feat(bpm): 待办任务站内信实时推送

gaoluyang

5 天以前 c55aca77e84607238c3354168a8642176c575ca1

feat(bpm): 待办任务站内信实时推送

- BPM 任务分配时生成站内信并通过 WebSocket 实时推送，铃铛无需点击即可显示未读
- 修复 WebSocket 握手 userType 为空导致会话被断开的问题
- 新增待办通知消息模板配置 SQL

Co-Authored-By: Claude Code <noreply@anthropic.com>

 sql/mysql/20261110_bpm_task_notify_template.sql

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,25 @@
-- =============================================================================
-- BPM å¾…办任务站内信通知模板
-- ç›®æ ‡åº“：mom-ntsh
-- èƒŒæ™¯ï¼šBpmMessageServiceImpl#sendMessageWhenTaskAssigned å·²æ”¹ä¸ºå‘送站内信，
--       ä¾èµ– system_notify_template ä¸­å­˜åœ¨ code=bpm_task_assigned çš„模板，
--       ç¼ºå¤±æ—¶ä¼šæŠ›ã€Œç«™å†…信模板不存在」并跳过发送（已捕获异常，不影响流程提交）。
-- å‚数：processName（流程名称）、taskName（任务节点）、startUserName（提交人）
-- å¹‚等：code å·²å­˜åœ¨åˆ™è·³è¿‡ï¼Œå¯é‡å¤æ‰§è¡Œã€‚
-- =============================================================================

SET NAMES utf8mb4;

INSERT INTO system_notify_template
    (name, code, nickname, content, type, params, status, remark, creator, updater)
SELECT '流程待办任务提醒', 'bpm_task_assigned', '审批中心',
       '您有新的待办任务：{processName} -「{taskName}」，提交人：{startUserName}。请到「审批中心 â†’ å¾…办任务」及时处理。',
       1, '["processName","taskName","startUserName"]', 0,
       'BPM ä»»åŠ¡åˆ†é…åˆ°å®¡æ‰¹äººæ—¶å‘送，点击通知直接跳转待办任务', '1', '1'
 WHERE NOT EXISTS (SELECT 1 FROM system_notify_template
                    WHERE code = 'bpm_task_assigned' AND deleted = b'0');

-- æ ¡éªŒ
SELECT id, name, code, nickname, content, params, status
  FROM system_notify_template
 WHERE code = 'bpm_task_assigned' AND deleted = b'0';

 yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/security/LoginUserHandshakeInterceptor.java

@@ -1,5 +1,6 @@
package cn.iocoder.yudao.framework.websocket.core.security;

import cn.iocoder.yudao.framework.common.enums.UserTypeEnum;
import cn.iocoder.yudao.framework.security.core.LoginUser;
import cn.iocoder.yudao.framework.security.core.filter.TokenAuthenticationFilter;
import cn.iocoder.yudao.framework.security.core.util.SecurityFrameworkUtils;
@@ -30,6 +31,12 @@
        if (loginUser == null) {
            return false;
        }
        // WebSocket åœ°å€ï¼ˆå¦‚ /infra/ws）不带 /admin-api、/app-api å‰ç¼€ï¼Œ
        // å› æ­¤ mock ç™»å½•æ—¶ WebFrameworkUtils#getLoginUserType å–不到用户类型。
        // è¿™é‡Œå…œåº•ä¸ºç®¡ç†å‘˜ç±»åž‹ï¼Œé¿å… WebSocketSessionManagerImpl ç”¨ null ä½œä¸º key è®°å½•ä¼šè¯æ—¶æŠ› NPE æ–­å¼€è¿žæŽ¥ã€‚
        if (loginUser.getUserType() == null) {
            loginUser.setUserType(UserTypeEnum.ADMIN.getValue());
        }
        WebSocketFrameworkUtils.setLoginUser(loginUser, attributes);
        return true;
    }

 yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/session/WebSocketSessionManagerImpl.java

@@ -43,7 +43,8 @@
        idSessions.put(session.getId(), session);
        // æ·»åŠ åˆ° userSessions ä¸­
        LoginUser user = WebSocketFrameworkUtils.getLoginUser(session);
        if (user == null) {
        // ç”¨æˆ·æˆ–用户类型为空时不处理：ConcurrentHashMap ä¸å…è®¸ null key，会直接抛 NPE æ–­å¼€è¿žæŽ¥
        if (user == null || user.getUserType() == null || user.getId() == null) {
            return;
        }
        ConcurrentMap<Long, CopyOnWriteArrayList<WebSocketSession>> userSessionsMap = userSessions.get(user.getUserType());
@@ -69,7 +70,8 @@
        idSessions.remove(session.getId());
        // ç§»é™¤ä»Ž idSessions ä¸­
        LoginUser user = WebSocketFrameworkUtils.getLoginUser(session);
        if (user == null) {
        // ç”¨æˆ·æˆ–用户类型为空时不处理：ConcurrentHashMap ä¸å…è®¸ null key，会直接抛 NPE æ–­å¼€è¿žæŽ¥
        if (user == null || user.getUserType() == null || user.getId() == null) {
            return;
        }
        ConcurrentMap<Long, CopyOnWriteArrayList<WebSocketSession>> userSessionsMap = userSessions.get(user.getUserType());

 yudao-module-bpm/src/main/java/cn/iocoder/yudao/module/bpm/service/message/BpmMessageServiceImpl.java

@@ -1,18 +1,25 @@
package cn.iocoder.yudao.module.bpm.service.message;

import cn.hutool.core.util.StrUtil;
import cn.iocoder.yudao.module.bpm.service.message.dto.BpmMessageSendWhenProcessInstanceApproveReqDTO;
import cn.iocoder.yudao.module.bpm.service.message.dto.BpmMessageSendWhenProcessInstanceRejectReqDTO;
import cn.iocoder.yudao.module.bpm.service.message.dto.BpmMessageSendWhenTaskCreatedReqDTO;
import cn.iocoder.yudao.module.bpm.service.message.dto.BpmMessageSendWhenTaskTimeoutReqDTO;
import cn.iocoder.yudao.module.system.service.notify.NotifySendService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.flowable.engine.TaskService;
import org.springframework.stereotype.Service;
import org.springframework.validation.annotation.Validated;

import java.util.HashMap;
import java.util.Map;

/**
 * BPM æ¶ˆæ¯ Service å®žçŽ°ç±»
 *
 * æ³¨æ„ï¼šå·²å…³é—­çŸ­ä¿¡é€šçŸ¥åŠŸèƒ½ï¼Œç›¸å…³æ–¹æ³•ä»…记录日志，不再调用短信发送接口，
 * ä»¥é¿å…â€œçŸ­ä¿¡æ¨¡æ¿ä¸å­˜åœ¨â€ç­‰é—®é¢˜ã€‚
 * å½“前只保留【待办任务】的站内信通知：任务分配到具体审批人时，给审批人发一条站内信（系统右上角小铃铛可见）。
 * çŸ­ä¿¡é€šçŸ¥å·²æ•´ä½“关闭，其余方法仅记录日志。
 *
 * @author è¶…级管理员
 */
@@ -20,6 +27,24 @@
@Validated
@Slf4j
public class BpmMessageServiceImpl implements BpmMessageService {

    /**
     * å¾…办任务站内信模板编码，对应 system_notify_template.code
     */
    private static final String TEMPLATE_CODE_TASK_ASSIGNED = "bpm_task_assigned";

    /**
     * ç«™å†…信参数值的最大长度。
     * è¯´æ˜Žï¼šsystem_notify_message.template_params åªæœ‰ varchar(255)，参数值过长会导致插入失败，
     * è¿™é‡Œç»Ÿä¸€æˆªæ–­ï¼Œä¿è¯ä¸€æ¡å¾…办必定能发出去。
     */
    private static final int PARAM_VALUE_MAX_LENGTH = 60;

    @Resource
    private NotifySendService notifySendService;

    @Resource
    private TaskService flowTaskService;

    @Override
    public void sendMessageWhenProcessInstanceApprove(BpmMessageSendWhenProcessInstanceApproveReqDTO reqDTO) {
@@ -31,9 +56,35 @@
        log.info("[sendMessageWhenProcessInstanceReject][已关闭短信通知，跳过发送][processInstanceId: {}]", reqDTO.getProcessInstanceId());
    }

    /**
     * ä»»åŠ¡åˆ†é…åˆ°å®¡æ‰¹äººæ—¶ï¼Œå‘送站内信通知。
     *
     * è¯´æ˜Žï¼šæ‰€æœ‰æµç¨‹èŠ‚点的待办任务都会走 BpmTaskServiceImpl#processTaskAssigned çš„本次调用（加签、转办同样会触发），
     * å› æ­¤è¿™é‡Œå®žçŽ°ä¸€æ¬¡å³å¯è¦†ç›–全部流程的待办通知。
     */
    @Override
    public void sendMessageWhenTaskAssigned(BpmMessageSendWhenTaskCreatedReqDTO reqDTO) {
        log.info("[sendMessageWhenTaskAssigned][已关闭短信通知，跳过发送][processInstanceId: {}]", reqDTO.getProcessInstanceId());
        if (reqDTO.getAssigneeUserId() == null) {
            return;
        }
        // ä»»åŠ¡å¯èƒ½åœ¨åŒä¸€äº‹åŠ¡é‡Œå·²è¢«è‡ªåŠ¨é€šè¿‡ã€è‡ªåŠ¨æ‹’绝或取消（免审批开关、或签被抢占等），此时发待办通知只会误导审批人
        if (flowTaskService.createTaskQuery().taskId(reqDTO.getTaskId()).count() == 0) {
            log.info("[sendMessageWhenTaskAssigned][taskId({}) å·²ä¸åœ¨å¾…办中，跳过待办通知]", reqDTO.getTaskId());
            return;
        }
        Map<String, Object> templateParams = new HashMap<>(4);
        templateParams.put("processName", StrUtil.maxLength(reqDTO.getProcessInstanceName(), PARAM_VALUE_MAX_LENGTH));
        templateParams.put("taskName", StrUtil.maxLength(reqDTO.getTaskName(), PARAM_VALUE_MAX_LENGTH));
        templateParams.put("startUserName", StrUtil.maxLength(reqDTO.getStartUserNickname(), PARAM_VALUE_MAX_LENGTH));
        try {
            notifySendService.sendSingleNotifyToAdmin(reqDTO.getAssigneeUserId(), TEMPLATE_CODE_TASK_ASSIGNED, templateParams);
            log.info("[sendMessageWhenTaskAssigned][已向用户({})发送待办站内信][taskId({}) taskName({})]",
                    reqDTO.getAssigneeUserId(), reqDTO.getTaskId(), reqDTO.getTaskName());
        } catch (Exception ex) {
            // ç«™å†…信发送失败不能影响审批任务本身的创建，否则整个流程提交事务会被回滚
            log.error("[sendMessageWhenTaskAssigned][向用户({})发送待办站内信失败][taskId({})]",
                    reqDTO.getAssigneeUserId(), reqDTO.getTaskId(), ex);
        }
    }

    @Override

 yudao-module-system/src/main/java/cn/iocoder/yudao/module/system/service/notify/NotifySendServiceImpl.java

@@ -2,7 +2,9 @@

import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
import cn.iocoder.yudao.framework.common.enums.UserTypeEnum;
import cn.iocoder.yudao.module.infra.api.websocket.WebSocketSenderApi;
import cn.iocoder.yudao.module.system.dal.dataobject.notify.NotifyTemplateDO;
import cn.iocoder.yudao.module.system.service.notify.dto.NotifyMessagePushDTO;
import com.google.common.annotations.VisibleForTesting;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
@@ -31,6 +33,9 @@
    @Resource
    private NotifyMessageService notifyMessageService;

    @Resource
    private WebSocketSenderApi webSocketSenderApi;

    @Override
    public Long sendSingleNotifyToAdmin(Long userId, String templateCode, Map<String, Object> templateParams) {
        return sendSingleNotify(userId, UserTypeEnum.ADMIN.getValue(), templateCode, templateParams);
@@ -54,7 +59,32 @@

        // å‘送站内信
        String content = notifyTemplateService.formatNotifyTemplateContent(template.getContent(), templateParams);
        return notifyMessageService.createNotifyMessage(userId, userType, template, content, templateParams);
        Long messageId = notifyMessageService.createNotifyMessage(userId, userType, template, content, templateParams);
        // å®žæ—¶æŽ¨é€ï¼šå‰ç«¯é“ƒé“›æ”¶åˆ°åŽç«‹å³åˆ·æ–°æœªè¯»æ•°ï¼Œé¿å…åªèƒ½ç­‰å¾…轮询
        pushToWebSocket(userId, userType, messageId, template, content);
        return messageId;
    }

    /**
     * é€šè¿‡ WebSocket æŠŠç«™å†…信实时推送给用户（该用户所有在线页面都会收到）
     *
     * è¯´æ˜Žï¼šæŽ¨é€åªæ˜¯â€œæå‰é€šçŸ¥å‰ç«¯åˆ·æ–°â€ï¼ŒæŽ¨é€å¤±è´¥ä¸å½±å“ç«™å†…信本身，因此这里吞掉异常仅记录日志。
     */
    private void pushToWebSocket(Long userId, Integer userType, Long messageId,
                                 NotifyTemplateDO template, String content) {
        if (messageId == null) {
            return;
        }
        NotifyMessagePushDTO message = new NotifyMessagePushDTO();
        message.setMessageId(messageId);
        message.setTemplateCode(template.getCode());
        message.setTemplateNickname(template.getNickname());
        message.setTemplateContent(content);
        try {
            webSocketSenderApi.sendObject(userType, userId, NotifyMessagePushDTO.MESSAGE_TYPE, message);
        } catch (Exception ex) {
            log.warn("[pushToWebSocket][向用户({}/{})推送站内信({})失败]", userId, userType, messageId, ex);
        }
    }

    @VisibleForTesting

 yudao-module-system/src/main/java/cn/iocoder/yudao/module/system/service/notify/dto/NotifyMessagePushDTO.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,40 @@
package cn.iocoder.yudao.module.system.service.notify.dto;

import lombok.Data;

/**
 * ç«™å†…ä¿¡ WebSocket å®žæ—¶æŽ¨é€çš„消息体
 *
 * å‰ç«¯æ”¶åˆ° type = notify-message çš„消息后，刷新未读数量和未读列表，无需等待轮询。
 *
 * @author è¶…级管理员
 */
@Data
public class NotifyMessagePushDTO {

    /**
     * æ¶ˆæ¯ç±»åž‹ï¼šç«™å†…ä¿¡
     */
    public static final String MESSAGE_TYPE = "notify-message";

    /**
     * ç«™å†…信编号
     */
    private Long messageId;

    /**
     * ç«™å†…信模板编码
     */
    private String templateCode;

    /**
     * ç«™å†…信发送人名称
     */
    private String templateNickname;

    /**
     * ç«™å†…信内容（已格式化）
     */
    private String templateContent;

}