fix(test): match getSession URL by suffix, not /getSession path

The PDS endpoint URL is /xrpc/com.atproto.server.getSession —
those are dots before getSession, not slashes, so the previous
includes("/getSession") filter never matched and the Authorization
header assertion silently short-circuited via optional chaining.

.dockerignore

@@ -19,6 +19,9 @@ remote-db/
 
 # admin / dev-only scripts (run locally, not in prod containers)
 scripts/
+# Backfill needs to run inside the api/jetstream container so it can read
+# the shared SQLite volume; ship it in the image.
+!scripts/backfill-opensearch.ts
 
 # logs and caches
 *.log

.gitea/workflows/ci.yml

@@ -0,0 +1,30 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+
+jobs:
+  check:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: denoland/setup-deno@v2
+        with:
+          deno-version: v2.x
+          cache: true
+          cache-hash: ${{ hashFiles('deno.lock') }}
+
+      - name: Format check
+        run: deno fmt --check
+
+      - name: Lint
+        run: deno lint
+
+      - name: Type check
+        run: deno check jetstream.ts server.ts scripts/*.ts src/migrations/init.ts tests/
+
+      - name: Test
+        run: deno test --allow-all tests/

.gitlab-ci.yml

@@ -1,23 +0,0 @@
-stages:
-  - build
-  - deploy
-
-variables:
-  DOCKER_DRIVER: overlay2
-  DOCKER_TLS_CERTDIR: "/certs"
-  IMAGE_TAG: $CI_REGISTRY/litenote:$CI_COMMIT_REF_SLUG
-  IMAGE_LATEST: $CI_REGISTRY/litenote:latest
-
-build:
-  stage: build
-  image: docker:24-dind
-  services:
-    - docker:24-dind
-  before_script:
-    - docker login -u $CI_REGISTRY_USER -p $CI_REGISTRY_PASSWORD $CI_REGISTRY
-  script:
-    - docker build -t $IMAGE_TAG -t $IMAGE_LATEST .
-    - docker push $IMAGE_TAG
-    - docker push $IMAGE_LATEST
-  only:
-    - main

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

@@ -1,4 +1,10 @@
 {
+  "fmt": {
+    "exclude": ["docs/"]
+  },
+  "lint": {
+    "exclude": ["docs/"]
+  },
   "tasks": {
     "jetstream:prod": "deno run --allow-net --allow-read --allow-write --allow-env --allow-ffi --unstable-ffi jetstream.ts",
     "jetstream": "deno run --watch --allow-net --allow-read --allow-write --allow-env --allow-ffi --unstable-ffi jetstream.ts",
@@ -6,12 +12,15 @@
     "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",
     "@oak/oak": "jsr:@oak/oak@^17.2.0",
+    "@opensearch-project/opensearch": "npm:@opensearch-project/opensearch@^3.6.0",
     "@skyware/jetstream": "npm:@skyware/jetstream@^0.2.5",
-    "@std/assert": "jsr:@std/assert@1"
+    "@std/assert": "jsr:@std/assert@1",
+    "@std/testing": "jsr:@std/testing@1"
   }
 }

deno.lock

@@ -7,8 +7,10 @@
     "jsr:@oak/commons@1": "1.0.0",
     "jsr:@oak/oak@^17.2.0": "17.2.0",
     "jsr:@std/assert@1": "1.0.18",
+    "jsr:@std/async@^1.4.0": "1.4.0",
     "jsr:@std/bytes@1": "1.0.4",
     "jsr:@std/crypto@1": "1.0.3",
+    "jsr:@std/data-structures@^1.1.0": "1.1.0",
     "jsr:@std/encoding@1": "1.0.6",
     "jsr:@std/encoding@^1.0.5": "1.0.6",
     "jsr:@std/fmt@1": "1.0.9",
@@ -20,6 +22,9 @@
     "jsr:@std/path@1": "1.1.4",
     "jsr:@std/path@1.0": "1.0.9",
     "jsr:@std/path@^1.1.1": "1.1.4",
+    "jsr:@std/testing@1": "1.0.19",
+    "npm:@opensearch-project/opensearch@3": "3.6.0",
+    "npm:@opensearch-project/opensearch@^3.6.0": "3.6.0",
     "npm:@skyware/jetstream@~0.2.5": "0.2.5",
     "npm:path-to-regexp@^6.3.0": "6.3.0"
   },
@@ -69,12 +74,18 @@
         "jsr:@std/internal@^1.0.12"
       ]
     },
+    "@std/async@1.4.0": {
+      "integrity": "4d70b008634f571cff9b554090d628c76141c32613aae0ff283fd5fa23d0c379"
+    },
     "@std/bytes@1.0.4": {
       "integrity": "11a0debe522707c95c7b7ef89b478c13fb1583a7cfb9a85674cd2cc2e3a28abc"
     },
     "@std/crypto@1.0.3": {
       "integrity": "a2a32f51ddef632d299e3879cd027c630dcd4d1d9a5285d6e6788072f4e51e7f"
     },
+    "@std/data-structures@1.1.0": {
+      "integrity": "c35ae4ad5d8e41a38573c2fe3e19b18ea2505f63cfea201edcb8720aca1f7f58"
+    },
     "@std/encoding@1.0.6": {
       "integrity": "ca87122c196e8831737d9547acf001766618e78cd8c33920776c7f5885546069"
     },
@@ -108,6 +119,13 @@
       "dependencies": [
         "jsr:@std/internal@^1.0.12"
       ]
+    },
+    "@std/testing@1.0.19": {
+      "integrity": "f4236172365b216728dc3cc8b5e80a9f4c33083d1e4ede7613d5b25b4014898e",
+      "dependencies": [
+        "jsr:@std/async",
+        "jsr:@std/data-structures"
+      ]
     }
   },
   "npm": {
@@ -142,6 +160,17 @@
         "unicode-segmenter"
       ]
     },
+    "@opensearch-project/opensearch@3.6.0": {
+      "integrity": "sha512-ow1A/Z7MBlNL/JZzTY4+M+8psiVh66g0Z7EQSl/qbT0U6zLBsVDYh6nvm8D4e4tIzQtn6ODbA2OsqYGvqZj76g==",
+      "dependencies": [
+        "aws4",
+        "debug",
+        "hpagent",
+        "json11",
+        "ms",
+        "secure-json-parse"
+      ]
+    },
     "@skyware/jetstream@0.2.5": {
       "integrity": "sha512-fM/zs03DLwqRyzZZJFWN20e76KrdqIp97Tlm8Cek+vxn96+tu5d/fx79V6H85L0QN6HvGiX2l9A8hWFqHvYlOA==",
       "dependencies": [
@@ -155,12 +184,31 @@
     "@standard-schema/spec@1.1.0": {
       "integrity": "sha512-l2aFy5jALhniG5HgqrD6jXLi/rUWrKvqN/qJx6yoJsgKhblVd+iqqU4RCXavm/jPityDo5TCvKMnpjKnOriy0w=="
     },
+    "aws4@1.13.2": {
+      "integrity": "sha512-lHe62zvbTB5eEABUVi/AwVh0ZKY9rMMDhmm+eeyuuUQbQ3+J+fONVQOZyj+DdrvD4BY33uYniyRJ4UJIaSKAfw=="
+    },
+    "debug@4.4.3": {
+      "integrity": "sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==",
+      "dependencies": [
+        "ms"
+      ]
+    },
     "esm-env@1.2.2": {
       "integrity": "sha512-Epxrv+Nr/CaL4ZcFGPJIYLWFom+YeV1DqMLHJoEd9SYRxNbaFruBwfEX/kkHUJf55j2+TUbmDcmuilbP1TmXHA=="
     },
     "event-target-polyfill@0.0.4": {
       "integrity": "sha512-Gs6RLjzlLRdT8X9ZipJdIZI/Y6/HhRLyq9RdDlCsnpxr/+Nn6bU2EFGuC94GjxqhM+Nmij2Vcq98yoHrU8uNFQ=="
     },
+    "hpagent@1.2.0": {
+      "integrity": "sha512-A91dYTeIB6NoXG+PxTQpCCDDnfHsW9kc06Lvpu1TEe9gnd6ZFeiBoRO9JvzEv6xK7EX97/dUE8g/vBMTqTS3CA=="
+    },
+    "json11@2.0.2": {
+      "integrity": "sha512-HIrd50UPYmP6sqLuLbFVm75g16o0oZrVfxrsY0EEys22klz8mRoWlX9KAEDOSOR9Q34rcxsyC8oDveGrCz5uLQ==",
+      "bin": true
+    },
+    "ms@2.1.3": {
+      "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA=="
+    },
     "partysocket@1.1.12": {
       "integrity": "sha512-079YDW1QZsFwJ8syLmr9xFMbE+VwbK4SpcKSOPVApX8rpIMennkxx0MeWl4oWutP/Zjgy8TMnQZ3FOXTQvu3DA==",
       "dependencies": [
@@ -170,6 +218,9 @@
     "path-to-regexp@6.3.0": {
       "integrity": "sha512-Yhpw4T9C6hPpgPeA28us07OJeqZ5EzQTkbfwuhsUg0c237RomFoETJgmp2sa3F/41gfLE6G5cqcYwznmeEeOlQ=="
     },
+    "secure-json-parse@2.7.0": {
+      "integrity": "sha512-6aU+Rwsezw7VR8/nyvKTx8QpWH9FrcYiXXlqC4z5d5XQBDRqtbfsRjnwGyqbi3gddNtWHuEk9OANUotL26qKUw=="
+    },
     "tiny-emitter@2.1.0": {
       "integrity": "sha512-NB6Dk1A9xgQPMoGqC5CVXn123gWyte215ONT5Pp5a0yt4nlEoO1ZWeCwpncaekPHXO60i47ihFnZPiRPjRMq4Q=="
     },
@@ -196,6 +247,8 @@
       "jsr:@db/sqlite@0.13",
       "jsr:@oak/oak@^17.2.0",
       "jsr:@std/assert@1",
+      "jsr:@std/testing@1",
+      "npm:@opensearch-project/opensearch@^3.6.0",
       "npm:@skyware/jetstream@~0.2.5"
     ]
   }

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:

docs/Remanso Jetstream API/admin/List every webhook subscription.bru

@@ -0,0 +1,114 @@
+meta {
+  name: List every webhook subscription
+  type: http
+  seq: 1
+}
+
+get {
+  url: {{baseUrl}}/admin/webhooks
+  body: none
+  auth: bearer
+}
+
+auth:bearer {
+  token: {{token}}
+}
+
+example {
+  name: 200 Response
+  description: All subscriptions across all DIDs.
+  
+  request: {
+    url: {{baseUrl}}/admin/webhooks
+    method: GET
+    mode: none
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 200
+      text: OK
+    }
+  
+    body: {
+      type: json
+      content: '''
+        [
+          {
+            "id": 0,
+            "did": "",
+            "method": "",
+            "url": "",
+            "verb": ""
+          }
+        ]
+      '''
+    }
+  }
+}
+
+example {
+  name: 401 Response
+  description: Missing or invalid bearer token.
+  
+  request: {
+    url: {{baseUrl}}/admin/webhooks
+    method: GET
+    mode: none
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 401
+      text: Unauthorized
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}
+
+example {
+  name: 403 Response
+  description: Token is valid but the caller is not authorized for this resource.
+  
+  request: {
+    url: {{baseUrl}}/admin/webhooks
+    method: GET
+    mode: none
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 403
+      text: Forbidden
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}

docs/Remanso Jetstream API/admin/folder.bru

@@ -0,0 +1,7 @@
+meta {
+  name: admin
+}
+
+auth {
+  mode: inherit
+}

docs/Remanso Jetstream API/auth/GitHub OAuth proxy.bru

@@ -0,0 +1,84 @@
+meta {
+  name: GitHub OAuth proxy
+  type: http
+  seq: 1
+}
+
+get {
+  url: {{baseUrl}}/auth/github
+  body: none
+  auth: inherit
+}
+
+params:query {
+  code: 
+  ~type: 
+}
+
+example {
+  name: 200 Response
+  description: GitHub access-token response (shape determined by GitHub).
+  
+  request: {
+    url: {{baseUrl}}/auth/github
+    method: GET
+    mode: none
+    params:query: {
+      code: 
+      ~type: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 200
+      text: OK
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {}
+      '''
+    }
+  }
+}
+
+example {
+  name: 400 Response
+  description: Malformed input.
+  
+  request: {
+    url: {{baseUrl}}/auth/github
+    method: GET
+    mode: none
+    params:query: {
+      code: 
+      ~type: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 400
+      text: Bad Request
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}

docs/Remanso Jetstream API/auth/folder.bru

@@ -0,0 +1,7 @@
+meta {
+  name: auth
+}
+
+auth {
+  mode: inherit
+}

docs/Remanso Jetstream API/bruno.json

@@ -0,0 +1,9 @@
+{
+  "version": "1",
+  "name": "Remanso Jetstream API",
+  "type": "collection",
+  "ignore": [
+    "node_modules",
+    ".git"
+  ]
+}

docs/Remanso Jetstream API/collection.bru

@@ -0,0 +1,7 @@
+meta {
+  name: Remanso Jetstream API
+}
+
+auth {
+  mode: none
+}

docs/Remanso Jetstream API/environments/Local development.bru

@@ -0,0 +1,3 @@
+vars {
+  baseUrl: http://localhost:8080
+}

docs/Remanso Jetstream API/environments/Production.bru

@@ -0,0 +1,3 @@
+vars {
+  baseUrl: https://api.remanso.space
+}

docs/Remanso Jetstream API/meta/Health probe.bru

@@ -0,0 +1,42 @@
+meta {
+  name: Health probe
+  type: http
+  seq: 2
+}
+
+get {
+  url: {{baseUrl}}/health
+  body: none
+  auth: inherit
+}
+
+example {
+  name: 200 Response
+  description: Service is up
+  
+  request: {
+    url: {{baseUrl}}/health
+    method: GET
+    mode: none
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 200
+      text: OK
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "status": ""
+        }
+      '''
+    }
+  }
+}

docs/Remanso Jetstream API/meta/Hello world.bru

@@ -0,0 +1,40 @@
+meta {
+  name: Hello world
+  type: http
+  seq: 1
+}
+
+get {
+  url: {{baseUrl}}/
+  body: none
+  auth: inherit
+}
+
+example {
+  name: 200 Response
+  description: Plain text greeting
+  
+  request: {
+    url: {{baseUrl}}/
+    method: GET
+    mode: none
+  }
+  
+  response: {
+    headers: {
+      Content-Type: text/plain
+    }
+  
+    status: {
+      code: 200
+      text: OK
+    }
+  
+    body: {
+      type: text
+      content: '''
+        Hello world
+      '''
+    }
+  }
+}

docs/Remanso Jetstream API/meta/folder.bru

@@ -0,0 +1,7 @@
+meta {
+  name: meta
+}
+
+auth {
+  mode: inherit
+}

docs/Remanso Jetstream API/notes/List discoverable notes across all users.bru

@@ -0,0 +1,61 @@
+meta {
+  name: List discoverable notes across all users
+  type: http
+  seq: 1
+}
+
+get {
+  url: {{baseUrl}}/notes
+  body: none
+  auth: inherit
+}
+
+params:query {
+  ~cursor: 
+  ~limit: 
+}
+
+example {
+  name: 200 Response
+  description: A page of notes ordered by rkey descending.
+  
+  request: {
+    url: {{baseUrl}}/notes
+    method: GET
+    mode: none
+    params:query: {
+      ~cursor: 
+      ~limit: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 200
+      text: OK
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "notes": [
+            {
+              "did": "",
+              "rkey": "",
+              "title": "",
+              "publishedAt": "",
+              "createdAt": "",
+              "language": ""
+            }
+          ],
+          "cursor": ""
+        }
+      '''
+    }
+  }
+}

docs/Remanso Jetstream API/notes/List discoverable notes for a single DID.bru

@@ -0,0 +1,69 @@
+meta {
+  name: List discoverable notes for a single DID
+  type: http
+  seq: 2
+}
+
+get {
+  url: {{baseUrl}}/:did/notes
+  body: none
+  auth: inherit
+}
+
+params:query {
+  ~cursor: 
+  ~limit: 
+}
+
+params:path {
+  did: 
+}
+
+example {
+  name: 200 Response
+  description: A page of notes for the given DID.
+  
+  request: {
+    url: {{baseUrl}}/:did/notes
+    method: GET
+    mode: none
+    params:query: {
+      ~cursor: 
+      ~limit: 
+    }
+  
+    params:path: {
+      did: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 200
+      text: OK
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "notes": [
+            {
+              "did": "",
+              "rkey": "",
+              "title": "",
+              "publishedAt": "",
+              "createdAt": "",
+              "language": ""
+            }
+          ],
+          "cursor": ""
+        }
+      '''
+    }
+  }
+}

docs/Remanso Jetstream API/notes/Multi-DID feed.bru

@@ -0,0 +1,105 @@
+meta {
+  name: Multi-DID feed
+  type: http
+  seq: 3
+}
+
+post {
+  url: {{baseUrl}}/notes/feed
+  body: json
+  auth: inherit
+}
+
+body:json {
+  {
+    "dids": [],
+    "cursor": "",
+    "limit": 0
+  }
+}
+
+example {
+  name: 200 Response
+  description: A page of notes from the requested DIDs.
+  
+  request: {
+    url: {{baseUrl}}/notes/feed
+    method: POST
+    mode: json
+    body:json: {
+      {
+        "dids": [],
+        "cursor": "",
+        "limit": 0
+      }
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 200
+      text: OK
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "notes": [
+            {
+              "did": "",
+              "rkey": "",
+              "title": "",
+              "publishedAt": "",
+              "createdAt": "",
+              "language": ""
+            }
+          ],
+          "cursor": ""
+        }
+      '''
+    }
+  }
+}
+
+example {
+  name: 400 Response
+  description: Malformed input.
+  
+  request: {
+    url: {{baseUrl}}/notes/feed
+    method: POST
+    mode: json
+    body:json: {
+      {
+        "dids": [],
+        "cursor": "",
+        "limit": 0
+      }
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 400
+      text: Bad Request
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}

docs/Remanso Jetstream API/notes/folder.bru

@@ -0,0 +1,7 @@
+meta {
+  name: notes
+}
+
+auth {
+  mode: inherit
+}

docs/Remanso Jetstream API/search/Owner-scoped full-text fuzzy search.bru

@@ -0,0 +1,195 @@
+meta {
+  name: Owner-scoped full-text fuzzy search
+  type: http
+  seq: 2
+}
+
+get {
+  url: {{baseUrl}}/:did/search
+  body: none
+  auth: bearer
+}
+
+params:query {
+  q: 
+  ~cursor: 
+  ~limit: 
+}
+
+params:path {
+  did: did:plc:4m3kouplb7s7xozjd3whinvl
+}
+
+auth:bearer {
+  token: {{token}}
+}
+
+example {
+  name: 200 Response
+  description: A page of search hits for the caller's DID.
+  
+  request: {
+    url: {{baseUrl}}/:did/search
+    method: GET
+    mode: none
+    params:query: {
+      q: 
+      ~cursor: 
+      ~limit: 
+    }
+  
+    params:path: {
+      did: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 200
+      text: OK
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "results": [
+            {
+              "did": "",
+              "rkey": "",
+              "title": "",
+              "snippet": "",
+              "score": 0,
+              "publishedAt": ""
+            }
+          ],
+          "cursor": ""
+        }
+      '''
+    }
+  }
+}
+
+example {
+  name: 400 Response
+  description: Malformed input.
+  
+  request: {
+    url: {{baseUrl}}/:did/search
+    method: GET
+    mode: none
+    params:query: {
+      q: 
+      ~cursor: 
+      ~limit: 
+    }
+  
+    params:path: {
+      did: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 400
+      text: Bad Request
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}
+
+example {
+  name: 401 Response
+  description: Missing or invalid bearer token.
+  
+  request: {
+    url: {{baseUrl}}/:did/search
+    method: GET
+    mode: none
+    params:query: {
+      q: 
+      ~cursor: 
+      ~limit: 
+    }
+  
+    params:path: {
+      did: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 401
+      text: Unauthorized
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}
+
+example {
+  name: 403 Response
+  description: Token is valid but the caller is not authorized for this resource.
+  
+  request: {
+    url: {{baseUrl}}/:did/search
+    method: GET
+    mode: none
+    params:query: {
+      q: 
+      ~cursor: 
+      ~limit: 
+    }
+  
+    params:path: {
+      did: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 403
+      text: Forbidden
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}

docs/Remanso Jetstream API/search/Public full-text fuzzy search.bru

@@ -0,0 +1,99 @@
+meta {
+  name: Public full-text fuzzy search
+  type: http
+  seq: 1
+}
+
+get {
+  url: {{baseUrl}}/search?q=
+  body: none
+  auth: inherit
+}
+
+params:query {
+  q: 
+  ~cursor: 
+  ~limit: 
+}
+
+example {
+  name: 200 Response
+  description: A page of search hits.
+  
+  request: {
+    url: {{baseUrl}}/search
+    method: GET
+    mode: none
+    params:query: {
+      q: 
+      ~cursor: 
+      ~limit: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 200
+      text: OK
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "results": [
+            {
+              "did": "",
+              "rkey": "",
+              "title": "",
+              "snippet": "",
+              "score": 0,
+              "publishedAt": ""
+            }
+          ],
+          "cursor": ""
+        }
+      '''
+    }
+  }
+}
+
+example {
+  name: 400 Response
+  description: Malformed input.
+  
+  request: {
+    url: {{baseUrl}}/search
+    method: GET
+    mode: none
+    params:query: {
+      q: 
+      ~cursor: 
+      ~limit: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 400
+      text: Bad Request
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}

docs/Remanso Jetstream API/search/folder.bru

@@ -0,0 +1,7 @@
+meta {
+  name: search
+}
+
+auth {
+  mode: inherit
+}

docs/Remanso Jetstream API/webhooks/Add one or more webhook subscriptions.bru

@@ -0,0 +1,206 @@
+meta {
+  name: Add one or more webhook subscriptions
+  type: http
+  seq: 2
+}
+
+post {
+  url: {{baseUrl}}/:did/webhooks
+  body: json
+  auth: bearer
+}
+
+params:path {
+  did: 
+}
+
+auth:bearer {
+  token: {{token}}
+}
+
+body:json {
+  {
+    "method": "",
+    "url": "",
+    "token": "",
+    "verb": ""
+  }
+}
+
+example {
+  name: 201 Response
+  description: One subscription per requested verb.
+  
+  request: {
+    url: {{baseUrl}}/:did/webhooks
+    method: POST
+    mode: json
+    params:path: {
+      did: 
+    }
+  
+    body:json: {
+      {
+        "method": "",
+        "url": "",
+        "token": "",
+        "verb": ""
+      }
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 201
+      text: Created
+    }
+  
+    body: {
+      type: json
+      content: '''
+        [
+          {
+            "id": 0,
+            "did": "",
+            "method": "",
+            "url": "",
+            "verb": ""
+          }
+        ]
+      '''
+    }
+  }
+}
+
+example {
+  name: 400 Response
+  description: Malformed input.
+  
+  request: {
+    url: {{baseUrl}}/:did/webhooks
+    method: POST
+    mode: json
+    params:path: {
+      did: 
+    }
+  
+    body:json: {
+      {
+        "method": "",
+        "url": "",
+        "token": "",
+        "verb": ""
+      }
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 400
+      text: Bad Request
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}
+
+example {
+  name: 401 Response
+  description: Missing or invalid bearer token.
+  
+  request: {
+    url: {{baseUrl}}/:did/webhooks
+    method: POST
+    mode: json
+    params:path: {
+      did: 
+    }
+  
+    body:json: {
+      {
+        "method": "",
+        "url": "",
+        "token": "",
+        "verb": ""
+      }
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 401
+      text: Unauthorized
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}
+
+example {
+  name: 403 Response
+  description: Token is valid but the caller is not authorized for this resource.
+  
+  request: {
+    url: {{baseUrl}}/:did/webhooks
+    method: POST
+    mode: json
+    params:path: {
+      did: 
+    }
+  
+    body:json: {
+      {
+        "method": "",
+        "url": "",
+        "token": "",
+        "verb": ""
+      }
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 403
+      text: Forbidden
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}

docs/Remanso Jetstream API/webhooks/Delete a single webhook subscription by id.bru

@@ -0,0 +1,189 @@
+meta {
+  name: Delete a single webhook subscription by id
+  type: http
+  seq: 4
+}
+
+delete {
+  url: {{baseUrl}}/:did/webhooks/:id
+  body: none
+  auth: bearer
+}
+
+params:path {
+  did: 
+  id: 
+}
+
+auth:bearer {
+  token: {{token}}
+}
+
+example {
+  name: 204 Response
+  description: Subscription deleted.
+  
+  request: {
+    url: {{baseUrl}}/:did/webhooks/:id
+    method: DELETE
+    mode: none
+    params:path: {
+      did: 
+      id: 
+    }
+  }
+  
+  response: {
+    status: {
+      code: 204
+      text: No Content
+    }
+  
+    body: {
+      type: text
+      content: '''
+  
+      '''
+    }
+  }
+}
+
+example {
+  name: 400 Response
+  description: Malformed input.
+  
+  request: {
+    url: {{baseUrl}}/:did/webhooks/:id
+    method: DELETE
+    mode: none
+    params:path: {
+      did: 
+      id: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 400
+      text: Bad Request
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}
+
+example {
+  name: 401 Response
+  description: Missing or invalid bearer token.
+  
+  request: {
+    url: {{baseUrl}}/:did/webhooks/:id
+    method: DELETE
+    mode: none
+    params:path: {
+      did: 
+      id: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 401
+      text: Unauthorized
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}
+
+example {
+  name: 403 Response
+  description: Token is valid but the caller is not authorized for this resource.
+  
+  request: {
+    url: {{baseUrl}}/:did/webhooks/:id
+    method: DELETE
+    mode: none
+    params:path: {
+      did: 
+      id: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 403
+      text: Forbidden
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}
+
+example {
+  name: 404 Response
+  description: No subscription with that id owned by this DID.
+  
+  request: {
+    url: {{baseUrl}}/:did/webhooks/:id
+    method: DELETE
+    mode: none
+    params:path: {
+      did: 
+      id: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 404
+      text: Not Found
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}

docs/Remanso Jetstream API/webhooks/Delete every webhook subscription for the caller.bru

@@ -0,0 +1,115 @@
+meta {
+  name: Delete every webhook subscription for the caller
+  type: http
+  seq: 3
+}
+
+delete {
+  url: {{baseUrl}}/:did/webhooks
+  body: none
+  auth: bearer
+}
+
+params:path {
+  did: 
+}
+
+auth:bearer {
+  token: {{token}}
+}
+
+example {
+  name: 204 Response
+  description: All subscriptions deleted.
+  
+  request: {
+    url: {{baseUrl}}/:did/webhooks
+    method: DELETE
+    mode: none
+    params:path: {
+      did: 
+    }
+  }
+  
+  response: {
+    status: {
+      code: 204
+      text: No Content
+    }
+  
+    body: {
+      type: text
+      content: '''
+  
+      '''
+    }
+  }
+}
+
+example {
+  name: 401 Response
+  description: Missing or invalid bearer token.
+  
+  request: {
+    url: {{baseUrl}}/:did/webhooks
+    method: DELETE
+    mode: none
+    params:path: {
+      did: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 401
+      text: Unauthorized
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}
+
+example {
+  name: 403 Response
+  description: Token is valid but the caller is not authorized for this resource.
+  
+  request: {
+    url: {{baseUrl}}/:did/webhooks
+    method: DELETE
+    mode: none
+    params:path: {
+      did: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 403
+      text: Forbidden
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}

docs/Remanso Jetstream API/webhooks/List a DID's webhook subscriptions.bru

@@ -0,0 +1,127 @@
+meta {
+  name: List a DID's webhook subscriptions
+  type: http
+  seq: 1
+}
+
+get {
+  url: {{baseUrl}}/:did/webhooks
+  body: none
+  auth: bearer
+}
+
+params:path {
+  did: did:plc:4m3kouplb7s7xozjd3whinvl
+}
+
+auth:bearer {
+  token: {{token}}
+}
+
+example {
+  name: 200 Response
+  description: All subscriptions owned by the caller.
+  
+  request: {
+    url: {{baseUrl}}/:did/webhooks
+    method: GET
+    mode: none
+    params:path: {
+      did: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 200
+      text: OK
+    }
+  
+    body: {
+      type: json
+      content: '''
+        [
+          {
+            "id": 0,
+            "did": "",
+            "method": "",
+            "url": "",
+            "verb": ""
+          }
+        ]
+      '''
+    }
+  }
+}
+
+example {
+  name: 401 Response
+  description: Missing or invalid bearer token.
+  
+  request: {
+    url: {{baseUrl}}/:did/webhooks
+    method: GET
+    mode: none
+    params:path: {
+      did: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 401
+      text: Unauthorized
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}
+
+example {
+  name: 403 Response
+  description: Token is valid but the caller is not authorized for this resource.
+  
+  request: {
+    url: {{baseUrl}}/:did/webhooks
+    method: GET
+    mode: none
+    params:path: {
+      did: 
+    }
+  }
+  
+  response: {
+    headers: {
+      Content-Type: application/json
+    }
+  
+    status: {
+      code: 403
+      text: Forbidden
+    }
+  
+    body: {
+      type: json
+      content: '''
+        {
+          "error": ""
+        }
+      '''
+    }
+  }
+}

docs/Remanso Jetstream API/webhooks/folder.bru

@@ -0,0 +1,7 @@
+meta {
+  name: webhooks
+}
+
+auth {
+  mode: inherit
+}

jetstream.ts

@@ -5,114 +5,53 @@ import {
   getWebhooksByDidAndVerb,
   saveCursor,
   upsertNote,
-  type WebhookVerb,
 } 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";
+import { dispatchAll, fireWebhooks } from "./src/jetstream/webhooks.ts";
+import { type BulkDeps, queueBulkCreate } from "./src/jetstream/bulk.ts";
 
-globalThis.addEventListener("unhandledrejection", (e) => {
+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);
+  }
+};
+
+const bulkDeps: BulkDeps = {
+  getSubs: getWebhooksByDidAndVerb,
+  dispatch: dispatchAll,
+};
+
+if (import.meta.main) {
+  globalThis.addEventListener("unhandledrejection", (e) => {
     log("[jetstream] unhandled rejection:", e.reason);
-});
+  });
 
-globalThis.addEventListener("error", (e) => {
+  globalThis.addEventListener("error", (e) => {
     log("[jetstream] uncaught error:", e.error);
-});
-
-type WebhookTarget = {
-  method: string;
-  url: string;
-  token?: string;
-};
-
-const dispatchAll = async (
-  webhooks: WebhookTarget[],
-  payload: Record<string, unknown>,
-  label: string,
-): Promise<void> => {
-  if (webhooks.length === 0) return;
-  const results = await Promise.allSettled(
-    webhooks.map(({ method, url, token }) => {
-      const hasBody = method !== "GET" && method !== "HEAD";
-      return fetch(url, {
-        method,
-        headers: {
-          ...(hasBody ? { "Content-Type": "application/json" } : {}),
-          ...(token ? { Authorization: `Bearer ${token}` } : {}),
-        },
-        body: hasBody ? JSON.stringify(payload) : undefined,
   });
-    }),
-  );
-  for (const result of results) {
-    if (result.status === "rejected") {
-      log(`[jetstream] ${label} webhook error:`, result.reason);
-    }
-  }
-};
 
-const fireWebhooks = async (
-  did: string,
-  verb: WebhookVerb,
-  payload: Record<string, unknown>,
-): Promise<void> => {
-  const webhooks = getWebhooksByDidAndVerb(did, verb);
-  await dispatchAll(webhooks, payload, `${verb} ${did}`);
-};
+  const cursor = getCursor();
+  log(`[jetstream] starting with cursor: ${cursor ?? "none"}`);
 
-const BULK_CREATE_DEBOUNCE_MS = 500;
-
-type BulkBuffer = {
-  records: Record<string, unknown>[];
-  timer: number;
-};
-const bulkBuffers = new Map<string, BulkBuffer>();
-
-const flushBulkCreate = async (did: string): Promise<void> => {
-  const buffer = bulkBuffers.get(did);
-  if (!buffer) return;
-  bulkBuffers.delete(did);
-  const webhooks = getWebhooksByDidAndVerb(did, "bulk-create");
-  log(`[jetstream] bulk-create ${did}: ${buffer.records.length} record(s)`);
-  await dispatchAll(
-    webhooks,
-    { event: "bulk-create", did, records: buffer.records },
-    `bulk-create ${did}`,
-  );
-};
-
-// Buffered records are not persisted: if jetstream restarts mid-window, those
-// `bulk-create` notifications are lost. Subscribers reconcile on cold start
-// because the underlying notes are already saved to the `note` table.
-const queueBulkCreate = (
-  did: string,
-  record: Record<string, unknown>,
-): void => {
-  const existing = bulkBuffers.get(did);
-  if (existing) {
-    clearTimeout(existing.timer);
-    existing.records.push(record);
-    existing.timer = setTimeout(
-      () => flushBulkCreate(did),
-      BULK_CREATE_DEBOUNCE_MS,
-    );
-    return;
-  }
-  bulkBuffers.set(did, {
-    records: [record],
-    timer: setTimeout(() => flushBulkCreate(did), BULK_CREATE_DEBOUNCE_MS),
-  });
-};
-
-const cursor = getCursor();
-log(`[jetstream] starting with cursor: ${cursor ?? "none"}`);
-
-const jetstream = new Jetstream({
+  const jetstream = new Jetstream({
     wantedCollections: ["space.remanso.note"],
     cursor: cursor ? Number(cursor) : undefined,
     endpoint: "https://jetstream2.fr.hose.cam/subscribe",
-});
+  });
 
-jetstream.onCreate("space.remanso.note", async (event) => {
+  jetstream.onCreate("space.remanso.note", async (event) => {
     try {
       const {
         did,
@@ -122,14 +61,22 @@ 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 });
-    queueBulkCreate(did, { rkey, ...note });
+      await Promise.allSettled([
+        fireWebhooks(
+          did,
+          "create",
+          { event: "create", did, rkey, ...note },
+          getWebhooksByDidAndVerb,
+        ),
+        safeIndex({ did, rkey, ...note }),
+      ]);
+      queueBulkCreate(did, { rkey, ...note }, bulkDeps);
     } catch (error) {
       log(`[jetstream] error on create:`, error);
     }
-});
+  });
 
-jetstream.onUpdate("space.remanso.note", async (event) => {
+  jetstream.onUpdate("space.remanso.note", async (event) => {
     try {
       const {
         did,
@@ -140,14 +87,22 @@ 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 });
-    queueBulkCreate(did, { rkey, ...note });
+      await Promise.allSettled([
+        fireWebhooks(
+          did,
+          "create",
+          { event: "create", did, rkey, ...note },
+          getWebhooksByDidAndVerb,
+        ),
+        safeIndex({ did, rkey, ...note }),
+      ]);
+      queueBulkCreate(did, { rkey, ...note }, bulkDeps);
     } catch (error) {
       log(`[jetstream] error on update:`, error);
     }
-});
+  });
 
-jetstream.onDelete("space.remanso.note", async (event) => {
+  jetstream.onDelete("space.remanso.note", async (event) => {
     try {
       const {
         did,
@@ -156,29 +111,44 @@ 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 },
+          getWebhooksByDidAndVerb,
+        ),
+        safeRemove(did, rkey),
+      ]);
     } catch (error) {
       log(`[jetstream] error on delete:`, error);
     }
-});
+  });
 
-jetstream.on("open", () => {
+  jetstream.on("open", () => {
     log("[jetstream] connected");
-});
+  });
 
-jetstream.on("close", () => {
+  jetstream.on("close", () => {
     log("[jetstream] connection closed");
-});
+  });
 
-jetstream.on("error", (error) => {
+  jetstream.on("error", (error) => {
     log("[jetstream] connection closed with error", error);
-});
+  });
 
-log("[jetstream] launching");
-jetstream.start();
+  try {
+    await ensureIndex();
+  } catch (error) {
+    log("[opensearch] ensureIndex failed at startup:", error);
+  }
 
-setInterval(() => {
+  log("[jetstream] launching");
+  jetstream.start();
+
+  setInterval(() => {
     if (jetstream.cursor) {
       saveCursor(jetstream.cursor);
     }
-}, 5000);
+  }, 5000);
+}

openapi.yaml

@@ -0,0 +1,449 @@
+openapi: 3.1.0
+info:
+  title: Remanso Jetstream API
+  version: 0.1.0
+  description: |
+    HTTP API for Remanso, a blogging platform on the AT Protocol. Exposes
+    paginated read access to `space.remanso.note` records, full-text fuzzy
+    search backed by OpenSearch, and per-user webhook subscriptions that fire
+    on note create/update/delete.
+
+    Authentication uses AT Protocol PDS-issued bearer JWTs. The token's `sub`
+    must equal the path `:did` for owner-scoped endpoints; admin endpoints
+    additionally require the verified DID to be in the server-side
+    `ADMIN_DIDS` allowlist.
+
+servers:
+  - url: https://api.remanso.space
+    description: Production
+  - url: http://localhost:8080
+    description: Local development
+
+tags:
+  - name: notes
+    description: Read access to stored notes
+  - name: search
+    description: Full-text fuzzy search via OpenSearch
+  - name: webhooks
+    description: Per-user delivery hooks for note events
+  - name: admin
+    description: Endpoints restricted to ADMIN_DIDS
+  - name: auth
+    description: OAuth helpers
+  - name: meta
+
+paths:
+  /:
+    get:
+      tags: [meta]
+      summary: Hello world
+      responses:
+        "200":
+          description: Plain text greeting
+          content:
+            text/plain:
+              schema: { type: string, example: "Hello world" }
+
+  /health:
+    get:
+      tags: [meta]
+      summary: Health probe
+      responses:
+        "200":
+          description: Service is up
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  status: { type: string, example: ok }
+
+  /auth/github:
+    get:
+      tags: [auth]
+      summary: GitHub OAuth proxy
+      description: |
+        Exchanges a GitHub OAuth `code` for a token, or refreshes an existing
+        one. The server adds its own client_id/client_secret and forwards to
+        `https://github.com/login/oauth/access_token`. Response body and
+        status are passed through from GitHub.
+      parameters:
+        - in: query
+          name: code
+          required: true
+          schema: { type: string }
+          description: GitHub OAuth code, or the refresh token when `type=refresh`.
+        - in: query
+          name: type
+          required: false
+          schema: { type: string, enum: [refresh] }
+          description: When `refresh`, performs a refresh-token grant instead.
+      responses:
+        "200":
+          description: GitHub access-token response (shape determined by GitHub).
+          content:
+            application/json:
+              schema: { type: object, additionalProperties: true }
+        "400":
+          $ref: "#/components/responses/BadRequest"
+
+  /notes:
+    get:
+      tags: [notes]
+      summary: List discoverable notes across all users
+      parameters:
+        - $ref: "#/components/parameters/Cursor"
+        - $ref: "#/components/parameters/Limit"
+      responses:
+        "200":
+          description: A page of notes ordered by rkey descending.
+          content:
+            application/json:
+              schema: { $ref: "#/components/schemas/NotePage" }
+
+  /{did}/notes:
+    get:
+      tags: [notes]
+      summary: List discoverable notes for a single DID
+      parameters:
+        - $ref: "#/components/parameters/Did"
+        - $ref: "#/components/parameters/Cursor"
+        - $ref: "#/components/parameters/Limit"
+      responses:
+        "200":
+          description: A page of notes for the given DID.
+          content:
+            application/json:
+              schema: { $ref: "#/components/schemas/NotePage" }
+
+  /notes/feed:
+    post:
+      tags: [notes]
+      summary: Multi-DID feed
+      description: Paginated discoverable notes across an arbitrary set of DIDs.
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              required: [dids]
+              properties:
+                dids:
+                  type: array
+                  minItems: 1
+                  items: { type: string, example: "did:plc:abc123..." }
+                cursor:
+                  type: string
+                  description: rkey returned as `cursor` from a previous page.
+                limit:
+                  type: integer
+                  minimum: 1
+                  default: 20
+      responses:
+        "200":
+          description: A page of notes from the requested DIDs.
+          content:
+            application/json:
+              schema: { $ref: "#/components/schemas/NotePage" }
+        "400":
+          $ref: "#/components/responses/BadRequest"
+
+  /search:
+    get:
+      tags: [search]
+      summary: Public full-text fuzzy search
+      description: |
+        Searches every discoverable note's title and content. Uses
+        OpenSearch `multi_match` with `fuzziness: AUTO`, so small typos still
+        match. Each hit includes a sentence-bounded `snippet` from the match
+        with query terms wrapped in `<em>` tags.
+      parameters:
+        - in: query
+          name: q
+          required: true
+          schema: { type: string, minLength: 1 }
+          description: Query string.
+        - in: query
+          name: cursor
+          required: false
+          schema: { type: string }
+          description: Opaque base64 cursor returned from a previous call.
+        - $ref: "#/components/parameters/Limit"
+      responses:
+        "200":
+          description: A page of search hits.
+          content:
+            application/json:
+              schema: { $ref: "#/components/schemas/SearchPage" }
+        "400":
+          $ref: "#/components/responses/BadRequest"
+
+  /{did}/search:
+    get:
+      tags: [search]
+      summary: Owner-scoped full-text fuzzy search
+      description: |
+        Same as `/search` but limited to a single DID and including
+        non-discoverable notes (drafts). Requires a bearer JWT whose `sub`
+        equals the path `:did`.
+      security:
+        - bearerAuth: []
+      parameters:
+        - $ref: "#/components/parameters/Did"
+        - in: query
+          name: q
+          required: true
+          schema: { type: string, minLength: 1 }
+        - in: query
+          name: cursor
+          required: false
+          schema: { type: string }
+        - $ref: "#/components/parameters/Limit"
+      responses:
+        "200":
+          description: A page of search hits for the caller's DID.
+          content:
+            application/json:
+              schema: { $ref: "#/components/schemas/SearchPage" }
+        "400":
+          $ref: "#/components/responses/BadRequest"
+        "401":
+          $ref: "#/components/responses/Unauthorized"
+        "403":
+          $ref: "#/components/responses/Forbidden"
+
+  /{did}/webhooks:
+    get:
+      tags: [webhooks]
+      summary: List a DID's webhook subscriptions
+      security:
+        - bearerAuth: []
+      parameters:
+        - $ref: "#/components/parameters/Did"
+      responses:
+        "200":
+          description: All subscriptions owned by the caller.
+          content:
+            application/json:
+              schema:
+                type: array
+                items: { $ref: "#/components/schemas/WebhookSubscription" }
+        "401": { $ref: "#/components/responses/Unauthorized" }
+        "403": { $ref: "#/components/responses/Forbidden" }
+
+    post:
+      tags: [webhooks]
+      summary: Add one or more webhook subscriptions
+      description: |
+        When `verb` is omitted, a pair of subscriptions is created — one for
+        `create`, one for `delete`. When `verb` is provided, only that
+        subscription is created. The response array contains every
+        subscription that was inserted.
+      security:
+        - bearerAuth: []
+      parameters:
+        - $ref: "#/components/parameters/Did"
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              required: [method, url]
+              properties:
+                method:
+                  type: string
+                  example: POST
+                url:
+                  type: string
+                  format: uri
+                token:
+                  type: string
+                  description: Optional bearer token sent to the webhook URL.
+                verb:
+                  type: string
+                  enum: [create, delete, bulk-create]
+      responses:
+        "201":
+          description: One subscription per requested verb.
+          content:
+            application/json:
+              schema:
+                type: array
+                items: { $ref: "#/components/schemas/WebhookSubscription" }
+        "400": { $ref: "#/components/responses/BadRequest" }
+        "401": { $ref: "#/components/responses/Unauthorized" }
+        "403": { $ref: "#/components/responses/Forbidden" }
+
+    delete:
+      tags: [webhooks]
+      summary: Delete every webhook subscription for the caller
+      security:
+        - bearerAuth: []
+      parameters:
+        - $ref: "#/components/parameters/Did"
+      responses:
+        "204": { description: All subscriptions deleted. }
+        "401": { $ref: "#/components/responses/Unauthorized" }
+        "403": { $ref: "#/components/responses/Forbidden" }
+
+  /{did}/webhooks/{id}:
+    delete:
+      tags: [webhooks]
+      summary: Delete a single webhook subscription by id
+      security:
+        - bearerAuth: []
+      parameters:
+        - $ref: "#/components/parameters/Did"
+        - in: path
+          name: id
+          required: true
+          schema: { type: integer, minimum: 1 }
+      responses:
+        "204": { description: Subscription deleted. }
+        "400": { $ref: "#/components/responses/BadRequest" }
+        "401": { $ref: "#/components/responses/Unauthorized" }
+        "403": { $ref: "#/components/responses/Forbidden" }
+        "404":
+          description: No subscription with that id owned by this DID.
+          content:
+            application/json:
+              schema: { $ref: "#/components/schemas/Error" }
+
+  /admin/webhooks:
+    get:
+      tags: [admin]
+      summary: List every webhook subscription
+      description: Requires the verified DID to be in `ADMIN_DIDS`.
+      security:
+        - bearerAuth: []
+      responses:
+        "200":
+          description: All subscriptions across all DIDs.
+          content:
+            application/json:
+              schema:
+                type: array
+                items: { $ref: "#/components/schemas/WebhookSubscription" }
+        "401": { $ref: "#/components/responses/Unauthorized" }
+        "403": { $ref: "#/components/responses/Forbidden" }
+
+components:
+  securitySchemes:
+    bearerAuth:
+      type: http
+      scheme: bearer
+      bearerFormat: JWT
+      description: |
+        AT Protocol PDS session JWT. The server resolves the DID from the
+        token's `sub`, looks up the PDS endpoint via plc.directory, and
+        verifies the session by calling `com.atproto.server.getSession`
+        against that PDS.
+
+  parameters:
+    Did:
+      in: path
+      name: did
+      required: true
+      schema: { type: string, example: "did:plc:abc123..." }
+      description: AT Protocol DID.
+    Cursor:
+      in: query
+      name: cursor
+      required: false
+      schema: { type: string }
+      description: rkey returned as `cursor` from the previous page.
+    Limit:
+      in: query
+      name: limit
+      required: false
+      schema: { type: integer, minimum: 1, default: 20 }
+      description: Maximum results per page. Search endpoints cap at 100.
+
+  responses:
+    BadRequest:
+      description: Malformed input.
+      content:
+        application/json:
+          schema: { $ref: "#/components/schemas/Error" }
+    Unauthorized:
+      description: Missing or invalid bearer token.
+      content:
+        application/json:
+          schema: { $ref: "#/components/schemas/Error" }
+    Forbidden:
+      description: Token is valid but the caller is not authorized for this resource.
+      content:
+        application/json:
+          schema: { $ref: "#/components/schemas/Error" }
+
+  schemas:
+    Error:
+      type: object
+      required: [error]
+      properties:
+        error: { type: string }
+
+    Note:
+      type: object
+      required: [did, rkey, title, publishedAt, createdAt]
+      properties:
+        did: { type: string }
+        rkey: { type: string }
+        title: { type: string }
+        publishedAt: { type: string, format: date-time }
+        createdAt: { type: string, format: date-time }
+        language:
+          type: string
+          description: ISO 639-3 code if known.
+
+    NotePage:
+      type: object
+      required: [notes]
+      properties:
+        notes:
+          type: array
+          items: { $ref: "#/components/schemas/Note" }
+        cursor:
+          type: string
+          nullable: true
+          description: Present only when more pages exist; pass back as `cursor`.
+
+    SearchHit:
+      type: object
+      required: [did, rkey, title, snippet, score]
+      properties:
+        did: { type: string }
+        rkey: { type: string }
+        title: { type: string }
+        snippet:
+          type: string
+          description: Sentence-bounded fragment with `<em>` around matches.
+        score: { type: number }
+        publishedAt: { type: string, format: date-time }
+
+    SearchPage:
+      type: object
+      required: [results]
+      properties:
+        results:
+          type: array
+          items: { $ref: "#/components/schemas/SearchHit" }
+        cursor:
+          type: string
+          nullable: true
+          description: Opaque base64 cursor; pass back as `cursor`.
+
+    WebhookSubscription:
+      type: object
+      required: [id, did, method, url, verb]
+      description: |
+        `token` is write-only — it is accepted by `POST /:did/webhooks` but
+        never returned in responses.
+      properties:
+        id: { type: integer, minimum: 1 }
+        did: { type: string }
+        method: { type: string, example: POST }
+        url: { type: string, format: uri }
+        verb: { type: string, enum: [create, delete, bulk-create] }

scripts/backfill-opensearch.ts

@@ -0,0 +1,123 @@
+// 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 { closeDb, listDistinctNoteDids } 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 dids = listDistinctNoteDids();
+  log(`[backfill] ${dids.length} DID(s) to process`);
+
+  let totalIndexed = 0;
+  let totalFailed = 0;
+  for (const did of dids) {
+    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}`);
+  closeDb();
+};
+
+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 = {
@@ -18,13 +19,25 @@ type AuthCtx = {
   response: { status: number; body: unknown };
 };
 
+let authenticator: typeof authenticateRequest = authenticateRequest;
+
+export const _setAuthenticatorForTest = (
+  fn: typeof authenticateRequest,
+): void => {
+  authenticator = fn;
+};
+
+export const _resetAuthenticatorForTest = (): void => {
+  authenticator = authenticateRequest;
+};
+
 const requireDidOwnership = async (
   ctx: AuthCtx,
   did: string,
 ): Promise<boolean> => {
   let verifiedDid: string;
   try {
-    verifiedDid = await authenticateRequest(
+    verifiedDid = await authenticator(
       ctx.request.headers.get("Authorization"),
     );
   } catch {
@@ -34,23 +47,24 @@ 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;
 };
 
-const ADMIN_DIDS = new Set(
+const adminDids = (): Set<string> =>
+  new Set(
     (Deno.env.get("ADMIN_DIDS") ?? "")
       .split(",")
       .map((d) => d.trim())
       .filter(Boolean),
-);
+  );
 
 const requireAdmin = async (ctx: AuthCtx): Promise<boolean> => {
   let verifiedDid: string;
   try {
-    verifiedDid = await authenticateRequest(
+    verifiedDid = await authenticator(
       ctx.request.headers.get("Authorization"),
     );
   } catch {
@@ -58,7 +72,7 @@ const requireAdmin = async (ctx: AuthCtx): Promise<boolean> => {
     ctx.response.body = { error: "Unauthorized" };
     return false;
   }
-  if (!ADMIN_DIDS.has(verifiedDid)) {
+  if (!adminDids().has(verifiedDid)) {
     ctx.response.status = 403;
     ctx.response.body = { error: "Admin only" };
     return false;
@@ -131,6 +145,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 ?? {};
@@ -222,9 +273,10 @@ router.delete("/:did/webhooks/:id", async (ctx) => {
 //   ctx.response.status = 204;
 // })
 
-const app = new Application();
+export const createApp = (): Application => {
+  const app = new Application();
 
-app.use(async (ctx, next) => {
+  app.use(async (ctx, next) => {
     ctx.response.headers.set("Access-Control-Allow-Origin", "*");
     ctx.response.headers.set(
       "Access-Control-Allow-Methods",
@@ -239,10 +291,16 @@ app.use(async (ctx, next) => {
       return;
     }
     await next();
-});
+  });
 
-app.use(router.routes());
-app.use(router.allowedMethods());
+  app.use(router.routes());
+  app.use(router.allowedMethods());
 
-log("[server] listening on port 8080");
-app.listen({ port: 8080 });
+  return app;
+};
+
+if (import.meta.main) {
+  const app = createApp();
+  log("[server] listening on port 8080");
+  app.listen({ port: 8080 });
+}

src/auth/verify.ts

@@ -7,14 +7,14 @@ interface DidDocument {
   service?: { id: string; serviceEndpoint: string }[];
 }
 
-function decodeJwtPayload(token: string): JwtPayload {
+export function decodeJwtPayload(token: string): JwtPayload {
   const parts = token.split(".");
   if (parts.length !== 3) throw new Error("Invalid JWT format");
   const payload = parts[1].replace(/-/g, "+").replace(/_/g, "/");
   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/db.ts

@@ -1,18 +1,39 @@
 import { Database } from "@db/sqlite";
 import type { Note } from "./note.ts";
 
-export const db = new Database(Deno.env.get("SQLITE_PATH") ?? "notes.db");
+let _db: Database | null = null;
 
-try {
-  db.exec("PRAGMA busy_timeout=10000");
-  db.exec("PRAGMA journal_mode=WAL");
-  const [row] = db.prepare("PRAGMA journal_mode").all<
+const openDefaultDb = (): Database => {
+  const handle = new Database(Deno.env.get("SQLITE_PATH") ?? "notes.db");
+  try {
+    handle.exec("PRAGMA busy_timeout=10000");
+    handle.exec("PRAGMA journal_mode=WAL");
+    const [row] = handle.prepare("PRAGMA journal_mode").all<
       { journal_mode: string }
     >();
     console.log(`[db] journal_mode=${row.journal_mode}, busy_timeout=10000`);
-} catch (e) {
+  } catch (e) {
     console.error("[db] failed to set PRAGMAs:", e);
-}
+  }
+  return handle;
+};
+
+const getDb = (): Database => _db ??= openDefaultDb();
+
+export const _setDbForTest = (handle: Database): void => {
+  _db = handle;
+};
+
+export const _resetDbForTest = (): void => {
+  _db = null;
+};
+
+export const closeDb = (): void => {
+  if (_db) {
+    _db.close();
+    _db = null;
+  }
+};
 
 type NoteRow = {
   did: string;
@@ -23,6 +44,7 @@ type NoteRow = {
 };
 
 export const getNotes = (cursor?: string, limit = 20) => {
+  const db = getDb();
   const notes = cursor
     ? db.prepare(
       "SELECT did, rkey, title, publishedAt, createdAt, language FROM note WHERE discoverable = 1 AND rkey < ? ORDER BY rkey DESC LIMIT ?",
@@ -38,6 +60,7 @@ export const getNotes = (cursor?: string, limit = 20) => {
 };
 
 export const getNotesByDid = (did: string, cursor?: string, limit = 20) => {
+  const db = getDb();
   const notes = cursor
     ? db.prepare(
       "SELECT did, rkey, title, publishedAt, createdAt, language FROM note WHERE discoverable = 1 AND did = ? AND rkey < ? ORDER BY rkey DESC LIMIT ?",
@@ -54,6 +77,7 @@ export const getNotesByDid = (did: string, cursor?: string, limit = 20) => {
 
 export const getNotesByDids = (dids: string[], cursor?: string, limit = 20) => {
   if (dids.length === 0) return { notes: [] };
+  const db = getDb();
   const placeholders = dids.map(() => "?").join(", ");
   const notes = cursor
     ? db.prepare(
@@ -70,18 +94,18 @@ export const getNotesByDids = (dids: string[], cursor?: string, limit = 20) => {
 };
 
 export const deleteNote = ({ did, rkey }: { did: string; rkey: string }) => {
-  db.exec("DELETE FROM note WHERE did = ? AND rkey = ?", did, rkey);
+  getDb().exec("DELETE FROM note WHERE did = ? AND rkey = ?", did, rkey);
 };
 
 export const getCursor = (): string | undefined => {
-  const row = db.prepare(
+  const row = getDb().prepare(
     "SELECT value FROM state WHERE key = 'cursor'",
   ).get<{ value: string }>();
   return row?.value;
 };
 
 export const saveCursor = (cursor: number) => {
-  db.exec(
+  getDb().exec(
     "INSERT OR REPLACE INTO state (key, value) VALUES ('cursor', ?)",
     String(cursor),
   );
@@ -101,6 +125,7 @@ type WebhookSubscriptionRow = {
 export const addWebhookSubscription = (
   { did, method, url, token, verb }: Omit<WebhookSubscriptionRow, "id">,
 ): WebhookSubscriptionRow => {
+  const db = getDb();
   db.exec(
     "INSERT INTO webhook_subscription (did, method, url, token, verb) VALUES (?, ?, ?, ?, ?)",
     did,
@@ -116,13 +141,13 @@ export const addWebhookSubscription = (
 };
 
 export const deleteWebhooksByDid = (did: string): void => {
-  db.exec("DELETE FROM webhook_subscription WHERE did = ?", did);
+  getDb().exec("DELETE FROM webhook_subscription WHERE did = ?", did);
 };
 
 export const deleteWebhookById = (
   { did, id }: { did: string; id: number },
 ): boolean => {
-  const result = db.prepare(
+  const result = getDb().prepare(
     "DELETE FROM webhook_subscription WHERE did = ? AND id = ?",
   ).run(did, id);
   return result > 0;
@@ -131,13 +156,13 @@ export const deleteWebhookById = (
 export const listWebhooksByDid = (
   did: string,
 ): Omit<WebhookSubscriptionRow, "token">[] => {
-  return db.prepare(
+  return getDb().prepare(
     "SELECT id, did, method, url, verb FROM webhook_subscription WHERE did = ? ORDER BY id DESC",
   ).all<Omit<WebhookSubscriptionRow, "token">>(did);
 };
 
 export const listAllWebhooks = (): Omit<WebhookSubscriptionRow, "token">[] => {
-  return db.prepare(
+  return getDb().prepare(
     "SELECT id, did, method, url, verb FROM webhook_subscription ORDER BY did, id",
   ).all<Omit<WebhookSubscriptionRow, "token">>();
 };
@@ -146,14 +171,20 @@ export const getWebhooksByDidAndVerb = (
   did: string,
   verb: WebhookVerb,
 ): WebhookSubscriptionRow[] => {
-  return db.prepare(
+  return getDb().prepare(
     "SELECT id, did, method, url, token, verb FROM webhook_subscription WHERE did = ? AND verb = ? ORDER BY id DESC LIMIT 10",
   ).all<WebhookSubscriptionRow>(did, verb);
 };
 
+export const listDistinctNoteDids = (): string[] => {
+  return getDb().prepare(
+    "SELECT DISTINCT did FROM note ORDER BY did",
+  ).all<{ did: string }>().map((row) => row.did);
+};
+
 export const upsertNote = (note: Note) => {
   const now = new Date().toISOString();
-  db.exec(
+  getDb().exec(
     `
     INSERT INTO note (
       title,

src/data/note.ts

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

src/jetstream/bulk.ts

@@ -0,0 +1,66 @@
+import { log } from "../log.ts";
+import type { GetSubsByVerb, WebhookTarget } from "./webhooks.ts";
+
+export const BULK_CREATE_DEBOUNCE_MS = 500;
+
+export type BulkDeps = {
+  getSubs: GetSubsByVerb;
+  dispatch: (
+    webhooks: WebhookTarget[],
+    payload: Record<string, unknown>,
+    label: string,
+  ) => Promise<void>;
+  debounceMs?: number;
+};
+
+type BulkBuffer = {
+  records: Record<string, unknown>[];
+  timer: ReturnType<typeof setTimeout>;
+};
+
+const bulkBuffers = new Map<string, BulkBuffer>();
+
+export const flushBulkCreate = async (
+  did: string,
+  deps: BulkDeps,
+): Promise<void> => {
+  const buffer = bulkBuffers.get(did);
+  if (!buffer) return;
+  bulkBuffers.delete(did);
+  const webhooks = deps.getSubs(did, "bulk-create");
+  log(`[jetstream] bulk-create ${did}: ${buffer.records.length} record(s)`);
+  await deps.dispatch(
+    webhooks,
+    { event: "bulk-create", did, records: buffer.records },
+    `bulk-create ${did}`,
+  );
+};
+
+// Buffered records are not persisted: if jetstream restarts mid-window, those
+// `bulk-create` notifications are lost. Subscribers reconcile on cold start
+// because the underlying notes are already saved to the `note` table.
+export const queueBulkCreate = (
+  did: string,
+  record: Record<string, unknown>,
+  deps: BulkDeps,
+): void => {
+  const ms = deps.debounceMs ?? BULK_CREATE_DEBOUNCE_MS;
+  const existing = bulkBuffers.get(did);
+  if (existing) {
+    clearTimeout(existing.timer);
+    existing.records.push(record);
+    existing.timer = setTimeout(() => flushBulkCreate(did, deps), ms);
+    return;
+  }
+  bulkBuffers.set(did, {
+    records: [record],
+    timer: setTimeout(() => flushBulkCreate(did, deps), ms),
+  });
+};
+
+export const _resetBulkBuffersForTest = (): void => {
+  for (const buffer of bulkBuffers.values()) {
+    clearTimeout(buffer.timer);
+  }
+  bulkBuffers.clear();
+};

src/jetstream/webhooks.ts

@@ -0,0 +1,49 @@
+import type { WebhookVerb } from "../data/db.ts";
+import { log } from "../log.ts";
+
+export type WebhookTarget = {
+  method: string;
+  url: string;
+  token?: string;
+};
+
+export const dispatchAll = async (
+  webhooks: WebhookTarget[],
+  payload: Record<string, unknown>,
+  label: string,
+): Promise<void> => {
+  if (webhooks.length === 0) return;
+  const results = await Promise.allSettled(
+    webhooks.map(({ method, url, token }) => {
+      const hasBody = method !== "GET" && method !== "HEAD";
+      return fetch(url, {
+        method,
+        headers: {
+          ...(hasBody ? { "Content-Type": "application/json" } : {}),
+          ...(token ? { Authorization: `Bearer ${token}` } : {}),
+        },
+        body: hasBody ? JSON.stringify(payload) : undefined,
+      });
+    }),
+  );
+  for (const result of results) {
+    if (result.status === "rejected") {
+      log(`[jetstream] ${label} webhook error:`, result.reason);
+    }
+  }
+};
+
+export type GetSubsByVerb = (
+  did: string,
+  verb: WebhookVerb,
+) => WebhookTarget[];
+
+export const fireWebhooks = async (
+  did: string,
+  verb: WebhookVerb,
+  payload: Record<string, unknown>,
+  getSubs: GetSubsByVerb,
+): Promise<void> => {
+  const webhooks = getSubs(did, verb);
+  await dispatchAll(webhooks, payload, `${verb} ${did}`);
+};

src/migrations/apply.ts

@@ -0,0 +1,83 @@
+import type { Database } from "@db/sqlite";
+
+export const applyMigrations = (db: Database): void => {
+  db.exec(`
+    CREATE TABLE IF NOT EXISTS note (
+      title TEXT NOT NULL,
+      publishedAt DATETIME,
+      createdAt DATETIME NOT NULL,
+      did TEXT NOT NULL,
+      rkey TEXT NOT NULL,
+      discoverable INTEGER NOT NULL DEFAULT 1,
+      PRIMARY KEY (did, rkey)
+    );
+  `);
+
+  try {
+    db.exec(
+      `ALTER TABLE note ADD COLUMN listed INTEGER NOT NULL DEFAULT 1;`,
+    );
+  } catch {
+    // Column already exists — no-op
+  }
+
+  try {
+    db.exec(
+      `ALTER TABLE note ADD COLUMN language STRING;`,
+    );
+  } catch {
+    // Column already exists — no-op
+  }
+
+  try {
+    db.exec(`ALTER TABLE note RENAME COLUMN listed TO discoverable;`);
+  } catch {
+    // Column already renamed or doesn't exist — no-op
+  }
+
+  db.exec(`
+    CREATE TABLE IF NOT EXISTS state (
+      key TEXT PRIMARY KEY,
+      value TEXT NOT NULL
+    );
+  `);
+
+  db.exec(`
+    CREATE TABLE IF NOT EXISTS webhook_subscription (
+      id     INTEGER PRIMARY KEY,
+      did    TEXT NOT NULL,
+      method TEXT NOT NULL,
+      url    TEXT NOT NULL
+    );
+  `);
+
+  db.exec(`
+    CREATE INDEX IF NOT EXISTS idx_webhook_subscription_did
+      ON webhook_subscription(did);
+  `);
+
+  try {
+    db.exec(`ALTER TABLE webhook_subscription ADD COLUMN token TEXT;`);
+  } catch {
+    // Column already exists — no-op
+  }
+
+  try {
+    db.exec(
+      `ALTER TABLE webhook_subscription ADD COLUMN verb TEXT NOT NULL DEFAULT 'create';`,
+    );
+    db.exec(`
+      INSERT INTO webhook_subscription (did, method, url, token, verb)
+      SELECT did, method, url, token, 'delete'
+      FROM webhook_subscription
+      WHERE verb = 'create';
+    `);
+  } catch {
+    // Column already exists — backfill already happened on a previous run.
+  }
+
+  db.exec(`
+    CREATE INDEX IF NOT EXISTS idx_webhook_subscription_did_verb
+      ON webhook_subscription(did, verb);
+  `);
+};

src/migrations/init.ts

@@ -1,83 +1,6 @@
-import { db } from "../data/db.ts";
-
-db.exec(`
-  CREATE TABLE IF NOT EXISTS note (
-    title TEXT NOT NULL,
-    publishedAt DATETIME,
-    createdAt DATETIME NOT NULL,
-    did TEXT NOT NULL,
-    rkey TEXT NOT NULL,
-    discoverable INTEGER NOT NULL DEFAULT 1,
-    PRIMARY KEY (did, rkey)
-  );
-`);
-
-try {
-  db.exec(
-    `ALTER TABLE note ADD COLUMN listed INTEGER NOT NULL DEFAULT 1;`,
-  );
-} catch {
-  // Column already exists — no-op
-}
-
-try {
-  db.exec(
-    `ALTER TABLE note ADD COLUMN language STRING;`,
-  );
-} catch {
-  // Column already exists — no-op
-}
-
-try {
-  db.exec(`ALTER TABLE note RENAME COLUMN listed TO discoverable;`);
-} catch {
-  // Column already renamed or doesn't exist — no-op
-}
-
-db.exec(`
-  CREATE TABLE IF NOT EXISTS state (
-    key TEXT PRIMARY KEY,
-    value TEXT NOT NULL
-  );
-`);
-
-db.exec(`
-  CREATE TABLE IF NOT EXISTS webhook_subscription (
-    id     INTEGER PRIMARY KEY,
-    did    TEXT NOT NULL,
-    method TEXT NOT NULL,
-    url    TEXT NOT NULL
-  );
-`);
-
-db.exec(`
-  CREATE INDEX IF NOT EXISTS idx_webhook_subscription_did
-    ON webhook_subscription(did);
-`);
-
-try {
-  db.exec(`ALTER TABLE webhook_subscription ADD COLUMN token TEXT;`);
-} catch {
-  // Column already exists — no-op
-}
-
-try {
-  db.exec(
-    `ALTER TABLE webhook_subscription ADD COLUMN verb TEXT NOT NULL DEFAULT 'create';`,
-  );
-  db.exec(`
-    INSERT INTO webhook_subscription (did, method, url, token, verb)
-    SELECT did, method, url, token, 'delete'
-    FROM webhook_subscription
-    WHERE verb = 'create';
-  `);
-} catch {
-  // Column already exists — backfill already happened on a previous run.
-}
-
-db.exec(`
-  CREATE INDEX IF NOT EXISTS idx_webhook_subscription_did_verb
-    ON webhook_subscription(did, verb);
-`);
+import { Database } from "@db/sqlite";
+import { applyMigrations } from "./apply.ts";
 
+const db = new Database(Deno.env.get("SQLITE_PATH") ?? "notes.db");
+applyMigrations(db);
 db.close();

src/search/opensearch.ts

@@ -0,0 +1,258 @@
+import type { API } from "@opensearch-project/opensearch";
+import { log } from "../log.ts";
+
+const MAX_LIMIT = 100;
+
+const openSearchUrl = (): string | undefined => Deno.env.get("OPENSEARCH_URL");
+const openSearchIndex = (): string =>
+  Deno.env.get("OPENSEARCH_INDEX") ?? "notes";
+
+const enabled = (): boolean => Boolean(openSearchUrl());
+
+const authHeaders = (): Record<string, string> => {
+  const username = Deno.env.get("OPENSEARCH_USERNAME");
+  const password = Deno.env.get("OPENSEARCH_PASSWORD");
+  if (username && password) {
+    const creds = btoa(`${username}:${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 = `${openSearchUrl()}${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 index = openSearchIndex();
+  const head = await request("HEAD", `/${index}`);
+  if (head.status === 200) return;
+  const res = await request("PUT", `/${index}`, indexMappings);
+  if (res.ok) {
+    log(`[opensearch] created index ${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",
+    `/${openSearchIndex()}/_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",
+    `/${openSearchIndex()}/_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;
+};
+
+export 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;
+};
+
+export const encodeCursor = (score: number, rkey: string): string =>
+  btoa(JSON.stringify([score, rkey]));
+
+export 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 searchAfter = opts.cursor ? decodeCursor(opts.cursor) : undefined;
+
+  const body: API.Search_RequestBody = {
+    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",
+        },
+        title: { number_of_fragments: 0 },
+      },
+    },
+    sort: [{ _score: "desc" }, { rkey: "desc" }],
+    _source: ["did", "rkey", "title", "content", "publishedAt"],
+    ...(searchAfter ? { search_after: searchAfter } : {}),
+  };
+
+  const res = await request("POST", `/${openSearchIndex()}/_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 };
+};

tests/_helpers/app.ts

@@ -0,0 +1,49 @@
+import type { Application } from "@oak/oak";
+
+export type RequestOptions = {
+  method?: string;
+  path: string;
+  headers?: HeadersInit;
+  body?: unknown;
+};
+
+export type TestResponse = {
+  status: number;
+  headers: Headers;
+  text: string;
+  json: () => unknown;
+};
+
+export const requestApp = async (
+  app: Application,
+  opts: RequestOptions,
+): Promise<TestResponse> => {
+  const method = opts.method ?? "GET";
+  const url = `http://test.local${opts.path}`;
+  const init: RequestInit = {
+    method,
+    headers: opts.headers,
+  };
+  if (opts.body !== undefined && method !== "GET" && method !== "HEAD") {
+    init.body = typeof opts.body === "string"
+      ? opts.body
+      : JSON.stringify(opts.body);
+    const headers = new Headers(opts.headers);
+    if (!headers.has("Content-Type")) {
+      headers.set("Content-Type", "application/json");
+    }
+    init.headers = headers;
+  }
+  const req = new Request(url, init);
+  const res = await app.handle(req);
+  if (!res) {
+    throw new Error("app.handle returned undefined");
+  }
+  const text = await res.text();
+  return {
+    status: res.status,
+    headers: res.headers,
+    text,
+    json: () => JSON.parse(text),
+  };
+};

tests/_helpers/auth.ts

@@ -0,0 +1,18 @@
+import {
+  _resetAuthenticatorForTest,
+  _setAuthenticatorForTest,
+} from "../../server.ts";
+
+export type AuthStub = { restore: () => void };
+
+export const stubAuthenticate = (didOrError: string | Error): AuthStub => {
+  _setAuthenticatorForTest(() => {
+    if (didOrError instanceof Error) {
+      return Promise.reject(didOrError);
+    }
+    return Promise.resolve(didOrError);
+  });
+  return {
+    restore: () => _resetAuthenticatorForTest(),
+  };
+};

tests/_helpers/db.ts

@@ -0,0 +1,60 @@
+import { Database } from "@db/sqlite";
+import {
+  _resetDbForTest,
+  _setDbForTest,
+  addWebhookSubscription,
+  upsertNote,
+  type WebhookVerb,
+} from "../../src/data/db.ts";
+import { applyMigrations } from "../../src/migrations/apply.ts";
+import type { Note } from "../../src/data/note.ts";
+
+export const withTestDb = (
+  fn: (db: Database) => Promise<void> | void,
+): () => Promise<void> => {
+  return async () => {
+    const db = new Database(":memory:");
+    applyMigrations(db);
+    _setDbForTest(db);
+    try {
+      await fn(db);
+    } finally {
+      _resetDbForTest();
+      db.close();
+    }
+  };
+};
+
+let seedCounter = 0;
+
+export const seedNote = (overrides: Partial<Note> = {}): Note => {
+  seedCounter++;
+  const note: Note = {
+    did: overrides.did ?? "did:plc:test",
+    rkey: overrides.rkey ?? `seed${seedCounter.toString().padStart(6, "0")}`,
+    title: overrides.title ?? "Test title",
+    publishedAt: overrides.publishedAt ?? "2026-01-01T00:00:00.000Z",
+    createdAt: overrides.createdAt ?? "2026-01-01T00:00:00.000Z",
+    discoverable: overrides.discoverable,
+    language: overrides.language,
+    content: overrides.content,
+  };
+  upsertNote(note);
+  return note;
+};
+
+export const seedWebhook = (overrides: {
+  did?: string;
+  method?: string;
+  url?: string;
+  token?: string;
+  verb?: WebhookVerb;
+} = {}) => {
+  return addWebhookSubscription({
+    did: overrides.did ?? "did:plc:test",
+    method: overrides.method ?? "POST",
+    url: overrides.url ?? "https://example.com/hook",
+    token: overrides.token,
+    verb: overrides.verb ?? "create",
+  });
+};

tests/_helpers/fetch_stub.ts

@@ -0,0 +1,57 @@
+export type FetchHandler = (
+  req: Request,
+) => Response | Promise<Response>;
+
+export type CapturedRequest = {
+  method: string;
+  url: string;
+  headers: Headers;
+  body: string;
+};
+
+export type FetchStub = {
+  calls: CapturedRequest[];
+  restore: () => void;
+};
+
+export const installFetchStub = (handler: FetchHandler): FetchStub => {
+  const original = globalThis.fetch;
+  const calls: CapturedRequest[] = [];
+  let restored = false;
+
+  globalThis.fetch = async (input, init) => {
+    const req = new Request(input as RequestInfo, init);
+    const body = req.method === "GET" || req.method === "HEAD"
+      ? ""
+      : await req.clone().text();
+    calls.push({
+      method: req.method,
+      url: req.url,
+      headers: new Headers(req.headers),
+      body,
+    });
+    return await handler(req);
+  };
+
+  return {
+    calls,
+    restore: () => {
+      if (restored) throw new Error("FetchStub.restore() called twice");
+      restored = true;
+      globalThis.fetch = original;
+    },
+  };
+};
+
+export const jsonResponse = (
+  body: unknown,
+  init: ResponseInit = {},
+): Response => {
+  return new Response(JSON.stringify(body), {
+    status: init.status ?? 200,
+    headers: { "Content-Type": "application/json", ...(init.headers ?? {}) },
+  });
+};
+
+export const errorResponse = (status: number, text = "error"): Response =>
+  new Response(text, { status });

tests/auth_verify_test.ts

@@ -0,0 +1,118 @@
+import { assertEquals, assertRejects, assertThrows } from "@std/assert";
+import { authenticateRequest, decodeJwtPayload } from "../src/auth/verify.ts";
+import { installFetchStub, jsonResponse } from "./_helpers/fetch_stub.ts";
+
+const makeJwt = (payload: Record<string, unknown>): string => {
+  const header = btoa(JSON.stringify({ alg: "HS256", typ: "JWT" }));
+  const body = btoa(JSON.stringify(payload));
+  return `${header}.${body}.sig`;
+};
+
+Deno.test("decodeJwtPayload parses sub from a 3-segment JWT", () => {
+  const token = makeJwt({ sub: "did:plc:abc", iat: 1 });
+  const payload = decodeJwtPayload(token);
+  assertEquals(payload.sub, "did:plc:abc");
+});
+
+Deno.test("decodeJwtPayload throws on malformed token", () => {
+  assertThrows(() => decodeJwtPayload("not.a.valid.jwt"));
+  assertThrows(() => decodeJwtPayload("onlytwo.parts"));
+});
+
+Deno.test("authenticateRequest rejects missing or non-Bearer Authorization", async () => {
+  await assertRejects(() => authenticateRequest(null));
+  await assertRejects(() => authenticateRequest(""));
+  await assertRejects(() => authenticateRequest("Basic abc"));
+});
+
+Deno.test(
+  "authenticateRequest resolves PDS then verifies session and returns did",
+  async () => {
+    const did = "did:plc:user";
+    const stub = installFetchStub((req) => {
+      if (req.url === `https://plc.directory/${did}`) {
+        return jsonResponse({
+          service: [{
+            id: "#atproto_pds",
+            serviceEndpoint: "https://pds.example.com",
+          }],
+        });
+      }
+      if (
+        req.url ===
+          "https://pds.example.com/xrpc/com.atproto.server.getSession"
+      ) {
+        return jsonResponse({ did });
+      }
+      return jsonResponse({ error: "unexpected" }, { status: 500 });
+    });
+
+    try {
+      const result = await authenticateRequest(
+        `Bearer ${makeJwt({ sub: did })}`,
+      );
+      assertEquals(result, did);
+      const sessionCall = stub.calls.find((c) => c.url.endsWith("getSession"));
+      assertEquals(
+        sessionCall?.headers.get("Authorization")?.startsWith("Bearer "),
+        true,
+      );
+    } finally {
+      stub.restore();
+    }
+  },
+);
+
+Deno.test(
+  "authenticateRequest throws when plc.directory or getSession is non-2xx",
+  async () => {
+    const did = "did:plc:user";
+
+    const plcFailStub = installFetchStub(() =>
+      new Response("not found", { status: 404 })
+    );
+    try {
+      await assertRejects(() =>
+        authenticateRequest(`Bearer ${makeJwt({ sub: did })}`)
+      );
+    } finally {
+      plcFailStub.restore();
+    }
+
+    const sessionFailStub = installFetchStub((req) => {
+      if (req.url.startsWith("https://plc.directory/")) {
+        return jsonResponse({
+          service: [{
+            id: "#atproto_pds",
+            serviceEndpoint: "https://pds.example.com",
+          }],
+        });
+      }
+      return new Response("nope", { status: 401 });
+    });
+    try {
+      await assertRejects(() =>
+        authenticateRequest(`Bearer ${makeJwt({ sub: did })}`)
+      );
+    } finally {
+      sessionFailStub.restore();
+    }
+  },
+);
+
+Deno.test(
+  "authenticateRequest throws when DID document has no atproto_pds service",
+  async () => {
+    const did = "did:plc:user";
+    const stub = installFetchStub(() =>
+      jsonResponse({ service: [{ id: "#other", serviceEndpoint: "x" }] })
+    );
+    try {
+      await assertRejects(() =>
+        authenticateRequest(`Bearer ${makeJwt({ sub: did })}`)
+      );
+    } finally {
+      stub.restore();
+    }
+  },
+);

tests/db_test.ts

@@ -0,0 +1,141 @@
+import { assertEquals, assertFalse } from "@std/assert";
+import {
+  addWebhookSubscription,
+  deleteWebhookById,
+  getNotes,
+  getNotesByDids,
+  getWebhooksByDidAndVerb,
+  listWebhooksByDid,
+  upsertNote,
+} from "../src/data/db.ts";
+import { seedNote, seedWebhook, withTestDb } from "./_helpers/db.ts";
+
+Deno.test(
+  "upsertNote inserts then updates on (did, rkey) conflict",
+  withTestDb(() => {
+    upsertNote({
+      did: "did:plc:user",
+      rkey: "rkey1",
+      title: "First",
+      publishedAt: "2026-01-01T00:00:00.000Z",
+      createdAt: "2026-01-01T00:00:00.000Z",
+    });
+    upsertNote({
+      did: "did:plc:user",
+      rkey: "rkey1",
+      title: "Second",
+      publishedAt: "2026-02-01T00:00:00.000Z",
+      createdAt: "2026-01-01T00:00:00.000Z",
+    });
+
+    const { notes } = getNotes(undefined, 10);
+    assertEquals(notes.length, 1);
+    assertEquals(notes[0].title, "Second");
+    assertEquals(notes[0].publishedAt, "2026-02-01T00:00:00.000Z");
+  }),
+);
+
+Deno.test(
+  "upsertNote treats undefined discoverable as 1 and false as 0",
+  withTestDb(() => {
+    upsertNote({
+      did: "did:plc:user",
+      rkey: "default",
+      title: "Default",
+      publishedAt: "2026-01-01T00:00:00.000Z",
+      createdAt: "2026-01-01T00:00:00.000Z",
+    });
+    upsertNote({
+      did: "did:plc:user",
+      rkey: "hidden",
+      title: "Hidden",
+      publishedAt: "2026-01-01T00:00:00.000Z",
+      createdAt: "2026-01-01T00:00:00.000Z",
+      discoverable: false,
+    });
+
+    const { notes } = getNotes(undefined, 10);
+    const rkeys = notes.map((n) => n.rkey);
+    assertEquals(rkeys, ["default"]);
+  }),
+);
+
+Deno.test(
+  "getNotes paginates rkey desc and returns cursor only when page is full",
+  withTestDb(() => {
+    for (const rkey of ["a", "b", "c", "d", "e"]) {
+      seedNote({ rkey });
+    }
+
+    const firstPage = getNotes(undefined, 3);
+    assertEquals(firstPage.notes.map((n) => n.rkey), ["e", "d", "c"]);
+    assertEquals(firstPage.cursor, "c");
+
+    const secondPage = getNotes(firstPage.cursor, 3);
+    assertEquals(secondPage.notes.map((n) => n.rkey), ["b", "a"]);
+    assertEquals(secondPage.cursor, undefined);
+  }),
+);
+
+Deno.test(
+  "getNotesByDids returns empty without running SQL on empty did list",
+  withTestDb(() => {
+    seedNote({ did: "did:plc:alice", rkey: "a" });
+    const result = getNotesByDids([], undefined, 10);
+    assertEquals(result, { notes: [] });
+  }),
+);
+
+Deno.test(
+  "webhook CRUD: add omits token, list orders desc, delete by id reports hit",
+  withTestDb(() => {
+    const created = addWebhookSubscription({
+      did: "did:plc:user",
+      method: "POST",
+      url: "https://example.com/a",
+      token: "secret",
+      verb: "create",
+    });
+    assertEquals(created.token, undefined);
+    assertEquals(created.url, "https://example.com/a");
+
+    seedWebhook({
+      did: "did:plc:user",
+      url: "https://example.com/b",
+      verb: "create",
+    });
+    seedWebhook({
+      did: "did:plc:user",
+      url: "https://example.com/c",
+      verb: "delete",
+    });
+
+    const list = listWebhooksByDid("did:plc:user");
+    assertEquals(list.length, 3);
+    const ids = list.map((w) => w.id);
+    assertEquals(ids, [...ids].sort((a, b) => b - a));
+
+    assertFalse(deleteWebhookById({ did: "did:plc:user", id: 9999 }));
+    const deleted = deleteWebhookById({ did: "did:plc:user", id: created.id });
+    assertEquals(deleted, true);
+    assertEquals(listWebhooksByDid("did:plc:user").length, 2);
+  }),
+);
+
+Deno.test(
+  "getWebhooksByDidAndVerb filters by verb and caps at 10",
+  withTestDb(() => {
+    for (let i = 0; i < 12; i++) {
+      seedWebhook({ url: `https://example.com/c${i}`, verb: "create" });
+    }
+    seedWebhook({ url: "https://example.com/d", verb: "delete" });
+
+    const creates = getWebhooksByDidAndVerb("did:plc:test", "create");
+    assertEquals(creates.length, 10);
+    for (const row of creates) assertEquals(row.verb, "create");
+
+    const deletes = getWebhooksByDidAndVerb("did:plc:test", "delete");
+    assertEquals(deletes.length, 1);
+    assertEquals(deletes[0].verb, "delete");
+  }),
+);

tests/jetstream_bulk_test.ts

@@ -0,0 +1,147 @@
+import { assertEquals } from "@std/assert";
+import { FakeTime } from "@std/testing/time";
+import {
+  _resetBulkBuffersForTest,
+  type BulkDeps,
+  queueBulkCreate,
+} from "../src/jetstream/bulk.ts";
+import type { WebhookTarget } from "../src/jetstream/webhooks.ts";
+import type { WebhookVerb } from "../src/data/db.ts";
+
+type DispatchCall = {
+  webhooks: WebhookTarget[];
+  payload: Record<string, unknown>;
+  label: string;
+};
+
+const flushMicrotasks = async (): Promise<void> => {
+  for (let i = 0; i < 5; i++) {
+    await Promise.resolve();
+  }
+};
+
+const makeDeps = (
+  webhooks: WebhookTarget[] = [],
+): {
+  deps: BulkDeps;
+  getSubsCalls: { did: string; verb: WebhookVerb }[];
+  dispatchCalls: DispatchCall[];
+} => {
+  const getSubsCalls: { did: string; verb: WebhookVerb }[] = [];
+  const dispatchCalls: DispatchCall[] = [];
+  const deps: BulkDeps = {
+    getSubs: (did, verb) => {
+      getSubsCalls.push({ did, verb });
+      return webhooks;
+    },
+    dispatch: (w, payload, label) => {
+      dispatchCalls.push({ webhooks: w, payload, label });
+      return Promise.resolve();
+    },
+    debounceMs: 500,
+  };
+  return { deps, getSubsCalls, dispatchCalls };
+};
+
+Deno.test("queueBulkCreate accumulates records and flushes once after debounce", async () => {
+  const time = new FakeTime();
+  try {
+    _resetBulkBuffersForTest();
+    const { deps, dispatchCalls } = makeDeps();
+
+    queueBulkCreate("did:plc:a", { rkey: "r1" }, deps);
+    queueBulkCreate("did:plc:a", { rkey: "r2" }, deps);
+    queueBulkCreate("did:plc:a", { rkey: "r3" }, deps);
+
+    assertEquals(dispatchCalls.length, 0);
+
+    await time.tickAsync(500);
+    await flushMicrotasks();
+
+    assertEquals(dispatchCalls.length, 1);
+    const payload = dispatchCalls[0].payload as {
+      records: { rkey: string }[];
+    };
+    assertEquals(payload.records.map((r) => r.rkey), ["r1", "r2", "r3"]);
+  } finally {
+    _resetBulkBuffersForTest();
+    time.restore();
+  }
+});
+
+Deno.test("queueBulkCreate restarts the timer on each push (debounce)", async () => {
+  const time = new FakeTime();
+  try {
+    _resetBulkBuffersForTest();
+    const { deps, dispatchCalls } = makeDeps();
+
+    queueBulkCreate("did:plc:a", { rkey: "r1" }, deps);
+    await time.tickAsync(400);
+    queueBulkCreate("did:plc:a", { rkey: "r2" }, deps);
+    await time.tickAsync(400);
+
+    assertEquals(dispatchCalls.length, 0);
+
+    await time.tickAsync(100);
+    await flushMicrotasks();
+
+    assertEquals(dispatchCalls.length, 1);
+  } finally {
+    _resetBulkBuffersForTest();
+    time.restore();
+  }
+});
+
+Deno.test("flushBulkCreate uses bulk-create verb and the standard payload shape", async () => {
+  const time = new FakeTime();
+  try {
+    _resetBulkBuffersForTest();
+    const targets: WebhookTarget[] = [
+      { method: "POST", url: "https://hook.example/bulk" },
+    ];
+    const { deps, getSubsCalls, dispatchCalls } = makeDeps(targets);
+
+    queueBulkCreate("did:plc:a", { rkey: "r1", title: "T" }, deps);
+    await time.tickAsync(500);
+    await flushMicrotasks();
+
+    assertEquals(getSubsCalls, [{ did: "did:plc:a", verb: "bulk-create" }]);
+    assertEquals(dispatchCalls.length, 1);
+    assertEquals(dispatchCalls[0].webhooks, targets);
+    assertEquals(dispatchCalls[0].label, "bulk-create did:plc:a");
+    assertEquals(dispatchCalls[0].payload, {
+      event: "bulk-create",
+      did: "did:plc:a",
+      records: [{ rkey: "r1", title: "T" }],
+    });
+  } finally {
+    _resetBulkBuffersForTest();
+    time.restore();
+  }
+});
+
+Deno.test("bulkBuffers are cleared after flush and isolated per did", async () => {
+  const time = new FakeTime();
+  try {
+    _resetBulkBuffersForTest();
+    const { deps, dispatchCalls } = makeDeps();
+
+    queueBulkCreate("did:plc:a", { rkey: "a1" }, deps);
+    queueBulkCreate("did:plc:b", { rkey: "b1" }, deps);
+    await time.tickAsync(500);
+    await flushMicrotasks();
+
+    assertEquals(dispatchCalls.length, 2);
+    const dids = dispatchCalls.map((c) => (c.payload as { did: string }).did)
+      .sort();
+    assertEquals(dids, ["did:plc:a", "did:plc:b"]);
+
+    // Another flush window should not re-dispatch the same records.
+    await time.tickAsync(500);
+    await flushMicrotasks();
+    assertEquals(dispatchCalls.length, 2);
+  } finally {
+    _resetBulkBuffersForTest();
+    time.restore();
+  }
+});

tests/jetstream_webhooks_test.ts

@@ -0,0 +1,94 @@
+import { assertEquals } from "@std/assert";
+import { dispatchAll } from "../src/jetstream/webhooks.ts";
+import { installFetchStub, jsonResponse } from "./_helpers/fetch_stub.ts";
+
+Deno.test("dispatchAll is a noop on empty webhook list", async () => {
+  const stub = installFetchStub(() =>
+    jsonResponse({ should: "not be called" })
+  );
+  try {
+    await dispatchAll([], { event: "create" }, "label");
+    assertEquals(stub.calls.length, 0);
+  } finally {
+    stub.restore();
+  }
+});
+
+Deno.test(
+  "dispatchAll sends Authorization Bearer only when token is present",
+  async () => {
+    const stub = installFetchStub(() => new Response(null, { status: 204 }));
+    try {
+      await dispatchAll(
+        [
+          { method: "POST", url: "https://a.example/hook" },
+          {
+            method: "POST",
+            url: "https://b.example/hook",
+            token: "shh",
+          },
+        ],
+        { event: "create" },
+        "label",
+      );
+      assertEquals(stub.calls.length, 2);
+      const [first, second] = stub.calls;
+      assertEquals(first.headers.get("Authorization"), null);
+      assertEquals(second.headers.get("Authorization"), "Bearer shh");
+      assertEquals(first.headers.get("Content-Type"), "application/json");
+    } finally {
+      stub.restore();
+    }
+  },
+);
+
+Deno.test(
+  "dispatchAll omits body and Content-Type for GET and HEAD",
+  async () => {
+    const stub = installFetchStub(() => new Response(null, { status: 204 }));
+    try {
+      await dispatchAll(
+        [
+          { method: "GET", url: "https://a.example/ping" },
+          { method: "HEAD", url: "https://b.example/ping" },
+        ],
+        { event: "create", data: "ignored" },
+        "label",
+      );
+      assertEquals(stub.calls.length, 2);
+      for (const call of stub.calls) {
+        assertEquals(call.body, "");
+        assertEquals(call.headers.get("Content-Type"), null);
+      }
+    } finally {
+      stub.restore();
+    }
+  },
+);
+
+Deno.test(
+  "dispatchAll uses Promise.allSettled — one rejection does not abort others",
+  async () => {
+    const stub = installFetchStub((req) => {
+      if (req.url === "https://reject.example/hook") {
+        return Promise.reject(new Error("connection refused"));
+      }
+      return Promise.resolve(new Response(null, { status: 204 }));
+    });
+    try {
+      await dispatchAll(
+        [
+          { method: "POST", url: "https://reject.example/hook" },
+          { method: "POST", url: "https://ok.example/hook" },
+        ],
+        { event: "create" },
+        "label",
+      );
+      assertEquals(stub.calls.length, 2);
+      const urls = stub.calls.map((c) => c.url);
+      assertEquals(urls.includes("https://ok.example/hook"), true);
+    } finally {
+      stub.restore();
+    }
+  },
+);

tests/migrations_test.ts

@@ -0,0 +1,115 @@
+import { assertEquals } from "@std/assert";
+import { Database } from "@db/sqlite";
+import { applyMigrations } from "../src/migrations/apply.ts";
+
+type TableInfoRow = { name: string };
+type ColumnInfoRow = { name: string };
+
+const tableNames = (db: Database): string[] =>
+  db.prepare(
+    "SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name",
+  ).all<TableInfoRow>().map((r) => r.name);
+
+const columnNames = (db: Database, table: string): string[] =>
+  db.prepare(`PRAGMA table_info(${table})`)
+    .all<ColumnInfoRow>()
+    .map((r) => r.name)
+    .sort();
+
+const indexNames = (db: Database): string[] =>
+  db.prepare(
+    "SELECT name FROM sqlite_master WHERE type = 'index' AND name NOT LIKE 'sqlite_%' ORDER BY name",
+  ).all<TableInfoRow>().map((r) => r.name);
+
+Deno.test("applyMigrations creates expected tables, columns, and indices", () => {
+  const db = new Database(":memory:");
+  try {
+    applyMigrations(db);
+
+    const tables = tableNames(db);
+    assertEquals(
+      tables.includes("note") &&
+        tables.includes("state") &&
+        tables.includes("webhook_subscription"),
+      true,
+    );
+
+    const noteCols = columnNames(db, "note");
+    for (
+      const col of [
+        "title",
+        "publishedAt",
+        "createdAt",
+        "did",
+        "rkey",
+        "discoverable",
+        "language",
+      ]
+    ) {
+      assertEquals(noteCols.includes(col), true, `note.${col} missing`);
+    }
+
+    const webhookCols = columnNames(db, "webhook_subscription");
+    for (const col of ["id", "did", "method", "url", "token", "verb"]) {
+      assertEquals(
+        webhookCols.includes(col),
+        true,
+        `webhook_subscription.${col} missing`,
+      );
+    }
+
+    const indices = indexNames(db);
+    assertEquals(indices.includes("idx_webhook_subscription_did"), true);
+    assertEquals(indices.includes("idx_webhook_subscription_did_verb"), true);
+  } finally {
+    db.close();
+  }
+});
+
+Deno.test("applyMigrations is idempotent", () => {
+  const db = new Database(":memory:");
+  try {
+    applyMigrations(db);
+    db.exec(
+      "INSERT INTO webhook_subscription (did, method, url, verb) VALUES ('did:plc:x', 'POST', 'https://e.com', 'create')",
+    );
+    applyMigrations(db);
+    applyMigrations(db);
+
+    const [{ count }] = db.prepare(
+      "SELECT COUNT(*) AS count FROM webhook_subscription WHERE did = 'did:plc:x' AND verb = 'create'",
+    ).all<{ count: number }>();
+    assertEquals(count, 1);
+  } finally {
+    db.close();
+  }
+});
+
+Deno.test(
+  "applyMigrations upgrades a legacy schema missing language and discoverable",
+  () => {
+    const db = new Database(":memory:");
+    try {
+      db.exec(`
+        CREATE TABLE note (
+          title TEXT NOT NULL,
+          publishedAt DATETIME,
+          createdAt DATETIME NOT NULL,
+          did TEXT NOT NULL,
+          rkey TEXT NOT NULL,
+          listed INTEGER NOT NULL DEFAULT 1,
+          PRIMARY KEY (did, rkey)
+        );
+      `);
+
+      applyMigrations(db);
+
+      const cols = columnNames(db, "note");
+      assertEquals(cols.includes("discoverable"), true);
+      assertEquals(cols.includes("language"), true);
+      assertEquals(cols.includes("listed"), false);
+    } finally {
+      db.close();
+    }
+  },
+);

tests/opensearch_test.ts

@@ -0,0 +1,204 @@
+import { assertEquals, assertRejects } from "@std/assert";
+import {
+  decodeCursor,
+  encodeCursor,
+  firstSnippet,
+  searchNotes,
+} from "../src/search/opensearch.ts";
+import { installFetchStub, jsonResponse } from "./_helpers/fetch_stub.ts";
+
+const withOpenSearchEnv = (fn: () => Promise<void> | void) => {
+  return async () => {
+    const previousUrl = Deno.env.get("OPENSEARCH_URL");
+    const previousIndex = Deno.env.get("OPENSEARCH_INDEX");
+    Deno.env.set("OPENSEARCH_URL", "http://opensearch.test");
+    Deno.env.set("OPENSEARCH_INDEX", "notes");
+    try {
+      await fn();
+    } finally {
+      if (previousUrl === undefined) Deno.env.delete("OPENSEARCH_URL");
+      else Deno.env.set("OPENSEARCH_URL", previousUrl);
+      if (previousIndex === undefined) Deno.env.delete("OPENSEARCH_INDEX");
+      else Deno.env.set("OPENSEARCH_INDEX", previousIndex);
+    }
+  };
+};
+
+Deno.test("encodeCursor and decodeCursor round-trip", () => {
+  const cursor = encodeCursor(3.14, "abc123");
+  assertEquals(decodeCursor(cursor), [3.14, "abc123"]);
+});
+
+Deno.test("decodeCursor returns undefined on garbage input", () => {
+  assertEquals(decodeCursor("not-base64-at-all!!!"), undefined);
+  assertEquals(decodeCursor(btoa("not-json")), undefined);
+  assertEquals(decodeCursor(btoa(JSON.stringify([1]))), undefined);
+  assertEquals(
+    decodeCursor(btoa(JSON.stringify(["wrong", "type"]))),
+    undefined,
+  );
+});
+
+Deno.test("firstSnippet prefers highlight.content > title > source content > title fallback", () => {
+  assertEquals(
+    firstSnippet(
+      { content: ["hl-content"], title: ["hl-title"] },
+      { title: "src-title", content: "src-content" },
+    ),
+    "hl-content",
+  );
+  assertEquals(
+    firstSnippet({ title: ["hl-title"] }, {
+      title: "src-title",
+      content: "src-content",
+    }),
+    "hl-title",
+  );
+  const longContent = "a".repeat(500);
+  assertEquals(
+    firstSnippet(undefined, { content: longContent, title: "ignored" }).length,
+    200,
+  );
+  assertEquals(firstSnippet(undefined, { title: "only-title" }), "only-title");
+});
+
+Deno.test(
+  "searchNotes returns empty when OPENSEARCH_URL is unset",
+  async () => {
+    const previousUrl = Deno.env.get("OPENSEARCH_URL");
+    Deno.env.delete("OPENSEARCH_URL");
+    try {
+      const stub = installFetchStub(() =>
+        jsonResponse({ should: "not be called" })
+      );
+      try {
+        const result = await searchNotes({
+          q: "hello",
+          discoverableOnly: true,
+          limit: 10,
+        });
+        assertEquals(result, { results: [] });
+        assertEquals(stub.calls.length, 0);
+      } finally {
+        stub.restore();
+      }
+    } finally {
+      if (previousUrl !== undefined) {
+        Deno.env.set("OPENSEARCH_URL", previousUrl);
+      }
+    }
+  },
+);
+
+Deno.test(
+  "searchNotes builds multi_match body with filters and cursor",
+  withOpenSearchEnv(async () => {
+    const stub = installFetchStub(() =>
+      jsonResponse({
+        hits: {
+          hits: [
+            {
+              _source: {
+                did: "did:plc:a",
+                rkey: "rk-a",
+                title: "T",
+                content: "c",
+              },
+              _score: 1.5,
+              highlight: { content: ["snippet"] },
+            },
+            {
+              _source: {
+                did: "did:plc:b",
+                rkey: "rk-b",
+                title: "U",
+                content: "d",
+              },
+              _score: 1.0,
+            },
+          ],
+        },
+      })
+    );
+    try {
+      const result = await searchNotes({
+        q: "hello world",
+        discoverableOnly: true,
+        did: "did:plc:a",
+        limit: 2,
+        cursor: encodeCursor(2.0, "rk-prev"),
+      });
+      assertEquals(result.results.length, 2);
+      assertEquals(result.results[0].snippet, "snippet");
+      assertEquals(result.results[1].snippet, "d");
+      assertEquals(result.cursor, encodeCursor(1.0, "rk-b"));
+
+      assertEquals(stub.calls.length, 1);
+      const sent = JSON.parse(stub.calls[0].body) as Record<string, unknown>;
+      assertEquals(sent.size, 2);
+      assertEquals(sent.search_after, [2.0, "rk-prev"]);
+      const query = sent.query as {
+        bool: { must: unknown[]; filter: unknown[] };
+      };
+      const multi =
+        (query.bool.must[0] as { multi_match: Record<string, unknown> })
+          .multi_match;
+      assertEquals(multi.query, "hello world");
+      assertEquals(multi.fields, ["title^2", "content"]);
+      assertEquals(multi.fuzziness, "AUTO");
+      assertEquals(query.bool.filter, [
+        { term: { discoverable: true } },
+        { term: { did: "did:plc:a" } },
+      ]);
+    } finally {
+      stub.restore();
+    }
+  }),
+);
+
+Deno.test(
+  "searchNotes returns cursor only when the page is full",
+  withOpenSearchEnv(async () => {
+    const stub = installFetchStub(() =>
+      jsonResponse({
+        hits: {
+          hits: [
+            {
+              _source: { did: "did:plc:a", rkey: "rk", title: "T" },
+              _score: 1,
+            },
+          ],
+        },
+      })
+    );
+    try {
+      const result = await searchNotes({
+        q: "hi",
+        discoverableOnly: false,
+        limit: 5,
+      });
+      assertEquals(result.results.length, 1);
+      assertEquals(result.cursor, undefined);
+    } finally {
+      stub.restore();
+    }
+  }),
+);
+
+Deno.test(
+  "searchNotes throws on non-ok response",
+  withOpenSearchEnv(async () => {
+    const stub = installFetchStub(() => new Response("boom", { status: 500 }));
+    try {
+      await assertRejects(() =>
+        searchNotes({
+          q: "hi",
+          discoverableOnly: false,
+          limit: 5,
+        })
+      );
+    } finally {
+      stub.restore();
+    }
+  }),
+);

tests/server_test.ts

@@ -0,0 +1,287 @@
+import { assertEquals } from "@std/assert";
+import { createApp } from "../server.ts";
+import { addWebhookSubscription, listWebhooksByDid } from "../src/data/db.ts";
+import { seedNote, withTestDb } from "./_helpers/db.ts";
+import { requestApp } from "./_helpers/app.ts";
+import { stubAuthenticate } from "./_helpers/auth.ts";
+import { installFetchStub, jsonResponse } from "./_helpers/fetch_stub.ts";
+
+const withAdminEnv = (
+  dids: string,
+  fn: () => Promise<void> | void,
+): () => Promise<void> => {
+  return async () => {
+    const previous = Deno.env.get("ADMIN_DIDS");
+    Deno.env.set("ADMIN_DIDS", dids);
+    try {
+      await fn();
+    } finally {
+      if (previous === undefined) Deno.env.delete("ADMIN_DIDS");
+      else Deno.env.set("ADMIN_DIDS", previous);
+    }
+  };
+};
+
+const withOpenSearchEnv = (fn: () => Promise<void> | void) => {
+  return async () => {
+    const previous = Deno.env.get("OPENSEARCH_URL");
+    Deno.env.set("OPENSEARCH_URL", "http://opensearch.test");
+    try {
+      await fn();
+    } finally {
+      if (previous === undefined) Deno.env.delete("OPENSEARCH_URL");
+      else Deno.env.set("OPENSEARCH_URL", previous);
+    }
+  };
+};
+
+Deno.test(
+  "GET /health returns 200 with status ok",
+  withTestDb(async () => {
+    const res = await requestApp(createApp(), { path: "/health" });
+    assertEquals(res.status, 200);
+    assertEquals(res.json(), { status: "ok" });
+  }),
+);
+
+Deno.test(
+  "GET /notes paginates rkey desc and honors cursor + limit",
+  withTestDb(async () => {
+    for (const rkey of ["a", "b", "c", "d", "e"]) {
+      seedNote({ rkey });
+    }
+    const app = createApp();
+
+    const first = await requestApp(app, { path: "/notes?limit=3" });
+    assertEquals(first.status, 200);
+    const firstBody = first.json() as {
+      notes: { rkey: string }[];
+      cursor?: string;
+    };
+    assertEquals(firstBody.notes.map((n) => n.rkey), ["e", "d", "c"]);
+    assertEquals(firstBody.cursor, "c");
+
+    const second = await requestApp(app, {
+      path: `/notes?limit=3&cursor=${firstBody.cursor}`,
+    });
+    const secondBody = second.json() as { notes: { rkey: string }[] };
+    assertEquals(secondBody.notes.map((n) => n.rkey), ["b", "a"]);
+  }),
+);
+
+Deno.test(
+  "POST /notes/feed rejects empty dids and accepts a valid list",
+  withTestDb(async () => {
+    seedNote({ did: "did:plc:alice", rkey: "rk1" });
+    const app = createApp();
+
+    const empty = await requestApp(app, {
+      method: "POST",
+      path: "/notes/feed",
+      body: { dids: [] },
+    });
+    assertEquals(empty.status, 400);
+
+    const notArray = await requestApp(app, {
+      method: "POST",
+      path: "/notes/feed",
+      body: { dids: "did:plc:alice" },
+    });
+    assertEquals(notArray.status, 400);
+
+    const ok = await requestApp(app, {
+      method: "POST",
+      path: "/notes/feed",
+      body: { dids: ["did:plc:alice"] },
+    });
+    assertEquals(ok.status, 200);
+    const body = ok.json() as { notes: { rkey: string }[] };
+    assertEquals(body.notes.length, 1);
+  }),
+);
+
+Deno.test(
+  "POST /:did/webhooks: 401 without auth, 403 on DID mismatch, 201 inserts create+delete by default",
+  withTestDb(async () => {
+    const app = createApp();
+
+    const noAuth = await requestApp(app, {
+      method: "POST",
+      path: "/did:plc:alice/webhooks",
+      body: { method: "POST", url: "https://hook.example/x" },
+    });
+    assertEquals(noAuth.status, 401);
+
+    const mismatch = stubAuthenticate("did:plc:bob");
+    try {
+      const res = await requestApp(app, {
+        method: "POST",
+        path: "/did:plc:alice/webhooks",
+        headers: { Authorization: "Bearer fake" },
+        body: { method: "POST", url: "https://hook.example/x" },
+      });
+      assertEquals(res.status, 403);
+    } finally {
+      mismatch.restore();
+    }
+
+    const owned = stubAuthenticate("did:plc:alice");
+    try {
+      const res = await requestApp(app, {
+        method: "POST",
+        path: "/did:plc:alice/webhooks",
+        headers: { Authorization: "Bearer fake" },
+        body: { method: "POST", url: "https://hook.example/x" },
+      });
+      assertEquals(res.status, 201);
+      const created = res.json() as { verb: string }[];
+      const verbs = created.map((w) => w.verb).sort();
+      assertEquals(verbs, ["create", "delete"]);
+    } finally {
+      owned.restore();
+    }
+  }),
+);
+
+Deno.test(
+  "POST /:did/webhooks rejects unknown verb with 400",
+  withTestDb(async () => {
+    const auth = stubAuthenticate("did:plc:alice");
+    try {
+      const res = await requestApp(createApp(), {
+        method: "POST",
+        path: "/did:plc:alice/webhooks",
+        headers: { Authorization: "Bearer fake" },
+        body: {
+          method: "POST",
+          url: "https://hook.example/x",
+          verb: "destroy",
+        },
+      });
+      assertEquals(res.status, 400);
+    } finally {
+      auth.restore();
+    }
+  }),
+);
+
+Deno.test(
+  "DELETE /:did/webhooks/:id: 400 on non-positive, 404 on unknown, 204 on success",
+  withTestDb(async () => {
+    const created = addWebhookSubscription({
+      did: "did:plc:alice",
+      method: "POST",
+      url: "https://hook.example/x",
+      verb: "create",
+    });
+    const auth = stubAuthenticate("did:plc:alice");
+    const app = createApp();
+    try {
+      const bad = await requestApp(app, {
+        method: "DELETE",
+        path: "/did:plc:alice/webhooks/0",
+        headers: { Authorization: "Bearer fake" },
+      });
+      assertEquals(bad.status, 400);
+
+      const missing = await requestApp(app, {
+        method: "DELETE",
+        path: "/did:plc:alice/webhooks/9999",
+        headers: { Authorization: "Bearer fake" },
+      });
+      assertEquals(missing.status, 404);
+
+      const ok = await requestApp(app, {
+        method: "DELETE",
+        path: `/did:plc:alice/webhooks/${created.id}`,
+        headers: { Authorization: "Bearer fake" },
+      });
+      assertEquals(ok.status, 204);
+      assertEquals(listWebhooksByDid("did:plc:alice").length, 0);
+    } finally {
+      auth.restore();
+    }
+  }),
+);
+
+Deno.test(
+  "GET /admin/webhooks: 403 for non-admin, 200 when ADMIN_DIDS contains caller",
+  withAdminEnv(
+    "did:plc:admin",
+    withTestDb(async () => {
+      const app = createApp();
+
+      const nonAdmin = stubAuthenticate("did:plc:other");
+      try {
+        const res = await requestApp(app, {
+          path: "/admin/webhooks",
+          headers: { Authorization: "Bearer fake" },
+        });
+        assertEquals(res.status, 403);
+      } finally {
+        nonAdmin.restore();
+      }
+
+      const admin = stubAuthenticate("did:plc:admin");
+      try {
+        const res = await requestApp(app, {
+          path: "/admin/webhooks",
+          headers: { Authorization: "Bearer fake" },
+        });
+        assertEquals(res.status, 200);
+        assertEquals(Array.isArray(res.json()), true);
+      } finally {
+        admin.restore();
+      }
+    }),
+  ),
+);
+
+Deno.test(
+  "GET /search: 400 on blank q, forwards discoverableOnly:true to opensearch",
+  withOpenSearchEnv(
+    withTestDb(async () => {
+      const app = createApp();
+
+      const blank = await requestApp(app, { path: "/search?q=" });
+      assertEquals(blank.status, 400);
+
+      const stub = installFetchStub(() => jsonResponse({ hits: { hits: [] } }));
+      try {
+        const res = await requestApp(app, { path: "/search?q=hello" });
+        assertEquals(res.status, 200);
+        assertEquals(stub.calls.length, 1);
+        const sent = JSON.parse(stub.calls[0].body) as {
+          query: { bool: { filter: Record<string, unknown>[] } };
+        };
+        assertEquals(sent.query.bool.filter, [
+          { term: { discoverable: true } },
+        ]);
+      } finally {
+        stub.restore();
+      }
+    }),
+  ),
+);
+
+Deno.test(
+  "OPTIONS preflight returns 204 with CORS headers",
+  withTestDb(async () => {
+    const res = await requestApp(createApp(), {
+      method: "OPTIONS",
+      path: "/notes",
+    });
+    assertEquals(res.status, 204);
+    assertEquals(res.headers.get("Access-Control-Allow-Origin"), "*");
+    assertEquals(
+      res.headers.get("Access-Control-Allow-Methods")?.includes("POST"),
+      true,
+    );
+    assertEquals(
+      res.headers.get("Access-Control-Allow-Headers")?.includes(
+        "Authorization",
+      ),
+      true,
+    );
+  }),
+);