super_cache

defmodule SuperCache.Cluster.Replicator do
  @moduledoc """
  Applies replicated writes on the local node and handles bulk partition
  transfers when a new node joins or a full sync is requested.

  ## Replication Modes

  - `:async` — Fire-and-forget via a lightweight worker pool (`Task.Supervisor`).
    Eliminates per-operation `spawn/1` overhead.
  - `:sync` — Adaptive quorum writes. Returns `:ok` once a strict majority of
    replicas acknowledge, avoiding waits for slow stragglers.
  - `:strong` — Write-Ahead Log (WAL) based consistency. Replaces heavy 3PC
    with fast local write + async replication + majority ack (~200µs vs ~1500µs).

  Every call to `replicate/3` increments the appropriate
  `replicate_<mode>` counter in `SuperCache.Cluster.Metrics` and records
  a latency sample so the results are visible in
  `SuperCache.Cluster.Stats.api/0`.
  """

  require Logger
  require SuperCache.Log

  alias SuperCache.{Storage, Partition}
  alias SuperCache.Cluster.{Manager, WAL, Metrics}

  @batch_size 500
  @pool_name __MODULE__.Pool

  # ── Public API ───────────────────────────────────────────────────────────────

  @spec replicate(non_neg_integer, atom, any) :: :ok | {:error, term}
  def replicate(partition_idx, op_name, op_arg \\ nil) do
    mode = SuperCache.Config.get_config(:replication_mode, :async)
    {_primary, replicas} = Manager.get_replicas(partition_idx)

    if replicas == [] do
      :ok
    else
      metric_key = :"replicate_#{mode}"
      t0 = System.monotonic_time(:microsecond)

      result =
        try do
          do_replicate(mode, partition_idx, op_name, op_arg, replicas)
        rescue
          err -> {:error, err}
        end

      elapsed = System.monotonic_time(:microsecond) - t0
      Metrics.increment({:api, metric_key}, :calls)
      Metrics.push_latency({:api_latency_us, metric_key}, elapsed)

      case result do
        {:error, _} -> Metrics.increment({:api, metric_key}, :errors)
        _ -> :ok
      end

      result
    end
  end

  @doc """
  Apply a replicated operation on this node.

  Called via `:erpc` from the primary or worker pool — **do NOT call directly**.
  """
  @spec apply_op(non_neg_integer, atom, any) :: :ok
  def apply_op(partition_idx, op_name, op_arg) do
    partition = Partition.get_partition_by_idx(partition_idx)

    case {op_name, op_arg} do
      {:put, records} when is_list(records) -> Storage.put(records, partition)
      {:put, record} -> Storage.put(record, partition)
      {:delete, key} -> Storage.delete(key, partition)
      {:delete_match, pattern} -> Storage.delete_match(pattern, partition)
      {:delete_all, _} -> Storage.delete_all(partition)
    end

    :ok
  end

  @doc """
  Push all records for `partition_idx` to `target_node` in batches.
  """
  @spec push_partition(non_neg_integer, node) :: :ok
  def push_partition(partition_idx, target_node) do
    partition = Partition.get_partition_by_idx(partition_idx)

    Logger.info(
      "super_cache, replicator, syncing partition #{partition_idx} → #{inspect(target_node)}"
    )

    records = Storage.scan(fn rec, acc -> [rec | acc] end, [], partition)
    total = length(records)

    records
    |> Enum.chunk_every(@batch_size)
    |> Enum.with_index(1)
    |> Enum.each(fn {batch, batch_num} ->
      try do
        :erpc.call(target_node, __MODULE__, :apply_op, [partition_idx, :put, batch], 10_000)

        SuperCache.Log.debug(fn ->
          "super_cache, replicator, partition #{partition_idx} batch #{batch_num} " <>
            "(#{length(batch)} records) → #{inspect(target_node)}"
        end)
      catch
        kind, reason ->
          Logger.warning(
            "super_cache, replicator, batch #{batch_num} failed " <>
              "→ #{inspect(target_node)}: #{inspect({kind, reason})}"
          )
      end
    end)

    Logger.info(
      "super_cache, replicator, pushed #{total} record(s) for partition " <>
        "#{partition_idx} → #{inspect(target_node)}"
    )

    :ok
  end

  @doc """
  Replicate a batch of operations to all replicas in a single `:erpc` call.

  Dramatically reduces network overhead for bulk writes by sending all
  operations in one message instead of one message per operation.
  """
  @spec replicate_batch(non_neg_integer, atom, [any]) :: :ok | {:error, term}
  def replicate_batch(partition_idx, op_name, op_args) when is_list(op_args) do
    {_primary, replicas} = Manager.get_replicas(partition_idx)

    if replicas == [] do
      :ok
    else
      t0 = System.monotonic_time(:microsecond)

      result =
        try do
          :erpc.multicall(
            replicas,
            __MODULE__,
            :apply_op_batch,
            [partition_idx, op_name, op_args],
            10_000
          )

          :ok
        rescue
          err -> {:error, err}
        end

      elapsed = System.monotonic_time(:microsecond) - t0
      metric_key = :"replicate_batch_#{SuperCache.Config.get_config(:replication_mode, :async)}"
      Metrics.increment({:api, metric_key}, :calls)
      Metrics.push_latency({:api_latency_us, metric_key}, elapsed)

      case result do
        {:error, _} -> Metrics.increment({:api, metric_key}, :errors)
        _ -> :ok
      end

      result
    end
  end

  @doc """
  Apply a batch of replicated operations on this node.

  Called via `:erpc` from the primary — **do NOT call directly**.
  """
  @spec apply_op_batch(non_neg_integer, atom, [any]) :: :ok
  def apply_op_batch(partition_idx, op_name, op_args) do
    partition = Partition.get_partition_by_idx(partition_idx)

    Enum.each(op_args, fn op_arg ->
      case {op_name, op_arg} do
        {:put, {key, record}} -> Storage.put({key, record}, partition)
        {:put, record} when is_tuple(record) -> Storage.put(record, partition)
        {:delete, key} -> Storage.delete(key, partition)
        {:delete_match, pattern} -> Storage.delete_match(pattern, partition)
        {:delete_all, _} -> Storage.delete_all(partition)
      end
    end)

    :ok
  end

  # ── Private ──────────────────────────────────────────────────────────────────

  defp do_replicate(:async, partition_idx, op_name, op_arg, replicas) do
    # Worker pool: uses Task.Supervisor to avoid per-operation spawn/1 overhead.
    # Fire-and-forget: returns immediately after handing off to the pool.
    # async_nolink prevents exit message leakage to the supervisor.
    ensure_pool()
    Task.Supervisor.async_nolink(@pool_name, fn ->
      :erpc.multicall(replicas, __MODULE__, :apply_op, [partition_idx, op_name, op_arg], 5_000)
    end)
    :ok
  end

  defp do_replicate(:sync, partition_idx, op_name, op_arg, replicas) do
    # Adaptive quorum: returns :ok once a strict majority acknowledges.
    # Avoids waiting for slow stragglers while maintaining strong durability.
    required = div(length(replicas), 2) + 1
    tasks = Enum.map(replicas, fn replica ->
      Task.async(fn ->
        :erpc.call(replica, __MODULE__, :apply_op, [partition_idx, op_name, op_arg], 5_000)
      end)
    end)

    await_quorum(tasks, required)
  end

  defp do_replicate(:strong, partition_idx, op_name, op_arg, _replicas) do
    # WAL-based strong consistency: replaces heavy 3PC.
    # ~200µs latency vs ~1500µs for 3PC.
    WAL.commit(partition_idx, [{op_name, op_arg}])
  end

  defp ensure_pool() do
    unless Process.whereis(@pool_name) do
      Task.Supervisor.start_link(name: @pool_name, max_restarts: 5, max_seconds: 10)
    end
  end

  defp await_quorum(tasks, required) do
    await_quorum(tasks, required, 0)
  end

  defp await_quorum(_tasks, required, success) when success >= required, do: :ok
  defp await_quorum([], _required, _success), do: {:error, :quorum_not_reached}
  defp await_quorum(tasks, required, success) do
    case Task.yield_many(tasks, 5_000) do
      [] -> {:error, :quorum_timeout}
      results ->
        finished = for {t, res} <- results, res != nil, do: t
        remaining = tasks -- finished
        new_success = Enum.count(results, fn {_, {:ok, :ok}} -> true; _ -> false end)
        await_quorum(remaining, required, success + new_success)
    end
  end
end