diff --git a/crates/nuncio-cli/src/args.rs b/crates/nuncio-cli/src/args.rs index a0235dd..17b21ee 100644 --- a/crates/nuncio-cli/src/args.rs +++ b/crates/nuncio-cli/src/args.rs @@ -369,6 +369,12 @@ pub enum FilterSubcommand { #[arg(short, long, help = "Rule identifier")] id: String, }, + /// Validate an NSQL query's syntax and semantics without persisting it. + Validate { + /// NSQL query string to validate. + #[arg(short, long, help = "NSQL query string")] + sql: String, + }, /// Test / dry-run NSQL query against an email. Test { /// NSQL query string to test. diff --git a/crates/nuncio-cli/src/runner.rs b/crates/nuncio-cli/src/runner.rs index cee2a2e..561f5c3 100644 --- a/crates/nuncio-cli/src/runner.rs +++ b/crates/nuncio-cli/src/runner.rs @@ -57,6 +57,23 @@ fn message_proto_to_json(message: &nuncio_proto::v1::Message) -> serde_json::Val }) } +/// Renders a `nuncio.v1.FilterRule` (as returned by the daemon's `Filters` +/// gRPC service, backlog story 2.A / GH #171) into the JSON shape used by +/// `filter list`/`filter create`'s `--json` output. +fn filter_rule_proto_to_json(rule: &nuncio_proto::v1::FilterRule) -> serde_json::Value { + json!({ + "id": rule.id, + "name": rule.name, + "target_account": rule.target_account, + "priority": rule.priority, + "enabled": rule.enabled, + "nsql_text": rule.nsql_text, + "actions": rule.actions, + "created_at": rule.created_at, + "updated_at": rule.updated_at, + }) +} + /// Errors emitted by the CLI headless runner. #[derive(Error, Debug)] pub enum RunnerError { @@ -368,60 +385,35 @@ impl HeadlessRunner { } }, Commands::Filter { action } => match action { - FilterSubcommand::List => { - let rules = self.db.list_filter_rules().await.unwrap_or_default(); - if json_mode { - format_json(&json!(rules)) - } else if rules.is_empty() { - "No filter rules configured.".to_string() - } else { - let mut out = String::from( - "ID PRIORITY ENABLED NAME NSQL\n", - ); - for r in rules { - out.push_str(&format!( - "{:<10} {:<8} {:<7} {:<20} {}\n", - r.id, r.priority, r.enabled, r.name, r.nsql_text - )); - } - out - } - } + FilterSubcommand::List => self.handle_filter_list(json_mode).await, FilterSubcommand::Create { name, sql, priority, - } => match nuncio_filter::NsqlParser::parse_rule(name, *priority, sql) { - Ok(rule) => { - let opts = nuncio_filter::ValidationOptions::default(); - if let Err(e) = nuncio_filter::NsqlValidator::validate(&rule, &opts) { - return if json_mode { - format_json_error(&e.to_string()) - } else { - format!("Validation Error: {e}") - }; - } - if let Err(e) = self.db.save_filter_rule(&rule).await { - return if json_mode { - format_json_error(&e.to_string()) - } else { - format!("Database Error: {e}") - }; - } - if json_mode { - format_json(&json!(rule)) - } else { - format!("✓ Created filter rule '{}' (ID: {}).", rule.name, rule.id) - } - } - Err(e) => { - if json_mode { - format_json_error(&e.to_string()) - } else { - format!("Syntax Error: {e}") - } - } - }, + } => { + self.handle_filter_create(name, sql, *priority, json_mode) + .await + } + FilterSubcommand::Delete { id } => self.handle_filter_delete(id, json_mode).await, + FilterSubcommand::Validate { sql } => { + self.handle_filter_validate(sql, json_mode).await + } + FilterSubcommand::Test { sql, message_id } => { + self.handle_filter_test(sql, message_id.as_deref(), json_mode) + .await + } + // `filter edit`/`export`/`import`/`logs` are NOT part of + // backlog story 2.A (GH #171)'s required `Filters` gRPC + // surface (`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 gap tracked for a later story, not a regression + // introduced silently here. FilterSubcommand::Edit { id, name, @@ -474,80 +466,6 @@ impl HeadlessRunner { format!("Rule '{}' not found.", id) } } - FilterSubcommand::Delete { id } => { - if let Err(e) = self.db.delete_filter_rule(id).await { - if json_mode { - format_json_error(&e.to_string()) - } else { - format!("Error: {e}") - } - } else if json_mode { - format_json(&json!({ "status": "deleted", "id": id })) - } else { - format!("✓ Filter rule '{}' deleted.", id) - } - } - FilterSubcommand::Test { sql, message_id } => { - match nuncio_filter::NsqlParser::parse_rule("Test Rule", 0, sql) { - Ok(rule) => { - let engine = match nuncio_filter::FilterEngine::new(vec![rule.clone()]) - { - Ok(engine) => engine, - Err(e) => { - return if json_mode { - format_json_error(&e) - } else { - format!("Filter Engine Error: {e}") - }; - } - }; - let sample_email = if let Some(mid) = message_id { - self.db.get_message(mid).await.unwrap_or_else(|_| { - nuncio_core::model::Email { - id: mid.clone(), - account_id: "acct-1".to_string(), - folder_id: "inbox".to_string(), - subject: "Test Subject".to_string(), - sender: "test@nuncio.mx".to_string(), - recipient: "me@nuncio.mx".to_string(), - received_at: chrono::Utc::now().timestamp(), - read: false, - body_plain: Some("Sample body text".to_string()), - body_html: None, - attachments: Vec::new(), - } - }) - } else { - nuncio_core::model::Email { - id: "msg-test".to_string(), - account_id: "acct-1".to_string(), - folder_id: "inbox".to_string(), - subject: "Test Subject".to_string(), - sender: "test@nuncio.mx".to_string(), - recipient: "me@nuncio.mx".to_string(), - received_at: chrono::Utc::now().timestamp(), - read: false, - body_plain: Some("Sample body text".to_string()), - body_html: None, - attachments: Vec::new(), - } - }; - let preview = engine.preview(&sample_email); - if json_mode { - format_json(&json!(preview)) - } else { - format!("Dry-run evaluation result: matched={}, actions={:?}, elapsed={}us", preview.matched, preview.actions_evaluated, preview.execution_time_us) - } - } - Err(e) => { - if json_mode { - format_json_error(&e.to_string()) - } else { - format!("Syntax Error: {e}") - } - } - } - } FilterSubcommand::Export { format } => { let rules = self.db.list_filter_rules().await.unwrap_or_default(); if format == "json" || json_mode { @@ -1023,6 +941,230 @@ impl HeadlessRunner { } } + /// `filter list`: a real thin gRPC client of the running `nunciod` + /// daemon's `nuncio.v1.Filters` API (backlog story 2.A, GH #171). Lists + /// every persisted filter rule from the daemon's real, persistent + /// store -- NOT this runner's own ephemeral local `db`, which is thrown + /// away when this CLI process exits. + async fn handle_filter_list(&self, 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 + .list_rules(nuncio_proto::v1::ListRulesRequest {}) + .await + { + Ok(response) => { + let rules = response.into_inner().rules; + if json_mode { + let rules_json: Vec = + rules.iter().map(filter_rule_proto_to_json).collect(); + format_json(&json!({ "rules": rules_json })) + } else if rules.is_empty() { + "No filter rules configured.".to_string() + } else { + let mut out = + String::from("ID PRIORITY ENABLED NAME NSQL\n"); + for r in rules { + out.push_str(&format!( + "{:<10} {:<8} {:<7} {:<20} {}\n", + r.id, r.priority, r.enabled, r.name, r.nsql_text + )); + } + out + } + } + Err(status) => Self::render_error( + &format!("nunciod daemon rejected list_rules: {status}"), + json_mode, + ), + } + } + + /// `filter create`: a real thin gRPC client of the running `nunciod` + /// daemon's `nuncio.v1.Filters` API (backlog story 2.A, GH #171). The + /// daemon parses, validates (6-pass `NsqlValidator`), and persists the + /// rule, then reloads its own live `FilterEngine` -- this runner's own + /// ephemeral local `db` is never touched, so the rule survives this CLI + /// process exiting. + async fn handle_filter_create( + &self, + name: &str, + sql: &str, + priority: i32, + 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 + .create_rule(nuncio_proto::v1::CreateRuleRequest { + name: name.to_string(), + nsql: sql.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!("✓ Created filter rule '{}' (ID: {}).", rule.name, rule.id) + } + } + None => Self::render_error( + "nunciod daemon accepted create_rule but returned no rule", + json_mode, + ), + }, + Err(status) if status.code() == tonic::Code::InvalidArgument => { + Self::render_error(status.message(), json_mode) + } + Err(status) => Self::render_error( + &format!("nunciod daemon rejected create_rule: {status}"), + json_mode, + ), + } + } + + /// `filter delete`: a real thin gRPC client of the running `nunciod` + /// daemon's `nuncio.v1.Filters` API (backlog story 2.A, GH #171). The + /// daemon deletes the persisted rule and reloads its own live + /// `FilterEngine` so the removal takes effect immediately. + async fn handle_filter_delete(&self, id: &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 + .delete_rule(nuncio_proto::v1::DeleteRuleRequest { id: id.to_string() }) + .await + { + Ok(_) => { + if json_mode { + format_json(&json!({ "status": "deleted", "id": id })) + } else { + format!("✓ Filter rule '{}' deleted.", id) + } + } + Err(status) => Self::render_error( + &format!("nunciod daemon rejected delete_rule: {status}"), + json_mode, + ), + } + } + + /// `filter validate`: a real thin gRPC client of the running `nunciod` + /// daemon's `nuncio.v1.Filters` API (backlog story 2.A, GH #171). Unlike + /// `filter create`, an invalid rule is never a connection/RPC failure + /// here -- it is the daemon's honestly reported `valid: false` result, + /// which this renders as a validation error without ever suggesting + /// the daemon itself was unreachable or misbehaving. + async fn handle_filter_validate(&self, sql: &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 + .validate_rule(nuncio_proto::v1::ValidateRuleRequest { + nsql: sql.to_string(), + }) + .await + { + Ok(response) => { + let response = response.into_inner(); + if json_mode { + format_json(&json!({ "valid": response.valid, "error": response.error })) + } else if response.valid { + "✓ NSQL rule is valid.".to_string() + } else { + format!("Validation Error: {}", response.error) + } + } + Err(status) => Self::render_error( + &format!("nunciod daemon rejected validate_rule: {status}"), + json_mode, + ), + } + } + + /// `filter test`: a real thin gRPC client of the running `nunciod` + /// daemon's `nuncio.v1.Filters` API (backlog story 2.A, GH #171). + /// Dry-run evaluates the given NSQL rule against a stored message (by + /// `message_id`, read from the daemon's real, persistent store) or a + /// fixed synthetic sample when none is given/resolvable -- the rule is + /// never persisted and no action is ever executed. + async fn handle_filter_test( + &self, + sql: &str, + message_id: Option<&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 + .preview_rule(nuncio_proto::v1::PreviewRuleRequest { + nsql: sql.to_string(), + message_id: message_id.map(str::to_string), + }) + .await + { + Ok(response) => { + let preview = response.into_inner(); + if json_mode { + format_json(&json!({ + "message_id": preview.message_id, + "matched": preview.matched, + "matched_rule_id": preview.matched_rule_id, + "matched_rule_name": preview.matched_rule_name, + "actions_evaluated": preview.actions_evaluated, + "execution_time_us": preview.execution_time_us, + "condition_traces": preview.condition_traces, + })) + } else { + format!( + "Dry-run evaluation result: matched={}, actions={:?}, elapsed={}us", + preview.matched, preview.actions_evaluated, preview.execution_time_us + ) + } + } + Err(status) if status.code() == tonic::Code::InvalidArgument => { + Self::render_error(status.message(), json_mode) + } + Err(status) => Self::render_error( + &format!("nunciod daemon rejected preview_rule: {status}"), + json_mode, + ), + } + } + + /// Resolves the gRPC bearer token from the injected vault and dials the + /// `nuncio.v1.Filters` service at `self.grpc_addr` (backlog story 2.A, + /// GH #171), shared by every `filter` handler above. + async fn connect_filters_client( + &self, + ) -> Result { + let token_bytes = self + .secrets + .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32) + .map_err(|e| format!("failed to read gRPC bearer token from vault: {e}"))?; + let token = hex::encode(token_bytes); + + nuncio_proto::client::connect_filters(&self.grpc_addr, &token) + .await + .map_err(|e| format!("nunciod daemon unreachable at {}: {e}", self.grpc_addr)) + } + /// `account add`: a real thin gRPC client of the running `nunciod` /// daemon's `nuncio.v1.Accounts` API (backlog story 1.C.1 / 1.C.2, /// GH #156 / GH #157). @@ -2043,6 +2185,268 @@ mod tests { assert_eq!(recorded_send.body_text, "Let's discuss the roadmap."); } + /// Reference-client proof for backlog story 2.A (GH #171): boots a stub + /// `nuncio.v1.Filters` gRPC server (mirroring the `Accounts`/`Mail` stub + /// pattern above) and drives the real `HeadlessRunner`'s + /// `filter list`/`filter create`/`filter delete`/`filter validate`/ + /// `filter test` gRPC client paths against it. + /// + /// Real persistence, `ArcSwap` engine reload, and honest + /// validation/preview semantics are proven by `nunciod`'s own + /// `grpc::tests`; this test exists purely to prove the CLI's connect + + /// call + JSON-format happy path, and its honest invalid-argument error + /// handling. + #[tokio::test] + async fn filter_rpcs_round_trip_over_grpc_to_a_stub_daemon() { + 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, + }; + 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. + #[derive(Default)] + struct StubFilters { + last_delete: Arc>>, + } + + #[tonic::async_trait] + impl FiltersService for StubFilters { + async fn create_rule( + &self, + request: tonic::Request, + ) -> Result, tonic::Status> { + let req = request.into_inner(); + if req.nsql.contains("BROKEN") { + return Err(tonic::Status::invalid_argument( + "NSQL syntax error: stub rejects BROKEN rules", + )); + } + Ok(tonic::Response::new(CreateRuleResponse { + rule: Some(FilterRuleProto { + id: "rule-stub-1".to_string(), + name: req.name, + target_account: "*".to_string(), + priority: req.priority, + enabled: true, + nsql_text: req.nsql, + actions: vec!["MARK READ".to_string()], + created_at: 1_700_000_000, + updated_at: 1_700_000_000, + }), + })) + } + + async fn list_rules( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> { + Ok(tonic::Response::new(ListRulesResponse { + rules: vec![FilterRuleProto { + id: "rule-stub-1".to_string(), + name: "Stub Rule".to_string(), + target_account: "*".to_string(), + priority: 5, + enabled: true, + nsql_text: "WHERE subject CONTAINS 'Urgent' ACTION MARK READ".to_string(), + actions: vec!["MARK READ".to_string()], + created_at: 1_700_000_000, + updated_at: 1_700_000_000, + }], + })) + } + + async fn delete_rule( + &self, + request: tonic::Request, + ) -> Result, tonic::Status> { + let req = request.into_inner(); + *self.last_delete.lock().unwrap_or_else(|e| e.into_inner()) = Some(req); + Ok(tonic::Response::new(DeleteRuleResponse {})) + } + + async fn validate_rule( + &self, + request: tonic::Request, + ) -> Result, tonic::Status> { + let req = request.into_inner(); + if req.nsql.contains("BROKEN") { + Ok(tonic::Response::new(ValidateRuleResponse { + valid: false, + error: "NSQL syntax error: stub rejects BROKEN rules".to_string(), + })) + } else { + Ok(tonic::Response::new(ValidateRuleResponse { + valid: true, + error: String::new(), + })) + } + } + + async fn preview_rule( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> { + Ok(tonic::Response::new(PreviewRuleResponse { + message_id: "msg-stub-1".to_string(), + matched: true, + matched_rule_id: Some("rule-stub-1".to_string()), + matched_rule_name: Some("Stub Rule".to_string()), + actions_evaluated: vec!["MARK READ".to_string()], + execution_time_us: 42, + condition_traces: vec!["Rule 'Stub Rule': MATCH".to_string()], + })) + } + } + + let stub = StubFilters::default(); + let delete_probe = stub.last_delete.clone(); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral loopback port"); + let addr = listener.local_addr().expect("listener has local addr"); + tokio::spawn(async move { + let _ = tonic::transport::Server::builder() + .add_service(FiltersServer::new(stub)) + .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener)) + .await; + }); + + let runner = + HeadlessRunner::ephemeral_with(Arc::new(SecretManager::mock()), addr.to_string()) + .await + .expect("ephemeral runner initializes"); + + // `filter create` happy path. + let create_out = runner + .execute_command( + &Commands::Filter { + action: FilterSubcommand::Create { + name: "My Rule".to_string(), + sql: "WHERE subject CONTAINS 'Urgent' ACTION MARK READ".to_string(), + priority: 5, + }, + }, + true, + ) + .await; + assert!(create_out.contains("rule-stub-1")); + assert!(create_out.contains("My Rule")); + + // `filter create` honest invalid-argument error. + let create_err = runner + .execute_command( + &Commands::Filter { + action: FilterSubcommand::Create { + name: "Broken".to_string(), + sql: "BROKEN NSQL".to_string(), + priority: 0, + }, + }, + true, + ) + .await; + assert!(create_err.contains(r#""status":"error""#)); + assert!(create_err.contains("stub rejects BROKEN rules")); + + // `filter list`. + let list_out = runner + .execute_command( + &Commands::Filter { + action: FilterSubcommand::List, + }, + true, + ) + .await; + assert!(list_out.contains("rule-stub-1")); + assert!(list_out.contains("Stub Rule")); + + // `filter delete`. + let delete_out = runner + .execute_command( + &Commands::Filter { + action: FilterSubcommand::Delete { + id: "rule-stub-1".to_string(), + }, + }, + true, + ) + .await; + assert!(delete_out.contains(r#""status":"deleted""#)); + let recorded_delete = delete_probe + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone() + .expect("stub daemon received a delete_rule request"); + assert_eq!(recorded_delete.id, "rule-stub-1"); + + // `filter validate` valid + invalid. + let validate_ok = runner + .execute_command( + &Commands::Filter { + action: FilterSubcommand::Validate { + sql: "WHERE subject CONTAINS 'Urgent' ACTION MARK READ".to_string(), + }, + }, + true, + ) + .await; + assert!(validate_ok.contains(r#""valid":true"#)); + + let validate_err = runner + .execute_command( + &Commands::Filter { + action: FilterSubcommand::Validate { + sql: "BROKEN NSQL".to_string(), + }, + }, + true, + ) + .await; + assert!(validate_err.contains(r#""valid":false"#)); + assert!(validate_err.contains("stub rejects BROKEN rules")); + + // `filter test` (PreviewRule) dry-run. + let test_out = runner + .execute_command( + &Commands::Filter { + action: FilterSubcommand::Test { + sql: "WHERE subject CONTAINS 'Urgent' ACTION MARK READ".to_string(), + message_id: Some("msg-stub-1".to_string()), + }, + }, + true, + ) + .await; + assert!(test_out.contains(r#""matched":true"#)); + assert!(test_out.contains("msg-stub-1")); + } + + #[tokio::test] + async fn filter_commands_report_honest_error_when_daemon_unreachable() { + let addr = reserve_unreachable_addr().await; + let runner = HeadlessRunner::ephemeral_with(Arc::new(SecretManager::mock()), addr) + .await + .expect("ephemeral runner initializes"); + + let out = runner + .execute_command( + &Commands::Filter { + action: FilterSubcommand::List, + }, + true, + ) + .await; + assert!(out.contains(r#""status":"error""#)); + assert!(out.contains("unreachable")); + } + #[test] fn runner_error_display() { let err = RunnerError::InitFailed("failed to open database".to_string()); diff --git a/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto b/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto index 76caea3..166c7e6 100644 --- a/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto +++ b/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto @@ -422,3 +422,162 @@ message SyncRequest { message SyncResponse { uint64 synced_count = 1; } + +// Filters exposes the NSQL filter engine's rule CRUD, validation, and +// dry-run preview operations over gRPC (backlog story 2.A, GH #171) -- the +// authenticated replacement for the existing hand-rolled `filter.*` JSON-RPC +// IPC methods (`nunciod::main`'s `CustomRpcHandler`), which remain in place +// (both transports coexist) until they are retired in a later story (2.C). +// +// This faithfully exposes the filter engine as it exists today, including +// its known correctness gaps (e.g. `target_account`/`ON ACCOUNT` matching +// against `folder_id` rather than the message's actual account field) -- +// fixing those is out of scope here (Phase 3). +// +// SECURITY (GH #165): this service MUST be mounted behind the SAME +// `BearerAuthInterceptor` that guards `System`/`Accounts`/`Mail` -- every +// gRPC service `nunciod` exposes MUST be authenticated, with no exceptions +// carved out for "just one more" unauthenticated service. See +// `nunciod::grpc::serve_on_listener`. +service Filters { + // CreateRule parses and validates an NSQL rule definition, then persists + // it and reloads the daemon's live `FilterEngine` rule set (via its + // `ArcSwap`-backed hot-reload) so the new rule is active immediately. + // Rejects a syntactically or semantically invalid rule with + // `Status::invalid_argument` -- nothing is ever persisted on failure. + rpc CreateRule(CreateRuleRequest) returns (CreateRuleResponse); + // ListRules returns every persisted filter rule, priority ascending. + rpc ListRules(ListRulesRequest) returns (ListRulesResponse); + // DeleteRule removes a persisted filter rule by ID and reloads the + // daemon's live `FilterEngine` rule set so the removal takes effect + // immediately. + rpc DeleteRule(DeleteRuleRequest) returns (DeleteRuleResponse); + // ValidateRule parses and runs the 6-pass `NsqlValidator` audit against + // an NSQL rule definition WITHOUT persisting it. Unlike `CreateRule`, + // an invalid rule is NOT an RPC failure: it is the expected, honestly + // reported result carried in `ValidateRuleResponse.valid`/`error`. + rpc ValidateRule(ValidateRuleRequest) returns (ValidateRuleResponse); + // PreviewRule dry-run evaluates an ad hoc, NOT-yet-persisted NSQL rule + // definition against a stored message (by `message_id`) -- or, if none is + // given, a synthetic sample message -- using `FilterEngine::preview`'s + // existing dry-run evaluation path. Never persists the rule and never + // executes any of its actions. + rpc PreviewRule(PreviewRuleRequest) returns (PreviewRuleResponse); +} + +// FilterRule mirrors `nuncio_filter::FilterRule`'s persisted/user-facing +// fields. The parsed condition AST is intentionally NOT mirrored here -- +// `nsql_text` is the single source of truth, and re-parsing it via +// `NsqlParser::parse_rule` deterministically reconstructs the identical AST, +// so there is no independent state to carry over the wire. `actions` is the +// NSQL-rendered form (`RuleAction::to_nsql`) of the rule's parsed action +// list, included for display purposes. +message FilterRule { + // Rule unique identifier. + string id = 1; + // Rule display name. + string name = 2; + // Target account pattern ("*" for all accounts, an exact address, or a + // wildcard like "%@kof22.com"). + string target_account = 3; + // Rule evaluation order (0 = highest). + int32 priority = 4; + // Active flag. + bool enabled = 5; + // Full NSQL source code text. + string nsql_text = 6; + // Rendered NSQL action clauses (e.g. "MOVE TO 'Archive'"), derived from + // the parsed AST. + repeated string actions = 7; + // Creation timestamp (unix seconds). + int64 created_at = 8; + // Last update timestamp (unix seconds). + int64 updated_at = 9; +} + +// CreateRuleRequest carries a new rule's display name, full NSQL source +// text, and evaluation priority. +message CreateRuleRequest { + // Display name for the rule. + string name = 1; + // Full NSQL source code text (may include an `ON ACCOUNT` clause and + // `ACTION` clause). + string nsql = 2; + // Evaluation order (0 = highest). + int32 priority = 3; +} + +// CreateRuleResponse carries the persisted rule, including its +// daemon-assigned `id` and timestamps. +message CreateRuleResponse { + FilterRule rule = 1; +} + +// ListRulesRequest currently carries no parameters. +message ListRulesRequest {} + +// ListRulesResponse. +message ListRulesResponse { + repeated FilterRule rules = 1; +} + +// DeleteRuleRequest. +message DeleteRuleRequest { + // Rule unique identifier. + string id = 1; +} + +// DeleteRuleResponse currently carries no parameters; a successful RPC +// response (as opposed to a `Status` error) is confirmation enough. +message DeleteRuleResponse {} + +// ValidateRuleRequest carries only the raw NSQL source text -- validation +// never persists anything, so no name/priority/id is needed. +message ValidateRuleRequest { + string nsql = 1; +} + +// ValidateRuleResponse. `valid == false` is an ordinary, successfully +// returned result (never a `Status` error) carrying the honest parse or +// `NsqlValidator` failure message in `error`. +message ValidateRuleResponse { + // Whether `nsql` parsed AND passed every `NsqlValidator` pass. + bool valid = 1; + // Empty when `valid` is true; otherwise the parse or validation error + // message. + string error = 2; +} + +// PreviewRuleRequest carries an ad hoc, NOT-yet-persisted NSQL rule +// definition to dry-run evaluate. `message_id` optionally names a stored +// message to evaluate against (looked up via the same store `Mail/ +// GetMessage` reads); when absent (or when `message_id` does not resolve to +// a stored message), a fixed synthetic sample message is evaluated instead, +// so this RPC can always return a result. +message PreviewRuleRequest { + // Full NSQL source code text of the rule to preview. + string nsql = 1; + // Optional stored message id to evaluate the rule against. + optional string message_id = 2; +} + +// PreviewRuleResponse mirrors `nuncio_filter::FilterPreviewResult` +// field-for-field, except `actions_evaluated` is rendered as NSQL strings +// (`RuleAction::to_nsql`) rather than a structured action AST. +message PreviewRuleResponse { + // Evaluated message id (either the requested `message_id`, or the + // synthetic sample message's fixed id when none was given/resolvable). + string message_id = 1; + // Whether the previewed rule matched the evaluated message. + bool matched = 2; + // Present only when `matched` is true. + optional string matched_rule_id = 3; + // Present only when `matched` is true. + optional string matched_rule_name = 4; + // NSQL-rendered actions that would be executed on a real match. + repeated string actions_evaluated = 5; + // Evaluation duration in microseconds. + uint64 execution_time_us = 6; + // Human-readable step-by-step match traces. + repeated string condition_traces = 7; +} diff --git a/crates/nuncio-proto/src/client.rs b/crates/nuncio-proto/src/client.rs index 35820f0..5f81ee7 100644 --- a/crates/nuncio-proto/src/client.rs +++ b/crates/nuncio-proto/src/client.rs @@ -10,6 +10,7 @@ //! server-side interceptor in `nunciod::grpc`. use crate::v1::accounts_client::AccountsClient; +use crate::v1::filters_client::FiltersClient; use crate::v1::mail_client::MailClient; use crate::v1::system_client::SystemClient; use crate::v1::{Event, SubscribeRequest}; @@ -94,6 +95,11 @@ pub type AuthenticatedAccountsClient = /// [`connect_mail`]. pub type AuthenticatedMailClient = MailClient>; +/// The authenticated `nuncio.v1.Filters` client type returned by +/// [`connect_filters`]. +pub type AuthenticatedFiltersClient = + FiltersClient>; + /// Dials `addr` (a `host:port` pair, e.g. `127.0.0.1:9420`) over plain HTTP /// (the loopback gRPC transport is never TLS-wrapped; auth is via bearer /// token instead) and returns the connected [`Channel`], shared by every @@ -164,6 +170,25 @@ pub async fn connect_mail( Ok(MailClient::with_interceptor(channel, interceptor)) } +/// Dials the `nuncio.v1.Filters` gRPC endpoint at `addr` and returns a +/// client that injects `authorization: Bearer ` metadata on every +/// call (backlog story 2.A, GH #171). +/// +/// `Filters` is guarded by the exact same `BearerAuthInterceptor` as +/// `System`/`Accounts`/`Mail` on the server side (see +/// `nunciod::grpc::serve_on_listener`, GH #165), so this shares +/// [`BearerTokenInterceptor`] and [`dial`] with [`connect_system`] / +/// [`connect_accounts`] / [`connect_mail`] rather than hand-rolling a fourth +/// auth handshake. +pub async fn connect_filters( + addr: &str, + token: &str, +) -> Result { + let interceptor = BearerTokenInterceptor::new(token)?; + let channel = dial(addr).await?; + Ok(FiltersClient::with_interceptor(channel, interceptor)) +} + /// Opens the `nuncio.v1.System/Subscribe` server-streaming RPC on an already /// authenticated client (as returned by [`connect_system`]), reusing the /// bearer-token interceptor already attached to `client` rather than @@ -272,6 +297,23 @@ mod tests { assert!(err.to_string().contains("127.0.0.1:1")); } + #[tokio::test] + async fn connect_filters_fails_closed_on_invalid_token() { + let err = connect_filters("127.0.0.1:0", "tok\ntoken") + .await + .expect_err("invalid token must fail before dialing"); + assert!(matches!(err, ConnectError::InvalidToken(_))); + } + + #[tokio::test] + async fn connect_filters_reports_transport_error_when_daemon_unreachable() { + let err = connect_filters("127.0.0.1:1", "abc123") + .await + .expect_err("connecting to an unreachable daemon must fail"); + assert!(matches!(err, ConnectError::Transport { .. })); + assert!(err.to_string().contains("127.0.0.1:1")); + } + #[tokio::test] async fn connect_system_reports_invalid_endpoint_for_a_malformed_address() { // An embedded space is not a legal URI authority character, so diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs index 6d10fac..e416ca9 100644 --- a/crates/nunciod/src/grpc.rs +++ b/crates/nunciod/src/grpc.rs @@ -8,27 +8,33 @@ //! transport is out of scope here and lands in a later story. use nuncio_core::{CoreCommand, CoreEvent, EventBus}; +use nuncio_filter::{FilterEngine, NsqlParser, NsqlValidator, ValidationOptions}; use nuncio_mail::{MailBackend, MessageSender}; use nuncio_proto::v1::accounts_server::{Accounts, AccountsServer}; use nuncio_proto::v1::event::Kind; +use nuncio_proto::v1::filters_server::{Filters, FiltersServer}; use nuncio_proto::v1::mail_server::{Mail, MailServer}; use nuncio_proto::v1::system_server::{System, SystemServer}; use nuncio_proto::v1::{ AccountConfig as AccountConfigProto, AccountProtocol as AccountProtocolProto, AddAccountRequest, AddAccountResponse, Attachment as AttachmentProto, BatchFilterProgress, - DatabaseRecovered, Event, EventError, FilterExecuted, Folder as FolderProto, GetMessageRequest, - GetMessageResponse, GetStatusRequest, GetStatusResponse, ListAccountsRequest, - ListAccountsResponse, ListFoldersRequest, ListFoldersResponse, ListMessagesRequest, - ListMessagesResponse, MarkReadRequest, MarkReadResponse, Message as MessageProto, - MessageFlagsChanged, MessageSearchHit, SearchMessagesRequest, SearchMessagesResponse, - SendMessageRequest, SendMessageResponse, ShuttingDown, SubscribeRequest, SyncCompleted, - SyncRequest, SyncResponse, SyncStarted, TlsMode as TlsModeProto, UpdateAvailable, + CreateRuleRequest, CreateRuleResponse, DatabaseRecovered, DeleteRuleRequest, + DeleteRuleResponse, Event, EventError, FilterExecuted, FilterRule as FilterRuleProto, + Folder as FolderProto, GetMessageRequest, GetMessageResponse, GetStatusRequest, + GetStatusResponse, ListAccountsRequest, ListAccountsResponse, ListFoldersRequest, + ListFoldersResponse, ListMessagesRequest, ListMessagesResponse, 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, }; use nuncio_store::db::DatabaseEngine; use nuncio_store::search::SearchEngine; use nuncio_store::vault::SecretManager; use std::pin::Pin; use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; use subtle::ConstantTimeEq; use thiserror::Error; use tokio::net::TcpListener; @@ -672,6 +678,234 @@ impl Mail for MailGrpcService { } } +/// Maps a `nuncio_filter::FilterRule` onto its wire-format +/// `nuncio.v1.FilterRule` representation (backlog story 2.A, GH #171). The +/// parsed condition AST is deliberately NOT carried over the wire -- see the +/// message's doc comment in `proto/nuncio/v1/nuncio.proto` -- `actions` is +/// rendered as NSQL strings via `RuleAction::to_nsql`. +fn map_filter_rule_to_proto(rule: nuncio_filter::FilterRule) -> FilterRuleProto { + FilterRuleProto { + id: rule.id, + name: rule.name, + target_account: rule.target_account, + priority: rule.priority, + enabled: rule.enabled, + nsql_text: rule.nsql_text, + actions: rule.actions.iter().map(|a| a.to_nsql()).collect(), + created_at: rule.created_at, + updated_at: rule.updated_at, + } +} + +/// Maps a `nuncio_filter::FilterPreviewResult` onto its wire-format +/// `nuncio.v1.PreviewRuleResponse` representation. +fn map_preview_result_to_proto(preview: nuncio_filter::FilterPreviewResult) -> PreviewRuleResponse { + PreviewRuleResponse { + message_id: preview.message_id, + matched: preview.matched, + matched_rule_id: preview.matched_rule_id, + matched_rule_name: preview.matched_rule_name, + actions_evaluated: preview + .actions_evaluated + .iter() + .map(|a| a.to_nsql()) + .collect(), + execution_time_us: preview.execution_time_us, + condition_traces: preview.condition_traces, + } +} + +/// Builds a fixed synthetic sample message for `PreviewRule` to evaluate +/// against when the caller supplies no `message_id` (or one that does not +/// resolve to a stored message), mirroring the exact fallback sample the +/// CLI's previous local `filter test` path used -- so this RPC can always +/// return a preview result rather than requiring pre-existing stored mail. +fn synthetic_preview_email(id: &str) -> nuncio_core::model::Email { + let received_at = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or_default(); + nuncio_core::model::Email { + id: id.to_string(), + account_id: "acct-1".to_string(), + folder_id: "inbox".to_string(), + subject: "Test Subject".to_string(), + sender: "test@nuncio.mx".to_string(), + recipient: "me@nuncio.mx".to_string(), + received_at, + read: false, + body_plain: Some("Sample body text".to_string()), + body_html: None, + attachments: Vec::new(), + } +} + +/// `nuncio.v1.Filters` gRPC service implementation backed by the daemon's +/// live [`DatabaseEngine`] filter-rule CRUD and the daemon's live +/// [`FilterEngine`] (backlog story 2.A, GH #171) -- the authenticated gRPC +/// replacement for the existing hand-rolled `filter.*` JSON-RPC IPC methods +/// (`nunciod::main`'s `CustomRpcHandler`), which remain in place (both +/// transports coexist) until retired in a later story (2.C). +/// +/// `CreateRule`/`DeleteRule` persist through `db` FIRST, then reload +/// `filter_engine`'s `ArcSwap`-backed rule set from the freshly persisted +/// state via `db.list_filter_rules()`, so the live engine and the persisted +/// store never disagree about which rules exist. A reload failure (e.g. a +/// rule whose regex no longer compiles) is logged but does NOT fail the +/// RPC: the mutation already succeeded in the store, so failing the RPC +/// here would misreport a successful write as an error. +struct FiltersGrpcService { + db: Arc, + filter_engine: Arc, +} + +impl FiltersGrpcService { + /// Reloads `self.filter_engine`'s active rule set from `self.db`'s + /// current persisted rules, logging (rather than propagating) a reload + /// failure -- shared by `create_rule` and `delete_rule` after each + /// persists its own mutation. + async fn reload_engine_from_store(&self) { + match self.db.list_filter_rules().await { + Ok(rules) => { + if let Err(e) = self.filter_engine.reload_rules(rules) { + tracing::warn!("Filters: failed to reload live FilterEngine rules: {}", e); + } + } + Err(e) => { + tracing::warn!( + "Filters: failed to re-list persisted filter rules for engine reload: {}", + e + ); + } + } + } +} + +#[tonic::async_trait] +impl Filters for FiltersGrpcService { + async fn create_rule( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + + let rule = NsqlParser::parse_rule(req.name, req.priority, &req.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}")))?; + + 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(CreateRuleResponse { + rule: Some(map_filter_rule_to_proto(rule)), + })) + } + + async fn list_rules( + &self, + _request: Request, + ) -> Result, Status> { + let rules = self + .db + .list_filter_rules() + .await + .map_err(|e| Status::internal(format!("failed to list filter rules: {e}")))? + .into_iter() + .map(map_filter_rule_to_proto) + .collect(); + + Ok(Response::new(ListRulesResponse { rules })) + } + + async fn delete_rule( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + if req.id.is_empty() { + return Err(Status::invalid_argument("id is required")); + } + + self.db + .delete_filter_rule(&req.id) + .await + .map_err(|e| Status::internal(format!("failed to delete filter rule: {e}")))?; + + self.reload_engine_from_store().await; + + Ok(Response::new(DeleteRuleResponse {})) + } + + /// Unlike `create_rule`, an invalid rule is never an RPC failure here -- + /// see this RPC's doc comment in `proto/nuncio/v1/nuncio.proto`. + async fn validate_rule( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + + let rule = match NsqlParser::parse_rule("Validation Preview", 0, &req.nsql) { + Ok(rule) => rule, + Err(e) => { + return Ok(Response::new(ValidateRuleResponse { + valid: false, + error: format!("NSQL syntax error: {e}"), + })); + } + }; + + match NsqlValidator::validate(&rule, &ValidationOptions::default()) { + Ok(()) => Ok(Response::new(ValidateRuleResponse { + valid: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(ValidateRuleResponse { + valid: false, + error: format!("NSQL validation error: {e}"), + })), + } + } + + /// Dry-run evaluates an ad hoc, NOT-yet-persisted NSQL rule against a + /// stored message (or a synthetic sample -- see + /// [`synthetic_preview_email`]), via a fresh one-off [`FilterEngine`] + /// built from just that single rule and its existing `preview` path. + /// This deliberately does NOT use `self.filter_engine` (the daemon's + /// live, persisted rule set): the whole point of a preview is to try out + /// a candidate rule that may not be, and may never be, saved. + async fn preview_rule( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + + let rule = NsqlParser::parse_rule("Preview Rule", 0, &req.nsql) + .map_err(|e| Status::invalid_argument(format!("NSQL syntax error: {e}")))?; + + let preview_engine = FilterEngine::new(vec![rule]) + .map_err(|e| Status::internal(format!("failed to build preview engine: {e}")))?; + + let email = match &req.message_id { + Some(message_id) if !message_id.is_empty() => { + match self.db.get_message(message_id).await { + Ok(email) => email, + Err(_) => synthetic_preview_email(message_id), + } + } + _ => synthetic_preview_email("msg-test"), + }; + + let preview = preview_engine.preview(&email); + Ok(Response::new(map_preview_result_to_proto(preview))) + } +} + /// Bearer-token authentication interceptor for the loopback `nuncio.v1` gRPC /// server. /// @@ -719,8 +953,9 @@ impl tonic::service::Interceptor for BearerAuthInterceptor { } /// Binds a loopback TCP listener at `addr` and serves the `nuncio.v1.System`, -/// `nuncio.v1.Accounts`, and `nuncio.v1.Mail` gRPC services on it, all -/// authenticated by `token`, until the transport server errors. +/// `nuncio.v1.Accounts`, `nuncio.v1.Mail`, and `nuncio.v1.Filters` gRPC +/// services on it, all authenticated by `token`, until the transport server +/// errors. /// /// `addr` MUST be a loopback address (e.g. `127.0.0.1:PORT`); callers are /// responsible for passing loopback-only addresses (see @@ -729,6 +964,7 @@ pub async fn serve( addr: &str, event_bus: Arc, db: Arc, + filter_engine: Arc, secrets: Arc, token: impl Into>, ) -> Result<(), GrpcServeError> { @@ -738,11 +974,11 @@ pub async fn serve( addr: addr.to_string(), source, })?; - serve_on_listener(listener, event_bus, db, secrets, token).await + serve_on_listener(listener, event_bus, db, filter_engine, secrets, token).await } -/// Serves the `nuncio.v1.System`, `nuncio.v1.Accounts`, and `nuncio.v1.Mail` -/// gRPC services on an already-bound [`TcpListener`]. +/// Serves the `nuncio.v1.System`, `nuncio.v1.Accounts`, `nuncio.v1.Mail`, and +/// `nuncio.v1.Filters` gRPC services on an already-bound [`TcpListener`]. /// /// # Security (GH #165) /// @@ -763,6 +999,7 @@ pub async fn serve_on_listener( listener: TcpListener, event_bus: Arc, db: Arc, + filter_engine: Arc, secrets: Arc, token: impl Into>, ) -> Result<(), GrpcServeError> { @@ -770,6 +1007,7 @@ pub async fn serve_on_listener( listener, event_bus, db, + filter_engine, secrets, token, MailEngineOverrides::default(), @@ -788,6 +1026,7 @@ pub async fn serve_on_listener_with_overrides( listener: TcpListener, event_bus: Arc, db: Arc, + filter_engine: Arc, secrets: Arc, token: impl Into>, overrides: MailEngineOverrides, @@ -812,18 +1051,27 @@ pub async fn serve_on_listener_with_overrides( // exactly like `System` and `Accounts` above -- see the hard invariant // documented on this function's doc comment (GH #165). let mail_service = MailGrpcService { - db, + db: db.clone(), event_bus, - secrets, + secrets: secrets.clone(), overrides, }; - let mail_interceptor = BearerAuthInterceptor::new(token); + let mail_interceptor = BearerAuthInterceptor::new(token.clone()); let mail_svc = MailServer::with_interceptor(mail_service, mail_interceptor); + // Filters (backlog story 2.A, GH #171): mounted behind its own + // `BearerAuthInterceptor`, exactly like `System`/`Accounts`/`Mail` above + // -- see the hard invariant documented on this function's doc comment + // (GH #165). + let filters_service = FiltersGrpcService { db, filter_engine }; + let filters_interceptor = BearerAuthInterceptor::new(token); + let filters_svc = FiltersServer::with_interceptor(filters_service, filters_interceptor); + Server::builder() .add_service(system_svc) .add_service(accounts_svc) .add_service(mail_svc) + .add_service(filters_svc) .serve_with_incoming(TcpListenerStream::new(listener)) .await .map_err(GrpcServeError::Transport) @@ -837,12 +1085,15 @@ mod tests { use nuncio_proto::v1::system_client::SystemClient; use tonic::Code; - /// Spawns a test server backed by a fresh ephemeral database and a - /// fresh [`SecretManager::mock`] vault -- the right default for every - /// test that only exercises `System` or doesn't care about pre-existing - /// account/keyring state. Tests that DO care (persistence-across-restart, - /// credential-secrecy) build their own `db`/`secrets` and call - /// [`spawn_test_server_with`] directly instead. + /// Spawns a test server backed by a fresh ephemeral database, a fresh + /// empty [`FilterEngine`], and a fresh [`SecretManager::mock`] vault -- + /// the right default for every test that only exercises + /// `System`/`Accounts`/`Mail` or doesn't care about pre-existing + /// account/keyring/filter-rule state. Tests that DO care + /// (persistence-across-restart, credential-secrecy, the `Filters` + /// service itself) build their own `db`/`secrets`/`filter_engine` and + /// call [`spawn_test_server_with`] / [`spawn_test_server_with_overrides`] + /// directly instead. async fn spawn_test_server( event_bus: Arc, token: &str, @@ -851,21 +1102,25 @@ mod tests { .await .expect("connect ephemeral test db"); let secrets = Arc::new(SecretManager::mock()); - spawn_test_server_with(event_bus, Arc::new(db), secrets, token).await + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); + spawn_test_server_with(event_bus, Arc::new(db), filter_engine, secrets, token).await } /// Spawns a test server on an ephemeral loopback port backed by the - /// given `db` and `secrets`, so tests can share (and re-open) the same - /// database path / mock keyring state across multiple server instances. + /// given `db`, `filter_engine`, and `secrets`, so tests can share (and + /// re-open) the same database path / engine instance / mock keyring + /// state across multiple server instances. async fn spawn_test_server_with( event_bus: Arc, db: Arc, + filter_engine: Arc, secrets: Arc, token: &str, ) -> (std::net::SocketAddr, tokio::task::JoinHandle<()>) { spawn_test_server_with_overrides( event_bus, db, + filter_engine, secrets, token, MailEngineOverrides::default(), @@ -874,14 +1129,15 @@ mod tests { } /// Spawns a test server on an ephemeral loopback port backed by the - /// given `db`/`secrets`/`overrides` (backlog story 1.C.6, GH #161): the - /// helper every test that injects a [`MailEngineOverrides::mail_backend`] - /// / [`MailEngineOverrides::message_sender`] uses to prove `Sync` / - /// `SendMessage` drive an injected test double over the real - /// authenticated gRPC API. + /// given `db`/`filter_engine`/`secrets`/`overrides` (backlog story + /// 1.C.6, GH #161): the helper every test that injects a + /// [`MailEngineOverrides::mail_backend`] / [`MailEngineOverrides::message_sender`] + /// uses to prove `Sync` / `SendMessage` drive an injected test double + /// over the real authenticated gRPC API. async fn spawn_test_server_with_overrides( event_bus: Arc, db: Arc, + filter_engine: Arc, secrets: Arc, token: &str, overrides: MailEngineOverrides, @@ -893,7 +1149,13 @@ mod tests { let token = token.to_string(); let handle = tokio::spawn(async move { let _ = serve_on_listener_with_overrides( - listener, event_bus, db, secrets, token, overrides, + listener, + event_bus, + db, + filter_engine, + secrets, + token, + overrides, ) .await; }); @@ -1021,8 +1283,17 @@ mod tests { .await .expect("connect ephemeral test db"); let secrets = Arc::new(SecretManager::mock()); + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); let _handle = tokio::spawn(async move { - let _ = serve("127.0.0.1:0", event_bus, Arc::new(db), secrets, "token").await; + let _ = serve( + "127.0.0.1:0", + event_bus, + Arc::new(db), + filter_engine, + secrets, + "token", + ) + .await; }); // Yield so the spawned task is polled at least once and actually // reaches the bind + delegate-to-`serve_on_listener` call before @@ -1037,10 +1308,12 @@ mod tests { .await .expect("connect ephemeral test db"); let secrets = Arc::new(SecretManager::mock()); + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); let err = serve( "not-a-valid-addr", event_bus, Arc::new(db), + filter_engine, secrets, "token", ) @@ -1313,6 +1586,7 @@ mod tests { let (addr, _handle) = spawn_test_server_with( Arc::new(EventBus::new()), db_first_run.clone(), + Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")), secrets.clone(), "correct-token", ) @@ -1348,6 +1622,7 @@ mod tests { let (addr2, _handle2) = spawn_test_server_with( Arc::new(EventBus::new()), db_second_run, + Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")), secrets.clone(), "correct-token", ) @@ -1392,6 +1667,7 @@ mod tests { let (addr, _handle) = spawn_test_server_with( Arc::new(EventBus::new()), db.clone(), + Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")), secrets.clone(), "correct-token", ) @@ -1657,8 +1933,15 @@ mod tests { let event_bus = Arc::new(EventBus::new()); let mut events = event_bus.subscribe_events(); let secrets = Arc::new(SecretManager::mock()); - let (addr, _handle) = - spawn_test_server_with(event_bus, Arc::new(db), secrets, "correct-token").await; + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); + let (addr, _handle) = spawn_test_server_with( + event_bus, + Arc::new(db), + filter_engine, + secrets, + "correct-token", + ) + .await; let mut client = MailClient::connect(format!("http://{addr}")) .await @@ -1853,8 +2136,15 @@ mod tests { .set_secret(&config.keyring_secret_key, "irrelevant-password") .expect("store credential in mock vault"); - let (addr, _handle) = - spawn_test_server_with(Arc::new(EventBus::new()), db, secrets, "correct-token").await; + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); + let (addr, _handle) = spawn_test_server_with( + Arc::new(EventBus::new()), + db, + filter_engine, + secrets, + "correct-token", + ) + .await; let mut client = MailClient::connect(format!("http://{addr}")) .await @@ -1944,9 +2234,11 @@ mod tests { mail_backend: Some(Arc::new(mock_backend)), message_sender: None, }; + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); let (addr, _handle) = spawn_test_server_with_overrides( Arc::new(EventBus::new()), db, + filter_engine, secrets, "correct-token", overrides, @@ -2017,9 +2309,11 @@ mod tests { mail_backend: None, message_sender: Some(Arc::new(mock_sender)), }; + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); let (addr, _handle) = spawn_test_server_with_overrides( Arc::new(EventBus::new()), db, + filter_engine, secrets, "correct-token", overrides, @@ -2071,9 +2365,11 @@ mod tests { mail_backend: Some(Arc::new(mock_backend)), message_sender: None, }; + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); let (addr, _handle) = spawn_test_server_with_overrides( Arc::new(EventBus::new()), db, + filter_engine, secrets, "correct-token", overrides, @@ -2126,9 +2422,11 @@ mod tests { mail_backend: None, message_sender: Some(Arc::new(mock_sender)), }; + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); let (addr, _handle) = spawn_test_server_with_overrides( Arc::new(EventBus::new()), db, + filter_engine, secrets, "correct-token", overrides, @@ -2145,4 +2443,406 @@ mod tests { assert_eq!(err.code(), Code::Internal); assert!(err.message().contains("failed to send message")); } + + // ---- Filters (backlog story 2.A, GH #171) ---- + + use nuncio_proto::v1::filters_client::FiltersClient; + use nuncio_proto::v1::{ + CreateRuleRequest, DeleteRuleRequest, ListRulesRequest, PreviewRuleRequest, + ValidateRuleRequest, + }; + + const SAMPLE_RULE_NSQL: &str = + "WHERE subject CONTAINS 'Urgent' ACTION MARK READ, MOVE TO 'Priority'"; + + /// Spawns a test server on an ephemeral loopback port backed by a fresh + /// ephemeral database and a fresh empty [`FilterEngine`], returning the + /// address AND the shared `filter_engine`/`db` handles so `Filters` + /// tests can assert directly on live engine state (proving the + /// `ArcSwap` reload actually happened) in addition to driving the gRPC + /// API. + async fn spawn_filters_test_server( + token: &str, + ) -> ( + std::net::SocketAddr, + tokio::task::JoinHandle<()>, + Arc, + Arc, + ) { + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("connect ephemeral test db"); + let db = Arc::new(db); + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); + let secrets = Arc::new(SecretManager::mock()); + let (addr, handle) = spawn_test_server_with( + Arc::new(EventBus::new()), + db.clone(), + filter_engine.clone(), + secrets, + token, + ) + .await; + (addr, handle, db, filter_engine) + } + + #[tokio::test] + async fn filters_rpcs_reject_missing_bearer_token() { + // Confirms `Filters` is mounted behind its own `BearerAuthInterceptor` + // exactly like `System`/`Accounts`/`Mail` (GH #165: no un-intercepted + // service). + let (addr, _handle, _db, _engine) = spawn_filters_test_server("correct-token").await; + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let err = client + .create_rule(CreateRuleRequest { + name: "Unauthed".to_string(), + nsql: SAMPLE_RULE_NSQL.to_string(), + priority: 0, + }) + .await + .expect_err("missing bearer token must be rejected"); + assert_eq!(err.code(), Code::Unauthenticated); + + let err = client + .list_rules(ListRulesRequest {}) + .await + .expect_err("missing bearer token must be rejected"); + assert_eq!(err.code(), Code::Unauthenticated); + + let err = client + .delete_rule(DeleteRuleRequest { + id: "rule-1".to_string(), + }) + .await + .expect_err("missing bearer token must be rejected"); + assert_eq!(err.code(), Code::Unauthenticated); + + let err = client + .validate_rule(ValidateRuleRequest { + nsql: SAMPLE_RULE_NSQL.to_string(), + }) + .await + .expect_err("missing bearer token must be rejected"); + assert_eq!(err.code(), Code::Unauthenticated); + + let err = client + .preview_rule(PreviewRuleRequest { + nsql: SAMPLE_RULE_NSQL.to_string(), + message_id: None, + }) + .await + .expect_err("missing bearer token must be rejected"); + assert_eq!(err.code(), Code::Unauthenticated); + } + + /// End-to-end proof for backlog story 2.A (GH #171): `CreateRule` + /// persists AND reloads the live `FilterEngine`'s `ArcSwap` rule set + /// (proven by evaluating the SAME `filter_engine` instance directly, + /// not just re-reading it back over `ListRules`); `ListRules` reflects + /// the persisted rule; `DeleteRule` removes it AND reloads the engine + /// back to empty. + #[tokio::test] + async fn create_list_and_delete_rule_round_trip_and_keep_the_live_engine_in_sync() { + let (addr, _handle, _db, filter_engine) = spawn_filters_test_server("correct-token").await; + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let sample_email = |subject: &str| nuncio_core::model::Email { + id: "msg-filters-1".to_string(), + account_id: "acct-1".to_string(), + folder_id: "inbox".to_string(), + subject: subject.to_string(), + sender: "alice@nuncio.mx".to_string(), + recipient: "bob@nuncio.mx".to_string(), + received_at: 1_700_000_000, + read: false, + body_plain: Some("body".to_string()), + body_html: None, + attachments: Vec::new(), + }; + + // Before creating anything, the live engine matches nothing. + assert!(filter_engine + .evaluate(&sample_email("Urgent Meeting")) + .is_empty()); + + let create_response = client + .create_rule(authed_bearer_request(CreateRuleRequest { + name: "Urgent Rule".to_string(), + nsql: SAMPLE_RULE_NSQL.to_string(), + priority: 5, + })) + .await + .expect("create_rule succeeds") + .into_inner(); + let created_rule = create_response.rule.expect("response carries the rule"); + assert_eq!(created_rule.name, "Urgent Rule"); + assert_eq!(created_rule.priority, 5); + assert!(created_rule.enabled); + assert_eq!(created_rule.nsql_text, SAMPLE_RULE_NSQL); + assert!(created_rule.actions.iter().any(|a| a.contains("MARK READ"))); + let rule_id = created_rule.id.clone(); + + // The live `FilterEngine`'s `ArcSwap` rule set was reloaded -- + // proven by evaluating the SAME instance the server holds, not a + // fresh one. + let matches = filter_engine.evaluate(&sample_email("Urgent Meeting")); + assert_eq!(matches.len(), 1); + assert_eq!(matches[0].0.id, rule_id); + + // ListRules reflects the persisted rule. + 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].id, rule_id); + + // DeleteRule removes it AND reloads the engine back to empty. + client + .delete_rule(authed_bearer_request(DeleteRuleRequest { + id: rule_id.clone(), + })) + .await + .expect("delete_rule succeeds"); + + let list_after_delete = client + .list_rules(authed_bearer_request(ListRulesRequest {})) + .await + .expect("list_rules succeeds") + .into_inner(); + assert!(list_after_delete.rules.is_empty()); + assert!(filter_engine + .evaluate(&sample_email("Urgent Meeting")) + .is_empty()); + } + + #[tokio::test] + async fn create_rule_rejects_invalid_nsql_and_persists_nothing() { + let (addr, _handle, _db, filter_engine) = spawn_filters_test_server("correct-token").await; + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let err = client + .create_rule(authed_bearer_request(CreateRuleRequest { + name: "Broken Rule".to_string(), + nsql: "THIS IS NOT VALID NSQL AT ALL {{{".to_string(), + priority: 0, + })) + .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!( + list_response.rules.is_empty(), + "an invalid rule must never be persisted" + ); + assert!(filter_engine + .evaluate(&nuncio_core::model::Email { + id: "msg-1".to_string(), + account_id: "acct-1".to_string(), + folder_id: "inbox".to_string(), + subject: "anything".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(), + }) + .is_empty()); + } + + #[tokio::test] + async fn delete_rule_rejects_empty_id() { + let (addr, _handle, _db, _engine) = spawn_filters_test_server("correct-token").await; + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let err = client + .delete_rule(authed_bearer_request(DeleteRuleRequest { + id: String::new(), + })) + .await + .expect_err("empty id must be rejected"); + assert_eq!(err.code(), Code::InvalidArgument); + } + + /// `ValidateRule` never fails the RPC itself -- an invalid rule is an + /// honest `valid: false` result, not a `Status` error (see this RPC's + /// doc comment). Also proves validation never persists anything. + #[tokio::test] + async fn validate_rule_reports_honest_ok_and_error_verdicts_without_persisting() { + let (addr, _handle, _db, _engine) = spawn_filters_test_server("correct-token").await; + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let valid_response = client + .validate_rule(authed_bearer_request(ValidateRuleRequest { + nsql: SAMPLE_RULE_NSQL.to_string(), + })) + .await + .expect("validate_rule call itself succeeds for a valid rule") + .into_inner(); + assert!(valid_response.valid); + assert!(valid_response.error.is_empty()); + + let invalid_response = client + .validate_rule(authed_bearer_request(ValidateRuleRequest { + nsql: "THIS IS NOT VALID NSQL AT ALL {{{".to_string(), + })) + .await + .expect("validate_rule call itself succeeds even for an invalid rule") + .into_inner(); + assert!(!invalid_response.valid); + assert!(!invalid_response.error.is_empty()); + + // Nothing was persisted by either validation call. + let list_response = client + .list_rules(authed_bearer_request(ListRulesRequest {})) + .await + .expect("list_rules succeeds") + .into_inner(); + assert!( + list_response.rules.is_empty(), + "validate_rule must never persist a rule" + ); + } + + /// `PreviewRule` dry-run evaluates an ad hoc rule against a stored + /// message (proving it reads REAL persisted mail via the same store + /// `Mail/GetMessage` reads) without ever persisting the rule itself. + #[tokio::test] + async fn preview_rule_dry_runs_against_a_stored_message_without_persisting() { + let (addr, _handle, db, _engine) = spawn_filters_test_server("correct-token").await; + + db.save_email(&nuncio_core::model::Email { + id: "msg-preview-1".to_string(), + account_id: "acct-1".to_string(), + folder_id: "inbox".to_string(), + subject: "Urgent Board Meeting".to_string(), + sender: "alice@nuncio.mx".to_string(), + recipient: "bob@nuncio.mx".to_string(), + received_at: 1_700_000_000, + read: false, + body_plain: Some("Please review the attached deck".to_string()), + body_html: None, + attachments: Vec::new(), + }) + .await + .expect("seed message"); + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let response = client + .preview_rule(authed_bearer_request(PreviewRuleRequest { + nsql: SAMPLE_RULE_NSQL.to_string(), + message_id: Some("msg-preview-1".to_string()), + })) + .await + .expect("preview_rule succeeds") + .into_inner(); + assert_eq!(response.message_id, "msg-preview-1"); + assert!(response.matched); + assert!(response + .actions_evaluated + .iter() + .any(|a| a.contains("MARK READ"))); + assert!(!response.condition_traces.is_empty()); + + // Preview evaluates a NON-matching message honestly too. + let non_matching = client + .preview_rule(authed_bearer_request(PreviewRuleRequest { + nsql: "WHERE subject CONTAINS 'Nonexistent Term' ACTION DELETE".to_string(), + message_id: Some("msg-preview-1".to_string()), + })) + .await + .expect("preview_rule succeeds") + .into_inner(); + assert!(!non_matching.matched); + + // Previewing must never persist the rule. + let list_response = client + .list_rules(authed_bearer_request(ListRulesRequest {})) + .await + .expect("list_rules succeeds") + .into_inner(); + assert!( + list_response.rules.is_empty(), + "preview_rule must never persist a rule" + ); + } + + /// When no `message_id` is given (or it does not resolve to a stored + /// message), `PreviewRule` falls back to a fixed synthetic sample + /// message rather than failing the RPC. + #[tokio::test] + async fn preview_rule_falls_back_to_a_synthetic_message_when_none_is_stored() { + let (addr, _handle, _db, _engine) = spawn_filters_test_server("correct-token").await; + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let response = client + .preview_rule(authed_bearer_request(PreviewRuleRequest { + nsql: "WHERE subject CONTAINS 'Test Subject' ACTION FLAG".to_string(), + message_id: None, + })) + .await + .expect("preview_rule succeeds with no message_id") + .into_inner(); + assert!(response.matched); + assert_eq!(response.message_id, "msg-test"); + + let response_missing_id = client + .preview_rule(authed_bearer_request(PreviewRuleRequest { + nsql: "WHERE subject CONTAINS 'Test Subject' ACTION FLAG".to_string(), + message_id: Some("does-not-exist".to_string()), + })) + .await + .expect("preview_rule succeeds with an unresolvable message_id") + .into_inner(); + assert!(response_missing_id.matched); + assert_eq!(response_missing_id.message_id, "does-not-exist"); + } + + #[tokio::test] + async fn preview_rule_and_validate_rule_reject_invalid_nsql_syntax() { + let (addr, _handle, _db, _engine) = spawn_filters_test_server("correct-token").await; + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let err = client + .preview_rule(authed_bearer_request(PreviewRuleRequest { + nsql: "THIS IS NOT VALID NSQL AT ALL {{{".to_string(), + message_id: None, + })) + .await + .expect_err("invalid NSQL must be rejected"); + assert_eq!(err.code(), Code::InvalidArgument); + } } diff --git a/crates/nunciod/src/main.rs b/crates/nunciod/src/main.rs index b077e2b..3db7493 100644 --- a/crates/nunciod/src/main.rs +++ b/crates/nunciod/src/main.rs @@ -380,16 +380,23 @@ async fn main() -> Result<(), Box> { let grpc_token = hex::encode(grpc_token_bytes); let grpc_addr = nunciod::grpc::grpc_addr_from_env(); tracing::info!( - "nunciod gRPC (nuncio.v1.System, nuncio.v1.Accounts) starting on {} (loopback only)", + "nunciod gRPC (nuncio.v1.System, nuncio.v1.Accounts, nuncio.v1.Mail, nuncio.v1.Filters) \ + starting on {} (loopback only)", grpc_addr ); let grpc_event_bus = event_bus.clone(); let grpc_db = db.clone(); + // The gRPC `Filters` service (backlog story 2.A, GH #171) shares this + // SAME `filter_engine` instance with the JSON-RPC `filter.*` handler + // above, so a rule created/deleted through EITHER transport reloads the + // one live `ArcSwap`-backed rule set both transports evaluate against. + let grpc_filter_engine = filter_engine.clone(); let _grpc_task = tokio::spawn(async move { if let Err(e) = nunciod::grpc::serve( &grpc_addr, grpc_event_bus, grpc_db, + grpc_filter_engine, grpc_secrets, grpc_token, ) diff --git a/crates/nunciod/tests/daemon_e2e_test.rs b/crates/nunciod/tests/daemon_e2e_test.rs index 8fc159b..94cbe14 100644 --- a/crates/nunciod/tests/daemon_e2e_test.rs +++ b/crates/nunciod/tests/daemon_e2e_test.rs @@ -14,6 +14,7 @@ use nuncio_cli::{AccountSubcommand, Commands, HeadlessRunner, SystemSubcommand}; use nuncio_core::ipc::{IpcClient, IpcDaemonServer}; use nuncio_core::{CoreCommand, EventBus}; +use nuncio_filter::FilterEngine; use nuncio_store::db::DatabaseEngine; use nuncio_store::vault::{SecretManager, GRPC_TOKEN_ACCOUNT}; use std::sync::Arc; @@ -157,11 +158,13 @@ async fn cli_system_status_round_trips_over_grpc_to_live_daemon() { .await .expect("connect ephemeral test db"); let server_secrets = secrets.clone(); + let server_filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); let _server_handle = tokio::spawn(async move { let _ = nunciod::grpc::serve_on_listener( listener, server_event_bus, Arc::new(server_db), + server_filter_engine, server_secrets, token, ) diff --git a/crates/nunciod/tests/grpc_auth_test.rs b/crates/nunciod/tests/grpc_auth_test.rs index 55ed7cd..23a5b2c 100644 --- a/crates/nunciod/tests/grpc_auth_test.rs +++ b/crates/nunciod/tests/grpc_auth_test.rs @@ -14,6 +14,7 @@ #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] use nuncio_core::{CoreCommand, EngineStatus, EventBus}; +use nuncio_filter::FilterEngine; use nuncio_proto::v1::system_client::SystemClient; use nuncio_proto::v1::GetStatusRequest; use nuncio_store::db::DatabaseEngine; @@ -40,8 +41,17 @@ async fn start_server(event_bus: Arc, token: String) -> std::net::Sock .await .expect("connect ephemeral test db"); let secrets = Arc::new(SecretManager::mock()); + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); tokio::spawn(async move { - let _ = serve_on_listener(listener, event_bus, Arc::new(db), secrets, token).await; + let _ = serve_on_listener( + listener, + event_bus, + Arc::new(db), + filter_engine, + secrets, + token, + ) + .await; }); addr } diff --git a/crates/nunciod/tests/spine_e2e_test.rs b/crates/nunciod/tests/spine_e2e_test.rs index f1277ac..69469ff 100644 --- a/crates/nunciod/tests/spine_e2e_test.rs +++ b/crates/nunciod/tests/spine_e2e_test.rs @@ -34,6 +34,7 @@ use nuncio_core::model::{Email, Folder}; use nuncio_core::EventBus; +use nuncio_filter::FilterEngine; use nuncio_mail::{MockMailBackend, MockMessageSender}; use nuncio_proto::v1::event::Kind; use nuncio_proto::v1::{ @@ -115,11 +116,13 @@ async fn boot_daemon( let event_bus = Arc::new(EventBus::new()); let server_secrets = secrets.clone(); let server_token = token.clone(); + let server_filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); tokio::spawn(async move { let _ = serve_on_listener_with_overrides( listener, event_bus, db, + server_filter_engine, server_secrets, server_token, overrides,