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. Only the authenticated, versioned envelope format is supported; records from earlier pre-release implementations must be recreated.

Lifecycle

setup(): Promise<void>

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

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

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[].

Throws: OptimisticConcurrencyError, CryptoKeyRevokedError, ReservedSnapshotEventTypeError.

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

load(streamId): Promise<ReplayedEvent[]>

Loads all events for a stream from version 1. Handles decryption and upcasting automatically.

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

Loads events starting from a specific version.

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

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): 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.

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

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

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.

See Projections & 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 & Outbox 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;

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[];
}

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

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

Outbox Type

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

Aggregate Types

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.

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.