feat: Add Gmail API sync workflow and integrate related data models

This commit is contained in:
rustmailer
2025-09-01 23:50:00 +08:00
parent 4b44a38f80
commit 7411233b48
50 changed files with 2811 additions and 200 deletions
+24 -1
View File
@@ -135,6 +135,10 @@ message Account {
int64 updated_at = 16;
// Method used to access and manage emails.
MailerType mailer_type = 17;
// Optional proxy ID for establishing the connection to external APIs (e.g., Gmail, Outlook).
// - If `None` or not provided, the client will connect directly to the API server.
// - If `Some(proxy_id)`, the client will use the pre-configured proxy with the given ID for API requests.
optional uint64 use_proxy = 18;
}
// PagedAccount represents a paginated list of Account messages.
@@ -173,6 +177,10 @@ message AccountCreateRequest {
int64 incremental_sync_interval_sec = 9;
// Method used to access and manage emails.
MailerType mailer_type = 10;
// Optional proxy ID for establishing the connection to external APIs (e.g., Gmail, Outlook).
// - If `None` or not provided, the client will connect directly to the API server.
// - If `Some(proxy_id)`, the client will use the pre-configured proxy with the given ID for API requests.
optional uint64 use_proxy = 11;
}
// AccountUpdateRequest defines the parameters for updating an existing email account.
@@ -195,6 +203,10 @@ message AccountUpdateRequest {
optional int64 full_sync_interval_min = 8;
// Optional: Update the interval (in seconds) for incremental synchronization updates.
optional int64 incremental_sync_interval_sec = 9;
// Optional proxy ID for establishing the connection to external APIs (e.g., Gmail, Outlook).
// - If `None` or not provided, the client will connect directly to the API server.
// - If `Some(proxy_id)`, the client will use the pre-configured proxy with the given ID for API requests.
optional uint64 use_proxy = 10;
}
// AccountError represents an error encountered during account processing.
@@ -338,7 +350,7 @@ service AutoConfigService {
// MailBox represents a mail folder on an IMAP server.
message MailBox {
// The unique hash of the mailbox, serving as a primary key.
uint64 mailbox_hash = 1;
uint64 mailbox_id = 1;
// The hash of the account to which this mailbox belongs, serving as a secondary key.
uint64 account_hash = 2;
// The unique name of the mailbox (e.g., "INBOX", "Sent Items").
@@ -696,6 +708,17 @@ message EmailEnvelope {
// The identifier of the thread this email belongs to.
// This is computed based on `in_reply_to` / `references` / `message_id`.
uint64 thread_id = 26;
// The `mid` field is reserved for potential integration with other backend models.
// For instance, it can be used to store the email index or ID from external services like the Gmail API.
// This ID could be used for reference or identification purposes in scenarios where an external service
// provides an identifier for the email in question.
// This field is optional, meaning that it may be `None` if no external service identifier is available.
optional string mid = 27;
// A list of labels applied to the message.
// Each element is a string representing a Gmail label ID (e.g., "INBOX", "UNREAD").
// This field reflects the current labels associated with the email.
// Note: This field is populated only for Gmail API accounts. For other account types, it will be empty.
repeated string label_ids = 28;
}
// FetchMessageContentRequest is used to fetch specific content sections of an email message.
+8
View File
@@ -60,6 +60,10 @@ pub struct AccountCreateRequest {
/// Incremental sync interval (seconds), default 60s
#[oai(validator(minimum(value = "10"), maximum(value = "3600")))]
pub incremental_sync_interval_sec: i64,
/// Optional proxy ID for establishing the connection to external APIs (e.g., Gmail, Outlook).
/// - If `None` or not provided, the client will connect directly to the API server.
/// - If `Some(proxy_id)`, the client will use the pre-configured proxy with the given ID for API requests.
pub use_proxy: Option<u64>,
}
impl AccountCreateRequest {
@@ -139,6 +143,10 @@ pub struct AccountUpdateRequest {
/// Incremental sync interval (seconds)
#[oai(validator(minimum(value = "10"), maximum(value = "3600")))]
pub incremental_sync_interval_sec: Option<i64>,
/// Optional proxy ID for establishing the connection to external APIs (e.g., Gmail, Outlook).
/// - If `None` or not provided, the client will connect directly to the API server.
/// - If `Some(proxy_id)`, the client will use the pre-configured proxy with the given ID for API requests.
pub use_proxy: Option<u64>,
}
impl AccountUpdateRequest {
+39 -4
View File
@@ -6,7 +6,7 @@ use crate::{
modules::error::{code::ErrorCode, RustMailerResult},
raise_error,
};
use chrono::{Datelike, Days, Local, Months, NaiveDate, Utc};
use chrono::{Datelike, Days, Local, Months, NaiveDate, TimeZone, Utc};
use poem_openapi::{Enum, Object};
use serde::{Deserialize, Serialize};
@@ -101,7 +101,7 @@ impl RelativeDate {
Ok(())
}
pub fn calculate_date(&self) -> RustMailerResult<String> {
fn compute_date(&self) -> RustMailerResult<chrono::DateTime<Local>> {
if self.value == 0 {
return Err(raise_error!(
"Value must be greater than 0".into(),
@@ -125,8 +125,6 @@ impl RelativeDate {
})?;
let naive_date = date.date_naive();
// Check if the date is before 1970
if naive_date.year() < 1970 {
return Err(raise_error!(
format!(
@@ -137,8 +135,18 @@ impl RelativeDate {
));
}
Ok(date)
}
pub fn calculate_date(&self) -> RustMailerResult<String> {
let date = self.compute_date()?;
Ok(date.format("%d-%b-%Y").to_string())
}
pub fn calculate_gmail_date(&self) -> RustMailerResult<String> {
let date = self.compute_date()?;
Ok(date.format("/%Y/%m/%d").to_string())
}
}
impl DateSince {
@@ -218,6 +226,20 @@ impl DateSince {
Ok(date.format("%d-%b-%Y").to_string())
}
pub fn format_for_gmail(&self, fixed: &str) -> RustMailerResult<String> {
let date = NaiveDate::parse_from_str(fixed, "%Y-%m-%d").map_err(|_| {
raise_error!(
format!(
"Invalid date format. Expected 'YYYY-MM-DD'. Example: '2024-11-19'. Provided: '{}'",
fixed
),
ErrorCode::InvalidParameter
)
})?;
Ok(date.format("/%Y/%m/%d").to_string())
}
pub fn since_date(&self) -> RustMailerResult<String> {
// Handle the case where only one of `fixed` or `relative` is provided
if let Some(r) = &self.relative {
@@ -234,6 +256,19 @@ impl DateSince {
))
}
}
pub fn since_gmail_date(&self) -> RustMailerResult<String> {
if let Some(r) = &self.relative {
r.calculate_gmail_date()
} else if let Some(f) = &self.fixed {
self.format_for_gmail(f)
} else {
Err(raise_error!(
"You must provide either a 'fixed' or 'relative' date.".to_string(),
ErrorCode::InvalidParameter
))
}
}
}
#[cfg(test)]
+12 -1
View File
@@ -18,6 +18,7 @@ use crate::{
since::DateSince,
status::AccountRunningState,
},
cache::imap::mailbox::MailBox,
database::{insert_impl, list_all_impl},
error::RustMailerResult,
},
@@ -28,7 +29,6 @@ use crate::id;
use crate::modules::account::payload::AccountCreateRequest;
use crate::modules::account::payload::AccountUpdateRequest;
use crate::modules::account::payload::MinimalAccount;
use crate::modules::cache::imap::mailbox::MailBox;
use crate::modules::cache::imap::manager::EnvelopeFlagsManager;
use crate::modules::cache::imap::task::IMAP_TASKS;
use crate::modules::context::controller::SYNC_CONTROLLER;
@@ -109,6 +109,10 @@ pub struct AccountV2 {
pub created_at: i64,
/// Last update timestamp (UNIX epoch milliseconds)
pub updated_at: i64,
/// Optional proxy ID for establishing the connection to external APIs (e.g., Gmail, Outlook).
/// - If `None` or not provided, the client will connect directly to the API server.
/// - If `Some(proxy_id)`, the client will use the pre-configured proxy with the given ID for API requests.
pub use_proxy: Option<u64>,
}
impl AccountV2 {
@@ -145,6 +149,7 @@ impl AccountV2 {
incremental_sync_interval_sec: request.incremental_sync_interval_sec,
created_at: utc_now!(),
updated_at: utc_now!(),
use_proxy: request.use_proxy,
})
}
@@ -411,6 +416,11 @@ impl AccountV2 {
if let Some(mailboxes) = request.sync_folders {
new.sync_folders = mailboxes;
}
if let Some(use_proxy) = request.use_proxy {
new.use_proxy = Some(use_proxy);
}
if let Some(full_sync_interval_min) = &request.full_sync_interval_min {
new.full_sync_interval_min = Some(*full_sync_interval_min);
}
@@ -469,6 +479,7 @@ impl From<Account> for AccountV2 {
known_folders: value.known_folders,
created_at: value.created_at,
updated_at: value.updated_at,
use_proxy: None,
}
}
}
+79 -3
View File
@@ -14,7 +14,7 @@ use tracing::info;
use crate::{
id,
modules::{
cache::imap::v2::EmailEnvelopeV2,
cache::{imap::v2::EmailEnvelopeV3, vendor::gmail::sync::envelope::GmailEnvelope},
database::{batch_delete_impl, filter_by_secondary_key_impl, manager::DB_MANAGER},
error::{code::ErrorCode, RustMailerResult},
utils::envelope_hash,
@@ -165,7 +165,7 @@ impl AddressEntity {
Ok(())
}
pub fn extract(envelope: &EmailEnvelopeV2) -> Vec<AddressEntity> {
pub fn extract(envelope: &EmailEnvelopeV3) -> Vec<AddressEntity> {
let from = envelope.from.as_ref().map(|f| f.address.clone()).flatten();
let envelope_hash = envelope.create_envelope_id();
let date = envelope.date.clone();
@@ -180,7 +180,83 @@ impl AddressEntity {
account_id,
mailbox_id,
id: id!(96),
from: from.clone(),
from,
to: None,
cc: None,
envelope_hash,
date,
internal_date,
});
}
(None, Some(cc)) => {
entities.extend(cc.iter().map(|c| {
let from = from.clone();
AddressEntity {
account_id,
mailbox_id,
id: id!(96),
from,
to: None,
cc: c.address.clone(),
envelope_hash,
date: date.clone(),
internal_date: internal_date.clone(),
}
}));
}
(Some(to), None) => {
entities.extend(to.iter().map(|t| {
let from = from.clone();
AddressEntity {
account_id,
mailbox_id,
id: id!(96),
from,
to: t.address.clone(),
cc: None,
envelope_hash,
date: date.clone(),
internal_date: internal_date.clone(),
}
}));
}
(Some(to), Some(cc)) => {
entities.extend(to.iter().flat_map(|t| {
let from = from.clone();
cc.iter().map(move |c| AddressEntity {
account_id,
mailbox_id,
id: id!(96),
from: from.clone(),
to: t.address.clone(),
cc: c.address.clone(),
envelope_hash,
date: date.clone(),
internal_date: internal_date.clone(),
})
}));
}
}
entities
}
pub fn extract2(envelope: &GmailEnvelope) -> Vec<AddressEntity> {
let from = envelope.from.as_ref().map(|f| f.address.clone()).flatten();
let envelope_hash = envelope.create_envelope_id();
let date = envelope.date.clone();
let internal_date = Some(envelope.internal_date.clone());
let account_id = envelope.account_id;
let mailbox_id = envelope.label_id;
let mut entities = Vec::new();
match (&envelope.to, &envelope.cc) {
(None, None) => {
entities.push(AddressEntity {
account_id,
mailbox_id,
id: id!(96),
from: from,
to: None,
cc: None,
envelope_hash,
+6 -6
View File
@@ -15,7 +15,7 @@ use crate::modules::cache::imap::flags_to_hash;
use crate::modules::cache::imap::mailbox::EnvelopeFlag;
use crate::modules::cache::imap::minimal::MinimalEnvelope;
use crate::modules::cache::imap::thread::EmailThread;
use crate::modules::cache::imap::v2::EmailEnvelopeV2;
use crate::modules::cache::imap::v2::EmailEnvelopeV3;
use crate::modules::context::Initialize;
use crate::modules::error::RustMailerResult;
use crate::modules::hook::channel::{Event, EVENT_CHANNEL};
@@ -75,7 +75,7 @@ impl EnvelopeFlagsManager {
pub async fn clean_account(account_id: u64) -> RustMailerResult<()> {
FLAGS_STATE_MAP.remove(&account_id);
EmailEnvelopeV2::clean_account(account_id).await?;
EmailEnvelopeV3::clean_account(account_id).await?;
MinimalEnvelope::clean_account(account_id).await?;
AddressEntity::clean_account(account_id).await?;
EmailThread::clean_account(account_id).await
@@ -99,7 +99,7 @@ impl EnvelopeFlagsManager {
FLAGS_STATE_MAP.remove(&account_id);
}
}
EmailEnvelopeV2::clean_envelopes(account_id, mailbox_id, to_delete_uid).await?;
EmailEnvelopeV3::clean_envelopes(account_id, mailbox_id, to_delete_uid).await?;
MinimalEnvelope::clean_envelopes(account_id, mailbox_id, to_delete_uid).await?;
AddressEntity::clean_envelopes(account_id, mailbox_id, to_delete_uid).await?;
EmailThread::clean_envelopes(account_id, mailbox_id, to_delete_uid).await
@@ -110,7 +110,7 @@ impl EnvelopeFlagsManager {
if let Some(mailbox_map) = FLAGS_STATE_MAP.get(&account_id) {
mailbox_map.remove(&mailbox_id);
}
EmailEnvelopeV2::clean_mailbox_envelopes(account_id, mailbox_id).await?;
EmailEnvelopeV3::clean_mailbox_envelopes(account_id, mailbox_id).await?;
MinimalEnvelope::clean_mailbox_envelopes(account_id, mailbox_id).await?;
AddressEntity::clean_mailbox_envelopes(account_id, mailbox_id).await?;
EmailThread::clean_mailbox_envelopes(account_id, mailbox_id).await
@@ -141,7 +141,7 @@ impl EnvelopeFlagsManager {
if !account.minimal_sync()
&& EventHookTask::event_watched(account.id, EventType::EmailFlagsChanged).await?
{
if let Some(current) = EmailEnvelopeV2::find(account.id, mailbox_id, uid).await? {
if let Some(current) = EmailEnvelopeV3::find(account.id, mailbox_id, uid).await? {
let (added, removed) = Self::diff_envelope_flags(&current.flags, &flags);
EVENT_CHANNEL
.queue(Event::new(
@@ -171,7 +171,7 @@ impl EnvelopeFlagsManager {
let flags_hash = flags_to_hash(&flags);
if !account.minimal_sync() {
EmailEnvelopeV2::update_flags(account.id, mailbox_id, uid, &flags, flags_hash)
EmailEnvelopeV3::update_flags(account.id, mailbox_id, uid, &flags, flags_hash)
.await?;
}
MinimalEnvelope::update_flags(account.id, mailbox_id, uid, flags_hash).await?;
+7 -7
View File
@@ -11,7 +11,7 @@ use tracing::{error, info};
use crate::{
modules::{
cache::imap::{v2::EmailEnvelopeV2, manager::EnvelopeFlagsManager},
cache::imap::{manager::EnvelopeFlagsManager, v2::EmailEnvelopeV3},
database::{
batch_delete_impl, batch_insert_impl, delete_impl, filter_by_secondary_key_impl,
manager::DB_MANAGER, update_impl,
@@ -94,7 +94,7 @@ impl MinimalEnvelope {
}
info!(
"Finished deleting envelopes for mailbox_hash={} account_id={} total_deleted={} in {:?}",
"Finished deleting envelopes for mailbox_id={} account_id={} total_deleted={} in {:?}",
mailbox_id,
account_id,
total_deleted,
@@ -105,11 +105,11 @@ impl MinimalEnvelope {
pub async fn clean_envelopes(
account_id: u64,
mailbox_hash: u64,
mailbox_id: u64,
to_delete_uid: &[u32],
) -> RustMailerResult<()> {
for uid in to_delete_uid {
let key = envelope_hash(account_id, mailbox_hash, *uid);
let key = envelope_hash(account_id, mailbox_id, *uid);
delete_impl(DB_MANAGER.envelope_db(), move |rw| {
rw.get()
.primary::<MinimalEnvelope>(key)
@@ -152,7 +152,7 @@ impl MinimalEnvelope {
.await
.map_err(|e| {
error!(
"Failed to update flags: account_id={}, mailbox_hash={}, uid={}, error={:?}",
"Failed to update flags: account_id={}, mailbox_id={}, uid={}, error={:?}",
account_id, mailbox_id, uid, e
);
e
@@ -196,8 +196,8 @@ impl MinimalEnvelope {
}
}
impl From<&EmailEnvelopeV2> for MinimalEnvelope {
fn from(value: &EmailEnvelopeV2) -> Self {
impl From<&EmailEnvelopeV3> for MinimalEnvelope {
fn from(value: &EmailEnvelopeV3) -> Self {
Self {
account_id: value.account_id,
mailbox_id: value.mailbox_id,
+16 -3
View File
@@ -7,9 +7,18 @@ use std::sync::LazyLock;
use crate::{
calculate_hash,
modules::{
cache::imap::{
address::AddressEntity, envelope::EmailEnvelope, minimal::MinimalEnvelope,
thread::EmailThread, v2::EmailEnvelopeV2,
cache::{
imap::{
address::AddressEntity,
envelope::EmailEnvelope,
minimal::MinimalEnvelope,
thread::EmailThread,
v2::{EmailEnvelopeV2, EmailEnvelopeV3},
},
vendor::gmail::sync::{
envelope::GmailEnvelope,
labels::{GmailCheckPoint, GmailLabels},
},
},
database::ModelsAdapter,
},
@@ -33,10 +42,14 @@ pub static ENVELOPE_MODELS: LazyLock<Models> = LazyLock::new(|| {
let mut adapter = ModelsAdapter::new();
adapter.register_model::<EmailEnvelope>();
adapter.register_model::<EmailEnvelopeV2>();
adapter.register_model::<EmailEnvelopeV3>();
adapter.register_model::<MailBox>();
adapter.register_model::<MinimalEnvelope>();
adapter.register_model::<AddressEntity>();
adapter.register_model::<EmailThread>();
adapter.register_model::<GmailEnvelope>();
adapter.register_model::<GmailLabels>();
adapter.register_model::<GmailCheckPoint>();
adapter.models
});
+6 -1
View File
@@ -3,7 +3,7 @@
// Unauthorized copying, modification, or distribution is prohibited.
use crate::modules::{
account::{status::AccountRunningState, v2::AccountV2},
account::{entity::MailerType, status::AccountRunningState, v2::AccountV2},
cache::imap::{mailbox::MailBox, manager::EnvelopeFlagsManager},
error::RustMailerResult,
hook::{
@@ -30,6 +30,11 @@ pub mod sync_type;
static SYNC_COUNTER: AtomicUsize = AtomicUsize::new(0);
pub async fn execute_imap_sync(account: &AccountV2) -> RustMailerResult<()> {
assert!(
matches!(account.mailer_type, MailerType::ImapSmtp),
"Bug: Unexpected mailer type, expected ImapSmtp, found: {:?}",
account.mailer_type
);
let start_time = Instant::now();
let account_id = account.id;
+3 -3
View File
@@ -122,11 +122,11 @@ pub async fn should_rebuild_cache(
mailbox_count: usize,
local_envelope_count: usize,
) -> RustMailerResult<bool> {
// If both remote mailboxes and local envelopes exist, no rebuild is needed.
// If both local mailboxes and local envelopes exist, no rebuild is needed.
if mailbox_count > 0 && local_envelope_count > 0 {
return Ok(false);
}
// If there are remote mailboxes but no local envelopes, clear the mailboxes.
// If there are local mailboxes but no local envelopes, clear the mailboxes.
if mailbox_count > 0 {
let mailboxes = MailBox::list_all(account.id).await?;
MailBox::batch_delete(mailboxes).await?;
@@ -134,7 +134,7 @@ pub async fn should_rebuild_cache(
if local_envelope_count > 0 {
EnvelopeFlagsManager::clean_account(account.id).await?
}
// If either remote mailboxes or local envelopes were missing, cache rebuild is required.
// If either local mailboxes or local envelopes were missing, cache rebuild is required.
Ok(true)
}
+30 -16
View File
@@ -4,7 +4,7 @@
use crate::{
modules::{
account::{status::AccountRunningState, v2::AccountV2},
account::{entity::MailerType, status::AccountRunningState, v2::AccountV2},
error::RustMailerResult,
},
utc_now,
@@ -27,24 +27,38 @@ pub async fn determine_sync_type(account: &AccountV2) -> RustMailerResult<SyncTy
Ok(match AccountRunningState::get(account.id).await? {
Some(info) => {
let now = utc_now!();
if is_time_for_full_sync(
now,
info.last_full_sync_start,
account
.full_sync_interval_min
.unwrap_or(DEFAULT_FULL_SYNC_INTERVAL_MIN),
) {
AccountRunningState::set_full_sync_start(account.id).await?;
SyncType::FullSync
} else if is_time_for_incremental_sync(
let incremental_sync = is_time_for_incremental_sync(
now,
info.last_incremental_sync_start,
account.incremental_sync_interval_sec,
) {
AccountRunningState::set_incremental_sync_start(account.id).await?;
SyncType::IncrementalSync
} else {
SyncType::SkipSync
);
match account.mailer_type {
MailerType::ImapSmtp => {
if is_time_for_full_sync(
now,
info.last_full_sync_start,
account
.full_sync_interval_min
.unwrap_or(DEFAULT_FULL_SYNC_INTERVAL_MIN),
) {
AccountRunningState::set_full_sync_start(account.id).await?;
SyncType::FullSync
} else if incremental_sync {
AccountRunningState::set_incremental_sync_start(account.id).await?;
SyncType::IncrementalSync
} else {
SyncType::SkipSync
}
}
MailerType::GmailApi => {
if incremental_sync {
AccountRunningState::set_incremental_sync_start(account.id).await?;
SyncType::IncrementalSync
} else {
SyncType::SkipSync
}
}
}
}
None => {
+4 -4
View File
@@ -12,7 +12,7 @@ use serde::{Deserialize, Serialize};
use crate::{
modules::{
cache::imap::v2::EmailEnvelopeV2,
cache::imap::v2::EmailEnvelopeV3,
database::{
batch_delete_impl, delete_impl, manager::DB_MANAGER, paginate_secondary_scan_impl,
},
@@ -139,7 +139,7 @@ impl EmailThread {
page: u64,
page_size: u64,
desc: bool,
) -> RustMailerResult<DataPage<EmailEnvelopeV2>> {
) -> RustMailerResult<DataPage<EmailEnvelopeV3>> {
let threads = paginate_secondary_scan_impl::<EmailThread>(
DB_MANAGER.envelope_db(),
Some(page),
@@ -151,7 +151,7 @@ impl EmailThread {
.await?;
let fetch_tasks = threads.items.into_iter().map(|thread| async move {
EmailEnvelopeV2::get(thread.envelope_id)
EmailEnvelopeV3::get(thread.envelope_id)
.await?
.ok_or_else(|| {
raise_error!(
@@ -161,7 +161,7 @@ impl EmailThread {
})
});
let results: RustMailerResult<Vec<EmailEnvelopeV2>> =
let results: RustMailerResult<Vec<EmailEnvelopeV3>> =
join_all(fetch_tasks).await.into_iter().collect();
let envelopes = results?;
+208 -21
View File
@@ -138,6 +138,125 @@ impl EmailEnvelopeV2 {
pub fn create_envelope_id(&self) -> u64 {
envelope_hash(self.account_id, self.mailbox_id, self.uid)
}
}
#[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize, Object)]
#[native_model(id = 1, version = 3, from = EmailEnvelopeV2)]
#[native_db(primary_key(pk -> String), secondary_key(create_envelope_id -> u64, unique))]
pub struct EmailEnvelopeV3 {
/// The ID of the account owning the email.
#[secondary_key]
pub account_id: u64,
/// The unique identifier of the mailbox where the email is stored (e.g., `MailBox::id`).
/// Used for indexing to avoid updating indexes when mailboxes are renamed.
#[secondary_key]
pub mailbox_id: u64,
/// The decoded, human-readable name of the mailbox (e.g., "INBOX", "Sent").
pub mailbox_name: String,
/// The unique identifier (IMAP UID) of the email within the mailbox.
pub uid: u32,
/// The date and time the email was received by the server, as a Unix timestamp in milliseconds.
/// If `None`, the internal date is unavailable.
pub internal_date: Option<i64>,
/// The size of the email in bytes.
pub size: u32,
/// The flags associated with the email (e.g., `\Seen`, `\Answered`, `\Flagged`).
/// Represented as a list of `EnvelopeFlag` for standard or custom flags.
pub flags: Vec<EnvelopeFlag>,
/// A hash of the email's flags for efficient comparison or indexing.
pub flags_hash: u64,
/// The blind carbon copy (BCC) recipient(s) of the email, if any.
pub bcc: Option<Vec<Addr>>,
/// The carbon copy (CC) recipient(s) of the email, if any.
pub cc: Option<Vec<Addr>>,
/// The date the email was sent, as a Unix timestamp in milliseconds, if available.
pub date: Option<i64>,
/// The sender's address, including name and email, if available.
pub from: Option<Addr>,
/// The message ID of the email to which this email is a reply, if applicable.
pub in_reply_to: Option<String>,
/// The actual sender's address, if different from the `from` field.
pub sender: Option<Addr>,
/// The return address for undeliverable emails, if specified.
pub return_address: Option<String>,
/// The unique message ID of the email, typically used for threading.
pub message_id: Option<String>,
/// The subject of the email, if available.
pub subject: Option<String>,
/// The name of the thread this email belongs to, if applicable.
pub thread_name: Option<String>,
/// The identifier of the thread this email belongs to.
/// This is computed based on `in_reply_to` / `references` / `message_id`.
#[secondary_key]
pub thread_id: u64,
/// The MIME version of the email (e.g., "1.0"), if specified.
pub mime_version: Option<String>,
/// A list of message IDs referenced by this email, used for threading.
pub references: Option<Vec<String>>,
/// The address(es) to which replies should be sent, if specified.
pub reply_to: Option<Vec<Addr>>,
/// The primary recipient(s) of the email, if any.
pub to: Option<Vec<Addr>>,
/// A list of attachments included in the email, if any.
///
/// Each `ImapAttachment` item contains metadata including the part ID and MIME type,
/// which indicates the exact location of the attachment in the raw message structure.
/// This allows the backend to directly fetch specific attachments without retrieving
/// the entire message content.
///
/// This is particularly useful for accounts configured with minimal sync, where full
/// message bodies are not cached locally. By including this data in the API response,
/// the client can request to download only the required attachment via a follow-up
/// API call, improving both efficiency and user experience.
///
/// Developers do not need to understand the internal IMAP part structure — this
/// metadata provides a clean abstraction for fetching specific attachments.
pub attachments: Option<Vec<ImapAttachment>>,
/// Metadata for the email's body parts (e.g., plain text, HTML), if available.
///
/// Each `EmailBodyPart` contains detailed metadata (such as part ID, content type,
/// and charset) describing a portion of the email body. This enables precise access
/// to body content, such as plain text or HTML sections, without downloading the full
/// raw message from the server.
///
/// This is especially helpful for lightweight clients or minimized-sync accounts that
/// do not cache full email content. The frontend can pass this metadata back to the
/// server to retrieve only the desired portion of the message (e.g., the HTML body),
/// which significantly reduces bandwidth and latency.
///
/// By abstracting the complexity of MIME part navigation, developers can efficiently
/// retrieve specific parts of an email without handling the low-level IMAP structure.
pub body_meta: Option<Vec<EmailBodyPart>>,
/// Details about how the email was received, if available.
pub received: Option<Received>,
/// The `mid` field is reserved for potential integration with other backend models.
/// For instance, it can be used to store the email index or ID from external services like the Gmail API.
/// This ID could be used for reference or identification purposes in scenarios where an external service
/// provides an identifier for the email in question.
///
/// This field is optional, meaning that it may be `None` if no external service identifier is available.
pub mid: Option<String>,
/// A list of labels applied to the message.
///
/// Each element is a string representing a Gmail label ID (e.g., "INBOX", "UNREAD").
/// This field reflects the current labels associated with the email.
///
/// Note: This field is populated only for Gmail API accounts. For other account types, it will be empty.
pub label_ids: Vec<String>,
}
impl EmailEnvelopeV3 {
pub fn pk(&self) -> String {
format!(
"{}_{}",
self.internal_date.unwrap_or(utc_now!()),
envelope_hash(self.account_id, self.mailbox_id, self.uid)
)
}
pub fn create_envelope_id(&self) -> u64 {
envelope_hash(self.account_id, self.mailbox_id, self.uid)
}
pub fn compute_thread_id(&self) -> u64 {
if self.in_reply_to.is_some() && self.references.as_ref().map_or(false, |r| !r.is_empty()) {
@@ -153,10 +272,10 @@ impl EmailEnvelopeV2 {
account_id: u64,
mailbox_id: u64,
uid: u32,
) -> RustMailerResult<Option<EmailEnvelopeV2>> {
) -> RustMailerResult<Option<EmailEnvelopeV3>> {
secondary_find_impl(
DB_MANAGER.envelope_db(),
EmailEnvelopeV2Key::create_envelope_id,
EmailEnvelopeV3Key::create_envelope_id,
envelope_hash(account_id, mailbox_id, uid),
)
.await
@@ -166,10 +285,10 @@ impl EmailEnvelopeV2 {
account_id: u64,
mailbox_id: u64,
thread_id: u64,
) -> RustMailerResult<Vec<EmailEnvelopeV2>> {
let envelopes = filter_by_secondary_key_impl::<EmailEnvelopeV2>(
) -> RustMailerResult<Vec<EmailEnvelopeV3>> {
let envelopes = filter_by_secondary_key_impl::<EmailEnvelopeV3>(
DB_MANAGER.envelope_db(),
EmailEnvelopeV2Key::thread_id,
EmailEnvelopeV3Key::thread_id,
thread_id,
)
.await?;
@@ -193,16 +312,16 @@ impl EmailEnvelopeV2 {
Ok(result)
}
pub async fn get(envelope_id: u64) -> RustMailerResult<Option<EmailEnvelopeV2>> {
pub async fn get(envelope_id: u64) -> RustMailerResult<Option<EmailEnvelopeV3>> {
secondary_find_impl(
DB_MANAGER.envelope_db(),
EmailEnvelopeV2Key::create_envelope_id,
EmailEnvelopeV3Key::create_envelope_id,
envelope_id,
)
.await
}
pub async fn save_envelopes(envelopes: Vec<EmailEnvelopeV2>) -> RustMailerResult<()> {
pub async fn save_envelopes(envelopes: Vec<EmailEnvelopeV3>) -> RustMailerResult<()> {
with_transaction(DB_MANAGER.envelope_db(), move |rw| {
for mut e in envelopes {
// --- Preprocessing ---
@@ -226,7 +345,7 @@ impl EmailEnvelopeV2 {
);
// --- Store full & minimal envelope ---
rw.insert::<EmailEnvelopeV2>(e)
rw.insert::<EmailEnvelopeV3>(e)
.map_err(|err| raise_error!(format!("{:#?}", err), ErrorCode::InternalError))?;
rw.insert::<MinimalEnvelope>(minimal)
.map_err(|err| raise_error!(format!("{:#?}", err), ErrorCode::InternalError))?;
@@ -272,13 +391,13 @@ impl EmailEnvelopeV2 {
page: u64,
page_size: u64,
desc: bool,
) -> RustMailerResult<DataPage<EmailEnvelopeV2>> {
) -> RustMailerResult<DataPage<EmailEnvelopeV3>> {
paginate_secondary_scan_impl(
DB_MANAGER.envelope_db(),
Some(page),
Some(page_size),
Some(desc),
EmailEnvelopeV2Key::mailbox_id,
EmailEnvelopeV3Key::mailbox_id,
mailbox_id,
)
.await
@@ -298,8 +417,8 @@ impl EmailEnvelopeV2 {
DB_MANAGER.envelope_db(),
move |rw| {
rw.get()
.secondary::<EmailEnvelopeV2>(
EmailEnvelopeV2Key::create_envelope_id,
.secondary::<EmailEnvelopeV3>(
EmailEnvelopeV3Key::create_envelope_id,
envelope_hash(account_id, mailbox_id, uid),
)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
@@ -320,7 +439,7 @@ impl EmailEnvelopeV2 {
.await
.map_err(|e| {
error!(
"Failed to update flags: account_id={}, mailbox_hash={}, uid={}, error={:?}",
"Failed to update flags: account_id={}, mailbox_id={}, uid={}, error={:?}",
account_id, mailbox_id, uid, e
);
e
@@ -335,15 +454,15 @@ impl EmailEnvelopeV2 {
let start_time = Instant::now();
loop {
let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
let to_delete: Vec<EmailEnvelopeV2> = rw
let to_delete: Vec<EmailEnvelopeV3> = rw
.scan()
.secondary(EmailEnvelopeV2Key::mailbox_id)
.secondary(EmailEnvelopeV3Key::mailbox_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.start_with(mailbox_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.take(BATCH_SIZE)
.filter_map(Result::ok) // filter only Ok values
.filter(|e: &EmailEnvelopeV2| e.account_id == account_id)
.filter(|e: &EmailEnvelopeV3| e.account_id == account_id)
.collect();
Ok(to_delete)
})
@@ -356,7 +475,7 @@ impl EmailEnvelopeV2 {
}
info!(
"Finished deleting envelopes for mailbox_hash={} account_id={} total_deleted={} in {:?}",
"Finished deleting envelopes for mailbox_id={} account_id={} total_deleted={} in {:?}",
mailbox_id,
account_id,
total_deleted,
@@ -374,7 +493,7 @@ impl EmailEnvelopeV2 {
let key = envelope_hash(account_id, mailbox_id, *uid);
delete_impl(DB_MANAGER.envelope_db(), move |rw| {
rw.get()
.secondary::<EmailEnvelopeV2>(EmailEnvelopeV2Key::create_envelope_id, key)
.secondary::<EmailEnvelopeV3>(EmailEnvelopeV3Key::create_envelope_id, key)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.ok_or_else(|| {
raise_error!("envelope missing".into(), ErrorCode::InternalError)
@@ -391,9 +510,9 @@ impl EmailEnvelopeV2 {
let start_time = Instant::now();
loop {
let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
let to_delete: Vec<EmailEnvelopeV2> = rw
let to_delete: Vec<EmailEnvelopeV3> = rw
.scan()
.secondary(EmailEnvelopeV2Key::account_id)
.secondary(EmailEnvelopeV3Key::account_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.start_with(account_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
@@ -484,3 +603,71 @@ impl From<EmailEnvelopeV2> for EmailEnvelope {
}
}
}
impl From<EmailEnvelopeV2> for EmailEnvelopeV3 {
fn from(value: EmailEnvelopeV2) -> Self {
Self {
account_id: value.account_id,
mailbox_id: value.mailbox_id,
mailbox_name: value.mailbox_name,
uid: value.uid,
internal_date: value.internal_date,
size: value.size,
flags: value.flags,
flags_hash: value.flags_hash,
bcc: value.bcc,
cc: value.cc,
date: value.date,
from: value.from,
in_reply_to: value.in_reply_to,
sender: value.sender,
return_address: value.return_address,
message_id: value.message_id,
subject: value.subject,
thread_name: value.thread_name,
thread_id: value.thread_id,
mime_version: value.mime_version,
references: value.references,
reply_to: value.reply_to,
to: value.to,
attachments: value.attachments,
body_meta: value.body_meta,
received: value.received,
mid: None,
label_ids: vec![],
}
}
}
impl From<EmailEnvelopeV3> for EmailEnvelopeV2 {
fn from(value: EmailEnvelopeV3) -> Self {
Self {
account_id: value.account_id,
mailbox_id: value.mailbox_id,
mailbox_name: value.mailbox_name,
uid: value.uid,
internal_date: value.internal_date,
size: value.size,
flags: value.flags,
flags_hash: value.flags_hash,
bcc: value.bcc,
cc: value.cc,
date: value.date,
from: value.from,
in_reply_to: value.in_reply_to,
sender: value.sender,
return_address: value.return_address,
message_id: value.message_id,
subject: value.subject,
thread_name: value.thread_name,
thread_id: value.thread_id,
mime_version: value.mime_version,
references: value.references,
reply_to: value.reply_to,
to: value.to,
attachments: value.attachments,
body_meta: value.body_meta,
received: value.received,
}
}
}
+1
View File
@@ -4,3 +4,4 @@
pub mod disk;
pub mod imap;
pub mod vendor;
+6
View File
@@ -0,0 +1,6 @@
// Copyright © 2025 rustmailer.com
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
pub mod model;
pub mod sync;
+67
View File
@@ -0,0 +1,67 @@
// Copyright © 2025 rustmailer.com
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct HistoryList {
#[serde(default)]
pub history: Vec<History>,
#[serde(rename = "historyId")]
pub history_id: String,
#[serde(rename = "nextPageToken")]
#[serde(default, skip_serializing_if = "Option::is_none")]
pub next_page_token: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct MessageIndex {
pub id: String,
#[serde(default, rename = "labelIds")]
pub label_ids: Vec<String>,
#[serde(rename = "threadId")]
pub thread_id: String,
}
#[derive(Debug, Serialize, Deserialize, Clone, Default)]
pub struct MessageObject {
pub message: MessageIndex,
}
#[derive(Debug, Serialize, Deserialize, Clone, Default)]
pub struct LabelMessageObject {
#[serde(rename = "labelIds")]
#[serde(default, skip_serializing_if = "Option::is_none")]
pub label_ids: Option<Vec<String>>,
pub message: MessageIndex,
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct History {
pub id: String,
#[serde(rename = "labelsAdded")]
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub labels_added: Vec<LabelMessageObject>,
#[serde(rename = "labelsRemoved")]
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub labels_removed: Vec<LabelMessageObject>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub messages: Vec<MessageIndex>,
#[serde(rename = "messagesAdded")]
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub messages_added: Vec<MessageObject>,
#[serde(rename = "messagesDeleted")]
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub messages_deleted: Vec<MessageObject>,
}
impl History {
pub fn has_changes(&self) -> bool {
!(self.labels_added.is_empty()
&& self.labels_removed.is_empty()
&& self.messages_added.is_empty()
&& self.messages_deleted.is_empty())
}
}
+73
View File
@@ -0,0 +1,73 @@
// Copyright © 2025 rustmailer.com
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use serde::{Deserialize, Serialize};
use crate::modules::cache::vendor::gmail::sync::labels::GmailLabels;
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct LabelList {
pub labels: Vec<Label>,
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct Label {
pub id: String,
pub name: String,
#[serde(rename = "type")]
pub label_type: String, // "system" or "user"
#[serde(rename = "labelListVisibility")]
pub label_list_visibility: Option<String>,
#[serde(rename = "messageListVisibility")]
pub message_list_visibility: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct LabelDetail {
/// Optional color configuration for user-created labels
#[serde(default, skip_serializing_if = "Option::is_none")]
pub color: Option<serde_json::Value>,
/// Unique identifier of the label
pub id: String,
/// Visibility of the label in Gmail's label list
#[serde(rename = "labelListVisibility")]
#[serde(default, skip_serializing_if = "Option::is_none")]
pub label_list_visibility: Option<String>,
/// Visibility of messages with this label in Gmail's message list
#[serde(rename = "messageListVisibility")]
#[serde(default, skip_serializing_if = "Option::is_none")]
pub message_list_visibility: Option<String>,
/// Total number of messages with this label
#[serde(rename = "messagesTotal")]
pub messages_total: u32,
/// Number of unread messages with this label
#[serde(rename = "messagesUnread")]
pub messages_unread: u32,
/// Display name of the label
pub name: String,
/// Total number of threads with this label
#[serde(rename = "threadsTotal")]
#[serde(default, skip_serializing_if = "Option::is_none")]
pub threads_total: Option<i64>,
/// Number of unread threads with this label
#[serde(rename = "threadsUnread")]
#[serde(default, skip_serializing_if = "Option::is_none")]
pub threads_unread: Option<i64>,
/// Type of the label ("user" or "system")
#[serde(rename = "type")]
#[serde(default, skip_serializing_if = "Option::is_none")]
pub type_: Option<String>,
}
impl From<LabelDetail> for GmailLabels {
fn from(label: LabelDetail) -> Self {
Self {
id: 0,
account_id: 0,
name: label.name,
exists: label.messages_total,
unseen: label.messages_unread,
label_id: label.id,
}
}
}
+144
View File
@@ -0,0 +1,144 @@
// Copyright © 2025 rustmailer.com
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use chrono::DateTime;
use serde::{Deserialize, Serialize};
use crate::{
modules::{
cache::vendor::gmail::sync::envelope::GmailEnvelope,
common::Addr,
error::{code::ErrorCode, RustMailerError},
},
raise_error,
};
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct MessageIndex {
pub id: String,
#[serde(rename = "threadId")]
pub thread_id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct MessageList {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub messages: Option<Vec<MessageIndex>>,
#[serde(rename = "nextPageToken")]
#[serde(default, skip_serializing_if = "Option::is_none")]
pub next_page_token: Option<String>,
#[serde(rename = "resultSizeEstimate")]
#[serde(default, skip_serializing_if = "Option::is_none")]
pub result_size_estimate: Option<i64>,
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct MessageMeta {
pub id: String,
#[serde(rename = "threadId")]
pub thread_id: String,
#[serde(rename = "historyId")]
pub history_id: String,
#[serde(rename = "internalDate")]
pub internal_date: String,
#[serde(rename = "labelIds")]
pub label_ids: Vec<String>,
pub payload: Payload,
#[serde(rename = "sizeEstimate")]
pub size_estimate: u32,
pub snippet: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct Payload {
#[serde(rename = "mimeType")]
pub mime_type: Option<String>,
pub headers: Vec<Header>,
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct Header {
pub name: String,
pub value: String,
}
impl TryFrom<MessageMeta> for GmailEnvelope {
type Error = RustMailerError;
fn try_from(value: MessageMeta) -> Result<Self, Self::Error> {
let payload = value.payload;
let mut envelope = Self {
account_id: 0,
label_id: 0,
label_name: "".into(),
id: value.id,
internal_date: value.internal_date.parse().map_err(|e| {
raise_error!(
format!("Failed to parse internal_date: {}", e),
ErrorCode::InternalError
)
})?,
size: value.size_estimate,
bcc: None,
cc: None,
date: None,
from: None,
in_reply_to: None,
sender: None,
message_id: None,
subject: None,
thread_id: 0,
mime_version: None,
references: None,
reply_to: None,
to: None,
snippet: value.snippet,
history_id: value.history_id,
gmail_thread_id: value.thread_id,
label_ids: value.label_ids,
};
for header in payload.headers {
match header.name.as_str() {
"Date" => {
let dt = DateTime::parse_from_rfc2822(&header.value).map_err(|e| {
raise_error!(
format!("Failed to parse Date: {}", e),
ErrorCode::InternalError
)
})?;
envelope.date = Some(dt.timestamp_millis());
}
"From" => envelope.from = Some(Addr::parse(&header.value)),
"Sender" => envelope.sender = Some(Addr::parse(&header.value)),
"Reply-To" => envelope.reply_to = Some(Self::parse_addr_list(&header.value)),
"In-Reply-To" => {
envelope.in_reply_to = Some(Self::clean_angle_brackets(&header.value).into())
}
"Message-ID" => {
envelope.message_id = Some(Self::clean_angle_brackets(&header.value).into())
}
"Mime-Version" => envelope.mime_version = Some(header.value),
"References" => {
envelope.references = Some(
header
.value
.split_whitespace()
.map(Self::clean_angle_brackets)
.filter(|id| !id.is_empty())
.map(|id| id.to_string())
.collect(),
)
}
"Subject" => envelope.subject = Some(header.value),
"To" => envelope.to = Some(Self::parse_addr_list(&header.value)),
"Bcc" => envelope.bcc = Some(Self::parse_addr_list(&header.value)),
"Cc" => envelope.cc = Some(Self::parse_addr_list(&header.value)),
_ => {}
}
}
Ok(envelope)
}
}
+7
View File
@@ -0,0 +1,7 @@
// Copyright © 2025 rustmailer.com
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
pub mod history;
pub mod labels;
pub mod messages;
+152
View File
@@ -0,0 +1,152 @@
// Copyright © 2025 rustmailer.com
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use crate::{
modules::{
cache::vendor::gmail::model::{
history::HistoryList,
labels::{LabelDetail, LabelList},
messages::{MessageList, MessageMeta},
},
error::{code::ErrorCode, RustMailerResult},
hook::http::HttpClient,
oauth2::token::OAuth2AccessToken,
},
raise_error,
};
pub struct GmailClient;
impl GmailClient {
async fn get_access_token(account_id: u64) -> RustMailerResult<String> {
let record = OAuth2AccessToken::get(account_id).await?;
record.and_then(|r| r.access_token).ok_or_else(|| {
raise_error!(
"Gmail API requires an OAuth2 access token, but authorization is incomplete."
.into(),
ErrorCode::MissingConfiguration
)
})
}
pub async fn list_labels(
account_id: u64,
use_proxy: Option<u64>,
) -> RustMailerResult<LabelList> {
let url = "https://gmail.googleapis.com/gmail/v1/users/me/labels";
let client = HttpClient::new(use_proxy).await?;
let access_token = Self::get_access_token(account_id).await?;
let value = client.get(url, &access_token).await?;
let list = serde_json::from_value::<LabelList>(value)
.map_err(|e| raise_error!(format!(
"Failed to deserialize Gmail API response into LabelList: {:#?}. Possible model mismatch or API change.",
e
), ErrorCode::InternalError))?;
Ok(list)
}
pub async fn get_label(
account_id: u64,
use_proxy: Option<u64>,
label_id: &str,
) -> RustMailerResult<LabelDetail> {
let url = format!(
"https://gmail.googleapis.com/gmail/v1/users/me/labels/{}",
label_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 detail = serde_json::from_value::<LabelDetail>(value)
.map_err(|e| raise_error!(format!(
"Failed to deserialize Gmail API response into LabelDetail: {:#?}. Possible model mismatch or API change.",
e
), ErrorCode::InternalError))?;
Ok(detail)
}
pub async fn list_messages(
account_id: u64,
use_proxy: Option<u64>,
label_id: &str,
page_token: Option<String>,
after: Option<&str>,
max_results: u32,
) -> RustMailerResult<MessageList> {
let mut url = format!(
"https://gmail.googleapis.com/gmail/v1/users/me/messages?labelIds={}&maxResults={}",
label_id, max_results
);
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::<MessageList>(value).map_err(|e| {
raise_error!(
format!(
"Failed to deserialize Gmail API response into MessageList: {:#?}. Possible model mismatch or API change.",
e
),
ErrorCode::InternalError
)
})?;
Ok(list)
}
pub async fn get_messages(
account_id: u64,
use_proxy: Option<u64>,
mid: &str,
) -> RustMailerResult<MessageMeta> {
let url = format!(
"https://gmail.googleapis.com/gmail/v1/users/me/messages/{}?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",
mid
);
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 message = serde_json::from_value::<MessageMeta>(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(message)
}
pub async fn list_history(
account_id: u64,
use_proxy: Option<u64>,
label_id: &str,
start_history_id: &str,
page_token: Option<&str>,
max_results: u32,
) -> RustMailerResult<HistoryList> {
let mut url = format!(
"https://gmail.googleapis.com/gmail/v1/users/me/history?labelIds={}&maxResults={}&startHistoryId={}",
label_id, max_results, start_history_id
);
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::<HistoryList>(value)
.map_err(|e| raise_error!(format!(
"Failed to deserialize Gmail API response into ListMessagesResponse: {:#?}. Possible model mismatch or API change.",
e
), ErrorCode::InternalError))?;
Ok(list)
}
}
+299
View File
@@ -0,0 +1,299 @@
// Copyright © 2025 rustmailer.com
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use std::time::Instant;
use crate::{
calculate_hash, id,
modules::{
cache::imap::{
address::AddressEntity,
thread::{EmailThread, EmailThreadKey},
v2::EmailEnvelopeV3,
},
common::Addr,
database::{
batch_delete_impl, delete_impl, manager::DB_MANAGER, secondary_find_impl, upsert_impl,
with_transaction,
},
error::{code::ErrorCode, RustMailerResult},
utils::envelope_hash_from_id,
},
raise_error,
};
use native_db::*;
use native_model::{native_model, Model};
use poem_openapi::Object;
use serde::{Deserialize, Serialize};
use tracing::info;
#[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize, Object)]
#[native_model(id = 6, version = 1)]
#[native_db(primary_key(pk -> String), secondary_key(create_envelope_id -> u64, unique))]
pub struct GmailEnvelope {
/// The ID of the account owning this email within RustMailer.
/// This corresponds to the local RustMailer account ID, not the Gmail account itself.
#[secondary_key]
pub account_id: u64,
/// The unique internal identifier of the label within RustMailers local cache
/// where the email is associated.
///
/// This is **not** the Gmail label ID; it only references the label stored locally.
#[secondary_key]
pub label_id: u64,
/// This is the human-readable label used in Gmail to categorize emails.
pub label_name: String,
/// The Gmail message ID as returned by the `messages.list` or `messages.get` API.
/// This ID uniquely identifies the email within the account and mailbox.
pub id: String,
/// The date and time when Gmail received the email, as a Unix timestamp in milliseconds.
/// Corresponds to the API field `internalDate`. May be `None` if unavailable.
pub internal_date: i64,
/// The size of the email in bytes. Corresponds to the `sizeEstimate` from the API.
pub size: u32,
/// Blind carbon copy (BCC) recipient(s), if any. Each `Addr` contains name and email.
pub bcc: Option<Vec<Addr>>,
/// Carbon copy (CC) recipient(s), if any. Each `Addr` contains name and email.
pub cc: Option<Vec<Addr>>,
/// The date the email was sent, as a Unix timestamp in milliseconds.
/// Extracted from the `Date` header if present. May be `None` if the header is missing or unparseable.
pub date: Option<i64>,
/// The sender's address, as specified in the `From` header.
pub from: Option<Addr>,
/// The message ID of the email to which this email is a reply, if applicable.
/// Corresponds to the `In-Reply-To` header.
pub in_reply_to: Option<String>,
/// The actual sender's address, if different from the `From` field.
/// Extracted from the `Sender` header, if present.
pub sender: Option<Addr>,
/// The globally unique message ID of the email.
/// Corresponds to the `Message-ID` header. Useful for threading and deduplication.
pub message_id: Option<String>,
/// The subject of the email, if present.
pub subject: Option<String>,
/// The identifier of the thread this email belongs to.
/// Derived from `in_reply_to`, `references`, or `message_id`.
#[secondary_key]
pub thread_id: u64,
/// The MIME version of the email (e.g., "1.0"), if specified.
/// Corresponds to the `Mime-Version` header.
pub mime_version: Option<String>,
/// List of message IDs referenced by this email, used for threading.
/// Corresponds to the `References` header.
pub references: Option<Vec<String>>,
/// The address(es) to which replies should be sent, if specified.
/// Corresponds to the `Reply-To` header.
pub reply_to: Option<Vec<Addr>>,
/// Primary recipient(s) of the email, corresponding to the `To` header.
pub to: Option<Vec<Addr>>,
/// A short snippet (preview) of the email body.
/// Corresponds to the API `snippet` field. Typically the first few hundred characters of the message body.
pub snippet: Option<String>,
/// The Gmail history ID associated with this email.
/// Useful for incremental synchronization via `history.list`.
pub history_id: String,
/// The Gmail API thread ID associated with this email.
pub gmail_thread_id: String,
/// A list of labels applied to the message.
///
/// Each element is a string representing a Gmail label ID (e.g., "INBOX", "UNREAD").
/// This field reflects the current labels associated with the email.
pub label_ids: Vec<String>,
}
impl GmailEnvelope {
pub fn pk(&self) -> String {
format!(
"{}_{}",
self.internal_date,
envelope_hash_from_id(self.account_id, self.label_id, &self.id)
)
}
pub fn create_envelope_id(&self) -> u64 {
envelope_hash_from_id(self.account_id, self.label_id, &self.id)
}
pub fn compute_thread_id(&self) -> u64 {
if self.in_reply_to.is_some() && self.references.as_ref().map_or(false, |r| !r.is_empty()) {
return calculate_hash!(&self.references.as_ref().unwrap()[0]);
}
if let Some(message_id) = self.message_id.as_ref() {
return calculate_hash!(message_id);
}
id!(128)
}
pub async fn delete(account_id: u64, label_id: u64, mid: &str) -> RustMailerResult<()> {
let mid = mid.to_string();
delete_impl(DB_MANAGER.envelope_db(), move |rw| {
rw.get()
.secondary::<GmailEnvelope>(
GmailEnvelopeKey::create_envelope_id,
envelope_hash_from_id(account_id, label_id, &mid),
)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.ok_or_else(|| {
raise_error!("gmail envelope missing".into(), ErrorCode::InternalError)
})
})
.await
}
pub async fn find(
account_id: u64,
label_id: u64,
mid: &str,
) -> RustMailerResult<Option<GmailEnvelope>> {
secondary_find_impl(
DB_MANAGER.envelope_db(),
GmailEnvelopeKey::create_envelope_id,
envelope_hash_from_id(account_id, label_id, mid),
)
.await
}
pub async fn upsert(envelope: GmailEnvelope) -> RustMailerResult<()> {
upsert_impl(DB_MANAGER.envelope_db(), envelope).await
}
pub async fn save_envelopes(envelopes: Vec<GmailEnvelope>) -> RustMailerResult<()> {
with_transaction(DB_MANAGER.envelope_db(), move |rw| {
for mut e in envelopes {
// --- Preprocessing ---
let address_entities = AddressEntity::extract2(&e);
e.thread_id = e.compute_thread_id();
let thread = EmailThread::new(
e.thread_id,
e.create_envelope_id(),
e.account_id,
e.label_id,
Some(e.internal_date),
e.date,
);
// --- Store envelope ---
rw.insert::<GmailEnvelope>(e)
.map_err(|err| raise_error!(format!("{:#?}", err), ErrorCode::InternalError))?;
// --- Thread upsert ---
match rw
.get()
.secondary::<EmailThread>(EmailThreadKey::thread_id, thread.thread_id)
.map_err(|err| raise_error!(format!("{:#?}", err), ErrorCode::InternalError))?
{
Some(current) => {
// Only replace if current.internal_date is older than new internal_date
if current.need_update(&thread) {
rw.remove(current).map_err(|err| {
raise_error!(format!("{:#?}", err), ErrorCode::InternalError)
})?;
rw.insert::<EmailThread>(thread).map_err(|err| {
raise_error!(format!("{:#?}", err), ErrorCode::InternalError)
})?;
}
}
None => {
rw.insert::<EmailThread>(thread).map_err(|err| {
raise_error!(format!("{:#?}", err), ErrorCode::InternalError)
})?;
}
}
// --- Store address entities ---
for addr in address_entities {
rw.insert::<AddressEntity>(addr).map_err(|err| {
raise_error!(format!("{:#?}", err), ErrorCode::InternalError)
})?;
}
}
Ok(())
})
.await
}
pub fn clean_angle_brackets(s: &str) -> &str {
s.trim().trim_matches(|c| c == '<' || c == '>')
}
pub fn parse_addr_list(s: &str) -> Vec<Addr> {
s.split(',')
.map(|part| part.trim())
.filter(|part| !part.is_empty())
.map(Addr::parse)
.collect()
}
pub async fn clean_label_envelopes(account_id: u64, label_id: u64) -> RustMailerResult<()> {
const BATCH_SIZE: usize = 200;
let mut total_deleted = 0usize;
let start_time = Instant::now();
loop {
let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
let to_delete: Vec<GmailEnvelope> = rw
.scan()
.secondary(GmailEnvelopeKey::label_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.start_with(label_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.take(BATCH_SIZE)
.filter_map(Result::ok) // filter only Ok values
.filter(|e: &GmailEnvelope| e.account_id == account_id)
.collect();
Ok(to_delete)
})
.await?;
total_deleted += deleted;
// If this batch is empty, break the loop
if deleted == 0 {
break;
}
}
info!(
"Finished deleting gmail envelopes for label_id={} account_id={} total_deleted={} in {:?}",
label_id,
account_id,
total_deleted,
start_time.elapsed()
);
Ok(())
}
}
impl From<GmailEnvelope> for EmailEnvelopeV3 {
fn from(value: GmailEnvelope) -> Self {
Self {
account_id: value.account_id,
mailbox_id: value.label_id,
mailbox_name: value.label_name,
uid: 0,
internal_date: Some(value.internal_date),
size: value.size,
flags: vec![],
flags_hash: 0,
bcc: value.bcc,
cc: value.cc,
date: value.date,
from: value.from,
in_reply_to: value.in_reply_to,
sender: value.sender,
return_address: None,
message_id: value.message_id,
subject: value.subject,
thread_name: None,
thread_id: value.thread_id,
mime_version: value.mime_version,
references: value.references,
reply_to: value.reply_to,
to: value.to,
attachments: None,
body_meta: None,
received: None,
mid: Some(value.id),
label_ids: value.label_ids,
}
}
}
+258
View File
@@ -0,0 +1,258 @@
// Copyright © 2025 rustmailer.com
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use std::sync::Arc;
use tokio::{sync::Semaphore, task::JoinHandle};
use tracing::error;
use crate::{
modules::{
account::{status::AccountRunningState, v2::AccountV2},
cache::vendor::gmail::model::messages::MessageMeta,
cache::vendor::gmail::sync::{
client::GmailClient, envelope::GmailEnvelope, labels::GmailLabels,
},
error::{code::ErrorCode, RustMailerResult},
},
raise_error,
};
const ENVELOPE_BATCH_SIZE: u32 = 500;
pub async fn fetch_and_save_since_date(
account: &AccountV2,
date: &str,
label: &GmailLabels,
initial: bool,
) -> RustMailerResult<(usize, Option<String>)> {
let page_size = ENVELOPE_BATCH_SIZE;
// 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;
let use_proxy = account.use_proxy;
// Gmail API pagination relies on pageToken.
// Each page returns message IDs, and we still need to fetch message details individually.
let mut page_token: Option<String> = None;
let mut page = 1; // Used only for tracking sync progress
let semaphore = Arc::new(Semaphore::new(10));
let mut max_history_id = None;
loop {
let resp = GmailClient::list_messages(
account_id,
use_proxy,
&label.label_id,
page_token,
Some(date),
ENVELOPE_BATCH_SIZE,
)
.await?;
// The total number of messages can only be retrieved via an API query
if page == 1 && initial {
let total = resp.result_size_estimate;
if let Some(total) = total {
let total_batches = (total as u32).div_ceil(page_size);
AccountRunningState::set_initial_current_syncing_folder(
account_id,
label.name.clone(),
total_batches,
)
.await?;
}
}
// Update page_token returned by Gmail API
page_token = resp.next_page_token;
// Concurrently fetch message details for this page, with concurrency limited to 10
if let Some(messages) = resp.messages {
let mut batch_messages = Vec::with_capacity(ENVELOPE_BATCH_SIZE as usize);
if initial {
AccountRunningState::set_current_sync_batch_number(account_id, page).await?;
}
let mut handles: Vec<JoinHandle<RustMailerResult<MessageMeta>>> = Vec::new();
for msg in messages {
match semaphore.clone().acquire_owned().await {
Ok(permit) => {
let handle: JoinHandle<RustMailerResult<MessageMeta>> =
tokio::spawn(async move {
// Permit will be released automatically when this task finishes
let _permit = permit;
GmailClient::get_messages(account_id, use_proxy, &msg.id).await
});
handles.push(handle);
}
Err(err) => error!("Failed to acquire semaphore permit, error: {:#?}", err),
}
}
for handle in handles {
match handle.await {
Ok(Ok(meta)) => batch_messages.push(meta),
Ok(Err(e)) => return Err(e),
Err(join_err) => {
return Err(raise_error!(
format!("tokio task join error: {:?}", join_err),
ErrorCode::InternalError
));
}
}
}
// All message details for this batch are collected, now convert and save them
let envelopes: Vec<GmailEnvelope> = batch_messages
.into_iter()
.map(|m| {
let mut envelope: GmailEnvelope = m.try_into()?;
envelope.account_id = account_id;
envelope.label_id = label.id;
envelope.label_name = label.name.clone();
Ok(envelope)
})
.collect::<RustMailerResult<Vec<GmailEnvelope>>>()?;
inserted_count += envelopes.len();
max_history_id = compute_max_history_id(&envelopes);
GmailEnvelope::save_envelopes(envelopes).await?;
}
// Break if API response has no next page
if page_token.is_none() {
break;
}
page += 1;
}
Ok((inserted_count, max_history_id))
}
pub async fn fetch_and_save_full_label(
account: &AccountV2,
label: &GmailLabels,
total: u32,
initial: bool,
) -> RustMailerResult<(usize, Option<String>)> {
let page_size = ENVELOPE_BATCH_SIZE;
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;
let use_proxy = account.use_proxy;
// If this is the first synchronization, set the initial state for the account
if initial {
AccountRunningState::set_initial_current_syncing_folder(
account_id,
label.name.clone(),
total_batches,
)
.await?;
}
// Gmail API pagination relies on pageToken.
// Each page returns message IDs, and we still need to fetch message details individually.
let mut page_token: Option<String> = None;
let mut page = 1; // Used only for tracking sync progress
let semaphore = Arc::new(Semaphore::new(10));
let mut max_history_id = None;
loop {
let resp = GmailClient::list_messages(
account_id,
use_proxy,
&label.label_id,
page_token,
None,
ENVELOPE_BATCH_SIZE,
)
.await?;
// Update page_token returned by Gmail API
page_token = resp.next_page_token;
// Concurrently fetch message details for this page, with concurrency limited to 10
if let Some(messages) = resp.messages {
let mut batch_messages = Vec::with_capacity(ENVELOPE_BATCH_SIZE as usize);
if initial {
AccountRunningState::set_current_sync_batch_number(account_id, page).await?;
}
let mut handles: Vec<JoinHandle<RustMailerResult<MessageMeta>>> = Vec::new();
for msg in messages {
match semaphore.clone().acquire_owned().await {
Ok(permit) => {
let handle: JoinHandle<RustMailerResult<MessageMeta>> =
tokio::spawn(async move {
// Permit will be released automatically when this task finishes
let _permit = permit;
GmailClient::get_messages(account_id, use_proxy, &msg.id).await
});
handles.push(handle);
}
Err(err) => error!("Failed to acquire semaphore permit, error: {:#?}", err),
}
}
for handle in handles {
match handle.await {
Ok(Ok(meta)) => batch_messages.push(meta),
Ok(Err(e)) => return Err(e),
Err(join_err) => {
return Err(raise_error!(
format!("tokio task join error: {:?}", join_err),
ErrorCode::InternalError
));
}
}
}
// All message details for this batch are collected, now convert and save them
let envelopes: Vec<GmailEnvelope> = batch_messages
.into_iter()
.map(|m| {
let mut envelope: GmailEnvelope = m.try_into()?;
envelope.account_id = account_id;
envelope.label_id = label.id;
envelope.label_name = label.name.clone();
Ok(envelope)
})
.collect::<RustMailerResult<Vec<GmailEnvelope>>>()?;
inserted_count += envelopes.len();
max_history_id = compute_max_history_id(&envelopes);
GmailEnvelope::save_envelopes(envelopes).await?;
}
// Break if API response has no next page
if page_token.is_none() {
break;
}
page += 1;
}
Ok((inserted_count, max_history_id))
}
fn max_history_id_fallback(a: &str, b: &str) -> String {
// Try to parse as u64
match (a.parse::<u64>(), b.parse::<u64>()) {
(Ok(a_num), Ok(b_num)) => {
if a_num >= b_num {
a.to_string()
} else {
b.to_string()
}
}
// If parsing fails, fall back to length + lexicographical comparison
_ => {
if a.len() > b.len() {
a.to_string()
} else if b.len() > a.len() {
b.to_string()
} else {
// Same length, compare lexicographically
if a >= b {
a.to_string()
} else {
b.to_string()
}
}
}
}
}
fn compute_max_history_id(envelopes: &[GmailEnvelope]) -> Option<String> {
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.as_str(), curr),
None => curr.to_string(),
})
})
}
+239
View File
@@ -0,0 +1,239 @@
// Copyright © 2025 rustmailer.com
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use ahash::{AHashSet, HashSet};
use tokio::task::JoinHandle;
use tracing::{info, warn};
use crate::{
modules::{
account::v2::AccountV2,
cache::vendor::gmail::{
model::history::History,
sync::{
cleanup_single_label,
client::GmailClient,
envelope::GmailEnvelope,
labels::{GmailCheckPoint, GmailLabels},
rebuild::rebuild_single_label_cache,
},
},
error::{code::ErrorCode, RustMailerError, RustMailerResult},
},
raise_error,
};
pub async fn handle_history(
account: &AccountV2,
local_labels: &[GmailLabels],
remote_labels: &[GmailLabels],
) -> RustMailerResult<()> {
let account_id = account.id;
let use_proxy = account.use_proxy.clone();
let remote_labels = find_existing_remote_labels(local_labels, remote_labels);
for remote in remote_labels {
let checkpoint = GmailCheckPoint::get(remote.id).await?;
let mut page_token = None;
loop {
let mut list = match GmailClient::list_history(
account_id,
use_proxy.clone(),
&remote.label_id,
&checkpoint.history_id,
page_token.as_deref(),
100, // 100 items per page
)
.await
{
Ok(list) => list,
Err(error) => match error {
RustMailerError::Generic {
message,
location: _,
code,
} => {
if code == ErrorCode::GmailApiInvalidHistoryId {
handle_invalid_history_id(account, &remote).await?;
continue;
} else {
return Err(raise_error!(message, code));
}
}
},
};
page_token = list.next_page_token.take();
let history_list: Vec<History> = list
.history
.into_iter()
.filter(|h| h.has_changes())
.collect();
apply_history(account_id, use_proxy, &remote, history_list).await?;
if page_token.is_none() {
GmailCheckPoint::new(account_id, remote.id, list.history_id)
.save()
.await?;
break;
}
}
GmailLabels::upsert(remote).await?;
}
Ok(())
}
pub fn find_existing_remote_labels(
local_labels: &[GmailLabels],
remote_labels: &[GmailLabels],
) -> Vec<GmailLabels> {
let local_ids: AHashSet<_> = local_labels.iter().map(|l| &l.id).collect();
remote_labels
.iter()
.filter(|remote| local_ids.contains(&remote.id))
.cloned()
.collect()
}
pub async fn apply_history(
account_id: u64,
use_proxy: Option<u64>,
label: &GmailLabels,
history_list: Vec<History>,
) -> RustMailerResult<()> {
for history in history_list {
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<String> = current.label_ids.into_iter().collect();
if let Some(to_add) = &item.label_ids {
merged.extend(to_add.iter().cloned());
}
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
);
}
}
}
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 {
if to_remove.contains(&label.id.to_string()) {
GmailEnvelope::delete(account_id, label.id, &current.id).await?;
} else {
current.label_ids.retain(|id| !to_remove.contains(id));
GmailEnvelope::upsert(current).await?;
}
}
}
None => {
warn!(
"Message {} not found in local cache, cannot merge labels.",
item.message.id
);
}
}
}
let len = history.messages_added.len();
let mut handles: Vec<JoinHandle<Option<GmailEnvelope>>> = Vec::with_capacity(len);
for item in history.messages_added {
let account_id = account_id;
let use_proxy = use_proxy.clone();
let label = label.clone();
handles.push(tokio::spawn(async move {
if !item.message.label_ids.contains(&label.label_id) {
return None;
}
let message_data = match GmailClient::get_messages(
account_id,
use_proxy.clone(),
&item.message.id,
)
.await
{
Ok(msg) => msg,
Err(_) => return None,
};
if !message_data.label_ids.contains(&label.label_id) {
return None;
}
let mut envelope: GmailEnvelope = match message_data.try_into() {
Ok(env) => env,
Err(_) => return None,
};
envelope.account_id = account_id;
envelope.label_id = label.id;
envelope.label_name = label.name.clone();
Some(envelope)
}));
}
let mut messages_added = Vec::with_capacity(len);
for handle in handles {
if let Ok(Some(envelope)) = handle.await {
messages_added.push(envelope);
}
}
if !messages_added.is_empty() {
GmailEnvelope::save_envelopes(messages_added).await?;
}
//Deletion events are temporarily not handled
// for item in history.messages_deleted {
// if item.message.label_ids.contains(&label.label_id) {
// let mid = item.message.id;
// let message_data =
// GmailClient::get_messages(account_id, use_proxy.clone(), &mid).await?;
// // We can directly delete it here
// if message_data.label_ids.contains(&label.label_id) {
// GmailEnvelope::delete(account_id, label.id, &message_data.id).await?;
// }
// }
// }
}
Ok(())
}
async fn handle_invalid_history_id(
account: &AccountV2,
label: &GmailLabels,
) -> RustMailerResult<()> {
info!(
"Account {}: Invalid history ID detected for label '{}'. Rebuilding local state...",
account.id, label.name
);
cleanup_single_label(account, label).await?;
info!(
"Account {}: Cleaned up local data for label '{}'",
account.id, label.name
);
GmailLabels::upsert(label.clone()).await?;
info!(
"Account {}: Upserted label '{}' into local database",
account.id, label.name
);
rebuild_single_label_cache(account, label).await?;
info!(
"Account {}: Rebuilt local cache for label '{}'",
account.id, label.name
);
Ok(())
}
+185
View File
@@ -0,0 +1,185 @@
// Copyright © 2025 rustmailer.com
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use itertools::Itertools;
use native_db::*;
use native_model::{native_model, Model};
use poem_openapi::Object;
use serde::{Deserialize, Serialize};
use crate::{
modules::{
cache::imap::mailbox::MailBox,
database::{
async_find_impl, batch_delete_impl, batch_insert_impl, delete_impl,
filter_by_secondary_key_impl, insert_impl, manager::DB_MANAGER, upsert_impl,
},
error::{code::ErrorCode, RustMailerResult},
},
raise_error, utc_now,
};
#[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize, Object)]
#[native_model(id = 7, version = 1)]
#[native_db]
pub struct GmailLabels {
/// This `id` **must be a hash value constructed from both `account_id` and `label_id`**,
/// ensuring global uniqueness across all accounts and labels.
#[primary_key]
pub id: u64,
#[secondary_key]
pub account_id: u64,
pub name: String,
pub exists: u32,
pub unseen: u32,
pub label_id: String,
}
impl GmailLabels {
pub async fn save(&self) -> RustMailerResult<()> {
insert_impl(DB_MANAGER.envelope_db(), self.to_owned()).await
}
pub async fn upsert(label: GmailLabels) -> RustMailerResult<()> {
upsert_impl(DB_MANAGER.envelope_db(), label).await
}
pub async fn batch_insert(labels: &[GmailLabels]) -> RustMailerResult<()> {
batch_insert_impl(DB_MANAGER.envelope_db(), labels.to_vec()).await
}
pub async fn delete(id: u64) -> RustMailerResult<()> {
delete_impl(DB_MANAGER.envelope_db(), move |rw| {
rw.get()
.primary::<GmailLabels>(id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.ok_or_else(|| raise_error!("label missing".into(), ErrorCode::InternalError))
})
.await
}
pub async fn list_all(account_id: u64) -> RustMailerResult<Vec<GmailLabels>> {
filter_by_secondary_key_impl(
DB_MANAGER.envelope_db(),
GmailLabelsKey::account_id,
account_id,
)
.await
}
pub async fn batch_delete(labels: Vec<GmailLabels>) -> RustMailerResult<()> {
batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
let mut to_deleted = Vec::new();
for label in labels {
let retrived = rw
.get()
.primary::<GmailLabels>(label.id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
if let Some(retrived) = retrived {
to_deleted.push(retrived);
}
}
Ok(to_deleted)
})
.await?;
Ok(())
}
}
impl From<GmailLabels> for MailBox {
fn from(value: GmailLabels) -> Self {
Self {
id: value.id,
account_id: value.account_id,
name: value.name,
delimiter: Some("/".into()),
attributes: vec![],
flags: vec![],
exists: value.exists,
unseen: Some(value.unseen),
permanent_flags: vec![],
uid_next: None,
uid_validity: None,
highest_modseq: None,
}
}
}
#[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize, Object)]
#[native_model(id = 8, version = 1)]
#[native_db]
pub struct GmailCheckPoint {
/// Primary key for the checkpoint.
/// Generated by hashing the combination of `account_id` and `label_id` (as a string).
/// Uniquely identifies the sync state of a specific label for a specific account.
#[primary_key]
pub id: u64,
/// The Gmail account ID this checkpoint belongs to.
#[secondary_key]
pub account_id: u64,
/// The latest Gmail `historyId` for incremental synchronization.
/// Used as `startHistoryId` in the next Gmail History API call.
pub history_id: String,
/// Creation timestamp in UNIX epoch milliseconds.
/// Records when this checkpoint was initially created.
pub created_at: i64,
/// Last update timestamp in UNIX epoch milliseconds.
/// Records the most recent time this checkpoint was updated.
pub updated_at: i64,
}
impl GmailCheckPoint {
pub async fn get(id: u64) -> RustMailerResult<GmailCheckPoint> {
let entity = async_find_impl(DB_MANAGER.envelope_db(), id).await?;
entity.ok_or_else(|| {
raise_error!(
format!("GmailCheckPoint not found for id={}", id),
ErrorCode::ResourceNotFound
)
})
}
pub fn new(account_id: u64, label_id: u64, max_history_id: String) -> Self {
Self {
id: label_id,
account_id,
history_id: max_history_id,
created_at: utc_now!(),
updated_at: utc_now!(),
}
}
// Upsert is used here to overwrite the existing record
pub async fn save(&self) -> RustMailerResult<()> {
upsert_impl(DB_MANAGER.envelope_db(), self.to_owned()).await
}
pub async fn list_all(account_id: u64) -> RustMailerResult<Vec<GmailLabels>> {
filter_by_secondary_key_impl(
DB_MANAGER.envelope_db(),
GmailLabelsKey::account_id,
account_id,
)
.await
}
pub async fn clean(account_id: u64) -> RustMailerResult<()> {
batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
let to_delete: Vec<GmailCheckPoint> = rw
.scan()
.secondary(GmailCheckPointKey::account_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.start_with(account_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.try_collect()
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
Ok(to_delete)
})
.await?;
Ok(())
}
}
+211
View File
@@ -0,0 +1,211 @@
// Copyright © 2025 rustmailer.com
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
pub mod client;
pub mod envelope;
pub mod flow;
pub mod history;
pub mod labels;
pub mod rebuild;
pub mod sync_labels;
use std::time::Instant;
use ahash::AHashSet;
use tracing::info;
use crate::modules::{
account::{entity::MailerType, status::AccountRunningState, v2::AccountV2},
cache::{
imap::{
address::AddressEntity,
sync::sync_type::{determine_sync_type, SyncType},
thread::EmailThread,
},
vendor::gmail::sync::{
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::{
channel::{Event, EVENT_CHANNEL},
events::{payload::AccountChange, EventPayload, EventType, RustMailerEvent},
task::EventHookTask,
},
utils::mailbox_id,
};
pub async fn execute_gmail_sync(account: &AccountV2) -> RustMailerResult<()> {
assert!(
matches!(account.mailer_type, MailerType::GmailApi),
"Bug: Unexpected mailer type, expected GmailApi, found: {:?}",
account.mailer_type
);
let sync_type = determine_sync_type(account).await?;
if matches!(sync_type, SyncType::SkipSync) {
return Ok(());
}
let remote_labels = get_sync_labels(account).await?;
let remote_labels: Vec<GmailLabels> = remote_labels
.into_iter()
.map(|label| {
let mut label: GmailLabels = label.into();
label.account_id = account.id;
label.id = mailbox_id(account.id, &label.label_id);
label
})
.collect();
let local_labels = GmailLabels::list_all(account.id).await?;
// How to determine if a rebuild is needed?
// Simplified rule: if the local label does not exist, trigger a rebuild.
// We do not check how many local message metadata entries exist,
// since that would be expensive.
let local_checkpoints = GmailCheckPoint::list_all(account.id).await?;
if should_rebuild_cache(account, local_labels.len(), local_checkpoints.len()).await? {
AccountRunningState::set_initial_sync_folders(
account.id,
remote_labels
.iter()
.map(|label| label.name.clone())
.collect(),
)
.await?;
match &account.date_since {
Some(date_since) => {
rebuild_cache_since_date(account, &remote_labels, date_since).await?;
}
None => {
rebuild_cache(account, &remote_labels).await?;
}
}
AccountRunningState::set_initial_sync_completed(account.id).await?;
if EventHookTask::event_watched(account.id, EventType::AccountFirstSyncCompleted).await? {
EVENT_CHANNEL
.queue(Event::new(
account.id,
&account.email,
RustMailerEvent::new(
EventType::AccountFirstSyncCompleted,
EventPayload::AccountFirstSyncCompleted(AccountChange {
account_id: account.id,
account_email: account.email.clone(),
}),
),
))
.await;
}
return Ok(());
}
handle_history(account, &local_labels, &remote_labels).await?;
let deleted_labels = find_deleted_labels(&local_labels, &remote_labels);
let missing_labels = find_missing_labels(&local_labels, &remote_labels);
if !deleted_labels.is_empty() {
info!(
"Account {}: Detected {} mailboxes missing from the IMAP server (not found in the LSUB response). \
Now cleaning up these mailboxes and their associated metadata locally.",
account.id, deleted_labels.len()
);
cleanup_deleted_labels(account, &deleted_labels).await?;
}
if !missing_labels.is_empty() {
GmailLabels::batch_insert(&missing_labels).await?;
for label in &missing_labels {
rebuild_single_label_cache(account, label).await?;
}
}
Ok(())
}
pub async fn should_rebuild_cache(
account: &AccountV2,
local_labels_count: usize,
local_checkpoints_count: usize,
) -> RustMailerResult<bool> {
// If both local labels and checkpoint exist, no rebuild is needed.
if local_labels_count > 0 && local_checkpoints_count > 0 {
return Ok(false);
}
// If there are local mailboxes but no checkpoints, clear the mailboxes.
if local_labels_count > 0 {
let mailboxes = GmailLabels::list_all(account.id).await?;
GmailLabels::batch_delete(mailboxes).await?;
}
if local_checkpoints_count > 0 {
//这个要清理,清理掉本地缓存的所有信息,包括关联的索引信息,比如thread, checkpoint也是
//EnvelopeFlagsManager::clean_account(account.id).await?
GmailCheckPoint::clean(account.id).await?;
}
// If either remote mailboxes or local envelopes were missing, cache rebuild is required.
Ok(true)
}
pub fn find_deleted_labels(
local_labels: &[GmailLabels],
remote_labels: &[GmailLabels],
) -> Vec<GmailLabels> {
let remote_ids: AHashSet<_> = remote_labels.iter().map(|l| &l.id).collect();
local_labels
.iter()
.filter(|l| !remote_ids.contains(&l.id))
.cloned()
.collect()
}
pub fn find_missing_labels(
local_labels: &[GmailLabels],
remote_labels: &[GmailLabels],
) -> Vec<GmailLabels> {
let local_ids: AHashSet<_> = local_labels.iter().map(|l| &l.id).collect();
remote_labels
.iter()
.filter(|l| !local_ids.contains(&l.id))
.cloned()
.collect()
}
async fn cleanup_deleted_labels(
account: &AccountV2,
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?;
}
GmailLabels::batch_delete(deleted_labels.to_vec()).await?;
let elapsed_time = start_time.elapsed().as_secs();
info!(
"Cleanup deleted GmailLabels completed: {} seconds elapsed.",
elapsed_time
);
Ok(())
}
async fn cleanup_single_label(account: &AccountV2, 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?;
GmailLabels::delete(label.id).await?;
let elapsed_time = start_time.elapsed().as_secs();
info!(
"Cleanup GmailLabels completed: {} seconds elapsed.",
elapsed_time
);
Ok(())
}
+180
View File
@@ -0,0 +1,180 @@
// Copyright © 2025 rustmailer.com
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use crate::modules::{
account::{since::DateSince, v2::AccountV2},
cache::{
vendor::gmail::sync::flow::{fetch_and_save_full_label, fetch_and_save_since_date},
vendor::gmail::sync::labels::{GmailCheckPoint, GmailLabels},
},
error::RustMailerResult,
};
use std::time::Instant;
use tracing::{error, info, warn};
pub async fn rebuild_cache(
account: &AccountV2,
remote_labels: &[GmailLabels],
) -> RustMailerResult<()> {
let start_time = Instant::now();
let mut total_inserted = 0;
GmailLabels::batch_insert(remote_labels).await?;
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 {
GmailCheckPoint::new(account.id, label.id, history_id)
.save()
.await?;
}
}
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
);
}
}
}
}
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(())
}
pub async fn rebuild_cache_since_date(
account: &AccountV2,
remote_labels: &[GmailLabels],
date_since: &DateSince,
) -> RustMailerResult<()> {
let start_time = Instant::now();
let mut total_inserted = 0;
let date = date_since.since_gmail_date()?;
GmailLabels::batch_insert(remote_labels).await?;
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;
// After each label finishes syncing, record its checkpoint individually.
// This avoids fetching a large amount of unnecessary history records.
if let Some(history_id) = max_history_id {
GmailCheckPoint::new(account.id, label.id, history_id)
.save()
.await?;
}
}
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
);
}
}
}
}
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_label_cache(
account: &AccountV2,
label: &GmailLabels,
) -> 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)) => {
info!(
"Account {}: Label '{}' synced successfully. {} messages inserted.",
account.id, label.name, inserted
);
if let Some(history_id) = max_history_id {
GmailCheckPoint::new(account.id, label.id, history_id)
.save()
.await?;
}
}
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
);
}
}
}
}
None => match fetch_and_save_full_label(account, label, label.exists, true).await {
Ok((inserted, max_history_id)) => {
info!(
"Account {}: Label '{}' synced successfully. {} messages inserted.",
account.id, label.name, inserted
);
if let Some(history_id) = max_history_id {
GmailCheckPoint::new(account.id, label.id, history_id)
.save()
.await?;
}
}
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
);
}
}
},
}
}
Ok(())
}
+118
View File
@@ -0,0 +1,118 @@
// Copyright © 2025 rustmailer.com
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use std::sync::Arc;
use tracing::warn;
use crate::{
modules::{
account::v2::AccountV2,
cache::vendor::gmail::model::labels::{Label, LabelDetail},
cache::{
imap::sync::sync_folders::detect_mailbox_changes,
vendor::gmail::sync::client::GmailClient,
},
error::{code::ErrorCode, RustMailerError, RustMailerResult},
},
raise_error,
};
pub async fn get_sync_labels(account: &AccountV2) -> RustMailerResult<Vec<LabelDetail>> {
let all_labels = GmailClient::list_labels(account.id, account.use_proxy).await?;
let visible_labels: Vec<Label> = all_labels
.labels
.into_iter()
.filter(|label| label.message_list_visibility.as_deref() != Some("hide"))
.collect();
// Exclude all labels that cannot retrieve messages via the message list,
// since we use the message API to fetch message details.
if visible_labels.is_empty() {
warn!(
"Account {}: No visible labels returned from Gmail API.",
account.id
);
return Err(
raise_error!(
format!(
"No visible labels returned from Gmail API for account {}. This is unexpected and may indicate an issue with the Gmail API or data.",
account.id
),
ErrorCode::InternalError
)
);
}
// Detect label changes through this method and send notifications.
detect_mailbox_changes(
account,
visible_labels
.iter()
.map(|label| label.name.clone())
.collect(),
)
.await?;
// Labels that need to be synced are stored by their ID, not by name,
// because the label name can be changed.
let subscribed = &account.sync_folders;
// Filter labels according to the subscription list; matched_labels will not include any labels outside of it.
let mut matched_labels: Vec<&Label> = if !subscribed.is_empty() {
visible_labels
.iter()
.filter(|label| subscribed.contains(&label.id))
.collect()
} else {
Vec::new()
};
// If there are no subscriptions, default to the two special labels: INBOX and SENT
if matched_labels.is_empty() {
matched_labels = visible_labels
.iter()
.filter(|label| label.id == "INBOX" || label.id == "SENT")
.collect();
if !matched_labels.is_empty() {
let sync_folders: Vec<String> = matched_labels.iter().map(|n| n.id.clone()).collect();
AccountV2::update_sync_folders(account.id, sync_folders).await?;
} else {
warn!("Account {}: No visible labels found from Gmail API. This is unexpected — Gmail API should at least provide INBOX.", account.id);
return Err(
raise_error!(
format!("No visible labels found for account {} via Gmail API. This is unexpected — Gmail API should at least provide INBOX.", &account.id),
ErrorCode::InternalError
)
);
}
}
retrieve_label_metadata(account, matched_labels).await
}
pub async fn retrieve_label_metadata(
account: &AccountV2,
labels: impl IntoIterator<Item = &Label>,
) -> RustMailerResult<Vec<LabelDetail>> {
let mut tasks = Vec::new();
let account = Arc::new(account.clone());
for label in labels.into_iter() {
let label_id = label.id.clone();
let account = account.clone();
let task: tokio::task::JoinHandle<Result<LabelDetail, RustMailerError>> =
tokio::spawn(async move {
GmailClient::get_label(account.id, account.use_proxy, label_id.as_str()).await
});
tasks.push(task);
}
let mut details = Vec::new();
for task in tasks {
match task.await {
Ok(Ok(detail)) => details.push(detail),
Ok(Err(err)) => return Err(err),
Err(e) => return Err(raise_error!(format!("{:#?}", e), ErrorCode::InternalError)),
}
}
Ok(details)
}
@@ -5,5 +5,4 @@
#[cfg(test)]
mod tests;
pub mod gmail;
+205
View File
@@ -0,0 +1,205 @@
// Copyright © 2025 rustmailer.com
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use poem_grpc::{ClientConfig, CompressionEncoding};
use reqwest::header::{AUTHORIZATION, CONTENT_TYPE};
use serde_json::Value;
use std::time::Duration;
use crate::{
modules::{
cache::{
imap::v2::EmailEnvelopeV3,
vendor::gmail::{
model::{
history::HistoryList,
messages::{MessageList, MessageMeta},
},
sync::envelope::GmailEnvelope,
},
},
common::{rustls::RustMailerTls, Addr},
context::Initialize,
grpc::service::rustmailer_grpc::{GetOAuth2TokensRequest, OAuth2ServiceClient},
},
rustmailer_version,
};
async fn access_token() -> String {
RustMailerTls::initialize().await.unwrap();
let cfg = ClientConfig::builder()
.uri("http://localhost:16630")
.build()
.unwrap();
let mut grpc_client = OAuth2ServiceClient::new(cfg);
grpc_client.set_accept_compressed([CompressionEncoding::GZIP]);
grpc_client.set_send_compressed(CompressionEncoding::GZIP);
let request = GetOAuth2TokensRequest {
account_id: 7397694139904449,
};
let mut request = poem_grpc::Request::new(request);
request.metadata_mut().insert(
AUTHORIZATION,
format!("Bearer {}", "2mY4irNCahQXeSarHYje1P1W"),
);
let result = grpc_client.get_o_auth2_tokens(request).await.unwrap();
result.access_token.clone().unwrap()
}
#[tokio::test]
async fn test1() {
let access_token = access_token().await;
let url = "https://gmail.googleapis.com/gmail/v1/users/me/labels/Label_6886728075529239043";
let url = "https://gmail.googleapis.com/gmail/v1/users/me/messages?labelIds=INBOX&maxResults=10&pageToken=08792416985640480557";
let url =
"https://gmail.googleapis.com/gmail/v1/users/me/messages/198e590baf688394?format=metadata";
let url =
"https://gmail.googleapis.com/gmail/v1/users/me/messages?labelIds=INBOX&maxResults=10";
let url = "https://gmail.googleapis.com/gmail/v1/users/me/labels/Label_6886728075529239043";
let url = "https://gmail.googleapis.com/gmail/v1/users/me/labels";
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();
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();
let pretty = serde_json::to_string_pretty(&body).unwrap();
println!("Response = {}", pretty);
} else {
eprintln!("Error: {} - {:?}", res.status(), res.text().await.unwrap());
}
}
#[tokio::test]
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";
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();
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();
let detail: MessageList = serde_json::from_value(body).unwrap();
println!("Response = {:#?}", detail);
} else {
eprintln!("Error: {} - {:?}", res.status(), res.text().await.unwrap());
}
}
#[tokio::test]
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";
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();
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();
let detail: MessageMeta = serde_json::from_value(body).unwrap();
let envelope: GmailEnvelope = detail.try_into().unwrap();
println!("Response = {:#?}", envelope);
let envelope: EmailEnvelopeV3 = envelope.into();
println!("Response = {:#?}", envelope);
} else {
eprintln!("Error: {} - {:?}", res.status(), res.text().await.unwrap());
}
}
#[tokio::test]
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";
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();
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();
let json = serde_json::to_string_pretty(&body).unwrap();
println!("Response = {}", json);
let list: HistoryList = serde_json::from_value(body).unwrap();
println!("Response = {:#?}", list);
} else {
eprintln!("Error: {} - {:?}", res.status(), res.text().await.unwrap());
}
}
#[tokio::test]
async fn test5() {
let examples = [
"Quinn Eckart <jira@lifebuoy.atlassian.net>",
"justemail@example.com",
"<only@example.com>",
];
for s in examples {
let addr = Addr::parse(s);
println!("{:?}", addr);
}
}
+35 -2
View File
@@ -11,20 +11,21 @@ use poem::error::ResponseError;
use poem::Body;
use poem::{http::StatusCode, Error, Response};
use poem_openapi::Object;
use regex::Regex;
use serde::{Deserialize, Serialize};
use std::borrow::Cow;
use std::ops::Deref;
use tracing::error;
pub mod auth;
pub mod error;
pub mod log;
pub mod paginated;
pub mod rustls;
pub mod signal;
pub mod timeout;
pub mod tls;
pub mod validator;
pub mod error;
pub mod signal;
#[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize, Object)]
pub struct Addr {
@@ -36,6 +37,38 @@ pub struct Addr {
pub address: Option<String>,
}
impl Addr {
pub fn parse(s: &str) -> Self {
let re = Regex::new(r#"(?:(?P<name>.*)\s*)?<(?P<email>[^<>]+)>"#).unwrap();
if let Some(caps) = re.captures(s) {
let name: Option<String> = caps.name("name").map(|m| m.as_str().trim().into());
let email: Option<String> = caps.name("email").map(|m| m.as_str().trim().into());
Addr {
name: if let Some(n) = name {
if n.is_empty() {
None
} else {
Some(n)
}
} else {
None
},
address: email,
}
} else {
let s_trimmed = s.trim();
Addr {
name: None,
address: if s_trimmed.is_empty() {
None
} else {
Some(s_trimmed.into())
},
}
}
}
}
impl std::fmt::Display for Addr {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match (&self.name, &self.address) {
+2 -2
View File
@@ -3,7 +3,7 @@
// Unauthorized copying, modification, or distribution is prohibited.
use crate::modules::account::v2::AccountV2;
use crate::modules::cache::imap::v2::EmailEnvelopeV2;
use crate::modules::cache::imap::v2::EmailEnvelopeV3;
use crate::modules::cache::imap::ENVELOPE_MODELS;
use crate::modules::context::Initialize;
use crate::modules::error::{code::ErrorCode, RustMailerError};
@@ -230,7 +230,7 @@ impl DatabaseManager {
let rw = database
.rw_transaction()
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
rw.migrate::<EmailEnvelopeV2>()
rw.migrate::<EmailEnvelopeV3>()
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
rw.commit()
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
+6 -4
View File
@@ -3,10 +3,10 @@
// Unauthorized copying, modification, or distribution is prohibited.
use crate::id;
use crate::modules::cache::imap::v2::EmailEnvelopeV2;
use crate::modules::cache::imap::flags_to_hash;
use crate::modules::cache::imap::mailbox::EnvelopeFlag;
use crate::modules::cache::imap::minimal::MinimalEnvelope;
use crate::modules::cache::imap::v2::EmailEnvelopeV3;
use crate::modules::common::AddrVec;
use crate::modules::envelope::MinimalEnvelopeMeta;
use crate::modules::error::code::ErrorCode;
@@ -22,7 +22,7 @@ pub fn extract_envelope(
fetch: &Fetch,
account_id: u64,
mailbox_name: &str,
) -> RustMailerResult<EmailEnvelopeV2> {
) -> RustMailerResult<EmailEnvelopeV3> {
let attachments: Option<Vec<crate::modules::imap::section::ImapAttachment>> =
SectionExtractor::new(fetch.bodystructure().ok_or_else(|| {
raise_error!(
@@ -67,7 +67,7 @@ pub fn extract_envelope(
)
})?;
let envelope = EmailEnvelopeV2 {
let envelope = EmailEnvelopeV3 {
account_id,
mailbox_id: mailbox_id(account_id, mailbox_name),
mailbox_name: mailbox_name.into(),
@@ -100,6 +100,8 @@ pub fn extract_envelope(
attachments,
body_meta: body,
received: message.received().map(Into::into),
mid: None,
label_ids: vec![],
};
Ok(envelope)
@@ -125,7 +127,7 @@ pub fn extract_rich_envelopes(
fetches: &Vec<Fetch>,
account_id: u64,
mailbox_name: &str,
) -> RustMailerResult<Vec<EmailEnvelopeV2>> {
) -> RustMailerResult<Vec<EmailEnvelopeV3>> {
let mut envelopes = Vec::with_capacity(fetches.len());
for fetch in fetches {
let envelope = extract_envelope(fetch, account_id, mailbox_name)?;
+4
View File
@@ -48,6 +48,8 @@ pub enum ErrorCode {
SmtpConnectionFailed = 50040,
MailBoxNotCached = 50050,
AutoconfigFetchFailed = 50060,
GmailApiCallFailed = 50070,
GmailApiInvalidHistoryId = 50080,
// Message queue errors (6000060999)
NatsRequestFailed = 60000,
@@ -83,6 +85,8 @@ impl ErrorCode {
ErrorCode::InternalError
| ErrorCode::AutoconfigFetchFailed
| ErrorCode::ImapCommandFailed
| ErrorCode::GmailApiCallFailed
| ErrorCode::GmailApiInvalidHistoryId
| ErrorCode::ImapUnexpectedResult
| ErrorCode::HttpResponseError
| ErrorCode::NatsRequestFailed
+2 -8
View File
@@ -28,28 +28,22 @@ impl From<RustMailerError> for Status {
| ErrorCode::ExceedsLimitation
| ErrorCode::EmlFileParseError
| ErrorCode::MissingContentLength => Code::InvalidArgument,
ErrorCode::PermissionDenied => Code::PermissionDenied,
ErrorCode::AccountDisabled
| ErrorCode::LicenseAccountLimitReached
| ErrorCode::LicenseExpired
| ErrorCode::InvalidLicense
| ErrorCode::OAuth2ItemDisabled => Code::PermissionDenied,
ErrorCode::ResourceNotFound => Code::NotFound,
ErrorCode::RequestTimeout => Code::DeadlineExceeded,
ErrorCode::AlreadyExists => Code::AlreadyExists,
ErrorCode::PayloadTooLarge => Code::ResourceExhausted,
ErrorCode::TooManyRequest => Code::ResourceExhausted,
ErrorCode::InternalError
| ErrorCode::AutoconfigFetchFailed
| ErrorCode::ImapCommandFailed
| ErrorCode::GmailApiCallFailed
| ErrorCode::GmailApiInvalidHistoryId
| ErrorCode::ImapUnexpectedResult
| ErrorCode::HttpResponseError
| ErrorCode::NatsRequestFailed
+4
View File
@@ -235,6 +235,7 @@ impl TryFrom<rustmailer_grpc::Account> for AccountV2 {
incremental_sync_interval_sec: value.incremental_sync_interval_sec,
created_at: value.created_at,
updated_at: value.updated_at,
use_proxy: value.use_proxy,
})
}
}
@@ -259,6 +260,7 @@ impl From<AccountV2> for rustmailer_grpc::Account {
incremental_sync_interval_sec: value.incremental_sync_interval_sec,
created_at: value.created_at,
updated_at: value.updated_at,
use_proxy: value.use_proxy,
}
}
}
@@ -278,6 +280,7 @@ impl TryFrom<rustmailer_grpc::AccountCreateRequest> for AccountCreateRequest {
minimal_sync: value.minimal_sync,
full_sync_interval_min: value.full_sync_interval_min,
incremental_sync_interval_sec: value.incremental_sync_interval_sec,
use_proxy: value.use_proxy,
})
}
}
@@ -295,6 +298,7 @@ impl TryFrom<rustmailer_grpc::AccountUpdateRequest> for AccountUpdateRequest {
incremental_sync_interval_sec: value.incremental_sync_interval_sec,
imap: value.imap.map(|imap| imap.try_into()).transpose()?,
smtp: value.smtp.map(|smtp| smtp.try_into()).transpose()?,
use_proxy: value.use_proxy,
})
}
}
+5 -3
View File
@@ -2,14 +2,16 @@
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use crate::modules::cache::imap::mailbox::{Attribute, AttributeEnum, EmailFlag, EnvelopeFlag};
use crate::modules::cache::imap::mailbox::{
Attribute, AttributeEnum, EmailFlag, EnvelopeFlag, MailBox,
};
use crate::modules::grpc::service::rustmailer_grpc;
use crate::modules::mailbox::rename::MailboxRenameRequest;
use crate::modules::{cache::imap::mailbox::MailBox, grpc::service::rustmailer_grpc};
impl From<MailBox> for rustmailer_grpc::MailBox {
fn from(value: MailBox) -> Self {
Self {
mailbox_hash: value.id,
mailbox_id: value.id,
account_hash: value.account_id,
name: value.name,
delimiter: value.delimiter,
+7 -5
View File
@@ -5,8 +5,8 @@
use crate::modules::{
cache::imap::{
envelope::Received,
v2::EmailEnvelopeV2,
mailbox::{EmailFlag, EnvelopeFlag},
v2::EmailEnvelopeV3,
},
common::Addr,
grpc::service::rustmailer_grpc::{self, PagedMessages},
@@ -132,8 +132,8 @@ impl TryFrom<rustmailer_grpc::FlagMessageRequest> for FlagMessageRequest {
}
}
impl From<DataPage<EmailEnvelopeV2>> for PagedMessages {
fn from(value: DataPage<EmailEnvelopeV2>) -> Self {
impl From<DataPage<EmailEnvelopeV3>> for PagedMessages {
fn from(value: DataPage<EmailEnvelopeV3>) -> Self {
Self {
current_page: value.current_page,
page_size: value.page_size,
@@ -144,8 +144,8 @@ impl From<DataPage<EmailEnvelopeV2>> for PagedMessages {
}
}
impl From<EmailEnvelopeV2> for rustmailer_grpc::EmailEnvelope {
fn from(value: EmailEnvelopeV2) -> Self {
impl From<EmailEnvelopeV3> for rustmailer_grpc::EmailEnvelope {
fn from(value: EmailEnvelopeV3) -> Self {
Self {
account_id: value.account_id,
mailbox_id: value.mailbox_id,
@@ -203,6 +203,8 @@ impl From<EmailEnvelopeV2> for rustmailer_grpc::EmailEnvelope {
.map(Into::into)
.collect(),
received: value.received.map(Into::into),
mid: value.mid,
label_ids: value.label_ids,
}
}
}
+70 -12
View File
@@ -2,6 +2,9 @@
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use dashmap::DashMap;
use http::header::{AUTHORIZATION, CONTENT_TYPE};
use crate::modules::error::code::ErrorCode;
use crate::modules::hook::entity::HttpMethod;
use crate::modules::settings::proxy::Proxy;
@@ -14,13 +17,8 @@ use std::time::Duration;
#[cfg(test)]
mod tests;
pub static HTTP_CLIENT: LazyLock<HttpClient> = LazyLock::new(|| {
let builder = HttpClient::base_builder();
let client = builder
.build()
.expect("Failed to build shared reqwest Client");
HttpClient::create(client)
});
// This will cache clients per proxy configuration.
static HTTP_CLIENTS_CACHE: LazyLock<DashMap<u64, reqwest::Client>> = LazyLock::new(DashMap::new);
pub struct HttpClient {
client: reqwest::Client,
@@ -39,9 +37,16 @@ impl HttpClient {
}
pub async fn new(use_proxy: Option<u64>) -> RustMailerResult<HttpClient> {
// Use proxy_id or 0 as the key for the cache
let proxy_id = use_proxy.unwrap_or(0);
// First, check if the HttpClient is already cached
if let Some(client) = HTTP_CLIENTS_CACHE.get(&proxy_id) {
return Ok(HttpClient::create(client.clone())); // Client is already cloneable, so clone the Arc here
}
// If not found in the cache, build a new HttpClient
let mut builder = Self::base_builder();
if let Some(proxy_id) = use_proxy {
if proxy_id != 0 {
// Only set the proxy if we have a valid proxy_id
let proxy = Proxy::get(proxy_id).await?;
let proxy_obj = reqwest::Proxy::all(&proxy.url).map_err(|e| {
raise_error!(
@@ -56,15 +61,16 @@ impl HttpClient {
.redirect(reqwest::redirect::Policy::none())
.proxy(proxy_obj);
}
// Build the HttpClient
let client = builder.build().map_err(|e| {
raise_error!(
format!("Failed to build HTTP client: {:#?}", e),
ErrorCode::InternalError
)
})?;
Ok(Self { client })
// Cache the newly created HttpClient
HTTP_CLIENTS_CACHE.insert(proxy_id, client.clone());
Ok(HttpClient::create(client))
}
pub async fn send_json_request(
@@ -101,4 +107,56 @@ impl HttpClient {
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
Ok(response)
}
/// Wrapper around the Gmail API `GET` request to fetch data.
pub async fn get(&self, url: &str, access_token: &str) -> RustMailerResult<serde_json::Value> {
let res = self
.client
.get(url)
.header(AUTHORIZATION, format!("Bearer {}", access_token))
.header(CONTENT_TYPE, "application/json")
.send()
.await
.map_err(|e| {
raise_error!(
format!("Request failed for URL {}: {:#?}", url, e),
ErrorCode::InternalError
)
})?;
if res.status().is_success() {
let json: serde_json::Value = res.json().await.map_err(|e| {
raise_error!(
format!("Failed to parse response from URL {}: {:#?}", url, e),
ErrorCode::InternalError
)
})?;
Ok(json)
} else {
let status = res.status();
let text = res.text().await.map_err(|e| {
raise_error!(
format!("Failed to read error response from URL {}: {:#?}", url, e),
ErrorCode::InternalError
)
})?;
if status.is_client_error() {
return Err(raise_error!(
format!(
"Gmail API returned client error (status {}) for {}: historyId may be invalid or expired. Response: {}",
status, url, text
),
ErrorCode::GmailApiInvalidHistoryId
));
}
// Return the error with status and response text for more context
Err(raise_error!(
format!(
"Gmail API call to {} failed with status {}: {}",
url, status, text
),
ErrorCode::GmailApiCallFailed
))
}
}
}
+1 -7
View File
@@ -7,7 +7,6 @@ use std::time::Instant;
use crate::modules::error::code::ErrorCode;
use crate::modules::error::RustMailerError;
use crate::modules::hook::http::HTTP_CLIENT;
use crate::modules::hook::vrl::payload::VrlScriptTestRequest;
use crate::modules::hook::vrl::resolve_vrl_input;
use crate::modules::hook::{entity::EventHooks, http::HttpClient};
@@ -256,12 +255,7 @@ async fn send_event(
let payload = process_payload(event, event_hook.vrl_script).await?;
if payload != serde_json::Value::Null {
let client = if event_hook.use_proxy.is_some() {
&HttpClient::new(event_hook.use_proxy).await?
} else {
&HTTP_CLIENT
};
let client = HttpClient::new(event_hook.use_proxy).await?;
let response = client
.send_json_request(
task,
+9 -9
View File
@@ -6,7 +6,7 @@ use crate::{
encode_mailbox_name,
modules::{
account::v2::AccountV2,
cache::imap::{mailbox::MailBox, thread::EmailThread, v2::EmailEnvelopeV2},
cache::imap::{mailbox::MailBox, thread::EmailThread, v2::EmailEnvelopeV3},
context::executors::RUST_MAIL_CONTEXT,
envelope::extractor::extract_envelope,
error::{code::ErrorCode, RustMailerResult},
@@ -23,7 +23,7 @@ pub async fn list_messages_in_mailbox(
page_size: u64,
remote: bool,
desc: bool,
) -> RustMailerResult<DataPage<EmailEnvelopeV2>> {
) -> RustMailerResult<DataPage<EmailEnvelopeV3>> {
let account = AccountV2::check_account_active(account_id).await?;
validate_pagination_params(page, page_size)?;
let remote = remote || account.minimal_sync();
@@ -57,7 +57,7 @@ async fn fetch_remote_messages(
page: u64,
page_size: u64,
desc: bool,
) -> RustMailerResult<DataPage<EmailEnvelopeV2>> {
) -> RustMailerResult<DataPage<EmailEnvelopeV3>> {
let excutor = RUST_MAIL_CONTEXT.imap(account_id).await?;
let (mut fetches, total_items) = excutor
.retrieve_metadata_paginated(
@@ -89,7 +89,7 @@ async fn process_fetches(
fetches: Vec<Fetch>,
account_id: u64,
mailbox_name: &str,
) -> RustMailerResult<Vec<EmailEnvelopeV2>> {
) -> RustMailerResult<Vec<EmailEnvelopeV3>> {
let mut envelopes = Vec::with_capacity(fetches.len());
for fetch in fetches {
let envelope = extract_envelope(&fetch, account_id, mailbox_name)?;
@@ -104,10 +104,10 @@ async fn fetch_local_messages(
page: u64,
page_size: u64,
desc: bool,
) -> RustMailerResult<DataPage<EmailEnvelopeV2>> {
) -> RustMailerResult<DataPage<EmailEnvelopeV3>> {
match MailBox::get(account.id, mailbox_name).await {
Ok(mailbox) => {
EmailEnvelopeV2::list_messages_in_mailbox(mailbox.id, page, page_size, desc).await
EmailEnvelopeV3::list_messages_in_mailbox(mailbox.id, page, page_size, desc).await
}
Err(_) => Err(raise_error!(
"This mailbox might not be included in the synchronized mailbox list of the account. \
@@ -124,7 +124,7 @@ pub async fn list_threads_in_mailbox(
page: u64,
page_size: u64,
desc: bool,
) -> RustMailerResult<DataPage<EmailEnvelopeV2>> {
) -> RustMailerResult<DataPage<EmailEnvelopeV3>> {
let account = AccountV2::check_account_active(account_id).await?;
validate_pagination_params(page, page_size)?;
if account.minimal_sync() {
@@ -159,7 +159,7 @@ pub async fn get_thread_messages(
account_id: u64,
mailbox_name: &str,
thread_id: u64,
) -> RustMailerResult<Vec<EmailEnvelopeV2>> {
) -> RustMailerResult<Vec<EmailEnvelopeV3>> {
let account = AccountV2::check_account_active(account_id).await?;
if account.minimal_sync() {
return Err(raise_error!(
@@ -174,7 +174,7 @@ pub async fn get_thread_messages(
}
match MailBox::get(account.id, mailbox_name).await {
Ok(mailbox) => EmailEnvelopeV2::get_thread(account_id, mailbox.id, thread_id).await,
Ok(mailbox) => EmailEnvelopeV3::get_thread(account_id, mailbox.id, thread_id).await,
Err(_) => Err(raise_error!(
format!(
"Mailbox '{}' not found in the synchronized mailbox list for account {}. \
+5 -5
View File
@@ -4,7 +4,7 @@
use crate::modules::cache::imap::address::AddressEntity;
use crate::modules::cache::imap::sync::flow::generate_uid_sequence_hashset;
use crate::modules::cache::imap::v2::EmailEnvelopeV2;
use crate::modules::cache::imap::v2::EmailEnvelopeV3;
use crate::modules::common::paginated::paginate_vec;
use crate::modules::database::Paginated;
use crate::modules::error::code::ErrorCode;
@@ -436,7 +436,7 @@ impl MessageSearchRequest {
page: u64,
page_size: u64,
desc: bool,
) -> RustMailerResult<DataPage<EmailEnvelopeV2>> {
) -> RustMailerResult<DataPage<EmailEnvelopeV3>> {
let account = AccountV2::check_account_active(account_id).await?;
self.search_remote(&account, page, page_size, desc).await
}
@@ -447,7 +447,7 @@ impl MessageSearchRequest {
page: u64,
page_size: u64,
desc: bool,
) -> RustMailerResult<DataPage<EmailEnvelopeV2>> {
) -> RustMailerResult<DataPage<EmailEnvelopeV3>> {
// Validate page and page_size
if page == 0 || page_size == 0 {
return Err(raise_error!(
@@ -590,7 +590,7 @@ impl UnifiedSearchRequest {
page: u64,
page_size: u64,
desc: bool,
) -> RustMailerResult<DataPage<EmailEnvelopeV2>> {
) -> RustMailerResult<DataPage<EmailEnvelopeV3>> {
if page == 0 || page_size == 0 {
return Err(raise_error!(
"'page' and 'page_size' must be greater than 0.".into(),
@@ -634,7 +634,7 @@ impl UnifiedSearchRequest {
let result = paginate_vec(&vec, Some(page), Some(page_size))?;
let mut items = Vec::new();
for (id, _) in result.items {
let envelope = EmailEnvelopeV2::get(id).await?.ok_or_else(|| {
let envelope = EmailEnvelopeV3::get(id).await?.ok_or_else(|| {
raise_error!(
format!("Failed to get EmailEnvelope for hash {id} in search operation"),
ErrorCode::InternalError
-1
View File
@@ -28,5 +28,4 @@ pub mod smtp;
pub mod tasks;
pub mod token;
pub mod utils;
pub mod vendor;
pub mod version;
+6 -6
View File
@@ -3,7 +3,7 @@
// Unauthorized copying, modification, or distribution is prohibited.
use crate::current_datetime;
use crate::modules::cache::imap::v2::EmailEnvelopeV2;
use crate::modules::cache::imap::v2::EmailEnvelopeV3;
use crate::modules::common::auth::ClientContext;
use crate::modules::message::append::AppendReplyToDraftRequest;
use crate::modules::message::attachment::{retrieve_email_attachment, AttachmentRequest};
@@ -133,7 +133,7 @@ impl MessageApi {
/// lists messages in descending order; otherwise, ascending. internal date
desc: Query<Option<bool>>,
context: ClientContext,
) -> ApiResult<Json<DataPage<EmailEnvelopeV2>>> {
) -> ApiResult<Json<DataPage<EmailEnvelopeV3>>> {
let remote = remote.0.unwrap_or(false);
let desc = desc.0.unwrap_or(false);
let account_id = account_id.0;
@@ -173,7 +173,7 @@ impl MessageApi {
/// lists messages in descending order; otherwise, ascending. internal date
desc: Query<Option<bool>>,
context: ClientContext,
) -> ApiResult<Json<DataPage<EmailEnvelopeV2>>> {
) -> ApiResult<Json<DataPage<EmailEnvelopeV3>>> {
let desc = desc.0.unwrap_or(false);
let account_id = account_id.0;
context.require_account_access(account_id)?;
@@ -201,7 +201,7 @@ impl MessageApi {
// Thread ID
thread_id: Query<u64>,
context: ClientContext,
) -> ApiResult<Json<Vec<EmailEnvelopeV2>>> {
) -> ApiResult<Json<Vec<EmailEnvelopeV3>>> {
let account_id = account_id.0;
context.require_account_access(account_id)?;
@@ -308,7 +308,7 @@ impl MessageApi {
/// specifying the search criteria (e.g., keywords, flags).
payload: Json<MessageSearchRequest>,
context: ClientContext,
) -> ApiResult<Json<DataPage<EmailEnvelopeV2>>> {
) -> ApiResult<Json<DataPage<EmailEnvelopeV3>>> {
let request = payload.0;
let desc = desc.0.unwrap_or(false);
let account_id = account_id.0;
@@ -344,7 +344,7 @@ impl MessageApi {
/// Request context (includes authentication and permissions).
context: ClientContext,
) -> ApiResult<Json<DataPage<EmailEnvelopeV2>>> {
) -> ApiResult<Json<DataPage<EmailEnvelopeV3>>> {
let mut request = payload.0;
let desc = desc.0.unwrap_or(false);
+10 -6
View File
@@ -2,7 +2,7 @@
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use crate::modules::cache::imap::v2::EmailEnvelopeV2;
use crate::modules::cache::imap::v2::EmailEnvelopeV3;
use scraper::{Html, Selector};
use time::{macros::format_description, OffsetDateTime};
use time_tz::timezones;
@@ -50,7 +50,7 @@ impl BodyComposer {
pub fn generate_html(
original_html: &str,
reply_content: &str,
envelope: &EmailEnvelopeV2,
envelope: &EmailEnvelopeV3,
timezone_name: &str,
reply: bool,
) -> String {
@@ -161,7 +161,7 @@ impl BodyComposer {
pub fn generate_text(
original_text: &str,
reply_content: &str,
envelope: &EmailEnvelopeV2,
envelope: &EmailEnvelopeV3,
timezone_name: &str,
reply: bool,
) -> String {
@@ -229,8 +229,8 @@ mod tests {
id,
modules::{
cache::imap::{
v2::EmailEnvelopeV2,
mailbox::{EmailFlag, EnvelopeFlag},
v2::EmailEnvelopeV3,
},
common::Addr,
},
@@ -289,7 +289,7 @@ mod tests {
let reply_content = "Thanks for your message!";
let envelope = EmailEnvelopeV2 {
let envelope = EmailEnvelopeV3 {
account_id: 0,
mailbox_id: 0,
mailbox_name: "inbox_001".to_string(),
@@ -330,6 +330,8 @@ mod tests {
attachments: None,
body_meta: None,
received: None,
mid: None,
label_ids: vec![],
};
let result = BodyComposer::generate_html(
@@ -347,7 +349,7 @@ mod tests {
let original_text = "Hello,\nThis is a test email.\nRegards,\nJohn";
let reply_content = "Hi John,\nThanks for your email!";
let envelope = EmailEnvelopeV2 {
let envelope = EmailEnvelopeV3 {
from: Some(Addr {
name: Some("John Doe".to_string()),
address: Some("john@example.com".to_string()),
@@ -377,6 +379,8 @@ mod tests {
attachments: None,
body_meta: None,
received: None,
mid: None,
label_ids: vec![],
};
let result = BodyComposer::generate_text(
+3 -3
View File
@@ -2,7 +2,7 @@
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use crate::modules::cache::imap::v2::EmailEnvelopeV2;
use crate::modules::cache::imap::v2::EmailEnvelopeV3;
use crate::modules::error::code::ErrorCode;
use crate::modules::smtp::request::builder::EmailBuilder;
use crate::modules::smtp::request::headers::HeaderValue;
@@ -226,7 +226,7 @@ impl ForwardEmailRequest {
fn apply_references(
&self,
builder: MessageBuilder<'static>,
envelope: &EmailEnvelopeV2,
envelope: &EmailEnvelopeV3,
) -> RustMailerResult<MessageBuilder<'static>> {
let mut references = envelope.references.clone().unwrap_or_default();
if let Some(message_id) = &envelope.message_id {
@@ -240,7 +240,7 @@ impl ForwardEmailRequest {
async fn apply_content(
&self,
mut builder: MessageBuilder<'static>,
envelope: &EmailEnvelopeV2,
envelope: &EmailEnvelopeV3,
account: &AccountV2,
) -> RustMailerResult<MessageBuilder<'static>> {
let timezone = self.timezone.as_deref().unwrap_or("UTC");
+5 -5
View File
@@ -8,7 +8,7 @@ use crate::modules::cache::disk::DISK_CACHE;
use crate::modules::cache::imap::mailbox::EmailFlag;
use crate::modules::cache::imap::mailbox::EnvelopeFlag;
use crate::modules::cache::imap::mailbox::MailBox;
use crate::modules::cache::imap::v2::EmailEnvelopeV2;
use crate::modules::cache::imap::v2::EmailEnvelopeV3;
use crate::modules::common::Addr;
use crate::modules::context::executors::RUST_MAIL_CONTEXT;
use crate::modules::envelope::extractor::extract_envelope;
@@ -601,7 +601,7 @@ impl EmailHandler {
pub async fn retrieve_message_content(
account: &AccountV2,
envelope: &EmailEnvelopeV2,
envelope: &EmailEnvelopeV3,
) -> RustMailerResult<Option<MessageContent>> {
let body_meta = match &envelope.body_meta {
Some(meta) => meta,
@@ -629,10 +629,10 @@ impl EmailHandler {
account: &AccountV2,
mailbox_name: &str,
uid: u32,
) -> RustMailerResult<EmailEnvelopeV2> {
) -> RustMailerResult<EmailEnvelopeV3> {
if let Ok(mailbox) = MailBox::get(account.id, mailbox_name).await {
if !account.minimal_sync() {
let envelope = EmailEnvelopeV2::find(account.id, mailbox.id, uid).await?;
let envelope = EmailEnvelopeV3::find(account.id, mailbox.id, uid).await?;
if let Some(envelope) = envelope {
return Ok(envelope);
}
@@ -703,7 +703,7 @@ impl EmailHandler {
async fn add_attachment(
builder: MessageBuilder<'static>,
attachment: &ImapAttachment,
envelope: &EmailEnvelopeV2,
envelope: &EmailEnvelopeV3,
inline: bool,
account: &AccountV2,
) -> RustMailerResult<MessageBuilder<'static>> {
+4 -4
View File
@@ -5,7 +5,7 @@
use crate::{
modules::{
account::v2::AccountV2,
cache::imap::v2::EmailEnvelopeV2,
cache::imap::v2::EmailEnvelopeV3,
error::{code::ErrorCode, RustMailerResult},
smtp::{
composer::BodyComposer,
@@ -189,7 +189,7 @@ impl ReplyEmailRequest {
fn apply_recipient_headers(
&self,
mut builder: MessageBuilder<'static>,
envelope: &EmailEnvelopeV2,
envelope: &EmailEnvelopeV3,
message_id: &str,
) -> RustMailerResult<MessageBuilder<'static>> {
if self.reply_all {
@@ -226,7 +226,7 @@ impl ReplyEmailRequest {
async fn apply_content(
&self,
mut builder: MessageBuilder<'static>,
envelope: &EmailEnvelopeV2,
envelope: &EmailEnvelopeV3,
account: &AccountV2,
) -> RustMailerResult<MessageBuilder<'static>> {
let timezone = self.timezone.as_deref().unwrap_or("UTC");
@@ -340,7 +340,7 @@ impl ReplyEmailRequest {
pub fn apply_references(
builder: MessageBuilder<'static>,
envelope: &EmailEnvelopeV2,
envelope: &EmailEnvelopeV3,
) -> RustMailerResult<MessageBuilder<'static>> {
let builder = if let Some(message_id) = &envelope.message_id {
builder.in_reply_to(message_id.clone())
+12 -1
View File
@@ -329,7 +329,6 @@ pub fn json_value_to_prost_value(json_value: serde_json::Value) -> prost_types::
prost_types::Value { kind }
}
/// Generates a 64-bit hash from a string, ensuring the output is within JavaScript's safe integer range (0 to 2^53 - 1).
pub fn hash(s: &str) -> u64 {
let mut cursor = Vec::new();
@@ -361,3 +360,15 @@ pub fn envelope_hash(account_id: u64, mailbox_id: u64, uid: u32) -> u64 {
let hash = murmur3::murmur3_x64_128(&mut cursor, 0).unwrap();
hash as u64
}
/// Generate a 64-bit hash for a GmailEnvelope using account_id, mailbox_id, and gmail api message id.
/// The `id` string is hashed to produce a consistent u64 value.
pub fn envelope_hash_from_id(account_id: u64, mailbox_id: u64, id: &str) -> u64 {
let mut buffer = Vec::with_capacity(8 + 8 + id.len());
buffer.extend_from_slice(&account_id.to_be_bytes());
buffer.extend_from_slice(&mailbox_id.to_be_bytes());
buffer.extend_from_slice(id.as_bytes());
let mut cursor = std::io::Cursor::new(buffer);
let hash128 = murmur3::murmur3_x64_128(&mut cursor, 0).unwrap();
hash128 as u64
}
-44
View File
@@ -1,44 +0,0 @@
// Copyright © 2025 rustmailer.com
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use reqwest::header::{AUTHORIZATION, CONTENT_TYPE};
use serde_json::Value;
use std::time::Duration;
use crate::rustmailer_version;
#[tokio::test]
async fn test1() {
let access_token = "xxx";
let url = "https://gmail.googleapis.com/gmail/v1/users/me/labels";
let url = "https://gmail.googleapis.com/gmail/v1/users/me/labels/Label_6886728075529239043";
let url = "https://gmail.googleapis.com/gmail/v1/users/me/messages?labelIds=INBOX&maxResults=10&pageToken=08792416985640480557";
let url = "https://gmail.googleapis.com/gmail/v1/users/me/messages/198e590baf688394?format=metadata";
let url = "https://gmail.googleapis.com/gmail/v1/users/me/messages?labelIds=INBOX&maxResults=10";
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();
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();
let pretty = serde_json::to_string_pretty(&body).unwrap();
println!("Response = {}", pretty);
} else {
eprintln!("Error: {} - {:?}", res.status(), res.text().await.unwrap());
}
}
@@ -46,6 +46,7 @@ const accountSchema = () =>
name: z.string().optional(),
email: z.string({ required_error: 'Email is required' }).email({ message: 'Invalid email address' }),
enabled: z.boolean(),
minimal_sync: z.boolean(),
date_since: dateSelectionSchema.optional(),
incremental_sync_interval_sec: z.number({ invalid_type_error: 'Incremental sync interval must be a number' }).int().min(1, { message: 'Incremental sync interval must be at least 1 second' }),
});
@@ -55,6 +56,7 @@ export type GmailApiAccount = {
name?: string;
email: string;
enabled: boolean;
minimal_sync: boolean;
date_since?: {
fixed?: string;
relative?: {
@@ -80,6 +82,7 @@ const defaultValues: GmailApiAccount = {
enabled: true,
date_since: undefined,
incremental_sync_interval_sec: 30,
minimal_sync: false
};
@@ -88,6 +91,7 @@ const mapCurrentRowToFormValues = (currentRow: AccountEntity): GmailApiAccount =
name: currentRow.name === null ? '' : currentRow.name,
email: currentRow.email,
enabled: currentRow.enabled,
minimal_sync: currentRow.minimal_sync ?? false,
date_since: currentRow.date_since ?? undefined,
incremental_sync_interval_sec: currentRow.incremental_sync_interval_sec,
};
@@ -160,6 +164,7 @@ export function GmailApiAccountDialog({ currentRow, open, onOpenChange }: Props)
name: data.name,
enabled: data.enabled,
date_since: data.date_since,
minimal_sync: data.minimal_sync,
incremental_sync_interval_sec: data.incremental_sync_interval_sec,
};
if (isEdit) {
@@ -190,7 +195,7 @@ export function GmailApiAccountDialog({ currentRow, open, onOpenChange }: Props)
Click save when you're done.
</DialogDescription>
</DialogHeader>
<ScrollArea className='h-[32rem] w-full pr-4 -mr-4 py-1'>
<ScrollArea className='h-[40rem] w-full pr-4 -mr-4 py-1'>
<Form {...form}>
<form
id='gmail-api-account-form'
@@ -242,6 +247,9 @@ export function GmailApiAccountDialog({ currentRow, open, onOpenChange }: Props)
<FormControl>
<Input type="number" placeholder="e.g 300" {...field} onChange={(e) => field.onChange(parseInt(e.target.value, 10))} />
</FormControl>
<FormDescription>
Set the interval (in seconds) for calling the Gmail History API for incremental sync. This determines how frequently updates are fetched for new or modified emails.
</FormDescription>
<FormMessage />
</FormItem>
)}
@@ -264,6 +272,30 @@ export function GmailApiAccountDialog({ currentRow, open, onOpenChange }: Props)
</FormItem>
)}
/>
<FormField
control={form.control}
name='minimal_sync'
render={({ field }) => (
<FormItem className='flex flex-col items-start gap-y-1'>
<FormLabel>Minimal Sync:</FormLabel>
<FormControl>
<Checkbox
className='mt-2'
checked={field.value}
onCheckedChange={isEdit ? undefined : field.onChange}
disabled={isEdit}
/>
</FormControl>
<FormDescription>
{isEdit ? (
"This setting cannot be modified after account creation."
) : (
"When enabled, Gmail metadata will not be cached locally, ensuring higher synchronization efficiency by syncing only essential basic metadata fields."
)}
</FormDescription>
</FormItem>
)}
/>
<FormLabel className="flex items-center justify-between">
Date Since:
</FormLabel>