From 27c4544406e9ca42ad3bbbb87a1b008cde04d3a5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gr=C3=A9goire=20Charvet=20=E9=BB=91=E7=93=9C?= Date: Sun, 2 Aug 2026 19:13:55 +0200 Subject: [PATCH 1/2] refactor shutting down party when empty Instead of using `:stop_and_reply`, use an internal event. This will work better when moving to event sourcing --- lib/teiserver/party/server.ex | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/lib/teiserver/party/server.ex b/lib/teiserver/party/server.ex index cd2852ea7..7e2817398 100644 --- a/lib/teiserver/party/server.ex +++ b/lib/teiserver/party/server.ex @@ -253,7 +253,8 @@ defmodule Teiserver.Party.Server do {: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()} + new_data = %{data | members: %{}} |> bump() + {:keep_state, new_data, [{:reply, from, :ok}, {:next_event, :internal, :empty}]} {_member, new_members} -> new_data = @@ -599,6 +600,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} From 9e84bffff9138449e8e0ba9ad70af9b14b74b454 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gr=C3=A9goire=20Charvet=20=E9=BB=91=E7=93=9C?= Date: Sun, 2 Aug 2026 19:13:55 +0200 Subject: [PATCH 2/2] Use event sourcing for leaving party --- lib/teiserver/party/event.ex | 13 ++++++ lib/teiserver/party/events/leave.ex | 39 ++++++++++++++++ lib/teiserver/party/server.ex | 62 ++++++++++++++++++-------- lib/teiserver/party/types/aggregate.ex | 52 +++++++++++++++++++++ 4 files changed, 148 insertions(+), 18 deletions(-) create mode 100644 lib/teiserver/party/event.ex create mode 100644 lib/teiserver/party/events/leave.ex create mode 100644 lib/teiserver/party/types/aggregate.ex diff --git a/lib/teiserver/party/event.ex b/lib/teiserver/party/event.ex new file mode 100644 index 000000000..d25bcfc48 --- /dev/null +++ b/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 diff --git a/lib/teiserver/party/events/leave.ex b/lib/teiserver/party/events/leave.ex new file mode 100644 index 000000000..974d78f09 --- /dev/null +++ b/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 diff --git a/lib/teiserver/party/server.ex b/lib/teiserver/party/server.ex index 7e2817398..943941b16 100644 --- a/lib/teiserver/party/server.ex +++ b/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,25 +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 -> - new_data = %{data | members: %{}} |> bump() - {:keep_state, new_data, [{:reply, from, :ok}, {:next_event, :internal, :empty}]} + aggregate = process_events(data, [%Events.Leave{leaver_id: user_id}]) - {_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 - - {: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 @@ -641,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)) diff --git a/lib/teiserver/party/types/aggregate.ex b/lib/teiserver/party/types/aggregate.ex new file mode 100644 index 000000000..4f9c3ed09 --- /dev/null +++ b/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