"""HRIS batch async-apply: ``applying`` status + heartbeat cols + resume cron.

Revision ID: 293
Revises: 292
Create Date: 2026-06-16

Background — full-bootstrap HRIS confirm timed out the frontend in prod
(2026-06-16): ``confirm_batch`` applied all ~500 staged events plus the
org-wide HRBP/Budget-Owner refresh, manager-backfill and annual-leave
tail passes *synchronously inside the HTTP request*. On sf-cn-prod that
ran for minutes; the browser's axios timeout aborted the connection
(nginx logged ``499``) and showed the admin an error toast — even though
the backend kept churning and the batch did eventually reach
``confirmed``.

Fix (this migration + PR): the admin-confirm HTTP path now flips the
batch to a new ``applying`` status and runs the heavy apply on a
background task, returning ``202`` immediately so the UI can poll to
terminal. Mirrors the proven inline-primary + cron-backstop shape from
migration 239 (PR #279). This migration:

1. Appends ``applying`` to the ``hr_sync_batch_status`` enum (appended
   last → in-place metadata-only ALTER on MySQL/TiDB).
2. Adds ``apply_started_at`` / ``apply_heartbeat_at`` so the restart-
   safety reconcile cron can distinguish a healthy in-flight apply
   (recent heartbeat) from one whose backend process died mid-run
   (stale heartbeat) and only resume the latter.
3. Seeds the ``hris.resume_stuck_apply`` scheduled job (60s interval) —
   the durable backstop that resumes any ``applying`` batch left stale
   by a backend restart. The per-event apply is idempotent/resumable,
   so re-running is safe.

Idempotent: column adds are guarded on information_schema; the job
INSERT is gated on a name-based NOT EXISTS.
"""
from __future__ import annotations

import uuid

import sqlalchemy as sa

from alembic import op

revision = "293"
down_revision = "292"
branch_labels = None
depends_on = None


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


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

    # 1) Append ``applying`` to the status enum (in-place on MySQL/TiDB).
    bind.execute(
        sa.text(
            "ALTER TABLE hr_sync_batches MODIFY COLUMN status "
            "ENUM('pending_review','confirmed','cancelled','failed','applying') "
            "NOT NULL DEFAULT 'pending_review'"
        )
    )

    # 2) Async-apply bookkeeping columns.
    if not _column_exists(bind, "hr_sync_batches", "apply_started_at"):
        op.add_column(
            "hr_sync_batches",
            sa.Column("apply_started_at", sa.DateTime(timezone=True), nullable=True),
        )
    if not _column_exists(bind, "hr_sync_batches", "apply_heartbeat_at"):
        op.add_column(
            "hr_sync_batches",
            sa.Column("apply_heartbeat_at", sa.DateTime(timezone=True), nullable=True),
        )

    # 3) Seed the restart-safety reconcile cron (60s interval).
    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, 'Asia/Shanghai', 1, :config, 0, 0, NOW(), NOW() "
                "FROM DUAL WHERE NOT EXISTS "
                "(SELECT 1 FROM scheduled_jobs WHERE name = :name)"
            ),
            {
                "id": str(uuid.uuid4()),
                "name": "hris.resume_stuck_apply",
                "display": "HRIS 批次异步应用 · 重启兜底",
                "desc": (
                    "每 60 秒扫描 status='applying' 且心跳 (apply_heartbeat_at) "
                    "已过期的 HRIS 同步批次 —— 这类批次的后台 apply 任务因 backend "
                    "进程重启/崩溃而中途丢失。重跑幂等的逐事件 apply 把剩余事件补完"
                    "并落终态。主路径是 confirm 接口 flip 到 applying 后用 "
                    "background.spawn 立即起后台任务;该 cron 是抗重启的第二条腰带"
                    "(对齐迁移 239 / PR #279 的 inline + cron 兜底形态)。"
                    "心跳新鲜的在途任务不会被打断,避免双跑。"
                ),
                "handler": "hris.resume_stuck_apply",
                "interval": 60,
                "config": "{}",
            },
        )


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

    if _table_exists(bind, "scheduled_jobs"):
        bind.execute(
            sa.text(
                "DELETE FROM scheduled_jobs "
                "WHERE handler_key = 'hris.resume_stuck_apply'"
            )
        )

    # Collapse any leftover ``applying`` rows so the narrowed enum accepts
    # the existing data, then drop the value.
    bind.execute(
        sa.text(
            "UPDATE hr_sync_batches SET status = 'pending_review' "
            "WHERE status = 'applying'"
        )
    )
    bind.execute(
        sa.text(
            "ALTER TABLE hr_sync_batches MODIFY COLUMN status "
            "ENUM('pending_review','confirmed','cancelled','failed') "
            "NOT NULL DEFAULT 'pending_review'"
        )
    )

    if _column_exists(bind, "hr_sync_batches", "apply_heartbeat_at"):
        op.drop_column("hr_sync_batches", "apply_heartbeat_at")
    if _column_exists(bind, "hr_sync_batches", "apply_started_at"):
        op.drop_column("hr_sync_batches", "apply_started_at")