Skip to main content
Version: Current

SqliteStore

@rotorsoft/act-root

@rotorsoft/act-root / act-sqlite/src / SqliteStore

Class: SqliteStore

Defined in: libs/act-sqlite/src/sqlite-store.ts:214

SQLite event store adapter for @rotorsoft/act.

Provides persistent event storage using SQLite via @libsql/client. All write operations use transactions for ACID guarantees.

SQLite is a single-writer store: concurrent write transactions raise SQLITE_BUSY rather than serializing transparently, so it does not support competing consumers the way PostgreSQL's FOR UPDATE SKIP LOCKED does (the TCK leaves concurrent_claim off for this reason). The intended model is a single drain worker per database file.

Store.notify is intentionally not implemented. The notify hook is a cross-process wake-up signal that lets a horizontally-scaled Act deployment wake settle() immediately on remote commits. SQLite is single-node by design — there is no remote writer to be notified of — so the Act orchestrator falls back to the existing debounce/poll path, which is correct for this topology.

Example​

import { store } from "@rotorsoft/act";
import { SqliteStore } from "@rotorsoft/act-sqlite";

store(new SqliteStore({ url: "file:myapp.db" }));
await store().seed();

Implements​

Constructors​

Constructor​

new SqliteStore(config?): SqliteStore

Defined in: libs/act-sqlite/src/sqlite-store.ts:225

Parameters​

config?​

Partial<SqliteConfig> = {}

Returns​

SqliteStore

Methods​

ack()​

ack(leases): Promise<Lease[]>

Defined in: libs/act-sqlite/src/sqlite-store.ts:746

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

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-sqlite/src/sqlite-store.ts:802

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

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-sqlite/src/sqlite-store.ts:618

Atomically discovers and leases streams for reaction processing.

Atomically discovers a stream and acquires a lease in one round-trip, eliminating the race that exists when discovery and locking are separate calls (a competing worker can grab the stream between the two).

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.

A subscription's source is matched against candidate streams by one of two rules in the has-work probe. A literal source (no regex metacharacter — the common case, and every autoclose/dynamic-resolver source) matches by string equality: the fast, index-friendly path, and exact so "s1" never matches "s12". A pattern source (carrying ^ $ . * + ? ( ) [ ] { } | \, e.g. a static ^(A|B)$ reaction) is compiled as a regex and matched against candidate stream names. Adapters that cannot faithfully run an arbitrary regex (SQLite) reject a non-portable pattern at subscribe time rather than silently never claiming the stream.

Every granted lease counts against the stream's retry budget: claim increments the stream's retry counter and only ack resets it, so a timed-out lease reclaimed by any worker marches the stream toward blockOnError exactly like a handler failure.

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

lane?​

string

Optional lane filter (ACT-1103)

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

Implementation of​

Store.claim

commit()​

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

Defined in: libs/act-sqlite/src/sqlite-store.ts:380

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

Implementation of​

Store.commit

defer()​

defer(input, deferred_at): Promise<number>

Defined in: libs/act-sqlite/src/sqlite-store.ts:847

Operator verb: bulk-pause streams until a future time without advancing their watermark. The drain itself never calls this — a reaction's defer outcome is persisted atomically by ack via due-marked leases; this standalone verb exists for operator-driven scheduling ("hold every webhook-.* stream until the maintenance window ends"), completing the recovery family: reset / unblock / prioritize / defer.

Sets deferred_at on each matched stream. claim skips any stream whose deferred_at is still in the future, so a paused stream is not re-claimed (and retry is never bumped) until the due-time passes, at which point the same pending events are re-delivered. Unlike in-process backoff, this is durable, shared store state — every competing worker honors the skip.

The schedule is cleared whenever the watermark moves or the stream is recovered: ack, block, reset, and unblock all reset deferred_at. Re-deferring simply overwrites it.

Accepts an explicit list of stream names or a StreamFilter (regex by default), the same shape as reset/unblock.

Parameters​

input​

string[] | StreamFilter

Stream names or a StreamFilter selecting streams

deferred_at​

number

Wall-clock time (ms since epoch) to revisit the streams

Returns​

Promise<number>

Count of streams whose deferred_at was set

Example​

// Pause every webhook delivery stream during downstream maintenance
await store().defer({ stream: "^webhook-" }, Date.now() + 30 * 60_000);

Implementation of​

Store.defer

dispose()​

dispose(): Promise<void>

Defined in: libs/act-sqlite/src/sqlite-store.ts:374

Returns​

Promise<void>

Implementation of​

Store.dispose

drop()​

drop(): Promise<void>

Defined in: libs/act-sqlite/src/sqlite-store.ts:369

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-sqlite/src/sqlite-store.ts:1585

Wipe the sensitive-data payload for every event on the stream — the physical-erasure side of the sensitive-data epic (#566). Sets events.pii to NULL for the stream's events; events.data and the rest of the row are never touched.

Single UPDATE under SQLite's writer lock, bounded by events-per- stream. Idempotent — the pii IS NOT NULL predicate filters out already-wiped rows so a second call returns 0.

SQLite doesn't auto-reclaim space; freed pages stay in the file until an operator-scheduled PRAGMA incremental_vacuum or a full VACUUM. The production checklist documents the cadence.

Parameters​

stream​

string

Target stream

Returns​

Promise<number>

Count of events whose pii was set to NULL

Implementation of​

Store.forget_pii

prioritize()​

prioritize(filter, priority): Promise<number>

Defined in: libs/act-sqlite/src/sqlite-store.ts:1390

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.

Examples​

Boost a specific replay

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

De-prioritize all background projections

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

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-sqlite/src/sqlite-store.ts:464

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.

Payload dates round-trip as Date. A Date committed inside event data (or meta) is returned as a Date, not an ISO string — every adapter revives ISO-8601 strings on read. A consequence: a plain string that happens to match ISO-8601 exactly is revived to a Date too, and a timezone-less ISO string is parsed in the reader's local time. Keep ISO-shaped strings you want to stay strings out of event payloads, or carry them in a wrapper field.

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

Optional filter criteria — see Query for fields (stream, name, after, before, created_after, created_before, limit, with_snaps, stream_exact).

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-sqlite/src/sqlite-store.ts:1089

Per-stream aggregated stats — see Store.query_stats.

Two code paths (mirrors the PostgresStore strategy):

  • Heads-only path (no count, no names): one or two queries using ROW_NUMBER() OVER (PARTITION BY stream ORDER BY version DESC|ASC) (SQLite lacks PG's DISTINCT ON). Window function + WHERE rn = 1 materializes the head (or tail) per stream from the (stream, version) unique index. Parallel Promise.all when tail is requested.

  • Full-scan path (count or names set): one CTE materializes the filtered events, then GROUP BY stream, name → json_group_object(name, n) for the names map plus SUM(n) for count. Heads (and tails when requested) come from the same scan.

SQLite specifics:

  • data and meta are stored as TEXT (JSON-encoded); the reader JSON-parses them when materializing the Committed rows.
  • blocked is stored as 0/1 integer; the filter converts.
  • Array input expands to a placeholder list (IN (?, ?, ...)) since SQLite has no native array type.

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-sqlite/src/sqlite-store.ts:992

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.

Examples​

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

Page through all positions

let after: string | undefined;
for (;;) {
  const page: StreamPosition[] = [];
  const { count } = await store().query_streams(
    (s) => page.push(s),
    { after, limit: 100 }
  );
  if (!count) break;
  // ... use page ...
  after = page.at(-1)?.stream;
}

Implementation of​

Store.query_streams

reset()​

reset(input): Promise<number>

Defined in: libs/act-sqlite/src/sqlite-store.ts:920

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-sqlite/src/sqlite-store.ts:1532

Atomically wipe-and-rebuild the store inside a single libsql write transaction.

On any throw inside the driver the transaction rolls back and the store is byte-for-byte unchanged. DELETE FROM events + DELETE FROM streams wipe both tables; DELETE FROM sqlite_sequence WHERE name = 'events' resets the autoincrement counter so the new sequence is dense 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-sqlite/src/sqlite-store.ts:281

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

Defined in: libs/act-sqlite/src/sqlite-store.ts:556

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 — a literal stream name (matched by equality in claim's has-work probe) or a regex pattern (compiled and matched against candidate streams); non-portable patterns are rejected here on SQLite

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", 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-sqlite/src/sqlite-store.ts:1408

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. Removes the stream's entry from the streams table
  3. Inserts a __snapshot__ (when snapshot is provided) or __tombstone__ event as the sole event on the stream

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.

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-sqlite/src/sqlite-store.ts:955

Clears the blocked flag on streams without replaying their history. Sets blocked = false, retry_count = 0, error = null, and clears any lease bookkeeping. The at watermark stays where it was — the stream resumes from the next event after the last successful ack, not from zero.

The distinction from reset matters: reset() is for projection rebuilds (replay from event 0); unblock() is for recovering from a poison message after the operator fixes the underlying issue. Use unblock() when you don't want to re-process history.

Prefer Act.unblock() over calling this directly. Like reset(), this primitive doesn't raise the orchestrator's internal "needs drain" flag — a settled Act instance will short-circuit and skip the resume. Act.unblock() wraps this and arms the flag.

Only streams that were actually blocked at call time count toward the return value; already-unblocked streams and unknown stream names are silently skipped. The atomic single-statement update makes the call safe to issue concurrently with claim() — workers holding a FOR UPDATE SKIP LOCKED lock won't see partial state.

Accepts either an explicit list of stream names or a StreamFilter for bulk recovery (e.g., "unblock every blocked order projection"). The blocked = true predicate is always applied — passing blocked: false in the filter matches nothing. An empty filter ({}) means "unblock everything that's blocked," which is a sane post-incident bulk recovery.

Parameters​

input​

string[] | StreamFilter

Stream names or a StreamFilter

Returns​

Promise<number>

Count of streams that were actually flipped (were blocked)

Example​

// By name (single targeted recovery)
await app.unblock(["webhooks-out-customer-42"]);

// By filter — unblock every blocked stream in a family
await app.unblock({ stream: "^webhooks-out-" });

// Post-incident: unblock everything that's blocked
await app.unblock({});

// Low-level (does NOT trigger resume on settled apps)
await store().unblock(["webhooks-out-customer-42"]);

See​

  • Act.unblock for the high-level recovery API
  • reset for the rebuild-from-zero alternative

Implementation of​

Store.unblock