mirror of
https://github.com/rustmailer/bichon.git
synced 2026-08-03 07:48:34 +02:00
refactor(workspace): decompose project into multiple crates
This commit is contained in:
rustmailer
@@ -0,0 +1,185 @@
|
||||
//
|
||||
// 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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
use std::collections::BTreeSet;
|
||||
|
||||
use crate::{
|
||||
decode_mailbox_name,
|
||||
{
|
||||
account::migration::{AccountModel, AccountType},
|
||||
cache::imap::mailbox::{AttributeEnum, MailBox},
|
||||
error::{code::ErrorCode, BichonResult},
|
||||
imap::{executor::ImapExecutor, session::SessionStream},
|
||||
mailbox::list::convert_names_to_mailboxes,
|
||||
},
|
||||
raise_error,
|
||||
};
|
||||
use async_imap::{types::Name, Session};
|
||||
use tracing::{debug, info, warn};
|
||||
|
||||
pub async fn get_download_folders(
|
||||
account: &AccountModel,
|
||||
session: &mut Session<Box<dyn SessionStream>>,
|
||||
) -> BichonResult<Vec<MailBox>> {
|
||||
assert_eq!(account.account_type, AccountType::IMAP);
|
||||
let names = ImapExecutor::list_all_mailboxes(session).await?;
|
||||
if names.is_empty() {
|
||||
warn!(
|
||||
"Account {}: No mailboxes returned from IMAP server.",
|
||||
account.id
|
||||
);
|
||||
return Err(raise_error!(format!(
|
||||
"No mailboxes returned from IMAP server for account {}. This is unexpected and may indicate an issue with the IMAP server.",
|
||||
&account.id
|
||||
), ErrorCode::ImapUnexpectedResult));
|
||||
}
|
||||
let mailboxes: Vec<(MailBox, Name)> = names.into_iter().map(|n| ((&n).into(), n)).collect();
|
||||
|
||||
for (mailbox, _) in &mailboxes {
|
||||
debug!(
|
||||
"[MAILBOX DEBUG] Account {}: mailbox='{}', attributes={:?}",
|
||||
account.id, mailbox.name, mailbox.attributes
|
||||
);
|
||||
}
|
||||
|
||||
detect_mailbox_changes(
|
||||
account,
|
||||
mailboxes.iter().map(|(m, _)| m.name.clone()).collect(),
|
||||
)
|
||||
.await?;
|
||||
let account = AccountModel::async_get(account.id).await?;
|
||||
let subscribed = &account.download_folders.unwrap_or_default();
|
||||
let is_noselect = |mailbox: &MailBox| {
|
||||
mailbox
|
||||
.attributes
|
||||
.iter()
|
||||
.any(|attr| matches!(attr.attr, AttributeEnum::NoSelect))
|
||||
};
|
||||
|
||||
let is_default_mailbox = |mailbox: &MailBox| {
|
||||
mailbox.name.eq_ignore_ascii_case("INBOX")
|
||||
|| mailbox
|
||||
.attributes
|
||||
.iter()
|
||||
.any(|attr| matches!(attr.attr, AttributeEnum::Sent))
|
||||
};
|
||||
|
||||
let mut matched_mailboxes: Vec<&Name> = if !subscribed.is_empty() {
|
||||
mailboxes
|
||||
.iter()
|
||||
.filter(|(mailbox, _)| subscribed.contains(&mailbox.name) && !is_noselect(mailbox))
|
||||
.map(|(_, name)| name)
|
||||
.collect()
|
||||
} else {
|
||||
Vec::new()
|
||||
};
|
||||
|
||||
if matched_mailboxes.is_empty() {
|
||||
matched_mailboxes = mailboxes
|
||||
.iter()
|
||||
.filter(|(mailbox, _)| !is_noselect(mailbox) && is_default_mailbox(mailbox))
|
||||
.map(|(_, name)| name)
|
||||
.collect();
|
||||
|
||||
debug!(
|
||||
"[MAILBOX DEBUG] Account {}: matched_mailboxes (default selection) = {:?}",
|
||||
account.id,
|
||||
matched_mailboxes
|
||||
.iter()
|
||||
.map(|n| decode_mailbox_name!(n.name().to_string()))
|
||||
.collect::<Vec<_>>()
|
||||
);
|
||||
|
||||
if !matched_mailboxes.is_empty() {
|
||||
let sync_folders: Vec<String> = matched_mailboxes
|
||||
.iter()
|
||||
.map(|n| decode_mailbox_name!(n.name().to_string()))
|
||||
.collect();
|
||||
AccountModel::update_download_folders(account.id, sync_folders).await?;
|
||||
} else {
|
||||
warn!(
|
||||
"Account {}: No subscribed mailboxes found. This is unexpected — IMAP server should at least provide INBOX.",
|
||||
account.id
|
||||
);
|
||||
return Err(raise_error!(format!(
|
||||
"No subscribed mailboxes found for account {}. This is unexpected — IMAP server should at least provide INBOX.",
|
||||
&account.id
|
||||
), ErrorCode::ImapUnexpectedResult));
|
||||
}
|
||||
}
|
||||
convert_names_to_mailboxes(account.id, session, matched_mailboxes).await
|
||||
}
|
||||
|
||||
pub async fn detect_mailbox_changes(
|
||||
account: &AccountModel,
|
||||
all_names: BTreeSet<String>,
|
||||
) -> BichonResult<()> {
|
||||
if account.known_folders.is_none() {
|
||||
// First time sync: just save without comparing
|
||||
AccountModel::update_known_folders(account.id, all_names).await?;
|
||||
return Ok(());
|
||||
}
|
||||
let known_folders = account.known_folders.clone().unwrap_or_default();
|
||||
// Compute differences
|
||||
let new_folders: Vec<String> = all_names.difference(&known_folders).cloned().collect();
|
||||
let deleted_folders: Vec<String> = known_folders.difference(&all_names).cloned().collect();
|
||||
|
||||
let has_changes = !new_folders.is_empty() || !deleted_folders.is_empty();
|
||||
let download_folders = account.download_folders.as_deref().unwrap_or_default();
|
||||
// Handle deleted folders in sync_folders
|
||||
if !deleted_folders.is_empty() {
|
||||
// Check if any deleted folders are in sync_folders
|
||||
let remaining_sync_folders: Vec<String> = download_folders
|
||||
.iter()
|
||||
.filter(|folder| !deleted_folders.contains(folder))
|
||||
.cloned()
|
||||
.collect();
|
||||
|
||||
// If sync_folders changed, update them
|
||||
if remaining_sync_folders.len() != download_folders.len() {
|
||||
let removed_count = download_folders.len() - remaining_sync_folders.len();
|
||||
info!(
|
||||
"Account {}: Removed {} deleted folders from sync_folders",
|
||||
account.id, removed_count
|
||||
);
|
||||
// Note: When all subscribed folders are deleted (remaining_sync_folders empty),
|
||||
// the system's default behavior is to automatically fall back to syncing
|
||||
// only the default folders (INBOX and Sent) in subsequent operations
|
||||
AccountModel::update_download_folders(account.id, remaining_sync_folders).await?;
|
||||
}
|
||||
|
||||
info!(
|
||||
"Account {}: Folders deleted: {:?}",
|
||||
account.id, deleted_folders
|
||||
);
|
||||
}
|
||||
|
||||
// Fire events for new folders if needed
|
||||
if !new_folders.is_empty() {
|
||||
info!(
|
||||
"Account {}: New folders detected: {:?}",
|
||||
account.id, new_folders
|
||||
);
|
||||
}
|
||||
|
||||
// Update known folders only if there were changes
|
||||
if has_changes {
|
||||
AccountModel::update_known_folders(account.id, all_names).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
//
|
||||
// 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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
use crate::{
|
||||
{
|
||||
account::{
|
||||
migration::AccountModel,
|
||||
state::{DownloadState, TriggerType},
|
||||
},
|
||||
error::BichonResult,
|
||||
},
|
||||
utc_now,
|
||||
};
|
||||
|
||||
#[derive(Clone, Debug, Eq, PartialEq)]
|
||||
pub enum DownloadTask {
|
||||
FullFetch,
|
||||
TraceFetch,
|
||||
Idle,
|
||||
}
|
||||
|
||||
pub async fn decide_next_download_task(account: &AccountModel) -> BichonResult<DownloadTask> {
|
||||
Ok(match DownloadState::get(account.id).await? {
|
||||
Some(state) => {
|
||||
let should_trigger = should_trigger_next_download(
|
||||
state.last_trigger_at,
|
||||
state.last_finished_at.unwrap_or(0),
|
||||
account.download_interval_min.unwrap(),
|
||||
);
|
||||
|
||||
if should_trigger {
|
||||
DownloadState::start_new_session(account.id, TriggerType::Scheduled).await?;
|
||||
DownloadTask::TraceFetch
|
||||
} else {
|
||||
DownloadTask::Idle
|
||||
}
|
||||
}
|
||||
None => {
|
||||
DownloadState::init(account.id).await?;
|
||||
DownloadTask::FullFetch
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
fn should_trigger_next_download(
|
||||
last_trigger_at: i64,
|
||||
last_finished_at: i64,
|
||||
sync_interval_min: i64,
|
||||
) -> bool {
|
||||
let now = utc_now!();
|
||||
now - last_trigger_at > (sync_interval_min * 60 * 1000) && now - last_finished_at > 60 * 1000
|
||||
}
|
||||
+701
@@ -0,0 +1,701 @@
|
||||
//
|
||||
// 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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
use crate::{
|
||||
{
|
||||
account::{
|
||||
migration::AccountModel,
|
||||
state::{DownloadState, DownloadStatus, FolderStatus},
|
||||
},
|
||||
cache::{
|
||||
imap::{
|
||||
download::rebuild::{rebuild_mailbox_cache, rebuild_mailbox_cache_by_date},
|
||||
find_intersecting_mailboxes, find_missing_mailboxes,
|
||||
mailbox::MailBox,
|
||||
},
|
||||
SEMAPHORE,
|
||||
},
|
||||
error::{code::ErrorCode, BichonResult},
|
||||
imap::executor::ImapExecutor,
|
||||
store::tantivy::envelope::ENVELOPE_MANAGER,
|
||||
},
|
||||
raise_error,
|
||||
};
|
||||
use std::time::Instant;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tracing::{debug, error, info, warn};
|
||||
|
||||
pub const DEFAULT_BATCH_SIZE: u32 = 30;
|
||||
|
||||
#[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,
|
||||
token: CancellationToken,
|
||||
) -> BichonResult<()> {
|
||||
let account_id = account.id;
|
||||
let mut session = match ImapExecutor::create_connection(account_id).await {
|
||||
Ok(session) => session,
|
||||
Err(e) => {
|
||||
let err_msg = format!("Connection failed for this folder: {:#?}", e);
|
||||
DownloadState::update_folder_progress(
|
||||
account_id,
|
||||
mailbox.name.clone(),
|
||||
0,
|
||||
0,
|
||||
FolderStatus::Failed,
|
||||
Some(err_msg.clone()),
|
||||
)
|
||||
.await?;
|
||||
DownloadState::append_session_error(account_id, err_msg).await?;
|
||||
return Err(e);
|
||||
}
|
||||
};
|
||||
|
||||
let search_criteria = match direction {
|
||||
FetchDirection::Since => format!("SINCE {date}"),
|
||||
FetchDirection::Before => format!("BEFORE {date}"),
|
||||
};
|
||||
|
||||
let uid_list =
|
||||
match ImapExecutor::uid_search(&mut session, &mailbox.encoded_name(), &search_criteria)
|
||||
.await
|
||||
{
|
||||
Ok(uid_list) => uid_list,
|
||||
Err(e) => {
|
||||
let err_msg = format!("UID search failed in [{}]: {:#?}", mailbox.name, e);
|
||||
DownloadState::update_folder_progress(
|
||||
account_id,
|
||||
mailbox.name.clone(),
|
||||
0,
|
||||
0,
|
||||
FolderStatus::Failed,
|
||||
Some(err_msg.clone()),
|
||||
)
|
||||
.await?;
|
||||
DownloadState::append_session_error(account_id, err_msg).await?;
|
||||
return Err(e);
|
||||
}
|
||||
};
|
||||
|
||||
let len = uid_list.len();
|
||||
if len == 0 {
|
||||
DownloadState::update_folder_progress(
|
||||
account_id,
|
||||
mailbox.name.clone(),
|
||||
0,
|
||||
0,
|
||||
FolderStatus::Success,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let folder_limit = account.folder_limit;
|
||||
// sort small -> bigger
|
||||
let mut uid_vec: Vec<u32> = 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 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,
|
||||
mailbox.name.clone(),
|
||||
planned,
|
||||
0,
|
||||
FolderStatus::Pending,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
|
||||
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() {
|
||||
DownloadState::update_session_status(
|
||||
account_id,
|
||||
DownloadStatus::Cancelled,
|
||||
Some("User stopped or system shutdown".to_string()),
|
||||
)
|
||||
.await?;
|
||||
DownloadState::update_folder_progress(
|
||||
account_id,
|
||||
mailbox.name.clone(),
|
||||
planned,
|
||||
current_processed,
|
||||
FolderStatus::Cancelled,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
has_error_or_cancel = true;
|
||||
break;
|
||||
}
|
||||
// Fetch metadata for the current batch of UIDs
|
||||
match ImapExecutor::uid_batch_retrieve_emails(
|
||||
&mut session,
|
||||
account_id,
|
||||
mailbox.id,
|
||||
&batch.0,
|
||||
token.clone(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(_) => {
|
||||
current_processed += batch.1;
|
||||
DownloadState::update_folder_progress(
|
||||
account_id,
|
||||
mailbox.name.clone(),
|
||||
planned,
|
||||
current_processed,
|
||||
FolderStatus::Downloading,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
Err(e) => {
|
||||
let err_msg = format!("Batch {} failed: {:#?}", index, e);
|
||||
DownloadState::append_session_error(account_id, err_msg.clone()).await?;
|
||||
DownloadState::update_folder_progress(
|
||||
account_id,
|
||||
mailbox.name.clone(),
|
||||
planned,
|
||||
current_processed,
|
||||
FolderStatus::Failed,
|
||||
Some(err_msg),
|
||||
)
|
||||
.await?;
|
||||
has_error_or_cancel = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
if !has_error_or_cancel {
|
||||
DownloadState::update_folder_progress(
|
||||
account_id,
|
||||
mailbox.name.clone(),
|
||||
planned,
|
||||
current_processed,
|
||||
FolderStatus::Success,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
session.logout().await.ok();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn fetch_and_save_full_mailbox(
|
||||
account: &AccountModel,
|
||||
mailbox: &MailBox,
|
||||
token: CancellationToken,
|
||||
) -> BichonResult<()> {
|
||||
let mailbox_id = mailbox.id;
|
||||
let account_id = account.id;
|
||||
|
||||
let mut session = match ImapExecutor::create_connection(account_id).await {
|
||||
Ok(session) => session,
|
||||
Err(e) => {
|
||||
let err_msg = format!("Connection failed for this folder: {:#?}", e);
|
||||
DownloadState::update_folder_progress(
|
||||
account_id,
|
||||
mailbox.name.clone(),
|
||||
0,
|
||||
0,
|
||||
FolderStatus::Failed,
|
||||
Some(err_msg.clone()),
|
||||
)
|
||||
.await?;
|
||||
DownloadState::append_session_error(account_id, err_msg).await?;
|
||||
return Err(e);
|
||||
}
|
||||
};
|
||||
|
||||
let total = match session.examine(&mailbox.encoded_name()).await {
|
||||
Ok(mailbox) => mailbox.exists as u64,
|
||||
Err(e) => {
|
||||
let err_msg = format!("Failed to examine folder [{}]: {:#?}", mailbox.name, e);
|
||||
DownloadState::update_folder_progress(
|
||||
account_id,
|
||||
mailbox.name.clone(),
|
||||
mailbox.exists as u64,
|
||||
0,
|
||||
FolderStatus::Failed,
|
||||
Some(err_msg.clone()),
|
||||
)
|
||||
.await?;
|
||||
|
||||
DownloadState::append_session_error(account_id, err_msg).await?;
|
||||
session.logout().await.ok();
|
||||
return Err(raise_error!(
|
||||
format!("{:#?}", e),
|
||||
ErrorCode::ImapCommandFailed
|
||||
));
|
||||
}
|
||||
};
|
||||
|
||||
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();
|
||||
|
||||
info!(
|
||||
"Starting full mailbox download for '{}', total={}, limit={:?}, batches={}, desc={}",
|
||||
mailbox.name, total, folder_limit, total_batches, desc
|
||||
);
|
||||
|
||||
let mut current_processed = 0u64;
|
||||
let mut has_error_or_cancel = false;
|
||||
|
||||
for page in 1..=total_batches {
|
||||
if token.is_cancelled() {
|
||||
DownloadState::update_session_status(
|
||||
account_id,
|
||||
DownloadStatus::Cancelled,
|
||||
Some("User stopped or system shutdown".to_string()),
|
||||
)
|
||||
.await?;
|
||||
DownloadState::update_folder_progress(
|
||||
account_id,
|
||||
mailbox.name.clone(),
|
||||
total_to_fetch,
|
||||
current_processed,
|
||||
FolderStatus::Cancelled,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
has_error_or_cancel = true;
|
||||
break;
|
||||
}
|
||||
|
||||
match ImapExecutor::batch_retrieve_emails(
|
||||
&mut session,
|
||||
account_id,
|
||||
mailbox_id,
|
||||
total_to_fetch,
|
||||
page as u64,
|
||||
page_size as u64,
|
||||
&mailbox.encoded_name(),
|
||||
desc,
|
||||
token.clone(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(count) => {
|
||||
current_processed += count as u64;
|
||||
DownloadState::update_folder_progress(
|
||||
account_id,
|
||||
mailbox.name.clone(),
|
||||
total_to_fetch,
|
||||
current_processed,
|
||||
FolderStatus::Downloading,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
Err(e) => {
|
||||
let err_msg = format!("Batch {} failed: {:#?}", page, e);
|
||||
DownloadState::append_session_error(account_id, err_msg.clone()).await?;
|
||||
DownloadState::update_folder_progress(
|
||||
account_id,
|
||||
mailbox.name.clone(),
|
||||
total_to_fetch,
|
||||
current_processed,
|
||||
FolderStatus::Failed,
|
||||
Some(err_msg),
|
||||
)
|
||||
.await?;
|
||||
has_error_or_cancel = true;
|
||||
break;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
if !has_error_or_cancel {
|
||||
DownloadState::update_folder_progress(
|
||||
account_id,
|
||||
mailbox.name.clone(),
|
||||
total_to_fetch,
|
||||
current_processed,
|
||||
FolderStatus::Success,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
session.logout().await.ok();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn generate_uid_sequence_hashset(
|
||||
unique_nums: Vec<u32>,
|
||||
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();
|
||||
|
||||
for chunk in nums.chunks(chunk_size) {
|
||||
let size = chunk.len() as u64;
|
||||
let compressed = compress_uid_list(chunk.to_vec());
|
||||
result.push((compressed, size));
|
||||
}
|
||||
|
||||
result
|
||||
}
|
||||
|
||||
pub fn compress_uid_list(nums: Vec<u32>) -> String {
|
||||
if nums.is_empty() {
|
||||
return String::new();
|
||||
}
|
||||
|
||||
let mut sorted_nums = nums;
|
||||
sorted_nums.sort();
|
||||
|
||||
let mut result = Vec::new();
|
||||
let mut current_range_start = sorted_nums[0];
|
||||
let mut current_range_end = sorted_nums[0];
|
||||
|
||||
for &n in sorted_nums.iter().skip(1) {
|
||||
if n == current_range_end + 1 {
|
||||
current_range_end = n;
|
||||
} else {
|
||||
if current_range_start == current_range_end {
|
||||
result.push(current_range_start.to_string());
|
||||
} else {
|
||||
result.push(format!("{}:{}", current_range_start, current_range_end));
|
||||
}
|
||||
current_range_start = n;
|
||||
current_range_end = n;
|
||||
}
|
||||
}
|
||||
|
||||
if current_range_start == current_range_end {
|
||||
result.push(current_range_start.to_string());
|
||||
} else {
|
||||
result.push(format!("{}:{}", current_range_start, current_range_end));
|
||||
}
|
||||
|
||||
result.join(",")
|
||||
}
|
||||
|
||||
pub async fn reconcile_mailboxes(
|
||||
account: &AccountModel,
|
||||
remote_mailboxes: &[MailBox],
|
||||
local_mailboxes: &[MailBox],
|
||||
token: CancellationToken,
|
||||
) -> BichonResult<()> {
|
||||
let start_time = Instant::now();
|
||||
let existing_mailboxes = find_intersecting_mailboxes(local_mailboxes, remote_mailboxes);
|
||||
let account_id = account.id;
|
||||
if !existing_mailboxes.is_empty() {
|
||||
let mut mailboxes_to_update = Vec::with_capacity(existing_mailboxes.len());
|
||||
|
||||
DownloadState::init_folder_details(
|
||||
account.id,
|
||||
remote_mailboxes.iter().map(|m| m.name.clone()).collect(),
|
||||
)
|
||||
.await?;
|
||||
|
||||
for (local_mailbox, remote_mailbox) in &existing_mailboxes {
|
||||
if token.is_cancelled() {
|
||||
DownloadState::update_session_status(
|
||||
account.id,
|
||||
DownloadStatus::Cancelled,
|
||||
Some("Received termination signal (User stop or System shutdown)".to_string()),
|
||||
)
|
||||
.await?;
|
||||
break;
|
||||
}
|
||||
|
||||
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.",
|
||||
local_mailbox.name
|
||||
);
|
||||
|
||||
warn!("Account {}: {}", account_id, err_msg);
|
||||
|
||||
DownloadState::update_folder_progress(
|
||||
account_id,
|
||||
remote_mailbox.name.clone(),
|
||||
0,
|
||||
0,
|
||||
FolderStatus::Failed,
|
||||
Some(err_msg.clone()),
|
||||
)
|
||||
.await?;
|
||||
DownloadState::append_session_error(account_id, err_msg).await?;
|
||||
continue;
|
||||
}
|
||||
info!(
|
||||
"Account {}: Mailbox '{}' detected with changed uid_validity (local: {:#?}, remote: {:#?}). \
|
||||
The mailbox data may be invalid, resetting its envelopes and rebuilding the cache.",
|
||||
account_id, local_mailbox.name, &local_mailbox.uid_validity, &remote_mailbox.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()),
|
||||
)
|
||||
.await?;
|
||||
|
||||
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?
|
||||
}
|
||||
},
|
||||
}
|
||||
} else {
|
||||
perform_incremental_sync(account, local_mailbox, remote_mailbox, token.clone())
|
||||
.await?;
|
||||
}
|
||||
|
||||
mailboxes_to_update.push(remote_mailbox.clone());
|
||||
}
|
||||
//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.
|
||||
MailBox::batch_upsert(&mailboxes_to_update).await?;
|
||||
}
|
||||
|
||||
debug!(
|
||||
"Checked mailbox folders for account ID: {}. Compared local and server folders to identify changes. Elapsed time: {} seconds",
|
||||
account.id,
|
||||
start_time.elapsed().as_secs()
|
||||
);
|
||||
|
||||
let missing_mailboxes = find_missing_mailboxes(local_mailboxes, remote_mailboxes);
|
||||
//Mail folders that are not locally need to be downloaded.
|
||||
if !missing_mailboxes.is_empty() {
|
||||
MailBox::batch_insert(&missing_mailboxes).await?;
|
||||
|
||||
let mut has_error = false;
|
||||
let mut last_err = None;
|
||||
for mailbox in &missing_mailboxes {
|
||||
if token.is_cancelled() {
|
||||
DownloadState::update_session_status(
|
||||
account.id,
|
||||
DownloadStatus::Cancelled,
|
||||
Some("Received termination signal (User stop or System shutdown)".to_string()),
|
||||
)
|
||||
.await?;
|
||||
break;
|
||||
}
|
||||
if mailbox.exists > 0 {
|
||||
let account = account.clone();
|
||||
let mailbox = mailbox.clone();
|
||||
|
||||
let _global_permit = match SEMAPHORE.clone().acquire_owned().await {
|
||||
Ok(permit) => permit,
|
||||
Err(err) => {
|
||||
error!(
|
||||
"Failed to acquire global semaphore permit for account {} mailbox '{}': {:#?}",
|
||||
account.id, &mailbox.name, err
|
||||
);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
let result = match &account.date_since {
|
||||
Some(date_since) => {
|
||||
rebuild_mailbox_cache_by_date(
|
||||
&account,
|
||||
mailbox.id,
|
||||
&date_since.since_date()?,
|
||||
&mailbox,
|
||||
FetchDirection::Since,
|
||||
token.clone(),
|
||||
)
|
||||
.await
|
||||
}
|
||||
None => match &account.date_before {
|
||||
Some(r) => {
|
||||
rebuild_mailbox_cache_by_date(
|
||||
&account,
|
||||
mailbox.id,
|
||||
&r.calculate_date()?,
|
||||
&mailbox,
|
||||
FetchDirection::Before,
|
||||
token.clone(),
|
||||
)
|
||||
.await
|
||||
}
|
||||
None => {
|
||||
rebuild_mailbox_cache(&account, &mailbox, &mailbox, token.clone()).await
|
||||
}
|
||||
},
|
||||
};
|
||||
|
||||
match result {
|
||||
Ok(_) => {}
|
||||
Err(err) => {
|
||||
has_error = true;
|
||||
tracing::error!("Folder sync task failed: {:#?}", err);
|
||||
last_err = Some(err);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if has_error {
|
||||
if let Some(e) = last_err {
|
||||
return Err(e);
|
||||
}
|
||||
return Err(raise_error!(
|
||||
"Some tasks failed".into(),
|
||||
ErrorCode::InternalError
|
||||
));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
//only check new emails and sync
|
||||
async fn perform_incremental_sync(
|
||||
account: &AccountModel,
|
||||
local_mailbox: &MailBox,
|
||||
remote_mailbox: &MailBox,
|
||||
token: CancellationToken,
|
||||
) -> BichonResult<()> {
|
||||
if remote_mailbox.exists > 0 {
|
||||
let local_max_uid = ENVELOPE_MANAGER
|
||||
.get_max_uid(account.id, local_mailbox.id)
|
||||
.await?;
|
||||
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();
|
||||
}
|
||||
None => {
|
||||
info!(
|
||||
"No maximum UID found in index for mailbox, assuming local cache is missing."
|
||||
);
|
||||
|
||||
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?;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
+138
@@ -0,0 +1,138 @@
|
||||
//
|
||||
// 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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
use crate::{
|
||||
account::{
|
||||
migration::{AccountModel, AccountType},
|
||||
state::{DownloadState, DownloadStatus},
|
||||
},
|
||||
cache::imap::{mailbox::MailBox, download::flow::FetchDirection},
|
||||
error::BichonResult,
|
||||
imap::executor::ImapExecutor,
|
||||
};
|
||||
use flow::reconcile_mailboxes;
|
||||
use rebuild::{rebuild_cache, rebuild_cache_by_date};
|
||||
use std::time::Instant;
|
||||
use download_folders::get_download_folders;
|
||||
use download_type::{decide_next_download_task, DownloadTask};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tracing::{debug, warn};
|
||||
|
||||
pub mod flow;
|
||||
pub mod rebuild;
|
||||
pub mod download_folders;
|
||||
pub mod download_type;
|
||||
|
||||
pub async fn process_imap_download(
|
||||
account: &AccountModel,
|
||||
token: CancellationToken,
|
||||
) -> BichonResult<()> {
|
||||
assert_eq!(account.account_type, AccountType::IMAP);
|
||||
let start_time = Instant::now();
|
||||
let account_id = account.id;
|
||||
let download_task = decide_next_download_task(account).await?;
|
||||
if matches!(download_task, DownloadTask::Idle) {
|
||||
return Ok(());
|
||||
}
|
||||
let mut session = match ImapExecutor::create_connection(account_id).await {
|
||||
Ok(session) => session,
|
||||
Err(e) => {
|
||||
DownloadState::update_session_status(
|
||||
account_id,
|
||||
DownloadStatus::Failed,
|
||||
Some(format!("Failed to connect to IMAP server: {}", e)),
|
||||
)
|
||||
.await?;
|
||||
return Err(e);
|
||||
}
|
||||
};
|
||||
let remote_mailboxes = match get_download_folders(account, &mut session).await {
|
||||
Ok(mailboxes) => mailboxes,
|
||||
Err(err) => {
|
||||
let err_msg = format!("Failed to fetch mailboxes: {}", err);
|
||||
warn!(account_id = account.id, error = %err, "{}", err_msg);
|
||||
DownloadState::update_session_status(account_id, DownloadStatus::Failed, Some(err_msg))
|
||||
.await?;
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
session.logout().await.ok();
|
||||
if matches!(download_task, DownloadTask::FullFetch) {
|
||||
let result = match &account.date_since {
|
||||
Some(date_since) => {
|
||||
rebuild_cache_by_date(
|
||||
account,
|
||||
&remote_mailboxes,
|
||||
&date_since.since_date()?,
|
||||
FetchDirection::Since,
|
||||
token,
|
||||
)
|
||||
.await
|
||||
}
|
||||
None => match &account.date_before {
|
||||
Some(r) => {
|
||||
rebuild_cache_by_date(
|
||||
account,
|
||||
&remote_mailboxes,
|
||||
&r.calculate_date()?,
|
||||
FetchDirection::Before,
|
||||
token,
|
||||
)
|
||||
.await
|
||||
}
|
||||
None => rebuild_cache(account, &remote_mailboxes, token).await,
|
||||
},
|
||||
};
|
||||
match result {
|
||||
Ok(_) => {
|
||||
DownloadState::update_session_status(account_id, DownloadStatus::Success, None)
|
||||
.await?;
|
||||
}
|
||||
Err(e) => {
|
||||
DownloadState::update_session_status(
|
||||
account_id,
|
||||
DownloadStatus::Failed,
|
||||
Some(format!("Email Download interrupted: {:#?}", e)),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let local_mailboxes = MailBox::list_all(account_id).await?;
|
||||
match reconcile_mailboxes(account, &remote_mailboxes, &local_mailboxes, token).await {
|
||||
Ok(_) => {
|
||||
DownloadState::update_session_status(account_id, DownloadStatus::Success, None).await?
|
||||
}
|
||||
Err(e) => {
|
||||
DownloadState::update_session_status(
|
||||
account_id,
|
||||
DownloadStatus::Failed,
|
||||
Some(format!("Email Download interrupted: {:#?}", e)),
|
||||
)
|
||||
.await?
|
||||
}
|
||||
}
|
||||
let elapsed_time = start_time.elapsed().as_secs();
|
||||
debug!(
|
||||
"Account{{{}}} Incremental sync completed: {} seconds elapsed.",
|
||||
account.email, elapsed_time
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
+267
@@ -0,0 +1,267 @@
|
||||
//
|
||||
// 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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
use crate::{
|
||||
{
|
||||
account::{
|
||||
migration::AccountModel,
|
||||
state::{DownloadState, DownloadStatus, FolderStatus},
|
||||
},
|
||||
cache::{
|
||||
imap::{
|
||||
download::flow::{
|
||||
fetch_and_save_by_date, fetch_and_save_full_mailbox, FetchDirection,
|
||||
},
|
||||
mailbox::MailBox,
|
||||
},
|
||||
SEMAPHORE,
|
||||
},
|
||||
error::{code::ErrorCode, BichonResult},
|
||||
store::tantivy::envelope::ENVELOPE_MANAGER,
|
||||
},
|
||||
raise_error,
|
||||
};
|
||||
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tracing::{error, info};
|
||||
|
||||
pub async fn rebuild_cache(
|
||||
account: &AccountModel,
|
||||
remote_mailboxes: &[MailBox],
|
||||
token: CancellationToken,
|
||||
) -> BichonResult<()> {
|
||||
MailBox::batch_insert(remote_mailboxes).await?;
|
||||
DownloadState::init_folder_details(
|
||||
account.id,
|
||||
remote_mailboxes.iter().map(|m| m.name.clone()).collect(),
|
||||
)
|
||||
.await?;
|
||||
|
||||
let mut has_error = false;
|
||||
let mut last_err = None;
|
||||
|
||||
for mailbox in remote_mailboxes {
|
||||
if token.is_cancelled() {
|
||||
DownloadState::update_session_status(
|
||||
account.id,
|
||||
DownloadStatus::Cancelled,
|
||||
Some("Received termination signal (User stop or System shutdown)".to_string()),
|
||||
)
|
||||
.await?;
|
||||
break;
|
||||
}
|
||||
if mailbox.exists == 0 {
|
||||
info!(
|
||||
"Account {}: Mailbox '{}' on the remote server has no emails. Skipping fetch for this mailbox.",
|
||||
account.id, &mailbox.name
|
||||
);
|
||||
DownloadState::update_folder_progress(
|
||||
account.id,
|
||||
mailbox.name.clone(),
|
||||
0,
|
||||
0,
|
||||
FolderStatus::Success,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
continue;
|
||||
}
|
||||
let account = account.clone();
|
||||
let mailbox = mailbox.clone();
|
||||
|
||||
let _global_permit = match SEMAPHORE.clone().acquire_owned().await {
|
||||
Ok(permit) => permit,
|
||||
Err(err) => {
|
||||
error!(
|
||||
"Failed to acquire global semaphore permit for account {} mailbox '{}': {:#?}",
|
||||
account.id, &mailbox.name, err
|
||||
);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
match fetch_and_save_full_mailbox(&account, &mailbox, token.clone()).await {
|
||||
Ok(_) => {},
|
||||
Err(err) => {
|
||||
has_error = true;
|
||||
tracing::error!("Folder sync task failed: {:#?}", err);
|
||||
last_err = Some(err);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if has_error {
|
||||
if let Some(e) = last_err {
|
||||
return Err(e);
|
||||
}
|
||||
return Err(raise_error!(
|
||||
"Some tasks failed".into(),
|
||||
ErrorCode::InternalError
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn rebuild_cache_by_date(
|
||||
account: &AccountModel,
|
||||
remote_mailboxes: &[MailBox],
|
||||
date: &str,
|
||||
direction: FetchDirection,
|
||||
token: CancellationToken,
|
||||
) -> BichonResult<()> {
|
||||
MailBox::batch_insert(remote_mailboxes).await?;
|
||||
DownloadState::init_folder_details(
|
||||
account.id,
|
||||
remote_mailboxes.iter().map(|m| m.name.clone()).collect(),
|
||||
)
|
||||
.await?;
|
||||
|
||||
let mut has_error = false;
|
||||
let mut last_err = None;
|
||||
|
||||
for mailbox in remote_mailboxes {
|
||||
if token.is_cancelled() {
|
||||
DownloadState::update_session_status(
|
||||
account.id,
|
||||
DownloadStatus::Cancelled,
|
||||
Some("Received termination signal (User stop or System shutdown)".to_string()),
|
||||
)
|
||||
.await?;
|
||||
break;
|
||||
}
|
||||
if mailbox.exists == 0 {
|
||||
info!(
|
||||
"Account {}: Mailbox '{}' on the remote server has no emails. Skipping fetch for this mailbox.",
|
||||
account.id, &mailbox.name
|
||||
);
|
||||
|
||||
DownloadState::update_folder_progress(
|
||||
account.id,
|
||||
mailbox.name.clone(),
|
||||
0,
|
||||
0,
|
||||
FolderStatus::Success,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
continue;
|
||||
}
|
||||
let account = account.clone();
|
||||
let mailbox = mailbox.clone();
|
||||
let date = date.to_string();
|
||||
let direction = direction.clone();
|
||||
|
||||
let _global_permit = match SEMAPHORE.clone().acquire_owned().await {
|
||||
Ok(permit) => permit,
|
||||
Err(err) => {
|
||||
error!(
|
||||
"Failed to acquire global semaphore permit for account {} mailbox '{}': {:#?}",
|
||||
account.id, &mailbox.name, err
|
||||
);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
match fetch_and_save_by_date(&account, date.as_str(), &mailbox, direction, token.clone())
|
||||
.await
|
||||
{
|
||||
Ok(_) => {}
|
||||
Err(err) => {
|
||||
has_error = true;
|
||||
tracing::error!("Folder sync task failed: {:#?}", err);
|
||||
last_err = Some(err);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if has_error {
|
||||
if let Some(e) = last_err {
|
||||
return Err(e);
|
||||
}
|
||||
return Err(raise_error!(
|
||||
"Some tasks failed".into(),
|
||||
ErrorCode::InternalError
|
||||
));
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn rebuild_mailbox_cache(
|
||||
account: &AccountModel,
|
||||
local_mailbox: &MailBox,
|
||||
remote_mailbox: &MailBox,
|
||||
token: CancellationToken,
|
||||
) -> BichonResult<()> {
|
||||
ENVELOPE_MANAGER
|
||||
.delete_mailbox_envelopes(account.id, vec![local_mailbox.id])
|
||||
.await?;
|
||||
|
||||
if remote_mailbox.exists == 0 {
|
||||
info!(
|
||||
"Account {}: Mailbox '{}' has no emails on the remote server. The mailbox is empty, no envelopes to fetch.",
|
||||
account.id,
|
||||
&local_mailbox.name
|
||||
);
|
||||
DownloadState::update_folder_progress(
|
||||
account.id,
|
||||
remote_mailbox.name.clone(),
|
||||
0,
|
||||
0,
|
||||
FolderStatus::Success,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
fetch_and_save_full_mailbox(account, remote_mailbox, token).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn rebuild_mailbox_cache_by_date(
|
||||
account: &AccountModel,
|
||||
local_mailbox_id: u64,
|
||||
date: &str,
|
||||
remote: &MailBox,
|
||||
direction: FetchDirection,
|
||||
token: CancellationToken,
|
||||
) -> BichonResult<()> {
|
||||
ENVELOPE_MANAGER
|
||||
.delete_mailbox_envelopes(account.id, vec![local_mailbox_id])
|
||||
.await?;
|
||||
if remote.exists == 0 {
|
||||
info!(
|
||||
"Account {}: Mailbox '{}' has no emails on the remote server. The mailbox is empty, no envelopes to fetch.",
|
||||
account.id,
|
||||
&remote.name
|
||||
);
|
||||
DownloadState::update_folder_progress(
|
||||
account.id,
|
||||
remote.name.clone(),
|
||||
0,
|
||||
0,
|
||||
FolderStatus::Success,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
fetch_and_save_by_date(account, date, remote, direction, token).await?;
|
||||
Ok(())
|
||||
}
|
||||
+214
@@ -0,0 +1,214 @@
|
||||
//
|
||||
// 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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
use crate::{
|
||||
decode_mailbox_name, encode_mailbox_name,
|
||||
{
|
||||
database::{
|
||||
async_filter_by_secondary_key_impl, async_find_impl, batch_delete_impl,
|
||||
batch_insert_impl, batch_upsert_impl, delete_impl, filter_by_secondary_key_impl,
|
||||
find_impl, manager::DB_MANAGER,
|
||||
},
|
||||
error::{code::ErrorCode, BichonResult},
|
||||
},
|
||||
raise_error,
|
||||
};
|
||||
use async_imap::types::{Name, NameAttribute};
|
||||
use itertools::Itertools;
|
||||
use native_db::*;
|
||||
use native_model::{native_model, Model};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
#[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize)]
|
||||
#[cfg_attr(feature = "web-api", derive(poem_openapi::Object))]
|
||||
#[native_model(id = 1, version = 1)]
|
||||
#[native_db]
|
||||
pub struct MailBox {
|
||||
/// The unique identifier for the mailbox
|
||||
#[primary_key]
|
||||
pub id: u64,
|
||||
/// The ID of the account associated with the mailbox
|
||||
#[secondary_key]
|
||||
pub account_id: u64,
|
||||
/// The unique, decoded, human-readable name of the mailbox (e.g., "INBOX", "Sent Items").
|
||||
/// This is the decoded name as presented to users, derived from the IMAP server's mailbox name
|
||||
/// (e.g., after decoding UTF-7 or other encodings per RFC 3501).
|
||||
pub name: String,
|
||||
/// Optional delimiter used to separate mailbox names in a hierarchy (e.g., "/" or ".").
|
||||
/// Used in IMAP to structure nested mailboxes (e.g., "INBOX/Archive").
|
||||
pub delimiter: Option<String>,
|
||||
/// List of attributes associated with the mailbox (e.g., `\NoSelect`, `\Deleted`).
|
||||
/// These indicate special properties, such as whether the mailbox can hold messages.
|
||||
pub attributes: Vec<Attribute>,
|
||||
/// The number of messages that currently exist in the mailbox.
|
||||
pub exists: u32,
|
||||
/// Optional number of unseen messages in the mailbox (i.e., messages without the `\Seen` flag).
|
||||
pub unseen: Option<u32>,
|
||||
/// The next unique identifier (UID) that will be assigned to a new message in the mailbox.
|
||||
/// If `None`, the IMAP server has not provided this information.
|
||||
pub uid_next: Option<u32>,
|
||||
/// 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<u32>,
|
||||
}
|
||||
|
||||
impl MailBox {
|
||||
pub fn encoded_name(&self) -> String {
|
||||
encode_mailbox_name!(&self.name)
|
||||
}
|
||||
|
||||
pub async fn async_get(id: u64) -> BichonResult<MailBox> {
|
||||
let result = async_find_impl::<MailBox>(DB_MANAGER.envelope_db(), id).await?;
|
||||
Ok(result.ok_or_else(|| {
|
||||
raise_error!(
|
||||
format!("mailbox {} not found", id),
|
||||
ErrorCode::InternalError
|
||||
)
|
||||
})?)
|
||||
}
|
||||
|
||||
pub fn get(id: u64) -> BichonResult<MailBox> {
|
||||
let result = find_impl::<MailBox>(DB_MANAGER.envelope_db(), id)?;
|
||||
Ok(result.ok_or_else(|| {
|
||||
raise_error!(
|
||||
format!("mailbox {} not found", id),
|
||||
ErrorCode::InternalError
|
||||
)
|
||||
})?)
|
||||
}
|
||||
|
||||
pub async fn delete(id: u64) -> BichonResult<()> {
|
||||
delete_impl(DB_MANAGER.envelope_db(), move |rw| {
|
||||
rw.get()
|
||||
.primary::<MailBox>(id)
|
||||
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
|
||||
.ok_or_else(|| raise_error!("mailbox missing".into(), ErrorCode::InternalError))
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn list_all(account_id: u64) -> BichonResult<Vec<MailBox>> {
|
||||
async_filter_by_secondary_key_impl(
|
||||
DB_MANAGER.envelope_db(),
|
||||
MailBoxKey::account_id,
|
||||
account_id,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
pub fn find_mailbox(account_id: u64, mailbox_id: u64) -> BichonResult<Option<MailBox>> {
|
||||
let all: Vec<MailBox> = filter_by_secondary_key_impl(
|
||||
DB_MANAGER.envelope_db(),
|
||||
MailBoxKey::account_id,
|
||||
account_id,
|
||||
)?;
|
||||
Ok(all.into_iter().find(|m| m.id == mailbox_id))
|
||||
}
|
||||
|
||||
pub async fn batch_insert(mailboxes: &[MailBox]) -> BichonResult<()> {
|
||||
batch_insert_impl(DB_MANAGER.envelope_db(), mailboxes.to_vec()).await
|
||||
}
|
||||
|
||||
pub async fn batch_upsert(mailboxes: &[MailBox]) -> BichonResult<()> {
|
||||
batch_upsert_impl(DB_MANAGER.envelope_db(), mailboxes.to_vec()).await
|
||||
}
|
||||
|
||||
pub async fn clean(account_id: u64) -> BichonResult<()> {
|
||||
batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
|
||||
let mailboxes: Vec<MailBox> = rw
|
||||
.scan()
|
||||
.secondary::<MailBox>(MailBoxKey::account_id)
|
||||
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
|
||||
.start_with(account_id)
|
||||
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
|
||||
.try_collect()
|
||||
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
|
||||
Ok(mailboxes)
|
||||
})
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
|
||||
#[cfg_attr(feature = "web-api", derive(poem_openapi::Object))]
|
||||
pub struct Attribute {
|
||||
pub attr: AttributeEnum,
|
||||
pub extension: Option<String>,
|
||||
}
|
||||
|
||||
impl Attribute {
|
||||
pub fn new(attr: AttributeEnum, extension: Option<String>) -> Self {
|
||||
Self { attr, extension }
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
|
||||
#[cfg_attr(feature = "web-api", derive(poem_openapi::Enum))]
|
||||
pub enum AttributeEnum {
|
||||
NoInferiors,
|
||||
NoSelect,
|
||||
Marked,
|
||||
Unmarked,
|
||||
All,
|
||||
Archive,
|
||||
Drafts,
|
||||
Flagged,
|
||||
Junk,
|
||||
Sent,
|
||||
Trash,
|
||||
Extension,
|
||||
Unknown,
|
||||
}
|
||||
|
||||
impl From<&Name> for MailBox {
|
||||
fn from(value: &Name) -> Self {
|
||||
let name = decode_mailbox_name!(value.name().to_string());
|
||||
let delimiter = value.delimiter().map(|f| f.to_owned());
|
||||
let attributes: Vec<Attribute> = value.attributes().iter().map(|na| na.into()).collect();
|
||||
//The remaining parts will be supplemented during the examine_mailbox process.
|
||||
MailBox {
|
||||
name,
|
||||
delimiter,
|
||||
attributes,
|
||||
..Default::default() //has_synced is initialized to false here
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<&NameAttribute<'_>> for Attribute {
|
||||
fn from(value: &NameAttribute) -> Self {
|
||||
match value {
|
||||
NameAttribute::NoInferiors => Attribute::new(AttributeEnum::NoInferiors, None),
|
||||
NameAttribute::NoSelect => Attribute::new(AttributeEnum::NoSelect, None),
|
||||
NameAttribute::Marked => Attribute::new(AttributeEnum::Marked, None),
|
||||
NameAttribute::Unmarked => Attribute::new(AttributeEnum::Unmarked, None),
|
||||
NameAttribute::All => Attribute::new(AttributeEnum::All, None),
|
||||
NameAttribute::Archive => Attribute::new(AttributeEnum::Archive, None),
|
||||
NameAttribute::Drafts => Attribute::new(AttributeEnum::Drafts, None),
|
||||
NameAttribute::Flagged => Attribute::new(AttributeEnum::Flagged, None),
|
||||
NameAttribute::Junk => Attribute::new(AttributeEnum::Junk, None),
|
||||
NameAttribute::Sent => Attribute::new(AttributeEnum::Sent, None),
|
||||
NameAttribute::Trash => Attribute::new(AttributeEnum::Trash, None),
|
||||
NameAttribute::Extension(s) => {
|
||||
Attribute::new(AttributeEnum::Extension, Some(s.to_string()))
|
||||
}
|
||||
_ => Attribute::new(AttributeEnum::Unknown, None),
|
||||
}
|
||||
}
|
||||
}
|
||||
Vendored
+71
@@ -0,0 +1,71 @@
|
||||
//
|
||||
// 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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
use std::{
|
||||
collections::{HashMap, HashSet},
|
||||
sync::LazyLock,
|
||||
};
|
||||
|
||||
use crate::{
|
||||
account::{old_state::AccountRunningState, state::DownloadState},
|
||||
database::ModelsAdapter,
|
||||
};
|
||||
use mailbox::MailBox;
|
||||
use native_db::Models;
|
||||
|
||||
pub mod download;
|
||||
pub mod mailbox;
|
||||
pub mod task;
|
||||
|
||||
pub static MAILBOX_MODELS: LazyLock<Models> = LazyLock::new(|| {
|
||||
let mut adapter = ModelsAdapter::new();
|
||||
adapter.register_model::<MailBox>();
|
||||
adapter.register_model::<AccountRunningState>();
|
||||
adapter.register_model::<DownloadState>();
|
||||
adapter.models
|
||||
});
|
||||
|
||||
pub fn find_missing_mailboxes(
|
||||
local_mailboxes: &[MailBox],
|
||||
server_mailboxes: &[MailBox],
|
||||
) -> Vec<MailBox> {
|
||||
let local_names: HashSet<_> = local_mailboxes.iter().map(|m| &m.name).collect();
|
||||
server_mailboxes
|
||||
.iter()
|
||||
.filter(|m| !local_names.contains(&m.name))
|
||||
.cloned()
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub fn find_intersecting_mailboxes(
|
||||
local_mailboxes: &[MailBox],
|
||||
remote_mailboxes: &[MailBox],
|
||||
) -> Vec<(MailBox, MailBox)> {
|
||||
let local_map: HashMap<_, _> = local_mailboxes
|
||||
.iter()
|
||||
.map(|m| (m.name.clone(), m.clone()))
|
||||
.collect();
|
||||
remote_mailboxes
|
||||
.iter()
|
||||
.filter_map(|m| {
|
||||
local_map
|
||||
.get(&m.name)
|
||||
.map(|local_mailbox| (local_mailbox.clone(), m.clone()))
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
Vendored
+154
@@ -0,0 +1,154 @@
|
||||
//
|
||||
// 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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
use crate::account::entity::AuthType;
|
||||
use crate::account::state::DownloadState;
|
||||
use crate::cache::imap::download::process_imap_download;
|
||||
use crate::common::periodic::{PeriodicTask, TaskHandle};
|
||||
use crate::oauth2::token::OAuth2AccessToken;
|
||||
use crate::{account::migration::AccountModel, error::BichonResult};
|
||||
use crate::utc_now;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::atomic::{AtomicI64, Ordering};
|
||||
use std::{sync::LazyLock, time::Duration};
|
||||
use tokio::sync::Mutex;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tracing::{error, info, warn};
|
||||
|
||||
static _DESCRIPTION: &str = "This task periodically synchronizes mailbox data for a specified account, ensuring that all local data is up-to-date.";
|
||||
const TASK_INTERVAL: Duration = Duration::from_secs(10);
|
||||
pub static SYNC_TASKS: LazyLock<AccountSyncTask> = LazyLock::new(AccountSyncTask::new);
|
||||
static LAST_WARN_TIME: AtomicI64 = AtomicI64::new(0);
|
||||
const WARN_INTERVAL_MS: i64 = 600_000;
|
||||
|
||||
pub struct AccountSyncTask {
|
||||
tasks: Mutex<Option<HashMap<u64, (TaskHandle, CancellationToken)>>>,
|
||||
}
|
||||
|
||||
impl AccountSyncTask {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
tasks: Mutex::new(Some(HashMap::new())),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn start_account_download_task(&self, account_id: u64, email: String) {
|
||||
let task_name = format!("account-download-task-{}-{}", account_id, &email);
|
||||
let periodic_task = PeriodicTask::new(&task_name);
|
||||
|
||||
let cancel_token = CancellationToken::new();
|
||||
let task_token = cancel_token.clone();
|
||||
|
||||
let task = move |param: Option<u64>| {
|
||||
let account_id = param.unwrap();
|
||||
let internal_token = task_token.clone();
|
||||
Box::pin(async move {
|
||||
let account = AccountModel::async_get(account_id).await.ok();
|
||||
match account {
|
||||
Some(account) => {
|
||||
if !account.enabled {
|
||||
let last = LAST_WARN_TIME.load(Ordering::Relaxed);
|
||||
let now = utc_now!();
|
||||
if now - last >= WARN_INTERVAL_MS {
|
||||
LAST_WARN_TIME.store(now, Ordering::Relaxed);
|
||||
warn!(
|
||||
"Account {}: download aborted. Account is currently disabled.",
|
||||
account_id
|
||||
);
|
||||
}
|
||||
} else {
|
||||
if let Some(imap) = &account.imap {
|
||||
if let AuthType::OAuth2 = imap.auth.auth_type {
|
||||
if OAuth2AccessToken::get(account.id).await?.is_none() {
|
||||
if utc_now!() % 300_000 == 0 {
|
||||
warn!("Account {}: download aborted. OAuth2 authorization not completed. Please visit the rustmailer admin page to authorize this account.", account_id);
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
}
|
||||
if let Err(e) = process_imap_download(&account, internal_token).await {
|
||||
DownloadState::append_global_error_message(
|
||||
account.id,
|
||||
format!("error in account download task: {:#?}", e),
|
||||
)
|
||||
.await?;
|
||||
error!(
|
||||
"Failed to download mailbox data for '{}': {:?}",
|
||||
account_id, e
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
None => {
|
||||
error!(
|
||||
"Account {}: download aborted. Account entity not found.",
|
||||
account_id
|
||||
);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
})
|
||||
};
|
||||
let handler = periodic_task.start(task, Some(account_id), TASK_INTERVAL, true, true);
|
||||
self.add_task(account_id, (handler, cancel_token)).await;
|
||||
}
|
||||
|
||||
pub async fn add_task(&self, account_id: u64, handler: (TaskHandle, CancellationToken)) {
|
||||
let mut guard = self.tasks.lock().await;
|
||||
if let Some(map) = guard.as_mut() {
|
||||
map.insert(account_id, handler);
|
||||
} else {
|
||||
tracing::error!("Failed to add task: HashMap has been taken during shutdown.");
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn stop(&self, account_id: u64) -> BichonResult<()> {
|
||||
let mut guard = self.tasks.lock().await;
|
||||
if let Some(map) = guard.as_mut() {
|
||||
if let Some((handler, token)) = map.remove(&account_id) {
|
||||
drop(guard);
|
||||
token.cancel();
|
||||
handler.cancel().await;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn shutdown(&self) {
|
||||
let mut guard = self.tasks.lock().await;
|
||||
if let Some(map) = guard.take() {
|
||||
drop(guard);
|
||||
for (account_id, (handler, token)) in map {
|
||||
info!(
|
||||
"Shutdown: Sending cancel signal to account {}...",
|
||||
account_id
|
||||
);
|
||||
token.cancel();
|
||||
if let Err(_) = tokio::time::timeout(Duration::from_secs(5), handler.stop()).await
|
||||
{
|
||||
error!(
|
||||
"Shutdown: Account {} download task forced timeout.",
|
||||
account_id
|
||||
);
|
||||
}
|
||||
}
|
||||
info!("Shutdown: All download tasks processed.");
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user