From 937e16e6029648c86b6f0970f2e6bb1f390f7789 Mon Sep 17 00:00:00 2001
From: Julien Calixte <julien.calixte@theodo.com>
Date: Sun, 7 Jun 2026 11:26:08 +0200
Subject: [PATCH] feat(jetstream): index notes in OpenSearch on write

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.
---
 jetstream.ts | 38 +++++++++++++++++++++++++++++++++++---
 1 file changed, 35 insertions(+), 3 deletions(-)

diff --git a/jetstream.ts b/jetstream.ts
index 4e3caf9..23fee68 100644
--- a/jetstream.ts
+++ b/jetstream.ts
@@ -9,6 +9,23 @@ import {
 } 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";
+
+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) => {
   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">;
     upsertNote({ did, rkey, ...note });
     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 });
   } catch (error) {
     log(`[jetstream] error on create:`, error);
@@ -140,7 +160,10 @@ jetstream.onUpdate("space.remanso.note", async (event) => {
     upsertNote({ did, rkey, ...note });
     log(`[jetstream] update ${did}/${rkey}: ${note.title}`);
     // 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 });
   } catch (error) {
     log(`[jetstream] error on update:`, error);
@@ -156,7 +179,10 @@ jetstream.onDelete("space.remanso.note", async (event) => {
     log(`[jetstream] deleting ${did}/${rkey}...`);
     deleteNote({ 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) {
     log(`[jetstream] error on delete:`, error);
   }
@@ -174,6 +200,12 @@ jetstream.on("error", (error) => {
   log("[jetstream] connection closed with error", error);
 });
 
+try {
+  await ensureIndex();
+} catch (error) {
+  log("[opensearch] ensureIndex failed at startup:", error);
+}
+
 log("[jetstream] launching");
 jetstream.start();