diff --git a/.sqlx/query-67719bd71584183785d05f28095debebac93f27189194b680dca887dea725dd9.json b/.sqlx/query-7963cce11a588f8b8697e596da24802e4a5556c388b84fe7a6102444a887c2cc.json similarity index 62% rename from .sqlx/query-67719bd71584183785d05f28095debebac93f27189194b680dca887dea725dd9.json rename to .sqlx/query-7963cce11a588f8b8697e596da24802e4a5556c388b84fe7a6102444a887c2cc.json index cc0cf1f..21eaf64 100644 --- a/.sqlx/query-67719bd71584183785d05f28095debebac93f27189194b680dca887dea725dd9.json +++ b/.sqlx/query-7963cce11a588f8b8697e596da24802e4a5556c388b84fe7a6102444a887c2cc.json @@ -1,6 +1,6 @@ { "db_name": "SQLite", - "query": "\n SELECT g.id AS \"grab_id!: i64\",\n g.infohash AS \"infohash!: String\",\n se.id AS \"season_id!: i64\",\n se.number AS \"season_number!: i64\",\n s.tmdb_id AS \"series_tmdb_id!: i64\",\n s.title AS \"series_title!: String\",\n s.year AS \"series_year\",\n s.original_language,\n r.name AS \"release_name!: String\"\n FROM grabs g\n JOIN seasons se ON se.id = g.target_id\n JOIN series s ON s.id = se.series_id\n JOIN releases r ON r.id = g.release_id\n WHERE g.state = 'downloaded' AND g.target_kind = 'season'\n ORDER BY g.id\n ", + "query": "\n SELECT g.id AS \"grab_id!: i64\",\n g.infohash AS \"infohash!: String\",\n se.id AS \"season_id!: i64\",\n se.number AS \"season_number!: i64\",\n s.id AS \"series_id!: i64\",\n s.tmdb_id AS \"series_tmdb_id!: i64\",\n s.title AS \"series_title!: String\",\n s.year AS \"series_year\",\n s.original_language,\n r.name AS \"release_name!: String\"\n FROM grabs g\n JOIN seasons se ON se.id = g.target_id\n JOIN series s ON s.id = se.series_id\n JOIN releases r ON r.id = g.release_id\n WHERE g.state = 'downloaded' AND g.target_kind = 'season'\n ORDER BY g.id\n ", "describe": { "columns": [ { @@ -24,28 +24,33 @@ "type_info": "Integer" }, { - "name": "series_tmdb_id!: i64", + "name": "series_id!: i64", "ordinal": 4, "type_info": "Integer" }, { - "name": "series_title!: String", + "name": "series_tmdb_id!: i64", "ordinal": 5, + "type_info": "Integer" + }, + { + "name": "series_title!: String", + "ordinal": 6, "type_info": "Text" }, { "name": "series_year", - "ordinal": 6, + "ordinal": 7, "type_info": "Integer" }, { "name": "original_language", - "ordinal": 7, + "ordinal": 8, "type_info": "Text" }, { "name": "release_name!: String", - "ordinal": 8, + "ordinal": 9, "type_info": "Text" } ], @@ -59,10 +64,11 @@ false, false, false, + false, true, true, false ] }, - "hash": "67719bd71584183785d05f28095debebac93f27189194b680dca887dea725dd9" + "hash": "7963cce11a588f8b8697e596da24802e4a5556c388b84fe7a6102444a887c2cc" } diff --git a/.sqlx/query-a71c62a2e9cb2ba19f37427347dbf37d194fed7b272371131f07b3ac4eb0bf33.json b/.sqlx/query-7f94d0bad8dd346c5e930c2606c8643e732850b0a1ec331e216c3aedf6987ad4.json similarity index 64% rename from .sqlx/query-a71c62a2e9cb2ba19f37427347dbf37d194fed7b272371131f07b3ac4eb0bf33.json rename to .sqlx/query-7f94d0bad8dd346c5e930c2606c8643e732850b0a1ec331e216c3aedf6987ad4.json index 565a89a..cb7b214 100644 --- a/.sqlx/query-a71c62a2e9cb2ba19f37427347dbf37d194fed7b272371131f07b3ac4eb0bf33.json +++ b/.sqlx/query-7f94d0bad8dd346c5e930c2606c8643e732850b0a1ec331e216c3aedf6987ad4.json @@ -1,6 +1,6 @@ { "db_name": "SQLite", - "query": "\n SELECT g.id AS \"grab_id!: i64\",\n g.infohash AS \"infohash!: String\",\n e.id AS \"episode_id!: i64\",\n se.id AS \"season_id!: i64\",\n se.number AS \"season_number!: i64\",\n s.tmdb_id AS \"series_tmdb_id!: i64\",\n s.title AS \"series_title!: String\",\n s.year AS \"series_year\",\n s.original_language,\n r.name AS \"release_name!: String\"\n FROM grabs g\n JOIN episodes e ON e.id = g.target_id\n JOIN seasons se ON se.id = e.season_id\n JOIN series s ON s.id = se.series_id\n JOIN releases r ON r.id = g.release_id\n WHERE g.state = 'downloaded' AND g.target_kind = 'episode'\n ORDER BY g.id\n ", + "query": "\n SELECT g.id AS \"grab_id!: i64\",\n g.infohash AS \"infohash!: String\",\n e.id AS \"episode_id!: i64\",\n se.id AS \"season_id!: i64\",\n se.number AS \"season_number!: i64\",\n s.id AS \"series_id!: i64\",\n s.tmdb_id AS \"series_tmdb_id!: i64\",\n s.title AS \"series_title!: String\",\n s.year AS \"series_year\",\n s.original_language,\n r.name AS \"release_name!: String\"\n FROM grabs g\n JOIN episodes e ON e.id = g.target_id\n JOIN seasons se ON se.id = e.season_id\n JOIN series s ON s.id = se.series_id\n JOIN releases r ON r.id = g.release_id\n WHERE g.state = 'downloaded' AND g.target_kind = 'episode'\n ORDER BY g.id\n ", "describe": { "columns": [ { @@ -29,28 +29,33 @@ "type_info": "Integer" }, { - "name": "series_tmdb_id!: i64", + "name": "series_id!: i64", "ordinal": 5, "type_info": "Integer" }, { - "name": "series_title!: String", + "name": "series_tmdb_id!: i64", "ordinal": 6, + "type_info": "Integer" + }, + { + "name": "series_title!: String", + "ordinal": 7, "type_info": "Text" }, { "name": "series_year", - "ordinal": 7, + "ordinal": 8, "type_info": "Integer" }, { "name": "original_language", - "ordinal": 8, + "ordinal": 9, "type_info": "Text" }, { "name": "release_name!: String", - "ordinal": 9, + "ordinal": 10, "type_info": "Text" } ], @@ -65,10 +70,11 @@ false, false, false, + false, true, true, false ] }, - "hash": "a71c62a2e9cb2ba19f37427347dbf37d194fed7b272371131f07b3ac4eb0bf33" + "hash": "7f94d0bad8dd346c5e930c2606c8643e732850b0a1ec331e216c3aedf6987ad4" } diff --git a/.sqlx/query-9bad020cbe19c8c5d5f0888c1f21cfd20b095d56b961b67e8599c3a0e8b9770e.json b/.sqlx/query-9bad020cbe19c8c5d5f0888c1f21cfd20b095d56b961b67e8599c3a0e8b9770e.json new file mode 100644 index 0000000..73f638d --- /dev/null +++ b/.sqlx/query-9bad020cbe19c8c5d5f0888c1f21cfd20b095d56b961b67e8599c3a0e8b9770e.json @@ -0,0 +1,32 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT id AS \"id!: i64\", title AS \"title!: String\", year\n FROM movies\n WHERE id IN (\n SELECT m.id\n FROM movies m\n JOIN roots root ON root.id = m.root_id\n WHERE root.audience = 'kids'\n AND m.wanted = 1 AND m.blocked = 0 AND m.state = 'missing'\n AND m.search_attempts > 0\n AND NOT EXISTS (\n SELECT 1 FROM movie_releases mr\n JOIN releases r ON r.id = mr.release_id\n WHERE mr.movie_id = m.id AND r.verdict IN ('eligible', 'waived')\n )\n )\n ", + "describe": { + "columns": [ + { + "name": "id!: i64", + "ordinal": 0, + "type_info": "Integer" + }, + { + "name": "title!: String", + "ordinal": 1, + "type_info": "Text" + }, + { + "name": "year", + "ordinal": 2, + "type_info": "Integer" + } + ], + "parameters": { + "Right": 0 + }, + "nullable": [ + true, + false, + true + ] + }, + "hash": "9bad020cbe19c8c5d5f0888c1f21cfd20b095d56b961b67e8599c3a0e8b9770e" +} diff --git a/.sqlx/query-ae085c6b2b9f7b207bdd839384123195a45e264af3fd8e87408b7f65b1e0cc42.json b/.sqlx/query-ae085c6b2b9f7b207bdd839384123195a45e264af3fd8e87408b7f65b1e0cc42.json new file mode 100644 index 0000000..bbe24ef --- /dev/null +++ b/.sqlx/query-ae085c6b2b9f7b207bdd839384123195a45e264af3fd8e87408b7f65b1e0cc42.json @@ -0,0 +1,32 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT id AS \"id!: i64\", title AS \"title!: String\", year\n FROM movies\n WHERE (SELECT count(DISTINCT g.release_id)\n FROM grabs g\n WHERE g.target_kind = 'movie' AND g.target_id = movies.id\n AND g.state = 'failed') >= 2\n ", + "describe": { + "columns": [ + { + "name": "id!: i64", + "ordinal": 0, + "type_info": "Integer" + }, + { + "name": "title!: String", + "ordinal": 1, + "type_info": "Text" + }, + { + "name": "year", + "ordinal": 2, + "type_info": "Integer" + } + ], + "parameters": { + "Right": 0 + }, + "nullable": [ + false, + false, + true + ] + }, + "hash": "ae085c6b2b9f7b207bdd839384123195a45e264af3fd8e87408b7f65b1e0cc42" +} diff --git a/.sqlx/query-f5a42a1fcc0361f58b7be7906962efa4b9a08d4aaaf8a6883ee9f949d6a5cfe7.json b/.sqlx/query-f5a42a1fcc0361f58b7be7906962efa4b9a08d4aaaf8a6883ee9f949d6a5cfe7.json new file mode 100644 index 0000000..4d78665 --- /dev/null +++ b/.sqlx/query-f5a42a1fcc0361f58b7be7906962efa4b9a08d4aaaf8a6883ee9f949d6a5cfe7.json @@ -0,0 +1,20 @@ +{ + "db_name": "SQLite", + "query": "SELECT o.ntfy_topic AS \"ntfy_topic!: String\"\n FROM owners o\n JOIN title_owners t ON t.owner_id = o.id\n WHERE t.title_kind = ? AND t.title_id = ?", + "describe": { + "columns": [ + { + "name": "ntfy_topic!: String", + "ordinal": 0, + "type_info": "Text" + } + ], + "parameters": { + "Right": 2 + }, + "nullable": [ + false + ] + }, + "hash": "f5a42a1fcc0361f58b7be7906962efa4b9a08d4aaaf8a6883ee9f949d6a5cfe7" +} diff --git a/crates/arr-daemon/src/attention.rs b/crates/arr-daemon/src/attention.rs new file mode 100644 index 0000000..a417fbc --- /dev/null +++ b/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>>, +} + +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, sqlx::Error> { + let mut queued: HashMap = 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) -> 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); + } +} diff --git a/crates/arr-daemon/src/broken.rs b/crates/arr-daemon/src/broken.rs new file mode 100644 index 0000000..015d86a --- /dev/null +++ b/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, + pub transmission_url: String, + pub tmdb_url: String, + pub tmdb_api_key: Option, +} + +#[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>>, +} + +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 { + 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 { + 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"); + } +} diff --git a/crates/arr-daemon/src/config.rs b/crates/arr-daemon/src/config.rs index 7a99cbf..8b0d30d 100644 --- a/crates/arr-daemon/src/config.rs +++ b/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, #[serde(default)] ntfy_url: Option, + #[serde(default)] + ntfy_operator_topic: Option, } #[derive(Debug, Clone, Copy, PartialEq, Deserialize)] @@ -127,6 +130,7 @@ pub struct EnvOverrides { pub jellyfin_url: Option, pub jellyfin_api_key: Option, pub ntfy_url: Option, + pub ntfy_operator_topic: Option, } 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, 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, } 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 { diff --git a/crates/arr-daemon/src/import.rs b/crates/arr-daemon/src/import.rs index 3c872d2..7ec1c0d 100644 --- a/crates/arr-daemon/src/import.rs +++ b/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, + /// 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>, /// 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, + ) -> 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) -> 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, 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, season_id: i64, season_number: i64, + series_id: i64, series_tmdb_id: i64, series_title: String, series_year: Option, @@ -780,6 +901,7 @@ async fn pending_tv_imports(database: &Db) -> Result, 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, 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, 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, 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 diff --git a/crates/arr-daemon/src/main.rs b/crates/arr-daemon/src/main.rs index 95d4dee..24e9147 100644 --- a/crates/arr-daemon/src/main.rs +++ b/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(), ¬ifier)?; // 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>, + notifier: &Notifier, ) -> Result { 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()))) } diff --git a/crates/arr-daemon/src/notify.rs b/crates/arr-daemon/src/notify.rs new file mode 100644 index 0000000..25c7335 --- /dev/null +++ b/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) -> Result { + 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(_)) + )); + } +}