fix: 资金链路并发与幂等加固——余额条件更新原子扣减防超扣，订单结算/支付回调/退款条件更新抢占+Redis锁，防重复扣费与超退

entity/src/main/java/com/kym/entity/common/RedisKeys.java

@@ -50,4 +50,9 @@ public interface RedisKeys {
      * 用户退款申请分布式锁
      */
     String REFUND_LOCK = "REFUND_LOCK:";
+
+    /**
+     * 微信支付回调处理分布式锁（key拼接商户订单号）
+     */
+    String PAY_NOTIFY_LOCK = "PAY_NOTIFY_LOCK:";
 }

service/src/main/java/com/kym/service/miniapp/AccountService.java

@@ -26,4 +26,13 @@ public interface AccountService extends MPJBaseService<Account> {
 
     @Transactional
     void batchHandleDiscountAmount(List<UserRechargeRights> userRechargeRightsList);
+
+    /**
+     * 条件扣减余额（原子操作）：余额充足时扣减并返回true，余额不足返回false
+     *
+     * @param userId 用户id
+     * @param amount 扣减金额（分），必须大于0
+     * @return 是否扣减成功
+     */
+    boolean deductBalance(Long userId, int amount);
 }

service/src/main/java/com/kym/service/miniapp/impl/AccountServiceImpl.java

@@ -90,6 +90,19 @@ public class AccountServiceImpl extends MyBaseServiceImpl<AccountMapper, Account
         return vo;
     }
 
+    @Override
+    public boolean deductBalance(Long userId, int amount) {
+        if (amount <= 0) {
+            return false;
+        }
+        // 条件更新原子扣减：余额充足才扣，防止并发读改写超扣
+        return lambdaUpdate()
+                .setSql("balance = balance - " + amount)
+                .eq(Account::getUserId, userId)
+                .ge(Account::getBalance, amount)
+                .update();
+    }
+
     /**
      * 批量处理用户权益的优惠金额
      *

service/src/main/java/com/kym/service/platform/impl/PlatformNotifyServiceImpl.java

@@ -323,12 +323,34 @@ public class PlatformNotifyServiceImpl implements PlatformNotifyService {
         LOGGER.info("【{}互联互通平台推送】收到充电订单信息推送：{}，解密数据：{}", platformName, json, data);
         var startChargeSeq = data.getString("StartChargeSeq");
         var chargeOrder = chargeOrderService.getChargingOrderByStartChargeSeq(startChargeSeq);
+        if (chargeOrder == null) {
+            LOGGER.error("【{}互联互通平台推送】充电订单信息推送：订单{}不存在，跳过处理", platformName, startChargeSeq);
+            var missParams = """
+                    {
+                        "StartChargeSeq":"%s",
+                        "ConnectorID":"%s",
+                        "ConfirmResult":%d
+                    }
+                    """.formatted(startChargeSeq, data.getString("ConnectorID"), 0);
+            var missPlatform = PlatformCache.INSTANCE.getPlatformByName(platformName);
+            return PlatformAesUtil.encrypt(missPlatform.getUserDataSecret(), missPlatform.getUserDataSecretIv(), missParams);
+        }
 
         // 账户
         var account = accountService.getAccountByUserId(chargeOrder.getUserId());
 
-        // EN+平台推送重试策略是当天失败第二天再推送一次，仅此一次。EN+订单页面可以多次手动推送，所以这里要先判断订单状态，避免重复处理。
-        if (chargeOrder.getChargeStatus() != ChargeOrder.CHARGE_STATUS_已结束 || chargeOrder.getOrderStatus() != ChargeOrder.ORDER_STATUS_成功) {
+        // EN+平台推送重试策略是当天失败第二天再推送一次，仅此一次。EN+订单页面可以多次手动推送，所以这里要防重复结算。
+        // 条件更新抢占结算权：仅当订单未同时处于“已结束+成功”终态时才能更新成功，防止并发/重复推送重复扣费
+        var claimed = chargeOrderService.lambdaUpdate()
+                .set(ChargeOrder::getChargeStatus, ChargeOrder.CHARGE_STATUS_已结束)
+                .set(ChargeOrder::getOrderStatus, ChargeOrder.ORDER_STATUS_成功)
+                .eq(ChargeOrder::getStartChargeSeq, startChargeSeq)
+                .and(w -> w.ne(ChargeOrder::getOrderStatus, ChargeOrder.ORDER_STATUS_成功)
+                        .or().ne(ChargeOrder::getChargeStatus, ChargeOrder.CHARGE_STATUS_已结束))
+                .update();
+        if (!claimed) {
+            LOGGER.info("【{}互联互通平台推送】订单{}已结算过，跳过重复处理", platformName, startChargeSeq);
+        } else {
             // 更新订单信息
             chargeOrder.setStartTime(LocalDateTime.parse(data.getString("StartTime"), DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")))
                     .setEndTime(LocalDateTime.parse(data.getString("EndTime"), DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")))
@@ -446,8 +468,17 @@ public class PlatformNotifyServiceImpl implements PlatformNotifyService {
         chargeOrder.setChargeStatus(ChargeOrder.CHARGE_STATUS_已结束);
         chargeOrderService.updateById(chargeOrder);
 
-        // 账户扣费
-        account.setBalance(account.getBalance() - chargeOrder.getPayAmount());
+        // 账户扣费：条件更新原子扣减，防止并发读改写超扣
+        if (chargeOrder.getPayAmount() > 0 && !accountService.deductBalance(chargeOrder.getUserId(), chargeOrder.getPayAmount())) {
+            // 余额不足时仍需扣减，允许余额为负挂账
+            LOGGER.error("用户：{}余额不足以支付订单：{}实付金额：{}，欠费挂账", chargeOrder.getUserId(), chargeOrder.getStartChargeSeq(), chargeOrder.getPayAmount());
+            accountService.lambdaUpdate()
+                    .setSql("balance = balance - " + chargeOrder.getPayAmount())
+                    .eq(Account::getUserId, chargeOrder.getUserId())
+                    .update();
+        }
+        // 余额已通过SQL原子扣减，置null避免updateById用旧值覆盖；仅落优惠金额等字段
+        account.setBalance(null);
         accountService.updateById(account);
 
         // 记录资金流水

service/src/main/java/com/kym/service/wechat/impl/WxPayServiceImpl.java

@@ -51,6 +51,7 @@ import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.interceptor.TransactionAspectSupport;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.io.ByteArrayOutputStream;
@@ -298,64 +299,77 @@ public class WxPayServiceImpl implements WxPayService {
             // 以支付通知回调为例，验签、解密并转换成 Transaction
             Transaction transaction = ((NotificationParser) notifyRes[1]).parse((RequestParam) notifyRes[0], Transaction.class);
             LOGGER.info("微信支付回调{}：验签解密完毕，数据：\n{}", notifyRes[2], transaction);
-            // 判断是否已经接收处理过通知
+            // 判断是否已经接收处理过通知（快路径）
             if (payLogService.lambdaQuery().eq(PayLog::getOutTradeNo, transaction.getOutTradeNo()).one() != null) {
                 return ResponseEntity.status(HttpStatus.OK).build();
             }
 
-            DateTime dt = DateUtil.parse(transaction.getSuccessTime());
-            LocalDateTime successTime = LocalDateTimeUtil.of(dt);
-
-            // 钱包流水
-            var walletDetail = walletDetailService.getWalletDetailByOrderNo(transaction.getOutTradeNo(), WalletDetail.TYPE_充值);
-            if (walletDetail != null) {
-                walletDetail.setStatus(WalletDetail.STATUS_已确认);  //已确认
-                walletDetail.setCurrency(transaction.getAmount().getCurrency());
-                walletDetail.setAmount(transaction.getAmount().getTotal());
-                walletDetail.setTransactionTime(successTime);
-                walletDetailService.updateById(walletDetail);
-
-                // 更新余额
-                accountService.lambdaUpdate().setSql("balance = (balance+%d)".formatted(transaction.getAmount().getTotal()))
-                        .eq(Account::getUserId, walletDetail.getUserId()).update();
-
-                // 异步处理充值服务费打折权益活动相关逻辑
-                activityService.handleRechargeActivity(walletDetail.getUserId(), transaction.getAmount().getTotal());
-
-                // 支付记录
-                var payLog = new PayLog();
-                payLog.setUserId(walletDetail.getUserId());
-                payLog.setOpenid(transaction.getPayer().getOpenid());
-                payLog.setBankType(transaction.getBankType());
-                payLog.setMchId(transaction.getMchid());
-                payLog.setOutTradeNo(transaction.getOutTradeNo());
-                payLog.setTransactionId(transaction.getTransactionId());
-                payLog.setSuccessTime(successTime);
-                payLog.setTradeType(transaction.getTradeType().name());
-                payLog.setTradeState(transaction.getTradeState().name());
-                payLog.setAttach(transaction.getAttach());
-                payLog.setTotal(transaction.getAmount().getTotal());
-                payLog.setCurrency(transaction.getAmount().getCurrency());
-                payLog.setPayerTotal(transaction.getAmount().getPayerTotal());
-                payLog.setPayerCurrency(transaction.getAmount().getPayerCurrency());
-                payLogService.save(payLog);
-                LOGGER.info("微信支付回调{}：业务处理结束", notifyRes[2]);
-
-                // 如果存在进行中的订单，则向en+更新用户余额（朗新不支持此操作）
-                var chargingOrder = chargeOrderService.getChargingOrderByUserId(walletDetail.getUserId());
-                if (chargingOrder != null && "EN_PLUS".equals(PlatformCache.INSTANCE.getPlatformNameByConnectorId(chargingOrder.getConnectorId()))) {
-                    var account = accountService.getAccountByUserId(walletDetail.getUserId());
-                    // TODO: 2023-11-30 快充这里考虑过充的金额要提高
-                    var res = platformApiService.updateBalanceByQueryEquipChargeStatus(PlatformCache.INSTANCE.getPlatformNameByConnectorId(chargingOrder.getConnectorId()), chargingOrder.getStartChargeSeq(), account.getBalance() - 50);
-                    LOGGER.info("用户：{}充电过程中充值，已更新EN_PLUS充电金额,en+返回数据：{}", account.getUserId(), res);
-                }
+            // Redis锁防并发重复处理（微信会对未应答的通知重试）
+            String notifyLockKey = RedisKeys.PAY_NOTIFY_LOCK + transaction.getOutTradeNo();
+            Boolean locked = stringRedisTemplate.opsForValue().setIfAbsent(notifyLockKey, "1", 30, TimeUnit.SECONDS);
+            if (Boolean.FALSE.equals(locked)) {
+                LOGGER.info("微信支付回调{}：订单{}正在处理中，忽略本次重复通知", notifyRes[2], transaction.getOutTradeNo());
                 return ResponseEntity.status(HttpStatus.OK).build();
+            }
+            try {
+                DateTime dt = DateUtil.parse(transaction.getSuccessTime());
+                LocalDateTime successTime = LocalDateTimeUtil.of(dt);
+
+                // 钱包流水
+                var walletDetail = walletDetailService.getWalletDetailByOrderNo(transaction.getOutTradeNo(), WalletDetail.TYPE_充值);
+                if (walletDetail != null) {
+                    walletDetail.setStatus(WalletDetail.STATUS_已确认);  //已确认
+                    walletDetail.setCurrency(transaction.getAmount().getCurrency());
+                    walletDetail.setAmount(transaction.getAmount().getTotal());
+                    walletDetail.setTransactionTime(successTime);
+                    walletDetailService.updateById(walletDetail);
+
+                    // 更新余额
+                    accountService.lambdaUpdate().setSql("balance = (balance+%d)".formatted(transaction.getAmount().getTotal()))
+                            .eq(Account::getUserId, walletDetail.getUserId()).update();
+
+                    // 异步处理充值服务费打折权益活动相关逻辑
+                    activityService.handleRechargeActivity(walletDetail.getUserId(), transaction.getAmount().getTotal());
+
+                    // 支付记录
+                    var payLog = new PayLog();
+                    payLog.setUserId(walletDetail.getUserId());
+                    payLog.setOpenid(transaction.getPayer().getOpenid());
+                    payLog.setBankType(transaction.getBankType());
+                    payLog.setMchId(transaction.getMchid());
+                    payLog.setOutTradeNo(transaction.getOutTradeNo());
+                    payLog.setTransactionId(transaction.getTransactionId());
+                    payLog.setSuccessTime(successTime);
+                    payLog.setTradeType(transaction.getTradeType().name());
+                    payLog.setTradeState(transaction.getTradeState().name());
+                    payLog.setAttach(transaction.getAttach());
+                    payLog.setTotal(transaction.getAmount().getTotal());
+                    payLog.setCurrency(transaction.getAmount().getCurrency());
+                    payLog.setPayerTotal(transaction.getAmount().getPayerTotal());
+                    payLog.setPayerCurrency(transaction.getAmount().getPayerCurrency());
+                    payLogService.save(payLog);
+                    LOGGER.info("微信支付回调{}：业务处理结束", notifyRes[2]);
+
+                    // 如果存在进行中的订单，则向en+更新用户余额（朗新不支持此操作）
+                    var chargingOrder = chargeOrderService.getChargingOrderByUserId(walletDetail.getUserId());
+                    if (chargingOrder != null && "EN_PLUS".equals(PlatformCache.INSTANCE.getPlatformNameByConnectorId(chargingOrder.getConnectorId()))) {
+                        var account = accountService.getAccountByUserId(walletDetail.getUserId());
+                        // TODO: 2023-11-30 快充这里考虑过充的金额要提高
+                        var res = platformApiService.updateBalanceByQueryEquipChargeStatus(PlatformCache.INSTANCE.getPlatformNameByConnectorId(chargingOrder.getConnectorId()), chargingOrder.getStartChargeSeq(), account.getBalance() - 50);
+                        LOGGER.info("用户：{}充电过程中充值，已更新EN_PLUS充电金额,en+返回数据：{}", account.getUserId(), res);
+                    }
+                    return ResponseEntity.status(HttpStatus.OK).build();
 
-            } else {
-                LOGGER.error("微信支付通知处理异常，资金流水为空，回调信息：{}", transaction);
-                return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(Map.of("code", HttpStatus.INTERNAL_SERVER_ERROR, "message", "资金流水为空"));
+                } else {
+                    LOGGER.error("微信支付通知处理异常，资金流水为空，回调信息：{}", transaction);
+                    return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(Map.of("code", HttpStatus.INTERNAL_SERVER_ERROR, "message", "资金流水为空"));
+                }
+            } finally {
+                stringRedisTemplate.delete(notifyLockKey);
             }
         } catch (Exception e) {
+            // 标记事务回滚，防止异常时部分提交
+            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
             if (e instanceof ValidationException) {
                 // 签名验证失败，返回 401 UNAUTHORIZED 状态码
                 LOGGER.error("微信支付通知验签失败", e);
@@ -669,6 +683,19 @@ public class WxPayServiceImpl implements WxPayService {
     public void wxRefund(long refundLogId) {
         // 通过退款申请id获取退款申请记录
         var refundLog = refundLogService.getById(refundLogId);
+        if (refundLog == null) {
+            throw new BusinessException("退款记录不存在");
+        }
+
+        // 条件更新占位：仅“已申请”状态可发起退款，防止重复提交导致超退
+        var claimed = refundLogService.lambdaUpdate()
+                .set(RefundLog::getStatus, RefundLog.STATUS_退款处理中)
+                .eq(RefundLog::getId, refundLogId)
+                .eq(RefundLog::getStatus, RefundLog.STATUS_退款已申请)
+                .update();
+        if (!claimed) {
+            throw new BusinessException("退款状态不正确，仅已申请状态的退款可发起退款");
+        }
 
         // 钱包流水
         var walletDetail = new WalletDetail()
@@ -740,15 +767,16 @@ public class WxPayServiceImpl implements WxPayService {
 
             //退款日志在申请时插入，接收通知时更新
             var refundLog = refundLogService.lambdaQuery().eq(RefundLog::getOutRefundNo, refundNotification.getOutRefundNo()).one();
-            // 防止重复处理消息
-            if (RefundLog.STATUS_退款成功.equals(refundLog.getStatus())) {
+            if (refundLog == null) {
+                LOGGER.error("微信退款回调{}：退款记录{}不存在，停止重试", notifyRes[2], refundNotification.getOutRefundNo());
                 return ResponseEntity.status(HttpStatus.OK).build();
             }
 
             DateTime dt = DateUtil.parse(refundNotification.getSuccessTime());
             LocalDateTime successTime = LocalDateTimeUtil.of(dt);
 
-            refundLogService.lambdaUpdate()
+            // 条件更新抢占处理权：仅非退款成功状态可更新，防止重复通知重复扣减冻结金额
+            var updated = refundLogService.lambdaUpdate()
                     .set(RefundLog::getRefundId, refundNotification.getRefundId())
                     .set(RefundLog::getTransactionId, refundNotification.getTransactionId())
                     .set(RefundLog::getUserReceivedAccount, refundNotification.getUserReceivedAccount())
@@ -756,7 +784,13 @@ public class WxPayServiceImpl implements WxPayService {
                     .set(RefundLog::getStatus, refundNotification.getRefundStatus().name())
                     .set(RefundLog::getTotal, refundNotification.getAmount().getTotal().intValue())
                     .set(RefundLog::getRefund, refundNotification.getAmount().getRefund().intValue())
-                    .eq(RefundLog::getId, refundLog.getId()).update();
+                    .eq(RefundLog::getId, refundLog.getId())
+                    .ne(RefundLog::getStatus, RefundLog.STATUS_退款成功)
+                    .update();
+            if (!updated) {
+                LOGGER.info("微信退款回调{}：退款记录{}已处理过，忽略本次通知", notifyRes[2], refundNotification.getOutRefundNo());
+                return ResponseEntity.status(HttpStatus.OK).build();
+            }
 
             if (RefundLog.STATUS_退款成功.equals(refundNotification.getRefundStatus().name())) {
                 // 冻结金额扣减此次（退款金额+优惠金额）,优惠金额字段减去申请退款时的优惠金额