fetch: support fetching mails before a specified date

This commit is contained in:
rustmailer
2025-12-29 12:06:04 +08:00
parent 06a126461b
commit 76ab16b55b
4 changed files with 119 additions and 35 deletions
+72 -16
View File
@@ -23,7 +23,7 @@ use crate::{
imap::{ imap::{
find_intersecting_mailboxes, find_missing_mailboxes, find_intersecting_mailboxes, find_missing_mailboxes,
mailbox::MailBox, mailbox::MailBox,
sync::rebuild::{rebuild_mailbox_cache, rebuild_mailbox_cache_since_date}, sync::rebuild::{rebuild_mailbox_cache, rebuild_mailbox_cache_by_date},
}, },
SEMAPHORE, SEMAPHORE,
}, },
@@ -38,15 +38,28 @@ use tracing::{debug, error, info, warn};
pub const DEFAULT_BATCH_SIZE: u32 = 50; 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, account: &AccountModel,
date: &str, date: &str,
mailbox: &MailBox, mailbox: &MailBox,
direction: FetchDirection,
) -> BichonResult<usize> { ) -> BichonResult<usize> {
let account_id = account.id; let account_id = account.id;
let executor = MAIL_CONTEXT.imap(account_id).await?; 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 let uid_list = executor
.uid_search(&mailbox.encoded_name(), format!("SINCE {date}").as_str()) .uid_search(&mailbox.encoded_name(), &search_criteria)
.await?; .await?;
let len = uid_list.len(); let len = uid_list.len();
@@ -62,7 +75,13 @@ pub async fn fetch_and_save_since_date(
if let Some(limit) = folder_limit { if let Some(limit) = folder_limit {
let limit = limit.max(100) as usize; let limit = limit.max(100) as usize;
if len > limit { 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 { match &account.date_since {
Some(date_since) => { Some(date_since) => {
rebuild_mailbox_cache_since_date( rebuild_mailbox_cache_by_date(
account, account,
local_mailbox.id, local_mailbox.id,
date_since, &date_since.since_date()?,
remote_mailbox, remote_mailbox,
FetchDirection::Since,
) )
.await?; .await?;
} }
None => { None => match &account.date_before {
rebuild_mailbox_cache(account, local_mailbox, remote_mailbox).await?; 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 { } else {
perform_incremental_sync(account, local_mailbox, remote_mailbox).await?; perform_incremental_sync(account, local_mailbox, remote_mailbox).await?;
@@ -310,14 +342,31 @@ pub async fn reconcile_mailboxes(
let _permit = permit; let _permit = permit;
match &account.date_since { match &account.date_since {
Some(date_since) => { Some(date_since) => {
rebuild_mailbox_cache_since_date( rebuild_mailbox_cache_by_date(
&account, mailbox.id, date_since, &mailbox, &account,
mailbox.id,
&date_since.since_date()?,
&mailbox,
FetchDirection::Since,
) )
.await .await
} }
None => { None => match &account.date_before {
rebuild_mailbox_cache(&account, &mailbox, &mailbox).await 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); handles.push(handle);
@@ -353,8 +402,14 @@ async fn perform_incremental_sync(
match local_max_uid { match local_max_uid {
Some(max_uid) => { Some(max_uid) => {
let executor = MAIL_CONTEXT.imap(account.id).await?; let executor = MAIL_CONTEXT.imap(account.id).await?;
let before_date = account
.date_before
.as_ref()
.map(|r| r.calculate_date())
.transpose()?;
executor executor
.fetch_new_mail(account, local_mailbox, max_uid + 1) .fetch_new_mail(account, local_mailbox, max_uid + 1, before_date.as_deref())
.await?; .await?;
} }
None => { None => {
@@ -364,10 +419,11 @@ async fn perform_incremental_sync(
match &account.date_since { match &account.date_since {
Some(date_since) => { Some(date_since) => {
fetch_and_save_since_date( fetch_and_save_by_date(
account, account,
date_since.since_date()?.as_str(), date_since.since_date()?.as_str(),
remote_mailbox, remote_mailbox,
FetchDirection::Since,
) )
.await?; .await?;
} }
+21 -4
View File
@@ -23,13 +23,13 @@ use crate::{
migration::{AccountModel, AccountType}, migration::{AccountModel, AccountType},
state::AccountRunningState, state::AccountRunningState,
}, },
cache::imap::mailbox::MailBox, cache::imap::{mailbox::MailBox, sync::flow::FetchDirection},
error::BichonResult, error::BichonResult,
}, },
utc_now, utc_now,
}; };
use flow::reconcile_mailboxes; 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 std::time::Instant;
use sync_folders::get_sync_folders; use sync_folders::get_sync_folders;
use sync_type::{determine_sync_type, SyncType}; 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?; // AccountRunningState::set_initial_sync_start(account_id).await?;
let result = match &account.date_since { let result = match &account.date_since {
Some(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 { match result {
Ok(_) => { Ok(_) => {
+18 -14
View File
@@ -16,14 +16,13 @@
// You should have received a copy of the GNU Affero General Public License // You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>. // along with this program. If not, see <http://www.gnu.org/licenses/>.
use crate::{ use crate::{
modules::{ modules::{
account::{migration::AccountModel, since::DateSince}, account::migration::AccountModel,
cache::{ cache::{
imap::{ imap::{
mailbox::MailBox, 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, SEMAPHORE,
}, },
@@ -86,14 +85,14 @@ pub async fn rebuild_cache(
Ok(()) Ok(())
} }
pub async fn rebuild_cache_since_date( pub async fn rebuild_cache_by_date(
account: &AccountModel, account: &AccountModel,
remote_mailboxes: &[MailBox], remote_mailboxes: &[MailBox],
date_since: &DateSince, date: &str,
direction: FetchDirection,
) -> BichonResult<()> { ) -> BichonResult<()> {
let start_time = Instant::now(); let start_time = Instant::now();
let mut total_inserted = 0; let mut total_inserted = 0;
let date = date_since.since_date()?;
MailBox::batch_insert(remote_mailboxes).await?; MailBox::batch_insert(remote_mailboxes).await?;
let mut handles = Vec::new(); let mut handles = Vec::new();
@@ -107,13 +106,14 @@ pub async fn rebuild_cache_since_date(
} }
let account = account.clone(); let account = account.clone();
let mailbox = mailbox.clone(); let mailbox = mailbox.clone();
let date = date.clone(); let date = date.to_string();
let direction = direction.clone();
match SEMAPHORE.clone().acquire_owned().await { match SEMAPHORE.clone().acquire_owned().await {
Ok(permit) => { Ok(permit) => {
let handle: tokio::task::JoinHandle<Result<usize, BichonError>> = let handle: tokio::task::JoinHandle<Result<usize, BichonError>> =
tokio::spawn(async move { tokio::spawn(async move {
let _permit = permit; // Ensure permit is released when task finishes 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); handles.push(handle);
} }
@@ -132,10 +132,14 @@ pub async fn rebuild_cache_since_date(
} }
} }
let elapsed_time = start_time.elapsed().as_secs(); 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!( info!(
"Rebuild account cache completed: {} envelopes inserted. {} secs elapsed. \ "Rebuild account cache completed: {} envelopes inserted. {} secs elapsed. \
Data fetched from server starting from the specified date: {}.", Data fetched from server {}: {}.",
total_inserted, elapsed_time, date total_inserted, elapsed_time, direction_desc, date
); );
Ok(()) Ok(())
} }
@@ -169,11 +173,12 @@ pub async fn rebuild_mailbox_cache(
Ok(()) Ok(())
} }
pub async fn rebuild_mailbox_cache_since_date( pub async fn rebuild_mailbox_cache_by_date(
account: &AccountModel, account: &AccountModel,
local_mailbox_id: u64, local_mailbox_id: u64,
date_since: &DateSince, date: &str,
remote: &MailBox, remote: &MailBox,
direction: FetchDirection,
) -> BichonResult<()> { ) -> BichonResult<()> {
ENVELOPE_INDEX_MANAGER ENVELOPE_INDEX_MANAGER
.delete_mailbox_envelopes(account.id, vec![local_mailbox_id]) .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 return Ok(()); // Skip if the mailbox has no emails
} }
let count = let count = fetch_and_save_by_date(account, date, remote, direction).await?;
fetch_and_save_since_date(account, date_since.since_date()?.as_str(), remote).await?;
info!( info!(
"Account {}: Successfully rebuild mailbox cache, inserted {} envelopes for mailbox '{}'.", "Account {}: Successfully rebuild mailbox cache, inserted {} envelopes for mailbox '{}'.",
account.id, count, &remote.name account.id, count, &remote.name
+8 -1
View File
@@ -84,12 +84,19 @@ impl ImapExecutor {
account: &AccountModel, account: &AccountModel,
mailbox: &MailBox, mailbox: &MailBox,
start_uid: u64, start_uid: u64,
before: Option<&str>
) -> BichonResult<()> { ) -> BichonResult<()> {
assert!(start_uid > 0, "start_uid must be greater than 0"); 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 let uid_list = self
.uid_search( .uid_search(
&mailbox.encoded_name(), &mailbox.encoded_name(),
format!("UID {start_uid}:*").as_str(), &query,
) )
.await?; .await?;