httpower

defmodule HTTPower.Middleware.Dedup do
  @behaviour HTTPower.Middleware

  @moduledoc """
  In-flight request deduplication to prevent duplicate operations.

  This module prevents duplicate requests from causing duplicate side effects
  (e.g., double charges, duplicate orders) by tracking in-flight requests and
  sharing responses with identical concurrent requests.

  ## How It Works

  1. **Request Fingerprinting** - Each request gets a hash based on method + URL + body,
     plus the auth-context headers (`authorization`, `cookie`, `x-api-key`) so responses
     are never shared across different credentials
  2. **In-Flight Tracking** - First request executes normally, subsequent identical requests wait
  3. **Response Sharing** - When the first request completes, all waiting requests receive the same response
  4. **Automatic Cleanup** - Completed responses are cached briefly so late-arriving
     duplicates still share them, then swept; orphaned in-flight rows (owner crashed
     without completing) are reaped as a safety net

  ## Use Cases

  - Prevent double charges from double-clicks on payment buttons
  - Prevent duplicate orders from retry storms or race conditions
  - Ensure idempotency for critical mutations (POST/PUT/DELETE)

  ## Configuration

      # Global configuration
      config :httpower, :deduplicate,
        enabled: true

      # Per-request configuration
      HTTPower.post(url,
        body: payment_data,
        deduplicate: true
      )

      # Or with options
      HTTPower.post(url,
        body: payment_data,
        deduplicate: [
          enabled: true,
          wait_timeout: 60_000,   # How long waiters block for an in-flight request (default: 30s)
          key: "custom-dedup-key"  # Optional: override hash generation
        ]
      )

  > #### Cache and reap windows are fixed, not per-request {: .info}
  >
  > The completed-response cache window (~0.5s) and the orphaned-in-flight reap
  > ceiling (60s) are internal constants. Cleanup runs in a shared process with no
  > per-request context, so these are not configurable per request. `wait_timeout`
  > and `key` are the per-request knobs.

  ## States

  - **`:in_flight`** - Request currently executing, other identical requests will wait
  - **`:completed`** - Brief window (~0.5s) after completion so late duplicates still share the response

  ## Thread Safety

  Hot path (claim, wait, complete, cancel) is lock-free on a public ETS table via
  `:ets.insert_new/2` and `:ets.select_replace/2` CAS — no GenServer round-trip.
  The GenServer owns the table lifecycle and runs periodic cleanup only.
  """

  use GenServer

  @table_name __MODULE__
  @completed_ttl 500
  @default_wait_timeout 30_000
  # Global ceiling for orphaned :in_flight rows (owner crash without complete/cancel).
  # Independent of per-request wait_timeout; cleanup has no access to caller config.
  @max_in_flight_ttl 60_000
  @cleanup_interval 5_000
  # Bounds only the vanished-row retry path (a row deleted between insert_new and
  # lookup). CAS contention among concurrent waiters on the same hash is bounded
  # separately by the number of racers — each failed CAS means a distinct waiter
  # won its append — so that path always terminates without consuming this budget.
  @claim_retries 32

  # Client API

  @doc """
  Starts the request deduplicator GenServer.
  """
  def start_link(opts \\ []) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @doc """
  Feature callback for the HTTPower pipeline.

  Checks for duplicate requests and either executes, waits, or returns cached response.

  Returns:
  - `{:ok, request}` with dedup info stored in private (first occurrence)
  - `{:halt, response}` if cached response available (short-circuit)
  - Waits and returns `{:halt, response}` for duplicate in-flight requests

  ## Examples

      iex> request = %HTTPower.Request{method: :post, url: "https://api.example.com/charge", body: "..."}
      iex> HTTPower.Dedup.handle_request(request, [enabled: true])
      {:ok, modified_request}
  """
  @impl HTTPower.Middleware
  def handle_request(request, config) do
    if deduplication_enabled?(config) do
      dedup_hash = extract_dedup_hash(request, config)

      case deduplicate(dedup_hash, config) do
        {:ok, :execute, ref} ->
          :telemetry.execute(
            [:httpower, :dedup, :execute],
            %{},
            %{dedup_key: dedup_hash}
          )

          modified_request =
            HTTPower.Request.put_private(request, :dedup, {dedup_hash, ref, config})

          {:ok, modified_request}

        {:waited, response, wait_time_ms} ->
          :telemetry.execute(
            [:httpower, :dedup, :wait],
            %{wait_time_ms: wait_time_ms, bypassed_rate_limit: 1},
            %{
              dedup_key: dedup_hash,
              coordination: :rate_limit_bypass
            }
          )

          {:halt, response}

        {:ok, cached_response} ->
          :telemetry.execute(
            [:httpower, :dedup, :cache_hit],
            %{bypassed_rate_limit: 1},
            %{
              dedup_key: dedup_hash,
              coordination: :rate_limit_bypass
            }
          )

          {:halt, cached_response}

        {:error, reason} ->
          {:error,
           %HTTPower.Error{reason: reason, message: "Deduplication error: #{inspect(reason)}"}}
      end
    else
      :ok
    end
  end

  defp extract_dedup_hash(request, config) do
    case Keyword.get(config, :key) do
      nil -> hash(request.method, URI.to_string(request.url), request.body, request.headers)
      custom_key -> custom_key
    end
  end

  # Internal middleware API — called by the Dedup middleware itself and by
  # `HTTPower.Client`. Operates on the internal SHA256 request hash, so it is
  # intentionally undocumented (not part of the public surface).
  #
  # Returns:
  # - `{:ok, :execute, ref}` — caller owns the hash; must complete/cancel with ref
  # - `{:ok, response}` — cache hit on a recently completed request
  # - `{:waited, response, wait_time_ms}` — waited for an in-flight owner
  # - `{:error, reason}` — wait timeout or cancel while waiting
  @doc false
  @spec deduplicate(String.t(), keyword()) ::
          {:ok, :execute, reference()}
          | {:ok, :execute}
          | {:ok, any()}
          | {:waited, any(), non_neg_integer()}
          | {:error, atom()}
  def deduplicate(request_hash, config \\ []) do
    if deduplication_enabled?(config) do
      claim(request_hash, config, @claim_retries)
    else
      # Disabled path has no ref; callers must not complete/cancel.
      {:ok, :execute}
    end
  end

  @doc false
  @spec complete(String.t(), reference(), any(), keyword()) :: :ok
  def complete(request_hash, ref, response, config \\ []) do
    if deduplication_enabled?(config) and is_reference(ref) do
      complete_loop(request_hash, ref, response)
    else
      :ok
    end
  end

  @doc false
  @spec cancel(String.t(), reference() | nil) :: :ok
  def cancel(request_hash, ref \\ nil)

  def cancel(_request_hash, nil), do: :ok

  def cancel(request_hash, ref) when is_reference(ref) do
    cancel_loop(request_hash, ref, :request_cancelled)
  end

  # Credentials are part of the request identity: without them, two callers
  # sending the same method+URL+body with different Authorization headers would
  # collide and share a response across auth contexts.
  @auth_context_headers ["authorization", "cookie", "x-api-key"]

  @doc false
  @spec hash(atom(), String.t(), String.t() | nil, map()) :: String.t()
  def hash(method, url, body, headers \\ %{}) do
    content = [Atom.to_string(method), ?:, url, ?:, body || "", auth_context(headers)]
    :crypto.hash(:sha256, content) |> Base.encode16(case: :lower)
  end

  defp auth_context(headers) do
    headers
    |> Enum.flat_map(fn {key, value} ->
      downcased = key |> to_string() |> String.downcase()
      if downcased in @auth_context_headers, do: [{downcased, value}], else: []
    end)
    |> Enum.sort()
    |> case do
      [] -> ""
      pairs -> :erlang.term_to_binary(pairs)
    end
  end

  # ---------------------------------------------------------------------------
  # Caller-side hot path (no GenServer)
  # ---------------------------------------------------------------------------

  defp claim(_hash, _config, 0), do: {:error, :dedup_claim_exhausted}

  defp claim(hash, config, retries) do
    ref = make_ref()
    ts = timestamp()

    if :ets.insert_new(@table_name, {hash, {:in_flight, [], self()}, ref, ts}) do
      {:ok, :execute, ref}
    else
      resolve_existing(hash, config, retries)
    end
  end

  defp resolve_existing(hash, config, retries) do
    case :ets.lookup(@table_name, hash) do
      [{^hash, {:completed, response}, _ref, _ts}] ->
        {:ok, response}

      [{^hash, {:in_flight, _waiters, _owner}, ref, _ts}] ->
        register_waiter(hash, ref, config, retries)

      [] ->
        # Row vanished between insert_new failure and lookup (completed+cleaned
        # or cancelled) — retry claim.
        claim(hash, config, retries - 1)
    end
  end

  defp register_waiter(hash, ref, config, retries) do
    case :ets.lookup(@table_name, hash) do
      [{^hash, {:in_flight, waiters, owner}, ^ref, ts} = old] ->
        # Register an alias, not a pid: on wait timeout, unalias makes the
        # runtime drop the owner's late notify instead of leaking it into the
        # caller's mailbox (same mechanism GenServer.call uses for late replies).
        waiter_alias = :erlang.alias([:reply])
        new = {hash, {:in_flight, [waiter_alias | waiters], owner}, ref, ts}

        if cas_replace(old, new) do
          await(ref, waiter_alias, config)
        else
          # Row changed between lookup and CAS — re-resolve.
          :erlang.unalias(waiter_alias)
          resolve_existing(hash, config, retries)
        end

      [{^hash, {:in_flight, _waiters, _owner}, _other_ref, _ts}] ->
        # Different ref (ABA) — re-resolve against current row.
        resolve_existing(hash, config, retries)

      [{^hash, {:completed, response}, _ref, _ts}] ->
        {:ok, response}

      [] ->
        claim(hash, config, retries - 1)
    end
  end

  # Once the append CAS succeeded, every later complete/cancel CAS matches a
  # row containing our alias, so a notify is guaranteed — no ETS recheck needed
  # (rechecking would leave the in-flight notify unconsumed).
  defp await(ref, waiter_alias, config) do
    wait_timeout = Keyword.get(config, :wait_timeout, @default_wait_timeout)
    wait_start = System.monotonic_time(:millisecond)

    receive do
      {:dedup_response, ^ref, response} ->
        wait_time_ms = System.monotonic_time(:millisecond) - wait_start
        {:waited, response, wait_time_ms}

      {:dedup_error, ^ref, reason} ->
        {:error, reason}
    after
      wait_timeout ->
        # Deactivate first so a concurrent notify is dropped by the runtime,
        # then drain anything delivered before the unalias took effect.
        :erlang.unalias(waiter_alias)

        receive do
          {:dedup_response, ^ref, response} ->
            wait_time_ms = System.monotonic_time(:millisecond) - wait_start
            {:waited, response, wait_time_ms}

          {:dedup_error, ^ref, reason} ->
            {:error, reason}
        after
          0 -> {:error, :dedup_timeout}
        end
    end
  end

  # lookup → guarded CAS → notify; retry when a waiter appends between steps.
  defp complete_loop(hash, ref, response) do
    case :ets.lookup(@table_name, hash) do
      [{^hash, {:in_flight, waiters, _owner}, ^ref, _ts} = old] ->
        new = {hash, {:completed, response}, ref, timestamp()}

        if cas_replace(old, new) do
          Enum.each(waiters, &send(&1, {:dedup_response, ref, response}))
          :ok
        else
          # A waiter appended (or row otherwise changed) — retry with fresh list.
          complete_loop(hash, ref, response)
        end

      _ ->
        # Already completed / cancelled / reaped / different ref (ABA).
        :ok
    end
  end

  # Same CAS discipline as complete: lookup exact row, CAS-delete, then notify.
  defp cancel_loop(hash, ref, reason) do
    case :ets.lookup(@table_name, hash) do
      [{^hash, {:in_flight, waiters, _owner}, ^ref, _ts} = old] ->
        if cas_delete(old) do
          Enum.each(waiters, &send(&1, {:dedup_error, ref, reason}))

          :telemetry.execute(
            [:httpower, :dedup, :abort],
            %{waiter_count: length(waiters)},
            %{dedup_key: hash, reason: reason}
          )

          :ok
        else
          cancel_loop(hash, ref, reason)
        end

      _ ->
        :ok
    end
  end

  # Atomic compare-and-swap on a full row. Body must be `[{:const, new}]`
  # (not `[{{:const, new}}]`) so the replacement is the object itself; maps,
  # refs, and pids are only legal as match-spec body terms via `const`.
  defp cas_replace(old, new) do
    :ets.select_replace(@table_name, [{old, [], [{:const, new}]}]) == 1
  end

  defp cas_delete(old) do
    :ets.select_delete(@table_name, [{old, [], [true]}]) == 1
  end

  # ---------------------------------------------------------------------------
  # GenServer — table ownership + cleanup only
  # ---------------------------------------------------------------------------

  @impl true
  def init(_opts) do
    # Format: {hash, state, ref, timestamp}
    # States: {:in_flight, [waiter_aliases], owner_pid} | {:completed, response}
    # heir: :none ensures table dies with process (prevents orphaning on crash)
    # read/write_concurrency: concurrent CAS writers on the hot path
    table =
      :ets.new(@table_name, [
        :set,
        :public,
        :named_table,
        {:read_concurrency, true},
        {:write_concurrency, true},
        {:heir, :none}
      ])

    schedule_cleanup()
    {:ok, %{table: table}}
  end

  @impl true
  def handle_info(:cleanup, state) do
    now = timestamp()
    completed_cutoff = now - @completed_ttl
    in_flight_cutoff = now - @max_in_flight_ttl

    :ets.select_delete(state.table, [
      {
        {:"$1", {:completed, :"$2"}, :"$3", :"$4"},
        [{:<, :"$4", completed_cutoff}],
        [true]
      }
    ])

    # D2: reap orphaned :in_flight rows (owner died without complete/cancel).
    # Staleness alone is not enough: a legitimately slow owner (60s per-attempt
    # timeout plus retry backoff) can hold a row past the TTL, and reaping it
    # would let a duplicate claim the hash and execute concurrently. Liveness
    # is checked here (owners are always local pids); the TTL is purely the
    # dead-owner backstop. cas_delete guards against the owner completing
    # between the select and the delete. No notify — waiters rely on
    # wait_timeout + ref ABA protection.
    stale_rows =
      :ets.select(state.table, [
        {
          {:"$1", {:in_flight, :"$2", :"$3"}, :"$4", :"$5"},
          [{:<, :"$5", in_flight_cutoff}],
          [:"$_"]
        }
      ])

    reaped =
      Enum.count(stale_rows, fn {_hash, {:in_flight, _waiters, owner}, _ref, _ts} = row ->
        not Process.alive?(owner) and cas_delete(row)
      end)

    if reaped > 0 do
      # Distinct from :abort (active cancel) — reaping is a silent GC of orphaned
      # rows whose waiters are left to time out, so it carries no dedup_key or
      # waiter_count and must not share :abort's payload shape.
      :telemetry.execute(
        [:httpower, :dedup, :reap],
        %{count: reaped},
        %{reason: :stale_in_flight}
      )
    end

    schedule_cleanup()
    {:noreply, state}
  end

  @impl true
  def handle_info(_msg, state) do
    {:noreply, state}
  end

  defp deduplication_enabled?(config) do
    Keyword.get(config, :enabled, false)
  end

  defp schedule_cleanup do
    Process.send_after(self(), :cleanup, @cleanup_interval)
  end

  defp timestamp do
    System.monotonic_time(:millisecond)
  end
end