Polish realtime: folder CRUD, out-of-sync fallback, model versions auto

Three Phase-3 polish items, bridge-only:

1. Folder CRUD events. A MailSet (typeId 429) event in a batch now flips
   the `Bucketed.folder_list_dirty` flag; the handler refreshes the
   folder list and prunes any folder that disappeared from the server
   (both in-memory and from LocalStore + .eml files). New helpers
   `MailStore::prune_unknown_folders` and `LocalStore::delete_folder_mails`.

2. Out-of-sync detection. The server only replays missed batches for
   ~44 days. At startup we now check the oldest `event_bus_state` row;
   if it predates that window we wipe the table so the syncer falls
   through to a bootstrap full sync instead of looping on a server
   refusal. New helpers `event_bus_state_min_updated_at_ms` and
   `clear_event_bus_state`.

3. Model versions. Drop the hard-coded `SYS_MODEL_VERSION = 150` /
   `TUTANOTA_MODEL_VERSION = 108` and read them at compile time from the
   vendored SDK's `type_models/{sys,tutanota}.json` via `include_str!` +
   `LazyLock`. They now track every SDK submodule bump automatically.

154/154 lib tests pass (6 new across bucket_marks_folder_list_dirty,
prune_unknown_folders, delete_folder_mails, event_bus_state min+clear,
parse_model_version + sanity check on the included JSON).

crates/bridge/src/bridge.rs

@@ -8,14 +8,42 @@ 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";
 
+// Tuta `modelVersions=` for the event-bus URL. Read at compile-time from the
+// vendored SDK's type-model JSONs so the values track the submodule bump
+// automatically — no more hard-coded constants going stale silently.
+const SYS_TYPE_MODELS_JSON: &str =
+    include_str!("../../../tuta-repo/tuta-sdk/rust/sdk/src/type_models/sys.json");
+const TUTANOTA_TYPE_MODELS_JSON: &str =
+    include_str!("../../../tuta-repo/tuta-sdk/rust/sdk/src/type_models/tutanota.json");
+
+fn parse_model_version(json: &str) -> u32 {
+    // The SDK guarantees every entry of an app's type-model JSON carries the
+    // same `version` field, so reading any one entry is enough.
+    let v: serde_json::Value =
+        serde_json::from_str(json).expect("type model JSON is malformed (build-time include)");
+    v.as_object()
+        .and_then(|m| m.values().next())
+        .and_then(|first| first.get("version"))
+        .and_then(|x| x.as_u64())
+        .map(|x| x as u32)
+        .expect("type model JSON has no version field")
+}
+
+fn sys_model_version() -> u32 {
+    static V: std::sync::LazyLock<u32> =
+        std::sync::LazyLock::new(|| parse_model_version(SYS_TYPE_MODELS_JSON));
+    *V
+}
+
+fn tutanota_model_version() -> u32 {
+    static V: std::sync::LazyLock<u32> =
+        std::sync::LazyLock::new(|| parse_model_version(TUTANOTA_TYPE_MODELS_JSON));
+    *V
+}
+
 #[derive(Debug, Clone, PartialEq, serde::Serialize)]
 pub enum BridgeStatus {
     Stopped,
@@ -166,12 +194,32 @@ impl BridgeHandle {
         // 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,
+            sys_model_version(),
+            tutanota_model_version(),
             tutasdk::CLIENT_VERSION.to_string(),
             CLIENT_NAME.to_string(),
         ));
         {
+            // OutOfSync detection: if the oldest cursor is older than the
+            // server's batch-replay window (~44 days), the server cannot
+            // catch us up — wipe the state so the syncer falls through to a
+            // bootstrap full sync.
+            let now_ms = std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .map(|d| d.as_millis() as i64)
+                .unwrap_or(0);
+            let expire_ms = tutasdk::event_bus::ENTITY_EVENT_BATCH_EXPIRE.as_millis() as i64;
+            if let Ok(Some(min_ms)) = local_store.event_bus_state_min_updated_at_ms() {
+                if now_ms - min_ms > expire_ms {
+                    self.emit_log(
+                        "Cached event-bus state is older than 44 days — wiping and forcing a full re-sync",
+                    );
+                    if let Err(e) = local_store.clear_event_bus_state() {
+                        self.emit_log(&format!("Could not clear event_bus_state: {e}"));
+                    }
+                }
+            }
+
             let ids_handle = bus_client.last_batch_ids();
             match local_store.load_event_bus_state() {
                 Ok(s) if !s.is_empty() => {
@@ -301,3 +349,25 @@ impl BridgeHandle {
         let _ = self.log_tx.send(msg.to_string());
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_model_version_extracts_first_entry_version() {
+        let json = r#"{
+            "0": {"name":"Foo","app":"sys","version":150,"id":0},
+            "1": {"name":"Bar","app":"sys","version":150,"id":1}
+        }"#;
+        assert_eq!(parse_model_version(json), 150);
+    }
+
+    #[test]
+    fn sys_and_tutanota_model_versions_are_positive() {
+        // The included JSONs must always carry a positive version; if this
+        // ever returns 0 something is very wrong with the vendored SDK.
+        assert!(sys_model_version() > 0);
+        assert!(tutanota_model_version() > 0);
+    }
+}

crates/bridge/src/event_handler.rs

@@ -24,6 +24,9 @@ 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).
+const MAIL_SET_TYPE_ID: i64 = 429;
 
 pub async fn run_event_handler(
 	store: Arc<MailStore>,
@@ -83,8 +86,8 @@ 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.
+/// metadata refresh, whether the folder list itself changed) is pure and
+/// has no I/O — that lets us test it in isolation.
 #[cfg_attr(test, derive(Debug))]
 #[derive(Default)]
 struct Bucketed<'a> {
@@ -93,6 +96,10 @@ struct Bucketed<'a> {
 	folder_entry_lists: std::collections::HashSet<&'a str>,
 	/// Mail-entity updates (read/unread, subject, delete, …).
 	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<'_> {
@@ -106,6 +113,7 @@ fn bucket_updates(updates: &[EntityUpdateEvent]) -> Bucketed<'_> {
 				out.folder_entry_lists.insert(ev.instance_list_id.as_str());
 			},
 			MAIL_TYPE_ID => out.mail_events.push(ev),
+			MAIL_SET_TYPE_ID => out.folder_list_dirty = true,
 			_ => {},
 		}
 	}
@@ -122,8 +130,38 @@ async fn apply_batch(
 	let Bucketed {
 		folder_entry_lists,
 		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.
+	if folder_list_dirty {
+		match backend.list_folders().await {
+			Ok(folders) => {
+				let known: std::collections::HashSet<String> =
+					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}"),
+		}
+	}
+
 	// 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
@@ -267,12 +305,26 @@ mod tests {
 			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");
+		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());
 	}
 }

crates/bridge/src/store.rs

@@ -419,6 +419,28 @@ impl LocalStore {
         Ok(out)
     }
 
+    /// Oldest `updated_at_ms` across all event-bus rows, or `None` if empty.
+    /// Used at startup to detect a cache older than the server's batch
+    /// replay window (~44 days) and force a full re-sync.
+    pub fn event_bus_state_min_updated_at_ms(&self) -> Result<Option<i64>, StoreError> {
+        let conn = self.conn.lock().unwrap();
+        let v: Option<i64> = conn.query_row(
+            "SELECT MIN(updated_at_ms) FROM event_bus_state",
+            [],
+            |row| row.get::<_, Option<i64>>(0),
+        )?;
+        Ok(v)
+    }
+
+    /// Wipe the per-group catch-up cursors. The next reconnect will not pass
+    /// `groupsToLastEventBatchIds`, and the syncer will see an empty state
+    /// and run the one-shot bootstrap.
+    pub fn clear_event_bus_state(&self) -> Result<(), StoreError> {
+        let conn = self.conn.lock().unwrap();
+        conn.execute("DELETE FROM event_bus_state", [])?;
+        Ok(())
+    }
+
     /// 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(
@@ -477,6 +499,23 @@ impl LocalStore {
         Ok(())
     }
 
+    /// Drop every cached mail row that belongs to `folder_id` and return
+    /// their element ids — the caller then deletes the matching .eml files.
+    /// Used by the event handler when a `MailSet` event tells us a folder
+    /// no longer exists on the server.
+    pub fn delete_folder_mails(&self, folder_id: &str) -> Result<Vec<String>, StoreError> {
+        let conn = self.conn.lock().unwrap();
+        let mut stmt = conn.prepare("SELECT element_id FROM mails WHERE folder_id = ?1")?;
+        let ids: Vec<String> = stmt
+            .query_map([folder_id], |row| row.get::<_, String>(0))?
+            .filter_map(|r| r.ok())
+            .collect();
+        if !ids.is_empty() {
+            conn.execute("DELETE FROM mails WHERE folder_id = ?1", [folder_id])?;
+        }
+        Ok(ids)
+    }
+
     /// 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> {
@@ -632,6 +671,27 @@ mod tests {
         assert!(store.verify_key());
     }
 
+    #[test]
+    fn event_bus_state_min_updated_at_and_clear() {
+        let store = open_memory_store();
+        assert!(store.event_bus_state_min_updated_at_ms().unwrap().is_none());
+
+        store.set_event_bus_batch_id("g1", "b1").unwrap();
+        // First write: min == that row's timestamp; just assert presence.
+        let min1 = store.event_bus_state_min_updated_at_ms().unwrap();
+        assert!(min1.is_some());
+        // Sleep a millisecond so the next write has a strictly larger ts.
+        std::thread::sleep(std::time::Duration::from_millis(2));
+        store.set_event_bus_batch_id("g2", "b2").unwrap();
+        let min2 = store.event_bus_state_min_updated_at_ms().unwrap();
+        // Still the original (oldest) row.
+        assert_eq!(min1, min2);
+
+        store.clear_event_bus_state().unwrap();
+        assert!(store.load_event_bus_state().unwrap().is_empty());
+        assert!(store.event_bus_state_min_updated_at_ms().unwrap().is_none());
+    }
+
     #[test]
     fn event_bus_state_roundtrip() {
         let store = open_memory_store();
@@ -650,6 +710,22 @@ mod tests {
         assert_eq!(s.get("group1"), Some(&"batchB".to_string()));
     }
 
+    #[test]
+    fn delete_folder_mails_returns_ids_and_drops_rows() {
+        let store = open_memory_store();
+        store.upsert_mail_metadata(&meta("a", "doomed", 1)).unwrap();
+        store.upsert_mail_metadata(&meta("b", "doomed", 2)).unwrap();
+        store.upsert_mail_metadata(&meta("c", "kept", 3)).unwrap();
+        let ids = store.delete_folder_mails("doomed").unwrap();
+        let mut ids = ids;
+        ids.sort();
+        assert_eq!(ids, vec!["a".to_string(), "b".to_string()]);
+        assert_eq!(store.mail_count("doomed").unwrap(), 0);
+        assert_eq!(store.mail_count("kept").unwrap(), 1);
+        // No rows for an unknown folder, returns empty.
+        assert!(store.delete_folder_mails("missing").unwrap().is_empty());
+    }
+
     #[test]
     fn delete_mail_removes_metadata_and_eml() {
         let store = open_memory_store();

crates/bridge/src/sync.rs

@@ -145,6 +145,30 @@ impl MailStore {
         }
     }
 
+    /// 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.
+    pub async fn prune_unknown_folders(
+        &self,
+        known: &std::collections::HashSet<String>,
+    ) -> Vec<String> {
+        let mut removed = Vec::new();
+        let mut folders = self.folders.write().await;
+        folders.retain(|fid, _| {
+            if known.contains(fid) {
+                true
+            } else {
+                removed.push(fid.clone());
+                false
+            }
+        });
+        drop(folders);
+        if !removed.is_empty() {
+            self.bump_generation();
+        }
+        removed
+    }
+
     /// 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;
@@ -705,4 +729,21 @@ mod tests {
         store.remove_mail_everywhere("unknown").await;
         assert_eq!(store.get_folder("folderA").await.len(), 1);
     }
+
+    #[tokio::test]
+    async fn prune_unknown_folders_drops_disappeared_ones() {
+        let store = MailStore::new();
+        store
+            .set_folder("keep", vec![stored(make_mail("L1", "M1", "k", true), 1)])
+            .await;
+        store
+            .set_folder("gone", vec![stored(make_mail("L1", "M2", "g", true), 2)])
+            .await;
+        let known: std::collections::HashSet<String> =
+            ["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);
+        assert!(store.get_folder("gone").await.is_empty());
+    }
 }