From 34127f75870560cf39d3d05e6d1a103c289ea9d4 Mon Sep 17 00:00:00 2001 From: Anthony Date: Thu, 28 May 2026 14:32:36 +0200 Subject: [PATCH] Event-driven realtime delta: skip the full folder re-sync on a MOVE MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The bridge no longer asks the server for a full folder listing on every MailSetEntry CREATE/DELETE event. Instead it uses the encoding of the entry id (4-byte timestamp + 9-byte Mail element id, see `tuta-sdk::mail_set_entry_id`) to recover the affected mail directly, and applies the delta to MailStore + LocalStore: - MailSetEntry CREATE first (so a MOVE clones from the source folder before the matching DELETE runs). Hit path = `find_mail_anywhere` → clone the already-decrypted StoredMail into the target folder with a fresh UID. Miss path = a single `load_mail` against the cached `Mail.list_id`. Any decode failure / unknown folder / `load_mail` error queues a fallback full `sync_folder` for that folder — no silent miss. - MailSetEntry DELETE second. Removes from the source folder only; the `.eml` and DB row are dropped only if no folder still holds the mail (multi-folder placement preserved, in-batch MOVE is correct because the target was upserted by the CREATE loop). - Mail UPDATE / DELETE unchanged. - MailSet folder-list dirty handling unchanged. Saves the `load_range(1000)` round-trip on the common MOVE case. The fallback path keeps the previous behaviour available so the change is strictly an optimisation, not a behavioural change. New MailStore helpers (5 new tests): - `find_mail_anywhere(eid)` / `is_mail_anywhere(eid)` — multi-folder lookup. - `remove_mail_from_folder(folder_id, eid)` — scoped remove (vs the existing `remove_mail_everywhere`). - `upsert_mail_in_folder` — idempotent insert/replace by element_id. - `mail_list_id()` — sniff once, cache; needed by the load_mail miss path. `mail_to_metadata` made `pub(crate)` for the handler. Bucketing in the event handler also split into `mail_set_entry_creates` vs `mail_set_entry_deletes` so the order is explicit; 8 handler tests cover empty / mixed / order-preserved / immutable-UPDATE-ignored shapes. 166/166 bridge lib tests pass. --- crates/bridge/src/event_handler.rs | 475 +++++++++++++++++++++-------- crates/bridge/src/sync.rs | 216 ++++++++++++- 2 files changed, 557 insertions(+), 134 deletions(-) diff --git a/crates/bridge/src/event_handler.rs b/crates/bridge/src/event_handler.rs index 14f23ee..391075d 100644 --- a/crates/bridge/src/event_handler.rs +++ b/crates/bridge/src/event_handler.rs @@ -2,20 +2,32 @@ //! //! Consumes [`EventBusMessage`] batches emitted by the SDK's WebSocket //! `EventBusClient` and applies them to `MailStore` (in-memory) and -//! `LocalStore` (encrypted SQLite + .eml files). After each batch the -//! `(group_id, batch_id)` pair is persisted so the next reconnect can -//! resume from there via `groupsToLastEventBatchIds`. +//! `LocalStore` (encrypted SQLite + `.eml` files). +//! +//! The hot path — `MailSetEntry` CREATE/DELETE — is handled **without** +//! re-listing the affected folder over REST: a `MailSetEntry`'s element id +//! is a Tuta-defined encoding of `(receivedDate, mail_element_id)`, so we +//! recover the mail id directly via `tutasdk::mail_set_entry_id::deconstruct` +//! and either move the already-decrypted Mail between folders in memory +//! (when it's a MOVE between two cached folders) or ask the backend for +//! the single Mail (`load_mail`) when it is a brand-new arrival. The full +//! `sync_folder` only runs as a safety-net fallback (decode failure, +//! unknown folder, `load_mail` error). +//! +//! After each batch the `(group_id, batch_id)` pair is persisted so the +//! next reconnect can resume from there via `groupsToLastEventBatchIds`. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::sync::{Arc, Mutex}; use log::{debug, info, warn}; use tokio::sync::{mpsc, watch}; use tutasdk::event_bus::{EntityUpdateBatch, EntityUpdateEvent, EventBusMessage, Operation}; +use tutasdk::{mail_set_entry_id, CustomId}; use crate::store::LocalStore; -use crate::sync::{sync_folder, MailStore}; -use crate::tuta::MailBackend; +use crate::sync::{sync_folder, MailStore, StoredMail}; +use crate::tuta::{FolderInfo, MailBackend}; /// Application tag for Tuta's mail-side entities. const TUTANOTA_APP: &str = "tutanota"; @@ -24,8 +36,8 @@ const TUTANOTA_APP: &str = "tutanota"; const MAIL_TYPE_ID: i64 = 97; /// `MailSetEntry` entity — placement of a mail inside a folder/MailSet. const MAIL_SET_ENTRY_TYPE_ID: i64 = 1450; -/// `MailSet` entity — the folder itself (custom folders are created/renamed/ -/// deleted by mutating MailSet entries). +/// `MailSet` entity — the folder itself (custom folders are created / +/// renamed / deleted by mutating MailSet entries). const MAIL_SET_TYPE_ID: i64 = 429; pub async fn run_event_handler( @@ -84,16 +96,15 @@ async fn process( } } -/// Bucketed view of the mail-relevant entity updates inside a batch. The -/// routing decision (which folders need a resync, which mails need a -/// metadata refresh, whether the folder list itself changed) is pure and -/// has no I/O — that lets us test it in isolation. +/// Bucketed view of the mail-relevant entity updates inside a batch. Pure; +/// no I/O. Splitting `MailSetEntry` events into creates and deletes lets +/// us process them in the right order (CREATEs first — so a move can clone +/// the mail from the source folder before the DELETE removes it). #[cfg_attr(test, derive(Debug))] #[derive(Default)] struct Bucketed<'a> { - /// `MailSetEntry.instance_list_id` for every CREATE / DELETE / UPDATE - /// we received — i.e. the folders whose contents changed. - folder_entry_lists: std::collections::HashSet<&'a str>, + mail_set_entry_creates: Vec<&'a EntityUpdateEvent>, + mail_set_entry_deletes: Vec<&'a EntityUpdateEvent>, /// Mail-entity updates (read/unread, subject, delete, …). mail_events: Vec<&'a EntityUpdateEvent>, /// A `MailSet` entity event arrived — the folder list itself changed @@ -109,8 +120,12 @@ fn bucket_updates(updates: &[EntityUpdateEvent]) -> Bucketed<'_> { continue; } match ev.type_id { - MAIL_SET_ENTRY_TYPE_ID => { - out.folder_entry_lists.insert(ev.instance_list_id.as_str()); + 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 => out.mail_events.push(ev), MAIL_SET_TYPE_ID => out.folder_list_dirty = true, @@ -128,108 +143,285 @@ async fn apply_batch( batch: &EntityUpdateBatch, ) { let Bucketed { - folder_entry_lists, + mail_set_entry_creates, + mail_set_entry_deletes, 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 the subsequent MailSetEntry - // re-sync (below) sees any newly created folder, and prune folders that - // disappeared from the server. + // 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 { - match backend.list_folders().await { - Ok(folders) => { - let known: std::collections::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}"), - } + refresh_folder_list(store, local_store, backend).await; } - // Any MailSetEntry CREATE/DELETE on a folder's entries list is the - // canonical signal that the folder's contents changed. Re-running - // `sync_folder` reuses the existing diff-and-update logic and is correct - // for all of CREATE / DELETE / multi-move at once. Cheaper than tracking - // per-entry id mappings; we can optimise later if it becomes a hot path. - if !folder_entry_lists.is_empty() { - let folders = store.list_folders().await; - for folder in folders - .iter() - .filter(|f| folder_entry_lists.contains(f.entries_list_id.as_str())) - { - debug!( - "Event bus: re-syncing folder {} (batch {})", - folder.imap_path, batch.batch_id - ); - if let Err(e) = sync_folder(store, local_store, backend, folder, sync_limit).await { - warn!( - "Event bus folder sync failed for {}: {}", - folder.imap_path, e - ); - } - } + // 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(); + + // 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, + 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 is - // already handled by the folder re-sync above; nothing to do here. + // CREATE on a Mail entity is paired with a MailSetEntry CREATE which + // the loop above already handled. for ev in mail_events { - 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 => { - match backend - .load_mail(&ev.instance_list_id, &ev.instance_id) - .await - { - Ok(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); - } - }, - Ok(None) => { - // Disappeared between event and our follow-up load — - // treat like a delete. - let _ = local_store.delete_mail(&ev.instance_id); - store.remove_mail_everywhere(&ev.instance_id).await; - }, - Err(e) => warn!("Mail UPDATE: failed to load {}: {}", ev.instance_id, e), - } - }, - Operation::Create | Operation::Other(_) => {}, + 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 + ); } } } +async fn refresh_folder_list( + 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}"), + } +} + +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, + 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; + }; + + // 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 { + assign_uid_and_upsert(store, local_store, target_folder, &mail_eid, &mut stored).await; + return; + } + + // MISS path: never seen this mail. Ask the backend for just that one + // mail. Needs the Mail's `list_id` (≠ the folder's entries_list_id) — + // sniff it from any cached Mail (single-MailGroup is the common case; + // caller falls back if the cache is still empty). + let Some(list_id) = store.mail_list_id().await else { + fallback_folders.insert(ev.instance_list_id.clone()); + return; + }; + match backend.load_mail(&list_id, &mail_eid).await { + Ok(Some(mail)) => { + let mut stored = StoredMail { + mail, + details: None, + rfc2822: None, + uid: 0, + }; + assign_uid_and_upsert(store, local_store, target_folder, &mail_eid, &mut stored).await; + }, + Ok(None) => { + debug!("MailSetEntry CREATE: mail {} not found on server", mail_eid); + }, + Err(e) => { + warn!( + "MailSetEntry CREATE: load_mail({}, {}) failed: {e} — falling back", + list_id, mail_eid + ); + fallback_folders.insert(ev.instance_list_id.clone()); + }, + } +} + +async fn apply_mail_set_entry_delete( + 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; + }; + store + .remove_mail_from_folder(&source_folder.id, &mail_eid) + .await; + // 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, +) { + 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 => match backend.load_mail(&ev.instance_list_id, &ev.instance_id).await { + Ok(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); + } + }, + Ok(None) => { + // Disappeared between event and our follow-up load — treat + // like a delete. + let _ = local_store.delete_mail(&ev.instance_id); + store.remove_mail_everywhere(&ev.instance_id).await; + }, + Err(e) => warn!("Mail UPDATE: failed to load {}: {}", ev.instance_id, e), + }, + Operation::Create | Operation::Other(_) => {}, + } +} + +/// 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, +) { + 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); + } + + store + .upsert_mail_in_folder(&target_folder.id, stored.clone()) + .await; +} + #[cfg(test)] mod tests { use super::*; @@ -249,41 +441,62 @@ mod tests { #[test] fn bucket_empty_batch() { let out = bucket_updates(&[]); - assert!(out.folder_entry_lists.is_empty()); + 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() { - // `sys`-app events (e.g. group/user changes) must not affect mail buckets. let updates = vec![ev("sys", 97, "L", "E", Operation::Create)]; let out = bucket_updates(&updates); - assert!(out.folder_entry_lists.is_empty()); + 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 (e.g. attachments, contacts) should pass through. + // Unrelated tutanota entities (attachments, contacts, …) pass through. let updates = vec![ev("tutanota", 999, "L", "E", Operation::Update)]; let out = bucket_updates(&updates); - assert!(out.folder_entry_lists.is_empty()); + 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_collects_mail_set_entry_lists() { - // Same list appearing twice (CREATE + DELETE) should de-duplicate. + 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, "inbox_entries", "e2", Operation::Delete), + 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.folder_entry_lists.len(), 2); - assert!(out.folder_entry_lists.contains("inbox_entries")); - assert!(out.folder_entry_lists.contains("sent_entries")); - assert!(out.mail_events.is_empty()); + 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] @@ -293,12 +506,24 @@ mod tests { ev("tutanota", MAIL_TYPE_ID, "mailL", "m2", Operation::Delete), ]; let out = bucket_updates(&updates); - assert!(out.folder_entry_lists.is_empty()); assert_eq!(out.mail_events.len(), 2); assert_eq!(out.mail_events[0].instance_id, "m1"); assert_eq!(out.mail_events[1].operation, Operation::Delete); } + #[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![ @@ -307,24 +532,8 @@ mod tests { ev("sys", 42, "X", "Y", Operation::Create), // ignored ]; let out = bucket_updates(&updates); - assert_eq!(out.folder_entry_lists.len(), 1); - assert!(out.folder_entry_lists.contains("inbox_entries")); + assert_eq!(out.mail_set_entry_creates.len(), 1); assert_eq!(out.mail_events.len(), 1); - assert_eq!(out.mail_events[0].instance_id, "m1"); assert!(!out.folder_list_dirty); } - - #[test] - fn bucket_marks_folder_list_dirty_on_mail_set_event() { - // Any CRUD on a MailSet (folder entity) flips the dirty flag once. - 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); - // MailSet events themselves are not bucketed as mail/entry events. - assert!(out.folder_entry_lists.is_empty()); - assert!(out.mail_events.is_empty()); - } } diff --git a/crates/bridge/src/sync.rs b/crates/bridge/src/sync.rs index 35653d7..0b0d675 100644 --- a/crates/bridge/src/sync.rs +++ b/crates/bridge/src/sync.rs @@ -32,6 +32,11 @@ pub struct MailStore { folder_list: RwLock>, generation: watch::Sender, gen_counter: std::sync::atomic::AtomicU64, + /// Cached `Mail.list_id` for this user. A user has one (per `MailGroup`) + /// and every Mail in the cache shares it; we sniff it lazily from any + /// existing Mail so the event handler can `load_mail` brand-new mail + /// ids without re-listing a folder. + mail_list_id_cache: RwLock>, } impl MailStore { @@ -42,6 +47,7 @@ impl MailStore { folder_list: RwLock::new(Vec::new()), generation: tx, gen_counter: std::sync::atomic::AtomicU64::new(0), + mail_list_id_cache: RwLock::new(None), }) } @@ -145,6 +151,114 @@ impl MailStore { } } + /// Find a mail by element id across all folders. Returns the source + /// folder id and a clone of the [`StoredMail`] entry — letting the + /// event handler reuse the already-decrypted Mail/details when the + /// same mail just hopped between two cached folders, no REST round-trip. + pub async fn find_mail_anywhere(&self, element_id: &str) -> Option<(String, StoredMail)> { + let folders = self.folders.read().await; + for (fid, mails) in folders.iter() { + for m in mails { + if m.mail + ._id + .as_ref() + .map(|id| id.element_id.to_string()) + .as_deref() + == Some(element_id) + { + return Some((fid.clone(), m.clone())); + } + } + } + None + } + + /// `true` if any folder still references this mail. Used after a + /// per-folder removal to decide whether the cached `.eml` can go. + pub async fn is_mail_anywhere(&self, element_id: &str) -> bool { + let folders = self.folders.read().await; + folders.values().any(|mails| { + mails.iter().any(|m| { + m.mail + ._id + .as_ref() + .map(|id| id.element_id.to_string()) + .as_deref() + == Some(element_id) + }) + }) + } + + /// Remove a single mail from one specific folder. Returns `true` if a + /// row was actually removed (the mail might already be gone from this + /// folder if we are reprocessing an event). + pub async fn remove_mail_from_folder(&self, folder_id: &str, element_id: &str) -> bool { + let mut folders = self.folders.write().await; + let Some(mails) = folders.get_mut(folder_id) else { + return false; + }; + let before = mails.len(); + mails.retain(|m| { + m.mail + ._id + .as_ref() + .map(|id| id.element_id.to_string()) + .as_deref() + != Some(element_id) + }); + let changed = mails.len() != before; + drop(folders); + if changed { + self.bump_generation(); + } + changed + } + + /// Insert or replace a mail in `folder_id`. Replace-by-element-id keeps + /// the operation idempotent — re-applying the same event leaves the + /// store unchanged. + pub async fn upsert_mail_in_folder(&self, folder_id: &str, mail: StoredMail) { + let Some(eid) = mail.mail._id.as_ref().map(|id| id.element_id.to_string()) else { + return; + }; + let mut folders = self.folders.write().await; + let entries = folders.entry(folder_id.to_string()).or_default(); + if let Some(slot) = entries.iter_mut().find(|m| { + m.mail + ._id + .as_ref() + .map(|id| id.element_id.to_string()) + .as_deref() + == Some(&eid) + }) { + *slot = mail; + } else { + entries.push(mail); + } + drop(folders); + self.bump_generation(); + } + + /// Return the cached `Mail.list_id` for this user, sniffing it from any + /// Mail already in the store on first call. `None` only if the store + /// is still empty (very first boot, no mail seen yet). + pub async fn mail_list_id(&self) -> Option { + if let Some(cached) = self.mail_list_id_cache.read().await.clone() { + return Some(cached); + } + let folders = self.folders.read().await; + let sniffed = folders.values().find_map(|mails| { + mails + .iter() + .find_map(|m| m.mail._id.as_ref().map(|id| id.list_id.to_string())) + }); + drop(folders); + if let Some(id) = &sniffed { + *self.mail_list_id_cache.write().await = Some(id.clone()); + } + sniffed + } + /// Drop in-memory state for folders that are no longer on the server. /// Returns the ids that were removed so the caller can clean up the /// LocalStore + .eml files for them. @@ -550,7 +664,7 @@ where unreachable!() } -fn mail_to_metadata(mail: &Mail, folder_id: &str, uid: u32) -> MailMetadata { +pub(crate) fn mail_to_metadata(mail: &Mail, folder_id: &str, uid: u32) -> MailMetadata { let (list_id, element_id) = mail ._id .as_ref() @@ -730,6 +844,106 @@ mod tests { assert_eq!(store.get_folder("folderA").await.len(), 1); } + #[tokio::test] + async fn find_mail_anywhere_returns_first_match() { + let store = MailStore::new(); + store + .set_folder("A", vec![stored(make_mail("L1", "shared", "x", true), 7)]) + .await; + store + .set_folder("B", vec![stored(make_mail("L1", "shared", "x", true), 11)]) + .await; + let found = store.find_mail_anywhere("shared").await.expect("must find"); + assert!(found.0 == "A" || found.0 == "B"); + assert_eq!( + found.1.mail._id.as_ref().unwrap().element_id.to_string(), + "shared" + ); + assert!(store.find_mail_anywhere("missing").await.is_none()); + } + + #[tokio::test] + async fn is_mail_anywhere_reflects_presence() { + let store = MailStore::new(); + store + .set_folder("A", vec![stored(make_mail("L1", "e1", "s", true), 1)]) + .await; + assert!(store.is_mail_anywhere("e1").await); + assert!(!store.is_mail_anywhere("e2").await); + } + + #[tokio::test] + async fn remove_mail_from_folder_only_touches_that_folder() { + let store = MailStore::new(); + store + .set_folder( + "A", + vec![ + stored(make_mail("L1", "k", "keep", true), 1), + stored(make_mail("L1", "g", "gone", true), 2), + ], + ) + .await; + store + .set_folder("B", vec![stored(make_mail("L1", "g", "gone", true), 5)]) + .await; + + let removed = store.remove_mail_from_folder("A", "g").await; + assert!(removed); + assert_eq!(store.get_folder("A").await.len(), 1); + // B still has the mail — `remove_mail_from_folder` is scoped. + assert_eq!(store.get_folder("B").await.len(), 1); + // Idempotent: re-removing the same mail from A is a no-op. + assert!(!store.remove_mail_from_folder("A", "g").await); + } + + #[tokio::test] + async fn upsert_mail_in_folder_replaces_existing_by_element_id() { + let store = MailStore::new(); + store + .set_folder("A", vec![stored(make_mail("L1", "e1", "old", true), 4)]) + .await; + store + .upsert_mail_in_folder("A", stored(make_mail("L1", "e1", "new", false), 4)) + .await; + let mails = store.get_folder("A").await; + // Same element_id → replaced in place, no duplicate. + assert_eq!(mails.len(), 1); + assert_eq!(mails[0].mail.subject, "new"); + assert!(!mails[0].mail.unread); + } + + #[tokio::test] + async fn upsert_mail_in_folder_appends_when_new() { + let store = MailStore::new(); + store + .set_folder("A", vec![stored(make_mail("L1", "e1", "one", true), 1)]) + .await; + store + .upsert_mail_in_folder("A", stored(make_mail("L1", "e2", "two", true), 2)) + .await; + assert_eq!(store.get_folder("A").await.len(), 2); + } + + #[tokio::test] + async fn mail_list_id_sniffs_lazily_and_caches() { + let store = MailStore::new(); + // No mails yet — nothing to sniff. + assert!(store.mail_list_id().await.is_none()); + store + .set_folder("A", vec![stored(make_mail("listAAA", "e1", "s", true), 1)]) + .await; + assert_eq!(store.mail_list_id().await.as_deref(), Some("listAAA")); + // Now the cache is populated — calling again returns the same value + // even if the store changes (cache is intentionally sticky for a + // session; a different list_id would only appear with a different + // MailGroup, which forces a full restart anyway). + store + .set_folder("A", vec![stored(make_mail("listBBB", "e1", "s", true), 1)]) + .await; + assert_eq!(store.mail_list_id().await.as_deref(), Some("listAAA")); + } + #[tokio::test] async fn prune_unknown_folders_drops_disappeared_ones() { let store = MailStore::new();