readJetstream

tcmts.atproto.readJetstream · #lxp2z3pk · function · runs on the server

uses process

Read the commits of collections (NSIDs, or prefix.*) from where this feed was left, until caught up (nothing new for idleMs, default 5 s) or for at most mostMs (default 50 s): each event given to take and the cursor kept in one transaction, so a job that runs it every minute goes on where the last left off (a record caught up from the relay may come again from the stream's last few seconds: taking it again should change nothing). With an API key (apiKey, else $JETSTREAM_API_KEY, for Jetstream's archive) the archive is replayed from the event read last. Without one, the live stream is read from a little before where it was left (LIVE_OVERLAP_MS: those taken already are skipped), if that's at most LIVE_REACH_MS ago; otherwise (never read, or left too long ago) what accounts hold of catchUp (default: the collections that aren't prefix.*) is read as the relay lists them (catchUpFromRelay), then the live stream from now. catchUp: false skips what was written meanwhile. Where it's kept: feed names it in jetstreamCursors (default: the collections), or cursor keeps it. (Records come as their author's PDS sent them, unchecked: verifiedRecord checks one.)

import { catchUpFromRelay, jetstream, jetstreamCursors, LIVE_OVERLAP_MS, LIVE_REACH_MS } from "#codebase/tcmts/atproto";
import type { JetstreamCursor } from "#codebase/tcmts/atproto";
import { transaction } from "#codebase/tcmts/atproto/lib/server";
import { Jetstream } from "@bsky/jetstream";
import type { CollectionFilter, RawEvent } from "@bsky/jetstream";

export async function readJetstream(o: {
  collections: readonly string[];
  take: (e: RawEvent) => Promise<void>;
  feed?: string;
  cursor?: {
    load(): Promise<JetstreamCursor | undefined>;
    save(c: JetstreamCursor): Promise<void>;
  };
  catchUp?: readonly string[] | false;
  apiKey?: string;
  idleMs?: number;
  mostMs?: number;
}): Promise<{ events: number }> {
  const feed = o.feed ?? o.collections.join(" ");
  const store = o.cursor ?? {
    load: () => jetstreamCursors.get(feed),
    save: (c: JetstreamCursor) => jetstreamCursors.put(feed, c),
  };
  const catchUp =
    o.catchUp === false ? [] : (o.catchUp ?? o.collections.filter((c) => !c.endsWith(".*")));
  const apiKey = o.apiKey ?? (process.env.JETSTREAM_API_KEY || undefined);
  const js = new Jetstream({ service: jetstream(), ...(apiKey ? { apiKey } : {}) });
  const kept = await store.load();
  const started = Date.now();
  let events = 0;
  let cut = false;
  // (the time it's read to, in microseconds, as Jetstream takes a time for its cursor)
  let time = kept?.time;
  let seq = kept?.seq ?? 0;
  const stop = new AbortController();
  const end = started + (o.mostMs ?? 50_000);
  let quiet: ReturnType<typeof setTimeout> | undefined;
  const hush = () => {
    clearTimeout(quiet);
    quiet = setTimeout(() => stop.abort(), o.idleMs ?? 5_000);
  };
  const at = (): JetstreamCursor => ({ seq, ...(time !== undefined ? { time } : {}) });
  // (those of the overlap already taken, by their number and time, skipped: taking one again may undo what was
  // done with it since)
  const read = async (stream: AsyncIterable<unknown>) => {
    hush();
    for await (const x of stream) {
      const e = x as RawEvent;
      const t = Date.parse((e as { time?: string }).time ?? "") * 1000;
      if (
        kept?.time !== undefined &&
        kept.seq > 0 &&
        e.seq <= kept.seq &&
        Number.isFinite(t) &&
        t <= kept.time
      ) {
        hush();
        continue;
      }
      if (Number.isFinite(t)) time = Math.max(time ?? 0, t);
      if (e.seq > 0) seq = Math.max(seq, e.seq);
      await transaction(async () => {
        await o.take(e);
        await store.save(at());
      });
      events++;
      hush();
      if (Date.now() > end) {
        cut = true;
        break;
      }
    }
  };
  const common = {
    collections: [...o.collections] as CollectionFilter[],
    kinds: ["commit"] as "commit"[],
    raw: true as const,
    signal: stop.signal,
  };
  const from = (cursor: number) => ({
    load: async () => cursor,
    save: async () => undefined,
  });
  try {
    if (apiKey) await read(js.replay({ ...common, afterSeq: kept?.seq ?? 0 }));
    else {
      if (time === undefined || started * 1000 - time > LIVE_REACH_MS * 1000) {
        // (what accounts hold now, as their PDSes say it, in place of what the stream would take too long to reach)
        if (catchUp.length)
          try {
            const n = await catchUpFromRelay(catchUp, (e) => transaction(() => o.take(e)));
            console.log(
              `atproto: read ${n} record(s) of ${feed} from the accounts the relay lists`,
            );
          } catch (e) {
            console.warn(
              `atproto: couldn't catch ${feed} up from the relay: ${(e as Error).message}`,
            );
          }
        time = started * 1000;
      }
      await read(js.live({ ...common, cursor: from(time - LIVE_OVERLAP_MS * 1000) }));
    }
  } catch (e) {
    if (!stop.signal.aborted) throw e;
  } finally {
    clearTimeout(quiet);
    stop.abort();
  }
  // caught up (nothing new for a while): read to when it started; cut short, to the last event read
  if (!apiKey && time !== undefined) {
    time = cut ? time : Math.max(time, started * 1000);
    await store.save(at());
  }
  return { events };
}

uses

used by