Contrail#

A library for indexing AT Protocol records. Define collections — get automatic Jetstream ingestion, PDS backfill, user discovery, and typed XRPC endpoints. Works with Cloudflare Workers + D1, SvelteKit, Node.js, or any JavaScript runtime.

Install#

npm install @atmo-dev/contrail

Usage#

import { Contrail } from "@atmo-dev/contrail";

const contrail = new Contrail({
  namespace: "com.example",
  db, // any Database-compatible instance (D1, SQLite, etc.)
  collections: {
    "community.lexicon.calendar.event": {
      queryable: {
        mode: {},                        // string → equality filter (?mode=online)
        name: {},                        // string → equality filter (?name=...)
        startsAt: { type: "range" },     // range → min/max filters (?startsAtMin=...&startsAtMax=...)
        endsAt: { type: "range" },
      },
      searchable: ["name", "description"],
      relations: {
        rsvps: {
          collection: "community.lexicon.calendar.rsvp",
          groupBy: "status",
          count: true,
          groups: {
            interested: "community.lexicon.calendar.rsvp#interested",
            going: "community.lexicon.calendar.rsvp#going",
            notgoing: "community.lexicon.calendar.rsvp#notgoing",
          },
        },
      },
    },
    "community.lexicon.calendar.rsvp": {
      queryable: {
        status: {},
        "subject.uri": {},
      },
      references: {
        event: {
          collection: "community.lexicon.calendar.event",
          field: "subject.uri",
        },
      },
    },
  },
});

await contrail.init();

Query records#

const { records, cursor } = await contrail.query(
  "community.lexicon.calendar.event",
  {
    filters: { mode: "in-person" },
    sort: { countType: "community.lexicon.calendar.rsvp", direction: "desc" },
    limit: 20,
  }
);

Ingest from Jetstream#

// Run one ingestion cycle (catches up to present, then stops)
await contrail.ingest();

Persistent ingestion#

// Long-lived Jetstream connection with automatic batching and reconnection
const controller = new AbortController();
await contrail.runPersistent({
  batchSize: 50,           // flush every N events (default: 50)
  flushIntervalMs: 5000,   // or every N ms (default: 5000)
  signal: controller.signal,
});

Call controller.abort() for graceful shutdown — the current batch is flushed and the cursor is saved.

Discover users and backfill#

// Find users from relays
await contrail.discover();

// Backfill their records from PDS
await contrail.backfill({ concurrency: 100 });

// Or both in one call
await contrail.sync({ concurrency: 100 });

Notify of immediate updates#

// After writing to a user's PDS, tell Contrail to fetch it now
await contrail.notify("at://did:plc:abc/community.lexicon.calendar.rsvp/123");

// Batch up to 25 URIs
await contrail.notify([uri1, uri2, uri3]);

HTTP handler (XRPC endpoints)#

Mount the full XRPC API in any framework:

import { createHandler } from "@atmo-dev/contrail/server";

const handle = createHandler(contrail);
// handle: (Request, db?) => Promise<Response>

SvelteKit:

// src/routes/xrpc/[...path]/+server.ts
export const GET = ({ request }) => handle(request);
export const POST = ({ request }) => handle(request);

Cloudflare Worker:

export default {
  async fetch(request, env) {
    return handle(request, env.DB);
  },
};

SQLite adapter (Node.js / local dev)#

import { createSqliteDatabase } from "@atmo-dev/contrail/sqlite";

const db = createSqliteDatabase("data.db");
const contrail = new Contrail({ ...config, db });

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.

PostgreSQL adapter (Node.js / server)#

import { createPostgresDatabase } from "@atmo-dev/contrail/postgres";
import pg from "pg";

const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
const db = createPostgresDatabase(pool);
const contrail = new Contrail({ ...config, db });

PostgreSQL uses JSONB for record storage, tsvector generated columns for full-text search (instead of FTS5), and BIGINT for timestamp columns.

Examples#

PostgreSQL (Node.js)#

See examples/postgres/ for a complete example with Docker Compose, persistent Jetstream ingestion, user discovery/backfill, and an HTTP API server.

Cloudflare Workers#

This repo includes a working example that indexes AT Protocol calendar events and RSVPs on Cloudflare Workers + D1.

Setup#

pnpm install
pnpm generate:pull   # pull lexicons from network, auto-detect fields, generate types

Dev#

pnpm sync            # discover users and backfill records from PDS
pnpm dev:auto        # start wrangler dev with auto-ingestion

Production#

npx wrangler d1 create contrail
# Add database_id to wrangler.jsonc
pnpm deploy
pnpm sync            # discover + backfill against prod D1

Ingestion runs automatically via cron (*/1 * * * *). Schema is auto-initialized.

Config#

Collection options#

Top-level options#

Profiles#

profiles is a top-level config array of collection NSIDs that contain profile records (rkey self). Defaults to ["app.bsky.actor.profile"]. These are auto-added to collections with { discover: false }. Use ?profiles=true on any endpoint to include a profiles map in the response, keyed by DID, with handle and profile record data.

XRPC API#

When using createHandler, all endpoints are available at /xrpc/{nsid}.{method}:

Query parameters#

Filtering:

Sorting — sort accepts any queryable field param name or a count field:

?sort=startsAt                 # by date (default: desc for range fields)
?sort=name&order=asc           # by name ascending
?sort=rsvpsCount               # by total RSVP count (default: desc)
?sort=rsvpsGoingCount&order=asc  # by going count ascending

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.

?search=meetup                          # basic search
?search=meetup&mode=online              # search + filter
?search=rust*&sort=startsAt&order=asc   # search + sort override

Hydration embeds related or referenced records inline:

?hydrateRsvps=5              # latest 5 RSVPs per group (going, interested, etc.)
?hydrateEvent=true           # embed the referenced event record
?hydrateRsvps=5&hydrateEvent=true   # combine both

Examples (events)#

# Upcoming events with 10+ going RSVPs, with RSVP records and profiles
/xrpc/community.lexicon.calendar.event.listRecords?startsAtMin=2026-03-16&rsvpsGoingCountMin=10&hydrateRsvps=5&profiles=true

# Events for a specific user (by handle)
/xrpc/community.lexicon.calendar.event.listRecords?actor=alice.bsky.social&profiles=true

# Single event with counts, RSVPs, and profiles
/xrpc/community.lexicon.calendar.event.getRecord?uri=at://did:plc:.../community.lexicon.calendar.event/...&hydrateRsvps=10&profiles=true

# Search for events by name/description
/xrpc/community.lexicon.calendar.event.listRecords?search=meetup&profiles=true

# RSVPs for a specific event, with the referenced event embedded
/xrpc/community.lexicon.calendar.rsvp.listRecords?subjectUri=at://did:plc:.../community.lexicon.calendar.event/...&hydrateEvent=true&profiles=true

Notify of Updates#

By default, Contrail ingests from Jetstream every minute (in the Worker example). If your app writes to a user's PDS and needs the change reflected immediately, use contrail.notify() or call the XRPC endpoint:

// Programmatic
await contrail.notify(uri);

// Or via HTTP
await fetch("https://your-contrail.workers.dev/xrpc/com.example.notifyOfUpdate", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ uri }),
});

Contrail fetches the record from the user's PDS and figures out what to do:

When Jetstream later delivers the same event, the duplicate is detected by CID and skipped.

Typesafe Client Usage#

You can get fully typed XRPC queries for any Contrail instance using @atcute/lex-cli. The lexicon files are committed to the repo, so you can pull them directly via the git source.

Setup#

npm install @atcute/client @atcute/lexicons @atcute/lex-cli

Create a lex.config.js pointing at the Contrail instance's repo:

import { defineLexiconConfig } from "@atcute/lex-cli";

export default defineLexiconConfig({
  outdir: "src/lexicon-types/",
  imports: ["@atcute/atproto"],
  files: ["lexicons/**/*.json"],
  pull: {
    outdir: "lexicons/",
    sources: [
      {
        type: "git",
        remote: "https://github.com/USER/REPO.git", // the Contrail instance repo
        pattern: ["lexicons-generated/**/*.json", "lexicons-pulled/**/*.json", "lexicons/**/*.json"],
      },
    ],
  },
});

Then pull and generate:

npx lex-cli pull && npx lex-cli generate

Usage#

Import the generated types (side-effect import registers them with @atcute/client), then query with full type safety:

import "./lexicon-types/index.js"; // registers ambient types
import { Client } from "@atcute/client";

const rpc = new Client({ handler: simpleFetchHandler({ service: /* your contrail url */ }) });

const response = await rpc.get("community.lexicon.calendar.event.getRecords", {
  params: { status: "going", limit: 10 }, // typed params
});

if (response.ok) {
  console.log(response.data.records); // typed
}