# Aggregates (/docs/aggregates)



The `defineAggregate` function is the **primary DX surface** for working with event-sourced aggregates. It produces a type-safe `AggregateHandle` that encapsulates stream ID derivation, state replay, and encryption configuration.

## Defining an Aggregate [#defining-an-aggregate]

### Step 1: Define the Event Map [#step-1-define-the-event-map]

The event map is a TypeScript type that maps event type names to their payload shapes:

```typescript
type OrderEvents = {
  OrderPlaced: { customerId: string; total: number };
  OrderShipped: { trackingNumber: string };
  OrderCancelled: { reason: string };
};
```

### Step 2: Define the Aggregate [#step-2-define-the-aggregate]

`defineAggregate` uses a **curried function pattern** for full TypeScript inference:

* `TState` is provided explicitly (the aggregate state)
* `TEvents` is provided explicitly (the event map)

```typescript
import { defineAggregate } from "alvyn";

type OrderState = {
  status: "pending" | "shipped" | "cancelled";
  total: number;
  customerId: string;
};

const Order = defineAggregate<OrderState, OrderEvents>()({
  streamPrefix: "Order", // stream_id = "Order-{entityId}"
  evolve: {
    OrderPlaced: (state, event) => ({
      ...state,
      status: "pending",
      total: event.data?.total ?? 0,
      customerId: event.data?.customerId ?? "",
    }),
    OrderShipped: (state) => ({
      ...state,
      status: "shipped",
    }),
    OrderCancelled: (state) => ({
      ...state,
      status: "cancelled",
    }),
  },
});
```

## AggregateDefinition Interface [#aggregatedefinition-interface]

```typescript
interface AggregateDefinition<TEvents, TState> {
  streamPrefix: string;
  evolve: {
    [K in keyof TEvents]: (
      state: TState,
      event: ReplayedEvent<TEvents[K]>,
    ) => TState;
  };
  encryption?: EncryptionConfig<TEvents>;
  upcasters?: Upcaster[];
}
```

### `streamPrefix` [#streamprefix]

Determines how stream IDs are derived: `"{streamPrefix}-{entityId}"`. For example, `streamPrefix: "Order"` with `entityId: "123"` produces `stream_id = "Order-123"`.

### `evolve` [#evolve]

A map of event type names to handler functions. Each handler receives the current state and the event, and returns the new state using an **immutable update pattern**.

Events without a handler in `evolve` are silently skipped. This is not a semantic forward-compatibility guarantee: old readers may compute incomplete state or miss invariants. Deploy compatible readers before enabling new event types or payload versions.

### `encryption` (optional) [#encryption-optional]

See [Crypto-Shredding](/docs/crypto-shredding) for details.

```typescript
encryption: {
  cryptoKeyId: (entityId) => `user:${entityId}`,
  encryptedFields: {
    UserRegistered: ["name", "email", "address.street"],
    UserRenamed: ["name"],
  },
}
```

### `upcasters` (optional) [#upcasters-optional]

See [Schema Evolution](/docs/schema-evolution) for details.

```typescript
upcasters: [userRenamedV1ToV2, userRenamedV2ToV3];
```

## AggregateHandle (Returned Object) [#aggregatehandle-returned-object]

The `defineAggregate` call returns an `AggregateHandle` with the following interface:

| Method / Property                                           | Description                                   |
| ----------------------------------------------------------- | --------------------------------------------- |
| `streamPrefix`                                              | The stream prefix string (readonly)           |
| `load(eventStore, entityId, options?)`                      | Returns `AggregateInstance<TState>`           |
| `loadEvents({ eventStore, entityId, maxEvents?, client? })` | Loads typed replayed domain events            |
| `append(eventStore, input, options?)`                       | Appends typed events to the stream            |
| `subscribe({ eventStore, entityId, options? })`             | Subscribes to typed live domain events        |
| `getUpcasters()`                                            | Returns upcasters for registration at startup |

### `load(eventStore, entityId, options?)` [#loadeventstore-entityid-options]

Loads the aggregate by replaying all events. Pass an optional `{ client }` to read within an existing PostgreSQL transaction:

```typescript
const order = await Order.load(eventStore, "order-123");

// Or within an active transaction:
const orderInTx = await Order.load(eventStore, "order-123", { client });

order.state; // { status: "shipped", total: 99.99, customerId: "cust-1" }
order.version; // 5 (current stream version)
order.streamId; // "Order-order-123"
```

For an empty stream, `state` is `null` and `version` is `0`. The first evolve
handler receives the runtime `null` state, so handlers should build the state
from the first event when necessary.

### `append(eventStore, input, options?)` [#appendeventstore-input-options]

Appends typed events to the aggregate's stream. Automatically maps encryption config if defined. Pass an optional `{ client }` to execute within an active transaction:

```typescript
await Order.append(eventStore, {
  entityId: "order-123",
  expectedVersion: order.version, // OCC: must match current version
  events: [{ type: "OrderShipped", data: { trackingNumber: "TRACK-456" } }],
  outboxTopics: ["orders"], // Optional: transactional outbox
  idempotencyKey: "cmd-ship-order-123", // Optional: deduplication key
});

// Or within an active transaction:
await Order.append(
  eventStore,
  {
    entityId: "order-123",
    expectedVersion: order.version,
    events: [{ type: "OrderShipped", data: { trackingNumber: "TRACK-456" } }],
  },
  { client },
);
```

### `getUpcasters()` [#getupcasters]

Returns all upcasters defined in the aggregate definition. Call this during startup to register them with the event store:

```typescript
eventStore.registerUpcasters(Order.getUpcasters());
```

## OCC Retry Pattern [#occ-retry-pattern]

When multiple writers may target the same stream, use a retry loop:

```typescript
import { OptimisticConcurrencyError } from "alvyn";

async function shipOrder(orderId: string, trackingNumber: string) {
  const maxRetries = 3;

  for (let attempt = 0; attempt < maxRetries; attempt++) {
    const order = await Order.load(eventStore, orderId);

    if (!order.state || order.state.status !== "pending") {
      throw new BadRequestError("Order cannot be shipped");
    }

    try {
      await Order.append(eventStore, {
        entityId: orderId,
        expectedVersion: order.version,
        events: [{ type: "OrderShipped", data: { trackingNumber } }],
      });
      return;
    } catch (error) {
      if (
        error instanceof OptimisticConcurrencyError &&
        attempt < maxRetries - 1
      ) {
        continue;
      }
      throw error;
    }
  }
}
```

## Idempotent Appends [#idempotent-appends]

When commands originate from network clients (e.g. HTTP POST requests or message retries), supply an optional `idempotencyKey` to ensure safe, deduplicated writes:

```typescript
await Order.append(eventStore, {
  entityId: orderId,
  expectedVersion: order.version,
  events: [{ type: "OrderShipped", data: { trackingNumber } }],
  idempotencyKey: `ship-${orderId}-${trackingNumber}`,
});
```

When an append with a known `idempotencyKey` is retried:

* No duplicate events are written to the database.
* No duplicate outbox entries or notifications are produced.
* Snapshot updates are automatically skipped.
* The original version boundaries are returned transparently.

If a key is reused with a different stream ID or different event payload, Alvyn throws `IdempotencyConflictError`.

## Handling Tombstoned Events [#handling-tombstoned-events]

When a crypto key is revoked, only encrypted fields become `null`; public fields remain in the original stored schema. Evolve handlers must check `tombstoned` and not rely on a complete, upcasted payload:

```typescript
evolve: {
  UserRegistered: (state, event) => ({
    ...state,
    name: "tombstoned" in event ? "" : event.data?.name ?? "",
    email: "tombstoned" in event ? "" : event.data?.email ?? "",
  }),
}
```

The first handler still receives `null` as its state, including when the creation event is redacted. Supply non-sensitive defaults for erased fields; never assume a previous state exists. Erasure of previously materialized copies requires separate cleanup.

## Explicit Transactions & Stream Locking [#explicit-transactions--stream-locking]

In distributed setups (such as Kubernetes replica sets), local clock skew and retry races can compromise strict temporal invariants (e.g. auction closing deadlines, soft-close anti-sniping, or reservation expiry).

To serialize command handling early and anchor business logic to authoritative database time, use `eventStore.lockStream(client, streamId)` inside an explicit transaction:

```typescript
await eventStore.withTransaction(async (client) => {
  // 1. Acquire transaction-scoped advisory lock & authoritative PostgreSQL clock_timestamp()
  const streamId = `Auction-${auctionId}`;
  const decisionAt = await eventStore.lockStream(client, streamId);

  // 2. Load the aggregate over the same locked connection
  const auction = await Auction.load(eventStore, auctionId, { client });

  // 3. Evaluate invariants deterministically
  if (decisionAt.getTime() >= new Date(auction.state.endsAt).getTime()) {
    throw new Error("Auction already closed");
  }

  // 4. Append events atomically within the transaction
  await Auction.append(
    eventStore,
    {
      entityId: auctionId,
      expectedVersion: auction.version,
      events: [
        {
          type: "BidAccepted",
          data: { amount: 200, placedAt: decisionAt.toISOString() },
        },
      ],
    },
    { client },
  );
});
```

Because `lockStream` and `Auction.append` use the exact same transaction-scoped PostgreSQL advisory lock (`pg_advisory_xact_lock`), acquiring the lock early forms an orderly queue at the database level and eliminates optimistic-concurrency retry storms under heavy contention.

## Full Example with All Options [#full-example-with-all-options]

```typescript
import { defineAggregate, type Upcaster } from "alvyn";

type UserEvents = {
  UserRegistered: {
    name: string;
    email: string;
    address: { street: string; city: string };
  };
  UserRenamed: { name: string; updatedAt: string | null };
  UserDeactivated: { reason: string };
};

type UserState = {
  name: string;
  email: string;
  address: { street: string; city: string };
  active: boolean;
};

const userRenamedV1ToV2: Upcaster = {
  eventType: "UserRenamed",
  fromSchemaVersion: 1,
  toSchemaVersion: 2,
  upcast(data: { name: string }) {
    return { name: data.name, updatedAt: null };
  },
};

const User = defineAggregate<UserState, UserEvents>()({
  streamPrefix: "User",
  evolve: {
    UserRegistered: (state, event) => ({
      ...state,
      name: event.data?.name ?? state?.name ?? "",
      email: event.data?.email ?? state?.email ?? "",
      address: event.data?.address ??
        state?.address ?? { street: "", city: "" },
      active: true,
    }),
    UserRenamed: (state, event) => ({
      ...state,
      name: event.data?.name ?? state?.name ?? "",
    }),
    UserDeactivated: (state) => ({
      ...state,
      active: false,
    }),
  },

  encryption: {
    cryptoKeyId: (entityId) => `user:${entityId}`,
    encryptedFields: {
      UserRegistered: ["name", "email", "address.street"],
      UserRenamed: ["name"],
    },
  },

  upcasters: [userRenamedV1ToV2],
});

// At startup: register upcasters
eventStore.registerUpcasters(User.getUpcasters());
```


# API Reference (/docs/api-reference)



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

## EventStore Class [#eventstore-class]

### Constructor [#constructor]

```typescript
import { EventStore } from "alvyn";

new EventStore(config: EventStoreConfig)
```

```typescript
interface EventStoreConfig {
  pool: Pool;
  schema?: string;
  migrationMode?: "migrate" | "verify";
  secrets?: CryptoSecretsConfig;
  defaultSource?: string;
  snapshots?: SnapshotHandle<unknown>[];
}
```

| Parameter       | Type                        | Required | Description                                                          |
| --------------- | --------------------------- | -------- | -------------------------------------------------------------------- |
| `pool`          | `Pool`                      | Yes      | PostgreSQL connection pool (caller manages lifecycle)                |
| `schema`        | `string`                    | No       | PostgreSQL schema name (default: `"event_store"`)                    |
| `migrationMode` | `"migrate" \| "verify"`     | No       | `setup()` runs migrations by default; `verify` performs no DDL       |
| `secrets`       | `CryptoSecretsConfig`       | No       | Versioned keyring and explicit version used for new encryption       |
| `defaultSource` | `string`                    | No       | CloudEvents `source` URI-reference applied to all events             |
| `snapshots`     | `SnapshotHandle<unknown>[]` | No       | Snapshot definitions maintained synchronously after matching appends |

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:

```typescript
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](/docs/crypto-shredding#rotating-secrets-without-downtime) 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 [#lifecycle]

#### `setup(): Promise<void>` [#setup-promisevoid]

By default, applies versioned schema migrations transactionally under a per-schema advisory lock, then verifies the schema. With `migrationMode: "verify"`, performs only read-only version and trigger checks (no DDL); use this after a deployment job has called `migrateEventStore`. &#x2A;*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. Concurrent migration calls across replicas serialize at the database lock.

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

#### `migrateEventStore({ pool, schema? }): Promise<void>` [#migrateeventstore-pool-schema--promisevoid]

Public standalone entry point for a deployment job. Uses the **same versioned migration routine** as `EventStore.setup()` in default mode. `schema` defaults to `"event_store"`. The supplied pool must have DDL privileges and connect to the PostgreSQL primary. See [Database Schema](/docs/database-schema#migrations) for a deployment example.

#### `isInitialized(): boolean` [#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>` [#getstreamversionstreamid-promisenumber]

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

### Stream Operations [#stream-operations]

#### `append(input, options?): Promise<AppendResult>` [#appendinput-options-promiseappendresult]

Appends events to a stream within an ACID transaction.

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

| Field             | Type                 | Description                                            |
| ----------------- | -------------------- | ------------------------------------------------------ |
| `streamId`        | `string`             | Target stream identifier                               |
| `expectedVersion` | `number`             | `-1` = new stream, `0` = no check, `N` = exact version |
| `events`          | `AppendEventInput[]` | Events to append                                       |
| `outboxTopics`    | `string[]`           | Optional: topics for transactional outbox              |
| `idempotencyKey`  | `string`             | Optional: deduplication key for safe retries           |

**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[]>` [#loadstreamid-options-promisereplayedevent]

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[]>` [#loadfromstreamid-options-promisereplayedevent]

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

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

#### `lockStream(client, streamId): Promise<Date>` [#lockstreamclient-streamid-promisedate]

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.

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

#### `readEventsPage(options): Promise<ReadEventsPage<T>>` [#readeventspageoptions-promisereadeventspaget]

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.

```typescript
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[]>` [#liststreamsoptions-promisestring]

Lists distinct stream IDs, optionally filtered by prefix.

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

| Field    | Type     | Default | Description                                                  |
| -------- | -------- | ------- | ------------------------------------------------------------ |
| `prefix` | `string` | -       | Stream ID prefix (separator `"-"` is appended automatically) |
| `limit`  | `number` | `100`   | Maximum number of stream IDs to return                       |

### Subscriptions [#subscriptions]

#### `subscribe(options?): AsyncIterable<SubscriptionEvent>` [#subscribeoptions-asynciterablesubscriptionevent]

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.

```typescript
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 [#snapshot-builder]

#### `defineSnapshot<TState, TEvents>()(definition)` [#definesnapshottstate-teventsdefinition]

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

```typescript
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:

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

#### `snapshot.load(eventStore, entityId, options?): Promise<SnapshotLoadResult<TState>>` [#snapshotloadeventstore-entityid-options-promisesnapshotloadresulttstate]

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.

```typescript
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 [#aggregate-builder]

#### `defineAggregate<TState, TEvents>()(definition): AggregateHandle<TState, TEvents>` [#defineaggregatetstate-teventsdefinition-aggregatehandletstate-tevents]

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

```typescript
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 [#crypto--gdpr]

#### `createCryptoKey(keyId): Promise<void>` [#createcryptokeykeyid-promisevoid]

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

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

#### `revokeKey(keyId): Promise<void>` [#revokekeykeyid-promisevoid]

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

**Throws:** `CryptoSecretsRequiredError`, `CryptoKeyNotFoundError`.

See [Crypto-Shredding](/docs/crypto-shredding) for details.

### Outbox & Maintenance [#outbox--maintenance]

#### `processOutbox(handler, limit?): Promise<number>` [#processoutboxhandler-limit-promisenumber]

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.

External effects are at-least-once: a successful publish can repeat if the database transaction subsequently fails. Only writes using the supplied `client` are atomic with marking entries processed.

#### `cleanupOutbox(olderThanMs?, batchSize?): Promise<number>` [#cleanupoutboxolderthanms-batchsize-promisenumber]

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>` [#cleanupidempotencykeysolderthanms-batchsize-promisenumber]

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

See [Transactional Outbox](/docs/outbox) for details.

### Projections [#projections]

#### `runProjection(projection, batchSize?): Promise<number>` [#runprojectionprojection-batchsize-promisenumber]

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

See [Projections](/docs/projections) for details.

### Upcasters [#upcasters]

#### `registerUpcaster(upcaster): void` [#registerupcasterupcaster-void]

Registers a single schema evolution transformer.

#### `registerUpcasters(upcasters): void` [#registerupcastersupcasters-void]

Registers multiple upcasters at once.

See [Schema Evolution](/docs/schema-evolution) for details.

### Transactions [#transactions]

#### `withTransaction(fn): Promise<T>` [#withtransactionfn-promiset]

Executes a function within a PostgreSQL transaction. Multiple low-level `append(input, { client })` calls can atomically update different streams when they share this client. Aggregate `append` handles do not expose a transaction-client option; use the low-level API and provide encryption settings explicitly where needed.

```typescript
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 [#type-definitions]

### Event Types [#event-types]

```typescript
interface StoredEvent<T = unknown>
  extends CloudEventRequiredAttributes, CloudEventOptionalAttributes {
  globalPosition: bigint;
  streamId: string;
  streamVersion: number;
  type: string;
  data: T;
  extensions: CloudEventExtensions;
  createdAt: Date;
}

interface TombstonedEvent
  extends CloudEventRequiredAttributes, CloudEventOptionalAttributes {
  globalPosition: bigint;
  streamId: string;
  streamVersion: number;
  type: string;
  data: unknown; // Public fields retained; encrypted fields set to null
  redactedPaths: readonly string[];
  schemaVersion: number;
  extensions: CloudEventExtensions;
  createdAt: Date;
  tombstoned: true;
}

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

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

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

Both event types inherit these CloudEvents context fields (in addition to the internal stream metadata above):

```typescript
interface CloudEventRequiredAttributes {
  id: string;
  source: string;
  specversion: string;
  type: string;
}

interface CloudEventOptionalAttributes {
  datacontenttype?: string;
  dataschema?: string;
  subject?: string;
  time?: string;
}

interface CloudEventExtensions {
  correlationid?: string;
  causationid?: string;
  actorid?: string;
  schemaversion?: number;
  [key: string]: unknown;
}
```

`time` is an RFC 3339 string, while `createdAt` is a `Date`. Serialize `globalPosition` as a decimal string for JSON. Subscriptions upcast complete unencrypted events by default without decrypting PII; encrypted events are returned as `RedactedSubscriptionEvent` in the stored schema with encrypted fields set to `null`. Pass `raw: true` to skip upcasting complete events. Projections decrypt available PII and then upcast before invoking typed handlers. Events with revoked keys instead use an explicit `onRedacted` handler or `onRedacted: "skip"`; their remaining public data stays in the original schema. Without a policy, the projection fails rather than silently losing an event.

### Bounded Multi-Stream Read Types [#bounded-multi-stream-read-types]

```typescript
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 [#stream-read-types]

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

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

### Append Types [#append-types]

```typescript
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 [#upcaster-type]

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

### Projection Types [#projection-types]

```typescript
interface Projection {
  projectionName: string;
  handle(event: StoredEvent, client: PoolClient): Promise<void>;
  handlesEventType?: (type: string) => boolean; // Filter before decrypting
  onRedacted?:
    | "skip"
    | ((
        event: RedactedProjectionEvent,
        client: PoolClient,
      ) => void | Promise<void>);
}

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

### Snapshot Types [#snapshot-types]

```typescript
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 [#outbox-type]

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

### Aggregate Types [#aggregate-types]

```typescript
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> = {
  [K in keyof TEvents & string]: ReplayedEvent<TEvents[K]> & { type: K };
}[keyof TEvents & string];

type RedactedSubscriptionEvent = Omit<StoredEvent, "data"> & {
  redacted: true;
  data: unknown; // Public fields in the stored schema; encrypted fields null
  redactedPaths: readonly string[];
  schemaVersion: number;
};

type SubscriptionEvent = StoredEvent | RedactedSubscriptionEvent;

type AggregateSubscriptionEvent<TEvents> =
  | AggregateStoredEvent<TEvents>
  | RedactedSubscriptionEvent;

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

## Error Reference [#error-reference]

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

### `OptimisticConcurrencyError` [#optimisticconcurrencyerror]

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

| Property          | Type     | Description                 |
| ----------------- | -------- | --------------------------- |
| `streamId`        | `string` | The conflicting stream      |
| `expectedVersion` | `number` | What the caller expected    |
| `actualVersion`   | `number` | The stream's actual version |

### `IdempotencyConflictError` [#idempotencyconflicterror]

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

| Property         | Type     | Description                     |
| ---------------- | -------- | ------------------------------- |
| `idempotencyKey` | `string` | The conflicting idempotency key |

### `StreamNotFoundError` [#streamnotfounderror]

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

### `CryptoKeyRevokedError` [#cryptokeyrevokederror]

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

### `CryptoKeyNotFoundError` [#cryptokeynotfounderror]

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

### `CryptoKeyIdRequiredError` [#cryptokeyidrequirederror]

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

### `CryptoSecretsRequiredError` [#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` [#invalidcryptosecretserror]

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

### `CryptoSecretVersionNotFoundError` [#cryptosecretversionnotfounderror]

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

### `EventStoreNotInitializedError` [#eventstorenotinitializederror]

Thrown when any method is called before `setup()`.

### `InvalidSchemaNameError` [#invalidschemanameerror]

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

### `ReservedSnapshotEventTypeError` [#reservedsnapshoteventtypeerror]

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


# GDPR & Data Privacy (/docs/crypto-shredding)



Event sourcing architectures are built on an immutable, append-only log: once written, events record historical facts that should never be modified or deleted. However, &#x2A;*Article 17 of the General Data Protection Regulation (GDPR)** grants individuals the "Right to Erasure" (the right to be forgotten).

Naively deleting rows from an event store breaks stream versioning, destroys causal integrity, and invalidates deterministic state replay. To satisfy regulatory erasure requirements without compromising the integrity of your event store, two production-proven patterns are available:

1. **Pattern A: Reference-Based PII (Keep PII Outside the Event Store)** — Store personal data in standard mutable PostgreSQL tables and reference only opaque, non-semantic identifiers inside event payloads.
2. **Pattern B: In-Stream Envelope Encryption (Crypto-Shredding)** — Encrypt personal fields inside event payloads using per-entity encryption keys. Revoking a key renders the ciphertext permanently unreadable (`tombstoned`).

## Architectural Decision Matrix [#architectural-decision-matrix]

| Dimension                     | Pattern A: Reference Tables                            | Pattern B: Alvyn Crypto-Shredding                            |
| :---------------------------- | :----------------------------------------------------- | :----------------------------------------------------------- |
| **Primary Mechanism**         | Store PII in mutable tables, reference via opaque ID   | Store PII encrypted in event payloads (AES-256-GCM)          |
| **Erasure Method**            | Standard SQL `DELETE FROM user_profiles WHERE id = $1` | Revoke entity key via `eventStore.revokeKey(keyId)`          |
| **Implementation Complexity** | Minimal (familiar CRUD + foreign keys)                 | Moderate (keyring setup, tombstone-aware evolve handlers)    |
| **Event Autonomy**            | Partial (requires lookup to resolve current PII)       | Full (events are self-contained while key is active)         |
| **Point-in-Time History**     | Lost upon deletion (or requires audit versioning)      | Preserved until key revocation, then tombstoned              |
| **SQL Indexability**          | Full (B-Tree indexes, `ILIKE`, full-text search)       | None on encrypted fields (`encrypted_data` JSONB)            |
| **Erasure Scope**             | $O(1)$ in local relational table                       | $O(1)$ across all streams and events in the system           |
| **Legal Classification**      | Definitive physical deletion                           | Cryptographic erasure (ciphertext remains in storage)        |
| **When to Choose**            | Standard applications with centralized user profiles   | Multi-stream architectures, distributed agents, audit trails |

## Pattern A: Reference-Based PII (Keep PII Outside the Event Store) [#pattern-a-reference-based-pii-keep-pii-outside-the-event-store]

The simplest and most resilient way to handle GDPR Article 17 is to **keep personal data entirely out of the event log**.

Instead of embedding names, email addresses, or physical addresses in event payloads, store PII in a standard mutable PostgreSQL table (`user_profiles`). The immutable events contain only an opaque, non-semantic identifier (such as `userId: "usr_123"`).

### 1. PostgreSQL Schema Design [#1-postgresql-schema-design]

Define a standard relational table for personal data. Because this table is separate from the event log, it supports standard relational features including uniqueness constraints, foreign keys, and indexes:

```sql
CREATE TABLE user_profiles (
  user_id TEXT PRIMARY KEY,
  email TEXT NOT NULL,
  full_name TEXT NOT NULL,
  phone TEXT,
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- Plaintext indexes for fast lookup and uniqueness
CREATE UNIQUE INDEX idx_user_profiles_email ON user_profiles (lower(email));
```

The event payload contains only non-personal business facts alongside the opaque identifier:

```json
{
  "userId": "usr_123",
  "orderId": "ord_987",
  "totalAmount": 14900,
  "currency": "EUR"
}
```

### 2. Atomic Writes with `eventStore.withTransaction` [#2-atomic-writes-with-eventstorewithtransaction]

To prevent inconsistencies where a profile is updated but the corresponding event fails to append (or vice-versa), execute both operations inside a single PostgreSQL transaction using `eventStore.withTransaction`:

```typescript
import { eventStore } from "./event-store";

await eventStore.withTransaction(async (client) => {
  // 1. Upsert mutable profile with PII
  await client.query(
    `INSERT INTO user_profiles (user_id, email, full_name, updated_at)
     VALUES ($1, $2, $3, NOW())
     ON CONFLICT (user_id) DO UPDATE
     SET email = EXCLUDED.email, full_name = EXCLUDED.full_name, updated_at = NOW()`,
    [userId, email, fullName],
  );

  // 2. Append event referencing only the opaque userId
  await eventStore.append(
    {
      streamId: `User-${userId}`,
      expectedVersion: 0,
      events: [
        {
          type: "UserRegistered",
          data: { registeredAt: new Date().toISOString() },
        },
      ],
    },
    { client }, // Atomic: commits or rolls back together with the profile upsert
  );
});
```

Because both writes share the same PostgreSQL client and transaction, your system guarantees atomic consistency with zero distributed transaction overhead.

### 3. Deletion Flow (Physical Deletion) [#3-deletion-flow-physical-deletion]

When a user exercises their GDPR Article 17 Right to Erasure, delete their row from `user_profiles` using standard SQL:

```typescript
await pool.query("DELETE FROM user_profiles WHERE user_id = $1", [userId]);
```

The event stream (`User-usr_123` or orders linked to `usr_123`) remains completely untouched in the append-only log. The event history continues to record the business fact that a registration or purchase occurred, but the identifier `usr_123` is now an orphaned pseudonym that can no longer be resolved to an identifiable natural person.

### 4. Handling Orphaned References During Replay [#4-handling-orphaned-references-during-replay]

When read models or downstream projections join events with `user_profiles`, they must handle missing profile records gracefully:

```typescript
async function getUserSummary(userId: string) {
  // Replay immutable business events
  const events = await eventStore.load(`User-${userId}`);

  // Resolve current profile (may be null if deleted)
  const { rows } = await pool.query(
    "SELECT email, full_name FROM user_profiles WHERE user_id = $1",
    [userId],
  );
  const profile = rows[0] ?? null;

  return {
    userId,
    isDeleted: !profile,
    email: profile?.email ?? "[DELETED]",
    fullName: profile?.full_name ?? "Anonymized User",
    registrationEvent: events.find((e) => e.type === "UserRegistered"),
  };
}
```

### 5. Architectural Trade-offs [#5-architectural-trade-offs]

**Advantages:**

* **Zero Cryptographic Overhead:** No root keys, keyrings, encryption envelopes, or performance penalty for encryption/decryption.
* **Full SQL Indexability:** Search, filter, and sort profiles by email or name using standard B-Tree or full-text indexes (`pg_trgm`).
* **Unambiguous Legal Erasure:** Physical deletion via `DELETE` is universally accepted by data protection authorities as true erasure under GDPR Article 17.
* **Simplicity (KISS):** Regular PostgreSQL CRUD logic without special evolve-handler fallbacks or tombstone types.

**Limitations:**

* **Loss of Historical Snapshot:** The profile table holds only the *current* state. You cannot see what a user's address was at the exact moment a past order occurred (unless you maintain a separate audited version history).
* **Reduced Event Autonomy:** Events are not self-contained. Any consumer or projection that requires personal details must query the relational profile table.
* **Identifier Hygiene:** An ID is not anonymous simply because it is called `id`. It must be an opaque, non-semantic UUID or random token that cannot be reversed or correlated with the person through external systems.

## Pattern B: In-Stream Envelope Encryption (Crypto-Shredding) [#pattern-b-in-stream-envelope-encryption-crypto-shredding]

When downstream consumers and projections require self-contained event payloads, or when personal data is distributed across multiple distinct streams, &#x2A;*per-entity envelope encryption (crypto-shredding)** enables instant, surgical data revocation.

Events store personal data encrypted under a unique, entity-specific AES-256-GCM key. Revoking that single key renders all historical ciphertext for that entity unreadable in $O(1)$ time, without modifying any append-only event rows.

### Two-Tier Envelope Encryption Architecture [#two-tier-envelope-encryption-architecture]

Alvyn implements a two-tier key hierarchy:

1. **Versioned Root Secrets (Key Encryption Key):** A configured server secret (`currentVersion`) wraps and unwraps per-entity keys using `scrypt` and AES-256-GCM. Rotating root secrets requires zero database re-encryption.
2. **Per-Entity AES Keys (Data Encryption Key):** Each entity (e.g. user, tenant) gets a unique 256-bit AES key stored encrypted in the `crypto_keys` table.
3. **Field-Level Authenticated Encryption:** Individual PII fields (e.g. `email`, `address.street`) are encrypted with the entity key, each with a unique random IV and GCM authentication tag.

AES-GCM also authenticates the meaning of each encrypted value. Entity-key envelopes are bound to their key ID, magic marker, and secret version. PII fields are bound to their event ID, entity-key ID, field path, and stored version as Additional Authenticated Data (AAD). Moving a valid ciphertext to another row, event, or field causes decryption to fail.

### How It Works (Write & Read Paths) [#how-it-works-write--read-paths]

#### Write Path (append) [#write-path-append]

1. For each event with `encryptedFields` + `cryptoKeyId`, the entity's AES key is retrieved and decrypted from the database using the configured secret version.
2. Each PII field (specified as dot-paths, e.g. `"address.street"`) is:
   * Extracted from the event data
   * JSON-stringified and encrypted with AES-256-GCM (random IV per field)
   * Stored in the `encrypted_data` column as `{ version, ciphertext, iv, authTag }` (base64-encoded)
3. The `data` column stores the event payload **with PII fields removed**.

**Example of what gets stored in the database:**

```
data column:            { "loginCount": 5 }
encrypted_data column:  { "name": { "version": 2, "ciphertext": "...", "iv": "...", "authTag": "..." },
                          "email": { "version": 2, "ciphertext": "...", "iv": "...", "authTag": "..." } }
crypto_key_id column:   "user:abc123"
```

#### Read Path (load) [#read-path-load]

1. If the entity's crypto key is **active**: fields are decrypted and merged back into the event data. The caller sees the complete event as if encryption was transparent.
2. If the key is **revoked** and fields were actually encrypted: the event is returned as a `TombstonedEvent` with non-encrypted fields preserved and encrypted field paths set to `null`. The payload is in its original schema and is not upcasted. A configured encrypted field that was absent from an event is not encrypted and does not cause a tombstone later. Missing key rows or missing crypto configuration are errors, not intentional erasure.

### Setting Up Encryption [#setting-up-encryption]

#### 1. Configure Versioned Secrets [#1-configure-versioned-secrets]

Generate secret values for the versions you will use:

```bash
openssl rand -base64 32
```

Configure the complete keyring and explicitly select the version used for new encryption. The order of entries is not significant:

```typescript
const eventStore = new EventStore({
  pool,
  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 the environment:

```bash
GDPR_CRYPTO_SECRETS=2:new-secret-base64,1:old-secret-base64
GDPR_CRYPTO_CURRENT_VERSION=2
```

Secret values are strings. All values are strengthened with `scrypt` to derive the 32-byte key used to protect entity keys. Use a high-entropy generated value; `scrypt` makes guessing more expensive but cannot add entropy to a weak secret. Keep every old secret needed to read existing versioned envelopes until that data has been intentionally retired; removing a configured version makes those envelopes unreadable.

Secret versions are unsigned 32-bit integers from `0` to `4294967295`. Gaps are allowed, and `currentVersion` must identify one of the configured secrets. Use a higher version for each new rotation, but do not encode the active version by reordering the keyring. Rotating between secret versions does not require a database migration.

#### 2. Create Entity Keys [#2-create-entity-keys]

Create a per-entity key before appending encrypted events:

```typescript
await eventStore.createCryptoKey("user:abc123");
```

This is **idempotent** — if the key already exists, it is a no-op.

#### 3. Configure Aggregate Encryption [#3-configure-aggregate-encryption]

In the aggregate definition, specify which fields contain PII per event type:

```typescript
type UserState = {
  name: string;
  email: string;
  address: { street: string; city: string };
};

const User = defineAggregate<UserState, UserEvents>()({
  streamPrefix: "User",
  // ... evolve ...

  encryption: {
    cryptoKeyId: (entityId) => `user:${entityId}`,
    encryptedFields: {
      UserRegistered: ["name", "email", "address.street"],
      UserRenamed: ["name"],
    },
  },
});
```

#### 4. Or Use Low-Level API [#4-or-use-low-level-api]

If not using aggregates, specify encryption per-event in the `append` call:

```typescript
await eventStore.append({
  streamId: "User-abc123",
  expectedVersion: 0,
  events: [
    {
      type: "UserRegistered",
      data: { name: "Alice", email: "alice@example.com", loginCount: 0 },
      encryptedFields: ["name", "email"],
      cryptoKeyId: "user:abc123",
    },
  ],
});
```

### Dot-Path Field Notation [#dot-path-field-notation]

Encrypted fields are specified using dot-path notation for nested objects:

```typescript
encryptedFields: ["name", "address.street", "address.city"];
```

Given event data:

```json
{
  "name": "Alice",
  "address": { "street": "123 Main", "city": "Berlin" },
  "active": true
}
```

After encryption, `data` column contains:

```json
{ "address": {}, "active": true }
```

And `encrypted_data` column contains:

```json
{
  "name": { "version": 2, "ciphertext": "...", "iv": "...", "authTag": "..." },
  "address.street": {
    "version": 2,
    "ciphertext": "...",
    "iv": "...",
    "authTag": "..."
  },
  "address.city": {
    "version": 2,
    "ciphertext": "...",
    "iv": "...",
    "authTag": "..."
  }
}
```

### Deletion Flow (Revoking Keys) [#deletion-flow-revoking-keys]

The complete flow for handling a user deletion request via crypto-shredding:

```typescript
// 1. When user registers — create their crypto key
await eventStore.createCryptoKey("user:abc123");

// 2. Normal operation — append events with encrypted PII
await User.append(eventStore, {
  entityId: "abc123",
  expectedVersion: -1,
  events: [
    {
      type: "UserRegistered",
      data: {
        name: "Alice",
        email: "alice@example.com",
        address: { street: "123 Main", city: "Berlin" },
      },
    },
  ],
});

// 3. User requests deletion — revoke the key
await eventStore.revokeKey("user:abc123");

// 4. All future reads return tombstones for encrypted events
const user = await User.load(eventStore, "abc123");
// user.state contains the result of your null-safe evolve handlers
```

#### What `revokeKey` Does [#what-revokekey-does]

Within a single transaction:

1. Sets `revoked_at = now()` on the crypto key.
2. Sets `encrypted_key = NULL`, permanently removing the database copy of the wrapped entity key.

The crypto-key row remains only to retain non-sensitive audit metadata such as `key_id`, `revoked_at`, `algorithm`, and `created_at`. Revocation is idempotent: repeating it keeps the row tombstoned and the key material absent.

Appends lock the entity-key row until their transaction commits. Revocation uses the same PostgreSQL row lock, so the ordering is safe across multiple application replicas: an append that acquired the lock first commits before revocation, while an append that starts after revocation sees the revoked key and fails with `CryptoKeyRevokedError`.

### Handling Tombstones in Evolve Handlers [#handling-tombstones-in-evolve-handlers]

After key revocation, encrypted events are returned as `TombstonedEvent`:

```typescript
interface TombstonedEvent {
  globalPosition: bigint;
  streamId: string;
  streamVersion: number;
  type: string; // Event type name — you know what happened
  data: unknown; // Original-schema public data; encrypted field paths are null
  redactedPaths: readonly string[];
  schemaVersion: number;
  extensions: object; // Extensions are NOT encrypted
  createdAt: Date;
  tombstoned: true;
}
```

**Important characteristics:**

* Only encrypted field paths within `data` become `null` — the PII is cryptographically irrecoverable; other fields remain available in the original schema.
* `extensions` is still available (extensions are never encrypted).
* `type` is still available — you know *what* business event happened, just not the personal details.
* Non-encrypted events in the same stream are **not affected**.

Evolve handlers must distinguish complete events from redacted, original-schema payloads. Validate public fields before using them and do not retain erased PII from previous state:

```typescript
evolve: {
  UserRegistered: (state, event) => {
    if ("tombstoned" in event) {
      const data = event.data;
      const active = typeof data === "object" && data !== null &&
        "active" in data && typeof data.active === "boolean"
          ? data.active : false;
      return { ...state, name: "", email: "", active };
    }
    return { ...state, name: event.data.name,
      email: event.data.email, active: event.data.active };
  },
}
```

The creation event can itself be redacted; initialize any encrypted fields from safe defaults while retaining public fields. Avoid copying old PII from an earlier state when a field has been explicitly redacted.

**Snapshot warning:** Aggregate encryption is not inherited by `defineSnapshot`. A snapshot can otherwise persist decrypted PII in plaintext, surviving source-key revocation. Configure snapshot encryption explicitly with the same entity key and all sensitive snapshot-state paths; see the [snapshot encryption example](/docs/snapshots#encrypt-sensitive-snapshot-state). Also erase or invalidate downstream read models, caches, and exported copies.

### Rotating Secrets Without Downtime [#rotating-secrets-without-downtime]

For a single application instance, add the new version to the keyring and set it as `currentVersion`. For an HA or rolling deployment, use two phases so every replica can read version `2` before any replica writes it.

<Callout type="warn">
  **Backup recommendation:** A restorable backup of the event-store database
  before every secret rotation is strongly recommended. Verify that the backup
  includes both `events` and `crypto_keys`, and keep the old secret securely
  available until the rollout and recovery procedure have been verified. A
  database backup without its corresponding secret cannot decrypt encrypted
  entity keys.
</Callout>

#### Phase 1: Distribute the New Secret [#phase-1-distribute-the-new-secret]

Deploy this configuration to every replica while version `1` remains current:

```typescript
secrets: {
  currentVersion: 1,
  secrets: [
    { version: 1, value: "old-secret" },
    { version: 2, value: "new-secret" },
  ],
};
```

Wait until all replicas have this configuration. They continue writing version `1`, but all of them can decrypt version `2`.

#### Phase 2: Activate the New Secret [#phase-2-activate-the-new-secret]

After phase 1 is complete, keep the same keyring and change only the active version:

```typescript
secrets: {
  currentVersion: 2,
  secrets: [
    { version: 1, value: "old-secret" },
    { version: 2, value: "new-secret" },
  ],
};
```

New entity-key envelopes and encrypted fields use version `2`. A replica still on phase 1 can read a version `2` envelope and will never downgrade it back to version `1`. Existing version `1` envelopes are decrypted by direct version lookup and are not rewritten during reads. The next append for an entity lazily re-wraps only its entity key with version `2`; existing event ciphertext is not rewritten. No full database migration or downtime is required.

#### Removing the Old Secret [#removing-the-old-secret]

Do not remove version `1` merely because phase 2 finished. Lazy rotation means an inactive entity may keep a version `1` envelope indefinitely. Before removal, verify that no active entity-key envelope still uses it:

```sql
SELECT
  get_byte(encrypted_key, 4)::bigint * 16777216
    + get_byte(encrypted_key, 5)::bigint * 65536
    + get_byte(encrypted_key, 6)::bigint * 256
    + get_byte(encrypted_key, 7)::bigint AS secret_version,
  count(*)
FROM event_store.crypto_keys
WHERE revoked_at IS NULL
GROUP BY secret_version
ORDER BY secret_version;
```

Use your configured schema instead of `event_store`. Keep the old secret until the old version count is zero, retain it indefinitely, or intentionally retire the remaining data. An inactive key is not automatically rewrapped by reads.

<Callout type="warn">
  **Migration warning:** Deploy the new secret configuration before removing the
  old secret. Removing a secret version makes envelopes using it permanently
  unreadable.
</Callout>

### Error Handling [#error-handling]

| Error                        | When                                                                 | Recovery                                      |
| ---------------------------- | -------------------------------------------------------------------- | --------------------------------------------- |
| `CryptoKeyRevokedError`      | Trying to encrypt **new** events with a revoked key                  | Do not write PII for deleted users            |
| `CryptoKeyNotFoundError`     | Crypto key does not exist in the key store                           | Create the key first with `createCryptoKey()` |
| `CryptoKeyIdRequiredError`   | `encryptedFields` has no non-empty `cryptoKeyId`                     | Provide the entity key ID                     |
| `CryptoSecretsRequiredError` | Crypto operation without `secrets` or a complete environment keyring | Provide configured secrets                    |

`CryptoKeyRevokedError` is only thrown on **write** operations. On **read**, revoked keys simply produce tombstoned events without throwing.

### Cryptographic Security Details [#cryptographic-security-details]

* **Algorithm:** AES-256-GCM (authenticated encryption) for both secret-to-entity and entity-to-field encryption.
* **IV:** 12 bytes, randomly generated per encryption operation.
* **Auth tag:** 16 bytes.
* **Entity key storage format:** Versioned envelope `[magic][secret version][iv][authTag][ciphertext]` stored as `BYTEA`; the header and key ID are authenticated as AAD.
* **Field storage format:** `{ version, ciphertext, iv, authTag }` as base64 strings in `JSONB`; event ID, key ID, field path, and version are authenticated as AAD.
* **Key cache:** Decrypted entity keys are cached in a per-read-call `Map` (not across requests) to avoid redundant DB lookups within a single stream load.

## Operational & Legal Realities (Backups, WAL, and Retention) [#operational--legal-realities-backups-wal-and-retention]

Regardless of whether you choose Pattern A (Reference Tables) or Pattern B (Crypto-Shredding), neither pattern physically alters historical database backups or Write-Ahead Logs (WAL). Understanding these operational boundaries is essential for an airtight compliance posture.

### 1. Database Backups & Point-in-Time Recovery (PITR) [#1-database-backups--point-in-time-recovery-pitr]

* **Pattern A (Reference Tables):** Executing `DELETE FROM user_profiles WHERE user_id = $1` deletes the row in the active database. However, a database dump or cloud snapshot taken yesterday still contains that row. If that backup is ever restored, deleted profiles will reappear unless your disaster recovery procedures re-apply deletion logs.
* **Pattern B (Crypto-Shredding):** Calling `eventStore.revokeKey(keyId)` zeroes the `encrypted_key` in the live database. Past backups still contain the encrypted key and root secret. If an old backup is restored, the key becomes active again unless the key revocation is replayed or root secrets have rotated.
* **Actionable Guidance:** Establish a clearly documented backup retention schedule (e.g., 14 to 30 days). Under GDPR guidance, data in static disaster-recovery archives is generally considered compliant if it is put beyond practical use and naturally purged as backup retention windows expire.

### 2. Write-Ahead Logs (WAL) & Streaming Replicas [#2-write-ahead-logs-wal--streaming-replicas]

PostgreSQL WAL segments record low-level disk page modifications. In Pattern A, WAL contains the pre-image of the deleted row; in Pattern B, WAL contains the pre-image of the revoked key.

* WAL files are transient and automatically recycled based on PostgreSQL configuration (`wal_keep_size`, `max_wal_size`, and archive cleanup scripts).
* Standby replicas replicate deletion transactions near-instantaneously: a SQL `DELETE` or `revokeKey` query executed on the primary propagates to all streaming replicas within milliseconds.

### 3. Legal Posture & Data Protection Authorities [#3-legal-posture--data-protection-authorities]

* **Physical Deletion (Pattern A):** Universally recognized by European Data Protection Authorities (including the CNIL, ICO, and BfDI) as definitive erasure under Article 17.
* **Crypto-Shredding (Pattern B):** Cryptographic erasure is widely accepted as an effective technical measure when key destruction is permanent, the algorithm is state-of-the-art (AES-256-GCM), and root keys are controlled. Because encrypted ciphertext remains in the immutable log, supervisory authorities classify the residual data as pseudonymized until the ciphertext itself is purged or becomes impossible to decrypt even against future computational advances.


# Database Schema (/docs/database-schema)



All event store tables live in a dedicated PostgreSQL schema (default: `event_store`), isolated from application tables. Schema migrations are versioned and serialized across application replicas. By default, `setup()` applies pending migrations on startup.

## Schema Isolation [#schema-isolation]

The event store uses its own PostgreSQL schema. This keeps event store tables completely separate from application tables (e.g. those managed by Prisma or other ORMs).

```sql
CREATE SCHEMA IF NOT EXISTS event_store;
```

The schema name defaults to `event_store` but can be configured:

```typescript
const eventStore = new EventStore({
  pool,
  schema: "my_custom_schema", // Must match /^[a-z_][a-z0-9_]{0,62}$/
});
```

## Tables [#tables]

### `events` [#events]

The core append-only event log. Columns are aligned with the **CloudEvents v1.0.2** specification. Event-backed snapshots are stored here as generated events with event types ending in `Snapshot`.

```sql
CREATE TABLE {schema}.events (
  global_position  BIGSERIAL        NOT NULL,
  stream_id        TEXT             NOT NULL,
  stream_version   INTEGER          NOT NULL,
  id               TEXT             NOT NULL,
  source           TEXT             NOT NULL,
  specversion      TEXT             NOT NULL,
  event_type       TEXT             NOT NULL,
  subject          TEXT             NOT NULL,
  time             TIMESTAMPTZ      NOT NULL,
  datacontenttype  TEXT             NOT NULL,
  data             JSONB            NOT NULL,
  extensions       JSONB            NOT NULL,
  encrypted_data   JSONB            NULL,
  crypto_key_id    TEXT             NULL,
  schema_version   INTEGER          NOT NULL,
  created_at       TIMESTAMPTZ      NOT NULL DEFAULT now(),
  txid             XID8             NOT NULL DEFAULT pg_current_xact_id(),

  PRIMARY KEY (global_position),
  UNIQUE (stream_id, stream_version)
);
```

| Column            | Type                   | Description                                                                                                                         |
| ----------------- | ---------------------- | ----------------------------------------------------------------------------------------------------------------------------------- |
| `global_position` | `BIGSERIAL PK`         | Unique sequence position; may have gaps and is not commit order                                                                     |
| `stream_id`       | `TEXT NOT NULL`        | Stream identifier (e.g. `"Order-123"`)                                                                                              |
| `stream_version`  | `INTEGER NOT NULL`     | Sequential version within the stream (1, 2, 3, ...)                                                                                 |
| `id`              | `TEXT NOT NULL`        | CloudEvents `id` — `"{streamId}/{streamVersion}"`                                                                                   |
| `source`          | `TEXT NOT NULL`        | CloudEvents `source` — URI-reference                                                                                                |
| `specversion`     | `TEXT NOT NULL`        | Always `"1.0"`                                                                                                                      |
| `event_type`      | `TEXT NOT NULL`        | CloudEvents `type` — event type name                                                                                                |
| `subject`         | `TEXT NOT NULL`        | CloudEvents `subject` — the stream ID                                                                                               |
| `time`            | `TIMESTAMPTZ NOT NULL` | CloudEvents `time` — ISO 8601 timestamp                                                                                             |
| `datacontenttype` | `TEXT NOT NULL`        | Always `"application/json"`                                                                                                         |
| `data`            | `JSONB NOT NULL`       | Event payload (PII fields removed if encryption is used)                                                                            |
| `extensions`      | `JSONB NOT NULL`       | CloudEvents extension attributes                                                                                                    |
| `encrypted_data`  | `JSONB NULL`           | Encrypted PII field blobs                                                                                                           |
| `crypto_key_id`   | `TEXT NULL`            | Reference to `crypto_keys.key_id`                                                                                                   |
| `schema_version`  | `INTEGER NOT NULL`     | Schema version for upcasting                                                                                                        |
| `created_at`      | `TIMESTAMPTZ NOT NULL` | Auto-set by the database                                                                                                            |
| `txid`            | `XID8 NOT NULL`        | Appending transaction id, used to compute the commit-safe watermark so cursor consumers never skip a late-committing lower position |

**Constraints:**

* `PRIMARY KEY (global_position)` — global ordering
* `UNIQUE (stream_id, stream_version)` — enforces sequential versions per stream, serves as OCC safety net

**Indexes:**

```sql
CREATE INDEX idx_events_stream_id
  ON {schema}.events (stream_id, stream_version);

CREATE INDEX idx_events_stream_global_position
  ON {schema}.events (stream_id, global_position, id);

CREATE INDEX idx_events_event_type
  ON {schema}.events (event_type);

CREATE INDEX idx_events_created_at
  ON {schema}.events (created_at);

CREATE INDEX idx_events_source_id
  ON {schema}.events (source, id);
```

`{schema}` is a documentation placeholder, not executable SQL. The `crypto_key_id` and `event_global_pos` references are logical references, not declared foreign-key constraints.

### Event-backed snapshots [#event-backed-snapshots]

Snapshots do not use a separate table. They are normal rows in `{schema}.events`, written to the same stream they optimize.

For example, a `BankAccountBalance` snapshot for stream `Transaction-account-123` is stored as an event with `event_type = 'BankAccountBalanceSnapshot'` in that same stream. Because it is a real event row, it advances `stream_version` and participates in the normal stream ordering and OCC rules.

### `outbox` [#outbox]

Transactional outbox for at-least-once delivery.

```sql
CREATE TABLE {schema}.outbox (
  id               BIGSERIAL        PRIMARY KEY,
  event_global_pos BIGINT           NOT NULL,
  topic            TEXT             NOT NULL,
  payload          JSONB            NOT NULL,
  processed_at     TIMESTAMPTZ      NULL,
  created_at       TIMESTAMPTZ      NOT NULL DEFAULT now()
);
```

| Column             | Type                   | Description                           |
| ------------------ | ---------------------- | ------------------------------------- |
| `id`               | `BIGSERIAL PK`         | Auto-incrementing outbox entry ID     |
| `event_global_pos` | `BIGINT NOT NULL`      | Reference to `events.global_position` |
| `topic`            | `TEXT NOT NULL`        | Message topic/channel name            |
| `payload`          | `JSONB NOT NULL`       | CloudEvents v1.0.2 JSON payload       |
| `processed_at`     | `TIMESTAMPTZ NULL`     | `NULL` = pending, set = processed     |
| `created_at`       | `TIMESTAMPTZ NOT NULL` | Auto-set by database                  |

```sql
CREATE INDEX idx_outbox_pending
  ON {schema}.outbox (created_at) WHERE processed_at IS NULL;
```

### `crypto_keys` [#crypto_keys]

Per-entity encryption keys for GDPR crypto-shredding.

```sql
CREATE TABLE {schema}.crypto_keys (
  key_id           TEXT             PRIMARY KEY,
  encrypted_key    BYTEA            NULL,
  algorithm        TEXT             NOT NULL DEFAULT 'aes-256-gcm',
  revoked_at       TIMESTAMPTZ      NULL,
  created_at       TIMESTAMPTZ      NOT NULL DEFAULT now()
);
```

| Column          | Type                   | Description                                  |
| --------------- | ---------------------- | -------------------------------------------- |
| `key_id`        | `TEXT PK`              | Entity key identifier (e.g. `"user:abc123"`) |
| `encrypted_key` | `BYTEA NULL`           | Versioned envelope; `NULL` after revocation  |
| `algorithm`     | `TEXT NOT NULL`        | Always `"aes-256-gcm"`                       |
| `revoked_at`    | `TIMESTAMPTZ NULL`     | `NULL` = active, set = revoked               |
| `created_at`    | `TIMESTAMPTZ NOT NULL` | Auto-set by database                         |

Revoked key rows are retained only for audit metadata. Their `encrypted_key` is set to `NULL` during revocation.

### `projections` [#projections]

Checkpoint tracking for projection runners.

```sql
CREATE TABLE {schema}.projections (
  projection_name  TEXT             PRIMARY KEY,
  last_position    BIGINT           NOT NULL DEFAULT 0,
  updated_at       TIMESTAMPTZ      NOT NULL DEFAULT now()
);
```

| Column            | Type                   | Description                      |
| ----------------- | ---------------------- | -------------------------------- |
| `projection_name` | `TEXT PK`              | Unique projection identifier     |
| `last_position`   | `BIGINT NOT NULL`      | Last processed `global_position` |
| `updated_at`      | `TIMESTAMPTZ NOT NULL` | Last checkpoint update time      |

### `idempotency_keys` [#idempotency_keys]

Deduplication table for idempotent stream appends. Prevents duplicate event writes and repeated outbox notifications on retries.

```sql
CREATE TABLE {schema}.idempotency_keys (
  key              TEXT             PRIMARY KEY,
  stream_id        TEXT             NOT NULL,
  request_hash     TEXT             NOT NULL,
  from_version     INTEGER          NOT NULL,
  to_version       INTEGER          NOT NULL,
  global_positions BIGINT[]         NOT NULL,
  created_at       TIMESTAMPTZ      NOT NULL DEFAULT now()
);
```

| Column             | Type                   | Description                                                |
| ------------------ | ---------------------- | ---------------------------------------------------------- |
| `key`              | `TEXT PK`              | Idempotency key supplied by the caller                     |
| `stream_id`        | `TEXT NOT NULL`        | Target stream identifier                                   |
| `request_hash`     | `TEXT NOT NULL`        | SHA-256 fingerprint of the appended event payload          |
| `from_version`     | `INTEGER NOT NULL`     | Stream version of the first event in the original append   |
| `to_version`       | `INTEGER NOT NULL`     | Stream version of the last event in the original append    |
| `global_positions` | `BIGINT[] NOT NULL`    | Global positions assigned to events in the original append |
| `created_at`       | `TIMESTAMPTZ NOT NULL` | Auto-set by database, used for age-based retention cleanup |

```sql
CREATE INDEX idx_idempotency_keys_created_at
  ON {schema}.idempotency_keys (created_at);
```

## Migrations [#migrations]

By default, `eventStore.setup()` applies pending migrations. A schema-local version table (`{schema}.schema_version`) tracks the installed version; migration DDL and its version update run in one transaction under a transaction-scoped PostgreSQL advisory lock. Concurrent replicas serialize on that lock and do not repeat DDL after the first migration completes.

```typescript
await eventStore.setup(); // Default: migrate, then verify
```

For deployments where application instances do not have DDL privileges, use **the same migration routine** in a separate deployment job, then start applications in verify-only mode:

```typescript
import { EventStore, migrateEventStore } from "alvyn";

// Deployment job, using a DDL-capable pool connected to the primary:
await migrateEventStore({ pool: migrationPool, schema: "event_store" });

// Application instances, using a pool with schema/table access but no DDL rights:
const store = new EventStore({ pool: appPool, migrationMode: "verify" });
await store.setup(); // No DDL; fails if the version or event-position trigger is invalid
```

This versioned baseline does not silently claim or modify an older, unversioned Alvyn schema: if Alvyn tables already exist without `{schema}.schema_version`, migration stops with an error and leaves existing data intact. Choose a fresh schema or explicitly plan a data migration before upgrading.

The app role needs `USAGE` on the schema and `SELECT` on `{schema}.schema_version` in addition to the privileges required for normal event-store operations. Deploy the migration job **before** starting instances with a newer Alvyn version. The migration path is tested with PostgreSQL 16. Plan schema/trigger changes as a deployment operation rather than running them concurrently with live writes. Verification fails closed if the event-position trigger or safe-watermark sequence is missing, or the schema version is unsupported. No session-scoped advisory locks are used during migrations.


# Full-stack Example (/docs/full-stack-example)



The repository includes a full-stack banking example in
`examples/example-alvyn-full-stack-app`. It demonstrates how Alvyn fits into a
web application rather than requiring a particular server or UI framework.

## Included Stack [#included-stack]

* Vite and Nitro for the development and server runtime
* React with Apollo Client for the browser UI
* Elysia and GraphQL Yoga for the GraphQL endpoint
* GraphQL-over-SSE for live subscriptions
* PostgreSQL with Alvyn event streams and an event-backed balance snapshot
* Testcontainers for a reusable local PostgreSQL development database

## Live Banking Flow [#live-banking-flow]

The example defines separate bank-account and transaction aggregates. Deposits
and withdrawals are appended to the transaction stream, while
`BankAccountBalance` maintains a derived balance snapshot. GraphQL subscription
resolvers consume the aggregate subscription API to publish live transaction
and balance updates to the selected account.

The withdrawal mutation also demonstrates typed GraphQL errors: insufficient
funds are returned with an `INSUFFICIENT_FUNDS` code and a payload containing
the current balance and requested amount.

## Run It Locally [#run-it-locally]

From the example directory:

<CodeBlockTabs defaultValue="npm" groupId="package-manager">
  <CodeBlockTabsList>
    <CodeBlockTabsTrigger value="npm">
      npm
    </CodeBlockTabsTrigger>

    <CodeBlockTabsTrigger value="pnpm">
      pnpm
    </CodeBlockTabsTrigger>

    <CodeBlockTabsTrigger value="yarn">
      yarn
    </CodeBlockTabsTrigger>

    <CodeBlockTabsTrigger value="bun">
      bun
    </CodeBlockTabsTrigger>
  </CodeBlockTabsList>

  <CodeBlockTab value="npm">
    ```bash
    npm install
    npm run dev
    ```
  </CodeBlockTab>

  <CodeBlockTab value="pnpm">
    ```bash
    pnpm install
    pnpm run dev
    ```
  </CodeBlockTab>

  <CodeBlockTab value="yarn">
    ```bash
    yarn install
    yarn dev
    ```
  </CodeBlockTab>

  <CodeBlockTab value="bun">
    ```bash
    bun install
    bun run dev
    ```
  </CodeBlockTab>
</CodeBlockTabs>

When `DATABASE_URL` is not set, the Vite configuration starts or reuses the
`alvyn-example-postgres-dev` PostgreSQL container through Docker. To use an
existing database, provide `DATABASE_URL` before starting the app.

The example is intended for learning and local development. It does not
provide authentication or authorization and should not be used as a production
banking service.

See the [example source on GitHub](https://github.com/lox-solutions/alvyn/tree/main/examples/example-alvyn-full-stack-app)
for the complete application.


# Introduction (/docs)



Alvyn is a **TypeScript library for event history on PostgreSQL**. Record meaningful facts, rebuild application state, and inspect how a workflow reached its current state. AI agents are an approachable starting point; orders, approvals, billing, and other domain applications use the same primitives. You do not need CQRS knowledge or an all-at-once rewrite.

## Use Alvyn in an application [#use-alvyn-in-an-application]

<CodeBlockTabs defaultValue="npm" groupId="package-manager">
  <CodeBlockTabsList>
    <CodeBlockTabsTrigger value="npm">
      npm
    </CodeBlockTabsTrigger>

    <CodeBlockTabsTrigger value="pnpm">
      pnpm
    </CodeBlockTabsTrigger>

    <CodeBlockTabsTrigger value="yarn">
      yarn
    </CodeBlockTabsTrigger>

    <CodeBlockTabsTrigger value="bun">
      bun
    </CodeBlockTabsTrigger>
  </CodeBlockTabsList>

  <CodeBlockTab value="npm">
    ```bash
    npm install alvyn pg
    ```
  </CodeBlockTab>

  <CodeBlockTab value="pnpm">
    ```bash
    pnpm add alvyn pg
    ```
  </CodeBlockTab>

  <CodeBlockTab value="yarn">
    ```bash
    yarn add alvyn pg
    ```
  </CodeBlockTab>

  <CodeBlockTab value="bun">
    ```bash
    bun add alvyn pg
    ```
  </CodeBlockTab>
</CodeBlockTabs>

You supply and close the connection pool. This snippet records a goal and a known result; it does not call a model or execute a tool. See the [Agent History, Replay & Audit playbook](/docs/playbooks/ai-agent-deterministic-memory-and-audit) for model integration boundaries.

```typescript
import { Pool } from "pg";
import { EventStore, defineAggregate } from "alvyn";

type AgentEvents = {
  GoalReceived: { goal: string };
  RunCompleted: { output: string };
};
type AgentState = { goal: string | null; output: string | null };

const AgentRun = defineAggregate<AgentState, AgentEvents>()({
  streamPrefix: "AgentRun",
  evolve: {
    GoalReceived: (_state, event) => ({
      goal: event.data?.goal ?? null,
      output: null,
    }),
    RunCompleted: (state, event) => ({
      goal: state?.goal ?? null,
      output: event.data?.output ?? null,
    }),
  },
});

const pool = new Pool({ connectionString: process.env.DATABASE_URL });
try {
  const store = new EventStore({ pool });
  await store.setup();
  const runId = crypto.randomUUID();
  const created = await AgentRun.append(store, {
    entityId: runId,
    expectedVersion: -1,
    events: [
      { type: "GoalReceived", data: { goal: "Find the returns policy" } },
    ],
  });
  await AgentRun.append(store, {
    entityId: runId,
    expectedVersion: created.toVersion,
    events: [
      {
        type: "RunCompleted",
        data: { output: "Returns accepted for 30 days." },
      },
    ],
  });
  console.log((await AgentRun.load(store, runId)).state);
} finally {
  await pool.end();
}
```

The first evolve handler receives `null` at runtime and must initialize the whole state. An empty aggregate loads with `state: null` and `version: 0`; use `expectedVersion: -1` to explicitly create a new stream. Later appends use the loaded version or the previous append's `toVersion`. Define a handler for every event in the event map.

## Building blocks [#building-blocks]

* **Typed aggregates** group related facts into streams and rebuild current state.
* **Historical replay** uses `load` or `loadFrom` plus your reducer over an explicit event prefix; current aggregate load is not a historical query.
* **Snapshots** cache expensive derived state as separate events.
* **Subscriptions and projections** support resumable consumers and query-oriented read models.
* **Transactional outbox** records delivery work atomically with events; delivery is at least once and external effects need idempotency.
* **Schema evolution** applies read-time upcasters to older event shapes.
* **Encryption and key revocation** support privacy workflows when explicitly configured.

Replay reproduces reducer state from retained, readable events, not a deterministic model rerun. Only instrumented actions are recorded, not hidden thoughts. Event history is neither automatic regulatory compliance nor tamper-proof evidence. Encryption does not automatically cover snapshots, projections, logs, backups, or provider copies; deletion can make original state unreconstructable.

## Requirements [#requirements]

* Node.js matching the package engines: `^22.18.0 || ^24.11.0 || >=26.0.0`.
* PostgreSQL 14 or later for persistence.
* `pg` 8 as a peer dependency.

## Next steps [#next-steps]

* [Aggregates](/docs/aggregates) — event maps and domain state.
* [Aggregate design](/docs/playbooks/aggregate-design-and-stream-boundaries) — stream boundaries and concurrency.
* [Snapshots](/docs/snapshots) — faster derived-state reads.
* [Subscriptions](/docs/subscriptions) — catch-up and live event delivery.
* [Transactional outbox](/docs/outbox) — at-least-once downstream delivery.
* [Projections](/docs/projections) — read models and checkpoints.
* [Data privacy](/docs/crypto-shredding) — encryption and deletion boundaries.
* [Schema evolution](/docs/schema-evolution) — handling older payloads.
* [API reference](/docs/api-reference) — supported methods and types.


# Production Load Testing (/docs/load-testing)



<Callout type="info">
  **Production sizing note:** These results demonstrate that Alvyn reliably
  handles the tested daily workload under the benchmarked configuration. Because
  performance in production depends on infrastructure, network topology, and
  database sizing, always validate capacity limits, write contention, and
  PostgreSQL high availability against your specific operational requirements.
</Callout>

The load-test suite measures Alvyn through a small HTTP application that models
bank-account traffic. It starts independent application processes, each with its
own `pg.Pool` and `EventStore`, and makes them share one PostgreSQL primary.
The logical workload is deterministic when the same configuration and `--seed`
are used; timings, scheduling interleavings, conflicts, and performance vary.

## What the suite tests [#what-the-suite-tests]

The HTTP harness has three profiles:

| Profile    | Purpose                                   | Default workload                                                      |
| ---------- | ----------------------------------------- | --------------------------------------------------------------------- |
| `daily`    | Simulate an accelerated day of activity   | 100,000 users, 10 operations each, one million requests over one hour |
| `capacity` | Find the highest sustainable offered rate | 30-second stages from 100 to 2,000 requests/second                    |
| `custom`   | Run a focused scenario                    | Explicit request count and concurrency settings                       |

The daily and capacity profiles use open-loop traffic. Requests are scheduled at
their target arrival times up to the configured `--concurrency`, so an overloaded
system cannot hide saturation by simply slowing down request generation. The
custom profile retains the original closed-loop behavior for small, focused
checks.

The example application exposes:

* `GET /health` for replica readiness;
* `GET /accounts/:id` for aggregate loads; and
* `POST /accounts/:id/deposit` for aggregate decisions and optimistic-concurrency
  appends.

The coordinator seeds account histories directly through Alvyn before measured
traffic begins. Measured requests still use the real HTTP and aggregate path.
After traffic, the verifier checks event counts, contiguous versions, ordering,
balances, generated snapshot events, and every successful logical deposit.
During traffic, every successful read response is also checked for the requested
stream ID, open account state, a stored version not behind the replay-derived
source-event count, and a balance equal to the replay-derived deposit total.
Successful deposits must echo their account, amount, operation token, and
one-event version range.

## Running the tests [#running-the-tests]

Run the commands from the repository root. Node.js 22.18.0 or newer, pnpm, and a
working Docker installation are required. The load tests start PostgreSQL with
Testcontainers; do not start a separate database for the default local runs.

### Verify the normal test suite first [#verify-the-normal-test-suite-first]

These checks do not run the long load profiles:

```bash
pnpm run build
pnpm test
```

The load test profiles are intentionally separate from `pnpm test` because
they seed large histories and may run for several minutes or hours. Fast
deterministic tests for harness configuration, scheduling, metrics, SLOs,
validation, deadlines, and reporting remain part of the normal suite.

### Run the direct EventStore smoke test [#run-the-direct-eventstore-smoke-test]

This scenario exercises multiple worker processes, seeded stream history,
appends, loads, optimistic-concurrency retries, and final stream verification
without the HTTP layer:

```bash
pnpm test:load -- \
  --workers 2 \
  --pool-size 2 \
  --streams 4 \
  --hot-streams 2 \
  --history 100 \
  --operations 50 \
  --append-percent 60 \
  --batch-size 2 \
  --max-retries 20 \
  --seed 42 \
  --verbose \
  --output .load-smoke-report.json
```

For a larger-history direct run:

```bash
pnpm test:load -- \
  --workers 4 \
  --pool-size 8 \
  --streams 100 \
  --hot-streams 4 \
  --history 10000 \
  --operations 1000 \
  --append-percent 60 \
  --batch-size 10 \
  --max-retries 30 \
  --seed 42 \
  --verbose \
  --output load-report.json
```

`--history` is the number of existing events per stream, while
`--operations` is the number of live operations per worker. A successful run
exits with code `0` and verifies that no required event is missing or duplicated.

### Run the HTTP smoke test [#run-the-http-smoke-test]

Use a short run to verify the HTTP workers and aggregate endpoints before
starting a long profile:

```bash
pnpm test:load:http -- \
  --workers 2 \
  --pool-size 4 \
  --accounts 20 \
  --active-users 20 \
  --operations-per-user 10 \
  --hot-accounts 4 \
  --hot-traffic-percent 75 \
  --history 100 \
  --seed-batch-size 50 \
  --requests 500 \
  --read-percent 35 \
  --concurrency 32 \
  --max-retries 20 \
  --seed 42 \
  --verbose \
  --snapshot-benchmark-requests 0 \
  --output .http-load-smoke-report.json
```

`--snapshot-benchmark-requests 0` skips the optional paired snapshot latency
benchmark. Snapshot events created during normal aggregate appends are still
verified.

For a value `N > 0`, the benchmark collects `N` paired samples and performs
`2N` HTTP requests: one full replay and one snapshot-aware read per sample. It
uses up to `min(N, --accounts)` distinct accounts and cycles through those
accounts when `N` exceeds the account count.

### Run the production daily profile [#run-the-production-daily-profile]

This is the default production-oriented local scenario:

```bash
pnpm test:load:http:daily -- \
  --seed 42 \
  --verbose \
  --output http-load-daily.json
```

The profile starts three independent local application processes, seeds 60
history events for each of 100,000 accounts, and measures one million requests:

* `10%` of requests during a `20%` off-peak period;
* `80%` during a `60%` peak period; and
* `10%` during a `20%` cool-down period.

The default SLOs are enforced:

* read `p95` no higher than `250 ms`;
* deposit `p95` no higher than `500 ms`; and
* unexpected error rate no higher than `0.1%`.

Handled optimistic-concurrency conflicts and successful retries are reported
separately and are not counted as unexpected errors. Override the budgets with
`--slo-read-p95-ms`, `--slo-deposit-p95-ms`, and
`--slo-error-rate-percent`. Use `--enforce-slos false` for a diagnostic run that
still records the SLO checks without failing on them.

The full profile takes about one hour for traffic alone. Seeding, worker
startup, verification, and cleanup add additional time, so use the smoke
profile first on a new machine.

### Run the capacity profile [#run-the-capacity-profile]

The capacity profile increases offered traffic in stages and reports the highest
stage that both reaches at least `95%` of its target rate and passes all SLOs:

```bash
pnpm test:load:http:capacity -- \
  --seed 42 \
  --verbose \
  --output http-load-capacity.json
```

The default stages run from `100` through `2,000` requests/second in increments
of `100`, with each stage lasting `30` seconds. Reduce the range for a quick
capacity smoke test:

```bash
pnpm test:load:http:capacity -- \
  --capacity-start-rps 100 \
  --capacity-step-rps 100 \
  --capacity-max-rps 300 \
  --capacity-stage-seconds 10 \
  --seed 42 \
  --verbose \
  --output .http-load-capacity-smoke.json
```

## Interpreting the JSON report [#interpreting-the-json-report]

Use `--output <path>` to save a machine-readable report. The most useful fields
are:

* `environment`: generation timestamp, commit revision, Alvyn, Node.js and
  PostgreSQL versions, OS/architecture, CPU model/count, and total host memory;
* `request`, `read`, and `deposit`: attempted, successful, failed, conflict,
  retry, and bounded latency-sample metrics;
* `throughput`: successful requests per second for the measured traffic phase;
* `trafficPhases`: per-phase throughput, latency, errors, and SLO checks;
* `workers`: request attempts routed to each application replica;
* `phaseDurations`: setup, seeding, startup, traffic, snapshot benchmark, and
  verification timing;
* `snapshotBenchmark`: full-replay versus snapshot-aware read latency when the
  optional benchmark is enabled; and
* `verification`: source event counts, stored event counts, snapshot event
  counts, successful deposits, and final balance.

Latency samples are bounded at 10,000 values per operation class, using a
deterministic reservoir sample across the measured run. The percentiles are
therefore useful for long runs without retaining every request in memory, but
rare-tail analysis should use additional observability in a real deployment.

HTTP requests, PostgreSQL connection waits, and PostgreSQL statements default
to a `30,000 ms` operation deadline. Worker readiness defaults to `60,000 ms`,
and the complete run defaults to two hours. Configure these with
`--operation-timeout-ms`, `--worker-ready-timeout-ms`, and `--run-timeout-ms`.
Deadline failures produce a failed diagnostic report when `--output` is set and
trigger worker, pool, and container cleanup.

## Recorded daily-profile result [#recorded-daily-profile-result]

The following result was produced by the default daily profile on 2026-07-15
and saved as `http-load-daily.json`.

### Workload configuration [#workload-configuration]

| Metric                    |                   Result |
| ------------------------- | -----------------------: |
| Application replicas      |      `3` local processes |
| Accounts and active users |    `100,000` / `100,000` |
| Operations per user       |                     `10` |
| Total HTTP requests       |              `1,000,000` |
| Read / deposit mix        |            `70%` / `30%` |
| History per account       |              `60` events |
| Measured traffic duration |                 `1 hour` |
| Peak offered rate         | `370.37 requests/second` |
| Configured connections    |                     `40` |
| Database topology         |   One PostgreSQL primary |

### Correctness and throughput [#correctness-and-throughput]

| KPI                        |                                   Result | Assessment                                 |
| -------------------------- | ---------------------------------------: | ------------------------------------------ |
| Requests succeeded         |                  `1,000,000 / 1,000,000` | No failed requests                         |
| Reads returned HTTP `200`  |                      `700,000 / 700,000` | Response bodies passed contract validation |
| Deposits succeeded         |                      `300,000 / 300,000` | Write path remained correct                |
| Unexpected errors          |                                      `0` | Within the `0.1%` SLO                      |
| Average traffic throughput |                 `277.77 requests/second` | Matches one million requests over one hour |
| Peak traffic throughput    |                 `370.37 requests/second` | Target rate achieved                       |
| Replica routing            | `333,334`, `333,333`, `333,333` attempts | Evenly distributed                         |

The run seeded `6,000,000` events and added `300,000` successful deposit
events. Verification found `6,300,000` source events and `6,400,000` stored
events, including `100,000` generated snapshot events. The final aggregate
balance was `7,550,000`, matching the expected value. This difference between
source and stored events is expected because snapshot events share the same
event streams.

### Latency and SLOs [#latency-and-slos]

| Operation    |      `p50` |      `p95` |      `p99` |                  SLO |
| ------------ | ---------: | ---------: | ---------: | -------------------: |
| Reads        |  `1.49 ms` |  `3.44 ms` |  `6.82 ms` |      `p95 <= 250 ms` |
| Deposits     | `10.97 ms` | `23.91 ms` | `33.18 ms` |      `p95 <= 500 ms` |
| All requests |  `2.03 ms` | `17.88 ms` | `26.49 ms` | Error rate `<= 0.1%` |

All three traffic phases passed their SLO checks. Their achieved throughput and
`p95` latency were:

| Phase     |               Throughput | Read `p95` | Deposit `p95` | All requests `p95` |
| --------- | -----------------------: | ---------: | ------------: | -----------------: |
| Off-peak  | `138.89 requests/second` |  `3.92 ms` |    `18.03 ms` |         `13.88 ms` |
| Peak      | `370.37 requests/second` |  `2.37 ms` |    `25.32 ms` |         `18.93 ms` |
| Cool-down | `138.89 requests/second` |  `3.84 ms` |    `10.38 ms` |          `8.75 ms` |

Overall percentiles are dominated by the larger peak phase, so phase-level
values should be reviewed when making capacity decisions. The run recorded no
optimistic-concurrency conflicts or retries because the daily profile assigns
one independent account to each active user and has no hot-account traffic.

The optional snapshot benchmark was disabled in this report with
`--snapshot-benchmark-requests 0`. The report therefore validates normal
aggregate behavior but does not provide a snapshot-versus-full-replay latency
comparison.

### Run provenance [#run-provenance]

The report was generated at `2026-07-15T10:41:06.867Z` from revision
`517b4e2add0fd803202347c5c22ce82baad17b02` with Alvyn `0.1.2`, Node.js
`v24.13.0`, and PostgreSQL `16.14`. The host was macOS `25.5.0` on `arm64`,
with an Apple M2 Pro (`12` logical CPUs) and `16 GiB` of memory.

The complete run took approximately `68.7 minutes`, including setup, seeding,
worker startup, traffic, and verification. The measured HTTP traffic phase was
exactly one hour; setup and verification are excluded from the throughput KPI.

## What this result proves [#what-this-result-proves]

This is strong evidence that, for the tested workload:

* independent Alvyn application processes can share one PostgreSQL primary;
* one million requests returned expected HTTP statuses and validated response
  bodies, while successful deposits reconciled with final stored state;
* traffic can be distributed evenly across three application processes;
* seeded long-lived histories can be verified without missing or duplicate
  successful events; and
* the observed latency stayed well below the configured balanced SLOs.

## What this result does not prove [#what-this-result-does-not-prove]

The local processes are not a production load balancer or a Kubernetes,
Cloud Run, or Docker Compose deployment. This run did not test:

* stopping or replacing a replica during traffic;
* load-balancer health checks, connection draining, or autoscaling;
* PostgreSQL replication, promotion, or database failover;
* multi-hour resource stability, memory growth, or connection leaks;
* high-contention shared accounts and resulting OCC retry exhaustion;
* authentication, authorization, TLS, gateways, or external dependencies; or
* transfers, withdrawals, idempotency after client timeouts, and other banking
  commands.

The daily configuration intentionally uses one account per user and no hot
accounts, so it recorded zero OCC conflicts and retries. That is appropriate for
the independent-account scenario, but it is not evidence that highly contended
aggregates never conflict. The capacity profile includes a controlled hot-account
subset for a more demanding write workload.

For a broader production-readiness decision, run the daily and capacity profiles
with the same database and resource limits intended for deployment, add a
contention scenario, collect CPU and memory metrics during a soak run, and test
the actual load-balancer and replica lifecycle. PostgreSQL high availability
requires a separate database-topology test.


# Transactional Outbox (/docs/outbox)



The **Transactional Outbox** pattern guarantees **at-least-once delivery** of events to external message brokers (e.g. NATS, Apache Kafka, RabbitMQ, AWS SQS) without distributed transactions.

Outbox entries are inserted in the **exact same PostgreSQL transaction** as the event store append. If the transaction commits, both the event and its corresponding outbox rows are guaranteed to exist.

```
1. Client Appends Event
   │
   ▼
┌────────────────────────────────────────┐
│ PostgreSQL ACID Transaction            │
│ ├─ INSERT INTO events (...)            │
│ └─ INSERT INTO outbox (...)            │
└────────────────────────────────────────┘
   │
   ▼ (committed)
┌────────────────────────────────────────┐
│ Relay Worker Fleet                     │
│ ├─ SELECT ... FOR UPDATE SKIP LOCKED   │
│ ├─ Dispatch to Message Broker (NATS)   │
│ └─ UPDATE outbox SET processed_at=now()│
└────────────────────────────────────────┘
```

## Solving the Dual-Write Problem [#solving-the-dual-write-problem]

A common architectural pitfall is trying to write to PostgreSQL and publish to a message broker in separate steps:

```typescript
// ❌ DANGEROUS: Dual-write hazard
await eventStore.append({
  streamId,
  expectedVersion: 0,
  events: [orderPlaced],
});
await natsBroker.publish("orders", orderPlaced); // If this fails or worker crashes, event is lost to broker!
```

If your node process crashes or the network blips between those two lines, your message broker will never receive the event.

With Alvyn's Transactional Outbox, recording the intent to publish is atomic and durable; external delivery happens later:

```typescript
// ✅ SAFE: Single ACID transaction
await eventStore.append({
  streamId: "Order-123",
  expectedVersion: 0,
  events: [{ type: "OrderPlaced", data: { total: 99.99 } }],
  outboxTopics: ["orders", "analytics.orders"],
});
```

## Publishing to the Outbox [#publishing-to-the-outbox]

### Direct Append [#direct-append]

Specify `outboxTopics` array on `eventStore.append()`:

```typescript
await eventStore.append({
  streamId: "Order-123",
  expectedVersion: 0,
  events: [
    {
      type: "OrderPlaced",
      data: { total: 99.99, customerId: "cust_482" },
      extensions: { correlationid: "req_8819" },
    },
  ],
  outboxTopics: ["orders", "notifications"],
});
```

### With Typed Aggregates [#with-typed-aggregates]

When using `defineAggregate`, pass `outboxTopics` directly to the aggregate append handle:

```typescript
await Order.append(eventStore, {
  entityId: "order-123",
  expectedVersion: order.version,
  events: [
    {
      type: "OrderShipped",
      data: { trackingNumber: "TRACK-99128" },
    },
  ],
  outboxTopics: ["shipping-events"],
});
```

## Outbox Payload Format (CloudEvents v1.0.2) [#outbox-payload-format-cloudevents-v102]

Every outbox record generates an industry-standard CloudEvents v1.0.2 compliant JSON envelope:

```json
{
  "specversion": "1.0",
  "id": "Order-123/1",
  "type": "OrderPlaced",
  "source": "urn:my-app:event-store",
  "subject": "Order-123",
  "time": "2026-08-24T10:00:00.000Z",
  "datacontenttype": "application/json",
  "correlationid": "req_8819",
  "actorid": "usr_9921",
  "schemaversion": 1,
  "data": {
    "total": 99.99,
    "customerId": "cust_482"
  }
}
```

Encrypted fields in newly created outbox payloads are represented as `null`; PII is never decrypted by the relay. Outbox messages are fixed at append time: revoking a key later does not rewrite previously stored payloads (including entries created by older versions of Alvyn).

## Worker Relay with `processOutbox()` [#worker-relay-with-processoutbox]

Build a relay worker that claims pending outbox entries and dispatches them to your external message broker:

```typescript
import { EventStore } from "alvyn";
import { connect } from "nats";

const nc = await connect({ servers: process.env.NATS_URL });
const jsm = await nc.jetstream();

async function runOutboxRelay() {
  try {
    // Process up to 100 pending messages per batch
    const processedCount = await eventStore.processOutbox(
      async (entries, client) => {
        for (const entry of entries) {
          // entry.topic matches the string from outboxTopics
          // entry.payload contains the CloudEvent JSON object
          await jsm.publish(
            entry.topic,
            new TextEncoder().encode(JSON.stringify(entry.payload)),
          );
        }
      },
      100,
    );

    if (processedCount > 0) {
      console.log(`Successfully dispatched ${processedCount} outbox events.`);
    }
  } catch (err) {
    console.error("Outbox relay error:", err);
  }
}

// Continuously poll for pending outbox items
setInterval(runOutboxRelay, 1000);
```

### Replica Safety (`FOR UPDATE SKIP LOCKED`) [#replica-safety-for-update-skip-locked]

`processOutbox()` uses PostgreSQL's row-level locking with `FOR UPDATE SKIP LOCKED`.

* Multiple background workers or server replicas can execute `runOutboxRelay()` concurrently without collision.
* Each pending row is locked by **one worker at a time**, not necessarily processed only once over its lifetime.
* Other workers skip already-locked rows and immediately process the next available records.
* If a worker crashes before finishing, PostgreSQL releases the row lock, allowing other workers to pick it up on the next cycle.

An external publish may succeed before the worker crashes or the database commit fails. The next attempt can publish the same entry again. Wait for durable broker acknowledgement, retry failed batches, and deduplicate downstream using the outbox ID or CloudEvents `source` + `id` (scoped by destination). SQL writes using the supplied `client` can commit atomically with `processed_at`; external effects cannot.

## Retention and Cleanup [#retention-and-cleanup]

Once an outbox entry is processed, its `processed_at` column is set to the current timestamp. Over time, these processed entries accumulate.

Use `cleanupOutbox()` in a scheduled maintenance cron job to delete old processed records:

```typescript
// Delete processed entries older than 7 days (default) in batches of 500
const deletedCount = await eventStore.cleanupOutbox(
  7 * 24 * 60 * 60 * 1000, // 7 days in milliseconds
  500, // Batch size
);

console.log(`Cleaned up ${deletedCount} archived outbox records.`);
```

## Outbox vs. Subscriptions [#outbox-vs-subscriptions]

| Feature         | Transactional Outbox (`processOutbox()`)               | Subscriptions (`subscribe()`)                     |
| :-------------- | :----------------------------------------------------- | :------------------------------------------------ |
| **Model**       | **Competing Consumers** (Distributed Workers)          | **Fan-Out** (Every Subscriber)                    |
| **Target**      | External Brokers (Kafka, NATS, RabbitMQ, SQS)          | In-process consumers, GraphQL, WebSockets, SSE    |
| **Concurrency** | One worker locks a row at a time; publishes may repeat | Every replica/subscriber receives all events      |
| **Locking**     | PostgreSQL `FOR UPDATE SKIP LOCKED`                    | Lock-free streaming via `LISTEN/NOTIFY` + Polling |
| **Storage**     | Dedicated `outbox` table (cleaned up periodically)     | Read directly from `events` table                 |

For building query-optimized read models inside the same PostgreSQL database, see [Projections](/docs/projections). For architectural guidance on choosing between direct streaming and broker relays, see the [Event Streaming & Resilient Consumer Playbook](/docs/playbooks/sse-and-consumer-scaling).


# Projections (/docs/projections)



Projections build **read models** from the global event stream. They transform raw, immutable business events into query-optimized PostgreSQL tables, search indexes, or materialized views.

Each projection tracks its own independent checkpoint position in PostgreSQL and processes events in `globalPosition` order. Sequence allocation order is not transaction commit order; the commit-safe watermark holds back positions until earlier in-flight writes are resolved.

## Projections vs. Snapshots [#projections-vs-snapshots]

| Feature       | Snapshot                               | Projection                                      |
| :------------ | :------------------------------------- | :---------------------------------------------- |
| **Purpose**   | Fast stream replay shortcut            | Queryable read tables, search, analytics        |
| **Scope**     | Single aggregate stream (`stream_id`)  | Global event stream across multiple aggregates  |
| **Storage**   | Generated snapshot event in the stream | Custom database tables (e.g. `order_summaries`) |
| **Execution** | Synchronous or on aggregate append     | Asynchronous batch processing with checkpoint   |
| **Example**   | Cached `Cart` state for `Cart-123`     | Cross-customer sales report table               |

## Defining a Projection [#defining-a-projection]

### Using the `defineProjection` Builder (Recommended) [#using-the-defineprojection-builder-recommended]

For typed projections tied to event-sourced aggregates, use `defineProjection`. It mirrors the `defineAggregate` pattern with type inference, automatic stream prefix filtering, entity ID extraction, and typed handlers:

```typescript
import { defineProjection } from "alvyn";

type OrderEvents = {
  OrderPlaced: { customerId: string; total: number };
  OrderShipped: { trackingNumber: string };
};

export const orderSummaryProjection = defineProjection<OrderEvents>()({
  projectionName: "order-summary",
  streamPrefix: "Order",

  handlers: {
    OrderPlaced: async (data, ctx) => {
      // ctx.client is part of the checkpoint transaction (atomic!)
      await ctx.client.query(
        `INSERT INTO order_summaries (id, customer_id, total, status)
         VALUES ($1, $2, $3, 'placed')
         ON CONFLICT (id) DO UPDATE SET total = $3, status = 'placed'`,
        [ctx.entityId, data.customerId, data.total],
      );
    },
    OrderShipped: async (_data, ctx) => {
      await ctx.client.query(
        `UPDATE order_summaries SET status = 'shipped' WHERE id = $1`,
        [ctx.entityId],
      );
    },
  },
});
```

The builder automatically:

* **Filters by stream prefix and handler type:** Events from unrelated aggregates or event types without a handler (including snapshot events) are skipped before decryption/upcasting.
* **Extracts the entity ID:** Strips the prefix from the stream ID (e.g. `"Order-123"` becomes `"123"`).
* **Provides typed data:** Handlers receive strongly typed event payloads.
* **Provides atomic `ctx.client`:** All SQL operations inside the handler execute in the same transaction as the projection checkpoint update.

### Redacted events after crypto-shredding [#redacted-events-after-crypto-shredding]

An event encrypted with a revoked key is delivered only through `onRedacted`, **not** through its regular typed handler. `event.data` contains the original stored, non-encrypted fields with encrypted paths set to `null`, while `event.redactedPaths` lists those field paths and `event.schemaVersion` identifies the original payload schema. This partial payload is **not upcasted**: an upcaster may require PII that no longer exists. Inspect or validate the remaining data before using it.

```typescript
const issueProjection = defineProjection<IssueEvents>()({
  projectionName: "issues",
  streamPrefix: "Issue",
  handlers: {
    /* handlers for complete events */
  },
  onRedacted: async (event, ctx) => {
    // Example only: handle each historical schema explicitly.
    if (event.type === "IssueCreated" && event.schemaVersion === 1) {
      const data = event.data as { title: string };
      await ctx.client.query(
        `INSERT INTO issues (id, title, assignee) VALUES ($1, $2, NULL)`,
        [ctx.entityId, data.title],
      );
      return;
    }
    throw new Error(
      `Unhandled redacted event: ${event.type} v${event.schemaVersion}`,
    );
  },
});
```

If the projection can safely discard every **handled** redacted event, set `onRedacted: "skip"` explicitly. Typed projections skip event types without a handler before decryption; custom `Projection` implementations may provide `handlesEventType(type)` for the same optimization. Without a policy, processing fails and the checkpoint rolls back rather than silently omitting an event. A missing crypto configuration or a missing key row is an error even with `"skip"`; only **revoked** keys count as intentional redaction. `onRedacted` can also be defined on the raw `Projection` interface with `(event, client)` arguments.

#### `ProjectionHandlerContext` [#projectionhandlercontext]

Each handler receives a context object:

```typescript
interface ProjectionHandlerContext {
  entityId: string; // Aggregate entity ID (e.g. "123")
  streamId: string; // Full stream identifier (e.g. "Order-123")
  globalPosition: bigint; // Global position in the event store
  streamVersion: number; // Version of the stream at this event
  createdAt: Date; // Timestamp when event was recorded
  client: PoolClient; // PostgreSQL transaction client
}
```

### Raw `Projection` Interface [#raw-projection-interface]

For cross-stream projections that listen to multiple aggregate prefixes or custom event shapes, implement the `Projection` interface directly:

```typescript
import type { Projection, StoredEvent } from "alvyn";
import type { PoolClient } from "pg";

const multiStreamProjection: Projection = {
  projectionName: "global-activity-feed",
  async handle(event: StoredEvent, client: PoolClient) {
    if (event.type === "OrderPlaced") {
      const data = event.data as { total: number };
      await client.query(
        `INSERT INTO activity_feed (stream_id, action, amount) VALUES ($1, $2, $3)`,
        [event.streamId, "order_placed", data.total],
      );
    }
  },
};
```

## Running Projections with `runProjection()` [#running-projections-with-runprojection]

Call `eventStore.runProjection(projection, batchSize)` to process the next batch of pending events:

```typescript
// Process up to 500 events in a single batch
const processedCount = await eventStore.runProjection(
  orderSummaryProjection,
  500,
);
console.log(`Processed ${processedCount} events.`);
```

### Background Scheduling Loop [#background-scheduling-loop]

Run your projections continuously in a background worker:

```typescript
async function startProjectionWorker() {
  while (true) {
    try {
      const count = await eventStore.runProjection(orderSummaryProjection, 500);

      // If we processed a full batch, process the next batch immediately
      if (count >= 500) {
        continue;
      }

      // Otherwise wait briefly before polling again
      await new Promise((resolve) => setTimeout(resolve, 1000));
    } catch (err) {
      console.error("Projection execution failed:", err);
      // Wait before retrying on database error
      await new Promise((resolve) => setTimeout(resolve, 5000));
    }
  }
}

startProjectionWorker();
```

## How Projections Work Internally [#how-projections-work-internally]

1. **Checkpoint Tracking:** The projection's progress is tracked in the `projections` table (`last_position` column).
2. **Row Locking:** On each `runProjection()` invocation, Alvyn acquires a row-level lock (`FOR UPDATE`) on the projection's checkpoint row.
3. **Commit-Safe Watermark:** Alvyn queries pending events with `WHERE global_position > last_position AND global_position <= safeWatermark ORDER BY global_position ASC LIMIT batchSize`. The watermark prevents skipping out-of-order commits from concurrent transactions.
4. **Sequential Processing:** Alvyn invokes your handler for each event sequentially.
5. **Atomic Commit:** The read-model updates and the `last_position` checkpoint update commit together in the single PostgreSQL transaction.

<Callout>
  Because the read model queries and the checkpoint update share the same
  transaction via `ctx.client`, projection processing is **exactly-once** within
  PostgreSQL. If the worker crashes or a SQL query fails, the entire batch rolls
  back and the checkpoint remains unchanged.
</Callout>

## Best Practices [#best-practices]

* **Idempotent Handlers:** Use `INSERT ... ON CONFLICT DO UPDATE` or idempotent SQL writes in handlers to remain resilient against rebuilds.
* **External Effects:** HTTP calls, broker publishes, and writes through another connection are not atomic with the checkpoint and may repeat after failure. Use an outbox and downstream deduplication for these effects.
* **Schema and PII:** Complete projection events are decrypted and then upcast. Redacted events retain only public fields in their original schema and use the explicit `onRedacted` policy. Projection handlers can receive PII: restrict access to read models accordingly. Revoking a key does not remove PII already stored in read models; clean those up explicitly.
* **Dedicated Workers:** Run heavy background projections on dedicated worker instances to prevent projection queries from competing with latency-critical API requests.

## Related Topics [#related-topics]

* [Subscriptions](/docs/subscriptions) — Fan-out streaming for GraphQL, WebSockets, and SSE.
* [Transactional Outbox](/docs/outbox) — Relay domain events to Kafka, NATS, or RabbitMQ.
* [Aggregates](/docs/aggregates) — Write-side domain models and event emission.


# Schema Evolution (/docs/schema-evolution)



Upcasters transform old event schemas into newer versions **at read time**. The stored event in the database is **never modified** — transformations happen in memory during `load()`.

## Why Upcasters? [#why-upcasters]

Event sourcing stores events permanently. When the shape of an event changes (e.g., adding a new required field, restructuring nested data), you have two options:

1. **Migrate stored events** — Modify existing rows in the database. This breaks the immutability guarantee and is error-prone at scale.
2. **Upcast at read time** — Transform old events to the new shape when they are loaded. The stored data stays unchanged.

Alvyn uses option 2. Upcasters are pure functions that convert event data from one schema version to the next.

## Defining an Upcaster [#defining-an-upcaster]

An upcaster implements the `Upcaster` interface:

```typescript
import type { Upcaster } from "alvyn";

const orderPlacedV1ToV2: Upcaster = {
  eventType: "OrderPlaced",
  fromSchemaVersion: 1,
  toSchemaVersion: 2,
  upcast(data: { total: number }) {
    return {
      ...data,
      currency: "EUR",
    };
  },
};
```

### Registration [#registration]

Register upcasters with the event store at startup:

```typescript
// Single upcaster
eventStore.registerUpcaster(orderPlacedV1ToV2);

// Multiple upcasters
eventStore.registerUpcasters([orderPlacedV1ToV2, addressV1ToV2]);

// From an aggregate definition
eventStore.registerUpcasters(Order.getUpcasters());
```

## Chained Upcasters [#chained-upcasters]

When an event type has gone through multiple schema versions, register upcasters for each step. They are automatically chained:

```typescript
// v1 -> v2: Added "city" field
const addressV1ToV2: Upcaster = {
  eventType: "AddressChanged",
  fromSchemaVersion: 1,
  toSchemaVersion: 2,
  upcast(data: { street: string }) {
    return { ...data, city: "Unknown" };
  },
};

// v2 -> v3: Restructured to nested object
const addressV2ToV3: Upcaster = {
  eventType: "AddressChanged",
  fromSchemaVersion: 2,
  toSchemaVersion: 3,
  upcast(data: { street: string; city: string }) {
    return {
      address: { street: data.street, city: data.city },
    };
  },
};

eventStore.registerUpcasters([addressV1ToV2, addressV2ToV3]);
```

A v1 event stored as `{ street: "123 Main St" }` is automatically transformed through both upcasters on read, producing `{ address: { street: "123 Main St", city: "Unknown" } }`.

## How Upcasters Work Internally [#how-upcasters-work-internally]

1. Events are stored with a `schema_version` column (default: `1`).
2. The `UpcasterRegistry` maintains a `Map<eventType, Upcaster[]>` sorted by `fromSchemaVersion` ascending.
3. On read, for each event:
   * Look up the upcaster chain for the event type
   * Starting from the stored `schema_version`, walk the chain
   * For each upcaster where `fromSchemaVersion === currentVersion`, apply `upcast()` and advance `currentVersion` to `toSchemaVersion`
   * Events already at the latest version skip upcasters entirely
   * Invalid or duplicate transitions are rejected at registration; missing links in a chain throw on read instead of returning an older schema as though it were current.
4. The database row is **never modified**.

| Stored Version | Upcasters Applied   | Result Version |
| -------------- | ------------------- | -------------- |
| 1              | v1->v2, then v2->v3 | 3              |
| 2              | v2->v3              | 3              |
| 3              | (none)              | 3              |

## Using Upcasters with Aggregates [#using-upcasters-with-aggregates]

Define upcasters in the aggregate definition to keep them co-located with the domain logic:

```typescript
type OrderState = { total: number; currency: string };

const Order = defineAggregate<OrderState, OrderEvents>()({
  streamPrefix: "Order",
  evolve: {
    /* ... */
  },

  upcasters: [
    {
      eventType: "OrderPlaced",
      fromSchemaVersion: 1,
      toSchemaVersion: 2,
      upcast(data: { total: number }) {
        return { ...data, currency: "EUR" };
      },
    },
  ],
});

// At startup: register all aggregate upcasters
eventStore.registerUpcasters(Order.getUpcasters());
```

## Writing New Events with Schema Versions [#writing-new-events-with-schema-versions]

When you create a new schema version, set the `schemaVersion` on new events so they don't get upcasted unnecessarily:

```typescript
await Order.append(eventStore, {
  entityId: orderId,
  expectedVersion: order.version,
  events: [
    {
      type: "OrderPlaced",
      data: { total: 99.99, currency: "USD" },
      schemaVersion: 2,
    },
  ],
});
```

If `schemaVersion` is omitted, it defaults to `1`.

## Best Practices [#best-practices]

1. **Keep upcasters deterministic** — No current date, randomness, external lookups, side effects, or async operations. Derive values from stored data or use a documented fixed default (such as `null` for an unknown historical timestamp).
2. **Never skip versions** — If you have v1, v2, and v3, you need both v1->v2 and v2->v3. Don't create v1->v3 shortcuts.
3. **Co-locate with aggregates** — Define upcasters in the aggregate definition.
4. **Register at startup** — All upcasters must be registered before any read operations.
5. **Test upcaster chains** — Verify that old events are correctly transformed through multiple versions. Missing transitions throw when such an event is read.

## Step-by-Step Migration Guide [#step-by-step-migration-guide]

When you need to change the shape of an existing event type:

### Step 1: Define the upcaster [#step-1-define-the-upcaster]

```typescript
type OrderState = { total: number; currency: string };

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

  upcasters: [
    {
      eventType: "OrderPlaced",
      fromSchemaVersion: 1,
      toSchemaVersion: 2,
      upcast(data: { total: number }) {
        return { ...data, currency: "EUR" };
      },
    },
  ],
});
```

### Step 2: Update the event type map [#step-2-update-the-event-type-map]

```typescript
type OrderEvents = {
  OrderPlaced: { total: number; currency: string }; // v2
};
```

### Step 3: Register upcasters at startup [#step-3-register-upcasters-at-startup]

```typescript
await eventStore.setup();
eventStore.registerUpcasters(Order.getUpcasters());
```

### Step 4: Update new event writes [#step-4-update-new-event-writes]

```typescript
await Order.append(eventStore, {
  entityId: orderId,
  expectedVersion: order.version,
  events: [
    {
      type: "OrderPlaced",
      data: { total: 99.99, currency: "USD" },
      schemaVersion: 2,
    },
  ],
});
```

### Step 5: Deploy [#step-5-deploy]

Upcasters let new readers understand old data; they do not make old readers forward-compatible. Optional chaining cannot repair renamed fields, changed types, or changed business meaning.

1. Deploy readers and upcasters that understand both versions while writers still emit v1. Include projections, subscription consumers, and rollback candidates; complete projection and subscription events upcast automatically, while redacted events and subscriptions with `raw: true` retain the stored schema.
2. Verify old-history replay and mixed-version fixtures, then enable v2 writes only after every required reader is compatible.
3. Keep the upcasters for historical data. After v2 writes begin, rollback only to a reader version that understands v2, or use a deliberately designed compatibility bridge.

## Common Migration Scenarios [#common-migration-scenarios]

### Adding a required field [#adding-a-required-field]

```typescript
{
  eventType: "UserCreated",
  fromSchemaVersion: 1,
  toSchemaVersion: 2,
  upcast(data: { name: string }) {
    return { ...data, email: "unknown@migrated.com" };
  },
}
```

### Renaming a field [#renaming-a-field]

```typescript
{
  eventType: "UserCreated",
  fromSchemaVersion: 1,
  toSchemaVersion: 2,
  upcast(data: { userName: string }) {
    const { userName, ...rest } = data;
    return { ...rest, name: userName };
  },
}
```

### Restructuring nested data [#restructuring-nested-data]

```typescript
{
  eventType: "AddressChanged",
  fromSchemaVersion: 1,
  toSchemaVersion: 2,
  upcast(data: { street: string; city: string }) {
    return { address: { street: data.street, city: data.city } };
  },
}
```

### Changing a field type [#changing-a-field-type]

```typescript
{
  eventType: "PaymentReceived",
  fromSchemaVersion: 1,
  toSchemaVersion: 2,
  upcast(data: { amount: string }) {
    return { ...data, amount: parseFloat(data.amount) };
  },
}
```

### Removing a field [#removing-a-field]

```typescript
{
  eventType: "UserCreated",
  fromSchemaVersion: 1,
  toSchemaVersion: 2,
  upcast(data: { name: string; legacyId: string }) {
    const { legacyId, ...rest } = data;
    return rest;
  },
}
```


# Snapshots (/docs/snapshots)



Snapshots are **domain-defined performance helpers**. They cache calculated state that can be derived from events, then use that cached state as the replay base on future loads.

Unlike projections, snapshots are not separate read models and do not require a worker. A snapshot is stored as a generated event in the same stream it optimizes.

Snapshots are independent from aggregates. They can snapshot aggregate-like state, but they can also snapshot derived state that is not the aggregate state itself.

## When to Use Snapshots [#when-to-use-snapshots]

Use `defineSnapshot` when all of these are true:

* The state can be derived from events in one stream.
* Replaying the full stream becomes expensive over time.
* The cached state is only a replay optimization, not a separately queried read model.

Use projections instead when you need cross-stream views, query tables, filtering, search, or asynchronous read models.

Choose `every` from measured replay latency, payload size, and write/storage cost. There is no universal event-count threshold such as 200 events.

### Snapshots vs Projections [#snapshots-vs-projections]

| Use case                                                                                    | Prefer     | Why                                                                                      |
| ------------------------------------------------------------------------------------------- | ---------- | ---------------------------------------------------------------------------------------- |
| Current balance for one bank account from one `Transaction-{accountId}` stream              | Snapshot   | It speeds up replay of one stream and can be loaded as `latest snapshot + later events`. |
| Dashboard table with all accounts, owners, balances, and risk status                        | Projection | It is a query model that combines and filters data for reads.                            |
| Customer total wealth across many account streams                                           | Projection | It depends on multiple streams, so there is no single same-stream snapshot version.      |
| Current aggregate state after thousands of address changes in one `BankAccount-{id}` stream | Snapshot   | It is still a one-stream replay optimization.                                            |

## Snapshot Relation to Aggregates [#snapshot-relation-to-aggregates]

The relation between aggregates and snapshots is not always `1:1`.

One aggregate stream can have multiple snapshots when different calculations become expensive:

```typescript
const BankAccountBalance = defineSnapshot<
  { balance: number },
  TransactionEvents
>()({
  streamPrefix: Transaction.streamPrefix,
  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),
    }),
  },
});

const BankAccountTransactionCount = defineSnapshot<
  { count: number },
  TransactionEvents
>()({
  streamPrefix: Transaction.streamPrefix,
  snapshotName: "BankAccountTransactionCount",
  every: 100,
  initialState: { count: 0 },
  evolve: {
    Deposit: (state) => ({ count: state.count + 1 }),
    Withdrawal: (state) => ({ count: state.count + 1 }),
  },
});
```

This is why snapshots are defined with `defineSnapshot` instead of only as a `snapshot` option on `defineAggregate`. The snapshot belongs to a stream prefix and a domain calculation, not necessarily to the aggregate state object.

You can still keep the DX clean by grouping related snapshots near the aggregate:

```typescript
export const transactionSnapshots = [
  BankAccountBalance,
  BankAccountTransactionCount,
];
```

## Define a Snapshot [#define-a-snapshot]

```typescript
import { EventStore, defineSnapshot } from "alvyn";

type TransactionEvents = {
  Deposit: { amount: number };
  Withdrawal: { amount: number };
};

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. In this example Alvyn writes `BankAccountBalanceSnapshot` events.

## Register Snapshots on the EventStore [#register-snapshots-on-the-eventstore]

Register snapshot handles once when creating the store. Alvyn then maintains matching snapshots synchronously after public appends.

```typescript
const eventStore = new EventStore({
  pool,
  snapshots: transactionSnapshots,
});
```

When a `Deposit` or `Withdrawal` is appended to `Transaction-{accountId}`, Alvyn checks the registered `BankAccountBalance` snapshot in the same transaction. Once `every` handled source events have been appended after the latest snapshot, Alvyn writes a generated `BankAccountBalanceSnapshot` event immediately.

## Load a Snapshot [#load-a-snapshot]

```typescript
const result = await BankAccountBalance.load(eventStore, accountId);

result.state.balance;
result.version; // true latest stream version, including snapshot events
result.snapshotVersion; // stream version of the snapshot event used
result.replayedEvents; // source events replayed after the snapshot base

// Or within an active transaction:
const txResult = await BankAccountBalance.load(eventStore, accountId, {
  client,
});
```

Loading follows this process:

1. Read the latest matching snapshot event in `Transaction-{accountId}`.
2. Use its saved state as the replay base, or `initialState` when none exists.
3. Replay only later source events with handlers in `evolve`.

Loading is read-only. Snapshot events are generated during appends when the snapshot is registered on the `EventStore`.

## Encrypt Sensitive Snapshot State [#encrypt-sensitive-snapshot-state]

Snapshot encryption is **not inherited** from an aggregate or its source events. Snapshot maintenance replays decrypted data, so an unencrypted snapshot can persist PII in plaintext even when every source event is encrypted. Configure every sensitive path in the snapshot state explicitly:

```typescript
import { defineSnapshot } from "alvyn";

type UserEvents = {
  UserRegistered: { name: string; email: string };
  UserRenamed: { name: string };
};

const UserProfile = defineSnapshot<
  { name: string; email: string },
  UserEvents
>()({
  streamPrefix: "User",
  snapshotName: "UserProfile",
  every: 50,
  initialState: { name: "", email: "" },
  evolve: {
    UserRegistered: (_state, event) => ({
      name: event.data?.name ?? "",
      email: event.data?.email ?? "",
    }),
    UserRenamed: (state, event) => ({
      ...state,
      name: event.data?.name ?? "",
    }),
  },
  encryption: {
    cryptoKeyId: (entityId) => `user:${entityId}`,
    encryptedFields: ["name", "email"],
  },
});
```

Register `UserProfile` in `EventStore.snapshots`, configure secrets, and create the same `user:${entityId}` key used for the source events before writing. Revoking that key tombstones the encrypted snapshot as well. When the latest snapshot is tombstoned, loading falls back to `initialState` and replays source history with null-safe handlers. This does not erase earlier plaintext snapshots or downstream copies; address those separately.

## Reserved Event Names [#reserved-event-names]

Event types ending in `Snapshot` are reserved for Alvyn-generated snapshot events. Public `EventStore.append()` and aggregate append helpers reject user-supplied event names such as `OrderSnapshot`.

This prevents domain events from colliding with generated snapshot events in the same stream.

## Storage Model [#storage-model]

Snapshots are stored in the normal `{schema}.events` table. There is no `snapshots` table and no separate migration.

Each generated snapshot appends a new row; previous snapshots remain in history. Storage grows with snapshot frequency and with the number of snapshot definitions per stream, rather than staying at one row per stream. `Aggregate.load()` still replays domain history; use the snapshot handle's `load()` to benefit from its cached state.

Because snapshot events share the source stream, they also advance `stream_version`. Use `eventStore.getStreamVersion()` or aggregate `load().version` when calculating expected versions after snapshot loads.


# Subscriptions (/docs/subscriptions)



`EventStore.subscribe()` is Alvyn's **fan-out** primitive for consumers that observe events independently. It returns an `AsyncIterable<SubscriptionEvent>` that first catches up on matching history and then seamlessly transitions to tailing new events as they commit in PostgreSQL.

Unlike the transactional outbox, subscriptions are **not** competing consumers. Every subscriber (e.g. every replica or connected client) receives every matching event and maintains its own in-memory cursor.

## Basic Usage [#basic-usage]

```typescript
const controller = new AbortController();

for await (const event of eventStore.subscribe({
  subject: "Order-",
  recursive: true,
  eventTypes: ["OrderPlaced", "OrderShipped"],
  signal: controller.signal,
})) {
  await handleOrderEvent(event);
  // Persist event.globalPosition after successful processing
}

// Stops iteration and releases the PostgreSQL LISTEN connection
controller.abort();
```

The iterator is ordered by `globalPosition`, delivers at least once, and is independent for each subscriber. Consumers should be idempotent and persist their cursor after handling an event.

## Catch-up and Live Tailing [#catch-up-and-live-tailing]

The subscription reads historical events first, then transitions to live tailing on the same iterator:

1. **Catch-up phase:** Queries historical events matching your filter ordered by `global_position ASC` in batches of `batchSize`.
2. **Live tailing phase:** Alvyn automatically registers a PostgreSQL `LISTEN` client on commit notifications (`NOTIFY`). When new events append, the listener wakes up immediately.
3. **Polling fallback:** A background timer (default `pollIntervalMs: 1000`) guarantees delivery even if a database notification is dropped or during network reconnection.

### Commit-Safe Watermark [#commit-safe-watermark]

The read side uses a **commit-safe watermark**. This prevents a concurrent transaction that commits later with a lower `globalPosition` from being skipped by a live cursor consumer:

```
Transaction A (position 101) ----------> commits at t=2
Transaction B (position 100) ---------------> commits at t=3 (delayed)
```

Without a watermark, a subscriber tailing at `t=2` would read `101`, advance its cursor past `101`, and permanently miss `100` when `B` commits at `t=3`. Alvyn calculates the safe watermark dynamically so consumers never miss out-of-order commits.

## Filters and Options [#filters-and-options]

```typescript
interface SubscribeOptions {
  subject?: string; // CloudEvents subject (streamId), e.g. "Order-123" or "Order-"
  recursive?: boolean; // When true, matches prefix (e.g. "Order-" matches "Order-1", "Order-2")
  eventTypes?: string[]; // Restrict delivery to specific event types
  lowerBound?: { id: string; type?: "exclusive" | "inclusive" }; // Resume cursor
  signal?: AbortSignal; // Abort signal to stop stream and release DB connection
  batchSize?: number; // Batch size for catch-up queries (default: 500)
  pollIntervalMs?: number; // Polling fallback cadence in ms (default: 1000)
}
```

`subject` and `eventTypes` can be combined. Without a subject filter, the subscription observes matching events from all streams in the store.

## Resuming with a Cursor [#resuming-with-a-cursor]

To resume processing after a worker restart or server disconnect, pass `lowerBound.id` set to the string representation of the last processed `globalPosition`:

```typescript
let lastProcessedPosition = "42";

for await (const event of eventStore.subscribe({
  subject: "Order-",
  recursive: true,
  lowerBound: {
    id: lastProcessedPosition,
    type: "exclusive", // Default: start with the event immediately AFTER 42
  },
})) {
  await handle(event);
  lastProcessedPosition = event.globalPosition.toString();
}
```

Use `type: "inclusive"` if the event at the cursor should be delivered again (e.g. if the previous consumer crashed mid-processing before saving its state).

## Aggregate Subscriptions [#aggregate-subscriptions]

An aggregate handle provides a strongly typed subscription for a specific entity stream. Stream subject derivation and snapshot event filtering happen automatically:

```typescript
import { defineAggregate } from "alvyn";

type OrderState = { status: "placed" | "shipped" };
type OrderEvents = {
  OrderPlaced: { total: number };
  OrderShipped: { trackingNumber: string };
};

const Order = defineAggregate<OrderState, OrderEvents>()({
  streamPrefix: "Order",
  evolve: {
    OrderPlaced: () => ({ status: "placed" }),
    OrderShipped: () => ({ status: "shipped" }),
  },
});

// Strongly typed: event.type and event.data match OrderEvents
for await (const event of Order.subscribe({
  eventStore,
  entityId: "order-123",
  options: { eventTypes: ["OrderShipped"] },
})) {
  if (event.type === "OrderShipped") {
    console.log(event.type, event.data.trackingNumber);
  }
}
```

## Real-time Integration Use Cases [#real-time-integration-use-cases]

### GraphQL Subscriptions [#graphql-subscriptions]

Because `subscribe()` returns a standard `AsyncIterable`, it integrates directly with GraphQL subscription resolvers (such as GraphQL Yoga, Apollo Server, or Mercurius):

```typescript
const resolvers = {
  Subscription: {
    orderEvents: {
      subscribe: (_parent, args, context) => {
        const ac = new AbortController();

        // Stream matching events to this specific connected client
        return mapAsyncIterator(
          context.eventStore.subscribe({
            subject: "Order-",
            recursive: true,
            eventTypes: args.types,
            signal: ac.signal,
          }),
          (event) => ({ orderEvents: event }),
        );
      },
    },
  },
};
```

### Server-Sent Events (SSE) Endpoint [#server-sent-events-sse-endpoint]

You can expose a real-time HTTP event stream where the standard `Last-Event-ID` header acts as the resumable cursor:

```typescript
import type { IncomingMessage, ServerResponse } from "node:http";

async function sseHandler(req: IncomingMessage, res: ServerResponse) {
  res.writeHead(200, {
    "content-type": "text/event-stream",
    "cache-control": "no-cache",
    connection: "keep-alive",
  });

  const ac = new AbortController();
  res.on("close", () => ac.abort());

  const lastEventId = req.headers["last-event-id"] as string | undefined;

  for await (const event of eventStore.subscribe({
    subject: "Order-",
    recursive: true,
    lowerBound: lastEventId
      ? { id: lastEventId, type: "exclusive" }
      : undefined,
    signal: ac.signal,
  })) {
    // The SSE `id:` doubles as the resume cursor for automatic client reconnects
    res.write(`id: ${event.globalPosition}\n`);
    res.write(
      `data: ${JSON.stringify({ ...event, globalPosition: event.globalPosition.toString() })}\n\n`,
    );
  }
}
```

This minimal endpoint serializes `globalPosition` as a decimal string because JSON cannot encode `bigint`. Before deployment, add authorization, cursor validation, error cleanup, and bounded backpressure handling. For implementation sketches and their operational limits, see the [Event Streaming & Resilient Consumer Playbook](/docs/playbooks/sse-and-consumer-scaling).

## Encryption and Schema Evolution [#encryption-and-schema-evolution]

Subscriptions upcast **complete, unencrypted** event payloads by default, including during catch-up. Set `raw: true` to receive the stored schema instead. Events with encrypted fields are always delivered as `RedactedSubscriptionEvent` in their original schema, regardless of whether the key is active or revoked. Neither mode decrypts GDPR-encrypted fields.

* The payload starts with the stored `data` column, with inaccessible encrypted fields represented as `null` (whether the key is active or revoked). Ciphertext lives separately in `encrypted_data` and is not returned by subscriptions. These redacted events carry `redacted: true`, `redactedPaths`, and `schemaVersion`. They are **never upcasted**: a registered upcaster may require encrypted fields. Check for `"redacted" in event` before treating an aggregate subscription event as a fully typed payload.
* `raw: true` skips upcasting complete events and returns the stored schema format; `Order.subscribe({ options: { raw: true }, ... })` does not promise typed current-schema payloads. Redacted events remain redacted in either mode.
* Aggregate event-map types do not validate runtime payloads or restore encrypted fields. `eventTypes` filters also do not narrow the TypeScript union; check `event.type` before accessing event-specific fields.

When decrypted and upcasted domain state is required, use `Order.load()` or `eventStore.load()`.

## Subscription vs. Outbox [#subscription-vs-outbox]

| Feature                | Subscriptions (`subscribe()`)               | Transactional Outbox (`processOutbox()`)                     |
| :--------------------- | :------------------------------------------ | :----------------------------------------------------------- |
| **Pattern**            | In-process Fan-out                          | Competing Consumers (Distributed Fleet)                      |
| **Delivery Target**    | Connected clients, WebSockets, GraphQL, SSE | Message Brokers (Kafka, NATS, RabbitMQ, SQS)                 |
| **Scale Behavior**     | Every replica receives every event          | One worker holds a row lock at a time; retries can republish |
| **State Tracking**     | Client in-memory cursor (`lowerBound`)      | PostgreSQL table lock (`FOR UPDATE SKIP LOCKED`)             |
| **Delivery Guarantee** | At-least-once per subscriber                | At-least-once to external message broker                     |

Use `subscribe()` when you need real-time streaming to all replicas or clients. Use the [Transactional Outbox](/docs/outbox) when publishing fleet-wide messages to external brokers.


# Aggregate Design & Stream Boundary Playbook (/docs/playbooks/aggregate-design-and-stream-boundaries)





When engineering event-sourced systems with **Alvyn** and **PostgreSQL**, determining where to draw aggregate boundaries is the single most critical architectural choice.

An aggregate is **not** a database table, a document store record, or an arbitrary bucket of JSON. In Domain-Driven Design (DDD) and Event Sourcing, an aggregate is an **invariant boundary** and a **unit of transactional consistency**.

Designing boundaries too small causes unbounded stream replays, connection pool exhaustion, and non-atomic multi-stream updates. Designing them too large leads to optimistic concurrency bottlenecks and bloated state rehydration.

This playbook provides an architectural decision framework, stream partitioning strategies, an anti-pattern catalog derived from real-world production incidents, and concrete TypeScript blueprints for Alvyn.

***

## 1. The Fundamental Laws of Aggregate Design [#1-the-fundamental-laws-of-aggregate-design]

To design resilient event-sourced systems, adhere to three foundational principles:

### Law 1: The Boundary of Transactional Atomicity [#law-1-the-boundary-of-transactional-atomicity]

In Alvyn, `store.append({ streamId, events, expectedVersion })` appends a batch to one stream atomically. Several streams can also share one PostgreSQL transaction through `store.withTransaction()` and low-level `append(input, { client })` calls:

```typescript
await store.withTransaction(async (client) => {
  await store.append(
    {
      streamId: "Account-A",
      expectedVersion: versionA,
      events: [{ type: "Debited", data: { amount: 100 } }],
    },
    { client },
  );
  await store.append(
    {
      streamId: "Account-B",
      expectedVersion: versionB,
      events: [{ type: "Credited", data: { amount: 100 } }],
    },
    { client },
  );
});
```

This is a local database transaction, not distributed two-phase commit. Validate invariants, use OCC on every participating stream, and acquire stream locks in a consistent order. Aggregate append helpers do not accept `{ client }`; use the low-level API for this pattern and explicitly include any encryption settings. A single aggregate remains the simplest boundary for tightly coupled invariants, but it is not an API requirement for atomicity.

### Law 2: Invariants Require In-Memory State [#law-2-invariants-require-in-memory-state]

A domain invariant is a rule that must always remain true (e.g., &#x2A;"Total breaks cannot exceed gross working hours"&#x2A;, &#x2A;"An order cannot be cancelled once shipped"*).
To validate an invariant before appending an event, the **Decider** must inspect the current aggregate state. If checking an invariant requires querying multiple streams across the database, the boundary is misaligned.

### Law 3: Address Aggregate Roots, Never Child Entities [#law-3-address-aggregate-roots-never-child-entities]

Commands must be directed to the **Aggregate Root**. Unterlying child entities (e.g., an individual time entry inside a monthly timesheet, or an item inside an order) do not have independent stream identity.

```
┌─────────────────────────────────────────────────────────────┐
│ Aggregate Root: Timesheet (timesheet-user123:2026-09)        │
│                                                             │
│   ┌───────────────────────────┐ ┌─────────────────────────┐ │
│   │ Child Entity: TimeEntry A │ │ Child Entity: Entry B   │ │
│   │ - date: 2026-09-02        │ │ - date: 2026-09-15      │ │
│   │ - breaks: [Break 1]       │ │ - breaks: []            │ │
│   └───────────────────────────┘ └─────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
```

Clients address the Aggregate Root using a deterministic routing key (such as `userId` and `date`). Routing selects one stream without discovery; loading still costs a replay of its history.

***

## 2. Stream Partitioning Strategies [#2-stream-partitioning-strategies]

Choosing how to partition event streams determines your system's scalability, concurrency limits, and query complexity.

### Comparison Matrix [#comparison-matrix]

| Dimension                 | 1. Fine-Grained (Item Stream)                     | 2. Time-Bucketed (Cycle Stream)                       | 3. Infinite Lifetime (Entity Stream)          |
| :------------------------ | :------------------------------------------------ | :---------------------------------------------------- | :-------------------------------------------- |
| **Stream Key Format**     | `item-${itemId}`                                  | `parent-${parentId}:${period}`                        | `entity-${entityId}`                          |
| **Real-World Examples**   | Individual calendar entry, comment, sensor ping   | Monthly timesheet, daily cash register, fiscal ledger | Customer account, subscription, shopping cart |
| **Stream Event Count**    | Very low (1–5 events)                             | Controlled (30–150 events)                            | Unbounded (thousands over years)              |
| **Batch Atomicity**       | Requires a shared transaction client              | Single-stream batch append                            | Single-stream batch append                    |
| **Read Complexity**       | Cross-item queries need discovery or a read model | One targeted stream; replay cost varies               | One targeted stream; replay cost grows        |
| **Concurrency Conflicts** | Very low (isolated per item)                      | Low (per-user / per-period)                           | High (every action competes for stream)       |
| **Snapshot Requirement**  | Usually unnecessary for short histories           | Measure replay cost                                   | Consider when measured replay is too costly   |

***

### Strategy A: Fine-Grained Streams (`item-${id}`) [#strategy-a-fine-grained-streams-item-id]

Each entity owns its own stream.

#### When to Use [#when-to-use]

* Entities with completely independent lifecycles.
* High concurrent writes from different actors on sibling entities (e.g., comments on a viral post).
* When entities are rarely or never queried as an aggregated batch in the write model.

#### Trade-Offs & Pitfalls [#trade-offs--pitfalls]

* **The Scan Trap:** Querying "all items for user X this month" requires either listing all streams (`store.listStreams({ prefix })`) or maintaining a relational read model.
* **Multi-Item Coordination:** Independent appends are not atomic together. Use a shared PostgreSQL transaction client and OCC on each stream when a command spans items.

***

### Strategy B: Time-Bucketed Streams (`entity-${id}:${yearMonth}`) [#strategy-b-time-bucketed-streams-entity-idyearmonth]

Group time-series items into natural domain cycles (e.g., month, week, or fiscal year).

#### When to Use [#when-to-use-1]

* Workspaces with cyclical reporting or closing periods (e.g., payroll timesheets, expense reports, billing cycles).
* When users frequently batch-create, adjust, or override entries for a period.
* When period totals (e.g., total hours worked, monthly cap) represent hard domain invariants.

#### Trade-Offs & Pitfalls [#trade-offs--pitfalls-1]

* **Boundary Transitions:** Operations that span the boundary (e.g., a night shift starting at 22:00 on October 31 and ending at 06:00 on November 1) must follow a clear domain rule (e.g., attribution by start timestamp).
* **Concurrency Scope:** Concurrent writes to different days within the same month share the same stream version. For single-user domains (like personal timesheets), this contention is negligible.

***

### Strategy C: Infinite Lifetime Streams (`entity-${id}`) [#strategy-c-infinite-lifetime-streams-entity-id]

A single stream represents the entire lifetime of an aggregate.

#### When to Use [#when-to-use-2]

* Entities with a clear lifecycle end (e.g., an Order: Placed $\rightarrow$ Paid $\rightarrow$ Shipped $\rightarrow$ Delivered).
* Low-frequency state updates over time (e.g., user profiles, customer master data).

#### Trade-Offs & Pitfalls [#trade-offs--pitfalls-2]

* **Rehydration Degradation:** Replaying an aggregate with thousands of events increases request latency and CPU overhead.
* **Mitigation:** Consider Alvyn snapshots (`defineSnapshot`) when measurements justify caching every $N$ handled events; read through the snapshot handle to use that cache.

***

## 3. Anti-Pattern Catalog from the Trenches [#3-anti-pattern-catalog-from-the-trenches]

The following failure modes illustrate real-world anti-patterns identified in production event-sourced systems and security reviews.

### Anti-Pattern 1: The "ListStreams on Request" Trap [#anti-pattern-1-the-liststreams-on-request-trap]

<Callout type="error">
  **The Bug:** Listing all streams with `store.listStreams({prefix})` on user
  read requests and replaying each stream concurrently with `Promise.all()`.
</Callout>

```typescript
// ❌ ANTI-PATTERN: Scans entire database history on every HTTP GET
export async function getTimesheetForWeek(
  userId: string,
  start: Date,
  end: Date,
) {
  // Lists EVERY time-entry stream across all users and all years!
  const streamIds = await store.listStreams({ prefix: "time-entry" });

  // Unbounded Promise.all fan-out: 50,000 streams = 50,000 database queries
  const allEntries = await Promise.all(
    streamIds.map((id) =>
      timeEntryAggregate.load(store, id.slice("time-entry-".length)),
    ),
  );

  // In-memory filtering after exhausting the database connection pool
  return allEntries.filter(
    (e) =>
      e.state &&
      e.state.userId === userId &&
      e.state.date >= start &&
      e.state.date <= end,
  );
}
```

#### Why It Fails [#why-it-fails]

* **Complexity:** $O(N)$ where $N$ represents all events ever recorded in company history.
* **Resource Exhaustion:** Rapidly exhausts the PostgreSQL connection pool, blocks the event loop, and causes cascading timeouts across unrelated services.

#### The Fix [#the-fix]

Partition by user and period (`timesheet-${userId}:${yearMonth}`). A week query touches at most **one or two streams**:

```typescript
// Target one month directly; a week spanning months needs both buckets
export async function getTimesheetForWeek(userId: string, yearMonth: string) {
  const { state } = await timesheetAggregate.load(
    store,
    `${userId}:${yearMonth}`,
  );
  return Object.values(state?.entries ?? {});
}
```

***

### Anti-Pattern 2: The Partial Batch Disaster (Dirty Overrides) [#anti-pattern-2-the-partial-batch-disaster-dirty-overrides]

<Callout type="error">
  **The Bug:** Discarding existing entries in a loop before appending
  replacement entries sequentially across individual streams.
</Callout>

```typescript
// ❌ ANTI-PATTERN: Non-atomic multi-stream updates
export async function overrideWeekEntries(
  userId: string,
  newEntries: EntryInput[],
) {
  const existingEntries = await getExistingEntries(userId);

  // Phase 1: Discard old entries immediately
  for (const old of existingEntries) {
    await store.append({
      streamId: `time-entry-${old.id}`,
      expectedVersion: 0,
      events: [{ type: "TimeEntryDiscarded", data: {} }],
    });
  }

  // Phase 2: Insert new entries one by one
  for (const next of newEntries) {
    // If this fails (network drop, validation error, process crash):
    // Old entries are already discarded, replacement entries are half-persisted.
    // Result: IRREVERSIBLE DATA LOSS.
    await store.append({
      streamId: `time-entry-${next.id}`,
      expectedVersion: -1,
      events: [{ type: "TimeEntryBooked", data: next }],
    });
  }
}
```

#### The Fix [#the-fix-1]

Consolidate the batch inside a single Aggregate Root. The Decider computes all discard and book events in memory, and Alvyn appends them in **one atomic database commit**:

```typescript
// ✅ BEST PRACTICE: 100% atomic batch append
export async function overrideWeekEntries(
  userId: string,
  yearMonth: string,
  newEntries: TimesheetEntry[],
) {
  const entityId = `${userId}:${yearMonth}`;
  const aggregate = await timesheetAggregate.load(store, entityId);

  // In-memory pure function generates ALL events (discards + bookings)
  const state = aggregate.state ?? { userId, yearMonth, entries: {} };
  const events = decideBatchOverride(state, newEntries);
  if (events.length === 0) return;

  // Single PostgreSQL commit: All succeed or none succeed
  await timesheetAggregate.append(store, {
    entityId,
    expectedVersion: aggregate.version === 0 ? -1 : aggregate.version,
    events,
  });
}
```

***

### Anti-Pattern 3: "CRUD over Event Sourcing" (Anemic Property Bags) [#anti-pattern-3-crud-over-event-sourcing-anemic-property-bags]

<Callout type="warn">
  **The Symptom:** Emitting generic `EntityUpdated` events containing raw
  dictionaries of modified fields instead of intention-revealing domain events.
</Callout>

```typescript
// ❌ ANEMIC: Generic property-bag update
{
  type: "TimeEntryUpdated",
  data: {
    breaks: [{ startTime: "12:00", durationMinutes: 30 }],
    durationMinutes: 450
  }
}
```

#### Why It Fails [#why-it-fails-1]

* **Audit Loss:** Regulators, compliance auditors, and downstream projectors cannot determine *what* happened. Did the user take a break? Correct a typo? Shorten work hours?
* **Concurrent Invariant Breaches:** If two clients submit partial property updates, applying them leads to inconsistent states (e.g., negative net duration).

#### The Fix [#the-fix-2]

Emit specific, intention-revealing domain events:

```typescript
// ✅ INTENTION-REVEALING DOMAIN EVENTS
// 1. When a break is recorded:
{
  type: "TimeEntryBreakRecorded",
  data: { entryId: "e1", breakId: "b1", startTime: "12:00", durationMinutes: 30 }
}

// 2. When working hours are corrected:
{
  type: "TimeEntryWorkingHoursCorrected",
  data: { entryId: "e1", startTime: "08:00", endTime: "16:30", durationMinutes: 480 }
}
```

***

### Anti-Pattern 4: The False Hope of Snapshots [#anti-pattern-4-the-false-hope-of-snapshots]

<Callout type="warn">
  **The Misconception:** Believing that adding snapshots to fine-grained item
  streams resolves query performance problems.
</Callout>

Snapshots cache the materialized state of a **single stream** to prevent replaying long event histories (e.g., replaying event 1 to 1,000).

Snapshots **do not help** when your query needs to aggregate data across hundreds or thousands of different streams. Querying 1,000 streams with snapshots still performs 1,000 database read operations.

* Use **Time-Bucketing** to group related events into a single stream.
* Use **CQRS Projections** (`defineProjection`) to index cross-stream search queries into relational tables.

***

## 4. Read-Side Strategy: Replay vs. Snapshots vs. Projections [#4-read-side-strategy-replay-vs-snapshots-vs-projections]

Choose the simplest query strategy that fulfills your performance and consistency requirements:

### Strategy Comparison [#strategy-comparison]

| Strategy                             | When to Choose                                                                       | Storage Cost                 | Read Latency                      |
| :----------------------------------- | :----------------------------------------------------------------------------------- | :--------------------------- | :-------------------------------- |
| **In-Memory Replay**                 | Reading single aggregates or bounded cycles (e.g. current month timesheet).          | Zero extra storage           | Depends on history and handlers   |
| **Snapshots (`defineSnapshot`)**     | Streams whose measured replay cost justifies caching.                                | Append-only snapshot history | Latest snapshot plus recent delta |
| **Projections (`defineProjection`)** | Complex cross-aggregate queries: company-wide dashboards, status filters, SQL joins. | Relational table storage     | Single indexed SQL query          |

***

## 5. End-to-End Blueprint: Time-Bucketed Timesheet Aggregate [#5-end-to-end-blueprint-time-bucketed-timesheet-aggregate]

Below is a complete implementation blueprint showing a time-bucketed aggregate in Alvyn with deterministic routing and atomic batch mutations.

<Tabs items="[&#x22;Events & State&#x22;, &#x22;Aggregate Definition&#x22;, &#x22;Decider Logic&#x22;, &#x22;Service Handler&#x22;]">
  <Tab value="Events & State">
    ```typescript
    // schema.ts
    export interface TimeEntryBreak {
      id: string;
      startTime?: string | null;
      endTime?: string | null;
      durationMinutes: number;
    }

    export interface TimesheetEntry {
      id: string;
      userId: string;
      date: string; // YYYY-MM-DD
      startTime?: string | null;
      endTime?: string | null;
      durationMinutes: number;
      breaks: TimeEntryBreak[];
      note?: string | null;
      billable: boolean;
    }

    export interface TimesheetState {
      userId: string;
      yearMonth: string; // YYYY-MM
      entries: Record<string, TimesheetEntry>;
    }

    export type TimesheetEvents = {
      TimeEntryBooked: TimesheetEntry;
      TimeEntryBreakRecorded: { entryId: string; break: TimeEntryBreak };
      TimeEntryWorkingHoursCorrected: { entryId: string; startTime?: string | null; endTime?: string | null; durationMinutes: number };
      TimeEntryDiscarded: { entryId: string };
    };

    export type TimesheetEvent = {
      [K in keyof TimesheetEvents]: { type: K; data: TimesheetEvents[K] };
    }[keyof TimesheetEvents];
    ```
  </Tab>

  <Tab value="Aggregate Definition">
    ```typescript
    // aggregate.ts
    import { defineAggregate } from "alvyn";
    import type { TimesheetState, TimesheetEvents } from "./schema";

    export const timesheetAggregate = defineAggregate<TimesheetState, TimesheetEvents>()({
      streamPrefix: "timesheet",
      evolve: {
        TimeEntryBooked: (state, event) => {
          const base = state ?? { userId: "", yearMonth: "", entries: {} };
          if (!event.data) return base;
          return {
            userId: event.data.userId,
            yearMonth: event.data.date.slice(0, 7),
            entries: { ...base.entries, [event.data.id]: event.data },
          };
        },
        TimeEntryBreakRecorded: (state, event) => {
          const data = event.data;
          if (!data || !state?.entries[data.entryId]) return state;
          const entry = state.entries[data.entryId];
          return { ...state, entries: { ...state.entries, [data.entryId]: {
            ...entry,
            breaks: [...entry.breaks, data.break],
            durationMinutes: entry.durationMinutes - data.break.durationMinutes,
          } } };
        },
        TimeEntryWorkingHoursCorrected: (state, event) => {
          const data = event.data;
          if (!data || !state?.entries[data.entryId]) return state;
          return { ...state, entries: { ...state.entries, [data.entryId]: {
            ...state.entries[data.entryId],
            startTime: data.startTime,
            endTime: data.endTime,
            durationMinutes: data.durationMinutes,
          } } };
        },
        TimeEntryDiscarded: (state, event) => {
          if (!event.data || !state) return state;
          const entries = { ...state.entries };
          delete entries[event.data.entryId];
          return { ...state, entries };
        },
      },
    });

    ```
  </Tab>

  <Tab value="Decider Logic">
    ```typescript
    // decider.ts
    import type { TimesheetState, TimesheetEvent, TimesheetEntry } from "./schema";

    function calculateDuration(start: string, end: string): number {
      const minutes = (time: string) => {
        if (!/^([01]\d|2[0-3]):[0-5]\d$/.test(time)) throw new Error("Invalid time");
        const [hours, mins] = time.split(":").map(Number);
        return hours * 60 + mins;
      };
      const difference = minutes(end) - minutes(start);
      return difference < 0 ? difference + 24 * 60 : difference;
    }

    export function decideBookTime(
      state: TimesheetState,
      entry: TimesheetEntry
    ): TimesheetEvent[] {
      // Invariant 1: Ensure entry belongs to this month's bucket
      if (entry.userId !== state.userId || entry.date.slice(0, 7) !== state.yearMonth) {
        throw new Error(`Entry date ${entry.date} does not belong to bucket ${state.yearMonth}`);
      }

      if (!Number.isFinite(entry.durationMinutes) || entry.durationMinutes < 0 ||
          entry.breaks.some((b) => !Number.isFinite(b.durationMinutes) || b.durationMinutes < 0)) {
        throw new Error("Durations must be finite and non-negative.");
      }

      // Invariant 2: Gross - Breaks = Net duration check
      if (entry.startTime && entry.endTime) {
        const grossMinutes = calculateDuration(entry.startTime, entry.endTime);
        const breakMinutes = entry.breaks.reduce((acc, b) => acc + b.durationMinutes, 0);
        if (grossMinutes - breakMinutes !== entry.durationMinutes) {
          throw new Error("Duration must equal gross time minus total break time.");
        }
      }

      return [{ type: "TimeEntryBooked", data: entry }];
    }

    export function decideBatchOverride(
      state: TimesheetState,
      newEntries: TimesheetEntry[]
    ): TimesheetEvent[] {
      const events: TimesheetEvent[] = [];

      // 1. Identify distinct days touched by the new batch
      const touchedDates = new Set(newEntries.map((e) => e.date));

      // 2. Discard existing entries for those specific days
      for (const [id, existing] of Object.entries(state.entries)) {
        if (touchedDates.has(existing.date)) {
          events.push({ type: "TimeEntryDiscarded", data: { entryId: id } });
        }
      }

      // 3. Book replacement entries
      for (const entry of newEntries) {
        events.push(...decideBookTime(state, entry));
      }

      return events;
    }
    ```
  </Tab>

  <Tab value="Service Handler">
    ```typescript
    // service.ts
    import { timesheetAggregate } from "./aggregate";
    import { decideBatchOverride, decideBookTime } from "./decider";
    import type { EventStore } from "alvyn";
    import type { TimesheetEntry } from "./schema";

    export class TimesheetService {
      constructor(private eventStore: EventStore) {}

    async trackBatch(
    userId: string,
    yearMonth: string,
    entries: TimesheetEntry[],
    override: boolean,
    ) {
    if (entries.length > 31) throw new Error("Batch exceeds 31 entries.");
    if (!/^\d{4}-(0[1-9]|1[0-2])$/.test(yearMonth)) throw new Error("Invalid month");
    if (new Set(entries.map((e) => e.id)).size !== entries.length) {
    throw new Error("Duplicate entry IDs in batch.");
    }

        const entityId = `${userId}:${yearMonth}`;
        const aggregate = await timesheetAggregate.load(this.eventStore, entityId);
        const state = aggregate.state ?? { userId, yearMonth, entries: {} };
        for (const entry of entries) {
          const existing = state.entries[entry.id];
          if (existing && (!override || existing.date !== entry.date)) {
            throw new Error("Entry ID already used.");
          }
        }
        const events = override
          ? decideBatchOverride(state, entries)
          : entries.flatMap((e) => decideBookTime(state, e));
        if (events.length === 0) return;

        await timesheetAggregate.append(this.eventStore, {
          entityId,
          expectedVersion: aggregate.version === 0 ? -1 : aggregate.version,
          events,
        });

    }
    }

    ```
  </Tab>
</Tabs>

The aggregate derives the `timesheet-` prefix itself: pass only `userId:yearMonth` as `entityId`. Empty streams load as `null`; the service initializes command state and uses `-1` for create-only OCC (`0` disables the check). This blueprint covers batch booking/replacement, not authorization, full calendar validation, or commands for break/hour corrections. Validate those commands before emitting events; the time helper assumes wall-clock minutes and does not model DST or shifts lasting 24 hours or more.

***

## 6. Edge Cases & Boundary Transitions [#6-edge-cases--boundary-transitions]

### Midnight & Month-End Rollovers [#midnight--month-end-rollovers]

What happens when an entry spans across midnight at the end of a time bucket (e.g., October 31, 22:00 to November 1, 06:00)?

In event sourcing, adopt **Shift-Start Attribution** (the industry standard in ERP and payroll systems):

1. **Rule:** An unbroken shift belongs to the date and time bucket in which it **started**.
2. **Stream Allocation:** The shift starting on October 31 at 22:00 is recorded in `timesheet-${userId}:2026-10`.
3. **Query Coverage:**
   * A monthly report for October captures all 8 hours of the shift.
   * For high-precision range queries overlapping bucket boundaries (e.g., November 1–7), the query inspects whether the final day of the preceding bucket contains a shift that extends past midnight.

***

## 7. Architectural Review Checklist [#7-architectural-review-checklist]

Before approving or deploying an aggregate design in Alvyn, verify every point on this checklist:

* [ ] **Atomicity:** Are coupled events appended in one stream batch, or do all low-level appends share a transaction client with OCC on each stream?
* [ ] **Deterministic Routing ($O(1)$):** Can the service derive the `streamId` directly from command inputs (e.g. `userId` + `date.slice(0, 7)`) without querying the database?
* [ ] **No Unbounded Scans:** Have you eliminated calls to `store.listStreams()` inside synchronous request/response HTTP handlers?
* [ ] **Discrete Domain Events:** Are events named after business facts (`TimeEntryBreakRecorded`) rather than generic CRUD mutations (`EntityUpdated`)?
* [ ] **Batch Size Constraints:** Does the API enforce hard upper bounds on batch array inputs before touching the event store?
* [ ] **Optimistic Locking:** Does `store.append()` specify `expectedVersion` to prevent lost updates from concurrent writers?
* [ ] **Decoupled Search:** Are complex filters, cross-user rollups, and full-text searches delegated to CQRS read models (`defineProjection`) rather than aggregate replays?

```
```


# Agent History, Replay & Audit (/docs/playbooks/ai-agent-deterministic-memory-and-audit)



An agent run is an approachable way to learn event history: record a goal, a tool request, a tool result, and a final response. The same Alvyn primitives work for orders, approvals, billing, and other domain applications. You do not need to adopt CQRS or replace all your CRUD tables.

**Replay reconstructs application state from recorded events using a reducer. It does not rerun a model deterministically.** Only the inputs, outputs, and actions your application explicitly instruments are recorded; hidden model thoughts are not available to the store.

## Record facts, not inferred thoughts [#record-facts-not-inferred-thoughts]

An application can model its workflow with these event contracts:

```typescript
type AgentEvents = {
  GoalReceived: { goal: string };
  ToolInvoked: { callId: string; tool: string };
  ToolCompleted: { callId: string; result: string };
  RunCompleted: { output: string };
  RunFailed: { code: string };
};
```

Define an evolve handler for every event type. The first handler receives `null` at runtime, even though the library's handler parameter is typed as the state type. `GoalReceived` must therefore create the entire state; other handlers need a null-safe baseline. An empty aggregate load has `state: null` and `version: 0&#x60;. Explicitly create a new stream with &#x2A;*`expectedVersion: -1`**, not the empty-load version. For an existing stream, use its loaded version or the preceding append's `toVersion`.

See the [introduction](/docs) for a minimal persistence snippet. In your integration, await each append: `GoalReceived`, `ToolInvoked` with a stable call ID, then execute the tool, record `ToolCompleted` with that same ID, and finally `RunCompleted`. Decide how to handle persistence failures before executing external actions.

Evolve handlers are pure state transformations, not command validation or tool executors. Production command handling must additionally validate payloads, allowed transitions, authorization, and call/result correlation before appending. A database transaction cannot atomically execute a remote tool.

## Reconstruct an earlier state [#reconstruct-an-earlier-state]

Aggregate `load` returns current state. To inspect an earlier version, load events with `EventStore.load` or `loadFrom` and pass the selected prefix through an application-defined reducer.

Without snapshots, start at `null` and apply ordered events only while `streamVersion <= version`, without calling a model or tool. Handle unknown event types explicitly. For large streams, use bounded `loadFrom(streamId, { fromVersion, maxEvents })` pages, where `fromVersion` is inclusive; advance to the last returned `streamVersion + 1`. Carry reducer state between pages and stop at the target version.

A version is an unambiguous stream boundary. A wall-clock timestamp is not necessarily causal order across streams or a commit-time boundary; if filtering on `createdAt` or CloudEvents `time`, specify those semantics yourself. There are no `toTime` or `toVersion` aggregate-load options in this recipe.

Reconstruction depends on the retained events, reducer version, upcasters, and decryptable data. Changing a reducer can change the result. Re-running a model with recorded inputs is a new experiment, not replay of the original response; store it under a separate run ID.

## Integrating a model or orchestration framework [#integrating-a-model-or-orchestration-framework]

Alvyn is framework-neutral. This guide does not provide a version-specific AI SDK integration, and Alvyn has **no existing LangGraph adapter**. An application integration must bridge the framework's lifecycle to event appends.

* Pin the model SDK and provider package versions and consult their matching documentation before implementing callbacks, tool schemas, stop conditions, or usage fields. Do not assume API names are interchangeable across AI SDK releases.
* Instrument the actual request and response boundaries: model/provider identifiers, prompt/template version, allowed input context, tool call IDs, sanitized arguments and outputs, finish reason, errors, and provider-reported usage when available. Do not invent token counts or a rationale the model did not expose.
* Await asynchronous logging writes, bound the number of steps, and choose whether logging failure stops execution. Background callbacks that are not awaited can lose records.
* Serialize appends per run or explicitly handle optimistic concurrency conflicts. Parallel tool callbacks must not share a mutable cached version without coordination. Reload and re-evaluate commands after a conflict; do not blindly repeat external effects.
* Keep operational logs and metrics for persistence failures too: if the database is unavailable, appending `RunFailed` can also fail. Preserve the original exception and report the logging failure separately.

## External effects and retries [#external-effects-and-retries]

For a real tool, first persist intent with a stable operation ID, then call the tool with that ID as its idempotency key when supported, then persist the result. A crash after the remote action but before the result append leaves an uncertain outcome. Reconcile using the remote operation ID or a status lookup before retrying. Never record success merely because a request was sent.

Alvyn append idempotency applies to database writes; it does not deduplicate payments, emails, or model requests. Transactional outbox messages are atomic with event appends, but delivery is **at least once**. Consumers need durable deduplication and external systems need idempotency or reconciliation. Competing-worker locks are not an exactly-once external-execution guarantee.

## Privacy, snapshots, and audit limits [#privacy-snapshots-and-audit-limits]

For personal data, minimize collection and configure encrypted fields and key management explicitly. Alvyn supports per-entity envelope encryption and key revocation, not automatic privacy compliance.

After revocation, affected reads may return a tombstone with encrypted field paths set to `null` and public fields preserved in the original schema. A historical reducer must handle this explicitly rather than pretending to reproduce the original private state; null-safe aggregate handlers merely avoid crashing and do not restore missing facts. Previously loaded plaintext is not removed from application memory.

Snapshots are separate stored events and can contain copied sensitive state. Configure their encryption separately; source-event encryption does not automatically protect every derived copy. Projections, exports, logs, caches, provider retention, backups, and retained keys need their own deletion and retention policy. Revoking an active key does not erase plaintext copies or make old backups with recoverable keys harmless. Do not promise identical replay after deletion.

An append-oriented application API is not tamper-proof storage against a privileged database operator. Access controls, role separation, backups, retention, monitoring, and any required independent integrity evidence remain deployment responsibilities. Structured event history can support an audit process, including an applicable AI Act logging assessment, but it neither automatically instruments the application nor establishes legal compliance. Determine applicable obligations with qualified reviewers rather than treating this recipe as a regulatory checklist.

## Next steps [#next-steps]

* [Aggregates](/docs/aggregates): domain state and typed events.
* [Transactional outbox](/docs/outbox): reliable delivery with retry-aware consumers.
* [Snapshots](/docs/snapshots): performance helpers and derived-state lifecycle.
* [Data privacy](/docs/crypto-shredding): encryption and key management.


# Event Streaming & Resilient Consumer Playbook (/docs/playbooks/sse-and-consumer-scaling)





Building distributed, event-driven microservices often brings architectural dilemmas: When is a dedicated message broker necessary, and when does it introduce accidental complexity? How do you stream events to downstream services reliably without data loss? And how do you handle Kubernetes consumer replica sets without duplicate processing or race conditions?

This playbook provides an architectural decision framework and bounded implementation sketches for **Alvyn**, &#x2A;*Server-Sent Events (SSE)**, and **Transactional Outbox relays**. The sketches are not a production-ready HA consumer framework; their limits and required operational safeguards are explicit below.

***

## 1. Architectural Blueprint: Direct Streaming vs. Message Brokers [#1-architectural-blueprint-direct-streaming-vs-message-brokers]

A common reflex in event-driven architecture is to place an external message broker (like Apache Kafka, RabbitMQ, or NATS) between services as soon as events need to be shared. While brokers solve critical problems at scale, they also introduce operational overhead, client library dependencies, and potential consistency hurdles.

In an event-sourced architecture powered by **Alvyn**, PostgreSQL acts as an immutable, ordered, append-only event log. Understanding the trade-offs helps you choose the right communication model:

### Communication Models: Fan-Out vs. Competing Consumers [#communication-models-fan-out-vs-competing-consumers]

| Dimension               | Direct HTTP Streaming (Alvyn + SSE)                                                                      | Message Brokers (Kafka / RabbitMQ / NATS)                                                               |
| :---------------------- | :------------------------------------------------------------------------------------------------------- | :------------------------------------------------------------------------------------------------------ |
| **Delivery Model**      | **Fan-Out (Broadcast):** Every connected client receives all events matching its filter.                 | **Competing Consumers:** Broker distributes individual messages across worker pods in a consumer group. |
| **ReplicaSet Handling** | Requires client-side strategy (In-Process Worker Pool, Active-Passive Leader Lock, or Subject Sharding). | Built into the broker (partition assignment or queue round-robin).                                      |
| **Protocol**            | Standard HTTP/1.1 or HTTP/2 (`text/event-stream`). Accessible by any language, `curl`, or browser.       | Proprietary binary protocols (AMQP, Kafka TCP wire protocol, NATS protocol).                            |
| **Infrastructure**      | **Zero additional infrastructure:** Runs directly on your existing API Gateway and database.             | Requires dedicated cluster deployment, monitoring, JVM/storage tuning, and backup routines.             |
| **Historical Replay**   | Native: reconnect with `Last-Event-ID: <position>`; omit `lowerBound` to start at the beginning.         | Supported on log-based brokers (Kafka/Pulsar/JetStream); limited on classic queues.                     |
| **Dual-Write Risk**     | **None:** Read directly from the source PostgreSQL event store.                                          | Mitigated via Transactional Outbox pattern (`eventStore.processOutbox`).                                |

### When to Use Direct HTTP Streaming (Alvyn + SSE) [#when-to-use-direct-http-streaming-alvyn--sse]

* **CQRS Read-Model Projections:** Replicating domain events to external databases (e.g., Elasticsearch, Redis, MongoDB).
* **Downstream Microservices & 3rd-Party APIs:** Allowing external partners or internal microservices to consume events using standard web standards without configuring VPNs or broker client libraries.
* **Low-to-Medium Fleet Complexity:** When consumers can process events using in-process concurrency or active-passive replica sets.
* **Operational Simplicity:** When you want to minimize the number of moving parts in your infrastructure stack.

### When to Introduce a Dedicated Message Broker [#when-to-introduce-a-dedicated-message-broker]

* **High-Volume Competing Consumers:** When hundreds of independent worker pods must pull work items from a shared queue in round-robin fashion.
* **Big Data & Analytics Pipelines:** Ingesting 200,000+ events per second into stream-processing engines like Apache Flink or Apache Spark (ideal for **Apache Kafka**).
* **Complex Transactional Task Queuing:** Fine-grained per-message TTLs, priority queues, and dead-letter exchanges (ideal for **RabbitMQ**).
* **Ultra-Low Latency RPC & Edge Computing:** Sub-millisecond synchronous request-reply and edge mesh synchronization via leaf nodes (ideal for **NATS JetStream**).

<Callout type="info">
  **Alvyn supports both models:** You can stream events directly via
  `eventStore.subscribe()` over SSE, or use Alvyn's transactional outbox
  (`eventStore.processOutbox`) to relay events to Kafka, RabbitMQ, or NATS with
  guaranteed at-least-once delivery.
</Callout>

***

## 2. Core Primitives & Guarantees (Demystified) [#2-core-primitives--guarantees-demystified]

Before implementing the endpoints, distinguish Alvyn's ordered read cursor from application-owned durable checkpoints, retries, and side effects.

### Concept 1: The Monotonic Bookmark (`globalPosition`) [#concept-1-the-monotonic-bookmark-globalposition]

Every event written to Alvyn receives a database-generated `BIGSERIAL` sequence number called `globalPosition`.

`globalPosition` is a unique sequence position, not transaction commit order or wall-clock order; gaps are normal. Alvyn's subscription watermark prevents advancing past unresolved lower positions. A checkpoint of `10522` acknowledges matching events through that position only if the consumer has successfully processed the entire delivered prefix. A completed parallel task alone is not a safe checkpoint.

### Concept 2: The W3C SSE Standard (`Last-Event-ID`) [#concept-2-the-w3c-sse-standard-last-event-id]

The W3C Server-Sent Events standard includes built-in resumption mechanics. When the server pushes an event, it includes an `id:` line:

```http
id: 10522
event: OrderPlaced
data: {"orderId":"ORD-99","total":89.99}
```

When a network drop occurs, compliant HTTP clients automatically attach the `Last-Event-ID` header upon reconnecting:

```http
GET /events HTTP/1.1
Host: api.example.com
Accept: text/event-stream
Last-Event-ID: 10522
```

### Concept 3: Alvyn's `lowerBound` Parameter [#concept-3-alvyns-lowerbound-parameter]

When invoking `eventStore.subscribe()`, you supply `lowerBound`:

```typescript
const stream = eventStore.subscribe({
  lowerBound: { id: "10522", type: "exclusive" },
});
```

This tells Alvyn: &#x2A;"Resume immediately after position 10522 (`global_position > 10522`)."*

### Concept 4: The Zero-Gap Single Query Loop [#concept-4-the-zero-gap-single-query-loop]

Alvyn does **not** maintain separate systems for "historical catch-up" and "live listening". Instead, it executes a continuous cursor query against PostgreSQL:

```sql
SELECT * FROM events
WHERE global_position > $1 AND global_position <= $2
ORDER BY global_position ASC
LIMIT 500;
```

1. **Catch-Up Phase:** While the consumer's cursor (`$1`) is behind the database head, Alvyn fetches full 500-event batches via primary key index scans.
2. **Live Transition:** When a query returns fewer than 500 events, the consumer is caught up. Alvyn pauses on an internal waker tied to PostgreSQL `LISTEN/NOTIFY` (with a periodic polling fallback).
3. **Instant Wake-Up:** The instant a new transaction commits, the waker triggers the exact same query from the updated cursor.

### Concept 5: The Commit-Safe Watermark [#concept-5-the-commit-safe-watermark]

In PostgreSQL, sequence numbers are allocated during insertion, but transactions commit in non-deterministic order:

```
Tx A (globalPosition 101) -------------> commits at t=2
Tx B (globalPosition 100) --------------------> commits at t=3 (delayed)
```

If a subscriber tailing at `t=2` read position `101`, advancing its cursor to `101`, it would &#x2A;*permanently miss event `100`** when Tx B commits at `t=3`.

Alvyn eliminates this via `computeSafeWatermark`. The query upper bound (`$2`) only advances up to the safe watermark, holding back positions until all preceding concurrent transactions have either committed or aborted.

***

## 3. Step-by-Step Producer Implementation (HTTP SSE with Alvyn) [#3-step-by-step-producer-implementation-http-sse-with-alvyn]

This minimal Express / Node.js endpoint illustrates bookmarks, disconnect cleanup, and heartbeats. Before deployment, add authentication and per-subject authorization, validate decimal cursor input before sending headers, bound slow-client buffering with backpressure/timeouts, and test proxy/disconnect behavior. It does not implement these production safeguards:

```typescript
import type { Request, Response } from "express";
import { eventStore } from "./event-store";

export async function sseEventsHandler(req: Request, res: Response) {
  // 1. Set required SSE headers
  res.writeHead(200, {
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-cache, no-transform",
    Connection: "keep-alive",
    "X-Accel-Buffering": "no", // Disable buffering in Nginx / AWS ALB
  });
  res.flushHeaders();

  // 2. Extract Last-Event-ID from standard header or query param fallback
  const lastEventId = (req.headers["last-event-id"] ||
    req.query.lastEventId) as string | undefined;

  // 3. AbortController to cleanly release PostgreSQL connections on disconnect
  const abortController = new AbortController();
  res.on("close", () => {
    abortController.abort();
  });

  // 4. Heartbeat interval to prevent intermediate load balancers from dropping idle connections
  const heartbeatTimer = setInterval(() => {
    if (!res.writableEnded) {
      res.write(": keepalive\n\n");
    }
  }, 15000);

  try {
    // 5. Start Alvyn subscription from the client's bookmark
    const eventStream = eventStore.subscribe({
      lowerBound: lastEventId
        ? { id: lastEventId, type: "exclusive" }
        : undefined,
      signal: abortController.signal,
    });

    // 6. Stream events formatted according to W3C SSE standard
    for await (const event of eventStream) {
      res.write(`id: ${event.globalPosition.toString()}\n`);
      res.write(`event: ${event.type}\n`);
      res.write(
        `data: ${JSON.stringify({ ...event, globalPosition: event.globalPosition.toString() })}\n\n`,
      );
    }
  } catch (error: any) {
    if (!abortController.signal.aborted) {
      console.error("SSE stream error:", error);
      res.end();
    }
  } finally {
    clearInterval(heartbeatTimer);
    abortController.abort();
    if (!res.writableEnded) res.end();
  }
}
```

<Callout type="info">
  **PostgreSQL Connection Pooling (e.g., PgBouncer):** Because SSE clients
  maintain continuous streaming connections, ensure database proxies or pools
  operate in **Session Pooling** mode (or connect directly to PostgreSQL).
  **Transaction Pooling** does not retain session-level notifications
  (`LISTEN/NOTIFY`) or Advisory Locks.
</Callout>

***

## 4. Step-by-Step Consumer Implementation (Resilient Client) [#4-step-by-step-consumer-implementation-resilient-client]

The following single-process demonstration uses a local file and parses only the single-line LF-delimited format emitted above. It is not an HA checkpoint store or a general SSE parser. Local checkpoint loss causes replay; downstream effects must be idempotent. Use a durable shared database checkpoint for failover and a standards-compliant parser with bounded frame size in production:

```typescript
import fs from "node:fs/promises";
import path from "node:path";

const CHECKPOINT_PATH = path.resolve("./consumer.checkpoint");

// Load the last saved globalPosition
async function loadCheckpoint(): Promise<string | undefined> {
  try {
    return (await fs.readFile(CHECKPOINT_PATH, "utf-8")).trim();
  } catch (error) {
    if ((error as NodeJS.ErrnoException).code === "ENOENT") return undefined;
    throw error; // Do not silently reset progress on I/O errors
  }
}

// Persist checkpoint after successful handling
async function saveCheckpoint(position: string): Promise<void> {
  await fs.writeFile(`${CHECKPOINT_PATH}.tmp`, position, "utf-8");
  await fs.rename(`${CHECKPOINT_PATH}.tmp`, CHECKPOINT_PATH);
}

export async function startResilientConsumer(sseUrl: string) {
  let backoffMs = 1000;

  while (true) {
    const lastEventId = await loadCheckpoint();
    const controller = new AbortController();
    console.log(
      `Connecting to ${sseUrl} (Last-Event-ID: ${lastEventId ?? "BEGINNING"})...`,
    );

    try {
      const response = await fetch(sseUrl, {
        signal: controller.signal,
        headers: {
          Accept: "text/event-stream",
          ...(lastEventId ? { "Last-Event-ID": lastEventId } : {}),
        },
      });

      if (!response.ok || !response.body) {
        throw new Error(
          `HTTP error ${response.status}: ${response.statusText}`,
        );
      }

      // Reset backoff upon successful connection
      backoffMs = 1000;

      const reader = response.body.getReader();
      const decoder = new TextDecoder();
      let buffer = "";

      while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, { stream: true });
        if (buffer.length > 1_048_576)
          throw new Error("SSE buffer limit exceeded");
        const chunks = buffer.split("\n\n");
        buffer = chunks.pop() ?? "";

        for (const rawChunk of chunks) {
          if (rawChunk.startsWith(":")) continue; // Ignore keepalive heartbeats

          const lines = rawChunk.split("\n");
          let eventId = "";
          let eventType = "";
          let eventDataRaw = "";

          for (const line of lines) {
            if (line.startsWith("id: ")) eventId = line.slice(4).trim();
            else if (line.startsWith("event: "))
              eventType = line.slice(7).trim();
            else if (line.startsWith("data: "))
              eventDataRaw = line.slice(6).trim();
          }

          if (eventId && eventDataRaw) {
            const event = JSON.parse(eventDataRaw);

            // Execute domain business logic (must be idempotent)
            await handleDomainEvent(eventType, event);

            // Acknowledge checkpoint
            await saveCheckpoint(eventId);
          }
        }
      }
      throw new Error("SSE stream ended"); // Reconnect with backoff on clean EOF too
    } catch (error) {
      controller.abort();
      const jitterMs = Math.floor(Math.random() * 300); // 0-300ms random offset to prevent thundering herds
      const sleepMs = backoffMs + jitterMs;
      console.error(
        `SSE connection dropped. Retrying in ${sleepMs}ms...`,
        error,
      );
      await new Promise((resolve) => setTimeout(resolve, sleepMs));
      backoffMs = Math.min(backoffMs * 2, 30000); // Exponential backoff capped at 30s
    } finally {
      controller.abort();
    }
  }
}

async function handleDomainEvent(type: string, data: any) {
  // Domain business logic here (e.g. update local read model)
}
```

<Callout type="info">
  **Production Parser Tip:** For production edge cases involving multi-line
  formatted JSON strings spanning multiple `data:` lines or custom SSE comments,
  consider using a lightweight, zero-dependency W3C parser.
</Callout>

***

## 5. Multi-Instance Deployments & High-Availability Patterns [#5-multi-instance-deployments--high-availability-patterns]

In production cloud environments (such as Kubernetes, AWS ECS, or Nomad), services are rarely deployed as single, isolated instances. Multiple replicas are standard for several reasons:

1. **High Availability & Zero-Downtime:** Running 2–3 replicas across availability zones ensures that rolling deployments, node maintenance, or spontaneous pod restarts do not disrupt service availability.
2. **Hybrid / Multi-Purpose Services:** A microservice often serves incoming REST or GraphQL traffic from web/mobile clients behind an ingress load balancer, while simultaneously running a background event consumer to update local caches, index search data, or trigger asynchronous workflows.
3. **High I/O & Compute Throughput:** Processing tasks (such as PDF generation, third-party API synchronization, or complex aggregations) require concurrent execution across CPU cores.

Because HTTP Server-Sent Events operates on standard HTTP streams, the server delivers events to every connected client (**Fan-Out / Broadcast**). To structure multi-instance deployments cleanly, choose the architectural pattern that aligns with your operational goals:

### Architectural Decision Matrix [#architectural-decision-matrix]

| Deployment Objective                                                                | Recommended Pattern                        | Key Characteristics                                                                                           |
| :---------------------------------------------------------------------------------- | :----------------------------------------- | :------------------------------------------------------------------------------------------------------------ |
| **HA / Hybrid Web API** (e.g. 3 pods serving REST traffic + background event tasks) | **Pattern A: Coordinated ownership**       | Requires durable shared progress, cancellation on ownership loss, and fencing or idempotent effects.          |
| **High I/O Concurrency** (e.g. 50ms per event for PDF gen or DB writes)             | **Pattern B: In-Process Hash Worker Pool** | Overlaps I/O across partitions; preserves input order within each partition until failure.                    |
| **Domain / Regional Partitioning** (e.g. EU vs. US orders, Billing vs. Logistics)   | **Pattern C: Database-Level Sharding**     | Logical shards use disjoint filters and separately managed ownership/checkpoints.                             |
| **Elastic Competing Worker Fleet** (e.g. 50 dynamically autoscaled worker pods)     | **Pattern D: Message Broker Relay**        | Bridge Alvyn to Kafka, RabbitMQ, or NATS via the Transactional Outbox for dynamic consumer group rebalancing. |

***

### Pattern A: Active-Passive Leader Election (HA & Hybrid Services) [#pattern-a-active-passive-leader-election-ha--hybrid-services]

**Use Case:** You run 2–3 replicas of a service for **High Availability** or as a **Hybrid Service** (serving HTTP/REST traffic behind a load balancer while running a background event listener). You want the service to survive pod restarts without executing background event tasks multiple times in parallel.

An advisory lock alone is not a fenced lease. If the lock connection fails while the SSE connection or an external request remains alive, an old owner can keep working while a new owner acquires the lock. The local file consumer above cannot provide shared HA progress and must not be wrapped in a lock loop and called production-safe.

#### Bounded Recipe and Limits [#bounded-recipe-and-limits]

1. For a PostgreSQL read model, prefer `runProjection()` and perform all writes through its transaction client. Checkpoint row locking and data updates then share a transaction without a custom SSE leader.
2. For external effects, prefer `processOutbox()` with durable broker acknowledgements, retries, and destination-side deduplication. Delivery remains at-least-once.
3. If an SSE leader is necessary, implement a separate owner lifecycle: acquire ownership on a dedicated connection, load a **shared durable** checkpoint, and pass cancellation through the fetch, parser, worker queues, and handlers. On connection loss or shutdown, stop accepting work and await/cancel in-flight tasks before releasing ownership. Never return a still-locked session to a pool.
4. Persist only the successfully handled contiguous prefix. Where possible, commit local effects and the checkpoint in one transaction. Require destination-enforced monotonic fencing tokens to reject stale owners, or idempotency keys where fencing is unavailable; cancellation alone cannot undo an already issued external request.
5. Test lock-session loss while SSE stays connected, process pauses, partial external success, checkpoint failures, and rolling restarts. Define retry, timeout, and replay budgets. Failover latency depends on failure detection; neither instant failover nor zero duplicates is promised.

This page deliberately does not supply an HA lock wrapper: Alvyn's SSE subscription does not implement a distributed ownership/fencing protocol for your external system.

***

### Pattern B: In-Process Hash-Partitioned Worker Pool (High I/O Concurrency) [#pattern-b-in-process-hash-partitioned-worker-pool-high-io-concurrency]

**Use Case:** A dedicated consumer pod needs to process a high volume of events where individual task execution is I/O-heavy (e.g., rendering PDFs, making external REST calls, or writing to downstream stores).

In most architectures, network transmission of JSON events over SSE is extremely fast (tens of thousands of events per second on a single connection); the bottleneck is downstream I/O. Instead of opening duplicate connections, maintain **one SSE connection per consumer pod** and dispatch events in-memory across an internal pool of worker queues by hashing the `streamId` (`hash(streamId) % numberOfWorkers`):

* **Per-Entity Input Order:** Events for `Order-123` are routed to the same queue and executed in delivery order, not wall-clock order.
* **I/O Concurrency:** Different partitions can overlap asynchronous I/O. Promise queues do not provide multi-core CPU execution or eliminate races on shared resources.

```typescript
import { createHash } from "node:crypto";

export class HashPartitionedWorkerPool {
  private queues: Array<Promise<void>> = [];

  constructor(private concurrency: number = 16) {
    if (!Number.isSafeInteger(concurrency) || concurrency < 1) {
      throw new Error("concurrency must be a positive safe integer");
    }
    this.queues = Array.from({ length: concurrency }, () => Promise.resolve());
  }

  private getPartition(key: string): number {
    const hash = createHash("md5").update(key).digest().readUInt32BE(0);
    return hash % this.concurrency;
  }

  public enqueue(streamId: string, task: () => Promise<void>): Promise<void> {
    const partition = this.getPartition(streamId);

    // Chain the task onto the specific queue for this streamId
    const chain = this.queues[partition].then(task);

    this.queues[partition] = chain;
    return chain;
  }
}
```

The returned promise rejects on failure; the affected partition stays failed so later work cannot pass the failed event. Observe every returned promise immediately. Bound each dispatch batch and await all outcomes before advancing the checkpoint:

```typescript
// batch is a bounded, ordered list of consecutive delivered events
const outcomes = await Promise.allSettled(
  batch.map((event) => workers.enqueue(event.streamId, () => handle(event))),
);
const failed = outcomes.find((outcome) => outcome.status === "rejected");
if (failed?.status === "rejected") throw failed.reason;
if (batch.length > 0) {
  await saveCheckpoint(batch[batch.length - 1].globalPosition.toString());
}
```

Do not dispatch another batch after failure: stop intake, settle in-flight work, recreate the pool, and replay from the last durable checkpoint. Successful external effects in a failed batch may repeat, so handlers must be idempotent. The queue class alone does not bound memory; the caller must bound batch size and avoid accumulating unawaited batches.

***

### Pattern C: Database-Level Sharding via Subject / Type Filters (Domain & Regional Partitioning) [#pattern-c-database-level-sharding-via-subject--type-filters-domain--regional-partitioning]

**Use Case:** Workloads are naturally segregated by business domain or geographic region, and you want **each pod in your fleet to handle a dedicated slice of the event catalog**.

Partition the stream at the database query level using Alvyn's `subject` prefix or `eventTypes` filters:

<Tabs items="[&#x22;Pod A: EU Orders&#x22;, &#x22;Pod B: US Orders&#x22;, &#x22;Pod C: Logistics Events&#x22;]">
  <Tab value="Pod A: EU Orders">
    ```typescript
    // Dedicated pod processing only European orders
    const stream = eventStore.subscribe({
      subject: "Order-EU-",
      recursive: true,
      lowerBound: { id: lastProcessedEuId },
    });
    ```
  </Tab>

  <Tab value="Pod B: US Orders">
    ```typescript
    // Dedicated pod processing only US orders
    const stream = eventStore.subscribe({
      subject: "Order-US-",
      recursive: true,
      lowerBound: { id: lastProcessedUsId },
    });
    ```
  </Tab>

  <Tab value="Pod C: Logistics Events">
    ```typescript
    // Dedicated pod processing only warehouse and fulfillment event types
    const stream = eventStore.subscribe({
      eventTypes: ["OrderShipped", "OrderDelivered", "ParcelReturned"],
      lowerBound: { id: lastProcessedLogisticsId },
    });
    ```
  </Tab>
</Tabs>

Each logical shard needs a durable checkpoint and explicit ownership. Filters must be disjoint to avoid overlap; replicas with the same filter still receive the same events. Retries can duplicate effects, and changing shard filters requires a planned checkpoint/replay migration.

***

### Pattern D: Competing Consumers via Message Broker (Alvyn Transactional Outbox) [#pattern-d-competing-consumers-via-message-broker-alvyn-transactional-outbox]

**Use Case:** You require automatic, dynamic partition rebalancing across an elastic fleet of dozens or hundreds of worker pods without maintaining manual shard filters or leader locks.

Bridge Alvyn to Apache Kafka, RabbitMQ, or NATS JetStream using Alvyn's Transactional Outbox, as detailed in the next section.

***

## 6. The Hybrid Bridge: Relaying to Message Brokers via Alvyn Outbox [#6-the-hybrid-bridge-relaying-to-message-brokers-via-alvyn-outbox]

When your architecture requires **true competing consumers** across dozens of dynamically scaled pods, or integrates with an existing enterprise Kafka / RabbitMQ / NATS cluster, use Alvyn's **Transactional Outbox**.

This pattern eliminates the Dual-Write problem by atomically storing outbox entries within the same database transaction as the domain event:

1. **Atomic Transaction:** Domain events and outbox messages are committed together in one transaction.
2. **Non-Blocking Polling:** Background workers call `eventStore.processOutbox()`, which fetches pending messages using `FOR UPDATE SKIP LOCKED`.
3. **Broker Publishing with ACK:** Events are dispatched to Kafka, RabbitMQ, or NATS. Entries are marked processed only when the handler promise resolves.

### Outbox Publisher Implementation [#outbox-publisher-implementation]

```typescript
import { EventStore } from "alvyn";
import { kafkaProducer } from "./kafka-client"; // or RabbitMQ / NATS

export async function startOutboxRelay(
  eventStore: EventStore,
  signal: AbortSignal,
) {
  while (!signal.aborted) {
    const processed = await eventStore.processOutbox(async (entries) => {
      const messages = entries.map((entry) => ({
        key: entry.topic,
        value: JSON.stringify(entry.payload),
        headers: {
          globalPosition: entry.eventGlobalPosition.toString(),
        },
      }));

      // Send batch to Kafka topic with at-least-once delivery guarantee
      await kafkaProducer.send({
        topic: "domain-events",
        messages,
      });
    }, 100);

    if (processed === 0) {
      await new Promise((resolve) => setTimeout(resolve, 250));
    }
  }
}
```

<Callout type="info">
  **Outbox Table Retention & Purging:** In high-throughput architectures, purge processed outbox rows periodically to keep the outbox table compact:

  ```typescript
  // Delete processed outbox rows older than 7 days
  const deletedCount = await eventStore.cleanupOutbox(7 * 24 * 60 * 60 * 1000);
  ```
</Callout>

***

## 7. Observability, Monitoring & Disaster Recovery [#7-observability-monitoring--disaster-recovery]

The relay example assumes a KafkaJS-compatible producer. Retry failed `processOutbox()` batches with bounded backoff in your supervisor. A broker acknowledgement followed by a failed PostgreSQL commit causes redelivery; the row lock is not an exactly-once guarantee for external effects.

### Monitoring Consumer Lag (SQL) [#monitoring-consumer-lag-sql]

To detect if a consumer is falling behind during high-traffic spikes, query the lag directly in PostgreSQL:

```sql
SELECT
  MAX(global_position) AS head_position,
  $1::BIGINT AS consumer_checkpoint,
  (MAX(global_position) - $1::BIGINT) AS consumer_lag
FROM event_store.events;
```

* This is a position-distance estimate, not an event count: sequences have gaps, filters exclude events, and the visible head can exceed the safe watermark. Use your configured schema in place of `event_store`.
* Alert on sustained lag growth and processing latency against a measured service objective; investigate slow handlers and watermark stalls before increasing concurrency.

### Full Event Replays (Disaster Recovery & New Services) [#full-event-replays-disaster-recovery--new-services]

When deploying a brand new downstream microservice (or recovering from a corrupted database), set the consumer checkpoint to `0` or omit `Last-Event-ID`.

Alvyn will stream every historical event from the beginning of time in 500-event chunks, then automatically transition into the live stream once caught up.

### Idempotency Checklist [#idempotency-checklist]

Because network retries guarantee **at-least-once delivery**, make your downstream handlers idempotent:

1. **Deduplication Key:** Store the processed `globalPosition` or event `id` in a unique PostgreSQL table (`processed_events`).
2. **Transactional State Updates:** Commit your local domain state change and the consumer checkpoint in the same database transaction.
3. **Monotonic Version Checks:** When updating read models, ensure updates are only applied if `event.streamVersion > current_entity_version`.