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

import asyncio
import json
from datetime import datetime, timedelta, timezone
from threading import Barrier
from unittest.mock import AsyncMock, patch

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

from database import Base
from models import Avatar, ChatAttachment, TakeoverCursor, TakeoverMessage, TakeoverReplyTask, User
from services.boxim_client import BoxIMError
from services.boxim_image_service import DownloadedBoxIMImage
from services.takeover_service import (
    AVATAR_LOCAL_ID_PREFIX,
    TakeoverService,
    _avatar_local_id,
    _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)}


class ConcurrentPollingBoxIM(FakeBoxIM):
    def __init__(self):
        super().__init__()
        self.active_polls = 0
        self.peak_active_polls = 0

    async def exchange_access_token(self, huihui_token):
        return {"accessToken": huihui_token, "accessTokenExpiresIn": 3600}

    async def get_self(self, access_token):
        return {"id": 100 if access_token == "prod-huihui-token" else 101}

    async def fetch_private_messages(self, access_token, min_id="0"):
        self.active_polls += 1
        self.peak_active_polls = max(self.peak_active_polls, self.active_polls)
        await asyncio.sleep(0.05)
        self.active_polls -= 1
        return []


@pytest.fixture
def service_context(tmp_path):
    engine = create_engine(
        f"sqlite:///{tmp_path / 'takeover.db'}",
        connect_args={"check_same_thread": False},
    )
    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"],
            "takeoverReplyDelaySeconds": 3,
        },
    )
    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": "你好"}
    )

    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)
    with patch("routers.chat._resolve_reply", return_value={"answer": "**你好**\n\n很高兴见到你"}):
        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_incoming_image_is_analyzed_and_used_in_takeover_reply(service_context):
    session_factory, service, boxim, clock = service_context
    await service.poll_and_process_messages()
    boxim.messages.append(
        {
            "id": 111,
            "localId": 111,
            "sendId": 200,
            "recvId": 100,
            "sendTime": clock.millis(),
            "type": 1,
            "content": json.dumps(
                {
                    "originUrl": "https://cdn.example/case.png",
                    "thumbUrl": "https://cdn.example/case-thumb.png",
                }
            ),
        }
    )

    await service.poll_and_process_messages()
    db = session_factory()
    try:
        scheduled = db.query(TakeoverReplyTask).filter_by(trigger_message_id="111").one()
        assert scheduled.status == "pending"
        assert scheduled.prompt == "请看看这张图片。"
    finally:
        db.close()
    clock.advance(3)

    def analyze(db, avatar, content, **kwargs):
        assert content == b"image-content"
        attachment = ChatAttachment(
            avatar_id=avatar.id,
            uploader_kind=kwargs["uploader_kind"],
            filename=kwargs["filename"],
            mime_type="image/jpeg",
            file_size=len(content),
            status="ready",
            category="medical_document",
            summary="一张门诊病例",
            extracted_text="主诉：咳嗽三天",
            structured_data={"medical": {"chief_complaint": "咳嗽三天"}},
            warning="请核对原始资料",
            expires_at=clock.now() + timedelta(hours=24),
        )
        db.add(attachment)
        db.commit()
        db.refresh(attachment)
        return attachment

    downloaded = DownloadedBoxIMImage(
        content=b"image-content",
        filename="case.png",
        mime_type="image/png",
        source_url="https://cdn.example/case.png",
    )
    with (
        patch("services.takeover_service.download_boxim_image", return_value=downloaded),
        patch("routers.chat._analyze_image_bytes", side_effect=analyze) as analyzer,
        patch("routers.chat._resolve_reply", return_value={"answer": "这份资料里写的是咳嗽三天。"}) as resolver,
    ):
        await service.poll_and_process_messages()

    analyzer.assert_called_once()
    assert resolver.call_args.args[2] == "请看看这张图片。"
    image_contexts = resolver.call_args.kwargs["image_contexts"]
    assert image_contexts[0]["summary"] == "一张门诊病例"
    assert image_contexts[0]["extractedText"] == "主诉：咳嗽三天"
    assert [item["content"] for item in boxim.sent] == ["这份资料里写的是咳嗽三天。"]

    db = session_factory()
    try:
        event = db.query(TakeoverMessage).filter_by(boxim_message_id="111").one()
        task = db.query(TakeoverReplyTask).filter_by(trigger_message_id="111").one()
        assert event.attachment_id
        assert db.get(ChatAttachment, event.attachment_id).uploader_kind == "boxim"
        assert task.status == "sent"
        with patch(
            "services.takeover_service.download_boxim_image",
            side_effect=AssertionError("cached image must not be downloaded again"),
        ):
            cached = service._takeover_image_attachment(db, db.get(Avatar, "avatar-1"), event)
        assert cached.id == event.attachment_id
    finally:
        db.close()


@pytest.mark.asyncio
async def test_invalid_image_message_is_recorded_but_not_scheduled(service_context):
    session_factory, service, boxim, clock = service_context
    await service.poll_and_process_messages()
    boxim.messages.append(
        {
            "id": 112,
            "localId": 112,
            "sendId": 200,
            "recvId": 100,
            "sendTime": clock.millis(),
            "type": 1,
            "content": json.dumps({"width": 100, "height": 100}),
        }
    )

    await service.poll_and_process_messages()

    db = session_factory()
    try:
        assert db.query(TakeoverMessage).filter_by(boxim_message_id="112").one()
        assert db.query(TakeoverReplyTask).count() == 0
    finally:
        db.close()


@pytest.mark.asyncio
async def test_default_reply_delay_is_three_minutes(service_context):
    session_factory, service, boxim, clock = service_context
    db = session_factory()
    try:
        avatar = db.query(Avatar).one()
        avatar.config = {"authorizationPermissions": ["chat", "takeover"]}
        db.commit()
    finally:
        db.close()

    await service.poll_and_process_messages()
    boxim.messages.append(
        {"id": 12, "localId": 12, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "三分钟后回复"}
    )
    await service.poll_and_process_messages()

    db = session_factory()
    try:
        task = db.query(TakeoverReplyTask).one()
        assert task.scheduled_at == clock.now() + timedelta(seconds=180)
    finally:
        db.close()

    clock.advance(179)
    await service.process_reply_tasks()
    assert boxim.sent == []
    clock.advance(1)
    with patch("routers.chat._resolve_reply", return_value={"answer": "好的"}):
        await service.poll_and_process_messages()
    assert [item["content"] for item in boxim.sent] == ["好的"]


@pytest.mark.asyncio
async def test_multiple_avatar_owners_are_polled_concurrently(service_context):
    session_factory, _service, _boxim, clock = service_context
    db = session_factory()
    try:
        db.add_all(
            [
                User(
                    id="owner-local-2",
                    huihui_user_id="owner-huihui-2",
                    huihui_token="prod-huihui-token-2",
                    app_token="app-token-2",
                ),
                Avatar(
                    id="avatar-2",
                    owner_id="owner-huihui-2",
                    name="分身二",
                    status="active",
                    config={"authorizationPermissions": ["chat", "takeover"]},
                ),
            ]
        )
        db.commit()
    finally:
        db.close()

    boxim = ConcurrentPollingBoxIM()
    service = TakeoverService(
        session_factory,
        boxim,
        poll_concurrency=2,
        now=clock.now,
    )

    await service.poll_messages()

    assert boxim.peak_active_polls == 2
    db = session_factory()
    try:
        assert db.query(TakeoverCursor).filter(TakeoverCursor.initialized.is_(True)).count() == 2
    finally:
        db.close()


@pytest.mark.asyncio
async def test_delayed_poll_still_schedules_recent_message(service_context):
    session_factory, service, boxim, clock = service_context
    await service.poll_messages()
    delayed_send_time = int(
        (clock.value - timedelta(seconds=150)).replace(tzinfo=timezone.utc).timestamp()
        * 1000
    )
    boxim.messages.append(
        {
            "id": 13,
            "localId": 13,
            "sendId": 200,
            "recvId": 100,
            "sendTime": delayed_send_time,
            "type": 0,
            "content": "排队后仍需回复",
        }
    )

    await service.poll_messages()

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

    with patch("routers.chat._resolve_reply", return_value={"answer": "已经收到"}):
        await service.process_reply_tasks()
    assert [item["content"] for item in boxim.sent] == ["已经收到"]


@pytest.mark.asyncio
async def test_avatar_origin_message_never_schedules_a_reply(service_context):
    session_factory, service, boxim, clock = service_context
    await service.poll_and_process_messages()
    local_id = _avatar_local_id("peer-owner", "peer-trigger")
    boxim.messages.append(
        {"id": 15, "localId": local_id, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "另一端分身回复"}
    )

    with patch("routers.chat._resolve_reply") as resolver:
        await service.poll_and_process_messages()

    resolver.assert_not_called()
    db = session_factory()
    try:
        event = db.query(TakeoverMessage).filter(TakeoverMessage.boxim_message_id == "15").one()
        assert event.is_avatar is True
        assert db.query(TakeoverReplyTask).count() == 0
    finally:
        db.close()


@pytest.mark.asyncio
async def test_peer_avatar_messages_are_excluded_from_later_human_context(service_context):
    _session_factory, service, boxim, clock = service_context
    await service.poll_and_process_messages()
    boxim.messages.append(
        {
            "id": 16,
            "localId": _avatar_local_id("peer-owner", "peer-trigger"),
            "sendId": 200,
            "recvId": 100,
            "sendTime": clock.millis(),
            "type": 0,
            "content": "分身生成的夸张长文",
        }
    )
    await service.poll_and_process_messages()

    clock.advance(1)
    boxim.messages.append(
        {
            "id": 17,
            "localId": 17,
            "sendId": 200,
            "recvId": 100,
            "sendTime": clock.millis(),
            "type": 0,
            "content": "真人的新问题",
        }
    )
    await service.poll_and_process_messages()
    clock.advance(3)
    with patch("routers.chat._resolve_reply", return_value={"answer": "正常回复"}) as resolver:
        await service.poll_and_process_messages()

    assert resolver.call_args.args[3] == []


@pytest.mark.asyncio
async def test_owner_message_pauses_future_takeover_for_ten_minutes(service_context):
    session_factory, service, boxim, clock = service_context
    await service.poll_and_process_messages()
    boxim.messages.append(
        {"id": 18, "localId": 18, "sendId": 100, "recvId": 200, "sendTime": clock.millis(), "type": 0, "content": "我先来回复"}
    )
    await service.poll_and_process_messages()

    clock.advance(30)
    boxim.messages.append(
        {"id": 19, "localId": 19, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "收到"}
    )
    with patch("routers.chat._resolve_reply") as resolver:
        await service.poll_and_process_messages()

    resolver.assert_not_called()
    db = session_factory()
    try:
        assert db.query(TakeoverReplyTask).count() == 0
    finally:
        db.close()


def test_avatar_local_id_is_deterministic_and_self_describing():
    first = _avatar_local_id("owner", "message-1")
    assert first == _avatar_local_id("owner", "message-1")
    assert first != _avatar_local_id("owner", "message-2")
    assert first.startswith(AVATAR_LOCAL_ID_PREFIX)
    assert len(first) == 18
    assert first.isdigit()


@pytest.mark.asyncio
async def test_different_contacts_generate_without_blocking_each_other(service_context):
    session_factory, service, boxim, clock = service_context
    await service.poll_and_process_messages()
    boxim.messages.extend(
        [
            {"id": 13, "localId": 31, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "联系人甲"},
            {"id": 14, "localId": 32, "sendId": 300, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "联系人乙"},
        ]
    )
    both_generating = Barrier(2, timeout=2)

    def resolve(_db, _avatar, prompt, _history, **_kwargs):
        both_generating.wait()
        return {"answer": f"回复{prompt[-1]}"}

    await service.poll_and_process_messages()

    clock.advance(3)
    with patch("routers.chat._resolve_reply", side_effect=resolve):
        await service.poll_and_process_messages()
    assert {(item["peerId"], item["content"]) for item in boxim.sent} == {
        ("200", "回复甲"),
        ("300", "回复乙"),
    }

    db = session_factory()
    try:
        assert {task.status for task in db.query(TakeoverReplyTask).all()} == {"sent"}
    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_owner_message_in_final_second_wins_before_generation(service_context):
    session_factory, service, boxim, clock = service_context
    await service.poll_and_process_messages()
    boxim.messages.append(
        {"id": 23, "localId": 23, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "还在吗"}
    )
    await service.poll_and_process_messages()

    clock.advance(2)
    boxim.messages.append(
        {"id": 24, "localId": 24, "sendId": 100, "recvId": 200, "sendTime": clock.millis(), "type": 0, "content": "我来处理"}
    )
    clock.advance(1)
    with patch("routers.chat._resolve_reply") as resolver:
        await service.poll_and_process_messages()

    resolver.assert_not_called()
    assert boxim.sent == []
    db = session_factory()
    try:
        task = db.query(TakeoverReplyTask).filter(TakeoverReplyTask.trigger_message_id == "23").one()
        assert task.status == "cancelled"
        assert task.cancel_reason == "owner_replied"
    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": "第一句"}
    )
    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": "第二句"}
    )
    await service.poll_and_process_messages()

    clock.advance(3)
    with patch("routers.chat._resolve_reply", return_value={"answer": "合并回复"}) as resolver:
        await service.poll_and_process_messages()
    assert resolver.call_args.args[2] == "第一句\n第二句"
    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")


@pytest.mark.asyncio
async def test_transient_connection_failure_keeps_takeover_and_recovers(service_context):
    session_factory, service, boxim, _ = service_context
    boxim.exchange_access_token = AsyncMock(
        side_effect=[
            BoxIMError("连接超时"),
            {"accessToken": "box-token", "accessTokenExpiresIn": 3600},
        ]
    )

    await service.poll_messages()

    db = session_factory()
    try:
        avatar = db.query(Avatar).one()
        cursor = db.query(TakeoverCursor).one()
        assert "takeover" in avatar.config["authorizationPermissions"]
        assert cursor.initialized is False
        assert "暂时连接失败" in cursor.last_error
    finally:
        db.close()

    await service.poll_messages()

    db = session_factory()
    try:
        avatar = db.query(Avatar).one()
        cursor = db.query(TakeoverCursor).one()
        assert "takeover" in avatar.config["authorizationPermissions"]
        assert cursor.initialized is True
        assert cursor.last_error == ""
    finally:
        db.close()
    assert boxim.exchange_access_token.await_count == 2


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