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

@@ -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",

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();