diff --git a/db/20261030_sync_wallet_department_after_user_transaction.sql b/db/20261030_sync_wallet_department_after_user_transaction.sql new file mode 100644 index 00000000..08a335c7 --- /dev/null +++ b/db/20261030_sync_wallet_department_after_user_transaction.sql @@ -0,0 +1,105 @@ +-- 人员调动审批通过后的历史钱包部门补同步（2026-10-30，仅生成，尚未执行）。 +-- 范围：指定租户内，存在“审批通过且目标部门等于用户当前部门”的调动记录， +-- 用户、当前部门及钱包均未删除，钱包部门与当前用户部门不一致（包含 NULL）。 +-- 以 sys_user.dept_id 为准，不用历史调动单覆盖用户后来再次变更的部门。 +-- 仅更新 wallet_manage.dept_id 和 update_time，不处理余额、流水、饭卡商户或正元第三方同步。 +-- 重复执行会跳过已经一致的钱包；所有跨表关联均同时校验租户和用户。 +-- 默认微服务库名：UPMS 为 pjcloudx，钱包为 pjcloudx_wallet，要求位于同一 MySQL 实例。 +-- 单体部署或自定义库名请先统一替换下方库名前缀，并确认执行账号具备跨库读写权限。 +-- 先执行第 1 节并导出旧值作为修复前快照，核对范围后在同一连接执行第 2、3 节。 +-- 执行期间暂停人员部门变更及钱包部门同步；遇到 SQL 错误立即 ROLLBACK，不继续 COMMIT。 +-- 本文件不修改表结构；此 SQL 只补钱包部门，饭卡商户和第三方信息仍需按业务接口同步。 + +SET NAMES utf8mb4; + +-- 目标租户，默认限定本次问题所在的租户 1；切换租户时必须先重新核对预览结果。 +SET @repair_tenant_id = 1; +-- NULL 表示修复该租户内所有符合条件的人员；只修董馨悦时改为 2087822558359871489。 +SET @repair_user_id = NULL; + +-- 1. 修复前预览：每行对应一个待更新钱包，部门不一致判断使用 NULL 安全比较。 +SELECT w.tenant_id, + w.id AS wallet_id, + w.user_id, + u.name AS user_name, + w.dept_id AS old_wallet_dept_id, + u.dept_id AS new_wallet_dept_id, + d.name AS new_wallet_dept_name, + w.update_time AS old_update_time +FROM pjcloudx_wallet.wallet_manage w +JOIN pjcloudx.sys_user u + ON u.user_id = w.user_id AND u.tenant_id = w.tenant_id +JOIN pjcloudx.sys_dept d + ON d.dept_id = u.dept_id AND d.tenant_id = u.tenant_id +WHERE w.tenant_id = @repair_tenant_id + AND (@repair_user_id IS NULL OR w.user_id = @repair_user_id) + AND w.del_flag = '0' + AND u.del_flag = '0' + AND d.del_flag = '0' + AND u.dept_id IS NOT NULL + AND NOT (w.dept_id <=> u.dept_id) + AND EXISTS ( + SELECT 1 + FROM pjcloudx.sys_user_transaction t + WHERE t.tenant_id = u.tenant_id + AND t.user_id = u.user_id + AND t.del_flag = '0' + AND t.process_status = '2' + AND t.target_dept_id = u.dept_id + ) +ORDER BY w.user_id, w.id; + +-- 2. 补同步：更新时重新核对当前用户部门及审批记录，不依赖预览中可能过期的部门值。 +START TRANSACTION; + +UPDATE pjcloudx_wallet.wallet_manage w +JOIN pjcloudx.sys_user u + ON u.user_id = w.user_id AND u.tenant_id = w.tenant_id +JOIN pjcloudx.sys_dept d + ON d.dept_id = u.dept_id AND d.tenant_id = u.tenant_id +SET w.dept_id = u.dept_id, + w.update_time = NOW() +WHERE w.tenant_id = @repair_tenant_id + AND (@repair_user_id IS NULL OR w.user_id = @repair_user_id) + AND w.del_flag = '0' + AND u.del_flag = '0' + AND d.del_flag = '0' + AND u.dept_id IS NOT NULL + AND NOT (w.dept_id <=> u.dept_id) + AND EXISTS ( + SELECT 1 + FROM pjcloudx.sys_user_transaction t + WHERE t.tenant_id = u.tenant_id + AND t.user_id = u.user_id + AND t.del_flag = '0' + AND t.process_status = '2' + AND t.target_dept_id = u.dept_id + ); + +SELECT ROW_COUNT() AS updated_wallet_count; + +COMMIT; + +-- 3. 修复后复核：同一范围的剩余不一致钱包数量应为 0；也可重跑第 1 节查看明细。 +SELECT COUNT(*) AS remaining_mismatched_wallet_count +FROM pjcloudx_wallet.wallet_manage w +JOIN pjcloudx.sys_user u + ON u.user_id = w.user_id AND u.tenant_id = w.tenant_id +JOIN pjcloudx.sys_dept d + ON d.dept_id = u.dept_id AND d.tenant_id = u.tenant_id +WHERE w.tenant_id = @repair_tenant_id + AND (@repair_user_id IS NULL OR w.user_id = @repair_user_id) + AND w.del_flag = '0' + AND u.del_flag = '0' + AND d.del_flag = '0' + AND u.dept_id IS NOT NULL + AND NOT (w.dept_id <=> u.dept_id) + AND EXISTS ( + SELECT 1 + FROM pjcloudx.sys_user_transaction t + WHERE t.tenant_id = u.tenant_id + AND t.user_id = u.user_id + AND t.del_flag = '0' + AND t.process_status = '2' + AND t.target_dept_id = u.dept_id + ); diff --git a/pjcloud-upms/pjcloud-upms-biz/src/main/java/com/bsd/pjcloud/admin/handler/UserTransactionWalletSyncEvent.java b/pjcloud-upms/pjcloud-upms-biz/src/main/java/com/bsd/pjcloud/admin/handler/UserTransactionWalletSyncEvent.java new file mode 100644 index 00000000..410cf6b9 --- /dev/null +++ b/pjcloud-upms/pjcloud-upms-biz/src/main/java/com/bsd/pjcloud/admin/handler/UserTransactionWalletSyncEvent.java @@ -0,0 +1,58 @@ +package com.bsd.pjcloud.admin.handler; + +/** + * 人员调动审批通过后的钱包同步事件。 + * + *

在审批事务中发布，仅保留跨线程所需的稳定标识；不携带请求对象或可能过期的用户部门快照。

+ */ +public final class UserTransactionWalletSyncEvent { + + /** 已通过审批的人员调动单主键，用于关联同步失败日志。 */ + private final Long userTransactionId; + + /** 调动用户主键；每次同步均由钱包服务重新查询当前已提交的用户资料。 */ + private final Long userId; + + /** 发布事件时的租户主键，异步调用时通过 TenantBroker 恢复；为空时沿用下游默认租户。 */ + private final Long tenantId; + + /** + * 创建人员调动钱包同步事件。 + * + * @param userTransactionId 已通过审批的调动单主键 + * @param userId 需要同步钱包部门的用户主键 + * @param tenantId 审批事务的租户上下文 + */ + public UserTransactionWalletSyncEvent(Long userTransactionId, Long userId, Long tenantId) { + this.userTransactionId = userTransactionId; + this.userId = userId; + this.tenantId = tenantId; + } + + /** + * 获取调动单主键。 + * + * @return 用于日志关联的调动单主键 + */ + public Long getUserTransactionId() { + return userTransactionId; + } + + /** + * 获取需要同步钱包的用户主键。 + * + * @return 调动用户主键 + */ + public Long getUserId() { + return userId; + } + + /** + * 获取审批事务的租户上下文。 + * + * @return 租户主键，非 Web 调用场景下可能为空 + */ + public Long getTenantId() { + return tenantId; + } +} diff --git a/pjcloud-upms/pjcloud-upms-biz/src/main/java/com/bsd/pjcloud/admin/handler/UserTransactionWalletSyncEventListener.java b/pjcloud-upms/pjcloud-upms-biz/src/main/java/com/bsd/pjcloud/admin/handler/UserTransactionWalletSyncEventListener.java new file mode 100644 index 00000000..c724545c --- /dev/null +++ b/pjcloud-upms/pjcloud-upms-biz/src/main/java/com/bsd/pjcloud/admin/handler/UserTransactionWalletSyncEventListener.java @@ -0,0 +1,79 @@ +package com.bsd.pjcloud.admin.handler; + +import com.bsd.pjcloud.common.core.util.R; +import com.bsd.pjcloud.common.data.tenant.TenantBroker; +import com.bsd.pjcloud.wallet.api.feign.RemoteWalletService; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Async; +import org.springframework.stereotype.Component; +import org.springframework.transaction.event.TransactionPhase; +import org.springframework.transaction.event.TransactionalEventListener; + +/** + * 人员调动提交后的钱包同步监听器。 + * + *

复用 Spring 管理的异步线程池，按用户当前资料幂等同步；有限重试全部失败后记录错误，供人工补同步。 + * 重试只保存在当前进程中，服务重启不会恢复尚未完成的调用。

+ */ +@Slf4j +@Component +public class UserTransactionWalletSyncEventListener { + + /** 每个调动事件最多调用三次，包含首次调用和两次失败重试。 */ + private static final int MAX_SYNC_ATTEMPTS = 3; + + /** 重试等待基数，单位毫秒；第一次、第二次重试分别等待一秒、两秒。 */ + private static final long RETRY_DELAY_MILLIS = 1000L; + + /** 钱包内部接口，负责回查当前用户并同步钱包部门及关联饭卡商户。 */ + private final RemoteWalletService remoteWalletService; + + /** + * 创建人员调动钱包同步监听器。 + * + * @param remoteWalletService 钱包服务远程接口 + */ + public UserTransactionWalletSyncEventListener(RemoteWalletService remoteWalletService) { + this.remoteWalletService = remoteWalletService; + } + + /** + * 在调动事务提交成功后异步同步钱包，遇到异常、空响应或业务失败时有限重试。 + * + *

不启用无事务兜底，避免绕过审批事务提交边界。每次调用均恢复并释放租户上下文， + * 不绑定 HTTP 请求；钱包接口已有 NoToken 内部调用标记。中断时保留中断标志并停止重试。

+ * + * @param event 已通过审批的调动单、用户主键及租户上下文 + */ + @Async + @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) + public void syncWalletAfterCommit(UserTransactionWalletSyncEvent event) { + for (int attempt = 1; attempt <= MAX_SYNC_ATTEMPTS; attempt++) { + try { + R result = TenantBroker.applyAs(event.getTenantId(), tenantId -> remoteWalletService.userChange(event.getUserId(), Boolean.FALSE)); + if (result == null) { + throw new IllegalStateException("钱包服务未返回同步结果"); + } + if (!result.isOk() || !Boolean.TRUE.equals(result.getData())) { + throw new IllegalStateException("钱包同步业务失败，code:" + result.getCode() + "，data:" + result.getData() + "，msg:" + result.getMsg()); + } + log.info("人员调动提交后同步钱包成功，userTransactionId:{}，userId:{}，tenantId:{}，attempt:{}", event.getUserTransactionId(), event.getUserId(), event.getTenantId(), attempt); + return; + } catch (Exception exception) { + if (attempt == MAX_SYNC_ATTEMPTS) { + log.error("人员调动提交后同步钱包重试耗尽，需补同步，userTransactionId:{}，userId:{}，tenantId:{}，attempt:{}", event.getUserTransactionId(), event.getUserId(), event.getTenantId(), attempt, exception); + return; + } + long retryDelayMillis = RETRY_DELAY_MILLIS * attempt; + log.warn("人员调动提交后同步钱包失败，将重试，userTransactionId:{}，userId:{}，tenantId:{}，attempt:{}，retryDelayMillis:{}", event.getUserTransactionId(), event.getUserId(), event.getTenantId(), attempt, retryDelayMillis, exception); + try { + Thread.sleep(retryDelayMillis); + } catch (InterruptedException interruptedException) { + Thread.currentThread().interrupt(); + log.error("人员调动钱包同步重试被中断，需补同步，userTransactionId:{}，userId:{}，tenantId:{}", event.getUserTransactionId(), event.getUserId(), event.getTenantId(), interruptedException); + return; + } + } + } + } +} diff --git a/pjcloud-upms/pjcloud-upms-biz/src/main/java/com/bsd/pjcloud/admin/service/impl/SysUserTransactionServiceImpl.java b/pjcloud-upms/pjcloud-upms-biz/src/main/java/com/bsd/pjcloud/admin/service/impl/SysUserTransactionServiceImpl.java index 3fb9c2be..f3083ed4 100644 --- a/pjcloud-upms/pjcloud-upms-biz/src/main/java/com/bsd/pjcloud/admin/service/impl/SysUserTransactionServiceImpl.java +++ b/pjcloud-upms/pjcloud-upms-biz/src/main/java/com/bsd/pjcloud/admin/service/impl/SysUserTransactionServiceImpl.java @@ -19,6 +19,7 @@ import com.bsd.pjcloud.admin.api.entity.SysDept; import com.bsd.pjcloud.admin.api.entity.SysUser; import com.bsd.pjcloud.admin.api.entity.SysUserTransactionEntity; import com.bsd.pjcloud.admin.api.vo.SysUserTransactionVo; +import com.bsd.pjcloud.admin.handler.UserTransactionWalletSyncEvent; import com.bsd.pjcloud.admin.mapper.SysUserTransactionMapper; import com.bsd.pjcloud.admin.service.SysDeptService; import com.bsd.pjcloud.admin.service.SysUserService; @@ -36,22 +37,19 @@ import com.bsd.pjcloud.flow.task.constant.TaskCheckStatusEnum; import com.bsd.pjcloud.flow.task.dto.ProcessNodeRecordAssignUserParamDto; import com.bsd.pjcloud.flow.task.dto.TaskParamDto; import com.bsd.pjcloud.wallet.api.feign.RemoteWalletRuleTransactionService; -import com.bsd.pjcloud.wallet.api.feign.RemoteWalletService; import com.bsd.pjcloud.wallet.api.feign.RemoteWalletThirdZyService; import lombok.AllArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.cache.CacheManager; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.security.oauth2.server.authorization.OAuth2AuthorizationService; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; -import org.springframework.web.context.request.RequestAttributes; -import org.springframework.web.context.request.RequestContextHolder; import java.time.LocalDate; import java.time.LocalDateTime; import java.util.ArrayList; import java.util.List; -import java.util.concurrent.CompletableFuture; /** * 人员调动表 @@ -71,11 +69,13 @@ public class SysUserTransactionServiceImpl extends ServiceImpl scopePageJoin(Page page, SysUserTransactionDto dto, DataScope dataScope) { return baseMapper.scopePageJoin(page, dto, dataScope); @@ -189,7 +189,13 @@ public class SysUserTransactionServiceImpl extends ServiceImpl事件只携带稳定标识；用户部门和审批状态成功提交后才异步回查用户，避免钱包读取旧部门。 + * 钱包调用失败由监听器记录并重试，不再跨线程持有已结束的 HTTP 请求。

+ * + * @param dto 流程引擎回调，包含流程实例及审批结果 + * @param tenantId 流程所属租户；为空时沿用请求租户上下文 */ @Override @Transactional(rollbackFor = Exception.class) @@ -224,12 +230,8 @@ public class SysUserTransactionServiceImpl extends ServiceImpl { - RequestContextHolder.setRequestAttributes(requestAttributes); - remoteWalletService.userChange(detail.getUserId(), Boolean.FALSE); - }); + // 提交前只登记事件，事务回滚时不向钱包发送用户部门变更。 + applicationEventPublisher.publishEvent(new UserTransactionWalletSyncEvent(detail.getId(), detail.getUserId(), TenantContextHolder.getTenantId())); } } catch (Exception e) { e.printStackTrace(); @@ -295,4 +297,4 @@ public class SysUserTransactionServiceImpl extends ServiceImpl用户资料查询与食堂商户解析（均含远程调用）在事务外完成，本地更新经短事务 * {@link #userChangeInTx} 提交；正元第三方同步放在事务提交后执行， - * 避免远程调用和第三方逻辑拉长事务持锁时间，同步异常也不再回滚本地部门变更。

+ * 避免远程调用和第三方逻辑拉长事务持锁时间，同步异常也不再回滚本地部门变更。 + * 用户查询业务失败、返回空资料或未分配部门时抛出异常，避免向调动重试调用方报告虚假成功。

* * @param userId 变更用户 ID * @param isDel 是否删除事件；仅处理非删除事件 @@ -877,9 +878,14 @@ public class WalletManageServiceImpl extends ServiceImpl userResult = userService.getInnerByUserId(userId); + if (userResult == null || !userResult.isOk() || userResult.getData() == null || userResult.getData().getSysUser() == null) { + throw new IllegalStateException("查询待同步用户资料失败，userId:" + userId + "，原因:" + (userResult == null ? "用户服务未返回结果" : userResult.getMsg())); + } + SysUser sysUser = userResult.getData().getSysUser(); + if (!ObjectUtil.equals(userId, sysUser.getUserId()) || sysUser.getDeptId() == null) { + throw new IllegalStateException("待同步用户主键不匹配或未分配部门，userId:" + userId); + } // 查询用户所属于的食堂商户，解析含远程调用，先于事务执行 MchManageVo mch = mchManageService.getFirstMchByDeptId(sysUser.getDeptId(), List.of(MerchantTypeEnum.CANTEEN)); getSelf().userChangeInTx(sysUser, mch); @@ -890,6 +896,7 @@ public class WalletManageServiceImpl extends ServiceImpl历史钱包部门为 NULL 时也需要回填；已经一致的钱包不更新，以便失败后的重复通知保持幂等。

* * @param sysUser 已查询的用户资料，取部门和用户 ID * @param mch 按用户部门解析的食堂商户；为空时不调整已绑饭卡的商户归属 @@ -897,11 +904,14 @@ public class WalletManageServiceImpl extends ServiceImpl 不匹配 NULL，必须把历史空部门纳入同一用户的更新范围。 baseMapper.update(new LambdaUpdateWrapper() .set(WalletManageEntity::getDeptId, sysUser.getDeptId()) - .ne(WalletManageEntity::getDeptId, sysUser.getDeptId()) .eq(WalletManageEntity::getUserId, sysUser.getUserId()) + .and(wrapper -> wrapper.isNull(WalletManageEntity::getDeptId).or().ne(WalletManageEntity::getDeptId, sysUser.getDeptId())) ); // 修改钱包冻结开卡商户 if (ObjectUtil.isNotEmpty(mch)) {