defmodule CCXT.HTTP.Client do @moduledoc """ HTTP client for exchange API requests. This module wraps Req with signing middleware and telemetry. It provides a consistent interface for making signed and unsigned requests to exchanges. ## Debug Mode Two debug options are available: - `config :ccxt_client, :debug, true` - Log exceptions with full stack traces - Pass `debug_request: true` to any request - Log full request details before sending **Security Warning**: Both modes may log sensitive data including API credentials. Only use in development environments. Never enable in production. ## Why Manual Query Encoding? This module uses `URI.encode_query/1` instead of Req's `:params` step because: 1. **Signing requires raw params** - Signing patterns need access to params before URL encoding to construct the signature string 2. **Sorted encoding** - Some exchanges require alphabetically sorted params for signature verification. The signing module handles this sorting. 3. **Consistency** - Both public and private requests use the same encoding approach, avoiding subtle bugs from different code paths. The "manual" encoding is intentional architecture, not technical debt. ## Features - **Signing middleware** - Automatically signs requests using the pattern library - **Telemetry events** - Emits `[:ccxt, :request, :start | :stop | :exception]` - **Error normalization** - Converts HTTP/exchange errors to `CCXT.Error` structs - **Retry support** - Uses Req's built-in retry with configurable attempts ## Usage spec = %CCXT.Spec{...} credentials = %CCXT.Credentials{api_key: "...", secret: "..."} # Signed request (private endpoint) {:ok, response} = CCXT.HTTP.Client.request( spec, :get, "/v5/account/wallet-balance", params: %{accountType: "UNIFIED"}, credentials: credentials ) # Unsigned request (public endpoint) {:ok, response} = CCXT.HTTP.Client.request( spec, :get, "/v5/market/tickers", params: %{category: "spot"} ) ## Telemetry Events See `CCXT.Telemetry` for the full event contract. """ alias CCXT.CircuitBreaker alias CCXT.Credentials alias CCXT.Defaults alias CCXT.Error alias CCXT.HTTP.RateLimiter alias CCXT.HTTP.RateLimitHeaders alias CCXT.HTTP.RateLimitState alias CCXT.Signing alias CCXT.Spec require Logger @default_cost 1 @html_preview_length 200 @base_client_key {__MODULE__, :base_client} @debug_body_inspect_limit 500 @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 - `spec` - Exchange specification 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: `%{}`) - `:credentials` - `CCXT.Credentials` for authenticated endpoints - `:timeout` - Request timeout in milliseconds (default: see `CCXT.Defaults.request_timeout_ms/0`) - `:retry` - Retry strategy (default: `:safe_transient` - retries on safe, transient errors) - `:cost` - Request weight/cost for rate limiting (default: 1) - `:rate_limit` - Override rate limit config (default: uses spec.rate_limits) ## Returns - `{:ok, response}` - Successful response with `:status`, `:headers`, `:body` - `{:error, %CCXT.Error{}}` - Normalized error """ @spec request(Spec.t(), atom(), String.t(), keyword()) :: {:ok, response()} | {:error, Error.t()} def request(%Spec{} = spec, method, path, opts \\ []) do # Note: Param mappings are applied at endpoint level in the generator (endpoints.ex:125) # Spec-level mappings caused conflicts (e.g., KuCoin symbol→symbols applied to fetch_ticker) params = Keyword.get(opts, :params, %{}) credentials = Keyword.get(opts, :credentials) timeout = Keyword.get(opts, :timeout, Defaults.request_timeout_ms()) retry = Keyword.get(opts, :retry, Defaults.retry_policy()) cost = Keyword.get(opts, :cost, @default_cost) rate_limit = Keyword.get(opts, :rate_limit, spec.rate_limits) # Custom base URL for endpoints using different API sections (e.g., Kraken Futures history) custom_base_url = Keyword.get(opts, :base_url) # Debug request logging (logs full request details before sending) debug_request = Keyword.get(opts, :debug_request, false) # Extra opts for testing (e.g., plug: {Req.Test, :stub_name}) extra_opts = Keyword.drop(opts, [ :params, :credentials, :timeout, :retry, :cost, :rate_limit, :base_url, :debug_request, :normalize ]) # exchange_id is pre-computed at compile time by the generator macro exchange = spec.exchange_id sandbox = credentials && credentials.sandbox # Circuit breaker check FIRST: fast fail if exchange is down # This must come before rate limiting to avoid consuming capacity for requests # that will be rejected anyway case CircuitBreaker.check(exchange) do :blown -> {:error, Error.circuit_open(exchange: exchange)} :ok -> # Rate limiting: key is {exchange, api_key} for auth, {exchange, :public} for public rate_key = build_rate_key(exchange, credentials) :ok = RateLimiter.wait_for_capacity(rate_key, rate_limit, cost) # Use custom base URL if provided (for endpoints with different API sections), # otherwise use spec's default URL (with sandbox support) base_url = custom_base_url || Spec.api_url(spec, sandbox || false) request_opts = %{ base_url: base_url, timeout: timeout, retry: retry, extra_opts: extra_opts, debug_request: debug_request } do_request(spec, method, path, params, credentials, request_opts) end end # Performs the actual HTTP request after rate limiting and circuit breaker checks pass @doc false defp do_request(spec, method, path, params, credentials, request_opts) do %{base_url: base_url, timeout: timeout, retry: retry, extra_opts: extra_opts, debug_request: debug_request} = request_opts exchange = spec.exchange_id start_time = System.monotonic_time() emit_start(exchange, method, path) result = try do req_opts = build_request(spec, method, path, params, credentials, base_url) req_opts = Keyword.merge(req_opts, receive_timeout: timeout, retry: retry) req_opts = Keyword.merge(req_opts, extra_opts) maybe_log_debug_request(debug_request, exchange, method, req_opts, timeout, retry) base_client = get_base_client() result = Req.request(base_client, req_opts) # Record circuit breaker result using should_melt?/1 logic CircuitBreaker.record_result(exchange, result) case result do {:ok, %Req.Response{status: status, headers: headers, body: body}} -> rate_limit_info = maybe_parse_rate_limit(exchange, credentials, headers, spec.rate_limits) emit_stop(exchange, method, path, status, start_time, rate_limit_info) handle_response(status, headers, body, spec) {:error, %Req.TransportError{reason: reason}} -> emit_exception(exchange, method, path, :transport, reason, start_time) {:error, Error.network_error(message: "Transport error: #{inspect(reason)}", exchange: exchange)} {:error, reason} -> emit_exception(exchange, method, path, :request, reason, start_time) {:error, Error.network_error(message: "Request failed: #{inspect(reason)}", exchange: exchange)} end rescue e -> if Application.get_env(:ccxt_client, :debug, false) do Logger.error(""" CCXT request exception: Exchange: #{exchange} Method: #{method} Path: #{path} #{Exception.format(:error, e, __STACKTRACE__)} """) end emit_exception(exchange, method, path, :exception, e, start_time) # Exceptions trip the circuit breaker CircuitBreaker.record_failure(exchange) {:error, Error.network_error(message: "Exception: #{Exception.message(e)}", exchange: exchange)} end result end @doc """ Makes a raw HTTP request without signing or error normalization. Use this for debugging or when you need full control over the request. ## Parameters - `method` - HTTP method - `url` - Full URL to request - `headers` - Request headers - `body` - Request body (or nil) ## Options - `:timeout` - Request timeout in milliseconds """ @spec raw_request(atom(), String.t(), [{String.t(), String.t()}], String.t() | nil, keyword()) :: {:ok, response()} | {:error, term()} def raw_request(method, url, headers, body, opts \\ []) do timeout = Keyword.get(opts, :timeout, Defaults.request_timeout_ms()) extra_opts = Keyword.delete(opts, :timeout) req_opts = [ method: method, url: url, headers: headers, body: body, receive_timeout: timeout, retry: false ] ++ extra_opts case Req.request(req_opts) do {:ok, %Req.Response{status: status, headers: resp_headers, body: resp_body}} -> {:ok, %{status: status, headers: resp_headers, body: resp_body}} {:error, reason} -> {:error, reason} end end # Build request options, applying signing if credentials provided @doc false # Builds request options for public (unsigned) requests defp build_request(spec, method, path, params, nil, base_url) do # Public request - no signing query_string = if params == %{}, do: "", else: "?" <> URI.encode_query(params) url = base_url <> path <> query_string # Base headers with Content-Type headers = [{"Content-Type", "application/json"}] # Apply static headers and user agent from spec.http_config (Task 16d) headers = apply_http_config(headers, spec.http_config) [ method: method, url: url, headers: headers ] end defp build_request(spec, method, path, params, %Credentials{} = credentials, base_url) do # Private request - apply signing signing_config = spec.signing || %{} pattern = Map.get(signing_config, :pattern, :hmac_sha256_headers) request = %{ method: method, path: path, body: if(method in [:post, :put] && params != %{}, do: Jason.encode!(params)), params: params } signed = Signing.sign(pattern, request, credentials, signing_config) url = base_url <> signed.url # Apply static headers and user agent from spec.http_config (Task 16d) # These are merged after signing to ensure required headers are always present headers = apply_http_config(signed.headers, spec.http_config) # Apply broker header for volume attribution (Task 100) headers = apply_broker_header(headers, signing_config[:broker_config], spec.options) body_opts = if signed.body, do: [body: signed.body], else: [] [ method: signed.method, url: url, headers: headers ] ++ body_opts end # Apply static headers and user agent from http_config (Task 16d) # Some exchanges require specific headers (partner IDs, API versions) or user agents @spec apply_http_config([{String.t(), String.t()}], map() | nil) :: [{String.t(), String.t()}] defp apply_http_config(headers, nil), do: headers defp apply_http_config(headers, http_config) when is_map(http_config) do headers |> apply_static_headers(http_config[:headers]) |> apply_user_agent(http_config[:user_agent]) end # Apply static headers from http_config (e.g., partner IDs, API versions) @spec apply_static_headers([{String.t(), String.t()}], map() | nil) :: [{String.t(), String.t()}] defp apply_static_headers(headers, nil), do: headers defp apply_static_headers(headers, static_headers) when is_map(static_headers) do # Convert map to list of tuples and append to existing headers static_list = Enum.map(static_headers, fn {k, v} -> {to_string(k), to_string(v)} end) headers ++ static_list end # Apply custom User-Agent from http_config @spec apply_user_agent([{String.t(), String.t()}], String.t() | nil) :: [{String.t(), String.t()}] defp apply_user_agent(headers, nil), do: headers defp apply_user_agent(headers, user_agent) when is_binary(user_agent) do # Only add User-Agent if not already present if Enum.any?(headers, fn {k, _} -> String.downcase(k) == "user-agent" end) do headers else headers ++ [{"User-Agent", user_agent}] end end # Apply broker header for volume attribution (Task 100) # Exchanges inject broker headers for referral program volume tracking. # Priority: 1) Application config override, 2) spec.options default from CCXT @spec apply_broker_header([{String.t(), String.t()}], map() | nil, map() | nil) :: [{String.t(), String.t()}] defp apply_broker_header(headers, nil, _options), do: headers defp apply_broker_header(headers, broker_config, options) when is_map(broker_config) do header_name = broker_config[:header] option_key = broker_config[:option_key] # Priority 1: Global application config override broker_id = Application.get_env(:ccxt_client, :broker_id) # Priority 2: Default from spec.options (extracted from CCXT) broker_id = if is_nil(broker_id) and is_map(options) do # Try both string and atom keys since options may have either options[option_key] || options[String.to_atom(option_key)] else broker_id end if broker_id && header_name do headers ++ [{header_name, to_string(broker_id)}] else headers end end # Handle response, normalizing errors # Task 37: Also check for body-level errors in 2xx responses @spec handle_response(integer(), response_headers(), term(), Spec.t()) :: {:ok, response()} | {:error, Error.t()} defp handle_response(status, headers, body, spec) when status >= 200 and status < 300 do # First check for HTML responses (indicates geo-blocking or access restriction) case detect_html_response(body, headers) do {:html, context} -> {:error, build_access_restricted_error(status, context, spec.exchange_id)} :not_html -> # Ensure body is decoded as JSON (some exchanges return text/plain with JSON content) decoded_body = ensure_json_decoded(body) # Check for body-level errors (many exchanges return HTTP 200 with error in body) case check_body_error( decoded_body, spec.response_error, spec.error_codes, spec.error_code_details, spec.exchange_id ) do nil -> {:ok, %{status: status, headers: headers, body: decoded_body}} error -> {:error, error} end end end defp handle_response(status, headers, body, spec) do # Check for HTML responses first (403/404 with HTML error pages) case detect_html_response(body, headers) do {:html, context} -> {:error, build_access_restricted_error(status, context, spec.exchange_id)} :not_html -> # exchange_id is pre-computed at compile time by the generator macro error = normalize_error( status, body, spec.error_codes, spec.error_code_details, spec.exchange_id, headers ) {:error, error} end end # ============================================================================= # HTML Response Detection (Geographic/Access Restrictions) # # When exchanges block access (geo-restrictions, IP blocks, Cloudflare), # they often return HTML instead of JSON. This detects such responses # and provides clear error messages instead of dumping raw HTML. # ============================================================================= @doc false # Detects if response body is HTML (indicates geo-blocking or access restriction) @spec detect_html_response(term(), response_headers()) :: {:html, map()} | :not_html 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 (e.g., BitMEX orderBook). # Req only auto-decodes when Content-Type is application/json, so we need to # handle this case manually. # ============================================================================= @doc false # Ensures body is decoded as JSON, handling text/plain responses with JSON content @spec ensure_json_decoded(term()) :: term() 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 @doc false # Checks if body starts with HTML markers defp html_body?(body) do trimmed = String.trim_leading(body) String.starts_with?(trimmed, " content from HTML defp extract_html_title(html) do case Regex.run(~r/