feat:批量充值扣款审批回调优化

db/20261102_wallet_recharge_idempotency.md

@@ -0,0 +1,42 @@
+# 批量充值、扣减防重上线说明
+
+本次只修代码和新增防重表。脚本不删除历史流水、不回填防重记录、不调整余额。历史差额、补偿扣款必须另行核账；不能根据 biz_id 重复就直接删除或再次扣钱。
+
+## 修复范围
+
+- 提交、编辑、删除、回调和补记先锁定申请。已经提交或存在资金流水的申请不能再次发起、重置审批字段或更改金额明细；同一发票只能绑定一次。
+- 回调在锁内通过“租户 + 申请 + 流程号 + 审核中”条件变更状态，并检查影响行数。状态、发票、整批余额、流水和防重登记使用同一本地事务。任一明细失败全部回滚；钱包缺失不再跳过部分入账。
+- `addWalletBalance` 在数据库钱包行锁内读取最新余额，检查更新/流水写入结果和整数溢出。类型 2（批量更新余额）与 8（退还各自食堂）按 `tenant_id + wallet_id + biz_type + biz_id` 防重。正金额、负金额、零金额均覆盖。
+- 类型 1 人工调账仍允许使用同一用户业务号多次操作。未对整张 wallet_transaction 添加 biz_id 唯一限制，也未将本次防重规则扩展到所有业务类型。
+- 补记仅允许审批通过的申请，与正常回调使用相同金额方向、流水类型和入账防重。草稿、审核中、拒绝的申请不能补记。
+- 正元同步改到资金事务提交后，并用独立事务执行；失败记录申请号和租户，不再影响已提交的资金。此同步仍为提交后的同步调用，未新增持久化重试队列，也未保证回调响应在 10 秒内完成。
+
+## 历史数据兼容
+
+防重表上线时为空。首次收到旧业务号时，会在数据库当前读中查询相同租户/钱包/类型/业务号的历史流水，包括逻辑删除的记录：
+
+| 情况 | 行为 |
+| --- | --- |
+| 无历史流水 | 余额、流水、防重键一起提交 |
+| 唯一有效流水，金额相同 | 登记原流水，直接返回原结果，余额不变 |
+| 多条流水、金额不同、流水已删除，或防重登记与流水不一致 | 拒绝处理，整笔回滚并记录异常，需要人工核查 |
+
+不要通过删除防重表记录或修改审批状态来“重试”。若正元同步失败，应重试同步用户信息；重放一个已处理回调不会再执行同步。恢复历史数据必须以核账结果为依据。
+
+## 上线顺序
+
+1. 暂停批量充值/扣减提交、审批回调和补记入口，排空在途请求。保留资金数据备份和当前对账结果。
+2. 连接目标钱包库，在该库执行 `20261102_wallet_recharge_idempotency.sql`。确认表为 InnoDB，复合主键与脚本一致。脚本可重复执行；不要对历史流水直接添加全局唯一 biz_id。
+3. 更新并重启**全部钱包服务实例**后恢复入口。新旧版本混用期间，旧实例不会遵守新的防重协议，因此不能在混用时恢复资金写入。
+4. 在预发布环境使用指定测试账户依次验证充值、记账扣款、退还食堂、重复回调、补记，以及中途失败回滚；对照余额、累计金额、流水数量和申请状态。生产验证须使用经业务确认的真实交易，并进行对应对账。
+5. 监控“批量入账金额冲突”“批量入账历史流水异常”“充值回调未找到有效申请”“充值回调失败”和“正元同步失败”日志。异常时保留 `tenantId / applyId / processId / taskId / walletId / bizId`，核查后处理。
+
+回退应用版本也必须先停写并排空请求；旧版本本身没有防重能力。保留新增防重表，不能把删表作为回退步骤。此改动未修改工作流引擎的通知发送机制；历史重复通知或重新发起审批的具体触发原因仍需生产请求/审批日志确认。
+
+## 本地验证
+
+`WalletRechargeIdempotencyMySqlTest` 使用真实 MySQL、InnoDB 行锁、项目租户拦截器/自动填充配置、生产 Mapper SQL 和 Spring 事务代理。仅新建并删除 `codex_recharge_*` 隔离测试库，不读取或改动生产数据。需显式提供 `WALLET_RECHARGE_TEST_JDBC_URL`、`MYSQL_USER`、`MYSQL_PWD`，没有连接时该组测试会跳过，不能将跳过当作通过。
+
+主要覆盖并发/顺序重复入账（正数、负数、零、类型8）、同钱包不同业务并发、旧事务快照、历史重复/删除/金额冲突、流水失败/钱包更新失败回滚、租户隔离、重复人工调账、重复提交/重复回调、整批和发票状态回滚、补记方向与审批限制、发票重复绑定、提交后同步与同步失败隔离。
+
+2026-11-02 本地验证结果：39 项真实 MySQL 用例及 20 项相关回归用例全部通过（共 59 项，0 失败、0 错误、0 跳过）；钱包服务及依赖模块 Maven package 成功。测试库已清理。未部署、未执行生产 SQL。

db/20261102_wallet_recharge_idempotency.sql

@@ -0,0 +1,11 @@
+-- 在部署新版 wallet 服务之前，对钱包库执行。只新增空表，不清理或修正历史流水、余额。
+CREATE TABLE IF NOT EXISTS wallet_transaction_idempotency (
+    tenant_id BIGINT NOT NULL,
+    wallet_id BIGINT NOT NULL,
+    biz_type VARCHAR(8) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '仅批量更新2、退还食堂8',
+    biz_id VARCHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
+    price INT NOT NULL COMMENT '有符号变动金额，分',
+    transaction_id BIGINT NOT NULL,
+    create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    PRIMARY KEY (tenant_id, wallet_id, biz_type, biz_id)
+) ENGINE=InnoDB COMMENT='批量钱包入账幂等键，与余额和流水同事务提交';

db/800pjcloudx_wallet.sql

@@ -388,6 +388,18 @@ CREATE TABLE wallet_transaction
   DEFAULT CHARSET = utf8mb4 COMMENT ='会员钱包流水表';
 CREATE INDEX idx_dept_id ON wallet_transaction (wallet_id);
 
+-- 批量充值/扣减持久化防重；已有环境执行 20261102_wallet_recharge_idempotency.sql。
+CREATE TABLE IF NOT EXISTS wallet_transaction_idempotency (
+    tenant_id BIGINT NOT NULL,
+    wallet_id BIGINT NOT NULL,
+    biz_type VARCHAR(8) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '仅批量更新2、退还食堂8',
+    biz_id VARCHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
+    price INT NOT NULL COMMENT '有符号变动金额，分',
+    transaction_id BIGINT NOT NULL,
+    create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    PRIMARY KEY (tenant_id, wallet_id, biz_type, biz_id)
+) ENGINE=InnoDB COMMENT='批量钱包入账幂等键，与余额和流水同事务提交';
+
 -- ----------------------------
 -- Table structure for wallet_recharge_apply
 -- ----------------------------

pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/controller/recharge/WalletRechargeImportController.java

@@ -96,7 +96,7 @@ public class WalletRechargeImportController {
     @PostMapping
     @PreAuthorize("@pms.hasPermission('wallet_walletRechargeImport_add')")
     public R save(@RequestBody WalletRechargeImportEntity walletRechargeImport) {
-        return R.ok(walletRechargeImportService.save(walletRechargeImport));
+        return R.ok(walletRechargeImportService.saveDraft(walletRechargeImport));
     }
 
     @Operation(summary = "修改充值导入表", description = "修改充值导入表")
@@ -104,7 +104,7 @@ public class WalletRechargeImportController {
     @PutMapping
     @PreAuthorize("@pms.hasPermission('wallet_walletRechargeImport_edit')")
     public R updateById(@RequestBody WalletRechargeImportEntity walletRechargeImport) {
-        return R.ok(walletRechargeImportService.updateById(walletRechargeImport));
+        return R.ok(walletRechargeImportService.updateDraft(walletRechargeImport));
     }
 
     @Operation(summary = "通过id删除充值导入表", description = "通过id删除充值导入表")
@@ -112,7 +112,7 @@ public class WalletRechargeImportController {
     @DeleteMapping
     @PreAuthorize("@pms.hasPermission('wallet_walletRechargeImport_del')")
     public R removeById(@RequestBody Long[] ids) {
-        return R.ok(walletRechargeImportService.removeBatchByIds(CollUtil.toList(ids)));
+        return R.ok(walletRechargeImportService.removeDrafts(CollUtil.toList(ids)));
     }
 
     @PostMapping("/export")
@@ -127,4 +127,4 @@ public class WalletRechargeImportController {
     public R<com.bsd.pjcloud.common.asynctask.vo.AsyncTaskProgressVO> exportAsync(@RequestBody WalletRechargeImportDto dto) {
         return R.ok(batchExportService.submitImport(dto), "导出任务已提交");
     }
-}
+}

pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/mapper/WalletTransactionIdempotencyMapper.java

@@ -0,0 +1,53 @@
+package com.bsd.pjcloud.wallet.mapper;
+
+import com.baomidou.mybatisplus.annotation.InterceptorIgnore;
+import com.bsd.pjcloud.wallet.api.entity.WalletTransactionEntity;
+import lombok.Data;
+import org.apache.ibatis.annotations.*;
+
+import java.util.List;
+
+/** 批量资金变动的持久化业务键。所有 SQL 显式限定租户，历史查询包含逻辑删除记录。 */
+@Mapper
+@InterceptorIgnore(tenantLine = "true")
+public interface WalletTransactionIdempotencyMapper {
+
+    @Insert("""
+            INSERT INTO wallet_transaction_idempotency(tenant_id,wallet_id,biz_type,biz_id,price,transaction_id)
+            VALUES(#{tenantId},#{walletId},#{bizType},#{bizId},#{price},#{transactionId})
+            ON DUPLICATE KEY UPDATE biz_id=wallet_transaction_idempotency.biz_id
+            """)
+    int reserve(@Param("tenantId") Long tenantId, @Param("walletId") Long walletId,
+                @Param("bizType") String bizType, @Param("bizId") String bizId,
+                @Param("price") Integer price, @Param("transactionId") Long transactionId);
+
+    @Select("""
+            SELECT price,transaction_id FROM wallet_transaction_idempotency
+            WHERE tenant_id=#{tenantId} AND wallet_id=#{walletId} AND biz_type=#{bizType} AND biz_id=#{bizId}
+            FOR UPDATE
+            """)
+    Entry lock(@Param("tenantId") Long tenantId, @Param("walletId") Long walletId,
+               @Param("bizType") String bizType, @Param("bizId") String bizId);
+
+    // 当前读，避免外层 REPEATABLE READ 事务的旧快照遗漏已提交流水；禁止删除流水后重新入账。
+    @Select("""
+            SELECT * FROM wallet_transaction
+            WHERE tenant_id=#{tenantId} AND wallet_id=#{walletId} AND biz_type=#{bizType} AND biz_id=#{bizId}
+            ORDER BY id LIMIT 2 FOR UPDATE
+            """)
+    List<WalletTransactionEntity> lockHistory(@Param("tenantId") Long tenantId, @Param("walletId") Long walletId,
+                                              @Param("bizType") String bizType, @Param("bizId") String bizId);
+
+    @Update("""
+            UPDATE wallet_transaction_idempotency SET transaction_id=#{transactionId}
+            WHERE tenant_id=#{tenantId} AND wallet_id=#{walletId} AND biz_type=#{bizType} AND biz_id=#{bizId}
+            """)
+    int bind(@Param("tenantId") Long tenantId, @Param("walletId") Long walletId,
+             @Param("bizType") String bizType, @Param("bizId") String bizId, @Param("transactionId") Long transactionId);
+
+    @Data
+    class Entry {
+        private Integer price;
+        private Long transactionId;
+    }
+}

pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/mapper/recharge/WalletRechargeApplyMapper.java

@@ -17,12 +17,29 @@ import com.github.yulichang.wrapper.MPJLambdaWrapper;
 
 import org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Select;
+import com.baomidou.mybatisplus.annotation.InterceptorIgnore;
 
 import java.util.List;
 
 @Mapper
 public interface WalletRechargeApplyMapper extends PjcloudBaseMapper<WalletRechargeApplyEntity> {
 
+    default WalletRechargeApplyEntity lockById(Long id) {
+        return selectOne(new LambdaQueryWrapper<WalletRechargeApplyEntity>()
+                .eq(WalletRechargeApplyEntity::getId, id).last("FOR UPDATE"));
+    }
+
+    /** 即使历史流水/导入记录已逻辑删除，也不允许把已经入账的申请重置后再次发起。 */
+    @InterceptorIgnore(tenantLine = "true")
+    @Select("""
+            SELECT t.id FROM wallet_transaction t JOIN wallet_recharge_import i
+              ON t.biz_id=CAST(i.id AS CHAR CHARACTER SET ascii) AND t.tenant_id=i.tenant_id
+            WHERE i.apply_id=#{applyId} AND i.tenant_id=#{tenantId} AND t.biz_type IN ('2','8')
+            LIMIT 1 FOR UPDATE
+            """)
+    List<Long> lockPostedTransactions(@Param("applyId") Long applyId, @Param("tenantId") Long tenantId);
+
 
     Page<WalletRechargeApplyVo> scopePageList(Page page, @Param("dto") WalletRechargeApplyDto dto, DataScope dataScope);
 
@@ -60,4 +77,4 @@ public interface WalletRechargeApplyMapper extends PjcloudBaseMapper<WalletRecha
         list.setRecords(BeanUtils.toBean(list.getRecords(), WalletRechargeApplyVo.class));
         return list;
     }
-}
+}

pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/WalletRechargeImportService.java

@@ -24,4 +24,11 @@ public interface WalletRechargeImportService extends IService<WalletRechargeImpo
     List<WalletRechargeImportVo> scopeList(WalletRechargeImportDto dto, DataScope dataScope);
 
     Boolean recover(Long id);
-}
+
+    /** 后台人工维护入口，仅允许修改尚未提交审批的明细。 */
+    boolean saveDraft(WalletRechargeImportEntity entity);
+
+    boolean updateDraft(WalletRechargeImportEntity entity);
+
+    boolean removeDrafts(List<Long> ids);
+}

pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/WalletTransactionServiceImpl.java

@@ -13,6 +13,7 @@ import com.alibaba.fastjson.JSON;
 import com.baomidou.lock.annotation.Lock4j;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
+import com.baomidou.mybatisplus.core.toolkit.IdWorker;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.bsd.pjcloud.admin.api.entity.SysDept;
@@ -25,6 +26,7 @@ import com.bsd.pjcloud.common.core.util.BeanUtils;
 import com.bsd.pjcloud.common.core.util.CollectionUtils;
 import com.bsd.pjcloud.common.core.util.R;
 import com.bsd.pjcloud.common.data.datascope.DataScope;
+import com.bsd.pjcloud.common.data.tenant.TenantContextHolder;
 import com.bsd.pjcloud.hard.api.constant.HardTypeEnum;
 import com.bsd.pjcloud.wallet.api.constant.WalletBizEnum;
 import com.bsd.pjcloud.wallet.api.constant.WalletTypeEnum;
@@ -47,6 +49,7 @@ import com.bsd.pjcloud.wallet.api.vo.mch.WalletMchDeptViewVo;
 import com.bsd.pjcloud.wallet.api.vo.transaction.WalletTransactionVo;
 import com.bsd.pjcloud.wallet.mapper.WalletManageMapper;
 import com.bsd.pjcloud.wallet.mapper.WalletTransactionMapper;
+import com.bsd.pjcloud.wallet.mapper.WalletTransactionIdempotencyMapper;
 import com.bsd.pjcloud.wallet.mapper.WalletTypeMapper;
 import com.bsd.pjcloud.wallet.mapper.mch.view.MchDeptViewMapper;
 import com.bsd.pjcloud.wallet.service.MchManageService;
@@ -70,6 +73,7 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -98,6 +102,7 @@ public class WalletTransactionServiceImpl extends ServiceImpl<WalletTransactionM
     private final WalletThirdZyService walletThirdZyService;
 
     private final WalletManageMapper walletManageMapper;
+    private final WalletTransactionIdempotencyMapper idempotencyMapper;
     private final WalletTypeMapper walletTypeMapper;
     private final MchDeptViewMapper mchDeptViewMapper;
 
@@ -179,19 +184,34 @@ public class WalletTransactionServiceImpl extends ServiceImpl<WalletTransactionM
     // @CacheEvict(value = CacheConstants.WALLET_DETAIL, key = "#walletId")
     @Lock4j(keys = {"'add_wallet_balance'", "#walletId", "#bizId"}, acquireTimeout = 5000, expire = 60000)
     public WalletTransactionEntity addWalletBalance(Long walletId, Integer price, String bizId, WalletBizEnum bizType, String tags, String remark) {
-        WalletManageEntity wallet = walletManageService.getById(walletId);
+        if (walletId == null || price == null || bizType == null) {
+            throw new IllegalArgumentException("钱包、金额和业务类型不能为空");
+        }
+        // 数据库锁覆盖完整本地事务。Redis 锁到期或释放后也不能让两个业务使用同一个旧余额。
+        WalletManageEntity wallet = walletManageMapper.selectOne(new LambdaQueryWrapper<WalletManageEntity>()
+                .eq(WalletManageEntity::getId, walletId).last("FOR UPDATE"));
         if (ObjectUtil.isEmpty(wallet)) {
             log.error("|{}|，用户钱包id({})不存在.", TAG, walletId);
             throw new RuntimeException("扣款失败,用户钱包不存在");
         }
+        Long transactionId = IdWorker.getId();
+        boolean batchPosting = bizType == WalletBizEnum.BATCH || bizType == WalletBizEnum.RETURN_CANTEEN;
+        if (batchPosting) {
+            WalletTransactionEntity previous = reserveBatchPosting(wallet, price, bizId, bizType, transactionId);
+            if (previous != null) return previous;
+        }
+        int afterBalance = Math.addExact(wallet.getBalance(), price);
+        int updated;
         switch (bizType) {
             case UPDATE:
             case RECHARGE, BATCH, TRANSFER_ACCOUNTS: {
-                walletManageMapper.updateWhenRecharge(wallet.getId(), price, wallet.getVersion());
+                Math.addExact(wallet.getTotalRecharge(), price);
+                updated = walletManageMapper.updateWhenRecharge(wallet.getId(), price, wallet.getVersion());
                 break;
             }
             case REFUND, TRANSFER_ACCOUNTS_REFUND, RETURN_CANTEEN: {
-                walletManageMapper.updateWhenConsumptionRefund(wallet.getId(), price);
+                Math.subtractExact(wallet.getTotalExpense(), price);
+                updated = walletManageMapper.updateWhenConsumptionRefund(wallet.getId(), price);
                 break;
             }
             default: {
@@ -199,12 +219,15 @@ public class WalletTransactionServiceImpl extends ServiceImpl<WalletTransactionM
                 throw new UnsupportedOperationException("待实现");
             }
         }
+        if (updated != 1) throw new IllegalStateException("钱包更新失败，请重试");
         WalletTransactionEntity insetObj = new WalletTransactionEntity();
+        insetObj.setId(transactionId);
+        insetObj.setTenantId(wallet.getTenantId());
         insetObj.setUserId(wallet.getUserId());
         insetObj.setDeptId(wallet.getDeptId());
         insetObj.setWalletId(walletId);
         insetObj.setPrice(price);
-        insetObj.setBalance(wallet.getBalance() + price);
+        insetObj.setBalance(afterBalance);
         insetObj.setBizId(String.valueOf(bizId));
         insetObj.setBizType(bizType.getCode());
         insetObj.setTitle(bizType.getDesc());
@@ -212,10 +235,49 @@ public class WalletTransactionServiceImpl extends ServiceImpl<WalletTransactionM
         insetObj.setTags(tags);
         String normalizedRemark = remark == null ? null : remark.strip();
         insetObj.setRemark(ObjectUtil.isEmpty(normalizedRemark) ? null : normalizedRemark);
-        save(insetObj);
+        if (!save(insetObj)) throw new IllegalStateException("钱包流水写入失败");
         return insetObj;
     }
 
+    /** 仅类型2/8具有一次性导入业务号；人工调账类型1允许同一用户多次操作，不能全表唯一 biz_id。 */
+    private WalletTransactionEntity reserveBatchPosting(WalletManageEntity wallet, Integer price, String bizId,
+                                                         WalletBizEnum bizType, Long transactionId) {
+        Long tenantId = TenantContextHolder.getTenantId();
+        if (tenantId == null || !Objects.equals(tenantId, wallet.getTenantId())
+                || Boolean.TRUE.equals(TenantContextHolder.getTenantSkip())) {
+            throw new IllegalStateException("批量入账租户上下文错误");
+        }
+        if (bizId == null || bizId.isBlank() || bizId.length() > 64 || "null".equals(bizId)) {
+            throw new IllegalArgumentException("批量入账业务号错误");
+        }
+        idempotencyMapper.reserve(tenantId, wallet.getId(), bizType.getCode(), bizId, price, transactionId);
+        WalletTransactionIdempotencyMapper.Entry entry = idempotencyMapper.lock(tenantId, wallet.getId(), bizType.getCode(), bizId);
+        if (entry == null || !Objects.equals(price, entry.getPrice())) {
+            log.error("批量入账金额冲突 tenantId={}, walletId={}, bizType={}, bizId={}, requestPrice={}, storedPrice={}",
+                    tenantId, wallet.getId(), bizType.getCode(), bizId, price, entry == null ? null : entry.getPrice());
+            throw new IllegalStateException("同一业务号的金额发生变化，已停止入账，请人工核查");
+        }
+        List<WalletTransactionEntity> history = idempotencyMapper.lockHistory(tenantId, wallet.getId(), bizType.getCode(), bizId);
+        if (history.isEmpty() && Objects.equals(entry.getTransactionId(), transactionId)) return null;
+        if (history.size() != 1 || !Objects.equals(price, history.get(0).getPrice())
+                || !"0".equals(history.get(0).getDelFlag())
+                || (!Objects.equals(entry.getTransactionId(), transactionId)
+                    && !Objects.equals(entry.getTransactionId(), history.get(0).getId()))) {
+            log.error("批量入账历史流水异常 tenantId={}, walletId={}, bizType={}, bizId={}, registeredTransactionId={}, historyIds={}",
+                    tenantId, wallet.getId(), bizType.getCode(), bizId, entry.getTransactionId(),
+                    history.stream().map(WalletTransactionEntity::getId).toList());
+            throw new IllegalStateException("业务流水存在重复、删除或不一致，已停止入账，请人工核查");
+        }
+        WalletTransactionEntity previous = history.get(0);
+        if (Objects.equals(entry.getTransactionId(), transactionId)
+                && idempotencyMapper.bind(tenantId, wallet.getId(), bizType.getCode(), bizId, previous.getId()) != 1) {
+            throw new IllegalStateException("业务流水防重登记失败");
+        }
+        log.info("批量入账重复请求已忽略 tenantId={}, walletId={}, bizType={}, bizId={}, transactionId={}",
+                tenantId, wallet.getId(), bizType.getCode(), bizId, previous.getId());
+        return previous;
+    }
+
     @Override
     // @CacheEvict(value = CacheConstants.WALLET_DETAIL, key = "#dto.getWalletId()")
     public Boolean updateWalletBalance(WalletTransactionDto dto) {

pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/recharge/WalletRechargeAfterCommitSyncService.java

@@ -0,0 +1,52 @@
+package com.bsd.pjcloud.wallet.service.impl.recharge;
+
+import com.bsd.pjcloud.common.data.tenant.TenantContextHolder;
+import com.bsd.pjcloud.wallet.service.WalletThirdZyService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import java.util.List;
+
+/** 资金落库后再同步第三方，失败不能回滚已入账资金，也不能占用资金行锁。 */
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class WalletRechargeAfterCommitSyncService {
+    private final WalletThirdZyService walletThirdZyService;
+    private final PlatformTransactionManager transactionManager;
+
+    public void register(Long applyId, Long tenantId, List<Long> userIds) {
+        if (!TransactionSynchronizationManager.isActualTransactionActive()
+                || !TransactionSynchronizationManager.isSynchronizationActive()) {
+            throw new IllegalStateException("资金同步必须在入账事务中注册");
+        }
+        List<Long> users = List.copyOf(userIds);
+        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
+            @Override
+            public void afterCommit() {
+                Long previousTenant = TenantContextHolder.getTenantId();
+                boolean previousSkip = TenantContextHolder.getTenantSkip();
+                try {
+                    TenantContextHolder.setTenantId(tenantId);
+                    TenantContextHolder.setTenantSkip(false);
+                    // afterCommit 仍绑定旧事务资源；第三方同步可能创建钱包，必须使用独立事务。
+                    TransactionTemplate transaction = new TransactionTemplate(transactionManager);
+                    transaction.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
+                    transaction.executeWithoutResult(status -> walletThirdZyService.syncAllUser(users));
+                } catch (Exception failure) {
+                    log.error("充值入账已提交，正元同步失败，需重试同步 applyId={}, tenantId={}, userCount={}",
+                            applyId, tenantId, users.size(), failure);
+                } finally {
+                    TenantContextHolder.setTenantId(previousTenant);
+                    TenantContextHolder.setTenantSkip(previousSkip);
+                }
+            }
+        });
+    }
+}

pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/recharge/WalletRechargeApplyServiceImpl.java

@@ -5,7 +5,6 @@ import cn.hutool.core.date.LocalDateTimeUtil;
 import cn.hutool.core.io.resource.ResourceUtil;
 import cn.hutool.core.lang.Validator;
 import cn.hutool.core.util.ObjectUtil;
-import cn.hutool.extra.spring.SpringUtil;
 
 import com.alibaba.excel.EasyExcel;
 import com.alibaba.excel.ExcelWriter;
@@ -13,6 +12,7 @@ import com.alibaba.excel.write.metadata.WriteSheet;
 import com.alibaba.excel.write.metadata.fill.FillConfig;
 import com.alibaba.fastjson.JSON;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.bsd.pjcloud.admin.api.dto.UserInfo;
@@ -105,12 +105,13 @@ public class WalletRechargeApplyServiceImpl extends ServiceImpl<WalletRechargeAp
     private final WalletManageService walletManageService;
     private final WalletTransactionService walletTransactionService;
     private final WalletTypeService walletTypeService;
-    private final WalletThirdZyService walletThirdZyService;
+    private final WalletRechargeAfterCommitSyncService afterCommitSyncService;
 
     private final WalletTransactionMapper walletTransactionMapper;
     private final WalletManageMapper walletManageMapper;
     private final WalletTypeMapper walletTypeMapper;
     private final WalletRechargeInvoiceService walletRechargeInvoiceService;
+    private final WalletRechargeInvoiceMapper walletRechargeInvoiceMapper;
 
     private final RemoteUserService userService;
     private final RemoteDeptService deptService;
@@ -125,7 +126,8 @@ public class WalletRechargeApplyServiceImpl extends ServiceImpl<WalletRechargeAp
     @Transactional(rollbackFor = Exception.class)
     public void saveTopup(WalletRechargeSaveTopupDto dto) {
         // 查询一下发票记录
-        WalletRechargeInvoiceEntity invoice = walletRechargeInvoiceService.getById(dto.getInvoiceId());
+        WalletRechargeInvoiceEntity invoice = walletRechargeInvoiceMapper.selectOne(new LambdaQueryWrapper<WalletRechargeInvoiceEntity>()
+                .eq(WalletRechargeInvoiceEntity::getId, dto.getInvoiceId()).last("FOR UPDATE"));
         if (ObjectUtil.isEmpty(invoice)) throw new RuntimeException("发票记录不存在");
         if (ObjectUtil.isNotEmpty(invoice.getApplyId())) throw new RuntimeException("该发票已被绑定");
         if (ObjectUtil.notEqual(invoice.getProcessStatus(), TaskCheckStatusEnum.RESULT.getCode())) {
@@ -140,7 +142,7 @@ public class WalletRechargeApplyServiceImpl extends ServiceImpl<WalletRechargeAp
         apply.setDeptId(invoice.getApplyDeptId());
         apply.setImportTime(invoice.getCreateTime().toLocalDate());
         apply.setRemark(invoice.getRemark());
-        baseMapper.insert(apply);
+        if (baseMapper.insert(apply) != 1) throw new IllegalStateException("新增充值申请失败");
         // 更新import记录表的applyId
         List<WalletRechargeImportEntity> list = rechargeImportService.list(new LambdaQueryWrapper<WalletRechargeImportEntity>().eq(WalletRechargeImportEntity::getInvoiceId, invoice.getId()).isNull(WalletRechargeImportEntity::getApplyId));
         if (ObjectUtil.isNotEmpty(list)) {
@@ -150,14 +152,14 @@ public class WalletRechargeApplyServiceImpl extends ServiceImpl<WalletRechargeAp
                 item.setApplyId(apply.getId());
                 return item;
             }).toList();
-            rechargeImportService.updateBatchById(updateObjList);
+            if (!rechargeImportService.updateBatchById(updateObjList)) throw new IllegalStateException("绑定充值明细失败");
         }
         // 更新发票状态
         WalletRechargeInvoiceEntity updateObj = new WalletRechargeInvoiceEntity();
         updateObj.setId(invoice.getId());
         updateObj.setStatus(WalletRechargeInvoiceStatusEnum.BINDED.getCode());
         updateObj.setApplyId(apply.getId());
-        walletRechargeInvoiceService.updateById(updateObj);
+        if (!walletRechargeInvoiceService.updateById(updateObj)) throw new IllegalStateException("绑定发票失败");
     }
 
     @Override
@@ -171,16 +173,24 @@ public class WalletRechargeApplyServiceImpl extends ServiceImpl<WalletRechargeAp
     @Override
     @Transactional(rollbackFor = Exception.class)
     public Boolean edit(WalletRechargeApplyDto dto) {
-        WalletRechargeApplyEntity detail = baseMapper.selectById(dto.getId());
+        WalletRechargeApplyEntity detail = baseMapper.lockById(dto.getId());
         if (ObjectUtil.isEmpty(detail)) {
             throw new RuntimeException("申请不存在");
         }
+        requireUnsubmitted(detail);
         if (ObjectUtil.equals(dto.getIsAuthorization(), YesNoEnum.NO.getCode())) {
             throw new RuntimeException("请同意授权协议");
         }
-        WalletRechargeApplyEntity updateObj = BeanUtils.toBean(dto, WalletRechargeApplyEntity.class);
-        baseMapper.updateById(updateObj);
-        getSelf().updateBeforePrice(baseMapper.selectById(dto.getId()));
+        if (dto.getBizType() != null && (WalletBizRechargeEnum.valueOfByType(dto.getBizType()) == null
+                || (detail.getInvoiceId() != null && !dto.getBizType().equals(detail.getBizType())))) {
+            throw new IllegalArgumentException("充值业务类型错误，发票关联申请不能变更业务类型");
+        }
+        // 只允许编辑业务字段，不接受客户端传入的流程号、审批状态、租户和申请人。
+        WalletRechargeApplyEntity updateObj = WalletRechargeApplyEntity.builder().id(detail.getId())
+                .walletTypeId(dto.getWalletTypeId()).importTime(dto.getImportTime()).bizType(dto.getBizType())
+                .remark(dto.getRemark()).isAuthorization(dto.getIsAuthorization()).build();
+        if (baseMapper.updateById(updateObj) != 1) throw new IllegalStateException("编辑申请失败");
+        updateBeforePrice(baseMapper.selectById(dto.getId()));
         return true;
     }
 
@@ -192,10 +202,11 @@ public class WalletRechargeApplyServiceImpl extends ServiceImpl<WalletRechargeAp
     @Transactional(rollbackFor = Exception.class)
     public void delByIds(ArrayList<Long> list) {
         if (ObjectUtil.isEmpty(list)) return;
-        for (Long v : list) {
+        for (Long v : list.stream().distinct().sorted().toList()) {
             // 只有未审批和审核中能删除
-            WalletRechargeApplyEntity detail = baseMapper.selectById(v);
+            WalletRechargeApplyEntity detail = baseMapper.lockById(v);
             if (ObjectUtil.isEmpty(detail)) continue;
+            requireNoPostedTransactions(detail);
             if (ObjectUtil.isNotEmpty(detail.getProcessStatus()) && ObjectUtil.notEqual(detail.getProcessStatus(), TaskCheckStatusEnum.PADDING.getCode())) {
                 UserInfo userInfo = userService.getByUserId(detail.getUserId()).getData();
                 throw new RuntimeException("用户<" + userInfo.getSysUser().getName() + ">发起的审批不能删除");
@@ -211,10 +222,10 @@ public class WalletRechargeApplyServiceImpl extends ServiceImpl<WalletRechargeAp
                 WalletRechargeInvoiceEntity updateObj = new WalletRechargeInvoiceEntity();
                 updateObj.setId(detail.getInvoiceId());
                 updateObj.setStatus(WalletRechargeInvoiceStatusEnum.DEPT_CANCELLATION.getCode());
-                walletRechargeInvoiceService.updateById(updateObj);
+                if (!walletRechargeInvoiceService.updateById(updateObj)) throw new IllegalStateException("发票状态更新失败");
             }
             // 删除申请记录
-            baseMapper.deleteById(v);
+            if (baseMapper.deleteById(v) != 1) throw new IllegalStateException("删除申请失败");
             // 删除申请记录条目
             rechargeImportService.remove(new LambdaQueryWrapper<WalletRechargeImportEntity>().eq(WalletRechargeImportEntity::getApplyId, v));
         }
@@ -256,10 +267,11 @@ public class WalletRechargeApplyServiceImpl extends ServiceImpl<WalletRechargeAp
     @Transactional(rollbackFor = Exception.class)
     public void apply(Long id) {
         // 查询详情
-        WalletRechargeApplyEntity detail = baseMapper.selectById(id);
+        WalletRechargeApplyEntity detail = baseMapper.lockById(id);
         if (ObjectUtil.isEmpty(detail)) {
             throw new RuntimeException("充值申请不存在");
         }
+        requireUnsubmitted(detail);
         if (!ObjectUtil.isAllNotEmpty(detail.getImportTime(), detail.getWalletTypeId())) {
             throw new RuntimeException("请先编辑导入时间等信息");
         }
@@ -296,21 +308,22 @@ public class WalletRechargeApplyServiceImpl extends ServiceImpl<WalletRechargeAp
         flowDto.setImportTime(LocalDateTimeUtil.format(detail.getImportTime(), DatePattern.NORM_MONTH_PATTERN));
         flowDto.setRemark(detail.getRemark());
         String processId = flowUtil.startFlow("TASK_RECHARGE_APPLY_FLOW", user.getSysUser(), detail.getId(), flowDto);
+        if (processId == null || processId.isBlank()) throw new IllegalStateException("发起审批失败，未返回流程号");
         WalletRechargeApplyEntity updateObj = new WalletRechargeApplyEntity();
         updateObj.setId(detail.getId());
         updateObj.setProcessId(processId);
         updateObj.setProcessStatus(TaskCheckStatusEnum.PADDING.getCode());
         updateObj.setApplyTime(LocalDateTime.now());
-        baseMapper.updateById(updateObj);
+        if (baseMapper.updateById(updateObj) != 1) throw new IllegalStateException("保存审批流程失败");
         // 如果有绑定的发票
         if (ObjectUtil.isNotEmpty(detail.getInvoiceId())) {
             WalletRechargeInvoiceEntity invoiceUpdate = new WalletRechargeInvoiceEntity();
             invoiceUpdate.setId(detail.getInvoiceId());
             invoiceUpdate.setStatus(WalletRechargeInvoiceStatusEnum.UNCHECKED.getCode());
-            walletRechargeInvoiceService.updateById(invoiceUpdate);
+            if (!walletRechargeInvoiceService.updateById(invoiceUpdate)) throw new IllegalStateException("发票状态更新失败");
         }
         // 更新之前的总金额
-        getSelf().updateBeforePrice(detail);
+        updateBeforePrice(detail);
     }
 
     @Override
@@ -435,69 +448,97 @@ public class WalletRechargeApplyServiceImpl extends ServiceImpl<WalletRechargeAp
     @Override
     @Transactional(rollbackFor = Exception.class)
     public void applyCallback(ProcessNodeRecordAssignUserParamDto dto, Long tenantId) {
-        if (!flowUtil.checkCallback(dto)) {
-            return;
-        }
-        if (ObjectUtil.isNotEmpty(tenantId) && ObjectUtil.notEqual(tenantId, 1L)) {
-            TenantContextHolder.setTenantId(tenantId);
+        if (dto == null || !Boolean.TRUE.equals(flowUtil.checkCallback(dto))
+                || ObjectUtil.isEmpty(dto.getProcessInstanceId()) || ObjectUtil.isEmpty(dto.getNodeId())) {
+            throw new IllegalArgumentException("充值审批回调参数不完整");
         }
+        Long previousTenant = TenantContextHolder.getTenantId();
+        boolean previousSkip = TenantContextHolder.getTenantSkip();
+        Long callbackTenant = tenantId == null ? previousTenant : tenantId;
+        if (callbackTenant == null) throw new IllegalArgumentException("充值审批回调缺少租户");
         String processId = dto.getProcessInstanceId();
-        WalletRechargeApplyEntity detail = baseMapper.selectOne(new LambdaQueryWrapper<WalletRechargeApplyEntity>().eq(WalletRechargeApplyEntity::getProcessId, processId));
-        if (ObjectUtil.isEmpty(detail)) {
-            return;
-        }
-        if (!ObjectUtil.equals(detail.getProcessStatus(), TaskCheckStatusEnum.PADDING.getCode())) {
-            log.error("|{}|重复的回调,dto:|{}|", TAG, JSON.toJSONString(dto));
-            return;
-        }
         try {
-            Boolean isConsent = flowUtil.isConsent(dto);
+            TenantContextHolder.setTenantId(callbackTenant);
+            TenantContextHolder.setTenantSkip(false);
+            WalletRechargeApplyEntity detail = baseMapper.selectOne(new LambdaQueryWrapper<WalletRechargeApplyEntity>()
+                    .eq(WalletRechargeApplyEntity::getTenantId, callbackTenant)
+                    .eq(WalletRechargeApplyEntity::getProcessId, processId).last("FOR UPDATE"));
+            if (detail == null) {
+                log.warn("充值回调未找到有效申请 tenantId={}, processId={}, taskId={}", callbackTenant, processId, dto.getTaskId());
+                return;
+            }
+            if (!TaskCheckStatusEnum.PADDING.getCode().equals(detail.getProcessStatus())) {
+                log.info("充值回调已处理，忽略重复通知 applyId={}, processId={}, taskId={}, status={}",
+                        detail.getId(), processId, dto.getTaskId(), detail.getProcessStatus());
+                return;
+            }
+            // 无效通知不能当作驳回，否则会永久抢占合法回调。
+            com.alibaba.fastjson.JSONObject data = JSON.parseObject(dto.getData());
+            Object decision = data == null ? null : data.get(dto.getNodeId() + "_approve_condition");
+            if (decision == null || !("true".equalsIgnoreCase(decision.toString()) || "false".equalsIgnoreCase(decision.toString()))) {
+                throw new IllegalArgumentException("充值审批回调缺少有效审批结论");
+            }
+            boolean isConsent = Boolean.parseBoolean(decision.toString());
             String processStatus = isConsent ? TaskCheckStatusEnum.RESULT.getCode() : TaskCheckStatusEnum.REJECT.getCode();
-            baseMapper.updateById(WalletRechargeApplyEntity.builder().id(detail.getId()).processTime(LocalDateTime.now()).processStatus(processStatus).build());
-
-            // 如果有绑定的发票
+            int claimed = baseMapper.update(WalletRechargeApplyEntity.builder().processTime(LocalDateTime.now()).processStatus(processStatus).build(),
+                    Wrappers.<WalletRechargeApplyEntity>lambdaUpdate().eq(WalletRechargeApplyEntity::getId, detail.getId())
+                            .eq(WalletRechargeApplyEntity::getTenantId, callbackTenant)
+                            .eq(WalletRechargeApplyEntity::getProcessId, processId)
+                            .eq(WalletRechargeApplyEntity::getProcessStatus, TaskCheckStatusEnum.PADDING.getCode()));
+            if (claimed != 1) throw new IllegalStateException("审批状态变更失败，请重试");
             if (ObjectUtil.isNotEmpty(detail.getInvoiceId())) {
                 WalletRechargeInvoiceEntity invoiceUpdate = new WalletRechargeInvoiceEntity();
                 invoiceUpdate.setId(detail.getInvoiceId());
                 invoiceUpdate.setStatus(WalletRechargeInvoiceStatusEnum.CHECKED.getCode());
-                walletRechargeInvoiceService.updateById(invoiceUpdate);
+                if (!walletRechargeInvoiceService.updateById(invoiceUpdate)) throw new IllegalStateException("发票状态更新失败");
             }
-
             if (isConsent) {
-                List<WalletRechargeImportEntity> importList = rechargeImportService.list(new LambdaQueryWrapper<WalletRechargeImportEntity>().eq(WalletRechargeImportEntity::getApplyId, detail.getId()));
-                if (ObjectUtil.isEmpty(importList)) {
-                    return;
+                List<WalletRechargeImportEntity> importList = rechargeImportService.list(new LambdaQueryWrapper<WalletRechargeImportEntity>()
+                        .eq(WalletRechargeImportEntity::getApplyId, detail.getId()));
+                if (ObjectUtil.isEmpty(importList)) throw new IllegalStateException("充值申请没有有效明细，已停止入账");
+                updateBeforePrice(detail);
+                WalletBizEnum walletBiz = WalletRechargePosting.transactionType(detail.getBizType());
+                record Posting(Long walletId, WalletRechargeImportEntity item, int price) { }
+                List<Posting> postings = new ArrayList<>();
+                for (WalletRechargeImportEntity item : importList) {
+                    WalletManageEntity wallet = walletManageService.getWallet(item.getUserId(), detail.getWalletTypeId());
+                    if (wallet == null) throw new IllegalStateException("充值明细的钱包不存在，已停止整批入账，明细号：" + item.getId());
+                    postings.add(new Posting(wallet.getId(), item, WalletRechargePosting.signedPrice(detail.getBizType(), item.getPrice())));
                 }
-                getSelf().updateBeforePrice(detail);
-                if (WalletBizRechargeEnum.reductionCodes().contains(detail.getBizType())) {
-                    importList.forEach(v -> v.setPrice(-1 * Math.abs(v.getPrice())));
+                // 所有批次按同样的钱包顺序加锁，任一明细失败则整批回滚。
+                postings.sort(Comparator.comparing(Posting::walletId).thenComparing(p -> p.item().getId()));
+                for (Posting posting : postings) {
+                    walletTransactionService.addWalletBalance(posting.walletId(), posting.price(), String.valueOf(posting.item().getId()), walletBiz);
                 }
-                for (WalletRechargeImportEntity v : importList) {
-                    // 查询每个人的钱包余额
-                    WalletManageEntity wallet = walletManageService.getWallet(v.getUserId(), detail.getWalletTypeId());
-                    if (ObjectUtil.isEmpty(wallet)) {
-                        continue;
-                    }
-                    // 判断是`退还各自食堂`还是`批量更新余额`
-                    WalletBizEnum walletBiz = ObjectUtil.equal(WalletBizRechargeEnum.RETURN_CANTEEN.getCode(), detail.getBizType()) ? WalletBizEnum.RETURN_CANTEEN : WalletBizEnum.BATCH;
-                    walletTransactionService.addWalletBalance(wallet.getId(), v.getPrice(), String.valueOf(v.getId()), walletBiz);
-                }
-                // 同步到正元钱包状态
                 List<Long> userIds = importList.stream().map(WalletRechargeImportEntity::getUserId).distinct().toList();
-                if (ObjectUtil.isNotEmpty(userIds)) {
-                    try {
-                        walletThirdZyService.syncAllUser(userIds);
-                    } catch (Exception e) {
-                        // 不要抛出异常
-                        log.error("|{}|审批回调同步到正元钱包状态失败,dto:|{}|error:|{}|", TAG, JSON.toJSONString(dto), e.getMessage());
-                    }
-                }
-                log.info("|{}|审批回调成功,dto:|{}", TAG, JSON.toJSONString(dto));
+                afterCommitSyncService.register(detail.getId(), callbackTenant, userIds);
+                log.info("充值回调入账完成，等待事务提交 applyId={}, processId={}, taskId={}, count={}",
+                        detail.getId(), processId, dto.getTaskId(), postings.size());
             }
-        } catch (Exception e) {
-            e.printStackTrace();
-            log.error("|{}|审批回调意外的错误,dto:|{}|error:|{}|", TAG, JSON.toJSONString(dto), e.getMessage());
-            throw new RuntimeException(e);
+        } catch (Exception failure) {
+            log.error("充值回调失败，回滚本次处理 tenantId={}, processId={}, taskId={}", callbackTenant, processId, dto.getTaskId(), failure);
+            throw failure;
+        } finally {
+            TenantContextHolder.setTenantId(previousTenant);
+            TenantContextHolder.setTenantSkip(previousSkip);
+        }
+    }
+
+    private void requireUnsubmitted(WalletRechargeApplyEntity detail) {
+        if (ObjectUtil.isNotEmpty(detail.getProcessId()) || ObjectUtil.isNotEmpty(detail.getProcessStatus())
+                || detail.getApplyTime() != null || detail.getProcessTime() != null) {
+            throw new IllegalStateException("申请已提交或已处理，不能重复发起或修改，请新建申请");
+        }
+        requireNoPostedTransactions(detail);
+    }
+
+    private void requireNoPostedTransactions(WalletRechargeApplyEntity detail) {
+        Long tenantId = TenantContextHolder.getTenantId();
+        if (tenantId == null || !tenantId.equals(detail.getTenantId()) || TenantContextHolder.getTenantSkip()) {
+            throw new IllegalStateException("充值申请租户上下文错误");
+        }
+        if (!baseMapper.lockPostedTransactions(detail.getId(), tenantId).isEmpty()) {
+            throw new IllegalStateException("申请已有资金流水，不能重置或删除，请人工核查");
         }
     }
 
@@ -933,7 +974,4 @@ public class WalletRechargeApplyServiceImpl extends ServiceImpl<WalletRechargeAp
         }
     }
 
-    private WalletRechargeApplyServiceImpl getSelf() {
-        return SpringUtil.getBean(getClass());
-    }
-}
+}

pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/recharge/WalletRechargeImportServiceImpl.java

@@ -13,29 +13,32 @@ import com.bsd.pjcloud.admin.api.vo.UserVO;
 import com.bsd.pjcloud.common.core.util.CollectionUtils;
 import com.bsd.pjcloud.common.core.util.R;
 import com.bsd.pjcloud.common.data.datascope.DataScope;
-import com.bsd.pjcloud.wallet.api.constant.WalletBizEnum;
+import com.bsd.pjcloud.common.data.tenant.TenantContextHolder;
+import com.bsd.pjcloud.flow.task.constant.TaskCheckStatusEnum;
 import com.bsd.pjcloud.wallet.api.constant.WalletTypeEnum;
 import com.bsd.pjcloud.wallet.api.dto.WalletTypeDto;
 import com.bsd.pjcloud.wallet.api.dto.recharge.WalletRechargeImportDto;
 import com.bsd.pjcloud.wallet.api.entity.WalletManageEntity;
-import com.bsd.pjcloud.wallet.api.entity.WalletTransactionEntity;
 import com.bsd.pjcloud.wallet.api.entity.WalletTypeEntity;
 import com.bsd.pjcloud.wallet.api.entity.recharge.WalletRechargeApplyEntity;
 import com.bsd.pjcloud.wallet.api.entity.recharge.WalletRechargeImportEntity;
+import com.bsd.pjcloud.wallet.api.entity.recharge.WalletRechargeInvoiceEntity;
 import com.bsd.pjcloud.wallet.api.utils.WalletDataUtil;
 import com.bsd.pjcloud.wallet.api.vo.recharge.WalletRechargeImportVo;
 import com.bsd.pjcloud.wallet.mapper.WalletTypeMapper;
 import com.bsd.pjcloud.wallet.mapper.recharge.WalletRechargeImportMapper;
+import com.bsd.pjcloud.wallet.mapper.recharge.WalletRechargeApplyMapper;
+import com.bsd.pjcloud.wallet.mapper.recharge.WalletRechargeInvoiceMapper;
 import com.bsd.pjcloud.wallet.service.*;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.validation.annotation.Validated;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Objects;
 import java.util.stream.Collectors;
 
 /**
@@ -50,12 +53,12 @@ import java.util.stream.Collectors;
 @Service
 public class WalletRechargeImportServiceImpl extends ServiceImpl<WalletRechargeImportMapper, WalletRechargeImportEntity> implements WalletRechargeImportService {
 
-    @Lazy
-    private final WalletRechargeApplyService walletRechargeApplyService;
     private final WalletTransactionService walletTransactionService;
     private final WalletManageService walletManageService;
 
     private final WalletTypeMapper walletTypeMapper;
+    private final WalletRechargeApplyMapper walletRechargeApplyMapper;
+    private final WalletRechargeInvoiceMapper walletRechargeInvoiceMapper;
 
     private final RemoteUserService userService;
     private final RemoteDeptService deptService;
@@ -111,23 +114,112 @@ public class WalletRechargeImportServiceImpl extends ServiceImpl<WalletRechargeI
     public Boolean recover(Long id) {
         WalletRechargeImportEntity rechargeImport = baseMapper.selectById(id);
         if (ObjectUtil.isEmpty(rechargeImport)) throw new RuntimeException("充值记录不存在");
-        WalletRechargeApplyEntity rechargeApply = walletRechargeApplyService.getById(rechargeImport.getApplyId());
+        // 与回调、提交、删除按相同顺序先锁申请，再处理钱包。
+        WalletRechargeApplyEntity rechargeApply = walletRechargeApplyMapper.lockById(rechargeImport.getApplyId());
         if (ObjectUtil.isEmpty(rechargeApply)) throw new RuntimeException("充值申请不存在");
+        if (!TaskCheckStatusEnum.RESULT.getCode().equals(rechargeApply.getProcessStatus())) {
+            throw new IllegalStateException("只有审批通过的申请才能补记资金流水");
+        }
+        if (TenantContextHolder.getTenantId() == null || TenantContextHolder.getTenantSkip()
+                || !TenantContextHolder.getTenantId().equals(rechargeApply.getTenantId())) {
+            throw new IllegalStateException("充值申请租户上下文错误");
+        }
+        rechargeImport = baseMapper.selectOne(new LambdaQueryWrapper<WalletRechargeImportEntity>()
+                .eq(WalletRechargeImportEntity::getId, id)
+                .eq(WalletRechargeImportEntity::getApplyId, rechargeApply.getId()).last("FOR UPDATE"));
+        if (rechargeImport == null) throw new IllegalStateException("充值明细已变更，请刷新后重试");
         WalletTypeEntity walletType = walletTypeMapper.selectById(rechargeApply.getWalletTypeId());
-        WalletTypeEnum walletTypeEnum = WalletTypeEnum.valueOfByCode(walletType.getUniqueCode());
-        if (ObjectUtil.isEmpty(walletType) || ObjectUtil.isEmpty(walletTypeEnum)) {
+        WalletTypeEnum walletTypeEnum = walletType == null ? null : WalletTypeEnum.valueOfByCode(walletType.getUniqueCode());
+        if (ObjectUtil.isEmpty(walletTypeEnum)) {
             throw new RuntimeException("钱包类型错误");
         }
-        WalletTransactionEntity walletTransaction = walletTransactionService.getOne(new LambdaQueryWrapper<WalletTransactionEntity>()
-                .eq(WalletTransactionEntity::getBizId, String.valueOf(rechargeImport.getId()))
-        );
-        if (ObjectUtil.isNotEmpty(walletTransaction)) throw new RuntimeException("钱包已经充值过");
         WalletManageEntity wallet = walletManageService.getOrCreateWallet(rechargeImport.getUserId(), walletTypeEnum);
         if (ObjectUtil.isEmpty(wallet)) throw new RuntimeException("钱包错误");
-        walletTransactionService.addWalletBalance(wallet.getId(), rechargeImport.getPrice(), String.valueOf(rechargeImport.getId()), WalletBizEnum.BATCH);
+        walletTransactionService.addWalletBalance(wallet.getId(), WalletRechargePosting.signedPrice(rechargeApply.getBizType(), rechargeImport.getPrice()),
+                String.valueOf(rechargeImport.getId()), WalletRechargePosting.transactionType(rechargeApply.getBizType()));
         return true;
     }
 
+    @Override
+    @Transactional(rollbackFor = Exception.class)
+    public boolean saveDraft(WalletRechargeImportEntity entity) {
+        lockDraftParent(entity);
+        WalletRechargeImportEntity insert = WalletRechargeImportEntity.builder().applyId(entity.getApplyId())
+                .invoiceId(entity.getInvoiceId()).userId(entity.getUserId()).userDeptId(entity.getUserDeptId())
+                .price(entity.getPrice()).tenantId(TenantContextHolder.getTenantId()).build();
+        if (insert.getUserId() == null || insert.getPrice() == null) throw new IllegalArgumentException("用户和金额不能为空");
+        if (baseMapper.insert(insert) != 1) throw new IllegalStateException("新增充值明细失败");
+        return true;
+    }
+
+    @Override
+    @Transactional(rollbackFor = Exception.class)
+    public boolean updateDraft(WalletRechargeImportEntity entity) {
+        WalletRechargeImportEntity current = lockDraftItem(entity.getId());
+        if ((entity.getApplyId() != null && !Objects.equals(entity.getApplyId(), current.getApplyId()))
+                || (entity.getInvoiceId() != null && !Objects.equals(entity.getInvoiceId(), current.getInvoiceId()))) {
+            throw new IllegalArgumentException("不能变更充值明细所属申请或发票");
+        }
+        WalletRechargeImportEntity update = WalletRechargeImportEntity.builder().id(current.getId())
+                .userId(entity.getUserId()).userDeptId(entity.getUserDeptId()).price(entity.getPrice()).build();
+        if (baseMapper.updateById(update) != 1) throw new IllegalStateException("修改充值明细失败");
+        return true;
+    }
+
+    @Override
+    @Transactional(rollbackFor = Exception.class)
+    public boolean removeDrafts(List<Long> ids) {
+        if (ids == null || ids.isEmpty()) return true;
+        // 先收集并按申请/发票排序，避免两个批量维护请求以相反顺序持锁。
+        List<WalletRechargeImportEntity> items = baseMapper.selectList(new LambdaQueryWrapper<WalletRechargeImportEntity>()
+                .in(WalletRechargeImportEntity::getId, ids).orderByAsc(WalletRechargeImportEntity::getApplyId)
+                .orderByAsc(WalletRechargeImportEntity::getInvoiceId).orderByAsc(WalletRechargeImportEntity::getId));
+        for (WalletRechargeImportEntity item : items) {
+            lockDraftItem(item.getId());
+            if (baseMapper.deleteById(item.getId()) != 1) throw new IllegalStateException("删除充值明细失败");
+        }
+        return true;
+    }
+
+    private WalletRechargeImportEntity lockDraftItem(Long id) {
+        WalletRechargeImportEntity before = baseMapper.selectById(id);
+        if (before == null) throw new IllegalStateException("充值明细不存在");
+        lockDraftParent(before);
+        WalletRechargeImportEntity current = baseMapper.selectOne(new LambdaQueryWrapper<WalletRechargeImportEntity>()
+                .eq(WalletRechargeImportEntity::getId, id).last("FOR UPDATE"));
+        if (current == null || !Objects.equals(before.getApplyId(), current.getApplyId())
+                || !Objects.equals(before.getInvoiceId(), current.getInvoiceId())) {
+            throw new IllegalStateException("充值明细已变更，请刷新后重试");
+        }
+        return current;
+    }
+
+    private void lockDraftParent(WalletRechargeImportEntity item) {
+        Long tenantId = TenantContextHolder.getTenantId();
+        if (tenantId == null || TenantContextHolder.getTenantSkip()) throw new IllegalStateException("充值明细缺少租户上下文");
+        if (item.getApplyId() != null) {
+            WalletRechargeApplyEntity apply = walletRechargeApplyMapper.lockById(item.getApplyId());
+            if (apply == null || !tenantId.equals(apply.getTenantId())) throw new IllegalStateException("充值申请不存在");
+            if (item.getInvoiceId() != null && !Objects.equals(item.getInvoiceId(), apply.getInvoiceId())) {
+                throw new IllegalArgumentException("充值明细的发票与申请不一致");
+            }
+            if (ObjectUtil.isNotEmpty(apply.getProcessId()) || ObjectUtil.isNotEmpty(apply.getProcessStatus())
+                    || apply.getApplyTime() != null || apply.getProcessTime() != null
+                    || !walletRechargeApplyMapper.lockPostedTransactions(apply.getId(), tenantId).isEmpty()) {
+                throw new IllegalStateException("已提交审批或已入账的明细不能修改");
+            }
+        } else if (item.getInvoiceId() != null) {
+            WalletRechargeInvoiceEntity invoice = walletRechargeInvoiceMapper.selectOne(new LambdaQueryWrapper<WalletRechargeInvoiceEntity>()
+                    .eq(WalletRechargeInvoiceEntity::getId, item.getInvoiceId()).last("FOR UPDATE"));
+            if (invoice == null || !tenantId.equals(invoice.getTenantId()) || invoice.getApplyId() != null
+                    || ObjectUtil.isNotEmpty(invoice.getProcessId()) || ObjectUtil.isNotEmpty(invoice.getProcessStatus())) {
+                throw new IllegalStateException("发票已提交审批或绑定申请，不能修改明细");
+            }
+        } else {
+            throw new IllegalArgumentException("充值明细必须属于有效申请或发票");
+        }
+    }
+
     /** 导出分页在原查询基础上校验远程补全结果，失败时禁止生成缺失字段的文件。 */
     public Page<WalletRechargeImportVo> scopeExportPageList(Page page, WalletRechargeImportDto dto, DataScope dataScope) {
         Page<WalletRechargeImportVo> result = baseMapper.scopePageList(page, dto, dataScope);
@@ -180,4 +272,4 @@ public class WalletRechargeImportServiceImpl extends ServiceImpl<WalletRechargeI
             }
         }
     }
-}
+}

pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/recharge/WalletRechargePosting.java

@@ -0,0 +1,25 @@
+package com.bsd.pjcloud.wallet.service.impl.recharge;
+
+import com.bsd.pjcloud.wallet.api.constant.WalletBizEnum;
+import com.bsd.pjcloud.wallet.api.constant.WalletBizRechargeEnum;
+
+/** 审批入账与补记使用同一金额方向和流水类型。 */
+final class WalletRechargePosting {
+    private WalletRechargePosting() { }
+
+    static int signedPrice(String applyBizType, Integer price) {
+        if (price == null || WalletBizRechargeEnum.valueOfByType(applyBizType) == null) {
+            throw new IllegalArgumentException("充值业务类型或金额错误");
+        }
+        return WalletBizRechargeEnum.reductionCodes().contains(applyBizType)
+                ? Math.negateExact(Math.abs(price)) : price;
+    }
+
+    static WalletBizEnum transactionType(String applyBizType) {
+        if (WalletBizRechargeEnum.valueOfByType(applyBizType) == null) {
+            throw new IllegalArgumentException("充值业务类型错误");
+        }
+        return WalletBizRechargeEnum.RETURN_CANTEEN.getCode().equals(applyBizType)
+                ? WalletBizEnum.RETURN_CANTEEN : WalletBizEnum.BATCH;
+    }
+}

pjcloud-wallet/pjcloud-wallet-biz/src/test/java/com/bsd/pjcloud/wallet/recharge/WalletRechargeIdempotencyMySqlTest.java

@@ -0,0 +1,624 @@
+package com.bsd.pjcloud.wallet.recharge;
+
+import com.baomidou.mybatisplus.annotation.TableField;
+import com.baomidou.mybatisplus.core.MybatisConfiguration;
+import com.baomidou.mybatisplus.core.MybatisSqlSessionFactoryBuilder;
+import com.baomidou.mybatisplus.core.toolkit.GlobalConfigUtils;
+import com.baomidou.mybatisplus.extension.plugins.MybatisPlusInterceptor;
+import com.baomidou.mybatisplus.extension.plugins.inner.TenantLineInnerInterceptor;
+import com.bsd.pjcloud.admin.api.dto.UserInfo;
+import com.bsd.pjcloud.admin.api.entity.SysDept;
+import com.bsd.pjcloud.admin.api.entity.SysUser;
+import com.bsd.pjcloud.admin.api.feign.RemoteDeptService;
+import com.bsd.pjcloud.admin.api.feign.RemoteUserService;
+import com.bsd.pjcloud.common.core.util.R;
+import com.bsd.pjcloud.common.data.tenant.*;
+import com.bsd.pjcloud.common.data.mybatis.MybatisPlusMetaObjectHandler;
+import com.bsd.pjcloud.flow.task.constant.TaskCheckStatusEnum;
+import com.bsd.pjcloud.flow.task.dto.ProcessNodeRecordAssignUserParamDto;
+import com.bsd.pjcloud.wallet.api.constant.*;
+import com.bsd.pjcloud.wallet.api.dto.recharge.WalletRechargeApplyDto;
+import com.bsd.pjcloud.wallet.api.dto.recharge.WalletRechargeSaveTopupDto;
+import com.bsd.pjcloud.wallet.api.entity.*;
+import com.bsd.pjcloud.wallet.api.entity.recharge.*;
+import com.bsd.pjcloud.wallet.mapper.*;
+import com.bsd.pjcloud.wallet.mapper.recharge.*;
+import com.bsd.pjcloud.wallet.service.*;
+import com.bsd.pjcloud.wallet.service.impl.WalletTransactionServiceImpl;
+import com.bsd.pjcloud.wallet.service.impl.recharge.*;
+import com.bsd.pjcloud.wallet.utils.flow.FlowUtil;
+import org.apache.ibatis.datasource.pooled.PooledDataSource;
+import org.apache.ibatis.mapping.Environment;
+import org.junit.jupiter.api.*;
+import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.mybatis.spring.SqlSessionTemplate;
+import org.mybatis.spring.transaction.SpringManagedTransactionFactory;
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource;
+import org.springframework.transaction.interceptor.TransactionInterceptor;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import java.lang.reflect.Modifier;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.*;
+import java.time.*;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.function.IntFunction;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+import static org.mockito.AdditionalAnswers.delegatesTo;
+
+/**
+ * Opt-in real InnoDB transactions, tenant interceptor and production MyBatis mappers.
+ * Only creates/drops codex_recharge_* schemas. Does not copy or mutate application data.
+ */
+@EnabledIfEnvironmentVariable(named = "WALLET_RECHARGE_TEST_JDBC_URL", matches = ".+")
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class WalletRechargeIdempotencyMySqlTest {
+    private Connection admin;
+    private String schema;
+    private PooledDataSource source;
+    private JdbcTemplate db;
+    private DataSourceTransactionManager manager;
+    private WalletManageMapper wallets;
+    private WalletTransactionServiceImpl money, moneyTarget;
+    private WalletRechargeApplyServiceImpl applications;
+    private WalletRechargeImportServiceImpl imports;
+    private WalletRechargeApplyMapper applyMapper;
+    private WalletThirdZyService sync;
+    private FlowUtil flow;
+
+    @BeforeAll
+    void isolatedSchema() throws Exception {
+        String url = System.getenv("WALLET_RECHARGE_TEST_JDBC_URL");
+        String user = System.getenv("MYSQL_USER"), password = System.getenv("MYSQL_PWD");
+        admin = DriverManager.getConnection(url, user, password);
+        schema = "codex_recharge_" + System.currentTimeMillis();
+        try (Statement sql = admin.createStatement()) {
+            sql.execute("CREATE DATABASE " + schema + " CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci");
+        }
+        int slash = url.indexOf('/', "jdbc:mysql://".length());
+        int query = url.indexOf('?', slash);
+        String testUrl = url.substring(0, slash) + "/" + schema + (query < 0 ? "" : url.substring(query));
+        source = new PooledDataSource("com.mysql.cj.jdbc.Driver", testUrl, user, password);
+        source.setPoolMaximumActiveConnections(16);
+        db = new JdbcTemplate(source);
+        createEntityTable("wallet_manage", WalletManageEntity.class);
+        createEntityTable("wallet_transaction", WalletTransactionEntity.class);
+        createEntityTable("wallet_recharge_apply", WalletRechargeApplyEntity.class);
+        createEntityTable("wallet_recharge_import", WalletRechargeImportEntity.class);
+        createEntityTable("wallet_recharge_invoice", WalletRechargeInvoiceEntity.class);
+        db.execute("CREATE INDEX idx_history ON wallet_transaction(tenant_id,biz_id,biz_type)");
+        db.execute("CREATE INDEX idx_process ON wallet_recharge_apply(tenant_id,process_id)");
+        db.execute("CREATE INDEX idx_apply ON wallet_recharge_import(tenant_id,apply_id)");
+        String migration = Files.readString(Path.of("../../db/20261102_wallet_recharge_idempotency.sql"));
+        db.execute(migration.replaceAll("(?m)^--.*$", "").strip().replaceAll(";$", ""));
+    }
+
+    /** Match entity column types without depending on any existing local database or personal data. */
+    private void createEntityTable(String table, Class<?> entity) {
+        List<String> columns = new ArrayList<>();
+        for (var field : entity.getDeclaredFields()) {
+            if (Modifier.isStatic(field.getModifiers())) continue;
+            TableField annotation = field.getAnnotation(TableField.class);
+            if (annotation != null && !annotation.exist()) continue;
+            String name = field.getName().replaceAll("([a-z])([A-Z])", "$1_$2").toLowerCase(Locale.ROOT);
+            Class<?> type = field.getType();
+            String sqlType = type == Long.class ? "BIGINT" : type == Integer.class ? "INT"
+                    : type == LocalDate.class ? "DATE" : type == LocalDateTime.class ? "DATETIME" : "VARCHAR(512)";
+            if (Set.of("biz_id", "biz_type", "process_id").contains(name)) sqlType = "VARCHAR(64)";
+            columns.add(name + " " + sqlType + (name.equals("id") ? " PRIMARY KEY" : name.equals("del_flag") ? " DEFAULT '0'" : ""));
+        }
+        db.execute("CREATE TABLE " + table + "(" + String.join(",", columns) + ") ENGINE=InnoDB");
+    }
+
+    @AfterAll
+    void dropOnlyTestSchema() throws Exception {
+        TenantContextHolder.clear();
+        if (source != null) source.forceCloseAll();
+        if (admin != null) {
+            try (Connection connection = admin; Statement sql = connection.createStatement()) {
+                if (schema != null && schema.matches("codex_recharge_[0-9]+")) sql.execute("DROP DATABASE " + schema);
+            }
+        }
+    }
+
+    @BeforeEach
+    void setup() {
+        TenantContextHolder.setTenantId(1L);
+        TenantContextHolder.setTenantSkip(false);
+        for (String table : List.of("wallet_transaction_idempotency", "wallet_transaction", "wallet_recharge_import", "wallet_recharge_apply", "wallet_recharge_invoice", "wallet_manage")) db.update("DELETE FROM " + table);
+        MybatisConfiguration config = new MybatisConfiguration();
+        config.setMapUnderscoreToCamelCase(true);
+        GlobalConfigUtils.getGlobalConfig(config).setMetaObjectHandler(new MybatisPlusMetaObjectHandler());
+        config.setEnvironment(new Environment("recharge", new SpringManagedTransactionFactory(), source));
+        var tenant = new PjcloudTenantConfigProperties();
+        tenant.setTables(List.of("wallet_manage", "wallet_transaction", "wallet_recharge_apply", "wallet_recharge_import", "wallet_recharge_invoice"));
+        var interceptors = new MybatisPlusInterceptor();
+        interceptors.addInnerInterceptor(new TenantLineInnerInterceptor(new PjcloudTenantHandler(tenant)));
+        config.addInterceptor(interceptors);
+        for (Class<?> mapper : List.of(WalletManageMapper.class, WalletTransactionMapper.class,
+                WalletTransactionIdempotencyMapper.class, WalletRechargeApplyMapper.class,
+                WalletRechargeImportMapper.class, WalletRechargeInvoiceMapper.class)) config.addMapper(mapper);
+        var sessions = new SqlSessionTemplate(new MybatisSqlSessionFactoryBuilder().build(config));
+        manager = new DataSourceTransactionManager(source);
+        wallets = mock(WalletManageMapper.class, delegatesTo(sessions.getMapper(WalletManageMapper.class)));
+        applyMapper = sessions.getMapper(WalletRechargeApplyMapper.class);
+        var importMapper = sessions.getMapper(WalletRechargeImportMapper.class);
+        var invoiceMapper = sessions.getMapper(WalletRechargeInvoiceMapper.class);
+        moneyTarget = mock(WalletTransactionServiceImpl.class, CALLS_REAL_METHODS);
+        field(moneyTarget, "baseMapper", sessions.getMapper(WalletTransactionMapper.class));
+        field(moneyTarget, "walletManageMapper", wallets);
+        field(moneyTarget, "idempotencyMapper", sessions.getMapper(WalletTransactionIdempotencyMapper.class));
+        money = proxy(moneyTarget);
+        // Exercise the real proxied six-argument entry, without loading Redis or the full application.
+        doAnswer(call -> money.addWalletBalance(call.getArgument(0), call.getArgument(1), call.getArgument(2), call.getArgument(3), null, null))
+                .when(moneyTarget).addWalletBalance(anyLong(), anyInt(), anyString(), any(WalletBizEnum.class));
+        var manage = mock(WalletManageService.class);
+        when(manage.getWallet(anyLong(), anyLong())).thenAnswer(call -> wallets.selectOne(new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper<WalletManageEntity>()
+                .eq(WalletManageEntity::getUserId, call.<Long>getArgument(0)).eq(WalletManageEntity::getWalletTypeId, call.<Long>getArgument(1))));
+        when(manage.getOrCreateWallet(anyLong(), any(WalletTypeEnum.class))).thenAnswer(call -> manage.getWallet(call.getArgument(0), 10L));
+        var type = new WalletTypeEntity();
+        type.setId(10L); type.setName("饭卡"); type.setUniqueCode(WalletTypeEnum.EAT.getCode());
+        var typeMapper = mock(WalletTypeMapper.class);
+        when(typeMapper.selectById(anyLong())).thenReturn(type);
+        var types = mock(WalletTypeService.class);
+        when(types.getById(anyLong())).thenReturn(type);
+        sync = mock(WalletThirdZyService.class);
+        flow = mock(FlowUtil.class);
+        when(flow.checkCallback(any())).thenReturn(true);
+        when(flow.startFlow(anyString(), any(), anyLong(), any())).thenReturn("started-process");
+        imports = proxy(mock(WalletRechargeImportServiceImpl.class, CALLS_REAL_METHODS));
+        field(imports, "baseMapper", importMapper);
+        field(imports, "walletRechargeApplyMapper", applyMapper);
+        field(imports, "walletRechargeInvoiceMapper", invoiceMapper);
+        field(imports, "walletTransactionService", money);
+        field(imports, "walletManageService", manage);
+        field(imports, "walletTypeMapper", typeMapper);
+        var invoices = mock(WalletRechargeInvoiceService.class);
+        when(invoices.updateById(any())).thenAnswer(call -> invoiceMapper.updateById(call.<WalletRechargeInvoiceEntity>getArgument(0)) == 1);
+        var users = mock(RemoteUserService.class);
+        var user = new SysUser(); user.setUserId(7L); user.setUsername("fixture");
+        var userInfo = new UserInfo(); userInfo.setSysUser(user);
+        when(users.getInnerByUserId(anyLong())).thenReturn(R.ok(userInfo));
+        var departments = mock(RemoteDeptService.class);
+        var dept = new SysDept(); dept.setDeptId(9L); dept.setName("fixture");
+        when(departments.getDeptByDeptId(anyLong())).thenReturn(R.ok(dept));
+        var applicationTarget = mock(WalletRechargeApplyServiceImpl.class, CALLS_REAL_METHODS);
+        field(applicationTarget, "baseMapper", applyMapper);
+        field(applicationTarget, "rechargeImportService", imports);
+        field(applicationTarget, "walletManageService", manage);
+        field(applicationTarget, "walletTransactionService", money);
+        field(applicationTarget, "walletTypeService", types);
+        field(applicationTarget, "walletRechargeInvoiceService", invoices);
+        field(applicationTarget, "walletRechargeInvoiceMapper", invoiceMapper);
+        field(applicationTarget, "flowUtil", flow);
+        field(applicationTarget, "userService", users);
+        field(applicationTarget, "deptService", departments);
+        field(applicationTarget, "afterCommitSyncService", new WalletRechargeAfterCommitSyncService(sync, manager));
+        doNothing().when(applicationTarget).updateBeforePrice(any()); // Reporting snapshot only, no funds writes.
+        applications = proxy(applicationTarget);
+        wallet(1, 7, 1);
+        wallet(2, 8, 1);
+        application("1", "1");
+        item(101, 7, 150000);
+        item(102, 8, 150000);
+    }
+
+    private <T> T proxy(T target) {
+        ProxyFactory factory = new ProxyFactory(target);
+        factory.setProxyTargetClass(true);
+        factory.addAdvice(new TransactionInterceptor(manager, new AnnotationTransactionAttributeSource()));
+        @SuppressWarnings("unchecked") T value = (T) factory.getProxy();
+        return value;
+    }
+
+    private void field(Object target, String name, Object value) { ReflectionTestUtils.setField(target, name, value); }
+    private void wallet(long id, long user, long tenant) {
+        db.update("INSERT INTO wallet_manage(id,user_id,wallet_type_id,dept_id,balance,total_expense,total_recharge,version,tenant_id) VALUES(?,?,10,9,1000000,0,0,0,?)", id, user, tenant);
+    }
+    private void application(String status, String bizType) {
+        db.update("DELETE FROM wallet_recharge_apply");
+        db.update("INSERT INTO wallet_recharge_apply(id,process_id,process_status,biz_type,wallet_type_id,user_id,dept_id,tenant_id,create_time,import_time) VALUES(50,?,?,?,10,7,9,1,NOW(),'2026-10-01')", status == null ? null : "process", status, bizType);
+    }
+    private void item(long id, long user, int price) {
+        db.update("INSERT INTO wallet_recharge_import(id,apply_id,user_id,user_dept_id,price,tenant_id) VALUES(?,50,?,9,?,1)", id, user, price);
+    }
+    private ProcessNodeRecordAssignUserParamDto callback(boolean consent, int task) {
+        var dto = new ProcessNodeRecordAssignUserParamDto();
+        dto.setFlowId("flow"); dto.setProcessInstanceId("process"); dto.setNodeId("approve"); dto.setTaskId("task-" + task);
+        dto.setData("{\"approve_approve_condition\":" + consent + "}");
+        return dto;
+    }
+    private int scalar(String sql) { return db.queryForObject(sql, Integer.class); }
+    private int balance(long id) { return scalar("SELECT balance FROM wallet_manage WHERE id=" + id); }
+    private WalletTransactionEntity post(int price, String bizId, WalletBizEnum type) { return money.addWalletBalance(1L, price, bizId, type, null, null); }
+
+    /** Start all requests behind a barrier, assert every future completes; no timing sleeps. */
+    private <T> List<T> concurrent(int count, IntFunction<T> action) throws Exception {
+        var executor = Executors.newFixedThreadPool(count);
+        var start = new CountDownLatch(1);
+        try {
+            List<Future<T>> futures = new ArrayList<>();
+            for (int i = 0; i < count; i++) {
+                final int index = i;
+                futures.add(executor.submit(() -> {
+                    TenantContextHolder.setTenantId(1L);
+                    TenantContextHolder.setTenantSkip(false);
+                    try { assertTrue(start.await(5, TimeUnit.SECONDS)); return action.apply(index); }
+                    finally { TenantContextHolder.clear(); }
+                }));
+            }
+            start.countDown();
+            List<T> results = new ArrayList<>();
+            for (var future : futures) results.add(future.get(30, TimeUnit.SECONDS));
+            return results;
+        } finally {
+            executor.shutdownNow();
+            assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS), "All financial transactions must finish before fixture cleanup");
+        }
+    }
+
+    @ParameterizedTest
+    @CsvSource({"BATCH,150000", "BATCH,-150000", "BATCH,0", "RETURN_CANTEEN,-150000"})
+    void concurrentAndSequentialReplaysOnlyPostOnce(WalletBizEnum type, int price) throws Exception {
+        var results = concurrent(10, i -> post(price, "101", type));
+        assertEquals(1, results.stream().map(WalletTransactionEntity::getId).distinct().count());
+        assertEquals(results.get(0).getId(), post(price, "101", type).getId());
+        assertEquals(1000000 + price, balance(1));
+        assertEquals(1, scalar("SELECT count(*) FROM wallet_transaction"));
+        assertEquals(1, scalar("SELECT count(*) FROM wallet_transaction_idempotency"));
+        assertEquals(1, scalar("SELECT version FROM wallet_manage WHERE id=1"));
+        assertEquals(type == WalletBizEnum.BATCH ? price : 0, scalar("SELECT total_recharge FROM wallet_manage WHERE id=1"));
+        assertEquals(type == WalletBizEnum.RETURN_CANTEEN ? -price : 0, scalar("SELECT total_expense FROM wallet_manage WHERE id=1"));
+    }
+
+    @Test
+    void distinctBusinessIdsSerializeTheBalanceAndLedger() throws Exception {
+        concurrent(10, i -> post(100, "biz-" + i, WalletBizEnum.BATCH));
+        assertEquals(1001000, balance(1));
+        assertEquals(10, scalar("SELECT count(*) FROM wallet_transaction"));
+        assertEquals(List.of(1000100,1000200,1000300,1000400,1000500,1000600,1000700,1000800,1000900,1001000),
+                db.queryForList("SELECT balance FROM wallet_transaction ORDER BY balance", Integer.class));
+    }
+
+    @Test
+    void callerWithAnOlderRepeatableReadSnapshotStillSeesCommittedPosting() {
+        Long previousId = new TransactionTemplate(manager).execute(status -> {
+            assertEquals(0, scalar("SELECT count(*) FROM wallet_transaction"));
+            try {
+                var committed = concurrent(1, i -> post(150000, "101", WalletBizEnum.BATCH)).get(0);
+                WalletTransactionEntity replay = post(150000, "101", WalletBizEnum.BATCH);
+                assertEquals(committed.getId(), replay.getId());
+                return replay.getId();
+            } catch (Exception failure) { throw new RuntimeException(failure); }
+        });
+        assertNotNull(previousId);
+        assertEquals(1150000, balance(1));
+        assertEquals(1, scalar("SELECT count(*) FROM wallet_transaction"));
+    }
+
+    @Test
+    void changedAmountAndIntegerOverflowFailWithoutChangingFunds() {
+        post(100, "biz", WalletBizEnum.BATCH);
+        assertThrows(IllegalStateException.class, () -> post(101, "biz", WalletBizEnum.BATCH));
+        assertThrows(ArithmeticException.class, () -> post(Integer.MAX_VALUE, "overflow", WalletBizEnum.BATCH));
+        assertEquals(1000100, balance(1));
+        assertEquals(1, scalar("SELECT count(*) FROM wallet_transaction_idempotency"));
+        assertEquals(1, scalar("SELECT count(*) FROM wallet_transaction"));
+    }
+
+    @Test
+    void conflictingConcurrentAmountsOnlyAcceptOneRequest() throws Exception {
+        var results = concurrent(2, i -> {
+            int price = i == 0 ? 100 : 200;
+            try { post(price, "101", WalletBizEnum.BATCH); return price; }
+            catch (IllegalStateException failure) { assertTrue(failure.getMessage().contains("金额发生变化")); return 0; }
+        });
+        assertEquals(1, results.stream().filter(price -> price != 0).count());
+        assertEquals(1000000 + results.stream().mapToInt(Integer::intValue).sum(), balance(1));
+        assertEquals(1, scalar("SELECT count(*) FROM wallet_transaction"));
+    }
+
+    @Test
+    void missingGuardTableFailsBeforeAnyMoneyChanges() {
+        db.execute("RENAME TABLE wallet_transaction_idempotency TO guard_temporarily_unavailable");
+        try {
+            assertThrows(RuntimeException.class, () -> post(150000, "101", WalletBizEnum.BATCH));
+            assertEquals(1000000, balance(1));
+            assertEquals(0, scalar("SELECT count(*) FROM wallet_transaction"));
+        } finally {
+            db.execute("RENAME TABLE guard_temporarily_unavailable TO wallet_transaction_idempotency");
+        }
+    }
+
+    @Test
+    void ledgerFailureRollsBackGuardAndBalanceThenRetrySucceeds() {
+        doReturn(false).when(moneyTarget).save(any(WalletTransactionEntity.class));
+        assertThrows(IllegalStateException.class, () -> post(150000, "101", WalletBizEnum.BATCH));
+        assertEquals(1000000, balance(1));
+        assertEquals(0, scalar("SELECT count(*) FROM wallet_transaction_idempotency"));
+        doCallRealMethod().when(moneyTarget).save(any(WalletTransactionEntity.class));
+        post(150000, "101", WalletBizEnum.BATCH);
+        assertEquals(1150000, balance(1));
+    }
+
+    @Test
+    void zeroAffectedWalletUpdateCannotWriteALedger() {
+        doReturn(0).when(wallets).updateWhenRecharge(anyLong(), anyInt(), anyInt());
+        assertThrows(IllegalStateException.class, () -> post(150000, "101", WalletBizEnum.BATCH));
+        assertEquals(1000000, balance(1));
+        assertEquals(0, scalar("SELECT count(*) FROM wallet_transaction"));
+        assertEquals(0, scalar("SELECT count(*) FROM wallet_transaction_idempotency"));
+    }
+
+    private void legacy(long id, int price, String deleted) {
+        db.update("INSERT INTO wallet_transaction(id,wallet_id,user_id,biz_id,biz_type,price,balance,tenant_id,del_flag) VALUES(?,1,7,'101','2',?,1000000,1,?)", id, price, deleted);
+    }
+
+    @Test
+    void singleLegacyPostingIsAdoptedWithoutReplayingMoney() {
+        legacy(1001, 150000, "0");
+        assertEquals(1001L, post(150000, "101", WalletBizEnum.BATCH).getId());
+        assertEquals(1001L, post(150000, "101", WalletBizEnum.BATCH).getId());
+        assertEquals(1000000, balance(1));
+        assertEquals(1001L, db.queryForObject("SELECT transaction_id FROM wallet_transaction_idempotency", Long.class));
+    }
+
+    @ParameterizedTest
+    @CsvSource({"duplicate", "deleted", "changed"})
+    void ambiguousLegacyDataFailsClosed(String condition) {
+        legacy(1001, condition.equals("changed") ? 100 : 150000, condition.equals("deleted") ? "1" : "0");
+        if (condition.equals("duplicate")) legacy(1002, 150000, "0");
+        assertThrows(IllegalStateException.class, () -> post(150000, "101", WalletBizEnum.BATCH));
+        assertEquals(1000000, balance(1));
+        assertEquals(0, scalar("SELECT count(*) FROM wallet_transaction_idempotency"));
+    }
+
+    @Test
+    void deletingAPostedLedgerCannotEnableReplay() {
+        post(150000, "101", WalletBizEnum.BATCH);
+        db.update("DELETE FROM wallet_transaction");
+        assertThrows(IllegalStateException.class, () -> post(150000, "101", WalletBizEnum.BATCH));
+        assertEquals(1150000, balance(1));
+    }
+
+    @Test
+    void logicallyDeletingAPostedLedgerCannotEnableReplay() {
+        post(150000, "101", WalletBizEnum.BATCH);
+        db.update("UPDATE wallet_transaction SET del_flag='1'");
+        assertThrows(IllegalStateException.class, () -> post(150000, "101", WalletBizEnum.BATCH));
+        assertEquals(1150000, balance(1));
+    }
+
+    @Test
+    void manualAdjustmentsMayLegitimatelyReuseAUserBusinessId() {
+        post(100, "7", WalletBizEnum.UPDATE);
+        post(100, "7", WalletBizEnum.UPDATE);
+        assertEquals(1000200, balance(1));
+        assertEquals(2, scalar("SELECT count(*) FROM wallet_transaction"));
+        assertEquals(0, scalar("SELECT count(*) FROM wallet_transaction_idempotency"));
+    }
+
+    @Test
+    void tenantIsolationAndMissingContext() {
+        post(100, "101", WalletBizEnum.BATCH);
+        wallet(3, 7, 2);
+        TenantContextHolder.setTenantId(2L);
+        money.addWalletBalance(3L, 200, "101", WalletBizEnum.BATCH, null, null);
+        assertThrows(RuntimeException.class, () -> post(100, "101", WalletBizEnum.BATCH));
+        assertEquals(1000100, balance(1)); assertEquals(1000200, balance(3));
+        TenantContextHolder.clear();
+        assertThrows(IllegalStateException.class, () -> post(100, "missing", WalletBizEnum.BATCH));
+        TenantContextHolder.setTenantId(1L); TenantContextHolder.setTenantSkip(true);
+        assertThrows(IllegalStateException.class, () -> post(100, "skip", WalletBizEnum.BATCH));
+        assertEquals(2, scalar("SELECT count(*) FROM wallet_transaction"));
+    }
+
+    @Test
+    void concurrentDifferentTaskCallbacksOnlyPostOneBatchAndSyncAfterCommit() throws Exception {
+        doAnswer(call -> {
+            // Read through a separate physical connection, proving money was committed before sync.
+            try (Connection connection = source.getConnection(); Statement statement = connection.createStatement();
+                 ResultSet rows = statement.executeQuery("SELECT SUM(balance) FROM wallet_manage")) {
+                assertTrue(rows.next()); assertEquals(2300000, rows.getInt(1));
+            }
+            assertEquals(1L, TenantContextHolder.getTenantId());
+            return null;
+        }).when(sync).syncAllUser(anyList());
+        concurrent(6, i -> { applications.applyCallback(callback(true, i), 1L); return true; });
+        applications.applyCallback(callback(true, 99), 1L);
+        assertEquals(1150000, balance(1)); assertEquals(1150000, balance(2));
+        assertEquals(2, scalar("SELECT count(*) FROM wallet_transaction"));
+        assertEquals("2", db.queryForObject("SELECT process_status FROM wallet_recharge_apply", String.class));
+        verify(sync, times(1)).syncAllUser(anyList());
+    }
+
+    @Test
+    void secondItemFailureRollsBackWholeBatchStatusAndGuard() {
+        doAnswer(call -> {
+            if (call.<WalletTransactionEntity>getArgument(0).getWalletId().equals(2L)) throw new IllegalStateException("injected failure");
+            return call.callRealMethod();
+        }).when(moneyTarget).save(any(WalletTransactionEntity.class));
+        assertThrows(IllegalStateException.class, () -> applications.applyCallback(callback(true, 1), 1L));
+        assertEquals(1000000, balance(1)); assertEquals(1000000, balance(2));
+        assertEquals(0, scalar("SELECT count(*) FROM wallet_transaction"));
+        assertEquals(0, scalar("SELECT count(*) FROM wallet_transaction_idempotency"));
+        assertEquals("1", db.queryForObject("SELECT process_status FROM wallet_recharge_apply", String.class));
+        verify(sync, never()).syncAllUser(anyList());
+        doCallRealMethod().when(moneyTarget).save(any(WalletTransactionEntity.class));
+        applications.applyCallback(callback(true, 2), 1L);
+        assertEquals(2, scalar("SELECT count(*) FROM wallet_transaction"));
+    }
+
+    @Test
+    void missingWalletAndInvalidDecisionNeverClaimAnApplication() {
+        var invalid = callback(true, 1); invalid.setData("{}");
+        assertThrows(IllegalArgumentException.class, () -> applications.applyCallback(invalid, 1L));
+        db.update("DELETE FROM wallet_manage WHERE id=2");
+        assertThrows(IllegalStateException.class, () -> applications.applyCallback(callback(true, 2), 1L));
+        assertEquals("1", db.queryForObject("SELECT process_status FROM wallet_recharge_apply", String.class));
+        assertEquals(0, scalar("SELECT count(*) FROM wallet_transaction"));
+        verify(sync, never()).syncAllUser(anyList());
+    }
+
+    @Test
+    void rejectionIsFinalAndDoesNotMoveFunds() {
+        applications.applyCallback(callback(false, 1), 1L);
+        applications.applyCallback(callback(true, 2), 1L);
+        assertEquals("3", db.queryForObject("SELECT process_status FROM wallet_recharge_apply", String.class));
+        assertEquals(0, scalar("SELECT count(*) FROM wallet_transaction"));
+    }
+
+    @ParameterizedTest
+    @CsvSource({"1,150000,2", "2,-150000,2", "3,-150000,8"})
+    void callbackUsesCorrectSignedAmountForAllApprovalBusinessTypes(String type, int price, String ledgerType) {
+        application("1", type);
+        applications.applyCallback(callback(true, 1), 1L);
+        applications.applyCallback(callback(true, 2), 1L);
+        assertEquals(1000000 + price, balance(1));
+        assertEquals(1000000 + price, balance(2));
+        assertEquals(List.of(price, price), db.queryForList("SELECT price FROM wallet_transaction", Integer.class));
+        assertEquals(List.of(ledgerType, ledgerType), db.queryForList("SELECT biz_type FROM wallet_transaction", String.class));
+    }
+
+    @Test
+    void callbackInvoiceStatusAlsoRollsBackOnPostingFailure() {
+        db.update("INSERT INTO wallet_recharge_invoice(id,status,tenant_id) VALUES(70,'fixture',1)");
+        db.update("UPDATE wallet_recharge_apply SET invoice_id=70");
+        doReturn(false).when(moneyTarget).save(any(WalletTransactionEntity.class));
+        assertThrows(IllegalStateException.class, () -> applications.applyCallback(callback(true, 1), 1L));
+        assertEquals("fixture", db.queryForObject("SELECT status FROM wallet_recharge_invoice", String.class));
+        assertEquals("1", db.queryForObject("SELECT process_status FROM wallet_recharge_apply", String.class));
+        assertEquals(1000000, balance(1));
+    }
+
+    @Test
+    void recoveryRacingACallbackCannotPostTheSameItemTwice() throws Exception {
+        concurrent(2, i -> {
+            if (i == 0) applications.applyCallback(callback(true, 1), 1L);
+            else {
+                try { imports.recover(101L); }
+                catch (IllegalStateException failure) { assertTrue(failure.getMessage().contains("审批通过")); }
+            }
+            return true;
+        });
+        imports.recover(101L);
+        assertEquals(1150000, balance(1));
+        assertEquals(2, scalar("SELECT count(*) FROM wallet_transaction"));
+    }
+
+    @Test
+    void synchronizationFailureCannotRollBackPostedMoneyAndCallerTenantIsRestored() {
+        doThrow(new IllegalStateException("offline")).when(sync).syncAllUser(anyList());
+        TenantContextHolder.setTenantId(88L); TenantContextHolder.setTenantSkip(true);
+        applications.applyCallback(callback(true, 1), 1L);
+        assertEquals(88L, TenantContextHolder.getTenantId()); assertTrue(TenantContextHolder.getTenantSkip());
+        assertEquals(1150000, balance(1));
+        assertEquals("2", db.queryForObject("SELECT process_status FROM wallet_recharge_apply", String.class));
+    }
+
+    @Test
+    void concurrentSubmissionsStartOnlyOneProcessAndNeverOverwriteIt() throws Exception {
+        application(null, "1");
+        var results = concurrent(2, i -> {
+            try { applications.apply(50L); return true; }
+            catch (IllegalStateException failure) { assertTrue(failure.getMessage().contains("已提交")); return false; }
+        });
+        assertEquals(1, results.stream().filter(Boolean::booleanValue).count());
+        verify(flow, times(1)).startFlow(anyString(), any(), anyLong(), any());
+        assertEquals("started-process", db.queryForObject("SELECT process_id FROM wallet_recharge_apply", String.class));
+    }
+
+    @Test
+    void processedOrHistoricallyPostedApplicationsCannotBeResetOrEdited() {
+        application("2", "1");
+        assertThrows(IllegalStateException.class, () -> applications.apply(50L));
+        var dto = new WalletRechargeApplyDto(); dto.setId(50L); dto.setProcessStatus(""); dto.setProcessId("");
+        assertThrows(IllegalStateException.class, () -> applications.edit(dto));
+        application(null, "1"); legacy(1001, 150000, "1");
+        assertThrows(IllegalStateException.class, () -> applications.apply(50L));
+        verify(flow, never()).startFlow(anyString(), any(), anyLong(), any());
+    }
+
+    @Test
+    void editingDraftIgnoresClientWorkflowAndTenantFields() {
+        application(null, "1");
+        var dto = new WalletRechargeApplyDto(); dto.setId(50L); dto.setWalletTypeId(10L);
+        dto.setProcessStatus("2"); dto.setProcessId("forged"); dto.setTenantId(9L); dto.setUserId(999L);
+        dto.setImportTime(LocalDate.of(2026,10,1));
+        applications.edit(dto);
+        assertNull(db.queryForObject("SELECT process_id FROM wallet_recharge_apply", String.class));
+        assertNull(db.queryForObject("SELECT process_status FROM wallet_recharge_apply", String.class));
+        assertEquals(1, scalar("SELECT tenant_id FROM wallet_recharge_apply"));
+        assertEquals(7, scalar("SELECT user_id FROM wallet_recharge_apply"));
+    }
+
+    @ParameterizedTest
+    @CsvSource({"2,2", "3,8", "1,2"})
+    void recoveryUsesApprovalSignAndBusinessTypeAndIsIdempotent(String applyType, String transactionType) {
+        application("2", applyType);
+        imports.recover(101L); imports.recover(101L);
+        assertEquals(applyType.equals("1") ? 1150000 : 850000, balance(1));
+        assertEquals(transactionType, db.queryForObject("SELECT biz_type FROM wallet_transaction", String.class));
+        assertEquals(1, scalar("SELECT count(*) FROM wallet_transaction"));
+    }
+
+    @Test
+    void recoveryAndDirectItemChangesAreForbiddenOutsideDraftLifecycle() {
+        assertThrows(IllegalStateException.class, () -> imports.recover(101L));
+        var dto = WalletRechargeImportEntity.builder().id(101L).price(1).build();
+        assertThrows(IllegalStateException.class, () -> imports.updateDraft(dto));
+        assertThrows(IllegalStateException.class, () -> imports.removeDrafts(List.of(101L)));
+        application("3", "2"); assertThrows(IllegalStateException.class, () -> imports.recover(101L));
+        assertEquals(150000, scalar("SELECT price FROM wallet_recharge_import WHERE id=101"));
+        assertEquals(0, scalar("SELECT count(*) FROM wallet_transaction"));
+    }
+
+    @Test
+    void draftItemMaintenancePreservesOwnershipAndRejectsReparenting() {
+        application(null, "2");
+        imports.updateDraft(WalletRechargeImportEntity.builder().id(101L).price(120000).tenantId(99L).build());
+        assertEquals(120000, scalar("SELECT price FROM wallet_recharge_import WHERE id=101"));
+        assertEquals(1, scalar("SELECT tenant_id FROM wallet_recharge_import WHERE id=101"));
+        assertThrows(IllegalArgumentException.class, () -> imports.updateDraft(WalletRechargeImportEntity.builder().id(101L).applyId(999L).build()));
+        imports.saveDraft(WalletRechargeImportEntity.builder().applyId(50L).userId(7L).userDeptId(9L).price(50).tenantId(99L).build());
+        assertEquals(3, scalar("SELECT count(*) FROM wallet_recharge_import WHERE apply_id=50 AND tenant_id=1"));
+        imports.removeDrafts(List.of(101L));
+        assertEquals("1", db.queryForObject("SELECT del_flag FROM wallet_recharge_import WHERE id=101", String.class));
+        assertEquals(0, scalar("SELECT count(*) FROM wallet_transaction"));
+    }
+
+    @Test
+    void directInvoiceItemMaintenanceStopsWhenInvoiceIsSubmitted() {
+        db.update("INSERT INTO wallet_recharge_invoice(id,tenant_id) VALUES(70,1)");
+        var item = WalletRechargeImportEntity.builder().invoiceId(70L).userId(7L).price(100).build();
+        imports.saveDraft(item);
+        db.update("UPDATE wallet_recharge_invoice SET process_status='1',process_id='invoice-process' WHERE id=70");
+        assertThrows(IllegalStateException.class, () -> imports.saveDraft(item));
+        assertEquals(1, scalar("SELECT count(*) FROM wallet_recharge_import WHERE invoice_id=70"));
+    }
+
+    @Test
+    void invoiceCanOnlyBeBoundOnceUnderConcurrentRequests() throws Exception {
+        db.update("INSERT INTO wallet_recharge_invoice(id,process_status,apply_user_id,apply_dept_id,tenant_id,create_time) VALUES(70,'2',7,9,1,NOW())");
+        db.update("UPDATE wallet_recharge_import SET apply_id=NULL,invoice_id=70");
+        var dto = new WalletRechargeSaveTopupDto(); dto.setInvoiceId(70L); dto.setBizType("1");
+        var results = concurrent(2, i -> {
+            try { applications.saveTopup(dto); return true; }
+            catch (RuntimeException failure) { assertTrue(failure.getMessage().contains("已被绑定")); return false; }
+        });
+        assertEquals(1, results.stream().filter(Boolean::booleanValue).count());
+        assertEquals(2, scalar("SELECT count(*) FROM wallet_recharge_apply"));
+        assertEquals(2, scalar("SELECT count(*) FROM wallet_recharge_import WHERE apply_id=(SELECT apply_id FROM wallet_recharge_invoice WHERE id=70)"));
+    }
+}