"""Admin API (`x-aia-admin-token`): overview, connectors, runs/errors, documents & snapshots (cleaned text only), queues, LLM
accounting, review queue with entity merging, duplicates, curation, infrastructure, cache. Never returns raw archive paths."""
from __future__ import annotations

import platform
import socket
import time
from datetime import UTC, datetime
from typing import Any

from fastapi import APIRouter, Depends, Query, Request
from pydantic import BaseModel, Field

from aiatlas.api.common import ADMIN_DEPENDENCIES, PAGINATION, ApiError, Pagination, audit, client_ip, page
from aiatlas.config import settings
from aiatlas.connectors import registry as connector_registry
from aiatlas.db import connection, execute, fetch_all, fetch_one, fetch_val, jsonb, transaction
from aiatlas.ids import new_id
from aiatlas.sdk import archive
from aiatlas.services import cache
from aiatlas.services.jobs import enqueue, queue_depth
from aiatlas.services.llm import gateway
from aiatlas.services.merge import merge_entities
from aiatlas.services.stats import live_counts

SKIP_AUDIT_GET = ("/api/v1/admin/overview", "/api/v1/admin/infrastructure", "/api/v1/admin/llm/health", "/api/v1/admin/audit")


async def audit_request(request: Request) -> None:
    """Every admin call leaves a row in `admin_audit_log` (except the dashboard polling GETs). Runs AFTER the rate limit and the token check."""
    if request.method == "GET" and request.url.path in SKIP_AUDIT_GET:
        return
    payload: dict[str, Any] = {"method": request.method, "path": request.url.path}
    if request.query_params:
        payload["query"] = dict(request.query_params)
    if request.method in ("POST", "PATCH", "PUT", "DELETE"):
        try:
            body = await request.body()
            if body:
                import orjson

                payload["body"] = orjson.loads(body) if len(body) < 64 * 1024 else {"truncated": True, "bytes": len(body)}
        except Exception:  # noqa: BLE001
            payload["body"] = {"unparsed": True}
    target = request.path_params.get("name") or request.path_params.get("review_id") or request.path_params.get("job_id") or request.path_params.get("snap_id") \
        or request.path_params.get("doc_id") or request.path_params.get("entity_id") or request.path_params.get("run_id") or request.path_params.get("anomaly_id") \
        or request.path_params.get("quarantine_id") or request.path_params.get("a")
    await audit(f"{request.method} {request.url.path}", str(target) if target else None, payload, client_ip(request))


router = APIRouter(prefix="/api/v1/admin", tags=["admin"], dependencies=[*ADMIN_DEPENDENCIES, Depends(audit_request)])
_STARTED = time.time()
SNAPSHOT_TEXT_LIMIT = 20 * 1024
HIDDEN = ("raw_path", "text_path")


def _public(row: dict[str, Any]) -> dict[str, Any]:
    return {k: v for k, v in row.items() if k not in HIDDEN}


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


@router.get("/overview")
async def overview() -> dict[str, Any]:
    async with connection() as conn:
        stats = await live_counts(conn)
        queue = await queue_depth(conn)
        health = await fetch_all(conn, "select health, count(*) as n from connectors group by 1")
        errors = await fetch_val(conn, "select count(*) from connector_errors where created_at > now() - interval '24 hours'")
        llm = await fetch_one(conn, """select count(*) as jobs_24h, coalesce(sum(input_tokens), 0) + coalesce(sum(output_tokens), 0) as tokens_24h,
                                       count(*) filter (where status <> 'ok') as failed_24h from llm_jobs where created_at > now() - interval '24 hours'""")
        by_stage = await fetch_all(conn, "select stage, status, count(*) as n from llm_jobs where created_at > now() - interval '24 hours' group by 1, 2 order by 1, 2")
        recent_runs = await fetch_all(conn, "select * from connector_runs order by started_at desc limit 12")
        review_kinds = await fetch_all(conn, "select kind, count(*) as n from review_queue where status = 'pending' group by 1")
    counts = {h["health"]: int(h["n"]) for h in health}
    return {"stats": stats, "queue": queue, "heartbeats": await cache.heartbeats(),
            "connectors": {k: counts.get(k, 0) for k in ("ok", "degraded", "failing", "disabled", "unknown")}, "review_pending": stats.get("review_pending", 0),
            "review_by_kind": {r["kind"]: int(r["n"]) for r in review_kinds}, "recent_errors": int(errors or 0),
            "llm": {"available": gateway.available, "jobs_24h": int(llm["jobs_24h"] or 0), "tokens_24h": int(llm["tokens_24h"] or 0), "failed_24h": int(llm["failed_24h"] or 0),
                    "by_stage": [{**r, "n": int(r["n"])} for r in by_stage]},
            "recent_runs": recent_runs, "archive": archive.archive_size(), "computed_at": datetime.now(UTC)}


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


@router.get("/connectors")
async def connectors() -> dict[str, Any]:
    async with connection() as conn:
        rows = await fetch_all(conn, """
            select c.*, s.key as source_key, s.name as source_name, s.tier as source_tier, s.domain as source_domain,
                   (select row_to_json(r) from connector_runs r where r.connector_name = c.name order by r.started_at desc limit 1) as last_run,
                   (select count(*) from documents d where d.connector_name = c.name) as documents,
                   (select count(*) from snapshots x join documents d on d.id = x.document_id where d.connector_name = c.name) as snapshots,
                   (select count(*) from connector_errors e where e.connector_name = c.name and e.created_at > now() - interval '7 days') as errors_7d
            from connectors c left join sources s on s.id = c.source_id order by c.priority, c.name""")
    known = connector_registry()
    return {"items": [{**r, "documents": int(r["documents"]), "snapshots": int(r["snapshots"]), "errors_7d": int(r["errors_7d"]), "in_code": r["name"] in known,
                       "run_now_pending": await _run_now_pending(r["name"])} for r in rows],
            "unregistered_in_db": sorted(set(known) - {r["name"] for r in rows})}


async def _run_now_pending(name: str) -> bool:
    try:
        return bool(await cache.redis().exists(f"aia:run-now:{name}"))
    except Exception:  # noqa: BLE001
        return False


class RunBody(BaseModel):
    force: bool = False


@router.post("/connectors/{name}/run")
async def run_connector(name: str, body: RunBody | None = None) -> dict[str, Any]:
    async with connection() as conn:
        exists = await fetch_one(conn, "select name from connectors where name = :n", n=name)
    if not exists and name not in connector_registry():
        raise ApiError(404, f"unknown connector {name!r}")
    try:
        await cache.redis().set(f"aia:run-now:{name}", b"force" if (body and body.force) else b"1", ex=6 * 3600)
    except Exception as exc:
        raise ApiError(503, f"redis unavailable: {exc.__class__.__name__}") from exc
    return {"queued": True, "connector": name, "force": bool(body and body.force), "note": "consumed by the scheduler tick (aia:run-now:<name>)"}


class ConnectorPatch(BaseModel):
    enabled: bool | None = None
    interval_seconds: int | None = Field(None, ge=60, le=30 * 86400)
    priority: int | None = Field(None, ge=0, le=9)


@router.patch("/connectors/{name}")
async def patch_connector(name: str, body: ConnectorPatch) -> dict[str, Any]:
    sets, params = [], {"n": name}
    if body.enabled is not None:
        sets.append("enabled = :enabled")
        sets.append("health = case when :enabled then (case when health = 'disabled' then 'unknown' else health end) else 'disabled' end")
        sets.append("circuit_open_until = case when :enabled then null else circuit_open_until end")
        params["enabled"] = body.enabled
    if body.interval_seconds is not None:
        sets.append("interval_seconds = :iv")
        sets.append("next_run_at = least(coalesce(next_run_at, now()), coalesce(last_success_at, now()) + make_interval(secs => :iv))")
        params["iv"] = body.interval_seconds
    if body.priority is not None:
        sets.append("priority = :pr")
        params["pr"] = body.priority
    if not sets:
        raise ApiError(400, "nothing to update")
    async with transaction() as conn:
        row = await fetch_one(conn, f"update connectors set {', '.join(sets)}, updated_at = now() where name = :n returning *", **params)
    if not row:
        raise ApiError(404, f"unknown connector {name!r}")
    return row


@router.get("/runs")
async def runs(connector: str | None = None, status: str | None = None, limit: int = Query(50, ge=1, le=500), offset: int = Query(0, ge=0)) -> dict[str, Any]:
    where = ["true"]
    params: dict[str, Any] = {"lim": limit, "off": offset}
    if connector:
        where.append("connector_name = :c")
        params["c"] = connector
    if status:
        where.append("status = :s")
        params["s"] = status
    async with connection() as conn:
        rows = await fetch_all(conn, f"select * from connector_runs where {' and '.join(where)} order by started_at desc limit :lim offset :off", **params)
        total = await fetch_val(conn, f"select count(*) from connector_runs where {' and '.join(where)}", **{k: v for k, v in params.items() if k not in ('lim', 'off')})
    return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset}


@router.get("/errors")
async def errors(connector: str | None = None, limit: int = Query(100, ge=1, le=1000), offset: int = Query(0, ge=0)) -> dict[str, Any]:
    where = "connector_name = :c" if connector else "true"
    async with connection() as conn:
        rows = await fetch_all(conn, f"select * from connector_errors where {where} order by created_at desc limit :lim offset :off", c=connector, lim=limit, off=offset)
        total = await fetch_val(conn, f"select count(*) from connector_errors where {where}", c=connector)
    return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset}


# ------------------------------------------------------------------------------------------------------------------ documents & snapshots


@router.get("/documents")
async def documents(connector: str | None = None, status: str | None = None, q: str | None = Query(None, max_length=300), entity: str | None = None,
                    needs_llm: int | None = Query(None, ge=0, le=1), p: Pagination = PAGINATION) -> dict[str, Any]:
    where = ["true"]
    params: dict[str, Any] = {}
    if connector:
        where.append("d.connector_name = :c")
        params["c"] = connector
    if status:
        where.append("d.status = :s")
        params["s"] = status
    if q:
        where.append("(d.url ilike :q or d.title ilike :q)")
        params["q"] = f"%{q}%"
    if entity:
        where.append("(d.entity_id = :e or e.slug = :e)")
        params["e"] = entity
    if needs_llm is not None:
        where.append("d.needs_llm = :nl")
        params["nl"] = bool(needs_llm)
    where_sql = " and ".join(where)
    async with connection() as conn:
        rows = await fetch_all(conn, f"""select d.*, e.slug as entity_slug, e.canonical_name as entity_name, e.entity_type,
                                                (select count(*) from snapshots x where x.document_id = d.id) as snapshots,
                                                (select x.processing_status from snapshots x where x.document_id = d.id order by x.observed_at desc limit 1) as last_processing_status
                                         from documents d left join entities e on e.id = d.entity_id where {where_sql}
                                         order by d.last_fetched_at desc nulls last, d.first_seen_at desc limit :lim offset :off""", lim=p.limit, off=p.offset, **params)
        total = await fetch_val(conn, f"select count(*) from documents d left join entities e on e.id = d.entity_id where {where_sql}", **params)
    return page([{**r, "snapshots": int(r["snapshots"])} for r in rows], int(total or 0), p)


@router.get("/documents/{doc_id}")
async def document(doc_id: str) -> dict[str, Any]:
    async with connection() as conn:
        doc = await fetch_one(conn, """select d.*, e.slug as entity_slug, e.canonical_name as entity_name, e.entity_type, s.name as source_name, s.tier as source_tier
                                       from documents d left join entities e on e.id = d.entity_id left join sources s on s.id = d.source_id where d.id = :id or d.url = :id limit 1""", id=doc_id)
        if not doc:
            raise ApiError(404, "document not found")
        snaps = await fetch_all(conn, """select id, run_id, url, final_url, observed_at, http_status, content_type, content_hash, byte_size, text_hash, parser_version, connector_version,
                                         transport, changed, processing_status, created_at, structured is not null as has_structured, diff is not null as has_diff, text_path is not null as has_text
                                         from snapshots where document_id = :id order by observed_at desc limit 200""", id=doc["id"])
        claims = await fetch_val(conn, "select count(*) from claims c join snapshots s on s.id = c.snapshot_id where s.document_id = :id", id=doc["id"])
    return {**doc, "snapshots": snaps, "claims_from_document": int(claims or 0)}


@router.get("/snapshots/{snap_id}")
async def snapshot(snap_id: str, text_limit: int = Query(SNAPSHOT_TEXT_LIMIT, ge=0, le=SNAPSHOT_TEXT_LIMIT)) -> dict[str, Any]:
    async with connection() as conn:
        snap = await fetch_one(conn, """select s.*, d.url as document_url, d.doc_type, d.connector_name, d.entity_id, e.slug as entity_slug, e.canonical_name as entity_name
                                        from snapshots s join documents d on d.id = s.document_id left join entities e on e.id = d.entity_id where s.id = :id""", id=snap_id)
        if not snap:
            raise ApiError(404, "snapshot not found")
        claims = await fetch_all(conn, """select c.id, c.entity_id, e.slug as entity_slug, c.property, c.value, c.unit, c.status, c.confidence, c.extractor, c.observed_at
                                          from claims c left join entities e on e.id = c.entity_id where c.snapshot_id = :id order by e.slug, c.property limit 500""", id=snap_id)
        events = await fetch_all(conn, "select id, entity_id, event_type, category, property, summary, importance, observed_at from change_events where snapshot_id = :id order by observed_at desc limit 200", id=snap_id)
        llm = await fetch_all(conn, "select id, task_type, stage, model, status, input_tokens, output_tokens, duration_ms, error, created_at from llm_jobs where snapshot_id = :id order by created_at desc limit 50", id=snap_id)
    text, text_error, text_len = None, None, 0
    if snap.get("text_path") and text_limit:
        try:
            full = archive.load_text(snap["text_path"])
            text_len = len(full)
            text = full[:text_limit]
        except OSError as exc:
            text_error = f"cleaned text unavailable ({exc.__class__.__name__})"
    out = _public(snap)
    out.update({"text": text, "text_chars": text_len, "text_truncated": text_len > text_limit, "text_error": text_error, "has_raw": bool(snap.get("raw_path")),
                "claims": claims, "events": events, "llm_jobs": llm})
    return out


# ------------------------------------------------------------------------------------------------------------------ queues


@router.get("/jobs")
async def jobs(status: str | None = None, kind: str | None = None, limit: int = Query(100, ge=1, le=1000), offset: int = Query(0, ge=0)) -> dict[str, Any]:
    where = ["true"]
    params: dict[str, Any] = {}
    if status:
        where.append("status = any(cast(:s as text[]))")
        params["s"] = [x.strip() for x in status.split(",") if x.strip()]
    if kind:
        where.append("kind = :k")
        params["k"] = kind
    where_sql = " and ".join(where)
    async with connection() as conn:
        rows = await fetch_all(conn, f"select * from jobs where {where_sql} order by case status when 'running' then 0 when 'queued' then 1 when 'failed' then 2 when 'dead' then 3 else 4 end, "
                                     f"priority, coalesce(finished_at, run_after) desc limit :lim offset :off", lim=limit, off=offset, **params)
        total = await fetch_val(conn, f"select count(*) from jobs where {where_sql}", **params)
        depth = await queue_depth(conn)
    return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset, "depth": depth}


@router.post("/jobs/{job_id}/retry")
async def retry_job(job_id: str) -> dict[str, Any]:
    async with transaction() as conn:
        row = await fetch_one(conn, """update jobs set status = 'queued', attempts = 0, error = null, locked_by = null, locked_at = null, finished_at = null, run_after = now()
                                       where id = :id and status in ('failed', 'dead', 'done') returning id, kind, status""", id=job_id)
    if not row:
        raise ApiError(404, "job not found or not retryable (must be failed, dead or done)")
    return {"ok": True, **row}


@router.post("/jobs/requeue-dead")
async def requeue_dead(kind: str | None = None) -> dict[str, Any]:
    where = "j.status = 'dead'" + (" and j.kind = :k" if kind else "")
    async with transaction() as conn:
        # respect the partial unique index on dedupe_key (queued|running): skip dead jobs already superseded by a live one,
        # and requeue only one dead job per dedupe_key (the most recent); the other duplicates are closed as superseded.
        n = await fetch_val(conn, f"""with pick as (
                select distinct on (coalesce(j.dedupe_key, j.id)) j.id from jobs j
                where {where} and (j.dedupe_key is null or not exists (select 1 from jobs q where q.dedupe_key = j.dedupe_key and q.status in ('queued','running')))
                order by coalesce(j.dedupe_key, j.id), j.created_at desc),
            u as (update jobs set status = 'queued', attempts = 0, error = null, locked_by = null, locked_at = null, finished_at = null, run_after = now()
                  where id in (select id from pick) returning 1)
            select count(*) from u""", k=kind)
        await execute(conn, f"update jobs j set status = 'done', error = coalesce(error, '') || ' [superseded]' where {where}", k=kind)
    return {"requeued": int(n or 0)}


@router.get("/llm-jobs")
async def llm_jobs(limit: int = Query(100, ge=1, le=1000), offset: int = Query(0, ge=0), status: str | None = None, task: str | None = None) -> dict[str, Any]:
    where = ["true"]
    params: dict[str, Any] = {}
    if status:
        where.append("status = :s")
        params["s"] = status
    if task:
        where.append("task_type = :t")
        params["t"] = task
    where_sql = " and ".join(where)
    async with connection() as conn:
        rows = await fetch_all(conn, f"select id, job_id, task_type, stage, engine, model, node, schema_name, snapshot_id, entity_id, input_tokens, output_tokens, duration_ms, status, error, created_at "
                                     f"from llm_jobs where {where_sql} order by created_at desc limit :lim offset :off", lim=limit, off=offset, **params)
        total = await fetch_val(conn, f"select count(*) from llm_jobs where {where_sql}", **params)
        totals = await fetch_all(conn, """select stage, model, status, count(*) as n, coalesce(sum(input_tokens), 0) as input_tokens, coalesce(sum(output_tokens), 0) as output_tokens,
                                          coalesce(avg(duration_ms), 0)::int as avg_ms from llm_jobs group by 1, 2, 3 order by 1, 2, 3""")
    return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset, "totals": [{**t, "n": int(t["n"]), "input_tokens": int(t["input_tokens"]), "output_tokens": int(t["output_tokens"])} for t in totals]}


@router.get("/llm/health")
async def llm_health() -> dict[str, Any]:
    h = await gateway.health()
    h["base_url_configured"] = bool(settings.llm_base_url)
    return h


class LLMEnqueue(BaseModel):
    limit: int = Field(200, ge=1, le=5000)
    task: str | None = None
    connector: str | None = None


@router.post("/llm/enqueue")
async def llm_enqueue(body: LLMEnqueue | None = None) -> dict[str, Any]:
    body = body or LLMEnqueue()
    where = "s.processing_status in ('stored', 'extracted', 'llm_pending') and s.text_path is not null and (d.needs_llm or s.processing_status = 'llm_pending')"
    params: dict[str, Any] = {"lim": body.limit}
    if body.connector:
        where += " and d.connector_name = :c"
        params["c"] = body.connector
    queued = 0
    async with transaction() as conn:
        snaps = await fetch_all(conn, f"select s.id from snapshots s join documents d on d.id = s.document_id where {where} order by d.priority, s.observed_at desc limit :lim", **params)
        for s in snaps:
            payload = {"snapshot_id": s["id"], **({"task": body.task} if body.task else {})}
            if await enqueue(conn, "llm_extract", payload, priority=6, dedupe_key=f"llm:{s['id']}"):
                queued += 1
        if snaps:
            await execute(conn, "update snapshots set processing_status = 'llm_pending' where id = any(cast(:ids as text[])) and processing_status <> 'llm_pending'", ids=[s["id"] for s in snaps])
    return {"queued": queued, "candidates": len(snaps), "llm_available": gateway.available}


class ReprocessBody(BaseModel):
    connector: str
    url: str | None = None


@router.post("/reprocess")
async def reprocess(body: ReprocessBody) -> dict[str, Any]:
    if body.connector not in connector_registry():
        raise ApiError(404, f"unknown connector {body.connector!r}")
    async with transaction() as conn:
        jid = await enqueue(conn, "reprocess_snapshot", {"connector": body.connector, **({"url": body.url} if body.url else {})}, priority=3,
                            dedupe_key=f"reprocess:{body.connector}:{body.url or '*'}")
    return {"queued": jid is not None, "job_id": jid}


# ------------------------------------------------------------------------------------------------------------------ review queue & curation


@router.get("/review")
async def review(status: str = "pending", kind: str | None = None, limit: int = Query(100, ge=1, le=1000), offset: int = Query(0, ge=0)) -> dict[str, Any]:
    where = ["r.status = :s"] if status != "all" else ["true"]
    params: dict[str, Any] = {"s": status}
    if kind:
        where.append("r.kind = :k")
        params["k"] = kind
    where_sql = " and ".join(where)
    async with connection() as conn:
        rows = await fetch_all(conn, f"""select r.*, (select jsonb_agg(jsonb_build_object('id', e.id, 'slug', e.slug, 'name', e.canonical_name, 'entity_type', e.entity_type, 'status', e.status) order by e.slug)
                                                       from entities e where e.id = any(r.entity_ids)) as entities
                                         from review_queue r where {where_sql} order by r.created_at desc limit :lim offset :off""", lim=limit, off=offset, **params)
        total = await fetch_val(conn, f"select count(*) from review_queue r where {where_sql}", **params)
        kinds = await fetch_all(conn, "select kind, status, count(*) as n from review_queue group by 1, 2 order by 1, 2")
    return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset, "by_kind": [{**k, "n": int(k["n"])} for k in kinds]}


class ReviewAction(BaseModel):
    action: str = Field(..., pattern="^(approve|reject|edit)$")
    resolution: dict[str, Any] | None = None


@router.post("/review/{review_id}")
async def review_action(review_id: str, body: ReviewAction) -> dict[str, Any]:
    async with transaction() as conn:
        item = await fetch_one(conn, "select * from review_queue where id = :id for update", id=review_id)
        if not item:
            raise ApiError(404, "review item not found")
        if item["status"] != "pending":
            raise ApiError(409, f"review item already {item['status']}")
        resolution: dict[str, Any] = dict(body.resolution or {})
        effect: dict[str, Any] | None = None
        if body.action == "approve" and item["kind"] == "merge_candidate":
            payload = item["payload"] or {}
            ids = list(item["entity_ids"] or [])
            source = resolution.get("source_id") or payload.get("source_id") or (ids[0] if len(ids) > 1 else None)
            target = resolution.get("target_id") or payload.get("target_id") or (ids[1] if len(ids) > 1 else None)
            if not source or not target:
                raise ApiError(400, "merge_candidate needs source_id and target_id (payload or resolution)")
            try:
                effect = await merge_entities(conn, source, target)
            except (ValueError, LookupError) as exc:
                raise ApiError(400, str(exc)) from exc
            resolution.update({"merged": effect})
        elif body.action == "approve" and item["kind"] == "conflict":
            keep = resolution.get("keep_claim_id")
            if keep:
                effect = await _accept_claim(conn, keep)
                resolution["accepted"] = effect
        status = {"approve": "approved", "reject": "rejected", "edit": "edited"}[body.action]
        await execute(conn, "update review_queue set status = :st, resolution = cast(:r as jsonb), resolved_at = now() where id = :id", st=status, r=jsonb(resolution), id=review_id)
    await cache.cache_invalidate()
    return {"ok": True, "id": review_id, "status": status, "effect": effect}


async def _accept_claim(conn: Any, claim_id: str) -> dict[str, Any]:
    """Conflict resolution: promote one claim to current, supersede the others for the same property."""
    c = await fetch_one(conn, "select * from claims where id = :id", id=claim_id)
    if not c:
        raise ApiError(404, "claim not found")
    await execute(conn, "update claims set status = 'superseded', valid_to = now() where entity_id = :e and property = :p and id <> :id and status in ('current', 'conflicting')",
                  e=c["entity_id"], p=c["property"], id=claim_id)
    await execute(conn, "update claims set status = 'current', confidence = case when confidence = 'conflicted' then 'high' else confidence end, valid_to = null where id = :id", id=claim_id)
    prov = {"source_id": c["source_id"], "snapshot_id": c["snapshot_id"], "url": c["source_url"], "tier": c["tier"], "confidence": "high", "extractor": c["extractor"],
            "observed_at": c["observed_at"].isoformat() if c["observed_at"] else None, "resolved_by": "review"}
    await execute(conn, "update entities set attributes = attributes || jsonb_build_object(:p, cast(:v as jsonb)), provenance = provenance || jsonb_build_object(:p, cast(:pv as jsonb)), updated_at = now() where id = :e",
                  p=c["property"], v=jsonb(c["value"]), pv=jsonb(prov), e=c["entity_id"])
    return {"claim_id": claim_id, "entity_id": c["entity_id"], "property": c["property"]}


@router.get("/entities/duplicates")
async def duplicates(type: str | None = Query(None, alias="type"), limit: int = Query(100, ge=1, le=500), threshold: float = Query(0.8, ge=0.3, le=1.0)) -> dict[str, Any]:
    where = "a.entity_type = :t" if type else "true"
    async with connection() as conn:
        # `%` uses the pg_trgm GIN index at the session default threshold (0.3, a superset of any threshold ≥ 0.3); the exact bound is applied
        # with similarity() > :th — no set_limit(), which used to leak a GUC change into the pooled connection.
        rows = await fetch_all(conn, f"""
            select a.id as a_id, a.slug as a_slug, a.canonical_name as a_name, a.entity_type, a.first_seen_at as a_first_seen_at, oa.canonical_name as a_org,
                   b.id as b_id, b.slug as b_slug, b.canonical_name as b_name, b.first_seen_at as b_first_seen_at, ob.canonical_name as b_org,
                   similarity(a.canonical_name, b.canonical_name) as similarity,
                   (select count(*) from claims c where c.entity_id = a.id and c.status = 'current') as a_claims,
                   (select count(*) from claims c where c.entity_id = b.id and c.status = 'current') as b_claims
            from entities a join entities b on b.entity_type = a.entity_type and b.id > a.id and a.canonical_name % b.canonical_name
            left join entities oa on oa.id = a.organization_id left join entities ob on ob.id = b.organization_id
            where {where} and a.merged_into is null and b.merged_into is null and similarity(a.canonical_name, b.canonical_name) > :th
            order by similarity desc, a.canonical_name limit :lim""", t=type, th=threshold, lim=limit)
    return {"threshold": threshold, "items": [{"entity_type": r["entity_type"], "similarity": round(float(r["similarity"]), 3),
                                               "a": {"id": r["a_id"], "slug": r["a_slug"], "name": r["a_name"], "organization": r["a_org"], "first_seen_at": r["a_first_seen_at"], "claims": int(r["a_claims"])},
                                               "b": {"id": r["b_id"], "slug": r["b_slug"], "name": r["b_name"], "organization": r["b_org"], "first_seen_at": r["b_first_seen_at"], "claims": int(r["b_claims"])}}
                                              for r in rows]}


class MergeBody(BaseModel):
    source_id: str
    target_id: str


@router.post("/entities/merge")
async def merge(body: MergeBody) -> dict[str, Any]:
    async with transaction() as conn:
        src = await fetch_one(conn, "select id from entities where id = :k or slug = :k limit 1", k=body.source_id)
        dst = await fetch_one(conn, "select id from entities where id = :k or slug = :k limit 1", k=body.target_id)
        if not src or not dst:
            raise ApiError(404, "source or target entity not found")
        try:
            result = await merge_entities(conn, src["id"], dst["id"])
        except (ValueError, LookupError) as exc:
            raise ApiError(400, str(exc)) from exc
    await cache.cache_invalidate()
    return result


@router.post("/entities/{entity_id}/claims/{claim_id}/retract")
async def retract_claim(entity_id: str, claim_id: str) -> dict[str, Any]:
    async with transaction() as conn:
        ent = await fetch_one(conn, "select id from entities where id = :k or slug = :k limit 1", k=entity_id)
        if not ent:
            raise ApiError(404, "entity not found")
        c = await fetch_one(conn, "select * from claims where id = :c and entity_id = :e", c=claim_id, e=ent["id"])
        if not c:
            raise ApiError(404, "claim not found for this entity")
        was_current = c["status"] == "current"
        await execute(conn, "update claims set status = 'retracted', valid_to = coalesce(valid_to, now()) where id = :c", c=claim_id)
        restored = None
        if was_current:
            prev = await fetch_one(conn, """select * from claims where entity_id = :e and property = :p and id <> :c and status in ('superseded', 'conflicting')
                                            order by tier, valid_from desc limit 1""", e=ent["id"], p=c["property"], c=claim_id)
            if prev:
                await execute(conn, "update claims set status = 'current', valid_to = null where id = :id", id=prev["id"])
                prov = {"source_id": prev["source_id"], "snapshot_id": prev["snapshot_id"], "url": prev["source_url"], "tier": prev["tier"], "confidence": prev["confidence"],
                        "extractor": prev["extractor"], "observed_at": prev["observed_at"].isoformat() if prev["observed_at"] else None, "restored_by": "retraction"}
                await execute(conn, "update entities set attributes = attributes || jsonb_build_object(:p, cast(:v as jsonb)), provenance = provenance || jsonb_build_object(:p, cast(:pv as jsonb)), updated_at = now() where id = :e",
                              p=c["property"], v=jsonb(prev["value"]), pv=jsonb(prov), e=ent["id"])
                restored = prev["id"]
            else:
                await execute(conn, "update entities set attributes = attributes - :p, provenance = provenance - :p, updated_at = now() where id = :e", p=c["property"], e=ent["id"])
        await execute(conn, """insert into change_events (id, entity_id, event_type, category, property, old_value, summary, importance, connector_name, dedupe_key)
                               values (:id, :e, 'CLAIM_RETRACTED', 'source', :p, cast(:v as jsonb), :s, 1, 'curation', :dk) on conflict (dedupe_key) do nothing""",
                      id=new_id("change_event"), e=ent["id"], p=c["property"], v=jsonb(c["value"]),
                      s=f"Retracted claim {c['property']}", dk=f"retract:{claim_id}")
    await cache.cache_invalidate()
    return {"ok": True, "claim_id": claim_id, "property": c["property"], "was_current": was_current, "restored_claim_id": restored}


# ------------------------------------------------------------------------------------------------------------------ infrastructure & maintenance


@router.get("/infrastructure")
async def infrastructure() -> dict[str, Any]:
    async with connection() as conn:
        db = await fetch_one(conn, """select pg_database_size(current_database()) as db_bytes, current_database() as database, version() as pg_version,
                                      (select count(*) from pg_stat_activity where datname = current_database()) as connections,
                                      (select setting from pg_settings where name = 'max_connections') as max_connections""")
        tables = await fetch_all(conn, """select c.relname as table, pg_total_relation_size(c.oid) as total_bytes, pg_relation_size(c.oid) as data_bytes,
                                          coalesce(s.n_live_tup, 0) as rows_estimate from pg_class c join pg_namespace n on n.oid = c.relnamespace
                                          left join pg_stat_user_tables s on s.relid = c.oid where n.nspname = 'public' and c.relkind = 'r'
                                          order by pg_total_relation_size(c.oid) desc limit 20""")
        extensions = await fetch_all(conn, "select extname, extversion from pg_extension order by 1")
    redis_info: dict[str, Any] = {"ok": False}
    try:
        info = await cache.redis().info("memory")
        redis_info = {"ok": True, "used_memory": info.get("used_memory"), "used_memory_human": info.get("used_memory_human"),
                      "api_cache_keys": sum([1 async for _ in cache.redis().scan_iter(match="aia:api:*", count=1000)])}
    except Exception as exc:  # noqa: BLE001
        redis_info = {"ok": False, "error": exc.__class__.__name__}
    return {"hostname": socket.gethostname(), "python": platform.python_version(), "platform": platform.platform(), "api_uptime_s": int(time.time() - _STARTED),
            "env": settings.app_env, "heartbeats": await cache.heartbeats(), "archive": archive.archive_size(), "data_dir_exists": settings.data_dir.exists(),
            "database": {**(db or {}), "db_bytes": int((db or {}).get("db_bytes") or 0), "tables": tables, "extensions": extensions}, "redis": redis_info,
            "llm": {"available": gateway.available, "engine": gateway.engine.name}, "scheduler_tick_s": settings.scheduler_tick_s}


CACHE_PREFIXES_AFTER_RUN = ("/api/v1/stats", "/api/v1/changes", "/api/v1/benchmarks", "/api/v1/prices", "/api/v1/models", "/api/v1/deployments", "/api/v1/frontier",
                            "/api/v1/pulse", "/api/v1/open", "/api/v1/families", "/api/v1/timeline", "/api/v1/diff", "facets:", "frontier:")


@router.post("/cache/flush")
async def cache_flush(prefix: str = "", after_run: int = Query(0, ge=0, le=1)) -> dict[str, Any]:
    """Flush `aia:api:<prefix>*`. `after_run=1` flushes the read paths a connector run invalidates (the scheduler calls `services.cache.cache_invalidate`
    with these prefixes after every run — see docs/API.md)."""
    if after_run:
        flushed = {p: await cache.cache_invalidate(p) for p in CACHE_PREFIXES_AFTER_RUN}
        return {"flushed": sum(flushed.values()), "by_prefix": flushed}
    return {"flushed": await cache.cache_invalidate(prefix), "prefix": prefix or "*"}


@router.post("/stats/recompute")
async def stats_recompute() -> dict[str, Any]:
    from aiatlas.services.stats import compute_stats

    counts = await compute_stats()
    await cache.cache_invalidate("/api/v1/stats")
    return {"ok": True, "entities_total": counts.get("entities_total")}


class QualityBody(BaseModel):
    entity_ids: list[str] | None = None
    limit: int = Field(20000, ge=1, le=200000)


@router.post("/quality/recompute")
async def quality_recompute(body: QualityBody | None = None) -> dict[str, Any]:
    from aiatlas.services.quality import recompute

    body = body or QualityBody()
    res = await recompute(entity_ids=body.entity_ids, limit=body.limit)
    await cache.cache_invalidate()
    return {"ok": True, **res}