Command Handling

A command is an intention to run business logic: add an item to a cart, confirm an order. Handling it means loading the stream, rebuilding the current state from its events, running your decision, and appending the events the decision returns under an optimistic concurrency check. This guide shows you how to set up a command handler, write the decision that holds your business logic, and control the concurrency, retries, and HTTP layer around it. For the full API surface, see the Command Handler reference.

Create a Handler

Configure a handler once with evolve and initialState, then reuse it across the application:

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

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

Write the Decision

A decision receives the current state and the command, and returns the events to append. It holds your business logic: it decides what happens, and it is the one place a business rule is enforced.

Return one event

When a command produces a single outcome, return one event:

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

Return several events

Return an array when one command produces several outcomes. The handler appends them together in a single write, so the stream cannot be 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 } },
  ];
};

Guard against an invalid state

Throw IllegalStateError before building any event, so an invalid transition never produces one. The same decision returns an empty array when the state leaves nothing to do, which keeps re-sending the command safe:

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

Validate the input

Throw ValidationError when the command carries bad input, again before producing any event:

const applyDiscount = (
  command: ApplyDiscount,
  _state: ShoppingCart,
): ShoppingCartEvent => {
  // Reject invalid input before producing any event
  if (command.data.percent <= 0 || command.data.percent > 1)
    throw new ValidationError('Discount percent has to be between 0 and 1');

  return { type: 'DiscountApplied', data: { percent: command.data.percent } };
};

Handle a Command

Call the handler with the event store, the stream id, and your decision. It loads the stream, runs the decision, and appends the result, returning nextExpectedStreamVersion to carry into the next call:

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

Run Several Decisions in Order

Pass an array of decisions to run them in sequence. Each runs on the state left by the ones 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),
  ]);

Control Which Decision Results Are Appended

Suppose a command reports that a product is out of stock. You may need to return that outcome to the caller without appending it, or stop the remaining commands after appending the changes made so far.

You can configure a reusable rule on either CommandHandler or DeciderCommandHandler. The rule runs for each command outcome and chooses whether to append it, continue with the next command, or reject the whole batch. Outcomes that are not appended are still returned, so you can use them in the response.

Emmett calls these rules decision middleware. Use the helpers below for the common cases, or write custom middleware when you need different behavior.

The business logic remains responsible only for describing what happened. These decisions return ordinary cart events for unavailable stock, duplicate products, item limits, and failed payment authorization:

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

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

const addProductItemWithStock = (
  command: AddProductItem,
  availableQuantity: number,
  state: ShoppingCart,
): ShoppingCartEvent =>
  command.data.productItem.quantity > availableQuantity
    ? {
        type: 'ProductItemOutOfStock',
        data: {
          productId: command.data.productItem.productId,
          requestedQuantity: command.data.productItem.quantity,
          availableQuantity,
        },
      }
    : addProductItem(command, state);

const addProductItemWithinLimit = (
  command: AddProductItem,
  maximumItems: number,
  state: ShoppingCart,
): ShoppingCartEvent =>
  state.productItems.length >= maximumItems
    ? {
        type: 'ShoppingCartItemLimitReached',
        data: {
          maximumItems,
          requestedProductId: command.data.productItem.productId,
        },
      }
    : addProductItem(command, state);

const addProductItemOnce = (
  command: AddProductItem,
  state: ShoppingCart,
): ShoppingCartEvent =>
  state.productItems.some(
    (item) => item.productId === command.data.productItem.productId,
  )
    ? {
        type: 'ProductItemAlreadyInCart',
        data: { productId: command.data.productItem.productId },
      }
    : addProductItem(command, state);

const confirmAfterPaymentAuthorization = (
  command: ConfirmShoppingCart,
  paymentAuthorized: boolean,
  state: ShoppingCart,
): ShoppingCartEvent[] | ShoppingCartEvent =>
  paymentAuthorized
    ? confirm(command, state)
    : {
        type: 'ShoppingCartConfirmationFailed',
        data: { reason: 'PaymentAuthorizationFailed' },
      };

The handler configuration below decides which of those events belong in the shopping-cart stream and whether the rest of a command batch should run.

Reject the complete batch

Use rejectOn when none of the changes from the current call should be appended after a matching outcome. This configuration rejects the complete batch when a decision returns ProductItemOutOfStock:

const handle = CommandHandler<ShoppingCart, ShoppingCartEvent>({
  evolve,
  initialState,
  middleware: [rejectOn((event) => event.type === 'ProductItemOutOfStock')],
});

rejectOn leaves the cart unchanged and does not run later decisions. It still returns ProductItemOutOfStock, so the endpoint can use the available quantity to return status 409. Use this when the commands belong to one operation and a failure in any of them should cancel the whole operation. Map Returned Events to Error Responses shows the complete Express route.

If one command produces several events, they remain one unit. A match rejects all events produced by that command.

Commit earlier decisions and stop

Use stopOn when earlier changes should be appended, but the matching outcome and later commands should not be. This configuration stops when payment authorization returns ShoppingCartConfirmationFailed:

const handle = CommandHandler<ShoppingCart, ShoppingCartEvent>({
  evolve,
  initialState,
  middleware: [
    stopOn((event) => event.type === 'ShoppingCartConfirmationFailed'),
  ],
});

Use stopOn when earlier commands may stand on their own. The caller still receives ShoppingCartConfirmationFailed, so it can explain why processing stopped. Use rejectOn instead when those earlier changes must also be cancelled.

Skip one outcome and continue

Use skipOn when you want to ignore the matching outcome and continue. This configuration skips ProductItemAlreadyInCart and runs the next decision:

const handle = CommandHandler<ShoppingCart, ShoppingCartEvent>({
  evolve,
  initialState,
  middleware: [
    skipOn((event) => event.type === 'ProductItemAlreadyInCart'),
  ],
});

The caller receives ProductItemAlreadyInCart, but that event is not appended to the shopping-cart stream. Processing continues, and the confirmation sees a cart with one copy of the product.

Record a terminal failure and stop

Use stopAfter when the matching outcome should be appended and no later command should run. This configuration appends ShoppingCartItemLimitReached before stopping:

const handle = CommandHandler<ShoppingCart, ShoppingCartEvent>({
  evolve,
  initialState,
  middleware: [
    stopAfter((event) => event.type === 'ShoppingCartItemLimitReached'),
  ],
});

ProductItemAdded and ShoppingCartItemLimitReached are appended together. The third product decision does not run. stopOn would omit ShoppingCartItemLimitReached; stopAfter includes it.

Turn a produced event into an exception

Use throwOn when the matching outcome should enter an existing exception-based error path, such as HTTP Problem Details mapping:

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

Nothing from the call is appended. Unlike rejectOn, throwOn does not return the matching event to the caller; the created error follows the application's exception-handling path.

Check the complete input before handling

Use middleware.beforeAll when a check should run once for the whole call instead of once per command. Request authorization is one example. It runs before the current state is read. If it throws, no commands run and nothing is appended.

For CommandHandler, it receives the decision or decision array. For DeciderCommandHandler, it receives the command or command array. For WorkflowHandler, it receives the input message. The second argument contains the resolved stream name and handle options.

Put checks that need a command or rebuilt state in middleware.decision. Pass the decision middleware array directly as middleware when beforeAll is not needed.

Observe the completed result

Use middleware.afterAll when a measurement or notification should describe the completed operation rather than each command separately. It receives the final result, so it can record values such as the number of appended events or the resulting stream version. The decider example below records the appended-event count.

Do not use afterAll for validation or for anything that must prevent the append. The operation has already completed, so an error from this callback cannot undo it. Use the event-store hooks when you need to observe each store commit rather than the completed handler result.

Use middleware with DeciderCommandHandler

The handling choices shown above work the same way with CommandHandler and DeciderCommandHandler. The difference is what each middleware receives. Raw command handling supplies the current state because the command is captured by the decision function. DeciderCommandHandler supplies the command and current state, so a shared handler can authorize or measure commands without wrapping every call.

Use before for work before decide, and after for work based on its result:

const handle = DeciderCommandHandler<Cart, CartCommand, CartEvent>({
  evolve,
  initialState,
  decide,
  middleware: {
    beforeAll: authorizeCommands,
    afterAll: recordInvocation,
    decision: [
      before(authorizeCommand),
      after((result, command) => {
        logResult(command, result);
        return result;
      }),
    ],
  },
});

const commands: CartCommand[] = [
  { type: 'AddItem', data: { productId: 'product-1' } },
  { type: 'Confirm', data: { confirmedAt: new Date() } },
];
const result = await handle(eventStore, shoppingCartId, commands);

Callbacks may omit state, as authorizeCommand does here. Accept it as the second argument when needed: before((command, state) => ...). Decision middleware runs again on concurrency retry.

Write custom decision middleware

Use custom middleware when one handler needs several handling rules or when a helper predicate is not enough. The middleware receives next, calls it with the command and current state, then returns the handling selected for the produced outcome:

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

This middleware uses the same rules as the preceding examples:

• ProductItemAlreadyInCart is returned but not appended, and the next command runs.
• ShoppingCartConfirmationFailed is returned without being appended; earlier changes are appended and later commands do not run.
• ProductItemOutOfStock rejects the complete command batch.
• ShoppingCartItemLimitReached is appended before the remaining commands are stopped.
• Other outcomes keep the default append-and-continue behavior returned by next.

The first configured middleware is the outermost wrapper. Put authorization or timing middleware before outcome-selection middleware when it must observe the complete decision call.

Control Concurrency

By default the handler reuses the version it read from the stream, so you do not pass one and concurrent writers cannot overwrite each other:

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

Guard with a client's version

When the version comes from outside, such as a client's If-Match header, pass it as expectedStreamVersion. 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 },
);

Create a stream exactly once

Pass STREAM_DOES_NOT_EXIST so the append succeeds only when the stream does not exist yet:

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

A stale version throws ExpectedVersionConflictError. See Optimistic Concurrency in the reference for the full behaviour.

Retry on Conflict

Set retry so Emmett re-runs the handler when a version conflict occurs. { onVersionConflict: true } applies the default policy:

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

For a different backoff, or to retry on errors other than version conflicts, pass your own shouldRetryError:

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

Make decision middleware retry-safe

beforeAll runs once before retry processing. Decision middleware and the decision run again for every attempt:

const handleWithRetrySafeMiddleware = CommandHandler<
  ShoppingCart,
  ShoppingCartEvent
>({
  evolve,
  initialState,
  middleware: {
    beforeAll: authorizeRequest,
    afterAll: () => {
      measuredInvocations++;
    },
    decision: [before(recordDecisionMetric)],
  },
  retry: { onVersionConflict: true },
});

await handleWithRetrySafeMiddleware(eventStore, shoppingCartId, (state) =>
  addProductItem(product, state),
);

The caller receives only the outcome of the attempt that completes. Any external work performed by decision middleware may run more than once, so make it safe to repeat. Use beforeAll for a check that must run once and does not need the rebuilt state. afterAll runs once after an attempt completes and receives its final result.

Make It Idempotent

The same command can reach the handler twice. A shopper double-clicks "Add to cart", a request times out and the browser sends it again, a retry re-runs it after a conflict. Handling it twice shouldn't add the product twice, and two things you already have keep it safe.

First, a decision that returns [] once its outcome is already in the state. Cancelling a cart that is already cancelled, for instance, has nothing left 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 } },
  ];
};

Resending the command is then a no-op: the second call appends nothing and returns the version unchanged.

// cancel once
const { nextExpectedStreamVersion: cancelledVersion } = await handleCommand(
  eventStore,
  shoppingCartId,
  (state) => cancel(cancelCart, state),
);

// resending the same command is a no-op: the cart is already cancelled,
// so the decision returns [] and nothing is appended
const { newEvents, nextExpectedStreamVersion, createdNewStream } =
  await handleCommand(eventStore, shoppingCartId, (state) =>
    cancel(cancelCart, state),
  );

Second, optimistic concurrency. When you carry the version, a duplicate write that lost the race fails with ExpectedVersionConflictError instead of appending twice, and STREAM_DOES_NOT_EXIST makes stream creation happen exactly once. The no-op decision and the version guard together make a resent command safe.

Integrate with Express

emmett-expressjs wraps a handler with on. Fetch any data the decision needs, such as the unit price, before calling the handler and pass it into the command, which 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();
  }),
);

When decision middleware returns a business failure, use ResponseFromEvents to return the response expected by Express on without assuming the failure is the final event. The failure callback returns an ordinary response helper such as Conflict(...). If no event selects a failure response, success may be a status code or a callback returning NoContent(...), Created(...), or another response helper. The callback receives the complete handler result when headers such as an ETag depend on it:

router.post(
  '/shopping-carts/:shoppingCartId/product-items',
  on(async (request: AddProductItemRequest) => {
    const result = await handleAddProductItem(
      eventStore,
      request.params.shoppingCartId,
      {
        type: 'AddProductItem',
        data: {
          productId: String(request.body.productId),
          quantity: Number(request.body.quantity),
          availableQuantity: 2,
        },
      },
    );

    return ResponseFromEvents({
      events: result,
      success: 204,
      failure: (event) => {
        switch (event.type) {
          case 'ProductItemOutOfStock':
            return Conflict({
              problemDetails: `Only ${event.data.availableQuantity} items are available`,
            });
        }
        return undefined;
      },
    });
  }),
);

Carry the version in an ETag

To expose optimistic concurrency over HTTP, read the client's version from the If-Match header with getETagValueFromIfMatch, pass it as expectedStreamVersion, and return the new version with toWeakETag. A stale If-Match maps to HTTP 412 Precondition Failed:

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

Best Practices

Keep the Decision Pure

A decision that only reads state and returns events is easy to test and safe to retry. When it needs external data, such as a price or an exchange rate, fetch it before the handler and pass it in. Avoid I/O inside the decision: on a version conflict the whole handler re-runs, so the call fires 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,
    );
  },
);

Type Events as a Discriminated Union

Declare your events as a discriminated union and pass it as the handler's StreamEvent. evolve and every decision are then checked against it, and unknown event types surface at compile time:

type ShoppingCartEvent =
  | ProductItemAdded
  | DiscountApplied
  | ShoppingCartConfirmed
  | ShoppingCartCancelled
  | ProductItemOutOfStock
  | ShoppingCartItemLimitReached
  | ProductItemAlreadyInCart
  | ShoppingCartConfirmationFailed;

Troubleshooting

Version Conflict on Every Write

If every write throws ExpectedVersionConflictError, the expected version you pass is stale. When you track the version yourself, pass the nextExpectedStreamVersion from the previous result into the next call. When a client supplies it, return the new version as an ETag so the next request carries the current one. To let Emmett recover on its own, set retry.

Nothing Is Appended

If a command runs without error yet no events land, the decision returned an empty array. That is the no-op path, taken when the state leaves nothing to do, such as confirming an already-confirmed cart. Check the guard that returns [].

A Side Effect Fires Twice

If an external call inside a decision fires more than once, a version conflict retried the handler and re-ran the decision. Move the I/O out of the decision and pass its result in, as Keep the Decision Pure shows.

Further Readings

• Getting Started - Command Handling
• API Reference: Command Handler
• API Reference: Command
• Decider Pattern
• Event Store
• Error Handling
• Testing
• Optimistic Concurrency for Pessimistic Times