@rotorsoft/act-pg

@rotorsoft/act-pg

PostgreSQL event store adapter for @rotorsoft/act. Provides persistent, production-ready event storage with ACID guarantees, connection pooling, and distributed stream processing.

Installation

npm install @rotorsoft/act @rotorsoft/act-pg
# or
pnpm add @rotorsoft/act @rotorsoft/act-pg

Requirements: Node.js >= 22.18.0, PostgreSQL >= 14

Usage

import { act, state, store } from "@rotorsoft/act";
import { PostgresStore } from "@rotorsoft/act-pg";
import { z } from "zod";

// Inject the PostgreSQL store before building your app
store(new PostgresStore({
  host: "localhost",
  port: 5432,
  database: "myapp",
  user: "postgres",
  password: "secret",
}));

// Initialize tables (creates schema, events table, streams table, and indexes)
await store().seed();

// Build and use your app as normal
const Counter = state({ Counter: z.object({ count: z.number() }) })
  .init(() => ({ count: 0 }))
  .emits({ Incremented: z.object({ amount: z.number() }) })
  .patch({ Incremented: ({ data }, s) => ({ count: s.count + data.amount }) })  // optional — only for custom reducers
  .on({ increment: z.object({ by: z.number() }) })
    .emit((action) => ["Incremented", { amount: action.by }])
  .build();

const app = act().withState(Counter).build();
await app.do("increment", { stream: "counter1", actor: { id: "1", name: "User" } }, { by: 1 });

Configuration

All configuration fields are optional and have sensible defaults:

Option	Default	Description
host	localhost	PostgreSQL host
port	5432	PostgreSQL port
database	postgres	Database name
user	postgres	Database user
password	postgres	Database password
schema	public	Schema for event tables
table	events	Base name for event tables

Custom Schema and Table Names

const pgStore = new PostgresStore({
  host: "db.example.com",
  database: "production",
  user: "app_user",
  password: process.env.DB_PASSWORD,
  schema: "events",       // custom schema
  table: "act_events",    // creates act_events and act_events_streams tables
});

Environment-Based Configuration

if (process.env.NODE_ENV === "production") {
  store(new PostgresStore({
    host: process.env.DB_HOST,
    port: parseInt(process.env.DB_PORT || "5432"),
    database: process.env.DB_NAME,
    user: process.env.DB_USER,
    password: process.env.DB_PASSWORD,
  }));
}
// In development, the default InMemoryStore is used

Features

• ACID Transactions - Events are committed atomically within PostgreSQL transactions
• Optimistic Concurrency - Version-based conflict detection prevents lost updates
• Connection Pooling - Uses node-postgres Pool for efficient connection management
• Atomic Stream Claiming - Zero-contention competing consumers via FOR UPDATE SKIP LOCKED
• Auto Schema Setup - seed() creates all required tables, indexes, and schema
• NOTIFY/LISTEN - Real-time event notifications via PostgreSQL channels
• Multi-Tenant - Isolate tenants using separate schemas

Database Schema

Calling seed() creates two tables:

Events table ({schema}.{table}) - stores all committed events:

• id (serial) - global event sequence
• name - event type name
• data (jsonb) - event payload
• stream - stream identifier
• version - per-stream sequence number
• created (timestamptz) - event timestamp
• meta (jsonb) - correlation, causation, and actor metadata

Streams table ({schema}.{table}_streams) - tracks stream processing state for reactions:

• stream - stream identifier
• at - last processed event position
• leased_by / leased_until - distributed processing claim info
• blocked / error - error tracking for failed streams

Competing Consumer Pattern

The PostgreSQL adapter uses FOR UPDATE SKIP LOCKED for atomic stream claiming — the idiomatic PostgreSQL competing consumer pattern. The claim() method discovers streams with pending events and locks them in a single query:

• Workers never block each other — locked rows are silently skipped
• No race between discovery and locking (unlike a separate poll + lease)
• Same pattern used by pgBoss, Graphile Worker, and other production job queues
• Enables horizontal scaling by simply adding more workers

This replaces the previous two-step poll/lease approach, eliminating contention and simplifying the drain cycle.

Related

• @rotorsoft/act - Core framework
• Documentation
• Examples

License

MIT