diff --git a/video-gen-api/app/api/v1/payments.py b/video-gen-api/app/api/v1/payments.py index 10d47cfc..75bbd39b 100644 --- a/video-gen-api/app/api/v1/payments.py +++ b/video-gen-api/app/api/v1/payments.py @@ -18,7 +18,8 @@ from app.services.payment import ( verify_alipay_callback, process_payment_success_by_order_no, _get_payment_configs, - _close_alipay_order, + PaymentCloseResult, + _resolve_order_for_cancellation, ) router = APIRouter(prefix="/payments", tags=["payments"]) @@ -421,21 +422,28 @@ async def cancel_order( if order.status != "pending": raise HTTPException(status_code=400, detail=f"订单状态为{order.status}，无法取消") - # Alipay/WeChat must confirm remote close before local cancellation. - # If close fails or the remote state is uncertain, keep the local order - # pending so a later gateway query can still discover a successful payment. + # Resolve the authoritative gateway state before local cancellation. + # If WeChat explicitly reports ORDERPAID, the payment service queries only + # this order once and runs the normal atomic payment-success fulfillment. db_configs = await _get_payment_configs(db) - closed = True - if order.payment_method == "alipay": - closed = await _close_alipay_order(db, order, db_configs) - elif order.payment_method == "wechat": - from app.services.payment import _close_wechat_order - closed = await _close_wechat_order(db, order, db_configs) + log_user_id = str(current_user.id) + payment_method = str(order.payment_method) + close_result = await _resolve_order_for_cancellation(db, order, db_configs) - if not closed: + if close_result == PaymentCloseResult.PAID: + logger.info( + f"ORDER_CANCEL_PAID_RECOVERED order_no={order_no} " + f"user={log_user_id} method={payment_method}" + ) + raise HTTPException( + status_code=409, + detail="订单已支付并完成支付处理，无法取消", + ) + + if close_result != PaymentCloseResult.CLOSED: logger.warning( f"ORDER_CANCEL_CLOSE_PENDING order_no={order_no} " - f"user={current_user.id} method={order.payment_method}" + f"user={log_user_id} method={payment_method}" ) raise HTTPException( status_code=409, diff --git a/video-gen-api/app/services/payment.py b/video-gen-api/app/services/payment.py index f9aeadb8..f2ed4c04 100644 --- a/video-gen-api/app/services/payment.py +++ b/video-gen-api/app/services/payment.py @@ -2,6 +2,7 @@ import logging import os import json from datetime import datetime, timedelta +from enum import Enum # 尝试设置 SSL 证书路径 try: @@ -115,6 +116,19 @@ if not logger.handlers: DEFAULT_ORDER_EXPIRE_SECONDS = 180 +class PaymentCloseResult(str, Enum): + """Result of attempting to close a remote payment order. + + ``PAID`` is intentionally distinct from ``FAILED``: when a gateway + explicitly says an order is already paid, callers must recover the paid + transaction instead of cancelling locally or retrying close forever. + """ + + CLOSED = "closed" + PAID = "paid" + FAILED = "failed" + + def _get_order_expire_seconds(db_configs: dict[str, str]) -> int: """Get order expire time in seconds from config, with fallback to 180.""" try: @@ -174,26 +188,48 @@ async def _close_order_for_cancellation( db: AsyncSession, order: PaymentOrder, db_configs: dict[str, str], -) -> bool: +) -> PaymentCloseResult: """Close a gateway order before changing the local order to cancelled. - For Alipay/WeChat, local cancellation is only allowed after the gateway - confirms that the order has been closed. Unknown/legacy payment methods - keep the historical local-only cancellation behaviour. + A gateway may explicitly report that the order has already been paid. + That is not a close failure: it must be routed into paid-order recovery so + the normal payment fulfillment pipeline can run. """ if order.payment_method == "alipay": - return await _close_alipay_order(db, order, db_configs) + closed = await _close_alipay_order(db, order, db_configs) + return PaymentCloseResult.CLOSED if closed else PaymentCloseResult.FAILED if order.payment_method == "wechat": return await _close_wechat_order(db, order, db_configs) - return True + return PaymentCloseResult.CLOSED + + +async def _resolve_order_for_cancellation( + db: AsyncSession, + order: PaymentOrder, + db_configs: dict[str, str], +) -> PaymentCloseResult: + """Resolve a cancellation attempt into closed / paid / failed. + + Normal orders are simply closed. For WeChat ``ORDERPAID``, query only + this order once and feed the verified result into the existing atomic + payment-success handler. + """ + close_result = await _close_order_for_cancellation(db, order, db_configs) + if close_result != PaymentCloseResult.PAID: + return close_result + + if order.payment_method != "wechat": + return PaymentCloseResult.FAILED + + recovered = await _recover_wechat_paid_order(db, order, db_configs) + return PaymentCloseResult.PAID if recovered else PaymentCloseResult.FAILED async def _check_and_expire_order(db: AsyncSession, order: PaymentOrder) -> bool: - """Expire a pending order only after the payment gateway confirms close. + """Expire one pending order after resolving the remote gateway state. - Returns True only when the local order was actually changed to cancelled. - If the gateway close result is uncertain/failed, the order remains pending - so that later query/reconciliation can still discover a successful payment. + If WeChat reports ``ORDERPAID``, the order is queried once and recovered + through the normal payment-success pipeline instead of being cancelled. """ if order.status != "pending": return False @@ -204,11 +240,21 @@ async def _check_and_expire_order(db: AsyncSession, order: PaymentOrder) -> bool if utc_now() < expiry: return False - closed = await _close_order_for_cancellation(db, order, db_configs) - if not closed: + log_order_no = str(order.order_no) + log_user_id = str(order.user_id) + log_amount = order.amount + close_result = await _resolve_order_for_cancellation(db, order, db_configs) + if close_result == PaymentCloseResult.PAID: + logger.info( + f"ORDER_EXPIRE_PAID_RECOVERED order_no={log_order_no} " + f"user={log_user_id} amount={log_amount}" + ) + return False + + if close_result != PaymentCloseResult.CLOSED: logger.warning( - f"ORDER_EXPIRE_CLOSE_PENDING order_no={order.order_no} " - f"user={order.user_id} amount={order.amount}" + f"ORDER_EXPIRE_CLOSE_PENDING order_no={log_order_no} " + f"user={log_user_id} amount={log_amount}" ) return False @@ -224,44 +270,87 @@ async def _check_and_expire_order(db: AsyncSession, order: PaymentOrder) -> bool async def expire_all_pending_orders(db: AsyncSession) -> int: - """Background task: close and cancel expired pending orders. + """Background task: resolve and expire overdue pending orders. - Alipay/WeChat orders remain pending when the remote close cannot be - confirmed. This prevents a real paid order from being persisted locally - as cancelled. + Each order is handled in its own database transaction. This keeps one + gateway or fulfillment failure from invalidating ORM state for the rest of + the batch. A WeChat ``ORDERPAID`` result is recovered as a paid order, + never persisted as cancelled. """ db_configs = await _get_payment_configs(db) expire_seconds = _get_order_expire_seconds(db_configs) threshold = utc_now() - timedelta(seconds=expire_seconds) + result = await db.execute( - select(PaymentOrder).where( + select(PaymentOrder.order_no).where( PaymentOrder.status == "pending", PaymentOrder.created_at <= threshold, ) ) - orders = result.scalars().all() + order_nos = [str(row[0]) for row in result.all()] + + # End the read-only candidate transaction before processing individual + # orders. Subsequent commit/rollback never leaves us reusing stale ORM + # instances from this initial scan. + await db.rollback() + expired_count = 0 - for order in orders: - closed = await _close_order_for_cancellation(db, order, db_configs) - if not closed: - logger.warning( - f"ORDER_EXPIRE_CLOSE_PENDING order_no={order.order_no} " - f"user={order.user_id} amount={order.amount}" + for order_no in order_nos: + try: + result = await db.execute( + select(PaymentOrder) + .where( + PaymentOrder.order_no == order_no, + PaymentOrder.status == "pending", + ) + .with_for_update() + .limit(1) ) - continue + order = result.scalar_one_or_none() + if order is None: + await db.rollback() + continue - order.status = "cancelled" - if order.upgrade_period_ids_json: - await release_upgrade_reservation(db, order=order, released_at=utc_now()) - expired_count += 1 - logger.info( - f"ORDER_EXPIRED order_no={order.order_no} user={order.user_id} " - f"amount={order.amount}" - ) + close_result = await _resolve_order_for_cancellation( + db, order, db_configs + ) + + if close_result == PaymentCloseResult.PAID: + logger.info( + f"ORDER_EXPIRE_PAID_RECOVERED order_no={order_no}" + ) + # Paid recovery commits inside process_payment_success_by_order_no. + continue + + if close_result != PaymentCloseResult.CLOSED: + logger.warning( + f"ORDER_EXPIRE_CLOSE_PENDING order_no={order_no} " + f"user={order.user_id} amount={order.amount}" + ) + await db.rollback() + continue + + log_user_id = str(order.user_id) + log_amount = order.amount + order.status = "cancelled" + if order.upgrade_period_ids_json: + await release_upgrade_reservation( + db, order=order, released_at=utc_now() + ) + + await db.commit() + expired_count += 1 + logger.info( + f"ORDER_EXPIRED order_no={order_no} user={log_user_id} " + f"amount={log_amount}" + ) + except Exception: + await db.rollback() + logger.exception( + f"ORDER_EXPIRE_PROCESS_FAILED order_no={order_no}" + ) - if expired_count > 0: - await db.commit() return expired_count @@ -598,10 +687,12 @@ def _create_wechat_order(order: PaymentOrder, db_configs: dict[str, str]) -> str return None -async def _close_wechat_order(db: AsyncSession, order: PaymentOrder, db_configs: dict[str, str]) -> bool: - """Call WeChat Pay close API to close an unpaid order. - Returns True if the order was closed successfully. - """ +async def _close_wechat_order( + db: AsyncSession, + order: PaymentOrder, + db_configs: dict[str, str], +) -> PaymentCloseResult: + """Call WeChat Pay close API and classify the authoritative result.""" mch_id = db_configs.get("payment_wechat_mch_id", "") private_key = db_configs.get("payment_wechat_private_key", "") cert_serial_no = db_configs.get("payment_wechat_cert_serial_no", "") @@ -613,18 +704,18 @@ async def _close_wechat_order(db: AsyncSession, order: PaymentOrder, db_configs: client = _get_wechat_client(mch_id, private_key, cert_serial_no, api_v3_key, appid, notify_url, public_key, public_key_id) if client is None: - return False + return PaymentCloseResult.FAILED mock_mode = _is_mock_mode(db_configs) if mock_mode: logger.info(f"Mock mode: skipping close_wechat_order for {order.order_no}") - return True + return PaymentCloseResult.CLOSED try: code, result = client.close(out_trade_no=order.order_no) if code == 204: logger.info(f"WeChat order closed: order_no={order.order_no}") - return True + return PaymentCloseResult.CLOSED data = _parse_wechat_result(result) error_code = str( @@ -645,16 +736,26 @@ async def _close_wechat_order(db: AsyncSession, order: PaymentOrder, db_configs: f"WeChat order not exist, treat as closed for local expiry: " f"order_no={order.order_no}, code={code}, error_code={error_code or 'unknown'}" ) - return True + return PaymentCloseResult.CLOSED + + # ORDERPAID means the remote transaction is already paid. Do not + # cancel locally and do not retry close forever; the caller will query + # this one order once and recover it through the normal paid pipeline. + if error_code == "ORDERPAID" or "ORDERPAID" in raw_result: + logger.info( + f"WeChat close reports order already paid: " + f"order_no={order.order_no}, code={code}" + ) + return PaymentCloseResult.PAID logger.error( f"WeChat close failed: order_no={order.order_no}, " f"code={code}, result={result}" ) - return False + return PaymentCloseResult.FAILED except Exception as e: logger.exception(f"WeChat close exception: order_no={order.order_no}") - return False + return PaymentCloseResult.FAILED async def _query_wechat_order(db: AsyncSession, order: PaymentOrder, db_configs: dict[str, str]) -> dict | None: @@ -699,6 +800,95 @@ async def _query_wechat_order(db: AsyncSession, order: PaymentOrder, db_configs: return None +async def _recover_wechat_paid_order( + db: AsyncSession, + order: PaymentOrder, + db_configs: dict[str, str], +) -> bool: + """Recover one WeChat order after close reports ``ORDERPAID``. + + This is not a periodic query of all pending orders. It runs only after + WeChat itself explicitly says this specific order has already been paid. + The query supplies the authoritative transaction id and amount, then the + existing atomic payment-success handler performs all subscription / upgrade + / credit fulfillment. + """ + local_order_no = str(order.order_no) + data = await _query_wechat_order(db, order, db_configs) + if not data: + logger.warning( + f"WECHAT_ORDERPAID_QUERY_FAILED order_no={local_order_no}" + ) + return False + + trade_state = str(data.get("trade_state") or "").upper() + remote_order_no = str(data.get("out_trade_no") or "") + transaction_id = str(data.get("transaction_id") or "") + amount_info = data.get("amount") or {} + currency = str(amount_info.get("currency") or "CNY").upper() + total_fen = amount_info.get("total") + + if trade_state != "SUCCESS": + logger.warning( + f"WECHAT_ORDERPAID_QUERY_NOT_SUCCESS order_no={local_order_no} " + f"trade_state={trade_state or 'unknown'}" + ) + return False + + if remote_order_no and remote_order_no != local_order_no: + logger.error( + f"WECHAT_ORDERPAID_ORDER_MISMATCH local={local_order_no} " + f"remote={remote_order_no}" + ) + return False + + if currency != "CNY": + logger.error( + f"WECHAT_ORDERPAID_CURRENCY_MISMATCH order_no={local_order_no} " + f"currency={currency}" + ) + return False + + if total_fen is None: + logger.error( + f"WECHAT_ORDERPAID_AMOUNT_MISSING order_no={local_order_no}" + ) + return False + + try: + total_amount = float(to_credit_decimal(total_fen) / to_credit_decimal(100)) + except Exception: + logger.exception( + f"WECHAT_ORDERPAID_AMOUNT_INVALID order_no={local_order_no} " + f"total={total_fen}" + ) + return False + + if not transaction_id: + logger.error( + f"WECHAT_ORDERPAID_TRANSACTION_ID_MISSING order_no={local_order_no}" + ) + return False + + processed = await process_payment_success_by_order_no( + db, + local_order_no, + transaction_id, + total_amount, + ) + if processed: + logger.info( + f"WECHAT_ORDERPAID_RECOVERED order_no={local_order_no} " + f"transaction_id={transaction_id} amount={total_amount}" + ) + else: + logger.warning( + f"WECHAT_ORDERPAID_RECOVERY_NOT_PROCESSED order_no={local_order_no} " + f"transaction_id={transaction_id} amount={total_amount}" + ) + return processed + + async def _refund_wechat_order( db: AsyncSession, order: PaymentOrder, @@ -1388,10 +1578,13 @@ async def process_payment_success_by_order_no( ), ) + log_user_id = str(order.user_id) + log_amount = order.amount + log_credits = order.credits await db.commit() logger.info( - f"PAYMENT_SUCCESS order_no={order_no} user={order.user_id} " - f"amount={order.amount} credits={order.credits} trade_no={trade_no}" + f"PAYMENT_SUCCESS order_no={order_no} user={log_user_id} " + f"amount={log_amount} credits={log_credits} trade_no={trade_no}" ) return True except Exception: