"""
check_live_perf.py -- READ-ONLY check of the perf changes (slim list, gzip, per-lead detail, bell endpoints)
against a backend you already have running -- your dev server on your real Neon data, or a deployed Space.

    uv run python scripts/check_live_perf.py                         # http://127.0.0.1:8000
    uv run python scripts/check_live_perf.py --base-url https://<user>-<space>.hf.space
    uv run python scripts/check_live_perf.py --repeat 5

Only GET requests. Nothing is created, edited or deleted; the backend does the DB reads it always does.
Auth: uses API_KEY from --api-key, the environment, or outreach-backend/.env (the server-to-server key the
backend already accepts). If the backend has no API_KEY set and login is off, no key is needed.

What it reports:
  * which DB endpoint the backend .env points at (pooled / direct -- host only, never the password)
  * GET /health
  * GET /leads (full) vs GET /leads?slim=1: rows, decoded bytes, bytes actually on the wire, Content-Encoding,
    and how long each took (first call, then the median of --repeat)
  * that slim rows really are empty on the six long-text fields, and everything else matches the full list
  * GET /leads/{id} for a lead that has a draft: full, slim=false, text present
  * GET /notifications/unread-count latency (the bell's only poll now)
Exit code = number of FAILED checks.
"""

from __future__ import annotations

import argparse
import json
import statistics
import sys
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import httpx  # noqa: E402

from app.config import is_pooled_url, settings  # noqa: E402
from app.schemas.lead import SLIM_DEFERRED  # noqa: E402

FAILED = 0


def check(label: str, ok: bool, detail: str = "") -> None:
    global FAILED
    FAILED += 0 if ok else 1
    print(f"  [{'ok  ' if ok else 'FAIL'}] {label}" + ("" if ok else f"   {detail}"))


def fetch(client: httpx.Client, path: str, params: dict | None = None) -> dict:
    """One GET asking for gzip; returns timing, wire bytes, decoded bytes, encoding and parsed JSON."""
    t0 = time.perf_counter()
    with client.stream("GET", path, params=params) as r:
        wire = b"".join(r.iter_raw())
        status, enc = r.status_code, r.headers.get("content-encoding") or "none"
    ms = (time.perf_counter() - t0) * 1000
    body = wire
    if enc == "gzip":
        import gzip

        body = gzip.decompress(wire)
    data = None
    if status == 200:
        try:
            data = json.loads(body)
        except ValueError:
            data = None
    return {"status": status, "ms": ms, "wire": len(wire), "decoded": len(body), "enc": enc, "data": data}


def timed(client: httpx.Client, path: str, params: dict | None, repeat: int) -> tuple[dict, float]:
    first = fetch(client, path, params)
    times = [fetch(client, path, params)["ms"] for _ in range(max(0, repeat - 1))]
    return first, (statistics.median(times) if times else first["ms"])


def kb(n: int) -> str:
    return f"{n / 1024:,.0f} KB"


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--base-url", default="http://127.0.0.1:8000")
    ap.add_argument("--api-key", default="")
    ap.add_argument("--repeat", type=int, default=3, help="calls per endpoint after the first (median reported)")
    args = ap.parse_args()

    url = settings.sqlalchemy_url
    host = url.split("@")[-1].split("/")[0] if "@" in url else "(sqlite / not set)"
    print(f"\nBackend .env database: {host}  ->  {'POOLED (PgBouncer)' if is_pooled_url(url) else 'direct'}")
    print(f"Testing: {args.base_url}")

    key = args.api_key or settings.api_key
    headers = {"Accept-Encoding": "gzip"}
    if key:
        headers["X-API-Key"] = key
    client = httpx.Client(base_url=args.base_url.rstrip("/"), headers=headers, timeout=60)

    print("\nHealth")
    try:
        h = fetch(client, "/health")
    except httpx.HTTPError as exc:
        print(f"  [FAIL] cannot reach {args.base_url}: {type(exc).__name__} -- is the backend running?")
        return 1
    check("GET /health ok", h["status"] == 200 and (h["data"] or {}).get("db") == "ok", str(h["data"]))

    print("\nLead list: full vs slim")
    full, full_med = timed(client, "/leads", None, args.repeat)
    slim, slim_med = timed(client, "/leads", {"slim": 1}, args.repeat)
    if full["status"] != 200 or slim["status"] != 200:
        check("GET /leads both 200", False, f"full {full['status']}, slim {slim['status']} (401 = API key missing/wrong)")
        return FAILED
    fr, sr = full["data"], slim["data"]
    print(f"  {len(fr)} leads")
    print(f"  {'':10} {'decoded':>10} {'on the wire':>12} {'encoding':>9} {'first':>9} {'median':>9}")
    print(f"  {'full':10} {kb(full['decoded']):>10} {kb(full['wire']):>12} {full['enc']:>9} {full['ms']:>7.0f}ms {full_med:>7.0f}ms")
    print(f"  {'slim':10} {kb(slim['decoded']):>10} {kb(slim['wire']):>12} {slim['enc']:>9} {slim['ms']:>7.0f}ms {slim_med:>7.0f}ms")
    if full["wire"]:
        print(f"  -> slim on the wire is {slim['wire'] / full['wire'] * 100:.0f}% of full on the wire; "
              f"gzip alone shrinks full by {(1 - full['wire'] / full['decoded']) * 100:.0f}%")
    check("same number of rows", len(fr) == len(sr), f"{len(fr)} vs {len(sr)}")
    check("gzip on the wire (Content-Encoding: gzip)", full["enc"] == "gzip" and slim["enc"] == "gzip",
          f"full={full['enc']} slim={slim['enc']} -- gzip middleware not deployed, or a proxy is stripping it")
    check("slim rows all say slim=true", all(x.get("slim") is True for x in sr))
    check("full rows say slim=false", all(x.get("slim") is False for x in fr))
    leaked = [f for x in sr for f in SLIM_DEFERRED if x.get(f) not in ("", [], None)]
    check("slim rows are empty on the six long-text fields", not leaked, f"{len(leaked)} non-empty values")
    light = [k for k in (fr[0] if fr else {}) if k not in SLIM_DEFERRED and k != "slim"]
    by_id = {x["id"]: x for x in sr}
    diff = [x["id"] for x in fr if x["id"] in by_id and any(x.get(k) != by_id[x["id"]].get(k) for k in light)]
    # A lead touched between the two calls can differ legitimately; more than a couple means a real mismatch.
    check("every other field matches the full list", len(diff) <= 2, f"{len(diff)} leads differ, e.g. ids {diff[:5]}")
    has_text = {f: sum(1 for x in fr if x.get(f)) for f in SLIM_DEFERRED}
    print(f"  full list has text in: " + ", ".join(f"{f} {n}" for f, n in has_text.items()))

    print("\nSingle lead (what a card fetches when opened)")
    pick = next((x for x in fr if x.get("body") or x.get("phone_script")), fr[0] if fr else None)
    if pick is None:
        print("  (no leads at all -- skipped)")
    else:
        one, one_med = timed(client, f"/leads/{pick['id']}", None, args.repeat)
        d = one["data"] or {}
        print(f"  GET /leads/{pick['id']}: {one['status']}, {kb(one['wire'])} on the wire, first {one['ms']:.0f}ms, median {one_med:.0f}ms")
        check("200", one["status"] == 200, str(one["status"]))
        check("full (slim=false)", d.get("slim") is False)
        check("long text present and equal to the full list's",
              all(d.get(f) == pick.get(f) for f in SLIM_DEFERRED))
        check("slim list row for the same lead is empty where this one has text",
              not any(d.get(f) and by_id.get(pick["id"], {}).get(f) for f in SLIM_DEFERRED))

    print("\nBell poll (GET /notifications/unread-count)")
    n, n_med = timed(client, "/notifications/unread-count", None, max(args.repeat, 5))
    print(f"  {n['status']} first {n['ms']:.0f}ms, median {n_med:.0f}ms  -> {n['data']}")
    print("  (with only the API key there is no signed-in user, so count is 0; the latency is what matters here)")
    check("200", n["status"] == 200, str(n["status"]))

    print(f"\n{'FAILED: ' + str(FAILED) if FAILED else 'All checks passed.'}")
    return FAILED


if __name__ == "__main__":
    raise SystemExit(main())