mirror

tcmts.registry.mirror · #npcimgy5 · term · runs on the server

uses tcmts.registry.releases, which uses tcmts.server.table, which uses tcmts.server.Table, which uses #tcm/host (database)

Mirror a release (a job, queued as it's announced): check its record against its author's signed repository, then fetch its namespace by hash from its sources (re-hashed as it arrives) into the registry's codebase, mounted read-only at mountFor, and note the libraries it holds (and, if the version after it mirrored first, say that one's changes since it). A record that doesn't check out fails it, and it never shows as that author's; a source that doesn't answer is retried, as a job is.

import { NotTheirs, resolveIdentity } from "#codebase/tcmts/atproto";
import { IDENT_PATH } from "#codebase/tcmts/cli/codebase";
import {
  fetchFromSources, librariesIn, librariesOf, namesNow, recordAt, recordFor, within, withRecord,
} from "#codebase/tcmts/cli/libraries";
import type { LibraryEntry } from "#codebase/tcmts/cli/libraries";
import { publicRemote } from "#codebase/tcmts/cli/sync";
import { IR_VERSION } from "#codebase/tcmts/core/hash";
import { compareVersions } from "#codebase/tcmts/core/semver";
import { recordKey } from "#codebase/tcmts/records";
import type { ReleaseRecord } from "#codebase/tcmts/records";
import {
  mountFor, people, REGISTRY_LIMITS, registryCodebase, releaseFacts, releaseFields, releases, signedRelease, uses,
} from "#codebase/tcmts/registry";
import type { RegistryRelease } from "#codebase/tcmts/registry";
import { transaction } from "#codebase/tcmts/server";

export async function mirror(uri: string): Promise<void> {
  const found = await releases.findOne({ uri });
  if (!found || found.value.state !== "pending") return;
  const r = found.value;
  const fail = (reason: string) =>
    releases.update(found.id, (v) => ({ ...v, state: "failed", reason }));
  let signed: { cid: string; record: ReleaseRecord };
  try {
    signed = await signedRelease(r.did, r.rkey);
  } catch (e) {
    if (e instanceof NotTheirs) return void (await fail(e.message));
    throw e;
  }
  if (signed.record.namespace !== r.namespace)
    return void (await fail(
      `the signed record names #${signed.record.namespace}, not #${r.namespace}`,
    ));
  if (recordKey(signed.record) !== r.rkey)
    return void (await fail(`the signed record is ${recordKey(signed.record)}'s, not ${r.rkey}'s`));
  if (signed.record.hashFormat !== IR_VERSION)
    return void (await fail(
      `announced in hash format ${signed.record.hashFormat}; this registry's is ${IR_VERSION}`,
    ));
  // (who they are, for showing: a handle that names the DID back, as the DID's document says it now)
  const handle = await resolveIdentity(r.did).then(
    (i) => i.handle,
    () => undefined,
  );
  const person = await people.findOne({ did: r.did });
  if (person) {
    const { handle: _, ...kept } = person.value;
    await people.put(person.id, { ...kept, ...(handle ? { handle } : {}) });
  }
  const cb = registryCodebase();
  const at = mountFor(r.did, r.package, r.version);
  if (!IDENT_PATH.test(at)) return void (await fail(`no name for it: ${at}`));
  const mounted = librariesIn(cb.store).find((l) => l.path === at);
  let from = mounted?.sources[0];
  if (!mounted) {
    // (what one account can make a registry hold: so many releases, of so many definitions each)
    if (
      (await releases.find({ did: r.did, state: "mirrored" })).length >=
      REGISTRY_LIMITS.releasesPerAccount
    )
      return void (await fail(
        `the registry mirrors at most ${REGISTRY_LIMITS.releasesPerAccount} releases of one account`,
      ));
    // (sources an announcement names: asked as a registry asks anyone's, with no credentials, public addresses only)
    const got = await fetchFromSources(
      cb,
      signed.record.sources,
      `#${r.namespace}`,
      (s) => publicRemote(s, `${r.did}'s source`),
      { maxNames: REGISTRY_LIMITS.definitionsPerRelease },
    );
    from = got.from;
    const names = namesNow(cb.store);
    if (names.some((n) => within(n.path, at)))
      return void (await fail(`${at} is taken in the registry's codebase`));
    const rows = cb.store
      .namespaceRows(r.namespace)
      .map((x) => ({ ...x, path: `${at}.${x.path}` }));
    const record = recordFor(at);
    const entry: LibraryEntry = {
      namespace: r.namespace,
      release: `${r.package}@${r.version}`,
      sources: signed.record.sources,
      did: r.did,
      ...(handle ? { handle } : {}),
    };
    cb.store.setBranchNames(
      cb.store.branch().name,
      [
        ...withRecord(cb, names, record, {
          ...recordAt(cb.store, record),
          [at.slice(at.lastIndexOf(".") + 1)]: entry,
        }),
        ...rows,
      ],
      `mirror ${at}`,
    );
  } else if (mounted.namespace !== r.namespace)
    return void (await fail(`${at} holds #${mounted.namespace} already`));
  // the libraries it holds, by hash: who depends on what, exactly
  const held = librariesOf(
    cb.store.namespaceRows(r.namespace),
    (h) => cb.store.getDef(h)?.sourceText ?? "",
  );
  // what it is, from its code; and how its API changed since the version before it (the highest mirrored below it)
  const rows = cb.store.namespaceRows(r.namespace);
  const earlier = (await releases.find({ did: r.did, package: r.package, state: "mirrored" }))
    .map((x) => x.value)
    .filter((x) => x.uri !== uri && compareVersions(x.version, r.version) < 0)
    .sort((a, b) => compareVersions(b.version, a.version))[0];
  const facts = releaseFacts(
    cb,
    rows,
    earlier && { version: earlier.version, rows: cb.store.namespaceRows(earlier.namespace) },
  );
  await transaction(async () => {
    for (const u of await uses.find({ uri })) await uses.delete(u.id);
    for (const l of held) await uses.insert({ uri, namespace: l.namespace, at: l.path });
    await releases.update(found.id, (v) => {
      // (what it says, as its author's repository signs it: not as the event that brought it said)
      const next: RegistryRelease = {
        ...v,
        ...releaseFields(signed.record),
        cid: signed.cid,
        state: "mirrored",
        mount: at,
        ...(from ? { from } : {}),
        verifiedAt: new Date(),
        facts,
      };
      delete next.reason;
      return next;
    });
  });
  // the version after it, if that mirrored first (jobs run side by side): its changes are since this one now
  const later = (await releases.find({ did: r.did, package: r.package, state: "mirrored" }))
    .filter((x) => compareVersions(x.value.version, r.version) > 0)
    .sort((a, b) => compareVersions(a.value.version, b.value.version))[0];
  const since = later?.value.facts?.after;
  if (later && (since === undefined || compareVersions(since, r.version) < 0)) {
    const laterFacts = releaseFacts(cb, cb.store.namespaceRows(later.value.namespace), {
      version: r.version,
      rows,
    });
    await releases.update(later.id, (v) => ({ ...v, facts: laterFacts }));
  }
}

uses

used by