qihuaiflow/docs/architecture/market-billing-webhook-design.md

应用市场产品计费 Webhook 设计 — #939

任务:#939 — 市场产品计费 webhook 设计 日期:2026-08-09 状态:✅ 设计完成 / 留待 MuMuAINovel 上线后接入

---

1. 业务诉求

平台应用市场(market_products)中,产品可以是:

• 免费 SaaS:无计费,接入即可用
• 按量计费(metered):每次调用扣账户余额
• 订阅(subscription):月度/年固定费
• 免费增值(freemium):基础免费 / 高级付费

当前 4 个计费方案中,前两种(free)零成本。metered / subscription / freemium 需要:

1. 计费事件触发(产品方在用户用了功能后回调)
2. 对账(平台账与产品方账一致)
3. 重试(网络抖动不能丢单)

---

2. 目标架构

2.1 单向事件流(平台 = 计费接收方)

产品方后端 → POST /app/api/market/products/{id}/billing-event
              → 平台校验签名(HMAC-SHA256)
              → 落 im_billing_events 表
              → 异步 reconcile worker → 扣用户余额 / 开订阅
              → 200 OK + {accepted: true, event_id}

关键设计点:

• Webhook 是单向事件流,不假设产品方按序回调(用 event_id 去重)
• 平台幂等:同一 event_id 重复回调不重复扣费
• 平台有状态:余额扣减事务化,失败可重试

2.2 数据表 im_billing_events

CREATE TABLE IF NOT EXISTS im_billing_events (
    id                  BIGSERIAL PRIMARY KEY,
    event_id            TEXT NOT NULL UNIQUE,         -- 产品方传的去重键(UUID)
    product_id          UUID NOT NULL,                -- market_products.id
    user_id             UUID,                         -- 关联平台用户(可空:订阅场景)
    event_type          TEXT NOT NULL,                -- 'usage' / 'subscription_start' / 'subscription_end' / 'refund'
    quantity            DECIMAL(18, 6) NOT NULL DEFAULT 1,
    unit_price          DECIMAL(18, 6),               -- 单价(可选,平台基于 product.billing_plan 校验)
    amount              DECIMAL(18, 6) NOT NULL,      -- 扣减金额(分)
    currency            TEXT NOT NULL DEFAULT 'CNY',
    raw_payload         JSONB NOT NULL,               -- 产品方完整 payload,供对账
    signature           TEXT NOT NULL,                -- HMAC-SHA256
    received_at         TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    processed_at        TIMESTAMPTZ,                  -- worker 处理成功时间
    reconcile_status    TEXT NOT NULL DEFAULT 'pending',
    error_message       TEXT,
    retry_count         INT NOT NULL DEFAULT 0,
    UNIQUE (event_id)
);
CREATE INDEX idx_im_billing_events_product ON im_billing_events(product_id, received_at DESC);
CREATE INDEX idx_im_billing_events_user ON im_billing_events(user_id, received_at DESC);
CREATE INDEX idx_im_billing_events_pending ON im_billing_events(reconcile_status) WHERE reconcile_status = 'pending';

2.3 Webhook 端点

# app/api/market_billing_webhook.py
from fastapi import APIRouter, Request, HTTPException, Depends
import hmac, hashlib, json
from datetime import datetime

router = APIRouter(prefix="/app/api/market/products", tags=["billing-webhook"])


def verify_signature(product_id: str, body: bytes, signature: str) -> bool:
    """HMAC-SHA256 验签。
    每个产品有独立 secret(存 market_products.metadata.billing_secret)。
    """
    product = db.query(MarketProduct).filter(...).first()
    secret = (product.metadata or {}).get("billing_secret", "")
    expected = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)


@router.post("/{product_id}/billing-event")
async def receive_billing_event(product_id: str, request: Request):
    body = await request.body()
    sig = request.headers.get("X-Billing-Signature", "")
    if not verify_signature(product_id, body, sig):
        raise HTTPException(401, "invalid signature")
    payload = json.loads(body)
    event_id = payload.get("event_id")
    if not event_id:
        raise HTTPException(400, "missing event_id")
    # 幂等:已有 event_id 直接返回
    existing = db.query(ImBillingEvent).filter(ImBillingEvent.event_id == event_id).first()
    if existing:
        return {"accepted": True, "event_id": existing.id, "dedup": True}
    # 落库
    evt = ImBillingEvent(
        event_id=event_id,
        product_id=product_id,
        user_id=payload.get("user_id"),
        event_type=payload["event_type"],
        amount=payload["amount"],
        raw_payload=payload,
        signature=sig,
    )
    db.add(evt); db.commit()
    return {"accepted": True, "event_id": evt.id}

2.4 Reconcile Worker(异步)

# app/api/market_billing_worker.py
import asyncio
async def process_pending_events():
    """每 30s 扫一次 reconcile_status='pending' 的事件,处理:
    - 'usage': 扣 user_balance.amount += amount
    - 'subscription_start': 创建 billing_subscription 行
    - 'subscription_end': 标记 subscription cancelled
    - 'refund': 反向扣减
    """
    while True:
        events = db.query(ImBillingEvent).filter(
            ImBillingEvent.reconcile_status == "pending",
            ImBillingEvent.retry_count < 5,
        ).limit(100).all()
        for evt in events:
            try:
                apply_to_user_balance(evt)
                evt.reconcile_status = "applied"
                evt.processed_at = datetime.utcnow()
                db.commit()
            except Exception as e:
                evt.retry_count += 1
                evt.error_message = str(e)
                db.commit()
        await asyncio.sleep(30)

---

3. 安全

项	实现
防重放	event_id 唯一约束
防伪造	HMAC-SHA256(每产品独立 secret)
防篡改	数据库行级 hash(可选,signature + body 二次 hash)
防过载	Rate limit:每产品 100 events/min
鉴权	IP 白名单(可选,Casdoor OIDC token 也可)

---

4. 重试 / 死信

• 平台处理失败:retry_count 累加,error_message 记录
• 5 次重试仍失败:reconcile_status = 'dead_letter',后台告警
• 产品方重发:用同一 event_id,平台识别为重复

---

5. 行动项

5.1 P1 — 落库 + 端点 stub(0.5 天)

• 迁移 im_billing_events 表
• 写 webhook 端点 + 幂等检查
• 后台 admin 加 "Billing Events" 列表页

5.2 P2 — Worker + 计费对接(1 天)

• 异步 reconcile worker(后台 task)
• 余额扣减事务化(SQLAlchemy with FOR UPDATE)
• 计费异常告警(Grafana + Alertmanager)

5.3 P3 — MuMuAINovel 接入(0.5 天)

• 文档化 webhook 接入规范
• MuMuAINovel 后端按规范加事件推送

当前不实现任何代码,只保留设计 + 接口契约。等 MuMuAINovel 上线后再拉 P1。