From 746367bc587d0b1c2752c2a7fd7cda2e0c9235fb Mon Sep 17 00:00:00 2001 From: Florian <45694132+flo-bit@users.noreply.github.com> Date: Fri, 28 Aug 2026 12:58:43 +0200 Subject: [PATCH] small fixes, update docs --- README.md | 17 +- apps/atmo-rsvp/README.md | 2 - apps/cloudflare-workers/README.md | 2 +- docs/00-getting-started.md | 32 ++ docs/01-client.md | 34 ++ docs/01-indexing.md | 184 ----------- docs/02-configure-and-query.md | 83 +++++ docs/02-querying.md | 195 ----------- docs/03-deploy-cloudflare.md | 101 ++++++ docs/04-feeds.md | 122 ------- docs/09-labels.md | 142 -------- docs/advanced/README.md | 9 + docs/advanced/feeds.md | 41 +++ docs/advanced/labels.md | 37 +++ docs/advanced/outbox.md | 94 ++++++ docs/frameworks/sveltekit-cloudflare.md | 213 ------------ docs/public-services/api-atmo-rsvp.md | 246 -------------- docs/public-services/creating.md | 311 ------------------ docs/public-services/using.md | 164 --------- packages/contrail/README.md | 1 - .../contrail/src/core/change-bootstrap.ts | 22 +- packages/contrail/src/core/change-log.ts | 1 + packages/contrail/src/core/changes.ts | 109 +++++- packages/contrail/src/core/db/schema.ts | 6 + .../contrail/tests/change-bootstrap.test.ts | 40 +++ .../contrail/tests/change-consumers.test.ts | 28 +- packages/contrail/tests/change-log.test.ts | 14 +- 27 files changed, 629 insertions(+), 1621 deletions(-) create mode 100644 docs/00-getting-started.md create mode 100644 docs/01-client.md delete mode 100644 docs/01-indexing.md create mode 100644 docs/02-configure-and-query.md delete mode 100644 docs/02-querying.md create mode 100644 docs/03-deploy-cloudflare.md delete mode 100644 docs/04-feeds.md delete mode 100644 docs/09-labels.md create mode 100644 docs/advanced/README.md create mode 100644 docs/advanced/feeds.md create mode 100644 docs/advanced/labels.md create mode 100644 docs/advanced/outbox.md delete mode 100644 docs/frameworks/sveltekit-cloudflare.md delete mode 100644 docs/public-services/api-atmo-rsvp.md delete mode 100644 docs/public-services/creating.md delete mode 100644 docs/public-services/using.md diff --git a/README.md b/README.md index 7c6c4f0..deaa28d 100644 --- a/README.md +++ b/README.md @@ -133,19 +133,16 @@ import { createSqliteDatabase } from "@atmo-dev/contrail/sqlite"; import { createPostgresDatabase } from "@atmo-dev/contrail/postgres"; ``` -The runnable [`apps/sqlite`](apps/sqlite) example wires the standard backfill CLI to a local SQLite file, including the optional Alluvium base/archive path. See [Indexing](docs/01-indexing.md) for adapter setup and [Querying](docs/02-querying.md) for the query and hydration model. +The runnable [`apps/sqlite`](apps/sqlite) example wires the standard backfill CLI to a local SQLite file, including the optional Alluvium base/archive path. ## Documentation -- [Indexing](docs/01-indexing.md) -- [Querying](docs/02-querying.md) -- [Feeds](docs/04-feeds.md) -- [Labels](docs/09-labels.md) -- Public Contrail services: - - [Creating a service](docs/public-services/creating.md) - - [Using a service](docs/public-services/using.md) - - [Example: api.atmo.rsvp](docs/public-services/api-atmo-rsvp.md) -- [SvelteKit + Cloudflare](docs/frameworks/sveltekit-cloudflare.md) +1. [Get started locally](docs/00-getting-started.md) +2. [Add the typed client](docs/01-client.md) +3. [Configure and query](docs/02-configure-and-query.md) +4. [Deploy to Cloudflare Workers](docs/03-deploy-cloudflare.md) + +See [advanced topics](docs/advanced/README.md) for other runtimes and features. ## Repository layout diff --git a/apps/atmo-rsvp/README.md b/apps/atmo-rsvp/README.md index bdb8cf5..83a75ae 100644 --- a/apps/atmo-rsvp/README.md +++ b/apps/atmo-rsvp/README.md @@ -92,5 +92,3 @@ pnpx @atmo-dev/contrail connect https://api.atmo.rsvp ``` That verifies the anonymous and service-auth contracts, verifies the canonical contract and Lexicon digests, writes a provider lock, installs provider-owned Lexicons, and runs Atcute TypeScript generation. Reconnecting an existing project requires `--update`. - -See [Example: api.atmo.rsvp](../../docs/public-services/api-atmo-rsvp.md) for the complete method, authentication, acquisition, and deployment walkthrough. diff --git a/apps/cloudflare-workers/README.md b/apps/cloudflare-workers/README.md index 114d5cf..7db07bc 100644 --- a/apps/cloudflare-workers/README.md +++ b/apps/cloudflare-workers/README.md @@ -71,4 +71,4 @@ This reports the durable state without exposing account DIDs or raw upstream err - **add a collection:** append to `collections` in `src/contrail.config.ts`; redeploy; `pnpm contrail backfill --remote` to backfill the new one. - **add full-text search:** `searchable: ["field1", "field2"]`, redeploy, no backfill needed (fts indexes repopulate on ingest). -- **add relations / references:** see [indexing docs](../../docs/01-indexing.md). +- **add relations / references:** see [configuration guide](../../docs/02-configure-and-query.md). diff --git a/docs/00-getting-started.md b/docs/00-getting-started.md new file mode 100644 index 0000000..2e5cad0 --- /dev/null +++ b/docs/00-getting-started.md @@ -0,0 +1,32 @@ +# Get started locally + +Run a local AppView for a public AT Protocol collection with Node.js 22.13 or newer. + +Create an empty directory with one file: + +```ts +// contrail.config.ts +export default { + namespace: "com.example", + collections: { + event: { + collection: "community.lexicon.calendar.event", + queryable: { startsAt: { type: "range" } }, + }, + }, +}; +``` + +Then run: + +```bash +pnpx @atmo-dev/contrail dev +``` + +Contrail resolves the Lexicons, backfills existing records, follows new records, and serves a resumable SQLite AppView at `http://127.0.0.1:8787`. + +```bash +curl 'http://127.0.0.1:8787/xrpc/com.example.event.listRecords?limit=10' +``` + +Replace the collection and fields with your own. Next: [add the typed client to your app](./01-client.md). diff --git a/docs/01-client.md b/docs/01-client.md new file mode 100644 index 0000000..e498f93 --- /dev/null +++ b/docs/01-client.md @@ -0,0 +1,34 @@ +# Add the typed client + +With your [local AppView](./00-getting-started.md) set up, add Contrail and Atcute to your application: + +```bash +pnpm add @atmo-dev/contrail @atcute/client @atcute/lexicons +pnpx @atmo-dev/contrail connect ../my-appview +``` + +Point `connect` at the directory containing `contrail.config.ts`. It resolves the source and query Lexicons, then uses Atcute to generate a typed client in `src/contrail/`. + +Use it from your app: + +```ts +import { createLocalContrailClient } from "./contrail/index.js"; + +const contrail = createLocalContrailClient(); +const response = await contrail.get("com.example.event.listRecords", { + params: { + startsAtMin: new Date().toISOString(), + limit: 20, + }, +}); + +if (!response.ok) throw new Error(`Contrail returned ${response.status}`); + +for (const event of response.data.records) { + console.log(event.value.name, event.value.startsAt); +} +``` + +The method name, parameters, and response are all typed from the Lexicons. Re-run `connect` when the AppView config changes. + +Next: [configure filters, sorting, and hydration](./02-configure-and-query.md). To use an existing deployed AppView, pass its HTTPS URL to `contrail connect` instead of a config path. diff --git a/docs/01-indexing.md b/docs/01-indexing.md deleted file mode 100644 index f4117ba..0000000 --- a/docs/01-indexing.md +++ /dev/null @@ -1,184 +0,0 @@ -# Indexing - -Contrail's core job: mirror atproto records into your DB and expose them via XRPC. You describe what to index with a config object; everything else is automatic. - -## Collection shape - -A realistic two-collection example: events and RSVPs. RSVPs point at events via `subject.uri`; events expose per-status RSVP counts. - -```ts -collections: { - event: { - collection: "community.lexicon.calendar.event", // full NSID - queryable: { - mode: {}, // ?mode=online - startsAt: { type: "range" }, // ?startsAtMin=...&startsAtMax=... - }, - searchable: ["name", "description"], // FTS5 / tsvector - relations: { - rsvps: { - collection: "rsvp", // short name of the child collection - groupBy: "status", // field on the child record - groups: { - going: "community.lexicon.calendar.rsvp#going", - interested: "community.lexicon.calendar.rsvp#interested", - }, - }, - }, - }, - rsvp: { - collection: "community.lexicon.calendar.rsvp", - queryable: { status: {} }, - references: { - event: { collection: "event", field: "subject.uri" }, // RSVP's field → event's URI - }, - }, -} -``` - -- **queryable** — string equality or range, exposed as query params. -- **searchable** — FTS5 on D1/Postgres. Not available on `node:sqlite`. -- **relations** — many-to-one with materialized counts. The `event` collection gains `rsvpsCount`, `rsvpsGoingCount`, `rsvpsInterestedCount` columns — filter (`?rsvpsGoingCountMin=10`) and sort (`?sort=rsvpsGoingCount`) on them. Hydrate inline with `?hydrateRsvps=5`. -- **references** — forward lookups from child → parent. `?hydrateEvent=true` on an RSVP query embeds the referenced event record. - -## Backfill (historical data) - -Run once at setup to pull every record that exists today. - -```ts -await contrail.backfillAll({ concurrency: 100 }); // discover + backfill, logs progress -``` - -Under the hood this is two steps you can call separately if you want finer control: - -```ts -await contrail.discover(); // walk relays, register DIDs -await contrail.backfill({ - concurrency: 100, // identity resolution - pdsConcurrency: 20, // active PDS hosts - didsPerPds: 3, // accounts per active PDS -}); -``` - -`backfill()` picks up each account/collection at its saved PDS cursor. A row is marked complete only after the PDS listing reaches its end. Timeouts, failed identity resolution, `429`, and `5xx` responses leave the row pending with its last error. - -Each initial invocation attempts a failed account once by default, then gets out of the way. Failed rows retain their cursors and receive an exponential `next_retry_at`. Scheduled retries start at 15 minutes, double to a maximum of 48 hours, and stop after ten failed scheduled attempts. Cloudflare's scheduled Worker retries a small due slice after each live-ingest cycle; an explicit later invocation resets exhausted rows and forces another pass. `backfillAll()` returns a durable `status` summary alongside the number of discovered accounts and accepted records. - -Historical loading writes canonical records first, then rebuilds FTS and materialized relation counts with set-based SQL. A durable dirty marker keeps status `incomplete` if the process stops between those phases; the next manual or scheduled backfill repairs the projections before reporting readiness. Live ingestion and scheduled account retries continue maintaining both projections incrementally. - -### Workers CLI - -For Cloudflare Workers deploys, `@atmo-dev/contrail` ships a `contrail` bin that handles the `wrangler.getPlatformProxy` dance — no script file, no package.json alias needed: - -```bash -pnpm contrail backfill # local D1 (wrangler dev's bindings) -pnpm contrail backfill --remote # production D1 -``` - -Auto-detects configs at `contrail.config.ts`, `src/contrail.config.ts`, `src/lib/contrail.config.ts`, or `app/contrail.config.ts` (first match wins). Override with `--config `. Other flags include `--binding ` (default `DB`), `--concurrency ` for identity resolution (default 100), `--pds-concurrency ` (default 20), `--dids-per-pds ` (default 3), and `--max-attempts ` (default 1). Once every known account has either completed or received a deferred failure, the initial pass is complete and scheduled retries continue in the background. Interrupted or undiscovered work still reports the pass as incomplete. - -If you'd rather embed backfill inside your own script, `@atmo-dev/contrail/workers` exports the same logic as a function: - -```ts -import { backfillAll } from "@atmo-dev/contrail/workers"; -import { config } from "../src/contrail.config"; - -await backfillAll({ config, remote: process.argv.includes("--remote") }); -``` - -For node/postgres deploys, skip both — you already have a `db` in hand; just `await contrail.backfillAll({}, db)` directly. - -## Ingestion (ongoing new records) - -After the initial `backfillAll()`, keep the index fresh with new records as they're published. Pick the mode that matches your runtime. - -### Cron-driven (cloudflare workers) - -Workers can't hold long-lived connections, so run one catch-up cycle per cron fire: - -```ts -// wrangler.jsonc: "triggers": { "crons": ["*/1 * * * *"] } -async scheduled(_ev, env, ctx) { - ctx.waitUntil((async () => { - await contrail.ingest({}, env.DB); - await contrail.retryBackfill({}, env.DB); // small due slice - })()); -} -``` - -`ingest()` connects to Jetstream, streams events since the saved cursor, stops when caught up. Running every minute is fine — the next fire resumes where this one left off. Each cycle is bounded, so it can't blow past the Worker time limit. - -**Local dev:** wrangler's cron scheduler only runs in deployed production. For local dev use `pnpm contrail dev` — it runs `wrangler dev --test-scheduled`, fires `/__scheduled` on your configured cron interval, and offers to start or resume backfill whenever known work remains. - -### Persistent (node / any long-lived server) - -If your runtime can keep a socket open, skip the cron entirely: - -```ts -const ac = new AbortController(); -await contrail.runPersistent({ - batchSize: 50, // flush every N events (default: 50) - flushIntervalMs: 5000, // or every N ms, whichever first - signal: ac.signal, -}); -// ac.abort() flushes the current batch and saves the cursor before returning -``` - -One process, one socket, auto-reconnect on drops. Lower latency than cron mode (events land within seconds instead of up-to-a-minute), but needs a runtime that can run indefinitely. - -### Immediate (`notify()`) - -Use this when your own app writes to a user's PDS and needs the change indexed *now* — waiting for the next cron / Jetstream flush is too slow: - -```ts -await contrail.notify(uri); // one record -await contrail.notify([u1, u2, u3]); // batch, up to 25 -``` - -Fetches directly from the user's PDS and indexes synchronously. When Jetstream later delivers the same event, the duplicate is detected by CID and skipped. - -### Which one do I use? - -| | backfillAll | ingest | runPersistent | notify | -|---|---|---|---|---| -| when | once, at setup | every cron fire | start once, runs forever | per-write, on demand | -| runtime | local script | cloudflare workers | node / long-lived server | anywhere | -| scope | all historical records | events since last cursor | events since last cursor, live | specific URIs | -| latency | — | ~minute | ~seconds | immediate | - -Typical combos: -- **workers app:** `backfillAll()` once + `ingest()` on cron + optional `notify()` for self-writes -- **node server:** `backfillAll()` once + `runPersistent()` forever + optional `notify()` for self-writes - -## Recovery after an outage - -Normal ingestion resumes from its saved Jetstream cursor, so a short outage needs no special command: restart `ingest()` or `runPersistent()` and let it catch up. - -Contrail does not perform a full PDS sweep as a repair mechanism. Such a sweep is expensive, cannot discover repositories it never knew about, and cannot safely infer remote deletions after partial failures. If the saved cursor is older than the source's retained history, rebuild into a fresh database with `backfillAll()` rather than trusting a partial reconciliation. A replay-capable source and first-class projection rebuild command are planned follow-up work. - -Reading the indexed data — filters, sorts, hydration, search, pagination — has its own doc: [Querying](./02-querying.md). - -## Adapters - -| Adapter | Use when | FTS | -|---|---|---| -| Cloudflare D1 | Workers | ✅ | -| `@atmo-dev/contrail/sqlite` | Node 22+ local dev | ❌ | -| `@atmo-dev/contrail/postgres` | Node server | ✅ | - -```ts -import { createPostgresDatabase } from "@atmo-dev/contrail/postgres"; -const db = createPostgresDatabase(pool); -``` - -## Top-level config - -| Key | Default | | -|---|---|---| -| `namespace` | — | Reverse-domain for XRPC paths | -| `profiles` | `["app.bsky.actor.profile"]` | Profile NSIDs, auto-hydrated via `?profiles=true` | -| `jetstreams` | Bluesky | Jetstream URLs | -| `relays` | Bluesky | Relay URLs for discovery | -| `notify` | off | Prefer an in-process call or a secret string requiring `Bearer`; open `true` mode is not recommended | -| `feeds` | — | See [Feeds](./04-feeds.md) | -| `labels` | — | See [Labels](./09-labels.md) | diff --git a/docs/02-configure-and-query.md b/docs/02-configure-and-query.md new file mode 100644 index 0000000..2b5e27f --- /dev/null +++ b/docs/02-configure-and-query.md @@ -0,0 +1,83 @@ +# Configure and query + +Each entry in `collections` becomes typed `getRecord` and `listRecords` methods. Declare only the fields and relationships your app needs: + +```ts +// contrail.config.ts +export default { + namespace: "com.example", + collections: { + event: { + collection: "community.lexicon.calendar.event", + queryable: { + mode: {}, + startsAt: { type: "range" }, + }, + searchable: ["name", "description"], + relations: { + rsvps: { + collection: "rsvp", + groupBy: "status", + groups: { + going: "community.lexicon.calendar.rsvp#going", + }, + }, + }, + }, + rsvp: { + collection: "community.lexicon.calendar.rsvp", + queryable: { + status: {}, + "subject.uri": {}, + }, + references: { + event: { collection: "event", field: "subject.uri" }, + }, + }, + }, +}; +``` + +This produces the following client parameters: + +| Config | Query parameter | +|---|---| +| `mode: {}` | `mode` | +| `startsAt: { type: "range" }` | `startsAtMin`, `startsAtMax` | +| `searchable` | `search` | +| `relations.rsvps` | `rsvpsCountMin`, `hydrateRsvps` | +| `groups.going` | `rsvpsGoingCountMin` | +| `references.event` | `hydrateEvent` | + +Dotted fields become camel case: `subject.uri` becomes `subjectUri`. Every list method also supports `actor` (a DID or handle), `sort`, `order`, `limit`, `cursor`, and `profiles`. + +## Query from the client + +After changing the config, re-run `contrail connect` in your app. The generated Atcute client now knows the new parameters and response types: + +```ts +const response = await contrail.get("com.example.event.listRecords", { + params: { + mode: "in-person", + startsAtMin: new Date().toISOString(), + rsvpsGoingCountMin: 5, + sort: "startsAt", + order: "asc", + hydrateRsvps: 3, + profiles: true, + limit: 20, + }, +}); + +if (!response.ok) throw new Error(`Contrail returned ${response.status}`); + +const { records, profiles, cursor } = response.data; +``` + +Every record has `uri`, `cid`, and its original record body in `value`. Hydrated relations and references are added to that record; requested profiles are returned once in the top-level `profiles` array. + +Pass the returned opaque `cursor` into the same query to get the next page. `limit` defaults to 50 and may be 1–200. + +Full-text `search` works with D1 and PostgreSQL. The zero-config local SQLite AppView does not provide full-text search. + +Next: [deploy to Cloudflare Workers](./03-deploy-cloudflare.md). diff --git a/docs/02-querying.md b/docs/02-querying.md deleted file mode 100644 index 46dac29..0000000 --- a/docs/02-querying.md +++ /dev/null @@ -1,195 +0,0 @@ -# Querying - -Once [indexing](./01-indexing.md) is set up, every collection you declared gets a pair of XRPC endpoints under `/xrpc/{namespace}.{short}.*`: - -| Endpoint | Returns | -|---|---| -| `{namespace}.{short}.listRecords` | Paginated list with filters, sorts, hydration | -| `{namespace}.{short}.getRecord?uri=…` | Single record by AT-URI | - -Top-level methods include `{namespace}.getProfile`, `{namespace}.getCursor`, `{namespace}.notifyOfUpdate`, and optionally `{namespace}.getFeed` and `{namespace}.lexicons`. - -## HTTP (what most callers use) - -Every config field becomes a predictable URL param: - -``` -/xrpc/com.example.event.listRecords?mode=online&startsAtMin=2026-01-01&rsvpsGoingCountMin=10&sort=startsAt&order=asc&hydrateRsvps=5 -/xrpc/com.example.event.getRecord?uri=at://did:plc:.../...&hydrateRsvps=5 -``` - -| Config produces | URL param | -|---|---| -| `queryable: { field: {} }` | `?field=value` (equality) | -| `queryable: { field: { type: "range" } }` | `?fieldMin=…`, `?fieldMax=…` | -| `relations: { rel: {...} }` | `?relCountMin=N`, `?sort=relCount`, `?hydrateRel=N` | -| `relations: { rel: { groups: { going } } }` | `?relGoingCountMin=N`, `?sort=relGoingCount` | -| `references: { ref: {...} }` | `?hydrateRef=true` | - -Dotted field names become camelCase params — `queryable: { "subject.uri": {} }` → `?subjectUri=…`. - -## Operational status - -`GET /status` returns the current JSON overview. It includes indexed record totals, live-ingest freshness, and durable backfill state: - -- discovery source progress; -- mutually exclusive account totals for `complete`, `pending`, `retrying`, and `failed`; -- known-account completion percentage; -- the same mutually exclusive totals per collection; and -- scheduled/due account retries plus the next retry time. - -`state` is `running` while a manual or scheduled slice holds the backfill lease. It becomes `complete` after discovery and the initial pass finish, even when account-level `retrying` or `failed` counts are non-zero. `pending` means no failure has occurred yet, `retrying` means at least one attempt failed but automatic attempts remain, and `failed` means the ten-attempt automatic budget is exhausted. An explicit backfill resets that budget. `incomplete` means discovery or an initial account attempt has not finished. - -"Known" is deliberate: while relay discovery is incomplete, Contrail cannot honestly claim how many accounts remain undiscovered. `/health` remains a lightweight liveness response and does not claim that historical backfill is complete. - -`GET /xrpc/{namespace}.getCursor` returns the committed primary ordered-source position when `orderedSource` is configured. The `{ source, epoch, cursor }` tuple is opaque: compare complete tuples for equality only, and treat a source or epoch change as a full reset. A consumer that needs a stable query snapshot can read the position before and after its query and retry when the two positions differ. - -## Programmatic - -```ts -const { records, cursor } = await contrail.query("event", { - filters: { mode: "online" }, - rangeFilters: { startsAt: { min: "2026-01-01" } }, - countFilters: { rsvp: 10 }, // keyed by child collection short name - sort: { recordField: "startsAt", direction: "asc" }, - limit: 20, -}); -``` - -The programmatic shape doesn't use the URL param names — keys are the underlying field/collection identifiers: - -- `filters` / `rangeFilters` — keyed by the field name from your config (`startsAt`, `subject.uri`), not the camelCased URL param. -- `countFilters` — keyed by the target collection's short name for totals, or by the full `nsid#group` token for group counts. E.g., `{ rsvp: 10 }` for "at least 10 RSVPs total," or `{ "community.lexicon.calendar.rsvp#going": 10 }` for "at least 10 going." -- `sort` — `{ recordField, direction }` for field sorts, `{ countType, direction }` for count sorts (where `countType` is the same collection-short-name or `nsid#group` as above). Field sorts preserve the SQL value type, and records whose field is missing or `null` sort last in either direction. - -For count filters / sorts, the HTTP side is nicer than the programmatic side — consider going through `createHandler` + `fetch` even for in-process calls if you want the friendly names. Or use `createServerClient` from `@atmo-dev/contrail/server` for a typed XRPC client that runs in-process (no fetch roundtrip). - -## Pagination - -``` -?limit=25&cursor= -``` - -`cursor` is opaque — pass back whatever `listRecords` returned in its `cursor` field. `limit` is 1–200 (default 50). Cursors embed the complete ordering, including relevance rank for search and URI as the final unique tiebreaker. They also embed the sort kind, so a cursor from a `sort=startsAt` query is ignored by a `sort=rsvpsCount` query instead of silently returning wrong results. - -```ts -let cursor: string | undefined; -do { - const page = await contrail.query("event", { limit: 100, cursor }); - // process page.records - cursor = page.cursor; -} while (cursor); -``` - -## Hydration - -Each record response is a flat shape: - -```jsonc -{ - "uri": "at://did:plc:.../community.lexicon.calendar.event/...", - "cid": "...", - "value": { "name": "Rust meetup", "startsAt": "2026-03-16T...", ... }, - "rsvpsCount": 42, // from relations - "rsvpsGoingCount": 30, - // relations + references appear here only when hydrated -} -``` - -The `value` field carries the record body — same shape as atproto's `com.atproto.repo.listRecords#record`. `did`, `collection`, `rkey`, and `time_us` are also returned alongside as optional extras. - -### `?hydrateRel=N` (relations) - -Embeds the latest N child records per group, inline under the parent: - -``` -/xrpc/com.example.event.listRecords?hydrateRsvps=5 -``` - -Returns: - -```jsonc -{ - "records": [{ - "uri": "at://.../event/...", - "value": { "name": "..." }, - "rsvpsCount": 42, - "rsvps": { - "going": [ {uri, cid, value}, ... 5 items ], - "interested":[ {uri, cid, value}, ... 5 items ] - } - }] -} -``` - -Max 50 per group. For grouped relations you get one array per group value; for ungrouped relations just a flat array. - -### `?hydrateRef=true` (references) - -Embeds the single referenced parent record — useful for RSVP lists that need to show event details: - -``` -/xrpc/com.example.rsvp.listRecords?subjectUri=at://.../event/...&hydrateEvent=true -``` - -Each RSVP record in the response gains an `event: {uri, cid, value}` field. - -### `?profiles=true` - -Opt in to profile + handle hydration for every DID referenced in the result: - -``` -/xrpc/com.example.event.listRecords?profiles=true -``` - -Response grows a top-level `profiles` array, one entry per (DID, configured profile NSID): - -```jsonc -{ - "records": [...], - "profiles": [ - { - "did": "did:plc:alice...", - "handle": "alice.bsky.social", - "uri": "at://did:plc:alice.../app.bsky.actor.profile/self", - "cid": "...", - "collection": "app.bsky.actor.profile", - "rkey": "self", - "value": { /* profile record body */ } - } - ] -} -``` - -A DID with no profile record (or whose handle resolved but profile didn't) shows up as a bare `{ did, handle }` entry — `uri`/`cid`/`value` are omitted. With multiple profile NSIDs configured, you'll see one entry per (DID × NSID) that resolved. - -Which profile NSID(s) to hydrate from is configured at the top level of Contrail's config (`profiles`, defaults to `["app.bsky.actor.profile"]`). - -## Full-text search - -``` -?search=meetup -?search=meetup* -?search="rust meetup" -?search=rust OR typescript -``` - -Combinable with every other filter and sort. Backed by SQLite FTS5 (D1) or Postgres tsvector (Postgres adapter). Not available on `node:sqlite` — that adapter doesn't ship FTS5. - -When searching, results are ranked by relevance by default. Override with an explicit `sort` param. - -## Examples - -``` -# Upcoming events with 10+ going RSVPs, with RSVP records + profiles -?startsAtMin=2026-03-16&rsvpsGoingCountMin=10&hydrateRsvps=5&profiles=true - -# Events for a specific user (by handle — triggers on-demand backfill) -?actor=alice.bsky.social&profiles=true - -# RSVPs for one event, with the event record embedded -?subjectUri=at://did:plc:.../event/...&hydrateEvent=true&profiles=true - -# Search + filter + sort -?search=meetup&mode=online&sort=startsAt&order=asc -``` diff --git a/docs/03-deploy-cloudflare.md b/docs/03-deploy-cloudflare.md new file mode 100644 index 0000000..5252755 --- /dev/null +++ b/docs/03-deploy-cloudflare.md @@ -0,0 +1,101 @@ +# Deploy to Cloudflare Workers + +Turn the same local AppView into a public Worker backed by D1. + +## Install + +From the directory containing `contrail.config.ts`: + +```bash +pnpm init +pnpm add @atmo-dev/contrail +pnpm add -D @atcute/lex-cli wrangler typescript +``` + +Add an ordered source to the config. Keep its epoch stable for the lifetime of this database: + +```ts +// contrail.config.ts +import type { ContrailConfig } from "@atmo-dev/contrail"; + +const config: ContrailConfig = { + namespace: "com.example", + orderedSource: { + source: "jetstream", + epoch: "my-appview-v1", + }, + collections: { + // ...your existing collections + }, +}; + +export default config; +``` + +Generate the public query Lexicons and their referenced record Lexicons: + +```bash +pnpm contrail lexicons all --public +``` + +## Create the Worker + +```ts +// worker.ts +import { createWorker } from "@atmo-dev/contrail/worker"; +import config from "./contrail.config"; +import { lexicons } from "./lexicons/generated"; + +export default createWorker(config, { + lexicons, + publicService: { + endpoint: "https://my-appview.example.com", + }, +}); +``` + +Use the Worker's actual `workers.dev` or custom-domain URL as `endpoint`. + +Create `wrangler.jsonc`: + +```jsonc +{ + "name": "my-appview", + "main": "worker.ts", + "compatibility_date": "2025-12-25", + "observability": { "enabled": true }, + "d1_databases": [ + { + "binding": "DB", + "database_name": "my-appview", + "database_id": "PASTE_DATABASE_ID_HERE" + } + ], + "triggers": { "crons": ["*/1 * * * *"] } +} +``` + +## Deploy and backfill + +```bash +pnpm wrangler d1 create my-appview # copy its ID into wrangler.jsonc +pnpm wrangler deploy +pnpm contrail backfill --remote +``` + +The one-minute cron keeps the AppView current. Check it with: + +```bash +curl https://my-appview.example.com/status +curl 'https://my-appview.example.com/xrpc/com.example.event.listRecords?limit=10' +``` + +Finally, point the application from the previous guide at the deployment: + +```bash +pnpx @atmo-dev/contrail connect https://my-appview.example.com +``` + +When the config changes, regenerate the Lexicons, deploy, backfill any new collections, and reconnect the client with `--update`. + +For optional outbox deliveries, feeds, and labels, see [advanced topics](./advanced/README.md). diff --git a/docs/04-feeds.md b/docs/04-feeds.md deleted file mode 100644 index 0f5c5e1..0000000 --- a/docs/04-feeds.md +++ /dev/null @@ -1,122 +0,0 @@ -# Feeds - -Personalized "what the people I follow are doing" timelines, fanned out at write time. Opt-in; no cost if you don't enable it. - -## Mental model - -> A feed is a (follow-collection, [target-collections]) pair, named by you. Every time someone an *actor* follows posts to a target collection, contrail inserts one row into `feed_items` for that actor. - -Reading a feed is a join through `feed_items` plus the standard pipeline (filters, sorts, hydration, references). The actor parameter on a read is *whose feed* you want — there is no anonymous feed read. - -## Enable - -```ts -import type { ContrailConfig } from "@atmo-dev/contrail"; - -const config: ContrailConfig = { - namespace: "com.example", - collections: { - follow: { collection: "app.bsky.graph.follow" }, - post: { collection: "app.bsky.feed.post", queryable: { /* ... */ } }, - }, - feeds: { - timeline: { - follow: "follow", // short name (key in `collections`), NOT the NSID - targets: ["post"], - maxItems: 500, // optional, default 200 - }, - }, -}; -``` - -Both the follow collection and every target collection must be declared in `collections`. Names in `feeds` are the **short names** (the keys of `collections`), not NSIDs. Config validation throws if you reference an unknown short name. - -## Follow-record shape - -The follow collection's record must have a `subject` field at the top level whose value is the followed DID. `app.bsky.graph.follow` matches this naturally: - -```json -{ "subject": "did:plc:abc...", "createdAt": "2026-01-01T00:00:00Z" } -``` - -Custom follow lexicons work as long as `subject` is the followed DID at JSON path `$.subject`. Contrail extracts via that path during ingest fan-out and during follow-event backfill. - -## Schema - -Two tables, one shared across all feeds: - -| Table | Purpose | -|---|---| -| `feed_items (actor, uri, collection, time_us)` | One row per (viewer, target record). Primary key `(actor, uri)` so a single target record can appear in many feeds. | -| `feed_backfills (actor, feed, completed)` | Marker so first-read backfill only runs once per (actor, feed). | - -Indexes: `(actor, collection, time_us DESC)` and `(actor, time_us DESC)` on `feed_items`, plus a JSON `subject` index on each follow collection's records table for the fan-out join. - -## Read - -``` -GET /xrpc/{namespace}.getFeed?feed=timeline&actor=&limit=50 -``` - -| Param | Meaning | -|---|---| -| `feed` | Feed name from `config.feeds` (required) | -| `actor` | Whose feed — DID or handle (required) | -| `collection` | Restrict to one target collection's short name (default: first in `targets`) | -| `limit`, `cursor`, filters from the target's `queryable`, hydration flags, sort/order | Same as `listRecords` on the target collection | - -The `actor` parameter is **whose feed** you're reading, not a filter on record creator. Feeds are always per-user. - -```ts -const feed = await fetch( - `/xrpc/com.example.getFeed?feed=timeline&actor=${did}&limit=50&profiles=true` -).then((r) => r.json()); -// feed.records — target records by users `actor` follows, newest first -// feed.profiles — hydrated profile records for record authors -``` - -## How fan-out works - -Three moments: - -1. **A target write.** Someone followed by N actors posts to a target collection. Contrail inserts N `feed_items` rows in one statement (`INSERT … SELECT … FROM WHERE subject = ?`). Cost is linear in N — there is no max-followers cap; a viral author with 1M followers is 1M inserts. - -2. **A follow write.** An actor follows a new user. Contrail backfills the most recent **100** target records from that user into the new follower's feed. The 100 is hardcoded in `core/router/feed.ts` — separate from the per-feed `maxItems` cap, and not tunable per feed today. - -3. **First read for an (actor, feed) pair.** Contrail backfills the actor's follow records from their PDS (so their `feed_items` rows can be computed), then populates `feed_items` from existing target records by users they already follow. Marked complete in `feed_backfills` so it runs once per pair. - -## Pruning - -Feeds are capped: each actor keeps at most `maxItems` rows per target collection (default 200, newest first). Older rows past the cap are deleted by a background cleanup that piggybacks on ingestion — there is no separate prune job. - -A few terms used below: - -- **Tick** — one cycle of the ingest loop. In cron mode the worker wakes on a schedule (e.g. once a minute) and each wake-up is a tick; in the persistent loop it's each batch flush. -- **Sweep** — the cleanup that walks `feed_items` actor by actor and deletes whatever is over an actor's cap. -- **Slice** — a sweep doesn't scan the whole table at once. Each tick it handles a chunk of up to `FEED_PRUNE_SWEEP_ACTORS` actors (default 500). That chunk is one slice. -- **Cursor / full pass** — a bookmark for the last actor a slice stopped on, so the next slice resumes after it instead of restarting. When the cursor reaches the last actor it *wraps* back to the start; one start-to-end trip is a *full pass*. - -**When the sweep runs.** A feed can only go over its cap right after a feed-mutating record (a target fan-out or a follow backfill) is applied, so the sweep is skipped entirely on ticks that ingested nothing feed-relevant. It runs when the current tick — a cron run, a persistent-loop flush, or a `notifyOfUpdate` call — applied a feed-mutating record. As a safety net it also runs on a recovery interval (`FEED_PRUNE_RECOVERY_INTERVAL_MS`, 6h), so rows that went over cap without a fresh ingest (a lowered cap, a bulk import) still get cleaned up — including on a stream that is otherwise idle. - -Doing one slice per tick keeps each tick's cost flat no matter how big the table grows. The recovery timer measures from the last *completed full pass* (not the last slice): a fresh pass becomes due one recovery interval after the previous one finished, then advances a slice per tick until the cursor wraps. So a full pass *completes* roughly every `recovery interval + lap time`, where lap time is `ceil(actors / FEED_PRUNE_SWEEP_ACTORS)` ticks — e.g. with 100k actors and one-minute cron ticks, ~6h + ~3h20m. That keeps the whole table draining on a bounded cadence; it is not a hard "fully clean every 6h" guarantee. Raise `FEED_PRUNE_SWEEP_ACTORS` if you need the lap time shorter at large actor counts. - -**Fan-out isn't cleaned up instantly.** A slice cleans up whatever actors the cursor lands on next — not specifically the actors whose feeds just changed. So when a popular author posts and fans out to many followers: a follower the cursor *hasn't reached yet* this pass is trimmed later in the same pass (soon), but a follower the cursor has *already passed* waits for the next pass — and on a quiet stream the next pass only starts on the recovery interval. So the worst case for an over-cap follower is roughly one recovery interval (`FEED_PRUNE_RECOVERY_INTERVAL_MS`, 6h), not the next tick. - -This is on purpose: an author can have unboundedly many followers, and trimming every one on the spot would either overrun the per-tick request budget (one delete per follower) or overrun D1's per-query CPU limit (one big delete over all of them, which can reset the shared Durable Object). `feed_items` is just a cache, so a follower sitting a little over cap for up to an interval does no harm. Deployments with fewer than `FEED_PRUNE_SWEEP_ACTORS` (500) distinct feed actors clean the whole table on every triggered tick, so they never see this lag at all. Pruning the touched actors directly (instead of the rolling cursor) would remove the lag but trade the bounded per-tick cost for cost proportional to fan-out size; see the issue tracker for that trade-off. - -## Deletes - -Deleting a target record removes its `feed_items` rows across all actors. Deleting a follow record currently does not retroactively prune the feed_items inserted during the original follow backfill — they age out via the global pruner instead. - -## XRPCs - -- `{namespace}.getFeed` — read - -That's it. Feeds are read-only over XRPC; writes to follow / target collections happen through `com.atproto.repo.putRecord` on the user's PDS as normal, and Jetstream ingestion drives the fan-out. - -## What's not here - -- No per-feed prune cap; the global pruner uses the largest `maxItems` across all feeds. -- The 100-record backfill on a new follow is hardcoded — not tunable per feed. -- No max-followers cap on target writes — a target record by a user with 1M followers means 1M `feed_items` inserts. For apps expecting that scale, partition feeds or rate-limit upstream. -- Feeds contain only public target records. diff --git a/docs/09-labels.md b/docs/09-labels.md deleted file mode 100644 index d34b702..0000000 --- a/docs/09-labels.md +++ /dev/null @@ -1,142 +0,0 @@ -# Labels - -Atproto-native moderation hydration. Subscribe to one or more labelers, index their labels, and attach them to records and profiles in your XRPC responses. Opt-in; zero cost if you don't enable it. - -## Mental model - -> A **label** is a `(src, uri, val)` triple authored by a labeler DID. A **labeler** is a regular atproto account that publishes signed annotations about other accounts and records via `com.atproto.label.subscribeLabels`. - -- One contrail deployment can subscribe to many labelers. -- The caller of your XRPC picks which subset to honor per request via the `atproto-accept-labelers` header (or `?labelers=` query param when headers are awkward — SSE/WS). -- Labels hydrate onto every `listRecords`, `getRecord`, `getProfile`, and `?profiles=true` response without changing your collection config. -- This module only consumes labels. Producing them — your appview emitting its own labels — is a separate question. See *Future work* below. - -## Enable - -```ts -import type { ContrailConfig } from "@atmo-dev/contrail"; - -const config: ContrailConfig = { - namespace: "com.example", - collections: { /* ... */ }, - labels: { - sources: [ - { did: "did:plc:ar7c4by46qjdydhdevvrndac" }, // bsky moderation - { did: "did:plc:newsmast" }, - ], - }, -}; -``` - -`initSchema` creates a `labels` table and a `labeler_cursors` table. Both live on the main DB; nothing per-collection. - -## Caller selection - -Per request, contrail picks accepted labelers in this order: - -1. `atproto-accept-labelers: did:plc:a, did:plc:b` — the spec's HTTP header. -2. `?labelers=did:plc:a,did:plc:b` — fallback for transports that can't set headers easily. -3. `config.labels.defaults` — operator policy. -4. Every entry in `config.labels.sources`. - -The list is intersected with what's actually configured (unknowns dropped — only labelers we've subscribed to have rows to hydrate from) and capped at `maxPerRequest` (default 20). Contrail echoes the applied set back via `atproto-content-labelers`. - -``` -GET /xrpc/com.example.event.listRecords - atproto-accept-labelers: did:plc:ar7c4by46qjdydhdevvrndac -``` - -→ - -```jsonc -// Response: atproto-content-labelers: did:plc:ar7c4by46qjdydhdevvrndac -{ - "records": [ - { - "uri": "at://did:plc:.../com.example.event/...", - "value": { /* ... */ }, - "labels": [ - { - "src": "did:plc:ar7c4by46qjdydhdevvrndac", - "uri": "at://did:plc:.../com.example.event/...", - "val": "spam", - "cts": "2026-04-25T00:00:00.000Z" - } - ] - } - ] -} -``` - -`labels` matches `com.atproto.label.defs#label` field-for-field — pass it straight to atproto SDK moderation helpers. - -### `defaults: []` - -Set defaults to an empty array if you want strict opt-in: callers that send no header / param see no labels at all. - -## Hydration semantics - -For each `(src, uri, val)` tuple visible to the caller, hydration picks the row with the highest `cts`. If that row has `neg=true`, the label is treated as retracted and dropped. Expired rows (`exp` past `now`) are filtered at the SQL level. CID-pinned labels apply only when the indexed record's CID matches. - -Account-level labels (subject = bare DID) hydrate onto profiles. They appear inside each `ProfileEntry.labels` of the `profiles` array on `?profiles=true` responses, and on `getProfile`. - -## Ingestion - -`com.atproto.label.subscribeLabels` is a per-labeler WebSocket firehose with a CBOR frame envelope. Contrail mirrors its existing Jetstream pipeline: - -| Mode | Function | When | -|---|---|---| -| Cron-driven | `contrail.ingestLabels()` | Cloudflare Workers — one drain per cron tick | -| Persistent | `contrail.runPersistentLabels()` | Node / long-lived servers — one socket per labeler, auto-reconnect | -| One-shot backfill | `pnpm contrail backfill --only labels [--remote]` | Local script, drains until each labeler reports caught up. (`pnpm contrail backfill` runs both records and labels.) | - -When `config.labels` is set, the bundled `createWorker` already calls `ingestLabels()` from `scheduled()` alongside `ingest()` — no boilerplate. - -```ts -// node / long-lived -const ac = new AbortController(); -await Promise.all([ - contrail.runPersistent({ signal: ac.signal }), - contrail.runPersistentLabels({ signal: ac.signal }), -]); -``` - -Per-labeler cursors live in `labeler_cursors` (`{did, cursor, endpoint, resolved_at}`). Endpoints are resolved from the DID doc's `service[id="#atproto_labeler"]` and cached for 6h. On `#info { name: "OutdatedCursor" }` frames, contrail resets the cursor to `0` so the next cycle re-backfills. - -### `backfill: false` - -Per source. Default: backfill from `cursor=0` on first sight. Set `false` to start at "now" — useful for very chatty labelers where you don't need history. - -```ts -labels: { - sources: [{ did: "did:plc:somenoisylabeler", backfill: false }], -} -``` - -## Storage - -```sql -CREATE TABLE labels ( - src TEXT NOT NULL, -- labeler DID - uri TEXT NOT NULL, -- subject: at://... or did:... - val TEXT NOT NULL, -- label value - cid TEXT, -- optional record-version pin - neg INTEGER NOT NULL DEFAULT 0, - exp INTEGER, -- expiry, unix sec - cts INTEGER NOT NULL, -- creation time, unix sec - sig BLOB, -- signature bytes (stored, not verified in v1) - PRIMARY KEY (src, uri, val, cts) -); -``` - -The PK includes `cts`, so a `neg=true` retraction is a *new row* that replaces the previous decision via the read-time collapse rule above — never an in-place mutation. This matches the spec, tolerates out-of-order delivery, and survives a labeler that flip-flops. - -## What's not here - -- **Signature verification.** `sig` is stored if the labeler supplies it, but contrail does not verify it in v1. Document as TODO; most appviews skip it. -- **Outbound `subscribeLabels`.** Contrail consumes labels but does not act as a labeler or republish them. -- **Label definitions / preferences UX.** Custom label names, blur behaviors, severity, and per-user preference state belong on the *client*, fetched directly from each labeler. Contrail intentionally stays out of this. - -## Design - -Follows the [atproto label spec](https://atproto.com/specs/label) literally. The wire format on responses matches `com.atproto.label.defs#label` so existing atproto SDKs can consume it directly. Storage is the data model normalized into rows; ingestion mirrors Jetstream both in code shape and in operator UX. diff --git a/docs/advanced/README.md b/docs/advanced/README.md new file mode 100644 index 0000000..49e9282 --- /dev/null +++ b/docs/advanced/README.md @@ -0,0 +1,9 @@ +# Advanced topics + +Start with the four-page happy path: [local AppView](../00-getting-started.md), [typed client](../01-client.md), [configuration and queries](../02-configure-and-query.md), and [Cloudflare deployment](../03-deploy-cloudflare.md). + +Optional features: + +- [Outbox](./outbox.md) +- [Feeds](./feeds.md) +- [Labels](./labels.md) diff --git a/docs/advanced/feeds.md b/docs/advanced/feeds.md new file mode 100644 index 0000000..6e57cf6 --- /dev/null +++ b/docs/advanced/feeds.md @@ -0,0 +1,41 @@ +# Feeds + +Feeds provide a per-user timeline of records authored by people that user follows. + +## Configure + +```ts +export default { + namespace: "com.example", + collections: { + event: { collection: "community.lexicon.calendar.event" }, + }, + feeds: { + network: { + targets: [{ collection: "event", maxItems: 200 }], + }, + }, +}; +``` + +The default follow collection is `app.bsky.graph.follow`. Contrail adds it internally with `discover: false`; declare a different collection and set `follow` only when your app uses another follow record type. + +## Query + +```ts +const response = await contrail.get("com.example.getFeed", { + params: { + feed: "network", + actor: signedInDid, + collection: "community.lexicon.calendar.event", + profiles: true, + limit: 20, + }, +}); +``` + +`actor` means “whose feed,” not “record author.” It accepts a DID or handle. Filters, sorting, pagination, and hydration work like `listRecords` for the selected target collection. + +The first read starts a bounded backfill of that actor's follows, so its initial result may be partial. New target records are then fanned out during normal ingestion. Old items are pruned to each target's `maxItems` cap. + +Fan-out cost grows with an author's number of indexed followers. Feeds are therefore a projection for bounded application communities, not a replacement for a network-wide timeline service. diff --git a/docs/advanced/labels.md b/docs/advanced/labels.md new file mode 100644 index 0000000..a636329 --- /dev/null +++ b/docs/advanced/labels.md @@ -0,0 +1,37 @@ +# Labels + +Contrail can subscribe to AT Protocol labelers and attach their labels to records and profiles. + +## Configure + +```ts +export default { + namespace: "com.example", + collections: { + event: { collection: "community.lexicon.calendar.event" }, + }, + labels: { + sources: [ + { did: "did:plc:ar7c4by46qjdydhdevvrndac" }, + ], + }, +}; +``` + +Normal `contrail backfill`, Worker cron ingestion, and `runPersistent()` include configured labelers automatically. + +## Select labelers + +A request chooses labelers with the standard header: + +```text +Atproto-Accept-Labelers: did:plc:ar7c4by46qjdydhdevvrndac +``` + +Use `?labelers=did:plc:...` when setting a header is inconvenient. Without either, Contrail uses `labels.defaults`, or all configured sources when defaults are omitted. Set `defaults: []` to require callers to opt in. + +Selected labels appear as `record.labels`. Account labels appear on hydrated profile entries. The response's `Atproto-Content-Labelers` header reports which configured labelers were applied. + +Contrail drops expired labels, applies CID-pinned labels only to the matching record version, and treats newer `neg: true` labels as retractions. + +Label signatures are stored but are not currently verified. Contrail consumes labels; it does not publish a label stream or provide moderation-preference UI. diff --git a/docs/advanced/outbox.md b/docs/advanced/outbox.md new file mode 100644 index 0000000..db069e1 --- /dev/null +++ b/docs/advanced/outbox.md @@ -0,0 +1,94 @@ +# Outbox + +> Experimental. Enable the outbox only on a fresh, empty Contrail database. + +The outbox delivers indexed record changes to external projections such as search indexes, webhooks, or caches. Contrail appends each change in the same database transaction as the canonical record and source cursor; destination failures never roll back ingestion. + +## Configure a consumer + +```ts +const config: ContrailConfig = { + namespace: "com.example", + collections: { + event: { collection: "community.lexicon.calendar.event" }, + }, + changes: { + consumers: { + search: { + collections: ["community.lexicon.calendar.event"], + phases: ["historical", "live"], + initial: "history", + }, + }, + }, +}; +``` + +Collections are full NSIDs, not config short names. Omitting `phases` includes both historical backfill and live ingestion. + +## Deliver from a Worker + +Add one handler for every configured consumer: + +```ts +type Env = { SEARCH_ENDPOINT: string }; + +export default createWorker(config, { + deliveries: { + search: async (batch, { env, signal }) => { + const response = await fetch(env.SEARCH_ENDPOINT, { + method: "POST", + signal, + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + cursor: batch.cursor, + upserts: batch.currentRecords, + deletes: batch.absentUris, + }), + }); + if (!response.ok) throw new Error(`Search returned ${response.status}`); + }, + }, +}); +``` + +`createWorker` runs bounded delivery rounds after scheduled ingestion. Throwing retries the batch with backoff; returning successfully acknowledges it. + +Delivery is **at least once**. A destination may apply a batch before the acknowledgement fails, so handlers must be idempotent. Upsert and delete by record URI rather than incrementing counters. + +Claims coalesce repeated changes to the same URI. `currentRecords` contains the latest indexed values when the batch is delivered; `absentUris` contains records that are currently deleted. + +## Initial state + +| `initial` | Starts with | +|---|---| +| `history` | All retained historical and live changes | +| `future` | Changes written after the consumer is registered | +| `current` | A current-state snapshot, a fixed catch-up tail, then atomic destination activation | + +`current` is intended for building a candidate index without a read gap. It additionally requires matching `changeBootstraps` snapshot and activation handlers. + +For a long-lived Node process, run delivery beside ingestion: + +```ts +await Promise.all([ + contrail.runPersistent({ signal }), + contrail.runPersistentDeliveries({ + env, + deliveries: { search: deliverSearch }, + runtime: { signal }, + }), +]); +``` + +## Operate + +```bash +pnpm contrail changes status --remote +pnpm contrail changes retry search --remote +pnpm contrail changes prune --remote +``` + +Status reports each consumer's position, backlog, lease, and retry state. Pruning never passes the slowest durable consumer. `changes skip` is an explicit audited data-loss operation and should be reserved for recovery. + +Once enabled, ordinary startup fails closed if a consumer is removed or changed incompatibly. Adding a consumer is safe only when its collection/phase coverage was already retained; otherwise build a fresh database generation. diff --git a/docs/frameworks/sveltekit-cloudflare.md b/docs/frameworks/sveltekit-cloudflare.md deleted file mode 100644 index dd384dd..0000000 --- a/docs/frameworks/sveltekit-cloudflare.md +++ /dev/null @@ -1,213 +0,0 @@ -# SvelteKit + Cloudflare Workers - -How to add contrail to an existing SvelteKit project deployed on Cloudflare Workers (via `@sveltejs/adapter-cloudflare`). Gives you XRPC endpoints alongside your pages, Jetstream ingestion on cron, and a typed in-process client for server loaders. - -Assumes you already have a SvelteKit app with `@sveltejs/adapter-cloudflare` and a D1 binding. If you don't, [`apps/sveltekit-cloudflare-workers`](https://github.com/flo-bit/contrail/tree/main/apps/sveltekit-cloudflare-workers) is a complete starting point. - -## Install - -```bash -pnpm add @atmo-dev/contrail @atcute/client -``` - -## Project layout - -``` -src/ - lib/ - contrail.config.ts # your config — auto-detected by the CLI - contrail/ - index.ts # Contrail instance + ensureInit + server client - routes/ - xrpc/[...path]/+server.ts # mounts all contrail XRPC endpoints - api/cron/+server.ts # hit by the cron trigger (see below) -wrangler.jsonc -``` - -## 1. Declare the config - -```ts -// src/lib/contrail.config.ts -import type { ContrailConfig } from "@atmo-dev/contrail"; - -export const config: ContrailConfig = { - namespace: "com.example", - collections: { - event: { - collection: "community.lexicon.calendar.event", - queryable: { startsAt: { type: "range" } }, - searchable: ["name", "description"], - }, - }, -}; -``` - -## 2. The Contrail instance - -```ts -// src/lib/contrail/index.ts -import { Contrail } from "@atmo-dev/contrail"; -import { createHandler, createServerClient } from "@atmo-dev/contrail/server"; -import type { Client } from "@atcute/client"; -import { config } from "../contrail.config"; - -export const contrail = new Contrail(config); - -let initialized = false; -export async function ensureInit(db: D1Database) { - if (!initialized) { await contrail.init(db); initialized = true; } -} - -const handle = createHandler(contrail); - -/** In-process XRPC client for loaders and actions. */ -export function getServerClient(db: D1Database): Client { - return createServerClient(async (req) => { - await ensureInit(db); - return handle(req, db) as Promise; - }); -} -``` - -Why the lazy `ensureInit`: Workers cold-start many times; doing schema init on the first request keeps the boot path fast and means `contrail.init()` doesn't need top-level `await` (which the adapter doesn't love). - -## 3. Mount the XRPC routes - -One catch-all that forwards to contrail's handler: - -```ts -// src/routes/xrpc/[...path]/+server.ts -import type { RequestHandler } from "./$types"; -import { createHandler } from "@atmo-dev/contrail/server"; -import { contrail, ensureInit } from "$lib/contrail"; - -const handle = createHandler(contrail); - -async function h(req: Request, platform: App.Platform | undefined) { - const db = platform!.env.DB; - await ensureInit(db); - return handle(req, db) as Promise; -} - -export const GET: RequestHandler = ({ request, platform }) => h(request, platform); -export const POST: RequestHandler = ({ request, platform }) => h(request, platform); -``` - -Now every `com.example.*.listRecords` / `com.example.*.getRecord` / `com.example.notifyOfUpdate` / etc. is served under `/xrpc/...`. - -## 4. Using the typed client in loaders - -```ts -// src/routes/+page.server.ts -import { getServerClient } from "$lib/contrail"; -import type { PageServerLoad } from "./$types"; - -export const load: PageServerLoad = async ({ platform }) => { - const rpc = getServerClient(platform!.env.DB); - const res = await rpc.get("com.example.event.listRecords", { - params: { startsAtMin: "2026-01-01", limit: 20 }, - }); - return { events: res.ok ? res.data.records : [] }; -}; -``` - -`createServerClient` bypasses the network — the loader runs Contrail's public XRPC handler in-process. - -## 5. Cron ingest — the workaround - -SvelteKit's `@sveltejs/adapter-cloudflare` doesn't expose a `scheduled()` export on the generated worker ([issue #4841](https://github.com/sveltejs/kit/issues/4841)). The fix is an HTTP endpoint that does the ingest, plus a post-build patch on `_worker.js` that appends a `scheduled` handler calling it. The patch is what `contrail append-scheduled` does. - -**Endpoint:** - -```ts -// src/routes/api/cron/+server.ts -import type { RequestHandler } from "./$types"; -import { contrail, ensureInit } from "$lib/contrail"; - -export const POST: RequestHandler = async ({ request, platform }) => { - if (request.headers.get("X-Cron-Secret") !== platform!.env.CRON_SECRET) { - return new Response("Unauthorized", { status: 401 }); - } - const db = platform!.env.DB; - await ensureInit(db); - await contrail.ingest({}, db); - return new Response("OK"); -}; -``` - -**Wire `contrail append-scheduled` into your `build` script:** - -```jsonc -// package.json -"scripts": { - "build": "vite build && contrail append-scheduled" -} -``` - -`contrail append-scheduled` patches `.svelte-kit/cloudflare/_worker.js` to append a `scheduled()` export that POSTs to `/api/cron` with `env.CRON_SECRET`. Override with `--worker `, `--cron-path `, or `--secret-env ` if your project diverges. - -`CRON_SECRET` is any random string — generate one, set it as a secret with `wrangler secret put CRON_SECRET`. The cron handler self-auths with it so nobody external can trigger your ingest. - -## 6. Wrangler config - -```jsonc -// wrangler.jsonc -{ - "main": ".svelte-kit/cloudflare/_worker.js", - "compatibility_date": "2025-12-25", - "compatibility_flags": ["nodejs_compat_v2"], - "assets": { "binding": "ASSETS", "directory": ".svelte-kit/cloudflare" }, - "d1_databases": [ - { "binding": "DB", "database_name": "yourapp", "database_id": "..." } - ], - "triggers": { "crons": ["*/1 * * * *"] } -} -``` - -Type the D1 binding in `src/app.d.ts`: - -```ts -declare global { - namespace App { - interface Platform { - env: { - DB: D1Database; - CRON_SECRET: string; - // ...other bindings - }; - } - } -} -``` - -## 7. Deploy + backfill - -```bash -pnpm wrangler d1 create yourapp # copy the id into wrangler.jsonc -pnpm build && pnpm wrangler deploy -pnpm wrangler secret put CRON_SECRET # paste any random string -pnpm contrail backfill --remote # one-time historical backfill -``` - -From now on: - -- Pages and XRPC endpoints are served under your domain. -- The cron fires every minute, hitting `/api/cron`, which runs `contrail.ingest()`. -- Loaders that need live data use `getServerClient()` for zero-overhead typed calls. -- After a short outage, ingestion resumes from its saved cursor. If source history has expired, rebuild into a fresh database with `pnpm backfill:remote`. - -## Where to go next - -- [Indexing](../01-indexing.md) — config options, adapter choices -- [Querying](../02-querying.md) — filters, sorts, hydration, search -- [Feeds](../04-feeds.md) — personalized timelines via follow + target collections -- [Labels](../09-labels.md) — moderation label hydration - -Use Atcute directly for Lexicon pulling, validation, and TypeScript generation. - -## Common gotchas - -- **Top-level await in `$lib/contrail/index.ts`** will fail to bundle — use the lazy `ensureInit` pattern above. -- **`ensureInit` is per-isolate, not global.** Cloudflare cold-starts spin new isolates; each one pays one init call on its first request. `contrail.init()` is idempotent so this is safe, just not instant. -- **SvelteKit's `adapter-cloudflare` regenerates `_worker.js` on every build**, so `contrail append-scheduled` has to run *after* `vite build`. Don't try to put it in `prebuild`. -- **`D1Database` type in platform env** needs `@cloudflare/workers-types` in `devDependencies` and `types` in your tsconfig. diff --git a/docs/public-services/api-atmo-rsvp.md b/docs/public-services/api-atmo-rsvp.md deleted file mode 100644 index cfb2345..0000000 --- a/docs/public-services/api-atmo-rsvp.md +++ /dev/null @@ -1,246 +0,0 @@ -# Example: api.atmo.rsvp - -[`https://api.atmo.rsvp`](https://api.atmo.rsvp) is a public Contrail read-through service for AT Protocol calendar events and RSVPs. It demonstrates anonymous collection queries, profile hydration, a personalized network feed, authenticated update notifications, verified remote discovery, and an immutable D1 deployment generation. - -## Discovery - -```text -https://api.atmo.rsvp/.well-known/contrail -https://api.atmo.rsvp/.well-known/did.json -https://api.atmo.rsvp/lexicons -https://api.atmo.rsvp/status -``` - -The XRPC namespace is DNS-authoritative: - -```text -rsvp.atmo.* -``` - -The base service DID and exact service-auth audience are: - -```text -service DID: did:web:api.atmo.rsvp -audience: did:web:api.atmo.rsvp#contrail -``` - -The fragmented service reference is the OAuth and JWT audience. The base DID identifies the DID document published by the API. The API does not use either value to sign user records and does not act as a PDS. - -## Anonymous methods - -```text -rsvp.atmo.getCursor -rsvp.atmo.getProfile -rsvp.atmo.event.getRecord -rsvp.atmo.event.listRecords -rsvp.atmo.rsvp.getRecord -rsvp.atmo.rsvp.listRecords -``` - -Event queries support equality and date-range filtering, full-text search over names and descriptions, stable keyset pagination, RSVP relation counts, RSVP hydration, and actor profile hydration. - -RSVP queries support status, subject URI, and creation-time filtering. They can hydrate the referenced event and actor profiles. - -`getProfile` resolves an actor and reads their indexed `app.bsky.actor.profile` record. Missing public profile data may be fetched from that actor's PDS as part of the read-through request. - -## Protected methods - -Discovery lists these separately under AT Protocol service auth: - -```text -rsvp.atmo.getFeed -rsvp.atmo.notifyOfUpdate -``` - -The module generated by `contrail connect` exposes the required OAuth permission: - -```ts -import { contrail } from "./contrail/index.js"; - -export const scopes = ["atproto", contrail.scope]; -// contrail.scope is: -// rpc?aud=did:web:api.atmo.rsvp%23contrail&lxm=rsvp.atmo.getFeed&lxm=rsvp.atmo.notifyOfUpdate -``` - -After login, derive one client from the user's existing authenticated AT Protocol client: - -```ts -const client = contrail.authenticated(authenticatedClient, { - onNotificationError(error, { uris }) { - console.warn("Contrail notification failed", uris, error); - }, -}); -``` - -Advertised service methods route to Contrail; other methods route to the PDS. Protected methods automatically obtain and cache exact method-bound tokens. - -Missing, expired, wrong-audience, wrong-method, and invalid-signature tokens receive `401` with a `WWW-Authenticate` challenge. - -## Personalized network feed - -The configured feed is: - -```text -feed=network -``` - -It contains recent events and RSVPs authored by actors followed by the signed-in user. Per-actor projection caps are: - -| Collection | Maximum retained items | -|---|---:| -| `community.lexicon.calendar.event` | 100 | -| `community.lexicon.calendar.rsvp` | 250 | - -A typed query looks like: - -```ts -const response = await client.get( - "rsvp.atmo.getFeed", - { - params: { - feed: "network", - actor: signedInDid, - collection: "community.lexicon.calendar.event", - profiles: true, - limit: 20, - }, - }, -); -``` - -The requested actor must resolve to the token issuer. Service auth therefore prevents one authenticated account from creating or refreshing arbitrary personalized feed projections for other actors. - -The underlying `app.bsky.graph.follow` records are internal. The service does not advertise raw follow collection methods. It indexes follows authored by known actors only when the follow subject is already in the service's acquisition scope. Constellation enrichment helps connect newly observed calendar authors to existing in-scope followers. - -The feed endpoint may start a bounded background follow backfill the first time an actor requests their feed. Until that finishes, an initial response may be empty or partial. The operation remains a read-through cache fill rather than a network-wide social graph crawl. - -## Immediate update notification - -Successful event and RSVP writes through the combined client automatically notify Contrail: - -```ts -const response = await client.post("com.atproto.repo.createRecord", { - input: { - repo: signedInDid, - collection: "community.lexicon.calendar.event", - record: event, - }, -}); -``` - -The original PDS response is returned unchanged. Notification failures are nonfatal and reported through `onNotificationError`. The protected `rsvp.atmo.notifyOfUpdate` procedure remains available for explicit batches or records written elsewhere. - -The endpoint enforces all of the following: - -- at most 25 URIs per request; -- every URI is a canonical record AT URI; -- every URI belongs to the service-token issuer; -- the collection is tracked by this deployment; -- the record body and CID are fetched from the issuer's current PDS; and -- only an explicit XRPC `RecordNotFound` response is treated as deletion. - -Transient DNS, identity, network, timeout, PDS, or malformed-response failures preserve existing indexed state and are returned as bounded per-record errors to the authenticated caller. - -## Profiles - -The deployment indexes: - -```text -app.bsky.actor.profile -``` - -but keeps the underlying profile collection methods internal. Profiles are exposed through: - -- `rsvp.atmo.getProfile`; -- `profiles=true` on event and RSVP reads; and -- `profiles=true` on network feed reads. - -This avoids a redundant raw profile-record API while still providing typed display names, handles, avatars, and profile values alongside calendar data. - -## Acquisition scope - -Relay discovery starts from: - -```text -community.lexicon.calendar.event -community.lexicon.calendar.rsvp -``` - -Profiles and follows are dependent collections. They do not independently discover every Bluesky repository. - -The service is intentionally a shared, possibly incomplete cache. Unavailable identities and PDSes stay visibly pending, retrying, or failed instead of being silently marked complete. Unknown dependent subjects are scope exclusions and do not create tombstones. - -## Runtime validation policy - -This deployment publishes verified API and record Lexicons for discovery and TypeScript generation, but does not enable Contrail's optional runtime record/CID validation during ingestion. - -That distinction is deliberate. A matched benchmark using the current calendar Lexicons rejected 12,291 historical records, reducing indexed events from roughly 14,600 to 4,800 and RSVPs from roughly 6,300 to 3,800. Those records include historical shapes that predate the current published definitions, so enabling latest-schema validation would silently discard most of the useful archive. - -Under the compatibility policy: - -- provider and consumer contracts are still canonical and digest-verified; -- generated Atcute types describe the current expected response values, but are not a runtime guarantee for every historical record; -- startup still rejects an inconsistent advertised API; -- records and CIDs are fetched from authoritative sources rather than accepted from callers; but -- ingestion does not reject records based on runtime Lexicon or canonical-CID checks. - -A future strict deployment needs version-aware historical schemas or an explicitly looser response value, rather than pretending the compatibility loss does not exist. - -## Ordered source position - -The primary ordered source is one pinned Jetstream endpoint with an operator-owned continuity epoch: - -```json -{ - "source": "jetstream", - "epoch": "api-atmo-rsvp-primary-2026-08" -} -``` - -`rsvp.atmo.getCursor` returns the currently committed opaque position. Consumers compare the complete source, epoch, and cursor for equality only. A source or epoch change requires a full refetch. - -Backfills do not send historical notifications. Jetstream projection advances the serving position atomically with accepted live mutations. - -## Production generation - -The current expanded generation was built in native SQLite before activation. Its initial canonical projection contained approximately: - -| Collection | Records | -|---|---:| -| Events | 14,751 | -| RSVPs | 6,299 | -| Profiles | 1,408 | -| Scoped follows | 65,798 | - -The exact live totals change as Jetstream and read-through acquisition continue. - -The provisioning process: - -1. captured a replay boundary before relay discovery; -2. ran the resumable native-SQLite backfill; -3. retained 43 unavailable accounts as explicit scheduled retries; -4. replayed Jetstream to the present; -5. imported canonical tables into a fresh D1 database; -6. rebuilt FTS and materialized RSVP counts; -7. verified all visible rows against durable record versions; -8. exercised discovery, profiles, search, CORS, and protected-route rejection through a candidate Worker; and -9. activated the new Worker and D1 binding together. - -The previous D1 generation remains separate for rollback. No percentage traffic split is used between databases with independent serving positions. - -## Connect a consumer - -```bash -pnpx @atmo-dev/contrail connect https://api.atmo.rsvp -``` - -The provider lock records: - -- the HTTPS endpoint; -- `rsvp.atmo` namespace; -- anonymous methods; -- protected methods and their audience; -- the content-addressed Lexicon digest; and -- the provider-owned Lexicon directory. - -See [Using a public Contrail service](./using.md) for a framework-neutral consumer walkthrough. diff --git a/docs/public-services/creating.md b/docs/public-services/creating.md deleted file mode 100644 index a520a0c..0000000 --- a/docs/public-services/creating.md +++ /dev/null @@ -1,311 +0,0 @@ -# Creating a public Contrail service - -A public Contrail service lets independent applications query one Contrail AppView from a stable HTTPS origin. The provider chooses the indexed collections, projections, query methods, and authentication policy. Consumers discover that API surface, verify its Lexicons, generate local TypeScript types, and make ordinary XRPC requests. - -Public service mode does not turn Contrail into a PDS. Records remain in their authors' repositories, and applications still authenticate users and publish writes through those users' PDSes. - -## Define the index - -Start with a normal Contrail configuration: - -```ts -// src/contrail.config.ts -import type { ContrailConfig } from "@atmo-dev/contrail"; - -export const config: ContrailConfig = { - namespace: "events.example", - orderedSource: { - source: "jetstream", - epoch: "primary-2026-08", - }, - collections: { - event: { - collection: "community.lexicon.calendar.event", - queryable: { - mode: {}, - startsAt: { type: "range" }, - }, - searchable: ["name", "description"], - }, - }, -}; -``` - -The namespace becomes the prefix of generated methods such as: - -```text -events.example.getCursor -events.example.event.getRecord -events.example.event.listRecords -``` - -Use one stable `orderedSource.epoch` for one continuity history. Change the epoch when the Jetstream endpoint set, retention assumptions, or cursor meaning changes. Consumers treat a source or epoch change as a full-refetch boundary. - -## Generate the public Lexicons - -Add Atcute's generator configuration: - -```js -// lex.config.js -import { defineLexiconConfig } from "@atcute/lex-cli"; - -export default defineLexiconConfig({ - generate: { - files: [ - "lexicons/custom/**/*.json", - "lexicons/pulled/**/*.json", - "lexicons/generated/**/*.json", - ], - outdir: "src/lexicon-types/", - }, -}); -``` - -Generate the provider API, pull referenced record Lexicons, and generate TypeScript types: - -```bash -pnpm contrail lexicons all --public -``` - -Check generated drift in CI: - -```bash -pnpm contrail lexicons check --public -``` - -The public surface includes anonymous queries plus explicitly configured service-auth queries and procedures. Private full-surface procedures are not included merely because they exist in application code. - -## Publish discovery from a Worker - -Pass the generated documents and canonical HTTPS origin to `createWorker`: - -```ts -// src/worker.ts -import { createWorker } from "@atmo-dev/contrail/worker"; -import { lexicons } from "../lexicons/generated"; -import { config } from "./contrail.config"; - -export default createWorker(config, { - lexicons, - publicService: { - endpoint: "https://api.example.com", - }, -}); -``` - -This publishes: - -```text -GET /.well-known/contrail -GET /lexicons -GET /lexicons/ -GET /status -``` - -The version-2 discovery manifest contains the endpoint, namespace, methods, collections, service-auth declaration, and a content-addressed Lexicon bundle. It does not hash the complete method set. Startup still fails when advertised methods, capabilities, and bundled Lexicons disagree, and the immutable Lexicon URL retains its digest. - -The public `/status` response contains aggregate readiness and freshness information. It omits DIDs, record bodies, source cursors, raw upstream errors, and other private operational details. - -## Anonymous read-through methods - -Collection queries, `getCursor`, profiles, feeds, and authored custom queries are anonymous unless explicitly protected. An anonymous query may still improve the shared cache by: - -- resolving an actor; -- fetching a missing public record from its PDS; -- populating profile or feed projections; or -- running a trusted custom query handler. - -This is read-through acquisition, not a caller-controlled write API. Only configured collections and trusted provider code can affect the projection. - -## Protecting feeds and notifications with service auth - -AT Protocol service auth lets any suitably authorized AT Protocol client call selected methods without distributing a shared application secret. - -```ts -export const config: ContrailConfig = { - namespace: "events.example", - notify: true, - serviceAuth: { - audience: "did:web:api.example.com#contrail", - methods: ["getFeed", "notifyOfUpdate"], - }, - collections, - feeds: { - network: { - targets: [{ collection: "event", maxItems: 100 }], - }, - }, -}; -``` - -The provider verifies: - -- the JWT signature against the issuer DID's `#atproto` key; -- the exact fragmented service audience; -- the token's exact `lxm` method claim; -- expiration and maximum token age; and -- the route-specific ownership rule. - -For protected feeds, the requested actor must resolve to the token issuer. For `notifyOfUpdate`, every submitted AT URI must belong to the token issuer. Notify still fetches the current authoritative record from that issuer's PDS; callers never submit a record body for Contrail to trust. - -The default PLC/`did:web` resolver keeps a bounded five-minute in-process cache and deduplicates concurrent lookups. Signature failure forces an uncached refresh so key rotation does not remain hidden behind a stale entry. Deployments can still provide their own resolver policy. - -The authenticated methods are listed separately from anonymous methods in discovery. Their query or procedure Lexicons remain in the provider bundle, so consumers still get generated types. - -## Start from owned source - -A consumer developed alongside the provider can generate its initial API surface before any deployment exists: - -```bash -pnpx @atmo-dev/contrail connect ./src/contrail.config.ts -# or discover the standard config beneath another project directory -pnpx @atmo-dev/contrail connect ../api -``` - -This compiles the config directly, writes generated Lexicons and types, and exports local/target client factories. It does not create or modify `contrail.lock.json`. Run the service separately: - -```bash -pnpx @atmo-dev/contrail dev --config ../api/src/contrail.config.ts -``` - -After deploying, `contrail connect https://api.example.com` creates the version-2 provider lock and makes that deployment the generated default target. Future config-source connections can refresh local API types without changing the production lock. - -### OAuth permission versus token binding - -Contrail generates one least-privilege OAuth permission containing the exact fragmented audience and one sorted `lxm` parameter per protected method: - -```text -rpc?aud=did:web:api.example.com%23contrail&lxm=events.example.getFeed&lxm=events.example.notifyOfUpdate -``` - -Current granular OAuth RPC syntax requires the absolute DID service reference; a plain DID may be silently removed from an authorization request. The `%23` is the encoded `#` delimiter. Wildcard `lxm=*` permissions are not generated because they are broader than necessary and can trigger misleading consent descriptions. - -Each call to `com.atproto.server.getServiceAuth` still passes the specific method NSID as `lxm` and the decoded `did:web:api.example.com#contrail` audience, producing a short-lived method-bound token. - -For example, a feed token uses: - -```text -lxm=events.example.getFeed -``` - -and cannot be reused for: - -```text -lxm=events.example.notifyOfUpdate -``` - -### Service DID and audience - -For this configuration, the identities are distinct: - -```text -base service DID: did:web:api.example.com -exact audience: did:web:api.example.com#contrail -public endpoint: https://api.example.com -``` - -Contrail publishes the base DID document at: - -```text -https://api.example.com/.well-known/did.json -``` - -The document `id` is the base DID and its Contrail service entry `id` is the exact fragmented audience. Startup fails if an automatically hosted `did:web` audience resolves to a different DID-document URL. A `did:plc` audience remains externally managed and does not create a local DID route. - -The exact service reference—not the base DID—is the JWT audience passed to `getServiceAuth` and verified by Contrail. The AppView does not need a signing key merely to receive and verify user-issued service tokens. - -## Profiles and internal follow projections - -Profiles can be enabled without exposing raw profile collection methods: - -```ts -profiles: ["app.bsky.actor.profile"], -collections: { - event, - profile: { - collection: "app.bsky.actor.profile", - discover: false, - methods: [], - }, -} -``` - -Likewise, a follow collection can remain an internal feed input: - -```ts -follow: { - collection: "app.bsky.graph.follow", - discover: false, - subjectField: "subject", - methods: [], -} -``` - -`discover: false` prevents network-wide relay discovery for dependent collections. `subjectField: "subject"` excludes follows whose subject is outside the known acquisition scope. Those exclusions do not create tombstones. - -Profiles can then appear through `getProfile` and `profiles=true` hydration, while follows power `getFeed` without creating a public social-graph directory. - -## Runtime validation is explicit per collection - -The deployment already ships one reviewed generated Lexicon bundle. Select which collection policies use it directly: - -```ts -collections: { - event: { - collection: "community.lexicon.calendar.event", - validate: true, - }, - legacy: { - collection: "community.example.legacy", - // Omitted or false means no runtime validation. - }, -}, -validation: { - strict: true, - verifyCid: true, -} -``` - -`createWorker(config, { lexicons })` binds the exact deployment bundle to opted-in collections; no second validation-specific array is needed. Validation applies across every acquisition source and startup fails if an opted-in record schema or transitive reference is absent. Collections without `validate: true` retain compatibility behavior. - -## CORS - -Public services allow browser requests and explicitly permit the `Authorization`, `Content-Type`, and `Atproto-Accept-Labelers` headers. Authentication failures expose `WWW-Authenticate` so browser clients can distinguish missing, expired, wrong-audience, and wrong-method tokens. - -Never place a reusable application secret in browser code. AT Protocol service tokens are short-lived and minted for the authenticated user's DID. - -## Provisioning and activation - -Local development can use the normal local backfill command: - -```bash -pnpm contrail backfill -``` - -For a substantial D1 production deployment, do not run a long bulk load through Wrangler's remote development proxy. Prefer a fresh generation: - -1. capture the ordered-source replay boundary; -2. build canonical state in native SQLite; -3. leave unavailable accounts visibly pending, retrying, or failed; -4. catch up through the ordered source; -5. import canonical tables into a fresh D1 database; -6. rebuild FTS, relation counts, and other derived projections; -7. verify record/version consistency, status, discovery, and representative queries; -8. test the candidate through a non-production Worker; and -9. activate the matching Worker and D1 binding together. - -Keep the previous D1 generation available for rollback. Do not split percentage traffic between independent databases with different serving positions. - -## Provider checklist - -Before announcing an origin: - -- run public Lexicon drift checking and TypeScript typechecking; -- verify the manifest and Lexicon digests differ and both recompute correctly; -- verify every advertised method has a matching query or procedure Lexicon; -- verify protected methods reject missing, wrong-audience, and wrong-`lxm` tokens; -- verify feed actors and notify URIs are bound to the token issuer; -- verify browser CORS preflight with `Authorization`; -- verify `/status` contains no sensitive operational detail; -- verify `getCursor` reports the committed ordered-source position; and -- connect and compile an independent consumer project. diff --git a/docs/public-services/using.md b/docs/public-services/using.md deleted file mode 100644 index 69903ef..0000000 --- a/docs/public-services/using.md +++ /dev/null @@ -1,164 +0,0 @@ -# Using a public Contrail service - -A public Contrail service is a typed, read-through API over public AT Protocol records. This guide uses `https://api.atmo.rsvp`; replace it with the provider you want to use. - -## Install and connect - -```bash -pnpm add @atcute/client @atcute/lexicons @atmo-dev/contrail -pnpx @atmo-dev/contrail connect https://api.atmo.rsvp -``` - -`connect` validates the provider description and content-addressed Lexicon bundle, writes a version-2 `contrail.lock.json`, and generates: - -```text -lex.config.js -src/contrail/ - index.ts - lexicons/ - types/ -``` - -The generated configuration contains everything needed to identify the service and generate its types: - -```js -export default { - contrail: { - endpoint: "https://api.atmo.rsvp", - serviceDid: "did:web:api.atmo.rsvp", - serviceAudience: "did:web:api.atmo.rsvp#contrail", - scope: - "rpc?aud=did:web:api.atmo.rsvp%23contrail&lxm=rsvp.atmo.getFeed&lxm=rsvp.atmo.notifyOfUpdate", - protectedMethods: [ - "rsvp.atmo.getFeed", - "rsvp.atmo.notifyOfUpdate", - ], - collections: [ - "app.bsky.actor.profile", - "app.bsky.graph.follow", - "community.lexicon.calendar.event", - "community.lexicon.calendar.rsvp", - ], - }, - generate: { - files: ["src/contrail/lexicons/**/*.json"], - outdir: "src/contrail/types/", - }, -}; -``` - -For JavaScript output, choose a `.js` client path: - -```bash -pnpx @atmo-dev/contrail connect https://api.atmo.rsvp \ - --client src/contrail/index.js -``` - -Commit `contrail.lock.json` and the generated files. - -When the application owns or can read the provider config, connect directly to that source before a deployment exists: - -```bash -pnpx @atmo-dev/contrail connect ../api/src/contrail.config.ts -pnpx @atmo-dev/contrail dev --config ../api/src/contrail.config.ts -``` - -The config connection compiles Lexicons and types without creating or modifying `contrail.lock.json`. `contrail dev` only runs the loopback service. The generated module exports `createLocalContrailClient()`: - -```ts -import { - contrail as productionContrail, - createLocalContrailClient, -} from "./contrail/index.js"; - -export const contrail = process.env.CONTRAIL_URL - ? createLocalContrailClient(process.env.CONTRAIL_URL) - : productionContrail; -``` - -The helper permits HTTP only for `localhost`, `127.0.0.1`, or `[::1]` and does not inherit a production service-auth audience. Local notification and configured protected methods are loopback-only operations with a null OAuth scope. A deployed service still requires HTTPS and its real service auth. - -## Query anonymous methods - -```ts -import { contrail } from "./contrail/index.js"; - -const response = await contrail.get("rsvp.atmo.event.listRecords", { - params: { - limit: 20, - sort: "startsAt", - order: "asc", - profiles: true, - }, -}); - -if (!response.ok) { - throw new Error(`Contrail query failed: ${response.status}`); -} - -for (const event of response.data.records) { - console.log(event.value.name, event.value.startsAt); -} -``` - -The generated Lexicons provide typed method names, parameters, and responses. - -## Use one authenticated client - -Add the provider's generated least-privilege scope to the application's OAuth scopes. It contains the encoded fragmented audience and only the protected methods advertised by this provider: - -```ts -import { contrail } from "./contrail/index.js"; - -export const scopes = ["atproto", contrail.scope]; -``` - -After login, combine Contrail with the existing authenticated AT Protocol client: - -```ts -const client = contrail.authenticated(authenticatedClient, { - onNotificationError(error, { uris }) { - console.warn("Contrail notification failed", uris, error); - }, -}); - -const response = await client.get("rsvp.atmo.getFeed", { - params: { - feed: "network", - actor: signedInDid, - collection: "community.lexicon.calendar.event", - profiles: true, - limit: 20, - }, -}); -``` - -The same client still handles ordinary PDS calls. Successful writes to connected collections automatically notify Contrail: - -```ts -await client.post("com.atproto.repo.createRecord", { - input: { - repo: signedInDid, - collection: "community.lexicon.calendar.event", - record: event, - }, -}); -``` - -Contrail returns the original PDS response. A notification failure is reported through `onNotificationError` but never turns a committed PDS write into a failed write. Handle-form `deleteRecord` inputs are resolved through the PDS to construct the canonical DID record URI before notification. The login session remains application-owned. - -## Update the connection - -Anonymous generated clients call the endpoint directly without fetching discovery first. Protected calls lazily discover service auth; transient discovery failures remain retryable, while endpoint, base-DID, exact-audience, scope, or protected-method mismatches fail closed. Adding anonymous provider methods does not interrupt methods already known by a generated client. Regenerate when application code wants new API surface or the protected contract changes: - -```bash -pnpx @atmo-dev/contrail connect https://api.atmo.rsvp --update -``` - -Review changes to `contrail.lock.json`, `lex.config.js`, and `src/contrail/`, then run the application's typecheck and tests. A change from the old plain-DID/wildcard permission requires OAuth reauthorization; an existing grant cannot mint tokens for the corrected fragmented audience. `--update` cannot repoint an existing lock or abandon its provider-owned Lexicon root; remove the existing connection deliberately before switching providers or output roots. Version-1 provider locks are intentionally unsupported after the clean manifest-v2 cut and must be removed before reconnecting. A version-2 lock written before exact audiences existed is reported separately and needs the same removal, reconnection, and OAuth reauthorization. - -## Completeness - -A public Contrail service is a shared, possibly incomplete read-through cache. Reads may fetch missing public records or profiles, but an empty result does not prove that no matching record exists on the network. - -Applications still authenticate users and publish records through their PDS. Contrail service auth only authorizes the protected methods advertised by that service. diff --git a/packages/contrail/README.md b/packages/contrail/README.md index 947a24d..a4a42f6 100644 --- a/packages/contrail/README.md +++ b/packages/contrail/README.md @@ -293,7 +293,6 @@ The returned cursor is opaque. Compare the complete `{ source, epoch, cursor }` Connect an independent consumer with `contrail connect `. The version-2 provider lock records the deployment and exact Lexicon bundle, but generated clients do not pin the provider's complete method set at runtime. Existing anonymous methods therefore continue working when a provider adds methods. A repeated connection to the same endpoint and provider-owned output root requires `--update`; provider files and the lock are staged and swapped without deleting consumer-owned Lexicons. Version-1 locks must be removed and reconnected. -See [Creating a public service](../../docs/public-services/creating.md), [Using a public service](../../docs/public-services/using.md), and [Example: api.atmo.rsvp](../../docs/public-services/api-atmo-rsvp.md) for complete provider and consumer walkthroughs. ## Runtime record validation diff --git a/packages/contrail/src/core/change-bootstrap.ts b/packages/contrail/src/core/change-bootstrap.ts index 2115bfc..4385bad 100644 --- a/packages/contrail/src/core/change-bootstrap.ts +++ b/packages/contrail/src/core/change-bootstrap.ts @@ -142,7 +142,8 @@ async function release( await db .prepare( `UPDATE change_consumers - SET lease_owner = NULL, lease_expires_at = NULL, updated_at = ? + SET lease_owner = NULL, lease_expires_at = NULL, + lease_through_position = NULL, updated_at = ? WHERE consumer_id = ? AND generation_id = ? AND lease_owner = ?`, ) .bind(timestamp, consumerId, generation, owner) @@ -224,7 +225,8 @@ export async function claimCurrentSnapshotPage( bootstrap_scan_cursor = CASE WHEN bootstrap_state = 'pending' THEN NULL ELSE bootstrap_scan_cursor END, - lease_owner = ?, lease_expires_at = ?, updated_at = ? + lease_owner = ?, lease_expires_at = ?, + lease_through_position = NULL, updated_at = ? WHERE consumer_id = ? AND initial_mode = 'current' AND bootstrap_state IN ('pending', 'scanning') AND generation_id = ( @@ -349,7 +351,8 @@ export async function claimCurrentSnapshotPage( .prepare( `UPDATE change_consumers SET bootstrap_scan_collection = ?, bootstrap_scan_cursor = NULL, - lease_owner = NULL, lease_expires_at = NULL, updated_at = ? + lease_owner = NULL, lease_expires_at = NULL, + lease_through_position = NULL, updated_at = ? WHERE consumer_id = ? AND generation_id = ? AND bootstrap_state = 'scanning' AND lease_owner = ?`, ) @@ -372,7 +375,8 @@ export async function claimCurrentSnapshotPage( SELECT head_position FROM change_log_state WHERE id = 1 ), bootstrap_scan_collection = NULL, bootstrap_scan_cursor = NULL, - lease_owner = NULL, lease_expires_at = NULL, updated_at = ? + lease_owner = NULL, lease_expires_at = NULL, + lease_through_position = NULL, updated_at = ? WHERE consumer_id = ? AND generation_id = ? AND bootstrap_state = 'scanning' AND lease_owner = ?`, ) @@ -393,7 +397,8 @@ export async function acknowledgeCurrentSnapshotPage( .prepare( `UPDATE change_consumers SET bootstrap_scan_cursor = ?, lease_owner = NULL, - lease_expires_at = NULL, attempts = 0, next_attempt_at = NULL, + lease_expires_at = NULL, lease_through_position = NULL, + attempts = 0, next_attempt_at = NULL, last_success_at = ?, last_error_code = NULL, last_error_at = NULL, updated_at = ? WHERE consumer_id = ? AND generation_id = ? @@ -449,6 +454,7 @@ async function failBootstrapLease( .prepare( `UPDATE change_consumers SET lease_owner = NULL, lease_expires_at = NULL, + lease_through_position = NULL, attempts = attempts + 1, next_attempt_at = ?, last_error_code = ?, last_error_at = ?, updated_at = ? WHERE consumer_id = ? AND generation_id = ? @@ -544,7 +550,8 @@ export async function claimCurrentActivation( const row = await db .prepare( `UPDATE change_consumers - SET lease_owner = ?, lease_expires_at = ?, updated_at = ? + SET lease_owner = ?, lease_expires_at = ?, + lease_through_position = NULL, updated_at = ? WHERE consumer_id = ? AND initial_mode = 'current' AND bootstrap_state = 'activating' AND generation_id = ( @@ -596,7 +603,8 @@ export async function completeCurrentActivation( .prepare( `UPDATE change_consumers SET bootstrap_state = 'ready', lease_owner = NULL, - lease_expires_at = NULL, attempts = 0, next_attempt_at = NULL, + lease_expires_at = NULL, lease_through_position = NULL, + attempts = 0, next_attempt_at = NULL, last_success_at = ?, last_error_code = NULL, last_error_at = NULL, updated_at = ? WHERE consumer_id = ? AND generation_id = ? diff --git a/packages/contrail/src/core/change-log.ts b/packages/contrail/src/core/change-log.ts index 1220a3a..937463e 100644 --- a/packages/contrail/src/core/change-log.ts +++ b/packages/contrail/src/core/change-log.ts @@ -163,6 +163,7 @@ export function buildChangeLogSchema( bootstrap_token TEXT, lease_owner TEXT, lease_expires_at ${bigint}, + lease_through_position ${bigint}, attempts INTEGER NOT NULL DEFAULT 0, next_attempt_at ${bigint}, last_success_at ${bigint}, diff --git a/packages/contrail/src/core/changes.ts b/packages/contrail/src/core/changes.ts index 0008639..a94038d 100644 --- a/packages/contrail/src/core/changes.ts +++ b/packages/contrail/src/core/changes.ts @@ -306,7 +306,8 @@ async function releaseLease( await db .prepare( `UPDATE change_consumers - SET lease_owner = NULL, lease_expires_at = NULL, updated_at = ? + SET lease_owner = NULL, lease_expires_at = NULL, + lease_through_position = NULL, updated_at = ? WHERE consumer_id = ? AND generation_id = ? AND acknowledged_position = ? AND lease_owner = ?`, ) @@ -427,7 +428,8 @@ async function claimChangeRange( const leased = await db .prepare( `UPDATE change_consumers - SET lease_owner = ?, lease_expires_at = ?, updated_at = ? + SET lease_owner = ?, lease_expires_at = ?, + lease_through_position = NULL, updated_at = ? WHERE consumer_id = ? AND bootstrap_state = '${requiredState}' AND generation_id = (SELECT generation_id FROM change_log_state WHERE id = 1) @@ -508,7 +510,8 @@ async function claimChangeRange( .prepare( `UPDATE change_consumers SET bootstrap_state = 'activating', lease_owner = NULL, - lease_expires_at = NULL, updated_at = ? + lease_expires_at = NULL, lease_through_position = NULL, + updated_at = ? WHERE consumer_id = ? AND generation_id = ? AND bootstrap_state = 'catching-up' AND acknowledged_position = bootstrap_target_position @@ -585,6 +588,33 @@ async function claimChangeRange( } const through = String(selected.at(-1)!.position); + // Persist the exact selected upper bound before exposing the lease. Ack, + // renew, and fail all compare it so a mutated claim cannot skip delivery. + const bounded = await db + .prepare( + `UPDATE change_consumers + SET lease_through_position = ?, updated_at = ? + WHERE consumer_id = ? AND generation_id = ? + AND acknowledged_position = ? AND lease_owner = ? + AND lease_expires_at > ? + RETURNING consumer_id`, + ) + .bind( + through, + limits.now, + consumerId, + generation, + from, + owner, + limits.now, + ) + .first<{ consumer_id: string }>(); + if (!bounded) { + throw new ChangeLeaseLostError( + `Change claim for ${consumerId} expired while its range was being bound`, + ); + } + const rows = await db .prepare( `SELECT position, phase, change_count, encoded_bytes, changes_json @@ -759,12 +789,13 @@ export async function acknowledgeChanges( AND ? = bootstrap_target_position THEN 'activating' ELSE bootstrap_state END, lease_owner = NULL, lease_expires_at = NULL, + lease_through_position = NULL, attempts = 0, next_attempt_at = NULL, last_success_at = ?, last_error_code = NULL, last_error_at = NULL, updated_at = ? WHERE consumer_id = ? AND generation_id = ? AND acknowledged_position = ? AND lease_owner = ? - AND lease_expires_at > ? + AND lease_expires_at > ? AND lease_through_position = ? AND ? > acknowledged_position AND ? <= ( SELECT head_position FROM change_log_state @@ -784,6 +815,7 @@ export async function acknowledgeChanges( now, claim.through, claim.through, + claim.through, claim.generation, ) .first<{ consumer_id: string }>(); @@ -814,7 +846,7 @@ export async function renewChangeClaim( SET lease_expires_at = ?, updated_at = ? WHERE consumer_id = ? AND generation_id = ? AND acknowledged_position = ? AND lease_owner = ? - AND lease_expires_at > ? + AND lease_expires_at > ? AND lease_through_position = ? RETURNING consumer_id`, ) .bind( @@ -825,6 +857,7 @@ export async function renewChangeClaim( claim.from, claim.leaseOwner, now, + claim.through, ) .first<{ consumer_id: string }>(); if (!renewed) { @@ -860,10 +893,12 @@ export async function failChanges( .prepare( `UPDATE change_consumers SET lease_owner = NULL, lease_expires_at = NULL, + lease_through_position = NULL, attempts = attempts + 1, next_attempt_at = ?, last_error_code = ?, last_error_at = ?, updated_at = ? WHERE consumer_id = ? AND generation_id = ? AND acknowledged_position = ? AND lease_owner = ? + AND lease_through_position = ? RETURNING attempts, next_attempt_at`, ) .bind( @@ -875,6 +910,7 @@ export async function failChanges( claim.generation, claim.from, claim.leaseOwner, + claim.through, ) .first<{ attempts: number | string; next_attempt_at: number | string | null }>(); if (!failed) { @@ -1117,7 +1153,8 @@ export async function skipChangeConsumer( .prepare( `UPDATE change_consumers SET acknowledged_position = ?, lease_owner = NULL, - lease_expires_at = NULL, attempts = 0, next_attempt_at = NULL, + lease_expires_at = NULL, lease_through_position = NULL, + attempts = 0, next_attempt_at = NULL, last_error_code = ?, last_error_at = ?, updated_at = ? WHERE consumer_id = ? AND generation_id = ? AND acknowledged_position = ? @@ -1230,25 +1267,63 @@ export async function pruneChanges( }; } + // created_at is captured before serialized position allocation, so age and + // position order may differ under overlapping projectors. Inspect only the + // bounded leading range and stop at the first age blocker; deleting every + // independently old row would create holes that retainedFloor cannot express. + const candidates = await db + .prepare( + `SELECT position, created_at FROM change_batches + WHERE generation_id = ? AND position > ? AND position <= ? + ORDER BY position LIMIT ?`, + ) + .bind(state.generation, state.retainedFloor, safeThrough, maxBatches) + .all<{ position: number | string; created_at: number | string }>(); + + let expected = BigInt(state.retainedFloor) + 1n; + let deleteThrough = state.retainedFloor; + for (const candidate of candidates.results) { + const position = BigInt(String(candidate.position)); + if (position !== expected) { + throw new ChangeHistoryGapError( + `Change log has a gap after retained floor ${state.retainedFloor}`, + ); + } + if ( + options.olderThan !== undefined && + Number(candidate.created_at) >= options.olderThan + ) { + break; + } + deleteThrough = String(candidate.position); + expected++; + } + + if (deleteThrough === state.retainedFloor) { + return { + pruned: 0, + retainedFloor: state.retainedFloor, + safeThrough, + done: true, + }; + } + const results = await db.batch([ db .prepare( `DELETE FROM change_batches - WHERE generation_id = ? AND position IN ( - SELECT position FROM change_batches - WHERE generation_id = ? AND position > ? AND position <= ? - AND (? IS NULL OR created_at < ?) - ORDER BY position LIMIT ? - )`, + WHERE generation_id = ? AND position > ? AND position <= ? + AND NOT EXISTS ( + SELECT 1 FROM change_consumers + WHERE generation_id = ? AND acknowledged_position < ? + )`, ) .bind( - state.generation, state.generation, state.retainedFloor, - safeThrough, - options.olderThan ?? null, - options.olderThan ?? null, - maxBatches, + deleteThrough, + state.generation, + deleteThrough, ), db .prepare( diff --git a/packages/contrail/src/core/db/schema.ts b/packages/contrail/src/core/db/schema.ts index d412723..8e89aea 100644 --- a/packages/contrail/src/core/db/schema.ts +++ b/packages/contrail/src/core/db/schema.ts @@ -789,6 +789,12 @@ export async function initSchema( "encoded_bytes", "INTEGER", ); + await addColumnIfNotExists( + db, + "change_consumers", + "lease_through_position", + dialect.bigintType, + ); await db .prepare( "UPDATE change_batches SET encoded_bytes = LENGTH(changes_json) WHERE encoded_bytes IS NULL", diff --git a/packages/contrail/tests/change-bootstrap.test.ts b/packages/contrail/tests/change-bootstrap.test.ts index c4733aa..73b2fea 100644 --- a/packages/contrail/tests/change-bootstrap.test.ts +++ b/packages/contrail/tests/change-bootstrap.test.ts @@ -356,4 +356,44 @@ describe("consumer-aware change pruning", () => { (await getChangesStatus(db)).consumers.find((item) => item.id === "late"), ).toMatchObject({ position: "5", backlogBatches: 0 }); }); + + it("keeps age-limited pruning to a contiguous retained prefix", async () => { + const db = createSqliteDatabase(":memory:"); + const resolved = config({ + keeper: { collections: [EVENT], initial: "history" }, + }); + await initSchema(db, resolved); + for (let position = 1; position <= 3; position++) { + await apply(db, resolved, event({ rkey: String(position), time: position })); + } + const claim = await claimChanges(db, "keeper", { now: 100 }); + await acknowledgeChanges(db, claim!, { now: 101 }); + + // Allocation order is serialized, but created_at is captured before that + // lock and can therefore be non-monotonic under overlapping projectors. + await db + .prepare( + `UPDATE change_batches SET created_at = CASE position + WHEN 1 THEN 100 WHEN 2 THEN 300 ELSE 100 END`, + ) + .run(); + + const pruned = await pruneChanges(db, { + maxBatches: 10, + olderThan: 200, + }); + expect(pruned).toEqual({ + pruned: 1, + retainedFloor: "1", + safeThrough: "3", + done: true, + }); + expect( + ( + await db + .prepare("SELECT position FROM change_batches ORDER BY position") + .all<{ position: number }>() + ).results.map((row) => row.position), + ).toEqual([2, 3]); + }); }); diff --git a/packages/contrail/tests/change-consumers.test.ts b/packages/contrail/tests/change-consumers.test.ts index f1d1528..3ec818a 100644 --- a/packages/contrail/tests/change-consumers.test.ts +++ b/packages/contrail/tests/change-consumers.test.ts @@ -343,16 +343,34 @@ describe("durable change consumers", () => { expect(claim?.changes).toHaveLength(2); }); - it("rejects forged generation cursors and never regresses a checkpoint", async () => { + it("rejects forged generation and range cursors without skipping delivery", async () => { const db = createSqliteDatabase(":memory:"); const config = readyEventConsumer(); await initSchema(db, config); - await apply(db, config, [mutation({ rkey: "one", sourceTime: 1 })]); - const claim = await claimChanges(db, "search", { now: 100 }); - const forged: ChangeClaim = { ...claim!, generation: crypto.randomUUID() }; + for (let position = 1; position <= 3; position++) { + await apply(db, config, [ + mutation({ rkey: String(position), sourceTime: position }), + ]); + } + const claim = await claimChanges(db, "search", { + now: 100, + maxBatches: 1, + }); + expect(claim?.through).toBe("1"); + + const wrongGeneration: ChangeClaim = { + ...claim!, + generation: crypto.randomUUID(), + }; await expect( - acknowledgeChanges(db, forged, { now: 101 }), + acknowledgeChanges(db, wrongGeneration, { now: 101 }), ).rejects.toBeInstanceOf(ChangeLeaseLostError); + + const beyondClaimedRange: ChangeClaim = { ...claim!, through: "3" }; + await expect( + acknowledgeChanges(db, beyondClaimedRange, { now: 101 }), + ).rejects.toBeInstanceOf(ChangeLeaseLostError); + await acknowledgeChanges(db, claim!, { now: 101 }); expect((await getChangesStatus(db)).consumers[0].position).toBe("1"); }); diff --git a/packages/contrail/tests/change-log.test.ts b/packages/contrail/tests/change-log.test.ts index 17f3303..851a83c 100644 --- a/packages/contrail/tests/change-log.test.ts +++ b/packages/contrail/tests/change-log.test.ts @@ -479,7 +479,7 @@ describe("transactional projection change log", () => { ); }); - it("upgrades retained pre-byte-count change batches without resetting consumers", async () => { + it("upgrades retained change-log columns without resetting consumers", async () => { const db = createSqliteDatabase(":memory:"); const resolved = loggedConfig(); await initSchema(db, resolved); @@ -487,6 +487,11 @@ describe("transactional projection change log", () => { await db .prepare("ALTER TABLE change_batches DROP COLUMN encoded_bytes") .run(); + await db + .prepare( + "ALTER TABLE change_consumers DROP COLUMN lease_through_position", + ) + .run(); await db .prepare( "UPDATE _contrail_meta SET value = 'old-change-schema' WHERE key = 'schema_fingerprint'", @@ -500,6 +505,13 @@ describe("transactional projection change log", () => { expect(row?.encoded_bytes).toBe( new TextEncoder().encode(row!.changes_json).byteLength, ); + expect( + ( + await db + .prepare("PRAGMA table_info(change_consumers)") + .all<{ name: string }>() + ).results.map((column) => column.name), + ).toContain("lease_through_position"); expect((await getChangeLogState(db))?.head).toBe("1"); }); -- 2.51.2