From 039d24a5c31b2e86d09479046ef85f1c797a22b8 Mon Sep 17 00:00:00 2001
From: grabowski <berwn@buildfor.life>
Date: Sat, 12 Sep 2026 00:22:03 +0200
Subject: [PATCH] fix: init ntfy only in the collection leader

Every uvicorn worker ran the notification_state DDL at startup; on
Postgres the losers of that race get UniqueViolation on pg_type and the
whole init was skipped (notifications off). Only the leader publishes, so
only the leader initialises, after election. The DDL also tolerates a
concurrent creator now: on failure it verifies the table exists instead
of giving up.
---
 src/notify.py  | 23 ++++++++++------
 src/web_api.py | 72 ++++++++++++++++++++++++++++----------------------
 2 files changed, 56 insertions(+), 39 deletions(-)

diff --git a/src/notify.py b/src/notify.py
index 2668033..62be094 100644
--- a/src/notify.py
+++ b/src/notify.py
@@ -136,14 +136,21 @@ class NotificationState:
     def _ensure(self) -> None:
         from sqlalchemy import text
 
-        with self.engine.begin() as conn:
-            conn.execute(
-                text(
-                    "CREATE TABLE IF NOT EXISTS notification_state ("
-                    "key VARCHAR(64) PRIMARY KEY, state VARCHAR(16) NOT NULL, "
-                    "value NUMERIC(8,3), updated_at TIMESTAMP NOT NULL)"
-                )
-            )
+        ddl = (
+            "CREATE TABLE IF NOT EXISTS notification_state ("
+            "key VARCHAR(64) PRIMARY KEY, state VARCHAR(16) NOT NULL, "
+            "value NUMERIC(8,3), updated_at TIMESTAMP NOT NULL)"
+        )
+        try:
+            with self.engine.begin() as conn:
+                conn.execute(text(ddl))
+        except Exception as error:
+            # Postgres: two sessions racing CREATE TABLE IF NOT EXISTS can
+            # both pass the existence check; the loser fails with a unique
+            # violation on pg_type. The table exists either way; verify.
+            with self.engine.connect() as conn:
+                conn.execute(text("SELECT 1 FROM notification_state WHERE 1=0"))
+            logger.debug(f"notification_state DDL raced, table present: {error}")
 
     def get(self, key: str) -> Optional[str]:
         from sqlalchemy import text
diff --git a/src/web_api.py b/src/web_api.py
index d6d5ffb..d5719c1 100644
--- a/src/web_api.py
+++ b/src/web_api.py
@@ -175,37 +175,6 @@ async def lifespan(app: FastAPI):
         app_state["forecast_store"] = None
         logger.error(f"Forecast history store init failed: {e}")
 
-    # Public flood notifications over ntfy (leader only, transitions only).
-    # Off unless NTFY_SERVER is set; then state lives in notification_state
-    # next to the measurements so a restart never re-sends.
-    app_state["notify"] = None
-    if Config.NTFY_SERVER:
-        try:
-            from . import notify as notify_mod
-
-            store = app_state.get("forecast_store")
-            if store and not store.engine:
-                store.connect()
-            state = (
-                notify_mod.NotificationState(store.engine, store.db_type)
-                if store and store.engine
-                else notify_mod.InMemoryState()
-            )
-            app_state["notify"] = (
-                notify_mod.NtfyPublisher(
-                    Config.NTFY_SERVER,
-                    prefix=Config.NTFY_TOPIC_PREFIX,
-                    token=Config.NTFY_TOKEN or None,
-                    dashboard_url=Config.PUBLIC_URL,
-                ),
-                state,
-            )
-            logger.info(
-                f"ntfy notifications: {Config.NTFY_SERVER} topics {Config.NTFY_TOPIC_PREFIX}-*"
-            )
-        except Exception as e:
-            logger.error(f"ntfy init failed (notifications off): {e}")
-
     # Initialize HII/ThaiWater collector (rainfall + backup water level)
     try:
         from .hii_collector import create_collector_from_config
@@ -241,7 +210,9 @@ async def lifespan(app: FastAPI):
     app_state["leader_lock"] = _acquire_collection_leadership(
         Config.COLLECTION_LEADER_PORT
     )
+    app_state["notify"] = None
     if app_state["leader_lock"]:
+        app_state["notify"] = _init_notifications()
         app_state["scraping_task"] = asyncio.create_task(background_scraping_task())
         logger.info("This worker is the background-collection leader")
     else:
@@ -327,6 +298,45 @@ async def _persist_rain():
         logger.warning(f"rain persistence failed: {e}")
 
 
+def _init_notifications():
+    """Publisher + persisted state for ntfy, or None if off/unavailable.
+
+    Called only by the collection leader: it is the one process that
+    publishes, so the notification_state DDL runs exactly once per host.
+    """
+    if not Config.NTFY_SERVER:
+        return None
+    try:
+        from . import notify as notify_mod
+
+        store = app_state.get("forecast_store")
+        if store and not store.engine:
+            store.connect()
+        state = (
+            notify_mod.NotificationState(store.engine, store.db_type)
+            if store and store.engine
+            else notify_mod.InMemoryState()
+        )
+        if isinstance(state, notify_mod.InMemoryState):
+            logger.warning(
+                "ntfy: no SQL store; notification state is in-memory "
+                "(a restart may re-send the current level)"
+            )
+        publisher = notify_mod.NtfyPublisher(
+            Config.NTFY_SERVER,
+            prefix=Config.NTFY_TOPIC_PREFIX,
+            token=Config.NTFY_TOKEN or None,
+            dashboard_url=Config.PUBLIC_URL,
+        )
+        logger.info(
+            f"ntfy notifications: {Config.NTFY_SERVER} topics {Config.NTFY_TOPIC_PREFIX}-*"
+        )
+        return publisher, state
+    except Exception as e:
+        logger.error(f"ntfy init failed (notifications off): {e}")
+        return None
+
+
 async def _notify_transitions():
     """Publish flood/outlook/feed transitions to ntfy (leader only, fail-safe)."""
     cfg = app_state.get("notify")