Party event sourcing

lib/teiserver/party/event.ex

@@ -0,0 +1,13 @@
+defprotocol Teiserver.Party.Event do
+  alias Teiserver.Party.Types, as: PT
+
+  @doc """
+  Given an event and an aggregate, returns the new aggregate.
+
+  This function must be side-effect free, because many events may be folded
+  together or rolled back before arriving to the final aggregate to be used
+  by the lobby.
+  """
+  @spec apply_event(term(), PT.Aggregate.t()) :: PT.Aggregate.t()
+  def apply_event(event, data)
+end

lib/teiserver/party/events/leave.ex

@@ -0,0 +1,39 @@
+defmodule Teiserver.Party.Events.Leave do
+  @moduledoc """
+  leave the party
+  """
+  alias Teiserver.Account.User
+
+  @enforce_keys [:leaver_id]
+  defstruct [:leaver_id]
+
+  @type t() :: %__MODULE__{
+          leaver_id: User.id()
+        }
+end
+
+defimpl Teiserver.Party.Event, for: Teiserver.Party.Events.Leave do
+  alias Teiserver.Helpers.MonitorCollection, as: MC
+  alias Teiserver.Party.Events.Leave
+  alias Teiserver.Party.Types.Aggregate
+  import Access, only: [key!: 1]
+
+  def apply_event(%Leave{} = ev, %Aggregate{} = agg) do
+    case Map.pop(agg.data.members, ev.leaver_id) do
+      {nil, _members} ->
+        agg
+
+      {_member, rest} when map_size(rest) == 0 ->
+        put_in(agg.data.members, %{})
+
+      {_member, new_members} ->
+        new_aggregate =
+          agg
+          |> Aggregate.put_in_data([key!(:members)], new_members)
+          |> Aggregate.update_in_data([key!(:monitors)], &MC.demonitor_by_val(&1, ev.leaver_id))
+          |> Aggregate.add_side_effect(:notify_updated)
+
+        new_aggregate
+    end
+  end
+end

lib/teiserver/party/server.ex

@@ -11,6 +11,8 @@ defmodule Teiserver.Party.Server do
   alias Teiserver.Matchmaking
   alias Teiserver.Messaging
   alias Teiserver.Party
+  alias Teiserver.Party.Event
+  alias Teiserver.Party.Events
   alias Teiserver.Party.Types, as: PT
   alias Teiserver.Player
   alias Teiserver.Tachyon
@@ -248,24 +250,12 @@ defmodule Teiserver.Party.Server do
   end
 
   def handle_event({:call, from}, {:leave, user_id}, :running, %PT.Data{} = data) do
-    case Map.pop(data.members, user_id) do
-      {nil, _members} ->
-        {:keep_state, data, [{:reply, from, {:error, :not_a_member}}]}
-
-      {_member, rest} when map_size(rest) == 0 ->
-        {:stop_and_reply, :normal, [{:reply, from, :ok}], %{data | members: %{}} |> bump()}
-
-      {_member, new_members} ->
-        new_data =
-          %{data | members: new_members}
-          |> bump()
-          |> Map.update!(:monitors, &MC.demonitor_by_val(&1, user_id))
-
-        for id <- Map.keys(data.invited) |> Stream.concat(Map.keys(data.members)) do
-          Player.party_notify_updated(id, overview_from_data(new_data))
-        end
+    aggregate = process_events(data, [%Events.Leave{leaver_id: user_id}])
 
-        {:keep_state, new_data, {:reply, from, :ok}}
+    if aggregate.data == data do
+      {:keep_state, data, [{:reply, from, {:error, :not_a_member}}]}
+    else
+      {:keep_state, aggregate.data, [{:reply, from, :ok} | aggregate.actions]}
     end
   end
 
@@ -599,6 +589,10 @@ defmodule Teiserver.Party.Server do
     {:keep_state, data, [:postpone]}
   end
 
+  def handle_event(:internal, :empty, _state, %PT.Data{} = data) do
+    {:stop, {:shutdown, :empty}, data}
+  end
+
   def handle_event(:state_timeout, :snapshot_timeout, :starting_up, %PT.Data{} = data) do
     Logger.warning("failed to recover before time out. Missing #{inspect(data.ids_to_rejoin)}")
     {:stop, :normal}
@@ -636,6 +630,43 @@ defmodule Teiserver.Party.Server do
 
   def terminate(_reason, _state, %PT.Data{} = _data), do: nil
 
+  defp process_events(%PT.Data{} = data, events) do
+    new_aggregate = compute_aggregate(data, events)
+
+    Enum.each(new_aggregate.side_effects, fn effect ->
+      process_side_effect(new_aggregate.data, effect)
+    end)
+
+    new_aggregate
+  end
+
+  defp compute_aggregate(%PT.Data{} = data, events) do
+    aggregate = %PT.Aggregate{data: data}
+    final_aggregate = Enum.reduce(events, aggregate, &Event.apply_event/2)
+
+    if aggregate.data == final_aggregate.data do
+      final_aggregate
+    else
+      final_aggregate = update_in(final_aggregate.data.version, &(&1 + 1))
+
+      if map_size(final_aggregate.data.members) == 0 do
+        update_in(final_aggregate.actions, &(&1 ++ [{:next_event, :internal, :empty}]))
+      else
+        final_aggregate
+      end
+    end
+  end
+
+  defp process_side_effect(%PT.Data{} = data, :notify_updated) do
+    overview = overview_from_data(data)
+
+    Map.keys(data.invited)
+    |> Enum.concat(Map.keys(data.members))
+    |> Enum.each(fn id ->
+      Player.party_notify_updated(id, overview)
+    end)
+  end
+
   defp notify_updated(data) do
     for id <- Map.keys(data.invited) |> Stream.concat(Map.keys(data.members)) do
       Player.party_notify_updated(id, overview_from_data(data))

lib/teiserver/party/types/aggregate.ex

@@ -0,0 +1,52 @@
+defmodule Teiserver.Party.Types.Aggregate do
+  @moduledoc """
+  An aggregate used when processing events related to parties
+  """
+
+  alias Teiserver.Party.Types, as: PT
+
+  @enforce_keys [:data]
+  defstruct [:data, side_effects: [], actions: []]
+
+  @type t() :: %__MODULE__{
+          data: PT.Data.t(),
+          side_effects: list(),
+          actions: list(:gen_statem.action())
+        }
+
+  @doc """
+  helper to add nested data in the aggregate, similar to `Kernel.put_in/3`
+  """
+  def put_in_data(aggregate, keys, datum) do
+    put_in(aggregate, [Access.key!(:data)] ++ keys, datum)
+  end
+
+  @doc """
+  helper to modify data in the aggregate, similar to `Kernel.update_in/3`
+  """
+  def update_in_data(aggregate, keys, fun) do
+    update_in(aggregate.data, fn data ->
+      update_in(data, keys, fun)
+    end)
+  end
+
+  @doc """
+  helper to append a new side effect to the aggregate
+  """
+  @spec add_side_effect(t(), term()) :: t()
+  def add_side_effect(aggregate, effect) do
+    update_in(aggregate, [Access.key!(:side_effects)], fn side_effects ->
+      side_effects ++ [effect]
+    end)
+  end
+
+  @doc """
+  helper to append a new action to the aggregate
+  """
+  @spec add_side_action(t(), :gen_statem.action()) :: t()
+  def add_side_action(aggregate, action) do
+    update_in(aggregate, [Access.key!(:actions)], fn actions ->
+      actions ++ [action]
+    end)
+  end
+end