diff --git a/crates/nuncio-cli/src/args.rs b/crates/nuncio-cli/src/args.rs index 554f691..4994093 100644 --- a/crates/nuncio-cli/src/args.rs +++ b/crates/nuncio-cli/src/args.rs @@ -449,14 +449,13 @@ pub enum SystemSubcommand { #[derive(Subcommand, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "kebab-case")] pub enum AuditSubcommand { - /// List persisted WORM audit ledger records, sequence ascending. + /// List persisted WORM audit ledger records, sequence ascending. Follows + /// keyset pagination to the end, fetching `page_size` records per request. List { - /// Max records to fetch (0 = server default). - #[arg(short, long, default_value_t = 50, help = "Max records to fetch")] - limit: u32, - /// Pagination offset. - #[arg(short, long, default_value_t = 0, help = "Pagination offset")] - offset: u32, + /// Records fetched per page (0 = server default). All pages are + /// followed via the keyset cursor and printed together. + #[arg(short, long, default_value_t = 50, help = "Records fetched per page")] + page_size: u32, }, /// Verify the WORM audit ledger's cryptographic HMAC hash-chain /// integrity. @@ -716,25 +715,18 @@ mod tests { cli_list.command, Commands::System { action: SystemSubcommand::Audit { - action: AuditSubcommand::List { - limit: 50, - offset: 0, - } + action: AuditSubcommand::List { page_size: 50 } } } ); - let cli_list_paged = Cli::parse_from([ - "nuncio", "system", "audit", "list", "--limit", "10", "--offset", "5", - ]); + let cli_list_paged = + Cli::parse_from(["nuncio", "system", "audit", "list", "--page-size", "10"]); assert_eq!( cli_list_paged.command, Commands::System { action: SystemSubcommand::Audit { - action: AuditSubcommand::List { - limit: 10, - offset: 5, - } + action: AuditSubcommand::List { page_size: 10 } } } ); diff --git a/crates/nuncio-cli/src/runner.rs b/crates/nuncio-cli/src/runner.rs index 9bf5dbe..33b51e1 100644 --- a/crates/nuncio-cli/src/runner.rs +++ b/crates/nuncio-cli/src/runner.rs @@ -498,8 +498,8 @@ impl HeadlessRunner { Commands::System { action } => match action { SystemSubcommand::Status => self.handle_system_status(json_mode).await, SystemSubcommand::Audit { action } => match action { - AuditSubcommand::List { limit, offset } => { - self.handle_audit_list(*limit, *offset, json_mode).await + AuditSubcommand::List { page_size } => { + self.handle_audit_list(*page_size, json_mode).await } AuditSubcommand::Verify => self.handle_audit_verify(json_mode).await, }, @@ -726,33 +726,48 @@ impl HeadlessRunner { Err(e) => return Self::render_error(&e, json_mode), }; - match client - .list_messages(nuncio_proto::v1::ListMessagesRequest { - folder_id: folder.to_string(), - limit: 0, - }) - .await - { - Ok(response) => { - let messages = response.into_inner().messages; - if json_mode { - let messages_json: Vec = - messages.iter().map(message_proto_to_json).collect(); - format_json(&json!({ - "folder": folder, - "messages": messages_json - })) - } else { - format!("Folder '{}': {} message(s) found", folder, messages.len()) + // Follow `next_page_token` to the end so the command shows the whole + // folder, transparently exercising the keyset pagination convention. + let mut messages = Vec::new(); + let mut page_token = String::new(); + loop { + match client + .list_messages(nuncio_proto::v1::ListMessagesRequest { + folder_id: folder.to_string(), + page_size: 0, + page_token: page_token.clone(), + }) + .await + { + Ok(response) => { + let page = response.into_inner(); + messages.extend(page.messages); + if page.next_page_token.is_empty() { + break; + } + page_token = page.next_page_token; + } + Err(status) => { + return Self::render_error( + &format!( + "nunciod daemon rejected list_messages: {}", + Self::clean_status_message(&status) + ), + json_mode, + ); } } - Err(status) => Self::render_error( - &format!( - "nunciod daemon rejected list_messages: {}", - Self::clean_status_message(&status) - ), - json_mode, - ), + } + + if json_mode { + let messages_json: Vec = + messages.iter().map(message_proto_to_json).collect(); + format_json(&json!({ + "folder": folder, + "messages": messages_json + })) + } else { + format!("Folder '{}': {} message(s) found", folder, messages.len()) } } @@ -824,40 +839,55 @@ impl HeadlessRunner { Err(e) => return Self::render_error(&e, json_mode), }; - match client - .search_messages(nuncio_proto::v1::SearchMessagesRequest { - query: query.to_string(), - }) - .await - { - Ok(response) => { - let hits = response.into_inner().hits; - if json_mode { - let hits_json: Vec = hits - .iter() - .map(|h| { - json!({ - "id": h.id, - "title": h.title, - "snippet": h.snippet, - }) - }) - .collect(); - format_json(&json!({ - "query": query, - "results": hits_json - })) - } else { - format!("Search complete for '{}' ({} matches)", query, hits.len()) + // Follow `next_page_token` to the end so search returns every hit. + let mut hits = Vec::new(); + let mut page_token = String::new(); + loop { + match client + .search_messages(nuncio_proto::v1::SearchMessagesRequest { + query: query.to_string(), + page_size: 0, + page_token: page_token.clone(), + }) + .await + { + Ok(response) => { + let page = response.into_inner(); + hits.extend(page.hits); + if page.next_page_token.is_empty() { + break; + } + page_token = page.next_page_token; + } + Err(status) => { + return Self::render_error( + &format!( + "nunciod daemon rejected search_messages: {}", + Self::clean_status_message(&status) + ), + json_mode, + ); } } - Err(status) => Self::render_error( - &format!( - "nunciod daemon rejected search_messages: {}", - Self::clean_status_message(&status) - ), - json_mode, - ), + } + + if json_mode { + let hits_json: Vec = hits + .iter() + .map(|h| { + json!({ + "id": h.id, + "title": h.title, + "snippet": h.snippet, + }) + }) + .collect(); + format_json(&json!({ + "query": query, + "results": hits_json + })) + } else { + format!("Search complete for '{}' ({} matches)", query, hits.len()) } } @@ -869,36 +899,52 @@ impl HeadlessRunner { Err(e) => return Self::render_error(&e, json_mode), }; - match client - .list_folders(nuncio_proto::v1::ListFoldersRequest {}) - .await - { - Ok(response) => { - let folders = response.into_inner().folders; - if json_mode { - let folders_json: Vec = folders - .iter() - .map(|f| { - json!({ - "id": f.id, - "name": f.name, - "total_messages": f.total_messages, - "unread_messages": f.unread_messages, - }) - }) - .collect(); - format_json(&json!({ "folders": folders_json })) - } else { - format!("Available Mailbox Folders: {} folders found", folders.len()) + // Follow `next_page_token` to the end so every folder is listed. + let mut folders = Vec::new(); + let mut page_token = String::new(); + loop { + match client + .list_folders(nuncio_proto::v1::ListFoldersRequest { + page_size: 0, + page_token: page_token.clone(), + }) + .await + { + Ok(response) => { + let page = response.into_inner(); + folders.extend(page.folders); + if page.next_page_token.is_empty() { + break; + } + page_token = page.next_page_token; + } + Err(status) => { + return Self::render_error( + &format!( + "nunciod daemon rejected list_folders: {}", + Self::clean_status_message(&status) + ), + json_mode, + ); } } - Err(status) => Self::render_error( - &format!( - "nunciod daemon rejected list_folders: {}", - Self::clean_status_message(&status) - ), - json_mode, - ), + } + + if json_mode { + let folders_json: Vec = folders + .iter() + .map(|f| { + json!({ + "id": f.id, + "name": f.name, + "total_messages": f.total_messages, + "unread_messages": f.unread_messages, + }) + }) + .collect(); + format_json(&json!({ "folders": folders_json })) + } else { + format!("Available Mailbox Folders: {} folders found", folders.len()) } } @@ -1093,46 +1139,63 @@ impl HeadlessRunner { /// `system audit list`: a real thin gRPC client of the running /// `nunciod` daemon's `nuncio.v1.Audit` API. Lists a page of the /// daemon's real, persisted WORM audit ledger, sequence ascending. - async fn handle_audit_list(&self, limit: u32, offset: u32, json_mode: bool) -> String { + async fn handle_audit_list(&self, page_size: u32, json_mode: bool) -> String { let mut client = match self.connect_audit_client().await { Ok(client) => client, Err(e) => return Self::render_error(&e, json_mode), }; - match client - .list_records(nuncio_proto::v1::ListRecordsRequest { limit, offset }) - .await - { - Ok(response) => { - let records = response.into_inner().records; - if json_mode { - let records_json: Vec = - records.iter().map(audit_record_proto_to_json).collect(); - format_json(&json!({ "records": records_json })) - } else if records.is_empty() { - "No audit records recorded.".to_string() - } else { - let mut out = - String::from("SEQ ACTOR ACTION TIMESTAMP\n"); - for r in records { - out.push_str(&format!( - "{:<4} {:<20} {:<20} {}\n", - r.sequence, - r.actor, - r.action, - format_timestamp(&r.timestamp) - )); + // Follow `next_page_token` to the end so the whole ledger is listed, + // fetching `page_size` records per keyset page. + let mut records = Vec::new(); + let mut page_token = String::new(); + loop { + match client + .list_records(nuncio_proto::v1::ListRecordsRequest { + page_size, + page_token: page_token.clone(), + }) + .await + { + Ok(response) => { + let page = response.into_inner(); + records.extend(page.records); + if page.next_page_token.is_empty() { + break; } - out + page_token = page.next_page_token; + } + Err(status) => { + return Self::render_error( + &format!( + "nunciod daemon rejected list_records: {}", + Self::clean_status_message(&status) + ), + json_mode, + ); } } - Err(status) => Self::render_error( - &format!( - "nunciod daemon rejected list_records: {}", - Self::clean_status_message(&status) - ), - json_mode, - ), + } + + if json_mode { + let records_json: Vec = + records.iter().map(audit_record_proto_to_json).collect(); + format_json(&json!({ "records": records_json })) + } else if records.is_empty() { + "No audit records recorded.".to_string() + } else { + let mut out = + String::from("SEQ ACTOR ACTION TIMESTAMP\n"); + for r in records { + out.push_str(&format!( + "{:<4} {:<20} {:<20} {}\n", + r.sequence, + r.actor, + r.action, + format_timestamp(&r.timestamp) + )); + } + out } } @@ -1208,37 +1271,52 @@ impl HeadlessRunner { Err(e) => return Self::render_error(&e, json_mode), }; - match client - .list_rules(nuncio_proto::v1::ListRulesRequest {}) - .await - { - Ok(response) => { - let rules = response.into_inner().rules; - if json_mode { - let rules_json: Vec = - rules.iter().map(filter_rule_proto_to_json).collect(); - format_json(&json!({ "rules": rules_json })) - } else if rules.is_empty() { - "No filter rules configured.".to_string() - } else { - let mut out = - String::from("ID PRIORITY ENABLED NAME NSQL\n"); - for r in rules { - out.push_str(&format!( - "{:<10} {:<8} {:<7} {:<20} {}\n", - r.id, r.priority, r.enabled, r.name, r.nsql_text - )); + // Follow `next_page_token` to the end so every rule is listed. + let mut rules = Vec::new(); + let mut page_token = String::new(); + loop { + match client + .list_rules(nuncio_proto::v1::ListRulesRequest { + page_size: 0, + page_token: page_token.clone(), + }) + .await + { + Ok(response) => { + let page = response.into_inner(); + rules.extend(page.rules); + if page.next_page_token.is_empty() { + break; } - out + page_token = page.next_page_token; + } + Err(status) => { + return Self::render_error( + &format!( + "nunciod daemon rejected list_rules: {}", + Self::clean_status_message(&status) + ), + json_mode, + ); } } - Err(status) => Self::render_error( - &format!( - "nunciod daemon rejected list_rules: {}", - Self::clean_status_message(&status) - ), - json_mode, - ), + } + + if json_mode { + let rules_json: Vec = + rules.iter().map(filter_rule_proto_to_json).collect(); + format_json(&json!({ "rules": rules_json })) + } else if rules.is_empty() { + "No filter rules configured.".to_string() + } else { + let mut out = String::from("ID PRIORITY ENABLED NAME NSQL\n"); + for r in rules { + out.push_str(&format!( + "{:<10} {:<8} {:<7} {:<20} {}\n", + r.id, r.priority, r.enabled, r.name, r.nsql_text + )); + } + out } } @@ -2240,36 +2318,52 @@ impl HeadlessRunner { Err(e) => return Self::render_error(&e, json_mode), }; - match client - .list_events(nuncio_proto::v1::ListEventsRequest { - account_id: account.to_string(), - calendar_id: calendar.to_string(), - start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(start)), - end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(end)), - }) - .await - { - Ok(response) => { - let events = response.into_inner().events; - if json_mode { - let events_json: Vec = - events.iter().map(calendar_event_proto_to_json).collect(); - format_json(&json!({ - "account": account, - "calendar": calendar, - "events": events_json - })) - } else { - format!("Calendar Events: {} event(s) found", events.len()) + // Follow `next_page_token` to the end so every event in the window is + // listed, transparently exercising keyset pagination. + let mut events = Vec::new(); + let mut page_token = String::new(); + loop { + match client + .list_events(nuncio_proto::v1::ListEventsRequest { + account_id: account.to_string(), + calendar_id: calendar.to_string(), + start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(start)), + end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(end)), + page_size: 0, + page_token: page_token.clone(), + }) + .await + { + Ok(response) => { + let page = response.into_inner(); + events.extend(page.events); + if page.next_page_token.is_empty() { + break; + } + page_token = page.next_page_token; + } + Err(status) => { + return Self::render_error( + &format!( + "nunciod daemon rejected list_events: {}", + Self::clean_status_message(&status) + ), + json_mode, + ); } } - Err(status) => Self::render_error( - &format!( - "nunciod daemon rejected list_events: {}", - Self::clean_status_message(&status) - ), - json_mode, - ), + } + + if json_mode { + let events_json: Vec = + events.iter().map(calendar_event_proto_to_json).collect(); + format_json(&json!({ + "account": account, + "calendar": calendar, + "events": events_json + })) + } else { + format!("Calendar Events: {} event(s) found", events.len()) } } @@ -2348,35 +2442,59 @@ impl HeadlessRunner { Err(e) => return Self::render_error(&e, json_mode), }; - match client - .list_contacts(nuncio_proto::v1::ListContactsRequest { - account_id: account.to_string(), - }) - .await - { - Ok(response) => { - let contacts = response.into_inner().contacts; - if json_mode { - let contacts_json: Vec = - contacts.iter().map(contact_proto_to_json).collect(); - format_json(&json!({ "account": account, "contacts": contacts_json })) - } else { - format!( - "Contacts: {} contact(s) found in address book.", - contacts.len() - ) - } + let contacts = match self.fetch_all_contacts(&mut client, account).await { + Ok(contacts) => contacts, + Err(status) => { + return Self::render_error( + &format!( + "nunciod daemon rejected list_contacts: {}", + Self::clean_status_message(&status) + ), + json_mode, + ); } - Err(status) => Self::render_error( - &format!( - "nunciod daemon rejected list_contacts: {}", - Self::clean_status_message(&status) - ), - json_mode, - ), + }; + + if json_mode { + let contacts_json: Vec = + contacts.iter().map(contact_proto_to_json).collect(); + format_json(&json!({ "account": account, "contacts": contacts_json })) + } else { + format!( + "Contacts: {} contact(s) found in address book.", + contacts.len() + ) } } + /// Follows `next_page_token` to fetch every contact in `account`'s address + /// book over the keyset-paginated `ListContacts` RPC, shared by + /// [`Self::handle_contact_list`] and [`Self::handle_contact_search`]. + async fn fetch_all_contacts( + &self, + client: &mut nuncio_proto::client::AuthenticatedContactsClient, + account: &str, + ) -> Result, tonic::Status> { + let mut contacts = Vec::new(); + let mut page_token = String::new(); + loop { + let page = client + .list_contacts(nuncio_proto::v1::ListContactsRequest { + account_id: account.to_string(), + page_size: 0, + page_token: page_token.clone(), + }) + .await? + .into_inner(); + contacts.extend(page.contacts); + if page.next_page_token.is_empty() { + break; + } + page_token = page.next_page_token; + } + Ok(contacts) + } + /// `contact search`: fetches `account`'s full `ListContacts` result over /// the real gRPC API, then filters it client-side by `query` against /// display name, organization, and email address (case-insensitive) -- @@ -2387,42 +2505,37 @@ impl HeadlessRunner { Err(e) => return Self::render_error(&e, json_mode), }; - match client - .list_contacts(nuncio_proto::v1::ListContactsRequest { - account_id: account.to_string(), - }) - .await - { - Ok(response) => { - let matches: Vec<_> = response - .into_inner() - .contacts - .into_iter() - .filter(|c| contact_matches_query(c, query)) - .collect(); - if json_mode { - let contacts_json: Vec = - matches.iter().map(contact_proto_to_json).collect(); - format_json(&json!({ - "account": account, - "query": query, - "contacts": contacts_json - })) - } else { - format!( - "Search ('{}'): {} contact(s) matched.", - query, - matches.len() - ) - } + let contacts = match self.fetch_all_contacts(&mut client, account).await { + Ok(contacts) => contacts, + Err(status) => { + return Self::render_error( + &format!( + "nunciod daemon rejected list_contacts: {}", + Self::clean_status_message(&status) + ), + json_mode, + ); } - Err(status) => Self::render_error( - &format!( - "nunciod daemon rejected list_contacts: {}", - Self::clean_status_message(&status) - ), - json_mode, - ), + }; + + let matches: Vec<_> = contacts + .into_iter() + .filter(|c| contact_matches_query(c, query)) + .collect(); + if json_mode { + let contacts_json: Vec = + matches.iter().map(contact_proto_to_json).collect(); + format_json(&json!({ + "account": account, + "query": query, + "contacts": contacts_json + })) + } else { + format!( + "Search ('{}'): {} contact(s) matched.", + query, + matches.len() + ) } } @@ -3231,6 +3344,7 @@ mod tests { total_messages: 3, unread_messages: 1, }], + next_page_token: String::new(), })) } @@ -3243,6 +3357,7 @@ mod tests { message.folder_id = req.folder_id; Ok(tonic::Response::new(ListMessagesResponse { messages: vec![message], + next_page_token: String::new(), })) } @@ -3286,6 +3401,7 @@ mod tests { title: "Stub Subject".to_string(), snippet: format!("...{}...", req.query), }], + next_page_token: String::new(), })) } @@ -3693,6 +3809,7 @@ mod tests { .unwrap_or_else(|e| e.into_inner()) = Some(req); Ok(tonic::Response::new(ListEventsResponse { events: vec![stub_event()], + next_page_token: String::new(), })) } @@ -3875,6 +3992,7 @@ mod tests { stub_contact("ct-stub-1", "Alice Stub", "Kof22", "alice@nuncio.mx"), stub_contact("ct-stub-2", "Bob Other", "OtherCo", "bob@nuncio.mx"), ], + next_page_token: String::new(), })) } @@ -4114,6 +4232,7 @@ mod tests { 1_700_000_000, )), }], + next_page_token: String::new(), })) } @@ -4723,6 +4842,7 @@ mod tests { previous_block_hash: "GENESIS".to_string(), record_hmac: "cafebabe".to_string(), }], + next_page_token: String::new(), })) } @@ -4758,10 +4878,7 @@ mod tests { .execute_command( &Commands::System { action: SystemSubcommand::Audit { - action: AuditSubcommand::List { - limit: 0, - offset: 0, - }, + action: AuditSubcommand::List { page_size: 0 }, }, }, true, @@ -4824,10 +4941,7 @@ mod tests { .execute_command( &Commands::System { action: SystemSubcommand::Audit { - action: AuditSubcommand::List { - limit: 0, - offset: 0, - }, + action: AuditSubcommand::List { page_size: 0 }, }, }, true, diff --git a/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin b/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin index 94ba2c2..828d294 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 368af82..2024e64 100644 --- a/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto +++ b/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto @@ -375,11 +375,11 @@ message TestAccountConnectionResponse { // carved out for "just one more" unauthenticated service. See // `nunciod::grpc::serve_on_listener`. service Mail { - // ListFolders returns every mailbox folder known to the store, with - // total/unread message counts. + // ListFolders returns a keyset-paginated page of mailbox folders known to + // the store, folder id ascending, each with total/unread message counts. rpc ListFolders(ListFoldersRequest) returns (ListFoldersResponse); - // ListMessages returns messages in a single folder, newest first, capped - // at `limit`. + // ListMessages returns a keyset-paginated page of messages in a single + // folder, newest first. rpc ListMessages(ListMessagesRequest) returns (ListMessagesResponse); // GetMessage returns a single message by ID, including its (decrypted) // body. @@ -387,7 +387,8 @@ service Mail { // MarkRead updates a message's read/unread flag. rpc MarkRead(MarkReadRequest) returns (MarkReadResponse); // SearchMessages runs a full-text (FTS5) search over cached message - // subjects, senders, and bodies. + // subjects, senders, and bodies, returning a keyset-paginated page of hits + // ordered by relevance rank. rpc SearchMessages(SearchMessagesRequest) returns (SearchMessagesResponse); // SendMessage composes and sends a real outbound email over SMTP, built // from the configured account's SMTP endpoint @@ -462,26 +463,53 @@ message Folder { uint64 unread_messages = 4; } -// ListFoldersRequest currently carries no parameters. -message ListFoldersRequest {} +// ListFoldersRequest pages the folder listing (folder id ascending) using +// the shared keyset convention: `page_size` bounds a page and `page_token` +// is the opaque cursor returned as the previous page's `next_page_token`. +// See `ListMessagesRequest` for the full semantics. +message ListFoldersRequest { + // Maximum folders to return in this page; 0 means "server default". The + // server clamps this to its documented maximum. + uint32 page_size = 1; + // Opaque keyset cursor from a prior response's `next_page_token`; empty + // starts from the first page. Not an offset -- resumes strictly after the + // last folder of the previous page. + string page_token = 2; +} // ListFoldersResponse. message ListFoldersResponse { repeated Folder folders = 1; + // Opaque cursor for the next page, or empty when this is the last page. + string next_page_token = 2; } -// ListMessagesRequest. `limit` caps the number of messages returned -// (newest first); a value of 0 means "use the server's default limit". +// ListMessagesRequest pages a folder's messages, newest first, using the +// shared keyset pagination convention used by every list/search RPC: +// - `page_size` bounds one page; 0 selects the server default and the +// server clamps any value to its documented maximum. +// - `page_token` is an OPAQUE cursor, NOT a numeric offset. It encodes the +// stable sort key (here `(received_at, id)`) of the last row of the +// previous page, so paging resumes strictly AFTER that row and stays +// correct under concurrent inserts (no skipped or duplicated rows). +// Leave empty to fetch the first page; otherwise pass a prior response's +// `next_page_token` verbatim. A token the server cannot decode is +// rejected with a `VALIDATION_FAILED` error, never silently ignored. message ListMessagesRequest { // Mailbox folder identifier (e.g. "inbox"). string folder_id = 1; // Maximum number of messages to return; 0 means "server default". - uint32 limit = 2; + uint32 page_size = 2; + // Opaque keyset cursor from a prior response's `next_page_token`; empty + // starts from the newest message. + string page_token = 3; } // ListMessagesResponse. message ListMessagesResponse { repeated Message messages = 1; + // Opaque cursor for the next page, or empty when this is the last page. + string next_page_token = 2; } // GetMessageRequest. @@ -507,10 +535,19 @@ message MarkReadRequest { // response (as opposed to a `Status` error) is confirmation enough. message MarkReadResponse {} -// SearchMessagesRequest. +// SearchMessagesRequest. Pages hits using the shared keyset convention (see +// `ListMessagesRequest`); the cursor here encodes `(rank, id)` so paging +// preserves the relevance ordering while resuming strictly after the last +// hit of the previous page. message SearchMessagesRequest { // Full-text search query string. string query = 1; + // Maximum hits to return in this page; 0 means "server default". Clamped + // to the server's documented maximum. + uint32 page_size = 2; + // Opaque keyset cursor from a prior response's `next_page_token`; empty + // starts from the highest-ranked hit. + string page_token = 3; } // Mirrors `nuncio_store::search::SearchHit`. @@ -526,6 +563,8 @@ message MessageSearchHit { // SearchMessagesResponse. message SearchMessagesResponse { repeated MessageSearchHit hits = 1; + // Opaque cursor for the next page, or empty when this is the last page. + string next_page_token = 2; } // SendMessageRequest carries a composed outbound email. The sending @@ -595,7 +634,8 @@ service Filters { // Rejects a syntactically or semantically invalid rule with // `Status::invalid_argument` -- nothing is ever persisted on failure. rpc CreateRule(CreateRuleRequest) returns (CreateRuleResponse); - // ListRules returns every persisted filter rule, priority ascending. + // ListRules returns a keyset-paginated page of persisted filter rules, + // priority ascending. rpc ListRules(ListRulesRequest) returns (ListRulesResponse); // DeleteRule removes a persisted filter rule by ID and reloads the // daemon's live `FilterEngine` rule set so the removal takes effect @@ -690,12 +730,23 @@ message CreateRuleResponse { FilterRule rule = 1; } -// ListRulesRequest currently carries no parameters. -message ListRulesRequest {} +// ListRulesRequest pages rules (priority ascending, id as tiebreaker) using +// the shared keyset convention; the cursor encodes `(priority, id)`. See +// `ListMessagesRequest` for the full semantics. +message ListRulesRequest { + // Maximum rules to return in this page; 0 means "server default". Clamped + // to the server's documented maximum. + uint32 page_size = 1; + // Opaque keyset cursor from a prior response's `next_page_token`; empty + // starts from the highest-priority rule. + string page_token = 2; +} // ListRulesResponse. message ListRulesResponse { repeated FilterRule rules = 1; + // Opaque cursor for the next page, or empty when this is the last page. + string next_page_token = 2; } // DeleteRuleRequest. @@ -956,9 +1007,9 @@ service Calendar { // returns `Status::internal` describing that gap unless a test double has // been injected -- never a fabricated `synced_count`. rpc Sync(CalendarSyncRequest) returns (CalendarSyncResponse); - // ListEvents returns persisted events for `account_id` whose window - // overlaps `[start_window, end_window]`, ordered by start time -- always - // genuinely persisted data, never fabricated. + // ListEvents returns a keyset-paginated page of persisted events for + // `account_id` whose window overlaps `[start_window, end_window]`, ordered + // by start time -- always genuinely persisted data, never fabricated. rpc ListEvents(ListEventsRequest) returns (ListEventsResponse); // GetEvent returns a single persisted event by ID. rpc GetEvent(GetEventRequest) returns (GetEventResponse); @@ -1001,17 +1052,29 @@ message CalendarSyncResponse { uint64 synced_count = 1; } -// ListEventsRequest. +// ListEventsRequest pages events (start time ascending, id as tiebreaker) +// using the shared keyset convention; the cursor encodes `(start_time, id)`. +// Recurrence expansion happens server-side before paging, so an expanded +// occurrence's `(start_time, id)` is what the cursor tracks. See +// `ListMessagesRequest` for the full semantics. message ListEventsRequest { string account_id = 1; string calendar_id = 2; google.protobuf.Timestamp start_window = 3; google.protobuf.Timestamp end_window = 4; + // Maximum events to return in this page; 0 means "server default". Clamped + // to the server's documented maximum. + uint32 page_size = 5; + // Opaque keyset cursor from a prior response's `next_page_token`; empty + // starts from the earliest event in the window. + string page_token = 6; } // ListEventsResponse. message ListEventsResponse { repeated CalendarEvent events = 1; + // Opaque cursor for the next page, or empty when this is the last page. + string next_page_token = 2; } // GetEventRequest. @@ -1052,9 +1115,9 @@ service Contacts { // describing that gap unless a test double has been injected -- never a // fabricated `synced_count`. rpc Sync(ContactsSyncRequest) returns (ContactsSyncResponse); - // ListContacts returns persisted contacts for `account_id`, ordered by - // interaction count then display name -- always genuinely persisted data, - // never fabricated. + // ListContacts returns a keyset-paginated page of persisted contacts for + // `account_id`, ordered by interaction count (descending) then display + // name -- always genuinely persisted data, never fabricated. rpc ListContacts(ListContactsRequest) returns (ListContactsResponse); // GetContact returns a single persisted contact by ID. rpc GetContact(GetContactRequest) returns (GetContactResponse); @@ -1132,14 +1195,24 @@ message ContactsSyncResponse { // ListContactsRequest. Contacts are always scoped to the account whose // address book they belong to, mirroring the store's `list_contacts` -// signature. +// signature. Pages using the shared keyset convention; the cursor encodes +// `(interaction_count, display_name, id)` to match the sort order. See +// `ListMessagesRequest` for the full semantics. message ListContactsRequest { string account_id = 1; + // Maximum contacts to return in this page; 0 means "server default". + // Clamped to the server's documented maximum. + uint32 page_size = 2; + // Opaque keyset cursor from a prior response's `next_page_token`; empty + // starts from the most-interacted contact. + string page_token = 3; } // ListContactsResponse. message ListContactsResponse { repeated Contact contacts = 1; + // Opaque cursor for the next page, or empty when this is the last page. + string next_page_token = 2; } // GetContactRequest. @@ -1183,8 +1256,8 @@ message CreateContactResponse { // `BearerAuthInterceptor` that guards every other service `nunciod` // exposes -- see `nunciod::grpc::serve_on_listener`. service Audit { - // ListRecords returns a page of the real, persisted WORM audit ledger, - // sequence ascending. + // ListRecords returns a keyset-paginated page of the real, persisted WORM + // audit ledger, sequence ascending. rpc ListRecords(ListRecordsRequest) returns (ListRecordsResponse); // VerifyChain re-verifies the ENTIRE persisted WORM audit ledger's // HMAC hash-chain integrity, using the ledger's real WORM HMAC key @@ -1215,17 +1288,25 @@ message AuditRecord { string record_hmac = 7; } -// ListRecordsRequest. `limit` caps the number of records returned -// (sequence ascending); a value of 0 means "use the server's default -// limit". +// ListRecordsRequest pages the WORM audit ledger, sequence ascending, using +// the shared keyset convention that every list/search RPC uses -- replacing +// the previous limit/offset paging, whose offset was unstable under +// concurrent appends. The cursor encodes the last record's `sequence`. See +// `ListMessagesRequest` for the full semantics. message ListRecordsRequest { - uint32 limit = 1; - uint32 offset = 2; + // Maximum records to return in this page; 0 means "server default". + // Clamped to the server's documented maximum. + uint32 page_size = 1; + // Opaque keyset cursor from a prior response's `next_page_token`; empty + // starts from the lowest sequence. + string page_token = 2; } // ListRecordsResponse. message ListRecordsResponse { repeated AuditRecord records = 1; + // Opaque cursor for the next page, or empty when this is the last page. + string next_page_token = 2; } // VerifyChainRequest currently carries no parameters. diff --git a/crates/nuncio-store/src/db.rs b/crates/nuncio-store/src/db.rs index 7cba21a..c58912c 100644 --- a/crates/nuncio-store/src/db.rs +++ b/crates/nuncio-store/src/db.rs @@ -1291,6 +1291,102 @@ impl DatabaseEngine { .collect::, DatabaseError>>() } + /// Keyset-paginated listing of a folder's messages, newest first + /// (`received_at DESC, id DESC`). `after` is the `(received_at, id)` of + /// the last message of the previous page; `None` starts from the newest. + /// Fetches `page_size + 1` rows to detect a following page: returns at + /// most `page_size` messages plus the `(received_at, id)` cursor of the + /// last returned message when more remain (else `None`). + #[allow(clippy::type_complexity)] + pub async fn list_messages_page( + &self, + folder_id: &str, + after: Option<(i64, String)>, + page_size: usize, + ) -> Result<(Vec, Option<(i64, String)>), DatabaseError> { + let fetch = page_size.saturating_add(1); + let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new( + "SELECT id, account_id, folder_id, subject, sender, recipient, received_at, read_flag, body_plain, body_html, remote_id, uid_validity FROM messages WHERE folder_id = ", + ); + builder.push_bind(folder_id.to_string()); + if let Some((ts, id)) = &after { + builder.push(" AND (received_at < "); + builder.push_bind(*ts); + builder.push(" OR (received_at = "); + builder.push_bind(*ts); + builder.push(" AND id < "); + builder.push_bind(id.clone()); + builder.push("))"); + } + builder.push(" ORDER BY received_at DESC, id DESC LIMIT "); + builder.push_bind(fetch as i64); + + let rows = builder + .build_query_as::<( + String, + String, + String, + String, + String, + String, + i64, + i64, + Option, + Option, + String, + String, + )>() + .fetch_all(&self.pool) + .await + .map_err(DatabaseError::Query)?; + + let has_more = rows.len() > page_size; + let mut emails = rows + .into_iter() + .take(page_size) + .map(|r| { + let dec_plain = + r.8.map(|p| { + crate::cipher::PayloadCipher::decrypt_text_at_rest(&self.storage_key, &p) + }) + .transpose() + .map_err(|e| DatabaseError::Decryption(e.to_string()))?; + let dec_html = + r.9.map(|h| { + crate::cipher::PayloadCipher::decrypt_text_at_rest(&self.storage_key, &h) + }) + .transpose() + .map_err(|e| DatabaseError::Decryption(e.to_string()))?; + Ok(nuncio_core::model::Email { + id: r.0, + account_id: r.1, + folder_id: r.2, + remote_id: r.10, + uid_validity: r.11, + subject: r.3, + sender: r.4, + recipient: r.5, + received_at: r.6, + read: r.7 != 0, + body_plain: dec_plain, + body_html: dec_html, + attachments: Vec::new(), + }) + }) + .collect::, DatabaseError>>()?; + + let next = if has_more { + emails + .last() + .map(|e| (e.received_at, e.id.clone())) + .filter(|_| !emails.is_empty()) + } else { + None + }; + emails.shrink_to_fit(); + Ok((emails, next)) + } + /// Retrieve a single message by ID. #[allow(clippy::type_complexity)] pub async fn get_message( @@ -1632,6 +1728,69 @@ impl DatabaseEngine { Ok(rows.iter().map(contact_from_row).collect()) } + /// Keyset-paginated listing of `account_id`'s contacts, in the same order + /// as [`Self::list_contacts`] (`interaction_count DESC, display_name ASC, + /// id ASC`). `after` is the `(interaction_count, display_name, id)` of the + /// last contact of the previous page. Fetches `page_size + 1` rows to + /// detect a following page and returns the `(interaction_count, + /// display_name, id)` cursor of the last returned contact when more remain. + /// + /// See the confidentiality/dependency note on [`Self::save_contact`]. + #[allow(clippy::type_complexity)] + pub async fn list_contacts_page( + &self, + account_id: &str, + after: Option<(i64, String, String)>, + page_size: usize, + ) -> Result<(Vec, Option<(i64, String, String)>), DatabaseError> { + let fetch = page_size.saturating_add(1); + let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new( + "SELECT id, account_id, display_name, given_name, family_name, organization, \ + job_title, notes, avatar_url, emails_json, phones_json, is_favorite, \ + interaction_count, last_interacted_at, created_at, updated_at \ + FROM contacts WHERE account_id = ", + ); + builder.push_bind(account_id.to_string()); + if let Some((ic, dn, id)) = &after { + builder.push(" AND (interaction_count < "); + builder.push_bind(*ic); + builder.push(" OR (interaction_count = "); + builder.push_bind(*ic); + builder.push(" AND (display_name > "); + builder.push_bind(dn.clone()); + builder.push(" OR (display_name = "); + builder.push_bind(dn.clone()); + builder.push(" AND id > "); + builder.push_bind(id.clone()); + builder.push("))))"); + } + builder.push(" ORDER BY interaction_count DESC, display_name ASC, id ASC LIMIT "); + builder.push_bind(fetch as i64); + + let rows = builder + .build() + .fetch_all(&self.pool) + .await + .map_err(DatabaseError::Query)?; + + let has_more = rows.len() > page_size; + let contacts: Vec = + rows.iter().take(page_size).map(contact_from_row).collect(); + + let next = if has_more { + contacts.last().map(|c| { + ( + c.interaction_count as i64, + c.display_name.clone(), + c.id.clone(), + ) + }) + } else { + None + }; + Ok((contacts, next)) + } + /// Retrieve a single contact by ID. /// /// See the confidentiality/dependency note on [`Self::save_contact`]. @@ -1704,6 +1863,52 @@ impl DatabaseEngine { .collect()) } + /// Keyset-paginated listing of folders with message counts, folder id + /// ascending. `after` is the id of the last folder of the previous page. + /// Fetches `page_size + 1` groups to detect a following page and returns + /// the id of the last returned folder as the cursor when more remain. + pub async fn list_folders_page( + &self, + after: Option, + page_size: usize, + ) -> Result<(Vec, Option), DatabaseError> { + let fetch = page_size.saturating_add(1); + let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new( + "SELECT folder_id, COUNT(*) as total, SUM(CASE WHEN read_flag = 0 THEN 1 ELSE 0 END) as unread FROM messages ", + ); + if let Some(id) = &after { + builder.push("WHERE folder_id > "); + builder.push_bind(id.clone()); + } + builder.push(" GROUP BY folder_id ORDER BY folder_id ASC LIMIT "); + builder.push_bind(fetch as i64); + + let rows = builder + .build_query_as::<(String, i64, i64)>() + .fetch_all(&self.pool) + .await + .map_err(DatabaseError::Query)?; + + let has_more = rows.len() > page_size; + let folders: Vec = rows + .into_iter() + .take(page_size) + .map(|(folder_id, total, unread)| nuncio_core::model::Folder { + id: folder_id.clone(), + name: folder_id, + total_messages: total as usize, + unread_messages: unread as usize, + }) + .collect(); + + let next = if has_more { + folders.last().map(|f| f.id.clone()) + } else { + None + }; + Ok((folders, next)) + } + /// Save or replace a [`nuncio_filter::FilterRule`]. pub async fn save_filter_rule( &self, @@ -1772,6 +1977,77 @@ impl DatabaseEngine { Ok(rules) } + /// Keyset-paginated listing of persisted filter rules, `priority ASC, id + /// ASC`. `after` is the `(priority, id)` of the last rule of the previous + /// page. Fetches `page_size + 1` rows to detect a following page. + /// + /// The following-page detection and the returned cursor are computed from + /// the RAW rows (their stored `(priority, id)`), before parsing -- a row + /// whose stored NSQL no longer parses is skipped from the returned rules + /// exactly as [`Self::list_filter_rules`] does, but it still counts toward + /// the page boundary, so a page with skipped rows still advances the + /// cursor correctly and never re-emits or gaps around a rule. + #[allow(clippy::type_complexity)] + pub async fn list_filter_rules_page( + &self, + after: Option<(i32, String)>, + page_size: usize, + ) -> Result<(Vec, Option<(i32, String)>), DatabaseError> { + let fetch = page_size.saturating_add(1); + let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new( + "SELECT id, name, priority, enabled, nsql_text, created_at, updated_at FROM filter_rules ", + ); + if let Some((priority, id)) = &after { + builder.push("WHERE (priority > "); + builder.push_bind(*priority as i64); + builder.push(" OR (priority = "); + builder.push_bind(*priority as i64); + builder.push(" AND id > "); + builder.push_bind(id.clone()); + builder.push("))"); + } + builder.push(" ORDER BY priority ASC, id ASC LIMIT "); + builder.push_bind(fetch as i64); + + let rows = builder + .build_query_as::<(String, String, i64, i64, String, i64, i64)>() + .fetch_all(&self.pool) + .await + .map_err(DatabaseError::Query)?; + + let has_more = rows.len() > page_size; + let page: Vec<(String, String, i64, i64, String, i64, i64)> = + rows.into_iter().take(page_size).collect(); + + let next = if has_more { + page.last().map(|r| (r.2 as i32, r.0.clone())) + } else { + None + }; + + let mut rules = Vec::new(); + for (id, name, priority, enabled, nsql_text, created_at, updated_at) in page { + match nuncio_filter::NsqlParser::parse_rule(&name, priority as i32, &nsql_text) { + Ok(mut parsed) => { + parsed.id = id; + parsed.enabled = enabled != 0; + parsed.created_at = created_at; + parsed.updated_at = updated_at; + rules.push(parsed); + } + Err(err) => { + tracing::warn!( + rule_id = %id, + rule_name = %name, + error = %err, + "stored filter rule failed to parse and will not be enforced" + ); + } + } + } + Ok((rules, next)) + } + /// Delete a [`nuncio_filter::FilterRule`] by ID. pub async fn delete_filter_rule(&self, rule_id: &str) -> Result<(), DatabaseError> { sqlx::query("DELETE FROM filter_rules WHERE id = ?") @@ -2261,6 +2537,57 @@ impl DatabaseEngine { .collect()) } + /// Keyset-paginated listing of the WORM audit ledger, `sequence ASC`. + /// `after` is the sequence of the last record of the previous page; + /// `None` starts from the lowest sequence. Fetches `page_size + 1` rows to + /// detect a following page and returns the sequence of the last returned + /// record as the cursor when more remain. Keyset paging on the + /// monotonically increasing sequence is stable under concurrent appends, + /// unlike the previous limit/offset paging. + pub async fn list_worm_audit_records_page( + &self, + after: Option, + page_size: usize, + ) -> Result<(Vec, Option), DatabaseError> { + let fetch = page_size.saturating_add(1); + let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new( + "SELECT sequence, timestamp_ns, actor, action, data_hash, previous_block_hash, record_hmac FROM worm_audit_records ", + ); + if let Some(seq) = after { + builder.push("WHERE sequence > "); + builder.push_bind(seq as i64); + } + builder.push(" ORDER BY sequence ASC LIMIT "); + builder.push_bind(fetch as i64); + + let rows = builder + .build_query_as::<(i64, i64, String, String, String, String, String)>() + .fetch_all(&self.pool) + .await?; + + let has_more = rows.len() > page_size; + let records: Vec = rows + .into_iter() + .take(page_size) + .map(|r| nuncio_core::WormAuditRecord { + sequence: r.0 as u64, + timestamp_ns: r.1, + actor: r.2, + action: r.3, + data_hash: r.4, + previous_block_hash: r.5, + record_hmac: r.6, + }) + .collect(); + + let next = if has_more { + records.last().map(|r| r.sequence) + } else { + None + }; + Ok((records, next)) + } + /// Verify the entire WORM cryptographic audit log chain, using the WORM HMAC key /// provisioned for this engine. pub async fn verify_worm_audit_chain(&self) -> Result<(), DatabaseError> { diff --git a/crates/nuncio-store/src/search.rs b/crates/nuncio-store/src/search.rs index 0befdf5..dc25f64 100644 --- a/crates/nuncio-store/src/search.rs +++ b/crates/nuncio-store/src/search.rs @@ -68,6 +68,71 @@ impl<'a> SearchEngine<'a> { .collect()) } + /// Keyset-paginated full-text search over message subjects, senders, and + /// bodies, ordered by relevance (`rank ASC, id ASC`). `after` is the + /// `(rank, id)` of the last hit of the previous page; `None` starts from + /// the highest-ranked hit. Fetches `page_size + 1` hits to detect a + /// following page and returns the `(rank, id)` cursor of the last returned + /// hit when more remain. + /// + /// The bm25 `rank` is materialized in an inner query (where the FTS5 + /// `MATCH` is in scope) into an ordinary column, so the outer query can + /// both keyset-filter and order by `(rank, id)` -- preserving relevance + /// order across pages while resuming strictly after the previous page's + /// last hit. + pub async fn search_messages_page( + &self, + query: &str, + after: Option<(f64, String)>, + page_size: usize, + ) -> Result<(Vec, Option<(f64, String)>), DatabaseError> { + let clean_query = Self::sanitize_fts5_query(query); + if clean_query.is_empty() { + return Ok((Vec::new(), None)); + } + let fetch = page_size.saturating_add(1); + + let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new( + "SELECT id, title, snippet, rk FROM ( \ + SELECT id, subject AS title, \ + snippet(messages_fts, 3, '', '', '...', 10) AS snippet, rank AS rk \ + FROM messages_fts WHERE messages_fts MATCH ", + ); + builder.push_bind(clean_query); + builder.push(") "); + if let Some((rank, id)) = &after { + builder.push("WHERE (rk > "); + builder.push_bind(*rank); + builder.push(" OR (rk = "); + builder.push_bind(*rank); + builder.push(" AND id > "); + builder.push_bind(id.clone()); + builder.push("))"); + } + builder.push(" ORDER BY rk ASC, id ASC LIMIT "); + builder.push_bind(fetch as i64); + + let rows = builder + .build_query_as::<(String, String, String, f64)>() + .fetch_all(self.db.pool()) + .await + .map_err(DatabaseError::Query)?; + + let has_more = rows.len() > page_size; + let mut last_key: Option<(f64, String)> = None; + let hits: Vec = rows + .into_iter() + .take(page_size) + .map(|(id, title, snippet, rk)| { + last_key = Some((rk, id.clone())); + SearchHit { id, title, snippet } + }) + .collect(); + + let next = if has_more { last_key } else { None }; + Ok((hits, next)) + } + /// Perform a full-text trigram search over calendar event summaries and locations. pub async fn search_events(&self, query: &str) -> Result, DatabaseError> { let clean_query = Self::sanitize_fts5_query(query); @@ -224,6 +289,52 @@ mod tests { assert_eq!(hits[0].title, "Legacy Roadmap Notes"); } + /// Paging FTS search results across multiple pages via the `(rank, id)` + /// keyset must return every matching hit EXACTLY ONCE, no dupes and no + /// gaps -- proving the rank-materializing subquery keyset works end to + /// end over the real encrypt-then-index write path. + #[tokio::test] + async fn search_messages_page_returns_every_hit_once_no_gaps() { + let (db, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap(); + for i in 0..5 { + db.save_email(&sample_email( + &format!("msg-{i}"), + "Budget Planning Meeting", + "annual budget revenue forecast discussion", + )) + .await + .unwrap(); + } + + let search = SearchEngine::new(&db); + let mut seen = Vec::new(); + let mut after: Option<(f64, String)> = None; + let mut pages = 0; + loop { + let (hits, next) = search + .search_messages_page("budget", after.clone(), 2) + .await + .unwrap(); + pages += 1; + assert!(hits.len() <= 2, "page must not exceed page_size"); + seen.extend(hits.iter().map(|h| h.id.clone())); + match next { + Some(cursor) => after = Some(cursor), + None => break, + } + assert!(pages < 100, "pagination must terminate"); + } + + seen.sort(); + seen.dedup(); + assert_eq!( + seen.len(), + 5, + "every matching message returned exactly once" + ); + assert!(pages >= 3, "5 hits at page_size 2 must span multiple pages"); + } + #[tokio::test] async fn fts5_event_search_and_triggers() { let (db, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap(); diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs index 7ac4718..234b4e1 100644 --- a/crates/nunciod/src/grpc.rs +++ b/crates/nunciod/src/grpc.rs @@ -5,6 +5,7 @@ //! //! This is the daemon's sole client-facing transport. +use crate::pagination::{self, CursorField}; use nuncio_cal::CalendarBackend; use nuncio_contacts::ContactsBackend; use nuncio_core::{CoreCommand, CoreEvent, EventBus}; @@ -745,10 +746,6 @@ impl Accounts for AccountsGrpcService { } } -/// Default cap on messages returned by `ListMessages` when the caller -/// supplies `limit: 0` ("use the server default"). -const DEFAULT_LIST_MESSAGES_LIMIT: usize = 50; - /// Maps a `nuncio_core::model::Attachment` onto its wire-format /// `nuncio.v1.Attachment` representation. fn map_attachment_to_proto(attachment: nuncio_core::model::Attachment) -> AttachmentProto { @@ -1083,6 +1080,11 @@ impl Calendar for CalendarGrpcService { } let start_window = require_window_bound(req.start_window, "start_window")?; let end_window = require_window_bound(req.end_window, "end_window")?; + let page_size = pagination::clamp_page_size(req.page_size); + let after = pagination::parse_token(&req.page_token, |f| match f { + [CursorField::I(ts), CursorField::S(id)] => Some((*ts, id.clone())), + _ => None, + })?; let all_in_window = self .db @@ -1101,12 +1103,45 @@ impl Calendar for CalendarGrpcService { events.retain(|e| req.calendar_id.is_empty() || e.calendar_id == req.calendar_id); + // Recurrence expansion materializes the result set in memory, so keyset + // pagination is applied to the sorted expanded list here rather than at + // the store. Each occurrence has a distinct `(start_time, id)` (an + // occurrence shares its master's id but never its start time), so the + // pair is a stable, unique keyset over the whole set. + events.sort_by(|a, b| { + a.start_time + .cmp(&b.start_time) + .then_with(|| a.id.cmp(&b.id)) + }); + if let Some((ts, id)) = &after { + events.retain(|e| e.start_time > *ts || (e.start_time == *ts && e.id > *id)); + } + + let has_more = events.len() > page_size; + events.truncate(page_size); + let next_page_token = if has_more { + events + .last() + .map(|e| { + pagination::encode_cursor(&[ + CursorField::I(e.start_time), + CursorField::S(e.id.clone()), + ]) + }) + .unwrap_or_default() + } else { + String::new() + }; + let events = events .into_iter() .map(map_calendar_event_to_proto) .collect(); - Ok(Response::new(ListEventsResponse { events })) + Ok(Response::new(ListEventsResponse { + events, + next_page_token, + })) } async fn get_event( @@ -1212,17 +1247,35 @@ impl Contacts for ContactsGrpcService { if req.account_id.is_empty() { return Err(Status::invalid_argument("account_id is required")); } + let page_size = pagination::clamp_page_size(req.page_size); + let after = pagination::parse_token(&req.page_token, |f| match f { + [CursorField::I(ic), CursorField::S(dn), CursorField::S(id)] => { + Some((*ic, dn.clone(), id.clone())) + } + _ => None, + })?; - let contacts = self + let (contacts, next) = self .db - .list_contacts(&req.account_id) + .list_contacts_page(&req.account_id, after, page_size) .await - .map_err(|e| Status::internal(format!("failed to list contacts: {e}")))? - .into_iter() - .map(map_contact_to_proto) - .collect(); + .map_err(|e| Status::internal(format!("failed to list contacts: {e}")))?; - Ok(Response::new(ListContactsResponse { contacts })) + let next_page_token = next + .map(|(ic, dn, id)| { + pagination::encode_cursor(&[ + CursorField::I(ic), + CursorField::S(dn), + CursorField::S(id), + ]) + }) + .unwrap_or_default(); + let contacts = contacts.into_iter().map(map_contact_to_proto).collect(); + + Ok(Response::new(ListContactsResponse { + contacts, + next_page_token, + })) } async fn get_contact( @@ -1364,18 +1417,30 @@ struct MailGrpcService { impl Mail for MailGrpcService { async fn list_folders( &self, - _request: Request, + request: Request, ) -> Result, Status> { - let folders = self - .db - .list_folders() - .await - .map_err(|e| Status::internal(format!("failed to list folders: {e}")))? - .into_iter() - .map(map_folder_to_proto) - .collect(); + let req = request.into_inner(); + let page_size = pagination::clamp_page_size(req.page_size); + let after = pagination::parse_token(&req.page_token, |f| match f { + [CursorField::S(id)] => Some(id.clone()), + _ => None, + })?; - Ok(Response::new(ListFoldersResponse { folders })) + let (folders, next) = self + .db + .list_folders_page(after, page_size) + .await + .map_err(|e| Status::internal(format!("failed to list folders: {e}")))?; + + let next_page_token = next + .map(|id| pagination::encode_cursor(&[CursorField::S(id)])) + .unwrap_or_default(); + let folders = folders.into_iter().map(map_folder_to_proto).collect(); + + Ok(Response::new(ListFoldersResponse { + folders, + next_page_token, + })) } async fn list_messages( @@ -1386,22 +1451,27 @@ impl Mail for MailGrpcService { if req.folder_id.is_empty() { return Err(Status::invalid_argument("folder_id is required")); } - let limit = if req.limit == 0 { - DEFAULT_LIST_MESSAGES_LIMIT - } else { - req.limit as usize - }; + let page_size = pagination::clamp_page_size(req.page_size); + let after = pagination::parse_token(&req.page_token, |f| match f { + [CursorField::I(ts), CursorField::S(id)] => Some((*ts, id.clone())), + _ => None, + })?; - let messages = self + let (messages, next) = self .db - .list_messages(&req.folder_id, limit) + .list_messages_page(&req.folder_id, after, page_size) .await - .map_err(|e| Status::internal(format!("failed to list messages: {e}")))? - .into_iter() - .map(map_email_to_proto) - .collect(); + .map_err(|e| Status::internal(format!("failed to list messages: {e}")))?; - Ok(Response::new(ListMessagesResponse { messages })) + let next_page_token = next + .map(|(ts, id)| pagination::encode_cursor(&[CursorField::I(ts), CursorField::S(id)])) + .unwrap_or_default(); + let messages = messages.into_iter().map(map_email_to_proto).collect(); + + Ok(Response::new(ListMessagesResponse { + messages, + next_page_token, + })) } async fn get_message( @@ -1467,11 +1537,24 @@ impl Mail for MailGrpcService { request: Request, ) -> Result, Status> { let req = request.into_inner(); + let page_size = pagination::clamp_page_size(req.page_size); + let after = pagination::parse_token(&req.page_token, |f| match f { + [CursorField::F(rank), CursorField::S(id)] => Some((*rank, id.clone())), + _ => None, + })?; + let search = SearchEngine::new(&self.db); - let hits = search - .search_messages(&req.query) + let (hits, next) = search + .search_messages_page(&req.query, after, page_size) .await - .map_err(|e| Status::internal(format!("search failed: {e}")))? + .map_err(|e| Status::internal(format!("search failed: {e}")))?; + + let next_page_token = next + .map(|(rank, id)| { + pagination::encode_cursor(&[CursorField::F(rank), CursorField::S(id)]) + }) + .unwrap_or_default(); + let hits = hits .into_iter() .map(|hit| MessageSearchHit { id: hit.id, @@ -1480,7 +1563,10 @@ impl Mail for MailGrpcService { }) .collect(); - Ok(Response::new(SearchMessagesResponse { hits })) + Ok(Response::new(SearchMessagesResponse { + hits, + next_page_token, + })) } /// SendMessage composes and sends a real outbound email over SMTP via @@ -1753,18 +1839,32 @@ impl Filters for FiltersGrpcService { async fn list_rules( &self, - _request: Request, + request: Request, ) -> Result, Status> { - let rules = self - .db - .list_filter_rules() - .await - .map_err(|e| Status::internal(format!("failed to list filter rules: {e}")))? - .into_iter() - .map(map_filter_rule_to_proto) - .collect(); + let req = request.into_inner(); + let page_size = pagination::clamp_page_size(req.page_size); + let after = pagination::parse_token(&req.page_token, |f| match f { + [CursorField::I(priority), CursorField::S(id)] => Some((*priority as i32, id.clone())), + _ => None, + })?; - Ok(Response::new(ListRulesResponse { rules })) + let (rules, next) = self + .db + .list_filter_rules_page(after, page_size) + .await + .map_err(|e| Status::internal(format!("failed to list filter rules: {e}")))?; + + let next_page_token = next + .map(|(priority, id)| { + pagination::encode_cursor(&[CursorField::I(priority as i64), CursorField::S(id)]) + }) + .unwrap_or_default(); + let rules = rules.into_iter().map(map_filter_rule_to_proto).collect(); + + Ok(Response::new(ListRulesResponse { + rules, + next_page_token, + })) } async fn delete_rule( @@ -2178,8 +2278,6 @@ impl Export for ExportGrpcService { /// Default cap on records returned by `Audit/ListRecords` when the caller /// supplies `limit: 0` ("use the server default"). -const DEFAULT_LIST_RECORDS_LIMIT: u32 = 100; - /// Maps a `nuncio_core::WormAuditRecord` onto its wire-format /// `nuncio.v1.AuditRecord` representation. `record_hmac` is a verification /// MAC output, never the WORM HMAC signing key itself -- see the message's @@ -2214,22 +2312,27 @@ impl Audit for AuditGrpcService { request: Request, ) -> Result, Status> { let req = request.into_inner(); - let limit = if req.limit == 0 { - DEFAULT_LIST_RECORDS_LIMIT - } else { - req.limit - }; + let page_size = pagination::clamp_page_size(req.page_size); + let after = pagination::parse_token(&req.page_token, |f| match f { + [CursorField::U(seq)] => Some(*seq), + _ => None, + })?; - let records = self + let (records, next) = self .db - .list_worm_audit_records(limit, req.offset) + .list_worm_audit_records_page(after, page_size) .await - .map_err(|e| Status::internal(format!("failed to list audit records: {e}")))? - .into_iter() - .map(map_audit_record_to_proto) - .collect(); + .map_err(|e| Status::internal(format!("failed to list audit records: {e}")))?; - Ok(Response::new(ListRecordsResponse { records })) + let next_page_token = next + .map(|seq| pagination::encode_cursor(&[CursorField::U(seq)])) + .unwrap_or_default(); + let records = records.into_iter().map(map_audit_record_to_proto).collect(); + + Ok(Response::new(ListRecordsResponse { + records, + next_page_token, + })) } /// Re-verifies the ENTIRE persisted WORM audit ledger via @@ -3721,6 +3824,110 @@ mod tests { } } + /// Paging a folder larger than `page_size` across multiple pages via + /// `next_page_token` must return every message EXACTLY ONCE, with no + /// duplicates and no gaps, and in the stable newest-first keyset order + /// (`received_at DESC, id DESC`). Some messages deliberately share a + /// `received_at` so the `(received_at, id)` tiebreaker is exercised across + /// a page boundary. + #[tokio::test] + async fn list_messages_keyset_pagination_returns_every_message_once_no_gaps() { + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("connect ephemeral test db"); + let seed = [ + ("m1", 1_700_000_000_i64), + ("m2", 1_700_000_000), + ("m3", 1_700_000_010), + ("m4", 1_700_000_010), + ("m5", 1_700_000_020), + ("m6", 1_700_000_030), + ("m7", 1_700_000_030), + ]; + for (id, ts) in seed { + let mut email = sample_email(id, "inbox", "subject", "body"); + email.received_at = ts; + db.save_email(&email).await.expect("seed message"); + } + + let event_bus = Arc::new(EventBus::new()); + let secrets = Arc::new(SecretManager::mock()); + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); + let (addr, _handle) = spawn_test_server_with( + event_bus, + Arc::new(db), + filter_engine, + secrets, + "correct-token", + ) + .await; + let mut client = MailClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let mut seen = Vec::new(); + let mut page_token = String::new(); + let mut pages = 0; + loop { + let resp = client + .list_messages(authed_bearer_request(ListMessagesRequest { + folder_id: "inbox".to_string(), + page_size: 2, + page_token: page_token.clone(), + })) + .await + .expect("list_messages page succeeds") + .into_inner(); + pages += 1; + assert!( + resp.messages.len() <= 2, + "a page must never exceed the requested page_size" + ); + seen.extend(resp.messages.iter().map(|m| m.id.clone())); + if resp.next_page_token.is_empty() { + break; + } + page_token = resp.next_page_token; + assert!(pages < 100, "pagination must terminate"); + } + + // Every seeded message appears exactly once (no dupes, no gaps), in + // the stable newest-first order including the id tiebreaker. + assert_eq!( + seen, + vec!["m7", "m6", "m5", "m4", "m3", "m2", "m1"], + "keyset paging must yield the full set once, newest-first" + ); + assert!(pages >= 4, "a 7-item set at page_size 2 must span >1 page"); + } + + /// A `page_token` the server cannot decode is rejected with a typed + /// `VALIDATION_FAILED` `ErrorInfo` mapped to `InvalidArgument` -- never a + /// panic and never a silently-empty first page. + #[tokio::test] + async fn list_messages_rejects_malformed_page_token() { + let event_bus = Arc::new(EventBus::new()); + let (addr, _handle, _dir) = spawn_test_server(event_bus, "correct-token").await; + let mut client = MailClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let err = client + .list_messages(authed_bearer_request(ListMessagesRequest { + folder_id: "inbox".to_string(), + page_size: 2, + page_token: "not-a-valid-cursor".to_string(), + })) + .await + .expect_err("a malformed page_token must be rejected"); + assert_eq!(err.code(), Code::InvalidArgument); + assert_eq!( + nuncio_proto::errors::error_reason(&err), + Some(ErrorReason::ValidationFailed), + "malformed page_token must carry a typed VALIDATION_FAILED reason" + ); + } + #[tokio::test] async fn mail_rpcs_reject_missing_bearer_token() { // Confirms `Mail` is mounted behind its own `BearerAuthInterceptor` @@ -3733,7 +3940,10 @@ mod tests { .expect("client connects"); let err = client - .list_folders(ListFoldersRequest {}) + .list_folders(ListFoldersRequest { + page_size: 0, + page_token: String::new(), + }) .await .expect_err("missing bearer token must be rejected"); assert_eq!(err.code(), Code::Unauthenticated); @@ -3741,7 +3951,8 @@ mod tests { let err = client .list_messages(ListMessagesRequest { folder_id: "inbox".to_string(), - limit: 10, + page_size: 10, + page_token: String::new(), }) .await .expect_err("missing bearer token must be rejected"); @@ -3767,6 +3978,8 @@ mod tests { let err = client .search_messages(SearchMessagesRequest { query: "hello".to_string(), + page_size: 0, + page_token: String::new(), }) .await .expect_err("missing bearer token must be rejected"); @@ -3814,6 +4027,8 @@ mod tests { calendar_id: "cal-1".to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(0)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(i64::MAX)), + page_size: 0, + page_token: String::new(), }) .await .expect_err("missing bearer token must be rejected"); @@ -3885,7 +4100,8 @@ mod tests { let err = client .list_messages(authed_bearer_request(ListMessagesRequest { folder_id: String::new(), - limit: 10, + page_size: 10, + page_token: String::new(), })) .await .expect_err("empty folder_id must be rejected"); @@ -3942,7 +4158,10 @@ mod tests { // ListFolders reports the seeded folder with the right counts. let folders = client - .list_folders(authed_bearer_request(ListFoldersRequest {})) + .list_folders(authed_bearer_request(ListFoldersRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_folders succeeds") .into_inner() @@ -3956,7 +4175,8 @@ mod tests { let messages = client .list_messages(authed_bearer_request(ListMessagesRequest { folder_id: "inbox".to_string(), - limit: 10, + page_size: 10, + page_token: String::new(), })) .await .expect("list_messages succeeds") @@ -3988,6 +4208,8 @@ mod tests { let hits = client .search_messages(authed_bearer_request(SearchMessagesRequest { query: "revenue".to_string(), + page_size: 0, + page_token: String::new(), })) .await .expect("search_messages succeeds") @@ -4513,7 +4735,10 @@ mod tests { assert_eq!(err.code(), Code::Unauthenticated); let err = client - .list_rules(ListRulesRequest {}) + .list_rules(ListRulesRequest { + page_size: 0, + page_token: String::new(), + }) .await .expect_err("missing bearer token must be rejected"); assert_eq!(err.code(), Code::Unauthenticated); @@ -4639,7 +4864,10 @@ mod tests { // ListRules reflects the persisted rule. let list_response = client - .list_rules(authed_bearer_request(ListRulesRequest {})) + .list_rules(authed_bearer_request(ListRulesRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_rules succeeds") .into_inner(); @@ -4655,7 +4883,10 @@ mod tests { .expect("delete_rule succeeds"); let list_after_delete = client - .list_rules(authed_bearer_request(ListRulesRequest {})) + .list_rules(authed_bearer_request(ListRulesRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_rules succeeds") .into_inner(); @@ -4685,7 +4916,10 @@ mod tests { assert_eq!(err.code(), Code::InvalidArgument); let list_response = client - .list_rules(authed_bearer_request(ListRulesRequest {})) + .list_rules(authed_bearer_request(ListRulesRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_rules succeeds") .into_inner(); @@ -4756,7 +4990,10 @@ mod tests { ); let list_response = client - .list_rules(authed_bearer_request(ListRulesRequest {})) + .list_rules(authed_bearer_request(ListRulesRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_rules succeeds") .into_inner(); @@ -4821,7 +5058,10 @@ mod tests { // Nothing was persisted by either validation call. let list_response = client - .list_rules(authed_bearer_request(ListRulesRequest {})) + .list_rules(authed_bearer_request(ListRulesRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_rules succeeds") .into_inner(); @@ -4889,7 +5129,10 @@ mod tests { // Previewing must never persist the rule. let list_response = client - .list_rules(authed_bearer_request(ListRulesRequest {})) + .list_rules(authed_bearer_request(ListRulesRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_rules succeeds") .into_inner(); @@ -4998,7 +5241,10 @@ mod tests { ); let list_response = client - .list_rules(authed_bearer_request(ListRulesRequest {})) + .list_rules(authed_bearer_request(ListRulesRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_rules succeeds") .into_inner(); @@ -5077,7 +5323,10 @@ mod tests { assert_eq!(err.code(), Code::InvalidArgument); let list_response = client - .list_rules(authed_bearer_request(ListRulesRequest {})) + .list_rules(authed_bearer_request(ListRulesRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_rules succeeds") .into_inner(); @@ -5154,7 +5403,10 @@ mod tests { assert_eq!(response.errors.len(), 1); let list_response = client - .list_rules(authed_bearer_request(ListRulesRequest {})) + .list_rules(authed_bearer_request(ListRulesRequest { + page_size: 0, + page_token: String::new(), + })) .await .expect("list_rules succeeds") .into_inner(); @@ -5406,8 +5658,8 @@ mod tests { let err = client .list_records(ListRecordsRequest { - limit: 10, - offset: 0, + page_size: 10, + page_token: String::new(), }) .await .expect_err("missing bearer token must be rejected"); @@ -5455,8 +5707,8 @@ mod tests { let response = client .list_records(authed_bearer_request(ListRecordsRequest { - limit: 0, - offset: 0, + page_size: 0, + page_token: String::new(), })) .await .expect("list_records succeeds") @@ -5471,6 +5723,81 @@ mod tests { assert_eq!(response.records[1].action, "test.action.two"); } + /// Keyset paging the WORM ledger across multiple pages must return every + /// record EXACTLY ONCE, in ascending sequence, with no dupes or gaps. + #[tokio::test] + async fn list_records_keyset_pagination_returns_every_record_once_no_gaps() { + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("connect ephemeral test db"); + for i in 0..7 { + db.append_worm_audit_record( + "system.test", + &format!("test.action.{i}"), + format!("payload-{i}").as_bytes(), + ) + .await + .expect("append audit record"); + } + + let event_bus = Arc::new(EventBus::new()); + let secrets = Arc::new(SecretManager::mock()); + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); + let (addr, _handle) = spawn_test_server_with( + event_bus, + Arc::new(db), + filter_engine, + secrets, + "correct-token", + ) + .await; + let mut client = AuditClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let mut seen = Vec::new(); + let mut page_token = String::new(); + let mut pages = 0; + loop { + let resp = client + .list_records(authed_bearer_request(ListRecordsRequest { + page_size: 2, + page_token: page_token.clone(), + })) + .await + .expect("list_records page succeeds") + .into_inner(); + pages += 1; + assert!(resp.records.len() <= 2, "page must not exceed page_size"); + seen.extend(resp.records.iter().map(|r| r.sequence)); + if resp.next_page_token.is_empty() { + break; + } + page_token = resp.next_page_token; + assert!(pages < 100, "pagination must terminate"); + } + + assert_eq!( + seen, + vec![1, 2, 3, 4, 5, 6, 7], + "keyset paging must yield every sequence once, ascending" + ); + + // A malformed page_token is a typed VALIDATION_FAILED error. + let err = client + .list_records(authed_bearer_request(ListRecordsRequest { + page_size: 2, + page_token: "garbage".to_string(), + })) + .await + .expect_err("malformed page_token must be rejected"); + assert_eq!(err.code(), Code::InvalidArgument); + assert_eq!( + nuncio_proto::errors::error_reason(&err), + Some(ErrorReason::ValidationFailed) + ); + } + /// Proves `Audit/VerifyChain` reports `valid: true` and the real /// record count on an intact, honestly-signed chain -- never a /// fabricated `valid`. @@ -5645,6 +5972,8 @@ mod tests { calendar_id: "cal-work".to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_699_999_000)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_700_004_000)), + page_size: 0, + page_token: String::new(), })) .await .expect("list_events succeeds") @@ -5704,6 +6033,8 @@ mod tests { calendar_id: "cal-work".to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_699_999_000)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_700_004_000)), + page_size: 0, + page_token: String::new(), })) .await .expect("list_events succeeds") @@ -5730,6 +6061,8 @@ mod tests { calendar_id: "cal-work".to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(0)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(i64::MAX)), + page_size: 0, + page_token: String::new(), })) .await .expect_err("empty account_id must be rejected"); @@ -5894,6 +6227,8 @@ mod tests { calendar_id: "cal-work".to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_699_999_000)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_700_004_000)), + page_size: 0, + page_token: String::new(), })) .await .expect("list_events succeeds") @@ -6068,6 +6403,8 @@ mod tests { let err = client .list_contacts(ListContactsRequest { account_id: "acct-1".to_string(), + page_size: 0, + page_token: String::new(), }) .await .expect_err("missing bearer token must be rejected"); @@ -6126,6 +6463,8 @@ mod tests { let response = client .list_contacts(authed_bearer_request(ListContactsRequest { account_id: "acct-contacts-list".to_string(), + page_size: 0, + page_token: String::new(), })) .await .expect("list_contacts succeeds") @@ -6239,6 +6578,8 @@ mod tests { let list_response = client .list_contacts(authed_bearer_request(ListContactsRequest { account_id: "acct-contacts-injected".to_string(), + page_size: 0, + page_token: String::new(), })) .await .expect("list_contacts succeeds") @@ -6350,7 +6691,10 @@ mod tests { .await .expect("Mail client connects"); let err = client - .list_folders(ListFoldersRequest {}) + .list_folders(ListFoldersRequest { + page_size: 0, + page_token: String::new(), + }) .await .expect_err("Mail must reject an unauthenticated call"); assert_eq!(err.code(), Code::Unauthenticated); @@ -6360,7 +6704,10 @@ mod tests { .await .expect("Filters client connects"); let err = client - .list_rules(ListRulesRequest {}) + .list_rules(ListRulesRequest { + page_size: 0, + page_token: String::new(), + }) .await .expect_err("Filters must reject an unauthenticated call"); assert_eq!(err.code(), Code::Unauthenticated); @@ -6385,8 +6732,8 @@ mod tests { .expect("Audit client connects"); let err = client .list_records(ListRecordsRequest { - limit: 10, - offset: 0, + page_size: 10, + page_token: String::new(), }) .await .expect_err("Audit must reject an unauthenticated call"); @@ -6402,6 +6749,8 @@ mod tests { calendar_id: "cal-1".to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(0)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(i64::MAX)), + page_size: 0, + page_token: String::new(), }) .await .expect_err("Calendar must reject an unauthenticated call"); @@ -6414,6 +6763,8 @@ mod tests { let err = client .list_contacts(ListContactsRequest { account_id: "acct-1".to_string(), + page_size: 0, + page_token: String::new(), }) .await .expect_err("Contacts must reject an unauthenticated call"); diff --git a/crates/nunciod/src/lib.rs b/crates/nunciod/src/lib.rs index 0f3ef2d..93fcc13 100644 --- a/crates/nunciod/src/lib.rs +++ b/crates/nunciod/src/lib.rs @@ -9,6 +9,7 @@ pub mod lifecycle; pub mod lock; pub mod orchestrator; pub mod outbox; +pub mod pagination; pub mod send; pub mod sync; pub use orchestrator::SelfHealingSyncOrchestrator; diff --git a/crates/nunciod/src/pagination.rs b/crates/nunciod/src/pagination.rs new file mode 100644 index 0000000..958341a --- /dev/null +++ b/crates/nunciod/src/pagination.rs @@ -0,0 +1,168 @@ +//! Shared keyset ("cursor") pagination for the daemon's list/search RPCs. +//! +//! Every `nuncio.v1` list/search RPC pages results with the same convention +//! (AIP-158): a request carries `page_size` + `page_token` and a response +//! carries `next_page_token`. The `page_token` is an OPAQUE cursor, never a +//! numeric offset: it encodes the stable sort key (and id tiebreaker) of the +//! last row of the previous page so the next page resumes strictly AFTER that +//! row. Keyset paging is stable under concurrent inserts -- rows are never +//! skipped or returned twice as the underlying table changes between page +//! fetches, which offset paging cannot guarantee. +//! +//! The wire encoding is deliberately private to the daemon: it is a +//! hex-encoded JSON array of typed [`CursorField`]s. Clients must treat the +//! token as opaque and echo it back verbatim; a token this module cannot +//! decode is rejected with a typed [`ErrorReason::ValidationFailed`] error +//! (mapped to `InvalidArgument`), never silently ignored or treated as the +//! first page. + +use nuncio_proto::errors; +use nuncio_proto::v1::ErrorReason; +use serde::{Deserialize, Serialize}; +use tonic::Status; + +/// Page size used when a request leaves `page_size` at 0 ("server default"). +pub const DEFAULT_PAGE_SIZE: u32 = 50; + +/// Hard upper bound on a single page, so a client cannot force the daemon to +/// materialize an unbounded result set in one call. +pub const MAX_PAGE_SIZE: u32 = 500; + +/// Resolves a requested `page_size` into a concrete row count: 0 selects +/// [`DEFAULT_PAGE_SIZE`], and any value is clamped to [`MAX_PAGE_SIZE`]. +#[must_use] +pub fn clamp_page_size(requested: u32) -> usize { + let n = if requested == 0 { + DEFAULT_PAGE_SIZE + } else { + requested.min(MAX_PAGE_SIZE) + }; + n as usize +} + +/// One typed component of an opaque keyset cursor. A cursor is an ordered +/// list of these, matching a list's stable sort key followed by its id +/// tiebreaker (e.g. `[I(received_at), S(id)]` for messages). +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub enum CursorField { + /// A signed integer sort key (e.g. a unix timestamp or a priority). + I(i64), + /// An unsigned integer sort key (e.g. an audit sequence). + U(u64), + /// A floating-point sort key (e.g. an FTS5 relevance rank). + F(f64), + /// A textual sort key or id (e.g. a message id or display name). + S(String), +} + +/// Encodes a cursor's component list into an opaque `page_token`. +#[must_use] +pub fn encode_cursor(fields: &[CursorField]) -> String { + // The component list is a fixed, small set of primitives, so + // serialization cannot realistically fail; fall back to an empty payload + // rather than panicking, keeping this crate free of `unwrap`/`expect`. + let json = serde_json::to_vec(fields).unwrap_or_default(); + hex::encode(json) +} + +fn malformed_token() -> Status { + errors::status( + ErrorReason::ValidationFailed, + "malformed page_token: not a valid pagination cursor", + ) +} + +/// Decodes an opaque `page_token` back into its component list, rejecting a +/// token that is not a hex-encoded cursor JSON payload with a typed +/// `VALIDATION_FAILED` error. +pub fn decode_cursor(token: &str) -> Result, Status> { + let bytes = hex::decode(token).map_err(|_| malformed_token())?; + serde_json::from_slice(&bytes).map_err(|_| malformed_token()) +} + +/// Decodes an optional `page_token` into a strongly-typed cursor value. +/// +/// An empty token means "first page" and yields `Ok(None)`. A non-empty +/// token is decoded and handed to `shape`, which pattern-matches the expected +/// component layout for a specific list and returns `Some(value)`; a token +/// that decodes but does not match the expected layout (e.g. a cursor minted +/// for a different list) is rejected with the same typed `VALIDATION_FAILED` +/// error as an undecodable token, so a caller can never resume from a cursor +/// shaped for the wrong query. +pub fn parse_token( + token: &str, + shape: impl FnOnce(&[CursorField]) -> Option, +) -> Result, Status> { + if token.is_empty() { + return Ok(None); + } + let fields = decode_cursor(token)?; + shape(&fields).map(Some).ok_or_else(malformed_token) +} + +#[cfg(test)] +mod tests { + use super::*; + use tonic::Code; + + #[test] + fn clamp_page_size_applies_default_and_max() { + assert_eq!(clamp_page_size(0), DEFAULT_PAGE_SIZE as usize); + assert_eq!(clamp_page_size(10), 10); + assert_eq!(clamp_page_size(MAX_PAGE_SIZE + 1), MAX_PAGE_SIZE as usize); + assert_eq!(clamp_page_size(u32::MAX), MAX_PAGE_SIZE as usize); + } + + #[test] + fn cursor_round_trips_through_encode_decode() { + let fields = vec![ + CursorField::I(1_700_000_000), + CursorField::S("msg-9".into()), + ]; + let token = encode_cursor(&fields); + assert!(!token.is_empty()); + assert_eq!(decode_cursor(&token).unwrap(), fields); + } + + #[test] + fn empty_token_parses_as_first_page() { + let parsed = parse_token("", |f| match f { + [CursorField::I(n)] => Some(*n), + _ => None, + }) + .unwrap(); + assert_eq!(parsed, None); + } + + #[test] + fn malformed_token_is_validation_failed() { + // Not valid hex. + let err = decode_cursor("zzz not-hex").unwrap_err(); + assert_eq!(err.code(), Code::InvalidArgument); + assert_eq!( + errors::error_reason(&err), + Some(ErrorReason::ValidationFailed) + ); + + // Valid hex, but not a cursor JSON payload. + let err = decode_cursor(&hex::encode(b"not json")).unwrap_err(); + assert_eq!(err.code(), Code::InvalidArgument); + } + + #[test] + fn wrong_shape_token_is_rejected() { + // A cursor with a single int component handed to a shape expecting + // (int, string) must be rejected, not silently accepted. + let token = encode_cursor(&[CursorField::I(1)]); + let err = parse_token(&token, |f| match f { + [CursorField::I(n), CursorField::S(s)] => Some((*n, s.clone())), + _ => None, + }) + .unwrap_err(); + assert_eq!(err.code(), Code::InvalidArgument); + assert_eq!( + errors::error_reason(&err), + Some(ErrorReason::ValidationFailed) + ); + } +} diff --git a/crates/nunciod/tests/caldav_production_e2e_test.rs b/crates/nunciod/tests/caldav_production_e2e_test.rs index 51a9d1f..6a210f3 100644 --- a/crates/nunciod/tests/caldav_production_e2e_test.rs +++ b/crates/nunciod/tests/caldav_production_e2e_test.rs @@ -198,6 +198,8 @@ async fn caldav_production_sync_builds_real_client_and_persists_events() { calendar_id: CALENDAR_ID.to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_700_000_000)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_800_000_000)), + page_size: 0, + page_token: String::new(), }) .await .expect("list_events succeeds") @@ -234,6 +236,8 @@ async fn caldav_production_sync_builds_real_client_and_persists_events() { calendar_id: CALENDAR_ID.to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(0)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(i64::MAX)), + page_size: 0, + page_token: String::new(), }) .await .expect_err("missing bearer token must be rejected"); diff --git a/crates/nunciod/tests/calendar_e2e_test.rs b/crates/nunciod/tests/calendar_e2e_test.rs index 3c3ea63..f9cf5c9 100644 --- a/crates/nunciod/tests/calendar_e2e_test.rs +++ b/crates/nunciod/tests/calendar_e2e_test.rs @@ -154,6 +154,8 @@ async fn calendar_sync_then_list_and_get_round_trip_over_authenticated_grpc() { calendar_id: CALENDAR_ID.to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_699_999_000)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(1_700_020_000)), + page_size: 0, + page_token: String::new(), }) .await .expect("list_events succeeds") @@ -189,6 +191,8 @@ async fn calendar_sync_then_list_and_get_round_trip_over_authenticated_grpc() { calendar_id: CALENDAR_ID.to_string(), start_window: Some(nuncio_proto::time::timestamp_from_unix_secs(0)), end_window: Some(nuncio_proto::time::timestamp_from_unix_secs(i64::MAX)), + page_size: 0, + page_token: String::new(), }) .await .expect_err("missing bearer token must be rejected"); diff --git a/crates/nunciod/tests/contacts_e2e_test.rs b/crates/nunciod/tests/contacts_e2e_test.rs index f231f39..8cc1cf5 100644 --- a/crates/nunciod/tests/contacts_e2e_test.rs +++ b/crates/nunciod/tests/contacts_e2e_test.rs @@ -146,6 +146,8 @@ async fn contacts_sync_create_and_list_round_trip_over_authenticated_grpc() { let list_response = contacts_client .list_contacts(ListContactsRequest { account_id: ACCOUNT_ID.to_string(), + page_size: 0, + page_token: String::new(), }) .await .expect("list_contacts succeeds") @@ -201,6 +203,8 @@ async fn contacts_sync_create_and_list_round_trip_over_authenticated_grpc() { let second_list = second_client .list_contacts(ListContactsRequest { account_id: ACCOUNT_ID.to_string(), + page_size: 0, + page_token: String::new(), }) .await .expect("list_contacts succeeds on second connection") @@ -231,6 +235,8 @@ async fn contacts_sync_create_and_list_round_trip_over_authenticated_grpc() { let err = unauthenticated_client .list_contacts(ListContactsRequest { account_id: ACCOUNT_ID.to_string(), + page_size: 0, + page_token: String::new(), }) .await .expect_err("missing bearer token must be rejected"); diff --git a/crates/nunciod/tests/spine_e2e_test.rs b/crates/nunciod/tests/spine_e2e_test.rs index 42f786c..d7ad21f 100644 --- a/crates/nunciod/tests/spine_e2e_test.rs +++ b/crates/nunciod/tests/spine_e2e_test.rs @@ -226,7 +226,8 @@ async fn full_mail_spine_round_trips_over_authenticated_grpc_with_mocked_protoco let list_messages_response = mail_client .list_messages(ListMessagesRequest { folder_id: "inbox".to_string(), - limit: 0, + page_size: 0, + page_token: String::new(), }) .await .expect("list_messages succeeds")