Northern-Thailand-Ping-River-Monitor/rust/collector/src/notify.rs

//! Public flood notifications over ntfy, and the same transitions to a
//! Matrix room: the port of src/notify.py, run by the collector when
//! FORECAST_IN_COLLECTOR=true. (Matrix replaces the old `run.py
//! --alert-check` cron job, which repeated an alert on every run while a
//! level stayed high.)
//!
//! Same topics, texts, hysteresis, capacity guard, outlook and persisted
//! state keys (`notification_state`) as Python, so a cutover neither repeats
//! nor drops a message. One change: for stations whose HII/ThaiWater twin is
//! the same sensor, a reading newer than RID's (RID's report often lags 2-10 h
//! at night) is used for the level alerts and the feed check. The feed is
//! "stale" only when neither source has a reading for 3 h; RID lagging alone
//! is reported separately after RID_STALE_ALERT_HOURS.

use crate::config::MatrixConfig;
use crate::db::{self, LatestReading};
use anyhow::Result;
use chrono::{Duration, NaiveDateTime};
use ping_forecast::Forecast;
use ping_forecast::config::config;
use ping_forecast::hii::FillInfo;
use sqlx::PgPool;
use std::collections::HashMap;

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Source {
    Rid,
    /// HII/ThaiWater telemetry of the same sensor, offset to the RID datum.
    Hii,
}

#[derive(Clone, Debug, PartialEq)]
pub struct Reading {
    pub station_code: String,
    pub timestamp: NaiveDateTime,
    pub water_level: Option<f64>,
    pub discharge_percent: Option<f64>,
    pub source: Source,
    /// The station's newest RID reading with a level (for the RID lag check).
    pub rid_timestamp: Option<NaiveDateTime>,
}

#[derive(Clone, Debug, PartialEq)]
pub struct Notification {
    pub topic: String,
    pub title: String,
    pub message: String,
    pub priority: u8,
    pub tags: Vec<String>,
}

/// A notification and the state to record once it was delivered.
#[derive(Clone, Debug, PartialEq)]
pub struct Planned {
    pub notification: Notification,
    pub state: Option<(String, String, Option<f64>)>,
}

/// The readings the notifier judges: the newest RID row per station, replaced
/// by the newest HII fill when that is newer (or RID's newest row has no level).
pub fn readings_with_hii(rid: &[LatestReading], fills: &[FillInfo]) -> Vec<Reading> {
    let by_code: HashMap<&str, &FillInfo> =
        fills.iter().map(|f| (f.station_code.as_str(), f)).collect();
    rid.iter()
        .map(|r| {
            let fill = by_code.get(r.station_code.as_str()).and_then(|f| f.latest);
            match fill {
                Some((t, level))
                    if t > r.timestamp || (t == r.timestamp && r.water_level.is_none()) =>
                {
                    Reading {
                        station_code: r.station_code.clone(),
                        timestamp: t,
                        water_level: Some(level),
                        // capacity drifts slowly: RID's figure if it is recent
                        discharge_percent: r
                            .discharge_percent
                            .filter(|_| t - r.timestamp <= Duration::hours(6)),
                        source: Source::Hii,
                        rid_timestamp: r.water_level.map(|_| r.timestamp),
                    }
                }
                _ => Reading {
                    station_code: r.station_code.clone(),
                    timestamp: r.timestamp,
                    water_level: r.water_level,
                    discharge_percent: r.discharge_percent,
                    source: Source::Rid,
                    rid_timestamp: r.water_level.map(|_| r.timestamp),
                },
            }
        })
        .collect()
}

/// `(a - b).total_seconds() / 3600` as Python computes it.
fn hours_between(a: NaiveDateTime, b: NaiveDateTime) -> f64 {
    (a - b)
        .num_microseconds()
        .map_or(f64::INFINITY, |us| us as f64 / 1e6 / 3600.0)
}

fn slug(code: &str) -> String {
    code.to_lowercase().replace('.', "")
}

/// 'clear' | 'warning' | 'danger', with hysteresis on the way down.
fn level_state(level: f64, warn: f64, danger: f64, prev: &str) -> &'static str {
    let margin = config().notify.clear_margin_m;
    if level >= danger {
        return "danger";
    }
    if level >= warn {
        if prev == "danger" && level >= danger - margin {
            return "danger";
        }
        return "warning";
    }
    if (prev == "warning" || prev == "danger") && level >= warn - margin {
        return "warning";
    }
    "clear"
}

pub struct Settings {
    pub prefix: String,
    /// 0 = off
    pub rid_stale_alert_hours: f64,
}

impl Settings {
    fn topic(&self, parts: &[&str]) -> String {
        std::iter::once(self.prefix.as_str())
            .chain(parts.iter().copied())
            .collect::<Vec<_>>()
            .join("-")
    }
}

fn note(n: Notification) -> Planned {
    Planned {
        notification: n,
        state: None,
    }
}

fn n(
    topic: String,
    title: impl Into<String>,
    message: String,
    priority: u8,
    tags: &[&str],
) -> Notification {
    Notification {
        topic,
        title: title.into(),
        message,
        priority,
        tags: tags.iter().map(|t| t.to_string()).collect(),
    }
}

/// `notify.evaluate` as a plan: every transition against the stored states.
pub fn evaluate(
    readings: &[Reading],
    forecasts: &[Forecast],
    states: &HashMap<String, String>,
    settings: &Settings,
    now: NaiveDateTime,
) -> Vec<Planned> {
    let cfg = config();
    let nc = &cfg.notify;
    let state = |k: &str| states.get(k).map(String::as_str);
    let mut plan: Vec<Planned> = Vec::new();
    let (mut b_danger, mut b_warning, mut b_clear) = (vec![], vec![], vec![]);
    let mut latest: Option<NaiveDateTime> = None;
    let mut latest_rid: Option<NaiveDateTime> = None;

    for r in readings {
        latest_rid = latest_rid.max(r.rid_timestamp);
        let Some(level) = r.water_level else {
            continue;
        };
        let ts = r.timestamp;
        latest = latest.max(Some(ts));
        let code = r.station_code.as_str();
        let (warn, danger) = cfg.thresholds(code);
        let key = format!("level:{code}");
        let prev = state(&key).unwrap_or("clear");
        let cur = level_state(level, warn, danger, prev);
        if cur != "clear"
            && prev == "clear"
            && !nc.capacity_guard_exempt.iter().any(|c| c == code)
            && let Some(pct) = r.discharge_percent
            && pct < nc.capacity_guard_min_pct
        {
            tracing::info!(
                "{code}: level {level:.2} m >= {warn:.2} but only {pct:.0}% capacity; \
                 threshold looks stale, not alerting"
            );
            continue;
        }
        if cur == prev {
            continue;
        }
        let name = nc.station_names.get(code).map_or(code, String::as_str);
        let slug = slug(code);
        let mut when = ts.format("%d %b %H:%M").to_string();
        if r.source == Source::Hii {
            when.push_str(" (ThaiWater telemetry; the RID report is not in yet)");
        }
        let (topic_kind, title, message, priority, tag) = match (cur, prev) {
            ("danger", _) => {
                b_danger.push(format!("{code} {level:.2} m"));
                (
                    "danger",
                    format!("DANGER level at {code}"),
                    format!(
                        "{name}: {level:.2} m at {when}, above the danger level of {danger:.2} m."
                    ),
                    5,
                    "rotating_light",
                )
            }
            ("warning", "danger") => {
                b_clear.push(format!("{code} below danger ({level:.2} m)"));
                (
                    "danger",
                    format!("{code} back below danger level"),
                    format!(
                        "{name}: {level:.2} m at {when}; still above the warning level of {warn:.2} m."
                    ),
                    3,
                    "arrow_down",
                )
            }
            ("warning", _) => {
                b_warning.push(format!("{code} {level:.2} m"));
                (
                    "warning",
                    format!("Warning level at {code}"),
                    format!(
                        "{name}: {level:.2} m at {when}, above the warning level of {warn:.2} m."
                    ),
                    4,
                    "warning",
                )
            }
            _ => {
                b_clear.push(format!("{code} normal ({level:.2} m)"));
                (
                    "warning",
                    format!("{code} back to normal"),
                    format!(
                        "{name}: {level:.2} m at {when}, below the warning level of {warn:.2} m."
                    ),
                    2,
                    "white_check_mark",
                )
            }
        };
        plan.push(Planned {
            notification: n(
                settings.topic(&[&slug, topic_kind]),
                title,
                message,
                priority,
                &[tag, code],
            ),
            // remembered only once delivered, so ntfy being down means a retry
            state: Some((key, cur.to_string(), Some(level))),
        });
    }

    if !b_danger.is_empty() {
        plan.push(note(n(
            settings.topic(&["danger"]),
            "Ping River: danger level reached",
            b_danger.join("; "),
            5,
            &["rotating_light"],
        )));
    }
    if !b_warning.is_empty() {
        plan.push(note(n(
            settings.topic(&["warning"]),
            "Ping River: warning level reached",
            b_warning.join("; "),
            4,
            &["warning"],
        )));
    }
    if !b_clear.is_empty() {
        plan.push(note(n(
            settings.topic(&["warning"]),
            "Ping River: levels falling",
            b_clear.join("; "),
            2,
            &["white_check_mark"],
        )));
    }

    // model outlook for the city gauge (opt-in topic, experimental)
    if let Some(p1) = forecasts.iter().find(|f| {
        f.station_code == "P.1" && f.horizon_hours == nc.outlook_horizon && f.source == "model"
    }) {
        let p = p1.p_warning;
        let key = "outlook:P.1";
        let prev = state(key).unwrap_or("off");
        let cur = if p >= nc.outlook_on || (prev == "on" && p >= nc.outlook_off) {
            "on"
        } else {
            "off"
        };
        if cur != prev {
            let (warn, _) = cfg.thresholds("P.1");
            let notification = if cur == "on" {
                n(
                    settings.topic(&["p1-outlook"]),
                    "Early warning: Chiang Mai flood risk rising",
                    format!(
                        "The forecast model gives a {:.0}% chance that Nawarat Bridge (P.1) reaches \
                         {warn:.2} m within 24 h (expected peak {:.2} m). Experimental model output, \
                         not an official warning; follow ThaiWater/TMD for official alerts.",
                        p * 100.0,
                        p1.predicted_max_level
                    ),
                    4,
                    &["crystal_ball"],
                )
            } else {
                n(
                    settings.topic(&["p1-outlook"]),
                    "Chiang Mai flood risk easing",
                    format!(
                        "The model's 24 h probability of reaching {warn:.2} m at P.1 has dropped to {:.0}%.",
                        p * 100.0
                    ),
                    2,
                    &["crystal_ball"],
                )
            };
            plan.push(Planned {
                notification,
                state: Some((key.into(), cur.into(), Some(p))),
            });
        }
    }

    // feed health: stale only when no source has a recent reading
    let mut feed_stale = false;
    if let Some(latest) = latest {
        let age_h = hours_between(now, latest);
        let prev = state("feed").unwrap_or("ok");
        feed_stale = age_h >= nc.feed_stale_after_h;
        let cur = if feed_stale { "stale" } else { "ok" };
        if cur != prev {
            let notification = if feed_stale {
                n(
                    settings.topic(&["status"]),
                    "Ping River monitor: gauge feed stale",
                    format!(
                        "No new readings for {age_h:.0} h (last {}). Levels and forecasts on the \
                         dashboard are not current.",
                        latest.format("%d %b %H:%M")
                    ),
                    3,
                    &["hourglass"],
                )
            } else {
                n(
                    settings.topic(&["status"]),
                    "Ping River monitor: feed recovered",
                    format!(
                        "Readings are current again (latest {}).",
                        latest.format("%d %b %H:%M")
                    ),
                    2,
                    &["white_check_mark"],
                )
            };
            plan.push(Planned {
                notification,
                state: Some(("feed".into(), cur.into(), Some(age_h))),
            });
        }
    }

    // RID itself far behind while ThaiWater keeps the levels current
    if settings.rid_stale_alert_hours > 0.0
        && let Some(rid_ts) = latest_rid
    {
        let age_h = hours_between(now, rid_ts);
        let prev = state("feed:rid").unwrap_or("ok");
        let cur = if age_h >= settings.rid_stale_alert_hours {
            "stale"
        } else {
            "ok"
        };
        // while the whole feed is stale, that message covers RID too
        if cur != prev && !(cur == "stale" && feed_stale) {
            let notification = if cur == "stale" {
                n(
                    settings.topic(&["status"]),
                    "Ping River monitor: RID reports delayed",
                    format!(
                        "No RID gauge report for {age_h:.0} h (last {}). Levels and alerts continue \
                         from ThaiWater telemetry for the stations it mirrors.",
                        rid_ts.format("%d %b %H:%M")
                    ),
                    3,
                    &["hourglass"],
                )
            } else {
                n(
                    settings.topic(&["status"]),
                    "Ping River monitor: RID reports current again",
                    format!("Latest RID report {}.", rid_ts.format("%d %b %H:%M")),
                    2,
                    &["white_check_mark"],
                )
            };
            plan.push(Planned {
                notification,
                state: Some(("feed:rid".into(), cur.into(), Some(age_h))),
            });
        }
    }
    plan
}

/// ntfy publishing target.
pub struct Ntfy {
    pub publish_url: String,
    pub token: Option<String>,
}

/// State keys of the Matrix channel: the same transitions as ntfy, tracked
/// separately so a failed delivery on one channel is retried on that one
/// only.
const MATRIX_PREFIX: &str = "matrix:";

/// Posts to ntfy and Matrix and records delivered transitions.
pub struct Notifier {
    pub http: reqwest::Client,
    pub ntfy: Option<Ntfy>,
    pub matrix: Option<MatrixConfig>,
    pub dashboard_url: String,
    pub settings: Settings,
}

/// ntfy tag shortcodes the notifications use, as emoji for Matrix.
fn emoji(tags: &[String]) -> &'static str {
    match tags.first().map(String::as_str) {
        Some("rotating_light") => "🚨",
        Some("warning") => "⚠️",
        Some("arrow_down") => "⬇️",
        Some("white_check_mark") => "✅",
        Some("crystal_ball") => "🔮",
        Some("hourglass") => "⏳",
        _ => "📊",
    }
}

/// The Matrix message: the ntfy title and text, in the small Markdown
/// subset `markdown_to_html` renders.
pub fn matrix_text(n: &Notification, dashboard_url: &str) -> String {
    format!(
        "{} **{}**\n{}\n\n📈 Dashboard: {dashboard_url}",
        emoji(&n.tags),
        n.title,
        n.message
    )
}

/// `**bold**` removed, for Matrix's plain-text `body`.
pub fn strip_markdown(text: &str) -> String {
    text.replace("**", "")
}

/// `alerting.markdown_to_matrix_html`: escape, `**bold**`, bare URLs as
/// links, newlines as `<br/>`.
pub fn markdown_to_html(text: &str) -> String {
    let escaped = text
        .replace('&', "&amp;")
        .replace('<', "&lt;")
        .replace('>', "&gt;");
    let mut bold = String::new();
    let mut parts = escaped.split("**").peekable();
    let mut open = false;
    while let Some(part) = parts.next() {
        bold.push_str(part);
        if parts.peek().is_some() {
            bold.push_str(if open { "</strong>" } else { "<strong>" });
            open = !open;
        }
    }
    if open {
        // an unpaired ** stays literal
        if let Some(i) = bold.rfind("<strong>") {
            bold.replace_range(i..i + "<strong>".len(), "**");
        }
    }
    let mut out = String::new();
    let mut rest = bold.as_str();
    while let Some(i) = rest
        .find("http://")
        .into_iter()
        .chain(rest.find("https://"))
        .min()
    {
        out.push_str(&rest[..i]);
        let url_len = rest[i..]
            .find(|c: char| c.is_whitespace() || c == '<')
            .unwrap_or(rest.len() - i);
        let url = &rest[i..i + url_len];
        out.push_str(&format!("<a href=\"{url}\">{url}</a>"));
        rest = &rest[i + url_len..];
    }
    out.push_str(rest);
    out.replace('\n', "<br/>")
}

impl Notifier {
    pub async fn publish(&self, ntfy: &Ntfy, n: &Notification) -> bool {
        let mut req = self
            .http
            .post(format!(
                "{}/{}",
                ntfy.publish_url.trim_end_matches('/'),
                n.topic
            ))
            .timeout(std::time::Duration::from_secs(10))
            .header("Title", &n.title)
            .header("Priority", n.priority.to_string())
            .header("Click", &self.dashboard_url)
            .header(
                "Actions",
                format!("view, Open dashboard, {}", self.dashboard_url),
            )
            .body(n.message.clone());
        if !n.tags.is_empty() {
            req = req.header("Tags", n.tags.join(","));
        }
        if let Some(t) = &ntfy.token {
            req = req.bearer_auth(t);
        }
        match req.send().await {
            Ok(r) if r.status().as_u16() < 300 => true,
            Ok(r) => {
                let status = r.status();
                let body = r.text().await.unwrap_or_default();
                tracing::warn!(
                    "ntfy {}: HTTP {status} {}",
                    n.topic,
                    body.chars().take(120).collect::<String>()
                );
                false
            }
            Err(e) => {
                tracing::warn!("ntfy {}: {e}", n.topic);
                false
            }
        }
    }

    /// One Matrix room message (`MatrixNotifier.send_message`).
    pub async fn send_matrix(&self, m: &MatrixConfig, n: &Notification) -> bool {
        static TXN: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
        let text = matrix_text(n, &self.dashboard_url);
        let txn = format!(
            "ping-{}-{}",
            chrono::Utc::now().timestamp_micros(),
            TXN.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
        );
        let url = format!(
            "{}/_matrix/client/v3/rooms/{}/send/m.room.message/{txn}",
            m.homeserver.trim_end_matches('/'),
            m.room_id
        );
        let body = serde_json::json!({
            "msgtype": "m.text",
            "body": strip_markdown(&text),
            "format": "org.matrix.custom.html",
            "formatted_body": markdown_to_html(&text),
        });
        match self
            .http
            .put(url)
            .bearer_auth(&m.access_token)
            .timeout(std::time::Duration::from_secs(10))
            .json(&body)
            .send()
            .await
        {
            Ok(r) if r.status().is_success() => true,
            Ok(r) => {
                let status = r.status();
                let body = r.text().await.unwrap_or_default();
                tracing::warn!(
                    "Matrix: HTTP {status} {}",
                    body.chars().take(120).collect::<String>()
                );
                false
            }
            Err(e) => {
                tracing::warn!("Matrix: {e}");
                false
            }
        }
    }

    /// The Matrix channel's states (keys without the prefix). On first use
    /// they start as a copy of the ntfy states, so what ntfy already
    /// announced is not sent again.
    pub async fn matrix_states(
        &self,
        pool: &PgPool,
        states: &HashMap<String, String>,
        now: NaiveDateTime,
        seed: bool,
    ) -> Result<HashMap<String, String>> {
        let own: HashMap<String, String> = states
            .iter()
            .filter_map(|(k, v)| {
                k.strip_prefix(MATRIX_PREFIX)
                    .map(|k| (k.to_string(), v.clone()))
            })
            .collect();
        if !own.is_empty() {
            return Ok(own);
        }
        let copy: HashMap<String, String> = states
            .iter()
            .filter(|(k, _)| !k.starts_with(MATRIX_PREFIX))
            .map(|(k, v)| (k.clone(), v.clone()))
            .collect();
        if seed {
            for (k, v) in &copy {
                db::set_notification_state(pool, &format!("{MATRIX_PREFIX}{k}"), v, None, now)
                    .await?;
            }
            db::set_notification_state(pool, "matrix:seeded", "yes", None, now).await?;
        }
        Ok(copy)
    }

    /// What the Matrix channel would send: the transitions that carry a
    /// state (per station, outlook, feed). The per-topic summaries ntfy also
    /// gets ("Ping River: warning level reached") repeat those, so the room
    /// does not get them.
    pub fn matrix_plan(
        &self,
        readings: &[Reading],
        forecasts: &[Forecast],
        states: &HashMap<String, String>,
        now: NaiveDateTime,
    ) -> Vec<Planned> {
        evaluate(readings, forecasts, states, &self.settings, now)
            .into_iter()
            .filter(|p| p.state.is_some())
            .collect()
    }

    /// Evaluate and publish; returns the notifications delivered (Matrix
    /// messages with topic "matrix").
    pub async fn run(
        &self,
        pool: &PgPool,
        readings: &[Reading],
        forecasts: &[Forecast],
        now: NaiveDateTime,
    ) -> Result<Vec<Notification>> {
        let states = db::notification_states(pool).await?;
        let mut sent = vec![];
        if let Some(ntfy) = &self.ntfy {
            for p in evaluate(readings, forecasts, &states, &self.settings, now) {
                if !self.publish(ntfy, &p.notification).await {
                    continue;
                }
                if let Some((key, state, value)) = &p.state {
                    db::set_notification_state(pool, key, state, *value, now).await?;
                }
                sent.push(p.notification);
            }
        }
        if let Some(m) = &self.matrix {
            let mstates = self.matrix_states(pool, &states, now, true).await?;
            for p in self.matrix_plan(readings, forecasts, &mstates, now) {
                if !self.send_matrix(m, &p.notification).await {
                    continue;
                }
                if let Some((key, state, value)) = &p.state {
                    let key = format!("{MATRIX_PREFIX}{key}");
                    db::set_notification_state(pool, &key, state, *value, now).await?;
                }
                sent.push(Notification {
                    topic: "matrix".into(),
                    ..p.notification
                });
            }
        }
        Ok(sent)
    }
}

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

    #[test]
    fn matrix_message_rendering() {
        let n = Notification {
            topic: "ping-p1-warning".into(),
            title: "Warning level at P.1".into(),
            message:
                "Nawarat Bridge: 3.75 m at 18 Oct 14:00, above the warning level of 3.70 m <x> & y."
                    .into(),
            priority: 4,
            tags: vec!["warning".into(), "P.1".into()],
        };
        let text = matrix_text(&n, "https://water.buildfor.life/");
        assert_eq!(
            markdown_to_html(&text),
            "⚠️ <strong>Warning level at P.1</strong><br/>Nawarat Bridge: 3.75 m at 18 Oct 14:00, \
             above the warning level of 3.70 m &lt;x&gt; &amp; y.<br/><br/>📈 Dashboard: \
             <a href=\"https://water.buildfor.life/\">https://water.buildfor.life/</a>"
        );
        assert!(strip_markdown(&text).starts_with("⚠️ Warning level at P.1\nNawarat"));
        assert_eq!(markdown_to_html("a ** b"), "a ** b");
    }

    fn now() -> NaiveDateTime {
        NaiveDate::from_ymd_opt(2026, 10, 1)
            .unwrap()
            .and_hms_opt(12, 0, 0)
            .unwrap()
    }

    fn settings() -> Settings {
        Settings {
            prefix: "ping".into(),
            rid_stale_alert_hours: 18.0,
        }
    }

    fn reading(code: &str, level: f64, hours_ago: i64) -> Reading {
        Reading {
            station_code: code.into(),
            timestamp: now() - Duration::hours(hours_ago),
            water_level: Some(level),
            discharge_percent: None,
            source: Source::Rid,
            rid_timestamp: Some(now() - Duration::hours(hours_ago)),
        }
    }

    /// Run `evaluate` and apply the state of every planned transition, as a
    /// successful publish would.
    fn step(
        rs: &[Reading],
        fcs: &[Forecast],
        st: &mut HashMap<String, String>,
        at: NaiveDateTime,
    ) -> Vec<Notification> {
        let plan = evaluate(rs, fcs, st, &settings(), at);
        for p in &plan {
            if let Some((k, s, _)) = &p.state {
                st.insert(k.clone(), s.clone());
            }
        }
        plan.into_iter().map(|p| p.notification).collect()
    }

    fn titles(ns: &[Notification]) -> Vec<(&str, &str)> {
        ns.iter()
            .map(|n| (n.topic.as_str(), n.title.as_str()))
            .collect()
    }

    #[test]
    fn quiet_river_sends_nothing() {
        let mut st = HashMap::new();
        assert!(step(&[reading("P.1", 1.2, 1)], &[], &mut st, now()).is_empty());
    }

    #[test]
    fn warning_once_then_silence_then_clear_with_hysteresis() {
        let mut st = HashMap::new();
        let sent = step(&[reading("P.1", 3.75, 0)], &[], &mut st, now());
        assert_eq!(
            titles(&sent),
            [
                ("ping-p1-warning", "Warning level at P.1"),
                ("ping-warning", "Ping River: warning level reached")
            ]
        );
        assert!(
            sent[0]
                .message
                .starts_with("Nawarat Bridge, Chiang Mai city: 3.75 m at 01 Oct 12:00")
        );
        assert!(step(&[reading("P.1", 3.8, 0)], &[], &mut st, now()).is_empty());
        // within the 0.10 m margin: still warning
        assert!(step(&[reading("P.1", 3.65, 0)], &[], &mut st, now()).is_empty());
        let sent = step(&[reading("P.1", 3.5, 0)], &[], &mut st, now());
        assert_eq!(titles(&sent)[0], ("ping-p1-warning", "P.1 back to normal"));
        assert_eq!(
            titles(&sent)[1],
            ("ping-warning", "Ping River: levels falling")
        );
    }

    #[test]
    fn danger_escalation_and_deescalation() {
        let mut st = HashMap::new();
        step(&[reading("P.1", 3.8, 0)], &[], &mut st, now());
        let sent = step(&[reading("P.1", 4.3, 0)], &[], &mut st, now());
        assert_eq!(titles(&sent)[0], ("ping-p1-danger", "DANGER level at P.1"));
        assert_eq!(sent[0].priority, 5);
        assert!(step(&[reading("P.1", 4.15, 0)], &[], &mut st, now()).is_empty());
        let sent = step(&[reading("P.1", 4.0, 0)], &[], &mut st, now());
        assert_eq!(
            titles(&sent)[0],
            ("ping-p1-danger", "P.1 back below danger level")
        );
    }

    #[test]
    fn capacity_guard_blocks_a_stale_threshold_but_not_p1() {
        let mut st = HashMap::new();
        let mut r = reading("P.77", 4.5, 0);
        r.discharge_percent = Some(22.0);
        assert!(step(&[r.clone()], &[], &mut st, now()).is_empty());
        r.discharge_percent = Some(80.0);
        assert_eq!(step(&[r], &[], &mut st, now()).len(), 2);
        let mut p1 = reading("P.1", 3.8, 0);
        p1.discharge_percent = Some(10.0);
        assert_eq!(step(&[p1], &[], &mut st, now()).len(), 2);
    }

    fn outlook(p: f64) -> Forecast {
        Forecast {
            station_code: "P.1".into(),
            horizon_hours: 24,
            p_warning: p,
            p_danger: 0.0,
            predicted_max_level: 3.9,
            current_level: 3.0,
            as_of: now(),
            model_version: "hgb-v3+test".into(),
            trained_at: None,
            source: "model".into(),
            threshold_warning: 3.7,
            threshold_danger: 4.2,
            stages: None,
        }
    }

    #[test]
    fn outlook_on_off_with_hysteresis_and_heuristic_ignored() {
        let mut st = HashMap::new();
        let r = [reading("P.1", 3.0, 0)];
        let sent = step(&r, &[outlook(0.62)], &mut st, now());
        assert_eq!(
            titles(&sent),
            [(
                "ping-p1-outlook",
                "Early warning: Chiang Mai flood risk rising"
            )]
        );
        assert!(sent[0].message.contains("62% chance") && sent[0].message.contains("3.90 m"));
        assert!(step(&r, &[outlook(0.3)], &mut st, now()).is_empty());
        assert_eq!(
            step(&r, &[outlook(0.2)], &mut st, now())[0].title,
            "Chiang Mai flood risk easing"
        );
        let mut h = outlook(0.9);
        h.source = "heuristic".into();
        assert!(step(&r, &[h], &mut st, now()).is_empty());
    }

    fn rid_row(code: &str, hours_ago: i64, level: Option<f64>, pct: Option<f64>) -> LatestReading {
        LatestReading {
            timestamp: now() - Duration::hours(hours_ago),
            station_code: code.into(),
            water_level: level,
            discharge_percent: pct,
        }
    }

    fn fill(code: &str, hours_ago: i64, level: f64) -> FillInfo {
        FillInfo {
            station_code: code.into(),
            offset: 300.0,
            filled_hours: 1,
            latest: Some((now() - Duration::hours(hours_ago), level)),
        }
    }

    #[test]
    fn hii_reading_replaces_a_lagging_rid_reading() {
        let rid = [
            rid_row("P.1", 5, Some(3.2), Some(40.0)),
            rid_row("P.21", 5, Some(1.0), None),
            rid_row("P.67", 1, None, None),
        ];
        let rs = readings_with_hii(
            &rid,
            &[
                fill("P.1", 0, 3.8),
                fill("P.67", 1, 2.0),
                fill("P.103", 0, 1.0),
            ],
        );
        assert_eq!(rs[0].source, Source::Hii);
        assert_eq!(rs[0].water_level, Some(3.8));
        assert_eq!(rs[0].discharge_percent, Some(40.0)); // RID figure 5 h old
        assert_eq!(rs[1].source, Source::Rid); // no twin
        assert_eq!((rs[2].source, rs[2].water_level), (Source::Hii, Some(2.0))); // RID row had no level
        let mut st = HashMap::new();
        let sent = step(&rs, &[], &mut st, now());
        assert!(
            sent[0]
                .message
                .contains("(ThaiWater telemetry; the RID report is not in yet)")
        );
        // an older HII value never replaces a newer RID one
        let rs = readings_with_hii(
            &[rid_row("P.1", 0, Some(1.0), None)],
            &[fill("P.1", 2, 3.9)],
        );
        assert_eq!(rs[0].source, Source::Rid);
    }

    #[test]
    fn feed_is_stale_only_when_both_sources_are() {
        let mut st = HashMap::new();
        // RID 5 h behind, ThaiWater current: no stale alert (the 2026-10 nights)
        let rs = readings_with_hii(
            &[rid_row("P.1", 5, Some(1.2), None)],
            &[fill("P.1", 0, 1.25)],
        );
        assert!(step(&rs, &[], &mut st, now()).is_empty());
        // both behind: stale, then recovered
        let rs = readings_with_hii(
            &[rid_row("P.1", 5, Some(1.2), None)],
            &[fill("P.1", 4, 1.25)],
        );
        let sent = step(&rs, &[], &mut st, now());
        assert_eq!(
            titles(&sent),
            [("ping-status", "Ping River monitor: gauge feed stale")]
        );
        assert!(
            sent[0]
                .message
                .starts_with("No new readings for 4 h (last 01 Oct 08:00)")
        );
        let rs = readings_with_hii(
            &[rid_row("P.1", 5, Some(1.2), None)],
            &[fill("P.1", 0, 1.25)],
        );
        assert_eq!(
            step(&rs, &[], &mut st, now())[0].title,
            "Ping River monitor: feed recovered"
        );
    }

    #[test]
    fn rid_lagging_a_day_is_reported_once_while_hii_is_current() {
        let mut st = HashMap::new();
        let rs = readings_with_hii(
            &[rid_row("P.1", 20, Some(1.2), None)],
            &[fill("P.1", 0, 1.25)],
        );
        let sent = step(&rs, &[], &mut st, now());
        assert_eq!(
            titles(&sent),
            [("ping-status", "Ping River monitor: RID reports delayed")]
        );
        assert!(step(&rs, &[], &mut st, now()).is_empty());
        let rs = readings_with_hii(&[rid_row("P.1", 0, Some(1.2), None)], &[]);
        assert_eq!(
            step(&rs, &[], &mut st, now())[0].title,
            "Ping River monitor: RID reports current again"
        );
        // whole feed stale: only the general message
        let mut st = HashMap::new();
        let sent = step(&[reading("P.1", 1.2, 30)], &[], &mut st, now());
        assert_eq!(
            titles(&sent),
            [("ping-status", "Ping River monitor: gauge feed stale")]
        );
    }

    #[test]
    fn undelivered_transitions_are_retried() {
        let mut st = HashMap::new();
        let plan = evaluate(&[reading("P.1", 3.8, 0)], &[], &st, &settings(), now());
        assert_eq!(plan[0].state.as_ref().unwrap().1, "warning");
        // nothing recorded (publish failed): the next run plans it again
        let again = evaluate(&[reading("P.1", 3.8, 0)], &[], &st, &settings(), now());
        assert_eq!(plan, again);
        st.insert("level:P.1".into(), "warning".into());
        assert!(
            evaluate(&[reading("P.1", 3.8, 0)], &[], &st, &settings(), now())
                .iter()
                .all(|p| p.notification.topic == "ping-warning")
        );
    }

    /// 652 hourly steps from src/notify.py itself (testdata/gen_notify_cases.py):
    /// the same notifications, word for word, in the same order.
    #[test]
    fn matches_python_notify_on_generated_sequences() {
        use std::io::Read;
        let raw = include_bytes!("../testdata/notify_python_cases.json.gz");
        let mut text = String::new();
        flate2::read::GzDecoder::new(&raw[..])
            .read_to_string(&mut text)
            .unwrap();
        let fixture: serde_json::Value = serde_json::from_str(&text).unwrap();
        let ts = |v: &serde_json::Value| {
            NaiveDateTime::parse_from_str(v.as_str().unwrap(), "%Y-%m-%dT%H:%M:%S").unwrap()
        };
        let mut compared = 0;
        for seq in fixture["sequences"].as_array().unwrap() {
            let mut states: HashMap<String, String> = HashMap::new();
            for step in seq.as_array().unwrap() {
                let readings: Vec<Reading> = step["readings"]
                    .as_array()
                    .unwrap()
                    .iter()
                    .map(|r| {
                        let level = r["water_level"].as_f64();
                        Reading {
                            station_code: r["station_code"].as_str().unwrap().into(),
                            timestamp: ts(&r["timestamp"]),
                            water_level: level,
                            discharge_percent: r["discharge_percent"].as_f64(),
                            source: Source::Rid,
                            rid_timestamp: level.map(|_| ts(&r["timestamp"])),
                        }
                    })
                    .collect();
                let forecasts: Vec<Forecast> = step["forecasts"]
                    .as_array()
                    .unwrap()
                    .iter()
                    .map(|f| Forecast {
                        horizon_hours: f["horizon_hours"].as_u64().unwrap() as u32,
                        p_warning: f["p_warning"].as_f64().unwrap(),
                        predicted_max_level: f["predicted_max_level"].as_f64().unwrap(),
                        source: f["source"].as_str().unwrap().into(),
                        ..outlook(0.0)
                    })
                    .collect();
                let down: Vec<&str> = step["down"]
                    .as_array()
                    .unwrap()
                    .iter()
                    .map(|v| v.as_str().unwrap())
                    .collect();
                let mut delivered = vec![];
                for p in evaluate(
                    &readings,
                    &forecasts,
                    &states,
                    &settings(),
                    ts(&step["now"]),
                ) {
                    if down.contains(&p.notification.topic.as_str()) {
                        continue;
                    }
                    if let Some((k, s, _)) = &p.state {
                        states.insert(k.clone(), s.clone());
                    }
                    delivered.push(p.notification);
                }
                let want: Vec<Notification> = step["sent"]
                    .as_array()
                    .unwrap()
                    .iter()
                    .map(|n| Notification {
                        topic: n["topic"].as_str().unwrap().into(),
                        title: n["title"].as_str().unwrap().into(),
                        message: n["message"].as_str().unwrap().into(),
                        priority: n["priority"].as_u64().unwrap() as u8,
                        tags: n["tags"]
                            .as_array()
                            .unwrap()
                            .iter()
                            .map(|t| t.as_str().unwrap().into())
                            .collect(),
                    })
                    .collect();
                assert_eq!(delivered, want, "at {}", step["now"]);
                compared += want.len();
            }
        }
        assert!(compared > 700, "only {compared} notifications compared");
    }
}