spiral-sdk 0.1.10

//! Shared authenticated HTTP transport for control-plane tools.

use std::collections::HashSet;
use std::fmt::{Debug, Formatter, Result as FmtResult};
use std::time::Duration;

use reqwest::{
    Client as ReqwestClient, ClientBuilder as ReqwestClientBuilder, Method, Response, Url,
};
use reqwest_middleware::{ClientBuilder, ClientWithMiddleware, RequestBuilder};
use serde::Serialize;
use serde::de::DeserializeOwned;
use serde_json::from_slice;
use spiral_auth::install_default_crypto_provider;
use spiral_contract::auth::AuthHeader;

use crate::auth::TokenSource;
use crate::{Error, List};

/// The `User-Agent` of every SDK request.
const USER_AGENT: &str = concat!("spiral-sdk/", env!("CARGO_PKG_VERSION"));

/// An HTTP client builder that identifies the SDK and leaves responses compressed.
///
/// Response bytes and content headers remain intact even if another dependency enables a
/// reqwest compression feature. API clients opt into gzip with `.gzip(true)`.
#[expect(
    clippy::disallowed_methods,
    reason = "The SDK defines its own HTTP client policy"
)]
pub(crate) fn http_client_builder() -> ReqwestClientBuilder {
    install_default_crypto_provider();
    ReqwestClient::builder()
        .user_agent(USER_AGENT)
        .no_gzip()
        .no_brotli()
        .no_deflate()
        .no_zstd()
}

/// Wrap `client` for generated service clients.
pub(crate) fn service_client(client: ReqwestClient) -> ClientWithMiddleware {
    ClientBuilder::new(client).build()
}

/// Apply the current bearer token, if any, to an outgoing request.
pub(crate) async fn apply_auth_token(
    provider: Option<&TokenSource>,
    request: RequestBuilder,
) -> Result<RequestBuilder, Error> {
    match provider {
        Some(provider) => match provider.token().await? {
            Some(token) => Ok(request.bearer_auth(token.expose_secret())),
            None => Ok(request),
        },
        None => Ok(request),
    }
}

/// Return `response` if it succeeded, else the status and body it was rejected with.
pub(crate) async fn accepted(response: Response) -> Result<Response, Error> {
    let status = response.status();
    if status.is_success() {
        return Ok(response);
    }
    let url = response.url().clone();
    let body = response.text().await?;
    Err(Error::Rejected { url, status, body })
}

/// An empty request body or query value with an inferred unit type.
pub const fn empty_value() -> Option<()> {
    None
}

/// Shared connection pool and credentials for control-plane HTTP requests.
#[derive(Clone)]
pub struct HttpClient {
    base_url: Url,
    http: ClientWithMiddleware,
    auth: Option<TokenSource>,
}

impl Debug for HttpClient {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> FmtResult {
        formatter
            .debug_struct("HttpClient")
            .field("base_url", &self.base_url)
            .finish_non_exhaustive()
    }
}

impl HttpClient {
    pub(crate) fn middleware_client(&self) -> ClientWithMiddleware {
        self.http.clone()
    }

    /// Resolve a fresh credential for one generated service call.
    pub(crate) async fn auth_header(&self) -> Result<AuthHeader, Error> {
        resolve_auth_header(self.auth.as_ref()).await
    }

    pub(crate) fn with_client(mut self, client: ReqwestClient) -> Self {
        self.http = service_client(client);
        self
    }

    pub(crate) fn with_token_source(mut self, provider: TokenSource) -> Self {
        self.auth = Some(provider);
        self
    }
    pub(crate) fn new(base_url: Url, http: ReqwestClient, auth: Option<TokenSource>) -> Self {
        Self {
            base_url,
            http: service_client(http),
            auth,
        }
    }

    pub(crate) fn connect(base: &str) -> Result<Self, Error> {
        let base = Url::parse(base).map_err(Error::invalid)?;
        let http = http_client_builder()
            .gzip(true)
            .redirect(reqwest::redirect::Policy::none())
            .connect_timeout(Duration::from_secs(10))
            .timeout(Duration::from_secs(60))
            .build()
            .map_err(Error::invalid)?;
        Ok(Self::new(base, http, None))
    }

    /// The configured API endpoint, including any deployment prefix.
    pub fn base_url(&self) -> &Url {
        &self.base_url
    }

    /// Read a JSON resource with optional query parameters.
    pub async fn get<Q: Serialize, R: DeserializeOwned>(
        &self,
        path: &str,
        query: Option<Q>,
    ) -> Result<R, Error> {
        self.request(Method::GET, path, None::<()>, query).await
    }

    /// Create or invoke a JSON resource.
    pub async fn post<B: Serialize, Q: Serialize, R: DeserializeOwned>(
        &self,
        path: &str,
        body: Option<B>,
        query: Option<Q>,
    ) -> Result<R, Error> {
        self.request(Method::POST, path, body, query).await
    }

    /// Replace a JSON resource.
    pub async fn put<B: Serialize, Q: Serialize, R: DeserializeOwned>(
        &self,
        path: &str,
        body: Option<B>,
        query: Option<Q>,
    ) -> Result<R, Error> {
        self.request(Method::PUT, path, body, query).await
    }

    /// Delete a resource, decoding its JSON response.
    pub async fn delete<Q: Serialize, R: DeserializeOwned>(
        &self,
        path: &str,
        query: Option<Q>,
    ) -> Result<R, Error> {
        self.request(Method::DELETE, path, None::<()>, query).await
    }

    /// Read a JSON resource without retaining more than `max_bytes` of metadata.
    pub async fn get_bounded<R: DeserializeOwned>(
        &self,
        path: &str,
        max_bytes: usize,
    ) -> Result<R, Error> {
        self.request_bounded(Method::GET, path, None::<()>, None::<()>, max_bytes)
            .await
    }

    /// Invoke a bodyless resource binding with a bounded JSON response.
    pub async fn put_bounded<R: DeserializeOwned>(
        &self,
        path: &str,
        max_bytes: usize,
    ) -> Result<R, Error> {
        self.request_bounded(Method::PUT, path, None::<()>, None::<()>, max_bytes)
            .await
    }

    async fn request<B: Serialize, Q: Serialize, R: DeserializeOwned>(
        &self,
        method: Method,
        path: &str,
        body: Option<B>,
        query: Option<Q>,
    ) -> Result<R, Error> {
        self.request_bounded(method, path, body, query, usize::MAX)
            .await
    }

    async fn request_bounded<B: Serialize, Q: Serialize, R: DeserializeOwned>(
        &self,
        method: Method,
        path: &str,
        body: Option<B>,
        query: Option<Q>,
        max_bytes: usize,
    ) -> Result<R, Error> {
        let url = Url::parse(&format!(
            "{}{}",
            self.base_url,
            path.trim_start_matches('/')
        ))
        .map_err(Error::invalid)?;
        let mut request = self
            .http
            .request(method, url)
            .timeout(Duration::from_secs(60));
        if let Some(body) = body {
            request = request.json(&body);
        }
        if let Some(query) = query {
            request = request.query(&query);
        }
        let response = apply_auth_token(self.auth.as_ref(), request)
            .await?
            .send()
            .await?;
        if max_bytes == usize::MAX {
            return Ok(accepted(response).await?.json().await?);
        }
        let mut response = response;
        let status = response.status();
        let url = response.url().clone();
        let mut bytes = Vec::new();
        while let Some(chunk) = response.chunk().await? {
            if bytes.len().saturating_add(chunk.len()) > max_bytes {
                return Err(Error::Protocol(
                    "resource metadata exceeds its byte budget".into(),
                ));
            }
            bytes.extend_from_slice(&chunk);
        }
        if !status.is_success() {
            return Err(Error::Rejected {
                url,
                status,
                body: String::from_utf8_lossy(&bytes).into_owned(),
            });
        }
        from_slice(&bytes).map_err(|error| Error::Protocol(error.to_string()))
    }

    /// Build a paginated GET request, retaining filters and page size on every request.
    pub fn list<T: DeserializeOwned + Send + 'static>(
        &self,
        path: String,
        query: Option<Vec<(String, String)>>,
    ) -> List<'_, T> {
        List::new(move |options| {
            let path = path.clone();
            let mut query = query.clone().unwrap_or_default();
            query.retain(|(key, _)| key != "page_token" && key != "page_size");
            if let Some(token) = options.token {
                query.push(("page_token".to_owned(), token));
            }
            if let Some(size) = options.size {
                query.push(("page_size".to_owned(), size.to_string()));
            }
            Box::pin(async move { self.get(&path, Some(query)).await })
        })
    }
}

pub(crate) async fn resolve_auth_header(
    provider: Option<&TokenSource>,
) -> Result<AuthHeader, Error> {
    let provider = provider.ok_or(Error::CredentialsNotFound)?;
    let token = provider.token().await?.ok_or(Error::CredentialsNotFound)?;
    Ok(AuthHeader::new(format!("Bearer {}", token.expose_secret())))
}

pub(crate) fn continuation(
    token: Option<String>,
    seen: &mut HashSet<String>,
) -> Result<Option<String>, Error> {
    if let Some(token) = &token
        && (token.is_empty() || !seen.insert(token.clone()))
    {
        return Err(Error::Protocol("invalid pagination token".to_owned()));
    }
    Ok(token)
}