✨ feat(im): 添加 WebSocket 服务和消息推送功能

新增 IM WebSocket 服务接口及实现，支持异步推送私聊和群聊消息。同时，重构相关消息发送逻辑，提升代码可读性和维护性。 - 新增 ImWebSocketService 接口 - 实现 ImWebSocketServiceImpl 类 - 更新 ImPrivateMessageService 和 ImGroupMessageService 以使用 WebSocket 推送

✨ feat(im): 添加 WebSocket 服务和消息推送功能
新增 IM WebSocket 服务接口及实现，支持异步推送私聊和群聊消息。同时，重构相关消息发送逻辑，提升代码可读性和维护性。 - 新增 ImWebSocketService 接口 - 实现 ImWebSocketServiceImpl 类 - 更新 ImPrivateMessageService 和 ImGroupMessageService 以使用 WebSocket 推送
YunaiV
1 parent e2eab4fe
Showing 19 changed files with 279 additions and 103 deletions
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/group/ImGroupMessageRespVO.java
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/friend/ImFriendServiceImpl.java
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageService.java
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImpl.java
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/websocket/ImWebSocketService.java
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/websocket/ImWebSocketServiceImpl.java
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/friend/FriendUpdateMessage.java
yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImWebSocketEventTest.java
@@ -46,4 +46,7 @@ public class ImGroupMessageRespVO {
     @Schema(description = "回执状态", example = "0")
     private Integer receiptStatus;
  
+    @Schema(description = "已读人数（回执消息、且发送人为当前用户时有值）", example = "3")
+    private Integer readCount;
+
 }
 package cn.iocoder.yudao.module.im.service.friend;
  
+import cn.hutool.extra.spring.SpringUtil;
 import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
 import cn.iocoder.yudao.module.im.controller.admin.friend.vo.ImFriendUpdateReqVO;
 import cn.iocoder.yudao.module.im.dal.dataobject.friend.ImFriendDO;
 import cn.iocoder.yudao.module.im.dal.mysql.friend.ImFriendMapper;
+import cn.iocoder.yudao.module.im.service.message.ImPrivateMessageService;
+import cn.iocoder.yudao.module.im.service.websocket.ImWebSocketService;
+import cn.iocoder.yudao.module.im.service.websocket.dto.friend.FriendUpdateMessage;
+import cn.iocoder.yudao.module.im.service.websocket.dto.ImPrivateMessageDTO;
 import cn.iocoder.yudao.module.system.api.user.AdminUserApi;
-import cn.hutool.extra.spring.SpringUtil;
 import jakarta.annotation.Resource;
 import org.springframework.cache.annotation.CacheEvict;
 import org.springframework.cache.annotation.Cacheable;
@@ -19,13 +23,17 @@ import java.util.Objects;
  
 import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
 import static cn.iocoder.yudao.module.im.dal.redis.RedisKeyConstants.FRIEND;
-import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.*;
+import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.FRIEND_ADD_SELF;
+import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.FRIEND_NOT_FRIEND;
+import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.FRIEND_ADD_TIP_MESSAGE;
+import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.FRIEND_DELETE_TIP_MESSAGE;
  
 /**
  * IM 好友关系 Service 实现类
  *
  * @author 芋道源码
  */
+@Slf4j
 @Service
 @Validated
 public class ImFriendServiceImpl implements ImFriendService {
@@ -36,6 +44,11 @@ public class ImFriendServiceImpl implements ImFriendService {
     @Resource
     private AdminUserApi adminUserApi;
  
+    @Resource
+    private ImWebSocketService imWebSocketService;
+    @Resource
+    private ImPrivateMessageService privateMessageService;
+
     @Override
     @Cacheable(cacheNames = FRIEND, key = "#userId + '_' + #friendUserId", unless = "#result == null")
     public boolean isFriend(Long userId, Long friendUserId) {
@@ -62,10 +75,13 @@ public class ImFriendServiceImpl implements ImFriendService {
         }
  
         // 2. 更新好友属性（目前仅免打扰）
-        ImFriendDO updateObj = new ImFriendDO();
-        updateObj.setId(friend.getId());
-        updateObj.setMuted(reqVO.getMuted());
-        imFriendMapper.updateById(updateObj);
+        imFriendMapper.updateById(new ImFriendDO().setId(friend.getId()).setMuted(reqVO.getMuted()));
+
+        // 3. 推送好友更新通知（多端同步）
+        FriendUpdateMessage websocketContent = new FriendUpdateMessage()
+                .setFriendUserId(reqVO.getFriendUserId()).setMuted(reqVO.getMuted());
+        imWebSocketService.sendPrivateMessageAsync(userId,
+                ImPrivateMessageDTO.ofFriendUpdate(userId, reqVO.getFriendUserId(), websocketContent));
     }
  
     @Override
@@ -82,19 +98,29 @@ public class ImFriendServiceImpl implements ImFriendService {
         getSelf().addFriend0(userId, friendUserId);
         getSelf().addFriend0(friendUserId, userId);
  
-        // TODO @芋艿：好友添加成功后：1）插入 TIP_TEXT 系统提示消息并推送双方；2）通知双方好友列表变更。
-        //  依赖 WebSocket 事件类型扩展（当前无 FRIEND_NEW），待统一后补上。
+        // 3.1 插入 TIP_TEXT 系统提示消息并推送双方
+        privateMessageService.sendTipPrivateMessage(userId, friendUserId, FRIEND_ADD_TIP_MESSAGE);
+        // 3.2 推送好友添加通知给双方
+        imWebSocketService.sendPrivateMessageAsync(userId,
+                ImPrivateMessageDTO.ofFriendAdd(friendUserId, userId));
+        imWebSocketService.sendPrivateMessageAsync(friendUserId,
+                ImPrivateMessageDTO.ofFriendAdd(userId, friendUserId));
     }
  
     @Override
     @Transactional(rollbackFor = Exception.class)
     public void deleteFriend(Long userId, Long friendUserId) {
-        // 双向标记为 DISABLE + 记录 deleteTime
+        // 1. 双向标记为 DISABLE + 记录 deleteTime
         getSelf().deleteFriend0(userId, friendUserId);
         getSelf().deleteFriend0(friendUserId, userId);
  
-        // TODO @芋艿：【对齐】这里缺少了好友删除的系统提示 + FRIEND_DEL WebSocket 推送，
-        //  待 WS 事件类型扩展后补上。
+        // 2.1 插入 TIP_TEXT 系统提示消息并推送双方
+        privateMessageService.sendTipPrivateMessage(userId, friendUserId, FRIEND_DELETE_TIP_MESSAGE);
+        // 2.2 推送好友删除通知给双方
+        imWebSocketService.sendPrivateMessageAsync(userId,
+                ImPrivateMessageDTO.ofFriendDelete(userId, friendUserId));
+        imWebSocketService.sendPrivateMessageAsync(friendUserId,
+                ImPrivateMessageDTO.ofFriendDelete(friendUserId, userId));
     }
  
     /**
@@ -102,11 +128,14 @@ public class ImFriendServiceImpl implements ImFriendService {
      * - 首次：新增记录（status=ENABLE，addTime=now）
      * - 已存在且 ENABLE：无需重复操作
      * - 已存在但 DISABLE：恢复关系（status=ENABLE，刷新 addTime，清空 deleteTime）
+     * <p>
+     * 并发安全：依靠 im_friend 表的唯一索引 uk_user_friend(user_id, friend_user_id) 保证幂等，
+     * 当并发 insert 触发 {@link DuplicateKeyException} 时降级为 select + update。
      */
     @CacheEvict(cacheNames = FRIEND, key = "#userId + '_' + #friendUserId")
     public void addFriend0(Long userId, Long friendUserId) {
         ImFriendDO exists = imFriendMapper.selectByUserIdAndFriendUserId(userId, friendUserId);
-        // 情况一：恢复已删除的好友关系
+        // 情况一：已存在记录 → 恢复或跳过
         if (exists != null) {
             if (CommonStatusEnum.isEnable(exists.getStatus())) {
                 return;
@@ -115,10 +144,16 @@ public class ImFriendServiceImpl implements ImFriendService {
                     .setStatus(CommonStatusEnum.ENABLE.getStatus()).setAddTime(LocalDateTime.now()));
             return;
         }
-        // 情况二：新增好有关系
+        // 情况二：新增好友关系
         ImFriendDO friend = ImFriendDO.builder().userId(userId).friendUserId(friendUserId)
                 .muted(false).status(CommonStatusEnum.ENABLE.getStatus()).addTime(LocalDateTime.now()).build();
-        imFriendMapper.insert(friend);
+        try {
+            imFriendMapper.insert(friend);
+        } catch (DuplicateKeyException e) {
+            // 并发场景：另一个请求已先一步插入，降级走已存在逻辑
+            log.warn("[addFriend0][userId({}) friendUserId({}) 并发插入冲突，降级处理]", userId, friendUserId);
+            addFriend0(userId, friendUserId);
+        }
     }
  
     /**
@@ -6,15 +6,13 @@ import cn.hutool.core.util.IdUtil;
 import cn.hutool.core.util.ObjUtil;
 import cn.hutool.extra.spring.SpringUtil;
 import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
-import cn.iocoder.yudao.framework.common.enums.UserTypeEnum;
 import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
 import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
-import cn.iocoder.yudao.framework.websocket.core.sender.WebSocketMessageSender;
 import cn.iocoder.yudao.module.im.controller.admin.message.vo.group.ImGroupMessageListReqVO;
 import cn.iocoder.yudao.module.im.controller.admin.message.vo.group.ImGroupMessageSendReqVO;
 import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupMemberDO;
 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO;
-import cn.iocoder.yudao.module.im.dal.dataobject.message.content.RecallMessage;
+import cn.iocoder.yudao.module.im.service.websocket.dto.message.RecallMessage;
 import cn.iocoder.yudao.module.im.dal.mysql.message.ImGroupMessageMapper;
 import cn.iocoder.yudao.module.im.dal.redis.message.GroupMessageReadRedisDAO;
 import cn.iocoder.yudao.module.im.enums.message.ImGroupMessageReceiptStatusEnum;
@@ -23,7 +21,8 @@ import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
 import cn.iocoder.yudao.module.im.service.group.ImGroupMemberService;
 import cn.iocoder.yudao.module.im.service.group.ImGroupService;
 import cn.iocoder.yudao.module.im.service.sensitiveword.ImSensitiveWordService;
-import cn.iocoder.yudao.module.im.websocket.ImGroupMessageDTO;
+import cn.iocoder.yudao.module.im.service.websocket.ImWebSocketService;
+import cn.iocoder.yudao.module.im.service.websocket.dto.ImGroupMessageDTO;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.scheduling.annotation.Async;
@@ -69,7 +68,7 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
     private ImSensitiveWordService sensitiveWordService;
  
     @Resource
-    private WebSocketMessageSender webSocketMessageSender;
+    private ImWebSocketService imWebSocketService;
  
     @Override
     public ImGroupMessageDO sendGroupMessage(Long senderId, ImGroupMessageSendReqVO reqVO) {
@@ -98,7 +97,7 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
         groupMessageMapper.insert(message);
  
         // 3. WebSocket 推送给群成员（异步）
-        getSelf().sendGroupMessageEvent(message);
+        sendGroupMessageEvent(message);
         return message;
     }
  
@@ -161,7 +160,7 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
  
             // 4. 按本轮消息最大 id 推进内部游标，跳过这段不可见区间；若游标未前进则直接停止
             Long maxMessageId = getMaxValue(messages, ImGroupMessageDO::getId);
-            if (maxMessageId <= activeMinId) {
+            if (maxMessageId == null || maxMessageId <= activeMinId) {
                 return Collections.emptyList();
             }
             activeMinId = maxMessageId;
@@ -314,7 +313,7 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
         groupMessageMapper.insert(recallMessage);
  
         // 4. 异步推送撤回提示消息（前端据此更新原消息状态 + 插入撤回提示）
-        getSelf().sendGroupMessageEvent(recallMessage);
+        sendGroupMessageEvent(recallMessage);
         return recallMessage;
     }
  
@@ -363,21 +362,12 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
     /**
      * 发送群聊消息 WebSocket 事件
      */
-    @Async
     public void sendGroupMessageEvent(ImGroupMessageDO message) {
         ImGroupMessageDTO dto = ImGroupMessageDTO.ofSend(message);
-        // 广播给群内有效成员（含发送方自己，支持多端同步）
+        // 广播给群内有效成员中的可见用户（含发送方自己，支持多端同步；定向消息自动过滤）
         List<ImGroupMemberDO> members = groupMemberService.getActiveGroupMemberListByGroupId(message.getGroupId());
-        // 定向消息仍需过滤，非目标用户不应收到推送
-        List<Long> receiverList = message.getReceiverUserIds();
-        for (ImGroupMemberDO member : members) {
-            if (CollUtil.isNotEmpty(receiverList) && !receiverList.contains(member.getUserId())
-                    && ObjUtil.notEqual(member.getUserId(), message.getSenderId())) {
-                continue;
-            }
-            webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), member.getUserId(),
-                    ImGroupMessageDTO.TYPE, dto);
-        }
+        Set<Long> targetUserIds = getVisibleUserIds(message, members);
+        imWebSocketService.sendGroupMessageAsync(targetUserIds, dto);
     }
  
     /**
@@ -389,11 +379,11 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
      * @param newMaxMessageId   本次已读位置
      */
     @Async
+    @SuppressWarnings("DataFlowIssue")
     public void readGroupMessageEvent(Long userId, Long groupId, Long prevMaxMessageId, Long newMaxMessageId) {
         // 1. 发送 READ 事件给自己的其他终端（多端同步）
-        ImGroupMessageDTO readDTO = ImGroupMessageDTO.ofRead(userId, groupId, newMaxMessageId);
-        webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), userId,
-                ImGroupMessageDTO.TYPE, readDTO);
+        imWebSocketService.sendGroupMessageAsync(userId,
+                ImGroupMessageDTO.ofRead(userId, groupId, newMaxMessageId));
  
         // 2. 刷新 (prevMaxMessageId, newMaxMessageId] 区间内的待回执消息
         List<ImGroupMessageDO> pendingMessages = groupMessageMapper.selectListByGroupIdAndPendingReceipt(
@@ -422,10 +412,8 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
             }
  
             // 2.2 发送 RECEIPT 事件给消息发送方（只有 ta 关心已读进度）
-            ImGroupMessageDTO receiptDTO = ImGroupMessageDTO.ofReceipt(
-                    message.getId(), groupId, readCount, newReceiptStatus);
-            webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), message.getSenderId(),
-                    ImGroupMessageDTO.TYPE, receiptDTO);
+            imWebSocketService.sendGroupMessageAsync(message.getSenderId(),
+                    ImGroupMessageDTO.ofReceipt(message.getId(), groupId, readCount, newReceiptStatus));
         }
     }
  
@@ -483,6 +471,7 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
                 member -> isMessageVisible(msg, member, member.getUserId()));
     }
  
+
     private ImGroupMessageServiceImpl getSelf() {
         return SpringUtil.getBean(getClass());
     }
@@ -58,4 +58,15 @@ public interface ImPrivateMessageService {
      */
     List<ImPrivateMessageDO> getPrivateMessageList(Long userId, ImPrivateMessageListReqVO reqVO);
  
+    /**
+     * 发送系统提示消息（TIP_TEXT）
+     * <p>
+     * 跳过好友校验、敏感词过滤，用于系统自动生成的提示消息。
+     *
+     * @param senderId   发送人编号
+     * @param receiverId 接收人编号
+     * @param content    提示文本
+     */
+    void sendTipPrivateMessage(Long senderId, Long receiverId, String content);
+
 }
@@ -3,24 +3,21 @@ package cn.iocoder.yudao.module.im.service.message;
 import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.util.IdUtil;
 import cn.hutool.core.util.ObjUtil;
-import cn.hutool.extra.spring.SpringUtil;
-import cn.iocoder.yudao.framework.common.enums.UserTypeEnum;
 import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
 import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
-import cn.iocoder.yudao.framework.websocket.core.sender.WebSocketMessageSender;
 import cn.iocoder.yudao.module.im.controller.admin.message.vo.privates.ImPrivateMessageListReqVO;
 import cn.iocoder.yudao.module.im.controller.admin.message.vo.privates.ImPrivateMessageSendReqVO;
 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImPrivateMessageDO;
-import cn.iocoder.yudao.module.im.dal.dataobject.message.content.RecallMessage;
 import cn.iocoder.yudao.module.im.dal.mysql.message.ImPrivateMessageMapper;
 import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
 import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
 import cn.iocoder.yudao.module.im.service.friend.ImFriendService;
 import cn.iocoder.yudao.module.im.service.sensitiveword.ImSensitiveWordService;
-import cn.iocoder.yudao.module.im.websocket.ImPrivateMessageDTO;
+import cn.iocoder.yudao.module.im.service.websocket.ImWebSocketService;
+import cn.iocoder.yudao.module.im.service.websocket.dto.ImPrivateMessageDTO;
+import cn.iocoder.yudao.module.im.service.websocket.dto.message.RecallMessage;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.validation.annotation.Validated;
@@ -31,9 +28,7 @@ import java.util.List;
 import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
 import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList;
 import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.*;
-import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.MESSAGE_MAX_PULL_SIZE;
-import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.MESSAGE_PRIVATE_PULL_MAX_DAYS;
-import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.MESSAGE_RECALL_TIMEOUT_MINUTES;
+import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.*;
  
 /**
  * IM 私聊消息 Service 实现类
@@ -54,7 +49,7 @@ public class ImPrivateMessageServiceImpl implements ImPrivateMessageService {
     private ImSensitiveWordService sensitiveWordService;
  
     @Resource
-    private WebSocketMessageSender webSocketMessageSender;
+    private ImWebSocketService imWebSocketService;
  
     @Override
     public ImPrivateMessageDO sendPrivateMessage(Long senderId, ImPrivateMessageSendReqVO reqVO) {
@@ -80,8 +75,10 @@ public class ImPrivateMessageServiceImpl implements ImPrivateMessageService {
                 .setSenderId(senderId).setStatus(ImMessageStatusEnum.UNREAD.getStatus()).setSendTime(LocalDateTime.now()));
         privateMessageMapper.insert(message);
  
-        // 3. WebSocket 推送给接收方（异步）
-        getSelf().sendPrivateMessageEvent(message);
+        // 3. WebSocket 异步推送（接收方 + 发送方多端同步）
+        ImPrivateMessageDTO websocketMessage = ImPrivateMessageDTO.ofSend(message);
+        imWebSocketService.sendPrivateMessageAsync(message.getReceiverId(), websocketMessage);
+        imWebSocketService.sendPrivateMessageAsync(message.getSenderId(), websocketMessage);
         return message;
     }
  
@@ -116,7 +113,10 @@ public class ImPrivateMessageServiceImpl implements ImPrivateMessageService {
  
         // 2. 异步发送 READ + RECEIPT 事件（携带已读位置）
         Long maxReadId = CollUtil.max(messageIds);
-        getSelf().readPrivateMessageEvent(userId, receiverId, maxReadId);
+        imWebSocketService.sendPrivateMessageAsync(userId,
+                ImPrivateMessageDTO.ofRead(userId, receiverId, maxReadId));
+        imWebSocketService.sendPrivateMessageAsync(receiverId,
+                ImPrivateMessageDTO.ofReceipt(userId, receiverId, maxReadId));
     }
  
     @Override
@@ -153,52 +153,30 @@ public class ImPrivateMessageServiceImpl implements ImPrivateMessageService {
         privateMessageMapper.insert(tipMessage);
  
         // 4. 异步推送撤回提示消息（前端据此更新原消息状态 + 插入撤回提示）
-        getSelf().sendPrivateMessageEvent(tipMessage);
+        ImPrivateMessageDTO websocketMessage = ImPrivateMessageDTO.ofSend(tipMessage);
+        imWebSocketService.sendPrivateMessageAsync(tipMessage.getReceiverId(), websocketMessage);
+        imWebSocketService.sendPrivateMessageAsync(tipMessage.getSenderId(), websocketMessage);
         return tipMessage;
     }
  
-    /**
-     * 发送私聊消息 WebSocket 事件
-     */
-    @Async
-    public void sendPrivateMessageEvent(ImPrivateMessageDO message) {
-        ImPrivateMessageDTO dto = ImPrivateMessageDTO.ofSend(message);
-        // 推送给接收方
-        webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), message.getReceiverId(),
-                ImPrivateMessageDTO.TYPE, dto);
-
-        // 推送给发送方自己的其他终端（多端同步）
-        webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), message.getSenderId(),
-                ImPrivateMessageDTO.TYPE, dto);
-    }
-
-    /**
-     * 发送已读 + 已读回执 WebSocket 事件
-     *
-     * @param userId     当前用户编号
-     * @param receiverId 对方用户编号
-     * @param maxReadId  已读位置（最大已读消息编号）
-     */
-    @Async
-    public void readPrivateMessageEvent(Long userId, Long receiverId, Long maxReadId) {
-        // 1. 发送 READ 事件给自己的其他终端（多端同步）
-        ImPrivateMessageDTO readDTO = ImPrivateMessageDTO.ofRead(userId, receiverId, maxReadId);
-        webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), userId,
-                ImPrivateMessageDTO.TYPE, readDTO);
-
-        // 2. 发送 RECEIPT 事件给对方（已读回执）
-        ImPrivateMessageDTO receiptDTO = ImPrivateMessageDTO.ofReceipt(userId, receiverId, maxReadId);
-        webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), receiverId,
-                ImPrivateMessageDTO.TYPE, receiptDTO);
-    }
-
     @Override
     public List<ImPrivateMessageDO> getPrivateMessageList(Long userId, ImPrivateMessageListReqVO reqVO) {
         return privateMessageMapper.selectHistoryList(userId, reqVO.getReceiverId(), reqVO.getMaxId(), reqVO.getLimit());
     }
  
-    private ImPrivateMessageServiceImpl getSelf() {
-        return SpringUtil.getBean(getClass());
+    @Override
+    public void sendTipPrivateMessage(Long senderId, Long receiverId, String content) {
+        // 1. 插入 TIP_TEXT 消息
+        ImPrivateMessageDO tipMessage = new ImPrivateMessageDO()
+                .setClientMessageId(IdUtil.fastSimpleUUID()).setSenderId(senderId).setReceiverId(receiverId)
+                .setType(ImMessageTypeEnum.TIP_TEXT.getType()).setContent(content)
+                .setStatus(ImMessageStatusEnum.UNREAD.getStatus()).setSendTime(LocalDateTime.now());
+        privateMessageMapper.insert(tipMessage);
+
+        // 2. 推送给双方
+        ImPrivateMessageDTO dto = ImPrivateMessageDTO.ofSend(tipMessage);
+        imWebSocketService.sendPrivateMessageAsync(receiverId, dto);
+        imWebSocketService.sendPrivateMessageAsync(senderId, dto);
     }
  
 }
+package cn.iocoder.yudao.module.im.service.websocket;
+
+import cn.iocoder.yudao.module.im.service.websocket.dto.ImGroupMessageDTO;
+import cn.iocoder.yudao.module.im.service.websocket.dto.ImPrivateMessageDTO;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * IM WebSocket 推送 Service 接口
+ * <p>
+ * 统一封装 WebSocket 消息推送，所有方法默认异步执行。
+ *
+ * @author 芋道源码
+ */
+public interface ImWebSocketService {
+
+    /**
+     * 异步推送私聊消息给指定用户
+     *
+     * @param userId 目标用户编号
+     * @param dto    私聊消息 DTO
+     */
+    void sendPrivateMessageAsync(Long userId, ImPrivateMessageDTO dto);
+
+    /**
+     * 异步推送群聊消息给指定用户
+     *
+     * @param userId 目标用户编号
+     * @param dto    群聊消息 DTO
+     */
+    default void sendGroupMessageAsync(Long userId, ImGroupMessageDTO dto) {
+        sendGroupMessageAsync(Collections.singleton(userId), dto);
+    }
+
+    /**
+     * 异步批量推送群聊消息给多个用户
+     *
+     * @param userIds 目标用户编号列表
+     * @param dto     群聊消息 DTO
+     */
+    void sendGroupMessageAsync(Collection<Long> userIds, ImGroupMessageDTO dto);
+
+}
+package cn.iocoder.yudao.module.im.service.websocket;
+
+import cn.iocoder.yudao.framework.common.enums.UserTypeEnum;
+import cn.iocoder.yudao.framework.websocket.core.sender.WebSocketMessageSender;
+import cn.iocoder.yudao.module.im.service.websocket.dto.ImGroupMessageDTO;
+import cn.iocoder.yudao.module.im.service.websocket.dto.ImPrivateMessageDTO;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Service;
+
+import java.util.Collection;
+
+/**
+ * IM WebSocket 推送 Service 实现类
+ *
+ * @author 芋道源码
+ */
+@Service
+@Slf4j
+public class ImWebSocketServiceImpl implements ImWebSocketService {
+
+    @Resource
+    private WebSocketMessageSender webSocketMessageSender;
+
+    @Async
+    @Override
+    public void sendPrivateMessageAsync(Long userId, ImPrivateMessageDTO dto) {
+        try {
+            webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), userId,
+                    ImPrivateMessageDTO.TYPE, dto);
+        } catch (Exception e) {
+            log.error("[sendPrivateMessageAsync][userId({}) dto({}) 发送失败]", userId, dto, e);
+        }
+    }
+
+    @Async
+    @Override
+    public void sendGroupMessageAsync(Collection<Long> userIds, ImGroupMessageDTO dto) {
+        for (Long userId : userIds) {
+            try {
+                webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), userId,
+                        ImGroupMessageDTO.TYPE, dto);
+            } catch (Exception e) {
+                log.error("[sendGroupMessageAsync][userId({}) dto({}) 发送失败]", userId, dto, e);
+            }
+        }
+    }
+
+}
-package cn.iocoder.yudao.module.im.websocket;
+package cn.iocoder.yudao.module.im.service.websocket.dto;
  
 import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO;
-package cn.iocoder.yudao.module.im.websocket;
+package cn.iocoder.yudao.module.im.service.websocket.dto;
  
+import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
 import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImPrivateMessageDO;
 import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
+import cn.iocoder.yudao.module.im.service.websocket.dto.friend.FriendUpdateMessage;
 import lombok.Data;
 import lombok.experimental.Accessors;
  
@@ -100,4 +102,46 @@ public class ImPrivateMessageDTO {
                 .setSenderId(senderId).setReceiverId(receiverId);
     }
  
+    /**
+     * 构建好友添加推送 DTO（通知对方"你有一位新好友"）
+     *
+     * @param userId       当前用户编号（即发起添加的用户）
+     * @param friendUserId 好友的用户编号（即接收通知的用户）
+     * @return 私聊 DTO
+     */
+    public static ImPrivateMessageDTO ofFriendAdd(Long userId, Long friendUserId) {
+        return new ImPrivateMessageDTO()
+                .setType(ImMessageTypeEnum.FRIEND_ADD.getType())
+                .setSenderId(userId).setReceiverId(friendUserId)
+                .setSendTime(LocalDateTime.now());
+    }
+
+    /**
+     * 构建好友删除推送 DTO（通知对方"好友关系已解除"）
+     *
+     * @param userId       当前用户编号（即发起删除的用户）
+     * @param friendUserId 好友的用户编号（即接收通知的用户）
+     * @return 私聊 DTO
+     */
+    public static ImPrivateMessageDTO ofFriendDelete(Long userId, Long friendUserId) {
+        return new ImPrivateMessageDTO()
+                .setType(ImMessageTypeEnum.FRIEND_DELETE.getType())
+                .setSenderId(userId).setReceiverId(friendUserId)
+                .setSendTime(LocalDateTime.now());
+    }
+    /**
+     * 构建好友更新推送 DTO（通知自己多端同步好友属性变更）
+     *
+     * @param userId       当前用户编号
+     * @param friendUserId 好友的用户编号
+     * @param content      更新后的好友属性
+     * @return 私聊 DTO
+     */
+    public static ImPrivateMessageDTO ofFriendUpdate(Long userId, Long friendUserId, FriendUpdateMessage content) {
+        return new ImPrivateMessageDTO()
+                .setType(ImMessageTypeEnum.FRIEND_UPDATE.getType())
+                .setSenderId(userId).setReceiverId(friendUserId)
+                .setContent(JsonUtils.toJsonString(content)).setSendTime(LocalDateTime.now());
+    }
+
 }
+package cn.iocoder.yudao.module.im.service.websocket.dto.friend;
+
+import lombok.Data;
+import lombok.experimental.Accessors;
+
+/**
+ * 好友更新消息内容（目前仅免打扰）
+ */
+@Data
+@Accessors(chain = true)
+public class FriendUpdateMessage {
+
+    /**
+     * 好友的用户编号
+     */
+    private Long friendUserId;
+    /**
+     * 是否免打扰
+     */
+    private Boolean muted;
+
+}
-package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
+package cn.iocoder.yudao.module.im.service.websocket.dto.message;
  
 import lombok.Data;
  
-package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
+package cn.iocoder.yudao.module.im.service.websocket.dto.message;
  
 import lombok.Data;
  
-package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
+package cn.iocoder.yudao.module.im.service.websocket.dto.message;
  
 import lombok.Data;
  
-package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
+package cn.iocoder.yudao.module.im.service.websocket.dto.message;
  
 import lombok.Data;
  
-package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
+package cn.iocoder.yudao.module.im.service.websocket.dto.message;
  
 import lombok.Data;
  
-package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
+package cn.iocoder.yudao.module.im.service.websocket.dto.message;
  
 import lombok.Data;
 import lombok.experimental.Accessors;
-package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
+package cn.iocoder.yudao.module.im.service.websocket.dto.message;
  
 import lombok.Data;
  
-package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
+package cn.iocoder.yudao.module.im.service.websocket.dto.message;
  
 import lombok.Data;
  
@@ -9,7 +9,7 @@ import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
 import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
 import cn.iocoder.yudao.module.im.service.friend.ImFriendService;
 import cn.iocoder.yudao.module.im.service.sensitiveword.ImSensitiveWordService;
-import cn.iocoder.yudao.module.im.websocket.ImPrivateMessageDTO;
+import cn.iocoder.yudao.module.im.service.websocket.dto.ImPrivateMessageDTO;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.InjectMocks;