@kronos-ts/postgres

The full Kronos persistence family on Postgres — event store (with the snapshotting and scheduling capability tiers that wrap it), token store, dead-letter queue, unit of work and handler wrapper — over raw SQL. No ORM required.

Install

bun add @kronos-ts/postgres
# Choose one driver:
bun add pg                 # for pgAdapter (Node + Bun) — also what a bare
                           # connection string loads
bun add postgres           # for postgresAdapter (porsager — Node + Bun)
# bunSqlAdapter has no external dep (uses Bun.sql built-in, requires Bun >= 1.2)

Requires Postgres 14+ (for xid8 and pg_snapshot_xmin).

Usage

One pool has a lifetime. Everything else is a plain function of it:

import { unitOfWork, localCommandBus, jsonSerializer } from "@kronos-ts/core"
import {
  postgresPool,
  postgresEventStore,
  postgresSnapshottingEventStore,
  postgresTokenStore,
  postgresDeadLetterQueue,
  postgresUnitOfWork,
} from "@kronos-ts/postgres"

const pg = postgresPool("postgresql://user:pass@host/db")
await pg.start()          // connects + bootstraps the schema

const eventStore  = postgresEventStore(pg)
const tokenStore  = postgresTokenStore(pg)
const deadLetters = postgresDeadLetterQueue(pg)
const uow         = postgresUnitOfWork(unitOfWork, pg)

// …and IF this deployment caches folds, one more line AROUND the log. The base
// store has never heard of snapshots; wrapping it adds the capability, and the
// wrapper serves a cached read in ONE round trip because it owns the query.
const cachingStore = postgresSnapshottingEventStore(eventStore, pg, jsonSerializer())

const commandBus = localCommandBus(uow)
// …
await pg.close()

Name only what this deployment uses — nothing else is constructed. postgresPool also takes an adapter you built yourself, which is how you pick a different driver or tune the pool:

postgresPool(pgAdapter({ connectionString, poolConfig: { max: 40 } }))
postgresPool(bunSqlAdapter({ connectionString }), { bootstrap: false })

The family shares one transaction

Every function above is built from the same pg, and that is what makes them one transaction. postgresUnitOfWork puts a (lazily opened) transaction on each unit of work; the token store, the dead-letter queue, the event store and a event handler's own ctx.sql() all read it. A crash cannot advance a processor's token while losing the work that token accounts for. A command handler's ctx.sql() is the exception: it runs on the pool (see below).

Never mix families within one processor: a token store or client built over a different pool or connection than the one the unit of work's transaction runs on is two clients, hence two transactions, hence no atomicity.

ctx.sql() in a handler

import { postgresHandler, type PostgresCapability } from "@kronos-ts/postgres"
import type { CommandHandlerContext } from "@kronos-ts/core"

const editWidget = commandHandler(EditWidget, async ({ payload }, ctx: CommandHandlerContext & PostgresCapability) => {
  await ctx.sql().query(
    "INSERT INTO widget_names (id, name) VALUES ($1, $2) ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name",
    [payload.id, payload.name],
  )                                    // commits now, on the pool
  ctx.append(WidgetUpdated, payload)   // commits at the end of the command
})

kronos({
  commandHandlers: [editWidget]
    .map((h) => ({ ...h, handler: postgresHandler(h.handler, pg) }))
    .map((h) => ({ ...h, eventStore, commandBus, queryBus })),
})

postgresHandler wraps the handler FUNCTION, not the entry — it is a plain (next, pg) => (message, ctx) => result with <M, C, R> inferred, so nothing about a handler entry appears in its type and the host keeps ownership of the spread. One wrapper covers command, event and query handlers alike: they differ only in the context they receive, and the capability is added the same way to each. A slice names what it uses by intersecting PostgresCapability onto the base context; there is no per-kind alias to import.

The erasure is DIRECTIONAL — a handler that ASKS for sql() goes in, one that asks only for the base context comes out — so wrapping twice, or wrapping a handler that never asked, is a compile error rather than a silent no-op.

In an event handler, ctx.sql() returns the unit of work's transaction when one is open and the pool otherwise, and postgresHandler opens the lazy transaction before the handler runs, so a batch's statements commit with the token. In a command handler, ctx.sql() is always the pool. A command's own writes commit as they run, before its events are appended; a rejected append does not undo them, and a retry runs them again, so write them idempotently — an upsert keyed by the aggregate's id. For the writer that must be inside a transaction whether or not anything else has touched it yet, use the accessor pair directly:

• postgresTransaction(uow) — opens the lazy transaction if it has not begun. Rejects if uow did not come from postgresUnitOfWork.
• activePostgresTransaction(uow) — observes, never opens; undefined when there is nothing to join.

Adapters

Three reference adapters ship with the package:

| Adapter | Runtime | Driver | LISTEN support | |---------|---------|--------|----------------| | pgAdapter | Node, Bun | pg (node-postgres) 8.20+ | native | | postgresAdapter | Node, Bun | postgres (porsager) 3.4+ | native | | bunSqlAdapter | Bun 1.2+ | built-in Bun.sql | native (1.2+) or polling shim |

You can also implement your own — PostgresAdapter from @kronos-ts/postgres is the contract.

pgAdapter

import { pgAdapter } from "@kronos-ts/postgres/adapters/pg"

const adapter = pgAdapter({
  connectionString: "postgresql://user:pass@host/db",
  poolConfig: { max: 10 }, // optional pg.Pool overrides
  // Per-transaction safety timeouts, armed via SET LOCAL on every transaction
  // this adapter opens. Defaults: 30s idle-in-transaction, statement off.
  idleInTransactionTimeoutMs: 30_000, // 0 disables
  statementTimeoutMs: 0, // opt in per deployment
})

Safety timeouts live on the adapter. Every transaction opened through the adapter (UoW-scoped commits, the event store's own-tx appends, the scheduler worker) is bounded, and each adapter instance is configured independently — so an event-store adapter and an event-processing adapter pointed at two different databases stay fully decoupled. idleInTransactionTimeoutMs is the important one: it stops a stalled transaction from pinning a connection (and pg_snapshot_xmin, which gates the tailing query) open indefinitely. The same two options are available on postgresAdapter and bunSqlAdapter.

postgresAdapter

import { postgresAdapter } from "@kronos-ts/postgres/adapters/postgres"

const adapter = postgresAdapter({
  connectionString: "postgresql://user:pass@host/db",
  clientOptions: { idle_timeout: 20 }, // optional porsager/postgres options
})

bunSqlAdapter

import { bunSqlAdapter } from "@kronos-ts/postgres/adapters/bun-sql"

// Bun runtime only — throws at connect() if run under Node
const adapter = bunSqlAdapter({
  connectionString: "postgresql://user:pass@host/db",
})

Transactions

Every handler runs inside a single Unit of Work, and with this extension that UoW carries a Postgres transaction. It is lazy: it opens on the first writer and never opens for pure-read work, so a read claims no connection it does not use.

In a processor, the batch is the transaction. postgresHandler opens it before the handler runs, and the handler's statements, the token update and any dead letters commit — or roll back — together.

In a command, the transaction is the append. A handler's appended events are buffered and flushed as one bulk write at commit, in a transaction that opens and commits within milliseconds. The handler's own writes are not part of it: ctx.sql() runs them on the pool, while the handler runs, so write them idempotently. Do not open the unit of work's transaction from a command handler with postgresTransaction(ctx.unitOfWork): it pins a connection for the rest of the handler, and a ctx.load after it waits for a second one from the same pool — a burst of commands the size of the pool deadlocks.

This is the framework's transaction, exposed for you to enroll work into. The package does not ship an ORM client — you bring (or wrap) your own. The two pieces you need are the accessor pair:

• postgresTransaction(uow) — returns the unit of work's transaction, opening the lazy one on first call. Use on write paths.
• activePostgresTransaction(uow) — returns it only if already open (else undefined), without forcing one. Use on read-only paths that shouldn't provoke a connection.

Running your own SQL — no ORM needed

The transaction handle runs parameterised SQL directly (it's what the event store appends with):

import { postgresTransaction } from "@kronos-ts/postgres"

// inside an event handler, in a processor:
const tx = await postgresTransaction(ctx.unitOfWork)
await tx.query("UPDATE widget_views SET name = $1 WHERE id = $2", [name, id])   // commits with the token

Handing the transaction to an ORM

tx.unwrap<T>() returns the live driver connection backing the transaction — the same connection the event store appends on. Bind your ORM to it and its writes join the transaction. The handle type depends on the adapter you wired:

| Adapter | unwrap() returns | |---------|--------------------| | pgAdapter | pg PoolClient | | postgresAdapter | scoped Sql (porsager) | | bunSqlAdapter | scoped Bun SQL |

Drizzle (the easy case)

drizzle(connection) accepts an already-open connection, so binding is a one-liner — pick the import matching your adapter:

// pgAdapter
import { drizzle } from "drizzle-orm/node-postgres"
import type { PoolClient } from "pg"
const db = drizzle(tx.unwrap<PoolClient>(), { schema })

// postgresAdapter
import { drizzle } from "drizzle-orm/postgres-js"
import type { Sql } from "postgres"
const db = drizzle(tx.unwrap<Sql>(), { schema })

// bunSqlAdapter
import { drizzle } from "drizzle-orm/bun-sql"
import type { SQL } from "bun"
const db = drizzle(tx.unwrap<SQL>(), { schema })

drizzle(conn) is a thin wrapper, so a small per-call helper is the ergonomic shape:

async function uowDb(ctx: { unitOfWork: UnitOfWork }) {
  const tx = await postgresTransaction(ctx.unitOfWork)
  return drizzle(tx.unwrap<PoolClient>(), { schema })
}

// in an event handler, in a processor:
const db = await uowDb(ctx)
await db.insert(widgetViews).values({ id, name })
  .onConflictDoUpdate({ target: widgetViews.id, set: { name } })

On ctx.db, as a handler wrapper

Kronos ships no builder integration, so the builder's types are always the version your app installed. To hand handlers a ready-built ctx.db, write the step next to your composition root — a wrapper that goes INSIDE postgresHandler:

import { describe } from "@kronos-ts/core"
import type { PostgresCapability } from "@kronos-ts/postgres"
import { drizzle, type PostgresJsDatabase } from "drizzle-orm/postgres-js"
import type { Sql } from "postgres"

export type DbCapability = { readonly db: PostgresJsDatabase }

export const drizzleHandler = <M, C extends DbCapability, R>(next: (message: M, context: C) => R) =>
  describe(
    (message: M, context: Omit<C, "db"> & PostgresCapability): R =>
      next(message, { ...context, db: drizzle(context.sql().unwrap<Sql>()) } as unknown as C),
    { name: "drizzleHandler", supplies: ["db"], uses: ["sql"], next } as const,
  )

// composition root
const wrap = (h) => postgresHandler(drizzleHandler(h), pg)

// in a slice
eventHandler(OrderCreated, async ({ payload }, ctx: EventHandlerContext & DbCapability) => {
  await ctx.db.insert(orderViews).values({ … })
})

describe is optional; it names the step in the boot walk, which then refuses the wrapper outside postgresHandler. With a relational { schema }, memoize the built client per ctx.sql() handle (a WeakMap) rather than rebuilding it per call. Do not call db.transaction() on it — it is already inside the task's transaction.

Pool-owning ORMs

ORMs that manage their own connection pool are more work — they don't take a borrowed connection directly. The route is a thin custom dialect/driver whose acquireConnection() hands back unwrap()'s connection and whose begin/commit/release are no-ops (the UoW owns the transaction lifecycle). Doable, just more plumbing than Drizzle. For occasional statements, the raw tx.query(...) path above is usually simpler than wiring a dialect.

The one rule

In a processor, bind your client to unwrap()'s connection — never to a separately-created pool. A separate pool is a separate connection, hence a separate transaction, and your projection writes silently stop being atomic with the token. unwrap() is the guarantee that you're on the framework's connection.

DCB Semantics

The engine implements Dynamic Consistency Boundaries:

• Atomic multi-tag conflict check on append with advisory-lock serialization
• Criteria-based sourcing with @> (contains-all) tag semantics
• Gap-free tailing using xid8 + pg_snapshot_xmin(pg_current_snapshot())

Concurrent writers on disjoint tags run in parallel (advisory-lock taxonomy permits this). Concurrent writers on the same tag serialise — exactly one commits; the other receives AppendConditionError.

A writer locks the tags its condition reads and the tags its events carry, with or without a condition. A position is assigned at insert and becomes visible at commit, so unserialised writers of one tag could commit out of position order and leave a read unable to see an event below its own marker. A read that matches nothing is anchored below its start rather than at the log head, for the same reason. When a condition carries reads, each read is checked against its own marker.

Gap-free streaming

The open() streaming method uses a two-phase cursor to prevent the concurrent-commit gap bug:

• Initial fetch: WHERE sequence_position > $cursor AND transaction_id < pg_snapshot_xmin(pg_current_snapshot())
• Subsequent fetches: WHERE (transaction_id, sequence_position) > ($xid, $pos) AND transaction_id < pg_snapshot_xmin(pg_current_snapshot())

The pg_snapshot_xmin watermark hides events from in-flight transactions, preventing the cursor from advancing past events that haven't committed yet. When a slow transaction finally commits, its events become visible in the correct xid8 order — no gaps, no re-ordering.

Schema

postgresPool(..., { bootstrap: true }) (the default) auto-creates these tables on start():

• kronos_events — append-only event log with xid8 watermark, BIGSERIAL position, JSONB payload, GIN index on tags
• kronos_snapshots — composite-PK (state_name, state_id) with BYTEA payload
• kronos_scheduled_events — durable schedules, tombstoned on fire/cancel
• kronos_token_entries — processor tokens, PK (processor_name, segment)
• kronos_dead_letters — parked letters, partitioned by processing_group

The last two are the SAME column sets the ORM families write, so a deployment can move between families without a migration.

Custom table names are a property of the pool, not of each store:

postgresPool(connectionString, {
  tableNames: { events: "my_events", snapshots: "my_snaps", scheduled: "my_sched",
                tokens: "my_tokens", deadLetters: "my_dead_letters" },
})

SQLSTATE

DCB violations raise SQLSTATE KR001. The adapter catches this and throws core's AppendConditionError, the conflict every store throws. To detect violations from your own code, ask with core's guard:

import { isConflict } from "@kronos-ts/core"

try {
  await eventStore.append(events, condition)
} catch (e) {
  if (isConflict(e)) {
    // Re-source the state to get a fresh marker, then retry.
  }
}

If you're working directly against the driver (bypassing the adapter), isDcbViolation(err) from @kronos-ts/postgres inspects a raw pg / postgres.js / Bun.sql error's .code field for the KR001 SQLSTATE.

FAQ

Q: Can I run my own migrations? Yes — set bootstrap: false and apply equivalent DDL via your own migration tooling. Every builder is exported: buildEventsTableDDL, buildEventsIndexesDDL, buildSnapshotsTableDDL, buildScheduledEventsTableDDL, buildScheduledEventsIndexesDDL, buildTokensTableDDL, buildDeadLettersTableDDL, buildDeadLettersIndexesDDL — plus bootstrapSchema if you want the whole thing in one call.

Q: PgBouncer compatibility? Yes — the adapter uses xact-scoped advisory locks (never session-scoped), so transaction-pooling mode is safe.

Q: Older Postgres support? No. xid8 requires PG14+. PG14 has been GA since 2021 and is universally supported across managed services.

Q: Is Bun.sql LISTEN available in all Bun versions? Bun.SQL was introduced in Bun 1.2. The bunSqlAdapter feature-detects native LISTEN; if unavailable (older Bun), it falls back to a 250ms polling shim. For production streaming workloads, use Bun 1.2+ or switch to pgAdapter / postgresAdapter which always have native LISTEN.

Q: TokenStore? Dead-letter queue? Both ship here — postgresTokenStore(pg) and postgresDeadLetterQueue(pg). Use them rather than an ORM family's: they write through the same client handle as your appends, so they join the same transaction. The dead-letter queue takes the processing group per CALL, not in its constructor — one queue object is one table, and which partition a call touches is the caller's business.