aerospike_driver

defmodule Aerospike.Cluster.Router do
  @moduledoc false

  alias Aerospike.Cluster.PartitionMap
  alias Aerospike.Cluster.PartitionMap.PartitionOwners
  alias Aerospike.Key

  @type replica_policy :: :master | :sequence
  @type node_name :: String.t()
  @type reason :: :cluster_not_ready | :no_master

  @typedoc """
  ETS table names published through `Aerospike.Cluster.tables/1`. The router reads
  `:owners` for partition ownership and `:meta` for the `:ready` flag.
  `:node_gens` is accepted for API symmetry with the cluster tables map and is currently
  unused.
  """
  @type tables :: %{owners: atom(), node_gens: atom(), meta: atom(), txn_tracking: atom()}

  @doc """
  Returns the master node name for writing `key`.

  Refuses with `{:error, :cluster_not_ready}` while the cluster has not
  filled every configured namespace's partition map, and with
  `{:error, :no_master}` if the specific partition has no master replica.
  """
  @spec pick_for_write(tables(), Key.t()) :: {:ok, node_name()} | {:error, reason()}
  def pick_for_write(tables, %Key{} = key) do
    if ready?(tables) do
      pick_for_write_ready(tables, key)
    else
      {:error, :cluster_not_ready}
    end
  end

  @doc """
  Returns the node name for reading `key` under `policy` with the
  caller-supplied `attempt` counter.
  """
  @spec pick_for_read(tables(), Key.t(), replica_policy(), non_neg_integer()) ::
          {:ok, node_name()} | {:error, reason()}
  def pick_for_read(tables, %Key{} = key, policy, attempt)
      when policy in [:master, :sequence] and is_integer(attempt) and attempt >= 0 do
    if ready?(tables) do
      pick_for_read_ready(tables, key, policy, attempt)
    else
      {:error, :cluster_not_ready}
    end
  end

  @doc false
  @spec pick_for_write_ready(tables(), Key.t()) :: {:ok, node_name()} | {:error, :no_master}
  def pick_for_write_ready(%{owners: owners}, %Key{} = key) do
    resolve_master(owners, key)
  end

  @doc false
  @spec pick_for_read_ready(tables(), Key.t(), replica_policy(), non_neg_integer()) ::
          {:ok, node_name()} | {:error, :no_master}
  def pick_for_read_ready(%{owners: owners}, %Key{} = key, policy, attempt)
      when policy in [:master, :sequence] and is_integer(attempt) and attempt >= 0 do
    resolve_read(owners, key, policy, attempt)
  end

  @doc false
  @spec ready?(tables()) :: boolean()
  def ready?(%{meta: meta}) do
    case :ets.lookup(meta, :ready) do
      [{:ready, true}] -> true
      _ -> false
    end
  end

  defp resolve_master(owners, key) do
    case PartitionMap.owners(owners, key.namespace, Key.partition_id(key)) do
      {:ok, %PartitionOwners{replicas: [master | _]}} when is_binary(master) ->
        {:ok, master}

      {:ok, %PartitionOwners{}} ->
        {:error, :no_master}

      {:error, :unknown_partition} ->
        {:error, :no_master}
    end
  end

  defp resolve_read(owners, key, :master, _attempt), do: resolve_master(owners, key)

  defp resolve_read(owners, key, :sequence, attempt) do
    case PartitionMap.owners(owners, key.namespace, Key.partition_id(key)) do
      {:ok, %PartitionOwners{replicas: replicas}} ->
        pick_sequence(replicas, attempt)

      {:error, :unknown_partition} ->
        {:error, :no_master}
    end
  end

  defp pick_sequence(replicas, attempt) do
    case available_count(replicas, 0) do
      0 ->
        {:error, :no_master}

      count ->
        {:ok, pick_available(replicas, rem(attempt, count))}
    end
  end

  defp available_count([], count), do: count
  defp available_count([nil | rest], count), do: available_count(rest, count)

  defp available_count([node_name | rest], count) when is_binary(node_name),
    do: available_count(rest, count + 1)

  defp pick_available([nil | rest], index), do: pick_available(rest, index)
  defp pick_available([node_name | _rest], 0) when is_binary(node_name), do: node_name

  defp pick_available([node_name | rest], index)
       when is_binary(node_name) and index > 0 do
    pick_available(rest, index - 1)
  end
end