hunjing/backend/app/marketplace/routes.py

"""交易域路由（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}