Command Handler

Overview

A command handler processes a command against a single event stream. The command is an intention, such as adding an item to a cart; your business logic decides what should happen; the events it returns are the outcome that gets stored.

Use it whenever a command writes to a single stream. It saves you from repeating this flow by hand, and it applies the optimistic concurrency check that stops two commands on the same stream from overwriting each other.

On each call it loads the stream and folds its events into the current state with evolve and initialState, runs your decision function, and appends the returned events using the version it read as the expected version. It builds on the event store's aggregateStream and appendToStream; when you'd rather keep decide, evolve, and initialState together as one object, the Decider is the more structured alternative.

Basic Usage

Configure a handler once with evolve and initialState, then reuse it across the app, calling it for each command with a stream id and one or more decision functions.

Creating a handler

import { CommandHandler } from '@event-driven-io/emmett';
import { evolve, initialState } from './shoppingCart';

export const handle = CommandHandler({ evolve, initialState });

Handling a command

import { getInMemoryEventStore } from '@event-driven-io/emmett';

const eventStore = getInMemoryEventStore();

const command: AddProductItemToShoppingCart = {
  type: 'AddProductItemToShoppingCart',
  data: {
    shoppingCartId,
    productItem,
  },
};

const { nextExpectedStreamVersion } = await handle(
  eventStore,
  shoppingCartId,
  (state) => addProductItem(command, state),
);

Type Definitions

CommandHandler takes three type parameters: State, StreamEvent (the discriminated union written to the stream), and an optional EventPayloadType (the stored shape when it differs from StreamEvent, used with schema versioning). It returns an async function whose third argument is either a single handler or an array of handlers. The full signatures are shown under Type Source; the tables below summarise them.

CommandHandlerOptions

Property	Type	Description
evolve	(state: State, event: StreamEvent) => State	Folds an event into state. Required.
initialState	() => State	Starting state for a stream with no events. Required.
mapToStreamId	(id: string) => string	Maps the business id to the stream name. Defaults to the identity function.
retry	CommandHandlerRetryOptions	Retry policy for version conflicts. See Retry Configuration.
schema.versioning	{ upcast?; downcast? }	Converts stored events to and from StreamEvent for schema evolution.
serialization	{ serializer?; ... }	Custom serialiser for reading and writing events.
name / commandType	string / string | string[]	Labels used for observability and as the default command type when none is passed.
observability	CommandObservabilityConfig	Tracing and metrics configuration.

CommandHandlerResult

Each call returns the new state, the events it appended, and the version to pass on the next call. For the stores shipped with Emmett:

Property	Type	Description
newState	State	State after applying the newly produced events.
newEvents	StreamEvent[]	Events that were appended, empty when the handler produced none.
nextExpectedStreamVersion	StreamPosition (bigint for the built-in stores)	Pass this as expectedStreamVersion on the next call.
createdNewStream	boolean	Whether this call created the stream.

nextExpectedStreamVersion and createdNewStream come from the store you're using, so their exact type depends on it.

Stream ID Mapping

The id you pass to the handler is not always the stream's name. A cart id, for example, might map to a stream named shopping_cart:{id}. Set mapToStreamId to build the stream name from the id; you keep passing the business id, and the handler applies the mapping.

const handle = CommandHandler<ShoppingCart, ShoppingCartEvent>({
  evolve,
  initialState,
  mapToStreamId: (id) => `shopping_cart-${id}`,
});

Handler Functions

A handler receives the current state and returns one or more events to append.

Single event

A decision that produces one outcome returns a single event:

const addProductItem = (
  command: AddProductItem,
  _state: ShoppingCart,
): ShoppingCartEvent => {
  return {
    type: 'ProductItemAdded',
    data: { productItem: command.data.productItem },
  };
};

The handler appends it and reports the new state:

const { nextExpectedStreamVersion, newState, newEvents, createdNewStream } =
  await handleCommand(eventStore, shoppingCartId, (state) =>
    addProductItem(command, state),
  );

Multiple events

Return an array when a decision produces several events:

const addProductItemWithDiscount = (
  command: AddProductItem,
  _state: ShoppingCart,
): ShoppingCartEvent[] => {
  return [
    {
      type: 'ProductItemAdded',
      data: { productItem: command.data.productItem },
    },
    { type: 'DiscountApplied', data: { percent: defaultDiscount } },
  ];
};

The call is unchanged; the events are appended in a single operation, so the stream cannot be left holding only some of them:

const {
  nextExpectedStreamVersion,
  newState,
  newEvents,
  createdNewStream,
} = await handleCommand(eventStore, shoppingCartId, (state) =>
  addProductItemWithDiscount(command, state),
);

Sequential handlers

Pass an array of handlers to run several decisions in order. Each runs on the state produced by the handlers before it, and all their events are appended in a single write:

const { newState, newEvents, nextExpectedStreamVersion } =
  await handleCommand(eventStore, shoppingCartId, [
    (state) => addProductItem(addProduct, state),
    (state) => confirm(confirmCart, state),
  ]);

Optimistic Concurrency

The handler keeps the version it read from the stream and requires the stream to still be at that version when it appends. If another writer has changed the stream in between, the append fails with ExpectedVersionConflictError (a ConcurrencyError) instead of overwriting their events.

Automatic version tracking

By default the handler reuses the version it read, so you don't need to pass one:

const { newState: state1, nextExpectedStreamVersion } = await handleCommand(
  eventStore,
  shoppingCartId,
  (state) => addProductItem(command, state),
);

nextExpectedStreamVersion in the result is the stream's version after this append. The ETag concurrency example returns it to the client, which sends it back to guard its next write.

Explicit version

Pass expectedStreamVersion when the version to check comes from outside the handler, such as a client's If-Match header. The append then succeeds only if the stream is still at that version:

const { nextExpectedStreamVersion: version2 } = await handleCommand(
  eventStore,
  shoppingCartId,
  (state) => addProductItem(command, state),
  { expectedStreamVersion: nextExpectedStreamVersion },
);

New-stream requirement

Pass STREAM_DOES_NOT_EXIST as the expected version so the append succeeds only if the stream does not exist yet. This enforces that the stream is created exactly once:

const { nextExpectedStreamVersion, newState, newEvents, createdNewStream } =
  await handleCommand(
    eventStore,
    shoppingCartId,
    (state) => addProductItem(command, state),
    { expectedStreamVersion: 'STREAM_DOES_NOT_EXIST' },
  );

Retry Configuration

Set retry so Emmett re-runs the handler when a version conflict occurs.

{ onVersionConflict: true } applies Emmett's default conflict policy (3 retries, a 100 ms minimum timeout, and a 1.5× backoff factor), retrying only on ExpectedVersionConflictError:

const { newEvents } = await handleCommand(
  eventStore,
  shoppingCartId,
  (state) => addProductItem(command, state),
  { retry: { onVersionConflict: true } },
);

Pass a number ({ onVersionConflict: 5 }) to keep that policy but change the retry count. For full control, such as a different backoff or retrying on errors other than version conflicts, pass AsyncRetryOptions with your own shouldRetryError:

const handle = CommandHandler<ShoppingCart, ShoppingCartEvent>({
  evolve,
  initialState,
  retry: {
    retries: 5,
    minTimeout: 50,
    factor: 2,
    shouldRetryError: (error) => error instanceof ConcurrencyError,
  },
});

You can also set retry per call in the handle options, which overrides the handler-level policy for that call.

No-Op Handling

A decision returns an empty array when the current state leaves it nothing to do. Confirming a cart that is already confirmed is the common case:

const confirm = (
  command: ConfirmShoppingCart,
  state: ShoppingCart,
): ShoppingCartEvent[] => {
  // Already confirmed: nothing left to do, so append nothing
  if (state.status === 'Confirmed') return [];

  if (state.productItems.length === 0)
    throw new IllegalStateError('Cannot confirm an empty shopping cart');

  return [
    { type: 'ShoppingCartConfirmed', data: { confirmedAt: command.data.now } },
  ];
};

The handler then appends nothing and returns the current version with newEvents: [] and createdNewStream: false, so re-sending the command is safe:

// Confirming an already-confirmed cart is a no-op, so nothing is appended
const { newEvents, nextExpectedStreamVersion, createdNewStream } =
  await handleCommand(eventStore, shoppingCartId, (state) =>
    confirm(confirmCart, state),
  );

Error Handling

Business-rule errors

If a command breaks a business rule, throw from the decision: IllegalStateError for an invalid state transition, ValidationError for bad input. Nothing gets appended, and the error reaches the caller unchanged:

handleCommand(eventStore, shoppingCartId, () => {
  throw new IllegalStateError('Shopping Cart already closed');
});

Concurrency errors

A version conflict throws ExpectedVersionConflictError, which extends ConcurrencyError. Both carry the expected and current versions as strings, so you can report what happened to the caller:

try {
  await handleCommand(
    eventStore,
    shoppingCartId,
    (state) => addProductItem(command, state),
    { expectedStreamVersion: STREAM_DOES_NOT_EXIST },
  );
} catch (error) {
  if (error instanceof ConcurrencyError) {
    // error.expected: the version the command required
    // error.current:  the version the stream is actually at
    caught = error;
  }
}

Integration with Web Frameworks

Express.js

emmett-expressjs wraps a handler with on: call the command handler, then return a response helper. Fetch any data the decision needs, such as the unit price, before calling the handler and pass it into the command; this keeps the decision pure. The first write creates the stream:

// Add Product Item
router.post(
  '/clients/:clientId/shopping-carts/current/product-items',
  on(async (request: AddProductItemRequest) => {
    const shoppingCartId = getShoppingCartId(
      assertNotEmptyString(request.params.clientId),
    );
    const productId = assertNotEmptyString(request.body.productId);

    const command: AddProductItemToShoppingCart = {
      type: 'AddProductItemToShoppingCart',
      data: {
        shoppingCartId,
        productItem: {
          productId,
          quantity: assertPositiveNumber(request.body.quantity),
          unitPrice: await getUnitPrice(productId),
        },
      },
      metadata: { now: getCurrentTime() },
    };

    await handle(eventStore, shoppingCartId, (state) =>
      addProductItem(command, state),
    );

    return NoContent();
  }),
);

ETag concurrency

To expose optimistic concurrency over HTTP, carry the version in a weak ETag. Read the client's version from the If-Match header with getETagValueFromIfMatch, pass it as expectedStreamVersion, and return the new version via toWeakETag(result.nextExpectedStreamVersion):

// Add a product item, guarded by the version carried in the If-Match header
router.post(
  '/clients/:clientId/shopping-carts/:shoppingCartId/product-items',
  on(async (request: AddProductItemRequest) => {
    const shoppingCartId = assertNotEmptyString(
      request.params.shoppingCartId,
    );
    const productItem: ProductItem = {
      productId: assertNotEmptyString(request.body.productId),
      quantity: assertPositiveNumber(request.body.quantity),
    };
    const unitPrice = dummyPriceProvider(productItem.productId);

    const command: AddProductItemToShoppingCart = {
      type: 'AddProductItemToShoppingCart',
      data: { shoppingCartId, productItem: { ...productItem, unitPrice } },
    };

    const result = await handle(
      eventStore,
      shoppingCartId,
      (state) => decide(command, state),
      {
        expectedStreamVersion: assertUnsignedBigInt(
          getETagValueFromIfMatch(request),
        ),
      },
    );

    return NoContent({
      eTag: toWeakETag(result.nextExpectedStreamVersion),
    });
  }),
);

A stale If-Match makes the handler throw ExpectedVersionConflictError, which emmett-expressjs maps to HTTP 412 Precondition Failed.

Best Practices

Keep the Decision Pure

A handler that only reads state and returns events is easy to test and safe to retry. When a decision needs external data, such as a price or an exchange rate, fetch it before the handler and pass it in, as the Express.js example does with the unit price.

The handler can be async, and Emmett awaits it. Avoid I/O inside it, because on a version conflict the whole handler re-runs, so the call is made again on every retry:

const { newState, newEvents } = await handleCommand(
  eventStore,
  shoppingCartId,
  async (state) => {
    // ❌ Avoid: on a version-conflict retry the whole handler re-runs, so this call fires again
    const price = await getPrice(command.data.productItem.productId);
    return addProductItem(
      {
        ...command,
        data: { productItem: { ...command.data.productItem, price } },
      },
      state,
    );
  },
);

Guard Before Deciding

Throw IllegalStateError or ValidationError before building any event, so an invalid command never produces one. The confirm decision does this: it refuses to confirm an empty cart. See Error Handling for how those errors reach the caller.

Type Events as a Discriminated Union

Declare your events as a discriminated union and pass it as CommandHandler's StreamEvent:

type ShoppingCartEvent =
  ProductItemAdded | DiscountApplied | ShoppingCartConfirmed;

evolve and every handler's return value are then checked against that union, and unknown event types surface at compile time.

Type Source

For the full signatures, see handleCommand.ts in the source.

See also

• Getting Started - Command Handling
• Command
• Decider Pattern
• Event Store
• Error Handling
• Optimistic Concurrency for Pessimistic Times