vela_cache

defmodule Vela.Distributed.Cluster do
  @moduledoc """
  Monitors cluster membership and maintains a list of connected nodes.

  Subscribes to Erlang node events via :net_kernel.monitor_nodes/2.
  When nodes join or leave, broadcasts to all processes that joined
  the :vela_cluster_events pg group.

  This is a global singleton — one per BEAM node, shared by all caches.
  """

  use GenServer

  @pg_scope :vela_pg
  @pg_group :vela_cluster_events

  # ---- Public API ----

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

  @doc "Returns all currently connected nodes including the local node."
  def nodes do
    GenServer.call(__MODULE__, :nodes)
  end

  @doc """
  Subscribe the calling process to cluster events.
  The process will receive:
    {:vela_cluster_event, :node_up, node}
    {:vela_cluster_event, :node_down, node}
  """
  def subscribe do
    :pg.join(@pg_scope, @pg_group, self())
  end

  @doc "Unsubscribe the calling process from cluster events."
  def unsubscribe do
    :pg.leave(@pg_scope, @pg_group, self())
  end

  # ---- GenServer Callbacks ----

  @impl true
  def init(_opts) do
    # Subscribe to node join/leave events from the BEAM
    :net_kernel.monitor_nodes(true, node_type: :visible)

    # Start with all currently connected nodes plus ourselves
    connected = [node() | Node.list()]

    {:ok, %{nodes: MapSet.new(connected)}}
  end

  @impl true
  def handle_call(:nodes, _from, state) do
    {:reply, MapSet.to_list(state.nodes), state}
  end

  @impl true
  def handle_info({:nodeup, new_node, _info}, state) do
    new_nodes = MapSet.put(state.nodes, new_node)

    :telemetry.execute(
      [:vela, :cache, :node, :up],
      %{},
      %{node: new_node}
    )

    broadcast(:node_up, new_node)

    {:noreply, %{state | nodes: new_nodes}}
  end

  @impl true
  def handle_info({:nodedown, lost_node, _info}, state) do
    new_nodes = MapSet.delete(state.nodes, lost_node)

    :telemetry.execute(
      [:vela, :cache, :node, :down],
      %{},
      %{node: lost_node}
    )

    broadcast(:node_down, lost_node)

    {:noreply, %{state | nodes: new_nodes}}
  end

  defp broadcast(event, node) do
    members = :pg.get_local_members(@pg_scope, @pg_group)
    Enum.each(members, &send(&1, {:vela_cluster_event, event, node}))
  end
end