"""Scheduled-job ledger hygiene + execution-capacity model.

Two questions the admin console could not answer before this module, both
of which cost a real production investigation on 2026-09-05:

1. **"Why is this row not doing anything?"** A ``scheduled_jobs`` row can
   be enabled, green, and completely inert — because the cloud
   integration its ``config.integration_id`` points at is disabled (35
   such rows in prod), because its ``handler_key`` no longer exists in
   the registry (a handler renamed without retiring the row), or because
   every recent run short-circuited to ``skipped``. All three render
   identically in the list: "enabled, last status ok/skipped". The
   health resolver below names the condition so the UI can say it.

2. **"Why did this job start 20 minutes late?"** The execution gate
   (``SCHEDULED_JOB_MAX_CONCURRENCY`` = 8 global / 4 heavy) is a hard
   ceiling, and a cluster of long cloud sweeps coming due in the same
   hour queues everything behind it — measured in prod: ``cmdb.grants``
   for aws-sg runs 9 s and waited 1384 s. Nothing in the product showed
   that. :func:`capacity_summary` turns the run ledger into a demand-vs-
   capacity picture (per weight class, per hour of day, per job) so the
   pile-up is visible before it becomes a late-night incident.

Both are computed from rows that already exist — no new table, no extra
bookkeeping on the hot path.

**Observed, not declared.** Every number here is derived from
``scheduled_job_runs`` history rather than from parsing ``cron_expr``.
A cron parser would tell us what the schedule *claims*; the ledger tells
us what actually happened, including misfires, gate queueing, disabled
windows and manual triggers. When the two disagree the ledger is the one
an operator needs.
"""
from __future__ import annotations

import logging
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import Any, Literal
from uuid import UUID

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

from app.scheduled_jobs.models import (
    TRIGGER_CHAINED,
    ScheduledJob,
    ScheduledJobRun,
)
from app.scheduled_jobs.registry import get_handler

logger = logging.getLogger(__name__)


# ── Health ────────────────────────────────────────────────────────

JobHealth = Literal[
    "ok",
    "handler_missing",
    "integration_missing",
    "chain_orphaned",
    "integration_disabled",
    "chain_stalled",
    "never_run",
    "idle_skipping",
]

#: Resolution order — the first condition that matches wins in
#: :func:`resolve_job_health`, and this tuple is that order **verbatim**.
#: Ordered by "how badly is this row broken": a row whose handler vanished
#: can never run again, whereas a row that is merely skipping is doing
#: exactly what it was asked to.
#:
#: 🔴 It also seeds the census's ``counts`` / ``offenders`` keys, so it is
#: easy to reorder for presentation and leave the resolver saying something
#: else. It listed ``integration_disabled`` before ``chain_stalled`` while
#: the resolver returned ``chain_stalled`` first — reachable for a chained
#: row whose predecessor is disabled AND whose integration is disabled.
#: ``test_health_priority_matches_the_resolver`` pins the two together, so
#: changing one without the other is a red test rather than a doc that
#: quietly stops describing the code.
_HEALTH_PRIORITY: tuple[JobHealth, ...] = (
    "handler_missing",
    "integration_missing",
    "chain_orphaned",
    "chain_stalled",
    "integration_disabled",
    "never_run",
    "idle_skipping",
)

#: A row is called ``never_run`` only after it has had a fair chance to
#: fire. A monthly cron seeded this morning is not broken.
NEVER_RUN_GRACE = timedelta(days=2)

#: ``idle_skipping`` needs enough evidence that one flaky window cannot
#: trip it: at least this many terminal-ish runs inside the window, all
#: of them ``skipped``.
IDLE_MIN_RUNS = 10
IDLE_WINDOW_DAYS = 14


@dataclass(slots=True, frozen=True)
class JobRunFacts:
    """Per-job aggregate over the health window (one grouped query)."""

    runs: int = 0
    ok: int = 0
    failed: int = 0
    skipped: int = 0


def config_integration_id_of(config: Any) -> str | None:
    """The dict-level half of :func:`config_integration_id`.

    Separate because callers that read ``config`` as a plain column
    (``chain.plan_advances``) must reach the SAME verdict as the one
    ``execute_job``'s pre-gate reaches from an ORM instance. A second,
    slightly different predicate is how "this row is integration-bound"
    starts meaning two things: the looser reading lets a job run while
    the stricter one quietly excludes it from the chain reconciler, and
    the successor stops advancing with nothing anywhere saying why.

    Tolerant by design: ``config`` is operator-editable JSON, so the
    value can legitimately be absent, empty, or a non-string. Anything
    that is not a plain non-empty string means "not integration-bound"
    — never an error, because a malformed config must not change how a
    job's health renders (and must never block execution; see
    :func:`integration_gate_state`).
    """
    if not isinstance(config, dict):
        return None
    raw = config.get("integration_id")
    if not isinstance(raw, str):
        return None
    value = raw.strip()
    return value or None


def config_integration_id(job: ScheduledJob) -> str | None:
    """Return the ``integration_id`` this row is bound to, if any."""
    return config_integration_id_of(job.config or {})


def resolve_job_health(
    job: ScheduledJob,
    *,
    integration_states: dict[str, bool],
    facts: JobRunFacts | None = None,
    now: datetime | None = None,
    parent_enabled: dict[str, bool] | None = None,
) -> JobHealth:
    """Classify one job row.

    ``integration_states`` maps ``integration_id`` → ``enabled``; an id
    that is absent from the mapping is treated as deleted. ``facts`` is
    the run aggregate for this job over :data:`IDLE_WINDOW_DAYS`;
    ``None`` means "no runs in the window", which is not by itself a
    fault (a monthly cron has none for 29 days out of 30).
    ``parent_enabled`` maps a chain predecessor's id → its ``enabled``
    flag, for the two chain states below.
    """
    now = now or datetime.now(UTC)

    if get_handler(job.handler_key) is None:
        return "handler_missing"

    integration_id = config_integration_id(job)
    if integration_id is not None:
        enabled = integration_states.get(integration_id)
        if enabled is None:
            return "integration_missing"

    # A chained row's trigger is another row. Both ways that can stop
    # being true look exactly like a healthy enabled job in the list,
    # which is the whole reason this census exists.
    if job.trigger_type == TRIGGER_CHAINED:
        if job.chain_after_job_id is None:
            # The FK clears the edge instead of cascading the delete, so
            # this is what a deleted predecessor leaves behind. It can
            # never fire again under any state of the system.
            return "chain_orphaned"
        if parent_enabled is not None:
            upstream = parent_enabled.get(str(job.chain_after_job_id))
            if upstream is False:
                # Recoverable and possibly deliberate — re-enabling the
                # predecessor resumes this row by itself — so it ranks
                # alongside ``integration_disabled`` rather than with the
                # states that can never run again.
                return "chain_stalled"

    if integration_id is not None and integration_states.get(integration_id) is False:
        return "integration_disabled"

    # Everything below is only meaningful for a row that is supposed to
    # be firing. A deliberately disabled row is not "unhealthy".
    if not job.enabled:
        return "ok"

    created = job.created_at
    if created is not None and created.tzinfo is None:
        created = created.replace(tzinfo=UTC)
    # Both signals have to agree that nothing ever ran: the job row's
    # own counters AND the run ledger. They are kept in lockstep by
    # ``execute_job`` (every terminal write-back bumps ``run_count``),
    # so requiring both costs nothing — and it means a row whose
    # counters were seeded or repaired by hand, while real runs exist in
    # the ledger, is not reported as broken. ``facts`` is already loaded
    # for the whole page, so this adds no query.
    ledger_empty = facts is None or facts.runs == 0
    if (
        job.run_count == 0
        and job.last_run_at is None
        and ledger_empty
        and (created is None or now - created > NEVER_RUN_GRACE)
    ):
        return "never_run"

    if (
        facts is not None
        and facts.runs >= IDLE_MIN_RUNS
        and facts.skipped == facts.runs
    ):
        return "idle_skipping"

    return "ok"


async def load_enabled_flags(session: AsyncSession) -> dict[str, bool]:
    """``{job_id: enabled}`` over every row, for the chain states.

    One query for the whole ledger (~350 rows). Resolving a
    predecessor's enabled flag per row would turn the cheapest column on
    the page into the most expensive.
    """
    rows = (
        await session.execute(select(ScheduledJob.id, ScheduledJob.enabled))
    ).all()
    return {str(row[0]): bool(row[1]) for row in rows}


async def load_integration_states(session: AsyncSession) -> dict[str, bool]:
    """``{integration_id: enabled}`` for every integration row.

    One query, ~20 rows in prod. Keys are the string form of the id so
    they compare directly against the JSON ``config.integration_id``
    written by the seeding migrations, whatever the DB's id column type.
    """
    from app.integrations.models import Integration

    rows = (
        await session.execute(select(Integration.id, Integration.enabled))
    ).all()
    return {str(row[0]): bool(row[1]) for row in rows}


async def load_run_facts(
    session: AsyncSession,
    job_ids: list[UUID],
    *,
    window_days: int = IDLE_WINDOW_DAYS,
    now: datetime | None = None,
) -> dict[str, JobRunFacts]:
    """Grouped run aggregate for the given jobs, keyed by ``str(job_id)``.

    Restricted to the caller's page of ids so the admin list stays O(1)
    round-trips regardless of how many rows the ledger holds; the
    ``ix_scheduled_job_runs_job_started`` composite index covers both
    the ``job_id IN (...)`` narrowing and the ``started_at`` range.
    """
    if not job_ids:
        return {}
    now = now or datetime.now(UTC)
    window_start = now - timedelta(days=window_days)
    rows = (
        await session.execute(
            select(
                ScheduledJobRun.job_id,
                func.count(),
                func.sum(case_status(ScheduledJobRun.status, "ok")),
                func.sum(case_status(ScheduledJobRun.status, "failed")),
                func.sum(case_status(ScheduledJobRun.status, "skipped")),
            )
            .where(
                ScheduledJobRun.job_id.in_(job_ids),
                ScheduledJobRun.started_at >= window_start,
            )
            .group_by(ScheduledJobRun.job_id)
        )
    ).all()
    return {
        str(row[0]): JobRunFacts(
            runs=int(row[1] or 0),
            ok=int(row[2] or 0),
            failed=int(row[3] or 0),
            skipped=int(row[4] or 0),
        )
        for row in rows
    }


def case_status(column: Any, value: str) -> Any:
    """``1`` when ``column == value`` else ``0``, portable across engines.

    ``SUM(col = 'x')`` is MySQL-only shorthand; the explicit CASE keeps
    the same aggregate readable and engine-neutral.
    """
    from sqlalchemy import case

    return case((column == value, 1), else_=0)


async def annotate_health(
    session: AsyncSession,
    jobs: list[ScheduledJob],
    *,
    now: datetime | None = None,
) -> dict[str, JobHealth]:
    """Resolve health for a page of jobs in two queries total."""
    if not jobs:
        return {}
    now = now or datetime.now(UTC)
    integration_states = await load_integration_states(session)
    facts = await load_run_facts(session, [j.id for j in jobs], now=now)
    parent_enabled = (
        await load_enabled_flags(session)
        if any(j.chain_after_job_id is not None for j in jobs)
        else {}
    )
    return {
        str(job.id): resolve_job_health(
            job,
            integration_states=integration_states,
            facts=facts.get(str(job.id)),
            now=now,
            parent_enabled=parent_enabled,
        )
        for job in jobs
    }


async def health_census(
    session: AsyncSession, *, now: datetime | None = None,
) -> dict[str, Any]:
    """Fleet-wide health counts + the offending rows, for the console.

    Unlike :func:`annotate_health` this deliberately scans every row —
    it is the "is my ledger clean" view, and a census that only looked
    at one page would answer a different question.
    """
    now = now or datetime.now(UTC)
    jobs = list((await session.execute(select(ScheduledJob))).scalars().all())
    integration_states = await load_integration_states(session)
    facts = await load_run_facts(session, [j.id for j in jobs], now=now)
    parent_enabled = {str(j.id): bool(j.enabled) for j in jobs}

    counts: dict[str, int] = {"ok": 0}
    counts.update({state: 0 for state in _HEALTH_PRIORITY})
    offenders: dict[str, list[dict[str, Any]]] = {
        state: [] for state in _HEALTH_PRIORITY
    }
    for job in jobs:
        state = resolve_job_health(
            job,
            integration_states=integration_states,
            facts=facts.get(str(job.id)),
            now=now,
            parent_enabled=parent_enabled,
        )
        counts[state] = counts.get(state, 0) + 1
        if state != "ok":
            offenders[state].append(
                {
                    "id": str(job.id),
                    "name": job.name,
                    "handler_key": job.handler_key,
                    "enabled": bool(job.enabled),
                },
            )
    return {
        "checked_at": now.isoformat(),
        "total": len(jobs),
        "counts": counts,
        "offenders": offenders,
    }


# ── Execution-gate pre-check ──────────────────────────────────────


@dataclass(slots=True, frozen=True)
class IntegrationGateState:
    """Verdict of the pre-gate integration check for one execution."""

    #: ``None`` when the job is not integration-bound.
    integration_id: str | None
    #: ``True`` only when the job may proceed to acquire a slot.
    runnable: bool
    #: Stable machine-readable reason, mirrored into the skipped run row.
    reason: str | None = None


async def integration_gate_state(
    session: AsyncSession, job: ScheduledJob,
) -> IntegrationGateState:
    """Decide whether ``job`` can do any work at all, before it queues.

    **Why this runs before the concurrency gate.** Every per-scope cloud
    handler already returns ``{"status": "skipped", "reason":
    "integration disabled"}`` — but it does so *from inside the handler*,
    which means the run has already taken one of the four ``heavy``
    slots and held it while opening a session and loading the
    integration. In prod that is 35 enabled rows bound to 5 disabled
    integrations, each taking a slot every day to discover it has
    nothing to do, in the exact hours where live sweeps are queueing.
    Checking here costs one indexed primary-key read on a session we do
    not hold open, and the handler-level checks stay as defence in depth.

    Fails **open**: any error resolving the integration (malformed id in
    operator-edited config, a transient read) yields ``runnable=True``
    so the handler decides. A hygiene optimisation must never be the
    reason a sync did not run.
    """
    integration_id = config_integration_id(job)
    if integration_id is None:
        return IntegrationGateState(integration_id=None, runnable=True)

    from app.integrations.models import Integration

    # Read-only: a single indexed SELECT. Nothing is added to the session and
    # nothing flushes, so a failure cannot deactivate the caller's transaction.
    # The handler this runs in front of re-checks the same fact anyway, so
    # swallowing here can only cost the optimisation, never correctness.
    # savepoint-exempt: read-only SELECT, fails open with runnable=True
    try:
        enabled = (
            await session.execute(
                select(Integration.enabled).where(
                    Integration.id == integration_id,
                ),
            )
        ).scalar_one_or_none()
    except Exception:  # noqa: BLE001 - never block a sync on this check
        logger.warning(
            "integration pre-gate check failed for job=%s integration=%s; "
            "letting the handler decide",
            job.name, integration_id, exc_info=True,
        )
        return IntegrationGateState(
            integration_id=integration_id, runnable=True,
        )

    if enabled is None:
        return IntegrationGateState(
            integration_id=integration_id,
            runnable=False,
            reason="integration_missing",
        )
    if not enabled:
        return IntegrationGateState(
            integration_id=integration_id,
            runnable=False,
            reason="integration_disabled",
        )
    return IntegrationGateState(integration_id=integration_id, runnable=True)


def should_record_gate_skip(job: ScheduledJob, reason: str) -> bool:
    """Whether a pre-gate skip deserves its own run row.

    The first skip after a state change is evidence an operator wants
    ("this stopped running on the 3rd, when the integration went off").
    The 400th identical skip is landfill that evicts that evidence —
    ``MAX_RUN_HISTORY_PER_JOB`` keeps only the last 100 runs, so a
    permanently-off integration would erase the job's entire real
    history within days. Record the transition, then stay quiet.
    """
    if job.last_status != "skipped":
        return True
    summary = job.last_result_summary
    if not isinstance(summary, dict):
        return True
    return summary.get("reason") != reason


# ── Capacity ──────────────────────────────────────────────────────

#: Weight classes and how many concurrent slots each is allowed. Read
#: from settings at call time so an operator who raises the limit sees
#: the utilisation figure move with it.
def _slot_capacity() -> dict[str, int]:
    from app.core.config import settings

    global_slots = max(1, int(settings.SCHEDULED_JOB_MAX_CONCURRENCY))
    heavy_slots = min(
        max(1, int(settings.SCHEDULED_JOB_HEAVY_MAX_CONCURRENCY)),
        global_slots,
    )
    return {"global": global_slots, "heavy": heavy_slots}


def handler_weight(handler_key: str) -> str:
    handler = get_handler(handler_key)
    return getattr(handler, "weight", "light") if handler else "light"


async def capacity_summary(
    session: AsyncSession,
    *,
    window_days: int = 7,
    top_n: int = 15,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Demand-vs-capacity picture over the trailing ``window_days``.

    Three slices, one query each:

    * **per weight class** — occupied slot-seconds per day against the
      slot ceiling, i.e. "how full is the pipe".
    * **per hour of day** — where the demand actually lands, which is
      what makes a nightly pile-up legible. Bucketed in UTC because
      that is the clock the executor runs on; the UI labels it.
    * **per job** — the ranked consumers, so the answer to "what should
      I move" is on screen instead of in a DB session.

    Only ``ok`` and ``skipped`` runs count toward demand. A ``failed``
    run's duration is real wall-clock too, but it is unrepresentative
    (a run killed at the 1800 s timeout would dominate the ranking with
    time the job does not normally need), and ``running`` rows have no
    duration yet.
    """
    now = now or datetime.now(UTC)
    window_start = now - timedelta(days=window_days)
    counted_statuses = ("ok", "skipped")

    gate_wait_ms = cast(
        func.json_extract(ScheduledJobRun.result_summary, "$._gate_wait_ms"),
        Integer,
    )

    # ── per job ───────────────────────────────────────────────────
    per_job_rows = (
        await session.execute(
            select(
                ScheduledJob.id,
                ScheduledJob.name,
                ScheduledJob.display_name,
                ScheduledJob.handler_key,
                ScheduledJob.enabled,
                func.count(),
                func.avg(ScheduledJobRun.duration_ms),
                func.max(ScheduledJobRun.duration_ms),
                func.sum(ScheduledJobRun.duration_ms),
                func.avg(gate_wait_ms),
                func.max(gate_wait_ms),
            )
            .join(ScheduledJobRun, ScheduledJobRun.job_id == ScheduledJob.id)
            .where(
                ScheduledJobRun.started_at >= window_start,
                ScheduledJobRun.status.in_(counted_statuses),
                ScheduledJobRun.duration_ms.isnot(None),
            )
            .group_by(
                ScheduledJob.id,
                ScheduledJob.name,
                ScheduledJob.display_name,
                ScheduledJob.handler_key,
                ScheduledJob.enabled,
            )
        )
    ).all()

    jobs: list[dict[str, Any]] = []
    demand_by_weight: dict[str, float] = {"heavy": 0.0, "light": 0.0}
    for row in per_job_rows:
        (
            job_id, name, display_name, handler_key, enabled,
            run_count, avg_ms, max_ms, sum_ms, avg_gate, max_gate,
        ) = row
        weight = handler_weight(handler_key)
        daily_seconds = float(sum_ms or 0) / 1000.0 / max(1, window_days)
        demand_by_weight[weight] = demand_by_weight.get(weight, 0.0) + daily_seconds
        jobs.append(
            {
                "id": str(job_id),
                "name": name,
                "display_name": display_name,
                "handler_key": handler_key,
                "enabled": bool(enabled),
                "weight": weight,
                "runs": int(run_count or 0),
                "runs_per_day": round(int(run_count or 0) / max(1, window_days), 2),
                "avg_duration_ms": int(avg_ms or 0),
                "max_duration_ms": int(max_ms or 0),
                "daily_seconds": round(daily_seconds, 1),
                "avg_gate_wait_ms": int(avg_gate or 0),
                "max_gate_wait_ms": int(max_gate or 0),
            },
        )
    jobs.sort(key=lambda j: j["daily_seconds"], reverse=True)

    # ── per hour of day (UTC) ─────────────────────────────────────
    hour_rows = (
        await session.execute(
            select(
                func.hour(ScheduledJobRun.started_at),
                func.count(),
                func.sum(ScheduledJobRun.duration_ms),
                func.avg(gate_wait_ms),
            )
            .where(
                ScheduledJobRun.started_at >= window_start,
                ScheduledJobRun.status.in_(counted_statuses),
                ScheduledJobRun.duration_ms.isnot(None),
            )
            .group_by(func.hour(ScheduledJobRun.started_at))
        )
    ).all()
    by_hour_map = {
        int(row[0]): {
            "hour_utc": int(row[0]),
            "runs_per_day": round(int(row[1] or 0) / max(1, window_days), 2),
            "seconds_per_day": round(
                float(row[2] or 0) / 1000.0 / max(1, window_days), 1,
            ),
            "avg_gate_wait_ms": int(row[3] or 0),
        }
        for row in hour_rows
    }
    by_hour = [
        by_hour_map.get(
            hour,
            {
                "hour_utc": hour,
                "runs_per_day": 0.0,
                "seconds_per_day": 0.0,
                "avg_gate_wait_ms": 0,
            },
        )
        for hour in range(24)
    ]

    # ── headline ─────────────────────────────────────────────────
    slots = _slot_capacity()
    light_slots = max(0, slots["global"] - slots["heavy"])
    # Two gates, two rows — and the ``all`` row counts HEAVY demand too.
    # A heavy run holds the heavy sub-gate *and* a global slot (strict
    # acquisition order, see ``_execution_slot``), so a global-gate
    # utilisation figure that summed only the light jobs would report a
    # near-empty pipe on exactly the nights the pipe is full. Caught in
    # browser verification: the tile read 0% while heavy sat at 13%.
    classes: list[dict[str, Any]] = []
    for scope, scope_slots, scope_demand in (
        ("heavy", slots["heavy"], demand_by_weight.get("heavy", 0.0)),
        ("all", slots["global"], sum(demand_by_weight.values())),
    ):
        demand_seconds = round(scope_demand, 1)
        capacity_seconds = scope_slots * 86400.0
        classes.append(
            {
                "scope": scope,
                "slots": scope_slots,
                "demand_seconds_per_day": demand_seconds,
                "capacity_seconds_per_day": round(capacity_seconds, 1),
                "utilisation": (
                    round(demand_seconds / capacity_seconds, 4)
                    if capacity_seconds > 0
                    else None
                ),
            },
        )

    queued = [j for j in jobs if j["avg_gate_wait_ms"] > 1000]
    queued.sort(key=lambda j: j["avg_gate_wait_ms"], reverse=True)

    return {
        "checked_at": now.isoformat(),
        "window_days": window_days,
        "slots": {
            "global": slots["global"],
            "heavy": slots["heavy"],
            "light_reserved": light_slots,
        },
        "classes": classes,
        "by_hour_utc": by_hour,
        "top_jobs": jobs[:top_n],
        "most_queued": queued[:top_n],
    }