"""Scheduled-job capacity reclaim + ledger-hygiene cron.

Revision ID: 613
Revises: 612
Create Date: 2026-09-05

Five cadence corrections and one new row. Every cadence edit is
**conditional on the row still carrying the value the code seeded** —
the same discipline migration 418 used for the resource-sync cadence
tiers. An operator who has already tuned one of these rows has made a
decision this migration knows nothing about, and silently overwriting it
would be worse than leaving the default in place.

Measured against production (30 days of ``scheduled_job_runs``, read-only):

``cmdb.resource_directory_refresh`` — every 15 min, 400-450 s per run,
DB-only. **41,095 slot-seconds/day: 77% of the entire heavy-class
demand**, permanently occupying one of the four heavy slots. Of 94
successful runs, 27 found any change at all — because the tables it
projects are themselves refreshed on a 4 h / 12 h cadence by
``cmdb.resource_sync`` (migration 418), so three of every four passes
re-hash 106,518 rows to confirm nothing moved. Hourly keeps the
projection well inside the freshness of its own sources and returns
three heavy slot-hours a day to the fleet.

``cmdb.resource_inventory_refresh`` — hourly, heavy, and every one of
its last 100 runs reported ``inventory_skipped=8, seen=0``: broad
inventory is opt-in per integration and nobody has opted in. Moved to
daily rather than disabled, deliberately — a disabled row needs a human
to remember it when someone finally opts in, whereas a daily row
discovers the opt-in by itself within 24 h at a cost of half a second.

``approval_registry.owner_confirmation_semiannual`` and
``auditor.access_self_attestation_auto_open_semiannual`` — both fire
daily and both open a campaign only inside ``[boundary, boundary+7d)``
for Jun 1 / Dec 1 (``confirmation.due_boundary`` /
``auto_open_semiannual_self_attestation``). 351 of every 365 runs exist
to answer "no". The cron now expresses the predicate the handler already
enforces, which also stops two ``heavy`` rows from taking a slot to say
nothing. The window arithmetic is preserved exactly: ``1-7`` of months
``6,12`` is the same set of days the handlers accept, so a run that used
to open a round still opens it.

``cmdb.auto_owner_grant_backfill`` — every 10 minutes, 16-51 s per run,
4,143 slot-seconds/day; the last 100 runs scanned 1 row and fixed 0.
It is a backstop behind the synchronous post-provision path, not the
path itself, so hourly is the honest cadence for it.

New row ``scheduler.ledger_reconcile`` — the ledger has no owner and no
foreign key to the integrations it references; this is the daily census
that keeps it honest (see ``scheduled_jobs/hygiene_handlers.py``).

Downgrade restores every previous value and removes the new row.
"""
from __future__ import annotations

import json
import uuid

import sqlalchemy as sa

from alembic import op

revision = "613"
down_revision = "612"
branch_labels = None
depends_on = None


LEDGER_RECONCILE_JOB = {
    "name": "scheduler.ledger_reconcile",
    "display_name": "定时任务台账体检",
    "description": (
        "每日体检 scheduled_jobs 台账:统计每一行的健康态(处理器缺失 / "
        "集成已删除 / 集成已停用 / 从未运行 / 长期空转),结果写进运行摘要"
        "供管理台「台账体检」面板展示。只对一种情况动手 —— 绑定的集成行"
        "已被删除、该行在任何未来状态下都不可能再跑,才自动停用并记审计;"
        "其余只报不改。"
    ),
    "handler_key": "scheduler.ledger_reconcile",
    "cron_expr": "50 2 * * *",
    "timezone": "Asia/Shanghai",
}

#: ``(name, column, expected_current, new_value)`` — the edit applies
#: only when the row still holds ``expected_current``.
CRON_CHANGES: tuple[tuple[str, str, str, str], ...] = (
    (
        "cmdb.resource_directory_refresh",
        "cron_expr",
        "7,22,37,52 * * * *",
        "7 * * * *",
    ),
    (
        "cmdb.resource_inventory_refresh",
        "cron_expr",
        "17 * * * *",
        "17 4 * * *",
    ),
    (
        "approval_registry.owner_confirmation_semiannual",
        "cron_expr",
        "30 8 * * *",
        "30 8 1-7 6,12 *",
    ),
    (
        "auditor.access_self_attestation_auto_open_semiannual",
        "cron_expr",
        "0 8 * * *",
        "0 8 1-7 6,12 *",
    ),
)

INTERVAL_CHANGES: tuple[tuple[str, int, int], ...] = (
    ("cmdb.auto_owner_grant_backfill", 600, 3600),
)


def _table_exists(bind, table: str) -> bool:
    return bool(
        bind.execute(
            sa.text(
                "SELECT 1 FROM information_schema.TABLES "
                "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t LIMIT 1",
            ),
            {"t": table},
        ).scalar(),
    )


def _apply_cron(bind, name: str, expected: str, new: str) -> None:
    bind.execute(
        sa.text(
            "UPDATE scheduled_jobs SET cron_expr = :new, updated_at = NOW() "
            "WHERE name = :n AND cron_expr = :expected",
        ),
        {"n": name, "expected": expected, "new": new},
    )


def _apply_interval(bind, name: str, expected: int, new: int) -> None:
    bind.execute(
        sa.text(
            "UPDATE scheduled_jobs SET interval_seconds = :new, "
            "updated_at = NOW() "
            "WHERE name = :n AND interval_seconds = :expected",
        ),
        {"n": name, "expected": expected, "new": new},
    )


def upgrade() -> None:
    bind = op.get_bind()
    if not _table_exists(bind, "scheduled_jobs"):
        return

    for name, _column, expected, new in CRON_CHANGES:
        _apply_cron(bind, name, expected, new)
    for name, expected, new in INTERVAL_CHANGES:
        _apply_interval(bind, name, expected, new)

    already = bind.execute(
        sa.text("SELECT 1 FROM scheduled_jobs WHERE name = :n LIMIT 1"),
        {"n": LEDGER_RECONCILE_JOB["name"]},
    ).scalar()
    if not already:
        bind.execute(
            sa.text(
                "INSERT INTO scheduled_jobs "
                "(id, name, display_name, description, handler_key, "
                " trigger_type, cron_expr, timezone, enabled, config, "
                " run_count, failure_count, created_at, updated_at) "
                "VALUES (:id, :n, :dn, :d, :hk, 'cron', :ce, :tz, 1, :cfg, "
                " 0, 0, NOW(), NOW())",
            ),
            {
                "id": str(uuid.uuid4()),
                "n": LEDGER_RECONCILE_JOB["name"],
                "dn": LEDGER_RECONCILE_JOB["display_name"],
                "d": LEDGER_RECONCILE_JOB["description"],
                "hk": LEDGER_RECONCILE_JOB["handler_key"],
                "ce": LEDGER_RECONCILE_JOB["cron_expr"],
                "tz": LEDGER_RECONCILE_JOB["timezone"],
                "cfg": json.dumps({}),
            },
        )


def downgrade() -> None:
    bind = op.get_bind()
    if not _table_exists(bind, "scheduled_jobs"):
        return
    # Mirror-image guard: restore only rows that still carry the value
    # this migration wrote, so a post-upgrade operator edit survives a
    # downgrade the same way it survived the upgrade.
    for name, _column, expected, new in CRON_CHANGES:
        _apply_cron(bind, name, new, expected)
    for name, expected, new in INTERVAL_CHANGES:
        _apply_interval(bind, name, new, expected)
    bind.execute(
        sa.text("DELETE FROM scheduled_jobs WHERE name = :n"),
        {"n": LEDGER_RECONCILE_JOB["name"]},
    )