微服务重构：基础对话（一对一、群）版本

resources/database-sql/webchat-payment.sql

@@ -1,44 +1,44 @@
 -- 红包信息表
 CREATE TABLE webchat_payment.`web_chat_red_packet` (
-                                                       `ID` bigint unsigned NOT NULL AUTO_INCREMENT COMMENT '自增ID',
-                                                       `send_user_id` char(100) NOT NULL COMMENT '红包发送人',
-                                                       `receiver_user_id` char(100) NOT NULL COMMENT '接受人',
-                                                       `type` int(4) NOT NULL COMMENT '消息类型',
-                                                       `count` int(4) NOT NULL COMMENT '红包个数',
-                                                       `status` int(4) NOT NULL COMMENT '状态',
-                                                       `total_money` DECIMAL(10, 2) default '0.00' COMMENT '金额',
-                                                       `CREATE_BY` char(100) DEFAULT NULL COMMENT '创建人',
-                                                       `CREATE_DATE` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
-                                                       `expire_date` datetime NOT NULL COMMENT '过期时间',
-                                                       `UPDATE_BY` char(100) DEFAULT NULL COMMENT '更新人',
-                                                       `UPDATE_DATE` datetime DEFAULT NULL COMMENT '更新时间',
-                                                       `VERSION` int DEFAULT '0' COMMENT '版本',
-                                                       PRIMARY KEY (`ID`),
-                                                       KEY `INDEX_SEND_USER_ID` (`send_user_id`),
-                                                       KEY `INDEX_STATUS_EXPIRE_DATE` (`status`, `expire_date`)
+   `ID` bigint unsigned NOT NULL AUTO_INCREMENT COMMENT '自增ID',
+   `send_user_id` char(100) NOT NULL COMMENT '红包发送人',
+   `receiver_user_id` char(100) NOT NULL COMMENT '接受人',
+   `type` int(4) NOT NULL COMMENT '消息类型',
+   `count` int(4) NOT NULL COMMENT '红包个数',
+   `status` int(4) NOT NULL COMMENT '状态',
+   `total_money` DECIMAL(10, 2) default '0.00' COMMENT '金额',
+   `CREATE_BY` char(100) DEFAULT NULL COMMENT '创建人',
+   `CREATE_DATE` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+   `expire_date` datetime NOT NULL COMMENT '过期时间',
+   `UPDATE_BY` char(100) DEFAULT NULL COMMENT '更新人',
+   `UPDATE_DATE` datetime DEFAULT NULL COMMENT '更新时间',
+   `VERSION` int DEFAULT '0' COMMENT '版本',
+   PRIMARY KEY (`ID`),
+   KEY `INDEX_SEND_USER_ID` (`send_user_id`),
+   KEY `INDEX_STATUS_EXPIRE_DATE` (`status`, `expire_date`)
 ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='红包信息表';
 
 -- 红包拆分记录明细表
 CREATE TABLE webchat_payment.`web_chat_red_packet_record` (
-                                                              `ID` bigint unsigned NOT NULL AUTO_INCREMENT COMMENT '自增ID',
-                                                              `red_packet_id` bigint NOT NULL COMMENT '红包id',
-                                                              `user_id` char(100) NOT NULL COMMENT '领取人',
-                                                              `money` DECIMAL(10, 2) default '0.00' COMMENT '领取金额',
-                                                              `CREATE_DATE` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
-                                                              PRIMARY KEY (`ID`),
-                                                              KEY `INDEX_RED_PACKET_ID` (`red_packet_id`),
-                                                              KEY `INDEX_USER_ID` (`user_id`)
+      `ID` bigint unsigned NOT NULL AUTO_INCREMENT COMMENT '自增ID',
+      `red_packet_id` bigint NOT NULL COMMENT '红包id',
+      `user_id` char(100) NOT NULL COMMENT '领取人',
+      `money` DECIMAL(10, 2) default '0.00' COMMENT '领取金额',
+      `CREATE_DATE` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+      PRIMARY KEY (`ID`),
+      KEY `INDEX_RED_PACKET_ID` (`red_packet_id`),
+      KEY `INDEX_USER_ID` (`user_id`)
 ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='红包拆分记录明细表';
 
 -- 用户钱包
 CREATE TABLE webchat_payment.`web_chat_user_wallet` (
-                                                        `ID` bigint unsigned NOT NULL AUTO_INCREMENT COMMENT '自增ID',
-                                                        `trans_event` int(4) NOT NULL COMMENT '事件类型',
-                                                        `trans_type` int(4) NOT NULL COMMENT '收入/支出',
-                                                        `user_id` char(100) NOT NULL COMMENT '用户id',
-                                                        `target_user_id` char(100) NOT NULL COMMENT '目标用户',
-                                                        `money` DECIMAL(10, 2) default '0.00' COMMENT '流转金额',
-                                                        `trans_date` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '流转时间',
-                                                        PRIMARY KEY (`ID`),
-                                                        KEY `INDEX_USER_ID` (`user_id`)
+    `ID` bigint unsigned NOT NULL AUTO_INCREMENT COMMENT '自增ID',
+    `trans_event` int(4) NOT NULL COMMENT '事件类型',
+    `trans_type` int(4) NOT NULL COMMENT '收入/支出',
+    `user_id` char(100) NOT NULL COMMENT '用户id',
+    `target_user_id` char(100) NOT NULL COMMENT '目标用户',
+    `money` DECIMAL(10, 2) default '0.00' COMMENT '流转金额',
+    `trans_date` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '流转时间',
+    PRIMARY KEY (`ID`),
+    KEY `INDEX_USER_ID` (`user_id`)
 ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='用户钱包';

resources/database-sql/webchat-ugc.sql

@@ -1,2 +1,17 @@
--- 用户剩下内容库：主要存放对话消息等等
-
+-- 消息持久化数据表
+CREATE TABLE webchat_ugc.`web_chat_message` (
+     `ID` bigint unsigned NOT NULL AUTO_INCREMENT COMMENT '自增ID',
+     `sender` char(100) NOT NULL COMMENT '发送人',
+     `receiver` char(100) NOT NULL COMMENT '接收人',
+     `proxy_sender` char(100) DEFAULT NULL COMMENT '消息代理发送人（应用在群聊场景）',
+     `message` text DEFAULT NULL COMMENT '消息内容',
+     `image` varchar(300) DEFAULT NULL COMMENT '图片',
+     `type` tinyint(1) DEFAULT 0 COMMENT '消息类型',
+     `IS_READ` tinyint(1) DEFAULT 0 COMMENT '是否已读',
+     `SEND_DATE` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '消息时间',
+     `UPDATE_DATE` datetime DEFAULT NULL COMMENT '更新时间',
+     `VERSION` int DEFAULT '0' COMMENT '版本',
+     PRIMARY KEY (`ID`),
+     KEY `INDEX_SENDER_PROXY_SENDER` (`sender`, `proxy_sender`),
+     KEY `INDEX_RECEIVER` (`receiver`)
+) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='消息持久化数据表';

webchat-client-chat/src/main/java/com/webchat/client/chat/controller/ChatMessageController.java

@@ -6,8 +6,10 @@ import com.webchat.common.bean.APIResponseBean;
 import com.webchat.common.bean.APIResponseBeanUtil;
 import com.webchat.common.helper.SessionHelper;
 import com.webchat.domain.vo.response.chatting.ChattingListResponseVO;
+import com.webchat.domain.vo.response.mess.ChatMessageResponseVO;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
@@ -35,4 +37,18 @@ public class ChatMessageController {
         return APIResponseBeanUtil.success(chatMessageService.listChatting(userId, lastChatTime, size));
     }
 
+    /***
+     * 查询跟用户聊天记录
+     * @return
+     */
+    @GetMapping("/list/{chatUserId}")
+    public APIResponseBean<List<ChatMessageResponseVO>> list(@PathVariable String chatUserId,
+                                                             @RequestParam(value = "size", required = false, defaultValue = "50") Integer size,
+                                                             @RequestParam(value = "lastId", required = false) Long lastId,
+                                                             @RequestParam(value = "fixedMessageId", required = false) Long fixedMessageId) {
+        String currUserId = SessionHelper.getCurrentUserId();
+        return APIResponseBeanUtil.success(
+                chatMessageService.list(currUserId, chatUserId, lastId, fixedMessageId, size));
+    }
+
 }

webchat-client-chat/src/main/java/com/webchat/client/chat/service/ChatMessageService.java

@@ -6,11 +6,15 @@ import com.webchat.common.bean.APIResponseBeanUtil;
 import com.webchat.common.exception.BusinessException;
 import com.webchat.common.util.JsonUtil;
 import com.webchat.domain.vo.response.chatting.ChattingListResponseVO;
+import com.webchat.domain.vo.response.mess.ChatMessageResponseVO;
 import com.webchat.rmi.ugc.ChatMessageClient;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestParam;
 
+import java.util.Collections;
 import java.util.List;
 
 
@@ -22,6 +26,14 @@ public class ChatMessageService {
     @Autowired
     private ChatMessageClient chatMessageClient;
 
+    public List<ChatMessageResponseVO> list(String currentUserId, String chatUserId, Long lastId, Long fixedMessageId, Integer size) {
+        APIResponseBean<List<ChatMessageResponseVO>> messageResponse = chatMessageClient.list(currentUserId, chatUserId, size, lastId, fixedMessageId);
+        if (APIResponseBeanUtil.isOk(messageResponse)) {
+            return messageResponse.getData();
+        }
+        return Collections.emptyList();
+    }
+
     public List<ChattingListResponseVO> listChatting(String userId, Long lastChatTime, Integer size) {
 
         // RPC 远程调用，请求 UGC 核心服务获取对话列表数据

webchat-common/src/main/java/com/webchat/common/enums/messagequeue/MessageQueueEnum.java

@@ -13,6 +13,8 @@ public enum MessageQueueEnum {
 
     QUEUE_CHAT_MESSAGE("queue_chat_message", "聊天消息队列"),
 
+    QUEUE_PERSISTENT_MESSAGE("queue_persistent_message", "消息持久化队列"),
+
     QUEUE_CHAT_VIDEO_P2P("queue_chat_video_p2p", "P2P（一对一）音视频聊天信令消息队列"),
 
     QUEUE_CHAT_VIDEO_MESH("queue_chat_video_mesh", "基于Mesh模式的多人音视频聊天信令消息队列");

webchat-common/src/main/java/com/webchat/common/service/RedisService.java

@@ -451,7 +451,7 @@ public class RedisService {
         } catch (Exception e) {
             log.error("redis zrange error,errorMessage:{}", e.getMessage());
         }
-        return null;
+        return new HashSet<>();
     }
 
     /**

webchat-connect/src/main/java/com/webchat/connect/messagequeue/consumer/service/ChatMessageConsumeService.java

@@ -3,11 +3,14 @@ package com.webchat.connect.messagequeue.consumer.service;
 
 import com.webchat.common.constants.ConnectConstants;
 import com.webchat.common.enums.RoleCodeEnum;
+import com.webchat.common.enums.messagequeue.MessageQueueEnum;
+import com.webchat.common.service.messagequeue.producer.MessageQueueProducer;
 import com.webchat.common.util.JsonUtil;
 import com.webchat.connect.service.AccountService;
 import com.webchat.connect.websocket.handler.ChatWebSocketEndPointServletHandler;
 import com.webchat.domain.vo.request.mess.ChatMessageRequestVO;
 import com.webchat.domain.vo.response.UserBaseResponseInfoVO;
+import com.webchat.domain.vo.response.mess.ChatMessageResponseVO;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections.MapUtils;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -16,6 +19,7 @@ import org.springframework.web.socket.TextMessage;
 import org.springframework.web.socket.WebSocketSession;
 
 import java.io.IOException;
+import java.util.Date;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
@@ -32,6 +36,9 @@ public class ChatMessageConsumeService {
     @Autowired
     private AccountService accountService;
 
+    @Autowired
+    private MessageQueueProducer<ChatMessageRequestVO, Long> messageQueueProducer;
+
     /**
      * 处理来自对话topic 广播的消息
      * @param message
@@ -44,22 +51,37 @@ public class ChatMessageConsumeService {
         }
         // 获取对话消息接受人
         String receiverId = chatMessage.getReceiverId();
-        UserBaseResponseInfoVO accountInfo = accountService.accountInfo(receiverId);
-        if (accountInfo == null) {
+        UserBaseResponseInfoVO receiver = accountService.accountInfo(receiverId);
+        if (receiver == null) {
             return;
         }
         // 最终要推送的用户
         Set<String> receivers = new HashSet<>();
-
         /**
          * 处理一对一、群聊
          */
-        if (RoleCodeEnum.GROUP.getCode().equals(accountInfo.getRoleCode())) {
+        // 构建消息响应VO，用于推送客户端
+        ChatMessageResponseVO messageResponse = new ChatMessageResponseVO();
+        messageResponse.setMessage(chatMessage.getMessage());
+        messageResponse.setSenderId(chatMessage.getSenderId());
+        messageResponse.setReceiverId(chatMessage.getReceiverId());
+        messageResponse.setTime(new Date().getTime());
+        messageResponse.setReceiverId(receiverId);
+        messageResponse.setReceiver(receiver);
+        messageResponse.setType(chatMessage.getType());
+        if (RoleCodeEnum.GROUP.getCode().equals(receiver.getRoleCode())) {
             /**
-             * 群聊
+             * 群聊(消息代理人，角色反转，原消息接受人-群聊反转为消息发送人，实际消息发送人作为被代理发送人)
              */
-            // TODO
-        } else if (RoleCodeEnum.isUserRole(accountInfo.getRoleCode())) {
+            // 由群组代理消息发送
+            messageResponse.setSenderId(receiverId);
+            // 被代理消息发送人
+            messageResponse.setProxySenderId(chatMessage.getSenderId());
+            UserBaseResponseInfoVO sender = accountService.accountInfo(chatMessage.getSenderId());
+            messageResponse.setProxySender(sender);
+            Set<String> groupUserIds = accountService.getGroupUserIds(receiverId);
+            receivers.addAll(groupUserIds);
+        } else if (RoleCodeEnum.isUserRole(receiver.getRoleCode())) {
             /**
              * 用户账号间一对一对话
              */
@@ -68,14 +90,21 @@ public class ChatMessageConsumeService {
             log.error("不支持的对话场景 =====> receiverId:{}", receiverId);
             return;
         }
-        // 消息推送
-        this.doPushMessage2Client(chatMessage, receivers);
 
         /**
-         * 消息持久化 TODO
+         * 《实时场景》消息推送
          */
-    }
+        this.doPushMessage2Client(messageResponse, receivers);
 
+        /**
+         * 《离线场景》持久化消息队列，保存离线消息，同时会将数据同步到ES用于后续的RAG问答和消息搜索
+         *
+         *  Consumer在UGC服务：
+         *  com.webchat.ugc.messaegqueue.consumer.redis.PersistentMessageRedisMQConsumer
+         *  com.webchat.ugc.messaegqueue.consumer.rocketmq.PersistentMessageRocketMQConsumer
+         */
+        messageQueueProducer.send(MessageQueueEnum.QUEUE_PERSISTENT_MESSAGE, chatMessage);
+    }
 
     /**
      * 推送消息到客户端
@@ -83,7 +112,7 @@ public class ChatMessageConsumeService {
      * @param chatMessage
      * @param receivers
      */
-    private void doPushMessage2Client(ChatMessageRequestVO chatMessage, Set<String> receivers) {
+    private void doPushMessage2Client(ChatMessageResponseVO chatMessage, Set<String> receivers) {
 
         Set<String> bizCodes = ConnectConstants.ConnectBiz.getBizCode(ConnectConstants.BizEnum.CHAT);
         for (String bizCode : bizCodes) {
@@ -92,8 +121,12 @@ public class ChatMessageConsumeService {
                  continue;
             }
             chatMessage.setTime(System.currentTimeMillis());
-            String message = JsonUtil.toJsonString(chatMessage);
             for (String receiver : receivers) {
+                if (receiver.equals(chatMessage.getProxySenderId())) {
+                    continue;
+                }
+                chatMessage.setReceiverId(receiver);
+                String message = JsonUtil.toJsonString(chatMessage);
                 WebSocketSession socketSession = wsSessionMap.get(receiver);
                 if (socketSession == null || !socketSession.isOpen()) {
                     // 当前接受人可能离线、或者当前接受人的ws 链接在其他节点（

webchat-connect/src/main/java/com/webchat/connect/service/AccountService.java

@@ -11,6 +11,8 @@ import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import java.util.Set;
+
 @Slf4j
 @Service
 public class AccountService {
@@ -35,4 +37,19 @@ public class AccountService {
                    account, JsonUtil.toJsonString(responseBean));
         return null;
     }
+
+    /**
+     * 获取群组下的群成员用户id集合
+     *
+     * @param groupAccount
+     * @return
+     */
+    public Set<String> getGroupUserIds(String groupAccount) {
+
+        APIResponseBean<Set<String>> responseBean = userServiceClient.getGroupUserIds(groupAccount);
+        if (APIResponseBeanUtil.isOk(responseBean)) {
+            return responseBean.getData();
+        }
+        return null;
+    }
 }

webchat-connect/src/main/java/com/webchat/connect/websocket/handler/ChatWebSocketEndPointServletHandler.java

@@ -2,10 +2,8 @@ package com.webchat.connect.websocket.handler;
 
 import com.webchat.common.enums.RoleCodeEnum;
 import com.webchat.common.enums.messagequeue.MessageBroadChannelEnum;
-import com.webchat.common.enums.messagequeue.MessageQueueEnum;
 import com.webchat.common.service.messagequeue.producer.MessageQueueProducer;
 import com.webchat.common.util.JsonUtil;
-import com.webchat.common.util.StringUtil;
 import com.webchat.connect.service.AccountService;
 import com.webchat.domain.vo.request.mess.ChatMessageRequestVO;
 import com.webchat.domain.vo.response.UserBaseResponseInfoVO;
@@ -28,7 +26,7 @@ import java.util.concurrent.ConcurrentHashMap;
 public class ChatWebSocketEndPointServletHandler extends TextWebSocketHandler {
 
     @Autowired
-    private MessageQueueProducer<ChatMessageRequestVO, Long> messageQueueProducer;
+    private MessageQueueProducer<Object, Long> messageQueueProducer;
 
     @Autowired
     private AccountService accountService;
@@ -88,19 +86,6 @@ public class ChatWebSocketEndPointServletHandler extends TextWebSocketHandler {
         if (accountInfo == null) {
             return;
         }
-        // 刷新聊天对话列表
-        Long msgTime = System.currentTimeMillis();
-        ChatMessageRequestVO refreshChattingMessage = new ChatMessageRequestVO();
-        refreshChattingMessage.setSenderId(chatMessage.getSenderId());
-        refreshChattingMessage.setReceiverId(chatMessage.getReceiverId());
-        refreshChattingMessage.setTime(msgTime);
-        messageQueueProducer.send(MessageQueueEnum.QUEUE_CHATTING_LIST_REFRESH, refreshChattingMessage);
-        if (RoleCodeEnum.isUserRole(accountInfo.getRoleCode())) {
-            // 如果消息接受账号是用户类型，也需要刷新用户账号列表
-            refreshChattingMessage.setSenderId(chatMessage.getReceiverId());
-            refreshChattingMessage.setReceiverId(chatMessage.getSenderId());
-            messageQueueProducer.send(MessageQueueEnum.QUEUE_CHATTING_LIST_REFRESH, refreshChattingMessage);
-        }
         if (RoleCodeEnum.ROBOT.getCode().equals(accountInfo.getRoleCode())) {
             /**
              * 机器人AGENT对话，走AGENT工作流处理
@@ -146,7 +131,12 @@ public class ChatWebSocketEndPointServletHandler extends TextWebSocketHandler {
             return Collections.emptyMap();
         }
         Map<String, WebSocketSession> userSessionMap = new ConcurrentHashMap<>();
-        userIds.forEach(uid -> userSessionMap.put(uid, userSessions.get(uid)));
+        userIds.forEach(uid -> {
+            WebSocketSession ws = userSessions.get(uid);
+            if (ws != null) {
+                userSessionMap.put(uid, ws);
+            }
+        });
         return userSessionMap;
     }
 }

webchat-domain/src/main/java/com/webchat/domain/vo/request/mess/ChatMessageRequestVO.java

@@ -3,7 +3,7 @@ package com.webchat.domain.vo.request.mess;
 import lombok.Data;
 
 
-/***
+/**
  * 聊天消息发送VO
  */
 @Data

webchat-domain/src/main/java/com/webchat/domain/vo/request/mess/MessageBaseVO.java

@@ -1,6 +1,7 @@
 package com.webchat.domain.vo.request.mess;
 
 import com.webchat.domain.dto.queue.BaseQueueDTO;
+import com.webchat.domain.vo.response.UserBaseResponseInfoVO;
 import lombok.Data;
 
 /**
@@ -24,6 +25,8 @@ public class MessageBaseVO extends BaseQueueDTO {
      */
     private String proxySenderId;
 
+    private UserBaseResponseInfoVO proxySender;
+
     /**
      * 消息文本正文
      */

webchat-domain/src/main/java/com/webchat/domain/vo/response/chatting/ChattingListResponseVO.java

@@ -10,6 +10,11 @@ import lombok.Data;
 public class ChattingListResponseVO {
 
     /**
+     * 是否未读
+     */
+    private boolean unread;
+
+    /**
      * 最新对话时间，默认13位时间戳
      */
     private Long lastChatTime;

webchat-remote/src/main/java/com/webchat/rmi/ugc/ChatMessageClient.java

@@ -4,6 +4,7 @@ package com.webchat.rmi.ugc;
 import com.webchat.common.bean.APIResponseBean;
 import com.webchat.domain.vo.request.ChattingRequestVO;
 import com.webchat.domain.vo.response.chatting.ChattingListResponseVO;
+import com.webchat.domain.vo.response.mess.ChatMessageResponseVO;
 import org.springframework.cloud.openfeign.FeignClient;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
@@ -25,7 +26,7 @@ public interface ChatMessageClient {
      *
      * @param userId        当前登录用户uid
      * @param lastChatTime  开始查询时间
-     * @param size          最大获取对话列表条数
+     * @param size          最大获取对话列表条数, 默认取100条
      * @return
      */
     @GetMapping("/ugc-service/chat/message/chatting/list/{userId}")
@@ -44,4 +45,21 @@ public interface ChatMessageClient {
      */
     @PostMapping("/ugc-service/chat/message/chatting/list/add")
     APIResponseBean<Boolean> addChattingList(@RequestBody ChattingRequestVO chattingRequest);
+
+    /**
+     * 查询与用户的对话列表
+     *
+     * @param sourceAccount
+     * @param targetAccount
+     * @param size
+     * @param lastId
+     * @param fixedMessageId
+     * @return
+     */
+    @GetMapping("/ugc-service/chat/message/list")
+    APIResponseBean<List<ChatMessageResponseVO>> list(@RequestParam(value = "sourceAccount", required = false) String sourceAccount,
+                                                      @RequestParam(value = "targetAccount", required = false) String targetAccount,
+                                                      @RequestParam(value = "size", required = false, defaultValue = "50") Integer size,
+                                                      @RequestParam(value = "lastId", required = false) Long lastId,
+                                                      @RequestParam(value = "fixedMessageId", required = false) Long fixedMessageId);
 }

webchat-remote/src/main/java/com/webchat/rmi/user/UserServiceClient.java

@@ -37,6 +37,15 @@ public interface UserServiceClient {
     APIResponseBean<UserBaseResponseInfoVO> userInfo(@PathVariable String userId);
 
     /**
+     * 获取群聊下所有用户id
+     *
+     * @param groupAccount
+     * @return
+     */
+    @GetMapping("/user-service/group/userIds/{groupAccount}")
+    APIResponseBean<Set<String>> getGroupUserIds(@PathVariable String groupAccount);
+
+    /**
      * 根据手机号查询用户基础信息
      *
      * @param mobile

webchat-ugc/src/main/java/com/webchat/ugc/WebchatUGCApplication.java

@@ -1,6 +1,7 @@
 package com.webchat.ugc;
 
 import com.webchat.common.util.SpringContextUtil;
+import com.webchat.ugc.messaegqueue.consumer.redis.PersistentMessageRedisMQConsumer;
 import com.webchat.ugc.messaegqueue.consumer.redis.RefreshChattingRedisMQConsumer;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -18,7 +19,11 @@ public class WebchatUGCApplication {
 
         SpringApplication.run(WebchatUGCApplication.class, args);
 
+        /**
+         * 启动Redis MQ
+         */
         SpringContextUtil.getBean(RefreshChattingRedisMQConsumer.class).initBean();
+        SpringContextUtil.getBean(PersistentMessageRedisMQConsumer.class).initBean();
     }
 
 }

webchat-ugc/src/main/java/com/webchat/ugc/controller/ChatMessageController.java

@@ -3,8 +3,11 @@ package com.webchat.ugc.controller;
 import com.webchat.common.bean.APIResponseBean;
 import com.webchat.common.bean.APIResponseBeanUtil;
 import com.webchat.domain.vo.request.ChattingRequestVO;
+import com.webchat.domain.vo.request.mess.ChatMessageRequestVO;
 import com.webchat.domain.vo.response.chatting.ChattingListResponseVO;
+import com.webchat.domain.vo.response.mess.ChatMessageResponseVO;
 import com.webchat.rmi.ugc.ChatMessageClient;
+import com.webchat.ugc.messaegqueue.service.PersistentMessageService;
 import com.webchat.ugc.service.ChatMessageService;
 import jakarta.validation.Valid;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -22,6 +25,9 @@ public class ChatMessageController implements ChatMessageClient {
     @Autowired
     private ChatMessageService chatMessageService;
 
+    @Autowired
+    private PersistentMessageService persistentMessageService;
+
     /**
      * 查询用户对话列表
      *
@@ -42,4 +48,10 @@ public class ChatMessageController implements ChatMessageClient {
 
         return APIResponseBeanUtil.success(chatMessageService.addChattingList(chattingRequest));
     }
+
+    @Override
+    public APIResponseBean<List<ChatMessageResponseVO>> list(String sourceAccount, String targetAccount, Integer size, Long lastId, Long fixedMessageId) {
+        List<ChatMessageResponseVO> chatMessages= persistentMessageService.getChatMessListFromCache(sourceAccount, targetAccount, lastId, fixedMessageId, size);
+        return APIResponseBeanUtil.success(chatMessages);
+    }
 }

webchat-ugc/src/main/java/com/webchat/ugc/messaegqueue/consumer/redis/PersistentMessageRedisMQConsumer.java

@@ -0,0 +1,62 @@
+package com.webchat.ugc.messaegqueue.consumer.redis;
+
+import com.webchat.common.enums.messagequeue.MessageQueueEnum;
+import com.webchat.common.service.messagequeue.consumer.AbstractRedisQueueConsumer;
+import com.webchat.common.util.JsonUtil;
+import com.webchat.domain.vo.request.mess.ChatMessageRequestVO;
+import com.webchat.ugc.messaegqueue.service.PersistentMessageService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.stereotype.Component;
+import org.springframework.util.Assert;
+
+/**
+ * 消息持久化队列消费者
+ *
+ */
+@Component
+@Lazy(value = false)
+@Slf4j
+public class PersistentMessageRedisMQConsumer extends AbstractRedisQueueConsumer<ChatMessageRequestVO> {
+
+    @Autowired
+    private PersistentMessageService persistentMessageService;
+
+
+    public void initBean() {
+        new Thread(new Runnable() {
+            @Override
+            public void run() {
+                PersistentMessageRedisMQConsumer.this.schedule();
+            }
+        }).start();
+    }
+
+    @Override
+    protected ChatMessageRequestVO convert(String s) {
+
+        return JsonUtil.fromJson(s, ChatMessageRequestVO.class);
+    }
+
+    @Override
+    protected MessageQueueEnum getMessageQueue() {
+
+        return MessageQueueEnum.QUEUE_PERSISTENT_MESSAGE;
+    }
+
+    @Override
+    protected void receive(ChatMessageRequestVO data) {
+        if (data == null) {
+            return;
+        }
+        boolean result = persistentMessageService.persistent(data);
+        // 抛出异常，重新入队
+        Assert.isTrue(result, "消息持久化失败");
+    }
+
+    @Override
+    protected void error(ChatMessageRequestVO data, Exception ex) {
+
+    }
+}

webchat-ugc/src/main/java/com/webchat/ugc/messaegqueue/consumer/rocketmq/PersistentMessageRocketMQConsumer.java

@@ -0,0 +1,39 @@
+package com.webchat.ugc.messaegqueue.consumer.rocketmq;
+
+import com.webchat.common.util.JsonUtil;
+import com.webchat.domain.vo.request.mess.ChatMessageRequestVO;
+import com.webchat.ugc.messaegqueue.service.PersistentMessageService;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.util.Assert;
+
+
+/**
+ * 消息持久化RocketMQ队列
+ *
+ */
+@Component
+@RocketMQMessageListener(consumerGroup = "web_chat", topic = "queue_persistent_message")
+public class PersistentMessageRocketMQConsumer implements RocketMQListener<String> {
+
+    @Autowired
+    private PersistentMessageService persistentMessageService;
+
+    /**
+     * 消息持久化
+     * @param message
+     */
+    @Override
+    public void onMessage(String message) {
+        System.out.println("Received message: " + message);
+        ChatMessageRequestVO data = JsonUtil.fromJson(message, ChatMessageRequestVO.class);
+        if (data == null) {
+            return;
+        }
+        boolean result = persistentMessageService.persistent(data);
+        // 抛出异常，重新入队
+        Assert.isTrue(result, "消息持久化失败");
+    }
+}

webchat-ugc/src/main/java/com/webchat/ugc/messaegqueue/service/PersistentMessageService.java

@@ -0,0 +1,366 @@
+package com.webchat.ugc.messaegqueue.service;
+
+
+import com.webchat.common.bean.APIPageResponseBean;
+import com.webchat.common.enums.ChatMessageTypeEnum;
+import com.webchat.common.enums.RedisKeyEnum;
+import com.webchat.common.enums.RoleCodeEnum;
+import com.webchat.common.enums.messagequeue.MessageQueueEnum;
+import com.webchat.common.service.RedisService;
+import com.webchat.common.service.messagequeue.producer.MessageQueueProducer;
+import com.webchat.common.util.DateUtils;
+import com.webchat.common.util.HtmlUtil;
+import com.webchat.common.util.JsonUtil;
+import com.webchat.domain.vo.request.mess.ChatMessageRequestVO;
+import com.webchat.domain.vo.response.UserBaseResponseInfoVO;
+import com.webchat.domain.vo.response.mess.ChatMessageResponseVO;
+import com.webchat.ugc.repository.dao.IChatMessageDAO;
+import com.webchat.ugc.repository.entity.ChatMessageEntity;
+import com.webchat.ugc.service.AccountService;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
+import org.springframework.stereotype.Service;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+@Slf4j
+@Service
+public class PersistentMessageService {
+
+
+    @Autowired
+    private IChatMessageDAO chatMessageDAO;
+
+    @Autowired
+    private RedisService redisService;
+
+    @Autowired
+    private AccountService accountService;
+
+    @Autowired
+    private MessageQueueProducer<Object, Long> messageQueueProducer;
+
+    /**
+     * 《离线场景》持久化消息队列，保存离线消息，同时会将数据同步到ES用于后续的RAG问答和消息搜索
+     *
+     * @param messVo
+     * @return
+     */
+    public boolean persistent(ChatMessageRequestVO messVo) {
+
+        ChatMessageEntity mess = convert(messVo);
+        // 取消息接收人信息，判断是否群聊场景
+        String receiverId = messVo.getReceiverId();
+        UserBaseResponseInfoVO receiver = accountService.accountInfo(receiverId);
+        boolean groupMessage = RoleCodeEnum.GROUP.getCode().equals(receiver.getRoleCode());
+        List<String> receivers = new ArrayList<>();
+        if (groupMessage) {
+            // 角色翻转，已群组作为发送人
+            mess.setSender(receiverId);
+            // 设置代理消息发送用户
+            mess.setProxySender(messVo.getSenderId());
+            // 查询实际接收人，群组下所有用户
+            Set<String> groupUserIds = accountService.getGroupUserIds(receiverId);
+            receivers.addAll(groupUserIds);
+        } else {
+            receivers.add(receiverId);
+        }
+        receivers.remove(messVo.getSenderId());
+        for (String receiverUserId : receivers) {
+            mess.setReceiver(receiverUserId);
+            // 这里有优化空间，可以改为批量一次入库（先简单实现功能）
+            mess = chatMessageDAO.save(mess);
+            // 加入聊天缓存
+            this.addUserMessCache(mess);
+            // 加入聊天列表
+            if (groupMessage) {
+                this.addOrRefreshMessListCache(mess.getProxySender(), messVo.getReceiverId());
+                this.addOrRefreshMessListCache(receiverUserId, messVo.getReceiverId());
+                // 群聊未读消息+1
+                addUnreadMessCountCache(receiverUserId, mess.getSender());
+            } else {
+                this.addOrRefreshMessListCache(messVo.getSenderId(), receiverUserId);
+                this.addOrRefreshMessListCache(receiverUserId, messVo.getSenderId());
+                // 未读消息+1
+                addUnreadMessCountCache(receiverUserId, messVo.getSenderId());
+            }
+        }
+
+        /**
+         * 刷新用户对话列表
+         *
+         */
+        this.refreshChattingCache(messVo);
+        return true;
+    }
+
+
+    private void refreshChattingCache(ChatMessageRequestVO chatMessage) {
+        // 刷新聊天对话列表
+        Long msgTime = System.currentTimeMillis();
+        ChatMessageRequestVO chattingRequest = new ChatMessageRequestVO();
+        chattingRequest.setSenderId(chatMessage.getSenderId());
+        chattingRequest.setReceiverId(chatMessage.getReceiverId());
+        chattingRequest.setTime(msgTime);
+        messageQueueProducer.send(MessageQueueEnum.QUEUE_CHATTING_LIST_REFRESH, chattingRequest);
+        UserBaseResponseInfoVO accountInfo = accountService.accountInfo(chatMessage.getReceiverId());
+        if (RoleCodeEnum.isUserRole(accountInfo.getRoleCode())) {
+            // 如果消息接受账号是用户类型，也需要刷新用户账号列表
+            chattingRequest.setSenderId(chatMessage.getReceiverId());
+            chattingRequest.setReceiverId(chatMessage.getSenderId());
+            messageQueueProducer.send(MessageQueueEnum.QUEUE_CHATTING_LIST_REFRESH, chattingRequest);
+        } else if (RoleCodeEnum.GROUP.getCode().equals(accountInfo.getRoleCode())) {
+            // 接受人是群组，需要刷新群组下所有用户对话列表
+            Set<String> groupUsers = accountService.getGroupUserIds(chatMessage.getReceiverId());
+            if (CollectionUtils.isNotEmpty(groupUsers)) {
+                for (String groupUserId : groupUsers) {
+                    if (groupUserId.equals(chatMessage.getSenderId())) {
+                        continue;
+                    }
+                    chattingRequest.setSenderId(groupUserId);
+                    chattingRequest.setReceiverId(chatMessage.getReceiverId());
+                    messageQueueProducer.send(MessageQueueEnum.QUEUE_CHATTING_LIST_REFRESH, chattingRequest);
+                }
+            }
+        }
+    }
+
+    /**
+     * 刷新未读消息数缓存，用于新消息红点🔴通知
+     *
+     * @param currUserId
+     * @param chatUserId
+     * @return
+     */
+    private Long addUnreadMessCountCache(String currUserId, String chatUserId) {
+        String unreadUserCacheCountKey = RedisKeyEnum.UN_READ_MESS_USER_SET_CACHE.getKey(currUserId);
+        redisService.sadd(unreadUserCacheCountKey, chatUserId);
+        String unreadCacheCountKey = RedisKeyEnum.UN_READ_MESS_COUNT_CACHE.getKey(currUserId, chatUserId);
+        return redisService.increx(unreadCacheCountKey, RedisKeyEnum.UN_READ_MESS_COUNT_CACHE.getExpireTime());
+    }
+
+    private void clearUnreadMessCountCache(String currUserId, String chatUserId) {
+        String unreadCacheCountKey = RedisKeyEnum.UN_READ_MESS_COUNT_CACHE.getKey(currUserId, chatUserId);
+        redisService.set(unreadCacheCountKey, "0", RedisKeyEnum.UN_READ_MESS_COUNT_CACHE.getExpireTime());
+        String unreadUserCacheCountKey = RedisKeyEnum.UN_READ_MESS_USER_SET_CACHE.getKey(currUserId);
+        redisService.sremove(unreadUserCacheCountKey, chatUserId);
+    }
+
+    public Long getUnreadMessUserCountFromCache(String currUserId) {
+        String unreadUserCacheCountKey = RedisKeyEnum.UN_READ_MESS_USER_SET_CACHE.getKey(currUserId);
+        return redisService.ssize(unreadUserCacheCountKey);
+    }
+
+    public Set<String> getUnreadMessUserSetFromCache(String currUserId) {
+        String unreadUserCacheCountKey = RedisKeyEnum.UN_READ_MESS_USER_SET_CACHE.getKey(currUserId);
+        return redisService.smembers(unreadUserCacheCountKey);
+    }
+
+    /***
+     * 查询两个人的聊天记录
+     * @param currUserId
+     * @param chatUserId
+     * @param lastId
+     * @param size
+     * @return
+     */
+    public List<ChatMessageResponseVO> getChatMessListFromCache(String currUserId, String chatUserId, Long lastId,
+                                                                Long fixedMessageId, int size) {
+
+        // 查询后清理未读消息数
+        clearUnreadMessCountCache(currUserId, chatUserId);
+
+        lastId = lastId == null ? Long.MAX_VALUE : lastId;
+        String cacheKey = getUserMessRedisKey(currUserId, chatUserId);
+        Set<String> cacheSet = redisService.zreverseRangeByScore(cacheKey, lastId, 0, size);
+        if (CollectionUtils.isEmpty(cacheSet)) {
+            return Collections.emptyList();
+        }
+        if (fixedMessageId != null) {
+            cacheSet.remove(fixedMessageId.toString());
+        }
+        List<ChatMessageResponseVO> chatMessageResponseVOList = cacheSet.stream().map(cache -> {
+            ChatMessageResponseVO messageResponse = getChatMessDetailFromCache(Long.valueOf(cache));
+            messageResponse.setReceiver(accountService.accountInfo(messageResponse.getReceiverId()));
+            messageResponse.setSender(accountService.accountInfo(messageResponse.getSenderId()));
+            if (StringUtils.isNotBlank(messageResponse.getProxySenderId())) {
+                messageResponse.setProxySender(accountService.accountInfo(messageResponse.getProxySenderId()));
+            }
+            return messageResponse;
+        }).sorted(Comparator.comparing(ChatMessageResponseVO::getMessId)).collect(Collectors.toList());
+
+        if (fixedMessageId != null) {
+            // 处理消息定位，简单处理，默认插入到消息尾部
+            ChatMessageResponseVO fixedMessage = getChatMessDetailFromCache(fixedMessageId);
+            fixedMessage.setReceiver(accountService.accountInfo(fixedMessage.getReceiverId()));
+            fixedMessage.setSender(accountService.accountInfo(fixedMessage.getSenderId()));
+            if (StringUtils.isNotBlank(fixedMessage.getProxySenderId())) {
+                fixedMessage.setProxySender(accountService.accountInfo(fixedMessage.getProxySenderId()));
+            }
+            chatMessageResponseVOList.add(fixedMessage);
+        }
+
+        return chatMessageResponseVOList;
+    }
+    public Map<String, ChatMessageResponseVO> batchGetUserLastMess(String currUserId, Set<String> userIds) {
+        Map<String, ChatMessageResponseVO> map = new HashMap<>();
+        for (String userId : userIds) {
+            String cacheKey = getUserMessRedisKey(currUserId, userId);
+            Set<String> cacheSet =
+                    redisService.zreverseRangeByScore(cacheKey, Long.MAX_VALUE, 0, 1);
+            if (CollectionUtils.isNotEmpty(cacheSet)) {
+                Long lastMessId = Long.valueOf(new ArrayList<>(cacheSet).get(0));
+                ChatMessageResponseVO chatMessageResponseVO = getChatMessDetailFromCache(lastMessId);
+                if (chatMessageResponseVO != null) {
+                    map.put(userId, chatMessageResponseVO);
+                }
+            }
+        }
+        return map;
+    }
+
+    private void addOrRefreshMessListCache(String currUserId, String chatUserId) {
+        String messUserKey = RedisKeyEnum.MESS_USER_LIST_KEY.getKey(currUserId);
+        redisService.zadd(messUserKey, chatUserId, DateUtils.getCurrentTimeMillis());
+    }
+
+    private void addUserMessCache(ChatMessageEntity mess) {
+        long messId = mess.getId();
+        /***
+         * 刷新消息详情缓存
+         */
+        refreshMessCache(mess);
+        /***
+         * 加入用户消息列表
+         */
+        addUserMessCache(
+                mess.getProxySender() != null ? mess.getProxySender() : mess.getSender(),
+                mess.getProxySender() != null ? mess.getSender() : mess.getReceiver(),
+                messId, messId);
+        addUserMessCache(mess.getReceiver(), mess.getSender(), messId, messId);
+    }
+
+    private void addUserMessCache(String sender, String receiver, Long messId, Long score) {
+        String cacheKey = getUserMessRedisKey(sender, receiver);
+        redisService.zadd(cacheKey, messId.toString(), score);
+    }
+
+    /***
+     * 刷新消息缓存
+     * @param mess
+     */
+    private void refreshMessCache(ChatMessageEntity mess) {
+        String messKey = RedisKeyEnum.MESS_DETAIL_CACHE_KEY.getKey();
+        redisService.hset(messKey, String.valueOf(mess.getId()), JsonUtil.toJsonString(mess),
+                RedisKeyEnum.MESS_DETAIL_CACHE_KEY.getExpireTime());
+    }
+
+    private ChatMessageResponseVO getChatMessDetailFromCache(Long messId) {
+        String messKey = RedisKeyEnum.MESS_DETAIL_CACHE_KEY.getKey();
+        String messCache = redisService.hget(messKey, String.valueOf(messId));
+        if (StringUtils.isBlank(messCache)) {
+            return null;
+        }
+        ChatMessageEntity userMessEntity = JsonUtil.fromJson(messCache, ChatMessageEntity.class);
+        ChatMessageResponseVO messageResponse = new ChatMessageResponseVO();
+        messageResponse.setMessId(userMessEntity.getId());
+        messageResponse.setMessage(userMessEntity.getMessage());
+        messageResponse.setSenderId(userMessEntity.getSender());
+        messageResponse.setProxySenderId(userMessEntity.getProxySender());
+        messageResponse.setReceiverId(userMessEntity.getReceiver());
+        messageResponse.setTime(userMessEntity.getSendDate().getTime());
+        messageResponse.setIsRead(userMessEntity.getIsRead());
+        messageResponse.setType(userMessEntity.getType());
+        if (ChatMessageTypeEnum.RED_PACKET.getType().equals(userMessEntity.getType())) {
+//            messageResponse.setRedPacketDetail(redPacketService.getRedPacketDetailCache(Long.valueOf(userMessEntity.getMessage())));
+        } else if (ChatMessageTypeEnum.PUBLIC_ACCOUNT_ARTICLE.getType().equals(userMessEntity.getType())) {
+//            messageResponse.setPublicAccountArticle(articleService.getPublicAccountArticleMessage(Long.valueOf(userMessEntity.getMessage())));
+        }
+        messageResponse.setGroupMessage(StringUtils.isNotBlank(userMessEntity.getProxySender()));
+        return messageResponse;
+    }
+
+    public String getUserMessRedisKey(String sender, String receiver) {
+        return RedisKeyEnum.USER_CHAT_MESS_CACHE_KEY.getKey(sender, receiver);
+    }
+
+    public APIPageResponseBean<List<ChatMessageResponseVO>> pageMessage(String mess, int pageNo, int pageSize) {
+        Pageable pageable = PageRequest.of(pageNo - 1, pageSize, Sort.by(Sort.Order.desc("id")));
+        Page<ChatMessageEntity> chatMessEntityPage;
+        if (StringUtils.isBlank(mess)) {
+            chatMessEntityPage = chatMessageDAO.findAll(pageable);
+        } else {
+            chatMessEntityPage = chatMessageDAO.findAllByMessageLike("%"+mess+"%", pageable);
+        }
+        List<ChatMessageResponseVO> chatMessageResponseVOList = convertChatMessageResponseList(chatMessEntityPage.getContent());
+        return APIPageResponseBean.success(pageNo, pageSize, chatMessEntityPage.getTotalElements(), chatMessageResponseVOList);
+    }
+
+    private List<ChatMessageResponseVO> convertChatMessageResponseList(List<ChatMessageEntity> chatMessEntities) {
+        if (CollectionUtils.isEmpty(chatMessEntities)) {
+            return Collections.emptyList();
+        }
+        /**
+         * 批量查询用户信息
+         */
+        Set<String> senderUserIds = chatMessEntities.stream().map(ChatMessageEntity::getSender).collect(Collectors.toSet());
+        Set<String> proxySenderUserIds = chatMessEntities.stream().map(ChatMessageEntity::getProxySender).filter(Objects::nonNull).collect(Collectors.toSet());
+        Set<String> receiverUserIds = chatMessEntities.stream().map(ChatMessageEntity::getReceiver).collect(Collectors.toSet());
+        senderUserIds.addAll(receiverUserIds);
+        if (CollectionUtils.isNotEmpty(proxySenderUserIds)) {
+            senderUserIds.addAll(proxySenderUserIds);
+        }
+        Map<String, UserBaseResponseInfoVO> userMap = accountService.batchGet(senderUserIds);
+        return chatMessEntities.stream().map(chat -> {
+            ChatMessageResponseVO chatMessageResponse = new ChatMessageResponseVO();
+            chatMessageResponse.setTime(chat.getSendDate().getTime());
+            chatMessageResponse.setSender(userMap.get(chat.getSender()));
+            if (StringUtils.isNotBlank(chat.getProxySender())) {
+                chatMessageResponse.setProxySender(userMap.get(chat.getProxySender()));
+            }
+            chatMessageResponse.setReceiver(userMap.get(chat.getReceiver()));
+            chatMessageResponse.setMessage(chat.getMessage());
+            return chatMessageResponse;
+        }).collect(Collectors.toList());
+    }
+
+    private ChatMessageEntity convert(ChatMessageRequestVO messVo) {
+        ChatMessageEntity mess = new ChatMessageEntity();
+        mess.setSender(messVo.getSenderId());
+        mess.setReceiver(messVo.getReceiverId());
+        mess.setMessage(this.handleSpecialHtmlTag(HtmlUtil.xssEscape(messVo.getMessage())));
+        mess.setSendDate(new Date());
+        mess.setIsRead(false);
+        mess.setType(messVo.getType());
+        return mess;
+    }
+
+    /***
+     * 处理特殊字符
+     * @param content
+     * @return
+     */
+    private String handleSpecialHtmlTag(String content) {
+        if (StringUtils.isBlank(content)) {
+            return content;
+        }
+        content = content.replaceAll("&lt;br&gt;", "<br>");
+        content = content.replaceAll("&lt;b&gt;", "<b>");
+        return content;
+    }
+}

webchat-ugc/src/main/java/com/webchat/ugc/repository/dao/IChatMessageDAO.java

@@ -0,0 +1,23 @@
+package com.webchat.ugc.repository.dao;
+
+import com.webchat.ugc.repository.entity.ChatMessageEntity;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.stereotype.Repository;
+
+import java.util.List;
+
+
+@Repository
+public interface IChatMessageDAO extends JpaSpecificationExecutor<ChatMessageEntity>, JpaRepository<ChatMessageEntity, Long> {
+
+    @Query(value = "select distinct mess from ChatMessageEntity mess where "
+            + "(mess.sender = :sender or mess.receiver = :receiver) and "
+            + "(mess.receiver = :sender or mess.sender = :receiver) order by mess.id desc")
+    List<ChatMessageEntity> findAllBySenderAndReceiver(String sender, String receiver);
+
+    Page<ChatMessageEntity> findAllByMessageLike(String message, Pageable pageable);
+}

webchat-ugc/src/main/java/com/webchat/ugc/repository/entity/BaseEntity.java

@@ -0,0 +1,28 @@
+package com.webchat.ugc.repository.entity;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.MappedSuperclass;
+import jakarta.persistence.PreUpdate;
+import lombok.Data;
+
+import java.io.Serializable;
+import java.util.Date;
+
+/**
+ * 基础实体类
+ */
+@MappedSuperclass
+@Data
+public class BaseEntity extends SimpleBaseEntity implements Serializable {
+
+    @Column(name = "update_by")
+    private String updateBy;
+
+    @Column(name = "update_date")
+    private Date updateDate;
+
+    @PreUpdate
+    public void preUpdate() {
+        this.updateDate = new Date();
+    }
+}

webchat-ugc/src/main/java/com/webchat/ugc/repository/entity/ChatMessageEntity.java

@@ -0,0 +1,87 @@
+package com.webchat.ugc.repository.entity;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.GenerationType;
+import jakarta.persistence.Id;
+import jakarta.persistence.PrePersist;
+import jakarta.persistence.PreUpdate;
+import jakarta.persistence.Table;
+import jakarta.persistence.Version;
+import lombok.Data;
+import java.util.Date;
+
+@Data
+@Entity
+@Table(name = "web_chat_message")
+public class ChatMessageEntity {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    protected Long id;
+
+    /**
+     * 发送人
+     */
+    @Column(name = "sender")
+    private String sender;
+
+    /**
+     * 代理发送人（群聊真实消息发送人）点对点消息，proxySender为空
+     */
+    @Column(name = "proxy_sender")
+    private String proxySender;
+
+    /**
+     * 接收人
+     */
+    @Column(name = "receiver")
+    private String receiver;
+
+    /**
+     * 消息类型
+     *
+     * @see com.webchat.common.enums.ChatMessageTypeEnum
+     */
+    @Column(name = "type")
+    private Integer type;
+
+    /**
+     * 正文
+     */
+    @Column(name = "message")
+    private String message;
+
+    /**
+     * 状态
+     */
+    @Column(name = "is_read")
+    private Boolean isRead;
+
+    /**
+     * 发送时间
+     */
+    @Column(name = "send_date")
+    private Date sendDate;
+
+    @Column(name = "update_date")
+    private Date updateDate;
+
+    @Version
+    @Column(name = "version")
+    private Integer version;
+
+    @PrePersist
+    public void prePersist() {
+        Date now = new Date();
+        if (this.sendDate == null) {
+            this.sendDate = now;
+        }
+    }
+
+    @PreUpdate
+    public void preUpdate() {
+        this.updateDate = new Date();
+    }
+}

webchat-ugc/src/main/java/com/webchat/ugc/repository/entity/SimpleBaseEntity.java

@@ -0,0 +1,36 @@
+package com.webchat.ugc.repository.entity;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.MappedSuperclass;
+import jakarta.persistence.PrePersist;
+import jakarta.persistence.Version;
+import lombok.Data;
+
+import java.io.Serializable;
+import java.util.Date;
+
+/**
+ * 基础实体类
+ */
+@MappedSuperclass
+@Data
+public class SimpleBaseEntity implements Serializable {
+
+    @Column(name = "create_by")
+    private String createBy;
+
+    @Column(name = "create_date")
+    private Date createDate;
+
+    @Version
+    @Column(name = "version")
+    private Integer version;
+
+    @PrePersist
+    public void prePersist() {
+        Date now = new Date();
+        if (this.createDate == null) {
+            this.createDate = now;
+        }
+    }
+}

webchat-ugc/src/main/java/com/webchat/ugc/service/AccountService.java

@@ -11,6 +11,7 @@ import org.springframework.stereotype.Service;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 @Service
 public class AccountService {
@@ -19,6 +20,22 @@ public class AccountService {
     private UserServiceClient userServiceClient;
 
     /**
+     * 查询当前账号信息
+     *
+     * @param account
+     * @return
+     */
+
+    public UserBaseResponseInfoVO accountInfo(String account) {
+
+        APIResponseBean<UserBaseResponseInfoVO> responseBean = userServiceClient.userInfo(account);
+        if (APIResponseBeanUtil.isOk(responseBean)) {
+            return responseBean.getData();
+        }
+        return null;
+    }
+
+    /**
      *  批量查询账号详情数据
      *
      * @param accounts
@@ -35,4 +52,30 @@ public class AccountService {
         }
         throw new RuntimeException(responseBean.getMsg());
     }
+
+    public Map<String, UserBaseResponseInfoVO> batchGet(Set<String> accounts) {
+        if (CollectionUtils.isEmpty(accounts)) {
+            return Collections.emptyMap();
+        }
+        APIResponseBean<Map<String, UserBaseResponseInfoVO>> responseBean =
+                userServiceClient.batchGet(accounts);
+        if (APIResponseBeanUtil.isOk(responseBean)) {
+            return responseBean.getData();
+        }
+        throw new RuntimeException(responseBean.getMsg());
+    }
+
+    /**
+     * 获取群组下的群成员用户id集合
+     *
+     * @param groupAccount
+     * @return
+     */
+    public Set<String> getGroupUserIds(String groupAccount) {
+        APIResponseBean<Set<String>> responseBean = userServiceClient.getGroupUserIds(groupAccount);
+        if (APIResponseBeanUtil.isOk(responseBean)) {
+            return responseBean.getData();
+        }
+        return null;
+    }
 }

webchat-ugc/src/main/java/com/webchat/ugc/service/ChatMessageService.java

@@ -10,7 +10,10 @@ import com.webchat.domain.vo.request.ChattingRequestVO;
 import com.webchat.domain.vo.request.mess.MessageNotifyVO;
 import com.webchat.domain.vo.response.UserBaseResponseInfoVO;
 import com.webchat.domain.vo.response.chatting.ChattingListResponseVO;
+import com.webchat.domain.vo.response.mess.ChatMessageResponseVO;
+import com.webchat.ugc.messaegqueue.service.PersistentMessageService;
 import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections.MapUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
@@ -26,12 +29,13 @@ import java.util.stream.Collectors;
 @Service
 public class ChatMessageService {
 
-
     @Autowired
     private RedisService redisService;
     @Autowired
     private AccountService accountService;
     @Autowired
+    private PersistentMessageService persistentMessageService;
+    @Autowired
     private MessageQueueProducer<MessageNotifyVO, Long> messageQueueProducer;
 
     /**
@@ -60,11 +64,19 @@ public class ChatMessageService {
          */
         Map<String, Long> lastTimeMap = redisService.zscoreTomap(cacheKey, chattingUserList);
         /**
+         * 查询未读消息的用户列表，用于标记小红点
+         */
+        Set<String> unreadUsers = persistentMessageService.getUnreadMessUserSetFromCache(userId);
+        /**
          * 查询对话列表最新对话消息内容数据
-         * TODO
          */
+        Map<String, ChatMessageResponseVO> lastMessageVOMap = persistentMessageService.batchGetUserLastMess(userId, chattingUsers);
         Map<String, String> lastMessageMap = new HashMap<>();
-
+        if (MapUtils.isNotEmpty(lastMessageVOMap)) {
+            for (Map.Entry<String, ChatMessageResponseVO> entry : lastMessageVOMap.entrySet()) {
+                lastMessageMap.put(entry.getKey(), entry.getValue() != null ? entry.getValue().getPrintMessage() : "有新消息");
+            }
+        }
         /**
          * 构造用户对话列表
          */
@@ -73,6 +85,7 @@ public class ChatMessageService {
                 .map(u -> {
                     return ChattingListResponseVO.builder()
                                 .user(u)
+                                .unread(unreadUsers.contains(u.getUserId()))
                                 .lastChatTime(lastTimeMap.get(u.getUserId()))
                                 .lastOfflineMessage(lastMessageMap.get(u.getUserId()))
                                 .build();

webchat-user/src/main/java/com/webchat/user/controller/UserServiceController.java

@@ -38,6 +38,12 @@ public class UserServiceController implements UserServiceClient {
     }
 
     @Override
+    public APIResponseBean<Set<String>> getGroupUserIds(String groupAccount) {
+
+        return APIResponseBeanUtil.success(userService.getGroupUserIds(groupAccount));
+    }
+
+    @Override
     public APIResponseBean<UserBaseResponseVO> userBaseInfo(String mobile) {
 
         return APIResponseBeanUtil.success(userService.getUserBaseByMobile(mobile));

webchat-user/src/main/java/com/webchat/user/service/UserService.java

@@ -25,6 +25,7 @@ import com.webchat.user.repository.dao.IUserDAO;
 import com.webchat.user.repository.entity.GroupUserEntity;
 import com.webchat.user.repository.entity.UserEntity;
 import com.webchat.user.service.relation.User2FileSenderAccountRelationService;
+import com.webchat.user.service.relation.User2GroupAccountRelationService;
 import jakarta.persistence.criteria.Predicate;
 import jakarta.servlet.http.HttpServletRequest;
 import jakarta.servlet.http.HttpServletResponse;
@@ -593,6 +594,17 @@ public class UserService {
     }
 
     /**
+     * 查询群聊下的所有用户
+     *
+     * @param groupAccount
+     * @return
+     */
+    public Set<String> getGroupUserIds(String groupAccount) {
+        return SpringContextUtil.getBean(User2GroupAccountRelationService.class)
+                .getAllSubscriber(groupAccount);
+    }
+
+    /**
      * 用户信息编辑
      *
      * @param updateUserInfoRequest

webchat-user/src/main/java/com/webchat/user/service/relation/AbstractAccountRelationService.java

@@ -26,6 +26,7 @@ import org.springframework.util.Assert;
 
 import java.util.ArrayList;
 import java.util.Date;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -216,6 +217,26 @@ public abstract class AbstractAccountRelationService implements AccountRelationV
         return APIPageResponseBean.success(pageNo, pageSize, total, relactionAccountList);
     }
 
+    /**
+     * 查询账号下的所有订阅用户列表
+     *
+     * @param account
+     * @return
+     */
+    @Override
+    public Set<String> getAllSubscriber(String account) {
+        Integer type = this.getRelationType().getType();
+        String prefix = "SAFE";
+        String cacheKey = this.getRelationListRedisKey(account, type);
+        String safeCacheKey = this.getRelationListRedisKey(prefix, account, type);
+        if (!redisService.exists(cacheKey) && !redisService.exists(safeCacheKey)) {
+            // 失效/没有好友关系, 刷新账号管理缓存列表
+            this.initAccountRelationListCache(account, type);
+            redisService.set(safeCacheKey, WebConstant.CACHE_NONE, 5 * 60);
+        }
+        return redisService.zreverseRange(cacheKey, 0, Integer.MAX_VALUE);
+    }
+
 
     /**
      * 添加targetAccount到sourceAccount的好友列表缓存

webchat-user/src/main/java/com/webchat/user/service/relation/AccountRelationWrapper.java

@@ -3,7 +3,7 @@ package com.webchat.user.service.relation;
 import com.webchat.common.bean.APIPageResponseBean;
 import com.webchat.domain.vo.response.UserBaseResponseInfoVO;
 
-import java.util.List;
+import java.util.Set;
 
 public interface AccountRelationWrapper {
 
@@ -36,4 +36,12 @@ public interface AccountRelationWrapper {
      */
     APIPageResponseBean<UserBaseResponseInfoVO> listRelations(String account, Integer pageNo, Integer pageSize);
 
+    /**
+     * 查询订阅了当前账号的所有账号
+     * 如：获取群组下的所有用户、获取公众号的所有订阅用户
+     *
+     * @param account
+     * @return
+     */
+    Set<String> getAllSubscriber(String account);
 }

webchat-user/src/main/java/com/webchat/user/service/relation/User2GroupAccountRelationService.java

@@ -51,10 +51,11 @@ public class User2GroupAccountRelationService extends AbstractAccountRelationSer
     protected void doAfterComplete(Long id, String sourceAccount, String targetAccount, boolean subscribe) {
         // 添加群聊到用户账号关系列表
         if (subscribe) {
-            // 文件传输助手，用户注册后默认订阅，无需审核
-            // 添加文件传输助手到用户好友列表缓存
+            super.init(targetAccount, sourceAccount);
+            super.addTargetAccountRelationListCache(targetAccount, sourceAccount);
+            super.init(sourceAccount, targetAccount);
             super.addTargetAccountRelationListCache(sourceAccount, targetAccount);
-            // 添加文件传输助手到用户聊天对话列表缓存
+            // 添加群聊到用户聊天对话列表缓存
             super.addTargetAccount2SourceLastChattingList(sourceAccount, targetAccount);
         }
     }