harness_zbt/backend/app/domains/enrollment/service.py

import hmac
import json
from collections.abc import Callable
from dataclasses import replace
from datetime import datetime, timedelta
from hashlib import sha256
from secrets import token_urlsafe
from typing import Any, Protocol

from app.core.errors import AppError
from app.core.identifiers import new_ulid
from app.domains.catalog.service import ProductCatalogService
from app.domains.enrollment.models import (
    AsyncTask,
    EnrollmentDraft,
    EnrollmentOrder,
    OutboxEvent,
    PaymentTransaction,
    Policy,
    Quote,
    UserConfirmation,
)
from app.domains.enrollment.repository import EnrollmentRepository
from app.domains.enrollment.tasks import InlinePolicyTaskDispatcher, TaskDispatcher
from app.domains.identity.models import H5User


class OrderAttributionRecorder(Protocol):
    def record_order(self, order: EnrollmentOrder) -> None: ...


class EnrollmentService:
    def __init__(
        self,
        repository: EnrollmentRepository,
        catalog_service: ProductCatalogService,
        clock: Callable[[], datetime],
        callback_secret: str,
        attribution_recorder: OrderAttributionRecorder | None = None,
        task_dispatcher: TaskDispatcher | None = None,
    ) -> None:
        self._repository = repository
        self._catalog = catalog_service
        self._clock = clock
        self._callback_secret = callback_secret.encode("utf-8")
        self._attribution_recorder = attribution_recorder
        self._task_dispatcher = task_dispatcher or InlinePolicyTaskDispatcher(
            repository,
            clock,
        )

    def create_quote(
        self,
        user: H5User,
        *,
        product_id: str,
        plan_id: str,
        age: int,
        region_code: str,
        occupation_code: str,
        relationship: str,
    ) -> dict[str, Any]:
        product = next(
            (item for item in self._catalog.list_available() if item["product_id"] == product_id),
            None,
        )
        if product is None:
            raise AppError("PRODUCT_NOT_AVAILABLE", "产品当前不可投保", 404)
        plan = next(
            (item for item in product["plans"] if item["id"] == plan_id),
            None,
        )
        if plan is None:
            raise AppError("PLAN_NOT_AVAILABLE", "保障计划当前不可投保", 404)

        premium = self._deterministic_premium(
            str(product["product_code"]),
            str(plan["code"]),
            age,
            region_code,
            occupation_code,
            configured_premium=int(plan.get("premium_cents", 0)),
            configured_min_age=int(plan.get("min_age", 0)),
            configured_max_age=int(plan.get("max_age", 100)),
        )
        now = self._clock()
        quote = Quote(
            id=new_ulid(),
            user_id=user.id,
            product_id=product_id,
            product_version_id=str(product["product_version_id"]),
            plan_id=plan_id,
            insured_age=age,
            insured_region_code=region_code,
            occupation_code=occupation_code,
            relationship=relationship,
            premium_cents=premium,
            currency="CNY",
            rule_version="eligibility-v1",
            rate_version="rate-v1",
            status="VALID",
            expires_at=now + timedelta(minutes=30),
            created_at=now,
        )
        self._repository.save_quote(quote)
        return {
            "eligible": True,
            "eligibility_reasons": [],
            "quote_id": quote.id,
            "product_id": quote.product_id,
            "product_version_id": quote.product_version_id,
            "plan_id": quote.plan_id,
            "premium_cents": quote.premium_cents,
            "currency": quote.currency,
            "rule_version": quote.rule_version,
            "rate_version": quote.rate_version,
            "expires_at": quote.expires_at,
        }

    def create_draft(
        self,
        user: H5User,
        *,
        quote_id: str,
        applicant: dict[str, Any],
        insured: dict[str, Any],
        contact: dict[str, Any],
    ) -> dict[str, Any]:
        quote = self._repository.get_quote(quote_id)
        now = self._clock()
        if (
            quote is None
            or quote.user_id != user.id
            or quote.status != "VALID"
            or quote.expires_at <= now
        ):
            raise AppError("QUOTE_NOT_AVAILABLE", "报价不存在或已失效", 410)
        draft = EnrollmentDraft(
            id=new_ulid(),
            user_id=user.id,
            quote_id=quote.id,
            applicant=applicant,
            insured=insured,
            contact=contact,
            status="EDITABLE",
            expires_at=now + timedelta(hours=24),
            created_at=now,
        )
        self._repository.save_draft(draft)
        return {
            "draft_id": draft.id,
            "quote_id": draft.quote_id,
            "status": draft.status,
            "expires_at": draft.expires_at,
        }

    def confirm_draft(self, user: H5User, draft_id: str) -> dict[str, Any]:
        draft = self._repository.get_draft(draft_id)
        now = self._clock()
        if (
            draft is None
            or draft.user_id != user.id
            or draft.status != "EDITABLE"
            or draft.expires_at <= now
        ):
            raise AppError("DRAFT_NOT_AVAILABLE", "投保草稿不存在或不可确认", 409)
        raw_token = token_urlsafe(32)
        confirmation = UserConfirmation(
            id=new_ulid(),
            user_id=user.id,
            draft_id=draft.id,
            token_hash=self._hash_token(raw_token),
            status="PENDING",
            expires_at=now + timedelta(minutes=10),
            created_at=now,
        )
        self._repository.save_confirmation(confirmation)
        return {
            "confirmation_id": confirmation.id,
            "confirmation_token": raw_token,
            "expires_at": confirmation.expires_at,
        }

    def create_order(
        self,
        user: H5User,
        *,
        draft_id: str,
        confirmation_token: str,
        idempotency_key: str,
    ) -> tuple[dict[str, Any], bool]:
        existing = self._repository.find_order_by_idempotency(user.id, idempotency_key)
        if existing is not None:
            if existing.draft_id != draft_id:
                raise AppError(
                    "IDEMPOTENCY_CONFLICT",
                    "相同幂等键已用于其他投保请求",
                    409,
                )
            return self._order_data(existing), False

        now = self._clock()
        draft = self._repository.get_draft(draft_id)
        confirmation = self._repository.get_confirmation_by_token_hash(
            self._hash_token(confirmation_token)
        )
        if draft is None or draft.user_id != user.id:
            raise AppError("DRAFT_NOT_AVAILABLE", "投保草稿不存在", 404)
        if (
            confirmation is None
            or confirmation.user_id != user.id
            or confirmation.draft_id != draft.id
            or confirmation.status != "PENDING"
            or confirmation.expires_at <= now
        ):
            raise AppError(
                "CONFIRMATION_REQUIRED",
                "创建订单前需要有效的用户确认",
                409,
            )
        quote = self._repository.get_quote(draft.quote_id)
        if quote is None or quote.expires_at <= now:
            raise AppError("QUOTE_EXPIRED", "报价已过期", 410)
        order = EnrollmentOrder(
            id=new_ulid(),
            order_no=f"ORD-{now:%Y%m%d}-{new_ulid()[-8:]}",
            user_id=user.id,
            quote_id=quote.id,
            draft_id=draft.id,
            confirmation_id=confirmation.id,
            idempotency_key=idempotency_key,
            product_id=quote.product_id,
            product_version_id=quote.product_version_id,
            plan_id=quote.plan_id,
            applicant_snapshot=draft.applicant,
            insured_snapshot=draft.insured,
            amount_cents=quote.premium_cents,
            currency=quote.currency,
            status="PENDING_PAYMENT",
            created_at=now,
        )
        self._repository.save_order(order)
        if self._attribution_recorder is not None:
            self._attribution_recorder.record_order(order)
        return self._order_data(order), True

    def create_payment(
        self,
        user: H5User,
        *,
        order_id: str,
        idempotency_key: str,
    ) -> tuple[dict[str, Any], bool]:
        order = self._repository.get_order(order_id)
        if order is None or order.user_id != user.id:
            raise AppError("ORDER_NOT_FOUND", "未找到订单", 404)
        if order.status != "PENDING_PAYMENT":
            raise AppError("ORDER_NOT_PAYABLE", "订单当前不可支付", 409)
        existing = self._repository.find_payment_by_idempotency(order.id, idempotency_key)
        if existing is not None:
            return self._payment_data(existing), False
        now = self._clock()
        payment = PaymentTransaction(
            id=new_ulid(),
            payment_no=f"PAY-{now:%Y%m%d}-{new_ulid()[-8:]}",
            order_id=order.id,
            user_id=user.id,
            transaction_type="PAYMENT",
            provider="LOCAL_MOCK_PAY",
            idempotency_key=idempotency_key,
            amount_cents=order.amount_cents,
            currency=order.currency,
            status="CREATED",
            provider_transaction_no=None,
            succeeded_at=None,
            created_at=now,
        )
        self._repository.save_payment(payment)
        return self._payment_data(payment), True

    def complete_mock_payment(
        self,
        user: H5User,
        *,
        payment_id: str,
    ) -> dict[str, Any]:
        payment = self._repository.get_payment(payment_id)
        if payment is None or payment.user_id != user.id:
            raise AppError("PAYMENT_NOT_FOUND", "未找到支付流水", 404)
        now = self._clock()
        payload = {
            "callback_no": f"CB-{now:%Y%m%d}-{payment.id[-8:]}",
            "provider_transaction_no": f"MP-{now:%Y%m%d}-{payment.id[-8:]}",
            "payment_no": payment.payment_no,
            "status": "SUCCEEDED",
            "amount_cents": payment.amount_cents,
            "occurred_at": now.isoformat(),
        }
        signature = self.sign_mock_callback(payload)
        return self.handle_mock_callback(payload, signature)

    def handle_mock_callback(
        self,
        payload: dict[str, Any],
        signature: str,
    ) -> dict[str, Any]:
        expected_signature = self.sign_mock_callback(payload)
        if not hmac.compare_digest(signature, expected_signature):
            raise AppError("INVALID_CALLBACK_SIGNATURE", "支付回调签名无效", 401)
        payment = self._repository.get_payment_by_no(str(payload["payment_no"]))
        if payment is None:
            raise AppError("PAYMENT_NOT_FOUND", "未找到支付流水", 404)
        if payload["status"] != "SUCCEEDED":
            raise AppError("PAYMENT_CALLBACK_REJECTED", "支付结果不是成功状态", 409)
        if int(payload["amount_cents"]) != payment.amount_cents:
            raise AppError("PAYMENT_AMOUNT_MISMATCH", "支付回调金额不一致", 409)
        order = self._repository.get_order(payment.order_id)
        if order is None:
            raise AppError("ORDER_NOT_FOUND", "未找到支付对应订单", 404)
        existing_policy = self._repository.get_policy_by_order(order.id)
        if payment.status == "SUCCEEDED" and existing_policy is not None:
            return self._payment_completion_data(payment, order, existing_policy)

        now = self._clock()
        succeeded_payment = replace(
            payment,
            status="SUCCEEDED",
            provider_transaction_no=str(payload["provider_transaction_no"]),
            succeeded_at=now,
        )
        paid_order = replace(order, status="PAID")
        self._repository.save_payment(succeeded_payment)
        self._repository.save_order(paid_order)

        event = OutboxEvent(
            id=new_ulid(),
            aggregate_type="PAYMENT",
            aggregate_id=payment.id,
            event_type="payment.succeeded",
            payload={"payment_id": payment.id, "order_id": order.id},
            status="PENDING",
            created_at=now,
        )
        task_key = f"policy.issue:{order.id}"
        task = self._repository.get_task_by_idempotency(task_key) or AsyncTask(
            id=new_ulid(),
            task_type="policy.issue",
            business_key=order.id,
            idempotency_key=task_key,
            payload={"order_id": order.id},
            status="PENDING",
            attempt_count=0,
            created_at=now,
        )
        self._repository.save_outbox_event(event)
        self._repository.save_task(task)
        policy = existing_policy or self._task_dispatcher.dispatch(task)
        self._repository.save_outbox_event(replace(event, status="DISPATCHED"))
        issued_order = self._repository.get_order(order.id)
        if issued_order is None:
            raise AppError("ORDER_NOT_FOUND", "保单签发后未找到订单", 500)
        return self._payment_completion_data(succeeded_payment, issued_order, policy)

    def sign_mock_callback(self, payload: dict[str, Any]) -> str:
        canonical = json.dumps(
            payload,
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
        ).encode("utf-8")
        return hmac.new(
            self._callback_secret,
            canonical,
            digestmod=sha256,
        ).hexdigest()

    def list_policies(self, user: H5User) -> dict[str, Any]:
        policies = self._repository.list_policies(user.id)
        items = [self._policy_data(policy) for policy in policies]
        return {"items": items, "total": len(items)}

    def list_all_policies(self) -> dict[str, Any]:
        policies = self._repository.list_policies()
        items = [self._policy_data(policy) for policy in policies]
        return {"items": items, "total": len(items)}

    def list_orders(self, user_id: str | None = None) -> dict[str, Any]:
        orders = self._repository.list_orders(user_id)
        items = [self._order_data(order) for order in orders]
        return {"items": items, "total": len(items)}

    @staticmethod
    def _hash_token(token: str) -> str:
        return sha256(token.encode("utf-8")).hexdigest()

    def _order_data(self, order: EnrollmentOrder) -> dict[str, Any]:
        draft = self._repository.get_draft(order.draft_id)
        quote = self._repository.get_quote(order.quote_id)
        payment = self._repository.get_payment_by_order(order.id)
        policy = self._repository.get_policy_by_order(order.id)
        product = next(
            (
                item
                for item in self._catalog.list_available()
                if item["product_id"] == order.product_id
            ),
            None,
        )
        plan = (
            next(
                (item for item in product["plans"] if item["id"] == order.plan_id),
                None,
            )
            if product is not None
            else None
        )
        return {
            "order_id": order.id,
            "order_no": order.order_no,
            "user_id": order.user_id,
            "product_name": product["name"] if product else "智保通保障计划",
            "plan_name": plan["name"] if plan else "保障计划",
            "relationship": quote.relationship if quote else "SELF",
            "applicant": order.applicant_snapshot,
            "insured": order.insured_snapshot,
            "contact": draft.contact if draft else {},
            "status": order.status,
            "amount_cents": order.amount_cents,
            "currency": order.currency,
            "payment": (
                {
                    "payment_id": payment.id,
                    "payment_no": payment.payment_no,
                    "provider": payment.provider,
                    "status": payment.status,
                    "provider_transaction_no": payment.provider_transaction_no,
                    "succeeded_at": payment.succeeded_at,
                }
                if payment
                else None
            ),
            "policy_no": policy.policy_no if policy else None,
            "created_at": order.created_at,
        }

    @staticmethod
    def _payment_data(payment: PaymentTransaction) -> dict[str, Any]:
        return {
            "payment_id": payment.id,
            "payment_no": payment.payment_no,
            "order_id": payment.order_id,
            "transaction_type": payment.transaction_type,
            "status": payment.status,
            "amount_cents": payment.amount_cents,
            "currency": payment.currency,
        }

    def _policy_data(self, policy: Policy) -> dict[str, Any]:
        order = self._repository.get_order(policy.order_id)
        quote = self._repository.get_quote(order.quote_id) if order else None
        product = next(
            (
                item
                for item in self._catalog.list_available()
                if item["product_id"] == policy.product_id
            ),
            None,
        )
        plan = (
            next(
                (item for item in product["plans"] if item["id"] == policy.plan_id),
                None,
            )
            if product is not None
            else None
        )
        return {
            "policy_id": policy.id,
            "policy_no": policy.policy_no,
            "order_id": policy.order_id,
            "order_no": order.order_no if order else "",
            "product_name": product["name"] if product else "智保通保障计划",
            "plan_name": plan["name"] if plan else "保障计划",
            "relationship": quote.relationship if quote else "SELF",
            "applicant": order.applicant_snapshot if order else {},
            "insured": order.insured_snapshot if order else {},
            "status": policy.status,
            "premium_cents": policy.premium_cents,
            "currency": policy.currency,
            "coverage_start": policy.coverage_start,
            "coverage_end": policy.coverage_end,
            "issued_at": policy.issued_at,
        }

    def _payment_completion_data(
        self,
        payment: PaymentTransaction,
        order: EnrollmentOrder,
        policy: Policy,
    ) -> dict[str, Any]:
        return {
            "payment_id": payment.id,
            "payment_status": payment.status,
            "order_id": order.id,
            "order_status": order.status,
            "policy": self._policy_data(policy),
        }

    @staticmethod
    def _deterministic_premium(
        product_code: str,
        plan_code: str,
        age: int,
        region_code: str,
        occupation_code: str,
        *,
        configured_premium: int = 0,
        configured_min_age: int = 0,
        configured_max_age: int = 100,
    ) -> int:
        if region_code != "510100":
            raise AppError(
                "ELIGIBILITY_REJECTED",
                "当前产品仅支持成都地区",
                422,
                details={"reason_code": "REGION_NOT_SUPPORTED"},
            )
        if occupation_code != "GENERAL":
            raise AppError(
                "ELIGIBILITY_REJECTED",
                "当前职业类别不在可投保范围",
                422,
                details={"reason_code": "OCCUPATION_NOT_SUPPORTED"},
            )
        rates: dict[tuple[str, str], tuple[int, int, int]] = {
            ("MED-SENIOR", "SENIOR_STANDARD"): (50, 75, 19900),
            ("MED-BASIC", "BASIC"): (0, 65, 23900),
            ("MED-UPGRADE", "STANDARD"): (0, 65, 39900),
            ("MED-UPGRADE", "ENHANCED"): (0, 65, 59900),
            ("ACC-FAMILY", "INDIVIDUAL"): (0, 70, 9900),
            ("ACC-FAMILY", "FAMILY"): (0, 70, 15900),
        }
        rule = rates.get((product_code, plan_code))
        min_age = configured_min_age if configured_premium > 0 else (rule[0] if rule else 0)
        max_age = configured_max_age if configured_premium > 0 else (rule[1] if rule else -1)
        if (configured_premium <= 0 and rule is None) or not min_age <= age <= max_age:
            raise AppError(
                "ELIGIBILITY_REJECTED",
                "被保人年龄不在该计划可投保范围",
                422,
                details={"reason_code": "AGE_OUT_OF_RANGE"},
            )
        if configured_premium > 0:
            return configured_premium
        if rule is None:
            raise AppError("RATE_NOT_FOUND", "未找到该保障计划的费率", 422)
        return rule[2]