"""End-to-end service tests for BOXIM takeover timing and human priority."""

from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock, patch

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from database import Base
from models import Avatar, TakeoverCursor, TakeoverMessage, TakeoverReplyTask, User
from services.boxim_client import BoxIMError
from services.takeover_service import TakeoverService, _plain_text_reply


class Clock:
    def __init__(self):
        self.value = datetime(2026, 8, 19, 10, 0, 0)

    def now(self):
        return self.value

    def advance(self, seconds: int):
        self.value += timedelta(seconds=seconds)

    def millis(self):
        return int(self.value.replace(tzinfo=timezone.utc).timestamp() * 1000)


class FakeBoxIM:
    def __init__(self):
        self.messages = []
        self.sent = []
        self.read_receipts = []

    async def exchange_access_token(self, huihui_token):
        assert huihui_token == "prod-huihui-token"
        return {"accessToken": "box-token", "accessTokenExpiresIn": 3600}

    async def get_self(self, access_token):
        assert access_token == "box-token"
        return {"id": 100}

    async def fetch_private_messages(self, access_token, min_id="0"):
        assert access_token == "box-token"
        return [item.copy() for item in self.messages if int(item["id"]) > int(min_id)]

    async def mark_private_messages_read(self, access_token, friend_id, message_id):
        assert access_token == "box-token"
        self.read_receipts.append(
            {"friendId": str(friend_id), "messageId": str(message_id)}
        )

    async def send_private_message(self, access_token, peer_id, content, *, local_id=None):
        self.sent.append({"peerId": str(peer_id), "content": content, "localId": str(local_id)})
        return {"id": 900 + len(self.sent), "localId": int(local_id)}


@pytest.fixture
def service_context():
    engine = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    session_factory = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
    Base.metadata.create_all(engine)
    db = session_factory()
    user = User(
        id="owner-local",
        huihui_user_id="owner-huihui",
        huihui_token="prod-huihui-token",
        app_token="app-token",
    )
    avatar = Avatar(
        id="avatar-1",
        owner_id=user.huihui_user_id,
        name="分身",
        status="active",
        config={"authorizationPermissions": ["chat", "takeover"]},
    )
    db.add_all([user, avatar])
    db.commit()
    db.close()

    clock = Clock()
    boxim = FakeBoxIM()
    service = TakeoverService(session_factory, boxim, now=clock.now)
    return session_factory, service, boxim, clock


@pytest.mark.asyncio
async def test_first_sync_primes_cursor_without_replying_to_history(service_context):
    session_factory, service, boxim, clock = service_context
    boxim.messages = [
        {"id": 10, "localId": 1, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "旧消息"}
    ]

    with patch("routers.chat._resolve_reply", return_value={"answer": "不应发送"}):
        await service.poll_and_process_messages()

    db = session_factory()
    try:
        cursor = db.query(TakeoverCursor).one()
        assert cursor.initialized is True
        assert cursor.last_message_id == "10"
        assert db.query(TakeoverMessage).count() == 1
        assert db.query(TakeoverReplyTask).count() == 0
        assert boxim.sent == []
        assert boxim.read_receipts == [{"friendId": "200", "messageId": "10"}]
    finally:
        db.close()


@pytest.mark.asyncio
async def test_incoming_message_is_prepared_then_sent_at_three_seconds(service_context):
    session_factory, service, boxim, clock = service_context
    await service.poll_and_process_messages()
    boxim.messages.append(
        {"id": 11, "localId": 2, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "你好"}
    )

    with patch("routers.chat._resolve_reply", return_value={"answer": "**你好**\n\n很高兴见到你"}):
        await service.poll_and_process_messages()
    assert boxim.sent == []
    assert boxim.read_receipts == [{"friendId": "200", "messageId": "11"}]

    clock.advance(2)
    await service.poll_and_process_messages()
    assert boxim.sent == []

    clock.advance(1)
    await service.poll_and_process_messages()
    assert boxim.sent == [{"peerId": "200", "content": "你好\n很高兴见到你", "localId": boxim.sent[0]["localId"]}]

    db = session_factory()
    try:
        task = db.query(TakeoverReplyTask).one()
        assert task.status == "sent"
        assert task.sent_at == clock.now()
    finally:
        db.close()


@pytest.mark.asyncio
async def test_read_receipt_failure_does_not_advance_cursor(service_context):
    session_factory, service, boxim, clock = service_context
    await service.poll_and_process_messages()
    boxim.messages.append(
        {"id": 12, "localId": 3, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "未读消息"}
    )
    boxim.mark_private_messages_read = AsyncMock(side_effect=BoxIMError("回执失败"))

    with patch("routers.chat._resolve_reply", return_value={"answer": "稍后回复"}):
        await service.poll_and_process_messages()

    db = session_factory()
    try:
        cursor = db.query(TakeoverCursor).one()
        assert cursor.last_message_id == "0"
        assert db.query(TakeoverMessage).count() == 0
        assert db.query(TakeoverReplyTask).count() == 0
    finally:
        db.close()

    boxim.mark_private_messages_read = AsyncMock(return_value=None)
    with patch("routers.chat._resolve_reply", return_value={"answer": "稍后回复"}):
        await service.poll_and_process_messages()

    db = session_factory()
    try:
        assert db.query(TakeoverCursor).one().last_message_id == "12"
        assert db.query(TakeoverMessage).count() == 1
        assert db.query(TakeoverReplyTask).count() == 1
    finally:
        db.close()


@pytest.mark.asyncio
async def test_owner_message_cancels_pending_reply(service_context):
    session_factory, service, boxim, clock = service_context
    await service.poll_and_process_messages()
    boxim.messages.append(
        {"id": 21, "localId": 3, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "在吗"}
    )
    with patch("routers.chat._resolve_reply", return_value={"answer": "在的"}):
        await service.poll_and_process_messages()

    clock.advance(2)
    boxim.messages.append(
        {"id": 22, "localId": 4, "sendId": 100, "recvId": 200, "sendTime": clock.millis(), "type": 0, "content": "我来回复"}
    )
    await service.poll_and_process_messages()
    clock.advance(2)
    await service.poll_and_process_messages()

    db = session_factory()
    try:
        task = db.query(TakeoverReplyTask).filter(TakeoverReplyTask.trigger_message_id == "21").one()
        assert task.status == "cancelled"
        assert task.cancel_reason == "owner_replied"
        assert boxim.sent == []
    finally:
        db.close()


@pytest.mark.asyncio
async def test_quick_successive_messages_are_coalesced_into_one_reply(service_context):
    session_factory, service, boxim, clock = service_context
    await service.poll_and_process_messages()
    boxim.messages.append(
        {"id": 31, "localId": 5, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "第一句"}
    )
    with patch("routers.chat._resolve_reply", return_value={"answer": "第一版"}):
        await service.poll_and_process_messages()

    clock.advance(1)
    boxim.messages.append(
        {"id": 32, "localId": 6, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "第二句"}
    )
    with patch("routers.chat._resolve_reply", return_value={"answer": "合并回复"}) as resolver:
        await service.poll_and_process_messages()
    assert resolver.call_args.args[2] == "第一句\n第二句"

    clock.advance(3)
    await service.poll_and_process_messages()
    assert [item["content"] for item in boxim.sent] == ["合并回复"]

    db = session_factory()
    try:
        tasks = db.query(TakeoverReplyTask).order_by(TakeoverReplyTask.created_at).all()
        assert [task.status for task in tasks] == ["cancelled", "sent"]
        assert tasks[0].cancel_reason == "newer_incoming_message"
    finally:
        db.close()


@pytest.mark.asyncio
async def test_connection_failure_disables_takeover_and_stops_retrying(service_context):
    session_factory, service, boxim, _ = service_context
    boxim.exchange_access_token = AsyncMock(
        side_effect=BoxIMError("无效的访问令牌", code=40101, auth_error=True)
    )

    await service.poll_and_process_messages()
    await service.poll_and_process_messages()

    db = session_factory()
    try:
        avatar = db.query(Avatar).one()
        cursor = db.query(TakeoverCursor).one()
        assert "takeover" not in avatar.config["authorizationPermissions"]
        assert cursor.initialized is False
        assert "重新登录" in cursor.last_error
        assert db.query(TakeoverReplyTask).count() == 0
    finally:
        db.close()
    boxim.exchange_access_token.assert_awaited_once_with("prod-huihui-token")


def test_plain_text_reply_removes_markdown_and_empty_lines():
    assert _plain_text_reply("## 建议\n\n**不能自行用药**\n`必要时就医`") == "建议\n不能自行用药\n必要时就医"