syntropy

defmodule Syntropy.ClusterInventory do
  @moduledoc """
  RPC-safe cluster inventory and invalidation boundary for Syntropy.

  Each node remains the source of truth for its own agents and local runtime
  state. This module aggregates those local exports without replicating
  supervisor state.
  """

  alias Syntropy.{ClusterInfo, LatticeMath, LatticeSupervisor, TaskScheduler}

  @cluster_topic "lattice:cluster"
  @rpc_timeout 5_000

  @type agent_ref :: %{id: String.t(), runtime_id: String.t(), node_id: String.t()}

  @type agent_export :: map()

  @type active_task_export :: %{
          task_id: String.t(),
          coordinator_node_id: String.t(),
          selected_agents: [agent_ref()],
          started_at: DateTime.t()
        }

  @type node_export :: %{
          node_id: String.t(),
          node_name: String.t(),
          services: [ClusterInfo.service_health()],
          agents: [agent_export()],
          active_tasks: [active_task_export()]
        }

  @type cluster_export :: %{
          status: String.t(),
          local_node_id: String.t(),
          local_node_name: String.t(),
          nodes: [node_export()],
          unavailable_nodes: [map()]
        }

  @spec local_export(keyword()) :: node_export()
  def local_export(opts \\ []) do
    %{
      node_id: ClusterInfo.node_id(),
      node_name: ClusterInfo.node_name(),
      services: ClusterInfo.service_health(),
      agents: Enum.map(LatticeSupervisor.list_agents(), &serialize_agent/1),
      active_tasks:
        if(Keyword.get(opts, :include_active_tasks, true),
          do: TaskScheduler.active_tasks_snapshot(),
          else: []
        )
    }
  end

  @spec cluster_export(keyword()) :: cluster_export()
  def cluster_export(opts \\ []) do
    local =
      local_export(include_active_tasks: Keyword.get(opts, :include_local_active_tasks, true))

    {nodes, unavailable_nodes} =
      Node.list()
      |> Enum.reduce({[local], []}, fn node, {nodes, unavailable_nodes} ->
        case :rpc.call(node, __MODULE__, :local_export, [], @rpc_timeout) do
          %{} = export ->
            {[export | nodes], unavailable_nodes}

          {:badrpc, reason} ->
            {nodes,
             [
               %{
                 node_name: Atom.to_string(node),
                 reason: inspect(reason)
               }
               | unavailable_nodes
             ]}

          other ->
            {nodes,
             [
               %{
                 node_name: Atom.to_string(node),
                 reason: inspect(other)
               }
               | unavailable_nodes
             ]}
        end
      end)

    %{
      status: if(unavailable_nodes == [], do: "ok", else: "degraded"),
      local_node_id: local.node_id,
      local_node_name: local.node_name,
      nodes: Enum.sort_by(nodes, &{&1.node_id, &1.node_name}),
      unavailable_nodes: Enum.sort_by(unavailable_nodes, & &1.node_name)
    }
  end

  @spec cluster_agents() :: [agent_export()]
  def cluster_agents do
    cluster_export(include_local_active_tasks: false)
    |> Map.fetch!(:nodes)
    |> Enum.flat_map(& &1.agents)
    |> LatticeMath.sort_agents()
  end

  @spec active_tasks() :: [active_task_export()]
  def active_tasks do
    cluster_export()
    |> Map.fetch!(:nodes)
    |> Enum.flat_map(& &1.active_tasks)
    |> Enum.sort_by(& &1.task_id)
  end

  @spec busy_runtime_ids([active_task_export()]) :: MapSet.t(String.t())
  def busy_runtime_ids(local_active_tasks \\ []) do
    remote_active_tasks =
      cluster_export(include_local_active_tasks: false)
      |> Map.fetch!(:nodes)
      |> Enum.reject(&(&1.node_id == ClusterInfo.node_id()))
      |> Enum.flat_map(& &1.active_tasks)

    (local_active_tasks ++ remote_active_tasks)
    |> Enum.flat_map(& &1.selected_agents)
    |> Enum.map(& &1.runtime_id)
    |> MapSet.new()
  end

  @spec fetch_agent(agent_ref() | String.t()) :: agent_export() | nil
  def fetch_agent(runtime_id) when is_binary(runtime_id) do
    case ClusterInfo.parse_runtime_id(runtime_id) do
      {:ok, ref} -> fetch_agent(ref)
      :error -> nil
    end
  end

  def fetch_agent(%{id: local_id, node_id: owner_node_id} = ref) do
    case node_name_for_ref(ref) do
      nil ->
        nil

      node_name ->
        fetch_agent_on_node(node_name, local_id, owner_node_id)
    end
  end

  @spec notify_change(String.t() | atom()) :: :ok
  def notify_change(reason) do
    payload = %{
      reason: to_string(reason),
      node_id: ClusterInfo.node_id(),
      node_name: ClusterInfo.node_name(),
      occurred_at: DateTime.utc_now()
    }

    # The PG PubSub adapter already fans this broadcast out to every
    # connected node, so no RPC re-broadcast is needed (it would deliver
    # duplicate :cluster_refresh messages to remote subscribers).
    broadcast_refresh(payload)

    :ok
  end

  @spec find_agent(String.t()) :: agent_export() | nil
  def find_agent(identifier) do
    Enum.find(cluster_agents(), fn agent ->
      agent.id == identifier or agent.runtime_id == identifier
    end)
  end

  defp node_name_for_ref(%{node_name: node_name}) when is_binary(node_name), do: node_name

  defp node_name_for_ref(%{node_id: owner_node_id}) do
    cluster_export(include_local_active_tasks: false)
    |> Map.fetch!(:nodes)
    |> Enum.find_value(fn node ->
      if node.node_id == owner_node_id, do: node.node_name
    end)
  end

  defp serialize_agent(agent) do
    agent
    |> Map.from_struct()
    |> Map.put(:node_id, Map.get(agent, :node_id, ClusterInfo.node_id()))
    |> Map.put(:runtime_id, Map.get(agent, :runtime_id, ClusterInfo.runtime_id(agent.id)))
    |> Map.put(:node_name, ClusterInfo.node_name())
  end

  defp maybe_serialize_agent(nil), do: nil
  defp maybe_serialize_agent(agent), do: maybe_serialize_agent(agent, ClusterInfo.node_name())

  defp maybe_serialize_agent(agent, node_name) do
    agent
    |> Map.from_struct()
    |> Map.put(:node_id, Map.get(agent, :node_id, ClusterInfo.node_id()))
    |> Map.put(:runtime_id, Map.get(agent, :runtime_id, ClusterInfo.runtime_id(agent.id)))
    |> Map.put(:node_name, node_name)
  end

  defp fetch_agent_on_node(node_name, local_id, owner_node_id) do
    if node_name == ClusterInfo.node_name() do
      local_id
      |> LatticeSupervisor.get_agent()
      |> maybe_serialize_agent()
    else
      case :rpc.call(
             node_name_to_atom(node_name),
             LatticeSupervisor,
             :get_agent,
             [local_id],
             @rpc_timeout
           ) do
        nil -> nil
        {:badrpc, _reason} -> nil
        agent -> agent |> Map.put(:node_id, owner_node_id) |> maybe_serialize_agent(node_name)
      end
    end
  end

  defp broadcast_refresh(payload) do
    Phoenix.PubSub.broadcast(Syntropy.PubSub, @cluster_topic, {:cluster_refresh, payload})
  end

  defp node_name_to_atom(node_name) when is_binary(node_name) do
    Enum.find([Node.self() | Node.list()], &(Atom.to_string(&1) == node_name)) ||
      String.to_atom(node_name)
  end
end