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
  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** - Tracking data is automatically removed after configurable TTL

  ## 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,
        ttl: 5_000  # 5 seconds - how long to track in-flight requests

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

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

  ## States

  - **`:in_flight`** - Request currently executing, other identical requests will wait
  - **`:completed`** - Brief period after completion to catch race conditions (100-500ms)

  ## Thread Safety

  Uses ETS for thread-safe storage and GenServer for coordination.
  """

  use GenServer

  @completed_ttl 500
  @default_wait_timeout 30_000
  @cleanup_interval 5_000

  # 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} ->
          :telemetry.execute(
            [:httpower, :dedup, :execute],
            %{},
            %{dedup_key: dedup_hash}
          )

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

        {:ok, :wait, ref} ->
          # The ref is shared across all waiters for the same in-flight request.
          # It's created when the first request starts and used to correlate the
          # response broadcast in handle_cast(:complete). If the ETS entry is
          # cleaned up and recreated between this point and the receive (e.g., the
          # original request completes and a new one starts with the same hash),
          # the waiter will not match the new ref and will fall through to the
          # wait timeout. This is safe — the timeout produces a clean error.
          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

              :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}

            {:dedup_error, ^ref, reason} ->
              {:error,
               %HTTPower.Error{
                 reason: reason,
                 message: "Deduplication failed: #{reason}"
               }}
          after
            wait_timeout ->
              {:error,
               %HTTPower.Error{reason: :dedup_timeout, message: "Request deduplication timeout"}}
          end

        {: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)
      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).
  @doc false
  @spec deduplicate(String.t(), keyword()) ::
          {:ok, :execute} | {:ok, :wait, reference()} | {:ok, any()} | {:error, atom()}
  def deduplicate(request_hash, config \\ []) do
    if deduplication_enabled?(config) do
      GenServer.call(__MODULE__, {:deduplicate, request_hash, self()})
    else
      {:ok, :execute}
    end
  end

  @doc false
  @spec complete(String.t(), any(), keyword()) :: :ok
  def complete(request_hash, response, config \\ []) do
    if deduplication_enabled?(config) do
      GenServer.cast(__MODULE__, {:complete, request_hash, response})
    else
      :ok
    end
  end

  @doc false
  @spec cancel(String.t()) :: :ok
  def cancel(request_hash) do
    GenServer.cast(__MODULE__, {:cancel, request_hash})
  end

  @doc false
  @spec hash(atom(), String.t(), String.t() | nil) :: String.t()
  def hash(method, url, body) do
    content = "#{method}:#{url}:#{body || ""}"
    :crypto.hash(:sha256, content) |> Base.encode16(case: :lower)
  end

  # Server Callbacks

  @impl true
  def init(_opts) do
    # ETS table for tracking requests
    # Format: {hash, state, ref, timestamp}
    # States: {:in_flight, [waiter_pids], owner_pid} | {:completed, response}
    # heir: :none ensures table dies with process (prevents orphaning on crash)
    # read/write_concurrency improves performance under high concurrent load (2-3x throughput)
    table =
      :ets.new(__MODULE__, [
        :set,
        :public,
        :named_table,
        {:read_concurrency, true},
        {:write_concurrency, true},
        {:heir, :none}
      ])

    # Schedule periodic cleanup
    schedule_cleanup()

    # pid_to_hashes: reverse index for O(1) lookup on process death
    #   %{pid => MapSet.t(hash)} — supports multiple concurrent hashes per PID
    # monitors: %{pid => monitor_ref} for cleanup (one monitor per PID)
    {:ok, %{table: table, pid_to_hashes: %{}, monitors: %{}}}
  end

  @impl true
  def handle_call({:deduplicate, hash, caller_pid}, _from, state) do
    case :ets.lookup(state.table, hash) do
      [] ->
        ref = make_ref()
        :ets.insert(state.table, {hash, {:in_flight, [], caller_pid}, ref, timestamp()})
        {:reply, {:ok, :execute}, monitor_pid(state, caller_pid, hash)}

      [{^hash, {:in_flight, waiters, owner}, ref, _ts}] ->
        :ets.update_element(state.table, hash, {2, {:in_flight, [caller_pid | waiters], owner}})
        {:reply, {:ok, :wait, ref}, monitor_pid(state, caller_pid, hash)}

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

  @impl true
  def handle_cast({:complete, hash, response}, state) do
    case :ets.lookup(state.table, hash) do
      [{^hash, {:in_flight, waiters, owner}, ref, _ts}] ->
        Enum.each(waiters, fn pid ->
          send(pid, {:dedup_response, ref, response})
        end)

        # Mark as completed with short TTL for race conditions
        :ets.insert(state.table, {hash, {:completed, response}, ref, timestamp()})

        # Clean up monitors for all participants of this hash
        state = demonitor_pids(state, [owner | waiters], hash)
        {:noreply, state}

      _ ->
        {:noreply, state}
    end
  end

  @impl true
  def handle_cast({:cancel, hash}, state) do
    case :ets.lookup(state.table, hash) do
      [{^hash, {:in_flight, _waiters, _owner}, _ref, _ts}] ->
        {:noreply, abort_in_flight(state, hash, :request_cancelled)}

      _ ->
        :ets.delete(state.table, hash)
        {:noreply, state}
    end
  end

  @impl true
  def handle_info({:DOWN, _mon_ref, :process, pid, _reason}, state) do
    {hashes, pid_to_hashes} = Map.pop(state.pid_to_hashes, pid, MapSet.new())
    {_mon_ref, monitors} = Map.pop(state.monitors, pid)
    state = %{state | pid_to_hashes: pid_to_hashes, monitors: monitors}

    state =
      Enum.reduce(hashes, state, fn hash, acc ->
        case :ets.lookup(acc.table, hash) do
          [{^hash, {:in_flight, _waiters, ^pid}, _ref, _ts}] ->
            # Original requester died — notify all waiters and clean up
            abort_in_flight(acc, hash, :requester_down)

          [{^hash, {:in_flight, waiters, owner}, ref, ts}] ->
            # A waiter died — remove from waiter list
            new_waiters = List.delete(waiters, pid)
            :ets.insert(acc.table, {hash, {:in_flight, new_waiters, owner}, ref, ts})
            acc

          _ ->
            acc
        end
      end)

    {:noreply, state}
  end

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

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

    schedule_cleanup()

    {:noreply, state}
  end

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

  # Private Functions

  defp monitor_pid(state, pid, hash) do
    hashes = Map.get(state.pid_to_hashes, pid, MapSet.new())
    state = put_in(state, [:pid_to_hashes, pid], MapSet.put(hashes, hash))

    if Map.has_key?(state.monitors, pid) do
      # Already monitoring this PID, just track the new hash
      state
    else
      mon_ref = Process.monitor(pid)
      put_in(state, [:monitors, pid], mon_ref)
    end
  end

  defp abort_in_flight(state, hash, reason) do
    [{^hash, {:in_flight, waiters, owner}, ref, _ts}] = :ets.lookup(state.table, hash)

    Enum.each(waiters, fn pid ->
      send(pid, {:dedup_error, ref, reason})
    end)

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

    :ets.delete(state.table, hash)
    demonitor_pids(state, [owner | waiters], hash)
  end

  defp demonitor_pids(state, pids, hash) do
    Enum.reduce(pids, state, fn pid, acc -> cleanup_pid_hash(acc, pid, hash) end)
  end

  defp cleanup_pid_hash(state, pid, hash) do
    case Map.get(state.pid_to_hashes, pid) do
      nil ->
        state

      hashes ->
        hashes
        |> MapSet.delete(hash)
        |> cleanup_remaining_hashes(pid, state)
    end
  end

  defp cleanup_remaining_hashes(remaining, pid, state) do
    if Enum.empty?(remaining) do
      # No more active hashes for this PID — fully clean up
      {mon_ref, monitors} = Map.pop(state.monitors, pid)
      if mon_ref, do: Process.demonitor(mon_ref, [:flush])
      %{state | monitors: monitors, pid_to_hashes: Map.delete(state.pid_to_hashes, pid)}
    else
      # PID still has other active hashes — keep monitor, update set
      put_in(state, [:pid_to_hashes, pid], remaining)
    end
  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