1pub mod batch;
4mod calendars;
5pub mod delta;
6pub mod events;
7mod mail;
8mod me;
9
10pub use calendars::list_calendars;
11pub use events::{
12 EventsPage, NewEvent, RsvpKind, cancel_event, create_event, delete_event, get_event,
13 list_calendar_view, list_events_at, move_event_to_calendar, move_time, rsvp_event,
14 update_event,
15};
16pub use mail::{
17 InboxPage, MailFolder, Outgoing, add_attachment, create_child_folder, create_draft,
18 create_forward_draft, create_mail_folder, create_reply_all_draft, create_reply_draft,
19 delete_attachment, delete_mail_folder, delete_message, fetch_message_headers, forward_message,
20 get_attachment_bytes, get_categories, get_message, list_attachments, list_child_folders,
21 list_drafts, list_folder_messages, list_inbox, list_mail_folders, list_messages_at, mark_read,
22 mark_unread, move_message, reply_all_message, reply_message, search_messages, send_draft,
23 send_mail, set_categories, set_flag, update_draft,
24};
25pub use me::{Me, get_me};
26
27use crate::auth::AuthClient;
28use crate::auth::config;
29use crate::error::ClientError;
30use pidge_core::Message;
31
32const MAX_ATTEMPTS: u32 = 4;
34
35fn is_transient(status: reqwest::StatusCode) -> bool {
37 matches!(status.as_u16(), 429 | 503 | 504)
38}
39
40pub(crate) async fn send_with_retry(
51 req: reqwest::RequestBuilder,
52) -> Result<reqwest::Response, ClientError> {
53 let mut attempt: u32 = 0;
54 loop {
55 let this_try = match req.try_clone() {
56 Some(clone) => clone,
57 None => return Ok(req.send().await?),
58 };
59 let resp = this_try.send().await?;
60 let status = resp.status();
61 if !is_transient(status) {
62 return Ok(resp);
63 }
64 let retry_after = resp
65 .headers()
66 .get(reqwest::header::RETRY_AFTER)
67 .and_then(|v| v.to_str().ok())
68 .and_then(|v| v.parse::<u64>().ok());
69 attempt += 1;
70 if attempt >= MAX_ATTEMPTS {
71 return Err(ClientError::Throttled { retry_after });
72 }
73 let backoff = retry_after
74 .map(std::time::Duration::from_secs)
75 .unwrap_or_else(|| {
76 let jitter = std::time::Duration::from_millis(u64::from(attempt) * 83 % 250);
77 std::time::Duration::from_secs(1u64 << attempt.min(4)) / 2 + jitter
78 });
79 tracing::debug!(
80 status = status.as_u16(),
81 attempt,
82 ?backoff,
83 "retrying Graph request"
84 );
85 tokio::time::sleep(backoff).await;
86 }
87}
88
89pub struct GraphClient {
91 auth: AuthClient,
92 http: reqwest::Client,
93 base_url: String,
94}
95
96impl GraphClient {
97 pub fn new(auth: AuthClient) -> Result<Self, ClientError> {
98 Ok(Self {
99 auth,
100 http: reqwest::Client::builder()
101 .user_agent(format!("pidge/{}", env!("CARGO_PKG_VERSION")))
102 .build()?,
103 base_url: config::GRAPH_BASE.to_string(),
104 })
105 }
106
107 pub fn for_test(auth: AuthClient, base_url: impl Into<String>) -> Self {
108 Self {
109 auth,
110 http: reqwest::Client::new(),
111 base_url: base_url.into(),
112 }
113 }
114
115 pub fn auth(&self) -> &AuthClient {
116 &self.auth
117 }
118
119 pub async fn me(&self, access_token: &str) -> Result<Me, ClientError> {
121 get_me(&self.http, &self.base_url, access_token).await
122 }
123
124 pub async fn list_inbox(
127 &self,
128 account: &str,
129 limit: usize,
130 skip: usize,
131 unread_only: bool,
132 ) -> Result<InboxPage, ClientError> {
133 let token = self.auth.get_valid_token(account).await?;
134 list_inbox(
135 &self.http,
136 &self.base_url,
137 &token,
138 account,
139 limit,
140 skip,
141 unread_only,
142 )
143 .await
144 }
145
146 pub async fn list_folder(
148 &self,
149 account: &str,
150 folder_id: &str,
151 limit: usize,
152 skip: usize,
153 unread_only: bool,
154 ) -> Result<InboxPage, ClientError> {
155 let token = self.auth.get_valid_token(account).await?;
156 list_folder_messages(
157 &self.http,
158 &self.base_url,
159 &token,
160 account,
161 folder_id,
162 limit,
163 skip,
164 unread_only,
165 )
166 .await
167 }
168
169 pub async fn mail_delta_bootstrap(
172 &self,
173 account: &str,
174 folder: &str,
175 ) -> Result<(Vec<Message>, String), ClientError> {
176 let token = self.auth.get_valid_token(account).await?;
177 delta::mail_delta_bootstrap(&self.http, &self.base_url, &token, account, folder).await
178 }
179
180 pub async fn mail_delta(
182 &self,
183 account: &str,
184 delta_link: &str,
185 ) -> Result<(Vec<delta::MailDeltaEvent>, String), ClientError> {
186 let token = self.auth.get_valid_token(account).await?;
187 delta::mail_delta(&self.http, &token, account, delta_link).await
188 }
189
190 pub async fn calendar_delta_bootstrap(
192 &self,
193 account: &str,
194 start: chrono::DateTime<chrono::Utc>,
195 end: chrono::DateTime<chrono::Utc>,
196 ) -> Result<(Vec<pidge_core::Event>, String), ClientError> {
197 let token = self.auth.get_valid_token(account).await?;
198 delta::calendar_delta_bootstrap(&self.http, &self.base_url, &token, account, start, end)
199 .await
200 }
201
202 pub async fn calendar_delta(
204 &self,
205 account: &str,
206 delta_link: &str,
207 ) -> Result<(Vec<delta::CalendarDeltaEvent>, String), ClientError> {
208 let token = self.auth.get_valid_token(account).await?;
209 delta::calendar_delta(&self.http, &token, account, delta_link).await
210 }
211
212 pub async fn batch_all(
214 &self,
215 account: &str,
216 requests: Vec<batch::BatchRequest>,
217 ) -> Result<Vec<batch::BatchResponse>, ClientError> {
218 let token = self.auth.get_valid_token(account).await?;
219 batch::batch_all(&self.http, &self.base_url, &token, requests).await
220 }
221
222 pub async fn list_conversation(
224 &self,
225 account: &str,
226 conversation_id: &str,
227 ) -> Result<Vec<Message>, ClientError> {
228 let token = self.auth.get_valid_token(account).await?;
229 mail::list_conversation(&self.http, &self.base_url, &token, account, conversation_id).await
230 }
231
232 pub async fn list_messages_at(
234 &self,
235 account: &str,
236 url: &str,
237 ) -> Result<InboxPage, ClientError> {
238 let token = self.auth.get_valid_token(account).await?;
239 list_messages_at(&self.http, &token, account, url).await
240 }
241
242 pub async fn search_messages(
243 &self,
244 account: &str,
245 query: &str,
246 limit: usize,
247 ) -> Result<InboxPage, ClientError> {
248 let token = self.auth.get_valid_token(account).await?;
249 search_messages(&self.http, &self.base_url, &token, account, query, limit).await
250 }
251
252 pub async fn mark_unread(&self, account: &str, message_id: &str) -> Result<(), ClientError> {
254 let token = self.auth.get_valid_token(account).await?;
255 mail::mark_unread(&self.http, &self.base_url, &token, message_id).await
256 }
257
258 pub async fn set_flag(
260 &self,
261 account: &str,
262 message_id: &str,
263 flagged: bool,
264 ) -> Result<(), ClientError> {
265 let token = self.auth.get_valid_token(account).await?;
266 mail::set_flag(&self.http, &self.base_url, &token, message_id, flagged).await
267 }
268
269 pub async fn get_categories(
271 &self,
272 account: &str,
273 message_id: &str,
274 ) -> Result<Vec<String>, ClientError> {
275 let token = self.auth.get_valid_token(account).await?;
276 mail::get_categories(&self.http, &self.base_url, &token, message_id).await
277 }
278
279 pub async fn set_categories(
281 &self,
282 account: &str,
283 message_id: &str,
284 categories: &[String],
285 ) -> Result<(), ClientError> {
286 let token = self.auth.get_valid_token(account).await?;
287 mail::set_categories(&self.http, &self.base_url, &token, message_id, categories).await
288 }
289
290 pub async fn move_message(
292 &self,
293 account: &str,
294 message_id: &str,
295 destination: &str,
296 ) -> Result<(), ClientError> {
297 let token = self.auth.get_valid_token(account).await?;
298 mail::move_message(&self.http, &self.base_url, &token, message_id, destination).await
299 }
300
301 pub async fn list_mail_folders(&self, account: &str) -> Result<Vec<MailFolder>, ClientError> {
303 let token = self.auth.get_valid_token(account).await?;
304 mail::list_mail_folders(&self.http, &self.base_url, &token).await
305 }
306
307 pub async fn create_mail_folder(
309 &self,
310 account: &str,
311 display_name: &str,
312 ) -> Result<MailFolder, ClientError> {
313 let token = self.auth.get_valid_token(account).await?;
314 mail::create_mail_folder(&self.http, &self.base_url, &token, display_name).await
315 }
316
317 pub async fn list_child_folders(
319 &self,
320 account: &str,
321 parent_id: &str,
322 ) -> Result<Vec<MailFolder>, ClientError> {
323 let token = self.auth.get_valid_token(account).await?;
324 mail::list_child_folders(&self.http, &self.base_url, &token, parent_id).await
325 }
326
327 pub async fn create_child_folder(
329 &self,
330 account: &str,
331 parent_id: &str,
332 display_name: &str,
333 ) -> Result<MailFolder, ClientError> {
334 let token = self.auth.get_valid_token(account).await?;
335 mail::create_child_folder(&self.http, &self.base_url, &token, parent_id, display_name).await
336 }
337
338 pub async fn delete_mail_folder(
341 &self,
342 account: &str,
343 folder_id: &str,
344 ) -> Result<(), ClientError> {
345 let token = self.auth.get_valid_token(account).await?;
346 mail::delete_mail_folder(&self.http, &self.base_url, &token, folder_id).await
347 }
348
349 pub async fn send_mail(&self, account: &str, message: &Outgoing) -> Result<(), ClientError> {
351 let token = self.auth.get_valid_token(account).await?;
352 mail::send_mail(&self.http, &self.base_url, &token, message).await
353 }
354
355 pub async fn reply_message(
357 &self,
358 account: &str,
359 message_id: &str,
360 comment: &str,
361 ) -> Result<(), ClientError> {
362 let token = self.auth.get_valid_token(account).await?;
363 mail::reply_message(&self.http, &self.base_url, &token, message_id, comment).await
364 }
365
366 pub async fn reply_all_message(
368 &self,
369 account: &str,
370 message_id: &str,
371 comment: &str,
372 ) -> Result<(), ClientError> {
373 let token = self.auth.get_valid_token(account).await?;
374 mail::reply_all_message(&self.http, &self.base_url, &token, message_id, comment).await
375 }
376
377 pub async fn forward_message(
379 &self,
380 account: &str,
381 message_id: &str,
382 to: &[String],
383 comment: &str,
384 ) -> Result<(), ClientError> {
385 let token = self.auth.get_valid_token(account).await?;
386 mail::forward_message(&self.http, &self.base_url, &token, message_id, to, comment).await
387 }
388
389 pub async fn list_drafts(
391 &self,
392 account: &str,
393 limit: usize,
394 skip: usize,
395 ) -> Result<InboxPage, ClientError> {
396 let token = self.auth.get_valid_token(account).await?;
397 mail::list_drafts(&self.http, &self.base_url, &token, account, limit, skip).await
398 }
399
400 pub async fn create_draft(
402 &self,
403 account: &str,
404 message: &Outgoing,
405 ) -> Result<String, ClientError> {
406 let token = self.auth.get_valid_token(account).await?;
407 mail::create_draft(&self.http, &self.base_url, &token, message).await
408 }
409
410 pub async fn create_reply_draft(
412 &self,
413 account: &str,
414 message_id: &str,
415 comment: &str,
416 ) -> Result<String, ClientError> {
417 let token = self.auth.get_valid_token(account).await?;
418 mail::create_reply_draft(&self.http, &self.base_url, &token, message_id, comment).await
419 }
420
421 pub async fn create_reply_all_draft(
423 &self,
424 account: &str,
425 message_id: &str,
426 comment: &str,
427 ) -> Result<String, ClientError> {
428 let token = self.auth.get_valid_token(account).await?;
429 mail::create_reply_all_draft(&self.http, &self.base_url, &token, message_id, comment).await
430 }
431
432 pub async fn create_forward_draft(
434 &self,
435 account: &str,
436 message_id: &str,
437 to: &[String],
438 comment: &str,
439 ) -> Result<String, ClientError> {
440 let token = self.auth.get_valid_token(account).await?;
441 mail::create_forward_draft(&self.http, &self.base_url, &token, message_id, to, comment)
442 .await
443 }
444
445 pub async fn send_draft(&self, account: &str, message_id: &str) -> Result<(), ClientError> {
447 let token = self.auth.get_valid_token(account).await?;
448 mail::send_draft(&self.http, &self.base_url, &token, message_id).await
449 }
450
451 pub async fn update_draft(
453 &self,
454 account: &str,
455 message_id: &str,
456 message: &Outgoing,
457 ) -> Result<(), ClientError> {
458 let token = self.auth.get_valid_token(account).await?;
459 mail::update_draft(&self.http, &self.base_url, &token, message_id, message).await
460 }
461
462 pub async fn delete_message(&self, account: &str, message_id: &str) -> Result<(), ClientError> {
465 let token = self.auth.get_valid_token(account).await?;
466 mail::delete_message(&self.http, &self.base_url, &token, message_id).await
467 }
468
469 pub async fn add_attachment(
471 &self,
472 account: &str,
473 message_id: &str,
474 name: &str,
475 content_type: &str,
476 bytes: &[u8],
477 ) -> Result<String, ClientError> {
478 let token = self.auth.get_valid_token(account).await?;
479 mail::add_attachment(
480 &self.http,
481 &self.base_url,
482 &token,
483 message_id,
484 name,
485 content_type,
486 bytes,
487 )
488 .await
489 }
490
491 pub async fn delete_attachment(
493 &self,
494 account: &str,
495 message_id: &str,
496 attachment_id: &str,
497 ) -> Result<(), ClientError> {
498 let token = self.auth.get_valid_token(account).await?;
499 mail::delete_attachment(
500 &self.http,
501 &self.base_url,
502 &token,
503 message_id,
504 attachment_id,
505 )
506 .await
507 }
508
509 pub async fn get_message(
511 &self,
512 account: &str,
513 message_id: &str,
514 ) -> Result<pidge_core::FullMessage, ClientError> {
515 let token = self.auth.get_valid_token(account).await?;
516 mail::get_message(&self.http, &self.base_url, &token, account, message_id).await
517 }
518
519 pub async fn fetch_message_headers(
521 &self,
522 account: &str,
523 message_id: &str,
524 ) -> Result<Vec<(String, String)>, ClientError> {
525 let token = self.auth.get_valid_token(account).await?;
526 mail::fetch_message_headers(&self.http, &self.base_url, &token, message_id).await
527 }
528
529 pub async fn list_attachments(
531 &self,
532 account: &str,
533 message_id: &str,
534 ) -> Result<Vec<pidge_core::Attachment>, ClientError> {
535 let token = self.auth.get_valid_token(account).await?;
536 mail::list_attachments(&self.http, &self.base_url, &token, message_id).await
537 }
538
539 pub async fn get_attachment_bytes(
541 &self,
542 account: &str,
543 message_id: &str,
544 attachment_id: &str,
545 ) -> Result<Vec<u8>, ClientError> {
546 let token = self.auth.get_valid_token(account).await?;
547 mail::get_attachment_bytes(
548 &self.http,
549 &self.base_url,
550 &token,
551 message_id,
552 attachment_id,
553 )
554 .await
555 }
556
557 pub async fn mark_read(&self, account: &str, message_id: &str) -> Result<(), ClientError> {
559 let token = self.auth.get_valid_token(account).await?;
560 mail::mark_read(&self.http, &self.base_url, &token, message_id).await
561 }
562
563 pub async fn list_calendars(
567 &self,
568 account: &str,
569 ) -> Result<Vec<pidge_core::Calendar>, ClientError> {
570 let token = self.auth.get_valid_token(account).await?;
571 calendars::list_calendars(&self.http, &self.base_url, &token, account).await
572 }
573
574 pub async fn list_calendar_view(
576 &self,
577 account: &str,
578 calendar_id: Option<&str>,
579 start: chrono::DateTime<chrono::Utc>,
580 end: chrono::DateTime<chrono::Utc>,
581 limit: usize,
582 ) -> Result<events::EventsPage, ClientError> {
583 let token = self.auth.get_valid_token(account).await?;
584 events::list_calendar_view(
585 &self.http,
586 &self.base_url,
587 &token,
588 account,
589 calendar_id,
590 start,
591 end,
592 limit,
593 )
594 .await
595 }
596
597 pub async fn list_events_at(
599 &self,
600 account: &str,
601 url: &str,
602 ) -> Result<EventsPage, ClientError> {
603 let token = self.auth.get_valid_token(account).await?;
604 list_events_at(&self.http, &token, account, url).await
605 }
606
607 pub async fn get_event(
609 &self,
610 account: &str,
611 event_id: &str,
612 ) -> Result<pidge_core::Event, ClientError> {
613 let token = self.auth.get_valid_token(account).await?;
614 events::get_event(&self.http, &self.base_url, &token, account, event_id).await
615 }
616
617 pub async fn create_event(
619 &self,
620 account: &str,
621 calendar_id: Option<&str>,
622 new_event: &events::NewEvent,
623 ) -> Result<String, ClientError> {
624 let token = self.auth.get_valid_token(account).await?;
625 events::create_event(&self.http, &self.base_url, &token, calendar_id, new_event).await
626 }
627
628 pub async fn update_event(
630 &self,
631 account: &str,
632 event_id: &str,
633 new_event: &events::NewEvent,
634 ) -> Result<(), ClientError> {
635 let token = self.auth.get_valid_token(account).await?;
636 events::update_event(&self.http, &self.base_url, &token, event_id, new_event).await
637 }
638
639 pub async fn move_time(
641 &self,
642 account: &str,
643 event_id: &str,
644 start: chrono::DateTime<chrono::Utc>,
645 end: chrono::DateTime<chrono::Utc>,
646 tz: &str,
647 ) -> Result<(), ClientError> {
648 let token = self.auth.get_valid_token(account).await?;
649 events::move_time(&self.http, &self.base_url, &token, event_id, start, end, tz).await
650 }
651
652 pub async fn delete_event(&self, account: &str, event_id: &str) -> Result<(), ClientError> {
654 let token = self.auth.get_valid_token(account).await?;
655 events::delete_event(&self.http, &self.base_url, &token, event_id).await
656 }
657
658 pub async fn cancel_event(
660 &self,
661 account: &str,
662 event_id: &str,
663 comment: &str,
664 ) -> Result<(), ClientError> {
665 let token = self.auth.get_valid_token(account).await?;
666 events::cancel_event(&self.http, &self.base_url, &token, event_id, comment).await
667 }
668
669 pub async fn rsvp_event(
671 &self,
672 account: &str,
673 event_id: &str,
674 kind: events::RsvpKind,
675 comment: &str,
676 send_response: bool,
677 ) -> Result<(), ClientError> {
678 let token = self.auth.get_valid_token(account).await?;
679 events::rsvp_event(
680 &self.http,
681 &self.base_url,
682 &token,
683 event_id,
684 kind,
685 comment,
686 send_response,
687 )
688 .await
689 }
690
691 pub async fn move_event_to_calendar(
693 &self,
694 account: &str,
695 event_id: &str,
696 destination_calendar_id: &str,
697 ) -> Result<(), ClientError> {
698 let token = self.auth.get_valid_token(account).await?;
699 events::move_event_to_calendar(
700 &self.http,
701 &self.base_url,
702 &token,
703 event_id,
704 destination_calendar_id,
705 )
706 .await
707 }
708}
709
710#[cfg(test)]
711mod retry_tests {
712 use super::*;
713 use wiremock::matchers::{method, path};
714 use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate};
715
716 struct FlakyResponder {
718 failures: std::sync::atomic::AtomicU32,
719 }
720
721 impl Respond for FlakyResponder {
722 fn respond(&self, _req: &Request) -> ResponseTemplate {
723 let n = self
724 .failures
725 .fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
726 if n > 0 {
727 ResponseTemplate::new(429).insert_header("Retry-After", "0")
728 } else {
729 ResponseTemplate::new(200).set_body_json(serde_json::json!({"ok": true}))
730 }
731 }
732 }
733
734 #[tokio::test]
735 async fn retries_transient_429_until_success() {
736 let server = MockServer::start().await;
737 Mock::given(method("GET"))
738 .and(path("/thing"))
739 .respond_with(FlakyResponder {
740 failures: std::sync::atomic::AtomicU32::new(2),
741 })
742 .expect(3)
743 .mount(&server)
744 .await;
745
746 let http = reqwest::Client::new();
747 let resp = send_with_retry(http.get(format!("{}/thing", server.uri())))
748 .await
749 .unwrap();
750 assert_eq!(resp.status(), 200);
751 }
752
753 #[tokio::test]
754 async fn persistent_429_becomes_throttled_after_max_attempts() {
755 let server = MockServer::start().await;
756 Mock::given(method("GET"))
757 .and(path("/thing"))
758 .respond_with(ResponseTemplate::new(429).insert_header("Retry-After", "0"))
759 .expect(4)
760 .mount(&server)
761 .await;
762
763 let http = reqwest::Client::new();
764 let err = send_with_retry(http.get(format!("{}/thing", server.uri())))
765 .await
766 .unwrap_err();
767 match err {
768 ClientError::Throttled { retry_after } => assert_eq!(retry_after, Some(0)),
769 other => panic!("expected Throttled, got {other:?}"),
770 }
771 }
772
773 #[tokio::test]
774 async fn non_transient_errors_pass_through_without_retry() {
775 let server = MockServer::start().await;
776 Mock::given(method("GET"))
777 .and(path("/thing"))
778 .respond_with(ResponseTemplate::new(400).set_body_string("bad"))
779 .expect(1)
780 .mount(&server)
781 .await;
782
783 let http = reqwest::Client::new();
784 let resp = send_with_retry(http.get(format!("{}/thing", server.uri())))
785 .await
786 .unwrap();
787 assert_eq!(resp.status(), 400);
788 }
789}