PacketStation/server/packetstation/notify/watch.py

"""The standing conditions notify-send checks itself (at most every 15 minutes), and the version check.

drawing_failing: drawing lookups failing for more than the rule's hours (subject `drawings`, the list in the body).
co_waiting: a change order pending review for more than the rule's days (subject `co:<id>`, the packet's department).
version_deployed: the worker runs a build other than the last one noted in the settings row `notify.last_build`.
"""

from __future__ import annotations

import logging
from typing import Any

from psycopg.types.json import Jsonb

from .catalog import EVENT_TYPES
from .events import raise_event, resolve_event
from .routing import load_rules

log = logging.getLogger(__name__)

BUILD_KEY = "notify.last_build"
LIST_MAX = 50


def _param(conn: Any, event_type: str) -> int:
    rule = load_rules(conn, event_type).get(event_type)
    value = rule.param_value if rule is not None else None
    default = EVENT_TYPES[event_type].param_default or 1
    return int(value or default)


def check_drawings(conn: Any) -> dict[str, Any]:
    hours = _param(conn, "drawing_failing")
    rows = conn.execute(
        "SELECT dwg, failing_since, fail_count, note FROM drawing_lookups"
        " WHERE failing_since IS NOT NULL AND failing_since < now() - make_interval(hours => %s)"
        " ORDER BY failing_since, dwg",
        (hours,),
    ).fetchall()
    if not rows:
        rid = resolve_event(conn, "drawing_failing", subject="drawings", body="Every drawing lookup works again.")
        return {"drawings_failing": 0, "drawings_cleared": rid is not None}
    n = len(rows)
    lines = [f"{r[0]}: failing since {r[1]:%Y-%m-%d %H:%M} UTC, {r[2]} tries. {r[3] or ''}".rstrip() for r in rows]
    if n > LIST_MAX:
        lines = [*lines[:LIST_MAX], f"and {n - LIST_MAX} more"]
    raise_event(
        conn,
        "drawing_failing",
        subject="drawings",
        title=f"{n} drawing lookup{'s' if n != 1 else ''} failing for more than {hours} hours",
        body="\n".join(lines),
        detail={"count": n, "hours": hours, "drawings": [r[0] for r in rows[:LIST_MAX]]},
        condition=True,
    )
    return {"drawings_failing": n}


def check_change_orders(conn: Any) -> dict[str, Any]:
    days = _param(conn, "co_waiting")
    rows = conn.execute(
        "SELECT co.id, co.co_no, co.issued_at, p.id, p.dept_code, p.title, p.drawing, pr.number, u.lot,"
        " floor(extract(epoch FROM now() - co.issued_at) / 86400)::int"
        " FROM change_orders co JOIN packets p ON p.id = co.packet_id JOIN units u ON u.id = p.unit_id"
        " JOIN projects pr ON pr.id = u.project_id"
        " WHERE co.review = 'pending' AND p.removed_at IS NULL AND co.issued_at < now() - make_interval(days => %s)"
        " ORDER BY co.id",
        (days,),
    ).fetchall()
    waiting = {f"co:{r[0]}" for r in rows}
    raised = 0
    for co_id, co_no, _issued, pid, dept, title, drawing, number, lot, age in rows:
        what = " ".join(x for x in (f"{number}-{lot}", dept, title or drawing or "") if x)
        new = raise_event(
            conn,
            "co_waiting",
            subject=f"co:{co_id}",
            title=f"Change order CO-{co_no} waiting {age} days: {what}",
            body=f"CO-{co_no} on packet {pid} ({what}) has waited for review for {age} days.",
            detail={"co_id": co_id, "packet_id": pid, "days": age},
            dept_code=dept,
            condition=True,
        )
        raised += new is not None
    cleared = 0
    for (subject,) in conn.execute(
        "SELECT subject FROM notification_events WHERE kind = 'co_waiting' AND resolved_at IS NULL"
    ).fetchall():
        if subject not in waiting:
            cleared += (
                resolve_event(conn, "co_waiting", subject=subject, body="Reviewed or no longer waiting.") is not None
            )
    return {"co_waiting": len(rows), "co_raised": raised, "co_cleared": cleared}


def note_build(conn: Any, build: str) -> int | None:
    """Raise version_deployed when `build` differs from the last build noted. Returns the event id or None."""
    build = (build or "").strip()
    if not build:
        return None
    with conn.transaction():
        row = conn.execute("SELECT value FROM settings WHERE key = %s FOR UPDATE", (BUILD_KEY,)).fetchone()
        before = (row[0] or {}).get("build") if row is not None else None
        if before == build:
            return None
        if row is None:
            conn.execute(
                "INSERT INTO settings (key, value, version, updated_at) VALUES (%s, %s, 1, now())"
                " ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, version = settings.version + 1,"
                " updated_at = now()",
                (BUILD_KEY, Jsonb({"build": build})),
            )
        else:
            conn.execute(
                "UPDATE settings SET value = %s, version = version + 1, updated_at = now() WHERE key = %s",
                (Jsonb({"build": build}), BUILD_KEY),
            )
        log.info("version deployed: %s (was %s)", build, before or "none")
        return raise_event(
            conn,
            "version_deployed",
            subject=f"build:{build}",
            title=f"Version {build} deployed",
            body=f"The worker started on build {build}. The build before was {before or 'not recorded'}.",
            detail={"build": build, "before": before},
            severity="info",
        )


FEED_JOBS = ("visual-feed", "drawing-lookup")


def feed_run(conn: Any, job: str, status: str, errors: list[str], run_id: int | None = None) -> None:
    """feed_failed and feed_degraded after a visual-feed or drawing-lookup run."""
    if job not in FEED_JOBS or status not in ("ok", "degraded", "failed"):
        return
    first = errors[0] if errors else ""
    text = "\n".join(errors[:10])
    if status == "failed":
        raise_event(
            conn,
            "feed_failed",
            subject=job,
            title=f"{job} run failed",
            body=f"Run {run_id} failed.\n{text}".strip(),
            detail={"job": job, "run_id": run_id, "error": first},
            severity="critical",
            condition=True,
        )
        return
    resolve_event(conn, "feed_failed", subject=job, body=f"Run {run_id} finished {status}.")
    if job != "visual-feed":
        return
    if status == "degraded":
        raise_event(
            conn,
            "feed_degraded",
            subject=job,
            title=f"{job} run degraded",
            body=f"Run {run_id} finished with errors.\n{text}".strip(),
            detail={"job": job, "run_id": run_id, "error": first},
            condition=True,
        )
    else:
        resolve_event(conn, "feed_degraded", subject=job, body=f"Run {run_id} finished ok.")