feat: add state module

Author: yordis

guides/explanations/fork-differences.md

@@ -243,3 +243,46 @@ end
 - Visualize event handler execution in your tracing backend
 - Correlate event processing with command dispatch using span links
 - Configurable span relationships (`:link`, `:child`, `:none`)
+
+### **Separate State Module for Aggregates**
+[PR #49](https://github.com/straw-hat-team/commanded/pull/49)
+
+**Changes:**
+- Added `state` option to command router's `dispatch` macro
+- Allows specifying a separate module for aggregate state struct
+- Useful for protobuf-generated modules or external struct definitions
+
+**Usage:**
+```elixir
+defmodule MyRouter do
+  use Commanded.Commands.Router
+
+  # BankAccount has execute/2 and apply/2 functions
+  # BankAccountState defines the state struct (e.g., protobuf-generated)
+  dispatch [OpenAccount, DepositMoney],
+    to: BankAccount,
+    state: BankAccountState,
+    identity: :account_number
+end
+
+# State module (could be protobuf-generated)
+defmodule BankAccountState do
+  defstruct [:account_number, :balance, :status]
+end
+
+# Aggregate module (behavior only, no struct)
+defmodule BankAccount do
+  def execute(%BankAccountState{status: nil}, %OpenAccount{} = cmd) do
+    %AccountOpened{account_number: cmd.account_number}
+  end
+
+  def apply(%BankAccountState{} = state, %AccountOpened{} = event) do
+    %BankAccountState{state | account_number: event.account_number, status: :open}
+  end
+end
+```
+
+**Benefits:**
+- Decouples aggregate behavior from state representation
+- Enables use of protobuf-generated structs as aggregate state
+- Backwards compatible - if `state` option is omitted, aggregate module is used for both behavior and state (existing behavior)

lib/application.ex

@@ -214,18 +214,26 @@ defmodule Commanded.Application do
       Retrieving aggregate state is done by calling to the opened aggregate,
       or querying the event store for an optional state snapshot
       and then replaying the aggregate's event stream.
+
+      ## Options
+
+        - `:timeout` - timeout in milliseconds (default: 5000)
+        - `:state_module` - module to use for initial state struct when rebuilding
+          from events. Defaults to `aggregate_module`. Use this when the aggregate
+          uses a separate state module.
+
       """
       @spec aggregate_state(
               aggregate_module :: module(),
               aggregate_uuid :: Aggregate.uuid(),
-              timeout :: integer
+              timeout_or_opts :: integer | Keyword.t()
             ) :: Aggregate.state()
-      def aggregate_state(aggregate_module, aggregate_uuid, timeout \\ 5000) do
+      def aggregate_state(aggregate_module, aggregate_uuid, timeout_or_opts \\ 5000) do
         Aggregate.aggregate_state(
           __MODULE__,
           aggregate_module,
           aggregate_uuid,
-          timeout
+          timeout_or_opts
         )
       end
 

lib/commanded.ex

@@ -38,19 +38,27 @@ defmodule Commanded do
   Retrieving aggregate state is done by calling to the opened aggregate,
   or querying the event store for an optional state snapshot
   and then replaying the aggregate's event stream.
+
+  ## Options
+
+    - `:timeout` - timeout in milliseconds (default: 5000)
+    - `:state_module` - module to use for initial state struct when rebuilding
+      from events. Defaults to `aggregate_module`. Use this when the aggregate
+      uses a separate state module.
+
   """
   @spec aggregate_state(
           application :: Commanded.Application.t(),
           aggregate_module :: module(),
           aggregate_uuid :: Aggregate.uuid(),
-          timeout :: integer
+          timeout_or_opts :: integer | Keyword.t()
         ) :: Aggregate.state()
-  def aggregate_state(application, aggregate_module, aggregate_uuid, timeout \\ 5_000) do
+  def aggregate_state(application, aggregate_module, aggregate_uuid, timeout_or_opts \\ 5_000) do
     Aggregate.aggregate_state(
       application,
       aggregate_module,
       aggregate_uuid,
-      timeout
+      timeout_or_opts
     )
   end
 end

lib/commanded/aggregates/aggregate.ex

@@ -147,6 +147,7 @@ defmodule Commanded.Aggregates.Aggregate do
   defstruct [
     :application,
     :aggregate_module,
+    :state_module,
     :aggregate_uuid,
     :aggregate_state,
     :snapshotting,
@@ -160,6 +161,7 @@ defmodule Commanded.Aggregates.Aggregate do
 
     aggregate_module = Keyword.fetch!(aggregate_opts, :aggregate_module)
     aggregate_uuid = Keyword.fetch!(aggregate_opts, :aggregate_uuid)
+    state_module = Keyword.get(aggregate_opts, :state_module) || aggregate_module
 
     unless is_atom(aggregate_module),
       do: raise(ArgumentError, message: "aggregate module must be an atom")
@@ -174,6 +176,7 @@ defmodule Commanded.Aggregates.Aggregate do
     state = %Aggregate{
       application: application,
       aggregate_module: aggregate_module,
+      state_module: state_module,
       aggregate_uuid: aggregate_uuid,
       snapshotting: Snapshotting.new(application, aggregate_uuid, snapshot_options)
     }
@@ -230,7 +233,16 @@ defmodule Commanded.Aggregates.Aggregate do
   end
 
   @doc false
-  def aggregate_state(application, aggregate_module, aggregate_uuid, timeout \\ 5_000) do
+  def aggregate_state(application, aggregate_module, aggregate_uuid, timeout_or_opts \\ 5_000)
+
+  def aggregate_state(application, aggregate_module, aggregate_uuid, timeout)
+      when is_integer(timeout) or timeout == :infinity do
+    aggregate_state(application, aggregate_module, aggregate_uuid, timeout: timeout)
+  end
+
+  def aggregate_state(application, aggregate_module, aggregate_uuid, opts) when is_list(opts) do
+    timeout = Keyword.get(opts, :timeout, 5_000)
+    state_module = Keyword.get(opts, :state_module, aggregate_module)
     name = via_name(application, aggregate_module, aggregate_uuid)
 
     try do
@@ -249,6 +261,7 @@ defmodule Commanded.Aggregates.Aggregate do
             %Aggregate{
               application: application,
               aggregate_module: aggregate_module,
+              state_module: state_module,
               aggregate_uuid: aggregate_uuid,
               snapshotting: Snapshotting.new(application, aggregate_uuid, snapshot_options)
             }

lib/commanded/aggregates/aggregate_state_builder.ex

@@ -40,9 +40,12 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do
   If the snapshot exists, fetch any subsequent events to rebuild its state.
   Otherwise start with the aggregate struct and stream all existing events for
   the aggregate from the event store to rebuild its state from those events.
+
+  The `state_module` field determines which module is used to create the initial
+  empty state struct.
   """
   def populate(%Aggregate{} = state) do
-    %Aggregate{aggregate_module: aggregate_module, snapshotting: snapshotting} = state
+    %Aggregate{state_module: state_module, snapshotting: snapshotting} = state
 
     aggregate =
       case Snapshotting.read_snapshot(snapshotting) do
@@ -55,7 +58,7 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do
 
         {:error, _error} ->
           # No snapshot present, or exists but for outdated state, so use initial empty state
-          %Aggregate{state | aggregate_version: 0, aggregate_state: struct(aggregate_module)}
+          %Aggregate{state | aggregate_version: 0, aggregate_state: struct(state_module)}
       end
 
     rebuild_from_events(aggregate)

lib/commanded/aggregates/supervisor.ex

@@ -23,8 +23,16 @@ defmodule Commanded.Aggregates.Supervisor do
 
   Returns `{:ok, aggregate_uuid}` when a process is successfully started, or is
   already running.
+
+  ## Options
+
+    - `:state_module` - optional module to use for the aggregate's state struct.
+      If not provided, defaults to the aggregate module itself.
+
   """
-  def open_aggregate(application, aggregate_module, aggregate_uuid)
+  def open_aggregate(application, aggregate_module, aggregate_uuid, opts \\ [])
+
+  def open_aggregate(application, aggregate_module, aggregate_uuid, opts)
       when is_atom(application) and is_atom(aggregate_module) and is_binary(aggregate_uuid) do
     Logger.debug(fn ->
       "Locating aggregate process for `#{inspect(aggregate_module)}` with UUID " <>
@@ -34,10 +42,13 @@ defmodule Commanded.Aggregates.Supervisor do
     supervisor_name = Module.concat([application, __MODULE__])
     aggregate_name = Aggregate.name(application, aggregate_module, aggregate_uuid)
 
+    state_module = Keyword.get(opts, :state_module)
+
     args = [
       application: application,
       aggregate_module: aggregate_module,
-      aggregate_uuid: aggregate_uuid
+      aggregate_uuid: aggregate_uuid,
+      state_module: state_module
     ]
 
     case Registration.start_child(application, aggregate_name, supervisor_name, {Aggregate, args}) do
@@ -55,7 +66,7 @@ defmodule Commanded.Aggregates.Supervisor do
     end
   end
 
-  def open_aggregate(_application, _aggregate_module, aggregate_uuid),
+  def open_aggregate(_application, _aggregate_module, aggregate_uuid, _opts),
     do: {:error, {:unsupported_aggregate_identity_type, aggregate_uuid}}
 
   def init(args) do

lib/commanded/commands/dispatcher.ex

@@ -24,6 +24,7 @@ defmodule Commanded.Commands.Dispatcher do
       :handler_function,
       :handler_before_execute,
       :aggregate_module,
+      :state_module,
       :identity,
       :identity_prefix,
       :timeout,
@@ -83,13 +84,16 @@ defmodule Commanded.Commands.Dispatcher do
   @dialyzer {:nowarn_function, execute: 3}
   defp execute(%Pipeline{} = pipeline, %Payload{} = payload, %ExecutionContext{} = context) do
     %Pipeline{application: application, assigns: %{aggregate_uuid: aggregate_uuid}} = pipeline
-    %Payload{aggregate_module: aggregate_module, timeout: timeout} = payload
+
+    %Payload{aggregate_module: aggregate_module, state_module: state_module, timeout: timeout} =
+      payload
 
     {:ok, ^aggregate_uuid} =
       Commanded.Aggregates.Supervisor.open_aggregate(
         application,
         aggregate_module,
-        aggregate_uuid
+        aggregate_uuid,
+        state_module: state_module
       )
 
     task_dispatcher_name = Module.concat([application, Commanded.Commands.TaskDispatcher])

lib/commanded/commands/router.ex

@@ -73,6 +73,34 @@ defmodule Commanded.Commands.Router do
         dispatch OpenAccount, to: BankAccount, function: :open_account, identity: :account_number
       end
 
+  ## Separate state module
+
+  By default, the aggregate module is expected to define a struct that represents
+  the aggregate's state. However, you can specify a separate module to use for the
+  state struct using the `state` option. This is useful when you want to use a
+  different struct (such as a protobuf-generated module) as the aggregate's state.
+
+  ### Example
+
+      defmodule BankRouter do
+        use Commanded.Commands.Router
+
+        # BankAccount module has execute/2 and apply/2 functions
+        # BankAccountState module defines the state struct
+        dispatch OpenAccount,
+          to: BankAccount,
+          state: BankAccountState,
+          identity: :account_number
+      end
+
+  When using a separate state module:
+
+    - The `state` module must define a struct (i.e., use `defstruct`)
+    - The aggregate module's `execute/2` and `apply/2` functions receive and
+      return the state module's struct
+    - If `state` is not specified, the aggregate module is used for both
+      behavior and state (the default behavior)
+
   ## Define aggregate identity
 
   You can define the identity field for an aggregate once using the `identify` macro.
@@ -506,6 +534,7 @@ defmodule Commanded.Commands.Router do
 
       for {command_module, command_opts} <- @registered_commands do
         @aggregate Keyword.fetch!(command_opts, :aggregate)
+        @state Keyword.get(command_opts, :state, @aggregate)
         @handler Keyword.fetch!(command_opts, :to)
         @function Keyword.fetch!(command_opts, :function)
         @before_execute Keyword.get(command_opts, :before_execute)
@@ -576,6 +605,7 @@ defmodule Commanded.Commands.Router do
             handler_function: @function,
             handler_before_execute: @before_execute,
             aggregate_module: @aggregate,
+            state_module: @state,
             identity: identity,
             identity_prefix: identity_prefix,
             returning: returning,
@@ -659,6 +689,7 @@ defmodule Commanded.Commands.Router do
     :function,
     :before_execute,
     :aggregate,
+    :state,
     :identity,
     :identity_prefix,
     :timeout,

test/commands/routing_commands_test.exs

@@ -226,7 +226,7 @@ defmodule Commanded.Commands.RoutingCommandsTest do
     assert_raise RuntimeError,
                  """
                  unexpected dispatch parameter "id"
-                 available params are: to, function, before_execute, aggregate, identity, identity_prefix, timeout, lifespan, consistency
+                 available params are: to, function, before_execute, aggregate, state, identity, identity_prefix, timeout, lifespan, consistency
                  """,
                  fn ->
                    Code.eval_string("""