diff --git a/crates/nuncio-mail/src/jmap.rs b/crates/nuncio-mail/src/jmap.rs index 83d5227..9f307de 100644 --- a/crates/nuncio-mail/src/jmap.rs +++ b/crates/nuncio-mail/src/jmap.rs @@ -1,7 +1,7 @@ //! JMAP (RFC 8620 / RFC 8621) protocol engine, session discovery, and differential update parser. use async_trait::async_trait; -use nuncio_core::model::{Email, Folder, Placement, RemoteIdentity}; +use nuncio_core::model::{Email, Folder, Placement, PlacementKey, RemoteIdentity}; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; @@ -19,6 +19,47 @@ use crate::parser::MailError; /// JMAP mutation, which addresses messages by object id. const JMAP_UID_VALIDITY_SENTINEL: &str = "jmap"; +/// Scheme tag prefixing every folder checkpoint this engine writes: +/// `"v1:jmapstate:{state}"`. +/// +/// A JMAP state string is an opaque server token (RFC 8620 section 1.6.2) with +/// no self-describing shape, so a stored checkpoint that carries no scheme +/// cannot be told apart from one written by a different protocol engine or a +/// different meaning of the same field. Tagging makes an unrecognised token +/// resolve to a full re-enumeration instead of being handed to `Email/changes` +/// as a `sinceState` it never came from -- over-fetching is recoverable, +/// resuming from a token the server reads differently is not. +const JMAP_STATE_SCHEME: &str = "v1:jmapstate"; + +/// Separator between the scheme tag and the opaque state it wraps. The state +/// itself may contain further separators, so only the first one is structural. +const JMAP_STATE_DELIM: char = ':'; + +/// Ceiling on `Email/changes` pages followed in one pass. +/// +/// `hasMoreChanges` is a server-driven loop, and a server that answers `true` +/// forever -- or returns a `newState` that never advances -- would otherwise +/// pin the sync task in place. The bound is far above any real paging depth, so +/// hitting it means the server is misbehaving and the pass says so rather than +/// spinning. +const MAX_CHANGES_PAGES: usize = 256; + +/// Wrap an opaque JMAP state token in this build's checkpoint scheme. +fn encode_jmap_checkpoint(state: &str) -> String { + format!("{JMAP_STATE_SCHEME}{JMAP_STATE_DELIM}{state}") +} + +/// Recover the opaque JMAP state from a stored checkpoint, or `None` when the +/// token carries no scheme this build recognises. Only the scheme's own +/// delimiter is consumed; the remainder is the server's token verbatim. +fn parse_jmap_checkpoint(raw: &str) -> Option<&str> { + let state = raw + .trim() + .strip_prefix(JMAP_STATE_SCHEME)? + .strip_prefix(JMAP_STATE_DELIM)?; + (!state.is_empty()).then_some(state) +} + /// JMAP Session Object (RFC 8620 Section 2). #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] @@ -54,18 +95,44 @@ pub struct JmapEmailGetResponse { pub list: Vec, } -/// JMAP `Email/changes` response payload wrapper. -#[derive(Debug, Clone, Serialize, Deserialize)] +/// One page of a JMAP `Email/changes` response (RFC 8621 section 4.2, over the +/// RFC 8620 section 5.2 core method). +#[derive(Debug, Clone, Default, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct JmapEmailChangesResponse { /// Old state token queried against. + #[serde(default)] pub old_state: String, - /// New state token after differential sync. + /// New state token after this page of changes. pub new_state: String, - /// List of created or updated email IDs. + /// Ids of objects created since `oldState`. + #[serde(default)] + pub created: Vec, + /// Ids of objects updated since `oldState`. + #[serde(default)] pub updated: Vec, - /// List of destroyed/deleted email IDs. + /// Ids of objects destroyed since `oldState`. + #[serde(default)] pub destroyed: Vec, + /// Whether the server truncated this page. When `true`, `newState` is an + /// intermediate checkpoint and the caller must ask again from it. + #[serde(default)] + pub has_more_changes: bool, +} + +/// What an `Email/changes` call established. +/// +/// The `cannotCalculateChanges` arm is a first-class outcome rather than an +/// error: RFC 8620 section 5.2 defines it as the server declining to compute a +/// delta from the given state (the state is too old, or the server never kept +/// enough history), and the prescribed response is to re-enumerate rather than +/// to fail the sync. +#[derive(Debug, Clone)] +pub enum EmailChangesPage { + /// The server answered with a page of changes. + Changes(JmapEmailChangesResponse), + /// The server cannot compute a delta from the supplied `sinceState`. + CannotCalculateChanges, } /// JMAP `Email/query` response payload wrapper (RFC 8620 Section 5.5), narrowed @@ -99,6 +166,16 @@ pub struct JmapEmail { /// than one `Message-ID` header. The first entry is the one used. #[serde(default)] pub message_id: Option>, + /// Mailboxes this message currently belongs to, as `{mailboxId: true}` + /// (RFC 8621 section 4.1.1). JMAP models mailbox membership as a set, so + /// this is the only thing that says which folders a message is placed in -- + /// and, for a message the server reports as changed, whether it is still + /// placed in the folder being synced at all. + /// + /// `None` means the property was not requested or not returned, which is + /// "unknown membership", never "belongs to nothing". + #[serde(default)] + pub mailbox_ids: Option>, } /// JMAP email address object. @@ -191,7 +268,10 @@ impl JmapEngine { { "accountId": account_id, "ids": ids, - "properties": ["id", "subject", "from", "to", "receivedAt", "isUnread", "bodySnippet", "messageId"] + // `mailboxIds` is requested because a changed message's + // folder membership is what decides whether this pass + // places it or reconciles it away. + "properties": ["id", "subject", "from", "to", "receivedAt", "isUnread", "bodySnippet", "messageId", "mailboxIds"] }, "c1" ] @@ -376,6 +456,24 @@ impl JmapEngine { pub fn parse_email_get_response( &self, raw_json: &str, + ) -> Result<(Vec, String), MailError> { + self.parse_email_get_response_in_folder(raw_json, "inbox") + } + + /// Parse a raw JMAP `Email/get` JSON response payload into the occupancies + /// it establishes for `folder_id`, plus the account's Email state string. + /// + /// A returned message whose `mailboxIds` is present and does NOT name + /// `folder_id` is dropped rather than placed: `Email/changes` is + /// account-wide (RFC 8621 section 4.2 offers no mailbox filter), so a + /// folder pass sees ids that belong to other mailboxes, and placing them + /// here would invent an occupancy the server never reported. Membership + /// that the server did not return at all is left alone -- unknown is not + /// absent. + fn parse_email_get_response_in_folder( + &self, + raw_json: &str, + folder_id: &str, ) -> Result<(Vec, String), MailError> { let val: Value = serde_json::from_str(raw_json) .map_err(|e| MailError::ParseFailed(format!("invalid JMAP JSON response: {e}")))?; @@ -386,6 +484,10 @@ impl JmapEngine { let emails = resp .list .into_iter() + .filter(|item| match &item.mailbox_ids { + Some(ids) => ids.get(folder_id).copied().unwrap_or(false), + None => true, + }) .map(|item| { let sender = item .from @@ -401,7 +503,7 @@ impl JmapEngine { .map(|a| a.email.clone()) .unwrap_or_else(|| "me@nuncio.mx".to_string()); - let folder_id = "inbox".to_string(); + let folder_id = folder_id.to_string(); let remote_id = item.id; let message_id = item .message_id @@ -461,18 +563,31 @@ impl JmapEngine { Ok((emails, resp.state)) } - /// Parse raw JMAP `Email/changes` JSON response payload. - pub fn parse_email_changes_response( - &self, - raw_json: &str, - ) -> Result<(Vec, Vec, String), MailError> { + /// Parse one raw JMAP `Email/changes` JSON response payload. + /// + /// A method-level error carries a `type` (RFC 8620 section 3.6.2), which a + /// successful `Email/changes` payload never does, so the discriminator + /// works whether the caller hands over the full `methodResponses` envelope + /// or a bare payload. `cannotCalculateChanges` is returned as an outcome so + /// the caller can fall back to a full enumeration; any other error type is + /// a genuine failure and surfaces as one. + pub fn parse_email_changes_response(raw_json: &str) -> Result { let val: Value = serde_json::from_str(raw_json) .map_err(|e| MailError::ParseFailed(format!("invalid JMAP JSON response: {e}")))?; let payload = Self::extract_method_response_payload(&val)?; + + if let Some(error_type) = payload.get("type").and_then(|v| v.as_str()) { + if error_type == "cannotCalculateChanges" { + return Ok(EmailChangesPage::CannotCalculateChanges); + } + return Err(MailError::OperationFailed(format!( + "JMAP Email/changes failed with error type '{error_type}'" + ))); + } + let resp: JmapEmailChangesResponse = serde_json::from_value(payload) .map_err(|e| MailError::ParseFailed(format!("invalid Email/changes payload: {e}")))?; - - Ok((resp.updated, resp.destroyed, resp.new_state)) + Ok(EmailChangesPage::Changes(resp)) } /// Parse raw JMAP `Email/query` JSON response payload into a matched id list. @@ -556,6 +671,202 @@ impl JmapEngine { .cloned() .unwrap_or_else(|| self.account_id.clone()) } + + /// Address one occupancy of `remote_id` in `folder_id`. JMAP has no + /// UIDVALIDITY, so the sentinel stands in for that half of the key. + fn placement_key(&self, folder_id: &str, remote_id: &str) -> PlacementKey { + PlacementKey { + account_id: self.account_id.clone(), + folder_id: folder_id.to_string(), + uid_validity: JMAP_UID_VALIDITY_SENTINEL.to_string(), + remote_id: remote_id.to_string(), + } + } + + /// Follow `Email/changes` from `since_state` until the server stops setting + /// `hasMoreChanges`, accumulating every page. + /// + /// Returns `None` when the server answered `cannotCalculateChanges`, which + /// is the caller's cue to re-enumerate instead of failing. Stopping at the + /// first page would silently drop every change past the server's page + /// limit, so the loop is the contract, not an optimisation. + async fn collect_email_changes( + &self, + api_url: &str, + jmap_account_id: &str, + since_state: &str, + ) -> Result, MailError> { + let mut accumulated = EmailChangeSet { + changed: Vec::new(), + destroyed: Vec::new(), + new_state: since_state.to_string(), + }; + let mut seen_changed: std::collections::HashSet = std::collections::HashSet::new(); + let mut seen_destroyed: std::collections::HashSet = + std::collections::HashSet::new(); + let mut cursor = since_state.to_string(); + + for _ in 0..MAX_CHANGES_PAGES { + let request = Self::build_email_changes_request(jmap_account_id, &cursor); + let raw = self.post_jmap(api_url, &request).await?; + let page = match Self::parse_email_changes_response(&raw)? { + EmailChangesPage::CannotCalculateChanges => return Ok(None), + EmailChangesPage::Changes(page) => page, + }; + + for id in page.created.into_iter().chain(page.updated) { + if seen_changed.insert(id.clone()) { + accumulated.changed.push(id); + } + } + for id in page.destroyed { + if seen_destroyed.insert(id.clone()) { + accumulated.destroyed.push(id); + } + } + accumulated.new_state = page.new_state.clone(); + + if !page.has_more_changes { + // A message created and then destroyed within the same window + // appears in both lists; it is gone, so absence wins. + accumulated + .changed + .retain(|id| !seen_destroyed.contains(id)); + return Ok(Some(accumulated)); + } + if page.new_state == cursor { + return Err(MailError::OperationFailed( + "JMAP Email/changes reported more changes without advancing newState" + .to_string(), + )); + } + cursor = page.new_state; + } + + Err(MailError::OperationFailed(format!( + "JMAP Email/changes did not converge within {MAX_CHANGES_PAGES} pages" + ))) + } + + /// Enumerate a folder in full: every message it currently holds, and a + /// `present` set that says so. + /// + /// This is the only pass entitled to report presence, because it is the + /// only one that genuinely looked at the whole mailbox. + async fn full_folder_enumeration( + &self, + api_url: &str, + jmap_account_id: &str, + folder_id: &str, + ) -> Result { + let query_request = Self::build_email_query_request(jmap_account_id, folder_id); + let query_raw = self.post_jmap(api_url, &query_request).await?; + let ids = Self::parse_email_query_response(&query_raw)?; + + let get_request = Self::build_email_get_request(jmap_account_id, Some(ids)); + let get_raw = self.post_jmap(api_url, &get_request).await?; + let (emails, state) = self.parse_email_get_response_in_folder(&get_raw, folder_id)?; + + let present: Vec = emails.iter().map(|m| m.placement.key()).collect(); + + tracing::info!( + account_id = %self.account_id, + folder_id, + fetched = emails.len(), + "jmap full folder enumeration complete" + ); + + Ok(FolderChanges { + upserts: emails, + // Everything absent is expressed through `present`; a full pass has + // no separate report of removals to make. + removals: Vec::new(), + present: Some(present), + next_state: encode_jmap_checkpoint(&state), + }) + } + + /// Enumerate one folder's changes since `since_state` via `Email/changes`. + /// + /// Returns `None` when the server cannot compute the delta, so the caller + /// can fall back to a full enumeration. + async fn incremental_folder_pass( + &self, + api_url: &str, + jmap_account_id: &str, + folder_id: &str, + since_state: &str, + ) -> Result, MailError> { + let Some(changes) = self + .collect_email_changes(api_url, jmap_account_id, since_state) + .await? + else { + return Ok(None); + }; + + let upserts = if changes.changed.is_empty() { + Vec::new() + } else { + let get_request = + Self::build_email_get_request(jmap_account_id, Some(changes.changed.clone())); + let get_raw = self.post_jmap(api_url, &get_request).await?; + let (emails, _state) = self.parse_email_get_response_in_folder(&get_raw, folder_id)?; + emails + }; + + // Destruction is account-wide, so a destroyed id takes its occupancy + // here with it. A changed id that came back placed somewhere other than + // this folder -- or did not come back at all, having been destroyed + // between the two calls -- has left this folder, and its occupancy goes + // too. The message itself survives as long as another placement does; + // that is the caller's decision, not this one's. + let placed: std::collections::HashSet<&str> = upserts + .iter() + .map(|m| m.placement.remote_id.as_str()) + .collect(); + let removals: Vec = changes + .destroyed + .iter() + .map(String::as_str) + .chain( + changes + .changed + .iter() + .map(String::as_str) + .filter(|id| !placed.contains(id)), + ) + .map(|id| self.placement_key(folder_id, id)) + .collect(); + + tracing::info!( + account_id = %self.account_id, + folder_id, + fetched = upserts.len(), + removed = removals.len(), + "jmap incremental folder sync complete" + ); + + Ok(Some(FolderChanges { + upserts, + removals, + // An `Email/changes` pass sees only what moved. It never enumerates + // the mailbox, so it cannot speak to absence beyond what the server + // explicitly reported, and claiming otherwise would delete every + // message that simply did not change. + present: None, + next_state: encode_jmap_checkpoint(&changes.new_state), + })) + } +} + +/// Every `Email/changes` page for one pass, collapsed. +struct EmailChangeSet { + /// Created and updated ids, deduplicated, in first-seen order. + changed: Vec, + /// Destroyed ids, deduplicated, in first-seen order. + destroyed: Vec, + /// The final `newState`, once the server stopped paging. + new_state: String, } #[async_trait] @@ -593,14 +904,14 @@ impl MailBackend for JmapEngine { } #[tracing::instrument( - skip(self, _since_state), + skip(self, since_state), fields(account_id = %self.account_id, folder_id = %folder_id), err )] async fn sync_changes( &self, folder_id: &str, - _since_state: Option<&str>, + since_state: Option<&str>, ) -> Result { if !self.has_credentials() { return Err(MailError::AuthError( @@ -610,31 +921,25 @@ impl MailBackend for JmapEngine { let session = self.discover_session().await?; let account_id = self.resolve_account_id(&session); - let query_request = Self::build_email_query_request(&account_id, folder_id); - let query_raw = self.post_jmap(&session.api_url, &query_request).await?; - let ids = Self::parse_email_query_response(&query_raw)?; + // A checkpoint this build wrote resumes differentially. Anything else + // -- absent, or carrying no scheme this build recognises -- enumerates + // the folder in full, which is always safe. + if let Some(state) = since_state.and_then(parse_jmap_checkpoint) { + if let Some(changes) = self + .incremental_folder_pass(&session.api_url, &account_id, folder_id, state) + .await? + { + return Ok(changes); + } + tracing::info!( + account_id = %self.account_id, + folder_id, + "jmap server cannot calculate changes from the stored state; re-enumerating" + ); + } - let get_request = Self::build_email_get_request(&account_id, Some(ids)); - let get_raw = self.post_jmap(&session.api_url, &get_request).await?; - let (emails, state) = self.parse_email_get_response(&get_raw)?; - - tracing::info!( - account_id = %self.account_id, - folder_id, - fetched = emails.len(), - "jmap message sync complete" - ); - - Ok(FolderChanges { - upserts: emails, - removals: Vec::new(), - // `Email/query` was filtered to one mailbox, but this engine does - // not yet use `Email/changes`, so it cannot claim to have seen the - // complete set. Reporting `None` keeps the caller from deleting - // anything on the strength of a partial view. - present: None, - next_state: state, - }) + self.full_folder_enumeration(&session.api_url, &account_id, folder_id) + .await } #[tracing::instrument(skip(self), fields(account_id = %self.account_id), err)] @@ -743,19 +1048,84 @@ mod tests { #[test] fn parse_email_changes_response_valid_payload() { - let engine = JmapEngine::new("acct-1"); let raw = r#"{ "oldState": "s-1", "newState": "s-2", + "created": ["msg-12"], "updated": ["msg-10", "msg-11"], - "destroyed": ["msg-5"] + "destroyed": ["msg-5"], + "hasMoreChanges": true }"#; - let (updated, destroyed, new_state) = engine - .parse_email_changes_response(raw) - .expect("parse changes"); - assert_eq!(updated.len(), 2); - assert_eq!(destroyed.len(), 1); - assert_eq!(new_state, "s-2"); + let EmailChangesPage::Changes(page) = + JmapEngine::parse_email_changes_response(raw).expect("parse changes") + else { + panic!("a well-formed page must not read as cannotCalculateChanges"); + }; + assert_eq!(page.created, vec!["msg-12".to_string()]); + assert_eq!(page.updated.len(), 2); + assert_eq!(page.destroyed.len(), 1); + assert_eq!(page.new_state, "s-2"); + assert!(page.has_more_changes); + } + + /// `cannotCalculateChanges` is an outcome, not a failure: RFC 8620 s5.2 + /// defines it as "re-enumerate", so it must not surface as an error that + /// aborts the sync. Any other method error still must. + #[test] + fn parse_email_changes_response_distinguishes_cannot_calculate_from_failure() { + let raw = r#"{"methodResponses":[["error",{"type":"cannotCalculateChanges"},"c1"]]}"#; + assert!(matches!( + JmapEngine::parse_email_changes_response(raw), + Ok(EmailChangesPage::CannotCalculateChanges) + )); + + let failed = r#"{"methodResponses":[["error",{"type":"serverFail"},"c1"]]}"#; + assert!(JmapEngine::parse_email_changes_response(failed).is_err()); + } + + #[test] + fn jmap_checkpoints_round_trip_through_their_scheme() { + // A JMAP state is opaque and may itself contain the delimiter, so only + // the scheme's own separator is structural. + let encoded = encode_jmap_checkpoint("state:with:colons"); + assert_eq!(encoded, "v1:jmapstate:state:with:colons"); + assert_eq!(parse_jmap_checkpoint(&encoded), Some("state:with:colons")); + } + + #[test] + fn an_untagged_or_unknown_scheme_checkpoint_is_not_interpreted() { + // A bare token written before tagging existed, or one from another + // scheme, must resolve to a full re-enumeration rather than being + // handed to the server as a sinceState it never issued. + assert_eq!(parse_jmap_checkpoint("sync-state-900"), None); + assert_eq!(parse_jmap_checkpoint("v1:uidnext:42:105"), None); + assert_eq!(parse_jmap_checkpoint("v1:jmapstate:"), None); + } + + #[test] + fn parse_email_get_response_drops_messages_placed_in_another_mailbox() { + // `Email/changes` is account-wide, so a folder pass sees ids that + // belong elsewhere. Membership the server did return is authoritative; + // membership it never returned is unknown, not absent. + let raw = r#"{ + "methodResponses": [ + ["Email/get", {"state": "s-1", "list": [ + {"id": "m-here", "mailboxIds": {"mb-inbox": true}}, + {"id": "m-elsewhere", "mailboxIds": {"mb-archive": true}}, + {"id": "m-unknown"} + ]}, "c1"] + ] + }"#; + let engine = JmapEngine::new("acct-1"); + let (emails, _state) = engine + .parse_email_get_response_in_folder(raw, "mb-inbox") + .expect("parse get"); + let ids: Vec<&str> = emails + .iter() + .map(|m| m.placement.remote_id.as_str()) + .collect(); + assert_eq!(ids, vec!["m-here", "m-unknown"]); + assert_eq!(emails[0].placement.folder_id, "mb-inbox"); } #[test] diff --git a/crates/nuncio-mail/src/lib.rs b/crates/nuncio-mail/src/lib.rs index 6155b1d..b2cb9ef 100644 --- a/crates/nuncio-mail/src/lib.rs +++ b/crates/nuncio-mail/src/lib.rs @@ -22,7 +22,7 @@ pub use backend::{ RemoteMutationKind, RemoteMutationSpec, }; pub use imap::{IdleSocketState, ImapDualSocketManager, ImapEngine}; -pub use jmap::JmapEngine; +pub use jmap::{EmailChangesPage, JmapEngine}; pub use mock::{MockMailBackend, MockMessageSender}; pub use parser::{MailError, MimeParserAdapter}; pub use smtp::SmtpTransportEngine; diff --git a/crates/nuncio-mail/tests/wiremock_jmap_test.rs b/crates/nuncio-mail/tests/wiremock_jmap_test.rs index 46b9415..d330f76 100644 --- a/crates/nuncio-mail/tests/wiremock_jmap_test.rs +++ b/crates/nuncio-mail/tests/wiremock_jmap_test.rs @@ -2,7 +2,7 @@ #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] -use nuncio_core::model::IdentitySource; +use nuncio_core::model::{IdentitySource, PlacementKey}; use nuncio_mail::{JmapEngine, MailBackend, RemoteMutationKind, RemoteMutationSpec}; use serde_json::json; use wiremock::matchers::{body_partial_json, method, path}; @@ -209,7 +209,10 @@ async fn jmap_engine_sync_folders_and_messages_through_mail_backend_trait() { .expect("sync_messages succeeds"); assert_eq!(emails.len(), 1); assert_eq!(emails[0].placement.remote_id, "msg-wm-200"); - assert_eq!(emails[0].placement.folder_id, "inbox"); + // The occupancy is labelled with the mailbox that was actually synced. + // Removals are addressed by the same key, so a placement stored under any + // other folder id could never be reconciled away again. + assert_eq!(emails[0].placement.folder_id, "mb-inbox"); assert_eq!(emails[0].source, IdentitySource::EmailId); assert_ne!(emails[0].email.id, "msg-wm-200"); assert_eq!(emails[0].email.id.len(), 64); @@ -222,7 +225,9 @@ async fn jmap_engine_sync_folders_and_messages_through_mail_backend_trait() { emails[0].email.body_plain, Some("Fetched through the MailBackend trait.".to_string()) ); - assert_eq!(new_state, "sync-state-900"); + // The checkpoint is scheme-tagged, so a later build can tell what the + // opaque server token inside it actually means. + assert_eq!(new_state, "v1:jmapstate:sync-state-900"); } /// A session-discovery failure (non-2xx status) must surface as a genuine `MailError`, never a @@ -392,3 +397,385 @@ async fn jmap_apply_mutation_surfaces_server_rejection_as_error() { .expect_err("a server-rejected mutation must not be reported as success"); assert!(err.to_string().contains("jmap-msg-1")); } + +// --- Email/changes differential sync (RFC 8621 s4.2 over RFC 8620 s5.2) --- + +/// Mount one `Email/changes` page, matched on the exact `sinceState` the engine +/// must send. A pass that resumes from the wrong token matches no mock and the +/// request fails, so the matcher is itself the round-trip assertion. +async fn mount_email_changes_page( + mock_server: &MockServer, + since_state: &str, + body: serde_json::Value, +) { + Mock::given(method("POST")) + .and(path("/jmap/api")) + .and(body_partial_json(json!({ + "methodCalls": [["Email/changes", { "sinceState": since_state }, "c1"]] + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "methodResponses": [["Email/changes", body, "c1"]] + }))) + .mount(mock_server) + .await; +} + +/// Mount an `Email/get` answering with `list` verbatim and the given state. +async fn mount_email_get(mock_server: &MockServer, state: &str, list: serde_json::Value) { + Mock::given(method("POST")) + .and(path("/jmap/api")) + .and(body_partial_json(json!({ + "methodCalls": [["Email/get", {}, "c1"]] + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "methodResponses": [["Email/get", {"state": state, "list": list}, "c1"]] + }))) + .mount(mock_server) + .await; +} + +fn message(id: &str, mailbox_id: &str) -> serde_json::Value { + json!({ + "id": id, + "subject": format!("Message {id}"), + "from": [{ "email": "sender@nuncio.mx" }], + "to": [{ "email": "james.maes@kof22.com" }], + "receivedAt": 1700003000i64, + "isUnread": true, + "mailboxIds": { mailbox_id: true } + }) +} + +fn placement_key(folder_id: &str, remote_id: &str) -> PlacementKey { + PlacementKey { + account_id: "acct-100".to_string(), + folder_id: folder_id.to_string(), + uid_validity: "jmap".to_string(), + remote_id: remote_id.to_string(), + } +} + +/// The whole point of the differential pass: ids the server reports as +/// `destroyed` come back as placement removals, addressed exactly as the +/// occupancies this engine stores are -- so the sync path deletes those +/// placements and nothing else. An incremental pass still reports +/// `present: None`, because it never enumerated the folder. +#[tokio::test] +async fn jmap_sync_changes_turns_destroyed_ids_into_placement_removals() { + let mock_server = MockServer::start().await; + mount_session(&mock_server).await; + mount_email_changes_page( + &mock_server, + "srv-state-1", + json!({ + "oldState": "srv-state-1", + "newState": "srv-state-2", + "created": ["m-new"], + "updated": [], + "destroyed": ["m-gone-1", "m-gone-2"], + "hasMoreChanges": false + }), + ) + .await; + mount_email_get( + &mock_server, + "srv-state-2", + json!([message("m-new", "mb-inbox")]), + ) + .await; + + let engine = jmap_engine(&mock_server); + let changes = engine + .sync_changes("mb-inbox", Some("v1:jmapstate:srv-state-1")) + .await + .expect("incremental sync succeeds"); + + assert_eq!(changes.upserts.len(), 1); + assert_eq!(changes.upserts[0].placement.remote_id, "m-new"); + assert_eq!(changes.upserts[0].placement.folder_id, "mb-inbox"); + assert_eq!(changes.upserts[0].source, IdentitySource::EmailId); + + // Exactly the destroyed ids, keyed the same way the stored placement is. + assert_eq!( + changes.removals, + vec![ + placement_key("mb-inbox", "m-gone-1"), + placement_key("mb-inbox", "m-gone-2"), + ] + ); + // A message the pass placed must never also be reconciled away. + assert!(!changes.removals.iter().any(|key| key.remote_id == "m-new")); + + // An incremental pass cannot speak to absence beyond what it was told. + assert!( + changes.present.is_none(), + "an incremental pass must not claim to have enumerated the folder" + ); + assert_eq!(changes.next_state, "v1:jmapstate:srv-state-2"); +} + +/// `hasMoreChanges: true` means the server truncated the page. Stopping there +/// silently drops every change past the first page, so the pass must follow the +/// intermediate `newState` until the server stops paging, and accumulate. +#[tokio::test] +async fn jmap_sync_changes_follows_has_more_changes_to_convergence() { + let mock_server = MockServer::start().await; + mount_session(&mock_server).await; + mount_email_changes_page( + &mock_server, + "page-1", + json!({ + "oldState": "page-1", + "newState": "page-2", + "created": ["m-page1"], + "destroyed": ["m-gone-page1"], + "hasMoreChanges": true + }), + ) + .await; + mount_email_changes_page( + &mock_server, + "page-2", + json!({ + "oldState": "page-2", + "newState": "page-3", + "updated": ["m-page2"], + "destroyed": ["m-gone-page2"], + "hasMoreChanges": false + }), + ) + .await; + mount_email_get( + &mock_server, + "page-3", + json!([ + message("m-page1", "mb-inbox"), + message("m-page2", "mb-inbox") + ]), + ) + .await; + + let engine = jmap_engine(&mock_server); + let changes = engine + .sync_changes("mb-inbox", Some("v1:jmapstate:page-1")) + .await + .expect("paged incremental sync succeeds"); + + let upserted: Vec<&str> = changes + .upserts + .iter() + .map(|m| m.placement.remote_id.as_str()) + .collect(); + assert_eq!( + upserted, + vec!["m-page1", "m-page2"], + "both pages' changed ids must be fetched, not just the first" + ); + assert_eq!( + changes.removals, + vec![ + placement_key("mb-inbox", "m-gone-page1"), + placement_key("mb-inbox", "m-gone-page2"), + ], + "removals must accumulate across pages" + ); + // The last page's state is the resume point; an intermediate one would + // replay the tail of the change log forever. + assert_eq!(changes.next_state, "v1:jmapstate:page-3"); +} + +/// `cannotCalculateChanges` (RFC 8620 s5.2) is the server declining to compute +/// a delta. The pass must re-enumerate the folder rather than fail -- and only +/// then may it report `present`, because only then has it genuinely seen +/// everything the folder holds. +#[tokio::test] +async fn jmap_sync_changes_falls_back_to_full_enumeration_and_only_then_reports_presence() { + let mock_server = MockServer::start().await; + mount_session(&mock_server).await; + + Mock::given(method("POST")) + .and(path("/jmap/api")) + .and(body_partial_json(json!({ + "methodCalls": [["Email/changes", {}, "c1"]] + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "methodResponses": [["error", {"type": "cannotCalculateChanges"}, "c1"]] + }))) + .mount(&mock_server) + .await; + + Mock::given(method("POST")) + .and(path("/jmap/api")) + .and(body_partial_json(json!({ + "methodCalls": [["Email/query", {"filter": {"inMailboxes": ["mb-inbox"]}}, "c1"]] + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "methodResponses": [["Email/query", {"ids": ["m-a", "m-b"]}, "c1"]] + }))) + .mount(&mock_server) + .await; + + mount_email_get( + &mock_server, + "srv-full-9", + json!([message("m-a", "mb-inbox"), message("m-b", "mb-inbox")]), + ) + .await; + + let engine = jmap_engine(&mock_server); + let changes = engine + .sync_changes("mb-inbox", Some("v1:jmapstate:far-too-old")) + .await + .expect("a stale state must re-enumerate, not fail the sync"); + + assert_eq!(changes.upserts.len(), 2); + assert_eq!( + changes.present, + Some(vec![ + placement_key("mb-inbox", "m-a"), + placement_key("mb-inbox", "m-b"), + ]), + "a genuine full enumeration is the one pass entitled to report presence" + ); + assert!( + changes.removals.is_empty(), + "a full pass expresses absence through `present`, not through removals" + ); + assert_eq!(changes.next_state, "v1:jmapstate:srv-full-9"); +} + +/// The `newState` a pass returns is the token the next pass must resume from. +/// The second pass's `Email/changes` mock only matches that exact state, so a +/// token that failed to round-trip through `next_state` would match nothing. +#[tokio::test] +async fn jmap_sync_changes_resumes_from_the_state_the_previous_pass_returned() { + let mock_server = MockServer::start().await; + mount_session(&mock_server).await; + mount_email_changes_page( + &mock_server, + "round-1", + json!({ + "oldState": "round-1", + "newState": "round-2", + "destroyed": ["m-gone"], + "hasMoreChanges": false + }), + ) + .await; + mount_email_changes_page( + &mock_server, + "round-2", + json!({ + "oldState": "round-2", + "newState": "round-3", + "destroyed": ["m-gone-later"], + "hasMoreChanges": false + }), + ) + .await; + + let engine = jmap_engine(&mock_server); + let first = engine + .sync_changes("mb-inbox", Some("v1:jmapstate:round-1")) + .await + .expect("first pass succeeds"); + assert_eq!(first.next_state, "v1:jmapstate:round-2"); + + let second = engine + .sync_changes("mb-inbox", Some(&first.next_state)) + .await + .expect("the returned checkpoint must be accepted verbatim on the next pass"); + assert_eq!( + second.removals, + vec![placement_key("mb-inbox", "m-gone-later")] + ); + assert_eq!(second.next_state, "v1:jmapstate:round-3"); +} + +/// A checkpoint carrying no scheme this build recognises -- a bare token +/// written before tagging existed -- must never be handed to the server as a +/// `sinceState`. It resolves to a full enumeration instead: over-fetching is +/// recoverable, resuming from a token the server reads differently is not. +#[tokio::test] +async fn jmap_sync_changes_re_enumerates_when_the_stored_checkpoint_is_untagged() { + let mock_server = MockServer::start().await; + mount_session(&mock_server).await; + + // No `Email/changes` mock is mounted on purpose: such a call would 404 and + // fail the sync, which is what proves none was attempted. + Mock::given(method("POST")) + .and(path("/jmap/api")) + .and(body_partial_json(json!({ + "methodCalls": [["Email/query", {}, "c1"]] + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "methodResponses": [["Email/query", {"ids": ["m-a"]}, "c1"]] + }))) + .mount(&mock_server) + .await; + mount_email_get( + &mock_server, + "srv-full-1", + json!([message("m-a", "mb-inbox")]), + ) + .await; + + let engine = jmap_engine(&mock_server); + let changes = engine + .sync_changes("mb-inbox", Some("sync-state-900")) + .await + .expect("an unrecognised checkpoint must re-enumerate"); + assert_eq!( + changes.present, + Some(vec![placement_key("mb-inbox", "m-a")]) + ); + assert_eq!(changes.next_state, "v1:jmapstate:srv-full-1"); +} + +/// `Email/changes` is account-wide (RFC 8621 s4.2 has no mailbox filter), so a +/// folder pass sees ids that now live in other mailboxes. A message another +/// client moved out has left this folder: its occupancy here is reconciled +/// away, and it is NOT placed here on the strength of having merely changed. +#[tokio::test] +async fn jmap_sync_changes_reconciles_a_message_moved_out_of_the_folder() { + let mock_server = MockServer::start().await; + mount_session(&mock_server).await; + mount_email_changes_page( + &mock_server, + "srv-state-1", + json!({ + "oldState": "srv-state-1", + "newState": "srv-state-2", + "updated": ["m-moved", "m-stayed"], + "hasMoreChanges": false + }), + ) + .await; + mount_email_get( + &mock_server, + "srv-state-2", + json!([ + message("m-moved", "mb-archive"), + message("m-stayed", "mb-inbox") + ]), + ) + .await; + + let engine = jmap_engine(&mock_server); + let changes = engine + .sync_changes("mb-inbox", Some("v1:jmapstate:srv-state-1")) + .await + .expect("incremental sync succeeds"); + + let upserted: Vec<&str> = changes + .upserts + .iter() + .map(|m| m.placement.remote_id.as_str()) + .collect(); + assert_eq!(upserted, vec!["m-stayed"]); + assert_eq!( + changes.removals, + vec![placement_key("mb-inbox", "m-moved")], + "a message whose mailboxes no longer include this folder must lose its occupancy here" + ); +} diff --git a/crates/nunciod/src/sync.rs b/crates/nunciod/src/sync.rs index 3b9dbe4..f30e8b6 100644 --- a/crates/nunciod/src/sync.rs +++ b/crates/nunciod/src/sync.rs @@ -940,10 +940,10 @@ mod tests { assert_eq!(synced, 1); // The message persists under an opaque surrogate id; it is recovered by - // folder, and its protocol-native JMAP object id round-trips in - // `remote_id`. + // the mailbox it was actually synced from, and its protocol-native JMAP + // object id round-trips in `remote_id`. let persisted = db - .list_messages("acct-jmap-real-1", "inbox", 10) + .list_messages("acct-jmap-real-1", "jmap-inbox", 10) .await .expect("jmap message persisted"); assert_eq!(persisted.len(), 1);