···7979await contrail.ingest();
8080```
81818282+### Persistent ingestion
8383+8484+```ts
8585+// Long-lived Jetstream connection with automatic batching and reconnection
8686+const controller = new AbortController();
8787+await contrail.runPersistent({
8888+ batchSize: 50, // flush every N events (default: 50)
8989+ flushIntervalMs: 5000, // or every N ms (default: 5000)
9090+ signal: controller.signal,
9191+});
9292+```
9393+9494+Call `controller.abort()` for graceful shutdown — the current batch is flushed and the cursor is saved.
9595+8296### Discover users and backfill
83978498```ts
···140154const contrail = new Contrail({ ...config, db });
141155```
142156143143-> **Note:** The SQLite adapter uses Node's built-in `node:sqlite` (Node 22+). Full-text search (`searchable`) is not supported with this adapter because `node:sqlite` doesn't include the FTS5 extension. Search works on Cloudflare D1, which has FTS5 enabled.
157157+> **Note:** The SQLite adapter uses Node's built-in `node:sqlite` (Node 22+). Full-text search (`searchable`) is not supported with this adapter because `node:sqlite` doesn't include the FTS5 extension. Search works on Cloudflare D1 and PostgreSQL.
144158145145-## Running the example (Cloudflare Workers)
159159+### PostgreSQL adapter (Node.js / server)
160160+161161+```ts
162162+import { createPostgresDatabase } from "@atmo-dev/contrail/postgres";
163163+import pg from "pg";
164164+165165+const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
166166+const db = createPostgresDatabase(pool);
167167+const contrail = new Contrail({ ...config, db });
168168+```
169169+170170+PostgreSQL uses JSONB for record storage, tsvector generated columns for full-text search (instead of FTS5), and `BIGINT` for timestamp columns.
171171+172172+## Examples
173173+174174+### PostgreSQL (Node.js)
175175+176176+See [`examples/postgres/`](examples/postgres/) for a complete example with Docker Compose, persistent Jetstream ingestion, user discovery/backfill, and an HTTP API server.
177177+178178+### Cloudflare Workers
146179147180This repo includes a working example that indexes AT Protocol calendar events and RSVPs on Cloudflare Workers + D1.
148181···190223| `references.*.field` | — | Field containing the target record's AT URI |
191224| `queries` | `{}` | Custom query handlers (raw Response) |
192225| `pipelineQueries` | `{}` | Custom query handlers that go through the standard filter/sort/hydration pipeline |
193193-| `searchable` | disabled | FTS5 search fields. Provide `string[]` to enable, omit to disable |
226226+| `searchable` | disabled | Full-text search fields. SQLite uses FTS5 virtual tables; PostgreSQL uses tsvector generated columns with GIN indexes. Provide `string[]` to enable, omit to disable |
194227195228### Top-level options
196229···251284?sort=rsvpsGoingCount&order=asc # by going count ascending
252285```
253286254254-**Search** uses SQLite FTS5 for ranked full-text search. To enable, set `searchable: ["field1", "field2"]` on a collection. Supports FTS5 syntax including prefix (`meetup*`), phrases (`"rust meetup"`), and boolean (`rust OR typescript`). Combinable with all other filters.
287287+**Search** uses SQLite FTS5 or PostgreSQL tsvector for ranked full-text search. To enable, set `searchable: ["field1", "field2"]` on a collection. Supports FTS5 syntax including prefix (`meetup*`), phrases (`"rust meetup"`), and boolean (`rust OR typescript`). Combinable with all other filters.
255288256289```
257290?search=meetup # basic search
···33export default defineConfig({
44 test: {
55 include: ["tests/**/*.test.ts"],
66+ // PostgreSQL tests share a single database and cannot run in parallel
77+ fileParallelism: false,
68 },
79});
+5
.changeset/tall-seals-sink.md
···11+---
22+"@atmo-dev/contrail": patch
33+---
44+55+add postgres adapter and example
+94
bin/ingest.ts
···11+/**
22+ * Persistent Jetstream ingestion with runtime configuration.
33+ *
44+ * Connects to an externally-provided database and subscribes to Jetstream.
55+ * The container does not own the database — it receives a connection URL.
66+ *
77+ * Environment:
88+ * JETSTREAM_URL - Jetstream WebSocket URL (required)
99+ * CONTRAIL_ADAPTER - "postgres" or "sqlite" (default: "postgres")
1010+ * DATABASE_URL - PostgreSQL connection string (required when adapter=postgres)
1111+ * SQLITE_PATH - SQLite file path (required when adapter=sqlite)
1212+ * CONTRAIL_CONFIG - Path to config file (default: "/app/config/contrail.config.ts")
1313+ * FLUSH_INTERVAL_MS - Batch flush interval in ms (default: 500)
1414+ */
1515+1616+import { Contrail } from "../src/index.ts";
1717+import type { ContrailConfig, Database } from "../src/index.ts";
1818+1919+const ADAPTER = process.env.CONTRAIL_ADAPTER ?? "postgres";
2020+const JETSTREAM_URL = process.env.JETSTREAM_URL;
2121+const CONFIG_PATH = process.env.CONTRAIL_CONFIG ?? "/app/config/contrail.config.ts";
2222+const FLUSH_INTERVAL_MS = parseInt(process.env.FLUSH_INTERVAL_MS ?? "500", 10);
2323+2424+async function createDatabase(): Promise<{ db: Database; cleanup: () => Promise<void> }> {
2525+ if (ADAPTER === "postgres") {
2626+ const url = process.env.DATABASE_URL;
2727+ if (!url) {
2828+ console.error("DATABASE_URL is required when CONTRAIL_ADAPTER=postgres");
2929+ process.exit(1);
3030+ }
3131+ const pg = await import("pg");
3232+ const { createPostgresDatabase } = await import("../src/adapters/postgres.ts");
3333+ const pool = new pg.default.Pool({ connectionString: url });
3434+ return { db: createPostgresDatabase(pool), cleanup: () => pool.end() };
3535+ }
3636+3737+ if (ADAPTER === "sqlite") {
3838+ const path = process.env.SQLITE_PATH;
3939+ if (!path) {
4040+ console.error("SQLITE_PATH is required when CONTRAIL_ADAPTER=sqlite");
4141+ process.exit(1);
4242+ }
4343+ const { createSqliteDatabase } = await import("../src/adapters/sqlite.ts");
4444+ return { db: createSqliteDatabase(path), cleanup: async () => {} };
4545+ }
4646+4747+ console.error(`Unknown adapter: ${ADAPTER}. Use "postgres" or "sqlite".`);
4848+ process.exit(1);
4949+}
5050+5151+async function loadConfig(): Promise<ContrailConfig> {
5252+ const module = await import(CONFIG_PATH);
5353+ return module.config ?? module.default;
5454+}
5555+5656+async function main() {
5757+ if (!JETSTREAM_URL) {
5858+ console.error("JETSTREAM_URL is required");
5959+ process.exit(1);
6060+ }
6161+6262+ const config = await loadConfig();
6363+ config.jetstreams = [JETSTREAM_URL];
6464+6565+ const { db, cleanup } = await createDatabase();
6666+ const contrail = new Contrail({ ...config, db });
6767+6868+ const controller = new AbortController();
6969+ const shutdown = () => {
7070+ console.log("\nShutting down gracefully...");
7171+ controller.abort();
7272+ };
7373+ process.on("SIGTERM", shutdown);
7474+ process.on("SIGINT", shutdown);
7575+7676+ console.log("Contrail ingester starting");
7777+ console.log(` Adapter: ${ADAPTER}`);
7878+ console.log(` Jetstream: ${JETSTREAM_URL}`);
7979+ console.log(` Config: ${CONFIG_PATH}`);
8080+ console.log(` Flush: ${FLUSH_INTERVAL_MS}ms`);
8181+8282+ await contrail.runPersistent({
8383+ signal: controller.signal,
8484+ flushIntervalMs: FLUSH_INTERVAL_MS,
8585+ });
8686+8787+ await cleanup();
8888+ console.log("Done.");
8989+}
9090+9191+main().catch((err) => {
9292+ console.error(err);
9393+ process.exit(1);
9494+});
+7
src/contrail.ts
···99import type { BackfillAllOptions, BackfillProgress } from "./core/backfill";
1010import { processNotifyUris } from "./core/router/notify";
1111import type { NotifyResult } from "./core/router/notify";
1212+import { runPersistent as runPersistentIngestion } from "./core/persistent";
1313+import type { PersistentIngestOptions } from "./core/persistent";
12141315export interface ContrailOptions extends ContrailConfig {
1416 db?: Database;
···4951 /** Run one Jetstream ingestion cycle (catches up to present, then stops). */
5052 async ingest(options?: { timeoutMs?: number }, db?: Database): Promise<void> {
5153 await runIngestCycle(this.getDb(db), this.config, options?.timeoutMs, this._ingestState);
5454+ }
5555+5656+ /** Run persistent Jetstream ingestion (long-lived, stays connected). */
5757+ async runPersistent(options?: Omit<PersistentIngestOptions, 'logger'>, db?: Database): Promise<void> {
5858+ await runPersistentIngestion(this.getDb(db), this.config, { ...options, logger: this.config.logger });
5259 }
53605461 /** Discover users from relays. Returns discovered DIDs. */
+3
src/index.ts
···2626export type { QueryOptions, SortOption } from "./core/db/records";
2727export type { BackfillProgress, BackfillAllOptions } from "./core/backfill";
2828export type { NotifyResult } from "./core/router/notify";
2929+3030+export { runPersistent } from "./core/persistent";
3131+export type { PersistentIngestOptions } from "./core/persistent";
···11+# Contrail — PostgreSQL Example
22+33+A complete example of using Contrail to index AT Protocol calendar events and RSVPs with PostgreSQL. Includes persistent Jetstream ingestion (long-running listener) and user discovery/backfill.
44+55+## Setup
66+77+```bash
88+# Copy this folder to a new project
99+cp -r examples/postgres my-contrail-app
1010+cd my-contrail-app
1111+1212+# Install dependencies
1313+npm install
1414+```
1515+1616+> **Note:** The `contrail` dependency in `package.json` points at `github:flo-bit/contrail`.
1717+> If you're using a fork with PostgreSQL support that hasn't been merged yet, update the
1818+> dependency to point at your fork's branch:
1919+>
2020+> ```bash
2121+> npm install github:your-username/contrail#your-branch
2222+> ```
2323+>
2424+> Or install from a local checkout:
2525+>
2626+> ```bash
2727+> npm install /path/to/your/contrail
2828+> ```
2929+3030+### Start PostgreSQL
3131+3232+**Option A: Docker (recommended)**
3333+3434+```bash
3535+docker compose up -d
3636+```
3737+3838+This starts PostgreSQL on port 5433 (to avoid conflicts with a local instance) with a `contrail` database ready to use. Override the port with `PG_PORT=5432 docker compose up -d`.
3939+4040+**Option B: Native PostgreSQL**
4141+4242+```bash
4343+createdb contrail
4444+```
4545+4646+Set `DATABASE_URL` to point at your local instance:
4747+4848+```bash
4949+export DATABASE_URL="postgresql://user:password@localhost:5432/contrail"
5050+```
5151+5252+## Configure
5353+5454+Edit `config.ts` to define your collections, queryable fields, relations, and references. See the [Contrail README](https://github.com/flo-bit/contrail) for all options.
5555+5656+## Run
5757+5858+### 1. Discover users and backfill records
5959+6060+```bash
6161+npm run sync
6262+```
6363+6464+This finds users from ATProto relays and backfills their existing records from PDS. Safe to interrupt and restart — progress is saved per-DID in the database.
6565+6666+### 2. Start persistent ingestion
6767+6868+```bash
6969+npm run ingest
7070+```
7171+7272+This opens a long-lived Jetstream connection and continuously indexes new records as they appear on the network. Events are batched and flushed every 5 seconds (or every 50 events, whichever comes first). Handles reconnection automatically.
7373+7474+Press `Ctrl+C` for graceful shutdown — the current batch is flushed and the cursor is saved so the next run picks up where it left off.
7575+7676+### 3. Serve the XRPC API
7777+7878+```bash
7979+npm run serve
8080+```
8181+8282+Your XRPC API is now available at `http://localhost:3000`:
8383+8484+```
8585+# List events sorted by RSVP count
8686+/xrpc/community.lexicon.calendar.event.listRecords?sort=rsvpsCount
8787+8888+# Upcoming events with 10+ going RSVPs
8989+/xrpc/community.lexicon.calendar.event.listRecords?startsAtMin=2026-03-16&rsvpsGoingCountMin=10
9090+9191+# Single event with hydrated RSVPs and profiles
9292+/xrpc/community.lexicon.calendar.event.getRecord?uri=at://...&hydrateRsvps=10&profiles=true
9393+9494+# Search events
9595+/xrpc/community.lexicon.calendar.event.listRecords?search=meetup
9696+9797+# RSVPs for a specific event
9898+/xrpc/community.lexicon.calendar.rsvp.listRecords?subjectUri=at://...
9999+```
100100+101101+## Running everything together
102102+103103+In production you'd typically run sync once (or periodically), then keep `ingest` and `serve` running as separate processes:
104104+105105+```bash
106106+# Initial sync (run once, or periodically to discover new users)
107107+npm run sync
108108+109109+# In separate terminals (or use a process manager)
110110+npm run ingest
111111+npm run serve
112112+```