Define type-safe event-sourced aggregates with full TypeScript inference, encryption, and schema evolution.

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

Step 1: Define the Event Map

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

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

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)

import { defineAggregate } from "@lox-solutions/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

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

`streamPrefix`

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

`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 supports forward compatibility — new event types can be added to the stream without breaking existing aggregate definitions.

`encryption` (optional)

See Crypto-Shredding for details.

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

`upcasters` (optional)

See Schema Evolution for details.

upcasters: [userRenamedV1ToV2, userRenamedV2ToV3];

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?)`

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

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?)`

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

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()`

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

eventStore.registerUpcasters(Order.getUpcasters());

OCC Retry Pattern

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

import { OptimisticConcurrencyError } from "@lox-solutions/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

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

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

When a crypto key is revoked (GDPR erasure), encrypted events return with data: null. Evolve handlers must gracefully handle null data using optional chaining with fallbacks:

evolve: {
  UserRegistered: (state, event) => ({
    ...state,
    name: event.data?.name ?? state.name,
    email: event.data?.email ?? state.email,
  }),
}

This ensures the aggregate can still be loaded after key revocation. Evolve handlers should preserve the existing state when encrypted fields are missing.

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:

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

import { defineAggregate, type Upcaster } from "@lox-solutions/alvyn";

type UserEvents = {
  UserRegistered: {
    name: string;
    email: string;
    address: { street: string; city: string };
  };
  UserRenamed: { name: string };
  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: new Date().toISOString() };
  },
};

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

Aggregates

On this page