steam-user 0.2.4

//! Remote Steam user client with round-robin load balancing and retry logic.

use std::{
    sync::atomic::{AtomicUsize, Ordering},
    time::Duration,
};

use reqwest::Client;
use serde::{Deserialize, Serialize};

use super::error::RemoteSteamUserError;
use super::health::{self, HealthTransition, NodeOutcome};

/// Default maximum number of retry attempts per request.
const DEFAULT_MAX_RETRIES: usize = 5;

/// Default HTTP request timeout.
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(300);

/// Default connect timeout for the HTTP client.
const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(30);

/// Process-global round-robin seed.
///
/// Each new [`RemoteSteamUser`] seeds its per-instance `index` from this so
/// freshly-constructed clients start at *different* nodes. Without it, every
/// client starts at `index = 0`, so callers that build a fresh client per
/// request (the common pattern) send their first attempt — and, for terminal
/// 4xx responses that never rotate, their *only* attempt — to `urls[0]` every
/// time. That funnels all single-shot traffic onto one node, which Steam then
/// rate-limits hardest.
static ROUND_ROBIN_SEED: AtomicUsize = AtomicUsize::new(0);

/// Build a reqwest [`Client`] with the project-standard timeouts and TLS hardening.
///
/// On failure (which is rare — only triggered by TLS backend init issues),
/// logs the error and returns a non-functional client built from
/// [`Client::new`] so callers do not panic. Subsequent requests will surface
/// the underlying problem on first use.
fn build_http_client() -> Client {
    Client::builder()
        .connect_timeout(DEFAULT_CONNECT_TIMEOUT)
        .timeout(DEFAULT_TIMEOUT)
        .min_tls_version(reqwest::tls::Version::TLS_1_2)
        .https_only(true)
        .build()
        .unwrap_or_else(|e| {
            tracing::error!(error = %e, "RemoteSteamUser: failed to build reqwest client; using default client");
            Client::new()
        })
}

/// API response envelope returned by `steam-user-api`.
#[derive(Debug, Deserialize)]
struct ApiResponse {
    success: bool,
    data: Option<serde_json::Value>,
    error: Option<String>,
}

/// Auth payload sent with every request.
#[derive(Debug, Serialize, Clone)]
struct AuthPayload {
    cookies: Vec<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    access_token: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    refresh_token: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    mobile_access_token: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    identity_secret: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    shared_secret: Option<String>,
}

/// Remote Steam user client that delegates operations to a `steam-user-api`
/// server.
///
/// Supports multiple server URLs with round-robin selection and automatic
/// retry.
///
/// # Example
///
/// ```rust,no_run
/// use steam_user::remote::RemoteSteamUser;
///
/// # #[tokio::main]
/// # async fn main() -> Result<(), steam_user::remote::RemoteSteamUserError> {
/// // Single server
/// let user = RemoteSteamUser::new(
///     "https://api.example.com",
///     &[
///         "steamLoginSecure=76561198012345678||TOKEN",
///         "sessionid=SESSION_ID",
///     ],
/// );
///
/// // Multiple servers with round-robin
/// let user = RemoteSteamUser::with_urls(
///     &["https://api-1.example.com", "https://api-2.example.com"],
///     &[
///         "steamLoginSecure=76561198012345678||TOKEN",
///         "sessionid=SESSION_ID",
///     ],
/// );
///
/// let balance = user.get_steam_wallet_balance().await?;
/// println!("{:?}", balance);
/// # Ok(())
/// # }
/// ```
pub struct RemoteSteamUser {
    /// Server URLs for round-robin rotation.
    urls: Vec<String>,
    /// Atomic index for round-robin (wraps around).
    index: AtomicUsize,
    /// Maximum retry attempts per request.
    max_retries: usize,
    /// Reusable HTTP client.
    http: Client,
    /// Auth payload included in every request.
    auth: AuthPayload,
    /// Optional API key sent as `Authorization: Bearer <key>`.
    api_key: Option<String>,
}

impl RemoteSteamUser {
    /// Creates a `RemoteSteamUser` pointing to a single API server.
    ///
    /// # Arguments
    ///
    /// * `base_url` - The API server URL (e.g. `"https://api.example.com"`)
    /// * `cookies` - Steam cookie strings (e.g. `["steamLoginSecure=...",
    ///   "sessionid=..."]`)
    pub fn new(base_url: &str, cookies: &[&str]) -> Self {
        Self::with_urls(&[base_url], cookies)
    }

    /// Creates a `RemoteSteamUser` with multiple API servers for round-robin
    /// load balancing.
    ///
    /// Requests will rotate across the provided URLs. On failure, the next URL
    /// is tried automatically up to [`max_retries`](Self::set_max_retries)
    /// times (default: 5).
    ///
    /// # Arguments
    ///
    /// * `urls` - Slice of API server URLs
    /// * `cookies` - Steam cookie strings
    pub fn with_urls(urls: &[&str], cookies: &[&str]) -> Self {
        let http = build_http_client();

        Self {
            urls: urls.iter().map(|u| u.trim_end_matches('/').to_string()).collect(),
            // Seed from the process-global counter so per-request clients fan
            // out across nodes instead of all starting at urls[0].
            index: AtomicUsize::new(ROUND_ROBIN_SEED.fetch_add(1, Ordering::Relaxed)),
            max_retries: DEFAULT_MAX_RETRIES,
            http,
            auth: AuthPayload {
                cookies: cookies.iter().map(|c| c.to_string()).collect(),
                access_token: None,
                refresh_token: None,
                mobile_access_token: None,
                identity_secret: None,
                shared_secret: None,
            },
            api_key: None,
        }
    }

    /// Sets the maximum number of retry attempts per request (default: 5).
    pub fn set_max_retries(&mut self, max: usize) {
        self.max_retries = max;
    }

    /// Sets the OAuth access token.
    pub fn set_access_token(&mut self, token: String) {
        self.auth.access_token = Some(token);
    }

    /// Sets the OAuth refresh token.
    pub fn set_refresh_token(&mut self, token: String) {
        self.auth.refresh_token = Some(token);
    }

    /// Sets the mobile access token for 2FA operations.
    pub fn set_mobile_access_token(&mut self, token: String) {
        self.auth.mobile_access_token = Some(token);
    }

    /// Sets the identity secret for trade confirmations.
    pub fn set_identity_secret(&mut self, secret: String) {
        self.auth.identity_secret = Some(secret);
    }

    /// Sets the shared secret for 2FA TOTP generation.
    pub fn set_shared_secret(&mut self, secret: String) {
        self.auth.shared_secret = Some(secret);
    }

    /// Sets the API key for server authentication.
    ///
    /// When set, every request includes an `Authorization: Bearer <key>` header.
    pub fn set_api_key(&mut self, key: String) {
        self.api_key = Some(key);
    }

    // ========================================================================
    // Internal helpers
    // ========================================================================

    /// Returns the next server URL using atomic round-robin, skipping
    /// quarantined nodes.
    ///
    /// The round-robin counter advances over the *eligible* (non-quarantined)
    /// subset reported by [`health::eligible_indices`]. When every node is
    /// quarantined the breaker fails open and the full URL list is eligible, so
    /// selection never hard-fails solely because the breaker closed every door.
    fn next_url(&self) -> &str {
        if self.urls.is_empty() {
            return "";
        }
        let eligible = health::eligible_indices(&self.urls);
        // `eligible` is non-empty whenever `urls` is non-empty (fail-open
        // guarantees at least every index), but guard defensively.
        if eligible.is_empty() {
            let idx = self.index.fetch_add(1, Ordering::Relaxed) % self.urls.len();
            return &self.urls[idx];
        }
        let pos = self.index.fetch_add(1, Ordering::Relaxed) % eligible.len();
        &self.urls[eligible[pos]]
    }

    /// Records a node outcome against the process-wide circuit breaker and
    /// emits the corresponding `tracing` event for any state transition.
    fn record_node(base: &str, outcome: NodeOutcome) {
        match health::record(base, outcome) {
            HealthTransition::Quarantined { consecutive } => {
                tracing::warn!(
                    url = %base,
                    consecutive = consecutive,
                    quarantine_secs = health::QUARANTINE_DURATION.as_secs(),
                    "proxy node quarantined"
                );
            }
            HealthTransition::Recovered => {
                tracing::info!(url = %base, "proxy node recovered");
            }
            HealthTransition::None => {}
        }
    }

    /// Internal method that sends a POST request with auth + extra fields,
    /// using round-robin server selection with retry-on-failure.
    ///
    /// Returns the `data` field from the API response on success.
    pub(crate) async fn call(&self, path: &str, extra: serde_json::Value) -> Result<serde_json::Value, RemoteSteamUserError> {
        if self.urls.is_empty() {
            return Err(RemoteSteamUserError::NoUrls);
        }

        // Build the request body: merge auth + extra fields
        let mut body = match extra {
            serde_json::Value::Object(map) => map,
            _ => serde_json::Map::new(),
        };
        body.insert("auth".to_string(), serde_json::to_value(&self.auth).map_err(RemoteSteamUserError::Json)?);

        let body = serde_json::Value::Object(body);

        let mut last_error: Option<RemoteSteamUserError> = None;
        let backoff_base = Duration::from_millis(250);
        let backoff_cap = Duration::from_secs(8);

        for attempt in 0..self.max_retries {
            // Exponential backoff: no sleep before the first attempt.
            if attempt > 0 {
                let delay = std::cmp::min(
                    backoff_base * 2u32.saturating_pow(attempt as u32 - 1),
                    backoff_cap,
                );
                tokio::time::sleep(delay).await;
            }

            let base = self.next_url().to_string();
            let url = format!("{}{}", base, path);

            tracing::debug!("POST {} (attempt {})", url, attempt + 1);

            let mut req = self.http.post(&url).json(&body);
            if let Some(ref key) = self.api_key {
                req = req.bearer_auth(key);
            }

            let resp = match req.send().await {
                Ok(r) => r,
                Err(e) => {
                    // A transport-level error (connect/timeout) is a node fault.
                    tracing::warn!("Request to {} failed: {}", url, e);
                    Self::record_node(&base, NodeOutcome::RetryableFault);
                    last_error = Some(RemoteSteamUserError::Http(e));
                    continue;
                }
            };

            let status = resp.status();
            let text = match resp.text().await {
                Ok(t) => t,
                Err(e) => {
                    // Failure reading the body mid-stream — treat as a node fault.
                    tracing::warn!("Failed to read response body from {}: {}", url, e);
                    Self::record_node(&base, NodeOutcome::RetryableFault);
                    last_error = Some(RemoteSteamUserError::Http(e));
                    continue;
                }
            };

            if text.is_empty() {
                tracing::warn!("Empty response from {}", url);
                Self::record_node(&base, NodeOutcome::RetryableFault);
                last_error = Some(RemoteSteamUserError::Api { status: status.as_u16(), message: "Empty response".into(), url: url.clone() });
                continue;
            }

            let api_resp: ApiResponse = match serde_json::from_str(&text) {
                Ok(r) => r,
                Err(e) => {
                    // Garbage body — the node misbehaved.
                    tracing::warn!("Failed to parse response from {}: {} (status {})", url, e, status);
                    Self::record_node(&base, NodeOutcome::RetryableFault);
                    last_error = Some(RemoteSteamUserError::Json(e));
                    continue;
                }
            };

            if !api_resp.success {
                let msg = api_resp.error.clone().unwrap_or_else(|| format!("Unknown error (HTTP {})", status));

                // Decide whether to retry or terminate based on the HTTP status
                // and error message content. 5xx-like and transient errors are
                // retried; 4xx-like client errors are terminal.
                let should_retry = if status.is_server_error() || status == reqwest::StatusCode::TOO_MANY_REQUESTS {
                    true
                } else if status.is_client_error() {
                    false
                } else {
                    // No clear HTTP signal — inspect the error string for known
                    // transient keywords.
                    let lower = msg.to_lowercase();
                    lower.contains("timeout")
                        || lower.contains("unavailable")
                        || lower.contains("503")
                        || lower.contains("502")
                        || lower.contains("429")
                };

                // Distinguish a *node* fault from a *Steam-tunneled* fault. When
                // the proxy reached Steam and relayed Steam's own error (e.g.
                // "Steam error: HTTP 429"), the node is healthy — the fault is
                // Steam's. Only genuine node-level faults feed the breaker.
                let outcome = if is_steam_tunneled_error(&msg) {
                    NodeOutcome::Success
                } else {
                    NodeOutcome::RetryableFault
                };
                Self::record_node(&base, outcome);

                if should_retry {
                    tracing::warn!("Retryable failure from {} (HTTP {}): {}", url, status, msg);
                    last_error = Some(RemoteSteamUserError::Api { status: status.as_u16(), message: msg, url: url.clone() });
                    continue;
                } else {
                    tracing::warn!("Terminal failure from {} (HTTP {}): {}", url, status, msg);
                    return Err(RemoteSteamUserError::Api { status: status.as_u16(), message: msg, url: url.clone() });
                }
            }

            // Genuine success — the node served the request.
            Self::record_node(&base, NodeOutcome::Success);
            return Ok(api_resp.data.unwrap_or(serde_json::Value::Null));
        }

        match last_error {
            Some(e) => Err(e),
            None => Err(RemoteSteamUserError::AllRetriesFailed),
        }
    }
    /// Typed variant of [`call`] that deserializes the JSON `data` field into
    /// `T`.
    pub(crate) async fn call_typed<T: serde::de::DeserializeOwned>(&self, path: &str, extra: serde_json::Value) -> Result<T, RemoteSteamUserError> {
        let value = self.call(path, extra).await?;
        serde_json::from_value(value).map_err(RemoteSteamUserError::Json)
    }

    /// Variant of [`call`] for void methods — discards the response data.
    pub(crate) async fn call_void(&self, path: &str, extra: serde_json::Value) -> Result<(), RemoteSteamUserError> {
        self.call(path, extra).await?;
        Ok(())
    }
}

/// Returns `true` if a proxy error message indicates the fault originated at
/// *Steam* (relayed through the proxy), not at the proxy node itself.
///
/// The proxy may answer `HTTP 502` while its body carries an inner
/// `Steam error: HTTP 429` (or similar). That means the node successfully
/// reached Steam — the node is healthy and must not be quarantined. There is
/// no structured field for this, so we match on message content.
///
/// Contract: the marker is the `steam-user-api` proxy's `ApiError::Steam`
/// Display prefix — `#[error("Steam error: {0:#}")]` (steam-user-api repo,
/// `src/auth.rs`). If the server ever rewords that prefix, this breaker
/// silently stops recognizing tunneled faults and would start quarantining
/// healthy nodes for Steam-side errors — keep the two in sync.
fn is_steam_tunneled_error(message: &str) -> bool {
    message.to_lowercase().contains("steam error")
}

impl std::fmt::Debug for RemoteSteamUser {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RemoteSteamUser").field("urls", &self.urls).field("max_retries", &self.max_retries).field("cookies_count", &self.auth.cookies.len()).field("has_api_key", &self.api_key.is_some()).finish()
    }
}

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

    #[test]
    fn detects_steam_tunneled_errors() {
        // Fixtures in the exact production format: the steam-user-api proxy's
        // `ApiError::Steam` Display prefix ("Steam error: {0:#}") wrapping the
        // inner steam-user error. If the proxy rewords the prefix these break.
        assert!(is_steam_tunneled_error("Steam error: HTTP 429 from http://no.url.provided.local/"));
        assert!(is_steam_tunneled_error("Steam error: Not logged in"));
        // Genuine node faults → not tunneled.
        assert!(!is_steam_tunneled_error("HTTP 502 Bad Gateway"));
        assert!(!is_steam_tunneled_error("connection refused"));
        assert!(!is_steam_tunneled_error("service unavailable"));
    }
}