"""Prometheus exposition for the platform.

Hand-rolled text/plain renderer because Helix runs a single FastAPI
process per container, so the dependency surface (`prometheus_client`
multiprocess machinery) is not worth pulling in. The exposition
format is stable + simple — header, type, sample lines, blank line
between families — and easy to test against.

Metric families exposed:

  * ``helix_scheduled_job_runs_total{handler_key, status}``  Counter
        Lifetime ok / failed counts per handler, sourced from
        monotonic ``ScheduledJob.run_count`` (ok = run_count -
        failure_count) and ``ScheduledJob.failure_count``. Round-62
        rewrite — the previous COUNT(*) GROUP BY over
        ``scheduled_job_runs`` violated Prometheus counter semantics
        because that table is retention-pruned. ``skipped`` runs are
        single-flight collisions, not real attempts, and are excluded;
        ``running`` runs are exposed separately as
        ``helix_scheduled_job_runs_active``.
  * ``helix_scheduled_job_runs_active{handler_key}``         Gauge
        Currently-running runs per handler (status=running). The
        transient counterpart to the ``_runs_total`` counter.
  * ``helix_scheduled_job_runs_interrupted_24h{handler_key}`` Gauge
        Runs cut short by a worker lifecycle event (deploy drain or
        hard kill) in the last 24h. Deliberately kept out of
        ``_runs_total``: an interrupted run is not a handler outcome.
  * ``helix_scheduled_job_last_status{handler_key}``         Gauge
        1=ok, 0=failed, -1=never_run / running / skipped.
  * ``helix_scheduled_job_failure_count{handler_key}``       Gauge
        Lifetime failure count (mirror of ``ScheduledJob.failure_count``).
  * ``helix_scheduled_job_unacked_failures{handler_key}``    Gauge
        ``failed`` runs with ``acknowledged_at IS NULL``.
  * ``helix_scheduled_job_slo_status{handler_key, tier}``    Gauge
        1.0=ok, 0.5=warn, 0.0=breach, NaN=no_data.
  * ``helix_offboarding_in_flight{sla}``                     Gauge
        In-flight processes bucketed by sla_breach_level.
  * ``helix_notification_outbox_by_status{status}``          Gauge
  * ``helix_compliance_incidents_open``                      Gauge
  * ``helix_under_licensed_plans``                           Gauge

The endpoint is **pull-based**; values are computed in the request
path. Helix has roughly 50 handlers and a few hundred-thousand
``scheduled_job_runs`` rows, so the cost is bounded — but the
endpoint should not be scraped sub-second.
"""
from __future__ import annotations

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

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

from app.compliance.models import INCIDENT_OPEN, ComplianceIncident
from app.domains.asset.models import (
    SoftwareLicensePlan,
    SoftwareSeatSyncStat,
)
from app.domains.offboarding.models import (
    PROC_CANCELLED,
    PROC_COMPLETED,
    OffboardingProcess,
)
from app.notifications.models import NotificationOutbox
from app.scheduled_jobs.models import (
    RUN_STATUS_INTERRUPTED,
    ScheduledJob,
    ScheduledJobRun,
)
from app.scheduled_jobs.registry import list_handlers
from app.scheduled_jobs.slo_evaluator import evaluate_all_slos


def _escape_label_value(v: str) -> str:
    """Escape a Prometheus label value per the exposition format."""
    return v.replace("\\", "\\\\").replace("\n", "\\n").replace('"', '\\"')


def _fmt_labels(labels: dict[str, str]) -> str:
    if not labels:
        return ""
    parts = [f'{k}="{_escape_label_value(v)}"' for k, v in labels.items()]
    return "{" + ",".join(parts) + "}"


def _line(name: str, labels: dict[str, str], value: float) -> str:
    if math.isnan(value):
        rendered = "NaN"
    elif math.isinf(value):
        rendered = "+Inf" if value > 0 else "-Inf"
    else:
        rendered = f"{value:g}"
    return f"{name}{_fmt_labels(labels)} {rendered}"


def _slo_status_value(status: str) -> float:
    if status == "ok":
        return 1.0
    if status == "warn":
        return 0.5
    if status == "breach":
        return 0.0
    return float("nan")  # no_data


_LAST_STATUS_VALUE = {
    "ok": 1.0,
    "failed": 0.0,
    "running": -1.0,
    "skipped": -1.0,
    None: -1.0,
    "never_run": -1.0,
}


async def render_metrics(db: AsyncSession) -> str:
    """Compute + render the Prometheus exposition for one scrape."""
    families: list[str] = []

    # ── Scheduled jobs ──────────────────────────────────────────
    # ``helix_scheduled_job_runs_total`` is sourced from monotonic
    # ScheduledJob columns (``run_count`` for the all-attempts total,
    # ``failure_count`` for the failed total). The previous COUNT(*)
    # GROUP BY over scheduled_job_runs violated Prometheus counter
    # semantics in TWO ways:
    #   (1) Round-62 P0/P1: ``status='running'`` rows get UPDATEd in
    #       place to a terminal status, making the running count
    #       decrement.
    #   (2) Round-62 P1: ``scheduled_job_runs`` is retention-pruned to
    #       MAX_RUN_HISTORY_PER_JOB=100 by ``_trim_runs`` (oldest first,
    #       ordered by started_at regardless of status). Once a job
    #       passes 100 runs, the ok-row count actively DECREASES every
    #       time a new ok row pushes an old ok row over the retention
    #       edge — failures stay fresh and survive longer, so the
    #       counter drifts in a way that breaks ``rate()`` exactly on
    #       the busiest handlers (cmdb sync, leaver scan).
    #
    # ScheduledJob.run_count + failure_count are incremented in the
    # post-handler write-back path of the runner (service.py around
    # ``_write_run_outcome``) within the same tx as the run row
    # itself, so they're durable and never decremented — the right
    # source for a Prometheus counter.
    #
    # The transient ``running`` count moves to a sibling gauge
    # (``helix_scheduled_job_runs_active``) below.
    job_counts = (
        await db.execute(
            select(
                ScheduledJob.handler_key,
                func.coalesce(func.sum(ScheduledJob.run_count), 0),
                func.coalesce(func.sum(ScheduledJob.failure_count), 0),
            )
            .group_by(ScheduledJob.handler_key)
        )
    ).all()
    families.append(
        "# HELP helix_scheduled_job_runs_total Lifetime cron attempts by "
        "handler and terminal status (sourced from monotonic ScheduledJob.run_count "
        "/ failure_count — retention-pruning of scheduled_job_runs cannot decrement)."
    )
    families.append("# TYPE helix_scheduled_job_runs_total counter")
    for hk, all_count, fail_count in job_counts:
        total = int(all_count or 0)
        failed = int(fail_count or 0)
        ok = max(total - failed, 0)
        families.append(
            _line(
                "helix_scheduled_job_runs_total",
                {"handler_key": hk, "status": "ok"},
                ok,
            )
        )
        families.append(
            _line(
                "helix_scheduled_job_runs_total",
                {"handler_key": hk, "status": "failed"},
                failed,
            )
        )
    families.append("")

    # Transient running rows as a gauge — ok to fluctuate.
    running_rows = (
        await db.execute(
            select(ScheduledJob.handler_key, func.count())
            .join(ScheduledJobRun, ScheduledJobRun.job_id == ScheduledJob.id)
            .where(ScheduledJobRun.status == "running")
            .group_by(ScheduledJob.handler_key)
        )
    ).all()
    running_by_key = {hk: int(n) for hk, n in running_rows}
    families.append("# HELP helix_scheduled_job_runs_active Currently-running runs per handler (status=running).")
    families.append("# TYPE helix_scheduled_job_runs_active gauge")
    for h in list_handlers():
        families.append(
            _line(
                "helix_scheduled_job_runs_active",
                {"handler_key": h.key},
                running_by_key.get(h.key, 0),
            )
        )
    families.append("")

    # Interrupted runs in the last 24 h — the worker-lifecycle signal.
    #
    # An interrupted run is not a handler failure, so it is excluded from
    # ``_runs_total`` / ``failure_count`` / the SLO denominators on
    # purpose. That exclusion would make a worker which dies every night
    # completely invisible in metrics, which is why this gauge exists: a
    # deploy-day spike is expected and self-explaining, a steady non-zero
    # floor on a quiet day means the worker is being killed for a reason
    # nobody has looked at (OOM, an overrun stop grace). Windowed rather
    # than lifetime because the run ledger is retention-pruned and a
    # monotonic counter sourced from it would violate counter semantics —
    # the exact bug the Round-62 rewrite of ``_runs_total`` fixed.
    interrupted_since = datetime.now(UTC) - timedelta(hours=24)
    interrupted_rows = (
        await db.execute(
            select(ScheduledJob.handler_key, func.count())
            .join(ScheduledJobRun, ScheduledJobRun.job_id == ScheduledJob.id)
            .where(
                ScheduledJobRun.status == RUN_STATUS_INTERRUPTED,
                ScheduledJobRun.started_at >= interrupted_since,
            )
            .group_by(ScheduledJob.handler_key)
        )
    ).all()
    interrupted_by_key = {hk: int(n) for hk, n in interrupted_rows}
    families.append(
        "# HELP helix_scheduled_job_runs_interrupted_24h Runs cut short by a "
        "worker lifecycle event (deploy drain / hard kill) in the last 24h."
    )
    families.append("# TYPE helix_scheduled_job_runs_interrupted_24h gauge")
    for h in list_handlers():
        families.append(
            _line(
                "helix_scheduled_job_runs_interrupted_24h",
                {"handler_key": h.key},
                interrupted_by_key.get(h.key, 0),
            )
        )
    families.append("")

    # last_status + failure_count + unacked failures per handler_key.
    job_rows = (
        await db.execute(
            select(
                ScheduledJob.handler_key,
                ScheduledJob.last_status,
                ScheduledJob.failure_count,
            )
        )
    ).all()
    last_status_by_key: dict[str, str] = {}
    fail_count_by_key: dict[str, int] = {}
    for hk, last_status, fail_count in job_rows:
        # Multi-instance handlers (per-provider): pick worst signal.
        prior = last_status_by_key.get(hk)
        if last_status == "failed":
            last_status_by_key[hk] = "failed"
        elif prior != "failed" and last_status:
            last_status_by_key[hk] = last_status
        elif hk not in last_status_by_key:
            last_status_by_key[hk] = last_status or "never_run"
        fail_count_by_key[hk] = fail_count_by_key.get(hk, 0) + int(fail_count or 0)

    families.append("# HELP helix_scheduled_job_last_status Last run status per handler. 1=ok 0=failed -1=skipped/running/never_run.")
    families.append("# TYPE helix_scheduled_job_last_status gauge")
    for hk, status in last_status_by_key.items():
        families.append(
            _line(
                "helix_scheduled_job_last_status",
                {"handler_key": hk},
                _LAST_STATUS_VALUE.get(status, -1.0),
            )
        )
    families.append("")

    families.append("# HELP helix_scheduled_job_failure_count Lifetime failure count per handler.")
    families.append("# TYPE helix_scheduled_job_failure_count gauge")
    for hk, n in fail_count_by_key.items():
        families.append(
            _line("helix_scheduled_job_failure_count", {"handler_key": hk}, n)
        )
    families.append("")

    unacked_rows = (
        await db.execute(
            select(ScheduledJob.handler_key, func.count())
            .join(ScheduledJobRun, ScheduledJobRun.job_id == ScheduledJob.id)
            .where(
                ScheduledJobRun.status == "failed",
                ScheduledJobRun.acknowledged_at.is_(None),
            )
            .group_by(ScheduledJob.handler_key)
        )
    ).all()
    unacked_by_key = {hk: int(n) for hk, n in unacked_rows}
    families.append("# HELP helix_scheduled_job_unacked_failures Failed runs without acknowledged_at.")
    families.append("# TYPE helix_scheduled_job_unacked_failures gauge")
    # Emit a sample per registered handler so absence vs zero is
    # explicit (consumers can alert on missing series).
    for h in list_handlers():
        families.append(
            _line(
                "helix_scheduled_job_unacked_failures",
                {"handler_key": h.key},
                unacked_by_key.get(h.key, 0),
            )
        )
    families.append("")

    # ── SLO status per handler ──────────────────────────────────
    slos = await evaluate_all_slos(db, window_days=7)
    families.append("# HELP helix_scheduled_job_slo_status SLO health: 1=ok 0.5=warn 0=breach NaN=no_data.")
    families.append("# TYPE helix_scheduled_job_slo_status gauge")
    for row in slos:
        families.append(
            _line(
                "helix_scheduled_job_slo_status",
                {"handler_key": row["handler_key"], "tier": row["tier"]},
                _slo_status_value(row["status"]),
            )
        )
    families.append("")

    # ── Offboarding in-flight by SLA ─────────────────────────────
    in_flight_rows = (
        await db.execute(
            select(OffboardingProcess.sla_breach_level, func.count())
            .where(OffboardingProcess.status.notin_([PROC_COMPLETED, PROC_CANCELLED]))
            .group_by(OffboardingProcess.sla_breach_level)
        )
    ).all()
    by_sla = {sla: int(n) for sla, n in in_flight_rows}
    families.append("# HELP helix_offboarding_in_flight In-flight offboarding processes by SLA tier.")
    families.append("# TYPE helix_offboarding_in_flight gauge")
    for sla in ("none", "warn", "urgent", "breach"):
        families.append(
            _line(
                "helix_offboarding_in_flight",
                {"sla": sla},
                by_sla.get(sla, 0),
            )
        )
    families.append("")

    # ── Notification outbox by status ────────────────────────────
    outbox_rows = (
        await db.execute(
            select(NotificationOutbox.status, func.count())
            .group_by(NotificationOutbox.status)
        )
    ).all()
    families.append("# HELP helix_notification_outbox_by_status Outbox row count grouped by status.")
    families.append("# TYPE helix_notification_outbox_by_status gauge")
    for status, n in outbox_rows:
        families.append(
            _line(
                "helix_notification_outbox_by_status",
                {"status": status or "unknown"},
                int(n),
            )
        )
    families.append("")

    # ── Data health ─────────────────────────────────────────────
    open_incidents = int(
        (
            await db.execute(
                select(func.count(ComplianceIncident.id)).where(
                    ComplianceIncident.status == INCIDENT_OPEN
                )
            )
        ).scalar()
        or 0
    )
    families.append("# HELP helix_compliance_incidents_open Open compliance incidents.")
    families.append("# TYPE helix_compliance_incidents_open gauge")
    families.append(_line("helix_compliance_incidents_open", {}, open_incidents))
    families.append("")

    under_licensed = int(
        (
            await db.execute(
                select(func.count(SoftwareSeatSyncStat.id))
                .join(
                    SoftwareLicensePlan,
                    SoftwareLicensePlan.id == SoftwareSeatSyncStat.plan_id,
                )
                .where(
                    SoftwareLicensePlan.max_seats.isnot(None),
                    SoftwareSeatSyncStat.vendor_license_holders
                    > SoftwareLicensePlan.max_seats,
                )
            )
        ).scalar()
        or 0
    )
    families.append("# HELP helix_under_licensed_plans Plans where vendor holders > contract max_seats.")
    families.append("# TYPE helix_under_licensed_plans gauge")
    families.append(_line("helix_under_licensed_plans", {}, under_licensed))
    families.append("")

    return "\n".join(families) + "\n"


__all__: list[str] = ["render_metrics"]


# Annotation helper used by router so the dependency-injection layer
# doesn't have to import the heavy module.
_: Any = None