feat(graph): add Graph API support and GraphAPI account type

src/modules/cache/imap/address.rs

@@ -352,7 +352,7 @@ impl AddressEntity {
         let from = envelope.from.as_ref().map(|f| f.address.clone()).flatten();
         let envelope_hash = envelope.create_envelope_id();
         let date = envelope.date.clone();
-        let internal_date = Some(envelope.internal_date.clone());
+        let internal_date = envelope.internal_date.clone();
         let account_id = envelope.account_id;
         let mailbox_id = envelope.folder_id;
         let mut entities = Vec::new();

src/modules/cache/imap/mod.rs

@@ -11,13 +11,18 @@ use crate::{
             imap::{
                 address::AddressEntity,
                 envelope::EmailEnvelope,
+                migration::{EmailEnvelopeV2, EmailEnvelopeV3},
                 minimal::MinimalEnvelope,
                 thread::EmailThread,
-                migration::{EmailEnvelopeV2, EmailEnvelopeV3},
             },
-            vendor::gmail::sync::{
-                envelope::GmailEnvelope,
-                labels::{GmailCheckPoint, GmailLabels},
+            vendor::{
+                gmail::sync::{
+                    envelope::GmailEnvelope,
+                    labels::{GmailCheckPoint, GmailLabels},
+                },
+                outlook::sync::{
+                    delta::FolderDeltaLink, envelope::OutlookEnvelope, folders::OutlookFolder,
+                },
             },
         },
         database::ModelsAdapter,
@@ -30,13 +35,13 @@ pub mod address;
 pub mod envelope;
 pub mod mailbox;
 pub mod manager;
+pub mod migration;
 pub mod minimal;
 pub mod sync;
 pub mod task;
 #[cfg(test)]
 mod tests;
 pub mod thread;
-pub mod migration;
 
 pub static ENVELOPE_MODELS: LazyLock<Models> = LazyLock::new(|| {
     let mut adapter = ModelsAdapter::new();
@@ -50,6 +55,9 @@ pub static ENVELOPE_MODELS: LazyLock<Models> = LazyLock::new(|| {
     adapter.register_model::<GmailEnvelope>();
     adapter.register_model::<GmailLabels>();
     adapter.register_model::<GmailCheckPoint>();
+    adapter.register_model::<OutlookFolder>();
+    adapter.register_model::<FolderDeltaLink>();
+    adapter.register_model::<OutlookEnvelope>();
     adapter.models
 });
 

src/modules/cache/imap/sync/flow.rs

@@ -15,7 +15,9 @@ use crate::{
                 migration::EmailEnvelopeV3,
                 minimal::MinimalEnvelope,
                 sync::rebuild::{rebuild_mailbox_cache, rebuild_mailbox_cache_since_date},
-            }, sync_type::SyncType, SEMAPHORE
+            },
+            sync_type::SyncType,
+            SEMAPHORE,
         },
         common::AddrVec,
         context::executors::RUST_MAIL_CONTEXT,
@@ -91,7 +93,7 @@ pub async fn fetch_and_save_since_date(
         AccountRunningState::set_initial_current_syncing_folder(
             account_id,
             mailbox.name.clone(),
-            uid_batches.len() as u32,
+            Some(uid_batches.len() as u32),
         )
         .await?;
     }
@@ -177,7 +179,7 @@ pub async fn fetch_and_save_full_mailbox(
         AccountRunningState::set_initial_current_syncing_folder(
             account_id,
             mailbox.name.clone(),
-            total_batches,
+            Some(total_batches),
         )
         .await?;
     }

src/modules/cache/imap/task.rs

@@ -5,6 +5,7 @@
 use crate::modules::account::entity::{AuthType, MailerType};
 use crate::modules::cache::imap::sync::execute_imap_sync;
 use crate::modules::cache::vendor::gmail::sync::execute_gmail_sync;
+use crate::modules::cache::vendor::outlook::sync::execute_outlook_sync;
 use crate::modules::oauth2::token::OAuth2AccessToken;
 use crate::modules::scheduler::periodic::TaskHandle;
 use crate::modules::{
@@ -98,6 +99,26 @@ impl AccountSyncTask {
                                         )
                                     }
                                 }
+                                MailerType::GraphApi => {
+                                    if OAuth2AccessToken::get(account.id).await?.is_none() {
+                                        if utc_now!() % 300_000 == 0 {
+                                            warn!("Account {}: Sync aborted. OAuth2 authorization not completed. Please visit the rustmailer admin page to authorize this account.", account_id);
+                                        }
+                                        return Ok(());
+                                    }
+                                    if let Err(e) = execute_outlook_sync(&account).await {
+                                        STATUS_DISPATCHER
+                                            .append_error(
+                                                account_id,
+                                                format!("error in account sync task: {:#?}", e),
+                                            )
+                                            .await;
+                                        error!(
+                                            "Failed to synchronize mailbox data for '{}': {:?}",
+                                            account_id, e
+                                        )
+                                    }
+                                }
                             }
                         }
                     }

src/modules/cache/model.rs

@@ -4,11 +4,18 @@
 use poem_openapi::Object;
 use serde::{Deserialize, Serialize};
 
-use crate::{calculate_hash, id, modules::{
-    cache::imap::{envelope::Received, mailbox::EnvelopeFlag, migration::EmailEnvelopeV3},
-    common::Addr,
-    imap::section::{EmailBodyPart, ImapAttachment},
-}};
+use crate::{
+    calculate_hash, id,
+    modules::{
+        cache::imap::{
+            envelope::Received,
+            mailbox::{EmailFlag, EnvelopeFlag},
+            migration::EmailEnvelopeV3,
+        },
+        common::Addr,
+        imap::section::{EmailBodyPart, ImapAttachment},
+    },
+};
 
 #[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize, Object)]
 pub struct Envelope {
@@ -111,6 +118,8 @@ pub struct Envelope {
     ///
     /// **Note:** This field is populated only for Gmail API accounts. For other account types, it will be empty.
     pub labels: Vec<String>,
+
+    pub is_read: bool,
 }
 
 impl Envelope {
@@ -134,6 +143,10 @@ impl From<EmailEnvelopeV3> for Envelope {
             mailbox_name: value.mailbox_name,
             internal_date: value.internal_date,
             size: value.size,
+            is_read: value
+                .flags
+                .iter()
+                .any(|f| matches!(f.flag, EmailFlag::Seen)),
             flags: Some(value.flags),
             flags_hash: Some(value.flags_hash),
             bcc: value.bcc,

src/modules/cache/sync_type.rs

@@ -4,7 +4,7 @@
 
 use crate::{
     modules::{
-        account::{entity::MailerType, status::AccountRunningState, migration::AccountModel},
+        account::{entity::MailerType, migration::AccountModel, status::AccountRunningState},
         error::RustMailerResult,
     },
     utc_now,
@@ -51,7 +51,7 @@ pub async fn determine_sync_type(account: &AccountModel) -> RustMailerResult<Syn
                         SyncType::SkipSync
                     }
                 }
-                MailerType::GmailApi => {
+                MailerType::GmailApi | MailerType::GraphApi => {
                     if incremental_sync {
                         AccountRunningState::set_incremental_sync_start(account.id).await?;
                         SyncType::IncrementalSync

src/modules/cache/vendor/gmail/sync/envelope.rs

@@ -8,8 +8,8 @@ use crate::{
         cache::{
             imap::{
                 address::AddressEntity,
-                thread::{EmailThread, EmailThreadKey},
                 migration::EmailEnvelopeV3,
+                thread::{EmailThread, EmailThreadKey},
             },
             model::Envelope,
         },
@@ -398,6 +398,8 @@ impl GmailEnvelope {
     }
 
     pub fn into_envelope(self, label_map: &AHashMap<String, String>) -> Envelope {
+        let is_read = self.label_ids.iter().any(|f| f == "UNREAD");
+
         let labels: Vec<String> = self
             .label_ids
             .into_iter()
@@ -431,6 +433,7 @@ impl GmailEnvelope {
             attachments: None,
             body_meta: None,
             received: None,
+            is_read,
             labels,
         }
     }

src/modules/cache/vendor/gmail/sync/flow.rs

@@ -76,7 +76,7 @@ pub async fn fetch_and_save_since_date(
                 AccountRunningState::set_initial_current_syncing_folder(
                     account_id,
                     label.name.clone(),
-                    total_batches,
+                    Some(total_batches),
                 )
                 .await?;
             }
@@ -134,7 +134,7 @@ pub async fn fetch_and_save_since_date(
                 history_ids.push(hid.to_string());
             }
             GmailEnvelope::save_envelopes(envelopes).await?;
-        } 
+        }
         // Break if API response has no next page
         if page_token.is_none() {
             break;
@@ -175,7 +175,7 @@ pub async fn fetch_and_save_full_label(
         AccountRunningState::set_initial_current_syncing_folder(
             account_id,
             label.name.clone(),
-            total_batches,
+            Some(total_batches),
         )
         .await?;
     }

src/modules/cache/vendor/gmail/sync/mod.rs

@@ -103,7 +103,7 @@ pub async fn execute_gmail_sync(account: &AccountModel) -> RustMailerResult<()>
 
     if !deleted_labels.is_empty() {
         info!(
-            "Account {}: Detected {} mailboxes missing from the IMAP server (not found in the LSUB response). \
+            "Account {}: Detected {} mailboxes missing from the Gmail server (not found in the Gmail API response). \
             Now cleaning up these mailboxes and their associated metadata locally.",
             account.id, deleted_labels.len()
         );

src/modules/cache/vendor/outlook/model.rs

@@ -128,3 +128,35 @@ pub struct Attachment {
     #[serde(rename = "microsoft.graph.fileAttachment/contentId")]
     pub content_id: Option<String>,
 }
+
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+pub struct DeltaResponse {
+    #[serde(rename = "@odata.context")]
+    pub context: Option<String>,
+
+    #[serde(rename = "@odata.nextLink")]
+    pub next_link: Option<String>,
+
+    #[serde(rename = "@odata.deltaLink")]
+    pub delta_link: Option<String>,
+
+    pub value: Option<Vec<PartialMessage>>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+pub struct PartialMessage {
+    #[serde(rename = "@odata.etag")]
+    pub etag: Option<String>,
+
+    #[serde(rename = "@odata.type")]
+    pub odata_type: Option<String>,
+
+    pub id: String,
+    #[serde(rename = "@removed")]
+    pub removed: Option<RemovedInfo>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+pub struct RemovedInfo {
+    pub reason: String,
+}

src/modules/cache/vendor/outlook/sync/client.rs

@@ -1,6 +1,8 @@
 use crate::{
     modules::{
-        cache::vendor::outlook::model::{MailFolder, MailFoldersResponse, MessageListResponse},
+        cache::vendor::outlook::model::{
+            MailFolder, MailFoldersResponse, Message, MessageListResponse,
+        },
         error::{code::ErrorCode, RustMailerResult},
         hook::http::HttpClient,
         oauth2::token::OAuth2AccessToken,
@@ -12,7 +14,7 @@ use std::{future::Future, pin::Pin};
 pub struct OutlookClient;
 
 impl OutlookClient {
-    async fn get_access_token(account_id: u64) -> RustMailerResult<String> {
+    pub async fn get_access_token(account_id: u64) -> RustMailerResult<String> {
         let record = OAuth2AccessToken::get(account_id).await?;
         record.and_then(|r| r.access_token).ok_or_else(|| {
             raise_error!(
@@ -45,7 +47,7 @@ impl OutlookClient {
         prefix: &'a str,
         output: &'a mut Vec<MailFolder>,
         access_token: &'a str,
-    ) -> Pin<Box<dyn Future<Output = RustMailerResult<()>> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = RustMailerResult<()>> + Send + 'a>> {
         Box::pin(async move {
             let mut url = match folder_id {
                 Some(id) => {
@@ -84,13 +86,15 @@ impl OutlookClient {
         })
     }
 
-    async fn get_folder<'a>(
-        client: &'a HttpClient,
-        default_folder_name: &'a str,
-        access_token: &'a str,
+    pub async fn get_folder(
+        account_id: u64,
+        use_proxy: Option<u64>,
+        default_folder_name: &str,
     ) -> RustMailerResult<MailFolder> {
+        let client = HttpClient::new(use_proxy).await?;
+        let access_token = Self::get_access_token(account_id).await?;
         let url = format!("https://graph.microsoft.com/v1.0/me/mailFolders/{default_folder_name}");
-        let value = client.get(&url, access_token).await.map_err(|e| {
+        let value = client.get(&url, &access_token).await.map_err(|e| {
             raise_error!(format!("Request error: {e:#?}"), ErrorCode::InternalError)
         })?;
         let folder = serde_json::from_value::<MailFolder>(value)
@@ -109,16 +113,6 @@ impl OutlookClient {
         let access_token = Self::get_access_token(account_id).await?;
         let mut result = Vec::new();
         Self::fetch_recursive(&client, None, "", &mut result, &access_token).await?;
-        let inbox = Self::get_folder(&client, "inbox", &access_token).await?;
-        let sentitems = Self::get_folder(&client, "sentitems", &access_token).await?;
-        for folder in &mut result {
-            if folder.id == inbox.id {
-                folder.display_name = "inbox".to_string();
-            }
-            if folder.id == sentitems.id {
-                folder.display_name = "sentitems".to_string();
-            }
-        }
         Ok(result)
     }
 
@@ -171,12 +165,9 @@ impl OutlookClient {
         use_proxy: Option<u64>,
         folder_id: &str,
     ) -> RustMailerResult<String> {
-        let mut url = format!("https://graph.microsoft.com/v1.0/me/mailFolders/{folder_id}/messages/delta?\
-            $select=id,isRead,conversationId,internetMessageId,from,body,toRecipients,ccRecipients,\
-            bccRecipients,replyTo,sender,subject,receivedDateTime,sentDateTime,isRead,bodyPreview,categories&\
-            $expand=attachments($select=id,name,contentType,size,isInline,microsoft.graph.fileAttachment/contentId)&\
-            $orderBy=receivedDateTime desc
-        ");
+        let mut url = format!(
+            "https://graph.microsoft.com/v1.0/me/mailFolders/{folder_id}/messages/delta?$select=id"
+        );
         let client = HttpClient::new(use_proxy).await?;
         let access_token = Self::get_access_token(account_id).await?;
         loop {
@@ -211,45 +202,28 @@ impl OutlookClient {
         }
     }
 
-    pub async fn list_delta(
+    pub async fn get_message(
         account_id: u64,
         use_proxy: Option<u64>,
-        delta_link: &str,
-    ) -> RustMailerResult<String> {
-        let mut url = delta_link.to_string();
+        id: &str,
+    ) -> RustMailerResult<Message> {
+        let url = format!("https://graph.microsoft.com/v1.0/me/messages/{id}?\
+               $select=id,isRead,conversationId,internetMessageId,from,body,toRecipients,ccRecipients,\
+               bccRecipients,replyTo,sender,subject,receivedDateTime,sentDateTime,isRead,bodyPreview,categories&\
+               $expand=attachments($select=id,name,contentType,size,isInline,microsoft.graph.fileAttachment/contentId)");
+
         let client = HttpClient::new(use_proxy).await?;
         let access_token = Self::get_access_token(account_id).await?;
-        loop {
-            let value = client.get(url.as_str(), &access_token).await?;
-            if let Some(next_link) = value.get("@odata.nextLink") {
-                //处理这一页的delta数据
-                url = next_link
-                    .as_str()
-                    .ok_or_else(|| {
-                        raise_error!(
-                            format!("unexpected type for @odata.nextLink in response at URL={url}"),
-                            ErrorCode::InternalError
-                        )
-                    })?
-                    .to_string();
-            } else if let Some(delta_link) = value.get("@odata.deltaLink") {
-                //delta处理完了拿到新的delta link，持久化
-                return Ok(delta_link
-                    .as_str()
-                    .ok_or_else(|| {
-                        raise_error!(
-                            format!(
-                                "unexpected type for @odata.deltaLink in response at URL={url}"
-                            ),
-                            ErrorCode::InternalError
-                        )
-                    })?
-                    .to_string());
-            } else {
-                return Err(raise_error!(format!(
-                    "neither @odata.nextLink nor @odata.deltaLink found in Graph API response at URL={url}"
-                ), ErrorCode::InternalError));
-            }
-        }
+        let value = client.get(url.as_str(), &access_token).await?;
+        let message = serde_json::from_value::<Message>(value).map_err(|e| {
+            raise_error!(
+                format!(
+                    "Failed to deserialize Graph API response into MessageListResponse: {:#?}. Possible model mismatch or API change.",
+                    e
+                ),
+                ErrorCode::InternalError
+            )
+        })?;
+        Ok(message)
     }
 }

src/modules/cache/vendor/outlook/sync/delta.rs

@@ -1,3 +1,4 @@
+use ahash::AHashSet;
 use itertools::Itertools;
 use native_db::*;
 use native_model::{native_model, Model};
@@ -6,11 +7,17 @@ use serde::{Deserialize, Serialize};
 
 use crate::{
     modules::{
+        account::migration::AccountModel,
+        cache::vendor::outlook::{
+            model::DeltaResponse,
+            sync::{client::OutlookClient, envelope::OutlookEnvelope, folders::OutlookFolder},
+        },
         database::{
             async_find_impl, batch_delete_impl, delete_impl, filter_by_secondary_key_impl,
             manager::DB_MANAGER, upsert_impl,
         },
         error::{code::ErrorCode, RustMailerResult},
+        hook::http::HttpClient,
         utils::mailbox_id,
     },
     raise_error, utc_now,
@@ -92,3 +99,72 @@ impl FolderDeltaLink {
         Ok(())
     }
 }
+
+pub async fn handle_delta(
+    account: &AccountModel,
+    local_folders: &[OutlookFolder],
+    remote_folders: &[OutlookFolder],
+) -> RustMailerResult<()> {
+    let account_id = account.id;
+    let use_proxy = account.use_proxy.clone();
+    let remote_folders = find_existing_remote_folders(local_folders, remote_folders);
+    for remote in remote_folders {
+        let mut url = FolderDeltaLink::get(account_id, &remote.folder_id)
+            .await?
+            .link;
+        let client = HttpClient::new(use_proxy).await?;
+        let access_token = OutlookClient::get_access_token(account_id).await?;
+        let mut batch = Vec::new();
+        loop {
+            let value = client.get(url.as_str(), &access_token).await?;
+            let resp = serde_json::from_value::<DeltaResponse>(value).map_err(|e| {
+                raise_error!(
+                    format!(
+                        "Failed to deserialize Graph API response into MessageListResponse: {:#?}. Possible model mismatch or API change.",
+                        e
+                    ),
+                    ErrorCode::InternalError
+                )
+            })?;
+            if let Some(items) = resp.value {
+                for item in items {
+                    if item.removed.is_none() {
+                        let message =
+                            OutlookClient::get_message(account_id, use_proxy, &item.id).await?;
+                        let mut envelope: OutlookEnvelope = message.try_into()?;
+                        envelope.account_id = account_id;
+                        envelope.folder_id = remote.id;
+                        envelope.folder_name = remote.name.clone();
+                        batch.push(envelope);
+                    }
+                }
+            }
+            if let Some(next_link) = resp.next_link {
+                url = next_link;
+            } else if let Some(delta_link) = resp.delta_link {
+                let new_delta_link = delta_link;
+                FolderDeltaLink::upsert(account_id, &remote.folder_id, &new_delta_link).await?;
+                break;
+            } else {
+                return Err(raise_error!(format!(
+                    "neither @odata.nextLink nor @odata.deltaLink found in Graph API response at URL={url}"
+                ), ErrorCode::InternalError));
+            }
+        }
+        OutlookEnvelope::save_envelopes(batch).await?;
+        OutlookFolder::upsert(remote).await?;
+    }
+    Ok(())
+}
+
+pub fn find_existing_remote_folders(
+    local_folders: &[OutlookFolder],
+    remote_folders: &[OutlookFolder],
+) -> Vec<OutlookFolder> {
+    let local_ids: AHashSet<_> = local_folders.iter().map(|l| &l.id).collect();
+    remote_folders
+        .iter()
+        .filter(|remote| local_ids.contains(&remote.id))
+        .cloned()
+        .collect()
+}

src/modules/cache/vendor/outlook/sync/envelope.rs

@@ -16,12 +16,16 @@ use crate::{
                 address::AddressEntity,
                 thread::{EmailThread, EmailThreadKey},
             },
+            model::Envelope,
             vendor::outlook::model::{Message, Recipient},
         },
         common::Addr,
-        database::{batch_delete_impl, manager::DB_MANAGER, with_transaction},
+        database::{
+            batch_delete_impl, manager::DB_MANAGER, paginate_secondary_scan_impl,
+            secondary_find_impl, with_transaction,
+        },
         error::{code::ErrorCode, RustMailerError, RustMailerResult},
-        message::attachment,
+        rest::response::DataPage,
         utils::envelope_hash_from_id,
     },
     raise_error,
@@ -59,7 +63,7 @@ pub struct OutlookEnvelope {
     ///
     /// Corresponds to the Microsoft Graph API field `receivedDateTime`.
     /// May be `None` if the value is unavailable.
-    pub internal_date: i64,
+    pub internal_date: Option<i64>,
     /// The estimated size of the email in bytes.
     ///
     /// This is calculated locally as the sum of the email body length in bytes
@@ -133,13 +137,15 @@ pub struct OutlookEnvelope {
     /// Each element is a string representing an Outlook category name. This field
     /// reflects the current categories assigned to the email in Outlook.
     pub categories: Vec<String>,
+
+    pub is_read: bool,
 }
 
 impl OutlookEnvelope {
     pub fn pk(&self) -> String {
         format!(
             "{}_{}",
-            self.internal_date,
+            self.internal_date.unwrap_or_default(),
             envelope_hash_from_id(self.account_id, self.folder_id, &self.id)
         )
     }
@@ -148,6 +154,34 @@ impl OutlookEnvelope {
         envelope_hash_from_id(self.account_id, self.folder_id, &self.id)
     }
 
+    pub async fn exists(&self) -> RustMailerResult<bool> {
+        let target = secondary_find_impl::<OutlookEnvelope>(
+            DB_MANAGER.envelope_db(),
+            OutlookEnvelopeKey::create_envelope_id,
+            self.create_envelope_id(),
+        )
+        .await?;
+        Ok(target.is_some())
+    }
+
+    pub async fn list_messages_in_folder(
+        folder_id: u64,
+        page: u64,
+        page_size: u64,
+        desc: bool,
+    ) -> RustMailerResult<DataPage<OutlookEnvelope>> {
+        paginate_secondary_scan_impl(
+            DB_MANAGER.envelope_db(),
+            Some(page),
+            Some(page_size),
+            Some(desc),
+            OutlookEnvelopeKey::folder_id,
+            folder_id,
+        )
+        .await
+        .map(DataPage::from)
+    }
+
     pub async fn clean_account(account_id: u64) -> RustMailerResult<()> {
         const BATCH_SIZE: usize = 200;
         let mut total_deleted = 0usize;
@@ -182,6 +216,42 @@ impl OutlookEnvelope {
         Ok(())
     }
 
+    pub async fn clean_folder_envelopes(account_id: u64, folder_id: u64) -> RustMailerResult<()> {
+        const BATCH_SIZE: usize = 200;
+        let mut total_deleted = 0usize;
+        let start_time = Instant::now();
+        loop {
+            let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
+                let to_delete: Vec<OutlookEnvelope> = rw
+                    .scan()
+                    .secondary(OutlookEnvelopeKey::folder_id)
+                    .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
+                    .start_with(folder_id)
+                    .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
+                    .filter_map(Result::ok) // filter only Ok values
+                    .filter(|e: &OutlookEnvelope| e.account_id == account_id)
+                    .take(BATCH_SIZE)
+                    .collect();
+                Ok(to_delete)
+            })
+            .await?;
+            total_deleted += deleted;
+            // If this batch is empty, break the loop
+            if deleted == 0 {
+                break;
+            }
+        }
+
+        info!(
+            "Finished deleting outlook envelopes for folder_id={} account_id={} total_deleted={} in {:?}",
+            folder_id,
+            account_id,
+            total_deleted,
+            start_time.elapsed()
+        );
+        Ok(())
+    }
+
     pub async fn save_envelopes(envelopes: Vec<OutlookEnvelope>) -> RustMailerResult<()> {
         with_transaction(DB_MANAGER.envelope_db(), move |rw| {
             for e in envelopes {
@@ -206,7 +276,7 @@ impl OutlookEnvelope {
                         envelope_id,
                         e.account_id,
                         e.folder_id,
-                        Some(e.internal_date),
+                        e.internal_date,
                         e.date,
                     );
                     // --- Thread upsert ---
@@ -240,6 +310,10 @@ impl OutlookEnvelope {
                             raise_error!(format!("{:#?}", err), ErrorCode::InternalError)
                         })?;
                     }
+                } else {
+                    rw.upsert::<OutlookEnvelope>(e.clone()).map_err(|err| {
+                        raise_error!(format!("{:#?}", err), ErrorCode::InternalError)
+                    })?;
                 }
             }
             Ok(())
@@ -252,18 +326,19 @@ impl TryFrom<Message> for OutlookEnvelope {
     type Error = RustMailerError;
 
     fn try_from(msg: Message) -> Result<Self, Self::Error> {
-        fn parse_datetime(dt: &Option<String>) -> RustMailerResult<i64> {
-            if let Some(s) = dt {
-                let parsed: DateTime<Utc> = s.parse().map_err(|e| {
-                    raise_error!(
-                        format!("Invalid datetime {}: {}", s, e),
-                        ErrorCode::InternalError
-                    )
-                })?;
-                Ok(parsed.timestamp_millis())
-            } else {
-                Ok(0)
-            }
+        fn parse_datetime(dt: &Option<String>) -> RustMailerResult<Option<i64>> {
+            dt.as_ref()
+                .map(|s| {
+                    s.parse::<DateTime<Utc>>()
+                        .map(|dt| dt.timestamp_millis())
+                        .map_err(|e| {
+                            raise_error!(
+                                format!("Invalid datetime {}: {}", s, e),
+                                ErrorCode::InternalError
+                            )
+                        })
+                })
+                .transpose()
         }
 
         fn recipient_to_addr(r: &Option<Recipient>) -> Option<Addr> {
@@ -284,7 +359,7 @@ impl TryFrom<Message> for OutlookEnvelope {
             })
         }
         let internal_date = parse_datetime(&msg.received_date_time)?;
-        let date = parse_datetime(&msg.sent_date_time).ok();
+        let date = parse_datetime(&msg.sent_date_time)?;
         let body_len = msg
             .body
             .as_ref()
@@ -312,11 +387,16 @@ impl TryFrom<Message> for OutlookEnvelope {
             size,
             bcc: recipients_to_addrs(&msg.bcc_recipients),
             cc: recipients_to_addrs(&msg.cc_recipients),
-            date: Some(date.unwrap_or(0)),
+            date: date,
             from: recipient_to_addr(&msg.from),
-            in_reply_to: msg.internet_message_id.clone(),
+            in_reply_to: None,
             sender: recipient_to_addr(&msg.sender),
-            message_id: msg.internet_message_id.clone(),
+            message_id: msg.internet_message_id.as_ref().map(|s| {
+                s.strip_prefix('<')
+                    .and_then(|s| s.strip_suffix('>'))
+                    .unwrap_or(s)
+                    .to_string()
+            }),
             subject: msg.subject.clone(),
             thread_id,
             mime_version: None,
@@ -326,6 +406,42 @@ impl TryFrom<Message> for OutlookEnvelope {
             snippet: msg.body_preview.clone(),
             conversation_id: msg.conversation_id.clone(),
             categories: msg.categories.clone().unwrap_or_default(),
+            is_read: msg.is_read.unwrap_or_default(),
         })
     }
 }
+
+impl From<OutlookEnvelope> for Envelope {
+    fn from(value: OutlookEnvelope) -> Self {
+        Self {
+            id: value.id,
+            account_id: value.account_id,
+            mailbox_id: value.folder_id,
+            mailbox_name: value.folder_name,
+            internal_date: value.internal_date,
+            size: value.size,
+            flags: None,
+            flags_hash: None,
+            bcc: value.bcc,
+            cc: value.cc,
+            date: value.date,
+            from: value.from,
+            in_reply_to: value.in_reply_to,
+            sender: value.sender,
+            return_address: None,
+            message_id: value.message_id,
+            subject: value.subject,
+            thread_name: None,
+            thread_id: value.thread_id,
+            mime_version: value.mime_version,
+            references: value.references,
+            reply_to: value.reply_to,
+            to: value.to,
+            attachments: None,
+            body_meta: None,
+            received: None,
+            labels: value.categories,
+            is_read: value.is_read,
+        }
+    }
+}

src/modules/cache/vendor/outlook/sync/flow.rs

@@ -43,7 +43,7 @@ pub async fn fetch_and_save_since_date(
             AccountRunningState::set_initial_current_syncing_folder(
                 account_id,
                 folder.name.clone(),
-                0, //无法知道到底多少条，因为graph api并不返回总数，只能一页页去获取
+                None,
             )
             .await?;
         }
@@ -96,7 +96,7 @@ pub async fn fetch_and_save_full_folder(
         AccountRunningState::set_initial_current_syncing_folder(
             account_id,
             folder.name.clone(),
-            total_batches,
+            Some(total_batches),
         )
         .await?;
     }

src/modules/cache/vendor/outlook/sync/mod.rs

@@ -1,15 +1,18 @@
+use std::time::Instant;
+
+use ahash::AHashSet;
 use tracing::info;
 
 use crate::modules::{
-    account::{migration::AccountModel, status::AccountRunningState},
+    account::{entity::MailerType, migration::AccountModel, status::AccountRunningState},
     cache::{
         imap::{address::AddressEntity, thread::EmailThread},
         sync_type::{determine_sync_type, SyncType},
         vendor::outlook::sync::{
-            delta::FolderDeltaLink,
+            delta::{handle_delta, FolderDeltaLink},
             envelope::OutlookEnvelope,
             folders::OutlookFolder,
-            rebuild::{rebuild_cache, rebuild_cache_since_date},
+            rebuild::{rebuild_cache, rebuild_cache_since_date, rebuild_single_folder_cache},
             sync_folders::get_sync_folders,
         },
     },
@@ -30,11 +33,11 @@ pub mod folders;
 pub mod rebuild;
 pub mod sync_folders;
 pub async fn execute_outlook_sync(account: &AccountModel) -> RustMailerResult<()> {
-    // assert!(
-    //     matches!(account.mailer_type, MailerType::GraphApi),
-    //     "Bug: Unexpected mailer type, expected GraphApi, found: {:?}",
-    //     account.mailer_type
-    // );
+    assert!(
+        matches!(account.mailer_type, MailerType::GraphApi),
+        "Bug: Unexpected mailer type, expected GraphApi, found: {:?}",
+        account.mailer_type
+    );
 
     let sync_type = determine_sync_type(account).await?;
     if matches!(sync_type, SyncType::SkipSync) {
@@ -86,7 +89,34 @@ pub async fn execute_outlook_sync(account: &AccountModel) -> RustMailerResult<()
         }
         return Ok(());
     }
-    todo!()
+
+    handle_delta(account, &local_folders, &remote_folders).await?;
+
+    let deleted_folders = find_deleted_labels(&local_folders, &remote_folders);
+    let missing_folders = find_missing_labels(&local_folders, &remote_folders);
+    if !deleted_folders.is_empty() {
+        info!(
+            "Account {}: Detected {} mailboxes missing from the Graph API server. \
+            Now cleaning up these mailboxes and their associated metadata locally.",
+            account.id,
+            deleted_folders.len()
+        );
+        cleanup_deleted_folders(account, &deleted_folders).await?;
+    }
+
+    if !missing_folders.is_empty() {
+        info!(
+            count = missing_folders.len(),
+            labels = ?missing_folders,
+            "Inserting missing folders into database"
+        );
+        OutlookFolder::batch_insert(&missing_folders).await?;
+        for folder in &missing_folders {
+            rebuild_single_folder_cache(account, folder).await?;
+        }
+    }
+    AccountRunningState::set_incremental_sync_end(account.id).await?;
+    Ok(())
 }
 
 pub async fn should_rebuild_cache(
@@ -117,3 +147,65 @@ pub async fn should_rebuild_cache(
     info!(account_id = account.id, "Cache cleaning completed");
     Ok(true)
 }
+
+pub fn find_deleted_labels(
+    local_folders: &[OutlookFolder],
+    remote_folders: &[OutlookFolder],
+) -> Vec<OutlookFolder> {
+    let remote_ids: AHashSet<_> = remote_folders.iter().map(|l| &l.id).collect();
+
+    local_folders
+        .iter()
+        .filter(|l| !remote_ids.contains(&l.id))
+        .cloned()
+        .collect()
+}
+
+pub fn find_missing_labels(
+    local_folders: &[OutlookFolder],
+    remote_folders: &[OutlookFolder],
+) -> Vec<OutlookFolder> {
+    let local_ids: AHashSet<_> = local_folders.iter().map(|l| &l.id).collect();
+
+    remote_folders
+        .iter()
+        .filter(|l| !local_ids.contains(&l.id))
+        .cloned()
+        .collect()
+}
+
+async fn cleanup_deleted_folders(
+    account: &AccountModel,
+    deleted_folders: &[OutlookFolder],
+) -> RustMailerResult<()> {
+    let start_time = Instant::now();
+    for folder in deleted_folders {
+        OutlookEnvelope::clean_folder_envelopes(account.id, folder.id).await?;
+        AddressEntity::clean_mailbox_envelopes(account.id, folder.id).await?;
+        EmailThread::clean_mailbox_envelopes(account.id, folder.id).await?;
+    }
+    OutlookFolder::batch_delete(deleted_folders.to_vec()).await?;
+    let elapsed_time = start_time.elapsed().as_secs();
+    info!(
+        "Cleanup deleted OutlookFolders completed: {} seconds elapsed.",
+        elapsed_time
+    );
+    Ok(())
+}
+
+async fn cleanup_single_label(
+    account: &AccountModel,
+    folder: &OutlookFolder,
+) -> RustMailerResult<()> {
+    let start_time = Instant::now();
+    OutlookEnvelope::clean_folder_envelopes(account.id, folder.id).await?;
+    AddressEntity::clean_mailbox_envelopes(account.id, folder.id).await?;
+    EmailThread::clean_mailbox_envelopes(account.id, folder.id).await?;
+    OutlookFolder::delete(folder.id).await?;
+    let elapsed_time = start_time.elapsed().as_secs();
+    info!(
+        "Cleanup OutlookFolders completed: {} seconds elapsed.",
+        elapsed_time
+    );
+    Ok(())
+}

src/modules/cache/vendor/outlook/sync/rebuild.rs

@@ -71,7 +71,7 @@ pub async fn rebuild_cache_since_date(
 ) -> RustMailerResult<()> {
     let start_time = Instant::now();
     let mut total_inserted = 0;
-    let date = date_since.since_gmail_date()?;
+    let date = date_since.since_outlook_date()?;
 
     let account_id = account.id;
     let use_proxy = account.use_proxy;
@@ -123,7 +123,7 @@ pub async fn rebuild_single_folder_cache(
     if folder.exists > 0 {
         match &account.date_since {
             Some(date_since) => {
-                let date = date_since.since_gmail_date()?;
+                let date = date_since.since_outlook_date()?;
                 match fetch_and_save_since_date(account, date.as_str(), folder, true).await {
                     Ok(inserted) => {
                         info!(

src/modules/cache/vendor/outlook/sync/sync_folders.rs

@@ -19,7 +19,7 @@ use crate::{
 pub async fn get_sync_folders(account: &AccountModel) -> RustMailerResult<Vec<MailFolder>> {
     let all_mail_folders = OutlookClient::list_mailfolders(account.id, account.use_proxy).await?;
     debug!(
-        "Account {}: Retrieved {} visible labels from Gmail API: {:?}",
+        "Account {}: Retrieved {} visible folders from Graph API: {:?}",
         account.id,
         all_mail_folders.len(),
         all_mail_folders
@@ -42,7 +42,7 @@ pub async fn get_sync_folders(account: &AccountModel) -> RustMailerResult<Vec<Ma
             )
         );
     }
-    //目前邮件夹的变更检测，仍然是根据name来做的，而不是根据id
+
     detect_mailbox_changes(
         account,
         all_mail_folders
@@ -52,16 +52,16 @@ pub async fn get_sync_folders(account: &AccountModel) -> RustMailerResult<Vec<Ma
     )
     .await?;
 
-    //sync_folders stores the mailbox names for IMAP accounts, whereas for Gmail API accounts it stores the label IDs.
     let subscribed = &account.sync_folders;
     debug!(
         "Account {}: Current subscribed sync folders: {:?}",
         account.id, subscribed
     );
     // Filter folders according to the subscription list; matched_folders will not include any folders outside of it.
-    let mut matched_folders: Vec<&MailFolder> = if !subscribed.is_empty() {
+    let mut matched_folders: Vec<MailFolder> = if !subscribed.is_empty() {
         all_mail_folders
-            .iter()
+            .clone()
+            .into_iter()
             .filter(|f| subscribed.contains(&f.id))
             .collect()
     } else {
@@ -77,13 +77,17 @@ pub async fn get_sync_folders(account: &AccountModel) -> RustMailerResult<Vec<Ma
     );
     // If there are no subscriptions, default to the two special folders: inbox and sentitems
     if matched_folders.is_empty() {
+        let inbox = OutlookClient::get_folder(account.id, account.use_proxy, "inbox").await?;
+        let sentitems =
+            OutlookClient::get_folder(account.id, account.use_proxy, "sentitems").await?;
+
         matched_folders = all_mail_folders
-            .iter()
-            .filter(|label| label.display_name == "inbox" || label.display_name == "sentitems")
+            .into_iter()
+            .filter(|folder| folder.id == inbox.id || folder.id == sentitems.id)
             .collect();
 
         debug!(
-            "Account {}: Matched labels after default inbox/sentitems filter: {:?}",
+            "Account {}: Matched folders after default inbox/sentitems filter: {:?}",
             account.id,
             matched_folders
                 .iter()
@@ -103,5 +107,5 @@ pub async fn get_sync_folders(account: &AccountModel) -> RustMailerResult<Vec<Ma
             );
         }
     }
-    Ok(all_mail_folders)
+    Ok(matched_folders)
 }

src/modules/cache/vendor/outlook/tests.rs

@@ -9,7 +9,9 @@ use std::{future::Future, pin::Pin, time::Duration};
 
 use crate::{
     modules::{
-        cache::vendor::outlook::model::{MailFolder, MailFoldersResponse, MessageListResponse},
+        cache::vendor::outlook::model::{
+            MailFolder, MailFoldersResponse, Message, MessageListResponse,
+        },
         common::rustls::RustMailerTls,
         context::Initialize,
         error::{code::ErrorCode, RustMailerResult},
@@ -239,14 +241,7 @@ async fn test9() {
 #[tokio::test]
 async fn fetch_delta() {
     let access_token = access_token().await;
-    let mut url =
-        "https://graph.microsoft.com/v1.0/me/mailFolders/inbox/messages/delta?\
-            $select=id,isRead,conversationId,internetMessageId,from,body,toRecipients,ccRecipients,\
-            bccRecipients,replyTo,sender,subject,receivedDateTime,sentDateTime,isRead,bodyPreview,categories&\
-            $expand=attachments($select=id,name,contentType,size,isInline,microsoft.graph.fileAttachment/contentId)&\
-            $orderBy=receivedDateTime desc
-        "
-            .to_string();
+    let mut url = "https://graph.microsoft.com/v1.0/me/mailFolders/inbox/messages/delta?$select=id&$orderBy=receivedDateTime desc".to_string();
     let client = reqwest::Client::builder()
         .user_agent(rustmailer_version!())
         .timeout(Duration::from_secs(10))
@@ -299,7 +294,7 @@ async fn fetch_delta() {
 async fn fetch_delta2() {
     let access_token = access_token().await;
     let mut url =
-        "https://graph.microsoft.com/v1.0/me/mailFolders('inbox')/messages/delta?$deltatoken=EYiPJwhbWv_RaMa6Pmw5E6U1QisCj5AIWyPQxgvPgSDyuXXU857vfYaf26QiigAqyySdRPjjG994PaAGYFUP21FRTx72KYI0d_u9hmqzjaLodClcNv6iZoz_JWKmce_Lh_14n7M6GlNlrwJLfZk8iKYMu_92_U_tJ0nd-mU1gJkW7-4w6ntpwPdUhYJgrK56-hxHDhKX8Jp7hX1Lzx04UoXkZVp6pTZF4gcWmk9Xhe7aksxG78IAMRuuM8hQiuDX6iZsexO2Mr2uyZ5MdbqCVu0v_4dEfUBcP7n4W05KmmxYzYtloJ23DwLVxOf6COpnESHUuVDrvkIpAvoGWP53vM5vKY4anIjRR8NE0xTEnBJ7xK8cd3G-Iqh5W4ERY-BoXCMwR45HFc7pROR8Uy9eOkybW8gCn4ZjNboykOwv5SWcqBNnpoZWNthOKjfrHs-0JwCKJJhAOauOym_0Eyxx2V-C_9-pt9IYLwZ3EorNBi4ehtpMQecYi0mv0NbELpX36mfTwAQsCEU8lARAX_rU9XqWD2Spfuf9tWb5Dmc1hFhzZi8p6dGTOokLwlovOPhMm1mS_bF-dkhil9YqIqSlTg.ZGam2nknrDorPIfcCajMuRW0Vj_4l3eouiYvVaB48d8"
+        "https://graph.microsoft.com/v1.0/me/mailFolders('inbox')/messages/delta?$deltatoken=gGpE9bWe1qFWcyG3HvbWkkunspLvA4xxmWqkeJWDye811LoJPSEs2ZjUPG0pKudlwejI_rsTDgNJAu_zA2ex-FgAPO2Di5TtbAbBZHRD2W4wUQqFja-VKSwMWTD2WTQ40_86JhGVgDhA_GpxYC-BBtGflJttseEOr6eCZyMVGlI.i7MKxGEvGLJfbZxhsubwycVsTVIyTbv4AIJH78wjsog"
             .to_string();
     let client = reqwest::Client::builder()
         .user_agent(rustmailer_version!())
@@ -389,3 +384,39 @@ async fn list_messages() {
         eprintln!("Error: {} - {:?}", res.status(), res.text().await.unwrap());
     }
 }
+
+#[tokio::test]
+async fn get_message() {
+    let access_token = access_token().await;
+    let mut url =
+        "https://graph.microsoft.com/v1.0/me/messages/AQMkADAwATMwMAItNzE0OC1jZTEzLTAwAi0wMAoARgAAA_KUk7xWPSBEntPHShr61lgHAOo9V4GwHndCjf0x1uoIcwUAAAIBDAAAAOo9V4GwHndCjf0x1uoIcwUAAYiJVH0AAAA=?\
+               $select=id,isRead,conversationId,internetMessageId,from,body,toRecipients,ccRecipients,\
+               bccRecipients,replyTo,sender,subject,receivedDateTime,sentDateTime,isRead,bodyPreview,categories&\
+               $expand=attachments($select=id,name,contentType,size,isInline,microsoft.graph.fileAttachment/contentId)"
+            .to_string();
+    let client = reqwest::Client::builder()
+        .user_agent(rustmailer_version!())
+        .timeout(Duration::from_secs(10))
+        .connect_timeout(Duration::from_secs(10))
+        .proxy(reqwest::Proxy::all("http://127.0.0.1:22307").unwrap())
+        .redirect(reqwest::redirect::Policy::none())
+        .build()
+        .unwrap();
+
+    let mut nextlink_count = 0;
+
+    let res = client
+        .get(&url)
+        .header(AUTHORIZATION, format!("Bearer {}", access_token))
+        .header(CONTENT_TYPE, "application/json")
+        .send()
+        .await
+        .unwrap();
+
+    if res.status().is_success() {
+        let body: Message = res.json().await.unwrap();
+        println!("{:#?}", body);
+    } else {
+        eprintln!("Error: {} - {:?}", res.status(), res.text().await.unwrap());
+    }
+}