fix: reconcile Huihui payment callbacks after provider confirmation

digital-avatar-app/backend/main.py

@@ -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=30),
+        id="reconcile_huihui_org_notices",
+        max_instances=1,
+        coalesce=True,
+    )
     maintenance_scheduler.start()
 
 

digital-avatar-app/backend/models.py

@@ -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"
 

digital-avatar-app/backend/routers/tokens.py

@@ -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,32 @@ 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")
+    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")
+    if not user.huihui_token or not user.huihui_user_id:
+        _save_huihui_org_notice(db, order, payment_no, amount_cents, expected_product_id, "unavailable")
+        return ok({"received": True, "paid": False, "awaitingProvider": True})
     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
+        provider_status = _huihui_org_provider_status(order, user, expected_product_id, payment_no)
+    except HuihuiPaymentError:
+        _save_huihui_org_notice(db, order, payment_no, amount_cents, expected_product_id, "unavailable")
+        return ok({"received": True, "paid": False, "awaitingProvider": True})
+    if provider_status not in {2, 3, 4}:
+        _save_huihui_org_notice(db, order, payment_no, amount_cents, expected_product_id, str(provider_status))
+        return ok({"received": True, "paid": False, "awaitingProvider": True})
+    _settle_verified_huihui_org_order(db, order, payment_no, notice)
+    return ok({"received": True, "paid": 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 +925,58 @@ 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, provider_status: 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 = provider_status
+    notice.last_error = ""
+    notice.attempts = int(notice.attempts or 0) + 1
+    notice.next_check_at = datetime.utcnow() + timedelta(seconds=_huihui_notice_retry_seconds(notice.attempts))
+    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 +985,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")

digital-avatar-app/backend/tests/test_takeover_scheduler.py

@@ -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() == 30
     maintenance_scheduler.start.assert_called_once_with()
 
     assert scheduler.add_job.call_count == 2

digital-avatar-app/backend/tests/test_virtual_payment_flow.py

@@ -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,42 @@ 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
+        pending = send(payload, callback_path=legacy_path)
+        assert pending.status_code == 200
+        assert pending.json()["data"] == {"received": True, "paid": False, "awaitingProvider": True}
+        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()
         provider.get_order.return_value = {**provider_data, "payStatus": 2}
-        assert send({**payload, "paymentNo": "wrong"}).status_code == 422
+        assert send({**payload, "paymentNo": "wrong"}).status_code == 409
         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}
+        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
+            assert reconcile_huihui_org_notices(db) == 0
+            assert db.get(HuihuiOrgPaymentNotice, order_no).attempts == 2
+            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: