KofTwentyTwo/nuncio
2
0 Fork 0
Code Issues 73 Pull Requests 12 Actions Packages Projects Releases Wiki Activity

feat(filter): move edit/export/import/logs onto Filters gRPC service #224

Merged
KofTwentyTwo merged 1 commits from story/m4-filters-grpc-mgmt into dev 2026-07-30 00:37:20 +00:00
Conversation 0 Commits 1 Files Changed 5
+982 -128
5 changed files with 982 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/nuncio-cli/src/runner.rs
+400 -121
View File
@@ -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<String> = 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<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
.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<serde_json::Value> = 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<Mutex<Option<DeleteRuleRequest>>>,
last_update: Arc<Mutex<Option<UpdateRuleRequest>>>,
last_import: Arc<Mutex<Option<ImportRulesRequest>>>,
}
#[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<UpdateRuleRequest>,
) -> Result<tonic::Response<UpdateRuleResponse>, 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<ExportRulesRequest>,
) -> Result<tonic::Response<ExportRulesResponse>, 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<ImportRulesRequest>,
) -> Result<tonic::Response<ImportRulesResponse>, 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<GetExecutionLogsRequest>,
) -> Result<tonic::Response<GetExecutionLogsResponse>, 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]
crates/nuncio-proto/proto/nuncio/v1/descriptor.bin
BIN
View File
Binary file not shown.
crates/nuncio-proto/proto/nuncio/v1/nuncio.proto
+98
View File
@@ -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
crates/nunciod/src/grpc.rs
+451 -7
View File
@@ -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<UpdateRuleRequest>,
) -> Result<Response<UpdateRuleResponse>, 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<ExportRulesRequest>,
) -> Result<Response<ExportRulesResponse>, 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::<Vec<_>>()
.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<ImportRulesRequest>,
) -> Result<Response<ImportRulesResponse>, 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<GetExecutionLogsRequest>,
) -> Result<Response<GetExecutionLogsResponse>, 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;
crates/nunciod/tests/grpc_auth_test.rs
+33
View File
@@ -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);
}