From 01048c753e6dc9cfe07d18a224bc7607b658593d Mon Sep 17 00:00:00 2001 From: Julien Calixte Date: Sun, 7 Jun 2026 22:06:19 +0200 Subject: [PATCH] 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 | 290 +++++++++++++++----------------------- src/jetstream/bulk.ts | 66 +++++++++ src/jetstream/webhooks.ts | 49 +++++++ 3 files changed, 229 insertions(+), 176 deletions(-) create mode 100644 src/jetstream/bulk.ts create mode 100644 src/jetstream/webhooks.ts diff --git a/jetstream.ts b/jetstream.ts index 23fee68..4b409e4 100644 --- a/jetstream.ts +++ b/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 => { try { @@ -27,190 +28,127 @@ const safeRemove = async (did: string, rkey: string): Promise => { } }; -globalThis.addEventListener("unhandledrejection", (e) => { - log("[jetstream] unhandled rejection:", e.reason); -}); - -globalThis.addEventListener("error", (e) => { - log("[jetstream] uncaught error:", e.error); -}); - -type WebhookTarget = { - method: string; - url: string; - token?: string; +const bulkDeps: BulkDeps = { + getSubs: getWebhooksByDidAndVerb, + dispatch: dispatchAll, }; -const dispatchAll = async ( - webhooks: WebhookTarget[], - payload: Record, - label: string, -): Promise => { - 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 ( - did: string, - verb: WebhookVerb, - payload: Record, -): Promise => { - const webhooks = getWebhooksByDidAndVerb(did, verb); - await dispatchAll(webhooks, payload, `${verb} ${did}`); -}; - -const BULK_CREATE_DEBOUNCE_MS = 500; - -type BulkBuffer = { - records: Record[]; - timer: number; -}; -const bulkBuffers = new Map(); - -const flushBulkCreate = async (did: string): Promise => { - 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, -): 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), +if (import.meta.main) { + globalThis.addEventListener("unhandledrejection", (e) => { + log("[jetstream] unhandled rejection:", e.reason); }); -}; -const cursor = getCursor(); -log(`[jetstream] starting with cursor: ${cursor ?? "none"}`); + globalThis.addEventListener("error", (e) => { + log("[jetstream] uncaught error:", e.error); + }); -const jetstream = new Jetstream({ - wantedCollections: ["space.remanso.note"], - cursor: cursor ? Number(cursor) : undefined, - endpoint: "https://jetstream2.fr.hose.cam/subscribe", -}); + 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) => { + try { + const { + did, + commit: { rkey, record }, + } = event; + log(`[jetstream] creating ${did}/${rkey}...`); + const note = record as unknown as Omit; + upsertNote({ did, rkey, ...note }); + log(`[jetstream] create ${did}/${rkey}: ${note.title}`); + await Promise.allSettled([ + fireWebhooks( + did, + "create", + { event: "create", did, rkey, ...note }, + getWebhooksByDidAndVerb, + ), + safeIndex({ did, rkey, ...note }), + ]); + queueBulkCreate(did, { rkey, ...note }, bulkDeps); + } catch (error) { + log(`[jetstream] error on create:`, error); + } + }); + + jetstream.onUpdate("space.remanso.note", async (event) => { + try { + const { + did, + commit: { rkey, record }, + } = event; + log(`[jetstream] updating ${did}/${rkey}...`); + const note = record as unknown as Omit; + upsertNote({ did, rkey, ...note }); + 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 }, + getWebhooksByDidAndVerb, + ), + safeIndex({ did, rkey, ...note }), + ]); + queueBulkCreate(did, { rkey, ...note }, bulkDeps); + } catch (error) { + log(`[jetstream] error on update:`, error); + } + }); + + jetstream.onDelete("space.remanso.note", async (event) => { + try { + const { + did, + commit: { rkey }, + } = event; + log(`[jetstream] deleting ${did}/${rkey}...`); + deleteNote({ did, rkey }); + log(`[jetstream] delete ${did}/${rkey}`); + await Promise.allSettled([ + fireWebhooks( + did, + "delete", + { event: "delete", did, rkey }, + getWebhooksByDidAndVerb, + ), + safeRemove(did, rkey), + ]); + } catch (error) { + log(`[jetstream] error on delete:`, error); + } + }); + + jetstream.on("open", () => { + log("[jetstream] connected"); + }); + + jetstream.on("close", () => { + log("[jetstream] connection closed"); + }); + + jetstream.on("error", (error) => { + log("[jetstream] connection closed with error", error); + }); -jetstream.onCreate("space.remanso.note", async (event) => { try { - const { - did, - commit: { rkey, record }, - } = event; - log(`[jetstream] creating ${did}/${rkey}...`); - const note = record as unknown as Omit; - upsertNote({ did, rkey, ...note }); - log(`[jetstream] create ${did}/${rkey}: ${note.title}`); - await Promise.allSettled([ - fireWebhooks(did, "create", { event: "create", did, rkey, ...note }), - safeIndex({ did, rkey, ...note }), - ]); - queueBulkCreate(did, { rkey, ...note }); + await ensureIndex(); } catch (error) { - log(`[jetstream] error on create:`, error); + log("[opensearch] ensureIndex failed at startup:", error); } -}); -jetstream.onUpdate("space.remanso.note", async (event) => { - try { - const { - did, - commit: { rkey, record }, - } = event; - log(`[jetstream] updating ${did}/${rkey}...`); - const note = record as unknown as Omit; - upsertNote({ did, rkey, ...note }); - 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 }), - safeIndex({ did, rkey, ...note }), - ]); - queueBulkCreate(did, { rkey, ...note }); - } catch (error) { - log(`[jetstream] error on update:`, error); - } -}); + log("[jetstream] launching"); + jetstream.start(); -jetstream.onDelete("space.remanso.note", async (event) => { - try { - const { - did, - commit: { rkey }, - } = event; - log(`[jetstream] deleting ${did}/${rkey}...`); - deleteNote({ did, rkey }); - log(`[jetstream] delete ${did}/${rkey}`); - await Promise.allSettled([ - fireWebhooks(did, "delete", { event: "delete", did, rkey }), - safeRemove(did, rkey), - ]); - } catch (error) { - log(`[jetstream] error on delete:`, error); - } -}); - -jetstream.on("open", () => { - log("[jetstream] connected"); -}); - -jetstream.on("close", () => { - log("[jetstream] connection closed"); -}); - -jetstream.on("error", (error) => { - log("[jetstream] connection closed with error", error); -}); - -try { - await ensureIndex(); -} catch (error) { - log("[opensearch] ensureIndex failed at startup:", error); + setInterval(() => { + if (jetstream.cursor) { + saveCursor(jetstream.cursor); + } + }, 5000); } - -log("[jetstream] launching"); -jetstream.start(); - -setInterval(() => { - if (jetstream.cursor) { - saveCursor(jetstream.cursor); - } -}, 5000); diff --git a/src/jetstream/bulk.ts b/src/jetstream/bulk.ts new file mode 100644 index 0000000..a4d83b4 --- /dev/null +++ b/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, + label: string, + ) => Promise; + debounceMs?: number; +}; + +type BulkBuffer = { + records: Record[]; + timer: number; +}; + +const bulkBuffers = new Map(); + +export const flushBulkCreate = async ( + did: string, + deps: BulkDeps, +): Promise => { + 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, + 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(); +}; diff --git a/src/jetstream/webhooks.ts b/src/jetstream/webhooks.ts new file mode 100644 index 0000000..070cc5f --- /dev/null +++ b/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, + label: string, +): Promise => { + 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, + getSubs: GetSubsByVerb, +): Promise => { + const webhooks = getSubs(did, verb); + await dispatchAll(webhooks, payload, `${verb} ${did}`); +};