vela_cache

defmodule Vela.Lock.Manager do
  @moduledoc """
  Per-key fetch locks to prevent cache stampedes (thundering herd).

  When multiple processes call get_or_fetch for the same key simultaneously:
  - The first one acquires the lock and fetches the value
  - All others wait
  - When the lock is released, waiters re-check the cache before deciding
    whether to fetch themselves (usually the value is already there)
  """

  use GenServer

  @default_timeout 5_000

  # ---- Public API ----

  def start_link(opts) do
    cache_name = Keyword.fetch!(opts, :name)
    GenServer.start_link(__MODULE__, opts, name: manager_name(cache_name))
  end

  def manager_name(cache_name), do: :"vela_lock_manager_#{cache_name}"

  @doc """
  Acquire the lock for `key`. Blocks until the lock is granted or timeout.
  Returns :granted or {:error, :timeout}.
  """
  def acquire(cache_name, key, timeout \\ @default_timeout) do
    manager_name(cache_name)
    |> GenServer.call({:acquire, key}, timeout + 100)
  catch
    :exit, {:timeout, _} -> {:error, :timeout}
  end

  @doc "Release the lock for `key`."
  def release(cache_name, key) do
    GenServer.cast(manager_name(cache_name), {:release, key})
  end

  # ---- GenServer Callbacks ----

  @impl true
  def init(opts) do
    config = Keyword.fetch!(opts, :config)
    {:ok, %{config: config, locks: %{}}}
  end

  # locks map: %{key => %{owner: pid, monitor: ref, waiters: :queue.queue(), timer: ref}}

  @impl true
  def handle_call({:acquire, key}, {caller_pid, _} = from, state) do
    case Map.get(state.locks, key) do
      nil ->
        # No lock held — grant immediately
        monitor = Process.monitor(caller_pid)

        timer =
          Process.send_after(
            self(),
            {:lock_timeout, key, caller_pid},
            state.config.stampede_timeout
          )

        lock = %{owner: caller_pid, monitor: monitor, waiters: :queue.new(), timer: timer}
        {:reply, :granted, put_in(state.locks[key], lock)}

      lock ->
        # Lock held by another process — add caller to the wait queue
        new_lock = update_in(lock.waiters, &:queue.in(from, &1))
        {:noreply, put_in(state.locks[key], new_lock)}
    end
  end

  @impl true
  def handle_cast({:release, key}, state) do
    {:noreply, release_lock(state, key)}
  end

  # Auto-release if a monitored process dies while holding the lock
  @impl true
  def handle_info({:DOWN, _ref, :process, pid, _reason}, state) do
    state =
      Enum.reduce(state.locks, state, fn {key, lock}, acc ->
        if lock.owner == pid, do: release_lock(acc, key), else: acc
      end)

    {:noreply, state}
  end

  # Force-release if a process holds the lock for too long
  @impl true
  def handle_info({:lock_timeout, key, owner_pid}, state) do
    case Map.get(state.locks, key) do
      %{owner: ^owner_pid} -> {:noreply, release_lock(state, key)}
      _ -> {:noreply, state}
    end
  end

  # ---- Helpers ----

  defp release_lock(state, key) do
    case Map.get(state.locks, key) do
      nil ->
        state

      lock ->
        Process.cancel_timer(lock.timer)
        Process.demonitor(lock.monitor, [:flush])

        case :queue.out(lock.waiters) do
          {:empty, _} ->
            %{state | locks: Map.delete(state.locks, key)}

          {{:value, next_from}, remaining_waiters} ->
            {next_pid, _} = next_from
            new_monitor = Process.monitor(next_pid)

            new_timer =
              Process.send_after(
                self(),
                {:lock_timeout, key, next_pid},
                state.config.stampede_timeout
              )

            new_lock = %{
              lock
              | owner: next_pid,
                monitor: new_monitor,
                waiters: remaining_waiters,
                timer: new_timer
            }

            GenServer.reply(next_from, :granted)

            put_in(state.locks[key], new_lock)
        end
    end
  end
end