Merge pull request '修复会会支付回调先于查单状态更新导致的 409' (#42) from codex/avatar-org-callback-reconcile-20261020 into main

Reviewed-on: #42
2026-10-20 14:41:50 +08:00
parent 23d9a33a6d 721f8b990d
commit cce5130136
6 changed files with 221 additions and 26 deletions
@@ -282,6 +282,19 @@ def close_expired_pending_orders_job():
        db.close()


+def reconcile_huihui_org_notices_job():
+    db = SessionLocal()
+    try:
+        count = routers.tokens.reconcile_huihui_org_notices(db)
+        if count:
+            logger.info("Confirmed %s Huihui organization payment notice(s)", count)
+    except Exception as exc:
+        db.rollback()
+        logger.warning("Failed to reconcile Huihui organization payment notices: %s", exc)
+    finally:
+        db.close()
+
+
 def start_maintenance_scheduler():
    global maintenance_scheduler

@@ -305,6 +318,13 @@ def start_maintenance_scheduler():
        max_instances=1,
        coalesce=True,
    )
+    maintenance_scheduler.add_job(
+        reconcile_huihui_org_notices_job,
+        trigger=IntervalTrigger(seconds=10),
+        id="reconcile_huihui_org_notices",
+        max_instances=1,
+        coalesce=True,
+    )
    maintenance_scheduler.start()


@@ -437,6 +437,24 @@ class PaymentTransaction(Base):
        }


+class HuihuiOrgPaymentNotice(Base):
+    """Signed success notice awaiting independent confirmation from Huihui."""
+
+    __tablename__ = "huihui_org_payment_notices"
+
+    order_no = Column(String, primary_key=True)
+    payment_no = Column(String, nullable=False)
+    amount_cents = Column(Integer, nullable=False)
+    product_id = Column(String, nullable=False)
+    status = Column(String, nullable=False, default="awaiting_provider", index=True)
+    attempts = Column(Integer, nullable=False, default=0)
+    next_check_at = Column(DateTime, index=True)
+    last_provider_status = Column(String, default="")
+    last_error = Column(String, default="")
+    created_at = Column(DateTime, server_default=func.now())
+    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
+
+
 class PaymentRefund(Base):
    __tablename__ = "payment_refunds"

@@ -4,16 +4,18 @@ import json
 import os
 import time
 import uuid
-from datetime import datetime
+from datetime import datetime, timedelta
 from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
 from urllib.parse import parse_qs

 from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query, Request, Response
 from sqlalchemy import func
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session

 from database import get_db
 from models import (
+    HuihuiOrgPaymentNotice,
    InvoiceApplication,
    PaymentRefund,
    PaymentTransaction,
@@ -889,16 +891,23 @@ def _process_huihui_org_notice(payload: dict, db: Session):
        raise HTTPException(status_code=422, detail="payment details mismatch")
    if order.status in {"paid", "refunded"}:
        return ok({"received": True, "duplicate": True})
-    if not order.provider_order_no or not user.huihui_token or not user.huihui_user_id:
-        raise HTTPException(status_code=503, detail="provider order verification unavailable")
-    try:
-        provider_order = HuihuiOrgPaymentClient().get_order(
-            token=user.huihui_token,
-            user_id=user.huihui_user_id,
-            order_no=order.provider_order_no,
-        )
-    except HuihuiPaymentError as exc:
-        raise HTTPException(status_code=503, detail="provider order verification unavailable") from exc
+    notice = db.get(HuihuiOrgPaymentNotice, order.order_no)
+    if notice and notice.payment_no != payment_no:
+        raise HTTPException(status_code=409, detail="payment number conflicts with prior notice")
+    if not order.provider_order_no:
+        raise HTTPException(status_code=422, detail="provider order number missing")
+    # The provider may not mark its order paid until it receives our HTTP 200.
+    # Never synchronously query the provider before acknowledging a signed notice.
+    _save_huihui_org_notice(db, order, payment_no, amount_cents, expected_product_id)
+    return ok({"received": True, "paid": False, "awaitingProvider": True})
+
+
+def _huihui_org_provider_status(order: TokenPaymentOrder, user: User, product_id: str, payment_no: str) -> int:
+    provider_order = HuihuiOrgPaymentClient().get_order(
+        token=user.huihui_token,
+        user_id=user.huihui_user_id,
+        order_no=order.provider_order_no,
+    )
    try:
        provider_status = int(provider_order.get("payStatus"))
        provider_type = int(provider_order.get("chargeType"))
@@ -907,14 +916,59 @@ def _process_huihui_org_notice(payload: dict, db: Session):
    if (
        str(provider_order.get("orderNo") or "") != order.provider_order_no
        or str(provider_order.get("paymentNo") or "") != payment_no
-        or str(provider_order.get("productId") or "") != expected_product_id
+        or str(provider_order.get("productId") or "") != product_id
        or str(provider_order.get("userId") or "") != user.huihui_user_id
        or provider_type != 9
        or _callback_amount_cents({"chargeAmount": provider_order.get("chargeAmount")}) != order.price_cents
    ):
        raise HTTPException(status_code=422, detail="provider order details mismatch")
-    if provider_status not in {2, 3, 4}:
-        raise HTTPException(status_code=409, detail="provider payment not confirmed")
+    return provider_status
+
+
+def _save_huihui_org_notice(
+    db: Session, order: TokenPaymentOrder, payment_no: str, amount_cents: int,
+    product_id: str,
+) -> None:
+    notice = db.get(HuihuiOrgPaymentNotice, order.order_no)
+    if notice and (
+        notice.payment_no != payment_no
+        or notice.amount_cents != amount_cents
+        or notice.product_id != product_id
+    ):
+        raise HTTPException(status_code=409, detail="payment notice conflicts with prior notice")
+    if notice is None:
+        notice = HuihuiOrgPaymentNotice(
+            order_no=order.order_no, payment_no=payment_no,
+            amount_cents=amount_cents, product_id=product_id,
+        )
+        db.add(notice)
+    notice.status = "awaiting_provider"
+    notice.last_provider_status = "not_checked"
+    notice.last_error = ""
+    first_check = datetime.utcnow() + timedelta(seconds=5)
+    if notice.next_check_at is None or notice.next_check_at > first_check:
+        notice.next_check_at = first_check
+    try:
+        db.commit()
+    except IntegrityError:
+        db.rollback()
+        concurrent = db.get(HuihuiOrgPaymentNotice, order.order_no)
+        if not concurrent or (
+            concurrent.payment_no != payment_no
+            or concurrent.amount_cents != amount_cents
+            or concurrent.product_id != product_id
+        ):
+            raise HTTPException(status_code=409, detail="payment notice conflicts with prior notice")
+
+
+def _huihui_notice_retry_seconds(attempts: int) -> int:
+    return min(15 * (2 ** min(max(0, attempts - 1), 8)), 3600)
+
+
+def _settle_verified_huihui_org_order(
+    db: Session, order: TokenPaymentOrder, payment_no: str,
+    notice: HuihuiOrgPaymentNotice | None = None,
+) -> None:
    existing_payment = db.query(PaymentTransaction).filter(
        PaymentTransaction.provider == "huihui_org",
        PaymentTransaction.event_type == "payment",
@@ -923,8 +977,78 @@ def _process_huihui_org_notice(payload: dict, db: Session):
    if existing_payment and existing_payment.order_no != order.order_no:
        raise HTTPException(status_code=409, detail="payment number belongs to another order")
    _settle_paid_order(db, order, provider_status="HUIHUI_SUCCESS", transaction_no=payment_no)
+    if notice:
+        notice.status = "confirmed"
+        notice.next_check_at = None
+        notice.last_provider_status = "paid"
+        notice.last_error = ""
    db.commit()
-    return ok({"received": True, "paid": True})
+
+
+def reconcile_huihui_org_notices(db: Session, limit: int = 20) -> int:
+    """Retry only signed notices whose provider confirmation lagged the callback."""
+    now = datetime.utcnow()
+    due = db.query(HuihuiOrgPaymentNotice).filter(
+        HuihuiOrgPaymentNotice.status == "awaiting_provider",
+        HuihuiOrgPaymentNotice.next_check_at <= now,
+    ).order_by(HuihuiOrgPaymentNotice.next_check_at).limit(limit).all()
+    confirmed = 0
+    for notice in due:
+        order = db.query(TokenPaymentOrder).filter(
+            TokenPaymentOrder.order_no == notice.order_no,
+            TokenPaymentOrder.provider == "huihui_org",
+        ).first()
+        if not order:
+            notice.status = "rejected"
+            notice.last_error = "order missing"
+            db.commit()
+            continue
+        if order.status in {"paid", "refunded"}:
+            notice.status = "confirmed"
+            notice.next_check_at = None
+            db.commit()
+            continue
+        user = db.query(User).filter(User.id == order.user_id).first()
+        plan = db.query(TokenPlan).filter(TokenPlan.id == order.plan_id).first()
+        try:
+            product_id = product_id_for_plan(plan) if plan else ""
+        except WechatVirtualPaymentError:
+            product_id = ""
+        if not user or not product_id or product_id != notice.product_id or order.price_cents != notice.amount_cents:
+            notice.status = "rejected"
+            notice.last_error = "local order details mismatch"
+            db.commit()
+            continue
+        try:
+            status = _huihui_org_provider_status(order, user, product_id, notice.payment_no)
+        except HuihuiPaymentError:
+            status = None
+        except HTTPException as exc:
+            notice.status = "rejected"
+            notice.last_error = str(exc.detail)[:200]
+            db.commit()
+            continue
+        if status in {2, 3, 4}:
+            try:
+                _settle_verified_huihui_org_order(db, order, notice.payment_no, notice)
+                confirmed += 1
+            except HTTPException as exc:
+                db.rollback()
+                notice = db.get(HuihuiOrgPaymentNotice, notice.order_no)
+                notice.status = "rejected"
+                notice.last_error = str(exc.detail)[:200]
+                db.commit()
+            continue
+        notice.last_provider_status = str(status) if status is not None else "unavailable"
+        notice.attempts = int(notice.attempts or 0) + 1
+        if notice.created_at and now - notice.created_at >= timedelta(hours=24):
+            notice.status = "unconfirmed"
+            notice.next_check_at = None
+            notice.last_error = "provider did not confirm payment within 24 hours"
+        else:
+            notice.next_check_at = now + timedelta(seconds=_huihui_notice_retry_seconds(notice.attempts))
+        db.commit()
+    return confirmed


@router.post("/token/payment/huihui/virtual/notify")
@@ -53,12 +53,15 @@ def test_scheduler_uses_boxim_and_restart_safe_service(
        max_message_age_seconds=600,
    )

-    assert maintenance_scheduler.add_job.call_count == 2
-    cleanup_call, close_orders_call = maintenance_scheduler.add_job.call_args_list
+    assert maintenance_scheduler.add_job.call_count == 3
+    cleanup_call, close_orders_call, reconcile_call = maintenance_scheduler.add_job.call_args_list
    assert cleanup_call.kwargs["id"] == "chat_attachment_cleanup"
    assert close_orders_call.args[0] is main.close_expired_pending_orders_job
    assert close_orders_call.kwargs["id"] == "close_expired_token_orders"
    assert close_orders_call.kwargs["trigger"].interval.total_seconds() == 60
+    assert reconcile_call.args[0] is main.reconcile_huihui_org_notices_job
+    assert reconcile_call.kwargs["id"] == "reconcile_huihui_org_notices"
+    assert reconcile_call.kwargs["trigger"].interval.total_seconds() == 10
    maintenance_scheduler.start.assert_called_once_with()

    assert scheduler.add_job.call_count == 2
@@ -3,13 +3,14 @@ import hmac
 import json
 import os
 import time
+from datetime import datetime, timedelta
 from unittest.mock import patch

 from fastapi.testclient import TestClient

 from database import SessionLocal
 from main import app, seed
-from models import InvoiceApplication, PaymentRefund, TokenAccount, TokenPaymentOrder, TokenPlan, User
+from models import HuihuiOrgPaymentNotice, InvoiceApplication, PaymentRefund, TokenAccount, TokenPaymentOrder, TokenPlan, User
 from services.token_billing import DEFAULT_TOKEN_GRANT


@@ -129,17 +130,46 @@ def test_huihui_org_callback_verifies_provider_before_granting_points(authorizat
        assert send(payload, signature=False, callback_path=legacy_path).status_code == 401
        assert send({**payload, "orderNo": "unknown"}, callback_path=legacy_path).status_code == 404
        assert send({**payload, "amountCents": 99}).status_code == 422
-        assert send(payload).status_code == 409
-        provider.get_order.return_value = {**provider_data, "payStatus": 2}
-        assert send({**payload, "paymentNo": "wrong"}).status_code == 422
-        provider.get_order.return_value = {**provider_data, "payStatus": 2, "chargeAmount": 2}
-        assert send(payload).status_code == 422
-        provider.get_order.return_value = {**provider_data, "payStatus": 2}
+        pending = send(payload, callback_path=legacy_path)
+        assert pending.status_code == 200
+        assert pending.json()["data"] == {"received": True, "paid": False, "awaitingProvider": True}
+        provider.get_order.assert_not_called()
+        db = SessionLocal()
+        try:
+            assert db.get(HuihuiOrgPaymentNotice, order_no).status == "awaiting_provider"
+            assert db.query(TokenPaymentOrder).filter(TokenPaymentOrder.order_no == order_no).one().status == "pending"
+        finally:
+            db.close()
+        assert send({**payload, "paymentNo": "wrong"}).status_code == 409
+        db = SessionLocal()
+        try:
+            notice = db.get(HuihuiOrgPaymentNotice, order_no)
+            notice.next_check_at = datetime.utcnow() - timedelta(seconds=1)
+            db.commit()
+            from routers.tokens import reconcile_huihui_org_notices
+            provider.get_order.return_value = {**provider_data, "payStatus": 2, "chargeAmount": 2}
+            assert reconcile_huihui_org_notices(db) == 0
+            assert db.get(HuihuiOrgPaymentNotice, order_no).status == "rejected"
+            assert db.query(TokenPaymentOrder).filter(TokenPaymentOrder.order_no == order_no).one().status == "pending"
+            assert send(payload).status_code == 200
+            notice.next_check_at = datetime.utcnow() - timedelta(seconds=1)
+            db.commit()
+            provider.get_order.return_value = provider_data
+            assert reconcile_huihui_org_notices(db) == 0
+            assert db.get(HuihuiOrgPaymentNotice, order_no).attempts == 1
+            assert db.query(TokenPaymentOrder).filter(TokenPaymentOrder.order_no == order_no).one().status == "pending"
+            notice.next_check_at = datetime.utcnow() - timedelta(seconds=1)
+            db.commit()
+            provider.get_order.return_value = {**provider_data, "payStatus": 2}
+            assert reconcile_huihui_org_notices(db) == 1
+            assert db.get(HuihuiOrgPaymentNotice, order_no).status == "confirmed"
+        finally:
+            db.close()
        first = send(payload, callback_path=legacy_path)
        second = send(payload)

    assert first.status_code == 200
-    assert first.json()["data"] == {"received": True, "paid": True}
+    assert first.json()["data"] == {"received": True, "duplicate": True}
    assert second.json()["data"] == {"received": True, "duplicate": True}
    db = SessionLocal()
    try:
@@ -28,6 +28,6 @@
 - `X-Huihui-Timestamp`：Unix 秒级时间戳，允许与本服务时间相差不超过 300 秒。
 - `X-Huihui-Signature`：小写十六进制 HMAC-SHA256，待签名数据为 `时间戳 + 换行符 + 原始请求体字节`。组织充值订单使用服务端配置的 `HUIHUI_ORG_CALLBACK_SECRET`（至少 16 字符），微信直连虚拟支付订单使用独立的 `HUIHUI_VIRTUAL_CALLBACK_SECRET`（至少 32 字符）。不要重排 JSON 后签名，也不要把密钥发给前端。

-成功接收返回 HTTP 200，`{"code":200,"message":"success","data":{"received":true,"paid":true}}`。重复通知返回 `duplicate: true`。验签、字段、订单或金额不符返回 HTTP 4xx，会会可按其重试策略处理。服务端未配置密钥时返回 HTTP 503。
+成功通知经签名及本地订单信息核对后，先持久保存并立即返回 HTTP 200、`{"received":true,"paid":false,"awaitingProvider":true}`；回调入口**不等待会会查单**，也**不立即发放积分**。后台仅针对已保存的通知做有上限的延迟复核，待会会查单确认已支付、订单号与支付流水等字段一致后才入账。已入账订单的重复通知返回 `duplicate: true`，不会重复发放积分。验签、字段、订单或金额不符返回 HTTP 4xx。服务端未配置密钥时返回 HTTP 503。

 上线前需由会会确认：其服务端是否能按此格式签名并发送通知、支付成功状态与支付流水号字段的正式定义、失败通知及重试/超时策略。小程序支付发起和退款仍需独立对接和验收；仅完成此回调不代表虚拟支付全流程已可用。