@structure-ai/eventsourcing-pg

PostgreSQL adapters (@effect/sql-pg) for the @structure-ai/eventsourcing ports, including EventStore, HistoryImporter, StreamEraser, snapshots, checkpoints, outbox, and inbox, plus a durable IdempotencyStore for @structure-ai/cqrs and a PostgreSQL LISTEN/NOTIFY event bus for projections in separate processes.

The package exports TypeScript source. Its runtime dependencies include the PostgreSQL driver declarations so consumers can typecheck the adapter without installing them separately.

Usage

import { layer } from "@structure-ai/eventsourcing-pg";
import { CommandBus } from "@structure-ai/cqrs";
import { Layer } from "effect";

// PgClient from `url` (or DATABASE_URL), migration, adapters, and a listening EventBus.
const durable = layer({ url: databaseUrl, idempotencyTtl: "24 hours" });

// The command bus takes its IdempotencyStore from the same layer.
const bus = CommandBus.layer.pipe(Layer.provide(Layer.mergeAll(handlers, authorizer, durable)));

On an existing SqlClient (shared with view models and migrations): storesLayer(options) merges every adapter; run migrate(options) yourself first, once, from the designated migration process. For notification-driven workers, also merge eventBusLayer(options) on the same ambient PgClient, after migration. storesLayer alone keeps its existing storage-only contract. Individual adapters — eventStoreLayer, snapshotStoreLayer, checkpointStoreLayer, outboxLayer, inboxLayer, idempotencyStoreLayer — compose the same way.

Exports

| Export | What it is | | --- | --- | | layer(config?) | PgClient + migrate + every adapter + EventBus, and the client itself. Acquisition can fail with SqlError or PersistenceError. config: url, maxConnections, applicationName, plus the adapter options. | | eventBusLayer(options?) | Scoped cross-process EventBus on an ambient PgClient. Requires schema revision 5; opens one dedicated listening connection outside the query pool. Acquisition and delivery failures are PersistenceError. | | storesLayer(options?) | Every adapter on top of an ambient SqlClient (no migration). | | migrate(options?) | Applies every step of migrations in order, idempotently: the tables for events, history-import bookkeeping, snapshots, checkpoints, outbox, inbox, and idempotency (rev 1), then the generated partition column and its index on events (rev 2), then outbox.available_at with its partial index for scheduled delivery (rev 3), then outbox.claim_token/lease_until with the claim index for delivery ownership (rev 4), then the event-insert notification trigger and its function (rev 5), all isolated by tablePrefix. | | migrations | The same schema as ordered { rev, name, apply(options?) } steps, for consumers keeping their own migration ledger: record each rev as its own entry and append the next one on upgrade. | | tableNames(options?) | Resolved table names for a prefix — use it for test isolation and cleanup. | | appendWithOutbox(stream, expectedVersion, events, messages) | Events and outbox rows committed in one transaction. | | withUnitOfWork(effect) | Application unit of work: every write effect performs through the ambient SqlClient — appends, appendWithOutbox, outbox/inbox/idempotency/snapshot writes, and application-owned SQL — commits as one transaction; any failure rolls it all back. Nested calls become SAVEPOINTs, so an inner unit can fail while the outer one continues. | | HistoryImporter from eventStoreLayer/storesLayer | Preserves frozen source positions and versions in atomic, resumable batches. Import bookkeeping makes identical retries no-ops and rejects divergence. | | StreamEraser from eventStoreLayer/storesLayer | Destructive per-stream retention (eraseStream) backed by the erased_streams ledger: payload-bearing rows become Erased tombstones (positions kept, checkpoints valid), the stream's snapshot is deleted in the same transaction, and the stream is pinned against future appends. Append and erasure are serialized per stream with a transaction-scoped advisory lock, so the mutual exclusion holds across pool connections and processes. See the @structure-ai/eventsourcing README for the full semantics. | | idempotencyStoreLayer(options?) | @structure-ai/cqrs IdempotencyStore over the idempotency table. | | purgeExpiredIdempotency(options?) | Deletes idempotency records past their TTL; returns the count. | | AdapterOptions | tablePrefix (default none) and idempotencyTtl (default 24 hours). |

Projections in a separate process

layer() supplies the event bus automatically. Run your projections in a worker entrypoint using the same database and tablePrefix as the API:

// src/projections.ts. Start separately with: bun src/projections.ts
import { Projection } from "@structure-ai/eventsourcing";
import { layer } from "@structure-ai/eventsourcing-pg";
import { launch } from "@structure-ai/runtime";
import { ordersProjection } from "./orders-projection.js";

launch(Projection.run(ordersProjection), {
  layers: layer({ url: databaseUrl, tablePrefix: "app_" }),
});

Application code owns the worker process; PostgreSQL carries notifications between it and the API, CLI commands, webhooks, or other writers. For a view model, use its ViewProjection.run() with the worker's view-model store layer too. Keep one active worker per projection name and read model. Stop that worker before inline catch-up or rebuilding the same projection.

Revision 5 installs an AFTER INSERT statement trigger on the events table. It calls pg_notify only when rows were inserted; PostgreSQL delivers after the outermost transaction commits. Rollbacks and rolled-back savepoints send nothing. Appends, appendWithOutbox, new history-import batches and direct SQL inserts signal automatically, including when a workflow commits an event and later fails. Command wrappers with EventBus.notifyAfter are unnecessary. Direct SQL writers must still honor the commit-order contract.

Layer acquisition completes LISTEN before returning. Workers then subscribe locally before their initial catch-up. Signals contain only a constant wake-up marker, coalesce per transaction and subscriber, and never carry event payloads. Workers read durable history from their checkpoints. Channels use the events table's PostgreSQL relation OID, so prefixed tables and tables in different schemas have independent channels.

Omit pollInterval to perform no continuous event-store polling; an explicitly configured interval still causes reconciliation reads. A listener disconnect fails pending and future waits with PersistenceError (EventBus.listen), so the worker stops. Restart the whole worker and its layer to reconnect and catch up from the checkpoint, including writes committed while it was offline. The adapter does not reconnect silently. Notification connection teardown is scoped and bounded. Use a direct PostgreSQL connection or session pooling for this dedicated listener; transaction pooling cannot retain LISTEN subscriptions.

Apply revision 5 before starting a worker that uses eventBusLayer. The migration is idempotent and preserves existing history. layer() runs it automatically; applications with their own migration ledger append revision 5 there. This revision requires PostgreSQL 14 or newer (CREATE OR REPLACE TRIGGER). When dropping a test table set, also drop its trigger function; the trigger disappears with the table, but the function remains.

Partition column

Rev 2 adds partition TEXT GENERATED ALWAYS AS (metadata->>'partition') STORED to events with an index on (partition, position). The envelope stays the single source of truth (appends never write the column); readAll({ partition }) filters on it before ordering by position. Upgrading a store that already holds events rewrites the table once, under an exclusive lock for the duration of the ALTER TABLE — plan it in a maintenance window on a large store, either by running migrations[1].apply() ahead of the deploy or by accepting the pause at the first migrate() of the new version. Rows written before the upgrade get NULL unless their envelope already carried a partition.

Outbox delivery ownership

Rev 4 adds outbox.claim_token and outbox.lease_until, backing Outbox.claim(limit, lease): one UPDATE … WHERE … FOR UPDATE SKIP LOCKED statement atomically leases due entries to a relay, so concurrent relays get disjoint batches and never publish the same entry within a lease. Settlements (markPublished/markFailed/markDead) carry the claim's token and are fenced: a settlement arriving after its lease lapsed and was recovered by another relay matches no row and writes nothing. ALTER TABLE … ADD COLUMN IF NOT EXISTS only — no table rewrite, no exclusive lock; existing rows keep NULL (unclaimed), matching pre-rev-4 behavior.

Unit of work

withUnitOfWork(effect) binds all writes of one command handler to a single PostgreSQL transaction: aggregate appends, outbox enqueue, inbox dedupe, idempotency claims, snapshots, and any application SQL issued through the same SqlClient commit together or not at all. Concurrency conflicts stay typed (ConcurrencyConflict), and because a rolled-back unit never committed, a retry after a conflict starts from a clean slate. Staged outbox rows become visible to the relay only after the unit commits, so a rolled-back unit never publishes.

Nesting is re-entrant: a withUnitOfWork (or any sql.withTransaction, including the adapters' own appends) started inside an open unit becomes a SAVEPOINT rather than a second transaction. An inner unit that fails rolls back only its own writes; the outer unit may catch the failure and keep going, while an uncaught inner failure rolls back the whole outer unit. Keep units short — every lock the unit took (row locks, and any advisory locks appends acquire) is held until the outermost unit commits.

Idempotency store

One row per (tag, actor, key) claim; the anonymous scope is stored as the empty actor. begin is a single conditional upsert (insert a fresh claim, or replace an expired one) followed, when a live record exists, by a read that reports Completed (with the JSON result), InFlight or Mismatch — so concurrent begin calls for one context yield exactly one Claimed, across instances. complete stores the encoded success and refreshes expires_at; release deletes a claim that never completed.

Records expire idempotencyTtl after their last claim or completion: begin treats an expired record as absent — it reclaims the row in place (whatever its old payload hash or status), and the reclaimed row is live again — and purgeExpiredIdempotency deletes the rows still expired when it runs, returning exactly that count. Size the TTL to the callers' retry window. A claim left by a crashed process blocks its key (callers see IdempotencyInFlight, 409) until it expires — bound handlers with a dispatch timeout so claims are released on the normal path. Run the purge periodically from any instance; it is safe to run concurrently.

Commit order

position is drawn from a sequence at insert time, inside the append transaction, so two concurrent appends could commit in the opposite order of their positions; a projection polling in between would checkpoint past a position whose transaction had not committed yet and never deliver that event (it is committed, visible to read, and replayed by a rebuild — the live view and the rebuilt view diverge silently). EventStore.append and appendWithOutbox therefore take a transaction-scoped advisory lock (pg_advisory_xact_lock(0x5f455654, hashtext(<events table>))) after the version check and before the first insert; Postgres releases it as part of the commit, after the transaction is visible to new snapshots, so positions become visible in the order they were drawn and a reader that has seen position N has seen every committed event below it. Only the inserts and the commit are serialized, never the version check or the caller's own work, and an empty (version-checked) append takes no lock. One consequence to plan around: an append nested inside a caller's own withTransaction holds the lock until the caller's commit, so a caller that keeps its transaction open after appending blocks every other append to that events table for as long as it stays open — append last, and keep such transactions short. The lock has no timeout of its own; a stalled writer is a stalled store until its transaction ends.

Measured on a local Postgres 17 (two-event appends to distinct streams): a single writer is unchanged (about 2 ms per append); the serialized section costs about 1.6 ms per append and is a hard throughput ceiling of roughly 600 appends per second whatever the number of writers, so sixteen concurrent writers go from about 2,600 to about 600 appends per second with p95 latency rising from about 10 ms to about 33 ms, and thirty-two writers see a p95 of about 69 ms, pure queueing. Keep that advisory key for this package; HistoryImporter locks the events table exclusively and composes with it.

Tests

bun test runs the suite against DATABASE_URL and skips it otherwise. Each test creates a uniquely prefixed table set and drops it afterwards, so one database serves parallel runs.

Persistence failures

The storage, history-import and idempotency adapters expose PersistenceError for SQL and JSON failures, retaining the operation and diagnostic cause. Version races remain ConcurrencyConflict. See the shared error contract.

SQL-specific helpers such as migrate and appendWithOutbox retain their explicit SqlError channel. JSON encoding in appendWithOutbox can also fail with PersistenceError.