tango/

transport.rs

//! Request loop, retry policy, header parsing, and validation-message extraction.
//!
//! This module is the core of the SDK. Public surface ([`RateLimitInfo`]) is
//! re-exported from `crate`. Internals are crate-private — resource methods
//! call [`Client::request_json`](crate::Client::request_json) and friends, not
//! this module directly.

use crate::error::{Error, ErrorBody, Result};
use reqwest::{header::HeaderMap, Method, RequestBuilder, Response, StatusCode};
use serde::de::DeserializeOwned;
use serde_json::Value;
use std::time::Duration;

pub(crate) const MAX_BACKOFF: Duration = Duration::from_secs(10);

pub(crate) const API_KEY_HEADER: &str = "X-API-KEY";
pub(crate) const RATE_LIMIT_REMAINING_HDR: &str = "x-ratelimit-remaining";
pub(crate) const RATE_LIMIT_LIMIT_HDR: &str = "x-ratelimit-limit";
pub(crate) const RATE_LIMIT_RESET_HDR: &str = "x-ratelimit-reset";
pub(crate) const RATE_LIMIT_TYPE_HDR: &str = "x-ratelimit-type";
pub(crate) const RETRY_AFTER_HDR: &str = "retry-after";

/// A snapshot of the rate-limit headers from the most recent response.
///
/// Returned by [`Client::rate_limit_info`](crate::Client::rate_limit_info).
/// Any field set to `None` was either absent from the response or could not
/// be parsed as an integer.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct RateLimitInfo {
    /// `X-RateLimit-Remaining` — calls remaining in the current bucket.
    pub remaining: Option<i64>,
    /// `X-RateLimit-Limit` — the maximum calls in the current bucket.
    pub limit: Option<i64>,
    /// `X-RateLimit-Reset` — seconds until the bucket refills.
    pub reset_in: Option<i64>,
    /// `Retry-After` — server-suggested wait before retrying, in seconds.
    pub retry_after: Option<i64>,
    /// `X-RateLimit-Type` — bucket discriminator (e.g. `"minute"`, `"hour"`).
    pub limit_type: Option<String>,
}

impl RateLimitInfo {
    pub(crate) fn from_headers(h: &HeaderMap) -> Self {
        Self {
            remaining: parse_int_header(h, RATE_LIMIT_REMAINING_HDR),
            limit: parse_int_header(h, RATE_LIMIT_LIMIT_HDR),
            reset_in: parse_int_header(h, RATE_LIMIT_RESET_HDR),
            retry_after: parse_int_header(h, RETRY_AFTER_HDR),
            limit_type: h
                .get(RATE_LIMIT_TYPE_HDR)
                .and_then(|v| v.to_str().ok())
                .map(str::to_string),
        }
    }
}

fn parse_int_header(h: &HeaderMap, key: &str) -> Option<i64> {
    h.get(key)
        .and_then(|v| v.to_str().ok())
        .map(str::trim)
        .and_then(|s| s.parse::<i64>().ok())
}

/// Parse the `Retry-After` header into a duration. Accepts both delta-seconds
/// and HTTP-date forms; returns `Duration::ZERO` on parse failure. The value
/// is capped at [`MAX_BACKOFF`].
pub(crate) fn parse_retry_after(h: &HeaderMap) -> Duration {
    let Some(raw) = h.get(RETRY_AFTER_HDR).and_then(|v| v.to_str().ok()) else {
        return Duration::ZERO;
    };
    let raw = raw.trim();
    if let Ok(secs) = raw.parse::<u64>() {
        return Duration::from_secs(secs).min(MAX_BACKOFF);
    }
    // HTTP-date form. We don't pull in chrono; httpdate would be a dep we want
    // to avoid. The Tango server emits delta-seconds in practice, so we accept
    // a parse failure here and let the exponential backoff take over.
    Duration::ZERO
}

/// Extract a human-readable validation message from a 400 response body.
///
/// Mirrors the Go SDK's `extractValidationMessage`. Tries the envelope keys
/// `detail`, `message`, `error` in priority order; falls back to sorted-key
/// iteration over the remaining keys (preferring array values over strings),
/// so the surfaced message is deterministic across runs.
pub(crate) fn extract_validation_message(body: Option<&Value>) -> String {
    const FALLBACK: &str = "invalid request parameters";
    let Some(Value::Object(map)) = body else {
        return FALLBACK.to_string();
    };
    for key in ["detail", "message", "error"] {
        if let Some(Value::String(s)) = map.get(key) {
            if !s.is_empty() {
                return format!("invalid request parameters: {s}");
            }
        }
    }
    // Field-error shape — walk keys in sorted order for determinism.
    let mut keys: Vec<&String> = map.keys().collect();
    keys.sort();
    for k in keys {
        match map.get(k) {
            Some(Value::Array(arr)) if !arr.is_empty() => {
                if let Some(Value::String(s)) = arr.first() {
                    if !s.is_empty() {
                        return format!("invalid request parameters: {s}");
                    }
                }
            }
            Some(Value::String(s)) if !s.is_empty() => {
                return format!("invalid request parameters: {s}");
            }
            _ => {}
        }
    }
    FALLBACK.to_string()
}

/// Body the SDK sends with a request. Used by [`Client::request_json`] and
/// [`Client::request_bytes`] callers.
#[derive(Debug)]
pub(crate) enum Body<'a> {
    None,
    Json(&'a Value),
}

pub(crate) async fn send_with_retries(
    inner: &crate::client::ClientInner,
    method: Method,
    url: reqwest::Url,
    body: Body<'_>,
) -> Result<Vec<u8>> {
    send_with_retries_and_headers(inner, method, url, body)
        .await
        .map(|(bytes, _)| bytes)
}

/// Like [`send_with_retries`], but also return the successful response's headers.
pub(crate) async fn send_with_retries_and_headers(
    inner: &crate::client::ClientInner,
    method: Method,
    url: reqwest::Url,
    body: Body<'_>,
) -> Result<(Vec<u8>, HeaderMap)> {
    // Pre-serialize the JSON body once; we re-use the bytes across attempts.
    let body_bytes = match body {
        Body::None => None,
        Body::Json(v) => Some(serde_json::to_vec(v)?),
    };
    retry(inner, || {
        attempt_once(inner, method.clone(), url.clone(), body_bytes.as_deref())
    })
    .await
}

/// GET `url` on the no-redirect client and return the resolved `Location` of
/// its 3xx response, under the same retry policy as every other request.
pub(crate) async fn redirect_location_with_retries(
    inner: &crate::client::ClientInner,
    url: reqwest::Url,
) -> Result<String> {
    retry(inner, || attempt_redirect(inner, url.clone())).await
}

async fn retry<T, F, Fut>(inner: &crate::client::ClientInner, mut op: F) -> Result<T>
where
    F: FnMut() -> Fut,
    Fut: std::future::Future<Output = Result<T>>,
{
    let max_attempts = inner.retries.saturating_add(1);
    let mut attempt: u32 = 0;

    loop {
        let err = match op().await {
            Ok(v) => return Ok(v),
            Err(e) => e,
        };

        if !err.is_retryable() || attempt + 1 >= max_attempts {
            return Err(err);
        }

        let wait = if let Error::RateLimit { retry_after, .. } = &err {
            let r = u64::from(*retry_after);
            if r > 0 {
                Duration::from_secs(r).min(MAX_BACKOFF)
            } else {
                backoff_for(inner.retry_backoff, attempt)
            }
        } else {
            backoff_for(inner.retry_backoff, attempt)
        };

        tokio::time::sleep(wait).await;
        attempt += 1;
    }
}

fn backoff_for(base: Duration, attempt: u32) -> Duration {
    let mult = 1u32 << attempt.min(6); // cap shift at 6 to avoid overflow
    base.saturating_mul(mult).min(MAX_BACKOFF)
}

fn build_request(
    inner: &crate::client::ClientInner,
    http: &reqwest::Client,
    method: Method,
    url: reqwest::Url,
    body_bytes: Option<&[u8]>,
) -> RequestBuilder {
    let mut req: RequestBuilder = http.request(method, url);
    req = req.header(reqwest::header::ACCEPT, "application/json");
    if !inner.api_key.is_empty() {
        req = req.header(API_KEY_HEADER, &inner.api_key);
    }
    if !inner.user_agent.is_empty() {
        req = req.header(reqwest::header::USER_AGENT, &inner.user_agent);
    }
    if let Some(bytes) = body_bytes {
        req = req
            .header(reqwest::header::CONTENT_TYPE, "application/json")
            .body(bytes.to_vec());
    }
    if !inner.timeout.is_zero() {
        req = req.timeout(inner.timeout);
    }
    req
}

/// Send `req` and return its status, headers and body, recording the headers
/// for [`Client::rate_limit_info`](crate::Client::rate_limit_info).
async fn send_raw(
    inner: &crate::client::ClientInner,
    req: RequestBuilder,
) -> Result<(StatusCode, HeaderMap, Vec<u8>)> {
    let resp: Response = match req.send().await {
        Ok(r) => r,
        Err(e) => {
            if e.is_timeout() {
                return Err(Error::Timeout {
                    timeout: inner.timeout,
                });
            }
            return Err(Error::Transport(e));
        }
    };

    let status = resp.status();
    let headers = resp.headers().clone();
    // Snapshot rate-limit/headers for observability.
    inner.set_last_response(&headers);

    let bytes = match resp.bytes().await {
        Ok(b) => b.to_vec(),
        Err(e) => return Err(Error::Transport(e)),
    };
    Ok((status, headers, bytes))
}

async fn attempt_once(
    inner: &crate::client::ClientInner,
    method: Method,
    url: reqwest::Url,
    body_bytes: Option<&[u8]>,
) -> Result<(Vec<u8>, HeaderMap)> {
    let req = build_request(inner, &inner.http, method, url, body_bytes);
    let (status, headers, bytes) = send_raw(inner, req).await?;
    if status.is_success() {
        return Ok((bytes, headers));
    }
    Err(decode_error(status, &headers, &bytes))
}

async fn attempt_redirect(inner: &crate::client::ClientInner, url: reqwest::Url) -> Result<String> {
    let req = build_request(
        inner,
        &inner.http_no_redirect,
        Method::GET,
        url.clone(),
        None,
    );
    let (status, headers, bytes) = send_raw(inner, req).await?;
    if status.is_redirection() {
        let location = headers
            .get(reqwest::header::LOCATION)
            .and_then(|v| v.to_str().ok())
            .filter(|s| !s.is_empty())
            .ok_or_else(|| Error::Api {
                status: status.as_u16(),
                message: "redirect response carried no Location header".into(),
                response: None,
            })?;
        return url
            .join(location)
            .map(String::from)
            .map_err(|e| Error::Build(format!("parse redirect location {location}: {e}")));
    }
    if status.is_success() {
        return Err(Error::Api {
            status: status.as_u16(),
            message: format!("expected a redirect, got status {}", status.as_u16()),
            response: None,
        });
    }
    Err(decode_error(status, &headers, &bytes))
}

fn decode_error(status: StatusCode, headers: &HeaderMap, body: &[u8]) -> Error {
    let parsed_value: Option<Value> = if body.is_empty() {
        None
    } else {
        serde_json::from_slice(body).ok()
    };
    let body_message = parsed_value.as_ref().and_then(extract_reason);
    let response = parsed_value.as_ref().map(|v| ErrorBody {
        message: body_message.clone().unwrap_or_default(),
        raw: Some(v.clone()),
    });
    match status.as_u16() {
        401 => Error::Auth { response },
        404 => Error::NotFound { response },
        400 => Error::Validation {
            message: extract_validation_message(parsed_value.as_ref()),
            response,
        },
        429 => {
            let retry_after = parse_retry_after(headers).as_secs();
            let detail = parsed_value
                .as_ref()
                .and_then(|v| v.get("detail"))
                .and_then(Value::as_str);
            Error::RateLimit {
                retry_after: u32::try_from(retry_after).unwrap_or(u32::MAX),
                limit_type: headers
                    .get(RATE_LIMIT_TYPE_HDR)
                    .and_then(|v| v.to_str().ok())
                    .filter(|s| !s.is_empty())
                    .map(str::to_string)
                    .or_else(|| detail.and_then(limit_type_from_detail).map(str::to_string)),
                response,
            }
        }
        code => Error::Api {
            status: code,
            message: body_message
                .unwrap_or_else(|| format!("API request failed with status {code}")),
            response,
        },
    }
}

/// Which limit a 429 tripped, read from the API's detail: `"daily"`, `"burst"`, or `None` when the detail names neither.
///
/// The API names the rate in `Rate limit exceeded for daily. Please try again in 16 hours, 58 minutes.`, so only the name after `exceeded for` is read: the wait can say "minutes" on a daily lockout.
fn limit_type_from_detail(detail: &str) -> Option<&'static str> {
    const MARKER: &str = "exceeded for ";
    let lower = detail.to_lowercase();
    let is_name_char = |c: char| c.is_alphanumeric() || c == '_' || c == '-';
    let name = lower.find(MARKER).and_then(|at| {
        let rest = &lower[at + MARKER.len()..];
        let end = rest.find(|c: char| !is_name_char(c)).unwrap_or(rest.len());
        (end > 0).then(|| &rest[..end])
    });
    let text = name.unwrap_or(&lower);
    if text.contains("daily") || contains_word(text, "day") {
        Some("daily")
    } else if text.contains("burst") || text.contains("minute") {
        Some("burst")
    } else {
        None
    }
}

fn contains_word(text: &str, word: &str) -> bool {
    let is_word_char = |c: char| c.is_alphanumeric() || c == '_';
    text.match_indices(word).any(|(at, _)| {
        let before = text[..at].chars().next_back();
        let after = text[at + word.len()..].chars().next();
        !before.is_some_and(is_word_char) && !after.is_some_and(is_word_char)
    })
}

/// The server's own explanation from an error body, e.g. why a 403 refused the request.
///
/// The API sends either `{"detail": "..."}` or a structured body such as a tier gate's `{"error": "...", "code": "tier_required", ...}`, sometimes nested under `detail`.
fn extract_reason(v: &Value) -> Option<String> {
    let Value::Object(map) = v else { return None };
    let envelope_reason = |map: &serde_json::Map<String, Value>| {
        ["error", "message", "detail"]
            .iter()
            .find_map(|key| match map.get(*key) {
                Some(Value::String(s)) if !s.is_empty() => Some(s.clone()),
                _ => None,
            })
    };
    envelope_reason(map).or_else(|| match map.get("detail") {
        Some(Value::Object(nested)) => envelope_reason(nested),
        _ => None,
    })
}

/// Decode a JSON byte slice into a typed `T`, wrapping decode errors so they
/// surface as [`Error::Decode`].
pub(crate) fn decode_json<T: DeserializeOwned>(bytes: &[u8]) -> Result<T> {
    serde_json::from_slice(bytes).map_err(Error::Decode)
}

/// Decode a JSON byte slice, treating an empty body as a default-constructed
/// value. Useful for endpoints that legally return `204 No Content`.
pub(crate) fn decode_json_or_default<T>(bytes: &[u8]) -> Result<T>
where
    T: DeserializeOwned + Default,
{
    if bytes.is_empty() {
        return Ok(T::default());
    }
    decode_json(bytes)
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn extract_envelope_detail() {
        let body = json!({"detail": "no soup for you"});
        assert_eq!(
            extract_validation_message(Some(&body)),
            "invalid request parameters: no soup for you"
        );
    }

    #[test]
    fn extract_envelope_message() {
        let body = json!({"message": "bad input"});
        assert_eq!(
            extract_validation_message(Some(&body)),
            "invalid request parameters: bad input"
        );
    }

    #[test]
    fn extract_field_errors_sorted() {
        let body = json!({
            "zebra": ["last alphabetically"],
            "apple": ["first alphabetically"],
        });
        // sorted keys => `apple` wins, deterministically.
        assert_eq!(
            extract_validation_message(Some(&body)),
            "invalid request parameters: first alphabetically"
        );
    }

    #[test]
    fn extract_falls_back() {
        let body = json!({});
        assert_eq!(
            extract_validation_message(Some(&body)),
            "invalid request parameters"
        );
        assert_eq!(
            extract_validation_message(None),
            "invalid request parameters"
        );
    }

    #[test]
    fn extract_prefers_envelope_over_field() {
        let body = json!({
            "detail": "envelope wins",
            "apple": ["field loses"],
        });
        assert_eq!(
            extract_validation_message(Some(&body)),
            "invalid request parameters: envelope wins"
        );
    }

    #[test]
    fn extract_string_field_error() {
        let body = json!({"piid": "must be present"});
        assert_eq!(
            extract_validation_message(Some(&body)),
            "invalid request parameters: must be present"
        );
    }

    #[test]
    fn limit_type_reads_the_rate_name_not_the_wait() {
        let cases = [
            (
                "Rate limit exceeded for daily. Please try again in 16 hours, 58 minutes.",
                Some("daily"),
            ),
            (
                "Rate limit exceeded for daily. Please try again in 1 minute.",
                Some("daily"),
            ),
            (
                "Rate limit exceeded for burst. Please try again in 45 seconds.",
                Some("burst"),
            ),
            (
                "Rate limit exceeded for burst. Please try again in 1 minute.",
                Some("burst"),
            ),
            (
                "Rate limit exceeded for premium_daily. Please try again in 3 hours.",
                Some("daily"),
            ),
            (
                "Rate limit exceeded for burst. Please try again in 1 day.",
                Some("burst"),
            ),
            ("Too many requests.", None),
            ("", None),
        ];
        for (detail, expected) in cases {
            assert_eq!(limit_type_from_detail(detail), expected, "{detail}");
        }
    }

    fn rate_limit_type(headers: &HeaderMap, body: &[u8]) -> Option<String> {
        match decode_error(StatusCode::TOO_MANY_REQUESTS, headers, body) {
            Error::RateLimit { limit_type, .. } => limit_type,
            other => panic!("expected Error::RateLimit, got {other:?}"),
        }
    }

    #[test]
    fn rate_limit_type_from_body_detail() {
        let body = br#"{"detail": "Rate limit exceeded for daily. Please try again in 16 hours, 58 minutes.", "wait_in_seconds": 61080}"#;
        assert_eq!(
            rate_limit_type(&HeaderMap::new(), body).as_deref(),
            Some("daily")
        );
    }

    #[test]
    fn rate_limit_type_none_without_a_recognisable_detail() {
        let h = HeaderMap::new();
        assert_eq!(
            rate_limit_type(&h, br#"{"detail": "Too many requests."}"#),
            None
        );
        assert_eq!(
            rate_limit_type(&h, br#"{"detail": {"error": "daily"}}"#),
            None
        );
        assert_eq!(rate_limit_type(&h, b"<html>daily burst</html>"), None);
        assert_eq!(rate_limit_type(&h, b""), None);
    }

    #[test]
    fn rate_limit_type_header_wins_over_detail() {
        let body = br#"{"detail": "Rate limit exceeded for daily. Please try again in 2 hours."}"#;
        let mut h = HeaderMap::new();
        h.insert(RATE_LIMIT_TYPE_HDR, "burst".parse().unwrap());
        assert_eq!(rate_limit_type(&h, body).as_deref(), Some("burst"));
        h.insert(RATE_LIMIT_TYPE_HDR, "".parse().unwrap());
        assert_eq!(rate_limit_type(&h, body).as_deref(), Some("daily"));
    }

    fn api_error(status: u16, body: &[u8]) -> (u16, String, Option<ErrorBody>, String) {
        let err = decode_error(
            StatusCode::from_u16(status).unwrap(),
            &HeaderMap::new(),
            body,
        );
        let display = err.to_string();
        match err {
            Error::Api {
                status,
                message,
                response,
            } => (status, message, response, display),
            other => panic!("expected Error::Api, got {other:?}"),
        }
    }

    #[test]
    fn api_error_carries_the_server_reason() {
        let body = json!({"detail": "These shape fields are not allowed for your account."});
        let (status, message, response, display) = api_error(403, body.to_string().as_bytes());
        assert_eq!(status, 403);
        assert_eq!(
            message,
            "These shape fields are not allowed for your account."
        );
        assert_eq!(
            display,
            "tango: API error (status 403): These shape fields are not allowed for your account."
        );
        let response = response.unwrap();
        assert_eq!(response.message, message);
        assert_eq!(response.raw, Some(body));
    }

    #[test]
    fn api_error_tier_gate_reason_flat_or_nested() {
        let reason = "The 'psc' filter requires a Small plan or higher.";
        for body in [
            json!({"error": reason, "code": "tier_required"}),
            json!({"detail": {"error": reason, "code": "tier_required"}}),
        ] {
            let (_, message, response, _) = api_error(403, body.to_string().as_bytes());
            assert_eq!(message, reason, "body: {body}");
            assert_eq!(response.unwrap().raw, Some(body));
        }
    }

    #[test]
    fn api_error_json_body_without_a_reason_keeps_the_body() {
        let body = json!({"code": "forbidden"});
        let (_, message, response, _) = api_error(403, body.to_string().as_bytes());
        assert_eq!(message, "API request failed with status 403");
        let response = response.unwrap();
        assert_eq!(response.message, "");
        assert_eq!(response.raw, Some(body));
    }

    #[test]
    fn api_error_non_json_body_keeps_the_bare_message() {
        let (status, message, response, _) = api_error(502, b"<html>Bad Gateway</html>");
        assert_eq!(status, 502);
        assert_eq!(message, "API request failed with status 502");
        assert_eq!(response, None);
    }

    #[test]
    fn backoff_doubles_and_caps() {
        let base = Duration::from_millis(250);
        assert_eq!(backoff_for(base, 0), Duration::from_millis(250));
        assert_eq!(backoff_for(base, 1), Duration::from_millis(500));
        assert_eq!(backoff_for(base, 2), Duration::from_secs(1));
        assert_eq!(backoff_for(base, 3), Duration::from_secs(2));
        assert_eq!(backoff_for(base, 4), Duration::from_secs(4));
        assert_eq!(backoff_for(base, 5), Duration::from_secs(8));
        assert_eq!(backoff_for(base, 6), MAX_BACKOFF);
        assert_eq!(backoff_for(base, 50), MAX_BACKOFF);
    }
}