"""迁移 633 —— 已经存在的工单群必须继续被认作工单群。

回填是这条迁移里唯一会**伤到生产**的部分。不回填的后果不是「少了几行历史数据」:
`group_chat.resolve_group_context` 从这一版起只认 `assistant_ticket_chats`,所以
没搬过来的房间会掉进「任意群」那条**默认关着**的闸 —— 屋里的人第二天开始
得不到任何回应,而群、工单、负责人都还在那儿。

三件事:

1. **搬过来了**,并且带着当时的提问者与时间。
2. **一张工单只搬一行**。同一张工单可能被转过两次(第二次拉了新群),而
   `ticket_id` 是唯一约束 —— 取最新那一行,那才是现在还在用的房间。
3. **没群的 handoff 行不搬**。降级路径(没指派负责人 / 建群失败)上
   `feishu_chat_id` 为空,搬过来会造出一个指向空房间的注册行。
"""
from __future__ import annotations

import importlib.util
import pathlib
import uuid
from datetime import UTC, datetime, timedelta

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" / "633_assistant_ticket_chats.py"
)

TABLE = "assistant_ticket_chats"


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


def _backfill(conn) -> None:
    """只跑回填那一半。

    建表那一半在测试库上早已由 ``Base.metadata.create_all`` 做过(表已存在),
    而迁移的第一行守卫正是「表在就返回」——所以这里直接把 SQL 再跑一遍,盯的是
    回填本身的行为。
    """
    module = _load()
    ctx = MigrationContext.configure(conn)
    with Operations.context(ctx):
        src = _MIGRATION.read_text()
    start = src.index('    bind.execute(sa.text(\n        f"""\n        INSERT INTO')
    end = src.index('    ))', start)
    body = src[start:end].split('f"""', 1)[1].rsplit('"""', 1)[0]
    conn.execute(sa.text(body.replace("{TABLE}", module.TABLE)))


@pytest.fixture
def conn():
    engine = sa.create_engine(_TEST_SYNC_URL)
    with engine.begin() as c:
        c.execute(sa.text(f"DELETE FROM {TABLE}"))
        c.execute(sa.text("DELETE FROM assistant_handoffs"))
        try:
            yield c
        finally:
            c.execute(sa.text(f"DELETE FROM {TABLE}"))
            c.execute(sa.text("DELETE FROM assistant_handoffs"))
    engine.dispose()


def _person(conn) -> str:
    pid = str(uuid.uuid4())
    conn.execute(
        sa.text(
            "INSERT INTO identity_persons "
            "(id, source_system, hris_person_id, work_email, display_name, "
            " employment_status, is_china_employee) "
            "VALUES (:id, 'successfactors', :hris, :email, '员工', 'active', 0)",
        ),
        {
            "id": pid,
            "hris": f"sf-{uuid.uuid4().hex[:8]}",
            "email": f"m633-{uuid.uuid4().hex[:8]}@test.com",
        },
    )
    return pid


def _handoff(
    conn, *, person_id: str, chat_id: str | None, ticket_id: str | None,
    created_at: datetime,
) -> str:
    """一行 handoff 台账。``message_id`` 上有唯一约束,所以每行一个新消息。"""
    conv_id = str(uuid.uuid4())
    msg_id = str(uuid.uuid4())
    conn.execute(
        sa.text(
            "INSERT INTO assistant_conversations "
            "(id, person_id, channel, chat_id, turn_count) "
            "VALUES (:id, :pid, 'feishu', 'oc-dm', 0)",
        ),
        {"id": conv_id, "pid": person_id},
    )
    conn.execute(
        sa.text(
            "INSERT INTO assistant_messages "
            "(id, conversation_id, seq, role, content) "
            "VALUES (:id, :conv, :seq, 'assistant', '答不上来')",
        ),
        {"id": msg_id, "conv": conv_id, "seq": abs(hash(msg_id)) % 1000},
    )
    row_id = str(uuid.uuid4())
    conn.execute(
        sa.text(
            "INSERT INTO assistant_handoffs "
            "(id, message_id, person_id, feishu_chat_id, ticket_id, created_at) "
            "VALUES (:id, :msg, :pid, :chat, :ticket, :created)",
        ),
        {
            "id": row_id, "msg": msg_id, "pid": person_id,
            "chat": chat_id, "ticket": ticket_id, "created": created_at,
        },
    )
    return row_id


def _rows(conn) -> list[dict]:
    return [
        dict(r) for r in conn.execute(
            sa.text(
                f"SELECT ticket_id, feishu_chat_id, state, opened_reason, "
                f"opened_by_person_id FROM {TABLE}",
            ),
        ).mappings().all()
    ]


def test_an_existing_room_keeps_being_a_ticket_room(conn):
    """不回填的房间会掉进「任意群」那条默认关着的闸 —— 屋里的人失去回应。"""
    person = _person(conn)
    ticket = str(uuid.uuid4())
    _handoff(
        conn, person_id=person, chat_id="oc_live", ticket_id=ticket,
        created_at=datetime.now(UTC),
    )
    _backfill(conn)

    rows = _rows(conn)
    assert len(rows) == 1
    assert rows[0]["ticket_id"] == ticket
    assert rows[0]["feishu_chat_id"] == "oc_live"
    assert rows[0]["state"] == "open"
    assert rows[0]["opened_reason"] == "handoff"
    assert rows[0]["opened_by_person_id"] == person


def test_a_ticket_transferred_twice_lands_exactly_once(conn):
    """``ticket_id`` 唯一 —— 取最新那一行,那才是现在还在用的房间。"""
    person = _person(conn)
    ticket = str(uuid.uuid4())
    now = datetime.now(UTC)
    _handoff(
        conn, person_id=person, chat_id="oc_old", ticket_id=ticket,
        created_at=now - timedelta(days=3),
    )
    _handoff(
        conn, person_id=person, chat_id="oc_new", ticket_id=ticket,
        created_at=now,
    )
    _backfill(conn)

    rows = _rows(conn)
    assert len(rows) == 1
    assert rows[0]["feishu_chat_id"] == "oc_new"


def test_a_degraded_handoff_is_not_backfilled(conn):
    """降级路径没有群 —— 搬过来会造出一个指向空房间的注册行。"""
    person = _person(conn)
    _handoff(
        conn, person_id=person, chat_id=None, ticket_id=str(uuid.uuid4()),
        created_at=datetime.now(UTC),
    )
    _handoff(
        conn, person_id=person, chat_id="oc_orphan", ticket_id=None,
        created_at=datetime.now(UTC),
    )
    _backfill(conn)
    assert _rows(conn) == []