PostgreSQL

@ocoda/event-sourcing-postgres provides the PostgresEventStore and the PostgresSnapshotStore. It needs the pg and pg-cursor drivers, which are peer dependencies: see Installation and Database drivers are peer dependencies.

Configuration

The config of each store is a pg pool config plus the store’s own options: driver, useDefaultPool and ddl. The store removes its own options before it creates the pool.

EventSourcingModule.forRoot<PostgresEventStoreConfig, PostgresSnapshotStoreConfig>({
  eventStore: {
    driver: PostgresEventStore,
    connectionString: process.env.DATABASE_URL,
    max: 20, // connections in the pool
    ddl: 'auto', // default: create missing tables; 'none' only checks them
  },
  snapshotStore: {
    driver: PostgresSnapshotStore,
    connectionString: process.env.DATABASE_URL,
  },
});

The tables live in the current schema of the connection, the first schema of its search_path (usually public). To use another schema, set it for the connections: options: '-c search_path=events_app'. Behind a connection pooler in transaction mode, set it for the role instead: see Managed PostgreSQL.

connect() fails when the database can’t be reached. disconnect() ends the pool; calling it twice, or before connect(), does nothing.

Schema

Every pool has its own table, in the current schema: events and snapshots for the default pool, <pool>-events and <pool>-snapshots for the others. A table name is at most 63 bytes, the PostgreSQL limit: ensureCollection rejects a longer pool name with an EventStoreCollectionCreationException instead of letting PostgreSQL truncate it.

The catalog. One table per schema, event_sourcing_collections, registers every table the stores use, with its kind and schema version. For an event table it also holds the counter of the pool’s global positions. listCollections() reads it, so a 3.x table that isn’t migrated is not listed.

CREATE TABLE event_sourcing_collections (
  name TEXT PRIMARY KEY,
  kind TEXT NOT NULL CHECK (kind IN ('events', 'snapshots')),
  schema_version INTEGER NOT NULL,
  last_position BIGINT NOT NULL DEFAULT 0 CHECK (last_position >= 0)
) WITH (fillfactor = 50);

Events. The columns that 3.x didn’t have come last, so that a created table and a migrated one have the same columns.

CREATE TABLE "events" (
  stream_id TEXT NOT NULL,
  version INTEGER NOT NULL,
  event TEXT NOT NULL,
  payload JSONB NOT NULL,
  event_id TEXT NOT NULL,
  aggregate_id TEXT NOT NULL,
  occurred_on TIMESTAMPTZ NOT NULL,
  correlation_id TEXT,
  causation_id TEXT,
  global_position BIGINT NOT NULL,
  headers JSONB,
  event_version INTEGER,
  PRIMARY KEY (stream_id, version)
);
CREATE UNIQUE INDEX "idx_events_global_position" ON "events" (global_position);

Snapshots. The last snapshot of a stream is the one with the highest version. It also carries the stream’s latest flag, which a unique index keeps to one row per stream, and which SnapshotRepository.loadAll() pages through in descending binary order of the aggregate ids.

CREATE TABLE "snapshots" (
  stream_id TEXT NOT NULL,
  version INTEGER NOT NULL,
  payload JSONB NOT NULL,
  snapshot_id TEXT NOT NULL,
  aggregate_id TEXT NOT NULL,
  registered_on TIMESTAMPTZ NOT NULL,
  aggregate_name TEXT NOT NULL,
  latest TEXT COLLATE "C",
  PRIMARY KEY (stream_id, version)
);
CREATE UNIQUE INDEX "idx_snapshots_latest" ON "snapshots" (aggregate_name, latest) WHERE latest IS NOT NULL;

ensureCollection(pool) creates the tables of a pool and registers them, and on an existing table only registers it. With ddl: 'none' it never creates anything: a missing table or catalog throws, with the statements that create it in the message. Index names that don’t fit in 63 bytes are shortened with a hash.

Capabilities

PostgresEventStore claims every capability:

Capability	Value	Why
`atomicAppend`	`true`	An append is one transaction.
`headers`	`true`	Stored in the `headers` column.
`globalOrder`	`'gap-safe'`	See below.

An append first moves the pool’s counter row in the catalog, then inserts its events, in one READ COMMITTED transaction. The row lock on the counter is held until the transaction ends, so an append that takes later positions waits until the earlier one committed or rolled back. Appends therefore commit in the order of their positions, and a rolled-back append returns its positions, so there are no holes. A reader that tails readAll() and resumes after the last position it read never skips an event that commits later. readAll() reads in keyset batches (WHERE global_position >= $1 ORDER BY global_position LIMIT $2), each its own statement, and holds no connection while you process a batch. The evidence is in ADR 0002. For managed services, see Managed PostgreSQL.

Limits and data types

Stream ids, aggregate ids, event names, correlation ids and causation ids are TEXT; the core limits them to 255 characters, and headers to 8 KiB (see Limits).
Payloads are JSONB. A Date in a payload is stored as an ISO string and read back as a string. A bigint in a payload can’t be stored: the append fails with an EventStorePersistenceException (outcome: 'not-persisted'), so store large integers as strings.
Global positions are BIGINT. The store reads them as text and returns bigints, without a global pg type parser.
An append sends its events as arrays, so its size isn’t limited by the 65,535 parameters of a statement (3.x stopped at about 6,500 events).

Throughput

Appends to one pool serialize on the pool’s counter row, from the counter update to the commit. That caps the appends per second of a pool, whatever the number of connections; appends to different pools don’t wait for each other. Pools are the unit of write scale.

Appends per second, with 1 to 3 events per append, each writer on streams of its own:

Setup	1 pool, 1 writer	1 pool, 8 writers	4 pools, 8 writers each (per pool)
PostgreSQL 14 and 17, data on tmpfs, through `appendEvents`	317–331	1,053–1,057	502–534
PostgreSQL 14, data on disk with fsync, the append transaction alone	–	390	–

These are laptop numbers (Apple silicon, PostgreSQL in a 6-CPU VM, all writers in one Node.js process); measure your own setup. The disk figure is dominated by the commit’s WAL flush.

An append holds the counter row from its first statement to its commit, so a writer that stalls in between (a frozen process, a lost network, a long pause of the event loop) holds up every append to the pool, from every instance, until its session ends. Without a timeout, a lost client can take until TCP keepalive notices, about 2 hours by default. Let the server end such sessions, and let pg notice a dead connection:

ALTER ROLE app SET idle_in_transaction_session_timeout = '10s';

eventStore: { driver: PostgresEventStore, connectionString: process.env.DATABASE_URL, keepAlive: true },

ExpectedVersion.Any makes up to 16 attempts while concurrent appends take the versions of a stream. Under sustained contention on one stream, an Any append can still end in an EventStoreVersionConflictException (12 of about 8,000 appends to one hot stream in a soak test), so retry such a conflict in your application.

Privileges

Task	Needs
Appending and reading (`ddl: 'none'`)	`USAGE` on the schema; `SELECT` and `INSERT` on the event tables; `SELECT`, `INSERT` and `UPDATE` on the snapshot tables; `SELECT`, `INSERT` and `UPDATE` on `event_sourcing_collections`
Creating missing tables (`ddl: 'auto'`, the default)	The above, plus `CREATE` on the schema. On a schema whose tables exist, `'auto'` creates nothing and needs no more than `'none'`
`migrate()`	Ownership of the tables, `TEMPORARY` on the database, `SELECT`, `INSERT` and `UPDATE` on `event_sourcing_collections`, and `CREATE` on the schema if the catalog doesn’t exist yet. The dry run reports a missing privilege as `blocked`

The stores also take advisory locks, which need no privilege. Advisory locks are per database; these are the keys, so you can recognize them in pg_locks and keep your own apart:

Lock	Key	Held
Creating the catalog	`hashtext('ocoda:event_sourcing_collections')`	for its transaction
Creating or registering a table	`hashtext('<table>')`, for example `hashtext('events')`	for its transaction
Appending a snapshot	`hashtext('<table>'), hashtext('<stream id>')`	for its transaction
Migrating a table	`hashtext('ocoda:migrate'), hashtext('<schema>.<table>')`, with the names as `format('%I.%I', …)` writes them	for the migration of the table

migrate() works on the current schema: run it once per schema, with that schema first in the search_path of the connections. Migrations of different schemas can run at the same time.

With ddl: 'none', create the schema yourself: from the statements in the error of ensureCollection, from the steps of a dry-run migrate(), or, for the default pools of a 3.x database, from node_modules/@ocoda/event-sourcing-postgres/migrations/4.0.sql. Then call ensureCollection, which registers the tables in the catalog.

Managed PostgreSQL

The stores use plain PostgreSQL: READ COMMITTED transactions, the counter row, transaction-scoped advisory locks, keyset reads, and a server-side cursor for stream reads. CI tests community PostgreSQL 14 to 18; no managed service is tested.

Service	Status
Amazon RDS for PostgreSQL, Google Cloud SQL for PostgreSQL, Azure Database for PostgreSQL, Supabase, Neon	Expected to work, and to be gap-safe on the primary: they run PostgreSQL (Neon on an open-source storage layer of its own).
Amazon Aurora PostgreSQL, Google AlloyDB for PostgreSQL	Expected to work, and to be gap-safe on the writer, but less certain: they are closed-source forks with storage of their own, and the evidence is for community PostgreSQL.
Amazon Aurora DSQL	Not compatible, see below.
Serverless drivers over HTTP, such as Neon’s	Not supported, see below.

Connect to the primary. Point both stores at the primary, or at the writer endpoint of a cluster: a read replica refuses appends. Don’t send the stores’ reads to replicas either, through a reader endpoint or a proxy that balances reads over replicas (Pgpool-II’s load balancing, for instance): the version check before an append would read stale versions, and the gap-safe claim covers reads from the server the appends commit on. A failover to a replica that lacks the newest commits, or a restore, can hand positions out again: see Failover and replicas.

Connection poolers. A pooler in session mode, which keeps a client on one server connection for as long as it is connected, changes nothing. In transaction mode (PgBouncer with pool_mode = transaction, or the transaction-mode endpoints of services such as Supabase and Neon), consecutive transactions of a client can run on different server connections, so what a session sets doesn’t follow the client. Transaction mode isn’t tested. What depends on the session:

The schema. The stores find their tables in the current schema. A search_path in the startup options (see Configuration) reaches the server only when the pooler passes it on. Recent PgBouncer releases pass on the parameters they track: search_path on PostgreSQL 18 and later, or when it is listed in track_extra_parameters. Otherwise, and in older releases, PgBouncer refuses the connection, or, with options in ignore_startup_parameters, drops the setting, and the stores create and read their tables in the role’s default schema. Set the schema for the role instead, which every new server connection of the role picks up:
```
ALTER ROLE app SET search_path = events_app;
```
migrate() keeps state on its connection from one transaction to the next: it sets statement_timeout and work_mem for the session (also for a dry run) and resets them at the end, and it holds a session advisory lock, which keeps a second migration of a table out, from before the table’s transaction until after it. Through a transaction-mode pooler, those settings would reach the sessions of other clients, and the lock could stay on a server connection the migration no longer uses. Run migrate() over a direct connection or a session-mode endpoint.
Stream reads (getEvents, a stream’s snapshots, SnapshotRepository.loadAll()) read a server-side cursor in batches, on one connection, outside a transaction. PgBouncer should keep the client on its server connection until the cursor is done (not tested); other poolers aren’t verified.

readAll runs one statement per batch, the stores’ own advisory locks are transaction-scoped, and the stores use no named prepared statements, so none of these depends on the session.

Serverless drivers. The stores create their own pg pool from the config, and speak PostgreSQL’s protocol over TCP. They can’t use a serverless driver, such as Neon’s @neondatabase/serverless or @vercel/postgres, whose queries over HTTP have no interactive transactions and no cursors. On Neon, use the regular connection string, from a runtime that can open TCP connections.

Amazon Aurora DSQL is not compatible. It takes no locks: it detects conflicts when a transaction commits, and fails the transaction that commits last with a serialization error (40001) instead of making it wait (concurrency control in Aurora DSQL). Every append to a pool updates the pool’s counter row, so of two appends to a pool that overlap, one fails. ensureCollection fails too: it creates a table and its index and registers them in one transaction, and Aurora DSQL allows one DDL statement per transaction, and no other writes with it.

For CockroachDB, YugabyteDB and other servers that speak PostgreSQL’s protocol, see Other databases.

Migrating from 3.x

The 3.x tables have an event_date column and no global positions. ensureCollection refuses a 3.x event table with an EventStoreSchemaException (found: 'v1'), so an application whose default pool is a 3.x table fails to bootstrap until it is migrated. It never migrates a table by itself. A 3.x snapshot table keeps working, with a warning, until it is migrated.

The migration is offline: stop every 3.x instance first.

Take a backup.
Run a dry run. It needs no Nest application and writes nothing:
```
import { PostgresEventStore, PostgresSnapshotStore } from '@ocoda/event-sourcing-postgres';

const config = { connectionString: process.env.DATABASE_URL };
const events = await PostgresEventStore.migrate(config, { dryRun: true });
const snapshots = await PostgresSnapshotStore.migrate(config, {
  dryRun: true,
  legacyTimeZone: 'Europe/Brussels', // the time zone your 3.x instances ran in
});
console.log(JSON.stringify([events, snapshots], null, 2));
if ([...events.collections, ...snapshots.collections].some(({ action }) => action === 'blocked')) {
  throw new Error('Resolve the blocking issues before the cutover');
}
```
It finds the tables by their name and columns (pools limits it to some pools) and reports, per table:
- action: migrate, skip (already migrated) or blocked, with the reasons in blocking;
- rows, bytes, and in warnings the free disk the migration needs and the objects it drops with event_date (constraints, statistics);
- gappedStreams: streams whose versions don’t run from 1 without gaps;
- duplicateEventIds and nonCrockfordEventIds, which are migrated as they are;
- dependents (views, rules, triggers, publications, foreign keys) and droppedIndexes (every index on event_date, also through an expression or a predicate);
- steps: the exact statements the migration runs, with the lock each one takes;
- for snapshots, snapshotFlags: streams with several snapshots flagged as the latest, or none;
- environment.timeZones: the time zones of the process, the server and the session.
Stop every 3.x instance.
Migrate the events, then the snapshots, with the same options without dryRun. onProgress reports each step. A blocked table is left as it is and nothing is thrown: check the reports as in the dry run, and stop the cutover while any table is blocked. A second run reports every table as skip.
Deploy 4.0.

Prefer the static PostgresEventStore.migrate(config, options) to migrate() on a connected store when the store’s config sets query_timeout: pg enforces it on every query of the pool, and the static migrate() leaves it out. The migration clears the statement_timeout of its own connection (a role or config setting would stop the inspection or the VACUUM of a large table) and restores it afterwards.

What the migration does to an event table. One transaction per table, under ACCESS EXCLUSIVE:

Copy the rows into a temporary table, numbered in 3.x’s order: (event_date, event_id, stream_id, version), adjusted so that every stream stays in version order (two events of a stream appended in the same millisecond could be listed in the wrong order by 3.x).
Add global_position, headers and event_version, change the VARCHAR columns to TEXT, empty the table, drop event_date (and with it every index, constraint and statistics object on it), and insert the numbered rows back in position order. Columns you added to a 3.x table keep their values; generated columns are computed again.
Create the unique index on global_position, register the table with its counter at the highest position, and commit. Then run VACUUM (ANALYZE); if that fails, the migration has committed anyway and the report has a warning with the statement to run by hand.

The table keeps its OID, so its grants, its publication memberships, and the views that don’t use event_date or the changed columns survive. A migration that stops at any point, for instance because its connection is lost, rolls back and leaves the 3.x table as it was: run it again.

Snapshot tables are migrated in place, in one transaction: the flags are normalized to exactly one per stream on its highest version, registered_on becomes a TIMESTAMPTZ, and the unique index on the flags replaces the 3.x index. 3.x stored registered_on as the wall time of the process, without a time zone: pass that time zone as legacyTimeZone (an IANA name the server knows; default: the time zone of the process that migrates). With UTC, the column is converted without rewriting the table.

What blocks a table. The migration leaves a table untouched, and reports it as blocked, when:

a view, rule, policy, trigger (its WHEN or UPDATE OF columns), publication (its row filter or column list) or generated column uses event_date or a column whose type changes (drop it, migrate, create it again);
a foreign key of another table references it (the migration empties and refills the table);
the current role doesn’t own it, may not create temporary tables, or may not read and write the catalog;
its name was truncated: 3.x created the table of a pool whose table name is longer than 63 bytes under the first 63 bytes of that name, and 4.0 refuses such pools. Rename the table for a pool name of at most 56 bytes (<pool>-events; 53 bytes for <pool>-snapshots) and use that pool. A dry run without pools lists these tables too;
another session still uses it after lockTimeoutMs (default 10 seconds): a 3.x instance may still be running;
another migrate() is migrating it.

Plan for:

Downtime. The table is locked from the start of its transaction to the commit. Plan for about 0.5 minutes per million events under the lock, on disk, and about 5 more seconds per million for the inspection before it and the VACUUM after it. These figures are provisional: they are extrapolated from 100,000 events (a 70 MB table: 1.1 to 1.4 seconds under the lock, 1.6 to 1.8 seconds in all, on tmpfs) and from the 1-million-row spike, not measured at a million rows with the statements that ship. Snapshots take about 15 seconds per million on tmpfs.
Disk. Keep 3 times the event table’s size free while it runs: until the commit, the old files of the table stay next to the numbered copy (in temp_tablespaces), the sort files of the numbering, and the new table and indexes (measured at 1.9 to 2.2 times). The migration writes about as much WAL as the table’s size. Converting the snapshots’ registered_on from a time zone other than UTC rewrites the snapshot table: keep its size free as well.
Replication and CDC. Publications and logical decoding receive a TRUNCATE of the table followed by an insert of every row. INSERT and TRUNCATE triggers fire; disable them for the migration if they must not. DDL isn’t replicated: migrate the tables of a logical replication subscriber first (the numbering is deterministic, so both sides get the same positions), or disable the subscription, migrate both sides and enable it again. Otherwise the subscriber can’t apply the reinserted rows, and the publisher keeps its WAL.
Other sessions. TRUNCATE isn’t MVCC-safe: a REPEATABLE READ or SERIALIZABLE transaction that started before the migration committed, and reads the table for the first time after it, sees it empty. Stop such readers too.
Time zones of the snapshots. A wall time in the hour that repeats when the clocks go back is ambiguous, and may move by an hour. A 4.x process stores the snapshots it appends to a 3.x table in its own time zone: run 4.x in the time zone of 3.x until the snapshots are migrated, or migrate them before you deploy 4.x in another time zone.
Gapped streams conflict on their next append, because 4.0 checks that the stream is exactly at the expected version: load the aggregate from its events and append with its actual version.
3.x writers that are still running fail after the migration with column "event_date" … does not exist (42703): nothing is written without a position.

Without migrate(). A DBA can run the statements of the dry run’s steps instead, in order, in one session, with the steps from begin to commit in one transaction: never in autocommit, and never from a tool that opens a connection per statement (the truncate step refuses to empty the table when the numbered copy of its transaction is missing). For the default pools of a 3.0.1 or 3.0.2 schema, migrations/4.0.sql of the package holds the statements, for psql 10 or later: psql -v legacy_time_zone='Europe/Brussels' -f migrations/4.0.sql. It stops before any change when legacy_time_zone isn’t set, and migrates each table only while it still has its 3.x shape, so it can run again after a failure.