arr/crates/arr-daemon/src/broken.rs

//! §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");
    }
}