aerospike_driver

defmodule Aerospike.Command.BatchRouter do
  @moduledoc false

  alias Aerospike.Cluster.Router
  alias Aerospike.Command.BatchCommand.Entry
  alias Aerospike.Command.BatchCommand.NodeRequest
  alias Aerospike.Key

  defmodule RoutingFailure do
    @moduledoc false

    @enforce_keys [:entry, :reason]
    defstruct [:entry, :reason]

    @type t :: %__MODULE__{
            entry: Entry.t(),
            reason: Router.reason()
          }
  end

  defmodule Grouping do
    @moduledoc false

    @enforce_keys [:node_requests, :routing_failures]
    defstruct [:node_requests, :routing_failures]

    @type t :: %__MODULE__{
            node_requests: [NodeRequest.t()],
            routing_failures: [RoutingFailure.t()]
          }
  end

  @type payload_fun :: (Key.t(), non_neg_integer() -> term())
  @type reason :: :cluster_not_ready

  @doc """
  Groups typed batch entries into per-node requests while preserving input
  indices for later merge ordering.
  """
  @spec group_entries(Router.tables(), [Entry.t()]) :: {:ok, Grouping.t()} | {:error, reason()}
  def group_entries(tables, entries) when is_list(entries) do
    case reduce_entries(entries, tables) do
      {:ok, grouped, node_order, routing_failures} ->
        {:ok,
         %Grouping{
           node_requests: build_node_requests(grouped, node_order),
           routing_failures: Enum.reverse(routing_failures)
         }}

      {:error, :cluster_not_ready} = err ->
        err
    end
  end

  @doc """
  Groups `keys` into per-node batch requests while preserving input indices.

  Global readiness refusal (`:cluster_not_ready`) aborts the whole grouping
  because no key can be routed honestly against an incomplete partition view.
  Partition-local misses (`:no_master`) are retained as indexed failures so a
  later merge step can surface mixed outcomes deterministically.
  """
  @spec group_keys(Router.tables(), [Key.t()], keyword()) ::
          {:ok, Grouping.t()} | {:error, reason()}
  def group_keys(tables, keys, opts \\ []) when is_list(keys) and is_list(opts) do
    dispatch = Keyword.get(opts, :dispatch, {:read, :master, 0})
    payload_fun = Keyword.get(opts, :payload_fun, fn _key, _index -> nil end)
    kind = Keyword.get(opts, :kind, default_kind(dispatch))

    entries =
      keys
      |> Enum.with_index()
      |> Enum.map(fn {key, index} ->
        %Entry{
          index: index,
          key: key,
          kind: kind,
          dispatch: dispatch,
          payload: payload_fun.(key, index)
        }
      end)

    group_entries(tables, entries)
  end

  defp reduce_entries([], _tables), do: {:ok, %{}, [], []}

  defp reduce_entries(entries, tables) do
    if Router.ready?(tables),
      do: reduce_ready_entries(entries, tables),
      else: {:error, :cluster_not_ready}
  end

  defp reduce_ready_entries(entries, tables) do
    Enum.reduce(entries, {:ok, %{}, [], []}, &reduce_ready_entry(&1, &2, tables))
  end

  defp reduce_ready_entry(%Entry{} = entry, {:ok, grouped, node_order, failures}, tables) do
    case route_entry_ready(tables, entry) do
      {:ok, node_name} ->
        {next_grouped, next_node_order} = put_entry(grouped, node_order, node_name, entry)
        {:ok, next_grouped, next_node_order, failures}

      {:error, reason} ->
        failure = %RoutingFailure{entry: entry, reason: reason}
        {:ok, grouped, node_order, [failure | failures]}
    end
  end

  defp put_entry(grouped, node_order, node_name, entry) do
    case grouped do
      %{^node_name => entries} ->
        {Map.put(grouped, node_name, [entry | entries]), node_order}

      %{} ->
        {Map.put(grouped, node_name, [entry]), [node_name | node_order]}
    end
  end

  defp build_node_requests(grouped, node_order) do
    node_order
    |> Enum.reverse()
    |> Enum.map(fn node_name ->
      %NodeRequest{
        node_name: node_name,
        entries: grouped |> Map.fetch!(node_name) |> Enum.reverse(),
        payload: nil
      }
    end)
  end

  defp route_entry_ready(tables, %Entry{key: %Key{} = key, dispatch: :write}) do
    Router.pick_for_write_ready(tables, key)
  end

  defp route_entry_ready(tables, %Entry{
         key: %Key{} = key,
         dispatch: {:read, replica_policy, attempt}
       }) do
    Router.pick_for_read_ready(tables, key, replica_policy, attempt)
  end

  defp default_kind(:write), do: :put
  defp default_kind({:read, _, _}), do: :read
end