super_cache

# =============================================================================
# lib/super_cache/stack.ex
#
# Unified LIFO stack — absorbs SuperCache.Distributed.Stack.
# Same dual-implementation pattern as Queue.
# =============================================================================
defmodule SuperCache.Stack do
  @moduledoc """
  Named LIFO stacks backed by SuperCache ETS partitions.

  Works transparently in both **local** and **distributed** modes — the
  mode is determined by the `:cluster` option passed to `SuperCache.start!/1`.

  In distributed mode, structural mutations (`push`, `pop`, `get_all`) are
  routed to the partition's primary node.

  ## Example

      alias SuperCache.Stack

      Stack.push("history", :page_a)
      Stack.push("history", :page_b)
      Stack.pop("history")          # => :page_b
      Stack.get_all("history")      # => [:page_a]
  """

  require Logger
  require SuperCache.Log

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

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

  @doc "Push `value` onto `stack_name`. Creates the stack if it does not exist."
  @spec push(any, any) :: true
  def push(stack_name, value) do
    if distributed?() do
      route_write(stack_name, :dist_push, [stack_name, value])
    else
      local_push(Partition.get_partition(stack_name), stack_name, value)
    end
  end

  @doc "Pop and return the top value. Returns `default` (`nil`) when empty."
  @spec pop(any, any) :: any
  def pop(stack_name, default \\ nil) do
    if distributed?() do
      route_write(stack_name, :dist_pop, [stack_name, default])
    else
      local_pop(Partition.get_partition(stack_name), stack_name, default)
    end
  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
    if distributed?() do
      route_read(stack_name, :dist_count, [stack_name], opts)
    else
      local_count(stack_name)
    end
  end

  @doc "Drain all items top-first. Returns `[]` for an empty stack."
  @spec get_all(any) :: list
  def get_all(stack_name) do
    if distributed?() do
      route_write(stack_name, :dist_get_all, [stack_name])
    else
      local_drain(Partition.get_partition(stack_name), stack_name)
    end
  end

  ## ── Remote entry points (distributed primary, called via :erpc) ─────────────

  @doc false
  def dist_push(stack_name, value) do
    dist_do_push(Partition.get_partition(stack_name), stack_name, value)
  end

  @doc false
  def dist_pop(stack_name, default) do
    dist_do_pop(Partition.get_partition(stack_name), stack_name, default)
  end

  @doc false
  def dist_count(stack_name), do: local_count(stack_name)

  @doc false
  def dist_get_all(stack_name) do
    dist_do_drain(Partition.get_partition(stack_name), stack_name)
  end

  ## ── Private — local mode (Storage.take for atomicity) ───────────────────────

  defp local_push(partition, stack_name, value) do
    case Storage.take({:stack, :counter, stack_name}, partition) do
      [] ->
        case Storage.get({:stack, :updating, stack_name}, partition) do
          [] -> local_init(stack_name); local_push(partition, stack_name, value)
          _  -> :erlang.yield();        local_push(partition, stack_name, value)
        end
      [{_, counter}] ->
        next = counter + 1
        Storage.put({{:stack, :updating, stack_name}, true}, partition)
        Storage.put({{:stack, :counter, stack_name}, next}, partition)
        Storage.put({{:stack, stack_name, next}, value}, partition)
        Storage.delete({:stack, :updating, stack_name}, partition)
        true
    end
  end

  defp local_pop(partition, stack_name, default) do
    case Storage.take({:stack, :counter, stack_name}, partition) do
      [] ->
        case Storage.get({:stack, :updating, stack_name}, partition) do
          [] -> default
          _  -> :erlang.yield(); local_pop(partition, stack_name, default)
        end
      [{_, 0}] -> default
      [{_, counter}] ->
        Storage.put({{:stack, :updating, stack_name}, true}, partition)
        value = case Storage.take({:stack, stack_name, counter}, partition) do
          []       -> Storage.put({{:stack, :counter, stack_name}, 0}, partition); default
          [{_, v}] -> Storage.put({{:stack, :counter, stack_name}, counter - 1}, partition); v
        end
        Storage.delete({:stack, :updating, stack_name}, partition)
        value
    end
  end

  defp local_drain(partition, stack_name) do
    case Storage.take({:stack, :counter, stack_name}, partition) do
      [] ->
        case Storage.get({:stack, :updating, stack_name}, partition) do
          [] -> []
          _  -> :erlang.yield(); local_drain(partition, stack_name)
        end
      [{_, 0}] -> []
      [{_, counter}] ->
        Storage.put({{:stack, :updating, stack_name}, true}, partition)
        values = Enum.reduce(counter..1//-1, [], fn x, acc ->
          case Storage.take({:stack, stack_name, x}, partition) do
            []       -> acc
            [{_, v}] -> [v | acc]
          end
        end)
        Storage.put({{:stack, :counter, stack_name}, 0}, partition)
        Storage.delete({:stack, :updating, stack_name}, partition)
        values
    end
  end

  defp local_init(stack_name) do
    part = Partition.get_partition(stack_name)
    Storage.put({{:stack, :counter, stack_name}, 0}, part)
  end

  defp local_count(stack_name) do
    part = Partition.get_partition(stack_name)
    case Storage.get({:stack, :counter, stack_name}, part) do
      []             -> 0
      [{_, counter}] -> counter
    end
  end

  ## ── Private — distributed mode (Storage.get + ops list for 3PC) ─────────────

  defp dist_do_push(partition, stack_name, value) do
    case Storage.get({:stack, :counter, stack_name}, partition) do
      [] ->
        case Storage.get({:stack, :updating, stack_name}, partition) do
          [] -> dist_init(stack_name); dist_do_push(partition, stack_name, value)
          _  -> :erlang.yield();       dist_do_push(partition, stack_name, value)
        end
      [{_, counter}] ->
        next = counter + 1
        lock(partition, stack_name)
        ops = [{:put, {{:stack, :counter, stack_name}, next}},
               {:put, {{:stack, stack_name, next}, value}}]
        apply_write(idx(stack_name), partition, ops)
        unlock(partition, stack_name)
        true
    end
  end

  defp dist_do_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(); dist_do_pop(partition, stack_name, default)
        end
      [{_, 0}] -> default
      [{_, counter}] ->
        lock(partition, stack_name)
        {value, ops} = case Storage.get({:stack, stack_name, counter}, partition) do
          [] ->
            {default, [{:put, {{:stack, :counter, stack_name}, 0}}]}
          [{_, v}] ->
            {v, [{:delete, {:stack, stack_name, counter}},
                 {:put, {{:stack, :counter, stack_name}, counter - 1}}]}
        end
        apply_write(idx(stack_name), partition, ops)
        unlock(partition, stack_name)
        value
    end
  end

  defp dist_do_drain(partition, stack_name) do
    case Storage.get({:stack, :counter, stack_name}, partition) do
      []        -> []
      [{_, 0}]  -> []
      [{_, counter}] ->
        lock(partition, stack_name)
        {values, del_ops} = Enum.reduce(1..counter, {[], []}, fn x, {vs, ops} ->
          case Storage.get({:stack, stack_name, x}, partition) do
            []       -> {vs, ops}
            [{_, v}] -> {[v | vs], [{:delete, {:stack, stack_name, x}} | ops]}
          end
        end)
        reset_ops = [{:put, {{:stack, :counter, stack_name}, 0}}]
        apply_write(idx(stack_name), partition, del_ops ++ reset_ops)
        unlock(partition, stack_name)
        values
    end
  end

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

  ## ── Private — shared helpers ─────────────────────────────────────────────────

  defp lock(partition, name),   do: Storage.put({{:stack, :updating, name}, true}, partition)
  defp unlock(partition, name), do: Storage.delete({:stack, :updating, name}, partition)

  defp distributed?(), do: Config.get_config(:cluster, :local) == :distributed
  defp idx(name),      do: Partition.get_partition_order(name)

  defp apply_write(idx, partition, ops) do
    case Manager.replication_mode() do
      :strong ->
        case ThreePhaseCommit.commit(idx, ops) do
          :ok         -> :ok
          {:error, r} -> Logger.error("super_cache, stack, 3pc failed: #{inspect(r)}"); {:error, r}
        end
      _ ->
        Enum.each(ops, fn
          {:put, r}          -> Storage.put(r, partition);          Replicator.replicate(idx, :put, r)
          {:delete, k}       -> Storage.delete(k, partition);       Replicator.replicate(idx, :delete, k)
          {:delete_match, p} -> Storage.delete_match(p, partition); Replicator.replicate(idx, :delete_match, p)
          {:delete_all, _}   -> Storage.delete_all(partition);      Replicator.replicate(idx, :delete_all, nil)
        end)
        :ok
    end
  end

  defp route_write(stack_name, fun, args) do
    {primary, _} = Manager.get_replicas(idx(stack_name))
    if primary == node() do
      apply(__MODULE__, fun, args)
    else
      SuperCache.Log.debug(fn -> "super_cache, stack #{inspect(stack_name)}, fwd #{fun} → #{inspect(primary)}" end)
      :erpc.call(primary, __MODULE__, fun, args, 5_000)
    end
  end

  defp route_read(stack_name, fun, args, opts) do
    mode = Keyword.get(opts, :read_mode, :local)
    eff  = if mode == :local and not has_partition?(stack_name), do: :primary, else: mode
    case eff do
      :local   -> apply(__MODULE__, fun, args)
      :primary ->
        {primary, _} = Manager.get_replicas(idx(stack_name))
        if primary == node(), do: apply(__MODULE__, fun, args),
        else: :erpc.call(primary, __MODULE__, fun, args, 5_000)
      :quorum ->
        {primary, replicas} = Manager.get_replicas(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, r} -> [r]; _ -> [] 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

  defp has_partition?(name) do
    {p, rs} = Manager.get_replicas(idx(name))
    node() in [p | rs]
  end
end