spacetraders-client 0.1.0

use std::future::Future;
use std::sync::atomic::Ordering;
use std::time::Instant;

use tokio::time::{Duration, sleep};
use tracing::{info, warn};

use super::utils::{
    LOW_PRIORITY_DEFERRED_TOTAL, MAX_RETRIES, REQUESTS_TOTAL, backoff_delay, map_progenitor_err,
};
use super::{ClientError, PError, RequestPriority, RetryPolicy, StClient};

impl StClient {
    /// Issue an **idempotent read** (GET) and return the unwrapped response body.
    ///
    /// Transient failures (429 / 5xx / communication errors) are retried
    /// transparently with backoff up to [`MAX_RETRIES`] times: a 429's
    /// `Retry-After` hint is honored when present, otherwise exponential backoff
    /// with jitter is used. The caller only ever sees a success or a final typed
    /// [`ClientError`]. Use [`StClient::send_mutating`] for non-idempotent writes.
    pub(crate) async fn send<T, E, F, Fut>(
        &self,
        endpoint: &'static str,
        priority: RequestPriority,
        op: F,
    ) -> Result<T, ClientError>
    where
        E: std::fmt::Debug,
        F: Fn() -> Fut,
        Fut: Future<Output = Result<progenitor_client::ResponseValue<T>, PError<E>>>,
    {
        self.send_with_policy(endpoint, priority, RetryPolicy::Idempotent, op)
            .await
    }

    /// Issue a **non-idempotent mutation** (POST/PATCH) and return the unwrapped
    /// response body.
    ///
    /// Identical to [`StClient::send`] except for the retry policy: a 5xx or
    /// communication failure is *not* replayed, because the server may already
    /// have applied the change and replaying the POST would double-apply it
    /// (e.g. selling cargo twice). Only a 429 — which the rate limiter rejects
    /// before the action runs — is retried. Any other failure is surfaced to the
    /// caller so the executor/actor can log it and back off instead of repeating
    /// the request.
    pub(crate) async fn send_mutating<T, E, F, Fut>(
        &self,
        endpoint: &'static str,
        priority: RequestPriority,
        op: F,
    ) -> Result<T, ClientError>
    where
        E: std::fmt::Debug,
        F: Fn() -> Fut,
        Fut: Future<Output = Result<progenitor_client::ResponseValue<T>, PError<E>>>,
    {
        self.send_with_policy(endpoint, priority, RetryPolicy::Mutating, op)
            .await
    }

    /// Shared body of [`StClient::send`] / [`StClient::send_mutating`]: map the
    /// progenitor result onto a typed [`ClientError`], then run the retry loop
    /// under the given `policy`.
    async fn send_with_policy<T, E, F, Fut>(
        &self,
        endpoint: &'static str,
        priority: RequestPriority,
        policy: RetryPolicy,
        op: F,
    ) -> Result<T, ClientError>
    where
        E: std::fmt::Debug,
        F: Fn() -> Fut,
        Fut: Future<Output = Result<progenitor_client::ResponseValue<T>, PError<E>>>,
    {
        self.send_retrying(endpoint, priority, policy, || async {
            match op().await {
                Ok(rv) => Ok(rv.into_inner()),
                Err(e) => Err(map_progenitor_err(e).await),
            }
        })
        .await
    }

    /// Retry/backoff loop over an already error-mapped operation. Kept separate
    /// from [`StClient::send`] so the retry policy can be tested without faking
    /// an HTTP response. `policy` decides which transient failures are eligible
    /// for an automatic retry (see [`RetryPolicy`]).
    pub(crate) async fn send_retrying<T, F, Fut>(
        &self,
        endpoint: &'static str,
        priority: RequestPriority,
        policy: RetryPolicy,
        op: F,
    ) -> Result<T, ClientError>
    where
        F: Fn() -> Fut,
        Fut: Future<Output = Result<T, ClientError>>,
    {
        let mut attempt: u32 = 0;
        let mut queued = Duration::ZERO;
        let started = Instant::now();
        loop {
            let permit = self.scheduler.acquire(priority).await;
            queued += permit.queued;
            let scheduler_queue_depth = permit.queue_depth;
            let rate_budget_remaining = permit.rate_budget_remaining;
            match op().await {
                Ok(value) => {
                    REQUESTS_TOTAL.fetch_add(1, Ordering::Relaxed);
                    if priority == RequestPriority::Low && !queued.is_zero() {
                        LOW_PRIORITY_DEFERRED_TOTAL.fetch_add(1, Ordering::Relaxed);
                    }
                    info!(
                        endpoint,
                        ?priority,
                        queued_ms = queued.as_millis() as u64,
                        attempts = attempt + 1,
                        status = "ok",
                        retry_after_ms = 0u64,
                        scheduler_queue_depth,
                        rate_budget_remaining,
                        duration_ms = started.elapsed().as_millis() as u64,
                        "st_request"
                    );
                    return Ok(value);
                }
                Err(err) if err.is_retryable_under(policy) && attempt < MAX_RETRIES => {
                    let delay = err.retry_after().unwrap_or_else(|| backoff_delay(attempt));
                    attempt += 1;
                    warn!(
                        ?priority,
                        attempt,
                        max_retries = MAX_RETRIES,
                        delay_ms = delay.as_millis() as u64,
                        error = %err,
                        "request failed with transient error; retrying after backoff"
                    );
                    sleep(delay).await;
                }
                Err(err) => {
                    REQUESTS_TOTAL.fetch_add(1, Ordering::Relaxed);
                    if priority == RequestPriority::Low && !queued.is_zero() {
                        LOW_PRIORITY_DEFERRED_TOTAL.fetch_add(1, Ordering::Relaxed);
                    }
                    let retry_after_ms = err
                        .retry_after()
                        .map(|delay| delay.as_millis() as u64)
                        .unwrap_or(0);
                    info!(
                        endpoint,
                        ?priority,
                        queued_ms = queued.as_millis() as u64,
                        attempts = attempt + 1,
                        status = err.class(),
                        retry_after_ms,
                        scheduler_queue_depth,
                        rate_budget_remaining,
                        duration_ms = started.elapsed().as_millis() as u64,
                        error = %err,
                        "st_request"
                    );
                    return Err(err);
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::sync::atomic::AtomicU32;

    use super::super::{RequestScheduler, StClient};
    use super::*;

    fn test_client() -> StClient {
        StClient::new(
            "http://localhost",
            "test-token".to_string(),
            RequestScheduler::with_burst(1000.0, 1000.0),
        )
    }

    #[tokio::test(start_paused = true)]
    async fn retries_transient_then_succeeds() {
        let client = test_client();
        let calls = Arc::new(AtomicU32::new(0));
        let counter = calls.clone();

        let result: Result<u32, ClientError> = client
            .send_retrying(
                "test_endpoint",
                RequestPriority::Normal,
                RetryPolicy::Idempotent,
                || {
                    let counter = counter.clone();
                    async move {
                        let n = counter.fetch_add(1, Ordering::SeqCst);
                        if n < 2 {
                            Err(ClientError::RateLimited {
                                retry_after: Some(Duration::from_millis(10)),
                            })
                        } else {
                            Ok(42)
                        }
                    }
                },
            )
            .await;

        assert_eq!(result.unwrap(), 42);
        assert_eq!(calls.load(Ordering::SeqCst), 3);
    }

    #[tokio::test(start_paused = true)]
    async fn gives_up_after_retry_budget() {
        let client = test_client();
        let calls = Arc::new(AtomicU32::new(0));
        let counter = calls.clone();

        let result: Result<u32, ClientError> = client
            .send_retrying(
                "test_endpoint",
                RequestPriority::Normal,
                RetryPolicy::Idempotent,
                || {
                    let counter = counter.clone();
                    async move {
                        counter.fetch_add(1, Ordering::SeqCst);
                        Err(ClientError::RetryableServer {
                            status: 503,
                            message: "down".into(),
                        })
                    }
                },
            )
            .await;

        assert!(matches!(result, Err(ClientError::RetryableServer { .. })));
        assert_eq!(calls.load(Ordering::SeqCst), MAX_RETRIES + 1);
    }

    #[tokio::test(start_paused = true)]
    async fn mutating_policy_does_not_retry_server_error() {
        let client = test_client();
        let calls = Arc::new(AtomicU32::new(0));
        let counter = calls.clone();

        let result: Result<u32, ClientError> = client
            .send_retrying(
                "sell_cargo",
                RequestPriority::Critical,
                RetryPolicy::Mutating,
                || {
                    let counter = counter.clone();
                    async move {
                        counter.fetch_add(1, Ordering::SeqCst);
                        Err(ClientError::RetryableServer {
                            status: 500,
                            message: "internal".into(),
                        })
                    }
                },
            )
            .await;

        assert!(matches!(result, Err(ClientError::RetryableServer { .. })));
        assert_eq!(calls.load(Ordering::SeqCst), 1);
    }

    #[tokio::test(start_paused = true)]
    async fn mutating_policy_still_retries_rate_limit() {
        let client = test_client();
        let calls = Arc::new(AtomicU32::new(0));
        let counter = calls.clone();

        let result: Result<u32, ClientError> = client
            .send_retrying(
                "sell_cargo",
                RequestPriority::Critical,
                RetryPolicy::Mutating,
                || {
                    let counter = counter.clone();
                    async move {
                        let n = counter.fetch_add(1, Ordering::SeqCst);
                        if n < 1 {
                            Err(ClientError::RateLimited {
                                retry_after: Some(Duration::from_millis(10)),
                            })
                        } else {
                            Ok(7)
                        }
                    }
                },
            )
            .await;

        assert_eq!(result.unwrap(), 7);
        assert_eq!(calls.load(Ordering::SeqCst), 2);
    }

    #[tokio::test(start_paused = true)]
    async fn non_retryable_error_is_not_retried() {
        let client = test_client();
        let calls = Arc::new(AtomicU32::new(0));
        let counter = calls.clone();

        let result: Result<u32, ClientError> = client
            .send_retrying(
                "test_endpoint",
                RequestPriority::Normal,
                RetryPolicy::Idempotent,
                || {
                    let counter = counter.clone();
                    async move {
                        counter.fetch_add(1, Ordering::SeqCst);
                        Err(ClientError::Api("bad request".into()))
                    }
                },
            )
            .await;

        assert!(matches!(result, Err(ClientError::Api(_))));
        assert_eq!(calls.load(Ordering::SeqCst), 1);
    }
}