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

feat(proto): keyset pagination for all list/search RPCs (AIP-158) #350

Merged
KofTwentyTwo merged 1 commits from story/ws-a3-pagination into dev 2026-07-31 21:42:25 +00:00
Conversation 0 Commits 1 Files Changed 13
+1536 -376
13 changed files with 1536 additions and 376 deletions
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/nuncio-cli/src/args.rs
+10 -18
View File
@@ -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 }
}
}
);
crates/nuncio-cli/src/runner.rs
+357 -243
View File
@@ -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<serde_json::Value> =
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<serde_json::Value> =
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<serde_json::Value> = 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<serde_json::Value> = 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<serde_json::Value> = 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<serde_json::Value> = 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<serde_json::Value> =
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<serde_json::Value> =
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<serde_json::Value> =
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<serde_json::Value> =
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<serde_json::Value> =
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<serde_json::Value> =
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<serde_json::Value> =
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<serde_json::Value> =
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<Vec<nuncio_proto::v1::Contact>, 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<serde_json::Value> =
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<serde_json::Value> =
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,
crates/nuncio-proto/proto/nuncio/v1/descriptor.bin
BIN
View File
Binary file not shown.
crates/nuncio-proto/proto/nuncio/v1/nuncio.proto
+110 -29
View File
@@ -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.
crates/nuncio-store/src/db.rs
+327
View File
@@ -1291,6 +1291,102 @@ impl DatabaseEngine {
.collect::<Result<Vec<_>, 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<nuncio_core::model::Email>, 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<String>,
Option<String>,
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::<Result<Vec<_>, 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<nuncio_contacts::Contact>, 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<nuncio_contacts::Contact> =
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<String>,
page_size: usize,
) -> Result<(Vec<nuncio_core::model::Folder>, Option<String>), 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<nuncio_core::model::Folder> = 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<nuncio_filter::FilterRule>, 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<u64>,
page_size: usize,
) -> Result<(Vec<nuncio_core::WormAuditRecord>, Option<u64>), 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<nuncio_core::WormAuditRecord> = 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> {
crates/nuncio-store/src/search.rs
+111
View File
@@ -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<SearchHit>, 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, '<b>', '</b>', '...', 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<SearchHit> = 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<Vec<SearchHit>, 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();
crates/nunciod/src/grpc.rs
+436 -85
View File
@@ -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<ListFoldersRequest>,
request: Request<ListFoldersRequest>,
) -> Result<Response<ListFoldersResponse>, 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<SearchMessagesRequest>,
) -> Result<Response<SearchMessagesResponse>, 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<ListRulesRequest>,
request: Request<ListRulesRequest>,
) -> Result<Response<ListRulesResponse>, 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<ListRecordsRequest>,
) -> Result<Response<ListRecordsResponse>, 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");
crates/nunciod/src/lib.rs
+1
View File
@@ -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;
crates/nunciod/src/pagination.rs
+168
View File
@@ -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<Vec<CursorField>, 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<T>(
token: &str,
shape: impl FnOnce(&[CursorField]) -> Option<T>,
) -> Result<Option<T>, 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)
);
}
}
crates/nunciod/tests/caldav_production_e2e_test.rs
+4
View File
@@ -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");
crates/nunciod/tests/calendar_e2e_test.rs
+4
View File
@@ -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");
crates/nunciod/tests/contacts_e2e_test.rs
+6
View File
@@ -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");
crates/nunciod/tests/spine_e2e_test.rs
+2 -1
View File
@@ -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")