initial commit

src/modules/imap/capabilities.rs

@@ -0,0 +1,51 @@
+//
+// Copyright (c) 2025 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::modules::error::code::ErrorCode;
+use crate::modules::imap::session::SessionStream;
+use crate::{modules::error::BichonResult, raise_error};
+use async_imap::types::Capability;
+use async_imap::{types::Capabilities, Session};
+
+pub async fn fetch_capabilities(
+    session: &mut Session<Box<dyn SessionStream>>,
+) -> BichonResult<Capabilities> {
+    session
+        .capabilities()
+        .await
+        .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))
+}
+
+pub fn check_capabilities(capabilities: &Capabilities) -> BichonResult<()> {
+    if !capabilities.has_str("IMAP4rev1") {
+        return Err(raise_error!(
+            "Server does not support IMAP4rev1".into(),
+            ErrorCode::Incompatible
+        ));
+    }
+    Ok(())
+}
+
+pub fn capability_to_string(capability: &Capability) -> String {
+    match capability {
+        Capability::Imap4rev1 => "IMAP4rev1".into(),
+        Capability::Auth(v) => format!("AUTH={}", v),
+        Capability::Atom(v) => v.into(),
+    }
+}

src/modules/imap/client.rs

@@ -0,0 +1,251 @@
+//
+// Copyright (c) 2025 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::modules::account::entity::Encryption;
+use crate::modules::error::code::ErrorCode;
+use crate::modules::error::BichonResult;
+use crate::modules::imap::session::SessionStream;
+use crate::modules::imap::stats::StatsWrapper;
+use crate::modules::utils::net::establish_tcp_connection_with_timeout;
+use crate::modules::utils::net::establish_tls_connection;
+use crate::modules::utils::tls::establish_tls_stream;
+use crate::raise_error;
+use async_imap::Client as ImapClient;
+use async_imap::Session as ImapSession;
+use std::net::SocketAddr;
+use std::net::ToSocketAddrs;
+use std::ops::Deref;
+use std::ops::DerefMut;
+use tokio::io::BufWriter;
+use tracing::debug;
+
+#[derive(Debug)]
+pub(crate) struct Client {
+    inner: ImapClient<Box<dyn SessionStream>>,
+}
+
+impl Deref for Client {
+    type Target = ImapClient<Box<dyn SessionStream>>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.inner
+    }
+}
+
+impl DerefMut for Client {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.inner
+    }
+}
+
+fn alpn(port: u16) -> &'static [&'static str] {
+    if port == 993 {
+        &[]
+    } else {
+        &["imap"]
+    }
+}
+
+impl Client {
+    fn new(stream: Box<dyn SessionStream>) -> Self {
+        Self {
+            inner: ImapClient::new(stream),
+        }
+    }
+
+    pub(crate) async fn login(
+        self,
+        username: &str,
+        password: &str,
+    ) -> BichonResult<ImapSession<Box<dyn SessionStream>>> {
+        let Client { inner, .. } = self;
+        let session = inner.login(username, password).await.map_err(|(e, _)| {
+            raise_error!(format!("{:#?}", e), ErrorCode::ImapAuthenticationFailed)
+        })?;
+        Ok(session)
+    }
+
+    pub(crate) async fn authenticate(
+        self,
+        authenticator: impl async_imap::Authenticator,
+    ) -> BichonResult<ImapSession<Box<dyn SessionStream>>> {
+        let Client { inner, .. } = self;
+        let session = inner
+            .authenticate("XOAUTH2", authenticator)
+            .await
+            .map_err(|(e, _)| {
+                raise_error!(format!("{:#?}", e), ErrorCode::ImapAuthenticationFailed)
+            })?;
+        Ok(session)
+    }
+
+    pub async fn connection(
+        domain: &str,
+        encryption: &Encryption,
+        port: u16,
+        use_proxy: Option<u64>,
+    ) -> BichonResult<Self> {
+        let resolved_addr = Self::resolve_to_socket_addr(domain, port)?;
+        debug!("Attempting IMAP connection to {domain} ({resolved_addr}).");
+        match encryption {
+            Encryption::Ssl => {
+                Self::establish_secure_connection(resolved_addr, domain, use_proxy).await
+            }
+            Encryption::StartTls => {
+                Self::establish_starttls_connection(resolved_addr, domain, use_proxy).await
+            }
+            Encryption::None => Self::establish_insecure_connection(resolved_addr, use_proxy).await,
+        }
+    }
+
+    async fn establish_secure_connection(
+        address: SocketAddr,
+        server_hostname: &str,
+        use_proxy: Option<u64>,
+    ) -> BichonResult<Self> {
+        // Establish the TLS connection with the specified parameters
+        let tls_stream =
+            establish_tls_connection(address, server_hostname, alpn(address.port()), use_proxy)
+                .await?;
+        let stats_stream = StatsWrapper::new(tls_stream);
+        // Wrap the TLS stream in a buffered writer for efficient IO
+        let buffered_stream = BufWriter::new(stats_stream);
+        // Create a SessionStream trait object for further communication
+        let session_stream = Box::new(buffered_stream);
+        // Initialize the client with the session stream
+        let mut client = Client::new(session_stream);
+        // Read and validate the greeting response
+        let _greeting = client
+            .read_response()
+            .await
+            .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?
+            .ok_or_else(|| {
+                raise_error!(
+                    "failed to read greeting".into(),
+                    ErrorCode::ImapCommandFailed
+                )
+            })?;
+
+        // Return the established client
+        Ok(client)
+    }
+
+    async fn establish_insecure_connection(
+        address: SocketAddr,
+        use_proxy: Option<u64>,
+    ) -> BichonResult<Self> {
+        // Establish the TCP connection without encryption
+        let tcp_stream = establish_tcp_connection_with_timeout(address, use_proxy).await?;
+        let stats_stream = StatsWrapper::new(tcp_stream);
+        // Wrap the TCP stream in a buffered writer for efficient IO
+        let buffered_stream = BufWriter::new(stats_stream);
+        // Create a SessionStream trait object for further communication
+        let session_stream: Box<dyn SessionStream> = Box::new(buffered_stream);
+        // Initialize the client with the session stream
+        let mut client = Client::new(session_stream);
+
+        // Read and validate the greeting response
+        let _greeting = client
+            .read_response()
+            .await
+            .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?
+            .ok_or_else(|| {
+                raise_error!(
+                    "failed to read greeting".into(),
+                    ErrorCode::ImapCommandFailed
+                )
+            })?;
+
+        // Return the established client
+        Ok(client)
+    }
+
+    async fn establish_starttls_connection(
+        address: SocketAddr,
+        server_hostname: &str,
+        use_proxy: Option<u64>,
+    ) -> BichonResult<Self> {
+        // Establish the initial TCP connection
+        let tcp_stream = establish_tcp_connection_with_timeout(address, use_proxy).await?;
+        let stats_stream = StatsWrapper::new(tcp_stream);
+        // Wrap the TCP stream in a buffered writer for efficient IO
+        let buffered_tcp_stream = BufWriter::new(stats_stream);
+
+        // Create a client for communication
+        let mut client = async_imap::Client::new(buffered_tcp_stream);
+
+        // Read and validate the greeting response
+        let _greeting = client
+            .read_response()
+            .await
+            .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?
+            .ok_or_else(|| {
+                raise_error!(
+                    "failed to read greeting".into(),
+                    ErrorCode::ImapCommandFailed
+                )
+            })?;
+
+        // Run the STARTTLS command to upgrade the connection to TLS
+        client
+            .run_command_and_check_ok("STARTTLS", None)
+            .await
+            .map_err(|_| {
+                raise_error!(
+                    "STARTTLS command failed".into(),
+                    ErrorCode::ImapCommandFailed
+                )
+            })?;
+
+        // Extract the TCP stream after running STARTTLS
+        let buffered_tcp_stream = client.into_inner();
+        let tcp_stream = buffered_tcp_stream.into_inner();
+        // Wrap the TCP stream in TLS encryption
+        let tls_stream = establish_tls_stream(server_hostname, &[], tcp_stream).await?;
+        // Wrap the TLS stream in a buffered writer
+        let buffered_stream = BufWriter::new(tls_stream);
+        // Create a SessionStream trait object for further communication
+        let session_stream: Box<dyn SessionStream> = Box::new(buffered_stream);
+        // Initialize the client with the session stream
+        let client = Client::new(session_stream);
+        // Return the established client
+        Ok(client)
+    }
+
+    fn resolve_to_socket_addr(domain: &str, port: u16) -> BichonResult<SocketAddr> {
+        if domain.is_empty() || domain.contains(|c: char| !c.is_ascii() && c != '.') {
+            return Err(raise_error!(
+                "Invalid domain format".into(),
+                ErrorCode::InvalidParameter
+            ));
+        }
+        // Combine domain and port into a single address string
+        let address = format!("{}:{}", domain, port);
+
+        // Resolve the address into a SocketAddr
+        let socket_addrs = address
+            .to_socket_addrs()
+            .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::NetworkError))?;
+
+        // Return the first valid SocketAddr
+        socket_addrs.into_iter().next().ok_or_else(|| {
+            raise_error!("Unable to resolve address".into(), ErrorCode::NetworkError)
+        })
+    }
+}

src/modules/imap/executor.rs

@@ -0,0 +1,241 @@
+//
+// Copyright (c) 2025 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::modules::account::state::AccountRunningState;
+use crate::modules::cache::imap::mailbox::MailBox;
+use crate::modules::cache::imap::sync::flow::{generate_uid_sequence_hashset, BATCH_SIZE};
+use crate::modules::envelope::extractor::extract_envelope;
+use crate::modules::error::code::ErrorCode;
+use crate::modules::indexer::manager::{EML_INDEX_MANAGER, ENVELOPE_INDEX_MANAGER};
+use crate::modules::indexer::schema::SchemaTools;
+use crate::modules::{error::BichonResult, imap::manager::ImapConnectionManager};
+use crate::raise_error;
+use async_imap::types::{Mailbox, Name};
+use bb8::Pool;
+use futures::TryStreamExt;
+use std::collections::HashSet;
+use tantivy::doc;
+use tracing::info;
+
+const BODY_FETCH_COMMAND: &str = "(UID INTERNALDATE RFC822.SIZE BODY.PEEK[])";
+
+pub struct ImapExecutor {
+    pool: Pool<ImapConnectionManager>,
+}
+
+impl ImapExecutor {
+    pub fn new(pool: Pool<ImapConnectionManager>) -> Self {
+        Self { pool }
+    }
+
+    pub async fn list_all_mailboxes(&self) -> BichonResult<Vec<Name>> {
+        let mut session = self.pool.get().await?;
+        let list = session
+            .list(Some(""), Some("*"))
+            .await
+            .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?;
+        let result = list
+            .try_collect::<Vec<Name>>()
+            .await
+            .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?;
+        Ok(result)
+    }
+
+    pub async fn examine_mailbox(&self, mailbox_name: &str) -> BichonResult<Mailbox> {
+        let mut session = self.pool.get().await?;
+        session
+            .examine(mailbox_name)
+            .await
+            .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))
+    }
+
+    pub async fn uid_search(&self, mailbox_name: &str, query: &str) -> BichonResult<HashSet<u32>> {
+        let mut session = self.pool.get().await?;
+        session
+            .examine(mailbox_name)
+            .await
+            .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?;
+        let result = session
+            .uid_search(query)
+            .await
+            .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?;
+        Ok(result)
+    }
+
+    pub async fn fetch_new_mail(
+        &self,
+        account_id: u64,
+        mailbox: &MailBox,
+        start_uid: u64,
+    ) -> BichonResult<()> {
+        assert!(start_uid > 0, "start_uid must be greater than 0");
+        let uid_list = self
+            .uid_search(
+                &mailbox.encoded_name(),
+                format!("UID {start_uid}:*").as_str(),
+            )
+            .await?;
+
+        let len = uid_list.len();
+        if len == 0 {
+            return Ok(());
+        }
+        info!(
+            "[account {}][mailbox {}] {} envelopes need to be fetched",
+            account_id, mailbox.name, len
+        );
+
+        let mut uid_vec: Vec<u32> = uid_list.into_iter().collect();
+        uid_vec.sort();
+        let uid_batches = generate_uid_sequence_hashset(uid_vec, BATCH_SIZE as usize, false);
+
+        let too_many = len as u32 > 10 * BATCH_SIZE;
+        if too_many {
+            AccountRunningState::set_initial_current_syncing_folder(
+                account_id,
+                mailbox.name.clone(),
+                uid_batches.len() as u32,
+            )
+            .await?;
+        }
+
+        for (index, batch) in uid_batches.into_iter().enumerate() {
+            if too_many {
+                AccountRunningState::set_current_sync_batch_number(
+                    account_id,
+                    mailbox.name.clone(),
+                    (index + 1) as u32,
+                )
+                .await?;
+            }
+            self.uid_batch_retrieve_emails(account_id, mailbox.id, &batch, &mailbox.encoded_name())
+                .await?;
+        }
+        Ok(())
+    }
+
+    pub async fn batch_retrieve_emails(
+        &self,
+        account_id: u64,
+        mailbox_id: u64,
+        page: u64,
+        page_size: u64,
+        encoded_mailbox_name: &str,
+        desc: bool,
+    ) -> BichonResult<usize> {
+        assert!(page > 0, "Page number must be greater than 0");
+        assert!(page_size > 0, "Page size must be greater than 0");
+
+        let mut session = self.pool.get().await?;
+        let total = session
+            .examine(encoded_mailbox_name)
+            .await
+            .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?
+            .exists as u64;
+
+        if total == 0 {
+            return Ok(0);
+        }
+
+        let (start, end) = if desc {
+            // Fetch messages starting from the newest (descending order)
+            let end = total.saturating_sub((page - 1) * page_size);
+            if end == 0 {
+                return Ok(0);
+            }
+            // Calculate start as end - page_size + 1 to avoid off-by-one errors
+            let start = end.saturating_sub(page_size - 1).max(1);
+            (start, end)
+        } else {
+            // Fetch messages starting from the oldest (ascending order)
+            let start = (page - 1) * page_size + 1;
+            if start > total {
+                return Ok(0);
+            }
+            // Calculate end, capped by the total number of messages
+            let end = (start + page_size - 1).min(total);
+            (start, end)
+        };
+
+        let sequence_set = format!("{}:{}", start, end);
+        info!(
+            "Fetching mailbox '{}' messages: sequence {} (page {}, page_size {}, desc={})",
+            encoded_mailbox_name, sequence_set, page, page_size, desc
+        );
+
+        let mut stream = session
+            .fetch(sequence_set.as_str(), BODY_FETCH_COMMAND)
+            .await
+            .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?;
+
+        let mut count = 0;
+        let fields = SchemaTools::eml_fields();
+        while let Some(fetch) = stream
+            .try_next()
+            .await
+            .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?
+        {
+            let envelope = extract_envelope(&fetch, account_id, mailbox_id)?;
+            ENVELOPE_INDEX_MANAGER
+                .add_document(envelope.id, envelope.to_document(mailbox_id)?)
+                .await;
+            let body = fetch.body().ok_or_else(|| {
+                raise_error!("missing a body".into(), ErrorCode::ImapUnexpectedResult)
+            })?;
+            EML_INDEX_MANAGER.add_document( envelope.id, doc!(fields.f_id => envelope.id, fields.f_account_id => account_id, fields.f_mailbox_id => mailbox_id, fields.f_eml => body)).await;
+            count += 1;
+        }
+        Ok(count)
+    }
+
+    pub async fn uid_batch_retrieve_emails(
+        &self,
+        account_id: u64,
+        mailbox_id: u64,
+        uid_set: &str,
+        encoded_mailbox_name: &str,
+    ) -> BichonResult<()> {
+        let mut session = self.pool.get().await?;
+        session
+            .examine(encoded_mailbox_name)
+            .await
+            .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?;
+
+        let mut stream = session
+            .uid_fetch(uid_set, BODY_FETCH_COMMAND)
+            .await
+            .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?;
+        let fields = SchemaTools::eml_fields();
+        while let Some(fetch) = stream
+            .try_next()
+            .await
+            .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?
+        {
+            let envelope = extract_envelope(&fetch, account_id, mailbox_id)?;
+            ENVELOPE_INDEX_MANAGER
+                .add_document(envelope.id, envelope.to_document(mailbox_id)?)
+                .await;
+            let body = fetch.body().ok_or_else(|| {
+                raise_error!("missing a body".into(), ErrorCode::ImapUnexpectedResult)
+            })?;
+            EML_INDEX_MANAGER.add_document( envelope.id, doc!(fields.f_id => envelope.id, fields.f_account_id => account_id, fields.f_mailbox_id => mailbox_id, fields.f_eml => body)).await;
+        }
+        Ok(())
+    }
+}

src/modules/imap/manager.rs

@@ -0,0 +1,155 @@
+//
+// Copyright (c) 2025 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::modules::account::dispatcher::STATUS_DISPATCHER;
+use crate::modules::account::entity::AuthType;
+use crate::modules::account::migration::{AccountModel, AccountType};
+use crate::modules::error::code::ErrorCode;
+use crate::modules::error::BichonResult;
+use crate::modules::imap::capabilities::{
+    capability_to_string, check_capabilities, fetch_capabilities,
+};
+use crate::modules::imap::client::Client;
+use crate::modules::imap::oauth2::OAuth2;
+use crate::modules::imap::session::SessionStream;
+use crate::modules::oauth2::token::OAuth2AccessToken;
+use crate::{decrypt, raise_error};
+use async_imap::Session;
+use tracing::error;
+
+#[derive(Debug)]
+pub struct ImapConnectionManager {
+    pub account_id: u64,
+}
+
+impl ImapConnectionManager {
+    pub fn new(account_id: u64) -> Self {
+        Self { account_id }
+    }
+
+    pub async fn fetch_account(&self) -> BichonResult<AccountModel> {
+        // Fetch the account entity in non-test environment
+        AccountModel::get(self.account_id).await
+    }
+
+    async fn create_client(&self, account: &AccountModel) -> BichonResult<Client> {
+        assert_eq!(account.account_type, AccountType::IMAP);
+        let imap = account.imap.as_ref().unwrap();
+        Client::connection(&imap.host, &imap.encryption, imap.port, imap.use_proxy).await
+    }
+
+    async fn authenticate(
+        &self,
+        client: Client,
+        account: &AccountModel,
+    ) -> BichonResult<Session<Box<dyn SessionStream>>> {
+        assert_eq!(account.account_type, AccountType::IMAP);
+        let imap = account.imap.as_ref().unwrap();
+        match &imap.auth.auth_type {
+            AuthType::Password => {
+                let password = &imap.auth.password.clone().ok_or_else(|| {
+                    raise_error!(
+                        "Imap auth type is Passwd, but password not set".into(),
+                        ErrorCode::MissingConfiguration
+                    )
+                })?;
+
+                let password = decrypt!(&password)?;
+                client.login(&account.email, &password).await
+            }
+            AuthType::OAuth2 => {
+                let record = OAuth2AccessToken::get(self.account_id).await?;
+                let access_token = record.and_then(|r| r.access_token).ok_or_else(|| {
+                    raise_error!(
+                        "Imap auth type is OAuth2, but OAuth2 authorization is not yet complete."
+                            .into(),
+                        ErrorCode::MissingConfiguration
+                    )
+                })?;
+                client
+                    .authenticate(OAuth2::new(account.email.clone(), access_token))
+                    .await
+            }
+        }
+    }
+
+    pub async fn build(&self) -> BichonResult<Session<Box<dyn SessionStream>>> {
+        let account = self.fetch_account().await?;
+        let client = match self.create_client(&account).await {
+            Ok(client) => client,
+            Err(error) => {
+                error!(
+                    "Failed to create IMAP {}'s client: {:#?}",
+                    &account.email, error
+                );
+                STATUS_DISPATCHER
+                    .append_error(
+                        self.account_id,
+                        format!("imap client connect error: {:#?}", error),
+                    )
+                    .await;
+                return Err(error);
+            }
+        };
+
+        let mut session = match self.authenticate(client, &account).await {
+            Ok(session) => session,
+            Err(error) => {
+                error!("Failed to authenticate IMAP session: {:#?}", error);
+
+                STATUS_DISPATCHER
+                    .append_error(
+                        self.account_id,
+                        format!("imap client authenticate error: {:#?}", error),
+                    )
+                    .await;
+                return Err(error);
+            }
+        };
+
+        match fetch_capabilities(&mut session).await {
+            Ok(capabilities) => {
+                let to_save: Vec<String> = capabilities.iter().map(capability_to_string).collect();
+                AccountModel::update_capabilities(self.account_id, to_save).await?;
+                if let Err(error) = check_capabilities(&capabilities) {
+                    error!("Failed to check IMAP capabilities: {:#?}", error);
+                    STATUS_DISPATCHER
+                        .append_error(
+                            self.account_id,
+                            format!("imap client check capabilities error: {:#?}", error),
+                        )
+                        .await;
+                    return Err(error);
+                }
+            }
+            Err(error) => {
+                error!("Failed to fetch IMAP capabilities: {:#?}", error);
+                STATUS_DISPATCHER
+                    .append_error(
+                        self.account_id,
+                        format!("imap client fetch capabilities error: {:#?}", error),
+                    )
+                    .await;
+                return Err(error);
+            }
+        }
+
+        Ok(session)
+    }
+}

src/modules/imap/mod.rs

@@ -0,0 +1,29 @@
+//
+// Copyright (c) 2025 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/>.
+
+
+pub mod capabilities;
+pub mod client;
+pub mod executor;
+pub mod manager;
+pub mod oauth2;
+pub mod pool;
+pub mod session;
+pub mod stats;
+#[cfg(test)]
+mod tests;

src/modules/imap/oauth2.rs

@@ -0,0 +1,41 @@
+//
+// Copyright (c) 2025 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/>.
+
+
+#[derive(Debug)]
+pub struct OAuth2 {
+    user: String,
+    access_token: String,
+}
+
+impl OAuth2 {
+    pub fn new(user: String, access_token: String) -> Self {
+        Self { user, access_token }
+    }
+}
+
+impl async_imap::Authenticator for OAuth2 {
+    type Response = String;
+
+    fn process(&mut self, _data: &[u8]) -> Self::Response {
+        format!(
+            "user={}\x01auth=Bearer {}\x01\x01",
+            self.user, self.access_token
+        )
+    }
+}

src/modules/imap/pool.rs

@@ -0,0 +1,60 @@
+//
+// Copyright (c) 2025 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::modules::error::code::ErrorCode;
+use crate::modules::error::{BichonError, BichonResult};
+use crate::modules::imap::{manager::ImapConnectionManager, session::SessionStream};
+use crate::raise_error;
+use async_imap::Session;
+use bb8::Pool;
+use std::time::Duration;
+
+impl bb8::ManageConnection for ImapConnectionManager {
+    type Connection = Session<Box<dyn SessionStream>>;
+
+    type Error = BichonError;
+
+    async fn connect(&self) -> BichonResult<Self::Connection> {
+        self.build().await
+    }
+    // call this function before using the connection
+    async fn is_valid(&self, conn: &mut Self::Connection) -> BichonResult<()> {
+        conn.noop()
+            .await
+            .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))
+    }
+
+    fn has_broken(&self, _: &mut Self::Connection) -> bool {
+        false
+    }
+}
+
+pub async fn build_imap_pool(account_id: u64) -> BichonResult<Pool<ImapConnectionManager>> {
+    let manager = ImapConnectionManager::new(account_id);
+    let pool = Pool::builder()
+        .connection_timeout(Duration::from_secs(30))
+        .idle_timeout(Duration::from_secs(120))
+        .retry_connection(true)
+        .max_size(10)
+        .test_on_check_out(true)
+        .build(manager)
+        .await?;
+
+    Ok(pool)
+}

src/modules/imap/session.rs

@@ -0,0 +1,52 @@
+//
+// Copyright (c) 2025 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::pin::Pin;
+use tokio::io::{AsyncRead, AsyncWrite, BufWriter};
+use tokio_io_timeout::TimeoutStream;
+
+pub trait SessionStream: AsyncRead + AsyncWrite + Unpin + Send + Sync + std::fmt::Debug {
+    //  Change the read timeout on the session stream.
+    // fn set_read_timeout(&mut self, timeout: Option<Duration>);
+}
+
+impl SessionStream for Box<dyn SessionStream> {
+    // fn set_read_timeout(&mut self, timeout: Option<Duration>) {
+    //     self.as_mut().set_read_timeout(timeout);
+    // }
+}
+
+impl<T: SessionStream> SessionStream for tokio_rustls::client::TlsStream<T> {
+    // fn set_read_timeout(&mut self, timeout: Option<Duration>) {
+    //     self.get_mut().0.set_read_timeout(timeout);
+    // }
+}
+
+impl<T: SessionStream> SessionStream for BufWriter<T> {
+    // fn set_read_timeout(&mut self, timeout: Option<Duration>) {
+    //     self.get_mut().set_read_timeout(timeout);
+    // }
+}
+impl<T: AsyncRead + AsyncWrite + Send + Sync + std::fmt::Debug> SessionStream
+    for Pin<Box<TimeoutStream<T>>>
+{
+    // fn set_read_timeout(&mut self, timeout: Option<Duration>) {
+    //     self.as_mut().set_read_timeout_pinned(timeout);
+    // }
+}

src/modules/imap/stats.rs

@@ -0,0 +1,86 @@
+//
+// Copyright (c) 2025 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::pin::Pin;
+use std::task::{Context, Poll};
+// use std::time::Duration;
+
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
+
+use crate::modules::imap::session::SessionStream;
+
+pub struct StatsWrapper<T> {
+    inner: T,
+}
+
+impl<T> StatsWrapper<T> {
+    pub fn new(inner: T) -> Self {
+        Self { inner }
+    }
+}
+
+impl<T: AsyncRead + Unpin> AsyncRead for StatsWrapper<T> {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<std::io::Result<()>> {
+        // let before = buf.filled().len();
+        let result = Pin::new(&mut self.inner).poll_read(cx, buf);
+        // if let Poll::Ready(Ok(())) = &result {
+        //     // let bytes_read = buf.filled().len() - before;
+        // }
+        result
+    }
+}
+
+impl<T: AsyncWrite + Unpin> AsyncWrite for StatsWrapper<T> {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<std::io::Result<usize>> {
+        let result = Pin::new(&mut self.inner).poll_write(cx, buf);
+        // if let Poll::Ready(Ok(bytes_written)) = &result {
+        // }
+        result
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut self.inner).poll_flush(cx)
+    }
+
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut self.inner).poll_shutdown(cx)
+    }
+}
+
+impl<T: SessionStream> SessionStream for StatsWrapper<T> {
+    // fn set_read_timeout(&mut self, timeout: Option<Duration>) {
+    //     self.inner.set_read_timeout(timeout);
+    // }
+}
+
+impl<T: SessionStream> std::fmt::Debug for StatsWrapper<T> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("StatsWrapper")
+            .field("inner", &self.inner)
+            .finish()
+    }
+}

src/modules/imap/tests.rs

@@ -0,0 +1,48 @@
+//
+// Copyright (c) 2025 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 mail_parser::MessageParser;
+
+use crate::{base64_encode_url_safe, modules::{account::entity::Encryption, imap::client::Client}};
+
+#[tokio::test]
+async fn testxx() {
+    rustls::crypto::CryptoProvider::install_default(rustls::crypto::ring::default_provider())
+        .unwrap();
+    let client = Client::connection("imap.zoho.com".into(), &Encryption::Ssl, 993, None)
+        .await
+        .unwrap();
+    let mut session = client.login("pollybase@zohomail.com", "xxx").await.unwrap();
+    session.select("INBOX").await.unwrap();
+    let result = session.uid_search("LARGER 1024").await.unwrap();
+    println!("{:#?}", result);
+}
+
+#[tokio::test]
+async fn test1() {
+    let path = r"C:\Users\polly\Downloads\test.eml";
+    let eml_data = std::fs::read(path).unwrap();
+    let input = base64_encode_url_safe!(eml_data);
+    let message = MessageParser::default().parse(&input).unwrap();
+    let parts = message.parts;
+    for part in parts {
+        println!("{}", part.is_message());
+        println!("{}", part.is_multipart());
+    }
+}