Extract a pure bucket_updates and unit-test the new mail-side code

Pull the event-routing decision out of event_handler::apply_batch into a
pure bucket_updates(&[EntityUpdateEvent]) -> Bucketed function so it can
be tested without standing up a MailStore or hitting the network. Six
new tests cover empty batches, foreign apps, unknown type ids, the
MailSetEntry-list de-duplication, mail-event ordering and a mixed batch.

Add four MailStore tests for the new helpers: refresh_mail_in_place
must update the metadata in every folder that holds the mail (Tuta's
model allows multi-folder placement) while preserving the per-folder
UID, and must no-op on an unknown id; remove_mail_everywhere drops from
all folders and no-ops on an unknown id.

Drop the dead `let _ = bus_event_groups;` in bridge.rs and document why
event_groups() is not passed to the bus: the WebSocket subscribes
implicitly via the auth, and the URL's `groupsToLastEventBatchIds=` is
purely a per-group catch-up cursor.

148/148 lib tests pass.

crates/bridge/src/bridge.rs

@@ -137,12 +137,14 @@ impl BridgeHandle {
         self.emit_log("Local store opened");
 
         // Seed the realtime event bus before we move `session` into the
-        // backend Arc: we need its access_token / user_id / membership groups.
+        // backend Arc. We do not pass `event_groups()` to the bus: the URL's
+        // `groupsToLastEventBatchIds=` is purely a per-group catch-up cursor
+        // built from `last_batch_ids`, and the authenticated WebSocket already
+        // implicitly subscribes to every group the user is a member of.
         let bus_access_token = session.access_token.clone();
         let bus_user_id = session
             .user_id()
             .ok_or_else(|| "Missing user id from session".to_string())?;
-        let bus_event_groups = session.event_groups();
         let bus_base_url = config.api_url.clone();
 
         let backend: Arc<dyn MailBackend> = Arc::new(session);
@@ -208,12 +210,6 @@ impl BridgeHandle {
                 let token = bus_access_token;
                 let uid = bus_user_id;
                 let shutdown = shutdown_sync_rx.clone();
-                // Multiple memberships are still surfaced as their group ids
-                // via `bus_event_groups`; even if our handler only acts on
-                // mail-related events, the bus query string already lists
-                // every group from `last_batch_ids`, so the server replays
-                // missed events for all of them.
-                let _ = bus_event_groups;
                 tokio::spawn(async move {
                     if let Err(e) = client.run(token, uid, event_tx, shutdown).await {
                         match e {

crates/bridge/src/event_handler.rs

@@ -81,6 +81,37 @@ 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) is pure and has no I/O — that lets us test it in
+/// isolation.
+#[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-entity updates (read/unread, subject, delete, …).
+	mail_events: Vec<&'a EntityUpdateEvent>,
+}
+
+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 => {
+				out.folder_entry_lists.insert(ev.instance_list_id.as_str());
+			},
+			MAIL_TYPE_ID => out.mail_events.push(ev),
+			_ => {},
+		}
+	}
+	out
+}
+
 async fn apply_batch(
 	store: &MailStore,
 	local_store: &LocalStore,
@@ -88,23 +119,12 @@ async fn apply_batch(
 	sync_limit: usize,
 	batch: &EntityUpdateBatch,
 ) {
-	// 1) Bucket updates by what they affect.
-	let mut folder_entry_lists: std::collections::HashSet<String> = Default::default();
-	let mut mail_events: Vec<&EntityUpdateEvent> = Vec::new();
-	for ev in &batch.updates {
-		if ev.application != TUTANOTA_APP {
-			continue;
-		}
-		match ev.type_id {
-			MAIL_SET_ENTRY_TYPE_ID => {
-				folder_entry_lists.insert(ev.instance_list_id.clone());
-			},
-			MAIL_TYPE_ID => mail_events.push(ev),
-			_ => {},
-		}
-	}
+	let Bucketed {
+		folder_entry_lists,
+		mail_events,
+	} = bucket_updates(&batch.updates);
 
-	// 2) Any MailSetEntry CREATE/DELETE on a folder's entries list is the
+	// 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
@@ -113,7 +133,7 @@ async fn apply_batch(
 		let folders = store.list_folders().await;
 		for folder in folders
 			.iter()
-			.filter(|f| folder_entry_lists.contains(&f.entries_list_id))
+			.filter(|f| folder_entry_lists.contains(f.entries_list_id.as_str()))
 		{
 			debug!(
 				"Event bus: re-syncing folder {} (batch {})",
@@ -171,3 +191,88 @@ async fn apply_batch(
 		}
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	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,
+		}
+	}
+
+	#[test]
+	fn bucket_empty_batch() {
+		let out = bucket_updates(&[]);
+		assert!(out.folder_entry_lists.is_empty());
+		assert!(out.mail_events.is_empty());
+	}
+
+	#[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_events.is_empty());
+	}
+
+	#[test]
+	fn bucket_ignores_unknown_type_ids_in_tutanota() {
+		// Unrelated tutanota entities (e.g. attachments, contacts) should 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_events.is_empty());
+	}
+
+	#[test]
+	fn bucket_collects_mail_set_entry_lists() {
+		// Same list appearing twice (CREATE + DELETE) should de-duplicate.
+		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, "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());
+	}
+
+	#[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!(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_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
+			ev("tutanota", 429, "folderL", "f1", Operation::Create), // MailSet — unhandled here
+		];
+		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_events.len(), 1);
+		assert_eq!(out.mail_events[0].instance_id, "m1");
+	}
+}

crates/bridge/src/sync.rs

@@ -558,3 +558,151 @@ fn backoff(current: Duration) -> Duration {
     };
     next.min(Duration::from_secs(120))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tutasdk::date::DateTime;
+    use tutasdk::entities::generated::tutanota::MailAddress;
+    use tutasdk::{GeneratedId, IdTupleGenerated};
+
+    fn id(s: &str) -> GeneratedId {
+        GeneratedId(s.to_string())
+    }
+
+    /// Minimal `Mail` fixture for `MailStore` tests. Only the `_id` and a
+    /// couple of metadata fields are read by the helpers under test; the rest
+    /// is filled with defaults.
+    fn make_mail(list: &str, element: &str, subject: &str, unread: bool) -> Mail {
+        Mail {
+            _id: Some(IdTupleGenerated::new(id(list), id(element))),
+            _permissions: id("perm"),
+            _format: 0,
+            _ownerEncSessionKey: None,
+            subject: subject.to_string(),
+            receivedDate: DateTime::from_millis(1735130245000),
+            state: 2,
+            unread,
+            confidential: false,
+            replyType: 0,
+            _ownerGroup: None,
+            differentEnvelopeSender: None,
+            listUnsubscribe: false,
+            movedTime: None,
+            phishingStatus: 0,
+            authStatus: None,
+            method: 0,
+            recipientCount: 1,
+            encryptionAuthStatus: None,
+            _ownerKeyVersion: None,
+            processingState: 0,
+            processNeeded: false,
+            sendAt: None,
+            serverClassificationData: None,
+            _kdfNonce: None,
+            sender: MailAddress {
+                _id: None,
+                name: "Sender".to_string(),
+                address: "sender@tuta.com".to_string(),
+                contact: None,
+                _errors: Default::default(),
+            },
+            attachments: vec![],
+            conversationEntry: IdTupleGenerated::new(id("conv_list"), id("conv_elem")),
+            firstRecipient: None,
+            mailDetails: None,
+            mailDetailsDraft: None,
+            bucketKey: None,
+            sets: vec![],
+            clientSpamClassifierResult: None,
+            _errors: Default::default(),
+        }
+    }
+
+    fn stored(mail: Mail, uid: u32) -> StoredMail {
+        StoredMail {
+            mail,
+            details: None,
+            rfc2822: None,
+            uid,
+        }
+    }
+
+    #[tokio::test]
+    async fn refresh_mail_in_place_updates_metadata_in_every_folder() {
+        let store = MailStore::new();
+        // Same mail referenced in two folders (Tuta's model allows this via
+        // MailSet membership). Both rows must be updated when the entity
+        // changes — e.g. an "unread" toggle from the webmail.
+        let m_a = make_mail("L1", "M1", "Hello", true);
+        let m_b = m_a.clone();
+        store.set_folder("folderA", vec![stored(m_a, 7)]).await;
+        store.set_folder("folderB", vec![stored(m_b, 12)]).await;
+
+        let mut updated = make_mail("L1", "M1", "Hello [updated]", false);
+        // also tweak subject to verify the whole entity is swapped in.
+        updated.subject = "Hello [updated]".into();
+        store.refresh_mail_in_place(&updated).await;
+
+        let a = store.get_folder("folderA").await;
+        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!(b[0].mail.subject, "Hello [updated]");
+        assert_eq!(b[0].uid, 12);
+    }
+
+    #[tokio::test]
+    async fn refresh_mail_in_place_no_match_is_noop() {
+        let store = MailStore::new();
+        store
+            .set_folder("folderA", vec![stored(make_mail("L1", "M1", "S", true), 1)])
+            .await;
+        let stranger = make_mail("L1", "OTHER", "X", false);
+        store.refresh_mail_in_place(&stranger).await;
+        let a = store.get_folder("folderA").await;
+        assert_eq!(a.len(), 1);
+        assert_eq!(a[0].mail.subject, "S"); // unchanged
+    }
+
+    #[tokio::test]
+    async fn remove_mail_everywhere_drops_from_all_folders() {
+        let store = MailStore::new();
+        store
+            .set_folder(
+                "folderA",
+                vec![
+                    stored(make_mail("L1", "keep", "k", true), 1),
+                    stored(make_mail("L1", "gone", "g", true), 2),
+                ],
+            )
+            .await;
+        store
+            .set_folder(
+                "folderB",
+                vec![stored(make_mail("L1", "gone", "g", true), 5)],
+            )
+            .await;
+        store.remove_mail_everywhere("gone").await;
+
+        let a = store.get_folder("folderA").await;
+        let b = store.get_folder("folderB").await;
+        assert_eq!(a.len(), 1);
+        assert_eq!(
+            a[0].mail._id.as_ref().unwrap().element_id.to_string(),
+            "keep"
+        );
+        assert!(b.is_empty());
+    }
+
+    #[tokio::test]
+    async fn remove_mail_everywhere_unknown_id_is_noop() {
+        let store = MailStore::new();
+        store
+            .set_folder("folderA", vec![stored(make_mail("L1", "M1", "s", true), 1)])
+            .await;
+        store.remove_mail_everywhere("unknown").await;
+        assert_eq!(store.get_folder("folderA").await.len(), 1);
+    }
+}