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

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

Parameters​

config?​

Partial<Config> = {}

Returns​

PostgresStore

Properties​

config​

readonly config: Config

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

notify?​

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

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

Present only when config.notify is set. See Config.notify.

Parameters​

handler​

(notification) => void

Returns​

Promise<NotifyDisposer>

Implementation of​

Store.notify

Methods​

ack()​

ack(leases): Promise<Lease[]>

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

Finalizes leased streams atomically: acknowledges the ones that processed successfully and persists defer schedules for the ones that chose to be re-visited later — one call, one transaction.

Every entry advances the watermark to at (the last event handled this cycle) and releases the lease. An entry without Lease.due is a plain ack: it also resets retry to -1 and clears any schedule. An entry with due additionally defers the remainder: the stream's deferred_at is set to due (ms since epoch) and retry to the entry's own Lease.retry. Advance and defer are independent legs, so a partial-progress backoff/defer advances past the events it handled — that prefix never re-runs — while the failing tail waits for deferred_at (a hold passes at = the current watermark, making the advance a no-op). A caller deferring a deliberate re-visit passes retry: -1 (the same semantics as defer — a defer is not a failure); a caller pacing a retry backoff passes the climbing counter so the retry budget keeps accruing across windows until it blocks. All-or-nothing is the contract: a failure must leave every watermark and every schedule untouched, so a drain cycle's outcomes can never land partially (an acked close request must not survive a lost defer, and vice versa).

Parameters​

leases​

Lease[]

Leases to finalize; due-carrying entries defer, the rest ack. At most one entry per stream. Each lease is for one stream, and an adapter that applies the batch in one statement cannot apply two entries to the same row (UPDATE … FROM may touch a target row once, so Postgres keeps an arbitrary one). Callers de-duplicate before calling, as the drain does; adapters may assume uniqueness.

Returns​

Promise<Lease[]>

The acknowledged leases (deferred entries are not returned)

Example​

const leased = await store().claim(5, 5, randomUUID(), 10000);
// Ack most streams at ID 150; hold order-42 until half past
await store().ack(leased.map(l =>
  l.stream === "order-42"
    ? { ...l, due: Date.now() + 30 * 60_000 }
    : { ...l, at: 150 }
));

See​

  • claim for acquiring leases
  • defer for the standalone (operator-facing) schedule write

Implementation of​

Store.ack

block()​

block(leases): Promise<BlockedLease[]>

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

Blocks streams after persistent processing failures.

Blocked streams won't be returned by claim until manually unblocked. This prevents poison messages from repeatedly failing and consuming resources.

Streams are typically blocked when:

  • Max retries reached
  • blockOnError option is true
  • Handler throws an error

Parameters​

leases​

BlockedLease[]

Leases to block with error messages. At most one entry per stream, for the same reason as ack; adapters may assume uniqueness.

Returns​

Promise<BlockedLease[]>

Blocked leases

Example​

try {
  await processEvents(lease);
  await store().ack([lease]);
} catch (error) {
  if (lease.retry >= 3) {
    await store().block([{
      ...lease,
      error: error.message
    }]);
  }
}

See​

claim for lease management

Implementation of​

Store.block

claim()​

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

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

FOR UPDATE SKIP LOCKED: workers never block each other.

Parameters​

lagging​

number

leading​

number

by​

string

millis​

number

lane?​

string

Returns​

Promise<Lease[]>

Implementation of​

Store.claim

commit()​

commit<E>(appends, meta): Promise<Committed<E, keyof E>[]>

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

Appends events to one or more streams atomically: every append lands, or none does.

Each append names its stream, its events and, optionally, the version the stream must be at. The store must:

  • check every expectedVersion against its own stream before appending anything (-1 expects an empty stream, undefined skips the check), and throw ConcurrencyError naming the first stream that moved;
  • refuse a stream named twice and an append with no events (a plain Error), so a commit is never a version check alone;
  • assign global ids in append order, then event order, and make the events visible in that order;
  • continue each stream's versions from its head, and attach meta to every event.

No appends at all commits nothing and returns [].

Type Parameters​

E​

E extends Schemas

Event schemas

Parameters​

appends​

readonly Append<E>[]

One Append per stream

meta​

EventMeta

Event metadata (correlation, causation), shared by every event

Returns​

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

The committed events of every append, in append order

Throws​

If a stream's version doesn't match its expectedVersion

Example​

const events = await store().commit(
  [{ stream: "user-123", msgs: [{ name: "UserCreated", data: { email: "user@example.com" } }], expectedVersion: -1 }],
  { correlation: "req-456", causation: { action: {...} } }
);

Implementation of​

Store.commit

defer()​

defer(input, deferred_at): Promise<number>

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

claim compares deferred_at with the worker's clock, the one its defer timer uses.

Parameters​

input​

string[] | StreamFilter

deferred_at​

number

Returns​

Promise<number>

Implementation of​

Store.defer

dispose()​

dispose(): Promise<void>

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

Returns​

Promise<void>

Implementation of​

Store.dispose

drop()​

drop(): Promise<void>

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

Drops all data from the store.

Dangerous operation that deletes all events and state. Use with extreme caution, primarily for testing or development environments.

Returns​

Promise<void>

Example​

// Clean up after tests
afterAll(async () => {
  await store().drop();
});

Implementation of​

Store.drop

forget_pii()​

forget_pii(stream): Promise<number>

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

Row-level locks only. Disk space returns via autovacuum (VACUUM FULL for strict deletion).

Parameters​

stream​

string

Returns​

Promise<number>

Implementation of​

Store.forget_pii

prioritize()​

prioritize(filter, priority): Promise<number>

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

Bulk-update the scheduling priority of streams matching a filter.

Used by Act.prioritize for operator runtime control over lagging-frontier claim() ordering. Unlike subscribe, which keeps the per-stream priority at the max() of all registered reactions targeting that stream, prioritize sets the priority directly to priority for matching rows — letting operators override the build-time scheduling policy.

Filter semantics mirror query_streams: stream/source are regex by default, exact with the *_exact flags. blocked restricts to blocked or unblocked rows. Omitted fields don't filter. An empty filter ({}) updates every registered stream — useful for "reset all priorities to N" but a footgun otherwise.

Parameters​

filter​

StreamFilter

PrioritizeFilter selecting which streams to update. Required (use {} to target all).

priority​

number

New priority value. Set as-is — no max(), no clamp.

Returns​

Promise<number>

Count of streams whose priority was changed.

Example​

Boost a specific replay

await store().prioritize(
  { stream: "^projection-orders$", stream_exact: false },
  10
);

See​

  • Act.prioritize for the orchestrator-level wrapper
  • claim for how priority biases stream scheduling

Implementation of​

Store.prioritize

query()​

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

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

Queries events from the store with optional filtering.

Calls the callback for each matching event. The callback approach allows processing large result sets without loading everything into memory.

The adapter returns what it stored. A Date is stored as its ISO string and comes back as that string; the framework turns the fields an event's schema declares z.date() back into Dates, so an ISO-shaped z.string() stays a string. See the Store contract in docs/docs/architecture/extension-points.md.

Type Parameters​

E​

E extends Schemas

Event schemas

Parameters​

callback​

(event) => void

Function invoked for each matching event

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 filter criteria; see Query for the fields.

Returns​

Promise<number>

Total number of events processed

Example​

Query all events for a stream

let count = 0;
await store().query(
  (event) => {
    console.log(event.name, event.data);
    count++;
  },
  { stream: "user-123" }
);
console.log(`Found ${count} events`);

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

Heads only when no count/names is asked for; otherwise one full scan.

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

Streams registered subscription positions to a callback, plus the highest event id in the store.

Read-only introspection for operational dashboards (Store / Subscriptions tab, projection lag, blocked subscriptions). Avoids forcing apps to open a second connection and run raw SQL against adapter-specific schemas.

Mirrors the Store.query callback pattern: the callback is invoked once per matching position, allowing large result sets to be processed without buffering. Results are ordered by stream name; use query.after (last seen stream name) for keyset pagination on big tables (dynamic reactions can produce one subscription per aggregate).

Parameters​

callback​

(position) => void

Invoked once per matching StreamPosition.

query?​

QueryStreams

Optional QueryStreams filter (default limit: 100).

Returns​

Promise<QueryStreamsResult>

maxEventId and the count of positions emitted.

Example​

List blocked streams with their lag

const { maxEventId } = await store().query_streams(
  (s) => console.log(`${s.stream}: lag=${maxEventId - s.at} ${s.error}`),
  { blocked: true, limit: 50 }
);

Implementation of​

Store.query_streams

reset()​

reset(input): Promise<number>

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

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 drain-armed 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 schedules the replay

Implementation of​

Store.reset

restore()​

restore(driver): Promise<void>

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

One transaction; RESTART IDENTITY makes the serial assign dense ids from 1.

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

Initializes or resets the store.

Used primarily for testing to ensure a clean state between tests. For production stores, this might create necessary tables or indexes.

Returns​

Promise<void>

Example​

// Reset store between tests
beforeEach(async () => {
  await store().seed();
});

Implementation of​

Store.seed

subscribe()​

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

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

Registers streams for event processing.

Upserts stream entries so they become visible to claim. Used by correlate() to register dynamically discovered reaction target streams.

Also returns the current maximum watermark across all subscribed streams, used internally for correlation checkpoint initialization on cold start.

Parameters​

streams​

SubscribeInput[]

Streams to register with optional source hint — a literal stream name (matched by equality when the drain fetches the stream's events) or a regex pattern (compiled and matched against stream names); non-portable patterns are rejected here on SQLite.

At most one entry per stream. A subscription row is keyed by stream and its source is set once at creation, so a second entry for the same stream could only ever contribute its priority — and an adapter that merges the batch in one statement cannot do even that (UPDATE ... FROM may touch a target row only once, so Postgres picks an arbitrary source row). Callers de-duplicate before calling; adapters may assume uniqueness.

correlated_at?​

number

correlator?​
by​

string

key​

string

millis​

number

Returns​

Promise<SubscribeResult>

subscribed count of newly registered streams, watermark max at across all streams

Example​

const { subscribed, watermark } = await store().subscribe([
  { stream: "stats-user-1", source: "user-1" },
  { stream: "stats-user-2", source: "user-2", priority: 10 },
]);

See​

  • claim for discovering and leasing registered streams
  • prioritize for changing priority after subscription

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

Atomically truncates streams and seeds each with a final event.

For each full target (no before), in a single transaction:

  1. Deletes all events for the stream
  2. Inserts a __snapshot__ (when snapshot is provided) or __tombstone__ event as the sole event on the stream

Subscriptions are not touched, so the event log and the subscriptions can live on different systems. A retired stream's subscription is inert (commits to a tombstoned stream are refused, so its work mark never rises and claim never returns it) and keeps each consumer's final watermark. Deleting those rows is optional housekeeping.

A windowed target (before set) is a pure prefix delete behind a real snapshot the app wrote — no seed, no tombstone, and the streams table is left untouched. The store finds the closest safe boundary — the latest __snapshot__ with created < before and, when max_id is given, id <= max_id — and deletes events with id < that snapshot's id, keeping the snapshot + tail. No qualifying snapshot ⇒ no-op (the stream is absent from the result). Because load() resets state at each snapshot on replay, events below the boundary contribute nothing to any load result — deleting them cannot change what load() returns. snapshot/meta must be omitted on windowed targets; before takes precedence when both appear.

Windowed targets are applied before full ones, and each full target pairs its own delete with its own seed. Both only matter for inputs the framework never produces — close-cycle dedups by stream before calling — but Store is a public port, so the order is contract rather than accident: a stream listed as both windowed and full ends with its FULL entry in the result, and a stream listed twice as full leaves exactly one seed carrying the last target's snapshot, reporting only that pass's own delete count. Deleting once for a deduped set and seeding once per entry would write two version: 0 rows, which the SQL adapters' unique (stream, version) constraint rejects outright.

Parameters​

targets​

object[]

Streams to truncate; full targets carry optional snapshot state and meta, windowed targets carry before (and optionally max_id, the min consumer watermark cap)

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 (the new seed, or the surviving boundary snapshot on windowed entries, which also echo before)

See​

Act.close for the high-level close-the-books API that orchestrates safety checks, archive callbacks, and atomic truncate+seed

Implementation of​

Store.truncate

unblock()​

unblock(input): Promise<number>

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

retry = -1, so the first claim after unblock reports attempt 0.

Parameters​

input​

string[] | StreamFilter

Returns​

Promise<number>

Implementation of​

Store.unblock