diff --git a/config/test.exs b/config/test.exs index ec55c6ca..87e75c98 100644 --- a/config/test.exs +++ b/config/test.exs @@ -38,7 +38,7 @@ config :commanded, TestEventStore, username: "postgres", password: "postgres", database: "eventstore_test", - hostname: "localhost", + hostname: System.get_env("PG_HOST", "localhost"), pool_size: 5, pool_overflow: 0 @@ -49,5 +49,5 @@ config :commanded, Commanded.Projections.Repo, database: "commanded_projections_test", username: "postgres", password: "postgres", - hostname: "localhost", + hostname: System.get_env("PG_HOST", "localhost"), pool: Ecto.Adapters.SQL.Sandbox diff --git a/guides/explanations/fork-differences.md b/guides/explanations/fork-differences.md index b2011318..25d2a3ba 100644 --- a/guides/explanations/fork-differences.md +++ b/guides/explanations/fork-differences.md @@ -284,3 +284,81 @@ end **Benefits:** - All event handlers (projectors, sagas, notification handlers) get processing latency visibility for free - Enables SLA dashboards and alerting without any application-level code + +### **Custom Initial State for Aggregates** +[PR #49](https://github.com/straw-hat-team/commanded/pull/49) + +**Changes:** +- Added `initial_state` option to command router's `dispatch` macro +- Allows specifying a module that implements `initial_state/0` callback +- Useful when using protobuf-generated messages as aggregate state + +**Usage:** + +```elixir +# State module with initial_state/0 callback +defmodule BankAccountState do + defstruct [:account_number, :balance, status: :uninitialized] + + def initial_state, do: %__MODULE__{} +end + +# Aggregate module (behavior only, no struct) +defmodule BankAccount do + def execute(%BankAccountState{status: :uninitialized}, %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 + +defmodule MyRouter do + use Commanded.Commands.Router + + # Calls BankAccountState.initial_state/0 to create initial state + dispatch [OpenAccount, DepositMoney], + to: BankAccount, + initial_state: BankAccountState, + identity: :account_number +end +``` + +**Benefits:** +- Decouples aggregate behavior from state representation +- State module controls its own initialization (like `to:` pattern) +- Enables use of protobuf-generated messages as aggregate state instead of being forced to use Elixir structs +- Backwards compatible - if `initial_state` is omitted, `struct(AggregateModule)` is used + +**Rationale:** + +In the upstream Commanded, the aggregate module serves dual purposes: it defines both the state struct and the behavior (`execute/2` and `apply/2` functions). This coupling becomes problematic when you want to use protobuf-generated modules for state. + +Protobuf modules are code-generated and shouldn't be manually modified—any changes would be overwritten on regeneration. To add aggregate behavior to a protobuf struct, you'd need to write custom protobuf extensions or use workarounds, adding complexity to your build pipeline. + +By separating the aggregate (behavior) from the state (data structure), you can: + +```elixir +# Generated by protobuf - don't modify +defmodule MyApp.Proto.BankAccountState do + use Protobuf, syntax: :proto3 + # ... generated fields ... +end + +# Your code - aggregate behavior with initial_state returning the protobuf struct +defmodule MyApp.BankAccount do + alias MyApp.Proto.BankAccountState + + def initial_state, do: %BankAccountState{status: :STATUS_UNINITIALIZED} + + def execute(%BankAccountState{} = state, %OpenAccount{} = cmd), do: ... + def apply(%BankAccountState{} = state, %AccountOpened{} = event), do: ... +end +``` + +**Why a callback instead of `struct/1`?** + +The main reason is to use protobuf-generated messages as aggregate state instead of being forced to use Elixir structs. Protobuf modules are code-generated and have their own initialization semantics. The `initial_state/0` callback lets you return whatever your state module needs—including protobuf messages—rather than relying on `struct(AggregateModule)`, which only works for Elixir structs. + +This separation follows the principle that data representation and business logic are distinct concerns that benefit from being in separate modules. It also aligns more closely with the [Functional Decider Pattern](https://thinkbeforecoding.com/post/2021/12/17/functional-event-sourcing-decider), where the aggregate is a set of pure functions (`execute`, `apply`, `initial_state`) operating on state, rather than a stateful object that owns its data structure. diff --git a/lib/application.ex b/lib/application.ex index 2f62f41f..886241e6 100644 --- a/lib/application.ex +++ b/lib/application.ex @@ -214,18 +214,25 @@ 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) + - `:initial_state` - module that implements `initial_state/0` callback. + Used when rebuilding state from events. Defaults to `aggregate_module`. + """ @spec aggregate_state( aggregate_module :: module(), aggregate_uuid :: Aggregate.uuid(), - timeout :: integer + timeout_or_opts :: timeout() | Aggregate.aggregate_state_opts() ) :: 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 diff --git a/lib/commanded.ex b/lib/commanded.ex index 76e5bc30..dece94d7 100644 --- a/lib/commanded.ex +++ b/lib/commanded.ex @@ -38,19 +38,26 @@ 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) + - `:initial_state` - module that implements `initial_state/0` callback. + Used when rebuilding state from events. Defaults to `aggregate_module`. + """ @spec aggregate_state( application :: Commanded.Application.t(), aggregate_module :: module(), aggregate_uuid :: Aggregate.uuid(), - timeout :: integer + timeout_or_opts :: timeout() | Aggregate.aggregate_state_opts() ) :: 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 diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex index ccfe9010..813c19e8 100644 --- a/lib/commanded/aggregates/aggregate.ex +++ b/lib/commanded/aggregates/aggregate.ex @@ -177,6 +177,9 @@ defmodule Commanded.Aggregates.Aggregate do @type return_event :: struct() | list(struct()) | {:ok, struct()} | {:ok, list(struct())} @type no_return_event :: :ok | {:ok, []} | nil | [] + @type aggregate_state_opt :: {:timeout, timeout()} | {:initial_state, module()} + @type aggregate_state_opts :: [aggregate_state_opt()] + @doc """ Optionally execute a command against the aggregate. Returns either no event, one event, a list of events, or an error tuple. @@ -194,6 +197,7 @@ defmodule Commanded.Aggregates.Aggregate do defstruct [ :application, :aggregate_module, + :initial_state, :aggregate_uuid, :aggregate_state, :snapshotting, @@ -208,6 +212,8 @@ defmodule Commanded.Aggregates.Aggregate do aggregate_module = Keyword.fetch!(aggregate_opts, :aggregate_module) aggregate_uuid = Keyword.fetch!(aggregate_opts, :aggregate_uuid) + initial_state = Keyword.get(aggregate_opts, :initial_state) + validate_initial_state_module!(initial_state) unless is_atom(aggregate_module), do: raise(ArgumentError, message: "aggregate module must be an atom") @@ -222,6 +228,7 @@ defmodule Commanded.Aggregates.Aggregate do state = %Aggregate{ application: application, aggregate_module: aggregate_module, + initial_state: initial_state, aggregate_uuid: aggregate_uuid, snapshotting: Snapshotting.new(application, aggregate_uuid, snapshot_options) } @@ -278,7 +285,17 @@ 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) + initial_state = Keyword.get(opts, :initial_state) + validate_initial_state_module!(initial_state) name = via_name(application, aggregate_module, aggregate_uuid) try do @@ -297,6 +314,7 @@ defmodule Commanded.Aggregates.Aggregate do %Aggregate{ application: application, aggregate_module: aggregate_module, + initial_state: initial_state, aggregate_uuid: aggregate_uuid, snapshotting: Snapshotting.new(application, aggregate_uuid, snapshot_options) } @@ -308,6 +326,9 @@ defmodule Commanded.Aggregates.Aggregate do {:ok, result} -> result + {:exit, reason} -> + exit(reason) + nil -> exit({:timeout, {GenServer, :call, [name, :aggregate_state, timeout]}}) end @@ -320,6 +341,12 @@ defmodule Commanded.Aggregates.Aggregate do GenServer.call(name, :aggregate_version, timeout) end + @doc false + def initial_state_module(application, aggregate_module, aggregate_uuid, timeout \\ 5_000) do + name = via_name(application, aggregate_module, aggregate_uuid) + GenServer.call(name, :initial_state_module, timeout) + end + @doc false def take_snapshot(application, aggregate_module, aggregate_uuid, timeout \\ 5_000) do name = via_name(application, aggregate_module, aggregate_uuid) @@ -437,6 +464,14 @@ defmodule Commanded.Aggregates.Aggregate do reply_with_lifespan(aggregate_version, state) end + @doc false + @impl GenServer + def handle_call(:initial_state_module, _from, %Aggregate{} = state) do + %Aggregate{initial_state: initial_state} = state + + reply_with_lifespan(initial_state, state) + end + @doc false @impl GenServer def handle_info({:events, events}, %Aggregate{} = state) do @@ -813,4 +848,27 @@ defmodule Commanded.Aggregates.Aggregate do lifespan_timeout -> {:noreply, state, lifespan_timeout} end end + + defp validate_initial_state_module!(nil), do: :ok + + defp validate_initial_state_module!(initial_state) when is_atom(initial_state) do + case Code.ensure_compiled(initial_state) do + {:module, _} -> + if function_exported?(initial_state, :initial_state, 0) do + :ok + else + raise ArgumentError, + "initial_state module #{inspect(initial_state)} must export initial_state/0 function." + end + + {:error, reason} -> + raise ArgumentError, + "initial_state module #{inspect(initial_state)} could not be loaded: #{inspect(reason)}" + end + end + + defp validate_initial_state_module!(initial_state) do + raise ArgumentError, + "initial_state must be a module but got: #{inspect(initial_state)}" + end end diff --git a/lib/commanded/aggregates/aggregate_state_builder.ex b/lib/commanded/aggregates/aggregate_state_builder.ex index a267abc3..7f630c2d 100644 --- a/lib/commanded/aggregates/aggregate_state_builder.ex +++ b/lib/commanded/aggregates/aggregate_state_builder.ex @@ -71,9 +71,13 @@ 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 initial state is determined by the `initial_state` field: + - If a module is provided, `initial_state.initial_state()` is called + - Otherwise, falls back to `struct(aggregate_module)` """ def populate(%Aggregate{} = state) do - %Aggregate{aggregate_module: aggregate_module, snapshotting: snapshotting} = state + %Aggregate{snapshotting: snapshotting} = state {aggregate, snapshot_used, snapshot_source_version} = case Snapshotting.read_snapshot(snapshotting) do @@ -90,7 +94,7 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do agg = %Aggregate{ state | aggregate_version: 0, - aggregate_state: struct(aggregate_module) + aggregate_state: create_initial_state(state) } {agg, false, nil} @@ -102,6 +106,15 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do ) end + defp create_initial_state(%Aggregate{initial_state: nil, aggregate_module: aggregate_module}) do + struct(aggregate_module) + end + + defp create_initial_state(%Aggregate{initial_state: initial_state}) + when is_atom(initial_state) do + initial_state.initial_state() + end + @doc """ Load events from the event store, in batches, to rebuild the aggregate state. diff --git a/lib/commanded/aggregates/supervisor.ex b/lib/commanded/aggregates/supervisor.ex index 66431c48..f4a42162 100644 --- a/lib/commanded/aggregates/supervisor.ex +++ b/lib/commanded/aggregates/supervisor.ex @@ -23,8 +23,24 @@ defmodule Commanded.Aggregates.Supervisor do Returns `{:ok, aggregate_uuid}` when a process is successfully started, or is already running. + + ## Options + + - `:initial_state` - optional module that implements `initial_state/0` callback. + If not provided, defaults to `aggregate_module`. + """ - def open_aggregate(application, aggregate_module, aggregate_uuid) + @type open_aggregate_opt :: {:initial_state, module()} | {:timeout, timeout()} + + @spec open_aggregate( + application :: module(), + aggregate_module :: module(), + aggregate_uuid :: String.t(), + opts :: [open_aggregate_opt()] + ) :: {:ok, String.t()} | {:error, term()} + 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,30 +50,102 @@ defmodule Commanded.Aggregates.Supervisor do supervisor_name = Module.concat([application, __MODULE__]) aggregate_name = Aggregate.name(application, aggregate_module, aggregate_uuid) + initial_state = Keyword.get(opts, :initial_state) + timeout = Keyword.get(opts, :timeout, 5_000) + args = [ application: application, aggregate_module: aggregate_module, - aggregate_uuid: aggregate_uuid + aggregate_uuid: aggregate_uuid, + initial_state: initial_state ] case Registration.start_child(application, aggregate_name, supervisor_name, {Aggregate, args}) do {:ok, _pid} -> - {:ok, aggregate_uuid} + ensure_initial_state_or_return( + application, + aggregate_module, + aggregate_uuid, + initial_state, + timeout + ) {:ok, _pid, _info} -> - {:ok, aggregate_uuid} + ensure_initial_state_or_return( + application, + aggregate_module, + aggregate_uuid, + initial_state, + timeout + ) {:error, {:already_started, _pid}} -> - {:ok, aggregate_uuid} + ensure_initial_state_or_return( + application, + aggregate_module, + aggregate_uuid, + initial_state, + timeout + ) reply -> reply 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}} + defp ensure_initial_state_or_return( + application, + aggregate_module, + aggregate_uuid, + initial_state, + timeout + ) do + case ensure_initial_state_consistency( + application, + aggregate_module, + aggregate_uuid, + initial_state, + timeout + ) do + :ok -> {:ok, aggregate_uuid} + {:error, _reason} = error -> error + end + end + + defp ensure_initial_state_consistency( + application, + aggregate_module, + aggregate_uuid, + initial_state, + timeout + ) do + running_initial_state = + Aggregate.initial_state_module(application, aggregate_module, aggregate_uuid, timeout) + + if running_initial_state == initial_state do + :ok + else + {:error, + {:conflicting_initial_state, + %{ + aggregate_module: aggregate_module, + aggregate_uuid: aggregate_uuid, + requested_initial_state: initial_state, + running_initial_state: running_initial_state + }}} + end + catch + :exit, {reason, _context} when reason in [:normal, :noproc] -> + # If aggregate exited between registration and validation, let caller retry. + :ok + + :exit, reason -> + {:error, {:cannot_validate_initial_state, reason}} + end + def init(args) do DynamicSupervisor.init(strategy: :one_for_one, extra_arguments: [args]) end diff --git a/lib/commanded/commands/dispatcher.ex b/lib/commanded/commands/dispatcher.ex index 411fa497..c7bf7030 100644 --- a/lib/commanded/commands/dispatcher.ex +++ b/lib/commanded/commands/dispatcher.ex @@ -24,6 +24,7 @@ defmodule Commanded.Commands.Dispatcher do :handler_function, :handler_before_execute, :aggregate_module, + :initial_state, :identity, :identity_prefix, :timeout, @@ -77,79 +78,88 @@ 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 - - {:ok, ^aggregate_uuid} = - Commanded.Aggregates.Supervisor.open_aggregate( - application, - aggregate_module, - aggregate_uuid - ) - - task_dispatcher_name = Module.concat([application, Commanded.Commands.TaskDispatcher]) - - task = - Task.Supervisor.async_nolink(task_dispatcher_name, Aggregate, :execute, [ - application, - aggregate_module, - aggregate_uuid, - context, - timeout - ]) - - result = - case Task.yield(task, timeout) || Task.shutdown(task) do - {:ok, result} -> - result - - {:exit, {:normal, :aggregate_stopped}} = result -> - result - - {:exit, {{:nodedown, _node_name}, {GenServer, :call, _}}} -> - {:error, :remote_node_down} - - {:exit, _reason} -> - {:error, :aggregate_execution_failed} - - nil -> - {:error, :aggregate_execution_timeout} - end - - case result do - {:ok, aggregate_version, events, aggregate_state} -> - pipeline - |> Pipeline.assign(:aggregate_version, aggregate_version) - |> Pipeline.assign(:events, events) - |> Pipeline.assign(:aggregate_state, aggregate_state) - |> after_dispatch(payload) - |> Pipeline.respond(:ok) - - {:ok, aggregate_version, events, aggregate_state, reply} -> - pipeline - |> Pipeline.assign(:aggregate_version, aggregate_version) - |> Pipeline.assign(:events, events) - |> Pipeline.assign(:aggregate_state, aggregate_state) - |> after_dispatch(payload) - |> Pipeline.respond({:ok, reply}) - - {:exit, {:normal, :aggregate_stopped}} -> - # Maybe retry command when aggregate process stopped by lifespan timeout - maybe_retry(pipeline, payload, context) - {:error, :remote_node_down} -> - # Maybe retry command when aggregate process not found on a remote node - maybe_retry(pipeline, payload, context) + %Payload{aggregate_module: aggregate_module, initial_state: initial_state, timeout: timeout} = + payload + + case Commanded.Aggregates.Supervisor.open_aggregate( + application, + aggregate_module, + aggregate_uuid, + initial_state: initial_state, + timeout: timeout + ) do + {:ok, ^aggregate_uuid} -> + task_dispatcher_name = Module.concat([application, Commanded.Commands.TaskDispatcher]) + + task = + Task.Supervisor.async_nolink(task_dispatcher_name, Aggregate, :execute, [ + application, + aggregate_module, + aggregate_uuid, + context, + timeout + ]) + + result = + case Task.yield(task, timeout) || Task.shutdown(task) do + {:ok, result} -> + result + + {:exit, {:normal, :aggregate_stopped}} = result -> + result + + {:exit, {{:nodedown, _node_name}, {GenServer, :call, _}}} -> + {:error, :remote_node_down} + + {:exit, _reason} -> + {:error, :aggregate_execution_failed} + + nil -> + {:error, :aggregate_execution_timeout} + end + + case result do + {:ok, aggregate_version, events, aggregate_state} -> + pipeline + |> Pipeline.assign(:aggregate_version, aggregate_version) + |> Pipeline.assign(:events, events) + |> Pipeline.assign(:aggregate_state, aggregate_state) + |> after_dispatch(payload) + |> Pipeline.respond(:ok) + + {:ok, aggregate_version, events, aggregate_state, reply} -> + pipeline + |> Pipeline.assign(:aggregate_version, aggregate_version) + |> Pipeline.assign(:events, events) + |> Pipeline.assign(:aggregate_state, aggregate_state) + |> after_dispatch(payload) + |> Pipeline.respond({:ok, reply}) + + {:exit, {:normal, :aggregate_stopped}} -> + # Maybe retry command when aggregate process stopped by lifespan timeout + maybe_retry(pipeline, payload, context) + + {:error, :remote_node_down} -> + # Maybe retry command when aggregate process not found on a remote node + maybe_retry(pipeline, payload, context) + + {:error, error} -> + pipeline + |> Pipeline.respond({:error, error}) + |> after_failure(payload) + + {:error, error, reason} -> + pipeline + |> Pipeline.assign(:error_reason, reason) + |> Pipeline.respond({:error, error}) + |> after_failure(payload) + end {:error, error} -> pipeline |> Pipeline.respond({:error, error}) |> after_failure(payload) - - {:error, error, reason} -> - pipeline - |> Pipeline.assign(:error_reason, reason) - |> Pipeline.respond({:error, error}) - |> after_failure(payload) end end diff --git a/lib/commanded/commands/router.ex b/lib/commanded/commands/router.ex index 36916321..948dc73a 100644 --- a/lib/commanded/commands/router.ex +++ b/lib/commanded/commands/router.ex @@ -1,3 +1,4 @@ +# credo:disable-for-this-file Credo.Check.Refactor.LongQuoteBlocks defmodule Commanded.Commands.Router do @moduledoc """ Command routing macro to allow configuration of each command to its command handler. @@ -73,6 +74,42 @@ defmodule Commanded.Commands.Router do dispatch OpenAccount, to: BankAccount, function: :open_account, identity: :account_number end + ## Custom initial state + + By default, the aggregate module is expected to define a struct that represents + the aggregate's state, and `struct(AggregateModule)` is used to create the initial + state. You can customize the initial state using the `initial_state` option, + which specifies a module that implements an `initial_state/0` function. + + This is useful when: + - You want to use protobuf-generated messages as aggregate state + - You want to separate aggregate behavior from state representation + + ### Example + + defmodule BankAccountState do + defstruct [:account_number, :balance, status: :uninitialized] + + def initial_state, do: %__MODULE__{} + end + + defmodule BankRouter do + use Commanded.Commands.Router + + dispatch OpenAccount, + to: BankAccount, + initial_state: BankAccountState, + identity: :account_number + end + + When using a custom initial state: + + - `initial_state.initial_state()` is called to create the initial state + - The aggregate module's `execute/2` and `apply/2` functions receive and + return the state struct type + - If `initial_state` is not specified, `struct(AggregateModule)` is used + (the default behavior) + ## Define aggregate identity You can define the identity field for an aggregate once using the `identify` macro. @@ -504,8 +541,36 @@ defmodule Commanded.Commands.Router do [middleware | acc] end) + @registered_commands + |> Enum.group_by( + fn {_command_module, opts} -> Keyword.fetch!(opts, :aggregate) end, + fn {command_module, opts} -> {command_module, Keyword.get(opts, :initial_state)} end + ) + |> Enum.each(fn {aggregate, commands_with_state} -> + commands_with_state + |> Enum.map(&elem(&1, 1)) + |> Enum.uniq() + |> case do + [_] -> + :ok + + _ -> + [{conflicting_command, conflicting_state} | _] = commands_with_state + [{_, existing_state} | _] = Enum.reverse(commands_with_state) + + raise ArgumentError, """ + aggregate #{inspect(aggregate)} must use the same `:initial_state` option across all dispatched commands. + + Existing `:initial_state`: #{inspect(existing_state)} + Conflicting command: #{inspect(conflicting_command)} + Conflicting `:initial_state`: #{inspect(conflicting_state)} + """ + end + end) + for {command_module, command_opts} <- @registered_commands do @aggregate Keyword.fetch!(command_opts, :aggregate) + @initial_state Keyword.get(command_opts, :initial_state) @handler Keyword.fetch!(command_opts, :to) @function Keyword.fetch!(command_opts, :function) @before_execute Keyword.get(command_opts, :before_execute) @@ -513,6 +578,30 @@ defmodule Commanded.Commands.Router do @identity Keyword.get(command_opts, :identity) @identity_prefix Keyword.get(command_opts, :identity_prefix) + if @initial_state do + case Code.ensure_compiled(@initial_state) do + {:module, _} -> + # credo:disable-for-next-line Credo.Check.Refactor.Nesting + unless function_exported?(@initial_state, :initial_state, 0) do + raise ArgumentError, """ + initial_state module #{inspect(@initial_state)} must export initial_state/0 function. + + Example: + + defmodule #{inspect(@initial_state)} do + defstruct [:field] + + def initial_state, do: %__MODULE__{} + end + """ + end + + {:error, reason} -> + raise ArgumentError, + "initial_state module #{inspect(@initial_state)} could not be loaded: #{inspect(reason)}" + end + end + @command_module command_module @command_opts command_opts @@ -576,6 +665,7 @@ defmodule Commanded.Commands.Router do handler_function: @function, handler_before_execute: @before_execute, aggregate_module: @aggregate, + initial_state: @initial_state, identity: identity, identity_prefix: identity_prefix, returning: returning, @@ -659,6 +749,7 @@ defmodule Commanded.Commands.Router do :function, :before_execute, :aggregate, + :initial_state, :identity, :identity_prefix, :timeout, diff --git a/test/commands/routing_commands_test.exs b/test/commands/routing_commands_test.exs index a82cae5d..b165d6a3 100644 --- a/test/commands/routing_commands_test.exs +++ b/test/commands/routing_commands_test.exs @@ -227,7 +227,7 @@ defmodule Commanded.Commands.RoutingCommandsTest do assert_raise ArgumentError, """ 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, initial_state, identity, identity_prefix, timeout, lifespan, consistency """, fn -> Code.eval_string(""" @@ -258,6 +258,39 @@ defmodule Commanded.Commands.RoutingCommandsTest do end end + test "should reject mixed initial_state configuration for the same aggregate" do + assert_raise ArgumentError, + ~r/must use the same `:initial_state` option across all dispatched commands/, + fn -> + Code.eval_string(""" + alias Commanded.ExampleDomain.BankAccount + alias Commanded.ExampleDomain.BankAccount.Commands.{OpenAccount, DepositMoney} + alias Commanded.ExampleDomain.{OpenAccountHandler, DepositMoneyHandler} + + defmodule CustomInitialState do + defstruct [:dummy] + + def initial_state, do: %__MODULE__{} + end + + defmodule InvalidInitialStateRouter do + use Commanded.Commands.Router + + dispatch OpenAccount, + to: OpenAccountHandler, + aggregate: BankAccount, + initial_state: CustomInitialState, + identity: :account_number + + dispatch DepositMoney, + to: DepositMoneyHandler, + aggregate: BankAccount, + identity: :account_number + end + """) + end + end + defmodule MultiCommandRouter do use Commanded.Commands.Router diff --git a/test/commands/state_module_test.exs b/test/commands/state_module_test.exs new file mode 100644 index 00000000..3e8d6a22 --- /dev/null +++ b/test/commands/state_module_test.exs @@ -0,0 +1,406 @@ +defmodule Commanded.Commands.StateModuleTest do + use ExUnit.Case + + alias Commanded.Aggregates.{Aggregate, Supervisor} + alias Commanded.Commands.StateModuleAggregate + alias Commanded.Commands.StateModuleAggregate.{CreateCommand, UpdateCommand} + alias Commanded.Commands.StateModuleAggregate.CreatedEvent + alias Commanded.Commands.StateModuleAggregate.State + alias Commanded.Commands.StateModuleRouter + alias Commanded.{DefaultApp, EventStore} + alias Commanded.EventStore.SnapshotData + alias Commanded.UUID + + @dispatch_opts [application: DefaultApp] + + defp attach_snapshot_listener(application, aggregate_uuid) do + ref = make_ref() + subscriber = self() + + handler = fn _event, _measurements, meta, _config -> + if meta[:aggregate_uuid] == aggregate_uuid and meta[:application] == application do + send(subscriber, {:snapshot_complete, ref}) + end + end + + :telemetry.attach(ref, [:commanded, :aggregate, :snapshot, :stop], handler, []) + ref + end + + defp wait_for_snapshot(ref, timeout \\ 5_000) do + assert_receive {:snapshot_complete, ^ref}, timeout + :telemetry.detach(ref) + end + + setup do + start_supervised!(DefaultApp) + + :ok + end + + describe "routing with separate state module" do + test "fetching aggregate state when GenServer is down fails cryptically if initial_state is omitted" do + uuid = UUID.uuid4() + + # 1. Dispatch to create the stream and process + assert :ok = + StateModuleRouter.dispatch( + %CreateCommand{uuid: uuid, name: "Test"}, + @dispatch_opts + ) + + # 2. Shut down the aggregate so it MUST be rebuilt from events + :ok = Aggregate.shutdown(DefaultApp, StateModuleAggregate, uuid) + + # 3. This should throw a UndefinedFunctionError since the default is `struct(aggregate_module)` + # The task will crash and the exit reason will be returned by our case statement + # Under Task.shutdown with exit(reason) the error is deeply nested in the Task result + Process.flag(:trap_exit, true) + + catch_exit(Commanded.aggregate_state(DefaultApp, StateModuleAggregate, uuid)) + + assert_receive {:EXIT, _pid, {:undef, stacktrace}} + + assert [{Commanded.Commands.StateModuleAggregate, :__struct__, [], []} | _] = stacktrace + + # 4. This should work perfectly + state = + Commanded.aggregate_state(DefaultApp, StateModuleAggregate, uuid, initial_state: State) + + assert %State{uuid: ^uuid, name: "Test"} = state + end + + test "should dispatch command and use state module for initial state" do + uuid = UUID.uuid4() + + assert :ok = + StateModuleRouter.dispatch( + %CreateCommand{uuid: uuid, name: "Test"}, + @dispatch_opts + ) + + recorded_events = EventStore.stream_forward(DefaultApp, uuid, 0) |> Enum.to_list() + assert length(recorded_events) == 1 + + [event] = recorded_events + assert %CreatedEvent{uuid: ^uuid, name: "Test"} = event.data + end + + test "should return aggregate state with correct state module struct" do + uuid = UUID.uuid4() + + assert {:ok, %State{uuid: ^uuid, name: "Test"}} = + StateModuleRouter.dispatch( + %CreateCommand{uuid: uuid, name: "Test"}, + Keyword.merge(@dispatch_opts, returning: :aggregate_state) + ) + end + + test "should apply multiple events to state module struct" do + uuid = UUID.uuid4() + + assert :ok = + StateModuleRouter.dispatch( + %CreateCommand{uuid: uuid, name: "Initial"}, + @dispatch_opts + ) + + assert {:ok, %State{uuid: ^uuid, name: "Updated"}} = + StateModuleRouter.dispatch( + %UpdateCommand{uuid: uuid, name: "Updated"}, + Keyword.merge(@dispatch_opts, returning: :aggregate_state) + ) + + recorded_events = EventStore.stream_forward(DefaultApp, uuid, 0) |> Enum.to_list() + assert length(recorded_events) == 2 + end + + test "should return events from dispatch" do + uuid = UUID.uuid4() + + assert {:ok, [%CreatedEvent{uuid: ^uuid, name: "Test"}]} = + StateModuleRouter.dispatch( + %CreateCommand{uuid: uuid, name: "Test"}, + Keyword.merge(@dispatch_opts, returning: :events) + ) + end + + test "should return execution result with state module struct" do + uuid = UUID.uuid4() + + assert {:ok, result} = + StateModuleRouter.dispatch( + %CreateCommand{uuid: uuid, name: "Test"}, + Keyword.merge(@dispatch_opts, returning: :execution_result) + ) + + assert result.aggregate_uuid == uuid + assert result.aggregate_version == 1 + assert %State{uuid: ^uuid, name: "Test"} = result.aggregate_state + assert [%CreatedEvent{uuid: ^uuid, name: "Test"}] = result.events + end + end + + describe "snapshotting with separate state module" do + setup do + stop_supervised!(DefaultApp) + + start_supervised!( + {DefaultApp, + snapshotting: %{StateModuleAggregate => [snapshot_every: 2, snapshot_version: 1]}} + ) + + :ok + end + + test "should create snapshot with state module struct" do + uuid = UUID.uuid4() + + ref = attach_snapshot_listener(DefaultApp, uuid) + + assert :ok = + StateModuleRouter.dispatch( + %CreateCommand{uuid: uuid, name: "Initial"}, + @dispatch_opts + ) + + assert :ok = + StateModuleRouter.dispatch( + %UpdateCommand{uuid: uuid, name: "Updated"}, + @dispatch_opts + ) + + wait_for_snapshot(ref) + + assert {:ok, snapshot} = EventStore.read_snapshot(DefaultApp, uuid) + + assert %SnapshotData{ + source_uuid: ^uuid, + source_version: 2, + source_type: "Elixir.Commanded.Commands.StateModuleAggregate.State", + data: %State{uuid: ^uuid, name: "Updated"}, + metadata: %{"snapshot_module_version" => 1} + } = snapshot + end + + test "should restore state from snapshot after aggregate restart" do + uuid = UUID.uuid4() + + ref = attach_snapshot_listener(DefaultApp, uuid) + + assert :ok = + StateModuleRouter.dispatch( + %CreateCommand{uuid: uuid, name: "Initial"}, + @dispatch_opts + ) + + assert :ok = + StateModuleRouter.dispatch( + %UpdateCommand{uuid: uuid, name: "Snapshotted"}, + @dispatch_opts + ) + + wait_for_snapshot(ref) + + assert {:ok, _snapshot} = EventStore.read_snapshot(DefaultApp, uuid) + + :ok = Aggregate.shutdown(DefaultApp, StateModuleAggregate, uuid) + + {:ok, ^uuid} = + Supervisor.open_aggregate(DefaultApp, StateModuleAggregate, uuid, initial_state: State) + + state = Aggregate.aggregate_state(DefaultApp, StateModuleAggregate, uuid) + assert %State{uuid: ^uuid, name: "Snapshotted"} = state + assert Aggregate.aggregate_version(DefaultApp, StateModuleAggregate, uuid) == 2 + end + + test "should restore state from snapshot and apply newer events" do + uuid = UUID.uuid4() + + ref = attach_snapshot_listener(DefaultApp, uuid) + + assert :ok = + StateModuleRouter.dispatch( + %CreateCommand{uuid: uuid, name: "Initial"}, + @dispatch_opts + ) + + assert :ok = + StateModuleRouter.dispatch( + %UpdateCommand{uuid: uuid, name: "Snapshotted"}, + @dispatch_opts + ) + + wait_for_snapshot(ref) + + assert :ok = + StateModuleRouter.dispatch( + %UpdateCommand{uuid: uuid, name: "AfterSnapshot"}, + @dispatch_opts + ) + + assert {:ok, snapshot} = EventStore.read_snapshot(DefaultApp, uuid) + assert snapshot.source_version == 2 + + :ok = Aggregate.shutdown(DefaultApp, StateModuleAggregate, uuid) + + {:ok, ^uuid} = + Supervisor.open_aggregate(DefaultApp, StateModuleAggregate, uuid, initial_state: State) + + state = Aggregate.aggregate_state(DefaultApp, StateModuleAggregate, uuid) + assert %State{uuid: ^uuid, name: "AfterSnapshot"} = state + assert Aggregate.aggregate_version(DefaultApp, StateModuleAggregate, uuid) == 3 + end + + test "should use initial_state when no snapshot exists" do + uuid = UUID.uuid4() + + assert :ok = + StateModuleRouter.dispatch( + %CreateCommand{uuid: uuid, name: "Initial"}, + @dispatch_opts + ) + + assert {:error, :snapshot_not_found} = EventStore.read_snapshot(DefaultApp, uuid) + + :ok = Aggregate.shutdown(DefaultApp, StateModuleAggregate, uuid) + + {:ok, ^uuid} = + Supervisor.open_aggregate(DefaultApp, StateModuleAggregate, uuid, initial_state: State) + + state = Aggregate.aggregate_state(DefaultApp, StateModuleAggregate, uuid) + assert %State{uuid: ^uuid, name: "Initial"} = state + assert Aggregate.aggregate_version(DefaultApp, StateModuleAggregate, uuid) == 1 + end + end + + describe "backward compatibility" do + defmodule LegacyAggregate do + @derive Jason.Encoder + defstruct [:uuid, :name] + + defmodule Command do + @derive Jason.Encoder + defstruct [:uuid, :name] + end + + defmodule Event do + @derive Jason.Encoder + defstruct [:uuid, :name] + end + + def execute(%__MODULE__{}, %Command{uuid: uuid, name: name}) do + %Event{uuid: uuid, name: name} + end + + def apply(%__MODULE__{} = state, %Event{uuid: uuid, name: name}) do + %__MODULE__{state | uuid: uuid, name: name} + end + end + + defmodule LegacyRouter do + use Commanded.Commands.Router + + alias Commanded.Commands.StateModuleTest.LegacyAggregate + alias Commanded.Commands.StateModuleTest.LegacyAggregate.Command + + dispatch Command, + to: LegacyAggregate, + identity: :uuid + end + + test "should work without state option (backward compatible)" do + uuid = UUID.uuid4() + + assert {:ok, %LegacyAggregate{uuid: ^uuid, name: "Test"}} = + LegacyRouter.dispatch( + %LegacyAggregate.Command{uuid: uuid, name: "Test"}, + Keyword.merge(@dispatch_opts, returning: :aggregate_state) + ) + end + end + + describe "initial_state validation" do + defmodule InvalidStateModule do + @derive Jason.Encoder + defstruct [:uuid] + end + + test "should raise when aggregate_state receives a module without initial_state/0" do + uuid = UUID.uuid4() + + assert_raise ArgumentError, + "initial_state module Commanded.Commands.StateModuleTest.InvalidStateModule must export initial_state/0 function.", + fn -> + Aggregate.aggregate_state( + DefaultApp, + StateModuleAggregate, + uuid, + initial_state: InvalidStateModule + ) + end + end + + test "should reject opening a running aggregate with conflicting initial_state" do + uuid = UUID.uuid4() + + assert :ok = + StateModuleRouter.dispatch( + %CreateCommand{uuid: uuid, name: "Initial"}, + @dispatch_opts + ) + + assert {:error, + {:conflicting_initial_state, + %{ + aggregate_module: StateModuleAggregate, + aggregate_uuid: ^uuid, + requested_initial_state: nil, + running_initial_state: State + }}} = + Supervisor.open_aggregate(DefaultApp, StateModuleAggregate, uuid) + end + + test "should return conflict error when dispatching from router with mismatched initial_state" do + uuid = UUID.uuid4() + + Code.eval_string(""" + defmodule Commanded.Commands.StateModuleTest.RuntimeMismatchedInitialStateRouter do + use Commanded.Commands.Router + + alias Commanded.Commands.StateModuleAggregate + alias Commanded.Commands.StateModuleAggregate.CreateCommand + + dispatch CreateCommand, + to: StateModuleAggregate, + identity: :uuid + end + """) + + assert :ok = + StateModuleRouter.dispatch( + %CreateCommand{uuid: uuid, name: "Initial"}, + @dispatch_opts + ) + + router = + Module.concat([ + Commanded.Commands.StateModuleTest, + RuntimeMismatchedInitialStateRouter + ]) + + assert {:error, + {:conflicting_initial_state, + %{ + aggregate_module: StateModuleAggregate, + aggregate_uuid: ^uuid, + requested_initial_state: nil, + running_initial_state: State + }}} = + router.dispatch( + %CreateCommand{uuid: uuid, name: "Mismatch"}, + @dispatch_opts + ) + end + end +end diff --git a/test/commands/support/state_module/state_module_aggregate.ex b/test/commands/support/state_module/state_module_aggregate.ex new file mode 100644 index 00000000..a0887403 --- /dev/null +++ b/test/commands/support/state_module/state_module_aggregate.ex @@ -0,0 +1,47 @@ +defmodule Commanded.Commands.StateModuleAggregate do + @moduledoc """ + An aggregate that uses a separate state module. + This module defines only the behavior (execute/apply), not the state struct. + The initial_state option in the router specifies which state module to use. + """ + + alias Commanded.Commands.StateModuleAggregate.State + alias Commanded.Commands.StateModuleAggregate.{CreateCommand, UpdateCommand} + alias Commanded.Commands.StateModuleAggregate.{CreatedEvent, UpdatedEvent} + + defmodule CreateCommand do + @derive Jason.Encoder + defstruct [:uuid, :name] + end + + defmodule UpdateCommand do + @derive Jason.Encoder + defstruct [:uuid, :name] + end + + defmodule CreatedEvent do + @derive Jason.Encoder + defstruct [:uuid, :name] + end + + defmodule UpdatedEvent do + @derive Jason.Encoder + defstruct [:uuid, :name] + end + + def execute(%State{uuid: nil}, %CreateCommand{uuid: uuid, name: name}) do + %CreatedEvent{uuid: uuid, name: name} + end + + def execute(%State{uuid: _uuid}, %UpdateCommand{uuid: uuid, name: name}) do + %UpdatedEvent{uuid: uuid, name: name} + end + + def apply(%State{} = state, %CreatedEvent{uuid: uuid, name: name}) do + %State{state | uuid: uuid, name: name} + end + + def apply(%State{} = state, %UpdatedEvent{name: name}) do + %State{state | name: name} + end +end diff --git a/test/commands/support/state_module/state_module_router.ex b/test/commands/support/state_module/state_module_router.ex new file mode 100644 index 00000000..1e10de14 --- /dev/null +++ b/test/commands/support/state_module/state_module_router.ex @@ -0,0 +1,13 @@ +defmodule Commanded.Commands.StateModuleRouter do + @moduledoc false + use Commanded.Commands.Router + + alias Commanded.Commands.StateModuleAggregate + alias Commanded.Commands.StateModuleAggregate.State + alias Commanded.Commands.StateModuleAggregate.{CreateCommand, UpdateCommand} + + dispatch [CreateCommand, UpdateCommand], + to: StateModuleAggregate, + initial_state: State, + identity: :uuid +end diff --git a/test/commands/support/state_module/state_module_state.ex b/test/commands/support/state_module/state_module_state.ex new file mode 100644 index 00000000..7ce462e9 --- /dev/null +++ b/test/commands/support/state_module/state_module_state.ex @@ -0,0 +1,11 @@ +defmodule Commanded.Commands.StateModuleAggregate.State do + @moduledoc """ + A separate state module that defines the aggregate's state struct + and the initial_state/0 callback. + This could be a protobuf-generated module or any other struct module. + """ + @derive Jason.Encoder + defstruct [:uuid, :name] + + def initial_state, do: %__MODULE__{} +end diff --git a/test/opentelemetry/event_handler_test.exs b/test/opentelemetry/event_handler_test.exs index c4b1f665..adea4b76 100644 --- a/test/opentelemetry/event_handler_test.exs +++ b/test/opentelemetry/event_handler_test.exs @@ -280,8 +280,7 @@ defmodule Commanded.OpenTelemetry.EventHandlerTest do assert map_size(exc_attrs_map) == 3 assert exc_attrs_map[:"exception.type"] == "Elixir.KeyError" - assert exc_attrs_map[:"exception.message"] == - "key :account_number not found in:\n\n %{balance: 100}\n" + assert exc_attrs_map[:"exception.message"] =~ "key :account_number not found in:" # Version number in stacktrace changes per release stacktrace = exc_attrs_map[:"exception.stacktrace"] diff --git a/test/subscriptions/subscriptions_test.exs b/test/subscriptions/subscriptions_test.exs index 975af0e6..f3152446 100644 --- a/test/subscriptions/subscriptions_test.exs +++ b/test/subscriptions/subscriptions_test.exs @@ -25,10 +25,15 @@ defmodule Commanded.SubscriptionsTest do test "should not remove PID when process terminates" do pid = spawn_link(fn -> + receive do + :continue -> :ok + end + :ok = Subscriptions.register(DefaultApp, "handler1", Handler1, :strong) end) ref = Process.monitor(pid) + send(pid, :continue) assert_receive {:DOWN, ^ref, :process, _, :normal} assert Subscriptions.all(DefaultApp) == [{"handler1", Handler1, pid}]