catchUpFromRelay

tcmts.registry.catchUpFromRelay · #bo54ywza · term · runs on the server

uses tcmts.registry.relayService, which uses process

Catch up without Jetstream's archive: ask the relay which accounts have records of each collection the registry keeps (com.atproto.sync.listReposByCollection), and read each one's from its PDS, as Jetstream would have sent them. Each is given to take as a commit event (seq 0: it moves no cursor). An account that can't be read is said, and skipped.

import { didDocument, guardedFetch, pdsOf } from "#codebase/tcmts/atproto";
import { REGISTRY_COLLECTIONS, REGISTRY_LIMITS, relayService } from "#codebase/tcmts/registry";
import type { RawEvent } from "@bsky/jetstream";

export async function catchUpFromRelay(take: (e: RawEvent) => Promise<void>): Promise<number> {
  let n = 0;
  for (const collection of REGISTRY_COLLECTIONS) {
    let cursor: string | undefined;
    let pages = 0;
    do {
      const q = new URLSearchParams({ collection, limit: "500", ...(cursor ? { cursor } : {}) });
      const res = await fetch(`${relayService()}/xrpc/com.atproto.sync.listReposByCollection?${q}`);
      if (!res.ok) throw new Error(`${relayService()}: listReposByCollection ${res.status}`);
      const page = (await res.json()) as { repos: { did: string }[]; cursor?: string };
      for (const { did } of page.repos) {
        try {
          const pds = pdsOf(await didDocument(did), did);
          let rc: string | undefined;
          let read = 0;
          do {
            const rq = new URLSearchParams({
              repo: did,
              collection,
              limit: "100",
              ...(rc ? { cursor: rc } : {}),
            });
            const got = await guardedFetch(
              `${pds}/xrpc/com.atproto.repo.listRecords?${rq}`,
              `${did}'s PDS`,
            );
            if (!got.ok) throw new Error(`listRecords ${got.status}`);
            const records = (await got.json()) as {
              records: { uri: string; cid: string; value: unknown }[];
              cursor?: string;
            };
            for (const r of records.records) {
              await take({
                kind: "commit",
                did,
                seq: 0,
                time: new Date().toISOString(),
                commit: {
                  operation: "create",
                  collection,
                  rkey: r.uri.slice(r.uri.lastIndexOf("/") + 1),
                  rev: "",
                  cid: r.cid,
                  record: r.value,
                },
              } as unknown as RawEvent);
              n++;
            }
            // (an account's records, a listing's pages: so many, however many a server says there are)
            rc =
              records.records.length && ++read < REGISTRY_LIMITS.pagesPerListing
                ? records.cursor
                : undefined;
          } while (rc);
        } catch (e) {
          console.warn(
            `registry: couldn't read ${did}'s ${collection} records: ${(e as Error).message}`,
          );
        }
      }
      cursor =
        page.repos.length && ++pages < REGISTRY_LIMITS.pagesPerListing ? page.cursor : undefined;
    } while (cursor);
  }
  return n;
}

uses

used by