From 34c481ab9e3e917beeba1941b604c99c13cc6fbf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E6=AD=86?= <2685475410@qq.com> Date: Fri, 11 Sep 2026 11:36:04 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20feat:=20WP-13=20=E5=B8=82=E5=9C=BA?= =?UTF-8?q?=E4=B8=8E=E4=BA=A4=E6=98=93=E2=80=94=E2=80=94=E4=B8=8D=E5=8F=AF?= =?UTF-8?q?=E5=8F=98=E8=B4=A6=E6=9C=AC=E3=80=81=E9=80=80=E6=AC=BE=E3=80=81?= =?UTF-8?q?Webhook=20=E4=B8=8E=E4=BF=9D=E5=85=A8=E6=80=81=EF=BC=88ADR-009/?= =?UTF-8?q?021=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 迁移 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）：双录平衡、订单幂等、退款反向与对账、验签/幂等、保全只读与恢复 --- ...32451_市场交易_商品_订单_账本与_webhook.py | 107 +++++++ backend/app/auth/dependencies.py | 20 +- backend/app/main.py | 5 + backend/app/marketplace/__init__.py | 1 + backend/app/marketplace/routes.py | 169 +++++++++++ backend/app/marketplace/service.py | 272 ++++++++++++++++++ backend/app/models.py | 123 ++++++++ backend/tests/test_conversations.py | 128 +++++++++ 8 files changed, 823 insertions(+), 2 deletions(-) create mode 100644 backend/alembic/versions/b7f1a0032451_市场交易_商品_订单_账本与_webhook.py create mode 100644 backend/app/marketplace/__init__.py create mode 100644 backend/app/marketplace/routes.py create mode 100644 backend/app/marketplace/service.py diff --git a/backend/alembic/versions/b7f1a0032451_市场交易_商品_订单_账本与_webhook.py b/backend/alembic/versions/b7f1a0032451_市场交易_商品_订单_账本与_webhook.py new file mode 100644 index 0000000..8f4355b --- /dev/null +++ b/backend/alembic/versions/b7f1a0032451_市场交易_商品_订单_账本与_webhook.py @@ -0,0 +1,107 @@ +"""市场交易：商品、订单、账本与 Webhook + +Revision ID: b7f1a0032451 +Revises: b3a26cb61923 +Create Date: 2026-09-11 11:31:22.345265 + +""" +from collections.abc import Sequence + +import sqlalchemy as sa + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = 'b7f1a0032451' +down_revision: str | Sequence[str] | None = 'b3a26cb61923' +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('webhook_event', + sa.Column('provider_event_id', sa.String(), nullable=False), + sa.Column('provider', sa.String(), nullable=False), + sa.Column('event_type', sa.String(), nullable=False), + sa.Column('payload_hash', sa.String(), nullable=False), + sa.Column('status', sa.String(), nullable=False), + sa.Column('received_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.CheckConstraint("status IN ('processed', 'ignored')", name=op.f('ck_webhook_event_status_values')), + sa.PrimaryKeyConstraint('provider_event_id', name=op.f('pk_webhook_event')) + ) + op.create_table('listing', + sa.Column('id', sa.String(), nullable=False), + sa.Column('creator_domain_id', sa.String(), nullable=False), + sa.Column('title', sa.String(), nullable=False), + sa.Column('price_minor', sa.Integer(), nullable=False), + sa.Column('currency', sa.String(), nullable=False), + sa.Column('status', sa.String(), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.CheckConstraint('price_minor >= 0', name=op.f('ck_listing_price_non_negative')), + sa.ForeignKeyConstraint(['creator_domain_id'], ['personal_domain.id'], name=op.f('fk_listing_personal_domain'), ondelete='cascade'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_listing')) + ) + op.create_index(op.f('ix_listing_creator_domain_id'), 'listing', ['creator_domain_id'], unique=False) + op.create_table('order', + sa.Column('id', sa.String(), nullable=False), + sa.Column('buyer_domain_id', sa.String(), nullable=False), + sa.Column('status', sa.String(), nullable=False), + sa.Column('total_minor', sa.Integer(), nullable=False), + sa.Column('currency', sa.String(), nullable=False), + sa.Column('idempotency_key', sa.String(), nullable=True), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.Column('paid_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('refunded_at', sa.DateTime(timezone=True), nullable=True), + sa.CheckConstraint("status IN ('pending_payment', 'paid', 'partially_refunded', 'refunded', 'disputed', 'cancelled')", name=op.f('ck_order_status_values')), + sa.ForeignKeyConstraint(['buyer_domain_id'], ['personal_domain.id'], name=op.f('fk_order_personal_domain'), ondelete='cascade'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_order')) + ) + op.create_index(op.f('ix_order_buyer_domain_id'), 'order', ['buyer_domain_id'], unique=False) + op.create_index('uq_order_buyer_idempotency', 'order', ['buyer_domain_id', 'idempotency_key'], unique=True) + op.create_table('ledger_entry', + sa.Column('id', sa.String(), nullable=False), + sa.Column('order_id', sa.String(), nullable=False), + sa.Column('entry_type', sa.String(), nullable=False), + sa.Column('amount_minor', sa.Integer(), nullable=False), + sa.Column('currency', sa.String(), nullable=False), + sa.Column('note', sa.String(), nullable=True), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.CheckConstraint("entry_type IN ('charge', 'refund', 'platform_fee', 'creator_payable', 'tax', 'reserve', 'payout', 'reversal')", name=op.f('ck_ledger_entry_entry_type_values')), + sa.ForeignKeyConstraint(['order_id'], ['order.id'], name=op.f('fk_ledger_entry_order'), ondelete='restrict'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_ledger_entry')) + ) + op.create_index(op.f('ix_ledger_entry_order_id'), 'ledger_entry', ['order_id'], unique=False) + op.create_index('ix_ledger_order_type', 'ledger_entry', ['order_id', 'entry_type'], unique=False) + op.create_table('order_item', + sa.Column('id', sa.String(), nullable=False), + sa.Column('order_id', sa.String(), nullable=False), + sa.Column('listing_id', sa.String(), nullable=False), + sa.Column('title_snapshot', sa.String(), nullable=False), + sa.Column('price_minor_snapshot', sa.Integer(), nullable=False), + sa.ForeignKeyConstraint(['listing_id'], ['listing.id'], name=op.f('fk_order_item_listing'), ondelete='restrict'), + sa.ForeignKeyConstraint(['order_id'], ['order.id'], name=op.f('fk_order_item_order'), ondelete='cascade'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_order_item')) + ) + op.create_index(op.f('ix_order_item_order_id'), 'order_item', ['order_id'], unique=False) + op.add_column('personal_domain', sa.Column('billing_status', sa.String(), nullable=False, server_default='ok')) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_column('personal_domain', 'billing_status') + op.drop_index(op.f('ix_order_item_order_id'), table_name='order_item') + op.drop_table('order_item') + op.drop_index('ix_ledger_order_type', table_name='ledger_entry') + op.drop_index(op.f('ix_ledger_entry_order_id'), table_name='ledger_entry') + op.drop_table('ledger_entry') + op.drop_index('uq_order_buyer_idempotency', table_name='order') + op.drop_index(op.f('ix_order_buyer_domain_id'), table_name='order') + op.drop_table('order') + op.drop_index(op.f('ix_listing_creator_domain_id'), table_name='listing') + op.drop_table('listing') + op.drop_table('webhook_event') + # ### end Alembic commands ### diff --git a/backend/app/auth/dependencies.py b/backend/app/auth/dependencies.py index d6fec3a..74e8317 100644 --- a/backend/app/auth/dependencies.py +++ b/backend/app/auth/dependencies.py @@ -6,7 +6,7 @@ from fastapi import Depends, Header, HTTPException, Request, status from sqlalchemy.ext.asyncio import AsyncSession from app.auth.service import SESSION_COOKIE, resolve_session -from app.db.session import get_db +from app.db.session import DbSession, get_db from app.models import User ALLOWED_ORIGINS: set[str] = set() # 由配置注入；空集表示本地开发放行同源 @@ -33,10 +33,11 @@ async def get_current_user( async def require_state_change( request: Request, + db: DbSession, user: Annotated[User, Depends(get_current_user)], hunjing_csrf_token: Annotated[str | None, Header(alias="Hunjing-CSRF-Token")] = None, ) -> User: - """Cookie 状态变更：Origin 校验 + CSRF 双提交（契约 2.1 节）。""" + """Cookie 状态变更：Origin 校验 + CSRF 双提交 + 保全态拦截（契约 2.1 / ADR-021）。""" if ALLOWED_ORIGINS: origin = request.headers.get("Origin") if origin is not None and origin not in ALLOWED_ORIGINS: @@ -44,6 +45,21 @@ async def require_state_change( csrf_hash = getattr(request.state, "session_csrf_hash", None) if csrf_hash is None or not _csrf_matches(csrf_hash, hunjing_csrf_token): raise HTTPException(status.HTTP_403_FORBIDDEN, "CSRF 校验失败") + # 保全态：欠费只读保全，状态变更拒绝（数据不删、恢复即解锁） + from sqlalchemy import select + + from app.models import PersonalDomain + + domain = ( + await db.execute( + select(PersonalDomain).where(PersonalDomain.user_id == user.id).limit(1) + ) + ).scalar_one_or_none() + if domain is not None and domain.billing_status == "preserved": + raise HTTPException( + status.HTTP_403_FORBIDDEN, + "账户处于保全态（只读）：恢复付费后自动解锁，数据不会删除", + ) return user diff --git a/backend/app/main.py b/backend/app/main.py index 9353fdb..3e6254d 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -10,6 +10,8 @@ from app.core.config import get_settings from app.core.errors import install_error_handlers from app.core.middleware import RequestContextMiddleware, configure_logging from app.db.session import init_engine +from app.marketplace.routes import finance_router, webhook_router +from app.marketplace.routes import router as marketplace_router from app.memories.routes import cycle_router from app.memories.routes import router as memories_router from app.tools.routes import calls_router, tools_router @@ -37,6 +39,9 @@ def create_app() -> FastAPI: app.include_router(memories_router) app.include_router(tools_router) app.include_router(calls_router) + app.include_router(marketplace_router) + app.include_router(finance_router) + app.include_router(webhook_router) @app.get("/healthz", tags=["ops"]) async def healthz() -> dict[str, str]: diff --git a/backend/app/marketplace/__init__.py b/backend/app/marketplace/__init__.py new file mode 100644 index 0000000..0ca3448 --- /dev/null +++ b/backend/app/marketplace/__init__.py @@ -0,0 +1 @@ +"""交易域：商品、订单、账本、退款与 Webhook。""" diff --git a/backend/app/marketplace/routes.py b/backend/app/marketplace/routes.py new file mode 100644 index 0000000..eba2c11 --- /dev/null +++ b/backend/app/marketplace/routes.py @@ -0,0 +1,169 @@ +"""交易域路由（WP-13）。""" + +import hashlib +import json +from typing import Annotated, Any + +from fastapi import APIRouter, Depends, Header, HTTPException, Request, status +from fastapi.responses import JSONResponse +from sqlalchemy import select + +from app.auth.dependencies import get_current_user, require_state_change +from app.conversations.routes import _domain_of +from app.db.session import DbSession +from app.marketplace import service +from app.models import Listing, Order, User + +router = APIRouter(prefix="/v1/marketplace", tags=["marketplace"]) +finance_router = APIRouter(prefix="/v1/finance", tags=["finance"]) +webhook_router = APIRouter(prefix="/v1/webhooks/v1", tags=["webhooks"]) + + +def _listing_out(item: Listing) -> dict[str, Any]: + return { + "id": item.id, "title": item.title, "price_minor": item.price_minor, + "currency": item.currency, "status": item.status, + } + + +def _order_out(o: Order) -> dict[str, Any]: + return { + "id": o.id, "status": o.status, "total_minor": o.total_minor, + "currency": o.currency, + "paid_at": o.paid_at.isoformat() if o.paid_at else None, + "refunded_at": o.refunded_at.isoformat() if o.refunded_at else None, + } + + +@router.post("/listings", status_code=201) +async def create_listing( + db: DbSession, + user: Annotated[User, Depends(require_state_change)], + request: Request, + body: dict[str, Any], +) -> dict[str, Any]: + domain = await _domain_of(db, user) + listing = await service.create_listing( + db, domain.id, body, getattr(request.state, "trace_id", None) + ) + return _listing_out(listing) + + +@router.get("/listings") +async def list_listings( + db: DbSession, user: Annotated[User, Depends(get_current_user)] +) -> dict[str, Any]: + rows = list((await db.scalars(select(Listing).limit(50))).all()) + return {"data": [_listing_out(item) for item in rows]} + + +@router.post("/orders", status_code=201, response_model=None) +async def create_order( + db: DbSession, + user: Annotated[User, Depends(require_state_change)], + request: Request, + body: dict[str, Any], + idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key")] = None, +) -> dict[str, Any] | JSONResponse: + if idempotency_key: + body = {**body, "idempotency_key": idempotency_key} + domain = await _domain_of(db, user) + order, replayed = await service.create_order( + db, domain.id, body, getattr(request.state, "trace_id", None) + ) + if replayed: + return JSONResponse(status_code=200, content=_order_out(order)) + return _order_out(order) + + +@router.post("/orders/{order_id}/payment-confirmations", status_code=201) +async def confirm_payment( + order_id: str, + db: DbSession, + user: Annotated[User, Depends(require_state_change)], + request: Request, +) -> dict[str, Any]: + domain = await _domain_of(db, user) + order = await service.confirm_payment( + db, domain.id, order_id, getattr(request.state, "trace_id", None) + ) + return _order_out(order) + + +@router.post("/orders/{order_id}/refunds", status_code=201) +async def refund( + order_id: str, + db: DbSession, + user: Annotated[User, Depends(require_state_change)], + request: Request, +) -> dict[str, Any]: + domain = await _domain_of(db, user) + order = await service.refund_order( + db, domain.id, order_id, getattr(request.state, "trace_id", None) + ) + return _order_out(order) + + +@finance_router.get("/summary") +async def finance_summary( + db: DbSession, user: Annotated[User, Depends(get_current_user)] +) -> dict[str, Any]: + domain = await _domain_of(db, user) + return await service.finance_summary(db, domain.id) + + +@finance_router.post("/preservation", response_model=None) +async def set_preservation( + db: DbSession, + user: Annotated[User, Depends(get_current_user)], + request: Request, + body: dict[str, bool], +) -> dict[str, str]: + """保全态切换（欠费只读；开发档由计费系统触发，验证用）。 + + 不经 require_state_change：保全期间必须能解除保全（否则自锁）。 + """ + domain = await _domain_of(db, user) + value = await service.set_preservation( + db, domain.id, bool(body.get("preserved")), getattr(request.state, "trace_id", None) + ) + return {"billing_status": value} + + +@webhook_router.post("/payments/mock", status_code=200) +async def payment_webhook( + request: Request, + db: DbSession, + x_signature: Annotated[str | None, Header(alias="X-Signature")] = None, +) -> dict[str, str]: + """支付 Provider Webhook：先验签（原始字节），再按事件 ID 幂等处理。""" + raw = await request.body() + service.verify_webhook_signature(raw, x_signature) + try: + payload: dict[str, Any] = json.loads(raw) + except json.JSONDecodeError as exc: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "payload 不是 JSON") from exc + event_type = str(payload.get("type", "payment.succeeded")) + provider_event_id = str(payload.get("provider_event_id", "")) + order_id = str(payload.get("order_id", "")) + if not provider_event_id or not order_id: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "缺少事件标识") + order = await db.get(Order, order_id) + if order is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "订单不存在") + buyer_domain_id = order.buyer_domain_id + + async def handler() -> None: + if event_type == "payment.succeeded": + await service.confirm_payment(db, buyer_domain_id, order_id, None) + elif event_type == "payment.refunded": + await service.refund_order(db, buyer_domain_id, order_id, None) + else: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "未知事件类型") + + payload_hash = hashlib.sha256(raw).hexdigest() + result = await service.process_webhook( + db, "mock", provider_event_id, event_type, payload_hash, + handler, getattr(request.state, "trace_id", None), + ) + return {"result": result} diff --git a/backend/app/marketplace/service.py b/backend/app/marketplace/service.py new file mode 100644 index 0000000..160b446 --- /dev/null +++ b/backend/app/marketplace/service.py @@ -0,0 +1,272 @@ +"""交易域：商品、订单、不可变账本、退款与 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 diff --git a/backend/app/models.py b/backend/app/models.py index b1c126b..558d5a9 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -42,6 +42,8 @@ class PersonalDomain(Base): name: Mapped[str] = mapped_column(String, nullable=False) # 区域驻留（ADR-024 之前的数据驻留决策）：默认区域，迁移走正式协议 data_region: Mapped[str] = mapped_column(String, nullable=False, default="default") + # 保全态（ADR-021）：欠费/封禁只读保全，不删除数据 + billing_status: Mapped[str] = mapped_column(String, nullable=False, default="ok") created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) @@ -673,3 +675,124 @@ class ToolCall(Base): unique=True, ), ) + + +class Listing(Base): + """市场商品（WP-13）：带版本、价格与许可；此处为最小商品形态。""" + + __tablename__ = "listing" + + id: Mapped[str] = mapped_column( + String, primary_key=True, default=lambda: new_id("listing_") + ) + creator_domain_id: Mapped[str] = mapped_column( + ForeignKey("personal_domain.id", ondelete="cascade"), nullable=False, index=True + ) + title: Mapped[str] = mapped_column(String, nullable=False) + price_minor: Mapped[int] = mapped_column(nullable=False, default=0) # 0=免费 + currency: Mapped[str] = mapped_column(String, nullable=False, default="CNY") + status: Mapped[str] = mapped_column(String, nullable=False, default="published") + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = ( + CheckConstraint("price_minor >= 0", name="price_non_negative"), + ) + + +class Order(Base): + """订单（契约 order_status 六态）；下单即价格快照，客户端金额不可信。""" + + __tablename__ = "order" + + id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: new_id("order_")) + buyer_domain_id: Mapped[str] = mapped_column( + ForeignKey("personal_domain.id", ondelete="cascade"), nullable=False, index=True + ) + status: Mapped[str] = mapped_column(String, nullable=False, default="pending_payment") + total_minor: Mapped[int] = mapped_column(nullable=False, default=0) + currency: Mapped[str] = mapped_column(String, nullable=False, default="CNY") + idempotency_key: Mapped[str | None] = mapped_column(String, nullable=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + paid_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + refunded_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + + __table_args__ = ( + CheckConstraint( + "status IN ('pending_payment', 'paid', 'partially_refunded', 'refunded', " + "'disputed', 'cancelled')", + name="status_values", + ), + Index("uq_order_buyer_idempotency", "buyer_domain_id", "idempotency_key", unique=True), + ) + + +class OrderItem(Base): + """订单行：商品与价格快照（下单时冻结，后续改价不影响）。""" + + __tablename__ = "order_item" + + id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: new_id("item_")) + order_id: Mapped[str] = mapped_column( + ForeignKey("order.id", ondelete="cascade"), nullable=False, index=True + ) + listing_id: Mapped[str] = mapped_column( + ForeignKey("listing.id", ondelete="restrict"), nullable=False + ) + title_snapshot: Mapped[str] = mapped_column(String, nullable=False) + price_minor_snapshot: Mapped[int] = mapped_column(nullable=False) + + +class LedgerEntry(Base): + """不可变账务分录（ADR-009）：修正只写反向分录，禁止更新删除。 + + 带符号金额：charge/payable/fee 为正，reversal 为负； + 平衡不变量：每笔订单 sum(charge+reversal) == sum(fee+fee_rev+payable+pay_rev)。 + """ + + __tablename__ = "ledger_entry" + + id: Mapped[str] = mapped_column( + String, primary_key=True, default=lambda: new_id("ledger_") + ) + order_id: Mapped[str] = mapped_column( + ForeignKey("order.id", ondelete="restrict"), nullable=False, index=True + ) + entry_type: Mapped[str] = mapped_column(String, nullable=False) + amount_minor: Mapped[int] = mapped_column(nullable=False) # 带符号 + currency: Mapped[str] = mapped_column(String, nullable=False, default="CNY") + note: Mapped[str | None] = mapped_column(String, nullable=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = ( + CheckConstraint( + "entry_type IN ('charge', 'refund', 'platform_fee', 'creator_payable', " + "'tax', 'reserve', 'payout', 'reversal')", + name="entry_type_values", + ), + Index("ix_ledger_order_type", "order_id", "entry_type"), + ) + + +class WebhookEvent(Base): + """支付 Webhook 事件：按 Provider 事件 ID 幂等，允许乱序到达。""" + + __tablename__ = "webhook_event" + + provider_event_id: Mapped[str] = mapped_column(String, primary_key=True) + provider: Mapped[str] = mapped_column(String, nullable=False) + event_type: Mapped[str] = mapped_column(String, nullable=False) + payload_hash: Mapped[str] = mapped_column(String, nullable=False) + status: Mapped[str] = mapped_column(String, nullable=False, default="processed") + received_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = ( + CheckConstraint("status IN ('processed', 'ignored')", name="status_values"), + ) diff --git a/backend/tests/test_conversations.py b/backend/tests/test_conversations.py index 7136f3c..db2fe59 100644 --- a/backend/tests/test_conversations.py +++ b/backend/tests/test_conversations.py @@ -1016,3 +1016,131 @@ async def _wait_run_terminal_headers(client, headers, limit_s: float = 10.0) -> import asyncio await asyncio.sleep(0) + + +@pytest.mark.usefixtures("client") +class TestMarketplace: + async def _make_listing(self, client, headers, price=1000) -> str: + resp = await client.post( + "/v1/marketplace/listings", + json={"title": "角色包·昭星", "price_minor": price}, headers=headers, + ) + assert resp.status_code == 201, resp.text + return resp.json()["id"] + + async def _make_order(self, client, headers, listing_id, key) -> dict: + resp = await client.post( + "/v1/marketplace/orders", + json={"listing_ids": [listing_id]}, + headers={**headers, "Idempotency-Key": key}, + ) + assert resp.status_code in (200, 201), resp.text + return resp.json() + + async def test_purchase_ledger_double_entry(self, client, db) -> None: + headers = await login(client, "mkt1@example.com") + lid = await self._make_listing(client, headers, price=1000) + order = await self._make_order(client, headers, lid, "ord-1") + assert order["status"] == "pending_payment" and order["total_minor"] == 1000 + paid = await client.post( + f"/v1/marketplace/orders/{order['id']}/payment-confirmations", headers=headers + ) + assert paid.json()["status"] == "paid" + summary = await client.get("/v1/finance/summary", headers=headers) + totals = summary.json()["totals_minor"] + assert totals["charge"] == 1000 + assert totals["platform_fee"] == 100 # 10% + assert totals["creator_payable"] == 900 + assert summary.json()["balanced"] is True + + async def test_order_idempotent_replay(self, client) -> None: + headers = await login(client, "mkt2@example.com") + lid = await self._make_listing(client, headers) + first = await self._make_order(client, headers, lid, "ord-replay") + second = await client.post( + "/v1/marketplace/orders", json={"listing_ids": [lid]}, + headers={**headers, "Idempotency-Key": "ord-replay"}, + ) + assert second.status_code == 200 + assert second.json()["id"] == first["id"] + + async def test_refund_writes_reversal_entries(self, client) -> None: + headers = await login(client, "mkt3@example.com") + lid = await self._make_listing(client, headers, price=2000) + order = await self._make_order(client, headers, lid, "ord-refund") + await client.post( + f"/v1/marketplace/orders/{order['id']}/payment-confirmations", headers=headers + ) + refunded = await client.post( + f"/v1/marketplace/orders/{order['id']}/refunds", headers=headers + ) + assert refunded.json()["status"] == "refunded" + # 重复退款幂等 + again = await client.post( + f"/v1/marketplace/orders/{order['id']}/refunds", headers=headers + ) + assert again.json()["status"] == "refunded" + summary = await client.get("/v1/finance/summary", headers=headers) + totals = summary.json()["totals_minor"] + # charge=2000, reversal=-2000 → 净 0；fee/payable 各自正负抵消 + assert totals.get("charge", 0) + totals.get("reversal", 0) == 0 + assert summary.json()["balanced"] is True + + async def test_webhook_signature_and_idempotency(self, client) -> None: + import hashlib + import hmac as hmac_mod + import os + + headers = await login(client, "mkt4@example.com") + lid = await self._make_listing(client, headers, price=500) + order = await self._make_order(client, headers, lid, "ord-wh") + payload = json.dumps({ + "provider_event_id": "evt-001", "type": "payment.succeeded", + "order_id": order["id"], + }) + secret = os.environ.get("HUNJING_WEBHOOK_SECRET", "whsec_test") + os.environ["HUNJING_WEBHOOK_SECRET"] = "whsec_test" + sig = hmac_mod.new(secret.encode(), payload.encode(), hashlib.sha256).hexdigest() + # 错误签名：401 + bad = await client.post( + "/v1/webhooks/v1/payments/mock", content=payload, + headers={"X-Signature": "deadbeef", "Content-Type": "application/json"}, + ) + assert bad.status_code == 401 + # 正确签名：200 处理 + ok = await client.post( + "/v1/webhooks/v1/payments/mock", content=payload, + headers={"X-Signature": sig, "Content-Type": "application/json"}, + ) + assert ok.status_code == 200 and ok.json()["result"] == "processed" + # 重复投递：幂等 duplicate + dup = await client.post( + "/v1/webhooks/v1/payments/mock", content=payload, + headers={"X-Signature": sig, "Content-Type": "application/json"}, + ) + assert dup.json()["result"] == "duplicate" + + async def test_preservation_readonly_and_recovery(self, client) -> None: + headers = await login(client, "mkt5@example.com") + await client.post( + "/v1/finance/preservation", json={"preserved": True}, headers=headers + ) + # 状态变更被拒绝（保全只读） + blocked = await client.post( + "/v1/conversations", json={"title": "保全期"}, headers=headers + ) + assert blocked.status_code == 403 + assert "保全" in blocked.json()["detail"] + # 读取不受影响：数据仍在 + listing_resp = await client.post( # 验证写拦截一致 + "/v1/marketplace/listings", json={"title": "x"}, headers=headers + ) + assert listing_resp.status_code == 403 + # 恢复：写恢复 + await client.post( + "/v1/finance/preservation", json={"preserved": False}, headers=headers + ) + recovered = await client.post( + "/v1/conversations", json={"title": "恢复"}, headers=headers + ) + assert recovered.status_code == 201