"""Ledger-hygiene cron: keep ``scheduled_jobs`` honest about itself.

The scheduled-job ledger is the one table in Helix that nobody owns.
Rows are seeded by migrations, bound to integrations by an untyped JSON
field, and left behind when the thing they synced goes away — there is
no foreign key to cascade and no service layer to notice. Three drift
shapes have all been observed in prod:

* a handler renamed in code while its row stayed (``jamf.seat_sync``
  survived the rename to ``jamf.device_sync``; the row is dead weight
  that can never run again),
* a one-shot bootstrap row that finished its job and now exists only to
  confuse the next reader (``intelligence.cloud_scope_baseline_bootstrap``),
* rows bound to an integration that was disabled or deleted — 35 of
  them, still firing daily to discover they have nothing to do.

This job does not try to be clever about any of that. It **reports**
everything and **acts** on exactly one case: a row whose integration row
no longer exists at all. That row cannot run under any future state of
the system, so leaving it enabled is strictly worse than disabling it,
and re-enabling costs one click if a human disagrees. Every other drift
shape is surfaced in the console and left to a human, because "disabled
integration" and "quiet job" are both legitimate steady states an
operator may have chosen on purpose.

**Why this does not raise when it disables something.** The obvious way
to get an alert out of a cron in this codebase is to raise, which is how
``identity_sync.staleness_watchdog`` reports drift. It is the wrong tool
here: raising would record a run that did exactly its job as ``failed``,
inflate ``failure_count``, and trip this handler's SLO — the same class
of ledger lie that ``_run_status_for_summary`` exists to prevent in the
other direction. The auto-disable is durably recorded in three places an
operator already reads (the audit log, this run's summary, and the
console's hygiene panel), and none of them require a red card.
"""
from __future__ import annotations

import logging
import uuid
from datetime import UTC, datetime
from typing import Any

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

from app.scheduled_jobs.hygiene import (
    config_integration_id,
    health_census,
    load_integration_states,
)
from app.scheduled_jobs.models import ScheduledJob
from app.scheduled_jobs.registry import register_scheduled_job
from app.scheduled_jobs.slo import SLO_DAILY_DIGEST

logger = logging.getLogger(__name__)

LEDGER_RECONCILE_HANDLER = "scheduler.ledger_reconcile"


@register_scheduled_job(
    key=LEDGER_RECONCILE_HANDLER,
    display_name="定时任务台账体检",
    description=(
        "每日体检 scheduled_jobs 台账:统计每一行的健康态(处理器缺失 / "
        "集成已删除 / 集成已停用 / 从未运行 / 长期空转),把结果写进运行摘要供"
        "管理台展示。只对**一种**情况动手 —— 绑定的集成行已经被删除、这一行"
        "在任何未来状态下都不可能再跑,才自动停用并记审计;其余一律只报不改,"
        "因为「集成停用」和「安静的任务」都可能是管理员刻意为之。"
        "自动停用不发红卡:它是一次成功的运行,证据落在审计日志、本次运行摘要"
        "和管理台体检面板三处;把成功记成失败会污染失败率与 SLO。"
    ),
    slo=SLO_DAILY_DIGEST,
)
async def run_ledger_reconcile(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.audit.service import AuditService

    dry_run = bool(config.get("dry_run", False))
    now = datetime.now(UTC)

    census = await health_census(db, now=now)
    integration_states = await load_integration_states(db)

    # Orphans: enabled rows whose ``config.integration_id`` names an
    # integration that is not in the table at all. ``integration_states``
    # holds every integration row, so absence here means deleted — not
    # "disabled", which is a different (and legitimate) state.
    orphan_rows = [
        job
        for job in (await db.execute(
            select(ScheduledJob).where(ScheduledJob.enabled.is_(True)),
        )).scalars().all()
        if (iid := config_integration_id(job)) is not None
        and iid not in integration_states
    ]

    disabled: list[dict[str, str]] = []
    if orphan_rows and not dry_run:
        audit = AuditService(db)
        for job in orphan_rows:
            job.enabled = False
            disabled.append(
                {
                    "id": str(job.id),
                    "name": job.name,
                    "handler_key": job.handler_key,
                    "integration_id": config_integration_id(job) or "",
                },
            )
            await audit.log(
                trace_id=uuid.uuid4(),
                actor_type="system",
                action="scheduled_job.auto_disable",
                status="success",
                resource_type="scheduled_job",
                resource_id=str(job.id),
                detail={
                    "reason": "integration_missing",
                    "name": job.name,
                    "handler_key": job.handler_key,
                    "integration_id": config_integration_id(job),
                    "source": LEDGER_RECONCILE_HANDLER,
                },
            )
        await db.commit()

    summary: dict[str, Any] = {
        "checked_at": census["checked_at"],
        "total_jobs": census["total"],
        "counts": census["counts"],
        "auto_disabled": disabled,
        "auto_disabled_count": len(disabled),
        "dry_run": dry_run,
        # Kept small on purpose: the console reads the full offender
        # lists from ``GET /scheduled-jobs/hygiene``. Persisting them
        # all into every run summary would bloat the ledger it audits.
        "offender_samples": {
            state: rows[:5] for state, rows in census["offenders"].items() if rows
        },
    }

    if disabled:
        logger.warning(
            "%s auto-disabled %d job row(s) bound to deleted integrations: %s",
            LEDGER_RECONCILE_HANDLER,
            len(disabled),
            ", ".join(row["name"] for row in disabled[:10]),
        )
    return summary