#!/usr/bin/env python3
"""Emit Windy Git CI telemetry to admin.windyword.ai (Windy Telemetry 40's ledger).

Runs on Veron after every sync (root; reads the gitea DB via `docker exec`).
Shapes are declared with Telemetry 40 (2026-09-23) — do not add keys or enum
values without re-declaring: a declared family quarantines any row that
doesn't match.

  ci.run          one row per FINISHED job, exactly once — cursor on
                  (finish time, job id) in STATE; jobs finish out of id order
  service.health  one row per invocation: CI plane counts for the interval

Privacy: ids, names of repos/jobs, codes, counts, durations. No commit
messages, no logs, no author names.

  --dry-run   print the batch instead of posting (and don't advance STATE)
"""

from __future__ import annotations

import json
import os
import re
import subprocess
import sys
import time
import urllib.error
import urllib.request
from datetime import UTC, datetime

INGEST = os.environ.get("TELEMETRY_INGEST_URL", "https://admin.windyword.ai/v1/events")
TOKEN = os.environ.get("WINDYGIT_TELEMETRY_TOKEN", "")
STATE = os.environ.get("TELEMETRY_STATE", "/var/lib/windy-git/telemetry-state.json")
PLATFORM, SERVICE = "windy-git", "ci"
OUTCOME = {1: "success", 2: "failure", 3: "cancelled", 4: "skipped"}
EVENTS = {"push", "pull_request", "pull_request_sync", "schedule", "workflow_dispatch"}
RUNNERS_EXPECTED = 6


def sql(query: str) -> list[dict]:
    """Rows as dicts, via psql's json_agg — no driver needed on the host."""
    wrapped = f"select coalesce(json_agg(t), '[]'::json) from ({query}) t;"
    out = subprocess.run(
        [
            "docker",
            "exec",
            "-i",
            "windy-git-db-1",
            "sh",
            "-c",
            'psql -U "$POSTGRES_USER" -d gitea -At -v ON_ERROR_STOP=1',
        ],
        input=wrapped,
        capture_output=True,
        text=True,
        check=True,
    ).stdout.strip()
    return json.loads(out or "[]")


def load_state() -> dict:
    try:
        with open(STATE) as f:
            return json.load(f)
    except (OSError, ValueError):
        return {}


def iso(epoch: float) -> str:
    return datetime.fromtimestamp(epoch, UTC).isoformat().replace("+00:00", "Z")


# ---- push velocity: DETECT + ALERT ONLY (G3.4, 2026-09-23) -----------------
# `git push` goes straight to Gitea and never touches our API, so throttle.py
# cannot see it (NOT_ENFORCED_HERE). Gitea's own `action` table does record
# every push, so we read it here, emit `forge.push_velocity` when an account
# crosses a threshold, and let Telemetry Boss's detector page. Nothing here sits
# in the push path; nothing is ever refused (orchestrator, 09-23).
#
# Thresholds = the STANDARD-band bases from config.py (500 pushes/day; the
# force-push base of 10/day is used for ref deletes, the closest thing we can
# see). EI band multipliers are NOT applied: a platinum agent over 500/day is
# still flagged, for a human to look at, not blocked. Gitea records no
# "forced" flag, so force pushes cannot be told apart from pushes: named, not
# guessed.
PV_RULES = (  # (rule, row key, window_s, threshold)
    ("pushes_1h", "p1h", 3600, 60),
    ("pushes_24h", "p24h", 86400, 500),
    ("ref_deletes_24h", "d24h", 86400, 10),
)
# The GitHub -> Windy Git sync pushes as windyadmin every 5 min, by design.
PV_EXEMPT = {"windyadmin"}
# Gitea op_type: 5 commit push, 9 tag push, 16 tag delete, 17 branch delete.
# One action row per WATCHER is written for each push; user_id = act_user_id
# keeps exactly the actor's own copy.
PV_QUERY = """
    select a.act_user_id as uid, u.lower_name as login,
           (select el.external_id from external_login_user el
             where el.user_id = u.id order by el.external_id limit 1) as wid,
           count(*) filter (where a.op_type in (5, 9) and a.created_unix > {h1}) as p1h,
           count(*) filter (where a.op_type in (5, 9)) as p24h,
           count(*) filter (where a.op_type in (16, 17)) as d24h,
           count(distinct a.repo_id) as repos
      from action a join "user" u on u.id = a.act_user_id
     where a.created_unix > {h24} and a.user_id = a.act_user_id
       and a.op_type in (5, 9, 16, 17)
     group by 1, 2"""


def passport_from_login(login: str) -> str | None:
    """agent-et26abcd1234 -> ET26-ABCD-1234 (repos.py _owner_login, reversed)."""
    m = re.fullmatch(r"agent-([a-z0-9]{4})([a-z0-9]{4})([a-z0-9]{4})", login)
    return "-".join(g.upper() for g in m.groups()) if m else None


def push_velocity_events(rows: list[dict], now: float, alerted: dict) -> tuple[list[dict], dict]:
    """(events, alerted') — one row per account per rule per window while over.

    `alerted` maps "<uid>:<rule>" -> epoch of the last row. An account still over
    the line is re-reported once per window, not every 5 minutes; one that drops
    back under is forgotten, so a later burst reports again.
    """
    events, keep = [], {}
    for r in rows:
        login = str(r["login"])
        if login in PV_EXEMPT:
            continue
        agent = login.startswith("agent-")
        for rule, key, window, limit in PV_RULES:
            n = int(r[key])
            if n <= limit:
                continue
            k = f"{r['uid']}:{rule}"
            last = alerted.get(k)
            if last is not None and now - float(last) < window:
                keep[k] = last
                continue
            keep[k] = now
            ev = {
                "ts": iso(now),
                "platform": PLATFORM,
                "service": "forge",
                "event_type": "forge.push_velocity",
                "metadata": {
                    "rule": rule,
                    "window_s": window,
                    "count": n,
                    "threshold": limit,
                    "repos": int(r["repos"]),
                    "gitea_user_id": int(r["uid"]),
                },
            }
            # Actor rule (telemetry UPDATE 2): agent/human rows MUST carry an
            # actor_id. Humans sign in to the forge only via Windy SSO, so the
            # external login id IS their windy_identity_id. No id we can prove
            # -> actor_type system + metadata.caller, never an invented id (I-12).
            actor_id = passport_from_login(login) if agent else (r.get("wid") or None)
            if actor_id:
                ev["actor_type"], ev["actor_id"] = ("agent" if agent else "human"), str(actor_id)
            else:
                ev["actor_type"] = "system"
                ev["metadata"]["caller"] = "unknown"
            events.append(ev)
    return events, keep


def main() -> int:
    dry = "--dry-run" in sys.argv
    state = load_state()
    now = time.time()
    since = float(state.get("last_ts", now - 300))
    # Cursor = (finish time, job id), NOT job id alone: jobs finish out of id
    # order, so an id high-water mark silently drops every long job that started
    # before the mark and finished after it (Telemetry Boss caught this: 43
    # finished vs 8 ci.run rows). Finish time = stopped, or updated for jobs
    # Gitea/the janitor skipped without a stop time.
    if "last_fin" in state:
        last_fin, last_id = int(state["last_fin"]), int(state["last_id"])
    else:  # first run or pre-cursor state: start now, never replay history
        last_fin, last_id = int(state.get("last_ts", now)), 0
    cutoff = int(now) - 5  # leave the current second alone; late writers land next run
    FIN = "coalesce(nullif(j.stopped, 0), j.updated)"

    jobs = sql(f"""
        select j.id, j.name as job, j.status, j.started, j.stopped, {FIN} as fin,
               p.lower_name as repo, p.default_branch, r.workflow_id, r.event,
               r.ref, r.index as run, left(r.commit_sha, 7) as sha
          from action_run_job j
          join action_run r on r.id = j.run_id
          join repository p on p.id = r.repo_id
         where j.status in (1, 2, 3, 4)
           and ({FIN}, j.id) > ({last_fin}, {last_id})
           and {FIN} <= {cutoff}
         order by {FIN}, j.id
         limit 2000""")

    try:  # posted_to_github: the bridge's own rules, from the same checkout
        sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
        import pr_status_bridge as bridge
    except Exception:  # noqa: BLE001
        bridge = None

    events = []
    for j in jobs:
        ref = j["ref"] or ""
        if ref.startswith("refs/pull/"):
            kind = "pr"
        elif ref == f"refs/heads/{j['default_branch']}":
            kind = "default"
        else:
            kind = "other"
        # Gitea stores whole seconds; duration_ms is seconds*1000 (so 10000 = 10 s).
        dur = (j["stopped"] - j["started"]) * 1000 if j["started"] and j["stopped"] else None
        ev = {
            "ts": iso(j["stopped"]),
            "platform": PLATFORM,
            "service": SERVICE,
            "event_type": "ci.run",
            "actor_type": "system",
            "metadata": {
                "repo": j["repo"],
                "workflow": (j["workflow_id"] or "").removesuffix(".yml").removesuffix(".yaml"),
                "job": j["job"],
                "outcome": OUTCOME[j["status"]],
                "event": j["event"] if j["event"] in EVENTS else "other",
                "branch_kind": kind,
                "run": j["run"],
                "sha": j["sha"],
            },
        }
        if dur is not None and dur >= 0:
            ev["duration_ms"] = int(dur)
        if bridge is not None:
            wf = ev["metadata"]["workflow"]
            ev["metadata"]["posted_to_github"] = bool(
                j["repo"] in {r.lower() for r in bridge.REPOS}
                and kind in ("default", "pr")
                and j["status"] != 4
                and not bridge.NO_DAEMON_JOB.search(j["job"])
                and f"{wf}/{j['job']}" not in bridge.NON_BLOCKING.get(j["repo"], set())
            )
        events.append(ev)

    # --- heartbeat: counts since the previous invocation --------------------
    # Interval counts come from EXACTLY the rows emitted above, so
    # sum(jobs_finished) over any window == count(ci.run) in it, by construction.
    h = sql(f"""
        select
          (select count(*) from action_run_job where status in (5, 7)) as jobs_waiting,
          (select count(*) from action_run_job where status = 6) as jobs_running,
          (select count(*) from action_runner where deleted is null and last_online >= {int(now) - 120}) as runners_online,
          (select coalesce(extract(epoch from now())::bigint - min(created), 0)
             from action_run_job where status in (5, 7)) as oldest_waiting_s""")[0]
    h["jobs_finished"] = len(jobs)
    h["jobs_failed"] = sum(1 for j in jobs if j["status"] == 2)
    h["jobs_cancelled"] = sum(1 for j in jobs if j["status"] == 3)
    meta = {k: int(v) for k, v in h.items()}
    meta["interval_s"] = int(now - since)  # ecosystem-standard key
    # UPDATE 7. Quarantines seen on earlier sends (the ledger answers 202 anyway)
    # are carried in the state file until a heartbeat reports them. Dropped is 0
    # by construction: a failed send keeps the cursor and the spool, so every row
    # is re-sent next run (a partial failure can duplicate, never lose).
    meta["telemetry_quarantined"] = int(state.get("quarantined_unreported", 0))
    meta["telemetry_dropped"] = 0
    for k in ("repos_synced", "repos_sync_failed", "statuses_posted", "bridge_errors"):
        v = os.environ.get(f"TELEMETRY_{k.upper()}")
        if v is not None and v.isdigit():  # absent = couldn't count; never invent 0
            meta[k] = int(v)
    events.append(
        {
            "ts": iso(now),
            "platform": PLATFORM,
            "service": SERVICE,
            "event_type": "service.health",
            "actor_type": "system",
            "metadata": meta,
        }
    )

    # Isolated: a failing push-velocity query must never cost the ci.run rows.
    pv_alerted = state.get("pv_alerted", {})
    try:
        pv_rows = sql(PV_QUERY.format(h1=int(now) - 3600, h24=int(now) - 86400))
        pv_events, pv_alerted = push_velocity_events(pv_rows, now, pv_alerted)
    except (subprocess.CalledProcessError, ValueError, KeyError) as e:
        print(f"[telemetry] push velocity check FAILED (non-fatal): {type(e).__name__}")
        pv_events = []
    for e in pv_events:
        m = e["metadata"]
        print(f"[telemetry] WARNING push velocity: gitea user {m['gitea_user_id']} "
              f"{m['rule']} = {m['count']} > {m['threshold']}")
    events += pv_events

    # ci.job_cancelled: spooled by the janitor (cancel_unrunnable.sh), one JSON per job.
    spool = os.environ.get("JANITOR_SPOOL", "/var/lib/windy-git/janitor-cancelled.jsonl")
    spooled = 0
    try:
        with open(spool) as f:
            for line in f:
                try:
                    m = json.loads(line)
                except ValueError:
                    continue
                events.append(
                    {
                        "ts": iso(now),
                        "platform": PLATFORM,
                        "service": SERVICE,
                        "event_type": "ci.job_cancelled",
                        "actor_type": "system",
                        "metadata": {
                            k: m[k]
                            for k in ("repo", "workflow", "job", "reason", "runs_on", "waited_s")
                        },
                    }
                )
                spooled += 1
    except OSError:
        pass

    if dry:
        out = os.environ.get("TELEMETRY_DRY_OUT")
        if out:
            with open(out, "w") as f:
                json.dump({"events": events}, f)
        else:
            print(json.dumps({"events": events}, indent=1)[:4000])
        print(f"[telemetry] DRY RUN: {len(events)} events ({len(jobs)} ci.run)")
        return 0
    if not TOKEN:
        print("[telemetry] WINDYGIT_TELEMETRY_TOKEN unset — not sending (not a failure)")
        return 0

    quarantined = 0
    for i in range(0, len(events), 500):
        req = urllib.request.Request(
            INGEST,
            data=json.dumps({"events": events[i : i + 500]}).encode(),
            method="POST",
            headers={
                "Authorization": f"Bearer {TOKEN}",
                "Content-Type": "application/json",
                "User-Agent": "windy-git-telemetry/1",
            },
        )
        try:
            with urllib.request.urlopen(req, timeout=30) as r:
                body = r.read()
                if r.status >= 300:
                    raise urllib.error.HTTPError(
                        INGEST, r.status, body[:300].decode(errors="replace"), None, None
                    )
            try:
                resp = json.loads(body or b"{}")
            except ValueError:
                resp = {}
            q = resp.get("quarantined") if isinstance(resp, dict) else None
            if isinstance(q, int) and q > 0:
                quarantined += q
                reasons = "; ".join(map(str, resp.get("rejections") or [])) or "no reason given"
                print(f"[telemetry] WARNING {q} row(s) QUARANTINED by the ledger: {reasons}")
        except urllib.error.HTTPError as e:
            print(f"[telemetry] FAILED ingest HTTP {e.code}: {e.read()[:200]!r}")
            return 1  # state NOT advanced: the same rows retry next run
        except urllib.error.URLError as e:
            print(f"[telemetry] FAILED ingest: {e.reason}")
            return 1

    if spooled:
        open(spool, "w").close()  # only after every batch was accepted
    os.makedirs(os.path.dirname(STATE), exist_ok=True)
    new_fin, new_id = (jobs[-1]["fin"], jobs[-1]["id"]) if jobs else (last_fin, last_id)
    with open(STATE + ".tmp", "w") as f:
        json.dump(
            {"last_fin": new_fin, "last_id": new_id, "last_ts": now,
             "quarantined_unreported": quarantined, "pv_alerted": pv_alerted},
            f,
        )
    os.replace(STATE + ".tmp", STATE)
    print(f"[telemetry] sent {len(events)} events ({len(jobs)} ci.run)")
    return 0


if __name__ == "__main__":
    sys.exit(main())