API Reference

Complete reference for the EventStore class, all exported builders, exported types, and error classes.

EventStore Class

Constructor

import { EventStore } from "@lox-solutions/alvyn";

new EventStore(config: EventStoreConfig)

interface EventStoreConfig {
  pool: Pool;
  schema?: string;
  secrets?: CryptoSecretsConfig;
  defaultSource?: string;
  snapshots?: SnapshotHandle<unknown>[];
}

Register independent snapshot handles here when you want Alvyn to maintain them on incoming writes. A snapshot is tied to a streamPrefix, but it is not always a 1:1 aggregate feature; one stream prefix can have multiple snapshots for different expensive calculations.

The schema name is validated against /^[a-z_][a-z0-9_]{0,62}$/. Invalid names throw InvalidSchemaNameError.

secrets contains the complete keyring and an explicit currentVersion; the order of entries is not significant:

secrets: {
  currentVersion: 2,
  secrets: [
    { version: 1, value: process.env.GDPR_CRYPTO_SECRET_V1! },
    { version: 2, value: process.env.GDPR_CRYPTO_SECRET_V2! },
  ],
}

The same configuration can be supplied through GDPR_CRYPTO_SECRETS=version:value,... and GDPR_CRYPTO_CURRENT_VERSION=2. Keep old entries during rotation; no database migration or downtime is required, and entity keys are lazily re-wrapped on their next encrypted write. Versions must be unsigned 32-bit integers, and currentVersion must be present in the keyring. Use a higher version for each new secret; gaps are allowed. In an HA deployment, deploy the complete keyring to every replica first, then change only currentVersion. See Crypto-Shredding for the rollout and old-secret removal procedure. All secret values are strengthened with scrypt; use a high-entropy generated value because a KDF cannot add entropy to a weak secret.

Lifecycle

setup(): Promise<void>

Runs idempotent schema migrations (CREATE TABLE IF NOT EXISTS). Safe on every startup. Must be called before any other method.

Concurrent in-flight calls within the same process are automatically deduplicated — returning the active initialization promise and avoiding redundant database connections.

Throws EventStoreNotInitializedError if other methods are called before setup().

isInitialized(): boolean

Returns whether setup() has completed successfully. Useful for synchronous health/readiness checks (e.g. /ready or /healthz endpoints) and defensive guards.

getStreamVersion(streamId): Promise<number>

Returns the current stream version, or 0 when the stream does not exist.

Stream Operations

append(input, options?): Promise<AppendResult>

Appends events to a stream within an ACID transaction.

const result = await eventStore.append({
  streamId: "Order-123",
  expectedVersion: 5,
  events: [
    {
      type: "OrderShipped",
      data: { trackingNumber: "TRACK-456" },
      extensions: { actorid: "user-789", correlationid: "cmd-abc" },
    },
  ],
  outboxTopics: ["orders"],
});

Options: Pass { client } to use an existing transaction (from withTransaction()).

Returns: AppendResult with streamId, fromVersion, toVersion, globalPositions[], and isDuplicate?: boolean.

Throws: OptimisticConcurrencyError, IdempotencyConflictError, CryptoKeyRevokedError, ReservedSnapshotEventTypeError.

Event types ending in Snapshot are reserved for Alvyn-generated snapshot events and cannot be appended through public write APIs.

load(streamId, options?): Promise<ReplayedEvent[]>

Loads all events for a stream from version 1. Handles decryption and upcasting automatically. Accepts an optional options object { maxEvents?: number; client?: PoolClient } (or LoadOptions) to limit the number of events or read over an active transaction client.

loadFrom(streamId, options): Promise<ReplayedEvent[]>

Loads events starting from a specific version. Accepts an optional client?: PoolClient to read within an active transaction.

const events = await eventStore.loadFrom("Order-123", {
  fromVersion: 4,
  maxEvents: 100,
});

lockStream(client, streamId): Promise<Date>

Acquires a transaction-scoped PostgreSQL advisory lock (pg_advisory_xact_lock) on the specified stream and returns the authoritative database timestamp (clock_timestamp()).

Must be called within an active transaction on the provided PoolClient. The lock is released automatically when the transaction commits or rolls back.

const decisionAt = await eventStore.lockStream(client, "Auction-123");

readEventsPage(options): Promise<ReadEventsPage<T>>

Reads a bounded page from an explicit allowlist of stream IDs, without loading complete streams or reading unrelated streams. Unknown, deleted, and empty stream IDs are ignored; an allowlist containing only those IDs returns an empty page.

const page = await eventStore.readEventsPage<OrderEvent>({
  streamIds: ["Order-123", "Order-456"],
  limit: 50,
  order: "asc",
});

if (page.hasNextPage && page.nextCursor !== null) {
  const next = await eventStore.readEventsPage({
    streamIds: ["Order-123", "Order-456"],
    limit: 50,
    cursor: page.nextCursor,
    order: "asc",
  });
}

limit must be a finite positive safe integer no greater than 1,000 (also exported as MAX_READ_EVENTS_PAGE_LIMIT). Pages are ordered by global_position, then stream_id, then event id; the latter fields are a deterministic tie-breaker for the cursor boundary. The cursor is opaque and must be reused with the same stream allowlist and order.

The first page captures an immutable PostgreSQL MVCC snapshot together with a high-water position. Both are encoded in every continuation cursor, and each page reapplies that snapshot using the transaction ID stored with every event. Transactions that were still in progress at the first request remain excluded even if they later commit below the high-water position; concurrent appends are also excluded. This guarantees stable repeated cursors with no skipped or duplicated events; start a new read to include later commits. Ascending and descending reads use the same rule.

Cursor positions and watermarks are encoded as decimal strings in the base64url cursor, so positions larger than JavaScript's safe integer range are not rounded. Returned events preserve the existing globalPosition: bigint metadata contract; use a JSON replacer or convert that field to a decimal string when serializing events because native JSON.stringify does not accept bigint values.

listStreams(options?): Promise<string[]>

Lists distinct stream IDs, optionally filtered by prefix.

const orderStreams = await eventStore.listStreams({
  prefix: "Order",
  limit: 50,
});

Subscriptions

subscribe(options?): AsyncIterable<StoredEvent>

Creates an independent fan-out subscription. It catches up on matching historical events and then tails live events in globalPosition order. Each subscriber receives its own copy of matching events; use the transactional outbox instead for competing-consumer delivery.

const controller = new AbortController();

for await (const event of eventStore.subscribe({
  subject: "Order-",
  recursive: true,
  eventTypes: ["OrderPlaced"],
  lowerBound: { id: "42", type: "exclusive" },
  signal: controller.signal,
})) {
  await handle(event);
}

Successful appends wake subscribers through transactional PostgreSQL LISTEN/NOTIFY, with polling as a fallback. Delivery is at least once, so persist a cursor after processing and make consumers idempotent. The read side uses a commit-safe watermark to avoid skipping events from late-committing transactions.

Snapshot Builder

defineSnapshot<TState, TEvents>()(definition)

Defines an event-backed snapshot over one source stream prefix.

const BankAccountBalance = defineSnapshot<
  { balance: number },
  TransactionEvents
>()({
  streamPrefix: "Transaction",
  snapshotName: "BankAccountBalance",
  every: 50,
  initialState: { balance: 0 },
  evolve: {
    Deposit: (state, event) => ({
      balance: state.balance + Number(event.data?.amount ?? 0),
    }),
    Withdrawal: (state, event) => ({
      balance: state.balance - Number(event.data?.amount ?? 0),
    }),
  },
});

snapshotName generates the snapshot event type ${snapshotName}Snapshot. Snapshot events are stored in the same stream and advance stream_version like any other event.

Register snapshot handles on the EventStore to maintain them on incoming events:

const eventStore = new EventStore({
  pool,
  snapshots: [BankAccountBalance],
});

snapshot.load(eventStore, entityId, options?): Promise<SnapshotLoadResult<TState>>

Finds the latest generated snapshot event and replays only later handled source events. Loading is read-only; registered snapshots are updated synchronously after matching public appends. Pass an optional { client } to read within an active transaction.

const result = await BankAccountBalance.load(eventStore, "account-123");
result.state.balance;

// Within an active transaction:
const txResult = await BankAccountBalance.load(eventStore, "account-123", {
  client,
});

Aggregate Builder

defineAggregate<TState, TEvents>()(definition): AggregateHandle<TState, TEvents>

Defines a typed aggregate. Empty streams load with state: null and version: 0; the first evolve handler receives the runtime null state.

const Order = defineAggregate<OrderState, OrderEvents>()({
  streamPrefix: "Order",
  evolve: {
    OrderPlaced: (state, event) => ({
      ...state,
      status: "placed",
      total: event.data?.total ?? 0,
    }),
  },
});

const order = await Order.load(eventStore, "order-123");
await Order.append(eventStore, {
  entityId: "order-123",
  expectedVersion: order.version,
  events: [{ type: "OrderPlaced", data: { total: 99.99 } }],
});

The handle also exposes typed loadEvents({ eventStore, entityId, maxEvents? }) and subscribe({ eventStore, entityId, options? }) methods. Aggregate event maps infer stored and replayed payloads, and generated snapshot events are filtered from those domain-facing methods.

Crypto / GDPR

createCryptoKey(keyId): Promise<void>

Creates a per-entity AES-256 encryption key. Idempotent.

Throws: CryptoSecretsRequiredError if no secrets or complete environment keyring was configured.

revokeKey(keyId): Promise<void>

Revokes a crypto key (GDPR erasure). Encrypted events become tombstones on read.

Throws: CryptoSecretsRequiredError, CryptoKeyNotFoundError.

See Crypto-Shredding for details.

Outbox & Maintenance

processOutbox(handler, limit?): Promise<number>

Claims and processes a batch of pending outbox entries in one transaction. The handler receives the entries and the transaction client; entries are marked processed only after the handler succeeds. Uses FOR UPDATE SKIP LOCKED for replica-safe concurrent processing.

cleanupOutbox(olderThanMs?, batchSize?): Promise<number>

Deletes processed outbox entries older than the supplied age (seven days by default) in bounded batches and returns the number deleted.

cleanupIdempotencyKeys(olderThanMs?, batchSize?): Promise<number>

Deletes recorded idempotency keys older than the supplied age (seven days by default) in bounded batches and returns the number deleted.

See Transactional Outbox for details.

Projections

runProjection(projection, batchSize?): Promise<number>

Processes the next batch of events for a projection (default batch: 500). Returns the count of events processed.

See Projections for details.

Upcasters

registerUpcaster(upcaster): void

Registers a single schema evolution transformer.

registerUpcasters(upcasters): void

Registers multiple upcasters at once.

See Schema Evolution for details.

Transactions

withTransaction(fn): Promise<T>

Executes a function within a PostgreSQL transaction.

await eventStore.withTransaction(async (client) => {
  await eventStore.append(
    {
      streamId: "Order-123",
      expectedVersion: 5,
      events: [{ type: "OrderPlaced", data: { total: 99.99 } }],
    },
    { client },
  );
  await client.query("INSERT INTO audit_log ...", [...]);
});

Type Definitions

Event Types

interface StoredEvent<T = unknown> {
  globalPosition: bigint;
  streamId: string;
  streamVersion: number;
  type: string;
  data: T;
  extensions: CloudEventExtensions;
  createdAt: Date;
}

interface TombstonedEvent {
  globalPosition: bigint;
  streamId: string;
  streamVersion: number;
  type: string;
  data: null;
  extensions: CloudEventExtensions;
  createdAt: Date;
  tombstoned: true;
}

interface CryptoSecret {
  version: number;
  value: string;
}

interface CryptoSecretsConfig {
  currentVersion: number;
  secrets: CryptoSecret[];
}

type ReplayedEvent<T = unknown> = StoredEvent<T> | TombstonedEvent;

Bounded Multi-Stream Read Types

type ReadEventsPageOrder = "asc" | "desc";

interface ReadEventsPageOptions {
  streamIds: readonly string[];
  limit: number;
  cursor?: string;
  order?: ReadEventsPageOrder;
}

interface ReadEventsPage<T = unknown> {
  events: ReplayedEvent<T>[];
  hasNextPage: boolean;
  nextCursor: string | null;
}

Stream Read Types

interface LoadOptions {
  maxEvents?: number;
  client?: PoolClient;
}

interface LoadFromOptions {
  fromVersion: number;
  maxEvents?: number;
  client?: PoolClient;
}

Append Types

interface AppendEventInput<T = unknown> {
  type: string;
  data: T;
  extensions?: Partial<CloudEventExtensions>;
  source?: string;
  encryptedFields?: string[];
  cryptoKeyId?: string;
  schemaVersion?: number;
}

interface AppendInput<T = unknown> {
  streamId: string;
  expectedVersion: number;
  events: AppendEventInput<T>[];
  outboxTopics?: string[];
  idempotencyKey?: string;
}

interface AppendOptions {
  client?: PoolClient;
}

interface AppendResult {
  streamId: string;
  fromVersion: number;
  toVersion: number;
  globalPositions: bigint[];
  isDuplicate?: boolean;
}

Upcaster Type

interface Upcaster<TIn = unknown, TOut = unknown> {
  eventType: string;
  fromSchemaVersion: number;
  toSchemaVersion: number;
  upcast(data: TIn): TOut;
}

Projection Types

interface Projection {
  projectionName: string;
  handle(event: StoredEvent, client: PoolClient): Promise<void>;
}

interface ProjectionHandlerContext {
  entityId: string;
  streamId: string;
  globalPosition: bigint;
  streamVersion: number;
  createdAt: Date;
  client: PoolClient;
}

Snapshot Types

interface SnapshotDefinition<TState, TEvents> {
  streamPrefix: string;
  snapshotName: string;
  every: number;
  initialState: TState;
  evolve: Partial<{
    [K in keyof TEvents & string]: (
      state: TState,
      event: ReplayedEvent<TEvents[K]>,
    ) => TState;
  }>;
  encryption?: {
    cryptoKeyId: (entityId: string) => string;
    encryptedFields: string[];
  };
}

interface SnapshotLoadResult<TState> {
  state: TState;
  streamId: string;
  version: number;
  snapshotVersion: number | null;
  replayedEvents: number;
}

interface SnapshotLoadOptions {
  client?: PoolClient;
}

Outbox Type

interface OutboxEntry {
  id: bigint;
  eventGlobalPosition: bigint;
  topic: string;
  payload: unknown;
  createdAt: Date;
}

Aggregate Types

interface AggregateLoadOptions {
  client?: PoolClient;
}

interface AggregateAppendOptions {
  client?: PoolClient;
}

interface AggregateAppendInput<TEvents> {
  entityId: string;
  expectedVersion: number;
  events: AggregateEventInput<TEvents>[];
  outboxTopics?: string[];
  idempotencyKey?: string;
}

interface AggregateLoadEventsOptions {
  eventStore: EventStore;
  entityId: string;
  maxEvents?: number;
  client?: PoolClient;
}

type AggregateEventInput<TEvents> = {
  [K in keyof TEvents & string]: {
    type: K;
    data: TEvents[K];
    extensions?: Partial<CloudEventExtensions>;
    schemaVersion?: number;
  };
}[keyof TEvents & string];

interface AggregateInstance<TState> {
  state: TState | null;
  version: number;
  streamId: string;
}

type AggregateStoredEvent<TEvents> = {
  [K in keyof TEvents & string]: StoredEvent<TEvents[K]> & { type: K };
}[keyof TEvents & string];

type AggregateReplayedEvent<TEvents> =
  | AggregateStoredEvent<TEvents>
  | TombstonedEvent;

interface SubscribeOptions {
  subject?: string;
  recursive?: boolean;
  eventTypes?: string[];
  lowerBound?: { id: string; type?: "exclusive" | "inclusive" };
  signal?: AbortSignal;
  batchSize?: number;
  pollIntervalMs?: number;
}

Error Reference

All errors extend Error and have a name property matching the class name for instanceof checks.

OptimisticConcurrencyError

Thrown when expectedVersion does not match the stream's current version.

IdempotencyConflictError

Thrown when an idempotencyKey is reused with a conflicting stream ID or with a different event payload.

StreamNotFoundError

Thrown when loading a stream that does not exist and the caller explicitly required existence.

CryptoKeyRevokedError

Thrown when attempting to encrypt new events with a revoked key. Not thrown during reads.

CryptoKeyNotFoundError

Thrown when a crypto key is not found in the key store.

CryptoKeyIdRequiredError

Thrown when encryptedFields are configured without a non-empty cryptoKeyId.

CryptoSecretsRequiredError

Thrown when crypto operations are attempted but no complete versioned crypto keyring was provided, or when crypto environment configuration is missing its current version.

InvalidCryptoSecretsError

Thrown when configured crypto secrets are empty, malformed, duplicated, or use an invalid version.

CryptoSecretVersionNotFoundError

Thrown when an encrypted entity-key envelope references a secret version that is not configured.

EventStoreNotInitializedError

Thrown when any method is called before setup().

InvalidSchemaNameError

Thrown when the schema name does not match /^[a-z_][a-z0-9_]{0,62}$/.

ReservedSnapshotEventTypeError

Thrown when user code attempts to append an event type ending in Snapshot. The suffix is reserved for Alvyn-generated snapshot events.