# contrail Full Documentation

# Contrail

> **pre-alpha.** Expect breaking changes.

a library for easily creating (serverless) atproto backends/appviews.

- declare collections
- get automatic jetstream backfill and ingestion, typed XRPC endpoints
- optional: permissioned spaces and group-controlled communities

mostly tested on cloudflare workers with d1 but should run in any node env too 
(+ has adapters for node:sqlite and postgres for the db).

## Install

```bash
pnpm add @atmo-dev/contrail
```

## Minimal example

a complete cloudflare worker that indexes public calendar events from the atproto network and serves them over a typed XRPC endpoint. two files + config. a runnable version lives in [`apps/cloudflare-workers`](https://github.com/flo-bit/contrail/tree/main/apps/cloudflare-workers) — clone, deploy, `pnpm contrail backfill --remote`, done.

**`src/contrail.config.ts`** — picked up automatically by the `contrail` CLI:

```ts
import type { ContrailConfig } from "@atmo-dev/contrail";

export const config: ContrailConfig = {
  namespace: "com.example",
  collections: {
    event: {
      collection: "community.lexicon.calendar.event", // NSID to index
      queryable: { startsAt: { type: "range" } },     // ?startsAtMin=...
      searchable: ["name", "description"],            // ?search=...
    },
  },
};
```

**`src/worker.ts`** — four lines. `createWorker` wires up fetch + scheduled + lazy init:

```ts
import { createWorker } from "@atmo-dev/contrail/worker";
import { config } from "./contrail.config";
import { lexicons } from "../lexicons/generated";

export default createWorker(config, { lexicons });
```

`lexicons/generated/` is produced by `contrail-lex generate`; passing `{ lexicons }` exposes them at `/lexicons` so consumer apps can typegen against your deployed service. Drop it if you don't need that.

and a d1 binding + cron in `wrangler.jsonc`:

```jsonc
{
  "main": "src/worker.ts",
  "d1_databases": [{ "binding": "DB", "database_name": "contrail", "database_id": "..." }],
  "triggers": { "crons": ["*/1 * * * *"] }
}
```

then:

```bash
npx wrangler d1 create contrail   # copy the id into wrangler.jsonc
pnpm wrangler deploy              # deploy the worker
pnpm contrail backfill --remote   # one-shot historical backfill
```

the worker keeps itself fresh from now on via the cron. hit:

```
GET https://<your-worker>.workers.dev/xrpc/com.example.event.listRecords?startsAtMin=2026-01-01&limit=10
```

returns every `community.lexicon.calendar.event` record published anywhere on atproto that matches, as JSON. that's it — no PDS setup, no lexicon publishing, no relay configuration. everything scales from there: add filters, add full-text search, add more collections, turn on [spaces](https://github.com/flo-bit/contrail/blob/main/docs/05-spaces.md) for private records, mount the handler in sveltekit instead, swap the adapter for postgres.

**not using workers?** same library, different `db`. see [adapters](https://github.com/flo-bit/contrail/blob/main/docs/01-indexing.md#adapters) for node:sqlite and postgres.

## Docs

- [Indexing](https://github.com/flo-bit/contrail/blob/main/docs/01-indexing.md) — the core: collections, ingestion, adapters
- [Querying](https://github.com/flo-bit/contrail/blob/main/docs/02-querying.md) — filters, sorts, hydration, search, pagination
- [Lexicons](https://github.com/flo-bit/contrail/blob/main/docs/03-lexicons.md) — `contrail-lex` CLI, codegen, publishing
- [Auth](https://github.com/flo-bit/contrail/blob/main/docs/04-auth.md) — service-auth JWTs, invite tokens, watch tickets, OAuth permission sets
- [Spaces](https://github.com/flo-bit/contrail/blob/main/docs/05-spaces.md) — permissioned records stored by the appview
- [Communities](https://github.com/flo-bit/contrail/blob/main/docs/06-communities.md) — group-controlled atproto DIDs
- [Sync](https://github.com/flo-bit/contrail/blob/main/docs/07-sync.md) — reactive client-side store over `watchRecords`
- [Labels](https://github.com/flo-bit/contrail/blob/main/docs/08-labels.md) — atproto-native moderation hydration from external labelers
- Frameworks: [SvelteKit + Cloudflare](https://github.com/flo-bit/contrail/blob/main/docs/frameworks/sveltekit-cloudflare.md)

## Packages

| Package | |
|---|---|
| `@atmo-dev/contrail` | Core library — indexing, XRPC server, spaces, communities, realtime |
| `@atmo-dev/contrail-sync` | Client-side reactive watch-store with optional IndexedDB cache |
| `@atmo-dev/contrail-lexicons` | Codegen + `contrail-lex` CLI |

Working in this repo? See [development.md](https://github.com/flo-bit/contrail/blob/main/development.md) for the monorepo layout and commands.

# 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 }); // fetch history for registered DIDs
```

`backfill()` picks up where it left off across runs — safe to re-run.

### 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 <path>`. Other flags: `--binding <name>` (default `DB`), `--concurrency <n>` (default 100).

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(contrail.ingest({}, env.DB));
}
```

`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 prompts you to run backfill or refresh if the local DB looks stale on start.

### 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

## Refresh (catch-up after outages / dev idle)

When Jetstream drops events — you went offline for a few days in dev, there was an outage, or you just want reassurance nothing was lost — `refresh` walks every known DID's PDS and reconciles against your DB:

```bash
pnpm contrail refresh                    # totals only
pnpm contrail refresh --by-collection    # totals + per-collection breakdown
pnpm contrail refresh --ignore-window 30 # grace seconds (default: 60)
```

Each record is classified as:

- **missing** — PDS has it, DB doesn't. Inserted.
- **stale update** — DB has it with a different CID, *and* the DB row is older than the ignore window. Upserted.
- **in sync** — same CID, or DB row is within the ignore window (jetstream probably just hadn't caught up yet).

The ignore window is there so a refresh run seconds after a normal jetstream cycle doesn't double-count records that are about to sync anyway. Records inside the window are still written if they differ; they just don't show up in the stats.

Report shape (`--by-collection`):

```
by collection:
  community.lexicon.calendar.event
    3 missing, 1 stale updates, 842 in sync
  community.lexicon.calendar.rsvp
    12 missing, 0 stale updates, 4108 in sync

total:
  15 missing, 1 stale updates, 4950 in sync
  234 users scanned, 1 failed in 85.3s
  (ignore window: 60s)
```

Safe to run repeatedly — each pass converges toward zero missing / stale. Programmatic equivalent: `contrail.refresh({ ignoreWindowMs, concurrency })` returns the same structure.

Refresh is **not** a replacement for `ingest`/`runPersistent` — it walks every user's full history, which is expensive. Use it after outages or during dev idle, not as a continuous freshness mechanism.

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 | `true` opens `notifyOfUpdate`; a string requires `Bearer` |
| `spaces` | — | See [Spaces](./05-spaces.md) |
| `community` | — | See [Communities](./06-communities.md) |
| `realtime` | — | See [Sync](./07-sync.md) |
| `labels` | — | See [Labels](./08-labels.md) |

# 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 |

Plus a few top-level ones: `{namespace}.getProfile`, `{namespace}.getCursor`, `{namespace}.getOverview`, `{namespace}.notifyOfUpdate`, `{namespace}.permissionSet`, `{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=…`.

## 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).

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=<opaque>
```

`cursor` is opaque — pass back whatever `listRecords` returned in its `cursor` field. `limit` is 1–200 (default 50). Cursors 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
```

# Lexicons

Contrail emits lexicon JSON for every XRPC method it exposes. Your app needs those files for two things: **generating typed TypeScript clients** and **publishing to a PDS** so other apps can discover your schemas. The `contrail-lex` CLI handles both.

```bash
pnpm add -D @atmo-dev/contrail-lexicons @atcute/lex-cli
```

`@atcute/lex-cli` is a peer dep — you pin the version.

## CLI

```bash
contrail-lex generate       # emit lexicon JSON from your Contrail config
contrail-lex pull           # wraps `lex-cli pull` (fetch external lexicons)
contrail-lex types          # wraps `lex-cli generate` (JSON → TS types)
contrail-lex all            # generate → pull → generate → pull → types
contrail-lex all --no-types # same, skip the type step
contrail-lex publish        # publish lexicons to your PDS (add --dry-run to preview)
contrail-lex pull-service <url>  # consume a deployed contrail's /lexicons endpoint
```

Config is auto-detected at `contrail.config.ts`, `src/contrail.config.ts`, `src/lib/contrail.config.ts`, or `app/contrail.config.ts`. Override with `--config <path>`.

## What lands where

Running `contrail-lex all` organises everything under one `lexicons/` directory:

```
lexicons/
  custom/               # hand-authored lexicons you write (optional)
  generated/            # JSON emitted from your Contrail config
  pulled/               # external NSIDs fetched by lex-cli pull
lex.config.js           # regenerated each run (gitignore)
src/lexicon-types/      # TS types from lex-cli generate
```

`lexicons/custom/`, `lexicons/generated/`, and `lexicons/pulled/` should be **committed** — that way CI doesn't need network access. `lex.config.js` and `src/lexicon-types/` are regenerated on demand and safe to gitignore.

`lexicons/generated/index.ts` is also emitted on every run — a barrel that imports every lexicon the deployment speaks (generated + pulled + custom). Pass it to `createWorker(config, { lexicons })` to expose them at `/xrpc/<namespace>.lexicons` on your deployed service; consumer apps can then `pull-service` against it.

## Consuming a deployed contrail

If you're building a frontend that talks to someone's (or your own) deployed contrail, you don't need the backend's source code to get typed XRPC calls. Have the operator pass `{ lexicons }` to `createWorker`, then:

```bash
contrail-lex pull-service https://my-contrail.dev/xrpc/com.example.lexicons
# or: contrail-lex pull-service https://my-contrail.dev --namespace com.example

npx lex-cli generate
```

`pull-service` hits the manifest endpoint, writes each lexicon under `lexicons/pulled/<nsid>.json`, and `lex-cli generate` emits TypeScript types for `@atcute/client`. Override the output dir with `--out <path>`.

The manifest includes the service's generated lexicons *plus* any external NSIDs the generator `$ref`s (e.g., `app.bsky.actor.profile`, `community.lexicon.calendar.event`) — so typegen resolves cleanly with no additional fetching from bsky / atproto registries.

No PDS setup required, no DNS, no published lexicon records — just an HTTP endpoint and typegen.

## Publishing

Once your lexicons are generated and committed, publish them as `com.atproto.lexicon.schema` records on your PDS so other apps can resolve them:

```bash
contrail-lex publish <handle-or-did> <app-password>

# or via env vars (nicer for CI):
LEXICON_ACCOUNT_IDENTIFIER=you.bsky.social \
LEXICON_ACCOUNT_PASSWORD=xxxx-xxxx-xxxx-xxxx \
  contrail-lex publish
```

Writes each lexicon to `at://<did>/com.atproto.lexicon.schema/<nsid>`. You'll also need DNS TXT records for your NSID authorities — the command prints the exact records you need and won't proceed until you confirm (override with `--skip-confirm` in CI).

Preview first with `--dry-run`:

```bash
contrail-lex publish --dry-run
```

Walks your `lexicons/generated/` dir, prints the NSIDs it would publish and the exact TXT records you'd need. Doesn't log in or write anything. Credentials aren't required for dry-run, so it's safe to run without secrets configured — useful for sanity-checking what a release would push.

**One-off flow:** generate once, commit, `contrail-lex publish --dry-run` to preview, `contrail-lex publish` once per version bump. You don't need to re-publish unless your lexicon JSON changes.

## Programmatic API

If you need to call the generator from code — e.g. to derive the XRPC method list for an OAuth permission set:

```ts
import { generateLexicons, extractXrpcMethods } from "@atmo-dev/contrail-lexicons";

const generated = generateLexicons({
  config,
  rootDir: process.cwd(),
  outputDir: "lexicons/generated",
});

const methods = extractXrpcMethods(generated); // every query + procedure NSID
```

`publishLexicons` is also exported for custom flows, but the CLI covers the common case.

# Auth

Contrail has four auth mechanisms. Which one applies depends on who's calling and what they're asking for.

| Mechanism | Used by | For |
|---|---|---|
| Anonymous | anyone | public reads |
| Service-auth JWT | third-party apps acting on behalf of a user | anything permissioned (spaces, communities) |
| In-process server client | your own server code | loaders / actions that skip HTTP entirely |
| Invite token | anonymous bearers | read-only access to a specific space |
| Watch ticket | browsers | realtime subscriptions (`watchRecords`) |

## Service-auth JWTs

The standard atproto mechanism. When a third-party app wants to call your contrail service as a user, it:

1. Asks the user's PDS to mint a service-auth JWT with `com.atproto.server.getServiceAuth` — the token's claims include `iss` (user's DID), `aud` (your service DID), `lxm` (the specific method NSID), and a short expiry.
2. Sends the request to your service with `Authorization: Bearer <jwt>` and `Atproto-Proxy: <did>#<service-id>` so the PDS knows where to route.

Contrail verifies every request against the public key in the issuer's DID doc (`@atcute/xrpc-server` does the heavy lifting). It checks:

- Signature valid
- `aud` matches the `serviceDid` you configured
- `lxm` covers the method being called
- Token hasn't expired

On pass, your handler sees a populated `serviceAuth = { issuer, audience, lxm, clientId }` context and can proceed. On fail, 401 or 403 with a structured reason.

### The `serviceDid` gotcha

Use the **plain DID** (no `#fragment`) when configuring contrail:

```ts
spaces: { serviceDid: "did:web:example.com" }  // right
spaces: { serviceDid: "did:web:example.com#com_example_space" }  // wrong
```

Many PDS implementations reject `aud` values containing `#fragment` in `com.atproto.server.getServiceAuth`, and contrail does strict string equality on `aud`. The fragment form belongs only in your DID doc's `service` entry, where PDSes use it to resolve the service endpoint URL for `Atproto-Proxy` routing — that's separate from JWT audience validation.

## In-process server client

When your own server code wants to call contrail, the service-auth dance is pointless — it's your code talking to your code. `createServerClient` skips it:

```ts
import { createServerClient } from "@atmo-dev/contrail/server";

const client = createServerClient(async (req) => handle(req, env.DB), userDid);

// Calls bypass fetch entirely; acts as `userDid` for ACL purposes.
const res = await client.get("com.example.event.listRecords", { params: {...} });
```

Pass `did` to act as that user; omit it for anonymous calls against public endpoints. This is a trust boundary — anything that actually crosses a network needs a real service-auth JWT, not this shortcut.

See [SvelteKit + Cloudflare](./frameworks/sveltekit-cloudflare.md) for the typical loader pattern.

## Invite tokens

First-class auth for spaces. When a space owner creates an invite:

```
com.example.space.invite.create  { spaceUri, ttl?, maxUses? }
  → { token: "...plaintext..." }  // returned once, never again
```

The plaintext token is handed to the user out-of-band (link, QR, email). Contrail stores only a SHA-256 hash. Redemption is one atomic UPDATE: `used_count++ WHERE hash = ? AND !revoked AND !expired AND !exhausted`.

Three invite kinds, depending on what the token does:

- **`join`** — redeemed via `com.example.space.invite.redeem` with a service-auth JWT. Adds the caller's DID to the member list. Members have full read + write inside the space; there's no per-member permission axis beyond "is a member."
- **`read`** — bearer-only. The token itself grants read access when passed as `?inviteToken=<plaintext>`, no DID, no redemption. Good for sharing a read-only link that doesn't add anyone to the member list.
- **`read-join`** — both. Works anonymously as a read token; can also be redeemed with a JWT to promote the caller to member.

Tokens can be revoked (`invite.revoke`), expire automatically (`ttl`), and be exhausted (`maxUses`).

## Watch tickets

Realtime subscriptions (`watchRecords`) can't use regular service-auth JWTs for two reasons: the WebSocket upgrade can't carry arbitrary headers, and an open socket would outlive a 60s JWT TTL. So contrail uses separate short-lived tickets.

Server-side minting comes in two flavours:

- `com.example.realtime.ticket` — POST `{ topic }` (e.g. `"space:ats://..."`) → `{ ticket, topics, expiresAt }`. Bare topic-list ticket, used with the generic `<ns>.realtime.subscribe` endpoint.
- `<collection>.watchRecords?mode=ws&spaceUri=…` (or `&actor=…`) handshake — returns `{ snapshot, ticket, wsUrl, sinceTs, ticketTtlMs, querySpec }`. The ticket is bound to `(did, topics, querySpec)` and is the one to use for the per-collection `watchRecords` stream — both for SSE (`?ticket=…`) and the subsequent WS upgrade.

Both flavours are signed by `realtime.ticketSecret` (a 32-byte random, configured once). Clients hand the ticket off via `?ticket=...` on connect.

In the `@atmo-dev/contrail-sync` client:

```ts
createWatchStore({
  url: "/xrpc/com.example.message.watchRecords?spaceUri=ats://...",
  mintTicket: async () => (await fetch("/api/ticket")).then((r) => r.text()),
});
```

Each reconnect mints a fresh ticket, so expiry doesn't matter for long-lived subscriptions. See [Sync](./07-sync.md) for the full flow.

## OAuth permission sets

When a user grants a third-party app permission to act as them in your contrail service, the consent screen is driven by a **permission set** — a lexicon that bundles every XRPC method you expose. Contrail auto-generates `{namespace}.permissionSet` for you; `contrail-lex` publishes it alongside the other lexicons.

A third-party app requests scope by referencing your permission set's NSID in its OAuth metadata:

```jsonc
"scope": "include:com.example.permissionSet"
```

The user's PDS fetches that lexicon (via DNS-backed NSID resolution), shows the user what methods are being requested, and mints scoped service-auth JWTs on confirmation.

### DNS requirements

Permission sets live under *your* namespace (`com.example.permissionSet`), so PDSes resolve the NSID via DNS. Resolution does **not** walk up subdomains — every authority in your NSID tree needs its own `TXT` record at `_lexicon.<reversed-domain-path>`.

`contrail-lex publish` prints the exact records you need (also works as a `--dry-run`). Without them, permission sets can't be fetched, and users get errors instead of a consent screen.

## Anonymous / public

No auth needed for:

- `listRecords` / `getRecord` without `spaceUri` — returns public records
- `getProfile`, `getCursor`, `getOverview`
- `notifyOfUpdate` (unless you set `notify: "some-bearer-token"` in config; then it needs `Authorization: Bearer <that>`)

Public requests skip all verification middleware — no JWT parsing, no DID-doc fetch. Fast path.

## How the pieces fit

A typical flow for a third-party app acting as a user in a space:

1. App registers OAuth client pointing at your permission set NSID.
2. User grants consent — PDS fetches your permission set lexicon via DNS, shows the user the methods, records the scope.
3. App calls `com.atproto.server.getServiceAuth` on the user's PDS: `{ aud: "did:web:example.com", lxm: "com.example.space.putRecord", exp: <60s> }`. PDS signs, returns JWT.
4. App sends `PUT /xrpc/com.example.space.putRecord` with `Authorization: Bearer <jwt>` and `Atproto-Proxy: did:web:example.com#com_example_space`.
5. User's PDS reads the `Atproto-Proxy` header, resolves the service endpoint from your DID doc, forwards the request.
6. Contrail verifies the JWT (signature, `aud`, `lxm`, expiry), runs ACL check (is this DID a member of that space?), dispatches the write.

For your own loaders/actions, steps 3–6 collapse into a single `createServerClient({did}).post(...)` call. For a browser subscribing to a feed, steps 3–5 are replaced by a ticket mint from your server. Same auth model, different surface.

# Spaces

Auth-gated store for records that can't live on public PDSes — private events, invite-only groups, members-only chat. Opt-in; zero cost if you don't enable it.

## Mental model

> A **space** is a bag of records with one lock. The **member list** says who has the key.

- One owner (DID), one type (NSID), one key. Identified by `ats://<owner>/<type>/<key>` — distinct scheme from atproto record URIs (`at://`) so the two can't be confused at any layer.
- Every member (including owner) has read + write inside the space. Delete is scoped to your own records — no one can remove records they didn't author, owner included. To wipe everything, delete the space.
- Optional **app policy** gates which OAuth clients can act in the space.

Every permission boundary is its own space. No nested ACLs. Richer roles = more spaces or app-layer checks.

## Enable

```ts
import type { ContrailConfig } from "@atmo-dev/contrail";

const config: ContrailConfig = {
  namespace: "com.example",
  collections: { /* ... */ },
  spaces: {
    type: "com.example.event.space",
    serviceDid: "did:web:example.com",
  },
};
```

Each collection gets a parallel `spaces_records_<short>` table. Opt out per-collection:

```ts
public_only: { collection: "com.example.public", allowInSpaces: false }
```

## Auth

Spaces use the standard contrail auth surface — service-auth JWTs for third-party apps, in-process server clients for your own loaders, invite tokens for anonymous read-grant links. See [Auth](./04-auth.md) for the full picture.

Space-specific wiring:

- `serviceDid` in the config is the `aud` contrail expects on incoming JWTs. Plain DID, no `#fragment`.
- Apps acting in a space send `Atproto-Proxy: <serviceDid>#<service-id-from-your-did-doc>` so the user's PDS routes correctly.
- Invite redemption via service-auth JWT grants membership; via `?inviteToken=...` query param grants read-only bearer access to that space.

## Unified `listRecords`

| Call | Returns |
|---|---|
| no auth, no `spaceUri` | public only |
| `?spaceUri=…` + JWT | one space (ACL-gated) |
| JWT, no `spaceUri` | public **unioned** with every space the caller is a member of |

Filters, sorts, hydration, and references work across all three. Records from a space carry a `space: <spaceUri>` field — same on `listRecords`/`getRecord` responses and `watchRecords` stream events.

## Invites

First-class primitive — see [Auth § Invite tokens](./04-auth.md#invite-tokens) for the mechanism. Space-specific: create via `com.example.space.invite.create`, redeem via `.redeem` (membership grant) or `?inviteToken=...` query param (read-only bearer grant).

## XRPCs

- `com.example.space.create | get | list | delete`
- `com.example.space.putRecord | deleteRecord | listRecords | getRecord`
- `com.example.space.invite.create | redeem | revoke | list`
- `com.example.space.listMembers | removeMember`

## What's not here

- No E2EE (data is operator-readable).
- No FTS on `?spaceUri=…` yet.
- No per-space sharding — one DB, one operator.
- Not a long-term replacement for real atproto permissioned repos.

The design follows Daniel Holmgren's [permissioned data rough spec](https://dholms.leaflet.pub/3mhj6bcqats2o). The goal is that when real atproto permissioned repos ship, migration is mostly data movement — the API your app speaks doesn't change.

# Communities

Group-controlled atproto DIDs. A community is a DID whose signing/rotation keys are held by the appview on behalf of multiple members, with tiered access levels. Built on top of [spaces](./05-spaces.md).

## When to use this

When you want atproto records published under a *shared* identity — a team, a project, a channel — not a single user. Think: a group's published calendar events, a community's published posts.

## Two modes

- **Minted** — contrail creates a fresh `did:plc` for the community, holds the signing key plus one rotation key (a second rotation key is returned to the creator once for recovery), and publishes from it.
- **Adopted** — contrail takes over an existing account by holding an **app password** issued from its PDS. The owner's identity, signing key, and rotation keys are unchanged; contrail just gets PDS write access via the app password.

Either way, the result is the same: a DID that multiple members can act through, gated by access levels.

## Access levels

Each member has a level (ranked). Levels map to write permissions. Owners can grant/revoke levels. Two reserved levels exist: `owner` and `member`. Your deployment defines the rest:

```ts
community: {
  masterKey: ENV.COMMUNITY_MASTER_KEY,  // 32-byte encryption key for stored credentials
  serviceDid: "did:web:example.com",
  levels: ["admin", "moderator"],        // ranked, highest-first
}
```

Stored credentials (app passwords for adopted communities, signing keys for minted) are envelope-encrypted with `masterKey`. Never ship the placeholder.

## How it composes with spaces

A community *owns* spaces. Members of the community get access to community-owned spaces based on their access level. Grant access per-space per-level:

```
community.space.grant  { spaceUri, subject: { did: "did:plc:..." }, accessLevel: "admin" }
```

The spaces layer stays ignorant of access levels — it just sees "this DID is a member." The community layer projects member × level → membership in specific spaces. Once a DID is a member of a space (through a community grant or otherwise), they have full read + write inside it.

## XRPCs

- `com.example.community.mint | adopt | list | delete`
- `com.example.community.invite.create | redeem | revoke | list`
- `com.example.community.setAccessLevel | revoke | listMembers`
- `com.example.community.space.create | grant | revoke | ...` — community-owned spaces
- `com.example.community.putRecord | deleteRecord` — publish records as the community DID

## What's not here

- No per-record per-level ACLs. Model as spaces.
- No auto-rotation on key compromise yet.
- Adoption can be revoked unilaterally by the owner — they revoke the app password on their PDS and contrail loses write access. (Mint mode is the irreversible one: the creator's recovery rotation key, returned once at mint time, is the only path back if contrail's signing/rotation key is compromised.)

The design follows zicklag's [Arbiter design sketch](https://zicklag.leaflet.pub/3mjrvb5pul224) for group management on atproto. The post is an early design note; our implementation will track it as the spec firms up.

# Sync

Client-side reactive store over contrail's `watchRecords` endpoints. Subscribes once, reconciles forever, ships with optimistic updates and an optional IndexedDB cache.

Lives in its own package:

```bash
pnpm add @atmo-dev/contrail-sync
```

## Basic use

```ts
import { createWatchStore } from "@atmo-dev/contrail-sync";

const store = createWatchStore({
  url: "/xrpc/com.example.message.watchRecords?roomUri=at://...",
  transport: "sse", // or "ws"
});

store.subscribe(({ records, status }) => { /* re-render */ });
store.start();
```

Framework-agnostic — wrap it in Svelte `$state`, React `useSyncExternalStore`, Vue `ref`, whatever.

## Transports

- **SSE** (default) — one HTTP request, simplest. Works everywhere.
- **WS** — a two-step handshake: HTTP GET returns a snapshot + watch-scoped ticket, then you upgrade to WS. On Cloudflare, the WS terminates on a Durable Object that hibernates idle connections. Same event stream either way.

## Authenticated watches

Pass `mintTicket` for any non-public endpoint. One-shot string for SSR-minted tickets, function for fresh tickets per reconnect:

```ts
mintTicket: async () => (await fetch("/api/ticket")).then((r) => r.text()),
```

Tickets are minted server-side via `com.example.realtime.ticket` (or any app-specific route).

## Optimistic updates

```ts
store.addOptimistic({ rkey, did, value: { text: "hi" } });
// later, on mutation failure:
store.markFailed(rkey, err);
// or explicit rollback:
store.removeOptimistic(rkey);
```

When a real record with the same `rkey` arrives via the stream, the optimistic entry is dropped automatically.

## IndexedDB cache

Instant first paint from last session's records:

```ts
import { createIndexedDBCache } from "@atmo-dev/contrail-sync/cache-idb";

createWatchStore({
  url,
  cache: createIndexedDBCache(),
  cacheMaxRecords: 200,
});
```

Cached records show immediately; the live snapshot reconciles when the connection opens.

## Server-side config

Enable `watchRecords` emission in your Contrail config:

```ts
realtime: {
  ticketSecret: ENV.REALTIME_TICKET_SECRET, // 32 bytes
  pubsub: new DurableObjectPubSub(env.PUBSUB), // or in-memory for dev
}
```

See [Indexing](./01-indexing.md) for the full config surface.

## Lifecycle

```
idle → connecting → snapshot → live
                         ↓ (disconnect)
                    reconnecting → snapshot → live
                         ↓ (stop)
                       closed
```

Stale records stay visible across reconnects until the fresh snapshot arrives, at which point anything the server didn't re-send is evicted. Survives offline periods cleanly.

# 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 labels-backfill [--remote]` | Local script, drains until each labeler reports caught up |

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.
- **Live label updates on `watchRecords`.** The realtime stream snapshots labels with the initial query but does not push label deltas. Adding this means publishing `labels:<src>` topic events from the ingest worker and merging them in `runQueryStream` — future work.
- **Spaces / community labels.** Hydration already runs on the spaces read paths, so labels emitted by a labeler-DID member of a space (or under a community DID) will show up if you write them to the `labels` table. The auth surface for "make this DID a labeler in this space" is not yet exposed as XRPCs.
- **Outbound `subscribeLabels`.** Contrail does not republish labels. Communities-as-labelers (using the community DID as `src`) is a natural extension once you want to act as a labeler instead of just consume.
- **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.

# 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
pnpm add -D @atmo-dev/contrail-lexicons @atcute/lex-cli
```

## 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
scripts/
  append-scheduled.ts        # workaround — see "Cron" below
```

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

/** Typed in-process XRPC client for loaders / actions. Pass `did` to act as
 *  that user (no JWT / PDS roundtrip); omit for anonymous public reads. */
export function getServerClient(db: D1Database, did?: string): Client {
  return createServerClient(async (req) => {
    await ensureInit(db);
    return handle(req, db) as Promise<Response>;
  }, did);
}
```

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<Response>;
}

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, locals }) => {
  const rpc = getServerClient(platform!.env.DB, locals.did ?? undefined);
  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 fetch — the loader runs contrail's XRPC handler in-process, no extra network hop. `did` sets the caller identity without requiring a signed JWT (it's a server-to-server trust boundary; anything crossing an untrusted boundary still needs real service-auth).

## 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 easiest fix: an HTTP endpoint that does the ingest, plus a post-build script that appends a `scheduled` handler calling that endpoint.

**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");
};
```

**Post-build patch:**

```ts
// scripts/append-scheduled.ts
import { readFileSync, writeFileSync } from "fs";
import { join, dirname } from "path";
import { fileURLToPath } from "url";

const root = join(dirname(fileURLToPath(import.meta.url)), "..");
const workerPath = join(root, ".svelte-kit", "cloudflare", "_worker.js");

writeFileSync(
  workerPath,
  readFileSync(workerPath, "utf-8") +
    `
worker_default.scheduled = async function (event, env, ctx) {
  const req = new Request("http://localhost/api/cron", {
    method: "POST",
    headers: { "X-Cron-Secret": env.CRON_SECRET ?? "" },
  });
  ctx.waitUntil(this.fetch(req, env, ctx));
};
`
);
```

**Wire it into `build`:**

```jsonc
// package.json
"scripts": {
  "build": "vite build && tsx scripts/append-scheduled.ts"
}
```

`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
npx wrangler d1 create yourapp       # copy the id into wrangler.jsonc
pnpm build && pnpm wrangler deploy
npx 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.
- Need to reconcile after an outage? `pnpm contrail refresh --remote`.

## Where to go next

- [Indexing](../01-indexing.md) — config options, adapter choices
- [Querying](../02-querying.md) — filters, sorts, hydration, search
- [Lexicons](../03-lexicons.md) — generate TS types for your XRPC surface
- [Auth](../04-auth.md) — service-auth JWTs, invite tokens, watch tickets, OAuth permission sets
- [Spaces](../05-spaces.md) / [Communities](../06-communities.md) — private records + group-controlled DIDs, which both slot into the same handler you just mounted
- [Sync](../07-sync.md) — reactive client-side subscriptions (`createWatchStore`) wrapped in Svelte `$state`

## 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 the `append-scheduled.ts` patch 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.