diff --git a/db/20261102_wallet_recharge_idempotency.md b/db/20261102_wallet_recharge_idempotency.md new file mode 100644 index 000000000..54bdbc768 --- /dev/null +++ b/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。 diff --git a/db/20261102_wallet_recharge_idempotency.sql b/db/20261102_wallet_recharge_idempotency.sql new file mode 100644 index 000000000..1e5c6a563 --- /dev/null +++ b/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='批量钱包入账幂等键，与余额和流水同事务提交'; diff --git a/db/800pjcloudx_wallet.sql b/db/800pjcloudx_wallet.sql index 76c5c763b..b717ac59a 100644 --- a/db/800pjcloudx_wallet.sql +++ b/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 -- ---------------------------- diff --git a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/controller/recharge/WalletRechargeImportController.java b/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/controller/recharge/WalletRechargeImportController.java index 09bf37695..392568d24 100644 --- a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/controller/recharge/WalletRechargeImportController.java +++ b/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 exportAsync(@RequestBody WalletRechargeImportDto dto) { return R.ok(batchExportService.submitImport(dto), "导出任务已提交"); } -} \ No newline at end of file +} diff --git a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/mapper/WalletTransactionIdempotencyMapper.java b/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/mapper/WalletTransactionIdempotencyMapper.java new file mode 100644 index 000000000..68e757fd5 --- /dev/null +++ b/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 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; + } +} diff --git a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/mapper/recharge/WalletRechargeApplyMapper.java b/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/mapper/recharge/WalletRechargeApplyMapper.java index 555fe42d0..83ccc81ce 100644 --- a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/mapper/recharge/WalletRechargeApplyMapper.java +++ b/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 { + default WalletRechargeApplyEntity lockById(Long id) { + return selectOne(new LambdaQueryWrapper() + .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 lockPostedTransactions(@Param("applyId") Long applyId, @Param("tenantId") Long tenantId); + Page scopePageList(Page page, @Param("dto") WalletRechargeApplyDto dto, DataScope dataScope); @@ -60,4 +77,4 @@ public interface WalletRechargeApplyMapper extends PjcloudBaseMapper scopeList(WalletRechargeImportDto dto, DataScope dataScope); Boolean recover(Long id); -} \ No newline at end of file + + /** 后台人工维护入口，仅允许修改尚未提交审批的明细。 */ + boolean saveDraft(WalletRechargeImportEntity entity); + + boolean updateDraft(WalletRechargeImportEntity entity); + + boolean removeDrafts(List ids); +} diff --git a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/WalletTransactionServiceImpl.java b/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/WalletTransactionServiceImpl.java index a885ec019..7c4a05442 100644 --- a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/WalletTransactionServiceImpl.java +++ b/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() + .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 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 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) { diff --git a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/recharge/WalletRechargeAfterCommitSyncService.java b/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/recharge/WalletRechargeAfterCommitSyncService.java new file mode 100644 index 000000000..790587ca5 --- /dev/null +++ b/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 userIds) { + if (!TransactionSynchronizationManager.isActualTransactionActive() + || !TransactionSynchronizationManager.isSynchronizationActive()) { + throw new IllegalStateException("资金同步必须在入账事务中注册"); + } + List 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); + } + } + }); + } +} diff --git a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/recharge/WalletRechargeApplyServiceImpl.java b/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/recharge/WalletRechargeApplyServiceImpl.java index 48bde9e84..253092149 100644 --- a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/recharge/WalletRechargeApplyServiceImpl.java +++ b/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() + .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 list = rechargeImportService.list(new LambdaQueryWrapper().eq(WalletRechargeImportEntity::getInvoiceId, invoice.getId()).isNull(WalletRechargeImportEntity::getApplyId)); if (ObjectUtil.isNotEmpty(list)) { @@ -150,14 +152,14 @@ public class WalletRechargeApplyServiceImpl extends ServiceImpl 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().eq(WalletRechargeImportEntity::getApplyId, v)); } @@ -256,10 +267,11 @@ public class WalletRechargeApplyServiceImpl extends ServiceImpl().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() + .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.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 importList = rechargeImportService.list(new LambdaQueryWrapper().eq(WalletRechargeImportEntity::getApplyId, detail.getId())); - if (ObjectUtil.isEmpty(importList)) { - return; + List importList = rechargeImportService.list(new LambdaQueryWrapper() + .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 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 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 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() + .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() - .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 ids) { + if (ids == null || ids.isEmpty()) return true; + // 先收集并按申请/发票排序，避免两个批量维护请求以相反顺序持锁。 + List items = baseMapper.selectList(new LambdaQueryWrapper() + .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() + .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() + .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 scopeExportPageList(Page page, WalletRechargeImportDto dto, DataScope dataScope) { Page result = baseMapper.scopePageList(page, dto, dataScope); @@ -180,4 +272,4 @@ public class WalletRechargeImportServiceImpl extends ServiceImpl entity) { + List 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() + .eq(WalletManageEntity::getUserId, call.getArgument(0)).eq(WalletManageEntity::getWalletTypeId, call.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.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 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 List concurrent(int count, IntFunction action) throws Exception { + var executor = Executors.newFixedThreadPool(count); + var start = new CountDownLatch(1); + try { + List> 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 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.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)")); + } +}