super_cache

defmodule SuperCache.KeyValue do
  @moduledoc """
  In-memory key-value namespaces 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`.

  Multiple independent namespaces coexist using different `kv_name` values.

  ## Read modes (distributed)

  Pass `read_mode: :primary` or `read_mode: :quorum` when you need to read
  your own writes from any node.

  ## Example

      alias SuperCache.KeyValue

      # Works identically in local and distributed mode
      KeyValue.add("session", :user_id, 42)
      KeyValue.get("session", :user_id)       # => 42
      KeyValue.keys("session")                # => [:user_id]
      KeyValue.remove("session", :user_id)
      KeyValue.remove_all("session")
  """

  require Logger
  require SuperCache.Log

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

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

  @spec add(any, any, any) :: true
  def add(kv_name, key, value) do
    SuperCache.Log.debug(fn -> "super_cache, kv #{inspect(kv_name)}, add key=#{inspect(key)}" end)

    if distributed?() do
      route_write(kv_name, :local_put, [kv_name, key, value])
    else
      Storage.put({{:kv, kv_name, key}, value}, Partition.get_partition(kv_name))
    end
  end

  @spec get(any, any, any, keyword) :: any
  def get(kv_name, key, default \\ nil, opts \\ []) do
    if distributed?() do
      route_read(kv_name, :local_get, [kv_name, key, default], opts)
    else
      case Storage.get({:kv, kv_name, key}, Partition.get_partition(kv_name)) do
        []           -> default
        [{_, value}] -> value
      end
    end
  end

  @spec keys(any, keyword) :: [any]
  def keys(kv_name, opts \\ []) do
    if distributed?() do
      route_read(kv_name, :local_keys, [kv_name], opts)
    else
      local_keys(kv_name)
    end
  end

  @spec values(any, keyword) :: [any]
  def values(kv_name, opts \\ []) do
    if distributed?() do
      route_read(kv_name, :local_values, [kv_name], opts)
    else
      local_values(kv_name)
    end
  end

  @spec count(any, keyword) :: non_neg_integer
  def count(kv_name, opts \\ []) do
    if distributed?() do
      route_read(kv_name, :local_count, [kv_name], opts)
    else
      local_count(kv_name)
    end
  end

  @spec to_list(any, keyword) :: [{any, any}]
  def to_list(kv_name, opts \\ []) do
    if distributed?() do
      route_read(kv_name, :local_to_list, [kv_name], opts)
    else
      local_to_list(kv_name)
    end
  end

  @spec remove(any, any) :: :ok
  def remove(kv_name, key) do
    SuperCache.Log.debug(fn -> "super_cache, kv #{inspect(kv_name)}, remove key=#{inspect(key)}" end)

    if distributed?() do
      route_write(kv_name, :local_delete, [kv_name, key])
    else
      Storage.delete({:kv, kv_name, key}, Partition.get_partition(kv_name))
      :ok
    end
  end

  @spec remove_all(any) :: :ok
  def remove_all(kv_name) do
    if distributed?() do
      route_write(kv_name, :local_delete_all, [kv_name])
    else
      SuperCache.delete_by_match!(kv_name, {{:kv, kv_name, :_}, :_})
    end
  end

  @doc """
  Add multiple key-value pairs in a single batch operation.

  Groups entries by partition and sends each group in a single `:erpc` call
  in distributed mode, dramatically reducing network overhead.

  ## Example

      KeyValue.add_batch("session", [
        {:user_1, %{name: "Alice"}},
        {:user_2, %{name: "Bob"}}
      ])
  """
  @spec add_batch(any, [{any, any}]) :: :ok
  def add_batch(kv_name, pairs) when is_list(pairs) do
    if distributed?() do
      # Group by partition and batch-write each group
      partition = Partition.get_partition(kv_name)
      records = Enum.map(pairs, fn {key, value} ->
        {{:kv, kv_name, key}, value}
      end)

      SuperCache.put_batch!(records)
    else
      Enum.each(pairs, fn {key, value} ->
        Storage.put({{:kv, kv_name, key}, value}, Partition.get_partition(kv_name))
      end)
    end

    :ok
  end

  @doc """
  Remove multiple keys in a single batch operation.

  Groups entries by partition and sends each group in a single `:erpc` call
  in distributed mode.

  ## Example

      KeyValue.remove_batch("session", [:user_1, :user_2])
  """
  @spec remove_batch(any, [any]) :: :ok
  def remove_batch(kv_name, keys) when is_list(keys) do
    if distributed?() do
      partition = Partition.get_partition(kv_name)
      # Build delete records for batch routing
      records = Enum.map(keys, fn key ->
        {:kv, kv_name, key}
      end)

      # Use Router for distributed deletes
      Enum.each(keys, fn key ->
        ets_key = {:kv, kv_name, key}
        Router.route_delete_by_key_partition!(ets_key, kv_name)
      end)
    else
      partition = Partition.get_partition(kv_name)
      Enum.each(keys, fn key ->
        Storage.delete({:kv, kv_name, key}, partition)
      end)
    end

    :ok
  end

  ## ── Remote entry points — writes (called via :erpc on primary) ──────────────

  @doc false
  def local_put(kv_name, key, value) do
    apply_write(idx(kv_name), Partition.get_partition(kv_name),
                [{:put, {{:kv, kv_name, key}, value}}])
    true
  end

  @doc false
  def local_delete(kv_name, key) do
    apply_write(idx(kv_name), Partition.get_partition(kv_name),
                [{:delete, {:kv, kv_name, key}}])
    :ok
  end

  @doc false
  def local_delete_all(kv_name) do
    apply_write(idx(kv_name), Partition.get_partition(kv_name),
                [{:delete_match, {{:kv, kv_name, :_}, :_}}])
    :ok
  end

  ## ── Remote entry points — reads (called via :erpc in quorum/primary reads) ──

  @doc false
  def local_get(kv_name, key, default) do
    case Storage.get({:kv, kv_name, key}, Partition.get_partition(kv_name)) do
      []           -> default
      [{_, value}] -> value
    end
  end

  @doc false
  def local_keys(kv_name) do
    do_match(kv_name) |> Enum.map(fn {{:kv, ^kv_name, k}, _} -> k end)
  end

  @doc false
  def local_values(kv_name) do
    do_match(kv_name) |> Enum.map(fn {{:kv, ^kv_name, _}, v} -> v end)
  end

  @doc false
  def local_count(kv_name), do: do_match(kv_name) |> length()

  @doc false
  def local_to_list(kv_name) do
    do_match(kv_name) |> Enum.map(fn {{:kv, ^kv_name, k}, v} -> {k, v} end)
  end

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

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

  defp do_match(kv_name) do
    Storage.get_by_match_object({{:kv, kv_name, :_}, :_}, Partition.get_partition(kv_name))
  end

  defp apply_write(idx, partition, ops) do
    case Manager.replication_mode() do
      :strong ->
        ThreePhaseCommit.commit(idx, ops)
      _ ->
        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(kv_name, fun, args) do
    {primary, _} = Manager.get_replicas(idx(kv_name))
    if primary == node() do
      apply(__MODULE__, fun, args)
    else
      SuperCache.Log.debug(fn -> "super_cache, kv #{inspect(kv_name)}, fwd #{fun} → #{inspect(primary)}" end)
      :erpc.call(primary, __MODULE__, fun, args, 5_000)
    end
  end

  defp route_read(kv_name, fun, args, opts) do
    mode = Keyword.get(opts, :read_mode, :local)
    eff  = if mode == :local and not has_partition?(kv_name), do: :primary, else: mode
    case eff do
      :local   -> apply(__MODULE__, fun, args)
      :primary -> read_from_primary(kv_name, fun, args)
      :quorum  -> read_from_quorum(kv_name, fun, args)
    end
  end

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

  defp read_from_primary(kv_name, fun, args) do
    {primary, _} = Manager.get_replicas(idx(kv_name))
    if primary == node(), do: apply(__MODULE__, fun, args),
    else: :erpc.call(primary, __MODULE__, fun, args, 5_000)
  end

  defp read_from_quorum(kv_name, fun, args) do
    {primary, replicas} = Manager.get_replicas(idx(kv_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