neiam /oh-my-ex-oh-ircd

defmodule OhMyExOhIrcd.IRC.MessageLog do
  @moduledoc """
  One per node. Holds every channel's PRIVMSG log in a single
  `DeltaCrdt.AWLWWMap` keyed by `{channel, hlc_ts}`.

  Why per-node (not per-channel-per-node):

    * Horde-supervised Channel processes are singletons; during a
      netsplit each partition spawns its own Channel. If the CRDT were
      owned by the Channel pid its state would be lost when Horde
      resolves the conflict on heal. A per-node CRDT means every node
      already has a full copy of the data — when Horde moves a Channel
      to a different node, the new owner just reads the local CRDT.

  All node-local `MessageLog` processes mesh together via
  `DeltaCrdt.set_neighbours/2`, driven by
  `OhMyExOhIrcd.Cluster.NodeListener`.

  Replay model:

    * Append entries with `append/3`. We tag each entry with an HLC
      timestamp so cross-partition ordering is well-defined.
    * Each node tracks its own `seen` cursor (a MapSet of CRDT keys).
      Keys not in `seen` are considered "new to this node" and are
      delivered to local subscribers in HLC order. The cursor itself is
      *not* in the CRDT — it's local bookkeeping so each node fans out
      messages exactly once to its locally connected clients.
    * On a split→heal the merged CRDT exposes the other side's entries
      as "new keys" on every node, which is exactly the replay we want.
  """

  use GenServer
  require Logger

  alias OhMyExOhIrcd.IRC.HLC

  @type entry :: %{
          ts: HLC.t(),
          sender: String.t(),
          target: String.t(),
          text: String.t(),
          kind: :privmsg | :notice
        }

  @crdt_name __MODULE__.CRDT
  @pubsub OhMyExOhIrcd.PubSub

  def crdt_name, do: @crdt_name

  ## Public API

  def start_link(_), do: GenServer.start_link(__MODULE__, [], name: __MODULE__)

  @spec append(String.t(), String.t(), String.t(), :privmsg | :notice) :: HLC.t()
  def append(channel, sender, text, kind \\ :privmsg),
    do: GenServer.call(__MODULE__, {:append, channel, sender, text, kind})

  @doc "Return all log entries for `channel`, sorted by HLC."
  def entries(channel), do: GenServer.call(__MODULE__, {:entries, channel})

  @doc """
  Update CRDT neighbour list. Called by NodeListener whenever
  `Node.list/0` changes.
  """
  def set_neighbours(nodes) do
    GenServer.cast(__MODULE__, {:set_neighbours, nodes})
  end

  @doc "Force a diff/drain cycle. Mostly for tests."
  def sync, do: GenServer.call(__MODULE__, :sync)

  ## Server

  @impl true
  def init(_) do
    {:ok, crdt} =
      DeltaCrdt.start_link(DeltaCrdt.AWLWWMap,
        sync_interval: 100,
        name: @crdt_name,
        on_diffs: {__MODULE__, :on_diffs, [self()]}
      )

    {:ok, %{crdt: crdt, seen: MapSet.new()}}
  end

  @impl true
  def handle_call({:append, channel, sender, text, kind}, _from, state) do
    ts = HLC.now()
    entry = %{ts: ts, sender: sender, target: channel, text: text, kind: kind}
    DeltaCrdt.put(state.crdt, {channel, ts}, entry)
    {:reply, ts, state}
  end

  def handle_call({:entries, channel}, _from, state) do
    list =
      state.crdt
      |> DeltaCrdt.to_map()
      |> Enum.flat_map(fn
        {{^channel, _ts}, entry} -> [entry]
        _ -> []
      end)
      |> Enum.sort_by(& &1.ts)

    {:reply, list, state}
  end

  def handle_call(:sync, _from, state) do
    new_state = drain(state)
    {:reply, :ok, new_state}
  end

  @impl true
  def handle_cast({:set_neighbours, nodes}, state) do
    neighbours = for n <- nodes, n != node(), do: {@crdt_name, n}
    DeltaCrdt.set_neighbours(state.crdt, neighbours)
    {:noreply, state}
  end

  @impl true
  def handle_info(:diffs, state), do: {:noreply, drain(state)}
  def handle_info(_, state), do: {:noreply, state}

  ## Diff callback (runs in DeltaCrdt's process; just signal us).
  ## DeltaCrdt invokes `apply(M, F, ExtraArgs ++ [diffs])`, so ExtraArgs
  ## come first.
  @doc false
  def on_diffs(pid, _diffs), do: send(pid, :diffs)

  defp drain(state) do
    map = DeltaCrdt.to_map(state.crdt)
    all_keys = MapSet.new(Map.keys(map))
    new_keys = MapSet.difference(all_keys, state.seen)

    new_keys
    |> MapSet.to_list()
    |> Enum.sort()
    |> Enum.map(&Map.fetch!(map, &1))
    |> Enum.each(fn entry ->
      HLC.observe(entry.ts)

      # local_broadcast, NOT broadcast: every node runs this same drain
      # against its own (fully replicated) CRDT, so each node must only
      # fan out to its *locally* connected subscribers. A cluster-wide
      # broadcast here delivers every message once per node.
      Phoenix.PubSub.local_broadcast(@pubsub, channel_topic(entry.target), {:irc_privmsg, entry})
    end)

    %{state | seen: all_keys}
  end

  def channel_topic(channel), do: "channel:" <> String.downcase(channel)
end