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]