import asyncio
import hashlib
import json
import os
from contextlib import AsyncExitStack
from datetime import UTC, datetime, timedelta
from uuid import UUID

import httpx
import pytest
from httpx_sse import aconnect_sse
from sqlalchemy import select
from test_after_sales import scenario  # noqa: F401

from zhifu.core.errors import BusinessError
from zhifu.modules.agent_runtime.models import AgentRun
from zhifu.modules.agent_runtime.runtime import locked_run
from zhifu.modules.agent_runtime.worker import claim
from zhifu.modules.conversations.models import Conversation, ConversationEvent
from zhifu.modules.conversations.stream import read_events
from zhifu.modules.identity.access_models import StoreGrant
from zhifu.modules.identity.auth import now
from zhifu.operations.conversation_retention import advance_window

pytestmark = pytest.mark.integration


async def setup(context):
    async with context["app"].state.sessions.begin() as session:
        for name in ("agent", "reviewer"):
            for permission in ("conversations:read", "conversations:manage", "conversations:reply"):
                session.add(
                    StoreGrant(
                        tenant_id=context["tenant"].bytes,
                        principal_id=context["accounts"][name].bytes,
                        store_id=context["store"].bytes,
                        permission=permission,
                    )
                )
    response = await context["call"](
        "POST", "/openapi/v1/conversations", payload={"store_id": str(context["store"])}, key="new"
    )
    assert response.status_code == 201, response.text
    return "/openapi/v1/conversations/" + response.json()["conversation_id"]


async def test_takeover_invalidates_agent_and_release_restores_control(scenario):  # noqa: F811
    path = await setup(scenario)
    response = await scenario["call"](
        "POST", path + "/messages", payload={"text": "订单查询", "expected_control_version": 1}
    )
    assert response.status_code == 201
    sessions = scenario["app"].state.sessions
    run_id, token = await claim(sessions, 2)
    response = await scenario["call"](
        "POST", path + "/takeover", "agent", {"expected_control_version": 1}
    )
    assert response.status_code == 200, response.text
    async with sessions.begin() as session:
        with pytest.raises(BusinessError, match="控制权已变更"):
            await locked_run(session, run_id, token)
    response = await scenario["call"](
        "POST", path + "/messages", payload={"text": "客服你好", "expected_control_version": 2}
    )
    assert response.json()["payload"]["run_id"] is None
    response = await scenario["call"](
        "POST", path + "/messages", "reviewer", {"text": "回复", "expected_control_version": 2}
    )
    assert response.status_code == 403
    response = await scenario["call"](
        "POST", path + "/messages", "agent", {"text": "正在处理", "expected_control_version": 2}
    )
    assert response.status_code == 201
    response = await scenario["call"](
        "POST", path + "/release", "agent", {"expected_control_version": 2}
    )
    assert response.json()["control_version"] == 3
    response = await scenario["call"](
        "POST", path + "/messages", payload={"text": "继续查询", "expected_control_version": 3}
    )
    assert response.json()["payload"]["run_id"] is not None


async def test_competing_takeovers_and_message_idempotency(scenario):  # noqa: F811
    path = await setup(scenario)
    original = (await scenario["call"]("GET", path)).json()
    assert original["channel"] == "h5"
    created_at = datetime.fromisoformat(original["created_at"])
    assert created_at.utcoffset() == timedelta(0)
    responses = await asyncio.gather(
        *(
            scenario["call"]("POST", path + "/takeover", actor, {"expected_control_version": 1})
            for actor in ("agent", "reviewer")
        )
    )
    assert sorted(response.status_code for response in responses) == [200, 409]
    body = {"text": "同一消息", "expected_control_version": 2}
    first = await scenario["call"]("POST", path + "/messages", payload=body, key="message")
    again = await scenario["call"]("POST", path + "/messages", payload=body, key="message")
    assert first.json() == again.json()
    history = (await scenario["call"]("GET", path + "/messages")).json()["items"]
    assert history[-1] == first.json()
    # 通过实际数据库记录验证展示时间来源，接管和重复发送保留原会话与事件时间。
    async with scenario["app"].state.sessions.begin() as session:
        conversation_id = UUID(original["conversation_id"]).bytes
        conversation = await session.get(Conversation, conversation_id)
        assert created_at == conversation.created_at.replace(tzinfo=UTC)
        persisted = (await session.scalars(select(ConversationEvent).where(
            ConversationEvent.tenant_id == scenario["tenant"].bytes,
            ConversationEvent.conversation_id == conversation_id,
        ).order_by(ConversationEvent.sequence))).all()
        assert [datetime.fromisoformat(item["created_at"]) for item in history] == [
            item.created_at.replace(tzinfo=UTC) for item in persisted
        ]
    response = await scenario["call"]("GET", path, "other")
    assert response.status_code == 404
    response = await scenario["call"]("GET", path + "/messages", "foreign")
    assert response.status_code == 404
    listing = f"/openapi/v1/conversations?store_id={scenario['store']}"
    response = await scenario["call"]("GET", listing, "other")
    assert response.json()["items"] == []
    response = await scenario["call"]("GET", listing, "agent")
    assert len(response.json()["items"]) == 1
    listed = response.json()["items"][0]
    assert listed["channel"] == "h5" and listed["created_at"] == original["created_at"]
    assert (await scenario["call"]("GET", path)).json()["created_at"] == original["created_at"]


async def test_sse_http_reconnect_and_revocation(scenario):  # noqa: F811
    path = await setup(scenario)
    await scenario["call"]("POST", path + "/takeover", "agent", {"expected_control_version": 1})
    async with httpx.AsyncClient(
        base_url=os.environ.get("ACCEPTANCE_API_URL", "http://127.0.0.1:18000"), timeout=10
    ) as client:
        async with aconnect_sse(
            client, "GET", path + "/events", headers=scenario["headers"]["customer"]
        ) as stream:
            iterator = stream.aiter_sse()
            event = await anext(iterator)
            assert event.event == "control.changed" and event.id == "1"
            history = (await scenario["call"]("GET", path + "/messages")).json()["items"]
            assert json.loads(event.data)["created_at"] == history[0]["created_at"]
        await scenario["call"](
            "POST", path + "/messages", "agent", {"text": "处理进度", "expected_control_version": 2}
        )
        headers = {**scenario["headers"]["customer"], "Last-Event-ID": "1"}
        async with aconnect_sse(client, "GET", path + "/events", headers=headers) as stream:
            iterator = stream.aiter_sse()
            event = await anext(iterator)
            assert event.id == "2" and event.event == "message.staff"
            history = (await scenario["call"]("GET", path + "/messages")).json()["items"]
            assert json.loads(event.data)["created_at"] == history[-1]["created_at"]
            await scenario["call"]("DELETE", "/openapi/v1/auth/sessions/current")
            event = await anext(iterator)
            assert event.event == "access.revoked"


async def test_sse_connection_limit_and_cleanup(scenario):  # noqa: F811
    path = await setup(scenario)
    await scenario["call"]("POST", path + "/takeover", "agent", {"expected_control_version": 1})
    async with httpx.AsyncClient(
        base_url=os.environ.get("ACCEPTANCE_API_URL", "http://127.0.0.1:18000"), timeout=10
    ) as client:
        async with AsyncExitStack() as connections:
            iterators = []
            for _ in range(3):
                stream = await connections.enter_async_context(
                    aconnect_sse(
                        client, "GET", path + "/events", headers=scenario["headers"]["customer"]
                    )
                )
                iterators.append(stream.aiter_sse())
                assert (await anext(iterators[-1])).event == "control.changed"
            async with client.stream(
                "GET", path + "/events", headers=scenario["headers"]["customer"]
            ) as response:
                assert response.status_code == 429
        key = f"sse:actor:{scenario['tenant'].hex}:{scenario['accounts']['customer'].hex}"
        async with asyncio.timeout(5):
            while True:
                if await scenario["app"].state.state.zcard(key) == 0:
                    break
                await asyncio.sleep(0.05)


async def test_expired_and_cancelled_runs_release_conversation(scenario):  # noqa: F811
    path = await setup(scenario)
    sessions = scenario["app"].state.sessions
    body = {"text": "查询订单", "expected_control_version": 1}
    response = await scenario["call"]("POST", path + "/messages", payload=body)
    run_id = UUID(response.json()["payload"]["run_id"])
    async with sessions.begin() as session:
        run = await session.get(AgentRun, run_id.bytes)
        run.deadline_at = now() - timedelta(seconds=1)
    assert await claim(sessions, 2) is None
    response = await scenario["call"]("GET", path)
    assert response.json()["active_run_id"] is None
    response = await scenario["call"]("POST", path + "/messages", payload=body)
    run_id = response.json()["payload"]["run_id"]
    response = await scenario["call"]("POST", f"/openapi/v1/agent/runs/{run_id}/cancel")
    assert response.status_code == 200
    response = await scenario["call"]("GET", path)
    assert response.json()["active_run_id"] is None


async def test_stream_window_requires_state_reload_and_keeps_history(scenario):  # noqa: F811
    path = await setup(scenario)
    await scenario["call"]("POST", path + "/takeover", "agent", {"expected_control_version": 1})
    conversation_id = UUID(path.rsplit("/", 1)[1])
    sessions = scenario["app"].state.sessions
    cursor = await advance_window(sessions, scenario["tenant"].bytes,
                                  conversation_id.bytes, now())
    assert cursor == 1
    response = await scenario["call"]("GET", path + "/events")
    assert response.status_code == 410 and response.json()["code"] == "EVENT_CURSOR_EXPIRED"
    assert (await scenario["call"]("GET", path)).json()["stream_cursor"] == 1
    assert len((await scenario["call"]("GET", path + "/messages")).json()["items"]) == 1
    token = scenario["headers"]["customer"]["Authorization"].removeprefix("Bearer ")
    with pytest.raises(BusinessError, match="事件游标已过期"):
        await read_events(sessions, hashlib.sha256(token.encode()).digest(), "window-test",
                          conversation_id.bytes, 0, asyncio.Semaphore(1))