Repository: slashdotdash/commanded
Pull request #98 (2017-10-18 11:39:51)

Add `Commanded.Aggregate.Multi`
Use `Commanded.Aggregate.Multi` to generate multiple events from a single command.

This can be useful when you want to emit multiple events that depend upon the aggregate state being updated.

## Example

In the example below, money is withdrawn from the bank account and the
updated balance is used to check whether the account is overdrawn.

```elixir
defmodule BankAccount do
  defstruct [
    account_number: nil,
    balance: 0,
    state: nil,
  ]

  alias Commanded.Aggregate.Multi

  def withdraw(
    %BankAccount{state: :active} = account,
    %WithdrawMoney{amount: amount})
    when is_number(amount) and amount > 0
  do
    account
    |> Multi.new()
    |> Multi.execute(&withdraw_money(&1, amount))
    |> Multi.execute(&check_balance/1)
  end

  def apply(%BankAccount{} = state, %MoneyWithdrawn{balance: balance}),
    do: %BankAccount{state | balance: balance}
 
  def apply(%BankAccount{} = state, %AccountOverdrawn{}),
    do: %BankAccount{state | state: :overdrawn}

  # private helpers

  defp withdraw_money(%BankAccount{account_number: account_number, balance: balance}, amount) do
    %MoneyWithdrawn{
      account_number: account_number,
      amount: amount,
      balance: balance - amount
    }
  end

  defp check_balance(%BankAccount{account_number: account_number, balance: balance}) when balance < 0 do
    %AccountOverdrawn{account_number: account_number, balance: balance}
  end
  defp check_balance(%BankAccount{}), do: []
end
```

---

Inspired by [Chronik](https://github.com/parody/chronik/blob/4f9753c9912c29d3652f76b4d8f272ddd03e6216/lib/chronik/aggregate/multi.ex) and [Ecto.Multi](https://hexdocs.pm/ecto/Ecto.Multi.html).


Patch:
diff --git a/guides/Aggregate Roots.md b/guides/Aggregate Roots.md
index 904b446..adf836c 100644
--- a/guides/Aggregate Roots.md
+++ b/guides/Aggregate Roots.md
@@ -55,7 +55,6 @@ An alternative approach is to expose one or more public command functions, `exec
 
 In this case the example function matches on the `OpenAccount` command module:
 
-
 ```elixir
 defmodule BankAccount do
   defstruct [account_number: nil, balance: nil]
@@ -68,11 +67,8 @@ defmodule BankAccount do
     %BankAccountOpened{account_number: account_number, initial_balance: initial_balance}
   end
 
-  def execute(%BankAccount{} = account,  %OpenAccount{initial_balance: initial_balance})
-    when initial_balance <= 0
-  do
-    {:error, :initial_balance_must_be_above_zero}
-  end
+  def execute(%BankAccount{} = account, %OpenAccount{initial_balance: initial_balance})
+    do: {:error, :initial_balance_must_be_above_zero}
 
   # state mutators
 
@@ -84,3 +80,55 @@ defmodule BankAccount do
   end
 end
 ```
+
+## Using `Commanded.Aggregate.Multi` to return multiple events
+
+Sometimes you need to create multiple events from a single command. You can use `Commanded.Aggregate.Multi` to help track the events and update the aggregate state. This can be useful when you want to emit multiple events that depend upon the aggregate state being updated.
+
+Any errors encountered will be returned to the caller, the modified aggregate state and any pending events are discarded.
+
+## Example
+
+In the example below, money is withdrawn from the bank account and the updated balance is used to check whether the account is overdrawn.
+
+```elixir
+defmodule BankAccount do
+  defstruct [
+    account_number: nil,
+    balance: 0,
+    state: nil,
+  ]
+
+  alias Commanded.Aggregate.Multi
+
+  # public command API
+
+  def withdraw(
+    %BankAccount{state: :active} = account,
+    %WithdrawMoney{amount: amount})
+    when is_number(amount) and amount > 0
+  do
+    account
+    |> Multi.new()
+    |> Multi.execute(&withdraw_money(&1, amount))
+    |> Multi.execute(&check_balance/1)
+  end
+
+  # private helpers
+
+  defp withdraw_money(%BankAccount{account_number: account_number, balance: balance}, amount) do
+    %MoneyWithdrawn{
+      account_number: account_number,
+      amount: amount,
+      balance: balance - amount
+    }
+  end
+
+  defp check_balance(%BankAccount{account_number: account_number, balance: balance})
+    when balance < 0
+  do
+    %AccountOverdrawn{account_number: account_number, balance: balance}
+  end
+  defp check_balance(%BankAccount{}), do: []
+end
+```
diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex
index 081d246..5053c2f 100644
--- a/lib/commanded/aggregates/aggregate.ex
+++ b/lib/commanded/aggregates/aggregate.ex
@@ -178,19 +178,20 @@ defmodule Commanded.Aggregates.Aggregate do
       none when none in [nil, []] ->
         {{:ok, expected_version, []}, state}
 
-      events ->
-        pending_events = List.wrap(events)
-
-        updated_state = apply_events(aggregate_module, aggregate_state, pending_events)
+      %Commanded.Aggregate.Multi{} = multi ->
+        case Commanded.Aggregate.Multi.run(multi) do
+          {:error, _reason} = reply ->
+            {reply, state}
 
-        {:ok, stream_version} = persist_events(pending_events, stream_uuid(state), expected_version, metadata)
+          {aggregate_state, pending_events} ->
+            persist_events(pending_events, metadata, %Aggregate{state | aggregate_state: aggregate_state})
+        end
 
-        state = %Aggregate{state |
-          aggregate_state: updated_state,
-          aggregate_version: stream_version
-        }
+      events ->
+        pending_events = List.wrap(events)
+        aggregate_state = apply_events(aggregate_module, aggregate_state, pending_events)
 
-        {{:ok, stream_version, pending_events}, state}
+        persist_events(pending_events, metadata, %Aggregate{state | aggregate_state: aggregate_state})
     end
   end
 
@@ -198,8 +199,14 @@ defmodule Commanded.Aggregates.Aggregate do
     Enum.reduce(events, aggregate_state, &aggregate_module.apply(&2, &1))
   end
 
-  defp persist_events([], _stream_uuid, expected_version, _metadata), do: {:ok, expected_version}
-  defp persist_events(pending_events, stream_uuid, expected_version, metadata) do
+  defp persist_events(pending_events, metadata, %Aggregate{aggregate_version: expected_version} = state) do
+    {:ok, stream_version} = append_to_stream(pending_events, stream_uuid(state), expected_version, metadata)
+
+    {{:ok, stream_version, pending_events}, %Aggregate{state | aggregate_version: stream_version}}
+  end
+
+  defp append_to_stream([], _stream_uuid, expected_version, _metadata), do: {:ok, expected_version}
+  defp append_to_stream(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)
 
diff --git a/lib/commanded/aggregates/multi.ex b/lib/commanded/aggregates/multi.ex
new file mode 100644
index 0000000..1a17e51
--- /dev/null
+++ b/lib/commanded/aggregates/multi.ex
@@ -0,0 +1,107 @@
+defmodule Commanded.Aggregate.Multi do
+  @moduledoc """
+  Use `Commanded.Aggregate.Multi` to generate multiple events from a single
+  command.
+
+  This can be useful when you want to emit multiple events that depend upon the
+  aggregate state being updated.
+
+  ## Example
+
+  In the example below, money is withdrawn from the bank account and the
+  updated balance is used to check whether the account is overdrawn.
+
+      defmodule BankAccount do
+        defstruct [
+          account_number: nil,
+          balance: 0,
+          state: nil,
+        ]
+
+        alias Commanded.Aggregate.Multi
+
+        def withdraw(
+          %BankAccount{state: :active} = account,
+          %WithdrawMoney{amount: amount})
+          when is_number(amount) and amount > 0
+        do
+          account
+          |> Multi.new()
+          |> Multi.execute(&withdraw_money(&1, amount))
+          |> Multi.execute(&check_balance/1)
+        end
+
+        defp withdraw_money(%BankAccount{account_number: account_number, balance: balance}, amount) do
+          %MoneyWithdrawn{
+            account_number: account_number,
+            amount: amount,
+            balance: balance - amount
+          }
+        end
+
+        defp check_balance(%BankAccount{account_number: account_number, balance: balance})
+          when balance < 0
+        do
+          %AccountOverdrawn{account_number: account_number, balance: balance}
+        end
+        defp check_balance(%BankAccount{}), do: []
+      end
+  """
+
+  alias Commanded.Aggregate.Multi
+
+  @type t :: %__MODULE__{
+    aggregate: struct(),
+    executions: list(function()),
+  }
+
+  defstruct [
+    aggregate: nil,
+    executions: [],
+  ]
+
+  @doc """
+  Create a new `Commanded.Aggregate.Multi` struct.
+  """
+  @spec new(aggregate :: struct()) :: Multi.t
+  def new(aggregate), do: %Multi{aggregate: aggregate}
+
+  @doc """
+  Adds a command execute function to the multi.
+  """
+  @spec execute(Multi.t, function()) :: Multi.t
+  def execute(%Multi{executions: executions} = multi, execute_fun)
+    when is_function(execute_fun)
+  do
+    %Multi{multi |
+      executions: [execute_fun | executions],
+    }
+  end
+
+  @doc """
+  Run the execute functions contained within the multi, returning the updated
+  aggregate state and any created events.
+  """
+  @spec run(Multi.t) :: {aggregate :: struct(), list(event :: struct())} | {:error, reason :: any()}
+  def run(%Multi{aggregate: aggregate, executions: executions}) do
+    try do
+      executions
+      |> Enum.reverse()
+      |> Enum.reduce({aggregate, []}, fn (execute_fun, {aggregate, events}) ->
+        pending_events =
+          case execute_fun.(aggregate) do
+            {:error, _reason} = error -> throw(error)
+            pending_events -> List.wrap(pending_events)
+          end
+
+        {apply_events(aggregate, pending_events), events ++ pending_events}
+      end)
+    catch
+      {:error, _reason} = error -> error
+    end
+  end
+
+  defp apply_events(aggregate, events) do
+    Enum.reduce(events, aggregate, &aggregate.__struct__.apply(&2, &1))
+  end
+end
diff --git a/mix.exs b/mix.exs
index e307d07..8a17be6 100644
--- a/mix.exs
+++ b/mix.exs
@@ -30,7 +30,7 @@ defmodule Commanded.Mixfile do
     ]
   end
 
-  defp elixirc_paths(:test), do: ["lib", "test/example_domain", "test/helpers"]
+  defp elixirc_paths(:test), do: ["lib", "test/aggregates", "test/example_domain", "test/helpers"]
   defp elixirc_paths(_), do: ["lib", "test/helpers"]
 
   defp deps do


Test patch:
diff --git a/test/aggregates/multi_bank_account.ex b/test/aggregates/multi_bank_account.ex
new file mode 100644
index 0000000..5a70a90
--- /dev/null
+++ b/test/aggregates/multi_bank_account.ex
@@ -0,0 +1,75 @@
+defmodule Commanded.Aggregate.Multi.BankAccount do
+  defstruct [
+    account_number: nil,
+    balance: 0,
+    state: nil,
+  ]
+
+  alias Commanded.Aggregate.Multi
+  alias Commanded.Aggregate.Multi.BankAccount
+
+  defmodule Commands do
+    defmodule OpenAccount,        do: defstruct [:account_number, :initial_balance]
+    defmodule WithdrawMoney,      do: defstruct [:account_number, :transfer_uuid, :amount]
+  end
+
+  defmodule Events do
+    defmodule BankAccountOpened,  do: defstruct [:account_number, :balance]
+    defmodule MoneyWithdrawn,     do: defstruct [:account_number, :transfer_uuid, :amount, :balance]
+  end
+
+  alias Commands.{OpenAccount,WithdrawMoney}
+  alias Events.{BankAccountOpened,MoneyWithdrawn}
+
+  def execute(
+    %BankAccount{state: nil},
+    %OpenAccount{account_number: account_number, initial_balance: initial_balance})
+    when is_number(initial_balance) and initial_balance > 0
+  do
+    %BankAccountOpened{account_number: account_number, balance: initial_balance}
+  end
+
+  def execute(
+    %BankAccount{state: :active} = account,
+    %WithdrawMoney{amount: amount})
+    when is_number(amount) and amount > 0
+  do
+    account
+    |> Multi.new()
+    |> Multi.execute(&withdraw_money(&1, amount))
+    |> Multi.execute(&check_balance/1)
+  end
+
+  # state mutatators
+
+  def apply(
+    %BankAccount{} = state,
+    %BankAccountOpened{account_number: account_number, balance: balance})
+  do
+    %BankAccount{state |
+      account_number: account_number,
+      balance: balance,
+      state: :active,
+    }
+  end
+
+  def apply(%BankAccount{} = state, %MoneyWithdrawn{balance: balance}),
+    do: %BankAccount{state | balance: balance}
+
+  # private helpers
+
+  defp withdraw_money(%BankAccount{account_number: account_number, balance: balance}, amount) do
+    %MoneyWithdrawn{
+      account_number: account_number,
+      amount: amount,
+      balance: balance - amount
+    }
+  end
+
+  defp check_balance(%BankAccount{balance: balance})
+    when balance < 0
+  do
+    {:error, :insufficient_funds_available}
+  end
+  defp check_balance(%BankAccount{}), do: []
+end
diff --git a/test/aggregates/multi_test.exs b/test/aggregates/multi_test.exs
new file mode 100644
index 0000000..7b7e7c8
--- /dev/null
+++ b/test/aggregates/multi_test.exs
@@ -0,0 +1,75 @@
+defmodule Commanded.Aggregate.MultiTest do
+  use Commanded.StorageCase
+
+  import Commanded.Enumerable
+
+  alias Commanded.EventStore
+  alias Commanded.Aggregate.Multi
+  alias Commanded.Aggregate.Multi.BankAccount
+  alias Commanded.Aggregate.Multi.BankAccount.Commands.{OpenAccount,WithdrawMoney}
+  alias Commanded.Aggregate.Multi.BankAccount.Events.{BankAccountOpened,MoneyWithdrawn}
+
+  defmodule MultiBankRouter do
+    use Commanded.Commands.Router
+
+    alias Commanded.Aggregate.Multi.BankAccount
+    alias Commanded.Aggregate.Multi.BankAccount.Commands.{OpenAccount,WithdrawMoney}
+
+    dispatch [OpenAccount,WithdrawMoney],
+      to: BankAccount, identity: :account_number
+  end
+
+  test "should return `Commanded.Aggregate.Multi` from command" do
+    account_number = UUID.uuid4()
+    account = BankAccount.apply(%BankAccount{}, %BankAccountOpened{account_number: account_number, balance: 1_000})
+
+    assert %Multi{} = multi = BankAccount.execute(account, %WithdrawMoney{account_number: account_number, amount: 100})
+
+    assert {account, events} = Multi.run(multi)
+
+    assert account == %BankAccount{
+      account_number: account_number,
+      balance: 900,
+      state: :active,
+    }
+    assert events == [
+      %MoneyWithdrawn{account_number: account_number, amount: 100, balance: 900},
+    ]
+  end
+
+  test "should return errors encountered by `Commanded.Aggregate.Multi`" do
+    account_number = UUID.uuid4()
+    account = BankAccount.apply(%BankAccount{}, %BankAccountOpened{account_number: account_number, balance: 1_000})
+
+    assert %Multi{} = multi = BankAccount.execute(account, %WithdrawMoney{account_number: account_number, amount: 1_100})
+
+    assert {:error, :insufficient_funds_available} = Multi.run(multi)
+  end
+
+  test "should execute command using `Commanded.Aggregate.Multi` and return events" do
+    account_number = UUID.uuid4()
+
+    assert :ok = MultiBankRouter.dispatch(%OpenAccount{account_number: account_number, initial_balance: 1_000})
+    assert :ok = MultiBankRouter.dispatch(%WithdrawMoney{account_number: account_number, amount: 250})
+
+    recorded_events = EventStore.stream_forward(account_number, 0) |> Enum.to_list()
+
+    assert pluck(recorded_events, :data) == [
+      %BankAccountOpened{account_number: account_number, balance: 1_000},
+      %MoneyWithdrawn{account_number: account_number, amount: 250, balance: 750},
+    ]
+  end
+
+  test "should execute command using `Commanded.Aggregate.Multi` and return any error" do
+    account_number = UUID.uuid4()
+
+    assert :ok = MultiBankRouter.dispatch(%OpenAccount{account_number: account_number, initial_balance: 1_000})
+    assert {:error, :insufficient_funds_available} = MultiBankRouter.dispatch(%WithdrawMoney{account_number: account_number, amount: 1_100})
+
+    recorded_events = EventStore.stream_forward(account_number, 0) |> Enum.to_list()
+
+    assert pluck(recorded_events, :data) == [
+      %BankAccountOpened{account_number: account_number, balance: 1_000},
+    ]
+  end
+end