From af345b8a958ff07b090bb442408992bbdfd8f148 Mon Sep 17 00:00:00 2001 From: rustmailer Date: Thu, 30 Oct 2025 21:34:07 +0800 Subject: [PATCH] feat: Add EmailAddedToFolder event callback for Graph API accounts --- src/modules/cache/imap/sync/flow.rs | 8 +- .../cache/vendor/gmail/sync/history.rs | 6 +- .../cache/vendor/outlook/sync/delta.rs | 68 ++++++++++- .../cache/vendor/outlook/sync/envelope.rs | 110 +++++++++--------- src/modules/hook/events/mod.rs | 11 +- 5 files changed, 121 insertions(+), 82 deletions(-) diff --git a/src/modules/cache/imap/sync/flow.rs b/src/modules/cache/imap/sync/flow.rs index 89be557..7305461 100644 --- a/src/modules/cache/imap/sync/flow.rs +++ b/src/modules/cache/imap/sync/flow.rs @@ -32,9 +32,7 @@ use crate::{ hook::{ channel::{Event, EVENT_CHANNEL}, events::{ - payload::{ - Attachment, EmailAddedToFolder, EmailBounce, EmailFeedBackReport, MailboxChange, - }, + payload::{EmailAddedToFolder, EmailBounce, EmailFeedBackReport, MailboxChange}, EventPayload, EventType, RustMailerEvent, }, task::EventHookTask, @@ -932,10 +930,6 @@ async fn process_email_added_events( message: message_content, thread_name: envelope.thread_name, reply_to: envelope.reply_to, - attachments: envelope - .attachments - .as_ref() - .map(|atts| atts.iter().cloned().map(Attachment::from).collect()), thread_id, labels: vec![], }), diff --git a/src/modules/cache/vendor/gmail/sync/history.rs b/src/modules/cache/vendor/gmail/sync/history.rs index 27ab7e8..b1a0975 100644 --- a/src/modules/cache/vendor/gmail/sync/history.rs +++ b/src/modules/cache/vendor/gmail/sync/history.rs @@ -24,7 +24,7 @@ use crate::{ hook::{ channel::{Event, EVENT_CHANNEL}, events::{ - payload::{Attachment, EmailAddedToFolder, EmailFlagsChanged}, + payload::{EmailAddedToFolder, EmailFlagsChanged}, EventPayload, EventType, RustMailerEvent, }, task::EventHookTask, @@ -342,10 +342,6 @@ async fn dispatch_new_email_notification( message: message_content, thread_name: envelope.thread_name, reply_to: envelope.reply_to, - attachments: envelope - .attachments - .as_ref() - .map(|atts| atts.iter().cloned().map(Attachment::from).collect()), thread_id: envelope.thread_id, labels: envelope.labels, }), diff --git a/src/modules/cache/vendor/outlook/sync/delta.rs b/src/modules/cache/vendor/outlook/sync/delta.rs index a8225a8..14f8d49 100644 --- a/src/modules/cache/vendor/outlook/sync/delta.rs +++ b/src/modules/cache/vendor/outlook/sync/delta.rs @@ -17,7 +17,13 @@ use crate::{ upsert_impl, }, error::{code::ErrorCode, RustMailerResult}, - hook::http::HttpClient, + hook::{ + channel::{Event, EVENT_CHANNEL}, + events::{payload::EmailAddedToFolder, EventPayload, EventType, RustMailerEvent}, + http::HttpClient, + task::EventHookTask, + }, + message::content::FullMessageContent, utils::mailbox_id, }, raise_error, utc_now, @@ -114,7 +120,9 @@ pub async fn handle_delta( .link; let client = HttpClient::new(use_proxy).await?; let access_token = OutlookClient::get_access_token(account_id).await?; - let mut batch = Vec::new(); + //This includes both new and modified emails. For modified emails, a local comparison is needed to determine what has changed. + let mut updated = Vec::new(); + let mut added = Vec::new(); loop { let value = client.get(url.as_str(), &access_token).await?; let resp = serde_json::from_value::(value).map_err(|e| { @@ -128,14 +136,20 @@ pub async fn handle_delta( })?; if let Some(items) = resp.value { for item in items { + //The deletion scenario will not be handled for now. if item.removed.is_none() { let message = OutlookClient::get_message(account_id, use_proxy, &item.id).await?; + let full_message: FullMessageContent = message.clone().try_into()?; 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 envelope.exists().await? { + updated.push(envelope); + } else { + added.push((envelope, full_message)); + } } } } @@ -151,7 +165,9 @@ pub async fn handle_delta( ), ErrorCode::InternalError)); } } - OutlookEnvelope::save_envelopes(batch).await?; + notify_outlook_envelopes(&account, &added).await?; + OutlookEnvelope::save_envelopes(added.into_iter().map(|t| t.0).collect()).await?; + OutlookEnvelope::update_envelopes(updated).await?; OutlookFolder::upsert(remote).await?; } Ok(()) @@ -168,3 +184,47 @@ pub fn find_existing_remote_folders( .cloned() .collect() } + +pub async fn notify_outlook_envelopes( + account: &AccountModel, + envelopes: &[(OutlookEnvelope, FullMessageContent)], +) -> RustMailerResult<()> { + let account_id = account.id; + if EventHookTask::is_watching_email_add_event(account_id).await? { + for message in envelopes { + EVENT_CHANNEL + .queue(Event::new( + account_id, + &account.email, + RustMailerEvent::new( + EventType::EmailAddedToFolder, + EventPayload::EmailAddedToFolder(EmailAddedToFolder { + account_id: account.id, + account_email: account.email.clone(), + mailbox_name: message.0.folder_name.clone(), + id: message.0.id.clone(), + internal_date: message.0.internal_date, + date: message.0.date, + from: message.0.from.clone(), + subject: message.0.subject.clone(), + to: message.0.to.clone(), + size: message.0.size, + flags: vec![], + cc: message.0.cc.clone(), + bcc: message.0.bcc.clone(), + in_reply_to: message.0.in_reply_to.clone(), + sender: message.0.sender.clone(), + message_id: message.0.message_id.clone(), + message: message.1.clone(), + thread_name: None, + reply_to: message.0.reply_to.clone(), + thread_id: message.0.thread_id, + labels: message.0.categories.clone(), + }), + ), + )) + .await; + } + } + Ok(()) +} diff --git a/src/modules/cache/vendor/outlook/sync/envelope.rs b/src/modules/cache/vendor/outlook/sync/envelope.rs index 1e0a899..63d3cc3 100644 --- a/src/modules/cache/vendor/outlook/sync/envelope.rs +++ b/src/modules/cache/vendor/outlook/sync/envelope.rs @@ -21,7 +21,8 @@ use crate::{ }, common::Addr, database::{ - batch_delete_impl, manager::DB_MANAGER, paginate_secondary_scan_impl, with_transaction, + batch_delete_impl, manager::DB_MANAGER, paginate_secondary_scan_impl, + secondary_find_impl, with_transaction, }, error::{code::ErrorCode, RustMailerError, RustMailerResult}, rest::response::DataPage, @@ -153,15 +154,15 @@ impl OutlookEnvelope { envelope_hash_from_id(self.account_id, self.folder_id, &self.id) } - // pub async fn exists(&self) -> RustMailerResult { - // let target = secondary_find_impl::( - // DB_MANAGER.envelope_db(), - // OutlookEnvelopeKey::create_envelope_id, - // self.create_envelope_id(), - // ) - // .await?; - // Ok(target.is_some()) - // } + pub async fn exists(&self) -> RustMailerResult { + let target = secondary_find_impl::( + 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, @@ -255,62 +256,46 @@ impl OutlookEnvelope { with_transaction(DB_MANAGER.envelope_db(), move |rw| { for e in envelopes { let envelope_id = e.create_envelope_id(); - // Idempotent write - if rw - .get() - .secondary::( - OutlookEnvelopeKey::create_envelope_id, - envelope_id, - ) - .map_err(|err| raise_error!(format!("{:#?}", err), ErrorCode::InternalError))? - .is_none() - { - rw.insert::(e.clone()).map_err(|err| { - raise_error!(format!("{:#?}", err), ErrorCode::InternalError) - })?; + //new emails added here + rw.insert::(e.clone()) + .map_err(|err| raise_error!(format!("{:#?}", err), ErrorCode::InternalError))?; - let address_entities = AddressEntity::extract3(&e); - let thread = EmailThread::new( - e.thread_id, - envelope_id, - e.account_id, - e.folder_id, - e.internal_date, - e.date, - ); - // --- Thread upsert --- - match rw - .get() - .secondary::(EmailThreadKey::thread_id, thread.thread_id) - .map_err(|err| { - raise_error!(format!("{:#?}", err), ErrorCode::InternalError) - })? { - Some(current) => { - // Only replace if current.internal_date is older than new internal_date - if current.need_update(&thread) { - rw.remove(current).map_err(|err| { - raise_error!(format!("{:#?}", err), ErrorCode::InternalError) - })?; - rw.insert::(thread).map_err(|err| { - raise_error!(format!("{:#?}", err), ErrorCode::InternalError) - })?; - } - } - None => { + let address_entities = AddressEntity::extract3(&e); + let thread = EmailThread::new( + e.thread_id, + envelope_id, + e.account_id, + e.folder_id, + e.internal_date, + e.date, + ); + // --- Thread upsert --- + match rw + .get() + .secondary::(EmailThreadKey::thread_id, thread.thread_id) + .map_err(|err| raise_error!(format!("{:#?}", err), ErrorCode::InternalError))? + { + Some(current) => { + // Only replace if current.internal_date is older than new internal_date + if current.need_update(&thread) { + rw.remove(current).map_err(|err| { + raise_error!(format!("{:#?}", err), ErrorCode::InternalError) + })?; rw.insert::(thread).map_err(|err| { raise_error!(format!("{:#?}", err), ErrorCode::InternalError) })?; } } - - // --- Store address entities --- - for addr in address_entities { - rw.insert::(addr).map_err(|err| { + None => { + rw.insert::(thread).map_err(|err| { raise_error!(format!("{:#?}", err), ErrorCode::InternalError) })?; } - } else { - rw.upsert::(e.clone()).map_err(|err| { + } + + // --- Store address entities --- + for addr in address_entities { + rw.insert::(addr).map_err(|err| { raise_error!(format!("{:#?}", err), ErrorCode::InternalError) })?; } @@ -319,6 +304,17 @@ impl OutlookEnvelope { }) .await } + + pub async fn update_envelopes(envelopes: Vec) -> RustMailerResult<()> { + with_transaction(DB_MANAGER.envelope_db(), move |rw| { + for e in envelopes { + rw.upsert::(e) + .map_err(|err| raise_error!(format!("{:#?}", err), ErrorCode::InternalError))?; + } + Ok(()) + }) + .await + } } impl TryFrom for OutlookEnvelope { diff --git a/src/modules/hook/events/mod.rs b/src/modules/hook/events/mod.rs index 4532926..5a6ee6f 100644 --- a/src/modules/hook/events/mod.rs +++ b/src/modules/hook/events/mod.rs @@ -6,9 +6,8 @@ use core::convert::Into; use std::{collections::HashMap, fmt, sync::LazyLock}; use payload::{ - AccountChange, Attachment, EmailAddedToFolder, EmailBounce, EmailFeedBackReport, - EmailFlagsChanged, EmailSendingError, EmailSentSuccess, MailboxChange, MailboxCreation, - MailboxDeletion, + AccountChange, EmailAddedToFolder, EmailBounce, EmailFeedBackReport, EmailFlagsChanged, + EmailSendingError, EmailSentSuccess, MailboxChange, MailboxCreation, MailboxDeletion, }; use poem_openapi::Enum; use serde::{Deserialize, Serialize}; @@ -189,12 +188,6 @@ impl RustMailerEvent { thread_id: id!(64), reply_to: Some(vec![addr("reply@example.com")]), to: Some(vec![addr("recipient@example.com")]), - attachments: Some(vec![Attachment { - filename: Some("notes.pdf".into()), - inline: false, - size: 1024, - file_type: "application/pdf".into(), - }]), labels: vec![] } );