diff --git a/crates/bridge/src/backup.rs b/crates/bridge/src/backup.rs index 664ecb6..80f0ae1 100644 --- a/crates/bridge/src/backup.rs +++ b/crates/bridge/src/backup.rs @@ -82,7 +82,8 @@ pub async fn export_eml( output: &Path, mut progress: impl FnMut(&BackupProgress), ) -> Result { - std::fs::create_dir_all(output).map_err(|e| format!("Cannot create {}: {e}", output.display()))?; + std::fs::create_dir_all(output) + .map_err(|e| format!("Cannot create {}: {e}", output.display()))?; let folders = backend.list_folders().await?; let mut stats = BackupStats::default(); @@ -394,7 +395,11 @@ mod tests { folder: &FolderInfo, _limit: usize, ) -> Result, String> { - Ok(self.mails.get(&folder.entries_list_id).cloned().unwrap_or_default()) + Ok(self + .mails + .get(&folder.entries_list_id) + .cloned() + .unwrap_or_default()) } async fn load_mail_details(&self, _mail: &Mail) -> Result, String> { *self.server_loads.lock().unwrap() += 1; @@ -449,7 +454,8 @@ mod tests { fn temp_store() -> (LocalStore, std::path::PathBuf) { let randomizer = RandomizerFacade::from_core(rand_core::OsRng); let key: GenericAesKey = GenericAesKey::Aes256(Aes256Key::generate(&randomizer)); - let tmp = std::env::temp_dir().join(format!("tutabridge_backup_test_{}", rand::random::())); + let tmp = + std::env::temp_dir().join(format!("tutabridge_backup_test_{}", rand::random::())); std::fs::create_dir_all(&tmp).unwrap(); let store = LocalStore::open(&tmp.join("s.db"), &tmp.join("mails"), key).unwrap(); (store, tmp) @@ -486,7 +492,8 @@ mod tests { server_loads: std::sync::Mutex::new(0), }; - let out = std::env::temp_dir().join(format!("tutabridge_backup_out_{}", rand::random::())); + let out = + std::env::temp_dir().join(format!("tutabridge_backup_out_{}", rand::random::())); let mut progress_calls = 0; let stats = export_eml(&backend, &store, &out, |_p| progress_calls += 1) .await @@ -494,7 +501,10 @@ mod tests { assert_eq!(stats.folders, 2); assert_eq!(stats.mails_written, 3); - assert_eq!(stats.from_cache, 1, "the seeded mail should come from cache"); + assert_eq!( + stats.from_cache, 1, + "the seeded mail should come from cache" + ); assert_eq!(stats.from_server, 2, "the other two should be fetched"); assert_eq!(progress_calls, 3); assert!(stats.errors.is_empty()); @@ -504,7 +514,10 @@ mod tests { // Files landed in the right per-folder dirs with the date prefix. let inbox = out.join("INBOX"); let sent = out.join("Sent"); - let inbox_files: Vec<_> = std::fs::read_dir(&inbox).unwrap().filter_map(|e| e.ok()).collect(); + let inbox_files: Vec<_> = std::fs::read_dir(&inbox) + .unwrap() + .filter_map(|e| e.ok()) + .collect(); assert_eq!(inbox_files.len(), 2); assert_eq!(std::fs::read_dir(&sent).unwrap().count(), 1); @@ -512,13 +525,17 @@ mod tests { let cached_path = inbox.join("20241225-123725_Cached1--3-9.eml"); let cached_eml = std::fs::read_to_string(&cached_path).unwrap(); let from_cache_b64 = base64::engine::general_purpose::STANDARD.encode(b"

from cache

"); - assert!(cached_eml.contains(&from_cache_b64), "cached body should be served verbatim"); + assert!( + cached_eml.contains(&from_cache_b64), + "cached body should be served verbatim" + ); // --- second run resumes: everything is already on disk --- - let stats2 = export_eml(&backend, &store, &out, |_p| {}) - .await - .unwrap(); - assert_eq!(stats2.skipped, 3, "a re-run must skip every already-exported mail"); + let stats2 = export_eml(&backend, &store, &out, |_p| {}).await.unwrap(); + assert_eq!( + stats2.skipped, 3, + "a re-run must skip every already-exported mail" + ); assert_eq!(stats2.mails_written, 0); assert_eq!(stats2.from_server, 0, "no server fetch on a resume"); assert_eq!( @@ -538,7 +555,10 @@ mod tests { // First backup: one mail fetched. let mut m1 = HashMap::new(); - m1.insert("inbox_entries".to_string(), vec![make_mail("Old1--3-9", "old")]); + m1.insert( + "inbox_entries".to_string(), + vec![make_mail("Old1--3-9", "old")], + ); let backend1 = MockBackend { folders: vec![folder("inbox", "inbox_entries", "INBOX")], mails: m1, diff --git a/crates/bridge/src/bridge.rs b/crates/bridge/src/bridge.rs index d6e160c..828816a 100644 --- a/crates/bridge/src/bridge.rs +++ b/crates/bridge/src/bridge.rs @@ -189,7 +189,8 @@ impl BridgeHandle { self.emit_log("TLS initialized"); self.emit_log(&format!("Authenticating as {}...", config.email)); - let session = match tuta::login_with_2fa(&config, password.as_deref(), totp_callback).await { + let session = match tuta::login_with_2fa(&config, password.as_deref(), totp_callback).await + { Ok(s) => s, Err(e) => { let msg = format!("Login failed: {e}"); @@ -290,7 +291,7 @@ impl BridgeHandle { "Event bus catch-up state loaded ({} group(s))", m.len() )); - }, + } Ok(_) => self.emit_log("Event bus catch-up state is empty (first launch)"), Err(e) => self.emit_log(&format!("Could not load event_bus_state: {e}")), } @@ -356,7 +357,7 @@ impl BridgeHandle { tokio::spawn(async move { if let Err(e) = client.run(token, uid, event_tx, shutdown).await { match e { - tutasdk::event_bus::EventBusError::Stopped => {}, + tutasdk::event_bus::EventBusError::Stopped => {} _ => log::warn!("Event bus exited: {e}"), } } @@ -378,7 +379,8 @@ impl BridgeHandle { imap_tls, pw.clone(), )); - let mut smtp_handle = tokio::spawn(smtp::serve(smtp_port, backend.clone(), smtp_tls, pw)); + let mut smtp_handle = + tokio::spawn(smtp::serve(smtp_port, backend.clone(), smtp_tls, pw)); tokio::select! { _ = rx => { diff --git a/crates/bridge/src/config.rs b/crates/bridge/src/config.rs index dba4a7b..69ab20b 100644 --- a/crates/bridge/src/config.rs +++ b/crates/bridge/src/config.rs @@ -81,7 +81,9 @@ pub fn ensure_bridge_password(config: &mut Config) -> Result Result> { +pub fn regenerate_bridge_password( + config: &mut Config, +) -> Result> { let password = generate_bridge_password(); config.bridge_password = Some(password.clone()); save_config(config)?; diff --git a/crates/bridge/src/event_handler.rs b/crates/bridge/src/event_handler.rs index 396fafe..d22e565 100644 --- a/crates/bridge/src/event_handler.rs +++ b/crates/bridge/src/event_handler.rs @@ -41,59 +41,62 @@ const MAIL_SET_ENTRY_TYPE_ID: i64 = 1450; const MAIL_SET_TYPE_ID: i64 = 429; pub async fn run_event_handler( - store: Arc, - local_store: Arc, - backend: Arc, - sync_limit: usize, - last_batch_ids: Arc>>, - mut rx: mpsc::Receiver, - mut shutdown: watch::Receiver, + store: Arc, + local_store: Arc, + backend: Arc, + sync_limit: usize, + last_batch_ids: Arc>>, + mut rx: mpsc::Receiver, + mut shutdown: watch::Receiver, ) { - info!("Event handler started"); - loop { - tokio::select! { - biased; - _ = shutdown.changed() => break, - msg = rx.recv() => { - let Some(msg) = msg else { break }; - process(&store, &local_store, &*backend, sync_limit, &last_batch_ids, msg).await; - } - } - } - info!("Event handler shutting down"); + info!("Event handler started"); + loop { + tokio::select! { + biased; + _ = shutdown.changed() => break, + msg = rx.recv() => { + let Some(msg) = msg else { break }; + process(&store, &local_store, &*backend, sync_limit, &last_batch_ids, msg).await; + } + } + } + info!("Event handler shutting down"); } async fn process( - store: &MailStore, - local_store: &LocalStore, - backend: &dyn MailBackend, - sync_limit: usize, - last_batch_ids: &Mutex>, - msg: EventBusMessage, + store: &MailStore, + local_store: &LocalStore, + backend: &dyn MailBackend, + sync_limit: usize, + last_batch_ids: &Mutex>, + msg: EventBusMessage, ) { - let batch = match msg { - EventBusMessage::EntityUpdate(b) => b, - EventBusMessage::InitialSyncDone => { - info!("Event bus initial sync done"); - return; - }, - // Counter / leader / op-status / phishing / work-estimate / unknown: - // nothing to do at this layer. - _ => return, - }; + let batch = match msg { + EventBusMessage::EntityUpdate(b) => b, + EventBusMessage::InitialSyncDone => { + info!("Event bus initial sync done"); + return; + } + // Counter / leader / op-status / phishing / work-estimate / unknown: + // nothing to do at this layer. + _ => return, + }; - apply_batch(store, local_store, backend, sync_limit, &batch).await; + apply_batch(store, local_store, backend, sync_limit, &batch).await; - // Advance the in-memory catch-up state and persist it. The two must stay - // in sync — the in-memory map drives the next reconnect's query string, - // the on-disk row survives bridge restarts. - { - let mut ids = last_batch_ids.lock().unwrap(); - ids.insert(batch.group_id.clone(), batch.batch_id.clone()); - } - if let Err(e) = local_store.set_event_bus_batch_id(&batch.group_id, &batch.batch_id) { - warn!("Failed to persist last batch id for {}: {}", batch.group_id, e); - } + // Advance the in-memory catch-up state and persist it. The two must stay + // in sync — the in-memory map drives the next reconnect's query string, + // the on-disk row survives bridge restarts. + { + let mut ids = last_batch_ids.lock().unwrap(); + ids.insert(batch.group_id.clone(), batch.batch_id.clone()); + } + if let Err(e) = local_store.set_event_bus_batch_id(&batch.group_id, &batch.batch_id) { + warn!( + "Failed to persist last batch id for {}: {}", + batch.group_id, e + ); + } } /// Bucketed view of the mail-relevant entity updates inside a batch. Pure; @@ -111,42 +114,42 @@ async fn process( #[cfg_attr(test, derive(Debug))] #[derive(Default)] struct Bucketed<'a> { - mail_set_entry_creates: Vec<&'a EntityUpdateEvent>, - mail_set_entry_deletes: Vec<&'a EntityUpdateEvent>, - /// Mail CREATEs — kept separate so we can pre-decrypt Mail + blob - /// inline before the matching MailSetEntry CREATE asks for the mail. - mail_creates: Vec<&'a EntityUpdateEvent>, - /// Mail UPDATE / DELETE events. - mail_events: Vec<&'a EntityUpdateEvent>, - /// A `MailSet` entity event arrived — the folder list itself changed - /// (custom folder created / renamed / deleted). Triggers a refresh of - /// `store.list_folders()` and a prune of folders no longer on the server. - folder_list_dirty: bool, + mail_set_entry_creates: Vec<&'a EntityUpdateEvent>, + mail_set_entry_deletes: Vec<&'a EntityUpdateEvent>, + /// Mail CREATEs — kept separate so we can pre-decrypt Mail + blob + /// inline before the matching MailSetEntry CREATE asks for the mail. + mail_creates: Vec<&'a EntityUpdateEvent>, + /// Mail UPDATE / DELETE events. + mail_events: Vec<&'a EntityUpdateEvent>, + /// A `MailSet` entity event arrived — the folder list itself changed + /// (custom folder created / renamed / deleted). Triggers a refresh of + /// `store.list_folders()` and a prune of folders no longer on the server. + folder_list_dirty: bool, } fn bucket_updates(updates: &[EntityUpdateEvent]) -> Bucketed<'_> { - let mut out = Bucketed::default(); - for ev in updates { - if ev.application != TUTANOTA_APP { - continue; - } - match ev.type_id { - MAIL_SET_ENTRY_TYPE_ID => match ev.operation { - Operation::Create => out.mail_set_entry_creates.push(ev), - Operation::Delete => out.mail_set_entry_deletes.push(ev), - // The Tuta model treats `MailSetEntry` as immutable — only - // CREATE / DELETE happen. Ignore other operations defensively. - _ => {}, - }, - MAIL_TYPE_ID => match ev.operation { - Operation::Create => out.mail_creates.push(ev), - _ => out.mail_events.push(ev), - }, - MAIL_SET_TYPE_ID => out.folder_list_dirty = true, - _ => {}, - } - } - out + let mut out = Bucketed::default(); + for ev in updates { + if ev.application != TUTANOTA_APP { + continue; + } + match ev.type_id { + MAIL_SET_ENTRY_TYPE_ID => match ev.operation { + Operation::Create => out.mail_set_entry_creates.push(ev), + Operation::Delete => out.mail_set_entry_deletes.push(ev), + // The Tuta model treats `MailSetEntry` as immutable — only + // CREATE / DELETE happen. Ignore other operations defensively. + _ => {} + }, + MAIL_TYPE_ID => match ev.operation { + Operation::Create => out.mail_creates.push(ev), + _ => out.mail_events.push(ev), + }, + MAIL_SET_TYPE_ID => out.folder_list_dirty = true, + _ => {} + } + } + out } /// A Mail+details pair recovered by inline-decrypting a Mail CREATE event @@ -154,256 +157,260 @@ fn bucket_updates(updates: &[EntityUpdateEvent]) -> Bucketed<'_> { /// when the server also bundled `event.blob_instance` (it's `None` on /// drafts and on legacy events). struct PendingMail { - mail: tutasdk::entities::generated::tutanota::Mail, - details: Option, + mail: tutasdk::entities::generated::tutanota::Mail, + details: Option, } async fn apply_batch( - store: &MailStore, - local_store: &LocalStore, - backend: &dyn MailBackend, - sync_limit: usize, - batch: &EntityUpdateBatch, + store: &MailStore, + local_store: &LocalStore, + backend: &dyn MailBackend, + sync_limit: usize, + batch: &EntityUpdateBatch, ) { - let Bucketed { - mail_set_entry_creates, - mail_set_entry_deletes, - mail_creates, - mail_events, - folder_list_dirty, - } = bucket_updates(&batch.updates); + let Bucketed { + mail_set_entry_creates, + mail_set_entry_deletes, + mail_creates, + mail_events, + folder_list_dirty, + } = bucket_updates(&batch.updates); - // A MailSet event means the user added / renamed / deleted a folder in - // the webmail. Refresh the list first so a brand-new folder is known - // before we try to apply MailSetEntry events that reference it. - if folder_list_dirty { - refresh_folder_list(store, local_store, backend).await; - } + // A MailSet event means the user added / renamed / deleted a folder in + // the webmail. Refresh the list first so a brand-new folder is known + // before we try to apply MailSetEntry events that reference it. + if folder_list_dirty { + refresh_folder_list(store, local_store, backend).await; + } - // Snapshot the folder list once; the delta path matches events to - // folders by `entries_list_id`. - let folders = store.list_folders().await; - let folder_by_entries: HashMap<&str, &FolderInfo> = folders - .iter() - .map(|f| (f.entries_list_id.as_str(), f)) - .collect(); + // Snapshot the folder list once; the delta path matches events to + // folders by `entries_list_id`. + let folders = store.list_folders().await; + let folder_by_entries: HashMap<&str, &FolderInfo> = folders + .iter() + .map(|f| (f.entries_list_id.as_str(), f)) + .collect(); - // Folders we could not handle precisely — fall back to a full - // `sync_folder` at the end. - let mut fallback_folders: HashSet = HashSet::new(); + // Folders we could not handle precisely — fall back to a full + // `sync_folder` at the end. + let mut fallback_folders: HashSet = HashSet::new(); - // Pre-decrypt every Mail CREATE in this batch so its `event.instance` - // (and `event.blob_instance`, when present) gives us the Mail and its - // `MailDetails` without any REST call. The matching MailSetEntry CREATE - // below consumes from this pool; what's left is dropped at scope end. - let mut pending: HashMap = - predecrypt_mail_creates(backend, &mail_creates).await; + // Pre-decrypt every Mail CREATE in this batch so its `event.instance` + // (and `event.blob_instance`, when present) gives us the Mail and its + // `MailDetails` without any REST call. The matching MailSetEntry CREATE + // below consumes from this pool; what's left is dropped at scope end. + let mut pending: HashMap = + predecrypt_mail_creates(backend, &mail_creates).await; - // 1) MailSetEntry CREATEs first. Doing creates *before* the matching - // deletes lets a MOVE clone the already-decrypted Mail straight from - // the source folder (still present in the cache at this point) — no - // REST round-trip. - for ev in &mail_set_entry_creates { - apply_mail_set_entry_create( - store, - local_store, - backend, - &folder_by_entries, - &mut fallback_folders, - &mut pending, - ev, - ) - .await; - } + // 1) MailSetEntry CREATEs first. Doing creates *before* the matching + // deletes lets a MOVE clone the already-decrypted Mail straight from + // the source folder (still present in the cache at this point) — no + // REST round-trip. + for ev in &mail_set_entry_creates { + apply_mail_set_entry_create( + store, + local_store, + backend, + &folder_by_entries, + &mut fallback_folders, + &mut pending, + ev, + ) + .await; + } - // 2) MailSetEntry DELETEs. Per Tuta's wire model these arrive paired - // with the CREATEs (a MOVE = DELETE source + CREATE target in the same - // batch); a lone DELETE means a trash / hard-delete. - for ev in &mail_set_entry_deletes { - apply_mail_set_entry_delete(store, local_store, &folder_by_entries, ev).await; - } + // 2) MailSetEntry DELETEs. Per Tuta's wire model these arrive paired + // with the CREATEs (a MOVE = DELETE source + CREATE target in the same + // batch); a lone DELETE means a trash / hard-delete. + for ev in &mail_set_entry_deletes { + apply_mail_set_entry_delete(store, local_store, &folder_by_entries, ev).await; + } - // 3) Mail-entity events — UPDATE (read/unread, subject, …) and DELETE. - // CREATE on a Mail entity is paired with a MailSetEntry CREATE which - // the loop above already handled. - for ev in mail_events { - apply_mail_event(store, local_store, backend, ev).await; - } + // 3) Mail-entity events — UPDATE (read/unread, subject, …) and DELETE. + // CREATE on a Mail entity is paired with a MailSetEntry CREATE which + // the loop above already handled. + for ev in mail_events { + apply_mail_event(store, local_store, backend, ev).await; + } - // 4) Safety net: for every folder we couldn't precisely apply (decode - // failure, unknown folder, REST error during `load_mail`), re-run the - // classic full sync so the user never silently misses a mail. - for entries_list_id in &fallback_folders { - let Some(folder) = folder_by_entries.get(entries_list_id.as_str()).copied() else { - continue; - }; - debug!( - "Event bus: fallback full sync for {} (batch {})", - folder.imap_path, batch.batch_id - ); - if let Err(e) = sync_folder(store, local_store, backend, folder, sync_limit).await { - warn!( - "Event bus fallback sync failed for {}: {}", - folder.imap_path, e - ); - } - } + // 4) Safety net: for every folder we couldn't precisely apply (decode + // failure, unknown folder, REST error during `load_mail`), re-run the + // classic full sync so the user never silently misses a mail. + for entries_list_id in &fallback_folders { + let Some(folder) = folder_by_entries.get(entries_list_id.as_str()).copied() else { + continue; + }; + debug!( + "Event bus: fallback full sync for {} (batch {})", + folder.imap_path, batch.batch_id + ); + if let Err(e) = sync_folder(store, local_store, backend, folder, sync_limit).await { + warn!( + "Event bus fallback sync failed for {}: {}", + folder.imap_path, e + ); + } + } } async fn refresh_folder_list( - store: &MailStore, - local_store: &LocalStore, - backend: &dyn MailBackend, + store: &MailStore, + local_store: &LocalStore, + backend: &dyn MailBackend, ) { - match backend.list_folders().await { - Ok(folders) => { - let known: HashSet = folders.iter().map(|f| f.id.clone()).collect(); - store.set_folder_list(folders).await; - let removed = store.prune_unknown_folders(&known).await; - for fid in &removed { - debug!("Event bus: folder {} removed", fid); - match local_store.delete_folder_mails(fid) { - Ok(ids) => { - for eid in &ids { - if let Err(e) = local_store.delete_eml(eid) { - warn!("Failed to delete cached eml {}: {}", eid, e); - } - } - }, - Err(e) => warn!("Failed to delete folder cache {}: {}", fid, e), - } - } - }, - Err(e) => warn!("MailSet event: folder list refresh failed: {e}"), - } + match backend.list_folders().await { + Ok(folders) => { + let known: HashSet = folders.iter().map(|f| f.id.clone()).collect(); + store.set_folder_list(folders).await; + let removed = store.prune_unknown_folders(&known).await; + for fid in &removed { + debug!("Event bus: folder {} removed", fid); + match local_store.delete_folder_mails(fid) { + Ok(ids) => { + for eid in &ids { + if let Err(e) = local_store.delete_eml(eid) { + warn!("Failed to delete cached eml {}: {}", eid, e); + } + } + } + Err(e) => warn!("Failed to delete folder cache {}: {}", fid, e), + } + } + } + Err(e) => warn!("MailSet event: folder list refresh failed: {e}"), + } } async fn apply_mail_set_entry_create( - store: &MailStore, - local_store: &LocalStore, - backend: &dyn MailBackend, - folder_by_entries: &HashMap<&str, &FolderInfo>, - fallback_folders: &mut HashSet, - pending: &mut HashMap, - ev: &EntityUpdateEvent, + store: &MailStore, + local_store: &LocalStore, + backend: &dyn MailBackend, + folder_by_entries: &HashMap<&str, &FolderInfo>, + fallback_folders: &mut HashSet, + pending: &mut HashMap, + ev: &EntityUpdateEvent, ) { - let custom = CustomId(ev.instance_id.clone()); - let mail_eid = match mail_set_entry_id::deconstruct(&custom) { - Ok((_date, mail_id)) => mail_id.0, - Err(e) => { - warn!( - "MailSetEntry CREATE id {:?} could not be decoded: {e} — falling back to full sync", - ev.instance_id - ); - fallback_folders.insert(ev.instance_list_id.clone()); - return; - }, - }; - let Some(target_folder) = folder_by_entries.get(ev.instance_list_id.as_str()).copied() else { - // Folder unknown — typically a newly created custom folder whose - // `MailSet` event we have not yet processed. Falling back ensures - // we discover it via `list_folders` on the next batch. - fallback_folders.insert(ev.instance_list_id.clone()); - return; - }; + let custom = CustomId(ev.instance_id.clone()); + let mail_eid = match mail_set_entry_id::deconstruct(&custom) { + Ok((_date, mail_id)) => mail_id.0, + Err(e) => { + warn!( + "MailSetEntry CREATE id {:?} could not be decoded: {e} — falling back to full sync", + ev.instance_id + ); + fallback_folders.insert(ev.instance_list_id.clone()); + return; + } + }; + let Some(target_folder) = folder_by_entries.get(ev.instance_list_id.as_str()).copied() else { + // Folder unknown — typically a newly created custom folder whose + // `MailSet` event we have not yet processed. Falling back ensures + // we discover it via `list_folders` on the next batch. + fallback_folders.insert(ev.instance_list_id.clone()); + return; + }; - // HIT path: the mail already lives in another cached folder (typical - // MOVE between two known folders). Clone the StoredMail into the - // target, allocate a fresh UID and persist. - if let Some((source_folder, mut stored)) = store.find_mail_anywhere(&mail_eid).await { - debug!( - "Event bus: cloning mail {} from {} → {} (no REST)", - mail_eid, source_folder, target_folder.imap_path - ); - assign_uid_and_upsert(store, local_store, target_folder, &mail_eid, &mut stored).await; - return; - } + // HIT path: the mail already lives in another cached folder (typical + // MOVE between two known folders). Clone the StoredMail into the + // target, allocate a fresh UID and persist. + if let Some((source_folder, mut stored)) = store.find_mail_anywhere(&mail_eid).await { + debug!( + "Event bus: cloning mail {} from {} → {} (no REST)", + mail_eid, source_folder, target_folder.imap_path + ); + assign_uid_and_upsert(store, local_store, target_folder, &mail_eid, &mut stored).await; + return; + } - // HIT-FROM-BATCH path: a matching Mail CREATE rode in the same batch - // and we already inline-decrypted it (and possibly its MailDetails blob) - // into `pending`. Use that directly — no REST, and if we got the blob - // too the body is already RFC2822-rendered and written to disk so the - // prefetch loop has nothing left to do for this mail. - if let Some(PendingMail { mail, details }) = pending.remove(&mail_eid) { - debug!( - "Event bus: applying inline Mail CREATE {} → {} (no REST{})", - mail_eid, - target_folder.imap_path, - if details.is_some() { ", body inline too" } else { "" }, - ); - // Render the inline body to an RFC 2822 envelope only when the mail - // has zero attachments — those need a separate blob download per - // File that the event-bus handler cannot do synchronously. Mails - // with attachments stay at `has_details=0` so the prefetch loop - // picks them up and emits a proper `multipart/mixed` envelope. - let has_attachments = !mail.attachments.is_empty(); - let rfc2822 = details.as_ref().and_then(|d| { - if has_attachments { - None - } else { - Some(crate::mail::mail_to_rfc2822(&mail, Some(d), &[])) - } - }); - let mut stored = StoredMail { - mail, - details: details.clone(), - rfc2822: rfc2822.clone(), - uid: 0, - attachments_pending: false, - }; - // Persist the body straight away so the IMAP layer can serve FETCH - // BODY[] without a placeholder, and the prefetch loop skips it. - if let Some(eml) = rfc2822.as_deref() { - if let Err(e) = local_store.write_eml(&mail_eid, eml) { - warn!("Failed to cache eml for {}: {e}", mail_eid); - } else if let Err(e) = local_store.mark_has_details(&mail_eid) { - warn!("Failed to mark has_details for {}: {e}", mail_eid); - } - } - assign_uid_and_upsert(store, local_store, target_folder, &mail_eid, &mut stored).await; - return; - } + // HIT-FROM-BATCH path: a matching Mail CREATE rode in the same batch + // and we already inline-decrypted it (and possibly its MailDetails blob) + // into `pending`. Use that directly — no REST, and if we got the blob + // too the body is already RFC2822-rendered and written to disk so the + // prefetch loop has nothing left to do for this mail. + if let Some(PendingMail { mail, details }) = pending.remove(&mail_eid) { + debug!( + "Event bus: applying inline Mail CREATE {} → {} (no REST{})", + mail_eid, + target_folder.imap_path, + if details.is_some() { + ", body inline too" + } else { + "" + }, + ); + // Render the inline body to an RFC 2822 envelope only when the mail + // has zero attachments — those need a separate blob download per + // File that the event-bus handler cannot do synchronously. Mails + // with attachments stay at `has_details=0` so the prefetch loop + // picks them up and emits a proper `multipart/mixed` envelope. + let has_attachments = !mail.attachments.is_empty(); + let rfc2822 = details.as_ref().and_then(|d| { + if has_attachments { + None + } else { + Some(crate::mail::mail_to_rfc2822(&mail, Some(d), &[])) + } + }); + let mut stored = StoredMail { + mail, + details: details.clone(), + rfc2822: rfc2822.clone(), + uid: 0, + attachments_pending: false, + }; + // Persist the body straight away so the IMAP layer can serve FETCH + // BODY[] without a placeholder, and the prefetch loop skips it. + if let Some(eml) = rfc2822.as_deref() { + if let Err(e) = local_store.write_eml(&mail_eid, eml) { + warn!("Failed to cache eml for {}: {e}", mail_eid); + } else if let Err(e) = local_store.mark_has_details(&mail_eid) { + warn!("Failed to mark has_details for {}: {e}", mail_eid); + } + } + assign_uid_and_upsert(store, local_store, target_folder, &mail_eid, &mut stored).await; + return; + } - // MISS path: never seen this mail. The MailSetEntry payload is - // inline in `event.instance`; decrypting it gives us the referenced - // Mail's full `(list_id, element_id)` directly, so we can ask the - // backend for just that one mail. No global mail-list-id cache, no - // `sync_folder` fallback needed in the typical case. - let mail_id_tuple = resolve_mail_set_entry(backend, ev).await; - let Some(mail_tuple) = mail_id_tuple else { - // Inline decode failed AND we have no cached mail to clone — the - // folder will be brought up-to-date by the safety-net sync below. - fallback_folders.insert(ev.instance_list_id.clone()); - return; - }; - let list_id = mail_tuple.list_id.to_string(); - let elem_id = mail_tuple.element_id.to_string(); - match backend.load_mail(&list_id, &elem_id).await { - Ok(Some(mail)) => { - debug!( - "Event bus: targeted load_mail({}, {}) → {} (1 REST call)", - list_id, elem_id, target_folder.imap_path - ); - let mut stored = StoredMail { - mail, - details: None, - rfc2822: None, - uid: 0, - attachments_pending: false, - }; - assign_uid_and_upsert(store, local_store, target_folder, &elem_id, &mut stored).await; - }, - Ok(None) => { - debug!("MailSetEntry CREATE: mail {} not found on server", elem_id); - }, - Err(e) => { - warn!( - "MailSetEntry CREATE: load_mail({}, {}) failed: {e} — falling back", - list_id, elem_id - ); - fallback_folders.insert(ev.instance_list_id.clone()); - }, - } + // MISS path: never seen this mail. The MailSetEntry payload is + // inline in `event.instance`; decrypting it gives us the referenced + // Mail's full `(list_id, element_id)` directly, so we can ask the + // backend for just that one mail. No global mail-list-id cache, no + // `sync_folder` fallback needed in the typical case. + let mail_id_tuple = resolve_mail_set_entry(backend, ev).await; + let Some(mail_tuple) = mail_id_tuple else { + // Inline decode failed AND we have no cached mail to clone — the + // folder will be brought up-to-date by the safety-net sync below. + fallback_folders.insert(ev.instance_list_id.clone()); + return; + }; + let list_id = mail_tuple.list_id.to_string(); + let elem_id = mail_tuple.element_id.to_string(); + match backend.load_mail(&list_id, &elem_id).await { + Ok(Some(mail)) => { + debug!( + "Event bus: targeted load_mail({}, {}) → {} (1 REST call)", + list_id, elem_id, target_folder.imap_path + ); + let mut stored = StoredMail { + mail, + details: None, + rfc2822: None, + uid: 0, + attachments_pending: false, + }; + assign_uid_and_upsert(store, local_store, target_folder, &elem_id, &mut stored).await; + } + Ok(None) => { + debug!("MailSetEntry CREATE: mail {} not found on server", elem_id); + } + Err(e) => { + warn!( + "MailSetEntry CREATE: load_mail({}, {}) failed: {e} — falling back", + list_id, elem_id + ); + fallback_folders.insert(ev.instance_list_id.clone()); + } + } } /// Try to discover the Mail referenced by a `MailSetEntry` CREATE event @@ -412,107 +419,107 @@ async fn apply_mail_set_entry_create( /// in-memory cache by entry-id-derived mail id — if neither works, the /// caller queues a fallback `sync_folder`. async fn resolve_mail_set_entry( - backend: &dyn MailBackend, - ev: &EntityUpdateEvent, + backend: &dyn MailBackend, + ev: &EntityUpdateEvent, ) -> Option { - if let Some(json) = ev.instance.as_deref() { - match backend.decrypt_inline_mail_set_entry(json).await { - Ok(Some(entry)) => return Some(entry.mail), - Ok(None) => debug!("MailSetEntry inline: session key unresolved"), - Err(e) => warn!("MailSetEntry inline decrypt failed: {e}"), - } - } - None + if let Some(json) = ev.instance.as_deref() { + match backend.decrypt_inline_mail_set_entry(json).await { + Ok(Some(entry)) => return Some(entry.mail), + Ok(None) => debug!("MailSetEntry inline: session key unresolved"), + Err(e) => warn!("MailSetEntry inline decrypt failed: {e}"), + } + } + None } async fn apply_mail_set_entry_delete( - store: &MailStore, - local_store: &LocalStore, - folder_by_entries: &HashMap<&str, &FolderInfo>, - ev: &EntityUpdateEvent, + store: &MailStore, + local_store: &LocalStore, + folder_by_entries: &HashMap<&str, &FolderInfo>, + ev: &EntityUpdateEvent, ) { - let custom = CustomId(ev.instance_id.clone()); - let mail_eid = match mail_set_entry_id::deconstruct(&custom) { - Ok((_date, mail_id)) => mail_id.0, - Err(e) => { - // We cannot identify *which* mail left the folder without the - // decoded id; an upstream MailSetEntry CREATE may have queued a - // fallback already, otherwise the next periodic interaction - // (re-select, FETCH) will reconcile. - warn!( - "MailSetEntry DELETE id {:?} could not be decoded: {e}", - ev.instance_id - ); - return; - }, - }; - let Some(source_folder) = folder_by_entries.get(ev.instance_list_id.as_str()).copied() else { - return; - }; - let removed = store - .remove_mail_from_folder(&source_folder.id, &mail_eid) - .await; - if removed { - debug!( - "Event bus: removed mail {} from {} (no REST)", - mail_eid, source_folder.imap_path - ); - } - // Drop the on-disk row + `.eml` only if no folder still holds the - // mail. Multi-folder placement (rare with the current Tuta model) and - // MOVE-within-batch (the matching CREATE ran first, so the target - // folder still has it) are both preserved by this check. - if !store.is_mail_anywhere(&mail_eid).await { - if let Err(e) = local_store.delete_mail(&mail_eid) { - warn!("Failed to delete cached mail {}: {}", mail_eid, e); - } - } + let custom = CustomId(ev.instance_id.clone()); + let mail_eid = match mail_set_entry_id::deconstruct(&custom) { + Ok((_date, mail_id)) => mail_id.0, + Err(e) => { + // We cannot identify *which* mail left the folder without the + // decoded id; an upstream MailSetEntry CREATE may have queued a + // fallback already, otherwise the next periodic interaction + // (re-select, FETCH) will reconcile. + warn!( + "MailSetEntry DELETE id {:?} could not be decoded: {e}", + ev.instance_id + ); + return; + } + }; + let Some(source_folder) = folder_by_entries.get(ev.instance_list_id.as_str()).copied() else { + return; + }; + let removed = store + .remove_mail_from_folder(&source_folder.id, &mail_eid) + .await; + if removed { + debug!( + "Event bus: removed mail {} from {} (no REST)", + mail_eid, source_folder.imap_path + ); + } + // Drop the on-disk row + `.eml` only if no folder still holds the + // mail. Multi-folder placement (rare with the current Tuta model) and + // MOVE-within-batch (the matching CREATE ran first, so the target + // folder still has it) are both preserved by this check. + if !store.is_mail_anywhere(&mail_eid).await { + if let Err(e) = local_store.delete_mail(&mail_eid) { + warn!("Failed to delete cached mail {}: {}", mail_eid, e); + } + } } async fn apply_mail_event( - store: &MailStore, - local_store: &LocalStore, - backend: &dyn MailBackend, - ev: &EntityUpdateEvent, + store: &MailStore, + local_store: &LocalStore, + backend: &dyn MailBackend, + ev: &EntityUpdateEvent, ) { - match ev.operation { - Operation::Delete => { - if let Err(e) = local_store.delete_mail(&ev.instance_id) { - warn!("Failed to delete cached mail {}: {}", ev.instance_id, e); - } - store.remove_mail_everywhere(&ev.instance_id).await; - }, - Operation::Update => { - // Prefer the inline-decrypt path: the encrypted Mail is already - // in `event.instance`, no REST round-trip needed. Fall back to - // `load_mail` if the payload is absent or its session key is in a - // transient unresolvable state (e.g. post-reply attachment keys). - let mail = resolve_mail(backend, ev).await; - match mail { - Some(mail) => { - store.refresh_mail_in_place(&mail).await; - let mail_json = serde_json::to_string(&mail).unwrap_or_default(); - if let Err(e) = local_store.refresh_mail_fields( - &ev.instance_id, - &mail.subject, - &mail.sender.name, - &mail.sender.address, - mail.unread, - &mail_json, - ) { - debug!("Could not refresh metadata for {}: {}", ev.instance_id, e); - } - }, - None => { - // Disappeared / unresolvable — treat like a delete to - // keep the cache consistent. - let _ = local_store.delete_mail(&ev.instance_id); - store.remove_mail_everywhere(&ev.instance_id).await; - }, - } - }, - Operation::Create | Operation::Other(_) => {}, - } + match ev.operation { + Operation::Delete => { + if let Err(e) = local_store.delete_mail(&ev.instance_id) { + warn!("Failed to delete cached mail {}: {}", ev.instance_id, e); + } + store.remove_mail_everywhere(&ev.instance_id).await; + } + Operation::Update => { + // Prefer the inline-decrypt path: the encrypted Mail is already + // in `event.instance`, no REST round-trip needed. Fall back to + // `load_mail` if the payload is absent or its session key is in a + // transient unresolvable state (e.g. post-reply attachment keys). + let mail = resolve_mail(backend, ev).await; + match mail { + Some(mail) => { + store.refresh_mail_in_place(&mail).await; + let mail_json = serde_json::to_string(&mail).unwrap_or_default(); + if let Err(e) = local_store.refresh_mail_fields( + &ev.instance_id, + &mail.subject, + &mail.sender.name, + &mail.sender.address, + mail.unread, + &mail_json, + ) { + debug!("Could not refresh metadata for {}: {}", ev.instance_id, e); + } + } + None => { + // Disappeared / unresolvable — treat like a delete to + // keep the cache consistent. + let _ = local_store.delete_mail(&ev.instance_id); + store.remove_mail_everywhere(&ev.instance_id).await; + } + } + } + Operation::Create | Operation::Other(_) => {} + } } /// Pre-decrypt every Mail CREATE event in a batch into a `(eid -> Mail + details)` @@ -522,244 +529,289 @@ async fn apply_mail_event( /// event leaves the pool entry absent — the MailSetEntry handler then /// falls back to `load_mail`, so nothing is silently lost. async fn predecrypt_mail_creates( - backend: &dyn MailBackend, - events: &[&EntityUpdateEvent], + backend: &dyn MailBackend, + events: &[&EntityUpdateEvent], ) -> HashMap { - let mut out: HashMap = HashMap::with_capacity(events.len()); - for ev in events { - let Some(json) = ev.instance.as_deref() else { - continue; - }; - let mail = match backend.decrypt_inline_mail(json).await { - Ok(Some(m)) => m, - Ok(None) => continue, // session key transient — caller will REST-fallback - Err(e) => { - warn!( - "predecrypt Mail CREATE {}: decrypt_inline_mail failed: {e}", - ev.instance_id - ); - continue; - }, - }; - let details = match ev.blob_instance.as_deref() { - Some(blob_json) => match backend.decrypt_inline_mail_details_blob(blob_json).await { - Ok(opt) => opt, - Err(e) => { - warn!( + let mut out: HashMap = HashMap::with_capacity(events.len()); + for ev in events { + let Some(json) = ev.instance.as_deref() else { + continue; + }; + let mail = match backend.decrypt_inline_mail(json).await { + Ok(Some(m)) => m, + Ok(None) => continue, // session key transient — caller will REST-fallback + Err(e) => { + warn!( + "predecrypt Mail CREATE {}: decrypt_inline_mail failed: {e}", + ev.instance_id + ); + continue; + } + }; + let details = match ev.blob_instance.as_deref() { + Some(blob_json) => match backend.decrypt_inline_mail_details_blob(blob_json).await { + Ok(opt) => opt, + Err(e) => { + warn!( "predecrypt Mail CREATE {}: blob decrypt failed: {e} — keeping mail without body", ev.instance_id ); - None - }, - }, - None => None, - }; - out.insert(ev.instance_id.clone(), PendingMail { mail, details }); - } - out + None + } + }, + None => None, + }; + out.insert(ev.instance_id.clone(), PendingMail { mail, details }); + } + out } /// Decrypt-inline first, REST-fallback second. Centralises the policy so /// every event-bus consumer takes the same fast path when the payload is /// already inline, and the same safety-net otherwise. async fn resolve_mail( - backend: &dyn MailBackend, - ev: &EntityUpdateEvent, + backend: &dyn MailBackend, + ev: &EntityUpdateEvent, ) -> Option { - if let Some(json) = ev.instance.as_deref() { - match backend.decrypt_inline_mail(json).await { - Ok(Some(mail)) => { - debug!( - "Event bus: decrypted Mail {} inline (no REST)", - ev.instance_id - ); - return Some(mail); - }, - Ok(None) => { - debug!( - "Event bus: Mail {} session key unresolvable, falling back to load_mail", - ev.instance_id - ); - }, - Err(e) => { - warn!( - "Event bus: decrypt_inline_mail({}) failed: {e} — falling back", - ev.instance_id - ); - }, - } - } - match backend - .load_mail(&ev.instance_list_id, &ev.instance_id) - .await - { - Ok(opt) => opt, - Err(e) => { - warn!("Event bus: load_mail({}) failed: {e}", ev.instance_id); - None - }, - } + if let Some(json) = ev.instance.as_deref() { + match backend.decrypt_inline_mail(json).await { + Ok(Some(mail)) => { + debug!( + "Event bus: decrypted Mail {} inline (no REST)", + ev.instance_id + ); + return Some(mail); + } + Ok(None) => { + debug!( + "Event bus: Mail {} session key unresolvable, falling back to load_mail", + ev.instance_id + ); + } + Err(e) => { + warn!( + "Event bus: decrypt_inline_mail({}) failed: {e} — falling back", + ev.instance_id + ); + } + } + } + match backend + .load_mail(&ev.instance_list_id, &ev.instance_id) + .await + { + Ok(opt) => opt, + Err(e) => { + warn!("Event bus: load_mail({}) failed: {e}", ev.instance_id); + None + } + } } /// Allocate a fresh UID in `target_folder`, stamp it on `stored`, then /// upsert both `MailStore` and the `LocalStore` metadata row in one step. async fn assign_uid_and_upsert( - store: &MailStore, - local_store: &LocalStore, - target_folder: &FolderInfo, - mail_eid: &str, - stored: &mut StoredMail, + store: &MailStore, + local_store: &LocalStore, + target_folder: &FolderInfo, + mail_eid: &str, + stored: &mut StoredMail, ) { - let uid = match local_store.allocate_folder_uids(&target_folder.id, &[mail_eid]) { - Ok(map) => map.get(mail_eid).copied().unwrap_or(0), - Err(e) => { - warn!( - "Failed to allocate UID for {} in {}: {e}", - mail_eid, target_folder.imap_path - ); - 0 - }, - }; - stored.uid = uid; + let uid = match local_store.allocate_folder_uids(&target_folder.id, &[mail_eid]) { + Ok(map) => map.get(mail_eid).copied().unwrap_or(0), + Err(e) => { + warn!( + "Failed to allocate UID for {} in {}: {e}", + mail_eid, target_folder.imap_path + ); + 0 + } + }; + stored.uid = uid; - let meta = crate::sync::mail_to_metadata(&stored.mail, &target_folder.id, uid); - if let Err(e) = local_store.upsert_mail_metadata(&meta) { - warn!("Failed to persist {} in {}: {e}", mail_eid, target_folder.id); - } + let meta = crate::sync::mail_to_metadata(&stored.mail, &target_folder.id, uid); + if let Err(e) = local_store.upsert_mail_metadata(&meta) { + warn!( + "Failed to persist {} in {}: {e}", + mail_eid, target_folder.id + ); + } - store - .upsert_mail_in_folder(&target_folder.id, stored.clone()) - .await; + store + .upsert_mail_in_folder(&target_folder.id, stored.clone()) + .await; } #[cfg(test)] mod tests { - use super::*; + use super::*; - fn ev(app: &str, type_id: i64, list: &str, elem: &str, op: Operation) -> EntityUpdateEvent { - EntityUpdateEvent { - application: app.to_string(), - type_id, - instance_list_id: list.to_string(), - instance_id: elem.to_string(), - operation: op, - instance: None, - blob_instance: None, - } - } + fn ev(app: &str, type_id: i64, list: &str, elem: &str, op: Operation) -> EntityUpdateEvent { + EntityUpdateEvent { + application: app.to_string(), + type_id, + instance_list_id: list.to_string(), + instance_id: elem.to_string(), + operation: op, + instance: None, + blob_instance: None, + } + } - #[test] - fn bucket_empty_batch() { - let out = bucket_updates(&[]); - assert!(out.mail_set_entry_creates.is_empty()); - assert!(out.mail_set_entry_deletes.is_empty()); - assert!(out.mail_events.is_empty()); - assert!(!out.folder_list_dirty); - } + #[test] + fn bucket_empty_batch() { + let out = bucket_updates(&[]); + assert!(out.mail_set_entry_creates.is_empty()); + assert!(out.mail_set_entry_deletes.is_empty()); + assert!(out.mail_events.is_empty()); + assert!(!out.folder_list_dirty); + } - #[test] - fn bucket_ignores_other_applications() { - let updates = vec![ev("sys", 97, "L", "E", Operation::Create)]; - let out = bucket_updates(&updates); - assert!(out.mail_set_entry_creates.is_empty()); - assert!(out.mail_set_entry_deletes.is_empty()); - assert!(out.mail_events.is_empty()); - } + #[test] + fn bucket_ignores_other_applications() { + let updates = vec![ev("sys", 97, "L", "E", Operation::Create)]; + let out = bucket_updates(&updates); + assert!(out.mail_set_entry_creates.is_empty()); + assert!(out.mail_set_entry_deletes.is_empty()); + assert!(out.mail_events.is_empty()); + } - #[test] - fn bucket_ignores_unknown_type_ids_in_tutanota() { - // Unrelated tutanota entities (attachments, contacts, …) pass through. - let updates = vec![ev("tutanota", 999, "L", "E", Operation::Update)]; - let out = bucket_updates(&updates); - assert!(out.mail_set_entry_creates.is_empty()); - assert!(out.mail_set_entry_deletes.is_empty()); - assert!(out.mail_events.is_empty()); - } + #[test] + fn bucket_ignores_unknown_type_ids_in_tutanota() { + // Unrelated tutanota entities (attachments, contacts, …) pass through. + let updates = vec![ev("tutanota", 999, "L", "E", Operation::Update)]; + let out = bucket_updates(&updates); + assert!(out.mail_set_entry_creates.is_empty()); + assert!(out.mail_set_entry_deletes.is_empty()); + assert!(out.mail_events.is_empty()); + } - #[test] - fn bucket_splits_mail_set_entry_creates_and_deletes() { - let updates = vec![ - ev("tutanota", MAIL_SET_ENTRY_TYPE_ID, "inbox_entries", "e1", Operation::Create), - ev("tutanota", MAIL_SET_ENTRY_TYPE_ID, "source_entries", "e2", Operation::Delete), - ev("tutanota", MAIL_SET_ENTRY_TYPE_ID, "sent_entries", "e3", Operation::Create), - ]; - let out = bucket_updates(&updates); - assert_eq!(out.mail_set_entry_creates.len(), 2); - assert_eq!(out.mail_set_entry_deletes.len(), 1); - // Order within each bucket is preserved (Tuta guarantees batch order). - assert_eq!(out.mail_set_entry_creates[0].instance_id, "e1"); - assert_eq!(out.mail_set_entry_creates[1].instance_id, "e3"); - assert_eq!(out.mail_set_entry_deletes[0].instance_id, "e2"); - } + #[test] + fn bucket_splits_mail_set_entry_creates_and_deletes() { + let updates = vec![ + ev( + "tutanota", + MAIL_SET_ENTRY_TYPE_ID, + "inbox_entries", + "e1", + Operation::Create, + ), + ev( + "tutanota", + MAIL_SET_ENTRY_TYPE_ID, + "source_entries", + "e2", + Operation::Delete, + ), + ev( + "tutanota", + MAIL_SET_ENTRY_TYPE_ID, + "sent_entries", + "e3", + Operation::Create, + ), + ]; + let out = bucket_updates(&updates); + assert_eq!(out.mail_set_entry_creates.len(), 2); + assert_eq!(out.mail_set_entry_deletes.len(), 1); + // Order within each bucket is preserved (Tuta guarantees batch order). + assert_eq!(out.mail_set_entry_creates[0].instance_id, "e1"); + assert_eq!(out.mail_set_entry_creates[1].instance_id, "e3"); + assert_eq!(out.mail_set_entry_deletes[0].instance_id, "e2"); + } - #[test] - fn bucket_ignores_mail_set_entry_update_operations() { - // MailSetEntry is immutable per Tuta's model; UPDATE shouldn't - // happen, but if one ever sneaks through we ignore it rather than - // crash. - let updates = vec![ev( - "tutanota", - MAIL_SET_ENTRY_TYPE_ID, - "inbox_entries", - "e1", - Operation::Update, - )]; - let out = bucket_updates(&updates); - assert!(out.mail_set_entry_creates.is_empty()); - assert!(out.mail_set_entry_deletes.is_empty()); - } + #[test] + fn bucket_ignores_mail_set_entry_update_operations() { + // MailSetEntry is immutable per Tuta's model; UPDATE shouldn't + // happen, but if one ever sneaks through we ignore it rather than + // crash. + let updates = vec![ev( + "tutanota", + MAIL_SET_ENTRY_TYPE_ID, + "inbox_entries", + "e1", + Operation::Update, + )]; + let out = bucket_updates(&updates); + assert!(out.mail_set_entry_creates.is_empty()); + assert!(out.mail_set_entry_deletes.is_empty()); + } - #[test] - fn bucket_collects_mail_events_in_order() { - let updates = vec![ - ev("tutanota", MAIL_TYPE_ID, "mailL", "m1", Operation::Update), - ev("tutanota", MAIL_TYPE_ID, "mailL", "m2", Operation::Delete), - ]; - let out = bucket_updates(&updates); - assert_eq!(out.mail_events.len(), 2); - assert!(out.mail_creates.is_empty(), "no CREATE here"); - assert_eq!(out.mail_events[0].instance_id, "m1"); - assert_eq!(out.mail_events[1].operation, Operation::Delete); - } + #[test] + fn bucket_collects_mail_events_in_order() { + let updates = vec![ + ev("tutanota", MAIL_TYPE_ID, "mailL", "m1", Operation::Update), + ev("tutanota", MAIL_TYPE_ID, "mailL", "m2", Operation::Delete), + ]; + let out = bucket_updates(&updates); + assert_eq!(out.mail_events.len(), 2); + assert!(out.mail_creates.is_empty(), "no CREATE here"); + assert_eq!(out.mail_events[0].instance_id, "m1"); + assert_eq!(out.mail_events[1].operation, Operation::Delete); + } - #[test] - fn bucket_routes_mail_create_to_its_own_bucket() { - // Mail CREATEs are split out of `mail_events` so the inline pool - // can pre-decrypt them before MailSetEntry CREATEs consume them. - let updates = vec![ - ev("tutanota", MAIL_TYPE_ID, "mailL", "newM", Operation::Create), - ev("tutanota", MAIL_TYPE_ID, "mailL", "readM", Operation::Update), - ]; - let out = bucket_updates(&updates); - assert_eq!(out.mail_creates.len(), 1); - assert_eq!(out.mail_creates[0].instance_id, "newM"); - assert_eq!(out.mail_events.len(), 1); - assert_eq!(out.mail_events[0].instance_id, "readM"); - } + #[test] + fn bucket_routes_mail_create_to_its_own_bucket() { + // Mail CREATEs are split out of `mail_events` so the inline pool + // can pre-decrypt them before MailSetEntry CREATEs consume them. + let updates = vec![ + ev("tutanota", MAIL_TYPE_ID, "mailL", "newM", Operation::Create), + ev( + "tutanota", + MAIL_TYPE_ID, + "mailL", + "readM", + Operation::Update, + ), + ]; + let out = bucket_updates(&updates); + assert_eq!(out.mail_creates.len(), 1); + assert_eq!(out.mail_creates[0].instance_id, "newM"); + assert_eq!(out.mail_events.len(), 1); + assert_eq!(out.mail_events[0].instance_id, "readM"); + } - #[test] - fn bucket_marks_folder_list_dirty_on_mail_set_event() { - let updates = vec![ - ev("tutanota", MAIL_SET_TYPE_ID, "folderL", "f1", Operation::Create), - ev("tutanota", MAIL_SET_TYPE_ID, "folderL", "f2", Operation::Delete), - ]; - let out = bucket_updates(&updates); - assert!(out.folder_list_dirty); - assert!(out.mail_set_entry_creates.is_empty()); - assert!(out.mail_set_entry_deletes.is_empty()); - assert!(out.mail_events.is_empty()); - } + #[test] + fn bucket_marks_folder_list_dirty_on_mail_set_event() { + let updates = vec![ + ev( + "tutanota", + MAIL_SET_TYPE_ID, + "folderL", + "f1", + Operation::Create, + ), + ev( + "tutanota", + MAIL_SET_TYPE_ID, + "folderL", + "f2", + Operation::Delete, + ), + ]; + let out = bucket_updates(&updates); + assert!(out.folder_list_dirty); + assert!(out.mail_set_entry_creates.is_empty()); + assert!(out.mail_set_entry_deletes.is_empty()); + assert!(out.mail_events.is_empty()); + } - #[test] - fn bucket_mixed_batch() { - let updates = vec![ - ev("tutanota", MAIL_SET_ENTRY_TYPE_ID, "inbox_entries", "e1", Operation::Create), - ev("tutanota", MAIL_TYPE_ID, "mailL", "m1", Operation::Update), - ev("sys", 42, "X", "Y", Operation::Create), // ignored - ]; - let out = bucket_updates(&updates); - assert_eq!(out.mail_set_entry_creates.len(), 1); - assert_eq!(out.mail_events.len(), 1); - assert!(!out.folder_list_dirty); - } + #[test] + fn bucket_mixed_batch() { + let updates = vec![ + ev( + "tutanota", + MAIL_SET_ENTRY_TYPE_ID, + "inbox_entries", + "e1", + Operation::Create, + ), + ev("tutanota", MAIL_TYPE_ID, "mailL", "m1", Operation::Update), + ev("sys", 42, "X", "Y", Operation::Create), // ignored + ]; + let out = bucket_updates(&updates); + assert_eq!(out.mail_set_entry_creates.len(), 1); + assert_eq!(out.mail_events.len(), 1); + assert!(!out.folder_list_dirty); + } } diff --git a/crates/bridge/src/imap/mod.rs b/crates/bridge/src/imap/mod.rs index ef5d70a..3dcbcb4 100644 --- a/crates/bridge/src/imap/mod.rs +++ b/crates/bridge/src/imap/mod.rs @@ -1,8 +1,8 @@ mod session; mod utf7; +use log::{debug, error, info}; use std::sync::Arc; -use log::{info, error, debug}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::TcpListener; use tokio::sync::watch; @@ -56,7 +56,9 @@ async fn handle_connection( let mut store_watch: watch::Receiver = store.subscribe(); let mut session = ImapSession::new(store, backend, password_hash); - writer.write_all(b"* OK TutaBridge IMAP4rev1 ready\r\n").await?; + writer + .write_all(b"* OK TutaBridge IMAP4rev1 ready\r\n") + .await?; writer.flush().await?; let mut line = String::new(); diff --git a/crates/bridge/src/imap/session.rs b/crates/bridge/src/imap/session.rs index de68876..a1166aa 100644 --- a/crates/bridge/src/imap/session.rs +++ b/crates/bridge/src/imap/session.rs @@ -1,9 +1,9 @@ +use log::{debug, info}; use std::sync::Arc; -use log::{info, debug}; use tutasdk::entities::generated::tutanota::{Mail, MailDetails}; -use crate::mail::rfc2822::{extract_headers, format_internal_date}; use crate::mail::mail_to_rfc2822; +use crate::mail::rfc2822::{extract_headers, format_internal_date}; use crate::sync::MailStore; use crate::tuta::{FolderInfo, MailBackend}; @@ -72,18 +72,16 @@ impl ImapSession { None => return vec![], }; - let decoded = match base64::Engine::decode( - &base64::engine::general_purpose::STANDARD, - line.trim(), - ) { - Ok(d) => d, - Err(_) => { - return vec![format!( - "{} NO [AUTHENTICATIONFAILED] Invalid base64\r\n", - tag - )]; - } - }; + let decoded = + match base64::Engine::decode(&base64::engine::general_purpose::STANDARD, line.trim()) { + Ok(d) => d, + Err(_) => { + return vec![format!( + "{} NO [AUTHENTICATIONFAILED] Invalid base64\r\n", + tag + )]; + } + }; // PLAIN format: \0authcid\0password (authzid is empty) let parts: Vec<&[u8]> = decoded.splitn(3, |&b| b == 0).collect(); @@ -289,7 +287,10 @@ impl ImapSession { let folder = match self.store.folder_by_imap_path(&folder_name).await { Some(f) => f, None => { - return vec![format!("{} NO [NONEXISTENT] Mailbox does not exist\r\n", tag)]; + return vec![format!( + "{} NO [NONEXISTENT] Mailbox does not exist\r\n", + tag + )]; } }; let folder_id = folder.id.clone(); @@ -660,9 +661,10 @@ impl ImapSession { } let details = sm.details.or(old_details); - let rfc2822 = sm.rfc2822.or(old_rfc).unwrap_or_else(|| { - mail_to_rfc2822(&sm.mail, details.as_ref(), &[]) - }); + let rfc2822 = sm + .rfc2822 + .or(old_rfc) + .unwrap_or_else(|| mail_to_rfc2822(&sm.mail, details.as_ref(), &[])); self.mails.push(CachedMail { mail: sm.mail, @@ -673,7 +675,11 @@ impl ImapSession { }); } - debug!("Refreshed {} mails for {} from store", self.mails.len(), folder_id); + debug!( + "Refreshed {} mails for {} from store", + self.mails.len(), + folder_id + ); Ok(()) } @@ -717,7 +723,6 @@ impl ImapSession { None } } - } fn build_fetch_response(seq: usize, cached: &CachedMail, items: &str, uid_mode: bool) -> String { @@ -817,9 +822,7 @@ fn build_envelope(cached: &CachedMail) -> String { .map(|id| format!("<{}.{}@tutabridge.local>", id.list_id, id.element_id)) .unwrap_or_default(); - format!( - "(\"{date}\" \"{subject}\" ({from}) ({from}) ({from}) ({to}) NIL NIL NIL \"{msg_id}\")" - ) + format!("(\"{date}\" \"{subject}\" ({from}) ({from}) ({from}) ({to}) NIL NIL NIL \"{msg_id}\")") } fn imap_quote(s: &str) -> String { @@ -904,7 +907,10 @@ fn parse_imap_token(s: &str) -> (String, &str) { (result, "") } else { let end = s.find(char::is_whitespace).unwrap_or(s.len()); - (s[..end].to_string(), if end < s.len() { &s[end..] } else { "" }) + ( + s[..end].to_string(), + if end < s.len() { &s[end..] } else { "" }, + ) } } @@ -1201,10 +1207,7 @@ mod tests { contact: None, _errors: Default::default(), }; - assert_eq!( - format_envelope_addr(&addr), - "(NIL NIL \"localonly\" \"\")" - ); + assert_eq!(format_envelope_addr(&addr), "(NIL NIL \"localonly\" \"\")"); } #[test] @@ -1218,10 +1221,7 @@ mod tests { }; let result = format_envelope_addr(&addr); // Inner quotes replaced with single quotes, wrapped in IMAP string delimiters - assert_eq!( - result, - "(\"John 'JD' Doe\" NIL \"john\" \"example.com\")" - ); + assert_eq!(result, "(\"John 'JD' Doe\" NIL \"john\" \"example.com\")"); } // --- extract_headers (via rfc2822 module) --- @@ -1324,10 +1324,10 @@ mod tests { // Integration tests with MockBackend // ================================================================= - use std::sync::Mutex; + use crate::mail::ParsedMessage; use crate::sync::MailStore; use crate::tuta::MailBackend; - use crate::mail::ParsedMessage; + use std::sync::Mutex; use tutasdk::entities::generated::tutanota::{Body, Recipients}; struct MockBackend { @@ -1358,7 +1358,10 @@ mod tests { } fn add_details(&self, element_id: &str, details: MailDetails) { - self.details.lock().unwrap().insert(element_id.to_string(), details); + self.details + .lock() + .unwrap() + .insert(element_id.to_string(), details); } } @@ -1396,8 +1399,20 @@ mod tests { .set_folder( "inbox", vec![ - StoredMail { mail: m1, details: None, rfc2822: None, uid: 1, attachments_pending: false }, - StoredMail { mail: m2, details: None, rfc2822: None, uid: 2, attachments_pending: false }, + StoredMail { + mail: m1, + details: None, + rfc2822: None, + uid: 1, + attachments_pending: false, + }, + StoredMail { + mail: m2, + details: None, + rfc2822: None, + uid: 2, + attachments_pending: false, + }, ], ) .await; @@ -1406,7 +1421,10 @@ mod tests { session.handle_command("b SELECT INBOX").await; let resp = session.handle_command("c UID MOVE 1 Work").await; - assert!(resp.iter().any(|r| r.contains("EXPUNGE")), "expected EXPUNGE, got {resp:?}"); + assert!( + resp.iter().any(|r| r.contains("EXPUNGE")), + "expected EXPUNGE, got {resp:?}" + ); assert!(resp.last().unwrap().contains("OK UID MOVE")); let moved = backend.moved.lock().unwrap(); @@ -1422,7 +1440,10 @@ mod tests { session.handle_command("a LOGIN u p").await; session.handle_command("b SELECT INBOX").await; let resp = session.handle_command("c UID COPY 1 Work").await; - assert!(resp[0].contains("NO"), "COPY should be rejected, got {resp:?}"); + assert!( + resp[0].contains("NO"), + "COPY should be rejected, got {resp:?}" + ); } async fn populate_store(store: &MailStore, mails: &[Mail]) { @@ -1451,10 +1472,18 @@ mod tests { #[async_trait::async_trait] impl MailBackend for MockBackend { - async fn load_mail_ids_for_folder(&self, _folder: &FolderInfo, _limit: usize) -> Result, String> { + async fn load_mail_ids_for_folder( + &self, + _folder: &FolderInfo, + _limit: usize, + ) -> Result, String> { Ok(self.mails.lock().unwrap().clone()) } - async fn load_mail(&self, _list_id: &str, element_id: &str) -> Result, String> { + async fn load_mail( + &self, + _list_id: &str, + element_id: &str, + ) -> Result, String> { Ok(self .mails .lock() @@ -1487,13 +1516,23 @@ mod tests { Ok(None) } async fn load_mail_details(&self, mail: &Mail) -> Result, String> { - let key = mail._id.as_ref().map(|id| id.element_id.to_string()).unwrap_or_default(); + let key = mail + ._id + .as_ref() + .map(|id| id.element_id.to_string()) + .unwrap_or_default(); Ok(self.details.lock().unwrap().get(&key).cloned()) } async fn load_attachments( &self, _mail: &Mail, - ) -> Result)>, String> { + ) -> Result< + Vec<( + tutasdk::entities::generated::tutanota::TutanotaFile, + Vec, + )>, + String, + > { // The mock has no attachment fixtures; the IMAP-session tests // only exercise body/header paths. Ok(Vec::new()) @@ -1501,7 +1540,11 @@ mod tests { async fn list_folders(&self) -> Result, String> { Ok(vec![inbox_folder()]) } - async fn set_unread_status(&self, mail_ids: Vec, unread: bool) -> Result<(), String> { + async fn set_unread_status( + &self, + mail_ids: Vec, + unread: bool, + ) -> Result<(), String> { self.unread_calls.lock().unwrap().push((mail_ids, unread)); Ok(()) } @@ -1509,8 +1552,15 @@ mod tests { self.trashed.lock().unwrap().extend(mail_ids); Ok(()) } - async fn move_mails(&self, mail_ids: Vec, target: &FolderInfo) -> Result<(), String> { - self.moved.lock().unwrap().push((mail_ids, target.id.clone())); + async fn move_mails( + &self, + mail_ids: Vec, + target: &FolderInfo, + ) -> Result<(), String> { + self.moved + .lock() + .unwrap() + .push((mail_ids, target.id.clone())); Ok(()) } async fn send_mail(&self, msg: &ParsedMessage) -> Result<(), String> { @@ -1521,10 +1571,7 @@ mod tests { fn make_mail(element_id: &str, subject: &str, unread: bool) -> Mail { Mail { - _id: Some(IdTupleGenerated::new( - test_id("list1"), - test_id(element_id), - )), + _id: Some(IdTupleGenerated::new(test_id("list1"), test_id(element_id))), _permissions: test_id("perm1"), _format: 0, _ownerEncSessionKey: None, @@ -1557,10 +1604,7 @@ mod tests { _errors: Default::default(), }, attachments: vec![], - conversationEntry: IdTupleGenerated::new( - test_id("conv_list"), - test_id("conv_elem"), - ), + conversationEntry: IdTupleGenerated::new(test_id("conv_list"), test_id("conv_elem")), firstRecipient: Some(MailAddress { _id: None, name: "Recip".to_string(), @@ -1619,10 +1663,27 @@ mod tests { let rfc1 = crate::mail::mail_to_rfc2822(&m1, Some(&d1), &[]); let rfc2 = crate::mail::mail_to_rfc2822(&m2, Some(&d2), &[]); store.set_folder_list(vec![inbox_folder()]).await; - store.set_folder("inbox", vec![ - StoredMail { mail: m1, details: Some(d1), rfc2822: Some(rfc1), uid: 1, attachments_pending: false }, - StoredMail { mail: m2, details: Some(d2), rfc2822: Some(rfc2), uid: 2, attachments_pending: false }, - ]).await; + store + .set_folder( + "inbox", + vec![ + StoredMail { + mail: m1, + details: Some(d1), + rfc2822: Some(rfc1), + uid: 1, + attachments_pending: false, + }, + StoredMail { + mail: m2, + details: Some(d2), + rfc2822: Some(rfc2), + uid: 2, + attachments_pending: false, + }, + ], + ) + .await; let mut session = ImapSession::new(store, backend, None); // LOGIN @@ -1643,13 +1704,15 @@ mod tests { // FETCH FLAGS let resp = session.handle_command("A004 FETCH 1:* (FLAGS UID)").await; assert!(resp[0].contains("* 1 FETCH")); - assert!(resp[0].contains("FLAGS ()")); // unread → no \Seen + assert!(resp[0].contains("FLAGS ()")); // unread → no \Seen assert!(resp[1].contains("* 2 FETCH")); - assert!(resp[1].contains("FLAGS (\\Seen)")); // read → \Seen + assert!(resp[1].contains("FLAGS (\\Seen)")); // read → \Seen assert!(resp.last().unwrap().contains("OK FETCH")); // UID FETCH with BODY - let resp = session.handle_command("A005 UID FETCH 1 (BODY.PEEK[])").await; + let resp = session + .handle_command("A005 UID FETCH 1 (BODY.PEEK[])") + .await; assert!(resp[0].contains("BODY[]")); // Body is base64-encoded HTML, verify the literal is present let b64_body = base64::engine::general_purpose::STANDARD.encode(b"

Body 1

"); @@ -1658,9 +1721,11 @@ mod tests { #[tokio::test] async fn test_store_seen_flag_calls_backend() { - let backend = Arc::new(MockBackend::with_mails(vec![ - make_mail("m1", "Unread mail", true), - ])); + let backend = Arc::new(MockBackend::with_mails(vec![make_mail( + "m1", + "Unread mail", + true, + )])); let store = MailStore::new(); populate_store(&store, &backend.mails.lock().unwrap()).await; let mut session = ImapSession::new(store, backend.clone(), None); @@ -1697,7 +1762,9 @@ mod tests { session.handle_command("A002 SELECT INBOX").await; // Mark mail 2 as deleted - session.handle_command("A003 STORE 2 +FLAGS (\\Deleted)").await; + session + .handle_command("A003 STORE 2 +FLAGS (\\Deleted)") + .await; // Verify deleted flag in FETCH let resp = session.handle_command("A004 FETCH 2 (FLAGS)").await; @@ -1714,7 +1781,10 @@ mod tests { // Verify only 2 mails remain let resp = session.handle_command("A006 FETCH 1:* (FLAGS UID)").await; - let fetch_lines: Vec<_> = resp.iter().filter(|r| r.contains("* ") && r.contains("FETCH")).collect(); + let fetch_lines: Vec<_> = resp + .iter() + .filter(|r| r.contains("* ") && r.contains("FETCH")) + .collect(); assert_eq!(fetch_lines.len(), 2); } @@ -1760,15 +1830,17 @@ mod tests { fn extract_uid(line: &str) -> u32 { let uid_pos = line.find("UID ").unwrap(); let rest = &line[uid_pos + 4..]; - let end = rest.find(|c: char| !c.is_ascii_digit()).unwrap_or(rest.len()); + let end = rest + .find(|c: char| !c.is_ascii_digit()) + .unwrap_or(rest.len()); rest[..end].parse().unwrap() } #[tokio::test] async fn test_idle_end_with_done() { - let backend = Arc::new(MockBackend::with_mails(vec![ - make_mail("m1", "Test", false), - ])); + let backend = Arc::new(MockBackend::with_mails(vec![make_mail( + "m1", "Test", false, + )])); let (_store, mut session) = make_session(backend).await; session.handle_command("A001 LOGIN user pass").await; @@ -1847,9 +1919,9 @@ mod tests { #[tokio::test] async fn test_close_resets_state() { - let backend = Arc::new(MockBackend::with_mails(vec![ - make_mail("m1", "Test", false), - ])); + let backend = Arc::new(MockBackend::with_mails(vec![make_mail( + "m1", "Test", false, + )])); let (_store, mut session) = make_session(backend).await; session.handle_command("A001 LOGIN user pass").await; @@ -1875,13 +1947,20 @@ mod tests { session.handle_command("A002 SELECT INBOX").await; // Delete mails 1 and 3 - session.handle_command("A003 STORE 1 +FLAGS (\\Deleted)").await; - session.handle_command("A004 STORE 3 +FLAGS (\\Deleted)").await; + session + .handle_command("A003 STORE 1 +FLAGS (\\Deleted)") + .await; + session + .handle_command("A004 STORE 3 +FLAGS (\\Deleted)") + .await; let resp = session.handle_command("A005 EXPUNGE").await; // Mail 1 is expunged at seq 1, then mail 3 becomes seq 2 - let expunge_lines: Vec<_> = resp.iter().filter(|r| r.starts_with("* ") && r.contains("EXPUNGE")).collect(); + let expunge_lines: Vec<_> = resp + .iter() + .filter(|r| r.starts_with("* ") && r.contains("EXPUNGE")) + .collect(); assert_eq!(expunge_lines.len(), 2); assert!(expunge_lines[0].contains("* 1 EXPUNGE")); assert!(expunge_lines[1].contains("* 2 EXPUNGE")); @@ -1946,15 +2025,30 @@ mod tests { .set_folder( "inbox", vec![ - StoredMail { mail: m1, details: None, rfc2822: None, uid: 1, attachments_pending: false }, - StoredMail { mail: m2, details: None, rfc2822: None, uid: 2, attachments_pending: false }, + StoredMail { + mail: m1, + details: None, + rfc2822: None, + uid: 1, + attachments_pending: false, + }, + StoredMail { + mail: m2, + details: None, + rfc2822: None, + uid: 2, + attachments_pending: false, + }, ], ) .await; let resp = session.check_new_mail().await; // No EXPUNGE — only an added mail. EXISTS reports the new total. assert!(!resp.iter().any(|r| r.contains("EXPUNGE")), "got {resp:?}"); - assert!(resp.iter().any(|r| r.contains("* 2 EXISTS")), "got {resp:?}"); + assert!( + resp.iter().any(|r| r.contains("* 2 EXISTS")), + "got {resp:?}" + ); } #[tokio::test] @@ -1967,13 +2061,25 @@ mod tests { store .set_folder( "inbox", - vec![StoredMail { mail: m1, details: None, rfc2822: None, uid: 1, attachments_pending: false }], + vec![StoredMail { + mail: m1, + details: None, + rfc2822: None, + uid: 1, + attachments_pending: false, + }], ) .await; let resp = session.check_new_mail().await; // m2 was seqno 2 in the session view. - assert!(resp.iter().any(|r| r.contains("* 2 EXPUNGE")), "got {resp:?}"); - assert!(resp.iter().any(|r| r.contains("* 1 EXISTS")), "got {resp:?}"); + assert!( + resp.iter().any(|r| r.contains("* 2 EXPUNGE")), + "got {resp:?}" + ); + assert!( + resp.iter().any(|r| r.contains("* 1 EXISTS")), + "got {resp:?}" + ); } #[tokio::test] @@ -1990,16 +2096,34 @@ mod tests { .set_folder( "inbox", vec![ - StoredMail { mail: old1, details: None, rfc2822: None, uid: 1, attachments_pending: false }, - StoredMail { mail: new3, details: None, rfc2822: None, uid: 3, attachments_pending: false }, + StoredMail { + mail: old1, + details: None, + rfc2822: None, + uid: 1, + attachments_pending: false, + }, + StoredMail { + mail: new3, + details: None, + rfc2822: None, + uid: 3, + attachments_pending: false, + }, ], ) .await; let resp = session.check_new_mail().await; // e2 was at seqno 2 — must EXPUNGE it. Then EXISTS reports the // (unchanged) count so the client refetches and discovers e3. - assert!(resp.iter().any(|r| r.contains("* 2 EXPUNGE")), "got {resp:?}"); - assert!(resp.iter().any(|r| r.contains("* 2 EXISTS")), "got {resp:?}"); + assert!( + resp.iter().any(|r| r.contains("* 2 EXPUNGE")), + "got {resp:?}" + ); + assert!( + resp.iter().any(|r| r.contains("* 2 EXISTS")), + "got {resp:?}" + ); } #[tokio::test] @@ -2014,12 +2138,24 @@ mod tests { store .set_folder( "inbox", - vec![StoredMail { mail: m1, details: None, rfc2822: None, uid: 1, attachments_pending: false }], + vec![StoredMail { + mail: m1, + details: None, + rfc2822: None, + uid: 1, + attachments_pending: false, + }], ) .await; let resp = session.check_new_mail().await; - let expunge_idx_3 = resp.iter().position(|r| r.contains("* 3 EXPUNGE")).expect("missing 3"); - let expunge_idx_2 = resp.iter().position(|r| r.contains("* 2 EXPUNGE")).expect("missing 2"); + let expunge_idx_3 = resp + .iter() + .position(|r| r.contains("* 3 EXPUNGE")) + .expect("missing 3"); + let expunge_idx_2 = resp + .iter() + .position(|r| r.contains("* 2 EXPUNGE")) + .expect("missing 2"); assert!( expunge_idx_3 < expunge_idx_2, "EXPUNGE 3 must come before EXPUNGE 2, got {resp:?}", diff --git a/crates/bridge/src/lib.rs b/crates/bridge/src/lib.rs index 54f1c14..8198773 100644 --- a/crates/bridge/src/lib.rs +++ b/crates/bridge/src/lib.rs @@ -2,10 +2,10 @@ pub mod backup; pub mod bridge; pub mod config; pub mod event_handler; -pub mod store; -pub mod sync; -pub mod tuta; pub mod imap; pub mod mail; pub mod smtp; +pub mod store; +pub mod sync; pub mod tls; +pub mod tuta; diff --git a/crates/bridge/src/mail/bodystructure.rs b/crates/bridge/src/mail/bodystructure.rs index 2a833be..cbded7b 100644 --- a/crates/bridge/src/mail/bodystructure.rs +++ b/crates/bridge/src/mail/bodystructure.rs @@ -22,8 +22,8 @@ pub fn compute_bodystructure(rfc2822: &str) -> String { let headers = parse_headers(&headers_text); let content_type = get_header(&headers, "content-type") .unwrap_or_else(|| "text/html; charset=UTF-8".to_owned()); - let cte = get_header(&headers, "content-transfer-encoding") - .unwrap_or_else(|| "7bit".to_owned()); + let cte = + get_header(&headers, "content-transfer-encoding").unwrap_or_else(|| "7bit".to_owned()); bodystructure_for(&content_type, &cte, &body, &headers) } @@ -71,8 +71,8 @@ fn part_structure(part: &str) -> String { let headers = parse_headers(&headers_text); let content_type = get_header(&headers, "content-type").unwrap_or_else(|| "text/plain".to_owned()); - let cte = get_header(&headers, "content-transfer-encoding") - .unwrap_or_else(|| "7bit".to_owned()); + let cte = + get_header(&headers, "content-transfer-encoding").unwrap_or_else(|| "7bit".to_owned()); bodystructure_for(&content_type, &cte, &body, &headers) } diff --git a/crates/bridge/src/mail/parser.rs b/crates/bridge/src/mail/parser.rs index 0d959cf..fd75da8 100644 --- a/crates/bridge/src/mail/parser.rs +++ b/crates/bridge/src/mail/parser.rs @@ -52,7 +52,11 @@ pub fn parse_rfc2822(raw: &str) -> ParsedMessage { extract_multipart_body_and_attachments(&body_section, &content_type) } else { ( - decode_body(&body_section, &content_transfer_encoding, &content_type.to_lowercase()), + decode_body( + &body_section, + &content_transfer_encoding, + &content_type.to_lowercase(), + ), Vec::new(), ) }; @@ -90,20 +94,29 @@ pub(super) fn parse_headers(header_section: &str) -> Vec<(String, String)> { current_value.push_str(line.trim()); } else if let Some((name, value)) = line.split_once(':') { if !current_name.is_empty() { - headers.push((current_name.to_lowercase(), current_value.trim().to_string())); + headers.push(( + current_name.to_lowercase(), + current_value.trim().to_string(), + )); } current_name = name.trim().to_string(); current_value = value.to_string(); } } if !current_name.is_empty() { - headers.push((current_name.to_lowercase(), current_value.trim().to_string())); + headers.push(( + current_name.to_lowercase(), + current_value.trim().to_string(), + )); } headers } pub(super) fn get_header(headers: &[(String, String)], name: &str) -> Option { - headers.iter().find(|(n, _)| n == name).map(|(_, v)| v.clone()) + headers + .iter() + .find(|(n, _)| n == name) + .map(|(_, v)| v.clone()) } fn parse_address_single(raw: &str) -> (String, String) { @@ -202,10 +215,9 @@ fn decode_q_encoding(s: &str) -> String { let mut i = 0; while i < bytes.len() { if bytes[i] == b'=' && i + 2 < bytes.len() { - if let Ok(byte) = u8::from_str_radix( - std::str::from_utf8(&bytes[i + 1..i + 3]).unwrap_or(""), - 16, - ) { + if let Ok(byte) = + u8::from_str_radix(std::str::from_utf8(&bytes[i + 1..i + 3]).unwrap_or(""), 16) + { result.push(byte); i += 3; continue; @@ -228,7 +240,9 @@ pub(super) fn extract_boundary(content_type: &str) -> Option { let boundary = if rest.starts_with('"') { rest[1..].split('"').next().unwrap_or("") } else { - rest.split(|c: char| c.is_whitespace() || c == ';').next().unwrap_or("") + rest.split(|c: char| c.is_whitespace() || c == ';') + .next() + .unwrap_or("") }; if !boundary.is_empty() { return Some(boundary.to_string()); @@ -267,8 +281,7 @@ fn extract_multipart_body_and_attachments( let part_cd_lower = part_cd.to_lowercase(); let is_attachment = part_cd_lower.contains("attachment") - || (extract_param(&part_ct, "name").is_some() - && !part_ct_lower.contains("text/")); + || (extract_param(&part_ct, "name").is_some() && !part_ct_lower.contains("text/")); if part_ct_lower.contains("multipart/") { let (nested_body, nested_atts) = @@ -284,10 +297,10 @@ fn extract_multipart_body_and_attachments( base64::engine::general_purpose::STANDARD .decode(&clean) .unwrap_or_default() - }, + } cte if cte.contains("quoted-printable") => { decode_quoted_printable(&part_body).into_bytes() - }, + } _ => part_body.as_bytes().to_vec(), }; let filename = extract_param(&part_cd, "filename") @@ -307,7 +320,8 @@ fn extract_multipart_body_and_attachments( }); } else if part_ct_lower.contains("text/html") && html_part.is_none() { html_part = Some(decode_body(&part_body, &part_cte, &part_ct_lower)); - } else if part_ct_lower.contains("text/plain") && html_part.is_none() && text_part.is_none() { + } else if part_ct_lower.contains("text/plain") && html_part.is_none() && text_part.is_none() + { text_part = Some(decode_body(&part_body, &part_cte, &part_ct_lower)); } } @@ -348,13 +362,23 @@ pub(super) fn split_mime_parts(body: &str, boundary: &str) -> Vec { for line in body.lines() { if line.starts_with(&end_delimiter) { if in_part && !current.is_empty() { - parts.push(current.trim_start_matches("\r\n").trim_start_matches('\n').to_string()); + parts.push( + current + .trim_start_matches("\r\n") + .trim_start_matches('\n') + .to_string(), + ); } break; } if line.starts_with(&delimiter) { if in_part && !current.is_empty() { - parts.push(current.trim_start_matches("\r\n").trim_start_matches('\n').to_string()); + parts.push( + current + .trim_start_matches("\r\n") + .trim_start_matches('\n') + .to_string(), + ); } current = String::new(); in_part = true; @@ -401,10 +425,7 @@ fn decode_quoted_printable(s: &str) -> String { i += 2; } else if i + 2 < bytes.len() { let hex = [bytes[i + 1], bytes[i + 2]]; - if let Ok(val) = u8::from_str_radix( - std::str::from_utf8(&hex).unwrap_or(""), - 16, - ) { + if let Ok(val) = u8::from_str_radix(std::str::from_utf8(&hex).unwrap_or(""), 16) { result.push(val); } i += 3; diff --git a/crates/bridge/src/mail/rfc2822.rs b/crates/bridge/src/mail/rfc2822.rs index a26fb6f..f648520 100644 --- a/crates/bridge/src/mail/rfc2822.rs +++ b/crates/bridge/src/mail/rfc2822.rs @@ -86,7 +86,10 @@ pub fn mail_to_rfc2822( for (file, data) in attachments { msg.push_str(&format!("--{}\r\n", boundary)); - let mime = file.mimeType.as_deref().unwrap_or("application/octet-stream"); + let mime = file + .mimeType + .as_deref() + .unwrap_or("application/octet-stream"); let name_encoded = encode_header_value(&file.name); msg.push_str(&format!( "Content-Type: {}; name=\"{}\"\r\n", @@ -380,10 +383,7 @@ mod tests { use tutasdk::IdTupleGenerated; let mail = Mail { - _id: Some(IdTupleGenerated::new( - test_id("list1"), - test_id("elem1"), - )), + _id: Some(IdTupleGenerated::new(test_id("list1"), test_id("elem1"))), _permissions: test_id("perm1"), _format: 0, _ownerEncSessionKey: None, @@ -416,10 +416,7 @@ mod tests { _errors: Default::default(), }, attachments: vec![], - conversationEntry: IdTupleGenerated::new( - test_id("conv_list1"), - test_id("conv_elem1"), - ), + conversationEntry: IdTupleGenerated::new(test_id("conv_list1"), test_id("conv_elem1")), firstRecipient: Some(MailAddress { _id: None, name: "Bob".to_string(), @@ -456,10 +453,7 @@ mod tests { use tutasdk::IdTupleGenerated; let mail = Mail { - _id: Some(IdTupleGenerated::new( - test_id("list2"), - test_id("elem2"), - )), + _id: Some(IdTupleGenerated::new(test_id("list2"), test_id("elem2"))), _permissions: test_id("perm2"), _format: 0, _ownerEncSessionKey: None, @@ -492,10 +486,7 @@ mod tests { _errors: Default::default(), }, attachments: vec![], - conversationEntry: IdTupleGenerated::new( - test_id("conv_list2"), - test_id("conv_elem2"), - ), + conversationEntry: IdTupleGenerated::new(test_id("conv_list2"), test_id("conv_elem2")), firstRecipient: None, mailDetails: None, mailDetailsDraft: None, @@ -552,8 +543,7 @@ mod tests { assert!(rfc.contains("To: Bob , charlie@example.com\r\n")); assert!(rfc.contains("Cc: Dave \r\n")); // Body should be base64 of "

Hello World

" - let body_b64 = - base64::engine::general_purpose::STANDARD.encode(b"

Hello World

"); + let body_b64 = base64::engine::general_purpose::STANDARD.encode(b"

Hello World

"); assert!(rfc.contains(&body_b64)); } @@ -600,10 +590,7 @@ mod tests { _errors: Default::default(), }, attachments: vec![], - conversationEntry: IdTupleGenerated::new( - test_id("conv_l"), - test_id("conv_e"), - ), + conversationEntry: IdTupleGenerated::new(test_id("conv_l"), test_id("conv_e")), firstRecipient: Some(MailAddress { _id: None, name: "".to_string(), @@ -661,12 +648,13 @@ mod tests { let attachments: Vec = vec![(&file, data)]; let rfc = mail_to_rfc2822(&mail, Some(&details), &attachments); - assert!(rfc.contains("Content-Type: multipart/mixed; boundary=\"=_TutaBridge_list_att_elem_att\"")); + assert!(rfc.contains( + "Content-Type: multipart/mixed; boundary=\"=_TutaBridge_list_att_elem_att\"" + )); assert!(rfc.contains("--=_TutaBridge_list_att_elem_att\r\n")); // Body part: text/html base64 assert!(rfc.contains("Content-Type: text/html; charset=UTF-8\r\n")); - let body_b64 = - base64::engine::general_purpose::STANDARD.encode(b"

The body

"); + let body_b64 = base64::engine::general_purpose::STANDARD.encode(b"

The body

"); assert!(rfc.contains(&body_b64)); // Attachment part assert!(rfc.contains("Content-Type: application/pdf; name=\"doc.pdf\"")); diff --git a/crates/bridge/src/smtp/mod.rs b/crates/bridge/src/smtp/mod.rs index 07c485e..50e115e 100644 --- a/crates/bridge/src/smtp/mod.rs +++ b/crates/bridge/src/smtp/mod.rs @@ -1,6 +1,6 @@ -use std::sync::Arc; use base64::Engine; -use log::{info, error, debug}; +use log::{debug, error, info}; +use std::sync::Arc; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::TcpListener; use tokio_rustls::TlsAcceptor; @@ -13,9 +13,15 @@ enum SmtpState { Init, Greeted, MailFrom(String), - RcptTo { from: String, to: Vec }, + RcptTo { + from: String, + to: Vec, + }, #[allow(dead_code)] - Data { from: String, to: Vec }, + Data { + from: String, + to: Vec, + }, Quit, } @@ -139,9 +145,7 @@ async fn handle_connection( } Err(e) => { error!("SMTP: failed to send via Tuta: {}", e); - writer - .write_all(b"451 Temporary failure\r\n") - .await?; + writer.write_all(b"451 Temporary failure\r\n").await?; } } state = SmtpState::Greeted; @@ -157,7 +161,11 @@ async fn handle_connection( continue; } - let cmd = trimmed.split_whitespace().next().unwrap_or("").to_uppercase(); + let cmd = trimmed + .split_whitespace() + .next() + .unwrap_or("") + .to_uppercase(); let response = match cmd.as_str() { "EHLO" | "HELO" => { state = SmtpState::Greeted; @@ -208,19 +216,17 @@ async fn handle_connection( } "250 OK\r\n".to_string() } - "DATA" => { - match &state { - SmtpState::RcptTo { from, to } => { - state = SmtpState::Data { - from: from.clone(), - to: to.clone(), - }; - in_data = true; - "354 Start mail input; end with .\r\n".to_string() - } - _ => "503 Bad sequence\r\n".to_string(), + "DATA" => match &state { + SmtpState::RcptTo { from, to } => { + state = SmtpState::Data { + from: from.clone(), + to: to.clone(), + }; + in_data = true; + "354 Start mail input; end with .\r\n".to_string() } - } + _ => "503 Bad sequence\r\n".to_string(), + }, "RSET" => { state = SmtpState::Greeted; "250 OK\r\n".to_string() @@ -252,11 +258,7 @@ fn extract_address(line: &str) -> String { } } } - line.split(':') - .nth(1) - .unwrap_or("") - .trim() - .to_string() + line.split(':').nth(1).unwrap_or("").trim().to_string() } fn verify_smtp_plain_data(data: &str, expected: &Option) -> String { diff --git a/crates/bridge/src/store.rs b/crates/bridge/src/store.rs index cebef35..8e395e7 100644 --- a/crates/bridge/src/store.rs +++ b/crates/bridge/src/store.rs @@ -320,8 +320,7 @@ impl LocalStore { if !deleted.is_empty() { conn.execute_batch("BEGIN IMMEDIATE")?; { - let mut stmt = - conn.prepare_cached("DELETE FROM mails WHERE element_id = ?1")?; + let mut stmt = conn.prepare_cached("DELETE FROM mails WHERE element_id = ?1")?; for eid in &deleted { stmt.execute([eid])?; } @@ -397,8 +396,7 @@ impl LocalStore { pub fn total_count(&self) -> Result { let conn = self.conn.lock().unwrap(); - let count: i64 = - conn.query_row("SELECT COUNT(*) FROM mails", [], |row| row.get(0))?; + let count: i64 = conn.query_row("SELECT COUNT(*) FROM mails", [], |row| row.get(0))?; Ok(count as usize) } @@ -443,11 +441,7 @@ impl LocalStore { /// Persist the last processed batch id for a group (event-bus catch-up /// resumes from this point on the next reconnect). - pub fn set_event_bus_batch_id( - &self, - group_id: &str, - batch_id: &str, - ) -> Result<(), StoreError> { + pub fn set_event_bus_batch_id(&self, group_id: &str, batch_id: &str) -> Result<(), StoreError> { let conn = self.conn.lock().unwrap(); let now_ms = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -540,7 +534,8 @@ mod tests { fn open_memory_store() -> LocalStore { let key = test_key(); - let tmp_dir = std::env::temp_dir().join(format!("tutabridge_test_{}", rand::random::())); + let tmp_dir = + std::env::temp_dir().join(format!("tutabridge_test_{}", rand::random::())); std::fs::create_dir_all(&tmp_dir).unwrap(); let db_path = tmp_dir.join("test.db"); let mails_dir = tmp_dir.join("mails"); @@ -586,7 +581,9 @@ mod tests { #[test] fn test_upsert_and_load_metadata() { let store = open_memory_store(); - store.upsert_mail_metadata(&meta("abc123", "inbox", 1700000000000)).unwrap(); + store + .upsert_mail_metadata(&meta("abc123", "inbox", 1700000000000)) + .unwrap(); let loaded = store.load_folder_metadata("inbox").unwrap(); assert_eq!(loaded.len(), 1); @@ -600,7 +597,9 @@ mod tests { fn test_metadata_is_per_folder() { let store = open_memory_store(); store.upsert_mail_metadata(&meta("a", "inbox", 1)).unwrap(); - store.upsert_mail_metadata(&meta("b", "custom1", 2)).unwrap(); + store + .upsert_mail_metadata(&meta("b", "custom1", 2)) + .unwrap(); assert_eq!(store.load_folder_metadata("inbox").unwrap().len(), 1); assert_eq!(store.load_folder_metadata("custom1").unwrap().len(), 1); @@ -641,7 +640,10 @@ mod tests { let store = open_memory_store(); let rfc2822 = "From: test@example.com\r\nSubject: Hello\r\n\r\nBody text here"; store.write_eml("test_mail", rfc2822).unwrap(); - assert_eq!(store.read_eml("test_mail").unwrap(), Some(rfc2822.to_string())); + assert_eq!( + store.read_eml("test_mail").unwrap(), + Some(rfc2822.to_string()) + ); } #[test] @@ -662,7 +664,9 @@ mod tests { #[test] fn test_reset() { let store = open_memory_store(); - store.upsert_mail_metadata(&meta("abc", "inbox", 0)).unwrap(); + store + .upsert_mail_metadata(&meta("abc", "inbox", 0)) + .unwrap(); store.write_eml("abc", "content").unwrap(); store.reset().unwrap(); @@ -729,7 +733,9 @@ mod tests { #[test] fn delete_mail_removes_metadata_and_eml() { let store = open_memory_store(); - store.upsert_mail_metadata(&meta("del1", "inbox", 0)).unwrap(); + store + .upsert_mail_metadata(&meta("del1", "inbox", 0)) + .unwrap(); store.write_eml("del1", "content").unwrap(); assert_eq!(store.mail_count("inbox").unwrap(), 1); assert!(store.has_eml("del1")); @@ -742,7 +748,9 @@ mod tests { #[test] fn test_mark_has_details() { let store = open_memory_store(); - store.upsert_mail_metadata(&meta("det", "inbox", 0)).unwrap(); + store + .upsert_mail_metadata(&meta("det", "inbox", 0)) + .unwrap(); assert!(!store.load_folder_metadata("inbox").unwrap()[0].has_details); store.mark_has_details("det").unwrap(); diff --git a/crates/bridge/src/sync.rs b/crates/bridge/src/sync.rs index 73f6883..d72d55b 100644 --- a/crates/bridge/src/sync.rs +++ b/crates/bridge/src/sync.rs @@ -121,11 +121,7 @@ impl MailStore { /// Refresh an existing mail's metadata in every folder that holds it /// (body/details preserved). No-op if the mail is not cached. pub async fn refresh_mail_in_place(&self, mail: &Mail) { - let Some(eid) = mail - ._id - .as_ref() - .map(|id| id.element_id.to_string()) - else { + let Some(eid) = mail._id.as_ref().map(|id| id.element_id.to_string()) else { return; }; let mut folders = self.folders.write().await; @@ -508,19 +504,16 @@ async fn load_cached_folder( Ok(Some(eml)) => { if !meta.has_details { if let Err(e) = local_store.mark_has_details(&meta.element_id) { - warn!( - "Failed to heal has_details for {}: {e}", - meta.element_id - ); + warn!("Failed to heal has_details for {}: {e}", meta.element_id); } } Some(eml) - }, + } Ok(None) => Some(mail_to_rfc2822(&mail, None, &[])), Err(e) => { warn!("Failed to read cached eml {}: {e}", meta.element_id); Some(mail_to_rfc2822(&mail, None, &[])) - }, + } }; stored_mails.push(StoredMail { @@ -582,7 +575,12 @@ pub(crate) async fn sync_folder( let elem_id = mail._id.as_ref().map(|id| id.element_id.to_string()); let uid = elem_id .as_ref() - .and_then(|id| existing_map.get(id).map(|m| m.uid).or_else(|| new_uids.get(id).copied())) + .and_then(|id| { + existing_map + .get(id) + .map(|m| m.uid) + .or_else(|| new_uids.get(id).copied()) + }) .unwrap_or(0); if let Some(existing) = elem_id.as_ref().and_then(|id| existing_map.get(id)) { @@ -798,17 +796,13 @@ async fn prefetch_details( match backend.load_attachments(&stored.mail).await { Ok(attachments) => { - let refs: Vec<(&TutanotaFile, &[u8])> = attachments - .iter() - .map(|(f, d)| (f, d.as_slice())) - .collect(); + let refs: Vec<(&TutanotaFile, &[u8])> = + attachments.iter().map(|(f, d)| (f, d.as_slice())).collect(); let rfc2822 = mail_to_rfc2822(&stored.mail, Some(details), &refs); if let Err(e) = local_store.write_eml(&eid, &rfc2822) { warn!("Failed to rewrite eml {}: {}", eid, e); } - store - .update_mail_rfc2822(&folder.id, &eid, rfc2822) - .await; + store.update_mail_rfc2822(&folder.id, &eid, rfc2822).await; attachment_retry_history.remove(&eid); debug!("Attachment retry succeeded for {}", eid); } @@ -826,7 +820,11 @@ async fn prefetch_details( // Clear the pending flag so we don't keep retrying a // permanent failure (e.g. server returned 404). store - .update_mail_rfc2822(&folder.id, &eid, stored.rfc2822.clone().unwrap_or_default()) + .update_mail_rfc2822( + &folder.id, + &eid, + stored.rfc2822.clone().unwrap_or_default(), + ) .await; attachment_retry_history.remove(&eid); } @@ -855,7 +853,12 @@ where match f().await { Ok(v) => return Ok(v), Err(e) if attempt < MAX_RETRIES => { - warn!("Attempt {} failed: {}, retrying in {:?}", attempt + 1, e, delay); + warn!( + "Attempt {} failed: {}, retrying in {:?}", + attempt + 1, + e, + delay + ); tokio::time::sleep(delay).await; delay = backoff(delay); } @@ -988,7 +991,10 @@ mod tests { let b = store.get_folder("folderB").await; assert_eq!(a[0].mail.subject, "Hello [updated]"); assert!(!a[0].mail.unread); - assert_eq!(a[0].uid, 7, "UID is per-folder state, must survive a refresh"); + assert_eq!( + a[0].uid, 7, + "UID is per-folder state, must survive a refresh" + ); assert_eq!(b[0].mail.subject, "Hello [updated]"); assert_eq!(b[0].uid, 12); } @@ -1127,7 +1133,6 @@ mod tests { assert_eq!(store.get_folder("A").await.len(), 2); } - #[tokio::test] async fn prune_unknown_folders_drops_disappeared_ones() { let store = MailStore::new(); @@ -1137,8 +1142,7 @@ mod tests { store .set_folder("gone", vec![stored(make_mail("L1", "M2", "g", true), 2)]) .await; - let known: std::collections::HashSet = - ["keep".to_string()].into_iter().collect(); + let known: std::collections::HashSet = ["keep".to_string()].into_iter().collect(); let removed = store.prune_unknown_folders(&known).await; assert_eq!(removed, vec!["gone".to_string()]); assert_eq!(store.get_folder("keep").await.len(), 1); diff --git a/crates/bridge/src/tls.rs b/crates/bridge/src/tls.rs index 2ef928a..e2a1baa 100644 --- a/crates/bridge/src/tls.rs +++ b/crates/bridge/src/tls.rs @@ -18,13 +18,17 @@ fn key_path() -> PathBuf { cert_dir().join("key.pem") } -pub fn load_or_create_tls_acceptor() -> Result> { +pub fn load_or_create_tls_acceptor() -> Result> +{ let cert_file = cert_path(); let key_file = key_path(); let (cert_pem, key_pem) = if cert_file.exists() && key_file.exists() { log::info!("Loading TLS certificate from {}", cert_file.display()); - (std::fs::read_to_string(&cert_file)?, std::fs::read_to_string(&key_file)?) + ( + std::fs::read_to_string(&cert_file)?, + std::fs::read_to_string(&key_file)?, + ) } else { log::info!("Generating self-signed TLS certificate..."); let (cert, key) = generate_self_signed()?; @@ -61,20 +65,18 @@ fn load_certs( pem: &str, ) -> Result>, Box> { let mut reader = std::io::BufReader::new(pem.as_bytes()); - let certs: Vec> = rustls_pemfile::certs(&mut reader) - .collect::, _>>()?; + let certs: Vec> = + rustls_pemfile::certs(&mut reader).collect::, _>>()?; if certs.is_empty() { return Err("No certificates found in PEM".into()); } Ok(certs) } -fn load_key( - pem: &str, -) -> Result, Box> { +fn load_key(pem: &str) -> Result, Box> { let mut reader = std::io::BufReader::new(pem.as_bytes()); - let keys: Vec> = rustls_pemfile::pkcs8_private_keys(&mut reader) - .collect::, _>>()?; + let keys: Vec> = + rustls_pemfile::pkcs8_private_keys(&mut reader).collect::, _>>()?; let key = keys .into_iter() .next() diff --git a/crates/bridge/src/tuta.rs b/crates/bridge/src/tuta.rs index f4e9399..cf8a365 100644 --- a/crates/bridge/src/tuta.rs +++ b/crates/bridge/src/tuta.rs @@ -1,24 +1,24 @@ use base64::Engine; -use std::collections::HashMap; -use std::sync::Arc; use crypto_primitives::aes::{Aes256Key, Iv, AES_256_KEY_SIZE}; use crypto_primitives::blake3::blake3_kdf; use crypto_primitives::key::GenericAesKey; use crypto_primitives::randomizer_facade::RandomizerFacade; +use std::collections::HashMap; +use std::sync::Arc; use tutasdk::bindings::file_client::{FileClient, FileClientError}; use tutasdk::bindings::rest_client::RestClient; -use tutasdk::crypto_entity_client::CryptoEntityClient; use tutasdk::blobs::blob_facade::FileData; +use tutasdk::crypto_entity_client::CryptoEntityClient; use tutasdk::entities::generated::sys::BlobReferenceTokenWrapper; use tutasdk::entities::generated::tutanota::{ AttachmentKeyData, DraftAttachment, DraftCreateData, DraftData, DraftRecipient, Mail, MailBox, MailDetails, MailDetailsBlob, MailSetEntry, NewDraftAttachment, SendDraftData, SendDraftParameters, TutanotaFile, }; -use tutasdk::tutanota_constants::ArchiveDataType; use tutasdk::folder_system::{FolderSystem, MailSetKind}; use tutasdk::services::generated::tutanota::{DraftService, SendDraftService}; use tutasdk::services::ExtraServiceParams; +use tutasdk::tutanota_constants::ArchiveDataType; use tutasdk::{ApiCallError, CustomId, IdTupleGenerated, ListLoadDirection, LoggedInSdk, Sdk}; use crate::config::Config; @@ -48,7 +48,11 @@ pub const IMAP_DELIMITER: char = '/'; #[async_trait::async_trait] pub trait MailBackend: Send + Sync { - async fn load_mail_ids_for_folder(&self, folder: &FolderInfo, limit: usize) -> Result, String>; + async fn load_mail_ids_for_folder( + &self, + folder: &FolderInfo, + limit: usize, + ) -> Result, String>; /// Load a single mail by `(list_id, element_id)` — used by the event-bus /// handler to fetch a freshly-created or updated mail without re-listing /// its folder. `Ok(None)` means the entity is no longer on the server. @@ -77,16 +81,21 @@ pub trait MailBackend: Send + Sync { /// `mail.attachments`. Empty if the mail has no attachments. Errors are /// per-mail (no partial returns): if any one attachment fails the whole /// call returns `Err` so the caller can decide to retry the prefetch. - async fn load_attachments( - &self, - mail: &Mail, - ) -> Result)>, String>; + async fn load_attachments(&self, mail: &Mail) -> Result)>, String>; /// Enumerate all mail folders (system + custom, with hierarchy). async fn list_folders(&self) -> Result, String>; - async fn set_unread_status(&self, mail_ids: Vec, unread: bool) -> Result<(), String>; + async fn set_unread_status( + &self, + mail_ids: Vec, + unread: bool, + ) -> Result<(), String>; async fn trash_mails(&self, mail_ids: Vec) -> Result<(), String>; /// Move mails into the given target folder. - async fn move_mails(&self, mail_ids: Vec, target: &FolderInfo) -> Result<(), String>; + async fn move_mails( + &self, + mail_ids: Vec, + target: &FolderInfo, + ) -> Result<(), String>; async fn send_mail(&self, msg: &ParsedMessage) -> Result<(), String>; } @@ -221,8 +230,7 @@ impl TutaSession { "tutabridge local storage v1", AES_256_KEY_SIZE, ); - GenericAesKey::from_bytes(&derived) - .map_err(|e| format!("Key derivation error: {e:?}")) + GenericAesKey::from_bytes(&derived).map_err(|e| format!("Key derivation error: {e:?}")) } fn crypto_client(&self) -> Arc { @@ -301,7 +309,11 @@ impl TutaSession { .await { Ok(batch) => mails.extend(batch), - Err(e) => log::warn!("Failed to batch load mails from list {}: {}", list_id_str, e), + Err(e) => log::warn!( + "Failed to batch load mails from list {}: {}", + list_id_str, + e + ), } } @@ -322,7 +334,7 @@ impl TutaSession { Err(e) => { log::error!("Failed to load mail details draft: {e}"); Err(e) - }, + } } } else if mail.mailDetails.is_some() { match mail_facade.load_mail_details_blob(mail).await { @@ -330,7 +342,7 @@ impl TutaSession { Err(e) => { log::error!("Failed to load mail details blob: {e}"); Err(e) - }, + } } } else { // Legacy mail without either reference — body lives nowhere we @@ -477,9 +489,7 @@ impl TutaSession { let enc_mime_type = file_sk .encrypt_data(att.mime_type.as_bytes(), Iv::generate(randomizer)) .map_err(|e| { - ApiCallError::internal(format!( - "Failed to encrypt attachment mime type: {e}" - )) + ApiCallError::internal(format!("Failed to encrypt attachment mime type: {e}")) })?; let owner_enc_file_sk = mail_group_key.encrypt_key(file_sk, Iv::generate(randomizer)); @@ -556,8 +566,9 @@ impl TutaSession { .get_current_sym_group_key(&mail_group_id) .await?; - let owner_enc_session_key = - group_key.object.encrypt_key(&session_key, Iv::generate(&randomizer)); + let owner_enc_session_key = group_key + .object + .encrypt_key(&session_key, Iv::generate(&randomizer)); let owner_key_version = group_key.version as i64; // Upload every attachment first — the resulting `DraftAttachment` @@ -669,10 +680,7 @@ impl TutaSession { /// the File entities can't be decrypted. Returns `None` if this is not /// a self-send, the cache has nothing for the envelope, or the entry /// is older than the TTL. - async fn try_self_send_cache( - &self, - mail: &Mail, - ) -> Option)>> { + async fn try_self_send_cache(&self, mail: &Mail) -> Option)>> { if !mail.sender.address.eq_ignore_ascii_case(&self.email) { return None; } @@ -760,7 +768,11 @@ fn self_send_cache_key(subject: &str, sender: &str, recipient: &str) -> String { #[async_trait::async_trait] impl MailBackend for TutaSession { - async fn load_mail_ids_for_folder(&self, folder: &FolderInfo, limit: usize) -> Result, String> { + async fn load_mail_ids_for_folder( + &self, + folder: &FolderInfo, + limit: usize, + ) -> Result, String> { let entries_list_id = tutasdk::GeneratedId(folder.entries_list_id.clone()); self.load_mail_ids_for_folder_impl(&entries_list_id, limit) .await @@ -810,16 +822,18 @@ impl MailBackend for TutaSession { .map_err(|e| format!("{e}")) } - async fn load_attachments( - &self, - mail: &Mail, - ) -> Result)>, String> { - self.load_attachments_impl(mail).await.map_err(|e| format!("{e}")) + async fn load_attachments(&self, mail: &Mail) -> Result)>, String> { + self.load_attachments_impl(mail) + .await + .map_err(|e| format!("{e}")) } async fn list_folders(&self) -> Result, String> { let mailbox = self.load_mailbox().await.map_err(|e| format!("{e}"))?; - let folder_system = self.load_folders(&mailbox).await.map_err(|e| format!("{e}"))?; + let folder_system = self + .load_folders(&mailbox) + .await + .map_err(|e| format!("{e}"))?; let mut result = Vec::new(); for indented in folder_system.indented_list() { @@ -888,7 +902,11 @@ impl MailBackend for TutaSession { .map_err(|e| format!("{e}")) } - async fn move_mails(&self, mail_ids: Vec, target: &FolderInfo) -> Result<(), String> { + async fn move_mails( + &self, + mail_ids: Vec, + target: &FolderInfo, + ) -> Result<(), String> { let target_folder = IdTupleGenerated::new( tutasdk::GeneratedId(target.list_id.clone()), tutasdk::GeneratedId(target.id.clone()), @@ -998,7 +1016,10 @@ pub async fn login_with_2fa( .any(|c| c.r#type == i64::from(tutasdk::tutanota_constants::SecondFactorType::Totp)); if !has_totp { - return Err("Account requires U2F/WebAuthn 2FA which is not supported — only TOTP is supported".into()); + return Err( + "Account requires U2F/WebAuthn 2FA which is not supported — only TOTP is supported" + .into(), + ); } let totp_code = match &totp_callback { @@ -1018,7 +1039,9 @@ pub async fn login_with_2fa( .is_second_factor_pending(&access_token) .await .map_err(|e| { - Box::::from(format!("2FA poll failed: {e}")) + Box::::from(format!( + "2FA poll failed: {e}" + )) })?; if !pending { cleared = true; @@ -1084,18 +1107,15 @@ fn save_credentials(email: &str, creds: &tutasdk::login::Credentials) { fn load_credentials(email: &str) -> Option { let mut cache = CREDENTIALS_CACHE.lock().unwrap(); if let Some(cached) = cache.as_ref() { - return cached.clone(); } - let result = load_credentials_from_keyring(email); *cache = Some(result.clone()); result } fn load_credentials_from_keyring(email: &str) -> Option { - let entry = keyring::Entry::new(KEYRING_SERVICE, email).ok()?; let json_str = entry.get_password().ok()?; @@ -1118,14 +1138,12 @@ fn load_credentials_from_keyring(email: &str) -> Option Vec { @@ -1328,10 +1346,9 @@ mod transient_tests { #[test] fn missing_owner_key_is_transient() { let e = ApiCallError::InternalSdkError { - error_message: - "Failed to resolve session key for entity 'File' with ID: ...; \ + error_message: "Failed to resolve session key for entity 'File' with ID: ...; \ Session key resolution failure: instance missing owner key/group data" - .to_string(), + .to_string(), }; assert!(is_session_key_transient(&e)); } diff --git a/src-tauri/src/commands.rs b/src-tauri/src/commands.rs index e7165e2..e19317d 100644 --- a/src-tauri/src/commands.rs +++ b/src-tauri/src/commands.rs @@ -50,7 +50,8 @@ pub async fn start_bridge( _ => return Err("No config found — save config first".into()), }; - config::ensure_bridge_password(&mut cfg).map_err(|e| format!("Bridge password setup failed: {e}"))?; + config::ensure_bridge_password(&mut cfg) + .map_err(|e| format!("Bridge password setup failed: {e}"))?; let mut handle = state.lock().await; handle.start(cfg, password, None).await diff --git a/src-tauri/src/main.rs b/src-tauri/src/main.rs index f589d7d..1a268fb 100644 --- a/src-tauri/src/main.rs +++ b/src-tauri/src/main.rs @@ -2,15 +2,14 @@ mod commands; -use std::sync::Arc; use commands::BridgeState; +use std::sync::Arc; use tauri::Manager; use tokio::sync::Mutex; use tutabridge_core::bridge::BridgeHandle; fn main() { - env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("debug")) - .init(); + env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("debug")).init(); tokio_rustls::rustls::crypto::ring::default_provider() .install_default() @@ -114,10 +113,7 @@ async fn stream_stats( } } -async fn stream_logs( - app: tauri::AppHandle, - mut rx: tokio::sync::broadcast::Receiver, -) { +async fn stream_logs(app: tauri::AppHandle, mut rx: tokio::sync::broadcast::Receiver) { use tauri::Emitter; loop { match rx.recv().await { diff --git a/src/main.rs b/src/main.rs index ef2aca6..cff4ff3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,5 +1,5 @@ -use std::sync::Arc; use log::{info, warn}; +use std::sync::Arc; use tutabridge_core::{ backup, bridge as bridge_helpers, config, event_handler, imap, smtp, store::LocalStore, sync, tls, tuta, @@ -58,8 +58,8 @@ async fn main() -> anyhow::Result<()> { .map_err(|e| anyhow::anyhow!("Bridge password setup failed: {e}"))?; info!("TutaBridge starting..."); - let tls_acceptor = tls::load_or_create_tls_acceptor() - .map_err(|e| anyhow::anyhow!("TLS setup failed: {e}"))?; + let tls_acceptor = + tls::load_or_create_tls_acceptor().map_err(|e| anyhow::anyhow!("TLS setup failed: {e}"))?; info!("TLS initialized"); info!("IMAP will listen on 127.0.0.1:{}", cfg.imap_port); @@ -68,7 +68,9 @@ async fn main() -> anyhow::Result<()> { let session = login_session(&cfg).await?; info!("Logged in as {}", cfg.email); - let storage_key = session.derive_storage_key().await + let storage_key = session + .derive_storage_key() + .await .map_err(|e| anyhow::anyhow!("{e}"))?; info!("Storage encryption key derived"); @@ -76,7 +78,8 @@ async fn main() -> anyhow::Result<()> { &config::store_db_path(), &config::store_mails_dir(), storage_key, - ).map_err(|e| anyhow::anyhow!("{e}"))?; + ) + .map_err(|e| anyhow::anyhow!("{e}"))?; if !local_store.verify_key() { info!("Storage key changed — resetting local cache"); let _ = local_store.reset(); @@ -195,7 +198,11 @@ async fn main() -> anyhow::Result<()> { shutdown_rx.clone(), )); let imap_handle = tokio::spawn(imap::serve( - cfg.imap_port, store.clone(), backend.clone(), imap_tls, pw.clone(), + cfg.imap_port, + store.clone(), + backend.clone(), + imap_tls, + pw.clone(), )); let smtp_handle = tokio::spawn(smtp::serve(cfg.smtp_port, backend.clone(), smtp_tls, pw));