Skip to main content

fizzy_sdk/
client.rs

1//! The client, its builder, the account-scoped client and the request pipeline they
2//! share: hooks, credentials, retries, redirects and the response cache.
3
4use std::borrow::Cow;
5use std::fmt::Display;
6use std::sync::Arc;
7use std::time::{Duration, Instant};
8
9use bytes::Bytes;
10use serde::de::DeserializeOwned;
11use url::Url;
12
13use crate::auth::{AuthStrategy, BearerAuth, CookieAuth, StaticTokenProvider, TokenProvider};
14use crate::cache::{CachedResponse, FileCache, ResponseCache, cache_key};
15use crate::config::Config;
16use crate::error::{Error, ErrorCode, MAX_ERROR_BODY_BYTES, retry_after_seconds};
17use crate::http::header::{
18    ACCEPT, AUTHORIZATION, CONTENT_LENGTH, CONTENT_TYPE, COOKIE, IF_NONE_MATCH, USER_AGENT,
19};
20use crate::http::{
21    Body, HeaderMap, HeaderValue, HttpClient, Method, Request, Response as HttpResponse, StatusCode,
22};
23use crate::observability::{
24    Hooks, NoopHooks, OperationInfo, OperationState, RequestInfo, RequestResult,
25};
26use crate::operation::{DEFAULT_RETRY_ON, Operation, RetryPolicy};
27use crate::pagination::Page;
28use crate::route::Route;
29use crate::security::{is_same_origin, require_secure_endpoint};
30use crate::version::default_user_agent;
31
32/// How long the shipped HTTP client gives an answer to arrive.
33pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
34/// How many times a raw call is sent, the first attempt included, and the ceiling on what a
35/// modelled route may ask for. Three, as the behavior model gives every retried operation
36/// and as the Go client counts its `MaxRetries`.
37pub const DEFAULT_MAX_ATTEMPTS: u32 = 3;
38/// The first backoff for a raw call.
39pub const DEFAULT_BASE_DELAY: Duration = Duration::from_secs(1);
40/// The longest the client waits between attempts, however many it has made. The backoff's
41/// job is to stop hammering, and thirty seconds does it; a caller who raised the retry
42/// count would otherwise be waiting minutes on a server that is not coming back. Move it
43/// with [`ClientBuilder::max_delay`].
44pub const DEFAULT_MAX_DELAY: Duration = Duration::from_secs(30);
45/// The longest `Retry-After` the client sits out. Past it the answer is handed back as the
46/// rate-limit error it is, with the wait in the hint, rather than the call blocking for
47/// an hour. Move it with [`ClientBuilder::max_retry_after`].
48pub const DEFAULT_MAX_RETRY_AFTER: Duration = Duration::from_secs(60);
49/// The most random time added to a backoff.
50pub const DEFAULT_MAX_JITTER: Duration = Duration::from_millis(100);
51/// How many pages a walk reads before stopping.
52pub const DEFAULT_MAX_PAGES: usize = 10_000;
53/// The most of an answer the client holds in memory.
54pub const DEFAULT_MAX_RESPONSE_BODY_BYTES: usize = 10 << 20;
55
56/// How many redirects one request may go through before the client gives up on it.
57const MAX_REDIRECTS: usize = 10;
58
59/// A Fizzy client: one set of credentials on one origin. Most of the API is scoped to an
60/// account, reached with [`Client::for_account`]; what is not — sessions, identity, access
61/// tokens — hangs off the client itself.
62///
63/// Clients are cheap to clone and share their connection pool, credentials and cache.
64#[derive(Clone)]
65pub struct Client {
66    pub(crate) shared: Arc<Shared>,
67}
68
69pub(crate) struct Shared {
70    pub(crate) config: Config,
71    pub(crate) base_url: Url,
72    pub(crate) http: Arc<dyn HttpClient>,
73    pub(crate) auth: Arc<dyn AuthStrategy>,
74    pub(crate) user_agent: String,
75    pub(crate) max_attempts: u32,
76    pub(crate) base_delay: Duration,
77    pub(crate) max_delay: Duration,
78    pub(crate) max_retry_after: Duration,
79    pub(crate) max_jitter: Duration,
80    pub(crate) max_pages: usize,
81    pub(crate) max_response_body_bytes: usize,
82    pub(crate) cache: Option<Arc<dyn ResponseCache>>,
83    pub(crate) hooks: Arc<dyn Hooks>,
84}
85
86/// A client scoped to one account: every path it sends starts with the account id. Made
87/// with [`Client::for_account`].
88#[derive(Clone)]
89pub struct AccountClient {
90    client: Client,
91    account_id: String,
92}
93
94impl std::fmt::Debug for Client {
95    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
96        f.debug_struct("Client")
97            .field("base_url", &self.shared.base_url.as_str())
98            .field("user_agent", &self.shared.user_agent)
99            .finish_non_exhaustive()
100    }
101}
102
103impl std::fmt::Debug for AccountClient {
104    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
105        f.debug_struct("AccountClient")
106            .field("account_id", &self.account_id)
107            .field("client", &self.client)
108            .finish()
109    }
110}
111
112impl AccountClient {
113    /// The client underneath.
114    pub fn client(&self) -> &Client {
115        &self.client
116    }
117
118    /// The account every call is scoped to.
119    pub fn account_id(&self) -> &str {
120        &self.account_id
121    }
122
123    /// The scope generated services send through.
124    pub fn scope(&self) -> Scope<'_> {
125        Scope {
126            client: &self.client,
127            account_id: Some(&self.account_id),
128        }
129    }
130}
131
132/// What a generated service sends through: the client, and the account when there is one.
133#[derive(Clone, Copy)]
134pub struct Scope<'a> {
135    client: &'a Client,
136    account_id: Option<&'a str>,
137}
138
139impl<'a> Scope<'a> {
140    /// The client underneath.
141    pub fn client(&self) -> &'a Client {
142        self.client
143    }
144
145    /// The account, when the scope has one.
146    pub fn account_id(&self) -> Option<&'a str> {
147        self.account_id
148    }
149
150    /// Starts a request for a modelled route. An account-scoped route needs an account,
151    /// and asks for one as a usage error rather than sending a path with a hole in it.
152    pub fn operation(
153        &self,
154        route: &'static Route,
155        params: &[&dyn Display],
156    ) -> Result<Operation, Error> {
157        if route.account_scoped && self.account_id.is_none() {
158            return Err(Error::usage_with_hint(
159                format!("{} needs an account", route.id),
160                "reach it through Client::for_account",
161            ));
162        }
163        let account_id = if route.account_scoped {
164            self.account_id
165        } else {
166            None
167        };
168        Operation::for_route(route, account_id, params)
169    }
170}
171
172/// What came back from Fizzy, before it is decoded.
173#[derive(Clone)]
174#[non_exhaustive]
175pub struct Response {
176    /// The status.
177    pub status: StatusCode,
178    /// The headers.
179    pub headers: HeaderMap,
180    /// The body, read whole.
181    pub body: Bytes,
182    /// Where the answer came from, once any redirects were followed.
183    pub url: Url,
184    /// The body came out of the response cache: Fizzy answered 304 and the client read the
185    /// entry it was holding.
186    pub from_cache: bool,
187}
188
189/// The body never prints, and the headers print redacted: an answer may carry a session
190/// cookie or a person's details, and `{:?}` of a response is the kind of thing that ends
191/// up in a log.
192impl std::fmt::Debug for Response {
193    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
194        f.debug_struct("Response")
195            .field("status", &self.status)
196            .field("headers", &crate::security::redact_headers(&self.headers))
197            .field("body_len", &self.body.len())
198            .field("url", &self.url.as_str())
199            .field("from_cache", &self.from_cache)
200            .finish()
201    }
202}
203
204impl Response {
205    /// Decodes the body as JSON. A body that does not read as `T` is an API error carrying
206    /// the answer's status and request id, so the failure can still be traced.
207    pub fn json<T: DeserializeOwned>(&self) -> Result<T, Error> {
208        let status = self.status.as_u16();
209        let error = if self.body.is_empty() {
210            Error::api(status, "empty response body")
211        } else {
212            match serde_json::from_slice(&self.body) {
213                Ok(value) => return Ok(value),
214                Err(error) => Error::api(status, "unexpected JSON")
215                    .with_hint(error.to_string())
216                    .with_source(error),
217            }
218        };
219        Err(match self.header("x-request-id") {
220            Some(request_id) => error.with_request_id(request_id),
221            None => error,
222        })
223    }
224
225    /// A header, when it is there and reads as text.
226    pub fn header(&self, name: &str) -> Option<&str> {
227        self.headers.get(name).and_then(|value| value.to_str().ok())
228    }
229}
230
231/// How a raw call is sent, where the defaults are not what the caller wants.
232#[derive(Debug, Clone, Default, PartialEq, Eq)]
233pub struct RequestOptions {
234    /// Send once, whatever the method.
235    pub no_retry: bool,
236    /// Resend a POST the way a GET is resent.
237    pub idempotent: bool,
238}
239
240impl RequestOptions {
241    /// The defaults: everything but a POST is retried.
242    pub fn new() -> RequestOptions {
243        RequestOptions::default()
244    }
245
246    /// Send once.
247    pub fn no_retry(mut self) -> RequestOptions {
248        self.no_retry = true;
249        self
250    }
251
252    /// Treat the call as safe to resend.
253    pub fn idempotent(mut self) -> RequestOptions {
254        self.idempotent = true;
255        self
256    }
257
258    pub(crate) fn apply(&self, mut operation: Operation) -> Operation {
259        if self.idempotent {
260            operation.idempotent(true);
261        }
262        if self.no_retry {
263            operation.no_retry();
264        }
265        operation
266    }
267}
268
269/// Builds a [`Client`].
270pub struct ClientBuilder {
271    config: Config,
272    auth: Option<Arc<dyn AuthStrategy>>,
273    http: Option<Arc<dyn HttpClient>>,
274    user_agent: String,
275    timeout: Duration,
276    max_attempts: u32,
277    base_delay: Duration,
278    max_delay: Duration,
279    max_retry_after: Duration,
280    max_jitter: Duration,
281    max_pages: usize,
282    max_response_body_bytes: usize,
283    cache: Option<Arc<dyn ResponseCache>>,
284    pub(crate) hooks: Arc<dyn Hooks>,
285}
286
287impl ClientBuilder {
288    /// A builder over a config, with nothing else decided.
289    pub fn new(config: Config) -> ClientBuilder {
290        ClientBuilder {
291            config,
292            auth: None,
293            http: None,
294            user_agent: default_user_agent(),
295            timeout: DEFAULT_TIMEOUT,
296            max_attempts: DEFAULT_MAX_ATTEMPTS,
297            base_delay: DEFAULT_BASE_DELAY,
298            max_delay: DEFAULT_MAX_DELAY,
299            max_retry_after: DEFAULT_MAX_RETRY_AFTER,
300            max_jitter: DEFAULT_MAX_JITTER,
301            max_pages: DEFAULT_MAX_PAGES,
302            max_response_body_bytes: DEFAULT_MAX_RESPONSE_BODY_BYTES,
303            cache: None,
304            hooks: Arc::new(NoopHooks),
305        }
306    }
307
308    /// Sends `Authorization: Bearer` from a token provider.
309    pub fn token_provider(self, provider: impl TokenProvider + 'static) -> ClientBuilder {
310        self.auth_strategy(BearerAuth::new(provider))
311    }
312
313    /// Sends `Authorization: Bearer` with a fixed access token.
314    pub fn access_token(self, token: impl Into<crate::types::SensitiveString>) -> ClientBuilder {
315        self.token_provider(StaticTokenProvider::new(token))
316    }
317
318    /// Sends `Cookie: session_token=` with a fixed session token, as a magic-link login
319    /// hands out.
320    pub fn session_token(self, token: impl Into<crate::types::SensitiveString>) -> ClientBuilder {
321        self.auth_strategy(CookieAuth::new(StaticTokenProvider::new(token)))
322    }
323
324    /// Puts credentials on with something of the caller's own.
325    pub fn auth_strategy(mut self, strategy: impl AuthStrategy + 'static) -> ClientBuilder {
326        self.auth = Some(Arc::new(strategy));
327        self
328    }
329
330    /// Replaces the HTTP client every request goes out on. The one supplied must not
331    /// follow redirects; see [`HttpClient`]. The timeout set on the builder is then ignored
332    /// — a timeout belongs to the client that can enforce it.
333    pub fn http_client(mut self, http: impl HttpClient + 'static) -> ClientBuilder {
334        self.http = Some(Arc::new(http));
335        self
336    }
337
338    /// Replaces the `User-Agent`.
339    pub fn user_agent(mut self, user_agent: impl Into<String>) -> ClientBuilder {
340        self.user_agent = user_agent.into();
341        self
342    }
343
344    /// How long the HTTP client the SDK ships gives an answer to arrive. It has no effect on
345    /// one supplied with [`ClientBuilder::http_client`].
346    pub fn timeout(mut self, timeout: Duration) -> ClientBuilder {
347        self.timeout = timeout;
348        self
349    }
350
351    /// How many times a raw call is sent, the first attempt included, and the most a
352    /// modelled route may be sent whatever the behavior model says. One sends everything
353    /// once; zero reads as one.
354    pub fn max_attempts(mut self, max_attempts: u32) -> ClientBuilder {
355        self.max_attempts = max_attempts.max(1);
356        self
357    }
358
359    /// The first backoff for a raw call.
360    pub fn base_delay(mut self, base_delay: Duration) -> ClientBuilder {
361        self.base_delay = base_delay;
362        self
363    }
364
365    /// The longest wait between attempts, whatever the backoff or the route's own base
366    /// delay would have made it.
367    pub fn max_delay(mut self, max_delay: Duration) -> ClientBuilder {
368        self.max_delay = max_delay;
369        self
370    }
371
372    /// The longest `Retry-After` the client sits out before handing the answer back.
373    pub fn max_retry_after(mut self, max_retry_after: Duration) -> ClientBuilder {
374        self.max_retry_after = max_retry_after;
375        self
376    }
377
378    /// The most random time added to a backoff. Zero makes the waits exact.
379    pub fn max_jitter(mut self, max_jitter: Duration) -> ClientBuilder {
380        self.max_jitter = max_jitter;
381        self
382    }
383
384    /// How many pages a walk reads before stopping.
385    pub fn max_pages(mut self, max_pages: usize) -> ClientBuilder {
386        self.max_pages = max_pages;
387        self
388    }
389
390    /// The most an answer may deliver before the client refuses to hold it. Zero asks for
391    /// the default: the cap cannot be lifted, only moved.
392    pub fn max_response_body_bytes(mut self, bytes: usize) -> ClientBuilder {
393        self.max_response_body_bytes = bytes;
394        self
395    }
396
397    /// Caches JSON reads by `ETag`. Without this, `config.cache_enabled` decides whether a
398    /// [`FileCache`] in `config.cache_dir` is used.
399    pub fn cache(mut self, cache: impl ResponseCache + 'static) -> ClientBuilder {
400        self.cache = Some(Arc::new(cache));
401        self
402    }
403
404    /// Reports every operation and every request the client makes. Several sets of hooks
405    /// go on as one with [`crate::observability::ChainHooks`].
406    pub fn hooks(mut self, hooks: impl Hooks + 'static) -> ClientBuilder {
407        self.hooks = Arc::new(hooks);
408        self
409    }
410
411    /// Builds the client, refusing a base URL that would carry credentials over plain HTTP
412    /// anywhere but this machine.
413    pub fn build(self) -> Result<Client, Error> {
414        let base_url = parse_base_url(&self.config.base_url)?;
415        let auth = self
416            .auth
417            .ok_or_else(|| Error::usage("a token provider or auth strategy is required"))?;
418        if self.timeout.is_zero() {
419            return Err(Error::usage("timeout must be greater than zero"));
420        }
421        if self.max_pages == 0 {
422            return Err(Error::usage("max pages must be greater than zero"));
423        }
424        let http = match self.http {
425            Some(http) => http,
426            None => shipped_http_client(self.timeout)?,
427        };
428        let cache =
429            match (self.cache, self.config.cache_enabled) {
430                (Some(cache), _) => Some(cache),
431                (None, true) => Some(Arc::new(FileCache::new(self.config.cache_dir.clone()))
432                    as Arc<dyn ResponseCache>),
433                (None, false) => None,
434            };
435        let max_response_body_bytes = match self.max_response_body_bytes {
436            0 => DEFAULT_MAX_RESPONSE_BODY_BYTES,
437            bytes => bytes,
438        };
439        let shared = Shared {
440            config: self.config,
441            base_url,
442            http,
443            auth,
444            user_agent: self.user_agent,
445            max_attempts: self.max_attempts,
446            base_delay: self.base_delay.min(self.max_delay),
447            max_delay: self.max_delay,
448            max_retry_after: self.max_retry_after,
449            max_jitter: self.max_jitter,
450            max_pages: self.max_pages,
451            max_response_body_bytes,
452            cache,
453            hooks: self.hooks,
454        };
455        Ok(Client {
456            shared: Arc::new(shared),
457        })
458    }
459}
460
461impl Client {
462    /// A builder over a config.
463    pub fn builder(config: Config) -> ClientBuilder {
464        ClientBuilder::new(config)
465    }
466
467    /// A client with the default settings and a bearer token, on the HTTP client the SDK
468    /// ships. Without the `reqwest` feature there is no such client, and a
469    /// [`ClientBuilder`] with an [`HttpClient`] of the application's own is the way in.
470    #[cfg(feature = "reqwest")]
471    #[cfg_attr(docsrs, doc(cfg(feature = "reqwest")))]
472    pub fn new(config: Config, provider: impl TokenProvider + 'static) -> Result<Client, Error> {
473        Client::builder(config).token_provider(provider).build()
474    }
475
476    /// The config the client was built from.
477    pub fn config(&self) -> &Config {
478        &self.shared.config
479    }
480
481    /// The origin every call goes to.
482    pub fn base_url(&self) -> &Url {
483        &self.shared.base_url
484    }
485
486    /// How many pages a walk reads before stopping.
487    pub fn max_pages(&self) -> usize {
488        self.shared.max_pages
489    }
490
491    /// A client scoped to one account. The id is checked for shape here — it goes into
492    /// every path as one segment — and against Fizzy on the first call.
493    pub fn for_account(&self, account_id: impl Into<String>) -> Result<AccountClient, Error> {
494        let account_id = account_id.into();
495        if account_id.is_empty()
496            || account_id == "."
497            || account_id == ".."
498            || account_id.contains(['/', '?', '#', '%'])
499            || account_id
500                .chars()
501                .any(|c| c.is_whitespace() || c.is_control())
502        {
503            return Err(Error::usage(format!("invalid account id {account_id:?}")));
504        }
505        Ok(AccountClient {
506            client: self.clone(),
507            account_id,
508        })
509    }
510
511    /// A client scoped to the account the config names, when it names one.
512    pub fn for_configured_account(&self) -> Result<AccountClient, Error> {
513        match &self.shared.config.account {
514            Some(account_id) => self.for_account(account_id.clone()),
515            None => Err(Error::usage_with_hint(
516                "no account configured",
517                "set FIZZY_ACCOUNT or Config::account, or call Client::for_account",
518            )),
519        }
520    }
521
522    /// The scope the account-free services send through.
523    pub fn scope(&self) -> Scope<'_> {
524        Scope {
525            client: self,
526            account_id: None,
527        }
528    }
529
530    /// Starts a request for one of the modelled routes that needs no account. Generated
531    /// service methods go through [`Scope::operation`]; reach for this directly only to
532    /// add headers or query parameters they do not expose.
533    pub fn operation(
534        &self,
535        route: &'static Route,
536        params: &[&dyn Display],
537    ) -> Result<Operation, Error> {
538        self.scope().operation(route, params)
539    }
540
541    /// Starts a request for a path the model does not cover. The path is relative to the
542    /// base URL and gets the same credentials and retry treatment as a modelled one.
543    pub fn request(&self, method: Method, path: impl Into<String>) -> Operation {
544        Operation::raw(method, path.into())
545    }
546
547    /// Sends an operation and decodes its JSON body.
548    pub async fn send<T: DeserializeOwned>(&self, operation: Operation) -> Result<T, Error> {
549        self.execute(operation).await?.json()
550    }
551
552    /// Sends an operation whose answer carries no body worth reading.
553    pub async fn send_unit(&self, operation: Operation) -> Result<(), Error> {
554        self.execute(operation).await.map(|_| ())
555    }
556
557    /// Sends a paginated read and keeps the cursor Fizzy answered with.
558    pub async fn send_page<T: DeserializeOwned>(
559        &self,
560        operation: Operation,
561    ) -> Result<Page<T>, Error> {
562        let info = operation.info.clone();
563        let retry = Some(self.policy_for(&operation));
564        let response = self.execute(operation).await?;
565        Ok(Page::new(response.json()?, &response, info, retry))
566    }
567
568    /// Sends an operation: asks the hooks whether it may run, applies credentials, retries
569    /// transient failures under the operation's policy, and answers a cached body on 304.
570    /// Non-2xx statuses become errors.
571    pub async fn execute(&self, operation: Operation) -> Result<Response, Error> {
572        let work = self.dispatch(&operation);
573        #[cfg(feature = "tracing")]
574        let work = {
575            use tracing::Instrument;
576            let span = tracing::info_span!(
577                "fizzy.operation",
578                operation = %operation.info.operation,
579                service = %operation.info.service,
580                http.status = tracing::field::Empty,
581                request_id = tracing::field::Empty,
582            );
583            async move {
584                let outcome = work.await;
585                let span = tracing::Span::current();
586                match &outcome {
587                    Ok(response) => {
588                        span.record("http.status", response.status.as_u16());
589                        if let Some(id) = response.header("x-request-id") {
590                            span.record("request_id", id);
591                        }
592                    }
593                    Err(error) => {
594                        if let Some(status) = error.http_status() {
595                            span.record("http.status", status);
596                        }
597                        if let Some(id) = error.request_id() {
598                            span.record("request_id", id);
599                        }
600                    }
601                }
602                outcome
603            }
604            .instrument(span)
605        };
606        self.instrument(&operation, work).await
607    }
608
609    /// Runs one operation inside the hook lifecycle every call shares.
610    ///
611    /// The end is reported from a drop guard rather than after the await, because the await
612    /// may never return: a caller's `tokio::time::timeout` or `select!` can drop the future
613    /// mid-flight, and a start with no end leaves the bulkhead a permit short and the
614    /// circuit breaker a call short for the life of the client. Dropped that way, the
615    /// operation ends as [`Error::cancelled`].
616    async fn instrument<T>(
617        &self,
618        operation: &Operation,
619        work: impl Future<Output = Result<T, Error>>,
620    ) -> Result<T, Error> {
621        let hooks = &self.shared.hooks;
622        hooks.on_operation_gate(&operation.info).await?;
623
624        let mut running = Running {
625            hooks,
626            info: &operation.info,
627            state: Some(hooks.on_operation_start(&operation.info)),
628            started: Instant::now(),
629        };
630        let outcome = work.await;
631        running.finished(outcome.as_ref().map(|_| ()));
632        outcome
633    }
634
635    /// Reads the answer the retry loop settled on, and tells the hooks how it turned out
636    /// once its body has been dealt with.
637    async fn dispatch(&self, operation: &Operation) -> Result<Response, Error> {
638        let url = self.url_for(operation)?;
639        let answered = self.attempt(operation, &url).await?;
640        let status = answered.status;
641        let finished = self
642            .finish(
643                operation,
644                &url,
645                answered.url,
646                status,
647                answered.headers,
648                answered.body,
649                answered.cached,
650            )
651            .await;
652        self.shared.hooks.on_request_end(
653            &answered.info,
654            &RequestResult {
655                status: Some(status),
656                duration: answered.duration,
657                error: finished.as_ref().err(),
658                from_cache: finished.as_ref().is_ok_and(|response| response.from_cache),
659                retryable: answered.retryable,
660                retry_after: answered.retry_after,
661            },
662        );
663        finished
664    }
665
666    /// The policy an operation is sent under: its own, or the client's defaults for a raw
667    /// call, and in either case no more attempts and no longer a first wait than the
668    /// client allows. A call that is not idempotent is sent once whatever else says.
669    pub(crate) fn policy_for(&self, operation: &Operation) -> RetryPolicy {
670        let shared = &self.shared;
671        if !operation.idempotent {
672            return RetryPolicy::none();
673        }
674        let policy = operation.retry.clone().unwrap_or_else(|| RetryPolicy {
675            attempts: shared.max_attempts,
676            base_delay: shared.base_delay,
677            retry_on: Cow::Borrowed(DEFAULT_RETRY_ON),
678        });
679        RetryPolicy {
680            attempts: policy.attempts.min(shared.max_attempts).max(1),
681            base_delay: policy.base_delay.min(shared.max_delay),
682            retry_on: policy.retry_on,
683        }
684    }
685
686    /// Sends the operation as many times as its retry budget and Fizzy's answers call for,
687    /// and hands back the answer it stopped on with the body still unread.
688    async fn attempt(&self, operation: &Operation, url: &Url) -> Result<Answered, Error> {
689        let policy = self.policy_for(operation);
690        let mut backoff = Backoff {
691            attempt: 1,
692            delay: policy.base_delay,
693        };
694        // Looked up once and carried across the attempts: a resend would find the same
695        // entry, and the cache the SDK ships reads it off disk.
696        let mut cached = None;
697
698        loop {
699            let once = self.attempt_once(operation, url, &policy, &mut backoff, &mut cached);
700            if let Some(answered) = once.await? {
701                return Ok(answered);
702            }
703        }
704    }
705
706    /// One request, and what came of it: the answer the loop settles on, or `None` once
707    /// the wait before the next attempt is over.
708    async fn attempt_once(
709        &self,
710        operation: &Operation,
711        url: &Url,
712        policy: &RetryPolicy,
713        backoff: &mut Backoff,
714        cached: &mut Option<(String, CachedResponse)>,
715    ) -> Result<Option<Answered>, Error> {
716        let hooks = &self.shared.hooks;
717        let attempt = backoff.attempt;
718        let request = self.prepare(operation, url, cached).await?;
719        let info = RequestInfo {
720            method: operation.method.clone(),
721            url: url.clone(),
722            attempt,
723        };
724        hooks.on_request_start(&info);
725        let started = Instant::now();
726        let sent = self.transmit(operation, url.clone(), request).await;
727        let duration = started.elapsed();
728
729        let (final_url, response) = match sent {
730            Err(error) => {
731                let again = error.is_retryable() && attempt < policy.attempts;
732                hooks.on_request_end(
733                    &info,
734                    &RequestResult::failed(None, duration, &error, error.is_retryable(), None),
735                );
736                return if again {
737                    self.resend(backoff, &info, operation, &error, None, "request failed")
738                        .await;
739                    Ok(None)
740                } else {
741                    Err(error)
742                };
743            }
744            Ok(sent) => sent,
745        };
746
747        let status = response.status();
748        let retryable = policy.retry_on.contains(&status.as_u16());
749        let retry_after = retry_after_asked(status, response.headers());
750        let wait = match retry_after {
751            Some(seconds) if seconds > 0 => Some(Duration::from_secs(seconds)),
752            _ => None,
753        };
754        let too_long = wait.is_some_and(|wait| wait > self.shared.max_retry_after);
755        if retryable && attempt < policy.attempts && !too_long {
756            let cause = Error::from_response(status, &operation.method, response.headers(), &[]);
757            hooks.on_request_end(
758                &info,
759                &RequestResult::failed(Some(status), duration, &cause, retryable, retry_after),
760            );
761            self.resend(backoff, &info, operation, &cause, wait, "retryable status")
762                .await;
763            return Ok(None);
764        }
765
766        let (parts, body) = response.into_parts();
767        match self.read_answer(operation, url, status, body).await {
768            Ok(body) => Ok(Some(Answered {
769                url: final_url,
770                status,
771                headers: parts.headers,
772                body,
773                cached: cached.take(),
774                info,
775                duration,
776                retryable,
777                retry_after,
778            })),
779            Err(error) => {
780                let (error, retryable) = unread(operation, status, &parts.headers, error);
781                hooks.on_request_end(
782                    &info,
783                    &RequestResult::failed(Some(status), duration, &error, retryable, retry_after),
784                );
785                if retryable && attempt < policy.attempts {
786                    self.resend(backoff, &info, operation, &error, None, "body broke off")
787                        .await;
788                    Ok(None)
789                } else {
790                    Err(error)
791                }
792            }
793        }
794    }
795
796    /// Tells the hooks a resend is coming, waits it out — `wait` when Fizzy named one,
797    /// the backoff otherwise — and moves the loop on to the next attempt.
798    async fn resend(
799        &self,
800        backoff: &mut Backoff,
801        info: &RequestInfo,
802        operation: &Operation,
803        error: &Error,
804        wait: Option<Duration>,
805        why: &str,
806    ) {
807        crate::trace::debug(&operation.id, backoff.attempt, &format!("{why}, retrying"));
808        self.shared.hooks.on_retry(info, backoff.attempt + 1, error);
809        self.wait(wait.unwrap_or(backoff.delay)).await;
810        backoff.delay = self.next_delay(backoff.delay);
811        backoff.attempt += 1;
812    }
813
814    /// Reads the body of the answer an attempt settled on: whole, up to the cap, for a
815    /// success; no more than the diagnostic prefix a failure keeps, for anything else.
816    async fn read_answer(
817        &self,
818        operation: &Operation,
819        url: &Url,
820        status: StatusCode,
821        body: Body,
822    ) -> Result<Bytes, Error> {
823        if status.is_success() {
824            let bound = self.shared.max_response_body_bytes;
825            read_body(body, bound, &operation.method, url.path()).await
826        } else {
827            body.prefix(MAX_ERROR_BODY_BYTES).await
828        }
829    }
830
831    /// Where an operation goes. Whatever the path was — relative, absolute, pasted — the
832    /// resolved URL has to sit on the Fizzy origin the client was built for: every request
833    /// carries the credentials, and this is the one place all of them pass through.
834    pub(crate) fn url_for(&self, operation: &Operation) -> Result<Url, Error> {
835        let mut url = match &operation.url {
836            Some(url) => url.clone(),
837            None => self
838                .shared
839                .base_url
840                .join(operation.path.trim_start_matches('/'))?,
841        };
842        if !operation.query.is_empty() {
843            url.query_pairs_mut().extend_pairs(&operation.query);
844        }
845        require_secure_endpoint(&url)?;
846        if !url.username().is_empty() || url.password().is_some() {
847            return Err(Error::usage(format!(
848                "{} names a URL carrying credentials; use an access token or a session token",
849                operation.id
850            )));
851        }
852        if !is_same_origin(&url, &self.shared.base_url) {
853            return Err(Error::usage(format!(
854                "{} resolves off the Fizzy origin {}, onto {}",
855                operation.id,
856                self.shared.base_url.origin().ascii_serialization(),
857                url.origin().ascii_serialization()
858            )));
859        }
860        Ok(url)
861    }
862
863    /// Builds the request for one attempt, and looks the response cache up the first time
864    /// it is asked for a key. `cached` carries the entry — or the empty stand-in that says
865    /// "cacheable, nothing stored" — from one attempt to the next.
866    async fn prepare(
867        &self,
868        operation: &Operation,
869        url: &Url,
870        cached: &mut Option<(String, CachedResponse)>,
871    ) -> Result<Request<Bytes>, Error> {
872        let mut request = Request::builder()
873            .method(operation.method.clone())
874            .uri(url.as_str())
875            .body(Bytes::new())
876            .map_err(Error::from_std)?;
877        let headers = request.headers_mut();
878        headers.insert(USER_AGENT, header_value(&self.shared.user_agent)?);
879        headers.insert(ACCEPT, HeaderValue::from_static("application/json"));
880        for (name, value) in &operation.headers {
881            headers.append(name.clone(), value.clone());
882        }
883        if let Some(body) = &operation.body {
884            headers.insert(CONTENT_TYPE, header_value(&body.content_type)?);
885            *request.body_mut() = body.bytes.clone();
886        }
887        self.shared.auth.authenticate(&mut request).await?;
888
889        let key = match self.cacheable(operation) {
890            None => None,
891            Some(cache) => match credential(request.headers()) {
892                None => None,
893                Some(credential) => {
894                    let key = cache_key(url.as_str(), &credential);
895                    if cached.as_ref().is_none_or(|(held, _)| *held != key) {
896                        *cached = self.look_up(cache, &key).await;
897                    }
898                    Some(key)
899                }
900            },
901        };
902        if key.is_none() {
903            *cached = None;
904        }
905        if let Some((_, entry)) = cached.as_ref()
906            && !entry.etag.is_empty()
907        {
908            let validator = header_value(&entry.etag)?;
909            request.headers_mut().insert(IF_NONE_MATCH, validator);
910        }
911        Ok(request)
912    }
913
914    /// What the cache holds for a key, as the attempt should carry it: the stored entry, an
915    /// empty stand-in when there is nothing stored, and nothing at all when what is stored
916    /// is longer than the client would hold — which is thrown away on the way past.
917    async fn look_up(
918        &self,
919        cache: &Arc<dyn ResponseCache>,
920        key: &str,
921    ) -> Option<(String, CachedResponse)> {
922        match cache_get(cache, key).await {
923            Some(entry) if entry.body.len() <= self.shared.max_response_body_bytes => {
924                Some((key.to_string(), entry))
925            }
926            Some(_) => {
927                cache_invalidate(cache, key).await;
928                None
929            }
930            None => Some((
931                key.to_string(),
932                CachedResponse {
933                    etag: String::new(),
934                    body: Bytes::new(),
935                },
936            )),
937        }
938    }
939
940    /// The cache the operation reads and writes, when there is one to use. Cached bodies
941    /// are held per identity, so a request that goes out without credentials is not
942    /// cached: there would be nothing to tell one caller's copy from another's.
943    fn cacheable(&self, operation: &Operation) -> Option<&Arc<dyn ResponseCache>> {
944        if operation.no_cache || operation.method != Method::GET {
945            None
946        } else {
947            self.shared.cache.as_ref()
948        }
949    }
950
951    /// Sends one request and follows the redirects it is answered with, up to
952    /// [`MAX_REDIRECTS`] hops, as long as they stay on the Fizzy origin. Hands back the
953    /// URL the answer came from along with the answer.
954    ///
955    /// A hop off the origin is refused rather than followed: Fizzy's API never sends one,
956    /// and following it would carry the credentials — the ones the client puts on, and any
957    /// the transport adds of its own — somewhere they were never meant to go. A 301, 302
958    /// or 303 turns anything but a GET or HEAD into a GET without its body; a 307 or 308
959    /// keeps both.
960    async fn transmit(
961        &self,
962        operation: &Operation,
963        mut url: Url,
964        mut request: Request<Bytes>,
965    ) -> Result<(Url, HttpResponse<Body>), Error> {
966        let mut hops = 0;
967        loop {
968            let outgoing = (
969                request.method().clone(),
970                request.headers().clone(),
971                request.body().clone(),
972            );
973            let response = self.shared.http.send(request).await?;
974            match redirect_target(&url, &response) {
975                None => return Ok((url, response)),
976                Some(_) if hops == MAX_REDIRECTS => {
977                    return Err(Error::new(
978                        ErrorCode::Network,
979                        format!(
980                            "{} redirected more than {MAX_REDIRECTS} times",
981                            operation.id
982                        ),
983                    )
984                    .retryable());
985                }
986                Some(next) => {
987                    require_secure_endpoint(&next)?;
988                    if !next.username().is_empty() || next.password().is_some() {
989                        return Err(Error::usage(format!(
990                            "{} redirected to a URL carrying credentials",
991                            operation.id
992                        )));
993                    }
994                    if !is_same_origin(&next, &self.shared.base_url) {
995                        return Err(Error::usage(format!(
996                            "{} redirected off the Fizzy origin to {}",
997                            operation.id,
998                            next.origin().ascii_serialization()
999                        )));
1000                    }
1001                    request = redirected(outgoing, response.status(), &next)?;
1002                    url = next;
1003                    hops += 1;
1004                }
1005            }
1006        }
1007    }
1008
1009    #[allow(clippy::too_many_arguments)]
1010    async fn finish(
1011        &self,
1012        operation: &Operation,
1013        url: &Url,
1014        final_url: Url,
1015        status: StatusCode,
1016        headers: HeaderMap,
1017        body: Bytes,
1018        cached: Option<(String, CachedResponse)>,
1019    ) -> Result<Response, Error> {
1020        // An answer that arrived from somewhere other than where the request went is not
1021        // the document the cache holds under the request's key, and is not stored there.
1022        let cached = if final_url == *url { cached } else { None };
1023
1024        if status == StatusCode::NOT_MODIFIED {
1025            return match cached {
1026                Some((_, entry)) if !entry.etag.is_empty() => Ok(Response {
1027                    status: StatusCode::OK,
1028                    headers,
1029                    body: entry.body,
1030                    url: final_url,
1031                    from_cache: true,
1032                }),
1033                _ => Err(Error::api(
1034                    304,
1035                    "304 received but no cached response available",
1036                )),
1037            };
1038        }
1039
1040        if status.is_success() {
1041            if let (Some((key, _)), Some(cache)) = (cached, self.cacheable(operation))
1042                && let Some(etag) = headers.get("etag").and_then(|value| value.to_str().ok())
1043            {
1044                cache_set(
1045                    cache,
1046                    &key,
1047                    CachedResponse {
1048                        etag: etag.to_string(),
1049                        body: body.clone(),
1050                    },
1051                )
1052                .await;
1053            }
1054            Ok(Response {
1055                status,
1056                headers,
1057                body,
1058                url: final_url,
1059                from_cache: false,
1060            })
1061        } else {
1062            Err(Error::from_response(
1063                status,
1064                &operation.method,
1065                &headers,
1066                &body,
1067            ))
1068        }
1069    }
1070
1071    async fn wait(&self, delay: Duration) {
1072        let jitter = match u64::try_from(self.shared.max_jitter.as_millis()).unwrap_or(u64::MAX) {
1073            0 => Duration::ZERO,
1074            millis => Duration::from_millis(rand::random_range(0..millis)),
1075        };
1076        tokio::time::sleep(delay + jitter).await;
1077    }
1078
1079    fn next_delay(&self, delay: Duration) -> Duration {
1080        (delay * 2).min(self.shared.max_delay)
1081    }
1082}
1083
1084/// An operation the hooks have been told the start of and are still owed the end of. It
1085/// reports the end whichever way the operation leaves: [`Running::finished`] with the
1086/// outcome, or the drop that comes instead when the caller abandons the future.
1087struct Running<'a> {
1088    hooks: &'a Arc<dyn Hooks>,
1089    info: &'a OperationInfo,
1090    state: Option<OperationState>,
1091    started: Instant,
1092}
1093
1094impl Running<'_> {
1095    fn finished(&mut self, outcome: Result<(), &Error>) {
1096        if let Some(state) = self.state.take() {
1097            self.hooks
1098                .on_operation_end(self.info, state, outcome, self.started.elapsed());
1099        }
1100    }
1101}
1102
1103impl Drop for Running<'_> {
1104    fn drop(&mut self) {
1105        if self.state.is_some() {
1106            self.finished(Err(&Error::cancelled()));
1107        }
1108    }
1109}
1110
1111/// One answer from Fizzy, body read: what the retry loop settled on, the URL it came from
1112/// once any redirects were followed, and what the hooks still have to be told about it
1113/// once the body has been dealt with.
1114struct Answered {
1115    url: Url,
1116    status: StatusCode,
1117    headers: HeaderMap,
1118    body: Bytes,
1119    cached: Option<(String, CachedResponse)>,
1120    info: RequestInfo,
1121    duration: Duration,
1122    retryable: bool,
1123    retry_after: Option<u64>,
1124}
1125
1126/// The credential a request carries, for keying the cache: the bearer token or the cookie.
1127fn credential(headers: &HeaderMap) -> Option<String> {
1128    headers
1129        .get(AUTHORIZATION)
1130        .or_else(|| headers.get(COOKIE))
1131        .and_then(|value| value.to_str().ok())
1132        .map(str::to_string)
1133}
1134
1135/// The cache is whatever the caller supplied, and the one the SDK ships keeps its entries in
1136/// files. So every read and write of it goes to the blocking pool: a file read on the
1137/// runtime's own thread stalls every other task sharing that thread. A cache that cannot be
1138/// reached is a miss, which is what any other failure to read it is too.
1139async fn cache_get(cache: &Arc<dyn ResponseCache>, key: &str) -> Option<CachedResponse> {
1140    let cache = cache.clone();
1141    let key = key.to_string();
1142    tokio::task::spawn_blocking(move || cache.get(&key))
1143        .await
1144        .ok()
1145        .flatten()
1146}
1147
1148async fn cache_set(cache: &Arc<dyn ResponseCache>, key: &str, response: CachedResponse) {
1149    let cache = cache.clone();
1150    let key = key.to_string();
1151    let _ = tokio::task::spawn_blocking(move || cache.set(&key, response)).await;
1152}
1153
1154async fn cache_invalidate(cache: &Arc<dyn ResponseCache>, key: &str) {
1155    let cache = cache.clone();
1156    let key = key.to_string();
1157    let _ = tokio::task::spawn_blocking(move || cache.invalidate(&key)).await;
1158}
1159
1160#[cfg(feature = "reqwest")]
1161fn shipped_http_client(timeout: Duration) -> Result<Arc<dyn HttpClient>, Error> {
1162    Ok(Arc::new(crate::http::ReqwestClient::with_timeout(timeout)?))
1163}
1164
1165#[cfg(not(feature = "reqwest"))]
1166fn shipped_http_client(_timeout: Duration) -> Result<Arc<dyn HttpClient>, Error> {
1167    Err(Error::usage(
1168        "no HTTP client: supply one with ClientBuilder::http_client, or enable the reqwest feature",
1169    ))
1170}
1171
1172/// Where a redirect points, when the answer is one and says where. A 3xx without a
1173/// `Location`, or with one that is not a URL, is handed back as the answer it is.
1174fn redirect_target(url: &Url, response: &HttpResponse<Body>) -> Option<Url> {
1175    let status = response.status();
1176    if status.is_redirection() && status != StatusCode::NOT_MODIFIED {
1177        response
1178            .headers()
1179            .get("location")
1180            .and_then(|value| value.to_str().ok())
1181            .and_then(|location| url.join(location).ok())
1182    } else {
1183        None
1184    }
1185}
1186
1187/// The request to send to `next`: the same one, reduced to a GET when the status asks for
1188/// it, and without the cache validator, which belonged to the URL the request left.
1189fn redirected(
1190    (method, mut headers, body): (Method, HeaderMap, Bytes),
1191    status: StatusCode,
1192    next: &Url,
1193) -> Result<Request<Bytes>, Error> {
1194    headers.remove(IF_NONE_MATCH);
1195    let keeps_method = method == Method::GET
1196        || method == Method::HEAD
1197        || status == StatusCode::TEMPORARY_REDIRECT
1198        || status == StatusCode::PERMANENT_REDIRECT;
1199    let (method, body) = if keeps_method {
1200        (method, body)
1201    } else {
1202        headers.remove(CONTENT_TYPE);
1203        headers.remove(CONTENT_LENGTH);
1204        (Method::GET, Bytes::new())
1205    };
1206    let mut request = Request::builder()
1207        .method(method)
1208        .uri(next.as_str())
1209        .body(body)
1210        .map_err(Error::from_std)?;
1211    *request.headers_mut() = headers;
1212    Ok(request)
1213}
1214
1215fn parse_base_url(base_url: &str) -> Result<Url, Error> {
1216    let mut url = Url::parse(base_url)
1217        .map_err(|error| Error::usage(format!("base URL {base_url}: {error}")))?;
1218    require_secure_endpoint(&url)?;
1219    if !url.username().is_empty() || url.password().is_some() {
1220        return Err(Error::usage(
1221            "base URL must not carry credentials; use an access token or a session token",
1222        ));
1223    }
1224    if !url.path().ends_with('/') {
1225        url.set_path(&format!("{}/", url.path()));
1226    }
1227    Ok(url)
1228}
1229
1230/// The wait Fizzy asked for, on the two statuses that carry one.
1231fn retry_after_asked(status: StatusCode, headers: &HeaderMap) -> Option<u64> {
1232    if status == StatusCode::TOO_MANY_REQUESTS || status == StatusCode::SERVICE_UNAVAILABLE {
1233        retry_after_seconds(headers)
1234    } else {
1235        None
1236    }
1237}
1238
1239fn header_value(value: &str) -> Result<HeaderValue, Error> {
1240    HeaderValue::from_str(value)
1241        .map_err(|_| Error::usage(format!("{value:?} is not a valid header value")))
1242}
1243
1244/// Reads a body up to the bound and refuses it on the first byte past. A body exactly at
1245/// the bound reads whole; one declared past it never starts.
1246/// The error for a body that could not be read, and whether the SDK would ask for the
1247/// answer again given an attempt to spare: a success whose body broke off, it would; a
1248/// refusal it would not, and that keeps its status over the reason its body was lost.
1249fn unread(
1250    operation: &Operation,
1251    status: StatusCode,
1252    headers: &HeaderMap,
1253    error: Error,
1254) -> (Error, bool) {
1255    let retryable = status.is_success() && error.is_retryable();
1256    let error = if status.is_success() {
1257        error
1258    } else {
1259        Error::from_response(status, &operation.method, headers, &[]).refusing(error)
1260    };
1261    (error, retryable)
1262}
1263
1264/// Where the retry loop stands: which attempt is next, and how long the wait before it is.
1265struct Backoff {
1266    attempt: u32,
1267    delay: Duration,
1268}
1269
1270pub(crate) async fn read_body(
1271    body: Body,
1272    limit: usize,
1273    method: &Method,
1274    path: &str,
1275) -> Result<Bytes, Error> {
1276    body.collect(limit, || Error::response_too_large(limit, method, path))
1277        .await
1278}
1279
1280#[cfg(test)]
1281#[allow(clippy::unwrap_used)]
1282mod tests {
1283    use std::sync::Mutex;
1284
1285    use async_trait::async_trait;
1286    use serde_json::Value;
1287
1288    use super::*;
1289    use crate::auth::StaticTokenProvider;
1290
1291    /// An [`HttpClient`] with no network behind it: it answers each request from a closure
1292    /// and keeps what it was sent. This is the second implementation the trait exists for,
1293    /// so the client is exercised here with no `reqwest` in the picture.
1294    struct Canned {
1295        answer: Box<Answer>,
1296        sent: Mutex<Vec<(Method, String, HeaderMap)>>,
1297    }
1298
1299    type Answer = dyn Fn(&Request<Bytes>) -> HttpResponse<Body> + Send + Sync;
1300
1301    impl Canned {
1302        fn new(
1303            answer: impl Fn(&Request<Bytes>) -> HttpResponse<Body> + Send + Sync + 'static,
1304        ) -> Arc<Canned> {
1305            Arc::new(Canned {
1306                answer: Box::new(answer),
1307                sent: Mutex::new(Vec::new()),
1308            })
1309        }
1310
1311        fn sent(&self) -> Vec<(Method, String, HeaderMap)> {
1312            self.sent.lock().unwrap().clone()
1313        }
1314    }
1315
1316    #[async_trait]
1317    impl HttpClient for Arc<Canned> {
1318        async fn send(&self, request: Request<Bytes>) -> Result<HttpResponse<Body>, Error> {
1319            self.sent.lock().unwrap().push((
1320                request.method().clone(),
1321                request.uri().to_string(),
1322                request.headers().clone(),
1323            ));
1324            Ok((self.answer)(&request))
1325        }
1326    }
1327
1328    fn answer(status: u16, body: &'static str) -> HttpResponse<Body> {
1329        let mut response = HttpResponse::new(Body::from(body));
1330        *response.status_mut() = StatusCode::from_u16(status).unwrap();
1331        response
1332    }
1333
1334    fn redirect(location: &str) -> HttpResponse<Body> {
1335        let mut response = answer(302, "");
1336        response
1337            .headers_mut()
1338            .insert("location", HeaderValue::from_str(location).unwrap());
1339        response
1340    }
1341
1342    fn client_over(http: Arc<Canned>) -> Client {
1343        Client::builder(Config::default().with_base_url("https://fizzy.test"))
1344            .token_provider(StaticTokenProvider::new("secret"))
1345            .http_client(http)
1346            .max_attempts(1)
1347            .build()
1348            .unwrap()
1349    }
1350
1351    #[tokio::test]
1352    async fn a_request_goes_out_on_the_supplied_http_client_with_credentials() {
1353        let http = Canned::new(|_| answer(200, r#"{"ok":true}"#));
1354        let client = client_over(http.clone());
1355
1356        let body: Value = client
1357            .send(client.request(Method::GET, "/my/identity.json"))
1358            .await
1359            .unwrap();
1360
1361        assert_eq!(body, serde_json::json!({ "ok": true }));
1362        let sent = http.sent();
1363        assert_eq!(sent.len(), 1);
1364        assert_eq!(sent[0].0, Method::GET);
1365        assert_eq!(sent[0].1, "https://fizzy.test/my/identity.json");
1366        assert_eq!(sent[0].2[AUTHORIZATION], "Bearer secret");
1367    }
1368
1369    #[tokio::test]
1370    async fn a_redirect_on_the_same_origin_is_followed_with_credentials() {
1371        let http = Canned::new(|request| {
1372            if request.uri().path() == "/old.json" {
1373                redirect("/new.json")
1374            } else {
1375                answer(200, r#"{"moved":true}"#)
1376            }
1377        });
1378        let client = client_over(http.clone());
1379
1380        let response = client
1381            .execute(client.request(Method::GET, "/old.json"))
1382            .await
1383            .unwrap();
1384
1385        assert_eq!(response.url.as_str(), "https://fizzy.test/new.json");
1386        assert_eq!(response.body, r#"{"moved":true}"#);
1387        let sent = http.sent();
1388        assert_eq!(sent.len(), 2);
1389        assert_eq!(sent[1].2[AUTHORIZATION], "Bearer secret");
1390    }
1391
1392    #[tokio::test]
1393    async fn a_redirect_off_the_origin_is_refused_with_nothing_sent_there() {
1394        let http = Canned::new(|_| redirect("https://storage.test/blobs/1"));
1395        let client = client_over(http.clone());
1396
1397        let error = client.get("/blobs/1").await.unwrap_err();
1398
1399        assert_eq!(error.code(), ErrorCode::Usage);
1400        assert_eq!(http.sent().len(), 1);
1401    }
1402
1403    fn broken_body(status: u16) -> HttpResponse<Body> {
1404        let chunks = futures_util::stream::iter([
1405            Ok(Bytes::from_static(b"{\"ok\":")),
1406            Err(Error::new(ErrorCode::Network, "cut off").retryable()),
1407        ]);
1408        let mut response = HttpResponse::new(Body::from_stream(chunks, None));
1409        *response.status_mut() = StatusCode::from_u16(status).unwrap();
1410        response
1411    }
1412
1413    #[tokio::test]
1414    async fn a_body_that_breaks_off_is_asked_for_again() {
1415        let calls = Arc::new(Mutex::new(0));
1416        let seen = calls.clone();
1417        let http = Canned::new(move |_| {
1418            let mut calls = seen.lock().unwrap();
1419            *calls += 1;
1420            if *calls == 1 {
1421                broken_body(200)
1422            } else {
1423                answer(200, r#"{"ok":true}"#)
1424            }
1425        });
1426        let client = Client::builder(Config::default().with_base_url("https://fizzy.test"))
1427            .token_provider(StaticTokenProvider::new("secret"))
1428            .http_client(http.clone())
1429            .max_attempts(2)
1430            .max_jitter(Duration::ZERO)
1431            .base_delay(Duration::from_millis(1))
1432            .build()
1433            .unwrap();
1434
1435        let response = client.get("/x.json").await.unwrap();
1436
1437        assert_eq!(response.body, r#"{"ok":true}"#);
1438        assert_eq!(http.sent().len(), 2);
1439    }
1440
1441    #[tokio::test]
1442    async fn a_failure_whose_body_breaks_off_keeps_its_status_and_is_not_resent() {
1443        let http = Canned::new(|_| broken_body(422));
1444        let client = Client::builder(Config::default().with_base_url("https://fizzy.test"))
1445            .token_provider(StaticTokenProvider::new("secret"))
1446            .http_client(http.clone())
1447            .max_attempts(2)
1448            .max_jitter(Duration::ZERO)
1449            .base_delay(Duration::from_millis(1))
1450            .build()
1451            .unwrap();
1452
1453        let error = client.get("/x.json").await.unwrap_err();
1454
1455        assert_eq!(error.code(), ErrorCode::Validation);
1456        assert_eq!(error.http_status(), Some(422));
1457        assert_eq!(http.sent().len(), 1);
1458    }
1459
1460    /// Keeps what each request was reported as, resendable or not.
1461    #[derive(Default)]
1462    struct Retryability(Mutex<Vec<bool>>);
1463
1464    impl Hooks for Retryability {
1465        fn on_request_end(&self, _info: &RequestInfo, result: &RequestResult<'_>) {
1466            self.0.lock().unwrap().push(result.retryable);
1467        }
1468    }
1469
1470    #[tokio::test]
1471    async fn a_body_that_breaks_off_on_the_last_attempt_is_still_reported_resendable() {
1472        let http = Canned::new(|_| broken_body(200));
1473        let seen = Arc::new(Retryability::default());
1474        let client = Client::builder(Config::default().with_base_url("https://fizzy.test"))
1475            .token_provider(StaticTokenProvider::new("secret"))
1476            .http_client(http.clone())
1477            .hooks(seen.clone())
1478            .max_attempts(2)
1479            .max_jitter(Duration::ZERO)
1480            .base_delay(Duration::from_millis(1))
1481            .build()
1482            .unwrap();
1483
1484        let error = client.get("/x.json").await.unwrap_err();
1485
1486        assert_eq!(error.code(), ErrorCode::Network);
1487        assert_eq!(http.sent().len(), 2);
1488        assert_eq!(*seen.0.lock().unwrap(), [true, true]);
1489    }
1490
1491    #[tokio::test]
1492    async fn a_redirect_carrying_credentials_is_refused_without_echoing_them() {
1493        let http = Canned::new(|_| redirect("https://user:s3cret@fizzy.test/new.json"));
1494        let client = client_over(http.clone());
1495
1496        let error = client.get("/old.json").await.unwrap_err();
1497
1498        assert_eq!(error.code(), ErrorCode::Usage);
1499        assert!(!error.to_string().contains("s3cret"));
1500        assert_eq!(http.sent().len(), 1);
1501    }
1502
1503    #[tokio::test]
1504    async fn a_path_that_resolves_off_the_origin_is_refused_before_anything_is_sent() {
1505        let http = Canned::new(|_| answer(200, "{}"));
1506        let client = client_over(http.clone());
1507
1508        for path in [
1509            "http://evil.test/x",
1510            "HTTP://evil.test/x",
1511            "https://evil.test/x",
1512        ] {
1513            let error = client.get(path).await.unwrap_err();
1514            assert_eq!(error.code(), ErrorCode::Usage, "{path}");
1515            let error = client
1516                .execute(client.request(Method::GET, path))
1517                .await
1518                .unwrap_err();
1519            assert_eq!(error.code(), ErrorCode::Usage, "{path}");
1520        }
1521        assert!(http.sent().is_empty());
1522    }
1523
1524    #[tokio::test]
1525    async fn a_url_carrying_userinfo_is_refused_without_echoing_it() {
1526        let http = Canned::new(|_| answer(200, "{}"));
1527        let client = client_over(http.clone());
1528
1529        let error = client
1530            .get("https://user:s3cret@fizzy.test/x")
1531            .await
1532            .unwrap_err();
1533
1534        assert_eq!(error.code(), ErrorCode::Usage);
1535        assert!(!error.to_string().contains("s3cret"));
1536        assert!(http.sent().is_empty());
1537    }
1538
1539    #[tokio::test]
1540    async fn a_refused_redirect_is_not_retried() {
1541        let http = Canned::new(|_| redirect("http://evil.test/"));
1542        let client = Client::builder(Config::default().with_base_url("https://fizzy.test"))
1543            .token_provider(StaticTokenProvider::new("secret"))
1544            .http_client(http.clone())
1545            .max_jitter(Duration::ZERO)
1546            .base_delay(Duration::from_millis(1))
1547            .build()
1548            .unwrap();
1549
1550        let error = client.get("/anything").await.unwrap_err();
1551
1552        assert_eq!(error.code(), ErrorCode::Usage);
1553        assert_eq!(http.sent().len(), 1);
1554    }
1555
1556    #[test]
1557    fn a_response_and_an_operation_print_without_their_secrets() {
1558        let mut headers = HeaderMap::new();
1559        headers.insert(
1560            "set-cookie",
1561            HeaderValue::from_static("session_token=s3cret; HttpOnly"),
1562        );
1563        let response = Response {
1564            status: StatusCode::OK,
1565            headers,
1566            body: Bytes::from_static(br#"{"email_address":"jane@example.com"}"#),
1567            url: Url::parse("https://fizzy.test/x").unwrap(),
1568            from_cache: false,
1569        };
1570        let printed = format!("{response:?}");
1571        assert!(!printed.contains("s3cret"));
1572        assert!(!printed.contains("jane@example.com"));
1573        assert!(printed.contains("[REDACTED]"));
1574
1575        let mut operation = Operation::raw(Method::POST, "/session.json".into());
1576        operation
1577            .json(&serde_json::json!({"email_address": "jane@example.com"}))
1578            .unwrap();
1579        let printed = format!("{operation:?}");
1580        assert!(!printed.contains("jane@example.com"));
1581        assert!(printed.contains("len"));
1582    }
1583
1584    #[tokio::test]
1585    async fn a_redirect_to_plain_http_elsewhere_is_refused() {
1586        let http = Canned::new(|_| redirect("http://evil.test/"));
1587        let client = client_over(http.clone());
1588
1589        let error = client.get("/anything").await.unwrap_err();
1590
1591        assert_eq!(error.code(), ErrorCode::Usage);
1592        assert_eq!(http.sent().len(), 1);
1593    }
1594
1595    #[tokio::test]
1596    async fn a_redirect_loop_is_given_up_on() {
1597        let http = Canned::new(|_| redirect("/again"));
1598        let client = client_over(http.clone());
1599
1600        let error = client.get("/again").await.unwrap_err();
1601
1602        assert_eq!(error.code(), ErrorCode::Network);
1603        assert_eq!(http.sent().len(), MAX_REDIRECTS + 1);
1604    }
1605
1606    #[test]
1607    fn base_url_must_be_https_or_local() {
1608        assert!(parse_base_url("https://fizzy.do").is_ok());
1609        assert!(parse_base_url("http://127.0.0.1:3000").is_ok());
1610        assert_eq!(
1611            parse_base_url("http://evil.example.com")
1612                .unwrap_err()
1613                .code(),
1614            ErrorCode::Usage
1615        );
1616        assert_eq!(
1617            parse_base_url("https://user:secret@fizzy.do")
1618                .unwrap_err()
1619                .code(),
1620            ErrorCode::Usage
1621        );
1622    }
1623
1624    #[test]
1625    fn an_account_id_has_to_fit_in_a_path() {
1626        let http = Canned::new(|_| answer(200, "[]"));
1627        let client = client_over(http);
1628        assert!(client.for_account("999").is_ok());
1629        assert!(client.for_account("").is_err());
1630        assert!(client.for_account("a/b").is_err());
1631    }
1632}