diff --git a/crates/nuncio-cli/src/runner.rs b/crates/nuncio-cli/src/runner.rs index 6b10426..750169e 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,29 @@ 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) - .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 - } + self.handle_filter_logs(*limit, json_mode).await } }, Commands::Update { action } => match action { @@ -1395,6 +1306,184 @@ 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, + ), + } + } + /// 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 +3232,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, 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 +3338,81 @@ 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(), + }], + })) + } } 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 +3533,120 @@ 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""#)); } #[tokio::test] diff --git a/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin b/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin index 0494ff3..4da297f 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..7ab9792 100644 --- a/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto +++ b/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto @@ -458,6 +458,27 @@ 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); } // FilterRule mirrors `nuncio_filter::FilterRule`'s persisted/user-facing @@ -577,6 +598,83 @@ 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; +} + // 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/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs index 26605cd..2b7bf5f 100644 --- a/crates/nunciod/src/grpc.rs +++ b/crates/nunciod/src/grpc.rs @@ -27,18 +27,21 @@ 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, UpdateAvailable, UpdateRuleRequest, UpdateRuleResponse, + ValidateRuleRequest, ValidateRuleResponse, VerifyChainRequest, VerifyChainResponse, }; use nuncio_store::db::DatabaseEngine; use nuncio_store::search::SearchEngine; @@ -1108,6 +1111,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 { @@ -1315,6 +1334,144 @@ 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 })) + } } /// Maps a wire-format `nuncio.v1.ExportFormat` enum value back onto @@ -3148,7 +3305,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, }; @@ -3243,6 +3401,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 @@ -3554,6 +3745,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/tests/grpc_auth_test.rs b/crates/nunciod/tests/grpc_auth_test.rs index aaafaa7..166f913 100644 --- a/crates/nunciod/tests/grpc_auth_test.rs +++ b/crates/nunciod/tests/grpc_auth_test.rs @@ -154,3 +154,36 @@ 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); +}