"""Seed the ``asset.stale_reservation_sweep`` daily cron.

Why (2026-09-05). ``HardwarePrepService.stale_reservations`` — "this
device is earmarked for a hire who no longer has an open onboarding
process" — only ran when somebody opened the 待备机 tab. Nothing else in
the platform looked at reservations at all: no scheduled job referenced
them, and ``delete_asset`` neither refused nor cleared a reserved row,
so a soft delete made the earmark vanish silently. A rescinded offer
therefore held a laptop out of stock indefinitely, with the only
detection path being a human happening to visit one tab.

That became materially worse on 2026-09-05, when an earmark started
satisfying the Day-1 hardware dimension (see migration-adjacent notes in
``onboarding_hardware_coordinator``): a zombie earmark is now a device
that is BOTH unavailable to the next hire AND quietly counted as some
departed candidate's Day-1 readiness. So the sweep ships in the same
series as the change that made it necessary.

04:10 Asia/Shanghai: after the 03:15 asset-admin reconcile and well
before CN work hours, so an admin arriving in the morning sees stock that
is already truthful. Enabled on seed — the whole point is that it runs
without anybody remembering to look.

No schema change. Seed guarded by NOT EXISTS on ``handler_key``, so
re-running is a no-op and an operator who later disables the job does not
get it silently re-enabled by a redeploy.

Revision ID: 608
Revises: 607

⚠️ Renumbered TWICE while unpushed: 606 → 607 (the access-lifecycle
series' PR2 landed 606) → 608 (``607_secaw_material_machine_translation``
landed 607). Two branches on one number means two heads, which breaks
every backend shard and the image build. On a repo with several sessions
merging in parallel, re-checking ``origin/main`` once is not enough —
re-check it in the same breath as ``git push``.
"""
from __future__ import annotations

import uuid

import sqlalchemy as sa

from alembic import op

revision = "608"
down_revision = "607"
branch_labels = None
depends_on = None

_HANDLER = "asset.stale_reservation_sweep"
_DISPLAY = "备机预留僵尸清理"
_DESC = (
    "每日 04:10(Asia/Shanghai)释放「预留给某个新人、但那个新人已经没有"
    "在办入职单」的备机预留(撤销 offer / 入职单已完成或已取消)。以前这个"
    "判断只在有人打开「待备机」tab 时算一次,平台里没有任何定时任务看过"
    "预留,一个被撤销的 offer 会把一台笔记本无限期扣在库外;而 2026-09-05"
    "之后预留还会满足 Day-1 硬件维度,僵尸预留同时意味着设备不可用且被算成"
    "某个人的 Day-1 就绪。释放走与人工释放同一条凭证路径,锁下重新断言,"
    "不会抢掉刚建立的预留。"
)
_CRON = "10 4 * * *"


def _table_exists(bind, name: str) -> bool:
    """Existence guard scoped to the CURRENT database.

    ``DATABASE()`` matters: without it this matches a same-named table in
    any schema on the instance — a false positive on a shared test
    server.
    """
    return bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.tables "
            "WHERE table_schema = DATABASE() AND table_name = :n"
        ),
        {"n": name},
    ).scalar() is not None


def upgrade() -> None:
    bind = op.get_bind()
    if not _table_exists(bind, "scheduled_jobs"):
        return
    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) "
            "SELECT :id, :name, :display, :desc, :handler, 'cron', "
            ":cron, 'Asia/Shanghai', 1, '{}', 0, 0, NOW(), NOW() "
            "FROM DUAL WHERE NOT EXISTS "
            "(SELECT 1 FROM scheduled_jobs WHERE handler_key = :handler)"
        ),
        {
            "id": str(uuid.uuid4()),
            "name": _HANDLER,
            "display": _DISPLAY,
            "desc": _DESC,
            "handler": _HANDLER,
            "cron": _CRON,
        },
    )


def downgrade() -> None:
    bind = op.get_bind()
    if _table_exists(bind, "scheduled_jobs"):
        bind.execute(
            sa.text("DELETE FROM scheduled_jobs WHERE handler_key = :k"),
            {"k": _HANDLER},
        )