Repository: slashdotdash/commanded (elixir)

Issue:
Include custom metadata during command dispatch
The supported event stores provide storing metadata per event appended to a stream. It would be useful to allow this metadata to be provided during commanded dispatch. 

An example usage would be to include the identity of the user invoking the command:

```elixir
command = %OpenAccount{account_number: "ACC123", initial_balance: 1_000}
metadata = %{"user_id" => "user-12345"}

:ok = BankRouter.dispatch(command, metadata: metdata)
```

The *optional* metadata would be recorded against any persisted events.

Pull request:
Include custom metadata during command dispatch
Allow metadata to be provided when dispatching a command:

```elixir
command = %OpenAccount{account_number: "ACC123", initial_balance: 1_000}
metadata = %{"user_id" => "user-12345", "ip_address" => "127.0.0.1"}

:ok = BankRouter.dispatch(command, metadata: metdata)
```

The *optional* metadata is recorded against all events created by the command. This is useful for additional auditing purposes, such as recording the user's identity, their source IP address, etc.

Fixes #61.

Thanks to @sammkj for implementing this feature.

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 82f16be..fbf4c31 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,10 @@
 # Changelog
 
+## v0.14.0 (unreleased)
+
+- Dispatch command with `:eventual` or `:strong` consistency guarantee ([#82](https://github.com/slashdotdash/commanded/issues/82)).
+- Include custom metadata during command dispatch ([#61](https://github.com/slashdotdash/commanded/issues/61)).
+
 ## v0.13.0
 
 ### Enhancements
diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex
index ff1db76..582496c 100644
--- a/lib/commanded/aggregates/aggregate.ex
+++ b/lib/commanded/aggregates/aggregate.ex
@@ -1,19 +1,24 @@
 defmodule Commanded.Aggregates.Aggregate do
   @moduledoc """
-  Aggregate is a `GenServer` process used to provide access to an instance of an event sourced aggregate root.
-  It allows execution of commands against an aggregate instance, and handles persistence of events to the configured event store.
-
-  Concurrent commands sent to an aggregate instance are serialized and executed in the order received.
-
-  The `Commanded.Commands.Router` module will locate, or start, an aggregate instance when a command is dispatched.
-  By default, it will run indefinitely once started. Its lifespan may be controlled by using the `Commanded.Aggregates.AggregateLifespan` behaviour.
+  Aggregate is a `GenServer` process used to provide access to an
+  instance of an event sourced aggregate root. It allows execution of commands
+  against an aggregate instance, and handles persistence of created events to
+  the configured event store.
+
+  Concurrent commands sent to an aggregate instance are serialized and executed
+  in the order received.
+
+  The `Commanded.Commands.Router` module will locate, or start, an aggregate
+  instance when a command is dispatched. By default, an aggregate process will
+  run indefinitely once started. Its lifespan may be controlled by using the
+  `Commanded.Aggregates.AggregateLifespan` behaviour.
   """
 
   use GenServer
 
   require Logger
 
-  alias Commanded.Aggregates.{Aggregate,DefaultLifespan}
+  alias Commanded.Aggregates.{Aggregate,ExecutionContext}
   alias Commanded.Event.Mapper
   alias Commanded.EventStore
 
@@ -41,6 +46,7 @@ defmodule Commanded.Aggregates.Aggregate do
     {:via, @registry_provider, {@aggregate_registry_name, aggregate_uuid}}
   end
 
+  @doc false
   def init(%Aggregate{} = state) do
     # initial aggregate state is populated by loading events from event store
     GenServer.cast(self(), {:populate_aggregate_state})
@@ -49,28 +55,19 @@ defmodule Commanded.Aggregates.Aggregate do
   end
 
   @doc """
-  Execute the given command against the aggregate, optionally providing a timeout and lifespan.
-
-  - `aggregate_uuid` uniquely identifies an instance of the aggregate.
-
-  - `command` is the command to execute, typically a struct (e.g. `%OpenBankAccount{...}`).
-
-  - `handler` is the module that handles the command.
-    It may be the aggregate module, but does not have to be as you can specify and use a command handler module.
+  Execute the given command against the aggregate.
 
-  - `function` is the name of function, as an atom, that handles the command.
-    The default value is `:execute`, used to support command dispatch directly to the aggregate module.
-
-  - `timeout` is an integer greater than zero which specifies how many milliseconds to wait for a reply, or the atom :infinity to wait indefinitely.
+  - `aggregate_uuid` - uniquely identifies an instance of the aggregate.
+  - `context` - includes command execution arguments
+    (see `Commanded.Aggregates.ExecutionContext` for details).
+  - `timeout` - is an integer greater than zero which specifies how many
+    milliseconds to wait for a reply, or the atom :infinity to wait indefinitely.
     The default value is 5000.
 
-  - `lifespan` is a module implementing the `Commanded.Aggregates.AggregateLifespan` behaviour to control the aggregate instance process lifespan.
-    The default value, `Commanded.Aggregates.DefaultLifespan`, keeps the process running indefinitely.
-
   Returns `{:ok, aggregate_version}` on success, or `{:error, reason}` on failure.
   """
-  def execute(aggregate_uuid, command, handler, function \\ :execute, timeout \\ 5_000, lifespan \\ DefaultLifespan) do
-    GenServer.call(via_tuple(aggregate_uuid), {:execute_command, handler, function, command, lifespan}, timeout)
+  def execute(aggregate_uuid, %ExecutionContext{} = context, timeout \\ 5_000) do
+    GenServer.call(via_tuple(aggregate_uuid), {:execute_command, context}, timeout)
   end
 
   @doc false
@@ -90,20 +87,23 @@ defmodule Commanded.Aggregates.Aggregate do
   end
 
   @doc false
-  def handle_call({:execute_command, handler, function, command, lifespan}, _from, %Aggregate{} = state) do
-    {reply, state} = execute_command(handler, function, command, state)
+  def handle_call({:execute_command, %ExecutionContext{command: command, lifespan: lifespan} = context}, _from, %Aggregate{} = state) do
+    {reply, state} = execute_command(context, state)
 
     {:reply, reply, state, lifespan.after_command(command)}
   end
 
+  @doc false
   def handle_call({:aggregate_state}, _from, %Aggregate{aggregate_state: aggregate_state} = state) do
     {:reply, aggregate_state, state}
   end
 
+  @doc false
   def handle_call({:aggregate_version}, _from, %Aggregate{aggregate_version: aggregate_version} = state) do
     {:reply, aggregate_version, state}
   end
 
+  @doc false
   def handle_info(:timeout, %Aggregate{} = state) do
     {:stop, :normal, state}
   end
@@ -150,7 +150,10 @@ defmodule Commanded.Aggregates.Aggregate do
     end
   end
 
-  defp execute_command(handler, function, command, %Aggregate{aggregate_uuid: aggregate_uuid, aggregate_version: expected_version, aggregate_state: aggregate_state, aggregate_module: aggregate_module} = state) do
+  defp execute_command(
+    %ExecutionContext{handler: handler, function: function, command: command, metadata: metadata},
+    %Aggregate{aggregate_uuid: aggregate_uuid, aggregate_version: expected_version, aggregate_state: aggregate_state, aggregate_module: aggregate_module} = state)
+  do
     case Kernel.apply(handler, function, [aggregate_state, command]) do
       {:error, _reason} = reply -> {reply, state}
       events ->
@@ -158,7 +161,7 @@ defmodule Commanded.Aggregates.Aggregate do
 
         updated_state = apply_events(aggregate_module, aggregate_state, pending_events)
 
-        {:ok, stream_version} = persist_events(pending_events, aggregate_uuid, expected_version)
+        {:ok, stream_version} = persist_events(pending_events, aggregate_uuid, expected_version, metadata)
 
         state = %Aggregate{state |
           aggregate_state: updated_state,
@@ -173,10 +176,10 @@ defmodule Commanded.Aggregates.Aggregate do
     Enum.reduce(events, aggregate_state, &aggregate_module.apply(&2, &1))
   end
 
-  defp persist_events([], _aggregate_uuid, expected_version), do: {:ok, expected_version}
-  defp persist_events(pending_events, aggregate_uuid, expected_version) do
+  defp persist_events([], _aggregate_uuid, expected_version, _metadata), do: {:ok, expected_version}
+  defp persist_events(pending_events, aggregate_uuid, expected_version, metadata) do
     correlation_id = UUID.uuid4
-    event_data = Mapper.map_to_event_data(pending_events, correlation_id)
+    event_data = Mapper.map_to_event_data(pending_events, correlation_id, nil, metadata)
 
     EventStore.append_to_stream(aggregate_uuid, expected_version, event_data)
   end
diff --git a/lib/commanded/aggregates/execution_context.ex b/lib/commanded/aggregates/execution_context.ex
new file mode 100644
index 0000000..1301165
--- /dev/null
+++ b/lib/commanded/aggregates/execution_context.ex
@@ -0,0 +1,37 @@
+defmodule Commanded.Aggregates.ExecutionContext do
+  @moduledoc """
+  Defines the arguments used to execute a command for an aggregate.
+
+  The available options are:
+
+    - `command` - the command to execute, typically a struct
+      (e.g. `%OpenBankAccount{...}`).
+
+    - `metadata` - a map of key/value pairs containing the metadata to be
+      associated with all events created by the command.
+
+    - `handler` - the module that handles the command. It may be either the
+      aggregate module itself or a separate command handler module.
+
+    - `function` - the name of function, as an atom, that handles the command.
+      The default value is `:execute`, used to support command dispatch directly
+      to the aggregate module. For command handlers the `:handle` function is
+      used.
+
+    - `lifespan` - a module implementing the `Commanded.Aggregates.AggregateLifespan`
+      behaviour to control the aggregate instance process lifespan. The default
+      value, `Commanded.Aggregates.DefaultLifespan`, keeps the process running
+      indefinitely.
+
+  """
+
+  alias Commanded.Aggregates.DefaultLifespan
+
+  defstruct [
+    command: nil,
+    metadata: %{},
+    handler: nil,
+    function: nil,
+    lifespan: DefaultLifespan,
+  ]
+end
diff --git a/lib/commanded/commands/dispatcher.ex b/lib/commanded/commands/dispatcher.ex
index 36c5a14..ac8b880 100644
--- a/lib/commanded/commands/dispatcher.ex
+++ b/lib/commanded/commands/dispatcher.ex
@@ -4,6 +4,7 @@ defmodule Commanded.Commands.Dispatcher do
   require Logger
 
   alias Commanded.Aggregates
+  alias Commanded.Aggregates.ExecutionContext
   alias Commanded.Middleware.Pipeline
 
   defmodule Payload do
@@ -18,6 +19,7 @@ defmodule Commanded.Commands.Dispatcher do
       identity: nil,
       timeout: nil,
       lifespan: nil,
+      metadata: nil,
       middleware: [],
     ]
   end
@@ -54,12 +56,13 @@ defmodule Commanded.Commands.Dispatcher do
   end
 
   defp execute(
-    %Pipeline{assigns: %{aggregate_uuid: aggregate_uuid}, command: command} = pipeline,
-    %Payload{handler_module: handler_module, handler_function: handler_function, timeout: timeout, lifespan: lifespan} = payload)
+    %Pipeline{assigns: %{aggregate_uuid: aggregate_uuid}} = pipeline,
+    %Payload{timeout: timeout} = payload)
   do
     {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(payload.aggregate_module, aggregate_uuid)
 
-    task = Task.Supervisor.async_nolink(Commanded.Commands.TaskDispatcher, Aggregates.Aggregate, :execute, [aggregate_uuid, command, handler_module, handler_function, timeout, lifespan])
+    context = to_execution_context(pipeline, payload)
+    task = Task.Supervisor.async_nolink(Commanded.Commands.TaskDispatcher, Aggregates.Aggregate, :execute, [aggregate_uuid, context, timeout])
     task_result = Task.yield(task, timeout) || Task.shutdown(task)
 
     result =
@@ -89,6 +92,19 @@ defmodule Commanded.Commands.Dispatcher do
      end
   end
 
+  defp to_execution_context(
+    %Pipeline{command: command},
+    %Payload{handler_module: handler_module, handler_function: handler_function, lifespan: lifespan, metadata: metadata})
+  do
+    %ExecutionContext{
+      command: command,
+      metadata: metadata,
+      handler: handler_module,
+      function: handler_function,
+      lifespan: lifespan,
+    }
+  end
+
   defp respond_with_success(%Pipeline{} = pipeline, %Payload{include_aggregate_version: include_aggregate_version}, aggregate_version) do
     response =
       case include_aggregate_version do
diff --git a/lib/commanded/commands/router.ex b/lib/commanded/commands/router.ex
index 8b1072d..70af52a 100644
--- a/lib/commanded/commands/router.ex
+++ b/lib/commanded/commands/router.ex
@@ -62,6 +62,14 @@ defmodule Commanded.Commands.Router do
   This is useful when you need to wait for an event handler (e.g. a read model
   projection) to be up-to-date before continuing or querying its data.
 
+  ## Metadata
+
+  You can associate metadata with all events created by the command.
+
+  Supply a map containing key/value pairs comprising the metadata:
+
+      :ok = BankRouter.dispatch(command, metadata: %{"ip_address" => "127.0.0.1"})
+
   """
   defmacro __using__(_) do
     quote do
@@ -77,6 +85,7 @@ defmodule Commanded.Commands.Router do
       @default_consistency :eventual
       @default_dispatch_timeout 5_000
       @default_lifespan Commanded.Aggregates.DefaultLifespan
+      @default_metadata %{}
       @include_aggregate_version false
     end
   end
@@ -153,26 +162,31 @@ defmodule Commanded.Commands.Router do
 
       - `timeout_or_opts` is either an integer timeout or a keyword list of options.
         The timeout must be an integer greater than zero which specifies how
-        many milliseconds to allow the command to be handled, or the atom `:infinity` to wait indefinitely.
-        The default timeout value is 5000.
+        many milliseconds to allow the command to be handled, or the atom
+        `:infinity` to wait indefinitely. The default timeout value is 5000.
 
         Alternatively, an options keyword list can be provided, it supports:
 
         Options:
 
           - `:consistency` as one of `:eventual` (default) or `:strong`
-            By setting the consistency to `:strong` a successful command dispatch
-            will block until all strongly consistent event handlers and process managers
-            have handled all events created by the command.
+            By setting the consistency to `:strong` a successful command
+            dispatch will block until all strongly consistent event handlers and
+            process managers have handled all events created by the command.
 
-          - `:timeout` as described above
+          - `:timeout` as described above.
 
-          - `:include_aggregate_version` set to true to include the aggregate stream version in the success response: `{:ok, aggregate_version}`
+          - `:include_aggregate_version` set to true to include the aggregate
+            stream version in the success response: `{:ok, aggregate_version}`
             The default is false, to return just `:ok`
 
-      Returns `:ok` on success, unless `:include_aggregate_version` is enabled where it returns `{:ok, aggregate_version}`.
+          - `:metadata` - An optional map containing key/value pairs comprising
+            the metadata to be associated with all events created by the command.
+
+      Returns `:ok` on success, unless `:include_aggregate_version` is enabled
+      where it returns `{:ok, aggregate_version}`.
       """
-      @spec dispatch(command :: struct, timeout :: integer | :infinity | keyword()) :: :ok | {:error, :consistency_timeout} | {:error, reason :: term}
+      @spec dispatch(command :: struct, timeout_or_opts :: integer | :infinity | keyword()) :: :ok | {:error, :consistency_timeout} | {:error, reason :: term}
       def dispatch(command, timeout_or_opts)
       def dispatch(%unquote(command_module){} = command, :infinity), do: do_dispatch(command, timeout: :infinity)
       def dispatch(%unquote(command_module){} = command, timeout) when is_integer(timeout), do: do_dispatch(command, timeout: timeout)
@@ -180,6 +194,7 @@ defmodule Commanded.Commands.Router do
 
       defp do_dispatch(%unquote(command_module){} = command, opts) do
         consistency = Keyword.get(opts, :consistency, unquote(consistency) || @default_consistency)
+        metadata = Keyword.get(opts, :metadata, @default_metadata)
         timeout = Keyword.get(opts, :timeout, unquote(timeout) || @default_dispatch_timeout)
         include_aggregate_version = Keyword.get(opts, :include_aggregate_version, @include_aggregate_version)
 
@@ -193,6 +208,7 @@ defmodule Commanded.Commands.Router do
           include_aggregate_version: include_aggregate_version,
           timeout: timeout,
           lifespan: unquote(lifespan) || @default_lifespan,
+          metadata: metadata,
           middleware: @registered_middleware ++ @default_middleware,
         })
       end
diff --git a/lib/commanded/event_store/adapters/in_memory.ex b/lib/commanded/event_store/adapters/in_memory.ex
index 575ac6f..25added 100644
--- a/lib/commanded/event_store/adapters/in_memory.ex
+++ b/lib/commanded/event_store/adapters/in_memory.ex
@@ -220,11 +220,15 @@ defmodule Commanded.EventStore.Adapters.InMemory do
     {{:ok, length(stream_events)}, state}
   end
 
-  defp map_to_recorded_event(event_number, stream_uuid, stream_version, now, %EventData{correlation_id: correlation_id, event_type: event_type, data: data, metadata: metadata}) do
+  defp map_to_recorded_event(
+    event_number, stream_uuid, stream_version, now, 
+    %EventData{causation_id: causation_id, correlation_id: correlation_id, event_type: event_type, data: data, metadata: metadata})
+  do
     %RecordedEvent{
       event_number: event_number,
       stream_id: stream_uuid,
       stream_version: stream_version,
+      causation_id: causation_id,
       correlation_id: correlation_id,
       event_type: event_type,
       data: data,
diff --git a/mix.exs b/mix.exs
index 93e7458..59d628b 100644
--- a/mix.exs
+++ b/mix.exs
@@ -1,7 +1,7 @@
 defmodule Commanded.Mixfile do
   use Mix.Project
 
-  @version "0.13.0"
+  @version "0.14.0"
 
   def project do
     [


Test patch:
diff --git a/test/aggregates/event_persistence_test.exs b/test/aggregates/event_persistence_test.exs
index 658f2ee..85c2f6b 100644
--- a/test/aggregates/event_persistence_test.exs
+++ b/test/aggregates/event_persistence_test.exs
@@ -3,7 +3,7 @@ defmodule Commanded.Entities.EventPersistenceTest do
 
   import Commanded.Enumerable, only: [pluck: 2]
 
-  alias Commanded.Aggregates.Aggregate
+  alias Commanded.Aggregates.{Aggregate,ExecutionContext}
   alias Commanded.EventStore
 
   defmodule ExampleAggregate do
@@ -52,11 +52,33 @@ defmodule Commanded.Entities.EventPersistenceTest do
 
     {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
 
-    {:ok, 10} = Aggregate.execute(aggregate_uuid, %AppendItems{count: 10}, AppendItemsHandler, :handle)
+    {:ok, 10} = Aggregate.execute(aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 10}, handler: AppendItemsHandler, function: :handle})
 
     recorded_events = EventStore.stream_forward(aggregate_uuid, 0) |> Enum.to_list()
 
     assert recorded_events |> pluck(:data) |> pluck(:index) == Enum.to_list(1..10)
+    assert pluck(recorded_events, :event_number) == Enum.to_list(1..10)
+
+    Enum.each(recorded_events, fn recorded_event ->
+      assert recorded_event.stream_id == aggregate_uuid
+    end)
+  end
+
+  test "should persist event metadata" do
+    aggregate_uuid = UUID.uuid4
+
+    {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
+
+    metadata = %{"ip_address" => "127.0.0.1"}
+    context = %ExecutionContext{command: %AppendItems{count: 10}, metadata: metadata, handler: AppendItemsHandler, function: :handle}
+
+    {:ok, 10} = Aggregate.execute(aggregate_uuid, context)
+
+    recorded_events = EventStore.stream_forward(aggregate_uuid, 0) |> Enum.to_list()
+
+    Enum.each(recorded_events, fn recorded_event ->
+      assert recorded_event.metadata == metadata
+    end)
   end
 
   test "should reload persisted events when restarting aggregate process" do
@@ -64,7 +86,7 @@ defmodule Commanded.Entities.EventPersistenceTest do
 
     {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
 
-    {:ok, 10} = Aggregate.execute(aggregate_uuid, %AppendItems{count: 10}, AppendItemsHandler, :handle)
+    {:ok, 10} = Aggregate.execute(aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 10}, handler: AppendItemsHandler, function: :handle})
 
     Commanded.Helpers.Process.shutdown(aggregate_uuid)
 
@@ -82,9 +104,9 @@ defmodule Commanded.Entities.EventPersistenceTest do
 
     {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
 
-    {:ok, 100} = Aggregate.execute(aggregate_uuid, %AppendItems{count: 100}, AppendItemsHandler, :handle)
-    {:ok, 200} = Aggregate.execute(aggregate_uuid, %AppendItems{count: 100}, AppendItemsHandler, :handle)
-    {:ok, 201} = Aggregate.execute(aggregate_uuid, %AppendItems{count: 1}, AppendItemsHandler, :handle)
+    {:ok, 100} = Aggregate.execute(aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 100}, handler: AppendItemsHandler, function: :handle})
+    {:ok, 200} = Aggregate.execute(aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 100}, handler: AppendItemsHandler, function: :handle})
+    {:ok, 201} = Aggregate.execute(aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 1}, handler: AppendItemsHandler, function: :handle})
 
     Commanded.Helpers.Process.shutdown(aggregate_uuid)
 
diff --git a/test/aggregates/execute_command_for_aggregate_test.exs b/test/aggregates/execute_command_for_aggregate_test.exs
index 815a943..955748f 100644
--- a/test/aggregates/execute_command_for_aggregate_test.exs
+++ b/test/aggregates/execute_command_for_aggregate_test.exs
@@ -1,7 +1,7 @@
 defmodule Commanded.Entities.ExecuteCommandForAggregateTest do
   use Commanded.StorageCase
 
-  alias Commanded.Aggregates.Aggregate
+  alias Commanded.Aggregates.{Aggregate,ExecutionContext}
   alias Commanded.EventStore
   alias Commanded.ExampleDomain.{BankAccount,OpenAccountHandler,DepositMoneyHandler}
   alias Commanded.ExampleDomain.BankAccount.Commands.{OpenAccount,DepositMoney}
@@ -15,7 +15,10 @@ defmodule Commanded.Entities.ExecuteCommandForAggregateTest do
 
     {:ok, ^account_number} = Commanded.Aggregates.Supervisor.open_aggregate(BankAccount, account_number)
 
-    {:ok, 1} = Aggregate.execute(account_number, %OpenAccount{account_number: account_number, initial_balance: 1_000}, BankAccount, :open_account)
+    command = %OpenAccount{account_number: account_number, initial_balance: 1_000}
+    context = %ExecutionContext{command: command, handler: BankAccount, function: :open_account}
+
+    {:ok, 1} = Aggregate.execute(account_number, context)
 
     Helpers.Process.shutdown(account_number)
 
@@ -31,7 +34,10 @@ defmodule Commanded.Entities.ExecuteCommandForAggregateTest do
 
     {:ok, ^account_number} = Commanded.Aggregates.Supervisor.open_aggregate(BankAccount, account_number)
 
-    {:ok, 1} = Aggregate.execute(account_number, %OpenAccount{account_number: account_number, initial_balance: 1_000}, OpenAccountHandler, :handle)
+    command = %OpenAccount{account_number: account_number, initial_balance: 1_000}
+    context = %ExecutionContext{command: command, handler: OpenAccountHandler, function: :handle}
+
+    {:ok, 1} = Aggregate.execute(account_number, context)
 
     Helpers.Process.shutdown(account_number)
 
@@ -47,12 +53,18 @@ defmodule Commanded.Entities.ExecuteCommandForAggregateTest do
 
     {:ok, ^account_number} = Commanded.Aggregates.Supervisor.open_aggregate(BankAccount, account_number)
 
-    {:ok, 1} = Aggregate.execute(account_number, %OpenAccount{account_number: account_number, initial_balance: 1_000}, OpenAccountHandler, :handle)
+    command = %OpenAccount{account_number: account_number, initial_balance: 1_000}
+    context = %ExecutionContext{command: command, handler: OpenAccountHandler, function: :handle}
+
+    {:ok, 1} = Aggregate.execute(account_number, context)
 
     state_before = Aggregate.aggregate_state(account_number)
 
     assert_process_exit(account_number, fn ->
-      Aggregate.execute(account_number, %OpenAccount{account_number: account_number, initial_balance: 1}, OpenAccountHandler, :handle)
+      command = %OpenAccount{account_number: account_number, initial_balance: 1}
+      context = %ExecutionContext{command: command, handler: OpenAccountHandler, function: :handle}
+
+      Aggregate.execute(account_number, context)
     end)
 
     {:ok, ^account_number} = Commanded.Aggregates.Supervisor.open_aggregate(BankAccount, account_number)
@@ -76,7 +88,10 @@ defmodule Commanded.Entities.ExecuteCommandForAggregateTest do
     ])
 
     assert_process_exit(account_number, fn ->
-      Aggregate.execute(account_number, %DepositMoney{account_number: account_number, transfer_uuid: UUID.uuid4, amount: 50}, DepositMoneyHandler, :handle)
+      command = %DepositMoney{account_number: account_number, transfer_uuid: UUID.uuid4, amount: 50}
+      context = %ExecutionContext{command: command, handler: DepositMoneyHandler, function: :handle}
+
+      Aggregate.execute(account_number, context)
     end)
   end
 
diff --git a/test/commands/routing_commands_test.exs b/test/commands/routing_commands_test.exs
index bb4dbf3..d75f712 100644
--- a/test/commands/routing_commands_test.exs
+++ b/test/commands/routing_commands_test.exs
@@ -142,4 +142,11 @@ defmodule Commanded.Commands.RoutingCommandsTest do
       assert {:ok, 2} == MultiCommandHandlerRouter.dispatch(%DepositMoney{account_number: "ACC123", amount: 100}, include_aggregate_version: true)
     end
   end
+
+  test "should allow setting metadata" do
+    metadata = %{ip_address: "127.0.0.1"}
+
+    assert :ok == MultiCommandHandlerRouter.dispatch(%OpenAccount{account_number: "ACC123", initial_balance: 1_000}, metadata: metadata)
+    assert :ok == MultiCommandHandlerRouter.dispatch(%DepositMoney{account_number: "ACC123", amount: 100}, metadata: metadata)
+  end
 end
diff --git a/test/event_store_adapter/append_events_test.exs b/test/event_store_adapter/append_events_test.exs
index dc3f1a6..47b7c77 100644
--- a/test/event_store_adapter/append_events_test.exs
+++ b/test/event_store_adapter/append_events_test.exs
@@ -36,16 +36,24 @@ defmodule Commanded.EventStore.Adapter.AppendEventsTest do
 
   describe "stream events from an existing stream" do
     test "should read events" do
-      events = build_events(4)
+      correlation_id = UUID.uuid4()
+      causation_id = UUID.uuid4()
+      events = build_events(4, correlation_id, causation_id)
 
       assert {:ok, 4} == EventStore.append_to_stream("stream", 0, events)
 
       read_events = EventStore.stream_forward("stream") |> Enum.to_list()
       assert length(read_events) == 4
       assert coerce(events) == coerce(read_events)
-
       assert pluck(read_events, :stream_version) == [1, 2, 3, 4]
 
+      Enum.each(read_events, fn event ->
+        assert event.stream_id == "stream"
+        assert event.correlation_id == correlation_id
+        assert event.causation_id == causation_id
+        assert event.metadata == %{"metadata" => "value"}
+      end)
+
       read_events = EventStore.stream_forward("stream", 3) |> Enum.to_list()
       assert coerce(Enum.slice(events, 2, 2)) == coerce(read_events)
       assert pluck(read_events, :stream_version) == [3, 4]
@@ -80,18 +88,20 @@ defmodule Commanded.EventStore.Adapter.AppendEventsTest do
     end
   end
 
-  defp build_event(account_number) do
+  defp build_event(account_number, correlation_id, causation_id) do
     %EventData{
-      correlation_id: UUID.uuid4,
+      correlation_id: correlation_id,
+      causation_id: causation_id,
       event_type: "Elixir.Commanded.EventStore.Adapter.AppendEventsTest.BankAccountOpened",
       data: %BankAccountOpened{account_number: account_number, initial_balance: 1_000},
       metadata: %{"metadata" => "value"},
     }
   end
 
-  defp build_events(count) do
-    for account_number <- 1..count, do: build_event(account_number)
+  defp build_events(count, correlation_id \\ UUID.uuid4(), causation_id \\ UUID.uuid4())
+  defp build_events(count, correlation_id, causation_id) do
+    for account_number <- 1..count, do: build_event(account_number, correlation_id, causation_id)
   end
 
-  defp coerce(events), do: Enum.map(events, &(%{correlation_id: &1.correlation_id, data: &1.data, metadata: &1.metadata}))
+  defp coerce(events), do: Enum.map(events, &(%{causation_id: &1.causation_id, correlation_id: &1.correlation_id, data: &1.data, metadata: &1.metadata}))
 end