pipedash_plugin_api/
http.rs

1use std::borrow::Cow;
2use std::error::Error as StdError;
3use std::time::Duration;
4
5use chrono::{
6    DateTime,
7    Utc,
8};
9use reqwest::header::HeaderMap;
10use reqwest::Url;
11
12use crate::error::{
13    PluginError,
14    PluginResult,
15};
16use crate::utils::RetryPolicy;
17
18const MAX_RETRY_AFTER: Duration = Duration::from_secs(3600);
19const EPOCH_THRESHOLD: u64 = 1_000_000_000;
20const BODY_PREVIEW_CHARS: usize = 300;
21const BODY_PREVIEW_BYTES: usize = BODY_PREVIEW_CHARS * 4;
22
23pub async fn read_prefix(mut resp: reqwest::Response, limit: usize) -> reqwest::Result<Vec<u8>> {
24    let mut body = Vec::new();
25    while body.len() < limit {
26        let Some(chunk) = resp.chunk().await? else {
27            break;
28        };
29        let take = chunk.len().min(limit - body.len());
30        body.extend_from_slice(&chunk[..take]);
31    }
32    Ok(body)
33}
34
35pub async fn body_preview(resp: reqwest::Response) -> String {
36    let body = read_prefix(resp, BODY_PREVIEW_BYTES)
37        .await
38        .unwrap_or_default();
39    truncate_chars(String::from_utf8_lossy(&body).trim(), BODY_PREVIEW_CHARS).into_owned()
40}
41
42pub async fn check_response(resp: reqwest::Response) -> PluginResult<reqwest::Response> {
43    let status = resp.status();
44    if status.is_success() || status.is_redirection() {
45        return Ok(resp);
46    }
47
48    let url = redact_url(resp.url());
49    let headers = resp.headers().clone();
50    let body = body_preview(resp).await;
51    let message = format!("{} from {}: {}", status.as_u16(), url, body);
52    Err(from_response_parts(
53        status.as_u16(),
54        &headers,
55        message,
56        Utc::now(),
57    ))
58}
59
60pub fn from_response_parts(
61    status: u16, headers: &HeaderMap, message: String, now: DateTime<Utc>,
62) -> PluginError {
63    let retry_after = retry_after(headers, now);
64    let rate_limit_exhausted = headers
65        .get("x-ratelimit-remaining")
66        .and_then(|v| v.to_str().ok())
67        .is_some_and(|v| v.trim() == "0");
68    let has_retry_after = headers.contains_key(reqwest::header::RETRY_AFTER);
69
70    if status == 403 && (rate_limit_exhausted || has_retry_after) {
71        return PluginError::RateLimited {
72            message,
73            retry_after,
74        };
75    }
76
77    from_status(status, message, retry_after)
78}
79
80pub fn from_status(status: u16, message: String, retry_after: Option<Duration>) -> PluginError {
81    match status {
82        401 => PluginError::AuthenticationFailed(message),
83        403 => PluginError::PermissionDenied(message),
84        404 => PluginError::PipelineNotFound(message),
85        429 => PluginError::RateLimited {
86            message,
87            retry_after,
88        },
89        500..=599 => PluginError::ServiceUnavailable {
90            message,
91            retry_after,
92        },
93        _ => PluginError::ApiError(message),
94    }
95}
96
97pub fn retry_after(headers: &HeaderMap, now: DateTime<Utc>) -> Option<Duration> {
98    let header = |name: &str| {
99        headers
100            .get(name)
101            .and_then(|v| v.to_str().ok())
102            .map(str::trim)
103    };
104
105    let parsed = if let Some(value) = header("retry-after") {
106        parse_retry_after(value, now)
107    } else {
108        header("x-ratelimit-reset")
109            .or_else(|| header("ratelimit-reset"))
110            .and_then(|value| parse_reset(value, now))
111    };
112
113    parsed.map(|d| d.min(MAX_RETRY_AFTER))
114}
115
116fn parse_retry_after(value: &str, now: DateTime<Utc>) -> Option<Duration> {
117    if let Ok(secs) = value.parse::<u64>() {
118        return Some(Duration::from_secs(secs));
119    }
120    let date = DateTime::parse_from_rfc2822(value).ok()?;
121    Some(until(date.with_timezone(&Utc), now))
122}
123
124fn parse_reset(value: &str, now: DateTime<Utc>) -> Option<Duration> {
125    let secs = value.parse::<u64>().ok()?;
126    if secs < EPOCH_THRESHOLD {
127        return Some(Duration::from_secs(secs));
128    }
129    let now_secs = u64::try_from(now.timestamp()).unwrap_or(0);
130    Some(Duration::from_secs(secs.saturating_sub(now_secs)))
131}
132
133fn until(target: DateTime<Utc>, now: DateTime<Utc>) -> Duration {
134    (target - now).to_std().unwrap_or(Duration::ZERO)
135}
136
137pub fn map_send_error(err: reqwest::Error, url: &Url) -> PluginError {
138    let tls = is_tls_error(&err);
139    let network = err.is_timeout() || err.is_connect() || err.is_request();
140    let message = format!("{} ({})", err.without_url(), redact_url(url));
141
142    if tls {
143        PluginError::Tls(message)
144    } else if network {
145        PluginError::NetworkError(message)
146    } else {
147        PluginError::ApiError(message)
148    }
149}
150
151fn is_tls_error(err: &(dyn StdError + 'static)) -> bool {
152    let mut current = Some(err);
153    while let Some(e) = current {
154        if e.downcast_ref::<rustls::Error>().is_some() {
155            return true;
156        }
157        if let Some(inner) = e
158            .downcast_ref::<std::io::Error>()
159            .and_then(std::io::Error::get_ref)
160        {
161            if is_tls_error(inner) {
162                return true;
163            }
164        }
165        current = e.source();
166    }
167    false
168}
169
170pub fn truncate_chars(s: &str, max: usize) -> Cow<'_, str> {
171    match s.char_indices().nth(max) {
172        Some((idx, _)) => Cow::Owned(format!("{}…", &s[..idx])),
173        None => Cow::Borrowed(s),
174    }
175}
176
177pub fn redact_url(url: &Url) -> String {
178    let mut clean = url.clone();
179    let _ = clean.set_username("");
180    let _ = clean.set_password(None);
181    clean.set_query(None);
182    clean.set_fragment(None);
183    clean.to_string()
184}
185
186pub fn require_client(
187    client: Option<std::sync::Arc<reqwest::Client>>,
188) -> PluginResult<std::sync::Arc<reqwest::Client>> {
189    client.ok_or_else(|| PluginError::Internal("HTTP client was not provided".into()))
190}
191
192pub fn sensitive_header(value: &str, label: &str) -> PluginResult<reqwest::header::HeaderValue> {
193    let mut header = reqwest::header::HeaderValue::from_str(value)
194        .map_err(|_| PluginError::InvalidConfig(format!("Invalid {label} header")))?;
195    header.set_sensitive(true);
196    Ok(header)
197}
198
199pub fn parse_http_url(raw: &str, label: &str) -> PluginResult<Url> {
200    let url = Url::parse(raw)
201        .map_err(|e| PluginError::InvalidConfig(format!("Invalid {label} URL: {e}")))?;
202    if !matches!(url.scheme(), "http" | "https") || url.host_str().is_none() {
203        return Err(PluginError::InvalidConfig(format!(
204            "{label} URL must be an HTTP or HTTPS URL with a host"
205        )));
206    }
207    Ok(url)
208}
209
210pub async fn decode_json<T: serde::de::DeserializeOwned>(
211    response: reqwest::Response, context: &str,
212) -> PluginResult<T> {
213    let url = redact_url(response.url());
214    response.json().await.map_err(|e| {
215        PluginError::SerializationError(format!(
216            "Failed to parse {context} from {url}: {}",
217            e.without_url()
218        ))
219    })
220}
221
222pub fn api_url(base: &str, path: &str, query: &[(&str, String)], label: &str) -> PluginResult<Url> {
223    parse_http_url(&format!("{base}{path}"), label).map(|url| with_query(url, query))
224}
225
226pub fn with_query(mut url: Url, query: &[(&str, String)]) -> Url {
227    if !query.is_empty() {
228        url.query_pairs_mut().extend_pairs(query);
229    }
230    url
231}
232
233pub fn bearer_request(
234    http: &reqwest::Client, method: reqwest::Method, url: &Url, token: &str,
235) -> reqwest::RequestBuilder {
236    http.request(method, url.clone()).bearer_auth(token)
237}
238
239pub async fn send(url: &Url, request: reqwest::RequestBuilder) -> PluginResult<reqwest::Response> {
240    request.send().await.map_err(|e| map_send_error(e, url))
241}
242
243pub async fn get_json<T: serde::de::DeserializeOwned>(
244    retry: &RetryPolicy, url: &Url, context: &str, build: impl Fn() -> reqwest::RequestBuilder,
245) -> PluginResult<T> {
246    retry
247        .retry(|| async {
248            let response = send(url, build()).await?;
249            decode_json(check_response(response).await?, context).await
250        })
251        .await
252}
253
254pub async fn send_checked(
255    retry: &RetryPolicy, url: &Url, build: impl Fn() -> reqwest::RequestBuilder,
256) -> PluginResult<reqwest::Response> {
257    check_response(retry.retry_unsent(url, || build().send()).await?).await
258}
259
260#[cfg(test)]
261mod tests;
pipedash_plugin_api/http.rs

pipedash_plugin_api/
http.rs
