Promote to main: M3 (JMAP) + M4 (Filters made live) #228

Merged
KofTwentyTwo merged 6 commits from dev into main 2026-07-30 01:09:32 +00:00
20 changed files with 2857 additions and 293 deletions

No files matched your search

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