julien/remanso-jetstream
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(jetstream): index notes in OpenSearch on write

Browse Source 
Wires create/update/delete events to indexNote/removeNote alongside the
existing webhook dispatch via Promise.allSettled, so search-index
failures cannot break webhook delivery. Calls ensureIndex once at
startup behind a try/catch so a transient OpenSearch outage does not
crash the listener.
This commit is contained in:
Julien Calixte
2026-06-07 11:26:08 +02:00
parent 45db68c34d
commit 937e16e602
1 changed files with 35 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
jetstream.ts
View File

@@ -9,6 +9,23 @@ import {
} from "./src/data/db.ts"; } from "./src/data/db.ts";
import { Note } from "./src/data/note.ts"; import { Note } from "./src/data/note.ts";
import { log } from "./src/log.ts"; import { log } from "./src/log.ts";
import { ensureIndex, indexNote, removeNote } from "./src/search/opensearch.ts";
const safeIndex = async (note: Note): Promise<void> => {
try {
await indexNote(note);
} catch (error) {
log(`[opensearch] index ${note.did}/${note.rkey} failed:`, error);
}
};
const safeRemove = async (did: string, rkey: string): Promise<void> => {
try {
await removeNote(did, rkey);
} catch (error) {
log(`[opensearch] delete ${did}/${rkey} failed:`, error);
}
};
globalThis.addEventListener("unhandledrejection", (e) => { globalThis.addEventListener("unhandledrejection", (e) => {
log("[jetstream] unhandled rejection:", e.reason); log("[jetstream] unhandled rejection:", e.reason);
@@ -122,7 +139,10 @@ jetstream.onCreate("space.remanso.note", async (event) => {
const note = record as unknown as Omit<Note, "did" | "rkey">; const note = record as unknown as Omit<Note, "did" | "rkey">;
upsertNote({ did, rkey, ...note }); upsertNote({ did, rkey, ...note });
log(`[jetstream] create ${did}/${rkey}: ${note.title}`); log(`[jetstream] create ${did}/${rkey}: ${note.title}`);
await fireWebhooks(did, "create", { event: "create", did, rkey, ...note }); await Promise.allSettled([
fireWebhooks(did, "create", { event: "create", did, rkey, ...note }),
safeIndex({ did, rkey, ...note }),
]);
queueBulkCreate(did, { rkey, ...note }); queueBulkCreate(did, { rkey, ...note });
} catch (error) { } catch (error) {
log(`[jetstream] error on create:`, error); log(`[jetstream] error on create:`, error);
@@ -140,7 +160,10 @@ jetstream.onUpdate("space.remanso.note", async (event) => {
upsertNote({ did, rkey, ...note }); upsertNote({ did, rkey, ...note });
log(`[jetstream] update ${did}/${rkey}: ${note.title}`); log(`[jetstream] update ${did}/${rkey}: ${note.title}`);
// Updates fold into the `create` verb — subscribers reconcile by (did, rkey). // Updates fold into the `create` verb — subscribers reconcile by (did, rkey).
await fireWebhooks(did, "create", { event: "create", did, rkey, ...note }); await Promise.allSettled([
fireWebhooks(did, "create", { event: "create", did, rkey, ...note }),
safeIndex({ did, rkey, ...note }),
]);
queueBulkCreate(did, { rkey, ...note }); queueBulkCreate(did, { rkey, ...note });
} catch (error) { } catch (error) {
log(`[jetstream] error on update:`, error); log(`[jetstream] error on update:`, error);
@@ -156,7 +179,10 @@ jetstream.onDelete("space.remanso.note", async (event) => {
log(`[jetstream] deleting ${did}/${rkey}...`); log(`[jetstream] deleting ${did}/${rkey}...`);
deleteNote({ did, rkey }); deleteNote({ did, rkey });
log(`[jetstream] delete ${did}/${rkey}`); log(`[jetstream] delete ${did}/${rkey}`);
await fireWebhooks(did, "delete", { event: "delete", did, rkey }); await Promise.allSettled([
fireWebhooks(did, "delete", { event: "delete", did, rkey }),
safeRemove(did, rkey),
]);
} catch (error) { } catch (error) {
log(`[jetstream] error on delete:`, error); log(`[jetstream] error on delete:`, error);
} }
@@ -174,6 +200,12 @@ jetstream.on("error", (error) => {
log("[jetstream] connection closed with error", error); log("[jetstream] connection closed with error", error);
}); });
try {
await ensureIndex();
} catch (error) {
log("[opensearch] ensureIndex failed at startup:", error);
}
log("[jetstream] launching"); log("[jetstream] launching");
jetstream.start(); jetstream.start();