diff --git a/README.md b/README.md index 33ecc6a..0e7f0c8 100644 --- a/README.md +++ b/README.md @@ -97,6 +97,8 @@ void gmail draft create --reply-to --subject "Re: Q3" --body "LGTM, approve void gmail batch-modify --remove INBOX ``` +`void inbox` is the Inbox Zero surface for Gmail's `INBOX` label: after every sync it reconciles its archive state with Gmail, so mail you archive in the Gmail web UI disappears from `void inbox` (and vice versa). It lists one item per **thread** — the thread's latest unarchived message — while `void gmail search 'in:inbox'` lists individual **messages**; archiving the shown item removes that message's `INBOX` label and the thread's next message surfaces until the thread is fully out of Gmail's inbox. + ### Calendar Docs: [commands](docs/commands.md#calendar) diff --git a/crates/void-gmail/src/api/client.rs b/crates/void-gmail/src/api/client.rs index b215bf2..368a123 100644 --- a/crates/void-gmail/src/api/client.rs +++ b/crates/void-gmail/src/api/client.rs @@ -89,15 +89,15 @@ impl GmailApiClient { if let Some(q) = query { params.push(("q", q.to_string())); } - let resp: MessageListResponse = self + let resp = self .http .get(format!("{}/gmail/v1/users/me/messages", self.base_url)) .bearer_auth(&self.access_token) .query(¶ms) .send() .await? - .json() - .await?; + .error_for_status()?; + let resp: MessageListResponse = resp.json().await?; let count = resp.messages.as_ref().map(|m| m.len()).unwrap_or(0); debug!( message_count = count, @@ -109,7 +109,7 @@ impl GmailApiClient { pub async fn get_message(&self, message_id: &str) -> Result { debug!(message_id, "gmail: get_message"); - let resp: GmailMessage = self + let resp = self .http .get(format!( "{}/gmail/v1/users/me/messages/{message_id}", @@ -119,8 +119,8 @@ impl GmailApiClient { .query(&[("format", "full")]) .send() .await? - .json() - .await?; + .error_for_status()?; + let resp: GmailMessage = resp.json().await?; Ok(resp) } @@ -143,23 +143,29 @@ impl GmailApiClient { if let Some(pt) = &page_token { params.push(("pageToken", pt.clone())); } - let resp: HistoryListResponse = self + let resp = self .http .get(format!("{}/gmail/v1/users/me/history", self.base_url)) .bearer_auth(&self.access_token) .query(¶ms) .send() - .await? - .json() .await?; + // Gmail returns 404 once the startHistoryId is too old (history is + // only kept for a limited window). Surface that distinctly so the + // sync loop can fall back to a full INBOX refresh. + if resp.status() == reqwest::StatusCode::NOT_FOUND { + return Err(GmailError::HistoryExpired); + } + let resp = resp.error_for_status()?; + let page_resp: HistoryListResponse = resp.json().await?; - if let Some(records) = resp.history { + if let Some(records) = page_resp.history { let count = records.len(); all_records.extend(records); debug!(page, record_count = count, "gmail: listed history page"); } - latest_history_id = resp.history_id.or(latest_history_id); - page_token = resp.next_page_token; + latest_history_id = page_resp.history_id.or(latest_history_id); + page_token = page_resp.next_page_token; if page_token.is_none() { break; } diff --git a/crates/void-gmail/src/connector/sync.rs b/crates/void-gmail/src/connector/sync.rs index 806c517..8836701 100644 --- a/crates/void-gmail/src/connector/sync.rs +++ b/crates/void-gmail/src/connector/sync.rs @@ -5,11 +5,17 @@ use tracing::{debug, info, warn}; use void_core::db::Database; use void_core::models::{Conversation, ConversationKind, Message}; -use crate::api::GmailMessage; +use crate::api::{GmailApiClient, GmailMessage}; +use crate::error::GmailError; use super::compose::{html_to_markdown, looks_like_html, parse_email_address, parse_email_name}; use super::GmailConnector; +/// Page size for INBOX listings (Gmail API allows up to 500). +const INBOX_PAGE_SIZE: u32 = 500; +/// Safety cap on INBOX listing pages (20 × 500 = 10 000 messages). +const INBOX_MAX_PAGES: u32 = 20; + impl GmailConnector { pub(crate) async fn initial_sync(&self, db: &Database) -> anyhow::Result<()> { let api = self.get_client().await?; @@ -23,7 +29,7 @@ impl GmailConnector { if had_history { debug!(config_id = %self.config_id, "history_id exists, refreshing inbox state"); - self.refresh_inbox(db).await?; + self.refresh_inbox_with_api(db, &api).await?; return Ok(()); } @@ -33,71 +39,42 @@ impl GmailConnector { info!(config_id = %self.config_id, "starting Gmail initial sync"); - let mut page_token: Option = None; - let max_pages: u64 = 5; - - let mut progress = void_core::progress::BackfillProgress::new( - &format!("gmail:{}", self.config_id), - "messages", - ); - progress.set_pages(max_pages); - - loop { - let resp = api - .list_messages( - 100, - page_token.as_deref(), - Some(&["INBOX"]), - Some("newer_than:7d"), - ) - .await?; - progress.inc_page(); - - if let Some(msgs) = resp.messages { - for msg_ref in &msgs { - match api.get_message(&msg_ref.id).await { - Ok(msg) => { - self.store_message(db, &msg)?; - progress.inc(1); - } - Err(e) => { - warn!(message_id = %msg_ref.id, "failed to fetch message: {e}"); - } - } - } - } - - page_token = resp.next_page_token; - if page_token.is_none() || progress.pages_done >= max_pages { - break; - } - } + self.refresh_inbox_with_api(db, &api).await?; - progress.finish(); - info!(config_id = %self.config_id, messages = progress.items, "Gmail initial sync complete"); + info!(config_id = %self.config_id, "Gmail initial sync complete"); Ok(()) } - /// Refresh inbox state: fetch current INBOX message IDs from Gmail and - /// reconcile `is_archived` in the local DB so it mirrors Gmail exactly. - /// Also fetches any new INBOX messages not yet in the local DB. + /// Refresh inbox state: fetch the *complete* INBOX message ID list from + /// Gmail (no date filter) and reconcile `is_archived` in the local DB so + /// it mirrors Gmail exactly. Also fetches full bodies for any INBOX + /// messages not yet in the local DB, so `void inbox` matches + /// `gmail search 'in:inbox'`. pub(crate) async fn refresh_inbox(&self, db: &Database) -> anyhow::Result<()> { let api = self.get_client().await?; + self.refresh_inbox_with_api(db, &api).await + } + + pub(crate) async fn refresh_inbox_with_api( + &self, + db: &Database, + api: &GmailApiClient, + ) -> anyhow::Result<()> { let connection_id = self.display_connection_id(); let mut inbox_ids: HashSet = HashSet::new(); let mut new_msg_ids: Vec = Vec::new(); let mut page_token: Option = None; - let max_pages = 5u32; + let mut truncated = false; let mut pages = 0u32; loop { let resp = api .list_messages( - 100, + INBOX_PAGE_SIZE, page_token.as_deref(), Some(&["INBOX"]), - Some("newer_than:7d"), + None, ) .await?; pages += 1; @@ -112,7 +89,16 @@ impl GmailConnector { } page_token = resp.next_page_token; - if page_token.is_none() || pages >= max_pages { + if page_token.is_none() { + break; + } + if pages >= INBOX_MAX_PAGES { + warn!( + config_id = %self.config_id, + listed = inbox_ids.len(), + "Gmail INBOX listing hit page cap; older INBOX messages may be missed" + ); + truncated = true; break; } } @@ -128,7 +114,13 @@ impl GmailConnector { } } - let (unarchived, archived) = db.reconcile_inbox(&connection_id, "gmail", &inbox_ids)?; + // Reconcile only when the INBOX listing is complete: a partial listing + // would wrongly archive messages that Gmail still keeps in INBOX. + let (unarchived, archived) = if truncated { + (0, 0) + } else { + db.reconcile_inbox(&connection_id, "gmail", &inbox_ids)? + }; if unarchived > 0 || archived > 0 || !new_msg_ids.is_empty() { info!( @@ -144,14 +136,41 @@ impl GmailConnector { } pub(crate) async fn incremental_sync(&self, db: &Database) -> anyhow::Result<()> { + let api = self.get_client().await?; + self.incremental_sync_with_api(db, &api).await + } + + pub(crate) async fn incremental_sync_with_api( + &self, + db: &Database, + api: &GmailApiClient, + ) -> anyhow::Result<()> { let Some(history_id) = db.get_sync_state(&self.config_id, "history_id")? else { debug!("no history_id, skipping incremental sync"); return Ok(()); }; - let api = self.get_client().await?; let connection_id = self.display_connection_id(); - let resp = api.list_history(&history_id, Some("INBOX")).await?; + let resp = match api.list_history(&history_id, Some("INBOX")).await { + Ok(resp) => resp, + Err(GmailError::HistoryExpired) => { + // historyId is only valid for a limited window (e.g. daemon + // offline for a while). Reconcile against the full INBOX + // listing so local archive state mirrors Gmail again, then + // resume incremental sync from a fresh historyId. + warn!( + config_id = %self.config_id, + "gmail history expired, falling back to full inbox refresh" + ); + self.refresh_inbox_with_api(db, api).await?; + let profile = api.get_profile().await?; + if let Some(new_id) = profile.history_id { + db.set_sync_state(&self.config_id, "history_id", &new_id)?; + } + return Ok(()); + } + Err(e) => return Err(e.into()), + }; if let Some(records) = resp.history { for record in &records { diff --git a/crates/void-gmail/src/connector/tests.rs b/crates/void-gmail/src/connector/tests.rs index 0da704e..14291c5 100644 --- a/crates/void-gmail/src/connector/tests.rs +++ b/crates/void-gmail/src/connector/tests.rs @@ -394,6 +394,185 @@ async fn initial_sync_respects_max_pages() { drop(server); } +// --------------------------------------------------------------------------- +// refresh_inbox / incremental_sync — INBOX reconciliation +// --------------------------------------------------------------------------- + +fn full_inbox_message(id: &str) -> serde_json::Value { + serde_json::json!({ + "id": id, + "threadId": "t1", + "snippet": "Hi", + "internalDate": "1741700000000", + "labelIds": ["INBOX"], + "payload": { + "mimeType": "text/plain", + "headers": [ + {"name": "From", "value": "sender@example.com"}, + {"name": "Subject", "value": "Subj"}, + {"name": "Date", "value": "Wed, 11 Mar 2026 10:00:00 +0000"} + ], + "body": {"data": "SGk", "size": 2} + } + }) +} + +fn seed_gmail_message(db: &Database, ext_id: &str, is_archived: bool) { + let connection_id = "test-gmail"; + let conversation = Conversation { + id: format!("{connection_id}-t0"), + connection_id: connection_id.into(), + connector: "gmail".into(), + external_id: "t0".into(), + name: Some("Seed thread".into()), + kind: ConversationKind::Thread, + last_message_at: None, + unread_count: 0, + is_muted: false, + metadata: None, + }; + db.upsert_conversation(&conversation).unwrap(); + + let msg = Message { + id: format!("{connection_id}-{ext_id}"), + conversation_id: format!("{connection_id}-t0"), + connection_id: connection_id.into(), + connector: "gmail".into(), + external_id: ext_id.into(), + sender: "x@example.com".into(), + sender_name: None, + sender_avatar_url: None, + body: None, + timestamp: 0, + synced_at: None, + is_archived, + is_saved: false, + reply_to_id: None, + media_type: None, + metadata: None, + context_id: Some(format!("{connection_id}-thread-t0")), + context: None, + }; + db.upsert_message(&msg).unwrap(); +} + +fn test_connector() -> GmailConnector { + GmailConnector::new( + "test-gmail", + None, + std::path::Path::new("/tmp/void-gmail-test"), + 60, + ) +} + +/// Regression (issue #63): `void inbox` must mirror Gmail's INBOX label. +/// The refresh must list the *complete* INBOX (no `newer_than:7d` filter) and +/// reconcile local `is_archived` both ways: un-archive INBOX mail that was +/// locally archived, archive mail whose INBOX label is gone. +#[tokio::test] +async fn refresh_inbox_reconciles_complete_inbox_without_date_filter() { + let server = MockServer::start().await; + + // The absence of the `q` matcher is the point: any query filter (e.g. + // `newer_than:7d`) would break reconciliation for older INBOX mail. + Mock::given(method("GET")) + .and(path("/gmail/v1/users/me/messages")) + .and(query_param_is_missing("q")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "messages": [ + {"id": "old1", "threadId": "t1"}, + {"id": "new2", "threadId": "t2"} + ] + }))) + .mount(&server) + .await; + + // `new2` is INBOX-labeled but unknown locally → must be fetched in full. + Mock::given(method("GET")) + .and(path("/gmail/v1/users/me/messages/new2")) + .respond_with(ResponseTemplate::new(200).set_body_json(full_inbox_message("new2"))) + .mount(&server) + .await; + + let api = GmailApiClient::with_base_url("test-token", &server.uri()); + let db = Database::open_in_memory().unwrap(); + // Simulated drift: `old1` is still in Gmail INBOX but locally archived + // (the old 7d-window reconcile did this); `gone1` lost its INBOX label. + seed_gmail_message(&db, "old1", true); + seed_gmail_message(&db, "gone1", false); + + let connector = test_connector(); + connector.refresh_inbox_with_api(&db, &api).await.unwrap(); + + let old1 = db.get_message("test-gmail-old1").unwrap().unwrap(); + assert!(!old1.is_archived, "INBOX message must be un-archived"); + + let gone1 = db.get_message("test-gmail-gone1").unwrap().unwrap(); + assert!(gone1.is_archived, "non-INBOX message must be archived"); + + let new2 = db + .get_message("test-gmail-new2") + .unwrap() + .expect("new2 stored"); + assert!(!new2.is_archived, "new INBOX message stored un-archived"); +} + +/// Regression (issue #63): an expired historyId (Gmail returns 404) used to +/// fail silently forever. It must trigger a full INBOX refresh and resume +/// incremental sync from a fresh historyId. +#[tokio::test] +async fn incremental_sync_history_expired_falls_back_to_inbox_refresh() { + let server = MockServer::start().await; + + Mock::given(method("GET")) + .and(path("/gmail/v1/users/me/history")) + .respond_with(ResponseTemplate::new(404).set_body_json(serde_json::json!({ + "error": {"code": 404, "message": "HistoryId is invalid."} + }))) + .expect(1) + .mount(&server) + .await; + + Mock::given(method("GET")) + .and(path("/gmail/v1/users/me/profile")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "emailAddress": "test@example.com", + "historyId": "99999" + }))) + .mount(&server) + .await; + + Mock::given(method("GET")) + .and(path("/gmail/v1/users/me/messages")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "messages": [{"id": "still1", "threadId": "t1"}] + }))) + .mount(&server) + .await; + + let api = GmailApiClient::with_base_url("test-token", &server.uri()); + let db = Database::open_in_memory().unwrap(); + db.set_sync_state("test-gmail", "history_id", "12345") + .unwrap(); + seed_gmail_message(&db, "gone1", false); + seed_gmail_message(&db, "still1", true); + + let connector = test_connector(); + connector + .incremental_sync_with_api(&db, &api) + .await + .unwrap(); + + let gone1 = db.get_message("test-gmail-gone1").unwrap().unwrap(); + assert!(gone1.is_archived, "archive state reconciled with Gmail"); + + let still1 = db.get_message("test-gmail-still1").unwrap().unwrap(); + assert!(!still1.is_archived, "INBOX message un-archived"); + + let history_id = db.get_sync_state("test-gmail", "history_id").unwrap(); + assert_eq!(history_id, Some("99999".to_string())); +} + #[test] fn parse_email_address_extracts_email() { assert_eq!( diff --git a/crates/void-gmail/src/error.rs b/crates/void-gmail/src/error.rs index acfd5a5..03f0a93 100644 --- a/crates/void-gmail/src/error.rs +++ b/crates/void-gmail/src/error.rs @@ -11,6 +11,10 @@ pub enum GmailError { "insufficient OAuth scope for Gmail settings (need gmail.settings.basic); re-authenticate" )] InsufficientScope, + /// The stored `historyId` is too old: Gmail purges history after a limited + /// window, so incremental sync must fall back to a full INBOX refresh. + #[error("gmail history expired; full inbox refresh required")] + HistoryExpired, #[error("HTTP error: {0}")] Http(#[from] reqwest::Error), #[error("Parse error: {0}")]