super_cache

defmodule SuperCache.Distributed.Stack do
  @moduledoc """
  Cluster-aware LIFO stack.

  Structural mutations (`push`, `pop`, `get_all`) are routed to the primary
  node for the partition that owns `stack_name`.  Reads (`count`) are served
  from the local node by default and support `:read_mode`.

  Write replication mode is controlled by the cluster-wide `:replication_mode`
  setting (see `SuperCache.Cluster.Bootstrap`):

  | Mode      | Guarantee              | Extra latency    |
  |-----------|------------------------|------------------|
  | `:async`  | Eventual (default)     | None             |
  | `:sync`   | At-least-once delivery | +1 RTT per write |
  | `:strong` | Three-phase commit     | +3 RTTs per write|

  ## Design note — 3PC and local-state ordering

  In `:strong` mode, `ThreePhaseCommit.commit/2` applies all ops to the
  primary last (after replicas have acknowledged), so mutating functions
  must NOT write to local Storage before calling `apply_write/3`.  The
  counter is therefore read non-destructively (`Storage.get` instead of
  `Storage.take`), ops are collected into a list, and `apply_write/3`
  atomically applies them everywhere.  The `:updating` lock key is a
  coordinator-only primitive and is never included in replicated ops.
  """

  alias SuperCache.{Storage, Partition}
  alias SuperCache.Cluster.{Manager, Replicator, ThreePhaseCommit}
  require Logger

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

  @doc "Push `value` onto `stack_name`. Routed to primary."
  @spec push(any, any) :: true
  def push(stack_name, value) do
    route(stack_name, :local_push, [stack_name, value])
  end

  @doc "Pop and return the top value. Routed to primary."
  @spec pop(any, any) :: any
  def pop(stack_name, default \\ nil) do
    route(stack_name, :local_pop, [stack_name, default])
  end

  @doc """
  Return the number of items.

  ## Options

  - `:read_mode` — `:local` (default), `:primary`, or `:quorum`.
  """
  @spec count(any, keyword) :: non_neg_integer
  def count(stack_name, opts \\ []) do
    route_read(stack_name, :local_count, [stack_name], opts)
  end

  @doc "Return all items top-first and clear the stack. Routed to primary."
  @spec get_all(any) :: list
  def get_all(stack_name) do
    route(stack_name, :local_get_all, [stack_name])
  end

  ## Remote entry points — writes ───────────────────────────────────────────────

  @doc false
  def local_push(stack_name, value) do
    part = Partition.get_partition(stack_name)
    stack_push(part, stack_name, value)
  end

  @doc false
  def local_pop(stack_name, default) do
    part = Partition.get_partition(stack_name)
    stack_pop(part, stack_name, default)
  end

  @doc false
  def local_get_all(stack_name) do
    part = Partition.get_partition(stack_name)
    stack_drain(part, stack_name)
  end

  ## Remote entry points — reads ────────────────────────────────────────────────

  @doc false
  def local_count(stack_name) do
    part = Partition.get_partition(stack_name)

    case Storage.get({:stack, :counter, stack_name}, part) do
      [] -> 0
      [{_, count}] -> count
    end
  end

  ## Private — push ────────────────────────────────────────────────────────────
  #
  # Reads the counter with non-destructive Storage.get so that apply_write/3
  # in :strong mode can hand all ops to ThreePhaseCommit, which applies them
  # to the primary last (after replicas ACK).

  defp stack_push(partition, stack_name, value) do
    case Storage.get({:stack, :counter, stack_name}, partition) do
      [] ->
        case Storage.get({:stack, :updating, stack_name}, partition) do
          [] ->
            stack_init(stack_name)
            stack_push(partition, stack_name, value)

          _ ->
            :erlang.yield()
            stack_push(partition, stack_name, value)
        end

      [{_, counter}] ->
        next = counter + 1
        idx = partition_idx(stack_name)

        lock(partition, stack_name)

        ops = [
          {:put, {{:stack, :counter, stack_name}, next}},
          {:put, {{:stack, stack_name, next}, value}}
        ]

        apply_write(idx, partition, ops)
        unlock(partition, stack_name)
        true
    end
  end

  ## Private — pop ─────────────────────────────────────────────────────────────

  defp stack_pop(partition, stack_name, default) do
    case Storage.get({:stack, :counter, stack_name}, partition) do
      [] ->
        case Storage.get({:stack, :updating, stack_name}, partition) do
          [] ->
            default

          _ ->
            :erlang.yield()
            stack_pop(partition, stack_name, default)
        end

      [{_, 0}] ->
        default

      [{_, counter}] ->
        lock(partition, stack_name)
        idx = partition_idx(stack_name)
        next = counter - 1

        {value, ops} =
          case Storage.get({:stack, stack_name, counter}, partition) do
            [] ->
              # Counter/item mismatch — reset to empty.
              ops = [{:put, {{:stack, :counter, stack_name}, 0}}]
              {default, ops}

            [{_, v}] ->
              ops = [
                {:delete, {:stack, stack_name, counter}},
                {:put, {{:stack, :counter, stack_name}, next}}
              ]

              {v, ops}
          end

        apply_write(idx, partition, ops)
        unlock(partition, stack_name)
        value
    end
  end

  ## Private — drain (destructive, top-first) ──────────────────────────────────

  defp stack_drain(partition, stack_name) do
    case Storage.get({:stack, :counter, stack_name}, partition) do
      [] ->
        case Storage.get({:stack, :updating, stack_name}, partition) do
          [] ->
            []

          _ ->
            :erlang.yield()
            stack_drain(partition, stack_name)
        end

      [{_, 0}] ->
        []

      [{_, counter}] ->
        lock(partition, stack_name)
        idx = partition_idx(stack_name)

        # Traverse bottom→top (1..counter), prepending each value.
        # Prepending during an ascending traversal produces a top-first list
        # [counter, counter-1, …, 1] without a separate Enum.reverse/1 call.
        {values, delete_ops} =
          Enum.reduce(1..counter, {[], []}, fn x, {vals, ops} ->
            case Storage.get({:stack, stack_name, x}, partition) do
              [] -> {vals, ops}
              [{_, v}] -> {[v | vals], [{:delete, {:stack, stack_name, x}} | ops]}
            end
          end)

        reset_ops = [{:put, {{:stack, :counter, stack_name}, 0}}]

        apply_write(idx, partition, delete_ops ++ reset_ops)
        unlock(partition, stack_name)

        values
    end
  end

  ## Private — init / lock ──────────────────────────────────────────────────────

  defp stack_init(stack_name) do
    partition = Partition.get_partition(stack_name)
    idx = partition_idx(stack_name)
    apply_write(idx, partition, [{:put, {{:stack, :counter, stack_name}, 0}}])
  end

  defp lock(partition, stack_name) do
    Storage.put({{:stack, :updating, stack_name}, true}, partition)
  end

  defp unlock(partition, stack_name) do
    Storage.delete({:stack, :updating, stack_name}, partition)
  end

  ## Private — routing ──────────────────────────────────────────────────────────

  defp route(stack_name, fun, args) do
    {primary, _} = Manager.get_replicas(partition_idx(stack_name))

    if primary == node() do
      apply(__MODULE__, fun, args)
    else
      Logger.debug(fn ->
        "super_cache, dist.stack #{inspect(stack_name)}, fwd #{fun} → #{inspect(primary)}"
      end)

      :erpc.call(primary, __MODULE__, fun, args, 5_000)
    end
  end

  # Dispatch to 3PC or async/sync replication depending on replication_mode.
  # In :strong mode ThreePhaseCommit.commit/2 handles local apply + replication.
  # In other modes, Storage is written locally first then replicated.
  defp apply_write(idx, partition, ops) do
    case Manager.replication_mode() do
      :strong ->
        case ThreePhaseCommit.commit(idx, ops) do
          :ok ->
            :ok

          {:error, reason} ->
            Logger.error("super_cache, dist.stack, 3pc failed: #{inspect(reason)}")
            {:error, reason}
        end

      _ ->
        Enum.each(ops, fn
          {:put, record} ->
            Storage.put(record, partition)
            Replicator.replicate(idx, :put, record)

          {:delete, key} ->
            Storage.delete(key, partition)
            Replicator.replicate(idx, :delete, key)

          {:delete_match, pattern} ->
            Storage.delete_match(pattern, partition)
            Replicator.replicate(idx, :delete_match, pattern)

          {:delete_all, _} ->
            Storage.delete_all(partition)
            Replicator.replicate(idx, :delete_all, nil)
        end)

        :ok
    end
  end

  defp partition_idx(stack_name), do: Partition.get_partition_order(stack_name)

  ## Private — read routing ─────────────────────────────────────────────────────

  defp route_read(stack_name, fun, args, opts) do
    mode = Keyword.get(opts, :read_mode, :local)

    effective_mode =
      if mode == :local and not has_partition?(stack_name), do: :primary, else: mode

    case effective_mode do
      :local -> apply(__MODULE__, fun, args)
      :primary -> route_read_primary(stack_name, fun, args)
      :quorum -> route_read_quorum(stack_name, fun, args)
    end
  end

  defp has_partition?(stack_name) do
    idx = partition_idx(stack_name)
    {primary, replicas} = Manager.get_replicas(idx)
    node() in [primary | replicas]
  end

  defp route_read_primary(stack_name, fun, args) do
    {primary, _} = Manager.get_replicas(partition_idx(stack_name))

    if primary == node() do
      apply(__MODULE__, fun, args)
    else
      Logger.debug(fn ->
        "super_cache, dist.stack #{inspect(stack_name)}, read_primary #{fun} → #{inspect(primary)}"
      end)

      :erpc.call(primary, __MODULE__, fun, args, 5_000)
    end
  end

  defp route_read_quorum(stack_name, fun, args) do
    {primary, replicas} = Manager.get_replicas(partition_idx(stack_name))

    results =
      [primary | replicas]
      |> Task.async_stream(
        fn
          n when n == node() -> apply(__MODULE__, fun, args)
          n -> :erpc.call(n, __MODULE__, fun, args, 5_000)
        end,
        timeout: 5_000,
        on_timeout: :kill_task
      )
      |> Enum.flat_map(fn
        {:ok, result} -> [result]
        _ -> []
      end)

    majority = div(length(results), 2) + 1

    case Enum.find(Enum.frequencies(results), fn {_, c} -> c >= majority end) do
      {result, _} ->
        result

      nil ->
        if primary == node(),
          do: apply(__MODULE__, fun, args),
          else: :erpc.call(primary, __MODULE__, fun, args, 5_000)
    end
  end
end