Defines the behaviour an event handler must implement and provides a convenience macro that implements the behaviour, allowing you to handle only the events you are interested in processing.

You should start your event handlers using a Supervisor to ensure they are restarted on error.

Example

defmodule ExampleHandler do
  use Commanded.Event.Handler,
    application: ExampleApp,
    name: "ExampleHandler"

  def handle(%AnEvent{..}, _metadata) do
    # ... process the event
    :ok
  end
end

Start your event handler process (or use a Supervisor):

{:ok, _handler} = ExampleHandler.start_link()

Event handler name

The name you specify is used when subscribing to the event store. You must use a unique name for each event handler and process manager you start. Also, you should not change the name once the handler has been deployed. A new subscription will be created if you change the name and the event handler will receive already handled events.

You can use the module name of your event handler using the __MODULE__ special form:

defmodule ExampleHandler do
  use Commanded.Event.Handler,
    application: ExampleApp,
    name: __MODULE__
end

Subscription options

You can choose to start the event handler's event store subscription from :origin, :current position, or an exact event number using the start_from option. The default is to use the origin so your handler will receive all events.

Use the :current position when you don't want newly created event handlers to go through all previous events. An example would be adding an event handler to send transactional emails to an already deployed system containing many historical events.

The start_from option only applies when the subscription is initially created, the first time the handler starts. Whenever the handler restarts the subscription will resume from the next event after the last successfully processed event. Restarting an event handler does not restart its subscription.

Example

Set the start_from option (:origin, :current, or an explicit event number) when using Commanded.Event.Handler:

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

You can optionally override :start_from by passing it as option when starting your handler:

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

Subscribing to an individual stream

By default event handlers will subscribe to all events appended to any stream. Provide a subscribe_to option to subscribe to a single stream.

defmodule ExampleHandler do
  use Commanded.Event.Handler,
    application: ExampleApp,
    name: __MODULE__,
    subscribe_to: "stream1234"
end

This will ensure the handler only receives events appended to that stream.

Runtime event handler configuration

Runtime options can be provided to the event handler's start_link/1 function or its child spec. The init/1 callback function can also be used to define runtime configuration.

Example

Provide runtime configuration to start_link/1:

{:ok, _pid} =
  ExampleHandler.start_link(
    application: ExampleApp,
    name: "ExampleHandler"
  )

Or when supervised:

Supervisor.start_link([
  {ExampleHandler, application: ExampleApp, name: "ExampleHandler"}
], strategy: :one_for_one)

Event handler state

An event handler can define and update state which is held in the GenServer process memory. It is passed to the handle/2 (or handle_batch/1) function as part of the metadata using the :state key. The state is transient and will be lost whenever the process restarts.

Initial state can be set in the init/1 callback function by adding a :state key to the config. It can also be provided when starting the handler process:

ExampleHandler.start_link(state: initial_state)

Or when supervised:

Supervisor.start_link([
  {ExampleHandler, state: initial_state}
], strategy: :one_for_one)

State can be updated by returning {:ok, new_state} from any handle/2 (or handle_batch/2) function. Returning an :ok reply will keep the state unchanged.

Handler state is also included in the Commanded.Event.FailureContext struct passed to the error/3 callback function.

Example

defmodule StatefulEventHandler do
  use Commanded.Event.Handler,
    application: ExampleApp,
    name: __MODULE__

  def init(config) do
    config = Keyword.put_new(config, :state, %{})

    {:ok, config}
  end

  def handle(event, metadata) do
    %{state: state} = metadata

    new_state = mutate_state(state)

    {:ok, new_state}
  end
end

Concurrency

An event handler may be configured to start multiple processes to handle the events concurrently. By default one process will be started, processing events one at a time in order. The :concurrency option determines how many event handler processes are started. It must be a positive integer.

Note with concurrent processing events will likely by processed out of order. If you need to enforce an order, such as per stream or by using a field from an event, you can define a partition_by/2 callback function in the event handler module. The function will receive each event and its metadata and must return a consistent term indicating the event's partition. Events which return the same term are guaranteed to be processed in order by the same event handler instance. While events with different partitions may be processed concurrently by another instance. An attempt will be made to distribute events as evenly as possible to all running event handler instances.

Only :eventual consistency is supported when multiple handler processes are configured with a :concurrency of greater than one.

Example

defmodule ConcurrentProcssingEventHandler do
  alias Commanded.EventStore.RecordedEvent

  use Commanded.Event.Handler,
    application: ExampleApp,
    name: __MODULE__,
    concurrency: 10

  def init(config) do
    # Fetch the index of this event handler instance (0..9 in this example)
    index = Keyword.fetch!(config, :index)

    {:ok, config}
  end

  def handle(event, metadata) do
    :ok
  end

  # Partition events by their stream
  def partition_by(event, metadata) do
    %{stream_id: stream_id} = metadata

    stream_id
  end
end

Consistency

For each event handler you can define its consistency, as one of either :strong or :eventual.

This setting is used when dispatching commands and specifying the consistency option.

When you dispatch a command using :strong consistency, after successful command dispatch the process will block until all event handlers configured to use :strong consistency have processed the domain events created by the command. This is useful when you have a read model updated by an event handler that you wish to query for data affected by the command dispatch. With :strong consistency you are guaranteed that the read model will be up-to-date after the command has successfully dispatched. It can be safely queried for data updated by any of the events created by the command.

The default setting is :eventual consistency. Command dispatch will return immediately upon confirmation of event persistence, not waiting for any event handlers.

Note strong consistency does not imply a transaction covers the command dispatch and event handling. It only guarantees that the event handler will have processed all events produced by the command: if event handling fails the events will have still been persisted.

Example

Define an event handler with :strong consistency:

defmodule ExampleHandler do
  use Commanded.Event.Handler,
    application: ExampleApp,
    name: "ExampleHandler",
    consistency: :strong
end

Batching

Event handlers can process events in batches rather than one at a time. This allows writing to a target system in a single transaction and acknowledging all events at once, reducing overhead.

To enable batching, set the batch_size option and implement handle_batch/1 instead of handle/2. The callback receives a list of {event, metadata} tuples. On returning :ok, all events in the batch are acknowledged. There is no partial acknowledgement.

Batching and concurrency are not supported together; setting both will raise a compilation error.

batch_size

The batch_size option controls the EventStore subscription's in-flight buffer — how many events the subscription can deliver without waiting for acknowledgement. It is not an accumulation mechanism at the handler level. The handler processes each delivery from the subscription immediately.

During live operation, when events arrive one at a time, handle_batch/1 is called with a single event — regardless of the configured batch_size. Larger batches form naturally only when the subscription has multiple events queued, such as during catch-up replay, back-pressure recovery, or bulk appends.

When batch_timeout is configured, batch_size also serves as the maximum size of the handler's internal buffer — once the buffer reaches batch_size events, it is flushed immediately regardless of the timeout.

batch_timeout

The batch_timeout option adds time-based buffering at the handler level. When set, events are accumulated in the handler process and flushed when either condition is met first:

• batch_size events have accumulated, OR
• batch_timeout milliseconds have elapsed since the first buffered event

This enables use cases such as accumulating changes before writing them in one go to a projection, batching bulk database inserts, or reducing per-event overhead for side effects — even when events arrive individually during steady-state operation.

batch_timeout accepts a positive integer (milliseconds) or :infinity (the default — no buffering, backwards compatible with previous behaviour). Requires batch_size to be configured.

Example

defmodule OrderProjector do
  use Commanded.Event.Handler,
    application: ExampleApp,
    name: "OrderProjector",
    batch_size: 50,
    batch_timeout: 100

  def handle_batch(events) do
    # Receives up to 50 events, or whatever accumulated within 100ms
    :ok
  end
end

Telemetry

Batch telemetry events include a flush_reason field:

• :size — batch reached batch_size
• :timeout — batch_timeout elapsed before the batch filled
• :immediate — no batch_timeout configured; events processed as delivered

Example

Define an event handler with :strong consistency:

defmodule ExampleHandler do
  use Commanded.Event.Handler,
    application: ExampleApp,
    name: "ExampleHandler",
    consistency: :strong
end

Dynamic application

An event handler's application can be provided as an option to start_link/1. This can be used to start the same handler multiple times, each using a separate Commanded application and event store.

Example

Start an event handler process for each tenant in a multi-tenanted app, guaranteeing that the data and processing remains isolated between tenants.

for tenant <- [:tenant1, :tenant2, :tenant3] do
  {:ok, _app} = MyApp.Application.start_link(name: tenant)
  {:ok, _handler} = ExampleHandler.start_link(application: tenant)
end

Typically you would start the event handlers using a supervisor:

children =
  for tenant <- [:tenant1, :tenant2, :tenant3] do
    {ExampleHandler, application: tenant}
  end

Supervisor.start_link(children, strategy: :one_for_one)

The above example requires three named Commanded applications to have already been started.

Telemetry

• [:commanded, :event, :handle, :start]

  • Description: Emitted when an event handler starts handling an event
  • Measurements: %{system_time: integer()}
  • Metadata: %{application: Commanded.Application.t(), context: map(), handler_name: String.t(), handler_module: atom(), handler_state: map(), recorded_event: RecordedEvent.t()}

• [:commanded, :event, :handle, :stop]

  • Description: Emitted when an event handler stops handling an event
  • Measurements: %{duration: non_neg_integer()}
  • Metadata: %{:application => Commanded.Application.t(), :context => map(), :handler_name => String.t(), :handler_module => atom(), :handler_state => map(), :recorded_event => RecordedEvent.t(), optional(:error) => any()}

• [:commanded, :event, :handle, :exception]

  • Description: Emitted when an event handler raises an exception
  • Measurements: %{duration: non_neg_integer()}
  • Metadata: %{application: Commanded.Application.t(), context: map(), handler_name: String.t(), handler_module: atom(), handler_state: map(), recorded_event: RecordedEvent.t(), kind: :throw | :error | :exit, reason: any(), stacktrace: list()}

• [:commanded, :event, :batch, :start]

  • Description: Emitted when an event handler starts handling a batch of events
  • Measurements: %{system_time: integer()}
  • Metadata: %{application: Commanded.Application.t(), context: map(), handler_name: String.t(), handler_module: atom(), handler_state: map(), first_event_id: binary(), last_event_id: binary(), event_count: integer(), flush_reason: :size | :timeout | :immediate}

• [:commanded, :event, :batch, :stop]

  • Description: Emitted when an event handler stops handling a batch of events
  • Measurements: %{duration: non_neg_integer()}
  • Metadata: %{application: Commanded.Application.t(), context: map(), handler_name: String.t(), handler_module: atom(), handler_state: map(), first_event_id: binary(), last_event_id: binary(), event_count: integer(), recorded_event: RecordedEvent.t() | nil, flush_reason: :size | :timeout | :immediate, optional(:error) => any()}

• [:commanded, :event, :batch, :exception]

  • Description: Emitted when an event batch handler raises an exception
  • Measurements: %{duration: non_neg_integer()}
  • Metadata: %{application: Commanded.Application.t(), context: map(), handler_name: String.t(), handler_module: atom(), handler_state: map(), recorded_event: => RecordedEvent.t() | nil, first_event_id: binary(), last_event_id: binary(), event_count: integer(), flush_reason: :size | :timeout | :immediate, kind: :throw | :error | :exit, reason: any(), optional(:stacktrace) => list()}