import asyncio
from uuid import uuid4

import pytest
from sqlalchemy import delete, select

from app.core.contracts import AgentRequest, RequestContext
from app.core.errors import IdempotencyConflictError
from app.infrastructure.db import SessionFactory
from app.model.conversation import ConversationMessage
from app.model.platform import AgentRun, RequestIdempotency
from app.service.agent_run_application_service import AgentRunApplicationService

# 依赖真实 MySQL：必须打 integration marker，否则按 marker 过滤时会漏测这批用例。
pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("acceptance_registry")]


@pytest.mark.asyncio
async def test_accept_is_idempotent_and_persists_outbox() -> None:
    trace_id = str(uuid4())
    session_id = f"integration-{uuid4()}"
    key = f"key-{uuid4()}"
    context = RequestContext(user_id="1", trace_id=trace_id,
                             roles=("customer",), permissions=("agent:run",))
    request = AgentRequest(
        agent_type="customer_service", message="integration test", session_id=session_id,
        idempotency_key=key,
    )
    first = None
    async with SessionFactory() as session:
        try:
            service = AgentRunApplicationService(session)
            first = await service.accept(request, context)
            second = await service.accept(request, context)
            assert first.run_id == second.run_id
            assert second.status == "queued"
            with pytest.raises(IdempotencyConflictError):
                await service.accept(request.model_copy(update={"message": "different"}), context)
        finally:
            run = await session.scalar(
                select(AgentRun).where(AgentRun.run_id == first.run_id)
            ) if first is not None else None
            if run is not None:
                await session.execute(delete(AgentRun).where(AgentRun.id == run.id))
                await session.execute(
                    delete(RequestIdempotency).where(RequestIdempotency.id == run.idempotency_id)
                )
                await session.execute(
                    delete(ConversationMessage).where(ConversationMessage.session_id == session_id)
                )
                await session.commit()


@pytest.mark.asyncio
async def test_customer_service_accept_redacts_sensitive_message_before_persistence() -> None:
    """客服原始凭据只能存在于请求瞬间，数据库会话中必须是脱敏文本。"""
    session_id = f"privacy-{uuid4()}"
    key = f"privacy-key-{uuid4()}"
    raw_message = "验证码 123456，银行卡 6222021234567890123，登录密码: Secret123"
    context = RequestContext(
        user_id="1", trace_id=str(uuid4()), roles=("customer",), permissions=("agent:run",)
    )
    request = AgentRequest(
        agent_type="customer_service", message=raw_message, session_id=session_id,
        idempotency_key=key,
    )
    run_id = ""
    async with SessionFactory() as session:
        try:
            accepted = await AgentRunApplicationService(session).accept(request, context)
            run_id = accepted.run_id
            message = await session.scalar(select(ConversationMessage).where(
                ConversationMessage.session_id == session_id,
                ConversationMessage.role == "user",
            ))
            assert message is not None
            assert "123456" not in message.content
            assert "6222021234567890123" not in message.content
            assert "Secret123" not in message.content
            assert "验证码" in message.content
            assert "银行卡号已隐藏" in message.content
        finally:
            if run_id:
                run = await session.scalar(select(AgentRun).where(AgentRun.run_id == run_id))
                if run is not None:
                    await session.execute(delete(AgentRun).where(AgentRun.id == run.id))
                    await session.execute(delete(RequestIdempotency).where(
                        RequestIdempotency.id == run.idempotency_id
                    ))
            await session.execute(delete(ConversationMessage).where(
                ConversationMessage.session_id == session_id
            ))
            await session.commit()


@pytest.mark.asyncio
async def test_concurrent_same_key_creates_at_most_one_run() -> None:
    session_id = f"concurrent-{uuid4()}"
    key = f"concurrent-key-{uuid4()}"
    request = AgentRequest(
        agent_type="customer_service", message="concurrent test", session_id=session_id,
        idempotency_key=key,
    )

    async def submit() -> str | None:
        async with SessionFactory() as session:
            try:
                result = await AgentRunApplicationService(session).accept(
                    request, RequestContext(user_id="1", trace_id=str(uuid4()),
                                            roles=("customer",), permissions=("agent:run",))
                )
                return result.run_id
            except (IdempotencyConflictError, Exception):
                return None

    results = await asyncio.gather(*(submit() for _ in range(5)))
    run_ids = {run_id for run_id in results if run_id is not None}
    assert len(run_ids) == 1

    async with SessionFactory() as session:
        rows = await session.scalars(
            select(AgentRun).where(AgentRun.session_id == session_id)
        )
        runs = list(rows)
        assert len(runs) == 1
        if runs:
            await session.execute(delete(AgentRun).where(AgentRun.id == runs[0].id))
            await session.execute(
                delete(RequestIdempotency).where(RequestIdempotency.id == runs[0].idempotency_id)
            )
        await session.execute(
            delete(ConversationMessage).where(ConversationMessage.session_id == session_id)
        )
        await session.commit()