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)`	Returns `AggregateInstance<TState>`
`loadEvents({ eventStore, entityId, maxEvents? })`	Loads typed replayed domain events
`append(eventStore, input)`	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)`

Loads the aggregate by replaying all events.

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

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, entityId, input)`

Appends typed events to the aggregate's stream. Automatically maps encryption config if defined.

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

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

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.

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