testJetstreamIsReadFromWhereItWasLeft

tcmts.atproto.test.testJetstreamIsReadFromWhereItWasLeft · #jnyblubu · function

Jetstream read as a job reads it: the first time, what accounts already hold, from the relay; then each read goes on from where the last was left, taking what's new once (not what the overlap read again), deletes too; the cursor kept by the feed's name.

import { configure, jetstreamCursors, readJetstream, signInWithAppPassword } from "#codebase/tcmts/atproto";
import { fakeNetwork } from "#codebase/tcmts/atproto/testing";
import type { RawEvent } from "@bsky/jetstream";
import assert from "node:assert/strict";

export const testJetstreamIsReadFromWhereItWasLeft = async () => {
  const alice = {
    did: "did:plc:aliceaaaaaaaaaaaaaaaaaaa",
    handle: "alice.test",
    password: "a-pass",
  };
  const net = await fakeNetwork([alice]);
  configure(net.config);
  const feed = `test ${Date.now()}`;
  const taken: string[] = [];
  const read = () =>
    readJetstream({
      collections: ["app.example.*"],
      catchUp: ["app.example.post"],
      feed,
      idleMs: 300,
      take: async (e: RawEvent) => {
        if (e.kind === "commit")
          taken.push(`${e.commit.operation} ${e.commit.collection}/${e.commit.rkey}`);
      },
    });
  try {
    net.seed(alice.did, "app.example.post", "first", { text: "Hello" });
    net.seed(alice.did, "app.other.post", "elsewhere", { text: "Not read" });
    await read();
    // (from the relay, and again from the stream's last few seconds: taking one again changes nothing)
    assert.ok(taken.includes("create app.example.post/first"), taken.join(", "));
    assert.ok(!taken.some((t) => t.includes("app.other")), taken.join(", "));
    const kept = await jetstreamCursors.get(feed);
    assert.ok(kept && kept.seq > 0 && kept.time! > 0, JSON.stringify(kept));

    taken.length = 0;
    const s = await signInWithAppPassword("alice.test", alice.password);
    await s.putRecord("app.example.like", "second", { subject: "first" });
    await s.deleteRecord("app.example.post", "first");
    assert.deepEqual((await read()).events, 2);
    assert.deepEqual(taken, ["create app.example.like/second", "delete app.example.post/first"]);

    taken.length = 0;
    assert.equal((await read()).events, 0);
    assert.ok((await jetstreamCursors.get(feed))!.seq >= kept!.seq + 2);
  } finally {
    configure({
      plc: undefined,
      appview: undefined,
      allowLocalNetwork: undefined,
      jetstream: undefined,
      relay: undefined,
    });
    await net.close();
  }
}

testJetstreamIsReadFromWhereItWasLeft

uses