"""SLO evaluation — runtime check of each handler's SLO commitment.

Computed on demand (admin opens /admin/health SLO panel, or
/metrics scrape) — no background cron, no persisted breach rows.
The signals are derived live from ``scheduled_job_runs`` so a
freshly-changed SLO target reflects immediately.

A handler's SLO is in one of three states:

  * ``ok``     — every metric passes the target
  * ``warn``   — within 10% of the floor / 1 below max_consecutive
                 (not yet breaching but trending)
  * ``breach`` — one or more metrics fail the target

The detail dict carries actual vs target per axis so the panel can
show ``99.3% (target 99%)`` instead of a bare ``ok`` tag.
"""
from __future__ import annotations

from datetime import UTC, datetime, timedelta
from typing import Any, Literal

from sqlalchemy import desc, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.scheduled_jobs.models import ScheduledJob, ScheduledJobRun
from app.scheduled_jobs.registry import list_handlers
from app.scheduled_jobs.slo import SLOTarget

SLOStatus = Literal["ok", "warn", "breach", "no_data"]

# Trending-warn ratio — within 10% of the floor counts as warn.
_WARN_HEADROOM_RATIO = 0.10


def _percentile(values: list[int], pct: float) -> int | None:
    """Cheap percentile (linear interpolation). pct in [0, 100]."""
    if not values:
        return None
    s = sorted(values)
    if len(s) == 1:
        return s[0]
    k = (len(s) - 1) * (pct / 100.0)
    f = int(k)
    c = min(f + 1, len(s) - 1)
    if f == c:
        return s[f]
    return int(s[f] + (s[c] - s[f]) * (k - f))


async def evaluate_handler_slo(
    db: AsyncSession,
    *,
    handler_key: str,
    target: SLOTarget,
    window_days: int = 7,
) -> dict[str, Any]:
    """Evaluate ``target`` against actual runs in the last ``window_days``.

    Returns:
        ``{
            handler_key, tier, status,
            window_days,
            actual: {success_rate, p95_latency_ms, consecutive_failures, runs_count},
            target: {success_rate_floor, latency_p95_ms_ceiling, max_consecutive_failures},
            breaches: list[str],
        }``
    """
    now = datetime.now(UTC)
    window_start = now - timedelta(days=window_days)

    # Locate the job rows whose handler_key matches. Multiple
    # job-row instances can share one handler_key (e.g. per-provider
    # forks of cmdb.cost_facts_sync) — we aggregate across all of
    # them because a single failing instance must move the SLO needle
    # for that handler.
    job_ids_q = select(ScheduledJob.id).where(ScheduledJob.handler_key == handler_key)
    job_ids = [row[0] for row in (await db.execute(job_ids_q)).all()]
    if not job_ids:
        return {
            "handler_key": handler_key,
            "tier": target.tier,
            "status": "no_data",
            "window_days": window_days,
            "actual": {
                "success_rate": None,
                "p95_latency_ms": None,
                "consecutive_failures": 0,
                "runs_count": 0,
            },
            "target": _target_dict(target),
            "breaches": [],
        }

    # Status breakdown.
    status_rows = (
        await db.execute(
            select(ScheduledJobRun.status, func.count())
            .where(
                ScheduledJobRun.job_id.in_(job_ids),
                ScheduledJobRun.started_at >= window_start,
            )
            .group_by(ScheduledJobRun.status)
        )
    ).all()
    by_status = {s: int(n) for s, n in status_rows}
    # Round-62 P1 fix: denominator counts TERMINAL statuses only.
    # ``running`` is in-flight and not a completed attempt; counting
    # it inflates the denominator and depresses success_rate to a
    # spurious breach whenever a long-running cron is mid-flight
    # (1 ok + 1 running → 0.5 instead of 1.0). ``skipped`` is a
    # single-flight collision (worker tick collided with prior run
    # still holding the lock), not a real attempt — also excluded.
    # Mirrors the metrics.py exclusion of status='running' from the
    # _runs_total counter (PR-C Round-62 fix).
    ok_count = by_status.get("ok", 0)
    failed_count = by_status.get("failed", 0)
    runs_count = ok_count + failed_count
    success_rate = (ok_count / runs_count) if runs_count else None

    # p95 latency over successful runs in window — including failures
    # would mix "ran but slow" with "never finished" and skew the
    # ceiling check.
    durations = (
        await db.execute(
            select(ScheduledJobRun.duration_ms).where(
                ScheduledJobRun.job_id.in_(job_ids),
                ScheduledJobRun.started_at >= window_start,
                ScheduledJobRun.duration_ms.isnot(None),
                ScheduledJobRun.status == "ok",
            )
        )
    ).all()
    duration_values = [int(r[0]) for r in durations]
    p95_latency = _percentile(duration_values, 95.0)

    # Consecutive failures from the tail — most recent runs across
    # ALL job_ids for this handler within the SLO window, ordered by
    # start time desc.
    #
    # Without ``started_at >= window_start`` a handler that hasn't
    # run for months would surface its ancient failure tail forever
    # (the next ``runs_count==0 → no_data`` short-circuit never fires
    # if any history exists) — perma-breach noise that drowns real
    # signal. Window-bound the tail so "is this handler currently in
    # trouble?" stays a current-window read.
    #
    # Round-62 P2: filter to TERMINAL statuses (ok / failed) so a
    # ``running`` row at the top of the tail (multi-instance handler
    # with concurrent per-provider jobs) doesn't short-circuit the
    # count to 0 and mask a real failure burst on a different instance.
    recent_statuses = (
        await db.execute(
            select(ScheduledJobRun.status)
            .where(
                ScheduledJobRun.job_id.in_(job_ids),
                ScheduledJobRun.started_at >= window_start,
                ScheduledJobRun.status.in_(["ok", "failed"]),
            )
            .order_by(desc(ScheduledJobRun.started_at))
            .limit(target.max_consecutive_failures + 1)
        )
    ).all()
    consecutive_failures = 0
    for (status,) in recent_statuses:
        if status == "failed":
            consecutive_failures += 1
        else:
            break

    # Evaluate against target.
    breaches: list[str] = []
    warns: list[str] = []
    if runs_count > 0 and success_rate is not None:
        if success_rate < target.success_rate_floor:
            breaches.append(
                f"success_rate {success_rate:.3f} < floor "
                f"{target.success_rate_floor:.3f}"
            )
        elif success_rate < target.success_rate_floor + _WARN_HEADROOM_RATIO * (
            1.0 - target.success_rate_floor
        ):
            warns.append("success_rate trending toward floor")
    if (
        target.latency_p95_ms_ceiling is not None
        and p95_latency is not None
        and p95_latency > target.latency_p95_ms_ceiling
    ):
        breaches.append(
            f"p95 latency {p95_latency} ms > ceiling "
            f"{target.latency_p95_ms_ceiling} ms"
        )
    if consecutive_failures >= target.max_consecutive_failures:
        breaches.append(
            f"consecutive failures {consecutive_failures} >= "
            f"limit {target.max_consecutive_failures}"
        )
    elif (
        target.max_consecutive_failures > 1
        and consecutive_failures == target.max_consecutive_failures - 1
    ):
        # ``> 1`` guard: the "one away from limit" warn is only meaningful
        # when the breach limit is at least 2. ``SLOTarget`` permits
        # ``max_consecutive_failures == 1`` (``__post_init__`` only enforces
        # ``>= 1``); without this guard a limit of 1 would make the predicate
        # ``consecutive_failures == 0`` and raise a spurious WARN on a handler
        # with ZERO recent failures. No shipped target uses 1 today, but a
        # future owner setting it would otherwise trip a false alert.
        warns.append(
            f"consecutive failures {consecutive_failures} (one away from limit)"
        )

    if runs_count == 0:
        # job row exists but no runs yet in window — treat the same
        # as "no job row" for status purposes (the panel should not
        # claim "ok" without evidence). Breach checks above either
        # short-circuited on the None success_rate or evaluated the
        # consecutive-failures tail (also empty), so this branch is a
        # pure relabel from the default ``ok``.
        status: SLOStatus = "no_data"
    elif breaches:
        status = "breach"
    elif warns:
        status = "warn"
    else:
        status = "ok"

    return {
        "handler_key": handler_key,
        "tier": target.tier,
        "status": status,
        "window_days": window_days,
        "actual": {
            "success_rate": success_rate,
            "p95_latency_ms": p95_latency,
            "consecutive_failures": consecutive_failures,
            "runs_count": runs_count,
        },
        "target": _target_dict(target),
        "breaches": breaches + warns,
    }


def _target_dict(t: SLOTarget) -> dict[str, Any]:
    return {
        "success_rate_floor": t.success_rate_floor,
        "latency_p95_ms_ceiling": t.latency_p95_ms_ceiling,
        "max_consecutive_failures": t.max_consecutive_failures,
        "tier": t.tier,
    }


async def evaluate_all_slos(
    db: AsyncSession,
    *,
    window_days: int = 7,
) -> list[dict[str, Any]]:
    """Evaluate SLOs for every registered handler.

    Returned list is sorted: P0 / breach first, P2 / ok last. The
    admin UI renders the top N to focus attention.
    """
    out: list[dict[str, Any]] = []
    for h in list_handlers():
        out.append(
            await evaluate_handler_slo(
                db,
                handler_key=h.key,
                target=h.slo,
                window_days=window_days,
            )
        )

    tier_rank = {"P0": 0, "P1": 1, "P2": 2}
    status_rank = {"breach": 0, "warn": 1, "no_data": 2, "ok": 3}
    out.sort(
        key=lambda r: (
            tier_rank.get(r["tier"], 9),
            status_rank.get(r["status"], 9),
            r["handler_key"],
        )
    )
    return out