feat: ntfy notifications for imported, needs-decision, broken (#96)

crates/arr-daemon/src/attention.rs

@@ -0,0 +1,227 @@
+//! §9.5 *needs a decision* → the operator alone: a movie entered the
+//! no-PT-source queue, or hard-failed twice on different releases (the same
+//! two queues `GET /api/queues/attention` reports, §9.3).
+//!
+//! Edge-triggered: a movie notifies once when it enters either queue, and is
+//! forgotten once it leaves both, so a future re-entry notifies again. The
+//! notified set is in-memory and transient — a restart re-notifies whatever
+//! is currently queued, same tradeoff `ImportAction`'s probe cache makes.
+
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+
+use arr_db::Db;
+use tokio::sync::Mutex;
+
+use crate::notify::Notifier;
+use crate::reconcile::{Action, ActionFuture, Outcome};
+
+#[derive(Debug)]
+pub struct AttentionAction {
+    notifier: Notifier,
+    operator_topic: String,
+    notified: Arc<Mutex<HashSet<i64>>>,
+}
+
+impl AttentionAction {
+    #[must_use]
+    pub fn new(notifier: Notifier, operator_topic: String) -> Self {
+        Self {
+            notifier,
+            operator_topic,
+            notified: Arc::new(Mutex::new(HashSet::new())),
+        }
+    }
+
+    async fn tick(&self, database: &Db) -> Result<Vec<Outcome>, sqlx::Error> {
+        let mut queued: HashMap<i64, (String, &'static str)> = HashMap::new();
+
+        let no_pt_source = sqlx::query!(
+            r#"
+            SELECT id AS "id!: i64", title AS "title!: String", year
+            FROM   movies
+            WHERE  id IN (
+                SELECT m.id
+                FROM   movies m
+                JOIN   roots root ON root.id = m.root_id
+                WHERE  root.audience = 'kids'
+                  AND  m.wanted = 1 AND m.blocked = 0 AND m.state = 'missing'
+                  AND  m.search_attempts > 0
+                  AND  NOT EXISTS (
+                          SELECT 1 FROM movie_releases mr
+                          JOIN   releases r ON r.id = mr.release_id
+                          WHERE  mr.movie_id = m.id AND r.verdict IN ('eligible', 'waived')
+                      )
+            )
+            "#
+        )
+        .fetch_all(database.pool())
+        .await?;
+        for row in no_pt_source {
+            queued.insert(
+                row.id,
+                (
+                    title_with_year(&row.title, row.year),
+                    "no Portuguese-audio source found",
+                ),
+            );
+        }
+
+        let needs_decision = sqlx::query!(
+            r#"
+            SELECT id AS "id!: i64", title AS "title!: String", year
+            FROM   movies
+            WHERE  (SELECT count(DISTINCT g.release_id)
+                    FROM   grabs g
+                    WHERE  g.target_kind = 'movie' AND g.target_id = movies.id
+                      AND  g.state = 'failed') >= 2
+            "#
+        )
+        .fetch_all(database.pool())
+        .await?;
+        for row in needs_decision {
+            queued.entry(row.id).or_insert_with(|| {
+                (
+                    title_with_year(&row.title, row.year),
+                    "hard-failed twice on different releases",
+                )
+            });
+        }
+
+        let mut notified = self.notified.lock().await;
+        let mut outcomes = Vec::new();
+        for (&movie_id, (title, reason)) in &queued {
+            if !notified.insert(movie_id) {
+                continue;
+            }
+            match self
+                .notifier
+                .send(&self.operator_topic, title, reason)
+                .await
+            {
+                Ok(()) => outcomes.push(Outcome::new(
+                    format!("movie {movie_id} needs a decision"),
+                    format!("notified operator: {reason}"),
+                )),
+                Err(error) => {
+                    tracing::warn!(%error, movie_id, "needs-decision notification failed");
+                }
+            }
+        }
+        notified.retain(|movie_id| queued.contains_key(movie_id));
+
+        Ok(outcomes)
+    }
+}
+
+fn title_with_year(title: &str, year: Option<i64>) -> String {
+    match year {
+        Some(year) => format!("{title} ({year})"),
+        None => title.to_string(),
+    }
+}
+
+impl Action for AttentionAction {
+    fn name(&self) -> &'static str {
+        "attention"
+    }
+
+    fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a> {
+        Box::pin(async move { self.tick(database).await.map_err(Into::into) })
+    }
+}
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used)]
+mod tests {
+    use wiremock::matchers::method;
+    use wiremock::{Mock, MockServer, ResponseTemplate};
+
+    use super::*;
+
+    async fn seeded_database() -> (tempfile::TempDir, Db) {
+        let dir = tempfile::tempdir().unwrap();
+        let database = Db::connect(dir.path().join("arr.db")).await.unwrap();
+        database.migrate().await.unwrap();
+        (dir, database)
+    }
+
+    async fn insert_no_pt_source_movie(database: &Db) -> i64 {
+        let root_id: i64 =
+            sqlx::query_scalar("SELECT id FROM roots WHERE kind = 'movie' AND audience = 'kids'")
+                .fetch_one(database.pool())
+                .await
+                .unwrap();
+        sqlx::query(
+            "INSERT INTO movies
+                (tmdb_id, title, year, root_id, wanted, state, search_attempts)
+             VALUES (1, 'Encanto', 2021, ?, 1, 'missing', 1)
+             RETURNING id",
+        )
+        .bind(root_id)
+        .fetch_one(database.pool())
+        .await
+        .unwrap();
+        sqlx::query_scalar("SELECT id FROM movies WHERE tmdb_id = 1")
+            .fetch_one(database.pool())
+            .await
+            .unwrap()
+    }
+
+    #[tokio::test]
+    async fn a_movie_entering_the_queue_notifies_the_operator_once() {
+        let (_dir, database) = seeded_database().await;
+        insert_no_pt_source_movie(&database).await;
+
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .respond_with(ResponseTemplate::new(200))
+            .mount(&server)
+            .await;
+        let action = AttentionAction::new(
+            Notifier::new(server.uri()).unwrap(),
+            "operator-topic".to_string(),
+        );
+
+        let first = action.tick(&database).await.unwrap();
+        let second = action.tick(&database).await.unwrap();
+
+        assert_eq!(first.len(), 1, "notifies on the tick it enters the queue");
+        assert_eq!(second.len(), 0, "does not repeat while still queued");
+        assert_eq!(server.received_requests().await.unwrap().len(), 1);
+    }
+
+    #[tokio::test]
+    async fn leaving_and_re_entering_the_queue_notifies_again() {
+        let (_dir, database) = seeded_database().await;
+        let movie_id = insert_no_pt_source_movie(&database).await;
+
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .respond_with(ResponseTemplate::new(200))
+            .mount(&server)
+            .await;
+        let action = AttentionAction::new(
+            Notifier::new(server.uri()).unwrap(),
+            "operator-topic".to_string(),
+        );
+        action.tick(&database).await.unwrap();
+
+        sqlx::query("UPDATE movies SET blocked = 1 WHERE id = ?")
+            .bind(movie_id)
+            .execute(database.pool())
+            .await
+            .unwrap();
+        action.tick(&database).await.unwrap();
+
+        sqlx::query("UPDATE movies SET blocked = 0 WHERE id = ?")
+            .bind(movie_id)
+            .execute(database.pool())
+            .await
+            .unwrap();
+        let third = action.tick(&database).await.unwrap();
+
+        assert_eq!(third.len(), 1, "re-enters and notifies again");
+        assert_eq!(server.received_requests().await.unwrap().len(), 2);
+    }
+}

crates/arr-daemon/src/broken.rs

@@ -0,0 +1,220 @@
+//! §9.5 *broken* → the operator alone: Prowlarr, Transmission or TMDB
+//! unreachable.
+//!
+//! Edge-triggered: notifies once when an upstream stops answering, and
+//! silently re-arms once it answers again. There is no "fixed" notification —
+//! DESIGN.md names exactly three events, and a fourth needs a design change,
+//! not an issue.
+
+use std::collections::HashSet;
+use std::sync::Arc;
+use std::time::Duration;
+
+use arr_db::Db;
+use reqwest::Client;
+use tokio::sync::Mutex;
+
+use crate::notify::Notifier;
+use crate::reconcile::{Action, ActionFuture, Outcome};
+
+const PROBE_TIMEOUT: Duration = Duration::from_secs(5);
+
+/// Where the three upstreams live, and the keys the two that need one.
+#[derive(Debug, Clone)]
+pub struct Upstreams {
+    pub prowlarr_url: String,
+    pub prowlarr_api_key: Option<String>,
+    pub transmission_url: String,
+    pub tmdb_url: String,
+    pub tmdb_api_key: Option<String>,
+}
+
+#[derive(Debug)]
+pub struct BrokenAction {
+    http: Client,
+    upstreams: Upstreams,
+    notifier: Notifier,
+    operator_topic: String,
+    /// Which upstreams are currently notified as broken. Transient — a
+    /// restart re-probes and re-notifies whatever is still down.
+    broken: Arc<Mutex<HashSet<&'static str>>>,
+}
+
+impl BrokenAction {
+    #[must_use]
+    pub fn new(upstreams: Upstreams, notifier: Notifier, operator_topic: String) -> Self {
+        Self {
+            http: Client::new(),
+            upstreams,
+            notifier,
+            operator_topic,
+            broken: Arc::new(Mutex::new(HashSet::new())),
+        }
+    }
+
+    async fn tick(&self) -> Vec<Outcome> {
+        let (prowlarr, transmission, tmdb) = tokio::join!(
+            self.probe_prowlarr(),
+            self.probe_transmission(),
+            self.probe_tmdb(),
+        );
+
+        let mut outcomes = Vec::new();
+        outcomes.extend(self.notify_transition("prowlarr", prowlarr).await);
+        outcomes.extend(self.notify_transition("transmission", transmission).await);
+        outcomes.extend(self.notify_transition("tmdb", tmdb).await);
+        outcomes
+    }
+
+    /// `reachable` is `true` when the upstream answered, or when it needs no
+    /// key and none is configured (not an outage — see `probe_tmdb`).
+    async fn notify_transition(&self, name: &'static str, reachable: bool) -> Option<Outcome> {
+        let mut broken = self.broken.lock().await;
+        if reachable {
+            broken.remove(name);
+            return None;
+        }
+        if !broken.insert(name) {
+            return None;
+        }
+        match self
+            .notifier
+            .send(
+                &self.operator_topic,
+                "arr: broken",
+                &format!("{name} is unreachable"),
+            )
+            .await
+        {
+            Ok(()) => Some(Outcome::new(
+                format!("{name} unreachable"),
+                "notified operator".to_string(),
+            )),
+            Err(error) => {
+                tracing::warn!(%error, name, "broken notification failed");
+                None
+            }
+        }
+    }
+
+    /// Prowlarr answers `/ping` without a key; the key is sent anyway so a
+    /// misconfigured one still counts as unreachable rather than silently
+    /// searching nothing.
+    async fn probe_prowlarr(&self) -> bool {
+        let url = format!("{}/ping", self.upstreams.prowlarr_url.trim_end_matches('/'));
+        let mut request = self.http.get(url).timeout(PROBE_TIMEOUT);
+        if let Some(key) = &self.upstreams.prowlarr_api_key {
+            request = request.header("X-Api-Key", key);
+        }
+        matches!(request.send().await, Ok(response) if response.status().is_success())
+    }
+
+    /// Transmission answers an RPC call without a session id with `409` plus
+    /// the id to retry with. That is a live daemon, so it counts as
+    /// reachable.
+    async fn probe_transmission(&self) -> bool {
+        let request = self
+            .http
+            .post(&self.upstreams.transmission_url)
+            .timeout(PROBE_TIMEOUT)
+            .json(&serde_json::json!({ "method": "session-get" }));
+        matches!(
+            request.send().await,
+            Ok(response)
+                if response.status().is_success()
+                    || response.status() == reqwest::StatusCode::CONFLICT
+        )
+    }
+
+    /// No key configured is not an outage (DESIGN.md §9.5 is about
+    /// reachability, not setup) — TMDB is simply not probed.
+    async fn probe_tmdb(&self) -> bool {
+        let Some(key) = &self.upstreams.tmdb_api_key else {
+            return true;
+        };
+        let url = format!(
+            "{}/configuration",
+            self.upstreams.tmdb_url.trim_end_matches('/')
+        );
+        matches!(
+            self.http
+                .get(url)
+                .timeout(PROBE_TIMEOUT)
+                .query(&[("api_key", key)])
+                .send()
+                .await,
+            Ok(response) if response.status().is_success()
+        )
+    }
+}
+
+impl Action for BrokenAction {
+    fn name(&self) -> &'static str {
+        "broken"
+    }
+
+    fn run<'a>(&'a self, _database: &'a Db) -> ActionFuture<'a> {
+        Box::pin(async move { Ok(self.tick().await) })
+    }
+}
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used)]
+mod tests {
+    use wiremock::matchers::{method, path};
+    use wiremock::{Mock, MockServer, ResponseTemplate};
+
+    use super::*;
+
+    fn upstreams(prowlarr_url: String, transmission_url: String) -> Upstreams {
+        Upstreams {
+            prowlarr_url,
+            prowlarr_api_key: None,
+            transmission_url,
+            tmdb_url: "http://127.0.0.1:1".to_string(),
+            tmdb_api_key: None,
+        }
+    }
+
+    #[tokio::test]
+    async fn an_unreachable_upstream_notifies_once_then_re_arms() {
+        let prowlarr = MockServer::start().await;
+        // No mock mounted: every request 404s, which counts as unreachable.
+        let transmission = MockServer::start().await;
+        Mock::given(method("POST"))
+            .respond_with(ResponseTemplate::new(409))
+            .mount(&transmission)
+            .await;
+
+        let ntfy = MockServer::start().await;
+        Mock::given(method("POST"))
+            .respond_with(ResponseTemplate::new(200))
+            .mount(&ntfy)
+            .await;
+
+        let action = BrokenAction::new(
+            upstreams(prowlarr.uri(), transmission.uri()),
+            Notifier::new(ntfy.uri()).unwrap(),
+            "operator-topic".to_string(),
+        );
+
+        let first = action.tick().await;
+        let second = action.tick().await;
+
+        assert_eq!(first.len(), 1, "notifies on the tick it goes unreachable");
+        assert_eq!(second.len(), 0, "does not repeat while still broken");
+
+        Mock::given(method("GET"))
+            .and(path("/ping"))
+            .respond_with(ResponseTemplate::new(200))
+            .mount(&prowlarr)
+            .await;
+        let recovered = action.tick().await;
+        assert_eq!(recovered.len(), 0, "recovery is silent, no fourth event");
+
+        // Take it down again: a fresh outage re-arms and notifies again.
+        prowlarr.reset().await;
+        let broken_again = action.tick().await;
+        assert_eq!(broken_again.len(), 1, "re-arms after recovering");
+    }
+}

crates/arr-daemon/src/config.rs

@@ -29,6 +29,7 @@ pub const ENV_TMDB_URL: &str = "ARR_TMDB_URL";
 pub const ENV_JELLYFIN_URL: &str = "ARR_JELLYFIN_URL";
 pub const ENV_JELLYFIN_API_KEY: &str = "ARR_JELLYFIN_API_KEY";
 pub const ENV_NTFY_URL: &str = "ARR_NTFY_URL";
+pub const ENV_NTFY_OPERATOR_TOPIC: &str = "ARR_NTFY_OPERATOR_TOPIC";
 
 pub const DEFAULT_BIND_ADDR: &str = "0.0.0.0:7878";
 pub const DEFAULT_DATABASE_PATH: &str = "arr.db";
@@ -89,6 +90,8 @@ struct ConfigFile {
     jellyfin_url: Option<String>,
     #[serde(default)]
     ntfy_url: Option<String>,
+    #[serde(default)]
+    ntfy_operator_topic: Option<String>,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Deserialize)]
@@ -127,6 +130,7 @@ pub struct EnvOverrides {
     pub jellyfin_url: Option<String>,
     pub jellyfin_api_key: Option<String>,
     pub ntfy_url: Option<String>,
+    pub ntfy_operator_topic: Option<String>,
 }
 
 impl EnvOverrides {
@@ -147,6 +151,7 @@ impl EnvOverrides {
             jellyfin_url: std::env::var(ENV_JELLYFIN_URL).ok(),
             jellyfin_api_key: std::env::var(ENV_JELLYFIN_API_KEY).ok(),
             ntfy_url: std::env::var(ENV_NTFY_URL).ok(),
+            ntfy_operator_topic: std::env::var(ENV_NTFY_OPERATOR_TOPIC).ok(),
         }
     }
 }
@@ -173,6 +178,10 @@ pub struct Config {
     pub jellyfin_url: String,
     pub jellyfin_api_key: Option<String>,
     pub ntfy_url: String,
+    /// The operator's ntfy topic (DESIGN.md §9.5) for *needs a decision* and
+    /// *broken*. `None` means those two notifications are skipped — there is
+    /// no sensible default topic name to fall back to.
+    pub ntfy_operator_topic: Option<String>,
 }
 
 impl Config {
@@ -253,6 +262,7 @@ impl Config {
                 .ntfy_url
                 .or(file.ntfy_url)
                 .unwrap_or_else(|| DEFAULT_NTFY_URL.to_string()),
+            ntfy_operator_topic: env.ntfy_operator_topic.or(file.ntfy_operator_topic),
         })
     }
 }
@@ -294,6 +304,7 @@ mod tests {
         assert_eq!(config.jellyfin_url, DEFAULT_JELLYFIN_URL);
         assert!(config.tracker_seeding.is_empty());
         assert_eq!(config.ntfy_url, DEFAULT_NTFY_URL);
+        assert_eq!(config.ntfy_operator_topic, None);
         assert_eq!(config.prowlarr_api_key, None);
         assert_eq!(config.tmdb_api_key, None);
         assert_eq!(config.jellyfin_api_key, None);
@@ -466,6 +477,26 @@ prowlarr_url = "http://prowlarr.internal:9696"
         );
     }
 
+    #[test]
+    fn ntfy_operator_topic_comes_from_env_or_file() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("arr.toml");
+        std::fs::write(&path, "ntfy_operator_topic = \"from-file\"\n").unwrap();
+        let env = EnvOverrides {
+            config_file: Some(path.to_string_lossy().into_owned()),
+            ..EnvOverrides::default()
+        };
+        let config = Config::resolve(env.clone()).unwrap();
+        assert_eq!(config.ntfy_operator_topic.as_deref(), Some("from-file"));
+
+        let config = Config::resolve(EnvOverrides {
+            ntfy_operator_topic: Some("from-env".into()),
+            ..env
+        })
+        .unwrap();
+        assert_eq!(config.ntfy_operator_topic.as_deref(), Some("from-env"));
+    }
+
     #[test]
     fn invalid_env_bind_addr_returns_diagnostic() {
         let env = EnvOverrides {

crates/arr-daemon/src/import.rs

@@ -26,6 +26,7 @@ use arr_dl::TransmissionClient;
 use arr_probe::Prober;
 
 use crate::jellyfin::JellyfinClient;
+use crate::notify::Notifier;
 use crate::reconcile::{Action, ActionFuture, Outcome};
 
 /// A failure during one import tick.
@@ -75,6 +76,15 @@ pub struct ImportAction {
     transmission: TransmissionClient,
     prober: Prober,
     jellyfin: JellyfinClient,
+    notifier: Notifier,
+    /// The operator's ntfy topic (DESIGN.md §9.5), for the *broken*
+    /// notification a disk-full hardlink/copy failure raises. `None` when
+    /// unconfigured: the failure is still logged, just not notified.
+    operator_topic: Option<String>,
+    /// Debounces the disk-full *broken* notification so a stuck-full disk
+    /// notifies once, not every tick. Transient: a restart re-arms it, same
+    /// as the probe cache above.
+    disk_full_notified: std::sync::Arc<tokio::sync::Mutex<bool>>,
     /// Probe results by path, kept across ticks. The reconcile lane cancels
     /// the whole action after its 25 s budget while one probe alone may take
     /// up to 60 s, so without this a large multi-file torrent would restart
@@ -101,11 +111,20 @@ struct PendingImport {
 
 impl ImportAction {
     #[must_use]
-    pub fn new(transmission: TransmissionClient, prober: Prober, jellyfin: JellyfinClient) -> Self {
+    pub fn new(
+        transmission: TransmissionClient,
+        prober: Prober,
+        jellyfin: JellyfinClient,
+        notifier: Notifier,
+        operator_topic: Option<String>,
+    ) -> Self {
         Self {
             transmission,
             prober,
             jellyfin,
+            notifier,
+            operator_topic,
+            disk_full_notified: std::sync::Arc::new(tokio::sync::Mutex::new(false)),
             probed: std::sync::Arc::new(tokio::sync::Mutex::new(HashMap::new())),
         }
     }
@@ -170,33 +189,77 @@ impl ImportAction {
         let mut outcomes = Vec::new();
         for pending in pending_imports(database).await? {
             match self.import_one(database, &pending).await {
-                Ok(Some(outcome)) => outcomes.push(outcome),
+                Ok(Some(outcome)) => {
+                    self.clear_disk_full().await;
+                    outcomes.push(outcome);
+                }
                 Ok(None) => {}
                 // One grab's failure must not cost the rest of the tick.
-                Err(error) => tracing::error!(
-                    grab_id = pending.grab_id,
-                    movie_id = pending.movie_id,
-                    title = pending.title,
-                    %error,
-                    "import failed"
-                ),
+                Err(error) => {
+                    self.notify_if_disk_full(&error).await;
+                    tracing::error!(
+                        grab_id = pending.grab_id,
+                        movie_id = pending.movie_id,
+                        title = pending.title,
+                        %error,
+                        "import failed"
+                    );
+                }
             }
         }
         for pending in pending_tv_imports(database).await? {
             match self.import_tv_one(database, &pending).await {
-                Ok(Some(outcome)) => outcomes.push(outcome),
+                Ok(Some(outcome)) => {
+                    self.clear_disk_full().await;
+                    outcomes.push(outcome);
+                }
                 Ok(None) => {}
-                Err(error) => tracing::error!(
-                    grab_id = pending.grab_id,
-                    series = pending.series_title,
-                    %error,
-                    "tv import failed"
-                ),
+                Err(error) => {
+                    self.notify_if_disk_full(&error).await;
+                    tracing::error!(
+                        grab_id = pending.grab_id,
+                        series = pending.series_title,
+                        %error,
+                        "tv import failed"
+                    );
+                }
             }
         }
         Ok(outcomes)
     }
 
+    /// §9.5 *broken*: a hardlink or copy that failed because the target
+    /// filesystem is full. Debounced so a disk that stays full notifies once,
+    /// not every tick, and silently re-arms once space frees up.
+    async fn notify_if_disk_full(&self, error: &ImportError) {
+        let ImportError::Io { source, .. } = error else {
+            return;
+        };
+        if source.kind() != io::ErrorKind::StorageFull {
+            return;
+        }
+        let Some(topic) = &self.operator_topic else {
+            return;
+        };
+        let mut notified = self.disk_full_notified.lock().await;
+        if *notified {
+            return;
+        }
+        if let Err(notify_error) = self
+            .notifier
+            .send(topic, "arr: disk full", &error.to_string())
+            .await
+        {
+            tracing::warn!(%notify_error, "broken notification failed");
+            return;
+        }
+        *notified = true;
+    }
+
+    async fn clear_disk_full(&self) {
+        *self.disk_full_notified.lock().await = false;
+    }
+
     async fn import_one(
         &self,
         database: &Db,
@@ -292,12 +355,37 @@ impl ImportAction {
             waived = waiver.is_some(),
             "imported"
         );
+        self.notify_imported(
+            database,
+            "movie",
+            pending.movie_id,
+            &title_with_year(&pending.title, pending.year),
+        )
+        .await;
         Ok(Some(Outcome::new(
             format!("grab {} downloaded, not imported", pending.grab_id),
             format!("imported {path_text}"),
         )))
     }
 
+    /// §9.5 *imported*: the only good-news notification, sent to the title's
+    /// owners alone. A failure to reach ntfy must not fail the import, which
+    /// has already succeeded.
+    async fn notify_imported(&self, database: &Db, title_kind: &str, title_id: i64, title: &str) {
+        let topics = match owner_topics(database, title_kind, title_id).await {
+            Ok(topics) => topics,
+            Err(error) => {
+                tracing::warn!(%error, "could not load owners for imported notification");
+                return;
+            }
+        };
+        for topic in topics {
+            if let Err(error) = self.notifier.send(&topic, title, "imported").await {
+                tracing::warn!(%error, topic, "imported notification failed");
+            }
+        }
+    }
+
     /// The torrent's files as safe local paths, or `None` when Transmission
     /// no longer has the torrent.
     ///
@@ -423,15 +511,7 @@ impl ImportAction {
         if imports.is_empty() {
             // Everything the pack holds is already on disk. Nothing to
             // place; the grab is settled.
-            sqlx::query!(
-                "UPDATE grabs
-                 SET    state = 'imported',
-                        imported_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
-                 WHERE  id = ?",
-                pending.grab_id
-            )
-            .execute(database.pool())
-            .await?;
+            mark_grab_imported(database, pending.grab_id).await?;
             self.forget_probes(&paths).await;
             return Ok(Some(Outcome::new(
                 format!("grab {} downloaded, not imported", pending.grab_id),
@@ -443,17 +523,16 @@ impl ImportAction {
             .place_episodes(database, pending, &loaded.root_path, imports)
             .await?;
 
-        sqlx::query!(
-            "UPDATE grabs
-             SET    state = 'imported',
-                    imported_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
-             WHERE  id = ?",
-            pending.grab_id
-        )
-        .execute(database.pool())
-        .await?;
+        mark_grab_imported(database, pending.grab_id).await?;
         self.forget_probes(&paths).await;
         self.refresh_jellyfin().await;
+        self.notify_imported(
+            database,
+            "series",
+            pending.series_id,
+            &title_with_year(&pending.series_title, pending.series_year),
+        )
+        .await;
         Ok(Some(Outcome::new(
             format!("grab {} downloaded, not imported", pending.grab_id),
             format!(
@@ -646,6 +725,47 @@ impl Action for ImportAction {
     }
 }
 
+/// Settle a TV grab as imported, whether or not any file was placed — a
+/// pack entirely already on disk still needs its grab marked done.
+async fn mark_grab_imported(database: &Db, grab_id: i64) -> Result<(), ImportError> {
+    sqlx::query!(
+        "UPDATE grabs
+         SET    state = 'imported',
+                imported_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
+         WHERE  id = ?",
+        grab_id
+    )
+    .execute(database.pool())
+    .await?;
+    Ok(())
+}
+
+/// A notification title: the bare title, or with the release year appended.
+fn title_with_year(title: &str, year: Option<i64>) -> String {
+    match year {
+        Some(year) => format!("{title} ({year})"),
+        None => title.to_string(),
+    }
+}
+
+/// The ntfy topics of a title's owners (§4.3, §9.5), movie or series alike.
+async fn owner_topics(
+    database: &Db,
+    title_kind: &str,
+    title_id: i64,
+) -> Result<Vec<String>, ImportError> {
+    Ok(sqlx::query_scalar!(
+        r#"SELECT o.ntfy_topic AS "ntfy_topic!: String"
+           FROM   owners o
+           JOIN   title_owners t ON t.owner_id = o.id
+           WHERE  t.title_kind = ? AND t.title_id = ?"#,
+        title_kind,
+        title_id
+    )
+    .fetch_all(database.pool())
+    .await?)
+}
+
 /// Settle a placed file into the rows: the `media_files` record (§4), the
 /// grab and the movie. The upsert on path is the crash seam — a re-run after
 /// a death between the link and here converges instead of erroring.
@@ -745,6 +865,7 @@ struct PendingTvImport {
     episode_id: Option<i64>,
     season_id: i64,
     season_number: i64,
+    series_id: i64,
     series_tmdb_id: i64,
     series_title: String,
     series_year: Option<i64>,
@@ -780,6 +901,7 @@ async fn pending_tv_imports(database: &Db) -> Result<Vec<PendingTvImport>, Impor
                  e.id       AS "episode_id!: i64",
                  se.id      AS "season_id!: i64",
                  se.number  AS "season_number!: i64",
+                 s.id       AS "series_id!: i64",
                  s.tmdb_id  AS "series_tmdb_id!: i64",
                  s.title    AS "series_title!: String",
                  s.year     AS "series_year",
@@ -802,6 +924,7 @@ async fn pending_tv_imports(database: &Db) -> Result<Vec<PendingTvImport>, Impor
         episode_id: Some(row.episode_id),
         season_id: row.season_id,
         season_number: row.season_number,
+        series_id: row.series_id,
         series_tmdb_id: row.series_tmdb_id,
         series_title: row.series_title,
         series_year: row.series_year,
@@ -815,6 +938,7 @@ async fn pending_tv_imports(database: &Db) -> Result<Vec<PendingTvImport>, Impor
                  g.infohash AS "infohash!: String",
                  se.id      AS "season_id!: i64",
                  se.number  AS "season_number!: i64",
+                 s.id       AS "series_id!: i64",
                  s.tmdb_id  AS "series_tmdb_id!: i64",
                  s.title    AS "series_title!: String",
                  s.year     AS "series_year",
@@ -836,6 +960,7 @@ async fn pending_tv_imports(database: &Db) -> Result<Vec<PendingTvImport>, Impor
         episode_id: None,
         season_id: row.season_id,
         season_number: row.season_number,
+        series_id: row.series_id,
         series_tmdb_id: row.series_tmdb_id,
         series_title: row.series_title,
         series_year: row.series_year,
@@ -1114,6 +1239,44 @@ mod tests {
         ]
     }"#;
 
+    /// Tags a title with one owner, so a test can assert an *imported*
+    /// notification reaches that owner's topic alone (§9.5).
+    async fn insert_owner(database: &Db, title_kind: &str, title_id: i64, name: &str, topic: &str) {
+        let owner_id = sqlx::query("INSERT INTO owners (name, ntfy_topic) VALUES (?, ?)")
+            .bind(name)
+            .bind(topic)
+            .execute(database.pool())
+            .await
+            .unwrap()
+            .last_insert_rowid();
+        sqlx::query("INSERT INTO title_owners (title_kind, title_id, owner_id) VALUES (?, ?, ?)")
+            .bind(title_kind)
+            .bind(title_id)
+            .bind(owner_id)
+            .execute(database.pool())
+            .await
+            .unwrap();
+    }
+
+    async fn start_ntfy_server() -> MockServer {
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .respond_with(ResponseTemplate::new(200))
+            .mount(&server)
+            .await;
+        server
+    }
+
+    async fn start_jellyfin_server() -> MockServer {
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .and(path("/Library/Refresh"))
+            .respond_with(ResponseTemplate::new(204))
+            .mount(&server)
+            .await;
+        server
+    }
+
     struct Harness {
         _dir: tempfile::TempDir,
         database: Db,
@@ -1122,6 +1285,7 @@ mod tests {
         action: ImportAction,
         _server: MockServer,
         jellyfin_server: MockServer,
+        ntfy_server: MockServer,
     }
 
     /// An `ffprobe` stand-in: canned JSON for media, a `tty` document for the
@@ -1192,6 +1356,7 @@ mod tests {
         .execute(database.pool())
         .await
         .unwrap();
+        insert_owner(&database, "movie", 1, "Alice", "alice-topic").await;
 
         let server = MockServer::start().await;
         Mock::given(method("POST"))
@@ -1206,19 +1371,18 @@ mod tests {
             .mount(&server)
             .await;
 
-        let jellyfin_server = MockServer::start().await;
-        Mock::given(method("POST"))
-            .and(path("/Library/Refresh"))
-            .respond_with(ResponseTemplate::new(204))
-            .mount(&jellyfin_server)
-            .await;
+        let jellyfin_server = start_jellyfin_server().await;
+        let ntfy_server = start_ntfy_server().await;
 
         let prober = Prober::new().with_binary(fake_ffprobe(dir.path(), media_json));
         let jellyfin = JellyfinClient::new(jellyfin_server.uri(), None).unwrap();
+        let notifier = Notifier::new(ntfy_server.uri()).unwrap();
         let action = ImportAction::new(
             TransmissionClient::new(&server.uri()).unwrap(),
             prober,
             jellyfin,
+            notifier,
+            Some("operator-topic".to_string()),
         );
 
         Harness {
@@ -1229,6 +1393,7 @@ mod tests {
             action,
             _server: server,
             jellyfin_server,
+            ntfy_server,
         }
     }
 
@@ -1283,6 +1448,14 @@ mod tests {
             1,
             "§7.5: one refresh call at the end of a successful import"
         );
+
+        // The issue's acceptance case: an import notifies only that title's
+        // owners (§9.5), never the operator topic.
+        let notifications = h.ntfy_server.received_requests().await.unwrap();
+        assert_eq!(notifications.len(), 1);
+        assert_eq!(notifications[0].url.path(), "/alice-topic");
+        let body = String::from_utf8(notifications[0].body.clone()).unwrap();
+        assert!(body.contains("Dune: Part Two (2024)"), "{body}");
     }
 
     /// §5.3 through §5.7: Profile 5 is a hard fail — blacklisted, grab
@@ -1322,6 +1495,38 @@ mod tests {
             movie_state, "missing",
             "the gap reopens for the next candidate"
         );
+
+        assert!(
+            h.ntfy_server.received_requests().await.unwrap().is_empty(),
+            "§9.5: a hard fail is not notified"
+        );
+    }
+
+    /// §9.5 *broken*: a full disk notifies the operator once, not every
+    /// tick, and re-arms once space frees up.
+    #[tokio::test]
+    async fn disk_full_notifies_the_operator_once_until_it_clears() {
+        let h = harness(HDR10_PROBE).await;
+        let error = ImportError::Io {
+            action: "hardlink into",
+            path: PathBuf::from("/mnt/media/x.mkv"),
+            source: io::Error::from(io::ErrorKind::StorageFull),
+        };
+
+        h.action.notify_if_disk_full(&error).await;
+        h.action.notify_if_disk_full(&error).await;
+
+        let notifications = h.ntfy_server.received_requests().await.unwrap();
+        assert_eq!(notifications.len(), 1, "debounced while still full");
+        assert_eq!(notifications[0].url.path(), "/operator-topic");
+
+        h.action.clear_disk_full().await;
+        h.action.notify_if_disk_full(&error).await;
+        assert_eq!(
+            h.ntfy_server.received_requests().await.unwrap().len(),
+            2,
+            "re-arms once space frees up"
+        );
     }
 
     /// §5.7 soft fail: watchable but not what was asked. It imports, and the
@@ -1507,6 +1712,8 @@ mod tests {
             TransmissionClient::new("http://127.0.0.1:1").unwrap(),
             Prober::new().with_binary(&script),
             JellyfinClient::new("http://127.0.0.1:1", None).unwrap(),
+            Notifier::new("http://127.0.0.1:1").unwrap(),
+            None,
         );
         let paths = vec![media];
 
@@ -1552,6 +1759,8 @@ mod tests {
             TransmissionClient::new("http://127.0.0.1:1").unwrap(),
             Prober::new().with_binary(&script),
             JellyfinClient::new("http://127.0.0.1:1", None).unwrap(),
+            Notifier::new("http://127.0.0.1:1").unwrap(),
+            None,
         );
         let paths = vec![media];
 
@@ -1604,6 +1813,7 @@ mod tests {
         library: PathBuf,
         action: ImportAction,
         _server: MockServer,
+        ntfy_server: MockServer,
     }
 
     /// A downloaded season-pack grab for Fallout S01E01-E02, its two files
@@ -1670,6 +1880,7 @@ mod tests {
         .execute(database.pool())
         .await
         .unwrap();
+        insert_owner(&database, "series", 1, "Bob", "bob-topic").await;
 
         let server = MockServer::start().await;
         Mock::given(method("POST"))
@@ -1686,19 +1897,18 @@ mod tests {
             })))
             .mount(&server)
             .await;
-        let jellyfin_server = MockServer::start().await;
-        Mock::given(method("POST"))
-            .and(path("/Library/Refresh"))
-            .respond_with(ResponseTemplate::new(204))
-            .mount(&jellyfin_server)
-            .await;
+        let jellyfin_server = start_jellyfin_server().await;
+        let ntfy_server = start_ntfy_server().await;
 
         let prober = Prober::new().with_binary(fake_ffprobe(dir.path(), media_json));
         let jellyfin = JellyfinClient::new(jellyfin_server.uri(), None).unwrap();
+        let notifier = Notifier::new(ntfy_server.uri()).unwrap();
         let action = ImportAction::new(
             TransmissionClient::new(&server.uri()).unwrap(),
             prober,
             jellyfin,
+            notifier,
+            Some("operator-topic".to_string()),
         );
 
         TvHarness {
@@ -1708,6 +1918,7 @@ mod tests {
             library,
             action,
             _server: server,
+            ntfy_server,
         }
     }
 
@@ -1746,6 +1957,16 @@ mod tests {
             h.downloads.join("Fallout.S01/Fallout.S01E01.mkv").is_file(),
             "§7.3: the torrent keeps seeding"
         );
+
+        let notifications = h.ntfy_server.received_requests().await.unwrap();
+        assert_eq!(
+            notifications.len(),
+            1,
+            "one notification per grab, not per episode"
+        );
+        assert_eq!(notifications[0].url.path(), "/bob-topic");
+        let body = String::from_utf8(notifications[0].body.clone()).unwrap();
+        assert!(body.contains("Fallout (2024)"), "{body}");
     }
 
     /// The fourth acceptance case: a pack containing an episode already on

crates/arr-daemon/src/main.rs

@@ -1,10 +1,13 @@
 //! arr — reconcile loop and process entry point. See DESIGN.md §8.
 
+mod attention;
+mod broken;
 mod config;
 mod grab;
 mod import;
 mod indexers;
 mod jellyfin;
+mod notify;
 mod reaper;
 pub mod reconcile;
 mod rss;
@@ -18,9 +21,12 @@ use arr_api::{AppState, Upstreams};
 use arr_compat::CompatState;
 use arr_db::Db;
 use arr_meta::TmdbClient;
+use attention::AttentionAction;
+use broken::BrokenAction;
 use config::Config;
 use grab::{GrabAction, SeedingLimits, SeedingRules};
 use import::ImportAction;
+use notify::Notifier;
 use reaper::ReaperAction;
 use reconcile::{ReconcileLoop, Tick};
 use rss::RssAction;
@@ -85,6 +91,8 @@ enum Error {
     Transmission(#[from] arr_dl::Error),
     #[error("jellyfin client: {0}")]
     Jellyfin(#[from] jellyfin::Error),
+    #[error("ntfy client: {0}")]
+    Notify(#[from] notify::NotifyError),
     #[error("bind {addr}: {source}")]
     Bind {
         addr: std::net::SocketAddr,
@@ -111,7 +119,8 @@ async fn run() -> Result<(), Error> {
     } else {
         None
     };
-    let reconcile = reconcile_loop(&database, &config, &transmission, tmdb.as_ref())?;
+    let notifier = Notifier::new(config.ntfy_url.clone())?;
+    let reconcile = reconcile_loop(&database, &config, &transmission, tmdb.as_ref(), &notifier)?;
 
     // Jellyseerr's Radarr shim (DESIGN.md §9.4) reads the same database and
     // needs its own TMDB client for `movie/lookup`.
@@ -176,6 +185,7 @@ fn reconcile_loop(
     config: &Config,
     transmission: &arr_dl::TransmissionClient,
     tmdb: Option<&Arc<TmdbClient>>,
+    notifier: &Notifier,
 ) -> Result<ReconcileLoop, Error> {
     let mut reconcile = ReconcileLoop::new(database.clone());
     let seeding = SeedingRules::new(
@@ -252,8 +262,42 @@ fn reconcile_loop(
     // imported on this tick.
     reconcile = reconcile.register(
         Tick::Reconcile,
-        ImportAction::new(transmission.clone(), arr_probe::Prober::new(), jellyfin),
+        ImportAction::new(
+            transmission.clone(),
+            arr_probe::Prober::new(),
+            jellyfin,
+            notifier.clone(),
+            config.ntfy_operator_topic.clone(),
+        ),
     );
+
+    // §9.5 *needs a decision* and *broken* both go to the operator alone;
+    // without a topic configured there is nowhere to send them.
+    if let Some(operator_topic) = &config.ntfy_operator_topic {
+        reconcile = reconcile.register(
+            Tick::Reconcile,
+            AttentionAction::new(notifier.clone(), operator_topic.clone()),
+        );
+        let broken_upstreams = broken::Upstreams {
+            prowlarr_url: config.prowlarr_url.clone(),
+            prowlarr_api_key: config.prowlarr_api_key.clone(),
+            transmission_url: config.transmission_url.clone(),
+            tmdb_url: config
+                .tmdb_url
+                .clone()
+                .unwrap_or_else(|| arr_api::DEFAULT_TMDB_URL.to_string()),
+            tmdb_api_key: config.tmdb_api_key.clone(),
+        };
+        reconcile = reconcile.register(
+            Tick::Reconcile,
+            BrokenAction::new(broken_upstreams, notifier.clone(), operator_topic.clone()),
+        );
+    } else {
+        tracing::warn!(
+            "ARR_NTFY_OPERATOR_TOPIC is not configured: needs-a-decision and broken notifications are disabled"
+        );
+    }
+
     Ok(reconcile.register(Tick::Reaper, ReaperAction::new(transmission.clone())))
 }
 

crates/arr-daemon/src/notify.rs

@@ -0,0 +1,105 @@
+//! ntfy delivery for the three DESIGN.md §9.5 events: imported (to a title's
+//! owners), needs-a-decision and broken (both to the operator alone). See
+//! `attention.rs` and `broken.rs` for what decides *when* to send.
+
+use std::time::Duration;
+
+use reqwest::Client;
+
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
+
+#[derive(Debug, thiserror::Error)]
+pub enum NotifyError {
+    #[error("request: {0}")]
+    Request(#[from] reqwest::Error),
+    #[error("ntfy returned {0}")]
+    Status(reqwest::StatusCode),
+}
+
+/// A client for posting to one ntfy server, many topics.
+#[derive(Debug, Clone)]
+pub struct Notifier {
+    client: Client,
+    base_url: String,
+}
+
+impl Notifier {
+    pub fn new(base_url: impl Into<String>) -> Result<Self, NotifyError> {
+        let client = Client::builder().timeout(REQUEST_TIMEOUT).build()?;
+        Ok(Self {
+            client,
+            base_url: base_url.into(),
+        })
+    }
+
+    /// Post a plain-text message to one topic. The title is the message's
+    /// first line rather than ntfy's `Title` header — that header must be
+    /// ASCII, and titles here are TMDB titles in whatever language they were
+    /// released.
+    pub async fn send(&self, topic: &str, title: &str, body: &str) -> Result<(), NotifyError> {
+        let url = format!("{}/{topic}", self.base_url.trim_end_matches('/'));
+        let response = self
+            .client
+            .post(url)
+            .body(format!("{title}\n{body}"))
+            .send()
+            .await?;
+        if !response.status().is_success() {
+            return Err(NotifyError::Status(response.status()));
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used)]
+mod tests {
+    use wiremock::matchers::{body_string, method, path};
+    use wiremock::{Mock, MockServer, ResponseTemplate};
+
+    use super::*;
+
+    #[tokio::test]
+    async fn send_posts_title_and_body_to_the_topic() {
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .and(path("/owner-topic"))
+            .and(body_string("Dune: Part Two (2024)\nimported"))
+            .respond_with(ResponseTemplate::new(200))
+            .mount(&server)
+            .await;
+
+        let notifier = Notifier::new(server.uri()).unwrap();
+
+        notifier
+            .send("owner-topic", "Dune: Part Two (2024)", "imported")
+            .await
+            .unwrap();
+    }
+
+    #[tokio::test]
+    async fn a_non_success_status_is_an_error() {
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .respond_with(ResponseTemplate::new(500))
+            .mount(&server)
+            .await;
+
+        let notifier = Notifier::new(server.uri()).unwrap();
+
+        assert!(matches!(
+            notifier.send("topic", "title", "body").await,
+            Err(NotifyError::Status(_))
+        ));
+    }
+
+    #[tokio::test]
+    async fn an_unreachable_server_is_an_error_the_caller_can_swallow() {
+        let notifier = Notifier::new("http://127.0.0.1:1").unwrap();
+
+        assert!(matches!(
+            notifier.send("topic", "title", "body").await,
+            Err(NotifyError::Request(_))
+        ));
+    }
+}