Repository: slashdotdash/commanded (elixir)

Issue:
Additional stream prefix per aggregate
It would be useful to be able to define a prefix on a per-aggregate basis. This would reduce risk of collisions when using a lot of aggregates with generated IDs and ease separation of streams on a business logic level.

# Example

Let's assume we have a general User aggregate which uses UUIDs as identification. For our business logic we need to relate additional data to this user, as such we add additional aggregates which also use the user ids as identification for ease of use. While being related but separate entities this would result in a single stream containing all events from all aggregates per user.

# Proposal

Similar to process managers or event handlers the name of an aggregate could be defined in a use clause. This could internally define a `@behaviour` conformance with a callback providing the aggregate name.

To allow backwards compatibility the `Aggregate` module could use `function_exported?/3` to check for the behaviour conformance in advance and fall back to the current logic. The old behaviour could be deprecated if wished for and removed in a later version.

---

What do you think of this?

Pull request description:
Additional stream prefix per aggregate
Add support to command router to allow further specification of an aggregate identity and additionally an identity prefix.

1. Allow the aggregate identity prefix to be specified during command registration:

    ```elixir
    defmodule BankRouter do
      use Commanded.Commands.Router

      dispatch [OpenAccount,DepositMoney,WithdrawMoney] 
        to: BankAccount, 
        identity: :account_number,
        identity_prefix: "bank-account-"
    end
    ```

2. Allow a function to be used as the `identity` option; you construct the identity as required from the command being dispatched:

    ```elixir
    defmodule BankRouter do
      use Commanded.Commands.Router

      dispatch [RegisterUser] 
        to: User, 
        identity: fn %{uuid: uuid} -> "user-#{uuid}" end

      dispatch [OpenAccount,DepositMoney,WithdrawMoney] 
        to: BankAccount, 
        identity: &BankRouter.account_identity/1

      def account_identity(%{account_number: account_number}), do: "bank-account-#{account_number}"
    end
    ```

3. Allow an aggregate identity to be specified for all its commands, providing an optional prefix:

    ```elixir
    defmodule BankRouter do
      use Commanded.Commands.Router

      identify BankAccount, 
        by: :account_number, 
        prefix: "bank-account-"

      dispatch [OpenAccount,DepositMoney,WithdrawMoney] to: BankAccount
    end
    ```

Closes #77.

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 374ad1a..bc77f16 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,7 @@
 ### Enhancements
 
 - Dispatch command with `:eventual` or `:strong` consistency guarantee ([#82](https://github.com/slashdotdash/commanded/issues/82)).
+- Additional stream prefix per aggregate ([#77](https://github.com/slashdotdash/commanded/issues/77)).
 - Include custom metadata during command dispatch ([#61](https://github.com/slashdotdash/commanded/issues/61)).
 - Validate command dispatch registration in router ([59](https://github.com/slashdotdash/commanded/issues/59)).
 
diff --git a/guides/Commands.md b/guides/Commands.md
index 1a40892..ad664dd 100644
--- a/guides/Commands.md
+++ b/guides/Commands.md
@@ -2,7 +2,7 @@
 
 ## Commands
 
-Create a module per command and define the fields with `defstruct`. A command **must contain** a field to uniquely identify the aggregate instance (e.g. `account_number`).
+You need to create a module per command and define the fields using `defstruct`:
 
 ```elixir
 defmodule OpenAccount do
@@ -10,11 +10,39 @@ defmodule OpenAccount do
 end
 ```
 
+A command **must contain** a field to uniquely identify the aggregate instance (e.g. `account_number`).
+
+## Command dispatch and routing
+
+A router module is used to route and dispatch commands to their registered command handler or aggregate module.
+
+You create a router module, using `Commanded.Commands.Router`, and register each command with its associated handler:
+
+```elixir
+defmodule BankRouter do
+  use Commanded.Commands.Router
+
+  dispatch OpenAccount, to: OpenAccountHandler, aggregate: BankAccount, identity: :account_number
+  dispatch DepositMoney, to: DepositMoneyHandler, aggregate: BankAccount, identity: :account_number
+end
+```
+
+This can be more succinctly configured by excluding command handlers and [dispatching directly to the aggregate root](#dispatch-directly-to-aggregate), using [multi-command registration](#multi-command-registration), and with the [`identify`](#define-aggregate-identity) helper macro:
+
+```elixir
+defmodule BankRouter do
+  use Commanded.Commands.Router
+
+  identify BankAccount, by: :account_number
+  dispatch [OpenAccount, DepositMoney], to: BankAccount
+end
+```
+
 ## Command handlers
 
-Implement the `Commanded.Commands.Handler` behaviour consisting of a single `handle/2` function.
+A command handler receives the aggregate root and the command being executed. It allows you to validate, authorize, and/or enrich the command with additional data before executing the appropriate aggregate root module function.
 
-It receives the aggregate root state and the command to be handled. It must return the raised domain events from the aggregate root. It may return an `{:error, reason}` tuple on failure.
+The command handler must implement the `Commanded.Commands.Handler` behaviour consisting of a single `handle/2` function. It receives the aggregate root state and the command to be handled. It must return the raised domain events from the aggregate root. It may return an `{:error, reason}` tuple on failure.
 
 ```elixir
 defmodule OpenAccountHandler do
@@ -27,37 +55,62 @@ defmodule OpenAccountHandler do
 end
 ```
 
-## Command dispatch and routing
+Command handlers execute in the context of the dispatch call, as such they are limited to the timeout period specified. The default time is five seconds, the same as a `GenServer` call. You can increase the timeout value for individual commands as required - see the section on [Timeouts](#timeouts) below.
+
+### Dispatch directly to aggregate
 
-You must create a router to register each command with its associated handler.
+It is also possible to route a command directly to an aggregate root, without requiring an intermediate command handler.
 
 ```elixir
 defmodule BankRouter do
   use Commanded.Commands.Router
 
-  dispatch OpenAccount, to: OpenAccountHandler, aggregate: BankAccount, identity: :account_number
-  dispatch DepositMoney, to: DepositMoneyHandler, aggregate: BankAccount, identity: :account_number
+  dispatch OpenAccount, to: BankAccount, identity: :account_number
 end
 ```
 
-It is also possible to route a command directly to an aggregate root, without requiring an intermediate command handler.
+The aggregate root must implement an `execute/2` function that receives the aggregate's state and the command to execute.
+
+### Dispatching commands
+
+You can then dispatch a command using the router:
+
+```elixir
+:ok = BankRouter.dispatch(%OpenAccount{account_number: "ACC123", initial_balance: 1_000})
+```
+
+### Define aggregate identity
+
+You can define the identity field for an aggregate once using the `identify` macro. The configured identity will be used for all commands registered to the aggregate, unless overridden by a command registration.
+
+#### Example
 
 ```elixir
 defmodule BankRouter do
   use Commanded.Commands.Router
 
-  dispatch OpenAccount, to: BankAccount, identity: :account_number
+  identify BankAccount, by: :account_number
+
+  dispatch OpenAccount, to: BankAccount
 end
 ```
 
-The aggregate root must implement an `execute/2` function that receives the aggregate's state and the command to execute.
-
-You can then dispatch a command using the router.
+An optional identity prefix can be used to distinguish between different aggregates that  would otherwise share the same identity. As an example you might have a `User` and a `UserPreferences` aggregate that you wish to share the same identity. In this scenario you should specify a `prefix` for each aggregate (e.g. "user-" and "user-preference-").
 
 ```elixir
-:ok = BankRouter.dispatch(%OpenAccount{account_number: "ACC123", initial_balance: 1_000})
+defmodule BankRouter do
+  use Commanded.Commands.Router
+
+  identify BankAccount,
+    by: :account_number,
+    prefix: "bank-account-"
+
+  dispatch OpenAccount, to: BankAccount
+end
 ```
 
+The prefix is used as the stream identity when appending, and reading, the aggregate's events.
+
 ### Timeouts
 
 A command handler has a default timeout of 5 seconds. The same default as a `GenServer` process call. It must handle the command in this period, otherwise the call fails and the caller exits.
@@ -69,7 +122,11 @@ defmodule BankRouter do
   use Commanded.Commands.Router
 
   # configure a timeout of 1 second for the open account command handler
-  dispatch OpenAccount, to: OpenAccountHandler, aggregate: BankAccount, identity: :account_number, timeout: 1_000
+  dispatch OpenAccount,
+    to: OpenAccountHandler,
+    aggregate: BankAccount,
+    identity: :account_number,
+    timeout: 1_000
 end
 ```
 
@@ -81,13 +138,16 @@ You can override the timeout value during command dispatch. This example is disp
 
 ### Multi-command registration
 
-Command routers support multi command registration so you can group related command handlers into the same module.
+Command routers support multi command registration so you can group related command handlers into the same module:
 
 ```elixir
 defmodule BankRouter do
   use Commanded.Commands.Router
 
-  dispatch [OpenAccount,CloseAccount], to: BankAccountHandler, aggregate: BankAccount, identity: :account_number
+  dispatch [OpenAccount,CloseAccount],
+    to: BankAccountHandler,
+    aggregate: BankAccount,
+    identity: :account_number
 end
 ```
 
@@ -99,7 +159,7 @@ You can choose the consistency guarantee provided by a command dispatch using th
 :ok = BankRouter.dispatch(command, consistency: :strong)
 ```
 
-The available options are `:eventual` (default) and `:strong`:
+The available options are `:eventual` (default) and `:strong`.
 
 - *Strong consistency* offers up-to-date data but at the cost of high latency.
 - *Eventual consistency* offers low latency but read model queries may reply with stale data since they may not have processed the persisted events.
@@ -160,7 +220,10 @@ defmodule BankRouter do
   use Commanded.Commands.Router
 
   dispatch [OpenAccount,CloseAccount],
-    to: BankAccountHandler, aggregate: BankAccount, lifespan: BankAccountLifespan, identity: :account_number
+    to: BankAccountHandler,
+    aggregate: BankAccount,
+    lifespan: BankAccountLifespan,
+    identity: :account_number
 end
 ```
 
@@ -184,8 +247,10 @@ defmodule BankingRouter do
   middleware MyCommandValidator
   middleware AuthorizeCommand
 
-  dispatch OpenAccount, to: OpenAccountHandler, aggregate: BankAccount, identity: :account_number
-  dispatch DepositMoney, to: DepositMoneyHandler, aggregate: BankAccount, identity: :account_number
+  identify BankAccount, by: :account_number
+  
+  dispatch OpenAccount, to: OpenAccountHandler, aggregate: BankAccount
+  dispatch DepositMoney, to: DepositMoneyHandler, aggregate: BankAccount
 end
 ```
 
diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex
index 582496c..d1de0f6 100644
--- a/lib/commanded/aggregates/aggregate.ex
+++ b/lib/commanded/aggregates/aggregate.ex
@@ -31,25 +31,29 @@ defmodule Commanded.Aggregates.Aggregate do
     aggregate_uuid: nil,
     aggregate_state: nil,
     aggregate_version: 0,
+    identity_prefix: nil,
   ]
 
-  def start_link(aggregate_module, aggregate_uuid) do
-    name = via_tuple(aggregate_uuid)
+  def start_link(aggregate_module, aggregate_uuid, identity_prefix \\ nil)
+  def start_link(aggregate_module, aggregate_uuid, identity_prefix) do
+    name = via_tuple(aggregate_module, aggregate_uuid)
+
     GenServer.start_link(__MODULE__, %Aggregate{
       aggregate_module: aggregate_module,
-      aggregate_uuid: aggregate_uuid
+      aggregate_uuid: aggregate_uuid,
+      identity_prefix: identity_prefix,
     },
     name: name)
   end
 
-  defp via_tuple(aggregate_uuid) do
-    {:via, @registry_provider, {@aggregate_registry_name, aggregate_uuid}}
+  defp via_tuple(aggregate_module, aggregate_uuid) do
+    {:via, @registry_provider, {@aggregate_registry_name, {aggregate_module, 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})
+    GenServer.cast(self(), :populate_aggregate_state)
 
     {:ok, state}
   end
@@ -62,28 +66,32 @@ defmodule Commanded.Aggregates.Aggregate do
     (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.
+    The default value is five seconds (5,000ms).
 
   Returns `{:ok, aggregate_version}` on success, or `{:error, reason}` on failure.
   """
-  def execute(aggregate_uuid, %ExecutionContext{} = context, timeout \\ 5_000) do
-    GenServer.call(via_tuple(aggregate_uuid), {:execute_command, context}, timeout)
+  def execute(aggregate_module, aggregate_uuid, %ExecutionContext{} = context, timeout \\ 5_000) do
+    GenServer.call(via_tuple(aggregate_module, aggregate_uuid), {:execute_command, context}, timeout)
   end
 
   @doc false
-  def aggregate_state(aggregate_uuid), do: GenServer.call(via_tuple(aggregate_uuid), {:aggregate_state})
+  def aggregate_state(aggregate_module, aggregate_uuid) do
+    GenServer.call(via_tuple(aggregate_module, aggregate_uuid), :aggregate_state)
+  end
 
   @doc false
-  def aggregate_state(aggregate_uuid, timeout), do: GenServer.call(via_tuple(aggregate_uuid), {:aggregate_state}, timeout)
+  def aggregate_state(aggregate_module, aggregate_uuid, timeout) do
+    GenServer.call(via_tuple(aggregate_module, aggregate_uuid), :aggregate_state, timeout)
+  end
 
   @doc false
-  def aggregate_version(aggregate_uuid), do: GenServer.call(via_tuple(aggregate_uuid), {:aggregate_version})
+  def aggregate_version(aggregate_module, aggregate_uuid) do
+    GenServer.call(via_tuple(aggregate_module, aggregate_uuid), :aggregate_version)
+  end
 
   @doc false
-  def handle_cast({:populate_aggregate_state}, %Aggregate{} = state) do
-    state = populate_aggregate_state(state)
-
-    {:noreply, state}
+  def handle_cast(:populate_aggregate_state, %Aggregate{} = state) do
+    {:noreply, populate_aggregate_state(state)}
   end
 
   @doc false
@@ -94,12 +102,12 @@ defmodule Commanded.Aggregates.Aggregate do
   end
 
   @doc false
-  def handle_call({:aggregate_state}, _from, %Aggregate{aggregate_state: aggregate_state} = state) do
+  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
+  def handle_call(:aggregate_version, _from, %Aggregate{aggregate_version: aggregate_version} = state) do
     {:reply, aggregate_version, state}
   end
 
@@ -110,15 +118,17 @@ defmodule Commanded.Aggregates.Aggregate do
 
   # Load any existing events for the aggregate from storage and repopulate the state using those events
   defp populate_aggregate_state(%Aggregate{aggregate_module: aggregate_module} = state) do
-    rebuild_from_events(%Aggregate{state |
+    aggregate = %Aggregate{state |
       aggregate_version: 0,
       aggregate_state: struct(aggregate_module)
-    })
+    }
+
+    rebuild_from_events(aggregate)
   end
 
   # Load events from the event store, in batches, to rebuild the aggregate state
-  defp rebuild_from_events(%Aggregate{aggregate_uuid: aggregate_uuid, aggregate_module: aggregate_module} = state) do
-    case EventStore.stream_forward(aggregate_uuid, 0, @read_event_batch_size) do
+  defp rebuild_from_events(%Aggregate{aggregate_module: aggregate_module} = state) do
+    case EventStore.stream_forward(stream_uuid(state), 0, @read_event_batch_size) do
       {:error, :stream_not_found} ->
         # aggregate does not exist so return empty state
         state
@@ -152,7 +162,7 @@ defmodule Commanded.Aggregates.Aggregate 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)
+    %Aggregate{aggregate_module: aggregate_module, aggregate_version: expected_version, aggregate_state: aggregate_state} = state)
   do
     case Kernel.apply(handler, function, [aggregate_state, command]) do
       {:error, _reason} = reply -> {reply, state}
@@ -161,7 +171,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, metadata)
+        {:ok, stream_version} = persist_events(pending_events, stream_uuid(state), expected_version, metadata)
 
         state = %Aggregate{state |
           aggregate_state: updated_state,
@@ -176,11 +186,15 @@ defmodule Commanded.Aggregates.Aggregate do
     Enum.reduce(events, aggregate_state, &aggregate_module.apply(&2, &1))
   end
 
-  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
+  defp persist_events([], _stream_uuid, expected_version, _metadata), do: {:ok, expected_version}
+  defp persist_events(pending_events, stream_uuid, expected_version, metadata) do
+    correlation_id = UUID.uuid4()
     event_data = Mapper.map_to_event_data(pending_events, correlation_id, nil, metadata)
 
-    EventStore.append_to_stream(aggregate_uuid, expected_version, event_data)
+    EventStore.append_to_stream(stream_uuid, expected_version, event_data)
   end
+
+  # Get the stream indentity to read/append events for the aggregate, with an optional prefix
+  defp stream_uuid(%Aggregate{aggregate_uuid: aggregate_uuid, identity_prefix: nil}), do: aggregate_uuid
+  defp stream_uuid(%Aggregate{aggregate_uuid: aggregate_uuid, identity_prefix: identity_prefix}), do: identity_prefix <> aggregate_uuid
 end
diff --git a/lib/commanded/aggregates/supervisor.ex b/lib/commanded/aggregates/supervisor.ex
index fc31bf6..e090e30 100644
--- a/lib/commanded/aggregates/supervisor.ex
+++ b/lib/commanded/aggregates/supervisor.ex
@@ -15,14 +15,15 @@ defmodule Commanded.Aggregates.Supervisor do
 
   Returns `{:ok, aggregate_uuid}` when a process is sucessfully started, or is already running.
   """
-  def open_aggregate(aggregate_module, aggregate_uuid)
+  def open_aggregate(aggregate_module, aggregate_uuid, identity_prefix \\ nil)
+  def open_aggregate(aggregate_module, aggregate_uuid, identity_prefix)
     when is_integer(aggregate_uuid) or
          is_atom(aggregate_uuid) or
          is_bitstring(aggregate_uuid) do
 
     Logger.debug(fn -> "Locating aggregate process for `#{inspect aggregate_module}` with UUID #{inspect aggregate_uuid}" end)
 
-    case Supervisor.start_child(__MODULE__, [aggregate_module, aggregate_uuid]) do
+    case Supervisor.start_child(__MODULE__, [aggregate_module, aggregate_uuid, identity_prefix]) do
       {:ok, _pid} -> {:ok, aggregate_uuid}
       {:error, {:already_started, _pid}} -> {:ok, aggregate_uuid}
       other -> {:error, other}
diff --git a/lib/commanded/commands/dispatcher.ex b/lib/commanded/commands/dispatcher.ex
index ac8b880..0a3b075 100644
--- a/lib/commanded/commands/dispatcher.ex
+++ b/lib/commanded/commands/dispatcher.ex
@@ -17,6 +17,7 @@ defmodule Commanded.Commands.Dispatcher do
       aggregate_module: nil,
       include_aggregate_version: nil,
       identity: nil,
+      identity_prefix: nil,
       timeout: nil,
       lifespan: nil,
       metadata: nil,
@@ -57,12 +58,12 @@ defmodule Commanded.Commands.Dispatcher do
 
   defp execute(
     %Pipeline{assigns: %{aggregate_uuid: aggregate_uuid}} = pipeline,
-    %Payload{timeout: timeout} = payload)
+    %Payload{aggregate_module: aggregate_module, identity_prefix: identity_prefix, timeout: timeout} = payload)
   do
-    {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(payload.aggregate_module, aggregate_uuid)
+    {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(aggregate_module, aggregate_uuid, identity_prefix)
 
     context = to_execution_context(pipeline, payload)
-    task = Task.Supervisor.async_nolink(Commanded.Commands.TaskDispatcher, Aggregates.Aggregate, :execute, [aggregate_uuid, context, timeout])
+    task = Task.Supervisor.async_nolink(Commanded.Commands.TaskDispatcher, Aggregates.Aggregate, :execute, [aggregate_module, aggregate_uuid, context, timeout])
     task_result = Task.yield(task, timeout) || Task.shutdown(task)
 
     result =
diff --git a/lib/commanded/commands/router.ex b/lib/commanded/commands/router.ex
index 5bcac50..baa3564 100644
--- a/lib/commanded/commands/router.ex
+++ b/lib/commanded/commands/router.ex
@@ -27,7 +27,7 @@ defmodule Commanded.Commands.Router do
   You can route a command directly to an aggregate root, without requiring an
   intermediate command handler.
 
-  ## Example
+  ### Example
 
       defmodule BankRouter do
         use Commanded.Commands.Router
@@ -38,6 +38,33 @@ defmodule Commanded.Commands.Router do
   The aggregate root must implement an `execute/2` function that receives the
   aggregate's state and the command being executed.
 
+  ## Define aggregate identity
+
+  You can define the identity field for an aggregate once using the `identify` macro.
+  The configured identity will be used for all commands registered to the aggregate,
+  unless overridden by a command registration.
+
+  ### Example
+
+      defmodule BankRouter do
+        use Commanded.Commands.Router
+
+        identify BankAccount,
+          by: :account_number,
+          prefix: "bank-account-"
+
+        dispatch OpenAccount, to: BankAccount
+      end
+
+  An optional identity prefix can be used to distinguish between different
+  aggregates that  would otherwise share the same identity. As an example you
+  might have a `User` and a `UserPreferences` aggregate that you wish
+  to share the same identity. In this scenario you should specify a `prefix`
+  for each aggregate (e.g. "user-" and "user-preference-").
+
+  The prefix is used as the stream identity when appending, and reading, the
+  aggregate's events: "<identity_prefix><aggregate_uuid>".
+
   ## Consistency
 
   You can choose to dispatch commands using either `:eventual` or `:strong`
@@ -79,6 +106,7 @@ defmodule Commanded.Commands.Router do
       @before_compile unquote(__MODULE__)
       @registered_commands []
       @registered_middleware []
+      @registered_identities %{}
       @default_middleware [
         Commanded.Middleware.ExtractAggregateIdentity,
         Commanded.Middleware.ConsistencyGuarantee,
@@ -117,6 +145,23 @@ defmodule Commanded.Commands.Router do
     end
   end
 
+  @doc """
+  Define an aggregate's identity
+
+  You can define the identity field for an aggregate using the `identify` macro.
+  The configured identity will be used for all commands registered to the
+  aggregate, unless overridden by a command registration.
+  """
+  defmacro identify(aggregate_module, opts) do
+    quote location: :keep do
+      if Map.has_key?(@registered_identities, unquote(aggregate_module)) do
+        raise "aggregate #{inspect unquote(aggregate_module)} has already been identified by"
+      end
+
+      @registered_identities Map.put(@registered_identities, unquote(aggregate_module), unquote(opts))
+    end
+  end
+
   @doc """
   Configure the command, or list of commands, to be dispatched to the
   corresponding handler for a given aggregate root
@@ -133,10 +178,28 @@ defmodule Commanded.Commands.Router do
     end)
   end
 
-  @register_params [:to, :function, :aggregate, :identity, :timeout, :lifespan, :consistency]
+  @register_params [
+    :to,
+    :function,
+    :aggregate,
+    :identity,
+    :identity_prefix,
+    :timeout,
+    :lifespan,
+    :consistency,
+  ]
 
   @doc false
-  defmacro register(command_module, to: handler, function: function, aggregate: aggregate, identity: identity, timeout: timeout, lifespan: lifespan, consistency: consistency) do
+  defmacro register(command_module,
+    to: handler,
+    function: function,
+    aggregate: aggregate,
+    identity: identity,
+    identity_prefix: identity_prefix,
+    timeout: timeout,
+    lifespan: lifespan,
+    consistency: consistency)
+  do
     quote location: :keep do
       if Enum.member?(@registered_commands, unquote(command_module)) do
         raise "duplicate command registration for: #{inspect unquote(command_module)}"
@@ -199,10 +262,26 @@ defmodule Commanded.Commands.Router do
       def dispatch(%unquote(command_module){} = command, opts), do: do_dispatch(command, opts)
 
       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)
+        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
+        lifespan = unquote(lifespan) || @default_lifespan
+
+        default_identity = unquote(identity)
+        default_identity_prefix = unquote(identity_prefix)
+
+        {identity, identity_prefix} =
+          case Map.get(@registered_identities, unquote(aggregate)) do
+            nil ->
+              {default_identity, default_identity_prefix}
+
+            config ->
+              identity = Keyword.get(config, :by, default_identity)
+              prefix = default_identity_prefix || Keyword.get(config, :prefix, default_identity_prefix)
+
+              {identity, prefix}
+          end
 
         Commanded.Commands.Dispatcher.dispatch(%Commanded.Commands.Dispatcher.Payload{
           command: command,
@@ -210,10 +289,11 @@ defmodule Commanded.Commands.Router do
           handler_module: unquote(handler),
           handler_function: unquote(function),
           aggregate_module: unquote(aggregate),
-          identity: unquote(identity),
+          identity: identity,
+          identity_prefix: identity_prefix,
           include_aggregate_version: include_aggregate_version,
           timeout: timeout,
-          lifespan: unquote(lifespan) || @default_lifespan,
+          lifespan: lifespan,
           metadata: metadata,
           middleware: @registered_middleware ++ @default_middleware,
         })
diff --git a/lib/commanded/middleware/extract_aggregate_identity.ex b/lib/commanded/middleware/extract_aggregate_identity.ex
index ceeae73..fbc48a6 100644
--- a/lib/commanded/middleware/extract_aggregate_identity.ex
+++ b/lib/commanded/middleware/extract_aggregate_identity.ex
@@ -24,7 +24,13 @@ defmodule Commanded.Middleware.ExtractAggregateIdentity do
 
   def after_failure(%Pipeline{} = pipeline), do: pipeline
 
-  defp extract_aggregate_uuid(%Pipeline{command: command, identity: identity}) do
+  # extract identity using a user-provider function
+  defp extract_aggregate_uuid(%Pipeline{command: command, identity: identity}) when is_function(identity) do
+    identity.(command)
+  end
+
+  # extract identity using a field in the command
+  defp extract_aggregate_uuid(%Pipeline{command: command, identity: identity}) when is_atom(identity) do
     Map.get(command, identity)
   end
 end


Test patch:
diff --git a/test/aggregates/aggregate_lifespan_test.exs b/test/aggregates/aggregate_lifespan_test.exs
index 4685163..00d9395 100644
--- a/test/aggregates/aggregate_lifespan_test.exs
+++ b/test/aggregates/aggregate_lifespan_test.exs
@@ -36,7 +36,7 @@ defmodule Commanded.Aggregates.AggregateLifespanTest do
 
       {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(BankAccount, aggregate_uuid)
 
-      pid = apply(@registry_provider, :whereis_name, [{:aggregate_registry, aggregate_uuid}])
+      pid = apply(@registry_provider, :whereis_name, [{:aggregate_registry, {BankAccount, aggregate_uuid}}])
       ref = Process.monitor(pid)
 
       %{aggregate_uuid: aggregate_uuid, ref: ref}
diff --git a/test/aggregates/event_persistence_test.exs b/test/aggregates/event_persistence_test.exs
index 85c2f6b..bf4cb67 100644
--- a/test/aggregates/event_persistence_test.exs
+++ b/test/aggregates/event_persistence_test.exs
@@ -52,7 +52,7 @@ defmodule Commanded.Entities.EventPersistenceTest do
 
     {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
 
-    {:ok, 10} = Aggregate.execute(aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 10}, handler: AppendItemsHandler, function: :handle})
+    {:ok, 10} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 10}, handler: AppendItemsHandler, function: :handle})
 
     recorded_events = EventStore.stream_forward(aggregate_uuid, 0) |> Enum.to_list()
 
@@ -72,7 +72,7 @@ defmodule Commanded.Entities.EventPersistenceTest do
     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)
+    {:ok, 10} = Aggregate.execute(ExampleAggregate, aggregate_uuid, context)
 
     recorded_events = EventStore.stream_forward(aggregate_uuid, 0) |> Enum.to_list()
 
@@ -86,14 +86,14 @@ defmodule Commanded.Entities.EventPersistenceTest do
 
     {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
 
-    {:ok, 10} = Aggregate.execute(aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 10}, handler: AppendItemsHandler, function: :handle})
+    {:ok, 10} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 10}, handler: AppendItemsHandler, function: :handle})
 
-    Commanded.Helpers.Process.shutdown(aggregate_uuid)
+    Commanded.Helpers.Process.shutdown(ExampleAggregate, aggregate_uuid)
 
     {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
 
-    assert Aggregate.aggregate_version(aggregate_uuid) == 10
-    assert Aggregate.aggregate_state(aggregate_uuid) == %Commanded.Entities.EventPersistenceTest.ExampleAggregate{
+    assert Aggregate.aggregate_version(ExampleAggregate, aggregate_uuid) == 10
+    assert Aggregate.aggregate_state(ExampleAggregate, aggregate_uuid) == %Commanded.Entities.EventPersistenceTest.ExampleAggregate{
       items: 1..10 |> Enum.to_list(),
       last_index: 10,
     }
@@ -104,18 +104,34 @@ defmodule Commanded.Entities.EventPersistenceTest do
 
     {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
 
-    {: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})
+    {:ok, 100} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 100}, handler: AppendItemsHandler, function: :handle})
+    {:ok, 200} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 100}, handler: AppendItemsHandler, function: :handle})
+    {:ok, 201} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 1}, handler: AppendItemsHandler, function: :handle})
 
-    Commanded.Helpers.Process.shutdown(aggregate_uuid)
+    Commanded.Helpers.Process.shutdown(ExampleAggregate, aggregate_uuid)
 
     {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
 
-    assert Aggregate.aggregate_version(aggregate_uuid) == 201
-    assert Aggregate.aggregate_state(aggregate_uuid) == %Commanded.Entities.EventPersistenceTest.ExampleAggregate{
+    assert Aggregate.aggregate_version(ExampleAggregate, aggregate_uuid) == 201
+    assert Aggregate.aggregate_state(ExampleAggregate, aggregate_uuid) == %Commanded.Entities.EventPersistenceTest.ExampleAggregate{
       items: 1..201 |> Enum.to_list,
       last_index: 201,
     }
   end
+
+  test "should prefix stream UUID with aggregate indentity prefix" do
+    aggregate_uuid = UUID.uuid4()
+    prefix = "example-preifx-"
+
+    {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid, prefix)
+
+    context = %ExecutionContext{command: %AppendItems{count: 1}, handler: AppendItemsHandler, function: :handle}
+
+    {:ok, 1} = Aggregate.execute(ExampleAggregate, aggregate_uuid, context)
+
+    recorded_events = EventStore.stream_forward(prefix <> aggregate_uuid, 0) |> Enum.to_list()
+    assert length(recorded_events) == 1
+
+    assert {:error, :stream_not_found} == EventStore.stream_forward(aggregate_uuid, 0)
+  end
 end
diff --git a/test/aggregates/execute_command_for_aggregate_test.exs b/test/aggregates/execute_command_for_aggregate_test.exs
index 955748f..45028b9 100644
--- a/test/aggregates/execute_command_for_aggregate_test.exs
+++ b/test/aggregates/execute_command_for_aggregate_test.exs
@@ -18,15 +18,15 @@ defmodule Commanded.Entities.ExecuteCommandForAggregateTest do
     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)
+    {:ok, 1} = Aggregate.execute(BankAccount, account_number, context)
 
-    Helpers.Process.shutdown(account_number)
+    Helpers.Process.shutdown(BankAccount, account_number)
 
     # reload aggregate to fetch persisted events from event store and rebuild state by applying saved events
     {:ok, ^account_number} = Commanded.Aggregates.Supervisor.open_aggregate(BankAccount, account_number)
 
-    assert Aggregate.aggregate_version(account_number) == 1
-    assert Aggregate.aggregate_state(account_number) == %BankAccount{account_number: account_number, balance: 1_000, state: :active}
+    assert Aggregate.aggregate_version(BankAccount, account_number) == 1
+    assert Aggregate.aggregate_state(BankAccount, account_number) == %BankAccount{account_number: account_number, balance: 1_000, state: :active}
   end
 
   test "execute command via a command handler" do
@@ -37,15 +37,15 @@ defmodule Commanded.Entities.ExecuteCommandForAggregateTest do
     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)
+    {:ok, 1} = Aggregate.execute(BankAccount, account_number, context)
 
-    Helpers.Process.shutdown(account_number)
+    Helpers.Process.shutdown(BankAccount, account_number)
 
     # reload aggregate to fetch persisted events from event store and rebuild state by applying saved events
     {:ok, ^account_number} = Commanded.Aggregates.Supervisor.open_aggregate(BankAccount, account_number)
 
-    assert Aggregate.aggregate_version(account_number) == 1
-    assert Aggregate.aggregate_state(account_number) == %BankAccount{account_number: account_number, balance: 1_000, state: :active}
+    assert Aggregate.aggregate_version(BankAccount, account_number) == 1
+    assert Aggregate.aggregate_state(BankAccount, account_number) == %BankAccount{account_number: account_number, balance: 1_000, state: :active}
   end
 
   test "aggregate raising an exception should not persist pending events or state" do
@@ -56,19 +56,19 @@ defmodule Commanded.Entities.ExecuteCommandForAggregateTest do
     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)
+    {:ok, 1} = Aggregate.execute(BankAccount, account_number, context)
 
-    state_before = Aggregate.aggregate_state(account_number)
+    state_before = Aggregate.aggregate_state(BankAccount, account_number)
 
     assert_process_exit(account_number, fn ->
       command = %OpenAccount{account_number: account_number, initial_balance: 1}
       context = %ExecutionContext{command: command, handler: OpenAccountHandler, function: :handle}
 
-      Aggregate.execute(account_number, context)
+      Aggregate.execute(BankAccount, account_number, context)
     end)
 
     {:ok, ^account_number} = Commanded.Aggregates.Supervisor.open_aggregate(BankAccount, account_number)
-    assert state_before == Aggregate.aggregate_state(account_number)
+    assert state_before == Aggregate.aggregate_state(BankAccount, account_number)
   end
 
   test "executing a command against an aggregate with concurrency error should terminate aggregate process" do
@@ -77,7 +77,7 @@ defmodule Commanded.Entities.ExecuteCommandForAggregateTest do
     {:ok, ^account_number} = Commanded.Aggregates.Supervisor.open_aggregate(BankAccount, account_number)
 
     # block until aggregate has loaded its initial (empty) state
-    Aggregate.aggregate_state(account_number)
+    Aggregate.aggregate_state(BankAccount, account_number)
 
     # write an event to the aggregate's stream, bypassing the aggregate process (simulate concurrency error)
     {:ok, _} = EventStore.append_to_stream(account_number, 0, [
@@ -91,7 +91,7 @@ defmodule Commanded.Entities.ExecuteCommandForAggregateTest do
       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)
+      Aggregate.execute(BankAccount, account_number, context)
     end)
   end
 
@@ -102,6 +102,6 @@ defmodule Commanded.Entities.ExecuteCommandForAggregateTest do
 
     # process should exit
     assert_receive({:EXIT, _from, _reason})
-    assert apply(@registry_provider, :whereis_name, [{:aggregate_registry, aggregate_uuid}]) == :undefined
+    assert apply(@registry_provider, :whereis_name, [{:aggregate_registry, {BankAccount, aggregate_uuid}}]) == :undefined
   end
 end
diff --git a/test/aggregates/registry_test.exs b/test/aggregates/registry_test.exs
index cc642b3..37e1b47 100644
--- a/test/aggregates/registry_test.exs
+++ b/test/aggregates/registry_test.exs
@@ -3,7 +3,7 @@ defmodule SupervisorTest do
 
   test "should let use integer, atom or string as aggregate_uuid" do
     Enum.each([1, :atom, "string"], fn aggregate_uuid ->
-      {:ok, _aggregate} = Commanded.Aggregates.Supervisor.open_aggregate(Commanded.ExampleDomain.BankAccount, aggregate_uuid)
+      {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(Commanded.ExampleDomain.BankAccount, aggregate_uuid)
     end)
   end
 end
diff --git a/test/commands/routing_commands_test.exs b/test/commands/routing_commands_test.exs
index 5b089bd..dac00df 100644
--- a/test/commands/routing_commands_test.exs
+++ b/test/commands/routing_commands_test.exs
@@ -1,7 +1,7 @@
 defmodule Commanded.Commands.RoutingCommandsTest do
   use Commanded.StorageCase
-  doctest Commanded.Commands.Router
 
+  alias Commanded.EventStore
   alias Commanded.ExampleDomain.BankAccount
   alias Commanded.ExampleDomain.{OpenAccountHandler,DepositMoneyHandler,WithdrawMoneyHandler}
   alias Commanded.ExampleDomain.BankAccount.Commands.{OpenAccount,CloseAccount,DepositMoney,WithdrawMoney}
@@ -53,6 +53,75 @@ defmodule Commanded.Commands.RoutingCommandsTest do
     end
   end
 
+  describe "identify aggregate" do
+    defmodule IdentityCommand, do: defstruct [:uuid]
+    defmodule IdentityEvent, do: defstruct [:uuid]
+
+    defmodule IdentityAggregateRoot do
+      defstruct [uuid: nil]
+
+      def execute(%__MODULE__{}, %IdentityCommand{uuid: uuid}), do: %IdentityEvent{uuid: uuid}
+      def apply(aggregate, _event), do: aggregate
+    end
+
+    defmodule IdentityAggregateRouter do
+      use Commanded.Commands.Router
+
+      identify IdentityAggregateRoot,
+        by: :uuid,
+        prefix: "prefix-"
+
+      dispatch IdentityCommand, to: IdentityAggregateRoot
+    end
+
+    test "should dispatch command to registered handler" do
+      assert :ok = IdentityAggregateRouter.dispatch(%IdentityCommand{uuid: UUID.uuid4})
+    end
+
+    test "should append events to stream using identity prefix" do
+      uuid = UUID.uuid4()
+      assert :ok = IdentityAggregateRouter.dispatch(%IdentityCommand{uuid: uuid})
+
+      recorded_events = EventStore.stream_forward("prefix-" <> uuid, 0) |> Enum.to_list()
+      assert length(recorded_events) == 1
+    end
+  end
+
+  describe "identify aggregate using function" do
+    defmodule IdentityFunctionCommand, do: defstruct [:uuid]
+    defmodule IdentityFunctionEvent, do: defstruct [:uuid]
+
+    defmodule IdentityFunctionAggregateRoot do
+      defstruct [uuid: nil]
+
+      def execute(%__MODULE__{}, %IdentityFunctionCommand{uuid: uuid}), do: %IdentityFunctionEvent{uuid: uuid}
+      def apply(aggregate, _event), do: aggregate
+    end
+
+    defmodule IdentityFunctionAggregateRouter do
+      use Commanded.Commands.Router
+
+      identify IdentityFunctionAggregateRoot,
+        by: &IdentityFunctionAggregateRouter.aggregate_identity/1
+
+      dispatch IdentityFunctionCommand, to: IdentityFunctionAggregateRoot
+
+      def aggregate_identity(%{uuid: uuid}), do: "fun-prefix-" <> uuid
+    end
+
+    test "should dispatch command to registered handler" do
+      assert :ok = IdentityFunctionAggregateRouter.dispatch(%IdentityFunctionCommand{uuid: UUID.uuid4})
+    end
+
+    test "should append events to stream" do
+      uuid = UUID.uuid4()
+      assert :ok = IdentityFunctionAggregateRouter.dispatch(%IdentityFunctionCommand{uuid: uuid})
+
+      recorded_events = EventStore.stream_forward("fun-prefix-" <> uuid, 0) |> Enum.to_list()
+      assert length(recorded_events) == 1
+    end
+  end
+
   test "should prevent duplicate registrations for a command" do
     # compile time safety net to prevent duplicate command registrations
     assert_raise RuntimeError, "duplicate command registration for: Commanded.ExampleDomain.BankAccount.Commands.OpenAccount", fn ->
@@ -96,7 +165,7 @@ defmodule Commanded.Commands.RoutingCommandsTest do
   test "should show a help note when bad argument given to a `dispatch/2` function" do
     assert_raise RuntimeError, """
     unexpected dispatch parameter "id"
-    available params are: to, function, aggregate, identity, timeout, lifespan, consistency
+    available params are: to, function, aggregate, identity, identity_prefix, timeout, lifespan, consistency
     """,
     fn ->
       Code.eval_string """
diff --git a/test/helpers/process.ex b/test/helpers/process.ex
index a28cd7b..186bb41 100644
--- a/test/helpers/process.ex
+++ b/test/helpers/process.ex
@@ -15,8 +15,15 @@ defmodule Commanded.Helpers.Process do
     assert_receive {:DOWN, ^ref, _, _, _}, 5_000
   end
 
-  def shutdown(aggregate_uuid) do
-    pid = apply(@registry_provider, :whereis_name, [{:aggregate_registry, aggregate_uuid}])
+  def shutdown(name) when is_atom(name) do
+    case Process.whereis(name) do
+      nil -> :ok
+      pid -> shutdown(pid)
+    end
+  end
+
+  def shutdown(aggregate_module, aggregate_uuid) do
+    pid = apply(@registry_provider, :whereis_name, [{:aggregate_registry, {aggregate_module, aggregate_uuid}}])
     shutdown(pid)
   end
 end