ccxt_client

defmodule CCXT.HTTP do
  @moduledoc """
  HTTP client for exchange API requests.

  Wraps Req with circuit breaker integration, error normalization, and
  telemetry. All exchange HTTP communication goes through this module.

  ## Why Manual Query Encoding?

  Uses `URI.encode_query/1` instead of Req's `:params` step because:

  1. **Signing requires raw params** — signing patterns need params before URL encoding
  2. **Sorted encoding** — some exchanges require alphabetically sorted params
  3. **Consistency** — both public and private requests use the same encoding

  ## Features

  - **Circuit breaker** — per-exchange, trips on 500+ and transport errors
  - **Error normalization** — HTTP/exchange errors to `CCXT.Error` structs
  - **Body-level error detection** — many exchanges return HTTP 200 with error in body
  - **HTML response detection** — geo-blocks and Cloudflare return HTML instead of JSON
  - **Safe retry** — only GET/HEAD, never POST/PUT/DELETE
  - **Telemetry** — emits `[:ccxt, :request, :start | :stop | :exception]`

  ## Usage

      {:ok, exchange} = CCXT.Exchange.new("bybit")

      # Public endpoint
      {:ok, response} = CCXT.HTTP.request(exchange, :get, "/v5/market/tickers",
        params: %{"category" => "spot", "symbol" => "BTCUSDT"}
      )

  """

  alias CCXT.CircuitBreaker
  alias CCXT.Defaults
  alias CCXT.Error
  alias CCXT.Exchange
  alias CCXT.RateLimiter
  alias CCXT.RateLimiter.Headers, as: RateLimitHeaders
  alias CCXT.Signing

  @html_preview_length 200
  @base_client_key {__MODULE__, :base_client}

  # CCXT rateLimit is milliseconds between requests; converted to max req/min below
  @rate_limit_period_ms 60_000

  @typedoc "HTTP response with status, headers, and decoded body"
  @type response_headers :: %{optional(String.t()) => [String.t()]}
  @type response :: %{status: integer(), headers: response_headers(), body: term()}

  @doc """
  Makes an HTTP request to an exchange API.

  ## Parameters

  - `exchange` - Exchange configuration struct
  - `method` - HTTP method (`:get`, `:post`, `:put`, `:delete`)
  - `path` - API endpoint path (e.g., "/v5/market/tickers")

  ## Options

  - `:params` - Query parameters or request body (default: `%{}`)
  - `:headers` - Additional request headers (default: `[]`)
  - `:timeout` - Request timeout in milliseconds (default: from `CCXT.Defaults`)
  - `:base_url` - Override base URL (default: uses exchange.base_urls)

  Any additional options are passed through to Req (useful for `:plug` in tests).

  ## Returns

  - `{:ok, response}` - Successful response with `:status`, `:headers`, `:body`
  - `{:error, %CCXT.Error{}}` - Normalized error
  """
  @spec request(Exchange.t(), atom(), String.t(), keyword()) ::
          {:ok, response()} | {:error, Error.t()}
  def request(%Exchange{} = exchange, method, path, opts \\ []) do
    params = Keyword.get(opts, :params, %{})
    headers = Keyword.get(opts, :headers, [])
    timeout = Keyword.get(opts, :timeout, Defaults.request_timeout_ms())
    custom_base_url = Keyword.get(opts, :base_url)
    endpoint_weight = Keyword.get(opts, :endpoint_weight, 1)

    extra_opts = Keyword.drop(opts, [:params, :headers, :timeout, :base_url, :endpoint_weight])

    with :ok <- check_circuit_breaker(exchange),
         :ok <- maybe_rate_limit(build_rate_key(exchange), exchange, endpoint_weight) do
      request_opts = %{
        base_url: custom_base_url || default_base_url(exchange),
        timeout: timeout,
        retry: Defaults.retry_policy(),
        extra_opts: extra_opts
      }

      do_request(exchange, method, path, params, headers, request_opts)
    end
  end

  # ===========================================================================
  # Request Execution
  # ===========================================================================

  defp do_request(exchange, method, path, params, headers, request_opts) do
    %{base_url: base_url, timeout: timeout, retry: retry, extra_opts: extra_opts} = request_opts

    req_opts = build_request(method, path, params, headers, base_url)
    req_opts = Keyword.merge(req_opts, receive_timeout: timeout, retry: retry)
    req_opts = Keyword.merge(req_opts, extra_opts)

    execute_request(exchange, method, path, req_opts)
  end

  # Shared execution pipeline: telemetry, Req call, circuit breaker, response handling.
  # Used by both unsigned (do_request) and signed (signed_request) code paths.
  defp execute_request(exchange, method, path, req_opts) do
    exchange_id = exchange.id
    start_time = System.monotonic_time()

    emit_start(exchange_id, method, path)

    try do
      base_client = get_base_client()
      result = Req.request(base_client, req_opts)

      CircuitBreaker.record_result(exchange_id, result)

      case result do
        {:ok, %Req.Response{status: status, headers: resp_headers, body: body}} ->
          emit_stop(exchange_id, method, path, status, start_time)
          maybe_update_rate_limit_state(exchange, resp_headers)
          handle_response(method, status, resp_headers, body, exchange)

        {:error, %Req.TransportError{reason: reason}} ->
          emit_exception(exchange_id, method, path, :transport, reason, start_time)
          {:error, Error.network_error(message: "Transport error: #{inspect(reason)}", exchange: exchange_id)}

        {:error, reason} ->
          emit_exception(exchange_id, method, path, :request, reason, start_time)
          {:error, Error.network_error(message: "Request failed: #{inspect(reason)}", exchange: exchange_id)}
      end
    rescue
      e ->
        emit_exception(exchange_id, method, path, :exception, e, start_time)
        CircuitBreaker.record_failure(exchange_id)
        {:error, Error.network_error(message: "Exception: #{Exception.message(e)}", exchange: exchange_id)}
    end
  end

  # ===========================================================================
  # Signed Request Execution
  # ===========================================================================

  @doc """
  Executes a pre-signed request.

  Used by `CCXT.Dispatch` for private endpoints after `CCXT.Signing.sign/4`
  has built the signed URL, headers, and body.

  ## Parameters

  - `exchange` - Exchange configuration struct
  - `signed` - Signed request from `CCXT.Signing.sign/4` with `:url`, `:method`, `:headers`, `:body`
  - `base_url` - Base URL to prepend to the signed path
  - `opts` - Options passed through to Req (`:timeout`, `:plug` for tests, etc.)
  """
  @spec signed_request(Exchange.t(), Signing.signed_request(), String.t(), keyword()) ::
          {:ok, response()} | {:error, Error.t()}
  def signed_request(%Exchange{} = exchange, signed, base_url, opts \\ []) do
    timeout = Keyword.get(opts, :timeout, Defaults.request_timeout_ms())
    endpoint_weight = Keyword.get(opts, :endpoint_weight, 1)
    extra_opts = Keyword.drop(opts, [:timeout, :endpoint_weight])

    with :ok <- check_circuit_breaker(exchange),
         :ok <- maybe_rate_limit(build_rate_key(exchange), exchange, endpoint_weight) do
      url = base_url <> signed.url
      body_opts = if signed.body, do: [body: signed.body], else: []

      req_opts =
        [method: signed.method, url: url, headers: signed.headers] ++
          body_opts ++
          [receive_timeout: timeout, retry: Defaults.retry_policy()] ++
          extra_opts

      # Clean path for telemetry (strip query string from signed URL)
      [telemetry_path | _] = String.split(signed.url, "?", parts: 2)
      execute_request(exchange, signed.method, telemetry_path, req_opts)
    end
  end

  # ===========================================================================
  # Request Building
  # ===========================================================================

  # Builds Req options with manual query encoding for unsigned (public) requests.
  defp build_request(method, path, params, extra_headers, base_url) do
    case method do
      m when m in [:get, :head, :delete] ->
        query_string = if params == %{}, do: "", else: "?" <> URI.encode_query(params)
        url = base_url <> path <> query_string
        [method: method, url: url, headers: extra_headers]

      m when m in [:post, :put, :patch] ->
        url = base_url <> path
        headers = [{"content-type", "application/json"}] ++ extra_headers
        body = if params == %{}, do: nil, else: Jason.encode!(params)
        body_opts = if body, do: [body: body], else: []
        [method: m, url: url, headers: headers] ++ body_opts
    end
  end

  # ===========================================================================
  # Response Handling
  # ===========================================================================

  # 2xx responses — check for HTML and body-level errors
  defp handle_response(method, status, headers, body, exchange) when status >= 200 and status < 300 do
    case detect_html_response(body, headers) do
      {:html, context} -> {:error, classify_html_response(status, context, exchange.id)}
      :not_html -> handle_success_body(method, status, headers, body, exchange)
    end
  end

  # Non-2xx responses
  defp handle_response(_method, status, headers, body, exchange) do
    case detect_html_response(body, headers) do
      {:html, context} ->
        {:error, classify_html_response(status, context, exchange.id)}

      :not_html ->
        {:error, normalize_error(status, body, exchange)}
    end
  end

  defp handle_success_body(method, status, headers, body, exchange) do
    if empty_body?(body) and status != 204 and method != :head do
      {:error, empty_body_error(status, exchange.id)}
    else
      decoded_body = ensure_json_decoded(body)

      case check_body_error(decoded_body, exchange) do
        nil -> {:ok, %{status: status, headers: headers, body: decoded_body}}
        error -> {:error, error}
      end
    end
  end

  defp empty_body_error(status, exchange_id) do
    Error.network_error(
      message: "Empty response body from exchange",
      code: status,
      exchange: exchange_id,
      raw: %{status: status},
      hints: ["Exchange returned HTTP #{status} with empty body — likely a transient transport issue"]
    )
  end

  defp empty_body?(body) when is_binary(body), do: String.trim(body) == ""
  defp empty_body?(_), do: false

  # ===========================================================================
  # HTML Response Detection
  # ===========================================================================

  defp detect_html_response(body, headers) when is_binary(body) do
    content_type = get_content_type(headers)

    cond do
      String.contains?(content_type, "text/html") -> {:html, extract_html_context(body)}
      html_body?(body) -> {:html, extract_html_context(body)}
      true -> :not_html
    end
  end

  defp detect_html_response(_body, _headers), do: :not_html

  # ===========================================================================
  # JSON Decoding Fallback
  # ===========================================================================

  # Some exchanges return JSON with Content-Type: text/plain
  defp ensure_json_decoded(body) when is_map(body) or is_list(body), do: body

  defp ensure_json_decoded(body) when is_binary(body) do
    trimmed = String.trim_leading(body)

    if String.starts_with?(trimmed, "{") or String.starts_with?(trimmed, "[") do
      case Jason.decode(body) do
        {:ok, decoded} -> decoded
        {:error, _} -> body
      end
    else
      body
    end
  end

  defp ensure_json_decoded(body), do: body

  # ===========================================================================
  # Body-Level Error Detection
  #
  # Many exchanges return HTTP 200 with error information in the body.
  # Sentinel-aware spec checks run first, then we fall back to the historical
  # exact-code probe for exchanges that still use simple top-level error codes.
  # ===========================================================================

  defp check_body_error(body, exchange) when is_map(body) do
    case detect_body_error(body, exchange) do
      {:error, code} ->
        message = extract_message(body)

        error_type =
          Map.get(exchange.error_codes, to_string(code)) ||
            match_broad_error(message, exchange.broad_error_patterns)

        build_typed_error(error_type, message, code, exchange.id)

      :ok ->
        nil
    end
  end

  defp check_body_error(_body, _exchange), do: nil

  defp detect_body_error(body, exchange) do
    case evaluate_status_sentinels(body, exchange) do
      :success -> :ok
      {:error, code} -> {:error, code}
      :unknown -> detect_error_from_code_fields(body, exchange)
    end
  end

  defp detect_error_from_code_fields(body, exchange) do
    code = extract_error_code(body, exchange.error_code_fields)
    if not is_nil(code) and error_code?(code), do: {:error, code}, else: :ok
  end

  # Code is an error when it's a non-zero, non-"OK" string or integer
  defp error_code?(code) when is_integer(code), do: code != 0
  defp error_code?(code) when is_binary(code), do: code != "0" and code != "OK"
  defp error_code?(_), do: false

  # Sentinel-bearing checks let us suppress false positives from exchanges whose
  # success values are not "0"/"OK" (e.g. KuCoin "200000", Bithumb "0000").
  # We keep scanning on error candidates so a later exact-code field can win.
  defp evaluate_status_sentinels(body, exchange) do
    exchange.error_body_checks
    |> Enum.filter(&(:status_sentinel in &1.roles))
    |> Enum.reduce_while(:unknown, &reduce_sentinel_check(&1, &2, body, exchange))
  end

  defp reduce_sentinel_check(check, best, body, exchange) do
    case extract_check_value(body, check) do
      nil -> {:cont, best}
      value -> accumulate_sentinel_result(value, check, best, exchange.error_codes)
    end
  end

  defp accumulate_sentinel_result(value, check, best, error_codes) do
    case evaluate_sentinel_value(value, check, error_codes) do
      :success -> {:halt, :success}
      {:error, _code} = candidate -> {:cont, prefer_error_candidate(candidate, best, error_codes)}
      :unknown -> {:cont, best}
    end
  end

  defp evaluate_sentinel_value(value, check, error_codes) do
    case comparable_value(value) do
      nil -> :unknown
      comparable -> classify_sentinel(value, comparable, check, error_codes)
    end
  end

  defp classify_sentinel(value, comparable, check, error_codes) do
    ctx = %{
      value: value,
      comparable: comparable,
      eq_vals: sentinel_values(check, "==="),
      neq_vals: sentinel_values(check, "!=="),
      has_code_role: :error_code in check.roles,
      known_error?: not is_nil(lookup_error_type(comparable, error_codes))
    }

    classify_by_success(ctx)
  end

  # Success first: an inequality match or a role-gated equality match with no known error code wins.
  defp classify_by_success(%{neq_vals: [_ | _] = vals, comparable: c} = ctx) when is_binary(c) do
    if c in vals, do: :success, else: classify_by_eq_success(ctx)
  end

  defp classify_by_success(ctx), do: classify_by_eq_success(ctx)

  defp classify_by_eq_success(%{eq_vals: [_ | _] = vals, comparable: c, has_code_role: true, known_error?: false} = ctx) do
    if c in vals, do: :success, else: classify_by_error(ctx)
  end

  # Sentinel-only entries have no role inversion: a === match IS the error
  # indicator. (Code-bearing entries above invert — there the eq_val is a
  # success code that suppresses the default `error_code?` heuristic.)
  defp classify_by_eq_success(%{eq_vals: [_ | _] = vals, comparable: c, has_code_role: false, value: value} = ctx) do
    if c in vals, do: {:error, value}, else: classify_by_error(ctx)
  end

  defp classify_by_eq_success(ctx), do: classify_by_error(ctx)

  # No success match: any configured sentinel means the value is an error.
  defp classify_by_error(%{neq_vals: [_ | _], value: value}), do: {:error, value}
  defp classify_by_error(%{eq_vals: [_ | _], has_code_role: true, value: value}), do: {:error, value}
  defp classify_by_error(_ctx), do: :unknown

  defp sentinel_values(check, operator) do
    check.sentinel_values
    |> Enum.filter(&(&1.operator == operator))
    |> Enum.map(& &1.value)
  end

  defp prefer_error_candidate(candidate, :unknown, _error_codes), do: candidate

  defp prefer_error_candidate({:error, new_code} = candidate, {:error, current_code}, error_codes) do
    if is_nil(lookup_error_type(current_code, error_codes)) and
         not is_nil(lookup_error_type(new_code, error_codes)) do
      candidate
    else
      {:error, current_code}
    end
  end

  defp comparable_value(value) when is_binary(value), do: value
  defp comparable_value(value) when is_integer(value), do: Integer.to_string(value)
  defp comparable_value(value) when is_atom(value), do: Atom.to_string(value)
  defp comparable_value(_), do: nil

  # Tries exchange-configured error code field names in priority order.
  # Field names come from Exchange.error_code_fields.
  defp extract_error_code(body, fields) do
    Enum.find_value(fields, &extract_field_value(body, &1))
  end

  defp extract_check_value(body, %{field: field, field2: field2}) do
    case extract_field_value(body, field) do
      nil -> extract_field_value(body, field2)
      value -> value
    end
  end

  defp extract_field_value(body, field) when is_map(body) and is_binary(field) do
    if Map.has_key?(body, field), do: Map.get(body, field)
  end

  defp extract_field_value(_body, _field), do: nil

  # ===========================================================================
  # HTTP Status Error Normalization
  # ===========================================================================

  defp normalize_error(429, body, exchange) do
    retry_after = extract_retry_after(body)

    [message: extract_message(body), retry_after: retry_after, exchange: exchange.id, raw: body]
    |> Error.rate_limit_exceeded()
    |> with_http_status(429)
  end

  defp normalize_error(status, body, exchange) when status in [401, 403] do
    [message: extract_message(body), exchange: exchange.id, raw: body]
    |> Error.authentication_error()
    |> with_http_status(status)
  end

  defp normalize_error(status, body, exchange) when is_map(body) do
    code =
      case evaluate_status_sentinels(body, exchange) do
        {:error, sentinel_code} -> sentinel_code
        _ -> extract_error_code(body, exchange.error_code_fields)
      end

    message = extract_message(body)

    # Try: exact error code → broad message substring → HTTP status mapping
    error_type =
      lookup_error_type(code, exchange.error_codes) ||
        match_broad_error(message, exchange.broad_error_patterns) ||
        Map.get(exchange.http_exceptions, to_string(status))

    error_type
    |> build_typed_error(message, code || status, exchange.id)
    |> with_http_status(status)
  end

  defp normalize_error(status, body, exchange) do
    body
    |> extract_message()
    |> Error.exchange_error(exchange: exchange.id, raw: body)
    |> with_http_status(status)
  end

  defp with_http_status(%Error{} = err, status), do: %{err | http_status: status}

  # ===========================================================================
  # Error Building Helpers
  # ===========================================================================

  defp lookup_error_type(nil, _error_codes), do: nil

  defp lookup_error_type(code, error_codes) do
    Map.get(error_codes, code) || Map.get(error_codes, to_string(code))
  end

  # Matches error message against broad exception patterns (substring matching).
  # Broad patterns are keyed by message substrings, e.g. "Insufficient balance!" => :insufficient_funds.
  defp match_broad_error(message, broad_patterns) when is_binary(message) and map_size(broad_patterns) > 0 do
    Enum.find_value(broad_patterns, fn {pattern, error_type} ->
      if String.contains?(message, pattern), do: error_type
    end)
  end

  defp match_broad_error(_message, _broad_patterns), do: nil

  @known_error_types MapSet.new([
                       :rate_limit_exceeded,
                       :authentication_error,
                       :insufficient_funds,
                       :invalid_order,
                       :order_not_found,
                       :bad_request,
                       :bad_symbol,
                       :permission_denied,
                       :exchange_not_available,
                       :operation_failed,
                       :not_supported,
                       :access_restricted,
                       :cloudflare_challenge,
                       :network_error
                     ])

  # Dispatches to the matching Error factory function, or falls back to exchange_error.
  defp build_typed_error(type, message, code, exchange_id) do
    opts = [message: message, code: code, exchange: exchange_id]

    if type in @known_error_types do
      apply(Error, type, [opts])
    else
      Error.exchange_error(message, opts)
    end
  end

  # ===========================================================================
  # Message Extraction
  # ===========================================================================

  # Safely extracts error message from response body, handling various formats
  defp extract_message(body) when is_map(body) do
    case body["message"] || body["msg"] || body["retMsg"] || body["error"] do
      nil -> "Unknown error"
      msg when is_binary(msg) -> msg
      msg -> inspect(msg)
    end
  end

  defp extract_message(body) when is_binary(body), do: body
  defp extract_message(_), do: "Unknown error"

  # ===========================================================================
  # Retry-After Extraction
  # ===========================================================================

  # Extracts retry-after from response body (some exchanges include it there)
  # TODO: Also extract from Retry-After header when headers are available
  defp extract_retry_after(body) when is_map(body) do
    case body["retry_after"] || body["retryAfter"] do
      seconds when is_integer(seconds) -> seconds * 1000
      _ -> nil
    end
  end

  defp extract_retry_after(_), do: nil

  # ===========================================================================
  # HTML Detection Helpers
  # ===========================================================================

  defp html_body?(body) do
    trimmed = String.trim_leading(body)

    String.starts_with?(trimmed, "<!DOCTYPE") or
      String.starts_with?(trimmed, "<html") or
      String.starts_with?(trimmed, "<HTML") or
      String.starts_with?(trimmed, "<!doctype")
  end

  defp extract_html_context(body) do
    %{
      page_title: extract_html_title(body),
      body_preview: String.slice(body, 0, @html_preview_length)
    }
  end

  defp extract_html_title(html) do
    case Regex.run(~r/<title[^>]*>([^<]+)<\/title>/i, html) do
      [_, title] -> String.trim(title)
      _ -> nil
    end
  end

  defp get_content_type(headers) when is_map(headers) do
    case Map.get(headers, "content-type") do
      [value | _] -> String.downcase(value)
      _ -> ""
    end
  end

  # Conservative Cloudflare-challenge markers. Anything NOT matching these stays
  # classified as :access_restricted so genuine URL/prefix bugs keep flunking in
  # integration probes. Extend this list as new CF variants surface.
  @cloudflare_title_patterns [~r/just a moment/i, ~r/attention required/i]
  @cloudflare_body_markers ["cf-chl-bypass", "challenge-platform", "cf-browser-verification"]

  defp classify_html_response(status, context, exchange_id) do
    page_title = context[:page_title]
    preview = context[:body_preview]
    raw = %{status: status, page_title: page_title, body_preview: preview}

    if cloudflare_challenge?(page_title, preview) do
      Error.cloudflare_challenge(
        message: cloudflare_message(page_title),
        code: status,
        exchange: exchange_id,
        raw: raw,
        hints: [
          "Cloudflare challenge detected — exchange is reachable but requires browser/approved client",
          "Running from an approved IP or using a CF-friendly HTTP client path may help"
        ]
      )
    else
      Error.access_restricted(
        message: access_restricted_message(page_title),
        code: status,
        exchange: exchange_id,
        raw: raw,
        hints: [
          "Verify the API URL is correct",
          "Could be geographic/IP blocking - try VPN if curl works"
        ]
      )
    end
  end

  defp cloudflare_challenge?(page_title, preview) do
    title_match?(page_title) or body_match?(preview)
  end

  defp title_match?(nil), do: false

  defp title_match?(title) when is_binary(title) do
    Enum.any?(@cloudflare_title_patterns, &Regex.match?(&1, title))
  end

  defp body_match?(nil), do: false

  defp body_match?(preview) when is_binary(preview) do
    Enum.any?(@cloudflare_body_markers, &String.contains?(preview, &1))
  end

  defp cloudflare_message(nil), do: "Cloudflare challenge page received instead of JSON API response"

  defp cloudflare_message(title), do: "Cloudflare challenge page '#{title}' received instead of JSON API response"

  defp access_restricted_message(nil), do: "Received HTML instead of JSON API response"

  defp access_restricted_message(title), do: "Received HTML page '#{title}' instead of JSON API response"

  # ===========================================================================
  # Base URL Resolution
  # ===========================================================================

  # Picks the default base URL from exchange.base_urls.
  # Handles 3 URL structures from specs:
  #   Flat:   %{"public" => "https://api.bybit.com"}
  #   Nested: %{"public" => %{"spot" => "https://api.gateio.ws/api/v4", ...}}
  #   Keyed:  %{"spot" => %{"public" => "https://api.mexc.com", ...}}
  defp default_base_url(exchange) do
    urls = exchange.base_urls

    # TODO(Phase 2): Dispatch will select the correct API version/visibility URL per-endpoint.
    # This picks the first URL found via recursive scan — good enough for connectivity, not for routing.
    cond do
      is_binary(urls["public"]) -> urls["public"]
      is_binary(urls["rest"]) -> urls["rest"]
      is_map(urls["public"]) -> find_flat_url(urls["public"])
      is_map(urls["spot"]) -> find_flat_url(urls["spot"])
      true -> find_flat_url(urls)
    end ||
      ""
  end

  defp find_flat_url(urls) when is_map(urls) do
    Enum.find_value(urls, fn
      {_key, value} when is_binary(value) -> value
      _ -> nil
    end)
  end

  # ===========================================================================
  # Base Client Caching
  # ===========================================================================

  defp get_base_client do
    case :persistent_term.get(@base_client_key, nil) do
      nil ->
        client = build_base_client()
        :persistent_term.put(@base_client_key, client)
        client

      client ->
        client
    end
  end

  defp build_base_client do
    Req.new(
      decode_body: true,
      compressed: true,
      retry: false
    )
  end

  # ===========================================================================
  # Telemetry
  # ===========================================================================

  defp emit_start(exchange_id, method, path) do
    :telemetry.execute(
      CCXT.Telemetry.request_start(),
      %{system_time: System.system_time()},
      %{exchange: exchange_id, method: method, path: path}
    )
  end

  defp emit_stop(exchange_id, method, path, status, start_time) do
    duration = System.monotonic_time() - start_time

    :telemetry.execute(
      CCXT.Telemetry.request_stop(),
      %{duration: duration},
      %{exchange: exchange_id, method: method, path: path, status: status}
    )
  end

  defp emit_exception(exchange_id, method, path, kind, reason, start_time) do
    duration = System.monotonic_time() - start_time

    :telemetry.execute(
      CCXT.Telemetry.request_exception(),
      %{duration: duration},
      %{exchange: exchange_id, method: method, path: path, kind: kind, reason: reason}
    )
  end

  # ===========================================================================
  # Rate Limiting
  # ===========================================================================

  # Wraps CircuitBreaker.check/1 to return {:error, _} for use in `with` chains
  defp check_circuit_breaker(%Exchange{id: exchange_id}) do
    case CircuitBreaker.check(exchange_id) do
      :ok -> :ok
      :blown -> {:error, Error.circuit_open(exchange: exchange_id)}
    end
  end

  # Builds rate limiter key from exchange: {exchange_id, api_key | :public}
  defp build_rate_key(%Exchange{id: id, credentials: nil}), do: {id, :public}
  defp build_rate_key(%Exchange{id: id, credentials: creds}), do: {id, creds.api_key || :public}

  # Checks rate limit if enabled — blocks until capacity is available, returns :ok.
  # rate_limit_ms is "milliseconds between requests" (CCXT convention), so
  # max requests per period = period / rate_limit_ms.
  defp maybe_rate_limit(rate_key, exchange, weight) do
    if Defaults.rate_limiter_enabled?() and exchange.rate_limit_ms > 0 do
      max_requests = trunc(@rate_limit_period_ms / exchange.rate_limit_ms)
      rate_limit = %{requests: max_requests, period: @rate_limit_period_ms}

      case RateLimiter.check_rate(rate_key, rate_limit, weight) do
        :ok ->
          :ok

        {:delay, delay_ms} ->
          emit_rate_limit_throttled(exchange.id, delay_ms, weight)
          Process.sleep(delay_ms)
          maybe_rate_limit(rate_key, exchange, weight)
      end
    else
      :ok
    end
  end

  # Parses rate limit headers from response and updates the ETS state store.
  # Returns :none when exchange doesn't send rate limit headers (OKX, Kraken, etc.)
  # — this is the normal case for most exchanges, not an error.
  defp maybe_update_rate_limit_state(exchange, resp_headers) do
    case RateLimitHeaders.parse(exchange.id, resp_headers, exchange.rate_limit_ms) do
      {:ok, info} ->
        rate_key = build_rate_key(exchange)
        RateLimiter.State.update(rate_key, info)

      :none ->
        :ok
    end
  end

  defp emit_rate_limit_throttled(exchange_id, delay_ms, cost) do
    :telemetry.execute(
      CCXT.Telemetry.rate_limiter_throttled(),
      %{delay_ms: delay_ms, cost: cost},
      %{exchange: exchange_id}
    )
  end
end