feat: add manual download and cancel download for email accounts

crates/core/Cargo.toml

@@ -70,3 +70,4 @@ tracing-log.workspace = true
 tokio-util.workspace = true
 whichlang = "0.1.1"
 deunicode = "1.6.2"
+scopeguard = "1.2.0"

crates/core/src/account/migration.rs

@@ -318,7 +318,7 @@ impl AccountV4 {
 
         if matches!(cloned.account_type, AccountType::IMAP) {
             DOWNLOAD_CONTROLLER
-                .trigger_start(cloned.id, cloned.email.clone())
+                .trigger_schedule(cloned.id, cloned.email.clone())
                 .await;
         }
         Ok(cloned)

crates/core/src/cache/imap/download/download_type.rs

@@ -17,6 +17,7 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 use crate::{
+    utc_now,
     {
         account::{
             migration::AccountModel,
@@ -24,7 +25,6 @@ use crate::{
         },
         error::BichonResult,
     },
-    utc_now,
 };
 
 #[derive(Clone, Debug, Eq, PartialEq)]
@@ -34,27 +34,33 @@ pub enum DownloadTask {
     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
-            }
-        }
+pub async fn decide_next_download_task(
+    account: &AccountModel,
+    trigger_type: TriggerType,
+) -> BichonResult<DownloadTask> {
+    let state = match DownloadState::get(account.id).await? {
         None => {
             DownloadState::init(account.id).await?;
-            DownloadTask::FullFetch
+            return Ok(DownloadTask::FullFetch);
         }
-    })
+        Some(s) => s,
+    };
+
+    let should_start = match trigger_type {
+        TriggerType::Manual => true,
+        TriggerType::Scheduled => should_trigger_next_download(
+            state.last_trigger_at,
+            state.last_finished_at.unwrap_or(0),
+            account.download_interval_min.unwrap_or(60),
+        ),
+    };
+
+    if should_start {
+        DownloadState::start_new_session(account.id, trigger_type).await?;
+        Ok(DownloadTask::TraceFetch)
+    } else {
+        Ok(DownloadTask::Idle)
+    }
 }
 
 fn should_trigger_next_download(

crates/core/src/cache/imap/download/mod.rs

@@ -19,33 +19,34 @@
 use crate::{
     account::{
         migration::{AccountModel, AccountType},
-        state::{DownloadState, DownloadStatus},
+        state::{DownloadState, DownloadStatus, TriggerType},
     },
-    cache::imap::{mailbox::MailBox, download::flow::FetchDirection},
+    cache::imap::{download::flow::FetchDirection, mailbox::MailBox},
     error::BichonResult,
     imap::executor::ImapExecutor,
 };
+use download_folders::get_download_folders;
+use download_type::{decide_next_download_task, DownloadTask};
 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 mod flow;
+pub mod rebuild;
 
 pub async fn process_imap_download(
     account: &AccountModel,
     token: CancellationToken,
+    trigger_type: TriggerType,
 ) -> 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?;
+    let download_task = decide_next_download_task(account, trigger_type).await?;
     if matches!(download_task, DownloadTask::Idle) {
         return Ok(());
     }

crates/core/src/cache/imap/task.rs

@@ -17,37 +17,56 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 use crate::account::entity::AuthType;
-use crate::account::state::DownloadState;
+use crate::account::state::{DownloadState, TriggerType};
 use crate::cache::imap::download::process_imap_download;
 use crate::common::periodic::{PeriodicTask, TaskHandle};
+use crate::error::code::ErrorCode;
 use crate::oauth2::token::OAuth2AccessToken;
-use crate::utc_now;
 use crate::{account::migration::AccountModel, error::BichonResult};
-use std::collections::HashMap;
+use crate::{raise_error, utc_now};
+use std::collections::{HashMap, HashSet};
 use std::sync::atomic::{AtomicI64, Ordering};
 use std::{sync::LazyLock, time::Duration};
 use tokio::sync::Mutex;
+use tokio::task::JoinHandle;
 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);
+pub static SYNC_TASKS: LazyLock<AccountDownTask> = LazyLock::new(AccountDownTask::new);
 static LAST_WARN_TIME: AtomicI64 = AtomicI64::new(0);
 const WARN_INTERVAL_MS: i64 = 600_000;
 
-pub struct AccountSyncTask {
+pub struct AccountDownTask {
     tasks: Mutex<Option<HashMap<u64, (TaskHandle, CancellationToken)>>>,
+    manual_tasks: Mutex<HashMap<u64, (JoinHandle<()>, CancellationToken)>>,
+    busy_accounts: Mutex<HashSet<u64>>,
 }
 
-impl AccountSyncTask {
+impl AccountDownTask {
     pub fn new() -> Self {
         Self {
             tasks: Mutex::new(Some(HashMap::new())),
+            manual_tasks: Mutex::new(HashMap::new()),
+            busy_accounts: Mutex::new(HashSet::new()),
         }
     }
 
-    pub async fn start_account_download_task(&self, account_id: u64, email: String) {
+    async fn set_busy(&self, account_id: u64, is_busy: bool) {
+        let mut guard = self.busy_accounts.lock().await;
+        if is_busy {
+            guard.insert(account_id);
+        } else {
+            guard.remove(&account_id);
+        }
+    }
+
+    async fn is_busy(&self, account_id: u64) -> bool {
+        self.busy_accounts.lock().await.contains(&account_id)
+    }
+
+    pub async fn start_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);
 
@@ -58,6 +77,28 @@ impl AccountSyncTask {
             let account_id = param.unwrap();
             let internal_token = task_token.clone();
             Box::pin(async move {
+                if SYNC_TASKS.is_manual_running(account_id).await {
+                    info!(
+                        "Account {}: Scheduled task skipped (Manual task is running).",
+                        account_id
+                    );
+                    return Ok(());
+                }
+
+                if SYNC_TASKS.is_busy(account_id).await {
+                    warn!(
+                        "Account {}: Scheduled task skipped (Previous sync still active).",
+                        account_id
+                    );
+                    return Ok(());
+                }
+
+                SYNC_TASKS.set_busy(account_id, true).await;
+                let _busy_guard = scopeguard::guard(account_id, |id| {
+                    tokio::spawn(async move {
+                        SYNC_TASKS.set_busy(id, false).await;
+                    });
+                });
                 let account = AccountModel::async_get(account_id).await.ok();
                 match account {
                     Some(account) => {
@@ -82,7 +123,13 @@ impl AccountSyncTask {
                                     }
                                 }
                             }
-                            if let Err(e) = process_imap_download(&account, internal_token).await {
+                            if let Err(e) = process_imap_download(
+                                &account,
+                                internal_token,
+                                TriggerType::Scheduled,
+                            )
+                            .await
+                            {
                                 DownloadState::append_session_error(
                                     account.id,
                                     format!("error in account download task: {:#?}", e),
@@ -150,4 +197,70 @@ impl AccountSyncTask {
             info!("Shutdown: All download tasks processed.");
         }
     }
+
+    pub async fn start_manual_task(&self, account_id: u64) -> BichonResult<()> {
+        {
+            if self.is_manual_running(account_id).await {
+                return Err(raise_error!(
+                    "Manual task already running.".into(),
+                    ErrorCode::Forbidden
+                ));
+            }
+            if self.is_busy(account_id).await {
+                return Err(raise_error!(
+                    "The background synchronization is currently active. Please try again in a few seconds.".into(), 
+                    ErrorCode::Forbidden
+                ));
+            }
+        }
+
+        let cancel_token = CancellationToken::new();
+        let token_clone = cancel_token.clone();
+        let handle = tokio::spawn(async move {
+            SYNC_TASKS.set_busy(account_id, true).await;
+            let _cleanup = scopeguard::guard(account_id, |id| {
+                tokio::spawn(async move {
+                    SYNC_TASKS.set_busy(id, false).await;
+                    let mut guard = SYNC_TASKS.manual_tasks.lock().await;
+                    guard.remove(&id);
+                });
+            });
+            if token_clone.is_cancelled() {
+                return;
+            }
+            let account = match AccountModel::async_get(account_id).await {
+                Ok(acc) => acc,
+                Err(e) => {
+                    error!("Failed to fetch account {}: {:?}", account_id, e);
+                    return;
+                }
+            };
+
+            if let Err(e) = process_imap_download(&account, token_clone, TriggerType::Manual).await
+            {
+                error!("Manual download failed for {}: {:?}", account_id, e);
+                let error_msg = format!("error in account download task: {:#?}", e);
+                let _ = DownloadState::append_session_error(account.id, error_msg).await;
+            }
+        });
+        {
+            let mut guard = self.manual_tasks.lock().await;
+            guard.insert(account_id, (handle, cancel_token));
+        }
+
+        Ok(())
+    }
+
+    pub async fn cancel_manual_task(&self, account_id: u64) {
+        let mut guard = self.manual_tasks.lock().await;
+        if let Some((handle, token)) = guard.remove(&account_id) {
+            token.cancel();
+            let _ = handle.await;
+        }
+    }
+
+    pub async fn is_manual_running(&self, account_id: u64) -> bool {
+        let guard = self.manual_tasks.lock().await;
+        guard.contains_key(&account_id)
+    }
 }

crates/core/src/context/controller.rs

@@ -35,11 +35,10 @@ impl DownloadController {
         tokio::spawn(async move {
             while let Some((account_id, email)) = rx.recv().await {
                 match Self::start_download(account_id, email.clone()).await {
-                    Ok(Some(_)) => {}
-                    Ok(None) => {}
+                    Ok(_) => {}
                     Err(err) => {
                         error!(
-                            "Failed to prepare and start download of account {{{}-{}}}, error: {:#?}",
+                            "Failed to prepare and start scheduled download of account {{{}-{}}}, error: {:#?}",
                             &account_id, &email, err
                         );
                     }
@@ -51,7 +50,7 @@ impl DownloadController {
     }
 
     /// Trigger synchronization for a specific account
-    pub async fn trigger_start(&self, account_id: u64, email: String) {
+    pub async fn trigger_schedule(&self, account_id: u64, email: String) {
         if let Err(e) = self.channel.send((account_id, email)).await {
             error!(
                 "Failed to trigger download for account={{{}}}, error: {:?}",
@@ -60,13 +59,13 @@ impl DownloadController {
         }
     }
 
-    async fn start_download(account_id: u64, email: String) -> BichonResult<Option<()>> {
+    async fn start_download(account_id: u64, email: String) -> BichonResult<()> {
         info!(
             "Account download starting for account: {}-{}.",
             account_id, email
         );
-        SYNC_TASKS.start_account_download_task(account_id, email).await;
+        SYNC_TASKS.start_download_task(account_id, email).await;
         tokio::time::sleep(Duration::from_millis(100)).await;
-        Ok(Some(()))
+        Ok(())
     }
 }

crates/core/src/context/executors.rs

@@ -35,7 +35,7 @@ pub struct BichonContext {
 
 impl Initialize for BichonContext {
     async fn initialize() -> BichonResult<()> {
-        BICHON_CONTEXT.start_account_syncers().await
+        BICHON_CONTEXT.start_account_downloader().await
     }
 }
 
@@ -49,7 +49,7 @@ impl BichonContext {
         utc_now!() - self.start_at
     }
 
-    pub async fn start_account_syncers(&self) -> BichonResult<()> {
+    pub async fn start_account_downloader(&self) -> BichonResult<()> {
         let accounts = AccountModel::list_all().await?;
         let active_accounts: Vec<AccountModel> = accounts
             .into_iter()
@@ -66,7 +66,7 @@ impl BichonContext {
         );
         for account in active_accounts {
             DOWNLOAD_CONTROLLER
-                .trigger_start(account.id, account.email)
+                .trigger_schedule(account.id, account.email)
                 .await
         }
 

crates/server/src/rest/api/account.rs

@@ -20,14 +20,17 @@ use crate::common::auth::WrappedContext;
 use crate::rest::api::ApiTags;
 use crate::rest::ApiResult;
 use bichon_core::account::grant::BatchAccountRoleRequest;
-use bichon_core::account::migration::AccountModel;
+use bichon_core::account::migration::{AccountModel, AccountType};
 use bichon_core::account::payload::{
     filter_accessible_accounts, AccountCreateRequest, AccountUpdateRequest, MinimalAccount,
 };
 use bichon_core::account::state::DownloadState;
 use bichon_core::account::stats::AccountStats;
 use bichon_core::account::view::AccountResp;
+use bichon_core::cache::imap::task::SYNC_TASKS;
 use bichon_core::common::paginated::{paginate_vec, DataPage};
+use bichon_core::error::code::ErrorCode;
+use bichon_core::raise_error;
 use bichon_core::store::tantivy::envelope::ENVELOPE_MANAGER;
 use bichon_core::users::permissions::Permission;
 use bichon_core::users::UserModel;
@@ -204,6 +207,68 @@ impl AccountApi {
         Ok(Json(state))
     }
 
+    /// Start a manual download task for an account
+    #[oai(
+        path = "/accounts/:account_id/start-download",
+        method = "post",
+        operation_id = "accounts_start_download"
+    )]
+    async fn accounts_start_download(
+        &self,
+        /// The account ID to start download for
+        account_id: Path<u64>,
+        context: WrappedContext,
+    ) -> ApiResult<()> {
+        let account_id = account_id.0;
+        let account = AccountModel::check_account_exists(account_id).await?;
+        if !matches!(account.account_type, AccountType::IMAP) {
+            return Err(raise_error!(
+                format!("Manual download is not supported for '{:#?}' accounts. Only IMAP accounts are supported.", account.account_type),
+                ErrorCode::InvalidParameter
+            ))?;
+        }
+        context
+            .require_permission(Some(account_id), Permission::ACCOUNT_MANAGE)
+            .await?;
+        SYNC_TASKS.start_manual_task(account_id).await?;
+        Ok(())
+    }
+
+    /// Cancel a running manual download task
+    #[oai(
+        path = "/accounts/:account_id/cancel-download",
+        method = "post",
+        operation_id = "accounts_cancel_download"
+    )]
+    async fn accounts_cancel_download(
+        &self,
+        /// The account ID to cancel download for
+        account_id: Path<u64>,
+        context: WrappedContext,
+    ) -> ApiResult<()> {
+        let account_id = account_id.0;
+        let account = AccountModel::check_account_exists(account_id).await?;
+
+        if !matches!(account.account_type, AccountType::IMAP) {
+            return Err(raise_error!(
+                "This operation is only supported for IMAP accounts.".into(),
+                ErrorCode::InvalidParameter
+            ))?;
+        }
+        context
+            .require_permission(Some(account_id), Permission::ACCOUNT_MANAGE)
+            .await?;
+
+        if !SYNC_TASKS.is_manual_running(account_id).await {
+            return Err(raise_error!(
+                "No running manual task found for this account.".into(),
+                ErrorCode::ResourceNotFound
+            ))?;
+        }
+        SYNC_TASKS.cancel_manual_task(account_id).await;
+        Ok(())
+    }
+
     /// Get the stats of an account
     #[oai(
         path = "/accounts/:account_id/stats",