Postgres

The Postgres adapter is one small file; pg is the application’s dependency (a peer dependency of Atlas).

import pg from 'pg';
import { createPostgresStore, migrate } from '@ts-atlas/core/postgres';
import { createRuntime } from '@ts-atlas/core/runtime';
import { loadSystem } from '@ts-atlas/core/system';

const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
pool.on('error', (error) => console.error(error));   // errors of idle pooled clients are reported here

const system = await loadSystem('src/capabilities');
await migrate(pool);                                  // idempotent: CREATE TABLE IF NOT EXISTS ...
const runtime = createRuntime({
  capabilities: system.capabilities.map((c) => c.capability),
  store: createPostgresStore(pool),
});

createPostgresStore(pool) takes any pg-compatible pool. Run migrate(pool) on every boot; it never rewrites data. View tables are created by the runtime from the declarations.

Tables

Table	Holds
atlas_records	Every owned record: jsonb value, version, shape version, updated_at.
atlas_facts	The outbox: facts and sent commands, with v, cause, depth and the transaction id tx.
atlas_cursors	One cursor per consumer.
atlas_views	The spec and hash of every view.
atlas_schedule	Scheduled commands by due time, claimed with FOR UPDATE SKIP LOCKED.
atlas_leases	Consumer leases: one owner per consumer until expires_at (also presence rows for the fair share).
atlas_view_<name>	One real table per view: typed columns, text[] with a GIN index, jsonb for the rest, btree indexes from indexes.

dropViews(pool) drops every view table (they rebuild on the next boot): useful in tests and resets.

Delivery order

Appends never take a lock: any number of commits append to the outbox concurrently, so ids commit out of order. Each entry stores its transaction id (pg_current_xact_id()), and readers return only entries whose transaction is older than every transaction still in flight, ordered by (tx, id). That order only grows at its end, so a cursor never skips or repeats an entry.

The consequence: a long transaction anywhere in the cluster delays delivery (it never loses an entry), and drain() waits up to 5 s for entries committed before it to become readable.

Refusals and outages

commit has three outcomes: { ok: true }, { ok: false, conflict: true }, and { ok: false, poison: { code, message } } for a refusal the database would repeat for the same bytes. The adapter returns poison only for SQLSTATE class 22 (data exception) and the integrity codes 23503, 23514 and 23P01. Anything else, transient or fixable by a migration or a GRANT, throws and is retried.

A connection lost during an operation (Postgres restarted, a terminated backend) fails that operation (store-failed, or a worker retry with backoff) and destroys the client; the process lives (since 0.5.0). Attach pool.on('error', ...) for idle clients.

Operations

atlas rebuild <view> --db <postgres-url>                       # replay a view from the outbox (shadow + swap)
atlas cursor list --db <postgres-url>                          # every consumer cursor and its capability
atlas cursor copy <from> <to> --db <postgres-url> [--force]    # start <to> where <from> is
atlas compact --db <postgres-url> --older-than 30d             # delete old tombstones (at least 1h)

Inspect the leases with SELECT consumer, owner, expires_at FROM atlas_leases ORDER BY owner, consumer;.

Another database

Implement the Store interface and run the shared contract suite in test/store-contract.ts: storeContract('mydb', async () => ({ store })). See the adapters reference.