diff --git a/crates/nuncio-cli/src/args.rs b/crates/nuncio-cli/src/args.rs index b4059af..8e8bdf4 100644 --- a/crates/nuncio-cli/src/args.rs +++ b/crates/nuncio-cli/src/args.rs @@ -250,6 +250,26 @@ pub enum AccountSubcommand { #[arg(short, long, help = "Unique account identifier")] id: String, }, + /// Turn the filter engine on or off for one account. + Filters { + /// Unique account identifier. + #[arg(short, long, help = "Unique account identifier")] + id: String, + /// Run filters against this account's incoming mail. + #[arg( + long, + help = "Enable filters for this account", + conflicts_with = "disable" + )] + enable: bool, + /// Stop running filters against this account's incoming mail. + #[arg( + long, + help = "Disable filters for this account", + conflicts_with = "enable" + )] + disable: bool, + }, /// Edit an existing account profile configuration. Edit { /// Unique account identifier. @@ -343,6 +363,66 @@ pub enum MailSubcommand { #[arg(long, help = "Mark the message as unread", conflicts_with = "read")] unread: bool, }, + /// Move one occupancy of a message into another folder. + Move { + /// Unique message identifier. + #[arg(short, long, help = "Unique message identifier")] + id: String, + /// Destination folder identifier. + #[arg(short, long, help = "Destination folder identifier")] + to: String, + /// Which mailbox occupancy to act on, as + /// `account/folder/uidvalidity/uid`. Required when the message occupies + /// more than one mailbox; `mail read` lists them. + #[arg(long, help = "Occupancy as account/folder/uidvalidity/uid")] + placement: Option, + }, + /// Delete one occupancy of a message. Other occupancies survive. + Delete { + /// Unique message identifier. + #[arg(short, long, help = "Unique message identifier")] + id: String, + /// Which mailbox occupancy to act on, as + /// `account/folder/uidvalidity/uid`. + #[arg(long, help = "Occupancy as account/folder/uidvalidity/uid")] + placement: Option, + }, + /// Flag one occupancy of a message. + Flag { + /// Unique message identifier. + #[arg(short, long, help = "Unique message identifier")] + id: String, + /// Which mailbox occupancy to act on, as + /// `account/folder/uidvalidity/uid`. + #[arg(long, help = "Occupancy as account/folder/uidvalidity/uid")] + placement: Option, + }, + /// Remove the flag from one occupancy of a message. + Unflag { + /// Unique message identifier. + #[arg(short, long, help = "Unique message identifier")] + id: String, + /// Which mailbox occupancy to act on, as + /// `account/folder/uidvalidity/uid`. + #[arg(long, help = "Occupancy as account/folder/uidvalidity/uid")] + placement: Option, + }, + /// List mutations the server refused because another client changed the + /// message first. + Conflicts { + /// Also list conflicts that have already been resolved. + #[arg(long, help = "Include already-resolved conflicts")] + all: bool, + }, + /// Mark a conflict handled, recording how it was resolved. + Resolve { + /// Conflict identifier, as reported by `mail conflicts`. + #[arg(short, long, help = "Conflict identifier")] + id: String, + /// Note recording how the conflict was resolved. + #[arg(short, long, help = "How the conflict was resolved")] + note: String, + }, /// Export mailbox messages to a portable file format: MBOX, an EML /// zip archive, JSON, or JSON Lines. Export { diff --git a/crates/nuncio-cli/src/runner.rs b/crates/nuncio-cli/src/runner.rs index cf063c7..1d8ae7e 100644 --- a/crates/nuncio-cli/src/runner.rs +++ b/crates/nuncio-cli/src/runner.rs @@ -425,6 +425,14 @@ impl HeadlessRunner { .await } AccountSubcommand::Show { id } => self.handle_account_show(id, json_mode).await, + AccountSubcommand::Filters { + id, + enable, + disable, + } => { + self.handle_account_filters(id, *enable, *disable, json_mode) + .await + } AccountSubcommand::Edit { id, email, @@ -468,6 +476,28 @@ impl HeadlessRunner { MailSubcommand::Mark { id, read, unread } => { self.handle_mark_read(id, *read, *unread, json_mode).await } + MailSubcommand::Move { id, to, placement } => { + self.handle_mail_mutation("move", id, placement.as_ref(), Some(to), json_mode) + .await + } + MailSubcommand::Delete { id, placement } => { + self.handle_mail_mutation("delete", id, placement.as_ref(), None, json_mode) + .await + } + MailSubcommand::Flag { id, placement } => { + self.handle_mail_mutation("flag", id, placement.as_ref(), None, json_mode) + .await + } + MailSubcommand::Unflag { id, placement } => { + self.handle_mail_mutation("unflag", id, placement.as_ref(), None, json_mode) + .await + } + MailSubcommand::Conflicts { all } => { + self.handle_list_conflicts(*all, json_mode).await + } + MailSubcommand::Resolve { id, note } => { + self.handle_resolve_conflict(id, note, json_mode).await + } MailSubcommand::Export { format, out, @@ -1079,6 +1109,245 @@ impl HeadlessRunner { } } + /// Parse an `account/folder/uidvalidity/uid` occupancy selector. + /// + /// All four coordinates are required and none may be empty: a partial + /// selector addresses nothing, and silently completing it would aim a + /// destructive mutation at a mailbox the user did not name. + fn parse_placement(spec: &str) -> Result { + let parts: Vec<&str> = spec.split('/').collect(); + match parts.as_slice() { + [account_id, folder_id, uidvalidity, uid] + if !account_id.is_empty() + && !folder_id.is_empty() + && !uidvalidity.is_empty() + && !uid.is_empty() => + { + Ok(nuncio_proto::v1::PlacementRef { + account_id: (*account_id).to_string(), + folder_id: (*folder_id).to_string(), + uidvalidity: (*uidvalidity).to_string(), + uid: (*uid).to_string(), + }) + } + _ => Err(format!( + "placement '{spec}' is not account/folder/uidvalidity/uid" + )), + } + } + + /// `mail move`/`delete`/`flag`/`unflag`: queue a placement-addressed remote + /// mutation via the daemon and report the outbox id it was queued as. + /// + /// The reported outcome is "queued", never "done": the daemon's outbox is + /// what actually talks to the server, and only it can say whether the + /// mutation applied, conflicted, or could not be verified. + async fn handle_mail_mutation( + &self, + verb: &str, + id: &str, + placement: Option<&String>, + destination: Option<&String>, + json_mode: bool, + ) -> String { + let placement = match placement.map(|p| Self::parse_placement(p)) { + Some(Ok(placement)) => Some(placement), + Some(Err(e)) => return Self::render_error(&e, json_mode), + None => None, + }; + + let mut client = match self.connect_mail_client().await { + Ok(client) => client, + Err(e) => return Self::render_error(&e, json_mode), + }; + + let queued = match verb { + "move" => { + let Some(to) = destination else { + return Self::render_error("a destination folder is required", json_mode); + }; + client + .move_message(nuncio_proto::v1::MoveMessageRequest { + message_id: id.to_string(), + placement, + destination_folder_id: to.clone(), + }) + .await + .map(|r| r.into_inner().mutation_id) + } + "delete" => client + .delete_message(nuncio_proto::v1::DeleteMessageRequest { + message_id: id.to_string(), + placement, + }) + .await + .map(|r| r.into_inner().mutation_id), + _ => client + .flag_message(nuncio_proto::v1::FlagMessageRequest { + message_id: id.to_string(), + placement, + flagged: verb == "flag", + }) + .await + .map(|r| r.into_inner().mutation_id), + }; + + match queued { + Ok(mutation_id) => { + if json_mode { + format_json(&json!({ + "status": "queued", + "operation": verb, + "id": id, + "mutation_id": mutation_id, + })) + } else { + format!("Queued {verb} of '{id}' as mutation '{mutation_id}'") + } + } + Err(status) => Self::render_status_error(verb, &status, json_mode), + } + } + + /// `mail conflicts`: list mutations the server refused because another + /// client changed the message first. + async fn handle_list_conflicts(&self, include_resolved: bool, json_mode: bool) -> String { + let mut client = match self.connect_mail_client().await { + Ok(client) => client, + Err(e) => return Self::render_error(&e, json_mode), + }; + match client + .list_conflicts(nuncio_proto::v1::ListConflictsRequest { + include_resolved, + limit: 100, + }) + .await + { + Ok(response) => { + let conflicts = response.into_inner().conflicts; + if json_mode { + let rows: Vec<_> = conflicts + .iter() + .map(|c| { + json!({ + "id": c.id, + "mutation_id": c.mutation_id, + "message_id": c.message_id, + "operation": c.mutation_type, + "observed": c.observed, + "resolved": c.resolved_at.is_some(), + }) + }) + .collect(); + format_json(&json!({ "conflicts": rows })) + } else if conflicts.is_empty() { + "No conflicts recorded".to_string() + } else { + let mut out = String::new(); + for c in &conflicts { + let state = if c.resolved_at.is_some() { + "resolved" + } else { + "open" + }; + out.push_str(&format!( + "{} [{}] {} on '{}': {}\n", + c.id, state, c.mutation_type, c.message_id, c.observed + )); + } + out.trim_end().to_string() + } + } + Err(status) => Self::render_status_error("list_conflicts", &status, json_mode), + } + } + + /// `mail resolve`: record a human's decision about a conflict. + async fn handle_resolve_conflict(&self, id: &str, note: &str, json_mode: bool) -> String { + let mut client = match self.connect_mail_client().await { + Ok(client) => client, + Err(e) => return Self::render_error(&e, json_mode), + }; + match client + .resolve_conflict(nuncio_proto::v1::ResolveConflictRequest { + conflict_id: id.to_string(), + resolution: note.to_string(), + }) + .await + { + Ok(_) => { + if json_mode { + format_json(&json!({ "status": "resolved", "id": id })) + } else { + format!("Conflict '{id}' resolved") + } + } + Err(status) => Self::render_status_error("resolve_conflict", &status, json_mode), + } + } + + /// `account filters`: flip one account's `filters_enabled` and persist it + /// through the daemon's `UpdateAccount` RPC. + /// + /// The account is read back first so the update carries its real current + /// configuration; sending a locally-invented config would silently reset + /// every field the CLI did not know about. + async fn handle_account_filters( + &self, + id: &str, + enable: bool, + disable: bool, + json_mode: bool, + ) -> String { + if enable == disable { + return Self::render_error( + "exactly one of --enable or --disable must be specified", + json_mode, + ); + } + + let mut client = match self.connect_accounts_client().await { + Ok(client) => client, + Err(e) => return Self::render_error(&e, json_mode), + }; + + let accounts = match client + .list_accounts(nuncio_proto::v1::ListAccountsRequest {}) + .await + { + Ok(response) => response.into_inner().accounts, + Err(status) => return Self::render_status_error("list_accounts", &status, json_mode), + }; + let Some(mut config) = accounts.into_iter().find(|a| a.id == id) else { + return Self::render_error(&format!("account '{id}' not found"), json_mode); + }; + config.filters_enabled = enable; + + match client + .update_account(nuncio_proto::v1::UpdateAccountRequest { + config: Some(config), + password: None, + }) + .await + { + Ok(_) => { + if json_mode { + format_json(&json!({ + "status": "updated", + "id": id, + "filters_enabled": enable, + })) + } else { + format!( + "Filters {} for account '{id}'", + if enable { "enabled" } else { "disabled" } + ) + } + } + Err(status) => Self::render_status_error("update_account", &status, json_mode), + } + } + /// `mail export`: a real thin gRPC client of the running `nunciod` /// daemon's `nuncio.v1.Export` API. The /// daemon writes the export to `out` on ITS OWN host filesystem (the @@ -1826,6 +2095,9 @@ impl HeadlessRunner { keyring_secret_key: keyring_key.clone(), sync_interval: Some(nuncio_proto::time::duration_from_secs(300)), transport: Some(transport), + // Filters take irreversible remote actions, so adding an account + // never opts it in; `account filters` does, deliberately. + filters_enabled: false, }; let mut client = match self.connect_accounts_client().await { @@ -3147,6 +3419,7 @@ mod tests { email_address: "stub@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-stub-1".to_string(), sync_interval: Some(nuncio_proto::time::duration_from_secs(300)), + filters_enabled: false, transport: Some(nuncio_proto::v1::account_config::Transport::ImapSmtp( nuncio_proto::v1::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), @@ -3371,6 +3644,13 @@ mod tests { body_plain: Some("Stub body text".to_string()), body_html: None, attachments: Vec::new(), + placements: vec![nuncio_proto::v1::Placement { + account_id: "acct-stub-1".to_string(), + folder_id: "inbox".to_string(), + uidvalidity: "1".to_string(), + uid: "42".to_string(), + read: false, + }], } } @@ -3383,10 +3663,77 @@ mod tests { last_mark_read: Arc>>, last_send_message: Arc>>, last_sync: Arc>>, + last_move: Arc>>, + last_delete: Arc>>, + last_flag: Arc>>, } #[tonic::async_trait] impl MailService for StubMail { + async fn move_message( + &self, + request: tonic::Request, + ) -> Result, tonic::Status> + { + *self.last_move.lock().unwrap_or_else(|e| e.into_inner()) = + Some(request.into_inner()); + Ok(tonic::Response::new( + nuncio_proto::v1::MoveMessageResponse { + mutation_id: "mut-stub-1".to_string(), + }, + )) + } + + async fn delete_message( + &self, + request: tonic::Request, + ) -> Result, tonic::Status> + { + *self.last_delete.lock().unwrap_or_else(|e| e.into_inner()) = + Some(request.into_inner()); + Ok(tonic::Response::new( + nuncio_proto::v1::DeleteMessageResponse { + mutation_id: "mut-stub-2".to_string(), + }, + )) + } + + async fn flag_message( + &self, + request: tonic::Request, + ) -> Result, tonic::Status> + { + *self.last_flag.lock().unwrap_or_else(|e| e.into_inner()) = + Some(request.into_inner()); + Ok(tonic::Response::new( + nuncio_proto::v1::FlagMessageResponse { + mutation_id: "mut-stub-3".to_string(), + }, + )) + } + + async fn list_conflicts( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> + { + Ok(tonic::Response::new( + nuncio_proto::v1::ListConflictsResponse { + conflicts: Vec::new(), + }, + )) + } + + async fn resolve_conflict( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> + { + Ok(tonic::Response::new( + nuncio_proto::v1::ResolveConflictResponse {}, + )) + } + async fn list_folders( &self, _request: tonic::Request, @@ -3717,6 +4064,46 @@ mod tests { #[tonic::async_trait] impl MailService for StubMailAlwaysErrors { + async fn move_message( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> + { + Err(tonic::Status::internal("mailbox index is corrupt")) + } + + async fn delete_message( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> + { + Err(tonic::Status::internal("mailbox index is corrupt")) + } + + async fn flag_message( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> + { + Err(tonic::Status::internal("mailbox index is corrupt")) + } + + async fn list_conflicts( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> + { + Err(tonic::Status::internal("mailbox index is corrupt")) + } + + async fn resolve_conflict( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> + { + Err(tonic::Status::internal("mailbox index is corrupt")) + } + async fn list_folders( &self, _request: tonic::Request, @@ -3820,6 +4207,52 @@ mod tests { #[tonic::async_trait] impl MailService for StubMailTypedError { + /// The one typed-error path this stub exists to exercise on the + /// mutation surface: a genuine, server-evidenced conflict. + async fn move_message( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> + { + Err(nuncio_proto::errors::status_with_metadata( + ErrorReason::Conflict, + "message was moved by another client", + [("message_id".to_string(), "msg-stub-1".to_string())], + )) + } + + async fn delete_message( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> + { + Err(tonic::Status::internal("unused in this test")) + } + + async fn flag_message( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> + { + Err(tonic::Status::internal("unused in this test")) + } + + async fn list_conflicts( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> + { + Err(tonic::Status::internal("unused in this test")) + } + + async fn resolve_conflict( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> + { + Err(tonic::Status::internal("unused in this test")) + } + async fn list_folders( &self, _request: tonic::Request, diff --git a/crates/nuncio-core/src/model.rs b/crates/nuncio-core/src/model.rs index 0df4a26..3cf92fe 100644 --- a/crates/nuncio-core/src/model.rs +++ b/crates/nuncio-core/src/model.rs @@ -183,6 +183,38 @@ pub struct PlacementKey { pub remote_id: String, } +/// A mutation the server refused because another client had already changed the +/// message, recorded as durable state rather than as a passing notification. +/// +/// A conflict is raised only on positive evidence from the server; an +/// unverifiable outcome is never one. Because the intent was formed against +/// state that no longer exists, a human has to decide what happens next, and +/// that decision may come long after the event. The event stream is a lossy +/// broadcast -- it drops on lag and carries no sequence number to resume from -- +/// so it can only ever be an optimisation over a queryable table, never the +/// record itself. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct MutationConflict { + /// Stable identifier of this conflict record. + pub id: String, + /// The outbox mutation that conflicted. + pub mutation_id: String, + /// Content-addressed key of the message the mutation targeted. + pub message_id: String, + /// The operation that was refused (`MOVE`, `DELETE`, `FLAG`, ...). + pub mutation_type: String, + /// The occupancy the mutation was aimed at, when the row recorded one. + pub placement: Option, + /// The server's positive evidence that another client changed the message. + pub observed: String, + /// Unix timestamp (seconds) when the conflict was detected. + pub detected_at: i64, + /// Unix timestamp (seconds) when a human resolved it, if they have. + pub resolved_at: Option, + /// Free-form note recording how the conflict was resolved. + pub resolution: Option, +} + impl Placement { /// The addressing coordinates of this occupancy, dropping the read flag. pub fn key(&self) -> PlacementKey { diff --git a/crates/nuncio-filter/src/outbox.rs b/crates/nuncio-filter/src/outbox.rs index bb9b191..ada4a71 100644 --- a/crates/nuncio-filter/src/outbox.rs +++ b/crates/nuncio-filter/src/outbox.rs @@ -1,6 +1,7 @@ //! Transactional Outbox Pattern for Remote IMAP/JMAP Mutations. use crate::ast::PendingRemoteMutation; +use nuncio_core::model::PlacementKey; use serde::{Deserialize, Serialize}; /// Payload details stored for pending remote mutation. @@ -10,6 +11,17 @@ pub struct MutationPayload { pub action_type: String, /// Target parameter (e.g. folder name, target email address). pub target: Option, + /// The mailbox occupancy this mutation was formed against. + /// + /// Message identity is content-addressed and folder-independent, so a single + /// message can occupy several mailboxes at once (Gmail labels make that + /// ordinary). `MOVE` and `DELETE` act on *one* occupancy, and acting on the + /// wrong one is destructive and irreversible, so the caller that formed the + /// intent records which occupancy it meant. `None` preserves the older + /// rows' meaning -- "unrecorded" -- which the executor may resolve only when + /// the message occupies exactly one mailbox. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub placement: Option, } /// Transactional outbox helper managing retry backoffs and mutation payloads. @@ -31,17 +43,33 @@ impl OutboxManager { base + jitter } - /// Construct a new `PendingRemoteMutation` record. + /// Construct a new `PendingRemoteMutation` record whose target occupancy is + /// left unrecorded. The executor may still address it when the message + /// occupies exactly one mailbox, and refuses when it occupies several. pub fn create_mutation( rule_id: impl Into, message_id: impl Into, action_type: impl Into, target: Option, + ) -> PendingRemoteMutation { + Self::create_mutation_at(rule_id, message_id, action_type, target, None) + } + + /// Construct a new `PendingRemoteMutation` record that names the mailbox + /// occupancy it was formed against, so a message occupying several mailboxes + /// is still unambiguously addressable. + pub fn create_mutation_at( + rule_id: impl Into, + message_id: impl Into, + action_type: impl Into, + target: Option, + placement: Option, ) -> PendingRemoteMutation { let action_str = action_type.into(); let payload_struct = MutationPayload { action_type: action_str.clone(), target, + placement, }; let payload_json = serde_json::to_string(&payload_struct).unwrap_or_default(); let now = chrono::Utc::now().timestamp(); diff --git a/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin b/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin index d555779..94f9059 100644 Binary files a/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin and b/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin differ diff --git a/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto b/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto index 2abaec2..bba6752 100644 --- a/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto +++ b/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto @@ -45,6 +45,14 @@ enum ErrorReason { ERROR_REASON_UNSUPPORTED = 11; // An unexpected internal error the client cannot act on specifically. ERROR_REASON_INTERNAL = 12; + // Another client changed the target first and the server said so. + // + // Raised only on positive evidence; an outcome the server left unverifiable + // is never reported as a conflict. Retrying is not the remedy -- the intent + // was formed against state that no longer exists, so re-issuing it would + // either do nothing or overwrite a change the caller never saw. The conflict + // is also recorded durably and is retrievable via `Mail.ListConflicts`. + ERROR_REASON_CONFLICT = 13; } // ErrorInfo is the structured error payload the daemon attaches to a failed @@ -378,6 +386,11 @@ message AccountConfig { JmapTransport jmap = 16; DavTransport dav = 17; } + // Whether the filter engine runs against this account's incoming mail. + // + // Defaults to false: filters take irreversible remote actions, so an account + // does not get them by being added, only by being told to. + bool filters_enabled = 18; } // AddAccountRequest carries the full account configuration plus the raw @@ -478,6 +491,24 @@ service Mail { rpc GetMessage(GetMessageRequest) returns (GetMessageResponse); // MarkRead updates a message's read/unread flag. rpc MarkRead(MarkReadRequest) returns (MarkReadResponse); + // MoveMessage enqueues a verified remote move of one occupancy of a message + // into another folder. Returns as soon as the intent is durably queued; the + // outbox executes it against the server and only records success on positive + // proof. + rpc MoveMessage(MoveMessageRequest) returns (MoveMessageResponse); + // DeleteMessage enqueues a verified remote delete of one occupancy of a + // message. Other occupancies of the same message are untouched. + rpc DeleteMessage(DeleteMessageRequest) returns (DeleteMessageResponse); + // FlagMessage enqueues a verified remote flag/unflag of one occupancy of a + // message. + rpc FlagMessage(FlagMessageRequest) returns (FlagMessageResponse); + // ListConflicts returns mutations the server refused because another client + // had already changed the message. Conflicts are durable state, not a + // notification: the event stream is a lossy broadcast, so this table is the + // source of truth a client reconciles against. + rpc ListConflicts(ListConflictsRequest) returns (ListConflictsResponse); + // ResolveConflict marks a conflict handled, recording the human's decision. + rpc ResolveConflict(ResolveConflictRequest) returns (ResolveConflictResponse); // SearchMessages runs a full-text (FTS5) search over cached message // subjects, senders, and bodies, returning a keyset-paginated page of hits // ordered by relevance rank. @@ -542,8 +573,8 @@ message Message { // // A message can occupy several folders at once. This names the one the // message was reached through: the requested folder for a listing, and the - // lowest-sorting folder id for a lookup by id. A field carrying the complete - // placement set is not yet part of this contract. + // lowest-sorting folder id for a lookup by id. See `placements` for the + // complete set of mailboxes this message occupies. string folder_id = 3; // Subject line. string subject = 4; @@ -561,6 +592,48 @@ message Message { optional string body_html = 10; // List of attached files. repeated Attachment attachments = 11; + // Every mailbox this message occupies, folder id ascending. + // + // Message identity is content-addressed and folder-independent, so one + // message can sit in several mailboxes at once -- Gmail labels make that + // ordinary rather than exotic. Mutations that act on a single occupancy + // (`MoveMessage`, `DeleteMessage`) need one of these to address, and a client + // that only ever saw `folder_id` cannot tell a one-mailbox message from a + // ten-mailbox one. + repeated Placement placements = 12; +} + +// A single mailbox occupancy of a message: where one copy of it lives, and how +// to address that copy on the server that holds it. +message Placement { + // Account owning the mailbox. + string account_id = 1; + // Mailbox folder identifier. + string folder_id = 2; + // The UIDVALIDITY generation `uid` was captured under. Addressing is only + // valid within this generation; the server rejects a stale one rather than + // acting on whatever now bears that uid. + string uidvalidity = 3; + // Protocol-native id addressing the message in this mailbox. + string uid = 4; + // Read/unread flag of this occupancy specifically. + bool read = 5; +} + +// Names one occupancy for a mutation to act on. +// +// All four coordinates are required. This is deliberately not "the folder" -- +// UIDVALIDITY scopes the uid, and dropping it would let a mutation formed +// before a mailbox was recreated land on an unrelated message. +message PlacementRef { + // Account owning the mailbox. + string account_id = 1; + // Mailbox folder identifier. + string folder_id = 2; + // The UIDVALIDITY generation `uid` was captured under. + string uidvalidity = 3; + // Protocol-native id addressing the message in this mailbox. + string uid = 4; } // Mirrors `nuncio_core::model::Folder`. @@ -647,6 +720,106 @@ message MarkReadRequest { // response (as opposed to a `Status` error) is confirmation enough. message MarkReadResponse {} +// Request to move one occupancy of a message into another folder. +message MoveMessageRequest { + // Content-addressed key of the message to move. + string message_id = 1; + // Which occupancy to move. Optional only when the message occupies exactly + // one mailbox; with several, omitting it is refused rather than guessed, + // because moving the wrong copy is destructive and has no undo. + optional PlacementRef placement = 2; + // Destination folder id, within the same account as `placement`. + string destination_folder_id = 3; +} + +// Result of enqueuing a move. +message MoveMessageResponse { + // Identifier of the queued outbox mutation, for correlating with conflicts. + string mutation_id = 1; +} + +// Request to delete one occupancy of a message. +message DeleteMessageRequest { + // Content-addressed key of the message to delete. + string message_id = 1; + // Which occupancy to delete. Optional only when the message occupies exactly + // one mailbox. Deleting removes that copy; other occupancies survive. + optional PlacementRef placement = 2; +} + +// Result of enqueuing a delete. +message DeleteMessageResponse { + // Identifier of the queued outbox mutation. + string mutation_id = 1; +} + +// Request to flag or unflag one occupancy of a message. +message FlagMessageRequest { + // Content-addressed key of the message to flag. + string message_id = 1; + // Which occupancy to flag. Optional only when the message occupies exactly + // one mailbox. + optional PlacementRef placement = 2; + // Desired end state: true flags, false unflags. + bool flagged = 3; +} + +// Result of enqueuing a flag change. +message FlagMessageResponse { + // Identifier of the queued outbox mutation. + string mutation_id = 1; +} + +// A mutation the server refused because another client changed the message +// first, recorded durably so the decision can be made later. +message MutationConflict { + // Stable identifier of this conflict record. + string id = 1; + // The outbox mutation that conflicted. + string mutation_id = 2; + // Content-addressed key of the message the mutation targeted. + string message_id = 3; + // The operation that was refused (e.g. "MOVE", "DELETE", "FLAG"). + string mutation_type = 4; + // The occupancy the mutation was aimed at, when the queued intent named one. + optional PlacementRef placement = 5; + // The server's positive evidence that another client changed the message. + // A conflict is never inferred from an unverifiable outcome. + string observed = 6; + // When the conflict was detected. + google.protobuf.Timestamp detected_at = 7; + // When a human resolved it, if they have. + optional google.protobuf.Timestamp resolved_at = 8; + // Free-form note recording how it was resolved. + optional string resolution = 9; +} + +// Request to list recorded mutation conflicts. +message ListConflictsRequest { + // Include conflicts a human has already resolved. Resolved rows are kept as + // the audit trail of what was decided, so they are opt-in. + bool include_resolved = 1; + // Maximum number of conflicts to return; clamped to the server's page limit. + uint32 limit = 2; +} + +// A page of mutation conflicts, newest first. +message ListConflictsResponse { + // The conflicts. + repeated MutationConflict conflicts = 1; +} + +// Request to mark a conflict handled. +message ResolveConflictRequest { + // Identifier of the conflict to resolve. + string conflict_id = 1; + // Note recording how it was resolved. + string resolution = 2; +} + +// Result of resolving a conflict. +message ResolveConflictResponse {} + // SearchMessagesRequest. Pages hits using the shared keyset convention (see // `ListMessagesRequest`); the cursor here encodes `(rank, id)` so paging // preserves the relevance ordering while resuming strictly after the last diff --git a/crates/nuncio-proto/src/errors.rs b/crates/nuncio-proto/src/errors.rs index c13f1c6..7438b0a 100644 --- a/crates/nuncio-proto/src/errors.rs +++ b/crates/nuncio-proto/src/errors.rs @@ -37,6 +37,11 @@ impl ErrorReason { ErrorReason::BackendUnreachable => Code::Unavailable, ErrorReason::AuthRequired => Code::Unauthenticated, ErrorReason::Unsupported => Code::Unimplemented, + // `Aborted` is the canonical code for a concurrency conflict, and + // unlike `FailedPrecondition` it tells a client the call may be + // retried only after re-reading the current state -- which is + // exactly the remedy here. + ErrorReason::Conflict => Code::Aborted, } } } diff --git a/crates/nuncio-store/src/db.rs b/crates/nuncio-store/src/db.rs index 24ced58..946c8f5 100644 --- a/crates/nuncio-store/src/db.rs +++ b/crates/nuncio-store/src/db.rs @@ -11,6 +11,64 @@ use zeroize::Zeroizing; /// Raw row shape for a pending remote mutation record fetched from SQLite. type PendingMutationRow = (String, String, String, String, String, String, i64, i64); +/// `mutation_conflicts` row: id, mutation_id, message_id, mutation_type, then the +/// four nullable placement coordinates, observed evidence, and resolution state. +type ConflictRow = ( + String, + String, + String, + String, + Option, + Option, + Option, + Option, + String, + i64, + Option, + Option, +); + +/// Rebuild a conflict from its row, reassembling the placement only when all +/// four coordinates are present -- a partial placement addresses nothing. +fn conflict_from_row(row: ConflictRow) -> nuncio_core::model::MutationConflict { + let ( + id, + mutation_id, + message_id, + mutation_type, + account_id, + folder_id, + uid_validity, + remote_id, + observed, + detected_at, + resolved_at, + resolution, + ) = row; + let placement = match (account_id, folder_id, uid_validity, remote_id) { + (Some(account_id), Some(folder_id), Some(uid_validity), Some(remote_id)) => { + Some(nuncio_core::model::PlacementKey { + account_id, + folder_id, + uid_validity, + remote_id, + }) + } + _ => None, + }; + nuncio_core::model::MutationConflict { + id, + mutation_id, + message_id, + mutation_type, + placement, + observed, + detected_at, + resolved_at, + resolution, + } +} + /// Database errors emitted by `nuncio-store`. #[derive(Error, Debug)] pub enum DatabaseError { @@ -919,6 +977,21 @@ impl DatabaseEngine { created_at INTEGER NOT NULL ); + CREATE TABLE IF NOT EXISTS mutation_conflicts ( + id TEXT PRIMARY KEY NOT NULL, + mutation_id TEXT NOT NULL, + message_id TEXT NOT NULL, + mutation_type TEXT NOT NULL, + account_id TEXT, + folder_id TEXT, + uidvalidity TEXT, + uid TEXT, + observed TEXT NOT NULL, + detected_at INTEGER NOT NULL, + resolved_at INTEGER, + resolution TEXT + ); + CREATE TABLE IF NOT EXISTS filter_execution_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, rule_id TEXT NOT NULL, @@ -2670,6 +2743,103 @@ impl DatabaseEngine { Ok(()) } + /// Record a mutation conflict as durable, queryable state. + /// + /// `INSERT OR REPLACE` keyed on the conflict id makes a re-record of the + /// same conflict idempotent, so a drain pass that is retried after a crash + /// cannot multiply one refusal into several rows a human has to dismiss. + pub async fn record_conflict( + &self, + conflict: &nuncio_core::model::MutationConflict, + ) -> Result<(), DatabaseError> { + let placement = conflict.placement.as_ref(); + sqlx::query( + r#" + INSERT OR REPLACE INTO mutation_conflicts + (id, mutation_id, message_id, mutation_type, account_id, folder_id, + uidvalidity, uid, observed, detected_at, resolved_at, resolution) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + "#, + ) + .bind(&conflict.id) + .bind(&conflict.mutation_id) + .bind(&conflict.message_id) + .bind(&conflict.mutation_type) + .bind(placement.map(|p| p.account_id.clone())) + .bind(placement.map(|p| p.folder_id.clone())) + .bind(placement.map(|p| p.uid_validity.clone())) + .bind(placement.map(|p| p.remote_id.clone())) + .bind(&conflict.observed) + .bind(conflict.detected_at) + .bind(conflict.resolved_at) + .bind(conflict.resolution.as_deref()) + .execute(&self.pool) + .await + .map_err(DatabaseError::Query)?; + Ok(()) + } + + /// List recorded mutation conflicts, newest first. + /// + /// Resolved conflicts stay in the table -- they are the audit trail of what + /// a human decided -- so they are only returned when explicitly asked for. + pub async fn list_conflicts( + &self, + include_resolved: bool, + limit: usize, + ) -> Result, DatabaseError> { + let sql = if include_resolved { + r#" + SELECT id, mutation_id, message_id, mutation_type, account_id, folder_id, + uidvalidity, uid, observed, detected_at, resolved_at, resolution + FROM mutation_conflicts + ORDER BY detected_at DESC, id ASC + LIMIT ? + "# + } else { + r#" + SELECT id, mutation_id, message_id, mutation_type, account_id, folder_id, + uidvalidity, uid, observed, detected_at, resolved_at, resolution + FROM mutation_conflicts + WHERE resolved_at IS NULL + ORDER BY detected_at DESC, id ASC + LIMIT ? + "# + }; + let rows: Vec = sqlx::query_as(sql) + .bind(limit as i64) + .fetch_all(&self.pool) + .await + .map_err(DatabaseError::Query)?; + + Ok(rows.into_iter().map(conflict_from_row).collect()) + } + + /// Mark a conflict resolved. Returns `false` when no such unresolved + /// conflict exists, so a caller can tell "already handled" from "applied" + /// instead of reporting a success that never happened. + pub async fn resolve_conflict( + &self, + id: &str, + resolution: &str, + resolved_at: i64, + ) -> Result { + let result = sqlx::query( + r#" + UPDATE mutation_conflicts + SET resolved_at = ?, resolution = ? + WHERE id = ? AND resolved_at IS NULL + "#, + ) + .bind(resolved_at) + .bind(resolution) + .bind(id) + .execute(&self.pool) + .await + .map_err(DatabaseError::Query)?; + Ok(result.rows_affected() > 0) + } + /// Save a [`nuncio_filter::PendingRemoteMutation`] outbox record. pub async fn save_pending_mutation( &self, diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs index ce0986a..317df20 100644 --- a/crates/nunciod/src/grpc.rs +++ b/crates/nunciod/src/grpc.rs @@ -34,27 +34,30 @@ use nuncio_proto::v1::{ CalendarSyncRequest, CalendarSyncResponse, Contact as ContactProto, ContactEmail as ContactEmailProto, ContactPhone as ContactPhoneProto, ContactsSyncRequest, ContactsSyncResponse, CreateContactRequest, CreateContactResponse, CreateRuleRequest, - CreateRuleResponse, DatabaseRecovered, DavTransport as DavTransportProto, DeleteRuleRequest, - DeleteRuleResponse, Event, EventError, ExportFormat as ExportFormatProto, ExportRequest, - ExportResponse, ExportRulesRequest, ExportRulesResponse, FilterExecuted, - FilterExecutionLog as FilterExecutionLogProto, FilterRule as FilterRuleProto, - Folder as FolderProto, GetContactRequest, GetContactResponse, GetEventRequest, - GetEventResponse, GetExecutionLogsRequest, GetExecutionLogsResponse, GetMessageRequest, - GetMessageResponse, GetStatusRequest, GetStatusResponse, - ImapSmtpTransport as ImapSmtpTransportProto, ImportRulesRequest, ImportRulesResponse, - JmapTransport as JmapTransportProto, ListAccountsRequest, ListAccountsResponse, + CreateRuleResponse, DatabaseRecovered, DavTransport as DavTransportProto, DeleteMessageRequest, + DeleteMessageResponse, DeleteRuleRequest, DeleteRuleResponse, Event, EventError, + ExportFormat as ExportFormatProto, ExportRequest, ExportResponse, ExportRulesRequest, + ExportRulesResponse, FilterExecuted, FilterExecutionLog as FilterExecutionLogProto, + FilterRule as FilterRuleProto, FlagMessageRequest, FlagMessageResponse, Folder as FolderProto, + GetContactRequest, GetContactResponse, GetEventRequest, GetEventResponse, + GetExecutionLogsRequest, GetExecutionLogsResponse, GetMessageRequest, GetMessageResponse, + GetStatusRequest, GetStatusResponse, ImapSmtpTransport as ImapSmtpTransportProto, + ImportRulesRequest, ImportRulesResponse, JmapTransport as JmapTransportProto, + ListAccountsRequest, ListAccountsResponse, ListConflictsRequest, ListConflictsResponse, ListContactsRequest, ListContactsResponse, ListEventsRequest, ListEventsResponse, ListFoldersRequest, ListFoldersResponse, ListMessagesRequest, ListMessagesResponse, ListRecordsRequest, ListRecordsResponse, ListRulesRequest, ListRulesResponse, MarkReadRequest, MarkReadResponse, Message as MessageProto, MessageFlagsChanged, MessageSearchHit, - PreviewRuleRequest, PreviewRuleResponse, RemoveAccountRequest, RemoveAccountResponse, - RuleExportFormat as RuleExportFormatProto, SearchMessagesRequest, SearchMessagesResponse, - SendMessageRequest, SendMessageResponse, ShuttingDown, SubscribeRequest, SyncCompleted, - SyncRequest, SyncResponse, SyncStarted, SyncState as SyncStateProto, - TestAccountConnectionRequest, TestAccountConnectionResponse, TlsMode as TlsModeProto, - TriageProgress, TriageRequest, UpdateAccountRequest, UpdateAccountResponse, UpdateAvailable, - UpdateRuleRequest, UpdateRuleResponse, ValidateRuleRequest, ValidateRuleResponse, - VerifyChainRequest, VerifyChainResponse, + MoveMessageRequest, MoveMessageResponse, MutationConflict as MutationConflictProto, + Placement as PlacementProto, PlacementRef as PlacementRefProto, PreviewRuleRequest, + PreviewRuleResponse, RemoveAccountRequest, RemoveAccountResponse, ResolveConflictRequest, + ResolveConflictResponse, RuleExportFormat as RuleExportFormatProto, SearchMessagesRequest, + SearchMessagesResponse, SendMessageRequest, SendMessageResponse, ShuttingDown, + SubscribeRequest, SyncCompleted, SyncRequest, SyncResponse, SyncStarted, + SyncState as SyncStateProto, TestAccountConnectionRequest, TestAccountConnectionResponse, + TlsMode as TlsModeProto, TriageProgress, TriageRequest, UpdateAccountRequest, + UpdateAccountResponse, UpdateAvailable, UpdateRuleRequest, UpdateRuleResponse, + ValidateRuleRequest, ValidateRuleResponse, VerifyChainRequest, VerifyChainResponse, }; use nuncio_store::db::DatabaseEngine; use nuncio_store::search::SearchEngine; @@ -429,6 +432,7 @@ fn map_account_config_to_proto(config: nuncio_core::AccountConfig) -> AccountCon config.sync_interval_secs, )), transport: Some(transport), + filters_enabled: config.filters_enabled, } } @@ -490,7 +494,7 @@ fn map_account_config_from_proto( email_address: config.email_address, keyring_secret_key: config.keyring_secret_key, sync_interval_secs: nuncio_proto::time::duration_to_secs(&sync_interval), - filters_enabled: false, + filters_enabled: config.filters_enabled, transport, }) } @@ -993,6 +997,54 @@ fn map_email_to_proto(placed: nuncio_store::db::MessageWithPlacement) -> Message .into_iter() .map(map_attachment_to_proto) .collect(), + // Filled in by `fill_placements`, which needs the store. + placements: Vec::new(), + } +} + +/// Maps one stored occupancy onto its wire-format `nuncio.v1.Placement`. +fn map_placement_to_proto(placement: nuncio_core::model::Placement) -> PlacementProto { + PlacementProto { + account_id: placement.account_id, + folder_id: placement.folder_id, + uidvalidity: placement.uid_validity, + uid: placement.remote_id, + read: placement.read, + } +} + +/// Maps a wire-format `nuncio.v1.PlacementRef` onto the store's key type. +fn map_placement_ref_from_proto(placement: PlacementRefProto) -> nuncio_core::model::PlacementKey { + nuncio_core::model::PlacementKey { + account_id: placement.account_id, + folder_id: placement.folder_id, + uid_validity: placement.uidvalidity, + remote_id: placement.uid, + } +} + +/// Fills each message's complete placement set from the store. +/// +/// This costs one query per message rather than one per page. That is the +/// honest price of the field: occupancy is per-message state and a page can mix +/// messages that sit in one mailbox with messages that sit in ten. A failure to +/// read one message's placements leaves that message's set empty rather than +/// failing the whole listing, and is logged -- a partial page is more useful +/// than none, and the field is additive so a client that ignores it is +/// unaffected. +async fn fill_placements(db: &DatabaseEngine, messages: &mut [MessageProto]) { + for message in messages.iter_mut() { + match db.placements_of(&message.id).await { + Ok(placements) => { + message.placements = placements.into_iter().map(map_placement_to_proto).collect(); + } + Err(e) => { + tracing::warn!( + message_id = %message.id, + "Mail: failed to read the placement set: {e}" + ); + } + } } } @@ -1786,6 +1838,145 @@ async fn primary_placement( }) } +/// Rule id recorded on outbox rows a human queued directly rather than a filter. +/// +/// The outbox row's `rule_id` explains *why* a mutation exists; for a +/// hand-driven one the honest answer is "a person asked", not a rule that does +/// not exist. +const MANUAL_ORIGIN: &str = "manual"; + +/// Seconds since the Unix epoch, saturating at 0 before it. +fn unix_now_secs() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or_default() +} + +/// Resolve which occupancy a mutation should act on. +/// +/// A named placement is honoured only after confirming the message still +/// occupies it, so a client working from a stale listing cannot aim a `MOVE` or +/// `DELETE` at a mailbox the message has left. An unnamed placement is inferred +/// only when the message occupies exactly one mailbox; with several, this is a +/// typed validation failure that lists the candidates, because picking one +/// would act on mail the caller never selected and there is no undo. +async fn resolve_target_placement( + db: &DatabaseEngine, + message_id: &str, + named: Option, +) -> Result { + if message_id.is_empty() { + return Err(Status::invalid_argument("message_id is required")); + } + let placements = db.placements_of(message_id).await.map_err(|e| { + Status::internal(format!("failed to read placements for '{message_id}': {e}")) + })?; + + if let Some(named) = named { + let key = map_placement_ref_from_proto(named); + return if placements.iter().any(|p| p.key() == key) { + Ok(key) + } else { + Err(errors::status_with_metadata( + ErrorReason::MessageNotFound, + format!( + "message '{message_id}' does not occupy {}/{} (uidvalidity {}, uid {})", + key.account_id, key.folder_id, key.uid_validity, key.remote_id + ), + [("message_id".to_string(), message_id.to_string())], + )) + }; + } + + match placements.as_slice() { + [only] => Ok(only.key()), + [] => Err(errors::status_with_metadata( + ErrorReason::MessageNotFound, + format!("message '{message_id}' occupies no mailbox"), + [("message_id".to_string(), message_id.to_string())], + )), + several => { + let folders: Vec<&str> = several.iter().map(|p| p.folder_id.as_str()).collect(); + Err(errors::status_with_metadata( + ErrorReason::ValidationFailed, + format!( + "message '{message_id}' occupies {} mailboxes ({}); name one in `placement` \ + rather than leaving the target to be guessed", + several.len(), + folders.join(", ") + ), + [ + ("message_id".to_string(), message_id.to_string()), + ("placement_count".to_string(), several.len().to_string()), + ], + )) + } + } +} + +/// Maps a stored conflict onto its wire-format `nuncio.v1.MutationConflict`. +fn map_conflict_to_proto(conflict: nuncio_core::model::MutationConflict) -> MutationConflictProto { + MutationConflictProto { + id: conflict.id, + mutation_id: conflict.mutation_id, + message_id: conflict.message_id, + mutation_type: conflict.mutation_type, + placement: conflict.placement.map(|p| PlacementRefProto { + account_id: p.account_id, + folder_id: p.folder_id, + uidvalidity: p.uid_validity, + uid: p.remote_id, + }), + observed: conflict.observed, + detected_at: Some(nuncio_proto::time::timestamp_from_unix_secs( + conflict.detected_at, + )), + resolved_at: conflict + .resolved_at + .map(nuncio_proto::time::timestamp_from_unix_secs), + resolution: conflict.resolution, + } +} + +impl MailGrpcService { + /// Durably queue a placement-addressed remote mutation on the outbox. + /// + /// Every mutation RPC funnels through here rather than touching a backend + /// directly: the outbox is what makes an intent survive a crash, and what + /// turns the server's answer into one of three honest outcomes instead of a + /// hopeful assumption. The RPC returns once the intent is durable, which is + /// the only thing it can truthfully promise. + async fn enqueue_mutation( + &self, + message_id: &str, + op: &str, + target: Option, + placement: nuncio_core::model::PlacementKey, + ) -> Result { + let mutation = nuncio_filter::OutboxManager::create_mutation_at( + MANUAL_ORIGIN, + message_id, + op, + target, + Some(placement.clone()), + ); + self.db + .save_pending_mutation(&mutation) + .await + .map_err(|e| Status::internal(format!("failed to queue {op} mutation: {e}")))?; + tracing::info!( + mutation_id = %mutation.id, + op = %op, + message_id = %message_id, + account_id = %placement.account_id, + folder_id = %placement.folder_id, + "Mail: queued placement-addressed mutation" + ); + Ok(mutation.id) + } +} + #[tonic::async_trait] impl Mail for MailGrpcService { /// Lists folders across every **configured** account, merged by folder id. @@ -1870,7 +2061,9 @@ impl Mail for MailGrpcService { ]) }) .unwrap_or_default(); - let messages = messages.into_iter().map(map_email_to_proto).collect(); + let mut messages: Vec = + messages.into_iter().map(map_email_to_proto).collect(); + fill_placements(&self.db, &mut messages).await; Ok(Response::new(ListMessagesResponse { messages, @@ -1897,8 +2090,23 @@ impl Mail for MailGrpcService { let placement = primary_placement(&self.db, &req.message_id).await?; + let mut message = map_email_to_proto((email, placement)); + message.placements = self + .db + .placements_of(&req.message_id) + .await + .map_err(|e| { + Status::internal(format!( + "failed to read placements for '{}': {e}", + req.message_id + )) + })? + .into_iter() + .map(map_placement_to_proto) + .collect(); + Ok(Response::new(GetMessageResponse { - message: Some(map_email_to_proto((email, placement))), + message: Some(message), })) } @@ -1949,6 +2157,97 @@ impl Mail for MailGrpcService { Ok(Response::new(MarkReadResponse {})) } + async fn move_message( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + if req.destination_folder_id.is_empty() { + return Err(Status::invalid_argument( + "destination_folder_id is required", + )); + } + let placement = resolve_target_placement(&self.db, &req.message_id, req.placement).await?; + let mutation_id = self + .enqueue_mutation( + &req.message_id, + "MOVE", + Some(req.destination_folder_id.clone()), + placement, + ) + .await?; + Ok(Response::new(MoveMessageResponse { mutation_id })) + } + + async fn delete_message( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let placement = resolve_target_placement(&self.db, &req.message_id, req.placement).await?; + let mutation_id = self + .enqueue_mutation(&req.message_id, "DELETE", None, placement) + .await?; + Ok(Response::new(DeleteMessageResponse { mutation_id })) + } + + async fn flag_message( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let placement = resolve_target_placement(&self.db, &req.message_id, req.placement).await?; + let op = if req.flagged { "FLAG" } else { "UNFLAG" }; + let mutation_id = self + .enqueue_mutation(&req.message_id, op, None, placement) + .await?; + Ok(Response::new(FlagMessageResponse { mutation_id })) + } + + async fn list_conflicts( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let limit = pagination::clamp_page_size(req.limit); + let conflicts = self + .db + .list_conflicts(req.include_resolved, limit) + .await + .map_err(|e| Status::internal(format!("failed to list conflicts: {e}")))?; + Ok(Response::new(ListConflictsResponse { + conflicts: conflicts.into_iter().map(map_conflict_to_proto).collect(), + })) + } + + async fn resolve_conflict( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + if req.conflict_id.is_empty() { + return Err(Status::invalid_argument("conflict_id is required")); + } + let resolved_at = unix_now_secs(); + let applied = self + .db + .resolve_conflict(&req.conflict_id, &req.resolution, resolved_at) + .await + .map_err(|e| Status::internal(format!("failed to resolve conflict: {e}")))?; + if !applied { + // Either it never existed or someone already resolved it. Reporting + // success would tell the caller their decision was recorded when it + // was not. + return Err(errors::status_with_metadata( + ErrorReason::MessageNotFound, + format!("no unresolved conflict '{}' to resolve", req.conflict_id), + [("conflict_id".to_string(), req.conflict_id.clone())], + )); + } + tracing::info!(conflict_id = %req.conflict_id, "Mail: conflict resolved"); + Ok(Response::new(ResolveConflictResponse {})) + } + async fn search_messages( &self, request: Request, @@ -4298,6 +4597,7 @@ mod tests { email_address: format!("{id}@nuncio.mx"), keyring_secret_key: keyring_secret_key.to_string(), sync_interval: Some(nuncio_proto::time::duration_from_secs(60)), + filters_enabled: false, transport: Some(TransportProto::ImapSmtp(ImapSmtpTransportProto { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -4317,6 +4617,7 @@ mod tests { email_address: "no-transport@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-no-transport".to_string(), sync_interval: Some(nuncio_proto::time::duration_from_secs(60)), + filters_enabled: false, transport: None, }; let err = map_account_config_from_proto(config) diff --git a/crates/nunciod/src/outbox.rs b/crates/nunciod/src/outbox.rs index 05ed581..77fcaaf 100644 --- a/crates/nunciod/src/outbox.rs +++ b/crates/nunciod/src/outbox.rs @@ -222,6 +222,7 @@ pub async fn execute_pending_mutations( message_id = %item.message_id, "outbox: mutation conflicted with another client: {observed}" ); + record_conflict(db, &item, &observed).await; ("conflicted", item.retry_count, |s| s.failed += 1) } Disposition::Permanent(reason) => { @@ -260,6 +261,35 @@ pub async fn execute_pending_mutations( /// Execute a single mutation and report its honest disposition. Performs no /// status writes itself -- the caller persists the outcome. +/// Persist a conflict so it outlives the log line and the event stream. +/// +/// The conflict id is derived from the mutation id, which makes re-recording +/// the same refusal idempotent. A failure to persist is logged and swallowed: +/// the mutation's own status write is what stops it being re-attempted, and +/// losing the audit row must not take the drain pass down with it. +async fn record_conflict(db: &DatabaseEngine, item: &PendingRemoteMutation, observed: &str) { + let placement = serde_json::from_str::(&item.payload) + .ok() + .and_then(|p| p.placement); + let conflict = nuncio_core::model::MutationConflict { + id: format!("conflict-{}", item.id), + mutation_id: item.id.clone(), + message_id: item.message_id.clone(), + mutation_type: item.mutation_type.clone(), + placement, + observed: observed.to_string(), + detected_at: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or_default(), + resolved_at: None, + resolution: None, + }; + if let Err(e) = db.record_conflict(&conflict).await { + tracing::warn!(mutation_id = %item.id, "outbox: failed to record conflict: {e}"); + } +} + async fn execute_one( db: &DatabaseEngine, env: &dyn RemoteExecutionEnv, @@ -304,9 +334,12 @@ async fn execute_one( /// /// The addressing comes from a placement rather than the message row, because /// those coordinates are properties of a mailbox occupancy and a message can -/// hold several. The outbox row records only the message key, so which occupancy -/// the rule matched is not recoverable here -- and when the message occupies -/// more than one mailbox, this **refuses** rather than picking one. +/// hold several. When the row records the occupancy its intent was formed +/// against, that occupancy is used verbatim -- after confirming the message +/// still occupies it, so a mutation cannot be aimed at a mailbox the message has +/// since left. Otherwise the occupancy is inferred, and inference is only +/// allowed when it is unambiguous: with more than one placement this **refuses** +/// rather than picking one. /// /// Guessing is not the conservative option. A rule that matched the `INBOX` /// copy would `MOVE` or `DELETE` whichever copy happened to sort first, against @@ -317,9 +350,7 @@ async fn execute_one( /// guessing is destructive and irreversible. /// /// The refusal is permanent, not retryable: nothing about a later attempt makes -/// the outbox row remember a placement it never recorded. Carrying the -/// originating placement on the row is the real fix, and belongs with the -/// placement-aware wire surface. +/// a row remember a placement it never recorded. async fn dispatch_mailbox_mutation( db: &DatabaseEngine, env: &dyn RemoteExecutionEnv, @@ -359,6 +390,27 @@ async fn dispatch_mailbox_mutation( return Disposition::Retry(format!("failed to resolve the message's placements: {e}")) } }; + if let Some(named) = &payload.placement { + return match placements.iter().find(|p| &p.key() == named) { + Some(placement) => { + let placement = placement.clone(); + apply_mailbox_mutation(env, mutation, email, &placement, kind).await + } + // The named occupancy is gone. Re-addressing to a different one + // would act on mail the caller never selected, so this is terminal. + None => Disposition::Permanent(format!( + "message '{}' no longer occupies {}/{} (uidvalidity {}, uid {}); \ + refusing to re-aim a {} at a different mailbox", + email.id, + named.account_id, + named.folder_id, + named.uid_validity, + named.remote_id, + mutation.mutation_type + )), + }; + } + let placement = match placements.as_slice() { [only] => only, [] => { @@ -390,6 +442,18 @@ async fn dispatch_mailbox_mutation( } }; + apply_mailbox_mutation(env, mutation, email, placement, kind).await +} + +/// Issue an already-addressed mailbox mutation against the account's backend and +/// translate the backend's three honest outcomes into a disposition. +async fn apply_mailbox_mutation( + env: &dyn RemoteExecutionEnv, + mutation: &PendingRemoteMutation, + email: &nuncio_core::model::Email, + placement: &nuncio_core::model::Placement, + kind: RemoteMutationKind, +) -> Disposition { let backend = match env.mail_backend(&placement.account_id).await { Ok(backend) => backend, Err(e) => return Disposition::Retry(format!("failed to build mail backend: {e}")), @@ -723,6 +787,7 @@ mod tests { payload: serde_json::to_string(&MutationPayload { action_type: "FLAG".to_string(), target: None, + placement: None, }) .expect("serialize payload"), status: "pending".to_string(), @@ -826,6 +891,15 @@ mod tests { /// decided here anyway, so the narrower call is also the tighter assertion. async fn dispatch_with_placements( placements: &[Placement], + ) -> (Disposition, Vec) { + dispatch_with_placements_naming(placements, None).await + } + + /// As [`dispatch_with_placements`], but with the queued row's payload + /// naming `named` as the occupancy the intent was formed against. + async fn dispatch_with_placements_naming( + placements: &[Placement], + named: Option, ) -> (Disposition, Vec) { let (db, _dir) = DatabaseEngine::connect_ephemeral() .await @@ -840,7 +914,15 @@ mod tests { let env = SucceedingEnv { backend: backend.clone(), }; - let mutation = flag_mutation(); + let mut mutation = flag_mutation(); + if named.is_some() { + mutation.payload = serde_json::to_string(&MutationPayload { + action_type: mutation.mutation_type.clone(), + target: None, + placement: named, + }) + .expect("serialize payload"); + } let payload = serde_json::from_str::(&mutation.payload).expect("parse payload"); let disposition = @@ -900,4 +982,31 @@ mod tests { } } } + + /// The same two-mailbox message stops being ambiguous once the queued row + /// records which occupancy the intent was formed against: there is nothing + /// left to guess, so the mutation is addressed at exactly that copy -- + /// including when it is not the one a single-placement inference would have + /// reached. + #[tokio::test] + async fn a_named_placement_addresses_that_occupancy_rather_than_being_refused() { + let (disposition, applied) = dispatch_with_placements_naming( + &[sample_placement(), second_placement()], + Some(second_placement().key()), + ) + .await; + + assert!( + matches!(disposition, Disposition::Completed), + "a row that names its occupancy is fully addressed, not ambiguous" + ); + assert_eq!(applied.len(), 1, "the backend really was asked to act"); + assert_eq!( + applied[0].folder_id, "Archive", + "the named occupancy is the one acted on, not the other copy" + ); + assert_eq!(applied[0].remote_id, "78"); + assert_eq!(applied[0].uid_validity, "9"); + assert_eq!(applied[0].message_id, "msg-outbox-log"); + } } diff --git a/crates/nunciod/tests/accounts_lifecycle_e2e_test.rs b/crates/nunciod/tests/accounts_lifecycle_e2e_test.rs index 020bae1..268e905 100644 --- a/crates/nunciod/tests/accounts_lifecycle_e2e_test.rs +++ b/crates/nunciod/tests/accounts_lifecycle_e2e_test.rs @@ -90,6 +90,7 @@ fn sample_account_config() -> AccountConfig { email_address: ACCOUNT_EMAIL.to_string(), keyring_secret_key: KEYRING_KEY.to_string(), sync_interval: Some(nuncio_proto::time::duration_from_secs(60)), + filters_enabled: false, transport: Some(nuncio_proto::v1::account_config::Transport::ImapSmtp( nuncio_proto::v1::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), diff --git a/crates/nunciod/tests/caldav_production_e2e_test.rs b/crates/nunciod/tests/caldav_production_e2e_test.rs index 8576be5..7272b18 100644 --- a/crates/nunciod/tests/caldav_production_e2e_test.rs +++ b/crates/nunciod/tests/caldav_production_e2e_test.rs @@ -159,6 +159,7 @@ async fn caldav_production_sync_builds_real_client_and_persists_events() { email_address: ACCOUNT_EMAIL.to_string(), keyring_secret_key: KEYRING_KEY.to_string(), sync_interval: Some(nuncio_proto::time::duration_from_secs(300)), + filters_enabled: false, transport: Some(nuncio_proto::v1::account_config::Transport::Dav( nuncio_proto::v1::DavTransport { collection_url: collection_url.clone(), diff --git a/crates/nunciod/tests/mutation_rpcs_e2e_test.rs b/crates/nunciod/tests/mutation_rpcs_e2e_test.rs new file mode 100644 index 0000000..671428e --- /dev/null +++ b/crates/nunciod/tests/mutation_rpcs_e2e_test.rs @@ -0,0 +1,622 @@ +//! Offline end-to-end tests for the hand-driven mail mutation surface: the +//! placement-addressed `MoveMessage`/`DeleteMessage`/`FlagMessage` RPCs, the +//! durable conflict ledger behind `ListConflicts`/`ResolveConflict`, the +//! complete placement set on `Message`, and the per-account filter switch. +//! +//! Each test boots a real `nunciod` daemon gRPC server +//! (`nunciod::grpc::serve_on_listener_with_overrides`) on an ephemeral loopback +//! port, backed by a real temp file-backed [`DatabaseEngine`] and a +//! [`SecretManager::mock`] vault (never the real OS keyring), with every engine +//! override left at its default. The store handle is shared with the server, so +//! the assertions can seed state a sync pass would have written and read back +//! the durable rows the RPCs produced -- which is the only thing enqueuing can +//! truthfully promise, since nothing here reaches a remote server. +//! +//! Fully offline: no real network, no real OS keyring, no live server. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +use nuncio_core::model::{Email, IdentitySource, MutationConflict, Placement, PlacementKey}; +use nuncio_core::EventBus; +use nuncio_filter::{FilterEngine, MutationPayload}; +use nuncio_proto::errors; +use nuncio_proto::v1::{ + AccountConfig, AddAccountRequest, DeleteMessageRequest, ErrorReason, FlagMessageRequest, + GetMessageRequest, ListAccountsRequest, ListConflictsRequest, ListMessagesRequest, + MoveMessageRequest, PlacementRef, ResolveConflictRequest, UpdateAccountRequest, +}; +use nuncio_store::db::DatabaseEngine; +use nuncio_store::vault::{SecretManager, GRPC_TOKEN_ACCOUNT}; +use nunciod::grpc::{ + serve_on_listener_with_overrides, AccountsEngineOverrides, CalendarEngineOverrides, + ContactsEngineOverrides, MailEngineOverrides, +}; +use std::sync::Arc; +use tokio::net::TcpListener; +use tonic::Code; + +const ACCOUNT: &str = "acct-mutation-e2e"; +/// One UIDVALIDITY generation for every seeded occupancy: these tests are about +/// which mailbox a mutation lands in, not about generation bumps. +const UIDVALIDITY: &str = "7"; + +/// Boots the real daemon gRPC server on an ephemeral loopback port with every +/// override at its default. +/// +/// Returns the dialable address, the minted bearer token, and the SAME store +/// handle the server is using, so a test can seed rows and inspect what an RPC +/// durably wrote. The `TempDir` guard is returned so the caller controls its +/// lifetime for the whole test (the spawned server keeps using the db). +async fn boot_daemon(db_name: &str) -> (String, String, Arc, tempfile::TempDir) { + let secrets = Arc::new(SecretManager::mock()); + let dir = tempfile::tempdir().expect("create temp dir for file-backed db"); + let db_path = dir.path().join(db_name); + let db = Arc::new( + DatabaseEngine::connect_file(&db_path, &secrets) + .await + .expect("open temp file-backed database"), + ); + + let token = hex::encode( + secrets + .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32) + .expect("mint gRPC bearer token from mock vault"), + ); + + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral loopback port"); + let addr = listener.local_addr().expect("listener has local addr"); + + let event_bus = Arc::new(EventBus::new()); + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); + let server_db = db.clone(); + let server_token = token.clone(); + tokio::spawn(async move { + let _ = serve_on_listener_with_overrides( + listener, + event_bus, + server_db, + filter_engine, + secrets, + server_token, + MailEngineOverrides::default(), + CalendarEngineOverrides::default(), + ContactsEngineOverrides::default(), + AccountsEngineOverrides::default(), + ) + .await; + }); + + (addr.to_string(), token, db, dir) +} + +/// One seeded mailbox occupancy of the test account. +fn placement(folder_id: &str, uid: &str) -> Placement { + Placement { + account_id: ACCOUNT.to_string(), + folder_id: folder_id.to_string(), + uid_validity: UIDVALIDITY.to_string(), + remote_id: uid.to_string(), + read: false, + } +} + +/// The wire form a client uses to name that same occupancy. +fn placement_ref(folder_id: &str, uid: &str) -> PlacementRef { + PlacementRef { + account_id: ACCOUNT.to_string(), + folder_id: folder_id.to_string(), + uidvalidity: UIDVALIDITY.to_string(), + uid: uid.to_string(), + } +} + +/// The store-side key the queued outbox row is expected to carry. +fn placement_key(folder_id: &str, uid: &str) -> PlacementKey { + placement(folder_id, uid).key() +} + +/// The account row the read RPCs merge over. `Mail` reads carry no account +/// field, so mail belonging to no configured account is unreachable through +/// them -- seeding one keeps these tests on the production path. +async fn seed_account(db: &DatabaseEngine) { + db.save_account(&nuncio_core::AccountConfig { + id: ACCOUNT.to_string(), + name: "Mutation E2E".to_string(), + email_address: "mutation-e2e@nuncio.mx".to_string(), + keyring_secret_key: format!("nuncio/{ACCOUNT}"), + sync_interval_secs: 60, + filters_enabled: false, + transport: nuncio_core::Transport::Jmap(nuncio_core::JmapTransport { + endpoint_host: "jmap.nuncio.mx".to_string(), + }), + }) + .await + .expect("save the owning account"); +} + +/// Persist one message into each of the given occupancies, the way a sync pass +/// over several mailboxes would. +async fn seed_message(db: &DatabaseEngine, id: &str, placements: &[Placement]) { + let email = Email { + id: id.to_string(), + account_id: ACCOUNT.to_string(), + subject: format!("subject of {id}"), + sender: "alice@nuncio.mx".to_string(), + recipient: "owner@nuncio.mx".to_string(), + received_at: 1_700_000_000, + body_plain: Some(format!("body of {id}")), + body_html: None, + attachments: Vec::new(), + message_id: None, + content_hash: None, + }; + for p in placements { + db.save_email_at(&email, IdentitySource::Surrogate, p) + .await + .expect("seed the occupancy"); + } +} + +/// The queued outbox row with this id, decoded into its operation and payload. +async fn queued_mutation(db: &DatabaseEngine, mutation_id: &str) -> (String, MutationPayload) { + let pending = db + .list_pending_mutations(50) + .await + .expect("read the outbox queue"); + let row = pending + .into_iter() + .find(|m| m.id == mutation_id) + .unwrap_or_else(|| panic!("no queued outbox row '{mutation_id}'")); + let payload = + serde_json::from_str::(&row.payload).expect("decode the queued payload"); + assert_eq!( + row.rule_id, "manual", + "a hand-driven mutation is attributed to a person, not to a rule" + ); + (row.mutation_type, payload) +} + +/// A message in two mailboxes cannot be moved by naming only the message: one +/// of the copies would be picked for the caller, and a misaimed MOVE is +/// destructive with no undo. Naming the occupancy makes the same call succeed, +/// and the queued row records which copy the intent was formed against. +#[tokio::test] +async fn moving_a_multi_placement_message_requires_the_occupancy_to_be_named() { + let (addr, token, db, _dir) = boot_daemon("mutation_move.db").await; + seed_account(&db).await; + seed_message( + &db, + "msg-multi", + &[placement("INBOX", "101"), placement("Archive", "202")], + ) + .await; + + let mut mail = nuncio_proto::client::connect_mail(&addr, &token) + .await + .expect("mail client connects"); + + // ---- unnamed: refused, and the refusal names the candidates ---- + let status = mail + .move_message(MoveMessageRequest { + message_id: "msg-multi".to_string(), + placement: None, + destination_folder_id: "Trash".to_string(), + }) + .await + .expect_err("an ambiguously-addressed move must be refused"); + assert_eq!(status.code(), Code::InvalidArgument); + assert_eq!( + errors::error_reason(&status), + Some(ErrorReason::ValidationFailed) + ); + assert!( + status.message().contains("INBOX") && status.message().contains("Archive"), + "the refusal must name the candidate mailboxes: {}", + status.message() + ); + assert!( + db.list_pending_mutations(50) + .await + .expect("read the outbox queue") + .is_empty(), + "a refused move must not leave a queued intent behind" + ); + + // ---- named: accepted, and durable ---- + let mutation_id = mail + .move_message(MoveMessageRequest { + message_id: "msg-multi".to_string(), + placement: Some(placement_ref("INBOX", "101")), + destination_folder_id: "Trash".to_string(), + }) + .await + .expect("naming the occupancy makes the move well-defined") + .into_inner() + .mutation_id; + assert!( + !mutation_id.is_empty(), + "the caller needs an id to correlate the queued intent with a conflict" + ); + + let (mutation_type, payload) = queued_mutation(&db, &mutation_id).await; + assert_eq!(mutation_type, "MOVE"); + assert_eq!(payload.action_type, "MOVE"); + assert_eq!(payload.target.as_deref(), Some("Trash")); + assert_eq!( + payload.placement, + Some(placement_key("INBOX", "101")), + "the queued row must remember the occupancy the intent was formed against" + ); +} + +/// Delete, flag and unflag queue the operation they were asked for, addressed +/// at a specific occupancy: named explicitly when there are several to choose +/// between, and inferred only when the message occupies exactly one mailbox. +#[tokio::test] +async fn delete_and_flag_queue_placement_addressed_rows() { + let (addr, token, db, _dir) = boot_daemon("mutation_delete_flag.db").await; + seed_account(&db).await; + seed_message( + &db, + "msg-multi", + &[placement("INBOX", "101"), placement("Archive", "202")], + ) + .await; + seed_message(&db, "msg-solo", &[placement("INBOX", "303")]).await; + + let mut mail = nuncio_proto::client::connect_mail(&addr, &token) + .await + .expect("mail client connects"); + + let delete_id = mail + .delete_message(DeleteMessageRequest { + message_id: "msg-multi".to_string(), + placement: Some(placement_ref("Archive", "202")), + }) + .await + .expect("delete of a named occupancy is well-defined") + .into_inner() + .mutation_id; + let (mutation_type, payload) = queued_mutation(&db, &delete_id).await; + assert_eq!(mutation_type, "DELETE"); + assert_eq!(payload.action_type, "DELETE"); + assert_eq!(payload.target, None); + assert_eq!(payload.placement, Some(placement_key("Archive", "202"))); + + let flag_id = mail + .flag_message(FlagMessageRequest { + message_id: "msg-multi".to_string(), + placement: Some(placement_ref("INBOX", "101")), + flagged: true, + }) + .await + .expect("flag of a named occupancy is well-defined") + .into_inner() + .mutation_id; + let (mutation_type, payload) = queued_mutation(&db, &flag_id).await; + assert_eq!(mutation_type, "FLAG"); + assert_eq!(payload.action_type, "FLAG"); + assert_eq!(payload.placement, Some(placement_key("INBOX", "101"))); + + // `flagged: false` is a distinct operation, not a FLAG with a parameter -- + // the executor maps the two onto opposite remote calls. + let unflag_id = mail + .flag_message(FlagMessageRequest { + message_id: "msg-solo".to_string(), + placement: None, + flagged: false, + }) + .await + .expect("a single-occupancy message needs no placement to be unambiguous") + .into_inner() + .mutation_id; + let (mutation_type, payload) = queued_mutation(&db, &unflag_id).await; + assert_eq!(mutation_type, "UNFLAG"); + assert_eq!(payload.action_type, "UNFLAG"); + assert_eq!( + payload.placement, + Some(placement_key("INBOX", "303")), + "the inferred occupancy is recorded too, so the executor never re-infers it" + ); +} + +/// Naming an occupancy the message does not hold is a not-found error, never a +/// silent re-aim at whichever copy does exist: the caller formed its intent +/// against a mailbox that is no longer true, and acting on a different one +/// would touch mail it never selected. +#[tokio::test] +async fn naming_an_occupancy_the_message_does_not_hold_is_refused() { + let (addr, token, db, _dir) = boot_daemon("mutation_stale_placement.db").await; + seed_account(&db).await; + seed_message(&db, "msg-solo", &[placement("INBOX", "303")]).await; + + let mut mail = nuncio_proto::client::connect_mail(&addr, &token) + .await + .expect("mail client connects"); + + let status = mail + .move_message(MoveMessageRequest { + message_id: "msg-solo".to_string(), + placement: Some(placement_ref("Spam", "999")), + destination_folder_id: "Trash".to_string(), + }) + .await + .expect_err("a stale occupancy must not be honoured"); + assert_eq!(status.code(), Code::NotFound); + assert_eq!( + errors::error_reason(&status), + Some(ErrorReason::MessageNotFound) + ); + assert!( + status.message().contains("Spam"), + "the error must name the occupancy that was asked for: {}", + status.message() + ); + + // A delete aimed at the wrong uid within a real folder is equally refused: + // the uid is part of the address, not a hint. + let status = mail + .delete_message(DeleteMessageRequest { + message_id: "msg-solo".to_string(), + placement: Some(placement_ref("INBOX", "304")), + }) + .await + .expect_err("a wrong uid must not be rounded to the right one"); + assert_eq!( + errors::error_reason(&status), + Some(ErrorReason::MessageNotFound) + ); + + assert!( + db.list_pending_mutations(50) + .await + .expect("read the outbox queue") + .is_empty(), + "a refused mutation must leave nothing queued" + ); +} + +/// Both read RPCs report the message's complete occupancy set, folder id +/// ascending, while `folder_id` keeps its older meaning: the occupancy the +/// message was reached through. A client that only ever saw `folder_id` cannot +/// tell a one-mailbox message from a two-mailbox one. +#[tokio::test] +async fn read_rpcs_report_every_occupancy_of_a_message() { + let (addr, token, db, _dir) = boot_daemon("mutation_placements_read.db").await; + seed_account(&db).await; + seed_message( + &db, + "msg-multi", + &[placement("INBOX", "101"), placement("Archive", "202")], + ) + .await; + + let mut mail = nuncio_proto::client::connect_mail(&addr, &token) + .await + .expect("mail client connects"); + + let fetched = mail + .get_message(GetMessageRequest { + message_id: "msg-multi".to_string(), + }) + .await + .expect("get_message succeeds") + .into_inner() + .message + .expect("a found message is always populated"); + let folders: Vec<&str> = fetched + .placements + .iter() + .map(|p| p.folder_id.as_str()) + .collect(); + assert_eq!(folders, vec!["Archive", "INBOX"]); + let uids: Vec<&str> = fetched.placements.iter().map(|p| p.uid.as_str()).collect(); + assert_eq!(uids, vec!["202", "101"]); + assert!(fetched + .placements + .iter() + .all(|p| p.account_id == ACCOUNT && p.uidvalidity == UIDVALIDITY)); + assert_eq!( + fetched.folder_id, "Archive", + "folder_id still names the single occupancy this read reached the message through" + ); + + // Reached through INBOX this time: the same complete set, a different + // `folder_id`. + let listed = mail + .list_messages(ListMessagesRequest { + folder_id: "INBOX".to_string(), + page_size: 10, + page_token: String::new(), + }) + .await + .expect("list_messages succeeds") + .into_inner() + .messages; + assert_eq!(listed.len(), 1); + assert_eq!(listed[0].folder_id, "INBOX"); + let folders: Vec<&str> = listed[0] + .placements + .iter() + .map(|p| p.folder_id.as_str()) + .collect(); + assert_eq!(folders, vec!["Archive", "INBOX"]); +} + +/// A conflict is durable state, not a passing notification: it survives in a +/// queryable table, resolving it is recorded once and only once, and resolved +/// rows stay as the audit trail of what was decided. +#[tokio::test] +async fn conflicts_round_trip_from_record_through_resolution() { + let (addr, token, db, _dir) = boot_daemon("mutation_conflicts.db").await; + seed_account(&db).await; + seed_message(&db, "msg-solo", &[placement("INBOX", "303")]).await; + + db.record_conflict(&MutationConflict { + id: "conflict-1".to_string(), + mutation_id: "mut-conflicted".to_string(), + message_id: "msg-solo".to_string(), + mutation_type: "MOVE".to_string(), + placement: Some(placement_key("INBOX", "303")), + observed: "another client moved the message to Archive first".to_string(), + detected_at: 1_700_000_500, + resolved_at: None, + resolution: None, + }) + .await + .expect("record the conflict"); + + let mut mail = nuncio_proto::client::connect_mail(&addr, &token) + .await + .expect("mail client connects"); + + let conflicts = mail + .list_conflicts(ListConflictsRequest { + include_resolved: false, + limit: 10, + }) + .await + .expect("list_conflicts succeeds") + .into_inner() + .conflicts; + assert_eq!(conflicts.len(), 1); + assert_eq!(conflicts[0].id, "conflict-1"); + assert_eq!(conflicts[0].mutation_id, "mut-conflicted"); + assert_eq!(conflicts[0].mutation_type, "MOVE"); + assert_eq!( + conflicts[0].observed, + "another client moved the message to Archive first" + ); + assert_eq!( + conflicts[0].placement, + Some(placement_ref("INBOX", "303")), + "the conflict names the occupancy the refused mutation was aimed at" + ); + assert_eq!(conflicts[0].resolved_at, None); + + mail.resolve_conflict(ResolveConflictRequest { + conflict_id: "conflict-1".to_string(), + resolution: "kept the other client's move".to_string(), + }) + .await + .expect("resolve_conflict succeeds"); + + // Resolving it again is an honest error: reporting success would tell the + // caller their decision was recorded when the earlier one already stands. + let status = mail + .resolve_conflict(ResolveConflictRequest { + conflict_id: "conflict-1".to_string(), + resolution: "changed my mind".to_string(), + }) + .await + .expect_err("a conflict can only be resolved once"); + assert_eq!(status.code(), Code::NotFound); + assert_eq!( + errors::error_reason(&status), + Some(ErrorReason::MessageNotFound) + ); + + let unresolved = mail + .list_conflicts(ListConflictsRequest { + include_resolved: false, + limit: 10, + }) + .await + .expect("list_conflicts succeeds") + .into_inner() + .conflicts; + assert!( + unresolved.is_empty(), + "a handled conflict is no longer outstanding work" + ); + + let all = mail + .list_conflicts(ListConflictsRequest { + include_resolved: true, + limit: 10, + }) + .await + .expect("list_conflicts succeeds") + .into_inner() + .conflicts; + assert_eq!(all.len(), 1, "the audit trail survives the resolution"); + assert_eq!( + all[0].resolution.as_deref(), + Some("kept the other client's move") + ); + assert!( + all[0].resolved_at.is_some(), + "a resolved conflict records when the decision was made" + ); +} + +/// `filters_enabled` is settable over the wire rather than only by editing the +/// accounts row by hand. It still defaults to off, because filters take +/// irreversible remote actions and an account should not acquire them merely by +/// being added. +#[tokio::test] +async fn filters_enabled_round_trips_over_the_wire() { + let (addr, token, _db, _dir) = boot_daemon("mutation_filters_flag.db").await; + + let mut accounts = nuncio_proto::client::connect_accounts(&addr, &token) + .await + .expect("accounts client connects"); + + let config = AccountConfig { + id: ACCOUNT.to_string(), + name: "Mutation E2E".to_string(), + email_address: "mutation-e2e@nuncio.mx".to_string(), + keyring_secret_key: format!("nuncio/{ACCOUNT}"), + sync_interval: Some(nuncio_proto::time::duration_from_secs(60)), + filters_enabled: false, + transport: Some(nuncio_proto::v1::account_config::Transport::Jmap( + nuncio_proto::v1::JmapTransport { + endpoint_host: "jmap.nuncio.mx".to_string(), + }, + )), + }; + accounts + .add_account(AddAccountRequest { + config: Some(config.clone()), + password: "mutation-e2e-secret".to_string(), + }) + .await + .expect("add_account succeeds"); + + let listed = accounts + .list_accounts(ListAccountsRequest {}) + .await + .expect("list_accounts succeeds") + .into_inner() + .accounts; + assert_eq!(listed.len(), 1); + assert!( + !listed[0].filters_enabled, + "an account does not get filters by being added, only by being told to" + ); + + accounts + .update_account(UpdateAccountRequest { + config: Some(AccountConfig { + filters_enabled: true, + ..config + }), + password: None, + }) + .await + .expect("update_account succeeds"); + + let relisted = accounts + .list_accounts(ListAccountsRequest {}) + .await + .expect("list_accounts succeeds") + .into_inner() + .accounts; + assert_eq!(relisted.len(), 1); + assert!( + relisted[0].filters_enabled, + "the flip must be persisted and reported back, not silently discarded" + ); +} diff --git a/crates/nunciod/tests/no_secrets_canary_test.rs b/crates/nunciod/tests/no_secrets_canary_test.rs index 413a78c..92bf3b2 100644 --- a/crates/nunciod/tests/no_secrets_canary_test.rs +++ b/crates/nunciod/tests/no_secrets_canary_test.rs @@ -122,6 +122,7 @@ fn sample_proto_config() -> AccountConfig { email_address: ACCOUNT_EMAIL.to_string(), keyring_secret_key: KEYRING_KEY.to_string(), sync_interval: Some(nuncio_proto::time::duration_from_secs(60)), + filters_enabled: false, transport: Some(nuncio_proto::v1::account_config::Transport::ImapSmtp( nuncio_proto::v1::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), diff --git a/crates/nunciod/tests/spine_e2e_test.rs b/crates/nunciod/tests/spine_e2e_test.rs index 81f798e..398c213 100644 --- a/crates/nunciod/tests/spine_e2e_test.rs +++ b/crates/nunciod/tests/spine_e2e_test.rs @@ -60,6 +60,7 @@ fn sample_account_config() -> AccountConfig { email_address: ACCOUNT_EMAIL.to_string(), keyring_secret_key: KEYRING_KEY.to_string(), sync_interval: Some(nuncio_proto::time::duration_from_secs(60)), + filters_enabled: false, transport: Some(nuncio_proto::v1::account_config::Transport::ImapSmtp( nuncio_proto::v1::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(),