"""Admin API (`X-CA-Admin-Token`). Nothing is ever deleted: actions update `status` columns and append audit rows in payloads."""
from __future__ import annotations

from datetime import UTC, datetime
from typing import Any

from fastapi import APIRouter, Depends, HTTPException, Query, Response
from pydantic import BaseModel, Field

from companyatlas import archive
from companyatlas.api import aggregates as agg
from companyatlas.api import queries as q
from companyatlas.api import serializers as ser
from companyatlas.api.common import NO_STORE, PageDep, cache, page_payload, require_admin
from companyatlas.config import settings
from companyatlas.db import connection, execute, fetch_all, fetch_one, fetch_val, jsonb, transaction
from companyatlas.ids import new_id, slugify
from companyatlas.taxonomy import FailureClass, SensorStatus, tier_for_interval
from companyatlas.urls import canonicalize_url, registrable_domain

ORDER = 10
router = APIRouter(prefix="/api/v1/admin", tags=["admin"], dependencies=[Depends(require_admin)])
LOW_QUALITY = 30.0
STALE_DAYS = 7
WORKER_WINDOW_MIN = 15


def _ns(response: Response) -> None:
    response.headers["cache-control"] = NO_STORE


# ------------------------------------------------------------------------------------------------ overview


@router.get("/overview")
async def overview(response: Response) -> dict[str, Any]:
    _ns(response)
    today = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0)
    async with connection() as conn:
        cs = await fetch_all(conn, "select status, count(*) as n from companies group by status")
        cos = await fetch_all(conn, "select onboarding_status, count(*) as n from companies group by onboarding_status")
        ss = await fetch_all(conn, "select status, count(*) as n from sensors group by status")
        st = await fetch_all(conn, "select tier, count(*) as n from sensors where status <> 'retired' group by tier")
        qs = await fetch_one(conn, "select count(*) filter (where status = 'pending') as pending, count(*) filter (where status = 'running') as running, "
                                   "count(*) filter (where status = 'dead') as dead, count(*) filter (where status = 'failed') as failed, "
                                   "extract(epoch from (now() - min(run_at) filter (where status = 'pending' and run_at <= now()))) as oldest_pending_s from queue_jobs") or {}
        llm = await fetch_one(conn, "select count(*) filter (where status = 'pending') as pending, count(*) filter (where status = 'done' and finished_at >= :t) as done_today, "
                                    "count(*) filter (where status = 'failed' and finished_at >= :t) as failed_today from llm_jobs", t=today) or {}
        fails = await fetch_all(conn, "select failure_class, count(*) as n from failures where at >= now() - interval '24 hours' group by failure_class order by n desc")
        rates = await fetch_one(conn, "select (select count(*) from observations where fetched_at >= now() - interval '1 hour') as fetch_rate_1h, "
                                      "(select count(*) from changes where detected_at >= now() - interval '1 hour') as change_rate_1h, "
                                      "(select count(*) from changes where detected_at >= now() - interval '1 hour' and kind in ('meaningful','major','critical')) as meaningful_rate_1h") or {}
        workers = await fetch_all(conn, "select worker as name, max(fetched_at) as last_seen_at, count(*) as fetches_15m from observations "
                                        "where fetched_at >= now() - make_interval(mins => :m) and worker is not null group by worker order by last_seen_at desc limit 50",
                                  m=WORKER_WINDOW_MIN)
        inflight = {r["claimed_by"]: int(r["n"]) for r in await fetch_all(conn, "select claimed_by, count(*) as n from sensors where claimed_by is not null group by claimed_by")}
        runs = await fetch_all(conn, "select worker as name, max(started_at) as last_seen_at from crawl_runs where started_at >= now() - make_interval(mins => :m) "
                                     "and worker is not null group by worker", m=WORKER_WINDOW_MIN)
        cost = await fetch_all(conn, "select dimension, sum(cost_estimate) as cost, sum(units) as units from cost_ledger where day = current_date group by dimension")
        periodic = await q.settings_value(conn, "scheduler:heartbeat")
    names = {w["name"] for w in workers} | {r["name"] for r in runs} | set(inflight)
    seen = {w["name"]: w["last_seen_at"] for w in workers}
    for r in runs:
        if r["name"] not in seen or (r["last_seen_at"] and seen[r["name"]] and r["last_seen_at"] > seen[r["name"]]):
            seen[r["name"]] = r["last_seen_at"]
    done_today, failed_today = int(llm.get("done_today") or 0), int(llm.get("failed_today") or 0)
    costs = {r["dimension"]: round(float(r["cost"] or 0), 4) for r in cost}
    return {"companies_by_status": {r["status"]: int(r["n"]) for r in cs}, "companies_by_onboarding": {r["onboarding_status"]: int(r["n"]) for r in cos},
            "sensors_by_status": {r["status"]: int(r["n"]) for r in ss}, "sensors_by_tier": {(r["tier"] or "").strip(): int(r["n"]) for r in st},
            "queue": {"pending": int(qs.get("pending") or 0), "running": int(qs.get("running") or 0), "dead": int(qs.get("dead") or 0),
                      "failed": int(qs.get("failed") or 0), "oldest_pending_s": round(float(qs["oldest_pending_s"]), 1) if qs.get("oldest_pending_s") is not None else 0.0},
            "llm": {"pending": int(llm.get("pending") or 0), "done_today": done_today, "failed_today": failed_today,
                    "budget_left": max(0, settings.llm_daily_budget - done_today - failed_today), "budget": settings.llm_daily_budget, "configured": settings.llm_configured},
            "failures_24h_by_class": {r["failure_class"]: int(r["n"]) for r in fails}, "fetch_rate_1h": int(rates.get("fetch_rate_1h") or 0),
            "change_rate_1h": int(rates.get("change_rate_1h") or 0), "meaningful_rate_1h": int(rates.get("meaningful_rate_1h") or 0),
            "storage": await agg.archive_stats(),
            "workers": [{"name": n, "last_seen_at": seen.get(n), "inflight": inflight.get(n, 0)} for n in sorted(names)],
            "cost_today": {"fetch": costs.get("fetch", 0.0), "browser": costs.get("browser", 0.0), "llm": costs.get("llm", 0.0), "total": round(sum(costs.values()), 4)},
            "scheduler_heartbeat": periodic, "time": datetime.now(UTC)}


# ------------------------------------------------------------------------------------------------ connectors


@router.get("/connectors")
async def connectors(response: Response) -> dict[str, Any]:
    _ns(response)
    async with connection() as conn:
        rows = await fetch_all(conn, "select * from connectors order by category, id")
        sensors = await fetch_all(conn, "select connector_id, count(*) filter (where status = 'active') as active, "
                                        "count(*) filter (where status in ('failing','stale','blocked')) as failing, count(*) filter (where status <> 'retired') as total, "
                                        "max(last_run_at) as last_run_at from sensors group by connector_id")
        obs = await fetch_all(conn, "select s.connector_id, count(*) as n, count(*) filter (where o.failure_class is null) as ok, avg(o.duration_ms) as latency, "
                                    "count(*) filter (where o.changed) as changed, count(*) filter (where o.failure_class is not null) as errors "
                                    "from observations o join sensors s on s.id = o.sensor_id where o.fetched_at >= now() - interval '24 hours' group by s.connector_id")
    sm = {r["connector_id"]: r for r in sensors}
    om = {r["connector_id"]: r for r in obs}
    items = []
    known = {r["id"] for r in rows}
    for cid in list(known) + [k for k in sm if k not in known]:
        row = next((r for r in rows if r["id"] == cid), None)
        base = ser.connector(row) if row else {"id": cid, "name": cid, "version": "?", "category": "?", "enabled": True}
        s, o = sm.get(cid, {}), om.get(cid, {})
        n = int(o.get("n") or 0)
        base.update({"sensors_active": int(s.get("active") or 0), "sensors_failing": int(s.get("failing") or 0), "sensors_total": int(s.get("total") or 0),
                     "success_rate_24h": round(int(o.get("ok") or 0) / n, 4) if n else None, "avg_latency_ms": round(float(o["latency"]), 1) if o.get("latency") is not None else None,
                     "change_rate_24h": round(int(o.get("changed") or 0) / n, 4) if n else None, "errors_24h": int(o.get("errors") or 0), "fetches_24h": n,
                     "last_run_at": s.get("last_run_at")})
        items.append(base)
    items.sort(key=lambda x: (x.get("category") or "", x["id"]))
    return {"items": items}


# ------------------------------------------------------------------------------------------------ sensors

SENSOR_FILTERS: dict[str, str] = {
    "healthy": "s.status = 'active' and s.consecutive_failures = 0",
    "failing": "(s.status = 'failing' or s.consecutive_failures > 0)",
    "stale": "(s.status = 'stale' or (s.status in ('active','failing') and (s.last_success_at is null or s.last_success_at < now() - make_interval(days => :stale_days))))",
    "blocked": "(s.status = 'blocked' or s.last_failure_class in ('BOT_CHALLENGE','ROBOTS','BLOCKED_DESTINATION','RATE_LIMIT'))",
    "redirected": "(s.status = 'redirected' or s.last_failure_class = 'REDIRECT')",
    "low_quality": "s.quality_score < :low_quality",
    "high_activity": "s.last_change_at >= now() - interval '24 hours'",
}
SENSOR_SORTS = {"recent": "s.last_run_at desc nulls last, s.id", "next_run": "s.next_run_at asc, s.id", "failures": "s.consecutive_failures desc, s.last_run_at desc nulls last, s.id",
                "quality": "s.quality_score asc, s.id", "changes": "s.change_count desc, s.id", "created": "s.created_at desc, s.id"}
SENSOR_SELECT = ("select s.*, c.slug as company_slug, c.display_name as company_display_name, c.canonical_domain as company_domain, c.country as company_country, "
                 "c.logo_url as company_logo_url from sensors s join companies c on c.id = s.company_id")


@router.get("/sensors")
async def admin_sensors(response: Response, p: PageDep, status: str | None = None, domain: str | None = None, connector: str | None = None,
                        company: str | None = None, surface: str | None = None,
                        filter: str | None = Query(None, pattern="^(healthy|failing|stale|blocked|redirected|low_quality|high_activity)$"),
                        sort: str = Query("recent", pattern="^(recent|next_run|failures|quality|changes|created)$")) -> dict[str, Any]:
    _ns(response)
    where, params = ["true"], {"stale_days": STALE_DAYS, "low_quality": LOW_QUALITY}
    if status:
        where.append("s.status = cast(:status as text)")
        params["status"] = status.lower()[:20]
    if domain:
        where.append("s.domain = cast(:domain as text)")
        params["domain"] = domain.lower()[:253]
    if connector:
        where.append("s.connector_id = cast(:connector as text)")
        params["connector"] = connector[:80]
    if surface:
        where.append("s.surface = cast(:surface as text)")
        params["surface"] = surface.lower()[:40]
    if filter:
        where.append(SENSOR_FILTERS[filter])
        if filter == "high_activity":
            sort = "changes"
    async with connection() as conn:
        if company:
            params["company_id"] = (await q.require_company(conn, company))["id"]
            where.append("s.company_id = :company_id")
        wsql = " and ".join(where)
        rows = await fetch_all(conn, f"{SENSOR_SELECT} where {wsql} order by {SENSOR_SORTS[sort]} limit :limit offset :offset", **params, limit=p.per_page, offset=p.offset)
        total = await q.bounded_count(conn, f"from sensors s join companies c on c.id = s.company_id where {wsql}", params)
    items = []
    for r in rows:
        item = ser.sensor_admin(r)
        item["company"] = ser.company_ref(r)
        items.append(item)
    return page_payload(items, total, p)


class SensorActionBody(BaseModel):
    interval_s: int | None = Field(None, ge=60, le=90 * 86400)
    connector_id: str | None = Field(None, max_length=80)
    reason: str | None = Field(None, max_length=500)


SENSOR_ACTIONS = ("pause", "resume", "retry", "rediscover", "retire", "run_now", "set_interval", "set_connector")


@router.post("/sensors/{sensor_id}/{action}")
async def sensor_action(sensor_id: str, action: str, response: Response, body: SensorActionBody | None = None) -> dict[str, Any]:
    _ns(response)
    if action not in SENSOR_ACTIONS:
        raise HTTPException(status_code=404, detail=f"unknown action (one of {', '.join(SENSOR_ACTIONS)})")
    body = body or SensorActionBody()
    now = datetime.now(UTC)
    async with transaction() as conn:
        s = await fetch_one(conn, "select * from sensors where id = :id", id=sensor_id)
        if s is None:
            raise HTTPException(status_code=404, detail="sensor not found")
        extra: dict[str, Any] = {}
        if action == "pause":
            await execute(conn, "update sensors set status = :st, claimed_by = null, claimed_at = null, updated_at = :now where id = :id", st=SensorStatus.PAUSED.value, now=now, id=sensor_id)
        elif action == "resume":
            await execute(conn, "update sensors set status = :st, consecutive_failures = 0, next_run_at = :now, retired_at = null, updated_at = :now where id = :id",
                          st=SensorStatus.ACTIVE.value, now=now, id=sensor_id)
        elif action in ("retry", "run_now"):
            prio = ", priority = greatest(priority, 0.95)" if action == "run_now" else ""
            await execute(conn, f"update sensors set next_run_at = :now, claimed_by = null, claimed_at = null{prio}, updated_at = :now where id = :id", now=now, id=sensor_id)
        elif action == "retire":
            await execute(conn, "update sensors set status = :st, retired_at = :now, claimed_by = null, claimed_at = null, updated_at = :now where id = :id",
                          st=SensorStatus.RETIRED.value, now=now, id=sensor_id)
        elif action == "rediscover":
            jid = new_id("queue_job")
            key = f"discover:{s['company_id']}:{int(now.timestamp())}"
            await execute(conn, "insert into queue_jobs (id, kind, key, payload, priority) values (:id, 'discover', :key, cast(:p as jsonb), 0.9) on conflict (key) do nothing",
                          id=jid, key=key, p=jsonb({"company_id": s["company_id"], "sensor_id": sensor_id, "reason": body.reason or "admin:rediscover"}))
            extra["queued"] = {"id": jid, "key": key}
        elif action == "set_interval":
            if body.interval_s is None:
                raise HTTPException(status_code=422, detail="interval_s required")
            iv = max(settings.min_interval_s, min(settings.max_interval_s, body.interval_s))
            await execute(conn, "update sensors set base_interval_s = :iv, current_interval_s = :iv, tier = :tier, next_run_at = :now, updated_at = :now where id = :id",
                          iv=iv, tier=tier_for_interval(iv), now=now, id=sensor_id)
            extra["interval_s"] = iv
        elif action == "set_connector":
            if not body.connector_id:
                raise HTTPException(status_code=422, detail="connector_id required")
            ok = await fetch_val(conn, "select enabled from connectors where id = :c", c=body.connector_id)
            if ok is None:
                raise HTTPException(status_code=422, detail="unknown connector_id")
            await execute(conn, "update sensors set connector_id = :c, next_run_at = :now, updated_at = :now where id = :id", c=body.connector_id, now=now, id=sensor_id)
        row = await fetch_one(conn, f"{SENSOR_SELECT} where s.id = :id", id=sensor_id)
    out = ser.sensor_admin(row or {})
    out["company"] = ser.company_ref(row or {})
    return {"ok": True, "action": action, "sensor": out, **extra}


# ------------------------------------------------------------------------------------------------ companies


@router.get("/companies")
async def admin_companies(response: Response, p: PageDep, onboarding_status: str | None = None, status: str | None = None,
                          q_: str | None = Query(None, alias="q", max_length=200), country: str | None = None,
                          sort: str = Query("recent", pattern="^(activity|events|hiring|name|importance|recent)$")) -> dict[str, Any]:
    _ns(response)
    where, params = q.company_filters(q=q_, country=country, status=status, onboarding_status=onboarding_status)
    async with connection() as conn:
        ids, total = await q.company_page_ids(conn, where, params, sort=sort, limit=p.per_page, offset=p.offset)
        rows = await q.fetch_cards_by_ids(conn, ids)
    items = []
    for r in rows:
        card = ser.company_card(r)
        card.update({"onboarding_error": r.get("onboarding_error"), "indexed": bool(r.get("indexed")), "discovered_at": r.get("discovered_at"), "updated_at": r.get("updated_at")})
        items.append(card)
    return page_payload(items, total, p)


class CompanyCreate(BaseModel):
    website: str = Field(min_length=4, max_length=500)
    display_name: str | None = Field(None, max_length=200)
    country: str | None = Field(None, min_length=2, max_length=2)
    industries: list[str] | None = None
    importance: float | None = Field(None, ge=0, le=1)
    tier: int | None = Field(None, ge=1, le=4)


@router.post("/companies", status_code=201)
async def admin_create_company(body: CompanyCreate, response: Response) -> dict[str, Any]:
    _ns(response)
    website = body.website.strip()
    if "://" not in website:
        website = "https://" + website
    if not website.lower().startswith(("http://", "https://")):
        raise HTTPException(status_code=422, detail="website: only http(s) URLs are accepted")
    domain = registrable_domain(website)
    if not domain or "." not in domain:
        raise HTTPException(status_code=422, detail="website: could not derive a registrable domain")
    website = canonicalize_url(website)
    display_name = (body.display_name or "").strip() or domain.split(".")[0].capitalize()
    country = body.country.upper() if body.country else None
    industries = [slugify(i) for i in (body.industries or []) if i.strip()][:10]
    async with transaction() as conn:
        existing = await fetch_one(conn, "select id, slug from companies where canonical_domain = :d", d=domain)
        if existing:
            raise HTTPException(status_code=409, detail=f"company already exists: {existing['slug']}")
        if country and not await fetch_val(conn, "select 1 from countries where code = cast(:c as char(2))", c=country):
            raise HTTPException(status_code=422, detail="country: unknown ISO-2 code")
        if industries:
            known = {r["slug"] for r in await fetch_all(conn, "select slug from industries where slug = any(cast(:s as text[]))", s=industries)}
            missing = [i for i in industries if i not in known]
            if missing:
                raise HTTPException(status_code=422, detail=f"industries: unknown slugs {', '.join(missing)}")
        base = slugify(display_name)
        slug, n = base, 2
        while await fetch_val(conn, "select 1 from companies where slug = :s", s=slug):
            slug = f"{base}-{n}"
            n += 1
        cid = new_id("company")
        await execute(conn, "insert into companies (id, slug, display_name, canonical_domain, website, country, industries, industry_primary, importance, tier, "
                            "source_meta) values (:id, :slug, :name, :domain, :website, :country, cast(:inds as text[]), :ip, :imp, :tier, cast(:meta as jsonb))",
                      id=cid, slug=slug, name=display_name, domain=domain, website=website, country=country, inds=industries, ip=industries[0] if industries else None,
                      imp=body.importance if body.importance is not None else 0.2, tier=body.tier or 4, meta=jsonb({"source": "admin_api", "created_at": datetime.now(UTC)}))
        await execute(conn, "insert into domains (id, company_id, domain, kind) values (:id, :cid, :d, 'primary') on conflict do nothing", id=new_id("domain"), cid=cid, d=domain)
        jid = new_id("queue_job")
        await execute(conn, "insert into queue_jobs (id, kind, key, payload, priority) values (:id, 'discover', :key, cast(:p as jsonb), 0.8) on conflict (key) do nothing",
                      id=jid, key=f"discover:{cid}", p=jsonb({"company_id": cid, "reason": "admin:create"}))
        cards = await q.fetch_cards_by_ids(conn, [cid])
    return {"ok": True, "company": ser.company_card(cards[0]), "queued": {"id": jid, "kind": "discover"}}


@router.post("/companies/{key}/rediscover")
async def admin_rediscover(key: str, response: Response) -> dict[str, Any]:
    _ns(response)
    now = datetime.now(UTC)
    async with transaction() as conn:
        c = await q.require_company(conn, key)
        jid = new_id("queue_job")
        jkey = f"discover:{c['id']}:{int(now.timestamp())}"
        await execute(conn, "insert into queue_jobs (id, kind, key, payload, priority) values (:id, 'discover', :key, cast(:p as jsonb), 0.9)",
                      id=jid, key=jkey, p=jsonb({"company_id": c["id"], "reason": "admin:rediscover"}))
        if c["onboarding_status"] in ("failed", "no_website"):
            await execute(conn, "update companies set onboarding_status = 'pending', onboarding_error = null, updated_at = :now where id = :id", now=now, id=c["id"])
    return {"ok": True, "company": c["slug"], "queued": {"id": jid, "key": jkey}}


# ------------------------------------------------------------------------------------------------ failures / queue / llm / reviews


@router.get("/failures")
async def admin_failures(response: Response, p: PageDep, class_: str | None = Query(None, alias="class", max_length=40), since: str | None = None,
                         sensor: str | None = None, company: str | None = None) -> dict[str, Any]:
    _ns(response)
    where, params = ["true"], {}
    if class_:
        where.append("f.failure_class = cast(:fc as text)")
        params["fc"] = class_.upper()
    since_dt = q.parse_iso(since) or q.days_ago(7)
    where.append("f.at >= :since")
    params["since"] = since_dt
    if sensor:
        where.append("f.sensor_id = :sid")
        params["sid"] = sensor
    async with connection() as conn:
        if company:
            params["cid"] = (await q.require_company(conn, company))["id"]
            where.append("f.company_id = :cid")
        wsql = " and ".join(where)
        rows = await fetch_all(conn, f"select f.*, c.slug as company_slug, s.surface from failures f left join companies c on c.id = f.company_id "
                                     f"left join sensors s on s.id = f.sensor_id where {wsql} order by f.at desc limit :limit offset :offset", **params, limit=p.per_page, offset=p.offset)
        total = await q.bounded_count(conn, f"from failures f where {wsql}", params)
        by_class = await fetch_all(conn, f"select f.failure_class, count(*) as n from failures f where {wsql} group by 1 order by n desc", **params)
    out = page_payload([ser.failure(r) for r in rows], total, p)
    out["by_class"] = {r["failure_class"]: int(r["n"]) for r in by_class}
    out["classes"] = [c.value for c in FailureClass]
    return out


@router.get("/queue")
async def admin_queue(response: Response, kind: str | None = None, status: str | None = None, limit: int = Query(100, ge=1, le=500)) -> dict[str, Any]:
    _ns(response)
    where, params = ["true"], {"limit": limit}
    if kind:
        where.append("kind = cast(:kind as text)")
        params["kind"] = kind[:40]
    if status:
        where.append("status = cast(:status as text)")
        params["status"] = status[:20]
    async with connection() as conn:
        counts = await fetch_all(conn, "select kind, status, count(*) as n from queue_jobs group by kind, status order by kind, status")
        rows = await fetch_all(conn, f"select * from queue_jobs where {' and '.join(where)} order by case status when 'running' then 0 when 'pending' then 1 else 2 end, "
                                     "run_at desc limit :limit", **params)
    return {"counts": [{"kind": r["kind"], "status": r["status"], "n": int(r["n"])} for r in counts], "items": [ser.queue_job(r) for r in rows]}


class RequeueBody(BaseModel):
    kind: str | None = Field(None, max_length=40)


@router.post("/queue/requeue-dead")
async def admin_requeue_dead(response: Response, body: RequeueBody | None = None) -> dict[str, Any]:
    _ns(response)
    body = body or RequeueBody()
    extra, params = "", {}
    if body.kind:
        extra = " and kind = cast(:kind as text)"
        params["kind"] = body.kind
    async with transaction() as conn:
        n = await fetch_val(conn, "with u as (update queue_jobs set status = 'pending', attempts = 0, run_at = now(), locked_at = null, locked_by = null, "
                                  f"last_error = null, finished_at = null where status = 'dead'{extra} returning 1) select count(*) from u", **params)
    return {"ok": True, "requeued": int(n or 0)}


@router.get("/llm")
async def admin_llm(response: Response, p: PageDep, status: str | None = None, kind: str | None = None) -> dict[str, Any]:
    _ns(response)
    where, params = ["true"], {}
    if status:
        where.append("status = cast(:status as text)")
        params["status"] = status[:20]
    if kind:
        where.append("kind = cast(:kind as text)")
        params["kind"] = kind[:40]
    wsql = " and ".join(where)
    async with connection() as conn:
        rows = await fetch_all(conn, f"select * from llm_jobs where {wsql} order by created_at desc limit :limit offset :offset", **params, limit=p.per_page, offset=p.offset)
        total = await q.bounded_count(conn, f"from llm_jobs where {wsql}", params)
        stats = await fetch_one(conn, "select count(*) filter (where status = 'pending') as pending, count(*) filter (where status = 'running') as running, "
                                      "count(*) filter (where status = 'done' and finished_at >= current_date) as done_today, "
                                      "count(*) filter (where status = 'failed' and finished_at >= current_date) as failed_today, "
                                      "avg(latency_ms) filter (where status = 'done' and finished_at >= current_date) as avg_latency_ms, "
                                      "sum(request_tokens + coalesce(response_tokens, 0)) filter (where finished_at >= current_date) as tokens_today from llm_jobs") or {}
    out = page_payload([ser.llm_job(r) for r in rows], total, p)
    out["stats"] = {k: (round(float(v), 1) if k == "avg_latency_ms" and v is not None else (int(v) if v is not None else 0)) for k, v in stats.items()}
    out["stats"]["budget"] = settings.llm_daily_budget
    out["stats"]["configured"] = settings.llm_configured
    return out


@router.get("/reviews")
async def admin_reviews(response: Response, p: PageDep, kind: str | None = None, status: str = Query("open", max_length=20)) -> dict[str, Any]:
    _ns(response)
    where, params = ["true"], {}
    if kind:
        where.append("r.kind = cast(:kind as text)")
        params["kind"] = kind[:40]
    if status and status != "all":
        where.append("r.status = cast(:status as text)")
        params["status"] = status
    wsql = " and ".join(where)
    async with connection() as conn:
        rows = await fetch_all(conn, f"select r.*, c.slug as company_slug, c.display_name as company_display_name from review_queue r left join companies c on c.id = r.company_id "
                                     f"where {wsql} order by r.created_at asc limit :limit offset :offset", **params, limit=p.per_page, offset=p.offset)
        total = await q.bounded_count(conn, f"from review_queue r where {wsql}", params)
        by_kind = await fetch_all(conn, "select kind, count(*) as n from review_queue where status = 'open' group by kind order by n desc")
    out = page_payload([ser.review(r) for r in rows], total, p)
    out["open_by_kind"] = {r["kind"]: int(r["n"]) for r in by_kind}
    return out


class ReviewBody(BaseModel):
    resolution: str = Field(pattern="^(accepted|rejected)$")
    note: str | None = Field(None, max_length=1000)
    label: str | None = Field(None, pattern="^(correct|duplicate|noise|misclassified)$")


@router.post("/reviews/{review_id}")
async def admin_resolve_review(review_id: str, body: ReviewBody, response: Response) -> dict[str, Any]:
    _ns(response)
    async with transaction() as conn:
        r = await fetch_one(conn, "select * from review_queue where id = :id", id=review_id)
        if r is None:
            raise HTTPException(status_code=404, detail="review not found")
        if r["status"] != "open":
            raise HTTPException(status_code=409, detail=f"review already {r['status']}")
        payload = ser._dict(r["payload"])
        payload["resolution"] = {"status": body.resolution, "note": body.note, "label": body.label, "at": datetime.now(UTC)}
        await execute(conn, "update review_queue set status = :st, resolution = :res, resolved_at = now(), payload = cast(:p as jsonb) where id = :id",
                      st=body.resolution, res=body.label or body.note or body.resolution, p=jsonb(payload), id=review_id)
        row = await fetch_one(conn, "select * from review_queue where id = :id", id=review_id)
    return {"ok": True, "review": ser.review(row or {})}


# ------------------------------------------------------------------------------------------------ events (corrections)


class RetractBody(BaseModel):
    reason: str = Field(min_length=3, max_length=500)


async def _audit(conn: Any, event_id: str, action: str, reason: str | None) -> None:
    entry = jsonb([{"action": action, "reason": reason, "at": datetime.now(UTC)}])
    await execute(conn, "update events set payload = jsonb_set(payload, '{_audit}', coalesce(payload->'_audit', '[]'::jsonb) || cast(:e as jsonb), true) where id = :id",
                  e=entry, id=event_id)


@router.post("/events/{event_id}/retract")
async def admin_retract_event(event_id: str, body: RetractBody, response: Response) -> dict[str, Any]:
    _ns(response)
    async with transaction() as conn:
        ev = await fetch_one(conn, "select id, status from events where id = :id", id=event_id)
        if ev is None:
            raise HTTPException(status_code=404, detail="event not found")
        await execute(conn, "update events set status = 'retracted', retracted_reason = :r where id = :id", r=body.reason.strip(), id=event_id)
        await _audit(conn, event_id, "retract", body.reason.strip())
        row = await q.fetch_event(conn, event_id)
    cache.clear()
    return {"ok": True, "event": ser.event(row or {})}


@router.post("/events/{event_id}/restore")
async def admin_restore_event(event_id: str, response: Response) -> dict[str, Any]:
    _ns(response)
    async with transaction() as conn:
        ev = await fetch_one(conn, "select id, status, retracted_reason from events where id = :id", id=event_id)
        if ev is None:
            raise HTTPException(status_code=404, detail="event not found")
        await execute(conn, "update events set status = 'active', retracted_reason = null where id = :id", id=event_id)
        await _audit(conn, event_id, "restore", ev.get("retracted_reason"))
        row = await q.fetch_event(conn, event_id)
    cache.clear()
    return {"ok": True, "event": ser.event(row or {})}


# ------------------------------------------------------------------------------------------------ quality / costs / cache


@router.get("/quality")
async def admin_quality(response: Response) -> dict[str, Any]:
    _ns(response)
    async with connection() as conn:
        r = await fetch_one(conn, """
            select (select count(*) from companies) as companies,
                   (select count(*) from companies where onboarding_status = 'active' and status = 'ACTIVE') as companies_active,
                   (select count(*) from sensors where status <> 'retired') as sensors,
                   (select count(*) from sensors where status = 'active') as sensors_active,
                   (select count(*) from sensors where status <> 'retired' and last_run_at >= now() - interval '24 hours') as checked_24h,
                   (select count(*) from sensors where status = 'stale') as stale,
                   (select count(*) from sensors where status in ('failing', 'blocked')) as failed_sensors,
                   (select count(*) from sensors where surface = 'other' and status <> 'retired') as unknown_surfaces,
                   (select count(*) from events where detected_at >= now() - interval '30 days') as events_30d,
                   (select count(*) from events where detected_at >= now() - interval '30 days' and status = 'duplicate') as duplicates_30d,
                   (select avg(confidence) from events where detected_at >= now() - interval '30 days' and status = 'active') as confidence_avg
        """) or {}
        cal = await fetch_all(conn, "select coalesce(payload->'resolution'->>'label', resolution) as label, count(*) as n from review_queue "
                                    "where status in ('accepted','rejected','resolved') group by 1")
    companies, sensors = int(r.get("companies") or 0), int(r.get("sensors") or 0)
    ev30 = int(r.get("events_30d") or 0)
    labels = {c["label"]: int(c["n"]) for c in cal if c["label"]}
    return {"coverage": {"companies_active_pct": round(int(r.get("companies_active") or 0) / companies * 100, 1) if companies else None,
                         "sensors_active_pct": round(int(r.get("sensors_active") or 0) / sensors * 100, 1) if sensors else None,
                         "companies": companies, "sensors": sensors},
            "freshness": {"sensors_checked_24h_pct": round(int(r.get("checked_24h") or 0) / sensors * 100, 1) if sensors else None, "stale": int(r.get("stale") or 0)},
            "duplicate_rate": round(int(r.get("duplicates_30d") or 0) / ev30, 4) if ev30 else None,
            "event_confidence_avg": round(float(r["confidence_avg"]), 3) if r.get("confidence_avg") is not None else None,
            "unknown_surfaces": int(r.get("unknown_surfaces") or 0), "failed_sensors": int(r.get("failed_sensors") or 0),
            "calibration": {k: labels.get(k, 0) for k in ("correct", "duplicate", "noise", "misclassified")}, "events_30d": ev30}


@router.get("/costs")
async def admin_costs(response: Response, days: int = Query(30, ge=1, le=365)) -> dict[str, Any]:
    _ns(response)
    async with connection() as conn:
        items = await fetch_all(conn, "select day, dimension, key, units, cost_estimate from cost_ledger where day >= :d order by day desc, dimension, key limit 5000",
                                d=q.days_ago(days).date())
        denom = await fetch_one(conn, "select (select count(*) from companies where status = 'ACTIVE') as companies, "
                                      "(select count(*) from observations where fetched_at >= :d) as observations, "
                                      "(select count(*) from events where status = 'active' and detected_at >= :d and importance >= :imp) as meaningful_events",
                                d=q.days_ago(days), imp=settings.meaningful_threshold) or {}
    total = sum(float(i["cost_estimate"] or 0) for i in items)
    by_dim: dict[str, float] = {}
    for i in items:
        by_dim[i["dimension"]] = by_dim.get(i["dimension"], 0.0) + float(i["cost_estimate"] or 0)
    comp, obs, mev = int(denom.get("companies") or 0), int(denom.get("observations") or 0), int(denom.get("meaningful_events") or 0)
    return {"days": days, "items": [{"day": i["day"], "dimension": i["dimension"], "key": i["key"], "units": float(i["units"] or 0), "cost_estimate": float(i["cost_estimate"] or 0)}
                                    for i in items],
            "total": round(total, 4), "by_dimension": {k: round(v, 4) for k, v in by_dim.items()},
            "per_1000_companies": round(total / comp * 1000, 4) if comp else None, "per_million_observations": round(total / obs * 1_000_000, 4) if obs else None,
            "per_meaningful_event": round(total / mev, 4) if mev else None}


@router.post("/cache/clear")
async def admin_cache_clear(response: Response, prefix: str | None = Query(None, max_length=60)) -> dict[str, Any]:
    _ns(response)
    cache.clear(prefix)
    return {"ok": True, "cleared": prefix or "all"}


@router.get("/storage")
async def admin_storage(response: Response) -> dict[str, Any]:
    _ns(response)
    stats = await agg.archive_stats()
    return {"objects_dir": str(settings.objects_dir), "exists": archive.object_path("00" * 32).parent.parent.parent.exists(), **stats}