Skip to main content

max / goingson

23.6 KB · 642 lines History Blame Raw
1 //! IMAP client for fetching and parsing email messages.
2
3 use async_imap::Client;
4 use base64::{engine::general_purpose::STANDARD as BASE64, Engine};
5 use chrono::{DateTime, Utc};
6 use futures_util::StreamExt;
7 use goingson_core::{EmailAccount, FolderSyncState};
8 use tokio::net::TcpStream;
9 use tokio_native_tls::native_tls::TlsConnector as NativeTlsConnector;
10 use tokio_native_tls::TlsConnector;
11
12 use super::mime_parse::{build_parsed_email, mime_nesting_within_limit};
13
14 type ImapSession = async_imap::Session<tokio_native_tls::TlsStream<TcpStream>>;
15
16 /// Maximum email size to download (25 MB). Emails exceeding this are skipped during sync.
17 const MAX_EMAIL_SIZE: u32 = 25 * 1024 * 1024;
18
19
20 /// How many UIDs to body-fetch per IMAP command. Bounds the command length and
21 /// the in-flight fetch pipeline so a large mailbox doesn't issue one enormous
22 /// fetch.
23 const IMAP_FETCH_BATCH: usize = 50;
24
25 /// Raw attachment data extracted during IMAP RFC822 parse.
26 #[derive(Debug, Clone)]
27 pub struct AttachmentPart {
28 pub filename: String,
29 pub mime_type: String,
30 pub data: Vec<u8>,
31 }
32
33 #[derive(Debug, Clone)]
34 pub struct ParsedEmail {
35 pub message_id: Option<String>,
36 pub in_reply_to: Option<String>,
37 /// First entry from the References header (thread root).
38 pub references_root: Option<String>,
39 pub imap_uid: u32,
40 pub source_folder: String,
41 pub from: String,
42 pub to: String,
43 pub subject: String,
44 pub body: String,
45 /// Original HTML body for "Open in Browser" feature.
46 pub html_body: Option<String>,
47 pub date: DateTime<Utc>,
48 pub is_read: bool,
49 /// Attachment parts extracted from MIME structure.
50 pub attachments: Vec<AttachmentPart>,
51 }
52
53 /// Result of fetching from a single IMAP folder, including sync state metadata.
54 pub struct FolderFetchResult {
55 pub emails: Vec<ParsedEmail>,
56 pub uid_validity: Option<u32>,
57 pub max_uid_fetched: Option<u32>,
58 pub debug_info: String,
59 }
60
61 /// Authentication method for IMAP
62 #[derive(Debug, Clone)]
63 pub enum ImapAuth {
64 /// Traditional username/password
65 Password { username: String, password: String },
66 /// OAuth2 XOAUTH2 mechanism
67 XOAuth2 {
68 email: String,
69 access_token: String,
70 },
71 }
72
73 pub struct ImapClient {
74 server: String,
75 port: u16,
76 auth: ImapAuth,
77 }
78
79 impl ImapClient {
80 /// Create an IMAP client with explicit password authentication.
81 ///
82 /// Use this when retrieving credentials from secure storage (keychain).
83 pub fn with_password(account: &EmailAccount, password: &str) -> Self {
84 Self {
85 server: account.imap_server.trim().to_string(),
86 port: account.imap_port as u16,
87 auth: ImapAuth::Password {
88 username: account.username.trim().to_string(),
89 password: password.to_string(),
90 },
91 }
92 }
93
94 /// Create an IMAP client with OAuth2 XOAUTH2 authentication
95 pub fn with_oauth(server: &str, port: u16, email: &str, access_token: &str) -> Self {
96 Self {
97 server: server.to_string(),
98 port,
99 auth: ImapAuth::XOAuth2 {
100 email: email.to_string(),
101 access_token: access_token.to_string(),
102 },
103 }
104 }
105
106 /// Connect to IMAP server and return an authenticated session
107 #[tracing::instrument(skip_all)]
108 async fn connect(&self) -> Result<ImapSession, String> {
109 let addr = format!("{}:{}", self.server, self.port);
110
111 let tcp_stream = tokio::time::timeout(
112 std::time::Duration::from_secs(30),
113 TcpStream::connect(&addr),
114 )
115 .await
116 .map_err(|_| format!("Connection to {} timed out after 30s", addr))?
117 .map_err(|e| format!("Connection error: {}", e))?;
118
119 let tls_connector = NativeTlsConnector::new()
120 .map_err(|e| format!("TLS setup error: {}", e))?;
121 let tls = TlsConnector::from(tls_connector);
122 let tls_stream = tokio::time::timeout(
123 std::time::Duration::from_secs(30),
124 tls.connect(&self.server, tcp_stream),
125 )
126 .await
127 .map_err(|_| format!("TLS handshake with {} timed out after 30s", self.server))?
128 .map_err(|e| format!("TLS error: {}", e))?;
129
130 let client = Client::new(tls_stream);
131
132 match &self.auth {
133 ImapAuth::Password { username, password } => {
134 let session = client
135 .login(username, password)
136 .await
137 .map_err(|e| format!("Login error: {}", e.0))?;
138 Ok(session)
139 }
140 ImapAuth::XOAuth2 { email, access_token } => {
141 // XOAUTH2 format: base64("user=" + email + "\x01auth=Bearer " + token + "\x01\x01")
142 let auth_string = format!("user={}\x01auth=Bearer {}\x01\x01", email, access_token);
143 let auth_base64 = BASE64.encode(auth_string.as_bytes());
144
145 let session = client
146 .authenticate("XOAUTH2", XOAuth2Authenticator(auth_base64))
147 .await
148 .map_err(|e| format!("XOAUTH2 login error: {}", e.0))?;
149 Ok(session)
150 }
151 }
152 }
153
154 /// Fetch emails from a specific folder with debug info
155 #[tracing::instrument(skip_all, fields(folder = %folder))]
156 pub async fn fetch_emails_from_folder_debug(
157 &self,
158 folder: &str,
159 since: Option<DateTime<Utc>>,
160 ) -> Result<(Vec<ParsedEmail>, String), String> {
161 let mut debug = Vec::new();
162 let mut session = self.connect().await?;
163
164 session
165 .select(folder)
166 .await
167 .map_err(|e| format!("Select {} error: {}", folder, e))?;
168
169 // Build search query
170 let search_query = if let Some(since_date) = since {
171 format!("SINCE {}", since_date.format("%d-%b-%Y"))
172 } else {
173 "ALL".to_string()
174 };
175
176 debug.push(format!("search: {}", search_query));
177
178 let search_result = session
179 .search(&search_query)
180 .await
181 .map_err(|e| format!("Search error: {}", e))?;
182
183 // Limit to last 50 if no date filter
184 let sequence_nums: Vec<u32> = if since.is_none() {
185 let mut nums: Vec<u32> = search_result.into_iter().collect();
186 nums.sort();
187 nums.into_iter().rev().take(50).collect()
188 } else {
189 search_result.into_iter().collect()
190 };
191
192 debug.push(format!("seq_nums: {}", sequence_nums.len()));
193
194 if sequence_nums.is_empty() {
195 session.logout().await.ok();
196 return Ok((Vec::new(), debug.join(", ")));
197 }
198
199 let sequence_set = sequence_nums
200 .iter()
201 .map(|n| n.to_string())
202 .collect::<Vec<_>>()
203 .join(",");
204
205 // Pre-filter oversized emails
206 let mut size_stream = session
207 .fetch(&sequence_set, "(UID RFC822.SIZE)")
208 .await
209 .map_err(|e| format!("Size fetch error: {}", e))?;
210
211 let mut safe_seqs: Vec<u32> = Vec::new();
212 let mut skipped_large = 0usize;
213 // Build a set of UIDs that are safe to fetch
214 while let Some(result) = size_stream.next().await {
215 if let Ok(msg) = result {
216 let over_limit = msg.size.is_some_and(|s| s > MAX_EMAIL_SIZE);
217 if over_limit {
218 skipped_large += 1;
219 tracing::warn!(uid = ?msg.uid, size = ?msg.size, folder = %folder, "Skipping oversized email");
220 continue;
221 }
222 // Use the sequence number (message index in stream matches input order)
223 // Re-collect the UIDs we want, then re-fetch by UID
224 if let Some(uid) = msg.uid {
225 safe_seqs.push(uid);
226 }
227 }
228 }
229 drop(size_stream);
230
231 if skipped_large > 0 {
232 debug.push(format!("skipped_large: {}", skipped_large));
233 }
234
235 if safe_seqs.is_empty() {
236 session.logout().await.ok();
237 return Ok((Vec::new(), debug.join(", ")));
238 }
239
240 let mut emails = Vec::new();
241 let folder_name = folder.to_string();
242 let mut msg_count = 0;
243 let mut body_count = 0;
244 let mut parse_errors = 0;
245
246 // Fetch full bodies in bounded UID batches so neither the IMAP command
247 // nor the in-flight fetch pipeline grows with the size of the mailbox.
248 for chunk in safe_seqs.chunks(IMAP_FETCH_BATCH) {
249 let safe_uid_set = chunk.iter().map(|n| n.to_string()).collect::<Vec<_>>().join(",");
250 let mut messages = session
251 .uid_fetch(&safe_uid_set, "(UID FLAGS RFC822)")
252 .await
253 .map_err(|e| format!("Fetch error: {}", e))?;
254
255 while let Some(result) = messages.next().await {
256 msg_count += 1;
257 let message = match result {
258 Ok(m) => m,
259 Err(e) => {
260 debug.push(format!("msg_err: {}", e));
261 continue;
262 }
263 };
264 let uid = match message.uid {
265 Some(u) => u,
266 None => {
267 tracing::warn!(folder = %folder_name, "IMAP message missing UID, skipping");
268 continue;
269 }
270 };
271
272 // Check if \Seen flag is present (email has been read)
273 let is_read = message.flags().any(|f| matches!(f, async_imap::types::Flag::Seen));
274
275 if let Some(body) = message.body() {
276 body_count += 1;
277 if !mime_nesting_within_limit(body) {
278 tracing::warn!(uid, folder = %folder_name, "skipping email with excessive MIME nesting");
279 parse_errors += 1;
280 continue;
281 }
282 match mailparse::parse_mail(body) {
283 Ok(parsed) => emails.push(build_parsed_email(&parsed, uid, &folder_name, is_read)),
284 Err(e) => {
285 tracing::debug!(uid, folder = %folder_name, error = %e, "Failed to parse email");
286 parse_errors += 1;
287 }
288 }
289 }
290 }
291 }
292
293 debug.push(format!("msgs: {}, bodies: {}, parsed: {}, errs: {}", msg_count, body_count, emails.len(), parse_errors));
294 session.logout().await.ok();
295 Ok((emails, debug.join(", ")))
296 }
297
298 /// Fetch emails incrementally using IMAP UIDs when sync state is available,
299 /// falling back to SINCE-based search otherwise.
300 #[tracing::instrument(skip_all)]
301 pub async fn fetch_emails_incremental(
302 &self,
303 folder: &str,
304 sync_state: Option<&FolderSyncState>,
305 since_fallback: Option<DateTime<Utc>>,
306 ) -> Result<FolderFetchResult, String> {
307 let mut debug = Vec::new();
308 let mut session = self.connect().await?;
309
310 let mailbox = session
311 .select(folder)
312 .await
313 .map_err(|e| format!("Select {} error: {}", folder, e))?;
314
315 let server_uid_validity = mailbox.uid_validity;
316
317 // Decide whether to use UID-based or SINCE-based search
318 let use_uid = match (sync_state, server_uid_validity) {
319 (Some(state), Some(validity)) if validity == state.uid_validity => {
320 debug.push(format!("uid_mode: UID {}:* (validity {})", state.last_seen_uid + 1, validity));
321 true
322 }
323 (Some(state), Some(validity)) => {
324 debug.push(format!("uid_validity_mismatch: stored={} server={}, falling back to SINCE", state.uid_validity, validity));
325 false
326 }
327 (Some(_), None) => {
328 debug.push("server_no_uidvalidity, falling back to SINCE".to_string());
329 false
330 }
331 (None, _) => {
332 debug.push("no_sync_state, using SINCE fallback".to_string());
333 false
334 }
335 };
336
337 let uids: Vec<u32> = if use_uid {
338 // `use_uid` is only set true in the (Some, Some) uid_validity-match
339 // arm above, so `sync_state` is guaranteed Some here. Bind it with
340 // let-else to keep that invariant local — a future edit to the match
341 // can't silently turn this into a panic.
342 let Some(state) = &sync_state else {
343 return Err("internal error: use_uid set without sync_state".to_string());
344 };
345 let search_query = format!("UID {}:*", state.last_seen_uid.saturating_add(1));
346 debug.push(format!("uid_search: {}", search_query));
347
348 let search_result = session
349 .uid_search(&search_query)
350 .await
351 .map_err(|e| format!("UID search error: {}", e))?;
352
353 // Filter out the last_seen_uid itself (IMAP UID ranges are inclusive,
354 // and `n:*` returns n even if it's the only match)
355 search_result
356 .into_iter()
357 .filter(|&uid| uid > state.last_seen_uid)
358 .collect()
359 } else {
360 // Fall back to SINCE-based search, then get UIDs via uid_search
361 let search_query = if let Some(since_date) = since_fallback {
362 format!("SINCE {}", since_date.format("%d-%b-%Y"))
363 } else {
364 "ALL".to_string()
365 };
366 debug.push(format!("since_search: {}", search_query));
367
368 let search_result = session
369 .uid_search(&search_query)
370 .await
371 .map_err(|e| format!("Search error: {}", e))?;
372
373 let mut nums: Vec<u32> = search_result.into_iter().collect();
374 if since_fallback.is_none() {
375 nums.sort();
376 nums = nums.into_iter().rev().take(50).collect();
377 }
378 nums
379 };
380
381 debug.push(format!("uids_to_fetch: {}", uids.len()));
382
383 if uids.is_empty() {
384 session.logout().await.ok();
385 return Ok(FolderFetchResult {
386 emails: Vec::new(),
387 uid_validity: server_uid_validity,
388 max_uid_fetched: None,
389 debug_info: debug.join(", "),
390 });
391 }
392
393 // Pre-filter oversized emails by fetching sizes first
394 let uid_set = uids.iter().map(|n| n.to_string()).collect::<Vec<_>>().join(",");
395 let mut size_stream = session
396 .uid_fetch(&uid_set, "(UID RFC822.SIZE)")
397 .await
398 .map_err(|e| format!("UID size fetch error: {}", e))?;
399
400 let mut safe_uids: Vec<u32> = Vec::new();
401 let mut skipped_large = 0usize;
402 // Highest UID we may advance the watermark past without fetching a body:
403 // oversized messages are permanently over the size cap, so re-fetching
404 // them would skip them identically. Parseable messages advance the
405 // watermark only once saved (below), so a transient parse failure is
406 // retried next sync instead of being silently dropped forever.
407 let mut max_skipped_uid: Option<u32> = None;
408 while let Some(result) = size_stream.next().await {
409 if let Ok(msg) = result
410 && let Some(uid) = msg.uid {
411 if let Some(size) = msg.size
412 && size > MAX_EMAIL_SIZE {
413 skipped_large += 1;
414 max_skipped_uid = Some(max_skipped_uid.map_or(uid, |m: u32| m.max(uid)));
415 tracing::warn!(uid, size, folder = %folder, "Skipping oversized email ({} bytes)", size);
416 continue;
417 }
418 safe_uids.push(uid);
419 }
420 }
421 drop(size_stream);
422
423 if skipped_large > 0 {
424 debug.push(format!("skipped_large: {}", skipped_large));
425 }
426
427 if safe_uids.is_empty() {
428 session.logout().await.ok();
429 return Ok(FolderFetchResult {
430 emails: Vec::new(),
431 uid_validity: server_uid_validity,
432 max_uid_fetched: uids.iter().copied().max(),
433 debug_info: debug.join(", "),
434 });
435 }
436
437 let mut emails = Vec::new();
438 let folder_name = folder.to_string();
439 let mut msg_count = 0;
440 let mut body_count = 0;
441 let mut parse_errors = 0;
442 // Seed from the oversized (permanently-skipped) UIDs only. Parseable
443 // messages advance this as they are saved, so a message that fails to
444 // parse does not silently fold into the persisted watermark.
445 let mut max_uid: Option<u32> = max_skipped_uid;
446
447 // Fetch full bodies in bounded UID batches so neither the IMAP command
448 // nor the in-flight fetch pipeline grows with the size of the mailbox.
449 for chunk in safe_uids.chunks(IMAP_FETCH_BATCH) {
450 let safe_uid_set = chunk.iter().map(|n| n.to_string()).collect::<Vec<_>>().join(",");
451 let mut messages = session
452 .uid_fetch(&safe_uid_set, "(UID FLAGS RFC822)")
453 .await
454 .map_err(|e| format!("UID fetch error: {}", e))?;
455
456 while let Some(result) = messages.next().await {
457 msg_count += 1;
458 let message = match result {
459 Ok(m) => m,
460 Err(e) => {
461 debug.push(format!("msg_err: {}", e));
462 continue;
463 }
464 };
465 let uid = match message.uid {
466 Some(u) => u,
467 None => {
468 tracing::warn!(folder = %folder_name, "IMAP message missing UID, skipping");
469 continue;
470 }
471 };
472
473 let is_read = message.flags().any(|f| matches!(f, async_imap::types::Flag::Seen));
474
475 if let Some(body) = message.body() {
476 body_count += 1;
477 if !mime_nesting_within_limit(body) {
478 tracing::warn!(uid, folder = %folder_name, "skipping email with excessive MIME nesting");
479 parse_errors += 1;
480 // Deterministic skip (same message always trips the guard):
481 // safe to advance past so it isn't re-fetched every sync.
482 max_uid = Some(max_uid.map_or(uid, |m: u32| m.max(uid)));
483 continue;
484 }
485 match mailparse::parse_mail(body) {
486 Ok(parsed) => {
487 emails.push(build_parsed_email(&parsed, uid, &folder_name, is_read));
488 // Advance only once the message is saved, so a parse
489 // failure below is retried rather than silently lost.
490 max_uid = Some(max_uid.map_or(uid, |m: u32| m.max(uid)));
491 }
492 Err(e) => {
493 tracing::debug!(uid, folder = %folder_name, error = %e, "Failed to parse email");
494 parse_errors += 1;
495 }
496 }
497 }
498 }
499 }
500
501 debug.push(format!("msgs: {}, bodies: {}, parsed: {}, errs: {}", msg_count, body_count, emails.len(), parse_errors));
502 session.logout().await.ok();
503
504 Ok(FolderFetchResult {
505 emails,
506 uid_validity: server_uid_validity,
507 max_uid_fetched: max_uid,
508 debug_info: debug.join(", "),
509 })
510 }
511
512 /// Move a message between folders using IMAP UID MOVE or COPY+DELETE
513 #[tracing::instrument(skip_all)]
514 pub async fn move_message(
515 &self,
516 uid: u32,
517 from_folder: &str,
518 to_folder: &str,
519 ) -> Result<(), String> {
520 let mut session = self.connect().await?;
521
522 // Select source folder
523 session
524 .select(from_folder)
525 .await
526 .map_err(|e| format!("Failed to select {}: {}", from_folder, e))?;
527
528 let uid_str = uid.to_string();
529
530 // Try MOVE first (RFC 6851) - more efficient and atomic
531 let move_result = session.uid_mv(&uid_str, to_folder).await;
532
533 match move_result {
534 Ok(_) => {
535 session.logout().await.ok();
536 Ok(())
537 }
538 Err(_) => {
539 // Fallback to COPY + DELETE + EXPUNGE
540 session
541 .uid_copy(&uid_str, to_folder)
542 .await
543 .map_err(|e| format!("Copy failed: {}", e))?;
544
545 // Mark as deleted - drop the stream since operation is done
546 let _ = session
547 .uid_store(&uid_str, "+FLAGS (\\Deleted)")
548 .await
549 .map_err(|e| format!("Delete flag failed: {}", e))?;
550
551 // Expunge - drop the stream since operation is done
552 let _ = session
553 .expunge()
554 .await
555 .map_err(|e| format!("Expunge failed: {}", e))?;
556
557 session.logout().await.ok();
558 Ok(())
559 }
560 }
561 }
562
563 /// Archive a message (move from INBOX to Archive folder)
564 #[tracing::instrument(skip_all)]
565 pub async fn archive_message(&self, uid: u32, archive_folder: &str) -> Result<(), String> {
566 self.move_message(uid, "INBOX", archive_folder).await
567 }
568
569 /// Unarchive a message (move from Archive folder to INBOX)
570 #[tracing::instrument(skip_all)]
571 pub async fn unarchive_message(&self, uid: u32, archive_folder: &str) -> Result<(), String> {
572 self.move_message(uid, archive_folder, "INBOX").await
573 }
574
575 #[tracing::instrument(skip_all)]
576 pub async fn test_connection(&self) -> Result<(), String> {
577 let mut session = self.connect().await?;
578 session.logout().await.ok();
579 Ok(())
580 }
581
582 /// List all available folders on the IMAP server
583 #[tracing::instrument(skip_all)]
584 pub async fn list_folders(&self) -> Result<Vec<String>, String> {
585 let mut session = self.connect().await?;
586
587 let folders_stream = session
588 .list(Some(""), Some("*"))
589 .await
590 .map_err(|e| format!("List folders error: {}", e))?;
591
592 use futures_util::StreamExt;
593 let folders: Vec<String> = folders_stream
594 .filter_map(|result| async {
595 result.ok().map(|name| name.name().to_string())
596 })
597 .collect()
598 .await;
599
600 session.logout().await.ok();
601 Ok(folders)
602 }
603
604 /// Debug fetch - returns diagnostic info about what's in a folder
605 #[tracing::instrument(skip_all, fields(folder = %folder))]
606 pub async fn debug_folder(&self, folder: &str) -> Result<String, String> {
607 let mut session = self.connect().await?;
608
609 let mailbox = session
610 .select(folder)
611 .await
612 .map_err(|e| format!("Select {} error: {}", folder, e))?;
613
614 let exists = mailbox.exists;
615 let recent = mailbox.recent;
616
617 // Try a simple search
618 let search_result = session
619 .search("ALL")
620 .await
621 .map_err(|e| format!("Search error: {}", e))?;
622
623 let count: usize = search_result.len();
624
625 session.logout().await.ok();
626
627 Ok(format!("Folder '{}': exists={}, recent={}, search_all={}", folder, exists, recent, count))
628 }
629
630 }
631
632 struct XOAuth2Authenticator(String);
633
634 impl async_imap::Authenticator for XOAuth2Authenticator {
635 type Response = String;
636
637 fn process(&mut self, _challenge: &[u8]) -> Self::Response {
638 // Return the pre-computed base64 XOAUTH2 string
639 std::mem::take(&mut self.0)
640 }
641 }
642