chenxin
34c481ab9e
- 迁移 11：listing、order（六态 + 幂等唯一约束）、order_item（价格快照，客户端金额不可信）、ledger_entry（不可变 8 类型带符号分录）、webhook_event（Provider 事件 ID 幂等）、personal_domain.billing_status（保全态） - 购买闭环：下单冻结价格快照 → 支付确认同事务写三分录（charge = platform_fee 10% + creator_payable，双录平衡校验拒绝不平衡提交） - 退款：反向三条分录修正账本（Ledger 不删不改，ADR-009）；重复退款幂等；汇总端点对账 balanced=true - Webhook：HMAC-SHA256 验签（密钥经环境变量，错签 401）→ 按 Provider 事件 ID 幂等（重复投递 duplicate 不重复入账）→ 买方从订单反查（不信任 payload） - 保全态：欠费只读保全（全部状态变更 403 且明示不删数据；解除端点不被保全自锁）；读取不受影响，恢复即解锁——ADR-021 落地 - 测试 72 项全绿（+5）：双录平衡、订单幂等、退款反向与对账、验签/幂等、保全只读与恢复
273 lines
9.6 KiB
Python
273 lines
9.6 KiB
Python
"""交易域：商品、订单、不可变账本、退款与 Webhook（WP-13 / ADR-009/021）。"""
|
||
|
||
import hashlib
|
||
import hmac
|
||
import os
|
||
from datetime import UTC, datetime
|
||
from typing import Any
|
||
|
||
from fastapi import HTTPException, status
|
||
from sqlalchemy import func, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.models import (
|
||
AuditEvent,
|
||
LedgerEntry,
|
||
Listing,
|
||
Order,
|
||
OrderItem,
|
||
OutboxEvent,
|
||
WebhookEvent,
|
||
)
|
||
|
||
PLATFORM_FEE_BPS = 1000 # 平台费 10%（基点；正式费率由运营决策门登记）
|
||
|
||
|
||
def _ledger_balance_check(entries: list[tuple[str, int]]) -> None:
|
||
"""平衡不变量：sum(charge+reversal) == sum(fee+payable 及其 reversal)。"""
|
||
left = sum(a for t, a in entries if t in {"charge", "reversal"})
|
||
right = sum(
|
||
a for t, a in entries
|
||
if t in {"platform_fee", "creator_payable", "refund"}
|
||
)
|
||
if left != right:
|
||
raise HTTPException(
|
||
status.HTTP_500_INTERNAL_SERVER_ERROR, "账本不平衡：拒绝提交"
|
||
)
|
||
|
||
|
||
async def create_listing(
|
||
db: AsyncSession, domain_id: str, body: dict[str, Any], trace_id: str | None
|
||
) -> Listing:
|
||
listing = Listing(
|
||
creator_domain_id=domain_id,
|
||
title=str(body.get("title", "")),
|
||
price_minor=int(body.get("price_minor", 0)),
|
||
currency=str(body.get("currency", "CNY")),
|
||
)
|
||
if not listing.title.strip():
|
||
raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "title 不能为空")
|
||
db.add(listing)
|
||
await db.flush()
|
||
db.add(
|
||
AuditEvent(
|
||
actor_type="user", actor_id=domain_id, action="listing.created",
|
||
aggregate_type="listing", aggregate_id=listing.id, trace_id=trace_id,
|
||
)
|
||
)
|
||
await db.flush()
|
||
return listing
|
||
|
||
|
||
async def create_order(
|
||
db: AsyncSession, buyer_domain_id: str, body: dict[str, Any], trace_id: str | None
|
||
) -> tuple[Order, bool]:
|
||
"""下单：价格快照冻结（客户端金额不可信）；幂等键重放返回原订单。"""
|
||
key = body.get("idempotency_key")
|
||
if key:
|
||
existing = await db.execute(
|
||
select(Order).where(
|
||
Order.buyer_domain_id == buyer_domain_id,
|
||
Order.idempotency_key == key,
|
||
)
|
||
)
|
||
replay = existing.scalar_one_or_none()
|
||
if replay is not None:
|
||
return replay, True
|
||
listing_ids = body.get("listing_ids", [])
|
||
if not listing_ids:
|
||
raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "listing_ids 不能为空")
|
||
listings = list(
|
||
(
|
||
await db.execute(select(Listing).where(Listing.id.in_(listing_ids)))
|
||
).scalars().all()
|
||
)
|
||
if len(listings) != len(set(listing_ids)):
|
||
raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "存在无效商品")
|
||
total = sum(x.price_minor for x in listings)
|
||
currency = listings[0].currency
|
||
if any(x.currency != currency for x in listings):
|
||
raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "订单不能混合币种")
|
||
order = Order(
|
||
buyer_domain_id=buyer_domain_id,
|
||
total_minor=total,
|
||
currency=currency,
|
||
idempotency_key=key,
|
||
)
|
||
db.add(order)
|
||
await db.flush()
|
||
for listing in listings:
|
||
db.add(
|
||
OrderItem(
|
||
order_id=order.id,
|
||
listing_id=listing.id,
|
||
title_snapshot=listing.title,
|
||
price_minor_snapshot=listing.price_minor,
|
||
)
|
||
)
|
||
db.add(
|
||
AuditEvent(
|
||
actor_type="user", actor_id=buyer_domain_id, action="order.created",
|
||
aggregate_type="order", aggregate_id=order.id,
|
||
payload={"total_minor": total}, trace_id=trace_id,
|
||
)
|
||
)
|
||
await db.flush()
|
||
return order, False
|
||
|
||
|
||
async def _post_entries(
|
||
db: AsyncSession, order: Order, sign: int, note: str
|
||
) -> None:
|
||
"""写三分录（带符号）：charge / platform_fee / creator_payable。
|
||
|
||
charge=fee+payable（双录平衡）；退款以 sign=-1 写反向三条。
|
||
"""
|
||
fee = order.total_minor * PLATFORM_FEE_BPS // 10000
|
||
payable = order.total_minor - fee
|
||
entries = [
|
||
("charge" if sign > 0 else "reversal", order.total_minor * sign),
|
||
("platform_fee", fee * sign),
|
||
("creator_payable", payable * sign),
|
||
]
|
||
checked = [(t, a) for t, a in entries if a != 0]
|
||
_ledger_balance_check(checked)
|
||
for entry_type, amount in checked:
|
||
db.add(
|
||
LedgerEntry(
|
||
order_id=order.id, entry_type=entry_type,
|
||
amount_minor=amount, currency=order.currency,
|
||
note=note,
|
||
)
|
||
)
|
||
|
||
|
||
async def confirm_payment(
|
||
db: AsyncSession, buyer_domain_id: str, order_id: str, trace_id: str | None
|
||
) -> Order:
|
||
"""支付确认（开发档 Provider；真实 Provider 经 Webhook 驱动同一入账）。"""
|
||
order = await db.get(Order, order_id)
|
||
if order is None or order.buyer_domain_id != buyer_domain_id:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "订单不存在")
|
||
if order.status == "paid":
|
||
return order # 幂等
|
||
if order.status != "pending_payment":
|
||
raise HTTPException(status.HTTP_409_CONFLICT, f"订单状态 {order.status} 不可支付")
|
||
order.status = "paid"
|
||
order.paid_at = datetime.now(UTC)
|
||
await _post_entries(db, order, +1, "payment confirmed")
|
||
db.add(
|
||
OutboxEvent(
|
||
aggregate_type="order", aggregate_id=order.id,
|
||
event_type="order.paid", payload={"order_id": order.id},
|
||
)
|
||
)
|
||
db.add(
|
||
AuditEvent(
|
||
actor_type="user", actor_id=buyer_domain_id, action="order.paid",
|
||
aggregate_type="order", aggregate_id=order.id, trace_id=trace_id,
|
||
)
|
||
)
|
||
await db.flush()
|
||
return order
|
||
|
||
|
||
async def refund_order(
|
||
db: AsyncSession, buyer_domain_id: str, order_id: str, trace_id: str | None
|
||
) -> Order:
|
||
"""全额退款：反向分录修正账本（Ledger 不可变，不删不改，ADR-009）。"""
|
||
order = await db.get(Order, order_id)
|
||
if order is None or order.buyer_domain_id != buyer_domain_id:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "订单不存在")
|
||
if order.status == "refunded":
|
||
return order # 幂等
|
||
if order.status != "paid":
|
||
raise HTTPException(status.HTTP_409_CONFLICT, "仅已支付订单可退款")
|
||
order.status = "refunded"
|
||
order.refunded_at = datetime.now(UTC)
|
||
await _post_entries(db, order, -1, "refund")
|
||
db.add(
|
||
OutboxEvent(
|
||
aggregate_type="order", aggregate_id=order.id,
|
||
event_type="order.refunded", payload={"order_id": order.id},
|
||
)
|
||
)
|
||
db.add(
|
||
AuditEvent(
|
||
actor_type="user", actor_id=buyer_domain_id, action="order.refunded",
|
||
aggregate_type="order", aggregate_id=order.id, trace_id=trace_id,
|
||
)
|
||
)
|
||
await db.flush()
|
||
return order
|
||
|
||
|
||
def verify_webhook_signature(raw_body: bytes, signature: str | None) -> None:
|
||
"""HMAC-SHA256 验签：密钥经环境变量；失败 401（契约 realtime-and-webhooks）。"""
|
||
secret = os.environ.get("HUNJING_WEBHOOK_SECRET", "")
|
||
if not secret:
|
||
raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, "Webhook 密钥未配置")
|
||
if signature is None:
|
||
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "缺少签名")
|
||
expected = hmac.new(secret.encode(), raw_body, hashlib.sha256).hexdigest()
|
||
if not hmac.compare_digest(expected, signature):
|
||
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "签名校验失败")
|
||
|
||
|
||
async def process_webhook(
|
||
db: AsyncSession, provider: str, provider_event_id: str, event_type: str,
|
||
payload_hash: str, handler: Any, trace_id: str | None,
|
||
) -> str:
|
||
"""Webhook 处理：先验签（调用方）、按事件 ID 幂等；重复事件 200 跳过。"""
|
||
existing = await db.get(WebhookEvent, provider_event_id)
|
||
if existing is not None:
|
||
return "duplicate" # 幂等：重复投递不重复入账
|
||
db.add(
|
||
WebhookEvent(
|
||
provider_event_id=provider_event_id,
|
||
provider=provider,
|
||
event_type=event_type,
|
||
payload_hash=payload_hash,
|
||
)
|
||
)
|
||
await handler() # 业务处理（如入账）与本记录同一事务
|
||
await db.flush()
|
||
return "processed"
|
||
|
||
|
||
async def finance_summary(db: AsyncSession, domain_id: str) -> dict[str, Any]:
|
||
"""域内财务汇总与平衡校验（每订单 charge=reversal 与 fee+payable 对账）。"""
|
||
rows = (
|
||
await db.execute(
|
||
select(LedgerEntry.entry_type, func.sum(LedgerEntry.amount_minor))
|
||
.join(Order, LedgerEntry.order_id == Order.id)
|
||
.where(Order.buyer_domain_id == domain_id)
|
||
.group_by(LedgerEntry.entry_type)
|
||
)
|
||
).all()
|
||
totals = {t: int(a or 0) for t, a in rows}
|
||
balance_ok = totals.get("charge", 0) + totals.get("reversal", 0) == (
|
||
totals.get("platform_fee", 0) + totals.get("creator_payable", 0)
|
||
)
|
||
return {"totals_minor": totals, "balanced": balance_ok}
|
||
|
||
|
||
async def set_preservation(
|
||
db: AsyncSession, domain_id: str, preserved: bool, trace_id: str | None
|
||
) -> str:
|
||
"""保全态（ADR-021）：欠费只读保全；恢复付费即恢复，数据不删不降级。"""
|
||
from app.models import PersonalDomain
|
||
|
||
domain = await db.get(PersonalDomain, domain_id)
|
||
assert domain is not None
|
||
domain.billing_status = "preserved" if preserved else "ok"
|
||
db.add(
|
||
AuditEvent(
|
||
actor_type="system", actor_id="billing", action="domain.preservation_changed",
|
||
aggregate_type="personal_domain", aggregate_id=domain_id,
|
||
payload={"preserved": preserved}, trace_id=trace_id,
|
||
)
|
||
)
|
||
await db.flush()
|
||
return domain.billing_status
|