fix: dispatch notifications on their own task

platform/README.md

@@ -53,7 +53,7 @@ host-provided service:
 | `credentials` | System keychain (`SystemCredentialStore`) | Implements `CredentialStore` with the platform keystore |
 | `oauth_redirects` | Loopback redirect broker | Implements `OAuthRedirectBroker` (app links, custom schemes) |
 | `external_urls` | System URL launcher | `ExternalUrlOpener::new` with the host's native URL bridge |
-| `notification_publisher` | Disabled by default | `NotificationPublisher::new` with a callback that formats and raises the host notification; it receives raw sender, subject and thread ids so the host translates the text and chooses the notification slot |
+| `notification_publisher` | Disabled by default | `NotificationPublisher::new` with a callback that raises the host notification. Core decides which mail to notify for and passes the raw sender, subject and ids, so the host translates the text and chooses the notification slot |
 
 Mobile hosts use `PlatformContext::app_private(data_root, cache_dir, documents,
 credentials, oauth_redirects)`, which keeps durable data below

src/main.rs

@@ -31,10 +31,10 @@ mod mail_render_projection;
 mod mail_scope;
 mod mail_setup;
 mod mail_view_model;
-mod notifications;
-pub use notifications::{NewMailNotification, NotificationPublisher};
 mod mail_work;
 mod message_appearance;
+mod notifications;
+pub use notifications::{NewMailNotification, NotificationPublisher};
 mod number_format;
 mod oauth_browser;
 #[cfg(test)]
@@ -4356,6 +4356,10 @@ pub fn run(platform: PlatformContext) -> Result<(), Box<dyn std::error::Error>>
                         core.clone(),
                         Arc::clone(&startup_pending_core_updates),
                         startup_core_wake.clone(),
+                    );
+                    notifications::spawn_dispatcher(
+                        &startup_runtime,
+                        core.file_core(),
                         startup_notification_publisher.clone(),
                     );
                     if !startup_benchmark_disable_background {

src/notifications.rs

@@ -1,51 +1,17 @@
-//! New-mail notifications: the host publisher and the outbox dispatcher.
+//! New-mail notifications raised by the host.
 //!
-//! The sync engine enqueues one durable row per chime-worthy arrival. The
-//! dispatcher applies the user's Important/All/tabs scope, waits for pending AI
-//! routing, and hands raw message fields to the host publisher. Hosts format
-//! and translate the title and body themselves, so the shared layer never emits
-//! user-visible English text.
+//! Core keeps a durable outbox of new mail and decides what to notify for: the
+//! user's Important/All/tabs scope, mail that is still new and unread, and
+//! threads whose AI route has resolved. The host only raises each
+//! notification, formatting and translating the title and body itself, so the
+//! shared layer never emits user-visible English text.
 
-use std::sync::Arc;
+use std::{sync::Arc, time::Duration};
 
-use flectar_mail_core::models::{Settings, now_ms};
-use flectar_mail_core::{NotificationOutboxItem, RoutedTab};
+use flectar_mail_core::{Core, events::CoreEvent, models::now_ms};
+use tokio::sync::broadcast::{self, error::RecvError};
 
-use super::CoreMailSource;
-
-/// One new-mail event handed to a host publisher.
-///
-/// The fields are raw database values: hosts build the translated title and
-/// body, choose the notification slot, and attach the thread or message id to a
-/// `wantAgent` or the platform's equivalent action.
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct NewMailNotification {
-    pub account_id: i64,
-    pub message_id: i64,
-    pub thread_id: Option<i64>,
-    pub sender: String,
-    pub subject: String,
-}
-
-impl NewMailNotification {
-    fn from_outbox(item: &NotificationOutboxItem) -> Self {
-        let sender = item
-            .sender_name
-            .as_deref()
-            .map(str::trim)
-            .filter(|name| !name.is_empty())
-            .map(str::to_owned)
-            .or_else(|| item.sender_addr.clone())
-            .unwrap_or_default();
-        Self {
-            account_id: item.account_id,
-            message_id: item.message_id,
-            thread_id: item.thread_id,
-            sender,
-            subject: item.subject.clone(),
-        }
-    }
-}
+pub use flectar_mail_core::NewMailNotification;
 
 type PublishHandler = Arc<dyn Fn(&NewMailNotification) + Send + Sync + 'static>;
 
@@ -64,134 +30,64 @@ impl NotificationPublisher {
             handler: Some(Arc::new(publish)),
         }
     }
-
-    fn publish(&self, notification: &NewMailNotification) {
-        if let Some(handler) = &self.handler {
-            handler(notification);
-        }
-    }
-
-    fn is_enabled(&self) -> bool {
-        self.handler.is_some()
-    }
 }
 
-/// Apply the user's notification scope to a resolved tab key.
-fn scope_allows(settings: &Settings, tab: &str) -> bool {
-    match settings.notification_scope.as_str() {
-        "all" => true,
-        "tabs" => settings
-            .notification_tabs
-            .iter()
-            .any(|selected| selected == tab),
-        // "important" and any future or unknown value keep the historical
-        // Important-only behaviour rather than notifying for every message.
-        _ => tab == "important",
-    }
-}
+/// How long to wait before trying again after a dispatch failed.
+const RETRY_AFTER: Duration = Duration::from_secs(60);
 
-/// Drain the notification outbox for a host that can raise notifications.
-///
-/// A pending AI route is deferred and retried on the next dispatch; a message
-/// outside the selected scope is suppressed permanently. The publisher is
-/// called only after the row is claimed, and the row is marked delivered once
-/// the host callback returns.
-pub(crate) async fn dispatch_notifications(
-    core: &CoreMailSource,
-    publisher: &NotificationPublisher,
+/// Raise new-mail notifications through the host publisher: once at startup,
+/// after new mail arrives, and when a postponed notification falls due. A host
+/// without a publisher starts nothing.
+pub(crate) fn spawn_dispatcher(
+    runtime: &tokio::runtime::Runtime,
+    core: Arc<Core>,
+    publisher: NotificationPublisher,
 ) {
-    if !publisher.is_enabled() {
-        return;
-    }
-    let Ok(settings) = core.load_settings().await else {
+    let Some(publish) = publisher.handler else {
         return;
     };
-    if !settings.notifications_enabled {
-        return;
-    }
-    let file_core = core.file_core();
-    let Ok(items) = file_core.due_notifications(16).await else {
-        return;
-    };
-    for item in items {
-        let Some(thread_id) = item.thread_id else {
-            let _ = file_core
-                .suppress_notification_delivery(item.id, "thread was deleted")
-                .await;
-            continue;
-        };
-        let tab = match file_core.notification_thread_tab(thread_id).await {
-            Ok(Some(RoutedTab::Resolved(tab))) => tab,
-            Ok(Some(RoutedTab::Pending)) => {
-                let _ = file_core
-                    .defer_notification_delivery(item.id, now_ms() + 60_000)
-                    .await;
-                continue;
-            }
-            Ok(None) => {
-                let _ = file_core
-                    .suppress_notification_delivery(item.id, "thread was deleted")
-                    .await;
-                continue;
-            }
-            Err(_) => {
-                let _ = file_core
-                    .retry_notification_delivery(item.id, 60_000, "thread tab lookup failed")
-                    .await;
-                continue;
-            }
-        };
-        if !scope_allows(&settings, &tab) {
-            let _ = file_core
-                .suppress_notification_delivery(item.id, "outside notification scope")
-                .await;
-            continue;
+    let mut events = core.bus.subscribe();
+    runtime.spawn(async move {
+        if let Err(error) = core.recover_notification_deliveries().await {
+            tracing::warn!(%error, "notifications: could not recover interrupted deliveries");
         }
-        if !file_core
-            .claim_notification_delivery(item.id)
-            .await
-            .unwrap_or(false)
-        {
-            continue;
+        loop {
+            let wait = match core.dispatch_notifications(&*publish).await {
+                Ok(next) => {
+                    next.map(|at| Duration::from_millis(u64::try_from(at - now_ms()).unwrap_or(0)))
+                }
+                Err(error) => {
+                    tracing::warn!(%error, "notifications: dispatch failed");
+                    Some(RETRY_AFTER)
+                }
+            };
+            tokio::select! {
+                running = new_mail(&mut events) => {
+                    if !running {
+                        return;
+                    }
+                }
+                () = sleep(wait) => {}
+            }
+        }
+    });
+}
+
+/// Wait for new mail. A receiver that lagged may have missed some, so that
+/// counts as new mail too. Returns false once core shuts down.
+async fn new_mail(events: &mut broadcast::Receiver<CoreEvent>) -> bool {
+    loop {
+        match events.recv().await {
+            Ok(CoreEvent::MailNew { .. }) | Err(RecvError::Lagged(_)) => return true,
+            Ok(_) => {}
+            Err(RecvError::Closed) => return false,
         }
-        publisher.publish(&NewMailNotification::from_outbox(&item));
-        let _ = file_core.mark_notification_delivered(item.id).await;
     }
 }
 
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn important_scope_only_notifies_important() {
-        let settings = Settings::default();
-        assert!(scope_allows(&settings, "important"));
-        assert!(!scope_allows(&settings, "other"));
-        assert!(!scope_allows(&settings, "split:1"));
-    }
-
-    #[test]
-    fn all_scope_notifies_every_tab() {
-        let settings = Settings {
-            notification_scope: "all".into(),
-            ..Settings::default()
-        };
-        assert!(scope_allows(&settings, "important"));
-        assert!(scope_allows(&settings, "other"));
-        assert!(scope_allows(&settings, "label:7"));
-    }
-
-    #[test]
-    fn tabs_scope_honours_the_selected_tabs() {
-        let settings = Settings {
-            notification_scope: "tabs".into(),
-            notification_tabs: vec!["other".into(), "label:7".into()],
-            ..Settings::default()
-        };
-        assert!(!scope_allows(&settings, "important"));
-        assert!(scope_allows(&settings, "other"));
-        assert!(scope_allows(&settings, "label:7"));
-        assert!(!scope_allows(&settings, "split:1"));
+async fn sleep(wait: Option<Duration>) {
+    match wait {
+        Some(wait) => tokio::time::sleep(wait).await,
+        None => std::future::pending().await,
     }
 }