feat:部门调岗问题

2026-10-30 16:08:23 +08:00
1 parent ae5af61d6e
commit d95fb63317
@@ -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
+  );
@@ -0,0 +1,58 @@
+package com.bsd.pjcloud.admin.handler;
+
+/**
+ * 人员调动审批通过后的钱包同步事件。
+ *
+ * <p>在审批事务中发布，仅保留跨线程所需的稳定标识；不携带请求对象或可能过期的用户部门快照。</p>
+ */
+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;
+    }
+}
@@ -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;
+
+/**
+ * 人员调动提交后的钱包同步监听器。
+ *
+ * <p>复用 Spring 管理的异步线程池，按用户当前资料幂等同步；有限重试全部失败后记录错误，供人工补同步。
+ * 重试只保存在当前进程中，服务重启不会恢复尚未完成的调用。</p>
+ */
+@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;
+    }
+
+    /**
+     * 在调动事务提交成功后异步同步钱包，遇到异常、空响应或业务失败时有限重试。
+     *
+     * <p>不启用无事务兜底，避免绕过审批事务提交边界。每次调用均恢复并释放租户上下文，
+     * 不绑定 HTTP 请求；钱包接口已有 NoToken 内部调用标记。中断时保留中断标志并停止重试。</p>
+     *
+     * @param event 已通过审批的调动单、用户主键及租户上下文
+     */
+    @Async
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    public void syncWalletAfterCommit(UserTransactionWalletSyncEvent event) {
+        for (int attempt = 1; attempt <= MAX_SYNC_ATTEMPTS; attempt++) {
+            try {
+                R<Boolean> 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;
+                }
+            }
+        }
+    }
+}
@@ -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<SysUserTransactio

    private final RemoteFlowEngineService remoteFlowEngineService;
    private final RemoteWalletThirdZyService remoteWalletThirdZyService;
-    private final RemoteWalletService remoteWalletService;
    private final RemoteWalletRuleTransactionService walletRuleTransactionService;

    private final FlowUtil flowUtil;

+    /** 发布调岗钱包同步事件，监听器仅在当前审批事务提交成功后执行。 */
+    private final ApplicationEventPublisher applicationEventPublisher;
+
    @Override
    public Page<SysUserTransactionVo> scopePageJoin(Page page, SysUserTransactionDto dto, DataScope dataScope) {
        return baseMapper.scopePageJoin(page, dto, dataScope);
@@ -189,7 +189,13 @@ public class SysUserTransactionServiceImpl extends ServiceImpl<SysUserTransactio


    /**
-     * 回调
+     * 处理人员调动审批结果，并在审批事务中发布钱包同步事件。
+     *
+     * <p>事件只携带稳定标识；用户部门和审批状态成功提交后才异步回查用户，避免钱包读取旧部门。
+     * 钱包调用失败由监听器记录并重试，不再跨线程持有已结束的 HTTP 请求。</p>
+     *
+     * @param dto 流程引擎回调，包含流程实例及审批结果
+     * @param tenantId 流程所属租户；为空时沿用请求租户上下文
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
@@ -224,12 +230,8 @@ public class SysUserTransactionServiceImpl extends ServiceImpl<SysUserTransactio
                    // 通知正元
                    remoteWalletThirdZyService.syncUserById(detail.getUserId());
                }
-                // 通知钱包进行相关修改
-                RequestAttributes requestAttributes = RequestContextHolder.getRequestAttributes();
-                CompletableFuture.runAsync(() -> {
-                    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<SysUserTransactio
            baseMapper.updateById(updateList);
        }
    }
-}
+}
@@ -867,7 +867,8 @@ public class WalletManageServiceImpl extends ServiceImpl<WalletManageMapper, Wal
     * 用户资料变更后的钱包侧同步。
     * <p>用户资料查询与食堂商户解析（均含远程调用）在事务外完成，本地更新经短事务
     * {@link #userChangeInTx} 提交；正元第三方同步放在事务提交后执行，
-     * 避免远程调用和第三方逻辑拉长事务持锁时间，同步异常也不再回滚本地部门变更。</p>
+     * 避免远程调用和第三方逻辑拉长事务持锁时间，同步异常也不再回滚本地部门变更。
+     * 用户查询业务失败、返回空资料或未分配部门时抛出异常，避免向调动重试调用方报告虚假成功。</p>
     *
     * @param userId 变更用户 ID
     * @param isDel 是否删除事件；仅处理非删除事件
@@ -877,9 +878,14 @@ public class WalletManageServiceImpl extends ServiceImpl<WalletManageMapper, Wal
        if (ObjectUtil.isEmpty(userId)) return;
        if (ObjectUtil.isEmpty(isDel)) return;
        if (!isDel) {
-            UserInfo userInfo = userService.getInnerByUserId(userId).getData();
-            if (ObjectUtil.isEmpty(userInfo)) return;
-            SysUser sysUser = userInfo.getSysUser();
+            R<UserInfo> 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<WalletManageMapper, Wal

    /**
     * 用户变更同步的短事务实现：钱包部门与已绑饭卡商户的本地更新。
+     * <p>历史钱包部门为 NULL 时也需要回填；已经一致的钱包不更新，以便失败后的重复通知保持幂等。</p>
     *
     * @param sysUser 已查询的用户资料，取部门和用户 ID
     * @param mch 按用户部门解析的食堂商户；为空时不调整已绑饭卡的商户归属
@@ -897,11 +904,14 @@ public class WalletManageServiceImpl extends ServiceImpl<WalletManageMapper, Wal
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void userChangeInTx(SysUser sysUser, MchManageVo mch) {
-        // 修改钱包部门
+        if (sysUser == null || sysUser.getUserId() == null || sysUser.getDeptId() == null) {
+            throw new IllegalArgumentException("同步钱包部门必须提供用户主键和非空部门");
+        }
+        // SQL 的 <> 不匹配 NULL，必须把历史空部门纳入同一用户的更新范围。
        baseMapper.update(new LambdaUpdateWrapper<WalletManageEntity>()
                .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)) {