Skip to main content
Version: Current

Act

@rotorsoft/act-root

@rotorsoft/act-root / act/src / Act

Class: Act<TSchemaReg, TEvents, TActions, TStateMap, TActor>

Defined in: libs/act/src/act.ts:353

See​

Store

Main orchestrator for event-sourced state machines and workflows.

It manages the lifecycle of actions, reactions, and event streams, providing APIs for loading state, executing actions, querying events, and draining reactions.

Usage​

const app = act().withState(Counter).build();
await app.do("increment", { stream: "counter1", actor }, { by: 1 });
const snapshot = await app.load(Counter, "counter1");
await app.drain();
  • Register event listeners with .on("committed", ...) and .on("acked", ...) to react to lifecycle events.
  • Use .query() to analyze event streams for analytics or debugging.

Type Parameters​

TSchemaReg​

TSchemaReg extends SchemaRegister<TActions>

SchemaRegister for state

TEvents​

TEvents extends Schemas

Schemas for events

TActions​

TActions extends Schemas

Schemas for actions

TStateMap​

TStateMap extends Record<string, Schema> = Record<string, never>

Map of state names to state schemas

TActor​

TActor extends Actor = Actor

Actor type extending base Actor

Implements​

  • IAct<TEvents, TActions, TActor, TSchemaReg>

Constructors​

Constructor​

new Act<TSchemaReg, TEvents, TActions, TStateMap, TActor>(registry, states?, batch_handlers?, options?, lanes?, patch_fn?): Act<TSchemaReg, TEvents, TActions, TStateMap, TActor>

Defined in: libs/act/src/act.ts:562

Create a new Act orchestrator. Prefer the act builder over direct construction — act()...build() wires the registry, merges partial states, and collects batch handlers from registered slices and projections in one pass.

Parameters​

registry​

Registry<TSchemaReg, TEvents, TActions, keyof TStateMap & string>

Schemas for every event and action across registered states

states?​

Map<string, State<any, any, any>> = ...

Merged map of state name → state definition

batch_handlers?​

Map<string, BatchHandler<any>> = ...

Static-target projection batch handlers (target → handler)

options?​

ActOptions = {}

Tuning knobs — see ActOptions

lanes?​

readonly LaneConfig[] = []

Declared drain lanes. The builder collects these from .withLane(...) calls; one DrainController runs per lane, plus the implicit "default".

patch_fn?​

PatchFn = bare_patch

The per-event patch step selected once by the builder from ActOptions.validateFoldedState — bare_patch by default, validating_patch when the flag is on. The builder uses the same value for its projection-fold handlers, so there is a single selection site. Defaults to bare_patch for direct construction.

Returns​

Act<TSchemaReg, TEvents, TActions, TStateMap, TActor>

Properties​

registry​

readonly registry: Registry<TSchemaReg, TEvents, TActions, keyof TStateMap & string>

Defined in: libs/act/src/act.ts:400

Public registry — kept as-is per the no-prefix-on-public convention.

Accessors​

lanes​

Get Signature​

get lanes(): readonly LaneConfig[]

Defined in: libs/act/src/act.ts:538

Drain lanes declared via .withLane(...). Implicit default not included.

Returns​

readonly LaneConfig[]

Methods​

audit()​

audit(categories?, options?): AsyncIterable<AuditFinding>

Defined in: libs/act/src/act.ts:1801

Operator-driven store audit.

Walks the connected store and yields per-category findings — each tagged with the remediation it suggests. Same operator- driven category as app.close() / app.reset() / app.unblock() / app.blocked_streams(): never auto-invoked by the framework; the operator decides when to run it (CI gate, scheduled job, ad-hoc forensics) and what to do with the findings.

Categories are independent — pass a subset to scope the work, or omit to run everything:

// Targeted: schema drift + deprecated-event load only
for await (const f of app.audit(["schema", "deprecated-load"], {
  query: { created_after: lastScan },
  thresholds: { deprecatedLoadShareMin: 0.10 },
})) {
  await escalate(f);
}

// Full audit, default thresholds
for await (const f of app.audit()) console.log(f);

Returns an AsyncIterable so callers can break early — the underlying store paginations respect the iterator protocol and stop cleanly. Each finding is emitted independently, so pipelining into Slack / persistence / further analysis works without buffering the full report in memory.

Findings shape — see AuditFinding. The discriminated union carries enough context for the operator to act on each finding directly: stream id, event id, recommendation hints.

Parameters​

categories?​

AuditCategory[]

Subset of categories to run (default: all).

options?​

AuditOptions

Query window + per-category thresholds.

Returns​

AsyncIterable<AuditFinding>

Async iterable of AuditFinding.

blocked_streams()​

blocked_streams(options?): Promise<StreamPosition[]>

Defined in: libs/act/src/act.ts:1744

Return every currently-blocked stream position. Convenience wrapper around store().query_streams(cb, { blocked: true }) for the common "show me what's broken" operational query.

Results are ordered by stream name, paginated by limit (default 100). Pass after to fetch the next page (keyset cursor on the stream name). For richer queries — including blocked + source filters, or full unblocked introspection — drop to store().query_streams(...) directly.

Parameters​

options?​
after?​

string

limit?​

number

Returns​

Promise<StreamPosition[]>

Array of StreamPosition for currently-blocked streams.

Example​

Discover and recover

const blocked = await app.blocked_streams();
console.table(blocked.map(({ stream, retry, error }) => ({ stream, retry, error })));

// Operator investigates, then bulk-unblocks the family:
await app.unblock({ stream: "^webhooks-out-" });

close()​

close(targets): Promise<CloseResult>

Defined in: libs/act/src/act.ts:1945

Close the books — guard, archive, truncate, and optionally restart streams.

Safely removes historical events from the operational store:

  1. Correlate — discover pending reaction targets
  2. Safety check — skip streams with pending reactions (skipped when no reactive events)
  3. Guard — commit __tombstone__ with expectedVersion to block concurrent writes
  4. Load state — for streams in snapshots, load final state while guarded (no races)
  5. Archive — user callback per stream (abort-all on failure, streams are guarded)
  6. Truncate + seed — atomic: delete all events, insert __snapshot__ or __tombstone__
  7. Cache — invalidate (tombstoned) or warm (restarted)
  8. Emit "closed" — lifecycle event with results

Targets carrying before take the windowed branch instead — close the books on a rolling window: probe the min consumer watermark (so the boundary never rises past a lagging reaction), run the archive callback against the cutoff, then prune the prefix below the closest safe __snapshot__. No tombstone guard (the pre-cutoff prefix is immutable), no seed, cache untouched — the stream stays live and keeps accepting actions. Requires the state to snapshot via .snap(...); no qualifying snapshot ⇒ the stream lands in skipped (retry after the next snapshot). Windowed entries in the result echo before and carry the surviving boundary snapshot as committed.

Parameters​

targets​

CloseTarget[]

Per-stream close options (stream, restart?, archive?, before?)

Returns​

Promise<CloseResult>

{ truncated: TruncateResult, skipped: string[] }

Examples​

Archive and close

await app.close([
  { stream: "order-123", archive: async () => { await archiveToS3("order-123"); } },
  { stream: "order-456" },
]);

Close with restart (state loaded automatically after guard)

await app.close([
  { stream: "counter-1", restart: true },
  { stream: "counter-2" },  // tombstoned
]);

Windowed close — keep the last 180 days of real events

const cutoff = new Date();
cutoff.setDate(cutoff.getDate() - 180);
await app.close([
  {
    stream: "ledger-acme",
    before: cutoff,
    archive: async () => { await archiveToS3("ledger-acme", cutoff); },
  },
]);

correlate()​

correlate(query?): Promise<{ last_id: number; subscribed: number; }>

Defined in: libs/act/src/act.ts:1454

Discovers and registers new streams dynamically based on reaction resolvers.

Correlation enables "dynamic reactions" where target streams are determined at runtime based on event content. For example, you might create a stats stream for each user when they perform certain actions.

This method scans events matching the query and identifies new target streams based on reaction resolvers. It then registers these streams so they'll be picked up by the next drain cycle.

Parameters​

query?​

Query = ...

Query filter to scan for new correlations

Returns​

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

Object with newly leased streams and last scanned event ID

Examples​

Manual correlation

// Scan for new streams
const { leased, last_id } = await app.correlate({ after: 0, limit: 100 });
console.log(`Found ${leased.length} new streams`);

// Save last_id for next scan
await saveCheckpoint(last_id);

Dynamic stream creation

const app = act()
  .withState(User)
  .withState(UserStats)
  .on("UserLoggedIn")
    .do(async (event) => ["incrementLoginCount", {}])
    .to((event) => ({
      target: `stats-${event.stream}` // Dynamic target per user
    }))
  .build();

// Discover stats streams as users log in
await app.correlate();

See​

do()​

do<TKey>(action, target, payload, options?): Promise<Snapshot<TSchemaReg[TKey], TEvents>[]>

Defined in: libs/act/src/act.ts:1019

Executes an action on a state instance, committing resulting events.

This is the primary method for modifying state. It:

  1. Validates the action payload against the schema
  2. Loads the current state snapshot
  3. Checks invariants (business rules)
  4. Executes the action handler to generate events
  5. Applies events to create new state
  6. Commits events to the store with optimistic concurrency control

Type Parameters​

TKey​

TKey extends string | number | symbol

Action name from registered actions

Parameters​

action​

TKey

The name of the action to execute

target​

Target<TActor>

Target specification with stream ID and actor context

payload​

Readonly<TActions[TKey]>

Action payload matching the action's schema

options?​

DoOptions<TEvents>

Per-call dispatch options (DoOptions) — reactingTo to thread correlation, correlator to override the framework or orchestrator-level correlator for this call only.

Returns​

Promise<Snapshot<TSchemaReg[TKey], TEvents>[]>

Array of snapshots for all affected states (usually one)

Throws​

If payload doesn't match action schema

Throws​

If business rules are violated

Throws​

If another process modified the stream

Examples​

Basic action execution

const snapshots = await app.do(
  "increment",
  {
    stream: "counter-1",
    actor: { id: "user1", name: "Alice" }
  },
  { by: 5 }
);

console.log(snapshots[0].state.count); // Current count after increment

With error handling

try {
  await app.do(
    "withdraw",
    { stream: "account-123", actor: { id: "user1", name: "Alice" } },
    { amount: 1000 }
  );
} catch (error) {
  if (error instanceof InvariantError) {
    console.error("Business rule violated:", error.description);
  } else if (error instanceof ConcurrencyError) {
    console.error("Concurrent modification detected, retry...");
  } else if (error instanceof ValidationError) {
    console.error("Invalid payload:", error.details);
  }
}

Reaction triggering another action (reactingTo auto-injected)

const app = act()
  .withState(Order)
  .withState(Inventory)
  .on("OrderPlaced")
    .do(async function reduceInventory(event, _stream, app) {
      // Inside reaction handlers, reactingTo is auto-injected when omitted.
      // The triggering event is used by default, maintaining the correlation chain.
      await app.do(
        "reduceStock",
        { stream: "inventory-1", actor: { id: "sys", name: "system" } },
        { amount: event.data.items.length }
      );
      // To use a different correlation, pass reactingTo explicitly:
      // await app.do("reduceStock", target, payload, { reactingTo: customEvent });
    })
    .to("inventory-1")
  .build();

See​

Implementation of​

IAct.do

drain()​

drain(options?): Promise<Drain<TEvents>>

Defined in: libs/act/src/act.ts:1332

Processes pending reactions by draining uncommitted events from the event store.

Runs a single drain cycle:

  1. Polls the store for streams with uncommitted events
  2. Leases streams to prevent concurrent processing
  3. Fetches events for each leased stream
  4. Executes matching reaction handlers
  5. Acknowledges successful reactions or blocks failing ones

Drain uses a dual-frontier strategy to balance processing of new streams (lagging) vs active streams (leading). The ratio adapts based on event pressure.

Call correlate() before drain(). It is not only how dynamic targets are discovered: a stream is claimable while at < correlated_at, and correlate is the only component that raises that mark, so a commit no correlate has seen is not drainable, static targets included. For a higher-level API that handles debouncing, correlation, and signaling automatically, use settle.

Parameters​

options?​

DrainOptions = {}

Drain configuration — see DrainOptions for fields (streamLimit, eventLimit, leaseMillis).

Returns​

Promise<Drain<TEvents>>

Drain statistics with fetched, leased, acked, and blocked counts

Examples​

In tests and scripts

await app.do("createUser", target, payload);
await app.correlate();
await app.drain();

In production, prefer settle()

await app.do("CreateItem", target, input);
app.settle(); // debounced correlate→drain, emits "settled"

See​

emit()​

emit<E>(event, args): boolean

Defined in: libs/act/src/act.ts:420

Emit a lifecycle event. The payload type is inferred from the event name via ActLifecycleEvents.

Each listener is contained on its own: a throwing listener is logged, the remaining listeners still run, and the caller (do(), forget(), close(), the drain) never sees the throw, since its durable work has already landed.

Uses rawListeners so once wrappers still de-register themselves, and returns "had listeners" to preserve the EventEmitter.emit contract.

Type Parameters​

E​

E extends keyof ActLifecycleEvents<TSchemaReg, TEvents, TActions>

Parameters​

event​

E

args​

ActLifecycleEvents<TSchemaReg, TEvents, TActions>[E]

Returns​

boolean

forget()​

forget(stream): Promise<{ eventCount: number; }>

Defined in: libs/act/src/act.ts:1273

Wipe the sensitive-data payload for every event on the stream — see IAct.forget.

Throws on adapters without Store.forget_pii, invalidates the cache entry for the stream, emits the forgotten lifecycle event with the row count. Idempotent: a second call returns {eventCount: 0} and does NOT re-emit.

Parameters​

stream​

string

Target stream.

Returns​

Promise<{ eventCount: number; }>

{eventCount} — number of events whose PII column was wiped.

Implementation of​

IAct.forget

load()​

Call Signature​

load<TNewState, TNewEvents, TNewActions>(state, stream, callback?, asOf?): Promise<Snapshot<TNewState, TNewEvents>>

Defined in: libs/act/src/act.ts:1095

Loads the current state snapshot for a specific stream.

Reconstructs the current state by replaying events from the event store. Uses snapshots when available to optimize loading performance.

Accepts either a State definition object or a state name string. When using a string, the merged state (from partial states registered via .withState()) is resolved by name.

Type Parameters​
TNewState​

TNewState extends Schema

State schema type

TNewEvents​

TNewEvents extends Schemas

Event schemas type

TNewActions​

TNewActions extends Schemas

Action schemas type

Parameters​
state​

State<TNewState, TNewEvents, TNewActions>

The state definition or state name to load

stream​

string

The stream ID (state instance identifier)

callback?​

(snapshot) => void

Optional callback invoked with the loaded snapshot

asOf?​

AsOf

Returns​

Promise<Snapshot<TNewState, TNewEvents>>

The current state snapshot for the stream

Examples​

Load by state definition

const snapshot = await app.load(Counter, "counter-1");
console.log(snapshot.state.count);    // Current count
console.log(snapshot.patches);        // Events since last snapshot

Load by state name (useful with partial states)

const snapshot = await app.load("Ticket", "ticket-123");
console.log(snapshot.state.title);    // Merged state from all partials

Load multiple states

const [user, account] = await Promise.all([
  app.load(User, "user-123"),
  app.load(BankAccount, "account-456")
]);
See​

Snapshot for snapshot structure

Implementation of​

IAct.load

Call Signature​

load<TKey>(name, stream, callback?, asOf?): Promise<Snapshot<TStateMap[TKey], TEvents>>

Defined in: libs/act/src/act.ts:1105

Loads the current state snapshot for a specific stream.

Reconstructs the current state by replaying events from the event store. Uses snapshots when available to optimize loading performance.

Accepts either a State definition object or a state name string. When using a string, the merged state (from partial states registered via .withState()) is resolved by name.

Type Parameters​
TKey​

TKey extends string

Parameters​
name​

TKey

stream​

string

The stream ID (state instance identifier)

callback?​

(snapshot) => void

Optional callback invoked with the loaded snapshot

asOf?​

AsOf

Returns​

Promise<Snapshot<TStateMap[TKey], TEvents>>

The current state snapshot for the stream

Examples​

Load by state definition

const snapshot = await app.load(Counter, "counter-1");
console.log(snapshot.state.count);    // Current count
console.log(snapshot.patches);        // Events since last snapshot

Load by state name (useful with partial states)

const snapshot = await app.load("Ticket", "ticket-123");
console.log(snapshot.state.title);    // Merged state from all partials

Load multiple states

const [user, account] = await Promise.all([
  app.load(User, "user-123"),
  app.load(BankAccount, "account-456")
]);
See​

Snapshot for snapshot structure

Implementation of​

IAct.load

Call Signature​

load<TNewState, TNewEvents, TNewActions>(state, target, callback?): Promise<Snapshot<TNewState, TNewEvents>>

Defined in: libs/act/src/act.ts:1112

Loads the current state snapshot for a specific stream.

Reconstructs the current state by replaying events from the event store. Uses snapshots when available to optimize loading performance.

Accepts either a State definition object or a state name string. When using a string, the merged state (from partial states registered via .withState()) is resolved by name.

Type Parameters​
TNewState​

TNewState extends Schema

State schema type

TNewEvents​

TNewEvents extends Schemas

Event schemas type

TNewActions​

TNewActions extends Schemas

Action schemas type

Parameters​
state​

State<TNewState, TNewEvents, TNewActions>

The state definition or state name to load

target​

LoadTarget<TActor>

callback?​

(snapshot) => void

Optional callback invoked with the loaded snapshot

Returns​

Promise<Snapshot<TNewState, TNewEvents>>

The current state snapshot for the stream

Examples​

Load by state definition

const snapshot = await app.load(Counter, "counter-1");
console.log(snapshot.state.count);    // Current count
console.log(snapshot.patches);        // Events since last snapshot

Load by state name (useful with partial states)

const snapshot = await app.load("Ticket", "ticket-123");
console.log(snapshot.state.title);    // Merged state from all partials

Load multiple states

const [user, account] = await Promise.all([
  app.load(User, "user-123"),
  app.load(BankAccount, "account-456")
]);
See​

Snapshot for snapshot structure

Implementation of​

IAct.load

Call Signature​

load<TKey>(name, target, callback?): Promise<Snapshot<TStateMap[TKey], TEvents>>

Defined in: libs/act/src/act.ts:1121

Loads the current state snapshot for a specific stream.

Reconstructs the current state by replaying events from the event store. Uses snapshots when available to optimize loading performance.

Accepts either a State definition object or a state name string. When using a string, the merged state (from partial states registered via .withState()) is resolved by name.

Type Parameters​
TKey​

TKey extends string

Parameters​
name​

TKey

target​

LoadTarget<TActor>

callback?​

(snapshot) => void

Optional callback invoked with the loaded snapshot

Returns​

Promise<Snapshot<TStateMap[TKey], TEvents>>

The current state snapshot for the stream

Examples​

Load by state definition

const snapshot = await app.load(Counter, "counter-1");
console.log(snapshot.state.count);    // Current count
console.log(snapshot.patches);        // Events since last snapshot

Load by state name (useful with partial states)

const snapshot = await app.load("Ticket", "ticket-123");
console.log(snapshot.state.title);    // Merged state from all partials

Load multiple states

const [user, account] = await Promise.all([
  app.load(User, "user-123"),
  app.load(BankAccount, "account-456")
]);
See​

Snapshot for snapshot structure

Implementation of​

IAct.load

off()​

off<E>(event, listener): this

Defined in: libs/act/src/act.ts:469

Remove a previously registered lifecycle listener.

Type Parameters​

E​

E extends keyof ActLifecycleEvents<TSchemaReg, TEvents, TActions>

Parameters​

event​

E

listener​

(args) => void

Returns​

this

on()​

on<E>(event, listener): this

Defined in: libs/act/src/act.ts:456

Register a listener for a lifecycle event. The listener receives the event-specific payload.

Type Parameters​

E​

E extends keyof ActLifecycleEvents<TSchemaReg, TEvents, TActions>

Parameters​

event​

E

listener​

(args) => void

Returns​

this

prioritize()​

prioritize(filter, priority): Promise<number>

Defined in: libs/act/src/act.ts:1855

Bulk-update scheduling priority for streams matching filter.

Operator-grade override of the claim() lagging-frontier ordering. Useful when a long-running replay needs to jump ahead of other lagging streams, or when a no-longer-urgent job should yield slots back to the rest. Build-time priorities (set via the resolver's priority field) are subject to a max() invariant across reactions; this API ignores that and sets the priority outright on every matching row.

Filter shape mirrors query / Store.query_streams: stream / source are regex by default, exact with the *_exact flags; blocked restricts to blocked or unblocked rows. An empty filter ({}) updates every registered stream.

Parameters​

filter​

StreamFilter

Selection criteria (regex by default).

priority​

number

New priority value. Set as-is — no clamp.

Returns​

Promise<number>

Count of streams whose priority changed.

Examples​

Boost a specific projection mid-replay

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

Drop all audit projections to background

await app.prioritize({ source: "^audit-" }, -5);

Reset everyone to default

await app.prioritize({}, 0);

See​

  • Store.prioritize for the underlying primitive
  • claim for how priority biases scheduling

query()​

query(query, callback?): Promise<{ count: number; first?: Committed<TEvents, keyof TEvents>; last?: Committed<TEvents, keyof TEvents>; }>

Defined in: libs/act/src/act.ts:1202

Queries the event store for events matching a filter.

Use this for analyzing event streams, generating reports, or debugging. The callback is invoked for each matching event, and the method returns summary information (first event, last event, total count).

For small result sets, consider using query_array instead.

Parameters​

query​

Query

Filter criteria — see Query for available fields (stream, name, after, before, created_after, created_before, limit, with_snaps, stream_exact)

callback?​

(event) => void

Optional callback invoked for each matching event

Returns​

Promise<{ count: number; first?: Committed<TEvents, keyof TEvents>; last?: Committed<TEvents, keyof TEvents>; }>

Object with first event, last event, and total count

Examples​

Query all events for a stream

const { first, last, count } = await app.query(
  { stream: "counter-1" },
  (event) => console.log(event.name, event.data)
);
console.log(`Found ${count} events from ${first?.id} to ${last?.id}`);

Query specific event types

const { count } = await app.query(
  { name: "UserCreated", limit: 100 },
  (event) => {
    console.log("User created:", event.data.email);
  }
);

Query events in time range

const yesterday = new Date(Date.now() - 24 * 60 * 60 * 1000);
const { count } = await app.query({
  created_after: yesterday,
  stream: "user-123"
});
console.log(`User had ${count} events in last 24 hours`);

See​

query_array for loading events into memory

Implementation of​

IAct.query

query_array()​

query_array(query): Promise<Committed<TEvents, keyof TEvents>[]>

Defined in: libs/act/src/act.ts:1249

Queries the event store and returns all matching events in memory.

Use with caution - this loads all results into memory. For large result sets, use query with a callback instead to process events incrementally.

Parameters​

query​

Query

The query filter (same as query)

Returns​

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

Array of all matching events

Examples​

Load all events for a stream

const events = await app.query_array({ stream: "counter-1" });
console.log(`Loaded ${events.length} events`);
events.forEach(event => console.log(event.name, event.data));

Get recent events

const recent = await app.query_array({
  stream: "user-123",
  limit: 10
});

See​

query for large result sets

Implementation of​

IAct.query_array

reset()​

reset(input): Promise<number>

Defined in: libs/act/src/act.ts:1596

Reset reaction stream watermarks and request a drain on the next drain() / settle() cycle.

Use this to replay events through projections (or other reaction targets) after changing handler logic. Equivalent to calling store().reset(streams) directly, but also raises the orchestrator's internal "needs drain" flag — store().reset(...) alone leaves the flag untouched, so a settled app would short-circuit and skip the replay.

Pair with app.settle() (or a single app.drain() for small streams). settle() loops correlate→drain until no progress is made, so one call fully catches up paginated streams without forcing callers to roll their own loop.

Parameters​

input​

string[] | StreamFilter

Reaction target streams (e.g., projection names) to reset, or a StreamFilter for bulk operations

Returns​

Promise<number>

Count of streams that were actually reset

Examples​

Rebuild a projection (production)

await app.reset(["my-projection"]);
app.settle({ eventLimit: 1000 });   // emits "settled" when fully replayed

Rebuild a projection (tests / scripts)

await app.reset(["my-projection"]);
await app.drain({ eventLimit: 1000 });   // small streams: one pass is enough

See​

  • Store.reset for the underlying store primitive
  • settle for the debounced full-catch-up loop

restore()​

restore(source, opts?, sink?): Promise<ScanResult>

Defined in: libs/act/src/act.ts:1683

Atomically wipe the store and rebuild it from an event source. The framework owns iteration, validation, drop_snapshots filtering, on_progress, and the per-call old → new causation remap; the adapter's Store.restore driver supplies the transaction lifecycle and per-event insert.

Throws if the adapter has no restore capability. Throws on the first invalid event (negative version, malformed created) with the running index in the message; atomic transaction rollback in the adapter means a failing restore leaves the store byte-for-byte unchanged.

Parameters​

source​

EventSource

Any EventSource: a CsvFile backup or another store. Read in batches rather than buffered, so multi-million-event backups don't OOM. Each event's original id is used as a causation lookup key but never written through — adapters renumber densely.

opts?​

ScanOptions = {}

ScanOptions. drop_snapshots skips __snapshot__ events (counted in the result); on_progress fires once per event.

sink?​

EventSink

Where the events land. Defaults to this Act's store.

Returns​

Promise<ScanResult>

ScanResult with kept, duration_ms, and dropped per-category counters.

Examples​

Restore a CSV backup into the connected store

const result = await app.restore(new CsvFile({ path: "./backup.csv" }));
console.log(`Restored ${result.kept} events in ${result.duration_ms}ms`);
await cache().clear();   // operator's responsibility

Copy one store into another

await app.restore(pg, {}, sqlite);

See​

Store.restore for the underlying driver-pattern primitive.

settle()​

settle(options?): void

Defined in: libs/act/src/act.ts:1986

Debounced, non-blocking correlate→drain cycle.

Call this after app.do() (or app.reset()) to schedule a background drain. Multiple rapid calls within the debounce window are coalesced into a single cycle. Runs correlate→drain in a loop until a pass makes no progress — no new subscriptions, no acks, no blocks — then emits the "settled" lifecycle event. This means a single settle() call fully catches up paginated streams (e.g. after reset() on a long projection) without forcing callers to loop.

Parameters​

options?​

SettleOptions = {}

Settle configuration — see SettleOptions for fields: debounceMs (default 10), correlate (default { after: -1, limit: 100 }), maxPasses (default Infinity — kill-switch for runaway loops), streamLimit (default 10), eventLimit (default 10), leaseMillis (default 10000).

Returns​

void

Example​

API mutations

await app.do("CreateItem", target, input);
app.settle(); // non-blocking, returns immediately

app.on("settled", (drain) => {
  // notify SSE clients, invalidate caches, etc.
});

See​

  • drain for single synchronous drain cycles
  • correlate for manual correlation

shutdown()​

shutdown(options?): Promise<void>

Defined in: libs/act/src/act.ts:829

Per-instance teardown: stop scheduling new work, give drain cycles already in flight a bounded chance to finish, then remove lifecycle listeners and tear down the cross-process notify subscription.

Scheduling stops first, then in-flight cycles get up to graceMs to reach their ack (an abandoned handler holds its lease until it expires). Listeners come off after the wait, so they still see work that finished during it. The budget is a ceiling, not a delay; when it runs out, teardown proceeds anyway.

Idempotent — repeated calls return the same promise, and the first call's graceMs is the one that applies. Registered automatically with the global dispose() registry at construction, so process-wide dispose()() covers it; test helpers (or operators that mint short-lived Acts) call it explicitly for prompt cleanup.

Parameters​

options?​

ShutdownOptions

See ShutdownOptions. Defaults the grace budget to the largest lane leaseMillis (capped at 30s).

Returns​

Promise<void>

start_correlations()​

start_correlations(query?, frequency?, callback?): boolean

Defined in: libs/act/src/act.ts:1521

Starts a background poll that settles on a timer: each tick looks for new events, subscribes the streams their reactions target, and drains them.

Use it when this process can miss commits — another process writes to a store without notify, or a notification is lost. It is the safety net under notify: a missed wakeup costs at most one frequency of latency.

Only one poll runs per Act instance. On a writer-only instance (drain: false) each tick is a no-op, like settle().

Parameters​

query?​

Query = {}

Only limit is used: the most events each correlate pass reads (default 100). The scan always starts at the correlation checkpoint, so after is ignored.

frequency?​

number = 10_000

Milliseconds between ticks (default: 10000)

callback?​

(subscribed) => void

Called with the number of newly subscribed streams after each correlate pass that found some.

Returns​

boolean

true if the poll started, false if one is already running

Deprecated​

Listen to the "settled" lifecycle event instead.

Example​

app.start_correlations({ limit: 100 }, 5000);
// later
app.stop_correlations();

See​

stop_correlations()​

stop_correlations(): void

Defined in: libs/act/src/act.ts:1543

Stops the poll started by start_correlations. Idempotent, and called by shutdown().

Returns​

void

See​

start_correlations

stop_settling()​

stop_settling(): void

Defined in: libs/act/src/act.ts:1559

Cancels any pending or active settle cycle.

Returns​

void

See​

settle

unblock()​

unblock(input): Promise<number>

Defined in: libs/act/src/act.ts:1636

Clear the blocked flag on streams without replaying their history.

Use this to recover from a poison message after fixing the underlying issue — the stream resumes from the next event after the last successful ack, not from the beginning. Compare with reset, which rebuilds from event 0 (suitable for projection rebuilds, wrong for "I fixed the bug, please retry").

Wraps store().unblock(streams) and raises the orchestrator's internal "needs drain" flag so a settled app picks up the now-free streams on the next cycle. Equivalent to calling store().unblock(...) directly, but store().unblock(...) alone leaves the flag untouched.

Parameters​

input​

string[] | StreamFilter

Stream names to unblock, or a StreamFilter for bulk recovery

Returns​

Promise<number>

Count of streams that were actually flipped (were blocked)

Example​

Recover from a 4xx webhook after fixing the bug

await app.unblock(["webhooks-out-customer-42"]);
// The stream resumes from the next event, not from zero.

See​

  • Store.unblock for the underlying store primitive
  • reset for the rebuild-from-zero alternative