PostgreSQL table structure for the event store — events, event-backed snapshots, outbox, crypto keys, and projections.

Database Schema

All event store tables live in a dedicated PostgreSQL schema (default: event_store), isolated from application tables. Migrations are idempotent — safe to call on every application startup.

Schema Isolation

The event store uses its own PostgreSQL schema. This keeps event store tables completely separate from application tables (e.g. those managed by Prisma or other ORMs).

CREATE SCHEMA IF NOT EXISTS event_store;

The schema name defaults to event_store but can be configured:

const eventStore = new EventStore({
  pool,
  schema: "my_custom_schema", // Must match /^[a-z_][a-z0-9_]{0,62}$/
});

Tables

`events`

The core append-only event log. Columns are aligned with the CloudEvents v1.0.2 specification. Event-backed snapshots are stored here as generated events with event types ending in Snapshot.

CREATE TABLE IF NOT EXISTS {schema}.events (
  global_position  BIGSERIAL        NOT NULL,
  stream_id        TEXT             NOT NULL,
  stream_version   INTEGER          NOT NULL,
  id               TEXT             NOT NULL,
  source           TEXT             NOT NULL,
  specversion      TEXT             NOT NULL,
  event_type       TEXT             NOT NULL,
  subject          TEXT             NOT NULL,
  time             TIMESTAMPTZ      NOT NULL,
  datacontenttype  TEXT             NOT NULL,
  data             JSONB            NOT NULL,
  extensions       JSONB            NOT NULL,
  encrypted_data   JSONB            NULL,
  crypto_key_id    TEXT             NULL,
  schema_version   INTEGER          NOT NULL,
  created_at       TIMESTAMPTZ      NOT NULL DEFAULT now(),
  txid             XID8             NOT NULL DEFAULT pg_current_xact_id(),

  PRIMARY KEY (global_position),
  UNIQUE (stream_id, stream_version)
);

Column	Type	Description
`global_position`	`BIGSERIAL PK`	Monotonically increasing global ordering
`stream_id`	`TEXT NOT NULL`	Stream identifier (e.g. `"Order-123"`)
`stream_version`	`INTEGER NOT NULL`	Sequential version within the stream (1, 2, 3, ...)
`id`	`TEXT NOT NULL`	CloudEvents `id` — `"{streamId}/{streamVersion}"`
`source`	`TEXT NOT NULL`	CloudEvents `source` — URI-reference
`specversion`	`TEXT NOT NULL`	Always `"1.0"`
`event_type`	`TEXT NOT NULL`	CloudEvents `type` — event type name
`subject`	`TEXT NOT NULL`	CloudEvents `subject` — the stream ID
`time`	`TIMESTAMPTZ NOT NULL`	CloudEvents `time` — ISO 8601 timestamp
`datacontenttype`	`TEXT NOT NULL`	Always `"application/json"`
`data`	`JSONB NOT NULL`	Event payload (PII fields removed if encryption is used)
`extensions`	`JSONB NOT NULL`	CloudEvents extension attributes
`encrypted_data`	`JSONB NULL`	Encrypted PII field blobs
`crypto_key_id`	`TEXT NULL`	Reference to `crypto_keys.key_id`
`schema_version`	`INTEGER NOT NULL`	Schema version for upcasting
`created_at`	`TIMESTAMPTZ NOT NULL`	Auto-set by the database
`txid`	`XID8 NOT NULL`	Appending transaction id, used to compute the commit-safe watermark so cursor consumers never skip a late-committing lower position

Constraints:

PRIMARY KEY (global_position) — global ordering
UNIQUE (stream_id, stream_version) — enforces sequential versions per stream, serves as OCC safety net

Indexes:

CREATE INDEX IF NOT EXISTS idx_events_stream_id
  ON {schema}.events (stream_id, stream_version);

CREATE INDEX IF NOT EXISTS idx_events_event_type
  ON {schema}.events (event_type);

CREATE INDEX IF NOT EXISTS idx_events_created_at
  ON {schema}.events (created_at);

Event-backed snapshots

Snapshots do not use a separate table. They are normal rows in {schema}.events, written to the same stream they optimize.

For example, a BankAccountBalance snapshot for stream Transaction-account-123 is stored as an event with event_type = 'BankAccountBalanceSnapshot' in that same stream. Because it is a real event row, it advances stream_version and participates in the normal stream ordering and OCC rules.

`outbox`

Transactional outbox for at-least-once delivery.

CREATE TABLE IF NOT EXISTS {schema}.outbox (
  id               BIGSERIAL        PRIMARY KEY,
  event_global_pos BIGINT           NOT NULL,
  topic            TEXT             NOT NULL,
  payload          JSONB            NOT NULL,
  processed_at     TIMESTAMPTZ      NULL,
  created_at       TIMESTAMPTZ      NOT NULL DEFAULT now()
);

Column	Type	Description
`id`	`BIGSERIAL PK`	Auto-incrementing outbox entry ID
`event_global_pos`	`BIGINT NOT NULL`	Reference to `events.global_position`
`topic`	`TEXT NOT NULL`	Message topic/channel name
`payload`	`JSONB NOT NULL`	CloudEvents v1.0.2 JSON payload
`processed_at`	`TIMESTAMPTZ NULL`	`NULL` = pending, set = processed
`created_at`	`TIMESTAMPTZ NOT NULL`	Auto-set by database

CREATE INDEX IF NOT EXISTS idx_outbox_pending
  ON {schema}.outbox (created_at) WHERE processed_at IS NULL;

`crypto_keys`

Per-entity encryption keys for GDPR crypto-shredding.

CREATE TABLE IF NOT EXISTS {schema}.crypto_keys (
  key_id           TEXT             PRIMARY KEY,
  encrypted_key    BYTEA            NULL,
  algorithm        TEXT             NOT NULL DEFAULT 'aes-256-gcm',
  revoked_at       TIMESTAMPTZ      NULL,
  created_at       TIMESTAMPTZ      NOT NULL DEFAULT now()
);

Column	Type	Description
`key_id`	`TEXT PK`	Entity key identifier (e.g. `"user:abc123"`)
`encrypted_key`	`BYTEA NULL`	Versioned envelope; `NULL` after revocation
`algorithm`	`TEXT NOT NULL`	Always `"aes-256-gcm"`
`revoked_at`	`TIMESTAMPTZ NULL`	`NULL` = active, set = revoked
`created_at`	`TIMESTAMPTZ NOT NULL`	Auto-set by database

Revoked key rows are retained only for audit metadata. Their encrypted_key is set to NULL during revocation.

`projections`

Checkpoint tracking for projection runners.

CREATE TABLE IF NOT EXISTS {schema}.projections (
  projection_name  TEXT             PRIMARY KEY,
  last_position    BIGINT           NOT NULL DEFAULT 0,
  updated_at       TIMESTAMPTZ      NOT NULL DEFAULT now()
);

Column	Type	Description
`projection_name`	`TEXT PK`	Unique projection identifier
`last_position`	`BIGINT NOT NULL`	Last processed `global_position`
`updated_at`	`TIMESTAMPTZ NOT NULL`	Last checkpoint update time

Migrations

Migrations are run by calling eventStore.setup(). This executes CREATE TABLE IF NOT EXISTS and CREATE INDEX IF NOT EXISTS statements, making it safe to call on every application startup.

await eventStore.setup(); // Idempotent — run on every startup

Database Schema

On this page