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

不变量（契约 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 RUNTIME_BLOCK_TYPES, MessageIn
from app.core import lingtai
from app.core.ids import new_id
from app.models import (
    AuditEvent,
    Branch,
    Conversation,
    IdempotencyRecord,
    Message,
    OutboxEvent,
    Run,
    RunEvent,
)

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)

    forged = [b.type for b in body.content if b.type in RUNTIME_BLOCK_TYPES]
    if forged:
        raise HTTPException(
            status.HTTP_422_UNPROCESSABLE_ENTITY,
            f"系统产生类 block 不允许客户端提交：{', '.join(sorted(set(forged)))}",
        )

    blocks: list[dict[str, Any]] = []
    parser_version: str | None = None
    for b in body.content:
        if b.type == "markdown":
            # 契约第 4 节：markdown 输入经确定性 parser 规范化为 canonical blocks
            normalized = lingtai.transform_markdown(b.text)
            parser_version = str(normalized["parser_version"])
            blocks.extend(normalized["blocks"])
        else:
            blocks.append({**b.model_dump(exclude_none=True), "status": "completed"})
    for block in blocks:
        block["id"] = new_id("block_")
        block.setdefault("status", "completed")
    message = Message(
        conversation_id=conversation.id,
        branch_id=branch.id,
        role="user",
        content=blocks,
        parser_version=parser_version,
    )
    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
        _record_run_events(db, run, message, blocks)
    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


def _record_run_events(
    db: AsyncSession, run: Run, message: Message, blocks: list[dict[str, Any]]
) -> None:
    """空状态闭环的事件流：状态事件 + 每 block completed + message.completed。

    sequence 在 Run 内从 0 单调；事件与权威状态同事务提交（契约第 5 节）。
    delta/progress 事件在模型流式接入（WP-05）后由真实流产生，此处不伪造。
    """
    seq = 0
    events: list[tuple[str, dict[str, Any]]] = [
        ("run.status.changed", {"run_id": run.id, "status": run.status})
    ]
    for index, block in enumerate(blocks):
        events.append(
            (
                "content.block.completed",
                {"run_id": run.id, "message_id": message.id, "index": index,
                 "block": {"id": block["id"], "type": block["type"], "status": "completed"}},
            )
        )
    events.append(
        (
            "message.completed",
            {"run_id": run.id, "message_ids": [message.id], "final_status": run.status},
        )
    )
    for event_type, data in events:
        db.add(RunEvent(run_id=run.id, sequence=seq, type=event_type, data=data))
        seq += 1


async def list_messages(
    db: AsyncSession,
    domain_id: str,
    conversation_id: str,
    branch_id: str | None,
    limit: int,
    cursor: str | None,
) -> tuple[list[Message], str | None, bool]:
    """按创建时间升序的消息列表（opaque cursor 分页，契约第 4.1 节）。"""
    conversation = await _require_conversation(db, domain_id, conversation_id)
    stmt = select(Message).where(Message.conversation_id == conversation.id)
    if branch_id is not None:
        stmt = stmt.where(Message.branch_id == branch_id)
    after_created, after_id = decode_cursor(cursor)
    if after_created is not None:
        # (created_at, id) 复合键分页：严格大于游标位置
        stmt = stmt.where(
            (Message.created_at > after_created)
            | ((Message.created_at == after_created) & (Message.id > after_id))
        )
    stmt = stmt.order_by(Message.created_at.asc(), Message.id.asc()).limit(limit + 1)
    rows = list((await db.scalars(stmt)).all())
    has_more = len(rows) > limit
    page = rows[:limit]
    next_cursor = encode_cursor(page[-1]) if has_more and page else None
    return page, next_cursor, has_more


def encode_cursor(message: Message) -> str:
    import base64

    raw = f"{message.created_at.isoformat()}|{message.id}"
    return base64.urlsafe_b64encode(raw.encode()).decode()


def decode_cursor(cursor: str | None) -> tuple[Any | None, str]:
    import base64

    if cursor is None:
        return None, ""
    try:
        raw = base64.urlsafe_b64decode(cursor.encode()).decode()
        created_str, _, mid = raw.partition("|")
        return datetime.fromisoformat(created_str), mid
    except Exception as exc:
        raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "cursor 不合法") from exc


async def list_branches(
    db: AsyncSession, domain_id: str, conversation_id: str
) -> list[Branch]:
    conversation = await _require_conversation(db, domain_id, conversation_id)
    result = await db.execute(
        select(Branch).where(Branch.conversation_id == conversation.id).order_by(Branch.created_at)
    )
    return list(result.scalars().all())


async def set_current_branch(
    db: AsyncSession, domain_id: str, conversation_id: str, branch_id: str
) -> Conversation:
    """切换当前路径：分支必须属于该会话（跨域/跨会话统一 404/422 fail-closed）。"""
    conversation = await _require_conversation(db, domain_id, conversation_id)
    branch = await db.get(Branch, branch_id)
    if branch is None or branch.conversation_id != conversation.id:
        raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "分支不属于该会话")
    conversation.current_branch_id = branch_id
    await db.flush()
    return conversation