External/tutabridge
1
0
Fork 0
mirror of https://github.com/spartanz51/tutabridge.git synced 2026-06-24 10:54:32 +02:00
Code Issues Packages Projects Releases Wiki Activity

Event-driven realtime delta: skip the full folder re-sync on a MOVE

Browse Source 
The bridge no longer asks the server for a full folder listing on every
MailSetEntry CREATE/DELETE event. Instead it uses the encoding of the
entry id (4-byte timestamp + 9-byte Mail element id, see
`tuta-sdk::mail_set_entry_id`) to recover the affected mail directly,
and applies the delta to MailStore + LocalStore:

- MailSetEntry CREATE first (so a MOVE clones from the source folder
  before the matching DELETE runs). Hit path = `find_mail_anywhere` →
  clone the already-decrypted StoredMail into the target folder with a
  fresh UID. Miss path = a single `load_mail` against the cached
  `Mail.list_id`. Any decode failure / unknown folder / `load_mail`
  error queues a fallback full `sync_folder` for that folder — no
  silent miss.
- MailSetEntry DELETE second. Removes from the source folder only; the
  `.eml` and DB row are dropped only if no folder still holds the mail
  (multi-folder placement preserved, in-batch MOVE is correct because
  the target was upserted by the CREATE loop).
- Mail UPDATE / DELETE unchanged.
- MailSet folder-list dirty handling unchanged.

Saves the `load_range(1000)` round-trip on the common MOVE case. The
fallback path keeps the previous behaviour available so the change is
strictly an optimisation, not a behavioural change.

New MailStore helpers (5 new tests):
- `find_mail_anywhere(eid)` / `is_mail_anywhere(eid)` — multi-folder
  lookup.
- `remove_mail_from_folder(folder_id, eid)` — scoped remove (vs the
  existing `remove_mail_everywhere`).
- `upsert_mail_in_folder` — idempotent insert/replace by element_id.
- `mail_list_id()` — sniff once, cache; needed by the load_mail miss
  path.

`mail_to_metadata` made `pub(crate)` for the handler.

Bucketing in the event handler also split into `mail_set_entry_creates`
vs `mail_set_entry_deletes` so the order is explicit; 8 handler tests
cover empty / mixed / order-preserved / immutable-UPDATE-ignored shapes.

166/166 bridge lib tests pass.
This commit is contained in:
Anthony
2026-05-28 14:32:36 +02:00
parent d0457b24f6
commit 34127f7587
2 changed files with 557 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/bridge/src/event_handler.rs
+342 -133
View File
@@ -2,20 +2,32 @@
//!
//! 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`.
//! `LocalStore` (encrypted SQLite + `.eml` files).
//!
//! The hot path — `MailSetEntry` CREATE/DELETE — is handled **without**
//! re-listing the affected folder over REST: a `MailSetEntry`'s element id
//! is a Tuta-defined encoding of `(receivedDate, mail_element_id)`, so we
//! recover the mail id directly via `tutasdk::mail_set_entry_id::deconstruct`
//! and either move the already-decrypted Mail between folders in memory
//! (when it's a MOVE between two cached folders) or ask the backend for
//! the single Mail (`load_mail`) when it is a brand-new arrival. The full
//! `sync_folder` only runs as a safety-net fallback (decode failure,
//! unknown folder, `load_mail` error).
//!
//! 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::collections::{HashMap, HashSet};
use std::sync::{Arc, Mutex};
use log::{debug, info, warn};
use tokio::sync::{mpsc, watch};
use tutasdk::event_bus::{EntityUpdateBatch, EntityUpdateEvent, EventBusMessage, Operation};
use tutasdk::{mail_set_entry_id, CustomId};
use crate::store::LocalStore;
use crate::sync::{sync_folder, MailStore};
use crate::tuta::MailBackend;
use crate::sync::{sync_folder, MailStore, StoredMail};
use crate::tuta::{FolderInfo, MailBackend};
/// Application tag for Tuta's mail-side entities.
const TUTANOTA_APP: &str = "tutanota";
@@ -24,8 +36,8 @@ 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).
/// `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(
@@ -84,16 +96,15 @@ 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, whether the folder list itself changed) is pure and
/// has no I/O — that lets us test it in isolation.
/// 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).
#[cfg_attr(test, derive(Debug))]
#[derive(Default)]
struct Bucketed<'a> {
/// `MailSetEntry.instance_list_id` for every CREATE / DELETE / UPDATE
/// we received — i.e. the folders whose contents changed.
folder_entry_lists: std::collections::HashSet<&'a str>,
mail_set_entry_creates: Vec<&'a EntityUpdateEvent>,
mail_set_entry_deletes: Vec<&'a EntityUpdateEvent>,
/// Mail-entity updates (read/unread, subject, delete, …).
mail_events: Vec<&'a EntityUpdateEvent>,
/// A `MailSet` entity event arrived — the folder list itself changed
@@ -109,8 +120,12 @@ fn bucket_updates(updates: &[EntityUpdateEvent]) -> Bucketed<'_> {
continue;
}
match ev.type_id {
MAIL_SET_ENTRY_TYPE_ID => {
out.folder_entry_lists.insert(ev.instance_list_id.as_str());
MAIL_SET_ENTRY_TYPE_ID => match ev.operation {
Operation::Create => out.mail_set_entry_creates.push(ev),
Operation::Delete => out.mail_set_entry_deletes.push(ev),
// The Tuta model treats `MailSetEntry` as immutable — only
// CREATE / DELETE happen. Ignore other operations defensively.
_ => {},
},
MAIL_TYPE_ID => out.mail_events.push(ev),
MAIL_SET_TYPE_ID => out.folder_list_dirty = true,
@@ -128,108 +143,285 @@ async fn apply_batch(
batch: &EntityUpdateBatch,
) {
let Bucketed {
folder_entry_lists,
mail_set_entry_creates,
mail_set_entry_deletes,
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.
// the webmail. Refresh the list first so a brand-new folder is known
// before we try to apply MailSetEntry events that reference it.
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}"),
}
refresh_folder_list(store, local_store, backend).await;
}
// 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.as_str()))
{
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
);
}
}
// Snapshot the folder list once; the delta path matches events to
// folders by `entries_list_id`.
let folders = store.list_folders().await;
let folder_by_entries: HashMap<&str, &FolderInfo> = folders
.iter()
.map(|f| (f.entries_list_id.as_str(), f))
.collect();
// Folders we could not handle precisely — fall back to a full
// `sync_folder` at the end.
let mut fallback_folders: HashSet<String> = HashSet::new();
// 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
// REST round-trip.
for ev in &mail_set_entry_creates {
apply_mail_set_entry_create(
store,
local_store,
backend,
&folder_by_entries,
&mut fallback_folders,
ev,
)
.await;
}
// 2) MailSetEntry DELETEs. Per Tuta's wire model these arrive paired
// with the CREATEs (a MOVE = DELETE source + CREATE target in the same
// batch); a lone DELETE means a trash / hard-delete.
for ev in &mail_set_entry_deletes {
apply_mail_set_entry_delete(store, local_store, &folder_by_entries, ev).await;
}
// 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.
// CREATE on a Mail entity is paired with a MailSetEntry CREATE which
// the loop above already handled.
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(_) => {},
apply_mail_event(store, local_store, backend, ev).await;
}
// 4) Safety net: for every folder we couldn't precisely apply (decode
// failure, unknown folder, REST error during `load_mail`), re-run the
// classic full sync so the user never silently misses a mail.
for entries_list_id in &fallback_folders {
let Some(folder) = folder_by_entries.get(entries_list_id.as_str()).copied() else {
continue;
};
debug!(
"Event bus: fallback full sync for {} (batch {})",
folder.imap_path, batch.batch_id
);
if let Err(e) = sync_folder(store, local_store, backend, folder, sync_limit).await {
warn!(
"Event bus fallback sync failed for {}: {}",
folder.imap_path, e
);
}
}
}
async fn refresh_folder_list(
store: &MailStore,
local_store: &LocalStore,
backend: &dyn MailBackend,
) {
match backend.list_folders().await {
Ok(folders) => {
let known: 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}"),
}
}
async fn apply_mail_set_entry_create(
store: &MailStore,
local_store: &LocalStore,
backend: &dyn MailBackend,
folder_by_entries: &HashMap<&str, &FolderInfo>,
fallback_folders: &mut HashSet<String>,
ev: &EntityUpdateEvent,
) {
let custom = CustomId(ev.instance_id.clone());
let mail_eid = match mail_set_entry_id::deconstruct(&custom) {
Ok((_date, mail_id)) => mail_id.0,
Err(e) => {
warn!(
"MailSetEntry CREATE id {:?} could not be decoded: {e} — falling back to full sync",
ev.instance_id
);
fallback_folders.insert(ev.instance_list_id.clone());
return;
},
};
let Some(target_folder) = folder_by_entries.get(ev.instance_list_id.as_str()).copied() else {
// Folder unknown — typically a newly created custom folder whose
// `MailSet` event we have not yet processed. Falling back ensures
// we discover it via `list_folders` on the next batch.
fallback_folders.insert(ev.instance_list_id.clone());
return;
};
// HIT path: the mail already lives in another cached folder (typical
// MOVE between two known folders). Clone the StoredMail into the
// target, allocate a fresh UID and persist.
if let Some((_source_folder, mut stored)) = store.find_mail_anywhere(&mail_eid).await {
assign_uid_and_upsert(store, local_store, target_folder, &mail_eid, &mut stored).await;
return;
}
// MISS path: never seen this mail. Ask the backend for just that one
// mail. Needs the Mail's `list_id` (≠ the folder's entries_list_id) —
// sniff it from any cached Mail (single-MailGroup is the common case;
// caller falls back if the cache is still empty).
let Some(list_id) = store.mail_list_id().await else {
fallback_folders.insert(ev.instance_list_id.clone());
return;
};
match backend.load_mail(&list_id, &mail_eid).await {
Ok(Some(mail)) => {
let mut stored = StoredMail {
mail,
details: None,
rfc2822: None,
uid: 0,
};
assign_uid_and_upsert(store, local_store, target_folder, &mail_eid, &mut stored).await;
},
Ok(None) => {
debug!("MailSetEntry CREATE: mail {} not found on server", mail_eid);
},
Err(e) => {
warn!(
"MailSetEntry CREATE: load_mail({}, {}) failed: {e} — falling back",
list_id, mail_eid
);
fallback_folders.insert(ev.instance_list_id.clone());
},
}
}
async fn apply_mail_set_entry_delete(
store: &MailStore,
local_store: &LocalStore,
folder_by_entries: &HashMap<&str, &FolderInfo>,
ev: &EntityUpdateEvent,
) {
let custom = CustomId(ev.instance_id.clone());
let mail_eid = match mail_set_entry_id::deconstruct(&custom) {
Ok((_date, mail_id)) => mail_id.0,
Err(e) => {
// We cannot identify *which* mail left the folder without the
// decoded id; an upstream MailSetEntry CREATE may have queued a
// fallback already, otherwise the next periodic interaction
// (re-select, FETCH) will reconcile.
warn!(
"MailSetEntry DELETE id {:?} could not be decoded: {e}",
ev.instance_id
);
return;
},
};
let Some(source_folder) = folder_by_entries.get(ev.instance_list_id.as_str()).copied() else {
return;
};
store
.remove_mail_from_folder(&source_folder.id, &mail_eid)
.await;
// Drop the on-disk row + `.eml` only if no folder still holds the
// mail. Multi-folder placement (rare with the current Tuta model) and
// MOVE-within-batch (the matching CREATE ran first, so the target
// folder still has it) are both preserved by this check.
if !store.is_mail_anywhere(&mail_eid).await {
if let Err(e) = local_store.delete_mail(&mail_eid) {
warn!("Failed to delete cached mail {}: {}", mail_eid, e);
}
}
}
async fn apply_mail_event(
store: &MailStore,
local_store: &LocalStore,
backend: &dyn MailBackend,
ev: &EntityUpdateEvent,
) {
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(_) => {},
}
}
/// Allocate a fresh UID in `target_folder`, stamp it on `stored`, then
/// upsert both `MailStore` and the `LocalStore` metadata row in one step.
async fn assign_uid_and_upsert(
store: &MailStore,
local_store: &LocalStore,
target_folder: &FolderInfo,
mail_eid: &str,
stored: &mut StoredMail,
) {
let uid = match local_store.allocate_folder_uids(&target_folder.id, &[mail_eid]) {
Ok(map) => map.get(mail_eid).copied().unwrap_or(0),
Err(e) => {
warn!(
"Failed to allocate UID for {} in {}: {e}",
mail_eid, target_folder.imap_path
);
0
},
};
stored.uid = uid;
let meta = crate::sync::mail_to_metadata(&stored.mail, &target_folder.id, uid);
if let Err(e) = local_store.upsert_mail_metadata(&meta) {
warn!("Failed to persist {} in {}: {e}", mail_eid, target_folder.id);
}
store
.upsert_mail_in_folder(&target_folder.id, stored.clone())
.await;
}
#[cfg(test)]
mod tests {
use super::*;
@@ -249,41 +441,62 @@ mod tests {
#[test]
fn bucket_empty_batch() {
let out = bucket_updates(&[]);
assert!(out.folder_entry_lists.is_empty());
assert!(out.mail_set_entry_creates.is_empty());
assert!(out.mail_set_entry_deletes.is_empty());
assert!(out.mail_events.is_empty());
assert!(!out.folder_list_dirty);
}
#[test]
fn bucket_ignores_other_applications() {
// `sys`-app events (e.g. group/user changes) must not affect mail buckets.
let updates = vec![ev("sys", 97, "L", "E", Operation::Create)];
let out = bucket_updates(&updates);
assert!(out.folder_entry_lists.is_empty());
assert!(out.mail_set_entry_creates.is_empty());
assert!(out.mail_set_entry_deletes.is_empty());
assert!(out.mail_events.is_empty());
}
#[test]
fn bucket_ignores_unknown_type_ids_in_tutanota() {
// Unrelated tutanota entities (e.g. attachments, contacts) should pass through.
// Unrelated tutanota entities (attachments, contacts, …) pass through.
let updates = vec![ev("tutanota", 999, "L", "E", Operation::Update)];
let out = bucket_updates(&updates);
assert!(out.folder_entry_lists.is_empty());
assert!(out.mail_set_entry_creates.is_empty());
assert!(out.mail_set_entry_deletes.is_empty());
assert!(out.mail_events.is_empty());
}
#[test]
fn bucket_collects_mail_set_entry_lists() {
// Same list appearing twice (CREATE + DELETE) should de-duplicate.
fn bucket_splits_mail_set_entry_creates_and_deletes() {
let updates = vec![
ev("tutanota", MAIL_SET_ENTRY_TYPE_ID, "inbox_entries", "e1", Operation::Create),
ev("tutanota", MAIL_SET_ENTRY_TYPE_ID, "inbox_entries", "e2", Operation::Delete),
ev("tutanota", MAIL_SET_ENTRY_TYPE_ID, "source_entries", "e2", Operation::Delete),
ev("tutanota", MAIL_SET_ENTRY_TYPE_ID, "sent_entries", "e3", Operation::Create),
];
let out = bucket_updates(&updates);
assert_eq!(out.folder_entry_lists.len(), 2);
assert!(out.folder_entry_lists.contains("inbox_entries"));
assert!(out.folder_entry_lists.contains("sent_entries"));
assert!(out.mail_events.is_empty());
assert_eq!(out.mail_set_entry_creates.len(), 2);
assert_eq!(out.mail_set_entry_deletes.len(), 1);
// Order within each bucket is preserved (Tuta guarantees batch order).
assert_eq!(out.mail_set_entry_creates[0].instance_id, "e1");
assert_eq!(out.mail_set_entry_creates[1].instance_id, "e3");
assert_eq!(out.mail_set_entry_deletes[0].instance_id, "e2");
}
#[test]
fn bucket_ignores_mail_set_entry_update_operations() {
// MailSetEntry is immutable per Tuta's model; UPDATE shouldn't
// happen, but if one ever sneaks through we ignore it rather than
// crash.
let updates = vec![ev(
"tutanota",
MAIL_SET_ENTRY_TYPE_ID,
"inbox_entries",
"e1",
Operation::Update,
)];
let out = bucket_updates(&updates);
assert!(out.mail_set_entry_creates.is_empty());
assert!(out.mail_set_entry_deletes.is_empty());
}
#[test]
@@ -293,12 +506,24 @@ mod tests {
ev("tutanota", MAIL_TYPE_ID, "mailL", "m2", Operation::Delete),
];
let out = bucket_updates(&updates);
assert!(out.folder_entry_lists.is_empty());
assert_eq!(out.mail_events.len(), 2);
assert_eq!(out.mail_events[0].instance_id, "m1");
assert_eq!(out.mail_events[1].operation, Operation::Delete);
}
#[test]
fn bucket_marks_folder_list_dirty_on_mail_set_event() {
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);
assert!(out.mail_set_entry_creates.is_empty());
assert!(out.mail_set_entry_deletes.is_empty());
assert!(out.mail_events.is_empty());
}
#[test]
fn bucket_mixed_batch() {
let updates = vec![
@@ -307,24 +532,8 @@ mod tests {
ev("sys", 42, "X", "Y", Operation::Create), // ignored
];
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_set_entry_creates.len(), 1);
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/sync.rs
+215 -1
View File
@@ -32,6 +32,11 @@ pub struct MailStore {
folder_list: RwLock<Vec<FolderInfo>>,
generation: watch::Sender<u64>,
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 {
@@ -42,6 +47,7 @@ impl MailStore {
folder_list: RwLock::new(Vec::new()),
generation: tx,
gen_counter: std::sync::atomic::AtomicU64::new(0),
mail_list_id_cache: RwLock::new(None),
})
}
@@ -145,6 +151,114 @@ impl MailStore {
}
}
/// Find a mail by element id across all folders. Returns the source
/// folder id and a clone of the [`StoredMail`] entry — letting the
/// event handler reuse the already-decrypted Mail/details when the
/// same mail just hopped between two cached folders, no REST round-trip.
pub async fn find_mail_anywhere(&self, element_id: &str) -> Option<(String, StoredMail)> {
let folders = self.folders.read().await;
for (fid, mails) in folders.iter() {
for m in mails {
if m.mail
._id
.as_ref()
.map(|id| id.element_id.to_string())
.as_deref()
== Some(element_id)
{
return Some((fid.clone(), m.clone()));
}
}
}
None
}
/// `true` if any folder still references this mail. Used after a
/// per-folder removal to decide whether the cached `.eml` can go.
pub async fn is_mail_anywhere(&self, element_id: &str) -> bool {
let folders = self.folders.read().await;
folders.values().any(|mails| {
mails.iter().any(|m| {
m.mail
._id
.as_ref()
.map(|id| id.element_id.to_string())
.as_deref()
== Some(element_id)
})
})
}
/// Remove a single mail from one specific folder. Returns `true` if a
/// row was actually removed (the mail might already be gone from this
/// folder if we are reprocessing an event).
pub async fn remove_mail_from_folder(&self, folder_id: &str, element_id: &str) -> bool {
let mut folders = self.folders.write().await;
let Some(mails) = folders.get_mut(folder_id) else {
return false;
};
let before = mails.len();
mails.retain(|m| {
m.mail
._id
.as_ref()
.map(|id| id.element_id.to_string())
.as_deref()
!= Some(element_id)
});
let changed = mails.len() != before;
drop(folders);
if changed {
self.bump_generation();
}
changed
}
/// Insert or replace a mail in `folder_id`. Replace-by-element-id keeps
/// the operation idempotent — re-applying the same event leaves the
/// store unchanged.
pub async fn upsert_mail_in_folder(&self, folder_id: &str, mail: StoredMail) {
let Some(eid) = mail.mail._id.as_ref().map(|id| id.element_id.to_string()) else {
return;
};
let mut folders = self.folders.write().await;
let entries = folders.entry(folder_id.to_string()).or_default();
if let Some(slot) = entries.iter_mut().find(|m| {
m.mail
._id
.as_ref()
.map(|id| id.element_id.to_string())
.as_deref()
== Some(&eid)
}) {
*slot = mail;
} else {
entries.push(mail);
}
drop(folders);
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.
/// Returns the ids that were removed so the caller can clean up the
/// LocalStore + .eml files for them.
@@ -550,7 +664,7 @@ where
unreachable!()
}
fn mail_to_metadata(mail: &Mail, folder_id: &str, uid: u32) -> MailMetadata {
pub(crate) fn mail_to_metadata(mail: &Mail, folder_id: &str, uid: u32) -> MailMetadata {
let (list_id, element_id) = mail
._id
.as_ref()
@@ -730,6 +844,106 @@ mod tests {
assert_eq!(store.get_folder("folderA").await.len(), 1);
}
#[tokio::test]
async fn find_mail_anywhere_returns_first_match() {
let store = MailStore::new();
store
.set_folder("A", vec![stored(make_mail("L1", "shared", "x", true), 7)])
.await;
store
.set_folder("B", vec![stored(make_mail("L1", "shared", "x", true), 11)])
.await;
let found = store.find_mail_anywhere("shared").await.expect("must find");
assert!(found.0 == "A" || found.0 == "B");
assert_eq!(
found.1.mail._id.as_ref().unwrap().element_id.to_string(),
"shared"
);
assert!(store.find_mail_anywhere("missing").await.is_none());
}
#[tokio::test]
async fn is_mail_anywhere_reflects_presence() {
let store = MailStore::new();
store
.set_folder("A", vec![stored(make_mail("L1", "e1", "s", true), 1)])
.await;
assert!(store.is_mail_anywhere("e1").await);
assert!(!store.is_mail_anywhere("e2").await);
}
#[tokio::test]
async fn remove_mail_from_folder_only_touches_that_folder() {
let store = MailStore::new();
store
.set_folder(
"A",
vec![
stored(make_mail("L1", "k", "keep", true), 1),
stored(make_mail("L1", "g", "gone", true), 2),
],
)
.await;
store
.set_folder("B", vec![stored(make_mail("L1", "g", "gone", true), 5)])
.await;
let removed = store.remove_mail_from_folder("A", "g").await;
assert!(removed);
assert_eq!(store.get_folder("A").await.len(), 1);
// B still has the mail — `remove_mail_from_folder` is scoped.
assert_eq!(store.get_folder("B").await.len(), 1);
// Idempotent: re-removing the same mail from A is a no-op.
assert!(!store.remove_mail_from_folder("A", "g").await);
}
#[tokio::test]
async fn upsert_mail_in_folder_replaces_existing_by_element_id() {
let store = MailStore::new();
store
.set_folder("A", vec![stored(make_mail("L1", "e1", "old", true), 4)])
.await;
store
.upsert_mail_in_folder("A", stored(make_mail("L1", "e1", "new", false), 4))
.await;
let mails = store.get_folder("A").await;
// Same element_id → replaced in place, no duplicate.
assert_eq!(mails.len(), 1);
assert_eq!(mails[0].mail.subject, "new");
assert!(!mails[0].mail.unread);
}
#[tokio::test]
async fn upsert_mail_in_folder_appends_when_new() {
let store = MailStore::new();
store
.set_folder("A", vec![stored(make_mail("L1", "e1", "one", true), 1)])
.await;
store
.upsert_mail_in_folder("A", stored(make_mail("L1", "e2", "two", true), 2))
.await;
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]
async fn prune_unknown_folders_drops_disappeared_ones() {
let store = MailStore::new();