"""Identity-sync execution plane: ``sched_failure_streak`` column +
stale-running-run cleanup cron.

Two execution-plane fixes for the sync engine ride this migration:

1. ``identity_sync_targets.sched_failure_streak`` (INT, default 0).
   Powers same-day failure backoff in ``_finalise_run``. Before this,
   the cron cursor (``next_run_at``) advanced to the NEXT slot on ANY
   scheduled-run finalise — so a single transient vendor failure (or a
   deploy that cancelled the sweep) cost a full day of staleness, since
   the next reconcile only came 24h later. Now a failed scheduled run
   re-arms ``next_run_at`` to a short exponential backoff and retries
   the same day (up to a cap), and a cancelled run (worker restart /
   deploy) re-arms almost immediately. The streak tracks consecutive
   scheduled failures so the backoff can escalate and eventually fall
   back to the daily slot rather than hammering a hard-down vendor.

2. ``identity_sync.cleanup_stale_runs`` scheduled job (every 15 min).
   A hard-killed worker (SIGKILL, OOM) leaves an ``IdentitySyncRun`` row
   pinned at ``status='running'`` — the per-target Redis lock TTLs out
   but the row does not. The web-app startup watchdog only sweeps these
   on boot with a 30-min floor, so a run orphaned minutes before a boot
   lingers as a phantom "running" row until the *next* restart. This
   recurring job finalises any run stuck ``running`` past the threshold
   as ``cancelled`` (an infra event, NOT a data failure — excluded from
   the success-rate and failure alerting), independent of process boots.
"""
import json as _json
import uuid

import sqlalchemy as sa

from alembic import op

revision = "402"
down_revision = "401"
branch_labels = None
depends_on = None


def _table_exists(bind, name: str) -> bool:
    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 _column_exists(bind, table: str, column: str) -> bool:
    return bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.COLUMNS "
            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t "
            "AND COLUMN_NAME = :c"
        ),
        {"t": table, "c": column},
    ).scalar() is not None


HANDLER_KEY = "identity_sync.cleanup_stale_runs"
JOB_NAME = "identity_sync.cleanup_stale_runs"

_DESC = (
    "每 15 分钟将卡在 status='running' 且超过阈值(config.stale_minutes,默认 30)"
    "的 identity_sync_runs 收尾为 'cancelled'(worker 硬退出 / OOM 留下的幽灵"
    "运行行 —— 分布式锁会 TTL 过期但运行行不会)。cancelled 是基础设施事件,"
    "不计入失败率、不触发失败告警;对应目标的 next_run_at 保持不动(硬退出未推进"
    "游标),下一调度 tick 自然补跑。补的是 web 启动看门狗(仅开机扫、30 分钟下限)"
    "覆盖不到的运行期盲区。"
)


def upgrade() -> None:
    bind = op.get_bind()
    if _table_exists(bind, "identity_sync_targets") and not _column_exists(
        bind, "identity_sync_targets", "sched_failure_streak",
    ):
        op.add_column(
            "identity_sync_targets",
            sa.Column(
                "sched_failure_streak",
                sa.Integer(),
                nullable=False,
                server_default="0",
            ),
        )

    if _table_exists(bind, "scheduled_jobs"):
        bind.execute(
            sa.text(
                "INSERT INTO scheduled_jobs "
                "(id, name, display_name, description, handler_key, "
                " trigger_type, interval_seconds, timezone, enabled, config, "
                " run_count, failure_count, created_at, updated_at) "
                "SELECT :id, :name, :display, :desc, :handler, 'interval', "
                ":interval, '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": JOB_NAME,
                "display": "数据同步中心 · 幽灵运行行清理",
                "desc": _DESC,
                "handler": HANDLER_KEY,
                "interval": 900,
                "config": _json.dumps({"stale_minutes": 30}),
            },
        )


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_KEY},
        )
    if _table_exists(bind, "identity_sync_targets") and _column_exists(
        bind, "identity_sync_targets", "sched_failure_streak",
    ):
        op.drop_column("identity_sync_targets", "sched_failure_streak")