feat(grpc): mail read path over authenticated gRPC (list/read/mark/search)
Add an authenticated Mail service (ListFolders, ListMessages, GetMessage, MarkRead, SearchMessages), mounted with the bearer interceptor like System/Accounts. Backed by DatabaseEngine read methods + a new set_message_read; GetMessage returns the decrypted body; MarkRead persists then emits MessageFlagsChanged (so it streams via Subscribe). Wire nuncio-cli mail list/read/mark/search + folder list off the ephemeral runner onto the daemon over gRPC, with honest errors when the daemon is unreachable. Offline E2E round-trips all RPCs and proves mark-read persistence + streaming. Refs #159
This commit is contained in:
8 files changed
+1372
-145
No files matched your search
@@ -20,6 +20,7 @@ nuncio-proto = { path = "../nuncio-proto" }
|
||||
|
||||
tokio.workspace = true
|
||||
sqlx.workspace = true
|
||||
tonic.workspace = true
|
||||
clap.workspace = true
|
||||
serde.workspace = true
|
||||
serde_json.workspace = true
|
||||
@@ -30,7 +31,6 @@ chrono.workspace = true
|
||||
hex.workspace = true
|
||||
|
||||
[dev-dependencies]
|
||||
tonic.workspace = true
|
||||
tokio-stream = { workspace = true, features = ["net"] }
|
||||
|
||||
|
||||
@@ -290,6 +290,18 @@ pub enum MailSubcommand {
|
||||
#[arg(short, long, help = "Search query string")]
|
||||
query: String,
|
||||
},
|
||||
/// Mark a message read or unread.
|
||||
Mark {
|
||||
/// Unique message identifier.
|
||||
#[arg(short, long, help = "Unique message identifier")]
|
||||
id: String,
|
||||
/// Mark the message as read.
|
||||
#[arg(long, help = "Mark the message as read", conflicts_with = "unread")]
|
||||
read: bool,
|
||||
/// Mark the message as unread.
|
||||
#[arg(long, help = "Mark the message as unread", conflicts_with = "read")]
|
||||
unread: bool,
|
||||
},
|
||||
}
|
||||
|
||||
/// Folder subcommands (`nuncio folder <verb>`).
|
||||
@@ -447,6 +459,38 @@ mod tests {
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
let cli_mark_read =
|
||||
Cli::parse_from(["nuncio", "mail", "mark", "--id", "msg-123", "--read"]);
|
||||
assert_eq!(
|
||||
cli_mark_read.command,
|
||||
Commands::Mail {
|
||||
action: MailSubcommand::Mark {
|
||||
id: "msg-123".to_string(),
|
||||
read: true,
|
||||
unread: false,
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
let cli_mark_unread =
|
||||
Cli::parse_from(["nuncio", "mail", "mark", "--id", "msg-123", "--unread"]);
|
||||
assert_eq!(
|
||||
cli_mark_unread.command,
|
||||
Commands::Mail {
|
||||
action: MailSubcommand::Mark {
|
||||
id: "msg-123".to_string(),
|
||||
read: false,
|
||||
unread: true,
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
// `--read` and `--unread` are mutually exclusive at the Clap level.
|
||||
let conflict = Cli::try_parse_from([
|
||||
"nuncio", "mail", "mark", "--id", "msg-123", "--read", "--unread",
|
||||
]);
|
||||
assert!(conflict.is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
+490
-93
@@ -39,6 +39,24 @@ fn map_tls_mode_to_proto(mode: nuncio_core::TlsMode) -> nuncio_proto::v1::TlsMod
|
||||
}
|
||||
}
|
||||
|
||||
/// Renders a `nuncio.v1.Message` (as returned by the daemon's `Mail` gRPC
|
||||
/// service) into the JSON shape used by `mail list`/`mail read`'s
|
||||
/// `--json` output.
|
||||
fn message_proto_to_json(message: &nuncio_proto::v1::Message) -> serde_json::Value {
|
||||
json!({
|
||||
"id": message.id,
|
||||
"account_id": message.account_id,
|
||||
"folder_id": message.folder_id,
|
||||
"subject": message.subject,
|
||||
"sender": message.sender,
|
||||
"recipient": message.recipient,
|
||||
"received_at": message.received_at,
|
||||
"read": message.read,
|
||||
"body_plain": message.body_plain,
|
||||
"body_html": message.body_html,
|
||||
})
|
||||
}
|
||||
|
||||
/// Errors emitted by the CLI headless runner.
|
||||
#[derive(Error, Debug)]
|
||||
pub enum RunnerError {
|
||||
@@ -196,6 +214,9 @@ impl HeadlessRunner {
|
||||
self.handle_send_email(to, subject, body, json_mode).await
|
||||
}
|
||||
MailSubcommand::Search { query } => self.handle_search(query, json_mode).await,
|
||||
MailSubcommand::Mark { id, read, unread } => {
|
||||
self.handle_mark_read(id, *read, *unread, json_mode).await
|
||||
}
|
||||
},
|
||||
Commands::Banner => {
|
||||
crate::output::print_splash_banner();
|
||||
@@ -714,21 +735,41 @@ impl HeadlessRunner {
|
||||
}
|
||||
}
|
||||
|
||||
/// `mail list`: a real thin gRPC client of the running `nunciod`
|
||||
/// daemon's `nuncio.v1.Mail` API (backlog story 1.C.4, GH #159). Lists
|
||||
/// messages in `folder`, newest first, from the daemon's real,
|
||||
/// persistent store -- NOT this runner's own ephemeral local `db`, which
|
||||
/// is thrown away when this CLI process exits.
|
||||
async fn handle_list_folder(&self, folder: &str, json_mode: bool) -> String {
|
||||
let count: i64 = sqlx::query_as("SELECT COUNT(*) FROM messages WHERE folder_id = ?")
|
||||
.bind(folder)
|
||||
.fetch_one(self.db.pool())
|
||||
.await
|
||||
.map(|r: (i64,)| r.0)
|
||||
.unwrap_or(0);
|
||||
let mut client = match self.connect_mail_client().await {
|
||||
Ok(client) => client,
|
||||
Err(e) => return Self::render_error(&e, json_mode),
|
||||
};
|
||||
|
||||
if json_mode {
|
||||
format_json(&json!({
|
||||
"folder": folder,
|
||||
"total_messages": count
|
||||
}))
|
||||
} else {
|
||||
format!("Folder '{}': {} total messages", folder, count)
|
||||
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())
|
||||
}
|
||||
}
|
||||
Err(status) => Self::render_error(
|
||||
&format!("nunciod daemon rejected list_messages: {status}"),
|
||||
json_mode,
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -751,49 +792,176 @@ impl HeadlessRunner {
|
||||
}
|
||||
}
|
||||
|
||||
/// `mail search`: a real thin gRPC client of the running `nunciod`
|
||||
/// daemon's `nuncio.v1.Mail` API (backlog story 1.C.4, GH #159). Runs a
|
||||
/// full-text (FTS5) search over the daemon's real, persistent store.
|
||||
async fn handle_search(&self, query: &str, json_mode: bool) -> String {
|
||||
if json_mode {
|
||||
format_json(&json!({
|
||||
"query": query,
|
||||
"results": []
|
||||
}))
|
||||
} else {
|
||||
format!("Search complete for '{}' (0 matches)", query)
|
||||
}
|
||||
}
|
||||
let mut client = match self.connect_mail_client().await {
|
||||
Ok(client) => client,
|
||||
Err(e) => return Self::render_error(&e, json_mode),
|
||||
};
|
||||
|
||||
async fn handle_folders_list(&self, json_mode: bool) -> String {
|
||||
let folders = self.db.list_folders().await.unwrap_or_default();
|
||||
if json_mode {
|
||||
format_json(&json!({
|
||||
"folders": folders
|
||||
}))
|
||||
} else {
|
||||
format!("Available Mailbox Folders: {} folders found", folders.len())
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_read_message(&self, id: &str, json_mode: bool) -> String {
|
||||
match self.db.get_message(id).await {
|
||||
Ok(msg) => {
|
||||
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!({
|
||||
"message": msg
|
||||
"query": query,
|
||||
"results": hits_json
|
||||
}))
|
||||
} else {
|
||||
format!("Search complete for '{}' ({} matches)", query, hits.len())
|
||||
}
|
||||
}
|
||||
Err(status) => Self::render_error(
|
||||
&format!("nunciod daemon rejected search_messages: {status}"),
|
||||
json_mode,
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
/// `folder list`: a real thin gRPC client of the running `nunciod`
|
||||
/// daemon's `nuncio.v1.Mail` API (backlog story 1.C.4, GH #159).
|
||||
async fn handle_folders_list(&self, json_mode: bool) -> String {
|
||||
let mut client = match self.connect_mail_client().await {
|
||||
Ok(client) => client,
|
||||
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())
|
||||
}
|
||||
}
|
||||
Err(status) => Self::render_error(
|
||||
&format!("nunciod daemon rejected list_folders: {status}"),
|
||||
json_mode,
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
/// `mail read`: a real thin gRPC client of the running `nunciod`
|
||||
/// daemon's `nuncio.v1.Mail` API (backlog story 1.C.4, GH #159). Returns
|
||||
/// the full message, including its (decrypted) body, from the daemon's
|
||||
/// real, persistent store.
|
||||
async fn handle_read_message(&self, id: &str, json_mode: bool) -> String {
|
||||
let mut client = match self.connect_mail_client().await {
|
||||
Ok(client) => client,
|
||||
Err(e) => return Self::render_error(&e, json_mode),
|
||||
};
|
||||
|
||||
match client
|
||||
.get_message(nuncio_proto::v1::GetMessageRequest {
|
||||
message_id: id.to_string(),
|
||||
})
|
||||
.await
|
||||
{
|
||||
Ok(response) => match response.into_inner().message {
|
||||
Some(msg) => {
|
||||
if json_mode {
|
||||
format_json(&json!({ "message": message_proto_to_json(&msg) }))
|
||||
} else {
|
||||
format!(
|
||||
"Message {}: Subject: '{}', From: {}, Date: {}",
|
||||
msg.id, msg.subject, msg.sender, msg.received_at
|
||||
)
|
||||
}
|
||||
}
|
||||
None => Self::render_error(&format!("message '{}' not found", id), json_mode),
|
||||
},
|
||||
Err(status) if status.code() == tonic::Code::NotFound => {
|
||||
Self::render_error(&format!("message '{}' not found", id), json_mode)
|
||||
}
|
||||
Err(status) => Self::render_error(
|
||||
&format!("nunciod daemon rejected get_message: {status}"),
|
||||
json_mode,
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
/// `mail mark`: a real thin gRPC client of the running `nunciod`
|
||||
/// daemon's `nuncio.v1.Mail` API (backlog story 1.C.4, GH #159). Exactly
|
||||
/// one of `read`/`unread` must be set (enforced both by Clap's
|
||||
/// `conflicts_with` and this runtime check, so a caller invoking this
|
||||
/// programmatically without going through Clap still cannot request an
|
||||
/// ambiguous state).
|
||||
async fn handle_mark_read(
|
||||
&self,
|
||||
id: &str,
|
||||
read: bool,
|
||||
unread: bool,
|
||||
json_mode: bool,
|
||||
) -> String {
|
||||
if read == unread {
|
||||
return Self::render_error(
|
||||
"exactly one of --read or --unread must be specified",
|
||||
json_mode,
|
||||
);
|
||||
}
|
||||
|
||||
let mut client = match self.connect_mail_client().await {
|
||||
Ok(client) => client,
|
||||
Err(e) => return Self::render_error(&e, json_mode),
|
||||
};
|
||||
|
||||
match client
|
||||
.mark_read(nuncio_proto::v1::MarkReadRequest {
|
||||
message_id: id.to_string(),
|
||||
read,
|
||||
})
|
||||
.await
|
||||
{
|
||||
Ok(_) => {
|
||||
if json_mode {
|
||||
format_json(&json!({ "status": "marked", "id": id, "read": read }))
|
||||
} else {
|
||||
format!(
|
||||
"Message {}: Subject: '{}', From: {}, Date: {}",
|
||||
msg.id, msg.subject, msg.sender, msg.received_at
|
||||
"Message '{}' marked as {}",
|
||||
id,
|
||||
if read { "read" } else { "unread" }
|
||||
)
|
||||
}
|
||||
}
|
||||
Err(_) => {
|
||||
if json_mode {
|
||||
format_json_error(&format!("message '{}' not found", id))
|
||||
} else {
|
||||
format!("Error: message '{}' not found", id)
|
||||
}
|
||||
Err(status) if status.code() == tonic::Code::NotFound => {
|
||||
Self::render_error(&format!("message '{}' not found", id), json_mode)
|
||||
}
|
||||
Err(status) => Self::render_error(
|
||||
&format!("nunciod daemon rejected mark_read: {status}"),
|
||||
json_mode,
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -951,6 +1119,23 @@ impl HeadlessRunner {
|
||||
.map_err(|e| format!("nunciod daemon unreachable at {}: {e}", self.grpc_addr))
|
||||
}
|
||||
|
||||
/// Resolves the gRPC bearer token from the injected vault and dials the
|
||||
/// `nuncio.v1.Mail` service at `self.grpc_addr` (backlog story 1.C.4,
|
||||
/// GH #159), shared by every `mail`/`folder` read-path handler above.
|
||||
async fn connect_mail_client(
|
||||
&self,
|
||||
) -> Result<nuncio_proto::client::AuthenticatedMailClient, String> {
|
||||
let token_bytes = self
|
||||
.secrets
|
||||
.get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32)
|
||||
.map_err(|e| format!("failed to read gRPC bearer token from vault: {e}"))?;
|
||||
let token = hex::encode(token_bytes);
|
||||
|
||||
nuncio_proto::client::connect_mail(&self.grpc_addr, &token)
|
||||
.await
|
||||
.map_err(|e| format!("nunciod daemon unreachable at {}: {e}", self.grpc_addr))
|
||||
}
|
||||
|
||||
/// Formats an error message consistently for both JSON and human-text
|
||||
/// output modes.
|
||||
fn render_error(message: &str, json_mode: bool) -> String {
|
||||
@@ -1138,7 +1323,17 @@ mod tests {
|
||||
.await;
|
||||
assert!(acct_show.contains("Account 'missing' not found"));
|
||||
|
||||
// Mail Noun Commands
|
||||
// Mail Noun Commands: `Sync` and `Send` do not talk to the daemon
|
||||
// (yet), so they are safe to exercise against this ephemeral local
|
||||
// runner. `List`/`Read`/`Search`/`Mark` (and `Folder::List`) are now
|
||||
// real gRPC clients of the `nunciod` daemon's `Mail` API (backlog
|
||||
// story 1.C.4, GH #159) -- exactly like `Account::Add`/`List` and
|
||||
// `System::Status` above, they must never run against this
|
||||
// `ephemeral()`-constructed runner's production `SecretManager` or
|
||||
// its (unreachable in CI) default gRPC address. They are exercised
|
||||
// separately below via `ephemeral_with` + `SecretManager::mock()`
|
||||
// against a live stub `Mail` gRPC server
|
||||
// (`mail_rpcs_round_trip_over_grpc_to_a_stub_daemon`).
|
||||
let mail_sync = runner
|
||||
.execute_command(
|
||||
&Commands::Mail {
|
||||
@@ -1149,30 +1344,6 @@ mod tests {
|
||||
.await;
|
||||
assert!(mail_sync.contains(r#""status":"sync_started""#));
|
||||
|
||||
let mail_list = runner
|
||||
.execute_command(
|
||||
&Commands::Mail {
|
||||
action: MailSubcommand::List {
|
||||
folder: "INBOX".to_string(),
|
||||
},
|
||||
},
|
||||
false,
|
||||
)
|
||||
.await;
|
||||
assert!(mail_list.contains("INBOX"));
|
||||
|
||||
let mail_read_err = runner
|
||||
.execute_command(
|
||||
&Commands::Mail {
|
||||
action: MailSubcommand::Read {
|
||||
id: "missing".to_string(),
|
||||
},
|
||||
},
|
||||
true,
|
||||
)
|
||||
.await;
|
||||
assert!(mail_read_err.contains(r#""status":"error""#));
|
||||
|
||||
let mail_send = runner
|
||||
.execute_command(
|
||||
&Commands::Mail {
|
||||
@@ -1187,29 +1358,6 @@ mod tests {
|
||||
.await;
|
||||
assert!(mail_send.contains("alice@nuncio.mx"));
|
||||
|
||||
let mail_search = runner
|
||||
.execute_command(
|
||||
&Commands::Mail {
|
||||
action: MailSubcommand::Search {
|
||||
query: "roadmap".to_string(),
|
||||
},
|
||||
},
|
||||
true,
|
||||
)
|
||||
.await;
|
||||
assert!(mail_search.contains(r#""query":"roadmap""#));
|
||||
|
||||
// Folder Noun Commands
|
||||
let folder_list = runner
|
||||
.execute_command(
|
||||
&Commands::Folder {
|
||||
action: FolderSubcommand::List,
|
||||
},
|
||||
true,
|
||||
)
|
||||
.await;
|
||||
assert!(folder_list.contains(r#""folders":[]"#));
|
||||
|
||||
// Cal Noun Commands
|
||||
let cal_list = runner
|
||||
.execute_command(
|
||||
@@ -1521,6 +1669,255 @@ mod tests {
|
||||
assert!(list_out_text.contains("1 account(s) registered"));
|
||||
}
|
||||
|
||||
/// Reference-client proof for backlog story 1.C.4 (GH #159): boots a
|
||||
/// stub `nuncio.v1.Mail` gRPC server (mirroring the `Accounts` stub
|
||||
/// pattern above) and drives the real `HeadlessRunner`'s `mail
|
||||
/// list`/`mail read`/`mail mark`/`mail search`/`folder list` gRPC client
|
||||
/// paths against it.
|
||||
///
|
||||
/// Real persistence (`mark_read` actually flipping the flag in the
|
||||
/// daemon's store, `list`/`get` returning REAL synced data,
|
||||
/// `MessageFlagsChanged` streaming) is proven by `nunciod`'s own
|
||||
/// `grpc::tests`; this test exists purely to prove the CLI's connect +
|
||||
/// call + JSON-format happy path, its honest not-found error handling,
|
||||
/// and that `mail mark` rejects an ambiguous `--read`/`--unread`
|
||||
/// combination before ever dialing the daemon.
|
||||
#[tokio::test]
|
||||
async fn mail_rpcs_round_trip_over_grpc_to_a_stub_daemon() {
|
||||
use nuncio_proto::v1::mail_server::{Mail as MailService, MailServer};
|
||||
use nuncio_proto::v1::{
|
||||
Folder as FolderProto, GetMessageRequest, GetMessageResponse, ListFoldersRequest,
|
||||
ListFoldersResponse, ListMessagesRequest, ListMessagesResponse, MarkReadRequest,
|
||||
MarkReadResponse, Message as MessageProto, MessageSearchHit, SearchMessagesRequest,
|
||||
SearchMessagesResponse,
|
||||
};
|
||||
use std::sync::Mutex;
|
||||
|
||||
fn stub_message() -> MessageProto {
|
||||
MessageProto {
|
||||
id: "msg-stub-1".to_string(),
|
||||
account_id: "acct-stub-1".to_string(),
|
||||
folder_id: "inbox".to_string(),
|
||||
subject: "Stub Subject".to_string(),
|
||||
sender: "alice@nuncio.mx".to_string(),
|
||||
recipient: "bob@nuncio.mx".to_string(),
|
||||
received_at: 1_700_000_000,
|
||||
read: false,
|
||||
body_plain: Some("Stub body text".to_string()),
|
||||
body_html: None,
|
||||
attachments: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Minimal test-only stub of `nuncio.v1.Mail`: records the last
|
||||
/// `MarkReadRequest` it received (so this test can assert on
|
||||
/// exactly what the CLI sent over the wire) and otherwise returns
|
||||
/// fixed responses.
|
||||
#[derive(Default)]
|
||||
struct StubMail {
|
||||
last_mark_read: Arc<Mutex<Option<MarkReadRequest>>>,
|
||||
}
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl MailService for StubMail {
|
||||
async fn list_folders(
|
||||
&self,
|
||||
_request: tonic::Request<ListFoldersRequest>,
|
||||
) -> Result<tonic::Response<ListFoldersResponse>, tonic::Status> {
|
||||
Ok(tonic::Response::new(ListFoldersResponse {
|
||||
folders: vec![FolderProto {
|
||||
id: "inbox".to_string(),
|
||||
name: "inbox".to_string(),
|
||||
total_messages: 3,
|
||||
unread_messages: 1,
|
||||
}],
|
||||
}))
|
||||
}
|
||||
|
||||
async fn list_messages(
|
||||
&self,
|
||||
request: tonic::Request<ListMessagesRequest>,
|
||||
) -> Result<tonic::Response<ListMessagesResponse>, tonic::Status> {
|
||||
let req = request.into_inner();
|
||||
let mut message = stub_message();
|
||||
message.folder_id = req.folder_id;
|
||||
Ok(tonic::Response::new(ListMessagesResponse {
|
||||
messages: vec![message],
|
||||
}))
|
||||
}
|
||||
|
||||
async fn get_message(
|
||||
&self,
|
||||
request: tonic::Request<GetMessageRequest>,
|
||||
) -> Result<tonic::Response<GetMessageResponse>, tonic::Status> {
|
||||
let req = request.into_inner();
|
||||
if req.message_id == "msg-stub-1" {
|
||||
Ok(tonic::Response::new(GetMessageResponse {
|
||||
message: Some(stub_message()),
|
||||
}))
|
||||
} else {
|
||||
Err(tonic::Status::not_found(format!(
|
||||
"message '{}' not found",
|
||||
req.message_id
|
||||
)))
|
||||
}
|
||||
}
|
||||
|
||||
async fn mark_read(
|
||||
&self,
|
||||
request: tonic::Request<MarkReadRequest>,
|
||||
) -> Result<tonic::Response<MarkReadResponse>, tonic::Status> {
|
||||
let req = request.into_inner();
|
||||
*self
|
||||
.last_mark_read
|
||||
.lock()
|
||||
.unwrap_or_else(|e| e.into_inner()) = Some(req);
|
||||
Ok(tonic::Response::new(MarkReadResponse {}))
|
||||
}
|
||||
|
||||
async fn search_messages(
|
||||
&self,
|
||||
request: tonic::Request<SearchMessagesRequest>,
|
||||
) -> Result<tonic::Response<SearchMessagesResponse>, tonic::Status> {
|
||||
let req = request.into_inner();
|
||||
Ok(tonic::Response::new(SearchMessagesResponse {
|
||||
hits: vec![MessageSearchHit {
|
||||
id: "msg-stub-1".to_string(),
|
||||
title: "Stub Subject".to_string(),
|
||||
snippet: format!("...{}...", req.query),
|
||||
}],
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
let stub = StubMail::default();
|
||||
let probe = stub.last_mark_read.clone();
|
||||
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
|
||||
.await
|
||||
.expect("bind ephemeral loopback port");
|
||||
let addr = listener.local_addr().expect("listener has local addr");
|
||||
tokio::spawn(async move {
|
||||
let _ = tonic::transport::Server::builder()
|
||||
.add_service(MailServer::new(stub))
|
||||
.serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener))
|
||||
.await;
|
||||
});
|
||||
|
||||
let runner =
|
||||
HeadlessRunner::ephemeral_with(Arc::new(SecretManager::mock()), addr.to_string())
|
||||
.await
|
||||
.expect("ephemeral runner initializes");
|
||||
|
||||
// `folder list`
|
||||
let folder_list = runner
|
||||
.execute_command(
|
||||
&Commands::Folder {
|
||||
action: FolderSubcommand::List,
|
||||
},
|
||||
true,
|
||||
)
|
||||
.await;
|
||||
assert!(folder_list.contains(r#""id":"inbox""#));
|
||||
assert!(folder_list.contains(r#""total_messages":3"#));
|
||||
|
||||
// `mail list`
|
||||
let mail_list = runner
|
||||
.execute_command(
|
||||
&Commands::Mail {
|
||||
action: MailSubcommand::List {
|
||||
folder: "inbox".to_string(),
|
||||
},
|
||||
},
|
||||
true,
|
||||
)
|
||||
.await;
|
||||
assert!(mail_list.contains("msg-stub-1"));
|
||||
assert!(mail_list.contains(r#""folder_id":"inbox""#));
|
||||
|
||||
// `mail read` happy path
|
||||
let mail_read = runner
|
||||
.execute_command(
|
||||
&Commands::Mail {
|
||||
action: MailSubcommand::Read {
|
||||
id: "msg-stub-1".to_string(),
|
||||
},
|
||||
},
|
||||
true,
|
||||
)
|
||||
.await;
|
||||
assert!(mail_read.contains("Stub Subject"));
|
||||
assert!(mail_read.contains("Stub body text"));
|
||||
|
||||
// `mail read` honest not-found error
|
||||
let mail_read_missing = runner
|
||||
.execute_command(
|
||||
&Commands::Mail {
|
||||
action: MailSubcommand::Read {
|
||||
id: "missing".to_string(),
|
||||
},
|
||||
},
|
||||
true,
|
||||
)
|
||||
.await;
|
||||
assert!(mail_read_missing.contains(r#""status":"error""#));
|
||||
assert!(mail_read_missing.contains("not found"));
|
||||
|
||||
// `mail search`
|
||||
let mail_search = runner
|
||||
.execute_command(
|
||||
&Commands::Mail {
|
||||
action: MailSubcommand::Search {
|
||||
query: "roadmap".to_string(),
|
||||
},
|
||||
},
|
||||
true,
|
||||
)
|
||||
.await;
|
||||
assert!(mail_search.contains(r#""query":"roadmap""#));
|
||||
assert!(mail_search.contains("msg-stub-1"));
|
||||
|
||||
// `mail mark --read`
|
||||
let mail_mark = runner
|
||||
.execute_command(
|
||||
&Commands::Mail {
|
||||
action: MailSubcommand::Mark {
|
||||
id: "msg-stub-1".to_string(),
|
||||
read: true,
|
||||
unread: false,
|
||||
},
|
||||
},
|
||||
true,
|
||||
)
|
||||
.await;
|
||||
assert!(mail_mark.contains(r#""status":"marked""#));
|
||||
assert!(mail_mark.contains(r#""read":true"#));
|
||||
|
||||
let recorded = probe
|
||||
.lock()
|
||||
.unwrap_or_else(|e| e.into_inner())
|
||||
.clone()
|
||||
.expect("stub daemon received a mark_read request");
|
||||
assert_eq!(recorded.message_id, "msg-stub-1");
|
||||
assert!(recorded.read);
|
||||
|
||||
// `mail mark` with neither --read nor --unread fails BEFORE dialing
|
||||
// the daemon.
|
||||
let mark_no_flag = runner
|
||||
.execute_command(
|
||||
&Commands::Mail {
|
||||
action: MailSubcommand::Mark {
|
||||
id: "msg-stub-1".to_string(),
|
||||
read: false,
|
||||
unread: false,
|
||||
},
|
||||
},
|
||||
true,
|
||||
)
|
||||
.await;
|
||||
assert!(mark_no_flag.contains("exactly one of --read or --unread"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn runner_error_display() {
|
||||
let err = RunnerError::InitFailed("failed to open database".to_string());
|
||||
|
||||
@@ -37,44 +37,12 @@ async fn system_test_cli_noun_verb_execution_matrix() {
|
||||
// separately below via `ephemeral_with` + `SecretManager::mock()`
|
||||
// rather than through this `ephemeral()`-constructed runner.
|
||||
|
||||
// 3. Folder list
|
||||
let out: String = runner
|
||||
.execute_command(
|
||||
&Commands::Folder {
|
||||
action: FolderSubcommand::List,
|
||||
},
|
||||
true,
|
||||
)
|
||||
.await;
|
||||
let json: Value = serde_json::from_str(&out).expect("valid json");
|
||||
assert_eq!(json["status"], "ok");
|
||||
|
||||
// 4. Mail list & search & read
|
||||
let out: String = runner
|
||||
.execute_command(
|
||||
&Commands::Mail {
|
||||
action: MailSubcommand::List {
|
||||
folder: "INBOX".to_string(),
|
||||
},
|
||||
},
|
||||
true,
|
||||
)
|
||||
.await;
|
||||
let json: Value = serde_json::from_str(&out).expect("valid json");
|
||||
assert_eq!(json["status"], "ok");
|
||||
|
||||
let out: String = runner
|
||||
.execute_command(
|
||||
&Commands::Mail {
|
||||
action: MailSubcommand::Search {
|
||||
query: "Architecture".to_string(),
|
||||
},
|
||||
},
|
||||
true,
|
||||
)
|
||||
.await;
|
||||
let json: Value = serde_json::from_str(&out).expect("valid json");
|
||||
assert_eq!(json["status"], "ok");
|
||||
// 3. Folder list & 4. Mail list/search/read/mark are real gRPC clients
|
||||
// of the `nunciod` daemon's `Mail` API (backlog story 1.C.4, GH #159),
|
||||
// so -- exactly like `system status` and `account add`/`list` above --
|
||||
// they are exercised separately below via `ephemeral_with` +
|
||||
// `SecretManager::mock()` rather than through this
|
||||
// `ephemeral()`-constructed runner.
|
||||
|
||||
// 5. Calendar list & sync
|
||||
let out: String = runner
|
||||
@@ -179,3 +147,94 @@ async fn account_add_and_list_report_honest_errors_when_daemon_unreachable() {
|
||||
// The password must never leak into an error message either.
|
||||
assert!(!add_out.contains("irrelevant-unreachable-daemon"));
|
||||
}
|
||||
|
||||
/// `folder list`, `mail list`, `mail read`, `mail search`, and `mail mark`
|
||||
/// are real gRPC clients of the `nunciod` daemon's `Mail` API (backlog
|
||||
/// story 1.C.4, GH #159). With no daemon reachable, all of them must report
|
||||
/// a clear, honest error rather than fabricating an empty result.
|
||||
#[tokio::test]
|
||||
async fn mail_and_folder_report_honest_errors_when_daemon_unreachable() {
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
|
||||
.await
|
||||
.expect("bind ephemeral loopback port");
|
||||
let addr = listener.local_addr().expect("listener has local addr");
|
||||
drop(listener); // free the port; nothing is listening on it now
|
||||
|
||||
let runner = HeadlessRunner::ephemeral_with(Arc::new(SecretManager::mock()), addr.to_string())
|
||||
.await
|
||||
.expect("runner init");
|
||||
|
||||
let assert_honest_error = |out: String| {
|
||||
let json: Value = serde_json::from_str(&out).expect("valid json");
|
||||
assert_eq!(json["status"], "error");
|
||||
assert!(json["error"]
|
||||
.as_str()
|
||||
.expect("error message present")
|
||||
.contains("unreachable"));
|
||||
};
|
||||
|
||||
assert_honest_error(
|
||||
runner
|
||||
.execute_command(
|
||||
&Commands::Folder {
|
||||
action: FolderSubcommand::List,
|
||||
},
|
||||
true,
|
||||
)
|
||||
.await,
|
||||
);
|
||||
|
||||
assert_honest_error(
|
||||
runner
|
||||
.execute_command(
|
||||
&Commands::Mail {
|
||||
action: MailSubcommand::List {
|
||||
folder: "INBOX".to_string(),
|
||||
},
|
||||
},
|
||||
true,
|
||||
)
|
||||
.await,
|
||||
);
|
||||
|
||||
assert_honest_error(
|
||||
runner
|
||||
.execute_command(
|
||||
&Commands::Mail {
|
||||
action: MailSubcommand::Read {
|
||||
id: "msg-1".to_string(),
|
||||
},
|
||||
},
|
||||
true,
|
||||
)
|
||||
.await,
|
||||
);
|
||||
|
||||
assert_honest_error(
|
||||
runner
|
||||
.execute_command(
|
||||
&Commands::Mail {
|
||||
action: MailSubcommand::Search {
|
||||
query: "Architecture".to_string(),
|
||||
},
|
||||
},
|
||||
true,
|
||||
)
|
||||
.await,
|
||||
);
|
||||
|
||||
assert_honest_error(
|
||||
runner
|
||||
.execute_command(
|
||||
&Commands::Mail {
|
||||
action: MailSubcommand::Mark {
|
||||
id: "msg-1".to_string(),
|
||||
read: true,
|
||||
unread: false,
|
||||
},
|
||||
},
|
||||
true,
|
||||
)
|
||||
.await,
|
||||
);
|
||||
}
|
||||
@@ -214,3 +214,145 @@ message ListAccountsRequest {}
|
||||
message ListAccountsResponse {
|
||||
repeated AccountConfig accounts = 1;
|
||||
}
|
||||
|
||||
// Mail exposes the READ path over the daemon's real, persistent message
|
||||
// store (the data backlog story 1.C.3 / GH #158 syncs in via
|
||||
// `DatabaseEngine::save_email`) -- backlog story 1.C.4, GH #159.
|
||||
//
|
||||
// SECURITY (GH #165): this service MUST be mounted behind the SAME
|
||||
// `BearerAuthInterceptor` that guards `System` and `Accounts` -- every gRPC
|
||||
// service `nunciod` exposes MUST be authenticated, with no exceptions
|
||||
// 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.
|
||||
rpc ListFolders(ListFoldersRequest) returns (ListFoldersResponse);
|
||||
// ListMessages returns messages in a single folder, newest first, capped
|
||||
// at `limit`.
|
||||
rpc ListMessages(ListMessagesRequest) returns (ListMessagesResponse);
|
||||
// GetMessage returns a single message by ID, including its (decrypted)
|
||||
// body.
|
||||
rpc GetMessage(GetMessageRequest) returns (GetMessageResponse);
|
||||
// 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.
|
||||
rpc SearchMessages(SearchMessagesRequest) returns (SearchMessagesResponse);
|
||||
}
|
||||
|
||||
// Mirrors `nuncio_core::model::Attachment`.
|
||||
message Attachment {
|
||||
// Attachment filename.
|
||||
string filename = 1;
|
||||
// MIME content type (e.g. "application/pdf").
|
||||
string mime_type = 2;
|
||||
// Raw attachment content buffer.
|
||||
bytes content = 3;
|
||||
}
|
||||
|
||||
// Mirrors `nuncio_core::model::Email` field-for-field. Named `Message`
|
||||
// (rather than `Email`) to avoid colliding with common generated-code
|
||||
// identifiers in hand-written Rust using this module.
|
||||
message Message {
|
||||
// Unique message identifier.
|
||||
string id = 1;
|
||||
// Account identifier owning the message.
|
||||
string account_id = 2;
|
||||
// Mailbox folder identifier (e.g. "inbox").
|
||||
string folder_id = 3;
|
||||
// Subject line.
|
||||
string subject = 4;
|
||||
// Sender address.
|
||||
string sender = 5;
|
||||
// Recipient address.
|
||||
string recipient = 6;
|
||||
// Unix timestamp of message arrival.
|
||||
int64 received_at = 7;
|
||||
// Read/unread flag status.
|
||||
bool read = 8;
|
||||
// Plaintext message body (already decrypted by the store).
|
||||
optional string body_plain = 9;
|
||||
// HTML message body (already decrypted by the store).
|
||||
optional string body_html = 10;
|
||||
// List of attached files.
|
||||
repeated Attachment attachments = 11;
|
||||
}
|
||||
|
||||
// Mirrors `nuncio_core::model::Folder`.
|
||||
message Folder {
|
||||
// Folder identifier.
|
||||
string id = 1;
|
||||
// Folder display name.
|
||||
string name = 2;
|
||||
// Total message count.
|
||||
uint64 total_messages = 3;
|
||||
// Unread message count.
|
||||
uint64 unread_messages = 4;
|
||||
}
|
||||
|
||||
// ListFoldersRequest currently carries no parameters.
|
||||
message ListFoldersRequest {}
|
||||
|
||||
// ListFoldersResponse.
|
||||
message ListFoldersResponse {
|
||||
repeated Folder folders = 1;
|
||||
}
|
||||
|
||||
// ListMessagesRequest. `limit` caps the number of messages returned
|
||||
// (newest first); a value of 0 means "use the server's default limit".
|
||||
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;
|
||||
}
|
||||
|
||||
// ListMessagesResponse.
|
||||
message ListMessagesResponse {
|
||||
repeated Message messages = 1;
|
||||
}
|
||||
|
||||
// GetMessageRequest.
|
||||
message GetMessageRequest {
|
||||
// Unique message identifier.
|
||||
string message_id = 1;
|
||||
}
|
||||
|
||||
// GetMessageResponse.
|
||||
message GetMessageResponse {
|
||||
Message message = 1;
|
||||
}
|
||||
|
||||
// MarkReadRequest.
|
||||
message MarkReadRequest {
|
||||
// Unique message identifier.
|
||||
string message_id = 1;
|
||||
// Desired read state.
|
||||
bool read = 2;
|
||||
}
|
||||
|
||||
// MarkReadResponse currently carries no parameters; a successful RPC
|
||||
// response (as opposed to a `Status` error) is confirmation enough.
|
||||
message MarkReadResponse {}
|
||||
|
||||
// SearchMessagesRequest.
|
||||
message SearchMessagesRequest {
|
||||
// Full-text search query string.
|
||||
string query = 1;
|
||||
}
|
||||
|
||||
// Mirrors `nuncio_store::search::SearchHit`.
|
||||
message MessageSearchHit {
|
||||
// Matching message ID.
|
||||
string id = 1;
|
||||
// Matching subject/title.
|
||||
string title = 2;
|
||||
// Text snippet matching the search term.
|
||||
string snippet = 3;
|
||||
}
|
||||
|
||||
// SearchMessagesResponse.
|
||||
message SearchMessagesResponse {
|
||||
repeated MessageSearchHit hits = 1;
|
||||
}
|
||||
@@ -10,6 +10,7 @@
|
||||
//! server-side interceptor in `nunciod::grpc`.
|
||||
|
||||
use crate::v1::accounts_client::AccountsClient;
|
||||
use crate::v1::mail_client::MailClient;
|
||||
use crate::v1::system_client::SystemClient;
|
||||
use crate::v1::{Event, SubscribeRequest};
|
||||
use thiserror::Error;
|
||||
@@ -89,6 +90,10 @@ pub type AuthenticatedSystemClient =
|
||||
pub type AuthenticatedAccountsClient =
|
||||
AccountsClient<InterceptedService<Channel, BearerTokenInterceptor>>;
|
||||
|
||||
/// The authenticated `nuncio.v1.Mail` client type returned by
|
||||
/// [`connect_mail`].
|
||||
pub type AuthenticatedMailClient = MailClient<InterceptedService<Channel, BearerTokenInterceptor>>;
|
||||
|
||||
/// Dials `addr` (a `host:port` pair, e.g. `127.0.0.1:9420`) over plain HTTP
|
||||
/// (the loopback gRPC transport is never TLS-wrapped; auth is via bearer
|
||||
/// token instead) and returns the connected [`Channel`], shared by every
|
||||
@@ -141,6 +146,24 @@ pub async fn connect_accounts(
|
||||
Ok(AccountsClient::with_interceptor(channel, interceptor))
|
||||
}
|
||||
|
||||
/// Dials the `nuncio.v1.Mail` gRPC endpoint at `addr` and returns a client
|
||||
/// that injects `authorization: Bearer <token>` metadata on every call
|
||||
/// (backlog story 1.C.4, GH #159).
|
||||
///
|
||||
/// `Mail` is guarded by the exact same `BearerAuthInterceptor` as `System`
|
||||
/// and `Accounts` on the server side (see `nunciod::grpc::serve_on_listener`,
|
||||
/// GH #165), so this shares [`BearerTokenInterceptor`] and [`dial`] with
|
||||
/// [`connect_system`] / [`connect_accounts`] rather than hand-rolling a third
|
||||
/// auth handshake.
|
||||
pub async fn connect_mail(
|
||||
addr: &str,
|
||||
token: &str,
|
||||
) -> Result<AuthenticatedMailClient, ConnectError> {
|
||||
let interceptor = BearerTokenInterceptor::new(token)?;
|
||||
let channel = dial(addr).await?;
|
||||
Ok(MailClient::with_interceptor(channel, interceptor))
|
||||
}
|
||||
|
||||
/// Opens the `nuncio.v1.System/Subscribe` server-streaming RPC on an already
|
||||
/// authenticated client (as returned by [`connect_system`]), reusing the
|
||||
/// bearer-token interceptor already attached to `client` rather than
|
||||
@@ -218,6 +241,23 @@ mod tests {
|
||||
assert!(err.to_string().contains("127.0.0.1:1"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn connect_mail_fails_closed_on_invalid_token() {
|
||||
let err = connect_mail("127.0.0.1:0", "tok\ntoken")
|
||||
.await
|
||||
.expect_err("invalid token must fail before dialing");
|
||||
assert!(matches!(err, ConnectError::InvalidToken(_)));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn connect_mail_reports_transport_error_when_daemon_unreachable() {
|
||||
let err = connect_mail("127.0.0.1:1", "abc123")
|
||||
.await
|
||||
.expect_err("connecting to an unreachable daemon must fail");
|
||||
assert!(matches!(err, ConnectError::Transport { .. }));
|
||||
assert!(err.to_string().contains("127.0.0.1:1"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn connect_system_reports_transport_error_when_daemon_unreachable() {
|
||||
// Port 0 alone is not connectable; using an explicit closed port on
|
||||
|
||||
@@ -814,6 +814,31 @@ impl DatabaseEngine {
|
||||
})
|
||||
}
|
||||
|
||||
/// Update a single message's read/unread flag in place (backlog story
|
||||
/// 1.C.4, GH #159).
|
||||
///
|
||||
/// Returns `DatabaseError::Query(sqlx::Error::RowNotFound)` if no message
|
||||
/// with `message_id` exists, so callers can distinguish "flag flipped"
|
||||
/// from "message never existed" rather than silently succeeding on a
|
||||
/// no-op update.
|
||||
pub async fn set_message_read(
|
||||
&self,
|
||||
message_id: &str,
|
||||
read: bool,
|
||||
) -> Result<(), DatabaseError> {
|
||||
let result = sqlx::query("UPDATE messages SET read_flag = ? WHERE id = ?")
|
||||
.bind(if read { 1i64 } else { 0i64 })
|
||||
.bind(message_id)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(DatabaseError::Query)?;
|
||||
|
||||
if result.rows_affected() == 0 {
|
||||
return Err(DatabaseError::Query(sqlx::Error::RowNotFound));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Query available folders with message counts.
|
||||
pub async fn list_folders(&self) -> Result<Vec<nuncio_core::model::Folder>, DatabaseError> {
|
||||
let rows: Vec<(String, i64, i64)> = sqlx::query_as(
|
||||
@@ -1548,6 +1573,59 @@ mod tests {
|
||||
assert_eq!(folders[0].unread_messages, 1);
|
||||
}
|
||||
|
||||
/// Backlog story 1.C.4 (GH #159): `set_message_read` flips the persisted
|
||||
/// `read_flag` in place (both directions), and reports
|
||||
/// `sqlx::Error::RowNotFound` for a message ID that was never saved,
|
||||
/// rather than silently succeeding on a no-op update.
|
||||
#[tokio::test]
|
||||
async fn set_message_read_flips_flag_and_reports_not_found_for_missing_message() {
|
||||
let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap();
|
||||
|
||||
let email = nuncio_core::model::Email {
|
||||
id: "msg-mark-1".to_string(),
|
||||
account_id: "acct-1".to_string(),
|
||||
folder_id: "INBOX".to_string(),
|
||||
subject: "Mark Read Test".to_string(),
|
||||
sender: "alice@nuncio.mx".to_string(),
|
||||
recipient: "bob@nuncio.mx".to_string(),
|
||||
received_at: 1700000000,
|
||||
read: false,
|
||||
body_plain: None,
|
||||
body_html: None,
|
||||
attachments: Vec::new(),
|
||||
};
|
||||
engine.save_email(&email).await.expect("save email");
|
||||
|
||||
engine
|
||||
.set_message_read("msg-mark-1", true)
|
||||
.await
|
||||
.expect("mark read succeeds");
|
||||
let fetched = engine
|
||||
.get_message("msg-mark-1")
|
||||
.await
|
||||
.expect("get message succeeds");
|
||||
assert!(fetched.read);
|
||||
|
||||
engine
|
||||
.set_message_read("msg-mark-1", false)
|
||||
.await
|
||||
.expect("mark unread succeeds");
|
||||
let fetched = engine
|
||||
.get_message("msg-mark-1")
|
||||
.await
|
||||
.expect("get message succeeds");
|
||||
assert!(!fetched.read);
|
||||
|
||||
let err = engine
|
||||
.set_message_read("msg-does-not-exist", true)
|
||||
.await
|
||||
.expect_err("marking an unknown message must fail");
|
||||
assert!(matches!(
|
||||
err,
|
||||
DatabaseError::Query(sqlx::Error::RowNotFound)
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_save_and_list_accounts() {
|
||||
let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap();
|
||||
|
||||
+480
-13
@@ -7,18 +7,24 @@
|
||||
//! (`nuncio_core::ipc::IpcDaemonServer`); migrating callers off the JSON-RPC
|
||||
//! transport is out of scope here and lands in a later story.
|
||||
|
||||
use nuncio_core::{CoreEvent, EventBus};
|
||||
use nuncio_core::{CoreCommand, CoreEvent, EventBus};
|
||||
use nuncio_proto::v1::accounts_server::{Accounts, AccountsServer};
|
||||
use nuncio_proto::v1::event::Kind;
|
||||
use nuncio_proto::v1::mail_server::{Mail, MailServer};
|
||||
use nuncio_proto::v1::system_server::{System, SystemServer};
|
||||
use nuncio_proto::v1::{
|
||||
AccountConfig as AccountConfigProto, AccountProtocol as AccountProtocolProto,
|
||||
AddAccountRequest, AddAccountResponse, BatchFilterProgress, DatabaseRecovered, Event,
|
||||
EventError, FilterExecuted, GetStatusRequest, GetStatusResponse, ListAccountsRequest,
|
||||
ListAccountsResponse, MessageFlagsChanged, ShuttingDown, SubscribeRequest, SyncCompleted,
|
||||
SyncStarted, TlsMode as TlsModeProto, UpdateAvailable,
|
||||
AddAccountRequest, AddAccountResponse, Attachment as AttachmentProto, BatchFilterProgress,
|
||||
DatabaseRecovered, Event, EventError, FilterExecuted, Folder as FolderProto, GetMessageRequest,
|
||||
GetMessageResponse, GetStatusRequest, GetStatusResponse, ListAccountsRequest,
|
||||
ListAccountsResponse, ListFoldersRequest, ListFoldersResponse, ListMessagesRequest,
|
||||
ListMessagesResponse, MarkReadRequest, MarkReadResponse, Message as MessageProto,
|
||||
MessageFlagsChanged, MessageSearchHit, SearchMessagesRequest, SearchMessagesResponse,
|
||||
ShuttingDown, SubscribeRequest, SyncCompleted, SyncStarted, TlsMode as TlsModeProto,
|
||||
UpdateAvailable,
|
||||
};
|
||||
use nuncio_store::db::DatabaseEngine;
|
||||
use nuncio_store::search::SearchEngine;
|
||||
use nuncio_store::vault::SecretManager;
|
||||
use std::pin::Pin;
|
||||
use std::sync::Arc;
|
||||
@@ -336,6 +342,188 @@ 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 {
|
||||
AttachmentProto {
|
||||
filename: attachment.filename,
|
||||
mime_type: attachment.mime_type,
|
||||
content: attachment.content.to_vec(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Maps a `nuncio_core::model::Email` onto its wire-format `nuncio.v1.Message`
|
||||
/// representation. The store (`DatabaseEngine::get_message` /
|
||||
/// `list_messages`) already returns `body_plain`/`body_html` decrypted, so
|
||||
/// there is no further decryption to do here -- just field-for-field
|
||||
/// mapping.
|
||||
fn map_email_to_proto(email: nuncio_core::model::Email) -> MessageProto {
|
||||
MessageProto {
|
||||
id: email.id,
|
||||
account_id: email.account_id,
|
||||
folder_id: email.folder_id,
|
||||
subject: email.subject,
|
||||
sender: email.sender,
|
||||
recipient: email.recipient,
|
||||
received_at: email.received_at,
|
||||
read: email.read,
|
||||
body_plain: email.body_plain,
|
||||
body_html: email.body_html,
|
||||
attachments: email
|
||||
.attachments
|
||||
.into_iter()
|
||||
.map(map_attachment_to_proto)
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Maps a `nuncio_core::model::Folder` onto its wire-format `nuncio.v1.Folder`
|
||||
/// representation. `usize` counts are narrowed to `u64` for the wire; these
|
||||
/// are message counts that cannot realistically approach `u64::MAX`, and
|
||||
/// protobuf has no native `usize` type.
|
||||
fn map_folder_to_proto(folder: nuncio_core::model::Folder) -> FolderProto {
|
||||
FolderProto {
|
||||
id: folder.id,
|
||||
name: folder.name,
|
||||
total_messages: folder.total_messages as u64,
|
||||
unread_messages: folder.unread_messages as u64,
|
||||
}
|
||||
}
|
||||
|
||||
/// `nuncio.v1.Mail` gRPC service implementation backed by the daemon's live
|
||||
/// [`DatabaseEngine`] read/mark methods and [`SearchEngine`] FTS index
|
||||
/// (backlog story 1.C.4, GH #159).
|
||||
///
|
||||
/// This exposes the mail READ path (list folders, list messages, read a
|
||||
/// message, mark read/unread, search) over the real, persistent store that
|
||||
/// backlog story 1.C.3 (GH #158) syncs into -- as opposed to the CLI's
|
||||
/// previous local ephemeral `HeadlessRunner` database, which was thrown away
|
||||
/// when the CLI process exited.
|
||||
struct MailGrpcService {
|
||||
db: Arc<DatabaseEngine>,
|
||||
event_bus: Arc<EventBus>,
|
||||
}
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl Mail for MailGrpcService {
|
||||
async fn list_folders(
|
||||
&self,
|
||||
_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();
|
||||
|
||||
Ok(Response::new(ListFoldersResponse { folders }))
|
||||
}
|
||||
|
||||
async fn list_messages(
|
||||
&self,
|
||||
request: Request<ListMessagesRequest>,
|
||||
) -> Result<Response<ListMessagesResponse>, Status> {
|
||||
let req = request.into_inner();
|
||||
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 messages = self
|
||||
.db
|
||||
.list_messages(&req.folder_id, limit)
|
||||
.await
|
||||
.map_err(|e| Status::internal(format!("failed to list messages: {e}")))?
|
||||
.into_iter()
|
||||
.map(map_email_to_proto)
|
||||
.collect();
|
||||
|
||||
Ok(Response::new(ListMessagesResponse { messages }))
|
||||
}
|
||||
|
||||
async fn get_message(
|
||||
&self,
|
||||
request: Request<GetMessageRequest>,
|
||||
) -> Result<Response<GetMessageResponse>, Status> {
|
||||
let req = request.into_inner();
|
||||
if req.message_id.is_empty() {
|
||||
return Err(Status::invalid_argument("message_id is required"));
|
||||
}
|
||||
|
||||
let email = self.db.get_message(&req.message_id).await.map_err(|e| {
|
||||
Status::not_found(format!("message '{}' not found: {e}", req.message_id))
|
||||
})?;
|
||||
|
||||
Ok(Response::new(GetMessageResponse {
|
||||
message: Some(map_email_to_proto(email)),
|
||||
}))
|
||||
}
|
||||
|
||||
/// Persists the flag change via `DatabaseEngine::set_message_read` FIRST
|
||||
/// (so a stream subscriber never observes `MessageFlagsChanged` for a
|
||||
/// change that failed to persist), then drives it through
|
||||
/// `EventBus::process_command(CoreCommand::MarkRead { .. })` -- the same
|
||||
/// path the JSON-RPC IPC transport's `MarkRead` command uses -- so the
|
||||
/// in-memory unread-count state updates AND the existing
|
||||
/// `CoreEvent::MessageFlagsChanged` event is published, which is what
|
||||
/// makes this flow through `System/Subscribe` for free.
|
||||
async fn mark_read(
|
||||
&self,
|
||||
request: Request<MarkReadRequest>,
|
||||
) -> Result<Response<MarkReadResponse>, Status> {
|
||||
let req = request.into_inner();
|
||||
if req.message_id.is_empty() {
|
||||
return Err(Status::invalid_argument("message_id is required"));
|
||||
}
|
||||
|
||||
self.db
|
||||
.set_message_read(&req.message_id, req.read)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
Status::not_found(format!("message '{}' not found: {e}", req.message_id))
|
||||
})?;
|
||||
|
||||
self.event_bus.process_command(CoreCommand::MarkRead {
|
||||
message_id: req.message_id,
|
||||
read: req.read,
|
||||
});
|
||||
|
||||
Ok(Response::new(MarkReadResponse {}))
|
||||
}
|
||||
|
||||
async fn search_messages(
|
||||
&self,
|
||||
request: Request<SearchMessagesRequest>,
|
||||
) -> Result<Response<SearchMessagesResponse>, Status> {
|
||||
let req = request.into_inner();
|
||||
let search = SearchEngine::new(&self.db);
|
||||
let hits = search
|
||||
.search_messages(&req.query)
|
||||
.await
|
||||
.map_err(|e| Status::internal(format!("search failed: {e}")))?
|
||||
.into_iter()
|
||||
.map(|hit| MessageSearchHit {
|
||||
id: hit.id,
|
||||
title: hit.title,
|
||||
snippet: hit.snippet,
|
||||
})
|
||||
.collect();
|
||||
|
||||
Ok(Response::new(SearchMessagesResponse { hits }))
|
||||
}
|
||||
}
|
||||
|
||||
/// Bearer-token authentication interceptor for the loopback `nuncio.v1` gRPC
|
||||
/// server.
|
||||
///
|
||||
@@ -382,9 +570,9 @@ impl tonic::service::Interceptor for BearerAuthInterceptor {
|
||||
}
|
||||
}
|
||||
|
||||
/// Binds a loopback TCP listener at `addr` and serves the `nuncio.v1.System`
|
||||
/// and `nuncio.v1.Accounts` gRPC services on it, both authenticated by
|
||||
/// `token`, until the transport server errors.
|
||||
/// Binds a loopback TCP listener at `addr` and serves the `nuncio.v1.System`,
|
||||
/// `nuncio.v1.Accounts`, and `nuncio.v1.Mail` gRPC services on it, all
|
||||
/// authenticated by `token`, until the transport server errors.
|
||||
///
|
||||
/// `addr` MUST be a loopback address (e.g. `127.0.0.1:PORT`); callers are
|
||||
/// responsible for passing loopback-only addresses (see
|
||||
@@ -405,8 +593,8 @@ pub async fn serve(
|
||||
serve_on_listener(listener, event_bus, db, secrets, token).await
|
||||
}
|
||||
|
||||
/// Serves the `nuncio.v1.System` and `nuncio.v1.Accounts` gRPC services on
|
||||
/// an already-bound [`TcpListener`].
|
||||
/// Serves the `nuncio.v1.System`, `nuncio.v1.Accounts`, and `nuncio.v1.Mail`
|
||||
/// gRPC services on an already-bound [`TcpListener`].
|
||||
///
|
||||
/// # Security (GH #165)
|
||||
///
|
||||
@@ -432,17 +620,31 @@ pub async fn serve_on_listener(
|
||||
) -> Result<(), GrpcServeError> {
|
||||
let token: Arc<str> = token.into();
|
||||
|
||||
let system_service = SystemGrpcService { event_bus };
|
||||
let system_service = SystemGrpcService {
|
||||
event_bus: event_bus.clone(),
|
||||
};
|
||||
let system_interceptor = BearerAuthInterceptor::new(token.clone());
|
||||
let system_svc = SystemServer::with_interceptor(system_service, system_interceptor);
|
||||
|
||||
let accounts_service = AccountsGrpcService { db, secrets };
|
||||
let accounts_interceptor = BearerAuthInterceptor::new(token);
|
||||
let accounts_service = AccountsGrpcService {
|
||||
db: db.clone(),
|
||||
secrets,
|
||||
};
|
||||
let accounts_interceptor = BearerAuthInterceptor::new(token.clone());
|
||||
let accounts_svc = AccountsServer::with_interceptor(accounts_service, accounts_interceptor);
|
||||
|
||||
// Mail (backlog story 1.C.4, GH #159): mounted behind its own
|
||||
// `BearerAuthInterceptor`, exactly like `System` and `Accounts` above --
|
||||
// see the hard invariant documented on this function's doc comment
|
||||
// (GH #165).
|
||||
let mail_service = MailGrpcService { db, event_bus };
|
||||
let mail_interceptor = BearerAuthInterceptor::new(token);
|
||||
let mail_svc = MailServer::with_interceptor(mail_service, mail_interceptor);
|
||||
|
||||
Server::builder()
|
||||
.add_service(system_svc)
|
||||
.add_service(accounts_svc)
|
||||
.add_service(mail_svc)
|
||||
.serve_with_incoming(TcpListenerStream::new(listener))
|
||||
.await
|
||||
.map_err(GrpcServeError::Transport)
|
||||
@@ -1058,4 +1260,269 @@ mod tests {
|
||||
"password leaked into the ListAccounts response: {response_debug}"
|
||||
);
|
||||
}
|
||||
|
||||
// ---- Mail (backlog story 1.C.4, GH #159) ----
|
||||
|
||||
use nuncio_proto::v1::mail_client::MailClient;
|
||||
use nuncio_proto::v1::{
|
||||
GetMessageRequest, ListFoldersRequest, ListMessagesRequest, MarkReadRequest,
|
||||
SearchMessagesRequest,
|
||||
};
|
||||
|
||||
fn sample_email(
|
||||
id: &str,
|
||||
folder_id: &str,
|
||||
subject: &str,
|
||||
body: &str,
|
||||
) -> nuncio_core::model::Email {
|
||||
nuncio_core::model::Email {
|
||||
id: id.to_string(),
|
||||
account_id: "acct-mail-1".to_string(),
|
||||
folder_id: folder_id.to_string(),
|
||||
subject: subject.to_string(),
|
||||
sender: "alice@nuncio.mx".to_string(),
|
||||
recipient: "bob@nuncio.mx".to_string(),
|
||||
received_at: 1_700_000_000,
|
||||
read: false,
|
||||
body_plain: Some(body.to_string()),
|
||||
body_html: None,
|
||||
attachments: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn mail_rpcs_reject_missing_bearer_token() {
|
||||
// Confirms `Mail` is mounted behind its own `BearerAuthInterceptor`
|
||||
// exactly like `System` and `Accounts` (GH #165: no un-intercepted
|
||||
// service).
|
||||
let event_bus = Arc::new(EventBus::new());
|
||||
let (addr, _handle) = spawn_test_server(event_bus, "correct-token").await;
|
||||
|
||||
let mut client = MailClient::connect(format!("http://{addr}"))
|
||||
.await
|
||||
.expect("client connects");
|
||||
|
||||
let err = client
|
||||
.list_folders(ListFoldersRequest {})
|
||||
.await
|
||||
.expect_err("missing bearer token must be rejected");
|
||||
assert_eq!(err.code(), Code::Unauthenticated);
|
||||
|
||||
let err = client
|
||||
.list_messages(ListMessagesRequest {
|
||||
folder_id: "inbox".to_string(),
|
||||
limit: 10,
|
||||
})
|
||||
.await
|
||||
.expect_err("missing bearer token must be rejected");
|
||||
assert_eq!(err.code(), Code::Unauthenticated);
|
||||
|
||||
let err = client
|
||||
.get_message(GetMessageRequest {
|
||||
message_id: "msg-1".to_string(),
|
||||
})
|
||||
.await
|
||||
.expect_err("missing bearer token must be rejected");
|
||||
assert_eq!(err.code(), Code::Unauthenticated);
|
||||
|
||||
let err = client
|
||||
.mark_read(MarkReadRequest {
|
||||
message_id: "msg-1".to_string(),
|
||||
read: true,
|
||||
})
|
||||
.await
|
||||
.expect_err("missing bearer token must be rejected");
|
||||
assert_eq!(err.code(), Code::Unauthenticated);
|
||||
|
||||
let err = client
|
||||
.search_messages(SearchMessagesRequest {
|
||||
query: "hello".to_string(),
|
||||
})
|
||||
.await
|
||||
.expect_err("missing bearer token must be rejected");
|
||||
assert_eq!(err.code(), Code::Unauthenticated);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn get_message_reports_not_found_for_unknown_message_id() {
|
||||
let event_bus = Arc::new(EventBus::new());
|
||||
let (addr, _handle) = spawn_test_server(event_bus, "correct-token").await;
|
||||
|
||||
let mut client = MailClient::connect(format!("http://{addr}"))
|
||||
.await
|
||||
.expect("client connects");
|
||||
let err = client
|
||||
.get_message(authed_bearer_request(GetMessageRequest {
|
||||
message_id: "does-not-exist".to_string(),
|
||||
}))
|
||||
.await
|
||||
.expect_err("unknown message id must be rejected");
|
||||
assert_eq!(err.code(), Code::NotFound);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn mark_read_reports_not_found_for_unknown_message_id() {
|
||||
let event_bus = Arc::new(EventBus::new());
|
||||
let (addr, _handle) = spawn_test_server(event_bus, "correct-token").await;
|
||||
|
||||
let mut client = MailClient::connect(format!("http://{addr}"))
|
||||
.await
|
||||
.expect("client connects");
|
||||
let err = client
|
||||
.mark_read(authed_bearer_request(MarkReadRequest {
|
||||
message_id: "does-not-exist".to_string(),
|
||||
read: true,
|
||||
}))
|
||||
.await
|
||||
.expect_err("unknown message id must be rejected");
|
||||
assert_eq!(err.code(), Code::NotFound);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_messages_rejects_empty_folder_id() {
|
||||
let event_bus = Arc::new(EventBus::new());
|
||||
let (addr, _handle) = 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: String::new(),
|
||||
limit: 10,
|
||||
}))
|
||||
.await
|
||||
.expect_err("empty folder_id must be rejected");
|
||||
assert_eq!(err.code(), Code::InvalidArgument);
|
||||
}
|
||||
|
||||
/// End-to-end proof for backlog story 1.C.4 (GH #159): seeds the
|
||||
/// daemon's real, persistent store directly via `DatabaseEngine::
|
||||
/// save_email` (standing in for backlog story 1.C.3's real sync path,
|
||||
/// which uses the exact same write path), then proves every `Mail` RPC
|
||||
/// round-trips real data: `ListFolders` reports the seeded folder,
|
||||
/// `ListMessages` returns the seeded messages, `GetMessage` returns the
|
||||
/// full (decrypted) body, `SearchMessages` finds a body term via FTS5,
|
||||
/// and `MarkRead` persists across a second `GetMessage` call AND
|
||||
/// publishes `MessageFlagsChanged` on the live event bus (proving it
|
||||
/// would stream via `System/Subscribe`).
|
||||
#[tokio::test]
|
||||
async fn mail_rpcs_round_trip_real_seeded_data_and_mark_read_persists_and_streams() {
|
||||
let (db, _dir) = DatabaseEngine::connect_ephemeral()
|
||||
.await
|
||||
.expect("connect ephemeral test db");
|
||||
db.save_email(&sample_email(
|
||||
"msg-mail-1",
|
||||
"inbox",
|
||||
"Quarterly Roadmap",
|
||||
"Let's discuss the annual revenue forecast",
|
||||
))
|
||||
.await
|
||||
.expect("seed message 1");
|
||||
db.save_email(&sample_email(
|
||||
"msg-mail-2",
|
||||
"inbox",
|
||||
"Lunch Plans",
|
||||
"Sandwiches at noon",
|
||||
))
|
||||
.await
|
||||
.expect("seed message 2");
|
||||
|
||||
let event_bus = Arc::new(EventBus::new());
|
||||
let mut events = event_bus.subscribe_events();
|
||||
let secrets = Arc::new(SecretManager::mock());
|
||||
let (addr, _handle) =
|
||||
spawn_test_server_with(event_bus, Arc::new(db), secrets, "correct-token").await;
|
||||
|
||||
let mut client = MailClient::connect(format!("http://{addr}"))
|
||||
.await
|
||||
.expect("client connects");
|
||||
|
||||
// ListFolders reports the seeded folder with the right counts.
|
||||
let folders = client
|
||||
.list_folders(authed_bearer_request(ListFoldersRequest {}))
|
||||
.await
|
||||
.expect("list_folders succeeds")
|
||||
.into_inner()
|
||||
.folders;
|
||||
assert_eq!(folders.len(), 1);
|
||||
assert_eq!(folders[0].id, "inbox");
|
||||
assert_eq!(folders[0].total_messages, 2);
|
||||
assert_eq!(folders[0].unread_messages, 2);
|
||||
|
||||
// ListMessages returns both seeded messages, newest-first.
|
||||
let messages = client
|
||||
.list_messages(authed_bearer_request(ListMessagesRequest {
|
||||
folder_id: "inbox".to_string(),
|
||||
limit: 10,
|
||||
}))
|
||||
.await
|
||||
.expect("list_messages succeeds")
|
||||
.into_inner()
|
||||
.messages;
|
||||
assert_eq!(messages.len(), 2);
|
||||
let ids: Vec<&str> = messages.iter().map(|m| m.id.as_str()).collect();
|
||||
assert!(ids.contains(&"msg-mail-1"));
|
||||
assert!(ids.contains(&"msg-mail-2"));
|
||||
|
||||
// GetMessage returns the full, decrypted body.
|
||||
let fetched = client
|
||||
.get_message(authed_bearer_request(GetMessageRequest {
|
||||
message_id: "msg-mail-1".to_string(),
|
||||
}))
|
||||
.await
|
||||
.expect("get_message succeeds")
|
||||
.into_inner()
|
||||
.message
|
||||
.expect("message present in response");
|
||||
assert_eq!(fetched.subject, "Quarterly Roadmap");
|
||||
assert_eq!(
|
||||
fetched.body_plain.as_deref(),
|
||||
Some("Let's discuss the annual revenue forecast")
|
||||
);
|
||||
assert!(!fetched.read);
|
||||
|
||||
// SearchMessages finds a body term via the real FTS5 index.
|
||||
let hits = client
|
||||
.search_messages(authed_bearer_request(SearchMessagesRequest {
|
||||
query: "revenue".to_string(),
|
||||
}))
|
||||
.await
|
||||
.expect("search_messages succeeds")
|
||||
.into_inner()
|
||||
.hits;
|
||||
assert_eq!(hits.len(), 1);
|
||||
assert_eq!(hits[0].id, "msg-mail-1");
|
||||
|
||||
// MarkRead persists: a second GetMessage call reflects the flip.
|
||||
client
|
||||
.mark_read(authed_bearer_request(MarkReadRequest {
|
||||
message_id: "msg-mail-1".to_string(),
|
||||
read: true,
|
||||
}))
|
||||
.await
|
||||
.expect("mark_read succeeds");
|
||||
|
||||
let refetched = client
|
||||
.get_message(authed_bearer_request(GetMessageRequest {
|
||||
message_id: "msg-mail-1".to_string(),
|
||||
}))
|
||||
.await
|
||||
.expect("get_message succeeds after mark_read")
|
||||
.into_inner()
|
||||
.message
|
||||
.expect("message present in response");
|
||||
assert!(refetched.read);
|
||||
|
||||
// MarkRead publishes `MessageFlagsChanged` on the live event bus, so
|
||||
// it would stream via `System/Subscribe` for a connected client.
|
||||
let event = events.recv().await.expect("event bus yields an event");
|
||||
assert_eq!(
|
||||
event,
|
||||
CoreEvent::MessageFlagsChanged {
|
||||
message_id: "msg-mail-1".to_string(),
|
||||
read: true,
|
||||
}
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user