Commanded Documentation

Commanded provides building blocks for creating Elixir applications using the Command Query Responsibility Segregation (CQRS) and Event Sourcing (ES) patterns.

Key architectural characteristics:

• Write Model (Command Dispatch): Uses strong consistency. Receiving an :ok reply from dispatch indicates the command was successfully handled and domain events were persisted.
• Read Model (Event Handlers/Projectors): Uses eventual consistency by default, though you can opt into strong consistency for individual handlers or dispatch as needed.

Core components include:

• Aggregates: Handle commands, protect business invariants, and produce domain events.
• Commands: Data structures representing intent.
• Events: Data structures representing facts that have occurred.
• Routers: Map commands to specific aggregates.
• Applications: Host aggregates and supporting processes.
• Event Handlers: React to events to update read models or trigger side effects.

When dispatching commands, you can choose how much to wait for side effects (like event handlers or process managers) to complete:

• :eventual (default): Does not block. Returns immediately without waiting for any handlers. This offers low latency but means read models might be stale.
• :strong: Blocks until all strongly consistent event handlers and process managers have processed the resulting events. This ensures read models are up-to-date but increases latency. If handlers are not configured for strong consistency, this has no effect.
• Explicit list: You can pass a list of specific handler modules or names [Handler1, Handler2] to wait only for those specific handlers.

Handling Consistency Failures

If you use :strong consistency, a dispatch might return {:error, :consistency_timeout}. This means the command was successful, but the handlers did not finish within the dispatch_consistency_timeout period.

Configuration

You can set the global default consistency in your application config:

config :commanded, default_consistency: :strong

And the global timeout for strong consistency dispatch:

config :commanded, dispatch_consistency_timeout: 10_000

# Dispatch with strong consistency
case BankApp.dispatch(command, consistency: :strong) do
  :ok -> # ... all ok
  {:error, :consistency_timeout} -> # command ok, handlers have not yet executed
end

# Dispatch with specific handlers
:ok = BankApp.dispatch(command, consistency: [ExampleHandler, AnotherHandler])

Upcasting allows you to transform events at runtime before they reach a consumer (such as an aggregate, event handler, or process manager). This enables you to evolve your event schema without needing to migrate historical data manually. Handlers only need to support the latest version of an event if an upcaster is present.

You can use upcasting to:

1. Change the shape of an event: e.g., renaming a field.
2. Change the type of an event: e.g., replacing a HistoricalEvent with a NewEvent.

Implement the Commanded.Event.Upcaster protocol for your event types.

# Example: Renaming a field
defimpl Commanded.Event.Upcaster, for: AnEvent do
  def upcast(%AnEvent{} = event, _metadata) do
    %AnEvent{name: name} = event

    %AnEvent{event | first_name: name}
  end
end

# Example: Changing event type
defimpl Commanded.Event.Upcaster, for: HistoricalEvent do
  def upcast(%HistoricalEvent{} = event, _metadata) do
    %HistoricalEvent{id: id, name: name} = event

    %NewEvent{id: id, name: name}
  end
end

A router module is responsible for mapping commands to their respective command handlers and aggregate modules. You define a router by using Commanded.Commands.Router in your module.

There are several ways to configure routing:

1. Standard Dispatch: Explicitly map a command to a handler and an aggregate, specifying the identity field.
2. Direct Dispatch: Skip the handler module and dispatch commands directly to the aggregate module.
3. Multi-command Registration: Register a list of commands to a single destination at once.
4. Identity Helper: Use the identify macro to define the identity field for an aggregate once, rather than repeating it in every dispatch call.

# Standard dispatch
defmodule BankRouter do
  use Commanded.Commands.Router

  dispatch OpenAccount, to: OpenAccountHandler, aggregate: BankAccount, identity: :account_number
  dispatch DepositMoney, to: DepositMoneyHandler, aggregate: BankAccount, identity: :account_number
end

# Succinct configuration using identify and multi-command registration
defmodule BankRouter do
  use Commanded.Commands.Router

  identify BankAccount, by: :account_number
  dispatch [OpenAccount, DepositMoney], to: BankAccount
end

When starting an event handler, you can specify where in the event stream it should begin processing using the start_from option. This can be set in the module definition or overridden during start_link/1.

Supported values for start_from:

• :origin: (Default) Starts from the very beginning of the event store. The handler will receive all historical events.
• :current: Starts from the current position in the event store. Use this when adding new handlers to existing systems where you do not want to replay historical events (e.g., a new email notification handler).
• integer: An explicit event number (e.g., 1234) to start from a specific point.

# Via module definition
defmodule ExampleHandler do
  use Commanded.Event.Handler,
    application: ExampleApp,
    name: "ExampleHandler",
    start_from: :origin
end

# Via start_link override
{:ok, _handler} = ExampleHandler.start_link(start_from: :current)

An aggregate is a core building block in Commanded, composed of three main parts:

1. Aggregate State: A struct (defined via defstruct) that holds the current state of the aggregate.
2. Command Functions: Public functions that receive the current state and a command, then return resulting domain events. These functions enforce business rules.
3. State Mutators: An apply/2 function that receives the current state and a domain event, returning a new modified state.

In a CQRS architecture, all state must be derived from published domain events. If an aggregate needs data owned by another aggregate, you should look up that data from a projection and include it in the command before dispatching.

Aggregates can be used in two ways:

• Via a Command Handler: You define domain-specific functions (e.g., open_account/3) and use a separate module implementing @behaviour Commanded.Commands.Handler to route commands to them.
• Directly via execute/2: You implement the @behaviour Commanded.Aggregates.Aggregate and define an execute/2 function. This allows commands to be routed directly to the aggregate without an intermediate handler.

defmodule ExampleAggregate do
  defstruct [:uuid, :name]

  # Command function
  def execute(%ExampleAggregate{uuid: nil}, %Create{} = command) do
    event = %Created{uuid: command.uuid, name: command.name}
    {:ok, event}
  end

  # State mutator
  def apply(%ExampleAggregate{}, %Created{} = event) do
    %ExampleAggregate{uuid: event.uuid, name: event.name}
  end
end