sova_http/
client.rs

1//! Shared [`HttpClient`] and request builder.
2
3use crate::breaker::{BreakerConfig, CircuitBreaker};
4use crate::error::HttpError;
5use crate::fake::FakeTransport;
6use crate::reqwest_transport::{default_timeout, ReqwestTransport, DEFAULT_MAX_RESPONSE};
7use crate::retry::{full_jitter, method_idempotent, parse_retry_after};
8use crate::ssrf::SsrfPolicy;
9use crate::transport::{OutRequest, OutResponse, Transport};
10use bytes::Bytes;
11use http::{HeaderMap, HeaderName, HeaderValue, Method};
12use sova_core::extend::BoxFuture;
13use sova_core::{App, Plugin, Request};
14use serde::Serialize;
15use std::collections::HashMap;
16use std::sync::Arc;
17use std::time::Duration;
18use tracing::Instrument;
19
20/// Per-named-client overrides from `sova.toml`.
21#[derive(Debug, Clone, Default)]
22pub struct NamedClientConfig {
23    pub base_url: Option<String>,
24    pub timeout: Option<Duration>,
25    pub retries: Option<u32>,
26    pub max_response_size: Option<usize>,
27    pub deny_private_networks: Option<bool>,
28    pub bearer_env: Option<String>,
29    pub allow_hosts: Vec<String>,
30}
31
32/// Shared outbound HTTP handle stored in app state.
33#[derive(Clone)]
34pub struct HttpClient {
35    transport: Arc<dyn Transport>,
36    default_timeout: Duration,
37    #[allow(dead_code)]
38    max_response_size: usize,
39    #[allow(dead_code)]
40    ssrf: SsrfPolicy,
41    breaker: Arc<CircuitBreaker>,
42    named: Arc<HashMap<String, NamedClientConfig>>,
43    fake: Option<FakeTransport>,
44}
45
46impl HttpClient {
47    pub fn transport(&self) -> &Arc<dyn Transport> {
48        &self.transport
49    }
50
51    pub fn fake(&self) -> Option<&FakeTransport> {
52        self.fake.as_ref()
53    }
54
55    pub fn request(&self, method: Method, url: impl Into<String>) -> PendingRequest {
56        PendingRequest::new(self.clone(), method, url.into(), None)
57    }
58
59    pub fn get(&self, url: impl Into<String>) -> PendingRequest {
60        self.request(Method::GET, url)
61    }
62
63    pub fn post(&self, url: impl Into<String>) -> PendingRequest {
64        self.request(Method::POST, url)
65    }
66
67    pub fn put(&self, url: impl Into<String>) -> PendingRequest {
68        self.request(Method::PUT, url)
69    }
70
71    pub fn patch(&self, url: impl Into<String>) -> PendingRequest {
72        self.request(Method::PATCH, url)
73    }
74
75    pub fn delete(&self, url: impl Into<String>) -> PendingRequest {
76        self.request(Method::DELETE, url)
77    }
78
79    pub fn named(&self, name: &str) -> NamedClient {
80        NamedClient {
81            client: self.clone(),
82            name: name.to_string(),
83        }
84    }
85}
86
87/// Bound to a named config (`[default.http.payments]`).
88#[derive(Clone)]
89pub struct NamedClient {
90    client: HttpClient,
91    name: String,
92}
93
94impl NamedClient {
95    fn cfg(&self) -> NamedClientConfig {
96        self.client
97            .named
98            .get(&self.name)
99            .cloned()
100            .unwrap_or_default()
101    }
102
103    fn resolve_url(&self, path: &str) -> Result<String, HttpError> {
104        let cfg = self.cfg();
105        let base = cfg
106            .base_url
107            .as_deref()
108            .ok_or_else(|| HttpError::Other(format!("named http client `{}` missing base_url", self.name)))?;
109        if path.starts_with("http://") || path.starts_with("https://") {
110            return Ok(path.to_string());
111        }
112        Ok(format!(
113            "{}/{}",
114            base.trim_end_matches('/'),
115            path.trim_start_matches('/')
116        ))
117    }
118
119    pub fn get(&self, path: impl Into<String>) -> PendingRequest {
120        self.request(Method::GET, path)
121    }
122
123    pub fn post(&self, path: impl Into<String>) -> PendingRequest {
124        self.request(Method::POST, path)
125    }
126
127    pub fn put(&self, path: impl Into<String>) -> PendingRequest {
128        self.request(Method::PUT, path)
129    }
130
131    pub fn patch(&self, path: impl Into<String>) -> PendingRequest {
132        self.request(Method::PATCH, path)
133    }
134
135    pub fn delete(&self, path: impl Into<String>) -> PendingRequest {
136        self.request(Method::DELETE, path)
137    }
138
139    pub fn request(&self, method: Method, path: impl Into<String>) -> PendingRequest {
140        let path = path.into();
141        let url = self.resolve_url(&path).unwrap_or(path);
142        let cfg = self.cfg();
143        let mut pending = PendingRequest::new(self.client.clone(), method, url, Some(self.name.clone()));
144        if let Some(t) = cfg.timeout {
145            pending.timeout = Some(t);
146        }
147        if let Some(n) = cfg.retries {
148            pending.retries = Some(n);
149        }
150        if let Some(ref env_key) = cfg.bearer_env {
151            if let Ok(token) = std::env::var(env_key) {
152                pending = pending.bearer_auth(token);
153            }
154        }
155        pending
156    }
157}
158
159/// Fluent outbound request.
160pub struct PendingRequest {
161    client: HttpClient,
162    method: Method,
163    url: String,
164    pub(crate) headers: HeaderMap,
165    body: Option<Bytes>,
166    timeout: Option<Duration>,
167    retries: Option<u32>,
168    idempotency_key: Option<String>,
169    named: Option<String>,
170    budget: Option<Duration>,
171}
172
173impl PendingRequest {
174    fn new(client: HttpClient, method: Method, url: String, named: Option<String>) -> Self {
175        Self {
176            client,
177            method,
178            url,
179            headers: HeaderMap::new(),
180            body: None,
181            timeout: None,
182            retries: None,
183            idempotency_key: None,
184            named,
185            budget: None,
186        }
187    }
188
189    pub fn header(mut self, name: impl AsRef<str>, value: impl AsRef<str>) -> Self {
190        if let (Ok(n), Ok(v)) = (
191            HeaderName::from_bytes(name.as_ref().as_bytes()),
192            HeaderValue::from_str(value.as_ref()),
193        ) {
194            self.headers.insert(n, v);
195        }
196        self
197    }
198
199    pub fn bearer_auth(self, token: impl AsRef<str>) -> Self {
200        self.header("authorization", format!("Bearer {}", token.as_ref()))
201    }
202
203    pub fn json<T: Serialize>(mut self, body: &T) -> Self {
204        match serde_json::to_vec(body) {
205            Ok(bytes) => {
206                self.body = Some(Bytes::from(bytes));
207                self.headers
208                    .insert(http::header::CONTENT_TYPE, HeaderValue::from_static("application/json"));
209            }
210            Err(e) => {
211                tracing::error!("json serialize: {e}");
212            }
213        }
214        self
215    }
216
217    pub fn body(mut self, bytes: impl Into<Bytes>) -> Self {
218        self.body = Some(bytes.into());
219        self
220    }
221
222    pub fn timeout(mut self, dur: Duration) -> Self {
223        self.timeout = Some(dur);
224        self
225    }
226
227    /// Cap by remaining inbound request budget.
228    pub fn with_budget(mut self, remaining: Option<Duration>) -> Self {
229        self.budget = remaining;
230        self
231    }
232
233    pub fn retry(mut self, times: u32) -> Self {
234        self.retries = Some(times);
235        self
236    }
237
238    pub fn idempotency_key(mut self, key: impl Into<String>) -> Self {
239        let key = key.into();
240        self.headers.insert(
241            HeaderName::from_static("idempotency-key"),
242            HeaderValue::from_str(&key).unwrap_or_else(|_| HeaderValue::from_static("")),
243        );
244        self.idempotency_key = Some(key);
245        self
246    }
247
248    fn effective_timeout(&self) -> Duration {
249        let base = self
250            .timeout
251            .unwrap_or(self.client.default_timeout);
252        match self.budget {
253            Some(b) if b < base => b,
254            _ => base,
255        }
256    }
257
258    pub fn send(self) -> BoxFuture<Result<OutResponse, HttpError>> {
259        Box::pin(self.send_inner())
260    }
261
262    async fn send_inner(self) -> Result<OutResponse, HttpError> {
263        let host = url::Url::parse(&self.url)
264            .ok()
265            .and_then(|u| u.host_str().map(str::to_string))
266            .unwrap_or_else(|| self.url.clone());
267
268        if !self.client.breaker.guard(&host) {
269            return Err(HttpError::CircuitOpen(host));
270        }
271
272        let timeout = self.effective_timeout();
273        if timeout.is_zero() {
274            return Err(HttpError::Timeout);
275        }
276
277        let can_retry = method_idempotent(&self.method)
278            || (self.method == Method::POST && self.idempotency_key.is_some());
279        let max_retries = if can_retry {
280            self.retries.unwrap_or(0)
281        } else {
282            0
283        };
284
285        let span = tracing::info_span!(
286            "http.client",
287            otel.kind = "client",
288            http.method = %self.method,
289            http.url = %self.url,
290            http.host = %host,
291            named = self.named.as_deref().unwrap_or(""),
292        );
293
294        async move {
295            let mut attempt = 0u32;
296            loop {
297                let out = OutRequest {
298                    method: self.method.clone(),
299                    url: self.url.clone(),
300                    headers: self.headers.clone(),
301                    body: self.body.clone(),
302                    timeout: Some(timeout),
303                };
304                let started = std::time::Instant::now();
305                let result = self.client.transport.send(out).await;
306                let elapsed = started.elapsed();
307
308                match result {
309                    Ok(res) => {
310                        let code = res.status_u16();
311                        tracing::info!(
312                            status = code,
313                            duration_ms = elapsed.as_millis() as u64,
314                            "http.client done"
315                        );
316                        let retryable = code == 429 || (500..600).contains(&code);
317                        if retryable && attempt < max_retries {
318                            attempt += 1;
319                            let delay = parse_retry_after(res.headers())
320                                .unwrap_or_else(|| {
321                                    full_jitter(
322                                        Duration::from_millis(100),
323                                        Duration::from_secs(2),
324                                        attempt,
325                                    )
326                                });
327                            self.client.breaker.record_failure(&host);
328                            tokio::time::sleep(delay).await;
329                            continue;
330                        }
331                        if code >= 500 {
332                            self.client.breaker.record_failure(&host);
333                        } else {
334                            self.client.breaker.record_success(&host);
335                        }
336                        return Ok(res);
337                    }
338                    Err(e) => {
339                        tracing::info!(
340                            error = %e,
341                            duration_ms = elapsed.as_millis() as u64,
342                            "http.client error"
343                        );
344                        let retryable = matches!(
345                            e,
346                            HttpError::Timeout | HttpError::Connect(_)
347                        );
348                        if retryable && attempt < max_retries {
349                            attempt += 1;
350                            self.client.breaker.record_failure(&host);
351                            tokio::time::sleep(full_jitter(
352                                Duration::from_millis(100),
353                                Duration::from_secs(2),
354                                attempt,
355                            ))
356                            .await;
357                            continue;
358                        }
359                        self.client.breaker.record_failure(&host);
360                        return Err(e);
361                    }
362                }
363            }
364        }
365        .instrument(span)
366        .await
367    }
368}
369
370/// Plugin: shared client + request-id middleware.
371pub struct Http {
372    deny_private: bool,
373    allow_hosts: Vec<String>,
374    max_response_size: usize,
375    default_timeout: Duration,
376    breaker: BreakerConfig,
377    fake: Option<FakeTransport>,
378    named: HashMap<String, NamedClientConfig>,
379}
380
381impl Http {
382    pub fn new() -> Self {
383        Self {
384            deny_private: true,
385            allow_hosts: Vec::new(),
386            max_response_size: DEFAULT_MAX_RESPONSE,
387            default_timeout: default_timeout(),
388            breaker: BreakerConfig::default(),
389            fake: None,
390            named: HashMap::new(),
391        }
392    }
393
394    pub fn deny_private_networks(mut self) -> Self {
395        self.deny_private = true;
396        self
397    }
398
399    pub fn allow_private_networks(mut self) -> Self {
400        self.deny_private = false;
401        self
402    }
403
404    pub fn allow_hosts(mut self, hosts: impl IntoIterator<Item = impl Into<String>>) -> Self {
405        self.allow_hosts = hosts.into_iter().map(Into::into).collect();
406        self
407    }
408
409    pub fn max_response_size(mut self, n: usize) -> Self {
410        self.max_response_size = n;
411        self
412    }
413
414    pub fn timeout(mut self, d: Duration) -> Self {
415        self.default_timeout = d;
416        self
417    }
418
419    pub fn fake() -> Self {
420        let fake = FakeTransport::new();
421        Self {
422            fake: Some(fake),
423            deny_private: false,
424            ..Self::new()
425        }
426    }
427
428    pub fn with_fake(mut self, fake: FakeTransport) -> Self {
429        self.fake = Some(fake);
430        self.deny_private = false;
431        self
432    }
433
434    pub fn stub_get(mut self, url: impl Into<String>, body: impl Into<crate::fake::StubBody>) -> Self {
435        if let Some(f) = self.fake.take() {
436            self.fake = Some(f.get(url, body));
437        }
438        self
439    }
440
441    /// Alias for [`Self::stub_get`] (Laravel-style `Http::fake().get(...)`).
442    pub fn get(self, url: impl Into<String>, body: impl Into<crate::fake::StubBody>) -> Self {
443        self.stub_get(url, body)
444    }
445
446    pub fn stub_post(mut self, url: impl Into<String>, status: u16) -> Self {
447        if let Some(f) = self.fake.take() {
448            self.fake = Some(f.post(url, status));
449        }
450        self
451    }
452
453    /// Alias for [`Self::stub_post`].
454    pub fn post(self, url: impl Into<String>, status: u16) -> Self {
455        self.stub_post(url, status)
456    }
457
458    pub fn stub_fail(mut self, url: impl Into<String>, msg: impl Into<String>) -> Self {
459        if let Some(f) = self.fake.take() {
460            self.fake = Some(f.fail(url, msg));
461        }
462        self
463    }
464
465    /// Alias for [`Self::stub_fail`].
466    pub fn fail(self, url: impl Into<String>, msg: impl Into<String>) -> Self {
467        self.stub_fail(url, msg)
468    }
469
470    /// Merge named clients from [`sova_core::ConfigDoc`] already on `app`.
471    pub fn with_config_from_app(mut self, app: &App) -> Self {
472        if let Some(doc) = app.config_doc() {
473            for (name, table) in doc.http_clients() {
474                self.named.insert(name, parse_named(&table));
475            }
476        }
477        self
478    }
479}
480
481impl Default for Http {
482    fn default() -> Self {
483        Self::new()
484    }
485}
486
487fn parse_named(table: &toml::map::Map<String, toml::Value>) -> NamedClientConfig {
488    use sova_core::extend::parse_bytes;
489    use sova_core::extend::parse_duration;
490
491    let mut cfg = NamedClientConfig::default();
492    if let Some(v) = table.get("base_url").and_then(|v| v.as_str()) {
493        cfg.base_url = Some(v.to_string());
494    }
495    if let Some(v) = table.get("timeout").and_then(|v| v.as_str()) {
496        cfg.timeout = parse_duration(v).ok();
497    }
498    if let Some(v) = table.get("retries").and_then(|v| v.as_integer()) {
499        cfg.retries = Some(v as u32);
500    }
501    if let Some(v) = table.get("max_response_size").and_then(|v| v.as_str()) {
502        cfg.max_response_size = parse_bytes(v).ok();
503    }
504    if let Some(v) = table.get("deny_private_networks").and_then(|v| v.as_bool()) {
505        cfg.deny_private_networks = Some(v);
506    }
507    if let Some(v) = table.get("bearer_env").and_then(|v| v.as_str()) {
508        cfg.bearer_env = Some(v.to_string());
509    }
510    if let Some(toml::Value::Array(arr)) = table.get("allow_hosts") {
511        cfg.allow_hosts = arr
512            .iter()
513            .filter_map(|v| v.as_str().map(str::to_string))
514            .collect();
515    }
516    cfg
517}
518
519impl Plugin for Http {
520    fn id(&self) -> &'static str {
521        "http"
522    }
523
524    fn meta(&self) -> sova_core::PluginMeta {
525        sova_core::PluginMeta::new("HTTP client")
526            .description("Outbound HTTP client with SSRF guards and named configs")
527            .version(env!("CARGO_PKG_VERSION"))
528    }
529
530    fn install(self, app: &mut App) {
531        let mut named_clients = self.named;
532        if let Some(doc) = app.config_doc() {
533            for (name, table) in doc.http_clients() {
534                named_clients
535                    .entry(name)
536                    .or_insert_with(|| parse_named(&table));
537            }
538        }
539
540        let ssrf = SsrfPolicy {
541            deny_private: self.deny_private,
542            allow_hosts: self.allow_hosts,
543        };
544
545        let (transport, fake): (Arc<dyn Transport>, Option<FakeTransport>) =
546            if let Some(fake) = self.fake.clone() {
547                (Arc::new(fake.clone()) as Arc<dyn Transport>, Some(fake))
548            } else {
549                let t = ReqwestTransport::new(ssrf.clone(), self.max_response_size)
550                    .expect("reqwest client");
551                (Arc::new(t) as Arc<dyn Transport>, None)
552            };
553
554        let client = HttpClient {
555            transport,
556            default_timeout: self.default_timeout,
557            max_response_size: self.max_response_size,
558            ssrf,
559            breaker: Arc::new(CircuitBreaker::new(self.breaker)),
560            named: Arc::new(named_clients),
561            fake,
562        };
563        app.state(client);
564        // RequestId comes from core `request_id()` middleware (presets / app).
565    }
566}
567
568pub(crate) fn propagation_headers(req: &Request) -> HeaderMap {
569    let mut h = HeaderMap::new();
570    if let Some(tp) = req.header("traceparent") {
571        if let Ok(v) = HeaderValue::from_str(tp) {
572            h.insert(HeaderName::from_static("traceparent"), v);
573        }
574    }
575    if let Some(id) = req
576        .get::<sova_core::RequestId>()
577        .map(|r| r.0.as_str())
578        .or_else(|| req.header("x-request-id"))
579    {
580        if let Ok(v) = HeaderValue::from_str(id) {
581            h.insert(HeaderName::from_static("x-request-id"), v);
582        }
583    }
584    if let Some(al) = req.header("accept-language") {
585        if let Ok(v) = HeaderValue::from_str(al) {
586            h.insert(http::header::ACCEPT_LANGUAGE, v);
587        }
588    }
589    h
590}
sova_http/client.rs

sova_http/
client.rs
