Skip to main content

PostgresStore

@rotorsoft/act-root

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

Class: PostgresStore

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

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​

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();
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();
// PostgresStore uses node-postgres (pg) connection pooling
// Pool is created automatically with default settings
// For custom pool config, use environment variables:
//   PGHOST, PGPORT, PGDATABASE, PGUSER, PGPASSWORD
//   PGMAXCONNECTIONS, PGIDLETIMEOUT, etc.

const pgStore = new PostgresStore({
  host: "db.example.com",
  port: 5432,
  database: "production",
  user: "app_user",
  password: process.env.DB_PASSWORD
});
// 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();
}
// 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:257

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:214

notify?​

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

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

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:806

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:861

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:632

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:532

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

dispose()​

dispose(): Promise<void>

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

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:417

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

prioritize()​

prioritize(filter, priority): Promise<number>

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

Bulk-update priority of streams matching filter (ACT-102).

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:447

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_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:1162

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: filter form matches event-bearing streams (not subscription rows). When the filter sets source or blocked, the events table is joined against the streams subscription table since those concepts only exist for subscribed streams.

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:1050

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:958

Resets watermarks for the given streams to -1, making them eligible for replay from the beginning. Also clears retry, blocked, error, and lease state so the streams can be claimed immediately.

Prefer Act.reset() over calling this directly. This primitive only resets the store; it does not raise the orchestrator's internal "needs drain" flag, so a settled Act instance will short-circuit and skip the replay. Act.reset() wraps this and arms the flag.

Accepts either an explicit list of stream names or a StreamFilter for bulk operations (e.g., "rebuild every blocked stream"). The filter form is the same shape used by prioritize and query_streams. An empty filter ({}) matches every registered stream — typically a footgun for reset; prefer narrower filters like { blocked: true }.

Parameters​

input​

string[] | StreamFilter

Stream names or a StreamFilter

Returns​

Promise<number>

Count of streams that were actually reset

Example​

// By name
await app.reset(["my-projection"]);

// By filter — rebuild every blocked stream in a projection family
await app.reset({ stream: "^proj-orders-", blocked: true });

// Low-level (does NOT trigger replay on settled apps)
await store().reset(["my-projection"]);

See​

Act.reset for the high-level rebuild API that wraps this primitive and arms the orchestrator's drain flag

Implementation of​

Store.reset

restore()​

restore(driver): Promise<void>

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

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:311

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): Promise<{ subscribed: number; watermark: number; }>

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

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​

object[]

Streams to register with optional source.

Returns​

Promise<{ subscribed: number; watermark: number; }>

subscribed count and current max watermark.

Implementation of​

Store.subscribe

truncate()​

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

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

Atomically truncates streams and seeds each with a snapshot or tombstone.

Parameters​

targets​

object[]

Streams to truncate with optional snapshot state and meta.

Returns​

Promise<Map<string, { 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:992

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