omni_dev/gmail/
client.rs

1//! Gmail REST API client.
2//!
3//! Thin `reqwest` wrapper that attaches a Bearer access token (refreshed by
4//! an owned [`GmailSession`]) to every request, retries HTTP 429 via the
5//! shared [`retry_429`](crate::utils::http::retry_429) driver, and retries
6//! exactly once on HTTP 401 by forcing a session refresh. Modelled on
7//! [`crate::datadog::client::DatadogClient`]; the difference is Bearer-token
8//! auth with in-process refresh instead of two static API keys.
9
10use std::sync::{Arc, PoisonError, RwLock};
11
12use anyhow::{Context, Result};
13use reqwest::{Client, Response};
14use url::Url;
15
16use crate::gmail::auth::{GmailCredentials, GmailSession};
17use crate::gmail::error::GmailError;
18use crate::request_log;
19use crate::utils::env::{EnvSource, SystemEnv};
20use crate::utils::http::{connect_timeout, read_timeout, retry_if, RetryNotifyFn};
21
22/// HTTP client for the Gmail v1 REST API.
23pub struct GmailClient {
24    client: Client,
25    base_url: String,
26    session: GmailSession,
27    /// Set via [`Self::set_retry_notify`] to route rate-limit retry notices
28    /// to a caller's progress display (#1651) instead of `retry_if`'s
29    /// default `eprintln!`. An `RwLock` (rather than a plain field) because
30    /// a caller may only hold `&GmailClient` by the time it has a progress
31    /// channel to attach — see `sync.rs::run_sync_command`.
32    retry_notify: RwLock<Option<Arc<RetryNotifyFn>>>,
33}
34
35impl std::fmt::Debug for GmailClient {
36    // Hand-written, not derived: omits `session` entirely rather than
37    // relying on every nested `Secret` staying wrapped — the safest
38    // possible redaction is "not mentioned at all."
39    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
40        f.debug_struct("GmailClient")
41            .field("base_url", &self.base_url)
42            .finish_non_exhaustive()
43    }
44}
45
46impl GmailClient {
47    /// The real Gmail API host. Unlike Datadog, there is no per-tenant
48    /// site/region this is *derived* from — [`Self::DEFAULT_BASE_URL`] is
49    /// the one real host, overridable wholesale via `GMAIL_API_URL`
50    /// (`crate::gmail::auth::GMAIL_API_URL`; see
51    /// [`Self::from_credentials_with`]) rather than site-substituted like
52    /// Datadog's `DATADOG_API_URL`. [`Self::new`]'s `base_url` parameter is
53    /// the lower-level seam both the override and tests go through.
54    const DEFAULT_BASE_URL: &'static str = "https://gmail.googleapis.com";
55
56    /// Builds a client against `base_url` with already-loaded credentials.
57    ///
58    /// For production use, construct via [`Self::from_credentials`]; tests
59    /// pass a wiremock URL directly.
60    pub fn new(base_url: &str, credentials: &GmailCredentials) -> Result<Self> {
61        let client = Client::builder()
62            .connect_timeout(connect_timeout())
63            .read_timeout(read_timeout())
64            .build()
65            .context("Failed to build HTTP client")?;
66        let session = GmailSession::new(client.clone(), credentials);
67        Ok(Self {
68            client,
69            base_url: base_url.trim_end_matches('/').to_string(),
70            session,
71            retry_notify: RwLock::new(None),
72        })
73    }
74
75    /// Registers `notify` to receive rate-limit retry notices (#1651)
76    /// instead of the shared retry driver's default `eprintln!`.
77    ///
78    /// Takes `&self` (writing through the interior `RwLock`) rather than
79    /// `&mut self` so it can be attached after construction through a
80    /// shared reference — the account label/progress channel a caller
81    /// wants to attribute the notice with often isn't available until
82    /// after the client already exists.
83    pub(crate) fn set_retry_notify(&self, notify: Arc<RetryNotifyFn>) {
84        *self
85            .retry_notify
86            .write()
87            .unwrap_or_else(PoisonError::into_inner) = Some(notify);
88    }
89
90    /// Creates a client from stored credentials against the real Gmail API
91    /// host.
92    ///
93    /// Respects `GMAIL_API_URL` as an optional override: when set (and
94    /// non-empty) in the process environment it replaces
95    /// [`Self::DEFAULT_BASE_URL`] wholesale. Added per PR #1466 review —
96    /// without it, exercising the CLI's output shapes required a real
97    /// Google Cloud project, and there was no way to route through a forced
98    /// egress proxy.
99    pub fn from_credentials(credentials: &GmailCredentials) -> Result<Self> {
100        Self::from_credentials_with(&SystemEnv, credentials)
101    }
102
103    /// [`from_credentials`](Self::from_credentials) over an injected
104    /// [`EnvSource`], so tests can exercise the `GMAIL_API_URL` override via
105    /// `MapEnv` without mutating the process environment.
106    pub(crate) fn from_credentials_with(
107        env: &impl EnvSource,
108        credentials: &GmailCredentials,
109    ) -> Result<Self> {
110        let base_url = env
111            .var(crate::gmail::auth::GMAIL_API_URL)
112            .filter(|s| !s.is_empty())
113            .unwrap_or_else(|| Self::DEFAULT_BASE_URL.to_string());
114        Self::new(&base_url, credentials)
115    }
116
117    /// Returns the API base URL (without trailing slash).
118    #[must_use]
119    pub fn base_url(&self) -> &str {
120        &self.base_url
121    }
122
123    /// Builds an absolute API URL by joining `path` onto `base_url`.
124    ///
125    /// Takes `base_url` (rather than `&self`) so the free `build_*_url`
126    /// functions in the API façade modules — and their unit tests, which
127    /// pass literal base URLs — can call it without an instance.
128    pub(crate) fn api_url(base_url: &str, path: &str) -> Result<Url> {
129        Url::parse(&format!("{base_url}{path}")).context("Invalid Gmail base URL")
130    }
131
132    /// Checks `response` for success and deserialises its JSON body into `T`.
133    pub(crate) async fn parse_response<T: serde::de::DeserializeOwned>(
134        &self,
135        response: Response,
136        context: &'static str,
137    ) -> Result<T> {
138        if !response.status().is_success() {
139            return Err(Self::response_to_error(response).await.into());
140        }
141        response.json().await.context(context)
142    }
143
144    /// Sends an authenticated GET and deserialises the JSON body into `T`.
145    pub(crate) async fn get_parsed<T: serde::de::DeserializeOwned>(
146        &self,
147        url: &str,
148        context: &'static str,
149    ) -> Result<T> {
150        let response = self.get_json(url).await?;
151        self.parse_response(response, context).await
152    }
153
154    /// Sends an authenticated GET request and returns the raw response.
155    ///
156    /// Retries exactly once on HTTP 401 by forcing a session refresh — see
157    /// [`Self::send_authorized`] for why both a proactive and a reactive
158    /// refresh path exist.
159    pub async fn get_json(&self, url: &str) -> Result<Response> {
160        self.send_authorized(url, "GET", |client, token| {
161            client
162                .get(url)
163                .bearer_auth(token)
164                .header("Accept", "application/json")
165        })
166        .await
167    }
168
169    /// Sends an authenticated POST request with a JSON body and returns the
170    /// raw response.
171    pub async fn post_json<T: serde::Serialize + Sync + ?Sized>(
172        &self,
173        url: &str,
174        body: &T,
175    ) -> Result<Response> {
176        self.send_authorized(url, "POST", |client, token| {
177            client
178                .post(url)
179                .bearer_auth(token)
180                .header("Content-Type", "application/json")
181                .json(body)
182        })
183        .await
184    }
185
186    /// Sends an authenticated POST request with a raw byte body and returns
187    /// the raw response — `messages.insert`'s `uploadType=multipart` media
188    /// endpoint (`crate::gmail::messages_api::MessagesApi::insert`), which
189    /// needs a hand-assembled `multipart/related` body rather than a JSON
190    /// one. Mirrors [`crate::drive::client::DriveClient::post_bytes`].
191    ///
192    /// The closure passed to [`Self::send_authorized`] captures `body` by
193    /// reference and calls `.to_vec()` inside itself, so a retried or
194    /// 401-refreshed attempt re-materialises the body rather than sending a
195    /// second, empty request against an already-consumed `reqwest::Body` —
196    /// see [`Self::send_authorized`]'s doc comment for why a retry rebuilds
197    /// the request from scratch.
198    pub(crate) async fn post_bytes(
199        &self,
200        url: &str,
201        body: &[u8],
202        content_type: &str,
203    ) -> Result<Response> {
204        self.send_authorized(url, "POST", |client, token| {
205            client
206                .post(url)
207                .bearer_auth(token)
208                .header("Content-Type", content_type)
209                .body(body.to_vec())
210        })
211        .await
212    }
213
214    /// [`Self::post_bytes`] with `PUT`: `drafts.update`'s
215    /// `uploadType=multipart` media endpoint
216    /// (`crate::gmail::drafts_api::DraftsApi::update`), which replaces the
217    /// whole draft. The body is re-materialised per attempt for the same
218    /// reason.
219    pub(crate) async fn put_bytes(
220        &self,
221        url: &str,
222        body: &[u8],
223        content_type: &str,
224    ) -> Result<Response> {
225        self.send_authorized(url, "PUT", |client, token| {
226            client
227                .put(url)
228                .bearer_auth(token)
229                .header("Content-Type", content_type)
230                .body(body.to_vec())
231        })
232        .await
233    }
234
235    /// Sends a request built by `build`, retrying exactly once on HTTP 401.
236    ///
237    /// [`GmailSession::access_token`] already refreshes proactively when the
238    /// tracked expiry is near — this reactive path exists for what
239    /// proactive tracking can't see: clock skew against Google's clock, or
240    /// the token being invalidated server-side mid-run (revoked access). A
241    /// second 401 after the retry is authoritative: either the refresh
242    /// produced a token that was also rejected, or another caller's
243    /// already-current token was reused and still rejected — either way the
244    /// problem isn't staleness, so it surfaces as an ordinary
245    /// `ApiRequestFailed` rather than retrying again.
246    async fn send_authorized<F>(
247        &self,
248        url: &str,
249        method: &'static str,
250        build: F,
251    ) -> Result<Response>
252    where
253        F: Fn(&Client, &str) -> reqwest::RequestBuilder + Send + Sync,
254    {
255        let token = self
256            .session
257            .access_token()
258            .await
259            .context("Failed to obtain a Gmail access token")?;
260        let response = self
261            .send_once(url, method, &build, token.expose_secret())
262            .await?;
263        if response.status().as_u16() != 401 {
264            return Ok(response);
265        }
266        let refreshed = self
267            .session
268            .force_refresh(&token)
269            .await
270            .context("Failed to refresh the Gmail access token after a 401")?;
271        self.send_once(url, method, &build, refreshed.expose_secret())
272            .await
273    }
274
275    async fn send_once<F>(
276        &self,
277        url: &str,
278        method: &'static str,
279        build: &F,
280        token: &str,
281    ) -> Result<Response>
282    where
283        F: Fn(&Client, &str) -> reqwest::RequestBuilder + Send + Sync,
284    {
285        let notify = self
286            .retry_notify
287            .read()
288            .unwrap_or_else(PoisonError::into_inner)
289            .clone();
290        retry_if(
291            || build(&self.client, token),
292            |started, result| {
293                request_log::record_http_result("gmail", method, url, started, result);
294            },
295            |status, body| status == 429 || is_gmail_quota_exceeded(status, body),
296            notify.as_deref(),
297        )
298        .await
299        .with_context(|| format!("Failed to send {method} request to Gmail API"))
300    }
301
302    /// Consumes a non-success response into a [`GmailError`].
303    ///
304    /// Parses Gmail's `{"error":{"message":...,"errors":[{"reason":...}]}}`
305    /// envelope into a human message when present (falls back to the raw
306    /// body otherwise). Gmail signals quota exhaustion as **403**
307    /// `rateLimitExceeded`/`userRateLimitExceeded`, not `429` — unlike plain
308    /// 429s, that shape now also drives a retry (see [`is_gmail_quota_exceeded`]
309    /// via [`retry_if`](crate::utils::http::retry_if)), so this only sees the
310    /// error once retries are exhausted (or the reason didn't match).
311    pub async fn response_to_error(response: Response) -> GmailError {
312        let status = response.status().as_u16();
313        let raw = response.text().await.unwrap_or_default();
314        let value = serde_json::from_str::<serde_json::Value>(&raw).ok();
315        let reason = value.as_ref().and_then(gmail_error_reason);
316        let body = value
317            .as_ref()
318            .and_then(gmail_error_message)
319            .map(|message| match &reason {
320                Some(r) => format!("{message} (reason: {r})"),
321                None => message,
322            })
323            .unwrap_or(raw);
324        GmailError::ApiRequestFailed {
325            status,
326            body,
327            reason,
328        }
329    }
330}
331
332/// Extracts the `error.errors[0].reason` field from Gmail's already-parsed
333/// JSON error envelope, if present.
334fn gmail_error_reason(value: &serde_json::Value) -> Option<String> {
335    value
336        .get("error")
337        .and_then(|e| e.get("errors"))
338        .and_then(|e| e.as_array())
339        .and_then(|a| a.first())
340        .and_then(|e| e.get("reason"))
341        .and_then(|r| r.as_str())
342        .map(str::to_string)
343}
344
345/// Extracts the `error.message` field from Gmail's already-parsed JSON
346/// error envelope, if present.
347fn gmail_error_message(value: &serde_json::Value) -> Option<String> {
348    value
349        .get("error")?
350        .get("message")?
351        .as_str()
352        .map(str::to_string)
353}
354
355/// Whether a response is Gmail's quota-exhaustion signal — **403** with
356/// `reason` of `rateLimitExceeded` or `userRateLimitExceeded` specifically,
357/// not any 403 with a `reason` field: e.g. `insufficientPermissions` is also
358/// a 403 and must never be retried (retrying a scope/permission error just
359/// wastes the backoff window before failing anyway).
360fn is_gmail_quota_exceeded(status: u16, body: &[u8]) -> bool {
361    if status != 403 {
362        return false;
363    }
364    let Ok(text) = std::str::from_utf8(body) else {
365        return false;
366    };
367    let Ok(value) = serde_json::from_str::<serde_json::Value>(text) else {
368        return false;
369    };
370    matches!(
371        gmail_error_reason(&value).as_deref(),
372        Some("rateLimitExceeded" | "userRateLimitExceeded")
373    )
374}
375
376/// Test-only seam letting sibling API-façade test modules (which can't
377/// reach `GmailClient`'s private fields directly, unlike this module's own
378/// `tests` submodule) bootstrap a deterministic access token via wiremock.
379#[cfg(test)]
380pub(crate) mod test_support {
381    use super::GmailClient;
382    use crate::gmail::auth::{GmailCredentials, GmailSession};
383
384    /// Replaces `client`'s session with one pointed at an explicit token
385    /// endpoint.
386    pub(crate) fn replace_session(
387        client: &mut GmailClient,
388        credentials: &GmailCredentials,
389        token_endpoint: &str,
390    ) {
391        client.session = GmailSession::new_with_token_endpoint(
392            client.client.clone(),
393            credentials,
394            token_endpoint,
395        );
396    }
397}
398
399#[cfg(test)]
400#[allow(clippy::unwrap_used, clippy::expect_used)]
401mod tests {
402    use super::*;
403    use crate::gmail::auth::GmailScope;
404    use crate::utils::secret::Secret;
405
406    fn test_credentials() -> GmailCredentials {
407        GmailCredentials {
408            client_id: "client-1".to_string(),
409            client_secret: Secret::new("secret-1"),
410            refresh_token: Secret::new("refresh-1"),
411            scope: GmailScope::ReadOnly,
412        }
413    }
414
415    #[test]
416    fn new_client_strips_trailing_slash() {
417        let client =
418            GmailClient::new("https://gmail.googleapis.com/", &test_credentials()).unwrap();
419        assert_eq!(client.base_url(), "https://gmail.googleapis.com");
420    }
421
422    #[test]
423    fn new_client_preserves_clean_url() {
424        let client = GmailClient::new("https://gmail.googleapis.com", &test_credentials()).unwrap();
425        assert_eq!(client.base_url(), "https://gmail.googleapis.com");
426    }
427
428    #[test]
429    fn from_credentials_uses_gmail_api_host() {
430        // Via a fresh MapEnv, not from_credentials()'s real SystemEnv — a
431        // stray GMAIL_API_URL in the actual process environment must not
432        // make this test flaky (mirrors the Datadog precedent,
433        // from_credentials_builds_base_url_from_site).
434        let env = crate::test_support::env::MapEnv::new();
435        let client = GmailClient::from_credentials_with(&env, &test_credentials()).unwrap();
436        assert_eq!(client.base_url(), "https://gmail.googleapis.com");
437    }
438
439    #[test]
440    fn from_credentials_honours_api_url_override() {
441        let env = crate::test_support::env::MapEnv::new().with(
442            crate::gmail::auth::GMAIL_API_URL,
443            "http://proxy.example:8080",
444        );
445        let client = GmailClient::from_credentials_with(&env, &test_credentials()).unwrap();
446        assert_eq!(client.base_url(), "http://proxy.example:8080");
447    }
448
449    #[test]
450    fn from_credentials_ignores_empty_api_url_override() {
451        let env =
452            crate::test_support::env::MapEnv::new().with(crate::gmail::auth::GMAIL_API_URL, "");
453        let client = GmailClient::from_credentials_with(&env, &test_credentials()).unwrap();
454        assert_eq!(client.base_url(), "https://gmail.googleapis.com");
455    }
456
457    #[test]
458    fn client_debug_never_mentions_session_field() {
459        let client = GmailClient::new("https://gmail.googleapis.com", &test_credentials()).unwrap();
460        let debug = format!("{client:?}");
461        assert!(!debug.contains("secret-1"));
462        assert!(!debug.contains("refresh-1"));
463        assert!(!debug.contains("session"));
464        assert!(debug.contains("GmailClient"));
465    }
466
467    /// Mounts a bootstrap token-endpoint mock at the same base URL as the
468    /// Gmail API mock — `GmailSession` doesn't distinguish the two hosts in
469    /// these tests, so pointing the token endpoint at the wiremock server
470    /// too keeps the setup to one server per test.
471    async fn client_with_bootstrapped_token(server: &wiremock::MockServer) -> GmailClient {
472        // `up_to_n_times(1)` + `with_priority(1)` so a test's own follow-up
473        // POST /token mock (registered at `with_priority(2)`, matched only
474        // once this one is exhausted) can simulate a second, distinct
475        // refresh without either mock racing the other for every request.
476        wiremock::Mock::given(wiremock::matchers::method("POST"))
477            .and(wiremock::matchers::path("/token"))
478            .respond_with(
479                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
480                    "access_token": "bootstrap-token",
481                    "expires_in": 3600,
482                })),
483            )
484            .up_to_n_times(1)
485            .with_priority(1)
486            .mount(server)
487            .await;
488
489        let mut client = GmailClient::new(&server.uri(), &test_credentials()).unwrap();
490        client.session = GmailSession::new_with_token_endpoint(
491            client.client.clone(),
492            &test_credentials(),
493            &format!("{}/token", server.uri()),
494        );
495        client
496    }
497
498    #[tokio::test]
499    async fn get_json_sends_bearer_auth_header() {
500        let server = wiremock::MockServer::start().await;
501        let client = client_with_bootstrapped_token(&server).await;
502        wiremock::Mock::given(wiremock::matchers::method("GET"))
503            .and(wiremock::matchers::path("/test"))
504            .and(wiremock::matchers::header(
505                "Authorization",
506                "Bearer bootstrap-token",
507            ))
508            .respond_with(
509                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({"ok": true})),
510            )
511            .expect(1)
512            .mount(&server)
513            .await;
514
515        let resp = client
516            .get_json(&format!("{}/test", server.uri()))
517            .await
518            .unwrap();
519        assert!(resp.status().is_success());
520    }
521
522    #[tokio::test]
523    async fn post_json_sends_body_and_bearer_auth() {
524        let server = wiremock::MockServer::start().await;
525        let client = client_with_bootstrapped_token(&server).await;
526        wiremock::Mock::given(wiremock::matchers::method("POST"))
527            .and(wiremock::matchers::path("/test"))
528            .and(wiremock::matchers::header(
529                "Authorization",
530                "Bearer bootstrap-token",
531            ))
532            .and(wiremock::matchers::body_json(serde_json::json!({"k": "v"})))
533            .respond_with(wiremock::ResponseTemplate::new(200))
534            .expect(1)
535            .mount(&server)
536            .await;
537
538        let resp = client
539            .post_json(
540                &format!("{}/test", server.uri()),
541                &serde_json::json!({"k": "v"}),
542            )
543            .await
544            .unwrap();
545        assert!(resp.status().is_success());
546    }
547
548    // ── post_bytes ───────────────────────────────────────────────────
549
550    #[tokio::test]
551    async fn post_bytes_sends_body_and_content_type_and_bearer_auth() {
552        let server = wiremock::MockServer::start().await;
553        let client = client_with_bootstrapped_token(&server).await;
554        wiremock::Mock::given(wiremock::matchers::method("POST"))
555            .and(wiremock::matchers::path("/test"))
556            .and(wiremock::matchers::header(
557                "Authorization",
558                "Bearer bootstrap-token",
559            ))
560            .and(wiremock::matchers::header(
561                "Content-Type",
562                "multipart/related; boundary=B",
563            ))
564            .and(wiremock::matchers::body_bytes(b"raw-body".to_vec()))
565            .respond_with(wiremock::ResponseTemplate::new(200))
566            .expect(1)
567            .mount(&server)
568            .await;
569
570        let resp = client
571            .post_bytes(
572                &format!("{}/test", server.uri()),
573                b"raw-body",
574                "multipart/related; boundary=B",
575            )
576            .await
577            .unwrap();
578        assert!(resp.status().is_success());
579    }
580
581    #[tokio::test]
582    async fn post_bytes_replays_the_full_body_across_a_429_retry() {
583        let server = wiremock::MockServer::start().await;
584        let client = client_with_bootstrapped_token(&server).await;
585        // Both the throttled first attempt and the successful retry must
586        // see the same non-empty body — a consumed-body regression would
587        // silently send an empty second request instead.
588        wiremock::Mock::given(wiremock::matchers::method("POST"))
589            .and(wiremock::matchers::path("/test"))
590            .and(wiremock::matchers::body_bytes(b"raw-body".to_vec()))
591            .respond_with(wiremock::ResponseTemplate::new(429).append_header("Retry-After", "0"))
592            .up_to_n_times(1)
593            .with_priority(1)
594            .mount(&server)
595            .await;
596        wiremock::Mock::given(wiremock::matchers::method("POST"))
597            .and(wiremock::matchers::path("/test"))
598            .and(wiremock::matchers::body_bytes(b"raw-body".to_vec()))
599            .respond_with(wiremock::ResponseTemplate::new(200))
600            .with_priority(2)
601            .mount(&server)
602            .await;
603
604        let resp = client
605            .post_bytes(
606                &format!("{}/test", server.uri()),
607                b"raw-body",
608                "application/octet-stream",
609            )
610            .await
611            .unwrap();
612        assert_eq!(resp.status().as_u16(), 200);
613    }
614
615    #[tokio::test]
616    async fn post_bytes_refreshes_and_retries_once_on_401() {
617        let server = wiremock::MockServer::start().await;
618        let client = client_with_bootstrapped_token(&server).await;
619        wiremock::Mock::given(wiremock::matchers::method("POST"))
620            .and(wiremock::matchers::path("/token"))
621            .respond_with(
622                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
623                    "access_token": "refreshed-token",
624                    "expires_in": 3600,
625                })),
626            )
627            .up_to_n_times(1)
628            .with_priority(2)
629            .mount(&server)
630            .await;
631        wiremock::Mock::given(wiremock::matchers::method("POST"))
632            .and(wiremock::matchers::path("/test"))
633            .and(wiremock::matchers::header(
634                "Authorization",
635                "Bearer bootstrap-token",
636            ))
637            .and(wiremock::matchers::body_bytes(b"raw-body".to_vec()))
638            .respond_with(wiremock::ResponseTemplate::new(401))
639            .expect(1)
640            .mount(&server)
641            .await;
642        wiremock::Mock::given(wiremock::matchers::method("POST"))
643            .and(wiremock::matchers::path("/test"))
644            .and(wiremock::matchers::header(
645                "Authorization",
646                "Bearer refreshed-token",
647            ))
648            .and(wiremock::matchers::body_bytes(b"raw-body".to_vec()))
649            .respond_with(wiremock::ResponseTemplate::new(200))
650            .expect(1)
651            .mount(&server)
652            .await;
653
654        let resp = client
655            .post_bytes(
656                &format!("{}/test", server.uri()),
657                b"raw-body",
658                "application/octet-stream",
659            )
660            .await
661            .unwrap();
662        assert_eq!(resp.status().as_u16(), 200);
663    }
664
665    #[tokio::test]
666    async fn get_json_retries_on_429() {
667        let server = wiremock::MockServer::start().await;
668        let client = client_with_bootstrapped_token(&server).await;
669        wiremock::Mock::given(wiremock::matchers::method("GET"))
670            .and(wiremock::matchers::path("/test"))
671            .respond_with(wiremock::ResponseTemplate::new(429).append_header("Retry-After", "0"))
672            .up_to_n_times(1)
673            .with_priority(1)
674            .mount(&server)
675            .await;
676        wiremock::Mock::given(wiremock::matchers::method("GET"))
677            .and(wiremock::matchers::path("/test"))
678            .respond_with(wiremock::ResponseTemplate::new(200))
679            .with_priority(2)
680            .mount(&server)
681            .await;
682
683        let resp = client
684            .get_json(&format!("{}/test", server.uri()))
685            .await
686            .unwrap();
687        assert_eq!(resp.status().as_u16(), 200);
688    }
689
690    // ── set_retry_notify (#1651) ─────────────────────────────────────
691
692    #[tokio::test]
693    async fn set_retry_notify_receives_rate_limit_notices_instead_of_eprintln() {
694        let server = wiremock::MockServer::start().await;
695        let client = client_with_bootstrapped_token(&server).await;
696        wiremock::Mock::given(wiremock::matchers::method("GET"))
697            .and(wiremock::matchers::path("/test"))
698            .respond_with(wiremock::ResponseTemplate::new(429).append_header("Retry-After", "0"))
699            .up_to_n_times(1)
700            .with_priority(1)
701            .mount(&server)
702            .await;
703        wiremock::Mock::given(wiremock::matchers::method("GET"))
704            .and(wiremock::matchers::path("/test"))
705            .respond_with(wiremock::ResponseTemplate::new(200))
706            .with_priority(2)
707            .mount(&server)
708            .await;
709
710        let calls = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
711        let recorded = calls.clone();
712        client.set_retry_notify(std::sync::Arc::new(move |status, delay, attempt| {
713            recorded.lock().unwrap().push((status, delay, attempt));
714        }));
715
716        let resp = client
717            .get_json(&format!("{}/test", server.uri()))
718            .await
719            .unwrap();
720        assert_eq!(resp.status().as_u16(), 200);
721        assert_eq!(*calls.lock().unwrap(), vec![(429, 0, 1)]);
722    }
723
724    #[tokio::test]
725    async fn get_json_retries_403_rate_limit_exceeded_then_succeeds() {
726        let server = wiremock::MockServer::start().await;
727        let client = client_with_bootstrapped_token(&server).await;
728        wiremock::Mock::given(wiremock::matchers::method("GET"))
729            .and(wiremock::matchers::path("/test"))
730            .respond_with(
731                wiremock::ResponseTemplate::new(403)
732                    .append_header("Retry-After", "0")
733                    .set_body_json(serde_json::json!({
734                        "error": {"message": "Rate Limit Exceeded", "errors": [{"reason": "rateLimitExceeded"}]}
735                    })),
736            )
737            .up_to_n_times(1)
738            .with_priority(1)
739            .mount(&server)
740            .await;
741        wiremock::Mock::given(wiremock::matchers::method("GET"))
742            .and(wiremock::matchers::path("/test"))
743            .respond_with(wiremock::ResponseTemplate::new(200))
744            .with_priority(2)
745            .mount(&server)
746            .await;
747
748        let resp = client
749            .get_json(&format!("{}/test", server.uri()))
750            .await
751            .unwrap();
752        assert_eq!(resp.status().as_u16(), 200);
753    }
754
755    #[tokio::test]
756    async fn get_json_does_not_retry_insufficient_permissions_403() {
757        let server = wiremock::MockServer::start().await;
758        let client = client_with_bootstrapped_token(&server).await;
759        wiremock::Mock::given(wiremock::matchers::method("GET"))
760            .and(wiremock::matchers::path("/test"))
761            .respond_with(
762                wiremock::ResponseTemplate::new(403).set_body_json(serde_json::json!({
763                    "error": {"message": "Insufficient Permission", "errors": [{"reason": "insufficientPermissions"}]}
764                })),
765            )
766            .expect(1)
767            .mount(&server)
768            .await;
769
770        let resp = client
771            .get_json(&format!("{}/test", server.uri()))
772            .await
773            .unwrap();
774        assert_eq!(resp.status().as_u16(), 403);
775    }
776
777    #[tokio::test]
778    async fn get_json_refreshes_and_retries_once_on_401() {
779        let server = wiremock::MockServer::start().await;
780        let client = client_with_bootstrapped_token(&server).await;
781        // The refresh endpoint issues a second, distinct token.
782        wiremock::Mock::given(wiremock::matchers::method("POST"))
783            .and(wiremock::matchers::path("/token"))
784            .respond_with(
785                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
786                    "access_token": "refreshed-token",
787                    "expires_in": 3600,
788                })),
789            )
790            .up_to_n_times(1)
791            .with_priority(2)
792            .mount(&server)
793            .await;
794        wiremock::Mock::given(wiremock::matchers::method("GET"))
795            .and(wiremock::matchers::path("/test"))
796            .and(wiremock::matchers::header(
797                "Authorization",
798                "Bearer bootstrap-token",
799            ))
800            .respond_with(wiremock::ResponseTemplate::new(401))
801            .expect(1)
802            .mount(&server)
803            .await;
804        wiremock::Mock::given(wiremock::matchers::method("GET"))
805            .and(wiremock::matchers::path("/test"))
806            .and(wiremock::matchers::header(
807                "Authorization",
808                "Bearer refreshed-token",
809            ))
810            .respond_with(wiremock::ResponseTemplate::new(200))
811            .expect(1)
812            .mount(&server)
813            .await;
814
815        let resp = client
816            .get_json(&format!("{}/test", server.uri()))
817            .await
818            .unwrap();
819        assert_eq!(resp.status().as_u16(), 200);
820    }
821
822    #[tokio::test]
823    async fn get_json_does_not_retry_a_second_time_on_persistent_401() {
824        let server = wiremock::MockServer::start().await;
825        let client = client_with_bootstrapped_token(&server).await;
826        wiremock::Mock::given(wiremock::matchers::method("POST"))
827            .and(wiremock::matchers::path("/token"))
828            .respond_with(
829                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
830                    "access_token": "still-rejected-token",
831                    "expires_in": 3600,
832                })),
833            )
834            .up_to_n_times(1)
835            .with_priority(2)
836            .mount(&server)
837            .await;
838        wiremock::Mock::given(wiremock::matchers::method("GET"))
839            .and(wiremock::matchers::path("/test"))
840            .respond_with(
841                wiremock::ResponseTemplate::new(401).set_body_string("still unauthorized"),
842            )
843            .expect(2)
844            .mount(&server)
845            .await;
846
847        let resp = client
848            .get_json(&format!("{}/test", server.uri()))
849            .await
850            .unwrap();
851        assert_eq!(resp.status().as_u16(), 401);
852    }
853
854    #[tokio::test]
855    async fn response_to_error_extracts_gmail_message_and_reason() {
856        let server = wiremock::MockServer::start().await;
857        let client = client_with_bootstrapped_token(&server).await;
858        // `userRateLimitExceeded` is now retryable (`is_gmail_quota_exceeded`),
859        // so without a zero-delay `Retry-After` this test would wait through
860        // the real exponential backoff before giving up.
861        wiremock::Mock::given(wiremock::matchers::method("GET"))
862            .and(wiremock::matchers::path("/test"))
863            .respond_with(
864                wiremock::ResponseTemplate::new(403)
865                    .append_header("Retry-After", "0")
866                    .set_body_json(serde_json::json!({
867                        "error": {
868                            "message": "User Rate Limit Exceeded",
869                            "errors": [{"reason": "userRateLimitExceeded"}],
870                        }
871                    })),
872            )
873            .mount(&server)
874            .await;
875
876        let resp = client
877            .get_json(&format!("{}/test", server.uri()))
878            .await
879            .unwrap();
880        let err = GmailClient::response_to_error(resp).await;
881        let msg = err.to_string();
882        assert!(msg.contains("User Rate Limit Exceeded"));
883        assert!(msg.contains("userRateLimitExceeded"));
884        assert_eq!(err.reason(), Some("userRateLimitExceeded"));
885    }
886
887    #[tokio::test]
888    async fn response_to_error_omits_reason_suffix_when_absent() {
889        let server = wiremock::MockServer::start().await;
890        let client = client_with_bootstrapped_token(&server).await;
891        wiremock::Mock::given(wiremock::matchers::method("GET"))
892            .and(wiremock::matchers::path("/test"))
893            .respond_with(
894                wiremock::ResponseTemplate::new(400).set_body_json(serde_json::json!({
895                    "error": {
896                        "message": "Invalid request",
897                    }
898                })),
899            )
900            .mount(&server)
901            .await;
902
903        let resp = client
904            .get_json(&format!("{}/test", server.uri()))
905            .await
906            .unwrap();
907        let err = GmailClient::response_to_error(resp).await;
908        let msg = err.to_string();
909        assert!(msg.contains("Invalid request"));
910        assert!(!msg.contains("reason:"));
911        assert_eq!(err.reason(), None);
912    }
913
914    #[tokio::test]
915    async fn response_to_error_falls_back_to_raw_body_when_not_gmail_shaped() {
916        let server = wiremock::MockServer::start().await;
917        let client = client_with_bootstrapped_token(&server).await;
918        wiremock::Mock::given(wiremock::matchers::method("GET"))
919            .and(wiremock::matchers::path("/test"))
920            .respond_with(wiremock::ResponseTemplate::new(500).set_body_string("internal error"))
921            .mount(&server)
922            .await;
923
924        let resp = client
925            .get_json(&format!("{}/test", server.uri()))
926            .await
927            .unwrap();
928        let err = GmailClient::response_to_error(resp).await;
929        assert!(err.to_string().contains("internal error"));
930    }
931
932    #[tokio::test]
933    async fn get_json_propagates_network_errors() {
934        let client = GmailClient::new("http://127.0.0.1:1", &test_credentials()).unwrap();
935        let result = client.get_json("http://127.0.0.1:1/test").await;
936        assert!(result.is_err());
937    }
938
939    #[tokio::test]
940    async fn get_parsed_errors_on_malformed_json_response() {
941        let server = wiremock::MockServer::start().await;
942        let client = client_with_bootstrapped_token(&server).await;
943        wiremock::Mock::given(wiremock::matchers::method("GET"))
944            .and(wiremock::matchers::path("/test"))
945            .respond_with(wiremock::ResponseTemplate::new(200).set_body_string("not json"))
946            .mount(&server)
947            .await;
948
949        let result: Result<serde_json::Value> = client
950            .get_parsed(&format!("{}/test", server.uri()), "test context")
951            .await;
952        assert!(result.is_err());
953    }
954}
omni_dev/gmail/client.rs

omni_dev/gmail/
client.rs
