From d7e189c61f95c2c8e779e92cc41ffa39bd00c7ba Mon Sep 17 00:00:00 2001
From: Julien Calixte <julien.calixte@theodo.com>
Date: Sun, 7 Jun 2026 11:26:20 +0200
Subject: [PATCH] feat(scripts): add OpenSearch backfill task

Iterates distinct DIDs in SQLite, resolves each one's PDS, and reindexes
every space.remanso.note record into OpenSearch via
com.atproto.repo.listRecords. Needed because note content is not
persisted to SQLite, so the index would otherwise only cover notes
created after deployment.
---
 deno.json                      |   3 +-
 scripts/backfill-opensearch.ts | 125 +++++++++++++++++++++++++++++++++
 2 files changed, 127 insertions(+), 1 deletion(-)
 create mode 100644 scripts/backfill-opensearch.ts

diff --git a/deno.json b/deno.json
index c5ab1de..cef5364 100644
--- a/deno.json
+++ b/deno.json
@@ -6,7 +6,8 @@
     "server:prod": "deno run --allow-net --allow-read --allow-write --allow-env --allow-ffi --unstable-ffi server.ts",
     "migrate": "deno run --allow-net --allow-read --allow-write --allow-env --allow-ffi --unstable-ffi src/migrations/init.ts",
     "webhooks": "deno run --allow-net --allow-env scripts/manage-webhooks.ts",
-    "webhooks:all": "deno run --allow-net --allow-env scripts/list-all-webhooks.ts"
+    "webhooks:all": "deno run --allow-net --allow-env scripts/list-all-webhooks.ts",
+    "backfill:search": "deno run --allow-net --allow-read --allow-write --allow-env --allow-ffi --unstable-ffi scripts/backfill-opensearch.ts"
   },
   "imports": {
     "@db/sqlite": "jsr:@db/sqlite@^0.13.0",
diff --git a/scripts/backfill-opensearch.ts b/scripts/backfill-opensearch.ts
new file mode 100644
index 0000000..d273687
--- /dev/null
+++ b/scripts/backfill-opensearch.ts
@@ -0,0 +1,125 @@
+// Backfill OpenSearch from each known user's PDS.
+//
+//   deno task backfill:search
+//
+// Reads distinct DIDs from the local SQLite database, resolves each DID's PDS
+// via the PLC directory, paginates `com.atproto.repo.listRecords` for
+// `space.remanso.note`, and indexes every record into OpenSearch.
+//
+// Requires OPENSEARCH_URL to be set; falls back to a no-op otherwise.
+// SQLite path follows the same SQLITE_PATH convention as the server/jetstream.
+
+import { db } from "../src/data/db.ts";
+import { resolvePds } from "../src/auth/verify.ts";
+import { ensureIndex, indexNote } from "../src/search/opensearch.ts";
+import { log } from "../src/log.ts";
+
+type ListRecordsResponse = {
+  records: { uri: string; value: Record<string, unknown> }[];
+  cursor?: string;
+};
+
+const COLLECTION = "space.remanso.note";
+const PAGE_SIZE = 100;
+
+const rkeyFromUri = (uri: string): string => {
+  const parts = uri.split("/");
+  return parts[parts.length - 1];
+};
+
+const backfillDid = async (
+  did: string,
+): Promise<{ indexed: number; failed: number }> => {
+  let indexed = 0;
+  let failed = 0;
+  const pds = await resolvePds(did);
+  let cursor: string | undefined = undefined;
+
+  do {
+    const url = new URL(`${pds}/xrpc/com.atproto.repo.listRecords`);
+    url.searchParams.set("repo", did);
+    url.searchParams.set("collection", COLLECTION);
+    url.searchParams.set("limit", String(PAGE_SIZE));
+    if (cursor) url.searchParams.set("cursor", cursor);
+
+    const res = await fetch(url);
+    if (!res.ok) {
+      throw new Error(`listRecords ${did}: ${res.status} ${await res.text()}`);
+    }
+    const page = (await res.json()) as ListRecordsResponse;
+
+    for (const record of page.records) {
+      const rkey = rkeyFromUri(record.uri);
+      const value = record.value as {
+        title?: string;
+        content?: string;
+        language?: string;
+        discoverable?: boolean;
+        publishedAt?: string;
+        createdAt?: string;
+      };
+      if (!value.title) {
+        failed++;
+        log(`[backfill] skip ${did}/${rkey}: missing title`);
+        continue;
+      }
+      try {
+        await indexNote({
+          did,
+          rkey,
+          title: value.title,
+          content: value.content,
+          language: value.language,
+          discoverable: value.discoverable,
+          publishedAt: value.publishedAt,
+          createdAt: value.createdAt,
+        });
+        indexed++;
+      } catch (error) {
+        failed++;
+        log(`[backfill] index ${did}/${rkey} failed:`, error);
+      }
+    }
+
+    cursor = page.cursor;
+  } while (cursor);
+
+  return { indexed, failed };
+};
+
+const main = async () => {
+  if (!Deno.env.get("OPENSEARCH_URL")) {
+    console.error("OPENSEARCH_URL is not set; aborting backfill.");
+    Deno.exit(1);
+  }
+
+  try {
+    await ensureIndex();
+  } catch (error) {
+    console.error("Failed to ensure index:", error);
+    Deno.exit(1);
+  }
+
+  const rows = db.prepare("SELECT DISTINCT did FROM note ORDER BY did").all<
+    { did: string }
+  >();
+  log(`[backfill] ${rows.length} DID(s) to process`);
+
+  let totalIndexed = 0;
+  let totalFailed = 0;
+  for (const { did } of rows) {
+    try {
+      const { indexed, failed } = await backfillDid(did);
+      log(`[backfill] ${did}: indexed=${indexed} failed=${failed}`);
+      totalIndexed += indexed;
+      totalFailed += failed;
+    } catch (error) {
+      log(`[backfill] ${did} aborted:`, error);
+      totalFailed++;
+    }
+  }
+  log(`[backfill] done. indexed=${totalIndexed} failed=${totalFailed}`);
+  db.close();
+};
+
+await main();