readRecords

tcmts.registry.readRecords · #7zakivo7 · function · runs on the server

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

Read Jetstream's codes.tcm.* commits from where they were left until caught up (the atproto library's readJetstream): each event into the tables (#nauecrx6). Behind by more than the live stream reaches, every account's records of the collections the registry keeps are read as the relay lists them. Where it was left is kept in cursors, as feed jetstream.

import { readJetstream } from "#codebase/tcmts/atproto";
import { COLLECTIONS, cursors, REGISTRY_COLLECTIONS, takeEvent } from "#codebase/tcmts/registry";

export async function readRecords(
  o: { idleMs?: number; mostMs?: number } = {},
): Promise<{ events: number }> {
  return readJetstream({
    ...o,
    collections: COLLECTIONS,
    catchUp: REGISTRY_COLLECTIONS,
    take: takeEvent,
    cursor: {
      load: async () => (await cursors.findOne({ feed: "jetstream" }))?.value,
      save: async (c) => {
        const row = { feed: "jetstream", ...c };
        const had = await cursors.findOne({ feed: "jetstream" });
        if (had) await cursors.put(had.id, row);
        else await cursors.insert(row);
      },
    },
  });
}

uses

used by