readJetstream

tcmts.hub.readJetstream · #csojhmt4 · term · runs on the server

uses process

Read Jetstream's codes.tcm.* commits from the stored cursor until caught up (nothing new for a few seconds, or at most mostMs): each event's rows and the new cursor written in one transaction. With no cursor, it starts from the archive's beginning when there's an API key ($JETSTREAM_API_KEY, for Jetstream's archive); without one, it reads every account's records as the relay lists them (catchUpFromRelay), then the live stream from 36 hours back, as far as it reaches. A cursor the live stream no longer reaches (the hub was down for longer) is replayed the same way: from the archive, or the relay.

import {
  catchUpFromRelay, COLLECTIONS, cursors, jetstreamService, LIVE_LOOKBACK_MS, takeEvent,
} from "#codebase/tcmts/hub";
import { transaction } from "#codebase/tcmts/server";
import { Jetstream } from "@bsky/jetstream";
import type { RawEvent } from "@bsky/jetstream";

export async function readJetstream(
  o: { idleMs?: number; mostMs?: number } = {},
): Promise<{ events: number }> {
  const apiKey = process.env.JETSTREAM_API_KEY || undefined;
  const js = new Jetstream({ service: jetstreamService(), ...(apiKey ? { apiKey } : {}) });
  const kept = (await cursors.findOne({ feed: "jetstream" }))?.value.seq;
  let events = 0;
  const stop = new AbortController();
  const end = Date.now() + (o.mostMs ?? 50_000);
  let quiet: ReturnType<typeof setTimeout> | undefined;
  const hush = () => {
    clearTimeout(quiet);
    quiet = setTimeout(() => stop.abort(), o.idleMs ?? 5_000);
  };
  // (raw events: a record as the wire has it, for recordOf)
  const read = async (stream: AsyncIterable<unknown>) => {
    hush();
    for await (const x of stream) {
      const e = x as RawEvent;
      await transaction(async () => {
        await takeEvent(e);
        const c = await cursors.findOne({ feed: "jetstream" });
        if (c) await cursors.put(c.id, { feed: "jetstream", seq: e.seq });
        else await cursors.insert({ feed: "jetstream", seq: e.seq });
      });
      events++;
      hush();
      if (Date.now() > end) break;
    }
  };
  const common = {
    collections: COLLECTIONS,
    kinds: ["commit"] as "commit"[],
    raw: true as const,
    signal: stop.signal,
  };
  const from = (seq: number | undefined) => ({
    load: async () => seq,
    save: async () => undefined,
  });
  // (without an archive: what accounts hold now, as their PDSes say it, before what Jetstream still has)
  const catchUp = async () => {
    try {
      const n = await catchUpFromRelay((e) => transaction(() => takeEvent(e)));
      console.log(`hub: read ${n} record(s) from the accounts the relay lists`);
    } catch (e) {
      console.warn(`hub: couldn't catch up from the relay: ${(e as Error).message}`);
    }
  };
  try {
    if (apiKey) await read(js.replay({ ...common, afterSeq: kept ?? 0 }));
    else
      try {
        if (kept === undefined) await catchUp();
        await read(
          js.live({
            ...common,
            // (the SDK resumes after the cursor it's given: the last event delivered)
            cursor: from(kept === undefined ? (Date.now() - LIVE_LOOKBACK_MS) * 1000 : kept),
          }),
        );
      } catch (e) {
        if (
          kept === undefined ||
          stop.signal.aborted ||
          !/CursorTooOld|response: 400/.test(String((e as Error)?.cause ?? e))
        )
          throw e;
        console.warn(
          `hub: Jetstream's live stream no longer reaches seq ${kept}; catching up from the relay, then reading from 36 hours back`,
        );
        await catchUp();
        hush();
        await read(js.live({ ...common, cursor: from((Date.now() - LIVE_LOOKBACK_MS) * 1000) }));
      }
  } catch (e) {
    if (!stop.signal.aborted) throw e;
  } finally {
    clearTimeout(quiet);
    stop.abort();
  }
  return { events };
}

uses

used by