group_fqcd_jr/app/service/public_platform_service.py

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.errors import (
    FeedbackAlreadyExistsError,
    GenericResourceNotFoundError,
    InvalidStateError,
    RunNotCancellableError,
    RunNotFoundError,
)
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)
                row = ConversationSession(
                    session_id=str(uuid4()),
                    user_id=user_id,
                    agent_type=payload["agent_type"],
                    portal=context.portal,
                )
                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=payload.get("reason_detail"),
                        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),
                )
            )
            # Safe minimum projection; arbitrary snapshot JSON needs field-policy expansion.
            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": {},
            }
            return {"data": data, "meta": {"trace_id": context.trace_id}}