remanso-jetstream/tests/jetstream_bulk_test.ts

import { assertEquals } from "@std/assert";
import { FakeTime } from "@std/testing/time";
import {
  _resetBulkBuffersForTest,
  type BulkDeps,
  queueBulkCreate,
} from "../src/jetstream/bulk.ts";
import type { WebhookTarget } from "../src/jetstream/webhooks.ts";
import type { WebhookVerb } from "../src/data/db.ts";

type DispatchCall = {
  webhooks: WebhookTarget[];
  payload: Record<string, unknown>;
  label: string;
};

const flushMicrotasks = async (): Promise<void> => {
  for (let i = 0; i < 5; i++) {
    await Promise.resolve();
  }
};

const makeDeps = (
  webhooks: WebhookTarget[] = [],
): {
  deps: BulkDeps;
  getSubsCalls: { did: string; verb: WebhookVerb }[];
  dispatchCalls: DispatchCall[];
} => {
  const getSubsCalls: { did: string; verb: WebhookVerb }[] = [];
  const dispatchCalls: DispatchCall[] = [];
  const deps: BulkDeps = {
    getSubs: (did, verb) => {
      getSubsCalls.push({ did, verb });
      return webhooks;
    },
    dispatch: (w, payload, label) => {
      dispatchCalls.push({ webhooks: w, payload, label });
      return Promise.resolve();
    },
    debounceMs: 500,
  };
  return { deps, getSubsCalls, dispatchCalls };
};

Deno.test("queueBulkCreate accumulates records and flushes once after debounce", async () => {
  const time = new FakeTime();
  try {
    _resetBulkBuffersForTest();
    const { deps, dispatchCalls } = makeDeps();

    queueBulkCreate("did:plc:a", { rkey: "r1" }, deps);
    queueBulkCreate("did:plc:a", { rkey: "r2" }, deps);
    queueBulkCreate("did:plc:a", { rkey: "r3" }, deps);

    assertEquals(dispatchCalls.length, 0);

    await time.tickAsync(500);
    await flushMicrotasks();

    assertEquals(dispatchCalls.length, 1);
    const payload = dispatchCalls[0].payload as {
      records: { rkey: string }[];
    };
    assertEquals(payload.records.map((r) => r.rkey), ["r1", "r2", "r3"]);
  } finally {
    _resetBulkBuffersForTest();
    time.restore();
  }
});

Deno.test("queueBulkCreate restarts the timer on each push (debounce)", async () => {
  const time = new FakeTime();
  try {
    _resetBulkBuffersForTest();
    const { deps, dispatchCalls } = makeDeps();

    queueBulkCreate("did:plc:a", { rkey: "r1" }, deps);
    await time.tickAsync(400);
    queueBulkCreate("did:plc:a", { rkey: "r2" }, deps);
    await time.tickAsync(400);

    assertEquals(dispatchCalls.length, 0);

    await time.tickAsync(100);
    await flushMicrotasks();

    assertEquals(dispatchCalls.length, 1);
  } finally {
    _resetBulkBuffersForTest();
    time.restore();
  }
});

Deno.test("flushBulkCreate uses bulk-create verb and the standard payload shape", async () => {
  const time = new FakeTime();
  try {
    _resetBulkBuffersForTest();
    const targets: WebhookTarget[] = [
      { method: "POST", url: "https://hook.example/bulk" },
    ];
    const { deps, getSubsCalls, dispatchCalls } = makeDeps(targets);

    queueBulkCreate("did:plc:a", { rkey: "r1", title: "T" }, deps);
    await time.tickAsync(500);
    await flushMicrotasks();

    assertEquals(getSubsCalls, [{ did: "did:plc:a", verb: "bulk-create" }]);
    assertEquals(dispatchCalls.length, 1);
    assertEquals(dispatchCalls[0].webhooks, targets);
    assertEquals(dispatchCalls[0].label, "bulk-create did:plc:a");
    assertEquals(dispatchCalls[0].payload, {
      event: "bulk-create",
      did: "did:plc:a",
      records: [{ rkey: "r1", title: "T" }],
    });
  } finally {
    _resetBulkBuffersForTest();
    time.restore();
  }
});

Deno.test("bulkBuffers are cleared after flush and isolated per did", async () => {
  const time = new FakeTime();
  try {
    _resetBulkBuffersForTest();
    const { deps, dispatchCalls } = makeDeps();

    queueBulkCreate("did:plc:a", { rkey: "a1" }, deps);
    queueBulkCreate("did:plc:b", { rkey: "b1" }, deps);
    await time.tickAsync(500);
    await flushMicrotasks();

    assertEquals(dispatchCalls.length, 2);
    const dids = dispatchCalls.map((c) => (c.payload as { did: string }).did)
      .sort();
    assertEquals(dids, ["did:plc:a", "did:plc:b"]);

    // Another flush window should not re-dispatch the same records.
    await time.tickAsync(500);
    await flushMicrotasks();
    assertEquals(dispatchCalls.length, 2);
  } finally {
    _resetBulkBuffersForTest();
    time.restore();
  }
});