1 parent
47c3cd7f65
commit
cfc74d9db6
33 files changed
+850
-244
No files matched your search
@@ -162,6 +162,25 @@ pub fn refresh_unread_count(conn: &Connection, folder_name: &str) -> Result<(),
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn set_unread_count(
|
||||
conn: &Connection,
|
||||
folder_name: &str,
|
||||
unread_count: u32,
|
||||
) -> Result<(), String> {
|
||||
conn.execute(
|
||||
crate::dbq!(db_folders::Q_028),
|
||||
params![unread_count, folder_name],
|
||||
)
|
||||
.map_err(|e| format!("Failed to set unread_count: {e}"))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn mark_folder_messages_fresh(conn: &Connection, folder_name: &str) -> Result<(), String> {
|
||||
conn.execute(crate::dbq!(db_folders::Q_029), params![folder_name])
|
||||
.map_err(|e| format!("Failed to mark folder messages fresh: {e}"))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn adjust_unread_count(conn: &Connection, folder_name: &str, delta: i32) -> Result<(), String> {
|
||||
conn.execute(crate::dbq!(db_folders::Q_010), params![delta, folder_name])
|
||||
.map_err(|e| format!("Failed to adjust unread_count: {e}"))?;
|
||||
|
||||
@@ -235,6 +235,9 @@ pub mod db_folders {
|
||||
pub const Q_026: &str =
|
||||
"UPDATE folders SET messages_updated_at = '2000-01-01 00:00:00' WHERE name = 'Sent'";
|
||||
pub const Q_027: &str = "SELECT messages_updated_at FROM folders WHERE name = 'Sent'";
|
||||
pub const Q_028: &str = "UPDATE folders SET unread_count = ? WHERE name = ?";
|
||||
pub const Q_029: &str =
|
||||
"UPDATE folders SET messages_updated_at = CURRENT_TIMESTAMP WHERE name = ?";
|
||||
pub const FORMAT_001: &str = "DELETE FROM folders WHERE name NOT IN ({})";
|
||||
}
|
||||
|
||||
@@ -299,6 +302,10 @@ pub mod db_messages {
|
||||
pub const Q_011: &str = "DELETE FROM messages WHERE folder = ? AND uid = ?";
|
||||
pub const Q_012: &str = "SELECT folder, uid FROM messages
|
||||
WHERE body_cached = 0
|
||||
AND lower(folder) NOT IN ('trash', 'junk', 'spam')
|
||||
AND lower(folder) NOT LIKE '%/trash'
|
||||
AND lower(folder) NOT LIKE '%/junk'
|
||||
AND lower(folder) NOT LIKE '%/spam'
|
||||
ORDER BY date_epoch DESC
|
||||
LIMIT ?";
|
||||
pub const Q_013: &str = "SELECT uid, flags FROM messages WHERE folder = ?";
|
||||
|
||||
@@ -236,6 +236,9 @@ pub mod db_folders {
|
||||
pub const Q_026: &str =
|
||||
"UPDATE folders SET messages_updated_at = '2000-01-01 00:00:00' WHERE name = 'Sent'";
|
||||
pub const Q_027: &str = "SELECT messages_updated_at FROM folders WHERE name = 'Sent'";
|
||||
pub const Q_028: &str = "UPDATE folders SET unread_count = $1 WHERE name = $2";
|
||||
pub const Q_029: &str =
|
||||
"UPDATE folders SET messages_updated_at = CURRENT_TIMESTAMP::text WHERE name = $1";
|
||||
pub const FORMAT_001: &str = "DELETE FROM folders WHERE name NOT IN ({})";
|
||||
}
|
||||
|
||||
@@ -299,6 +302,10 @@ pub mod db_messages {
|
||||
pub const Q_011: &str = "DELETE FROM messages WHERE folder = $1 AND uid = $2";
|
||||
pub const Q_012: &str = "SELECT folder, uid FROM messages
|
||||
WHERE body_cached = 0
|
||||
AND lower(folder) NOT IN ('trash', 'junk', 'spam')
|
||||
AND lower(folder) NOT LIKE '%/trash'
|
||||
AND lower(folder) NOT LIKE '%/junk'
|
||||
AND lower(folder) NOT LIKE '%/spam'
|
||||
ORDER BY date_epoch DESC
|
||||
LIMIT $1";
|
||||
pub const Q_013: &str = "SELECT uid, flags FROM messages WHERE folder = $1";
|
||||
|
||||
@@ -230,6 +230,9 @@ pub mod db_folders {
|
||||
pub const Q_026: &str =
|
||||
"UPDATE folders SET messages_updated_at = '2000-01-01 00:00:00' WHERE name = 'Sent'";
|
||||
pub const Q_027: &str = "SELECT messages_updated_at FROM folders WHERE name = 'Sent'";
|
||||
pub const Q_028: &str = "UPDATE folders SET unread_count = ?1 WHERE name = ?2";
|
||||
pub const Q_029: &str =
|
||||
"UPDATE folders SET messages_updated_at = datetime('now') WHERE name = ?1";
|
||||
pub const FORMAT_001: &str = "DELETE FROM folders WHERE name NOT IN ({})";
|
||||
}
|
||||
|
||||
@@ -288,6 +291,10 @@ pub mod db_messages {
|
||||
pub const Q_011: &str = "DELETE FROM messages WHERE folder = ?1 AND uid = ?2";
|
||||
pub const Q_012: &str = "SELECT folder, uid FROM messages
|
||||
WHERE body_cached = 0
|
||||
AND lower(folder) NOT IN ('trash', 'junk', 'spam')
|
||||
AND lower(folder) NOT LIKE '%/trash'
|
||||
AND lower(folder) NOT LIKE '%/junk'
|
||||
AND lower(folder) NOT LIKE '%/spam'
|
||||
ORDER BY date_epoch DESC
|
||||
LIMIT ?1";
|
||||
pub const Q_013: &str = "SELECT uid, flags FROM messages WHERE folder = ?1";
|
||||
|
||||
@@ -20,6 +20,11 @@ pub enum MailEvent {
|
||||
},
|
||||
|
||||
FolderUpdated,
|
||||
|
||||
IndexingStatus {
|
||||
active: bool,
|
||||
pending: u32,
|
||||
},
|
||||
}
|
||||
|
||||
pub struct EventBus {
|
||||
|
||||
+21
-2
@@ -303,6 +303,7 @@ async fn fetch_new_after_idle(
|
||||
};
|
||||
|
||||
let server_exists = mailbox.exists;
|
||||
let server_unseen = mailbox.unseen;
|
||||
let server_modseq = mailbox.highest_modseq.unwrap_or(0);
|
||||
let uid_validity = mailbox.uid_validity.unwrap_or(0);
|
||||
|
||||
@@ -436,7 +437,16 @@ async fn fetch_new_after_idle(
|
||||
server_exists,
|
||||
server_modseq,
|
||||
);
|
||||
let _ = db::folders::refresh_unread_count(&conn, folder);
|
||||
if let Some(unseen) = server_unseen {
|
||||
let _ = db::folders::set_unread_count(&conn, folder, unseen);
|
||||
} else {
|
||||
let _ = db::folders::refresh_unread_count(&conn, folder);
|
||||
}
|
||||
if has_new && new_count == 0 {
|
||||
let _ = db::folders::invalidate_folder_freshness(&conn, folder);
|
||||
} else {
|
||||
let _ = db::folders::mark_folder_messages_fresh(&conn, folder);
|
||||
}
|
||||
let _ = transaction.commit();
|
||||
}
|
||||
|
||||
@@ -550,7 +560,16 @@ async fn fetch_new_after_idle(
|
||||
server_exists,
|
||||
server_modseq,
|
||||
);
|
||||
let _ = db::folders::refresh_unread_count(&conn, folder);
|
||||
if let Some(unseen) = server_unseen {
|
||||
let _ = db::folders::set_unread_count(&conn, folder, unseen);
|
||||
} else {
|
||||
let _ = db::folders::refresh_unread_count(&conn, folder);
|
||||
}
|
||||
if has_new && new_count == 0 {
|
||||
let _ = db::folders::invalidate_folder_freshness(&conn, folder);
|
||||
} else {
|
||||
let _ = db::folders::mark_folder_messages_fresh(&conn, folder);
|
||||
}
|
||||
let _ = transaction.commit();
|
||||
}
|
||||
|
||||
|
||||
+444
-127
@@ -9,7 +9,7 @@ use crate::realtime::events::{EventBus, MailEvent};
|
||||
use crate::search::engine::UserIndex;
|
||||
use crate::search::engine::{IndexableMessage, SearchEngine};
|
||||
|
||||
const SYNC_INTERVAL_SECS: u64 = 180;
|
||||
const SYNC_INTERVAL_SECS: u64 = 30;
|
||||
|
||||
pub async fn sync_loop(
|
||||
user_hash: String,
|
||||
@@ -20,8 +20,7 @@ pub async fn sync_loop(
|
||||
search_engine: Arc<SearchEngine>,
|
||||
) {
|
||||
let mut interval = tokio::time::interval(std::time::Duration::from_secs(SYNC_INTERVAL_SECS));
|
||||
|
||||
interval.tick().await;
|
||||
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
|
||||
|
||||
loop {
|
||||
interval.tick().await;
|
||||
@@ -62,6 +61,7 @@ async fn run_sync(
|
||||
name: f.name,
|
||||
uid_validity: f.uid_validity,
|
||||
highest_modseq: f.highest_modseq,
|
||||
total_count: f.total_count,
|
||||
})
|
||||
.collect::<Vec<_>>())
|
||||
})
|
||||
@@ -107,7 +107,12 @@ async fn run_sync(
|
||||
let cached_count = database
|
||||
.run(move |conn| db::messages::count_messages(conn, &name))
|
||||
.await?;
|
||||
if status.exists == cached_count {
|
||||
let known_count = if cached_count == 0 && folder.total_count > 0 {
|
||||
folder.total_count
|
||||
} else {
|
||||
cached_count
|
||||
};
|
||||
if status.exists == known_count {
|
||||
tracing::debug!(
|
||||
folder = %folder_name,
|
||||
modseq = status.highest_modseq,
|
||||
@@ -125,6 +130,8 @@ async fn run_sync(
|
||||
imap_client,
|
||||
folder_name,
|
||||
cached_modseq,
|
||||
folder.total_count,
|
||||
folder.uid_validity != 0,
|
||||
&status,
|
||||
event_bus,
|
||||
&search_engine,
|
||||
@@ -142,6 +149,8 @@ async fn run_sync(
|
||||
creds,
|
||||
imap_client,
|
||||
folder_name,
|
||||
folder.total_count,
|
||||
folder.uid_validity != 0,
|
||||
&status,
|
||||
event_bus,
|
||||
&search_engine,
|
||||
@@ -157,8 +166,15 @@ async fn run_sync(
|
||||
}
|
||||
|
||||
if search_engine.is_enabled()
|
||||
&& let Err(e) =
|
||||
index_message_bodies(user_hash, creds, config, imap_client, search_engine).await
|
||||
&& let Err(e) = index_message_bodies(
|
||||
user_hash,
|
||||
creds,
|
||||
config,
|
||||
imap_client,
|
||||
event_bus,
|
||||
search_engine,
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::debug!(error = %e, "Deep index phase skipped or failed");
|
||||
}
|
||||
@@ -173,6 +189,7 @@ async fn index_message_bodies(
|
||||
creds: &ImapCredentials,
|
||||
config: &AppConfig,
|
||||
imap_client: &dyn ImapClient,
|
||||
event_bus: &EventBus,
|
||||
search_engine: Arc<SearchEngine>,
|
||||
) -> Result<(), String> {
|
||||
let database = db::pool::UserDb::new(&config.data_dir, user_hash);
|
||||
@@ -186,81 +203,263 @@ async fn index_message_bodies(
|
||||
.await?;
|
||||
|
||||
if unindexed.is_empty() {
|
||||
event_bus
|
||||
.publish(
|
||||
user_hash,
|
||||
MailEvent::IndexingStatus {
|
||||
active: false,
|
||||
pending: 0,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
event_bus
|
||||
.publish(
|
||||
user_hash,
|
||||
MailEvent::IndexingStatus {
|
||||
active: true,
|
||||
pending: unindexed.len() as u32,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
tracing::debug!(count = unindexed.len(), "Deep indexing message bodies");
|
||||
|
||||
for (folder, uid) in &unindexed {
|
||||
if UserIndex::is_excluded_folder(folder) {
|
||||
continue;
|
||||
}
|
||||
|
||||
let body = match imap_client.fetch_body(creds, folder, *uid).await {
|
||||
Ok(b) => b,
|
||||
Err(e) => {
|
||||
tracing::debug!(folder = %folder, uid = uid, error = %e, "Deep index: failed to fetch body");
|
||||
let result: Result<(), String> = async {
|
||||
for (folder, uid) in &unindexed {
|
||||
if UserIndex::is_excluded_folder(folder) {
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
let folder = folder.clone();
|
||||
let uid = *uid;
|
||||
let engine = search_engine.clone();
|
||||
let hash = user_hash.to_owned();
|
||||
database
|
||||
.run(move |conn| {
|
||||
let transaction = conn.unchecked_transaction().map_err(|e| e.to_string())?;
|
||||
let att_json = serde_json::to_string(&body.attachments).ok();
|
||||
let detected_theme = body
|
||||
.text_html
|
||||
.as_ref()
|
||||
.and_then(|h| email_theme::detect_email_theme(h))
|
||||
.map(|t| t.as_i32());
|
||||
db::messages::cache_message_body(
|
||||
&transaction,
|
||||
&folder,
|
||||
uid,
|
||||
body.text_html.as_deref(),
|
||||
body.text_plain.as_deref(),
|
||||
att_json.as_deref(),
|
||||
Some(&body.raw_headers),
|
||||
detected_theme,
|
||||
)?;
|
||||
if let Some(text) = body.text_plain
|
||||
&& let Some(msg) = db::messages::get_single_message(&transaction, &folder, uid)?
|
||||
{
|
||||
engine
|
||||
.open_user_index(&hash)?
|
||||
.index_message_with_connection(
|
||||
&transaction,
|
||||
&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,
|
||||
date_epoch: db::messages::parse_date_to_epoch_public(&msg.date),
|
||||
has_attachments: msg.has_attachments,
|
||||
},
|
||||
)?;
|
||||
let body = match imap_client.fetch_body(creds, folder, *uid).await {
|
||||
Ok(b) => b,
|
||||
Err(e) => {
|
||||
tracing::debug!(folder = %folder, uid = uid, error = %e, "Deep index: failed to fetch body");
|
||||
continue;
|
||||
}
|
||||
transaction.commit().map_err(|e| e.to_string())?;
|
||||
Ok(())
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
};
|
||||
|
||||
tracing::debug!(count = unindexed.len(), "Deep index batch complete");
|
||||
Ok(())
|
||||
let folder = folder.clone();
|
||||
let uid = *uid;
|
||||
let engine = search_engine.clone();
|
||||
let hash = user_hash.to_owned();
|
||||
database
|
||||
.run(move |conn| {
|
||||
let transaction = conn.unchecked_transaction().map_err(|e| e.to_string())?;
|
||||
let att_json = serde_json::to_string(&body.attachments).ok();
|
||||
let detected_theme = body
|
||||
.text_html
|
||||
.as_ref()
|
||||
.and_then(|h| email_theme::detect_email_theme(h))
|
||||
.map(|t| t.as_i32());
|
||||
db::messages::cache_message_body(
|
||||
&transaction,
|
||||
&folder,
|
||||
uid,
|
||||
body.text_html.as_deref(),
|
||||
body.text_plain.as_deref(),
|
||||
att_json.as_deref(),
|
||||
Some(&body.raw_headers),
|
||||
detected_theme,
|
||||
)?;
|
||||
if let Some(text) = body.text_plain
|
||||
&& let Some(msg) =
|
||||
db::messages::get_single_message(&transaction, &folder, uid)?
|
||||
{
|
||||
engine
|
||||
.open_user_index(&hash)?
|
||||
.index_message_with_connection(
|
||||
&transaction,
|
||||
&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,
|
||||
date_epoch: db::messages::parse_date_to_epoch_public(
|
||||
&msg.date,
|
||||
),
|
||||
has_attachments: msg.has_attachments,
|
||||
},
|
||||
)?;
|
||||
}
|
||||
transaction.commit().map_err(|e| e.to_string())?;
|
||||
Ok(())
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
|
||||
tracing::debug!(count = unindexed.len(), "Deep index batch complete");
|
||||
Ok(())
|
||||
}
|
||||
.await;
|
||||
|
||||
event_bus
|
||||
.publish(
|
||||
user_hash,
|
||||
MailEvent::IndexingStatus {
|
||||
active: false,
|
||||
pending: 0,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
result
|
||||
}
|
||||
|
||||
struct FolderSnapshot {
|
||||
name: String,
|
||||
uid_validity: u32,
|
||||
highest_modseq: u64,
|
||||
total_count: u32,
|
||||
}
|
||||
|
||||
fn store_headers(
|
||||
conn: &crate::db::pool::Connection,
|
||||
folder: &str,
|
||||
headers: &[crate::imap::types::ImapMessageHeader],
|
||||
search_engine: &Arc<SearchEngine>,
|
||||
user_hash: &str,
|
||||
) -> Result<(), String> {
|
||||
if headers.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
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(|address| address.address.as_str())
|
||||
.unwrap_or("");
|
||||
let from_name = header
|
||||
.from
|
||||
.first()
|
||||
.and_then(|address| address.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(),
|
||||
)?;
|
||||
db::contacts::populate_known_addresses(conn, from_address, from_name, &to_json, &cc_json)?;
|
||||
if sync_keywords {
|
||||
db::tags::import_imap_keywords(conn, header.uid, folder, &header.flags)?;
|
||||
}
|
||||
}
|
||||
|
||||
if !UserIndex::is_excluded_folder(folder)
|
||||
&& let Ok(user_index) = search_engine.open_user_index(user_hash)
|
||||
{
|
||||
let indexable = headers
|
||||
.iter()
|
||||
.map(|header| {
|
||||
let from_address = header
|
||||
.from
|
||||
.first()
|
||||
.map(|address| address.address.as_str())
|
||||
.unwrap_or("");
|
||||
let from_name = header
|
||||
.from
|
||||
.first()
|
||||
.and_then(|address| address.name.as_deref())
|
||||
.unwrap_or("");
|
||||
let to_json =
|
||||
serde_json::to_string(&header.to).unwrap_or_else(|_| "[]".to_string());
|
||||
IndexableMessage {
|
||||
uid: header.uid,
|
||||
folder: folder.to_string(),
|
||||
subject: header.subject.clone().unwrap_or_default(),
|
||||
from_address: from_address.to_string(),
|
||||
from_name: from_name.to_string(),
|
||||
to_addresses: to_json,
|
||||
body_text: String::new(),
|
||||
date_epoch: db::messages::parse_date_to_epoch_public(
|
||||
header.date.as_deref().unwrap_or(""),
|
||||
),
|
||||
has_attachments: header.has_attachments,
|
||||
}
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
user_index.index_messages_batch_with_connection(conn, &indexable)?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn latest_header_metadata(
|
||||
headers: &[crate::imap::types::ImapMessageHeader],
|
||||
) -> (Option<u32>, Option<String>, Option<String>) {
|
||||
let Some(header) = headers.iter().max_by_key(|header| header.uid) else {
|
||||
return (None, None, None);
|
||||
};
|
||||
let sender = header.from.first().and_then(|address| {
|
||||
address
|
||||
.name
|
||||
.as_ref()
|
||||
.filter(|name| !name.is_empty())
|
||||
.cloned()
|
||||
.or_else(|| (!address.address.is_empty()).then(|| address.address.clone()))
|
||||
});
|
||||
let subject = header
|
||||
.subject
|
||||
.as_ref()
|
||||
.filter(|subject| !subject.is_empty())
|
||||
.cloned();
|
||||
(Some(header.uid), sender, subject)
|
||||
}
|
||||
|
||||
async fn fetch_new_headers(
|
||||
creds: &ImapCredentials,
|
||||
imap_client: &dyn ImapClient,
|
||||
folder_name: &str,
|
||||
max_cached_uid: u32,
|
||||
) -> (Vec<crate::imap::types::ImapMessageHeader>, bool) {
|
||||
let uid_range = format!("{}:*", max_cached_uid.saturating_add(1));
|
||||
match imap_client
|
||||
.fetch_headers(creds, folder_name, &uid_range)
|
||||
.await
|
||||
{
|
||||
Ok(mut headers) => {
|
||||
headers.retain(|header| header.uid > max_cached_uid);
|
||||
headers.sort_by_key(|header| header.uid);
|
||||
headers.dedup_by_key(|header| header.uid);
|
||||
(headers, false)
|
||||
}
|
||||
Err(error) => {
|
||||
tracing::debug!(
|
||||
folder = %folder_name,
|
||||
error = %error,
|
||||
"Periodic sync could not fetch new headers"
|
||||
);
|
||||
(Vec::new(), true)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
@@ -271,6 +470,8 @@ async fn sync_condstore(
|
||||
imap_client: &dyn ImapClient,
|
||||
folder_name: &str,
|
||||
cached_modseq: u64,
|
||||
previous_total_count: u32,
|
||||
baseline_known: bool,
|
||||
status: &crate::imap::types::FolderStatusExtended,
|
||||
event_bus: &EventBus,
|
||||
search_engine: &Arc<SearchEngine>,
|
||||
@@ -280,10 +481,10 @@ async fn sync_condstore(
|
||||
.await
|
||||
{
|
||||
Ok(result) => result,
|
||||
Err(e) => {
|
||||
Err(error) => {
|
||||
tracing::debug!(
|
||||
folder = %folder_name,
|
||||
error = %e,
|
||||
error = %error,
|
||||
"CONDSTORE fetch failed, falling back to full sync"
|
||||
);
|
||||
return sync_full(
|
||||
@@ -292,6 +493,8 @@ async fn sync_condstore(
|
||||
creds,
|
||||
imap_client,
|
||||
folder_name,
|
||||
previous_total_count,
|
||||
baseline_known,
|
||||
status,
|
||||
event_bus,
|
||||
search_engine,
|
||||
@@ -302,15 +505,39 @@ async fn sync_condstore(
|
||||
|
||||
let database = db::pool::UserDb::new(&config.data_dir, user_hash);
|
||||
let name = folder_name.to_owned();
|
||||
let cached_count = database
|
||||
.run(move |conn| db::messages::count_messages(conn, &name))
|
||||
let (cached_count, max_cached_uid) = database
|
||||
.run(move |conn| {
|
||||
Ok((
|
||||
db::messages::count_messages(conn, &name)?,
|
||||
db::messages::max_uid(conn, &name)?,
|
||||
))
|
||||
})
|
||||
.await?;
|
||||
|
||||
let cache_initialized = cached_count > 0 || (baseline_known && previous_total_count == 0);
|
||||
let server_has_new_uid = status.uid_next > max_cached_uid.saturating_add(1);
|
||||
let (new_headers, fetch_failed) = if cache_initialized && server_has_new_uid {
|
||||
fetch_new_headers(creds, imap_client, folder_name, max_cached_uid).await
|
||||
} else {
|
||||
(Vec::new(), false)
|
||||
};
|
||||
let expected_new_count = if baseline_known {
|
||||
status.exists.saturating_sub(previous_total_count)
|
||||
} else {
|
||||
0
|
||||
};
|
||||
let new_event_count = expected_new_count.max(new_headers.len() as u32);
|
||||
let has_unfetched_new = (!cache_initialized && expected_new_count > 0)
|
||||
|| (server_has_new_uid && (fetch_failed || new_headers.is_empty()))
|
||||
|| expected_new_count > new_headers.len() as u32;
|
||||
let (latest_uid, latest_sender, latest_subject) = latest_header_metadata(&new_headers);
|
||||
|
||||
let imap_uids = if status.exists < cached_count {
|
||||
Some(
|
||||
imap_client
|
||||
.fetch_uids_and_flags(creds, folder_name)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?
|
||||
.map_err(|error| error.to_string())?
|
||||
.into_iter()
|
||||
.map(|(uid, _)| uid)
|
||||
.collect::<HashSet<_>>(),
|
||||
@@ -318,17 +545,21 @@ async fn sync_condstore(
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
let name = folder_name.to_owned();
|
||||
let status = status.clone();
|
||||
let search_engine = Arc::clone(search_engine);
|
||||
let search_engine_for_write = Arc::clone(search_engine);
|
||||
let search_user_hash = user_hash.to_owned();
|
||||
let (folder_changed, new_messages) = database
|
||||
let headers_for_write = new_headers;
|
||||
let (folder_changed, non_new_change) = database
|
||||
.run(move |conn| {
|
||||
let sync_keywords = db::mail_preferences::get_preferences(conn)
|
||||
.map(|prefs| prefs.sync_imap_keywords)
|
||||
.unwrap_or(false);
|
||||
let transaction = conn.unchecked_transaction().map_err(|e| e.to_string())?;
|
||||
let mut folder_changed = !changed.is_empty();
|
||||
let transaction = conn
|
||||
.unchecked_transaction()
|
||||
.map_err(|error| error.to_string())?;
|
||||
let mut non_new_change = !changed.is_empty();
|
||||
for (uid, mut flags) in changed {
|
||||
flags.sort();
|
||||
db::messages::update_message_flags(&transaction, &name, uid, &flags.join(","))?;
|
||||
@@ -340,15 +571,23 @@ async fn sync_condstore(
|
||||
for (uid, _) in db::messages::get_all_uids_and_flags(&transaction, &name)? {
|
||||
if !imap_uids.contains(&uid) {
|
||||
db::messages::delete_message(&transaction, &name, uid)?;
|
||||
if let Ok(index) = search_engine.open_user_index(&search_user_hash) {
|
||||
if let Ok(index) =
|
||||
search_engine_for_write.open_user_index(&search_user_hash)
|
||||
{
|
||||
index.delete_message_with_connection(&transaction, uid, &name)?;
|
||||
}
|
||||
folder_changed = true;
|
||||
non_new_change = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
let max_uid = db::messages::max_uid(&transaction, &name)?;
|
||||
let new_messages = status.uid_next > max_uid.saturating_add(1);
|
||||
|
||||
store_headers(
|
||||
&transaction,
|
||||
&name,
|
||||
&headers_for_write,
|
||||
&search_engine_for_write,
|
||||
&search_user_hash,
|
||||
)?;
|
||||
let modseq = if new_modseq > 0 {
|
||||
new_modseq
|
||||
} else {
|
||||
@@ -361,30 +600,36 @@ async fn sync_condstore(
|
||||
status.exists,
|
||||
modseq,
|
||||
)?;
|
||||
if new_messages {
|
||||
db::folders::set_unread_count(&transaction, &name, status.unseen)?;
|
||||
if has_unfetched_new {
|
||||
db::folders::invalidate_folder_freshness(&transaction, &name)?;
|
||||
folder_changed = true;
|
||||
} else if cache_initialized {
|
||||
db::folders::mark_folder_messages_fresh(&transaction, &name)?;
|
||||
}
|
||||
transaction.commit().map_err(|e| e.to_string())?;
|
||||
Ok((folder_changed, new_messages))
|
||||
transaction.commit().map_err(|error| error.to_string())?;
|
||||
Ok((
|
||||
non_new_change || !headers_for_write.is_empty(),
|
||||
non_new_change,
|
||||
))
|
||||
})
|
||||
.await?;
|
||||
if new_messages {
|
||||
|
||||
if new_event_count > 0 {
|
||||
event_bus
|
||||
.publish(
|
||||
user_hash,
|
||||
MailEvent::NewMessages {
|
||||
folder: folder_name.to_owned(),
|
||||
count: 0,
|
||||
latest_uid: None,
|
||||
latest_sender: None,
|
||||
latest_subject: None,
|
||||
count: new_event_count,
|
||||
latest_uid,
|
||||
latest_sender,
|
||||
latest_subject,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
if folder_changed {
|
||||
if non_new_change {
|
||||
event_bus
|
||||
.publish(
|
||||
user_hash,
|
||||
@@ -395,7 +640,7 @@ async fn sync_condstore(
|
||||
.await;
|
||||
}
|
||||
|
||||
Ok(folder_changed)
|
||||
Ok(folder_changed || new_event_count > 0)
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
@@ -405,6 +650,8 @@ async fn sync_full(
|
||||
creds: &ImapCredentials,
|
||||
imap_client: &dyn ImapClient,
|
||||
folder_name: &str,
|
||||
previous_total_count: u32,
|
||||
baseline_known: bool,
|
||||
status: &crate::imap::types::FolderStatusExtended,
|
||||
event_bus: &EventBus,
|
||||
search_engine: &Arc<SearchEngine>,
|
||||
@@ -414,7 +661,44 @@ async fn sync_full(
|
||||
let cached = database
|
||||
.run(move |conn| db::messages::get_all_uids_and_flags(conn, &name))
|
||||
.await?;
|
||||
if cached.is_empty() {
|
||||
let cached_count = cached.len() as u32;
|
||||
let max_cached_uid = cached.iter().map(|(uid, _)| *uid).max().unwrap_or(0);
|
||||
let cache_initialized = cached_count > 0 || (baseline_known && previous_total_count == 0);
|
||||
let server_has_new_uid = status.uid_next > max_cached_uid.saturating_add(1);
|
||||
|
||||
let imap_state = if cached.is_empty() {
|
||||
Vec::new()
|
||||
} else {
|
||||
match imap_client.fetch_uids_and_flags(creds, folder_name).await {
|
||||
Ok(state) => state,
|
||||
Err(error) => {
|
||||
tracing::debug!(
|
||||
folder = %folder_name,
|
||||
error = %error,
|
||||
"Skipping folder sync (full fetch failed)"
|
||||
);
|
||||
return Ok(false);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
let (new_headers, fetch_failed) = if cache_initialized && server_has_new_uid {
|
||||
fetch_new_headers(creds, imap_client, folder_name, max_cached_uid).await
|
||||
} else {
|
||||
(Vec::new(), false)
|
||||
};
|
||||
let expected_new_count = if baseline_known {
|
||||
status.exists.saturating_sub(previous_total_count)
|
||||
} else {
|
||||
0
|
||||
};
|
||||
let new_event_count = expected_new_count.max(new_headers.len() as u32);
|
||||
let has_unfetched_new = (!cache_initialized && expected_new_count > 0)
|
||||
|| (server_has_new_uid && (fetch_failed || new_headers.is_empty()))
|
||||
|| expected_new_count > new_headers.len() as u32;
|
||||
let (latest_uid, latest_sender, latest_subject) = latest_header_metadata(&new_headers);
|
||||
|
||||
if cached.is_empty() && new_headers.is_empty() {
|
||||
let name = folder_name.to_owned();
|
||||
let status = status.clone();
|
||||
database
|
||||
@@ -425,73 +709,78 @@ async fn sync_full(
|
||||
status.uid_validity,
|
||||
status.exists,
|
||||
status.highest_modseq,
|
||||
)
|
||||
)?;
|
||||
db::folders::set_unread_count(conn, &name, status.unseen)?;
|
||||
if has_unfetched_new {
|
||||
db::folders::invalidate_folder_freshness(conn, &name)?;
|
||||
}
|
||||
Ok(())
|
||||
})
|
||||
.await?;
|
||||
return Ok(false);
|
||||
}
|
||||
|
||||
let imap_state = match imap_client.fetch_uids_and_flags(creds, folder_name).await {
|
||||
Ok(state) => state,
|
||||
Err(e) => {
|
||||
tracing::debug!(
|
||||
folder = %folder_name,
|
||||
error = %e,
|
||||
"Skipping folder sync (full fetch failed)"
|
||||
);
|
||||
return Ok(false);
|
||||
if new_event_count > 0 {
|
||||
event_bus
|
||||
.publish(
|
||||
user_hash,
|
||||
MailEvent::NewMessages {
|
||||
folder: folder_name.to_owned(),
|
||||
count: new_event_count,
|
||||
latest_uid: None,
|
||||
latest_sender: None,
|
||||
latest_subject: None,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
}
|
||||
};
|
||||
return Ok(new_event_count > 0);
|
||||
}
|
||||
|
||||
let name = folder_name.to_owned();
|
||||
let status = status.clone();
|
||||
let search_engine = Arc::clone(search_engine);
|
||||
let search_engine_for_write = Arc::clone(search_engine);
|
||||
let search_user_hash = user_hash.to_owned();
|
||||
let folder_changed = database
|
||||
let headers_for_write = new_headers;
|
||||
let non_new_change = database
|
||||
.run(move |conn| {
|
||||
let folder_name = &name;
|
||||
let sync_keywords = db::mail_preferences::get_preferences(conn)
|
||||
.map(|prefs| prefs.sync_imap_keywords)
|
||||
.unwrap_or(false);
|
||||
let transaction = conn.unchecked_transaction().map_err(|e| e.to_string())?;
|
||||
let transaction = conn
|
||||
.unchecked_transaction()
|
||||
.map_err(|error| error.to_string())?;
|
||||
let imap_map: HashMap<u32, Vec<String>> = imap_state
|
||||
.into_iter()
|
||||
.map(|(uid, flags)| {
|
||||
let mut sorted = flags;
|
||||
sorted.sort();
|
||||
(uid, sorted)
|
||||
.map(|(uid, mut flags)| {
|
||||
flags.sort();
|
||||
(uid, flags)
|
||||
})
|
||||
.collect();
|
||||
|
||||
let mut folder_changed = false;
|
||||
|
||||
let mut non_new_change = false;
|
||||
for (uid, cached_flags_csv) in &cached {
|
||||
match imap_map.get(uid) {
|
||||
None => {
|
||||
db::messages::delete_message(&transaction, folder_name, *uid)?;
|
||||
if let Ok(index) = search_engine.open_user_index(&search_user_hash) {
|
||||
if let Ok(index) =
|
||||
search_engine_for_write.open_user_index(&search_user_hash)
|
||||
{
|
||||
index.delete_message_with_connection(
|
||||
&transaction,
|
||||
*uid,
|
||||
folder_name,
|
||||
)?;
|
||||
}
|
||||
folder_changed = true;
|
||||
tracing::debug!(
|
||||
folder = %folder_name,
|
||||
uid = uid,
|
||||
"Removed deleted message from cache"
|
||||
);
|
||||
non_new_change = true;
|
||||
}
|
||||
Some(imap_flags) => {
|
||||
let mut cached_sorted: Vec<&str> = cached_flags_csv
|
||||
let mut cached_sorted = cached_flags_csv
|
||||
.split(',')
|
||||
.filter(|flag| !flag.is_empty())
|
||||
.collect();
|
||||
.collect::<Vec<_>>();
|
||||
cached_sorted.sort();
|
||||
let cached_normalized = cached_sorted.join(",");
|
||||
let imap_flags_csv = imap_flags.join(",");
|
||||
|
||||
if cached_normalized != imap_flags_csv {
|
||||
db::messages::update_message_flags(
|
||||
&transaction,
|
||||
@@ -499,7 +788,7 @@ async fn sync_full(
|
||||
*uid,
|
||||
&imap_flags_csv,
|
||||
)?;
|
||||
folder_changed = true;
|
||||
non_new_change = true;
|
||||
}
|
||||
if sync_keywords {
|
||||
db::tags::import_imap_keywords(
|
||||
@@ -513,6 +802,13 @@ async fn sync_full(
|
||||
}
|
||||
}
|
||||
|
||||
store_headers(
|
||||
&transaction,
|
||||
folder_name,
|
||||
&headers_for_write,
|
||||
&search_engine_for_write,
|
||||
&search_user_hash,
|
||||
)?;
|
||||
db::folders::update_folder_sync_status(
|
||||
&transaction,
|
||||
folder_name,
|
||||
@@ -520,12 +816,33 @@ async fn sync_full(
|
||||
status.exists,
|
||||
status.highest_modseq,
|
||||
)?;
|
||||
transaction.commit().map_err(|e| e.to_string())?;
|
||||
Ok(folder_changed)
|
||||
db::folders::set_unread_count(&transaction, folder_name, status.unseen)?;
|
||||
if has_unfetched_new {
|
||||
db::folders::invalidate_folder_freshness(&transaction, folder_name)?;
|
||||
} else if cache_initialized {
|
||||
db::folders::mark_folder_messages_fresh(&transaction, folder_name)?;
|
||||
}
|
||||
transaction.commit().map_err(|error| error.to_string())?;
|
||||
Ok(non_new_change)
|
||||
})
|
||||
.await?;
|
||||
|
||||
if folder_changed {
|
||||
if new_event_count > 0 {
|
||||
event_bus
|
||||
.publish(
|
||||
user_hash,
|
||||
MailEvent::NewMessages {
|
||||
folder: folder_name.to_owned(),
|
||||
count: new_event_count,
|
||||
latest_uid,
|
||||
latest_sender,
|
||||
latest_subject,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
if non_new_change {
|
||||
event_bus
|
||||
.publish(
|
||||
user_hash,
|
||||
@@ -536,5 +853,5 @@ async fn sync_full(
|
||||
.await;
|
||||
}
|
||||
|
||||
Ok(folder_changed)
|
||||
Ok(non_new_change || new_event_count > 0)
|
||||
}
|
||||
+53
-2
@@ -1,3 +1,4 @@
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
use axum::Extension;
|
||||
@@ -152,6 +153,8 @@ async fn handle_socket_multi_account(
|
||||
|
||||
drop(event_tx);
|
||||
|
||||
let mut watched_folders: HashMap<String, String> = HashMap::new();
|
||||
|
||||
let (mut ws_tx, mut ws_rx) = socket.split();
|
||||
let mut heartbeat_interval = tokio::time::interval(std::time::Duration::from_secs(25));
|
||||
heartbeat_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
|
||||
@@ -178,8 +181,51 @@ async fn handle_socket_multi_account(
|
||||
Some(Ok(Message::Close(_))) | None => break,
|
||||
Some(Ok(Message::Pong(_))) => { last_pong = tokio::time::Instant::now(); }
|
||||
Some(Ok(Message::Text(text))) => {
|
||||
if serde_json::from_str::<serde_json::Value>(&text).ok().and_then(|value| value.get("type").and_then(|kind| kind.as_str()).map(str::to_owned)).as_deref() == Some("heartbeat_ack") {
|
||||
last_pong = tokio::time::Instant::now();
|
||||
if let Ok(value) = serde_json::from_str::<serde_json::Value>(&text) {
|
||||
match value.get("type").and_then(|kind| kind.as_str()) {
|
||||
Some("heartbeat_ack") => {
|
||||
last_pong = tokio::time::Instant::now();
|
||||
}
|
||||
Some("watch_folder") => {
|
||||
let account_id = value.get("accountId").and_then(|v| v.as_str()).unwrap_or("");
|
||||
let folder = value.get("folder").and_then(|v| v.as_str()).unwrap_or("").trim();
|
||||
if !account_id.is_empty() && !folder.is_empty() && folder.len() <= 1024
|
||||
&& let Some(session) = accounts.iter().find(|item| item.account_id == account_id)
|
||||
{
|
||||
let previous = watched_folders
|
||||
.get(account_id)
|
||||
.cloned()
|
||||
.unwrap_or_else(|| "INBOX".to_string());
|
||||
if previous != folder {
|
||||
if !previous.eq_ignore_ascii_case("INBOX") {
|
||||
idle_manager.stop_idle(&session.user_hash, &previous).await;
|
||||
}
|
||||
if !folder.eq_ignore_ascii_case("INBOX") {
|
||||
let creds = 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(),
|
||||
};
|
||||
idle_manager
|
||||
.start_idle(
|
||||
session.user_hash.clone(),
|
||||
folder.to_string(),
|
||||
creds,
|
||||
event_bus.clone(),
|
||||
config.clone(),
|
||||
search_engine.clone(),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
watched_folders.insert(account_id.to_string(), folder.to_string());
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
Some(Ok(_)) => {}
|
||||
@@ -210,6 +256,11 @@ async fn handle_socket_multi_account(
|
||||
}
|
||||
|
||||
for session in &accounts {
|
||||
if let Some(folder) = watched_folders.get(&session.account_id)
|
||||
&& !folder.eq_ignore_ascii_case("INBOX")
|
||||
{
|
||||
idle_manager.stop_idle(&session.user_hash, folder).await;
|
||||
}
|
||||
idle_manager.stop_idle(&session.user_hash, "INBOX").await;
|
||||
event_bus.cleanup(&session.user_hash).await;
|
||||
}
|
||||
|
||||
+1
-12
@@ -25,7 +25,7 @@ struct FolderEntry {
|
||||
unread_count: u32,
|
||||
}
|
||||
|
||||
const FOLDER_LIST_TTL_SECS: u32 = 30;
|
||||
const FOLDER_LIST_TTL_SECS: u32 = 300;
|
||||
|
||||
pub async fn list_folders(
|
||||
Extension(session): Extension<SessionState>,
|
||||
@@ -72,17 +72,6 @@ pub async fn list_folders(
|
||||
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
||||
}
|
||||
|
||||
let all_folders = db::folders::get_all_folders(&conn)
|
||||
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
||||
for f in &all_folders {
|
||||
let invalidated = db::folders::is_folder_messages_invalidated(&conn, &f.name)
|
||||
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
||||
if !invalidated {
|
||||
db::folders::refresh_unread_count(&conn, &f.name)
|
||||
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
||||
}
|
||||
}
|
||||
|
||||
let cached = db::folders::get_all_folders(&conn)
|
||||
.map_err(|e| AppError::InternalError(format!("Database error: {e}")))?;
|
||||
|
||||
|
||||
+42
-22
@@ -121,7 +121,7 @@ pub async fn list_messages(
|
||||
.await
|
||||
.map_err(AppError::InternalError)?;
|
||||
|
||||
if !folder_fresh {
|
||||
if query.refresh || !folder_fresh {
|
||||
let creds = build_creds(&session, &config)?;
|
||||
|
||||
let status = imap_client
|
||||
@@ -254,27 +254,44 @@ pub async fn list_messages(
|
||||
}
|
||||
|
||||
let name = folder.clone();
|
||||
let (messages, total_count, tags_map, flagged_on_top) = 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)?;
|
||||
Ok((messages, total, tags, flagged_on_top))
|
||||
})
|
||||
.await
|
||||
.map_err(AppError::InternalError)?;
|
||||
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,
|
||||
@@ -317,6 +334,8 @@ pub async fn list_messages(
|
||||
per_page: query.per_page,
|
||||
syncing,
|
||||
flagged_on_top,
|
||||
folder_total_count,
|
||||
unread_count,
|
||||
})
|
||||
.into_response())
|
||||
}
|
||||
@@ -1636,6 +1655,7 @@ mod sync_tests {
|
||||
page: 0,
|
||||
per_page: 50,
|
||||
sort: "date_desc".to_string(),
|
||||
refresh: false,
|
||||
}),
|
||||
)
|
||||
.await
|
||||
|
||||
@@ -18,6 +18,8 @@ pub(crate) struct ListMessagesQuery {
|
||||
pub(crate) per_page: u32,
|
||||
#[serde(default = "default_sort")]
|
||||
pub(crate) sort: String,
|
||||
#[serde(default)]
|
||||
pub(crate) refresh: bool,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
@@ -48,6 +50,8 @@ pub(crate) struct ListMessagesResponse {
|
||||
pub(crate) per_page: u32,
|
||||
pub(crate) syncing: bool,
|
||||
pub(crate) flagged_on_top: bool,
|
||||
pub(crate) folder_total_count: u32,
|
||||
pub(crate) unread_count: u32,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Clone)]
|
||||
|
||||
@@ -147,6 +147,28 @@
|
||||
gap: 4px;
|
||||
}
|
||||
|
||||
.indexing-indicator {
|
||||
width: 30px;
|
||||
height: 30px;
|
||||
display: inline-grid;
|
||||
place-items: center;
|
||||
color: var(--accent);
|
||||
}
|
||||
|
||||
.indexing-indicator svg {
|
||||
width: 16px;
|
||||
height: 16px;
|
||||
animation: rustbox-indexing-spin .9s linear infinite;
|
||||
}
|
||||
|
||||
@keyframes rustbox-indexing-spin {
|
||||
to { transform: rotate(360deg); }
|
||||
}
|
||||
|
||||
@media (prefers-reduced-motion: reduce) {
|
||||
.indexing-indicator svg { animation: none; }
|
||||
}
|
||||
|
||||
.account-summary {
|
||||
display: grid;
|
||||
min-width: 120px;
|
||||
|
||||
@@ -80,6 +80,10 @@
|
||||
<button id="contacts-btn" class="icon-button" type="button" title="Contacts" aria-label="Contacts"><svg>
|
||||
<use href="__RUSTBOX_ASSET_PREFIX__/assets/icons.svg#icon-user-round"></use>
|
||||
</svg></button>
|
||||
<span id="indexing-indicator" class="indexing-indicator hidden" role="status" aria-live="polite"
|
||||
title="Indexing messages" aria-label="Indexing messages"><svg aria-hidden="true">
|
||||
<use href="__RUSTBOX_ASSET_PREFIX__/assets/icons.svg#icon-refresh-cw"></use>
|
||||
</svg></span>
|
||||
<button id="calendar-btn" class="icon-button" type="button" title="Calendar" aria-label="Calendar"><svg>
|
||||
<use href="__RUSTBOX_ASSET_PREFIX__/assets/icons.svg#icon-calendar-days"></use>
|
||||
</svg></button>
|
||||
|
||||
@@ -60,6 +60,9 @@
|
||||
realtimePollTimer: null,
|
||||
realtimeSyncPromise: null,
|
||||
realtimeSeenEvents: new Map(),
|
||||
realtimeNotifiedMessages: new Map(),
|
||||
realtimeWatchedFolder: null,
|
||||
indexingActive: false,
|
||||
calendarReminderTimer: null,
|
||||
composeDraftId: null,
|
||||
composeAttachments: [],
|
||||
|
||||
@@ -143,6 +143,7 @@ function renderFolders() {
|
||||
list.append(row);
|
||||
}
|
||||
renderMoveOptions();
|
||||
syncRealtimeWatchedFolder();
|
||||
}
|
||||
|
||||
function updateUnreadAppBadge() {
|
||||
|
||||
@@ -20,11 +20,16 @@ async function loadMessages(reset = false, options = {}) {
|
||||
let loaded = [];
|
||||
let totalCount = 0;
|
||||
let flaggedOnTop = state.currentFolderFlaggedOnTop;
|
||||
let folderTotalCount = null;
|
||||
let folderUnreadCount = null;
|
||||
let lastLoadedPage = state.page;
|
||||
for (const page of pages) {
|
||||
const data = await request(`/folders/${encodeURIComponent(state.folder)}/messages?page=${page}&per_page=${state.perPage}&sort=${sort}`, { method: "GET" });
|
||||
const forceRefresh = options.force ? "&refresh=true" : "";
|
||||
const data = await request(`/folders/${encodeURIComponent(state.folder)}/messages?page=${page}&per_page=${state.perPage}&sort=${sort}${forceRefresh}`, { method: "GET" });
|
||||
totalCount = data.total_count || 0;
|
||||
flaggedOnTop = Boolean(data.flagged_on_top);
|
||||
folderTotalCount = Number(data.folder_total_count);
|
||||
folderUnreadCount = Number(data.unread_count);
|
||||
loaded.push(...(data.messages || []));
|
||||
lastLoadedPage = page;
|
||||
if (loaded.length >= totalCount) break;
|
||||
@@ -33,6 +38,13 @@ async function loadMessages(reset = false, options = {}) {
|
||||
state.total = totalCount;
|
||||
state.currentFolderFlaggedOnTop = flaggedOnTop;
|
||||
state.messages = reset ? loaded : [...state.messages, ...loaded];
|
||||
const folderState = state.folders.find((item) => item.name === state.folder);
|
||||
if (folderState) {
|
||||
if (Number.isFinite(folderTotalCount)) folderState.total_count = folderTotalCount;
|
||||
if (Number.isFinite(folderUnreadCount)) folderState.unread_count = folderUnreadCount;
|
||||
updateUnreadAppBadge();
|
||||
renderFolders();
|
||||
}
|
||||
$("#connection-label").textContent = "Connected";
|
||||
renderMessages();
|
||||
const desiredScroll = savedState?.scrollTop ?? previousScrollTop;
|
||||
|
||||
@@ -160,6 +160,58 @@ function playNotificationSound() {
|
||||
} catch (_) { }
|
||||
}
|
||||
|
||||
function setIndexingIndicator(active, pending = 0) {
|
||||
state.indexingActive = Boolean(active);
|
||||
const indicator = $("#indexing-indicator");
|
||||
if (!indicator) return;
|
||||
indicator.classList.toggle("hidden", !state.indexingActive);
|
||||
const count = Math.max(0, Number(pending) || 0);
|
||||
const label = count > 0 ? `Indexing messages (${count} pending)` : "Indexing messages";
|
||||
indicator.title = label;
|
||||
indicator.setAttribute("aria-label", label);
|
||||
}
|
||||
|
||||
function syncRealtimeWatchedFolder(force = false) {
|
||||
const socket = state.ws;
|
||||
if (!socket || socket.readyState !== WebSocket.OPEN || !state.accountId || !state.folder) return;
|
||||
const key = `${state.accountId}|${state.folder}`;
|
||||
if (!force && state.realtimeWatchedFolder === key) return;
|
||||
try {
|
||||
socket.send(JSON.stringify({ type: "watch_folder", accountId: state.accountId, folder: state.folder }));
|
||||
state.realtimeWatchedFolder = key;
|
||||
} catch (_) { }
|
||||
}
|
||||
|
||||
function newMailNotificationKey(data, accountId) {
|
||||
const uid = Number(data?.latest_uid);
|
||||
if (Number.isFinite(uid) && uid > 0) return `${accountId || ""}|${data?.folder || "INBOX"}|${uid}`;
|
||||
return `${accountId || ""}|${data?.folder || "INBOX"}|${data?.latest_sender || ""}|${data?.latest_subject || ""}|${data?.count || 1}`;
|
||||
}
|
||||
|
||||
function isDuplicateNewMailNotification(data, accountId) {
|
||||
const now = Date.now();
|
||||
for (const [key, seenAt] of state.realtimeNotifiedMessages) {
|
||||
if (now - seenAt > 10 * 60 * 1000) state.realtimeNotifiedMessages.delete(key);
|
||||
}
|
||||
const key = newMailNotificationKey(data, accountId);
|
||||
if (state.realtimeNotifiedMessages.has(key)) return true;
|
||||
state.realtimeNotifiedMessages.set(key, now);
|
||||
return false;
|
||||
}
|
||||
|
||||
function toastNewMail(data, accountId) {
|
||||
const count = Math.max(1, Number(data?.count) || 1);
|
||||
const folder = data?.folder || "INBOX";
|
||||
const account = state.accounts.find((item) => item.id === accountId);
|
||||
const accountPrefix = accountId && accountId !== state.accountId && account?.email ? `${account.email} · ` : "";
|
||||
const sender = data?.latest_sender ? `${data.latest_sender}: ` : "";
|
||||
const subject = data?.latest_subject || (count === 1 ? "New message" : `${count} new messages`);
|
||||
const message = count === 1
|
||||
? `${accountPrefix}${folder} · ${sender}${subject}`
|
||||
: `${accountPrefix}${folder} · ${count} new messages${data?.latest_subject ? ` · ${data.latest_subject}` : ""}`;
|
||||
toast(message, "info", 8000, { label: "Open", onClick: () => openRealtimeNotificationTarget(accountId, data) });
|
||||
}
|
||||
|
||||
async function openRealtimeNotificationTarget(accountId, data) {
|
||||
window.focus();
|
||||
const folder = data?.folder || "INBOX";
|
||||
@@ -185,7 +237,8 @@ function notifyNewMail(data, accountId) {
|
||||
const account = state.accounts.find((item) => item.id === accountId);
|
||||
const title = data.latest_sender || account?.email || "New mail";
|
||||
const body = data.latest_subject || `${data.count || 1} new message(s) in ${data.folder}`;
|
||||
const notification = new Notification(title, { body, tag: `rustbox-${accountId}-${data.folder}`, renotify: true });
|
||||
const notificationTag = `rustbox-${accountId}-${data.folder}-${Number.isFinite(uid) && uid > 0 ? uid : "new"}`;
|
||||
const notification = new Notification(title, { body, tag: notificationTag, renotify: false });
|
||||
notification.onclick = () => { notification.close(); void openRealtimeNotificationTarget(accountId, data); };
|
||||
}
|
||||
if (prefs.sound) playNotificationSound();
|
||||
@@ -270,13 +323,14 @@ function scheduleRealtimeRefresh(event, accountId) {
|
||||
clearTimeout(state.realtimeRefreshTimer);
|
||||
state.realtimeRefreshTimer = setTimeout(async () => {
|
||||
const events = state.realtimeRefreshEvents.splice(0);
|
||||
await refreshFolderListOnly();
|
||||
const refreshCurrent = events.some((item) => ["NewMessages", "FlagsChanged"].includes(item.type) && item.data?.folder === state.folder);
|
||||
const refreshCurrent = events.some((item) => item.type === "FolderUpdated" || (["NewMessages", "FlagsChanged"].includes(item.type) && item.data?.folder === state.folder));
|
||||
const tasks = [refreshFolderListOnly()];
|
||||
if (refreshCurrent) {
|
||||
if (state.searchContext) await runSearch(new URLSearchParams(state.searchContext.params), state.searchContext.label, { restoreScrollTop: state.searchContext.scrollTop });
|
||||
else await loadMessages(true);
|
||||
if (state.searchContext) tasks.push(runSearch(new URLSearchParams(state.searchContext.params), state.searchContext.label, { restoreScrollTop: state.searchContext.scrollTop }));
|
||||
else tasks.push(loadMessages(true));
|
||||
}
|
||||
}, 180);
|
||||
await Promise.all(tasks);
|
||||
}, 80);
|
||||
}
|
||||
|
||||
function startRealtimeFallbackPolling() {
|
||||
@@ -325,7 +379,9 @@ function connectRealtime() {
|
||||
state.wsConnectedOnce = true;
|
||||
state.wsRetryAttempt = 0;
|
||||
state.wsLastHeartbeatAt = Date.now();
|
||||
state.realtimeWatchedFolder = null;
|
||||
startRealtimeWatchdog(socket);
|
||||
syncRealtimeWatchedFolder(true);
|
||||
if (reconnect) void resyncRealtimeState();
|
||||
};
|
||||
socket.onmessage = (message) => {
|
||||
@@ -339,13 +395,24 @@ function connectRealtime() {
|
||||
}
|
||||
if (isDuplicateRealtimeEvent(payload)) return;
|
||||
const event = payload.event || {};
|
||||
if (event.type === "NewMessages") notifyNewMail(event.data || {}, payload.accountId);
|
||||
if (event.type === "NewMessages") {
|
||||
const data = event.data || {};
|
||||
if (!isDuplicateNewMailNotification(data, payload.accountId)) {
|
||||
toastNewMail(data, payload.accountId);
|
||||
notifyNewMail(data, payload.accountId);
|
||||
}
|
||||
}
|
||||
if (event.type === "IndexingStatus" && payload.accountId === state.accountId) {
|
||||
setIndexingIndicator(Boolean(event.data?.active), event.data?.pending);
|
||||
}
|
||||
if (["NewMessages", "FlagsChanged", "FolderUpdated"].includes(event.type)) scheduleRealtimeRefresh(event, payload.accountId);
|
||||
} catch (error) { console.warn("Invalid realtime event", error); }
|
||||
};
|
||||
socket.onclose = () => {
|
||||
if (state.ws !== socket) return;
|
||||
state.ws = null;
|
||||
state.realtimeWatchedFolder = null;
|
||||
setIndexingIndicator(false);
|
||||
clearInterval(state.wsWatchdogTimer);
|
||||
state.wsWatchdogTimer = null;
|
||||
scheduleRealtimeReconnect();
|
||||
@@ -366,6 +433,8 @@ function disconnectRealtime() {
|
||||
state.realtimeRefreshEvents = [];
|
||||
state.wsRetryAttempt = 0;
|
||||
state.wsLastHeartbeatAt = 0;
|
||||
state.realtimeWatchedFolder = null;
|
||||
setIndexingIndicator(false);
|
||||
const socket = state.ws;
|
||||
state.ws = null;
|
||||
if (socket && socket.readyState < WebSocket.CLOSING) socket.close(1000, "Client disconnect");
|
||||
|
||||
@@ -61,21 +61,44 @@ function bindPaneResizer(selector, preferenceKey, minPx, maxPx) {
|
||||
});
|
||||
}
|
||||
|
||||
function toolbarMenuPopover(menu) {
|
||||
return menu?._rustboxPopover || menu?.querySelector(".toolbar-menu-popover") || null;
|
||||
}
|
||||
|
||||
function mountToolbarMenuPopover(menu) {
|
||||
const popover = toolbarMenuPopover(menu);
|
||||
if (!popover) return null;
|
||||
menu._rustboxPopover = popover;
|
||||
if (popover.parentElement !== document.body) document.body.append(popover);
|
||||
popover.dataset.toolbarMenuOwner = menu.id || "toolbar-menu";
|
||||
return popover;
|
||||
}
|
||||
|
||||
function restoreToolbarMenuPopover(menu) {
|
||||
const popover = toolbarMenuPopover(menu);
|
||||
if (!popover || !menu?.isConnected || popover.parentElement !== document.body) return;
|
||||
popover.removeAttribute("data-toolbar-menu-owner");
|
||||
menu.append(popover);
|
||||
}
|
||||
|
||||
function positionToolbarMenu(menu) {
|
||||
if (!menu?.open) return;
|
||||
const popover = menu.querySelector(".toolbar-menu-popover");
|
||||
const popover = mountToolbarMenuPopover(menu);
|
||||
const summary = menu.querySelector("summary");
|
||||
if (!popover || !summary) return;
|
||||
Object.assign(popover.style, { right: "auto", bottom: "auto" });
|
||||
const rect = summary.getBoundingClientRect();
|
||||
const margin = 8;
|
||||
const mobile = matchMedia("(max-width: 560px)").matches;
|
||||
const width = Math.min(Math.max(mobile ? 180 : 164, popover.offsetWidth || 164), window.innerWidth - (margin * 2));
|
||||
const viewportWidth = window.visualViewport?.width || window.innerWidth;
|
||||
const viewportHeight = window.visualViewport?.height || window.innerHeight;
|
||||
popover.style.maxWidth = `${Math.max(120, viewportWidth - (margin * 2))}px`;
|
||||
const width = Math.min(Math.max(mobile ? 180 : 164, popover.offsetWidth || 164), viewportWidth - (margin * 2));
|
||||
const height = popover.offsetHeight || 80;
|
||||
let left = Math.min(rect.left, window.innerWidth - width - margin);
|
||||
let left = Math.min(rect.left, viewportWidth - width - margin);
|
||||
left = Math.max(margin, left);
|
||||
let top = rect.bottom + 5;
|
||||
if (top + height > window.innerHeight - margin) top = Math.max(margin, rect.top - height - 5);
|
||||
if (top + height > viewportHeight - margin) top = Math.max(margin, rect.top - height - 5);
|
||||
popover.style.left = `${Math.round(left)}px`;
|
||||
popover.style.top = `${Math.round(top)}px`;
|
||||
}
|
||||
@@ -83,12 +106,19 @@ function positionToolbarMenu(menu) {
|
||||
function bindToolbarMenus() {
|
||||
const menus = $$(".toolbar-menu");
|
||||
menus.forEach((menu) => menu.addEventListener("toggle", () => {
|
||||
if (!menu.open) return;
|
||||
if (!menu.open) {
|
||||
restoreToolbarMenuPopover(menu);
|
||||
return;
|
||||
}
|
||||
menus.forEach((other) => { if (other !== menu) other.open = false; });
|
||||
mountToolbarMenuPopover(menu);
|
||||
requestAnimationFrame(() => positionToolbarMenu(menu));
|
||||
}));
|
||||
window.addEventListener("resize", () => menus.forEach(positionToolbarMenu), { passive: true });
|
||||
document.addEventListener("scroll", () => menus.forEach(positionToolbarMenu), true);
|
||||
const reposition = () => menus.forEach(positionToolbarMenu);
|
||||
window.addEventListener("resize", reposition, { passive: true });
|
||||
window.visualViewport?.addEventListener("resize", reposition, { passive: true });
|
||||
window.visualViewport?.addEventListener("scroll", reposition, { passive: true });
|
||||
document.addEventListener("scroll", reposition, true);
|
||||
}
|
||||
|
||||
function bindEvents() {
|
||||
@@ -203,7 +233,20 @@ function bindEvents() {
|
||||
$("#bulk-tag-select").addEventListener("change", async (event) => { const tagId = event.target.value; event.target.value = ""; if (tagId) await bulkAddTag(tagId); });
|
||||
$("#select-mode-btn").addEventListener("click", () => setSelectionMode(!state.selectionMode));
|
||||
$("#selection-done").addEventListener("click", () => setSelectionMode(false));
|
||||
$("#refresh-btn").addEventListener("click", () => loadFolders(false));
|
||||
$("#refresh-btn").addEventListener("click", async () => {
|
||||
const button = $("#refresh-btn");
|
||||
setBusy(button, true);
|
||||
try {
|
||||
const folderRefresh = refreshFolderListOnly();
|
||||
if (state.searchContext) {
|
||||
await Promise.all([folderRefresh, runSearch(new URLSearchParams(state.searchContext.params), state.searchContext.label, { restoreScrollTop: state.searchContext.scrollTop })]);
|
||||
} else {
|
||||
await Promise.all([folderRefresh, loadMessages(true, { force: true })]);
|
||||
}
|
||||
} finally {
|
||||
setBusy(button, false);
|
||||
}
|
||||
});
|
||||
$("#message-sort").addEventListener("change", async (event) => {
|
||||
const value = event.target.value;
|
||||
state.mailPrefs.default_sort = value;
|
||||
@@ -445,7 +488,7 @@ function bindEvents() {
|
||||
bindEditor();
|
||||
document.addEventListener("click", (event) => {
|
||||
if (!event.target.closest("#context-menu")) hideContextMenu();
|
||||
if (!event.target.closest(".toolbar-menu")) $$(".toolbar-menu[open]").forEach((menu) => { menu.open = false; });
|
||||
if (!event.target.closest(".toolbar-menu, .toolbar-menu-popover[data-toolbar-menu-owner]")) $$(".toolbar-menu[open]").forEach((menu) => { menu.open = false; });
|
||||
});
|
||||
window.addEventListener("blur", hideContextMenu);
|
||||
document.addEventListener("scroll", hideContextMenu, true);
|
||||
|
||||
Reference in new issue
Block a user