from collections.abc import Awaitable, Callable
from datetime import UTC, datetime
from typing import Any
from uuid import uuid4

from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.contracts import DomainEvent, RequestContext
from app.core.conversation_privacy import sanitize_customer_service_message
from app.core.errors import (
    FeedbackAlreadyExistsError,
    GenericResourceNotFoundError,
    InvalidStateError,
    RunNotCancellableError,
    RunNotFoundError,
)
from app.core.profile_projection import project_profile
from app.infrastructure.db import SessionFactory
from app.model.audit import InteractionAudit
from app.model.conversation import ConversationFeedback
from app.model.platform import AgentRun, HandoverTicket, RequestIdempotency
from app.model.session import ConversationSession
from app.repository.conversation_repository import ConversationRepository
from app.repository.outbox_repository import OutboxRepository
from app.repository.platform_repository import PlatformRepository
from app.repository.session_repository import SessionRepository
from app.service.admin_service import public
from app.service.agent.bootstrap import get_agent_factory
from app.service.api_transaction_service import ApiTransactionService
from app.service.authorization_service import AuthorizationService


class PublicPlatformService:
    async def session(self, session_id: str, context: RequestContext) -> dict[str, Any]:
        async with SessionFactory() as session:
            row = await SessionRepository(session).owned(session_id, int(context.user_id))
            return {"data": self._session_view(row), "meta": {"trace_id": context.trace_id}}

    async def handover(self, ticket_no: str, context: RequestContext) -> dict[str, Any]:
        async with SessionFactory() as session:
            row = await session.scalar(
                select(HandoverTicket).where(
                    HandoverTicket.ticket_no == ticket_no,
                    HandoverTicket.customer_id == int(context.user_id),
                )
            )
            if row is None:
                raise GenericResourceNotFoundError("转人工请求不存在")
            return {
                "data": {
                    "handover_id": row.ticket_no,
                    "status": row.status,
                    "created_at": public(row.created_at),
                },
                "meta": {"trace_id": context.trace_id},
            }

    async def write(
        self,
        operation: str,
        target: str,
        context: RequestContext,
        key: str | None,
        payload: dict[str, Any],
    ) -> dict[str, Any]:
        permission = {
            "create": "conversation:create",
            "close": "conversation:close",
            "cancel": "agent:cancel",
            "feedback": "conversation:feedback",
            "handover": "handover:create",
        }[operation]
        await AuthorizationService.require(context, permission)

        async def action(session: AsyncSession) -> dict[str, Any]:
            now = datetime.now(UTC).replace(tzinfo=None)
            user_id = int(context.user_id)
            session_id: str | None = None
            if operation == "create":
                get_agent_factory().authorize(payload["agent_type"], context)
                # 这四个时间列在模型里都是 `server_default=CURRENT_TIMESTAMP(6)`。
                # 不显式赋值的话，`flush()` 之后 SQLAlchemy 需要**回读**这些由数据库生成的
                # 值，而在 async session 里回读是异步 IO —— 紧接着 `_session_view(row)`
                # 以同步属性访问去读，就抛 `MissingGreenlet: greenlet_spawn has not been
                # called`，整个 `POST /api/v1/conversations` 500，连带转人工也做不了
                # （会话建不出来 ⇒ 后续 404 会话不存在）。显式传 `now` 与同文件
                # `ConversationFeedback(...)` 的写法一致，也贴合本项目"应用侧赋时间"的约定。
                row = ConversationSession(
                    session_id=str(uuid4()),
                    user_id=user_id,
                    agent_type=payload["agent_type"],
                    portal=context.portal,
                    status="active",
                    clarification_round=0,
                    message_count=0,
                    started_at=now,
                    last_active_at=now,
                    created_at=now,
                    updated_at=now,
                )
                session.add(row)
                await session.flush()
                session_id = row.session_id
                data = self._session_view(row)
            elif operation == "cancel":
                data, session_id = await self._cancel(session, target, user_id, now)
            elif operation == "feedback":
                message = await ConversationRepository(session).message(int(target), user_id)
                if message is None:
                    raise GenericResourceNotFoundError("消息不存在")
                if await ConversationRepository(session).feedback(int(target), user_id):
                    raise FeedbackAlreadyExistsError("消息已经反馈")
                feedback = ConversationFeedback(
                    feedback_no=f"fb-{uuid4().hex[:24]}",
                    session_id=message.session_id,
                    message_id=message.id,
                    customer_id=user_id,
                    created_at=now,
                    updated_at=now,
                    **payload,
                )
                session.add(feedback)
                await session.flush()
                session_id = message.session_id
                data = {"feedback_no": feedback.feedback_no, "status": feedback.status}
            else:
                row = await SessionRepository(session).owned(target, user_id, lock=True)
                session_id = row.session_id
                if operation == "close":
                    if row.status not in {"active", "ended"}:
                        raise InvalidStateError("当前会话状态不能关闭")
                    if row.status == "active":
                        row.status, row.ended_at = "ended", now
                    data = self._session_view(row)
                else:
                    if row.status != "active":
                        raise InvalidStateError("会话不在可转人工状态")
                    messages = await ConversationRepository(session).messages(target, user_id, 1)
                    ticket = HandoverTicket(
                        ticket_no=f"ticket-{uuid4().hex[:24]}",
                        session_id=target,
                        customer_id=user_id,
                        source_agent=row.agent_type or "customer_service",
                        source_message_id=messages[0].id if messages else None,
                        reason_code=payload["reason_code"],
                        # 用户自填原因同样是客服会话链路的一部分，不能把凭据原样落工单。
                        reason_detail=(
                            sanitize_customer_service_message(payload["reason_detail"])
                            if payload.get("reason_detail") is not None else None
                        ),
                        status="pending",
                        created_at=now,
                        updated_at=now,
                    )
                    session.add(ticket)
                    await session.flush()
                    await OutboxRepository(session).append(
                        DomainEvent(
                            event_id=str(uuid4()),
                            event_type="conversation.transfer_requested",
                            aggregate_type="conversation",
                            aggregate_id=target,
                            trace_id=context.trace_id,
                            payload={"ticket_no": ticket.ticket_no},
                            occurred_at=now,
                        )
                    )
                    data = {
                        "handover_id": ticket.ticket_no,
                        "status": ticket.status,
                        "session_id": target,
                        "created_at": public(now),
                    }
            session.add(
                InteractionAudit(
                    actor_type="user",
                    actor_id=user_id,
                    session_id=session_id,
                    portal=context.portal,
                    action_type=f"platform.{operation}",
                    detail={
                        "trace_id": context.trace_id,
                        "target": target,
                    },
                    created_at=now,
                )
            )
            return {"data": data, "meta": {"trace_id": context.trace_id}}

        return await self._transact(context, operation, target, key, payload, action)

    @staticmethod
    async def _transact(
        context: RequestContext,
        operation: str,
        target: str,
        key: str | None,
        payload: dict[str, Any],
        action: Callable[[AsyncSession], Awaitable[dict[str, Any]]],
    ) -> dict[str, Any]:
        """取消之外的写接口走幂等响应缓存。

        取消**刻意不走**响应缓存：文档 §6.4 要求重复取消返回同一状态，而运行状态是
        服务端权威状态——worker 把 `cancel_requested` 推进到 `cancelled` 之后，缓存的旧
        快照会变成过期数据。取消的幂等性由 `_cancel` 读实时状态保证。
        """
        if operation == "cancel":
            async with SessionFactory() as session, session.begin():
                return await action(session)
        return await ApiTransactionService().execute(
            context, f"public:{operation}:{target}", key, payload, action
        )

    async def _cancel(
        self, session: AsyncSession, run_id: str, user_id: int, now: datetime
    ) -> tuple[dict[str, Any], str]:
        """文档 §6.4 的取消语义。

        - 首次取消 `queued`/`running` → 置 `cancel_requested`，并把原请求的
          `request_idempotency` 写成 `failed + RUN_CANCELLED`；
        - 重复取消 → **幂等**，返回与首次一致的状态与 `cancel_requested_at`，绝不报错；
        - 已成功、已失败或已进入最终提交事务 → `409 RUN_NOT_CANCELLABLE`。

        `RUN_CANCELLED` 只落 `request_idempotency`，不作为 HTTP 响应错误码返回。
        """
        run = await session.scalar(
            select(AgentRun)
            .where(AgentRun.run_id == run_id, AgentRun.user_id == user_id)
            .with_for_update()
        )
        if run is None:
            raise RunNotFoundError("运行不存在或不可见")
        if run.status in {"succeeded", "failed"}:
            raise RunNotCancellableError("运行已进入不可取消阶段")
        if run.status in {"cancel_requested", "cancelled"}:
            # 重复取消：返回同一状态与同一受理时间，不重复改写幂等记录。
            return {
                "run_id": run_id,
                "status": run.status,
                "cancel_requested_at": public(run.cancel_requested_at),
            }, run.session_id
        run.status, run.cancel_requested_at = "cancel_requested", now
        # 文档 §6.4：取消成功后原请求在 request_idempotency 中以 failed + RUN_CANCELLED
        # 结束，不扩展其既有状态枚举。
        await session.execute(
            update(RequestIdempotency)
            .where(RequestIdempotency.id == run.idempotency_id)
            .values(status="failed", error_code="RUN_CANCELLED", updated_at=now)
        )
        return {
            "run_id": run_id,
            "status": run.status,
            "cancel_requested_at": public(run.cancel_requested_at),
        }, run.session_id

    @staticmethod
    def _session_view(row: ConversationSession) -> dict[str, Any]:
        return {
            "session_id": row.session_id,
            "agent_type": row.agent_type,
            "portal": row.portal,
            "status": row.status,
            "clarification_round": row.clarification_round,
            "message_count": row.message_count,
            "last_active_at": public(row.last_active_at),
            "started_at": public(row.started_at),
            "ended_at": public(row.ended_at),
            "created_at": public(row.created_at),
        }

    async def memory(self, customer_id: int, context: RequestContext) -> dict[str, Any]:
        own = str(customer_id) == context.user_id
        permission = "memory:read:self" if own else "memory:read:customer"
        await AuthorizationService.require(context, permission)
        scope = context.permission_scopes.get(permission, "self")
        if (
            not own
            and scope != "all"
            and (scope != "own_customers" or str(customer_id) not in context.customer_ids)
        ):
            raise GenericResourceNotFoundError("客户不可访问")
        async with SessionFactory() as session, session.begin():
            rows = await PlatformRepository(session).rows(
                "profile_snapshots", {"customer_id": customer_id, "is_current": 1}, limit=1
            )
            session.add(
                InteractionAudit(
                    actor_type="user",
                    actor_id=int(context.user_id),
                    target_customer_id=customer_id,
                    portal=context.portal,
                    action_type="memory.profile_read",
                    detail={"trace_id": context.trace_id},
                    created_at=datetime.now(UTC).replace(tzinfo=None),
                )
            )
            # 字段策略投影（白名单 + 测评有效期实时判定）；不再返回空字典。
            data: dict[str, Any] = {
                "customer_id": str(customer_id),
                "version": str(rows[0]["version"]) if rows else None,
                "generated_at": public(rows[0].get("generated_at")) if rows else None,
                "profile": project_profile(rows[0].get("snapshot")) if rows else {},
            }
            return {"data": data, "meta": {"trace_id": context.trace_id}}