diff --git a/crates/core/src/cache/imap/download/flow.rs b/crates/core/src/cache/imap/download/flow.rs index 025a6f3..1da97f9 100644 --- a/crates/core/src/cache/imap/download/flow.rs +++ b/crates/core/src/cache/imap/download/flow.rs @@ -54,7 +54,7 @@ pub async fn fetch_and_save_by_date( mailbox: &MailBox, direction: FetchDirection, token: CancellationToken, -) -> BichonResult<()> { +) -> BichonResult> { let account_id = account.id; let mut session = match ImapExecutor::create_connection(account_id).await { Ok(session) => session, @@ -108,32 +108,18 @@ pub async fn fetch_and_save_by_date( FolderStatus::Success, None, )?; - return Ok(()); + return Ok(None); } - let folder_limit = account.folder_limit; // sort small -> bigger let mut uid_vec: Vec = uid_list.into_iter().collect(); uid_vec.sort(); - if let Some(limit) = folder_limit { - let limit = limit.max(100) as usize; - if len > limit { - uid_vec = match direction { - FetchDirection::Since => uid_vec.split_off(len - limit), - FetchDirection::Before => { - uid_vec.truncate(limit); - uid_vec - } - }; - } - } - + let max_uid = uid_vec.last().copied(); let planned = uid_vec.len() as u64; let uid_batches = generate_uid_sequence_hashset( uid_vec, account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE) as usize, - false, ); DownloadState::update_folder_progress( account_id, @@ -212,14 +198,16 @@ pub async fn fetch_and_save_by_date( )?; } session.logout().await.ok(); - Ok(()) + Ok(max_uid) } +/// Fetches all messages from a mailbox. +/// Returns `Ok(Some(max_uid))` with the highest UID stored, or `Ok(None)` if empty. pub async fn fetch_and_save_full_mailbox( account: &AccountModel, mailbox: &MailBox, token: CancellationToken, -) -> BichonResult<()> { +) -> BichonResult> { let mailbox_id = mailbox.id; let account_id = account.id; @@ -262,33 +250,17 @@ pub async fn fetch_and_save_full_mailbox( } }; - let folder_limit = account.folder_limit; - let total_to_fetch = match folder_limit { - Some(limit) if (limit as u64) < total => { - let limit64 = limit as u64; - total.min(limit64.max(100)) - } - _ => total, - }; - - let page_size = if let Some(limit) = folder_limit { - limit - .max(100) - .min(account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE)) - } else { - account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE) - }; - - let total_batches = total_to_fetch.div_ceil(page_size as u64); - let desc = folder_limit.is_some(); + let page_size = account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE); + let total_batches = total.div_ceil(page_size as u64); info!( - "Starting full mailbox download for '{}', total={}, limit={:?}, batches={}, desc={}", - mailbox.name, total, folder_limit, total_batches, desc + "Starting full mailbox download for '{}', total={}, batches={}", + mailbox.name, total, total_batches ); let mut current_processed = 0u64; let mut has_error_or_cancel = false; + let mut max_uid: Option = None; for page in 1..=total_batches { if token.is_cancelled() { @@ -300,7 +272,7 @@ pub async fn fetch_and_save_full_mailbox( DownloadState::update_folder_progress( account_id, mailbox.name.clone(), - total_to_fetch, + total, current_processed, FolderStatus::Cancelled, None, @@ -313,12 +285,12 @@ pub async fn fetch_and_save_full_mailbox( &mut session, account_id, mailbox_id, - total_to_fetch, + total, page as u64, page_size as u64, &mailbox.encoded_name(), - desc, token.clone(), + &mut max_uid, ) .await { @@ -327,7 +299,7 @@ pub async fn fetch_and_save_full_mailbox( DownloadState::update_folder_progress( account_id, mailbox.name.clone(), - total_to_fetch, + total, current_processed, FolderStatus::Downloading, None, @@ -339,7 +311,7 @@ pub async fn fetch_and_save_full_mailbox( DownloadState::update_folder_progress( account_id, mailbox.name.clone(), - total_to_fetch, + total, current_processed, FolderStatus::Failed, Some(err_msg), @@ -354,28 +326,24 @@ pub async fn fetch_and_save_full_mailbox( DownloadState::update_folder_progress( account_id, mailbox.name.clone(), - total_to_fetch, + total, current_processed, FolderStatus::Success, None, )?; } session.logout().await.ok(); - Ok(()) + Ok(max_uid) } pub fn generate_uid_sequence_hashset( unique_nums: Vec, chunk_size: usize, - desc: bool, ) -> Vec<(String, u64)> { assert!(!unique_nums.is_empty()); - let mut nums = unique_nums; - if desc { - nums.reverse(); - } let mut result = Vec::new(); + let nums = unique_nums; for chunk in nums.chunks(chunk_size) { let size = chunk.len() as u64; @@ -448,7 +416,7 @@ pub async fn reconcile_mailboxes( break; } - if local_mailbox.uid_validity != remote_mailbox.uid_validity { + let new_highest_uid = if local_mailbox.uid_validity != remote_mailbox.uid_validity { if remote_mailbox.uid_validity.is_none() { let err_msg = format!( "Mailbox '{}' logic error: Server did not provide UIDVALIDITY.", @@ -493,7 +461,7 @@ pub async fn reconcile_mailboxes( FetchDirection::Since, token.clone(), ) - .await?; + .await? } None => match &account.date_before { Some(r) => { @@ -505,7 +473,7 @@ pub async fn reconcile_mailboxes( FetchDirection::Before, token.clone(), ) - .await?; + .await? } None => { rebuild_mailbox_cache( @@ -520,10 +488,12 @@ pub async fn reconcile_mailboxes( } } else { perform_incremental_sync(account, local_mailbox, remote_mailbox, token.clone()) - .await?; - } + .await? + }; - mailboxes_to_update.push(remote_mailbox.clone()); + let mut updated = remote_mailbox.clone(); + updated.highest_uid = new_highest_uid; + mailboxes_to_update.push(updated); } //The metadata of this mailbox must only be updated after a successful synchronization; //otherwise, it may cause synchronization errors and result in missing emails in the local sync results. @@ -598,7 +568,11 @@ pub async fn reconcile_mailboxes( }; match result { - Ok(_) => {} + Ok(new_highest_uid) => { + let mut updated = mailbox.clone(); + updated.highest_uid = new_highest_uid; + MailBox::batch_upsert(&[updated])?; + } Err(err) => { has_error = true; tracing::error!("Folder sync task failed: {:#?}", err); @@ -622,64 +596,88 @@ pub async fn reconcile_mailboxes( } //only check new emails and sync +/// Incrementally syncs a mailbox. +/// Returns the new highest UID after sync, or `None` if nothing changed. async fn perform_incremental_sync( account: &AccountModel, local_mailbox: &MailBox, remote_mailbox: &MailBox, token: CancellationToken, -) -> BichonResult<()> { +) -> BichonResult> { if remote_mailbox.exists > 0 { - let local_max_uid = ENVELOPE_MANAGER.get_max_uid(account.id, local_mailbox.id)?; - tracing::info!( - "[account {}][mailbox {}] perform_incremental_sync: local_max_uid={:?}, remote.exists={}", - account.id, - local_mailbox.name, - local_max_uid, - remote_mailbox.exists - ); - match local_max_uid { - Some(max_uid) => { - let mut session = ImapExecutor::create_connection(account.id).await?; - let before_date = account - .date_before - .as_ref() - .map(|r| r.calculate_date()) - .transpose()?; - - ImapExecutor::fetch_new_mail( - &mut session, - account, - local_mailbox, - max_uid + 1, - before_date.as_deref(), - token, - ) - .await?; - session.logout().await.ok(); + // Use stored highest_uid if available; otherwise fall back to Tantivy + // query once (backward compatibility with pre-existing databases). + let start_uid = match local_mailbox.highest_uid { + Some(uid) => { + tracing::info!( + "[account {}][mailbox {}] perform_incremental_sync: stored highest_uid={}, remote.exists={}", + account.id, + local_mailbox.name, + uid, + remote_mailbox.exists + ); + uid as u64 + 1 } None => { - info!( - "No maximum UID found in index for mailbox, assuming local cache is missing." + let local_max_uid = + ENVELOPE_MANAGER.get_max_uid(account.id, local_mailbox.id)?; + tracing::info!( + "[account {}][mailbox {}] perform_incremental_sync: highest_uid unset, Tantivy max_uid={:?}, remote.exists={}", + account.id, + local_mailbox.name, + local_max_uid, + remote_mailbox.exists ); - - match &account.date_since { - Some(date_since) => { - fetch_and_save_by_date( - account, - date_since.since_date()?.as_str(), - remote_mailbox, - FetchDirection::Since, - token, - ) - .await?; - } + match local_max_uid { + Some(uid) => uid + 1, None => { - fetch_and_save_full_mailbox(account, remote_mailbox, token).await?; + info!( + "No maximum UID found in index for mailbox, assuming local cache is missing." + ); + + let result = match &account.date_since { + Some(date_since) => { + fetch_and_save_by_date( + account, + date_since.since_date()?.as_str(), + remote_mailbox, + FetchDirection::Since, + token, + ) + .await? + } + None => { + fetch_and_save_full_mailbox(account, remote_mailbox, token) + .await? + } + }; + return Ok(result); } } } - } - } + }; - Ok(()) + let mut session = ImapExecutor::create_connection(account.id).await?; + let before_date = account + .date_before + .as_ref() + .map(|r| r.calculate_date()) + .transpose()?; + + let new_max_uid = ImapExecutor::fetch_new_mail( + &mut session, + account, + local_mailbox, + start_uid, + before_date.as_deref(), + token, + ) + .await?; + session.logout().await.ok(); + + // Keep existing highest_uid if no new mail was fetched. + Ok(new_max_uid.or(local_mailbox.highest_uid)) + } else { + Ok(local_mailbox.highest_uid) + } } diff --git a/crates/core/src/cache/imap/download/rebuild.rs b/crates/core/src/cache/imap/download/rebuild.rs index 85c175d..96d31d9 100644 --- a/crates/core/src/cache/imap/download/rebuild.rs +++ b/crates/core/src/cache/imap/download/rebuild.rs @@ -89,7 +89,11 @@ pub async fn rebuild_cache( }; match fetch_and_save_full_mailbox(&account, &mailbox, token.clone()).await { - Ok(_) => {} + Ok(new_highest_uid) => { + let mut updated = mailbox.clone(); + updated.highest_uid = new_highest_uid; + MailBox::batch_upsert(&[updated])?; + } Err(err) => { has_error = true; tracing::error!("Folder sync task failed: {:#?}", err); @@ -169,7 +173,11 @@ pub async fn rebuild_cache_by_date( match fetch_and_save_by_date(&account, date.as_str(), &mailbox, direction, token.clone()) .await { - Ok(_) => {} + Ok(new_highest_uid) => { + let mut updated = mailbox.clone(); + updated.highest_uid = new_highest_uid; + MailBox::batch_upsert(&[updated])?; + } Err(err) => { has_error = true; tracing::error!("Folder sync task failed: {:#?}", err); @@ -196,7 +204,7 @@ pub async fn rebuild_mailbox_cache( local_mailbox: &MailBox, remote_mailbox: &MailBox, token: CancellationToken, -) -> BichonResult<()> { +) -> BichonResult> { ENVELOPE_MANAGER .delete_mailbox_envelopes(account.id, vec![local_mailbox.id]) .await?; @@ -217,11 +225,11 @@ pub async fn rebuild_mailbox_cache( FolderStatus::Success, None, )?; - return Ok(()); + return Ok(None); } - fetch_and_save_full_mailbox(account, remote_mailbox, token).await?; - Ok(()) + let result = fetch_and_save_full_mailbox(account, remote_mailbox, token).await?; + Ok(result) } pub async fn rebuild_mailbox_cache_by_date( @@ -231,7 +239,7 @@ pub async fn rebuild_mailbox_cache_by_date( remote: &MailBox, direction: FetchDirection, token: CancellationToken, -) -> BichonResult<()> { +) -> BichonResult> { ENVELOPE_MANAGER .delete_mailbox_envelopes(account.id, vec![local_mailbox_id]) .await?; @@ -252,9 +260,9 @@ pub async fn rebuild_mailbox_cache_by_date( FolderStatus::Success, None, )?; - return Ok(()); + return Ok(None); } - fetch_and_save_by_date(account, date, remote, direction, token).await?; - Ok(()) + let result = fetch_and_save_by_date(account, date, remote, direction, token).await?; + Ok(result) } diff --git a/crates/core/src/cache/imap/mailbox.rs b/crates/core/src/cache/imap/mailbox.rs index b85e712..db11e4e 100644 --- a/crates/core/src/cache/imap/mailbox.rs +++ b/crates/core/src/cache/imap/mailbox.rs @@ -56,6 +56,10 @@ pub struct MailBox { /// The validity identifier for UIDs in this mailbox, used to ensure UID consistency across sessions. /// If `None`, the IMAP server has not provided this information. pub uid_validity: Option, + /// The highest UID that has been successfully downloaded and stored locally. + /// Used for incremental sync: next fetch starts from `highest_uid + 1`. + /// If `None`, a fallback query against the Tantivy index will be performed once. + pub highest_uid: Option, } impl MemDbModel for MailBox { diff --git a/crates/core/src/imap/executor.rs b/crates/core/src/imap/executor.rs index 422dfed..561fc56 100644 --- a/crates/core/src/imap/executor.rs +++ b/crates/core/src/imap/executor.rs @@ -17,8 +17,7 @@ // along with this program. If not, see . use crate::account::migration::AccountModel; -use crate::account::state::{DownloadState, FolderStatus}; -use crate::cache::imap::download::flow::{generate_uid_sequence_hashset, DEFAULT_BATCH_SIZE}; +use crate::account::state::{DownloadState, DownloadStatus, FolderStatus}; use crate::cache::imap::mailbox::MailBox; use crate::envelope::extractor::extract_envelope_and_store_it; use crate::error::code::ErrorCode; @@ -80,6 +79,14 @@ impl ImapExecutor { .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed)) } + /// Fetches new mail for a mailbox by directly issuing a ranged UID FETCH. + /// + /// Unlike the old two-step approach (UID SEARCH → batch UID FETCH), + /// this sends a single `UID FETCH {start}:*` and streams the results, + /// eliminating one IMAP round-trip. + /// + /// Returns `Ok(Some(max_uid))` with the highest UID fetched, or `Ok(None)` + /// if no new mail was found. pub async fn fetch_new_mail( session: &mut Session>, account: &AccountModel, @@ -87,122 +94,90 @@ impl ImapExecutor { start_uid: u64, before: Option<&str>, token: CancellationToken, - ) -> BichonResult<()> { + ) -> BichonResult> { assert!(start_uid > 0, "start_uid must be greater than 0"); - let query = match before { - Some(date) => format!("UID {start_uid}:* BEFORE {date}"), - None => format!("UID {start_uid}:*"), + // Select the mailbox (read-only) so UID FETCH works. + session + .examine(&mailbox.encoded_name()) + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?; + + // Build the UID range. IMAP UID FETCH accepts the same range syntax + // as UID SEARCH, so we can skip the separate SEARCH round-trip. + let uid_range = match before { + Some(date) => format!("{start_uid}:* BEFORE {date}"), + None => format!("{start_uid}:*"), }; - let uid_list = match Self::uid_search(session, &mailbox.encoded_name(), &query).await { - Ok(uid_list) => uid_list, - Err(e) => { - let err_msg = format!("UID search failed in [{}]: {:#?}", mailbox.name, e); - DownloadState::update_folder_progress( + info!( + "[account {}][mailbox {}] fetch_new_mail: direct UID FETCH {}", + account.id, mailbox.name, uid_range + ); + + let mut stream = session + .uid_fetch(&uid_range, BODY_FETCH_COMMAND) + .await + .map_err(|e| { + let err_msg = format!( + "UID FETCH failed in [{}]: {:#?}", + mailbox.name, e + ); + let _ = DownloadState::append_session_error(account.id, err_msg); + raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed) + })?; + + let mut count = 0u64; + let mut max_uid: Option = None; + while let Some(fetch) = stream + .try_next() + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))? + { + if token.is_cancelled() { + tracing::info!( + "Account {}: fetch_new_mail stream interrupted.", + account.id + ); + DownloadState::update_session_status( account.id, - mailbox.name.clone(), - 0, - 0, - FolderStatus::Failed, - Some(err_msg.clone()), + DownloadStatus::Cancelled, + Some("User stopped or system shutdown".to_string()), )?; - DownloadState::append_session_error(account.id, err_msg)?; - return Err(e); + return Err(raise_error!( + "Stream cancelled".into(), + ErrorCode::InternalError + )); } - }; - let len = uid_list.len(); - if len == 0 { - let msg = match before { - Some(date) => format!("No emails found before {}.", date), - None => "No new emails found.".into(), - }; + if let Some(uid) = fetch.uid { + max_uid = Some(max_uid.unwrap_or(0).max(uid)); + } + extract_envelope_and_store_it(fetch, account.id, mailbox.id).await?; + count += 1; + } + + if count == 0 { DownloadState::update_folder_progress( account.id, mailbox.name.clone(), 0, 0, FolderStatus::Success, - Some(msg), + Some("No new emails found.".into()), )?; - return Ok(()); - } - info!( - "[account {}][mailbox {}] {} envelopes need to be fetched (start_uid={})", - account.id, mailbox.name, len, start_uid - ); - tracing::debug!( - "[account {}][mailbox {}] fetch_new_mail UID range: {}..{} ({} uids)", - account.id, - mailbox.name, - start_uid, - uid_list.iter().max().unwrap_or(&0), - len - ); - - let mut uid_vec: Vec = uid_list.into_iter().collect(); - uid_vec.sort(); - let uid_batches = generate_uid_sequence_hashset( - uid_vec, - account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE) as usize, - false, - ); - let mut current_processed = 0u64; - let mut has_error_or_cancel = false; - for (index, batch) in uid_batches.into_iter().enumerate() { - if token.is_cancelled() { - break; - } - match Self::uid_batch_retrieve_emails( - session, - account.id, - mailbox.id, - &batch.0, - token.clone(), - ) - .await - { - Ok(_) => { - current_processed += batch.1; - DownloadState::update_folder_progress( - account.id, - mailbox.name.clone(), - len as u64, - current_processed, - 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, - mailbox.name.clone(), - len as u64, - current_processed, - FolderStatus::Failed, - Some(err_msg), - )?; - has_error_or_cancel = true; - break; - } - } - } - - if !has_error_or_cancel { + } else { DownloadState::update_folder_progress( account.id, mailbox.name.clone(), - len as u64, - current_processed, + count, + count, FolderStatus::Success, None, )?; } - Ok(()) + Ok(max_uid) } pub async fn batch_retrieve_emails( @@ -213,36 +188,23 @@ impl ImapExecutor { page: u64, page_size: u64, encoded_mailbox_name: &str, - desc: bool, token: CancellationToken, + max_uid: &mut Option, ) -> BichonResult { assert!(page > 0, "Page number must be greater than 0"); assert!(page_size > 0, "Page size must be greater than 0"); - let (start, end) = if desc { - // Fetch messages starting from the newest (descending order) - let end = total.saturating_sub((page - 1) * page_size); - if end == 0 { - return Ok(0); - } - // Calculate start as end - page_size + 1 to avoid off-by-one errors - let start = end.saturating_sub(page_size - 1).max(1); - (start, end) - } else { - // Fetch messages starting from the oldest (ascending order) - let start = (page - 1) * page_size + 1; - if start > total { - return Ok(0); - } - // Calculate end, capped by the total number of messages - let end = (start + page_size - 1).min(total); - (start, end) - }; + // Fetch messages starting from the oldest (ascending order). + let start = (page - 1) * page_size + 1; + if start > total { + return Ok(0); + } + let end = (start + page_size - 1).min(total); let sequence_set = format!("{}:{}", start, end); info!( - "Fetching mailbox '{}' messages: sequence {} (page {}, page_size {}, desc={})", - encoded_mailbox_name, sequence_set, page, page_size, desc + "Fetching mailbox '{}' messages: sequence {} (page {}, page_size {})", + encoded_mailbox_name, sequence_set, page, page_size ); let mut stream = session @@ -263,6 +225,9 @@ impl ImapExecutor { ErrorCode::InternalError )); } + if let Some(uid) = fetch.uid { + *max_uid = Some((*max_uid).unwrap_or(0).max(uid)); + } extract_envelope_and_store_it(fetch, account_id, mailbox_id).await?; count += 1; } diff --git a/crates/core/src/import/mod.rs b/crates/core/src/import/mod.rs index 8cd533e..13d85b0 100644 --- a/crates/core/src/import/mod.rs +++ b/crates/core/src/import/mod.rs @@ -105,6 +105,7 @@ impl ImportEmls { unseen: None, uid_next: None, uid_validity: None, + highest_uid: None, }; let mailbox_id = mailbox.id; // Upsert the mailbox, creating it if it doesn't exist diff --git a/crates/core/src/mailbox/list.rs b/crates/core/src/mailbox/list.rs index 667e220..5fb1068 100644 --- a/crates/core/src/mailbox/list.rs +++ b/crates/core/src/mailbox/list.rs @@ -157,8 +157,13 @@ async fn fetch_remote_with_progress(account_id: u64) -> BichonResult BichonResult<()> { unseen: None, uid_next: None, uid_validity: None, + highest_uid: None, }; let mailbox_id = mailbox.id;