vela_cache

defmodule Vela.Topology.Replicated do
  @moduledoc """
  Replicated topology: every node holds a complete copy of all data.

  Writes are applied locally first (fast), then broadcast to all other
  nodes asynchronously via :erpc.cast (eventual consistency).

  Reads always serve from the local node — no network hop needed.
  """

  @behaviour Vela.Topology

  alias Vela.Cache.Entry

  defstruct [:backend, :backend_state, :cache_name]

  @impl true
  def init(config) do
    {:ok, backend_state} = config.backend.init(config)

    state = %__MODULE__{
      backend: config.backend,
      backend_state: backend_state,
      cache_name: config.name
    }

    {:ok, state}
  end

  @impl true
  def get(%__MODULE__{backend: b, backend_state: bs}, key) do
    b.get(bs, key)
  end

  @impl true
  def put(%__MODULE__{backend: b, backend_state: bs} = state, %Entry{} = entry) do
    # Write locally first
    {:ok, new_bs} = b.put(bs, entry)

    # Broadcast to other nodes asynchronously
    broadcast_to_cluster(state, {:replicate_put, entry})

    {:ok, %{state | backend_state: new_bs}}
  end

  @impl true
  def delete(%__MODULE__{backend: b, backend_state: bs} = state, key) do
    {:ok, new_bs} = b.delete(bs, key)
    broadcast_to_cluster(state, {:replicate_delete, key})
    {:ok, %{state | backend_state: new_bs}}
  end

  @impl true
  def get_many(%__MODULE__{backend: b, backend_state: bs}, keys) do
    b.get_many(bs, keys)
  end

  @impl true
  def put_many(%__MODULE__{} = state, entries) do
    Enum.reduce(entries, {:ok, state}, fn entry, {:ok, acc} ->
      put(acc, entry)
    end)
  end

  @impl true
  def flush(%__MODULE__{backend: b, backend_state: bs} = state) do
    {:ok, new_bs} = b.flush(bs)
    broadcast_to_cluster(state, :replicate_flush)
    {:ok, %{state | backend_state: new_bs}}
  end

  @impl true
  def size(%__MODULE__{backend: b, backend_state: bs}) do
    b.size(bs)
  end

  @impl true
  def invalidate_tag(%__MODULE__{backend: b, backend_state: bs} = state, tag) do
    {:ok, count, new_bs} = b.delete_by_tag(bs, tag)
    broadcast_to_cluster(state, {:replicate_invalidate_tag, tag})
    {:ok, count, %{state | backend_state: new_bs}}
  end

  # ---- Replication Internals ----

  @doc """
  Handles replication messages arriving from other nodes.
  Reads topology state from persistent_term — no GenServer needed.
  Called via :erpc.cast on the remote node.
  """
  def handle_replication_message(message, cache_name) do
    ts = :persistent_term.get({:vela_topology_state, cache_name})

    case message do
      {:replicate_put, entry} ->
        ts.backend.put(ts.backend_state, entry)

      {:replicate_delete, key} ->
        ts.backend.delete(ts.backend_state, key)

      :replicate_flush ->
        ts.backend.flush(ts.backend_state)

      {:replicate_invalidate_tag, tag} ->
        ts.backend.delete_by_tag(ts.backend_state, tag)
    end
  end

  defp broadcast_to_cluster(state, message) do
    other_nodes = Node.list()

    Enum.each(other_nodes, fn target_node ->
      :erpc.cast(target_node, __MODULE__, :handle_replication_message, [
        message,
        state.cache_name
      ])
    end)
  end
end