feat: queue notifications for mail missed while the app was closed

crates/flectar-mail-core/src/db/repo/accounts.rs

@@ -565,6 +565,15 @@ pub fn set_caught_up(conn: &Connection, id: i64, started: i64) -> Result<()> {
     Ok(())
 }
 
+/// When each account's last complete sync started, for the accounts that
+/// have had one.
+pub fn caught_up(conn: &Connection) -> Result<std::collections::BTreeMap<i64, i64>> {
+    let mut statement =
+        conn.prepare("SELECT id, caught_up_at FROM accounts WHERE caught_up_at IS NOT NULL")?;
+    let rows = statement.query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?;
+    Ok(rows.collect::<rusqlite::Result<_>>()?)
+}
+
 /// Move one account before or after another and compact the persisted order.
 /// A complete rewrite keeps ordering deterministic after any number of moves
 /// and avoids fragile fractional positions.

crates/flectar-mail-core/src/db/repo/actions.rs

@@ -183,6 +183,25 @@ pub fn due_calendar(
     Ok(rows)
 }
 
+/// How many of these actions are still to be tried, or being tried: waiting
+/// without an error from an earlier try, or in flight. One that is gone
+/// counts as tried.
+pub fn untried(conn: &Connection, ids: &[i64]) -> Result<usize> {
+    let mut statement = conn.prepare_cached(
+        "SELECT EXISTS (SELECT 1 FROM pending_actions
+                        WHERE id = ?1
+                          AND (state = 'inflight'
+                               OR (state = 'pending' AND last_error IS NULL)))",
+    )?;
+    let mut untried = 0;
+    for id in ids {
+        if statement.query_row(params![id], |row| row.get::<_, bool>(0))? {
+            untried += 1;
+        }
+    }
+    Ok(untried)
+}
+
 pub fn next_due_at(conn: &Connection) -> Result<Option<i64>> {
     Ok(conn
         .query_row(
@@ -438,6 +457,38 @@ mod tests {
     use super::*;
     use crate::db::testutil;
 
+    #[test]
+    fn an_action_counts_as_tried_once_it_ended_or_failed_once() {
+        let conn = testutil::conn();
+        testutil::seed_account(&conn);
+        let action = || {
+            enqueue(
+                &conn,
+                1,
+                "archive",
+                None,
+                None,
+                &serde_json::json!({}),
+                None,
+            )
+            .unwrap()
+        };
+        let (waiting, inflight, done, failed, retrying) =
+            (action(), action(), action(), action(), action());
+        conn.execute_batch(&format!(
+            "UPDATE pending_actions SET state = 'inflight' WHERE id = {inflight};
+             UPDATE pending_actions SET state = 'done' WHERE id = {done};
+             UPDATE pending_actions SET state = 'failed', last_error = 'refused' WHERE id = {failed};
+             UPDATE pending_actions SET last_error = 'offline', not_before = 9 WHERE id = {retrying};"
+        ))
+        .unwrap();
+
+        assert_eq!(untried(&conn, &[waiting, inflight]).unwrap(), 2);
+        assert_eq!(untried(&conn, &[done, failed, retrying]).unwrap(), 0);
+        // One that is gone was tried, by whoever removed it.
+        assert_eq!(untried(&conn, &[999]).unwrap(), 0);
+    }
+
     #[test]
     fn due_is_stable_and_has_due_ignores_future_and_calendar_actions() {
         let conn = testutil::conn();

crates/flectar-mail-core/src/db/repo/notifications.rs

@@ -93,6 +93,66 @@ pub fn enqueue_at(
     .ok_or_else(|| CoreError::NotFound(format!("message {message_id}")))
 }
 
+/// Queue notifications for mail that reached an Inbox while no copy of the
+/// app was following its account, so that a check in the background tells of
+/// it as a running app would have. `since` holds, per account, when its last
+/// complete sync before this start began (see
+/// [`super::accounts::caught_up`]). Mail that came in after that, by the
+/// measure a running sync uses ([`crate::sync::ingest::live_since`]), and is
+/// incoming, recent ([`CHIME_RECENT_MS`]) and still new ([`STILL_NEW`]) is
+/// queued, once per conversation as a sync queues it: its newest message,
+/// unless the conversation has a notification for one as new or newer.
+/// Returns the accounts and conversations it queued for.
+pub fn enqueue_missed(
+    conn: &Connection,
+    since: &BTreeMap<i64, i64>,
+    now: i64,
+) -> Result<Vec<(i64, i64)>> {
+    let mut statement = conn.prepare(&format!(
+        "SELECT m.id, m.thread_id, m.is_automated, COALESCE(m.from_addr, '')
+         FROM messages m JOIN accounts a ON a.id = m.account_id
+         WHERE m.account_id = ?1
+           AND m.is_outgoing = 0
+           AND COALESCE(m.internal_date, m.date) >= ?3
+           AND COALESCE(m.internal_date, m.date) > ?4
+           AND {}
+           AND NOT EXISTS (
+                 SELECT 1 FROM notification_outbox o JOIN messages n ON n.id = o.message_id
+                 WHERE o.thread_id = m.thread_id
+                   AND COALESCE(n.internal_date, n.date) >= COALESCE(m.internal_date, m.date))
+         ORDER BY COALESCE(m.internal_date, m.date) DESC, m.id DESC",
+        still_new_condition(2)
+    ))?;
+    let mut queued = Vec::new();
+    for (&account_id, &caught_up_at) in since {
+        // Newest first, so the first message of a conversation is its newest.
+        let mut newest = Vec::new();
+        {
+            let mut threads = std::collections::HashSet::new();
+            let mut rows = statement.query(params![
+                account_id,
+                roles::INBOX,
+                crate::sync::ingest::live_since(caught_up_at),
+                now - CHIME_RECENT_MS,
+            ])?;
+            while let Some(row) = rows.next()? {
+                let thread_id: i64 = row.get(1)?;
+                if threads.insert(thread_id) {
+                    let from: String = row.get(3)?;
+                    let audible = crate::sync::ingest::chime_worthy(row.get(2)?, &from);
+                    newest.push((row.get::<_, i64>(0)?, thread_id, audible));
+                }
+            }
+        }
+        // Oldest first, the order a sync queues them in.
+        for (message_id, thread_id, audible) in newest.into_iter().rev() {
+            enqueue_at(conn, message_id, audible, now)?;
+            queued.push((account_id, thread_id));
+        }
+    }
+    Ok(queued)
+}
+
 /// How a thread currently routes, for notification-scope filtering.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum RoutedTab {
@@ -1182,4 +1242,152 @@ mod tests {
         assert!(scope_allows(&settings, "label:7"));
         assert!(!scope_allows(&settings, "split:1"));
     }
+
+    /// Unread incoming mail in the Inbox that the server received at `at`,
+    /// in a conversation of its own unless `thread` is given.
+    fn arrived(c: &Connection, from: &str, at: i64, thread: Option<i64>) -> (i64, i64) {
+        let thread_id = thread.unwrap_or_else(|| {
+            c.execute(
+                "INSERT INTO threads (account_id, subject_norm, unread_count, last_message_at)
+                 VALUES (1, 'news', 1, ?1)",
+                params![at],
+            )
+            .unwrap();
+            c.last_insert_rowid()
+        });
+        c.execute(
+            "INSERT INTO messages (thread_id, account_id, folder_id, subject, from_addr, date,
+             internal_date, is_read, is_automated, is_draft, is_outgoing)
+             VALUES (?1, 1, 1, 'News', ?2, ?3, ?3, 0, 0, 0, 0)",
+            params![thread_id, from, at],
+        )
+        .unwrap();
+        (thread_id, c.last_insert_rowid())
+    }
+
+    fn queued(c: &Connection) -> Vec<(i64, bool)> {
+        c.prepare("SELECT notified_message_id, audible FROM notification_outbox ORDER BY id")
+            .unwrap()
+            .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))
+            .unwrap()
+            .collect::<rusqlite::Result<_>>()
+            .unwrap()
+    }
+
+    #[test]
+    fn mail_that_came_while_nobody_followed_is_queued_once_per_conversation() {
+        let c = testutil::conn();
+        testutil::seed_account(&c);
+        c.execute(
+            "INSERT INTO folders (id, account_id, path, role) VALUES (2, 1, 'Archive', 'archive')",
+            [],
+        )
+        .unwrap();
+        let now = 10 * CHIME_RECENT_MS;
+        let minute = 60_000;
+        let caught_up_at = now - 60 * minute;
+        let since = BTreeMap::from([(1, caught_up_at)]);
+
+        // Stored by the sync that caught up, which said nothing about it.
+        arrived(&c, "old@test.dev", caught_up_at - 2 * minute, None);
+        // Within the minute a server's clock may lag, as a sync counts it.
+        let (_, skewed) = arrived(&c, "skew@test.dev", caught_up_at - minute / 2, None);
+        let (_, person) = arrived(&c, "alice@test.dev", caught_up_at + 10 * minute, None);
+        let (_, robot) = arrived(&c, "noreply@test.dev", caught_up_at + 11 * minute, None);
+        // Two in one conversation: only the newest.
+        let (thread, first) = arrived(&c, "bob@test.dev", caught_up_at + 12 * minute, None);
+        let (_, second) = arrived(
+            &c,
+            "carol@test.dev",
+            caught_up_at + 13 * minute,
+            Some(thread),
+        );
+        // Read, sent, moved out of the Inbox, or already told of.
+        let (_, read) = arrived(&c, "dave@test.dev", caught_up_at + 14 * minute, None);
+        let (_, sent) = arrived(&c, "me@test.dev", caught_up_at + 15 * minute, None);
+        let (_, archived) = arrived(&c, "erin@test.dev", caught_up_at + 16 * minute, None);
+        let (_, told) = arrived(&c, "frank@test.dev", caught_up_at + 17 * minute, None);
+        c.execute_batch(&format!(
+            "UPDATE messages SET is_read = 1 WHERE id = {read};
+             UPDATE messages SET is_outgoing = 1 WHERE id = {sent};
+             UPDATE messages SET folder_id = 2 WHERE id = {archived};"
+        ))
+        .unwrap();
+        enqueue_at(&c, told, true, caught_up_at + 17 * minute).unwrap();
+        // A conversation told of while the app ran gets its newer message.
+        let (earlier_thread, earlier) =
+            arrived(&c, "gina@test.dev", caught_up_at - 30 * minute, None);
+        let (_, followup) = arrived(
+            &c,
+            "gina@test.dev",
+            caught_up_at + 18 * minute,
+            Some(earlier_thread),
+        );
+        enqueue_at(&c, earlier, true, caught_up_at - 30 * minute).unwrap();
+
+        let threads = enqueue_missed(&c, &since, now).unwrap();
+        assert_eq!(threads.len(), 5, "{threads:?}");
+        assert!(threads.contains(&(1, thread)));
+        assert!(threads.contains(&(1, earlier_thread)));
+        assert_eq!(
+            queued(&c),
+            [
+                (told, true),
+                (earlier, true),
+                (skewed, true),
+                (person, true),
+                (robot, false),
+                (second, true),
+                (followup, true),
+            ]
+        );
+        assert!(!queued(&c).iter().any(|&(id, _)| id == first));
+
+        // Queued once: a second look finds nothing more.
+        assert!(enqueue_missed(&c, &since, now).unwrap().is_empty());
+        // Nor for an account that was never caught up.
+        assert!(
+            enqueue_missed(&c, &BTreeMap::new(), now)
+                .unwrap()
+                .is_empty()
+        );
+    }
+
+    #[test]
+    fn missed_mail_stays_within_the_day_a_sync_tells_of() {
+        let c = testutil::conn();
+        testutil::seed_account(&c);
+        let now = 10 * CHIME_RECENT_MS;
+        let hour = 60 * 60_000;
+        // Last followed three days ago.
+        let since = BTreeMap::from([(1, now - 72 * hour)]);
+        arrived(&c, "old@test.dev", now - 25 * hour, None);
+        let (_, recent) = arrived(&c, "new@test.dev", now - 2 * hour, None);
+
+        enqueue_missed(&c, &since, now).unwrap();
+        assert_eq!(queued(&c), [(recent, true)]);
+    }
+
+    #[test]
+    fn missed_gmail_mail_counts_while_it_carries_the_inbox_label() {
+        let c = testutil::conn();
+        testutil::seed_account(&c);
+        c.execute_batch(
+            "UPDATE accounts SET provider = 'gmail', mail_protocol = 'gmail', auth_kind = 'oauth2';
+             INSERT INTO folders (id, account_id, path, role) VALUES (2, 1, '[Gmail]/All Mail', 'all');",
+        )
+        .unwrap();
+        let now = 10 * CHIME_RECENT_MS;
+        let since = BTreeMap::from([(1, now - 60 * 60_000)]);
+        let (_, labelled) = arrived(&c, "a@test.dev", now - 30 * 60_000, None);
+        let (_, archived) = arrived(&c, "b@test.dev", now - 20 * 60_000, None);
+        c.execute_batch(&format!(
+            "UPDATE messages SET folder_id = 2 WHERE id IN ({labelled}, {archived});
+             INSERT INTO message_folders (message_id, folder_id) VALUES ({labelled}, 1);"
+        ))
+        .unwrap();
+
+        enqueue_missed(&c, &since, now).unwrap();
+        assert_eq!(queued(&c), [(labelled, true)]);
+    }
 }

crates/flectar-mail-core/src/lib.rs

@@ -827,6 +827,13 @@ pub struct Core {
     ai_router_lock: Arc<tokio::sync::Mutex<()>>,
     /// Wakes the loop that copies and moves mail to other accounts.
     transfers: Arc<tokio::sync::Notify>,
+    /// Set once the startup has started the account workers.
+    accounts_started: Arc<tokio::sync::watch::Sender<bool>>,
+    /// When each account's last complete sync before this start began: mail
+    /// that came in after it was caught up on silently, so a check in the
+    /// background tells of it instead (see [`Core::announce_missed`]).
+    /// `None` once the user has seen the Inbox.
+    missed_since: Arc<std::sync::Mutex<Option<std::collections::BTreeMap<i64, i64>>>>,
 }
 
 impl Core {
@@ -903,6 +910,9 @@ impl Core {
         let calendar_db = Db::deferred_calendar(paths.calendar_db_file());
         let files_db = Db::deferred_files(paths.files_db_file());
         let bus = EventBus::new();
+        // Read before any account worker starts, as each records how far it
+        // caught up.
+        let missed_since = db.read(|conn| repo::accounts::caught_up(conn)).await?;
         let core = Core {
             db,
             calendar_db,
@@ -923,6 +933,8 @@ impl Core {
             ui_ready: Arc::new(tokio::sync::Notify::new()),
             ai_router_lock: Arc::new(tokio::sync::Mutex::new(())),
             transfers: Arc::new(tokio::sync::Notify::new()),
+            accounts_started: Arc::new(tokio::sync::watch::Sender::new(false)),
+            missed_since: Arc::new(std::sync::Mutex::new(Some(missed_since))),
         };
 
         // Finish any destructive operation interrupted between the mail and
@@ -1044,6 +1056,7 @@ impl Core {
                     Err(e) => {
                         events::sync_diagnostic("startup_accounts", None, e.code());
                         tracing::error!("listing accounts for startup sync: {e}");
+                        core.accounts_started.send_replace(true);
                         return;
                     }
                 };
@@ -1056,6 +1069,7 @@ impl Core {
                     Err(error) => tracing::error!("finding local mail for startup: {error}"),
                 }
                 events::sync_diagnostic("startup_workers", None, "ready");
+                core.accounts_started.send_replace(true);
                 // Calendar sync tasks for accounts with a syncing calendar
                 // connection, once older CalDAV passwords have moved to theirs.
                 if core.paths.calendar_db_file().exists() {
@@ -2056,6 +2070,18 @@ impl Core {
         self.ui_ready.notify_one();
     }
 
+    /// Start the account workers now, without waiting for a window to report
+    /// ready, and return once they run: for a check in the background, where
+    /// no window shows.
+    pub async fn start_accounts_now(&self) {
+        self.notify_ui_ready();
+        let _ = self
+            .accounts_started
+            .subscribe()
+            .wait_for(|started| *started)
+            .await;
+    }
+
     /// Free the SQLite page caches of every open store once a burst of work
     /// (a sync, an import) is over. Stores that were never opened stay closed.
     pub async fn release_cache_memory(&self) -> Result<()> {
@@ -9001,6 +9027,77 @@ impl Core {
             .write(move |conn| repo::notifications::prune(conn, now))
             .await
     }
+
+    /// Queue new-mail notifications for mail that reached an Inbox while no
+    /// copy of the app was following its account. A start catches up on that
+    /// mail silently, which suits a launch the user sees; a phone that starts
+    /// the app in the background to check for mail calls this after the
+    /// check ([`Self::sync_accounts`]), so the mail shows as it would have
+    /// with the app running. Each message is told of once, and nothing once
+    /// [`Self::forget_missed`] ran. Returns how many notifications it queued.
+    pub async fn announce_missed(&self) -> Result<usize> {
+        let since = self
+            .missed_since
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .clone();
+        let Some(since) = since.filter(|since| !since.is_empty()) else {
+            return Ok(0);
+        };
+        let now = now_ms();
+        let queued = self
+            .db
+            .write(move |conn| {
+                let tx = conn.transaction()?;
+                let queued = repo::notifications::enqueue_missed(&tx, &since, now)?;
+                tx.commit()?;
+                Ok(queued)
+            })
+            .await?;
+        let mut by_account = std::collections::BTreeMap::<i64, Vec<i64>>::new();
+        for &(account_id, thread_id) in &queued {
+            by_account.entry(account_id).or_default().push(thread_id);
+        }
+        for (account_id, thread_ids) in by_account {
+            tracing::info!(
+                account_id,
+                count = thread_ids.len(),
+                "receive: queued notifications for mail that came while the app wasn't running",
+            );
+            self.bus.emit(CoreEvent::MailNew {
+                account_id,
+                thread_ids,
+            });
+        }
+        Ok(queued.len())
+    }
+
+    /// The user has seen the Inbox since this start, so mail caught up on
+    /// then is no news: [`Self::announce_missed`] does nothing from now on.
+    pub fn forget_missed(&self) {
+        self.missed_since
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .take();
+    }
+
+    /// Wait until each of these queued actions has been tried: done, failed
+    /// or cancelled, or waiting to be tried again after an error, such as
+    /// while offline. For a host that must say when a notification's button
+    /// has done its work, as a phone suspends the app soon after.
+    pub async fn wait_for_actions(&self, action_ids: Vec<i64>) -> Result<()> {
+        loop {
+            let ids = action_ids.clone();
+            let untried = self
+                .db
+                .read(move |conn| repo::actions::untried(conn, &ids))
+                .await?;
+            if untried == 0 {
+                return Ok(());
+            }
+            tokio::time::sleep(SEND_STATUS_POLL_INTERVAL).await;
+        }
+    }
 }
 
 /// Build a (incoming → the user's reply) example from one of their sent
@@ -9966,6 +10063,64 @@ mod notification_dispatch_tests {
         );
     }
 
+    #[tokio::test]
+    async fn mail_that_came_while_the_app_was_closed_is_announced_once() {
+        let tmp = tempfile::tempdir().unwrap();
+        let paths = Paths::for_tests(tmp.path());
+        let now = now_ms();
+        let minute = 60_000;
+        // An unread message in the Inbox, received while no copy of the app
+        // ran: after the last run caught up, an hour ago.
+        let arrived = |conn: &rusqlite::Connection, subject: &str, at: i64| -> Result<()> {
+            let (_, id) = db::testutil::seed_message(conn, "a@test.dev", subject, false);
+            conn.execute(
+                "UPDATE messages SET internal_date = ?1 WHERE id = ?2",
+                rusqlite::params![at, id],
+            )?;
+            Ok(())
+        };
+        {
+            let db = Db::open(&paths.db_file()).unwrap();
+            db.write(move |conn| {
+                db::testutil::seed_account(conn);
+                repo::accounts::set_caught_up(conn, 1, now - 60 * minute)?;
+                arrived(conn, "While away", now - 10 * minute)
+            })
+            .await
+            .unwrap();
+        }
+
+        let core = Core::start(paths).await.unwrap();
+        let mut events = core.bus.subscribe();
+        assert_eq!(core.announce_missed().await.unwrap(), 1);
+        assert!(matches!(
+            events.try_recv(),
+            Ok(CoreEvent::MailNew { account_id: 1, .. })
+        ));
+        assert_eq!(core.announce_missed().await.unwrap(), 0);
+
+        // Once the user has seen the Inbox, mail caught up on is no news.
+        core.db
+            .write(move |conn| arrived(conn, "Seen in the Inbox", now - 5 * minute))
+            .await
+            .unwrap();
+        core.forget_missed();
+        assert_eq!(core.announce_missed().await.unwrap(), 0);
+    }
+
+    #[tokio::test]
+    async fn accounts_start_without_a_window_for_a_background_check() {
+        let tmp = tempfile::tempdir().unwrap();
+        let core = Core::start(Paths::for_tests(tmp.path())).await.unwrap();
+        // Well before the minute the start otherwise waits for a window.
+        tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            core.start_accounts_now(),
+        )
+        .await
+        .expect("the account workers started");
+    }
+
     #[tokio::test]
     async fn notifications_the_host_cannot_show_are_settled() {
         let (_tmp, core, ids) = seeded(3).await;