feat(ai-gateway): webhook 支付确认后自动绑定用户（支付→激活闭环）

Author: Aparencia

server/ai-gateway/routers/license_webhook.py

@@ -2,9 +2,12 @@
 熵减 AI 网关 — 面包多订单通知 Webhook
 
 @ai-context: 接收支付平台订单通知：快速应答 200 → 异步验真（HMAC 签名 +
-order_id 查询确认双保险）→ 标记激活码 sold。重复通知幂等（状态更新幂等）。
+order_id 查询确认双保险）→ 标记激活码 sold → 若回调携带用户标识
+(from_user/buyer_email) 自动完成 sold → bound 绑定 + 更新 paid metadata，
+实现「支付→自动激活」闭环。重复通知幂等（状态更新幂等，已 bound 不再重复绑）。
 验真失败/平台不可用时入队 pending（内存队列 + 日志告警），由
-scripts/license-admin.mjs reconcile 人工对账兜底。
+scripts/license-admin.mjs reconcile 人工对账兜底。自动绑定失败不阻塞订单
+确认（仅记日志），用户可回落手动输入激活码流程。
 @ai-context: 本端点不要求用户登录（平台服务端回调），不注册限流（依赖验真
 而非频率限制防伪造）；原始 payload 落日志供排查。
 """
@@ -60,14 +63,36 @@ async def _handle_webhook(payload: dict[str, Any], signature: str | None = None)
     # 查询确认（主闸）：凭 order_id 主动查询订单真实状态
     ok, reason = await payment_adapter.verify_and_mark_sold(order_id, code)
 
-    if ok:
-        return {"code": 200, "message": "ok"}
-    if sig_ok and reason.startswith("订单未支付"):
-        # 签名有效但订单未支付：正常拒绝，不入队（不是系统故障）
-        return {"code": 200, "message": reason}
-    # 查询失败/平台不可用/验真失败：入队待对账，先应答 200 避免重试风暴
-    _enqueue_pending(order_id, reason)
-    return {"code": 200, "message": "accepted"}
+    if not ok:
+        if sig_ok and reason.startswith("订单未支付"):
+            # 签名有效但订单未支付：正常拒绝，不入队（不是系统故障）
+            return {"code": 200, "message": reason}
+        # 查询失败/平台不可用/验真失败：入队待对账，先应答 200 避免重试风暴
+        _enqueue_pending(order_id, reason)
+        return {"code": 200, "message": "accepted"}
+
+    # ✅ 订单确认 sold 成功后，尝试自动绑定用户（支付→激活闭环）
+    buyer = payment_adapter.extract_buyer_info(payload)
+    if buyer.get("user_id"):
+        from services.supabase_adapter import auto_bind_license
+
+        bound_ok = await auto_bind_license(code, buyer["user_id"])
+        if bound_ok:
+            logger.info("Webhook 自动绑定成功: code=%s, user=%s", (code or "")[:12] + "...", buyer["user_id"])
+        else:
+            logger.warning("Webhook 自动绑定失败（状态非 sold 或绑定失败）: code=%s", (code or "")[:12] + "...")
+    elif buyer.get("email"):
+        # 通过邮箱查询 user_id 后尝试自动绑定
+        from services.supabase_adapter import auto_bind_license, get_user_by_email
+
+        user = await get_user_by_email(buyer["email"])
+        user_id = (user or {}).get("id") or (user or {}).get("user_id")
+        if user_id:
+            bound_ok = await auto_bind_license(code, user_id)
+            if bound_ok:
+                logger.info("Webhook 通过 email 自动绑定成功: code=%s, user=%s", code[:12] + "...", user_id)
+
+    return {"code": 200, "message": "ok"}
 
 
 @router.post("/webhook")

server/ai-gateway/services/payment_adapter.py

@@ -193,3 +193,31 @@ async def verify_and_mark_sold(order_id: str, code: str | None = None) -> tuple[
 
     logger.info("订单已确认并标记 sold: order=%s, code=%s", order_id, sold_code[:12] + "...")
     return True, "ok"
+
+
+# ============================================================
+# 买家信息提取（Webhook 自动绑定闭环）
+# ============================================================
+
+
+def extract_buyer_info(payload: dict) -> dict:
+    """从 webhook payload 提取用户标识信息。
+
+    面包多卡密模式的自定义参数通过回调参数回传（from_user / buyer_email）。
+    缺失时返回空 dict，调用方回落手动激活码模式。
+
+    @ai-context: 用户标识不参与签名验真（平台回调原样透传），因此自动绑定
+    仅作为体验增强：绑定失败不阻塞订单确认，回落手动激活码流程兜底。
+
+    Args:
+        payload: 解析后的 webhook 请求体
+
+    Returns:
+        dict: {"user_id": str, "email": str}，均为空串表示无用户标识
+    """
+    from_user = payload.get("from_user") or ""
+    email = payload.get("buyer_email") or payload.get("email") or ""
+    return {
+        "user_id": from_user,
+        "email": email,
+    }

server/ai-gateway/services/supabase_adapter.py

@@ -217,3 +217,81 @@ async def update_paid_metadata(user_id: str, paid: dict[str, Any]) -> bool:
         json_body={"user_metadata": {"paid": paid}},
     )
     return bool(data)
+
+
+# ============================================================
+# Webhook 自动绑定闭环（支付 → 自动激活）
+# ============================================================
+
+
+async def get_user_by_email(email: str) -> dict | None:
+    """通过邮箱查询 Supabase 用户（Auth Admin API）。
+
+    @ai-context: webhook 携带 buyer_email 时，需要先解析出 user_id 才能写入
+    user_metadata.paid；查询失败返回 None，调用方跳过自动绑定回落手动激活。
+
+    Args:
+        email: 买家邮箱
+
+    Returns:
+        dict | None: 用户对象（含 id），未找到/查询失败返回 None
+    """
+    if _is_mock_mode():
+        return _mock_metadata.get(email)
+    data = await _http_request(
+        "GET",
+        f"{_config()[0]}/auth/v1/admin/users?email={email}",
+    )
+    if not data:
+        return None
+    if isinstance(data, list):
+        return data[0] if data else None
+    # Supabase Auth Admin API 返回 {"users": [...]} 结构，取首个匹配用户
+    if isinstance(data, dict) and data.get("users"):
+        return data["users"][0]
+    return data
+
+
+async def auto_bind_license(code: str, user_id: str) -> bool:
+    """Webhook 确认后自动绑定许可证（sold → bound）+ 更新 paid metadata。
+
+    这是「支付→自动激活」闭环的核心：无需用户手动输入激活码。
+    返回 False 表示激活码不存在、状态不是 sold 或绑定失败。
+
+    @ai-context: 自动绑定为幂等增强：绑定失败不抛异常、不阻塞订单确认，
+    由调用方记录日志，用户可回落手动输入激活码流程。
+
+    Args:
+        code: 激活码（webhook 已确认订单）
+        user_id: 买家用户 ID
+
+    Returns:
+        bool: 是否绑定成功
+    """
+    from datetime import datetime, timedelta, timezone
+
+    row = await get_license_by_code(code)
+    if not row:
+        return False
+
+    if row.get("status") != "sold":
+        logger.warning("自动绑定拒绝: 状态不是 sold (code=%s, status=%s)", code[:12], row.get("status"))
+        return False
+
+    duration = row.get("duration_days", 30)
+    now = datetime.now(timezone.utc)
+    expires_at = (now + timedelta(days=duration)).isoformat()
+
+    bound = await bind_license(code, user_id, "auto-webhook", expires_at)
+    if not bound:
+        return False
+
+    tier = "lifetime" if row.get("type") == "lifetime" else "pro"
+    paid = {
+        "tier": tier,
+        "expires_at": expires_at if tier == "pro" else None,
+        "updated_at": now.isoformat(),
+    }
+    await update_paid_metadata(user_id, paid)
+
+    return True

server/ai-gateway/tests/test_payment_webhook.py

@@ -7,6 +7,7 @@
 - 幂等：重复通知仍 200（不重复入队不报错）
 - 降级：查询失败 → 入队 pending，响应 200
 - 订单未支付（签名有效时）→ 正常拒绝不入队
+- 自动绑定：携带 from_user → sold → bound + paid metadata；无用户标识 → 仅 sold
 """
 
 import hashlib
@@ -145,3 +146,77 @@ async def test_missing_order_id_rejected(self):
 
         result = await _handle_webhook({})
         assert result["code"] == 400
+
+
+class TestAutoBind:
+    """webhook 自动绑定用户（支付→激活闭环）"""
+
+    @pytest.mark.asyncio
+    async def test_webhook_with_from_user_auto_binds(self):
+        """携带 from_user 的 webhook 自动完成 sold → bound + paid metadata"""
+        from routers.license_webhook import _handle_webhook
+        from services import payment_adapter, supabase_adapter
+
+        _seed_pool()
+        payment_adapter._seed_mock_order("MB-2001", "ENTROPY-PRO-AAAA-BBBB", status="paid")
+
+        result = await _handle_webhook({
+            "order_id": "MB-2001",
+            "code": "ENTROPY-PRO-AAAA-BBBB",
+            "from_user": "user-123",
+        })
+        assert result["code"] == 200
+        assert result["message"] == "ok"
+
+        # 自动绑定：sold → bound
+        row = supabase_adapter._mock_pool["ENTROPY-PRO-AAAA-BBBB"]
+        assert row["status"] == "bound"
+        assert row["bound_user_id"] == "user-123"
+        assert row["machine_id"] == "auto-webhook"
+
+        # paid metadata 已更新（mock 模式直接存 paid dict：tier/expires_at/updated_at）
+        meta = supabase_adapter._mock_metadata.get("user-123")
+        assert meta is not None
+        assert meta["tier"] == "pro"
+        assert meta["expires_at"] is not None
+
+    @pytest.mark.asyncio
+    async def test_webhook_without_from_user_no_bind(self):
+        """无用户标识的 webhook 仅标记 sold，不自动绑定（兼容手动激活码）"""
+        from routers.license_webhook import _handle_webhook
+        from services import payment_adapter, supabase_adapter
+
+        _seed_pool()
+        payment_adapter._seed_mock_order("MB-2002", "ENTROPY-PRO-AAAA-BBBB", status="paid")
+
+        result = await _handle_webhook({
+            "order_id": "MB-2002",
+            "code": "ENTROPY-PRO-AAAA-BBBB",
+        })
+        assert result["code"] == 200
+
+        # 仅 sold，未绑定（mock 记录中 bound_user_id/machine_id 保持 None）
+        row = supabase_adapter._mock_pool["ENTROPY-PRO-AAAA-BBBB"]
+        assert row["status"] == "sold"
+        assert row["bound_user_id"] is None
+        assert row["machine_id"] is None
+
+    @pytest.mark.asyncio
+    async def test_webhook_with_unknown_email_no_bind(self):
+        """携带 email 但未匹配到用户 → 跳过绑定不崩溃（回落手动激活）"""
+        from routers.license_webhook import _handle_webhook
+        from services import payment_adapter, supabase_adapter
+
+        _seed_pool()
+        payment_adapter._seed_mock_order("MB-2003", "ENTROPY-PRO-AAAA-BBBB", status="paid")
+
+        result = await _handle_webhook({
+            "order_id": "MB-2003",
+            "code": "ENTROPY-PRO-AAAA-BBBB",
+            "buyer_email": "buyer@example.com",
+        })
+        assert result["code"] == 200
+
+        row = supabase_adapter._mock_pool["ENTROPY-PRO-AAAA-BBBB"]
+        assert row["status"] == "sold"
+        assert row["bound_user_id"] is None