Skip to main content
Version: Current

PostgresStore

@rotorsoft/act-root

@rotorsoft/act-root / act-pg/src / PostgresStore

Class: PostgresStore

Defined in: libs/act-pg/src/postgres-store.ts:325

Production-ready PostgreSQL event store implementation.

PostgresStore provides persistent, scalable event storage using PostgreSQL. It implements the full Store interface with production-grade features:

Features:

  • Persistent event storage with ACID guarantees
  • Optimistic concurrency control via version numbers
  • Distributed stream processing with leasing
  • Snapshot support for performance optimization
  • Connection pooling for scalability
  • Automatic table and index creation

Database Schema:

  • Events table: Stores all committed events
  • Streams table: Tracks stream metadata and leases
  • Indexes on stream, version, and timestamps for fast queries

Examples​

Basic setup

import { store } from "@rotorsoft/act";
import { PostgresStore } from "@rotorsoft/act-pg";

store(new PostgresStore({
  host: "localhost",
  port: 5432,
  database: "myapp",
  user: "postgres",
  password: "secret"
}));

const app = act()
  .withState(Counter)
  .build();

With custom schema and table

import { PostgresStore } from "@rotorsoft/act-pg";

const pgStore = new PostgresStore({
  host: process.env.DB_HOST || "localhost",
  port: parseInt(process.env.DB_PORT || "5432"),
  database: process.env.DB_NAME || "myapp",
  user: process.env.DB_USER || "postgres",
  password: process.env.DB_PASSWORD,
  schema: "events",      // Custom schema
  table: "act_events"    // Custom table name
});

// Initialize tables
await pgStore.seed();

Connection pooling configuration

// PostgresStore uses node-postgres (pg) connection pooling with
// opinionated defaults: max 20, connectionTimeoutMillis 10s,
// idleTimeoutMillis 30s, statement_timeout 60s. Any pg.PoolConfig
// field passed to the constructor overrides the default.

const pgStore = new PostgresStore({
  host: "db.example.com",
  port: 5432,
  database: "production",
  user: "app_user",
  password: process.env.DB_PASSWORD,
  max: 40  // lanes × streamLimit + API concurrency + headroom
});

Multi-tenant setup

// Use separate schemas per tenant
const tenants = ["tenant1", "tenant2", "tenant3"];

for (const tenant of tenants) {
  const tenantStore = new PostgresStore({
    host: "localhost",
    database: "multitenant",
    schema: tenant,        // Each tenant gets own schema
    table: "events"
  });
  await tenantStore.seed();
}

Querying PostgreSQL directly

// For advanced queries, you can access pg client
const pgStore = new PostgresStore(config);
await pgStore.seed();

// Use the store's query method for standard queries
await pgStore.query(
  (event) => console.log(event),
  { stream: "user-123", limit: 100 }
);

See​

Implements​

Constructors​

Constructor​

new PostgresStore(config?): PostgresStore

Defined in: libs/act-pg/src/postgres-store.ts:411

Create a new PostgresStore instance.

Parameters​

config?​

Partial<Config> = {}

Partial configuration (host, port, user, password, schema, table, etc.)

Returns​

PostgresStore

Properties​

config​

readonly config: Config

Defined in: libs/act-pg/src/postgres-store.ts:327

notify?​

optional notify?: (handler) => Promise<NotifyDisposer>

Defined in: libs/act-pg/src/postgres-store.ts:394

Cross-process commit subscription. Present only when config.notify === true — the orchestrator's auto-wire path checks if (store.notify), so omitting the method keeps single-instance deployments free of any LISTEN/NOTIFY overhead (no dedicated client, no per-commit pg_notify).

Parameters​

handler​

(notification) => void

Returns​

Promise<NotifyDisposer>

See​

Config.notify for the rationale and the multi-process contract.

Implementation of​

Store.notify

Methods​

ack()​

ack(leases): Promise<Lease[]>

Defined in: libs/act-pg/src/postgres-store.ts:1505

Acknowledge and release leases after processing, updating stream positions.

Parameters​

leases​

Lease[]

Leases to acknowledge, including last processed watermark and lease holder.

Returns​

Promise<Lease[]>

Acked leases.

Implementation of​

Store.ack

block()​

block(leases): Promise<BlockedLease[]>

Defined in: libs/act-pg/src/postgres-store.ts:1574

Block a stream for processing after failing to process and reaching max retries with blocking enabled.

Parameters​

leases​

BlockedLease[]

Leases to block, including lease holder and last error message.

Returns​

Promise<BlockedLease[]>

Blocked leases.

Implementation of​

Store.block

claim()​

claim(lagging, leading, by, millis, lane?): Promise<Lease[]>

Defined in: libs/act-pg/src/postgres-store.ts:1186

Atomically discovers and leases streams for reaction processing.

Uses FOR UPDATE SKIP LOCKED to implement zero-contention competing consumers:

  • Workers never block each other — locked rows are silently skipped
  • Discovery and locking happen in a single atomic transaction
  • No wasted polls — every returned stream is exclusively owned

Parameters​

lagging​

number

Max streams from lagging frontier (ascending watermark)

leading​

number

Max streams from leading frontier (descending watermark)

by​

string

Lease holder identifier (UUID)

millis​

number

Lease duration in milliseconds

lane?​

string

Returns​

Promise<Lease[]>

Leased streams with metadata

Implementation of​

Store.claim

commit()​

commit<E>(stream, msgs, meta, expectedVersion?): Promise<Committed<E, keyof E>[]>

Defined in: libs/act-pg/src/postgres-store.ts:1002

Commit new events to the store for a given stream, with concurrency control.

Type Parameters​

E​

E extends Schemas

Parameters​

stream​

string

The stream name

msgs​

Message<E, keyof E>[]

Array of messages (event name and data)

meta​

EventMeta

Event metadata (correlation, causation, etc.)

expectedVersion?​

number

(Optional) Expected stream version for concurrency control

Returns​

Promise<Committed<E, keyof E>[]>

Array of committed events

Throws​

ConcurrencyError if the expected version does not match

Implementation of​

Store.commit

defer()​

defer(input, deferred_at): Promise<number>

Defined in: libs/act-pg/src/postgres-store.ts:1632

Hold the matched streams out of claim until deferred_at (ms since epoch) — see Store.defer. Persists deferred_at (as a timestamptz) so the skip is honored by every competing worker; claim filters on deferred_at against the claiming worker's clock, the same clock its defer timer runs on. Accepts an explicit list of names or a StreamFilter, mirroring reset/prioritize. Cleared by ack/block/reset/unblock.

Parameters​

input​

string[] | StreamFilter

deferred_at​

number

Returns​

Promise<number>

Count of streams whose deferred_at was set.

Implementation of​

Store.defer

dispose()​

dispose(): Promise<void>

Defined in: libs/act-pg/src/postgres-store.ts:459

Dispose of the store and close all database connections. Releases any active LISTEN client first so the pool can drain cleanly.

Returns​

Promise<void>

Promise that resolves when all connections are closed

Implementation of​

Store.dispose

drop()​

drop(): Promise<void>

Defined in: libs/act-pg/src/postgres-store.ts:840

Drop all tables and schema created by the store (for testing or cleanup).

Returns​

Promise<void>

Promise that resolves when the schema is dropped

Implementation of​

Store.drop

forget_pii()​

forget_pii(stream): Promise<number>

Defined in: libs/act-pg/src/postgres-store.ts:2560

Wipe the sensitive-data payload for every event on the stream — the physical-erasure side of the sensitive-data epic. Sets events.pii to NULL for the stream's events; events.data and the rest of the row are never touched.

Row-level locks (no table lock), bounded by events-per-stream. Idempotent — a second call on an already-wiped stream returns 0.

Disk reclamation is autovacuum-driven; for strict-deletion jurisdictions the production checklist documents VACUUM FULL as the operator step.

Parameters​

stream​

string

Target stream

Returns​

Promise<number>

Count of events whose pii was set to NULL

Implementation of​

Store.forget_pii

prioritize()​

prioritize(filter, priority): Promise<number>

Defined in: libs/act-pg/src/postgres-store.ts:1768

Bulk-update priority of streams matching filter.

Filter semantics mirror query_streams: regex on stream / source by default, exact match with the _exact flags, blocked restricts to blocked or unblocked rows. Empty filter ({}) updates every registered stream.

Unlike subscribe (which keeps max() of registered priorities), this sets the priority outright — operator override for the build-time scheduling policy.

Parameters​

filter​

StreamFilter

priority​

number

Returns​

Promise<number>

Count of streams whose priority changed.

Implementation of​

Store.prioritize

query()​

query<E>(callback, query?): Promise<number>

Defined in: libs/act-pg/src/postgres-store.ts:870

Query events from the store, optionally filtered by stream, event name, time, etc.

Type Parameters​

E​

E extends Schemas

Parameters​

callback​

(event) => void

Function called for each event found

query?​

Readonly<{ after?: number; backward?: boolean; before?: number; correlation?: string; created_after?: Date; created_before?: Date; limit?: number; names?: string[]; stream?: string; stream_exact?: boolean; with_pii?: boolean; with_snaps?: boolean; }>

(Optional) Query filter (stream, names, before, after, etc.)

Returns​

Promise<number>

The number of events found

Example​

await store.query((event) => console.log(event), { stream: "A" });

Implementation of​

Store.query

query_stats()​

query_stats<E>(input, options?): Promise<Map<string, StreamStats<E>>>

Defined in: libs/act-pg/src/postgres-store.ts:1891

Per-stream aggregated stats — see Store.query_stats.

Two code paths chosen by the requested stats:

  • Heads-only path (no count, no names): one or two SELECT DISTINCT ON (stream) ... ORDER BY stream, version DESC|ASC queries, executed in parallel when tail: true. The (stream, version) unique index gives index-only access — K rows touched per query (K = matched streams), not N (events). Ordering by version (not id) is equivalent within a stream (versions are monotonic per stream and events are committed sequentially) and is the column actually indexed.

  • Full-scan path (count or names set): one CTE materializes the filtered events, then GROUP BY stream, name → jsonb_object_agg(name, n) for the names map plus per-stream COUNT(*) for count. Heads (and tails when requested) come from DISTINCT ON over the same CTE — they ride free on the already-paid scan.

The stream universe is derived from the events table: the filter (stream / stream_exact only) matches event-bearing streams, not subscription rows.

Type Parameters​

E​

E extends Schemas

Parameters​

input​

string[] | Pick<StreamFilter, "stream" | "stream_exact">

options?​

QueryStatsOptions<E>

Returns​

Promise<Map<string, StreamStats<E>>>

Implementation of​

Store.query_stats

query_streams()​

query_streams(callback, query?): Promise<QueryStreamsResult>

Defined in: libs/act-pg/src/postgres-store.ts:1787

Streams subscription positions to a callback, ordered by stream name, along with the highest event id in the store.

Filters (stream, source, blocked, after, limit) are applied server-side. stream/source are regex by default (~), or exact with *_exact: true — same convention as Store.query.

Parameters​

callback​

(position) => void

query?​

QueryStreams

Returns​

Promise<QueryStreamsResult>

maxEventId and the count of positions emitted.

Implementation of​

Store.query_streams

reset()​

reset(input): Promise<number>

Defined in: libs/act-pg/src/postgres-store.ts:1719

Reset watermarks to -1, clearing retry, blocked, error, and lease state so the matched streams can be replayed from the beginning.

Parameters​

input​

string[] | StreamFilter

Returns​

Promise<number>

Count of streams that were actually reset.

Implementation of​

Store.reset

restore()​

restore(driver): Promise<void>

Defined in: libs/act-pg/src/postgres-store.ts:2489

Atomically wipe-and-rebuild the store inside a single BEGIN/COMMIT transaction.

On any throw inside the driver the transaction rolls back and the store ends byte-for-byte unchanged. TRUNCATE ... RESTART IDENTITY CASCADE wipes events + resets the serial sequence to 1; the streams table is cleared in the same statement via CASCADE-like DELETE. Events are inserted one at a time with explicit columns (skipping id) so the serial assigns dense ids from 1. created is preserved verbatim from the source.

Parameters​

driver​

(callback) => Promise<void>

Returns​

Promise<void>

Implementation of​

Store.restore

seed()​

seed(): Promise<void>

Defined in: libs/act-pg/src/postgres-store.ts:514

Seed the database with required tables, indexes, and schema for event storage.

Returns​

Promise<void>

Promise that resolves when seeding is complete

Throws​

Error if seeding fails

Implementation of​

Store.seed

subscribe()​

subscribe(streams, correlated_at?, correlator?): Promise<SubscribeResult>

Defined in: libs/act-pg/src/postgres-store.ts:1353

Registers streams for event processing. Upserts stream entries so they become visible to claim(). Also returns the current max watermark across all subscriptions.

Parameters​

streams​

SubscribeInput[]

Streams to register with optional source.

correlated_at?​

number

correlator?​
by​

string

key​

string

millis​

number

Returns​

Promise<SubscribeResult>

subscribed count and current max watermark.

Implementation of​

Store.subscribe

truncate()​

truncate(targets): Promise<Map<string, { before?: Date; committed: Committed<Schemas, string>; deleted: number; }>>

Defined in: libs/act-pg/src/postgres-store.ts:2368

Atomically truncates streams and seeds each with a snapshot or tombstone. Windowed targets (before set) prune the prefix below the closest safe __snapshot__ instead — no seed, subscriptions untouched, no-op when no snapshot qualifies.

Parameters​

targets​

object[]

Streams to truncate with optional snapshot state and meta, or a before/max_id boundary for a windowed prefix delete.

Returns​

Promise<Map<string, { before?: Date; committed: Committed<Schemas, string>; deleted: number; }>>

Map keyed by stream name, each entry with deleted count and committed event.

Implementation of​

Store.truncate

unblock()​

unblock(input): Promise<number>

Defined in: libs/act-pg/src/postgres-store.ts:1742

Clear blocked flag (and retry / error / lease state) on streams without touching the at watermark. blocked = true is always applied, so the return count reflects only streams that were actually flipped — already-unblocked rows, unknown streams, and filter matches that aren't blocked are silently skipped.

retry = -1 matches the InMemoryStore convention: claim() bumps retry on every acquisition, so storing -1 means the first claim after unblock returns retry=0 ("first attempt"). Storing 0 would mis-report the post-recovery attempt as a continuation of the failed sequence. See Store.unblock.

Parameters​

input​

string[] | StreamFilter

Returns​

Promise<number>

Count of streams that were actually flipped (were blocked).

Implementation of​

Store.unblock