"""迁移 634 —— 收尾任务那一行 cron,以及它**不带 config**。

三件事,每一件都对应一个这个仓库真实踩过的坑:

1. **cron 行必须真的存在,而且 `handler_key` 指向注册过的 handler。** 指向一个
   不存在的键时 worker 只打一行 warning 然后跳过 —— 任务永远不跑,而且没有任何
   失败记录。
2. **`config` 必须是空的。** 2026-09-08 刚踩过反过来的那一半:播种进 config 的
   一把钥匙赢过了平台设置,于是设置页上的改动被静默无视一周(#3520)。这一行
   两个阈值只从平台设置读,所以 config 里**任何**键都是一个未来的谎。
3. **`id` 必须是带横线的 uuid**:32 位无横线的 id 会让 GUID TypeDecorator 永远
   匹配不上,handler 在 run 包装器外面就抛,`scheduled_job_runs` 一行都不写。
   本仓踩过四次。
"""
from __future__ import annotations

import importlib.util
import json
import pathlib
import re

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" / "634_assistant_room_lifecycle.py"
)

JOB_NAME = "assistant.ticket_chat_sweep"
KEYS = (
    "assistant.ticket_chat_retention_days",
    "assistant.ticket_chat_idle_nudge_days",
)
_DASHED_UUID = re.compile(
    r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
)


def _load():
    spec = importlib.util.spec_from_file_location("m634", _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:
        c.execute(sa.text("DELETE FROM scheduled_jobs WHERE name = :n"), {"n": JOB_NAME})
        c.execute(
            sa.text(
                "DELETE FROM platform_settings WHERE `key` IN :keys",
            ).bindparams(sa.bindparam("keys", expanding=True)),
            {"keys": list(KEYS)},
        )
        try:
            yield c
        finally:
            c.execute(
                sa.text("DELETE FROM scheduled_jobs WHERE name = :n"),
                {"n": JOB_NAME},
            )
            c.execute(
                sa.text(
                    "DELETE FROM platform_settings WHERE `key` IN :keys",
                ).bindparams(sa.bindparam("keys", expanding=True)),
                {"keys": list(KEYS)},
            )
    engine.dispose()


def _job(conn):
    return conn.execute(
        sa.text(
            "SELECT id, handler_key, cron_expr, timezone, enabled, config "
            "FROM scheduled_jobs WHERE name = :n",
        ),
        {"n": JOB_NAME},
    ).mappings().all()


def test_the_cron_row_lands_and_points_at_a_registered_handler(conn):
    _run(conn)
    rows = _job(conn)
    assert len(rows) == 1
    row = rows[0]
    assert row["enabled"] == 1
    assert row["timezone"] == "Asia/Shanghai"
    assert _DASHED_UUID.match(str(row["id"])), (
        "32 位无横线的 id 会让 GUID TypeDecorator 永远匹配不上"
    )

    # handler_key 必须在注册表里 —— 指向一个不存在的键时 worker 只打一行
    # warning 然后跳过,任务永远不跑而且没有失败记录。
    import app.scheduled_jobs.assistant_room_handlers  # noqa: F401
    from app.scheduled_jobs.registry import get_handler

    assert get_handler(row["handler_key"]) is not None


def test_the_cron_row_carries_no_config(conn):
    """两个阈值只从平台设置读,所以 config 里任何键都是一个未来的谎。

    #3520:播种进 config 的一把钥匙赢过了平台设置,设置页上的改动被静默无视
    一周。这条断言是那次事故的直接产物。
    """
    _run(conn)
    config = _job(conn)[0]["config"]
    parsed = json.loads(config) if isinstance(config, str) else (config or {})
    assert parsed == {}, f"config 必须为空,实际是 {parsed!r}"


def test_the_two_thresholds_become_visible_rows(conn):
    _run(conn)
    rows = {
        r["key"]: json.loads(r["value"]) if isinstance(r["value"], str)
        else r["value"]
        for r in conn.execute(
            sa.text(
                "SELECT `key`, `value`, `description` FROM platform_settings "
                "WHERE `key` IN :keys",
            ).bindparams(sa.bindparam("keys", expanding=True)),
            {"keys": list(KEYS)},
        ).mappings().all()
    }
    assert rows == {
        "assistant.ticket_chat_retention_days": 7,
        "assistant.ticket_chat_idle_nudge_days": 14,
    }


def test_an_administrators_retention_choice_survives(conn):
    """管理员把保留期改成 30 了 —— 一次发布不能把它改回 7。"""
    conn.execute(
        sa.text(
            "INSERT INTO platform_settings (`key`, `value`, `description`) "
            "VALUES (:k, :v, 'managed by hand')",
        ),
        {"k": KEYS[0], "v": json.dumps(30)},
    )
    _run(conn)
    value = conn.execute(
        sa.text("SELECT `value` FROM platform_settings WHERE `key` = :k"),
        {"k": KEYS[0]},
    ).scalar()
    parsed = json.loads(value) if isinstance(value, str) else value
    assert parsed == 30


def test_running_it_twice_leaves_one_row(conn):
    _run(conn)
    _run(conn)
    assert len(_job(conn)) == 1