julien/remanso-jetstream
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: julien:main
Branches Tags
julien:main
..
compare: julien:bfa041eb978ffc090d309e55c67e0ef5b38887f4
Branches Tags
julien:main

1 Commits
main ... bfa041eb97

Author SHA1 Message Date
Julien Calixte
bfa041eb97 chore: add dockerignore for scripts, local artifacts, and env files 
The image previously inherited everything from a `COPY . .`, including
.env (secrets), local notes.db copies, and admin scripts that should
not run in prod containers.
 2026-05-05 14:07:35 +02:00
53 changed files with 330 additions and 4268 deletions
Expand all files Collapse all files

3
.dockerignore
View File

@@ -19,9 +19,6 @@ 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

30
.gitea/workflows/ci.yml
View File

@@ -1,30 +0,0 @@
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/

23
.gitlab-ci.yml Normal file
View File

@@ -0,0 +1,23 @@
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

39
README.md
View File

@@ -41,38 +41,13 @@ docker run -p 8080:8080 -v litenote-data:/data litenote-jetstream
## API
| Endpoint | Description |
| ------------------------------------ | ----------------------------------------------------------------------------------------------- |
| `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.
| Endpoint | Description |
| -------------------------------- | ---------------------------------- |
| `GET /notes?cursor=&limit=` | Paginated notes from all users |
| `GET /:did/notes?cursor=&limit=` | Paginated notes for a specific DID |
## 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>`).
| Variable | Default | Description |
| ------------- | ---------- | -------------------------------- |
| `SQLITE_PATH` | `notes.db` | Path to the SQLite database file |

13
deno.json
View File

@@ -1,10 +1,4 @@
{
"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",
@@ -12,15 +6,12 @@
"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",
"backfill:search": "deno run --allow-net --allow-read --allow-write --allow-env --allow-ffi --unstable-ffi scripts/backfill-opensearch.ts"
"webhooks:all": "deno run --allow-net --allow-env scripts/list-all-webhooks.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/testing": "jsr:@std/testing@1"
"@std/assert": "jsr:@std/assert@1"
}
}

53
deno.lock generated
View File

@@ -7,10 +7,8 @@
"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",
@@ -22,9 +20,6 @@
"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"
},
@@ -74,18 +69,12 @@
"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"
},
@@ -119,13 +108,6 @@
"dependencies": [
"jsr:@std/internal@^1.0.12"
]
},
"@std/testing@1.0.19": {
"integrity": "f4236172365b216728dc3cc8b5e80a9f4c33083d1e4ede7613d5b25b4014898e",
"dependencies": [
"jsr:@std/async",
"jsr:@std/data-structures"
]
}
},
"npm": {
@@ -160,17 +142,6 @@
"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": [
@@ -184,31 +155,12 @@
"@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": [
@@ -218,9 +170,6 @@
"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=="
},
@@ -247,8 +196,6 @@
"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"
]
}

30
docker-compose.yml
View File

@@ -6,27 +6,6 @@ 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
@@ -34,10 +13,6 @@ services:
depends_on:
migrate:
condition: service_completed_successfully
opensearch:
condition: service_healthy
environment:
- OPENSEARCH_URL=http://opensearch:9200
volumes:
- ${DATA_VOLUME:-data}:/data
@@ -48,10 +23,6 @@ services:
depends_on:
migrate:
condition: service_completed_successfully
opensearch:
condition: service_healthy
environment:
- OPENSEARCH_URL=http://opensearch:9200
expose:
- "8080"
volumes:
@@ -65,4 +36,3 @@ services:
volumes:
data:
opensearch-data:

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

@@ -1,114 +0,0 @@
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": ""
}
'''
}
}
}

7
docs/Remanso Jetstream API/admin/folder.bru
View File

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

84
docs/Remanso Jetstream API/auth/GitHub OAuth proxy.bru
View File

@@ -1,84 +0,0 @@
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": ""
}
'''
}
}
}

7
docs/Remanso Jetstream API/auth/folder.bru
View File

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

9
docs/Remanso Jetstream API/bruno.json
View File

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

7
docs/Remanso Jetstream API/collection.bru
View File

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

3
docs/Remanso Jetstream API/environments/Local development.bru
View File

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

3
docs/Remanso Jetstream API/environments/Production.bru
View File

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

42
docs/Remanso Jetstream API/meta/Health probe.bru
View File

@@ -1,42 +0,0 @@
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": ""
}
'''
}
}
}

40
docs/Remanso Jetstream API/meta/Hello world.bru
View File

@@ -1,40 +0,0 @@
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
'''
}
}
}

7
docs/Remanso Jetstream API/meta/folder.bru
View File

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

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

@@ -1,61 +0,0 @@
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": ""
}
'''
}
}
}

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

@@ -1,69 +0,0 @@
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": ""
}
'''
}
}
}

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

@@ -1,105 +0,0 @@
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": ""
}
'''
}
}
}

7
docs/Remanso Jetstream API/notes/folder.bru
View File

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

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

@@ -1,195 +0,0 @@
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": ""
}
'''
}
}
}

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

@@ -1,99 +0,0 @@
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": ""
}
'''
}
}
}

7
docs/Remanso Jetstream API/search/folder.bru
View File

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

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

@@ -1,206 +0,0 @@
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": ""
}
'''
}
}
}

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

@@ -1,189 +0,0 @@
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": ""
}
'''
}
}
}

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

@@ -1,115 +0,0 @@
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": ""
}
'''
}
}
}

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

@@ -1,127 +0,0 @@
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": ""
}
'''
}
}
}

7
docs/Remanso Jetstream API/webhooks/folder.bru
View File

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

296
jetstream.ts
View File

@@ -5,150 +5,180 @@ 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";
const safeIndex = async (note: Note): Promise<void> => {
try {
await indexNote(note);
} catch (error) {
log(`[opensearch] index ${note.did}/${note.rkey} failed:`, error);
globalThis.addEventListener("unhandledrejection", (e) => {
log("[jetstream] unhandled rejection:", e.reason);
});
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 safeRemove = async (did: string, rkey: string): Promise<void> => {
try {
await removeNote(did, rkey);
} catch (error) {
log(`[opensearch] delete ${did}/${rkey} failed:`, error);
}
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 bulkDeps: BulkDeps = {
getSubs: getWebhooksByDidAndVerb,
dispatch: dispatchAll,
const BULK_CREATE_DEBOUNCE_MS = 15000;
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}`,
);
};
if (import.meta.main) {
globalThis.addEventListener("unhandledrejection", (e) => {
log("[jetstream] unhandled rejection:", e.reason);
});
globalThis.addEventListener("error", (e) => {
log("[jetstream] uncaught error:", e.error);
});
const cursor = getCursor();
log(`[jetstream] starting with cursor: ${cursor ?? "none"}`);
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) => {
try {
const {
did,
commit: { rkey, record },
} = event;
log(`[jetstream] creating ${did}/${rkey}...`);
const note = record as unknown as Omit<Note, "did" | "rkey">;
upsertNote({ did, rkey, ...note });
log(`[jetstream] create ${did}/${rkey}: ${note.title}`);
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) => {
try {
const {
did,
commit: { rkey, record },
} = event;
log(`[jetstream] updating ${did}/${rkey}...`);
const note = record as unknown as Omit<Note, "did" | "rkey">;
upsertNote({ did, rkey, ...note });
log(`[jetstream] update ${did}/${rkey}: ${note.title}`);
// Updates fold into the `create` verb — subscribers reconcile by (did, rkey).
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) => {
try {
const {
did,
commit: { rkey },
} = event;
log(`[jetstream] deleting ${did}/${rkey}...`);
deleteNote({ did, rkey });
log(`[jetstream] 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", () => {
log("[jetstream] connected");
});
jetstream.on("close", () => {
log("[jetstream] connection closed");
});
jetstream.on("error", (error) => {
log("[jetstream] connection closed with error", error);
});
try {
await ensureIndex();
} catch (error) {
log("[opensearch] ensureIndex failed at startup:", error);
// 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),
});
};
log("[jetstream] launching");
jetstream.start();
const cursor = getCursor();
log(`[jetstream] starting with cursor: ${cursor ?? "none"}`);
setInterval(() => {
if (jetstream.cursor) {
saveCursor(jetstream.cursor);
}
}, 5000);
}
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) => {
try {
const {
did,
commit: { rkey, record },
} = event;
log(`[jetstream] creating ${did}/${rkey}...`);
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 });
} catch (error) {
log(`[jetstream] error on create:`, error);
}
});
jetstream.onUpdate("space.remanso.note", async (event) => {
try {
const {
did,
commit: { rkey, record },
} = event;
log(`[jetstream] updating ${did}/${rkey}...`);
const note = record as unknown as Omit<Note, "did" | "rkey">;
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 });
} catch (error) {
log(`[jetstream] error on update:`, error);
}
});
jetstream.onDelete("space.remanso.note", async (event) => {
try {
const {
did,
commit: { rkey },
} = event;
log(`[jetstream] deleting ${did}/${rkey}...`);
deleteNote({ did, rkey });
log(`[jetstream] delete ${did}/${rkey}`);
await fireWebhooks(did, "delete", { event: "delete", did, rkey });
} catch (error) {
log(`[jetstream] error on delete:`, error);
}
});
jetstream.on("open", () => {
log("[jetstream] connected");
});
jetstream.on("close", () => {
log("[jetstream] connection closed");
});
jetstream.on("error", (error) => {
log("[jetstream] connection closed with error", error);
});
log("[jetstream] launching");
jetstream.start();
setInterval(() => {
if (jetstream.cursor) {
saveCursor(jetstream.cursor);
}
}, 5000);

449
openapi.yaml
View File

@@ -1,449 +0,0 @@
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] }

123
scripts/backfill-opensearch.ts
View File

@@ -1,123 +0,0 @@
// 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();

120
server.ts
View File

@@ -11,7 +11,6 @@ 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 = {
@@ -19,25 +18,13 @@ 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 authenticator(
verifiedDid = await authenticateRequest(
ctx.request.headers.get("Authorization"),
);
} catch {
@@ -47,24 +34,23 @@ const requireDidOwnership = async (
}
if (verifiedDid !== did) {
ctx.response.status = 403;
ctx.response.body = { error: "Forbidden: DID mismatch" };
ctx.response.body = { error: "You can only manage your own webhooks" };
return false;
}
return true;
};
const adminDids = (): Set<string> =>
new Set(
(Deno.env.get("ADMIN_DIDS") ?? "")
.split(",")
.map((d) => d.trim())
.filter(Boolean),
);
const ADMIN_DIDS = 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 authenticator(
verifiedDid = await authenticateRequest(
ctx.request.headers.get("Authorization"),
);
} catch {
@@ -72,7 +58,7 @@ const requireAdmin = async (ctx: AuthCtx): Promise<boolean> => {
ctx.response.body = { error: "Unauthorized" };
return false;
}
if (!adminDids().has(verifiedDid)) {
if (!ADMIN_DIDS.has(verifiedDid)) {
ctx.response.status = 403;
ctx.response.body = { error: "Admin only" };
return false;
@@ -145,43 +131,6 @@ 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 ?? {};
@@ -273,34 +222,27 @@ router.delete("/:did/webhooks/:id", async (ctx) => {
// ctx.response.status = 204;
// })
export const createApp = (): Application => {
const app = new Application();
const app = new Application();
app.use(async (ctx, next) => {
ctx.response.headers.set("Access-Control-Allow-Origin", "*");
ctx.response.headers.set(
"Access-Control-Allow-Methods",
"GET, POST, PUT, DELETE, OPTIONS",
);
ctx.response.headers.set(
"Access-Control-Allow-Headers",
"Content-Type, Authorization",
);
if (ctx.request.method === "OPTIONS") {
ctx.response.status = 204;
return;
}
await next();
});
app.use(async (ctx, next) => {
ctx.response.headers.set("Access-Control-Allow-Origin", "*");
ctx.response.headers.set(
"Access-Control-Allow-Methods",
"GET, POST, PUT, DELETE, OPTIONS",
);
ctx.response.headers.set(
"Access-Control-Allow-Headers",
"Content-Type, Authorization",
);
if (ctx.request.method === "OPTIONS") {
ctx.response.status = 204;
return;
}
await next();
});
app.use(router.routes());
app.use(router.allowedMethods());
app.use(router.routes());
app.use(router.allowedMethods());
return app;
};
if (import.meta.main) {
const app = createApp();
log("[server] listening on port 8080");
app.listen({ port: 8080 });
}
log("[server] listening on port 8080");
app.listen({ port: 8080 });

4
src/auth/verify.ts
View File

@@ -7,14 +7,14 @@ interface DidDocument {
service?: { id: string; serviceEndpoint: string }[];
}
export function decodeJwtPayload(token: string): JwtPayload {
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));
}
export async function resolvePds(did: string): Promise<string> {
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();

71
src/data/db.ts
View File

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

3
src/data/note.ts
View File

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

66
src/jetstream/bulk.ts
View File

@@ -1,66 +0,0 @@
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();
};

49
src/jetstream/webhooks.ts
View File

@@ -1,49 +0,0 @@
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}`);
};

83
src/migrations/apply.ts
View File

@@ -1,83 +0,0 @@
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);
`);
};

85
src/migrations/init.ts
View File

@@ -1,6 +1,83 @@
import { Database } from "@db/sqlite";
import { applyMigrations } from "./apply.ts";
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);
`);
const db = new Database(Deno.env.get("SQLITE_PATH") ?? "notes.db");
applyMigrations(db);
db.close();

258
src/search/opensearch.ts
View File

@@ -1,258 +0,0 @@
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 };
};

49
tests/_helpers/app.ts
View File

@@ -1,49 +0,0 @@
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),
};
};

18
tests/_helpers/auth.ts
View File

@@ -1,18 +0,0 @@
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(),
};
};

60
tests/_helpers/db.ts
View File

@@ -1,60 +0,0 @@
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",
});
};

57
tests/_helpers/fetch_stub.ts
View File

@@ -1,57 +0,0 @@
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 });

118
tests/auth_verify_test.ts
View File

@@ -1,118 +0,0 @@
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();
}
},
);

141
tests/db_test.ts
View File

@@ -1,141 +0,0 @@
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");
}),
);

147
tests/jetstream_bulk_test.ts
View File

@@ -1,147 +0,0 @@
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();
}
});

94
tests/jetstream_webhooks_test.ts
View File

@@ -1,94 +0,0 @@
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();
}
},
);

115
tests/migrations_test.ts
View File

@@ -1,115 +0,0 @@
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();
}
},
);

204
tests/opensearch_test.ts
View File

@@ -1,204 +0,0 @@
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();
}
}),
);

287
tests/server_test.ts
View File

@@ -1,287 +0,0 @@
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,
);
}),
);