Elixir OTP에서 파이프라인으로 상태관리하기

def handle_call({:deposit, user_id, amount}, _from, state) do
  # 검증
  if amount <= 0 do
    {:reply, {:error, :invalid_amount}, state}
  else
    # 상태 읽기
    user = Map.get(state.users, user_id)

    # 상태 전이
    new_user = %{user | balance: user.balance + amount}
    new_state = put_in(state.users[user_id], new_user)

    # 부수효과
    :ok = Audit.log(user_id, {:deposit, amount})

    {:reply, {:ok, new_user.balance}, new_state}
  end
end

defmodule Wallet.Command do
  defstruct [:type, :user_id, :amount, :idempotency_key]
end

defmodule Wallet.Context do
  defstruct [:state, :cmd, :user, :events, :reply]
end

defmodule Wallet.Pipeline do
  alias Wallet.{Context, Command}

  def run(state, %Command{} = cmd) do
    %Context{state: state, cmd: cmd, events: []}
    |> ok()
    |> validate_amount()
    |> load_user()
    |> apply_deposit()
    |> build_reply()
  end

  defp ok(ctx), do: {:ok, ctx}

  defp validate_amount({:ok, %Context{cmd: %Command{amount: amount}}} = ok_ctx)
       when is_integer(amount) and amount > 0,
       do: ok_ctx

  defp validate_amount({:ok, _ctx}), do: {:error, :invalid_amount}
  defp validate_amount({:error, _} = err), do: err

  defp load_user({:ok, %Context{state: state, cmd: %Command{user_id: uid}} = ctx}) do
    case get_in(state, [:users, uid]) do
      nil -> {:error, :user_not_found}
      user -> {:ok, %Context{ctx | user: user}}
    end
  end

  defp load_user({:error, _} = err), do: err

  defp apply_deposit({:ok, %Context{state: state, user: user, cmd: %Command{amount: amount}} = ctx}) do
    new_user = %{user | balance: user.balance + amount}
    new_state = put_in(state, [:users, user.id], new_user)

    event = {:deposited, user.id, amount}

    {:ok, %Context{ctx | state: new_state, user: new_user, events: [event | ctx.events]}}
  end

  defp apply_deposit({:error, _} = err), do: err

  defp build_reply({:ok, %Context{user: user} = ctx}) do
    {:ok, %Context{ctx | reply: {:ok, user.balance}}}
  end

  defp build_reply({:error, reason}), do: {:error, reason}
end

defmodule Wallet.Server do
  use GenServer
  alias Wallet.{Command, Pipeline}

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def deposit(user_id, amount, idempotency_key \\ nil) do
    cmd = %Command{type: :deposit, user_id: user_id, amount: amount, idempotency_key: idempotency_key}
    GenServer.call(__MODULE__, {:command, cmd})
  end

  @impl true
  def init(_opts) do
    {:ok, %{users: %{}, processed: MapSet.new()}}
  end

  @impl true
  def handle_call({:command, cmd}, _from, state) do
    case Pipeline.run(state, cmd) do
      {:ok, ctx} ->
        # 부수효과는 여기서 분리해서 처리하거나, after-reply로 넘긴다
        new_state = ctx.state
        {:reply, ctx.reply, new_state}

      {:error, reason} ->
        {:reply, {:error, reason}, state}
    end
  end
end

def handle_call({:command, cmd}, _from, state) do
  case Pipeline.run(state, cmd) do
    {:ok, ctx} ->
      Enum.each(Enum.reverse(ctx.events), fn event ->
        Wallet.SideEffects.dispatch(event)
      end)

      {:reply, ctx.reply, ctx.state}

    {:error, reason} ->
      {:reply, {:error, reason}, state}
  end
end

defmodule Result do
  def then({:ok, v}, fun) when is_function(fun, 1), do: fun.(v)
  def then({:error, _} = err, _fun), do: err

  def map({:ok, v}, fun) when is_function(fun, 1), do: {:ok, fun.(v)}
  def map({:error, _} = err, _fun), do: err
end

alias Result

{:ok, %Wallet.Context{state: state, cmd: cmd, events: []}}
|> Result.then(&Wallet.PipelineSteps.validate_amount/1)
|> Result.then(&Wallet.PipelineSteps.load_user/1)
|> Result.then(&Wallet.PipelineSteps.apply_deposit/1)
|> Result.then(&Wallet.PipelineSteps.build_reply/1)

defp ensure_idempotent({:ok, %Context{state: state, cmd: %Command{idempotency_key: nil}} = ctx}) do
  {:ok, ctx}
end

defp ensure_idempotent({:ok, %Context{state: state, cmd: %Command{idempotency_key: key}} = ctx}) do
  if MapSet.member?(state.processed, key) do
    {:error, :duplicate_command}
  else
    new_state = %{state | processed: MapSet.put(state.processed, key)}
    {:ok, %Context{ctx | state: new_state}}
  end
end

defp ensure_idempotent({:error, _} = err), do: err

%Context{state: state, cmd: cmd, events: []}
|> ok()
|> ensure_idempotent()
|> validate_amount()
|> load_user()
|> apply_deposit()
|> build_reply()

defmodule Wallet.PipelineTest do
  use ExUnit.Case
  alias Wallet.{Command, Pipeline}

  test "deposit increases balance and emits event" do
    state = %{users: %{1 => %{id: 1, balance: 100}}, processed: MapSet.new()}
    cmd = %Command{type: :deposit, user_id: 1, amount: 50, idempotency_key: "k1"}

    assert {:ok, ctx} = Pipeline.run(state, cmd)
    assert ctx.reply == {:ok, 150}
    assert get_in(ctx.state, [:users, 1, :balance]) == 150
    assert Enum.member?(ctx.events, {:deposited, 1, 50})
  end
end