catchUpFromRelay

tcmts.atproto.catchUpFromRelay · #vddk45ja · function · runs on the server

uses tcmts.atproto.relay, which uses process

Read what accounts hold of each collection, without Jetstream: ask the relay which accounts have records of it (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, time now). An account that can't be read is said, and skipped. At most pages pages of each listing are read (of 500 accounts, or 100 records).

import { listRecords, relay } from "#codebase/tcmts/atproto";
import type { RawEvent } from "@bsky/jetstream";

export async function catchUpFromRelay(
  collections: readonly string[],
  take: (e: RawEvent) => Promise<void>,
  o: { pages?: number } = {},
): Promise<number> {
  const most = o.pages ?? 50;
  let n = 0;
  for (const collection of collections) {
    let cursor: string | undefined;
    let pages = 0;
    do {
      const q = new URLSearchParams({ collection, limit: "500", ...(cursor ? { cursor } : {}) });
      const res = await fetch(`${relay()}/xrpc/com.atproto.sync.listReposByCollection?${q}`);
      if (!res.ok) throw new Error(`${relay()}: listReposByCollection ${res.status}`);
      const page = (await res.json()) as { repos: { did: string }[]; cursor?: string };
      for (const { did } of page.repos) {
        try {
          let rc: string | undefined;
          let read = 0;
          do {
            const records = await listRecords(did, collection, { limit: 100, cursor: rc });
            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++;
            }
            rc = records.records.length && ++read < most ? records.cursor : undefined;
          } while (rc);
        } catch (e) {
          console.warn(
            `atproto: couldn't read ${did}'s ${collection} records: ${(e as Error).message}`,
          );
        }
      }
      cursor = page.repos.length && ++pages < most ? page.cursor : undefined;
    } while (cursor);
  }
  return n;
}

uses

used by