spiral/
http.rs

1//! Shared authenticated HTTP transport for control-plane tools.
2
3use std::collections::HashSet;
4use std::fmt::{Debug, Formatter, Result as FmtResult};
5use std::time::Duration;
6
7use reqwest::{ClientBuilder as ReqwestClientBuilder, Method, Response, Url};
8use reqwest_middleware::{ClientWithMiddleware, RequestBuilder};
9use serde::Serialize;
10use serde::de::DeserializeOwned;
11use serde_json::from_slice;
12use spiral_auth::install_default_crypto_provider;
13use spiral_contract::auth::AuthHeader;
14
15use crate::auth::TokenSource;
16use crate::{Error, List};
17
18/// The `User-Agent` of every SDK request.
19const USER_AGENT: &str = concat!("spiral-sdk/", env!("CARGO_PKG_VERSION"));
20
21/// An HTTP client builder that identifies the SDK and leaves responses compressed.
22///
23/// Response bytes and content headers remain intact even if another dependency enables a
24/// reqwest compression feature. API clients opt into gzip with `.gzip(true)`.
25#[expect(
26    clippy::disallowed_methods,
27    reason = "The public SDK owns its default transport; hosts can supply their own client"
28)]
29fn http_client_builder() -> ReqwestClientBuilder {
30    install_default_crypto_provider();
31    reqwest::Client::builder()
32        .redirect(reqwest::redirect::Policy::none())
33        .connect_timeout(Duration::from_secs(10))
34        .timeout(Duration::from_secs(60))
35        .user_agent(USER_AGENT)
36        .no_gzip()
37        .no_brotli()
38        .no_deflate()
39        .no_zstd()
40}
41
42/// Create the SDK's redirect-disabled default HTTP client without host middleware.
43#[expect(
44    clippy::disallowed_methods,
45    reason = "The SDK owns its default transport"
46)]
47pub fn default_http_client() -> Result<ClientWithMiddleware, reqwest::Error> {
48    http_client_builder()
49        .build()
50        .map(ClientWithMiddleware::from)
51}
52
53#[expect(
54    clippy::disallowed_methods,
55    reason = "The SDK owns its default API transport"
56)]
57pub(crate) fn api_http_client() -> Result<ClientWithMiddleware, reqwest::Error> {
58    http_client_builder()
59        .gzip(true)
60        .build()
61        .map(ClientWithMiddleware::from)
62}
63
64/// Wrap `client` for generated service clients.
65pub(crate) fn service_client(client: impl Into<ClientWithMiddleware>) -> ClientWithMiddleware {
66    client.into()
67}
68
69/// Apply the current bearer token, if any, to an outgoing request.
70pub(crate) async fn apply_auth_token(
71    provider: Option<&TokenSource>,
72    request: RequestBuilder,
73) -> Result<RequestBuilder, Error> {
74    match provider {
75        Some(provider) => match provider.token().await? {
76            Some(token) => Ok(request.bearer_auth(token.expose_secret())),
77            None => Ok(request),
78        },
79        None => Ok(request),
80    }
81}
82
83/// Return `response` if it succeeded, else the status and body it was rejected with.
84pub(crate) async fn accepted(response: Response) -> Result<Response, Error> {
85    let status = response.status();
86    if status.is_success() {
87        return Ok(response);
88    }
89    let url = response.url().clone();
90    let body = response.text().await?;
91    Err(Error::Rejected { url, status, body })
92}
93
94/// An empty request body or query value with an inferred unit type.
95pub const fn empty_value() -> Option<()> {
96    None
97}
98
99/// Shared connection pool and credentials for control-plane HTTP requests.
100#[derive(Clone)]
101pub struct HttpClient {
102    base_url: Url,
103    http: ClientWithMiddleware,
104    auth: Option<TokenSource>,
105}
106
107impl Debug for HttpClient {
108    fn fmt(&self, formatter: &mut Formatter<'_>) -> FmtResult {
109        formatter
110            .debug_struct("HttpClient")
111            .field("base_url", &self.base_url)
112            .finish_non_exhaustive()
113    }
114}
115
116impl HttpClient {
117    pub(crate) fn middleware_client(&self) -> ClientWithMiddleware {
118        self.http.clone()
119    }
120
121    /// Resolve a fresh credential for one generated service call.
122    pub(crate) async fn auth_header(&self) -> Result<AuthHeader, Error> {
123        resolve_auth_header(self.auth.as_ref()).await
124    }
125
126    pub(crate) fn with_client(mut self, client: impl Into<ClientWithMiddleware>) -> Self {
127        self.http = service_client(client);
128        self
129    }
130
131    pub(crate) fn with_token_source(mut self, provider: TokenSource) -> Self {
132        self.auth = Some(provider);
133        self
134    }
135    pub(crate) fn new(
136        base_url: Url,
137        http: impl Into<ClientWithMiddleware>,
138        auth: Option<TokenSource>,
139    ) -> Self {
140        Self {
141            base_url,
142            http: service_client(http),
143            auth,
144        }
145    }
146
147    pub(crate) fn connect(base: &str) -> Result<Self, Error> {
148        let base = Url::parse(base).map_err(Error::invalid)?;
149        let http = api_http_client().map_err(Error::invalid)?;
150        Ok(Self::new(base, http, None))
151    }
152
153    /// The configured API endpoint, including any deployment prefix.
154    pub fn base_url(&self) -> &Url {
155        &self.base_url
156    }
157
158    /// Read a JSON resource with optional query parameters.
159    pub async fn get<Q: Serialize, R: DeserializeOwned>(
160        &self,
161        path: &str,
162        query: Option<Q>,
163    ) -> Result<R, Error> {
164        self.request(Method::GET, path, None::<()>, query).await
165    }
166
167    /// Create or invoke a JSON resource.
168    pub async fn post<B: Serialize, Q: Serialize, R: DeserializeOwned>(
169        &self,
170        path: &str,
171        body: Option<B>,
172        query: Option<Q>,
173    ) -> Result<R, Error> {
174        self.request(Method::POST, path, body, query).await
175    }
176
177    /// Replace a JSON resource.
178    pub async fn put<B: Serialize, Q: Serialize, R: DeserializeOwned>(
179        &self,
180        path: &str,
181        body: Option<B>,
182        query: Option<Q>,
183    ) -> Result<R, Error> {
184        self.request(Method::PUT, path, body, query).await
185    }
186
187    /// Delete a resource, decoding its JSON response.
188    pub async fn delete<Q: Serialize, R: DeserializeOwned>(
189        &self,
190        path: &str,
191        query: Option<Q>,
192    ) -> Result<R, Error> {
193        self.request(Method::DELETE, path, None::<()>, query).await
194    }
195
196    /// Read a JSON resource without retaining more than `max_bytes` of metadata.
197    pub async fn get_bounded<R: DeserializeOwned>(
198        &self,
199        path: &str,
200        max_bytes: usize,
201    ) -> Result<R, Error> {
202        self.request_bounded(Method::GET, path, None::<()>, None::<()>, max_bytes)
203            .await
204    }
205
206    /// Invoke a bodyless resource binding with a bounded JSON response.
207    pub async fn put_bounded<R: DeserializeOwned>(
208        &self,
209        path: &str,
210        max_bytes: usize,
211    ) -> Result<R, Error> {
212        self.request_bounded(Method::PUT, path, None::<()>, None::<()>, max_bytes)
213            .await
214    }
215
216    async fn request<B: Serialize, Q: Serialize, R: DeserializeOwned>(
217        &self,
218        method: Method,
219        path: &str,
220        body: Option<B>,
221        query: Option<Q>,
222    ) -> Result<R, Error> {
223        self.request_bounded(method, path, body, query, usize::MAX)
224            .await
225    }
226
227    async fn request_bounded<B: Serialize, Q: Serialize, R: DeserializeOwned>(
228        &self,
229        method: Method,
230        path: &str,
231        body: Option<B>,
232        query: Option<Q>,
233        max_bytes: usize,
234    ) -> Result<R, Error> {
235        let url = Url::parse(&format!(
236            "{}{}",
237            self.base_url,
238            path.trim_start_matches('/')
239        ))
240        .map_err(Error::invalid)?;
241        let mut request = self
242            .http
243            .request(method, url)
244            .timeout(Duration::from_secs(60));
245        if let Some(body) = body {
246            request = request.json(&body);
247        }
248        if let Some(query) = query {
249            request = request.query(&query);
250        }
251        let response = apply_auth_token(self.auth.as_ref(), request)
252            .await?
253            .send()
254            .await?;
255        if max_bytes == usize::MAX {
256            return Ok(accepted(response).await?.json().await?);
257        }
258        let mut response = response;
259        let status = response.status();
260        let url = response.url().clone();
261        let mut bytes = Vec::new();
262        while let Some(chunk) = response.chunk().await? {
263            if bytes.len().saturating_add(chunk.len()) > max_bytes {
264                return Err(Error::Protocol(
265                    "resource metadata exceeds its byte budget".into(),
266                ));
267            }
268            bytes.extend_from_slice(&chunk);
269        }
270        if !status.is_success() {
271            return Err(Error::Rejected {
272                url,
273                status,
274                body: String::from_utf8_lossy(&bytes).into_owned(),
275            });
276        }
277        from_slice(&bytes).map_err(|error| Error::Protocol(error.to_string()))
278    }
279
280    /// Build a paginated GET request, retaining filters and page size on every request.
281    pub fn list<T: DeserializeOwned + Send + 'static>(
282        &self,
283        path: String,
284        query: Option<Vec<(String, String)>>,
285    ) -> List<'_, T> {
286        List::new(move |options| {
287            let path = path.clone();
288            let mut query = query.clone().unwrap_or_default();
289            query.retain(|(key, _)| key != "page_token" && key != "page_size");
290            if let Some(token) = options.token {
291                query.push(("page_token".to_owned(), token));
292            }
293            if let Some(size) = options.size {
294                query.push(("page_size".to_owned(), size.to_string()));
295            }
296            Box::pin(async move { self.get(&path, Some(query)).await })
297        })
298    }
299}
300
301pub(crate) async fn resolve_auth_header(
302    provider: Option<&TokenSource>,
303) -> Result<AuthHeader, Error> {
304    let provider = provider.ok_or(Error::CredentialsNotFound)?;
305    let token = provider.token().await?.ok_or(Error::CredentialsNotFound)?;
306    Ok(AuthHeader::new(format!("Bearer {}", token.expose_secret())))
307}
308
309pub(crate) fn continuation(
310    token: Option<String>,
311    seen: &mut HashSet<String>,
312) -> Result<Option<String>, Error> {
313    if let Some(token) = &token
314        && (token.is_empty() || !seen.insert(token.clone()))
315    {
316        return Err(Error::Protocol("invalid pagination token".to_owned()));
317    }
318    Ok(token)
319}
spiral/http.rs

spiral/
http.rs
