ai-agent/backend/notify.py

"""
First-class notifications — the ai-agent backend as the homelab's notification
source of truth.

Everything that pushes to the phone now flows through here: the ``notify-done``
skill POSTs ``/api/notify`` (plain pushes) and ``/api/ask`` (choice questions
with tappable buttons), the backend records the event in a JSON sidecar
(``/data/notifications.json``) and *forwards* it to the configured **webhooks**
— typically the Home Assistant webhook registered by the ``ai_agent`` custom
integration (see ``services/ai-agent/integrations/home-assistant/``), which
renders the actual mobile notification and, for asks, POSTs the tapped answer
back to ``/api/ask/{id}/answer``.

Store shape (single JSON file, atomic rewrite, same pattern as ``meta.py``):

    { "webhooks": [ {"id", "name", "url", "events": ["notify","ask"],
                     "enabled": true}, … ],
      "notifications": [ {"id", "kind": "notify"|"ask", "title", "body",
                          "type", "url", "image", "at", "sessionId",
                          "delivered": [{"webhook","ok","status"}],
                          # ask-only:
                          "options", "status", "answer", "answerIndex",
                          "answeredAt", "expiresAt"}, … ] }

The notification log is capped (newest kept). Ask long-polling is in-memory
(one ``threading.Event`` per pending ask) — a backend restart drops the wait,
and the asking script's own timeout handles that.
"""

import json
import os
import pathlib
import re
import threading
import urllib.error
import urllib.request
import uuid
from datetime import datetime, timedelta, timezone

# Default Home Assistant webhook seeded into an empty store: HA runs on the
# host network, so the container reaches it via the docker host-gateway alias.
DEFAULT_HA_WEBHOOK_URL = os.environ.get(
    "HA_WEBHOOK_URL", "http://host.docker.internal:8123/api/webhook/ai_agent")

WEBHOOK_EVENTS = ("notify", "ask")
_ID_RE = re.compile(r"[^a-z0-9-]+")
_MAX_LOG = 1000
_FORWARD_TIMEOUT_S = 8

# How long an ask waits for a tap by default (mirrors ask.sh's default).
DEFAULT_ASK_TIMEOUT_S = 180


def _now_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _slug(s: str) -> str:
    return _ID_RE.sub("-", (s or "").strip().lower()).strip("-")


def valid_webhooks(raw: list) -> list[dict]:
    """Validate + normalize a webhook list; raises ValueError on bad input."""
    out: list[dict] = []
    seen: set[str] = set()
    if not isinstance(raw, list):
        raise ValueError("webhooks must be a list")
    for w in raw:
        if not isinstance(w, dict):
            raise ValueError("each webhook must be an object")
        url = str(w.get("url") or "").strip()
        if not url.startswith(("http://", "https://")):
            raise ValueError(f"webhook url must be http(s): {url!r}")
        name = str(w.get("name") or "").strip() or "webhook"
        wid = _slug(str(w.get("id") or "")) or _slug(name) or "webhook"
        base, n = wid, 2
        while wid in seen:
            wid = f"{base}-{n}"
            n += 1
        seen.add(wid)
        events = [e for e in (w.get("events") or list(WEBHOOK_EVENTS))
                  if e in WEBHOOK_EVENTS]
        if not events:
            raise ValueError(f"webhook {wid}: events must include notify/ask")
        out.append({"id": wid, "name": name, "url": url, "events": events,
                    "enabled": bool(w.get("enabled", True))})
    return out


class NotifyStore:
    """Webhook config + notification/ask log in one JSON sidecar."""

    def __init__(self, path: str):
        self.path = pathlib.Path(path)
        self._lock = threading.Lock()
        self._waiters: dict[str, threading.Event] = {}
        self._data: dict = {"webhooks": [], "notifications": []}
        self._load()

    def _load(self) -> None:
        try:
            data = json.loads(self.path.read_text(encoding="utf-8"))
            self._data = {
                "webhooks": list(data.get("webhooks") or []),
                "notifications": list(data.get("notifications") or []),
            }
        except (OSError, ValueError):
            self._data = {"webhooks": [], "notifications": []}
        if not self._data["webhooks"] and not self.path.exists():
            # First run: seed the Home Assistant webhook so the phone works
            # out of the box once the HA-side integration is installed.
            self._data["webhooks"] = [{
                "id": "home-assistant", "name": "Home Assistant",
                "url": DEFAULT_HA_WEBHOOK_URL,
                "events": list(WEBHOOK_EVENTS), "enabled": True,
            }]
            self._save()

    def _save(self) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        tmp = self.path.with_suffix(".json.tmp")
        tmp.write_text(json.dumps(self._data, indent=2), encoding="utf-8")
        os.replace(tmp, self.path)

    # ── webhooks ────────────────────────────────────────────────────────────
    def webhooks(self) -> list[dict]:
        with self._lock:
            return json.loads(json.dumps(self._data["webhooks"]))

    def set_webhooks(self, webhooks: list[dict]) -> list[dict]:
        clean = valid_webhooks(webhooks)
        with self._lock:
            self._data["webhooks"] = clean
            self._save()
        return json.loads(json.dumps(clean))

    def targets(self, event: str) -> list[dict]:
        """Enabled webhooks subscribed to ``event``."""
        with self._lock:
            return [json.loads(json.dumps(w)) for w in self._data["webhooks"]
                    if w.get("enabled") and event in (w.get("events") or [])]

    # ── log ─────────────────────────────────────────────────────────────────
    def record(self, entry: dict) -> dict:
        with self._lock:
            self._data["notifications"].append(entry)
            if len(self._data["notifications"]) > _MAX_LOG:
                self._data["notifications"] = \
                    self._data["notifications"][-_MAX_LOG:]
            self._save()
        return entry

    def log(self, limit: int = 200) -> list[dict]:
        with self._lock:
            out = list(self._data["notifications"])[-max(1, limit):]
        out.reverse()  # newest first
        return json.loads(json.dumps(out))

    def get(self, nid: str) -> dict | None:
        with self._lock:
            for n in self._data["notifications"]:
                if n.get("id") == nid:
                    return json.loads(json.dumps(n))
        return None

    def _update(self, nid: str, patch: dict) -> dict | None:
        with self._lock:
            for n in self._data["notifications"]:
                if n.get("id") == nid:
                    n.update(patch)
                    self._save()
                    return json.loads(json.dumps(n))
        return None

    # ── asks ────────────────────────────────────────────────────────────────
    def create_ask(self, question: str, options: list[str], *,
                   type_: str = "", url: str = "", image: str = "",
                   session_id: str = "",
                   timeout_s: int = DEFAULT_ASK_TIMEOUT_S,
                   nid: str = "") -> dict:
        """``nid``: the notify CLI mints ids itself (``ask_`` + 12 hex) so an
        ask created through a worker's outbox keeps the id the CLI printed."""
        nid = nid or "ask_" + uuid.uuid4().hex[:12]
        expires = datetime.now(timezone.utc) + timedelta(
            seconds=max(10, timeout_s))
        entry = {
            "id": nid, "kind": "ask", "title": "Claude needs a choice",
            "body": question, "type": type_ or "", "url": url or "",
            "image": image or "",
            "at": _now_iso(), "sessionId": session_id or "",
            "delivered": [],
            "options": list(options), "status": "pending",
            "answer": None, "answerIndex": None, "answeredAt": None,
            "expiresAt": expires.strftime("%Y-%m-%dT%H:%M:%SZ"),
        }
        with self._lock:
            self._waiters[nid] = threading.Event()
        return self.record(entry)

    def answer_ask(self, nid: str, index: int | None,
                   label: str | None) -> dict | None:
        cur = self.get(nid)
        if not cur or cur.get("kind") != "ask":
            return None
        options = cur.get("options") or []
        if label is None and index is not None and 0 <= index < len(options):
            label = options[index]
        if index is None and label in options:
            index = options.index(label)
        out = self._update(nid, {
            "status": "answered", "answer": label, "answerIndex": index,
            "answeredAt": _now_iso(),
        })
        with self._lock:
            ev = self._waiters.pop(nid, None)
        if ev:
            ev.set()
        return out

    def ask_status(self, nid: str) -> dict | None:
        """The ask with its *effective* status (pending past expiry → expired)."""
        cur = self.get(nid)
        if not cur or cur.get("kind") != "ask":
            return None
        if cur.get("status") == "pending":
            try:
                exp = datetime.strptime(cur.get("expiresAt") or "",
                                        "%Y-%m-%dT%H:%M:%SZ")
                if datetime.now(timezone.utc).replace(tzinfo=None) > exp:
                    cur = self._update(nid, {"status": "expired"}) or cur
            except ValueError:
                pass
        return cur

    def wait_for_answer(self, nid: str, wait_s: float) -> dict | None:
        """Block up to ``wait_s`` for the ask to be answered; returns the ask."""
        with self._lock:
            ev = self._waiters.get(nid)
        if ev and wait_s > 0:
            ev.wait(min(wait_s, 55.0))
        return self.ask_status(nid)

    def set_delivered(self, nid: str, delivered: list[dict]) -> None:
        self._update(nid, {"delivered": delivered})

    # ── notify action buttons ───────────────────────────────────────────────
    # A plain push may carry extra buttons (``notify send --action-cmd``):
    # ``actions: [{action: "<nid>__<i>", title}]``. The HA integration POSTs
    # the tapped index to /api/notify/{id}/action exactly like an ask's
    # answer, and the CLI's detached waiter long-polls GET /api/notify/{id}.
    def watch_actions(self, nid: str) -> None:
        with self._lock:
            self._waiters[nid] = threading.Event()

    def record_action(self, nid: str, index: int | None,
                      title: str | None) -> dict | None:
        cur = self.get(nid)
        if not cur or cur.get("kind") != "notify":
            return None
        actions = cur.get("actions") or []
        if title is None and index is not None and 0 <= index < len(actions):
            title = actions[index].get("title")
        out = self._update(nid, {"status": "acted", "actionIndex": index,
                                 "actionTitle": title,
                                 "actedAt": _now_iso()})
        with self._lock:
            ev = self._waiters.pop(nid, None)
        if ev:
            ev.set()
        return out

    def wait_for_action(self, nid: str, wait_s: float) -> dict | None:
        with self._lock:
            ev = self._waiters.get(nid)
        if ev and wait_s > 0:
            ev.wait(min(wait_s, 55.0))
        return self.get(nid)


def forward(webhooks: list[dict], payload: dict) -> list[dict]:
    """POST ``payload`` to every webhook; returns per-webhook delivery status.

    Synchronous with a short timeout — the caller (notify.sh) uses the result
    to decide whether to fall back to pushing Home Assistant directly.
    """
    results: list[dict] = []
    body = json.dumps(payload).encode()
    for w in webhooks:
        ok, status = False, 0
        try:
            req = urllib.request.Request(
                w["url"], data=body,
                headers={"Content-Type": "application/json"}, method="POST")
            with urllib.request.urlopen(req, timeout=_FORWARD_TIMEOUT_S) as r:
                status = r.status
                ok = 200 <= r.status < 300
        except urllib.error.HTTPError as e:
            status = e.code
        except (urllib.error.URLError, OSError, ValueError):
            status = 0
        results.append({"webhook": w["id"], "ok": ok, "status": status})
    return results