From 49ba67a7d233eaf60bb5812f1b02259bd5f40fac Mon Sep 17 00:00:00 2001 From: rustmailer Date: Wed, 3 Dec 2025 02:43:02 +0800 Subject: [PATCH] feat(thread): Add support for fetching thread lists and thread messages from remote in Gmail and Graph API --- protos/rustmailer.proto | 10 +- src/modules/account/migration.rs | 2 +- src/modules/cache/imap/sync/flow.rs | 2 +- src/modules/cache/imap/thread.rs | 2 +- src/modules/cache/model.rs | 4 +- src/modules/cache/vendor/gmail/cache.rs | 11 + .../cache/vendor/gmail/model/history.rs | 2 +- src/modules/cache/vendor/gmail/model/mod.rs | 3 +- .../cache/vendor/gmail/model/thread.rs | 168 +++++++ src/modules/cache/vendor/gmail/sync/client.rs | 88 +++- .../cache/vendor/gmail/sync/envelope.rs | 2 +- src/modules/cache/vendor/gmail/sync/flow.rs | 87 +--- .../cache/vendor/gmail/sync/history.rs | 257 ++++++---- src/modules/cache/vendor/gmail/sync/mod.rs | 75 ++- .../cache/vendor/gmail/sync/rebuild.rs | 135 +++--- src/modules/cache/vendor/gmail/tests.rs | 92 +++- .../cache/vendor/outlook/sync/client.rs | 33 ++ .../cache/vendor/outlook/sync/delta.rs | 24 +- .../cache/vendor/outlook/sync/envelope.rs | 2 +- src/modules/cache/vendor/outlook/sync/mod.rs | 18 +- .../cache/vendor/outlook/sync/rebuild.rs | 190 ++++---- src/modules/cache/vendor/outlook/tests.rs | 47 +- src/modules/common/http/mod.rs | 2 +- src/modules/error/code.rs | 2 - src/modules/grpc/error.rs | 1 - src/modules/grpc/service/mailbox/mod.rs | 4 +- src/modules/grpc/service/message/mod.rs | 8 +- src/modules/grpc/tests.rs | 6 +- src/modules/hook/events/mod.rs | 2 +- src/modules/hook/events/payload.rs | 2 +- src/modules/imap/manager.rs | 14 +- src/modules/mailbox/delete.rs | 2 +- src/modules/mailbox/rename.rs | 4 +- src/modules/message/list.rs | 441 +++++++++++++++--- src/modules/message/search/payload.rs | 6 +- src/modules/message/tags.rs | 2 +- src/modules/message/transfer.rs | 2 +- src/modules/rest/api/access_token.rs | 8 - src/modules/rest/api/account.rs | 1 - src/modules/rest/api/auto_config.rs | 1 + src/modules/rest/api/event_hook.rs | 2 - src/modules/rest/api/license.rs | 2 - src/modules/rest/api/mailbox.rs | 14 +- src/modules/rest/api/message.rs | 36 +- src/modules/rest/api/mta.rs | 3 - src/modules/rest/api/oauth2.rs | 9 - src/modules/rest/api/send.rs | 6 - src/modules/rest/api/system.rs | 3 - src/modules/rest/api/templates.rs | 8 - src/modules/smtp/request/mod.rs | 2 +- 50 files changed, 1300 insertions(+), 547 deletions(-) create mode 100644 src/modules/cache/vendor/gmail/model/thread.rs diff --git a/protos/rustmailer.proto b/protos/rustmailer.proto index 2f79349..6e43d34 100644 --- a/protos/rustmailer.proto +++ b/protos/rustmailer.proto @@ -784,7 +784,7 @@ message EmailEnvelope { Received received = 25; // The identifier of the thread this email belongs to. // This is computed based on `in_reply_to` / `references` / `message_id`. - uint64 thread_id = 26; + string thread_id = 26; // A list of labels applied to the message. // Each element is a string representing a Gmail label name (e.g., "INBOX", "UNREAD"). // This field reflects the current labels associated with the email. @@ -1029,11 +1029,12 @@ message ListThreadsRequest { // The name of the mailbox to list messages from. string mailbox_name = 2; // The requested page number (1-based). - uint64 page = 3; + optional string next_page_token = 3; // The number of messages to return per page. uint64 page_size = 4; // If true, results will be returned in descending order. bool desc = 5; + bool remote = 6; } @@ -1042,7 +1043,8 @@ message GetThreadMessagesRequest { // The ID of the account. uint64 account_id = 1; // thread id. - uint64 thread_id = 2; + string thread_id = 2; + optional bool remote = 3; } // PagedMessages represents a paginated list of EmailEnvelope messages. @@ -1167,7 +1169,7 @@ service MessageService { // Lists messages within a mailbox with pagination. rpc ListMessages(ListMessagesRequest) returns (CursorDataPage); // Lists threads within a mailbox with pagination. - rpc ListThreads(ListThreadsRequest) returns (PagedMessages); + rpc ListThreads(ListThreadsRequest) returns (CursorDataPage); // Get thread's envelopes within a mailbox. rpc GetThreadMessages(GetThreadMessagesRequest) returns (EmailEnvelopeList); // Fetches specific content parts (e.g., plain text, HTML) of an email message. diff --git a/src/modules/account/migration.rs b/src/modules/account/migration.rs index 04a8797..c888dc0 100644 --- a/src/modules/account/migration.rs +++ b/src/modules/account/migration.rs @@ -338,7 +338,7 @@ impl AccountV3 { if let Some(_) = &request.sync_folders { if matches!(account.mailer_type, MailerType::GmailApi) { map = Some( - GmailClient::reverse_label_map(account_id, account.use_proxy, true).await?, + GmailClient::for_lookup_label_id(account_id, account.use_proxy, true).await?, ); } } diff --git a/src/modules/cache/imap/sync/flow.rs b/src/modules/cache/imap/sync/flow.rs index 7305461..60ed89f 100644 --- a/src/modules/cache/imap/sync/flow.rs +++ b/src/modules/cache/imap/sync/flow.rs @@ -930,7 +930,7 @@ async fn process_email_added_events( message: message_content, thread_name: envelope.thread_name, reply_to: envelope.reply_to, - thread_id, + thread_id: thread_id.to_string(), labels: vec![], }), ), diff --git a/src/modules/cache/imap/thread.rs b/src/modules/cache/imap/thread.rs index 43036a6..6ec6606 100644 --- a/src/modules/cache/imap/thread.rs +++ b/src/modules/cache/imap/thread.rs @@ -242,7 +242,7 @@ impl EmailThread { let results: RustMailerResult> = join_all(fetch_tasks).await.into_iter().collect(); - let map = GmailClient::label_map(account.id, account.use_proxy).await?; + let map = GmailClient::for_get_label_name(account.id, account.use_proxy).await?; let envelopes = results? .into_iter() .map(|e| e.into_envelope(&map)) diff --git a/src/modules/cache/model.rs b/src/modules/cache/model.rs index 640c618..7c0d247 100644 --- a/src/modules/cache/model.rs +++ b/src/modules/cache/model.rs @@ -70,7 +70,7 @@ pub struct Envelope { pub thread_name: Option, /// The identifier of the thread this email belongs to. /// This is computed based on `in_reply_to` / `references` / `message_id`. - pub thread_id: u64, + pub thread_id: String, /// The MIME version of the email (e.g., "1.0"), if specified. pub mime_version: Option, /// A list of message IDs referenced by this email, used for threading. @@ -162,7 +162,7 @@ impl From for Envelope { message_id: value.message_id, subject: value.subject, thread_name: value.thread_name, - thread_id: value.thread_id, + thread_id: value.thread_id.to_string(), mime_version: value.mime_version, references: value.references, reply_to: value.reply_to, diff --git a/src/modules/cache/vendor/gmail/cache.rs b/src/modules/cache/vendor/gmail/cache.rs index bf1e120..f3d5c52 100644 --- a/src/modules/cache/vendor/gmail/cache.rs +++ b/src/modules/cache/vendor/gmail/cache.rs @@ -7,7 +7,18 @@ use std::time::Duration; use ahash::AHashMap; +use crate::modules::cache::vendor::gmail::model::thread::ThreadMessages; use crate::modules::common::lru::TimedLruCache; pub static GMAIL_LABELS_CACHE: LazyLock>> = LazyLock::new(|| TimedLruCache::new(100, Duration::from_secs(60))); + +// Gmail thread message cache. +// Key format: "{account_id}:{thread_id}" +// Stores the full message list of a thread to avoid repeated Gmail API calls. +pub static GMAIL_THREADS_CACHE: LazyLock> = + LazyLock::new(|| TimedLruCache::new(2000, Duration::from_secs(180))); + +pub fn thread_cache_key(account_id: u64, thread_id: &str) -> String { + format!("{}:{}", account_id, thread_id) +} diff --git a/src/modules/cache/vendor/gmail/model/history.rs b/src/modules/cache/vendor/gmail/model/history.rs index 9efd39c..98a5532 100644 --- a/src/modules/cache/vendor/gmail/model/history.rs +++ b/src/modules/cache/vendor/gmail/model/history.rs @@ -9,7 +9,7 @@ pub struct HistoryList { #[serde(default)] pub history: Vec, #[serde(rename = "historyId")] - pub history_id: String, + pub history_id: String,//This is the current maximum history ID for the entire account. #[serde(rename = "nextPageToken")] #[serde(default, skip_serializing_if = "Option::is_none")] pub next_page_token: Option, diff --git a/src/modules/cache/vendor/gmail/model/mod.rs b/src/modules/cache/vendor/gmail/model/mod.rs index e6aec2b..de49590 100644 --- a/src/modules/cache/vendor/gmail/model/mod.rs +++ b/src/modules/cache/vendor/gmail/model/mod.rs @@ -4,4 +4,5 @@ pub mod history; pub mod labels; -pub mod messages; \ No newline at end of file +pub mod messages; +pub mod thread; \ No newline at end of file diff --git a/src/modules/cache/vendor/gmail/model/thread.rs b/src/modules/cache/vendor/gmail/model/thread.rs new file mode 100644 index 0000000..ebb7551 --- /dev/null +++ b/src/modules/cache/vendor/gmail/model/thread.rs @@ -0,0 +1,168 @@ +use std::sync::Arc; + +use serde::{Deserialize, Serialize}; + +use crate::{ + modules::{ + cache::{ + model::Envelope, + vendor::gmail::{ + cache::{thread_cache_key, GMAIL_THREADS_CACHE}, + model::messages::MessageMeta, + sync::{client::GmailClient, envelope::GmailEnvelope}, + }, + }, + error::{code::ErrorCode, RustMailerResult}, + rest::response::CursorDataPage, + }, + raise_error, +}; + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct ThreadIndex { + #[serde(rename = "historyId")] + pub history_id: String, + pub id: String, + #[serde(default)] + pub snippet: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct ThreadList { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub threads: Option>, + #[serde(rename = "nextPageToken")] + #[serde(default, skip_serializing_if = "Option::is_none")] + pub next_page_token: Option, + #[serde(rename = "resultSizeEstimate")] + #[serde(default, skip_serializing_if = "Option::is_none")] + pub result_size_estimate: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct ThreadMessages { + #[serde(rename = "historyId")] + pub history_id: String, + pub id: String, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub messages: Vec, + ///A short part of the message text. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub snippet: Option, +} + +pub async fn list_threads_impl( + account_id: u64, + use_proxy: Option, + label_id: Option<&str>, + page_token: Option<&str>, + after: Option<&str>, + max_results: u64, +) -> RustMailerResult> { + let list = GmailClient::list_threads_internal( + account_id, + use_proxy, + label_id, + page_token, + after, + max_results, + ) + .await?; + + let threads = match list.threads { + Some(t) => t, + None => { + return Ok(CursorDataPage::new( + None, + Some(max_results), + 0, + None, + vec![], + )) + } + }; + + let label_map = GmailClient::for_get_label_name(account_id, use_proxy).await?; + let mut envelopes = Vec::with_capacity(threads.len()); + for thread in threads { + let thread_id = &thread.id; + + let cache_key = thread_cache_key(account_id, thread_id); + let thread_messages = if let Some(cached) = GMAIL_THREADS_CACHE.get(&cache_key).await { + (*cached).clone() + } else { + let fetched = + GmailClient::get_thread_messages(account_id, use_proxy, thread_id).await?; + GMAIL_THREADS_CACHE + .set(cache_key.clone(), Arc::new(fetched.clone())) + .await; + fetched + }; + + let thread_envelopes: Vec = thread_messages + .messages + .into_iter() + .map(GmailEnvelope::try_from) + .collect::>()?; + + if let Some(mut envelope) = thread_envelopes + .into_iter() + .max_by_key(|env| env.internal_date) + { + envelope.account_id = account_id; + envelopes.push(envelope.into_envelope(&label_map)); + } else { + tracing::warn!("No valid Gmail message found in thread {}", thread_id); + } + } + + let total = list.result_size_estimate.ok_or_else(|| { + raise_error!( + "Missing 'resultSizeEstimate' in Gmail API response".into(), + ErrorCode::InternalError + ) + })?; + + let total_pages = (total as f64 / max_results as f64).ceil() as u64; + Ok(CursorDataPage::new( + list.next_page_token, + Some(max_results), + total, + Some(total_pages), + envelopes, + )) +} + +pub async fn get_thread_messages_impl( + account_id: u64, + use_proxy: Option, + thread_id: &str, +) -> RustMailerResult> { + let label_map = GmailClient::for_get_label_name(account_id, use_proxy).await?; + + let cache_key = thread_cache_key(account_id, thread_id); + let thread_messages = if let Some(cached) = GMAIL_THREADS_CACHE.get(&cache_key).await { + (*cached).clone() + } else { + let fetched = GmailClient::get_thread_messages(account_id, use_proxy, thread_id).await?; + GMAIL_THREADS_CACHE + .set(cache_key.clone(), Arc::new(fetched.clone())) + .await; + fetched + }; + + let thread_envelopes: Vec = thread_messages + .messages + .into_iter() + .map(GmailEnvelope::try_from) + .collect::>()?; + + let result = thread_envelopes + .into_iter() + .map(|mut e| { + e.account_id = account_id; // Labels are ignored in remote mode; label_id has no effect. + e.into_envelope(&label_map) + }) + .collect(); + Ok(result) +} diff --git a/src/modules/cache/vendor/gmail/sync/client.rs b/src/modules/cache/vendor/gmail/sync/client.rs index a5f5d75..093fde7 100644 --- a/src/modules/cache/vendor/gmail/sync/client.rs +++ b/src/modules/cache/vendor/gmail/sync/client.rs @@ -16,6 +16,7 @@ use crate::{ history::HistoryList, labels::{Label, LabelDetail, LabelList}, messages::{FullMessage, MessageList, MessageMeta, PartBody}, + thread::{ThreadList, ThreadMessages}, }, }, common::http::HttpClient, @@ -56,7 +57,7 @@ impl GmailClient { Ok(list.labels) } - pub async fn label_map( + pub async fn for_get_label_name( account_id: u64, use_proxy: Option, ) -> RustMailerResult>> { @@ -74,7 +75,7 @@ impl GmailClient { Ok(map) } - pub async fn reverse_label_map( + pub async fn for_lookup_label_id( account_id: u64, use_proxy: Option, skip_cache: bool, @@ -202,16 +203,20 @@ impl GmailClient { pub async fn list_messages( account_id: u64, use_proxy: Option, - label_id: &str, + label_id: Option<&str>, page_token: Option<&str>, after: Option<&str>, max_results: u32, ) -> RustMailerResult { let mut url = format!( - "https://gmail.googleapis.com/gmail/v1/users/me/messages?labelIds={}&maxResults={}", - label_id, max_results + "https://gmail.googleapis.com/gmail/v1/users/me/messages?maxResults={}", + max_results ); + if let Some(label_id) = label_id { + url.push_str(&format!("&labelIds={}", label_id)); + } + if let Some(after) = after { url.push_str(&format!("&q=after:{}", after)); } @@ -235,6 +240,66 @@ impl GmailClient { Ok(list) } + pub async fn list_threads_internal( + account_id: u64, + use_proxy: Option, + label_id: Option<&str>, + page_token: Option<&str>, + after: Option<&str>, + max_results: u64, + ) -> RustMailerResult { + let mut url = format!( + "https://gmail.googleapis.com/gmail/v1/users/me/threads?maxResults={}", + max_results + ); + + if let Some(label_id) = label_id { + url.push_str(&format!("&labelIds={}", label_id)); + } + + if let Some(after) = after { + url.push_str(&format!("&q=after:{}", after)); + } + + if let Some(page_token) = page_token { + url.push_str(&format!("&pageToken={}", page_token)); + } + + let client = HttpClient::new(use_proxy).await?; + let access_token = Self::get_access_token(account_id).await?; + let value = client.get(url.as_str(), &access_token).await?; + let list = serde_json::from_value::(value).map_err(|e| { + raise_error!( + format!( + "Failed to deserialize Gmail API response into ThreadList: {:#?}. Possible model mismatch or API change.", + e + ), + ErrorCode::InternalError + ) + })?; + Ok(list) + } + + pub async fn get_thread_messages( + account_id: u64, + use_proxy: Option, + thread_id: &str, + ) -> RustMailerResult { + let url = format!( + "https://gmail.googleapis.com/gmail/v1/users/me/threads/{}?format=metadata&metadataHeaders=Message-ID&metadataHeaders=From&metadataHeaders=To&metadataHeaders=Cc&metadataHeaders=Bcc&metadataHeaders=Subject&metadataHeaders=Date&metadataHeaders=Mime-Version&metadataHeaders=Reply-To&metadataHeaders=In-Reply-To&metadataHeaders=References&metadataHeaders=Sender", + thread_id, + ); + let client = HttpClient::new(use_proxy).await?; + let access_token = Self::get_access_token(account_id).await?; + let value = client.get(url.as_str(), &access_token).await?; + let messages = serde_json::from_value::(value) + .map_err(|e| raise_error!(format!( + "Failed to deserialize Gmail API response into MessageMeta: {:#?}. Possible model mismatch or API change.", + e + ), ErrorCode::InternalError))?; + Ok(messages) + } + pub async fn search_messages( account_id: u64, use_proxy: Option, @@ -381,15 +446,16 @@ impl GmailClient { pub async fn list_history( account_id: u64, use_proxy: Option, - label_id: &str, + label_id: Option<&str>, start_history_id: &str, page_token: Option<&str>, max_results: u32, ) -> RustMailerResult { - let mut url = format!( - "https://gmail.googleapis.com/gmail/v1/users/me/history?labelId={}&maxResults={}&startHistoryId={}", - label_id, max_results, start_history_id - ); + let mut url = format!("https://gmail.googleapis.com/gmail/v1/users/me/history?maxResults={}&startHistoryId={}", max_results, start_history_id); + + if let Some(label_id) = label_id { + url.push_str(&format!("&labelId={}", label_id)); + } if let Some(page_token) = page_token { url.push_str(&format!("&pageToken={}", page_token)); @@ -425,7 +491,7 @@ impl GmailClient { ErrorCode::InternalError ) })?; - let map = Self::label_map(account_id, use_proxy).await?; + let map = Self::for_get_label_name(account_id, use_proxy).await?; let name = map.get("DRAFT").ok_or_else(|| { raise_error!( "Cannot find 'DRAFT' label in Gmail account label map".into(), diff --git a/src/modules/cache/vendor/gmail/sync/envelope.rs b/src/modules/cache/vendor/gmail/sync/envelope.rs index 1aa5e53..1b01995 100644 --- a/src/modules/cache/vendor/gmail/sync/envelope.rs +++ b/src/modules/cache/vendor/gmail/sync/envelope.rs @@ -425,7 +425,7 @@ impl GmailEnvelope { message_id: self.message_id, subject: self.subject, thread_name: None, - thread_id: self.thread_id, + thread_id: self.gmail_thread_id, mime_version: self.mime_version, references: self.references, reply_to: self.reply_to, diff --git a/src/modules/cache/vendor/gmail/sync/flow.rs b/src/modules/cache/vendor/gmail/sync/flow.rs index a98d360..89b664e 100644 --- a/src/modules/cache/vendor/gmail/sync/flow.rs +++ b/src/modules/cache/vendor/gmail/sync/flow.rs @@ -27,7 +27,7 @@ pub async fn fetch_and_save_since_date( date: &str, label: &GmailLabels, initial: bool, -) -> RustMailerResult<(usize, Option)> { +) -> RustMailerResult { // let total_batches = total.div_ceil(page_size); // Calculate total number of batches, useful for tracking sync progress on UI let mut inserted_count = 0; let account_id = account.id; @@ -37,14 +37,13 @@ pub async fn fetch_and_save_since_date( let mut page_token: Option = None; let mut page = 1; // Used only for tracking sync progress // let semaphore = Arc::new(Semaphore::new(1)); - let mut history_ids = Vec::new(); let mut page_size = ENVELOPE_BATCH_SIZE; let mut total_to_fetch = 100; loop { let resp = GmailClient::list_messages( account_id, use_proxy, - &label.label_id, + Some(&label.label_id), page_token.as_deref(), Some(date), page_size, @@ -129,10 +128,6 @@ pub async fn fetch_and_save_since_date( }) .collect::>>()?; inserted_count += envelopes.len(); - let hid = compute_max_history_id(&envelopes); - if let Some(hid) = hid { - history_ids.push(hid.to_string()); - } GmailEnvelope::save_envelopes(envelopes).await?; } // Break if API response has no next page @@ -145,8 +140,7 @@ pub async fn fetch_and_save_since_date( } page += 1; } - let hid = max_history_id(&history_ids).map(|s| s.to_string()); - Ok((inserted_count, hid)) + Ok(inserted_count) } pub async fn fetch_and_save_full_label( @@ -154,7 +148,7 @@ pub async fn fetch_and_save_full_label( label: &GmailLabels, total: u32, initial: bool, -) -> RustMailerResult<(usize, Option)> { +) -> RustMailerResult { let folder_limit = account.folder_limit; let total_to_fetch = match folder_limit { Some(limit) if limit < total => total.min(limit.max(100)), @@ -184,7 +178,6 @@ pub async fn fetch_and_save_full_label( let mut page_token: Option = None; let mut page = 1; // Used only for tracking sync progress // let semaphore = Arc::new(Semaphore::new(1)); - let mut history_ids = Vec::new(); loop { // Stop if we have already fetched enough messages if inserted_count as u32 >= total_to_fetch { @@ -196,7 +189,7 @@ pub async fn fetch_and_save_full_label( let resp = GmailClient::list_messages( account_id, use_proxy, - &label.label_id, + Some(&label.label_id), page_token.as_deref(), None, page_size, @@ -250,10 +243,6 @@ pub async fn fetch_and_save_full_label( }) .collect::>>()?; inserted_count += envelopes.len(); - let hid = compute_max_history_id(&envelopes); - if let Some(hid) = hid { - history_ids.push(hid.to_string()); - } GmailEnvelope::save_envelopes(envelopes).await?; } // Break if API response has no next page @@ -262,68 +251,6 @@ pub async fn fetch_and_save_full_label( } page += 1; } - let hid = max_history_id(&history_ids).map(|s| s.to_string()); - Ok((inserted_count, hid)) -} - -fn max_history_id_fallback<'a>(a: &'a str, b: &'a str) -> &'a str { - match (a.parse::(), b.parse::()) { - (Ok(a_num), Ok(b_num)) => { - if a_num >= b_num { - a - } else { - b - } - } - _ => { - if a.len() > b.len() { - a - } else if b.len() > a.len() { - b - } else if a >= b { - a - } else { - b - } - } - } -} - -pub fn max_history_id(ids: &[String]) -> Option<&str> { - ids.iter() - .map(|s| s.as_str()) - .reduce(|a, b| max_history_id_fallback(a, b)) -} - -fn compute_max_history_id<'a>(envelopes: &'a [GmailEnvelope]) -> Option<&'a str> { - envelopes - .iter() - .map(|e| e.history_id.as_str()) - .fold(None, |max_id, curr| { - Some(match max_id { - Some(m) => max_history_id_fallback(m, curr), - None => curr, - }) - }) -} - -#[cfg(test)] -mod tests { - use crate::modules::cache::vendor::gmail::sync::flow::max_history_id_fallback; - - #[tokio::test] - async fn test1() { - let ids = vec![ - "2671855", "2671863", "2671871", "2671881", "2671891", "2671898", "100865", "81974", - "81967", "2671905", "531772", "531769", "3296", "1385924", - ]; - - let max_id = ids - .iter() - .cloned() - .reduce(|a, b| max_history_id_fallback(a, b)) - .unwrap(); - - assert_eq!(max_id, "2671905"); - } + + Ok(inserted_count) } diff --git a/src/modules/cache/vendor/gmail/sync/history.rs b/src/modules/cache/vendor/gmail/sync/history.rs index b1a0975..da3a53e 100644 --- a/src/modules/cache/vendor/gmail/sync/history.rs +++ b/src/modules/cache/vendor/gmail/sync/history.rs @@ -15,7 +15,6 @@ use crate::{ cleanup_single_label, client::GmailClient, envelope::GmailEnvelope, - flow::max_history_id, labels::{GmailCheckPoint, GmailLabels}, rebuild::rebuild_single_label_cache, }, @@ -43,14 +42,15 @@ pub async fn handle_history( let use_proxy = account.use_proxy.clone(); let remote_labels = find_existing_remote_labels(local_labels, remote_labels); let checkpoint = GmailCheckPoint::get(account_id).await?; - let mut history_ids = Vec::with_capacity(remote_labels.len()); + let mut current_max_history_id: Option = None; + for remote in remote_labels { let mut page_token = None; loop { let mut list = match GmailClient::list_history( account_id, use_proxy.clone(), - &remote.label_id, + Some(&remote.label_id), &checkpoint.history_id, page_token.as_deref(), 100, // 100 items per page @@ -64,11 +64,10 @@ pub async fn handle_history( location: _, code, } => { - if code == ErrorCode::GmailApiInvalidHistoryId { - let history_id = handle_invalid_history_id(account, &remote).await?; - if let Some(history_id) = history_id { - history_ids.push(history_id); - } + if code == ErrorCode::ApiCallFailed + && message.contains("Requested entity was not found") + { + handle_invalid_history_id(account, &remote).await?; break; } else { return Err(raise_error!(message, code)); @@ -76,29 +75,59 @@ pub async fn handle_history( } }, }; - page_token = list.next_page_token.take(); - let history_list: Vec = list - .history - .into_iter() - .filter(|h| h.has_changes()) - .collect(); + let history_id = parse_history_id(&list.history_id)?; + page_token = list.next_page_token.take(); + let history_items = list.history; + let history_list = match current_max_history_id { + Some(current) => { + if history_id > current { + history_items + .into_iter() + .filter(|h| { + h.has_changes() + && h.id.parse::().map(|id| id <= current).unwrap_or(false) + }) + .collect() + } else { + history_items + .into_iter() + .filter(|h| h.has_changes()) + .collect() + } + } + None => { + current_max_history_id = Some(history_id); + history_items + .into_iter() + .filter(|h| h.has_changes()) + .collect() + } + }; apply_history(account, &remote, history_list).await?; if page_token.is_none() { - history_ids.push(list.history_id); break; } } GmailLabels::upsert(remote).await?; } - let max = max_history_id(&history_ids); - if let Some(history_id) = max { - let checkpoint = GmailCheckPoint::new(account_id, history_id.to_string()); - checkpoint.save().await?; + if let Some(current_max_history_id) = current_max_history_id { + GmailCheckPoint::new(account_id, current_max_history_id.to_string()) + .save() + .await?; } Ok(()) } +fn parse_history_id(history_id: &str) -> RustMailerResult { + history_id.parse::().map_err(|e| { + raise_error!( + format!("Invalid Gmail historyId '{}': {}", history_id, e), + ErrorCode::InternalError + ) + }) +} + pub fn find_existing_remote_labels( local_labels: &[GmailLabels], remote_labels: &[GmailLabels], @@ -127,79 +156,126 @@ pub async fn apply_history( let mut label_changes: AHashMap = AHashMap::new(); // -- labels_added for item in history.labels_added { - let current = - GmailEnvelope::find(account.id, label.id, item.message.id.as_str()).await?; - match current { - Some(mut current) => { - let mut merged: HashSet = current.label_ids.into_iter().collect(); - let mut actually_added = Vec::new(); - if let Some(to_add) = &item.label_ids { - for l in to_add { - if !merged.contains(l) { - actually_added.push(l.clone()); - } - } - merged.extend(to_add.iter().cloned()); - } - if !actually_added.is_empty() { - let entry = label_changes - .entry(item.message.id.clone()) - .or_insert_with(|| LabelChange::default()); - entry.added.extend(actually_added); - } - current.label_ids = merged.into_iter().collect(); - GmailEnvelope::upsert(current).await?; + if account.minimal_sync() { + if let Some(to_add) = &item.label_ids { + let entry = label_changes + .entry(item.message.id.clone()) + .or_insert_with(LabelChange::default); + entry.added.extend(to_add.iter().cloned()); } - None => { - warn!( - "Message {} not found in local cache, cannot merge labels.", - item.message.id - ); + } else { + let current = + GmailEnvelope::find(account.id, label.id, item.message.id.as_str()).await?; + match current { + Some(mut current) => { + let mut merged: HashSet = current.label_ids.into_iter().collect(); + let mut actually_added = Vec::new(); + if let Some(to_add) = &item.label_ids { + for l in to_add { + if !merged.contains(l) { + actually_added.push(l.clone()); + } + } + merged.extend(to_add.iter().cloned()); + } + if !actually_added.is_empty() { + let entry = label_changes + .entry(item.message.id.clone()) + .or_insert_with(|| LabelChange::default()); + entry.added.extend(actually_added); + } + current.label_ids = merged.into_iter().collect(); + GmailEnvelope::upsert(current).await?; + } + None => { + warn!( + "Message {} not found in local cache, cannot merge labels.", + item.message.id + ); + } } } } // -- labels_removed for item in history.labels_removed { - let current = - GmailEnvelope::find(account.id, label.id, item.message.id.as_str()).await?; - match current { - Some(mut current) => { - if let Some(to_remove) = &item.label_ids { - let mut actually_removed = Vec::new(); - if to_remove.contains(&label.id.to_string()) { - GmailEnvelope::delete(account.id, label.id, ¤t.id).await?; - } else { - current.label_ids.retain(|id| { - let keep = !to_remove.contains(id); - if !keep { - actually_removed.push(id.clone()); - } - keep - }); - GmailEnvelope::upsert(current).await?; - } - if !actually_removed.is_empty() { - let entry = label_changes - .entry(item.message.id.clone()) - .or_insert_with(|| LabelChange::default()); - entry.removed.extend(actually_removed); + if account.minimal_sync() { + if let Some(to_remove) = &item.label_ids { + let entry = label_changes + .entry(item.message.id.clone()) + .or_insert_with(LabelChange::default); + entry.removed.extend(to_remove.iter().cloned()); + } + } else { + let current = + GmailEnvelope::find(account.id, label.id, item.message.id.as_str()).await?; + match current { + Some(mut current) => { + if let Some(to_remove) = &item.label_ids { + let mut actually_removed = Vec::new(); + if to_remove.contains(&label.id.to_string()) { + GmailEnvelope::delete(account.id, label.id, ¤t.id).await?; + } else { + current.label_ids.retain(|id| { + let keep = !to_remove.contains(id); + if !keep { + actually_removed.push(id.clone()); + } + keep + }); + GmailEnvelope::upsert(current).await?; + } + if !actually_removed.is_empty() { + let entry = label_changes + .entry(item.message.id.clone()) + .or_insert_with(|| LabelChange::default()); + entry.removed.extend(actually_removed); + } } } - } - None => { - warn!( - "Message {} not found in local cache, cannot merge labels.", - item.message.id - ); + None => { + warn!( + "Message {} not found in local cache, cannot merge labels.", + item.message.id + ); + } } } } - if !label_changes.is_empty() { - if !account.minimal_sync() - && EventHookTask::is_watching_email_flags_changed(account.id).await? - { - for entry in label_changes { + if !label_changes.is_empty() + && EventHookTask::is_watching_email_flags_changed(account.id).await? + { + for entry in label_changes { + if account.minimal_sync() { + let message_data = + GmailClient::get_message(account.id, account.use_proxy, entry.0.as_str()) + .await?; + let current: GmailEnvelope = message_data.try_into()?; + EVENT_CHANNEL + .queue(Event::new( + account.id, + &account.email, + RustMailerEvent::new( + EventType::EmailFlagsChanged, + EventPayload::EmailFlagsChanged(EmailFlagsChanged { + account_id: account.id, + account_email: account.email.clone(), + mailbox_name: label.name.clone(), + uid: None, + from: current.from, + to: current.to, + message_id: current.message_id, + subject: current.subject, + internal_date: Some(current.internal_date), + date: current.date, + flags_added: entry.1.added, + flags_removed: entry.1.removed, + mid: Some(entry.0), + }), + ), + )) + .await; + } else { if let Some(current) = GmailEnvelope::find(account.id, label.id, entry.0.as_str()).await? { @@ -280,7 +356,9 @@ pub async fn apply_history( messages_added.len(), &label.name ); - GmailEnvelope::save_envelopes(messages_added.clone()).await?; + if !account.minimal_sync() { + GmailEnvelope::save_envelopes(messages_added.clone()).await?; + } if EventHookTask::is_watching_email_add_event(account.id).await? { dispatch_new_email_notification(account, messages_added).await?; } @@ -306,13 +384,19 @@ async fn dispatch_new_email_notification( account: &AccountModel, messages: Vec, ) -> RustMailerResult<()> { - let label_map = GmailClient::label_map(account.id, account.use_proxy).await?; + let label_map = GmailClient::for_get_label_name(account.id, account.use_proxy).await?; for message in messages { let full_message = GmailClient::get_full_messages(account.id, account.use_proxy, &message.id).await?; + let gmail_thread_id = full_message.thread_id.clone(); let message_content: FullMessageContent = full_message.try_into()?; let mut envelope = message.into_envelope(&label_map); - envelope.thread_id = envelope.compute_thread_id(); + + envelope.thread_id = if account.minimal_sync() { + gmail_thread_id.unwrap_or_default() + } else { + envelope.compute_thread_id().to_string() + }; EVENT_CHANNEL .queue(Event::new( account.id, @@ -355,7 +439,7 @@ async fn dispatch_new_email_notification( async fn handle_invalid_history_id( account: &AccountModel, label: &GmailLabels, -) -> RustMailerResult> { +) -> RustMailerResult<()> { info!( "Account {}: Invalid history ID detected for label '{}'. Rebuilding local state...", account.id, label.name @@ -370,5 +454,8 @@ async fn handle_invalid_history_id( "Account {}: Upserted label '{}' into local database", account.id, label.name ); - rebuild_single_label_cache(account, label).await + if !account.minimal_sync() { + rebuild_single_label_cache(account, label).await?; + } + Ok(()) } diff --git a/src/modules/cache/vendor/gmail/sync/mod.rs b/src/modules/cache/vendor/gmail/sync/mod.rs index c37359f..c38a993 100644 --- a/src/modules/cache/vendor/gmail/sync/mod.rs +++ b/src/modules/cache/vendor/gmail/sync/mod.rs @@ -18,13 +18,16 @@ use tracing::info; use crate::modules::{ account::{entity::MailerType, migration::AccountModel, status::AccountRunningState}, cache::{ - imap::{address::AddressEntity, thread::EmailThread}, sync_type::{determine_sync_type, SyncType}, vendor::gmail::sync::{ + imap::{address::AddressEntity, thread::EmailThread}, + sync_type::{determine_sync_type, SyncType}, + vendor::gmail::sync::{ + client::GmailClient, envelope::GmailEnvelope, history::handle_history, labels::{GmailCheckPoint, GmailLabels}, rebuild::{rebuild_cache, rebuild_cache_since_date, rebuild_single_label_cache}, sync_labels::get_sync_labels, - } + }, }, error::RustMailerResult, hook::{ @@ -96,6 +99,7 @@ pub async fn execute_gmail_sync(account: &AccountModel) -> RustMailerResult<()> } return Ok(()); } + handle_history(account, &local_labels, &remote_labels).await?; let deleted_labels = find_deleted_labels(&local_labels, &remote_labels); @@ -117,9 +121,11 @@ pub async fn execute_gmail_sync(account: &AccountModel) -> RustMailerResult<()> "Inserting missing Gmail labels into database" ); GmailLabels::batch_insert(&missing_labels).await?; - for label in &missing_labels { - //During incremental synchronization, if any labels are found missing or not fully synchronized, the checkpoint does not need to be updated. - rebuild_single_label_cache(account, label).await?; + if !account.minimal_sync() { + for label in &missing_labels { + //During incremental synchronization, if any labels are found missing or not fully synchronized, the checkpoint does not need to be updated. + rebuild_single_label_cache(account, label).await?; + } } } AccountRunningState::set_incremental_sync_end(account.id).await?; @@ -148,9 +154,12 @@ pub async fn should_rebuild_cache( if checkpoint.is_some() { GmailCheckPoint::clean(account.id).await?; } - GmailEnvelope::clean_account(account.id).await?; - AddressEntity::clean_account(account.id).await?; - EmailThread::clean_account(account.id).await?; + + if !account.minimal_sync() { + GmailEnvelope::clean_account(account.id).await?; + AddressEntity::clean_account(account.id).await?; + EmailThread::clean_account(account.id).await?; + } info!(account_id = account.id, "Cache cleaning completed"); @@ -188,10 +197,13 @@ async fn cleanup_deleted_labels( deleted_labels: &[GmailLabels], ) -> RustMailerResult<()> { let start_time = Instant::now(); - for label in deleted_labels { - GmailEnvelope::clean_label_envelopes(account.id, label.id).await?; - AddressEntity::clean_mailbox_envelopes(account.id, label.id).await?; - EmailThread::clean_mailbox_envelopes(account.id, label.id).await?; + + if !account.minimal_sync() { + for label in deleted_labels { + GmailEnvelope::clean_label_envelopes(account.id, label.id).await?; + AddressEntity::clean_mailbox_envelopes(account.id, label.id).await?; + EmailThread::clean_mailbox_envelopes(account.id, label.id).await?; + } } GmailLabels::batch_delete(deleted_labels.to_vec()).await?; let elapsed_time = start_time.elapsed().as_secs(); @@ -204,9 +216,11 @@ async fn cleanup_deleted_labels( async fn cleanup_single_label(account: &AccountModel, label: &GmailLabels) -> RustMailerResult<()> { let start_time = Instant::now(); - GmailEnvelope::clean_label_envelopes(account.id, label.id).await?; - AddressEntity::clean_mailbox_envelopes(account.id, label.id).await?; - EmailThread::clean_mailbox_envelopes(account.id, label.id).await?; + if !account.minimal_sync() { + GmailEnvelope::clean_label_envelopes(account.id, label.id).await?; + AddressEntity::clean_mailbox_envelopes(account.id, label.id).await?; + EmailThread::clean_mailbox_envelopes(account.id, label.id).await?; + } GmailLabels::delete(label.id).await?; let elapsed_time = start_time.elapsed().as_secs(); info!( @@ -215,3 +229,34 @@ async fn cleanup_single_label(account: &AccountModel, label: &GmailLabels) -> Ru ); Ok(()) } + +async fn save_latest_history_id(account: &AccountModel) -> RustMailerResult<()> { + // Fetch the most recent message + let result = + GmailClient::list_messages(account.id, account.use_proxy, None, None, None, 1).await?; + let first_message = match result.messages.as_ref().and_then(|m| m.first()) { + Some(msg) => msg, + None => return Ok(()), + }; + + // Retrieve the message details to get its history_id + let message = + GmailClient::get_message(account.id, account.use_proxy, &first_message.id).await?; + let start_history_id = &message.history_id; + + // Fetch the latest history starting from the message's history_id + let history_list = GmailClient::list_history( + account.id, + account.use_proxy, + None, + start_history_id, + None, + 1, + ) + .await?; + + // Save the latest history_id to ensure future incremental sync starts from the correct point + GmailCheckPoint::new(account.id, history_list.history_id) + .save() + .await +} diff --git a/src/modules/cache/vendor/gmail/sync/rebuild.rs b/src/modules/cache/vendor/gmail/sync/rebuild.rs index 63bd532..6574556 100644 --- a/src/modules/cache/vendor/gmail/sync/rebuild.rs +++ b/src/modules/cache/vendor/gmail/sync/rebuild.rs @@ -3,10 +3,11 @@ // Unauthorized copying, modification, or distribution is prohibited. use crate::modules::{ - account::{since::DateSince, migration::AccountModel}, + account::{migration::AccountModel, since::DateSince}, cache::vendor::gmail::sync::{ - flow::{fetch_and_save_full_label, fetch_and_save_since_date, max_history_id}, - labels::{GmailCheckPoint, GmailLabels}, + flow::{fetch_and_save_full_label, fetch_and_save_since_date}, + labels::GmailLabels, + save_latest_history_id, }, error::RustMailerResult, }; @@ -19,51 +20,44 @@ pub async fn rebuild_cache( ) -> RustMailerResult<()> { let start_time = Instant::now(); let mut total_inserted = 0; - GmailLabels::batch_insert(remote_labels).await?; - let mut history_ids = Vec::with_capacity(remote_labels.len()); - - for label in remote_labels { - if label.exists == 0 { - info!( + if !account.minimal_sync() { + for label in remote_labels { + if label.exists == 0 { + info!( "Account {}: Label '{}' on the remote server has no emails. Skipping fetch for this label.", account.id, &label.name ); - continue; - } - match fetch_and_save_full_label(account, label, label.exists, true).await { - Ok((inserted, max_history_id)) => { - total_inserted += inserted; - - if let Some(history_id) = max_history_id { - history_ids.push(history_id); - } + continue; } - Err(e) => { - warn!( + match fetch_and_save_full_label(account, label, label.exists, true).await { + Ok(inserted) => { + total_inserted += inserted; + } + Err(e) => { + warn!( "Account {}: Failed to sync label '{}'. Error: {:#?}. Removing label entry.", account.id, &label.name, e ); - if let Err(del_err) = GmailLabels::delete(label.id).await { - error!( - "Account {}: Failed to delete label '{}' after sync error: {}", - account.id, &label.name, del_err - ); + if let Err(del_err) = GmailLabels::delete(label.id).await { + error!( + "Account {}: Failed to delete label '{}' after sync error: {}", + account.id, &label.name, del_err + ); + } } } } } - let max = max_history_id(&history_ids); - if let Some(history_id) = max { - let checkpoint = GmailCheckPoint::new(account.id, history_id.to_string()); - checkpoint.save().await?; - } - let elapsed_time = start_time.elapsed().as_secs(); - info!( - "Rebuild account cache completed: {} envelopes inserted. {} secs elapsed. \ + save_latest_history_id(account).await?; + if !account.minimal_sync() { + let elapsed_time = start_time.elapsed().as_secs(); + info!( + "Rebuild account cache completed: {} envelopes inserted. {} secs elapsed. \ This is a full data fetch as there was no local cache data available.", - total_inserted, elapsed_time - ); + total_inserted, elapsed_time + ); + } Ok(()) } @@ -77,67 +71,66 @@ pub async fn rebuild_cache_since_date( let date = date_since.since_gmail_date()?; GmailLabels::batch_insert(remote_labels).await?; - let mut history_ids = Vec::with_capacity(remote_labels.len()); - for label in remote_labels { - if label.exists == 0 { - info!( + + if !account.minimal_sync() { + for label in remote_labels { + if label.exists == 0 { + info!( "Account {}: Mailbox '{}' on the remote server has no emails. Skipping fetch for this mailbox.", account.id, &label.name ); - continue; - } - - match fetch_and_save_since_date(account, date.as_str(), label, true).await { - Ok((inserted, max_history_id)) => { - total_inserted += inserted; - if let Some(history_id) = max_history_id { - history_ids.push(history_id); - } + continue; } - Err(e) => { - warn!( + + match fetch_and_save_since_date(account, date.as_str(), label, true).await { + Ok(inserted) => { + total_inserted += inserted; + } + Err(e) => { + warn!( "Account {}: Failed to sync mailbox '{}'. Error: {}. Removing mailbox entry.", account.id, &label.name, e ); - if let Err(del_err) = GmailLabels::delete(label.id).await { - error!( - "Account {}: Failed to delete mailbox '{}' after sync error: {}", - account.id, &label.name, del_err - ); + if let Err(del_err) = GmailLabels::delete(label.id).await { + error!( + "Account {}: Failed to delete mailbox '{}' after sync error: {}", + account.id, &label.name, del_err + ); + } } } } } - let max = max_history_id(&history_ids); - if let Some(history_id) = max { - let checkpoint = GmailCheckPoint::new(account.id, history_id.to_string()); - checkpoint.save().await?; - } - let elapsed_time = start_time.elapsed().as_secs(); - info!( - "Rebuild account cache completed: {} envelopes inserted. {} secs elapsed. \ + save_latest_history_id(account).await?; + + if !account.minimal_sync() { + let elapsed_time = start_time.elapsed().as_secs(); + info!( + "Rebuild account cache completed: {} envelopes inserted. {} secs elapsed. \ Data fetched from server starting from the specified date: {}.", - total_inserted, elapsed_time, date - ); + total_inserted, elapsed_time, date + ); + } + Ok(()) } pub async fn rebuild_single_label_cache( account: &AccountModel, label: &GmailLabels, -) -> RustMailerResult> { +) -> RustMailerResult<()> { if label.exists > 0 { match &account.date_since { Some(date_since) => { let date = date_since.since_gmail_date()?; match fetch_and_save_since_date(account, date.as_str(), label, true).await { - Ok((inserted, max_history_id)) => { + Ok(inserted) => { info!( "Account {}: Label '{}' synced successfully. {} messages inserted.", account.id, label.name, inserted ); - return Ok(max_history_id); + return Ok(()); } Err(e) => { warn!( @@ -154,12 +147,12 @@ pub async fn rebuild_single_label_cache( } } None => match fetch_and_save_full_label(account, label, label.exists, true).await { - Ok((inserted, max_history_id)) => { + Ok(inserted) => { info!( "Account {}: Label '{}' synced successfully. {} messages inserted.", account.id, label.name, inserted ); - return Ok(max_history_id); + return Ok(()); } Err(e) => { warn!( @@ -176,5 +169,5 @@ pub async fn rebuild_single_label_cache( }, } } - Ok(None) + Ok(()) } diff --git a/src/modules/cache/vendor/gmail/tests.rs b/src/modules/cache/vendor/gmail/tests.rs index 4fb9f19..152bf62 100644 --- a/src/modules/cache/vendor/gmail/tests.rs +++ b/src/modules/cache/vendor/gmail/tests.rs @@ -21,6 +21,7 @@ use crate::{ model::{ history::HistoryList, messages::{FullMessage, MessageList, MessageMeta, PartBody}, + thread::{ThreadList, ThreadMessages}, }, sync::envelope::GmailEnvelope, }, @@ -44,7 +45,7 @@ async fn access_token() -> String { grpc_client.set_send_compressed(CompressionEncoding::GZIP); let request = GetOAuth2TokensRequest { - account_id: 658031352292246, + account_id: 4727355769996270, }; let mut request = poem_grpc::Request::new(request); @@ -143,13 +144,13 @@ async fn test11() { async fn test2() { let access_token = access_token().await; let url = - "https://gmail.googleapis.com/gmail/v1/users/me/messages?labelIds=INBOX&q=after:2025/08/28&maxResults=20"; + "https://gmail.googleapis.com/gmail/v1/users/me/messages?labelIds=INBOX&q=before:2025/08/28&maxResults=20"; let mut builder = reqwest::ClientBuilder::new() .user_agent(rustmailer_version!()) .timeout(Duration::from_secs(10)) .connect_timeout(Duration::from_secs(10)); - let proxy_obj = reqwest::Proxy::all("socks5://127.0.0.1:22308").unwrap(); + let proxy_obj = reqwest::Proxy::all("http://127.0.0.1:22307").unwrap(); builder = builder .redirect(reqwest::redirect::Policy::none()) .proxy(proxy_obj); @@ -175,13 +176,13 @@ async fn test2() { async fn test3() { let access_token = access_token().await; let url = - "https://gmail.googleapis.com/gmail/v1/users/me/messages/198f6735682a3870?format=metadata&metadataHeaders=Message-Id&metadataHeaders=From&metadataHeaders=To&metadataHeaders=Cc&metadataHeaders=Bcc&metadataHeaders=Subject&metadataHeaders=Date&metadataHeaders=Mime-Version&metadataHeaders=Reply-To&metadataHeaders=In-Reply-To&metadataHeaders=References"; + "https://gmail.googleapis.com/gmail/v1/users/me/messages/19abb5fe57ca4856?format=metadata&metadataHeaders=Message-Id&metadataHeaders=From&metadataHeaders=To&metadataHeaders=Cc&metadataHeaders=Bcc&metadataHeaders=Subject&metadataHeaders=Date&metadataHeaders=Mime-Version&metadataHeaders=Reply-To&metadataHeaders=In-Reply-To&metadataHeaders=References"; let mut builder = reqwest::ClientBuilder::new() .user_agent(rustmailer_version!()) .timeout(Duration::from_secs(10)) .connect_timeout(Duration::from_secs(10)); - let proxy_obj = reqwest::Proxy::all("socks5://127.0.0.1:22308").unwrap(); + let proxy_obj = reqwest::Proxy::all("http://127.0.0.1:22307").unwrap(); builder = builder .redirect(reqwest::redirect::Policy::none()) .proxy(proxy_obj); @@ -210,13 +211,13 @@ async fn test3() { async fn test4() { let access_token = access_token().await; let url = - "https://gmail.googleapis.com/gmail/v1/users/me/history?startHistoryId=42032&labelId=INBOX&maxResults=20"; + "https://gmail.googleapis.com/gmail/v1/users/me/history?startHistoryId=45284&labelId=INBOX&maxResults=2"; let mut builder = reqwest::ClientBuilder::new() .user_agent(rustmailer_version!()) .timeout(Duration::from_secs(10)) .connect_timeout(Duration::from_secs(10)); - let proxy_obj = reqwest::Proxy::all("socks5://127.0.0.1:22308").unwrap(); + let proxy_obj = reqwest::Proxy::all("http://127.0.0.1:22307").unwrap(); builder = builder .redirect(reqwest::redirect::Policy::none()) .proxy(proxy_obj); @@ -236,7 +237,11 @@ async fn test4() { let list: HistoryList = serde_json::from_value(body).unwrap(); println!("Response = {:#?}", list); } else { - eprintln!("Error: {} - {:?}", res.status(), res.text().await.unwrap()); + //eprintln!("Error: {} - {:?}", res.status(), res.text().await.unwrap()); + let text = res.text().await.unwrap(); + if text.contains("Requested entity was not found") { + println!("history id is invalid"); + } } } @@ -428,4 +433,73 @@ async fn test9() { } else { eprintln!("Error: {} - {:?}", res.status(), res.text().await.unwrap()); } -} \ No newline at end of file +} + +#[tokio::test] +async fn list_threads() { + let access_token = access_token().await; + let url = + "https://gmail.googleapis.com/gmail/v1/users/me/threads?labelIds=INBOX&q=after:2025/07/28&maxResults=20"; + let mut builder = reqwest::ClientBuilder::new() + .user_agent(rustmailer_version!()) + .timeout(Duration::from_secs(10)) + .connect_timeout(Duration::from_secs(10)); + + let proxy_obj = reqwest::Proxy::all("http://127.0.0.1:22307").unwrap(); + builder = builder + .redirect(reqwest::redirect::Policy::none()) + .proxy(proxy_obj); + let client = builder.build().unwrap(); + 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: Value = res.json().await.unwrap(); + //println!("Response = {:#?}", body); + let list: ThreadList = serde_json::from_value(body).unwrap(); + println!("Response = {:#?}", list); + } else { + eprintln!("Error: {} - {:?}", res.status(), res.text().await.unwrap()); + } +} + +#[tokio::test] +async fn test13() { + let access_token = access_token().await; + let url = + "https://gmail.googleapis.com/gmail/v1/users/me/threads/19ad6ed3d0afba92?format=metadata&metadataHeaders=Message-Id&metadataHeaders=From&metadataHeaders=To&metadataHeaders=Cc&metadataHeaders=Bcc&metadataHeaders=Subject&metadataHeaders=Date&metadataHeaders=Mime-Version&metadataHeaders=Reply-To&metadataHeaders=In-Reply-To&metadataHeaders=References"; + let mut builder = reqwest::ClientBuilder::new() + .user_agent(rustmailer_version!()) + .timeout(Duration::from_secs(10)) + .connect_timeout(Duration::from_secs(10)); + + let proxy_obj = reqwest::Proxy::all("http://127.0.0.1:22307").unwrap(); + builder = builder + .redirect(reqwest::redirect::Policy::none()) + .proxy(proxy_obj); + let client = builder.build().unwrap(); + 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: Value = res.json().await.unwrap(); + println!("Response = {:#?}", body); + let messages: ThreadMessages = serde_json::from_value(body).unwrap(); + println!("Response = {:#?}", messages); + + // let envelope: GmailEnvelope = detail.try_into().unwrap(); + // println!("Response = {:#?}", envelope); + } else { + eprintln!("Error: {} - {:?}", res.status(), res.text().await.unwrap()); + } +} diff --git a/src/modules/cache/vendor/outlook/sync/client.rs b/src/modules/cache/vendor/outlook/sync/client.rs index 5343053..3612e43 100644 --- a/src/modules/cache/vendor/outlook/sync/client.rs +++ b/src/modules/cache/vendor/outlook/sync/client.rs @@ -182,6 +182,39 @@ impl OutlookClient { Ok(list) } + + pub async fn get_thread_messages( + account_id: u64, + use_proxy: Option, + conversation_id: &str, + ) -> RustMailerResult { + let url = format!( + "https://graph.microsoft.com/v1.0/me/messages?\ + $filter=conversationId eq '{}'&\ + $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)", + conversation_id + ); + + let client = HttpClient::new(use_proxy).await?; + let access_token = Self::get_access_token(account_id).await?; + let value = client.get(url.as_str(), &access_token).await?; + let list = match serde_json::from_value::(value.clone()) { + Ok(res) => res, + Err(e) => { + error!( + "Failed to deserialize Graph API response into MessageListResponse: {:#?}", + e + ); + error!("Original JSON: {}", value); + return Err(raise_error!(format!("Failed to deserialize Graph API response into MessageListResponse: {:#?}. Possible model mismatch or API change.",e),ErrorCode::InternalError)); + } + }; + + Ok(list) + } + pub async fn get_delta_link( account_id: u64, use_proxy: Option, diff --git a/src/modules/cache/vendor/outlook/sync/delta.rs b/src/modules/cache/vendor/outlook/sync/delta.rs index e146cb7..1ee24ee 100644 --- a/src/modules/cache/vendor/outlook/sync/delta.rs +++ b/src/modules/cache/vendor/outlook/sync/delta.rs @@ -155,10 +155,14 @@ pub async fn handle_delta( envelope.account_id = account_id; envelope.folder_id = remote.id; envelope.folder_name = remote.name.clone(); - if envelope.exists().await? { - updated.push(envelope); + if account.minimal_sync() { + added.push((envelope, full_message)); //If the local system does not record the email ID, treat all messages—whether new or updated—as new and send them. } else { - added.push((envelope, full_message)); + if envelope.exists().await? { + updated.push(envelope); + } else { + added.push((envelope, full_message)); + } } } } @@ -176,8 +180,10 @@ pub async fn handle_delta( } } notify_outlook_envelopes(&account, &added).await?; - OutlookEnvelope::save_envelopes(added.into_iter().map(|t| t.0).collect()).await?; - OutlookEnvelope::update_envelopes(updated).await?; + if !account.minimal_sync() { + OutlookEnvelope::save_envelopes(added.into_iter().map(|t| t.0).collect()).await?; + OutlookEnvelope::update_envelopes(updated).await?; + } OutlookFolder::upsert(remote).await?; } Ok(()) @@ -202,6 +208,12 @@ pub async fn notify_outlook_envelopes( let account_id = account.id; if EventHookTask::is_watching_email_add_event(account_id).await? { for message in envelopes { + let thread_id = if account.minimal_sync() { + message.0.conversation_id.clone().unwrap() + } else { + message.0.thread_id.to_string() + }; + EVENT_CHANNEL .queue(Event::new( account_id, @@ -228,7 +240,7 @@ pub async fn notify_outlook_envelopes( message: message.1.clone(), thread_name: None, reply_to: message.0.reply_to.clone(), - thread_id: message.0.thread_id, + thread_id, labels: message.0.categories.clone(), }), ), diff --git a/src/modules/cache/vendor/outlook/sync/envelope.rs b/src/modules/cache/vendor/outlook/sync/envelope.rs index c0f8f8f..fb4d333 100644 --- a/src/modules/cache/vendor/outlook/sync/envelope.rs +++ b/src/modules/cache/vendor/outlook/sync/envelope.rs @@ -463,7 +463,7 @@ impl From for Envelope { message_id: value.message_id, subject: value.subject, thread_name: None, - thread_id: value.thread_id, + thread_id: value.thread_id.to_string(), mime_version: value.mime_version, references: value.references, reply_to: value.reply_to, diff --git a/src/modules/cache/vendor/outlook/sync/mod.rs b/src/modules/cache/vendor/outlook/sync/mod.rs index 51adc3c..17a58f7 100644 --- a/src/modules/cache/vendor/outlook/sync/mod.rs +++ b/src/modules/cache/vendor/outlook/sync/mod.rs @@ -141,9 +141,11 @@ pub async fn should_rebuild_cache( if has_delta { FolderDeltaLink::clean(account.id).await?; } - OutlookEnvelope::clean_account(account.id).await?; - AddressEntity::clean_account(account.id).await?; - EmailThread::clean_account(account.id).await?; + if !account.minimal_sync() { + OutlookEnvelope::clean_account(account.id).await?; + AddressEntity::clean_account(account.id).await?; + EmailThread::clean_account(account.id).await?; + } info!(account_id = account.id, "Cache cleaning completed"); Ok(true) } @@ -179,10 +181,12 @@ async fn cleanup_deleted_folders( 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?; + if !account.minimal_sync() { + 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(); diff --git a/src/modules/cache/vendor/outlook/sync/rebuild.rs b/src/modules/cache/vendor/outlook/sync/rebuild.rs index 50fd31c..c2af1cf 100644 --- a/src/modules/cache/vendor/outlook/sync/rebuild.rs +++ b/src/modules/cache/vendor/outlook/sync/rebuild.rs @@ -26,41 +26,46 @@ pub async fn rebuild_cache( let use_proxy = account.use_proxy; OutlookFolder::batch_insert(remote_folders).await?; for folder in remote_folders { - if folder.exists > 0 { - match fetch_and_save_full_folder(account, folder, folder.exists, true).await { - Ok(inserted) => { - total_inserted += inserted; - } - Err(e) => { - warn!( + if !account.minimal_sync() { + if folder.exists > 0 { + match fetch_and_save_full_folder(account, folder, folder.exists, true).await { + Ok(inserted) => { + total_inserted += inserted; + } + Err(e) => { + warn!( "Account {}: Failed to sync label '{}'. Error: {:#?}. Removing label entry.", account.id, &folder.name, e ); - if let Err(del_err) = OutlookFolder::delete(folder.id).await { - error!( - "Account {}: Failed to delete label '{}' after sync error: {}", - account.id, &folder.name, del_err - ); + if let Err(del_err) = OutlookFolder::delete(folder.id).await { + error!( + "Account {}: Failed to delete label '{}' after sync error: {}", + account.id, &folder.name, del_err + ); + } } } - } - } else { - warn!( + } else { + warn!( "Account {}: folder '{}' on the remote server has no emails. Skipping fetch for this folder.", account.id, &folder.name ); + } } let delta_link = OutlookClient::get_delta_link(account_id, use_proxy, &folder.folder_id).await?; FolderDeltaLink::upsert(account_id, &folder.folder_id, &delta_link).await?; } - let elapsed_time = start_time.elapsed().as_secs(); - info!( - "Rebuild account cache completed: {} envelopes inserted. {} secs elapsed. \ - This is a full data fetch as there was no local cache data available.", - total_inserted, elapsed_time - ); + + if !account.minimal_sync() { + let elapsed_time = start_time.elapsed().as_secs(); + info!( + "Rebuild account cache completed: {} envelopes inserted. {} secs elapsed. \ + This is a full data fetch as there was no local cache data available.", + total_inserted, elapsed_time + ); + } Ok(()) } @@ -78,64 +83,17 @@ pub async fn rebuild_cache_since_date( OutlookFolder::batch_insert(remote_folders).await?; for folder in remote_folders { - if folder.exists > 0 { - match fetch_and_save_since_date(account, date.as_str(), folder, true).await { - Ok(inserted) => { - total_inserted += inserted; - } - Err(e) => { - warn!( - "Account {}: Failed to sync mailbox '{}'. Error: {}. Removing mailbox entry.", - account.id, &folder.name, e - ); - if let Err(del_err) = OutlookFolder::delete(folder.id).await { - error!( - "Account {}: Failed to delete mailbox '{}' after sync error: {}", - account.id, &folder.name, del_err - ); - } - } - } - } else { - warn!( - "Account {}: Mailbox '{}' on the remote server has no emails. Skipping fetch for this mailbox.", - account.id, &folder.name - ); - } - let delta_link = - OutlookClient::get_delta_link(account_id, use_proxy, &folder.folder_id).await?; - FolderDeltaLink::upsert(account_id, &folder.folder_id, &delta_link).await?; - } - let elapsed_time = start_time.elapsed().as_secs(); - info!( - "Rebuild account cache completed: {} envelopes inserted. {} secs elapsed. \ - Data fetched from server starting from the specified date: {}.", - total_inserted, elapsed_time, date - ); - Ok(()) -} - -pub async fn rebuild_single_folder_cache( - account: &AccountModel, - folder: &OutlookFolder, -) -> RustMailerResult<()> { - if folder.exists > 0 { - match &account.date_since { - Some(date_since) => { - let date = date_since.since_outlook_date()?; + if !account.minimal_sync() { + if folder.exists > 0 { match fetch_and_save_since_date(account, date.as_str(), folder, true).await { Ok(inserted) => { - info!( - "Account {}: folder '{}' synced successfully. {} messages inserted.", - account.id, folder.name, inserted - ); - return Ok(()); + total_inserted += inserted; } Err(e) => { warn!( - "Account {}: Failed to sync mailbox '{}'. Error: {}. Removing folder entry.", - account.id, &folder.name, e - ); + "Account {}: Failed to sync mailbox '{}'. Error: {}. Removing mailbox entry.", + account.id, &folder.name, e + ); if let Err(del_err) = OutlookFolder::delete(folder.id).await { error!( "Account {}: Failed to delete mailbox '{}' after sync error: {}", @@ -144,30 +102,86 @@ pub async fn rebuild_single_folder_cache( } } } + } else { + warn!( + "Account {}: Mailbox '{}' on the remote server has no emails. Skipping fetch for this mailbox.", + account.id, &folder.name + ); } - None => match fetch_and_save_full_folder(account, folder, folder.exists, true).await { - Ok(inserted) => { - info!( + } + + let delta_link = + OutlookClient::get_delta_link(account_id, use_proxy, &folder.folder_id).await?; + FolderDeltaLink::upsert(account_id, &folder.folder_id, &delta_link).await?; + } + if !account.minimal_sync() { + let elapsed_time = start_time.elapsed().as_secs(); + info!( + "Rebuild account cache completed: {} envelopes inserted. {} secs elapsed. \ + Data fetched from server starting from the specified date: {}.", + total_inserted, elapsed_time, date + ); + } + Ok(()) +} + +pub async fn rebuild_single_folder_cache( + account: &AccountModel, + folder: &OutlookFolder, +) -> RustMailerResult<()> { + if folder.exists > 0 { + if !account.minimal_sync() { + match &account.date_since { + Some(date_since) => { + let date = date_since.since_outlook_date()?; + match fetch_and_save_since_date(account, date.as_str(), folder, true).await { + Ok(inserted) => { + info!( + "Account {}: folder '{}' synced successfully. {} messages inserted.", + account.id, folder.name, inserted + ); + return Ok(()); + } + Err(e) => { + warn!( + "Account {}: Failed to sync mailbox '{}'. Error: {}. Removing folder entry.", + account.id, &folder.name, e + ); + if let Err(del_err) = OutlookFolder::delete(folder.id).await { + error!( + "Account {}: Failed to delete mailbox '{}' after sync error: {}", + account.id, &folder.name, del_err + ); + } + } + } + } + None => { + match fetch_and_save_full_folder(account, folder, folder.exists, true).await { + Ok(inserted) => { + info!( "Account {}: folder '{}' synced successfully. {} messages inserted.", account.id, folder.name, inserted ); - return Ok(()); - } - Err(e) => { - warn!( + return Ok(()); + } + Err(e) => { + warn!( "Account {}: Failed to sync folder '{}'. Error: {:#?}. Removing folder entry.", account.id, &folder.name, e ); - if let Err(del_err) = OutlookFolder::delete(folder.id).await { - error!( - "Account {}: Failed to delete folder '{}' after sync error: {}", - account.id, &folder.name, del_err - ); + if let Err(del_err) = OutlookFolder::delete(folder.id).await { + error!( + "Account {}: Failed to delete folder '{}' after sync error: {}", + account.id, &folder.name, del_err + ); + } + } } } - }, + } } - //重建完不要忘记将deltalink保存 + let delta_link = OutlookClient::get_delta_link(account.id, account.use_proxy, &folder.folder_id).await?; FolderDeltaLink::upsert(account.id, &folder.folder_id, &delta_link).await?; diff --git a/src/modules/cache/vendor/outlook/tests.rs b/src/modules/cache/vendor/outlook/tests.rs index 3b3ab5e..85bd314 100644 --- a/src/modules/cache/vendor/outlook/tests.rs +++ b/src/modules/cache/vendor/outlook/tests.rs @@ -31,7 +31,7 @@ async fn access_token() -> String { grpc_client.set_send_compressed(CompressionEncoding::GZIP); let request = GetOAuth2TokensRequest { - account_id: 1770162132609912, + account_id: 1409297407117997, }; let mut request = poem_grpc::Request::new(request); @@ -557,9 +557,6 @@ async fn copy_message() { } } - - - #[tokio::test] async fn move_message() { let access_token = access_token().await; @@ -595,4 +592,44 @@ async fn move_message() { } else { eprintln!("Error: {} - {:?}", res.status(), res.text().await.unwrap()); } -} \ No newline at end of file +} + + + +#[tokio::test] +async fn get_thread_messages() { + let access_token = access_token().await; + let mut url = + "https://graph.microsoft.com/v1.0/me/mailFolders/inbox/messages?$top=3&\ + $orderBy=receivedDateTime desc&\ + $filter=receivedDateTime ge 2025-10-01T00:00:00Z&\ + $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: MessageListResponse = res.json().await.unwrap(); + println!("{:#?}", body); + } else { + eprintln!("Error: {} - {:?}", res.status(), res.text().await.unwrap()); + } +} diff --git a/src/modules/common/http/mod.rs b/src/modules/common/http/mod.rs index 9a84c51..e4e54b3 100644 --- a/src/modules/common/http/mod.rs +++ b/src/modules/common/http/mod.rs @@ -170,7 +170,7 @@ impl HttpClient { "API returned client error (status {}) for {}. Response: {}", status, url, text ), - ErrorCode::GmailApiInvalidHistoryId + ErrorCode::ApiCallFailed )); } diff --git a/src/modules/error/code.rs b/src/modules/error/code.rs index 88cc419..9b8c299 100644 --- a/src/modules/error/code.rs +++ b/src/modules/error/code.rs @@ -49,7 +49,6 @@ pub enum ErrorCode { MailBoxNotCached = 50050, AutoconfigFetchFailed = 50060, ApiCallFailed = 50070, - GmailApiInvalidHistoryId = 50080, // Message queue errors (60000–60999) NatsRequestFailed = 60000, @@ -86,7 +85,6 @@ impl ErrorCode { | ErrorCode::AutoconfigFetchFailed | ErrorCode::ImapCommandFailed | ErrorCode::ApiCallFailed - | ErrorCode::GmailApiInvalidHistoryId | ErrorCode::ImapUnexpectedResult | ErrorCode::HttpResponseError | ErrorCode::NatsRequestFailed diff --git a/src/modules/grpc/error.rs b/src/modules/grpc/error.rs index 814697a..9186419 100644 --- a/src/modules/grpc/error.rs +++ b/src/modules/grpc/error.rs @@ -43,7 +43,6 @@ impl From for Status { | ErrorCode::AutoconfigFetchFailed | ErrorCode::ImapCommandFailed | ErrorCode::ApiCallFailed - | ErrorCode::GmailApiInvalidHistoryId | ErrorCode::ImapUnexpectedResult | ErrorCode::HttpResponseError | ErrorCode::NatsRequestFailed diff --git a/src/modules/grpc/service/mailbox/mod.rs b/src/modules/grpc/service/mailbox/mod.rs index 63875bf..315f809 100644 --- a/src/modules/grpc/service/mailbox/mod.rs +++ b/src/modules/grpc/service/mailbox/mod.rs @@ -12,7 +12,7 @@ use crate::modules::mailbox::{ create::create_mailbox, delete::delete_mailbox, list::{get_account_mailboxes, list_subscribed_mailboxes}, - rename::update_mailbox, + rename::update_mailbox_impl, subscribe::{subscribe_mailbox, unsubscribe_mailbox}, }; use poem_grpc::{Request, Response, Status}; @@ -86,7 +86,7 @@ impl MailboxService for RustMailerMailboxService { request: Request, ) -> Result, Status> { let req = require_account_access(request, |r| r.account_id)?; - update_mailbox(req.account_id, req.into()).await?; + update_mailbox_impl(req.account_id, req.into()).await?; Ok(Response::new(Empty::default())) } } diff --git a/src/modules/grpc/service/message/mod.rs b/src/modules/grpc/service/message/mod.rs index 66df1ac..ca175d8 100644 --- a/src/modules/grpc/service/message/mod.rs +++ b/src/modules/grpc/service/message/mod.rs @@ -219,13 +219,14 @@ impl MessageService for RustMailerMessageService { async fn list_threads( &self, request: Request, - ) -> Result, Status> { + ) -> Result, Status> { let req = require_account_access(request, |r| r.account_id)?; let result = list_threads_in_mailbox( req.account_id, &req.mailbox_name, - req.page, + req.next_page_token.as_deref(), req.page_size, + req.remote, req.desc, ) .await?; @@ -238,7 +239,8 @@ impl MessageService for RustMailerMessageService { request: Request, ) -> Result, Status> { let req = require_account_access(request, |r| r.account_id)?; - let envelopes = get_thread_messages(req.account_id, req.thread_id).await?; + let envelopes = + get_thread_messages(req.account_id, req.thread_id, req.remote.unwrap_or(false)).await?; Ok(Response::new(EmailEnvelopeList { items: envelopes.into_iter().map(|e| e.into()).collect(), diff --git a/src/modules/grpc/tests.rs b/src/modules/grpc/tests.rs index 1a0398c..54085b5 100644 --- a/src/modules/grpc/tests.rs +++ b/src/modules/grpc/tests.rs @@ -131,8 +131,9 @@ async fn test4() { let request = ListThreadsRequest { account_id: 8869750310191797, mailbox_name: "INBOX".into(), - page: 1, + next_page_token: Some("1".into()), page_size: 15, + remote: false, desc: true, }; @@ -159,7 +160,8 @@ async fn test5() { let request = GetThreadMessagesRequest { account_id: 6606017263301165, - thread_id: 1572863359614161, + thread_id: "1572863359614161".into(), + remote: None, }; let mut request = poem_grpc::Request::new(request); diff --git a/src/modules/hook/events/mod.rs b/src/modules/hook/events/mod.rs index 5a6ee6f..5040afb 100644 --- a/src/modules/hook/events/mod.rs +++ b/src/modules/hook/events/mod.rs @@ -185,7 +185,7 @@ impl RustMailerEvent { attachments: None }, thread_name: Some("Meeting Thread".into()), - thread_id: id!(64), + thread_id: id!(64).to_string(), reply_to: Some(vec![addr("reply@example.com")]), to: Some(vec![addr("recipient@example.com")]), labels: vec![] diff --git a/src/modules/hook/events/payload.rs b/src/modules/hook/events/payload.rs index afca52e..d69d421 100644 --- a/src/modules/hook/events/payload.rs +++ b/src/modules/hook/events/payload.rs @@ -47,7 +47,7 @@ pub struct EmailAddedToFolder { pub message: FullMessageContent, /// The identifier of the thread this email belongs to. /// This is computed based on `in_reply_to` / `references` / `message_id`. - pub thread_id: u64, + pub thread_id: String, /// Optional name of the thread to which the email belongs. pub thread_name: Option, /// Optional list of reply-to addresses for the email. diff --git a/src/modules/imap/manager.rs b/src/modules/imap/manager.rs index 839563e..7cb8c3a 100644 --- a/src/modules/imap/manager.rs +++ b/src/modules/imap/manager.rs @@ -14,7 +14,7 @@ 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 crate::{decrypt, raise_error, rustmailer_version}; use async_imap::Session; use tracing::error; @@ -128,6 +128,18 @@ impl ImapConnectionManager { .await; return Err(error); } + if capabilities.has_str("ID") || capabilities.has_str("id") { + session + .id([ + ("name", Some("rustmailer")), + ("version", Some(rustmailer_version!())), + ("vendor", Some("rustmailer")), + ]) + .await + .map_err(|e| { + raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed) + })?; + } } Err(error) => { error!("Failed to fetch IMAP capabilities: {:#?}", error); diff --git a/src/modules/mailbox/delete.rs b/src/modules/mailbox/delete.rs index e433a13..7838ae7 100644 --- a/src/modules/mailbox/delete.rs +++ b/src/modules/mailbox/delete.rs @@ -23,7 +23,7 @@ pub async fn delete_mailbox(account_id: u64, mailbox_name: &str) -> RustMailerRe .await } MailerType::GmailApi => { - let map = GmailClient::reverse_label_map(account_id, account.use_proxy, true).await?; + let map = GmailClient::for_lookup_label_id(account_id, account.use_proxy, true).await?; let label_id = map.get(mailbox_name).ok_or_else(|| { raise_error!( format!( diff --git a/src/modules/mailbox/rename.rs b/src/modules/mailbox/rename.rs index c98fdc2..c91ca27 100644 --- a/src/modules/mailbox/rename.rs +++ b/src/modules/mailbox/rename.rs @@ -41,7 +41,7 @@ pub struct MailboxUpdateRequest { pub label_color: Option, } -pub async fn update_mailbox( +pub async fn update_mailbox_impl( account_id: u64, payload: MailboxUpdateRequest, ) -> RustMailerResult<()> { @@ -72,7 +72,7 @@ pub async fn update_mailbox( )); } - let map = GmailClient::reverse_label_map(account_id, account.use_proxy, true).await?; + let map = GmailClient::for_lookup_label_id(account_id, account.use_proxy, true).await?; let label_id = map.get(&payload.current_name).ok_or_else(|| { raise_error!( format!( diff --git a/src/modules/message/list.rs b/src/modules/message/list.rs index f240619..baa0832 100644 --- a/src/modules/message/list.rs +++ b/src/modules/message/list.rs @@ -10,7 +10,10 @@ use crate::{ imap::{mailbox::MailBox, migration::EmailEnvelopeV3, thread::EmailThread}, model::Envelope, vendor::{ - gmail::sync::{client::GmailClient, envelope::GmailEnvelope, labels::GmailLabels}, + gmail::{ + model::thread::{get_thread_messages_impl, list_threads_impl}, + sync::{client::GmailClient, envelope::GmailEnvelope, labels::GmailLabels}, + }, outlook::sync::{ client::OutlookClient, envelope::OutlookEnvelope, folders::OutlookFolder, }, @@ -42,9 +45,9 @@ pub async fn list_messages_in_mailbox( ErrorCode::InvalidParameter )); } - if page_size > 500 { + if page_size > 100 { return Err(raise_error!( - "The page_size exceeds the maximum allowed limit of 500.".into(), + "The page_size exceeds the maximum allowed limit of 100.".into(), ErrorCode::InvalidParameter )); } @@ -56,21 +59,21 @@ pub async fn list_messages_in_mailbox( } } -fn validate_pagination_params(page: u64, page_size: u64) -> RustMailerResult<()> { - if page == 0 || page_size == 0 { - return Err(raise_error!( - "Both page and page_size must be greater than 0.".into(), - ErrorCode::InvalidParameter - )); - } - if page_size > 500 { - return Err(raise_error!( - "The page_size exceeds the maximum allowed limit of 500.".into(), - ErrorCode::InvalidParameter - )); - } - Ok(()) -} +// fn validate_pagination_params(page: u64, page_size: u64) -> RustMailerResult<()> { +// if page == 0 || page_size == 0 { +// return Err(raise_error!( +// "Both page and page_size must be greater than 0.".into(), +// ErrorCode::InvalidParameter +// )); +// } +// if page_size > 500 { +// return Err(raise_error!( +// "The page_size exceeds the maximum allowed limit of 500.".into(), +// ErrorCode::InvalidParameter +// )); +// } +// Ok(()) +// } async fn fetch_remote_messages( account: &AccountModel, @@ -79,6 +82,12 @@ async fn fetch_remote_messages( page_size: u64, desc: bool, ) -> RustMailerResult> { + if page_size > 100 { + return Err(raise_error!( + "The page_size exceeds the maximum allowed limit of 100.".into(), + ErrorCode::InvalidParameter + )); + } match account.mailer_type { MailerType::ImapSmtp => { let page = decode_page_token(next_page_token)?; @@ -125,7 +134,7 @@ async fn fetch_remote_messages( } MailerType::GmailApi => { let label_map = - GmailClient::reverse_label_map(account.id, account.use_proxy, true).await?; + GmailClient::for_lookup_label_id(account.id, account.use_proxy, true).await?; let label_id = label_map.get(mailbox_name).ok_or_else(|| { raise_error!( format!("Label not found for mailbox: {}", mailbox_name), @@ -135,7 +144,7 @@ async fn fetch_remote_messages( let message_list = GmailClient::list_messages( account.id, account.use_proxy, - label_id, + Some(label_id), next_page_token, None, page_size as u32, @@ -171,7 +180,7 @@ async fn fetch_remote_messages( GmailClient::get_message(account_id, use_proxy, &index.id).await }) .await?; - + let label_map = GmailClient::for_get_label_name(account.id, account.use_proxy).await?; let envelopes: Vec = batch_messages .into_iter() .map(|m| { @@ -197,10 +206,21 @@ async fn fetch_remote_messages( let folder = folders .into_iter() .find(|f| f.display_name == mailbox_name) - .ok_or_else(|| raise_error!("".into(), ErrorCode::InvalidParameter))?; - let total_items = folder - .total_item_count - .ok_or_else(|| raise_error!("".into(), ErrorCode::InvalidParameter))?; + .ok_or_else(|| { + raise_error!( + format!("Outlook folder '{}' not found.", mailbox_name), + ErrorCode::InvalidParameter + ) + })?; + let total_items = folder.total_item_count.ok_or_else(|| { + raise_error!( + format!( + "Outlook folder '{}' does not provide 'total_item_count'.", + mailbox_name + ), + ErrorCode::InvalidParameter + ) + })?; if total_items == 0 { return Ok(CursorDataPage::new( @@ -224,7 +244,7 @@ async fn fetch_remote_messages( ) .await?; - let envelopes: Vec = resp + let envelopes: Vec = resp .value .into_iter() .map(|m| { @@ -232,9 +252,12 @@ async fn fetch_remote_messages( envelope.account_id = account.id; envelope.folder_id = mailbox_id(account.id, &folder.id); envelope.folder_name = mailbox_name.to_string(); - Ok(envelope) + let conversation_id = envelope.conversation_id.clone(); + let mut e = Envelope::from(envelope); + e.thread_id = conversation_id.unwrap_or("unknown".into()); + Ok(e) }) - .collect::>>()?; + .collect::>>()?; let next_page_token = if page == total_pages { None @@ -247,7 +270,7 @@ async fn fetch_remote_messages( Some(page_size), total_items as u64, Some(total_pages), - envelopes.into_iter().map(Into::into).collect(), + envelopes, )) } } @@ -328,7 +351,7 @@ async fn fetch_local_messages( total_pages, } = GmailEnvelope::list_messages_in_label(target_label.id, page, page_size, desc) .await?; - let map = GmailClient::label_map(account.id, account.use_proxy).await?; + let map = GmailClient::for_get_label_name(account.id, account.use_proxy).await?; if total_items == 0 { Ok(CursorDataPage::new(None, page_size, 0, None, vec![])) @@ -398,21 +421,22 @@ async fn fetch_local_messages( pub async fn list_threads_in_mailbox( account_id: u64, mailbox_name: &str, - page: u64, + next_page_token: Option<&str>, page_size: u64, + remote: bool, desc: bool, -) -> RustMailerResult> { +) -> RustMailerResult> { let account = AccountModel::check_account_active(account_id, false).await?; - validate_pagination_params(page, page_size)?; - if account.minimal_sync() { + if page_size == 0 { return Err(raise_error!( - format!( - "Account {} is in minimal sync mode. Listing threads in a mailbox is not supported. \ - To enable this feature, you must delete the email account configuration and set it up again \ - with minimal sync mode disabled.", - account_id - ), - ErrorCode::Incompatible + "page_size must be greater than 0.".into(), + ErrorCode::InvalidParameter + )); + } + if page_size > 100 { + return Err(raise_error!( + "The page_size exceeds the maximum allowed limit of 100.".into(), + ErrorCode::InvalidParameter )); } @@ -430,52 +454,335 @@ pub async fn list_threads_in_mailbox( match account.mailer_type { MailerType::ImapSmtp => { + if account.minimal_sync() { + return Err(raise_error!( + format!( + "Account {} is configured with minimal_sync. RustMailer does not sync the required \ + metadata for IMAP accounts in this mode, so thread information cannot be obtained. \ + IMAP servers do not provide native thread data, therefore you must recreate the \ + account without minimal_sync to enable thread listing.", + account_id + ), + ErrorCode::Incompatible + )); + } + + if remote { + return Err(raise_error!( + format!( + "IMAP accounts cannot use the 'remote' parameter because thread lists cannot be \ + retrieved directly from an IMAP server. The IMAP protocol does not provide any \ + native thread information, so remote thread listing is unsupported.", + ), + ErrorCode::Incompatible + )); + } + + let page = decode_page_token(next_page_token)?; let mailbox = MailBox::get(account.id, mailbox_name) .await .map_err(|_| not_found_err())?; - EmailThread::list_threads_in_mailbox(mailbox.id, page, page_size, desc).await + let DataPage { + current_page: _, + page_size, + total_items, + items, + total_pages, + } = EmailThread::list_threads_in_mailbox(mailbox.id, page, page_size, desc).await?; + + if total_items == 0 { + Ok(CursorDataPage::new(None, page_size, 0, None, vec![])) + } else { + let total_pages = total_pages.ok_or_else(|| { + raise_error!( + "Internal error: total_pages is None (this should never happen)".into(), + ErrorCode::InternalError + ) + })?; + + let next_page_token = if page == total_pages { + None + } else { + Some(base64_encode_url_safe!((page + 1).to_string())) + }; + + Ok(CursorDataPage::new( + next_page_token, + page_size, + total_items, + Some(total_pages), + items, + )) + } } MailerType::GmailApi => { - let label = GmailLabels::get_by_name(account_id, mailbox_name).await?; - EmailThread::list_threads_in_label(account, label.id, page, page_size, desc).await + if account.minimal_sync() || remote { + let label_map = + GmailClient::for_lookup_label_id(account.id, account.use_proxy, true).await?; + let label_id = label_map.get(mailbox_name).ok_or_else(|| { + raise_error!( + format!("Label not found for mailbox: {}", mailbox_name), + ErrorCode::ResourceNotFound + ) + })?; + + list_threads_impl( + account_id, + account.use_proxy, + Some(label_id), + next_page_token, + None, + page_size, + ) + .await + } else { + let page = decode_page_token(next_page_token)?; + let label = GmailLabels::get_by_name(account_id, mailbox_name).await?; + let DataPage { + current_page: _, + page_size, + total_items, + items, + total_pages, + } = EmailThread::list_threads_in_label(account, label.id, page, page_size, desc) + .await?; + + if total_items == 0 { + Ok(CursorDataPage::new(None, page_size, 0, None, vec![])) + } else { + let total_pages = total_pages.ok_or_else(|| { + raise_error!( + "Internal error: total_pages is None (this should never happen)".into(), + ErrorCode::InternalError + ) + })?; + + let next_page_token = if page == total_pages { + None + } else { + Some(base64_encode_url_safe!((page + 1).to_string())) + }; + + Ok(CursorDataPage::new( + next_page_token, + page_size, + total_items, + Some(total_pages), + items, + )) + } + } } MailerType::GraphApi => { - let folder = OutlookFolder::get_by_name(account_id, mailbox_name).await?; - EmailThread::list_threads_in_folder(folder.id, page, page_size, desc).await + let page = decode_page_token(next_page_token)?; + if account.minimal_sync() || remote { + let folders = + OutlookClient::list_mailfolders(account.id, account.use_proxy).await?; + let folder = folders + .into_iter() + .find(|f| f.display_name == mailbox_name) + .ok_or_else(|| { + raise_error!( + format!("Outlook folder '{}' not found.", mailbox_name), + ErrorCode::InvalidParameter + ) + })?; + let total_items = folder.total_item_count.ok_or_else(|| { + raise_error!( + format!( + "Outlook folder '{}' does not provide 'total_item_count'.", + mailbox_name + ), + ErrorCode::InvalidParameter + ) + })?; + + if total_items == 0 { + return Ok(CursorDataPage::new( + None, + Some(page_size), + 0, + Some(0), + vec![], + )); + } + + let total_pages = (total_items as f64 / page_size as f64).ceil() as u64; + let resp = OutlookClient::list_messages( + account.id, + account.use_proxy, + &folder.id, + page, + page_size, + None, + ) + .await?; + + let envelopes: Vec = resp + .value + .into_iter() + .map(|m| { + let mut envelope: OutlookEnvelope = m.try_into()?; + envelope.account_id = account.id; + envelope.folder_id = mailbox_id(account.id, &folder.id); + envelope.folder_name = mailbox_name.to_string(); + let conversation_id = envelope.conversation_id.clone(); + let mut e = Envelope::from(envelope); + e.thread_id = conversation_id.unwrap_or("unknown".into()); + Ok(e) + }) + .collect::>>()?; + + let next_page_token = if page == total_pages { + None + } else { + Some(base64_encode_url_safe!((page + 1).to_string())) + }; + + Ok(CursorDataPage::new( + next_page_token, + Some(page_size), + total_items as u64, + Some(total_pages), + envelopes, + )) + } else { + let folder = OutlookFolder::get_by_name(account_id, mailbox_name).await?; + let DataPage { + current_page: _, + page_size, + total_items, + items, + total_pages, + } = EmailThread::list_threads_in_folder(folder.id, page, page_size, desc).await?; + if total_items == 0 { + Ok(CursorDataPage::new(None, page_size, 0, None, vec![])) + } else { + let total_pages = total_pages.ok_or_else(|| { + raise_error!( + "Internal error: total_pages is None (this should never happen)".into(), + ErrorCode::InternalError + ) + })?; + + let next_page_token = if page == total_pages { + None + } else { + Some(base64_encode_url_safe!((page + 1).to_string())) + }; + + Ok(CursorDataPage::new( + next_page_token, + page_size, + total_items, + Some(total_pages), + items, + )) + } + } } } } pub async fn get_thread_messages( account_id: u64, - thread_id: u64, + thread_id: String, + remote: bool, ) -> RustMailerResult> { let account = AccountModel::check_account_active(account_id, false).await?; - if account.minimal_sync() { - return Err(raise_error!( - format!( - "Account {} is in minimal sync mode. Listing threads in a mailbox is not supported. \ - To enable this feature, you must delete the email account configuration and set it up again \ - with minimal sync mode disabled.", - account_id - ), - ErrorCode::Incompatible - )); - } match account.mailer_type { - MailerType::ImapSmtp => EmailEnvelopeV3::get_thread(account_id, thread_id).await, + MailerType::ImapSmtp => { + if account.minimal_sync() { + return Err(raise_error!( + format!( + "Account {} is currently using minimal sync mode. IMAP accounts cannot provide thread \ + information when minimal sync is enabled because IMAP servers do not supply thread IDs. \ + Thread data must be computed locally, which requires full metadata caching. \ + To use thread-related features, please delete and re-add the email account with minimal sync disabled.", + account_id + ), + ErrorCode::Incompatible + )); + } + + if remote { + return Err(raise_error!( + format!( + "IMAP accounts cannot use the 'remote' parameter because thread messages cannot be \ + retrieved directly from an IMAP server. The IMAP protocol does not provide any \ + native thread information, so get remote thread messages is unsupported.", + ), + ErrorCode::Incompatible + )); + } + + let thread_id_num: u64 = thread_id.parse().map_err(|_| { + raise_error!( + format!( + "Invalid thread_id: '{}'. IMAP thread_id must be a numeric string that can be parsed into a number.", + thread_id + ), + ErrorCode::InvalidParameter + ) + })?; + + EmailEnvelopeV3::get_thread(account_id, thread_id_num).await + } MailerType::GmailApi => { - let envelopes = GmailEnvelope::get_thread(account_id, thread_id).await?; - let map = GmailClient::label_map(account_id, account.use_proxy).await?; - Ok(envelopes - .into_iter() - .map(|e| e.into_envelope(&map)) - .collect()) + if account.minimal_sync() || remote { + get_thread_messages_impl(account_id, account.use_proxy, &thread_id).await + } else { + let thread_id_num: u64 = thread_id.parse().map_err(|_| { + raise_error!( + format!( + "Invalid thread_id: '{}'. thread_id must be a numeric string that can be parsed into a number.", + thread_id + ), + ErrorCode::InvalidParameter + ) + })?; + let envelopes = GmailEnvelope::get_thread(account_id, thread_id_num).await?; + let map = GmailClient::for_get_label_name(account_id, account.use_proxy).await?; + Ok(envelopes + .into_iter() + .map(|e| e.into_envelope(&map)) + .collect()) + } } MailerType::GraphApi => { - let envelopes = OutlookEnvelope::get_thread(account_id, thread_id).await?; - Ok(envelopes.into_iter().map(|e| e.into()).collect()) + if account.minimal_sync() || remote { + let resp = + OutlookClient::get_thread_messages(account_id, account.use_proxy, &thread_id) + .await?; + let envelopes: Vec = resp + .value + .into_iter() + .map(|m| { + let mut envelope: OutlookEnvelope = m.try_into()?; + envelope.account_id = account.id; + envelope.folder_id = 0; + envelope.folder_name = "".to_string(); + let conversation_id = envelope.conversation_id.clone(); + let mut e = Envelope::from(envelope); + e.thread_id = conversation_id.unwrap_or("unknown".into()); + Ok(e) + }) + .collect::>>()?; + Ok(envelopes) + } else { + let thread_id_num: u64 = thread_id.parse().map_err(|_| { + raise_error!( + format!( + "Invalid thread_id: '{}'. thread_id must be a numeric string that can be parsed into a number.", + thread_id + ), + ErrorCode::InvalidParameter + ) + })?; + let envelopes = OutlookEnvelope::get_thread(account_id, thread_id_num).await?; + Ok(envelopes.into_iter().map(Into::into).collect()) + } } } } diff --git a/src/modules/message/search/payload.rs b/src/modules/message/search/payload.rs index 9f114c9..01f7e3a 100644 --- a/src/modules/message/search/payload.rs +++ b/src/modules/message/search/payload.rs @@ -522,7 +522,7 @@ impl MessageSearchRequest { let query = self.search.to_gmail_api_search_command()?; let label_map: AHashMap = - GmailClient::reverse_label_map(account.id, account.use_proxy, false).await?; + GmailClient::for_lookup_label_id(account.id, account.use_proxy, false).await?; let label_id = match self.mailbox.as_deref() { Some(name) => label_map.get(name).cloned().map(Some).ok_or_else(|| { @@ -573,6 +573,8 @@ impl MessageSearchRequest { }) .await?; + let label_map = GmailClient::for_get_label_name(account.id, account.use_proxy).await?; + let envelopes: Vec = batch_messages .into_iter() .map(|m| { @@ -821,7 +823,7 @@ impl UnifiedSearchRequest { })? .into(), MailerType::GmailApi => { - let label_map = GmailClient::label_map(account_id, account.use_proxy).await?; + let label_map = GmailClient::for_get_label_name(account_id, account.use_proxy).await?; let envelope = GmailEnvelope::get(id).await?.ok_or_else(|| { raise_error!( format!( diff --git a/src/modules/message/tags.rs b/src/modules/message/tags.rs index 4ebfb2b..a3a76e2 100644 --- a/src/modules/message/tags.rs +++ b/src/modules/message/tags.rs @@ -172,7 +172,7 @@ pub async fn tag_messages_impl(account_id: u64, payload: BatchTagRequest) -> Rus } MailerType::GmailApi => { let labels_map = - GmailClient::reverse_label_map(account_id, account.use_proxy, true).await?; + GmailClient::for_lookup_label_id(account_id, account.use_proxy, true).await?; let tags_to_process = &payload.tags; let mut target_label_ids: Vec = Vec::with_capacity(tags_to_process.len()); for tag_name in tags_to_process { diff --git a/src/modules/message/transfer.rs b/src/modules/message/transfer.rs index ee4b1e9..438bed2 100644 --- a/src/modules/message/transfer.rs +++ b/src/modules/message/transfer.rs @@ -137,7 +137,7 @@ pub async fn transfer_messages( } let labels_map = - GmailClient::reverse_label_map(account_id, account.use_proxy, true).await?; + GmailClient::for_lookup_label_id(account_id, account.use_proxy, true).await?; match transfer { MessageTransfer::Move => { let target_label_id = diff --git a/src/modules/rest/api/access_token.rs b/src/modules/rest/api/access_token.rs index 60ad040..b4182f0 100644 --- a/src/modules/rest/api/access_token.rs +++ b/src/modules/rest/api/access_token.rs @@ -19,7 +19,6 @@ pub struct AccessTokenApi; #[OpenApi(prefix_path = "/api/v1", tag = "ApiTags::AccessToken")] impl AccessTokenApi { /// Lists all access tokens in the system. - /// /// Requires root privileges. #[oai( path = "/access-token-list", @@ -35,7 +34,6 @@ impl AccessTokenApi { } /// Lists access tokens for a specific account. - /// /// Requires root privileges. #[oai( path = "/access-token-list/:account_id", @@ -52,7 +50,6 @@ impl AccessTokenApi { Ok(Json(AccessToken::list_account_tokens(account_id.0).await?)) } /// Deletes a specific access token. - /// /// Requires root privileges. #[oai( path = "/access-token/:token", @@ -70,7 +67,6 @@ impl AccessTokenApi { } /// Creates a new access token. - /// /// Requires root privileges. #[oai( path = "/access-token", @@ -88,7 +84,6 @@ impl AccessTokenApi { } /// Updates an existing access token. - /// /// Requires root privileges. #[oai( path = "/access-token/:token", @@ -108,7 +103,6 @@ impl AccessTokenApi { } /// Regenerates the root access token. - /// /// Requires root privileges. #[oai( path = "/reset-root-token", @@ -121,7 +115,6 @@ impl AccessTokenApi { } /// Reset the Root user's password. - /// /// Only callable by an already authenticated Root user. /// This endpoint updates the Root password to `password_str` /// and regenerates the `root_token`, invalidating any previous token. @@ -140,7 +133,6 @@ impl AccessTokenApi { } // /// Login endpoint for the Root user. - // /// // /// Accepts the Root password and returns the `root_token` // /// which should be used in subsequent requests for authentication. // #[oai(path = "/login", method = "post", operation_id = "login")] diff --git a/src/modules/rest/api/account.rs b/src/modules/rest/api/account.rs index ea023a3..2089fe9 100644 --- a/src/modules/rest/api/account.rs +++ b/src/modules/rest/api/account.rs @@ -168,7 +168,6 @@ impl AccountApi { } /// Get a minimal list of active accounts for use in selectors when creating account-related resources - /// /// This endpoint provides a lightweight list of accounts containing only essential information (id and name). /// It's primarily designed for UI selectors/dropdowns when creating or associating resources with accounts. #[oai( diff --git a/src/modules/rest/api/auto_config.rs b/src/modules/rest/api/auto_config.rs index f510508..5a87964 100644 --- a/src/modules/rest/api/auto_config.rs +++ b/src/modules/rest/api/auto_config.rs @@ -17,6 +17,7 @@ pub struct AutoConfigApi; #[OpenApi(prefix_path = "/api/v1", tag = "ApiTags::AutoConfig")] impl AutoConfigApi { /// Retrieve mail server configuration for a given email address + /// #[oai( path = "/autoconfig/:email_address", method = "get", diff --git a/src/modules/rest/api/event_hook.rs b/src/modules/rest/api/event_hook.rs index fa4d055..0580026 100644 --- a/src/modules/rest/api/event_hook.rs +++ b/src/modules/rest/api/event_hook.rs @@ -147,7 +147,6 @@ impl EventHookApi { } /// List event hooks (root) - /// /// Requires root privileges. #[oai( path = "/event-hook-list", @@ -284,7 +283,6 @@ impl EventHookApi { } /// Mark a hook task for deletion from queue - /// /// Initiates asynchronous removal of an event hook task by marking it for deletion. /// The task will be: /// 1. Immediately marked as "cancelled" in the system diff --git a/src/modules/rest/api/license.rs b/src/modules/rest/api/license.rs index 20a678b..04dcb79 100644 --- a/src/modules/rest/api/license.rs +++ b/src/modules/rest/api/license.rs @@ -15,7 +15,6 @@ pub struct LicenseApi; #[OpenApi(prefix_path = "/api/v1", tag = "ApiTags::License")] impl LicenseApi { /// Retrieve current license information - /// /// Requires root privileges. #[oai(path = "/license", method = "get", operation_id = "get_license")] async fn get_license(&self, context: ClientContext) -> ApiResult> { @@ -30,7 +29,6 @@ impl LicenseApi { } /// Upload and activate a new license - /// /// Requires root privileges. #[oai(path = "/license", method = "post", operation_id = "set_license")] async fn set_license( diff --git a/src/modules/rest/api/mailbox.rs b/src/modules/rest/api/mailbox.rs index 24ee263..0f070fa 100644 --- a/src/modules/rest/api/mailbox.rs +++ b/src/modules/rest/api/mailbox.rs @@ -7,7 +7,7 @@ use crate::modules::common::auth::ClientContext; use crate::modules::mailbox::create::{create_mailbox, CreateMailboxRequest}; use crate::modules::mailbox::delete::delete_mailbox; use crate::modules::mailbox::list::{get_account_mailboxes, list_subscribed_mailboxes}; -use crate::modules::mailbox::rename::{update_mailbox, MailboxUpdateRequest}; +use crate::modules::mailbox::rename::{update_mailbox_impl, MailboxUpdateRequest}; use crate::modules::mailbox::subscribe::{subscribe_mailbox, unsubscribe_mailbox}; use crate::modules::rest::api::ApiTags; use crate::modules::rest::ApiResult; @@ -20,11 +20,9 @@ pub struct MailBoxApi; #[OpenApi(prefix_path = "/api/v1", tag = "ApiTags::Mailbox")] impl MailBoxApi { /// Returns all available mailboxes for the given account. - /// /// - For IMAP/SMTP accounts, this corresponds to folders/mailboxes. /// - For Gmail API accounts, this corresponds to labels visible via the /// `list messages` API (serving as mailbox equivalents). - /// /// Both account types support two modes: /// - Using the local cache of mailboxes/labels. /// - Querying the remote service directly for the latest state. @@ -48,9 +46,7 @@ impl MailBoxApi { } /// Returns a list of mailboxes that the user is currently subscribed to. - /// /// This is only applicable to IMAP/SMTP accounts. - /// /// In the IMAP protocol, this list reflects which mailboxes the user has /// chosen to subscribe to on the server side, as maintained by the IMAP server. /// This is not a synchronized list of all mail folders, but rather the @@ -72,13 +68,10 @@ impl MailBoxApi { } /// Subscribes to a mailbox with the specified name. - /// /// This operation is only applicable to IMAP/SMTP accounts. - /// /// In the IMAP protocol, it marks the mailbox as subscribed on the /// server side. It does not create or synchronize the mailbox, but /// only updates the server-maintained subscription list. - /// /// Unsupported for Gmail API accounts. #[oai( path = "/subscribe-mailbox/:account_id", @@ -99,13 +92,10 @@ impl MailBoxApi { } /// Unsubscribes from a mailbox with the specified name. - /// /// This operation is only applicable to IMAP/SMTP accounts. - /// /// In the IMAP protocol, it removes the mailbox from the subscription list /// on the server side. It does not delete the mailbox or stop synchronization, /// but only affects the server’s record of subscribed folders. - /// /// Unsupported for Gmail API accounts. #[oai( path = "/unsubscribe-mailbox/:account_id", @@ -179,6 +169,6 @@ impl MailBoxApi { ) -> ApiResult<()> { let account_id = account_id.0; context.require_account_access(account_id)?; - Ok(update_mailbox(account_id, payload.0).await?) + Ok(update_mailbox_impl(account_id, payload.0).await?) } } diff --git a/src/modules/rest/api/message.rs b/src/modules/rest/api/message.rs index c5375de..b3d2769 100644 --- a/src/modules/rest/api/message.rs +++ b/src/modules/rest/api/message.rs @@ -112,13 +112,11 @@ impl MessageApi { } /// Batch modifies the custom tags, categories, or keywords on messages. - /// /// This interface is dedicated to operating on **user-defined labels** and is separate /// from standard system status flags (like Read/Unread). /// It unifies tagging across different email services: /// - **Gmail/Graph API:** Operates on user-defined Label IDs or Category Names. /// - **IMAP/SMTP:** Operates on custom IMAP Keywords (Custom Flags). - /// /// **Note:** This is a high-level operation designed for user tag management. #[oai( path = "/tag-messages/:account_id", @@ -155,7 +153,6 @@ impl MessageApi { /// fetches messages from the IMAP server; otherwise, uses local data. remote: Query>, /// The token for fetching the next page of results in pagination. - /// /// - If `None`, this indicates that the first page should be returned. /// - If `Some(token)`, the page corresponding to this token will be fetched. next_page_token: Query>, @@ -197,21 +194,32 @@ impl MessageApi { /// This name is presented as it appears to users, with any encoding (e.g., UTF-7) automatically handled by the system, /// so no manual decoding is required. mailbox: Query, - /// The page number for pagination (1-based). - page: Query, + /// The token for fetching the next page of results in pagination. + /// - If `None`, this indicates that the first page should be returned. + /// - If `Some(token)`, the page corresponding to this token will be fetched. + next_page_token: Query>, /// The number of messages per page. page_size: Query, /// lists messages in descending order; otherwise, ascending. internal date desc: Query>, + /// fetches threads from the gmail/outlook server; otherwise, uses local data. + remote: Query>, context: ClientContext, - ) -> ApiResult>> { + ) -> ApiResult>> { let desc = desc.0.unwrap_or(false); let account_id = account_id.0; context.require_account_access(account_id)?; - + let remote = remote.0.unwrap_or(false); Ok(Json( - list_threads_in_mailbox(account_id, mailbox.0.trim(), page.0, page_size.0, desc) - .await?, + list_threads_in_mailbox( + account_id, + mailbox.0.trim(), + next_page_token.0.as_deref(), + page_size.0, + remote, + desc, + ) + .await?, )) } @@ -226,13 +234,15 @@ impl MessageApi { /// The ID of the account owning the mailbox. account_id: Path, // Thread ID - thread_id: Query, + thread_id: Query, + /// fetches threads from the gmail/outlook server; otherwise, uses local data. + remote: Query>, context: ClientContext, ) -> ApiResult>> { let account_id = account_id.0; context.require_account_access(account_id)?; - - Ok(Json(get_thread_messages(account_id, thread_id.0).await?)) + let remote = remote.0.unwrap_or(false); + Ok(Json(get_thread_messages(account_id, thread_id.0, remote).await?)) } /// Fetches the content of a specific email for the given account. @@ -330,7 +340,6 @@ impl MessageApi { /// The ID of the account owning the mailboxes. account_id: Path, /// The token for fetching the next page of results in pagination. - /// /// - If `None`, this indicates that the first page should be returned. /// - If `Some(token)`, the page corresponding to this token will be fetched. next_page_token: Query>, @@ -400,7 +409,6 @@ impl MessageApi { /// Creates a reply draft email for the specified account. /// The server internally constructs the reply email, automatically linking it to the /// original email thread by applying appropriate headers such as `References` and `In-Reply-To`. - /// /// The newly created draft is appended into the specified draft mailbox. #[oai( path = "/append-reply-to-draft/:account_id", diff --git a/src/modules/rest/api/mta.rs b/src/modules/rest/api/mta.rs index eedf08d..f1b33da 100644 --- a/src/modules/rest/api/mta.rs +++ b/src/modules/rest/api/mta.rs @@ -39,7 +39,6 @@ impl MTAApi { } /// Deletes an existing MTA configuration identified by its name. - /// /// Requires root privileges. #[oai(path = "/mta/:id", method = "delete", operation_id = "remove_mta")] async fn remove_mta( @@ -53,7 +52,6 @@ impl MTAApi { } /// Creates a new MTA configuration. - /// /// Requires root privileges. #[oai(path = "/mta", method = "post", operation_id = "create_mta")] async fn create_mta( @@ -68,7 +66,6 @@ impl MTAApi { } /// Updates an existing MTA configuration by its name. - /// /// Requires root privileges. #[oai(path = "/mta/:id", method = "post", operation_id = "update_mta")] async fn update_mta( diff --git a/src/modules/rest/api/oauth2.rs b/src/modules/rest/api/oauth2.rs index db9491c..8e1ad92 100644 --- a/src/modules/rest/api/oauth2.rs +++ b/src/modules/rest/api/oauth2.rs @@ -21,7 +21,6 @@ pub struct OAuth2Api; #[OpenApi(prefix_path = "/api/v1", tag = "ApiTags::OAuth2")] impl OAuth2Api { /// Retrieves the OAuth2 configuration for a specified name. - /// /// Requires root privileges. /// This endpoint fetches the OAuth2 configuration identified by the given name. #[oai( @@ -46,7 +45,6 @@ impl OAuth2Api { } /// Deletes an OAuth2 configuration by name. - /// /// Requires root privileges. /// This endpoint removes the OAuth2 configuration identified by the specified name. #[oai( @@ -65,7 +63,6 @@ impl OAuth2Api { } /// Creates a new OAuth2 configuration. - /// /// Requires root privileges. /// This endpoint creates a new OAuth2 configuration based on the provided request data. #[oai( @@ -85,7 +82,6 @@ impl OAuth2Api { } /// Updates an existing OAuth2 configuration. - /// /// Requires root privileges. /// This endpoint updates the OAuth2 configuration identified by the specified name #[oai( @@ -106,7 +102,6 @@ impl OAuth2Api { } /// Lists OAuth2 configurations with pagination and sorting options. - /// /// This endpoint retrieves a paginated list of OAuth2 configurations, allowing for /// optional pagination and sorting parameters. It requires root access. #[oai( @@ -131,7 +126,6 @@ impl OAuth2Api { } /// Generates an OAuth2 authorization URL for a specific account. - /// /// This endpoint creates an authorization URL for the specified OAuth2 configuration /// and account ID. It requires root access and returns the URL as plain text. #[oai( @@ -152,7 +146,6 @@ impl OAuth2Api { } /// Retrieves OAuth2 access tokens for a specified account. - /// /// This endpoint fetches the OAuth2 access tokens associated with the given account ID. #[oai( path = "/oauth2-tokens/:account_id", @@ -178,7 +171,6 @@ impl OAuth2Api { } /// Configures an external OAuth2 token for a specified account. - /// /// This endpoint allows two usage modes: /// 1. If only an `access_token` is provided, RustMailer will store it directly. /// - In this mode, RustMailer **cannot refresh** the token, since it has no @@ -190,7 +182,6 @@ impl OAuth2Api { /// - Since the OAuth2 configuration (including client_id and client_secret) /// is already stored in RustMailer, the service can use the refresh token /// to obtain new access tokens automatically. - /// /// Note: The `oauth2_id` must reference a valid OAuth2 configuration /// already created in RustMailer. #[oai( diff --git a/src/modules/rest/api/send.rs b/src/modules/rest/api/send.rs index 2a76d44..d1a6bd0 100644 --- a/src/modules/rest/api/send.rs +++ b/src/modules/rest/api/send.rs @@ -27,7 +27,6 @@ pub struct SendMailApi; #[OpenApi(prefix_path = "/api/v1", tag = "ApiTags::SendMail")] impl SendMailApi { /// Sends a new email for a specified account. - /// /// This endpoint constructs and sends a new email based on the provided request data. #[oai( path = "/send-mail/:account_id", @@ -49,7 +48,6 @@ impl SendMailApi { } /// Sends a reply to an existing email for a specified account. - /// /// This endpoint constructs and sends a reply to an email based on the provided request data. #[oai( path = "/reply-mail/:account_id", @@ -71,7 +69,6 @@ impl SendMailApi { } /// Forwards an existing email for a specified account. - /// /// This endpoint constructs and sends a forwarded email based on the provided request data. #[oai( path = "/forward-mail/:account_id", @@ -93,7 +90,6 @@ impl SendMailApi { } /// Lists email tasks with pagination, sorting, and optional status filtering. - /// /// This endpoint retrieves a paginated list of email tasks, filtered by accessible accounts /// and optionally by task status. It supports sorting in ascending or descending order by creation time. #[oai( @@ -167,7 +163,6 @@ impl SendMailApi { } /// Retrieves a specific email task by its ID. - /// /// This endpoint fetches the details of an email task identified by the provided ID. #[oai( path = "/send-email-task/:id", @@ -190,7 +185,6 @@ impl SendMailApi { } /// Mark a email task for deletion from queue - /// /// Initiates asynchronous removal of an email task by marking it for deletion. /// The task will be: /// 1. Immediately marked as "cancelled" in the system diff --git a/src/modules/rest/api/system.rs b/src/modules/rest/api/system.rs index f1f2829..2999cd0 100644 --- a/src/modules/rest/api/system.rs +++ b/src/modules/rest/api/system.rs @@ -20,7 +20,6 @@ pub struct SystemApi; #[OpenApi(prefix_path = "/api/v1", tag = "ApiTags::System")] impl SystemApi { /// Retrieves important system notifications for the RustMail service. - /// /// This endpoint returns a consolidated view of all critical system notifications including: /// - Available version updates /// - License expiration warnings @@ -37,7 +36,6 @@ impl SystemApi { } /// Retrieves an overview of RustMail service metrics. - /// /// This endpoint returns a consolidated view of all key metrics including: /// - IMAP traffic (sent and received) /// - Email sent counts (success and failure) @@ -109,7 +107,6 @@ impl SystemApi { } /// Delete all entries in the disk cache. Requires root permission. - /// /// The disk cache stores temporary files such as email bodies, attachments, /// and outgoing emails waiting to be sent. This operation will clear all /// cached data, freeing disk space but removing all temporary content. diff --git a/src/modules/rest/api/templates.rs b/src/modules/rest/api/templates.rs index be3170e..5f8e90a 100644 --- a/src/modules/rest/api/templates.rs +++ b/src/modules/rest/api/templates.rs @@ -20,7 +20,6 @@ pub struct TempaltesApi; #[OpenApi(prefix_path = "/api/v1", tag = "ApiTags::Template")] impl TempaltesApi { /// Retrieves an email template by its name. - /// /// Returns the template if found, or a `ResourceNotFound` error if no template matches the provided name. #[oai(path = "/template/:id", method = "get", operation_id = "get_template")] async fn get_template( @@ -37,7 +36,6 @@ impl TempaltesApi { } /// Deletes an email template by its name. - /// /// Removes the specified template if it exists and the client has access. Returns a `ResourceNotFound` error if the template is not found. #[oai( path = "/template/:id", @@ -59,7 +57,6 @@ impl TempaltesApi { } /// Creates a new email template. - /// /// Saves a new email template based on the provided request data. #[oai(path = "/template", method = "post", operation_id = "create_template")] async fn create_template( @@ -72,7 +69,6 @@ impl TempaltesApi { } /// Updates an existing email template by its name. - /// /// Modifies the specified template with the provided update data if it exists and the client has access. Returns a `ResourceNotFound` error if the template is not found. #[oai( path = "/template/:id", @@ -95,7 +91,6 @@ impl TempaltesApi { } /// Lists all email templates with pagination. - /// /// Retrieves a paginated list of all email templates. /// Requires root privileges. #[oai( @@ -120,7 +115,6 @@ impl TempaltesApi { } /// Lists email templates associated with a specific account. - /// /// Retrieves a paginated list of templates for the specified account ID. Requires access to the specified account. #[oai( path = "/account-templates/:account_id", @@ -147,7 +141,6 @@ impl TempaltesApi { } /// Deletes all email templates associated with a specific account. - /// /// Removes all templates linked to the specified account ID. Requires access to the specified account. #[oai( path = "/account-templates/:account_id", @@ -166,7 +159,6 @@ impl TempaltesApi { } /// Send a test email using a specific template - /// /// This endpoint allows sending a test email to verify template rendering and delivery. #[oai( path = "/template-send-test/:id", diff --git a/src/modules/smtp/request/mod.rs b/src/modules/smtp/request/mod.rs index 7b14e31..b908739 100644 --- a/src/modules/smtp/request/mod.rs +++ b/src/modules/smtp/request/mod.rs @@ -635,7 +635,7 @@ impl EmailHandler { label_name: &str, mid: &str, ) -> RustMailerResult { - let map = GmailClient::label_map(account.id, account.use_proxy).await?; + let map = GmailClient::for_get_label_name(account.id, account.use_proxy).await?; if let Ok(label) = GmailLabels::get_by_name(account.id, label_name).await { if !account.minimal_sync() { let envelope = GmailEnvelope::find(account.id, label.id, mid).await?;