jmap: poll Email/changes and Mailbox/changes
One task per credential polls every 30s, so new and changed messages show up within a minute (task lylzwoyo). Sync states live in memory because boot re-imports everything — persisting them is task morwpurn — and destroyed ids are counted while their rows stay cached, with adopting them as Done its own task (yvusvlzz). Assisted-by: GLM-5.3 via pi
diff --git a/src/bin/docket/main.rs b/src/bin/docket/main.rs
index 763de43..63414be 100644
--- a/src/bin/docket/main.rs
+++ b/src/bin/docket/main.rs
@@ -1,4 +1,5 @@
mod config;
+mod poll;
use clap::Parser as _;
use miette::IntoDiagnostic as _;
@@ -33,13 +34,16 @@ async fn main() -> miette::Result<()> {
}
// One session per credential, before serving: a credential that
// can't open one is a config or token problem to fix, not an account
- // to quietly leave out.
+ // to quietly leave out. Each sync then polls for changes in the
+ // background.
let jmap = docket::jmap::Client::fastmail().into_diagnostic()?;
- for credential in config.credentials.iter() {
- jmap.sync_account(credential, &store)
+ for credential in config.credentials.iter().cloned() {
+ let sync = jmap
+ .sync_account(&credential, &store)
.await
.into_diagnostic()
.wrap_err_with(|| format!("opening the JMAP session for {}", credential.name))?;
+ poll::spawn(jmap.clone(), credential, store.clone(), sync);
}
let state = docket::routes::AppState::new(store);
#[cfg(feature = "dev")]
diff --git a/src/bin/docket/poll.rs b/src/bin/docket/poll.rs
new file mode 100644
index 0000000..c7d1a6a
--- /dev/null
+++ b/src/bin/docket/poll.rs
@@ -0,0 +1,36 @@
+//! The polling loop (task lylzwoyo, "Sync changes by polling"): one
+//! task per credential, applying `/changes` every 30s so new and
+//! changed messages show up within a minute. Push (task nwylszul,
+//! "Sync changes via JMAP push") replaces the timer later; the changes
+//! application it triggers stays the same.
+
+use std::time::Duration;
+
+use docket::jmap::{Client, Credential, Sync};
+use docket::store::Store;
+
+/// "Within a minute or so" with headroom for the fetches a cycle makes.
+const INTERVAL: Duration = Duration::from_secs(30);
+
+/// Runs the loop in the background. Errors are logged and retried on
+/// the next tick — a blip must not take syncing down with it.
+pub fn spawn(client: Client, credential: Credential, store: Store, sync: Sync) {
+ tokio::spawn(async move {
+ let name = credential.name.clone();
+ let mut sync = sync;
+ loop {
+ tokio::time::sleep(INTERVAL).await;
+ match client.poll_once(&credential, &mut sync, &store).await {
+ Ok(counts) if counts.is_quiet() => {
+ tracing::debug!(credential = %name, ?counts, "poll");
+ }
+ Ok(counts) => {
+ tracing::info!(credential = %name, ?counts, "poll");
+ }
+ Err(err) => {
+ tracing::warn!(credential = %name, %err, "poll failed; retrying next tick");
+ }
+ }
+ }
+ });
+}
diff --git a/src/jmap.rs b/src/jmap.rs
index 1fb54ed..f388f86 100644
--- a/src/jmap.rs
+++ b/src/jmap.rs
@@ -3,6 +3,11 @@
//! scope surfaces in the session as capabilities and `isReadOnly`;
//! per-mailbox `myRights` report mailbox ACLs, so they gate per-mailbox
//! actions like filing (task rn), never account-wide scope.
+//!
+//! Keeping up with the server is polling for now (task lylzwoyo):
+//! `poll_once` applies `Email/changes` and `Mailbox/changes` since the
+//! states the last sync left. Push replaces the timer later (task
+//! nwylszul); the changes application stays.
use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
@@ -51,6 +56,9 @@ pub enum JmapError {
detail: String,
},
+ #[error("{method} cannot calculate changes since the given state")]
+ CannotCalculateChanges { method: &'static str },
+
#[error("credential {credential:?} offers no mail account in its session")]
NoMailAccount { credential: String },
}
@@ -194,9 +202,13 @@ fn parse_reply(method: &'static str, json: &str) -> Result<Value, JmapError> {
name if name == method => Ok(args.clone()),
"error" => {
let err: ApiError = serde_json::from_value(args.clone()).map_err(malformed)?;
- Err(match err.description {
- Some(description) => detail(format!("{}: {description}", err.kind)),
- None => detail(err.kind),
+ Err(if err.kind == "cannotCalculateChanges" {
+ JmapError::CannotCalculateChanges { method }
+ } else {
+ match err.description {
+ Some(description) => detail(format!("{}: {description}", err.kind)),
+ None => detail(err.kind),
+ }
})
}
other => Err(detail(format!("unexpected response {other:?}"))),
@@ -219,6 +231,52 @@ fn parse_list<T: DeserializeOwned>(method: &'static str, args: Value) -> Result<
})
}
+/// Pulls a list and its type state out of a `Foo/get` response: that
+/// `state` is the `sinceState` the next `/changes` call starts from
+/// (RFC 8620 §5.2).
+fn parse_stateful_list<T: DeserializeOwned>(
+ method: &'static str,
+ args: Value,
+) -> Result<(Vec<T>, String), JmapError> {
+ #[derive(Deserialize)]
+ struct Got<T> {
+ state: String,
+ list: Vec<T>,
+ }
+ serde_json::from_value::<Got<T>>(args)
+ .map(|got| (got.list, got.state))
+ .map_err(|source| JmapError::Malformed {
+ what: method,
+ source,
+ })
+}
+
+/// One page of a `Foo/changes` reply (RFC 8620 §5.2).
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct Changes {
+ has_more_changes: bool,
+ new_state: String,
+ created: Vec<String>,
+ updated: Vec<String>,
+ destroyed: Vec<String>,
+}
+
+impl Changes {
+ /// Whether anything at all moved: even a destroy must rebuild the
+ /// mailbox layout.
+ fn changed(&self) -> bool {
+ !(self.created.is_empty() && self.updated.is_empty() && self.destroyed.is_empty())
+ }
+}
+
+fn parse_changes(method: &'static str, args: Value) -> Result<Changes, JmapError> {
+ serde_json::from_value(args).map_err(|source| JmapError::Malformed {
+ what: method,
+ source,
+ })
+}
+
/// An RFC 5322 address as JMAP parses it (RFC 8621 §4.1.4).
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
@@ -412,6 +470,14 @@ impl Layout {
.is_some_and(|id| email.mailbox_ids.get(id) == Some(&true))
}
+ /// True when the message sits in the Inbox mailbox: what the poll
+ /// path imports as received.
+ pub fn is_inbox(&self, email: &Email) -> bool {
+ self.inbox
+ .as_deref()
+ .is_some_and(|id| email.mailbox_ids.get(id) == Some(&true))
+ }
+
/// Folder names in the session's order, for the folders table.
pub fn folder_names(&self) -> Vec<String> {
self.folders.values().cloned().collect()
@@ -425,11 +491,44 @@ struct Imported {
sent: usize,
}
+/// What the poll loop carries between cycles: the type states to call
+/// `/changes` since, and the layout mail classifies against. Held in
+/// memory — boot re-imports everything, so a restart re-derives it.
+#[derive(Debug)]
+pub struct Sync {
+ pub email_state: String,
+ pub mailbox_state: String,
+ layout: Layout,
+}
+
+/// What one poll cycle did, for the loop's log line.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct PollCounts {
+ /// Messages imported or refreshed.
+ pub imported: usize,
+ /// Destroyed ids seen. The rows stay cached — adopting destroys as
+ /// Done is its own task.
+ pub destroyed: usize,
+ /// Whether the layout was rebuilt off `Mailbox/changes`.
+ pub mailboxes: bool,
+ /// Whether the server disowned the stored states and a full
+ /// re-import ran instead.
+ pub resynced: bool,
+}
+
+impl PollCounts {
+ /// A cycle with nothing to show, logged at debug rather than info.
+ pub fn is_quiet(&self) -> bool {
+ *self == Self::default()
+ }
+}
+
/// One page of request and response; small enough to keep bodies cheap.
const PAGE: usize = 50;
/// Opens sessions: one per credential, against Fastmail unless a test
/// points the session URL at a stub.
+#[derive(Clone)]
pub struct Client {
http: reqwest::Client,
session_url: String,
@@ -463,13 +562,13 @@ impl Client {
}
/// Every mailbox of the account, with the rights the server reports
- /// for it.
+ /// for it, and the type state the next `/changes` starts from.
async fn mailboxes(
&self,
session: &Session,
token: &str,
account_id: &str,
- ) -> Result<Vec<Mailbox>, JmapError> {
+ ) -> Result<(Vec<Mailbox>, String), JmapError> {
let args = self
.call(
&session.api_url,
@@ -479,7 +578,7 @@ impl Client {
"properties": ["id", "name", "role", "myRights"]}),
)
.await?;
- parse_list("Mailbox/get", args)
+ parse_stateful_list("Mailbox/get", args)
}
/// Posts one method call and returns the paired response's arguments.
@@ -504,14 +603,15 @@ impl Client {
parse_reply(method, &body)
}
- /// The named emails, with bodies.
+ /// The named emails, with bodies, and the type state the next
+ /// `/changes` starts from.
async fn emails(
&self,
api_url: &str,
token: &str,
account_id: &str,
ids: &[String],
- ) -> Result<Vec<Email>, JmapError> {
+ ) -> Result<(Vec<Email>, String), JmapError> {
let args = self
.call(
api_url,
@@ -525,7 +625,28 @@ impl Client {
"bodyValues": ["text/plain"], "textBody": true}}),
)
.await?;
- parse_list("Email/get", args)
+ parse_stateful_list("Email/get", args)
+ }
+
+ /// One page of `Foo/changes` since `since_state`.
+ async fn changes(
+ &self,
+ api_url: &str,
+ token: &str,
+ account_id: &str,
+ method: &'static str,
+ since_state: &str,
+ ) -> Result<Changes, JmapError> {
+ let args = self
+ .call(
+ api_url,
+ token,
+ method,
+ json!({"accountId": account_id, "sinceState": since_state,
+ "maxChanges": PAGE}),
+ )
+ .await?;
+ parse_changes(method, args)
}
/// Opens the credential's session, records its account, and imports
@@ -533,15 +654,20 @@ impl Client {
/// [`Import::account`] is the seam; the upserts underneath make this
/// the refresh path too — rights and the mail cache re-read every
/// time a session opens, while Docket-owned values (state,
- /// assignees, reads) keep whatever they already hold.
+ /// assignees, reads) keep whatever they already hold. Returns the
+ /// states the import left the account in, for the poll loop.
///
/// [`Import::account`]: crate::store::Import::account
- pub async fn sync_account(&self, credential: &Credential, store: &Store) -> Result<(), Error> {
+ pub async fn sync_account(
+ &self,
+ credential: &Credential,
+ store: &Store,
+ ) -> Result<Sync, Error> {
let token = read_token(&credential.token_file)?;
let session = self.session(&token).await?;
let (id, account) = session.mail_account(&credential.name)?;
let rights = account.rights();
- let mailboxes = self.mailboxes(&session, &token, id).await?;
+ let (mailboxes, mailbox_state) = self.mailboxes(&session, &token, id).await?;
let layout = Layout::of(&mailboxes);
let users = store.users()?;
let folders = layout.folder_names();
@@ -560,6 +686,7 @@ impl Client {
}
Ok(())
})?;
+ let mut email_state = None;
let imported = self
.import_mail(
&session,
@@ -569,8 +696,15 @@ impl Client {
&layout,
&users,
store,
+ &mut email_state,
)
.await?;
+ // An Inbox-less account never fetched a page, so no Email/get
+ // reported a state; ask for nothing to learn one.
+ let email_state = match email_state {
+ Some(state) => state,
+ None => self.emails(&session.api_url, &token, id, &[]).await?.1,
+ };
tracing::info!(
credential = %credential.name,
account = %id,
@@ -581,7 +715,11 @@ impl Client {
sent = imported.sent,
"session opened",
);
- Ok(())
+ Ok(Sync {
+ email_state,
+ mailbox_state,
+ layout,
+ })
}
/// The account's Inbox mail plus our sent replies in its threads:
@@ -600,6 +738,7 @@ impl Client {
layout: &Layout,
users: &[User],
store: &Store,
+ email_state: &mut Option<String>,
) -> Result<Imported, Error> {
let mut imported = Imported::default();
let Some(inbox) = layout.inbox.clone() else {
@@ -627,13 +766,19 @@ impl Client {
if n == 0 {
break;
}
- for email in self
+ let (emails, state) = self
.emails(&session.api_url, token, account_id, &page.ids)
- .await?
- {
+ .await?;
+ // The first page's state is the baseline the poll loop
+ // starts from: mail landing mid-import is then changed-since
+ // this state, so the first poll sweeps it in.
+ if email_state.is_none() {
+ *email_state = Some(state);
+ }
+ for email in &emails {
known.insert(email.id.clone());
threads.insert(email.thread_id.clone());
- if let Some(mail) = incoming(&email, layout, users) {
+ if let Some(mail) = incoming(email, layout, users) {
store.import(|tx| tx.incoming(slug, &mail))?;
imported.received = imported.received.saturating_add(1);
}
@@ -661,12 +806,12 @@ impl Client {
}
}
for chunk in wanted.chunks(PAGE) {
- for email in self
+ let (emails, _) = self
.emails(&session.api_url, token, account_id, chunk)
- .await?
- {
- if layout.is_sent(&email)
- && let Some(mail) = incoming(&email, layout, users)
+ .await?;
+ for email in &emails {
+ if layout.is_sent(email)
+ && let Some(mail) = incoming(email, layout, users)
{
store.import(|tx| tx.incoming(slug, &mail))?;
imported.sent = imported.sent.saturating_add(1);
@@ -675,6 +820,133 @@ impl Client {
}
Ok(imported)
}
+
+ /// One poll cycle: `/changes` for mailboxes and email since the
+ /// states the account was left in, applying what moved. A
+ /// `cannotCalculateChanges` reply means the server disowns those
+ /// states (RFC 8620 §5.2), so the cycle re-imports from scratch
+ /// and the caller carries the fresh states forward.
+ pub async fn poll_once(
+ &self,
+ credential: &Credential,
+ sync: &mut Sync,
+ store: &Store,
+ ) -> Result<PollCounts, Error> {
+ let token = read_token(&credential.token_file)?;
+ let session = self.session(&token).await?;
+ let (id, account) = session.mail_account(&credential.name)?;
+ let mut counts = PollCounts::default();
+
+ let mailboxes = match self
+ .changes(
+ &session.api_url,
+ &token,
+ id,
+ "Mailbox/changes",
+ &sync.mailbox_state,
+ )
+ .await
+ {
+ Ok(page) => page,
+ Err(JmapError::CannotCalculateChanges { .. }) => {
+ *sync = self.sync_account(credential, store).await?;
+ counts.resynced = true;
+ return Ok(counts);
+ }
+ Err(other) => return Err(other.into()),
+ };
+ // An unchanged reply still advances the state (RFC 8620 §5.2).
+ sync.mailbox_state = mailboxes.new_state.clone();
+ if mailboxes.changed() {
+ let (mailboxes, mailbox_state) = self.mailboxes(&session, &token, id).await?;
+ sync.layout = Layout::of(&mailboxes);
+ sync.mailbox_state = mailbox_state;
+ let rights = account.rights();
+ let slug = credential.name.clone();
+ let name = account.name.clone();
+ let address = id.to_owned();
+ let folders = sync.layout.folder_names();
+ store.import(|tx| {
+ tx.account(&Account {
+ slug,
+ name,
+ address,
+ read_only: rights.read_only,
+ })?;
+ for folder in &folders {
+ tx.folder(folder)?;
+ }
+ Ok(())
+ })?;
+ counts.mailboxes = true;
+ }
+
+ loop {
+ let page = match self
+ .changes(
+ &session.api_url,
+ &token,
+ id,
+ "Email/changes",
+ &sync.email_state,
+ )
+ .await
+ {
+ Ok(page) => page,
+ Err(JmapError::CannotCalculateChanges { .. }) => {
+ *sync = self.sync_account(credential, store).await?;
+ counts.resynced = true;
+ return Ok(counts);
+ }
+ Err(other) => return Err(other.into()),
+ };
+ let Changes {
+ has_more_changes,
+ new_state,
+ created,
+ updated,
+ destroyed,
+ } = page;
+ if !destroyed.is_empty() {
+ // Rows stay cached; adopting destroys as Done is its own
+ // task.
+ tracing::debug!(?destroyed, "mail destroyed server-side");
+ }
+ counts.destroyed = counts.destroyed.saturating_add(destroyed.len());
+ let mut ids = created;
+ ids.extend(updated);
+ ids.sort();
+ ids.dedup();
+ if !ids.is_empty() {
+ let (emails, _) = self.emails(&session.api_url, &token, id, &ids).await?;
+ let users = store.users()?;
+ // Received first: a new Inbox message founds the thread
+ // its same-batch sent reply then lands in.
+ for email in &emails {
+ if sync.layout.is_inbox(email)
+ && let Some(mail) = incoming(email, &sync.layout, &users)
+ {
+ store.import(|tx| tx.incoming(&credential.name, &mail))?;
+ counts.imported = counts.imported.saturating_add(1);
+ }
+ }
+ for email in &emails {
+ if sync.layout.is_sent(email)
+ && store.has_thread(&credential.name, &email.thread_id)?
+ && let Some(mail) = incoming(email, &sync.layout, &users)
+ {
+ store.import(|tx| tx.incoming(&credential.name, &mail))?;
+ counts.imported = counts.imported.saturating_add(1);
+ }
+ }
+ }
+ sync.email_state = new_state;
+ if !has_more_changes {
+ break;
+ }
+ }
+ Ok(counts)
+ }
}
/// The token stays in its file (DESIGN.md, Storage), read fresh each
@@ -1073,6 +1345,7 @@ mod tests {
assert_eq!(layout.state_of(&plain), State::Inbox);
assert_eq!(layout.folder_of(&plain), None);
assert!(!layout.is_sent(&plain));
+ assert!(layout.is_inbox(&plain));
let labeled = email(&[("M-in", true), ("M-watch", true)]);
assert_eq!(layout.state_of(&labeled), State::Watch);
@@ -1081,8 +1354,14 @@ mod tests {
assert_eq!(layout.state_of(&filed), State::Inbox);
assert_eq!(layout.folder_of(&filed).as_deref(), Some("Receipts"));
+ // Not a member: filed out of Inbox, or a sent message.
+ let gone = email(&[("M-receipts", true)]);
+ assert!(!layout.is_inbox(&gone));
+ assert!(!layout.is_inbox(&email(&[("M-in", false)])));
+
let mine = email(&[("M-sent", true)]);
assert!(layout.is_sent(&mine));
+ assert!(!layout.is_inbox(&mine));
}
#[test]
@@ -1127,6 +1406,62 @@ mod tests {
assert!(err.to_string().contains("malformed Email/query"), "{err}");
}
+ #[test]
+ fn changes_parse_with_their_paging_fields() {
+ let args = json!({
+ "accountId": "a", "oldState": "e0", "newState": "e1",
+ "hasMoreChanges": true,
+ "created": ["E1"], "updated": ["E2"], "destroyed": ["E3"],
+ });
+ let page = parse_changes("Email/changes", args).unwrap();
+ assert!(page.changed());
+ assert!(page.has_more_changes);
+ assert_eq!(page.new_state, "e1");
+ assert_eq!(page.created, ["E1".to_owned()]);
+ assert_eq!(page.updated, ["E2".to_owned()]);
+ assert_eq!(page.destroyed, ["E3".to_owned()]);
+
+ let quiet = parse_changes(
+ "Email/changes",
+ json!({"accountId": "a", "oldState": "e1", "newState": "e1",
+ "hasMoreChanges": false,
+ "created": [], "updated": [], "destroyed": []}),
+ )
+ .unwrap();
+ assert!(!quiet.changed());
+
+ let err = parse_changes("Email/changes", json!({"created": 3})).unwrap_err();
+ assert!(err.to_string().contains("malformed Email/changes"), "{err}");
+ }
+
+ #[test]
+ fn a_stateful_list_carries_its_type_state() {
+ let args = json!({"state": "m3", "list": [{"id": "M1", "name": "Inbox",
+ "role": "inbox", "myRights": rights(true, true, true, true,
+ true, true, true, true)}]});
+ let (mailboxes, state) = parse_stateful_list::<Mailbox>("Mailbox/get", args).unwrap();
+ assert_eq!(state, "m3");
+ assert_eq!(mailboxes.len(), 1);
+
+ let err = parse_stateful_list::<Mailbox>("Mailbox/get", json!({"list": []})).unwrap_err();
+ assert!(err.to_string().contains("malformed Mailbox/get"), "{err}");
+ }
+
+ #[test]
+ fn a_cannot_calculate_reply_is_its_own_error() {
+ let reply = json!({
+ "methodResponses": [[
+ "error",
+ {"type": "cannotCalculateChanges"},
+ "0",
+ ]],
+ })
+ .to_string();
+ let err = parse_reply("Email/changes", &reply).unwrap_err();
+ assert!(matches!(err, JmapError::CannotCalculateChanges { .. }));
+ assert!(err.to_string().contains("cannot calculate"), "{err}");
+ }
+
#[test]
fn incoming_mail_carries_its_derived_values() {
let layout = Layout::of(&mailboxes());
diff --git a/src/store.rs b/src/store.rs
index b0c6506..cfb32f8 100644
--- a/src/store.rs
+++ b/src/store.rs
@@ -5,7 +5,7 @@
//! memory.
use std::collections::{BTreeMap, BTreeSet};
-use std::sync::{Mutex, MutexGuard, PoisonError};
+use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use camino::Utf8Path;
use jiff::civil::DateTime;
@@ -77,11 +77,11 @@ impl Clock {
/// One connection behind a mutex: two people's clicks never contend long
/// enough to need a pool, and no lock is held across an await.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
pub struct Store {
/// The clock events and comments are stamped with.
clock: Clock,
- inner: Mutex<Inner>,
+ inner: Arc<Mutex<Inner>>,
}
#[derive(Debug)]
@@ -110,11 +110,11 @@ impl Store {
migrate(&mut conn)?;
Ok(Self {
clock,
- inner: Mutex::new(Inner {
+ inner: Arc::new(Mutex::new(Inner {
conn,
undo: BTreeMap::new(),
flash: BTreeMap::new(),
- }),
+ })),
})
}
@@ -193,6 +193,18 @@ impl Store {
Ok(accounts)
}
+ /// True when the account's thread is already imported: the poll
+ /// path admits a sent email only into a thread it knows.
+ pub fn has_thread(&self, account: &str, jmap_thread_id: &str) -> Result<bool, Error> {
+ let found = self.lock().conn.query_row(
+ "SELECT EXISTS(SELECT 1 FROM threads
+ WHERE account = ?1 AND jmap_thread_id = ?2)",
+ params![account, jmap_thread_id],
+ |r| r.get(0),
+ )?;
+ Ok(found)
+ }
+
pub fn thread(&self, id: ThreadId) -> Result<Option<Thread>, Error> {
let thread = self
.lock()
@@ -1129,6 +1141,30 @@ mod tests {
}
}
+ #[test]
+ fn has_thread_reports_whether_a_threads_imported() {
+ let store = Store::open_in_memory(Clock::Fixed(fixtures::now())).unwrap();
+ store
+ .import(|tx| {
+ tx.account(&Account {
+ slug: "household".into(),
+ name: "Household".into(),
+ address: "household@example.com".into(),
+ read_only: false,
+ })?;
+ tx.incoming("household", &incoming(State::Inbox, None, "Body"))
+ })
+ .unwrap();
+
+ assert!(store.has_thread("household", "T9").unwrap());
+ assert!(!store.has_thread("household", "T8").unwrap());
+ // Threads are scoped to the account.
+ assert!(!store.has_thread("eli", "T9").unwrap());
+
+ exec(&store, "ALTER TABLE threads RENAME TO gone_threads;");
+ assert!(store.has_thread("household", "T9").is_err());
+ }
+
#[test]
fn incoming_mail_upserts_keeping_docket_owned_values() {
let store = Store::open_in_memory(Clock::Fixed(fixtures::now())).unwrap();
diff --git a/tests/jmap.rs b/tests/jmap.rs
index a98f4fa..0710003 100644
--- a/tests/jmap.rs
+++ b/tests/jmap.rs
@@ -2,7 +2,10 @@
//! opens a session, its mail imports, and the accounts land in the store
//! with the rights the server reported.
+use std::collections::{BTreeMap, BTreeSet};
+use std::hash::{Hash, Hasher};
use std::net::SocketAddr;
+use std::sync::{Arc, Mutex};
use axum::http::header::{AUTHORIZATION, HOST};
use axum::http::{HeaderMap, StatusCode};
@@ -88,14 +91,19 @@ async fn session(headers: HeaderMap) -> Response {
}
}
-/// The mailboxes both accounts carry: system boxes by role, `Docket/`
-/// labels, and role-less folders.
-fn mailboxes() -> Vec<Value> {
- let rights = json!({
+/// The ACLs of a mailbox the token fully controls.
+fn full_rights() -> Value {
+ json!({
"mayReadItems": true, "mayAddItems": true, "mayRemoveItems": true,
"maySetKeywords": true, "mayCreateChild": true, "mayRename": true,
"mayDelete": true, "maySubmit": true,
- });
+ })
+}
+
+/// The mailboxes both accounts carry: system boxes by role, `Docket/`
+/// labels, and role-less folders.
+fn mailboxes() -> Vec<Value> {
+ let rights = full_rights();
let restricted = json!({
"mayReadItems": true, "mayAddItems": false, "mayRemoveItems": false,
"maySetKeywords": false, "mayCreateChild": false, "mayRename": false,
@@ -279,7 +287,161 @@ fn threads_of(account: &str) -> Vec<Value> {
}
}
-async fn api(headers: HeaderMap, body: String) -> Response {
+/// The stub's mutable side: each account's mail as JSON, plus a
+/// snapshot of each type per state it has handed out, so `/changes`
+/// can diff since any state a client holds.
+#[derive(Default)]
+struct Stub {
+ accounts: BTreeMap<&'static str, AccountMail>,
+}
+
+#[derive(Default)]
+struct AccountMail {
+ mailboxes: Vec<Value>,
+ emails: Vec<Value>,
+ threads: Vec<Value>,
+ /// state → each email's hash at that state.
+ email_states: Vec<(String, BTreeMap<String, u64>)>,
+ mailbox_states: Vec<(String, BTreeMap<String, u64>)>,
+}
+
+type World = Arc<Mutex<Stub>>;
+
+impl Stub {
+ fn hash(mail: &Value) -> u64 {
+ let mut hasher = std::collections::hash_map::DefaultHasher::new();
+ serde_json::to_string(mail).unwrap().hash(&mut hasher);
+ hasher.finish()
+ }
+
+ /// The fixture mail under its first states.
+ fn seed() -> Self {
+ let mut stub = Self::default();
+ for account in [HOUSEHOLD, ELI, BARE] {
+ stub.accounts.insert(
+ account,
+ AccountMail {
+ mailboxes: mailbox_list(account),
+ emails: emails_of(account),
+ threads: threads_of(account),
+ ..AccountMail::default()
+ },
+ );
+ }
+ for account in [HOUSEHOLD, ELI, BARE] {
+ stub.snapshot(account);
+ }
+ stub
+ }
+
+ /// Records the current mail under a fresh state per type, the way
+ /// a server bumps its Foo/get `state` when anything changes.
+ fn snapshot(&mut self, account: &'static str) {
+ let mail = self.accounts.get_mut(account).unwrap();
+ let emails = mail
+ .emails
+ .iter()
+ .map(|e| (e["id"].as_str().unwrap().to_owned(), Self::hash(e)))
+ .collect();
+ let n = mail.email_states.len();
+ mail.email_states.push((format!("e{n}"), emails));
+ let mailboxes = mail
+ .mailboxes
+ .iter()
+ .map(|m| (m["id"].as_str().unwrap().to_owned(), Self::hash(m)))
+ .collect();
+ let n = mail.mailbox_states.len();
+ mail.mailbox_states.push((format!("m{n}"), mailboxes));
+ }
+
+ fn add_email(&mut self, account: &'static str, email: Value) {
+ self.accounts.get_mut(account).unwrap().emails.push(email);
+ self.snapshot(account);
+ }
+
+ fn update_email(&mut self, account: &'static str, id: &str, change: impl FnOnce(&mut Value)) {
+ let mail = self.accounts.get_mut(account).unwrap();
+ let email = mail
+ .emails
+ .iter_mut()
+ .find(|e| e["id"] == json!(id))
+ .unwrap();
+ change(email);
+ self.snapshot(account);
+ }
+
+ fn remove_email(&mut self, account: &'static str, id: &str) {
+ let mail = self.accounts.get_mut(account).unwrap();
+ mail.emails.retain(|e| e["id"] != json!(id));
+ self.snapshot(account);
+ }
+
+ fn add_mailbox(&mut self, account: &'static str, mailbox: Value) {
+ self.accounts
+ .get_mut(account)
+ .unwrap()
+ .mailboxes
+ .push(mailbox);
+ self.snapshot(account);
+ }
+}
+
+/// The `/changes` reply for one type: the diff from the named state to
+/// the latest, cut on a snapshot boundary when it exceeds
+/// `max_changes`, so `hasMoreChanges` pages like a real server. A
+/// `since` the stub never handed out is a state it cannot calculate.
+fn changes_reply(
+ account: &str,
+ states: &[(String, BTreeMap<String, u64>)],
+ since: &str,
+ max_changes: usize,
+) -> Result<Value, ()> {
+ let start = states
+ .iter()
+ .position(|(state, _)| state == since)
+ .ok_or(())?;
+ let mut now = states[start].1.clone();
+ let mut new_state = states[start].0.clone();
+ let (mut created, mut updated, mut destroyed) =
+ (BTreeSet::new(), BTreeSet::new(), BTreeSet::new());
+ for (state, snapshot) in &states[start + 1..] {
+ let mut touched = (BTreeSet::new(), BTreeSet::new(), BTreeSet::new());
+ for (id, hash) in snapshot {
+ match now.get(id) {
+ None => {
+ touched.0.insert(id.clone());
+ }
+ Some(old) if old != hash => {
+ touched.1.insert(id.clone());
+ }
+ _ => {}
+ }
+ }
+ for id in now.keys() {
+ if !snapshot.contains_key(id) {
+ touched.2.insert(id.clone());
+ }
+ }
+ let n = touched.0.len() + touched.1.len() + touched.2.len();
+ if created.len() + updated.len() + destroyed.len() + n > max_changes {
+ break;
+ }
+ created.extend(touched.0);
+ updated.extend(touched.1);
+ destroyed.extend(touched.2);
+ now = snapshot.clone();
+ new_state = state.clone();
+ }
+ Ok(json!({
+ "accountId": account, "oldState": since, "newState": new_state,
+ "hasMoreChanges": new_state != states.last().unwrap().0,
+ "created": created.iter().collect::<Vec<_>>(),
+ "updated": updated.iter().collect::<Vec<_>>(),
+ "destroyed": destroyed.iter().collect::<Vec<_>>(),
+ }))
+}
+
+async fn api(world: axum::extract::State<World>, headers: HeaderMap, body: String) -> Response {
let Some(account) = bearer(&headers).and_then(account_of) else {
return (StatusCode::UNAUTHORIZED, "unknown token").into_response();
};
@@ -289,15 +451,20 @@ async fn api(headers: HeaderMap, body: String) -> Response {
let method = call[0].as_str().expect("a method name").to_owned();
let args = &call[1];
+ let stub = world.lock().unwrap();
+ let mail = stub.accounts.get(account).unwrap();
let reply = match method.as_str() {
- "Mailbox/get" => json!({"accountId": account, "state": "s", "notFound": [],
- "list": mailbox_list(account)}),
+ "Mailbox/get" => json!({
+ "accountId": account,
+ "state": mail.mailbox_states.last().unwrap().0,
+ "notFound": [], "list": mail.mailboxes.clone(),
+ }),
"Email/query" => {
let inbox = args["filter"]["inMailbox"].as_str().expect("an inMailbox");
let position = args["position"].as_u64().unwrap_or_default() as usize;
let limit = args["limit"].as_u64().unwrap_or_default() as usize;
- let inbox_emails = emails_of(account);
- let ids: Vec<&str> = inbox_emails
+ let ids: Vec<&str> = mail
+ .emails
.iter()
.filter(|e| e["mailboxIds"][inbox] == json!(true))
.filter_map(|e| e["id"].as_str())
@@ -311,10 +478,11 @@ async fn api(headers: HeaderMap, body: String) -> Response {
.as_array()
.map(|ids| ids.iter().filter_map(Value::as_str).collect())
.unwrap_or_default();
- json!({"accountId": account, "state": "s", "notFound": [],
- "list": emails_of(account).into_iter()
+ json!({"accountId": account, "state": mail.email_states.last().unwrap().0,
+ "notFound": [],
+ "list": mail.emails.iter()
.filter(|e| wanted.contains(&e["id"].as_str().expect("an id")))
- .collect::<Vec<_>>()})
+ .cloned().collect::<Vec<_>>()})
}
"Thread/get" => {
let wanted: Vec<&str> = args["ids"]
@@ -322,9 +490,33 @@ async fn api(headers: HeaderMap, body: String) -> Response {
.map(|ids| ids.iter().filter_map(Value::as_str).collect())
.unwrap_or_default();
json!({"accountId": account, "state": "s", "notFound": [],
- "list": threads_of(account).into_iter()
+ "list": mail.threads.iter()
.filter(|t| wanted.contains(&t["id"].as_str().expect("an id")))
- .collect::<Vec<_>>()})
+ .cloned().collect::<Vec<_>>()})
+ }
+ "Email/changes" | "Mailbox/changes" => {
+ let since = args["sinceState"].as_str().expect("a sinceState");
+ let max = args["maxChanges"].as_u64().unwrap_or_default() as usize;
+ // A state the tests reserve for making the call fail
+ // outright, distinct from one the stub cannot calculate.
+ if since.ends_with("-fail") {
+ return Json(json!({"methodResponses":
+ [["error", {"type": "serverFail", "description": "stub"}, "0"]]}))
+ .into_response();
+ }
+ let states = if method == "Email/changes" {
+ &mail.email_states
+ } else {
+ &mail.mailbox_states
+ };
+ match changes_reply(account, states, since, max) {
+ Ok(reply) => reply,
+ Err(()) => {
+ return Json(json!({"methodResponses":
+ [["error", {"type": "cannotCalculateChanges"}, "0"]]}))
+ .into_response();
+ }
+ }
}
other => panic!("unexpected method {other}"),
};
@@ -344,16 +536,18 @@ fn mailbox_list(account: &str) -> Vec<Value> {
}
}
-async fn serve() -> SocketAddr {
+async fn serve() -> (SocketAddr, World) {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
+ let world: World = Arc::new(Mutex::new(Stub::seed()));
let app = Router::new()
.route("/session", get(session))
- .route("/api", post(api));
+ .route("/api", post(api))
+ .with_state(world.clone());
tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
- addr
+ (addr, world)
}
/// Writes the token to its file with a trailing newline, the way a secret
@@ -370,7 +564,7 @@ fn credential(dir: &tempfile::TempDir, name: &str, token: &str) -> Credential {
#[tokio::test]
async fn each_credential_opens_a_session_and_imports_its_mail() {
init_tracing();
- let addr = serve().await;
+ let (addr, _world) = serve().await;
let dir = tempfile::tempdir().unwrap();
let household = credential(&dir, "household", HOUSEHOLD_TOKEN);
let eli = credential(&dir, "eli", ELI_TOKEN);
@@ -463,7 +657,7 @@ async fn each_credential_opens_a_session_and_imports_its_mail() {
#[tokio::test]
async fn a_rejected_token_fails_the_sync() {
- let addr = serve().await;
+ let (addr, _world) = serve().await;
let dir = tempfile::tempdir().unwrap();
let wrong = credential(&dir, "household", "not-the-token");
@@ -477,3 +671,356 @@ async fn a_rejected_token_fails_the_sync() {
assert!(err.contains("401"), "{err}");
assert!(store.accounts().unwrap().is_empty());
}
+
+/// The imported message with the given Message-ID, if any.
+fn the(store: &Store, message_id: &str) -> Option<docket::model::Message> {
+ store
+ .messages(Filter::Search(""))
+ .unwrap()
+ .into_iter()
+ .find(|m| m.message_id == message_id)
+}
+
+/// A household account synced and ready to poll.
+async fn synced(
+ addr: &SocketAddr,
+ store: &Store,
+ credential: &Credential,
+) -> (Client, docket::jmap::Sync) {
+ let client = Client::new(format!("http://{addr}/session")).unwrap();
+ let sync = client.sync_account(credential, store).await.unwrap();
+ (client, sync)
+}
+
+/// Boots a test world: stub, store with Sam signed in, and the
+/// household credential synced.
+async fn household_ready() -> (
+ World,
+ Client,
+ Credential,
+ docket::jmap::Sync,
+ Store,
+ tempfile::TempDir,
+) {
+ init_tracing();
+ let (addr, world) = serve().await;
+ let dir = tempfile::tempdir().unwrap();
+ let household = credential(&dir, "household", HOUSEHOLD_TOKEN);
+ let store = Store::open_in_memory(Clock::System).unwrap();
+ store.sign_in("sam@example.com", "Sam").unwrap();
+ let (client, sync) = synced(&addr, &store, &household).await;
+ (world, client, household, sync, store, dir)
+}
+
+#[tokio::test]
+async fn a_failing_changes_call_surfaces_the_error() {
+ let (_world, client, household, mut sync, store, _dir) = household_ready().await;
+
+ sync.email_state = "e-fail".into();
+ let err = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains("Email/changes"), "{err}");
+ assert!(err.contains("serverFail"), "{err}");
+
+ sync.email_state = "e0".into();
+ sync.mailbox_state = "m-fail".into();
+ let err = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains("Mailbox/changes"), "{err}");
+}
+
+#[tokio::test]
+async fn a_quiet_poll_changes_nothing() {
+ let (world, client, household, mut sync, store, _dir) = household_ready().await;
+ let before = store.messages(Filter::Search("")).unwrap().len();
+
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+
+ assert!(counts.is_quiet(), "{counts:?}");
+ assert_eq!(store.messages(Filter::Search("")).unwrap().len(), before);
+ assert_eq!(store.accounts().unwrap().len(), 1);
+ drop(world);
+}
+
+#[tokio::test]
+async fn new_inbox_mail_arrives_within_one_poll() {
+ let (world, client, household, mut sync, store, _dir) = household_ready().await;
+ world.lock().unwrap().add_email(
+ HOUSEHOLD,
+ email_json(
+ "E-new",
+ "T-new",
+ ("City Water", "billing@citywater.gov"),
+ "household",
+ &["MB-in"],
+ "Water main flush notice",
+ Some("Flushing on Thursday."),
+ "Flushing on Thursday",
+ ),
+ );
+
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+
+ assert_eq!(counts.imported, 1, "{counts:?}");
+ let arrived = the(&store, "E-new@chislan.family").unwrap();
+ assert_eq!(arrived.values().unwrap().state, State::Inbox);
+ assert_eq!(arrived.values().unwrap().folder, None);
+}
+
+#[tokio::test]
+async fn edits_refresh_the_row_while_state_adoption_waits() {
+ let (world, client, household, mut sync, store, _dir) = household_ready().await;
+
+ // A folder picked up elsewhere refreshes the cached row.
+ world
+ .lock()
+ .unwrap()
+ .update_email(HOUSEHOLD, "E-roofer1", |mail| {
+ mail["mailboxIds"]["MB-receipts"] = json!(true);
+ });
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+ assert_eq!(counts.imported, 1, "{counts:?}");
+ let roofer1 = the(&store, "E-roofer1@chislan.family").unwrap();
+ assert_eq!(
+ roofer1.values().unwrap().folder.as_deref(),
+ Some("Receipts")
+ );
+
+ // A `Docket/` label added elsewhere is adoption's to apply, not the
+ // poll's: Docket-owned state keeps what it holds.
+ world
+ .lock()
+ .unwrap()
+ .update_email(HOUSEHOLD, "E-roofer1", |mail| {
+ mail["mailboxIds"]["MB-do"] = json!(true);
+ });
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+ assert_eq!(counts.imported, 1, "{counts:?}");
+ let roofer1 = the(&store, "E-roofer1@chislan.family").unwrap();
+ assert_eq!(roofer1.values().unwrap().state, State::Inbox);
+}
+
+#[tokio::test]
+async fn destroyed_mail_stays_cached() {
+ let (world, client, household, mut sync, store, _dir) = household_ready().await;
+ world.lock().unwrap().remove_email(HOUSEHOLD, "E-school");
+
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+
+ assert_eq!(counts.destroyed, 1, "{counts:?}");
+ assert_eq!(counts.imported, 0, "{counts:?}");
+ assert!(!counts.is_quiet());
+ assert!(the(&store, "E-school@chislan.family").is_some());
+}
+
+#[tokio::test]
+async fn a_reimported_message_matches_by_message_id() {
+ let (world, client, household, mut sync, store, _dir) = household_ready().await;
+ world.lock().unwrap().remove_email(HOUSEHOLD, "E-roofer1");
+ // The same message under a new server id, as a reimport or restore
+ // brings it back.
+ let mut again = email_json(
+ "E-roofer1b",
+ "T-roofer",
+ ("Northwind Roofing", "office@northwind.co"),
+ "household",
+ &["MB-in"],
+ "Gutter repair estimate",
+ Some("Estimate attached: $1,840."),
+ "Estimate attached",
+ );
+ again["messageId"] = json!("E-roofer1@chislan.family");
+ world.lock().unwrap().add_email(HOUSEHOLD, again);
+
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+
+ assert_eq!(counts.imported, 1, "{counts:?}");
+ assert_eq!(store.messages(Filter::Search("")).unwrap().len(), 5);
+ assert!(the(&store, "E-roofer1@chislan.family").is_some());
+}
+
+#[tokio::test]
+async fn sent_replies_join_known_threads_only() {
+ let (world, client, household, mut sync, store, _dir) = household_ready().await;
+ world.lock().unwrap().add_email(
+ HOUSEHOLD,
+ email_json(
+ "E-reply4",
+ "T-roofer",
+ ("Sam", "sam@example.com"),
+ "office@northwind.co",
+ &["MB-sent"],
+ "Re: Gutter repair estimate",
+ Some("Does the fascia need work too?"),
+ "Does the fascia",
+ ),
+ );
+ world.lock().unwrap().add_email(
+ HOUSEHOLD,
+ email_json(
+ "E-solo",
+ "T-unknown",
+ ("Sam", "sam@example.com"),
+ "someone@elsewhere.org",
+ &["MB-sent"],
+ "Re: unrelated",
+ Some("Hello"),
+ "Hello",
+ ),
+ );
+
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+
+ assert_eq!(counts.imported, 1, "{counts:?}");
+ let reply = the(&store, "E-reply4@chislan.family").unwrap();
+ assert!(
+ matches!(&reply.kind, Kind::Sent { by, .. } if by.as_deref() == Some("sam@example.com"))
+ );
+ assert!(the(&store, "E-solo@chislan.family").is_none());
+}
+
+#[tokio::test]
+async fn a_disowned_state_triggers_a_full_reimport() {
+ let (world, client, household, mut sync, store, _dir) = household_ready().await;
+ sync.email_state = "e-bogus".into();
+
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+
+ assert!(counts.resynced, "{counts:?}");
+ assert_eq!(store.messages(Filter::Search("")).unwrap().len(), 5);
+
+ // The fresh baseline carries forward: mail arriving after the
+ // re-import still shows up on the next poll.
+ world.lock().unwrap().add_email(
+ HOUSEHOLD,
+ email_json(
+ "E-after",
+ "T-after",
+ ("City Water", "billing@citywater.gov"),
+ "household",
+ &["MB-in"],
+ "Bill available",
+ Some("Your bill is attached."),
+ "Your bill",
+ ),
+ );
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+ assert_eq!(counts.imported, 1, "{counts:?}");
+ assert!(the(&store, "E-after@chislan.family").is_some());
+}
+
+#[tokio::test]
+async fn a_disowned_mailbox_state_reimports_too() {
+ let (_world, client, household, mut sync, store, _dir) = household_ready().await;
+ sync.mailbox_state = "m-bogus".into();
+
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+
+ assert!(counts.resynced, "{counts:?}");
+ assert_eq!(store.accounts().unwrap().len(), 1);
+}
+
+#[tokio::test]
+async fn mailbox_changes_rebuild_the_layout() {
+ let (world, client, household, mut sync, store, _dir) = household_ready().await;
+ world.lock().unwrap().add_mailbox(
+ HOUSEHOLD,
+ json!({"id": "MB-travel", "name": "Travel", "role": null,
+ "myRights": full_rights()}),
+ );
+
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+
+ assert!(counts.mailboxes, "{counts:?}");
+ assert_eq!(store.folders().unwrap(), ["Receipts", "School", "Travel"]);
+
+ // The rebuilt layout classifies against the new mailbox.
+ world.lock().unwrap().add_email(
+ HOUSEHOLD,
+ email_json(
+ "E-trip",
+ "T-trip",
+ ("Airline", "no-reply@airline.co"),
+ "household",
+ &["MB-in", "MB-travel"],
+ "Itinerary",
+ Some("Seats 14A and 14B."),
+ "Seats 14A",
+ ),
+ );
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+ assert_eq!(counts.imported, 1, "{counts:?}");
+ let trip = the(&store, "E-trip@chislan.family").unwrap();
+ assert_eq!(trip.values().unwrap().folder.as_deref(), Some("Travel"));
+}
+
+#[tokio::test]
+async fn many_changes_page_through() {
+ let (world, client, household, mut sync, store, _dir) = household_ready().await;
+ for n in 1..=60 {
+ let id = format!("E-many{n:02}");
+ world.lock().unwrap().add_email(
+ HOUSEHOLD,
+ email_json(
+ &id,
+ "T-many",
+ ("Payer", "billing@payer.co"),
+ "household",
+ &["MB-in"],
+ "Statement",
+ Some("Balance due."),
+ "Balance",
+ ),
+ );
+ }
+
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+
+ assert_eq!(counts.imported, 60, "{counts:?}");
+ assert_eq!(store.messages(Filter::Search("")).unwrap().len(), 65);
+}