launcher/src/home_assistant.rs

//! Home Assistant REST API client for `power on`/`power off`.
//!
//! Talks to HA via `POST /api/services/<domain>/<service>` and uses
//! `Authorization: Bearer <token>`.  The HA domain for a switch is the
//! part before the first dot in the entity_id (e.g. `switch.dgx1` →
//! domain `switch`, service `turn_off` or `turn_on`).

use anyhow::{Context, Result, anyhow};
use serde_json::json;
use std::sync::OnceLock;
use std::time::Duration;
use tracing::{debug, info, warn};

use crate::config::{Cluster, Config, HomeAssistantConfig};
use crate::entity_cache as liveness;

/// Process-wide shared HA client, built once. The fleet probe rebuilds HA
/// state every 30 s across several clusters — a fresh `Client` per call
/// re-parses the system CA bundle (rustls) and allocates ~1.7 MB of TLS
/// setup each time, ~2 GB/h of churn in the long-running dashboard.
static SHARED_CLIENT: OnceLock<Option<Client>> = OnceLock::new();

/// Entity states are cached because the dashboard probes every 30 s (and a
/// switch must not be asked to authenticate on every tick). HA state changes
/// within 2 min; RouterOS devices are checked at most every 5 min.
const HA_TTL: Duration = Duration::from_secs(120);
const MIKROTIK_TTL: Duration = Duration::from_secs(300);

/// The shared client for long-lived callers (the web dashboard's probe
/// loop); CLI paths (one-shot `power`) may keep using `Client::new`.
pub fn shared_client(config: &Config) -> Option<Client> {
    SHARED_CLIENT
        .get_or_init(|| {
            config
                .home_assistant
                .as_ref()
                .and_then(|ha| Client::new(ha).ok())
        })
        .clone()
}

/// Thin async wrapper around the HA REST API.
///
/// `base_url` is the HA root URL (`http://homeassistant.local:8123`),
/// `token` is the long-lived access token used as bearer auth.
#[derive(Clone)]
pub struct Client {
    base_url: String,
    token: String,
    http: reqwest::Client,
}

impl Client {
    /// Construct a new client.  The reqwest client is built once and
    /// reused so connection pooling works across the parallel calls in
    /// the `power` subcommand.
    pub fn new(cfg: &HomeAssistantConfig) -> Result<Self> {
        let base_url = cfg.url.trim_end_matches('/').to_string();
        if base_url.is_empty() {
            anyhow::bail!("[home_assistant].url must be set");
        }
        let http = reqwest::Client::builder()
            .timeout(Duration::from_secs(10))
            .build()
            .context("building reqwest client")?;
        Ok(Self {
            base_url,
            token: cfg.token.clone(),
            http,
        })
    }

    /// Build `https://.../api/services/<domain>/<service>` for the given
    /// `entity_id`.  Domain is the prefix before the first `.`; pass an
    /// empty fallback for ambiguous input.
    fn endpoint(entity_id: &str, service: &str) -> Result<(String, String)> {
        let (domain, entity) = entity_id.split_once('.').ok_or_else(|| {
            anyhow!(
                "entity_id '{}' has no domain (e.g. 'switch.foo')",
                entity_id
            )
        })?;
        if entity.is_empty() {
            anyhow::bail!("entity_id '{}' has empty entity name", entity_id);
        }
        Ok((
            domain.to_string(),
            format!("/api/services/{}/{}", domain, service),
        ))
    }

    /// POST `/api/services/<domain>/<service>` with `{"entity_id": ...}`.
    async fn call_service(&self, entity_id: &str, service: &str) -> Result<()> {
        let (_domain, path) = Self::endpoint(entity_id, service)?;
        let url = format!("{}{}", self.base_url, path);
        debug!("HA POST {}", url);
        let resp = self
            .http
            .post(&url)
            .bearer_auth(&self.token)
            .json(&json!({ "entity_id": entity_id }))
            .send()
            .await
            .with_context(|| format!("calling HA at {}", url))?;
        let status = resp.status();
        if !status.is_success() {
            let body = resp.text().await.unwrap_or_default();
            anyhow::bail!(
                "HA {} {} returned {}: {}",
                service,
                entity_id,
                status,
                body.trim()
            );
        }
        liveness::invalidate(&format!("ha:{}", entity_id));
        Ok(())
    }

    /// `switch.turn_off` for the given entity.
    pub async fn turn_off(&self, entity_id: &str) -> Result<()> {
        info!("HA turn_off {}", entity_id);
        self.call_service(entity_id, "turn_off").await
    }

    /// `switch.turn_on` for the given entity.
    pub async fn turn_on(&self, entity_id: &str) -> Result<()> {
        info!("HA turn_on {}", entity_id);
        self.call_service(entity_id, "turn_on").await
    }

    /// GET `/api/states/<entity_id>` and return the state string.
    /// Returns `None` if the entity doesn't exist or isn't reachable.
    /// Logs a warning when the HTTP request or JSON parsing fails so
    /// callers (e.g. `entity_is_on`) can distinguish "entity is off"
    /// from "HA is unreachable" via the log.
    pub async fn get_state(&self, entity_id: &str) -> Option<String> {
        let url = format!("{}/api/states/{}", self.base_url, entity_id);
        let resp = match self.http.get(&url).bearer_auth(&self.token).send().await {
            Ok(r) => r,
            Err(e) => {
                warn!("HA GET {} failed: {}", url, e);
                return None;
            }
        };
        if !resp.status().is_success() {
            warn!("HA GET {} returned {}", url, resp.status());
            return None;
        }
        let body: serde_json::Value = match resp.json().await {
            Ok(v) => v,
            Err(e) => {
                warn!("HA GET {} JSON parse failed: {}", url, e);
                return None;
            }
        };
        body.get("state").and_then(|s| s.as_str()).map(String::from)
    }
}

/// HA states for a cluster's `extra_entities` (parallel GETs).
#[derive(Clone, Debug, Default, PartialEq)]
pub struct EntityStates {
    /// A *blocking* entity reports anything other than "on" — including
    /// HA being unreachable (unknown is treated as off, same as the
    /// `health` command has always done). Blocking-off takes the whole
    /// cluster down.
    pub blocking_off: bool,
    /// Non-blocking entities (`blocks = false`) that are not "on" —
    /// tracked and displayed, but they never affect up/down.
    pub soft_off: Vec<String>,
}

/// Probe the cluster's `extra_entities`. No HA config and no `[mikrotik]`
/// backend for an entity → that entity is "nothing to check", not off
/// (legacy semantics). Mikrotik devices (exact name match from
/// [`crate::config`]) are probed through the RouterOS API instead of HA:
/// logged in = on, unreachable/rejected login = off (a dead router takes
/// the cluster down like any other blocking switch).
pub async fn extra_entities_states(config: &Config, cluster: &Cluster) -> EntityStates {
    if cluster.extra_entities.is_empty() {
        return EntityStates::default();
    }
    let ha_client: Option<Client> = shared_client(config);
    let has_ha = ha_client.is_some();
    let mut handles = Vec::new();
    for e in &cluster.extra_entities {
        let id = e.entity().to_string();
        let is_mikrotik = config
            .mikrotik
            .as_ref()
            .is_some_and(|m| m.devices.iter().any(|d| d.name == id.as_str()));
        // A non-mikrotik entity without an HA backend is "nothing to check",
        // not off — only entities with a configured backend flip the verdict.
        if !is_mikrotik && !has_ha {
            continue;
        }
        let config = config.clone();
        let client = ha_client.clone();
        let blocks = e.blocks();
        handles.push(crate::locale::spawn_with_locale(async move {
            (
                id.clone(),
                blocks,
                entity_state(&config, client.as_ref(), &id).await,
            )
        }));
    }
    let mut st = EntityStates::default();
    for h in futures::future::join_all(handles).await {
        let Ok((id, blocks, state)) = h else {
            continue;
        };
        let on = matches!(&state, Some(s) if s == "on");
        if !on {
            if blocks {
                st.blocking_off = true;
            } else {
                st.soft_off.push(id);
            }
        }
    }
    st
}

/// The state of one entity through the correct backend: a `[mikrotik]`
/// device is probed with a RouterOS API login (logged in = "on";
/// unreachable or rejected login = "off", like a switch that does not
/// answer), everything else goes through HA REST.
/// Returns `None` only when the entity has no configured backend at all
/// (unknown — callers keep the "nothing to check ≠ off" semantics).
pub async fn entity_state(
    config: &Config,
    client: Option<&Client>,
    entity_id: &str,
) -> Option<String> {
    if let Some(dev) = config
        .mikrotik
        .as_ref()
        .and_then(|m| m.devices.iter().find(|d| d.name == entity_id))
    {
        let key = format!("mikrotik:{}:{}", dev.name, dev.address);
        if let Some(Some(v)) = liveness::fresh(&key) {
            return Some((if v { "on" } else { "off" }).to_string());
        }
        let v = match crate::mikrotik::ping_device(dev).await {
            Ok(()) => true,
            Err(e) => {
                warn!("RouterOS ping {entity_id} failed: {e}");
                false
            }
        };
        liveness::store(&key, Some(v), MIKROTIK_TTL);
        return Some((if v { "on" } else { "off" }).to_string());
    }
    let client = client?;
    let key = format!("ha:{}", entity_id);
    if let Some(Some(v)) = liveness::fresh(&key) {
        return Some((if v { "on" } else { "off" }).to_string());
    }
    let state = client.get_state(entity_id).await;
    let v = state.as_deref() == Some("on");
    liveness::store(&key, Some(v), HA_TTL);
    Some((if v { "on" } else { "off" }).to_string())
}

/// Parse `switch.dgx1` into `("switch", "dgx1")` so callers can build
/// their own endpoints if needed.  Exposed for the `power` subcommand
/// and for tests.
pub fn split_entity(entity_id: &str) -> Result<(&str, &str)> {
    let (domain, entity) = entity_id.split_once('.').ok_or_else(|| {
        anyhow!(
            "entity_id '{}' has no domain (e.g. 'switch.foo')",
            entity_id
        )
    })?;
    if domain.is_empty() || entity.is_empty() {
        anyhow::bail!("entity_id '{}' has empty domain or entity name", entity_id);
    }
    Ok((domain, entity))
}

/// Emit a deprecation-style warning if a HA token looks suspiciously
/// short.  HA tokens are typically 100+ chars; anything shorter is
/// almost certainly a placeholder.
pub fn warn_if_token_looks_short(token: &str) {
    if token.len() < 20 {
        warn!(
            "[home_assistant].token looks suspiciously short ({} chars); \
             did you paste a placeholder?",
            token.len()
        );
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn mikrotik_entity_unreachable_is_blocking_off() {
        // A `[mikrotik]` device referenced from `extra_entities` is probed
        // through the RouterOS API: unreachable (closed port) = off, and a
        // blocking entity takes the cluster down — the r18-class "dead
        // entity silently skipped" failure mode must not exist.
        let toml = r#"
ssh_user = "u"
[[clusters]]
name = "c1"
nodes = [{ name = "n1", ssh_ip = "10.0.0.1", rank = 0 }]
extra_entities = ["rt"]
[mikrotik]
[[mikrotik.devices]]
name = "rt"
address = "127.0.0.1"
username = "x"
password = "y"
api_port = 1
"#;
        let config: crate::config::Config = toml::from_str(toml).unwrap();
        let cluster = &config.clusters[0];
        let st = extra_entities_states(&config, cluster).await;
        assert!(
            st.blocking_off,
            "unreachable mikrotik device must be blocking-off"
        );
    }

    #[tokio::test]
    async fn mikrotik_device_without_backend_not_reported_off() {
        // No [mikrotik] section, no HA → the entity has no backend at all:
        // "nothing to check", not off (legacy semantics).
        let toml = r#"
ssh_user = "u"
[[clusters]]
name = "c1"
nodes = [{ name = "n1", ssh_ip = "10.0.0.1", rank = 0 }]
extra_entities = ["rt"]
"#;
        let config: crate::config::Config = toml::from_str(toml).unwrap();
        let cluster = &config.clusters[0];
        let st = extra_entities_states(&config, cluster).await;
        assert!(!st.blocking_off);
        assert!(st.soft_off.is_empty());
    }

    #[test]
    fn split_entity_basic() {
        assert_eq!(split_entity("switch.dgx1").unwrap(), ("switch", "dgx1"));
        assert_eq!(
            split_entity("light.living_room").unwrap(),
            ("light", "living_room")
        );
    }

    #[test]
    fn split_entity_no_domain_errors() {
        assert!(split_entity("dgx1").is_err());
    }

    #[test]
    fn split_entity_empty_entity_errors() {
        assert!(split_entity("switch.").is_err());
    }

    #[test]
    fn split_entity_empty_domain_errors() {
        assert!(split_entity(".thing").is_err());
    }

    #[test]
    fn endpoint_extracts_domain() {
        let (_d, path) = Client::endpoint("switch.dgx1", "turn_off").unwrap();
        assert_eq!(path, "/api/services/switch/turn_off");
        let (_d, path) = Client::endpoint("light.kitchen", "turn_on").unwrap();
        assert_eq!(path, "/api/services/light/turn_on");
    }

    #[test]
    fn endpoint_rejects_no_domain() {
        assert!(Client::endpoint("plain", "turn_off").is_err());
    }
}