2404 lines
86 KiB
Rust
2404 lines
86 KiB
Rust
use std::sync::Arc;
|
|
|
|
use axum::extract::Multipart;
|
|
use axum::extract::Path;
|
|
use axum::extract::Query;
|
|
use axum::response::{IntoResponse, Response};
|
|
use axum::{Extension, Json};
|
|
|
|
pub mod types;
|
|
use types::*;
|
|
|
|
use crate::auth::session::SessionState;
|
|
use crate::config::AppConfig;
|
|
use crate::db;
|
|
use crate::db::header_sync::HeaderSync;
|
|
use crate::email_theme;
|
|
use crate::error::AppError;
|
|
use crate::imap::client::{ImapClient, ImapCredentials};
|
|
use crate::realtime::events::{EventBus, MailEvent, MessageFlagChange};
|
|
use crate::routes::folder_counts;
|
|
use crate::search::engine::{IndexableMessage, SearchEngine, UserIndex};
|
|
|
|
fn build_creds(session: &SessionState, _config: &AppConfig) -> Result<ImapCredentials, AppError> {
|
|
Ok(ImapCredentials {
|
|
host: session.imap_host.clone(),
|
|
port: session.imap_port,
|
|
tls: session.imap_tls,
|
|
tls_verify: session.imap_tls_verify,
|
|
email: session.email.clone(),
|
|
password: session.password.clone(),
|
|
})
|
|
}
|
|
|
|
fn is_valid_flag(flag: &str) -> bool {
|
|
if flag.is_empty() || flag.len() > 64 {
|
|
return false;
|
|
}
|
|
|
|
let atom = flag.strip_prefix('\\').unwrap_or(flag);
|
|
!atom.is_empty()
|
|
&& atom
|
|
.bytes()
|
|
.all(|b| b.is_ascii_alphanumeric() || matches!(b, b'$' | b'_' | b'-' | b'+' | b'.'))
|
|
}
|
|
|
|
fn validate_flags(flags: &[String]) -> Result<(), AppError> {
|
|
for flag in flags {
|
|
if !is_valid_flag(flag) {
|
|
return Err(AppError::BadRequest("invalid flag".to_string()));
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
const FOLDER_MESSAGES_TTL_SECS: u32 = 30;
|
|
|
|
pub async fn list_messages(
|
|
Extension(session): Extension<SessionState>,
|
|
Extension(config): Extension<Arc<AppConfig>>,
|
|
Extension(imap_client): Extension<Arc<dyn ImapClient>>,
|
|
Extension(search_engine): Extension<Arc<SearchEngine>>,
|
|
Extension(event_bus): Extension<Arc<EventBus>>,
|
|
Path(folder): Path<String>,
|
|
Query(query): Query<ListMessagesQuery>,
|
|
) -> Result<Response, AppError> {
|
|
if !(1..=300).contains(&query.per_page) {
|
|
return Err(AppError::BadRequest(
|
|
"per_page must be between 1 and 300".to_string(),
|
|
));
|
|
}
|
|
if !matches!(
|
|
query.sort.as_str(),
|
|
"date_desc"
|
|
| "date_asc"
|
|
| "sender_asc"
|
|
| "sender_desc"
|
|
| "subject_asc"
|
|
| "subject_desc"
|
|
| "size_asc"
|
|
| "size_desc"
|
|
) {
|
|
return Err(AppError::BadRequest("unsupported message sort".to_string()));
|
|
}
|
|
|
|
let mut syncing = false;
|
|
|
|
let database = db::pool::UserDb::new(&config.data_dir, &session.user_hash);
|
|
let cached_folder_name = folder.clone();
|
|
let (
|
|
folder_fresh,
|
|
cached_uid_validity,
|
|
cached_count,
|
|
cached_server_total,
|
|
cached_counter_generation,
|
|
cached_highest_modseq,
|
|
folder_invalidated,
|
|
mut header_sync,
|
|
) = database
|
|
.run(move |conn| {
|
|
let header_sync = HeaderSync::capture(conn)?;
|
|
let fresh =
|
|
db::folders::is_folder_fresh(conn, &cached_folder_name, FOLDER_MESSAGES_TTL_SECS)?;
|
|
let cached = db::folders::get_folder(conn, &cached_folder_name)?;
|
|
let validity = cached.as_ref().map(|f| f.uid_validity).unwrap_or(0);
|
|
let server_total = cached.as_ref().map(|f| f.total_count).unwrap_or(0);
|
|
let counter_generation = cached.as_ref().map(|f| f.counter_generation).unwrap_or(0);
|
|
let highest_modseq = cached.as_ref().map(|f| f.highest_modseq).unwrap_or(0);
|
|
if cached.is_none() {
|
|
db::folders::upsert_folder(
|
|
conn,
|
|
&cached_folder_name,
|
|
None,
|
|
None,
|
|
"",
|
|
true,
|
|
0,
|
|
0,
|
|
0,
|
|
0,
|
|
)?;
|
|
}
|
|
Ok((
|
|
fresh,
|
|
validity,
|
|
db::messages::count_messages(conn, &cached_folder_name)?,
|
|
server_total,
|
|
counter_generation,
|
|
highest_modseq,
|
|
db::folders::is_folder_messages_invalidated(conn, &cached_folder_name)?,
|
|
header_sync,
|
|
))
|
|
})
|
|
.await
|
|
.map_err(AppError::InternalError)?;
|
|
|
|
if query.refresh || !folder_fresh || folder_invalidated || cached_count < cached_server_total {
|
|
let creds = build_creds(&session, &config)?;
|
|
|
|
let status = imap_client
|
|
.folder_status_extended(&creds, &folder)
|
|
.await
|
|
.map_err(|e| AppError::ServiceUnavailable(format!("IMAP error: {e}")))?;
|
|
|
|
tracing::info!(
|
|
folder = %folder,
|
|
cached_uid_validity = cached_uid_validity,
|
|
imap_uid_validity = status.uid_validity,
|
|
cached_count = cached_count,
|
|
imap_exists = status.exists,
|
|
cached_highest_modseq = cached_highest_modseq,
|
|
imap_highest_modseq = status.highest_modseq,
|
|
"list_messages: folder status check"
|
|
);
|
|
|
|
let needs_full_sync =
|
|
cached_uid_validity != 0 && cached_uid_validity != status.uid_validity;
|
|
let modseq_changed = !needs_full_sync
|
|
&& cached_highest_modseq > 0
|
|
&& status.highest_modseq > 0
|
|
&& cached_highest_modseq != status.highest_modseq;
|
|
|
|
if !needs_full_sync {
|
|
let name = folder.clone();
|
|
let expected_generation = cached_counter_generation;
|
|
let total = status.exists;
|
|
let unread = status.unseen;
|
|
database
|
|
.run(move |conn| {
|
|
db::folders::set_folder_counts_if_unchanged(
|
|
conn,
|
|
&name,
|
|
expected_generation,
|
|
total,
|
|
unread,
|
|
)
|
|
.map(|_| ())
|
|
})
|
|
.await
|
|
.map_err(AppError::InternalError)?;
|
|
}
|
|
|
|
if needs_full_sync {
|
|
tracing::info!(folder = %folder, "UIDVALIDITY changed, clearing cache");
|
|
let name = folder.clone();
|
|
let uid_validity = status.uid_validity;
|
|
let exists = status.exists;
|
|
let highest_modseq = status.highest_modseq;
|
|
let unseen = status.unseen;
|
|
header_sync = database
|
|
.run(move |conn| {
|
|
let transaction = header_sync.begin_write(conn)?;
|
|
db::messages::delete_folder_messages(&transaction, &name)?;
|
|
db::header_sync::invalidate(&transaction)?;
|
|
db::folders::update_folder_sync_status(
|
|
&transaction,
|
|
&name,
|
|
uid_validity,
|
|
exists,
|
|
highest_modseq,
|
|
)?;
|
|
db::folders::set_unread_count(&transaction, &name, unseen)?;
|
|
let next = HeaderSync::capture(&transaction)?;
|
|
transaction.commit().map_err(|e| e.to_string())?;
|
|
Ok(next)
|
|
})
|
|
.await
|
|
.map_err(AppError::InternalError)?;
|
|
}
|
|
|
|
let needs_sync = needs_full_sync
|
|
|| cached_count == 0
|
|
|| status.exists != cached_count
|
|
|| folder_invalidated
|
|
|| modseq_changed;
|
|
|
|
if needs_sync {
|
|
if (needs_full_sync || cached_count == 0) && status.exists > 0 {
|
|
let recent_start = status.uid_next.saturating_sub(100).max(1);
|
|
let recent_range = format!("{recent_start}:*");
|
|
|
|
tracing::info!(folder = %folder, uid_range = %recent_range, "Partial sync: fetching recent headers");
|
|
|
|
let headers = imap_client
|
|
.fetch_headers(&creds, &folder, &recent_range)
|
|
.await
|
|
.map_err(|e| AppError::ServiceUnavailable(format!("IMAP error: {e}")))?;
|
|
|
|
tracing::info!(folder = %folder, fetched = headers.len(), "Partial sync: fetched recent headers");
|
|
|
|
let name = folder.clone();
|
|
let engine = search_engine.clone();
|
|
let hash = session.user_hash.clone();
|
|
let validity = status.uid_validity;
|
|
let highest_modseq = status.highest_modseq;
|
|
database
|
|
.run(move |conn| {
|
|
let transaction = header_sync.begin_write(conn)?;
|
|
upsert_and_index_headers(&transaction, &name, &headers, &engine, &hash)
|
|
.map_err(|e| e.to_string())?;
|
|
db::folders::update_folder_sync_cursor(
|
|
&transaction,
|
|
&name,
|
|
validity,
|
|
highest_modseq,
|
|
)?;
|
|
transaction.commit().map_err(|e| e.to_string())
|
|
})
|
|
.await
|
|
.map_err(AppError::InternalError)?;
|
|
|
|
if recent_start > 1 {
|
|
let remaining_range = (1, recent_start - 1);
|
|
tokio::spawn(sync_remaining_headers(BackgroundSyncParams {
|
|
creds,
|
|
folder: folder.clone(),
|
|
imap_client: imap_client.clone(),
|
|
config: config.clone(),
|
|
user_hash: session.user_hash.clone(),
|
|
search_engine: search_engine.clone(),
|
|
event_bus: event_bus.clone(),
|
|
uid_range: remaining_range,
|
|
header_sync,
|
|
uid_validity: status.uid_validity,
|
|
highest_modseq: status.highest_modseq,
|
|
}));
|
|
syncing = true;
|
|
}
|
|
} else {
|
|
let uid_range = (1, status.uid_next.saturating_sub(1));
|
|
|
|
tokio::spawn(sync_remaining_headers(BackgroundSyncParams {
|
|
creds,
|
|
folder: folder.clone(),
|
|
imap_client: imap_client.clone(),
|
|
config: config.clone(),
|
|
user_hash: session.user_hash.clone(),
|
|
search_engine: search_engine.clone(),
|
|
event_bus: event_bus.clone(),
|
|
uid_range,
|
|
header_sync,
|
|
uid_validity: status.uid_validity,
|
|
highest_modseq: status.highest_modseq,
|
|
}));
|
|
syncing = true;
|
|
}
|
|
} else {
|
|
let name = folder.clone();
|
|
let uid_validity = status.uid_validity;
|
|
let highest_modseq = status.highest_modseq;
|
|
database
|
|
.run(move |conn| {
|
|
let transaction = header_sync.begin_write(conn)?;
|
|
db::folders::update_folder_sync_cursor(
|
|
&transaction,
|
|
&name,
|
|
uid_validity,
|
|
highest_modseq,
|
|
)?;
|
|
db::folders::mark_folder_messages_fresh(&transaction, &name)?;
|
|
transaction.commit().map_err(|e| e.to_string())
|
|
})
|
|
.await
|
|
.map_err(AppError::InternalError)?;
|
|
}
|
|
}
|
|
|
|
let name = folder.clone();
|
|
let (messages, total_count, tags_map, flagged_on_top, folder_total_count, unread_count) =
|
|
database
|
|
.run(move |conn| {
|
|
let flagged_on_top = db::folder_preferences::effective_flagged_on_top(conn, &name)?;
|
|
let messages = db::messages::get_messages_sorted(
|
|
conn,
|
|
&name,
|
|
query.page,
|
|
query.per_page,
|
|
&query.sort,
|
|
flagged_on_top,
|
|
)?;
|
|
let total = db::messages::count_messages(conn, &name)?;
|
|
let refs: Vec<_> = messages
|
|
.iter()
|
|
.map(|m| (m.uid, m.folder.as_str()))
|
|
.collect();
|
|
let tags = db::tags::get_tags_for_messages(conn, &refs)?;
|
|
let folder_state = db::folders::get_folder(conn, &name)?;
|
|
let folder_total_count = folder_state
|
|
.as_ref()
|
|
.map(|item| item.total_count)
|
|
.unwrap_or(total);
|
|
let unread_count = folder_state
|
|
.as_ref()
|
|
.map(|item| item.unread_count)
|
|
.unwrap_or(0);
|
|
Ok((
|
|
messages,
|
|
total,
|
|
tags,
|
|
flagged_on_top,
|
|
folder_total_count,
|
|
unread_count,
|
|
))
|
|
})
|
|
.await
|
|
.map_err(AppError::InternalError)?;
|
|
|
|
tracing::info!(
|
|
folder = %folder,
|
|
total_count = total_count,
|
|
page_messages = messages.len(),
|
|
page = query.page,
|
|
per_page = query.per_page,
|
|
"list_messages: returning results"
|
|
);
|
|
|
|
let summaries: Vec<MessageSummary> = messages
|
|
.into_iter()
|
|
.map(|m| {
|
|
let msg_tags = tags_map
|
|
.get(&(m.uid, m.folder.clone()))
|
|
.cloned()
|
|
.unwrap_or_default();
|
|
MessageSummary {
|
|
uid: m.uid,
|
|
folder: m.folder,
|
|
subject: m.subject,
|
|
from_address: m.from_address,
|
|
from_name: m.from_name,
|
|
to_addresses: m.to_addresses,
|
|
date: m.date,
|
|
flags: m.flags,
|
|
size: m.size,
|
|
has_attachments: m.has_attachments,
|
|
snippet: m.snippet,
|
|
reaction: m.reaction,
|
|
tags: msg_tags,
|
|
}
|
|
})
|
|
.collect();
|
|
|
|
Ok(Json(ListMessagesResponse {
|
|
messages: summaries,
|
|
total_count,
|
|
page: query.page,
|
|
per_page: query.per_page,
|
|
syncing,
|
|
flagged_on_top,
|
|
folder_total_count,
|
|
unread_count,
|
|
})
|
|
.into_response())
|
|
}
|
|
|
|
fn upsert_and_index_headers(
|
|
conn: &crate::db::pool::Transaction,
|
|
folder: &str,
|
|
headers: &[crate::imap::types::ImapMessageHeader],
|
|
search_engine: &Arc<SearchEngine>,
|
|
user_hash: &str,
|
|
) -> Result<(), AppError> {
|
|
let sync_keywords = db::mail_preferences::get_preferences(conn)
|
|
.map(|prefs| prefs.sync_imap_keywords)
|
|
.unwrap_or(false);
|
|
for header in headers {
|
|
let from_address = header
|
|
.from
|
|
.first()
|
|
.map(|a| a.address.as_str())
|
|
.unwrap_or("");
|
|
let from_name = header
|
|
.from
|
|
.first()
|
|
.and_then(|a| a.name.as_deref())
|
|
.unwrap_or("");
|
|
let to_json = serde_json::to_string(&header.to).unwrap_or_else(|_| "[]".to_string());
|
|
let cc_json = serde_json::to_string(&header.cc).unwrap_or_else(|_| "[]".to_string());
|
|
let subject = header.subject.as_deref().unwrap_or("");
|
|
let date = header.date.as_deref().unwrap_or("");
|
|
let flags_csv = header.flags.join(",");
|
|
|
|
db::messages::upsert_message(
|
|
conn,
|
|
folder,
|
|
header.uid,
|
|
header.message_id.as_deref(),
|
|
header.in_reply_to.as_deref(),
|
|
header.references.as_deref(),
|
|
subject,
|
|
from_address,
|
|
from_name,
|
|
&to_json,
|
|
&cc_json,
|
|
date,
|
|
&flags_csv,
|
|
header.size,
|
|
header.has_attachments,
|
|
"",
|
|
header.reaction.as_deref(),
|
|
)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
|
|
|
db::contacts::populate_known_addresses(conn, from_address, from_name, &to_json, &cc_json)
|
|
.map_err(|e| AppError::InternalError(format!("Known addresses error: {e}")))?;
|
|
if sync_keywords {
|
|
db::tags::import_imap_keywords(conn, header.uid, folder, &header.flags)
|
|
.map_err(|e| AppError::InternalError(format!("IMAP keyword import error: {e}")))?;
|
|
}
|
|
}
|
|
|
|
if !headers.is_empty()
|
|
&& !UserIndex::is_excluded_folder(folder)
|
|
&& let Ok(user_index) = search_engine.open_user_index(user_hash)
|
|
{
|
|
let indexable: Vec<IndexableMessage> = headers
|
|
.iter()
|
|
.map(|h| {
|
|
let from_address = h.from.first().map(|a| a.address.as_str()).unwrap_or("");
|
|
let from_name = h.from.first().and_then(|a| a.name.as_deref()).unwrap_or("");
|
|
let subject = h.subject.as_deref().unwrap_or("");
|
|
let date = h.date.as_deref().unwrap_or("");
|
|
let to_json = serde_json::to_string(&h.to).unwrap_or_else(|_| "[]".to_string());
|
|
IndexableMessage {
|
|
uid: h.uid,
|
|
folder: folder.to_string(),
|
|
subject: subject.to_string(),
|
|
from_address: from_address.to_string(),
|
|
from_name: from_name.to_string(),
|
|
to_addresses: to_json,
|
|
body_text: String::new(),
|
|
date_epoch: crate::db::messages::parse_date_to_epoch_public(date),
|
|
has_attachments: h.has_attachments,
|
|
}
|
|
})
|
|
.collect();
|
|
user_index
|
|
.index_messages_batch_with_connection(conn, &indexable)
|
|
.map_err(|e| AppError::InternalError(format!("Search outbox error: {e}")))?;
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
fn header_chunks(uids: &[u32]) -> impl Iterator<Item = String> + '_ {
|
|
uids.chunks(250).map(|chunk| {
|
|
chunk
|
|
.iter()
|
|
.map(u32::to_string)
|
|
.collect::<Vec<_>>()
|
|
.join(",")
|
|
})
|
|
}
|
|
|
|
struct BackgroundSyncParams {
|
|
creds: ImapCredentials,
|
|
folder: String,
|
|
imap_client: Arc<dyn ImapClient>,
|
|
config: Arc<AppConfig>,
|
|
user_hash: String,
|
|
search_engine: Arc<SearchEngine>,
|
|
event_bus: Arc<EventBus>,
|
|
uid_range: (u32, u32),
|
|
header_sync: HeaderSync,
|
|
uid_validity: u32,
|
|
highest_modseq: u64,
|
|
}
|
|
|
|
async fn sync_remaining_headers(params: BackgroundSyncParams) {
|
|
let BackgroundSyncParams {
|
|
creds,
|
|
folder,
|
|
imap_client,
|
|
config,
|
|
user_hash,
|
|
search_engine,
|
|
event_bus,
|
|
uid_range,
|
|
header_sync,
|
|
uid_validity,
|
|
highest_modseq,
|
|
} = params;
|
|
let result: Result<
|
|
(
|
|
Option<crate::imap::types::FolderStatusExtended>,
|
|
Vec<MessageFlagChange>,
|
|
bool,
|
|
),
|
|
String,
|
|
> = crate::imap::pool::background(async {
|
|
let database = db::pool::UserDb::new(&config.data_dir, &user_hash);
|
|
let mut server_state = imap_client
|
|
.fetch_uids_and_flags(&creds, &folder)
|
|
.await
|
|
.map_err(|e| format!("IMAP error: {e}"))?;
|
|
for (_, flags) in &mut server_state {
|
|
flags.sort();
|
|
flags.dedup();
|
|
}
|
|
|
|
let name = folder.clone();
|
|
let cached_state = database
|
|
.run(move |conn| db::messages::get_all_uids_and_flags(conn, &name))
|
|
.await?;
|
|
let cached_uids = cached_state
|
|
.iter()
|
|
.map(|(uid, _)| *uid)
|
|
.collect::<std::collections::HashSet<_>>();
|
|
|
|
let mut uids = if uid_range.0 <= uid_range.1 {
|
|
server_state
|
|
.iter()
|
|
.map(|(uid, _)| *uid)
|
|
.filter(|uid| (uid_range.0..=uid_range.1).contains(uid))
|
|
.filter(|uid| !cached_uids.contains(uid))
|
|
.collect::<Vec<_>>()
|
|
} else {
|
|
Vec::new()
|
|
};
|
|
uids.sort_unstable();
|
|
uids.dedup();
|
|
|
|
event_bus
|
|
.publish(
|
|
&user_hash,
|
|
MailEvent::SyncStatus {
|
|
active: true,
|
|
folder: Some(folder.clone()),
|
|
pending: uids.len() as u32,
|
|
},
|
|
)
|
|
.await;
|
|
|
|
let mut pending_headers = uids.len() as u32;
|
|
for uid_set in header_chunks(&uids) {
|
|
let chunk_len = uid_set.split(',').count() as u32;
|
|
let headers = imap_client
|
|
.fetch_headers(&creds, &folder, &uid_set)
|
|
.await
|
|
.map_err(|e| format!("IMAP error: {e}"))?;
|
|
let name = folder.clone();
|
|
let engine = search_engine.clone();
|
|
let hash = user_hash.clone();
|
|
database
|
|
.run(move |conn| {
|
|
let transaction = header_sync.begin_write(conn)?;
|
|
upsert_and_index_headers(&transaction, &name, &headers, &engine, &hash)
|
|
.map_err(|e| e.to_string())?;
|
|
transaction.commit().map_err(|e| e.to_string())
|
|
})
|
|
.await?;
|
|
pending_headers = pending_headers.saturating_sub(chunk_len);
|
|
event_bus
|
|
.publish(
|
|
&user_hash,
|
|
MailEvent::SyncStatus {
|
|
active: true,
|
|
folder: Some(folder.clone()),
|
|
pending: pending_headers,
|
|
},
|
|
)
|
|
.await;
|
|
}
|
|
|
|
let expected_counts_name = folder.clone();
|
|
let expected_counts = database
|
|
.run(move |conn| {
|
|
Ok(
|
|
db::folders::get_folder(conn, &expected_counts_name)?.map(|state| {
|
|
(
|
|
state.total_count,
|
|
state.unread_count,
|
|
state.counter_generation,
|
|
)
|
|
}),
|
|
)
|
|
})
|
|
.await?;
|
|
let final_status = imap_client
|
|
.folder_status_extended(&creds, &folder)
|
|
.await
|
|
.ok();
|
|
let name = folder.clone();
|
|
let status_for_db = final_status.clone();
|
|
let server_flags = server_state
|
|
.into_iter()
|
|
.collect::<std::collections::HashMap<u32, Vec<String>>>();
|
|
let engine = search_engine.clone();
|
|
let hash = user_hash.clone();
|
|
let (flag_changes, structural_change) = database
|
|
.run(move |conn| {
|
|
let transaction = header_sync.begin_write(conn)?;
|
|
let sync_keywords = db::mail_preferences::get_preferences(&transaction)
|
|
.map(|prefs| prefs.sync_imap_keywords)
|
|
.unwrap_or(false);
|
|
let current = db::messages::get_all_uids_and_flags(&transaction, &name)?;
|
|
let user_index = engine.open_user_index(&hash).ok();
|
|
let mut flag_changes = Vec::new();
|
|
let mut structural_change = false;
|
|
|
|
for (uid, cached_flags_csv) in current {
|
|
let Some(server_flags_for_uid) = server_flags.get(&uid) else {
|
|
db::messages::delete_message(&transaction, &name, uid)?;
|
|
if let Some(index) = user_index.as_ref() {
|
|
index.delete_message_with_connection(&transaction, uid, &name)?;
|
|
}
|
|
structural_change = true;
|
|
continue;
|
|
};
|
|
|
|
let mut cached_flags = cached_flags_csv
|
|
.split(',')
|
|
.filter(|flag| !flag.is_empty())
|
|
.map(str::to_string)
|
|
.collect::<Vec<_>>();
|
|
cached_flags.sort();
|
|
cached_flags.dedup();
|
|
if cached_flags.as_slice() != server_flags_for_uid.as_slice() {
|
|
db::messages::update_message_flags(
|
|
&transaction,
|
|
&name,
|
|
uid,
|
|
&server_flags_for_uid.join(","),
|
|
)?;
|
|
flag_changes.push(MessageFlagChange {
|
|
uid,
|
|
flags: server_flags_for_uid.clone(),
|
|
});
|
|
}
|
|
if sync_keywords {
|
|
db::tags::import_imap_keywords(
|
|
&transaction,
|
|
uid,
|
|
&name,
|
|
server_flags_for_uid,
|
|
)?;
|
|
}
|
|
}
|
|
|
|
if let Some(status) = status_for_db {
|
|
if uid_validity == 0 || status.uid_validity == uid_validity {
|
|
db::folders::update_folder_sync_cursor(
|
|
&transaction,
|
|
&name,
|
|
status.uid_validity,
|
|
highest_modseq,
|
|
)?;
|
|
} else {
|
|
db::folders::invalidate_folder_freshness(&transaction, &name)?;
|
|
}
|
|
|
|
if let Some((_expected_total, _expected_unread, expected_generation)) =
|
|
expected_counts
|
|
{
|
|
let _ = db::folders::set_folder_counts_if_unchanged(
|
|
&transaction,
|
|
&name,
|
|
expected_generation,
|
|
status.exists,
|
|
status.unseen,
|
|
)?;
|
|
} else {
|
|
db::folders::set_folder_counts(
|
|
&transaction,
|
|
&name,
|
|
status.exists,
|
|
status.unseen,
|
|
)?;
|
|
}
|
|
|
|
let cached_after = db::messages::count_messages(&transaction, &name)?;
|
|
if (uid_validity == 0 || status.uid_validity == uid_validity)
|
|
&& cached_after == status.exists
|
|
{
|
|
db::folders::mark_folder_messages_fresh(&transaction, &name)?;
|
|
} else {
|
|
db::folders::invalidate_folder_freshness(&transaction, &name)?;
|
|
}
|
|
} else {
|
|
db::folders::invalidate_folder_freshness(&transaction, &name)?;
|
|
}
|
|
transaction.commit().map_err(|e| e.to_string())?;
|
|
Ok((flag_changes, structural_change))
|
|
})
|
|
.await?;
|
|
|
|
Ok((final_status, flag_changes, structural_change))
|
|
})
|
|
.await;
|
|
|
|
match result {
|
|
Ok((_final_status, flag_changes, structural_change)) => {
|
|
tracing::info!(folder = %folder, "Background sync: complete");
|
|
if !flag_changes.is_empty() {
|
|
event_bus
|
|
.publish(
|
|
&user_hash,
|
|
MailEvent::FlagsChanged {
|
|
folder: folder.clone(),
|
|
changes: flag_changes,
|
|
},
|
|
)
|
|
.await;
|
|
}
|
|
if structural_change {
|
|
event_bus
|
|
.publish(&user_hash, MailEvent::FolderUpdated)
|
|
.await;
|
|
}
|
|
|
|
let status_db = db::pool::UserDb::new(&config.data_dir, &user_hash);
|
|
let status_name = folder.clone();
|
|
if let Ok(Some(folder_state)) = status_db
|
|
.run(move |conn| db::folders::get_folder(conn, &status_name))
|
|
.await
|
|
{
|
|
event_bus
|
|
.publish(
|
|
&user_hash,
|
|
MailEvent::FolderStatus {
|
|
folder: folder.clone(),
|
|
total_count: folder_state.total_count,
|
|
unread_count: folder_state.unread_count,
|
|
},
|
|
)
|
|
.await;
|
|
}
|
|
event_bus
|
|
.publish(
|
|
&user_hash,
|
|
MailEvent::SyncStatus {
|
|
active: false,
|
|
folder: Some(folder.clone()),
|
|
pending: 0,
|
|
},
|
|
)
|
|
.await;
|
|
let database = db::pool::UserDb::new(&config.data_dir, &user_hash);
|
|
if let Ok(pending) = database.run(db::folders::pending_message_sync_count).await {
|
|
event_bus
|
|
.publish(&user_hash, MailEvent::SyncInventory { pending })
|
|
.await;
|
|
}
|
|
}
|
|
Err(e) => {
|
|
event_bus
|
|
.publish(
|
|
&user_hash,
|
|
MailEvent::SyncStatus {
|
|
active: false,
|
|
folder: Some(folder.clone()),
|
|
pending: 0,
|
|
},
|
|
)
|
|
.await;
|
|
let inventory_db = db::pool::UserDb::new(&config.data_dir, &user_hash);
|
|
if let Ok(pending) = inventory_db
|
|
.run(db::folders::pending_message_sync_count)
|
|
.await
|
|
{
|
|
event_bus
|
|
.publish(&user_hash, MailEvent::SyncInventory { pending })
|
|
.await;
|
|
}
|
|
tracing::warn!(folder = %folder, error = %e, "Background sync: failed (will retry on next request)");
|
|
let database = db::pool::UserDb::new(&config.data_dir, &user_hash);
|
|
let name = folder.clone();
|
|
let superseded = database
|
|
.run(move |conn| {
|
|
let superseded = HeaderSync::capture(conn)? != header_sync;
|
|
db::folders::invalidate_folder_freshness(conn, &name)?;
|
|
Ok(superseded)
|
|
})
|
|
.await;
|
|
|
|
if matches!(superseded, Ok(true)) {
|
|
event_bus
|
|
.publish(&user_hash, MailEvent::FolderUpdated)
|
|
.await;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
pub async fn get_message(
|
|
Extension(session): Extension<SessionState>,
|
|
Extension(config): Extension<Arc<AppConfig>>,
|
|
Extension(imap_client): Extension<Arc<dyn ImapClient>>,
|
|
Extension(search_engine): Extension<Arc<SearchEngine>>,
|
|
Path((folder, uid)): Path<(String, u32)>,
|
|
) -> Result<Response, AppError> {
|
|
let creds = build_creds(&session, &config)?;
|
|
|
|
let conn = db::pool::open_user_db(&config.data_dir, &session.user_hash)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
|
|
|
let cached_body = db::messages::get_cached_body(&conn, &folder, uid)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
|
|
|
let usable_cache = cached_body.filter(|c| c.attachments_json.is_some());
|
|
|
|
let (body_html, body_text, attachments, raw_headers, email_theme) = if let Some(cached) =
|
|
usable_cache
|
|
{
|
|
let attachments: Vec<AttachmentMeta> = cached
|
|
.attachments_json
|
|
.as_deref()
|
|
.and_then(|j| serde_json::from_str(j).ok())
|
|
.unwrap_or_default();
|
|
let theme = cached.email_theme;
|
|
if theme.is_none()
|
|
&& let Some(ref html) = cached.html
|
|
{
|
|
let detected = email_theme::detect_email_theme(html).map(|t| t.as_i32());
|
|
if let Some(t) = detected {
|
|
let _ = db::messages::update_email_theme(&conn, &folder, uid, t);
|
|
}
|
|
(
|
|
cached.html,
|
|
cached.text,
|
|
attachments,
|
|
cached.raw_headers.unwrap_or_default(),
|
|
detected,
|
|
)
|
|
} else {
|
|
(
|
|
cached.html,
|
|
cached.text,
|
|
attachments,
|
|
cached.raw_headers.unwrap_or_default(),
|
|
theme,
|
|
)
|
|
}
|
|
} else {
|
|
let body = imap_client
|
|
.fetch_body(&creds, &folder, uid)
|
|
.await
|
|
.map_err(|e| match e {
|
|
crate::imap::client::ImapError::MessageNotFound { .. } => {
|
|
if let Ok(transaction) = conn.unchecked_transaction() {
|
|
let _ = db::messages::delete_message(&transaction, &folder, uid);
|
|
let _ = db::folders::invalidate_folder_freshness(&transaction, &folder);
|
|
if let Ok(user_index) = search_engine.open_user_index(&session.user_hash) {
|
|
let _ = user_index.delete_message_with_connection(
|
|
&transaction,
|
|
uid,
|
|
&folder,
|
|
);
|
|
}
|
|
let _ = transaction.commit();
|
|
}
|
|
AppError::NotFound(format!("Message UID {uid} not found in folder {folder}"))
|
|
}
|
|
other => {
|
|
if let Ok(transaction) = conn.unchecked_transaction() {
|
|
let _ = db::messages::delete_message(&transaction, &folder, uid);
|
|
let _ = db::folders::invalidate_folder_freshness(&transaction, &folder);
|
|
if let Ok(user_index) = search_engine.open_user_index(&session.user_hash) {
|
|
let _ = user_index.delete_message_with_connection(
|
|
&transaction,
|
|
uid,
|
|
&folder,
|
|
);
|
|
}
|
|
let _ = transaction.commit();
|
|
}
|
|
AppError::ServiceUnavailable(format!("IMAP error: {other}"))
|
|
}
|
|
})?;
|
|
|
|
let sanitized_html = body.text_html.clone();
|
|
|
|
let attachment_meta: Vec<AttachmentMeta> = body
|
|
.attachments
|
|
.iter()
|
|
.enumerate()
|
|
.map(|(i, a)| AttachmentMeta {
|
|
id: i.to_string(),
|
|
filename: a.filename.clone(),
|
|
content_type: preview_content_type(&a.content_type, a.filename.as_deref(), &a.data),
|
|
size: a.size,
|
|
content_id: a.content_id.clone(),
|
|
})
|
|
.collect();
|
|
|
|
let resolved_html = sanitized_html.map(|mut html| {
|
|
for att in &body.attachments {
|
|
if let Some(ref cid) = att.content_id {
|
|
let cid_url = format!("cid:{cid}");
|
|
if html.contains(&cid_url) {
|
|
use base64::Engine;
|
|
let b64 = base64::engine::general_purpose::STANDARD.encode(&att.data);
|
|
let data_uri = format!("data:{};base64,{}", att.content_type, b64);
|
|
html = html.replace(&cid_url, &data_uri);
|
|
}
|
|
}
|
|
}
|
|
html
|
|
});
|
|
|
|
let att_json = serde_json::to_string(&attachment_meta).ok();
|
|
|
|
let detected_theme = resolved_html
|
|
.as_ref()
|
|
.and_then(|h| email_theme::detect_email_theme(h))
|
|
.map(|t| t.as_i32());
|
|
|
|
let transaction = conn
|
|
.unchecked_transaction()
|
|
.map_err(|e| AppError::InternalError(format!("Database transaction error: {e}")))?;
|
|
db::messages::cache_message_body(
|
|
&transaction,
|
|
&folder,
|
|
uid,
|
|
resolved_html.as_deref(),
|
|
body.text_plain.as_deref(),
|
|
att_json.as_deref(),
|
|
Some(&body.raw_headers),
|
|
detected_theme,
|
|
)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
|
|
|
if let Some(text) = body.text_plain.as_ref()
|
|
&& !UserIndex::is_excluded_folder(&folder)
|
|
&& let Some(msg) = db::messages::get_single_message(&transaction, &folder, uid)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?
|
|
&& let Ok(user_index) = search_engine.open_user_index(&session.user_hash)
|
|
{
|
|
let indexable = IndexableMessage {
|
|
uid: msg.uid,
|
|
folder: msg.folder,
|
|
subject: msg.subject,
|
|
from_address: msg.from_address,
|
|
from_name: msg.from_name,
|
|
to_addresses: msg.to_addresses,
|
|
body_text: text.clone(),
|
|
date_epoch: crate::db::messages::parse_date_to_epoch_public(&msg.date),
|
|
has_attachments: msg.has_attachments,
|
|
};
|
|
user_index
|
|
.index_message_with_connection(&transaction, &indexable)
|
|
.map_err(|e| AppError::InternalError(format!("Search outbox error: {e}")))?;
|
|
}
|
|
transaction
|
|
.commit()
|
|
.map_err(|e| AppError::InternalError(format!("Database commit error: {e}")))?;
|
|
|
|
(
|
|
resolved_html,
|
|
body.text_plain,
|
|
attachment_meta,
|
|
body.raw_headers,
|
|
detected_theme,
|
|
)
|
|
};
|
|
|
|
let msg = db::messages::get_single_message(&conn, &folder, uid)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
|
|
|
let msg = match msg {
|
|
Some(m) => m,
|
|
None => {
|
|
let parsed = mail_parser::MessageParser::default().parse(raw_headers.as_bytes());
|
|
let subject = parsed
|
|
.as_ref()
|
|
.and_then(|p| p.subject().map(|s| s.to_string()))
|
|
.unwrap_or_default();
|
|
let (from_address, from_name) = parsed
|
|
.as_ref()
|
|
.and_then(|p| p.from())
|
|
.and_then(|addr| match addr {
|
|
mail_parser::Address::List(addrs) => addrs.first().map(|a| {
|
|
(
|
|
a.address
|
|
.as_ref()
|
|
.map(|s| s.to_string())
|
|
.unwrap_or_default(),
|
|
a.name.as_ref().map(|s| s.to_string()).unwrap_or_default(),
|
|
)
|
|
}),
|
|
_ => None,
|
|
})
|
|
.unwrap_or_default();
|
|
let date = parsed
|
|
.as_ref()
|
|
.and_then(|p| p.date())
|
|
.map(|d| {
|
|
let sign = if d.tz_before_gmt { '-' } else { '+' };
|
|
format!(
|
|
"{:04}-{:02}-{:02}T{:02}:{:02}:{:02}{}{:02}:{:02}",
|
|
d.year,
|
|
d.month,
|
|
d.day,
|
|
d.hour,
|
|
d.minute,
|
|
d.second,
|
|
sign,
|
|
d.tz_hour,
|
|
d.tz_minute
|
|
)
|
|
})
|
|
.unwrap_or_default();
|
|
|
|
db::messages::CachedMessage {
|
|
uid,
|
|
folder: folder.clone(),
|
|
message_id: parsed
|
|
.as_ref()
|
|
.and_then(|p| p.message_id().map(|s| format!("<{s}>"))),
|
|
in_reply_to: parsed
|
|
.as_ref()
|
|
.and_then(|p| p.in_reply_to().as_text().map(|s| format!("<{s}>"))),
|
|
references_header: parsed.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}>")))
|
|
}),
|
|
subject,
|
|
from_address,
|
|
from_name,
|
|
to_addresses: String::from("[]"),
|
|
cc_addresses: String::from("[]"),
|
|
date,
|
|
flags: String::new(),
|
|
size: 0,
|
|
has_attachments: !attachments.is_empty(),
|
|
snippet: String::new(),
|
|
reaction: None,
|
|
}
|
|
}
|
|
};
|
|
|
|
let thread_messages = if let Some(ref message_id) = msg.message_id {
|
|
db::messages::get_full_thread(&conn, message_id, msg.references_header.as_deref())
|
|
.unwrap_or_default()
|
|
} else {
|
|
vec![]
|
|
};
|
|
|
|
let thread: Vec<ThreadMessage> = thread_messages
|
|
.into_iter()
|
|
.map(|m| ThreadMessage {
|
|
uid: m.uid,
|
|
folder: m.folder,
|
|
message_id: m.message_id,
|
|
in_reply_to: m.in_reply_to,
|
|
subject: m.subject,
|
|
from_address: m.from_address,
|
|
from_name: m.from_name,
|
|
to_addresses: m.to_addresses,
|
|
cc_addresses: m.cc_addresses,
|
|
date: m.date,
|
|
flags: m.flags,
|
|
size: m.size,
|
|
has_attachments: m.has_attachments,
|
|
snippet: m.snippet,
|
|
})
|
|
.collect();
|
|
|
|
Ok(Json(MessageDetailResponse {
|
|
uid: msg.uid,
|
|
folder: msg.folder.clone(),
|
|
message_id: msg.message_id.clone(),
|
|
in_reply_to: msg.in_reply_to.clone(),
|
|
references: msg.references_header.clone(),
|
|
subject: msg.subject,
|
|
from_address: msg.from_address,
|
|
from_name: msg.from_name,
|
|
to_addresses: parse_address_list(&msg.to_addresses),
|
|
cc_addresses: parse_address_list(&msg.cc_addresses),
|
|
date: msg.date,
|
|
flags: parse_flags(&msg.flags),
|
|
has_attachments: msg.has_attachments,
|
|
html: body_html,
|
|
text: body_text,
|
|
raw_headers,
|
|
attachments,
|
|
thread,
|
|
email_theme: email_theme.map(|t| match t {
|
|
0 => EmailTheme::Light,
|
|
1 => EmailTheme::Dark,
|
|
2 => EmailTheme::Transparent,
|
|
3 => EmailTheme::Adaptive,
|
|
_ => EmailTheme::Light,
|
|
}),
|
|
})
|
|
.into_response())
|
|
}
|
|
|
|
pub async fn update_flags(
|
|
Extension(session): Extension<SessionState>,
|
|
Extension(config): Extension<Arc<AppConfig>>,
|
|
Extension(imap_client): Extension<Arc<dyn ImapClient>>,
|
|
Extension(event_bus): Extension<Arc<EventBus>>,
|
|
Path((folder, uid)): Path<(String, u32)>,
|
|
Json(body): Json<UpdateFlagsRequest>,
|
|
) -> Result<Response, AppError> {
|
|
validate_flags(&body.flags)?;
|
|
|
|
let creds = build_creds(&session, &config)?;
|
|
let flag_refs: Vec<&str> = body.flags.iter().map(|s| s.as_str()).collect();
|
|
let seen_changed = body.flags.iter().any(|flag| flag == "\\Seen");
|
|
let counter_snapshot = if seen_changed {
|
|
folder_counts::snapshot(&config, &session.user_hash, &folder)?
|
|
} else {
|
|
None
|
|
};
|
|
|
|
if body.add {
|
|
imap_client
|
|
.add_flags(&creds, &folder, uid, &flag_refs)
|
|
.await
|
|
.map_err(|e| AppError::ServiceUnavailable(format!("IMAP error: {e}")))?;
|
|
} else {
|
|
imap_client
|
|
.remove_flags(&creds, &folder, uid, &flag_refs)
|
|
.await
|
|
.map_err(|e| AppError::ServiceUnavailable(format!("IMAP error: {e}")))?;
|
|
}
|
|
|
|
let authoritative_counts = if seen_changed {
|
|
folder_counts::reconcile_after_mutation(
|
|
&config,
|
|
&session.user_hash,
|
|
imap_client.as_ref(),
|
|
&creds,
|
|
&folder,
|
|
counter_snapshot,
|
|
)
|
|
.await?
|
|
} else {
|
|
None
|
|
};
|
|
|
|
let conn = db::pool::open_user_db(&config.data_dir, &session.user_hash)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
|
|
|
let cached_flags: Option<String> = conn
|
|
.query_row(
|
|
crate::dbq!(routes_messages_mod::Q_001),
|
|
crate::db::pool::params![&folder, uid],
|
|
|row| row.get(0),
|
|
)
|
|
.ok();
|
|
let current_flags_csv = cached_flags.clone().unwrap_or_default();
|
|
let mut current_flags: Vec<String> = if current_flags_csv.is_empty() {
|
|
vec![]
|
|
} else {
|
|
current_flags_csv
|
|
.split(',')
|
|
.filter(|flag| !flag.is_empty())
|
|
.map(|flag| flag.to_string())
|
|
.collect()
|
|
};
|
|
let was_seen = current_flags.iter().any(|flag| flag == "\\Seen");
|
|
|
|
if body.add {
|
|
for flag in &body.flags {
|
|
if !current_flags.contains(flag) {
|
|
current_flags.push(flag.clone());
|
|
}
|
|
}
|
|
} else {
|
|
current_flags.retain(|flag| !body.flags.contains(flag));
|
|
}
|
|
current_flags.sort();
|
|
current_flags.dedup();
|
|
let is_seen = current_flags.iter().any(|flag| flag == "\\Seen");
|
|
|
|
db::messages::update_message_flags(&conn, &folder, uid, ¤t_flags.join(","))
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
|
|
|
if authoritative_counts.is_none()
|
|
&& seen_changed
|
|
&& cached_flags.is_some()
|
|
&& was_seen != is_seen
|
|
&& let Some((total_count, unread_count, _generation)) = counter_snapshot
|
|
{
|
|
let next_unread = if is_seen {
|
|
unread_count.saturating_sub(1)
|
|
} else {
|
|
unread_count.saturating_add(1)
|
|
};
|
|
let _ = folder_counts::set_if_snapshot_matches(
|
|
&config,
|
|
&session.user_hash,
|
|
&folder,
|
|
counter_snapshot,
|
|
total_count,
|
|
next_unread,
|
|
)?;
|
|
}
|
|
|
|
let folder_state = db::folders::get_folder(&conn, &folder)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
|
let total_count = folder_state
|
|
.as_ref()
|
|
.map(|state| state.total_count)
|
|
.unwrap_or(0);
|
|
let unread_count = folder_state
|
|
.as_ref()
|
|
.map(|state| state.unread_count)
|
|
.unwrap_or(0);
|
|
|
|
event_bus
|
|
.publish(
|
|
&session.user_hash,
|
|
MailEvent::FlagsChanged {
|
|
folder: folder.clone(),
|
|
changes: vec![MessageFlagChange {
|
|
uid,
|
|
flags: current_flags.clone(),
|
|
}],
|
|
},
|
|
)
|
|
.await;
|
|
event_bus
|
|
.publish(
|
|
&session.user_hash,
|
|
MailEvent::FolderStatus {
|
|
folder: folder.clone(),
|
|
total_count,
|
|
unread_count,
|
|
},
|
|
)
|
|
.await;
|
|
|
|
Ok(Json(serde_json::json!({
|
|
"status": "ok",
|
|
"flags": current_flags,
|
|
"total_count": total_count,
|
|
"unread_count": unread_count,
|
|
}))
|
|
.into_response())
|
|
}
|
|
|
|
pub async fn move_message_handler(
|
|
Extension(session): Extension<SessionState>,
|
|
Extension(config): Extension<Arc<AppConfig>>,
|
|
Extension(imap_client): Extension<Arc<dyn ImapClient>>,
|
|
Extension(search_engine): Extension<Arc<SearchEngine>>,
|
|
Extension(event_bus): Extension<Arc<EventBus>>,
|
|
Json(body): Json<MoveMessageRequest>,
|
|
) -> Result<Response, AppError> {
|
|
if body.from_folder == body.to_folder {
|
|
return Err(AppError::BadRequest(
|
|
"Source and destination folders must differ".to_string(),
|
|
));
|
|
}
|
|
|
|
let creds = build_creds(&session, &config)?;
|
|
let source_counter_snapshot =
|
|
folder_counts::snapshot(&config, &session.user_hash, &body.from_folder)?;
|
|
let destination_counter_snapshot =
|
|
folder_counts::snapshot(&config, &session.user_hash, &body.to_folder)?;
|
|
|
|
let source_snapshot = {
|
|
let conn = db::pool::open_user_db(&config.data_dir, &session.user_hash)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
|
db::messages::get_single_message(&conn, &body.from_folder, body.uid)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?
|
|
};
|
|
let destination_uid_floor = if source_snapshot
|
|
.as_ref()
|
|
.and_then(|message| message.message_id.as_deref())
|
|
.is_some_and(|message_id| !message_id.trim().is_empty())
|
|
{
|
|
imap_client
|
|
.folder_status(&creds, &body.to_folder)
|
|
.await
|
|
.ok()
|
|
.map(|status| status.uid_next.max(1))
|
|
} else {
|
|
None
|
|
};
|
|
|
|
imap_client
|
|
.move_message(&creds, &body.from_folder, body.uid, &body.to_folder)
|
|
.await
|
|
.map_err(|e| AppError::ServiceUnavailable(format!("IMAP error: {e}")))?;
|
|
|
|
let (source_counts, destination_counts) = tokio::join!(
|
|
folder_counts::reconcile_after_mutation(
|
|
&config,
|
|
&session.user_hash,
|
|
imap_client.as_ref(),
|
|
&creds,
|
|
&body.from_folder,
|
|
source_counter_snapshot,
|
|
),
|
|
folder_counts::reconcile_after_mutation(
|
|
&config,
|
|
&session.user_hash,
|
|
imap_client.as_ref(),
|
|
&creds,
|
|
&body.to_folder,
|
|
destination_counter_snapshot,
|
|
),
|
|
);
|
|
let source_counts = source_counts?;
|
|
let destination_counts = destination_counts?;
|
|
|
|
let undo_destination_uid = if let (Some(uid_floor), Some(message_id)) = (
|
|
destination_uid_floor,
|
|
source_snapshot
|
|
.as_ref()
|
|
.and_then(|message| message.message_id.as_deref())
|
|
.map(str::trim)
|
|
.filter(|value| !value.is_empty()),
|
|
) {
|
|
let uid_range = format!("{uid_floor}:*");
|
|
match imap_client
|
|
.fetch_headers(&creds, &body.to_folder, &uid_range)
|
|
.await
|
|
{
|
|
Ok(headers) => {
|
|
let mut matches = headers
|
|
.into_iter()
|
|
.filter(|header| header.uid >= uid_floor)
|
|
.filter(|header| {
|
|
header.message_id.as_deref().map(str::trim) == Some(message_id)
|
|
})
|
|
.map(|header| header.uid);
|
|
let first = matches.next();
|
|
if first.is_some() && matches.next().is_none() {
|
|
first
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
Err(error) => {
|
|
tracing::warn!(?error, folder = %body.to_folder, uid_floor, "Moved message but could not resolve destination UID for Undo");
|
|
None
|
|
}
|
|
}
|
|
} else {
|
|
None
|
|
};
|
|
|
|
let conn = db::pool::open_user_db(&config.data_dir, &session.user_hash)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
|
|
|
let transaction = conn
|
|
.unchecked_transaction()
|
|
.map_err(|e| AppError::InternalError(format!("Database transaction error: {e}")))?;
|
|
db::messages::remove_after_mutation_in_transaction(&transaction, &body.from_folder, body.uid)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
|
|
|
if let Ok(index) = search_engine.open_user_index(&session.user_hash) {
|
|
index
|
|
.delete_message_with_connection(&transaction, body.uid, &body.from_folder)
|
|
.map_err(|e| AppError::InternalError(format!("Search outbox error: {e}")))?;
|
|
}
|
|
|
|
db::folders::invalidate_folder_freshness(&transaction, &body.to_folder)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
|
transaction
|
|
.commit()
|
|
.map_err(|e| AppError::InternalError(format!("Database commit error: {e}")))?;
|
|
|
|
let source_counts =
|
|
folder_counts::current_counts(&config, &session.user_hash, &body.from_folder)?
|
|
.or(source_counts);
|
|
let destination_counts =
|
|
folder_counts::current_counts(&config, &session.user_hash, &body.to_folder)?
|
|
.or(destination_counts);
|
|
|
|
for (folder_name, counts) in [
|
|
(&body.from_folder, source_counts),
|
|
(&body.to_folder, destination_counts),
|
|
] {
|
|
if let Some((total_count, unread_count)) = counts {
|
|
event_bus
|
|
.publish(
|
|
&session.user_hash,
|
|
MailEvent::FolderStatus {
|
|
folder: folder_name.to_string(),
|
|
total_count,
|
|
unread_count,
|
|
},
|
|
)
|
|
.await;
|
|
}
|
|
}
|
|
|
|
let undo_move = undo_destination_uid.map(|destination_uid| {
|
|
serde_json::json!({
|
|
"from_folder": body.to_folder,
|
|
"to_folder": body.from_folder,
|
|
"uid": destination_uid,
|
|
})
|
|
});
|
|
|
|
Ok(Json(serde_json::json!({ "status": "ok", "undo_move": undo_move })).into_response())
|
|
}
|
|
|
|
pub async fn copy_message_handler(
|
|
Extension(session): Extension<SessionState>,
|
|
Extension(config): Extension<Arc<AppConfig>>,
|
|
Extension(imap_client): Extension<Arc<dyn ImapClient>>,
|
|
Extension(event_bus): Extension<Arc<EventBus>>,
|
|
Json(body): Json<CopyMessageRequest>,
|
|
) -> Result<Response, AppError> {
|
|
if body.from_folder == body.to_folder {
|
|
return Err(AppError::BadRequest(
|
|
"Source and destination folders must differ".to_string(),
|
|
));
|
|
}
|
|
let creds = build_creds(&session, &config)?;
|
|
let destination_counter_snapshot =
|
|
folder_counts::snapshot(&config, &session.user_hash, &body.to_folder)?;
|
|
imap_client
|
|
.copy_message(&creds, &body.from_folder, body.uid, &body.to_folder)
|
|
.await
|
|
.map_err(|e| AppError::ServiceUnavailable(format!("IMAP error: {e}")))?;
|
|
|
|
let destination_counts = folder_counts::reconcile_after_mutation(
|
|
&config,
|
|
&session.user_hash,
|
|
imap_client.as_ref(),
|
|
&creds,
|
|
&body.to_folder,
|
|
destination_counter_snapshot,
|
|
)
|
|
.await?;
|
|
|
|
let conn = db::pool::open_user_db(&config.data_dir, &session.user_hash)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
|
db::folders::invalidate_folder_freshness(&conn, &body.to_folder)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
|
let destination_counts =
|
|
folder_counts::current_counts(&config, &session.user_hash, &body.to_folder)?
|
|
.or(destination_counts);
|
|
|
|
if let Some((total_count, unread_count)) = destination_counts {
|
|
event_bus
|
|
.publish(
|
|
&session.user_hash,
|
|
MailEvent::FolderStatus {
|
|
folder: body.to_folder.clone(),
|
|
total_count,
|
|
unread_count,
|
|
},
|
|
)
|
|
.await;
|
|
}
|
|
|
|
Ok(Json(serde_json::json!({ "status": "ok" })).into_response())
|
|
}
|
|
|
|
pub async fn download_attachment(
|
|
Extension(session): Extension<SessionState>,
|
|
Extension(config): Extension<Arc<AppConfig>>,
|
|
Extension(imap_client): Extension<Arc<dyn ImapClient>>,
|
|
Path((folder, uid, attachment_id)): Path<(String, u32, String)>,
|
|
) -> Result<Response, AppError> {
|
|
let creds = build_creds(&session, &config)?;
|
|
|
|
let index: usize = attachment_id
|
|
.parse()
|
|
.map_err(|_| AppError::BadRequest(format!("Invalid attachment id: {attachment_id}")))?;
|
|
|
|
let body = imap_client
|
|
.fetch_body(&creds, &folder, uid)
|
|
.await
|
|
.map_err(|e| match e {
|
|
crate::imap::client::ImapError::MessageNotFound { .. } => {
|
|
AppError::NotFound(format!("Message UID {uid} not found in folder {folder}"))
|
|
}
|
|
other => AppError::ServiceUnavailable(format!("IMAP error: {other}")),
|
|
})?;
|
|
|
|
let attachment = body.attachments.into_iter().nth(index).ok_or_else(|| {
|
|
AppError::NotFound(format!(
|
|
"Attachment {attachment_id} not found on message UID {uid}"
|
|
))
|
|
})?;
|
|
|
|
let raw_filename = attachment
|
|
.filename
|
|
.unwrap_or_else(|| format!("attachment_{index}"));
|
|
|
|
let filename: String = raw_filename
|
|
.chars()
|
|
.filter(|c| !c.is_ascii_control())
|
|
.collect();
|
|
|
|
let filename = if filename.is_empty() {
|
|
format!("attachment_{index}")
|
|
} else {
|
|
filename
|
|
};
|
|
let content_type =
|
|
preview_content_type(&attachment.content_type, Some(&filename), &attachment.data);
|
|
|
|
let is_inline = content_type == "application/pdf"
|
|
|| content_type.starts_with("image/")
|
|
|| content_type.starts_with("text/");
|
|
let disposition = if is_inline {
|
|
format!("inline; filename=\"{}\"", filename.replace('"', "\\\""))
|
|
} else {
|
|
format!("attachment; filename=\"{}\"", filename.replace('"', "\\\""))
|
|
};
|
|
|
|
Response::builder()
|
|
.header("content-type", &content_type)
|
|
.header("content-disposition", &disposition)
|
|
.body(axum::body::Body::from(attachment.data))
|
|
.map_err(|e| AppError::InternalError(format!("Failed to build response: {e}")))
|
|
}
|
|
|
|
fn preview_content_type(content_type: &str, filename: Option<&str>, data: &[u8]) -> String {
|
|
let media_type = content_type.split(';').next().unwrap_or("").trim();
|
|
if media_type.eq_ignore_ascii_case("application/pdf")
|
|
|| (media_type.eq_ignore_ascii_case("application/octet-stream")
|
|
&& filename.is_some_and(|name| name.to_ascii_lowercase().ends_with(".pdf"))
|
|
&& data.starts_with(b"%PDF-"))
|
|
{
|
|
"application/pdf".to_string()
|
|
} else {
|
|
content_type.to_string()
|
|
}
|
|
}
|
|
|
|
const MAX_EML_IMPORT_SIZE: usize = 50 * 1024 * 1024;
|
|
|
|
pub async fn import_eml(
|
|
Extension(session): Extension<SessionState>,
|
|
Extension(config): Extension<Arc<AppConfig>>,
|
|
Extension(imap_client): Extension<Arc<dyn ImapClient>>,
|
|
Extension(event_bus): Extension<Arc<EventBus>>,
|
|
Path(folder): Path<String>,
|
|
mut multipart: Multipart,
|
|
) -> Result<Response, AppError> {
|
|
let field = multipart
|
|
.next_field()
|
|
.await
|
|
.map_err(|e| AppError::BadRequest(format!("Invalid multipart data: {e}")))?
|
|
.ok_or_else(|| AppError::BadRequest("No .eml file uploaded".to_string()))?;
|
|
|
|
let filename = field.file_name().unwrap_or("message.eml").to_string();
|
|
if !filename.to_ascii_lowercase().ends_with(".eml") {
|
|
return Err(AppError::BadRequest(
|
|
"Uploaded file must use the .eml extension".to_string(),
|
|
));
|
|
}
|
|
|
|
let bytes = field
|
|
.bytes()
|
|
.await
|
|
.map_err(|e| AppError::BadRequest(format!("Failed to read uploaded message: {e}")))?;
|
|
|
|
if bytes.is_empty() {
|
|
return Err(AppError::BadRequest(
|
|
"Uploaded .eml file is empty".to_string(),
|
|
));
|
|
}
|
|
if bytes.len() > MAX_EML_IMPORT_SIZE {
|
|
return Err(AppError::BadRequest(
|
|
".eml file exceeds the 50 MB import limit".to_string(),
|
|
));
|
|
}
|
|
|
|
if mail_parser::MessageParser::default()
|
|
.parse(bytes.as_ref())
|
|
.is_none()
|
|
{
|
|
return Err(AppError::BadRequest(
|
|
"Uploaded file is not a parseable RFC822 message".to_string(),
|
|
));
|
|
}
|
|
|
|
let creds = build_creds(&session, &config)?;
|
|
imap_client
|
|
.append_message(&creds, &folder, &bytes, &[])
|
|
.await
|
|
.map_err(|e| AppError::ServiceUnavailable(format!("IMAP error: {e}")))?;
|
|
|
|
let conn = db::pool::open_user_db(&config.data_dir, &session.user_hash)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
|
db::folders::invalidate_folder_freshness(&conn, &folder)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
|
|
|
event_bus
|
|
.publish(&session.user_hash, MailEvent::FolderUpdated)
|
|
.await;
|
|
|
|
Ok(Json(serde_json::json!({
|
|
"status": "imported",
|
|
"folder": folder,
|
|
"filename": filename,
|
|
"size": bytes.len(),
|
|
}))
|
|
.into_response())
|
|
}
|
|
|
|
pub async fn download_eml_source(
|
|
Extension(session): Extension<SessionState>,
|
|
Extension(config): Extension<Arc<AppConfig>>,
|
|
Extension(imap_client): Extension<Arc<dyn ImapClient>>,
|
|
Path((folder, uid)): Path<(String, u32)>,
|
|
) -> Result<Response, AppError> {
|
|
let creds = build_creds(&session, &config)?;
|
|
let raw = imap_client
|
|
.fetch_raw_message(&creds, &folder, uid)
|
|
.await
|
|
.map_err(|e| AppError::ServiceUnavailable(format!("IMAP error: {e}")))?;
|
|
let filename = format!("message-{uid}.eml");
|
|
Response::builder()
|
|
.header("content-type", "message/rfc822")
|
|
.header(
|
|
"content-disposition",
|
|
format!("attachment; filename=\"{filename}\""),
|
|
)
|
|
.header("cache-control", "private, no-store")
|
|
.body(axum::body::Body::from(raw))
|
|
.map_err(|e| AppError::InternalError(format!("Failed to build response: {e}")))
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod preview_tests {
|
|
use super::{is_valid_flag, preview_content_type, validate_flags};
|
|
|
|
#[test]
|
|
fn accepts_common_system_flags_and_keywords() {
|
|
for flag in [
|
|
"\\Seen",
|
|
"\\Answered",
|
|
"\\Flagged",
|
|
"\\Deleted",
|
|
"\\Draft",
|
|
"Junk",
|
|
"$Forwarded",
|
|
"$MDNSent",
|
|
"custom_tag-1.0+",
|
|
] {
|
|
assert!(is_valid_flag(flag), "rejected {flag:?}");
|
|
}
|
|
assert!(validate_flags(&[]).is_ok());
|
|
assert!(is_valid_flag(&"a".repeat(64)));
|
|
}
|
|
|
|
#[test]
|
|
fn rejects_unsafe_flags_and_enforces_length_limit() {
|
|
for flag in [
|
|
"",
|
|
"\\",
|
|
"two flags",
|
|
"tag\r\nSTORE",
|
|
"tag\0",
|
|
"tag\t",
|
|
"(tag)",
|
|
"tag{1}",
|
|
"tag\\Seen",
|
|
"étiquette",
|
|
] {
|
|
assert!(!is_valid_flag(flag), "accepted {flag:?}");
|
|
}
|
|
assert!(!is_valid_flag(&"a".repeat(65)));
|
|
assert!(validate_flags(&["\\Seen".to_string(), "two flags".to_string()]).is_err());
|
|
}
|
|
|
|
#[test]
|
|
fn normalizes_pdf_media_types() {
|
|
assert_eq!(
|
|
preview_content_type("Application/PDF; name=test.pdf", None, b""),
|
|
"application/pdf"
|
|
);
|
|
assert_eq!(
|
|
preview_content_type("application/octet-stream", Some("TEST.PDF"), b"%PDF-1.7"),
|
|
"application/pdf"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn does_not_trust_filename_alone() {
|
|
assert_eq!(
|
|
preview_content_type("application/octet-stream", Some("test.pdf"), b"not a pdf"),
|
|
"application/octet-stream"
|
|
);
|
|
}
|
|
}
|
|
|
|
pub async fn delete_message_handler(
|
|
Extension(session): Extension<SessionState>,
|
|
Extension(config): Extension<Arc<AppConfig>>,
|
|
Extension(imap_client): Extension<Arc<dyn ImapClient>>,
|
|
Extension(event_bus): Extension<Arc<EventBus>>,
|
|
Path((folder, uid)): Path<(String, u32)>,
|
|
) -> Result<Response, AppError> {
|
|
let creds = build_creds(&session, &config)?;
|
|
let counter_snapshot = folder_counts::snapshot(&config, &session.user_hash, &folder)?;
|
|
|
|
imap_client
|
|
.expunge_message(&creds, &folder, uid)
|
|
.await
|
|
.map_err(|e| AppError::ServiceUnavailable(format!("IMAP error: {e}")))?;
|
|
|
|
let authoritative_counts = folder_counts::reconcile_after_mutation(
|
|
&config,
|
|
&session.user_hash,
|
|
imap_client.as_ref(),
|
|
&creds,
|
|
&folder,
|
|
counter_snapshot,
|
|
)
|
|
.await?;
|
|
|
|
let conn = db::pool::open_user_db(&config.data_dir, &session.user_hash)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
|
db::messages::remove_after_mutation(&conn, &folder, uid)
|
|
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
|
let authoritative_counts = folder_counts::current_counts(&config, &session.user_hash, &folder)?
|
|
.or(authoritative_counts);
|
|
|
|
if let Some((total_count, unread_count)) = authoritative_counts {
|
|
event_bus
|
|
.publish(
|
|
&session.user_hash,
|
|
MailEvent::FolderStatus {
|
|
folder: folder.clone(),
|
|
total_count,
|
|
unread_count,
|
|
},
|
|
)
|
|
.await;
|
|
}
|
|
|
|
Ok(Json(serde_json::json!({ "status": "ok" })).into_response())
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod sync_tests {
|
|
use super::*;
|
|
use crate::imap::client::mock::MockImapClient;
|
|
|
|
fn drain_events(events: &mut tokio::sync::broadcast::Receiver<MailEvent>) -> Vec<MailEvent> {
|
|
let mut drained = Vec::new();
|
|
while let Ok(event) = events.try_recv() {
|
|
drained.push(event);
|
|
}
|
|
drained
|
|
}
|
|
|
|
async fn wait_for_sync_completion(
|
|
events: &mut tokio::sync::broadcast::Receiver<MailEvent>,
|
|
folder: &str,
|
|
) {
|
|
tokio::time::timeout(std::time::Duration::from_secs(5), async {
|
|
loop {
|
|
match events.recv().await.unwrap() {
|
|
MailEvent::SyncStatus {
|
|
active: false,
|
|
folder: Some(event_folder),
|
|
..
|
|
} if event_folder == folder => break,
|
|
_ => {}
|
|
}
|
|
}
|
|
})
|
|
.await
|
|
.expect("background header sync did not complete in time");
|
|
}
|
|
|
|
fn sync_header(uid: u32) -> crate::imap::types::ImapMessageHeader {
|
|
crate::imap::types::ImapMessageHeader {
|
|
uid,
|
|
subject: Some(format!("Message {uid}")),
|
|
from: Vec::new(),
|
|
to: Vec::new(),
|
|
cc: Vec::new(),
|
|
date: None,
|
|
flags: Vec::new(),
|
|
has_attachments: false,
|
|
size: 42,
|
|
message_id: None,
|
|
in_reply_to: None,
|
|
references: None,
|
|
reaction: None,
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn fetched_background_headers_cannot_resurrect_completed_move() {
|
|
fetched_headers_after_removal(true).await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn fetched_background_headers_cannot_resurrect_completed_delete() {
|
|
fetched_headers_after_removal(false).await;
|
|
}
|
|
|
|
async fn fetched_headers_after_removal(move_message: bool) {
|
|
let data_dir = tempfile::tempdir().unwrap();
|
|
let path = data_dir.path().to_str().unwrap();
|
|
let user_hash = "header-race";
|
|
let conn = db::pool::open_user_db(path, user_hash).unwrap();
|
|
for folder in ["INBOX", "Archive"] {
|
|
db::folders::upsert_folder(&conn, folder, None, None, "", true, 1, 1, 1, 0).unwrap();
|
|
}
|
|
let search_engine = Arc::new(SearchEngine::new(data_dir.path().to_path_buf()));
|
|
let header_sync = HeaderSync::capture(&conn).unwrap();
|
|
let transaction = header_sync.begin_write(&conn).unwrap();
|
|
upsert_and_index_headers(
|
|
&transaction,
|
|
"INBOX",
|
|
&[sync_header(7)],
|
|
&search_engine,
|
|
user_hash,
|
|
)
|
|
.unwrap();
|
|
transaction.commit().unwrap();
|
|
drop(conn);
|
|
|
|
let creds = ImapCredentials {
|
|
host: "localhost".into(),
|
|
port: 143,
|
|
tls: false,
|
|
tls_verify: false,
|
|
email: "test@example.com".into(),
|
|
password: "test".into(),
|
|
};
|
|
let config = Arc::new(AppConfig {
|
|
imap_host: Some(creds.host.clone()),
|
|
data_dir: path.to_owned(),
|
|
..serde_json::from_str::<AppConfig>("{}").unwrap()
|
|
});
|
|
let session = SessionState {
|
|
account_id: "test".into(),
|
|
email: creds.email.clone(),
|
|
password: creds.password.clone(),
|
|
user_hash: user_hash.into(),
|
|
imap_host: creds.host.clone(),
|
|
imap_port: creds.port,
|
|
imap_tls: false,
|
|
imap_tls_verify: false,
|
|
smtp_host: "localhost".into(),
|
|
smtp_port: 587,
|
|
smtp_tls: false,
|
|
smtp_tls_verify: false,
|
|
last_accessed: std::time::Instant::now(),
|
|
timeout_override: None,
|
|
created_at_unix: 0,
|
|
last_accessed_unix: 0,
|
|
user_agent: String::new(),
|
|
};
|
|
let mut client = MockImapClient::new().with_headers(vec![sync_header(7), sync_header(8)]);
|
|
client.header_fetch_gate = Some((tokio::sync::Notify::new(), tokio::sync::Notify::new()));
|
|
let client = Arc::new(client);
|
|
let gate = client.header_fetch_gate.as_ref().unwrap();
|
|
let event_bus = Arc::new(EventBus::new());
|
|
let mut events = event_bus.subscribe(user_hash).await;
|
|
let sync = tokio::spawn(sync_remaining_headers(BackgroundSyncParams {
|
|
creds,
|
|
folder: "INBOX".into(),
|
|
imap_client: client.clone(),
|
|
config: config.clone(),
|
|
user_hash: user_hash.into(),
|
|
search_engine: search_engine.clone(),
|
|
event_bus: event_bus.clone(),
|
|
uid_range: (1, 100),
|
|
header_sync,
|
|
uid_validity: 1,
|
|
highest_modseq: 0,
|
|
}));
|
|
|
|
gate.0.notified().await;
|
|
let imap_client: Arc<dyn ImapClient> = client.clone();
|
|
let response = if move_message {
|
|
move_message_handler(
|
|
Extension(session),
|
|
Extension(config),
|
|
Extension(imap_client),
|
|
Extension(search_engine.clone()),
|
|
Extension(event_bus.clone()),
|
|
Json(MoveMessageRequest {
|
|
uid: 7,
|
|
from_folder: "INBOX".into(),
|
|
to_folder: "Archive".into(),
|
|
}),
|
|
)
|
|
.await
|
|
.unwrap()
|
|
} else {
|
|
delete_message_handler(
|
|
Extension(session),
|
|
Extension(config),
|
|
Extension(imap_client),
|
|
Extension(event_bus.clone()),
|
|
Path(("INBOX".into(), 7)),
|
|
)
|
|
.await
|
|
.unwrap()
|
|
};
|
|
assert_eq!(response.status(), axum::http::StatusCode::OK);
|
|
gate.1.notify_one();
|
|
sync.await.unwrap();
|
|
|
|
let conn = db::pool::open_user_db(path, user_hash).unwrap();
|
|
assert!(
|
|
db::messages::get_single_message(&conn, "INBOX", 7)
|
|
.unwrap()
|
|
.is_none()
|
|
);
|
|
assert!(db::folders::is_folder_messages_invalidated(&conn, "INBOX").unwrap());
|
|
let emitted = drain_events(&mut events);
|
|
assert!(
|
|
emitted
|
|
.iter()
|
|
.any(|event| matches!(event, MailEvent::FolderUpdated)),
|
|
"superseded sync must request fresh missing-UID reconciliation"
|
|
);
|
|
|
|
if move_message {
|
|
let index = search_engine.open_user_index(user_hash).unwrap();
|
|
assert!(
|
|
index
|
|
.search(&crate::search::engine::SearchQuery {
|
|
text: "Message".into(),
|
|
limit: 10,
|
|
..Default::default()
|
|
})
|
|
.await
|
|
.unwrap()
|
|
.0
|
|
.is_empty()
|
|
);
|
|
}
|
|
|
|
let next_sync = HeaderSync::capture(&conn).unwrap();
|
|
let transaction = next_sync.begin_write(&conn).unwrap();
|
|
upsert_and_index_headers(
|
|
&transaction,
|
|
"Archive",
|
|
&[sync_header(7), sync_header(42)],
|
|
&search_engine,
|
|
user_hash,
|
|
)
|
|
.unwrap();
|
|
upsert_and_index_headers(
|
|
&transaction,
|
|
"INBOX",
|
|
&[sync_header(8)],
|
|
&search_engine,
|
|
user_hash,
|
|
)
|
|
.unwrap();
|
|
transaction.commit().unwrap();
|
|
assert_eq!(db::messages::count_messages(&conn, "Archive").unwrap(), 2);
|
|
assert!(
|
|
db::messages::get_single_message(&conn, "INBOX", 8)
|
|
.unwrap()
|
|
.is_some()
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn interrupted_initial_backfill_recovers_after_same_folder_delete() {
|
|
interrupted_backfill_recovers(false).await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn interrupted_initial_backfill_recovers_after_unrelated_folder_delete() {
|
|
interrupted_backfill_recovers(true).await;
|
|
}
|
|
|
|
async fn interrupted_backfill_recovers(other_folder: bool) {
|
|
let data_dir = tempfile::tempdir().unwrap();
|
|
let path = data_dir.path().to_str().unwrap();
|
|
let user_hash = "backfill-recovery";
|
|
let conn = db::pool::open_user_db(path, user_hash).unwrap();
|
|
for folder in ["INBOX", "Archive"] {
|
|
db::folders::upsert_folder(&conn, folder, None, None, "", true, 1000, 0, 1, 0).unwrap();
|
|
}
|
|
let engine = Arc::new(SearchEngine::new(data_dir.path().to_path_buf()));
|
|
let header_sync = HeaderSync::capture(&conn).unwrap();
|
|
let transaction = header_sync.begin_write(&conn).unwrap();
|
|
upsert_and_index_headers(
|
|
&transaction,
|
|
"INBOX",
|
|
&(901..=1000).map(sync_header).collect::<Vec<_>>(),
|
|
&engine,
|
|
user_hash,
|
|
)
|
|
.unwrap();
|
|
upsert_and_index_headers(
|
|
&transaction,
|
|
"Archive",
|
|
&[sync_header(950)],
|
|
&engine,
|
|
user_hash,
|
|
)
|
|
.unwrap();
|
|
db::folders::update_folder_status(&transaction, "INBOX", 1, 1000).unwrap();
|
|
transaction.commit().unwrap();
|
|
drop(conn);
|
|
let creds = ImapCredentials {
|
|
host: "localhost".into(),
|
|
port: 143,
|
|
tls: false,
|
|
tls_verify: false,
|
|
email: "test@example.com".into(),
|
|
password: "test".into(),
|
|
};
|
|
let config = Arc::new(AppConfig {
|
|
imap_host: Some(creds.host.clone()),
|
|
data_dir: path.to_owned(),
|
|
..serde_json::from_str::<AppConfig>("{}").unwrap()
|
|
});
|
|
let session = SessionState {
|
|
account_id: "test".into(),
|
|
email: creds.email.clone(),
|
|
password: creds.password.clone(),
|
|
user_hash: user_hash.into(),
|
|
imap_host: creds.host.clone(),
|
|
imap_port: 143,
|
|
imap_tls: false,
|
|
imap_tls_verify: false,
|
|
smtp_host: "localhost".into(),
|
|
smtp_port: 587,
|
|
smtp_tls: false,
|
|
smtp_tls_verify: false,
|
|
last_accessed: std::time::Instant::now(),
|
|
timeout_override: None,
|
|
created_at_unix: 0,
|
|
last_accessed_unix: 0,
|
|
user_agent: String::new(),
|
|
};
|
|
let mut client = MockImapClient::new().with_headers((1..=900).map(sync_header).collect());
|
|
client.header_fetch_gate = Some((tokio::sync::Notify::new(), tokio::sync::Notify::new()));
|
|
let client = Arc::new(client);
|
|
let gate = client.header_fetch_gate.as_ref().unwrap();
|
|
let bus = Arc::new(EventBus::new());
|
|
let mut events = bus.subscribe(user_hash).await;
|
|
let sync = tokio::spawn(sync_remaining_headers(BackgroundSyncParams {
|
|
creds,
|
|
folder: "INBOX".into(),
|
|
imap_client: client.clone(),
|
|
config: config.clone(),
|
|
user_hash: user_hash.into(),
|
|
search_engine: engine.clone(),
|
|
event_bus: bus.clone(),
|
|
uid_range: (1, 900),
|
|
header_sync,
|
|
uid_validity: 1,
|
|
highest_modseq: 0,
|
|
}));
|
|
gate.0.notified().await;
|
|
let folder = if other_folder { "Archive" } else { "INBOX" };
|
|
delete_message_handler(
|
|
Extension(session.clone()),
|
|
Extension(config.clone()),
|
|
Extension(client.clone() as Arc<dyn ImapClient>),
|
|
Extension(bus.clone()),
|
|
Path((folder.into(), 950)),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
gate.1.notify_one();
|
|
sync.await.unwrap();
|
|
let conn = db::pool::open_user_db(path, user_hash).unwrap();
|
|
assert!(db::folders::is_folder_messages_invalidated(&conn, "INBOX").unwrap());
|
|
assert!(
|
|
db::messages::get_single_message(&conn, "INBOX", 1)
|
|
.unwrap()
|
|
.is_none()
|
|
);
|
|
drop(conn);
|
|
let emitted = drain_events(&mut events);
|
|
assert!(
|
|
emitted
|
|
.iter()
|
|
.any(|event| matches!(event, MailEvent::FolderUpdated)),
|
|
"superseded backfill must request a fresh reconciliation"
|
|
);
|
|
|
|
let remaining = (1..=1000)
|
|
.filter(|uid| other_folder || *uid != 950)
|
|
.map(sync_header)
|
|
.collect::<Vec<_>>();
|
|
let fresh_client = Arc::new(MockImapClient::new().with_headers(remaining));
|
|
let response = list_messages(
|
|
Extension(session),
|
|
Extension(config),
|
|
Extension(fresh_client.clone() as Arc<dyn ImapClient>),
|
|
Extension(engine.clone()),
|
|
Extension(bus),
|
|
Path("INBOX".into()),
|
|
Query(ListMessagesQuery {
|
|
page: 0,
|
|
per_page: 50,
|
|
sort: "date_desc".to_string(),
|
|
refresh: false,
|
|
}),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(response.status(), axum::http::StatusCode::OK);
|
|
wait_for_sync_completion(&mut events, "INBOX").await;
|
|
let conn = db::pool::open_user_db(path, user_hash).unwrap();
|
|
assert_eq!(
|
|
db::messages::count_messages(&conn, "INBOX").unwrap(),
|
|
if other_folder { 1000 } else { 999 }
|
|
);
|
|
assert!(
|
|
db::messages::get_single_message(&conn, "INBOX", 1)
|
|
.unwrap()
|
|
.is_some()
|
|
);
|
|
assert!(
|
|
db::messages::get_single_message(&conn, "INBOX", 900)
|
|
.unwrap()
|
|
.is_some()
|
|
);
|
|
assert!(
|
|
db::messages::get_single_message(&conn, folder, 950)
|
|
.unwrap()
|
|
.is_none()
|
|
);
|
|
assert!(db::folders::is_folder_fresh(&conn, "INBOX", 30).unwrap());
|
|
{
|
|
let requests = fresh_client.header_requests.lock().unwrap();
|
|
assert_eq!(requests.len(), 4);
|
|
assert!(requests.iter().all(|(_, uids)| {
|
|
uids.split(',')
|
|
.all(|uid| uid.parse::<u32>().unwrap() <= 900)
|
|
}));
|
|
}
|
|
let results = engine
|
|
.open_user_index(user_hash)
|
|
.unwrap()
|
|
.search(&crate::search::engine::SearchQuery {
|
|
text: "Message".into(),
|
|
folder: Some("INBOX".into()),
|
|
limit: 1100,
|
|
..Default::default()
|
|
})
|
|
.await
|
|
.unwrap()
|
|
.0;
|
|
assert!(results.iter().any(|message| message.uid == 1));
|
|
}
|
|
|
|
#[test]
|
|
fn uidvalidity_reset_rejects_old_batch_but_allows_reused_uid() {
|
|
let data_dir = tempfile::tempdir().unwrap();
|
|
let conn = db::pool::open_test_db();
|
|
db::folders::upsert_folder(&conn, "INBOX", None, None, "", true, 1, 1, 1, 0).unwrap();
|
|
let engine = Arc::new(SearchEngine::new(data_dir.path().to_path_buf()));
|
|
let stale = HeaderSync::capture(&conn).unwrap();
|
|
db::messages::remove_after_mutation(&conn, "INBOX", 7).unwrap();
|
|
let before_reset = HeaderSync::capture(&conn).unwrap();
|
|
let transaction = before_reset.begin_write(&conn).unwrap();
|
|
db::messages::delete_folder_messages(&transaction, "INBOX").unwrap();
|
|
db::header_sync::invalidate(&transaction).unwrap();
|
|
db::folders::update_folder_sync_status(&transaction, "INBOX", 2, 1, 0).unwrap();
|
|
let new_namespace = HeaderSync::capture(&transaction).unwrap();
|
|
transaction.commit().unwrap();
|
|
assert!(stale.begin_write(&conn).is_err());
|
|
assert!(before_reset.begin_write(&conn).is_err());
|
|
let transaction = new_namespace.begin_write(&conn).unwrap();
|
|
upsert_and_index_headers(&transaction, "INBOX", &[sync_header(7)], &engine, "test")
|
|
.unwrap();
|
|
transaction.commit().unwrap();
|
|
assert!(
|
|
db::messages::get_single_message(&conn, "INBOX", 7)
|
|
.unwrap()
|
|
.is_some()
|
|
);
|
|
assert_eq!(
|
|
db::folders::get_folder(&conn, "INBOX")
|
|
.unwrap()
|
|
.unwrap()
|
|
.uid_validity,
|
|
2
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn chunks_bound_existing_messages_even_with_sparse_uids() {
|
|
assert_eq!(header_chunks(&[]).count(), 0);
|
|
let sparse = [1, 10_000_000, u32::MAX];
|
|
assert_eq!(
|
|
header_chunks(&sparse).collect::<Vec<_>>(),
|
|
vec!["1,10000000,4294967295"]
|
|
);
|
|
|
|
let uids: Vec<_> = (1..=501).map(|uid| uid * 1_000_000).collect();
|
|
let chunks: Vec<_> = header_chunks(&uids).collect();
|
|
assert_eq!(chunks.len(), 3);
|
|
assert_eq!(chunks[0].split(',').count(), 250);
|
|
assert_eq!(chunks[1].split(',').count(), 250);
|
|
assert_eq!(chunks[2].split(',').count(), 1);
|
|
let fetched: Vec<u32> = chunks
|
|
.iter()
|
|
.flat_map(|chunk| chunk.split(',').map(|uid| uid.parse().unwrap()))
|
|
.collect();
|
|
assert_eq!(fetched, uids);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn empty_background_sync_records_completion() {
|
|
for uid_range in [(1, 0), (1, 100)] {
|
|
let data_dir = tempfile::tempdir().unwrap();
|
|
let path = data_dir.path().to_str().unwrap();
|
|
let user_hash = "empty-sync";
|
|
let conn = db::pool::open_user_db(path, user_hash).unwrap();
|
|
db::folders::upsert_folder(&conn, "Sent", None, None, "", true, 99, 7, 1, 0).unwrap();
|
|
db::folders::invalidate_folder_freshness(&conn, "Sent").unwrap();
|
|
assert!(!db::folders::is_folder_fresh(&conn, "Sent", 30).unwrap());
|
|
let header_sync = HeaderSync::capture(&conn).unwrap();
|
|
drop(conn);
|
|
|
|
let client = Arc::new(MockImapClient::new().with_folder_status_extended(
|
|
crate::imap::types::FolderStatusExtended {
|
|
uid_validity: 42,
|
|
exists: 0,
|
|
uid_next: 1,
|
|
unseen: 0,
|
|
highest_modseq: 0,
|
|
},
|
|
));
|
|
let config = Arc::new(AppConfig {
|
|
data_dir: path.to_owned(),
|
|
..serde_json::from_str::<AppConfig>("{}").unwrap()
|
|
});
|
|
let event_bus = Arc::new(EventBus::new());
|
|
let mut events = event_bus.subscribe(user_hash).await;
|
|
|
|
sync_remaining_headers(BackgroundSyncParams {
|
|
creds: ImapCredentials {
|
|
host: "localhost".into(),
|
|
port: 143,
|
|
tls: false,
|
|
tls_verify: false,
|
|
email: "test@example.com".into(),
|
|
password: "test".into(),
|
|
},
|
|
folder: "Sent".into(),
|
|
imap_client: client.clone(),
|
|
config,
|
|
user_hash: user_hash.into(),
|
|
search_engine: Arc::new(SearchEngine::new(data_dir.path().to_path_buf())),
|
|
event_bus,
|
|
uid_range,
|
|
header_sync,
|
|
uid_validity: 42,
|
|
highest_modseq: 0,
|
|
})
|
|
.await;
|
|
|
|
let conn = db::pool::open_user_db(path, user_hash).unwrap();
|
|
let folder = db::folders::get_folder(&conn, "Sent").unwrap().unwrap();
|
|
assert_eq!(folder.uid_validity, 42);
|
|
assert_eq!(folder.total_count, 0);
|
|
assert_eq!(folder.unread_count, 0);
|
|
assert!(db::folders::is_folder_fresh(&conn, "Sent", 30).unwrap());
|
|
assert!(!db::folders::is_folder_messages_invalidated(&conn, "Sent").unwrap());
|
|
assert!(client.header_requests.lock().unwrap().is_empty());
|
|
let emitted = drain_events(&mut events);
|
|
assert!(emitted.iter().any(|event| matches!(
|
|
event,
|
|
MailEvent::FolderStatus {
|
|
folder,
|
|
total_count: 0,
|
|
unread_count: 0,
|
|
} if folder == "Sent"
|
|
)));
|
|
assert!(emitted.iter().any(|event| matches!(
|
|
event,
|
|
MailEvent::SyncStatus {
|
|
active: false,
|
|
folder: Some(folder),
|
|
pending: 0,
|
|
} if folder == "Sent"
|
|
)));
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn sparse_background_sync_fetches_existing_uids_in_one_batch() {
|
|
let data_dir = tempfile::tempdir().unwrap();
|
|
let path = data_dir.path().to_str().unwrap();
|
|
let user_hash = "sparse-sync";
|
|
let conn = db::pool::open_user_db(path, user_hash).unwrap();
|
|
db::folders::upsert_folder(&conn, "INBOX", None, None, "", true, 0, 0, 1, 0).unwrap();
|
|
let header_sync = HeaderSync::capture(&conn).unwrap();
|
|
drop(conn);
|
|
let headers = [1, 9_999_899, 10_000_000].map(|uid| crate::imap::types::ImapMessageHeader {
|
|
uid,
|
|
subject: Some(format!("Message {uid}")),
|
|
from: Vec::new(),
|
|
to: Vec::new(),
|
|
cc: Vec::new(),
|
|
date: None,
|
|
flags: Vec::new(),
|
|
has_attachments: false,
|
|
size: 42,
|
|
message_id: None,
|
|
in_reply_to: None,
|
|
references: None,
|
|
reaction: None,
|
|
});
|
|
let client = Arc::new(MockImapClient::new().with_headers(headers.to_vec()));
|
|
let config = Arc::new(AppConfig {
|
|
data_dir: path.to_owned(),
|
|
..serde_json::from_str::<AppConfig>("{}").unwrap()
|
|
});
|
|
let event_bus = Arc::new(EventBus::new());
|
|
let mut events = event_bus.subscribe(user_hash).await;
|
|
|
|
sync_remaining_headers(BackgroundSyncParams {
|
|
creds: ImapCredentials {
|
|
host: "localhost".into(),
|
|
port: 143,
|
|
tls: false,
|
|
tls_verify: false,
|
|
email: "test@example.com".into(),
|
|
password: "test".into(),
|
|
},
|
|
folder: "INBOX".into(),
|
|
imap_client: client.clone(),
|
|
config,
|
|
user_hash: user_hash.into(),
|
|
search_engine: Arc::new(SearchEngine::new(data_dir.path().to_path_buf())),
|
|
event_bus,
|
|
uid_range: (1, 9_999_900),
|
|
header_sync,
|
|
uid_validity: 1,
|
|
highest_modseq: 0,
|
|
})
|
|
.await;
|
|
|
|
assert_eq!(
|
|
*client.header_requests.lock().unwrap(),
|
|
vec![("INBOX".into(), "1,9999899".into())]
|
|
);
|
|
let emitted = drain_events(&mut events);
|
|
assert!(emitted.iter().any(|event| matches!(
|
|
event,
|
|
MailEvent::SyncStatus {
|
|
active: false,
|
|
folder: Some(folder),
|
|
pending: 0,
|
|
} if folder == "INBOX"
|
|
)));
|
|
let conn = db::pool::open_user_db(path, user_hash).unwrap();
|
|
assert!(
|
|
db::messages::get_single_message(&conn, "INBOX", 1)
|
|
.unwrap()
|
|
.is_some()
|
|
);
|
|
assert!(
|
|
db::messages::get_single_message(&conn, "INBOX", 9_999_899)
|
|
.unwrap()
|
|
.is_some()
|
|
);
|
|
}
|
|
}
|