irontide_session_core/

notification.rs

//! M226: engine-side OS notification dispatcher.
//!
//! Subscribes to the existing alert broadcast stream and emits OS-level
//! desktop notifications on `TorrentFinished` and `TorrentError`. No new
//! alert kinds are introduced — the dispatcher is a pure consumer of
//! [`AlertKind::TorrentAdded`] / [`AlertKind::TorrentRemoved`] (for the
//! name cache) plus [`AlertKind::TorrentFinished`] / [`AlertKind::TorrentError`]
//! (for the notification trigger).
//!
//! The dispatcher is split from the OS-side delivery via the
//! [`NotificationSink`] trait so tests can inject [`InMemorySink`] and
//! observe the records without ever loading `notify-rust` into the test
//! binary. Production code uses [`LibNotifySink`], which wraps
//! `notify_rust::Notification::show()` in `tokio::task::spawn_blocking`
//! (the underlying D-Bus call is synchronous-blocking; running it on a
//! tokio worker thread would starve the runtime).
//!
//! Live-toggle semantics: the `notify_on_complete` / `notify_on_error`
//! gates are read **fresh per alert** from a snapshot of `Settings`
//! handed to the dispatcher via a `tokio::sync::watch` channel, so
//! `apply_settings` flips take effect on the next alert without
//! restarting the dispatcher (matches the `classify_immediate` contract).
//!
//! D-Bus absence is tolerated: the first `sink.show()` failure logs a
//! single `tracing::warn!` (gated by an `AtomicBool` flag); subsequent
//! failures degrade silently so headless deployments don't churn the log
//! stream.

use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use async_trait::async_trait;
use irontide_core::Id20;
use parking_lot::Mutex;
use thiserror::Error;
use tokio::sync::{broadcast, oneshot};
use tracing::{debug, warn};

use crate::alert::{Alert, AlertKind};
use irontide_settings::Settings;

/// Per-record state captured by [`InMemorySink`] for assertion in tests.
/// Not exposed in production callers — production callers only consume
/// the `Result` from [`NotificationSink::show`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NotificationRecord {
    /// First line / title of the notification (e.g. `"IronTide"`).
    pub summary: String,
    /// Body text — torrent name + status, already HTML-escaped via
    /// [`sanitize_notification_text`].
    pub body: String,
}

/// Failure modes for the sink contract. The dispatcher catches every
/// variant and logs once via `dbus_failure_logged`; nothing propagates
/// up the alert pipeline.
#[derive(Debug, Error)]
pub enum NotificationError {
    /// The underlying `notify-rust` call returned an error (no D-Bus
    /// session, daemon refused the message, etc.). String-coerced
    /// because `notify_rust::error::Error` is not `Clone`.
    #[error("notify-rust failed: {0}")]
    Backend(String),
    /// `tokio::task::spawn_blocking` join failure. Should never happen
    /// outside of runtime shutdown; we surface it so tests can observe.
    #[error("spawn_blocking join error: {0}")]
    JoinError(String),
    /// Injected by [`InMemorySink::with_failure`] for the
    /// `notification_dispatcher_handles_sink_failure` test.
    #[error("injected test failure: {0}")]
    Test(String),
}

/// The dispatcher → OS bridge. Production wires this through
/// [`LibNotifySink`]; tests inject [`InMemorySink`].
///
/// Why a trait at all: the F14 alternative (a `#[cfg(test)]`
/// environment-variable hack) couples the prod code path to test
/// configuration AND links `notify-rust` into the test binary. A trait
/// keeps the boundary explicit and decouples test runtime from D-Bus.
#[async_trait]
pub trait NotificationSink: Send + Sync + 'static {
    /// Emit a single notification. `summary` is the title; `body` is
    /// already-sanitized HTML-escaped text suitable for daemons that
    /// interpret a subset of HTML markup (KDE, GNOME).
    async fn show(&self, summary: &str, body: &str) -> Result<(), NotificationError>;
}

/// Production sink: wraps `notify-rust` and runs the blocking call on
/// the dedicated tokio blocking pool. Cheap to clone (it's a ZST).
#[derive(Debug, Default, Clone, Copy)]
pub struct LibNotifySink;

impl LibNotifySink {
    /// Construct a fresh production sink. Equivalent to `LibNotifySink`
    /// but reads like a constructor at the call site.
    #[must_use]
    pub fn new() -> Self {
        Self
    }
}

#[async_trait]
impl NotificationSink for LibNotifySink {
    async fn show(&self, summary: &str, body: &str) -> Result<(), NotificationError> {
        let summary = summary.to_string();
        let body = body.to_string();
        tokio::task::spawn_blocking(move || {
            notify_rust::Notification::new()
                .summary(&summary)
                .body(&body)
                .appname("irontide")
                .show()
                .map(|_handle| ())
                .map_err(|e| NotificationError::Backend(e.to_string()))
        })
        .await
        .map_err(|e| NotificationError::JoinError(e.to_string()))?
    }
}

/// Test-only sink: stores every emitted record in an `Arc<Mutex<Vec<_>>>`
/// so the test can drain + assert without parsing log output. Optionally
/// returns a stubbed failure to exercise the `dbus_failure_logged` path.
#[derive(Debug, Clone, Default)]
pub struct InMemorySink {
    /// Records accumulated across every [`Self::show`] call. Tests
    /// `lock()` + `clone()` to snapshot.
    pub records: Arc<Mutex<Vec<NotificationRecord>>>,
    /// When set, [`Self::show`] returns this error string verbatim
    /// (wrapped in [`NotificationError::Test`]) without recording.
    fail_with: Option<String>,
}

impl InMemorySink {
    /// Construct an empty in-memory sink with no failure injection.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Construct a sink that fails every `show()` with the given message.
    /// Used by `notification_dispatcher_handles_sink_failure`.
    #[must_use]
    pub fn with_failure(message: impl Into<String>) -> Self {
        Self {
            records: Arc::new(Mutex::new(Vec::new())),
            fail_with: Some(message.into()),
        }
    }

    /// Snapshot of the records vector. Convenient for assertions that
    /// don't want to hold the lock.
    #[must_use]
    pub fn snapshot(&self) -> Vec<NotificationRecord> {
        self.records.lock().clone()
    }
}

#[async_trait]
impl NotificationSink for InMemorySink {
    async fn show(&self, summary: &str, body: &str) -> Result<(), NotificationError> {
        if let Some(ref msg) = self.fail_with {
            return Err(NotificationError::Test(msg.clone()));
        }
        self.records.lock().push(NotificationRecord {
            summary: summary.to_string(),
            body: body.to_string(),
        });
        Ok(())
    }
}

/// HTML-escape a single string char-by-char so subsequent passes (e.g.
/// running `replace("<", "&lt;")` after `replace("&", "&amp;")`)
/// can't double-encode an already-escaped entity. Single pass, single
/// allocation. Five characters covered per the WHATWG HTML-fragment
/// rule (`<`, `>`, `&`, `"`, `'`). Anything else passes through verbatim
/// — non-ASCII (CJK, accents, emoji) survives the escape.
///
/// G4 from the M226 plan: sequential `.replace()` is **not** safe.
/// `"AT&T".replace("<", "&lt;").replace("&", "&amp;")` produces
/// `"AT&amp;T"`, but `"<AT&T>".replace("<", "&lt;").replace("&", "&amp;")`
/// produces `"&amp;lt;AT&amp;T&gt;"` — the second pass re-escapes the
/// `&` we just emitted. Char-by-char closes that hole.
#[must_use]
pub fn sanitize_notification_text(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    for c in s.chars() {
        match c {
            '<' => out.push_str("&lt;"),
            '>' => out.push_str("&gt;"),
            '&' => out.push_str("&amp;"),
            '"' => out.push_str("&quot;"),
            '\'' => out.push_str("&apos;"),
            _ => out.push(c),
        }
    }
    out
}

/// Pure-function variant of the per-alert dispatch logic, peeled out
/// from the dispatcher loop so it can be unit-tested in isolation
/// without spinning up a real session. Returns the body text to pass to
/// `sink.show()`, or `None` if the alert should be skipped (gate
/// disabled or unrelated alert kind).
fn dispatch_one(
    settings: &Settings,
    kind: &AlertKind,
    name_cache: &HashMap<Id20, String>,
) -> Option<(&'static str, String)> {
    match kind {
        AlertKind::TorrentFinished { info_hash } if settings.notify_on_complete => {
            let raw = name_cache
                .get(info_hash)
                .cloned()
                .unwrap_or_else(|| info_hash_short_hex(*info_hash));
            let name = sanitize_notification_text(&raw);
            Some(("IronTide", format!("{name} download complete")))
        }
        AlertKind::TorrentError { info_hash, message } if settings.notify_on_error => {
            let raw = name_cache
                .get(info_hash)
                .cloned()
                .unwrap_or_else(|| info_hash_short_hex(*info_hash));
            let name = sanitize_notification_text(&raw);
            let message = sanitize_notification_text(message);
            Some(("IronTide", format!("{name}: {message}")))
        }
        _ => None,
    }
}

/// Fallback name when no `TorrentAdded` event has populated the cache
/// AND the async `torrent_info` lookup also fails (rare: shutdown race).
fn info_hash_short_hex(hash: Id20) -> String {
    let hex = hash.to_hex();
    // Take the first 8 lowercase-hex chars — same convention the GUI
    // uses for "unnamed magnet" placeholders before metadata arrives.
    hex.chars().take(8).collect()
}

/// Dispatcher options for tests + production. Production callers
/// invoke [`DispatcherOptions::production`] which carries a
/// [`LibNotifySink`]; tests construct directly with an [`InMemorySink`].
pub struct DispatcherOptions {
    /// Box so the dispatcher stays object-safe across [`InMemorySink`]
    /// (tests) and [`LibNotifySink`] (production).
    pub sink: Box<dyn NotificationSink>,
    /// Live settings snapshot; the dispatcher reads `notify_on_complete`
    /// / `notify_on_error` from the most-recently-broadcast value on
    /// every alert. `tokio::sync::watch` is the natural fit — exactly
    /// one writer (`SessionActor::handle_apply_settings`) + many
    /// readers (each cheap clone).
    pub settings_rx: tokio::sync::watch::Receiver<Settings>,
    /// Broadcast subscription. The dispatcher takes ownership; passing
    /// it in lets the caller acquire the subscription BEFORE the
    /// dispatcher task is spawned (avoids the missed-alert race on
    /// session startup — H5 in the plan).
    pub alerts_rx: broadcast::Receiver<Alert>,
    /// Shutdown signal: drop the sender or call `.send(())` to ask the
    /// dispatcher to exit cleanly.
    pub shutdown_rx: oneshot::Receiver<()>,
}

/// Spawn a notification dispatcher task and return its `JoinHandle`. The
/// caller owns the handle; awaiting it during session shutdown waits for
/// the dispatcher's last in-flight `sink.show()` to drain.
#[must_use]
pub fn spawn_notification_dispatcher(opts: DispatcherOptions) -> tokio::task::JoinHandle<()> {
    let DispatcherOptions {
        sink,
        settings_rx,
        mut alerts_rx,
        mut shutdown_rx,
    } = opts;
    tokio::spawn(async move {
        let mut name_cache: HashMap<Id20, String> = HashMap::new();
        let dbus_failure_logged = AtomicBool::new(false);
        loop {
            tokio::select! {
                _ = &mut shutdown_rx => {
                    debug!("notification dispatcher: shutdown signal received");
                    break;
                }
                event = alerts_rx.recv() => {
                    let alert = match event {
                        Ok(alert) => alert,
                        Err(broadcast::error::RecvError::Lagged(n)) => {
                            // Lagged means we missed `n` alerts; the
                            // name cache may be inconsistent. Log + keep
                            // going — losing a few notifications is
                            // strictly preferable to crashing the
                            // dispatcher (which would silently kill OS
                            // toasts for the rest of the session).
                            warn!(lagged = n, "notification dispatcher: alert stream lagged");
                            continue;
                        }
                        Err(broadcast::error::RecvError::Closed) => {
                            debug!("notification dispatcher: alert stream closed");
                            break;
                        }
                    };
                    // Maintain the name cache before deciding whether to
                    // dispatch — a TorrentAdded immediately followed by
                    // TorrentFinished must observe the name in cache
                    // (single-thread channel ordering guarantees this).
                    match &alert.kind {
                        AlertKind::TorrentAdded { info_hash, name } => {
                            name_cache.insert(*info_hash, name.clone());
                            continue;
                        }
                        AlertKind::TorrentRemoved { info_hash } => {
                            name_cache.remove(info_hash);
                            continue;
                        }
                        _ => {}
                    }
                    let settings = settings_rx.borrow().clone();
                    let Some((summary, body)) = dispatch_one(&settings, &alert.kind, &name_cache)
                    else {
                        continue;
                    };
                    if let Err(e) = sink.show(summary, &body).await
                        && !dbus_failure_logged.swap(true, Ordering::Relaxed)
                    {
                        warn!(
                            error = %e,
                            "notification dispatcher: sink failed; degrading silently for the rest of the session"
                        );
                    }
                }
            }
        }
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::alert::Alert;
    use std::time::Duration;

    fn fake_hash(byte: u8) -> Id20 {
        Id20([byte; 20])
    }

    #[test]
    fn sanitizer_escapes_all_five_html_metacharacters() {
        let input = r#"<b>"hi" & 'bye'</b>"#;
        let out = sanitize_notification_text(input);
        assert_eq!(
            out,
            "&lt;b&gt;&quot;hi&quot; &amp; &apos;bye&apos;&lt;/b&gt;"
        );
    }

    /// M226 G4 regression guard: sequential `.replace("<")` then
    /// `.replace("&")` produces `&amp;lt;` for input `<`. Char-by-char
    /// must NOT double-escape.
    #[test]
    fn sanitizer_does_not_double_escape_existing_entities() {
        // "AT&T" → "AT&amp;T" (NOT "AT&amp;amp;T")
        assert_eq!(sanitize_notification_text("AT&T"), "AT&amp;T");
        // "<AT&T>" → "&lt;AT&amp;T&gt;" (NOT "&amp;lt;AT&amp;amp;T&amp;gt;")
        assert_eq!(sanitize_notification_text("<AT&T>"), "&lt;AT&amp;T&gt;");
    }

    #[test]
    fn sanitizer_passes_through_non_ascii_unchanged() {
        // Canadian English orthography + CJK + emoji must survive verbatim.
        let input = "Façade — 管理 — 🦀";
        assert_eq!(sanitize_notification_text(input), input);
    }

    #[test]
    fn dispatch_one_skips_when_notify_on_complete_false() {
        let s = Settings {
            notify_on_complete: false,
            ..Default::default()
        };
        let hash = fake_hash(0xAA);
        let cache = HashMap::from([(hash, "test-torrent".to_string())]);
        let result = dispatch_one(&s, &AlertKind::TorrentFinished { info_hash: hash }, &cache);
        assert!(result.is_none(), "gate false must skip dispatch");
    }

    #[test]
    fn dispatch_one_emits_when_notify_on_complete_true() {
        let s = Settings {
            notify_on_complete: true,
            ..Default::default()
        };
        let hash = fake_hash(0xBB);
        let cache = HashMap::from([(hash, "my-movie".to_string())]);
        let (summary, body) =
            dispatch_one(&s, &AlertKind::TorrentFinished { info_hash: hash }, &cache)
                .expect("gate true + finished must dispatch");
        assert_eq!(summary, "IronTide");
        assert_eq!(body, "my-movie download complete");
    }

    #[test]
    fn dispatch_one_emits_error_body_with_sanitised_message() {
        let s = Settings {
            notify_on_error: true,
            ..Default::default()
        };
        let hash = fake_hash(0xCC);
        let cache = HashMap::from([(hash, "<evil>".to_string())]);
        let (summary, body) = dispatch_one(
            &s,
            &AlertKind::TorrentError {
                info_hash: hash,
                message: "disk full <again>".to_string(),
            },
            &cache,
        )
        .expect("gate true + error must dispatch");
        assert_eq!(summary, "IronTide");
        assert_eq!(body, "&lt;evil&gt;: disk full &lt;again&gt;");
    }

    #[test]
    fn dispatch_one_falls_back_to_hex_prefix_when_cache_miss() {
        let s = Settings {
            notify_on_complete: true,
            ..Default::default()
        };
        let hash = fake_hash(0xDE);
        let cache = HashMap::new();
        let (_, body) =
            dispatch_one(&s, &AlertKind::TorrentFinished { info_hash: hash }, &cache).unwrap();
        // First 8 chars of the lowercase hex of [0xDE; 20] is "dededede".
        assert!(
            body.starts_with("dededede"),
            "cache miss must fall back to hex prefix, got: {body}"
        );
    }

    #[tokio::test]
    async fn in_memory_sink_records_and_can_inject_failure() {
        let sink = InMemorySink::new();
        sink.show("title", "body").await.unwrap();
        let snap = sink.snapshot();
        assert_eq!(snap.len(), 1);
        assert_eq!(snap[0].summary, "title");

        let failing = InMemorySink::with_failure("boom");
        let err = failing.show("t", "b").await.unwrap_err();
        assert!(matches!(err, NotificationError::Test(_)));
    }

    /// Drive the dispatcher loop end-to-end with [`InMemorySink`] + a real
    /// broadcast channel. Confirms that the alert pump → cache lookup →
    /// [`dispatch_one`] → `sink.show` chain works.
    #[tokio::test]
    async fn dispatcher_emits_completion_notification_with_cached_name() {
        let (alert_tx, alert_rx) = broadcast::channel::<Alert>(16);
        let (shutdown_tx, shutdown_rx) = oneshot::channel();
        let settings = Settings {
            notify_on_complete: true,
            ..Default::default()
        };
        let (settings_tx, settings_rx) = tokio::sync::watch::channel(settings);
        // Suppress unused-var lint for the test-only writer handle.
        let _ = &settings_tx;

        let sink = InMemorySink::new();
        let join = spawn_notification_dispatcher(DispatcherOptions {
            sink: Box::new(sink.clone()),
            settings_rx,
            alerts_rx: alert_rx,
            shutdown_rx,
        });

        let hash = fake_hash(0xEE);
        alert_tx
            .send(Alert::new(AlertKind::TorrentAdded {
                info_hash: hash,
                name: "demo-torrent".to_string(),
            }))
            .unwrap();
        alert_tx
            .send(Alert::new(AlertKind::TorrentFinished { info_hash: hash }))
            .unwrap();

        // Give the dispatcher loop time to drain.
        tokio::time::sleep(Duration::from_millis(50)).await;
        let _ = shutdown_tx.send(());
        join.await.unwrap();

        let records = sink.snapshot();
        assert_eq!(records.len(), 1, "exactly one notification expected");
        assert_eq!(records[0].summary, "IronTide");
        assert_eq!(records[0].body, "demo-torrent download complete");
    }

    /// Sink failure must NOT crash the dispatcher and must log-once.
    #[tokio::test]
    async fn dispatcher_handles_sink_failure_without_crashing() {
        let (alert_tx, alert_rx) = broadcast::channel::<Alert>(16);
        let (shutdown_tx, shutdown_rx) = oneshot::channel();
        let settings = Settings {
            notify_on_complete: true,
            ..Default::default()
        };
        let (_settings_tx, settings_rx) = tokio::sync::watch::channel(settings);

        let sink = InMemorySink::with_failure("no dbus session");
        let join = spawn_notification_dispatcher(DispatcherOptions {
            sink: Box::new(sink),
            settings_rx,
            alerts_rx: alert_rx,
            shutdown_rx,
        });

        let hash = fake_hash(0xFF);
        alert_tx
            .send(Alert::new(AlertKind::TorrentAdded {
                info_hash: hash,
                name: "fail-test".to_string(),
            }))
            .unwrap();
        alert_tx
            .send(Alert::new(AlertKind::TorrentFinished { info_hash: hash }))
            .unwrap();
        alert_tx
            .send(Alert::new(AlertKind::TorrentFinished { info_hash: hash }))
            .unwrap();

        tokio::time::sleep(Duration::from_millis(50)).await;
        let _ = shutdown_tx.send(());
        // Awaiting the join handle is the assertion: a panicking
        // dispatcher would surface here as JoinError.
        join.await
            .expect("dispatcher must not panic on sink failure");
    }

    /// Live toggle: TorrentFinished#1 fires with gate=false (no record),
    /// then settings flip to true, TorrentFinished#2 fires (record).
    #[tokio::test]
    async fn dispatcher_respects_live_settings_toggle() {
        let (alert_tx, alert_rx) = broadcast::channel::<Alert>(16);
        let (shutdown_tx, shutdown_rx) = oneshot::channel();
        let settings = Settings {
            notify_on_complete: false,
            ..Default::default()
        };
        let (settings_tx, settings_rx) = tokio::sync::watch::channel(settings);

        let sink = InMemorySink::new();
        let join = spawn_notification_dispatcher(DispatcherOptions {
            sink: Box::new(sink.clone()),
            settings_rx,
            alerts_rx: alert_rx,
            shutdown_rx,
        });

        let hash_a = fake_hash(0xA1);
        alert_tx
            .send(Alert::new(AlertKind::TorrentAdded {
                info_hash: hash_a,
                name: "first".to_string(),
            }))
            .unwrap();
        alert_tx
            .send(Alert::new(AlertKind::TorrentFinished { info_hash: hash_a }))
            .unwrap();
        tokio::time::sleep(Duration::from_millis(30)).await;
        // Flip gate to true; subsequent finish must emit.
        settings_tx.send_modify(|s| s.notify_on_complete = true);

        let hash_b = fake_hash(0xB2);
        alert_tx
            .send(Alert::new(AlertKind::TorrentAdded {
                info_hash: hash_b,
                name: "second".to_string(),
            }))
            .unwrap();
        alert_tx
            .send(Alert::new(AlertKind::TorrentFinished { info_hash: hash_b }))
            .unwrap();
        tokio::time::sleep(Duration::from_millis(50)).await;
        let _ = shutdown_tx.send(());
        join.await.unwrap();

        let records = sink.snapshot();
        assert_eq!(
            records.len(),
            1,
            "only the second TorrentFinished must emit"
        );
        assert_eq!(records[0].body, "second download complete");
    }

    /// Cache eviction: after [`AlertKind::TorrentRemoved`] the next [`AlertKind::TorrentFinished`]
    /// for the same hash must fall through to the hex-prefix fallback.
    /// ([`AlertKind::TorrentFinished`] after removal isn't a normal real-world ordering
    /// but the dispatcher must tolerate the broadcast race regardless.)
    #[tokio::test]
    async fn dispatcher_evicts_name_cache_on_torrent_removed() {
        let (alert_tx, alert_rx) = broadcast::channel::<Alert>(16);
        let (shutdown_tx, shutdown_rx) = oneshot::channel();
        let settings = Settings {
            notify_on_complete: true,
            ..Default::default()
        };
        let (_settings_tx, settings_rx) = tokio::sync::watch::channel(settings);

        let sink = InMemorySink::new();
        let join = spawn_notification_dispatcher(DispatcherOptions {
            sink: Box::new(sink.clone()),
            settings_rx,
            alerts_rx: alert_rx,
            shutdown_rx,
        });

        let hash = fake_hash(0xCA);
        alert_tx
            .send(Alert::new(AlertKind::TorrentAdded {
                info_hash: hash,
                name: "cache-test".to_string(),
            }))
            .unwrap();
        alert_tx
            .send(Alert::new(AlertKind::TorrentRemoved { info_hash: hash }))
            .unwrap();
        // Now drive a Finished for the same hash; cache should be empty.
        alert_tx
            .send(Alert::new(AlertKind::TorrentFinished { info_hash: hash }))
            .unwrap();

        tokio::time::sleep(Duration::from_millis(50)).await;
        let _ = shutdown_tx.send(());
        join.await.unwrap();

        let records = sink.snapshot();
        assert_eq!(records.len(), 1);
        assert!(
            !records[0].body.contains("cache-test"),
            "after eviction the cached name must NOT appear; got {}",
            records[0].body
        );
        // hex prefix of [0xCA; 20] starts with "cacacaca".
        assert!(
            records[0].body.starts_with("cacacaca"),
            "expected hex-prefix fallback; got {}",
            records[0].body
        );
    }

    #[tokio::test]
    async fn dispatcher_exits_cleanly_when_alert_stream_closes() {
        let (alert_tx, alert_rx) = broadcast::channel::<Alert>(4);
        let (_shutdown_tx, shutdown_rx) = oneshot::channel();
        let (_settings_tx, settings_rx) = tokio::sync::watch::channel(Settings::default());
        let join = spawn_notification_dispatcher(DispatcherOptions {
            sink: Box::new(InMemorySink::new()),
            settings_rx,
            alerts_rx: alert_rx,
            shutdown_rx,
        });
        // Dropping the broadcast Sender closes the stream.
        drop(alert_tx);
        // Dispatcher must observe Closed and exit.
        tokio::time::timeout(Duration::from_secs(1), join)
            .await
            .expect("dispatcher must exit after broadcast Sender drops")
            .unwrap();
    }
}