refactor(jetstream): extract webhook dispatch and bulk-create modules

dispatchAll, fireWebhooks, and the bulk-create debounce buffer
move out of jetstream.ts into src/jetstream/{webhooks,bulk}.ts.
fireWebhooks and the bulk helpers now take their DB lookup and
dispatch as parameters so tests can inject stubs. Top-level
Jetstream construction, event handlers, ensureIndex call, and
the cursor-saving interval are gated behind import.meta.main so
importing jetstream.ts no longer opens a WebSocket.

jetstream.ts

@@ -5,11 +5,12 @@ import {
   getWebhooksByDidAndVerb,
   saveCursor,
   upsertNote,
-  type WebhookVerb,
 } from "./src/data/db.ts";
 import { Note } from "./src/data/note.ts";
 import { log } from "./src/log.ts";
 import { ensureIndex, indexNote, removeNote } from "./src/search/opensearch.ts";
+import { dispatchAll, fireWebhooks } from "./src/jetstream/webhooks.ts";
+import { type BulkDeps, queueBulkCreate } from "./src/jetstream/bulk.ts";
 
 const safeIndex = async (note: Note): Promise<void> => {
   try {
@@ -27,109 +28,30 @@ const safeRemove = async (did: string, rkey: string): Promise<void> => {
   }
 };
 
-globalThis.addEventListener("unhandledrejection", (e) => {
+const bulkDeps: BulkDeps = {
+  getSubs: getWebhooksByDidAndVerb,
+  dispatch: dispatchAll,
+};
+
+if (import.meta.main) {
+  globalThis.addEventListener("unhandledrejection", (e) => {
     log("[jetstream] unhandled rejection:", e.reason);
-});
+  });
 
-globalThis.addEventListener("error", (e) => {
+  globalThis.addEventListener("error", (e) => {
     log("[jetstream] uncaught error:", e.error);
-});
-
-type WebhookTarget = {
-  method: string;
-  url: string;
-  token?: string;
-};
-
-const dispatchAll = async (
-  webhooks: WebhookTarget[],
-  payload: Record<string, unknown>,
-  label: string,
-): Promise<void> => {
-  if (webhooks.length === 0) return;
-  const results = await Promise.allSettled(
-    webhooks.map(({ method, url, token }) => {
-      const hasBody = method !== "GET" && method !== "HEAD";
-      return fetch(url, {
-        method,
-        headers: {
-          ...(hasBody ? { "Content-Type": "application/json" } : {}),
-          ...(token ? { Authorization: `Bearer ${token}` } : {}),
-        },
-        body: hasBody ? JSON.stringify(payload) : undefined,
   });
-    }),
-  );
-  for (const result of results) {
-    if (result.status === "rejected") {
-      log(`[jetstream] ${label} webhook error:`, result.reason);
-    }
-  }
-};
 
-const fireWebhooks = async (
+  const cursor = getCursor();
-  did: string,
+  log(`[jetstream] starting with cursor: ${cursor ?? "none"}`);
-  verb: WebhookVerb,
-  payload: Record<string, unknown>,
-): Promise<void> => {
-  const webhooks = getWebhooksByDidAndVerb(did, verb);
-  await dispatchAll(webhooks, payload, `${verb} ${did}`);
-};
 
-const BULK_CREATE_DEBOUNCE_MS = 500;
+  const jetstream = new Jetstream({
-
-type BulkBuffer = {
-  records: Record<string, unknown>[];
-  timer: number;
-};
-const bulkBuffers = new Map<string, BulkBuffer>();
-
-const flushBulkCreate = async (did: string): Promise<void> => {
-  const buffer = bulkBuffers.get(did);
-  if (!buffer) return;
-  bulkBuffers.delete(did);
-  const webhooks = getWebhooksByDidAndVerb(did, "bulk-create");
-  log(`[jetstream] bulk-create ${did}: ${buffer.records.length} record(s)`);
-  await dispatchAll(
-    webhooks,
-    { event: "bulk-create", did, records: buffer.records },
-    `bulk-create ${did}`,
-  );
-};
-
-// Buffered records are not persisted: if jetstream restarts mid-window, those
-// `bulk-create` notifications are lost. Subscribers reconcile on cold start
-// because the underlying notes are already saved to the `note` table.
-const queueBulkCreate = (
-  did: string,
-  record: Record<string, unknown>,
-): void => {
-  const existing = bulkBuffers.get(did);
-  if (existing) {
-    clearTimeout(existing.timer);
-    existing.records.push(record);
-    existing.timer = setTimeout(
-      () => flushBulkCreate(did),
-      BULK_CREATE_DEBOUNCE_MS,
-    );
-    return;
-  }
-  bulkBuffers.set(did, {
-    records: [record],
-    timer: setTimeout(() => flushBulkCreate(did), BULK_CREATE_DEBOUNCE_MS),
-  });
-};
-
-const cursor = getCursor();
-log(`[jetstream] starting with cursor: ${cursor ?? "none"}`);
-
-const jetstream = new Jetstream({
     wantedCollections: ["space.remanso.note"],
     cursor: cursor ? Number(cursor) : undefined,
     endpoint: "https://jetstream2.fr.hose.cam/subscribe",
-});
+  });
 
-jetstream.onCreate("space.remanso.note", async (event) => {
+  jetstream.onCreate("space.remanso.note", async (event) => {
     try {
       const {
         did,
@@ -140,16 +62,21 @@ jetstream.onCreate("space.remanso.note", async (event) => {
       upsertNote({ did, rkey, ...note });
       log(`[jetstream] create ${did}/${rkey}: ${note.title}`);
       await Promise.allSettled([
-      fireWebhooks(did, "create", { event: "create", did, rkey, ...note }),
+        fireWebhooks(
+          did,
+          "create",
+          { event: "create", did, rkey, ...note },
+          getWebhooksByDidAndVerb,
+        ),
         safeIndex({ did, rkey, ...note }),
       ]);
-    queueBulkCreate(did, { rkey, ...note });
+      queueBulkCreate(did, { rkey, ...note }, bulkDeps);
     } catch (error) {
       log(`[jetstream] error on create:`, error);
     }
-});
+  });
 
-jetstream.onUpdate("space.remanso.note", async (event) => {
+  jetstream.onUpdate("space.remanso.note", async (event) => {
     try {
       const {
         did,
@@ -161,16 +88,21 @@ jetstream.onUpdate("space.remanso.note", async (event) => {
       log(`[jetstream] update ${did}/${rkey}: ${note.title}`);
       // Updates fold into the `create` verb — subscribers reconcile by (did, rkey).
       await Promise.allSettled([
-      fireWebhooks(did, "create", { event: "create", did, rkey, ...note }),
+        fireWebhooks(
+          did,
+          "create",
+          { event: "create", did, rkey, ...note },
+          getWebhooksByDidAndVerb,
+        ),
         safeIndex({ did, rkey, ...note }),
       ]);
-    queueBulkCreate(did, { rkey, ...note });
+      queueBulkCreate(did, { rkey, ...note }, bulkDeps);
     } catch (error) {
       log(`[jetstream] error on update:`, error);
     }
-});
+  });
 
-jetstream.onDelete("space.remanso.note", async (event) => {
+  jetstream.onDelete("space.remanso.note", async (event) => {
     try {
       const {
         did,
@@ -180,37 +112,43 @@ jetstream.onDelete("space.remanso.note", async (event) => {
       deleteNote({ did, rkey });
       log(`[jetstream] delete ${did}/${rkey}`);
       await Promise.allSettled([
-      fireWebhooks(did, "delete", { event: "delete", did, rkey }),
+        fireWebhooks(
+          did,
+          "delete",
+          { event: "delete", did, rkey },
+          getWebhooksByDidAndVerb,
+        ),
         safeRemove(did, rkey),
       ]);
     } catch (error) {
       log(`[jetstream] error on delete:`, error);
     }
-});
+  });
 
-jetstream.on("open", () => {
+  jetstream.on("open", () => {
     log("[jetstream] connected");
-});
+  });
 
-jetstream.on("close", () => {
+  jetstream.on("close", () => {
     log("[jetstream] connection closed");
-});
+  });
 
-jetstream.on("error", (error) => {
+  jetstream.on("error", (error) => {
     log("[jetstream] connection closed with error", error);
-});
+  });
 
-try {
+  try {
     await ensureIndex();
-} catch (error) {
+  } catch (error) {
     log("[opensearch] ensureIndex failed at startup:", error);
-}
+  }
 
-log("[jetstream] launching");
+  log("[jetstream] launching");
-jetstream.start();
+  jetstream.start();
 
-setInterval(() => {
+  setInterval(() => {
     if (jetstream.cursor) {
       saveCursor(jetstream.cursor);
     }
-}, 5000);
+  }, 5000);
+}

src/jetstream/bulk.ts

@@ -0,0 +1,66 @@
+import { log } from "../log.ts";
+import type { GetSubsByVerb, WebhookTarget } from "./webhooks.ts";
+
+export const BULK_CREATE_DEBOUNCE_MS = 500;
+
+export type BulkDeps = {
+  getSubs: GetSubsByVerb;
+  dispatch: (
+    webhooks: WebhookTarget[],
+    payload: Record<string, unknown>,
+    label: string,
+  ) => Promise<void>;
+  debounceMs?: number;
+};
+
+type BulkBuffer = {
+  records: Record<string, unknown>[];
+  timer: number;
+};
+
+const bulkBuffers = new Map<string, BulkBuffer>();
+
+export const flushBulkCreate = async (
+  did: string,
+  deps: BulkDeps,
+): Promise<void> => {
+  const buffer = bulkBuffers.get(did);
+  if (!buffer) return;
+  bulkBuffers.delete(did);
+  const webhooks = deps.getSubs(did, "bulk-create");
+  log(`[jetstream] bulk-create ${did}: ${buffer.records.length} record(s)`);
+  await deps.dispatch(
+    webhooks,
+    { event: "bulk-create", did, records: buffer.records },
+    `bulk-create ${did}`,
+  );
+};
+
+// Buffered records are not persisted: if jetstream restarts mid-window, those
+// `bulk-create` notifications are lost. Subscribers reconcile on cold start
+// because the underlying notes are already saved to the `note` table.
+export const queueBulkCreate = (
+  did: string,
+  record: Record<string, unknown>,
+  deps: BulkDeps,
+): void => {
+  const ms = deps.debounceMs ?? BULK_CREATE_DEBOUNCE_MS;
+  const existing = bulkBuffers.get(did);
+  if (existing) {
+    clearTimeout(existing.timer);
+    existing.records.push(record);
+    existing.timer = setTimeout(() => flushBulkCreate(did, deps), ms);
+    return;
+  }
+  bulkBuffers.set(did, {
+    records: [record],
+    timer: setTimeout(() => flushBulkCreate(did, deps), ms),
+  });
+};
+
+export const _resetBulkBuffersForTest = (): void => {
+  for (const buffer of bulkBuffers.values()) {
+    clearTimeout(buffer.timer);
+  }
+  bulkBuffers.clear();
+};

src/jetstream/webhooks.ts

@@ -0,0 +1,49 @@
+import type { WebhookVerb } from "../data/db.ts";
+import { log } from "../log.ts";
+
+export type WebhookTarget = {
+  method: string;
+  url: string;
+  token?: string;
+};
+
+export const dispatchAll = async (
+  webhooks: WebhookTarget[],
+  payload: Record<string, unknown>,
+  label: string,
+): Promise<void> => {
+  if (webhooks.length === 0) return;
+  const results = await Promise.allSettled(
+    webhooks.map(({ method, url, token }) => {
+      const hasBody = method !== "GET" && method !== "HEAD";
+      return fetch(url, {
+        method,
+        headers: {
+          ...(hasBody ? { "Content-Type": "application/json" } : {}),
+          ...(token ? { Authorization: `Bearer ${token}` } : {}),
+        },
+        body: hasBody ? JSON.stringify(payload) : undefined,
+      });
+    }),
+  );
+  for (const result of results) {
+    if (result.status === "rejected") {
+      log(`[jetstream] ${label} webhook error:`, result.reason);
+    }
+  }
+};
+
+export type GetSubsByVerb = (
+  did: string,
+  verb: WebhookVerb,
+) => WebhookTarget[];
+
+export const fireWebhooks = async (
+  did: string,
+  verb: WebhookVerb,
+  payload: Record<string, unknown>,
+  getSubs: GetSubsByVerb,
+): Promise<void> => {
+  const webhooks = getSubs(did, verb);
+  await dispatchAll(webhooks, payload, `${verb} ${did}`);
+};