super_cache

defmodule SuperCache.Cluster.WAL do
  @moduledoc """
  Write-Ahead Log for fast strong consistency.

  Replaces the heavy Three-Phase Commit (3PC) protocol with a lighter-weight
  WAL-based approach that:

  1. Writes to local ETS immediately
  2. Appends operation to WAL (in-memory ETS for speed)
  3. Async replicates WAL entries to replicas
  4. Tracks acknowledgments from replicas
  5. Returns success once majority has acked

  This reduces strong-mode latency from ~1500µs (3PC) to ~200µs (WAL).

  ## Design

  - WAL entries are stored in an ETS table for fast access
  - Each entry has a monotonically increasing sequence number
  - Replicas ack entries asynchronously via `:erpc.cast`
  - Majority acknowledgment determines commit success
  - Periodic cleanup of committed entries
  - Recovery on node restart replays uncommitted entries

  ## Usage

  This module is called internally by the Replicator when
  `replication_mode` is set to `:strong`. You should not call it directly.

  ## Configuration

  The WAL uses sensible defaults but can be tuned via application config:

      config :super_cache, :wal,
        majority_timeout: 2_000,  # ms to wait for majority ack
        cleanup_interval: 5_000,  # ms between cleanup cycles
        max_pending: 10_000       # max uncommitted entries before backpressure

  ## Example

      # Called internally by Replicator.replicate/3
      SuperCache.Cluster.WAL.commit(2, [{:put, {:user, 1, "Alice"}}])
      # => :ok
  """

  use GenServer, restart: :permanent, shutdown: 5_000

  require Logger
  require SuperCache.Log

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

  @table __MODULE__
  @ack_table :"#{@table}_acks"
  @seq_key {__MODULE__, :seq}

  @default_majority_timeout 2_000
  @default_cleanup_interval 5_000


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

  @doc """
  Starts the WAL GenServer.
  """
  @spec start_link(keyword) :: :ignore | {:error, any} | {:ok, pid}
  def start_link(opts \\ []) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @doc """
  Commit operations via WAL.

  1. Writes to local ETS immediately
  2. Appends to WAL
  3. Async replicates to replicas
  4. Waits for majority acknowledgment
  5. Returns `:ok` on success, `{:error, reason}` on failure

  This is the fast path for strong consistency — typically ~200µs vs ~1500µs for 3PC.

  ## Example

      WAL.commit(2, [{:put, {:user, 1, "Alice"}}])
      # => :ok
  """
  @spec commit(non_neg_integer, [{atom, any}]) :: :ok | {:error, term}
  def commit(partition_idx, ops) do
    {_primary, replicas} = Manager.get_replicas(partition_idx)

    if replicas == [] do
      # No replicas — apply locally and return immediately
      apply_local(partition_idx, ops)
      :ok
    else
      t0 = System.monotonic_time(:microsecond)

      # Apply locally first (write-ahead)
      apply_local(partition_idx, ops)

      # Get next sequence number
      seq = next_seq()

      # Write to WAL
      wal_entry = %{
        seq: seq,
        partition_idx: partition_idx,
        ops: ops,
        timestamp: System.monotonic_time(:millisecond)
      }

      :ets.insert(@table, {seq, wal_entry})

      # Initialize ack tracking
      required = div(length(replicas), 2) + 1
      :ets.insert(@ack_table, {seq, %{acked: 0, required: required, replicas: replicas}})

      # Async replicate to all replicas
      async_replicate(seq, partition_idx, ops, replicas)

      # Wait for majority ack
      result = await_majority(seq, required)

      elapsed = System.monotonic_time(:microsecond) - t0

      case result do
        :ok ->
          Metrics.increment({:wal, :committed}, :calls)
          Metrics.push_latency({:wal_latency_us, :commit}, elapsed)
          :ok

        {:error, _} = err ->
          Metrics.increment({:wal, :failed}, :calls)
          Metrics.push_latency({:wal_latency_us, :commit_failed}, elapsed)
          err
      end
    end
  end

  @doc """
  Handle replication acknowledgment from a replica.

  Called via `:erpc.cast` on the primary node when a replica has applied the WAL entry.
  """
  @spec ack(non_neg_integer, node) :: :ok
  def ack(seq, _replica_node) do
    case :ets.lookup(@ack_table, seq) do
      [{^seq, %{acked: acked, required: required} = state}] ->
        new_acked = acked + 1
        :ets.insert(@ack_table, {seq, %{state | acked: new_acked}})

        if new_acked >= required do
          # Majority reached — notify waiting process
          send(__MODULE__, {:majority_reached, seq})
        end

        :ok

      [] ->
        # Entry already cleaned up or not found (majority already reached)
        :ok
    end
  end

  @doc """
  Apply WAL operations on a replica and acknowledge.

  Called via `:erpc.cast` from the primary. Applies the operations locally
  then sends an ack back to the primary.
  """
  @spec replicate_and_ack(non_neg_integer, non_neg_integer, [{atom, any}]) :: :ok
  def replicate_and_ack(seq, partition_idx, ops) do
    apply_local(partition_idx, ops)

    # Send ack back to primary
    {primary, _} = Manager.get_replicas(partition_idx)

    if primary != node() do
      try do
        :erpc.cast(primary, __MODULE__, :ack, [seq, node()])
      catch
        kind, reason ->
          Logger.warning(
            "super_cache, wal, ack cast failed → #{inspect(primary)}: #{inspect({kind, reason})}"
          )
      end
    end

    :ok
  end

  @doc """
  Recover uncommitted WAL entries after restart.

  Replays any entries that haven't been fully committed to ensure consistency.
  """
  @spec recover() :: :ok
  def recover() do
    entries = :ets.tab2list(@table)
    count = length(entries)

    if count > 0 do
      Logger.info("super_cache, wal, recovering #{count} uncommitted entries")

      Enum.each(entries, fn {_seq, entry} ->
        # Re-apply locally
        apply_local(entry.partition_idx, entry.ops)

        # Re-replicate to replicas
        {_primary, replicas} = Manager.get_replicas(entry.partition_idx)
        async_replicate(entry.seq, entry.partition_idx, entry.ops, replicas)
      end)
    end

    :ok
  end

  @doc """
  Return WAL statistics.

  ## Example

      WAL.stats()
      # => %{pending: 42, committed: 1000}
  """
  @spec stats() :: %{pending: non_neg_integer, acks_pending: non_neg_integer}
  def stats() do
    %{
      pending: :ets.info(@table, :size),
      acks_pending: :ets.info(@ack_table, :size)
    }
  end

  # ── GenServer callbacks ──────────────────────────────────────────────────────

  @impl true
  def init(_opts) do
    table = :ets.new(@table, [
      :ordered_set,
      :public,
      :named_table,
      {:read_concurrency, true},
      {:write_concurrency, true}
    ])

    ack_table = :ets.new(@ack_table, [
      :set,
      :public,
      :named_table,
      {:read_concurrency, true},
      {:write_concurrency, true}
    ])

    # Initialize sequence counter
    :persistent_term.put(@seq_key, 0)

    # Start periodic cleanup
    schedule_cleanup()

    Logger.info("super_cache, wal, ETS tables ready (wal: #{inspect(table)}, acks: #{inspect(ack_table)})")
    {:ok, %{}}
  end

  @impl true
  def handle_info({:majority_reached, seq}, state) do
    # Clean up ack entry once majority is reached
    :ets.delete(@ack_table, seq)
    {:noreply, state}
  end

  def handle_info(:cleanup, state) do
    cleanup_old_entries()
    schedule_cleanup()
    {:noreply, state}
  end

  def handle_info(_msg, state), do: {:noreply, state}

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

  defp next_seq() do
    current = :persistent_term.get(@seq_key, 0)
    :persistent_term.put(@seq_key, current + 1)
    current + 1
  end

  defp apply_local(partition_idx, ops) do
    partition = Partition.get_partition_by_idx(partition_idx)

    Enum.each(ops, fn
      {: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)
  end

  defp async_replicate(seq, partition_idx, ops, replicas) do
    # Send to all replicas asynchronously using :erpc.cast (fire-and-forget)
    Enum.each(replicas, fn replica ->
      try do
        :erpc.cast(replica, __MODULE__, :replicate_and_ack, [seq, partition_idx, ops])
      catch
        kind, reason ->
          Logger.warning(
            "super_cache, wal, replication cast failed → #{inspect(replica)}: #{inspect({kind, reason})}"
          )
      end
    end)
  end

  defp await_majority(seq, required) do
    timeout = Application.get_env(:super_cache, :wal, [])[:majority_timeout] || @default_majority_timeout

    # Check if majority already reached (fast path)
    case :ets.lookup(@ack_table, seq) do
      [{^seq, %{acked: acked}}] when acked >= required ->
        :ok

      [{^seq, _}] ->
        # Wait for notification or timeout
        receive do
          {:majority_reached, ^seq} -> :ok
        after
          timeout -> {:error, :majority_timeout}
        end

      [] ->
        # Already cleaned up (majority reached before we checked)
        :ok
    end
  end

  defp schedule_cleanup() do
    interval = Application.get_env(:super_cache, :wal, [])[:cleanup_interval] || @default_cleanup_interval
    Process.send_after(self(), :cleanup, interval)
  end

  defp cleanup_old_entries() do
    # Remove entries older than 10 seconds
    cutoff = System.monotonic_time(:millisecond) - 10_000

    # Delete old WAL entries
    :ets.select_delete(@table, [
      {{:"$1", :"$2"}, [{:<, {:map_get, :timestamp, :"$2"}, cutoff}], [true]}
    ])

    # Delete old ack entries
    :ets.select_delete(@ack_table, [
      {{:"$1", :"$2"}, [{:<, {:map_get, :timestamp, :"$2"}, cutoff}], [true]}
    ])
  end
end