Skip to main content

Store

@rotorsoft/act-root

@rotorsoft/act-root / act/src / Store

Interface: Store

Defined in: libs/act/src/types/ports.ts:96

Interface for event store implementations.

The Store interface defines the contract for persistence adapters in Act. Implementations must provide event storage, querying, and distributed processing capabilities through leasing and watermark tracking.

Act includes two built-in implementations:

  • InMemoryStore: For development and testing
  • PostgresStore: For production use with PostgreSQL

Custom stores can be implemented for other databases or event log systems.

Example​

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

// Replace the default in-memory store
store(new PostgresStore({
  host: "localhost",
  port: 5432,
  database: "myapp",
  user: "postgres",
  password: "secret"
}));

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

See​

  • InMemoryStore for the default implementation
  • PostgresStore for the PostgreSQL implementation

Extends​

Properties​

ack​

ack: (leases) => Promise<Lease[]>

Defined in: libs/act/src/types/ports.ts:216

Acknowledges successful processing of leased streams.

Updates the watermark to indicate events have been processed successfully. Releases the lease so other workers can process subsequent events.

Parameters​

leases​

Lease[]

Leases to acknowledge with updated watermarks

Returns​

Promise<Lease[]>

Acknowledged leases

Example​

const leased = await store().claim(5, 5, randomUUID(), 10000);
// Process events up to ID 150
await store().ack(leased.map(l => ({ ...l, at: 150 })));

See​

claim for acquiring leases

block​

block: (leases) => Promise<Lease & object[]>

Defined in: libs/act/src/types/ports.ts:312

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​

Lease & object[]

Leases to block with error messages

Returns​

Promise<Lease & object[]>

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

claim​

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

Defined in: libs/act/src/types/ports.ts:248

Atomically discovers and leases streams for reaction processing.

Combines poll and lease into a single operation, eliminating the race condition where another worker can grab a stream between poll and lease.

PostgresStore uses FOR UPDATE SKIP LOCKED for zero-contention competing consumer semantics — workers never block each other, each grabbing different streams atomically. InMemoryStore fuses its poll+lease logic equivalently.

Used by Act.drain() as the primary stream acquisition method.

Parameters​

lagging​

number

Max streams from the lagging frontier (ascending watermark)

leading​

number

Max streams from the leading frontier (descending watermark)

by​

string

Unique lease holder identifier (UUID)

millis​

number

Lease duration in milliseconds

Returns​

Promise<Lease[]>

Array of successfully leased streams with metadata

Example​

const leased = await store().claim(5, 5, randomUUID(), 10000);
leased.forEach(({ stream, at, lagging }) => {
  console.log(`Leased ${stream} at ${at} (lagging: ${lagging})`);
});

See​

  • subscribe for registering new streams (used by correlate)
  • ack for acknowledging completion
  • block for blocking failed streams

commit​

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

Defined in: libs/act/src/types/ports.ts:157

Commits one or more events to a stream atomically.

This is the core method for persisting events. It must:

  • Assign global sequence IDs to events
  • Increment the stream version
  • Check optimistic concurrency if expectedVersion is provided
  • Store events atomically (all or nothing)
  • Attach metadata (id, stream, version, created timestamp)

Type Parameters​

E​

E extends Schemas

Event schemas

Parameters​

stream​

string

The stream ID to commit to

msgs​

Message<E, keyof E>[]

Array of messages (events) to commit

meta​

EventMeta

Event metadata (correlation, causation)

expectedVersion?​

number

Expected current version for optimistic concurrency

Returns​

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

Array of committed events with full metadata

Throws​

If expectedVersion doesn't match current version

Example​

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

dispose​

dispose: Disposer

Defined in: libs/act/src/types/ports.ts:59

Inherited from​

Disposable.dispose

drop​

drop: () => Promise<void>

Defined in: libs/act/src/types/ports.ts:126

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();
});

query​

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

Defined in: libs/act/src/types/ports.ts:193

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.

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; with_snaps?: boolean; }>

Optional filter criteria

Returns​

Promise<number>

Total number of events processed

Example​

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

seed​

seed: () => Promise<void>

Defined in: libs/act/src/types/ports.ts:111

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();
});

subscribe​

subscribe: (streams) => Promise<{ subscribed: number; watermark: number; }>

Defined in: libs/act/src/types/ports.ts:277

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​

object[]

Streams to register with optional source hint

Returns​

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

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" },
]);

See​

claim for discovering and leasing registered streams