diff --git a/crates/nuncio-cli/src/args.rs b/crates/nuncio-cli/src/args.rs index db359e8..abd2f38 100644 --- a/crates/nuncio-cli/src/args.rs +++ b/crates/nuncio-cli/src/args.rs @@ -517,6 +517,18 @@ pub enum FilterSubcommand { #[arg(short, long, default_value_t = 50, help = "Log limit")] limit: usize, }, + /// Retroactively apply the current rule set to every stored message, + /// streaming cumulative progress as the scan runs. + Triage { + /// Restrict the scan's reporting to a single rule ID. Not yet + /// honored by the daemon (see `nuncio.v1.Filters.Triage`) -- the + /// scan always evaluates every active rule. + #[arg(short, long, help = "Rule identifier (not yet honored)")] + rule_id: Option, + /// Messages fetched per streamed progress update. + #[arg(short, long, help = "Messages scanned per chunk")] + chunk_size: Option, + }, } #[cfg(test)] diff --git a/crates/nuncio-cli/src/runner.rs b/crates/nuncio-cli/src/runner.rs index 750169e..e5a3a53 100644 --- a/crates/nuncio-cli/src/runner.rs +++ b/crates/nuncio-cli/src/runner.rs @@ -512,6 +512,13 @@ impl HeadlessRunner { FilterSubcommand::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 + } }, Commands::Update { action } => match action { UpdateSubcommand::Check => match nuncio_core::UpdateEngine::new() { @@ -1484,6 +1491,93 @@ impl HeadlessRunner { } } + /// Drains the server-streaming `Filters.Triage` RPC, printing (or + /// accumulating, in `--json` mode) each `TriageProgress` update as it + /// arrives so the caller can watch a retroactive rescan of the whole + /// message store progress in real time rather than blocking silently + /// until it finishes. + async fn handle_filter_triage( + &self, + rule_id: Option, + chunk_size: Option, + json_mode: bool, + ) -> String { + let mut client = match self.connect_filters_client().await { + Ok(client) => client, + Err(e) => return Self::render_error(&e, json_mode), + }; + + let mut stream = match client + .triage(nuncio_proto::v1::TriageRequest { + rule_id, + chunk_size: chunk_size.unwrap_or(0), + }) + .await + { + Ok(response) => response.into_inner(), + Err(status) => { + return Self::render_error( + &format!("nunciod daemon rejected triage: {status}"), + json_mode, + ) + } + }; + + let mut updates = Vec::new(); + loop { + match stream.message().await { + Ok(Some(progress)) => { + if !json_mode { + println!( + "scanned={} matched={} actions_applied={} last_message_id={} done={}", + progress.scanned_count, + progress.matched_count, + progress.actions_applied_count, + progress.last_message_id, + progress.done + ); + } + let done = progress.done; + updates.push(progress); + if done { + break; + } + } + Ok(None) => break, + Err(status) => { + return Self::render_error( + &format!("nunciod daemon triage stream failed: {status}"), + json_mode, + ) + } + } + } + + if json_mode { + let updates_json: Vec = updates + .iter() + .map(|p| { + json!({ + "scanned_count": p.scanned_count, + "matched_count": p.matched_count, + "actions_applied_count": p.actions_applied_count, + "last_message_id": p.last_message_id, + "done": p.done, + }) + }) + .collect(); + format_json(&json!({ "updates": updates_json })) + } else { + match updates.last() { + Some(last) => format!( + "Triage complete: scanned={} matched={} actions_applied={}", + last.scanned_count, last.matched_count, last.actions_applied_count + ), + None => "Triage produced no progress updates.".to_string(), + } + } + } + /// Resolves the gRPC bearer token from the injected vault and dials the /// `nuncio.v1.Filters` service at `self.grpc_addr`, shared by every /// `filter` handler above. @@ -3235,8 +3329,8 @@ mod tests { ExportRulesRequest, ExportRulesResponse, FilterExecutionLog as FilterExecutionLogProto, FilterRule as FilterRuleProto, GetExecutionLogsRequest, GetExecutionLogsResponse, ImportRulesRequest, ImportRulesResponse, ListRulesRequest, ListRulesResponse, - PreviewRuleRequest, PreviewRuleResponse, UpdateRuleRequest, UpdateRuleResponse, - ValidateRuleRequest, ValidateRuleResponse, + PreviewRuleRequest, PreviewRuleResponse, TriageProgress, TriageRequest, + UpdateRuleRequest, UpdateRuleResponse, ValidateRuleRequest, ValidateRuleResponse, }; use std::sync::Mutex; @@ -3407,6 +3501,37 @@ mod tests { }], })) } + + type TriageStream = std::pin::Pin< + Box< + dyn tokio_stream::Stream> + + Send + + 'static, + >, + >; + + async fn triage( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> { + let updates = vec![ + Ok(TriageProgress { + scanned_count: 2, + matched_count: 1, + actions_applied_count: 1, + last_message_id: "msg-stub-2".to_string(), + done: false, + }), + Ok(TriageProgress { + scanned_count: 3, + matched_count: 2, + actions_applied_count: 2, + last_message_id: "msg-stub-3".to_string(), + done: true, + }), + ]; + Ok(tonic::Response::new(Box::pin(tokio_stream::iter(updates)))) + } } let stub = StubFilters::default(); @@ -3647,6 +3772,25 @@ mod tests { .await; assert!(logs_out.contains("msg-stub-1")); assert!(logs_out.contains(r#""action_taken":"MARK READ""#)); + + // `filter triage`: proves the CLI drains the server-streaming + // `Triage` RPC (its first real streaming-RPC consumer) and reports + // the final cumulative progress rather than only the first chunk. + let triage_out = runner + .execute_command( + &Commands::Filter { + action: FilterSubcommand::Triage { + rule_id: None, + chunk_size: Some(2), + }, + }, + true, + ) + .await; + assert!(triage_out.contains(r#""scanned_count":3"#)); + assert!(triage_out.contains(r#""matched_count":2"#)); + assert!(triage_out.contains(r#""actions_applied_count":2"#)); + assert!(triage_out.contains(r#""done":true"#)); } #[tokio::test] diff --git a/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin b/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin index 4da297f..a19a85f 100644 Binary files a/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin and b/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin differ diff --git a/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto b/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto index 7ab9792..0339047 100644 --- a/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto +++ b/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto @@ -479,6 +479,13 @@ service Filters { // 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 @@ -675,6 +682,36 @@ message FilterExecutionLog { string hash = 7; } +// TriageRequest configures a retroactive rescan of the whole message store. +message TriageRequest { + // Scope the scan to a single rule's actions. Currently NOT honored -- + // `Triage` always evaluates a message against the full live rule set, + // matching what a real (non-retroactive) sync run would do. Reserved for + // forward compatibility; a client that sets this today gets a whole-store + // scan against all active rules, not an error. + optional string rule_id = 1; + // Number of messages fetched per keyset page. 0 selects the server + // default (100). + uint32 chunk_size = 2; +} + +// TriageProgress reports the real, cumulative state of an in-progress (or +// just-finished) `Triage` scan. Every field is the true running total as of +// the chunk just processed -- never an estimate or a fabricated count. +message TriageProgress { + // Total messages scanned so far. + uint64 scanned_count = 1; + // Total messages that had at least one action applied or enqueued. + uint64 matched_count = 2; + // Total individual actions applied or enqueued across all messages. + uint64 actions_applied_count = 3; + // ID of the last message scanned in the most recently processed chunk. + string last_message_id = 4; + // True only on the final message of the stream, once the whole store has + // been scanned. + bool done = 5; +} + // Export exposes the daemon's real mailbox export engine // (`nuncio_core::export`, via `DatabaseEngine::export_messages_to_file`) // over authenticated gRPC. The daemon and its caller run on the SAME host diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs index 2b7bf5f..bab507a 100644 --- a/crates/nunciod/src/grpc.rs +++ b/crates/nunciod/src/grpc.rs @@ -40,8 +40,9 @@ use nuncio_proto::v1::{ Message as MessageProto, MessageFlagsChanged, MessageSearchHit, PreviewRuleRequest, PreviewRuleResponse, SearchMessagesRequest, SearchMessagesResponse, SendMessageRequest, SendMessageResponse, ShuttingDown, SubscribeRequest, SyncCompleted, SyncRequest, SyncResponse, - SyncStarted, TlsMode as TlsModeProto, UpdateAvailable, UpdateRuleRequest, UpdateRuleResponse, - 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; @@ -53,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}; @@ -1472,6 +1473,102 @@ impl Filters for FiltersGrpcService { Ok(Response::new(GetExecutionLogsResponse { logs })) } + + /// Server-streaming retroactive rescan of the whole message store. + /// Unlike `System::subscribe` (which wraps an already-live broadcast), + /// this RPC actively drives the scan: a spawned worker task walks + /// `db.get_message_chunk` in keyset pages, applies the shared + /// [`crate::sync::apply_filter_actions`] evaluate-and-route logic to + /// each message (the SAME side effects live sync produces), and pushes + /// cumulative progress onto an mpsc channel that this method wraps as + /// the returned stream. + /// + /// `rule_id` is accepted for forward compatibility but NOT currently + /// honored: `apply_filter_actions` always evaluates against the whole + /// live rule set, matching what a real sync pass would do, rather than + /// silently narrowing to one rule. + /// + /// A `get_message_chunk` failure sends a single `Err(Status::internal)` + /// on the stream and stops the scan -- never a fabricated `done: true` + /// on error. The receiver disconnecting (the caller dropped the call) + /// also stops the worker rather than continuing to scan for no one. + type TriageStream = + Pin> + Send + 'static>>; + + async fn triage( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let chunk_size = if req.chunk_size == 0 { + 100 + } else { + req.chunk_size as usize + }; + + let (tx, rx) = tokio::sync::mpsc::channel::>(16); + let db = self.db.clone(); + let engine = self.filter_engine.clone(); + + tokio::spawn(async move { + let mut last_id = String::new(); + let mut scanned: u64 = 0; + let mut matched: u64 = 0; + let mut applied: u64 = 0; + + loop { + let batch = match db.get_message_chunk(&last_id, chunk_size).await { + Ok(batch) => batch, + Err(e) => { + let _ = tx + .send(Err(Status::internal(format!( + "failed to read message chunk during triage: {e}" + )))) + .await; + return; + } + }; + let is_last_page = batch.len() < chunk_size; + + for email in &batch { + let actions_for_email = + crate::sync::apply_filter_actions(&db, &engine, email).await; + scanned += 1; + applied += actions_for_email as u64; + if actions_for_email > 0 { + matched += 1; + } + } + + if let Some(last) = batch.last() { + last_id = last.id.clone(); + } + + let done = is_last_page; + if tx + .send(Ok(TriageProgress { + scanned_count: scanned, + matched_count: matched, + actions_applied_count: applied, + last_message_id: last_id.clone(), + done, + })) + .await + .is_err() + { + // Receiver dropped: the caller disconnected, so there is + // no one left to report progress to. + return; + } + + if is_last_page { + return; + } + } + }); + + Ok(Response::new(Box::pin(ReceiverStream::new(rx)))) + } } /// Maps a wire-format `nuncio.v1.ExportFormat` enum value back onto diff --git a/crates/nunciod/tests/filters_triage_e2e_test.rs b/crates/nunciod/tests/filters_triage_e2e_test.rs new file mode 100644 index 0000000..bb34869 --- /dev/null +++ b/crates/nunciod/tests/filters_triage_e2e_test.rs @@ -0,0 +1,267 @@ +//! Offline end-to-end test for the `nuncio.v1.Filters.Triage` streaming RPC. +//! +//! Boots a real `nunciod` daemon gRPC server (`nunciod::grpc:: +//! serve_on_listener`) on an ephemeral loopback port, backed by a real +//! temp file-backed [`DatabaseEngine`] seeded directly via +//! [`DatabaseEngine::save_email`] (standing in for messages that were synced +//! before the active rule set existed) and a live [`FilterEngine`] carrying +//! a real persisted rule. +//! +//! Drives ONLY the authenticated `nuncio-proto` `FiltersClient::triage` call +//! against the live server, proving: +//! a. the final cumulative `matched_count`/`actions_applied_count` are +//! correct against a store containing both matching and non-matching +//! messages; +//! b. matching messages show the REAL side effects `apply_filter_actions` +//! produces (the read flag flips for an immediate `MARK READ` action; +//! an outbox row is queued for a remote `MOVE TO` action) -- not a +//! fabricated count with no backing state change; +//! c. a store bigger than one `chunk_size` genuinely streams more than one +//! non-final `TriageProgress` message, proving this is a real +//! multi-chunk stream rather than one batch dressed up as a stream. +//! +//! Fully offline and deterministic: no real network, no real OS keyring +//! (`SecretManager::mock`), no fixed sleeps. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +use nuncio_core::model::Email; +use nuncio_core::EventBus; +use nuncio_filter::{FilterEngine, NsqlParser}; +use nuncio_proto::v1::filters_client::FiltersClient; +use nuncio_proto::v1::TriageRequest; +use nuncio_store::db::DatabaseEngine; +use nuncio_store::vault::{SecretManager, GRPC_TOKEN_ACCOUNT}; +use nunciod::grpc::serve_on_listener; +use std::sync::Arc; +use tokio::net::TcpListener; + +const ACCOUNT_ID: &str = "acct-triage-e2e-1"; + +fn sample_email(id: &str, subject: &str) -> Email { + Email { + id: id.to_string(), + account_id: ACCOUNT_ID.to_string(), + folder_id: "inbox".to_string(), + subject: subject.to_string(), + sender: "alice@nuncio.mx".to_string(), + recipient: "triage-e2e@nuncio.mx".to_string(), + received_at: 1_700_000_000, + read: false, + body_plain: Some("body".to_string()), + body_html: None, + attachments: Vec::new(), + } +} + +/// Boots the real daemon gRPC server (bearer-token authenticated) on an +/// ephemeral loopback port, backed by a real temp file-backed +/// `DatabaseEngine` and the given live `FilterEngine`. Returns the server +/// address, the minted token, the `DatabaseEngine` handle (used to seed +/// messages before triggering `Triage`), and the `TempDir` guard, which the +/// caller must hold for as long as it talks to the server. +async fn boot_daemon( + secrets: Arc, + filter_engine: Arc, +) -> ( + std::net::SocketAddr, + String, + Arc, + tempfile::TempDir, +) { + let dir = tempfile::tempdir().expect("create temp dir for file-backed db"); + let db_path = dir.path().join("triage_e2e.db"); + let db = Arc::new( + DatabaseEngine::connect_file(&db_path, &secrets) + .await + .expect("open temp file-backed database"), + ); + + let token_bytes = secrets + .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32) + .expect("mint gRPC bearer token from mock vault"); + let token = hex::encode(token_bytes); + + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral loopback port"); + let addr = listener.local_addr().expect("listener has local addr"); + + let event_bus = Arc::new(EventBus::new()); + let server_db = db.clone(); + let server_secrets = secrets.clone(); + let server_token = token.clone(); + tokio::spawn(async move { + let _ = serve_on_listener( + listener, + event_bus, + server_db, + filter_engine, + server_secrets, + server_token, + ) + .await; + }); + + (addr, token, db, dir) +} + +fn authed_request(token: &str, message: T) -> tonic::Request { + let mut request = tonic::Request::new(message); + request.metadata_mut().insert( + "authorization", + format!("Bearer {token}") + .parse() + .expect("valid ascii metadata value"), + ); + request +} + +#[tokio::test] +async fn triage_applies_live_rule_retroactively_and_reports_real_cumulative_counts() { + let secrets = Arc::new(SecretManager::mock()); + + let mark_read_rule = NsqlParser::parse_rule( + "Urgent Auto-Read", + 1, + "WHERE subject CONTAINS 'Urgent' ACTION MARK READ", + ) + .expect("parse mark-read rule"); + let move_rule = NsqlParser::parse_rule( + "Archive Rule", + 2, + "WHERE subject CONTAINS 'Archive Me' ACTION MOVE TO 'Archive'", + ) + .expect("parse move rule"); + let filter_engine = + Arc::new(FilterEngine::new(vec![mark_read_rule, move_rule]).expect("compile rules")); + + let (addr, token, db, _dir) = boot_daemon(secrets, filter_engine).await; + + // Seed messages directly into the store, standing in for mail that was + // synced BEFORE these rules existed -- exactly the scenario `Triage` is + // for. Two match (one per rule), one does not. + db.save_email(&sample_email("m-urgent", "Urgent: server down")) + .await + .expect("seed urgent message"); + db.save_email(&sample_email("m-archive", "Please Archive Me")) + .await + .expect("seed archive message"); + db.save_email(&sample_email("m-plain", "Just saying hi")) + .await + .expect("seed non-matching message"); + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let mut stream = client + .triage(authed_request( + &token, + TriageRequest { + rule_id: None, + chunk_size: 100, + }, + )) + .await + .expect("triage call succeeds") + .into_inner(); + + let mut updates = Vec::new(); + while let Some(progress) = stream.message().await.expect("stream message") { + updates.push(progress); + } + + let last = updates.last().expect("at least one progress update"); + assert!(last.done, "final update must be marked done"); + assert_eq!(last.scanned_count, 3); + assert_eq!( + last.matched_count, 2, + "exactly the two messages with a rule match count as matched" + ); + assert_eq!( + last.actions_applied_count, 2, + "one action applied per matching message" + ); + + // Real side effects, not a fabricated count: the MARK READ action + // actually flipped the stored read flag... + let urgent = db.get_message("m-urgent").await.expect("get urgent"); + assert!(urgent.read, "MARK READ must genuinely flip the read flag"); + + // ...and the MOVE TO action actually queued a real outbox mutation. + let pending = db + .list_pending_mutations(10) + .await + .expect("list pending mutations"); + assert_eq!( + pending.len(), + 1, + "the remote MOVE action must enqueue exactly one outbox mutation" + ); + assert_eq!(pending[0].message_id, "m-archive"); + + let plain = db.get_message("m-plain").await.expect("get plain"); + assert!(!plain.read, "a non-matching message must be left untouched"); +} + +#[tokio::test] +async fn triage_streams_multiple_chunks_for_a_store_bigger_than_one_chunk() { + let secrets = Arc::new(SecretManager::mock()); + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); + + let (addr, token, db, _dir) = boot_daemon(secrets, filter_engine).await; + + // Five messages against a chunk size of two forces three chunks + // (2 + 2 + 1), so the stream must carry at least two non-final progress + // updates before the final `done: true` -- proof this is a real + // multi-chunk stream, not one batch disguised as one. + for i in 0..5 { + db.save_email(&sample_email( + &format!("m-{i:02}"), + &format!("Message number {i}"), + )) + .await + .expect("seed message"); + } + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let mut stream = client + .triage(authed_request( + &token, + TriageRequest { + rule_id: None, + chunk_size: 2, + }, + )) + .await + .expect("triage call succeeds") + .into_inner(); + + let mut updates = Vec::new(); + while let Some(progress) = stream.message().await.expect("stream message") { + updates.push(progress); + } + + let non_final: Vec<_> = updates.iter().filter(|p| !p.done).collect(); + assert!( + non_final.len() > 1, + "expected more than one non-final progress update for a multi-chunk scan, got {}", + non_final.len() + ); + + let last = updates.last().expect("at least one progress update"); + assert!(last.done); + assert_eq!(last.scanned_count, 5); + + // Cumulative counts must monotonically increase across the stream -- + // never reset or double-count between chunks. + let mut prev_scanned = 0u64; + for update in &updates { + assert!(update.scanned_count >= prev_scanned); + prev_scanned = update.scanned_count; + } +} diff --git a/crates/nunciod/tests/grpc_auth_test.rs b/crates/nunciod/tests/grpc_auth_test.rs index 166f913..767e051 100644 --- a/crates/nunciod/tests/grpc_auth_test.rs +++ b/crates/nunciod/tests/grpc_auth_test.rs @@ -187,3 +187,33 @@ async fn rejects_calls_to_new_filters_rpcs_without_a_bearer_token() { .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); +}