starred/rustmailer
1
0
Fork 0
mirror of https://github.com/rustmailer/rustmailer.git synced 2026-08-20 16:01:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Add EmailAddedToFolder event callback for Graph API accounts

Browse Source
This commit is contained in:
rustmailer
2025-10-30 21:34:07 +08:00
parent 170b428d36
commit af345b8a95
5 changed files with 121 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/modules/cache/imap/sync/flow.rs
Vendored
+1 -7
View File
@@ -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![],
}),
src/modules/cache/vendor/gmail/sync/history.rs
Vendored
+1 -5
View File
@@ -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,
}),
src/modules/cache/vendor/outlook/sync/delta.rs
Vendored
+64 -4
View File
@@ -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::<DeltaResponse>(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(())
}
src/modules/cache/vendor/outlook/sync/envelope.rs
Vendored
+53 -57
View File
@@ -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<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 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,
@@ -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::<OutlookEnvelope>(
OutlookEnvelopeKey::create_envelope_id,
envelope_id,
)
.map_err(|err| raise_error!(format!("{:#?}", err), ErrorCode::InternalError))?
.is_none()
{
rw.insert::<OutlookEnvelope>(e.clone()).map_err(|err| {
raise_error!(format!("{:#?}", err), ErrorCode::InternalError)
})?;
//new emails added here
rw.insert::<OutlookEnvelope>(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::<EmailThread>(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::<EmailThread>(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::<EmailThread>(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::<EmailThread>(thread).map_err(|err| {
raise_error!(format!("{:#?}", err), ErrorCode::InternalError)
})?;
}
}
// --- Store address entities ---
for addr in address_entities {
rw.insert::<AddressEntity>(addr).map_err(|err| {
None => {
rw.insert::<EmailThread>(thread).map_err(|err| {
raise_error!(format!("{:#?}", err), ErrorCode::InternalError)
})?;
}
} else {
rw.upsert::<OutlookEnvelope>(e.clone()).map_err(|err| {
}
// --- Store address entities ---
for addr in address_entities {
rw.insert::<AddressEntity>(addr).map_err(|err| {
raise_error!(format!("{:#?}", err), ErrorCode::InternalError)
})?;
}
@@ -319,6 +304,17 @@ impl OutlookEnvelope {
})
.await
}
pub async fn update_envelopes(envelopes: Vec<OutlookEnvelope>) -> RustMailerResult<()> {
with_transaction(DB_MANAGER.envelope_db(), move |rw| {
for e in envelopes {
rw.upsert::<OutlookEnvelope>(e)
.map_err(|err| raise_error!(format!("{:#?}", err), ErrorCode::InternalError))?;
}
Ok(())
})
.await
}
}
impl TryFrom<Message> for OutlookEnvelope {
src/modules/hook/events/mod.rs
+2 -9
View File
@@ -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![]
}
);