nebulex_local

defmodule Nebulex.Locks do
  @moduledoc """
  A simple, ETS-based locking mechanism for local synchronization.

  This module provides a lightweight locking system suitable for single-node
  use cases, such as implementing transactions in local cache adapters. It
  uses ETS for lock storage with atomic operations for acquiring and releasing
  locks.

  ## Features

    * **Atomic lock acquisition** - Uses `insert_new` for atomic operations.
    * **Deadlock prevention** - Locks are acquired in sorted key order.
    * **Dual stale lock cleanup** - Automatic cleanup via two mechanisms:
      * **On-demand**: During lock acquisition attempts
      * **Periodic**: Background cleanup at configurable intervals
    * **Fine-grained locking** - Lock specific keys for better concurrency.
    * **Retry with jitter** - Configurable retries with jitter to prevent
      thundering herd.
    * **Scalable cleanup** - Batch-based cleanup using ETS continuation to
      handle large numbers of locks efficiently.

  ## Usage

  The module supports two modes of operation:

  ### Named mode (recommended)

  When you provide a `:name` option, both the GenServer and the ETS table
  are registered with that name, allowing direct access without calling
  `get_table/1`:

      # Start with a name
      children = [
        {Nebulex.Locks, name: MyApp.Locks}
      ]

      Supervisor.start_link(children, strategy: :one_for_one)

      # Use the name directly for acquire/release
      case Nebulex.Locks.acquire(MyApp.Locks, [:key1, :key2], retries: 5) do
        :ok ->
          try do
            # Critical section
          after
            Nebulex.Locks.release(MyApp.Locks, [:key1, :key2])
          end

        {:error, :timeout} ->
          # Failed to acquire locks
      end

  ### Anonymous mode

  When `:name` is not provided or is `nil`, the GenServer starts unnamed
  and you must use `get_table/1` to obtain the table reference:

      # Start without a name
      {:ok, pid} = Nebulex.Locks.start_link([])

      # Get the table reference
      table = Nebulex.Locks.get_table(pid)

      # Use the table reference
      Nebulex.Locks.acquire(table, [:key1], retries: 3)

  ## Start Options

  #{Nebulex.Locks.Options.start_options_docs()}

  ## Lock Entry Structure

  Locks are stored in ETS as records:

      {:lock, lock_key, owner_pid, timestamp, timeout}

  Where:

    * `lock_key` - The key being locked (sorted to prevent deadlocks).
    * `owner_pid` - The PID of the process holding the lock.
    * `timestamp` - When the lock was acquired (for stale detection).
    * `timeout` - Maximum time the lock can be held before being stale.

  ## Stale Lock Detection and Cleanup

  This module implements a defense-in-depth approach to stale lock cleanup,
  combining two complementary mechanisms:

  ### On-Demand Cleanup

  When a process attempts to acquire a lock and finds an existing lock, it
  checks if the lock is stale before failing the acquisition:

    * **Dead process detection**: Uses `Process.alive?/1` to detect if the
      lock owner has crashed.
    * **Timeout detection**: Compares the lock's timestamp with its configured
      `lock_timeout` to detect locks held longer than allowed.

  If a stale lock is detected, it is immediately removed and the acquiring
  process can obtain the lock. This ensures low latency for lock acquisition
  on frequently accessed keys.

  ### Periodic Cleanup

  A background cleanup process runs at the configured `:cleanup_interval`
  (default: 5 minutes) to proactively remove stale locks. Locks are processed
  in batches (default: 100 locks per batch) using ETS continuation to avoid
  loading the entire table into memory at once.

  This periodic cleanup ensures that stale locks from infrequently accessed
  keys don't accumulate in memory, preventing memory leaks.

  ### Why Two Mechanisms?

    * **On-demand cleanup**: Provides immediate cleanup for actively used keys
      with minimal latency.
    * **Periodic cleanup**: Prevents memory leaks by cleaning up locks for keys
      that are no longer accessed.
    * **Together**: Ensures robust cleanup under all usage patterns.

  ## Performance Characteristics

  ### Lock Acquisition and Release

    * **No GenServer bottleneck**: Lock operations hit ETS directly via atomic
      `insert_new` operations. Only `get_table/1` requires a GenServer call.
    * **High concurrency**: Public ETS table with `:write_concurrency` and
      `:read_concurrency` enabled.

  ### Lock Fairness

    * **No FIFO guarantee**: When multiple processes compete for the same lock,
      the winner is determined by ETS race conditions and scheduler timing.
    * **Practical fairness**: With retry intervals and jitter, starvation is
      virtually impossible in practice.
    * **Trade-off**: Prioritizes performance over strict ordering. For use cases
      requiring FIFO fairness, consider `sleeplocks` or a queue-based approach.

  """

  use GenServer

  alias Nebulex.Locks.Options
  alias Nebulex.Time

  import Record

  ## Internals

  # Lock Entry
  defrecord(:lock,
    key: nil,
    owner: nil,
    timestamp: nil,
    timeout: nil
  )

  # Internal state
  defstruct table: nil, cleanup_interval: nil, cleanup_batch_size: nil

  ## API

  @doc """
  Starts the locks GenServer.

  ## Options

  #{Nebulex.Locks.Options.start_options_docs()}

  """
  @spec start_link(keyword()) :: GenServer.on_start()
  def start_link(opts) do
    opts = Options.validate_start_opts!(opts)

    case Keyword.get(opts, :name) do
      nil ->
        GenServer.start_link(__MODULE__, opts)

      name when is_atom(name) ->
        GenServer.start_link(__MODULE__, opts, name: name)
    end
  end

  @doc """
  Returns the locks table from a running GenServer.

  This function is needed when starting the GenServer without a name.
  For named GenServers, you can use the name directly with `acquire/3`
  and `release/2`.

  ## Examples

      # Named mode - table name is the same as GenServer name
      iex> {:ok, _pid} = Nebulex.Locks.start_link(name: MyLocks)
      iex> table = Nebulex.Locks.get_table(MyLocks)
      iex> table == MyLocks
      true

      # Anonymous mode - returns a table reference
      iex> {:ok, pid} = Nebulex.Locks.start_link([])
      iex> table = Nebulex.Locks.get_table(pid)
      iex> is_reference(table)
      true

  """
  @spec get_table(GenServer.server()) :: :ets.table()
  def get_table(server) do
    GenServer.call(server, :get_table)
  end

  @doc """
  Attempts to acquire locks for the given keys.

  Locks are acquired atomically in a sorted order to prevent deadlocks.
  If any lock cannot be acquired, all partially acquired locks are released
  and the operation is retried according to the options.

  The first parameter can be either a named table (atom) or a table reference.

  ## Options

  #{Nebulex.Locks.Options.options_docs()}

  ## Examples

      # Using a named table
      iex> {:ok, _pid} = Nebulex.Locks.start_link(name: MyLocks)
      iex> Nebulex.Locks.acquire(MyLocks, [:key1, :key2])
      :ok

      # Using a table reference
      iex> {:ok, pid} = Nebulex.Locks.start_link([])
      iex> table = Nebulex.Locks.get_table(pid)
      iex> Nebulex.Locks.acquire(table, [:key1])
      :ok

  """
  @spec acquire(:ets.table(), [any()], keyword()) :: :ok | {:error, :timeout}
  def acquire(table, keys, opts \\ []) do
    opts = Options.validate!(opts)
    retries = Keyword.fetch!(opts, :retries)
    retry_interval = Keyword.fetch!(opts, :retry_interval)
    lock_timeout = Keyword.fetch!(opts, :lock_timeout)
    lock_keys = build_lock_keys(keys)

    do_acquire(table, lock_keys, retries, retry_interval, lock_timeout, 0)
  end

  @doc """
  Releases locks for the given keys.

  This function is typically called in an `after` block to ensure
  locks are released even if an exception occurs.

  This API assumes trusted callers: it does not verify lock ownership
  when deleting lock entries. Releasing keys not owned by the current
  process is considered undefined behavior.

  The first parameter can be either a named table (atom) or a table reference.

  ## Examples

      # Using a named table
      iex> {:ok, _pid} = Nebulex.Locks.start_link(name: MyLocks)
      iex> Nebulex.Locks.acquire(MyLocks, [:key1, :key2])
      :ok
      iex> Nebulex.Locks.release(MyLocks, [:key1, :key2])
      :ok

      # Using a table reference
      iex> {:ok, pid} = Nebulex.Locks.start_link([])
      iex> table = Nebulex.Locks.get_table(pid)
      iex> Nebulex.Locks.acquire(table, [:key1])
      :ok
      iex> Nebulex.Locks.release(table, [:key1])
      :ok

  """
  @spec release(:ets.table(), [any()]) :: :ok
  def release(table, keys) do
    keys
    |> build_lock_keys()
    |> Enum.each(&:ets.delete(table, &1))
  end

  ## GenServer Callbacks

  @impl true
  def init(opts) do
    name = Keyword.get(opts, :name)
    cleanup_interval = Keyword.fetch!(opts, :cleanup_interval)
    cleanup_batch_size = Keyword.fetch!(opts, :cleanup_batch_size)
    init_callback = Keyword.get(opts, :init_callback)

    # Initialize the locks table
    table = init_table(name)

    # Invoke the init callback if provided
    if init_callback do
      {m, f, a} = init_callback

      apply(m, f, [table | a])
    end

    # Create the initial state
    state = %__MODULE__{
      table: table,
      cleanup_interval: cleanup_interval,
      cleanup_batch_size: cleanup_batch_size
    }

    {:ok, state, {:continue, :setup_cleanup}}
  end

  @impl true
  def handle_continue(:setup_cleanup, %__MODULE__{cleanup_interval: cleanup_interval} = state) do
    # Schedule first cleanup
    Process.send_after(self(), :cleanup, cleanup_interval)

    {:noreply, state}
  end

  @impl true
  def handle_call(:get_table, _from, %__MODULE__{table: table} = state) do
    {:reply, table, state}
  end

  @impl true
  def handle_info(
        :cleanup,
        %__MODULE__{
          table: table,
          cleanup_interval: cleanup_interval,
          cleanup_batch_size: cleanup_batch_size
        } = state
      ) do
    cleanup_stale_locks(table, cleanup_batch_size)

    # Schedule next cleanup
    Process.send_after(self(), :cleanup, cleanup_interval)

    {:noreply, state}
  end

  ## Helpers

  # Initialize a new locks ETS table
  defp init_table(nil) do
    :ets.new(:locks_table, [
      :set,
      :public,
      keypos: 2,
      read_concurrency: true,
      write_concurrency: true
    ])
  end

  defp init_table(name) do
    :ets.new(name, [
      :set,
      :public,
      :named_table,
      keypos: 2,
      read_concurrency: true,
      write_concurrency: true
    ])
  end

  # Build and sort lock keys to prevent deadlocks
  defp build_lock_keys([]) do
    # Global lock when no keys specified
    [:__global_lock__]
  end

  defp build_lock_keys(keys) do
    # Sort keys deterministically to prevent deadlocks
    Enum.sort(keys)
  end

  # Retry logic for lock acquisition
  defp do_acquire(_table, _lock_keys, 0, _retry_interval, _lock_timeout, _attempt) do
    {:error, :timeout}
  end

  defp do_acquire(table, lock_keys, :infinity, retry_interval, lock_timeout, attempt) do
    with :error <- try_acquire_all(table, lock_keys, lock_timeout) do
      sleep_with_jitter(retry_interval, attempt)

      do_acquire(table, lock_keys, :infinity, retry_interval, lock_timeout, attempt + 1)
    end
  end

  defp do_acquire(table, lock_keys, retries, retry_interval, lock_timeout, attempt)
       when is_integer(retries) and retries > 0 do
    with :error <- try_acquire_all(table, lock_keys, lock_timeout) do
      if attempt < retries do
        sleep_with_jitter(retry_interval, attempt)

        do_acquire(table, lock_keys, retries, retry_interval, lock_timeout, attempt + 1)
      else
        {:error, :timeout}
      end
    end
  end

  # Try to acquire all locks atomically (all-or-nothing)
  defp try_acquire_all(table, keys, timeout) do
    now = Time.now()
    owner = self()

    # Build all lock records upfront
    locks = Enum.map(keys, &lock(key: &1, owner: owner, timestamp: now, timeout: timeout))

    # Fast path: try to insert all locks at once (O(1))
    with :error <- insert_new(table, locks),
         # Slow path: check for stale locks with early termination
         :ok <- cleanup_stale_locks(table, keys, now) do
      # Stale locks were cleaned, retry batch insert (inner retry)
      insert_new(table, locks)
    end
  end

  # Insert new lock records atomically
  defp insert_new(table, lock_records) do
    case :ets.insert_new(table, lock_records) do
      true -> :ok
      false -> :error
    end
  end

  # Check all keys for stale locks with early termination on valid lock
  defp cleanup_stale_locks(table, lock_keys, now) do
    with stale_keys when is_list(stale_keys) <-
           Enum.reduce_while(lock_keys, [], &check_stale_key(&1, &2, table, now)) do
      # Delete all stale locks (could be empty if locks were released)
      Enum.each(stale_keys, &:ets.delete(table, &1))
    end
  end

  defp check_stale_key(lock_key, stale_acc, table, now) do
    case :ets.lookup(table, lock_key) do
      # This branch handles the race condition where a lock exists during
      # the batch insert attempt but is released before we check it during
      # stale lock detection. Difficult to test deterministically without
      # introducing flaky tests.
      # coveralls-ignore-start
      [] ->
        # No lock exists, continue checking
        {:cont, stale_acc}

      # coveralls-ignore-stop

      [lock(key: ^lock_key, owner: pid, timestamp: timestamp, timeout: timeout)] ->
        if stale_lock?(pid, timestamp, now, timeout) do
          # Stale lock found, mark for cleanup and continue
          {:cont, [lock_key | stale_acc]}
        else
          # Valid lock found, stop immediately
          {:halt, :error}
        end
    end
  end

  # Check if a lock is stale (dead process or timed out)
  defp stale_lock?(pid, timestamp, now, lock_timeout) do
    !Process.alive?(pid) or now - timestamp > lock_timeout
  end

  # Sleep with custom interval (function-based, no jitter)
  defp sleep_with_jitter(interval_fun, attempt) when is_function(interval_fun, 1) do
    case interval_fun.(attempt) do
      interval when is_integer(interval) and interval >= 0 ->
        Process.sleep(interval)

      other ->
        raise ArgumentError,
              "retry_interval function must return a non-negative integer, got: #{inspect(other)}"
    end
  end

  # Sleep with random jitter to prevent thundering herd (fixed interval)
  defp sleep_with_jitter(0, _attempt) do
    :ok
  end

  defp sleep_with_jitter(interval, _attempt) when is_integer(interval) and interval > 0 do
    jitter = :rand.uniform(interval)
    sleep_time = interval + jitter

    Process.sleep(sleep_time)
  end

  # Cleanup stale locks from the table
  defp cleanup_stale_locks(table, cleanup_batch_size) do
    now = Time.now()

    # Fix the table to prevent inconsistencies during cleanup
    :ets.safe_fixtable(table, true)

    # Select locks in batches using continuation
    case :ets.select(table, [{:"$1", [], [:"$1"]}], cleanup_batch_size) do
      :"$end_of_table" ->
        :ok

      {locks, continuation} ->
        cleanup_batch(table, locks, now)
        cleanup_with_continuation(table, continuation, now)
    end
  after
    # Unfix the table
    :ets.safe_fixtable(table, false)
  end

  # Cleanup locks using continuation
  defp cleanup_with_continuation(_table, :"$end_of_table", _now) do
    :ok
  end

  defp cleanup_with_continuation(table, continuation, now) do
    case :ets.select(continuation) do
      :"$end_of_table" ->
        :ok

      {locks, next_continuation} ->
        cleanup_batch(table, locks, now)
        cleanup_with_continuation(table, next_continuation, now)
    end
  end

  # Cleanup a batch of locks
  defp cleanup_batch(table, locks, now) do
    Enum.each(locks, fn lock(key: key, owner: pid, timestamp: ts, timeout: timeout) ->
      if stale_lock?(pid, ts, now, timeout) do
        :ets.delete(table, key)
      end
    end)
  end
end