acorn/io/http/
mod.rs

1//! HTTP client utilities for ACORN IO operations
2use super::ApiResult;
3use crate::prelude::{File, OpenOptions, Path, Write};
4use crate::util::constants::app::ACORN_USER_AGENT;
5use crate::util::Label;
6use async_trait::async_trait;
7use axum::http::header::{HeaderName, HeaderValue, ACCEPT_RANGES, AUTHORIZATION, RANGE, USER_AGENT};
8use axum::http::HeaderMap;
9use bon::Builder;
10use color_eyre::eyre::eyre;
11use core::fmt;
12use jiff::Timestamp;
13use serde::{Deserialize, Serialize};
14use tokio::time::sleep;
15use tower::{service_fn, ServiceExt};
16use tracing::{debug, warn};
17
18pub mod policy;
19
20#[derive(Builder)]
21#[builder(start_fn = init)]
22struct DownloadError {
23    status_code: Option<u16>,
24    retry_headers: Option<HeaderMap>,
25    report: color_eyre::Report,
26}
27/// Async HTTP service abstraction for ACORN I/O.
28#[async_trait]
29pub trait HttpService {
30    /// Execute an HTTP request.
31    async fn execute(&self, request: HttpRequest) -> ApiResult<HttpResponse>;
32}
33/// Supported HTTP methods for ACORN I/O requests.
34#[derive(Clone, Debug, Default, Deserialize, Serialize)]
35#[serde(rename_all = "lowercase")]
36pub enum HttpMethod {
37    /// HTTP GET
38    #[default]
39    Get,
40    /// HTTP DELETE
41    Delete,
42    /// HTTP PATCH
43    Patch,
44    /// HTTP POST
45    Post,
46    /// HTTP PUT
47    Put,
48}
49/// Internal HTTP request representation.
50#[derive(Clone, Debug)]
51pub struct HttpRequest {
52    /// Request headers
53    pub headers: HeaderMap,
54    /// Optional JSON body
55    pub json_body: Option<serde_json::Value>,
56    /// HTTP method
57    pub method: HttpMethod,
58    /// URL string
59    pub url: String,
60}
61/// Fluent async request builder backed by [`HttpService`].
62#[derive(Clone, Debug)]
63pub struct HttpRequestBuilder {
64    request: HttpRequest,
65    service: ReqwestHttpService,
66}
67/// Internal HTTP response representation.
68#[derive(Clone, Debug)]
69pub struct HttpResponse {
70    /// Response body bytes
71    pub body: Vec<u8>,
72    /// Response headers
73    pub headers: HeaderMap,
74    /// HTTP status code
75    pub status_code: u16,
76}
77/// Reqwest-backed async HTTP service adapter.
78#[derive(Clone, Debug)]
79pub struct ReqwestHttpService {
80    client: reqwest::Client,
81}
82impl Default for ReqwestHttpService {
83    fn default() -> Self {
84        let policy = policy::shared_http_policy();
85        let client = reqwest::Client::builder()
86            .timeout(policy.timeout.unsigned_abs())
87            .connect_timeout(policy.connect_timeout.unsigned_abs())
88            .build()
89            .unwrap_or_else(|_| reqwest::Client::new());
90        Self { client }
91    }
92}
93impl From<&str> for HttpMethod {
94    fn from(value: &str) -> Self {
95        match value.to_uppercase().as_str() {
96            | "DELETE" => HttpMethod::Delete,
97            | "PATCH" => HttpMethod::Patch,
98            | "POST" => HttpMethod::Post,
99            | "PUT" => HttpMethod::Put,
100            | _ => HttpMethod::Get,
101        }
102    }
103}
104impl From<HttpMethod> for reqwest::Method {
105    fn from(value: HttpMethod) -> Self {
106        match value {
107            | HttpMethod::Delete => reqwest::Method::DELETE,
108            | HttpMethod::Get => reqwest::Method::GET,
109            | HttpMethod::Patch => reqwest::Method::PATCH,
110            | HttpMethod::Post => reqwest::Method::POST,
111            | HttpMethod::Put => reqwest::Method::PUT,
112        }
113    }
114}
115impl fmt::Display for HttpRequestBuilder {
116    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
117        write!(f, "{} {}", self.request.method, self.request.url)
118    }
119}
120impl HttpRequestBuilder {
121    /// Add a single header to the request, ignoring invalid names or values.
122    pub fn header(mut self, name: &str, value: &str) -> Self {
123        self.request.headers.extend(header(name, value));
124        self
125    }
126    /// Add headers to the request.
127    pub fn headers(mut self, headers: HeaderMap) -> Self {
128        self.request.headers.extend(headers);
129        self
130    }
131    /// Add a Bearer `Authorization` header to the request when `token` is non-empty.
132    pub fn bearer_auth(mut self, token: &str) -> Self {
133        self.request.headers.extend(bearer_auth(token));
134        self
135    }
136    /// Add a JSON body to the request.
137    pub fn json(mut self, value: &serde_json::Value) -> Self {
138        self.request.json_body = Some(value.clone());
139        self
140    }
141    fn new(method: HttpMethod, url: impl Into<String>) -> Self {
142        Self {
143            request: HttpRequest {
144                headers: HeaderMap::new(),
145                json_body: None,
146                method,
147                url: url.into(),
148            },
149            service: ReqwestHttpService::default(),
150        }
151    }
152    /// Send the request.
153    pub async fn send(self) -> ApiResult<HttpResponse> {
154        self.service.execute(self.request).await
155    }
156}
157impl HttpResponse {
158    /// Read response body as bytes.
159    pub async fn bytes(self) -> ApiResult<Vec<u8>> {
160        Ok(self.body)
161    }
162    /// Read response body as text.
163    pub async fn text(self) -> ApiResult<String> {
164        match String::from_utf8(self.body) {
165            | Ok(value) => Ok(value),
166            | Err(why) => Err(eyre!("HTTP response body is not valid UTF-8 — {why}")),
167        }
168    }
169}
170#[async_trait]
171impl HttpService for ReqwestHttpService {
172    async fn execute(&self, request: HttpRequest) -> ApiResult<HttpResponse> {
173        execute_with_policy(self.client.clone(), request).await
174    }
175}
176impl ReqwestHttpService {
177    fn streaming() -> Self {
178        let policy = policy::shared_http_policy();
179        let client = reqwest::Client::builder()
180            .connect_timeout(policy.connect_timeout.unsigned_abs())
181            .build()
182            .unwrap_or_else(|_| reqwest::Client::new());
183        Self { client }
184    }
185}
186/// Streams response bytes from `url` to `output` and reports cumulative progress
187pub async fn download_with_progress(
188    url: &str,
189    output: &Path,
190    mut progress: impl FnMut(u64, Option<u64>),
191    headers: Option<HeaderMap>,
192    error_message: Option<&str>,
193    auth_error_message: Option<&str>,
194    resume_from: Option<u64>,
195) -> ApiResult<()> {
196    let service = ReqwestHttpService::streaming();
197    let request = HttpRequest {
198        headers: headers.unwrap_or_default(),
199        json_body: None,
200        method: HttpMethod::Get,
201        url: url.to_string(),
202    };
203    let policy = stream_request_with_policy(
204        service.client,
205        request,
206        output,
207        &mut progress,
208        resume_from,
209        error_message,
210        auth_error_message,
211    );
212    match policy.await {
213        | Ok(()) => Ok(()),
214        | Err(why) => Err(why),
215    }
216}
217/// Check if a URL advertises byte-range support via `Accept-Ranges: bytes`.
218pub async fn supports_byte_ranges(url: &str, headers: Option<HeaderMap>) -> ApiResult<bool> {
219    let service = ReqwestHttpService::streaming();
220    let request = service
221        .client
222        .head(url)
223        .header(USER_AGENT, ACORN_USER_AGENT)
224        .headers(headers.unwrap_or_default());
225    match request.send().await {
226        | Ok(response) => Ok(response
227            .headers()
228            .get(ACCEPT_RANGES)
229            .and_then(|value| value.to_str().ok())
230            .is_some_and(|value| value.eq_ignore_ascii_case("bytes"))),
231        | Err(why) => Err(eyre!("Failed to probe HTTP range support — {why}")),
232    }
233}
234async fn stream_request_with_policy(
235    client: reqwest::Client,
236    request: HttpRequest,
237    output: &Path,
238    progress: &mut impl FnMut(u64, Option<u64>),
239    resume_from: Option<u64>,
240    error_message: Option<&str>,
241    auth_error_message: Option<&str>,
242) -> ApiResult<()> {
243    let policy = policy::shared_http_policy();
244    let method = request.method.clone();
245    let url = request.url.clone();
246    let max_attempts = policy.max_attempts();
247    let mut outcome = None;
248    for attempt in 1..=max_attempts {
249        let started = Timestamp::now();
250        let attempt_resume_from = resume_from.and_then(|_| output.metadata().ok().map(|metadata| metadata.len()).filter(|size| *size > 0));
251        let result = stream_request_once(
252            client.clone(),
253            request.clone(),
254            output,
255            &mut *progress,
256            attempt_resume_from,
257            error_message,
258            auth_error_message,
259        )
260        .await;
261        let elapsed_ms = Timestamp::now().duration_since(started).as_millis();
262        match result {
263            | Ok(()) => {
264                debug!(attempt, elapsed_ms, url, "=> {} HTTP download", Label::using());
265                outcome = Some(Ok(()));
266                break;
267            }
268            | Err(DownloadError {
269                status_code,
270                retry_headers,
271                report,
272            }) => {
273                let retry = should_retry(&method, status_code);
274                if retry && attempt < max_attempts {
275                    let delay = policy::retry_delay(retry_headers.as_ref(), attempt, Timestamp::now());
276                    warn!(
277                        attempt,
278                        elapsed_ms,
279                        retry_after_ms = delay.as_millis(),
280                        url,
281                        "=> {} Retrying HTTP download — {report}",
282                        Label::using()
283                    );
284                    sleep(delay.unsigned_abs()).await;
285                } else {
286                    warn!(attempt, elapsed_ms, url, "=> {} HTTP download failed — {report}", Label::fail());
287                    outcome = Some(Err(report));
288                    break;
289                }
290            }
291        }
292    }
293    match outcome {
294        | Some(result) => result,
295        | None => Err(eyre!("HTTP download failed after retry attempts")),
296    }
297}
298async fn stream_request_once(
299    client: reqwest::Client,
300    request: HttpRequest,
301    output: &Path,
302    progress: &mut impl FnMut(u64, Option<u64>),
303    resume_from: Option<u64>,
304    error_message: Option<&str>,
305    auth_error_message: Option<&str>,
306) -> Result<(), DownloadError> {
307    let error_message = error_message.unwrap_or("Failed to download file");
308    let HttpRequest { headers, url, .. } = request;
309    let mut request = client.get(url).header(USER_AGENT, ACORN_USER_AGENT).headers(headers);
310    if let Some(offset) = resume_from {
311        request = request.header(RANGE, format!("bytes={offset}-"));
312    }
313    match request.send().await {
314        | Ok(response) if matches!(response.status().as_u16(), 401 | 403) => {
315            let status_code = response.status().as_u16();
316            Err(DownloadError::init()
317                .status_code(status_code)
318                .retry_headers(response.headers().clone())
319                .report(eyre!(
320                    "{}",
321                    auth_error_message.unwrap_or("Failed to download file — authentication required")
322                ))
323                .build())
324        }
325        | Ok(mut response) if response.status().is_success() => {
326            let append = response.status().as_u16() == 206 && resume_from.is_some();
327            let file_result = if append {
328                OpenOptions::new().create(true).append(true).open(output)
329            } else {
330                File::create(output)
331            };
332            match file_result {
333                | Ok(mut file) => {
334                    let resumed = if append { resume_from.unwrap_or_default() } else { 0 };
335                    let total = response.content_length().map(|value| value.saturating_add(resumed));
336                    let mut downloaded = resumed;
337                    progress(downloaded, total);
338                    loop {
339                        match response.chunk().await {
340                            | Ok(Some(chunk)) => match file.write_all(&chunk) {
341                                | Ok(_) => {
342                                    downloaded = downloaded.saturating_add(chunk.len() as u64);
343                                    progress(downloaded, total);
344                                }
345                                | Err(why) => break Err(eyre!("{error_message} — failed to write download chunk — {why}")),
346                            },
347                            | Ok(None) => break Ok(()),
348                            | Err(why) => break Err(eyre!("{error_message} — {why}")),
349                        }
350                    }
351                }
352                | Err(why) => Err(eyre!("{error_message} — failed to create output file {} — {why}", output.display())),
353            }
354            .map_err(|report| DownloadError::init().report(report).build())
355        }
356        | Ok(response) => {
357            let status = response.status();
358            Err(DownloadError::init()
359                .status_code(status.as_u16())
360                .retry_headers(response.headers().clone())
361                .report(eyre!("{error_message} — HTTP {status}"))
362                .build())
363        }
364        | Err(why) => Err(DownloadError::init().report(eyre!("{error_message} — {why}")).build()),
365    }
366}
367async fn execute_with_policy(client: reqwest::Client, request: HttpRequest) -> ApiResult<HttpResponse> {
368    let policy = policy::shared_http_policy();
369    let method = request.method.clone();
370    let url = request.url.clone();
371    let max_attempts = policy.max_attempts();
372    for attempt in 1..=max_attempts {
373        let started = Timestamp::now();
374        let result = execute_with_timeout(client.clone(), request.clone()).await;
375        let elapsed_ms = Timestamp::now().duration_since(started).as_millis();
376        match result {
377            | Ok(response) => {
378                let retry = should_retry(&method, Some(response.status_code));
379                if retry && attempt < max_attempts {
380                    let delay = policy::retry_delay(Some(&response.headers), attempt, Timestamp::now());
381                    warn!(
382                        attempt,
383                        status_code = response.status_code,
384                        elapsed_ms,
385                        retry_after_ms = delay.as_millis(),
386                        url,
387                        "=> {} Retrying HTTP request",
388                        Label::using()
389                    );
390                    sleep(delay.unsigned_abs()).await;
391                } else {
392                    debug!(
393                        attempt,
394                        status_code = response.status_code,
395                        elapsed_ms,
396                        url,
397                        "=> {} HTTP request",
398                        Label::using()
399                    );
400                    return Ok(response);
401                }
402            }
403            | Err(why) => {
404                let retry = should_retry(&method, None);
405                if retry && attempt < max_attempts {
406                    let delay = policy::retry_delay(None, attempt, Timestamp::now());
407                    warn!(
408                        attempt,
409                        elapsed_ms,
410                        retry_after_ms = delay.as_millis(),
411                        url,
412                        "=> {} Retrying HTTP request — {why}",
413                        Label::using()
414                    );
415                    sleep(delay.unsigned_abs()).await;
416                } else {
417                    warn!(attempt, elapsed_ms, url, "=> {} HTTP request failed — {why}", Label::fail());
418                    return Err(why);
419                }
420            }
421        }
422    }
423    Err(eyre!("HTTP request failed after retry attempts"))
424}
425async fn execute_with_timeout(client: reqwest::Client, request: HttpRequest) -> ApiResult<HttpResponse> {
426    let service = policy::http_service_builder().service(service_fn(move |value: HttpRequest| {
427        let client = client.clone();
428        async move { invoke_request(client, value).await }
429    }));
430    service
431        .oneshot(request)
432        .await
433        .map_err(|why| eyre!("HTTP service timeout or middleware error — {why}"))
434}
435/// Build a header map from string name-value pairs, skipping invalid entries.
436pub fn headers<'a>(values: impl IntoIterator<Item = (&'a str, &'a str)>) -> HeaderMap {
437    values.into_iter().fold(HeaderMap::new(), |mut headers, (name, value)| {
438        if let (Ok(name), Ok(mut value)) = (HeaderName::from_bytes(name.as_bytes()), HeaderValue::from_str(value)) {
439            value.set_sensitive(name == AUTHORIZATION);
440            headers.append(name, value);
441        }
442        headers
443    })
444}
445/// Build a one-entry header map, skipping invalid names or values.
446pub fn header(name: &str, value: &str) -> HeaderMap {
447    headers([(name, value)])
448}
449/// Build a Bearer authorization header when `token` is non-empty.
450pub fn bearer_auth(token: &str) -> HeaderMap {
451    match token.trim() {
452        | "" => HeaderMap::new(),
453        | value => header(AUTHORIZATION.as_str(), format!("Bearer {value}").as_str()),
454    }
455}
456/// Build an HTTP request for a dynamic method.
457pub fn request(method: HttpMethod, url: impl Into<String>) -> HttpRequestBuilder {
458    HttpRequestBuilder::new(method, url)
459}
460/// Utility method to employ best practices when making async HTTP GET requests.
461pub fn get(url: impl Into<String>) -> HttpRequestBuilder {
462    HttpRequestBuilder::new(HttpMethod::Get, url)
463}
464/// Utility method to employ best practices when making async HTTP DELETE requests.
465pub fn delete(url: impl Into<String>) -> HttpRequestBuilder {
466    HttpRequestBuilder::new(HttpMethod::Delete, url)
467}
468/// Utility method to employ best practices when making async HTTP PATCH requests.
469pub fn patch(url: impl Into<String>) -> HttpRequestBuilder {
470    HttpRequestBuilder::new(HttpMethod::Patch, url)
471}
472/// Utility method to employ best practices when making async HTTP POST requests.
473pub fn post(url: impl Into<String>) -> HttpRequestBuilder {
474    HttpRequestBuilder::new(HttpMethod::Post, url)
475}
476/// Utility method to employ best practices when making async HTTP PUT requests.
477pub fn put(url: impl Into<String>) -> HttpRequestBuilder {
478    HttpRequestBuilder::new(HttpMethod::Put, url)
479}
480async fn invoke_request(client: reqwest::Client, request: HttpRequest) -> ApiResult<HttpResponse> {
481    let HttpRequest {
482        headers,
483        json_body,
484        method,
485        url,
486    } = request;
487    let builder = client.request(method.into(), url).header(USER_AGENT, ACORN_USER_AGENT).headers(headers);
488    let builder = match json_body {
489        | Some(value) => builder.json(&value),
490        | None => builder,
491    };
492    match builder.send().await {
493        | Ok(response) => {
494            let status_code = response.status().as_u16();
495            let headers = response.headers().clone();
496            match response.bytes().await {
497                | Ok(body) => Ok(HttpResponse {
498                    body: body.to_vec(),
499                    headers,
500                    status_code,
501                }),
502                | Err(why) => Err(eyre!(why)),
503            }
504        }
505        | Err(why) => Err(eyre!(why)),
506    }
507}
508/// Reads response bytes or converts request, status, and body errors into a contextual error.
509pub async fn response_body_bytes(response: ApiResult<HttpResponse>, error_message: &str) -> ApiResult<Vec<u8>> {
510    match response {
511        | Ok(value) => match value.status_code {
512            | 200..=299 => value.bytes().await.map_err(|why| eyre!("{error_message} — {why}")),
513            | status => Err(eyre!("{error_message} — HTTP {status}")),
514        },
515        | Err(why) => Err(eyre!("{error_message} — {why}")),
516    }
517}
518pub(crate) fn should_retry(method: &HttpMethod, status_code: Option<u16>) -> bool {
519    policy::should_retry(method, status_code)
520}
acorn/io/http/mod.rs

acorn/io/http/
mod.rs
