v0.2.8

src/realtime/idle.rs

@@ -2,16 +2,25 @@ use std::sync::Arc;
 
 use dashmap::DashMap;
 use futures::StreamExt;
+use serde::Serialize;
 use tokio::task::JoinHandle;
 
 use crate::config::AppConfig;
 use crate::imap::client::ImapCredentials;
 use crate::imap::connection::ImapStream;
 use crate::imap::parse::{decode_rfc2047, flag_to_string, has_attachments, imap_address_to_email};
-use crate::realtime::events::{EventBus, MailEvent};
-use crate::search::engine::SearchEngine;
+use crate::realtime::events::{EventBus, MailEvent, MessageFlagChange};
+use crate::search::engine::{IndexableMessage, SearchEngine, UserIndex};
 use crate::{db, imap};
 
+#[derive(Debug, Clone, Serialize)]
+pub struct IdleTaskDiagnostics {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub user_hash: Option<String>,
+    pub folder: String,
+    pub subscribers: usize,
+}
+
 pub struct IdleManager {
     tasks: DashMap<(String, String), IdleTask>,
 }
@@ -90,6 +99,21 @@ impl IdleManager {
     pub fn subscriber_count(&self) -> usize {
         self.tasks.iter().map(|task| task.value().subscribers).sum()
     }
+
+    pub fn diagnostics_snapshot(&self, include_user_identifiers: bool) -> Vec<IdleTaskDiagnostics> {
+        let mut tasks = self
+            .tasks
+            .iter()
+            .map(|task| IdleTaskDiagnostics {
+                user_hash: include_user_identifiers.then(|| task.key().0.clone()),
+                folder: task.key().1.clone(),
+                subscribers: task.value().subscribers,
+            })
+            .collect::<Vec<_>>();
+        tasks.sort_by(|left, right| left.folder.cmp(&right.folder));
+        tasks.truncate(200);
+        tasks
+    }
 }
 
 impl Default for IdleManager {
@@ -201,12 +225,28 @@ async fn run_idle_session(
                                     },
                                 )
                                 .await;
-                        } else if result.flags_updated > 0 || result.deleted_count > 0 {
+                        }
+                        if result.flags_updated > 0 || result.deleted_count > 0 {
                             event_bus
                                 .publish(
                                     user_hash,
                                     MailEvent::FlagsChanged {
                                         folder: folder.to_string(),
+                                        changes: result.flag_changes,
+                                    },
+                                )
+                                .await;
+                        }
+                        if let Ok(conn) = db::pool::open_user_db(&config.data_dir, user_hash)
+                            && let Ok(Some(folder_state)) = db::folders::get_folder(&conn, folder)
+                        {
+                            event_bus
+                                .publish(
+                                    user_hash,
+                                    MailEvent::FolderStatus {
+                                        folder: folder.to_string(),
+                                        total_count: folder_state.total_count,
+                                        unread_count: folder_state.unread_count,
                                     },
                                 )
                                 .await;
@@ -249,6 +289,7 @@ struct ReconcileResult {
     latest_uid: Option<u32>,
     latest_sender: Option<String>,
     latest_subject: Option<String>,
+    flag_changes: Vec<MessageFlagChange>,
 }
 
 impl ReconcileResult {
@@ -260,6 +301,7 @@ impl ReconcileResult {
             latest_uid: None,
             latest_sender: None,
             latest_subject: None,
+            flag_changes: Vec::new(),
         }
     }
 }
@@ -271,28 +313,42 @@ async fn fetch_new_after_idle(
     config: &AppConfig,
     search_engine: &SearchEngine,
 ) -> (async_imap::Session<ImapStream>, ReconcileResult) {
-    let (cached_modseq, max_cached_uid, cached_count, cached_uid_validity, header_sync) =
-        match db::pool::open_user_db(&config.data_dir, user_hash) {
-            Ok(conn) => {
-                let header_sync = match db::header_sync::HeaderSync::capture(&conn) {
-                    Ok(sync) => sync,
-                    Err(e) => {
-                        tracing::warn!(error = %e, "IDLE fetch: failed to read header generation");
-                        return (session, ReconcileResult::empty());
-                    }
-                };
-                let cached = db::folders::get_folder(&conn, folder).ok().flatten();
-                let modseq = cached.as_ref().map(|f| f.highest_modseq).unwrap_or(0);
-                let validity = cached.as_ref().map(|f| f.uid_validity).unwrap_or(0);
-                let max_uid = db::messages::max_uid(&conn, folder).unwrap_or(0);
-                let count = db::messages::count_messages(&conn, folder).unwrap_or(0);
-                (modseq, max_uid, count, validity, header_sync)
-            }
-            Err(e) => {
-                tracing::warn!(error = %e, "IDLE fetch: failed to open DB");
-                return (session, ReconcileResult::empty());
-            }
-        };
+    let (
+        cached_modseq,
+        max_cached_uid,
+        cached_count,
+        cached_counter_generation,
+        cached_uid_validity,
+        header_sync,
+    ) = match db::pool::open_user_db(&config.data_dir, user_hash) {
+        Ok(conn) => {
+            let header_sync = match db::header_sync::HeaderSync::capture(&conn) {
+                Ok(sync) => sync,
+                Err(e) => {
+                    tracing::warn!(error = %e, "IDLE fetch: failed to read header generation");
+                    return (session, ReconcileResult::empty());
+                }
+            };
+            let cached = db::folders::get_folder(&conn, folder).ok().flatten();
+            let modseq = cached.as_ref().map(|f| f.highest_modseq).unwrap_or(0);
+            let validity = cached.as_ref().map(|f| f.uid_validity).unwrap_or(0);
+            let counter_generation = cached.as_ref().map(|f| f.counter_generation).unwrap_or(0);
+            let max_uid = db::messages::max_uid(&conn, folder).unwrap_or(0);
+            let count = db::messages::count_messages(&conn, folder).unwrap_or(0);
+            (
+                modseq,
+                max_uid,
+                count,
+                counter_generation,
+                validity,
+                header_sync,
+            )
+        }
+        Err(e) => {
+            tracing::warn!(error = %e, "IDLE fetch: failed to open DB");
+            return (session, ReconcileResult::empty());
+        }
+    };
 
     let mailbox = match session.select_condstore(folder).await {
         Ok(mb) => mb,
@@ -303,7 +359,13 @@ async fn fetch_new_after_idle(
     };
 
     let server_exists = mailbox.exists;
-    let server_unseen = mailbox.unseen;
+    let server_unseen = match session.uid_search("UNSEEN").await {
+        Ok(uids) => Some(uids.len() as u32),
+        Err(error) => {
+            tracing::debug!(folder = %folder, error = %error, "IDLE UNSEEN search failed");
+            None
+        }
+    };
     let server_modseq = mailbox.highest_modseq.unwrap_or(0);
     let uid_validity = mailbox.uid_validity.unwrap_or(0);
 
@@ -314,6 +376,18 @@ async fn fetch_new_after_idle(
         return (session, ReconcileResult::empty());
     }
 
+    if let Some(unseen) = server_unseen
+        && let Ok(conn) = db::pool::open_user_db(&config.data_dir, user_hash)
+    {
+        let _ = db::folders::set_folder_counts_if_unchanged(
+            &conn,
+            folder,
+            cached_counter_generation,
+            server_exists,
+            unseen,
+        );
+    }
+
     if server_modseq > 0 && server_modseq == cached_modseq && server_exists == cached_count {
         tracing::debug!(
             user_hash = %user_hash,
@@ -331,6 +405,7 @@ async fn fetch_new_after_idle(
                 Ok(stream) => {
                     let mut stream = std::pin::pin!(stream);
                     let mut items = Vec::new();
+                    let mut complete = true;
                     while let Some(result) = stream.next().await {
                         match result {
                             Ok(fetch) => {
@@ -345,11 +420,12 @@ async fn fetch_new_after_idle(
                                     error = %e,
                                     "IDLE fetch: error reading CHANGEDSINCE stream"
                                 );
+                                complete = false;
                                 break;
                             }
                         }
                     }
-                    Some(items)
+                    complete.then_some(items)
                 }
                 Err(e) => {
                     tracing::warn!(
@@ -378,18 +454,42 @@ async fn fetch_new_after_idle(
                 let sync_keywords = db::mail_preferences::get_preferences(&conn)
                     .map(|prefs| prefs.sync_imap_keywords)
                     .unwrap_or(false);
-                for (uid, flags) in &flag_changes {
-                    let mut sorted = flags.clone();
-                    sorted.sort();
-                    let flags_csv = sorted.join(",");
-                    if db::messages::update_message_flags(&conn, folder, *uid, &flags_csv).is_ok() {
-                        if sync_keywords
-                            && let Err(error) =
-                                db::tags::import_imap_keywords(&conn, *uid, folder, &sorted)
-                        {
-                            tracing::debug!(uid = uid, error = %error, "IDLE keyword import failed");
+                match header_sync.begin_write(&conn) {
+                    Ok(transaction) => {
+                        for (uid, flags) in &flag_changes {
+                            let mut sorted = flags.clone();
+                            sorted.sort();
+                            let flags_csv = sorted.join(",");
+                            if db::messages::update_message_flags(
+                                &transaction,
+                                folder,
+                                *uid,
+                                &flags_csv,
+                            )
+                            .is_ok()
+                            {
+                                if sync_keywords
+                                    && let Err(error) = db::tags::import_imap_keywords(
+                                        &transaction,
+                                        *uid,
+                                        folder,
+                                        &sorted,
+                                    )
+                                {
+                                    tracing::debug!(uid = uid, error = %error, "IDLE keyword import failed");
+                                }
+                                flags_updated += 1;
+                            }
                         }
-                        flags_updated += 1;
+                        if let Err(error) = transaction.commit() {
+                            tracing::debug!(error = %error, "IDLE flag reconciliation commit failed");
+                            flags_updated = 0;
+                            flag_changes.clear();
+                        }
+                    }
+                    Err(error) => {
+                        tracing::debug!(error = %error, "IDLE flag reconciliation discarded stale cache generation");
+                        flag_changes.clear();
                     }
                 }
             }
@@ -399,12 +499,15 @@ async fn fetch_new_after_idle(
                 let uid_range = format!("{}:*", max_cached_uid + 1);
                 new_count = fetch_and_store_headers(
                     &mut session,
-                    user_hash,
-                    folder,
-                    config,
+                    HeaderFetchContext {
+                        user_hash,
+                        folder,
+                        config,
+                        search_engine,
+                        header_sync,
+                    },
                     &uid_range,
                     max_cached_uid,
-                    header_sync,
                 )
                 .await;
             }
@@ -430,22 +533,16 @@ async fn fetch_new_after_idle(
             if let Ok(conn) = db::pool::open_user_db(&config.data_dir, user_hash)
                 && let Ok(transaction) = header_sync.begin_write(&conn)
             {
-                let _ = db::folders::update_folder_sync_status(
-                    &conn,
+                let _ = db::folders::update_folder_sync_cursor(
+                    &transaction,
                     folder,
                     uid_validity,
-                    server_exists,
                     server_modseq,
                 );
-                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);
+                    let _ = db::folders::invalidate_folder_freshness(&transaction, folder);
                 } else {
-                    let _ = db::folders::mark_folder_messages_fresh(&conn, folder);
+                    let _ = db::folders::mark_folder_messages_fresh(&transaction, folder);
                 }
                 let _ = transaction.commit();
             }
@@ -470,53 +567,136 @@ async fn fetch_new_after_idle(
                     latest_uid,
                     latest_sender,
                     latest_subject,
+                    flag_changes: flag_changes
+                        .into_iter()
+                        .map(|(uid, mut flags)| {
+                            flags.sort();
+                            MessageFlagChange { uid, flags }
+                        })
+                        .collect(),
                 },
             );
         }
     }
 
-    let cached_uids: std::collections::HashSet<u32> =
-        match db::pool::open_user_db(&config.data_dir, user_hash) {
-            Ok(conn) => {
-                let uids_flags =
-                    db::messages::get_all_uids_and_flags(&conn, folder).unwrap_or_default();
-                uids_flags.iter().map(|(uid, _)| *uid).collect()
-            }
-            Err(e) => {
-                tracing::warn!(error = %e, "IDLE fetch: failed to open DB for fallback");
-                return (session, ReconcileResult::empty());
-            }
-        };
+    let cached_state = match db::pool::open_user_db(&config.data_dir, user_hash) {
+        Ok(conn) => db::messages::get_all_uids_and_flags(&conn, folder).unwrap_or_default(),
+        Err(e) => {
+            tracing::warn!(error = %e, "IDLE fetch: failed to open DB for fallback");
+            return (session, ReconcileResult::empty());
+        }
+    };
+    let cached_uids = cached_state
+        .iter()
+        .map(|(uid, _)| *uid)
+        .collect::<std::collections::HashSet<_>>();
+    let cached_flags = cached_state
+        .into_iter()
+        .map(|(uid, flags)| {
+            let mut normalized = flags
+                .split(',')
+                .filter(|flag| !flag.is_empty())
+                .map(str::to_string)
+                .collect::<Vec<_>>();
+            normalized.sort();
+            normalized.dedup();
+            (uid, normalized)
+        })
+        .collect::<std::collections::HashMap<_, _>>();
 
-    let imap_uids: Option<std::collections::HashSet<u32>> =
-        match session.uid_fetch("1:*", "UID").await {
-            Ok(stream) => {
-                let mut stream = std::pin::pin!(stream);
-                let mut uids = std::collections::HashSet::new();
-                while let Some(result) = stream.next().await {
-                    match result {
-                        Ok(fetch) => {
-                            if let Some(uid) = fetch.uid {
-                                uids.insert(uid);
-                            }
-                        }
-                        Err(e) => {
-                            tracing::warn!(error = %e, "IDLE fetch: error reading UID stream");
-                            break;
+    let imap_state: Option<std::collections::HashMap<u32, Vec<String>>> = match session
+        .uid_fetch("1:*", "UID FLAGS")
+        .await
+    {
+        Ok(stream) => {
+            let mut stream = std::pin::pin!(stream);
+            let mut state = std::collections::HashMap::new();
+            let mut complete = true;
+            while let Some(result) = stream.next().await {
+                match result {
+                    Ok(fetch) => {
+                        if let Some(uid) = fetch.uid {
+                            let mut flags = fetch
+                                .flags()
+                                .map(|flag| flag_to_string(&flag))
+                                .collect::<Vec<_>>();
+                            flags.sort();
+                            flags.dedup();
+                            state.insert(uid, flags);
                         }
                     }
+                    Err(e) => {
+                        tracing::warn!(error = %e, "IDLE fetch: error reading UID/FLAGS stream");
+                        complete = false;
+                        break;
+                    }
                 }
-                Some(uids)
             }
-            Err(e) => {
-                tracing::warn!(error = %e, "IDLE fetch: uid_fetch(UIDs) failed");
-                None
-            }
-        };
+            complete.then_some(state)
+        }
+        Err(e) => {
+            tracing::warn!(error = %e, "IDLE fetch: uid_fetch(UID FLAGS) failed");
+            None
+        }
+    };
 
-    let Some(imap_uids) = imap_uids else {
+    let Some(imap_state) = imap_state else {
         return (session, ReconcileResult::empty());
     };
+    let imap_uids = imap_state
+        .keys()
+        .copied()
+        .collect::<std::collections::HashSet<_>>();
+
+    let mut flag_changes = cached_flags
+        .iter()
+        .filter_map(|(uid, cached)| {
+            imap_state
+                .get(uid)
+                .filter(|server| *server != cached)
+                .map(|server| (*uid, server.clone()))
+        })
+        .collect::<Vec<_>>();
+    flag_changes.sort_by_key(|(uid, _)| *uid);
+    let mut flags_updated = 0u32;
+    if !flag_changes.is_empty()
+        && let Ok(conn) = db::pool::open_user_db(&config.data_dir, user_hash)
+    {
+        let sync_keywords = db::mail_preferences::get_preferences(&conn)
+            .map(|prefs| prefs.sync_imap_keywords)
+            .unwrap_or(false);
+        match header_sync.begin_write(&conn) {
+            Ok(transaction) => {
+                for (uid, flags) in &flag_changes {
+                    if db::messages::update_message_flags(
+                        &transaction,
+                        folder,
+                        *uid,
+                        &flags.join(","),
+                    )
+                    .is_ok()
+                    {
+                        if sync_keywords
+                            && let Err(error) =
+                                db::tags::import_imap_keywords(&transaction, *uid, folder, flags)
+                        {
+                            tracing::debug!(uid = uid, error = %error, "IDLE fallback keyword import failed");
+                        }
+                        flags_updated += 1;
+                    }
+                }
+                if let Err(error) = transaction.commit() {
+                    tracing::debug!(error = %error, "IDLE fallback flag reconciliation commit failed");
+                    flags_updated = 0;
+                    flag_changes.clear();
+                }
+            }
+            Err(error) => {
+                tracing::debug!(error = %error, "IDLE fallback flag reconciliation discarded stale cache generation");
+                flag_changes.clear();
+            }
+        }
+    }
 
     let has_new = imap_uids.iter().any(|&uid| uid > max_cached_uid);
     let mut new_count = 0u32;
@@ -524,12 +704,15 @@ async fn fetch_new_after_idle(
         let uid_range = format!("{}:*", max_cached_uid + 1);
         new_count = fetch_and_store_headers(
             &mut session,
-            user_hash,
-            folder,
-            config,
+            HeaderFetchContext {
+                user_hash,
+                folder,
+                config,
+                search_engine,
+                header_sync,
+            },
             &uid_range,
             max_cached_uid,
-            header_sync,
         )
         .await;
     }
@@ -553,22 +736,16 @@ async fn fetch_new_after_idle(
     if let Ok(conn) = db::pool::open_user_db(&config.data_dir, user_hash)
         && let Ok(transaction) = header_sync.begin_write(&conn)
     {
-        let _ = db::folders::update_folder_sync_status(
-            &conn,
+        let _ = db::folders::update_folder_sync_cursor(
+            &transaction,
             folder,
             uid_validity,
-            server_exists,
             server_modseq,
         );
-        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);
+            let _ = db::folders::invalidate_folder_freshness(&transaction, folder);
         } else {
-            let _ = db::folders::mark_folder_messages_fresh(&conn, folder);
+            let _ = db::folders::mark_folder_messages_fresh(&transaction, folder);
         }
         let _ = transaction.commit();
     }
@@ -579,10 +756,11 @@ async fn fetch_new_after_idle(
         (None, None, None)
     };
 
-    if new_count > 0 || deleted_count > 0 {
+    if flags_updated > 0 || new_count > 0 || deleted_count > 0 {
         tracing::info!(
             user_hash = %user_hash,
             folder = %folder,
+            flags_updated = flags_updated,
             new = new_count,
             deleted = deleted_count,
             "Reconciled cache after IDLE (full scan)"
@@ -593,11 +771,15 @@ async fn fetch_new_after_idle(
         session,
         ReconcileResult {
             new_count,
-            flags_updated: 0,
+            flags_updated,
             deleted_count,
             latest_uid,
             latest_sender,
             latest_subject,
+            flag_changes: flag_changes
+                .into_iter()
+                .map(|(uid, flags)| MessageFlagChange { uid, flags })
+                .collect(),
         },
     )
 }
@@ -634,15 +816,27 @@ fn lookup_latest_message(
     }
 }
 
+struct HeaderFetchContext<'a> {
+    user_hash: &'a str,
+    folder: &'a str,
+    config: &'a AppConfig,
+    search_engine: &'a SearchEngine,
+    header_sync: db::header_sync::HeaderSync,
+}
+
 async fn fetch_and_store_headers(
     session: &mut async_imap::Session<ImapStream>,
-    user_hash: &str,
-    folder: &str,
-    config: &AppConfig,
+    context: HeaderFetchContext<'_>,
     uid_range: &str,
     min_uid: u32,
-    header_sync: db::header_sync::HeaderSync,
 ) -> u32 {
+    let HeaderFetchContext {
+        user_hash,
+        folder,
+        config,
+        search_engine,
+        header_sync,
+    } = context;
     let fetches: Option<Vec<_>> = match session
         .uid_fetch(
             uid_range,
@@ -653,16 +847,18 @@ async fn fetch_and_store_headers(
         Ok(stream) => {
             let mut stream = std::pin::pin!(stream);
             let mut items = Vec::new();
+            let mut complete = true;
             while let Some(result) = stream.next().await {
                 match result {
                     Ok(fetch) => items.push(fetch),
                     Err(e) => {
                         tracing::warn!(error = %e, "IDLE fetch: error reading header stream");
+                        complete = false;
                         break;
                     }
                 }
             }
-            Some(items)
+            complete.then_some(items)
         }
         Err(e) => {
             tracing::warn!(error = %e, "IDLE fetch: uid_fetch(headers) failed");
@@ -684,6 +880,11 @@ async fn fetch_and_store_headers(
         let sync_keywords = db::mail_preferences::get_preferences(&conn)
             .map(|prefs| prefs.sync_imap_keywords)
             .unwrap_or(false);
+        let user_index = if search_engine.is_enabled() && !UserIndex::is_excluded_folder(folder) {
+            search_engine.open_user_index(user_hash).ok()
+        } else {
+            None
+        };
         for fetch in &fetches {
             let uid = match fetch.uid {
                 Some(u) if u > min_uid => u,
@@ -787,7 +988,7 @@ async fn fetch_and_store_headers(
             let flags_csv = flags.join(",");
 
             if let Err(e) = db::messages::upsert_message(
-                &conn,
+                &transaction,
                 folder,
                 uid,
                 message_id.as_deref(),
@@ -808,7 +1009,7 @@ async fn fetch_and_store_headers(
                 tracing::warn!(uid = uid, error = %e, "IDLE fetch: failed to upsert message");
             } else {
                 if let Err(e) = db::contacts::populate_known_addresses(
-                    &conn,
+                    &transaction,
                     from_address,
                     from_name,
                     &to_json,
@@ -817,10 +1018,30 @@ async fn fetch_and_store_headers(
                     tracing::warn!(uid = uid, error = %e, "IDLE fetch: failed to populate known addresses");
                 }
                 if sync_keywords
-                    && let Err(error) = db::tags::import_imap_keywords(&conn, uid, folder, &flags)
+                    && let Err(error) =
+                        db::tags::import_imap_keywords(&transaction, uid, folder, &flags)
                 {
                     tracing::debug!(uid = uid, error = %error, "IDLE keyword import failed");
                 }
+                if let Some(index) = user_index.as_ref()
+                    && let Err(error) = index.index_message_with_connection(
+                        &transaction,
+                        &IndexableMessage {
+                            uid,
+                            folder: folder.to_string(),
+                            subject: subject_str.to_string(),
+                            from_address: from_address.to_string(),
+                            from_name: from_name.to_string(),
+                            to_addresses: to_json.clone(),
+                            body_text: String::new(),
+                            date_epoch: db::messages::parse_date_to_epoch_public(date_str),
+                            has_attachments: has_attach,
+                        },
+                    )
+                {
+                    tracing::warn!(uid = uid, error = %error, "IDLE fetch: failed to enqueue search indexing");
+                    return 0;
+                }
                 new_count += 1;
             }
         }
@@ -888,6 +1109,7 @@ async fn fetch_uids_and_remove_deleted(
             Ok(stream) => {
                 let mut stream = std::pin::pin!(stream);
                 let mut uids = std::collections::HashSet::new();
+                let mut complete = true;
                 while let Some(result) = stream.next().await {
                     match result {
                         Ok(fetch) => {
@@ -900,11 +1122,12 @@ async fn fetch_uids_and_remove_deleted(
                                 error = %e,
                                 "IDLE fetch: error reading UID stream for deletion check"
                             );
+                            complete = false;
                             break;
                         }
                     }
                 }
-                Some(uids)
+                complete.then_some(uids)
             }
             Err(e) => {
                 tracing::warn!(