jmap: push queued filings via Email/set
Routes write only the store, so the UI never blocks on the network and dev mode keeps working; the poll drains the queue on its 30s cycle. Assisted-by: GLM-5.3 via pi
diff --git a/migrations/0003_pending_files.sql b/migrations/0003_pending_files.sql
new file mode 100644
index 0000000..e04f357
--- /dev/null
+++ b/migrations/0003_pending_files.sql
@@ -0,0 +1,9 @@
+-- Filing is intent, recorded here and pushed by the poll loop via
+-- Email/set (task rn): the routes only write the database, so the UI
+-- never blocks on the network. One row per message — the latest
+-- filing wins.
+CREATE TABLE pending_files (
+ message INTEGER PRIMARY KEY REFERENCES messages (id) ON DELETE CASCADE,
+ -- NULL unfiles.
+ folder TEXT REFERENCES folders (name)
+) STRICT;
diff --git a/src/jmap.rs b/src/jmap.rs
index b6f48f5..4070677 100644
--- a/src/jmap.rs
+++ b/src/jmap.rs
@@ -6,8 +6,9 @@
//!
//! 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.
+//! states the last sync left, and drains queued filings (task rn) with
+//! `Email/set`. Push replaces the timer later (task nwylszul); the
+//! changes application stays.
use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
@@ -23,7 +24,7 @@ use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::Error;
-use crate::model::{Account, State, User};
+use crate::model::{Account, MessageId, State, User};
use crate::store::{Incoming, IncomingKind, Store};
/// RFC 8620 §9. The capability URNs Docket looks for.
@@ -436,8 +437,8 @@ struct Layout {
labels: BTreeMap<String, State>,
/// Mailbox ids in the Docket namespace: the labels and the parent.
docket: BTreeSet<String>,
- /// Folder mailbox id → name.
- folders: BTreeMap<String, String>,
+ /// Folder mailbox id → its name and ACLs.
+ folders: BTreeMap<String, Folder>,
}
impl Layout {
@@ -468,9 +469,13 @@ impl Layout {
layout.labels.insert(mailbox.id.clone(), state);
}
} else {
- layout
- .folders
- .insert(mailbox.id.clone(), mailbox.name.clone());
+ layout.folders.insert(
+ mailbox.id.clone(),
+ Folder {
+ name: mailbox.name.clone(),
+ rights: mailbox.my_rights,
+ },
+ );
}
}
}
@@ -493,7 +498,7 @@ impl Layout {
email
.mailbox_ids
.keys()
- .find_map(|id| self.folders.get(id).cloned())
+ .find_map(|id| self.folders.get(id).map(|f| f.name.clone()))
}
/// True when the message sits in the Sent mailbox.
@@ -513,8 +518,76 @@ impl Layout {
/// Folder names in the session's order, for the folders table.
pub fn folder_names(&self) -> Vec<String> {
- self.folders.values().cloned().collect()
+ self.folders.values().map(|f| f.name.clone()).collect()
}
+
+ /// The mailbox id a folder name files into.
+ fn folder_id(&self, name: &str) -> Option<&str> {
+ self.folders
+ .iter()
+ .find(|(_, f)| f.name == name)
+ .map(|(id, _)| id.as_str())
+ }
+}
+
+/// A role-less mailbox outside the Docket namespace: a folder, with
+/// the ACLs that gate moving mail into and out of it.
+#[derive(Debug, Clone)]
+struct Folder {
+ name: String,
+ rights: MailboxRights,
+}
+
+/// The memberships a filing leaves the message in: out of every
+/// folder, into the chosen one, with system boxes and `Docket/`
+/// labels untouched (task rn's done-when).
+fn filed_mailboxes(
+ current: &BTreeMap<String, bool>,
+ layout: &Layout,
+ target: Option<&str>,
+) -> BTreeMap<String, bool> {
+ let mut next: BTreeMap<String, bool> = current
+ .iter()
+ .filter(|(id, member)| **member && !layout.folders.contains_key(*id))
+ .map(|(id, _)| (id.clone(), true))
+ .collect();
+ if let Some(name) = target
+ && let Some(id) = layout.folder_id(name)
+ {
+ next.insert(id.to_owned(), true);
+ }
+ next
+}
+
+/// Whether the ACLs permit the filing: the target takes items, and
+/// every folder losing the message gives them up.
+fn filing_allowed(current: &BTreeMap<String, bool>, layout: &Layout, target: Option<&str>) -> bool {
+ let leaving = current
+ .iter()
+ .filter(|(id, member)| **member && layout.folders.contains_key(*id))
+ .all(|(id, _)| {
+ layout
+ .folders
+ .get(id)
+ .is_some_and(|f| f.rights.may_remove_items)
+ });
+ let entering = target
+ .and_then(|name| layout.folder_id(name))
+ .is_none_or(|id| {
+ layout
+ .folders
+ .get(id)
+ .is_some_and(|f| f.rights.may_add_items)
+ });
+ leaving && entering
+}
+
+/// Parses an `Email/set` reply's verdict on each id.
+fn parse_set_emails(args: Value) -> Result<SetEmailsReply, JmapError> {
+ serde_json::from_value(args).map_err(|source| JmapError::Malformed {
+ what: "Email/set",
+ source,
+ })
}
/// How much mail one sync brought in.
@@ -544,6 +617,8 @@ pub struct PollCounts {
pub destroyed: usize,
/// Whether the layout was rebuilt off `Mailbox/changes`.
pub mailboxes: bool,
+ /// Filings pushed server-side.
+ pub filed: usize,
/// Whether the server disowned the stored states and a full
/// re-import ran instead.
pub resynced: bool,
@@ -717,6 +792,33 @@ struct GetChanges<'a> {
/// One page of request and response; small enough to keep bodies cheap.
const PAGE: usize = 50;
+/// `Email/set` arguments (RFC 8621 §5): one patch per email.
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct SetEmails<'a> {
+ account_id: &'a str,
+ update: &'a BTreeMap<String, EmailPatch>,
+}
+
+/// One email's patch: setting `mailboxIds` replaces the whole
+/// membership set.
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct EmailPatch {
+ mailbox_ids: BTreeMap<String, bool>,
+}
+
+/// The outcome half of an `Email/set` reply (RFC 8621 §5.3): which ids
+/// took their patches, and which the server refused.
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct SetEmailsReply {
+ #[serde(default)]
+ updated: BTreeMap<String, Option<Value>>,
+ #[serde(default)]
+ not_updated: BTreeMap<String, Value>,
+}
+
/// Opens sessions: one per credential, against Fastmail unless a test
/// points the session URL at a stub.
#[derive(Clone)]
@@ -863,6 +965,26 @@ impl Client {
parse_changes(method, args)
}
+ /// Applies membership patches with one `Email/set`, returning the
+ /// server's verdict on each id.
+ async fn set_emails(
+ &self,
+ api_url: &str,
+ token: &str,
+ account_id: &str,
+ update: &BTreeMap<String, EmailPatch>,
+ ) -> Result<SetEmailsReply, JmapError> {
+ let args = self
+ .call(
+ api_url,
+ token,
+ "Email/set",
+ &SetEmails { account_id, update },
+ )
+ .await?;
+ parse_set_emails(args)
+ }
+
/// Opens the credential's session, records its account, and imports
/// its mail. What the server permits is the policy (ADR 1), and
/// [`Import::account`] is the seam; the upserts underneath make this
@@ -1050,10 +1172,11 @@ impl Client {
}
/// 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.
+ /// states the account was left in, applying what moved, then the
+ /// queued filings go out. 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,
@@ -1109,6 +1232,10 @@ impl Client {
counts.mailboxes = true;
}
+ counts.filed = self
+ .push_files(&session, &token, id, &credential.name, &sync.layout, store)
+ .await?;
+
loop {
let page = match self
.changes(
@@ -1175,6 +1302,86 @@ impl Client {
}
Ok(counts)
}
+
+ /// Drains the account's queued filings (task rn): each message's
+ /// memberships move it out of whatever folders hold it into the
+ /// chosen one, leaving system boxes and `Docket/` labels untouched.
+ /// Mailbox ACLs decide — entries the rights refuse stay queued and
+ /// nag the log until an admin grants them; ids the server refused
+ /// or no longer carries are dropped with a warning instead of
+ /// retried forever.
+ async fn push_files(
+ &self,
+ session: &Session,
+ token: &str,
+ account_id: &str,
+ slug: &str,
+ layout: &Layout,
+ store: &Store,
+ ) -> Result<usize, Error> {
+ let pending = store.pending_files(slug)?;
+ if pending.is_empty() {
+ return Ok(0);
+ }
+ // Memberships come fresh from the server: the store keeps only
+ // the folder they derive from.
+ let mut current: BTreeMap<String, BTreeMap<String, bool>> = BTreeMap::new();
+ for chunk in pending.chunks(PAGE) {
+ let ids: Vec<String> = chunk.iter().map(|p| p.jmap_id.clone()).collect();
+ let (emails, _) = self
+ .emails(&session.api_url, token, account_id, &ids)
+ .await?;
+ for email in emails {
+ current.insert(email.id.clone(), email.mailbox_ids);
+ }
+ }
+ let mut updates: BTreeMap<String, EmailPatch> = BTreeMap::new();
+ let mut clear: Vec<MessageId> = Vec::new();
+ for p in &pending {
+ let Some(memberships) = current.get(&p.jmap_id) else {
+ tracing::warn!(jmap_id = %p.jmap_id, "filing dropped: the mail is gone server-side");
+ clear.push(p.message);
+ continue;
+ };
+ if p.folder
+ .as_deref()
+ .is_some_and(|f| layout.folder_id(f).is_none())
+ {
+ tracing::warn!(?p.folder, "filing dropped: the folder vanished server-side");
+ clear.push(p.message);
+ continue;
+ }
+ if !filing_allowed(memberships, layout, p.folder.as_deref()) {
+ tracing::warn!(?p.folder, "filing held: mailbox ACLs refuse it");
+ continue;
+ }
+ updates.insert(
+ p.jmap_id.clone(),
+ EmailPatch {
+ mailbox_ids: filed_mailboxes(memberships, layout, p.folder.as_deref()),
+ },
+ );
+ }
+ let mut filed = 0usize;
+ let entries: Vec<(String, EmailPatch)> = updates.into_iter().collect();
+ for chunk in entries.chunks(PAGE) {
+ let update: BTreeMap<String, EmailPatch> = chunk.iter().cloned().collect();
+ let reply = self
+ .set_emails(&session.api_url, token, account_id, &update)
+ .await?;
+ for (id, error) in &reply.not_updated {
+ tracing::warn!(jmap_id = %id, %error, "the server refused a filing");
+ }
+ filed = filed.saturating_add(reply.updated.len());
+ // Refused ids clear too: the server's verdict stands, and
+ // retrying would only requeue the refusal.
+ clear.extend(chunk.iter().filter_map(|(id, _)| {
+ pending.iter().find(|p| &p.jmap_id == id).map(|p| p.message)
+ }));
+ }
+ store.clear_pending_files(&clear)?;
+ Ok(filed)
+ }
}
/// The token stays in its file (DESIGN.md, Storage), read fresh each
@@ -1594,6 +1801,128 @@ mod tests {
.collect()
}
+ #[test]
+ fn filings_move_folders_and_leave_labels_alone() {
+ let layout = Layout::of(&mailboxes());
+ let current = BTreeMap::from([
+ ("M-in".to_string(), true),
+ ("M-watch".to_string(), true),
+ ("M-school".to_string(), true),
+ ]);
+ let filed = filed_mailboxes(¤t, &layout, Some("Receipts"));
+ assert_eq!(
+ filed,
+ BTreeMap::from([
+ ("M-in".to_string(), true),
+ ("M-watch".to_string(), true),
+ ("M-receipts".to_string(), true),
+ ])
+ );
+
+ // Unfiling drops every folder membership, nothing else.
+ let unfiled = filed_mailboxes(¤t, &layout, None);
+ assert_eq!(
+ unfiled,
+ BTreeMap::from([("M-in".to_string(), true), ("M-watch".to_string(), true),])
+ );
+
+ // A folder the layout lost — renamed or removed server-side —
+ // files nothing, but the rest still comes through untouched.
+ let vanished = filed_mailboxes(¤t, &layout, Some("Gone"));
+ assert_eq!(
+ vanished,
+ BTreeMap::from([("M-in".to_string(), true), ("M-watch".to_string(), true),])
+ );
+ }
+
+ #[test]
+ fn mailbox_acls_decide_filings() {
+ let full = rights(true, true, true, true, true, true, true, true);
+ let in_school =
+ BTreeMap::from([("M-in".to_string(), true), ("M-school".to_string(), true)]);
+ let layout_of = |receipts: Value| {
+ Layout::of(
+ &[
+ mailbox("M-in", "Inbox", Some("inbox"), full.clone()),
+ mailbox("M-school", "School", None, full.clone()),
+ mailbox("M-receipts", "Receipts", None, receipts),
+ ]
+ .into_iter()
+ .map(|m| serde_json::from_value::<Mailbox>(m).unwrap())
+ .collect::<Vec<_>>(),
+ )
+ };
+
+ let allowed = layout_of(full.clone());
+ assert!(filing_allowed(&in_school, &allowed, Some("Receipts")));
+ assert!(filing_allowed(&in_school, &allowed, None));
+
+ // The target can't take items.
+ let shut = rights(true, false, true, true, true, true, true, true);
+ assert!(!filing_allowed(
+ &in_school,
+ &layout_of(shut),
+ Some("Receipts")
+ ));
+
+ // The folder losing the message can't give it up.
+ let locked = BTreeMap::from([("M-in".to_string(), true), ("M-school".to_string(), true)]);
+ let hoarder = Layout::of(
+ &[
+ mailbox("M-in", "Inbox", Some("inbox"), full.clone()),
+ mailbox("M-school", "School", None, {
+ let mut r = full.clone();
+ r["mayRemoveItems"] = json!(false);
+ r
+ }),
+ mailbox("M-receipts", "Receipts", None, full),
+ ]
+ .into_iter()
+ .map(|m| serde_json::from_value::<Mailbox>(m).unwrap())
+ .collect::<Vec<_>>(),
+ );
+ assert!(!filing_allowed(&locked, &hoarder, Some("Receipts")));
+ }
+
+ #[test]
+ fn set_emails_serializes_and_replies_parse() {
+ let update = BTreeMap::from([(
+ "E1".to_string(),
+ EmailPatch {
+ mailbox_ids: BTreeMap::from([("M1".to_string(), true)]),
+ },
+ )]);
+ let args = serde_json::to_value(SetEmails {
+ account_id: "a",
+ update: &update,
+ })
+ .unwrap();
+ assert_eq!(
+ args,
+ json!({"accountId": "a", "update": {"E1": {"mailboxIds": {"M1": true}}}})
+ );
+
+ let reply = serde_json::from_value::<SetEmailsReply>(json!({
+ "accountId": "a", "oldState": "s1", "newState": "s2",
+ "updated": {"E1": null},
+ "notUpdated": {"E2": {"type": "tooManyRequests"}},
+ }))
+ .unwrap();
+ assert!(reply.updated.contains_key("E1"));
+ assert_eq!(reply.not_updated.len(), 1);
+
+ // Both maps may be absent (RFC 8621 §5.3).
+ let bare = serde_json::from_value::<SetEmailsReply>(json!({"accountId": "a"})).unwrap();
+ assert!(bare.updated.is_empty());
+ assert!(bare.not_updated.is_empty());
+ }
+
+ #[test]
+ fn a_set_email_reply_that_wont_parse_fails() {
+ let err = parse_set_emails(json!({"updated": 3})).unwrap_err();
+ assert!(err.to_string().contains("malformed Email/set"), "{err}");
+ }
+
#[test]
fn layout_sorts_mailboxes_into_roles_labels_and_folders() {
let layout = Layout::of(&mailboxes());
diff --git a/src/store.rs b/src/store.rs
index cfb32f8..28ccec0 100644
--- a/src/store.rs
+++ b/src/store.rs
@@ -22,6 +22,7 @@ use crate::model::{
const MIGRATIONS: &[&str] = &[
include_str!("../migrations/0001_init.sql"),
include_str!("../migrations/0002_jmap_import.sql"),
+ include_str!("../migrations/0003_pending_files.sql"),
];
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -352,7 +353,7 @@ impl Store {
if read_only {
return Err(Error::Forbidden("this account is read-only"));
}
- match folder {
+ let text = match folder {
Some(f) if !folder_exists(&tx, &f)? => {
return Err(Error::NotFound("folder"));
}
@@ -365,7 +366,9 @@ impl Store {
next.folder = None;
"Removed from folder".to_owned()
}
- }
+ };
+ queue_file(&tx, id, next.folder.as_deref())?;
+ text
}
Change::ToggleAssignee(login) => {
let name = load_user(&tx, &login)?.ok_or(Error::NotFound("user"))?.slug;
@@ -400,6 +403,14 @@ impl Store {
.cloned()
.ok_or(Error::BadRequest("nothing to undo"))?;
let tx = inner.conn.transaction()?;
+ // An undone filing reverts server-side too: the queued intent
+ // points back at the folder the undo restores.
+ let msg = load_message(&tx, id)?.ok_or(Error::NotFound("message"))?;
+ if let Kind::Received { values, .. } = &msg.kind
+ && values.folder != prev.folder
+ {
+ queue_file(&tx, id, prev.folder.as_deref())?;
+ }
write_values(&tx, id, &prev)?;
insert_event(&tx, id, user, self.now(), "Undone")?;
tx.commit()?;
@@ -414,6 +425,42 @@ impl Store {
Ok(())
}
+ /// The account's queued filings, for the poll loop to push (task
+ /// rn). The latest intent per message is the only row there is.
+ pub fn pending_files(&self, account: &str) -> Result<Vec<PendingFile>, Error> {
+ let inner = self.lock();
+ let mut stmt = inner.conn.prepare_cached(
+ "SELECT p.message, m.jmap_id, p.folder
+ FROM pending_files p JOIN messages m ON m.id = p.message
+ WHERE m.account = ?1 ORDER BY p.message",
+ )?;
+ let pending = stmt
+ .query_map([account], |r| {
+ Ok(PendingFile {
+ message: r.get(0)?,
+ jmap_id: r.get(1)?,
+ folder: r.get(2)?,
+ })
+ })?
+ .collect::<Result<_, _>>()?;
+ Ok(pending)
+ }
+
+ /// Drops queued filings once the server has taken them — or
+ /// refused them, which no retry would fix.
+ pub fn clear_pending_files(&self, messages: &[MessageId]) -> Result<(), Error> {
+ if messages.is_empty() {
+ return Ok(());
+ }
+ let mut inner = self.lock();
+ let tx = inner.conn.transaction()?;
+ for id in messages {
+ tx.execute("DELETE FROM pending_files WHERE message = ?1", [id])?;
+ }
+ tx.commit()?;
+ Ok(())
+ }
+
pub fn take_flash(&self, user: &str) -> Option<Flash> {
self.lock().flash.remove(user)
}
@@ -476,6 +523,17 @@ pub enum IncomingKind {
Sent { by: Option<String>, to: Vec<String> },
}
+/// One queued filing, waiting for the poll loop to push it (task rn).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct PendingFile {
+ pub message: MessageId,
+ /// The JMAP Email id to move — cached on the message, read fresh at
+ /// drain time so a reimport's new id is the one that goes out.
+ pub jmap_id: String,
+ /// The folder to file into; None unfiles.
+ pub folder: Option<String>,
+}
+
/// Writes inside one transaction, from [`Store::import`].
pub struct Import<'a> {
tx: Transaction<'a>,
@@ -818,6 +876,19 @@ fn write_values(conn: &Connection, id: MessageId, values: &Values) -> Result<(),
Ok(())
}
+/// Records a filing for the poll loop to push (task rn). Fixture mail
+/// has no server side to move, so rows without a cached JMAP id stay
+/// out.
+fn queue_file(conn: &Connection, id: MessageId, folder: Option<&str>) -> Result<(), Error> {
+ conn.execute(
+ "INSERT INTO pending_files (message, folder)
+ SELECT ?1, ?2 FROM messages WHERE id = ?1 AND jmap_id IS NOT NULL
+ ON CONFLICT (message) DO UPDATE SET folder = excluded.folder",
+ params![id, folder],
+ )?;
+ Ok(())
+}
+
fn insert_event(
conn: &Connection,
message: MessageId,
@@ -928,6 +999,96 @@ mod tests {
);
}
+ #[test]
+ fn filing_queues_intent_for_the_poll_loop() {
+ 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.folder("House")?;
+ tx.incoming("household", &incoming(State::Inbox, None, "$1,840"))
+ })
+ .unwrap();
+ let id = store
+ .messages(Filter::Search("1,840"))
+ .unwrap()
+ .first()
+ .unwrap()
+ .id;
+ store.sign_in(SAM, "Sam").unwrap();
+
+ // Filing queues the intent, and the latest one wins.
+ store
+ .edit(SAM, id, Change::Folder(Some("House".into())))
+ .unwrap();
+ store.edit(SAM, id, Change::Folder(None)).unwrap();
+ assert_eq!(
+ store.pending_files("household").unwrap(),
+ [PendingFile {
+ message: id,
+ jmap_id: "E9".into(),
+ folder: None,
+ }]
+ );
+
+ // Another account's drain never sees them.
+ assert!(store.pending_files("eli").unwrap().is_empty());
+
+ // What the poll loop pushed, it clears.
+ store.clear_pending_files(&[id]).unwrap();
+ assert!(store.pending_files("household").unwrap().is_empty());
+ store.clear_pending_files(&[]).unwrap();
+
+ // Undoing a filing re-points the queue at the restored folder.
+ store
+ .edit(SAM, id, Change::Folder(Some("House".into())))
+ .unwrap();
+ assert_eq!(
+ store.pending_files("household").unwrap()[0].folder,
+ Some("House".into())
+ );
+ store.undo(SAM).unwrap();
+ assert_eq!(store.pending_files("household").unwrap()[0].folder, None);
+
+ // A filing that can't queue fails the whole edit, and the
+ // folder it would have set rolls back.
+ exec(
+ &store,
+ "CREATE TRIGGER no_pending BEFORE INSERT ON pending_files BEGIN SELECT RAISE(ABORT, 'no'); END;",
+ );
+ assert!(matches!(
+ store.edit(SAM, id, Change::Folder(Some("House".into()))),
+ Err(Error::Db(_))
+ ));
+ exec(&store, "DROP TRIGGER no_pending;");
+ assert_eq!(values(&store, id).folder, None);
+ }
+
+ #[test]
+ fn a_gone_pending_table_fails_the_drain() {
+ // A fresh store, so the query compiles against the missing
+ // table rather than replaying a cached statement.
+ let store = fixtures::store().unwrap();
+ exec(&store, "DROP TABLE pending_files;");
+ assert!(store.pending_files("household").is_err());
+ }
+
+ #[test]
+ fn fixture_mail_never_queues_a_filing() {
+ let store = fixtures::store().unwrap();
+ store
+ .edit(SAM, 4, Change::Folder(Some("Finance".into())))
+ .unwrap();
+ // No server side to move, so no row: fixture mail has no cached
+ // JMAP id.
+ assert!(store.pending_files("household").unwrap().is_empty());
+ }
+
#[test]
fn folders() {
let store = fixtures::store().unwrap();
diff --git a/tests/jmap.rs b/tests/jmap.rs
index 52e5ba4..7cd52c0 100644
--- a/tests/jmap.rs
+++ b/tests/jmap.rs
@@ -14,7 +14,7 @@ use axum::routing::{get, post};
use axum::{Json, Router};
use docket::jmap::{CORE, Client, Credential, MAIL, SUBMISSION};
use docket::model::{Kind, State};
-use docket::store::{Clock, Filter, Store};
+use docket::store::{Change, Clock, Filter, Store};
use serde_json::{Value, json};
use tokio::net::TcpListener;
@@ -307,6 +307,8 @@ fn threads_of(account: &str) -> Vec<Value> {
#[derive(Default)]
struct Stub {
accounts: BTreeMap<&'static str, AccountMail>,
+ /// Email ids `Email/set` refuses, to exercise the notUpdated path.
+ refuse: BTreeSet<String>,
}
#[derive(Default)]
@@ -398,6 +400,25 @@ impl Stub {
.push(mailbox);
self.snapshot(account);
}
+
+ /// Rewrites a mailbox's ACLs the way an admin would.
+ fn update_mailbox(&mut self, account: &'static str, id: &str, change: impl FnOnce(&mut Value)) {
+ let mail = self.accounts.get_mut(account).unwrap();
+ let mailbox = mail
+ .mailboxes
+ .iter_mut()
+ .find(|m| m["id"] == json!(id))
+ .unwrap();
+ change(mailbox);
+ self.snapshot(account);
+ }
+
+ /// Deletes a mailbox, the way a server does when a folder goes away.
+ fn remove_mailbox(&mut self, account: &'static str, id: &str) {
+ let mail = self.accounts.get_mut(account).unwrap();
+ mail.mailboxes.retain(|m| m["id"] != json!(id));
+ self.snapshot(account);
+ }
}
/// The `/changes` reply for one type: the diff from the named state to
@@ -465,6 +486,37 @@ async fn api(world: axum::extract::State<World>, headers: HeaderMap, body: Strin
let method = call[0].as_str().expect("a method name").to_owned();
let args = &call[1];
+ // The one mutating method, so it locks for itself and answers early.
+ if method == "Email/set" {
+ let mut stub = world.lock().unwrap();
+ let mut updated = serde_json::Map::new();
+ let mut not_updated = serde_json::Map::new();
+ for (id, patch) in args["update"].as_object().expect("an update map") {
+ if stub.refuse.contains(id) {
+ not_updated.insert(
+ id.clone(),
+ json!({"type": "serverFail", "description": "refused"}),
+ );
+ continue;
+ }
+ let mail = stub.accounts.get_mut(account).unwrap();
+ match mail.emails.iter_mut().find(|e| e["id"] == json!(id)) {
+ // A patch's mailboxIds replaces the whole membership set.
+ Some(email) => {
+ email["mailboxIds"] = patch["mailboxIds"].clone();
+ updated.insert(id.clone(), Value::Null);
+ }
+ None => {
+ not_updated.insert(id.clone(), json!({"type": "notFound"}));
+ }
+ }
+ }
+ stub.snapshot(account);
+ let reply = json!({"accountId": account, "oldState": "s", "newState": "s",
+ "updated": updated, "notUpdated": not_updated});
+ return Json(json!({"methodResponses": [["Email/set", reply, "0"]]})).into_response();
+ }
+
let stub = world.lock().unwrap();
let mail = stub.accounts.get(account).unwrap();
let reply = match method.as_str() {
@@ -1094,3 +1146,203 @@ async fn many_changes_page_through() {
assert_eq!(counts.imported, 60, "{counts:?}");
assert_eq!(store.messages(Filter::Search("")).unwrap().len(), 65);
}
+
+/// The school message, which sits in Inbox plus a Docket label plus a
+/// folder — the shape filing has to move without disturbing the rest.
+fn school_message(store: &Store) -> docket::model::Message {
+ store
+ .messages(Filter::Search("Buses now return"))
+ .unwrap()
+ .into_iter()
+ .next()
+ .unwrap()
+}
+
+#[tokio::test]
+async fn filing_pushes_through_the_poll() {
+ let (world, client, household, mut sync, store, _dir) = household_ready().await;
+ let school = school_message(&store);
+ store
+ .edit(
+ "sam@example.com",
+ school.id,
+ Change::Folder(Some("Receipts".into())),
+ )
+ .unwrap();
+ assert_eq!(store.pending_files("household").unwrap().len(), 1);
+
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+
+ assert_eq!(counts.filed, 1, "{counts:?}");
+
+ // Server side: out of School, into Receipts, with Inbox and the
+ // Docket label untouched.
+ let memberships = {
+ let world = world.lock().unwrap();
+ world.accounts[HOUSEHOLD]
+ .emails
+ .iter()
+ .find(|e| e["id"] == json!("E-school"))
+ .unwrap()["mailboxIds"]
+ .clone()
+ };
+ assert_eq!(
+ memberships,
+ json!({"MB-in": true, "MB-watch": true, "MB-receipts": true})
+ );
+
+ // The push cleared the queue, and the same cycle's /changes sweep
+ // confirmed the filing in the cached row.
+ assert!(store.pending_files("household").unwrap().is_empty());
+ let after = store.message(school.id).unwrap().unwrap();
+ assert_eq!(after.values().unwrap().folder.as_deref(), Some("Receipts"));
+ assert_eq!(after.values().unwrap().state, State::Watch);
+
+ // Unfiling queues again and empties every folder membership.
+ store
+ .edit("sam@example.com", school.id, Change::Folder(None))
+ .unwrap();
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+ assert_eq!(counts.filed, 1, "{counts:?}");
+ let memberships = {
+ let world = world.lock().unwrap();
+ world.accounts[HOUSEHOLD]
+ .emails
+ .iter()
+ .find(|e| e["id"] == json!("E-school"))
+ .unwrap()["mailboxIds"]
+ .clone()
+ };
+ assert_eq!(memberships, json!({"MB-in": true, "MB-watch": true}));
+}
+
+#[tokio::test]
+async fn acl_denied_filings_wait_for_the_rights() {
+ let (world, client, household, mut sync, store, _dir) = household_ready().await;
+ // Receipts stops taking items before the filing reaches it; the
+ // poll rebuilds the layout off Mailbox/changes and holds the queue.
+ world
+ .lock()
+ .unwrap()
+ .update_mailbox(HOUSEHOLD, "MB-receipts", |m| {
+ m["myRights"]["mayAddItems"] = json!(false);
+ });
+ let school = school_message(&store);
+ store
+ .edit(
+ "sam@example.com",
+ school.id,
+ Change::Folder(Some("Receipts".into())),
+ )
+ .unwrap();
+
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+ assert_eq!(counts.filed, 0, "{counts:?}");
+ assert_eq!(store.pending_files("household").unwrap().len(), 1);
+
+ // The rights come back and the waiting filing goes through.
+ world
+ .lock()
+ .unwrap()
+ .update_mailbox(HOUSEHOLD, "MB-receipts", |m| {
+ m["myRights"]["mayAddItems"] = json!(true);
+ });
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+ assert_eq!(counts.filed, 1, "{counts:?}");
+ assert!(store.pending_files("household").unwrap().is_empty());
+}
+
+#[tokio::test]
+async fn a_refused_filing_clears_rather_than_retries() {
+ let (world, client, household, mut sync, store, _dir) = household_ready().await;
+ let school = school_message(&store);
+ store
+ .edit(
+ "sam@example.com",
+ school.id,
+ Change::Folder(Some("Receipts".into())),
+ )
+ .unwrap();
+ world.lock().unwrap().refuse.insert("E-school".into());
+
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+
+ // The server's verdict stands: cleared, not requeued, and the
+ // cached row keeps the user's view until the server says otherwise.
+ assert_eq!(counts.filed, 0, "{counts:?}");
+ assert!(store.pending_files("household").unwrap().is_empty());
+ assert_eq!(
+ store
+ .message(school.id)
+ .unwrap()
+ .unwrap()
+ .values()
+ .unwrap()
+ .folder,
+ Some("Receipts".into())
+ );
+}
+
+#[tokio::test]
+async fn a_filing_for_mail_thats_gone_drops() {
+ let (world, client, household, mut sync, store, _dir) = household_ready().await;
+ let school = school_message(&store);
+ store
+ .edit(
+ "sam@example.com",
+ school.id,
+ Change::Folder(Some("Receipts".into())),
+ )
+ .unwrap();
+ world.lock().unwrap().remove_email(HOUSEHOLD, "E-school");
+
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+
+ assert_eq!(counts.filed, 0, "{counts:?}");
+ assert!(store.pending_files("household").unwrap().is_empty());
+}
+
+#[tokio::test]
+async fn a_filing_into_a_vanished_folder_drops() {
+ let (world, client, household, mut sync, store, _dir) = household_ready().await;
+ let school = school_message(&store);
+ store
+ .edit(
+ "sam@example.com",
+ school.id,
+ Change::Folder(Some("Receipts".into())),
+ )
+ .unwrap();
+ // The folder's mailbox disappears before the push lands: no id to
+ // file into, and nothing to retry.
+ world
+ .lock()
+ .unwrap()
+ .remove_mailbox(HOUSEHOLD, "MB-receipts");
+
+ let counts = client
+ .poll_once(&household, &mut sync, &store)
+ .await
+ .unwrap();
+
+ assert_eq!(counts.filed, 0, "{counts:?}");
+ assert!(store.pending_files("household").unwrap().is_empty());
+}