From 76ab16b55ba03344dd44a1d78299819c8747f0ff Mon Sep 17 00:00:00 2001 From: rustmailer Date: Mon, 29 Dec 2025 12:06:04 +0800 Subject: [PATCH] fetch: support fetching mails before a specified date --- src/modules/cache/imap/sync/flow.rs | 88 +++++++++++++++++++++----- src/modules/cache/imap/sync/mod.rs | 25 ++++++-- src/modules/cache/imap/sync/rebuild.rs | 32 ++++++---- src/modules/imap/executor.rs | 9 ++- 4 files changed, 119 insertions(+), 35 deletions(-) diff --git a/src/modules/cache/imap/sync/flow.rs b/src/modules/cache/imap/sync/flow.rs index 07e230e..fd16c69 100644 --- a/src/modules/cache/imap/sync/flow.rs +++ b/src/modules/cache/imap/sync/flow.rs @@ -23,7 +23,7 @@ use crate::{ imap::{ find_intersecting_mailboxes, find_missing_mailboxes, mailbox::MailBox, - sync::rebuild::{rebuild_mailbox_cache, rebuild_mailbox_cache_since_date}, + sync::rebuild::{rebuild_mailbox_cache, rebuild_mailbox_cache_by_date}, }, SEMAPHORE, }, @@ -38,15 +38,28 @@ use tracing::{debug, error, info, warn}; pub const DEFAULT_BATCH_SIZE: u32 = 50; -pub async fn fetch_and_save_since_date( +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum FetchDirection { + Since, + Before, +} + +pub async fn fetch_and_save_by_date( account: &AccountModel, date: &str, mailbox: &MailBox, + direction: FetchDirection, ) -> BichonResult { let account_id = account.id; let executor = MAIL_CONTEXT.imap(account_id).await?; + + let search_criteria = match direction { + FetchDirection::Since => format!("SINCE {date}"), + FetchDirection::Before => format!("BEFORE {date}"), + }; + let uid_list = executor - .uid_search(&mailbox.encoded_name(), format!("SINCE {date}").as_str()) + .uid_search(&mailbox.encoded_name(), &search_criteria) .await?; let len = uid_list.len(); @@ -62,7 +75,13 @@ pub async fn fetch_and_save_since_date( if let Some(limit) = folder_limit { let limit = limit.max(100) as usize; if len > limit { - uid_vec = uid_vec.split_off(len - limit as usize); + uid_vec = match direction { + FetchDirection::Since => uid_vec.split_off(len - limit), + FetchDirection::Before => { + uid_vec.truncate(limit); + uid_vec + } + }; } } @@ -256,17 +275,30 @@ pub async fn reconcile_mailboxes( match &account.date_since { Some(date_since) => { - rebuild_mailbox_cache_since_date( + rebuild_mailbox_cache_by_date( account, local_mailbox.id, - date_since, + &date_since.since_date()?, remote_mailbox, + FetchDirection::Since, ) .await?; } - None => { - rebuild_mailbox_cache(account, local_mailbox, remote_mailbox).await?; - } + None => match &account.date_before { + Some(r) => { + rebuild_mailbox_cache_by_date( + account, + local_mailbox.id, + &r.calculate_date()?, + remote_mailbox, + FetchDirection::Before, + ) + .await?; + } + None => { + rebuild_mailbox_cache(account, local_mailbox, remote_mailbox).await? + } + }, } } else { perform_incremental_sync(account, local_mailbox, remote_mailbox).await?; @@ -310,14 +342,31 @@ pub async fn reconcile_mailboxes( let _permit = permit; match &account.date_since { Some(date_since) => { - rebuild_mailbox_cache_since_date( - &account, mailbox.id, date_since, &mailbox, + rebuild_mailbox_cache_by_date( + &account, + mailbox.id, + &date_since.since_date()?, + &mailbox, + FetchDirection::Since, ) .await } - None => { - rebuild_mailbox_cache(&account, &mailbox, &mailbox).await - } + None => match &account.date_before { + Some(r) => { + rebuild_mailbox_cache_by_date( + &account, + mailbox.id, + &r.calculate_date()?, + &mailbox, + FetchDirection::Before, + ) + .await + } + None => { + rebuild_mailbox_cache(&account, &mailbox, &mailbox) + .await + } + }, } }); handles.push(handle); @@ -353,8 +402,14 @@ async fn perform_incremental_sync( match local_max_uid { Some(max_uid) => { let executor = MAIL_CONTEXT.imap(account.id).await?; + let before_date = account + .date_before + .as_ref() + .map(|r| r.calculate_date()) + .transpose()?; + executor - .fetch_new_mail(account, local_mailbox, max_uid + 1) + .fetch_new_mail(account, local_mailbox, max_uid + 1, before_date.as_deref()) .await?; } None => { @@ -364,10 +419,11 @@ async fn perform_incremental_sync( match &account.date_since { Some(date_since) => { - fetch_and_save_since_date( + fetch_and_save_by_date( account, date_since.since_date()?.as_str(), remote_mailbox, + FetchDirection::Since, ) .await?; } diff --git a/src/modules/cache/imap/sync/mod.rs b/src/modules/cache/imap/sync/mod.rs index 76473e8..d1873a7 100644 --- a/src/modules/cache/imap/sync/mod.rs +++ b/src/modules/cache/imap/sync/mod.rs @@ -23,13 +23,13 @@ use crate::{ migration::{AccountModel, AccountType}, state::AccountRunningState, }, - cache::imap::mailbox::MailBox, + cache::imap::{mailbox::MailBox, sync::flow::FetchDirection}, error::BichonResult, }, utc_now, }; use flow::reconcile_mailboxes; -use rebuild::{rebuild_cache, rebuild_cache_since_date}; +use rebuild::{rebuild_cache, rebuild_cache_by_date}; use std::time::Instant; use sync_folders::get_sync_folders; use sync_type::{determine_sync_type, SyncType}; @@ -54,9 +54,26 @@ pub async fn execute_imap_sync(account: &AccountModel) -> BichonResult<()> { // AccountRunningState::set_initial_sync_start(account_id).await?; let result = match &account.date_since { Some(date_since) => { - rebuild_cache_since_date(account, &remote_mailboxes, date_since).await + rebuild_cache_by_date( + account, + &remote_mailboxes, + &date_since.since_date()?, + FetchDirection::Since, + ) + .await } - None => rebuild_cache(account, &remote_mailboxes).await, + None => match &account.date_before { + Some(r) => { + rebuild_cache_by_date( + account, + &remote_mailboxes, + &r.calculate_date()?, + FetchDirection::Before, + ) + .await + } + None => rebuild_cache(account, &remote_mailboxes).await, + }, }; match result { Ok(_) => { diff --git a/src/modules/cache/imap/sync/rebuild.rs b/src/modules/cache/imap/sync/rebuild.rs index 77ec860..00a3c60 100644 --- a/src/modules/cache/imap/sync/rebuild.rs +++ b/src/modules/cache/imap/sync/rebuild.rs @@ -16,14 +16,13 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . - use crate::{ modules::{ - account::{migration::AccountModel, since::DateSince}, + account::migration::AccountModel, cache::{ imap::{ mailbox::MailBox, - sync::flow::{fetch_and_save_full_mailbox, fetch_and_save_since_date}, + sync::flow::{fetch_and_save_by_date, fetch_and_save_full_mailbox, FetchDirection}, }, SEMAPHORE, }, @@ -86,14 +85,14 @@ pub async fn rebuild_cache( Ok(()) } -pub async fn rebuild_cache_since_date( +pub async fn rebuild_cache_by_date( account: &AccountModel, remote_mailboxes: &[MailBox], - date_since: &DateSince, + date: &str, + direction: FetchDirection, ) -> BichonResult<()> { let start_time = Instant::now(); let mut total_inserted = 0; - let date = date_since.since_date()?; MailBox::batch_insert(remote_mailboxes).await?; let mut handles = Vec::new(); @@ -107,13 +106,14 @@ pub async fn rebuild_cache_since_date( } let account = account.clone(); let mailbox = mailbox.clone(); - let date = date.clone(); + let date = date.to_string(); + let direction = direction.clone(); match SEMAPHORE.clone().acquire_owned().await { Ok(permit) => { let handle: tokio::task::JoinHandle> = tokio::spawn(async move { let _permit = permit; // Ensure permit is released when task finishes - fetch_and_save_since_date(&account, date.as_str(), &mailbox).await + fetch_and_save_by_date(&account, date.as_str(), &mailbox, direction).await }); handles.push(handle); } @@ -132,10 +132,14 @@ pub async fn rebuild_cache_since_date( } } let elapsed_time = start_time.elapsed().as_secs(); + let direction_desc = match direction { + FetchDirection::Since => "starting from the specified date", + FetchDirection::Before => "ending before the specified date", + }; info!( "Rebuild account cache completed: {} envelopes inserted. {} secs elapsed. \ - Data fetched from server starting from the specified date: {}.", - total_inserted, elapsed_time, date + Data fetched from server {}: {}.", + total_inserted, elapsed_time, direction_desc, date ); Ok(()) } @@ -169,11 +173,12 @@ pub async fn rebuild_mailbox_cache( Ok(()) } -pub async fn rebuild_mailbox_cache_since_date( +pub async fn rebuild_mailbox_cache_by_date( account: &AccountModel, local_mailbox_id: u64, - date_since: &DateSince, + date: &str, remote: &MailBox, + direction: FetchDirection, ) -> BichonResult<()> { ENVELOPE_INDEX_MANAGER .delete_mailbox_envelopes(account.id, vec![local_mailbox_id]) @@ -190,8 +195,7 @@ pub async fn rebuild_mailbox_cache_since_date( return Ok(()); // Skip if the mailbox has no emails } - let count = - fetch_and_save_since_date(account, date_since.since_date()?.as_str(), remote).await?; + let count = fetch_and_save_by_date(account, date, remote, direction).await?; info!( "Account {}: Successfully rebuild mailbox cache, inserted {} envelopes for mailbox '{}'.", account.id, count, &remote.name diff --git a/src/modules/imap/executor.rs b/src/modules/imap/executor.rs index 85d02ef..95200d5 100644 --- a/src/modules/imap/executor.rs +++ b/src/modules/imap/executor.rs @@ -84,12 +84,19 @@ impl ImapExecutor { account: &AccountModel, mailbox: &MailBox, start_uid: u64, + before: Option<&str> ) -> 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}:*"), + }; + let uid_list = self .uid_search( &mailbox.encoded_name(), - format!("UID {start_uid}:*").as_str(), + &query, ) .await?;