cayce-ops/cayceops/components/monitor/monitor.py

"""monitor — static status page builder.

Reads the shared database and renders a self-contained HTML file (no external
assets, no JS, everything escaped). Contains NO sensitive data: daemon health,
proposal queue counts, and recent audit activity (actor/action/result only —
no email bodies, no detail payloads, no URLs).
"""

from __future__ import annotations

import html
from datetime import UTC, datetime

from sqlalchemy import func, select

from cayceops.config import Config
from cayceops.db import session_scope
from cayceops.models import (
    AuditEntry,
    ContentItem,
    DaemonRun,
    MailMessage,
    Proposal,
    ProposalStatus,
)

DAEMONS = ("scanner", "executor", "maild", "publisher", "monitor")

_STYLE = """<style>
 body{font-family:ui-monospace,Menlo,monospace;background:#0a0908;color:#f2ece0;margin:2rem}
 h1{color:#c9a84c;font-weight:600} h2{color:#c9a84c;margin-top:2rem;font-size:1.1rem}
 table{border-collapse:collapse;margin:.5rem 0}
 td,th{border:1px solid #333;padding:.35rem .8rem;text-align:left}
 .ok{color:#6fce6f}.error{color:#ef6b6b}.muted{color:#888;font-size:.85rem}
 .card{display:inline-block;background:#141210;border:1px solid #2a2a2a;
       border-radius:8px;padding:.8rem 1.4rem;margin:.3rem}
</style>"""


def _fmt_ts(dt) -> str:
    return dt.strftime("%Y-%m-%d %H:%M:%SZ") if dt else "—"


def _daemon_rows(sess) -> str:
    rows = []
    for name in DAEMONS:
        last = (
            sess.query(DaemonRun)
            .filter(DaemonRun.daemon == name)
            .order_by(DaemonRun.started_at.desc())
            .first()
        )
        if last is None:
            rows.append(
                "<tr><td>" + name + "</td><td class=muted>never run</td><td>—</td></tr>"
            )
        else:
            cls = "ok" if last.status == "ok" else "error"
            when = _fmt_ts(last.finished_at or last.started_at)
            rows.append(
                "<tr><td>" + html.escape(name) + "</td>"
                + '<td class="' + cls + '">' + html.escape(last.status) + "</td>"
                + "<td>" + when + "</td></tr>"
            )
    head = "<tr><th>daemon</th><th>last status</th><th>last run</th></tr>"
    return "<table>" + head + "".join(rows) + "</table>"


def _queue_cards(sess) -> str:
    counts = dict(
        sess.query(Proposal.status, func.count(Proposal.id)).group_by(Proposal.status).all()
    )
    cards = []
    for status in ProposalStatus:
        value = counts.get(status, 0)
        cards.append(
            "<div class=card>" + html.escape(status.value)
            + ": <strong>" + str(value) + "</strong></div>"
        )
    return "".join(cards)


def _recent_audit(sess, limit: int = 20) -> str:
    rows = []
    query = sess.query(AuditEntry).order_by(AuditEntry.ts.desc()).limit(limit)
    for entry in query.all():
        cls = "ok" if entry.result == "ok" else "error"
        rows.append(
            "<tr><td>" + _fmt_ts(entry.ts) + "</td>"
            + "<td>" + html.escape(entry.actor) + "</td>"
            + "<td>" + html.escape(entry.action) + "</td>"
            + '<td class="' + cls + '">' + html.escape(entry.result) + "</td></tr>"
        )
    body = "".join(rows) or "<tr><td colspan=4 class=muted>no activity yet</td></tr>"
    return (
        "<table><tr><th>time (utc)</th><th>actor</th><th>action</th><th>result</th></tr>"
        + body
        + "</table>"
    )


def _build(cfg: Config, sess) -> str:
    now = datetime.now(UTC)
    mail_unprocessed = (
        sess.query(func.count(MailMessage.id))
        .filter(
            MailMessage.direction == "in",
            MailMessage.draft_status.in_(["drafted", "none"]),
            MailMessage.classification.isnot(None),
        )
        .scalar()
    )
    content_pending = (
        sess.query(func.count(ContentItem.id))
        .filter(ContentItem.status.in_(["draft", "approved"]))
        .scalar()
    )
    generated = _fmt_ts(now)

    page = f"""<!doctype html>
<html lang=en><head><meta charset=utf-8><title>cayce-ops status</title>{_STYLE}</head>
<body>
<h1>cayce-ops · status</h1>
<p class=muted>generated {generated} · static page, no sensitive data by design</p>

<h2>daemon health</h2>
{_daemon_rows(sess)}

<h2>proposal queue</h2>
{_queue_cards(sess)}

<h2>gates</h2>
<div class=card>mail drafts awaiting approval: <strong>{mail_unprocessed}</strong></div>
<div class=card>content items draft/approved: <strong>{content_pending}</strong></div>

<h2>recent audit activity</h2>
{_recent_audit(sess)}

<p class=muted>audit entries: {sess.scalar(select(func.count(AuditEntry.id)))} total</p>
</body></html>"""

    import os

    os.makedirs(os.path.dirname(cfg.monitor_output) or ".", exist_ok=True)
    with open(cfg.monitor_output, "w", encoding="utf-8") as fh:
        fh.write(page)
    return cfg.monitor_output


def _anchor(cfg: Config, sess) -> int:
    """Anchor core (injectable session for tests). Returns 0 ok / 0 skip."""
    from cayceops.components.publisher import nostr
    from cayceops.config import ConfigError
    from cayceops.models import AuditEntry, audit

    if not cfg.nostr_private_key or not cfg.nostr_relays:
        raise ConfigError(
            "anchor requires CAYCE_OPS_NOSTR_PRIVATE_KEY and CAYCE_OPS_NOSTR_RELAYS"
        )
    if not cfg.deployment_id:
        raise ConfigError("anchor requires CAYCE_OPS_DEPLOYMENT_ID")

    last = (
        sess.query(AuditEntry)
        .filter(AuditEntry.entry_hash.isnot(None))
        .order_by(AuditEntry.id.desc())
        .first()
    )
    if last is None:
        print("nothing to anchor: chain empty")
        return 0
    today = datetime.now(UTC).date()
    last_anchor = (
        sess.query(AuditEntry)
        .filter(AuditEntry.action == "anchor")
        .order_by(AuditEntry.id.desc())
        .first()
    )
    if last_anchor is not None and last_anchor.ts.date() == today:
        print("already anchored today")
        return 0

    secret = nostr.normalize_key(cfg.nostr_private_key)
    event = nostr.build_event(
        secret,
        f"cayce-ops audit anchor deployment={cfg.deployment_id} "
        f"last_hash={last.entry_hash} entries={last.id}",
        tags=[["t", "cayce-ops"], ["t", "audit-anchor"]],
    )
    if not nostr.verify_event(event):
        raise RuntimeError("anchor event failed self-verification")
    results = nostr.publish(cfg.nostr_relays, event)
    audit(
        sess,
        actor="daemon:monitor",
        action="anchor",
        detail={
            "last_hash": last.entry_hash,
            "last_entry": last.id,
            "event_id": event["id"],
            "relays": results,
            "deployment": cfg.deployment_id,
        },
    )
    print(f"anchored: {event['id']}")
    return 0


def anchor(cfg: Config | None = None) -> int:
    """CLI: anchor once per day (external anchoring, audit round 2 point 4)."""
    from cayceops.config import ConfigError, load
    from cayceops.db import engine

    cfg = cfg or load()
    try:
        with session_scope() as sess:
            try:
                return _anchor(cfg, sess)
            except ConfigError as exc:
                print(f"anchor refused: {exc}")
                return 2
    except Exception as exc:  # noqa: BLE001 — CLI boundary
        print(f"anchor failed: {str(exc)[:200]}")
        return 1
    finally:
        engine().dispose()


def build(cfg: Config | None = None) -> int:
    """CLI entry: render the status page inside a DaemonRun bookkeeping wrap."""
    from cayceops.config import load
    from cayceops.db import engine

    cfg = cfg or load()
    try:
        with session_scope() as sess:
            run_row = DaemonRun(daemon="monitor")
            sess.add(run_row)
            sess.flush()
            try:
                out = _build(cfg, sess)
            except Exception as exc:  # noqa: BLE001 — daemon boundary
                run_row.status = "error"
                run_row.detail = {"error": str(exc)[:300]}
                run_row.finished_at = datetime.now(UTC)
                return 1
            run_row.status = "ok"
            run_row.detail = {"output": out}
            run_row.finished_at = datetime.now(UTC)
        print(out)
        return 0
    finally:
        engine().dispose()


def _latest_chain_hash(sess) -> str | None:
    """Last non-NULL entry_hash in the audit chain (None if chain empty)."""
    from cayceops.models import AuditEntry

    last = (
        sess.query(AuditEntry)
        .filter(AuditEntry.entry_hash.isnot(None))
        .order_by(AuditEntry.id.desc())
        .first()
    )
    return last.entry_hash if last else None


def _published_anchors(sess) -> list[dict]:
    """Anchor audit entries recorded locally, oldest first."""
    from cayceops.models import AuditEntry

    rows = (
        sess.query(AuditEntry)
        .filter(AuditEntry.action == "anchor")
        .order_by(AuditEntry.id.asc())
        .all()
    )
    return [
        {
            "event_id": (r.detail or {}).get("event_id"),
            "last_hash": (r.detail or {}).get("last_hash"),
            "last_entry": (r.detail or {}).get("last_entry"),
            "ts": r.ts.isoformat(),
        }
        for r in rows
    ]


def verify_chain_against_anchors(
    sess,
    relay_fetcher=None,
) -> dict:
    """Compare the local chain tail with the anchors published to Nostr.

    relay_fetcher(deployment) -> list of note contents (strings) found on
    the relays for this deployment's anchor tag. When None, only local
    coherence is checked and the result says so.

    Returns a report dict:
      status: "no_anchors" | "local_only" | "ok" | "mismatch"
      detail: human-readable explanation
      local_last_hash, local_anchors, relay_notes (when fetched)
    """
    local_hash = _latest_chain_hash(sess)
    anchors = _published_anchors(sess)

    report: dict = {
        "local_last_hash": local_hash,
        "local_anchors": anchors,
        "relay_notes": None,
    }

    if not anchors:
        report["status"] = "no_anchors"
        report["detail"] = (
            "the chain has never been anchored to Nostr — no external "
            "attestation exists yet (anchor once with `monitor anchor`)"
        )
        return report

    last_anchor = anchors[-1]
    if relay_fetcher is None:
        report["status"] = "local_only"
        report["detail"] = (
            "compared only against the LOCAL record of past anchors "
            "(no relay fetch configured); last anchored hash "
            f"{last_anchor['last_hash']}"
        )
        return report

    from cayceops.config import load as _load

    deployment = _load().deployment_id
    notes = relay_fetcher(deployment) or []
    report["relay_notes"] = notes

    if not notes:
        report["status"] = "no_anchors"
        report["detail"] = (
            "local anchor records exist but no anchor notes were found on "
            "the relays — the external attestation is missing or was deleted"
        )
        return report

    # the freshest relay note must attest the current chain tail
    expected_fragment = f"last_hash={local_hash}"
    latest_note = notes[-1]
    if expected_fragment in latest_note:
        report["status"] = "ok"
        report["detail"] = "latest relay anchor matches the current chain tail"
        return report

    # maybe the chain advanced after the last anchor — that's fine ONLY if
    # the note matches the hash recorded in the last local anchor entry
    if last_anchor["last_hash"] and last_anchor["last_hash"] in latest_note:
        report["status"] = "ok"
        report["detail"] = (
            "relay anchor matches the last local anchor; the chain has "
            "advanced since (anchor again to attest the new tail)"
        )
        return report

    report["status"] = "mismatch"
    report["detail"] = (
        f"relay anchor does not attest chain tail {local_hash} nor the last "
        f"local anchor {last_anchor['last_hash']} — the published anchor "
        "does not correspond to this database"
    )
    return report