"""Seed the access-certification revoke-fulfilment scheduled job.

Revision ID: 257
Revises: 256
Create Date: 2026-06-05

No schema change — registers the ``auditor.access_certification_revoke_fulfil``
cron (every 5 min) that drains ``revocation_status='pending'`` cert items by
reconciling them against their provisioning revoke task. Idempotent insert
(skips when the handler_key already exists), hyphenated ``UUID()`` id to
avoid the no-dash-UUID attach bug.
"""
from __future__ import annotations

import json as _json
import uuid

import sqlalchemy as sa

from alembic import op

revision = "257"
down_revision = "256"
branch_labels = None
depends_on = None

_HANDLER = "auditor.access_certification_revoke_fulfil"
_DESC = (
    "每 5 分钟对账访问认证的 revoke 决策:重试未落地的撤权入队,并把下游 "
    "provisioning 撤权任务的终态(完成/失败)镜像回 access_certification_items "
    "的 revocation_status,把复核员的撤权决策真正落到下游。"
)


def _table_exists(name: str) -> bool:
    return op.get_bind().execute(
        sa.text(
            "SELECT 1 FROM information_schema.TABLES "
            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n LIMIT 1"
        ),
        {"n": name},
    ).scalar() is not None


def upgrade() -> None:
    if not _table_exists("scheduled_jobs"):
        return
    op.get_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, 'UTC', 1, :config, 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": "访问认证 · 撤权执行与对账",
            "desc": _DESC,
            "handler": _HANDLER,
            "cron": "*/5 * * * *",
            "config": _json.dumps({"batch_size": 100}),
        },
    )


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