diff --git a/Cargo.lock b/Cargo.lock index f4191a1..ac6e67b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2652,6 +2652,7 @@ dependencies = [ "tonic", "tracing", "tracing-subscriber", + "wiremock", ] [[package]] diff --git a/crates/nuncio-cli/src/args.rs b/crates/nuncio-cli/src/args.rs index db359e8..abd2f38 100644 --- a/crates/nuncio-cli/src/args.rs +++ b/crates/nuncio-cli/src/args.rs @@ -517,6 +517,18 @@ pub enum FilterSubcommand { #[arg(short, long, default_value_t = 50, help = "Log limit")] limit: usize, }, + /// Retroactively apply the current rule set to every stored message, + /// streaming cumulative progress as the scan runs. + Triage { + /// Restrict the scan's reporting to a single rule ID. Not yet + /// honored by the daemon (see `nuncio.v1.Filters.Triage`) -- the + /// scan always evaluates every active rule. + #[arg(short, long, help = "Rule identifier (not yet honored)")] + rule_id: Option, + /// Messages fetched per streamed progress update. + #[arg(short, long, help = "Messages scanned per chunk")] + chunk_size: Option, + }, } #[cfg(test)] diff --git a/crates/nuncio-cli/src/runner.rs b/crates/nuncio-cli/src/runner.rs index 6b10426..e5a3a53 100644 --- a/crates/nuncio-cli/src/runner.rs +++ b/crates/nuncio-cli/src/runner.rs @@ -162,6 +162,20 @@ fn filter_rule_proto_to_json(rule: &nuncio_proto::v1::FilterRule) -> serde_json: }) } +fn filter_execution_log_proto_to_json( + log: &nuncio_proto::v1::FilterExecutionLog, +) -> serde_json::Value { + json!({ + "id": log.id, + "rule_id": log.rule_id, + "message_id": log.message_id, + "action_taken": log.action_taken, + "matched_at": log.matched_at, + "prev_hash": log.prev_hash, + "hash": log.hash, + }) +} + /// Errors emitted by the CLI headless runner. #[derive(Error, Debug)] pub enum RunnerError { @@ -474,132 +488,36 @@ impl HeadlessRunner { self.handle_filter_test(sql, message_id.as_deref(), json_mode) .await } - // `filter edit`/`export`/`import`/`logs` are NOT part of the - // `Filters` gRPC surface this runner implements - // (`CreateRule`/`ListRules`/`DeleteRule`/ - // `ValidateRule`/`PreviewRule`); they still read/write this - // runner's own ephemeral local `db` below, exactly as - // before. Because `List`/`Create`/`Delete` above now go - // through the daemon's real, persistent store instead, a - // rule created via `filter create` will NOT show up in - // `filter export`/`filter logs` (which only see this - // process's throwaway `db`) until these are migrated too -- - // a known, intentional gap, not a regression - // introduced silently here. FilterSubcommand::Edit { id, name, sql, priority, } => { - let existing = self - .db - .list_filter_rules() - .await - .unwrap_or_default() - .into_iter() - .find(|r| r.id == *id); - if let Some(rule) = existing { - let rule_name = name.clone().unwrap_or(rule.name); - let rule_sql = sql.clone().unwrap_or(rule.nsql_text); - let rule_priority = priority.unwrap_or(rule.priority); - - match nuncio_filter::NsqlParser::parse_rule( - &rule_name, - rule_priority, - &rule_sql, - ) { - Ok(mut updated) => { - updated.id = id.clone(); - if let Err(e) = self.db.save_filter_rule(&updated).await { - return if json_mode { - format_json_error(&e.to_string()) - } else { - format!("Database Error: {e}") - }; - } - if json_mode { - format_json(&json!(updated)) - } else { - format!("✓ Updated filter rule '{}'.", id) - } - } - Err(e) => { - if json_mode { - format_json_error(&e.to_string()) - } else { - format!("Syntax Error: {e}") - } - } - } - } else if json_mode { - format_json_error(&format!("Rule '{}' not found", id)) - } else { - format!("Rule '{}' not found.", id) - } + self.handle_filter_edit( + id, + name.as_deref(), + sql.as_deref(), + *priority, + json_mode, + ) + .await } FilterSubcommand::Export { format } => { - let rules = self.db.list_filter_rules().await.unwrap_or_default(); - if format == "json" || json_mode { - format_json(&json!(rules)) - } else { - let sqls: Vec = rules.iter().map(|r| r.to_nsql()).collect(); - sqls.join("\n") - } + self.handle_filter_export(format, json_mode).await + } + FilterSubcommand::Import { file } => { + self.handle_filter_import(file, json_mode).await } - FilterSubcommand::Import { file } => match std::fs::read_to_string(file) { - Ok(content) => { - let mut imported = 0; - for line in content.lines() { - let line_trim = line.trim(); - if line_trim.is_empty() || line_trim.starts_with("--") { - continue; - } - if let Ok(rule) = nuncio_filter::NsqlParser::parse_rule( - format!("Imported Rule {}", imported + 1), - 0, - line_trim, - ) { - if self.db.save_filter_rule(&rule).await.is_ok() { - imported += 1; - } - } - } - if json_mode { - format_json(&json!({ "imported_count": imported })) - } else { - format!("✓ Successfully imported {} filter rules.", imported) - } - } - Err(e) => { - if json_mode { - format_json_error(&e.to_string()) - } else { - format!("Failed to read file: {e}") - } - } - }, FilterSubcommand::Logs { limit } => { - let logs = self - .db - .list_filter_execution_logs(*limit) + self.handle_filter_logs(*limit, json_mode).await + } + FilterSubcommand::Triage { + rule_id, + chunk_size, + } => { + self.handle_filter_triage(rule_id.clone(), *chunk_size, json_mode) .await - .unwrap_or_default(); - if json_mode { - format_json(&json!(logs)) - } else if logs.is_empty() { - "No execution logs recorded.".to_string() - } else { - let mut out = - String::from("ID RULE_ID MSG_ID ACTION TIMESTAMP\n"); - for l in logs { - out.push_str(&format!( - "{:<4} {:<10} {:<10} {:<12} {}\n", - l.id, l.rule_id, l.message_id, l.action_taken, l.matched_at - )); - } - out - } } }, Commands::Update { action } => match action { @@ -1395,6 +1313,271 @@ impl HeadlessRunner { } } + /// `filter edit`: a real thin gRPC client of the running `nunciod` + /// daemon's `nuncio.v1.Filters` API. Only the supplied overrides are + /// sent; the daemon applies them to the existing persisted rule, + /// re-validates, persists, and reloads its own live `FilterEngine`. + async fn handle_filter_edit( + &self, + id: &str, + name: Option<&str>, + sql: Option<&str>, + priority: Option, + json_mode: bool, + ) -> String { + let mut client = match self.connect_filters_client().await { + Ok(client) => client, + Err(e) => return Self::render_error(&e, json_mode), + }; + + match client + .update_rule(nuncio_proto::v1::UpdateRuleRequest { + id: id.to_string(), + name: name.map(str::to_string), + nsql: sql.map(str::to_string), + priority, + }) + .await + { + Ok(response) => match response.into_inner().rule { + Some(rule) => { + if json_mode { + format_json(&json!({ "rule": filter_rule_proto_to_json(&rule) })) + } else { + format!("✓ Updated filter rule '{}'.", rule.id) + } + } + None => Self::render_error( + "nunciod daemon accepted update_rule but returned no rule", + json_mode, + ), + }, + Err(status) + if status.code() == tonic::Code::InvalidArgument + || status.code() == tonic::Code::NotFound => + { + Self::render_error(status.message(), json_mode) + } + Err(status) => Self::render_error( + &format!("nunciod daemon rejected update_rule: {status}"), + json_mode, + ), + } + } + + /// `filter export`: a real thin gRPC client of the running `nunciod` + /// daemon's `nuncio.v1.Filters` API, rendering every rule persisted in + /// the daemon's real, persistent store -- not this runner's own + /// ephemeral local `db`. + async fn handle_filter_export(&self, format: &str, json_mode: bool) -> String { + let mut client = match self.connect_filters_client().await { + Ok(client) => client, + Err(e) => return Self::render_error(&e, json_mode), + }; + + match client + .export_rules(nuncio_proto::v1::ExportRulesRequest { + format: format.to_string(), + }) + .await + { + Ok(response) => { + let content = response.into_inner().content; + if json_mode { + format_json(&json!({ "content": content })) + } else { + content + } + } + Err(status) => Self::render_error( + &format!("nunciod daemon rejected export_rules: {status}"), + json_mode, + ), + } + } + + /// `filter import`: reads the local file (the file lives on the CLI's + /// filesystem, which is why this stays a local read), then forwards the + /// raw content to the daemon for parsing -- so the same + /// `NsqlParser::parse_rule` and persistence path `filter create` uses is + /// exercised, and rules land in the daemon's real, persistent store. + async fn handle_filter_import(&self, file: &str, json_mode: bool) -> String { + let content = match std::fs::read_to_string(file) { + Ok(content) => content, + Err(e) => { + return if json_mode { + format_json_error(&e.to_string()) + } else { + format!("Failed to read file: {e}") + }; + } + }; + + let mut client = match self.connect_filters_client().await { + Ok(client) => client, + Err(e) => return Self::render_error(&e, json_mode), + }; + + match client + .import_rules(nuncio_proto::v1::ImportRulesRequest { content }) + .await + { + Ok(response) => { + let response = response.into_inner(); + if json_mode { + format_json(&json!({ + "imported_count": response.imported_count, + "errors": response.errors, + })) + } else { + let mut out = format!( + "✓ Successfully imported {} filter rules.", + response.imported_count + ); + for err in &response.errors { + out.push_str(&format!("\n ✗ {err}")); + } + out + } + } + Err(status) => Self::render_error( + &format!("nunciod daemon rejected import_rules: {status}"), + json_mode, + ), + } + } + + /// `filter logs`: a real thin gRPC client of the running `nunciod` + /// daemon's `nuncio.v1.Filters` API, reading the daemon's real, + /// persistent filter execution log ledger. + async fn handle_filter_logs(&self, limit: usize, json_mode: bool) -> String { + let mut client = match self.connect_filters_client().await { + Ok(client) => client, + Err(e) => return Self::render_error(&e, json_mode), + }; + + match client + .get_execution_logs(nuncio_proto::v1::GetExecutionLogsRequest { + limit: limit as u32, + }) + .await + { + Ok(response) => { + let logs = response.into_inner().logs; + if json_mode { + let logs_json: Vec = logs + .iter() + .map(filter_execution_log_proto_to_json) + .collect(); + format_json(&json!({ "logs": logs_json })) + } else if logs.is_empty() { + "No execution logs recorded.".to_string() + } else { + let mut out = + String::from("ID RULE_ID MSG_ID ACTION TIMESTAMP\n"); + for l in logs { + out.push_str(&format!( + "{:<4} {:<10} {:<10} {:<12} {}\n", + l.id, l.rule_id, l.message_id, l.action_taken, l.matched_at + )); + } + out + } + } + Err(status) => Self::render_error( + &format!("nunciod daemon rejected get_execution_logs: {status}"), + json_mode, + ), + } + } + + /// Drains the server-streaming `Filters.Triage` RPC, printing (or + /// accumulating, in `--json` mode) each `TriageProgress` update as it + /// arrives so the caller can watch a retroactive rescan of the whole + /// message store progress in real time rather than blocking silently + /// until it finishes. + async fn handle_filter_triage( + &self, + rule_id: Option, + chunk_size: Option, + json_mode: bool, + ) -> String { + let mut client = match self.connect_filters_client().await { + Ok(client) => client, + Err(e) => return Self::render_error(&e, json_mode), + }; + + let mut stream = match client + .triage(nuncio_proto::v1::TriageRequest { + rule_id, + chunk_size: chunk_size.unwrap_or(0), + }) + .await + { + Ok(response) => response.into_inner(), + Err(status) => { + return Self::render_error( + &format!("nunciod daemon rejected triage: {status}"), + json_mode, + ) + } + }; + + let mut updates = Vec::new(); + loop { + match stream.message().await { + Ok(Some(progress)) => { + if !json_mode { + println!( + "scanned={} matched={} actions_applied={} last_message_id={} done={}", + progress.scanned_count, + progress.matched_count, + progress.actions_applied_count, + progress.last_message_id, + progress.done + ); + } + let done = progress.done; + updates.push(progress); + if done { + break; + } + } + Ok(None) => break, + Err(status) => { + return Self::render_error( + &format!("nunciod daemon triage stream failed: {status}"), + json_mode, + ) + } + } + } + + if json_mode { + let updates_json: Vec = updates + .iter() + .map(|p| { + json!({ + "scanned_count": p.scanned_count, + "matched_count": p.matched_count, + "actions_applied_count": p.actions_applied_count, + "last_message_id": p.last_message_id, + "done": p.done, + }) + }) + .collect(); + format_json(&json!({ "updates": updates_json })) + } else { + match updates.last() { + Some(last) => format!( + "Triage complete: scanned={} matched={} actions_applied={}", + last.scanned_count, last.matched_count, last.actions_applied_count + ), + None => "Triage produced no progress updates.".to_string(), + } + } + } + /// Resolves the gRPC bearer token from the injected vault and dials the /// `nuncio.v1.Filters` service at `self.grpc_addr`, shared by every /// `filter` handler above. @@ -3143,18 +3326,23 @@ mod tests { use nuncio_proto::v1::filters_server::{Filters as FiltersService, FiltersServer}; use nuncio_proto::v1::{ CreateRuleRequest, CreateRuleResponse, DeleteRuleRequest, DeleteRuleResponse, - FilterRule as FilterRuleProto, ListRulesRequest, ListRulesResponse, PreviewRuleRequest, - PreviewRuleResponse, ValidateRuleRequest, ValidateRuleResponse, + ExportRulesRequest, ExportRulesResponse, FilterExecutionLog as FilterExecutionLogProto, + FilterRule as FilterRuleProto, GetExecutionLogsRequest, GetExecutionLogsResponse, + ImportRulesRequest, ImportRulesResponse, ListRulesRequest, ListRulesResponse, + PreviewRuleRequest, PreviewRuleResponse, TriageProgress, TriageRequest, + UpdateRuleRequest, UpdateRuleResponse, ValidateRuleRequest, ValidateRuleResponse, }; use std::sync::Mutex; /// Minimal test-only stub of `nuncio.v1.Filters`: records the last - /// `DeleteRuleRequest` it received (so this test can assert on - /// exactly what the CLI sent over the wire) and otherwise returns - /// fixed responses. + /// `DeleteRuleRequest`/`UpdateRuleRequest`/`ImportRulesRequest` it + /// received (so this test can assert on exactly what the CLI sent + /// over the wire) and otherwise returns fixed responses. #[derive(Default)] struct StubFilters { last_delete: Arc>>, + last_update: Arc>>, + last_import: Arc>>, } #[tonic::async_trait] @@ -3244,10 +3432,112 @@ mod tests { condition_traces: vec!["Rule 'Stub Rule': MATCH".to_string()], })) } + + async fn update_rule( + &self, + request: tonic::Request, + ) -> Result, tonic::Status> { + let req = request.into_inner(); + if req.id != "rule-stub-1" { + return Err(tonic::Status::not_found("no such rule")); + } + *self.last_update.lock().unwrap_or_else(|e| e.into_inner()) = Some(req.clone()); + Ok(tonic::Response::new(UpdateRuleResponse { + rule: Some(FilterRuleProto { + id: req.id, + name: req.name.unwrap_or_else(|| "Stub Rule".to_string()), + target_account: "*".to_string(), + priority: req.priority.unwrap_or(5), + enabled: true, + nsql_text: req + .nsql + .unwrap_or_else(|| "WHERE subject CONTAINS 'Urgent'".to_string()), + actions: vec!["MARK READ".to_string()], + created_at: 1_700_000_000, + updated_at: 1_700_000_001, + }), + })) + } + + async fn export_rules( + &self, + request: tonic::Request, + ) -> Result, tonic::Status> { + let req = request.into_inner(); + let content = if req.format == "json" { + r#"[{"id":"rule-stub-1"}]"#.to_string() + } else { + "SELECT * FROM emails WHERE subject CONTAINS 'Urgent' ACTION MARK READ" + .to_string() + }; + Ok(tonic::Response::new(ExportRulesResponse { content })) + } + + async fn import_rules( + &self, + request: tonic::Request, + ) -> Result, tonic::Status> { + let req = request.into_inner(); + *self.last_import.lock().unwrap_or_else(|e| e.into_inner()) = Some(req.clone()); + Ok(tonic::Response::new(ImportRulesResponse { + imported_count: 1, + errors: vec!["failed to parse 'garbage': stub parse error".to_string()], + })) + } + + async fn get_execution_logs( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> { + Ok(tonic::Response::new(GetExecutionLogsResponse { + logs: vec![FilterExecutionLogProto { + id: 1, + rule_id: "rule-stub-1".to_string(), + message_id: "msg-stub-1".to_string(), + action_taken: "MARK READ".to_string(), + matched_at: 1_700_000_002, + prev_hash: "prev-hash".to_string(), + hash: "hash".to_string(), + }], + })) + } + + type TriageStream = std::pin::Pin< + Box< + dyn tokio_stream::Stream> + + Send + + 'static, + >, + >; + + async fn triage( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> { + let updates = vec![ + Ok(TriageProgress { + scanned_count: 2, + matched_count: 1, + actions_applied_count: 1, + last_message_id: "msg-stub-2".to_string(), + done: false, + }), + Ok(TriageProgress { + scanned_count: 3, + matched_count: 2, + actions_applied_count: 2, + last_message_id: "msg-stub-3".to_string(), + done: true, + }), + ]; + Ok(tonic::Response::new(Box::pin(tokio_stream::iter(updates)))) + } } let stub = StubFilters::default(); let delete_probe = stub.last_delete.clone(); + let update_probe = stub.last_update.clone(); + let import_probe = stub.last_import.clone(); let listener = tokio::net::TcpListener::bind("127.0.0.1:0") .await @@ -3368,6 +3658,139 @@ mod tests { .await; assert!(test_out.contains(r#""matched":true"#)); assert!(test_out.contains("msg-stub-1")); + + // `filter edit`: proves the CLI forwards only the supplied + // overrides and renders the daemon's updated rule, and that a + // not-found `id` surfaces the daemon's honest `NotFound` error. + let edit_out = runner + .execute_command( + &Commands::Filter { + action: FilterSubcommand::Edit { + id: "rule-stub-1".to_string(), + name: Some("Renamed Rule".to_string()), + sql: None, + priority: Some(9), + }, + }, + true, + ) + .await; + assert!(edit_out.contains("Renamed Rule")); + assert!(edit_out.contains(r#""priority":9"#)); + let recorded_update = update_probe + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone() + .expect("stub daemon received an update_rule request"); + assert_eq!(recorded_update.id, "rule-stub-1"); + assert_eq!(recorded_update.name.as_deref(), Some("Renamed Rule")); + assert_eq!(recorded_update.nsql, None); + assert_eq!(recorded_update.priority, Some(9)); + + let edit_not_found = runner + .execute_command( + &Commands::Filter { + action: FilterSubcommand::Edit { + id: "no-such-rule".to_string(), + name: None, + sql: None, + priority: None, + }, + }, + true, + ) + .await; + assert!(edit_not_found.contains(r#""status":"error""#)); + + // `filter export`: proves the CLI forwards `format` and renders the + // daemon's rendered content as-is. + let export_sql = runner + .execute_command( + &Commands::Filter { + action: FilterSubcommand::Export { + format: "sql".to_string(), + }, + }, + false, + ) + .await; + assert!(export_sql.contains("SELECT * FROM emails")); + + let export_json = runner + .execute_command( + &Commands::Filter { + action: FilterSubcommand::Export { + format: "json".to_string(), + }, + }, + true, + ) + .await; + assert!(export_json.contains("rule-stub-1")); + + // `filter import`: proves the CLI reads the local file and forwards + // its raw content, and surfaces the daemon's per-line errors + // instead of silently dropping them. + let import_file = std::env::temp_dir().join(format!( + "nuncio-cli-filter-import-test-{}.sql", + std::process::id() + )); + std::fs::write(&import_file, "WHERE subject CONTAINS 'Urgent'\ngarbage\n") + .expect("write temp import file"); + let import_out = runner + .execute_command( + &Commands::Filter { + action: FilterSubcommand::Import { + file: import_file.to_string_lossy().to_string(), + }, + }, + true, + ) + .await; + let _ = std::fs::remove_file(&import_file); + assert!(import_out.contains(r#""imported_count":1"#)); + assert!(import_out.contains("stub parse error")); + let recorded_import = import_probe + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone() + .expect("stub daemon received an import_rules request"); + assert!(recorded_import + .content + .contains("WHERE subject CONTAINS 'Urgent'")); + assert!(recorded_import.content.contains("garbage")); + + // `filter logs`: proves the CLI reads the daemon's real execution + // log ledger, not any local/ephemeral state. + let logs_out = runner + .execute_command( + &Commands::Filter { + action: FilterSubcommand::Logs { limit: 10 }, + }, + true, + ) + .await; + assert!(logs_out.contains("msg-stub-1")); + assert!(logs_out.contains(r#""action_taken":"MARK READ""#)); + + // `filter triage`: proves the CLI drains the server-streaming + // `Triage` RPC (its first real streaming-RPC consumer) and reports + // the final cumulative progress rather than only the first chunk. + let triage_out = runner + .execute_command( + &Commands::Filter { + action: FilterSubcommand::Triage { + rule_id: None, + chunk_size: Some(2), + }, + }, + true, + ) + .await; + assert!(triage_out.contains(r#""scanned_count":3"#)); + assert!(triage_out.contains(r#""matched_count":2"#)); + assert!(triage_out.contains(r#""actions_applied_count":2"#)); + assert!(triage_out.contains(r#""done":true"#)); } #[tokio::test] diff --git a/crates/nuncio-filter/src/engine.rs b/crates/nuncio-filter/src/engine.rs index e3f4822..bce77e8 100644 --- a/crates/nuncio-filter/src/engine.rs +++ b/crates/nuncio-filter/src/engine.rs @@ -97,9 +97,12 @@ impl CompiledFilter { .unwrap_or(""); Self::eval_string_op(body, &leaf.operator, &leaf.value, regexes) } - FilterField::Folder | FilterField::Account => { + FilterField::Folder => { Self::eval_string_op(&email.folder_id, &leaf.operator, &leaf.value, regexes) } + FilterField::Account => { + Self::eval_string_op(&email.account_id, &leaf.operator, &leaf.value, regexes) + } FilterField::HasAttachment => { let has = !email.attachments.is_empty(); if let FilterValue::Boolean(b) = leaf.value { @@ -148,7 +151,11 @@ impl CompiledFilter { } } FilterField::Header(_name) => { - // Header evaluation fallback + // Unreachable defensive branch: `NsqlValidator::pass1_field_types` + // rejects any rule with a `header[...]` condition before it can + // reach a `CompiledFilter`, since `Email` has no parsed-headers + // field to match against. Kept only so this match stays + // exhaustive over `FilterField`. false } } @@ -475,6 +482,30 @@ mod tests { assert_eq!(engine.evaluate(&email).len(), 1); } + #[test] + fn test_account_condition_field_matches_account_id_not_folder_id() { + let nsql = "WHERE account = 'acct-a' ACTION MARK READ"; + let rule = crate::parser::NsqlParser::parse_rule("Account Filter", 1, nsql).unwrap(); + let engine = FilterEngine::new(vec![rule]).unwrap(); + + // Same folder_id, different account_id: a rule matching on the + // `account` condition field must key off email.account_id, never + // fall through to comparing folder_id. + let email_a = test_email("acct-a", "Anything", "shared-folder"); + assert_eq!( + engine.evaluate(&email_a).len(), + 1, + "rule on account = 'acct-a' must match a message with account_id 'acct-a'" + ); + + let email_b = test_email("acct-b", "Anything", "shared-folder"); + assert!( + engine.evaluate(&email_b).is_empty(), + "rule on account = 'acct-a' must not match a message with account_id 'acct-b', \ + even when it shares the same folder_id as an acct-a message" + ); + } + #[test] fn test_not_in_operator_is_complement_of_in() { let nsql = "WHERE folder NOT IN ('spam', 'trash') ACTION MARK READ"; diff --git a/crates/nuncio-filter/src/validator.rs b/crates/nuncio-filter/src/validator.rs index 145d836..ff67cbd 100644 --- a/crates/nuncio-filter/src/validator.rs +++ b/crates/nuncio-filter/src/validator.rs @@ -83,11 +83,14 @@ impl NsqlValidator { "field 'has_attachment' requires a boolean value".to_string(), )), }, + FilterField::Header(_) => Err(ValidationError::FieldTypeMismatch( + "header field conditions are not yet supported: message headers are not indexed" + .to_string(), + )), FilterField::Subject | FilterField::From | FilterField::To | FilterField::Body - | FilterField::Header(_) | FilterField::Folder | FilterField::Account => match &leaf.value { FilterValue::String(_) | FilterValue::List(_) => Ok(()), @@ -322,6 +325,18 @@ mod tests { assert!(matches!(err, ValidationError::FieldTypeMismatch(_))); } + #[test] + fn test_pass1_rejects_unsupported_header_field_conditions() { + let rule = crate::parser::NsqlParser::parse_rule( + "Header Spam Score", + 1, + "WHERE header['X-Spam-Score'] = '10' ACTION MARK READ", + ) + .unwrap(); + let err = NsqlValidator::validate(&rule, &ValidationOptions::default()).unwrap_err(); + assert!(matches!(err, ValidationError::FieldTypeMismatch(_))); + } + #[test] fn test_pass4_redos_validation() { let rule = crate::parser::NsqlParser::parse_rule( diff --git a/crates/nuncio-mail/Cargo.toml b/crates/nuncio-mail/Cargo.toml index 2f822f5..7099f8e 100644 --- a/crates/nuncio-mail/Cargo.toml +++ b/crates/nuncio-mail/Cargo.toml @@ -26,8 +26,8 @@ rustls.workspace = true webpki-roots.workspace = true rustls-pki-types.workspace = true tokio-stream.workspace = true +reqwest.workspace = true [dev-dependencies] tempfile.workspace = true wiremock.workspace = true -reqwest.workspace = true diff --git a/crates/nuncio-mail/src/jmap.rs b/crates/nuncio-mail/src/jmap.rs index ce68390..5ff136f 100644 --- a/crates/nuncio-mail/src/jmap.rs +++ b/crates/nuncio-mail/src/jmap.rs @@ -57,8 +57,25 @@ pub struct JmapEmailChangesResponse { pub destroyed: Vec, } -/// Raw JMAP email object representation. +/// JMAP `Email/query` response payload wrapper (RFC 8620 Section 5.5), narrowed +/// to just the matched id list -- `Email/get` needs explicit ids (or `None` +/// for "all", which real servers reject for large mailboxes), so a query must +/// always precede a get. #[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct JmapEmailQueryResponse { + /// Matched email ids, in server-defined order. + #[serde(default)] + pub ids: Vec, +} + +/// Raw JMAP email object representation (RFC 8621 Section 4.1). Field names +/// must stay `camelCase` on the wire -- `receivedAt`/`isUnread`/`bodySnippet` +/// are the actual JMAP property names; without `rename_all` these silently +/// fail to match and every parsed message gets a zero timestamp, `read: +/// true`, and no snippet. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] pub struct JmapEmail { pub id: String, pub subject: Option, @@ -75,19 +92,70 @@ pub struct JmapAddress { pub email: String, } -/// JMAP protocol engine implementing RFC 8620 / 8621. +/// Raw JMAP `Mailbox` object representation (RFC 8621 Section 2). +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct JmapMailbox { + pub id: String, + pub name: String, + #[serde(default)] + pub total_emails: u32, + #[serde(default)] + pub unread_emails: u32, +} + +/// JMAP `Mailbox/get` response payload wrapper. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct JmapMailboxGetResponse { + /// List of retrieved mailbox objects. + pub list: Vec, +} + +/// JMAP protocol engine implementing RFC 8620 / 8621. Mirrors [`crate::imap::ImapEngine`]'s +/// shape: `new` builds an uncredentialed engine that keeps the deterministic, network-free +/// fallback behaviour existing preview/dry-run callers and unit tests rely on; `with_credentials` +/// is what production (`nunciod::sync::build_mail_backend`) always uses, and performs genuine +/// HTTP session discovery and JSON-RPC calls against the real server. pub struct JmapEngine { account_id: String, + host: String, + username: Option, + password: Option, + http: reqwest::Client, } impl JmapEngine { - /// Create a new `JmapEngine` bound to an account ID. + /// Create a new, uncredentialed `JmapEngine` bound to an account ID. Never performs + /// network I/O -- callers that need a real server connection must use + /// [`JmapEngine::with_credentials`]. pub fn new(account_id: &str) -> Self { Self { account_id: account_id.to_string(), + host: String::new(), + username: None, + password: None, + http: reqwest::Client::new(), } } + /// Create a new `JmapEngine` with a server host and Basic-auth credentials. This is the + /// production path: `sync_folders`/`sync_messages` perform real HTTP session discovery and + /// JMAP JSON-RPC calls against `host`. + pub fn with_credentials(account_id: &str, host: &str, username: &str, password: &str) -> Self { + Self { + account_id: account_id.to_string(), + host: host.to_string(), + username: Some(username.to_string()), + password: Some(password.to_string()), + http: reqwest::Client::new(), + } + } + + fn has_credentials(&self) -> bool { + self.username.is_some() && self.password.is_some() + } + /// Construct standard JMAP Well-Known session discovery URL (RFC 8620 Section 2.1). pub fn build_session_url(host: &str) -> String { if host.starts_with("http://") || host.starts_with("https://") { @@ -116,6 +184,25 @@ impl JmapEngine { }) } + /// Build JSON-RPC request invocation for `Email/query`, filtered to a single mailbox. + /// `Email/get` requires an explicit id list (or `None` for "all", which real servers + /// reject for anything but tiny mailboxes), so a query must always precede a get. + pub fn build_email_query_request(account_id: &str, mailbox_id: &str) -> Value { + json!({ + "using": ["urn:ietf:params:jmap:core", "urn:ietf:params:jmap:mail"], + "methodCalls": [ + [ + "Email/query", + { + "accountId": account_id, + "filter": { "inMailboxes": [mailbox_id] } + }, + "c1" + ] + ] + }) + } + /// Build JSON-RPC request invocation for `Email/changes` (differential sync). pub fn build_email_changes_request(account_id: &str, since_state: &str) -> Value { json!({ @@ -133,12 +220,46 @@ impl JmapEngine { }) } + /// Build JSON-RPC request invocation for `Mailbox/get` (all mailboxes). + pub fn build_mailbox_get_request(account_id: &str) -> Value { + json!({ + "using": ["urn:ietf:params:jmap:core", "urn:ietf:params:jmap:mail"], + "methodCalls": [ + [ + "Mailbox/get", + { + "accountId": account_id, + "ids": null + }, + "c1" + ] + ] + }) + } + /// Parse JMAP RFC 8620 session response JSON. pub fn parse_session(raw_json: &str) -> Result { serde_json::from_str(raw_json) .map_err(|e| MailError::ParseFailed(format!("invalid JMAP session JSON: {e}"))) } + /// Extract the argument payload of the first method call from a JMAP JSON-RPC response, + /// supporting both the full `methodResponses` envelope and a bare response object (used by + /// tests that exercise a parser directly against a single method's payload). + fn extract_method_response_payload(val: &Value) -> Result { + if let Some(calls) = val.get("methodResponses").and_then(|v| v.as_array()) { + let first_call = calls + .first() + .ok_or_else(|| MailError::ParseFailed("empty methodResponses array".to_string()))?; + let args = first_call.get(1).ok_or_else(|| { + MailError::ParseFailed("missing method response payload".to_string()) + })?; + Ok(args.clone()) + } else { + Ok(val.clone()) + } + } + /// Parse raw JMAP `Email/get` JSON response payload into domain [`Email`] list and new state string. pub fn parse_email_get_response( &self, @@ -146,23 +267,9 @@ impl JmapEngine { ) -> Result<(Vec, String), MailError> { let val: Value = serde_json::from_str(raw_json) .map_err(|e| MailError::ParseFailed(format!("invalid JMAP JSON response: {e}")))?; - - // Support both direct object and RFC 8620 methodCalls wrapper - let resp: JmapEmailGetResponse = if let Some(calls) = - val.get("methodResponses").and_then(|v| v.as_array()) - { - let first_call = calls - .first() - .ok_or_else(|| MailError::ParseFailed("empty methodResponses array".to_string()))?; - let args = first_call.get(1).ok_or_else(|| { - MailError::ParseFailed("missing method response payload".to_string()) - })?; - serde_json::from_value(args.clone()) - .map_err(|e| MailError::ParseFailed(format!("invalid Email/get payload: {e}")))? - } else { - serde_json::from_value(val) - .map_err(|e| MailError::ParseFailed(format!("invalid Email/get payload: {e}")))? - }; + let payload = Self::extract_method_response_payload(&val)?; + let resp: JmapEmailGetResponse = serde_json::from_value(payload) + .map_err(|e| MailError::ParseFailed(format!("invalid Email/get payload: {e}")))?; let emails = resp .list @@ -208,31 +315,117 @@ impl JmapEngine { ) -> Result<(Vec, Vec, String), MailError> { let val: Value = serde_json::from_str(raw_json) .map_err(|e| MailError::ParseFailed(format!("invalid JMAP JSON response: {e}")))?; - - let resp: JmapEmailChangesResponse = - if let Some(calls) = val.get("methodResponses").and_then(|v| v.as_array()) { - let first_call = calls.first().ok_or_else(|| { - MailError::ParseFailed("empty methodResponses array".to_string()) - })?; - let args = first_call.get(1).ok_or_else(|| { - MailError::ParseFailed("missing method response payload".to_string()) - })?; - serde_json::from_value(args.clone()).map_err(|e| { - MailError::ParseFailed(format!("invalid Email/changes payload: {e}")) - })? - } else { - serde_json::from_value(val).map_err(|e| { - MailError::ParseFailed(format!("invalid Email/changes payload: {e}")) - })? - }; + let payload = Self::extract_method_response_payload(&val)?; + let resp: JmapEmailChangesResponse = serde_json::from_value(payload) + .map_err(|e| MailError::ParseFailed(format!("invalid Email/changes payload: {e}")))?; Ok((resp.updated, resp.destroyed, resp.new_state)) } + + /// Parse raw JMAP `Email/query` JSON response payload into a matched id list. + pub fn parse_email_query_response(raw_json: &str) -> Result, MailError> { + let val: Value = serde_json::from_str(raw_json) + .map_err(|e| MailError::ParseFailed(format!("invalid JMAP JSON response: {e}")))?; + let payload = Self::extract_method_response_payload(&val)?; + let resp: JmapEmailQueryResponse = serde_json::from_value(payload) + .map_err(|e| MailError::ParseFailed(format!("invalid Email/query payload: {e}")))?; + + Ok(resp.ids) + } + + /// Parse raw JMAP `Mailbox/get` JSON response payload into a mailbox list. + pub fn parse_mailbox_get_response(raw_json: &str) -> Result, MailError> { + let val: Value = serde_json::from_str(raw_json) + .map_err(|e| MailError::ParseFailed(format!("invalid JMAP JSON response: {e}")))?; + let payload = Self::extract_method_response_payload(&val)?; + let resp: JmapMailboxGetResponse = serde_json::from_value(payload) + .map_err(|e| MailError::ParseFailed(format!("invalid Mailbox/get payload: {e}")))?; + + Ok(resp.list) + } + + /// Issue an authenticated JMAP session discovery `GET /.well-known/jmap` against `self.host`. + /// A network failure or unparseable response surfaces as a genuine [`MailError`] -- there is + /// no fallback to canned data once credentials are present. + async fn discover_session(&self) -> Result { + let url = Self::build_session_url(&self.host); + let mut request = self.http.get(&url); + if let (Some(u), Some(p)) = (&self.username, &self.password) { + request = request.basic_auth(u, Some(p)); + } + + let response = request.send().await.map_err(|e| { + MailError::NetworkError(format!("JMAP session discovery at {url} failed: {e}")) + })?; + let status = response.status(); + let body = response.text().await.map_err(|e| { + MailError::NetworkError(format!("failed to read JMAP session response body: {e}")) + })?; + if !status.is_success() { + return Err(MailError::NetworkError(format!( + "JMAP session discovery at {url} returned status {status}" + ))); + } + + Self::parse_session(&body) + } + + /// POST a JMAP JSON-RPC request body to `api_url`, returning the raw response text. A + /// non-2xx status or transport failure surfaces as [`MailError::NetworkError`]. + async fn post_jmap(&self, api_url: &str, body: &Value) -> Result { + let mut request = self.http.post(api_url).json(body); + if let (Some(u), Some(p)) = (&self.username, &self.password) { + request = request.basic_auth(u, Some(p)); + } + + let response = request.send().await.map_err(|e| { + MailError::NetworkError(format!("JMAP request to {api_url} failed: {e}")) + })?; + let status = response.status(); + let text = response.text().await.map_err(|e| { + MailError::NetworkError(format!("failed to read JMAP response body: {e}")) + })?; + if !status.is_success() { + return Err(MailError::NetworkError(format!( + "JMAP server at {api_url} returned status {status}" + ))); + } + + Ok(text) + } + + /// Resolve the JMAP `accountId` to use for API calls: the session's advertised primary mail + /// account if present, else the Nuncio account id as a last resort. + fn resolve_account_id(&self, session: &JmapSession) -> String { + session + .primary_accounts + .get("urn:ietf:params:jmap:mail") + .cloned() + .unwrap_or_else(|| self.account_id.clone()) + } } #[async_trait] impl MailBackend for JmapEngine { async fn sync_folders(&self) -> Result, MailError> { + if self.has_credentials() { + let session = self.discover_session().await?; + let account_id = self.resolve_account_id(&session); + let request = Self::build_mailbox_get_request(&account_id); + let raw = self.post_jmap(&session.api_url, &request).await?; + let mailboxes = Self::parse_mailbox_get_response(&raw)?; + + return Ok(mailboxes + .into_iter() + .map(|mb| Folder { + id: mb.id, + name: mb.name, + total_messages: mb.total_emails as usize, + unread_messages: mb.unread_emails as usize, + }) + .collect()); + } + Ok(vec![ Folder { id: "inbox".to_string(), @@ -257,9 +450,22 @@ impl MailBackend for JmapEngine { async fn sync_messages( &self, - _folder_id: &str, + folder_id: &str, _since_state: Option<&str>, ) -> Result<(Vec, String), MailError> { + if self.has_credentials() { + let session = self.discover_session().await?; + let account_id = self.resolve_account_id(&session); + + let query_request = Self::build_email_query_request(&account_id, folder_id); + let query_raw = self.post_jmap(&session.api_url, &query_request).await?; + let ids = Self::parse_email_query_response(&query_raw)?; + + let get_request = Self::build_email_get_request(&account_id, Some(ids)); + let get_raw = self.post_jmap(&session.api_url, &get_request).await?; + return self.parse_email_get_response(&get_raw); + } + let sample_jmap = json!({ "state": "s-100", "list": [ @@ -314,6 +520,23 @@ mod tests { assert_eq!(req["methodCalls"][0][1]["sinceState"], "state-1"); } + #[test] + fn build_email_query_request_json_structure() { + let req = JmapEngine::build_email_query_request("acct-1", "mailbox-1"); + assert_eq!(req["methodCalls"][0][0], "Email/query"); + assert_eq!( + req["methodCalls"][0][1]["filter"]["inMailboxes"][0], + "mailbox-1" + ); + } + + #[test] + fn build_mailbox_get_request_json_structure() { + let req = JmapEngine::build_mailbox_get_request("acct-1"); + assert_eq!(req["methodCalls"][0][0], "Mailbox/get"); + assert_eq!(req["methodCalls"][0][1]["accountId"], "acct-1"); + } + #[test] fn parse_session_valid_payload() { let raw = r#"{ @@ -350,4 +573,69 @@ mod tests { let res = engine.parse_email_get_response("{ invalid json }"); assert!(res.is_err()); } + + #[test] + fn parse_email_query_response_extracts_ids() { + let raw = r#"{ + "methodResponses": [ + ["Email/query", {"ids": ["msg-1", "msg-2"]}, "c1"] + ] + }"#; + let ids = JmapEngine::parse_email_query_response(raw).expect("parse query"); + assert_eq!(ids, vec!["msg-1".to_string(), "msg-2".to_string()]); + } + + #[test] + fn parse_mailbox_get_response_extracts_mailboxes() { + let raw = r#"{ + "methodResponses": [ + ["Mailbox/get", {"list": [ + {"id": "mb-1", "name": "Inbox", "totalEmails": 5, "unreadEmails": 2} + ]}, "c1"] + ] + }"#; + let mailboxes = JmapEngine::parse_mailbox_get_response(raw).expect("parse mailboxes"); + assert_eq!(mailboxes.len(), 1); + assert_eq!(mailboxes[0].id, "mb-1"); + assert_eq!(mailboxes[0].total_emails, 5); + assert_eq!(mailboxes[0].unread_emails, 2); + } + + /// Regression test for the camelCase wire-format bug: `JmapEmail` must deserialize + /// `receivedAt`/`isUnread`/`bodySnippet` (the real JMAP wire property names) rather than + /// silently defaulting them to `None` because it only matched the Rust field names. + #[test] + fn jmap_email_deserializes_camel_case_wire_fields() { + let raw = r#"{ + "id": "msg-1", + "subject": "Test", + "from": [{"email": "a@nuncio.mx"}], + "to": [{"email": "b@nuncio.mx"}], + "receivedAt": 1700000000, + "isUnread": true, + "bodySnippet": "hello world" + }"#; + let email: JmapEmail = serde_json::from_str(raw).expect("parse camelCase JmapEmail"); + assert_eq!(email.received_at, Some(1700000000)); + assert_eq!(email.is_unread, Some(true)); + assert_eq!(email.body_snippet, Some("hello world".to_string())); + } + + #[tokio::test] + async fn uncredentialed_engine_keeps_static_fallback_behaviour() { + let engine = JmapEngine::new("acct-1"); + assert!(!engine.has_credentials()); + + let folders = engine.sync_folders().await.expect("sync folders"); + assert_eq!(folders.len(), 3); + assert_eq!(folders[0].id, "inbox"); + + let (emails, state) = engine + .sync_messages("inbox", None) + .await + .expect("sync messages"); + assert_eq!(state, "s-100"); + assert_eq!(emails.len(), 1); + assert_eq!(emails[0].id, "jmap-msg-1"); + } } diff --git a/crates/nuncio-mail/tests/wiremock_jmap_test.rs b/crates/nuncio-mail/tests/wiremock_jmap_test.rs index 4364d2a..fabf20e 100644 --- a/crates/nuncio-mail/tests/wiremock_jmap_test.rs +++ b/crates/nuncio-mail/tests/wiremock_jmap_test.rs @@ -2,7 +2,7 @@ #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] -use nuncio_mail::JmapEngine; +use nuncio_mail::{JmapEngine, MailBackend}; use serde_json::json; use wiremock::matchers::{method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; @@ -89,3 +89,151 @@ async fn wiremock_jmap_session_discovery_and_email_get_sync() { assert_eq!(emails[0].sender, "sender@nuncio.mx"); assert_eq!(new_state, "sync-state-500"); } + +/// Drives `JmapEngine` through the `MailBackend` trait itself (rather than its static parse +/// helpers) against a wiremock server, proving `with_credentials` performs genuine session +/// discovery, `Mailbox/get`, `Email/query`, and `Email/get` HTTP round trips end to end -- the +/// exact seam `nunciod::sync::build_mail_backend` wires into production. +#[tokio::test] +async fn jmap_engine_sync_folders_and_messages_through_mail_backend_trait() { + let mock_server = MockServer::start().await; + + let session_body = json!({ + "username": "james.maes@kof22.com", + "primaryAccounts": { + "urn:ietf:params:jmap:mail": "acct-100" + }, + "apiUrl": format!("{}/jmap/api", mock_server.uri()), + "state": "session-state-1" + }); + Mock::given(method("GET")) + .and(path("/.well-known/jmap")) + .respond_with(ResponseTemplate::new(200).set_body_json(&session_body)) + .mount(&mock_server) + .await; + + let mailbox_get_body = json!({ + "methodResponses": [ + [ + "Mailbox/get", + { + "accountId": "acct-100", + "list": [ + {"id": "mb-inbox", "name": "Inbox", "totalEmails": 3, "unreadEmails": 1} + ] + }, + "c1" + ] + ] + }); + Mock::given(method("POST")) + .and(path("/jmap/api")) + .and(wiremock::matchers::body_partial_json(json!({ + "methodCalls": [["Mailbox/get", {}, "c1"]] + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(&mailbox_get_body)) + .mount(&mock_server) + .await; + + let email_query_body = json!({ + "methodResponses": [ + ["Email/query", {"ids": ["msg-wm-200"]}, "c1"] + ] + }); + Mock::given(method("POST")) + .and(path("/jmap/api")) + .and(wiremock::matchers::body_partial_json(json!({ + "methodCalls": [["Email/query", {}, "c1"]] + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(&email_query_body)) + .mount(&mock_server) + .await; + + let email_get_body = json!({ + "methodResponses": [ + [ + "Email/get", + { + "accountId": "acct-100", + "state": "sync-state-900", + "list": [ + { + "id": "msg-wm-200", + "subject": "Trait-Level JMAP Sync", + "from": [{ "email": "sender@nuncio.mx" }], + "to": [{ "email": "james.maes@kof22.com" }], + "receivedAt": 1700002000i64, + "isUnread": true, + "bodySnippet": "Fetched through the MailBackend trait." + } + ] + }, + "c1" + ] + ] + }); + Mock::given(method("POST")) + .and(path("/jmap/api")) + .and(wiremock::matchers::body_partial_json(json!({ + "methodCalls": [["Email/get", {}, "c1"]] + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(&email_get_body)) + .mount(&mock_server) + .await; + + let engine = JmapEngine::with_credentials( + "acct-100", + &mock_server.uri(), + "james.maes@kof22.com", + "wiremock-password", + ); + + let folders = engine.sync_folders().await.expect("sync_folders succeeds"); + assert_eq!(folders.len(), 1); + assert_eq!(folders[0].id, "mb-inbox"); + assert_eq!(folders[0].name, "Inbox"); + assert_eq!(folders[0].total_messages, 3); + assert_eq!(folders[0].unread_messages, 1); + + let (emails, new_state) = engine + .sync_messages("mb-inbox", None) + .await + .expect("sync_messages succeeds"); + assert_eq!(emails.len(), 1); + assert_eq!(emails[0].id, "msg-wm-200"); + assert_eq!(emails[0].subject, "Trait-Level JMAP Sync"); + assert_eq!(emails[0].sender, "sender@nuncio.mx"); + assert_eq!(emails[0].received_at, 1700002000); + assert!(!emails[0].read); + assert_eq!( + emails[0].body_plain, + Some("Fetched through the MailBackend trait.".to_string()) + ); + assert_eq!(new_state, "sync-state-900"); +} + +/// A session-discovery failure (non-2xx status) must surface as a genuine `MailError`, never a +/// silent fallback to canned data, once credentials are present. +#[tokio::test] +async fn jmap_engine_sync_folders_surfaces_session_discovery_failure() { + let mock_server = MockServer::start().await; + + Mock::given(method("GET")) + .and(path("/.well-known/jmap")) + .respond_with(ResponseTemplate::new(500)) + .mount(&mock_server) + .await; + + let engine = JmapEngine::with_credentials( + "acct-100", + &mock_server.uri(), + "james.maes@kof22.com", + "wiremock-password", + ); + + let err = engine + .sync_folders() + .await + .expect_err("session discovery failure must not fall back to canned folders"); + assert!(err.to_string().contains("network I/O error")); +} diff --git a/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin b/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin index 0494ff3..a19a85f 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 a4027ec..0339047 100644 --- a/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto +++ b/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto @@ -458,6 +458,34 @@ service Filters { // existing dry-run evaluation path. Never persists the rule and never // executes any of its actions. rpc PreviewRule(PreviewRuleRequest) returns (PreviewRuleResponse); + // UpdateRule applies partial overrides (name/nsql/priority) to an existing + // persisted rule, re-validates the resulting rule exactly as `CreateRule` + // does, and reloads the daemon's live `FilterEngine` rule set so the + // change takes effect immediately. Rejects an unknown `id` with + // `Status::not_found` and an invalid resulting rule with + // `Status::invalid_argument` -- nothing is persisted on either failure. + rpc UpdateRule(UpdateRuleRequest) returns (UpdateRuleResponse); + // ExportRules renders every persisted filter rule in the requested + // `format` ("sql" or "json") and returns the rendered text inline -- + // small enough, unlike a mailbox export, to carry directly over the RPC + // rather than writing to a shared filesystem path. + rpc ExportRules(ExportRulesRequest) returns (ExportRulesResponse); + // ImportRules parses each non-empty, non-comment line of `content` as an + // independent NSQL rule definition and persists the ones that parse. + // Reports both the count of rules actually persisted and the parse error + // for every line that failed, rather than silently dropping failures. + rpc ImportRules(ImportRulesRequest) returns (ImportRulesResponse); + // GetExecutionLogs returns the most recent persisted filter execution + // log entries (hash-chain ledger), newest-eligible-first as stored, + // capped at `limit`. + rpc GetExecutionLogs(GetExecutionLogsRequest) returns (GetExecutionLogsResponse); + // Triage scans every stored message in bounded chunks, evaluating the + // daemon's live `FilterEngine` against each one and applying the same + // actions live sync would -- so the currently active rule set can be + // applied retroactively to mail that was synced before those rules + // existed. Streams cumulative progress as it works, ending with a final + // `done: true` message once the whole store has been scanned. + rpc Triage(TriageRequest) returns (stream TriageProgress); } // FilterRule mirrors `nuncio_filter::FilterRule`'s persisted/user-facing @@ -577,6 +605,113 @@ message PreviewRuleResponse { repeated string condition_traces = 7; } +// UpdateRuleRequest carries a persisted rule's `id` plus optional +// per-field overrides; an absent field leaves that field unchanged on the +// existing rule. +message UpdateRuleRequest { + // Rule unique identifier. + string id = 1; + // New display name, if changing. + optional string name = 2; + // New full NSQL source code text, if changing. + optional string nsql = 3; + // New evaluation order, if changing. + optional int32 priority = 4; +} + +// UpdateRuleResponse carries the rule as persisted after applying the +// requested overrides. +message UpdateRuleResponse { + FilterRule rule = 1; +} + +// ExportRulesRequest names the rendering format for `ExportRules`. +message ExportRulesRequest { + // "sql" or "json". + string format = 1; +} + +// ExportRulesResponse carries every persisted rule pre-rendered as a single +// text blob in the requested format. +message ExportRulesResponse { + string content = 1; +} + +// ImportRulesRequest carries raw rule-definition text to parse and persist, +// one NSQL rule definition per non-empty, non-comment (`--`-prefixed) line. +message ImportRulesRequest { + string content = 1; +} + +// ImportRulesResponse reports how many lines parsed and persisted +// successfully, and the parse error for every line that did not. +message ImportRulesResponse { + // Count of rule definitions successfully parsed and persisted. + int32 imported_count = 1; + // Parse error message for each line that failed, in encounter order. + repeated string errors = 2; +} + +// GetExecutionLogsRequest caps the number of log entries returned. +message GetExecutionLogsRequest { + uint32 limit = 1; +} + +// GetExecutionLogsResponse. +message GetExecutionLogsResponse { + repeated FilterExecutionLog logs = 1; +} + +// FilterExecutionLog mirrors `nuncio_filter::FilterExecutionLog` +// field-for-field -- the hash-chain ledger entry recorded each time a +// persisted rule matches and acts on a message. +message FilterExecutionLog { + // Log entry primary key. + int64 id = 1; + // Rule ID executed. + string rule_id = 2; + // Email message ID evaluated. + string message_id = 3; + // Action description string. + string action_taken = 4; + // Timestamp of execution (unix seconds). + int64 matched_at = 5; + // Hash of the preceding log record in the hash-chain ledger. + string prev_hash = 6; + // HMAC-SHA256 hash of this record. + string hash = 7; +} + +// TriageRequest configures a retroactive rescan of the whole message store. +message TriageRequest { + // Scope the scan to a single rule's actions. Currently NOT honored -- + // `Triage` always evaluates a message against the full live rule set, + // matching what a real (non-retroactive) sync run would do. Reserved for + // forward compatibility; a client that sets this today gets a whole-store + // scan against all active rules, not an error. + optional string rule_id = 1; + // Number of messages fetched per keyset page. 0 selects the server + // default (100). + uint32 chunk_size = 2; +} + +// TriageProgress reports the real, cumulative state of an in-progress (or +// just-finished) `Triage` scan. Every field is the true running total as of +// the chunk just processed -- never an estimate or a fabricated count. +message TriageProgress { + // Total messages scanned so far. + uint64 scanned_count = 1; + // Total messages that had at least one action applied or enqueued. + uint64 matched_count = 2; + // Total individual actions applied or enqueued across all messages. + uint64 actions_applied_count = 3; + // ID of the last message scanned in the most recently processed chunk. + string last_message_id = 4; + // True only on the final message of the stream, once the whole store has + // been scanned. + bool done = 5; +} + // Export exposes the daemon's real mailbox export engine // (`nuncio_core::export`, via `DatabaseEngine::export_messages_to_file`) // over authenticated gRPC. The daemon and its caller run on the SAME host diff --git a/crates/nuncio-store/src/db.rs b/crates/nuncio-store/src/db.rs index b280346..58b1b15 100644 --- a/crates/nuncio-store/src/db.rs +++ b/crates/nuncio-store/src/db.rs @@ -67,6 +67,17 @@ impl DatabaseError { _ => false, } } + + /// Returns true if this error means "no row matched the lookup" (e.g. a + /// `get_message`/`get_contact`/`get_calendar_event` call for an id that + /// was never saved), as opposed to a genuine I/O, corruption, or key + /// provisioning failure. + /// + /// Exposed so callers outside this crate can distinguish "not found" from + /// other failures without depending on `sqlx` directly. + pub fn is_not_found(&self) -> bool { + matches!(self, DatabaseError::Query(sqlx::Error::RowNotFound)) + } } /// SQLite database storage engine managing WAL connection pools and migrations. diff --git a/crates/nunciod/Cargo.toml b/crates/nunciod/Cargo.toml index 8a47d69..663ac49 100644 --- a/crates/nunciod/Cargo.toml +++ b/crates/nunciod/Cargo.toml @@ -37,3 +37,4 @@ bytes.workspace = true [dev-dependencies] nuncio-cli = { path = "../nuncio-cli" } sqlx.workspace = true +wiremock.workspace = true diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs index 487c286..bab507a 100644 --- a/crates/nunciod/src/grpc.rs +++ b/crates/nunciod/src/grpc.rs @@ -27,18 +27,22 @@ use nuncio_proto::v1::{ ContactEmail as ContactEmailProto, ContactPhone as ContactPhoneProto, ContactsSyncRequest, ContactsSyncResponse, CreateContactRequest, CreateContactResponse, CreateRuleRequest, CreateRuleResponse, DatabaseRecovered, DeleteRuleRequest, DeleteRuleResponse, Event, - EventError, ExportFormat as ExportFormatProto, ExportRequest, ExportResponse, FilterExecuted, - FilterRule as FilterRuleProto, Folder as FolderProto, GetContactRequest, GetContactResponse, - GetEventRequest, GetEventResponse, GetMessageRequest, GetMessageResponse, GetStatusRequest, - GetStatusResponse, ListAccountsRequest, ListAccountsResponse, ListContactsRequest, + 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, ImportRulesRequest, + ImportRulesResponse, ListAccountsRequest, ListAccountsResponse, ListContactsRequest, ListContactsResponse, ListEventsRequest, ListEventsResponse, ListFoldersRequest, ListFoldersResponse, ListMessagesRequest, ListMessagesResponse, ListRecordsRequest, ListRecordsResponse, ListRulesRequest, ListRulesResponse, MarkReadRequest, MarkReadResponse, Message as MessageProto, MessageFlagsChanged, MessageSearchHit, PreviewRuleRequest, PreviewRuleResponse, SearchMessagesRequest, SearchMessagesResponse, SendMessageRequest, SendMessageResponse, ShuttingDown, SubscribeRequest, SyncCompleted, SyncRequest, SyncResponse, - SyncStarted, TlsMode as TlsModeProto, UpdateAvailable, ValidateRuleRequest, - ValidateRuleResponse, VerifyChainRequest, VerifyChainResponse, + SyncStarted, TlsMode as TlsModeProto, TriageProgress, TriageRequest, UpdateAvailable, + UpdateRuleRequest, UpdateRuleResponse, ValidateRuleRequest, ValidateRuleResponse, + VerifyChainRequest, VerifyChainResponse, }; use nuncio_store::db::DatabaseEngine; use nuncio_store::search::SearchEngine; @@ -50,7 +54,7 @@ use subtle::ConstantTimeEq; use thiserror::Error; use tokio::net::TcpListener; use tokio_stream::wrappers::errors::BroadcastStreamRecvError; -use tokio_stream::wrappers::{BroadcastStream, TcpListenerStream}; +use tokio_stream::wrappers::{BroadcastStream, ReceiverStream, TcpListenerStream}; use tokio_stream::{Stream, StreamExt}; use tonic::transport::Server; use tonic::{Request, Response, Status}; @@ -865,6 +869,7 @@ struct MailGrpcService { db: Arc, event_bus: Arc, secrets: Arc, + filter_engine: Arc, overrides: MailEngineOverrides, } @@ -1053,15 +1058,33 @@ impl Mail for MailGrpcService { let account_id = request.into_inner().account_id; let synced = if let Some(backend) = &self.overrides.mail_backend { - crate::sync::sync_with_backend(&self.db, &self.event_bus, backend.as_ref(), account_id) - .await - .map_err(|e| Status::internal(format!("sync failed: {e}")))? + crate::sync::sync_with_backend( + &self.db, + &self.event_bus, + backend.as_ref(), + &self.filter_engine, + account_id, + ) + .await + .map_err(|e| Status::internal(format!("sync failed: {e}")))? } else if let Some(account_id) = account_id { - crate::sync::run_account_sync(&self.db, &self.secrets, &self.event_bus, &account_id) - .await - .map_err(|e| Status::internal(format!("sync failed: {e}")))? + crate::sync::run_account_sync( + &self.db, + &self.secrets, + &self.event_bus, + &self.filter_engine, + &account_id, + ) + .await + .map_err(|e| Status::internal(format!("sync failed: {e}")))? } else { - crate::sync::run_all_accounts_sync(&self.db, &self.secrets, &self.event_bus).await + crate::sync::run_all_accounts_sync( + &self.db, + &self.secrets, + &self.event_bus, + &self.filter_engine, + ) + .await }; Ok(Response::new(SyncResponse { @@ -1089,6 +1112,22 @@ fn map_filter_rule_to_proto(rule: nuncio_filter::FilterRule) -> FilterRuleProto } } +/// Maps a `nuncio_filter::FilterExecutionLog` onto its wire-format +/// `nuncio.v1.FilterExecutionLog` representation. +fn map_filter_execution_log_to_proto( + log: nuncio_filter::FilterExecutionLog, +) -> FilterExecutionLogProto { + FilterExecutionLogProto { + id: log.id, + rule_id: log.rule_id, + message_id: log.message_id, + action_taken: log.action_taken, + matched_at: log.matched_at, + prev_hash: log.prev_hash, + hash: log.hash, + } +} + /// Maps a `nuncio_filter::FilterPreviewResult` onto its wire-format /// `nuncio.v1.PreviewRuleResponse` representation. fn map_preview_result_to_proto(preview: nuncio_filter::FilterPreviewResult) -> PreviewRuleResponse { @@ -1296,6 +1335,240 @@ impl Filters for FiltersGrpcService { let preview = preview_engine.preview(&email); Ok(Response::new(map_preview_result_to_proto(preview))) } + + /// Applies partial overrides to an already-persisted rule, re-validates + /// the resulting rule with the same checks `create_rule` runs, and + /// reloads the live `FilterEngine` -- mirroring the store-then-reload + /// order every other mutating `Filters` RPC uses. There is no + /// single-rule store getter, so the existing rule is located the same + /// way the (now-retired) CLI-local implementation did: list then find + /// by `id`. + async fn update_rule( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + + let existing = self + .db + .list_filter_rules() + .await + .map_err(|e| Status::internal(format!("failed to list filter rules: {e}")))? + .into_iter() + .find(|r| r.id == req.id) + .ok_or_else(|| Status::not_found(format!("filter rule '{}' not found", req.id)))?; + + let name = req.name.unwrap_or(existing.name); + let nsql = req.nsql.unwrap_or(existing.nsql_text); + let priority = req.priority.unwrap_or(existing.priority); + + let mut rule = NsqlParser::parse_rule(name, priority, &nsql) + .map_err(|e| Status::invalid_argument(format!("NSQL syntax error: {e}")))?; + + NsqlValidator::validate(&rule, &ValidationOptions::default()) + .map_err(|e| Status::invalid_argument(format!("NSQL validation error: {e}")))?; + + rule.id = existing.id; + + self.db + .save_filter_rule(&rule) + .await + .map_err(|e| Status::internal(format!("failed to persist filter rule: {e}")))?; + + self.reload_engine_from_store().await; + + Ok(Response::new(UpdateRuleResponse { + rule: Some(map_filter_rule_to_proto(rule)), + })) + } + + /// Renders every persisted rule as either lossless NSQL source text + /// (`FilterRule::to_nsql`, one rule per line) or a JSON array, matching + /// the exact two renderings the (now-retired) CLI-local `filter export` + /// implementation produced. + async fn export_rules( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + + let rules = self + .db + .list_filter_rules() + .await + .map_err(|e| Status::internal(format!("failed to list filter rules: {e}")))?; + + let content = if req.format == "json" { + serde_json::to_string_pretty(&rules) + .map_err(|e| Status::internal(format!("failed to render rules as JSON: {e}")))? + } else { + rules + .iter() + .map(|r| r.to_nsql()) + .collect::>() + .join("\n") + }; + + Ok(Response::new(ExportRulesResponse { content })) + } + + /// Parses `content` one non-empty, non-comment (`--`-prefixed) line at a + /// time, persisting every rule that parses and collecting the parse + /// error for every line that does not -- unlike the (now-retired) + /// CLI-local implementation, a parse failure is never silently dropped. + /// The live `FilterEngine` is reloaded once at the end rather than once + /// per successfully imported line. + async fn import_rules( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + + let mut imported_count = 0i32; + let mut errors = Vec::new(); + + for line in req.content.lines() { + let line_trim = line.trim(); + if line_trim.is_empty() || line_trim.starts_with("--") { + continue; + } + + match NsqlParser::parse_rule( + format!("Imported Rule {}", imported_count + 1), + 0, + line_trim, + ) { + Ok(rule) => match self.db.save_filter_rule(&rule).await { + Ok(()) => imported_count += 1, + Err(e) => errors.push(format!("failed to persist '{line_trim}': {e}")), + }, + Err(e) => errors.push(format!("failed to parse '{line_trim}': {e}")), + } + } + + self.reload_engine_from_store().await; + + Ok(Response::new(ImportRulesResponse { + imported_count, + errors, + })) + } + + /// Returns the persisted filter execution log ledger, capped at + /// `limit`, in the store's natural order. + async fn get_execution_logs( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + + let logs = self + .db + .list_filter_execution_logs(req.limit as usize) + .await + .map_err(|e| Status::internal(format!("failed to list execution logs: {e}")))? + .into_iter() + .map(map_filter_execution_log_to_proto) + .collect(); + + Ok(Response::new(GetExecutionLogsResponse { logs })) + } + + /// Server-streaming retroactive rescan of the whole message store. + /// Unlike `System::subscribe` (which wraps an already-live broadcast), + /// this RPC actively drives the scan: a spawned worker task walks + /// `db.get_message_chunk` in keyset pages, applies the shared + /// [`crate::sync::apply_filter_actions`] evaluate-and-route logic to + /// each message (the SAME side effects live sync produces), and pushes + /// cumulative progress onto an mpsc channel that this method wraps as + /// the returned stream. + /// + /// `rule_id` is accepted for forward compatibility but NOT currently + /// honored: `apply_filter_actions` always evaluates against the whole + /// live rule set, matching what a real sync pass would do, rather than + /// silently narrowing to one rule. + /// + /// A `get_message_chunk` failure sends a single `Err(Status::internal)` + /// on the stream and stops the scan -- never a fabricated `done: true` + /// on error. The receiver disconnecting (the caller dropped the call) + /// also stops the worker rather than continuing to scan for no one. + type TriageStream = + Pin> + Send + 'static>>; + + async fn triage( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let chunk_size = if req.chunk_size == 0 { + 100 + } else { + req.chunk_size as usize + }; + + let (tx, rx) = tokio::sync::mpsc::channel::>(16); + let db = self.db.clone(); + let engine = self.filter_engine.clone(); + + tokio::spawn(async move { + let mut last_id = String::new(); + let mut scanned: u64 = 0; + let mut matched: u64 = 0; + let mut applied: u64 = 0; + + loop { + let batch = match db.get_message_chunk(&last_id, chunk_size).await { + Ok(batch) => batch, + Err(e) => { + let _ = tx + .send(Err(Status::internal(format!( + "failed to read message chunk during triage: {e}" + )))) + .await; + return; + } + }; + let is_last_page = batch.len() < chunk_size; + + for email in &batch { + let actions_for_email = + crate::sync::apply_filter_actions(&db, &engine, email).await; + scanned += 1; + applied += actions_for_email as u64; + if actions_for_email > 0 { + matched += 1; + } + } + + if let Some(last) = batch.last() { + last_id = last.id.clone(); + } + + let done = is_last_page; + if tx + .send(Ok(TriageProgress { + scanned_count: scanned, + matched_count: matched, + actions_applied_count: applied, + last_message_id: last_id.clone(), + done, + })) + .await + .is_err() + { + // Receiver dropped: the caller disconnected, so there is + // no one left to report progress to. + return; + } + + if is_last_page { + return; + } + } + }); + + Ok(Response::new(Box::pin(ReceiverStream::new(rx)))) + } } /// Maps a wire-format `nuncio.v1.ExportFormat` enum value back onto @@ -1607,6 +1880,7 @@ pub async fn serve_on_listener_with_overrides( db: db.clone(), event_bus, secrets: secrets.clone(), + filter_engine: filter_engine.clone(), overrides, }; let mail_interceptor = BearerAuthInterceptor::new(token.clone()); @@ -3128,7 +3402,8 @@ mod tests { use nuncio_proto::v1::filters_client::FiltersClient; use nuncio_proto::v1::{ - CreateRuleRequest, DeleteRuleRequest, ListRulesRequest, PreviewRuleRequest, + CreateRuleRequest, DeleteRuleRequest, ExportRulesRequest, GetExecutionLogsRequest, + ImportRulesRequest, ListRulesRequest, PreviewRuleRequest, UpdateRuleRequest, ValidateRuleRequest, }; @@ -3223,6 +3498,39 @@ mod tests { .await .expect_err("missing bearer token must be rejected"); assert_eq!(err.code(), Code::Unauthenticated); + + let err = client + .update_rule(UpdateRuleRequest { + id: "rule-1".to_string(), + name: None, + nsql: None, + priority: None, + }) + .await + .expect_err("missing bearer token must be rejected"); + assert_eq!(err.code(), Code::Unauthenticated); + + let err = client + .export_rules(ExportRulesRequest { + format: "sql".to_string(), + }) + .await + .expect_err("missing bearer token must be rejected"); + assert_eq!(err.code(), Code::Unauthenticated); + + let err = client + .import_rules(ImportRulesRequest { + content: SAMPLE_RULE_NSQL.to_string(), + }) + .await + .expect_err("missing bearer token must be rejected"); + assert_eq!(err.code(), Code::Unauthenticated); + + let err = client + .get_execution_logs(GetExecutionLogsRequest { limit: 10 }) + .await + .expect_err("missing bearer token must be rejected"); + assert_eq!(err.code(), Code::Unauthenticated); } /// End-to-end proof: `CreateRule` persists AND reloads the live @@ -3534,6 +3842,259 @@ mod tests { assert_eq!(err.code(), Code::InvalidArgument); } + /// `UpdateRule` happy path: applies a partial override, preserves the + /// original `id`, persists, AND reloads the live `FilterEngine` -- the + /// same proof pattern `create_list_and_delete_rule_round_trip...` uses. + #[tokio::test] + async fn update_rule_applies_partial_overrides_and_reloads_the_live_engine() { + let (addr, _handle, _db, filter_engine, _dir) = + spawn_filters_test_server("correct-token").await; + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let created = client + .create_rule(authed_bearer_request(CreateRuleRequest { + name: "Original Name".to_string(), + nsql: SAMPLE_RULE_NSQL.to_string(), + priority: 5, + })) + .await + .expect("create_rule succeeds") + .into_inner() + .rule + .expect("response carries the rule"); + let rule_id = created.id.clone(); + + let updated = client + .update_rule(authed_bearer_request(UpdateRuleRequest { + id: rule_id.clone(), + name: Some("Renamed".to_string()), + nsql: None, + priority: Some(1), + })) + .await + .expect("update_rule succeeds") + .into_inner() + .rule + .expect("response carries the updated rule"); + + assert_eq!(updated.id, rule_id, "id must be preserved across an edit"); + assert_eq!(updated.name, "Renamed"); + assert_eq!(updated.priority, 1); + assert_eq!( + updated.nsql_text, SAMPLE_RULE_NSQL, + "an omitted field must retain its previous value" + ); + + let list_response = client + .list_rules(authed_bearer_request(ListRulesRequest {})) + .await + .expect("list_rules succeeds") + .into_inner(); + assert_eq!(list_response.rules.len(), 1); + assert_eq!(list_response.rules[0].name, "Renamed"); + + // The live `FilterEngine` was reloaded with the updated rule. + let matches = filter_engine.evaluate(&nuncio_core::model::Email { + id: "msg-1".to_string(), + account_id: "acct-1".to_string(), + folder_id: "inbox".to_string(), + subject: "Urgent Meeting".to_string(), + sender: "a@b.com".to_string(), + recipient: "c@d.com".to_string(), + received_at: 0, + read: false, + body_plain: None, + body_html: None, + attachments: Vec::new(), + }); + assert_eq!(matches.len(), 1); + assert_eq!(matches[0].0.id, rule_id); + } + + #[tokio::test] + async fn update_rule_rejects_unknown_id() { + let (addr, _handle, _db, _engine, _dir) = spawn_filters_test_server("correct-token").await; + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let err = client + .update_rule(authed_bearer_request(UpdateRuleRequest { + id: "no-such-rule".to_string(), + name: Some("Whatever".to_string()), + nsql: None, + priority: None, + })) + .await + .expect_err("unknown id must be rejected"); + assert_eq!(err.code(), Code::NotFound); + } + + #[tokio::test] + async fn update_rule_rejects_invalid_nsql_and_persists_nothing() { + let (addr, _handle, _db, _engine, _dir) = spawn_filters_test_server("correct-token").await; + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let created = client + .create_rule(authed_bearer_request(CreateRuleRequest { + name: "Original Name".to_string(), + nsql: SAMPLE_RULE_NSQL.to_string(), + priority: 5, + })) + .await + .expect("create_rule succeeds") + .into_inner() + .rule + .expect("response carries the rule"); + + let err = client + .update_rule(authed_bearer_request(UpdateRuleRequest { + id: created.id, + name: None, + nsql: Some("THIS IS NOT VALID NSQL AT ALL {{{".to_string()), + priority: None, + })) + .await + .expect_err("invalid NSQL must be rejected"); + assert_eq!(err.code(), Code::InvalidArgument); + + let list_response = client + .list_rules(authed_bearer_request(ListRulesRequest {})) + .await + .expect("list_rules succeeds") + .into_inner(); + assert_eq!( + list_response.rules[0].nsql_text, SAMPLE_RULE_NSQL, + "a failed edit must never overwrite the persisted rule" + ); + } + + /// `ExportRules` renders every persisted rule as either lossless NSQL + /// text or a JSON array, reflecting the daemon's real, persistent + /// store -- not any ephemeral client-side state. + #[tokio::test] + async fn export_rules_renders_sql_and_json() { + let (addr, _handle, _db, _engine, _dir) = spawn_filters_test_server("correct-token").await; + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + client + .create_rule(authed_bearer_request(CreateRuleRequest { + name: "Export Me".to_string(), + nsql: SAMPLE_RULE_NSQL.to_string(), + priority: 0, + })) + .await + .expect("create_rule succeeds"); + + let sql_export = client + .export_rules(authed_bearer_request(ExportRulesRequest { + format: "sql".to_string(), + })) + .await + .expect("export_rules succeeds") + .into_inner(); + assert!(sql_export.content.contains("SELECT * FROM emails")); + assert!(sql_export.content.contains("Urgent")); + + let json_export = client + .export_rules(authed_bearer_request(ExportRulesRequest { + format: "json".to_string(), + })) + .await + .expect("export_rules succeeds") + .into_inner(); + assert!(json_export.content.contains("\"name\"")); + assert!(json_export.content.contains("Export Me")); + } + + /// `ImportRules` persists every line that parses AND reports the parse + /// error for every line that does not -- proving failures are surfaced + /// rather than silently dropped. + #[tokio::test] + async fn import_rules_reports_partial_failures_and_persists_valid_lines() { + let (addr, _handle, _db, filter_engine, _dir) = + spawn_filters_test_server("correct-token").await; + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let content = format!( + "-- a comment line, skipped\n\n{SAMPLE_RULE_NSQL}\nTHIS IS NOT VALID NSQL AT ALL {{{{{{\n" + ); + + let response = client + .import_rules(authed_bearer_request(ImportRulesRequest { content })) + .await + .expect("import_rules call itself succeeds") + .into_inner(); + + assert_eq!(response.imported_count, 1); + assert_eq!(response.errors.len(), 1); + + let list_response = client + .list_rules(authed_bearer_request(ListRulesRequest {})) + .await + .expect("list_rules succeeds") + .into_inner(); + assert_eq!(list_response.rules.len(), 1); + assert_eq!(list_response.rules[0].nsql_text, SAMPLE_RULE_NSQL); + + // The live engine was reloaded to include the imported rule. + let matches = filter_engine.evaluate(&nuncio_core::model::Email { + id: "msg-1".to_string(), + account_id: "acct-1".to_string(), + folder_id: "inbox".to_string(), + subject: "Urgent Meeting".to_string(), + sender: "a@b.com".to_string(), + recipient: "c@d.com".to_string(), + received_at: 0, + read: false, + body_plain: None, + body_html: None, + attachments: Vec::new(), + }); + assert_eq!(matches.len(), 1); + } + + /// `GetExecutionLogs` returns the persisted execution log ledger, + /// proving `filter logs` (via this RPC) now sees the SAME persistent + /// store that `filter create` (via `CreateRule`) writes into -- closing + /// the exact gap that motivated moving these operations onto `Filters`. + #[tokio::test] + async fn get_execution_logs_returns_the_persisted_ledger() { + let (addr, _handle, db, _engine, _dir) = spawn_filters_test_server("correct-token").await; + + db.save_filter_execution_log("rule-1", "msg-1", "MARK READ") + .await + .expect("seed an execution log entry"); + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let response = client + .get_execution_logs(authed_bearer_request(GetExecutionLogsRequest { limit: 10 })) + .await + .expect("get_execution_logs succeeds") + .into_inner(); + + assert_eq!(response.logs.len(), 1); + assert_eq!(response.logs[0].rule_id, "rule-1"); + assert_eq!(response.logs[0].message_id, "msg-1"); + assert_eq!(response.logs[0].action_taken, "MARK READ"); + } + // ---- Export ---- use nuncio_proto::v1::export_client::ExportClient; diff --git a/crates/nunciod/src/main.rs b/crates/nunciod/src/main.rs index 833d930..ecd4543 100644 --- a/crates/nunciod/src/main.rs +++ b/crates/nunciod/src/main.rs @@ -54,6 +54,7 @@ async fn main() -> Result<(), Box> { let db_sync = db.clone(); let secrets_sync = account_secrets.clone(); let event_bus_sync = event_bus.clone(); + let filter_engine_sync = filter_engine.clone(); let _sync_command_task = tokio::spawn(async move { while let Some(cmd) = command_rx.recv().await { match cmd { @@ -62,6 +63,7 @@ async fn main() -> Result<(), Box> { &db_sync, &secrets_sync, &event_bus_sync, + &filter_engine_sync, ) .await; tracing::info!("SyncAll completed: {} message(s) synced", synced); @@ -71,6 +73,7 @@ async fn main() -> Result<(), Box> { &db_sync, &secrets_sync, &event_bus_sync, + &filter_engine_sync, &account_id, ) .await diff --git a/crates/nunciod/src/sync.rs b/crates/nunciod/src/sync.rs index f9bc567..d260505 100644 --- a/crates/nunciod/src/sync.rs +++ b/crates/nunciod/src/sync.rs @@ -19,6 +19,7 @@ //! real backend; they then delegate the actual fetch/persist/event-emission //! work to [`sync_with_backend`]. use nuncio_core::{AccountConfig, AccountProtocol, CoreCommand, EventBus}; +use nuncio_filter::{FilterEngine, OutboxManager, RuleAction}; use nuncio_mail::{ImapEngine, JmapEngine, MailBackend, MailError}; use nuncio_store::db::DatabaseError; use nuncio_store::vault::{SecretManager, VaultError}; @@ -41,12 +42,193 @@ pub enum SyncError { Store(#[from] DatabaseError), } +/// Map a [`RuleAction`] to the short mutation tag persisted on +/// [`nuncio_filter::PendingRemoteMutation::mutation_type`] and the outbox +/// target parameter, for every action that is not applied immediately. +/// +/// Nothing downstream currently parses `mutation_type` back into a +/// `RuleAction`, so these tags are a fixed vocabulary owned by this module; +/// keep them stable once the outbox worker starts consuming them. +fn remote_action_tag(action: &RuleAction) -> Option<(&'static str, Option)> { + match action { + RuleAction::MoveTo(folder) => Some(("MOVE", Some(folder.clone()))), + RuleAction::CopyTo(folder) => Some(("COPY", Some(folder.clone()))), + RuleAction::Flag => Some(("FLAG", None)), + RuleAction::Unflag => Some(("UNFLAG", None)), + RuleAction::Delete => Some(("DELETE", None)), + RuleAction::ForwardTo(address) => Some(("FORWARD", Some(address.clone()))), + RuleAction::CallWebhook(url) => Some(("WEBHOOK", Some(url.clone()))), + RuleAction::MarkRead | RuleAction::MarkUnread => None, + } +} + +/// Evaluate `email` against `filter_engine` and route every matched rule's +/// actions to their real, persisted effect. +/// +/// Filters fire EXACTLY ONCE per newly-arrived message: callers (see +/// [`fetch_and_persist`]) must only invoke this for a message that was +/// genuinely new to the store on this sync pass, never for one that already +/// existed. Sync is not yet incremental -- every sync re-fetches and +/// re-persists messages the backend still reports -- so calling this +/// unconditionally on every persisted message would re-fire a rule's actions +/// on every single re-sync of the same mailbox: a fresh +/// `PendingRemoteMutation` and a fresh `FilterExecutionLog` entry per cycle +/// for a message that arrived once, growing the outbox and the audit ledger +/// without bound and (once a real mutation transport lands) duplicating the +/// real remote operation. This function itself has no way to tell "new" from +/// "already seen" -- that determination is the caller's responsibility. +/// +/// `MarkRead`/`MarkUnread` apply immediately to the stored message via +/// [`DatabaseEngine::set_message_read`]. Every other action enqueues a real +/// [`nuncio_filter::PendingRemoteMutation`] through the existing outbox +/// (`OutboxManager::create_mutation` + `DatabaseEngine::save_pending_mutation`) +/// for the background outbox worker to pick up -- this function never +/// performs the real remote IMAP/JMAP mutation or webhook call itself. +/// Every match also appends a [`nuncio_filter::FilterExecutionLog`] entry via +/// `DatabaseEngine::save_filter_execution_log`, so there is an audit trail of +/// what fired independent of whether the outbox item is ever drained. +/// +/// A bookkeeping failure on one action (a DB write error) is logged via +/// `tracing::warn!` and does not stop the remaining actions/rules from being +/// applied -- one broken write must never silently swallow the rest of a +/// sync's filtering. The return value counts only actions that were fully +/// applied AND logged; a partially-failed match (e.g. the mutation persists +/// but the execution log write fails) is not counted as success, keeping the +/// count honest for callers that log or assert on it. +/// +/// Standalone and reusable by design: this is the exact evaluate-and-route +/// logic other sync paths (e.g. a bulk retroactive triage over the whole +/// store) need, so it takes only a `DatabaseEngine`/`FilterEngine`/`Email` +/// and has no dependency on the live inbound-sync call chain. +pub async fn apply_filter_actions( + db: &nuncio_store::db::DatabaseEngine, + filter_engine: &FilterEngine, + email: &nuncio_core::model::Email, +) -> usize { + let mut applied = 0usize; + + for (rule, actions) in filter_engine.evaluate(email) { + for action in actions { + let immediate_ok = match &action { + RuleAction::MarkRead => match db.set_message_read(&email.id, true).await { + Ok(()) => true, + Err(e) => { + tracing::warn!( + "filter rule '{}' MARK READ failed for message '{}': {e}", + rule.id, + email.id + ); + false + } + }, + RuleAction::MarkUnread => match db.set_message_read(&email.id, false).await { + Ok(()) => true, + Err(e) => { + tracing::warn!( + "filter rule '{}' MARK UNREAD failed for message '{}': {e}", + rule.id, + email.id + ); + false + } + }, + RuleAction::MoveTo(_) + | RuleAction::CopyTo(_) + | RuleAction::Flag + | RuleAction::Unflag + | RuleAction::Delete + | RuleAction::ForwardTo(_) + | RuleAction::CallWebhook(_) => { + // Remote actions are handled below via the outbox; this + // arm only tracks whether the immediate-action write + // itself succeeded (there is none for a remote action, + // so it is unconditionally "not applicable" here). Listed + // explicitly rather than a wildcard so a future + // immediate-style action can't silently fall through to + // remote routing by accident. + true + } + }; + + if !immediate_ok { + continue; + } + + let mut action_ok = true; + if let Some((tag, target)) = remote_action_tag(&action) { + let mutation = OutboxManager::create_mutation(&rule.id, &email.id, tag, target); + if let Err(e) = db.save_pending_mutation(&mutation).await { + tracing::warn!( + "filter rule '{}' failed to enqueue outbox mutation for message '{}': {e}", + rule.id, + email.id + ); + action_ok = false; + } + } + + if !action_ok { + continue; + } + + let action_desc = action.to_nsql(); + if let Err(e) = db + .save_filter_execution_log(&rule.id, &email.id, &action_desc) + .await + { + tracing::warn!( + "filter rule '{}' matched message '{}' but failed to write execution log: {e}", + rule.id, + email.id + ); + continue; + } + + applied += 1; + } + } + + applied +} + +/// Reports whether `message_id` is not yet present in the store. +/// +/// Sync is not yet incremental (a re-sync re-fetches every message the +/// backend still reports, not just genuinely new ones), so this is the guard +/// that keeps filter actions firing exactly once per message: it must be +/// checked BEFORE `save_email` persists (or re-persists) the message, since +/// `save_email` is `INSERT OR REPLACE` and would otherwise erase the +/// distinction between "arriving for the first time" and "seen again". +/// +/// A lookup failure that is not "not found" (a genuine DB error) is logged +/// and treated as "not new", the conservative choice: firing filters again on +/// a message we can't positively identify as new risks duplicate remote +/// mutations, which is worse than occasionally missing a fire on a message +/// that was in fact new. +async fn is_new_message(db: &nuncio_store::db::DatabaseEngine, message_id: &str) -> bool { + match db.get_message(message_id).await { + Ok(_) => false, + Err(e) if e.is_not_found() => true, + Err(e) => { + tracing::warn!( + "could not determine whether message '{message_id}' is new before sync \ + (treating as not-new to avoid duplicate filter firing): {e}" + ); + false + } + } +} + /// Fetch every folder and every message in every folder from `backend`, -/// persisting each message via [`DatabaseEngine::save_email`]. Returns the -/// total number of messages processed (a message id "processed" more than -/// once, e.g. by a backend that returns the same id from multiple folders, -/// is counted once per occurrence -- `save_email` itself is `INSERT OR -/// REPLACE`, so storage stays deduplicated by message id regardless). +/// persisting each message via [`DatabaseEngine::save_email`] and, for +/// messages genuinely new to the store, evaluating it against +/// `filter_engine` via [`apply_filter_actions`] (see [`is_new_message`] for +/// why this must be gated rather than run on every persisted message). +/// Returns the total number of messages processed (a message id "processed" +/// more than once, e.g. by a backend that returns the same id from multiple +/// folders, is counted once per occurrence -- `save_email` itself is +/// `INSERT OR REPLACE`, so storage stays deduplicated by message id +/// regardless). /// /// Contains no event-bus or credential logic -- pure fetch-and-persist, so /// it is the smallest unit tests can exercise directly with a @@ -54,13 +236,18 @@ pub enum SyncError { async fn fetch_and_persist( db: &nuncio_store::db::DatabaseEngine, backend: &dyn MailBackend, + filter_engine: &FilterEngine, ) -> Result { let folders = backend.sync_folders().await?; let mut synced = 0usize; for folder in folders { let (emails, _state) = backend.sync_messages(&folder.id, None).await?; for email in emails { + let is_new = is_new_message(db, &email.id).await; db.save_email(&email).await?; + if is_new { + apply_filter_actions(db, filter_engine, &email).await; + } synced += 1; } } @@ -85,6 +272,7 @@ pub async fn sync_with_backend( db: &nuncio_store::db::DatabaseEngine, event_bus: &EventBus, backend: &dyn MailBackend, + filter_engine: &FilterEngine, account_id: Option, ) -> Result { match &account_id { @@ -94,7 +282,7 @@ pub async fn sync_with_backend( None => event_bus.process_command(CoreCommand::SyncAll), } - let result = fetch_and_persist(db, backend).await; + let result = fetch_and_persist(db, backend, filter_engine).await; if let Err(e) = &result { event_bus.process_command(CoreCommand::ReportError { @@ -132,7 +320,12 @@ fn build_mail_backend(config: &AccountConfig, password: &str) -> Box Box::new(JmapEngine::new(&config.id)), + AccountProtocol::Jmap => Box::new(JmapEngine::with_credentials( + &config.id, + &config.server_host, + &config.email_address, + password, + )), } } @@ -148,6 +341,7 @@ pub async fn run_account_sync( db: &nuncio_store::db::DatabaseEngine, secrets: &SecretManager, event_bus: &EventBus, + filter_engine: &FilterEngine, account_id: &str, ) -> Result { let setup = async { @@ -163,6 +357,7 @@ pub async fn run_account_sync( db, event_bus, backend.as_ref(), + filter_engine, Some(account_id.to_string()), ) .await @@ -187,6 +382,7 @@ pub async fn run_all_accounts_sync( db: &nuncio_store::db::DatabaseEngine, secrets: &SecretManager, event_bus: &EventBus, + filter_engine: &FilterEngine, ) -> usize { event_bus.process_command(CoreCommand::SyncAll); @@ -197,7 +393,7 @@ pub async fn run_all_accounts_sync( match secrets.get_secret(&config.keyring_secret_key) { Ok(password) => { let backend = build_mail_backend(&config, &password); - match fetch_and_persist(db, backend.as_ref()).await { + match fetch_and_persist(db, backend.as_ref(), filter_engine).await { Ok(count) => total_synced += count, Err(e) => { event_bus.process_command(CoreCommand::ReportError { @@ -238,14 +434,27 @@ mod tests { use nuncio_core::{CoreEvent, EngineStatus, TlsMode}; use nuncio_mail::MockMailBackend; use nuncio_store::db::DatabaseEngine; + use serde_json::json; + use wiremock::matchers::{body_partial_json, method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + /// A `FilterEngine` with no rules, for tests that exercise sync + /// mechanics unrelated to filtering. + fn empty_filter_engine() -> FilterEngine { + FilterEngine::new(Vec::new()).expect("empty rule set") + } fn sample_jmap_account(id: &str) -> AccountConfig { + sample_jmap_account_at(id, "jmap.nuncio.mx") + } + + fn sample_jmap_account_at(id: &str, host: &str) -> AccountConfig { AccountConfig { id: id.to_string(), name: "JMAP Test Account".to_string(), email_address: format!("{id}@nuncio.mx"), protocol: AccountProtocol::Jmap, - server_host: "jmap.nuncio.mx".to_string(), + server_host: host.to_string(), server_port: 443, smtp_host: "smtp.nuncio.mx".to_string(), smtp_port: 465, @@ -257,6 +466,79 @@ mod tests { } } + /// Mount a full JMAP session-discovery + `Mailbox/get` + `Email/query` + `Email/get` + /// stub set on `mock_server`, serving exactly one folder ("jmap-inbox") containing one + /// message ("jmap-msg-1"). Used by tests that need `JmapEngine::with_credentials` to + /// complete a real (wiremock-backed) fetch through `MailBackend`. + async fn mount_single_message_jmap_stubs(mock_server: &MockServer, account_id: &str) { + let session_body = json!({ + "username": format!("{account_id}@nuncio.mx"), + "primaryAccounts": { "urn:ietf:params:jmap:mail": account_id }, + "apiUrl": format!("{}/jmap/api", mock_server.uri()), + "state": "session-state-1" + }); + Mock::given(method("GET")) + .and(path("/.well-known/jmap")) + .respond_with(ResponseTemplate::new(200).set_body_json(&session_body)) + .mount(mock_server) + .await; + + let mailbox_get_body = json!({ + "methodResponses": [[ + "Mailbox/get", + { "list": [ { "id": "jmap-inbox", "name": "Inbox", "totalEmails": 1, "unreadEmails": 1 } ] }, + "c1" + ]] + }); + Mock::given(method("POST")) + .and(path("/jmap/api")) + .and(body_partial_json( + json!({"methodCalls": [["Mailbox/get", {}, "c1"]]}), + )) + .respond_with(ResponseTemplate::new(200).set_body_json(&mailbox_get_body)) + .mount(mock_server) + .await; + + let email_query_body = json!({ + "methodResponses": [["Email/query", {"ids": ["jmap-msg-1"]}, "c1"]] + }); + Mock::given(method("POST")) + .and(path("/jmap/api")) + .and(body_partial_json( + json!({"methodCalls": [["Email/query", {}, "c1"]]}), + )) + .respond_with(ResponseTemplate::new(200).set_body_json(&email_query_body)) + .mount(mock_server) + .await; + + let email_get_body = json!({ + "methodResponses": [[ + "Email/get", + { + "state": "sync-state-1", + "list": [{ + "id": "jmap-msg-1", + "subject": "Welcome to JMAP Sync", + "from": [{ "email": "support@nuncio.mx" }], + "to": [{ "email": format!("{account_id}@nuncio.mx") }], + "receivedAt": 1700000000i64, + "isUnread": true, + "bodySnippet": "Real wiremock-backed JMAP fetch." + }] + }, + "c1" + ]] + }); + Mock::given(method("POST")) + .and(path("/jmap/api")) + .and(body_partial_json( + json!({"methodCalls": [["Email/get", {}, "c1"]]}), + )) + .respond_with(ResponseTemplate::new(200).set_body_json(&email_get_body)) + .mount(mock_server) + .await; + } + fn sample_imap_account(id: &str) -> AccountConfig { AccountConfig { id: id.to_string(), @@ -309,9 +591,16 @@ mod tests { mock.add_message(mock_email("m1", "inbox", "Hello")); mock.add_message(mock_email("m2", "inbox", "World")); - let synced = sync_with_backend(&db, &event_bus, &mock, Some("acct-mock-1".to_string())) - .await - .expect("sync succeeds"); + let filter_engine = empty_filter_engine(); + let synced = sync_with_backend( + &db, + &event_bus, + &mock, + &filter_engine, + Some("acct-mock-1".to_string()), + ) + .await + .expect("sync succeeds"); assert_eq!(synced, 2); let persisted = db @@ -354,7 +643,8 @@ mod tests { unread_messages: 0, }); - let synced = sync_with_backend(&db, &event_bus, &mock, None) + let filter_engine = empty_filter_engine(); + let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) .await .expect("sync succeeds"); assert_eq!(synced, 0); @@ -380,9 +670,16 @@ mod tests { let mock = MockMailBackend::new(); mock.set_should_fail(true); - let err = sync_with_backend(&db, &event_bus, &mock, Some("acct-fail".to_string())) - .await - .expect_err("sync fails"); + let filter_engine = empty_filter_engine(); + let err = sync_with_backend( + &db, + &event_bus, + &mock, + &filter_engine, + Some("acct-fail".to_string()), + ) + .await + .expect_err("sync fails"); assert!(err.to_string().contains("mail backend error")); assert_eq!( @@ -447,12 +744,14 @@ mod tests { #[tokio::test] async fn run_account_sync_persists_real_backend_messages_and_reports_success() { - // Uses a real (non-mock) `JmapEngine` -- its `MailBackend` impl - // returns deterministic static data with no network I/O, so this - // proves `run_account_sync`'s full production wiring (account - // lookup -> keyring password -> real backend -> persist -> events) - // end-to-end without violating the "no live network calls in - // tests" rule. + // Uses a real (non-mock) `JmapEngine::with_credentials`, wired exactly like + // `build_mail_backend` wires it in production, pointed at a wiremock server standing + // in for the real JMAP host. This proves `run_account_sync`'s full production wiring + // (account lookup -> keyring password -> real HTTP backend -> persist -> events) + // end-to-end without violating the "no live network calls in tests" rule. + let mock_server = MockServer::start().await; + mount_single_message_jmap_stubs(&mock_server, "acct-jmap-real-1").await; + let (db, _dir) = DatabaseEngine::connect_ephemeral() .await .expect("ephemeral db"); @@ -460,20 +759,24 @@ mod tests { let mut events = event_bus.subscribe_events(); let secrets = SecretManager::mock(); - let config = sample_jmap_account("acct-jmap-real-1"); + let config = sample_jmap_account_at("acct-jmap-real-1", &mock_server.uri()); db.save_account(&config).await.expect("save account"); secrets .set_secret(&config.keyring_secret_key, "irrelevant-jmap-token") .expect("store credential in mock vault"); - let synced = run_account_sync(&db, &secrets, &event_bus, "acct-jmap-real-1") - .await - .expect("sync succeeds"); - // JmapEngine::sync_folders() returns 3 static folders; sync_messages - // returns the same 1 static message per folder regardless of - // folder_id, so 3 folders => 3 processed messages (deduplicated to - // 1 stored row by `save_email`'s INSERT OR REPLACE on message id). - assert_eq!(synced, 3); + let filter_engine = empty_filter_engine(); + let synced = run_account_sync( + &db, + &secrets, + &event_bus, + &filter_engine, + "acct-jmap-real-1", + ) + .await + .expect("sync succeeds"); + // The wiremock stubs advertise 1 folder containing 1 message. + assert_eq!(synced, 1); let persisted = db .get_message("jmap-msg-1") @@ -505,9 +808,16 @@ mod tests { let mut events = event_bus.subscribe_events(); let secrets = SecretManager::mock(); - let err = run_account_sync(&db, &secrets, &event_bus, "acct-does-not-exist") - .await - .expect_err("unknown account fails"); + let filter_engine = empty_filter_engine(); + let err = run_account_sync( + &db, + &secrets, + &event_bus, + &filter_engine, + "acct-does-not-exist", + ) + .await + .expect_err("unknown account fails"); assert!(matches!(err, SyncError::AccountNotFound(_))); match events.recv().await.expect("error event") { @@ -533,7 +843,8 @@ mod tests { db.save_account(&config).await.expect("save account"); // Deliberately never calls `secrets.set_secret(...)`. - let err = run_account_sync(&db, &secrets, &event_bus, "acct-no-cred-1") + let filter_engine = empty_filter_engine(); + let err = run_account_sync(&db, &secrets, &event_bus, &filter_engine, "acct-no-cred-1") .await .expect_err("missing credential fails"); assert!(matches!(err, SyncError::Vault(_))); @@ -556,7 +867,8 @@ mod tests { let mut events = event_bus.subscribe_events(); let secrets = SecretManager::mock(); - let total = run_all_accounts_sync(&db, &secrets, &event_bus).await; + let filter_engine = empty_filter_engine(); + let total = run_all_accounts_sync(&db, &secrets, &event_bus, &filter_engine).await; assert_eq!(total, 0); assert_eq!( @@ -571,14 +883,19 @@ mod tests { #[tokio::test] async fn run_all_accounts_sync_persists_across_multiple_accounts() { + let mock_server_a = MockServer::start().await; + let mock_server_b = MockServer::start().await; + mount_single_message_jmap_stubs(&mock_server_a, "acct-all-a").await; + mount_single_message_jmap_stubs(&mock_server_b, "acct-all-b").await; + let (db, _dir) = DatabaseEngine::connect_ephemeral() .await .expect("ephemeral db"); let event_bus = EventBus::new(); let secrets = SecretManager::mock(); - let config_a = sample_jmap_account("acct-all-a"); - let config_b = sample_jmap_account("acct-all-b"); + let config_a = sample_jmap_account_at("acct-all-a", &mock_server_a.uri()); + let config_b = sample_jmap_account_at("acct-all-b", &mock_server_b.uri()); db.save_account(&config_a).await.expect("save account a"); db.save_account(&config_b).await.expect("save account b"); secrets @@ -588,9 +905,10 @@ mod tests { .set_secret(&config_b.keyring_secret_key, "token-b") .expect("store credential b"); - let total = run_all_accounts_sync(&db, &secrets, &event_bus).await; - // 2 accounts * 3 folders each = 6 processed messages. - assert_eq!(total, 6); + let filter_engine = empty_filter_engine(); + let total = run_all_accounts_sync(&db, &secrets, &event_bus, &filter_engine).await; + // 2 accounts * 1 folder * 1 message each = 2 processed messages. + assert_eq!(total, 2); assert_eq!(event_bus.current_state().status, EngineStatus::Idle); } @@ -607,7 +925,8 @@ mod tests { db.save_account(&config).await.expect("save account"); // No credential stored in the vault for this account. - let total = run_all_accounts_sync(&db, &secrets, &event_bus).await; + let filter_engine = empty_filter_engine(); + let total = run_all_accounts_sync(&db, &secrets, &event_bus, &filter_engine).await; assert_eq!(total, 0); assert_eq!( @@ -625,4 +944,261 @@ mod tests { CoreEvent::SyncCompleted { account_id: None } ); } + + #[tokio::test] + async fn sync_with_backend_fires_immediate_mark_read_action_on_match() { + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("ephemeral db"); + let event_bus = EventBus::new(); + + let rule = nuncio_filter::NsqlParser::parse_rule( + "Urgent Auto-Read", + 1, + "WHERE subject CONTAINS 'Urgent' ACTION MARK READ", + ) + .expect("parse rule"); + let filter_engine = FilterEngine::new(vec![rule]).expect("compile rule"); + + let mock = MockMailBackend::new(); + mock.add_folder(Folder { + id: "inbox".to_string(), + name: "Inbox".to_string(), + total_messages: 1, + unread_messages: 1, + }); + mock.add_message(mock_email("m-urgent", "inbox", "Urgent: server down")); + + let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + .await + .expect("sync succeeds"); + assert_eq!(synced, 1); + + let stored = db.get_message("m-urgent").await.expect("message persisted"); + assert!( + stored.read, + "MARK READ action must genuinely flip the stored read flag" + ); + + let logs = db + .list_filter_execution_logs(10) + .await + .expect("list execution logs"); + assert_eq!(logs.len(), 1); + assert_eq!(logs[0].message_id, "m-urgent"); + + let pending = db + .list_pending_mutations(10) + .await + .expect("list pending mutations"); + assert!( + pending.is_empty(), + "an immediate action must not enqueue an outbox mutation" + ); + } + + #[tokio::test] + async fn sync_with_backend_enqueues_outbox_mutation_for_remote_action_on_match() { + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("ephemeral db"); + let event_bus = EventBus::new(); + + let rule = nuncio_filter::NsqlParser::parse_rule( + "Archive Rule", + 1, + "WHERE subject CONTAINS 'Archive Me' ACTION MOVE TO 'Archive'", + ) + .expect("parse rule"); + let filter_engine = FilterEngine::new(vec![rule]).expect("compile rule"); + + let mock = MockMailBackend::new(); + mock.add_folder(Folder { + id: "inbox".to_string(), + name: "Inbox".to_string(), + total_messages: 1, + unread_messages: 1, + }); + mock.add_message(mock_email("m-archive", "inbox", "Please Archive Me")); + + let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + .await + .expect("sync succeeds"); + assert_eq!(synced, 1); + + let pending = db + .list_pending_mutations(10) + .await + .expect("list pending mutations"); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].message_id, "m-archive"); + assert_eq!(pending[0].mutation_type, "MOVE"); + assert_eq!(pending[0].status, "pending"); + assert!(pending[0].payload.contains("Archive")); + + let logs = db + .list_filter_execution_logs(10) + .await + .expect("list execution logs"); + assert_eq!(logs.len(), 1); + assert_eq!(logs[0].message_id, "m-archive"); + + // The remote action must not have applied a local read-flag change. + let stored = db + .get_message("m-archive") + .await + .expect("message persisted"); + assert!(!stored.read); + } + + #[tokio::test] + async fn sync_with_backend_non_matching_message_produces_no_filter_side_effects() { + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("ephemeral db"); + let event_bus = EventBus::new(); + + let rule = nuncio_filter::NsqlParser::parse_rule( + "Urgent Auto-Read", + 1, + "WHERE subject CONTAINS 'Urgent' ACTION MARK READ", + ) + .expect("parse rule"); + let filter_engine = FilterEngine::new(vec![rule]).expect("compile rule"); + + let mock = MockMailBackend::new(); + mock.add_folder(Folder { + id: "inbox".to_string(), + name: "Inbox".to_string(), + total_messages: 1, + unread_messages: 1, + }); + mock.add_message(mock_email("m-plain", "inbox", "Just a normal update")); + + let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + .await + .expect("sync succeeds"); + assert_eq!(synced, 1); + + let stored = db.get_message("m-plain").await.expect("message persisted"); + assert!(!stored.read, "non-matching message must stay untouched"); + + let logs = db + .list_filter_execution_logs(10) + .await + .expect("list execution logs"); + assert!(logs.is_empty(), "no rule matched, so no log should exist"); + + let pending = db + .list_pending_mutations(10) + .await + .expect("list pending mutations"); + assert!( + pending.is_empty(), + "no rule matched, so no mutation should be enqueued" + ); + } + + #[tokio::test] + async fn sync_with_backend_does_not_refire_filters_on_a_repeat_sync_of_the_same_message() { + // Sync is not yet incremental: a real backend re-reports messages it + // already handed over on every sync pass. This proves a message's + // filter actions fire exactly once across repeated syncs rather than + // once per sync cycle -- the regression this guard exists for. + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("ephemeral db"); + let event_bus = EventBus::new(); + + let mark_read_rule = nuncio_filter::NsqlParser::parse_rule( + "Urgent Auto-Read", + 1, + "WHERE subject CONTAINS 'Urgent' ACTION MARK READ", + ) + .expect("parse rule"); + let move_rule = nuncio_filter::NsqlParser::parse_rule( + "Archive Rule", + 2, + "WHERE subject CONTAINS 'Urgent' ACTION MOVE TO 'Archive'", + ) + .expect("parse rule"); + let filter_engine = + FilterEngine::new(vec![mark_read_rule, move_rule]).expect("compile rules"); + + let mock = MockMailBackend::new(); + mock.add_folder(Folder { + id: "inbox".to_string(), + name: "Inbox".to_string(), + total_messages: 1, + unread_messages: 1, + }); + mock.add_message(mock_email("m-repeat", "inbox", "Urgent: server down")); + + // First sync: the message is genuinely new, so both rules must fire. + let synced_first = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + .await + .expect("first sync succeeds"); + assert_eq!(synced_first, 1); + + let stored_after_first = db + .get_message("m-repeat") + .await + .expect("message persisted after first sync"); + assert!(stored_after_first.read); + + let logs_after_first = db + .list_filter_execution_logs(10) + .await + .expect("list execution logs after first sync"); + assert_eq!(logs_after_first.len(), 2, "both rules should fire once"); + + let pending_after_first = db + .list_pending_mutations(10) + .await + .expect("list pending mutations after first sync"); + assert_eq!(pending_after_first.len(), 1); + + // Manually flip the message back to unread, exactly as a user + // reading and then un-reading it would -- proves the second sync + // doesn't re-flip it back to read via a repeat MARK READ fire. + db.set_message_read("m-repeat", false) + .await + .expect("manually mark unread"); + + // Second sync: the mock backend reports the SAME message again + // (unchanged id), simulating a non-incremental re-sync. + let synced_second = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + .await + .expect("second sync succeeds"); + assert_eq!(synced_second, 1); + + let stored_after_second = db + .get_message("m-repeat") + .await + .expect("message persisted after second sync"); + assert!( + !stored_after_second.read, + "a re-sync of an already-seen message must not re-fire MARK READ" + ); + + let logs_after_second = db + .list_filter_execution_logs(10) + .await + .expect("list execution logs after second sync"); + assert_eq!( + logs_after_second.len(), + 2, + "no new execution log entries should be written on a repeat sync" + ); + + let pending_after_second = db + .list_pending_mutations(10) + .await + .expect("list pending mutations after second sync"); + assert_eq!( + pending_after_second.len(), + 1, + "no new outbox mutation should be enqueued on a repeat sync" + ); + } } diff --git a/crates/nunciod/tests/filters_triage_e2e_test.rs b/crates/nunciod/tests/filters_triage_e2e_test.rs new file mode 100644 index 0000000..bb34869 --- /dev/null +++ b/crates/nunciod/tests/filters_triage_e2e_test.rs @@ -0,0 +1,267 @@ +//! Offline end-to-end test for the `nuncio.v1.Filters.Triage` streaming RPC. +//! +//! Boots a real `nunciod` daemon gRPC server (`nunciod::grpc:: +//! serve_on_listener`) on an ephemeral loopback port, backed by a real +//! temp file-backed [`DatabaseEngine`] seeded directly via +//! [`DatabaseEngine::save_email`] (standing in for messages that were synced +//! before the active rule set existed) and a live [`FilterEngine`] carrying +//! a real persisted rule. +//! +//! Drives ONLY the authenticated `nuncio-proto` `FiltersClient::triage` call +//! against the live server, proving: +//! a. the final cumulative `matched_count`/`actions_applied_count` are +//! correct against a store containing both matching and non-matching +//! messages; +//! b. matching messages show the REAL side effects `apply_filter_actions` +//! produces (the read flag flips for an immediate `MARK READ` action; +//! an outbox row is queued for a remote `MOVE TO` action) -- not a +//! fabricated count with no backing state change; +//! c. a store bigger than one `chunk_size` genuinely streams more than one +//! non-final `TriageProgress` message, proving this is a real +//! multi-chunk stream rather than one batch dressed up as a stream. +//! +//! Fully offline and deterministic: no real network, no real OS keyring +//! (`SecretManager::mock`), no fixed sleeps. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +use nuncio_core::model::Email; +use nuncio_core::EventBus; +use nuncio_filter::{FilterEngine, NsqlParser}; +use nuncio_proto::v1::filters_client::FiltersClient; +use nuncio_proto::v1::TriageRequest; +use nuncio_store::db::DatabaseEngine; +use nuncio_store::vault::{SecretManager, GRPC_TOKEN_ACCOUNT}; +use nunciod::grpc::serve_on_listener; +use std::sync::Arc; +use tokio::net::TcpListener; + +const ACCOUNT_ID: &str = "acct-triage-e2e-1"; + +fn sample_email(id: &str, subject: &str) -> Email { + Email { + id: id.to_string(), + account_id: ACCOUNT_ID.to_string(), + folder_id: "inbox".to_string(), + subject: subject.to_string(), + sender: "alice@nuncio.mx".to_string(), + recipient: "triage-e2e@nuncio.mx".to_string(), + received_at: 1_700_000_000, + read: false, + body_plain: Some("body".to_string()), + body_html: None, + attachments: Vec::new(), + } +} + +/// Boots the real daemon gRPC server (bearer-token authenticated) on an +/// ephemeral loopback port, backed by a real temp file-backed +/// `DatabaseEngine` and the given live `FilterEngine`. Returns the server +/// address, the minted token, the `DatabaseEngine` handle (used to seed +/// messages before triggering `Triage`), and the `TempDir` guard, which the +/// caller must hold for as long as it talks to the server. +async fn boot_daemon( + secrets: Arc, + filter_engine: Arc, +) -> ( + std::net::SocketAddr, + String, + Arc, + tempfile::TempDir, +) { + let dir = tempfile::tempdir().expect("create temp dir for file-backed db"); + let db_path = dir.path().join("triage_e2e.db"); + let db = Arc::new( + DatabaseEngine::connect_file(&db_path, &secrets) + .await + .expect("open temp file-backed database"), + ); + + let token_bytes = secrets + .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32) + .expect("mint gRPC bearer token from mock vault"); + let token = hex::encode(token_bytes); + + 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 server_db = db.clone(); + let server_secrets = secrets.clone(); + let server_token = token.clone(); + tokio::spawn(async move { + let _ = serve_on_listener( + listener, + event_bus, + server_db, + filter_engine, + server_secrets, + server_token, + ) + .await; + }); + + (addr, token, db, dir) +} + +fn authed_request(token: &str, message: T) -> tonic::Request { + let mut request = tonic::Request::new(message); + request.metadata_mut().insert( + "authorization", + format!("Bearer {token}") + .parse() + .expect("valid ascii metadata value"), + ); + request +} + +#[tokio::test] +async fn triage_applies_live_rule_retroactively_and_reports_real_cumulative_counts() { + let secrets = Arc::new(SecretManager::mock()); + + let mark_read_rule = NsqlParser::parse_rule( + "Urgent Auto-Read", + 1, + "WHERE subject CONTAINS 'Urgent' ACTION MARK READ", + ) + .expect("parse mark-read rule"); + let move_rule = NsqlParser::parse_rule( + "Archive Rule", + 2, + "WHERE subject CONTAINS 'Archive Me' ACTION MOVE TO 'Archive'", + ) + .expect("parse move rule"); + let filter_engine = + Arc::new(FilterEngine::new(vec![mark_read_rule, move_rule]).expect("compile rules")); + + let (addr, token, db, _dir) = boot_daemon(secrets, filter_engine).await; + + // Seed messages directly into the store, standing in for mail that was + // synced BEFORE these rules existed -- exactly the scenario `Triage` is + // for. Two match (one per rule), one does not. + db.save_email(&sample_email("m-urgent", "Urgent: server down")) + .await + .expect("seed urgent message"); + db.save_email(&sample_email("m-archive", "Please Archive Me")) + .await + .expect("seed archive message"); + db.save_email(&sample_email("m-plain", "Just saying hi")) + .await + .expect("seed non-matching message"); + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let mut stream = client + .triage(authed_request( + &token, + TriageRequest { + rule_id: None, + chunk_size: 100, + }, + )) + .await + .expect("triage call succeeds") + .into_inner(); + + let mut updates = Vec::new(); + while let Some(progress) = stream.message().await.expect("stream message") { + updates.push(progress); + } + + let last = updates.last().expect("at least one progress update"); + assert!(last.done, "final update must be marked done"); + assert_eq!(last.scanned_count, 3); + assert_eq!( + last.matched_count, 2, + "exactly the two messages with a rule match count as matched" + ); + assert_eq!( + last.actions_applied_count, 2, + "one action applied per matching message" + ); + + // Real side effects, not a fabricated count: the MARK READ action + // actually flipped the stored read flag... + let urgent = db.get_message("m-urgent").await.expect("get urgent"); + assert!(urgent.read, "MARK READ must genuinely flip the read flag"); + + // ...and the MOVE TO action actually queued a real outbox mutation. + let pending = db + .list_pending_mutations(10) + .await + .expect("list pending mutations"); + assert_eq!( + pending.len(), + 1, + "the remote MOVE action must enqueue exactly one outbox mutation" + ); + assert_eq!(pending[0].message_id, "m-archive"); + + let plain = db.get_message("m-plain").await.expect("get plain"); + assert!(!plain.read, "a non-matching message must be left untouched"); +} + +#[tokio::test] +async fn triage_streams_multiple_chunks_for_a_store_bigger_than_one_chunk() { + let secrets = Arc::new(SecretManager::mock()); + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); + + let (addr, token, db, _dir) = boot_daemon(secrets, filter_engine).await; + + // Five messages against a chunk size of two forces three chunks + // (2 + 2 + 1), so the stream must carry at least two non-final progress + // updates before the final `done: true` -- proof this is a real + // multi-chunk stream, not one batch disguised as one. + for i in 0..5 { + db.save_email(&sample_email( + &format!("m-{i:02}"), + &format!("Message number {i}"), + )) + .await + .expect("seed message"); + } + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let mut stream = client + .triage(authed_request( + &token, + TriageRequest { + rule_id: None, + chunk_size: 2, + }, + )) + .await + .expect("triage call succeeds") + .into_inner(); + + let mut updates = Vec::new(); + while let Some(progress) = stream.message().await.expect("stream message") { + updates.push(progress); + } + + let non_final: Vec<_> = updates.iter().filter(|p| !p.done).collect(); + assert!( + non_final.len() > 1, + "expected more than one non-final progress update for a multi-chunk scan, got {}", + non_final.len() + ); + + let last = updates.last().expect("at least one progress update"); + assert!(last.done); + assert_eq!(last.scanned_count, 5); + + // Cumulative counts must monotonically increase across the stream -- + // never reset or double-count between chunks. + let mut prev_scanned = 0u64; + for update in &updates { + assert!(update.scanned_count >= prev_scanned); + prev_scanned = update.scanned_count; + } +} diff --git a/crates/nunciod/tests/grpc_auth_test.rs b/crates/nunciod/tests/grpc_auth_test.rs index aaafaa7..767e051 100644 --- a/crates/nunciod/tests/grpc_auth_test.rs +++ b/crates/nunciod/tests/grpc_auth_test.rs @@ -154,3 +154,66 @@ async fn accepts_correct_bearer_token_and_returns_real_engine_status() { assert_eq!(response.engine_status, "Syncing"); assert_eq!(response.version, env!("CARGO_PKG_VERSION")); } + +/// Confirms `Filters` newly added management RPCs are mounted behind the +/// SAME `BearerAuthInterceptor` as every other service -- an unauthenticated +/// call to `UpdateRule` must be rejected exactly like an unauthenticated +/// call to `System/GetStatus` above, with no carve-out for "just one more" +/// RPC. +#[tokio::test] +async fn rejects_calls_to_new_filters_rpcs_without_a_bearer_token() { + let event_bus = Arc::new(EventBus::new()); + let secrets = SecretManager::mock(); + let token = hex::encode( + secrets + .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32) + .expect("token provisioned from mock vault"), + ); + + let (addr, _dir) = start_server(event_bus, token).await; + let mut client = + nuncio_proto::v1::filters_client::FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let err = client + .update_rule(nuncio_proto::v1::UpdateRuleRequest { + id: "rule-1".to_string(), + name: None, + nsql: None, + priority: None, + }) + .await + .expect_err("call without any authorization header must be rejected"); + assert_eq!(err.code(), Code::Unauthenticated); +} + +/// Confirms the streaming `Filters.Triage` RPC is mounted behind the SAME +/// `BearerAuthInterceptor` as every other `Filters` RPC -- an unauthenticated +/// call must be rejected before the stream is ever established, exactly +/// like an unauthenticated unary call. +#[tokio::test] +async fn rejects_triage_calls_without_a_bearer_token() { + let event_bus = Arc::new(EventBus::new()); + let secrets = SecretManager::mock(); + let token = hex::encode( + secrets + .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32) + .expect("token provisioned from mock vault"), + ); + + let (addr, _dir) = start_server(event_bus, token).await; + let mut client = + nuncio_proto::v1::filters_client::FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let err = client + .triage(nuncio_proto::v1::TriageRequest { + rule_id: None, + chunk_size: 0, + }) + .await + .expect_err("call without any authorization header must be rejected"); + assert_eq!(err.code(), Code::Unauthenticated); +} diff --git a/docs/HANDOFF.md b/docs/HANDOFF.md index 11d950b..48e3ba9 100644 --- a/docs/HANDOFF.md +++ b/docs/HANDOFF.md @@ -23,16 +23,24 @@ machine (or a new LLM) picks up the work cold. it to the definition of done, and open a PR. Do not merge your own work; do not start a second story until yours is merged or handed back. -## Current state of the code (2026-07-28) +## Current state of the code (2026-07-30) - **Phases 0–2 complete.** The daemon boots gRPC-only on loopback `127.0.0.1:9420` behind a keyring bearer token, serving six services (System, Accounts, Mail, Filters, Export, Audit), each behind the auth interceptor. The `IMAP → store → read → SMTP send` spine works end-to-end over gRPC via the `nuncio-cli` reference client, tested offline. `nuncio.v1` is published with a contract-stability golden. -- **Phase 3 in progress:** filter correctness (3.A) and the calendar **engine** - layer (3.B.1: real CalDAV transport, TZID fix, `CalendarBackend`, store CRUD) - are done and on `dev`. Everything else is the M1–M7 backlog. +- **Phase 3 (feature completeness) is now COMPLETE — M1–M4 all merged to `dev`.** + Filter correctness (3.A), Calendar (M1: CalDAV + RRULE recurrence, API vertical), + Contacts (M2: real CardDAV engine + API vertical), JMAP (M3: real client wired + into Mail sync), and Filters made live (M4: engine wired into sync with a + fire-once guard, the `account`-field bug fixed, edit/export/import/logs and + `Filters.Triage` bulk rescan on the gRPC service) are all done. +- **Next milestone: M5 — Mail lifecycle + incremental sync** (account + edit/delete/test-connection + persisted TLS mode, incremental IMAP sync with + progress + per-item FETCH timeout, explicit `SendMessage` account selection, + keyring rollback on failed `AddAccount`). Then M6 (security hardening), then + M7 (API freeze / client-readiness). - Full detail and the path to a client-ready API: `docs/ROADMAP.md`. ## Branches & preserved WIP @@ -86,24 +94,26 @@ GitHub milestones **M1–M7** mirror `docs/ROADMAP.md`. Each milestone holds `story`-labelled issues with a full, self-contained spec. Order: M1 → M7 (M1–M5 independent-ish feature work; M6 security; M7 the client-readiness gate). -**M1 — Finish Calendar (CalDAV)** · milestone #23 -- #185 — Add Calendar API vertical (proto, daemon, CLI, E2E) · **[PROTO]** -- #186 — Wire RRULE recurrence expansion into calendar event queries +**M1–M4 are DELIVERED — all stories merged to `dev`.** M5 is next. -**M2 — Contacts (CardDAV)** · milestone #24 -- #187 — Build a real Contacts (CardDAV) engine, replacing the fabricated client -- #188 — Add Contacts API vertical (proto, daemon, CLI, E2E) · **[PROTO]** +**M1 — Finish Calendar (CalDAV)** · milestone #23 · **DONE** +- #185 — Add Calendar API vertical (proto, daemon, CLI, E2E) · **[PROTO]** · merged +- #186 — Wire RRULE recurrence expansion into calendar event queries · merged -**M3 — JMAP** · milestone #25 -- #189 — Real JMAP client wired into the existing Mail sync path +**M2 — Contacts (CardDAV)** · milestone #24 · **DONE** +- #187 — Build a real Contacts (CardDAV) engine, replacing the fabricated client · merged +- #188 — Add Contacts API vertical (proto, daemon, CLI, E2E) · **[PROTO]** · merged -**M4 — Filters made live** · milestone #26 -- #190 — Wire FilterEngine into live sync so matched rules actually fire *(headline; #193 depends on its helper)* -- #191 — Fix filter header-field and account-field evaluation bugs -- #192 — Move filter edit/export/import/logs onto the Filters gRPC service · **[PROTO]** -- #193 — Re-expose bulk filter triage as a streaming Filters RPC · **[PROTO]** *(depends on #190)* +**M3 — JMAP** · milestone #25 · **DONE** +- #189 — Real JMAP client wired into the existing Mail sync path · merged -**M5 — Mail lifecycle + incremental sync** · milestone #27 +**M4 — Filters made live** · milestone #26 · **DONE** +- #190 — Wire FilterEngine into live sync so matched rules actually fire *(headline; #193 depends on its helper)* · merged +- #191 — Fix filter header-field and account-field evaluation bugs · merged +- #192 — Move filter edit/export/import/logs onto the Filters gRPC service · **[PROTO]** · merged +- #193 — Re-expose bulk filter triage as a streaming Filters RPC · **[PROTO]** *(depends on #190)* · merged + +**M5 — Mail lifecycle + incremental sync** · milestone #27 ← **next** - #194 — Account lifecycle over gRPC (edit/delete/test) + persist TLS mode · **[PROTO]** *(prior art: `wip/unreviewed-account-tls-sync`, reference-not-trust)* - #195 — Incremental IMAP sync with progress reporting and per-item FETCH timeout - #196 — SendMessage selects an explicit account instead of the first configured · **[PROTO]** @@ -135,3 +145,7 @@ independent-ish feature work; M6 security; M7 the client-readiness gate). those two issues if you consider that sensitive. - Decide the eventual fate of `wip/unreviewed-account-tls-sync` (review-and-adopt vs discard) once M5 lands a clean implementation. +- Follow-ups filed during M1–M4 that are open but not blocking (not scheduled + into a milestone yet): #216 (calendar windowing/predicate refinement), #219 + (contacts notes handling), #225 (`filter edit` silently re-enables a disabled + rule and resets `created_at`). diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 352dc0f..1eed593 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -41,41 +41,49 @@ These are non-negotiable and apply to every item below: --- -## Where we are today (2026-07-28) +## Where we are today (2026-07-30) **Phases 0–2 are complete and on `dev`.** The daemon boots gRPC-only on loopback -`127.0.0.1:9420` behind a keyring-minted bearer token, and serves **six services, -each mounted behind the auth interceptor**: `System`, `Accounts`, `Mail`, -`Filters`, `Export`, `Audit`. The IMAP→store→read→SMTP-send spine works +`127.0.0.1:9420` behind a keyring-minted bearer token, and serves **eight +services, each mounted behind the auth interceptor**: `System`, `Accounts`, +`Mail`, `Filters`, `Calendar`, `Contacts`, `Export`, `Audit` (the `Calendar` +and `Contacts` verticals landed in M1/M2). The IMAP→store→read→SMTP-send spine works end-to-end over gRPC, driven by the `nuncio-cli` reference client, tested offline. The legacy JSON-RPC IPC is removed. The `nuncio.v1` contract is published with a byte-deterministic `FileDescriptorSet` golden guarding it, plus external-codegen docs for Swift/C#/TypeScript. -- **Phase 0 — Honesty & foundation:** done. Fabricated docs archived; workspace - shrunk to engine crates + CLI (`_reference/` holds the old shells); real OS - keyring; fail-open auto-updater disabled; releases marked pre-alpha. -- **Phase 1 — gRPC skeleton + the spine:** done. Authenticated loopback gRPC; - CLI as gRPC client; real `IMAP fetch → store → read → SMTP send` against mock - servers; store correctness fixes (FTS-vs-encryption, backfill, salvage schema, - transient-error ≠ corruption). -- **Phase 2 — API completeness for the spine:** done. Full proto surface for the - spine; contract-stability test; codegen docs; IPC removed. - -**Phase 3 — Feature completeness — is in progress:** +**Phase 3 — Feature completeness — is now COMPLETE**, via milestones M1–M4, all +merged to `dev`: - **3.A Filter correctness** — **done** (account scoping, `ON ACCOUNT`, non-ASCII offsets, `NOT IN`). -- **3.B Calendar** — engine layer **done** (real CalDAV `REPORT` transport, - `TZID` fix, `CalendarBackend` trait, `calendar_events` store CRUD); the API - vertical + recurrence wiring remain (see M1). -- **3.C Contacts (CardDAV)** and **3.D JMAP** — not started. +- **3.B Calendar** — **done** (M1). Real CalDAV `REPORT` transport, `TZID` fix, + `CalendarBackend` trait, `calendar_events` store CRUD, the `Calendar` gRPC + service + CLI commands, and RRULE recurrence wired into the event-query path. +- **3.C Contacts (CardDAV)** — **done** (M2). A real, wiremock-tested CardDAV + client replaced the fabricated one; contacts persist through the daemon's + store and are exposed over a `Contacts` gRPC service + CLI. +- **3.D JMAP** — **done** (M3). A real JMAP HTTP client is wired into the same + `MailBackend` fetch→store path as IMAP. +- **Filters made live** — **done** (M4). `FilterEngine::evaluate` is wired into + the live sync path with a fire-once-per-new-message guard, the `account` + condition-field bug is fixed, unsupported `header[...]` conditions are + honestly rejected, filter edit/export/import/logs moved onto the `Filters` + gRPC service, and a `Filters.Triage` server-streaming RPC does retroactive + bulk rescan. -Dogfooding against a real mailbox and a fabrication audit surfaced additional -**real gaps** now folded into the milestones below: the filter match-and-act -engine is not yet wired into the live sync path; sync is not incremental; -account edit/delete/test-connection and TLS-mode selection are incomplete; and a -set of correctness/security items (see M5/M6). +**Next up: M5 — Mail lifecycle & incremental sync** (account edit/delete/test + +persisted TLS mode, incremental IMAP sync with progress + per-item FETCH +timeout, explicit `SendMessage` account selection, keyring rollback on failed +`AddAccount`). Then M6 (security hardening), then M7 (API freeze / client +readiness). + +A handful of follow-ups were filed during M1–M4 and remain open, tracked as +their own backlog items rather than blocking M1–M4's completion: calendar +windowing/predicate refinement (#216), contacts notes handling (#219), and a +`filter edit` bug where re-saving a rule silently re-enables it and resets +`created_at` (#225). **Explicitly deferred** (out of scope until the backend is client-ready and each is re-justified as a genuine engine capability): OpenPGP/S-MIME E2EE (#79), diff --git a/docs/roadmap/index.html b/docs/roadmap/index.html index d6f3146..c09c6fb 100644 --- a/docs/roadmap/index.html +++ b/docs/roadmap/index.html @@ -76,7 +76,7 @@ /* milestone rail */ .miles{position:relative;margin-top:8px} .miles::before{content:"";position:absolute;left:19px;top:12px;bottom:34px;width:2px; - background:linear-gradient(var(--done),var(--prog) 16%,var(--line) 22%);border-radius:2px} + background:linear-gradient(var(--done),var(--done) 57%,var(--prog) 57%,var(--prog) 71%,var(--line) 71%);border-radius:2px} @media (max-width:640px){.miles::before{display:none}} .m{position:relative;padding-left:52px;margin-bottom:16px} @media (max-width:640px){.m{padding-left:0}} @@ -133,24 +133,31 @@
-

Where we are · 2026-07-28

+

Where we are · 2026-07-30

-

Phases 0–2 complete — the spine works end to end

+

Phases 0–2 complete, Phase 3 feature completeness done — M1–M4 all merged

The daemon boots gRPC-only on loopback 127.0.0.1:9420 behind a keyring-minted bearer token. IMAP → store → read → SMTP send runs end-to-end over gRPC, driven by the nuncio-cli reference client, tested fully offline. The legacy JSON-RPC IPC is gone, - and nuncio.v1 is published with a byte-deterministic contract-stability golden.

+ and nuncio.v1 is published with a byte-deterministic contract-stability golden. + Milestones M1–M4 are now complete on dev: Calendar, Contacts, JMAP, and live + Filters are all real, wired, and API-complete. Next up: M5 (mail lifecycle + & incremental sync).

SystemAccountsMail - FiltersExportAudit - — 6 services, each behind the auth interceptor + FiltersCalendarContacts + ExportAudit + — 8 services, each behind the auth interceptor
Phase 0 · Honesty & foundation Phase 1 · gRPC skeleton + spine Phase 2 · API completeness (spine) 3.A · Filter correctness - 3.B.1 · Calendar engine (CalDAV + TZID) + M1 · Calendar (CalDAV + recurrence) + M2 · Contacts (CardDAV) + M3 · JMAP + M4 · Filters made live
@@ -159,72 +166,72 @@

The path · seven milestones to the goal line

-
+
M1

M1Finish Calendar

- In progress + Done
-

Publish the calendar engine over the API: a Calendar gRPC service +

The calendar engine is published over the API: a Calendar gRPC service (Sync / ListEvents / GetEvent — honest surface only) behind the auth interceptor, daemon wiring, CLI - commands replacing the stub, and recurrence (rrule) wired into the fetch/store path (today it's dead code).

+ commands replacing the stub, and recurrence (rrule) wired into the fetch/store path.

Exit: a Calendar view over gRPC, offline-tested sync→list→get, recurring events materialize.
-
#1783.B.1 done → 3.B.2 next
+
#178done — #185, #186 merged
-
+
M2

M2Contacts · CardDAV

- Planned + Done
-

Replace the fake CardDAV client (returns one hardcoded contact) with a real, wiremock-tested - client + vCard round-trip; persist through the daemon's store (the CLI currently writes a throwaway - in-memory DB); add a Contacts service, CLI, and offline E2E.

+

The fake CardDAV client (returned one hardcoded contact) is replaced with a real, wiremock-tested + client + vCard round-trip; contacts persist through the daemon's store (the CLI previously wrote a throwaway + in-memory DB); a Contacts service, CLI, and offline E2E are in place.

Exit: contacts sync and persist across restarts, driven over gRPC.
-
#179
+
#179done — #187, #188 merged
-
+
M3

M3JMAP

- Planned + Done
-

A real JMAP HTTP client (session, Email/query/get/set) with the camelCase wire fix, wired into +

A real JMAP HTTP client (session, Email/query/get/set) with the camelCase wire fix is wired into the same MailBackend fetch→store path as IMAP; wiremock-tested.

Exit: a JMAP account fetches → stores → lists offline, via the existing Mail service.
-
#180
+
#180done — #189 merged
-
+
M4

M4Filters made live

- Planned + Done
-

Today rules can be created, validated, and previewed as “MATCH” but never fire. Wire - FilterEngine::evaluate into the real sync path so matches trigger their action - (move/flag/delete/forward/webhook), fix the header + account-field bugs, and re-expose filter - edit/export/import/logs and bulk triage over the service.

+

Rules could previously be created, validated, and previewed as “MATCH” but never fired. + FilterEngine::evaluate is now wired into the real sync path so matches trigger their action + (move/flag/delete/forward/webhook), the header + account-field bugs are fixed, and filter + edit/export/import/logs plus bulk triage are re-exposed over the service.

Exit: an E2E proves a matching rule fires an action against synced mail — no fabricated preview.
-
#175#176#181
+
#175#176#181done — #190–#193 merged
-
+
M5

M5Mail lifecycle + incremental sync

- Planned + Next

Round out real account management: edit / delete / test-connection with TLS modes (implicit / STARTTLS / plain) actually wired; explicit send-account selection; and incremental sync