Command Handler

Overview

CommandHandler processes a command against a single event stream. Each call builds the current state from the stream's events with evolve and initialState, runs the decision, and appends the events it returns under an optimistic concurrency check.

It builds on the event store's aggregateStream and appendToStream. The Decider keeps decide, evolve, and initialState together as a single object. For setup and usage, see the Command Handling guide.

Construction

CommandHandler(options) returns the handler function. It 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).

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

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

DeciderCommandHandler(options) adds a CommandType parameter representing the command type. In DeciderCommandHandler we're not passing the decision callback.

CommandHandlerOptions

Property	Type	Description
`evolve`	`(state: State, event: StreamEvent) => State`	Applies an event to the state, returning the next 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.
`schema.versioning`	`{ upcast?; downcast? }`	Converts stored events to and from `StreamEvent`. See Advanced.
`serialization`	`{ serializer?; serializerOptions? }`	Custom serialiser for reading and writing events. See Advanced.
`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. See Advanced.
`middleware`	`Middleware[] \| { beforeAll?; afterAll?; decision? }`	Configures invocation-wide and per-decision middleware. An array is shorthand for `decision`.

Calling the Handler

The handler is called as handle(eventStore, id, decision, options?), with the event store, the business id, the decision, and optional per-call options. The third argument is a single decision or an array of decisions run in order (see Decisions).

HandleOptions

Property	Type	Description
`expectedStreamVersion`	`ExpectedStreamVersion`	Version the stream must be at for the append to succeed. Defaults to the version read at the start of the call.
`retry`	`CommandHandlerRetryOptions`	Retry policy for this call. Overrides the handler-level `retry`.

Decisions

A decision receives the current state and returns the events to append. It returns a single event, an array of events, or an empty array when there is nothing to append, and throws to reject the command. It may be synchronous or asynchronous, returning the events or a Promise of them.

A single event:

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

Several events, appended in one write so the stream is never left holding only some of them:

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

An array of decisions runs in order. Each receives the state left by the previous one, 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),
  ]);

An empty array is a no-op. The decision returns it when the current state leaves nothing to do:

const cancel = (
  command: CancelShoppingCart,
  state: ShoppingCart,
): ShoppingCartEvent[] => {
  // Already cancelled: nothing left to do, so append nothing
  if (state.status === 'Cancelled') return [];

  return [
    { type: 'ShoppingCartCancelled', data: { canceledAt: command.data.now } },
  ];
};

Result

Each call resolves to CommandHandlerResult:

Property	Type	Description
`newState`	`State`	State after applying the appended events.
`events`	`StreamEvent[]`	Every event produced by decisions that ran.
`appendedEvents`	`StreamEvent[]`	Events persisted by this call.
`newEvents`	`StreamEvent[]`	Deprecated alias of `appendedEvents`.
`nextExpectedStreamVersion`	`StreamPosition` (`bigint` for the built-in stores)	Version to pass as `expectedStreamVersion` on the next call.
`createdNewStream`	`boolean`	Whether this call created the stream.

nextExpectedStreamVersion and createdNewStream come from the store in use, so their exact type depends on it.

Decision Middleware

Ordinary decision results are treated as APPEND. Middleware can change how a complete decision result is handled with skipOn, stopOn, rejectOn, stopAfter, or a custom middleware. A predicate may inspect individual events, but a match always applies to every event returned by that decision.

APPEND stages and evolves the events, then continues.
SKIP exposes the events in events without staging or evolving them, then continues.
STOP discards the current events, commits earlier staged events, and stops.
REJECT discards every staged event, restores the state from before the batch, and stops.
APPEND_AND_STOP stages and evolves the current events, then commits and stops.

SKIP, STOP, REJECT, and APPEND_AND_STOP return normally; they do not throw. Produced events remain in events even when they are not present in appendedEvents. throwOn is the exception-producing helper.

const addUnavailableProduct = (state: ShoppingCart) =>
  addProductItemWithStock(unavailableProduct, 2, state);

const handle = CommandHandler<ShoppingCart, ShoppingCartEvent>({
  evolve,
  initialState,
  middleware: {
    beforeAll: authorizeRequest,
    decision: [
      throwOn(
        (event) => event.type === 'ProductItemOutOfStock',
        (event) => new ProductItemOutOfStockError(event.type),
      ),
    ],
  },
});

Configuration

middleware accepts a decision middleware array or an object:

Form	Behavior
`middleware: Middleware[]`	Applies the array to every decision.
`middleware: { decision: Middleware[] }`	Equivalent to the array form.
`middleware: { beforeAll, decision }`	Runs `beforeAll` once before aggregation/retries.
`middleware: { afterAll, decision }`	Runs `afterAll` once after the successful invocation.

beforeAll receives the complete handler input and an operation context containing streamName and handleOptions. Raw CommandHandler supplies the decision or decision array. DeciderCommandHandler supplies the command or command array. WorkflowHandler supplies the input message.

afterAll receives the final handler result and the operation context. Both lifecycle callbacks run outside retry processing. afterAll runs only after a successful invocation; throwing from it does not roll back events or messages that were already appended.

Decision middleware runs for every decision and every retry attempt. Its arguments depend on the handler:

Handler	First argument	Second argument
`CommandHandler`	Current state	None
`DeciderCommandHandler`	Command	Current state
`WorkflowHandler`	Input message	Current state

Helpers

Helper	Matching behavior
`before(callback)`	Runs `callback` before the decision.
`after(callback)`	Runs `callback` with the handling result after the decision.
`skipOn(predicate)`	Returns `SKIP`.
`stopOn(predicate)`	Returns `STOP`.
`rejectOn(predicate)`	Returns `REJECT`.
`stopAfter(predicate)`	Returns `APPEND_AND_STOP`.
`throwOn(predicate, map)`	Throws the error created by `map` before the batch is appended.

A helper predicate runs for each output. If one output matches, the helper applies its result to every output returned by that decision.

Custom middleware

Custom middleware can inspect the input, state, and result before selecting how the decision is handled:

const { appendAndStop, reject, skip, stop } = DecisionHandling.result;

const selectCartHandling: Middleware<CartCommand, Cart, CartEvent> =
  (next) => async (command, state) => {
    const result = await next(command, state);

    if (
      result.outputs.some((event) => event.type === 'ProductItemAlreadyInCart')
    )
      return skip(result.outputs);
    if (
      result.outputs.some(
        (event) => event.type === 'ShoppingCartConfirmationFailed',
      )
    )
      return stop(result.outputs);
    if (result.outputs.some((event) => event.type === 'ProductItemOutOfStock'))
      return reject(result.outputs);
    if (
      result.outputs.some(
        (event) => event.type === 'ShoppingCartItemLimitReached',
      )
    )
      return appendAndStop(result.outputs);
    return result;
  };

Use afterAll for logging, metrics or other measurements that need the returned result. Use event-store hooks for commit instrumentation that must reflect storage-level behavior.

When the decision returns an empty array, nothing is appended and the result carries newEvents: [], createdNewStream: false, and the current version:

// 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),
  );

Stream ID Mapping

mapToStreamId derives the stream name from the business id; the business id is still passed to the decision. It defaults to the identity function.

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

Optimistic Concurrency

The handler appends with an expected version and fails with ExpectedVersionConflictError (a ConcurrencyError) when the stream has moved on since it was read.

Version read from the stream

With no expectedStreamVersion passed, the handler expects the version it read from the stream at the start of the call. nextExpectedStreamVersion in the result is the version after the append.

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

Explicit expected version

expectedStreamVersion sets the expected version explicitly, such as one carried in a client's If-Match header. The append succeeds only if the stream is still at that version.

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

Expecting a new stream

STREAM_DOES_NOT_EXIST as the expected version makes the append succeed only if the stream does not exist yet.

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

Retry

retry re-runs the decision and append when a version conflict is retryable. CommandHandlerRetryOptions takes three forms:

{ onVersionConflict: true } applies the default policy.
{ onVersionConflict: number } applies the default policy with a different retry count.
AsyncRetryOptions is a full custom policy, including its own shouldRetryError.

Left undefined, retries are disabled. A per-call retry in the handle options overrides the handler-level policy. For deciding which errors are transient enough to retry, see Retry a Transient Failure in the Error Handling guide.

The default policy retries only on ExpectedVersionConflictError:

Field	Value
`retries`	`3`
`minTimeout`	`100` ms
`factor`	`1.5`

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

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

Idempotence

Re-running a command does not duplicate its effect. Two behaviours combine:

A decision returns an empty array once its outcome is already present in the state, so a repeat appends nothing. See Decisions.
Optimistic concurrency rejects a stale write. A retry carrying the version from before the first append, or STREAM_DOES_NOT_EXIST for a creation, fails with ExpectedVersionConflictError rather than appending twice. See Optimistic Concurrency.

The handler keeps no deduplication store and no idempotency key; idempotence comes from the decision and the expected version.

Error Handling

A decision throws to reject a command; the handler appends nothing and propagates the error unchanged. A version conflict is thrown by the append.

Error	Code	Raised when
`ValidationError`	`400`	The command carries invalid input.
`IllegalStateError`	`403`	The command is not valid for the current state.
`ExpectedVersionConflictError` (`ConcurrencyError`)	`412`	The stream moved on; carries `expected` and `current` versions as strings.

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

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

Advanced

Schema versioning

schema.versioning.upcast maps a stored event to the current StreamEvent shape on read; schema.versioning.downcast maps a StreamEvent back to its stored shape on write. Together they carry a stream through event schema evolution.

Pass the stored shape as the last type parameter, EventPayloadType, on either handler so both callbacks are checked against it. It defaults to StreamEvent, which is only correct while the stored and current shapes still match.

Define the transform against both shapes. Here older events stored the product item's fields flat and kept addedAt as the ISO string JSON persists a Date as; the current shape groups those fields and rebuilds the Date:

// The shape older events were stored with: productId and quantity as flat
// fields, addedAt as the ISO-8601 string JSON persists a Date as.
type ProductItemAddedV1 = Event<
  'ProductItemAdded',
  { productId: string; quantity: number; addedAt: string }
>;

// The current shape: the product item is grouped under its own object and
// addedAt is a Date. The event type name stays the same across versions.
type ProductItemAdded = Event<
  'ProductItemAdded',
  { productItem: { productId: string; quantity: number }; addedAt: Date }
>;

// The events the domain works with, and the shapes the stream can hold: a
// stream written across the change carries both versions.
type ShoppingCartEvent = ProductItemAdded;
type StoredShoppingCartEvent = ProductItemAddedV1 | ProductItemAdded;

const upcast = (event: StoredShoppingCartEvent): ShoppingCartEvent => {
  if ('productItem' in event.data)
    return { type: 'ProductItemAdded', data: event.data };

  return {
    type: 'ProductItemAdded',
    data: {
      productItem: {
        productId: event.data.productId,
        quantity: event.data.quantity,
      },
      addedAt: new Date(event.data.addedAt),
    },
  };
};

Then register it under schema.versioning on the handler:

const handle = DeciderCommandHandler<
  ShoppingCart,
  AddProductItem,
  ShoppingCartEvent,
  StoredShoppingCartEvent
>({
  decide,
  evolve,
  initialState,
  schema: { versioning: { upcast } },
});

Serialization

serialization.serializer replaces the default JSON serialiser used to read and write events; serialization.serializerOptions configures the default one.

Observability

name labels the handler in traces and metrics. commandType sets the default command type used when a call passes none. observability (CommandObservabilityConfig) configures tracing and metrics for the handler.

Type Source

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

Command Handler ​

Overview ​

Construction ​

CommandHandlerOptions ​

Calling the Handler ​

HandleOptions ​

Decisions ​

Result ​

Decision Middleware ​

Configuration ​

Helpers ​

Custom middleware ​

Stream ID Mapping ​

Optimistic Concurrency ​

Version read from the stream ​

Explicit expected version ​

Expecting a new stream ​

Retry ​

Idempotence ​

Error Handling ​

Advanced ​

Schema versioning ​

Serialization ​

Observability ​

Type Source ​

See also ​