docs(readme): document search endpoints and backfill

Single-node cluster with the security plugin disabled (in-cluster HTTP
only). jetstream and api now depend on it being healthy and receive
OPENSEARCH_URL via env.

README.md

@@ -42,12 +42,37 @@ docker run -p 8080:8080 -v litenote-data:/data litenote-jetstream
 ## API
 
 | Endpoint                             | Description                                                                                     |
-| -------------------------------- | ---------------------------------- |
-| `GET /notes?cursor=&limit=`      | Paginated notes from all users     |
-| `GET /:did/notes?cursor=&limit=` | Paginated notes for a specific DID |
+| ------------------------------------ | ----------------------------------------------------------------------------------------------- |
+| `GET /notes?cursor=&limit=`          | Paginated notes from all users (discoverable only)                                              |
+| `GET /:did/notes?cursor=&limit=`     | Paginated notes for a specific DID (discoverable only)                                          |
+| `GET /search?q=&cursor=&limit=`      | Full-text fuzzy search across discoverable notes. Returns `{ results, cursor? }`                |
+| `GET /:did/search?q=&cursor=&limit=` | Owner-scoped search incl. non-discoverable. Requires `Authorization: Bearer <jwt>` for that DID |
+
+Search hits include `{ did, rkey, title, snippet, score, publishedAt }`. The
+`snippet` is a sentence-bounded fragment from the matching content with the
+query terms wrapped in `<em>` tags. Queries use OpenSearch `multi_match` with
+`fuzziness: AUTO`, so small typos still match.
 
 ## Environment Variables
 
 | Variable              | Default    | Description                                                                                            |
-| ------------- | ---------- | -------------------------------- |
+| --------------------- | ---------- | ------------------------------------------------------------------------------------------------------ |
 | `SQLITE_PATH`         | `notes.db` | Path to the SQLite database file                                                                       |
+| `OPENSEARCH_URL`      | _(unset)_  | OpenSearch base URL (e.g. `http://opensearch:9200`). When unset, indexing and search degrade to no-ops |
+| `OPENSEARCH_USERNAME` | _(unset)_  | Optional basic-auth user                                                                               |
+| `OPENSEARCH_PASSWORD` | _(unset)_  | Optional basic-auth password                                                                           |
+| `OPENSEARCH_INDEX`    | `notes`    | Index name to read/write                                                                               |
+| `ADMIN_DIDS`          | _(empty)_  | Comma-separated DIDs allowed to hit `/admin/*`                                                         |
+
+## OpenSearch backfill
+
+Note content arrives via the jetstream firehose only for new and updated
+records. To populate the index with pre-existing notes, run:
+
+```bash
+OPENSEARCH_URL=http://localhost:9200 deno task backfill:search
+```
+
+The script reads every DID in the local SQLite database, resolves each one's
+PDS, and reindexes their notes via `com.atproto.repo.listRecords`. Safe to
+re-run; document IDs are deterministic (`<did>:<rkey>`).

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

docker-compose.yml

@@ -6,6 +6,27 @@ services:
     volumes:
       - ${DATA_VOLUME:-data}:/data
 
+  opensearch:
+    image: opensearchproject/opensearch:2.18.0
+    restart: unless-stopped
+    environment:
+      - discovery.type=single-node
+      - DISABLE_SECURITY_PLUGIN=true
+      - OPENSEARCH_JAVA_OPTS=-Xms512m -Xmx512m
+    volumes:
+      - ${OPENSEARCH_VOLUME:-opensearch-data}:/usr/share/opensearch/data
+    expose:
+      - "9200"
+    healthcheck:
+      test: [
+        "CMD-SHELL",
+        "curl -fs http://localhost:9200/_cluster/health || exit 1",
+      ]
+      interval: 30s
+      timeout: 5s
+      retries: 10
+      start_period: 30s
+
   jetstream:
     build: .
     restart: unless-stopped
@@ -13,6 +34,10 @@ services:
     depends_on:
       migrate:
         condition: service_completed_successfully
+      opensearch:
+        condition: service_healthy
+    environment:
+      - OPENSEARCH_URL=http://opensearch:9200
     volumes:
       - ${DATA_VOLUME:-data}:/data
 
@@ -23,6 +48,10 @@ services:
     depends_on:
       migrate:
         condition: service_completed_successfully
+      opensearch:
+        condition: service_healthy
+    environment:
+      - OPENSEARCH_URL=http://opensearch:9200
     expose:
       - "8080"
     volumes:
@@ -36,3 +65,4 @@ services:
 
 volumes:
   data:
+  opensearch-data:

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

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

server.ts

@@ -11,6 +11,7 @@ import {
   type WebhookVerb,
 } from "./src/data/db.ts";
 import { authenticateRequest } from "./src/auth/verify.ts";
+import { searchNotes } from "./src/search/opensearch.ts";
 import { log } from "./src/log.ts";
 
 type AuthCtx = {
@@ -34,7 +35,7 @@ const requireDidOwnership = async (
   }
   if (verifiedDid !== did) {
     ctx.response.status = 403;
-    ctx.response.body = { error: "You can only manage your own webhooks" };
+    ctx.response.body = { error: "Forbidden: DID mismatch" };
     return false;
   }
   return true;
@@ -131,6 +132,43 @@ router.get("/:did/notes", (ctx) => {
   ctx.response.body = getNotesByDid(did, cursor, limit);
 });
 
+router.get("/search", async (ctx) => {
+  const q = ctx.request.url.searchParams.get("q") ?? "";
+  if (!q.trim()) {
+    ctx.response.status = 400;
+    ctx.response.body = { error: "q is required" };
+    return;
+  }
+  const cursor = ctx.request.url.searchParams.get("cursor") ?? undefined;
+  const limit = Number(ctx.request.url.searchParams.get("limit")) || PAGINATION;
+  ctx.response.body = await searchNotes({
+    q,
+    discoverableOnly: true,
+    cursor,
+    limit,
+  });
+});
+
+router.get("/:did/search", async (ctx) => {
+  const { did } = ctx.params;
+  if (!(await requireDidOwnership(ctx, did))) return;
+  const q = ctx.request.url.searchParams.get("q") ?? "";
+  if (!q.trim()) {
+    ctx.response.status = 400;
+    ctx.response.body = { error: "q is required" };
+    return;
+  }
+  const cursor = ctx.request.url.searchParams.get("cursor") ?? undefined;
+  const limit = Number(ctx.request.url.searchParams.get("limit")) || PAGINATION;
+  ctx.response.body = await searchNotes({
+    q,
+    discoverableOnly: false,
+    did,
+    cursor,
+    limit,
+  });
+});
+
 router.post("/notes/feed", async (ctx) => {
   const body = await ctx.request.body.json();
   const { dids, cursor, limit } = body ?? {};

src/auth/verify.ts

@@ -14,7 +14,7 @@ function decodeJwtPayload(token: string): JwtPayload {
   return JSON.parse(atob(payload));
 }
 
-async function resolvePds(did: string): Promise<string> {
+export async function resolvePds(did: string): Promise<string> {
   const res = await fetch(`https://plc.directory/${did}`);
   if (!res.ok) throw new Error(`Failed to resolve DID: ${res.status}`);
   const doc: DidDocument = await res.json();

src/data/note.ts

@@ -5,5 +5,6 @@ export type Note = {
   publishedAt: string;
   createdAt: string;
   discoverable?: boolean;
-  language?: string
+  language?: string;
+  content?: string;
 };

src/search/opensearch.ts

@@ -0,0 +1,261 @@
+import { log } from "../log.ts";
+
+const OPENSEARCH_URL = Deno.env.get("OPENSEARCH_URL");
+const OPENSEARCH_USERNAME = Deno.env.get("OPENSEARCH_USERNAME");
+const OPENSEARCH_PASSWORD = Deno.env.get("OPENSEARCH_PASSWORD");
+const OPENSEARCH_INDEX = Deno.env.get("OPENSEARCH_INDEX") ?? "notes";
+
+const MAX_LIMIT = 100;
+
+const enabled = (): boolean => {
+  if (!OPENSEARCH_URL) return false;
+  return true;
+};
+
+const authHeaders = (): Record<string, string> => {
+  if (OPENSEARCH_USERNAME && OPENSEARCH_PASSWORD) {
+    const creds = btoa(`${OPENSEARCH_USERNAME}:${OPENSEARCH_PASSWORD}`);
+    return { Authorization: `Basic ${creds}` };
+  }
+  return {};
+};
+
+const docId = (did: string, rkey: string): string =>
+  encodeURIComponent(`${did}:${rkey}`);
+
+const request = async (
+  method: string,
+  path: string,
+  body?: unknown,
+): Promise<Response> => {
+  const url = `${OPENSEARCH_URL}${path}`;
+  const headers: Record<string, string> = { ...authHeaders() };
+  if (body !== undefined) headers["Content-Type"] = "application/json";
+  return await fetch(url, {
+    method,
+    headers,
+    body: body !== undefined ? JSON.stringify(body) : undefined,
+  });
+};
+
+const indexMappings = {
+  settings: {
+    "index.mapping.total_fields.limit": 64,
+  },
+  mappings: {
+    properties: {
+      did: { type: "keyword" },
+      rkey: { type: "keyword" },
+      title: { type: "text", analyzer: "standard" },
+      content: { type: "text", analyzer: "standard" },
+      language: { type: "keyword" },
+      discoverable: { type: "boolean" },
+      publishedAt: { type: "date" },
+      createdAt: { type: "date" },
+    },
+  },
+};
+
+export const ensureIndex = async (): Promise<void> => {
+  if (!enabled()) return;
+  const head = await request("HEAD", `/${OPENSEARCH_INDEX}`);
+  if (head.status === 200) return;
+  const res = await request("PUT", `/${OPENSEARCH_INDEX}`, indexMappings);
+  if (res.ok) {
+    log(`[opensearch] created index ${OPENSEARCH_INDEX}`);
+    return;
+  }
+  const text = await res.text();
+  if (text.includes("resource_already_exists_exception")) return;
+  throw new Error(`Failed to create index: ${res.status} ${text}`);
+};
+
+export type IndexedNote = {
+  did: string;
+  rkey: string;
+  title: string;
+  content?: string;
+  language?: string;
+  discoverable?: boolean;
+  publishedAt?: string;
+  createdAt?: string;
+};
+
+export const indexNote = async (note: IndexedNote): Promise<void> => {
+  if (!enabled()) return;
+  const doc = {
+    did: note.did,
+    rkey: note.rkey,
+    title: note.title,
+    content: note.content ?? "",
+    language: note.language,
+    discoverable: note.discoverable !== false,
+    publishedAt: note.publishedAt,
+    createdAt: note.createdAt,
+  };
+  const res = await request(
+    "PUT",
+    `/${OPENSEARCH_INDEX}/_doc/${docId(note.did, note.rkey)}`,
+    doc,
+  );
+  if (!res.ok) {
+    const text = await res.text();
+    throw new Error(
+      `Failed to index ${note.did}/${note.rkey}: ${res.status} ${text}`,
+    );
+  }
+};
+
+export const removeNote = async (
+  did: string,
+  rkey: string,
+): Promise<void> => {
+  if (!enabled()) return;
+  const res = await request(
+    "DELETE",
+    `/${OPENSEARCH_INDEX}/_doc/${docId(did, rkey)}`,
+  );
+  if (!res.ok && res.status !== 404) {
+    const text = await res.text();
+    throw new Error(`Failed to delete ${did}/${rkey}: ${res.status} ${text}`);
+  }
+};
+
+export type SearchHit = {
+  did: string;
+  rkey: string;
+  title: string;
+  snippet: string;
+  score: number;
+  publishedAt?: string;
+};
+
+export type SearchOptions = {
+  q: string;
+  discoverableOnly: boolean;
+  did?: string;
+  cursor?: string;
+  limit: number;
+};
+
+const decodeCursor = (cursor: string): [number, string] | undefined => {
+  try {
+    const parsed = JSON.parse(atob(cursor));
+    if (
+      Array.isArray(parsed) &&
+      parsed.length === 2 &&
+      typeof parsed[0] === "number" &&
+      typeof parsed[1] === "string"
+    ) {
+      return [parsed[0], parsed[1]];
+    }
+  } catch {
+    // fall through
+  }
+  log(`[opensearch] ignoring malformed cursor: ${cursor}`);
+  return undefined;
+};
+
+const encodeCursor = (score: number, rkey: string): string =>
+  btoa(JSON.stringify([score, rkey]));
+
+const firstSnippet = (
+  highlight: Record<string, string[]> | undefined,
+  source: { content?: string; title?: string },
+): string => {
+  if (highlight?.content?.length) return highlight.content[0];
+  if (highlight?.title?.length) return highlight.title[0];
+  if (source.content) return source.content.slice(0, 200);
+  return source.title ?? "";
+};
+
+export const searchNotes = async (
+  opts: SearchOptions,
+): Promise<{ results: SearchHit[]; cursor?: string }> => {
+  if (!enabled()) {
+    log("[opensearch] search requested but OPENSEARCH_URL is not set");
+    return { results: [] };
+  }
+  const limit = Math.max(1, Math.min(opts.limit, MAX_LIMIT));
+  const filters: Record<string, unknown>[] = [];
+  if (opts.discoverableOnly) {
+    filters.push({ term: { discoverable: true } });
+  }
+  if (opts.did) {
+    filters.push({ term: { did: opts.did } });
+  }
+
+  const body: Record<string, unknown> = {
+    size: limit,
+    query: {
+      bool: {
+        must: [
+          {
+            multi_match: {
+              query: opts.q,
+              fields: ["title^2", "content"],
+              fuzziness: "AUTO",
+            },
+          },
+        ],
+        filter: filters,
+      },
+    },
+    highlight: {
+      fields: {
+        content: {
+          fragment_size: 200,
+          number_of_fragments: 1,
+          boundary_scanner: "sentence",
+          boundary_scanner_locale: "en-US",
+        },
+        title: { number_of_fragments: 0 },
+      },
+    },
+    sort: [{ _score: "desc" }, { rkey: "desc" }],
+    _source: ["did", "rkey", "title", "content", "publishedAt"],
+  };
+
+  if (opts.cursor) {
+    const decoded = decodeCursor(opts.cursor);
+    if (decoded) body.search_after = decoded;
+  }
+
+  const res = await request("POST", `/${OPENSEARCH_INDEX}/_search`, body);
+  if (!res.ok) {
+    const text = await res.text();
+    throw new Error(`Search failed: ${res.status} ${text}`);
+  }
+  const json = await res.json();
+  const hits = json?.hits?.hits ?? [];
+
+  const results: SearchHit[] = hits.map(
+    (h: {
+      _source: {
+        did: string;
+        rkey: string;
+        title: string;
+        content?: string;
+        publishedAt?: string;
+      };
+      _score: number;
+      highlight?: Record<string, string[]>;
+    }) => ({
+      did: h._source.did,
+      rkey: h._source.rkey,
+      title: h._source.title,
+      snippet: firstSnippet(h.highlight, h._source),
+      score: h._score,
+      publishedAt: h._source.publishedAt,
+    }),
+  );
+
+  const cursor = results.length === limit
+    ? encodeCursor(
+      results[results.length - 1].score,
+      results[results.length - 1].rkey,
+    )
+    : undefined;
+
+  return { results, cursor };
+};