From cfb030272d48f1e32077eb07689b2b74af60d8f5 Mon Sep 17 00:00:00 2001 From: Anthony Date: Thu, 28 May 2026 12:56:34 +0200 Subject: [PATCH] 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 | 84 +++++++++++++++++++++++++++--- crates/bridge/src/event_handler.rs | 58 +++++++++++++++++++-- crates/bridge/src/store.rs | 76 +++++++++++++++++++++++++++ crates/bridge/src/sync.rs | 41 +++++++++++++++ 4 files changed, 249 insertions(+), 10 deletions(-) diff --git a/crates/bridge/src/bridge.rs b/crates/bridge/src/bridge.rs index 9d5c1b1..7a3c14a 100644 --- a/crates/bridge/src/bridge.rs +++ b/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 = + std::sync::LazyLock::new(|| parse_model_version(SYS_TYPE_MODELS_JSON)); + *V +} + +fn tutanota_model_version() -> u32 { + static V: std::sync::LazyLock = + 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); + } +} diff --git a/crates/bridge/src/event_handler.rs b/crates/bridge/src/event_handler.rs index 9a6370e..14f23ee 100644 --- a/crates/bridge/src/event_handler.rs +++ b/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, @@ -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 = + 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()); } } diff --git a/crates/bridge/src/store.rs b/crates/bridge/src/store.rs index cb7d765..cebef35 100644 --- a/crates/bridge/src/store.rs +++ b/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, StoreError> { + let conn = self.conn.lock().unwrap(); + let v: Option = conn.query_row( + "SELECT MIN(updated_at_ms) FROM event_bus_state", + [], + |row| row.get::<_, Option>(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, StoreError> { + let conn = self.conn.lock().unwrap(); + let mut stmt = conn.prepare("SELECT element_id FROM mails WHERE folder_id = ?1")?; + let ids: Vec = 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(); diff --git a/crates/bridge/src/sync.rs b/crates/bridge/src/sync.rs index 9a93837..35653d7 100644 --- a/crates/bridge/src/sync.rs +++ b/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, + ) -> Vec { + 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 = + ["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()); + } }