"""``scheduled_jobs.chain_after_job_id`` — ordering as a declaration.

Revision ID: 615
Revises: 614
Create Date: 2026-09-06

One nullable self-referential column plus its index. No data change: every
existing row keeps its cron or interval, and a NULL here means exactly
what it meant before this migration — "this job runs on a clock".

The column exists because at least eight ordering dependencies in this
codebase are currently expressed as a **gap between two cron minute
fields**, with the intent recorded only in prose:

* ``ai_spending.cost_drift_check`` — "每天 07:00 UTC(在
  ai_spending.period_rollup_refresh 06:30 之后)"
* ``feed.notify_celebrants_and_leads`` — "紧随扫描任务"
* ``hr.leave_insight_snapshot`` — "排在 hr.leave_account_sync 之后运行"
* ``auditor.compliance_posture_summary`` — "排在预编译(2 号 03:20)之后:
  先有证据,再谈覆盖率"
* ``offboarding.account_assurance_rosters`` — "每日 04:10 UTC(排在
  identity_sync.leaver_assurance 03:47 之后)"
* ``cmdb.cost_recon.aliyun_bq`` — "排在 cmdb.cost_facts_monthclose … 之后两天"
* ``bailian.usage_sync`` — "…先跑完,所以调度顺序排在它后面"
* plus the per-account cloud sweeps, whose whole nightly ordering is
  minute arithmetic.

A minute gap is a promise the platform cannot keep. The moment the
predecessor overruns it — and production has ``cmdb.roles`` at 1 055 s
average against a 9-minute gap — the successor runs on stale data and
nothing anywhere says so. Worse, the concurrency gate makes overrun the
*normal* case in the pre-dawn window: a job can wait 20+ minutes for a
slot before it even starts.

``ON DELETE SET NULL``, deliberately, rather than CASCADE: deleting a
parent should orphan the child (it stops firing and surfaces in the
hygiene census) not delete it. Losing a job row is never the right answer
to losing its predecessor.

Downgrade drops the index, the FK and the column, and converts any row
left on ``trigger_type='chained'`` back to a disabled cron so the
worker's boot guard does not meet a trigger type it cannot build. A
chained row without the column is unschedulable, and leaving it enabled
would make it silently never run.
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op
from app.core.migration_helpers import id_type_for

revision = "615"
down_revision = "614"
branch_labels = None
depends_on = None

TABLE = "scheduled_jobs"
COLUMN = "chain_after_job_id"
INDEX = "ix_scheduled_jobs_chain_after"
FK = "fk_scheduled_jobs_chain_after"


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


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


def _fk_exists(bind, table: str, name: str) -> bool:
    return bool(
        bind.execute(
            sa.text(
                "SELECT 1 FROM information_schema.TABLE_CONSTRAINTS "
                "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t "
                "AND CONSTRAINT_NAME = :n AND CONSTRAINT_TYPE = 'FOREIGN KEY' "
                "LIMIT 1",
            ),
            {"t": table, "n": name},
        ).scalar(),
    )


def _fk_names_on_column(bind, table: str, column: str) -> list[str]:
    """Every FK constraint sitting on ``(table, column)``, whatever its name.

    The downgrade has to drop by *column*, not by the name this migration
    happens to use. MySQL refuses to drop an index a foreign key still
    needs, and the same column can carry a differently-named FK — most
    obviously on a schema built from ORM metadata via ``create_all``,
    where SQLAlchemy auto-names it. Looking only for our own name left
    the ORM-named constraint in place and the index undroppable, which is
    exactly how the test for this downgrade failed the first time.
    """
    rows = bind.execute(
        sa.text(
            "SELECT k.CONSTRAINT_NAME FROM information_schema.KEY_COLUMN_USAGE k "
            "JOIN information_schema.TABLE_CONSTRAINTS c "
            "  ON c.CONSTRAINT_SCHEMA = k.CONSTRAINT_SCHEMA "
            " AND c.CONSTRAINT_NAME = k.CONSTRAINT_NAME "
            " AND c.TABLE_NAME = k.TABLE_NAME "
            "WHERE k.TABLE_SCHEMA = DATABASE() AND k.TABLE_NAME = :t "
            "  AND k.COLUMN_NAME = :c AND c.CONSTRAINT_TYPE = 'FOREIGN KEY'",
        ),
        {"t": table, "c": column},
    ).fetchall()
    return [row[0] for row in rows]


def upgrade() -> None:
    bind = op.get_bind()
    if not bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.TABLES WHERE TABLE_SCHEMA = "
            "DATABASE() AND TABLE_NAME = :t LIMIT 1",
        ),
        {"t": TABLE},
    ).scalar():
        return

    if not _column_exists(bind, TABLE, COLUMN):
        # ``id_type_for`` reads information_schema for the parent's real
        # DATA_TYPE *and collation*. Both halves matter: prod TiDB has
        # ``scheduled_jobs.id`` as CHAR(36) utf8mb4_unicode_ci while CI
        # MySQL can differ, and TiDB rejects a FK on either mismatch with
        # error 3780 — which is exactly how migration 612 wedged the prod
        # chain the day before this one.
        op.add_column(
            TABLE,
            sa.Column(COLUMN, id_type_for(TABLE), nullable=True),
        )
    if not _index_exists(bind, TABLE, INDEX):
        op.create_index(INDEX, TABLE, [COLUMN])
    if not _fk_exists(bind, TABLE, FK):
        op.create_foreign_key(
            FK, TABLE, TABLE, [COLUMN], ["id"], ondelete="SET NULL",
        )


def downgrade() -> None:
    bind = op.get_bind()
    if not _column_exists(bind, TABLE, COLUMN):
        return
    # A chained row is unschedulable without the column — the worker
    # would find a trigger type it cannot build. Park those rows as
    # disabled crons so the downgrade cannot leave an enabled row that
    # silently never fires.
    bind.execute(
        sa.text(
            f"UPDATE {TABLE} SET trigger_type = 'cron', "
            "    cron_expr = COALESCE(cron_expr, '0 4 * * *'), "
            "    enabled = 0, updated_at = NOW() "
            "WHERE trigger_type = 'chained'",
        ),
    )
    for fk_name in _fk_names_on_column(bind, TABLE, COLUMN):
        op.drop_constraint(fk_name, TABLE, type_="foreignkey")
    if _index_exists(bind, TABLE, INDEX):
        op.drop_index(INDEX, table_name=TABLE)
    op.drop_column(TABLE, COLUMN)