Repository: commanded/commanded
Pull request #396 (2020-08-09 23:04:06)

Include `application` and `handler_name` as additional event handler metadata
Include the associated Commanded Application and event handler name as additional system provided metadata key/values. This allows event handlers to determine the current application and handler name when started with these options configured at runtime.

## Example

```elixir
defmodule ExampleEventHandler do
  use Commanded.Event.Handler
  
  @impl Commanded.Event.Handler
  def handle(event, metadata) do
    application = Keyword.fetch!(metadata, :application)
    handler_name = Keyword.fetch!(metadata, :handler_name)

    :ok
  end
end
```

Usage:

```elixir
{:ok, _pid} = ExampleEventHandler.start_link(application: MyApp, name: "ExampleEventHandler1")
{:ok, _pid} = ExampleEventHandler.start_link(application: MyApp, name: "ExampleEventHandler2")
```

The above demonstrates using the same event handler module but configured at runtime with a unique name so that two event store subscriptions are created and both handlers receive events.


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 16ec6e9..ddbdb18 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,7 @@
 ### Enhancements
 
 - Add `init/1` callback function to event handlers and process managers ([#393](https://github.com/commanded/commanded/pull/393)).
+- Include `application` and `handler_name` as additional event handler metadata ([#396](https://github.com/commanded/commanded/pull/396)).
 
 ## v1.1.1
 
diff --git a/guides/Events.md b/guides/Events.md
index 5b3655f..24707e0 100644
--- a/guides/Events.md
+++ b/guides/Events.md
@@ -85,76 +85,11 @@ end
 
 This will ensure the handler only receives events appended to that stream.
 
-### `init/0` callback
+### Event handler callbacks
 
-You can define an `init/0` function in your handler to be called once it has started and successfully subscribed to the event store.
-
-This callback function must return `:ok`, any other return value will terminate the event handler with an error.
-
-```elixir
-defmodule ExampleHandler do
-  use Commanded.Event.Handler,
-    application: ExampleApp,
-    name: "ExampleHandler"
-
-  def init do
-    # optional initialisation
-    :ok
-  end
-
-  def handle(%AnEvent{..}, _metadata) do
-    # ... process the event
-    :ok
-  end
-end
-```
-
-### `error/3` callback
-
-You can define an `error/3` callback function to handle any exceptions or errors returned from your event handler's `handle/2` functions. The `error/3` function is passed the actual error (e.g. `{:error, :failure}`), the failed event, and a failure context.
-
-Use pattern matching on the error and/or failed event to explicitly handle certain errors or events. You can choose to retry, skip, or stop the event handler after an error.
-
-The default behaviour if you don't provide an `error/3` callback is to stop the event handler using the exact error reason returned from the `handle/2` function. You should supervise event handlers to ensure they are correctly restarted on error.
-
-#### Example error handling
-
-```elixir
-defmodule ExampleHandler do
-  use Commanded.Event.Handler,
-    application: ExampleApp,
-    name: __MODULE__
-
-  require Logger
-
-  alias Commanded.Event.FailureContext
-
-  def handle(%AnEvent{}, _metadata) do
-    # Simulate event handling failure
-    {:error, :failed}
-  end
-
-  def error({:error, :failed}, %AnEvent{} = event, %FailureContext{context: context}) do
-    context = record_failure(context)
-
-    case Map.get(context, :failures) do
-      too_many when too_many >= 3 ->
-        # Skip bad event after third failure
-        Logger.warn(fn -> "Skipping bad event, too many failures: " <> inspect(event) end)
-
-        :skip
-
-      _ ->
-        # Retry event, failure count is included in context map
-        {:retry, context}
-    end
-  end
-
-  defp record_failure(context) do
-    Map.update(context, :failures, 1, fn failures -> failures + 1 end)
-  end
-end
-```
+- `c:Commanded.Event.Handler.init/0` - (optional) initialisation callback function called when the handler starts.
+- `c:Commanded.Event.Handler.init/1` - (optional) used to configure the handler before it starts.
+- `c:Commanded.Event.Handler.error/3` - (optional) called when an event handle/2 callback returns an error.
 
 ### Metadata
 
@@ -166,6 +101,8 @@ The `handle/2` function in your handler receives the domain event and a map of m
 
 In addition to the metadata key/values you provide, the following system values will be included in the metadata passed to an event handler:
 
+- `application` - the `Commanded.Application` associated with the event handler.
+- `handler_name` - the name of the event handler.
 - `event_id` - a globally unique UUID to identify the event.
 - `event_number` - a globally unique, monotonically incrementing and gapless integer used to order the event amongst all events.
 - `stream_id` - the stream identity for the event.
diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex
index ac266fc..d83188f 100644
--- a/lib/commanded/aggregates/aggregate.ex
+++ b/lib/commanded/aggregates/aggregate.ex
@@ -276,14 +276,14 @@ defmodule Commanded.Aggregates.Aggregate do
   @doc false
   @impl GenServer
   def handle_info({:events, events}, %Aggregate{} = state) do
-    %Aggregate{lifespan_timeout: lifespan_timeout} = state
+    %Aggregate{application: application, lifespan_timeout: lifespan_timeout} = state
 
     Logger.debug(fn -> describe(state) <> " received events: #{inspect(events)}" end)
 
     try do
       state =
         events
-        |> Upcast.upcast_event_stream()
+        |> Upcast.upcast_event_stream(additional_metadata: %{application: application})
         |> Enum.reduce(state, &handle_event/2)
 
       state = Enum.reduce(events, state, &handle_event/2)
diff --git a/lib/commanded/event/failure_context.ex b/lib/commanded/event/failure_context.ex
index f199d4c..186ee64 100644
--- a/lib/commanded/event/failure_context.ex
+++ b/lib/commanded/event/failure_context.ex
@@ -4,6 +4,8 @@ defmodule Commanded.Event.FailureContext do
 
   The available fields are:
 
+    - `application` - the associated `Commanded.Application`.
+    - `handler_name` - the name of the event handler.
     - `context` - a map that is passed between each failure. Use it to store any
       transient state between failures. As an example it could be used to count
       error failures and stop or skip the problematic event after too many.
@@ -13,12 +15,16 @@ defmodule Commanded.Event.FailureContext do
   """
 
   @type t :: %__MODULE__{
+          application: Commanded.Application.t(),
+          handler_name: String.t(),
           context: map(),
           metadata: map(),
           stacktrace: Exception.stacktrace() | nil
         }
 
   defstruct [
+    :application,
+    :handler_name,
     :context,
     :metadata,
     :stacktrace
diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex
index dcf7d54..4f34fdc 100644
--- a/lib/commanded/event/handler.ex
+++ b/lib/commanded/event/handler.ex
@@ -7,7 +7,7 @@ defmodule Commanded.Event.Handler do
   You should start your event handlers using a [Supervisor](supervision.html) to
   ensure they are restarted on error.
 
-  ## Example
+  ### Example
 
       defmodule ExampleHandler do
         use Commanded.Event.Handler,
@@ -26,8 +26,9 @@ defmodule Commanded.Event.Handler do
 
   ## Event handler name
 
-  The name you specify is used when subscribing to the event store. Therefore
-  you *should not* change the name once the handler has been deployed. A new
+  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.
 
@@ -52,7 +53,7 @@ defmodule Commanded.Event.Handler do
   to send transactional emails to an already deployed system containing many
   historical events.
 
-  The `start_from` option only applies when the subscription is initially
+  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
@@ -107,113 +108,6 @@ defmodule Commanded.Event.Handler do
         {ExampleHandler, application: ExampleApp, name: "ExampleHandler"}
       ], strategy: :one_for_one)
 
-  ## `c:init/0` callback
-
-  You can define an `c:init/0` function in your handler to be called once it has
-  started and successfully subscribed to the event store.
-
-  This callback function must return `:ok`. Any other return value will
-  terminate the event handler with an error.
-
-      defmodule ExampleHandler do
-        use Commanded.Event.Handler,
-          application: ExampleApp,
-          name: "ExampleHandler"
-
-        # Optional initialisation
-        def init do
-          :ok
-        end
-
-        def handle(%AnEvent{..}, _metadata) do
-          # ... process the event
-          :ok
-        end
-      end
-
-    ## `c:init/1` callback
-
-    An `c:init/1` function can be defined in your handler which is used to
-    provide runtime configuration. This callback function must return
-    `{:ok, config}` with the updated config.
-
-    ### Example
-
-    The `c:init/1` function is used to define the handler's application and name
-    based upon a value provided at runtime:
-
-        defmodule ExampleHandler do
-          use Commanded.Event.Handler
-
-          def init(config) do
-            {tenant, config} = Keyword.pop!(config, :tenant)
-
-            config =
-              config
-              |> Keyword.put(:application, Module.concat([ExampleApp, tenant]))
-              |> Keyword.put(:name, Module.concat([__MODULE__, tenant]))
-
-            {:ok, config}
-          end
-        end
-
-    Usage:
-
-        {:ok, _pid} = ExampleHandler.start_link(tenant: :tenant1)
-
-  ## `c:error/3` callback
-
-  You can define an `c:error/3` callback function to handle any errors returned
-  from your event handler's `handle/2` functions. The `c:error/3` function is
-  passed the actual error (e.g. `{:error, :failure}`), the failed event, and a
-  failure context.
-
-  Use pattern matching on the error and/or failed event to explicitly handle
-  certain errors or events. You can choose to retry, skip, or stop the event
-  handler after an error.
-
-  The default behaviour if you don't provide an `c:error/3` callback is to stop
-  the event handler using the exact error reason returned from the `handle/2`
-  function. You should supervise event handlers to ensure they are correctly
-  restarted on error.
-
-  ### Example error handling
-
-      defmodule ExampleHandler do
-        use Commanded.Event.Handler,
-          application: ExampleApp,
-          name: __MODULE__
-
-        require Logger
-
-        alias Commanded.Event.FailureContext
-
-        def handle(%AnEvent{}, _metadata) do
-          # simulate event handling failure
-          {:error, :failed}
-        end
-
-        def error({:error, :failed}, %AnEvent{} = event, %FailureContext{context: context}) do
-          context = record_failure(context)
-
-          case Map.get(context, :failures) do
-            too_many when too_many >= 3 ->
-              # skip bad event after third failure
-              Logger.warn(fn -> "Skipping bad event, too many failures: " <> inspect(event) end)
-
-              :skip
-
-            _ ->
-              # retry event, failure count is included in context map
-              {:retry, context}
-          end
-        end
-
-        defp record_failure(context) do
-          Map.update(context, :failures, 1, fn failures -> failures + 1 end)
-        end
-      end
-
   ## Consistency
 
   For each event handler you can define its consistency, as one of either
@@ -235,6 +129,11 @@ defmodule Commanded.Event.Handler do
   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:
@@ -297,7 +196,28 @@ defmodule Commanded.Event.Handler do
 
   Can be used to start any related processes when the event handler is started.
 
-  Return `:ok` on success, or `{:stop, reason}` to stop the handler process.
+  This callback function must return `:ok`, or `{:stop, reason}` to stop the
+  handler process. Any other return value will terminate the event handler with
+  an error.
+
+  ### Example
+
+      defmodule ExampleHandler do
+        use Commanded.Event.Handler,
+          application: ExampleApp,
+          name: "ExampleHandler"
+
+        # Optional initialisation
+        def init do
+          :ok
+        end
+
+        def handle(%AnEvent{..}, _metadata) do
+          # ... process the event
+          :ok
+        end
+      end
+
   """
   @callback init() :: :ok | {:stop, reason :: any()}
 
@@ -305,7 +225,36 @@ defmodule Commanded.Event.Handler do
   Optional callback function called to configure the handler before it starts.
 
   It is passed the merged compile-time and runtime config, and must return the
-  updated config.
+  updated config as `{:ok, config}`.
+
+  Note this function is called before the event handler process is started and
+  *is not* run from the handler's process. You cannot use `self()` to access the
+  handler's PID.
+
+  ### Example
+
+  The `c:init/1` function is used to define the handler's application and name
+  based upon a value provided at runtime:
+
+      defmodule ExampleHandler do
+        use Commanded.Event.Handler
+
+        def init(config) do
+          {tenant, config} = Keyword.pop!(config, :tenant)
+
+          config =
+            config
+            |> Keyword.put(:application, Module.concat([ExampleApp, tenant]))
+            |> Keyword.put(:name, Module.concat([__MODULE__, tenant]))
+
+          {:ok, config}
+        end
+      end
+
+  Usage:
+
+      {:ok, _pid} = ExampleHandler.start_link(tenant: :tenant1)
+
   """
   @callback init(config :: Keyword.t()) :: {:ok, Keyword.t()}
 
@@ -326,8 +275,11 @@ defmodule Commanded.Event.Handler do
   The `c:error/3` function allows you to control how event handling failures
   are handled. The function is passed the error returned by the event handler
   (e.g. `{:error, :failure}`), the event causing the error, and a context map
-  containing state passed between retries. Use the context map to track any
-  transient state you need to access between retried failures.
+  containing state passed between retries.
+
+  Use pattern matching on the error and/or failed event to explicitly handle
+  certain errors or events. Use the context map to track any transient state you
+  need to access between retried failures.
 
   You can return one of the following responses depending upon the
   error severity:
@@ -344,6 +296,51 @@ defmodule Commanded.Event.Handler do
 
   - `{:stop, reason}` - stop the event handler with the given reason.
 
+  The default behaviour if you don't provide an `c:error/3` callback is to stop
+  the event handler using the exact error reason returned from the `handle/2`
+  function. If the event handler is supervised using restart `permanent` or
+  `transient` stopping on error will cause the handler to be restarted. It will
+  likely crash again as it will reprocesse the problematic event. This can lead
+  to cascading failures going up the supervision tree.
+
+  ### Example error handling
+
+      defmodule ExampleHandler do
+        use Commanded.Event.Handler,
+          application: ExampleApp,
+          name: __MODULE__
+
+        require Logger
+
+        alias Commanded.Event.FailureContext
+
+        def handle(%AnEvent{}, _metadata) do
+          # simulate event handling failure
+          {:error, :failed}
+        end
+
+        def error({:error, :failed}, %AnEvent{} = event, %FailureContext{context: context}) do
+          context = record_failure(context)
+
+          case Map.get(context, :failures) do
+            too_many when too_many >= 3 ->
+              # skip bad event after third failure
+              Logger.warn(fn -> "Skipping bad event, too many failures: " <> inspect(event) end)
+
+              :skip
+
+            _ ->
+              # retry event, failure count is included in context map
+              {:retry, context}
+          end
+        end
+
+        defp record_failure(context) do
+          Map.update(context, :failures, 1, fn failures -> failures + 1 end)
+        end
+      end
+
+
   """
   @callback error(
               error :: term(),
@@ -377,7 +374,7 @@ defmodule Commanded.Event.Handler do
       Provides a child specification to allow the event handler to be easily
       supervised.
 
-      ## Example
+      ### Example
 
           Supervisor.start_link([
             {ExampleHandler, []}
@@ -496,6 +493,7 @@ defmodule Commanded.Event.Handler do
     end
   end
 
+  @doc false
   def name(application, handler_name), do: {application, __MODULE__, handler_name}
 
   @doc false
@@ -510,14 +508,6 @@ defmodule Commanded.Event.Handler do
     {:noreply, subscribe_to_events(state)}
   end
 
-  @doc false
-  @impl GenServer
-  def handle_call(:last_seen_event, _from, %Handler{} = state) do
-    %Handler{last_seen_event: last_seen_event} = state
-
-    {:reply, last_seen_event, state}
-  end
-
   @doc false
   @impl GenServer
   def handle_info(:reset, %Handler{} = state) do
@@ -575,12 +565,14 @@ defmodule Commanded.Event.Handler do
   @doc false
   @impl GenServer
   def handle_info({:events, events}, %Handler{} = state) do
+    %Handler{application: application} = state
+
     Logger.debug(fn -> describe(state) <> " received events: #{inspect(events)}" end)
 
     try do
       state =
         events
-        |> Upcast.upcast_event_stream()
+        |> Upcast.upcast_event_stream(additional_metadata: %{application: application})
         |> Enum.reduce(state, &handle_event/2)
 
       {:noreply, state}
@@ -669,14 +661,14 @@ defmodule Commanded.Event.Handler do
       {:error, _reason} = error ->
         log_event_error(error, event, state)
 
-        failure_context = build_failure_context(event, context)
+        failure_context = build_failure_context(event, context, state)
 
         handle_event_error(error, event, failure_context, state)
 
       {:error, reason, stacktrace} ->
         log_event_error({:error, reason}, event, state)
 
-        failure_context = build_failure_context(event, context, stacktrace)
+        failure_context = build_failure_context(event, context, stacktrace, state)
 
         handle_event_error({:error, reason}, event, failure_context, state)
 
@@ -690,7 +682,7 @@ defmodule Commanded.Event.Handler do
             ", expected `:ok` or `{:error, term}`"
         end)
 
-        failure_context = build_failure_context(event, context)
+        failure_context = build_failure_context(event, context, state)
 
         handle_event_error(
           {:error, :invalid_return_value},
@@ -705,7 +697,7 @@ defmodule Commanded.Event.Handler do
     %RecordedEvent{data: data} = event
     %Handler{handler_module: handler_module} = state
 
-    metadata = RecordedEvent.enrich_metadata(event)
+    metadata = enrich_metadata(event, state)
 
     try do
       handler_module.handle(data, metadata)
@@ -718,10 +710,36 @@ defmodule Commanded.Event.Handler do
     end
   end
 
-  defp build_failure_context(%RecordedEvent{} = failed_event, context, stacktrace \\ nil) do
-    metadata = RecordedEvent.enrich_metadata(failed_event)
+  defp build_failure_context(
+         %RecordedEvent{} = failed_event,
+         context,
+         stacktrace \\ nil,
+         %Handler{} = state
+       ) do
+    %Handler{application: application, handler_name: handler_name} = state
+
+    metadata = enrich_metadata(failed_event, state)
+
+    %FailureContext{
+      application: application,
+      handler_name: handler_name,
+      context: context,
+      metadata: metadata,
+      stacktrace: stacktrace
+    }
+  end
 
-    %FailureContext{context: context, metadata: metadata, stacktrace: stacktrace}
+  # Enrich the metadata with additional fields from the recorded event, plus the
+  # associated Commanded application and the event handler's name.
+  defp enrich_metadata(%RecordedEvent{} = event, %Handler{} = state) do
+    %Handler{application: application, handler_name: handler_name} = state
+
+    RecordedEvent.enrich_metadata(event,
+      additional_metadata: %{
+        application: application,
+        handler_name: handler_name
+      }
+    )
   end
 
   defp handle_event_error(
diff --git a/lib/commanded/event/upcast.ex b/lib/commanded/event/upcast.ex
index 5920a1c..434f9db 100644
--- a/lib/commanded/event/upcast.ex
+++ b/lib/commanded/event/upcast.ex
@@ -4,18 +4,18 @@ defmodule Commanded.Event.Upcast do
   alias Commanded.Event.Upcaster
   alias Commanded.EventStore.RecordedEvent
 
-  def upcast_event_stream(%Stream{} = event_stream) do
-    Stream.map(event_stream, &upcast_event/1)
-  end
+  def upcast_event_stream(event_stream, opts \\ [])
 
-  def upcast_event_stream(event_stream) do
-    Enum.map(event_stream, &upcast_event/1)
-  end
+  def upcast_event_stream(%Stream{} = event_stream, opts),
+    do: Stream.map(event_stream, &upcast_event(&1, opts))
+
+  def upcast_event_stream(event_stream, opts),
+    do: Enum.map(event_stream, &upcast_event(&1, opts))
 
-  defp upcast_event(%RecordedEvent{} = event) do
+  defp upcast_event(%RecordedEvent{} = event, opts) do
     %RecordedEvent{data: data} = event
 
-    enriched_metadata = RecordedEvent.enrich_metadata(event)
+    enriched_metadata = RecordedEvent.enrich_metadata(event, opts)
 
     %RecordedEvent{event | data: Upcaster.upcast(data, enriched_metadata)}
   end
diff --git a/lib/commanded/event/upcaster.ex b/lib/commanded/event/upcaster.ex
index 4a039a2..a4a57d0 100644
--- a/lib/commanded/event/upcaster.ex
+++ b/lib/commanded/event/upcaster.ex
@@ -26,8 +26,9 @@ defprotocol Commanded.Event.Upcaster do
   associated with that event. The metadata is provided during command dispatch.
 
   In addition to the metadata key/values you provide, the following system
-  values will be included in the metadata passed to an event handler:
+  values will be included in the metadata:
 
+    - `application` - the `Commanded.Application` used to read the event.
     - `event_id` - a globally unique UUID to identify the event.
     - `event_number` - a globally unique, monotonically incrementing integer
       used to order the event amongst all events.
diff --git a/lib/commanded/event_store.ex b/lib/commanded/event_store.ex
index fb41e29..0ed12a9 100644
--- a/lib/commanded/event_store.ex
+++ b/lib/commanded/event_store.ex
@@ -36,8 +36,11 @@ defmodule Commanded.EventStore do
            start_version,
            read_batch_size
          ) do
-      {:error, _error} = error -> error
-      stream -> Upcast.upcast_event_stream(stream)
+      {:error, _error} = error ->
+        error
+
+      stream ->
+        Upcast.upcast_event_stream(stream, additional_metadata: %{application: application})
     end
   end
 
diff --git a/lib/commanded/event_store/recorded_event.ex b/lib/commanded/event_store/recorded_event.ex
index 27d79f7..a169352 100644
--- a/lib/commanded/event_store/recorded_event.ex
+++ b/lib/commanded/event_store/recorded_event.ex
@@ -60,7 +60,12 @@ defmodule Commanded.EventStore.RecordedEvent do
     :created_at
   ]
 
-  def enrich_metadata(%RecordedEvent{} = event) do
+  @doc """
+  Enrich the event's metadata with fields from the `RecordedEvent` struct and
+  any additional metadata passed as an option.
+  """
+  @spec enrich_metadata(t(), [{:additional_metadata, map()}]) :: map()
+  def enrich_metadata(%RecordedEvent{} = event, opts) do
     %RecordedEvent{
       event_id: event_id,
       event_number: event_number,
@@ -72,17 +77,16 @@ defmodule Commanded.EventStore.RecordedEvent do
       metadata: metadata
     } = event
 
-    Map.merge(
-      %{
-        event_id: event_id,
-        event_number: event_number,
-        stream_id: stream_id,
-        stream_version: stream_version,
-        correlation_id: correlation_id,
-        causation_id: causation_id,
-        created_at: created_at
-      },
-      metadata || %{}
-    )
+    %{
+      event_id: event_id,
+      event_number: event_number,
+      stream_id: stream_id,
+      stream_version: stream_version,
+      correlation_id: correlation_id,
+      causation_id: causation_id,
+      created_at: created_at
+    }
+    |> Map.merge(Keyword.get(opts, :additional_metadata, %{}))
+    |> Map.merge(metadata || %{})
   end
 end
diff --git a/lib/commanded/process_managers/process_router.ex b/lib/commanded/process_managers/process_router.ex
index 0bad09e..c1bbb1c 100644
--- a/lib/commanded/process_managers/process_router.ex
+++ b/lib/commanded/process_managers/process_router.ex
@@ -176,7 +176,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
 
   @impl GenServer
   def handle_info({:events, events}, %State{} = state) do
-    %State{pending_events: pending_events} = state
+    %State{application: application, pending_events: pending_events} = state
 
     Logger.debug(fn -> describe(state) <> " received #{length(events)} event(s)" end)
 
@@ -184,7 +184,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     unseen_events =
       events
       |> Enum.reject(&event_already_seen?(&1, state))
-      |> Upcast.upcast_event_stream()
+      |> Upcast.upcast_event_stream(additional_metadata: %{application: application})
 
     state =
       case {pending_events, unseen_events} do


Test patch:
diff --git a/test/event/event_handler_error_handling_test.exs b/test/event/event_handler_error_handling_test.exs
index d426b79..569b11e 100644
--- a/test/event/event_handler_error_handling_test.exs
+++ b/test/event/event_handler_error_handling_test.exs
@@ -4,7 +4,6 @@ defmodule Commanded.Event.EventHandlerErrorHandlingTest do
   import ExUnit.CaptureLog
 
   alias Commanded.DefaultApp
-  alias Commanded.Event.FailureContext
 
   alias Commanded.Event.ErrorAggregate.Events.{
     ErrorEvent,
@@ -13,6 +12,8 @@ defmodule Commanded.Event.EventHandlerErrorHandlingTest do
   }
 
   alias Commanded.Event.ErrorEventHandler
+  alias Commanded.Event.FailureContext
+  alias Commanded.Event.Handler
   alias Commanded.Helpers.EventFactory
 
   setup do
@@ -131,7 +132,8 @@ defmodule Commanded.Event.EventHandlerErrorHandlingTest do
     assert Process.alive?(handler)
 
     # Should ack errored event
-    assert GenServer.call(handler, :last_seen_event) == 1
+    %Handler{last_seen_event: last_seen_event} = :sys.get_state(handler)
+    assert last_seen_event == 1
   end
 
   defp send_error_event(handler, opts \\ []) do
diff --git a/test/event/event_handler_name_test.exs b/test/event/event_handler_name_test.exs
index e68eac6..33bb35f 100644
--- a/test/event/event_handler_name_test.exs
+++ b/test/event/event_handler_name_test.exs
@@ -49,11 +49,13 @@ defmodule Commanded.Event.EventHandlerNameTest do
     end
   end
 
+  defmodule ModuleNamedEventHandler do
+    use Commanded.Event.Handler, application: Commanded.DefaultApp, name: __MODULE__
+  end
+
   test "should allow using event handler module as name" do
-    Code.eval_string("""
-      defmodule EventHandler do
-        use Commanded.Event.Handler, application: Commanded.DefaultApp, name: __MODULE__
-      end
-    """)
+    start_supervised!(Commanded.DefaultApp)
+
+    assert {:ok, _pid} = ModuleNamedEventHandler.start_link()
   end
 end
diff --git a/test/event/event_handler_subscription_test.exs b/test/event/event_handler_subscription_test.exs
index 4629d50..3336efd 100644
--- a/test/event/event_handler_subscription_test.exs
+++ b/test/event/event_handler_subscription_test.exs
@@ -33,14 +33,11 @@ defmodule Commanded.Event.EventHandlerSubscriptionTest do
       reply_to = self()
 
       # First and second subscription attempts fail
-      expect(MockEventStore, :subscribe_to, 2, fn _event_store_meta,
-                                                  :all,
-                                                  "ExampleHandler",
-                                                  handler,
-                                                  :origin ->
-        send(reply_to, {:subscribe_to, handler})
-
-        {:error, :subscription_already_exists}
+      expect(MockEventStore, :subscribe_to, 2, fn
+        _event_store_meta, :all, "ExampleHandler", handler, :origin ->
+          send(reply_to, {:subscribe_to, handler})
+
+          {:error, :subscription_already_exists}
       end)
 
       # Third subscription attempt succeeds
@@ -89,15 +86,12 @@ defmodule Commanded.Event.EventHandlerSubscriptionTest do
   defp expect_subscribe_to(subscription) do
     reply_to = self()
 
-    expect(MockEventStore, :subscribe_to, fn _event_store_meta,
-                                             :all,
-                                             "ExampleHandler",
-                                             handler,
-                                             :origin ->
-      send(handler, {:subscribed, subscription})
-      send(reply_to, {:subscribed, subscription})
+    expect(MockEventStore, :subscribe_to, fn
+      _event_store_meta, :all, "ExampleHandler", handler, :origin ->
+        send(handler, {:subscribed, subscription})
+        send(reply_to, {:subscribed, subscription})
 
-      {:ok, subscription}
+        {:ok, subscription}
     end)
   end
 
diff --git a/test/event/handle_event_test.exs b/test/event/handle_event_test.exs
index 6078bc4..4f81bf6 100644
--- a/test/event/handle_event_test.exs
+++ b/test/event/handle_event_test.exs
@@ -5,52 +5,68 @@ defmodule Commanded.Event.HandleEventTest do
   import Commanded.Assertions.EventAssertions
 
   alias Commanded.DefaultApp
-  alias Commanded.EventStore
   alias Commanded.Event.AppendingEventHandler
+  alias Commanded.Event.EchoHandler
+  alias Commanded.Event.Handler
+  alias Commanded.Event.ReplyEvent
   alias Commanded.Event.UninterestingEvent
-  alias Commanded.Helpers.EventFactory
-  alias Commanded.Helpers.Wait
-  alias Commanded.ExampleDomain.BankApp
-  alias Commanded.ExampleDomain.BankAccount.AccountBalanceHandler
-  alias Commanded.ExampleDomain.BankAccount.BankAccountHandler
+  alias Commanded.EventStore
+  alias Commanded.EventStore.RecordedEvent
   alias Commanded.ExampleDomain.BankAccount.Events.BankAccountOpened
   alias Commanded.ExampleDomain.BankAccount.Events.MoneyDeposited
+  alias Commanded.Helpers.EventFactory
+  alias Commanded.Helpers.Wait
 
   describe "event handling" do
     setup do
-      start_supervised!(BankApp)
-
-      handler = start_supervised!(AccountBalanceHandler)
-
-      Wait.until(fn ->
-        assert AccountBalanceHandler.subscribed?()
-      end)
+      start_supervised!(DefaultApp)
+      handler = start_supervised!(EchoHandler)
 
       [handler: handler]
     end
 
     test "should handle received events", %{handler: handler} do
-      events = [
-        %BankAccountOpened{account_number: "ACC123", initial_balance: 1_000},
-        %MoneyDeposited{amount: 50, balance: 1_050}
-      ]
+      event1 = %ReplyEvent{reply_to: self(), value: 1}
+      event2 = %ReplyEvent{reply_to: self(), value: 2}
+      events = [event1, event2]
 
-      recorded_events = EventFactory.map_to_recorded_events(events)
+      metadata = %{"key" => "value"}
+
+      recorded_events = EventFactory.map_to_recorded_events(events, 1, metadata: metadata)
 
       send(handler, {:events, recorded_events})
 
-      Wait.until(fn ->
-        assert AccountBalanceHandler.current_balance() == 1_050
-      end)
+      assert_receive {:event, ^handler, ^event1, metadata1}
+
+      assert_enriched_metadata(metadata1, Enum.at(recorded_events, 0),
+        additional_metadata: %{
+          application: DefaultApp,
+          handler_name: "Commanded.Event.EchoHandler"
+        }
+      )
+
+      assert_receive {:event, ^handler, ^event2, metadata2}
+
+      assert_enriched_metadata(metadata2, Enum.at(recorded_events, 1),
+        additional_metadata: %{
+          application: DefaultApp,
+          handler_name: "Commanded.Event.EchoHandler"
+        }
+      )
+
+      refute_receive {:event, _handler, _event, _metadata}
     end
 
     test "should ignore uninterested events", %{handler: handler} do
+      interested_event1 = %ReplyEvent{reply_to: self(), value: 1}
+      interested_event2 = %ReplyEvent{reply_to: self(), value: 2}
+
       # Include uninterested events within those the handler is interested in
       events = [
         %UninterestingEvent{},
-        %BankAccountOpened{account_number: "ACC123", initial_balance: 1_000},
+        interested_event1,
         %UninterestingEvent{},
-        %MoneyDeposited{amount: 50, balance: 1_050},
+        interested_event2,
         %UninterestingEvent{}
       ]
 
@@ -58,9 +74,11 @@ defmodule Commanded.Event.HandleEventTest do
 
       send(handler, {:events, recorded_events})
 
-      Wait.until(fn ->
-        assert AccountBalanceHandler.current_balance() == 1_050
-      end)
+      # Receive only interested events
+      assert_receive {:event, ^handler, ^interested_event1, metadata1}
+      assert_receive {:event, ^handler, ^interested_event2, metadata2}
+
+      refute_receive {:event, _handler, _event, _metadata}
     end
 
     test "should ignore unexpected messages", %{handler: handler} do
@@ -75,65 +93,55 @@ defmodule Commanded.Event.HandleEventTest do
       end
 
       assert capture_log(send_unexpected_mesage) =~
-               "Commanded.ExampleDomain.BankAccount.AccountBalanceHandler received unexpected message: :unexpected_message"
+               "Commanded.Event.EchoHandler received unexpected message: :unexpected_message"
     end
   end
 
-  describe "reset event handler" do
+  describe "dynamic event handler" do
     setup do
-      start_supervised!(BankApp)
-
-      :ok
-    end
-
-    test "should be reset when starting from `:origin`" do
-      stream_uuid = UUID.uuid4()
-      initial_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}]
-
-      :ok = EventStore.append_to_stream(BankApp, stream_uuid, 0, to_event_data(initial_events))
-
-      handler = start_supervised!(BankAccountHandler)
-
-      Wait.until(fn ->
-        assert BankAccountHandler.current_accounts() == ["ACC123"]
-      end)
-
-      :ok = BankAccountHandler.change_prefix("PREF_")
+      start_supervised!({DefaultApp, name: :app1})
+      start_supervised!({DefaultApp, name: :app2})
 
-      send(handler, :reset)
+      handler1 = start_supervised!({EchoHandler, application: :app1, name: "handler1"})
+      handler2 = start_supervised!({EchoHandler, application: :app2, name: "handler2"})
 
-      Wait.until(fn ->
-        assert BankAccountHandler.current_accounts() == ["PREF_ACC123"]
-      end)
+      [handler1: handler1, handler2: handler2]
     end
 
-    test "should be reset when starting from `:current`" do
-      stream_uuid = UUID.uuid4()
+    test "should handle received events", %{handler1: handler1, handler2: handler2} do
+      event1 = %ReplyEvent{reply_to: self(), value: 1}
+      event2 = %ReplyEvent{reply_to: self(), value: 2}
 
-      # Ignored initial events
-      initial_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}]
-      :ok = EventStore.append_to_stream(BankApp, stream_uuid, 0, to_event_data(initial_events))
+      metadata = %{"key" => "value"}
 
-      handler = start_supervised!({BankAccountHandler, start_from: :current})
+      recorded_events1 = EventFactory.map_to_recorded_events([event1], 1, metadata: metadata)
+      recorded_events2 = EventFactory.map_to_recorded_events([event2], 1, metadata: metadata)
 
-      Wait.until(fn ->
-        assert BankAccountHandler.current_accounts() == []
-      end)
+      send(handler1, {:events, recorded_events1})
 
-      :ok = BankAccountHandler.change_prefix("PREF_")
+      assert_receive {:event, ^handler1, ^event1, metadata1}
+      refute_receive {:event, ^handler2, _event, _metadata}
 
-      send(handler, :reset)
+      assert_enriched_metadata(metadata1, Enum.at(recorded_events1, 0),
+        additional_metadata: %{
+          application: :app1,
+          handler_name: "handler1"
+        }
+      )
 
-      new_event = [%BankAccountOpened{account_number: "ACC1234", initial_balance: 1_000}]
-      :ok = EventStore.append_to_stream(BankApp, stream_uuid, 1, to_event_data(new_event))
+      send(handler2, {:events, recorded_events2})
 
-      wait_for_event(BankApp, BankAccountOpened, fn event, recorded_event ->
-        event.account_number == "ACC1234" and recorded_event.event_number == 2
-      end)
+      assert_receive {:event, ^handler2, ^event2, metadata2}
+      refute_receive {:event, ^handler1, _event, _metadata}
 
-      Wait.until(fn ->
-        assert BankAccountHandler.current_accounts() == ["PREF_ACC1234"]
-      end)
+      assert_enriched_metadata(metadata2, Enum.at(recorded_events2, 0),
+        additional_metadata: %{
+          application: :app2,
+          handler_name: "handler2"
+        }
+      )
+
+      refute_receive {:event, _handler, _event, _metadata}
     end
   end
 
@@ -155,7 +163,8 @@ defmodule Commanded.Event.HandleEventTest do
 
       handler = start_supervised!({AppendingEventHandler, start_from: :current})
 
-      assert GenServer.call(handler, :last_seen_event) == nil
+      %Handler{last_seen_event: last_seen_event} = :sys.get_state(handler)
+      assert is_nil(last_seen_event)
 
       :ok = EventStore.append_to_stream(DefaultApp, stream_uuid, 1, to_event_data(new_events))
 
@@ -173,7 +182,8 @@ defmodule Commanded.Event.HandleEventTest do
         assert Map.get(metadata, :stream_version) == 2
         assert %DateTime{} = Map.get(metadata, :created_at)
 
-        assert GenServer.call(handler, :last_seen_event) == 2
+        %Handler{last_seen_event: last_seen_event} = :sys.get_state(handler)
+        assert last_seen_event == 2
       end)
     end
 
@@ -232,6 +242,12 @@ defmodule Commanded.Event.HandleEventTest do
     end
   end
 
+  defp assert_enriched_metadata(metadata, %RecordedEvent{} = event, opts) do
+    enriched_metadata = RecordedEvent.enrich_metadata(event, opts)
+
+    assert metadata == enriched_metadata
+  end
+
   defp to_event_data(events) do
     Commanded.Event.Mapper.map_to_event_data(events,
       causation_id: UUID.uuid4(),
diff --git a/test/event/handler_init_test.exs b/test/event/handler_init_test.exs
index 8039035..5628add 100644
--- a/test/event/handler_init_test.exs
+++ b/test/event/handler_init_test.exs
@@ -4,7 +4,7 @@ defmodule Commanded.Event.HandlerInitTest do
   import Mox
 
   alias Commanded.DefaultApp
-  alias Commanded.Event.{InitHandler, RuntimeConfigHandler}
+  alias Commanded.Event.{Handler, InitHandler, RuntimeConfigHandler}
   alias Commanded.EventStore.Adapters.Mock, as: MockEventStore
 
   describe "event handler `init/1` callback" do
@@ -17,13 +17,21 @@ defmodule Commanded.Event.HandlerInitTest do
     end
 
     test "should be called at runtime" do
-      {:ok, _handler1} = RuntimeConfigHandler.start_link(tenant: :tenant1, reply_to: self())
-      {:ok, _handler2} = RuntimeConfigHandler.start_link(tenant: :tenant2, reply_to: self())
-      {:ok, _handler3} = RuntimeConfigHandler.start_link(tenant: :tenant3, reply_to: self())
+      {:ok, handler1} = RuntimeConfigHandler.start_link(tenant: :tenant1, reply_to: self())
+      {:ok, handler2} = RuntimeConfigHandler.start_link(tenant: :tenant2, reply_to: self())
+      {:ok, handler3} = RuntimeConfigHandler.start_link(tenant: :tenant3, reply_to: self())
 
       assert_receive {:init, :tenant1}
       assert_receive {:init, :tenant2}
       assert_receive {:init, :tenant3}
+
+      assert_handler_application(handler1, Module.concat([DefaultApp, :tenant1]))
+      assert_handler_application(handler2, Module.concat([DefaultApp, :tenant2]))
+      assert_handler_application(handler3, Module.concat([DefaultApp, :tenant3]))
+
+      assert_handler_name(handler1, "Commanded.Event.RuntimeConfigHandler.tenant1")
+      assert_handler_name(handler2, "Commanded.Event.RuntimeConfigHandler.tenant2")
+      assert_handler_name(handler3, "Commanded.Event.RuntimeConfigHandler.tenant3")
     end
   end
 
@@ -53,9 +61,21 @@ defmodule Commanded.Event.HandlerInitTest do
 
       assert_receive {:init, ^handler}
     end
+  end
 
-    defp send_subscribed(handler) do
-      send(handler, {:subscribed, handler})
-    end
+  defp assert_handler_application(handler, expected_application) do
+    %Handler{application: application} = :sys.get_state(handler)
+
+    assert application == expected_application
+  end
+
+  defp assert_handler_name(handler, expected_handler_name) do
+    %Handler{handler_name: handler_name} = :sys.get_state(handler)
+
+    assert handler_name == expected_handler_name
+  end
+
+  defp send_subscribed(handler) do
+    send(handler, {:subscribed, handler})
   end
 end
diff --git a/test/event/reset_event_handler_test.exs b/test/event/reset_event_handler_test.exs
new file mode 100644
index 0000000..6d25a1d
--- /dev/null
+++ b/test/event/reset_event_handler_test.exs
@@ -0,0 +1,76 @@
+defmodule Commanded.Event.ResetEventHandlerTest do
+  use ExUnit.Case
+
+  import Commanded.Assertions.EventAssertions
+
+  alias Commanded.EventStore
+  alias Commanded.Helpers.Wait
+  alias Commanded.ExampleDomain.BankApp
+  alias Commanded.ExampleDomain.BankAccount.BankAccountHandler
+  alias Commanded.ExampleDomain.BankAccount.Events.BankAccountOpened
+
+  describe "reset event handler" do
+    setup do
+      start_supervised!(BankApp)
+      :ok
+    end
+
+    test "should be reset when starting from `:origin`" do
+      stream_uuid = UUID.uuid4()
+      initial_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}]
+
+      :ok = EventStore.append_to_stream(BankApp, stream_uuid, 0, to_event_data(initial_events))
+
+      handler = start_supervised!(BankAccountHandler)
+
+      Wait.until(fn ->
+        assert BankAccountHandler.current_accounts() == ["ACC123"]
+      end)
+
+      :ok = BankAccountHandler.change_prefix("PREF_")
+
+      send(handler, :reset)
+
+      Wait.until(fn ->
+        assert BankAccountHandler.current_accounts() == ["PREF_ACC123"]
+      end)
+    end
+
+    test "should be reset when starting from `:current`" do
+      stream_uuid = UUID.uuid4()
+
+      # Ignored initial events
+      initial_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}]
+      :ok = EventStore.append_to_stream(BankApp, stream_uuid, 0, to_event_data(initial_events))
+
+      handler = start_supervised!({BankAccountHandler, start_from: :current})
+
+      Wait.until(fn ->
+        assert BankAccountHandler.current_accounts() == []
+      end)
+
+      :ok = BankAccountHandler.change_prefix("PREF_")
+
+      send(handler, :reset)
+
+      new_event = [%BankAccountOpened{account_number: "ACC1234", initial_balance: 1_000}]
+      :ok = EventStore.append_to_stream(BankApp, stream_uuid, 1, to_event_data(new_event))
+
+      wait_for_event(BankApp, BankAccountOpened, fn event, recorded_event ->
+        event.account_number == "ACC1234" and recorded_event.event_number == 2
+      end)
+
+      Wait.until(fn ->
+        assert BankAccountHandler.current_accounts() == ["PREF_ACC1234"]
+      end)
+    end
+  end
+
+  defp to_event_data(events) do
+    Commanded.Event.Mapper.map_to_event_data(events,
+      causation_id: UUID.uuid4(),
+      correlation_id: UUID.uuid4(),
+      metadata: %{}
+    )
+  end
+end
diff --git a/test/event/support/echo_handler.ex b/test/event/support/echo_handler.ex
new file mode 100644
index 0000000..6b44315
--- /dev/null
+++ b/test/event/support/echo_handler.ex
@@ -0,0 +1,16 @@
+defmodule Commanded.Event.EchoHandler do
+  use Commanded.Event.Handler,
+    application: Commanded.DefaultApp,
+    name: __MODULE__
+
+  alias Commanded.Event.ReplyEvent
+
+  @impl Commanded.Event.Handler
+  def handle(%ReplyEvent{} = event, metadata) do
+    %ReplyEvent{reply_to: reply_to} = event
+
+    send(reply_to, {:event, self(), event, metadata})
+
+    :ok
+  end
+end
diff --git a/test/event/support/init/runtime_config_handler.ex b/test/event/support/init/runtime_config_handler.ex
index 0bcb966..239c3c4 100644
--- a/test/event/support/init/runtime_config_handler.ex
+++ b/test/event/support/init/runtime_config_handler.ex
@@ -10,7 +10,7 @@ defmodule Commanded.Event.RuntimeConfigHandler do
     config =
       config
       |> Keyword.put(:application, Module.concat([DefaultApp, tenant]))
-      |> Keyword.put(:name, Module.concat([__MODULE__, tenant]))
+      |> Keyword.put(:name, inspect(__MODULE__) <> ".#{tenant}")
 
     send(reply_to, {:init, tenant})
 
diff --git a/test/event/support/reply_event.ex b/test/event/support/reply_event.ex
new file mode 100644
index 0000000..a0e5957
--- /dev/null
+++ b/test/event/support/reply_event.ex
@@ -0,0 +1,3 @@
+defmodule Commanded.Event.ReplyEvent do
+  defstruct [:reply_to, :value]
+end
diff --git a/test/event/support/upcast/event_handler.ex b/test/event/support/upcast/event_handler.ex
index 8ca715e..892ec6b 100644
--- a/test/event/support/upcast/event_handler.ex
+++ b/test/event/support/upcast/event_handler.ex
@@ -3,12 +3,13 @@ defmodule Commanded.Event.Upcast.EventHandler do
     application: Commanded.DefaultApp,
     name: __MODULE__
 
-  alias Commanded.Event.Upcast.Events.{EventOne, EventTwo, EventThree, EventFour}
+  alias Commanded.Event.Upcast.Events.{EventOne, EventTwo, EventThree, EventFour, EventFive}
 
   def handle(%EventOne{} = e, _), do: send_reply(e)
   def handle(%EventTwo{} = e, _), do: send_reply(e)
   def handle(%EventThree{} = e, _), do: send_reply(e)
   def handle(%EventFour{} = e, _), do: send_reply(e)
+  def handle(%EventFive{} = e, _), do: send_reply(e)
 
   defp send_reply(%{reply_to: reply_to} = e) do
     send(:erlang.list_to_pid(reply_to), e)
diff --git a/test/event/upcaster_test.exs b/test/event/upcaster_test.exs
index ee15f21..a87bec1 100644
--- a/test/event/upcaster_test.exs
+++ b/test/event/upcaster_test.exs
@@ -34,21 +34,7 @@ defmodule Event.UpcasterTest do
       assert %EventFive{event_metadata: metadata} =
                write_events(DefaultApp, struct(EventFive, version: 1)) |> read_event()
 
-      metadata_keys_not_included =
-        [
-          :event_id,
-          :event_number,
-          :stream_id,
-          :stream_version,
-          :correlation_id,
-          :causation_id,
-          :created_at
-        ] -- Map.keys(metadata)
-
-      assert metadata_keys_not_included == []
-      assert is_number(metadata.stream_version)
-      assert is_binary(metadata.event_id)
-      assert %DateTime{} = metadata.created_at
+      assert_metadata(metadata)
     end
 
     test "can adapt new event from old event" do
@@ -102,14 +88,18 @@ defmodule Event.UpcasterTest do
         [
           struct(EventOne, version: 1, reply_to: reply_to),
           struct(EventTwo, version: 1, reply_to: reply_to),
-          struct(EventThree, version: 1, reply_to: reply_to)
+          struct(EventThree, version: 1, reply_to: reply_to),
+          struct(EventFive, version: 1, reply_to: reply_to)
         ]
       )
 
       assert_receive %EventOne{version: 1}
       assert_receive %EventTwo{version: 2}
       assert_receive %EventFour{version: 2, name: "Chris"}
+      assert_receive %EventFive{version: 2, event_metadata: metadata}
       refute_receive %EventThree{}
+
+      assert_metadata(metadata)
     end
   end
 
@@ -143,6 +133,24 @@ defmodule Event.UpcasterTest do
     end
   end
 
+  defp assert_metadata(metadata) do
+    assert %{
+             application: Commanded.DefaultApp,
+             causation_id: nil,
+             correlation_id: nil,
+             created_at: %DateTime{},
+             event_id: event_id,
+             event_number: event_number,
+             stream_id: stream_id,
+             stream_version: stream_version
+           } = metadata
+
+    assert is_binary(event_id)
+    assert is_number(event_number)
+    assert is_binary(stream_id)
+    assert is_number(stream_version)
+  end
+
   defp write_events(application, events) do
     stream_uuid = UUID.uuid4()
     events = events |> List.wrap() |> Enum.map(&create_event/1)
diff --git a/test/helpers/event_factory.ex b/test/helpers/event_factory.ex
index 86d2e1c..b6912c1 100644
--- a/test/helpers/event_factory.ex
+++ b/test/helpers/event_factory.ex
@@ -2,11 +2,13 @@ defmodule Commanded.Helpers.EventFactory do
   @moduledoc false
   alias Commanded.EventStore.RecordedEvent
 
-  def map_to_recorded_events(events, initial_event_number \\ 1) do
+  def map_to_recorded_events(events, initial_event_number \\ 1, opts \\ []) do
     stream_id = UUID.uuid4()
-    causation_id = UUID.uuid4()
-    correlation_id = UUID.uuid4()
-    fields = [causation_id: causation_id, correlation_id: correlation_id, metadata: %{}]
+    causation_id = Keyword.get(opts, :causation_id, UUID.uuid4())
+    correlation_id = Keyword.get(opts, :correlation_id, UUID.uuid4())
+    metadata = Keyword.get(opts, :metadata, %{})
+
+    fields = [causation_id: causation_id, correlation_id: correlation_id, metadata: metadata]
 
     events
     |> Commanded.Event.Mapper.map_to_event_data(fields)