"""对话域服务：幂等、原子事务与越权防护。

不变量（契约 4.3/4.4 节）：Message 与可选 Run 在同一事务创建；权威状态、
AuditEvent 与 OutboxEvent 同事务提交；跨域对象按 404 fail-closed。
"""

import hashlib
import json
from datetime import UTC, datetime
from typing import Any

from fastapi import HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.conversations.schemas import MessageIn
from app.core.ids import new_id
from app.models import (
    AuditEvent,
    Branch,
    Conversation,
    IdempotencyRecord,
    Message,
    OutboxEvent,
    Run,
)

MESSAGE_ROUTE = "/v1/conversations/{conversation_id}/messages"


def _request_hash(body: MessageIn) -> str:
    canonical = json.dumps(
        {
            "content": [b.model_dump(exclude_none=True) for b in body.content],
            "start_run": body.start_run,
        },
        sort_keys=True,
        ensure_ascii=False,
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


async def _require_conversation(
    db: AsyncSession, domain_id: str, conversation_id: str
) -> Conversation:
    conversation = await db.get(Conversation, conversation_id)
    # 跨域对象统一 404，不泄露存在性（契约第 7 节）
    if conversation is None or conversation.personal_domain_id != domain_id:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "对话不存在")
    return conversation


async def _get_default_branch(db: AsyncSession, conversation_id: str) -> Branch:
    result = await db.execute(
        select(Branch).where(Branch.conversation_id == conversation_id).limit(1)
    )
    branch = result.scalar_one_or_none()
    assert branch is not None, "会话必须存在默认分支（创建事务保证）"
    return branch


async def create_conversation(
    db: AsyncSession, domain_id: str, title: str, trace_id: str | None
) -> Conversation:
    conversation = Conversation(personal_domain_id=domain_id, title=title)
    db.add(conversation)
    await db.flush()
    branch = Branch(conversation_id=conversation.id, title="主脉")
    db.add(branch)
    db.add(
        AuditEvent(
            actor_type="user",
            actor_id=domain_id,
            action="conversation.created",
            aggregate_type="conversation",
            aggregate_id=conversation.id,
            trace_id=trace_id,
        )
    )
    db.add(
        OutboxEvent(
            aggregate_type="conversation",
            aggregate_id=conversation.id,
            event_type="conversation.created",
            payload={"conversation_id": conversation.id, "title": title},
        )
    )
    await db.flush()
    return conversation


async def replay_or_execute_message(
    db: AsyncSession,
    domain_id: str,
    conversation_id: str,
    idempotency_key: str,
    body: MessageIn,
    trace_id: str | None,
) -> tuple[Message, Run | None, bool]:
    """返回（消息，Run，是否为重放）。幂等：相同键+相同请求返回原资源。"""
    route = MESSAGE_ROUTE
    existing = await db.execute(
        select(IdempotencyRecord).where(
            IdempotencyRecord.owner_domain_id == domain_id,
            IdempotencyRecord.route == route,
            IdempotencyRecord.key == idempotency_key,
        )
    )
    record = existing.scalar_one_or_none()
    request_hash = _request_hash(body)
    if record is not None:
        if record.request_hash != request_hash:
            raise HTTPException(status.HTTP_409_CONFLICT, "相同幂等键对应不同请求体")
        message = await db.get(Message, record.resource_ids.get("message_id"))
        run = None
        run_id = record.resource_ids.get("run_id")
        if run_id is not None:
            run = await db.get(Run, run_id)
        assert message is not None
        return message, run, True

    conversation = await _require_conversation(db, domain_id, conversation_id)
    branch = await _get_default_branch(db, conversation.id)

    blocks: list[dict[str, Any]] = [
        {**b.model_dump(exclude_none=True), "id": new_id("block_"), "status": "completed"}
        for b in body.content
    ]
    message = Message(
        conversation_id=conversation.id,
        branch_id=branch.id,
        role="user",
        content=blocks,
    )
    db.add(message)
    await db.flush()

    run = None
    if body.start_run:
        # 垂直切片空状态闭环：不调用模型，Run 原子创建并直接进入终态 completed。
        run = Run(
            personal_domain_id=domain_id,
            conversation_id=conversation.id,
            status="completed",
            completed_at=datetime.now(UTC),
        )
        db.add(run)
        await db.flush()

    resource_ids: dict[str, Any] = {"message_id": message.id}
    if run is not None:
        resource_ids["run_id"] = run.id
    db.add(
        AuditEvent(
            actor_type="user",
            actor_id=domain_id,
            action="message.created",
            aggregate_type="message",
            aggregate_id=message.id,
            trace_id=trace_id,
        )
    )
    db.add(
        OutboxEvent(
            aggregate_type="message",
            aggregate_id=message.id,
            event_type="message.created",
            payload={"message_id": message.id, "conversation_id": conversation.id},
        )
    )
    db.add(
        IdempotencyRecord(
            key=idempotency_key,
            owner_domain_id=domain_id,
            route=route,
            request_hash=request_hash,
            response_status=201,
            resource_ids=resource_ids,
        )
    )
    await db.flush()
    return message, run, False