aerospike_driver

defmodule Aerospike.Cluster.PartitionMapWriter do
  @moduledoc false

  use GenServer

  alias Aerospike.Cluster.PartitionMap
  alias Aerospike.Cluster.PartitionMapMerge
  alias Aerospike.RetryPolicy

  @type tables :: %{
          optional(atom()) => atom(),
          owners: atom(),
          node_gens: atom(),
          meta: atom()
        }

  @type option :: {:name, atom()} | {:tables, tables()}

  @doc false
  def child_spec(opts) do
    name = Keyword.fetch!(opts, :name)

    %{
      id: {__MODULE__, name},
      start: {__MODULE__, :start_link, [opts]},
      type: :worker,
      restart: :permanent,
      shutdown: 5_000
    }
  end

  @doc """
  Starts the writer.

  Options:

    * `:name` — cluster name atom. The writer registers itself under
      `via/1`.
    * `:tables` — `%{owners: atom(), node_gens: atom(), meta: atom()}`
      map of ETS table names as returned by
      `Aerospike.Cluster.TableOwner.tables/1`.
  """
  @spec start_link([option()]) :: GenServer.on_start()
  def start_link(opts) do
    name = Keyword.fetch!(opts, :name)
    GenServer.start_link(__MODULE__, opts, name: via(name))
  end

  @doc """
  Returns the registered name atom used by `start_link/1` for `name`.
  """
  @spec via(atom()) :: atom()
  def via(name) when is_atom(name), do: :"#{name}_partition_map_writer"

  @doc """
  Applies `segments` reported by `node_name` against the `owners` table
  via `Aerospike.Cluster.PartitionMapMerge.apply_segments/4`.

  Returns `true` when every in-scope segment was accepted by the regime
  guard, `false` when any segment was rejected as stale. The caller uses
  that boolean to decide whether to advance the node's `applied_gen`.
  """
  @spec apply_segments(
          GenServer.server(),
          PartitionMap.node_name(),
          [PartitionMapMerge.segment()],
          [PartitionMap.namespace()]
        ) :: boolean()
  def apply_segments(server, node_name, segments, namespaces)
      when is_binary(node_name) and is_list(segments) and is_list(namespaces) do
    GenServer.call(server, {:apply_segments, node_name, segments, namespaces})
  end

  @doc """
  Stores the latest `partition-generation` value for `node_name`.
  """
  @spec put_node_gen(GenServer.server(), PartitionMap.node_name(), non_neg_integer()) :: :ok
  def put_node_gen(server, node_name, gen)
      when is_binary(node_name) and is_integer(gen) and gen >= 0 do
    GenServer.call(server, {:put_node_gen, node_name, gen})
  end

  @doc """
  Removes every `owners` entry that lists `node_name` as a replica.
  """
  @spec drop_node(GenServer.server(), PartitionMap.node_name()) :: :ok
  def drop_node(server, node_name) when is_binary(node_name) do
    GenServer.call(server, {:drop_node, node_name})
  end

  @doc """
  Removes the per-node `partition-generation` entry for `node_name`.
  """
  @spec delete_node_gen(GenServer.server(), PartitionMap.node_name()) :: :ok
  def delete_node_gen(server, node_name) when is_binary(node_name) do
    GenServer.call(server, {:delete_node_gen, node_name})
  end

  @doc """
  Recomputes the `meta.:ready` flag from the current `owners` table.

  `:ready` flips to `true` only when every configured namespace has a
  complete partition map (`PartitionMap.complete?/2`). Returns the new
  ready value so callers can cache it without re-reading the table.
  """
  @spec recompute_ready(GenServer.server(), [PartitionMap.namespace(), ...]) :: boolean()
  def recompute_ready(server, namespaces) when is_list(namespaces) and namespaces != [] do
    GenServer.call(server, {:recompute_ready, namespaces})
  end

  @doc """
  Publishes the cluster-default retry policy to `meta.:retry_opts`.
  """
  @spec publish_retry_policy(GenServer.server(), RetryPolicy.t()) :: :ok
  def publish_retry_policy(
        server,
        %{
          max_retries: _,
          sleep_between_retries_ms: _,
          replica_policy: _
        } = policy
      ) do
    GenServer.call(server, {:publish_retry_policy, policy})
  end

  @doc """
  Publishes the sorted active node-name snapshot to `meta.:active_nodes`.
  """
  @spec publish_active_nodes(GenServer.server(), [PartitionMap.node_name()]) :: :ok
  def publish_active_nodes(server, node_names) when is_list(node_names) do
    GenServer.call(server, {:publish_active_nodes, node_names})
  end

  @impl GenServer
  def init(opts) do
    name = Keyword.fetch!(opts, :name)
    tables = Keyword.fetch!(opts, :tables)

    is_atom(name) or
      raise ArgumentError, "Aerospike.Cluster.PartitionMapWriter: :name must be an atom"

    %{owners: owners_tab, node_gens: node_gens_tab, meta: meta_tab} = tables

    {:ok,
     %{
       name: name,
       owners_tab: owners_tab,
       node_gens_tab: node_gens_tab,
       meta_tab: meta_tab
     }}
  end

  @impl GenServer
  def handle_call({:apply_segments, node_name, segments, namespaces}, _from, state) do
    applied? = PartitionMapMerge.apply_segments(state.owners_tab, node_name, segments, namespaces)
    {:reply, applied?, state}
  end

  def handle_call({:put_node_gen, node_name, gen}, _from, state) do
    :ok = PartitionMap.put_node_gen(state.node_gens_tab, node_name, gen)
    {:reply, :ok, state}
  end

  def handle_call({:drop_node, node_name}, _from, state) do
    :ok = PartitionMap.drop_node(state.owners_tab, node_name)
    {:reply, :ok, state}
  end

  def handle_call({:delete_node_gen, node_name}, _from, state) do
    :ok = PartitionMap.delete_node_gen(state.node_gens_tab, node_name)
    {:reply, :ok, state}
  end

  def handle_call({:recompute_ready, namespaces}, _from, state) do
    ready? =
      Enum.all?(namespaces, fn namespace ->
        PartitionMap.complete?(state.owners_tab, namespace)
      end)

    :ets.insert(state.meta_tab, {:ready, ready?})
    {:reply, ready?, state}
  end

  def handle_call({:publish_retry_policy, policy}, _from, state) do
    true = RetryPolicy.put(state.meta_tab, policy)
    {:reply, :ok, state}
  end

  def handle_call({:publish_active_nodes, node_names}, _from, state) do
    snapshot =
      node_names
      |> Enum.uniq()
      |> Enum.sort()

    :ets.insert(state.meta_tab, {:active_nodes, snapshot})
    {:reply, :ok, state}
  end
end