"""迁移 615 / 616 —— 依赖列本身,以及第一批被转换的顺序对。

615 是纯 schema(一列 + 索引 + 自引用外键),616 是数据(播种推进对账 cron +
转换六条边)。两者放在一个测试文件里,因为它们的**降级顺序**是耦合的:
616 的 downgrade 必须精确还原原来的 cron,615 的 downgrade 只是一张兜底网。

盯五件事:

1. **只转换仍然带着代码播种值的行。** 管理员调过节奏的行代表一个这次迁移
   不知道的判断,静默盖掉比留着旧默认更糟(沿用 418 / 613 的纪律)。
2. **精确还原。** 616 的 downgrade 必须把每条边各自原来的 cron 写回去 ——
   615 那条通用兜底(把 chained 行停成 `0 4 * * *`)会丢掉真实值,所以它
   只该接住「上线后管理员手工连的链」,不该接住这六条。
3. **幂等。** 重复 upgrade 不能重复插 cron 行,也不能把已经转过的行再动一次。
4. **播种的 cron 行必须指向一个已注册的 handler** —— worker 的开机守卫对
   「启用但 handler 不存在」是硬崩,所以播种/代码不一致会在下一次发布把
   整个 worker 带下去。
5. **每一条边的两端都必须真实存在于 handler 注册表里**,否则这条迁移在
   描述一个不存在的顺序。
"""
from __future__ import annotations

import importlib.util
import pathlib
import uuid

import pytest
import sqlalchemy as sa
from alembic.migration import MigrationContext
from alembic.operations import Operations

from ..conftest import _TEST_SYNC_URL

_VERSIONS = pathlib.Path(__file__).resolve().parents[2] / "alembic" / "versions"
JOBS = "scheduled_jobs"


def _load(stem: str):
    path = next(_VERSIONS.glob(f"{stem}_*.py"))
    spec = importlib.util.spec_from_file_location(f"m{stem}", path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _run(conn, stem: str, direction: str = "upgrade") -> None:
    ctx = MigrationContext.configure(conn)
    with Operations.context(ctx):
        getattr(_load(stem), direction)()


@pytest.fixture
def conn():
    engine = sa.create_engine(_TEST_SYNC_URL)
    with engine.begin() as c:
        try:
            yield c
        finally:
            c.execute(
                sa.text(f"DELETE FROM {JOBS} WHERE name LIKE 'probe616.%'"),
            )
            c.execute(
                sa.text(
                    f"DELETE FROM {JOBS} WHERE name = 'scheduler.chain_advance'",
                ),
            )
            for name in (
                "ai_spending.period_rollup_refresh",
                "ai_spending.cost_drift_check",
                "hr.leave_account_sync",
                "hr.leave_insight_snapshot",
            ):
                c.execute(
                    sa.text(f"DELETE FROM {JOBS} WHERE name = :n"), {"n": name},
                )
    engine.dispose()


def _seed(conn, name: str, cron: str) -> str:
    jid = str(uuid.uuid4())
    conn.execute(
        sa.text(
            f"INSERT INTO {JOBS} (id, name, display_name, handler_key, "
            " trigger_type, cron_expr, timezone, enabled, run_count, "
            " failure_count, created_at, updated_at) "
            "VALUES (:id, :n, :n, :n, 'cron', :c, 'UTC', 1, 0, 0, "
            " NOW(), NOW())",
        ),
        {"id": jid, "n": name, "c": cron},
    )
    return jid


def _row(conn, name: str):
    return conn.execute(
        sa.text(
            f"SELECT trigger_type, cron_expr, chain_after_job_id "
            f"FROM {JOBS} WHERE name = :n",
        ),
        {"n": name},
    ).fetchone()


def test_column_index_and_fk_are_idempotent(conn):
    """615 runs on every environment and may meet a partially-applied
    state; each of the three DDL steps is guarded independently."""
    _run(conn, "615", "upgrade")
    _run(conn, "615", "upgrade")
    cols = conn.execute(
        sa.text(
            "SELECT COUNT(*) FROM information_schema.COLUMNS WHERE "
            "TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t AND "
            "COLUMN_NAME = 'chain_after_job_id'",
        ),
        {"t": JOBS},
    ).scalar()
    assert cols == 1


def test_converts_only_rows_still_on_the_seeded_cron(conn):
    _run(conn, "615", "upgrade")
    _seed(conn, "ai_spending.period_rollup_refresh", "30 6 * * *")
    _seed(conn, "ai_spending.cost_drift_check", "0 7 * * *")
    _seed(conn, "hr.leave_account_sync", "30 6 * * *")
    # An operator already retuned this one. The migration knows nothing
    # about why, so it must leave it exactly as found.
    _seed(conn, "hr.leave_insight_snapshot", "45 8 * * *")

    _run(conn, "616", "upgrade")

    drift = _row(conn, "ai_spending.cost_drift_check")
    assert drift.trigger_type == "chained"
    assert drift.cron_expr is None
    assert drift.chain_after_job_id is not None

    untouched = _row(conn, "hr.leave_insight_snapshot")
    assert untouched.trigger_type == "cron"
    assert untouched.cron_expr == "45 8 * * *"
    assert untouched.chain_after_job_id is None


def test_edge_is_skipped_when_the_predecessor_is_absent(conn):
    """If the predecessor is not seeded in this deployment, the ordering
    it was supposed to guarantee does not exist here either — leaving the
    successor on its own cron is the correct answer, not chaining it to
    nothing."""
    _run(conn, "615", "upgrade")
    _seed(conn, "ai_spending.cost_drift_check", "0 7 * * *")
    _run(conn, "616", "upgrade")
    drift = _row(conn, "ai_spending.cost_drift_check")
    assert drift.trigger_type == "cron"
    assert drift.cron_expr == "0 7 * * *"


def test_seeding_the_reconciler_is_idempotent(conn):
    _run(conn, "615", "upgrade")
    _run(conn, "616", "upgrade")
    _run(conn, "616", "upgrade")
    count = conn.execute(
        sa.text(f"SELECT COUNT(*) FROM {JOBS} WHERE name = :n"),
        {"n": "scheduler.chain_advance"},
    ).scalar()
    assert count == 1


def test_reconciler_row_points_at_a_registered_handler(conn):
    """The worker's boot guard hard-crashes on an enabled row whose
    handler is unknown — a seed/code mismatch takes the whole worker down
    on the next deploy rather than failing quietly."""
    from app.scheduled_jobs.registry import get_handler

    _run(conn, "615", "upgrade")
    _run(conn, "616", "upgrade")
    handler_key = conn.execute(
        sa.text(f"SELECT handler_key FROM {JOBS} WHERE name = :n"),
        {"n": "scheduler.chain_advance"},
    ).scalar()
    assert get_handler(handler_key) is not None


def test_every_declared_edge_names_two_real_handlers():
    """A migration that chains a job which does not exist is describing
    an ordering nobody has. Checked against the live registry rather than
    a second list, so a rename cannot leave this passing."""
    from app.scheduled_jobs.registry import get_handler

    module = _load("616")
    for successor, predecessor, _cron in module.CHAIN_EDGES:
        assert get_handler(successor) is not None, successor
        assert get_handler(predecessor) is not None, predecessor


def test_downgrade_restores_each_original_cron(conn):
    """615's generic fallback parks any leftover chained row on a
    placeholder cron; that must never be what these six get, because it
    would silently change their schedule. 616 restores each one exactly.
    """
    _run(conn, "615", "upgrade")
    _seed(conn, "ai_spending.period_rollup_refresh", "30 6 * * *")
    _seed(conn, "ai_spending.cost_drift_check", "0 7 * * *")
    _run(conn, "616", "upgrade")
    assert _row(conn, "ai_spending.cost_drift_check").trigger_type == "chained"

    _run(conn, "616", "downgrade")
    restored = _row(conn, "ai_spending.cost_drift_check")
    assert restored.trigger_type == "cron"
    assert restored.cron_expr == "0 7 * * *"
    assert restored.chain_after_job_id is None
    assert conn.execute(
        sa.text(f"SELECT COUNT(*) FROM {JOBS} WHERE name = :n"),
        {"n": "scheduler.chain_advance"},
    ).scalar() == 0


def test_615_downgrade_parks_hand_built_chains_disabled(conn):
    """A chained row is unschedulable once the column is gone — the
    worker would meet a trigger type it cannot build. Parking them
    disabled means a downgrade cannot leave an enabled row that silently
    never fires."""
    _run(conn, "615", "upgrade")
    parent = _seed(conn, "probe616.parent", "0 3 * * *")
    _seed(conn, "probe616.child", "0 4 * * *")
    conn.execute(
        sa.text(
            f"UPDATE {JOBS} SET trigger_type = 'chained', cron_expr = NULL, "
            "chain_after_job_id = :p WHERE name = 'probe616.child'",
        ),
        {"p": parent},
    )

    _run(conn, "615", "downgrade")

    row = conn.execute(
        sa.text(
            f"SELECT trigger_type, enabled FROM {JOBS} "
            "WHERE name = 'probe616.child'",
        ),
    ).fetchone()
    assert row.trigger_type == "cron"
    assert int(row.enabled) == 0
    # Put the column back for the rest of the shard.
    _run(conn, "615", "upgrade")