feat: public flood notifications over self-hosted ntfy

Anyone can now get push alerts on their phone without an account: the
monitor publishes to an ntfy server (one Go binary, ~30 MB RSS) and
subscribers pick topics in the free iOS/Android/web app.

Semantics are transitions, never state. One message when a gauge crosses
its warning or danger threshold, one all-clear when it drops back (0.10 m
hysteresis), nothing while it sits above. A three-day flood is two
messages; a quiet season is zero. Topics: ping-warning / ping-danger
(basin digest), ping-<station>-warning / -danger, ping-p1-outlook (opt-in:
model P(warning within 24 h) at P.1 rises through 50 %, clears below 25 %,
message says it is experimental), ping-status (feed stale >= 3 h /
recovered). Priority 5 on danger so it rings through Do Not Disturb.

src/notify.py runs once per collection cycle in the API process (leader
only, after the forecast precompute, same data the dashboard shows).
Last-sent state lives in a notification_state table so a restart never
re-sends; a failed publish leaves state untouched so the crossing is
retried next cycle instead of lost. Off unless NTFY_SERVER is set.

Dashboard: a "Get alerts" button (only when configured) opens a panel
with the server, per-topic cards, ntfy:// deep links and web links, app
store links and a disclaimer. EN + TH. GET /api/notifications feeds it.

scripts/install_ntfy.sh: .deb install, server.yml (loopback listen,
anonymous read, token-only write scoped to ping-*, 72 h cache, signup/
login/metrics off, tight visitor limits), systemd, user + token, .env.
Verified against ntfy 2.28.0: anon publish 403, token publish 200, token
on foreign topic 403, anon read 200, and a seeded crossing through the
real _notify_transitions path arrived in the topic with priority, tags,
click and action button. docs/NOTIFICATIONS.md has the deployment and
reverse-proxy notes. Tests: 10 for the state machine (159 total).

tests/test_notify.py

@@ -0,0 +1,218 @@
+"""ntfy notification state machine: transitions only, hysteresis, restart-safe."""
+
+import datetime
+
+import pytest
+
+from src import notify
+
+
+class FakePublisher(notify.NtfyPublisher):
+    def __init__(self):
+        super().__init__("http://ntfy.test", prefix="ping")
+        self.sent = []
+
+    def publish(self, n):
+        self.sent.append(n)
+        return True
+
+
+@pytest.fixture
+def pub():
+    return FakePublisher()
+
+
+def _reading(code, level, ts="2026-09-24T12:00:00"):
+    return {"station_code": code, "water_level": level, "timestamp": ts}
+
+
+def _fc(p, peak=None):
+    return [
+        {
+            "station_code": "P.1",
+            "horizon_hours": 24,
+            "p_warning": p,
+            "predicted_max_level": peak,
+            "source": "model",
+        }
+    ]
+
+
+NOW = datetime.datetime(2026, 9, 24, 12, 30)
+
+
+def topics(pub):
+    return [n.topic for n in pub.sent]
+
+
+def test_quiet_river_sends_nothing(pub):
+    state = notify.InMemoryState()
+    for h in range(48):
+        notify.evaluate(
+            [_reading("P.1", 1.6), _reading("P.103", 3.2)],
+            _fc(0.01),
+            state,
+            pub,
+            now=NOW,
+        )
+    assert pub.sent == []
+
+
+def test_warning_crossing_once_then_silence_then_clear(pub):
+    state = notify.InMemoryState()
+    # rising through 3.70 (P.1 warning)
+    notify.evaluate([_reading("P.1", 3.65)], [], state, pub, now=NOW)
+    assert pub.sent == []
+    notify.evaluate([_reading("P.1", 3.72)], [], state, pub, now=NOW)
+    assert topics(pub) == ["ping-p1-warning", "ping-warning"]
+    assert pub.sent[0].priority == 4 and "3.72 m" in pub.sent[0].message
+    # stays above: no repeats for many hours
+    for level in (3.80, 3.95, 4.05, 3.90, 3.75):
+        notify.evaluate([_reading("P.1", level)], [], state, pub, now=NOW)
+    assert len(pub.sent) == 2
+    # dips to 3.65: within hysteresis, still no message
+    notify.evaluate([_reading("P.1", 3.65)], [], state, pub, now=NOW)
+    assert len(pub.sent) == 2
+    # 3.55: clear
+    notify.evaluate([_reading("P.1", 3.55)], [], state, pub, now=NOW)
+    assert topics(pub)[2:] == ["ping-p1-warning", "ping-warning"]
+    assert "back to normal" in pub.sent[2].title
+
+
+def test_danger_escalation_and_deescalation(pub):
+    state = notify.InMemoryState()
+    notify.evaluate([_reading("P.1", 3.9)], [], state, pub, now=NOW)  # warning
+    notify.evaluate(
+        [_reading("P.1", 4.25)], [], state, pub, now=NOW
+    )  # danger (>= 4.20)
+    assert topics(pub) == [
+        "ping-p1-warning",
+        "ping-warning",
+        "ping-p1-danger",
+        "ping-danger",
+    ]
+    assert pub.sent[2].priority == 5
+    notify.evaluate(
+        [_reading("P.1", 4.15)], [], state, pub, now=NOW
+    )  # hysteresis: still danger
+    assert len(pub.sent) == 4
+    notify.evaluate([_reading("P.1", 4.05)], [], state, pub, now=NOW)  # back to warning
+    assert topics(pub)[4:] == ["ping-p1-danger", "ping-warning"]
+    assert "below danger" in pub.sent[4].title
+
+
+def test_jump_straight_to_danger(pub):
+    state = notify.InMemoryState()
+    notify.evaluate(
+        [_reading("P.103", 7.0)], [], state, pub, now=NOW
+    )  # P.103 danger 6.75
+    assert topics(pub) == ["ping-p103-danger", "ping-danger"]
+
+
+def test_basin_digest_groups_stations(pub):
+    state = notify.InMemoryState()
+    notify.evaluate(
+        [_reading("P.1", 3.8), _reading("P.103", 6.0), _reading("P.67", 1.0)],
+        [],
+        state,
+        pub,
+        now=NOW,
+    )
+    basin = [n for n in pub.sent if n.topic == "ping-warning"]
+    assert len(basin) == 1 and "P.1" in basin[0].message and "P.103" in basin[0].message
+
+
+def test_outlook_on_off_with_hysteresis(pub):
+    state = notify.InMemoryState()
+    r = [_reading("P.1", 2.9)]
+    notify.evaluate(r, _fc(0.30), state, pub, now=NOW)
+    assert pub.sent == []
+    notify.evaluate(r, _fc(0.55, 3.9), state, pub, now=NOW)
+    assert topics(pub) == ["ping-p1-outlook"]
+    assert "55%" in pub.sent[0].message and "3.90 m" in pub.sent[0].message
+    assert "not an official warning" in pub.sent[0].message
+    notify.evaluate(
+        r, _fc(0.40), state, pub, now=NOW
+    )  # between OFF and ON: stays on, silent
+    assert len(pub.sent) == 1
+    notify.evaluate(r, _fc(0.20), state, pub, now=NOW)
+    assert len(pub.sent) == 2 and "easing" in pub.sent[1].title
+
+
+def test_heuristic_forecast_ignored(pub):
+    state = notify.InMemoryState()
+    fc = [
+        {
+            "station_code": "P.1",
+            "horizon_hours": 24,
+            "p_warning": 0.9,
+            "source": "heuristic",
+        }
+    ]
+    notify.evaluate([_reading("P.1", 2.0)], fc, state, pub, now=NOW)
+    assert pub.sent == []
+
+
+def test_stale_feed_and_recovery(pub):
+    state = notify.InMemoryState()
+    notify.evaluate(
+        [_reading("P.1", 1.6, "2026-09-24T12:00:00")], [], state, pub, now=NOW
+    )
+    assert pub.sent == []
+    later = NOW + datetime.timedelta(hours=4)
+    notify.evaluate(
+        [_reading("P.1", 1.6, "2026-09-24T12:00:00")], [], state, pub, now=later
+    )
+    assert topics(pub) == ["ping-status"] and "stale" in pub.sent[0].title
+    notify.evaluate(
+        [_reading("P.1", 1.6, "2026-09-24T12:00:00")],
+        [],
+        state,
+        pub,
+        now=later + datetime.timedelta(hours=1),
+    )
+    assert len(pub.sent) == 1  # still stale, no repeat
+    notify.evaluate(
+        [_reading("P.1", 1.6, "2026-09-24T17:00:00")],
+        [],
+        state,
+        pub,
+        now=later + datetime.timedelta(hours=1),
+    )
+    assert len(pub.sent) == 2 and "recovered" in pub.sent[1].title
+
+
+def test_state_survives_restart_via_sql(tmp_path, pub):
+    from sqlalchemy import create_engine
+
+    eng = create_engine(f"sqlite:///{tmp_path / 'n.db'}")
+    state = notify.NotificationState(eng, "sqlite")
+    notify.evaluate([_reading("P.1", 3.8)], [], state, pub, now=NOW)
+    assert len(pub.sent) == 2
+    # "restart": new state object on the same DB, same reading -> nothing re-sent
+    state2 = notify.NotificationState(eng, "sqlite")
+    notify.evaluate([_reading("P.1", 3.8)], [], state2, pub, now=NOW)
+    assert len(pub.sent) == 2
+
+
+def test_publish_failure_does_not_advance_state():
+    """If ntfy is down the transition must be retried next cycle, not lost."""
+
+    class Down(notify.NtfyPublisher):
+        def __init__(self):
+            super().__init__("http://ntfy.test")
+            self.calls = 0
+
+        def publish(self, n):
+            self.calls += 1
+            return False
+
+    pub = Down()
+    state = notify.InMemoryState()
+    notify.evaluate([_reading("P.1", 3.8)], [], state, pub, now=NOW)
+    assert pub.calls == 2 and state.get("level:P.1") is None
+    # next cycle, ntfy back: the crossing is delivered
+    good = FakePublisher()
+    notify.evaluate([_reading("P.1", 3.8)], [], state, good, now=NOW)
+    assert topics(good) == ["ping-p1-warning", "ping-warning"]
+    assert state.get("level:P.1") == "warning"