inspect_aggregate.ex

defmodule Mix.Tasks.InspectAggregate do
  use Mix.Task

  alias MyApp.EventStore

  def run(args) do
    Application.ensure_all_started(:eventstore)
    _ = EventStore.start_link()

    {opts, _} =
      OptionParser.parse!(args,
        strict: [aggregate_module: :string, stream: :string],
        aliases: [a: :aggregate_module, s: :stream]
      )

    stream_uuid = Keyword.fetch!(opts, :stream)
    aggregate_module = String.to_existing_atom("Elixir." <> Keyword.fetch!(opts, :aggregate_module))

    with {:ok, events} <- EventStore.read_stream_forward(stream_uuid) do
      events
      |> Enum.map(& &1.data)
      |> step_through(aggregate_module)
      |> IO.inspect(label: "step through all events and their effective state; in order")
    end
  end

  defp step_through(events, aggregate_module) do
    initial_state = struct(aggregate_module)
    acc = {initial_state, [{0, %{state: initial_state, event: nil}}]}

    {_current_state, states} =
      events
      |> Enum.with_index(1)
      |> Enum.reduce(acc, fn {event, index}, {current_state, states} ->
        next_state = aggregate_module.apply(current_state, event)
        {next_state, [{index, %{state: next_state, event: event}} | states]}
      end)

    states
    |> Enum.reverse()
    |> Enum.into(%{})
  end
end