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