From 20ce331a5531362222ae248276111ef666f25643 Mon Sep 17 00:00:00 2001 From: James Cleveland Date: Fri, 28 Aug 2026 09:38:27 +0100 Subject: [PATCH 1/2] feat(watch): stream arriving mail as NDJSON over JMAP push MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #64. `fastmail watch` blocks and emits one JSON object per line as mail lands, so incoming email can drive a real-time shell loop rather than a cron job that re-lists the inbox and diffs it by hand. JMAP already carried the transport: the session advertises an `eventSourceUrl` (RFC 8620 §7.3) that we parsed and discarded. Push is treated purely as a wake-up — the `Email` state cursor lives in the CLI, and every notification, poll tick and reconnect runs `Email/changes` against it, so all three paths converge on the same answer and a lost notification costs latency rather than mail. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_016ri9XAwMqKiHLrDyTuA3np --- CHANGELOG.md | 24 +++ README.md | 39 +++++ src/commands/mod.rs | 2 + src/commands/watch.rs | 237 +++++++++++++++++++++++++++ src/jmap/events.rs | 153 +++++++++++++++++ src/jmap/mod.rs | 373 +++++++++++++++++++++++++++++++++++++++++- src/main.rs | 28 ++++ src/models/mod.rs | 14 ++ 8 files changed, 867 insertions(+), 3 deletions(-) create mode 100644 src/commands/watch.rs create mode 100644 src/jmap/events.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 595a981..d9a8085 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,29 @@ # Changelog +## [Unreleased] + +### Added + +- **`fastmail watch`.** Blocks and emits one JSON object per line as mail + arrives, so incoming email can drive a shell loop instead of a cron job that + re-lists the inbox and diffs it by hand. `--mailbox` narrows to one folder, + `--full` includes bodies, and `--poll ` swaps the push connection for + periodic checks where a long-lived one will not survive the network. + + JMAP has offered the transport all along: the session advertises an + `eventSourceUrl` (RFC 8620 §7.3) that we parsed and threw away. The design + decision worth knowing is that push is only ever a wake-up. The `Email` state + cursor lives in the CLI, and every notification — or poll tick, or reconnect — + runs `Email/changes` against it, so all three paths converge on the same + answer and a lost notification costs latency rather than mail. When the server + has discarded history back past the cursor, the watcher resyncs and says so on + stderr rather than replaying the mailbox as new; stdout stays pure NDJSON + either way. + + Only creations are reported. Reporting updates too would replay every flag + change and folder move as an arrival, which is not what a mail loop means by + "new". + ## [3.4.0] - 2026-08-17 ### Added diff --git a/README.md b/README.md index a68e364..e76f4aa 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ CLI for Fastmail's JMAP API. Read, search, send, and manage emails from your ter | **Text Extraction** | 56 formats via [kreuzberg](https://github.com/kreuzberg-dev/kreuzberg) | | **Image Resizing** | `--max-size` to resize images on download | | **Masked Email** | Create, list, enable/disable aliases | +| **Watch** | Stream arriving mail as NDJSON over JMAP push, for real-time loops | | **MCP Server** | Claude integration via Model Context Protocol | | **Shell Completions** | Bash, Zsh, Fish, PowerShell | | **JSON Output** | All commands output JSON for scripting | @@ -169,6 +170,44 @@ fastmail search --from "boss" --has-attachment --after 2024-06-01 --limit 20 Available flags: `--text`, `--from`, `--to`, `--cc`, `--bcc`, `--subject`, `--body`, `--mailbox`, `--has-attachment`, `--min-size`, `--max-size`, `--before`, `--after`, `--unread`, `--flagged` +### Watch for New Mail + +Block and emit one JSON object per line as mail arrives, so a shell loop can act +on it: + +```bash +# Everything that arrives, anywhere in the account +fastmail watch + +# Just the inbox +fastmail watch --mailbox inbox + +# Pipe into a loop +fastmail watch --mailbox inbox | while read -r line; do + echo "$line" | jq -r '.data.subject' +done + +# Include bodies and attachment metadata, not just summaries +fastmail watch --full + +# Fall back to polling every 60s where a long-lived connection won't survive +fastmail watch --poll 60 +``` + +Output is the same `{"success":true,"data":{...}}` envelope as every other +command, one compact line per email, flushed as it is written — so `jq` filters +and `read` loops both work unbuffered. + +It uses JMAP's push channel (`eventSourceUrl`), but treats a notification purely +as a signal to look again: the state cursor lives in the CLI, and each wake-up +runs `Email/changes` against it. Dropped connections are reconciled on reconnect +and `--poll` takes the identical path, so a missed notification costs latency +rather than mail. Only *new* messages are reported — flag and folder changes to +existing mail are not arrivals. + +Reconnects, and the rare case where the server has discarded change history and +the cursor has to resync, are reported on stderr; stdout stays pure NDJSON. + ### List Identities View available sender identities (useful for `--from`): diff --git a/src/commands/mod.rs b/src/commands/mod.rs index ffc7f3b..e637850 100644 --- a/src/commands/mod.rs +++ b/src/commands/mod.rs @@ -12,6 +12,7 @@ mod search; mod send; mod spam; mod thread; +mod watch; pub use auth::*; pub use contacts::*; @@ -27,3 +28,4 @@ pub use search::*; pub use send::*; pub use spam::*; pub use thread::*; +pub use watch::*; diff --git a/src/commands/watch.rs b/src/commands/watch.rs new file mode 100644 index 0000000..3bdb0d2 --- /dev/null +++ b/src/commands/watch.rs @@ -0,0 +1,237 @@ +use crate::error::Error; +use crate::jmap::{EventParser, JmapClient, authenticated_client}; +use crate::models::{Email, Output}; +use std::time::Duration; +use tracing::debug; + +/// How often the server should send a keep-alive on the push channel. Also sets +/// the read timeout that detects a connection which died without saying so. +const PING_SECONDS: u32 = 30; + +/// Reconnect backoff bounds. The ceiling is deliberately below the ping +/// interval's own scale: a watcher that goes quiet for minutes after a blip is +/// indistinguishable from a broken one. +const BACKOFF_START: u64 = 1; +const BACKOFF_MAX: u64 = 30; + +pub struct WatchOptions { + /// Only report mail landing in this mailbox, by name or role. + pub mailbox: Option, + /// Fetch bodies and attachment metadata rather than summaries. + pub full: bool, + /// Check every N seconds instead of holding a push connection open. + pub poll: Option, +} + +/// Stream newly arrived emails as newline-delimited JSON, one object per line, +/// until interrupted. +/// +/// Arrivals are discovered through `Email/changes` against a state cursor this +/// function owns. The push channel only ever says *look again* — so a dropped +/// connection, a missed event or a `--poll` fallback all converge on the same +/// answer, and the cost of losing a notification is latency rather than mail. +pub async fn watch(opts: WatchOptions) -> anyhow::Result<()> { + let mut client = authenticated_client().await?; + + let mailbox_id = match opts.mailbox { + Some(ref name) => Some(client.find_mailbox(name).await?.id), + None => None, + }; + + // Start from now: the caller asked what arrives next, not what is already + // sitting there. + let mut state = client.email_state().await?; + + match opts.poll { + Some(seconds) => { + poll_loop( + &client, + &mut state, + mailbox_id.as_deref(), + opts.full, + seconds, + ) + .await + } + None => push_loop(&client, &mut state, mailbox_id.as_deref(), opts.full).await, + } +} + +async fn push_loop( + client: &JmapClient, + state: &mut String, + mailbox_id: Option<&str>, + full: bool, +) -> anyhow::Result<()> { + let mut backoff = BACKOFF_START; + let mut last_event_id: Option = None; + + loop { + match client + .open_event_stream(PING_SECONDS, last_event_id.as_deref()) + .await + { + Ok(mut resp) => { + let mut parser = EventParser::default(); + loop { + match resp.chunk().await { + Ok(Some(bytes)) => { + // Reset only once the connection has actually + // carried something. Resetting on connect alone + // would let a server that accepts and immediately + // hangs up spin at full rate. + backoff = BACKOFF_START; + + for event in parser.feed(&String::from_utf8_lossy(&bytes)) { + if let Some(id) = event.id { + last_event_id = Some(id); + } + // Keep-alives carry no state change. + if event.event.as_deref() == Some("ping") || event.data.is_empty() { + continue; + } + drain(client, state, mailbox_id, full).await?; + } + } + Ok(None) => { + debug!("Event source closed by server"); + break; + } + Err(e) => { + eprintln!("watch: event stream dropped ({e}); reconnecting"); + break; + } + } + } + } + Err(e) if fatal(&e) => return Err(e.into()), + Err(e) => eprintln!("watch: could not open event stream ({e}); retrying"), + } + + tokio::time::sleep(Duration::from_secs(backoff)).await; + backoff = (backoff * 2).min(BACKOFF_MAX); + + // Reconcile across the gap before waiting on the next push, so mail + // that landed while disconnected is reported on reconnect rather than + // on whatever arrives after it. + drain(client, state, mailbox_id, full).await?; + } +} + +async fn poll_loop( + client: &JmapClient, + state: &mut String, + mailbox_id: Option<&str>, + full: bool, + seconds: u64, +) -> anyhow::Result<()> { + loop { + tokio::time::sleep(Duration::from_secs(seconds)).await; + drain(client, state, mailbox_id, full).await?; + } +} + +/// Advance the cursor and print whatever arrived, oldest first. +/// +/// Transient failures are reported and swallowed: a watcher that exits on one +/// bad response is worse than useless in the loop it is meant to feed. A dead +/// credential is not transient, so it still ends the process. +async fn drain( + client: &JmapClient, + state: &mut String, + mailbox_id: Option<&str>, + full: bool, +) -> anyhow::Result<()> { + let changes = match client.email_changes(state).await { + Ok(changes) => changes, + Err(Error::Jmap { ref error_type, .. }) if error_type == "cannotCalculateChanges" => { + // The server has discarded history back to our cursor. There is no + // way to know what was missed, and replaying the mailbox as "new" + // would be a lie, so resync to now and say so. + *state = client.email_state().await?; + eprintln!( + "watch: server dropped change history; resynced, some arrivals may be missing" + ); + return Ok(()); + } + Err(e) if fatal(&e) => return Err(e.into()), + Err(e) => { + eprintln!("watch: could not read changes ({e})"); + return Ok(()); + } + }; + + *state = changes.new_state; + if changes.created.is_empty() { + return Ok(()); + } + + let fetched = if full { + client.get_emails(&changes.created).await + } else { + client.get_email_summaries(&changes.created).await + }; + + let mut emails = match fetched { + Ok(emails) => emails, + Err(e) if fatal(&e) => return Err(e.into()), + Err(e) => { + eprintln!( + "watch: could not fetch {} new email(s) ({e})", + changes.created.len() + ); + return Ok(()); + } + }; + + // `Email/get` makes no ordering guarantee, and a stream reads chronologically. + emails.sort_by(|a, b| a.received_at.cmp(&b.received_at)); + + for email in emails.iter().filter(|e| in_mailbox(e, mailbox_id)) { + Output::success(email).print_compact(); + } + + Ok(()) +} + +fn in_mailbox(email: &Email, mailbox_id: Option<&str>) -> bool { + mailbox_id.is_none_or(|id| email.mailbox_ids.contains_key(id)) +} + +/// Whether an error means the watcher can never succeed again. +fn fatal(e: &Error) -> bool { + matches!(e, Error::InvalidToken(_) | Error::NotAuthenticated) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn email_in(mailbox_ids: &[&str]) -> Email { + let ids: serde_json::Map<_, _> = mailbox_ids + .iter() + .map(|id| (id.to_string(), serde_json::json!(true))) + .collect(); + serde_json::from_value(serde_json::json!({ "id": "e1", "mailboxIds": ids })).unwrap() + } + + #[test] + fn no_mailbox_filter_accepts_everything() { + assert!(in_mailbox(&email_in(&["archive"]), None)); + } + + #[test] + fn mailbox_filter_matches_membership() { + let email = email_in(&["inbox", "important"]); + assert!(in_mailbox(&email, Some("inbox"))); + assert!(!in_mailbox(&email, Some("trash"))); + } + + #[test] + fn a_dead_credential_is_fatal_but_a_bad_response_is_not() { + assert!(fatal(&Error::NotAuthenticated)); + assert!(fatal(&Error::InvalidToken("nope"))); + assert!(!fatal(&Error::RateLimited)); + assert!(!fatal(&Error::Server("boom".into()))); + } +} diff --git a/src/jmap/events.rs b/src/jmap/events.rs new file mode 100644 index 0000000..6c37c81 --- /dev/null +++ b/src/jmap/events.rs @@ -0,0 +1,153 @@ +//! Server-Sent Events parsing for the JMAP push channel (RFC 8620 §7.3). +//! +//! The wire format is [SSE]: `field: value` lines, frames separated by a blank +//! line. Only `event`, `data` and `id` carry meaning here — `id` is what a +//! reconnect replays from, via the `Last-Event-ID` header. +//! +//! [SSE]: https://html.spec.whatwg.org/multipage/server-sent-events.html + +/// One complete SSE frame. +#[derive(Debug, Default, Clone, PartialEq, Eq)] +pub struct ServerEvent { + pub event: Option, + pub data: String, + pub id: Option, +} + +/// Reassembles frames from arbitrary byte chunks. +/// +/// Chunk boundaries fall wherever the network puts them — mid-frame, mid-line, +/// even between the `\r` and `\n` of a line ending — so partial input is held +/// until the blank line that terminates a frame actually arrives. +#[derive(Default)] +pub struct EventParser { + buf: String, +} + +impl EventParser { + /// Append a chunk and return every frame it completed. + pub fn feed(&mut self, chunk: &str) -> Vec { + self.buf.push_str(chunk); + + let mut out = Vec::new(); + while let Some((at, len)) = next_frame_end(&self.buf) { + let frame: String = self.buf.drain(..at + len).collect(); + if let Some(event) = parse_frame(&frame) { + out.push(event); + } + } + out + } +} + +/// Offset and length of the first frame separator: a blank line, in either +/// line-ending convention. Returns the earliest match so a stream that mixes +/// them cannot desynchronise. +fn next_frame_end(buf: &str) -> Option<(usize, usize)> { + let lf = buf.find("\n\n").map(|at| (at, 2)); + let crlf = buf.find("\r\n\r\n").map(|at| (at, 4)); + match (lf, crlf) { + (Some(a), Some(b)) => Some(if a.0 <= b.0 { a } else { b }), + (found, None) | (None, found) => found, + } +} + +/// A frame with no recognised field is not an event — that is how keep-alive +/// comments (`: ping`) stay invisible to callers. +fn parse_frame(frame: &str) -> Option { + let mut event = ServerEvent::default(); + let mut data = Vec::new(); + let mut recognised = false; + + for line in frame.lines() { + if line.is_empty() || line.starts_with(':') { + continue; + } + // A line with no colon is a field with an empty value. + let (field, value) = match line.split_once(':') { + Some((field, value)) => (field, value.strip_prefix(' ').unwrap_or(value)), + None => (line, ""), + }; + match field { + "event" => event.event = Some(value.to_string()), + "data" => data.push(value), + "id" => event.id = Some(value.to_string()), + // `retry` and unknown fields are ignored: reconnect backoff is the + // caller's, and it has better information than the server does. + _ => continue, + } + recognised = true; + } + + recognised.then(|| { + event.data = data.join("\n"); + event + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_a_whole_frame() { + let mut parser = EventParser::default(); + let events = parser.feed("event: state\ndata: {\"x\":1}\nid: abc\n\n"); + assert_eq!( + events, + vec![ServerEvent { + event: Some("state".into()), + data: "{\"x\":1}".into(), + id: Some("abc".into()), + }] + ); + } + + #[test] + fn holds_a_frame_split_across_chunks() { + let mut parser = EventParser::default(); + assert!(parser.feed("event: sta").is_empty()); + assert!(parser.feed("te\ndata: {\"x\":1}").is_empty()); + let events = parser.feed("\n\n"); + assert_eq!(events.len(), 1); + assert_eq!(events[0].data, "{\"x\":1}"); + } + + #[test] + fn holds_a_frame_split_between_cr_and_lf() { + let mut parser = EventParser::default(); + assert!(parser.feed("data: hi\r\n\r").is_empty()); + let events = parser.feed("\ndata: there\r\n\r\n"); + assert_eq!(events.len(), 2); + assert_eq!(events[0].data, "hi"); + assert_eq!(events[1].data, "there"); + } + + #[test] + fn joins_repeated_data_lines() { + let mut parser = EventParser::default(); + let events = parser.feed("data: one\ndata: two\n\n"); + assert_eq!(events[0].data, "one\ntwo"); + } + + #[test] + fn yields_several_frames_from_one_chunk() { + let mut parser = EventParser::default(); + let events = parser.feed("data: a\n\ndata: b\n\ndata: c\n\n"); + assert_eq!(events.len(), 3); + } + + #[test] + fn skips_comment_only_frames() { + let mut parser = EventParser::default(); + assert!(parser.feed(": ping\n\n").is_empty()); + } + + #[test] + fn tolerates_a_missing_space_after_the_colon() { + let mut parser = EventParser::default(); + let events = parser.feed("event:state\ndata:{}\n\n"); + assert_eq!(events[0].event.as_deref(), Some("state")); + assert_eq!(events[0].data, "{}"); + } +} diff --git a/src/jmap/mod.rs b/src/jmap/mod.rs index 0f8b194..bb703ae 100644 --- a/src/jmap/mod.rs +++ b/src/jmap/mod.rs @@ -1,3 +1,5 @@ +mod events; + use crate::commands::SearchFilter; use crate::error::{Error, Result}; use crate::models::*; @@ -8,6 +10,8 @@ use std::collections::HashMap; use std::time::Duration; use tracing::{debug, instrument}; +pub use events::{EventParser, ServerEvent}; + const SESSION_URL: &str = "https://api.fastmail.com/jmap/session"; const TIMEOUT: Duration = Duration::from_secs(30); @@ -255,6 +259,19 @@ impl ComposeContext { } } +/// How many changes to ask for per `Email/changes` call. The server may cap it +/// lower; `hasMoreChanges` then drives the next page. +const CHANGES_PAGE: u32 = 100; + +/// What arrived since a known `Email` state. +#[derive(Debug)] +pub struct EmailChanges { + /// The state to pass as `sinceState` next time. + pub new_state: String, + /// IDs created in that window, oldest change first. + pub created: Vec, +} + // Shared JMAP response types used across multiple methods #[derive(Deserialize)] struct GetResponse { @@ -885,6 +902,27 @@ impl JmapClient { /// email DataLoader. #[instrument(skip(self))] pub async fn get_emails(&self, ids: &[String]) -> Result> { + self.get_email_records(ids, EMAIL_FULL_PROPERTIES, true) + .await + } + + /// Summary records for known IDs — the cheap counterpart to [`Self::get_emails`]. + /// + /// `Email/query` already returns summaries for the page it matched; this is + /// for the callers that arrive holding IDs from somewhere else, such as + /// `Email/changes`. + #[instrument(skip(self))] + pub async fn get_email_summaries(&self, ids: &[String]) -> Result> { + self.get_email_records(ids, EMAIL_SUMMARY_PROPERTIES, false) + .await + } + + async fn get_email_records( + &self, + ids: &[String], + properties: &[&str], + fetch_bodies: bool, + ) -> Result> { if ids.is_empty() { return Ok(Vec::new()); } @@ -896,9 +934,9 @@ impl JmapClient { { "accountId": account_id, "ids": ids, - "properties": EMAIL_FULL_PROPERTIES, - "fetchTextBodyValues": true, - "fetchHTMLBodyValues": true + "properties": properties, + "fetchTextBodyValues": fetch_bodies, + "fetchHTMLBodyValues": fetch_bodies }, "g0" ])]) @@ -910,6 +948,148 @@ impl JmapClient { Ok(resp.list) } + /// The account's current `Email` state string: the cursor + /// [`Self::email_changes`] reads forward from. + /// + /// Fetched with an empty `ids` list, so the server returns the state and no + /// mail. + #[instrument(skip(self))] + pub async fn email_state(&self) -> Result { + let account_id = self.account_id()?; + + let responses = self + .request(vec![json!([ + "Email/get", + { "accountId": account_id, "ids": [] }, + "s0" + ])]) + .await?; + + #[derive(Deserialize)] + struct StateOnly { + state: String, + } + + let resp: StateOnly = + Self::parse_response(responses.first().unwrap_or(&Value::Null), "Email/get")?; + Ok(resp.state) + } + + /// IDs of emails created since `since_state`, and the state they leave the + /// caller at. + /// + /// Follows `hasMoreChanges` to the end, so the returned state is always + /// current: a partial read would silently drop everything past the first + /// page on the next call. Only creations are reported — a watcher wants + /// arrivals, and updates would replay every flag change as news. + /// + /// Fails with a `cannotCalculateChanges` JMAP error when the server has + /// discarded history back that far; the caller resyncs via + /// [`Self::email_state`]. + #[instrument(skip(self))] + pub async fn email_changes(&self, since_state: &str) -> Result { + let account_id = self.account_id()?; + let mut state = since_state.to_string(); + let mut created = Vec::new(); + + loop { + let responses = self + .request(vec![json!([ + "Email/changes", + { + "accountId": account_id, + "sinceState": state, + "maxChanges": CHANGES_PAGE + }, + "c0" + ])]) + .await?; + + #[derive(Deserialize)] + #[serde(rename_all = "camelCase")] + struct ChangesResponse { + new_state: String, + #[serde(default)] + has_more_changes: bool, + #[serde(default)] + created: Vec, + } + + let resp: ChangesResponse = + Self::parse_response(responses.first().unwrap_or(&Value::Null), "Email/changes")?; + + created.extend(resp.created); + state = resp.new_state; + + if !resp.has_more_changes { + return Ok(EmailChanges { + new_state: state, + created, + }); + } + } + } + + /// Open the JMAP push channel and return the live response to read frames + /// from. + /// + /// `last_event_id` asks the server to replay from where a dropped + /// connection left off. Missing it is not a correctness problem — the + /// caller holds its own `Email` state and reconciles through + /// [`Self::email_changes`] — but it saves a round trip. + #[instrument(skip(self))] + pub async fn open_event_stream( + &self, + ping: u32, + last_event_id: Option<&str>, + ) -> Result { + let template = self + .session()? + .event_source_url + .as_deref() + .ok_or_else(|| { + Error::Config( + "Server advertises no eventSourceUrl for push. Use --poll to fall back to \ + periodic checks." + .into(), + ) + })? + .to_string(); + + let url = template + .replace("{types}", "Email") + .replace("{closeafter}", "no") + .replace("{ping}", &ping.to_string()); + + // The shared client caps every request at 30s; a push channel is meant + // to stay open for days. A read timeout of a few ping intervals stands + // in for it, so silence reads as a dead connection rather than an idle + // one — the difference between reconnecting and hanging forever. + let client = Client::builder() + .read_timeout(Duration::from_secs(u64::from(ping) * 3)) + .build()?; + + let mut req = client + .get(&url) + .bearer_auth(&self.token) + .header("Accept", "text/event-stream"); + if let Some(id) = last_event_id { + req = req.header("Last-Event-ID", id); + } + + debug!(url = %url, "Opening JMAP event source"); + let resp = req.send().await?; + + match resp.status().as_u16() { + 401 => return Err(Error::InvalidToken("Token expired or invalid")), + 429 => return Err(Error::RateLimited), + 500..=599 => return Err(Error::Server(format!("Server error: {}", resp.status()))), + _ => {} + } + + Ok(resp) + } + #[instrument(skip(self))] pub async fn get_email(&self, email_id: &str) -> Result { let ids = [email_id.to_string()]; @@ -2081,6 +2261,193 @@ mod tests { // ============ upload_blob mock test ============ + /// Build a client pointed at a mock JMAP server. + fn mock_client(uri: &str) -> JmapClient { + let mut client = JmapClient::new("test-token".to_string()); + let mut session = create_test_session(vec![ + "urn:ietf:params:jmap:core", + "urn:ietf:params:jmap:mail", + ]); + session.api_url = format!("{uri}/jmap"); + client.available_capabilities = session.capabilities.keys().cloned().collect(); + client.session = Some(session); + client + } + + /// One `methodResponses` envelope around a single method result. + fn jmap_response(method: &str, result: serde_json::Value) -> serde_json::Value { + serde_json::json!({ "methodResponses": [[method, result, "c0"]] }) + } + + #[tokio::test] + async fn test_email_state_reads_state_without_fetching_mail() { + use wiremock::matchers::{body_string_contains, method}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let mock_server = MockServer::start().await; + + Mock::given(method("POST")) + .and(body_string_contains(r#""ids":[]"#)) + .respond_with(ResponseTemplate::new(200).set_body_json(jmap_response( + "Email/get", + serde_json::json!({ "state": "state-42", "list": [], "notFound": [] }), + ))) + .mount(&mock_server) + .await; + + let client = mock_client(&mock_server.uri()); + assert_eq!(client.email_state().await.unwrap(), "state-42"); + } + + #[tokio::test] + async fn test_email_changes_follows_has_more_changes() { + use wiremock::matchers::{body_string_contains, method}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let mock_server = MockServer::start().await; + + // Each page is matched by the state it was asked to start from, so the + // test asserts the cursor actually advances rather than trusting order. + Mock::given(method("POST")) + .and(body_string_contains(r#""sinceState":"s0""#)) + .respond_with(ResponseTemplate::new(200).set_body_json(jmap_response( + "Email/changes", + serde_json::json!({ + "oldState": "s0", + "newState": "s1", + "hasMoreChanges": true, + "created": ["e1", "e2"], + "updated": [], + "destroyed": [] + }), + ))) + .mount(&mock_server) + .await; + + Mock::given(method("POST")) + .and(body_string_contains(r#""sinceState":"s1""#)) + .respond_with(ResponseTemplate::new(200).set_body_json(jmap_response( + "Email/changes", + serde_json::json!({ + "oldState": "s1", + "newState": "s2", + "hasMoreChanges": false, + "created": ["e3"], + "updated": [], + "destroyed": [] + }), + ))) + .mount(&mock_server) + .await; + + let client = mock_client(&mock_server.uri()); + let changes = client.email_changes("s0").await.unwrap(); + + assert_eq!(changes.created, vec!["e1", "e2", "e3"]); + assert_eq!(changes.new_state, "s2"); + } + + #[tokio::test] + async fn test_email_changes_surfaces_cannot_calculate_changes() { + use wiremock::matchers::method; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let mock_server = MockServer::start().await; + + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(jmap_response( + "error", + serde_json::json!({ "type": "cannotCalculateChanges" }), + ))) + .mount(&mock_server) + .await; + + let client = mock_client(&mock_server.uri()); + let err = client.email_changes("ancient").await.unwrap_err(); + + // The watcher keys its resync off this type, so it has to survive the + // trip through parse_response intact. + assert!( + matches!(&err, Error::Jmap { error_type, .. } if error_type == "cannotCalculateChanges"), + "unexpected error: {err}" + ); + } + + #[tokio::test] + async fn test_get_email_summaries_skips_body_values() { + use wiremock::matchers::{body_string_contains, method}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let mock_server = MockServer::start().await; + + // Bodies are the expensive half of Email/get; a watcher fetching them + // by default would make every arrival cost a full document parse. + Mock::given(method("POST")) + .and(body_string_contains(r#""fetchTextBodyValues":false"#)) + .respond_with(ResponseTemplate::new(200).set_body_json(jmap_response( + "Email/get", + serde_json::json!({ + "state": "s1", + "list": [{ "id": "e1", "subject": "hi" }], + "notFound": [] + }), + ))) + .mount(&mock_server) + .await; + + let client = mock_client(&mock_server.uri()); + let emails = client + .get_email_summaries(&["e1".to_string()]) + .await + .unwrap(); + + assert_eq!(emails.len(), 1); + assert_eq!(emails[0].subject.as_deref(), Some("hi")); + } + + #[tokio::test] + async fn test_open_event_stream_fills_the_url_template() { + use wiremock::matchers::{header, method, path, query_param}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let mock_server = MockServer::start().await; + + // The session hands back a URI template; getting any placeholder wrong + // fails silently as "push just never fires". + Mock::given(method("GET")) + .and(path("/jmap/event-source/")) + .and(query_param("types", "Email")) + .and(query_param("closeafter", "no")) + .and(query_param("ping", "30")) + .and(header("Authorization", "Bearer test-token")) + .and(header("Last-Event-ID", "evt-7")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("Content-Type", "text/event-stream") + .set_body_string(": ping\n\n"), + ) + .mount(&mock_server) + .await; + + let mut client = mock_client(&mock_server.uri()); + client.session.as_mut().unwrap().event_source_url = Some(format!( + "{}/jmap/event-source/?types={{types}}&closeafter={{closeafter}}&ping={{ping}}", + mock_server.uri() + )); + + let resp = client.open_event_stream(30, Some("evt-7")).await.unwrap(); + assert!(resp.status().is_success()); + } + + #[tokio::test] + async fn test_open_event_stream_without_a_url_points_at_poll() { + // create_test_session advertises no eventSourceUrl, standing in for a + // server that does not offer push. + let client = mock_client("https://api.example.com"); + let err = client.open_event_stream(30, None).await.unwrap_err(); + assert!(err.to_string().contains("--poll"), "unhelpful error: {err}"); + } + #[tokio::test] async fn test_upload_blob_success() { use wiremock::matchers::{header, method}; diff --git a/src/main.rs b/src/main.rs index ffaf80d..d796d53 100644 --- a/src/main.rs +++ b/src/main.rs @@ -59,6 +59,21 @@ enum Commands { email_id: String, }, + /// Stream newly arrived emails as newline-delimited JSON, until interrupted + Watch { + /// Only report mail landing in this mailbox (name or role) + #[arg(short, long)] + mailbox: Option, + + /// Include bodies and attachment metadata, not just summaries + #[arg(long)] + full: bool, + + /// Check every N seconds instead of holding a push connection open + #[arg(long, value_name = "SECONDS")] + poll: Option, + }, + /// Get all emails in a thread/conversation Thread { /// Email ID (will fetch entire thread this email belongs to) @@ -528,6 +543,19 @@ async fn main() { Commands::Thread { email_id } => commands::get_thread(&email_id).await, + Commands::Watch { + mailbox, + full, + poll, + } => { + commands::watch(commands::WatchOptions { + mailbox, + full, + poll, + }) + .await + } + Commands::Search { text, from, diff --git a/src/models/mod.rs b/src/models/mod.rs index 37f1057..58c9b27 100644 --- a/src/models/mod.rs +++ b/src/models/mod.rs @@ -318,6 +318,20 @@ impl Output { Err(e) => eprintln!("{{\"success\":false,\"error\":\"Serialization failed: {e}\"}}"), } } + + /// One line, flushed: the newline-delimited form a stream consumer reads + /// incrementally. Same envelope as [`Self::print`], so `jq` filters written + /// against any other command still apply. + pub fn print_compact(&self) { + use std::io::Write; + match serde_json::to_string(self) { + Ok(json) => { + println!("{json}"); + let _ = std::io::stdout().flush(); + } + Err(e) => eprintln!("{{\"success\":false,\"error\":\"Serialization failed: {e}\"}}"), + } + } } #[cfg(test)] From 62732a1a510571f8369301dfa730a463c9d95cf7 Mon Sep 17 00:00:00 2001 From: James Cleveland Date: Fri, 28 Aug 2026 09:59:38 +0100 Subject: [PATCH 2/2] feat(graphql): stream arrivals as an `emails` subscription over SSE MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The watcher moves to `jmap::ArrivalWatcher` so the CLI command and the GraphQL subscription are two front ends on one implementation rather than two implementations that agree today. Cursor, backoff, reconnect and resync semantics are therefore identical by construction. Served at `/graphql/stream` over Server-Sent Events. SSE rather than WebSockets: the subscription is a server-to-client firehose, nothing is ever sent back up the socket, and SSE reconnects on its own. MCP gets nothing here on purpose — tools are request/response and a subscription never returns. The `graphql` tool's description now says so and points at the CLI and the HTTP surface instead. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_016ri9XAwMqKiHLrDyTuA3np --- CHANGELOG.md | 13 ++ Cargo.toml | 2 + README.md | 21 +- src/commands/watch.rs | 226 ++-------------------- src/jmap/mod.rs | 2 + src/jmap/watch.rs | 329 ++++++++++++++++++++++++++++++++ src/mcp/graphql/mod.rs | 8 +- src/mcp/graphql/subscription.rs | 104 ++++++++++ src/mcp/mod.rs | 106 +++++++--- 9 files changed, 573 insertions(+), 238 deletions(-) create mode 100644 src/jmap/watch.rs create mode 100644 src/mcp/graphql/subscription.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index d9a8085..5a8e85f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,19 @@ change and folder move as an arrival, which is not what a mail loop means by "new". +- **The `emails` GraphQL subscription**, at `/graphql/stream` over Server-Sent + Events when the HTTP surface is up. Same watcher as `fastmail watch`, so the + cursor semantics are identical rather than merely similar — the CLI and the + subscription are two front ends on one implementation, which is the only way + they stay that way. + + SSE rather than WebSockets: the subscription is a server-to-client firehose, + nothing is ever sent back up the socket, and SSE reconnects on its own. + + MCP deliberately has none of this. Tools are request/response and a + subscription never returns, so the `graphql` tool's description says so and + points at the other two ways to get it. + ## [3.4.0] - 2026-08-17 ### Added diff --git a/Cargo.toml b/Cargo.toml index bc07e7b..7c4b115 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -65,6 +65,8 @@ uuid = { version = "1", features = ["v4"] } [dev-dependencies] tempfile = "3" +# `test-util` for the paused clock, so backoff tests assert timing without waiting it out. +tokio = { version = "1", features = ["test-util"] } wiremock = "0.6" [profile.release] diff --git a/README.md b/README.md index e76f4aa..06fe034 100644 --- a/README.md +++ b/README.md @@ -487,7 +487,7 @@ Three independent surfaces, each opt-in, sharing one port (default | Flag | Serves | | ------------ | ----------------------------------------------------------- | | `--http` | MCP streamable-HTTP at `/mcp` | -| `--graphql` | plain GraphQL-over-HTTP at `/graphql` | +| `--graphql` | plain GraphQL-over-HTTP at `/graphql`, subscriptions at `/graphql/stream` | | `--graphiql` | the GraphiQL IDE at `/`, and its `/graphql` | | `--browser` | opens the IDE once the port is bound, implying `--graphiql` | @@ -508,6 +508,23 @@ MCP JSON-RPC, which it doesn't. That is why GraphiQL needs its own route rather than pointing at the MCP one. Both share the schema, the client cache and the credential resolution below, so the IDE sees exactly what a model sees. +`/graphql/stream` carries subscriptions over Server-Sent Events — POST the +operation, read events off the response: + +```bash +curl -N http://127.0.0.1:8080/graphql/stream \ + -H 'Content-Type: application/json' \ + -d '{"query":"subscription { emails(mailbox: \"inbox\") { id subject from { email } } }"}' +``` + +SSE rather than WebSockets because the only subscription is a server-to-client +firehose: nothing is ever sent back up the socket, and SSE reconnects on its +own. Body fields (`textBody`, `attachments`) want `full: true` — a subscription +has no request boundary at which the batching loaders reset, so the lazy path +resolves through a loader that lives as long as the subscription. `pollSeconds` +is the same fallback as the CLI's `--poll`. MCP has no equivalent: tools are +request/response, and a subscription never returns. + **Credential resolution is the same everywhere:** the request's own header wins, otherwise the local config (or the matching environment variable) is used. Running it yourself, that means your own credentials with no ceremony. In a @@ -787,6 +804,8 @@ size, surfaced in the field descriptions; it never refuses a query. Being told All operations are available as GraphQL queries and mutations: mailboxes, emails, search, threads, identities (with signatures), attachments (with text extraction and image resizing), contacts, masked email management, and send/reply/forward with the preview/confirm safety pattern. +One subscription, `emails`, streams arrivals over the same machinery as `fastmail watch`. + Token can be set via `FASTMAIL_API_TOKEN` env var or config file. ## Debug Logging diff --git a/src/commands/watch.rs b/src/commands/watch.rs index 3bdb0d2..b426a80 100644 --- a/src/commands/watch.rs +++ b/src/commands/watch.rs @@ -1,18 +1,6 @@ -use crate::error::Error; -use crate::jmap::{EventParser, JmapClient, authenticated_client}; -use crate::models::{Email, Output}; +use crate::jmap::{ArrivalWatcher, authenticated_client}; +use crate::models::Output; use std::time::Duration; -use tracing::debug; - -/// How often the server should send a keep-alive on the push channel. Also sets -/// the read timeout that detects a connection which died without saying so. -const PING_SECONDS: u32 = 30; - -/// Reconnect backoff bounds. The ceiling is deliberately below the ping -/// interval's own scale: a watcher that goes quiet for minutes after a blip is -/// indistinguishable from a broken one. -const BACKOFF_START: u64 = 1; -const BACKOFF_MAX: u64 = 30; pub struct WatchOptions { /// Only report mail landing in this mailbox, by name or role. @@ -25,213 +13,29 @@ pub struct WatchOptions { /// Stream newly arrived emails as newline-delimited JSON, one object per line, /// until interrupted. -/// -/// Arrivals are discovered through `Email/changes` against a state cursor this -/// function owns. The push channel only ever says *look again* — so a dropped -/// connection, a missed event or a `--poll` fallback all converge on the same -/// answer, and the cost of losing a notification is latency rather than mail. pub async fn watch(opts: WatchOptions) -> anyhow::Result<()> { - let mut client = authenticated_client().await?; - - let mailbox_id = match opts.mailbox { - Some(ref name) => Some(client.find_mailbox(name).await?.id), - None => None, - }; + let client = std::sync::Arc::new(tokio::sync::Mutex::new(authenticated_client().await?)); - // Start from now: the caller asked what arrives next, not what is already - // sitting there. - let mut state = client.email_state().await?; - - match opts.poll { - Some(seconds) => { - poll_loop( - &client, - &mut state, - mailbox_id.as_deref(), - opts.full, - seconds, - ) - .await - } - None => push_loop(&client, &mut state, mailbox_id.as_deref(), opts.full).await, - } -} - -async fn push_loop( - client: &JmapClient, - state: &mut String, - mailbox_id: Option<&str>, - full: bool, -) -> anyhow::Result<()> { - let mut backoff = BACKOFF_START; - let mut last_event_id: Option = None; + let mut watcher = ArrivalWatcher::new( + client, + opts.mailbox.as_deref(), + opts.full, + opts.poll.map(Duration::from_secs), + ) + .await?; loop { - match client - .open_event_stream(PING_SECONDS, last_event_id.as_deref()) - .await - { - Ok(mut resp) => { - let mut parser = EventParser::default(); - loop { - match resp.chunk().await { - Ok(Some(bytes)) => { - // Reset only once the connection has actually - // carried something. Resetting on connect alone - // would let a server that accepts and immediately - // hangs up spin at full rate. - backoff = BACKOFF_START; - - for event in parser.feed(&String::from_utf8_lossy(&bytes)) { - if let Some(id) = event.id { - last_event_id = Some(id); - } - // Keep-alives carry no state change. - if event.event.as_deref() == Some("ping") || event.data.is_empty() { - continue; - } - drain(client, state, mailbox_id, full).await?; - } - } - Ok(None) => { - debug!("Event source closed by server"); - break; - } - Err(e) => { - eprintln!("watch: event stream dropped ({e}); reconnecting"); - break; - } - } - } - } - Err(e) if fatal(&e) => return Err(e.into()), - Err(e) => eprintln!("watch: could not open event stream ({e}); retrying"), - } - - tokio::time::sleep(Duration::from_secs(backoff)).await; - backoff = (backoff * 2).min(BACKOFF_MAX); + let arrivals = watcher.next_arrivals().await?; - // Reconcile across the gap before waiting on the next push, so mail - // that landed while disconnected is reported on reconnect rather than - // on whatever arrives after it. - drain(client, state, mailbox_id, full).await?; - } -} - -async fn poll_loop( - client: &JmapClient, - state: &mut String, - mailbox_id: Option<&str>, - full: bool, - seconds: u64, -) -> anyhow::Result<()> { - loop { - tokio::time::sleep(Duration::from_secs(seconds)).await; - drain(client, state, mailbox_id, full).await?; - } -} - -/// Advance the cursor and print whatever arrived, oldest first. -/// -/// Transient failures are reported and swallowed: a watcher that exits on one -/// bad response is worse than useless in the loop it is meant to feed. A dead -/// credential is not transient, so it still ends the process. -async fn drain( - client: &JmapClient, - state: &mut String, - mailbox_id: Option<&str>, - full: bool, -) -> anyhow::Result<()> { - let changes = match client.email_changes(state).await { - Ok(changes) => changes, - Err(Error::Jmap { ref error_type, .. }) if error_type == "cannotCalculateChanges" => { - // The server has discarded history back to our cursor. There is no - // way to know what was missed, and replaying the mailbox as "new" - // would be a lie, so resync to now and say so. - *state = client.email_state().await?; + // Mail may have been lost, and stdout is reserved for mail that wasn't. + if arrivals.resynced { eprintln!( "watch: server dropped change history; resynced, some arrivals may be missing" ); - return Ok(()); - } - Err(e) if fatal(&e) => return Err(e.into()), - Err(e) => { - eprintln!("watch: could not read changes ({e})"); - return Ok(()); } - }; - - *state = changes.new_state; - if changes.created.is_empty() { - return Ok(()); - } - let fetched = if full { - client.get_emails(&changes.created).await - } else { - client.get_email_summaries(&changes.created).await - }; - - let mut emails = match fetched { - Ok(emails) => emails, - Err(e) if fatal(&e) => return Err(e.into()), - Err(e) => { - eprintln!( - "watch: could not fetch {} new email(s) ({e})", - changes.created.len() - ); - return Ok(()); + for email in &arrivals.emails { + Output::success(email).print_compact(); } - }; - - // `Email/get` makes no ordering guarantee, and a stream reads chronologically. - emails.sort_by(|a, b| a.received_at.cmp(&b.received_at)); - - for email in emails.iter().filter(|e| in_mailbox(e, mailbox_id)) { - Output::success(email).print_compact(); - } - - Ok(()) -} - -fn in_mailbox(email: &Email, mailbox_id: Option<&str>) -> bool { - mailbox_id.is_none_or(|id| email.mailbox_ids.contains_key(id)) -} - -/// Whether an error means the watcher can never succeed again. -fn fatal(e: &Error) -> bool { - matches!(e, Error::InvalidToken(_) | Error::NotAuthenticated) -} - -#[cfg(test)] -mod tests { - use super::*; - - fn email_in(mailbox_ids: &[&str]) -> Email { - let ids: serde_json::Map<_, _> = mailbox_ids - .iter() - .map(|id| (id.to_string(), serde_json::json!(true))) - .collect(); - serde_json::from_value(serde_json::json!({ "id": "e1", "mailboxIds": ids })).unwrap() - } - - #[test] - fn no_mailbox_filter_accepts_everything() { - assert!(in_mailbox(&email_in(&["archive"]), None)); - } - - #[test] - fn mailbox_filter_matches_membership() { - let email = email_in(&["inbox", "important"]); - assert!(in_mailbox(&email, Some("inbox"))); - assert!(!in_mailbox(&email, Some("trash"))); - } - - #[test] - fn a_dead_credential_is_fatal_but_a_bad_response_is_not() { - assert!(fatal(&Error::NotAuthenticated)); - assert!(fatal(&Error::InvalidToken("nope"))); - assert!(!fatal(&Error::RateLimited)); - assert!(!fatal(&Error::Server("boom".into()))); } } diff --git a/src/jmap/mod.rs b/src/jmap/mod.rs index bb703ae..fa8d3df 100644 --- a/src/jmap/mod.rs +++ b/src/jmap/mod.rs @@ -1,4 +1,5 @@ mod events; +mod watch; use crate::commands::SearchFilter; use crate::error::{Error, Result}; @@ -11,6 +12,7 @@ use std::time::Duration; use tracing::{debug, instrument}; pub use events::{EventParser, ServerEvent}; +pub use watch::{ArrivalWatcher, Arrivals, SharedJmapClient}; const SESSION_URL: &str = "https://api.fastmail.com/jmap/session"; const TIMEOUT: Duration = Duration::from_secs(30); diff --git a/src/jmap/watch.rs b/src/jmap/watch.rs new file mode 100644 index 0000000..3e8efdb --- /dev/null +++ b/src/jmap/watch.rs @@ -0,0 +1,329 @@ +//! Watching for newly arrived mail. +//! +//! The cursor is the point: JMAP's push channel says only *something changed*, +//! so this holds an `Email` state string and asks `Email/changes` what that +//! actually was. Push, polling and reconnect-after-a-drop all funnel through +//! the same call, which is why a lost notification costs latency rather than +//! mail — and why `--poll` is a timer swapped for a socket, not a second +//! implementation. + +use crate::error::{Error, Result}; +use crate::jmap::{EventParser, JmapClient}; +use crate::models::Email; +use std::time::Duration; +use tracing::debug; + +/// A client shared between a watcher and whatever else is using the connection. +/// +/// The lock is held only across individual JMAP calls, never across a read of +/// the push channel — that read blocks until mail arrives, which on a quiet +/// account is hours. +pub type SharedJmapClient = std::sync::Arc>; + +/// How often to ask the server for a keep-alive, and the basis for the read +/// timeout that notices a connection which died without saying so. +const PING_SECONDS: u32 = 30; + +/// Reconnect backoff bounds. The ceiling is deliberately small: a watcher that +/// goes quiet for minutes after a blip is indistinguishable from a broken one. +const BACKOFF_START: u64 = 1; +const BACKOFF_MAX: u64 = 30; + +/// What one wake-up turned up. +pub struct Arrivals { + /// New emails, oldest first. Empty is normal — a wake-up is a prompt to + /// look, not a promise of mail. + pub emails: Vec, + /// The server had discarded change history past our cursor, so it was reset + /// to the present. Anything that arrived in the gap was never reported and + /// now never will be. + pub resynced: bool, +} + +/// How the watcher learns it should look again. +enum Wake { + Push { + /// The live response, or `None` when it needs (re)opening. + stream: Option, + parser: EventParser, + backoff: u64, + last_event_id: Option, + }, + Poll(Duration), +} + +pub struct ArrivalWatcher { + client: SharedJmapClient, + state: String, + mailbox_id: Option, + full: bool, + wake: Wake, +} + +impl ArrivalWatcher { + /// Start watching from the present moment. + /// + /// `poll` swaps the push connection for a timer of that interval; without + /// it the watcher holds JMAP's event source open. + pub async fn new( + client: SharedJmapClient, + mailbox: Option<&str>, + full: bool, + poll: Option, + ) -> Result { + let (mailbox_id, state) = { + let mut locked = client.lock().await; + let mailbox_id = match mailbox { + Some(name) => Some(locked.find_mailbox(name).await?.id), + None => None, + }; + // Start from now: the caller asked what arrives next, not what is + // already sitting there. + (mailbox_id, locked.email_state().await?) + }; + + Ok(Self { + client, + state, + mailbox_id, + full, + wake: match poll { + Some(interval) => Wake::Poll(interval), + None => Wake::Push { + stream: None, + parser: EventParser::default(), + backoff: BACKOFF_START, + last_event_id: None, + }, + }, + }) + } + + /// Block until the next wake-up, then report what arrived. + /// + /// Errors are returned only when the watcher can never succeed again — a + /// dead credential. Everything else is transient by nature (a dropped + /// connection, a bad response, a server that has forgotten our cursor) and + /// is retried internally with backoff, because a watcher that exits on one + /// blip is useless in the loop it exists to feed. + pub async fn next_arrivals(&mut self) -> Result { + self.wait().await?; + self.drain().await + } + + /// Wait until there is reason to believe something changed. + async fn wait(&mut self) -> Result<()> { + let interval = match &mut self.wake { + Wake::Poll(interval) => *interval, + Wake::Push { .. } => return self.wait_for_push().await, + }; + tokio::time::sleep(interval).await; + Ok(()) + } + + async fn wait_for_push(&mut self) -> Result<()> { + loop { + let Wake::Push { + stream, + parser, + backoff, + last_event_id, + } = &mut self.wake + else { + unreachable!("wait_for_push is only entered for Wake::Push") + }; + + if stream.is_none() { + // Opened under the lock, read outside it: the read blocks for + // as long as the account is quiet. + let opened = { + let client = self.client.lock().await; + client + .open_event_stream(PING_SECONDS, last_event_id.as_deref()) + .await + }; + match opened { + Ok(resp) => *stream = Some(resp), + Err(e) if is_fatal(&e) => return Err(e), + Err(e) => { + debug!("Could not open event stream ({e}); retrying"); + sleep_backoff(backoff).await; + // Reconcile anyway: while push is down, each retry is + // also a poll tick, so mail still surfaces. + return Ok(()); + } + } + } + + let chunk = stream + .as_mut() + .expect("stream was just opened") + .chunk() + .await; + + match chunk { + Ok(Some(bytes)) => { + // Reset only once the connection has carried something. + // Resetting on connect alone would let a server that + // accepts and immediately hangs up spin at full rate. + *backoff = BACKOFF_START; + + let mut changed = false; + for event in parser.feed(&String::from_utf8_lossy(&bytes)) { + if let Some(id) = event.id { + *last_event_id = Some(id); + } + // Keep-alives carry no state change. + if event.event.as_deref() == Some("ping") || event.data.is_empty() { + continue; + } + changed = true; + } + if changed { + return Ok(()); + } + } + Ok(None) | Err(_) => { + debug!("Event stream ended; reconnecting"); + *stream = None; + sleep_backoff(backoff).await; + // Reconcile across the gap before waiting again, so mail + // that landed while disconnected is reported now rather + // than whenever the next message happens to arrive. + return Ok(()); + } + } + } + } + + /// Advance the cursor and collect whatever it turned up. + async fn drain(&mut self) -> Result { + let client = self.client.lock().await; + + let changes = match client.email_changes(&self.state).await { + Ok(changes) => changes, + Err(Error::Jmap { ref error_type, .. }) if error_type == "cannotCalculateChanges" => { + // History is gone back to our cursor. There is no way to know + // what was missed, and replaying the mailbox as "new" would be + // a lie, so reset to the present and say so. + self.state = client.email_state().await?; + return Ok(Arrivals { + emails: Vec::new(), + resynced: true, + }); + } + Err(e) if is_fatal(&e) => return Err(e), + Err(e) => { + debug!("Could not read changes ({e})"); + return Ok(Arrivals::none()); + } + }; + + self.state = changes.new_state; + if changes.created.is_empty() { + return Ok(Arrivals::none()); + } + + let fetched = if self.full { + client.get_emails(&changes.created).await + } else { + client.get_email_summaries(&changes.created).await + }; + + let mut emails = match fetched { + Ok(emails) => emails, + Err(e) if is_fatal(&e) => return Err(e), + Err(e) => { + debug!( + "Could not fetch {} new email(s) ({e})", + changes.created.len() + ); + return Ok(Arrivals::none()); + } + }; + + emails.retain(|email| in_mailbox(email, self.mailbox_id.as_deref())); + // `Email/get` makes no ordering guarantee, and a stream reads + // chronologically. + emails.sort_by(|a, b| a.received_at.cmp(&b.received_at)); + + Ok(Arrivals { + emails, + resynced: false, + }) + } +} + +impl Arrivals { + fn none() -> Self { + Self { + emails: Vec::new(), + resynced: false, + } + } +} + +async fn sleep_backoff(backoff: &mut u64) { + tokio::time::sleep(Duration::from_secs(*backoff)).await; + *backoff = (*backoff * 2).min(BACKOFF_MAX); +} + +fn in_mailbox(email: &Email, mailbox_id: Option<&str>) -> bool { + mailbox_id.is_none_or(|id| email.mailbox_ids.contains_key(id)) +} + +/// Whether an error means the watcher can never succeed again. +fn is_fatal(e: &Error) -> bool { + matches!(e, Error::InvalidToken(_) | Error::NotAuthenticated) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn email_in(mailbox_ids: &[&str]) -> Email { + let ids: serde_json::Map<_, _> = mailbox_ids + .iter() + .map(|id| (id.to_string(), serde_json::json!(true))) + .collect(); + serde_json::from_value(serde_json::json!({ "id": "e1", "mailboxIds": ids })).unwrap() + } + + #[test] + fn no_mailbox_filter_accepts_everything() { + assert!(in_mailbox(&email_in(&["archive"]), None)); + } + + #[test] + fn mailbox_filter_matches_membership() { + let email = email_in(&["inbox", "important"]); + assert!(in_mailbox(&email, Some("inbox"))); + assert!(!in_mailbox(&email, Some("trash"))); + } + + #[test] + fn a_dead_credential_is_fatal_but_a_bad_response_is_not() { + assert!(is_fatal(&Error::NotAuthenticated)); + assert!(is_fatal(&Error::InvalidToken("nope"))); + assert!(!is_fatal(&Error::RateLimited)); + assert!(!is_fatal(&Error::Server("boom".into()))); + } + + // Paused clock: the sleeps are the point of the function, but waiting out + // half a minute of them is not. + #[tokio::test(start_paused = true)] + async fn backoff_doubles_up_to_the_ceiling() { + let mut backoff = BACKOFF_START; + sleep_backoff(&mut backoff).await; + assert_eq!(backoff, BACKOFF_START * 2); + + let mut backoff = BACKOFF_MAX - 1; + sleep_backoff(&mut backoff).await; + assert_eq!(backoff, BACKOFF_MAX); + sleep_backoff(&mut backoff).await; + assert_eq!( + backoff, BACKOFF_MAX, + "backoff must not grow past the ceiling" + ); + } +} diff --git a/src/mcp/graphql/mod.rs b/src/mcp/graphql/mod.rs index 5f398df..cea5045 100644 --- a/src/mcp/graphql/mod.rs +++ b/src/mcp/graphql/mod.rs @@ -10,19 +10,21 @@ pub mod filter; pub mod loaders; mod mutation; mod query; +mod subscription; #[cfg(test)] mod tests; pub mod types; use mutation::MutationRoot; use query::QueryRoot; +use subscription::SubscriptionRoot; -pub type FastmailSchema = Schema; +pub type FastmailSchema = Schema; /// The per-request JMAP client, injected into each GraphQL execution as request /// data. Shared (`Arc`) so an authenticated client can be reused across requests /// for the same Fastmail token rather than re-authenticating every call. -pub type SharedClient = std::sync::Arc>; +pub type SharedClient = crate::jmap::SharedJmapClient; /// The credentials `contacts` needs. /// @@ -87,7 +89,7 @@ pub fn build_schema() -> FastmailSchema { // // Depth stays capped: the graph contains cycles, and nothing else bounds // them. - Schema::build(QueryRoot, MutationRoot, async_graphql::EmptySubscription) + Schema::build(QueryRoot, MutationRoot, SubscriptionRoot) .data(types::NonceStore::default()) .limit_depth(MAX_DEPTH) .finish() diff --git a/src/mcp/graphql/subscription.rs b/src/mcp/graphql/subscription.rs new file mode 100644 index 0000000..53bf4e0 --- /dev/null +++ b/src/mcp/graphql/subscription.rs @@ -0,0 +1,104 @@ +//! GraphQL subscription resolvers. +//! +//! One subscription: mail as it arrives. It runs on the same +//! [`ArrivalWatcher`](crate::jmap::ArrivalWatcher) as `fastmail watch`, so the +//! cursor semantics are identical — push is a wake-up, `Email/changes` is the +//! answer, and a dropped connection costs latency rather than mail. + +use async_graphql::futures_util::stream::{self, Stream, StreamExt}; +use async_graphql::{Context, Result, Subscription}; +use std::time::Duration; + +use super::SharedClient; +use super::loaders::to_gql_error; +use super::types::GqlEmail; +use crate::jmap::ArrivalWatcher; +use std::sync::Arc; + +pub struct SubscriptionRoot; + +/// Whether the watcher is still worth polling. A fatal error yields once and +/// then closes the stream — there is nothing to retry with a dead credential. +enum Watching { + Live(Box), + Done, +} + +#[Subscription] +impl SubscriptionRoot { + /// Emits each email as it arrives, indefinitely. + /// + /// Backed by JMAP's push channel, with the state cursor held server-side + /// here rather than by the subscriber: a reconnect or a missed + /// notification is reconciled through `Email/changes`, so the subscription + /// reports late rather than losing mail. Only *new* messages are emitted — + /// flag and folder changes to existing mail are not arrivals. + /// + /// Transient failures are retried internally and never reach the + /// subscriber. The stream ends only when the token stops authenticating. + /// + /// Set `full` to select body and attachment fields. Unlike a query, a + /// subscription has no request boundary at which the batching loaders + /// reset, so leaving it off and selecting `textBody` anyway resolves each + /// email through a loader that lives as long as the subscription. + async fn emails( + &self, + ctx: &Context<'_>, + #[graphql(desc = "Only emit mail landing in this mailbox, by name or role.")] + mailbox: Option, + #[graphql( + default = false, + desc = "Fetch bodies and attachment metadata with each arrival." + )] + full: bool, + #[graphql( + desc = "Check on this interval instead of holding a push connection open. For \ + networks that will not keep one alive; the results are the same." + )] + poll_seconds: Option, + ) -> Result> + use<>> { + let client = ctx.data::()?.clone(); + + let watcher = ArrivalWatcher::new( + client, + mailbox.as_deref(), + full, + poll_seconds.map(Duration::from_secs), + ) + .await + .map_err(|e| to_gql_error(Arc::new(e)))?; + + let batches = stream::unfold(Watching::Live(Box::new(watcher)), |state| async move { + let Watching::Live(mut watcher) = state else { + return None; + }; + loop { + match watcher.next_arrivals().await { + // A wake-up that turned up nothing is normal — keep + // waiting rather than emitting an empty tick. + Ok(arrivals) if arrivals.emails.is_empty() => continue, + Ok(arrivals) => { + return Some((Ok(arrivals.emails), Watching::Live(watcher))); + } + Err(e) => return Some((Err(to_gql_error(Arc::new(e))), Watching::Done)), + } + } + }); + + Ok(batches.flat_map(move |batch| match batch { + Ok(emails) => stream::iter( + emails + .into_iter() + .map(|email| { + Ok(if full { + GqlEmail::full(email) + } else { + GqlEmail::summary(email) + }) + }) + .collect::>(), + ), + Err(e) => stream::iter(vec![Err(e)]), + })) + } +} diff --git a/src/mcp/mod.rs b/src/mcp/mod.rs index 28753d0..74d5b37 100644 --- a/src/mcp/mod.rs +++ b/src/mcp/mod.rs @@ -266,6 +266,11 @@ mutation { sendEmail(action: PREVIEW, to: \"a@b.com\", subject: \"Hi\", body: \" mutation { sendEmail(action: CONFIRM, to: \"a@b.com\", subject: \"Hi\", body: \"...\", confirmationToken: \"\") { success emailId } } ``` +SUBSCRIPTIONS are not available through this tool — it is request/response, and +a subscription never returns. The schema defines one (`emails`, streaming mail +as it arrives) for callers on the HTTP surface, which serves it over SSE at +`/graphql/stream`; `fastmail watch` is the same thing as a CLI. + NOT LISTED ABOVE — ask `schema_sdl` for these rather than guessing: attachment payloads (base64/image/text), masked email, contacts and contact CRUD (CardDAV, so check `session { carddavConfigured }` first), identities, moveEmail, @@ -424,20 +429,17 @@ fn is_introspection_only(query: &str) -> bool { }) } -/// Plain GraphQL-over-HTTP, for browsers and anything else that speaks it -/// directly rather than through MCP's JSON-RPC envelope. Shares the server's -/// schema, client cache and token resolution with the `graphql` tool. -async fn graphql_endpoint( - axum::extract::State(mcp): axum::extract::State, - headers: http::HeaderMap, - axum::Json(req): axum::Json, -) -> axum::Json { - let error = |msg: String| { - axum::Json(async_graphql::Response::from_errors(vec![ - async_graphql::ServerError::new(msg, None), - ])) - }; - +/// Resolve credentials and build the GraphQL request an HTTP body describes. +/// +/// Shared by the query and subscription endpoints so they cannot drift on which +/// token wins or what counts as introspection. The error case is a message for +/// the caller, not a status code — GraphQL reports its own failures in the +/// response body. +async fn build_http_request( + mcp: &FastmailMcp, + headers: &http::HeaderMap, + req: HttpGraphqlRequest, +) -> std::result::Result { // Introspection is answered from the schema, so it neither needs a token nor // touches the network — the IDE stays usable while credentials are wrong. let mut request = if is_introspection_only(&req.query) { @@ -446,8 +448,8 @@ async fn graphql_endpoint( // Must keep honouring `mcp.default_token` here: GraphiQL runs in a // browser and cannot attach the token header, so making this // headers-only breaks local development. - let Some(token) = resolve_token(Some(&headers), mcp.default_token.as_deref()) else { - return error(format!( + let Some(token) = resolve_token(Some(headers), mcp.default_token.as_deref()) else { + return Err(format!( "No Fastmail token available. Configure one via `fastmail auth` \ or send the {TOKEN_HEADER} header." )); @@ -455,14 +457,13 @@ async fn graphql_endpoint( // Authenticated on first use rather than at startup, so a missing or // expired token surfaces in the response pane instead of stopping the // server booting. - let client = match client_for(&mcp.clients, &token).await { - Ok(client) => client, - Err(e) => return error(format!("Fastmail authentication failed: {e}")), - }; + let client = client_for(&mcp.clients, &token) + .await + .map_err(|e| format!("Fastmail authentication failed: {e}"))?; graphql::request( &req.query, client, - resolve_carddav(Some(&headers), &mcp.default_carddav), + resolve_carddav(Some(headers), &mcp.default_carddav), ) }; if let Some(vars) = req.variables { @@ -471,7 +472,60 @@ async fn graphql_endpoint( if let Some(name) = req.operation_name { request = request.operation_name(name); } - axum::Json(mcp.schema.execute(request).await) + Ok(request) +} + +/// Plain GraphQL-over-HTTP, for browsers and anything else that speaks it +/// directly rather than through MCP's JSON-RPC envelope. Shares the server's +/// schema, client cache and token resolution with the `graphql` tool. +async fn graphql_endpoint( + axum::extract::State(mcp): axum::extract::State, + headers: http::HeaderMap, + axum::Json(req): axum::Json, +) -> axum::Json { + match build_http_request(&mcp, &headers, req).await { + Ok(request) => axum::Json(mcp.schema.execute(request).await), + Err(msg) => axum::Json(async_graphql::Response::from_errors(vec![ + async_graphql::ServerError::new(msg, None), + ])), + } +} + +/// GraphQL subscriptions over Server-Sent Events, one event per response. +/// +/// SSE rather than WebSockets because the only subscription here is a +/// server-to-client firehose: nothing is ever sent back up the socket, and SSE +/// reconnects on its own. It is also the same shape the CLI consumes from +/// Fastmail, which keeps one mental model for the whole path. +async fn graphql_stream_endpoint( + axum::extract::State(mcp): axum::extract::State, + headers: http::HeaderMap, + axum::Json(req): axum::Json, +) -> axum::response::Response { + use async_graphql::futures_util::stream::StreamExt; + use axum::response::{IntoResponse, Sse, sse}; + + let request = match build_http_request(&mcp, &headers, req).await { + Ok(request) => request, + Err(msg) => { + return axum::Json(async_graphql::Response::from_errors(vec![ + async_graphql::ServerError::new(msg, None), + ])) + .into_response(); + } + }; + + let events = mcp.schema.execute_stream(request).map(|response| { + let data = serde_json::to_string(&response) + .unwrap_or_else(|e| format!(r#"{{"errors":[{{"message":"{e}"}}]}}"#)); + Ok::<_, std::convert::Infallible>(sse::Event::default().data(data)) + }); + + // Proxies drop connections that go quiet, and a mail subscription is quiet + // most of the time. + Sse::new(events) + .keep_alive(sse::KeepAlive::default()) + .into_response() } /// Which surfaces [`run_http_server`] mounts alongside MCP at `/mcp`. @@ -522,8 +576,14 @@ pub async fn run_http_server(addr: &str, surfaces: HttpSurfaces) -> anyhow::Resu tracing::info!("MCP streamable-HTTP listening on http://{addr}/mcp"); if surfaces.graphql || surfaces.graphiql { - router = router.route("/graphql", axum::routing::post(graphql_endpoint)); + router = router + .route("/graphql", axum::routing::post(graphql_endpoint)) + .route( + "/graphql/stream", + axum::routing::post(graphql_stream_endpoint), + ); tracing::info!("GraphQL endpoint on http://{addr}/graphql"); + tracing::info!("GraphQL subscriptions (SSE) on http://{addr}/graphql/stream"); } if surfaces.graphiql {