#!/usr/bin/env -S uv run --script --quiet
# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "httpx",
#     "websockets",
#     "atproto",
#     "pydantic-settings",
# ]
# ///
"""pub-search ingestion + serving watchdog.

Two independent checks, both off the serving path's failure domain:

1. FRESHNESS (turso): `max(indexed_at)` must be younger than
   FRESHNESS_ALERT_MINUTES. Catches the whole ingest chain being dead —
   ingester down, /channel wedged, backend worker stalled — regardless of
   which component ate it. This is the check that correctly detected the
   2026-06-10 seven-hour stall while latency probes were green.
2. SERVING (backend): /stats and a real keyword search must return 200 with
   sane bodies within 5s. Status codes alone are NOT trusted (Pages serves
   fake 200s; see docs/retro-2026-06-10-cutover-cascade.md).

tap is GONE (replaced by the ingester 2026-06-09; machine stopped with
autostart disabled). The old tap-outbox/cursor checks alerted on a zombie
machine that fly auto-started whenever this watchdog probed it.

Requires: ATPROTO_HANDLE, ATPROTO_PASSWORD, TURSO_DB_TOKEN.

Usage:
    ./scripts/ingestion-watchdog                 # check; post only if stalled
    ./scripts/ingestion-watchdog --test          # wiring test post (auto-deletes)
"""

import argparse
import datetime
import os
import sys
import time

import httpx
from atproto import Client, client_utils
from pydantic_settings import BaseSettings, SettingsConfigDict
from watchdog_checks import snapshot_age_problem

ALERT_HANDLE = "zzstoatzz.io"
ALERT_MARKER = "pub-search ingestion looks stalled"
BACKEND = "https://leaflet-search-backend.fly.dev"
TURSO_URL = "https://leaf-zzstoatzz.aws-us-east-1.turso.io/v2/pipeline"
FRESHNESS_ALERT_MINUTES = 60


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=os.environ.get("ENV_FILE", ".env"), extra="ignore"
    )
    atproto_handle: str = ""
    atproto_password: str = ""
    turso_db_token: str = ""
    prefect_api_url: str = ""
    prefect_api_auth_string: str = ""


def _max_indexed_at(token: str) -> str:
    r = httpx.post(
        TURSO_URL,
        headers={"Authorization": f"Bearer {token}"},
        json={"requests": [
            {"type": "execute", "stmt": {"sql": "SELECT max(indexed_at) FROM documents"}},
            {"type": "close"},
        ]},
        timeout=20,
    )
    r.raise_for_status()
    return r.json()["results"][0]["response"]["result"]["rows"][0][0]["value"]


def check(window_secs: int) -> tuple[bool, str]:
    del window_secs  # single-shot checks; kept for CLI compat
    problems = []

    # 1. ingestion freshness via turso
    settings = Settings()
    token = getattr(settings, "turso_db_token", None)
    if token:
        try:
            newest = _max_indexed_at(token)
            age_min = (
                datetime.datetime.now(datetime.timezone.utc)
                - datetime.datetime.fromisoformat(newest).replace(tzinfo=datetime.timezone.utc)
            ).total_seconds() / 60
            if age_min > FRESHNESS_ALERT_MINUTES:
                problems.append(f"no new docs indexed in {age_min:.0f}m (newest: {newest})")
            else:
                print(f"freshness ok: newest doc {age_min:.0f}m old")
        except Exception as e:
            problems.append(f"turso freshness check failed: {e!r}")
    else:
        print("TURSO_DB_TOKEN not set — skipping freshness check")

    # 2. snapshot freshness: serving adopts hourly builds from R2; the
    # /snapshot endpoint serves the live replica's manifest sidecar. A stale
    # created_at means the builder or the promote watcher is dead even though
    # turso ingestion looks fine.
    try:
        r = httpx.get(BACKEND + "/snapshot", timeout=5)
        if r.status_code == 200:
            manifest = r.json()
            now = datetime.datetime.now(datetime.timezone.utc).timestamp()
            if problem := snapshot_age_problem(manifest, now):
                problems.append(problem)
            else:
                snap_age_min = (now - manifest["created_at"]) / 60
                print(f"snapshot ok: {snap_age_min:.0f}m old ({manifest.get('build_id', '?')})")
        else:
            # 404 = pre-adoption replica; tolerated until the first adoption
            print(f"snapshot check: /snapshot -> {r.status_code} (pre-adoption replica?)")
    except Exception as e:
        problems.append(f"/snapshot -> {e!r}")

    # 3. serving health, bodies inspected. one retry after a pause: the
    # promote watcher restarts the machine once an hour to adopt a snapshot
    # (a seconds-wide window), and a single probe can race it — a real
    # outage fails twice 30s apart, a restart blip doesn't.
    for path, want in [("/stats", '"documents"'), ("/search?q=atproto&mode=keyword", '"uri"')]:
        last_err = None
        for attempt in range(2):
            try:
                r = httpx.get(BACKEND + path, timeout=5)
                if r.status_code == 200 and want in r.text[:500]:
                    print(f"serving ok: {path}" + (" (after retry)" if attempt else ""))
                    last_err = None
                    break
                last_err = f"{path} -> {r.status_code}, body unexpected"
            except Exception as e:
                last_err = f"{path} -> {e!r}"
            if attempt == 0:
                time.sleep(30)
        if last_err:
            problems.append(last_err)

    # 4. snapshot builder (heavypad prefect flow) — cause-level check so a
    # failing flow pages within 15min instead of waiting for the 180m
    # snapshot-age symptom. Skipped silently when the creds aren't set.
    if settings.prefect_api_url and settings.prefect_api_auth_string:
        try:
            import base64

            base_url = settings.prefect_api_url.rstrip("/")
            headers = {
                "Authorization": "Basic "
                + base64.b64encode(settings.prefect_api_auth_string.encode()).decode()
            }
            # the flow-name filter is ignored by this server version; the
            # deployment id is the unambiguous handle anyway.
            dep = httpx.get(
                base_url + "/deployments/name/pub-search-snapshot/pub-search-snapshot",
                headers=headers,
                timeout=10,
            )
            dep.raise_for_status()
            r = httpx.post(
                base_url + "/flow_runs/filter",
                headers=headers,
                json={
                    "deployments": {"id": {"any_": [dep.json()["id"]]}},
                    "flow_runs": {"state": {"type": {"any_": ["COMPLETED", "FAILED", "CRASHED"]}}},
                    "sort": "END_TIME_DESC",
                    "limit": 1,
                },
                timeout=10,
            )
            r.raise_for_status()
            runs = r.json()
            if runs:
                last = runs[0]
                state = last.get("state_type", "?")
                if state != "COMPLETED":
                    problems.append(
                        f"snapshot builder flow run {last.get('name', '?')} ended {state}"
                    )
                else:
                    print(f"builder ok: last run {last.get('name', '?')} COMPLETED")
            else:
                problems.append("snapshot builder: no terminal flow runs found")
        except Exception as e:
            # the prefect API being unreachable is itself worth knowing about,
            # but don't page for a transient blip alone — print, don't append.
            print(f"builder check skipped: prefect API unreachable ({e!r})")

    if problems:
        return True, "; ".join(problems)
    return False, "healthy: ingestion fresh, serving green"


def _post(client: Client, target_did: str, message: str):
    tb = client_utils.TextBuilder().mention(f"@{ALERT_HANDLE}", target_did).text(f" {message}")
    return client.send_post(tb)


def _alerted_within(client: Client, hours: int) -> bool:
    """True if this account already posted a stall alert in the last `hours`.

    Keeps a 15-min cron from spamming on a multi-hour stall. Errs toward posting
    (returns False) if anything is uncertain — better a dupe than a missed alert.
    """
    try:
        feed = client.get_author_feed(client.me.did, limit=15).feed
    except Exception:
        return False
    cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(hours=hours)
    for item in feed:
        rec = getattr(item.post, "record", None)
        text = getattr(rec, "text", "") or ""
        if ALERT_MARKER not in text:
            continue
        created = getattr(rec, "created_at", None)
        if not created:
            return True
        try:
            if datetime.datetime.fromisoformat(created.replace("Z", "+00:00")) > cutoff:
                return True
        except ValueError:
            return True
    return False


def main() -> None:
    ap = argparse.ArgumentParser(description="pub-search ingestion watchdog")
    ap.add_argument("--test", action="store_true", help="post a wiring-test mention, then delete it")
    ap.add_argument("--window", type=int, default=0, help="(unused, kept for CLI compat)")
    ap.add_argument("--dedup-hours", type=int, default=3, help="suppress repeat alerts within this window")
    args = ap.parse_args()

    settings = Settings()
    if not (settings.atproto_handle and settings.atproto_password):
        print("ATPROTO_HANDLE/ATPROTO_PASSWORD not set — skipping (configure to enable)")
        return

    client = Client()
    client.login(settings.atproto_handle, settings.atproto_password)
    target_did = client.resolve_handle(ALERT_HANDLE).did

    if args.test:
        resp = _post(client, target_did, "pub-search ingestion watchdog wiring test — ignore (auto-deleting).")
        print(f"posted test: {resp.uri}")
        client.delete_post(resp.uri)
        print("deleted test post — auth + posting confirmed")
        return

    stalled, detail = check(args.window)
    print(detail)
    if stalled:
        if _alerted_within(client, args.dedup_hours):
            print(f"stalled, but already alerted within {args.dedup_hours}h — not reposting")
        else:
            resp = _post(client, target_did, f"⚠️ {ALERT_MARKER} — {detail}. check logfire + fly logs -a leaflet-search-backend")
            print(f"ALERT posted: {resp.uri}")
        sys.exit(1)


if __name__ == "__main__":
    main()
