import asyncio
import hashlib
import hmac
import json
import os
import secrets
from datetime import UTC, datetime, timedelta
from uuid import UUID, uuid4

import pytest
import pytest_asyncio
from langchain_core.messages import AIMessage, message_to_dict
from sqlalchemy import func, select
from test_after_sales import make_proposal, scenario  # noqa: F401
from test_platform_management import receiver  # noqa: F401
from test_storefront import challenge, prepare_channel, proof

from zhifu.modules.after_sales.models import Approval, BusinessEvent
from zhifu.modules.agent_runtime.models import AgentRun
from zhifu.modules.agent_runtime.runtime import build_tool_graph
from zhifu.modules.agent_runtime.worker import claim
from zhifu.modules.channels.models import ChannelInbox, ChannelKey
from zhifu.modules.conversations.models import Conversation
from zhifu.modules.identity.access_models import StoreGrant
from zhifu.modules.merchant.models import ChannelApplication
from zhifu.modules.platform.delivery import enqueue, execute_one
from zhifu.modules.platform.models import ChannelDelivery
from zhifu.operations.channel_keys import KeyConfiguration, configure

pytestmark = pytest.mark.integration


@pytest_asyncio.fixture
async def channel(scenario):  # noqa: F811
    await prepare_channel(scenario)
    call = scenario["call"]
    item = await challenge(scenario)
    path = f"/openapi/v1/channel/identity-challenges/{item['challenge_id']}/verify"
    assert (await call("POST", path, "connector", proof(scenario))).status_code == 200
    confirm_path = f"/openapi/v1/me/identity-challenges/{item['challenge_id']}/confirm"
    assert (await call("POST", confirm_path)).status_code == 200
    sessions = scenario["app"].state.sessions
    async with sessions.begin() as session:
        row = await session.get(
            ChannelApplication, (scenario["tenant"].bytes, scenario["store"].bytes))
        row.scopes = ["conversations:messages", "customers:actions", "tickets:results"]
    env = "CHANNEL_TEST_" + uuid4().hex.upper()
    secret = secrets.token_urlsafe(32)
    os.environ[env] = secret
    config = KeyConfiguration(key_id="key-1", credential_env=env,
        valid_from=datetime.now(UTC) - timedelta(minutes=1),
        valid_until=datetime.now(UTC) + timedelta(hours=1))
    await configure(sessions, scenario["tenant"].bytes, scenario["store"].bytes, config)
    endpoint = f"/channel/v1/bindings/{scenario['store']}/messages"

    async def send(body, *, nonce=None, stamp=None, key_id="key-1", signing_secret=None):
        content = json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode()
        stamp = int(datetime.now(UTC).timestamp()) if stamp is None else stamp
        nonce = nonce or uuid4().hex
        signed = b"\n".join([str(stamp).encode(), nonce.encode(), b"POST", endpoint.encode(),
                             hashlib.sha256(content).hexdigest().encode()])
        signature = hmac.new(
            (signing_secret or secret).encode(), signed, hashlib.sha256).hexdigest()
        return await scenario["client"].post(endpoint, content=content, headers={
            "Content-Type": "application/json", "X-Key-Id": key_id,
            "X-Timestamp": str(stamp),
            "X-Nonce": nonce, "X-Signature": signature})
    try:
        yield {**scenario, "send": send, "config": config, "secret": secret}
    finally:
        os.environ.pop(env)


def envelope(kind="message", payload=None):
    return {"external_message_id": uuid4().hex, "external_user_id": "verified-channel-account",
            "external_conversation_id": "conversation-1",
            "occurred_at": datetime.now(UTC).isoformat(),
            "type": kind, "payload": payload or {"text": "请查询我的订单"}}


async def test_signed_message_concurrent_dedup_and_channel_history_isolation(channel):
    body = envelope()
    results = await asyncio.gather(*[channel["send"](body) for _ in range(4)])
    assert all(response.status_code == 202 for response in results), [r.text for r in results]
    assert len({response.json()["inbox_id"] for response in results}) == 1
    result = results[0].json()
    different = await channel["send"]({**body, "payload": {"text": "修改消息正文"}})
    assert different.status_code == 409 and different.json()["code"] == "CHANNEL_MESSAGE_CONFLICT"
    h5 = await channel["call"]("GET", "/openapi/v1/conversations?store_id=" + str(channel["store"]))
    assert not h5.json()["items"]
    history_path = f"/openapi/v1/conversations/{result['conversation_id']}"
    assert (await channel["call"]("GET", history_path)).status_code == 404
    async with channel["app"].state.sessions() as session:
        for model in (ChannelInbox, AgentRun, Conversation):
            assert await session.scalar(select(func.count()).select_from(model).where(
                model.tenant_id == channel["tenant"].bytes)) == 1


async def test_signature_replay_mapping_and_disabled_key(channel):
    body = envelope()
    invalid = await channel["send"](body, signing_secret="x" * 32)
    assert invalid.json()["code"] == "CHANNEL_SIGNATURE_INVALID"
    assert (await channel["send"](body, stamp=1)).json()["code"] == "CHANNEL_TIMESTAMP_EXPIRED"
    assert (await channel["send"](body, key_id="unknown")).json()["code"] == "CHANNEL_KEY_DISABLED"
    assert (await channel["send"]({**body, "external_user_id": "unlinked"})).status_code == 404
    assert (await channel["send"]({**body, "tenant_id": str(uuid4())})).status_code == 422
    nonce = uuid4().hex
    assert (await channel["send"](body, nonce=nonce)).status_code == 202
    assert (await channel["send"](body, nonce=nonce)).json()["code"] == "CHANNEL_REPLAY"
    sessions = channel["app"].state.sessions
    rotated = channel["config"].model_copy(update={"key_id": "key-2"})
    await configure(sessions, channel["tenant"].bytes, channel["store"].bytes, rotated)
    assert (await channel["send"](body, key_id="key-2")).status_code == 202
    await configure(sessions, channel["tenant"].bytes, channel["store"].bytes,
                    channel["config"], enabled=False)
    assert (await channel["send"](body)).json()["code"] == "CHANNEL_KEY_DISABLED"
    async with sessions.begin() as session:
        row = await session.get(
            ChannelKey, (channel["tenant"].bytes, channel["store"].bytes, "key-2"))
        row.valid_until = datetime.now(UTC).replace(tzinfo=None) - timedelta(seconds=1)
    assert (await channel["send"](body, key_id="key-2")).status_code == 401


async def test_explicit_channel_confirmation_uses_real_business_service(channel):
    response = await channel["send"](envelope(payload={"text": "确认及批准换货"}))
    assert response.status_code == 202, response.text
    async with channel["app"].state.sessions() as session:
        assert await session.scalar(select(func.count()).select_from(Approval).where(
            Approval.tenant_id == channel["tenant"].bytes)) == 0
    proposal = await make_proposal(channel)
    body = envelope("proposal_confirmation", {
        "proposal_id": proposal["proposal_id"], "proposal_version": proposal["proposal_version"],
        "expected_ticket_version": proposal["version"],
    })
    responses = await asyncio.gather(*[channel["send"](body) for _ in range(3)])
    assert all(item.status_code == 202 for item in responses), [r.text for r in responses]
    assert responses[0].json()["result"]["status"] == "WAITING_REVIEW"
    async with channel["app"].state.sessions() as session:
        assert await session.scalar(select(func.count()).select_from(Approval).where(
            Approval.tenant_id == channel["tenant"].bytes)) == 1
    invalid = await channel["send"]({**body, "payload": {**body["payload"], "approved": True}})
    assert invalid.status_code == 422
    async with channel["app"].state.sessions.begin() as session:
        app = await session.get(
            ChannelApplication, (channel["tenant"].bytes, channel["store"].bytes))
        app.enabled = False
    assert (await channel["send"](body)).json()["code"] == "CHANNEL_UNAVAILABLE"


async def pause_channel(channel):
    response = await channel["send"](envelope())
    assert response.status_code == 202, response.text
    conversation_id = UUID(response.json()["conversation_id"])
    sessions = channel["app"].state.sessions
    async with sessions.begin() as session:
        conversation = await session.get(Conversation, conversation_id.bytes)
        run = await session.get(AgentRun, conversation.active_run_id)
        # 直接提交工具命令，验证渠道身份与真实图节点，生成模型调用次数为0。
        run.messages = run.messages + [message_to_dict(AIMessage(content="", tool_calls=[{
            "name": "request_clarification", "id": uuid4().hex,
            "args": {"question": "请补充换货原因", "missing_fields": ["reason"]},
            "type": "tool_call",
        }]))]
        run.next_node = "tools"
    run_id, token = await claim(sessions, 2)
    result = await build_tool_graph(sessions).ainvoke({"run_id": run_id, "token": token})
    assert result["next_node"] == "wait"
    async with sessions() as session:
        run = await session.get(AgentRun, run_id)
        assert run.status == "WAITING_USER" and run.model_calls == 0
        return conversation_id, run_id, run.waiting


async def test_channel_clarification_resume_and_revoked_scope(channel):
    _, run_id, waiting = await pause_channel(channel)
    body = envelope("clarification", {
        "interrupt_id": waiting["interrupt_id"],
        "expected_run_version": waiting["run_version"], "answer": "按键失灵",
    })
    sessions = channel["app"].state.sessions
    async with sessions.begin() as session:
        app = await session.get(ChannelApplication,
                                (channel["tenant"].bytes, channel["store"].bytes))
        app.scopes = ["conversations:messages"]
    assert (await channel["send"](body)).status_code == 403
    async with sessions.begin() as session:
        app = await session.get(ChannelApplication,
                                (channel["tenant"].bytes, channel["store"].bytes))
        app.scopes = ["conversations:messages", "customers:actions"]
    responses = await asyncio.gather(*[channel["send"](body) for _ in range(3)])
    assert all(response.status_code == 202 for response in responses), [r.text for r in responses]
    assert len({response.json()["inbox_id"] for response in responses}) == 1
    assert responses[0].json()["result"]["run_id"] == str(UUID(bytes=run_id))
    async with sessions() as session:
        run = await session.get(AgentRun, run_id)
        assert run.status == "QUEUED" and run.waiting is None and run.model_calls == 0


async def test_channel_conversation_tls_order_and_takeover(channel, receiver):  # noqa: F811
    conversation_id, _, _ = await pause_channel(channel)
    sessions = channel["app"].state.sessions
    async with sessions.begin() as session:
        app = await session.get(ChannelApplication,
                                (channel["tenant"].bytes, channel["store"].bytes))
        app.callback_url, app.credential_env = receiver["url"], receiver["credential_env"]
        session.add(StoreGrant(
            tenant_id=channel["tenant"].bytes, principal_id=channel["accounts"]["agent"].bytes,
            store_id=channel["store"].bytes, permission="conversations:manage",
        ))
    assert await enqueue(sessions, tenant_id=channel["tenant"].bytes) == 2
    async with sessions() as session:
        deliveries = (await session.scalars(select(ChannelDelivery).join(
            BusinessEvent, BusinessEvent.id == ChannelDelivery.event_id
        ).where(
            ChannelDelivery.tenant_id == channel["tenant"].bytes
        ).order_by(BusinessEvent.aggregate_version))).all()
    first, second = deliveries
    assert await execute_one(sessions, second.id) is False
    receiver["accepted"].set()
    assert await execute_one(sessions, first.id) is True
    assert len(receiver["received"]) == 1
    received = receiver["received"][0]
    assert received["event_type"] == "conversation.message.assistant"
    assert received["payload"]["external_subject"] == "verified-channel-account"
    assert received["payload"]["event"]["payload"]["text"] == "请补充换货原因"
    path = f"/openapi/v1/conversations/{conversation_id}/takeover"
    response = await channel["call"]("POST", path, "agent", {"expected_control_version": 1})
    assert response.status_code == 200, response.text
    assert await execute_one(sessions, second.id) is True
    async with sessions() as session:
        stale = await session.get(ChannelDelivery, second.id)
        assert stale.status == "FAILED" and stale.last_error == "CONVERSATION_CONTROL_LOST"
        assert stale.attempts == 0
    assert len(receiver["received"]) == 1