Phase 3c: Mail CREATE inline → zero REST on a fresh mail, body included

A Mail CREATE event carries the encrypted Mail in `event.instance` and
its MailDetailsBlob in `event.blob_instance`. The bridge now pre-decrypts
both at the start of each batch into a `pending: HashMap<eid, PendingMail>`
pool, then the matching `MailSetEntry CREATE` consumes the entry by
element id — no `load_mail` REST call, and when the blob was present the
RFC 2822 `.eml` is rendered + written + `has_details = 1` on the spot,
so the prefetch loop never has to fetch the body either.

Total: a brand-new mail arriving over the event bus now needs 0 REST
calls when the server bundles the inline payloads (the common case).
Old path required 2 (load_mail + load_mail_details_blob).

New MailBackend method `decrypt_inline_mail_details_blob` delegates to
the SDK's `CryptoEntityClient::decrypt_inline_and_parse::<MailDetailsBlob>`
and extracts the `details` aggregate so the caller stays in terms of
`MailDetails`. The bucketer now routes Mail CREATEs to their own
`mail_creates` bucket so the pre-decrypt step runs ahead of the
MailSetEntry CREATE consumers. Existing CREATE-on-Mail behaviour
(nothing happens directly, MailSetEntry CREATE drives placement) is
preserved.

Fallback chain unchanged: a missing payload, an unresolvable session
key, or a decrypt error leaves the pool entry absent and the
MailSetEntry handler falls back to its existing inline-MailSetEntry →
`load_mail` ladder. 166 bridge lib tests pass (1 new bucket test
covers the routing).

crates/bridge/src/event_handler.rs

@@ -97,15 +97,26 @@ async fn process(
 }
 
 /// Bucketed view of the mail-relevant entity updates inside a batch. Pure;
-/// no I/O. Splitting `MailSetEntry` events into creates and deletes lets
-/// us process them in the right order (CREATEs first — so a move can clone
-/// the mail from the source folder before the DELETE removes it).
+/// no I/O. Splitting `MailSetEntry` events into creates and deletes (and
+/// peeling Mail CREATEs out of the generic `mail_events` bucket) lets us
+/// process them in the right order:
+///
+/// 1. Mail CREATEs pre-decrypt to a pending pool with their `MailDetails`
+///    blob inline.
+/// 2. MailSetEntry CREATEs consume the pool — no REST call on a brand-new
+///    mail when its payload was inline.
+/// 3. MailSetEntry DELETEs run after the CREATEs so a MOVE can clone from
+///    the source folder before it is wiped.
+/// 4. Mail UPDATEs / DELETEs apply last.
 #[cfg_attr(test, derive(Debug))]
 #[derive(Default)]
 struct Bucketed<'a> {
 	mail_set_entry_creates: Vec<&'a EntityUpdateEvent>,
 	mail_set_entry_deletes: Vec<&'a EntityUpdateEvent>,
-	/// Mail-entity updates (read/unread, subject, delete, …).
+	/// Mail CREATEs — kept separate so we can pre-decrypt Mail + blob
+	/// inline before the matching MailSetEntry CREATE asks for the mail.
+	mail_creates: Vec<&'a EntityUpdateEvent>,
+	/// Mail UPDATE / DELETE events.
 	mail_events: Vec<&'a EntityUpdateEvent>,
 	/// A `MailSet` entity event arrived — the folder list itself changed
 	/// (custom folder created / renamed / deleted). Triggers a refresh of
@@ -127,7 +138,10 @@ fn bucket_updates(updates: &[EntityUpdateEvent]) -> Bucketed<'_> {
 				// CREATE / DELETE happen. Ignore other operations defensively.
 				_ => {},
 			},
-			MAIL_TYPE_ID => out.mail_events.push(ev),
+			MAIL_TYPE_ID => match ev.operation {
+				Operation::Create => out.mail_creates.push(ev),
+				_ => out.mail_events.push(ev),
+			},
 			MAIL_SET_TYPE_ID => out.folder_list_dirty = true,
 			_ => {},
 		}
@@ -135,6 +149,15 @@ fn bucket_updates(updates: &[EntityUpdateEvent]) -> Bucketed<'_> {
 	out
 }
 
+/// A Mail+details pair recovered by inline-decrypting a Mail CREATE event
+/// before the matching MailSetEntry CREATE runs. `details` is only set
+/// when the server also bundled `event.blob_instance` (it's `None` on
+/// drafts and on legacy events).
+struct PendingMail {
+	mail: tutasdk::entities::generated::tutanota::Mail,
+	details: Option<tutasdk::entities::generated::tutanota::MailDetails>,
+}
+
 async fn apply_batch(
 	store: &MailStore,
 	local_store: &LocalStore,
@@ -145,6 +168,7 @@ async fn apply_batch(
 	let Bucketed {
 		mail_set_entry_creates,
 		mail_set_entry_deletes,
+		mail_creates,
 		mail_events,
 		folder_list_dirty,
 	} = bucket_updates(&batch.updates);
@@ -168,6 +192,13 @@ async fn apply_batch(
 	// `sync_folder` at the end.
 	let mut fallback_folders: HashSet<String> = HashSet::new();
 
+	// Pre-decrypt every Mail CREATE in this batch so its `event.instance`
+	// (and `event.blob_instance`, when present) gives us the Mail and its
+	// `MailDetails` without any REST call. The matching MailSetEntry CREATE
+	// below consumes from this pool; what's left is dropped at scope end.
+	let mut pending: HashMap<String, PendingMail> =
+		predecrypt_mail_creates(backend, &mail_creates).await;
+
 	// 1) MailSetEntry CREATEs first. Doing creates *before* the matching
 	// deletes lets a MOVE clone the already-decrypted Mail straight from
 	// the source folder (still present in the cache at this point) — no
@@ -179,6 +210,7 @@ async fn apply_batch(
 			backend,
 			&folder_by_entries,
 			&mut fallback_folders,
+			&mut pending,
 			ev,
 		)
 		.await;
@@ -252,6 +284,7 @@ async fn apply_mail_set_entry_create(
 	backend: &dyn MailBackend,
 	folder_by_entries: &HashMap<&str, &FolderInfo>,
 	fallback_folders: &mut HashSet<String>,
+	pending: &mut HashMap<String, PendingMail>,
 	ev: &EntityUpdateEvent,
 ) {
 	let custom = CustomId(ev.instance_id.clone());
@@ -286,6 +319,40 @@ async fn apply_mail_set_entry_create(
 		return;
 	}
 
+	// HIT-FROM-BATCH path: a matching Mail CREATE rode in the same batch
+	// and we already inline-decrypted it (and possibly its MailDetails blob)
+	// into `pending`. Use that directly — no REST, and if we got the blob
+	// too the body is already RFC2822-rendered and written to disk so the
+	// prefetch loop has nothing left to do for this mail.
+	if let Some(PendingMail { mail, details }) = pending.remove(&mail_eid) {
+		debug!(
+			"Event bus: applying inline Mail CREATE {} → {} (no REST{})",
+			mail_eid,
+			target_folder.imap_path,
+			if details.is_some() { ", body inline too" } else { "" },
+		);
+		let rfc2822 = details
+			.as_ref()
+			.map(|d| crate::mail::mail_to_rfc2822(&mail, Some(d)));
+		let mut stored = StoredMail {
+			mail,
+			details: details.clone(),
+			rfc2822: rfc2822.clone(),
+			uid: 0,
+		};
+		// Persist the body straight away so the IMAP layer can serve FETCH
+		// BODY[] without a placeholder, and the prefetch loop skips it.
+		if let (Some(eml), true) = (rfc2822.as_deref(), details.is_some()) {
+			if let Err(e) = local_store.write_eml(&mail_eid, eml) {
+				warn!("Failed to cache eml for {}: {e}", mail_eid);
+			} else if let Err(e) = local_store.mark_has_details(&mail_eid) {
+				warn!("Failed to mark has_details for {}: {e}", mail_eid);
+			}
+		}
+		assign_uid_and_upsert(store, local_store, target_folder, &mail_eid, &mut stored).await;
+		return;
+	}
+
 	// MISS path: never seen this mail. The MailSetEntry payload is
 	// inline in `event.instance`; decrypting it gives us the referenced
 	// Mail's full `(list_id, element_id)` directly, so we can ask the
@@ -436,6 +503,50 @@ async fn apply_mail_event(
 	}
 }
 
+/// Pre-decrypt every Mail CREATE event in a batch into a `(eid -> Mail + details)`
+/// pool, ready to be consumed by the matching MailSetEntry CREATE. Each
+/// event has its Mail in `event.instance` and (when the server bundles it)
+/// the MailDetails blob in `event.blob_instance`. Failure on any single
+/// event leaves the pool entry absent — the MailSetEntry handler then
+/// falls back to `load_mail`, so nothing is silently lost.
+async fn predecrypt_mail_creates(
+	backend: &dyn MailBackend,
+	events: &[&EntityUpdateEvent],
+) -> HashMap<String, PendingMail> {
+	let mut out: HashMap<String, PendingMail> = HashMap::with_capacity(events.len());
+	for ev in events {
+		let Some(json) = ev.instance.as_deref() else {
+			continue;
+		};
+		let mail = match backend.decrypt_inline_mail(json).await {
+			Ok(Some(m)) => m,
+			Ok(None) => continue, // session key transient — caller will REST-fallback
+			Err(e) => {
+				warn!(
+					"predecrypt Mail CREATE {}: decrypt_inline_mail failed: {e}",
+					ev.instance_id
+				);
+				continue;
+			},
+		};
+		let details = match ev.blob_instance.as_deref() {
+			Some(blob_json) => match backend.decrypt_inline_mail_details_blob(blob_json).await {
+				Ok(opt) => opt,
+				Err(e) => {
+					warn!(
+						"predecrypt Mail CREATE {}: blob decrypt failed: {e} — keeping mail without body",
+						ev.instance_id
+					);
+					None
+				},
+			},
+			None => None,
+		};
+		out.insert(ev.instance_id.clone(), PendingMail { mail, details });
+	}
+	out
+}
+
 /// 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.
@@ -594,10 +705,26 @@ mod tests {
 		];
 		let out = bucket_updates(&updates);
 		assert_eq!(out.mail_events.len(), 2);
+		assert!(out.mail_creates.is_empty(), "no CREATE here");
 		assert_eq!(out.mail_events[0].instance_id, "m1");
 		assert_eq!(out.mail_events[1].operation, Operation::Delete);
 	}
 
+	#[test]
+	fn bucket_routes_mail_create_to_its_own_bucket() {
+		// Mail CREATEs are split out of `mail_events` so the inline pool
+		// can pre-decrypt them before MailSetEntry CREATEs consume them.
+		let updates = vec![
+			ev("tutanota", MAIL_TYPE_ID, "mailL", "newM", Operation::Create),
+			ev("tutanota", MAIL_TYPE_ID, "mailL", "readM", Operation::Update),
+		];
+		let out = bucket_updates(&updates);
+		assert_eq!(out.mail_creates.len(), 1);
+		assert_eq!(out.mail_creates[0].instance_id, "newM");
+		assert_eq!(out.mail_events.len(), 1);
+		assert_eq!(out.mail_events[0].instance_id, "readM");
+	}
+
 	#[test]
 	fn bucket_marks_folder_list_dirty_on_mail_set_event() {
 		let updates = vec![

crates/bridge/src/imap/session.rs

@@ -1471,6 +1471,12 @@ mod tests {
         ) -> Result<Option<tutasdk::entities::generated::tutanota::MailSetEntry>, String> {
             Ok(None)
         }
+        async fn decrypt_inline_mail_details_blob(
+            &self,
+            _json: &str,
+        ) -> Result<Option<MailDetails>, String> {
+            Ok(None)
+        }
         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/tuta.rs

@@ -8,7 +8,7 @@ use tutasdk::bindings::file_client::{FileClient, FileClientError};
 use tutasdk::bindings::rest_client::RestClient;
 use tutasdk::crypto_entity_client::CryptoEntityClient;
 use tutasdk::entities::generated::tutanota::{
-    DraftCreateData, DraftData, DraftRecipient, Mail, MailBox, MailDetails,
+    DraftCreateData, DraftData, DraftRecipient, Mail, MailBox, MailDetails, MailDetailsBlob,
     MailSetEntry, SendDraftData, SendDraftParameters,
 };
 use tutasdk::folder_system::{FolderSystem, MailSetKind};
@@ -59,6 +59,13 @@ pub trait MailBackend: Send + Sync {
         &self,
         json: &str,
     ) -> Result<Option<MailSetEntry>, String>;
+    /// Inline-decrypt the `event.blob_instance` carried alongside a Mail
+    /// CREATE — that's the `MailDetailsBlob` (subject + body + recipients
+    /// envelope) the prefetch loop would otherwise have to fetch via REST.
+    async fn decrypt_inline_mail_details_blob(
+        &self,
+        json: &str,
+    ) -> Result<Option<MailDetails>, 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>;
@@ -361,6 +368,20 @@ impl MailBackend for TutaSession {
             .map_err(|e| format!("{e}"))
     }
 
+    async fn decrypt_inline_mail_details_blob(
+        &self,
+        json: &str,
+    ) -> Result<Option<MailDetails>, String> {
+        // `event.blob_instance` is the encrypted MailDetailsBlob. We decrypt
+        // it through the same inline pipeline as the Mail itself and pull out
+        // its `details` aggregate, which is what consumers actually want.
+        self.crypto_client()
+            .decrypt_inline_and_parse::<MailDetailsBlob>(json)
+            .await
+            .map(|opt| opt.map(|blob| blob.details))
+            .map_err(|e| format!("{e}"))
+    }
+
     async fn load_mail_details(&self, mail: &Mail) -> Result<Option<MailDetails>, String> {
         self.load_mail_details_impl(mail)
             .await