X-Financial/server/src/app/services/risk_dispositions.py

from __future__ import annotations

import hashlib
import json
import os
import uuid
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import Any

from pydantic import ValidationError
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.api.deps import CurrentUserContext
from app.core.logging import get_logger
from app.models.financial_record import ExpenseClaim
from app.models.risk_disposition import RiskDisposition, RiskDispositionEvent
from app.models.risk_observation import RiskObservation, RiskObservationFeedback
from app.schemas.risk_disposition import (
    RiskDispositionActionCreate,
    RiskDispositionEventRead,
    RiskDispositionMutationRead,
    RiskDispositionRead,
)
from app.services.expense_claim_tenant_scope import ExpenseClaimTenantScopeMixin
from app.services.risk_observation_access_policy import RiskObservationAccessPolicy

logger = get_logger("app.services.risk_dispositions")


class RiskDispositionConflictError(RuntimeError):
    pass


class RiskDispositionVersionConflictError(RiskDispositionConflictError):
    def __init__(self, current_version: int) -> None:
        self.current_version = current_version
        super().__init__(
            f"Risk disposition version conflict; current version is {current_version}."
        )


class RiskDispositionIdempotencyConflictError(RiskDispositionConflictError):
    pass


class RiskDispositionPermissionError(RuntimeError):
    pass


@dataclass(slots=True)
class RiskDispositionMutation:
    response: RiskDispositionMutationRead
    legacy_feedback: RiskObservationFeedback | None = None

    @property
    def disposition(self) -> RiskDispositionRead:
        return self.response.disposition

    @property
    def event(self) -> RiskDispositionEventRead:
        return self.response.event

    @property
    def replayed(self) -> bool:
        return self.response.replayed


class RiskDispositionService:
    def __init__(self, db: Session) -> None:
        self.db = db

    def get_disposition(
        self,
        observation_id: str,
        *,
        tenant_id: str,
    ) -> RiskDisposition | None:
        return self.db.scalar(
            select(RiskDisposition).where(
                RiskDisposition.tenant_id == _tenant(tenant_id),
                RiskDisposition.observation_id == observation_id,
            )
        )

    def get_current_version(self, observation_id: str, *, tenant_id: str) -> int:
        disposition = self.get_disposition(observation_id, tenant_id=tenant_id)
        return disposition.version if disposition is not None else 0

    def execute_action(
        self,
        observation_key_or_id: str,
        payload: RiskDispositionActionCreate,
        *,
        tenant_id: str,
        actor_id: str,
        actor_name: str,
        current_user: CurrentUserContext | None = None,
    ) -> RiskDispositionMutation:
        normalized_tenant = _tenant(tenant_id)
        normalized_actor_id = _text(actor_id) or "anonymous"
        observation_locator = self.db.execute(
            select(RiskObservation.id, RiskObservation.claim_id).where(
                RiskObservation.tenant_id == normalized_tenant,
                (
                    (RiskObservation.id == observation_key_or_id)
                    | (RiskObservation.observation_key == observation_key_or_id)
                ),
            )
        ).one_or_none()
        if observation_locator is None:
            raise LookupError("Risk observation not found.")
        observation_id, located_claim_id = observation_locator
        fingerprint = _payload_fingerprint(
            payload,
            observation_id=observation_id,
            actor_id=normalized_actor_id,
        )
        replay = self._find_replay(
            tenant_id=normalized_tenant,
            request_id=payload.request_id,
            fingerprint=fingerprint,
        )
        if replay is not None:
            return replay

        try:
            locked_claim = self._lock_claim(
                str(located_claim_id or "").strip(),
                tenant_id=normalized_tenant,
            )
            observation = self.db.scalar(
                select(RiskObservation)
                .where(
                    RiskObservation.tenant_id == normalized_tenant,
                    RiskObservation.id == observation_id,
                )
                .with_for_update()
            )
            if observation is None:
                raise LookupError("Risk observation not found.")
            if str(observation.claim_id or "").strip() != str(located_claim_id or "").strip():
                raise RiskDispositionConflictError(
                    "Risk observation claim changed concurrently; reload and retry."
                )
            if current_user is not None and not RiskObservationAccessPolicy(
                self.db
            ).can_manage_locked_disposition(
                observation,
                current_user,
                locked_claim=locked_claim,
            ):
                raise RiskDispositionPermissionError("当前用户已不再是该单据的有效审批人。")

            disposition = self.db.scalar(
                select(RiskDisposition)
                .where(
                    RiskDisposition.tenant_id == normalized_tenant,
                    RiskDisposition.observation_id == observation.id,
                )
                .with_for_update()
            )
            current_version = disposition.version if disposition is not None else 0
            if current_version != payload.expected_version:
                raise RiskDispositionVersionConflictError(current_version)

            mutation_at = datetime.now(UTC)
            if disposition is None:
                disposition = RiskDisposition(
                    id=str(uuid.uuid4()),
                    tenant_id=normalized_tenant,
                    observation_id=observation.id,
                    adjudication=_initial_adjudication(observation),
                    lifecycle_status=_initial_lifecycle_status(observation),
                    version=0,
                    created_at=mutation_at,
                    updated_at=mutation_at,
                )
                self.db.add(disposition)

            _validate_transition(disposition, payload)
            before = _state(disposition)
            _apply_action(disposition, observation, payload)
            disposition.version = current_version + 1
            disposition.updated_at = mutation_at
            event = RiskDispositionEvent(
                id=str(uuid.uuid4()),
                tenant_id=normalized_tenant,
                disposition_id=disposition.id,
                observation_id=observation.id,
                version=disposition.version,
                action=payload.action,
                actor_id=normalized_actor_id,
                actor_name=_text(actor_name) or _text(actor_id) or "anonymous",
                request_id=payload.request_id,
                payload_fingerprint=fingerprint,
                comment=payload.comment,
                before_json=before,
                after_json=_state(disposition),
                created_at=mutation_at,
            )
            response = self._build_response_snapshot(
                disposition,
                event,
                prior_version=current_version,
            )
            # PostgreSQL 会拒绝 append-only 事件的后续 UPDATE，因此完整响应
            # 必须和事件在同一条 INSERT 中原子写入。
            event.response_json = response.model_dump(mode="json")
            self.db.add(event)
            legacy_feedback = self._append_safe_feedback(
                observation,
                event,
                payload,
                actor_name=actor_name,
            )
            self.db.commit()
            self.db.refresh(disposition)
            self.db.refresh(event)
            if legacy_feedback is not None:
                self.db.refresh(legacy_feedback)
                self._ingest_feedback_sample(observation, legacy_feedback)
            return RiskDispositionMutation(
                response=response,
                legacy_feedback=legacy_feedback,
            )
        except (LookupError, RiskDispositionConflictError, RiskDispositionPermissionError):
            self.db.rollback()
            raise
        except IntegrityError as error:
            self.db.rollback()
            replay = self._find_replay(
                tenant_id=normalized_tenant,
                request_id=payload.request_id,
                fingerprint=fingerprint,
            )
            if replay is not None:
                return replay
            raise RiskDispositionConflictError(
                "Risk disposition was changed concurrently; reload and retry."
            ) from error

    def _lock_claim(
        self,
        claim_id: str,
        *,
        tenant_id: str,
    ) -> ExpenseClaim | None:
        if not claim_id:
            return None
        statement = select(ExpenseClaim).where(
            ExpenseClaim.id == claim_id,
            ExpenseClaimTenantScopeMixin.build_claim_tenant_condition(tenant_id),
        )
        bind = self.db.get_bind()
        if bind is not None and bind.dialect.name == "postgresql":
            statement = statement.with_for_update()
        return self.db.scalar(statement.execution_options(populate_existing=True))

    def _build_response_snapshot(
        self,
        disposition: RiskDisposition,
        event: RiskDispositionEvent,
        *,
        prior_version: int,
    ) -> RiskDispositionMutationRead:
        prior_events = list(
            self.db.scalars(
                select(RiskDispositionEvent)
                .where(
                    RiskDispositionEvent.tenant_id == disposition.tenant_id,
                    RiskDispositionEvent.disposition_id == disposition.id,
                    RiskDispositionEvent.version <= prior_version,
                )
                .order_by(RiskDispositionEvent.version.asc())
            ).all()
        )
        event_read = RiskDispositionEventRead.model_validate(event)
        event_history = [
            *[RiskDispositionEventRead.model_validate(item) for item in prior_events],
            event_read,
        ]
        disposition_read = RiskDispositionRead(
            id=disposition.id,
            tenant_id=disposition.tenant_id,
            observation_id=disposition.observation_id,
            adjudication=disposition.adjudication,
            lifecycle_status=disposition.lifecycle_status,
            version=disposition.version,
            assignee=disposition.assignee,
            due_at=disposition.due_at,
            resolution=disposition.resolution,
            created_at=disposition.created_at,
            updated_at=disposition.updated_at,
            events=event_history,
        )
        return RiskDispositionMutationRead(
            disposition=disposition_read,
            event=event_read,
            replayed=False,
        )

    def _find_replay(
        self,
        *,
        tenant_id: str,
        request_id: str,
        fingerprint: str,
    ) -> RiskDispositionMutation | None:
        event = self.db.scalar(
            select(RiskDispositionEvent).where(
                RiskDispositionEvent.tenant_id == tenant_id,
                RiskDispositionEvent.request_id == request_id,
            )
        )
        if event is None:
            return None
        if event.payload_fingerprint != fingerprint:
            raise RiskDispositionIdempotencyConflictError("request_id 已被不同的风险处置内容使用。")
        response = self._event_response(event)
        feedback = self.db.scalar(
            select(RiskObservationFeedback).where(
                RiskObservationFeedback.observation_id == event.observation_id,
                RiskObservationFeedback.action == f"disposition:{event.id}",
            )
        )
        return RiskDispositionMutation(
            response=response.model_copy(update={"replayed": True}),
            legacy_feedback=feedback,
        )

    def _event_response(
        self,
        event: RiskDispositionEvent,
    ) -> RiskDispositionMutationRead:
        if isinstance(event.response_json, dict) and event.response_json:
            try:
                response = RiskDispositionMutationRead.model_validate(event.response_json)
            except ValidationError as error:
                raise RiskDispositionConflictError(
                    "Stored risk disposition response snapshot is invalid."
                ) from error
            self._validate_snapshot_identity(event, response)
            return response
        return self._rebuild_legacy_event_response(event)

    def _rebuild_legacy_event_response(
        self,
        event: RiskDispositionEvent,
    ) -> RiskDispositionMutationRead:
        # 0012 之前的事件没有响应快照。只读取目标版本及以前的 append-only
        # 事件，并使用目标事件 after_json 重建；禁止读取当前 disposition。
        history = list(
            self.db.scalars(
                select(RiskDispositionEvent)
                .where(
                    RiskDispositionEvent.tenant_id == event.tenant_id,
                    RiskDispositionEvent.disposition_id == event.disposition_id,
                    RiskDispositionEvent.version <= event.version,
                )
                .order_by(RiskDispositionEvent.version.asc())
            ).all()
        )
        target = next((item for item in history if item.id == event.id), None)
        if target is None:
            raise RiskDispositionConflictError(
                "Legacy risk disposition replay event is missing from its audit chain."
            )
        state: dict[str, Any] = dict(event.after_json or {})
        event_reads = [RiskDispositionEventRead.model_validate(item) for item in history]
        target_read = RiskDispositionEventRead.model_validate(target)
        try:
            disposition = RiskDispositionRead(
                id=event.disposition_id,
                tenant_id=event.tenant_id,
                observation_id=event.observation_id,
                adjudication=state.get("adjudication"),
                lifecycle_status=state.get("lifecycle_status"),
                version=event.version,
                assignee=state.get("assignee"),
                due_at=state.get("due_at"),
                resolution=state.get("resolution"),
                created_at=history[0].created_at,
                updated_at=event.created_at,
                events=event_reads,
            )
            response = RiskDispositionMutationRead(
                disposition=disposition,
                event=target_read,
                replayed=False,
            )
        except ValidationError as error:
            raise RiskDispositionConflictError(
                "Legacy risk disposition response cannot be reconstructed safely."
            ) from error
        self._validate_snapshot_identity(event, response)
        return response

    @staticmethod
    def _validate_snapshot_identity(
        event: RiskDispositionEvent,
        response: RiskDispositionMutationRead,
    ) -> None:
        if (
            response.event.id != event.id
            or response.event.request_id != event.request_id
            or response.event.version != event.version
            or response.disposition.id != event.disposition_id
            or response.disposition.observation_id != event.observation_id
            or response.disposition.tenant_id != event.tenant_id
            or response.disposition.version != event.version
            or not any(item.id == event.id for item in response.disposition.events)
            or any(item.version > event.version for item in response.disposition.events)
        ):
            raise RiskDispositionConflictError(
                "Risk disposition response snapshot does not match its immutable event."
            )

    def _append_safe_feedback(
        self,
        observation: RiskObservation,
        event: RiskDispositionEvent,
        payload: RiskDispositionActionCreate,
        *,
        actor_name: str,
    ) -> RiskObservationFeedback | None:
        if payload.action not in {"confirm", "false_positive"}:
            return None
        feedback = RiskObservationFeedback(
            observation_id=observation.id,
            feedback_type=payload.action,
            action=f"disposition:{event.id}",
            actor=_text(actor_name) or "anonymous",
            comment=payload.comment,
            payload_json={
                "decision": payload.action,
                "source": "typed_risk_disposition",
            },
        )
        self.db.add(feedback)
        return feedback

    def _ingest_feedback_sample(
        self,
        observation: RiskObservation,
        feedback: RiskObservationFeedback,
    ) -> None:
        enabled = os.environ.get("FEW_SHOT_INJECTION_ENABLED", "true")
        if enabled.strip().lower() in {"0", "false", "no"}:
            return
        try:
            from app.services.few_shot_ingestion import FewShotIngestionService

            FewShotIngestionService(self.db).ingest_observation_feedback(
                observation,
                feedback,
            )
        except Exception:
            logger.exception(
                "few-shot ingestion failed for disposition event %s",
                feedback.action,
            )


def _apply_action(
    disposition: RiskDisposition,
    observation: RiskObservation,
    payload: RiskDispositionActionCreate,
) -> None:
    if payload.action == "confirm":
        disposition.adjudication = "confirmed"
        observation.status = "confirmed"
        observation.feedback_status = "confirmed"
        return
    if payload.action == "false_positive":
        disposition.adjudication = "false_positive"
        observation.status = "false_positive"
        observation.feedback_status = "false_positive"
        return

    lifecycle_by_action = {
        "request_supplement": "supplement_requested",
        "start_remediation": "remediation_in_progress",
        "request_waiver": "waiver_requested",
        "resolve": "resolved",
    }
    disposition.lifecycle_status = lifecycle_by_action[payload.action]
    if payload.assignee is not None:
        disposition.assignee = payload.assignee
    if payload.due_at is not None:
        disposition.due_at = payload.due_at
    if payload.action != "resolve":
        disposition.resolution = None
    else:
        disposition.resolution = payload.resolution
        observation.status = "resolved"


def _validate_transition(
    disposition: RiskDisposition,
    payload: RiskDispositionActionCreate,
) -> None:
    if disposition.lifecycle_status == "resolved":
        raise RiskDispositionConflictError("已解决的风险不能再次变更裁决或处置生命周期。")
    if payload.action == "confirm":
        if disposition.adjudication == "confirmed":
            raise RiskDispositionConflictError("该风险已经确认成立，请勿重复提交。")
        return
    if payload.action == "false_positive":
        if disposition.adjudication == "false_positive":
            raise RiskDispositionConflictError("该风险已经标记为误报，请勿重复提交。")
        return
    target_lifecycle = {
        "request_supplement": "supplement_requested",
        "start_remediation": "remediation_in_progress",
        "request_waiver": "waiver_requested",
        "resolve": "resolved",
    }[payload.action]
    if disposition.lifecycle_status == target_lifecycle:
        raise RiskDispositionConflictError("该风险已处于目标处置状态，请勿重复提交。")
    if payload.action == "request_supplement":
        return
    if disposition.adjudication != "confirmed":
        raise RiskDispositionConflictError("风险必须先确认成立，才能启动整改、申请豁免或标记解决。")


def _state(disposition: RiskDisposition) -> dict[str, object]:
    return {
        "adjudication": disposition.adjudication,
        "lifecycle_status": disposition.lifecycle_status,
        "assignee": disposition.assignee,
        "due_at": disposition.due_at.isoformat() if disposition.due_at else None,
        "resolution": disposition.resolution,
        "version": disposition.version,
    }


def _initial_adjudication(observation: RiskObservation) -> str:
    values = {
        _text(observation.feedback_status).lower(),
        _text(observation.status).lower(),
    }
    if "false_positive" in values:
        return "false_positive"
    if "confirmed" in values:
        return "confirmed"
    return "unreviewed"


def _initial_lifecycle_status(observation: RiskObservation) -> str:
    return "resolved" if _text(observation.status).lower() == "resolved" else "open"


def _payload_fingerprint(
    payload: RiskDispositionActionCreate,
    *,
    observation_id: str,
    actor_id: str,
) -> str:
    canonical_payload = payload.model_dump(
        mode="json",
        exclude={"request_id"},
        exclude_none=False,
    )
    canonical_payload["observation_id"] = observation_id
    canonical_payload["actor_id"] = actor_id
    canonical = json.dumps(
        canonical_payload,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def _tenant(value: str) -> str:
    return _text(value) or "default"


def _text(value: object) -> str:
    return str(value or "").strip()