Realtime sync via the SDK event bus, drop the periodic poll

The 60s list-sync loop is replaced by the WebSocket event bus from the
SDK (sdk-event-bus). On startup the syncer still does Phase 0 (load the
local store into memory), then a one-shot bootstrap sync only if no
event-bus catch-up state is cached. From there on:

- `EventBusClient` runs in its own task, streams `EventBusMessage`s into
  an mpsc channel and reconnects with backoff.
- `event_handler` consumes the channel: MailSetEntry CREATE/DELETE
  triggers a targeted `sync_folder` for the affected folder; Mail UPDATE
  refreshes metadata in place; Mail DELETE drops the cache + .eml.
- After each batch the `(group_id, batch_id)` is persisted in the new
  `event_bus_state` SQLite table (schema bumped to v4) and mirrored in
  the bus's in-memory map, so the next reconnect resumes catch-up via
  `groupsToLastEventBatchIds`.

`stop()` aborts and awaits the new bus + handler tasks alongside the
existing syncer/IMAP/SMTP teardown, so ports release before the next
start rebinds them.

138/138 bridge unit tests pass (incl. 2 new ones for the event-bus state
table). End-to-end behaviour to be verified against the live server.

Cargo.lock

@@ -929,6 +929,12 @@ dependencies = [
  "parking_lot_core",
 ]
 
+[[package]]
+name = "data-encoding"
+version = "2.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8"
+
 [[package]]
 name = "dbus"
 version = "0.9.11"
@@ -4173,6 +4179,17 @@ dependencies = [
  "stable_deref_trait",
 ]
 
+[[package]]
+name = "sha1"
+version = "0.10.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba"
+dependencies = [
+ "cfg-if",
+ "cpufeatures",
+ "digest",
+]
+
 [[package]]
 name = "sha2"
 version = "0.10.9"
@@ -4989,6 +5006,22 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "tokio-tungstenite"
+version = "0.24.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "edc5f74e248dc973e0dbb7b74c7e0d6fcc301c694ff50049504004ef4d0cdcd9"
+dependencies = [
+ "futures-util",
+ "log 0.4.29",
+ "rustls",
+ "rustls-pki-types",
+ "tokio",
+ "tokio-rustls",
+ "tungstenite",
+ "webpki-roots 0.26.11",
+]
+
 [[package]]
 name = "tokio-util"
 version = "0.7.18"
@@ -5249,6 +5282,26 @@ dependencies = [
  "syn 2.0.117",
 ]
 
+[[package]]
+name = "tungstenite"
+version = "0.24.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "18e5b8366ee7a95b16d32197d0b2604b43a0be89dc5fac9f8e96ccafbaedda8a"
+dependencies = [
+ "byteorder",
+ "bytes",
+ "data-encoding",
+ "http",
+ "httparse",
+ "log 0.4.29",
+ "rand 0.8.6",
+ "rustls",
+ "rustls-pki-types",
+ "sha1",
+ "thiserror 1.0.69",
+ "utf-8",
+]
+
 [[package]]
 name = "tuta-sdk"
 version = "348.260526.0"
@@ -5288,6 +5341,7 @@ dependencies = [
  "time",
  "time-tz",
  "tokio",
+ "tokio-tungstenite",
  "uniffi",
  "util",
  "x25519-dalek",
@@ -5869,6 +5923,24 @@ dependencies = [
  "system-deps",
 ]
 
+[[package]]
+name = "webpki-roots"
+version = "0.26.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9"
+dependencies = [
+ "webpki-roots 1.0.7",
+]
+
+[[package]]
+name = "webpki-roots"
+version = "1.0.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "52f5ee44c96cf55f1b349600768e3ece3a8f26010c05265ab73f945bb1a2eb9d"
+dependencies = [
+ "rustls-pki-types",
+]
+
 [[package]]
 name = "webview2-com"
 version = "0.38.2"

crates/bridge/src/bridge.rs

@@ -2,11 +2,20 @@ use std::sync::Arc;
 use tokio::sync::{broadcast, oneshot, watch, RwLock};
 
 use crate::config::{self, Config};
+use crate::event_handler;
 use crate::store::LocalStore;
 use crate::sync::{self, MailStore};
 use crate::tuta::{self, MailBackend, TwoFactorCallback};
 use crate::{imap, smtp, tls};
 
+// Tuta `modelVersions=` for the event-bus URL. The server uses these to
+// validate compatibility. Keep in sync with the vendored SDK
+// (`tuta-sdk/.../type_models/{sys,tutanota}.json` `version`).
+const SYS_MODEL_VERSION: u32 = 150;
+const TUTANOTA_MODEL_VERSION: u32 = 108;
+/// Identifier the server uses for telemetry/rate-limit bucketing.
+const CLIENT_NAME: &str = "tutabridge";
+
 #[derive(Debug, Clone, PartialEq, serde::Serialize)]
 pub enum BridgeStatus {
     Stopped,
@@ -127,6 +136,15 @@ impl BridgeHandle {
         let local_store = Arc::new(local_store);
         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.
+        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);
         let store = MailStore::new();
         self.store = Some(store.clone());
@@ -142,6 +160,32 @@ impl BridgeHandle {
         let sync_limit = config.sync_limit;
         let pw = config.bridge_password.clone();
 
+        // Build the realtime event bus and hydrate its catch-up state from
+        // disk so the next reconnect resumes from the last processed batch.
+        let bus_client = Arc::new(tutasdk::event_bus::EventBusClient::new(
+            bus_base_url,
+            SYS_MODEL_VERSION,
+            TUTANOTA_MODEL_VERSION,
+            tutasdk::CLIENT_VERSION.to_string(),
+            CLIENT_NAME.to_string(),
+        ));
+        {
+            let ids_handle = bus_client.last_batch_ids();
+            match local_store.load_event_bus_state() {
+                Ok(s) if !s.is_empty() => {
+                    let mut m = ids_handle.lock().unwrap();
+                    m.extend(s);
+                    self.emit_log(&format!(
+                        "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}")),
+            }
+        }
+        let bus_ids_for_handler = bus_client.last_batch_ids();
+
         let task = tokio::spawn(async move {
             let imap_tls = tls_acceptor.clone();
             let smtp_tls = tls_acceptor;
@@ -149,12 +193,44 @@ impl BridgeHandle {
             let _ = log_tx.send(format!("IMAP listening on 127.0.0.1:{imap_port}"));
             let _ = log_tx.send(format!("SMTP listening on 127.0.0.1:{smtp_port}"));
 
+            // mpsc channel from event bus -> handler.
+            let (event_tx, event_rx) = tokio::sync::mpsc::channel(64);
+
             let syncer_handle = tokio::spawn(sync::run_syncer(
+                store.clone(),
+                local_store.clone(),
+                backend.clone(),
+                sync_limit,
+                shutdown_sync_rx.clone(),
+            ));
+            let bus_handle = {
+                let client = Arc::clone(&bus_client);
+                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 {
+                            tutasdk::event_bus::EventBusError::Stopped => {},
+                            _ => log::warn!("Event bus exited: {e}"),
+                        }
+                    }
+                })
+            };
+            let handler_handle = tokio::spawn(event_handler::run_event_handler(
                 store.clone(),
                 local_store,
                 backend.clone(),
                 sync_limit,
-                shutdown_sync_rx,
+                bus_ids_for_handler,
+                event_rx,
+                shutdown_sync_rx.clone(),
             ));
             let mut imap_handle = tokio::spawn(imap::serve(
                 imap_port,
@@ -186,11 +262,19 @@ impl BridgeHandle {
             // a subsequent start rebinds them. Skip awaiting a handle that already
             // resolved in the select above (re-polling it would panic).
             syncer_handle.abort();
+            bus_handle.abort();
+            handler_handle.abort();
             imap_handle.abort();
             smtp_handle.abort();
             if !syncer_handle.is_finished() {
                 let _ = syncer_handle.await;
             }
+            if !bus_handle.is_finished() {
+                let _ = bus_handle.await;
+            }
+            if !handler_handle.is_finished() {
+                let _ = handler_handle.await;
+            }
             if !imap_handle.is_finished() {
                 let _ = imap_handle.await;
             }

crates/bridge/src/event_handler.rs

@@ -0,0 +1,173 @@
+//! Realtime event-bus handler.
+//!
+//! 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`.
+
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+
+use log::{debug, info, warn};
+use tokio::sync::{mpsc, watch};
+use tutasdk::event_bus::{EntityUpdateBatch, EntityUpdateEvent, EventBusMessage, Operation};
+
+use crate::store::LocalStore;
+use crate::sync::{sync_folder, MailStore};
+use crate::tuta::MailBackend;
+
+/// Application tag for Tuta's mail-side entities.
+const TUTANOTA_APP: &str = "tutanota";
+/// `Mail` entity (see `tuta-sdk/.../entities/generated/tutanota.rs`,
+/// `impl Entity for Mail`).
+const MAIL_TYPE_ID: i64 = 97;
+/// `MailSetEntry` entity — placement of a mail inside a folder/MailSet.
+const MAIL_SET_ENTRY_TYPE_ID: i64 = 1450;
+
+pub async fn run_event_handler(
+	store: Arc<MailStore>,
+	local_store: Arc<LocalStore>,
+	backend: Arc<dyn MailBackend>,
+	sync_limit: usize,
+	last_batch_ids: Arc<Mutex<HashMap<String, String>>>,
+	mut rx: mpsc::Receiver<EventBusMessage>,
+	mut shutdown: watch::Receiver<bool>,
+) {
+	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<HashMap<String, String>>,
+	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,
+	};
+
+	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);
+	}
+}
+
+async fn apply_batch(
+	store: &MailStore,
+	local_store: &LocalStore,
+	backend: &dyn MailBackend,
+	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),
+			_ => {},
+		}
+	}
+
+	// 2) 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))
+		{
+			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
+				);
+			}
+		}
+	}
+
+	// 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.
+	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(_) => {},
+		}
+	}
+}

crates/bridge/src/imap/session.rs

@@ -1397,6 +1397,21 @@ mod tests {
         async fn load_mail_ids_for_folder(&self, _folder: &FolderInfo, _limit: usize) -> Result<Vec<Mail>, String> {
             Ok(self.mails.lock().unwrap().clone())
         }
+        async fn load_mail(&self, _list_id: &str, element_id: &str) -> Result<Option<Mail>, String> {
+            Ok(self
+                .mails
+                .lock()
+                .unwrap()
+                .iter()
+                .find(|m| {
+                    m._id
+                        .as_ref()
+                        .map(|id| id.element_id.to_string())
+                        .as_deref()
+                        == Some(element_id)
+                })
+                .cloned())
+        }
         async fn load_mail_details(&self, mail: &Mail) -> Result<Option<MailDetails>, String> {
             let key = mail._id.as_ref().map(|id| id.element_id.to_string()).unwrap_or_default();
             Ok(self.details.lock().unwrap().get(&key).cloned())

crates/bridge/src/lib.rs

@@ -1,5 +1,6 @@
 pub mod bridge;
 pub mod config;
+pub mod event_handler;
 pub mod store;
 pub mod sync;
 pub mod tuta;

crates/bridge/src/store.rs

@@ -8,9 +8,9 @@ use log::{debug, warn};
 use rusqlite::{Connection, OptionalExtension};
 
 /// Bumped when the on-disk schema changes. A mismatch drops the cached tables
-/// (mails + sync_state) and triggers a full re-sync; encrypted .eml files are
+/// (mails + sync_state + event_bus_state) and triggers a full re-sync;
-/// keyed by element id and survive the migration.
+/// encrypted .eml files are keyed by element id and survive the migration.
-const SCHEMA_VERSION: &str = "3";
+const SCHEMA_VERSION: &str = "4";
 
 #[derive(Debug, thiserror::Error)]
 pub enum StoreError {
@@ -82,7 +82,11 @@ impl LocalStore {
                 "Local store schema {:?} != {SCHEMA_VERSION}, dropping cache tables",
                 version
             );
-            conn.execute_batch("DROP TABLE IF EXISTS mails; DROP TABLE IF EXISTS sync_state;")?;
+            conn.execute_batch(
+                "DROP TABLE IF EXISTS mails;
+                 DROP TABLE IF EXISTS sync_state;
+                 DROP TABLE IF EXISTS event_bus_state;",
+            )?;
         }
 
         conn.execute_batch(&format!(
@@ -106,6 +110,11 @@ impl LocalStore {
                 last_sync_ms  INTEGER NOT NULL DEFAULT 0,
                 next_uid      INTEGER NOT NULL DEFAULT 1
             );
+            CREATE TABLE IF NOT EXISTS event_bus_state (
+                group_id       TEXT PRIMARY KEY,
+                last_batch_id  TEXT NOT NULL,
+                updated_at_ms  INTEGER NOT NULL DEFAULT 0
+            );
             INSERT OR REPLACE INTO store_meta(key, value) VALUES ('schema_version', '{SCHEMA_VERSION}');"
         ))?;
 
@@ -392,6 +401,92 @@ impl LocalStore {
             conn.query_row("SELECT COUNT(*) FROM mails", [], |row| row.get(0))?;
         Ok(count as usize)
     }
+
+    /// Load the last processed event-batch id for every known group.
+    pub fn load_event_bus_state(
+        &self,
+    ) -> Result<std::collections::HashMap<String, String>, StoreError> {
+        let conn = self.conn.lock().unwrap();
+        let mut stmt = conn.prepare("SELECT group_id, last_batch_id FROM event_bus_state")?;
+        let rows = stmt.query_map([], |row| {
+            Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
+        })?;
+        let mut out = std::collections::HashMap::new();
+        for row in rows {
+            let (g, b) = row?;
+            out.insert(g, b);
+        }
+        Ok(out)
+    }
+
+    /// 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> {
+        let conn = self.conn.lock().unwrap();
+        let now_ms = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|d| d.as_millis() as i64)
+            .unwrap_or(0);
+        conn.execute(
+            "INSERT INTO event_bus_state(group_id, last_batch_id, updated_at_ms)
+             VALUES (?1, ?2, ?3)
+             ON CONFLICT(group_id) DO UPDATE SET
+                last_batch_id = excluded.last_batch_id,
+                updated_at_ms = excluded.updated_at_ms",
+            rusqlite::params![group_id, batch_id, now_ms],
+        )?;
+        Ok(())
+    }
+
+    /// Refresh the per-mail fields (read/unread, subject, sender, JSON blob)
+    /// for every row of this `element_id` — typically a mail lives in one
+    /// folder, but Tuta's model allows multi-folder placement, so we update
+    /// all matching rows. Folder placement itself is governed by
+    /// `MailSetEntry` events.
+    pub fn refresh_mail_fields(
+        &self,
+        element_id: &str,
+        subject: &str,
+        sender_name: &str,
+        sender_address: &str,
+        unread: bool,
+        mail_json: &str,
+    ) -> Result<(), StoreError> {
+        let conn = self.conn.lock().unwrap();
+        conn.execute(
+            "UPDATE mails SET
+                subject = ?1,
+                sender_name = ?2,
+                sender_address = ?3,
+                unread = ?4,
+                mail_json = ?5
+             WHERE element_id = ?6",
+            rusqlite::params![
+                subject,
+                sender_name,
+                sender_address,
+                unread as i64,
+                mail_json,
+                element_id,
+            ],
+        )?;
+        Ok(())
+    }
+
+    /// Drop a mail entirely (metadata + .eml). Used by the event handler on a
+    /// `DELETE` of the underlying mail entity.
+    pub fn delete_mail(&self, element_id: &str) -> Result<(), StoreError> {
+        {
+            let conn = self.conn.lock().unwrap();
+            conn.execute("DELETE FROM mails WHERE element_id = ?1", [element_id])?;
+        }
+        self.delete_eml(element_id)?;
+        Ok(())
+    }
 }
 
 #[cfg(test)]
@@ -537,6 +632,37 @@ mod tests {
         assert!(store.verify_key());
     }
 
+    #[test]
+    fn event_bus_state_roundtrip() {
+        let store = open_memory_store();
+        assert!(store.load_event_bus_state().unwrap().is_empty());
+
+        store.set_event_bus_batch_id("group1", "batchA").unwrap();
+        store.set_event_bus_batch_id("group2", "batchX").unwrap();
+
+        let s = store.load_event_bus_state().unwrap();
+        assert_eq!(s.get("group1"), Some(&"batchA".to_string()));
+        assert_eq!(s.get("group2"), Some(&"batchX".to_string()));
+
+        // Upsert overwrites.
+        store.set_event_bus_batch_id("group1", "batchB").unwrap();
+        let s = store.load_event_bus_state().unwrap();
+        assert_eq!(s.get("group1"), Some(&"batchB".to_string()));
+    }
+
+    #[test]
+    fn delete_mail_removes_metadata_and_eml() {
+        let store = open_memory_store();
+        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"));
+
+        store.delete_mail("del1").unwrap();
+        assert_eq!(store.mail_count("inbox").unwrap(), 0);
+        assert!(!store.has_eml("del1"));
+    }
+
     #[test]
     fn test_mark_has_details() {
         let store = open_memory_store();

crates/bridge/src/sync.rs

@@ -12,7 +12,6 @@ use crate::tuta::{FolderInfo, MailBackend};
 
 const INTER_REQUEST_DELAY: Duration = Duration::from_millis(150);
 const INTER_FOLDER_DELAY: Duration = Duration::from_millis(300);
-const SYNC_INTERVAL: Duration = Duration::from_secs(60);
 /// Pause between background body-prefetch sweeps.
 const PREFETCH_INTERVAL: Duration = Duration::from_secs(30);
 const MAX_RETRIES: u32 = 3;
@@ -115,6 +114,61 @@ impl MailStore {
         self.bump_generation();
     }
 
+    /// 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 {
+            return;
+        };
+        let mut folders = self.folders.write().await;
+        let mut changed = false;
+        for mails in folders.values_mut() {
+            if let Some(m) = mails.iter_mut().find(|m| {
+                m.mail
+                    ._id
+                    .as_ref()
+                    .map(|id| id.element_id.to_string())
+                    .as_deref()
+                    == Some(&eid)
+            }) {
+                m.mail = mail.clone();
+                changed = true;
+            }
+        }
+        drop(folders);
+        if changed {
+            self.bump_generation();
+        }
+    }
+
+    /// Drop a mail from every folder it appears in (handles DELETE events).
+    pub async fn remove_mail_everywhere(&self, element_id: &str) {
+        let mut folders = self.folders.write().await;
+        let mut changed = false;
+        for mails in folders.values_mut() {
+            let before = mails.len();
+            mails.retain(|m| {
+                m.mail
+                    ._id
+                    .as_ref()
+                    .map(|id| id.element_id.to_string())
+                    .as_deref()
+                    != Some(element_id)
+            });
+            if mails.len() != before {
+                changed = true;
+            }
+        }
+        drop(folders);
+        if changed {
+            self.bump_generation();
+        }
+    }
+
     async fn update_mail_details(
         &self,
         folder_id: &str,
@@ -186,68 +240,36 @@ pub async fn run_syncer(
         }
     }
 
-    // Run the fast list/folder sync and the slow body prefetch as independent
+    // Bootstrap: if we have no cached event-bus catch-up state, the on-disk
-    // loops: a long prefetch pass must never block folder / new-mail refresh.
+    // cache may be stale or empty. Run a one-shot full list sync of every
-    tokio::join!(
+    // folder so the store reflects current server state; from then on the
-        list_sync_loop(&store, &local_store, &*backend, sync_limit, shutdown.clone()),
+    // event bus drives all updates (no periodic polling).
-        prefetch_loop(&store, &local_store, &*backend, shutdown.clone()),
+    let needs_bootstrap = match local_store.load_event_bus_state() {
-    );
+        Ok(s) => s.is_empty(),
-    info!("Mail syncer shutting down");
+        Err(e) => {
-}
+            warn!("Could not read event_bus_state ({e}); assuming bootstrap needed");
-
+            true
-/// Fast loop: refresh the folder list and sync the mail-id list for every
-/// folder. This is what surfaces new folders and new mail (~every
-/// `SYNC_INTERVAL`), independent of the slow body prefetch.
-async fn list_sync_loop(
-    store: &MailStore,
-    local_store: &LocalStore,
-    backend: &dyn MailBackend,
-    sync_limit: usize,
-    mut shutdown: watch::Receiver<bool>,
-) {
-    let mut cycle_backoff = Duration::ZERO;
-    loop {
-        if *shutdown.borrow() {
-            return;
         }
-        let mut had_error = false;
+    };
-
+    if needs_bootstrap && !folders.is_empty() {
-        let folders = match retry(|| backend.list_folders()).await {
+        info!("Bootstrap sync (no cached event-bus state)");
-            Ok(folders) => {
-                store.set_folder_list(folders.clone()).await;
-                folders
-            }
-            Err(e) => {
-                warn!("Failed to refresh folder list: {e}");
-                had_error = true;
-                store.list_folders().await
-            }
-        };
-
         for folder in &folders {
             if *shutdown.borrow() {
                 return;
             }
-            if let Err(e) = sync_folder(store, local_store, backend, folder, sync_limit).await {
+            if let Err(e) = sync_folder(&store, &local_store, &*backend, folder, sync_limit).await {
-                warn!("Sync error for {}: {}", folder.imap_path, e);
+                warn!("Bootstrap sync failed for {}: {}", folder.imap_path, e);
-                had_error = true;
             }
             tokio::time::sleep(INTER_FOLDER_DELAY).await;
         }
-
+    } else if !needs_bootstrap {
-        cycle_backoff = if had_error {
+        debug!("Skipping bootstrap sync — event-bus catch-up will reconcile");
-            backoff(cycle_backoff)
-        } else {
-            Duration::ZERO
-        };
-        let wait = SYNC_INTERVAL + cycle_backoff;
-        debug!("Next list sync in {:?}", wait);
-
-        tokio::select! {
-            _ = tokio::time::sleep(wait) => {}
-            _ = shutdown.changed() => return,
-        }
     }
+
+    // From here on the syncer only owns the slow body prefetch. Folder /
+    // new-mail refresh is driven by the event bus + `event_handler`.
+    prefetch_loop(&store, &local_store, &*backend, shutdown.clone()).await;
+    info!("Mail syncer shutting down");
 }
 
 /// Slow loop: progressively prefetch mail bodies in the background, on its own
@@ -320,7 +342,7 @@ async fn load_cached_folder(
     Ok(count)
 }
 
-async fn sync_folder(
+pub(crate) async fn sync_folder(
     store: &MailStore,
     local_store: &LocalStore,
     backend: &dyn MailBackend,

crates/bridge/src/tuta.rs

@@ -44,6 +44,10 @@ 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<Vec<Mail>, 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.
+    async fn load_mail(&self, list_id: &str, element_id: &str) -> Result<Option<Mail>, String>;
     async fn load_mail_details(&self, mail: &Mail) -> Result<Option<MailDetails>, String>;
     /// Enumerate all mail folders (system + custom, with hierarchy).
     async fn list_folders(&self) -> Result<Vec<FolderInfo>, String>;
@@ -82,6 +86,9 @@ fn system_special_use(kind: MailSetKind) -> Option<&'static str> {
 pub struct TutaSession {
     pub logged_in: Arc<LoggedInSdk>,
     pub email: String,
+    /// Bearer token issued at login; used to authenticate REST requests and
+    /// the realtime event-bus WebSocket query string.
+    pub access_token: String,
 }
 
 impl TutaSession {
@@ -96,6 +103,46 @@ impl TutaSession {
             .await
     }
 
+    pub fn user_id(&self) -> Option<String> {
+        self.logged_in.get_user_id().map(|id| id.0)
+    }
+
+    /// Group ids the event bus should subscribe to: all of the user's
+    /// memberships except mailing lists, plus the user group itself
+    /// (mirrors `EventBusClient.eventGroups()` in the TS worker).
+    pub fn event_groups(&self) -> Vec<String> {
+        use tutasdk::tutanota_constants::GroupType;
+        let user = self.logged_in.get_user();
+        let mut groups: Vec<String> = user
+            .memberships
+            .iter()
+            .filter(|m| m.groupType != Some(GroupType::MailingList as i64))
+            .map(|m| m.group.to_string())
+            .collect();
+        groups.push(user.userGroup.group.to_string());
+        groups
+    }
+
+    pub async fn load_mail_by_id(
+        &self,
+        list_id: &str,
+        element_id: &str,
+    ) -> Result<Option<Mail>, ApiCallError> {
+        let id = IdTupleGenerated {
+            list_id: tutasdk::GeneratedId(list_id.to_string()),
+            element_id: tutasdk::GeneratedId(element_id.to_string()),
+        };
+        match self.crypto_client().load::<Mail, _>(&id).await {
+            Ok(mail) => Ok(Some(mail)),
+            // Treat "not found" gracefully — the entity may have just been
+            // deleted server-side between the event and our follow-up load.
+            Err(ApiCallError::ServerResponseError {
+                source: tutasdk::rest_error::HttpError::NotFoundError,
+            }) => Ok(None),
+            Err(e) => Err(e),
+        }
+    }
+
     pub async fn derive_storage_key(&self) -> Result<GenericAesKey, String> {
         let user_group_id = self.logged_in.get_user_group_id();
         let versioned_key = self
@@ -249,6 +296,12 @@ impl MailBackend for TutaSession {
             .map_err(|e| format!("{e}"))
     }
 
+    async fn load_mail(&self, list_id: &str, element_id: &str) -> Result<Option<Mail>, String> {
+        self.load_mail_by_id(list_id, element_id)
+            .await
+            .map_err(|e| format!("{e}"))
+    }
+
     async fn load_mail_details(&self, mail: &Mail) -> Result<Option<MailDetails>, String> {
         self.load_mail_details_impl(mail)
             .await
@@ -396,11 +449,13 @@ pub async fn login_with_2fa(
 
     if let Some(credentials) = load_credentials(&cfg.email) {
         log::info!("Resuming saved session...");
+        let access_token = credentials.access_token.clone();
         match sdk.login(credentials).await {
             Ok(logged_in) => {
                 return Ok(TutaSession {
                     logged_in,
                     email: cfg.email.clone(),
+                    access_token,
                 });
             }
             Err(e) => {
@@ -474,6 +529,7 @@ pub async fn login_with_2fa(
     Ok(TutaSession {
         logged_in,
         email: cfg.email.clone(),
+        access_token: credentials.access_token,
     })
 }