diff --git a/crates/core/src/cache/imap/download/flow.rs b/crates/core/src/cache/imap/download/flow.rs index 4ce3eac..3fc0b7b 100644 --- a/crates/core/src/cache/imap/download/flow.rs +++ b/crates/core/src/cache/imap/download/flow.rs @@ -33,7 +33,7 @@ use crate::{ }, error::{code::ErrorCode, BichonResult}, imap::executor::{ - generate_uid_sequence_hashset, ImapExecutor, DEFAULT_BATCH_SIZE, + compress_uid_list, generate_uid_sequence_hashset, ImapExecutor, DEFAULT_BATCH_SIZE, }, store::tantivy::envelope::ENVELOPE_MANAGER, }, @@ -433,12 +433,279 @@ pub async fn fetch_and_save_full_mailbox( fn generate_synthetic_uidvalidity(mailbox_name: &str) -> u32 { use std::collections::hash_map::DefaultHasher; use std::hash::{Hash, Hasher}; - + let mut hasher = DefaultHasher::new(); mailbox_name.hash(&mut hasher); (hasher.finish() as u32).wrapping_add(1) // Avoid 0, which might be reserved } +/// Retry fetching UIDVALIDITY via STATUS command when the initial listing returned None. +/// A None UIDVALIDITY while other STATUS fields (MESSAGES, UNSEEN, UIDNEXT) are present +/// can be caused by transient network issues corrupting just the UIDVALIDITY portion of +/// the response. Retrying avoids unnecessarily triggering a full reconcile. +async fn fetch_uid_validity_with_retry( + account_id: u64, + mailbox_name: &str, + max_retries: u32, +) -> BichonResult> { + let mailbox_name = mailbox_name.to_string(); + fetch_uid_validity_with_retry_inner(max_retries, move || { + let account_id = account_id; + let mailbox_name = mailbox_name.clone(); + async move { + let mut session = ImapExecutor::create_connection(account_id).await?; + let result = session + .status(&mailbox_name, "(UIDVALIDITY)") + .await + .map(|r| r.uid_validity) + .map_err(|e| { + let msg = format!("STATUS failed during UIDVALIDITY retry: {:#?}", e); + raise_error!(msg, ErrorCode::InternalError) + }); + session.logout().await.ok(); + result + } + }) + .await +} + +/// Generic retry loop: calls `fetch_fn` up to `max_retries` times with +/// backoff (500ms × attempt). Returns the first `Some(uid)`, or `Ok(None)` +/// if all attempts return `None` or error. +async fn fetch_uid_validity_with_retry_inner( + max_retries: u32, + mut fetch_fn: F, +) -> BichonResult> +where + F: FnMut() -> Fut, + Fut: std::future::Future>>, +{ + for attempt in 0..max_retries { + if attempt > 0 { + tokio::time::sleep(std::time::Duration::from_millis(500 * attempt as u64)).await; + } + + match fetch_fn().await { + Ok(Some(uid)) => return Ok(Some(uid)), + Ok(None) => { + warn!( + attempt = attempt + 1, + max_retries, + "STATUS returned no UIDVALIDITY" + ); + } + Err(e) => { + warn!( + attempt = attempt + 1, + max_retries, + "UIDVALIDITY fetch attempt failed: {:#?}", e + ); + } + } + } + + Ok(None) +} + +/// Handle uid_validity change without deleting local data. +/// Compares remote Message-IDs with local Tantivy index, downloads only +/// truly missing emails. DedupCache catches any remaining duplicates. +async fn reconcile_uid_validity_change( + account: &AccountModel, + local_mailbox: &MailBox, + remote_mailbox: &MailBox, + token: CancellationToken, +) -> BichonResult> { + let account_id = account.id; + + // Phase 1: connect + examine + let mut session = ImapExecutor::create_connection(account_id).await?; + session + .examine(&remote_mailbox.encoded_name()) + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + + // Phase 2: collect remote UIDs, respecting date constraints + let remote_uid_list: Vec = if let Some(date_since) = &account.date_since { + let date = date_since.since_date()?; + let results = session + .uid_search(&format!("SINCE {date}")) + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + let mut v: Vec = results.into_iter().collect(); + v.sort(); + v + } else if let Some(date_before) = &account.date_before { + let date = date_before.calculate_date()?; + let results = session + .uid_search(&format!("BEFORE {date}")) + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + let mut v: Vec = results.into_iter().collect(); + v.sort(); + v + } else { + let results = session + .uid_search("ALL") + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + let mut v: Vec = results.into_iter().collect(); + v.sort(); + v + }; + + if remote_uid_list.is_empty() { + DownloadState::update_folder_progress( + account_id, + remote_mailbox.name.clone(), + 0, + 0, + FolderStatus::Success, + Some("UIDVALIDITY changed but remote mailbox is empty.".into()), + )?; + session.logout().await.ok(); + return Ok(None); + } + + let max_uid = remote_uid_list.last().copied(); + + // Phase 3: fetch remote Message-IDs (headers only, no bodies) + let uid_set = compress_uid_list(remote_uid_list.clone()); + let remote_msg_ids = + ImapExecutor::fetch_uid_metadata(&mut session, &uid_set, token.clone()).await?; + session.logout().await.ok(); + + // Phase 4: query local Message-IDs from Tantivy. + // For a large mailbox this can allocate 50-100 MB of HashSet. + let local_msg_ids = + ENVELOPE_MANAGER.get_message_ids_for_mailbox(account_id, local_mailbox.id)?; + + // Phase 5: compute missing UIDs + let mut missing_uids: Vec = Vec::new(); + for uid in &remote_uid_list { + if token.is_cancelled() { + return Err(raise_error!("Cancelled".into(), ErrorCode::InternalError)); + } + match remote_msg_ids.get(uid) { + Some(Some(msg_id)) if local_msg_ids.contains(msg_id) => { + // already have this email locally + } + _ => missing_uids.push(*uid), + } + } + + // Phase 6: download missing + if missing_uids.is_empty() { + info!( + account_id, + mailbox = remote_mailbox.name, + "UIDVALIDITY changed but all {} emails already exist locally", + remote_uid_list.len() + ); + DownloadState::update_folder_progress( + account_id, + remote_mailbox.name.clone(), + 0, + 0, + FolderStatus::Success, + None, + )?; + } else { + let planned = missing_uids.len() as u64; + info!( + account_id, + mailbox = remote_mailbox.name, + total = remote_uid_list.len(), + missing = planned, + "UIDVALIDITY changed, downloading missing emails" + ); + DownloadState::update_folder_progress( + account_id, + remote_mailbox.name.clone(), + planned, + 0, + FolderStatus::Downloading, + Some("UIDVALIDITY changed, downloading missing emails...".into()), + )?; + + let mut session2 = ImapExecutor::create_connection(account_id).await?; + session2 + .examine(&remote_mailbox.encoded_name()) + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + + let batch_size = account + .download_batch_size + .unwrap_or(DEFAULT_BATCH_SIZE) as usize; + let batches = generate_uid_sequence_hashset(missing_uids, batch_size); + + let mut downloaded = 0u64; + for (index, batch) in batches.into_iter().enumerate() { + if token.is_cancelled() { + DownloadState::update_folder_progress( + account_id, + remote_mailbox.name.clone(), + planned, + downloaded, + FolderStatus::Cancelled, + None, + )?; + session2.logout().await.ok(); + return Err(raise_error!("Cancelled".into(), ErrorCode::InternalError)); + } + + match ImapExecutor::uid_batch_retrieve_emails( + &mut session2, + account_id, + remote_mailbox.id, + &batch.0, + account.max_email_size_bytes, + token.clone(), + ) + .await + { + Ok(processed) => { + downloaded += processed; + DownloadState::update_folder_progress( + account_id, + remote_mailbox.name.clone(), + planned, + downloaded, + FolderStatus::Downloading, + None, + )?; + } + Err(e) => { + let err_msg = format!("Batch {} failed: {:#?}", index, e); + DownloadState::append_session_error(account_id, err_msg.clone())?; + DownloadState::update_folder_progress( + account_id, + remote_mailbox.name.clone(), + planned, + downloaded, + FolderStatus::Failed, + Some(err_msg), + )?; + session2.logout().await.ok(); + return Err(e); + } + } + } + + DownloadState::update_folder_progress( + account_id, + remote_mailbox.name.clone(), + planned, + downloaded, + FolderStatus::Success, + None, + )?; + session2.logout().await.ok(); + } + + Ok(max_uid) +} + pub async fn reconcile_mailboxes( account: &AccountModel, remote_mailboxes: &[MailBox], @@ -471,71 +738,66 @@ pub async fn reconcile_mailboxes( let remote_uid_validity = match remote_mailbox.uid_validity { Some(uid) => uid, None => { - // Generate a synthetic UIDVALIDITY based on mailbox name - let synthetic_uid = generate_synthetic_uidvalidity(&remote_mailbox.name); - - warn!( - "Account {}: Mailbox '{}' - Server did not provide UIDVALIDITY. \ - Using synthetic UIDVALIDITY {} based on mailbox name. \ - This mailbox will be synced but may require periodic rebuilds if the server's mailbox structure changes.", - account_id, remote_mailbox.name, synthetic_uid - ); - - synthetic_uid + if local_mailbox.uid_validity.is_some() { + // We had a real UIDVALIDITY before; a None now is likely + // transient (network jitter). Retry before falling back. + warn!( + "Account {}: Mailbox '{}' - STATUS returned no UIDVALIDITY, retrying to rule out network jitter...", + account_id, remote_mailbox.name + ); + match fetch_uid_validity_with_retry( + account_id, + &remote_mailbox.encoded_name(), + 3, + ) + .await? + { + Some(uid) => { + info!( + "Account {}: Mailbox '{}' - UIDVALIDITY recovered after retry: {}", + account_id, remote_mailbox.name, uid + ); + uid + } + None => { + // All retries exhausted; keep the local value. + // Safer than synthetic because we know the server had one. + let fallback = local_mailbox.uid_validity.unwrap(); + warn!( + "Account {}: Mailbox '{}' - all retries failed, keeping local UIDVALIDITY {}", + account_id, remote_mailbox.name, fallback + ); + fallback + } + } + } else { + // First sync and server genuinely doesn't provide UIDVALIDITY + let synthetic_uid = generate_synthetic_uidvalidity(&remote_mailbox.name); + warn!( + "Account {}: Mailbox '{}' - Server did not provide UIDVALIDITY. \ + Using synthetic UIDVALIDITY {} based on mailbox name. \ + This mailbox will be synced but may require periodic rebuilds if the server's mailbox structure changes.", + account_id, remote_mailbox.name, synthetic_uid + ); + synthetic_uid + } } }; let new_highest_uid = if local_mailbox.uid_validity != Some(remote_uid_validity) { info!( "Account {}: Mailbox '{}' detected with changed uid_validity (local: {:#?}, remote: {:#?}). \ - The mailbox data may be invalid, resetting its envelopes and rebuilding the cache.", + Comparing by Message-ID to find missing emails.", account_id, local_mailbox.name, &local_mailbox.uid_validity, &remote_uid_validity ); - DownloadState::update_folder_progress( - account_id, - local_mailbox.name.clone(), - remote_mailbox.exists as u64, - 0, - FolderStatus::Downloading, - Some("UID validity changed, rebuilding...".into()), - )?; - - match &account.date_since { - Some(date_since) => { - rebuild_mailbox_cache_by_date( - account, - local_mailbox.id, - &date_since.since_date()?, - remote_mailbox, - FetchDirection::Since, - token.clone(), - ) - .await? - } - None => match &account.date_before { - Some(r) => { - rebuild_mailbox_cache_by_date( - account, - local_mailbox.id, - &r.calculate_date()?, - remote_mailbox, - FetchDirection::Before, - token.clone(), - ) - .await? - } - None => { - rebuild_mailbox_cache( - account, - local_mailbox, - remote_mailbox, - token.clone(), - ) - .await? - } - }, - } + reconcile_uid_validity_change( + account, + local_mailbox, + remote_mailbox, + token.clone(), + ) + .await? } else { perform_incremental_sync(account, local_mailbox, remote_mailbox, token.clone()) .await? @@ -749,3 +1011,648 @@ async fn perform_incremental_sync( Ok(local_mailbox.highest_uid) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::imap::session::SessionStream; + use std::sync::Arc; + use tokio_io_timeout::TimeoutStream; + + // ============================================================ + // Pure unit tests (no network) + // ============================================================ + + #[test] + fn test_generate_synthetic_uidvalidity_deterministic() { + let a = generate_synthetic_uidvalidity("INBOX"); + let b = generate_synthetic_uidvalidity("INBOX"); + assert_eq!(a, b, "same mailbox name must produce same uid_validity"); + } + + #[test] + fn test_generate_synthetic_uidvalidity_different_mailboxes() { + let inbox = generate_synthetic_uidvalidity("INBOX"); + let sent = generate_synthetic_uidvalidity("Sent"); + assert_ne!(inbox, sent, "different mailboxes should have different uid_validity"); + } + + #[test] + fn test_generate_synthetic_uidvalidity_non_zero() { + let uid = generate_synthetic_uidvalidity("INBOX"); + assert_ne!(uid, 0, "UIDVALIDITY should not be 0 (reserved)"); + } + + // ============================================================ + // Integration tests (real IMAP server required) + // ============================================================ + // Fill in your IMAP server details below to run these tests. + // cargo test -p bichon-core -- --ignored + + const TEST_IMAP_HOST: &str = "imap.zoho.com"; + const TEST_IMAP_PORT: u16 = 993; + const TEST_IMAP_USERNAME: &str = ""; + const TEST_IMAP_PASSWORD: &str = ""; + const TEST_MAILBOX: &str = "INBOX"; + + /// Build a direct IMAP session bypassing the database-dependent + /// ImapConnectionManager, for local testing with real credentials. + async fn direct_connect( + host: &str, + port: u16, + username: &str, + password: &str, + ) -> Result>, String> { + use rustls::ClientConfig; + use rustls_pki_types::ServerName; + use tokio::net::TcpStream; + use tokio_rustls::TlsConnector; + + // Ensure a rustls crypto provider is installed (ring). + // May already be installed by production code; ignore duplicate. + rustls::crypto::CryptoProvider::install_default( + rustls::crypto::ring::default_provider(), + ) + .ok(); + + let tcp = TcpStream::connect((host, port)) + .await + .map_err(|e| format!("TCP connect error: {e}"))?; + + // Wrap in Pin>> to satisfy SessionStream, + // matching the production path in establish_tcp_connection_with_timeout. + let timeout_stream = TimeoutStream::new(tcp); + let pinned = Box::pin(timeout_stream); + + let server_name = ServerName::try_from(host.to_owned()) + .map_err(|e| format!("Invalid hostname: {e}"))?; + + let config = ClientConfig::builder() + .with_root_certificates(rustls::RootCertStore { + roots: webpki_roots::TLS_SERVER_ROOTS.into(), + }) + .with_no_client_auth(); + + let connector = TlsConnector::from(Arc::new(config)); + let tls_stream = connector + .connect(server_name, pinned) + .await + .map_err(|e| format!("TLS error: {e}"))?; + + let client = async_imap::Client::new(Box::new(tls_stream) as Box); + let session = client + .login(username, password) + .await + .map_err(|(e, _)| format!("Login error: {e}"))?; + + Ok(session) + } + + /// Test STATUS UIDVALIDITY directly — verifies your server returns + /// a valid UIDVALIDITY for the given mailbox. + #[tokio::test] + #[ignore = "requires real IMAP credentials"] + async fn test_status_uid_validity_present() { + if TEST_IMAP_HOST.is_empty() || TEST_IMAP_USERNAME.is_empty() { + eprintln!("SKIP: fill in TEST_IMAP_* constants to run this test"); + return; + } + + let mut session = direct_connect( + TEST_IMAP_HOST, + TEST_IMAP_PORT, + TEST_IMAP_USERNAME, + TEST_IMAP_PASSWORD, + ) + .await + .expect("should connect"); + + let status = session + .status(TEST_MAILBOX, "(UIDVALIDITY)") + .await + .expect("STATUS command should succeed"); + + session.logout().await.ok(); + + match status.uid_validity { + Some(uid) => { + println!("[OK] Server returned UIDVALIDITY: {uid}"); + assert_ne!(uid, 0); + } + None => { + println!("[INFO] Server returned no UIDVALIDITY in STATUS response"); + println!(" This server may need the synthetic fallback or the retry logic."); + } + } + } + + /// Test STATUS with full attributes (MESSAGES UNSEEN UIDNEXT UIDVALIDITY), + /// mimicking the actual listing flow in mailbox::list::fetch_remote_with_progress. + #[tokio::test] + #[ignore = "requires real IMAP credentials"] + async fn test_status_full_attributes() { + if TEST_IMAP_HOST.is_empty() || TEST_IMAP_USERNAME.is_empty() { + eprintln!("SKIP: fill in TEST_IMAP_* constants to run this test"); + return; + } + + let mut session = direct_connect( + TEST_IMAP_HOST, + TEST_IMAP_PORT, + TEST_IMAP_USERNAME, + TEST_IMAP_PASSWORD, + ) + .await + .expect("should connect"); + + let status = session + .status(TEST_MAILBOX, "(MESSAGES UNSEEN UIDNEXT UIDVALIDITY)") + .await + .expect("STATUS with full attributes should succeed"); + + session.logout().await.ok(); + + println!("MESSAGES: {:?}", status.exists); + println!("UNSEEN: {:?}", status.unseen); + println!("UIDNEXT: {:?}", status.uid_next); + println!("UIDVALIDITY: {:?}", status.uid_validity); + } + + /// Simulate the retry flow: call STATUS multiple times and verify + /// UIDVALIDITY is consistently returned (or consistently absent). + #[tokio::test] + #[ignore = "requires real IMAP credentials"] + async fn test_uid_validity_consistency_over_multiple_status_calls() { + if TEST_IMAP_HOST.is_empty() || TEST_IMAP_USERNAME.is_empty() { + eprintln!("SKIP: fill in TEST_IMAP_* constants to run this test"); + return; + } + + let results: Vec> = Vec::with_capacity(5); + let results = std::cell::RefCell::new(results); + + for i in 0..5 { + let mut session = direct_connect( + TEST_IMAP_HOST, + TEST_IMAP_PORT, + TEST_IMAP_USERNAME, + TEST_IMAP_PASSWORD, + ) + .await + .expect("should connect"); + + let status = session + .status(TEST_MAILBOX, "(UIDVALIDITY)") + .await + .expect("STATUS should succeed"); + + session.logout().await.ok(); + + println!( + "Call {}: UIDVALIDITY = {:?}", + i + 1, + status.uid_validity + ); + results.borrow_mut().push(status.uid_validity); + } + + let results = results.into_inner(); + let first = results[0]; + let all_same = results.iter().all(|r| *r == first); + if all_same { + println!("[OK] All 5 STATUS calls returned consistent UIDVALIDITY: {first:?}"); + } else { + println!("[WARN] Inconsistent UIDVALIDITY across calls: {results:?}"); + println!(" Network jitter or server-side changes detected."); + } + } + + /// Test `fetch_uid_metadata` against a real IMAP server. + /// Fetches a few UIDs from the configured mailbox and verifies + /// that every returned UID has a valid, non-empty Message-ID. + #[tokio::test] + #[ignore = "requires real IMAP credentials"] + async fn test_fetch_uid_metadata_real_server() { + if TEST_IMAP_HOST.is_empty() || TEST_IMAP_USERNAME.is_empty() { + eprintln!("SKIP: fill in TEST_IMAP_* constants to run this test"); + return; + } + + let mut session = direct_connect( + TEST_IMAP_HOST, + TEST_IMAP_PORT, + TEST_IMAP_USERNAME, + TEST_IMAP_PASSWORD, + ) + .await + .expect("should connect"); + + // Examine to enter the mailbox + session + .examine(TEST_MAILBOX) + .await + .expect("EXAMINE should succeed"); + + // Find actual UIDs via SEARCH (don't assume contiguous 1..N) + let all_uids: Vec = { + let mut v: Vec = session + .uid_search("ALL") + .await + .expect("UID SEARCH ALL should succeed") + .into_iter() + .collect(); + v.sort(); + v + }; + + if all_uids.is_empty() { + eprintln!("SKIP: mailbox is empty, nothing to fetch"); + session.logout().await.ok(); + return; + } + + // Take up to 5 UIDs for a quick test + let sample: Vec = all_uids.into_iter().take(5).collect(); + let uid_set = compress_uid_list(sample.clone()); + + let result = ImapExecutor::fetch_uid_metadata( + &mut session, + &uid_set, + CancellationToken::new(), + ) + .await + .expect("fetch_uid_metadata should succeed"); + + session.logout().await.ok(); + + println!( + "[OK] Fetched {} UIDs from mailbox '{}':", + result.len(), + TEST_MAILBOX + ); + for uid in sample.iter() { + let mid = result + .get(uid) + .map(|o| o.as_deref().unwrap_or("")) + .unwrap_or(""); + println!(" UID {uid}: {mid}"); + } + + // Every UID we requested must be present with a valid Message-ID + for uid in &sample { + let msg_id = result + .get(uid) + .unwrap_or_else(|| panic!("UID {uid} missing from result")); + let mid = msg_id + .as_deref() + .unwrap_or_else(|| panic!("UID {uid} returned None Message-ID")); + assert!(!mid.is_empty(), "UID {uid} returned empty Message-ID"); + assert!( + mid.contains('@'), + "UID {uid}: Message-ID '{mid}' does not look like a valid Message-ID" + ); + } + } + + // ============================================================ + // Unit tests for fetch_uid_validity_with_retry_inner + // (pure logic, no network needed) + // ============================================================ + + /// Mock helper: returns the given results in sequence, then always None. + fn mock_results( + results: Vec>>, + ) -> impl FnMut() -> std::future::Ready>> { + let mut iter = results.into_iter(); + move || std::future::ready(iter.next().unwrap_or(Ok(None))) + } + + #[tokio::test] + async fn test_retry_first_attempt_succeeds() { + let result = fetch_uid_validity_with_retry_inner(3, mock_results(vec![Ok(Some(42))])) + .await; + assert_eq!(result.unwrap(), Some(42)); + } + + #[tokio::test] + async fn test_retry_succeeds_after_two_nones() { + // First two attempts return None, third returns Some + let result = fetch_uid_validity_with_retry_inner( + 3, + mock_results(vec![Ok(None), Ok(None), Ok(Some(99))]), + ) + .await; + assert_eq!(result.unwrap(), Some(99)); + } + + #[tokio::test] + async fn test_retry_succeeds_after_error_then_none() { + // Error, then None, then success + let result = fetch_uid_validity_with_retry_inner( + 3, + mock_results(vec![ + Err(raise_error!("boom".into(), ErrorCode::NetworkError)), + Ok(None), + Ok(Some(7)), + ]), + ) + .await; + assert_eq!(result.unwrap(), Some(7)); + } + + #[tokio::test] + async fn test_retry_returns_none_after_all_retries_exhausted() { + // All attempts return None + let result = fetch_uid_validity_with_retry_inner( + 3, + mock_results(vec![Ok(None), Ok(None), Ok(None)]), + ) + .await; + assert_eq!(result.unwrap(), None); + } + + #[tokio::test] + async fn test_retry_returns_none_when_all_attempts_error() { + // All attempts return Err — still returns Ok(None), not propagating the error + let result = fetch_uid_validity_with_retry_inner( + 3, + mock_results(vec![ + Err(raise_error!("e1".into(), ErrorCode::NetworkError)), + Err(raise_error!("e2".into(), ErrorCode::NetworkError)), + Err(raise_error!("e3".into(), ErrorCode::NetworkError)), + ]), + ) + .await; + assert!(result.is_ok()); + assert_eq!(result.unwrap(), None); + } + + #[tokio::test] + async fn test_retry_respects_max_retries() { + // max_retries=5, success on 5th attempt + let result = fetch_uid_validity_with_retry_inner( + 5, + mock_results(vec![ + Ok(None), + Ok(None), + Ok(None), + Ok(None), + Ok(Some(5)), + ]), + ) + .await; + assert_eq!(result.unwrap(), Some(5)); + } + + #[tokio::test] + async fn test_retry_stops_at_max_retries() { + // max_retries=2, third value is Some but should never be reached + let result = fetch_uid_validity_with_retry_inner( + 2, + mock_results(vec![Ok(None), Ok(None), Ok(Some(42))]), + ) + .await; + assert_eq!(result.unwrap(), None, "should stop after max_retries (2)"); + } + + #[tokio::test] + async fn test_retry_max_retries_zero() { + // max_retries=0 means no attempts at all + let result = fetch_uid_validity_with_retry_inner( + 0, + mock_results(vec![Ok(Some(42))]), + ) + .await; + assert_eq!(result.unwrap(), None); + } + + // ============================================================ + // Mock IMAP server integration tests + // ============================================================ + + use crate::imap::mock_server::{ + examine_response, uid_fetch_metadata_response, uid_fetch_rfc822_response, + minimal_eml, MockImapServer, MockImapServerHandle, + }; + + /// Build an `async_imap::Session` connected to the mock server, + /// authenticated and with the given mailbox examined. + async fn mock_session( + handle: &MockImapServerHandle, + ) -> async_imap::Session> { + let tcp = tokio::net::TcpStream::connect((handle.host(), handle.port())) + .await + .unwrap(); + let timeout_stream = TimeoutStream::new(tcp); + let pinned: std::pin::Pin>> = + Box::pin(timeout_stream); + let stream: Box = Box::new(pinned); + let mut client = async_imap::Client::new(stream); + + // Read greeting + client.read_response().await.unwrap(); + + // Login + let mut session = client.login("user", "pass").await.map_err(|(e, _)| { + panic!("Login failed: {e:?}") + }).unwrap(); + + // Examine + session.examine("INBOX").await.unwrap(); + + session + } + + #[tokio::test] + async fn fetch_uid_metadata_with_mock_server() { + let handle = MockImapServer::new() + .respond("LOGIN", "{TAG} OK LOGIN done\r\n") + .respond("EXAMINE", examine_response("INBOX", 3, 42, 4)) + .respond("UID FETCH", uid_fetch_metadata_response(&[ + (1, ""), + (2, ""), + (3, ""), + ])) + .start() + .await; + + let mut session = mock_session(&handle).await; + + let result = ImapExecutor::fetch_uid_metadata( + &mut session, + "1:3", + CancellationToken::new(), + ) + .await + .unwrap(); + + assert_eq!(result.len(), 3); + assert_eq!( + result.get(&1).unwrap().as_deref(), + Some("msg-a@test.com") + ); + assert_eq!( + result.get(&2).unwrap().as_deref(), + Some("msg-b@test.com") + ); + assert_eq!( + result.get(&3).unwrap().as_deref(), + Some("msg-c@test.com") + ); + + session.logout().await.ok(); + } + + /// Verify basic UID FETCH (FLAGS only, no body) works with the mock server. + #[tokio::test] + async fn fetch_uid_flags_with_mock_server() { + // Response with just UID and FLAGS — no body literal parsing needed. + let handle = MockImapServer::new() + .respond("LOGIN", "{TAG} OK LOGIN done\r\n") + .respond("EXAMINE", examine_response("INBOX", 2, 42, 3)) + .respond( + "UID FETCH", + b"* 1 FETCH (UID 1 FLAGS (\\Seen))\r\n\ +* 2 FETCH (UID 2 FLAGS (\\Flagged))\r\n\ +{TAG} OK FETCH completed\r\n" + .to_vec(), + ) + .start() + .await; + + let mut session = mock_session(&handle).await; + + let uids: Vec = { + let mut stream = session + .uid_fetch("1:2", "(UID FLAGS)") + .await + .unwrap(); + + use futures::TryStreamExt; + let mut uids = Vec::new(); + while let Some(fetch) = stream.try_next().await.unwrap() { + uids.push(fetch.uid.unwrap_or(0)); + } + uids + }; + + assert_eq!(uids, vec![1, 2]); + + session.logout().await.ok(); + } + + /// Verify UID FETCH with BODY[] (full RFC822) works with the mock server. + #[tokio::test] + async fn fetch_uid_rfc822_with_mock_server() { + let eml = minimal_eml("Test Subject", "test@example.com"); + let eml_len = eml.len(); + + let handle = MockImapServer::new() + .respond("LOGIN", "{TAG} OK LOGIN done\r\n") + .respond("EXAMINE", examine_response("INBOX", 1, 42, 2)) + .respond("UID FETCH", uid_fetch_rfc822_response(1, &eml)) + .start() + .await; + + let mut session = mock_session(&handle).await; + + let bodies: Vec<(u32, Vec)> = { + let mut stream = session + .uid_fetch("1:1", "(UID BODY[])") + .await + .unwrap(); + + use futures::TryStreamExt; + let mut bodies = Vec::new(); + while let Some(fetch) = stream.try_next().await.unwrap() { + let uid = fetch.uid.unwrap_or(0); + let body = fetch.body().map(|b| b.to_vec()).unwrap_or_default(); + bodies.push((uid, body)); + } + bodies + }; + + assert_eq!(bodies.len(), 1); + assert_eq!(bodies[0].0, 1); + assert_eq!(bodies[0].1.len(), eml_len); + assert!(String::from_utf8_lossy(&bodies[0].1).contains("Test Subject")); + + session.logout().await.ok(); + } + + #[tokio::test] + async fn fetch_uid_metadata_empty_mailbox() { + let handle = MockImapServer::new() + .respond("LOGIN", "{TAG} OK LOGIN done\r\n") + .respond("EXAMINE", examine_response("INBOX", 0, 42, 1)) + .respond("UID FETCH", b"{TAG} OK FETCH completed\r\n".to_vec()) + .start() + .await; + + let mut session = mock_session(&handle).await; + + let result = ImapExecutor::fetch_uid_metadata( + &mut session, + "1:*", + CancellationToken::new(), + ) + .await + .unwrap(); + + assert!( + result.is_empty(), + "empty mailbox should return empty map" + ); + + session.logout().await.ok(); + } + + #[tokio::test] + async fn fetch_uid_metadata_missing_message_id() { + // One entry has a Message-ID, the other has no header at all. + let header_with_msgid = + "From: sender@example.com\r\n\ + Date: Thu, 01 Jan 2025 00:00:00 +0000\r\n\ + Message-ID: \r\n\r\n"; + let header_without_msgid = "\r\n"; + let len1 = header_with_msgid.len(); + let len2 = header_without_msgid.len(); + + let handle = MockImapServer::new() + .respond("LOGIN", "{TAG} OK LOGIN done\r\n") + .respond("EXAMINE", examine_response("INBOX", 2, 42, 3)) + .respond( + "UID FETCH", + format!( + "* 1 FETCH (UID 1 BODY[HEADER] {{{len1}}}\r\n\ +{header_with_msgid})\r\n\ +* 2 FETCH (UID 2 BODY[HEADER] {{{len2}}}\r\n\ +{header_without_msgid})\r\n\ +{{TAG}} OK FETCH completed\r\n" + ) + .into_bytes(), + ) + .start() + .await; + + let mut session = mock_session(&handle).await; + + let result = ImapExecutor::fetch_uid_metadata( + &mut session, + "1:2", + CancellationToken::new(), + ) + .await + .unwrap(); + + assert_eq!(result.len(), 2); + assert_eq!( + result.get(&1).unwrap().as_deref(), + Some("ok@test.com") + ); + // UID 2 has no Message-ID header → None + assert_eq!(result.get(&2).unwrap().as_deref(), None); + + session.logout().await.ok(); + } +} diff --git a/crates/core/src/imap/client.rs b/crates/core/src/imap/client.rs index 1fd70b0..85cbcee 100644 --- a/crates/core/src/imap/client.rs +++ b/crates/core/src/imap/client.rs @@ -34,6 +34,25 @@ use std::ops::DerefMut; use tokio::io::BufWriter; use tracing::debug; +/// Classify an `io::Error` (from TLS stream I/O) for IMAP connection errors. +/// `UnexpectedEof` is treated as a network error because many servers skip +/// the TLS `close_notify` alert, causing rustls to emit this error when the +/// TCP connection is dropped normally. +fn classify_io_error(e: &std::io::Error) -> ErrorCode { + use std::io::ErrorKind; + matches!( + e.kind(), + ErrorKind::BrokenPipe + | ErrorKind::ConnectionReset + | ErrorKind::ConnectionAborted + | ErrorKind::TimedOut + | ErrorKind::UnexpectedEof + | ErrorKind::NotConnected + ) + .then_some(ErrorCode::NetworkError) + .unwrap_or(ErrorCode::ImapCommandFailed) +} + #[derive(Debug)] pub(crate) struct Client { inner: ImapClient>, @@ -141,7 +160,7 @@ impl Client { let _greeting = client .read_response() .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))? + .map_err(|e| raise_error!(format!("{:#?}", e), classify_io_error(&e)))? .ok_or_else(|| { raise_error!( "Failed to read IMAP greeting — this usually indicates an incorrect encryption setting (SSL vs. STARTTLS). Your current setting is SSL.".into(), @@ -171,7 +190,7 @@ impl Client { let _greeting = client .read_response() .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))? + .map_err(|e| raise_error!(format!("{:#?}", e), classify_io_error(&e)))? .ok_or_else(|| { raise_error!( "failed to read greeting".into(), @@ -202,7 +221,7 @@ impl Client { let _greeting = client .read_response() .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))? + .map_err(|e| raise_error!(format!("{:#?}", e), classify_io_error(&e)))? .ok_or_else(|| { raise_error!( "Failed to read IMAP greeting — this usually indicates an incorrect encryption setting (SSL vs. STARTTLS). Your current setting is STARTTLS.".into(), diff --git a/crates/core/src/imap/executor.rs b/crates/core/src/imap/executor.rs index 16e281b..3f5313e 100644 --- a/crates/core/src/imap/executor.rs +++ b/crates/core/src/imap/executor.rs @@ -27,7 +27,7 @@ use crate::{error::BichonResult, imap::manager::ImapConnectionManager}; use async_imap::types::Name; use async_imap::Session; use futures::TryStreamExt; -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use tokio_util::sync::CancellationToken; use tracing::info; @@ -260,7 +260,9 @@ impl ImapExecutor { let mut count = 0u64; let mut skipped = 0u64; let mut max_uid: Option = None; - let size_limit = account.max_email_size_bytes.unwrap_or(DEFAULT_MAX_EMAIL_SIZE); + let size_limit = account + .max_email_size_bytes + .unwrap_or(DEFAULT_MAX_EMAIL_SIZE); while let Some(fetch) = stream .try_next() .await @@ -363,14 +365,14 @@ impl ImapExecutor { let mut size_stream = session .fetch(sequence_set.as_str(), SIZE_ONLY_FETCH) .await - .map_err(|e| { - raise_error!(format!("{:#?}", e), classify_imap_error(&e)) - })?; + .map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?; let mut uids: Vec = Vec::new(); - while let Some(fetch) = size_stream.try_next().await.map_err(|e| { - raise_error!(format!("{:#?}", e), classify_imap_error(&e)) - })? { + while let Some(fetch) = size_stream + .try_next() + .await + .map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))? + { let uid = fetch.uid.unwrap_or(0); let msg_size = fetch.size.unwrap_or(0) as u64; if msg_size == 0 || msg_size <= limit { @@ -437,14 +439,14 @@ impl ImapExecutor { let mut size_stream = session .uid_fetch(uid_set, SIZE_ONLY_FETCH) .await - .map_err(|e| { - raise_error!(format!("{:#?}", e), classify_imap_error(&e)) - })?; + .map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?; let mut uids: Vec = Vec::new(); - while let Some(fetch) = size_stream.try_next().await.map_err(|e| { - raise_error!(format!("{:#?}", e), classify_imap_error(&e)) - })? { + while let Some(fetch) = size_stream + .try_next() + .await + .map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))? + { let uid = fetch.uid.unwrap_or(0); let msg_size = fetch.size.unwrap_or(0) as u64; if msg_size == 0 || msg_size <= limit { @@ -550,6 +552,37 @@ impl ImapExecutor { ) -> BichonResult>> { ImapConnectionManager::build(account_id).await } + + /// Fetch UID → Message-ID mapping without downloading bodies. + /// `uid_set` is an IMAP sequence-set string (e.g. "1:100" or "1,3,5"). + pub async fn fetch_uid_metadata( + session: &mut Session>, + uid_set: &str, + token: CancellationToken, + ) -> BichonResult>> { + let mut stream = session + .uid_fetch(uid_set, "(UID BODY.PEEK[HEADER])") + .await + .map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?; + + let mut result = HashMap::new(); + while let Some(fetch) = stream + .try_next() + .await + .map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))? + { + if token.is_cancelled() { + return Err(raise_error!( + "Stream cancelled".into(), + ErrorCode::InternalError + )); + } + let uid = fetch.uid.unwrap_or(0); + let msg_id = fetch.header().and_then(parse_message_id_header); + result.insert(uid, msg_id); + } + Ok(result) + } } pub const DEFAULT_BATCH_SIZE: u32 = 30; @@ -613,6 +646,27 @@ pub fn generate_uid_sequence_hashset( result } +fn parse_message_id_header(header_bytes: &[u8]) -> Option { + let header = std::str::from_utf8(header_bytes).ok()?; + for line in header.lines() { + if let Some(value) = line + .strip_prefix("Message-ID:") + .or_else(|| line.strip_prefix("Message-Id:")) + .or_else(|| line.strip_prefix("Message-id:")) + { + // mail_parser strips angle brackets, so we must do the same + // to ensure comparisons against the Tantivy index match. + let trimmed = value.trim(); + let stripped = trimmed.strip_prefix('<').unwrap_or(trimmed); + let stripped = stripped.strip_suffix('>').unwrap_or(stripped); + if !stripped.is_empty() { + return Some(stripped.to_string()); + } + } + } + None +} + #[cfg(test)] mod test { use super::*; @@ -668,4 +722,80 @@ mod test { assert_eq!(batches[2].0, "5"); assert_eq!(batches[2].1, 1); } + + // ── parse_message_id_header ───────────────────────────────────── + + #[test] + fn parse_standard_message_id() { + let header = b"Message-ID: \r\n"; + assert_eq!( + parse_message_id_header(header), + Some("abc123@example.com".into()) + ); + } + + #[test] + fn parse_message_id_lowercase() { + let header = b"Message-Id: \r\n"; + assert_eq!( + parse_message_id_header(header), + Some("foo@bar.com".into()) + ); + } + + #[test] + fn parse_message_id_extra_whitespace() { + let header = b"Message-ID: \r\n"; + assert_eq!( + parse_message_id_header(header), + Some("spaces@test.com".into()) + ); + } + + #[test] + fn parse_empty_message_id_returns_none() { + let header = b"Message-ID: <>\r\n"; + assert_eq!(parse_message_id_header(header), None); + } + + #[test] + fn parse_missing_header_returns_none() { + let header = b"X-Custom: something\r\n"; + assert_eq!(parse_message_id_header(header), None); + } + + #[test] + fn parse_empty_body_returns_none() { + assert_eq!(parse_message_id_header(b""), None); + } + + #[test] + fn parse_message_id_in_full_header() { + // The Message-ID line is in the middle, not at the start. + let header = b"From: sender@example.com\r\n\ +Date: Thu, 01 Jan 2025 00:00:00 +0000\r\n\ +Subject: test\r\n\ +Message-ID: \r\n\ +To: recipient@example.com\r\n\r\n"; + assert_eq!( + parse_message_id_header(header), + Some("mid@example.com".into()) + ); + } + + #[test] + fn parse_message_id_only_in_full_header() { + // Only a few headers, Message-ID is among them. + let header = b"From: a@b.com\r\nMessage-ID: \r\n\r\n"; + assert_eq!(parse_message_id_header(header), Some("x@y.com".into())); + } + + #[test] + fn parse_message_id_no_brackets_still_works() { + let header = b"Message-ID: plain@example.com\r\n"; + assert_eq!( + parse_message_id_header(header), + Some("plain@example.com".into()) + ); + } } diff --git a/crates/core/src/imap/manager.rs b/crates/core/src/imap/manager.rs index d32f517..e048edd 100644 --- a/crates/core/src/imap/manager.rs +++ b/crates/core/src/imap/manager.rs @@ -95,16 +95,40 @@ impl ImapConnectionManager { pub async fn build(account_id: u64) -> BichonResult>> { let account = AccountModel::get(account_id)?; - let client = match Self::create_client(&account).await { - Ok(client) => client, - Err(error) => { - error!( - "Failed to create IMAP {}'s client: {:#?}", - &account.email, error - ); - return Err(error); + let account_email = account.email.clone(); + + let mut client = None; + for attempt in 0..3u32 { + match Self::create_client(&account).await { + Ok(c) => { + client = Some(c); + break; + } + Err(error) if error.code() == ErrorCode::NetworkError && attempt < 2 => { + warn!( + "IMAP connection attempt {}/3 to {} failed (network error), retrying...", + attempt + 1, + account_email + ); + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + continue; + } + Err(error) => { + error!( + "Failed to create IMAP {}'s client: {:#?}", + account_email, error + ); + return Err(error); + } } - }; + } + + let client = client.ok_or_else(|| { + raise_error!( + format!("Failed to create IMAP {}'s client after 3 attempts", account_email), + ErrorCode::NetworkError + ) + })?; let mut session = match Self::authenticate(client, &account).await { Ok(session) => session, diff --git a/crates/core/src/imap/mock_server.rs b/crates/core/src/imap/mock_server.rs new file mode 100644 index 0000000..c37c345 --- /dev/null +++ b/crates/core/src/imap/mock_server.rs @@ -0,0 +1,538 @@ +// +// Copyright (c) 2025-2026 rustmailer.com (https://rustmailer.com) +// +// This file is part of the Bichon Email Archiving Project +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +//! A minimal scriptable IMAP server for integration testing. +//! +//! Each instance listens on a random localhost port and responds to a +//! pre-configured script of (expected_command, response) pairs. Commands +//! are matched by substring — the first matching pattern wins. +//! +//! # Example +//! ```ignore +//! let server = MockImapServer::new() +//! .greeting("* OK ready\r\n") +//! .respond("LOGIN", "A0 OK logged in\r\n") +//! .respond("CAPABILITY", "* CAPABILITY IMAP4rev1\r\nA0 OK done\r\n") +//! .respond("STATUS", "* STATUS INBOX (MESSAGES 10 UIDVALIDITY 42)\r\nA0 OK\r\n") +//! .respond("LOGOUT", "* BYE\r\nA0 OK\r\n") +//! .start() +//! .await; +//! +//! let (host, port) = server.addr(); +//! // connect to host:port with Encryption::None +//! ``` + +use std::net::SocketAddr; +use std::sync::Arc; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::{TcpListener, TcpStream}; + +type Response = Vec; + +pub struct MockImapServer { + greeting: Vec, + script: Vec<(String, Response)>, +} + +impl MockImapServer { + pub fn new() -> Self { + Self { + greeting: b"* OK Mock IMAP server ready\r\n".to_vec(), + script: Vec::new(), + } + } + + /// Set the greeting banner sent immediately after connection. + pub fn greeting(mut self, banner: impl Into>) -> Self { + self.greeting = banner.into(); + self + } + + /// Add a script step: when a client command *contains* `pattern` (case-insensitive), + /// respond with `response`. Steps are checked in insertion order. + pub fn respond(mut self, pattern: impl Into, response: impl Into>) -> Self { + self.script.push((pattern.into(), response.into())); + self + } + + /// Start the server on a random port. Returns a handle whose `addr()` gives + /// the `(host, port)` to connect to. + pub async fn start(self) -> MockImapServerHandle { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("local_addr"); + + let server = Arc::new(self); + + tokio::spawn(async move { + loop { + match listener.accept().await { + Ok((stream, _)) => { + let srv = server.clone(); + tokio::spawn(async move { + srv.handle_connection(stream).await; + }); + } + Err(_) => break, + } + } + }); + + MockImapServerHandle { addr } + } + + async fn handle_connection(&self, mut stream: TcpStream) { + let (reader, mut writer) = stream.split(); + let mut reader = BufReader::new(reader); + + // Send greeting + if writer.write_all(&self.greeting).await.is_err() { + return; + } + + let mut line = String::new(); + loop { + line.clear(); + match reader.read_line(&mut line).await { + Ok(0) => break, // EOF + Ok(_) => {} + Err(_) => break, + } + + let tag = extract_tag(&line).unwrap_or("A0"); + let matched = self.find_match(&line); + if let Some(response) = matched { + let substituted = substitute_tag(response, tag); + if writer.write_all(&substituted).await.is_err() { + break; + } + } else { + // Default: send tagged OK for commands we don't handle + let fallback = format!("{tag} OK done\r\n"); + if writer.write_all(fallback.as_bytes()).await.is_err() { + break; + } + } + } + } + + fn find_match(&self, line: &str) -> Option<&[u8]> { + let line_lower = line.to_lowercase(); + for (pattern, response) in &self.script { + if line_lower.contains(&pattern.to_lowercase()) { + return Some(response); + } + } + None + } +} + +impl Default for MockImapServer { + fn default() -> Self { + Self::new() + } +} + +/// Handle to a running mock IMAP server. The server stops when this handle +/// is dropped. +pub struct MockImapServerHandle { + addr: SocketAddr, +} + +impl MockImapServerHandle { + pub fn host(&self) -> String { + self.addr.ip().to_string() + } + + pub fn port(&self) -> u16 { + self.addr.port() + } +} + +fn extract_tag(line: &str) -> Option<&str> { + line.split_whitespace().next() +} + +/// Replace `{TAG}` placeholders in `response` with `tag`. +fn substitute_tag(response: &[u8], tag: &str) -> Vec { + let placeholder = b"{TAG}"; + if response.is_empty() || !contains_slice(response, placeholder) { + return response.to_vec(); + } + let tag_bytes = tag.as_bytes(); + let mut result = Vec::with_capacity(response.len()); + let mut pos = 0; + while let Some(idx) = find_slice(&response[pos..], placeholder) { + result.extend_from_slice(&response[pos..pos + idx]); + result.extend_from_slice(tag_bytes); + pos += idx + placeholder.len(); + } + result.extend_from_slice(&response[pos..]); + result +} + +fn contains_slice(haystack: &[u8], needle: &[u8]) -> bool { + haystack.windows(needle.len()).any(|w| w == needle) +} + +fn find_slice(haystack: &[u8], needle: &[u8]) -> Option { + haystack + .windows(needle.len()) + .position(|w| w == needle) +} + +// ============================================================ +// Pre-built response helpers +// ============================================================ + +/// Build a tagged OK response. +pub fn ok(tag: impl AsRef, msg: impl AsRef) -> Vec { + format!("{} OK {}\r\n", tag.as_ref(), msg.as_ref()).into_bytes() +} + +/// Build a STATUS response line. +pub fn status_response( + mailbox: &str, + messages: u32, + unseen: u32, + uid_next: u32, + uid_validity: Option, +) -> Vec { + let uv = uid_validity + .map(|v| format!(" UIDVALIDITY {v}")) + .unwrap_or_default(); + let text = format!( + "* STATUS \"{mailbox}\" (MESSAGES {messages} UNSEEN {unseen} UIDNEXT {uid_next}{uv})\r\n" + ); + // Clients expect a tagged response after the untagged STATUS line. + // We produce a generic OK that works for any tag. + let mut out = text.into_bytes(); + out.extend_from_slice(b"{TAG} OK STATUS completed\r\n"); + out +} + +/// Build an EXAMINE response with mailbox data. +pub fn examine_response( + _mailbox: &str, + exists: u32, + uid_validity: u32, + uid_next: u32, +) -> Vec { + format!( + "* FLAGS (\\Seen \\Answered \\Flagged \\Deleted \\Draft)\r\n\ + * OK [PERMANENTFLAGS ()]\r\n\ + * {exists} EXISTS\r\n\ + * 0 RECENT\r\n\ + * OK [UIDVALIDITY {uid_validity}]\r\n\ + * OK [UIDNEXT {uid_next}]\r\n\ + * OK [HIGHESTMODSEQ 1]\r\n\ + {{TAG}} OK [READ-ONLY] EXAMINE completed\r\n" + ) + .into_bytes() +} + +/// Build a UID SEARCH response for the given UID list. +pub fn uid_search_response(uids: &[u32]) -> Vec { + let uid_str = uids + .iter() + .map(|u| u.to_string()) + .collect::>() + .join(" "); + format!("* SEARCH {uid_str}\r\n{{TAG}} OK SEARCH completed\r\n").into_bytes() +} + +/// Build a UID FETCH response returning full headers (for BODY[HEADER]). +/// Each entry: (uid, message_id) +pub fn uid_fetch_metadata_response(entries: &[(u32, &str)]) -> Vec { + let mut out = Vec::new(); + for (uid, msg_id) in entries { + // Build a minimal header that contains the Message-ID line. + let header_data = format!( + "From: sender@example.com\r\n\ +To: recipient@example.com\r\n\ +Date: Thu, 01 Jan 2025 00:00:00 +0000\r\n\ +Subject: test\r\n\ +Message-ID: {msg_id}\r\n\r\n" + ); + let header_len = header_data.len(); + let line = format!( + "* {uid} FETCH (UID {uid} BODY[HEADER] {{{header_len}}}\r\n\ +{header_data}\ +)\r\n", + ); + out.extend_from_slice(line.as_bytes()); + } + out.extend_from_slice(b"{TAG} OK FETCH completed\r\n"); + out +} + +/// Build a UID FETCH RFC822 response with a full email body. +pub fn uid_fetch_rfc822_response(uid: u32, eml: &[u8]) -> Vec { + let header = format!( + "* {uid} FETCH (UID {uid} RFC822 {{{len}}}\r\n", + len = eml.len() + ); + let mut out = header.into_bytes(); + out.extend_from_slice(eml); + out.extend_from_slice(b")\r\n{TAG} OK FETCH completed\r\n"); + out +} + +/// A minimal RFC822 email fixture for testing. +pub fn minimal_eml(subject: &str, message_id: &str) -> Vec { + format!( + "From: sender@example.com\r\n\ + To: recipient@example.com\r\n\ + Subject: {subject}\r\n\ + Message-ID: <{message_id}>\r\n\ + Date: Thu, 01 Jan 2025 00:00:00 +0000\r\n\ + MIME-Version: 1.0\r\n\ + Content-Type: text/plain; charset=utf-8\r\n\ + \r\n\ + This is a test email: {subject}.\r\n" + ) + .into_bytes() +} + +// ============================================================ +// Self-tests for the mock server itself +// ============================================================ + +#[cfg(test)] +mod tests { + use super::*; + + async fn connect_and_read_greeting(host: &str, port: u16) -> String { + let mut stream = TcpStream::connect((host, port)).await.unwrap(); + let (reader, _writer) = stream.split(); + let mut reader = BufReader::new(reader); + let mut line = String::new(); + reader.read_line(&mut line).await.unwrap(); + line + } + + async fn send_and_recv(host: &str, port: u16, cmd: &str) -> String { + let mut stream = TcpStream::connect((host, port)).await.unwrap(); + let (reader, mut writer) = stream.split(); + let mut reader = BufReader::new(reader); + + // Read greeting + let mut line = String::new(); + reader.read_line(&mut line).await.unwrap(); + + // Send command + writer.write_all(cmd.as_bytes()).await.unwrap(); + writer.write_all(b"\r\n").await.unwrap(); + + // Read response (may be multi-line; read until tagged response) + let mut out = String::new(); + loop { + line.clear(); + reader.read_line(&mut line).await.unwrap(); + out.push_str(&line); + if line.starts_with("A0") || line.starts_with("A1") { + break; + } + } + out + } + + #[tokio::test] + async fn test_mock_greeting() { + let handle = MockImapServer::new().start().await; + let greeting = connect_and_read_greeting(&handle.host(), handle.port()).await; + assert!(greeting.starts_with("* OK")); + } + + #[tokio::test] + async fn test_mock_scripted_response() { + let handle = MockImapServer::new() + .respond( + "LOGIN", + "A0 OK LOGIN completed\r\n", + ) + .start() + .await; + + let resp = send_and_recv(&handle.host(), handle.port(), "A0 LOGIN u p").await; + assert!(resp.contains("LOGIN completed")); + } + + #[tokio::test] + async fn test_mock_fallback_on_unmatched() { + let handle = MockImapServer::new().start().await; + + // Send a command that has no scripted response + let resp = send_and_recv(&handle.host(), handle.port(), "A0 NOOP").await; + assert!(resp.contains("OK done"), "unmatched command should get fallback OK"); + } + + #[tokio::test] + async fn test_status_response_helper() { + let resp = status_response("INBOX", 10, 2, 11, Some(42)); + let text = String::from_utf8(resp).unwrap(); + assert!(text.contains("MESSAGES 10")); + assert!(text.contains("UNSEEN 2")); + assert!(text.contains("UIDNEXT 11")); + assert!(text.contains("UIDVALIDITY 42")); + } + + #[tokio::test] + async fn test_status_response_without_uidvalidity() { + let resp = status_response("INBOX", 10, 2, 11, None); + let text = String::from_utf8(resp).unwrap(); + assert!(!text.contains("UIDVALIDITY")); + assert!(text.contains("MESSAGES 10")); + } + + #[tokio::test] + async fn test_examine_response() { + let resp = examine_response("INBOX", 10, 42, 11); + let text = String::from_utf8(resp).unwrap(); + assert!(text.contains("UIDVALIDITY 42")); + assert!(text.contains("10 EXISTS")); + } + + #[tokio::test] + async fn test_uid_search_response() { + let resp = uid_search_response(&[1, 3, 5]); + let text = String::from_utf8(resp).unwrap(); + assert!(text.contains("SEARCH 1 3 5")); + } + + #[tokio::test] + async fn test_uid_fetch_metadata_response() { + let resp = uid_fetch_metadata_response(&[(1, "msg-a@x.com"), (2, "msg-b@x.com")]); + let text = String::from_utf8(resp).unwrap(); + assert!(text.contains("Message-ID: msg-a@x.com")); + assert!(text.contains("Message-ID: msg-b@x.com")); + } + + #[tokio::test] + async fn test_multiple_commands_in_sequence() { + let handle = MockImapServer::new() + .respond("LOGIN", "A0 OK LOGIN\r\n") + .respond("STATUS", status_response("INBOX", 5, 1, 6, Some(99))) + .respond("LOGOUT", "* BYE\r\nA0 OK\r\n") + .start() + .await; + + let mut stream = TcpStream::connect((handle.host(), handle.port())) + .await + .unwrap(); + let (reader, mut writer) = stream.split(); + let mut reader = BufReader::new(reader); + + // Read greeting + let mut buf = String::new(); + reader.read_line(&mut buf).await.unwrap(); + + // LOGIN + writer.write_all(b"A0 LOGIN u p\r\n").await.unwrap(); + buf.clear(); + reader.read_line(&mut buf).await.unwrap(); + assert!(buf.contains("LOGIN")); + + // STATUS + writer + .write_all(b"A0 STATUS INBOX (MESSAGES UNSEEN UIDNEXT UIDVALIDITY)\r\n") + .await + .unwrap(); + buf.clear(); + // Read multi-line STATUS response (untagged line + tagged OK) + loop { + reader.read_line(&mut buf).await.unwrap(); + if buf.contains("UIDVALIDITY 99") { + // Consume the tagged OK line that follows + buf.clear(); + reader.read_line(&mut buf).await.unwrap(); + break; + } + } + + // LOGOUT + writer.write_all(b"A0 LOGOUT\r\n").await.unwrap(); + buf.clear(); + reader.read_line(&mut buf).await.unwrap(); + assert!(buf.contains("BYE")); + } + + #[tokio::test] + async fn test_tag_substitution_in_response() { + // Use {TAG} placeholder in the response and verify it gets the + // client's actual tag ("A5") substituted in. + let handle = MockImapServer::new() + .respond("LOGIN", "{TAG} OK LOGIN succeeded\r\n") + .start() + .await; + + let mut stream = TcpStream::connect((handle.host(), handle.port())) + .await + .unwrap(); + let (reader, mut writer) = stream.split(); + let mut reader = BufReader::new(reader); + + // Read greeting + let mut buf = String::new(); + reader.read_line(&mut buf).await.unwrap(); + + // Send LOGIN with non-standard tag + writer.write_all(b"A5 LOGIN u p\r\n").await.unwrap(); + buf.clear(); + reader.read_line(&mut buf).await.unwrap(); + + assert!( + buf.contains("A5 OK LOGIN succeeded"), + "expected 'A5 OK LOGIN succeeded', got '{buf}'" + ); + } + + #[tokio::test] + async fn test_tag_substitution_multiple_placeholders() { + let handle = MockImapServer::new() + .respond("NOOP", "* 0 RECENT\r\n{TAG} OK NOOP done\r\n") + .start() + .await; + + let mut stream = TcpStream::connect((handle.host(), handle.port())) + .await + .unwrap(); + let (reader, mut writer) = stream.split(); + let mut reader = BufReader::new(reader); + + // Read greeting + let mut buf = String::new(); + reader.read_line(&mut buf).await.unwrap(); + + // Send with tag "B99" + writer.write_all(b"B99 NOOP\r\n").await.unwrap(); + + // Read all lines + let mut all = String::new(); + loop { + buf.clear(); + reader.read_line(&mut buf).await.unwrap(); + all.push_str(&buf); + if buf.starts_with("B99") { + break; + } + } + + assert!(all.contains("* 0 RECENT\r\n")); + assert!(all.contains("B99 OK NOOP done\r\n")); + } +} diff --git a/crates/core/src/imap/mod.rs b/crates/core/src/imap/mod.rs index 479c954..0866f1a 100644 --- a/crates/core/src/imap/mod.rs +++ b/crates/core/src/imap/mod.rs @@ -26,3 +26,5 @@ pub mod session; pub mod stats; #[cfg(test)] mod tests; +#[cfg(test)] +pub mod mock_server; diff --git a/crates/core/src/store/tantivy/envelope.rs b/crates/core/src/store/tantivy/envelope.rs index 1a5e97a..a7ccac7 100644 --- a/crates/core/src/store/tantivy/envelope.rs +++ b/crates/core/src/store/tantivy/envelope.rs @@ -278,6 +278,80 @@ impl IndexManager { Box::new(boolean_query) } + /// Return all Message-IDs stored in Tantivy for a given mailbox. + /// Prefer `mailbox_contains_message_id` for existence checks on large + /// mailboxes — this method loads everything into a HashSet. + pub fn get_message_ids_for_mailbox( + &self, + account_id: u64, + mailbox_id: u64, + ) -> BichonResult> { + let query = self.mailbox_query(account_id, mailbox_id); + let fields = SchemaTools::email_fields(); + let searcher = self.create_searcher()?; + + let docs = searcher + .search(&query, &DocSetCollector) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + + let mut result = HashSet::new(); + for doc_address in docs { + let doc = searcher + .doc::(doc_address) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + + if let Some(v) = doc.get_first(fields.f_message_id) { + if let Some(s) = v.as_str() { + if !s.is_empty() { + result.insert(s.to_string()); + } + } + } + } + Ok(result) + } + + /// Check whether a specific Message-ID exists in a mailbox. + /// Uses a TermQuery — O(1) per call, no allocation proportional to + /// mailbox size. Suitable for large mailboxes where + /// `get_message_ids_for_mailbox` would allocate too much memory. + pub fn mailbox_contains_message_id( + &self, + account_id: u64, + mailbox_id: u64, + message_id: &str, + ) -> BichonResult { + let fields = SchemaTools::email_fields(); + let query = BooleanQuery::new(vec![ + ( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_u64(fields.f_account_id, account_id), + IndexRecordOption::Basic, + )), + ), + ( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_u64(fields.f_mailbox_id, mailbox_id), + IndexRecordOption::Basic, + )), + ), + ( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_text(fields.f_message_id, message_id), + IndexRecordOption::Basic, + )), + ), + ]); + let searcher = self.create_searcher()?; + let count = searcher + .search(&query, &Count) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + Ok(count > 0) + } + fn envelope_query(&self, account_id: u64, eid: &str) -> Box { let account_id_query = TermQuery::new( Term::from_field_u64(SchemaTools::email_fields().f_account_id, account_id), @@ -2069,4 +2143,362 @@ mod tests { "body should be absent when EML is missing" ); } + + // ── get_message_ids_for_mailbox ───────────────────────────────── + + #[test] + fn get_message_ids_returns_stored_ids() { + let f = SchemaTools::email_fields(); + let index = Index::create_in_ram(SchemaTools::email_schema()); + index.tokenizers().register("euro", EuroTokenizer::new()); + + // Insert two docs for mailbox 10, one for mailbox 20 + { + let mut writer = index + .writer_with_num_threads(1, 15_000_000) + .expect("writer"); + + let mut doc1 = TantivyDocument::new(); + doc1.add_u64(f.f_account_id, 1); + doc1.add_u64(f.f_mailbox_id, 10); + doc1.add_text(f.f_message_id, ""); + doc1.add_text(f.f_id, "id-a"); + doc1.add_u64(f.f_uid, 1); + doc1.add_text(f.f_content_hash, "hash-a"); + writer.add_document(doc1).unwrap(); + + let mut doc2 = TantivyDocument::new(); + doc2.add_u64(f.f_account_id, 1); + doc2.add_u64(f.f_mailbox_id, 10); + doc2.add_text(f.f_message_id, ""); + doc2.add_text(f.f_id, "id-b"); + doc2.add_u64(f.f_uid, 2); + doc2.add_text(f.f_content_hash, "hash-b"); + writer.add_document(doc2).unwrap(); + + let mut doc3 = TantivyDocument::new(); + doc3.add_u64(f.f_account_id, 1); + doc3.add_u64(f.f_mailbox_id, 20); + doc3.add_text(f.f_message_id, ""); + doc3.add_text(f.f_id, "id-c"); + doc3.add_u64(f.f_uid, 3); + doc3.add_text(f.f_content_hash, "hash-c"); + writer.add_document(doc3).unwrap(); + + writer.commit().unwrap(); + } + + let reader = index.reader().unwrap(); + reader.reload().unwrap(); + let searcher = reader.searcher(); + + // We can't easily call ENVELOPE_MANAGER.get_message_ids_for_mailbox + // because it reads from ENVELOPE_MANAGER's own index, not our in-memory one. + // Instead, test the query pattern directly. + let query: Box = { + let account_query = TermQuery::new( + Term::from_field_u64(f.f_account_id, 1), + IndexRecordOption::Basic, + ); + let mailbox_query = TermQuery::new( + Term::from_field_u64(f.f_mailbox_id, 10), + IndexRecordOption::Basic, + ); + Box::new(BooleanQuery::new(vec![ + (Occur::Must, Box::new(account_query)), + (Occur::Must, Box::new(mailbox_query)), + ])) + }; + + let docs = searcher + .search(&query, &DocSetCollector) + .unwrap(); + + let mut ids: Vec = Vec::new(); + for addr in docs { + let doc: TantivyDocument = searcher.doc(addr).unwrap(); + if let Some(v) = doc.get_first(f.f_message_id) { + if let Some(s) = v.as_str() { + ids.push(s.to_string()); + } + } + } + ids.sort(); + + assert_eq!(ids, vec!["", ""]); + } + + #[test] + fn get_message_ids_empty_mailbox_returns_empty() { + let f = SchemaTools::email_fields(); + let index = Index::create_in_ram(SchemaTools::email_schema()); + index.tokenizers().register("euro", EuroTokenizer::new()); + + { + let mut writer = index + .writer_with_num_threads(1, 15_000_000) + .expect("writer"); + + // Doc for a different mailbox + let mut doc = TantivyDocument::new(); + doc.add_u64(f.f_account_id, 1); + doc.add_u64(f.f_mailbox_id, 99); + doc.add_text(f.f_message_id, ""); + doc.add_text(f.f_id, "id-other"); + doc.add_u64(f.f_uid, 1); + doc.add_text(f.f_content_hash, "hash-other"); + writer.add_document(doc).unwrap(); + writer.commit().unwrap(); + } + + let reader = index.reader().unwrap(); + reader.reload().unwrap(); + let searcher = reader.searcher(); + + let query: Box = { + let account_query = TermQuery::new( + Term::from_field_u64(f.f_account_id, 1), + IndexRecordOption::Basic, + ); + let mailbox_query = TermQuery::new( + Term::from_field_u64(f.f_mailbox_id, 10), + IndexRecordOption::Basic, + ); + Box::new(BooleanQuery::new(vec![ + (Occur::Must, Box::new(account_query)), + (Occur::Must, Box::new(mailbox_query)), + ])) + }; + + let docs = searcher.search(&query, &DocSetCollector).unwrap(); + assert!(docs.is_empty()); + } + + // ── mailbox_contains_message_id ─────────────────────────────── + + #[test] + fn mailbox_contains_message_id_finds_existing() { + let f = SchemaTools::email_fields(); + let index = Index::create_in_ram(SchemaTools::email_schema()); + index.tokenizers().register("euro", EuroTokenizer::new()); + + { + let mut writer = index + .writer_with_num_threads(1, 15_000_000) + .expect("writer"); + + let mut doc = TantivyDocument::new(); + doc.add_u64(f.f_account_id, 1); + doc.add_u64(f.f_mailbox_id, 10); + doc.add_text(f.f_message_id, "abc@example.com"); + doc.add_text(f.f_id, "id-1"); + doc.add_u64(f.f_uid, 1); + doc.add_text(f.f_content_hash, "hash-1"); + writer.add_document(doc).unwrap(); + writer.commit().unwrap(); + } + + // We test the query pattern directly (can't call ENVELOPE_MANAGER + // which uses a different index). + let reader = index.reader().unwrap(); + reader.reload().unwrap(); + let searcher = reader.searcher(); + + let query = BooleanQuery::new(vec![ + ( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_u64(f.f_account_id, 1), + IndexRecordOption::Basic, + )), + ), + ( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_u64(f.f_mailbox_id, 10), + IndexRecordOption::Basic, + )), + ), + ( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_text(f.f_message_id, "abc@example.com"), + IndexRecordOption::Basic, + )), + ), + ]); + + let count = searcher.search(&query, &Count).unwrap(); + assert_eq!(count, 1); + } + + #[test] + fn mailbox_contains_message_id_returns_zero_for_missing() { + let f = SchemaTools::email_fields(); + let index = Index::create_in_ram(SchemaTools::email_schema()); + index.tokenizers().register("euro", EuroTokenizer::new()); + + { + let mut writer = index + .writer_with_num_threads(1, 15_000_000) + .expect("writer"); + + let mut doc = TantivyDocument::new(); + doc.add_u64(f.f_account_id, 1); + doc.add_u64(f.f_mailbox_id, 10); + doc.add_text(f.f_message_id, "existing@example.com"); + doc.add_text(f.f_id, "id-1"); + doc.add_u64(f.f_uid, 1); + doc.add_text(f.f_content_hash, "hash-1"); + writer.add_document(doc).unwrap(); + writer.commit().unwrap(); + } + + let reader = index.reader().unwrap(); + reader.reload().unwrap(); + let searcher = reader.searcher(); + + let query = BooleanQuery::new(vec![ + ( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_u64(f.f_account_id, 1), + IndexRecordOption::Basic, + )), + ), + ( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_u64(f.f_mailbox_id, 10), + IndexRecordOption::Basic, + )), + ), + ( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_text(f.f_message_id, "nonexistent@example.com"), + IndexRecordOption::Basic, + )), + ), + ]); + + let count = searcher.search(&query, &Count).unwrap(); + assert_eq!(count, 0); + } + + #[test] + fn mailbox_contains_message_id_respects_mailbox_boundary() { + let f = SchemaTools::email_fields(); + let index = Index::create_in_ram(SchemaTools::email_schema()); + index.tokenizers().register("euro", EuroTokenizer::new()); + + { + let mut writer = index + .writer_with_num_threads(1, 15_000_000) + .expect("writer"); + + // Same Message-ID in mailbox 10 + let mut doc1 = TantivyDocument::new(); + doc1.add_u64(f.f_account_id, 1); + doc1.add_u64(f.f_mailbox_id, 10); + doc1.add_text(f.f_message_id, "shared@example.com"); + doc1.add_text(f.f_id, "id-1"); + doc1.add_u64(f.f_uid, 1); + doc1.add_text(f.f_content_hash, "hash-1"); + writer.add_document(doc1).unwrap(); + + // Same Message-ID in mailbox 20 (different mailbox) + let mut doc2 = TantivyDocument::new(); + doc2.add_u64(f.f_account_id, 1); + doc2.add_u64(f.f_mailbox_id, 20); + doc2.add_text(f.f_message_id, "shared@example.com"); + doc2.add_text(f.f_id, "id-2"); + doc2.add_u64(f.f_uid, 2); + doc2.add_text(f.f_content_hash, "hash-2"); + writer.add_document(doc2).unwrap(); + writer.commit().unwrap(); + } + + let reader = index.reader().unwrap(); + reader.reload().unwrap(); + let searcher = reader.searcher(); + + // Query mailbox 10: should find 1 + let q10 = BooleanQuery::new(vec![ + ( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_u64(f.f_account_id, 1), + IndexRecordOption::Basic, + )), + ), + ( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_u64(f.f_mailbox_id, 10), + IndexRecordOption::Basic, + )), + ), + ( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_text(f.f_message_id, "shared@example.com"), + IndexRecordOption::Basic, + )), + ), + ]); + assert_eq!(searcher.search(&q10, &Count).unwrap(), 1); + + // Query mailbox 20: should find 1 + let q20 = BooleanQuery::new(vec![ + ( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_u64(f.f_account_id, 1), + IndexRecordOption::Basic, + )), + ), + ( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_u64(f.f_mailbox_id, 20), + IndexRecordOption::Basic, + )), + ), + ( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_text(f.f_message_id, "shared@example.com"), + IndexRecordOption::Basic, + )), + ), + ]); + assert_eq!(searcher.search(&q20, &Count).unwrap(), 1); + + // Query mailbox 99 (no docs): should find 0 + let q99 = BooleanQuery::new(vec![ + ( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_u64(f.f_account_id, 1), + IndexRecordOption::Basic, + )), + ), + ( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_u64(f.f_mailbox_id, 99), + IndexRecordOption::Basic, + )), + ), + ( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_text(f.f_message_id, "shared@example.com"), + IndexRecordOption::Basic, + )), + ), + ]); + assert_eq!(searcher.search(&q99, &Count).unwrap(), 0); + } }