"""迁移 614 —— 把「被部署打断」从 failed 改判成 interrupted.

这是一批会改动**既有历史行**的数据迁移,所以谓词的窄度就是它的全部安全性。
盯三件事:

1. **只命中 reaper 自己写的那句话。** 一条被人诊断过、写了真实原因的 failed 行,
   绝不能被扫进这个桶 —— 那才是真正的「改写历史」。谓词只认 reaper 机器生成的
   前缀。
2. **幂等。** 迁移会在多个环境跑;重复执行不能把标记追加两遍,也不能把已经改判
   过的行再算一次。
3. **可逆且不越界。** downgrade 只还原带标记的行(即本迁移改过的那些),不去碰
   上线后由新代码正常写出来的 interrupted 行 —— 那些不是这条迁移的产物。
"""
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

_MIGRATION = (
    pathlib.Path(__file__).resolve().parents[2]
    / "alembic" / "versions" / "614_reclassify_interrupted_runs.py"
)

JOBS = "scheduled_jobs"
RUNS = "scheduled_job_runs"


def _load():
    spec = importlib.util.spec_from_file_location("m614", _MIGRATION)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


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


@pytest.fixture
def conn():
    engine = sa.create_engine(_TEST_SYNC_URL)
    with engine.begin() as c:
        job_id = str(uuid.uuid4())
        c.execute(
            sa.text(
                f"INSERT INTO {JOBS} (id, name, display_name, handler_key, "
                " trigger_type, interval_seconds, timezone, enabled, "
                " run_count, failure_count, created_at, updated_at) "
                "VALUES (:id, 'probe614.job', 'probe614', 'probe614.job', "
                " 'interval', 3600, 'UTC', 1, 5, 3, NOW(), NOW())",
            ),
            {"id": job_id},
        )
        c.execute(sa.text("SET @probe614 := :j"), {"j": job_id})
        try:
            yield c
        finally:
            c.execute(
                sa.text(f"DELETE FROM {RUNS} WHERE job_id = :j"), {"j": job_id},
            )
            c.execute(
                sa.text(f"DELETE FROM {JOBS} WHERE id = :j"), {"j": job_id},
            )
    engine.dispose()


def _job_id(conn) -> str:
    return conn.execute(sa.text(f"SELECT id FROM {JOBS} WHERE name = 'probe614.job'")).scalar()


def _seed_run(conn, *, status: str, error: str | None) -> str:
    run_id = str(uuid.uuid4())
    conn.execute(
        sa.text(
            f"INSERT INTO {RUNS} (id, job_id, started_at, finished_at, "
            " duration_ms, status, triggered_by, error_message) "
            "VALUES (:id, :j, NOW(), NOW(), 10, :s, 'scheduler', :e)",
        ),
        {"id": run_id, "j": _job_id(conn), "s": status, "e": error},
    )
    return run_id


def _status_of(conn, run_id: str) -> str:
    return conn.execute(
        sa.text(f"SELECT status FROM {RUNS} WHERE id = :i"), {"i": run_id},
    ).scalar()


def _error_of(conn, run_id: str) -> str | None:
    return conn.execute(
        sa.text(f"SELECT error_message FROM {RUNS} WHERE id = :i"), {"i": run_id},
    ).scalar()


def test_reclassifies_only_the_reaper_message(conn):
    module = _load()
    reaper = _seed_run(
        conn, status="failed",
        error=(
            "auto-fail: stale running row swept after 6h "
            "(worker likely died mid-run)"
        ),
    )
    # A real, human-legible failure. Sweeping this one would be the
    # actual "rewrote history" mistake the repo rule protects against.
    genuine = _seed_run(
        conn, status="failed",
        error="RuntimeError: Tencent ListUsers returned a user with no name",
    )
    # And a failure that merely *mentions* the words — the predicate is a
    # prefix on the reaper's own wording, not a substring search.
    mentions = _seed_run(
        conn, status="failed",
        error="operator note: this looks like a stale running row swept after 6h",
    )
    healthy = _seed_run(conn, status="ok", error=None)

    _run(conn, "upgrade")

    assert _status_of(conn, reaper) == "interrupted"
    assert module.MARKER in _error_of(conn, reaper)
    assert "worker likely died mid-run" in _error_of(conn, reaper)
    assert _status_of(conn, genuine) == "failed"
    assert _status_of(conn, mentions) == "failed"
    assert _status_of(conn, healthy) == "ok"


def test_is_idempotent(conn):
    module = _load()
    reaper = _seed_run(
        conn, status="failed",
        error="auto-fail: stale running row swept after 6h (worker likely died mid-run)",
    )
    _run(conn, "upgrade")
    _run(conn, "upgrade")
    assert _error_of(conn, reaper).count(module.MARKER) == 1


def test_counters_are_untouched(conn):
    """The bulk UPDATE that created these rows never incremented
    ``run_count`` / ``failure_count``, so the counters have always
    excluded them. Moving the status must not move a counter, or the
    Prometheus identity ``ok = run_count - failure_count`` shifts."""
    _seed_run(
        conn, status="failed",
        error="auto-fail: stale running row swept after 6h (worker likely died mid-run)",
    )
    before = conn.execute(
        sa.text(f"SELECT run_count, failure_count FROM {JOBS} WHERE name = 'probe614.job'"),
    ).fetchone()
    _run(conn, "upgrade")
    after = conn.execute(
        sa.text(f"SELECT run_count, failure_count FROM {JOBS} WHERE name = 'probe614.job'"),
    ).fetchone()
    assert tuple(before) == tuple(after) == (5, 3)


def test_downgrade_restores_only_rows_this_migration_wrote(conn):
    reaper = _seed_run(
        conn, status="failed",
        error="auto-fail: stale running row swept after 6h (worker likely died mid-run)",
    )
    _run(conn, "upgrade")
    # An interrupted row written by the NEW executor after the deploy —
    # not this migration's product, so downgrade must leave it alone.
    native = _seed_run(
        conn, status="interrupted",
        error="worker shutdown drain cancelled the run",
    )

    _run(conn, "downgrade")

    assert _status_of(conn, reaper) == "failed"
    assert "reclassified" not in (_error_of(conn, reaper) or "")
    assert _status_of(conn, native) == "interrupted"