mirror of
https://github.com/spartanz51/tutabridge.git
synced 2026-06-24 10:54:32 +02:00
Wire inline decrypt into the event handler — Mail UPDATEs go REST-free
Two new MailBackend methods, `decrypt_inline_mail` and `decrypt_inline_mail_set_entry`, delegate to the SDK's `CryptoEntityClient::decrypt_inline_and_parse<T>`. The event handler now takes the inline path first, and only falls back to `load_mail` if the payload was missing or its session key was unresolvable: - Mail UPDATE: a new `resolve_mail` helper centralises the decrypt-then-fallback policy. Most UPDATE events (read/unread, label moves, …) now do zero REST calls, since the encrypted Mail rides inside `event.instance`. - MailSetEntry CREATE miss path: decode `event.instance` of the MailSetEntry inline, read its `mail: IdTupleGenerated` field, then `load_mail` it with the correct list_id. Drops the previous `mail_list_id_cache` sniffing hack on MailStore (and its test) — the inline payload always carries the correct list_id, no need to guess from any cached Mail. `resolve_mail_set_entry` factors the same try-inline-first pattern for MailSetEntry events. Both helpers fall back to a `sync_folder` if every path fails, preserving the no-silent-miss guarantee from Phase 2. `MockBackend` returns `Ok(None)` from both new methods by default so existing handler tests keep their REST behaviour; the live SDK fixture test (`tests/decrypt_inline_test.rs` on the SDK side) covers the real decryption. 165 bridge lib tests pass.
This commit is contained in:
@@ -286,19 +286,25 @@ async fn apply_mail_set_entry_create(
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
// MISS path: never seen this mail. Ask the backend for just that one
|
// MISS path: never seen this mail. The MailSetEntry payload is
|
||||||
// mail. Needs the Mail's `list_id` (≠ the folder's entries_list_id) —
|
// inline in `event.instance`; decrypting it gives us the referenced
|
||||||
// sniff it from any cached Mail (single-MailGroup is the common case;
|
// Mail's full `(list_id, element_id)` directly, so we can ask the
|
||||||
// caller falls back if the cache is still empty).
|
// backend for just that one mail. No global mail-list-id cache, no
|
||||||
let Some(list_id) = store.mail_list_id().await else {
|
// `sync_folder` fallback needed in the typical case.
|
||||||
|
let mail_id_tuple = resolve_mail_set_entry(backend, ev).await;
|
||||||
|
let Some(mail_tuple) = mail_id_tuple else {
|
||||||
|
// Inline decode failed AND we have no cached mail to clone — the
|
||||||
|
// folder will be brought up-to-date by the safety-net sync below.
|
||||||
fallback_folders.insert(ev.instance_list_id.clone());
|
fallback_folders.insert(ev.instance_list_id.clone());
|
||||||
return;
|
return;
|
||||||
};
|
};
|
||||||
match backend.load_mail(&list_id, &mail_eid).await {
|
let list_id = mail_tuple.list_id.to_string();
|
||||||
|
let elem_id = mail_tuple.element_id.to_string();
|
||||||
|
match backend.load_mail(&list_id, &elem_id).await {
|
||||||
Ok(Some(mail)) => {
|
Ok(Some(mail)) => {
|
||||||
debug!(
|
debug!(
|
||||||
"Event bus: targeted load_mail({}, {}) → {} (1 REST call)",
|
"Event bus: targeted load_mail({}, {}) → {} (1 REST call)",
|
||||||
list_id, mail_eid, target_folder.imap_path
|
list_id, elem_id, target_folder.imap_path
|
||||||
);
|
);
|
||||||
let mut stored = StoredMail {
|
let mut stored = StoredMail {
|
||||||
mail,
|
mail,
|
||||||
@@ -306,21 +312,40 @@ async fn apply_mail_set_entry_create(
|
|||||||
rfc2822: None,
|
rfc2822: None,
|
||||||
uid: 0,
|
uid: 0,
|
||||||
};
|
};
|
||||||
assign_uid_and_upsert(store, local_store, target_folder, &mail_eid, &mut stored).await;
|
assign_uid_and_upsert(store, local_store, target_folder, &elem_id, &mut stored).await;
|
||||||
},
|
},
|
||||||
Ok(None) => {
|
Ok(None) => {
|
||||||
debug!("MailSetEntry CREATE: mail {} not found on server", mail_eid);
|
debug!("MailSetEntry CREATE: mail {} not found on server", elem_id);
|
||||||
},
|
},
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
warn!(
|
warn!(
|
||||||
"MailSetEntry CREATE: load_mail({}, {}) failed: {e} — falling back",
|
"MailSetEntry CREATE: load_mail({}, {}) failed: {e} — falling back",
|
||||||
list_id, mail_eid
|
list_id, elem_id
|
||||||
);
|
);
|
||||||
fallback_folders.insert(ev.instance_list_id.clone());
|
fallback_folders.insert(ev.instance_list_id.clone());
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Try to discover the Mail referenced by a `MailSetEntry` CREATE event
|
||||||
|
/// without a REST round-trip. Preferred path: decrypt `event.instance`
|
||||||
|
/// inline (the entry carries `mail: IdTupleGenerated`). Falls back to the
|
||||||
|
/// in-memory cache by entry-id-derived mail id — if neither works, the
|
||||||
|
/// caller queues a fallback `sync_folder`.
|
||||||
|
async fn resolve_mail_set_entry(
|
||||||
|
backend: &dyn MailBackend,
|
||||||
|
ev: &EntityUpdateEvent,
|
||||||
|
) -> Option<tutasdk::IdTupleGenerated> {
|
||||||
|
if let Some(json) = ev.instance.as_deref() {
|
||||||
|
match backend.decrypt_inline_mail_set_entry(json).await {
|
||||||
|
Ok(Some(entry)) => return Some(entry.mail),
|
||||||
|
Ok(None) => debug!("MailSetEntry inline: session key unresolved"),
|
||||||
|
Err(e) => warn!("MailSetEntry inline decrypt failed: {e}"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
None
|
||||||
|
}
|
||||||
|
|
||||||
async fn apply_mail_set_entry_delete(
|
async fn apply_mail_set_entry_delete(
|
||||||
store: &MailStore,
|
store: &MailStore,
|
||||||
local_store: &LocalStore,
|
local_store: &LocalStore,
|
||||||
@@ -378,33 +403,81 @@ async fn apply_mail_event(
|
|||||||
}
|
}
|
||||||
store.remove_mail_everywhere(&ev.instance_id).await;
|
store.remove_mail_everywhere(&ev.instance_id).await;
|
||||||
},
|
},
|
||||||
Operation::Update => match backend.load_mail(&ev.instance_list_id, &ev.instance_id).await {
|
Operation::Update => {
|
||||||
Ok(Some(mail)) => {
|
// Prefer the inline-decrypt path: the encrypted Mail is already
|
||||||
store.refresh_mail_in_place(&mail).await;
|
// in `event.instance`, no REST round-trip needed. Fall back to
|
||||||
let mail_json = serde_json::to_string(&mail).unwrap_or_default();
|
// `load_mail` if the payload is absent or its session key is in a
|
||||||
if let Err(e) = local_store.refresh_mail_fields(
|
// transient unresolvable state (e.g. post-reply attachment keys).
|
||||||
&ev.instance_id,
|
let mail = resolve_mail(backend, ev).await;
|
||||||
&mail.subject,
|
match mail {
|
||||||
&mail.sender.name,
|
Some(mail) => {
|
||||||
&mail.sender.address,
|
store.refresh_mail_in_place(&mail).await;
|
||||||
mail.unread,
|
let mail_json = serde_json::to_string(&mail).unwrap_or_default();
|
||||||
&mail_json,
|
if let Err(e) = local_store.refresh_mail_fields(
|
||||||
) {
|
&ev.instance_id,
|
||||||
debug!("Could not refresh metadata for {}: {}", ev.instance_id, e);
|
&mail.subject,
|
||||||
}
|
&mail.sender.name,
|
||||||
},
|
&mail.sender.address,
|
||||||
Ok(None) => {
|
mail.unread,
|
||||||
// Disappeared between event and our follow-up load — treat
|
&mail_json,
|
||||||
// like a delete.
|
) {
|
||||||
let _ = local_store.delete_mail(&ev.instance_id);
|
debug!("Could not refresh metadata for {}: {}", ev.instance_id, e);
|
||||||
store.remove_mail_everywhere(&ev.instance_id).await;
|
}
|
||||||
},
|
},
|
||||||
Err(e) => warn!("Mail UPDATE: failed to load {}: {}", ev.instance_id, e),
|
None => {
|
||||||
|
// Disappeared / unresolvable — treat like a delete to
|
||||||
|
// keep the cache consistent.
|
||||||
|
let _ = local_store.delete_mail(&ev.instance_id);
|
||||||
|
store.remove_mail_everywhere(&ev.instance_id).await;
|
||||||
|
},
|
||||||
|
}
|
||||||
},
|
},
|
||||||
Operation::Create | Operation::Other(_) => {},
|
Operation::Create | Operation::Other(_) => {},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Decrypt-inline first, REST-fallback second. Centralises the policy so
|
||||||
|
/// every event-bus consumer takes the same fast path when the payload is
|
||||||
|
/// already inline, and the same safety-net otherwise.
|
||||||
|
async fn resolve_mail(
|
||||||
|
backend: &dyn MailBackend,
|
||||||
|
ev: &EntityUpdateEvent,
|
||||||
|
) -> Option<tutasdk::entities::generated::tutanota::Mail> {
|
||||||
|
if let Some(json) = ev.instance.as_deref() {
|
||||||
|
match backend.decrypt_inline_mail(json).await {
|
||||||
|
Ok(Some(mail)) => {
|
||||||
|
debug!(
|
||||||
|
"Event bus: decrypted Mail {} inline (no REST)",
|
||||||
|
ev.instance_id
|
||||||
|
);
|
||||||
|
return Some(mail);
|
||||||
|
},
|
||||||
|
Ok(None) => {
|
||||||
|
debug!(
|
||||||
|
"Event bus: Mail {} session key unresolvable, falling back to load_mail",
|
||||||
|
ev.instance_id
|
||||||
|
);
|
||||||
|
},
|
||||||
|
Err(e) => {
|
||||||
|
warn!(
|
||||||
|
"Event bus: decrypt_inline_mail({}) failed: {e} — falling back",
|
||||||
|
ev.instance_id
|
||||||
|
);
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
match backend
|
||||||
|
.load_mail(&ev.instance_list_id, &ev.instance_id)
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
Ok(opt) => opt,
|
||||||
|
Err(e) => {
|
||||||
|
warn!("Event bus: load_mail({}) failed: {e}", ev.instance_id);
|
||||||
|
None
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Allocate a fresh UID in `target_folder`, stamp it on `stored`, then
|
/// Allocate a fresh UID in `target_folder`, stamp it on `stored`, then
|
||||||
/// upsert both `MailStore` and the `LocalStore` metadata row in one step.
|
/// upsert both `MailStore` and the `LocalStore` metadata row in one step.
|
||||||
async fn assign_uid_and_upsert(
|
async fn assign_uid_and_upsert(
|
||||||
|
|||||||
@@ -1460,6 +1460,17 @@ mod tests {
|
|||||||
})
|
})
|
||||||
.cloned())
|
.cloned())
|
||||||
}
|
}
|
||||||
|
async fn decrypt_inline_mail(&self, _json: &str) -> Result<Option<Mail>, String> {
|
||||||
|
// Tests stub: the inline path is exercised by handler tests with
|
||||||
|
// an explicit recorded payload, not through the mock's body.
|
||||||
|
Ok(None)
|
||||||
|
}
|
||||||
|
async fn decrypt_inline_mail_set_entry(
|
||||||
|
&self,
|
||||||
|
_json: &str,
|
||||||
|
) -> Result<Option<tutasdk::entities::generated::tutanota::MailSetEntry>, String> {
|
||||||
|
Ok(None)
|
||||||
|
}
|
||||||
async fn load_mail_details(&self, mail: &Mail) -> Result<Option<MailDetails>, String> {
|
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();
|
let key = mail._id.as_ref().map(|id| id.element_id.to_string()).unwrap_or_default();
|
||||||
Ok(self.details.lock().unwrap().get(&key).cloned())
|
Ok(self.details.lock().unwrap().get(&key).cloned())
|
||||||
|
|||||||
@@ -32,11 +32,6 @@ pub struct MailStore {
|
|||||||
folder_list: RwLock<Vec<FolderInfo>>,
|
folder_list: RwLock<Vec<FolderInfo>>,
|
||||||
generation: watch::Sender<u64>,
|
generation: watch::Sender<u64>,
|
||||||
gen_counter: std::sync::atomic::AtomicU64,
|
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<Option<String>>,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
impl MailStore {
|
impl MailStore {
|
||||||
@@ -47,7 +42,6 @@ impl MailStore {
|
|||||||
folder_list: RwLock::new(Vec::new()),
|
folder_list: RwLock::new(Vec::new()),
|
||||||
generation: tx,
|
generation: tx,
|
||||||
gen_counter: std::sync::atomic::AtomicU64::new(0),
|
gen_counter: std::sync::atomic::AtomicU64::new(0),
|
||||||
mail_list_id_cache: RwLock::new(None),
|
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -239,26 +233,6 @@ impl MailStore {
|
|||||||
self.bump_generation();
|
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<String> {
|
|
||||||
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.
|
/// 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
|
/// Returns the ids that were removed so the caller can clean up the
|
||||||
/// LocalStore + .eml files for them.
|
/// LocalStore + .eml files for them.
|
||||||
@@ -925,24 +899,6 @@ mod tests {
|
|||||||
assert_eq!(store.get_folder("A").await.len(), 2);
|
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]
|
#[tokio::test]
|
||||||
async fn prune_unknown_folders_drops_disappeared_ones() {
|
async fn prune_unknown_folders_drops_disappeared_ones() {
|
||||||
|
|||||||
@@ -48,6 +48,17 @@ pub trait MailBackend: Send + Sync {
|
|||||||
/// handler to fetch a freshly-created or updated mail without re-listing
|
/// 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.
|
/// 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(&self, list_id: &str, element_id: &str) -> Result<Option<Mail>, String>;
|
||||||
|
/// Decrypt the still-encrypted `event.instance` payload of a Mail event
|
||||||
|
/// directly (no REST round-trip). `Ok(None)` covers both "session key
|
||||||
|
/// transient" and "payload absent" so the handler can fall back to
|
||||||
|
/// `load_mail`.
|
||||||
|
async fn decrypt_inline_mail(&self, json: &str) -> Result<Option<Mail>, String>;
|
||||||
|
/// Same shape for a `MailSetEntry` event — gives the handler the
|
||||||
|
/// referenced `mail` IdTuple without ever asking the server.
|
||||||
|
async fn decrypt_inline_mail_set_entry(
|
||||||
|
&self,
|
||||||
|
json: &str,
|
||||||
|
) -> Result<Option<MailSetEntry>, String>;
|
||||||
async fn load_mail_details(&self, mail: &Mail) -> Result<Option<MailDetails>, String>;
|
async fn load_mail_details(&self, mail: &Mail) -> Result<Option<MailDetails>, String>;
|
||||||
/// Enumerate all mail folders (system + custom, with hierarchy).
|
/// Enumerate all mail folders (system + custom, with hierarchy).
|
||||||
async fn list_folders(&self) -> Result<Vec<FolderInfo>, String>;
|
async fn list_folders(&self) -> Result<Vec<FolderInfo>, String>;
|
||||||
@@ -333,6 +344,23 @@ impl MailBackend for TutaSession {
|
|||||||
.map_err(|e| format!("{e}"))
|
.map_err(|e| format!("{e}"))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn decrypt_inline_mail(&self, json: &str) -> Result<Option<Mail>, String> {
|
||||||
|
self.crypto_client()
|
||||||
|
.decrypt_inline_and_parse::<Mail>(json)
|
||||||
|
.await
|
||||||
|
.map_err(|e| format!("{e}"))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn decrypt_inline_mail_set_entry(
|
||||||
|
&self,
|
||||||
|
json: &str,
|
||||||
|
) -> Result<Option<MailSetEntry>, String> {
|
||||||
|
self.crypto_client()
|
||||||
|
.decrypt_inline_and_parse::<MailSetEntry>(json)
|
||||||
|
.await
|
||||||
|
.map_err(|e| format!("{e}"))
|
||||||
|
}
|
||||||
|
|
||||||
async fn load_mail_details(&self, mail: &Mail) -> Result<Option<MailDetails>, String> {
|
async fn load_mail_details(&self, mail: &Mail) -> Result<Option<MailDetails>, String> {
|
||||||
self.load_mail_details_impl(mail)
|
self.load_mail_details_impl(mail)
|
||||||
.await
|
.await
|
||||||
|
|||||||
Reference in New Issue
Block a user