Files
RustBox/src/imap/client.rs
T
Mateusz Gruszczyński 20d78ee7be
RustBox CI / rustbox (push) Canceled after 0s
v0.2.27
2026-11-06 11:15:43 +01:00

1436 lines
46 KiB
Rust

use async_trait::async_trait;
use futures::StreamExt;
use super::connection::map_imap_error;
use super::parse::{
decode_header_bytes, flag_to_string, has_attachments, imap_address_to_email,
name_attribute_to_string,
};
pub use super::error::ImapError;
pub use super::types::{
EmailAddress, FolderStatus, FolderStatusExtended, ImapAttachment, ImapCredentials, ImapFolder,
ImapMessageBody, ImapMessageHeader, MailboxQuota,
};
pub(crate) use super::connection::connect;
#[cfg(test)]
#[path = "mock.rs"]
pub mod mock;
async fn folder_status_extended_on_session(
session: &mut async_imap::Session<super::connection::ImapStream>,
folder: &str,
) -> Result<FolderStatusExtended, ImapError> {
if let Ok(mailbox) = session
.status(
folder,
"(MESSAGES UIDNEXT UIDVALIDITY UNSEEN HIGHESTMODSEQ)",
)
.await
&& let Some(unseen) = mailbox.unseen
{
return Ok(FolderStatusExtended {
uid_validity: mailbox.uid_validity.unwrap_or(0),
exists: mailbox.exists,
uid_next: mailbox.uid_next.unwrap_or(0),
unseen,
highest_modseq: mailbox.highest_modseq.unwrap_or(0),
});
}
if let Ok(mailbox) = session
.status(folder, "(MESSAGES UIDNEXT UIDVALIDITY UNSEEN)")
.await
&& let Some(unseen) = mailbox.unseen
{
return Ok(FolderStatusExtended {
uid_validity: mailbox.uid_validity.unwrap_or(0),
exists: mailbox.exists,
uid_next: mailbox.uid_next.unwrap_or(0),
unseen,
highest_modseq: mailbox.highest_modseq.unwrap_or(0),
});
}
let mailbox = match session.select_condstore(folder).await {
Ok(mailbox) => mailbox,
Err(_) => session.select(folder).await.map_err(map_imap_error)?,
};
let unseen = session
.uid_search("UNSEEN")
.await
.map_err(map_imap_error)?
.len() as u32;
Ok(FolderStatusExtended {
uid_validity: mailbox.uid_validity.unwrap_or(0),
exists: mailbox.exists,
uid_next: mailbox.uid_next.unwrap_or(0),
unseen,
highest_modseq: mailbox.highest_modseq.unwrap_or(0),
})
}
#[allow(dead_code)]
#[allow(clippy::double_must_use)]
#[async_trait]
pub trait ImapClient: Send + Sync {
async fn capabilities(&self, creds: &ImapCredentials) -> Result<Vec<String>, ImapError>;
async fn reset_connections(&self, _creds: &ImapCredentials) -> Result<(), ImapError> {
Ok(())
}
async fn list_folders(&self, creds: &ImapCredentials) -> Result<Vec<ImapFolder>, ImapError>;
async fn folder_status(
&self,
creds: &ImapCredentials,
folder: &str,
) -> Result<FolderStatus, ImapError>;
async fn fetch_headers(
&self,
creds: &ImapCredentials,
folder: &str,
uid_range: &str,
) -> Result<Vec<ImapMessageHeader>, ImapError>;
async fn fetch_body(
&self,
creds: &ImapCredentials,
folder: &str,
uid: u32,
) -> Result<ImapMessageBody, ImapError>;
async fn fetch_raw_message(
&self,
_creds: &ImapCredentials,
folder: &str,
uid: u32,
) -> Result<Vec<u8>, ImapError> {
Err(ImapError::MessageNotFound {
uid,
folder: folder.to_string(),
})
}
async fn add_flags(
&self,
creds: &ImapCredentials,
folder: &str,
uid: u32,
flags: &[&str],
) -> Result<(), ImapError>;
async fn remove_flags(
&self,
creds: &ImapCredentials,
folder: &str,
uid: u32,
flags: &[&str],
) -> Result<(), ImapError>;
async fn set_flags(
&self,
creds: &ImapCredentials,
folder: &str,
uid: u32,
flags: &[&str],
) -> Result<(), ImapError>;
async fn move_message(
&self,
creds: &ImapCredentials,
from_folder: &str,
uid: u32,
to_folder: &str,
) -> Result<(), ImapError>;
async fn copy_message(
&self,
creds: &ImapCredentials,
from_folder: &str,
uid: u32,
to_folder: &str,
) -> Result<(), ImapError>;
async fn mark_all_read(&self, creds: &ImapCredentials, folder: &str) -> Result<(), ImapError>;
async fn empty_folder(&self, creds: &ImapCredentials, folder: &str) -> Result<(), ImapError>;
async fn expunge_message(
&self,
creds: &ImapCredentials,
folder: &str,
uid: u32,
) -> Result<(), ImapError>;
async fn append_message(
&self,
creds: &ImapCredentials,
folder: &str,
message_bytes: &[u8],
flags: &[&str],
) -> Result<(), ImapError>;
async fn append_message_resolved_uid(
&self,
creds: &ImapCredentials,
folder: &str,
message_bytes: &[u8],
flags: &[&str],
identity_header: &str,
identity_value: &str,
) -> Result<(u32, u32), ImapError>;
async fn create_folder(
&self,
creds: &ImapCredentials,
folder_name: &str,
) -> Result<(), ImapError>;
async fn rename_folder(
&self,
creds: &ImapCredentials,
from: &str,
to: &str,
) -> Result<(), ImapError>;
async fn delete_folder(
&self,
creds: &ImapCredentials,
folder_name: &str,
) -> Result<(), ImapError>;
async fn subscribe_folder(
&self,
creds: &ImapCredentials,
folder_name: &str,
subscribe: bool,
) -> Result<(), ImapError>;
async fn fetch_uids_and_flags(
&self,
creds: &ImapCredentials,
folder: &str,
) -> Result<Vec<(u32, Vec<String>)>, ImapError>;
/// `UID SEARCH <criteria>` in `folder`. Callers must build `criteria`
/// from validated, quoted parts (see `routes::filters`).
async fn search_uids(
&self,
creds: &ImapCredentials,
folder: &str,
criteria: &str,
) -> Result<Vec<u32>, ImapError>;
async fn folder_status_extended(
&self,
creds: &ImapCredentials,
folder: &str,
) -> Result<FolderStatusExtended, ImapError>;
async fn folder_statuses_extended(
&self,
creds: &ImapCredentials,
folders: &[String],
) -> Result<Vec<(String, FolderStatusExtended)>, ImapError> {
let mut statuses = Vec::with_capacity(folders.len());
for folder in folders {
statuses.push((
folder.clone(),
self.folder_status_extended(creds, folder).await?,
));
}
Ok(statuses)
}
async fn fetch_changed_flags(
&self,
creds: &ImapCredentials,
folder: &str,
since_modseq: u64,
) -> Result<(Vec<(u32, Vec<String>)>, u64), ImapError>;
async fn get_quota(&self, creds: &ImapCredentials) -> Result<Option<MailboxQuota>, ImapError>;
async fn fetch_folder_size(
&self,
creds: &ImapCredentials,
folder: &str,
) -> Result<u64, ImapError>;
async fn folder_keyword_support(
&self,
_creds: &ImapCredentials,
_folder: &str,
) -> Result<bool, ImapError> {
Ok(false)
}
}
#[derive(Default)]
pub struct RealImapClient {
pool: super::pool::ConnectionPool,
}
fn resolve_message_date(
envelope_date: Option<String>,
internal_date: Option<String>,
) -> Option<String> {
envelope_date
.and_then(|value| {
let trimmed = value.trim();
if trimmed.is_empty() || mail_parser::DateTime::parse_rfc822(trimmed).is_none() {
None
} else {
Some(trimmed.to_string())
}
})
.or_else(|| internal_date.filter(|value| !value.trim().is_empty()))
}
#[async_trait]
impl ImapClient for RealImapClient {
async fn capabilities(&self, creds: &ImapCredentials) -> Result<Vec<String>, ImapError> {
let mut session = self.pool.checkout(creds).await?;
let caps = session.capabilities().await.map_err(map_imap_error)?;
let mut values: Vec<String> = caps
.iter()
.map(|cap| match cap {
async_imap::types::Capability::Imap4rev1 => "IMAP4REV1".to_string(),
async_imap::types::Capability::Auth(name) => format!("AUTH={name}"),
async_imap::types::Capability::Atom(name) => name.to_string(),
})
.collect();
values.sort_by_key(|value| value.to_ascii_uppercase());
values.dedup_by(|a, b| a.eq_ignore_ascii_case(b));
session.recycle();
Ok(values)
}
async fn reset_connections(&self, creds: &ImapCredentials) -> Result<(), ImapError> {
self.pool.clear_user(creds);
Ok(())
}
async fn folder_status(
&self,
creds: &ImapCredentials,
folder: &str,
) -> Result<FolderStatus, ImapError> {
let mut session = self.pool.checkout(creds).await?;
let mailbox = session.select(folder).await.map_err(|e| match &e {
async_imap::error::Error::No(msg)
if msg.to_lowercase().contains("not found")
|| msg.to_lowercase().contains("doesn't exist")
|| msg.to_lowercase().contains("does not exist")
|| msg.to_lowercase().contains("no such") =>
{
ImapError::FolderNotFound(folder.to_string())
}
_ => map_imap_error(e),
})?;
let uid_validity = mailbox.uid_validity.unwrap_or(0);
let exists = mailbox.exists;
let uid_next = mailbox.uid_next.unwrap_or(0);
session.recycle();
Ok(FolderStatus {
uid_validity,
exists,
uid_next,
})
}
async fn list_folders(&self, creds: &ImapCredentials) -> Result<Vec<ImapFolder>, ImapError> {
let mut session = self.pool.checkout(creds).await?;
let folders = {
let names_stream = session
.list(Some(""), Some("*"))
.await
.map_err(map_imap_error)?;
let mut names_stream = std::pin::pin!(names_stream);
let mut names = Vec::new();
while let Some(result) = names_stream.next().await {
names.push(result.map_err(map_imap_error)?);
}
names
.iter()
.map(|n| ImapFolder {
name: n.name().to_string(),
delimiter: n.delimiter().map(|d| d.to_string()),
attributes: n
.attributes()
.iter()
.map(name_attribute_to_string)
.collect(),
})
.collect()
};
session.recycle();
Ok(folders)
}
async fn fetch_headers(
&self,
creds: &ImapCredentials,
folder: &str,
uid_range: &str,
) -> Result<Vec<ImapMessageHeader>, ImapError> {
let mut session = self.pool.checkout(creds).await?;
session.select(folder).await.map_err(|e| match &e {
async_imap::error::Error::No(msg)
if msg.to_lowercase().contains("not found")
|| msg.to_lowercase().contains("doesn't exist")
|| msg.to_lowercase().contains("does not exist")
|| msg.to_lowercase().contains("no such") =>
{
ImapError::FolderNotFound(folder.to_string())
}
_ => map_imap_error(e),
})?;
let fetches = {
let mut fetch_stream = session
.uid_fetch(
uid_range,
"(UID ENVELOPE INTERNALDATE FLAGS BODYSTRUCTURE RFC822.SIZE BODY.PEEK[HEADER.FIELDS (Message-ID In-Reply-To References Content-Class x-ms-exchange-generated-message-class)])",
)
.await
.map_err(map_imap_error)?;
let mut fetches = Vec::new();
while let Some(result) = fetch_stream.next().await {
fetches.push(result.map_err(map_imap_error)?);
}
fetches
};
session.recycle();
let folder = folder.to_owned();
tokio::task::spawn_blocking(move || {
let mut headers = Vec::with_capacity(fetches.len());
for fetch in &fetches {
let uid = match fetch.uid {
Some(u) => u,
None => continue,
};
let raw_header_bytes = fetch.header();
let parsed_threading = raw_header_bytes
.and_then(|raw| mail_parser::MessageParser::default().parse(raw));
let internal_date = fetch.internal_date().map(|value| value.to_rfc2822());
let (subject, from, to, cc, date) = if let Some(env) = fetch.envelope() {
let subject = env.subject.as_ref().map(|b| decode_header_bytes(b));
let from: Vec<EmailAddress> = env
.from
.as_ref()
.map(|addrs| addrs.iter().map(imap_address_to_email).collect())
.unwrap_or_default();
let to: Vec<EmailAddress> = env
.to
.as_ref()
.map(|addrs| addrs.iter().map(imap_address_to_email).collect())
.unwrap_or_default();
let cc: Vec<EmailAddress> = env
.cc
.as_ref()
.map(|addrs| addrs.iter().map(imap_address_to_email).collect())
.unwrap_or_default();
let envelope_date = env
.date
.as_ref()
.and_then(|b| std::str::from_utf8(b).ok())
.map(str::to_string);
let date = resolve_message_date(envelope_date, internal_date.clone());
(subject, from, to, cc, date)
} else {
tracing::warn!(
uid = uid,
folder = %folder,
"ENVELOPE missing for message, headers will be empty until body is fetched"
);
(
None,
vec![],
vec![],
vec![],
resolve_message_date(None, internal_date.clone()),
)
};
let message_id = parsed_threading
.as_ref()
.and_then(|p| p.message_id().map(|s| format!("<{s}>")));
let in_reply_to = parsed_threading.as_ref().and_then(|p| {
let val = p.in_reply_to();
val.as_text().map(|s| format!("<{s}>"))
});
let references = parsed_threading.as_ref().and_then(|p| {
let val = p.references();
val.as_text_list()
.map(|list| {
list.iter()
.map(|s| format!("<{s}>"))
.collect::<Vec<_>>()
.join(" ")
})
.or_else(|| val.as_text().map(|s| format!("<{s}>")))
});
let reaction = raw_header_bytes.and_then(|raw| {
let header_str = std::str::from_utf8(raw).ok()?;
let lower = header_str.to_lowercase();
let is_reaction = lower.contains("content-class: activitynotification")
|| lower.contains("urn:content-class:reaction");
if !is_reaction {
return None;
}
let subj = subject.as_deref().unwrap_or("").to_lowercase();
let emoji = match subj.trim() {
s if s.contains("like") => "\u{1f44d}",
s if s.contains("heart") || s.contains("love") => "\u{2764}\u{fe0f}",
s if s.contains("laugh") => "\u{1f604}",
s if s.contains("surprised") || s.contains("wow") => "\u{1f62e}",
s if s.contains("sad") => "\u{1f622}",
s if s.contains("angry") => "\u{1f620}",
_ => "\u{1f44d}",
};
Some(emoji.to_string())
});
let flags: Vec<String> = fetch.flags().map(|f| flag_to_string(&f)).collect();
let has_attach = fetch
.bodystructure()
.map(|bs| has_attachments(bs))
.unwrap_or(false);
let size = fetch.size.unwrap_or(0);
headers.push(ImapMessageHeader {
uid,
subject,
from,
to,
date,
flags,
has_attachments: has_attach,
size,
message_id,
in_reply_to,
references,
cc,
reaction,
});
}
headers
})
.await
.map_err(|e| ImapError::ProtocolError(format!("Header parsing task failed: {e}")))
}
async fn fetch_body(
&self,
creds: &ImapCredentials,
folder: &str,
uid: u32,
) -> Result<ImapMessageBody, ImapError> {
let mut session = self.pool.checkout(creds).await?;
session.select(folder).await.map_err(|e| match &e {
async_imap::error::Error::No(msg)
if msg.to_lowercase().contains("not found")
|| msg.to_lowercase().contains("doesn't exist")
|| msg.to_lowercase().contains("does not exist")
|| msg.to_lowercase().contains("no such") =>
{
ImapError::FolderNotFound(folder.to_string())
}
_ => map_imap_error(e),
})?;
let uid_str = uid.to_string();
let body = {
let mut fetch_stream = session
.uid_fetch(&uid_str, "(UID BODY[])")
.await
.map_err(map_imap_error)?;
let mut fetches = Vec::new();
while let Some(result) = fetch_stream.next().await {
fetches.push(result.map_err(map_imap_error)?);
}
let fetch = fetches.first().ok_or(ImapError::MessageNotFound {
uid,
folder: folder.to_string(),
})?;
let raw = fetch.body().ok_or_else(|| {
ImapError::ProtocolError("BODY[] not returned by server".to_string())
})?;
use mail_parser::MimeHeaders;
let parsed = mail_parser::MessageParser::default()
.parse(raw)
.ok_or_else(|| {
ImapError::ProtocolError("failed to parse RFC822 message".to_string())
})?;
let text_plain: Option<String> = parsed.body_text(0).map(|s| s.to_string());
let has_html_part = parsed.parts.iter().any(|part| {
part.content_type()
.is_some_and(|ct| ct.ctype() == "text" && ct.subtype() == Some("html"))
});
let text_html: Option<String> = if has_html_part {
parsed.body_html(0).map(|s| s.to_string())
} else {
None
};
tracing::debug!(
uid = uid,
folder = %folder,
total_parts = parsed.parts.len(),
attachment_count = parsed.attachments().count(),
has_text = text_plain.is_some(),
has_html = text_html.is_some(),
"fetch_body: parsed message structure"
);
let mut attachments = Vec::new();
for attachment in parsed.attachments() {
let filename: Option<String> = attachment.attachment_name().map(|s| s.to_string());
let content_type: String = attachment.content_type().map_or_else(
|| "application/octet-stream".to_string(),
|ct: &mail_parser::ContentType<'_>| {
if let Some(subtype) = ct.subtype() {
format!("{}/{}", ct.ctype(), subtype)
} else {
ct.ctype().to_string()
}
},
);
let content_id = attachment
.content_id()
.map(|s| s.trim_matches(|c| c == '<' || c == '>').to_string());
let data = attachment.contents().to_vec();
let size = data.len();
attachments.push(ImapAttachment {
filename,
content_type,
size,
data,
content_id,
});
}
for part in parsed.parts.iter() {
if part.content_id().is_none() {
continue;
}
let is_text = part.content_type().is_some_and(|ct| ct.ctype() == "text");
if is_text {
continue;
}
let cid = part
.content_id()
.unwrap()
.trim_matches(|c| c == '<' || c == '>')
.to_string();
if attachments
.iter()
.any(|a| a.content_id.as_deref() == Some(&cid))
{
continue;
}
let content_type: String = part.content_type().map_or_else(
|| "application/octet-stream".to_string(),
|ct: &mail_parser::ContentType<'_>| {
if let Some(subtype) = ct.subtype() {
format!("{}/{}", ct.ctype(), subtype)
} else {
ct.ctype().to_string()
}
},
);
let data = part.contents().to_vec();
let size = data.len();
attachments.push(ImapAttachment {
filename: part.attachment_name().map(|s| s.to_string()),
content_type,
size,
data,
content_id: Some(cid),
});
}
let raw_str = String::from_utf8_lossy(raw);
let raw_headers = raw_str
.split_once("\r\n\r\n")
.or_else(|| raw_str.split_once("\n\n"))
.map_or_else(|| raw_str.to_string(), |(h, _)| h.to_string());
ImapMessageBody {
uid,
text_plain,
text_html,
attachments,
raw_headers,
}
};
session.recycle();
Ok(body)
}
async fn fetch_raw_message(
&self,
creds: &ImapCredentials,
folder: &str,
uid: u32,
) -> Result<Vec<u8>, ImapError> {
let mut session = self.pool.checkout(creds).await?;
session.select(folder).await.map_err(map_imap_error)?;
let uid_str = uid.to_string();
let raw = {
let mut stream = session
.uid_fetch(&uid_str, "(UID BODY.PEEK[])")
.await
.map_err(map_imap_error)?;
let mut bytes = None;
while let Some(result) = stream.next().await {
let fetch = result.map_err(map_imap_error)?;
if let Some(body) = fetch.body() {
bytes = Some(body.to_vec());
break;
}
}
bytes.ok_or(ImapError::MessageNotFound {
uid,
folder: folder.to_string(),
})?
};
session.recycle();
Ok(raw)
}
async fn add_flags(
&self,
creds: &ImapCredentials,
folder: &str,
uid: u32,
flags: &[&str],
) -> Result<(), ImapError> {
let mut session = connect(creds).await?;
session.select(folder).await.map_err(map_imap_error)?;
let uid_str = uid.to_string();
let flags_str = format!("+FLAGS ({})", flags.join(" "));
{
let mut store_stream = session
.uid_store(&uid_str, &flags_str)
.await
.map_err(map_imap_error)?;
while let Some(result) = store_stream.next().await {
result.map_err(map_imap_error)?;
}
}
let _ = session.logout().await;
Ok(())
}
async fn remove_flags(
&self,
creds: &ImapCredentials,
folder: &str,
uid: u32,
flags: &[&str],
) -> Result<(), ImapError> {
let mut session = connect(creds).await?;
session.select(folder).await.map_err(map_imap_error)?;
let uid_str = uid.to_string();
let flags_str = format!("-FLAGS ({})", flags.join(" "));
{
let mut store_stream = session
.uid_store(&uid_str, &flags_str)
.await
.map_err(map_imap_error)?;
while let Some(result) = store_stream.next().await {
result.map_err(map_imap_error)?;
}
}
let _ = session.logout().await;
Ok(())
}
async fn set_flags(
&self,
creds: &ImapCredentials,
folder: &str,
uid: u32,
flags: &[&str],
) -> Result<(), ImapError> {
let mut session = connect(creds).await?;
session.select(folder).await.map_err(map_imap_error)?;
let uid_str = uid.to_string();
let flags_str = format!("FLAGS ({})", flags.join(" "));
{
let mut store_stream = session
.uid_store(&uid_str, &flags_str)
.await
.map_err(map_imap_error)?;
while let Some(result) = store_stream.next().await {
result.map_err(map_imap_error)?;
}
}
let _ = session.logout().await;
Ok(())
}
async fn move_message(
&self,
creds: &ImapCredentials,
from_folder: &str,
uid: u32,
to_folder: &str,
) -> Result<(), ImapError> {
let mut session = connect(creds).await?;
session.select(from_folder).await.map_err(map_imap_error)?;
let uid_str = uid.to_string();
match session.uid_mv(&uid_str, to_folder).await {
Ok(()) => {}
Err(async_imap::error::Error::No(_) | async_imap::error::Error::Bad(_)) => {
session
.uid_copy(&uid_str, to_folder)
.await
.map_err(map_imap_error)?;
{
let mut store_stream = session
.uid_store(&uid_str, "+FLAGS (\\Deleted)")
.await
.map_err(map_imap_error)?;
while let Some(r) = store_stream.next().await {
r.map_err(map_imap_error)?;
}
}
{
let expunge_stream = session.expunge().await.map_err(map_imap_error)?;
let mut expunge_stream = std::pin::pin!(expunge_stream);
while let Some(r) = expunge_stream.next().await {
r.map_err(map_imap_error)?;
}
}
}
Err(e) => return Err(map_imap_error(e)),
}
let _ = session.logout().await;
Ok(())
}
async fn copy_message(
&self,
creds: &ImapCredentials,
from_folder: &str,
uid: u32,
to_folder: &str,
) -> Result<(), ImapError> {
let mut session = connect(creds).await?;
session.select(from_folder).await.map_err(map_imap_error)?;
session
.uid_copy(&uid.to_string(), to_folder)
.await
.map_err(map_imap_error)?;
let _ = session.logout().await;
Ok(())
}
async fn mark_all_read(&self, creds: &ImapCredentials, folder: &str) -> Result<(), ImapError> {
let mut session = connect(creds).await?;
session.select(folder).await.map_err(map_imap_error)?;
{
let mut stream = session
.uid_store("1:*", "+FLAGS (\\Seen)")
.await
.map_err(map_imap_error)?;
while let Some(result) = stream.next().await {
result.map_err(map_imap_error)?;
}
}
let _ = session.logout().await;
Ok(())
}
async fn empty_folder(&self, creds: &ImapCredentials, folder: &str) -> Result<(), ImapError> {
let mut session = connect(creds).await?;
session.select(folder).await.map_err(map_imap_error)?;
{
let mut stream = session
.uid_store("1:*", "+FLAGS (\\Deleted)")
.await
.map_err(map_imap_error)?;
while let Some(result) = stream.next().await {
result.map_err(map_imap_error)?;
}
}
{
let stream = session.expunge().await.map_err(map_imap_error)?;
let mut stream = std::pin::pin!(stream);
while let Some(result) = stream.next().await {
result.map_err(map_imap_error)?;
}
}
let _ = session.logout().await;
Ok(())
}
async fn expunge_message(
&self,
creds: &ImapCredentials,
folder: &str,
uid: u32,
) -> Result<(), ImapError> {
let mut session = connect(creds).await?;
session.select(folder).await.map_err(map_imap_error)?;
let uid_str = uid.to_string();
{
let mut store_stream = session
.uid_store(&uid_str, "+FLAGS (\\Deleted)")
.await
.map_err(map_imap_error)?;
while let Some(r) = store_stream.next().await {
r.map_err(map_imap_error)?;
}
}
let uid_expunge_ok = {
match session.uid_expunge(&uid_str).await {
Ok(stream) => {
let mut stream = std::pin::pin!(stream);
while let Some(r) = stream.next().await {
r.map_err(map_imap_error)?;
}
true
}
Err(_) => false,
}
};
if !uid_expunge_ok {
let stream = session.expunge().await.map_err(map_imap_error)?;
let mut stream = std::pin::pin!(stream);
while let Some(r) = stream.next().await {
r.map_err(map_imap_error)?;
}
}
let _ = session.logout().await;
Ok(())
}
async fn append_message(
&self,
creds: &ImapCredentials,
folder: &str,
message_bytes: &[u8],
flags: &[&str],
) -> Result<(), ImapError> {
let mut session = connect(creds).await?;
let flags_str: Vec<String> = flags.iter().map(|f| f.to_string()).collect();
let flags_joined = if flags_str.is_empty() {
None
} else {
Some(format!("({})", flags_str.join(" ")))
};
session
.append(folder, flags_joined.as_deref(), None, message_bytes)
.await
.map_err(map_imap_error)?;
let _ = session.logout().await;
Ok(())
}
async fn append_message_resolved_uid(
&self,
creds: &ImapCredentials,
folder: &str,
message_bytes: &[u8],
flags: &[&str],
identity_header: &str,
identity_value: &str,
) -> Result<(u32, u32), ImapError> {
fn quote_search(value: &str) -> String {
value.replace('\\', "\\\\").replace('"', "\\\"")
}
if identity_header.is_empty()
|| !identity_header
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || byte == b'-')
|| identity_value.contains('\r')
|| identity_value.contains('\n')
{
return Err(ImapError::ProtocolError(
"Invalid draft identity header".to_string(),
));
}
let mut session = connect(creds).await?;
let mailbox = session.select(folder).await.map_err(map_imap_error)?;
let uid_validity = mailbox.uid_validity.unwrap_or(0);
let uid_next_before = mailbox.uid_next.unwrap_or(0);
let flags_str: Vec<String> = flags.iter().map(|flag| flag.to_string()).collect();
let flags_joined = if flags_str.is_empty() {
None
} else {
Some(format!("({})", flags_str.join(" ")))
};
session
.append(folder, flags_joined.as_deref(), None, message_bytes)
.await
.map_err(map_imap_error)?;
let query = format!(
"HEADER {} \"{}\"",
identity_header,
quote_search(identity_value)
);
let matches = session.uid_search(query).await.map_err(map_imap_error)?;
let mut candidates: Vec<u32> = matches
.into_iter()
.filter(|uid| uid_next_before == 0 || *uid >= uid_next_before)
.collect();
candidates.sort_unstable();
let resolved = match candidates.as_slice() {
[uid] => *uid,
[] => {
let _ = session.logout().await;
return Err(ImapError::ProtocolError(
"IMAP APPEND succeeded but the new draft UID could not be resolved".to_string(),
));
}
_ => {
let _ = session.logout().await;
return Err(ImapError::ProtocolError(
"IMAP APPEND succeeded but more than one matching draft UID was returned"
.to_string(),
));
}
};
let _ = session.logout().await;
Ok((resolved, uid_validity))
}
async fn create_folder(
&self,
creds: &ImapCredentials,
folder_name: &str,
) -> Result<(), ImapError> {
let mut session = connect(creds).await?;
session.create(folder_name).await.map_err(map_imap_error)?;
session
.subscribe(folder_name)
.await
.map_err(map_imap_error)?;
session.logout().await.ok();
Ok(())
}
async fn rename_folder(
&self,
creds: &ImapCredentials,
from: &str,
to: &str,
) -> Result<(), ImapError> {
let mut session = connect(creds).await?;
session.rename(from, to).await.map_err(map_imap_error)?;
session.logout().await.ok();
Ok(())
}
async fn delete_folder(
&self,
creds: &ImapCredentials,
folder_name: &str,
) -> Result<(), ImapError> {
let mut session = connect(creds).await?;
session.delete(folder_name).await.map_err(map_imap_error)?;
session.logout().await.ok();
Ok(())
}
async fn subscribe_folder(
&self,
creds: &ImapCredentials,
folder_name: &str,
subscribe: bool,
) -> Result<(), ImapError> {
let mut session = connect(creds).await?;
if subscribe {
session
.subscribe(folder_name)
.await
.map_err(map_imap_error)?;
} else {
session
.unsubscribe(folder_name)
.await
.map_err(map_imap_error)?;
}
session.logout().await.ok();
Ok(())
}
async fn fetch_uids_and_flags(
&self,
creds: &ImapCredentials,
folder: &str,
) -> Result<Vec<(u32, Vec<String>)>, ImapError> {
let mut session = self.pool.checkout(creds).await?;
session.select(folder).await.map_err(|e| match &e {
async_imap::error::Error::No(msg)
if msg.to_lowercase().contains("not found")
|| msg.to_lowercase().contains("doesn't exist")
|| msg.to_lowercase().contains("does not exist")
|| msg.to_lowercase().contains("no such") =>
{
ImapError::FolderNotFound(folder.to_string())
}
_ => map_imap_error(e),
})?;
let results = {
let mut fetch_stream = session
.uid_fetch("1:*", "(UID FLAGS)")
.await
.map_err(map_imap_error)?;
let mut items = Vec::new();
while let Some(result) = fetch_stream.next().await {
let fetch = result.map_err(map_imap_error)?;
if let Some(uid) = fetch.uid {
let flags: Vec<String> = fetch.flags().map(|f| flag_to_string(&f)).collect();
items.push((uid, flags));
}
}
items
};
session.recycle();
Ok(results)
}
async fn search_uids(
&self,
creds: &ImapCredentials,
folder: &str,
criteria: &str,
) -> Result<Vec<u32>, ImapError> {
let mut session = self.pool.checkout(creds).await?;
session.select(folder).await.map_err(|e| match &e {
async_imap::error::Error::No(msg)
if msg.to_lowercase().contains("not found")
|| msg.to_lowercase().contains("doesn't exist")
|| msg.to_lowercase().contains("does not exist")
|| msg.to_lowercase().contains("no such") =>
{
ImapError::FolderNotFound(folder.to_string())
}
_ => map_imap_error(e),
})?;
let matches = session.uid_search(criteria).await.map_err(map_imap_error)?;
session.recycle();
let mut uids: Vec<u32> = matches.into_iter().collect();
uids.sort_unstable();
Ok(uids)
}
async fn folder_status_extended(
&self,
creds: &ImapCredentials,
folder: &str,
) -> Result<FolderStatusExtended, ImapError> {
let mut session = self.pool.checkout(creds).await?;
let result = folder_status_extended_on_session(&mut session, folder).await;
if result.is_ok() {
session.recycle();
}
result
}
async fn folder_statuses_extended(
&self,
creds: &ImapCredentials,
folders: &[String],
) -> Result<Vec<(String, FolderStatusExtended)>, ImapError> {
let mut session = self.pool.checkout(creds).await?;
let mut statuses = Vec::with_capacity(folders.len());
for folder in folders {
match folder_status_extended_on_session(&mut session, folder).await {
Ok(status) => statuses.push((folder.clone(), status)),
Err(error) => {
tracing::debug!(folder = %folder, error = %error, "Skipping folder STATUS refresh");
}
}
}
session.recycle();
Ok(statuses)
}
async fn fetch_changed_flags(
&self,
creds: &ImapCredentials,
folder: &str,
since_modseq: u64,
) -> Result<(Vec<(u32, Vec<String>)>, u64), ImapError> {
let mut session = self.pool.checkout(creds).await?;
let mailbox = session
.select_condstore(folder)
.await
.map_err(map_imap_error)?;
let new_modseq = mailbox.highest_modseq.unwrap_or(0);
let items = {
let mut fetch_stream = session
.uid_fetch("1:*", format!("(UID FLAGS) (CHANGEDSINCE {since_modseq})"))
.await
.map_err(map_imap_error)?;
let mut items = Vec::new();
while let Some(result) = fetch_stream.next().await {
let fetch = result.map_err(map_imap_error)?;
if let Some(uid) = fetch.uid {
let flags: Vec<String> = fetch.flags().map(|f| flag_to_string(&f)).collect();
items.push((uid, flags));
}
}
items
};
session.recycle();
Ok((items, new_modseq))
}
async fn get_quota(&self, creds: &ImapCredentials) -> Result<Option<MailboxQuota>, ImapError> {
use async_imap::imap_proto::{Response, Status};
let mut session = self.pool.checkout(creds).await?;
let req_id = session
.run_command("GETQUOTAROOT INBOX")
.await
.map_err(map_imap_error)?;
let (quota, reusable) = tokio::time::timeout(std::time::Duration::from_secs(15), async {
let mut quota = None;
while let Some(response) = session.read_response().await {
let response = response.map_err(|e| ImapError::ConnectionFailed(e.to_string()))?;
match response.parsed() {
Response::Quota(q) => {
for resource in &q.resources {
if matches!(
resource.name,
async_imap::imap_proto::types::QuotaResourceName::Storage
) {
quota = Some(MailboxQuota {
usage_bytes: resource.usage * 1024,
limit_bytes: resource.limit * 1024,
});
}
}
}
Response::Done { tag, status, .. } if *tag == req_id => {
return match status {
Status::Ok => Ok((quota, true)),
Status::No | Status::Bad => Ok((None, true)),
_ => Err(ImapError::ProtocolError(
"unexpected GETQUOTAROOT status".into(),
)),
};
}
Response::Data {
status: Status::Bye,
..
} => break,
_ => {}
}
}
Err(ImapError::ConnectionFailed(
"connection lost during GETQUOTAROOT".into(),
))
})
.await
.map_err(|_| ImapError::ConnectionFailed("GETQUOTAROOT timed out after 15s".into()))??;
if reusable {
session.recycle();
}
Ok(quota)
}
async fn fetch_folder_size(
&self,
creds: &ImapCredentials,
folder: &str,
) -> Result<u64, ImapError> {
let mut session = self.pool.checkout(creds).await?;
let mailbox = session.select(folder).await.map_err(map_imap_error)?;
if mailbox.exists == 0 {
session.recycle();
return Ok(0);
}
let mut total: u64 = 0;
let fetch_result = tokio::time::timeout(std::time::Duration::from_secs(60), async {
use async_imap::imap_proto::{Response, Status, types::AttributeValue};
let req_id = session
.run_command("UID FETCH 1:* (RFC822.SIZE)")
.await
.map_err(map_imap_error)?;
while let Some(response) = session.read_response().await {
let response = response.map_err(|e| ImapError::ConnectionFailed(e.to_string()))?;
match response.parsed() {
Response::Fetch(_, attributes) => {
for attribute in attributes {
if let AttributeValue::Rfc822Size(size) = attribute {
total += u64::from(*size);
}
}
}
Response::Done { tag, status, .. } if *tag == req_id => {
return if *status == Status::Ok {
Ok(())
} else {
Err(ImapError::ProtocolError(
"folder-size FETCH rejected".into(),
))
};
}
Response::Data {
status: Status::Bye,
..
} => break,
_ => {}
}
}
Err(ImapError::ConnectionFailed(
"connection lost during folder-size FETCH".into(),
))
})
.await;
match fetch_result {
Ok(Err(e)) => {
return Err(e);
}
Err(_) => {
tracing::warn!(folder = %folder, "fetch_folder_size timed out after 60s");
return Err(ImapError::ConnectionFailed(
"folder-size FETCH timed out after 60s".into(),
));
}
Ok(Ok(())) => session.recycle(),
}
Ok(total)
}
async fn folder_keyword_support(
&self,
creds: &ImapCredentials,
folder: &str,
) -> Result<bool, ImapError> {
let mut session = self.pool.checkout(creds).await?;
let mailbox = session.select(folder).await.map_err(map_imap_error)?;
let supported = mailbox.permanent_flags.is_empty()
|| mailbox
.permanent_flags
.iter()
.any(|flag| matches!(flag, async_imap::types::Flag::MayCreate));
session.recycle();
Ok(supported)
}
}
#[cfg(test)]
mod date_resolution_tests {
use super::resolve_message_date;
#[test]
fn uses_internal_date_when_message_date_is_missing() {
assert_eq!(
resolve_message_date(None, Some("Tue, 3 Nov 2026 14:38:39 +0100".to_string())),
Some("Tue, 3 Nov 2026 14:38:39 +0100".to_string())
);
}
#[test]
fn uses_internal_date_when_message_date_is_invalid() {
assert_eq!(
resolve_message_date(
Some("not-a-date".to_string()),
Some("Tue, 3 Nov 2026 14:38:39 +0100".to_string())
),
Some("Tue, 3 Nov 2026 14:38:39 +0100".to_string())
);
}
#[test]
fn keeps_valid_message_date() {
assert_eq!(
resolve_message_date(
Some("Mon, 2 Nov 2026 10:00:00 +0100".to_string()),
Some("Tue, 3 Nov 2026 14:38:39 +0100".to_string())
),
Some("Mon, 2 Nov 2026 10:00:00 +0100".to_string())
);
}
}