feat: fire webhooks on jetstream events, cap at 10 per DID

- db.ts: getWebhooksByDid returns the 10 most recent subscriptions (ORDER BY id DESC LIMIT 10)
- jetstream.ts: fireWebhooks fans out to registered URLs via Promise.allSettled after each create/update/delete event
This commit is contained in:
Julien Calixte
2026-02-25 22:51:48 +01:00
parent 373b7a6777
commit 62f981dd93
2 changed files with 35 additions and 3 deletions

View File

@@ -2,6 +2,7 @@ import { Jetstream } from "@skyware/jetstream";
import {
deleteNote,
getCursor,
getWebhooksByDid,
saveCursor,
upsertNote,
} from "./src/data/db.ts";
@@ -16,6 +17,28 @@ globalThis.addEventListener("error", (e) => {
log("[jetstream] uncaught error:", e.error);
});
const fireWebhooks = async (
did: string,
payload: Record<string, unknown>,
): Promise<void> => {
const webhooks = getWebhooksByDid(did);
if (webhooks.length === 0) return;
const results = await Promise.allSettled(
webhooks.map(({ method, url }) =>
fetch(url, {
method,
headers: { "Content-Type": "application/json" },
body: JSON.stringify(payload),
})
),
);
for (const result of results) {
if (result.status === "rejected") {
log(`[jetstream] webhook error for ${did}:`, result.reason);
}
}
};
const cursor = getCursor();
log(`[jetstream] starting with cursor: ${cursor ?? "none"}`);
@@ -25,36 +48,39 @@ const jetstream = new Jetstream({
endpoint: "https://jetstream2.fr.hose.cam/subscribe"
});
jetstream.onCreate("space.remanso.note", (event) => {
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<Note, "did" | "rkey">;
upsertNote({ did, rkey, ...note });
log(`[jetstream] create ${did}/${rkey}: ${note.title}`);
await fireWebhooks(did, { event: "create", did, rkey, ...note });
} catch (error) {
log(`[jetstream] error on create:`, error);
}
});
jetstream.onUpdate("space.remanso.note", (event) => {
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<Note, "did" | "rkey">;
upsertNote({ did, rkey, ...note });
log(`[jetstream] update ${did}/${rkey}: ${note.title}`);
await fireWebhooks(did, { event: "update", did, rkey, ...note });
} catch (error) {
log(`[jetstream] error on update:`, error);
}
});
jetstream.onDelete("space.remanso.note", (event) => {
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 fireWebhooks(did, { event: "delete", did, rkey });
} catch (error) {
log(`[jetstream] error on delete:`, error);
}