feat(outlook): implement retrieve_outlook_raw_email to fetch and save EML via Graph API

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

@@ -18,8 +18,8 @@ use crate::{
                 messages::{FullMessage, MessageList, MessageMeta, PartBody},
             },
         },
+        common::http::HttpClient,
         error::{code::ErrorCode, RustMailerResult},
-        hook::http::HttpClient,
         mailbox::{create::CreateMailboxRequest, rename::MailboxUpdateRequest},
         message::append::ReplyDraft,
         oauth2::token::OAuth2AccessToken,

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

@@ -1,3 +1,4 @@
+use bytes::Bytes;
 use serde_json::json;
 
 use crate::{
@@ -5,8 +6,8 @@ use crate::{
         cache::vendor::outlook::model::{
             MailFolder, MailFoldersResponse, Message, MessageListResponse,
         },
+        common::http::HttpClient,
         error::{code::ErrorCode, RustMailerResult},
-        hook::http::HttpClient,
         message::append::ReplyDraft,
         oauth2::token::OAuth2AccessToken,
     },
@@ -230,6 +231,18 @@ impl OutlookClient {
         Ok(message)
     }
 
+    pub async fn get_raw_message(
+        account_id: u64,
+        use_proxy: Option<u64>,
+        id: &str,
+    ) -> RustMailerResult<Bytes> {
+        let url = format!("https://graph.microsoft.com/v1.0/me/messages/{id}/$value");
+        let client = HttpClient::new(use_proxy).await?;
+        let access_token = Self::get_access_token(account_id).await?;
+        let value = client.get_bytes(url.as_str(), &access_token).await?;
+        Ok(value)
+    }
+
     pub async fn get_attachment(
         account_id: u64,
         use_proxy: Option<u64>,

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

@@ -7,24 +7,17 @@ use serde::{Deserialize, Serialize};
 
 use crate::{
     modules::{
-        account::migration::AccountModel,
-        cache::vendor::outlook::{
+        account::migration::AccountModel, cache::vendor::outlook::{
             model::DeltaResponse,
             sync::{client::OutlookClient, envelope::OutlookEnvelope, folders::OutlookFolder},
-        },
-        database::{
+        }, common::http::HttpClient, database::{
             async_find_impl, batch_delete_impl, filter_by_secondary_key_impl, manager::DB_MANAGER,
             upsert_impl,
-        },
-        error::{code::ErrorCode, RustMailerResult},
-        hook::{
-            channel::{Event, EVENT_CHANNEL},
-            events::{payload::EmailAddedToFolder, EventPayload, EventType, RustMailerEvent},
-            http::HttpClient,
+        }, error::{RustMailerResult, code::ErrorCode}, hook::{
+            channel::{EVENT_CHANNEL, Event},
+            events::{EventPayload, EventType, RustMailerEvent, payload::EmailAddedToFolder},
             task::EventHookTask,
-        },
-        message::content::FullMessageContent,
-        utils::mailbox_id,
+        }, message::content::FullMessageContent, utils::mailbox_id
     },
     raise_error, utc_now,
 };

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

@@ -2,21 +2,18 @@
 // Licensed under RustMailer License Agreement v1.0
 // Unauthorized copying, modification, or distribution is prohibited.
 
-use http::header::CONTENT_TYPE;
+use http::header::{ACCEPT, CONTENT_TYPE};
 use poem_grpc::{ClientConfig, CompressionEncoding};
 use reqwest::{header::AUTHORIZATION, Client};
-use std::{future::Future, pin::Pin, time::Duration};
+use std::{fs::File, future::Future, io::Write, pin::Pin, time::Duration};
 
 use crate::{
     modules::{
-        cache::vendor::outlook::model::{
-            MailFolder, MailFoldersResponse, Message, MessageListResponse,
-        },
+        cache::vendor::outlook::model::{MailFolder, MailFoldersResponse, MessageListResponse},
         common::rustls::RustMailerTls,
         context::Initialize,
         error::{code::ErrorCode, RustMailerResult},
         grpc::service::rustmailer_grpc::{GetOAuth2TokensRequest, OAuth2ServiceClient},
-        hook::http::HttpClient,
     },
     raise_error, rustmailer_version,
 };
@@ -488,3 +485,36 @@ async fn create_reply_draft() {
         eprintln!("Error: {} - {:?}", res.status(), res.text().await.unwrap());
     }
 }
+
+#[tokio::test]
+async fn get_raw_message() {
+    let access_token = access_token().await;
+    let url = format!(
+        "https://graph.microsoft.com/v1.0/me/messages/{}/$value",
+        "AQMkADAwATMwMAItNzE0OC1jZTEzLTAwAi0wMAoARgAAA_KUk7xWPSBEntPHShr61lgHAOo9V4GwHndCjf0x1uoIcwUAAAIBDAAAAOo9V4GwHndCjf0x1uoIcwUAAYiJVIEAAAA="
+    );
+    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 res = client
+        .get(&url)
+        .header(AUTHORIZATION, format!("Bearer {}", access_token))
+        .header(ACCEPT, "application/octet-stream")
+        .send()
+        .await
+        .unwrap();
+
+    if res.status().is_success() {
+        let bytes = res.bytes().await.unwrap();
+        let mut file = File::create("e:\\message.eml").unwrap();
+        file.write_all(&bytes).unwrap();
+    } else {
+        eprintln!("Error: {} - {:?}", res.status(), res.text().await.unwrap());
+    }
+}

src/modules/common/http/mod.rs

@@ -2,8 +2,9 @@
 // Licensed under RustMailer License Agreement v1.0
 // Unauthorized copying, modification, or distribution is prohibited.
 
+use bytes::Bytes;
 use dashmap::DashMap;
-use http::header::{AUTHORIZATION, CONTENT_LENGTH, CONTENT_TYPE};
+use http::header::{ACCEPT, AUTHORIZATION, CONTENT_LENGTH, CONTENT_TYPE};
 use http::StatusCode;
 use serde::Serialize;
 use tracing::error;
@@ -218,6 +219,84 @@ impl HttpClient {
         }
     }
 
+    pub async fn get_bytes(
+        &self,
+        url: &str,
+        access_token: &str,
+    ) -> RustMailerResult<Bytes> {
+        let mut attempt = 0;
+        let max_attempts = 4;
+        let mut delay_ms = 500;
+
+        loop {
+            attempt += 1;
+            let res_result = self
+                .client
+                .get(url)
+                .header(AUTHORIZATION, format!("Bearer {}", access_token))
+                .header(ACCEPT, "application/octet-stream")
+                .send()
+                .await;
+
+            match res_result {
+                Ok(res) => {
+                    if res.status().is_success() {
+                        let bytes = res.bytes().await.map_err(|e| {
+                            raise_error!(
+                                format!("Failed to parse response: {:#?}", e),
+                                ErrorCode::InternalError
+                            )
+                        })?;
+                        return Ok(bytes);
+                    } else {
+                        let status = res.status();
+                        let text = res.text().await.unwrap_or_default();
+
+                        if attempt < max_attempts && status.is_server_error() {
+                            tracing::warn!(
+                                "API call to {} returned server error {} on attempt {}. Retrying after {}ms...",
+                                url, status, attempt, delay_ms
+                            );
+                            tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
+                            delay_ms *= 2;
+                            continue;
+                        }
+
+                        return Err(raise_error!(
+                            format!(
+                                "API call to {} failed with status {}: {}",
+                                url, status, text
+                            ),
+                            ErrorCode::ApiCallFailed
+                        ));
+                    }
+                }
+                Err(e) => {
+                    if attempt < max_attempts {
+                        tracing::warn!(
+                            "Request to {} failed on attempt {}: {:#?}, retrying after {}ms",
+                            url,
+                            attempt,
+                            e,
+                            delay_ms
+                        );
+                        tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
+                        delay_ms *= 2;
+                        continue;
+                    } else {
+                        return Err(raise_error!(
+                            format!(
+                                "Request to {} failed after {} attempts: {:#?}",
+                                url, attempt, e
+                            ),
+                            ErrorCode::ApiCallFailed
+                        ));
+                    }
+                }
+            }
+        }
+    }
+
     /// Wrapper around the Gmail API `POST` request.
     pub async fn post<T: Serialize>(
         &self,

src/modules/common/http/tests.rs

@@ -4,10 +4,7 @@
 
 use std::time::Duration;
 
-use crate::{
-    modules::{error::code::ErrorCode, hook::http::HttpClient},
-    raise_error, rustmailer_version,
-};
+use crate::{modules::{common::http::HttpClient, error::code::ErrorCode}, raise_error, rustmailer_version};
 
 #[tokio::test]
 async fn test_connect_timeout() {
@@ -103,7 +100,7 @@ async fn test_connect_use_proxy() {
     let client = HttpClient::create(client);
     let url = "https://discord.com/api/webhooks/1397150752484622416/9yb6QJSJkszn-uiDge3No3ri9B2-shKMKOT1ruijnPbVtd_k9HAuqspn8C2cOXIqu4l5";
     let payload = json!({
-        "avatar_url": "https://github.com/rustmailer.png", 
+        "avatar_url": "https://github.com/rustmailer.png",
         "content": "hello world",
         "embeds": [
             {
@@ -145,7 +142,7 @@ async fn test_connect_use_proxy() {
     let result = client
         .send_json_request(None, HttpMethod::Post, url, &payload, None)
         .await;
- 
+
     match result {
         Err(e) => {
             let err_str = e.to_string();

src/modules/common/mod.rs

@@ -23,6 +23,7 @@ use tracing::error;
 
 pub mod auth;
 pub mod error;
+pub mod http;
 pub mod log;
 pub mod lru;
 pub mod paginated;

src/modules/hook/mod.rs

@@ -5,7 +5,6 @@
 pub mod channel;
 pub mod entity;
 pub mod events;
-pub mod http;
 pub mod nats;
 pub mod payload;
 pub mod task;

src/modules/hook/task.rs

@@ -5,11 +5,12 @@
 use std::collections::HashMap;
 use std::time::Instant;
 
+use crate::modules::common::http::HttpClient;
 use crate::modules::error::code::ErrorCode;
 use crate::modules::error::RustMailerError;
+use crate::modules::hook::entity::EventHooks;
 use crate::modules::hook::vrl::payload::VrlScriptTestRequest;
 use crate::modules::hook::vrl::resolve_vrl_input;
-use crate::modules::hook::{entity::EventHooks, http::HttpClient};
 use crate::modules::metrics::{
     FAILURE, RUSTMAILER_EVENT_DISPATCH_DURATION_SECONDS_BY_TYPE_STATUS_AND_DESTINATION,
     RUSTMAILER_EVENT_DISPATCH_TOTAL_BY_TYPE_STATUS_AND_DESTINATION, SUCCESS,

src/modules/message/full.rs

@@ -6,7 +6,10 @@ use crate::{
     base64_decode_url_safe,
     modules::{
         account::{entity::MailerType, migration::AccountModel},
-        cache::{disk::DISK_CACHE, vendor::gmail::sync::client::GmailClient},
+        cache::{
+            disk::DISK_CACHE,
+            vendor::{gmail::sync::client::GmailClient, outlook::sync::client::OutlookClient},
+        },
         context::executors::RUST_MAIL_CONTEXT,
         error::{code::ErrorCode, RustMailerResult},
     },
@@ -35,7 +38,11 @@ fn gmail_raw_email_diskcache_key(account_id: u64, mid: &str) -> String {
     format!("gmail_raw_email_{}_{}", account_id, mid)
 }
 
-pub async fn  retrieve_raw_email(
+fn outlook_raw_email_diskcache_key(account_id: u64, mid: &str) -> String {
+    format!("outlook_raw_email_{}_{}", account_id, mid)
+}
+
+pub async fn retrieve_raw_email(
     account_id: u64,
     mailbox: Option<&str>,
     id: &str,
@@ -58,7 +65,7 @@ pub async fn  retrieve_raw_email(
             retrieve_imap_raw_email(account_id, mailbox, uid).await
         }
         MailerType::GmailApi => retrieve_gmail_raw_email(&account, id).await,
-        MailerType::GraphApi => todo!(),
+        MailerType::GraphApi => retrieve_outlook_raw_email(&account, id).await,
     }
 }
 
@@ -148,3 +155,19 @@ async fn retrieve_gmail_raw_email(
         .await?
         .ok_or_else(|| raise_error!("Unexpected cache miss".into(), ErrorCode::InternalError))
 }
+
+async fn retrieve_outlook_raw_email(
+    account: &AccountModel,
+    mid: &str,
+) -> RustMailerResult<cacache::Reader> {
+    let cache_key = outlook_raw_email_diskcache_key(account.id, mid);
+    if let Some(reader) = DISK_CACHE.get_cache(&cache_key).await? {
+        return Ok(reader);
+    }
+    let data = OutlookClient::get_raw_message(account.id, account.use_proxy, mid).await?;
+    DISK_CACHE.put_cache(&cache_key, &data, false).await?;
+    DISK_CACHE
+        .get_cache(&cache_key)
+        .await?
+        .ok_or_else(|| raise_error!("Unexpected cache miss".into(), ErrorCode::InternalError))
+}