getMaxReadMessageId(@RequestParam("peerId") Long peerId) {
return success(privateMessageService.getMaxReadMessageId(getLoginUserId(), peerId));
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/channel/ImChannelMessagePullRespVO.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/channel/ImChannelMessagePullRespVO.java
index 41bbbf9..8a30fcb 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/channel/ImChannelMessagePullRespVO.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/channel/ImChannelMessagePullRespVO.java
@@ -24,8 +24,8 @@ public class ImChannelMessagePullRespVO {
@Schema(description = "消息内容；payload JSON 快照", requiredMode = Schema.RequiredMode.REQUIRED)
private String content;
- @Schema(description = "当前用户的已读态；按 Redis 游标计算填充", requiredMode = Schema.RequiredMode.REQUIRED, example = "0")
- private Integer status; // 参见 ImMessageStatusEnum 枚举类
+ @Schema(description = "当前用户回执 / 已读态；按 Redis 读位置计算（已读 DONE，未读 PENDING）", requiredMode = Schema.RequiredMode.REQUIRED, example = "2")
+ private Integer receiptStatus; // 参见 ImMessageReceiptStatusEnum 枚举类
@Schema(description = "发送时间", requiredMode = Schema.RequiredMode.REQUIRED)
private LocalDateTime sendTime;
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/group/ImGroupMessageSendReqVO.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/group/ImGroupMessageSendReqVO.java
index f02d919..813e35a 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/group/ImGroupMessageSendReqVO.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/group/ImGroupMessageSendReqVO.java
@@ -1,7 +1,7 @@
package cn.iocoder.yudao.module.im.controller.admin.message.vo.group;
import cn.iocoder.yudao.framework.common.validation.InEnum;
-import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
import com.fasterxml.jackson.annotation.JsonIgnore;
import io.swagger.v3.oas.annotations.media.Schema;
import jakarta.validation.constraints.AssertTrue;
@@ -28,7 +28,7 @@ public class ImGroupMessageSendReqVO {
@Schema(description = "消息类型", requiredMode = Schema.RequiredMode.REQUIRED, example = "0")
@NotNull(message = "消息类型不能为空")
- @InEnum(ImMessageTypeEnum.class)
+ @InEnum(ImContentTypeEnum.class)
private Integer type;
@Schema(description = "消息内容，JSON 格式", requiredMode = Schema.RequiredMode.REQUIRED, example = "{\"content\":\"你好\"}")
@@ -47,7 +47,7 @@ public class ImGroupMessageSendReqVO {
@AssertTrue(message = "消息类型不允许")
@JsonIgnore
public boolean isTypeNormal() {
- return type == null || ImMessageTypeEnum.validate(type).isNormal();
+ return type == null || ImContentTypeEnum.validate(type).isNormal();
}
}
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/privates/ImPrivateMessageRespVO.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/privates/ImPrivateMessageRespVO.java
index 3484b12..9b62689 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/privates/ImPrivateMessageRespVO.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/privates/ImPrivateMessageRespVO.java
@@ -31,7 +31,10 @@ public class ImPrivateMessageRespVO {
private String content;
@Schema(description = "消息状态", example = "0")
- private Integer status;
+ private Integer status; // 参见 ImMessageStatusEnum 枚举类
+
+ @Schema(description = "回执状态", example = "0")
+ private Integer receiptStatus; // 参见 ImMessageReceiptStatusEnum 枚举类
@Schema(description = "发送时间")
private LocalDateTime sendTime;
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/privates/ImPrivateMessageSendReqVO.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/privates/ImPrivateMessageSendReqVO.java
index d6c703f..6552234 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/privates/ImPrivateMessageSendReqVO.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/privates/ImPrivateMessageSendReqVO.java
@@ -1,7 +1,7 @@
package cn.iocoder.yudao.module.im.controller.admin.message.vo.privates;
import cn.iocoder.yudao.framework.common.validation.InEnum;
-import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
import com.fasterxml.jackson.annotation.JsonIgnore;
import io.swagger.v3.oas.annotations.media.Schema;
import jakarta.validation.constraints.AssertTrue;
@@ -26,20 +26,23 @@ public class ImPrivateMessageSendReqVO {
@Schema(description = "消息类型", requiredMode = Schema.RequiredMode.REQUIRED, example = "0")
@NotNull(message = "消息类型不能为空")
- @InEnum(ImMessageTypeEnum.class)
+ @InEnum(ImContentTypeEnum.class)
private Integer type;
@Schema(description = "消息内容，JSON 格式", requiredMode = Schema.RequiredMode.REQUIRED, example = "{\"content\":\"你好\"}")
@NotEmpty(message = "消息内容不能为空")
private String content;
+ @Schema(description = "是否需要回执", example = "false")
+ private Boolean receipt;
+
/**
* 仅允许用户消息（normal）类型
*/
@AssertTrue(message = "消息类型不允许")
@JsonIgnore
public boolean isTypeNormal() {
- return type == null || ImMessageTypeEnum.validate(type).isNormal();
+ return type == null || ImContentTypeEnum.validate(type).isNormal();
}
}
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/conversation/ImConversationReadDO.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/conversation/ImConversationReadDO.java
new file mode 100644
index 0000000..bdd9fa0
--- /dev/null
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/conversation/ImConversationReadDO.java
@@ -0,0 +1,68 @@
+package cn.iocoder.yudao.module.im.dal.dataobject.conversation;
+
+import cn.iocoder.yudao.framework.mybatis.core.dataobject.BaseDO;
+import cn.iocoder.yudao.module.im.dal.dataobject.channel.ImChannelDO;
+import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupDO;
+import cn.iocoder.yudao.module.im.dal.dataobject.message.ImChannelMessageDO;
+import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO;
+import cn.iocoder.yudao.module.im.dal.dataobject.message.ImPrivateMessageDO;
+import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
+import com.baomidou.mybatisplus.annotation.KeySequence;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.*;
+
+import java.time.LocalDateTime;
+
+/**
+ * IM 会话读位置 DO
+ *
+ * 只表达「用户在某个会话的最大已读位置」，私聊 / 群聊 / 频道统一落这张表，是读位置的唯一权威。
+ *
+ * @author 芋道源码
+ */
+@TableName("im_conversation_read")
+@KeySequence("im_conversation_read_seq")
+@Data
+@EqualsAndHashCode(callSuper = true)
+@ToString(callSuper = true)
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class ImConversationReadDO extends BaseDO {
+
+ /**
+ * 编号
+ */
+ @TableId
+ private Long id;
+ /**
+ * 用户编号
+ *
+ * 关联 AdminUserDO 的 id 字段
+ */
+ private Long userId;
+ /**
+ * 会话类型
+ *
+ * 枚举 {@link ImConversationTypeEnum}
+ */
+ private Integer conversationType;
+ /**
+ * 目标编号
+ *
+ * 私聊关联 AdminUserDO 的 id；群聊关联 {@link ImGroupDO#getId()}；频道关联 {@link ImChannelDO#getId()}
+ */
+ private Long targetId;
+ /**
+ * 最大已读消息编号
+ *
+ * 关联 {@link ImPrivateMessageDO#getId()}、{@link ImGroupMessageDO#getId()}、{@link ImChannelMessageDO#getId()}
+ */
+ private Long messageId;
+ /**
+ * 最近已读时间
+ */
+ private LocalDateTime readTime;
+
+}
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/ImChannelMessageDO.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/ImChannelMessageDO.java
index 2572828..d71ee7c 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/ImChannelMessageDO.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/ImChannelMessageDO.java
@@ -4,7 +4,7 @@ import cn.iocoder.yudao.framework.mybatis.core.dataobject.BaseDO;
import cn.iocoder.yudao.framework.mybatis.core.type.LongListTypeHandler;
import cn.iocoder.yudao.module.im.dal.dataobject.channel.ImChannelDO;
import cn.iocoder.yudao.module.im.dal.dataobject.channel.ImChannelMaterialDO;
-import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
import com.baomidou.mybatisplus.annotation.KeySequence;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
@@ -54,7 +54,7 @@ public class ImChannelMessageDO extends BaseDO {
/**
* 消息类型
*
- * 枚举 {@link ImMessageTypeEnum}
+ * 枚举 {@link ImContentTypeEnum}
*/
private Integer type;
/**
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/ImGroupMessageDO.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/ImGroupMessageDO.java
index f6c3d60..bdab8a8 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/ImGroupMessageDO.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/ImGroupMessageDO.java
@@ -3,9 +3,9 @@ package cn.iocoder.yudao.module.im.dal.dataobject.message;
import cn.iocoder.yudao.framework.mybatis.core.dataobject.BaseDO;
import cn.iocoder.yudao.framework.mybatis.core.type.LongListTypeHandler;
import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupDO;
-import cn.iocoder.yudao.module.im.enums.message.ImGroupMessageReceiptStatusEnum;
+import cn.iocoder.yudao.module.im.enums.message.ImMessageReceiptStatusEnum;
import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
-import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
import com.baomidou.mybatisplus.annotation.KeySequence;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
@@ -54,7 +54,7 @@ public class ImGroupMessageDO extends BaseDO {
/**
* 消息类型
*
- * 枚举 {@link ImMessageTypeEnum}
+ * 枚举 {@link ImContentTypeEnum}
*/
private Integer type;
/**
@@ -67,8 +67,6 @@ public class ImGroupMessageDO extends BaseDO {
* 消息状态
*
* 枚举 {@link ImMessageStatusEnum}
- *
- * 为什么没有 READ 状态？与单聊的差异：单聊用 UNREAD/READ 跟踪每条消息状态，群聊用 Redis 存储每个成员的已读位置（游标模型）
*/
private Integer status;
/**
@@ -94,7 +92,7 @@ public class ImGroupMessageDO extends BaseDO {
/**
* 回执状态
*
- * 枚举 {@link ImGroupMessageReceiptStatusEnum}
+ * 枚举 {@link ImMessageReceiptStatusEnum}
*/
private Integer receiptStatus;
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/ImPrivateMessageDO.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/ImPrivateMessageDO.java
index bb87939..808c1ee 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/ImPrivateMessageDO.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/ImPrivateMessageDO.java
@@ -1,8 +1,9 @@
package cn.iocoder.yudao.module.im.dal.dataobject.message;
import cn.iocoder.yudao.framework.mybatis.core.dataobject.BaseDO;
+import cn.iocoder.yudao.module.im.enums.message.ImMessageReceiptStatusEnum;
import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
-import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
import com.baomidou.mybatisplus.annotation.KeySequence;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
@@ -49,7 +50,7 @@ public class ImPrivateMessageDO extends BaseDO {
/**
* 消息类型
*
- * 枚举 {@link ImMessageTypeEnum}
+ * 枚举 {@link ImContentTypeEnum}
*/
private Integer type;
/**
@@ -65,6 +66,12 @@ public class ImPrivateMessageDO extends BaseDO {
*/
private Integer status;
/**
+ * 回执状态
+ *
+ * 枚举 {@link ImMessageReceiptStatusEnum}
+ */
+ private Integer receiptStatus;
+ /**
* 发送时间
*/
private LocalDateTime sendTime;
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/AudioMessage.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/AudioMessage.java
index dd37cc0..2422467 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/AudioMessage.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/AudioMessage.java
@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.im.service.websocket.dto.message;
+package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
import lombok.Data;
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/CardMessage.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/CardMessage.java
index 39bef39..d99ed12 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/CardMessage.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/CardMessage.java
@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.im.service.websocket.dto.message;
+package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
import lombok.Data;
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/FaceMessage.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/FaceMessage.java
index f5c6b6c..6d6a642 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/FaceMessage.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/FaceMessage.java
@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.im.service.websocket.dto.message;
+package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
import lombok.Data;
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/FileMessage.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/FileMessage.java
index 7feb9ff..bd4fcc3 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/FileMessage.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/FileMessage.java
@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.im.service.websocket.dto.message;
+package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
import lombok.Data;
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/ImageMessage.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/ImageMessage.java
index bc8e41e..052342a 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/ImageMessage.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/ImageMessage.java
@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.im.service.websocket.dto.message;
+package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
import lombok.Data;
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/LocationMessage.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/LocationMessage.java
index 8c36d1a..fdbb7c7 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/LocationMessage.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/LocationMessage.java
@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.im.service.websocket.dto.message;
+package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
import lombok.Data;
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/MaterialMessage.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/MaterialMessage.java
index 9ef4308..41fbdce 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/MaterialMessage.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/MaterialMessage.java
@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.im.service.websocket.dto.message;
+package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
import lombok.Data;
import lombok.experimental.Accessors;
@@ -6,7 +6,7 @@ import lombok.experimental.Accessors;
/**
* 频道素材消息 payload
*
- * 对应 {@link cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum#MATERIAL}（type=125）。
+ * 对应 {@link cn.iocoder.yudao.module.im.enums.ImContentTypeEnum#MATERIAL}（type=125）。
* 客户端按本字段集渲染图文卡片；
* 富文本正文不在本 payload 中传递，点击详情时另调 /im/channel/material/get-content?id= 按需拉取。
*/
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/MergeMessage.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/MergeMessage.java
index 62a051d..9bc675b 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/MergeMessage.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/MergeMessage.java
@@ -1,6 +1,6 @@
-package cn.iocoder.yudao.module.im.service.websocket.dto.message;
+package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
-import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
import lombok.Data;
import java.util.List;
@@ -47,7 +47,7 @@ public class MergeMessage {
/**
* 消息类型
*
- * 枚举 {@link ImMessageTypeEnum}
+ * 枚举 {@link ImContentTypeEnum}
*/
private Integer type;
/**
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/QuoteMessage.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/QuoteMessage.java
index 20fa9f3..a06d417 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/QuoteMessage.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/QuoteMessage.java
@@ -1,6 +1,6 @@
-package cn.iocoder.yudao.module.im.service.websocket.dto.message;
+package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
-import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
import lombok.Data;
import lombok.experimental.Accessors;
@@ -39,7 +39,7 @@ public class QuoteMessage {
/**
* 被引用消息类型
*
- * 枚举 {@link ImMessageTypeEnum}
+ * 枚举 {@link ImContentTypeEnum}
*/
private Integer type;
/**
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/RecallMessage.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/RecallMessage.java
index 5df78f2..a9ba575 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/RecallMessage.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/RecallMessage.java
@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.im.service.websocket.dto.message;
+package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
import lombok.Data;
import lombok.experimental.Accessors;
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/TextMessage.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/TextMessage.java
index 9d54b7d..4d8a66c 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/TextMessage.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/TextMessage.java
@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.im.service.websocket.dto.message;
+package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
import lombok.Data;
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/VideoMessage.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/VideoMessage.java
index a5ba53a..5bceef1 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/message/VideoMessage.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/message/content/VideoMessage.java
@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.im.service.websocket.dto.message;
+package cn.iocoder.yudao.module.im.dal.dataobject.message.content;
import lombok.Data;
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/conversation/ImConversationReadMapper.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/conversation/ImConversationReadMapper.java
new file mode 100644
index 0000000..45a702f
--- /dev/null
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/conversation/ImConversationReadMapper.java
@@ -0,0 +1,83 @@
+package cn.iocoder.yudao.module.im.dal.mysql.conversation;
+
+import cn.hutool.core.date.LocalDateTimeUtil;
+import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
+import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
+import cn.iocoder.yudao.module.im.dal.dataobject.conversation.ImConversationReadDO;
+import com.baomidou.mybatisplus.core.toolkit.Wrappers;
+import org.apache.ibatis.annotations.Mapper;
+
+import java.time.LocalDateTime;
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * IM 会话读位置 Mapper
+ *
+ * @author 芋道源码
+ */
+@Mapper
+public interface ImConversationReadMapper extends BaseMapperX {
+
+ default ImConversationReadDO selectByUserIdAndConversation(Long userId, Integer conversationType, Long conversationId) {
+ return selectOne(new LambdaQueryWrapperX()
+ .eq(ImConversationReadDO::getUserId, userId)
+ .eq(ImConversationReadDO::getConversationType, conversationType)
+ .eq(ImConversationReadDO::getTargetId, conversationId));
+ }
+
+ default List selectListByConversation(Integer conversationType, Long conversationId) {
+ return selectList(new LambdaQueryWrapperX()
+ .eq(ImConversationReadDO::getConversationType, conversationType)
+ .eq(ImConversationReadDO::getTargetId, conversationId));
+ }
+
+ default List selectListByUserIdAndConversations(Long userId, Integer conversationType,
+ Collection conversationIds) {
+ return selectList(new LambdaQueryWrapperX()
+ .eq(ImConversationReadDO::getUserId, userId)
+ .eq(ImConversationReadDO::getConversationType, conversationType)
+ .in(ImConversationReadDO::getTargetId, conversationIds));
+ }
+
+ /**
+ * 增量拉取当前用户的会话读位置（按 update_time + id 正向游标）
+ *
+ * @param userId 当前用户编号
+ * @param lastUpdateTime 上次拉取到的更新时间；首次拉取传 null
+ * @param lastId 上次拉取到的记录编号；首次拉取传 null
+ * @param limit 拉取数量
+ * @return 会话读位置列表
+ */
+ default List selectPullListByUserId(Long userId, Long lastUpdateTime, Long lastId, Integer limit) {
+ LambdaQueryWrapperX query = new LambdaQueryWrapperX()
+ .eq(ImConversationReadDO::getUserId, userId);
+ if (lastUpdateTime != null && lastId != null) {
+ LocalDateTime lastTime = LocalDateTimeUtil.of(lastUpdateTime);
+ query.and(w -> w.gt(ImConversationReadDO::getUpdateTime, lastTime)
+ .or(n -> n.eq(ImConversationReadDO::getUpdateTime, lastTime).gt(ImConversationReadDO::getId, lastId)));
+ }
+ return selectList(query.orderByAsc(ImConversationReadDO::getUpdateTime).orderByAsc(ImConversationReadDO::getId)
+ .last("LIMIT " + limit));
+ }
+
+ /**
+ * 单调递增更新读位置：仅当新位置更大时才更新
+ *
+ * 通过 {@code read_message_id < messageId} 的 CAS 条件，保证乱序 / 并发上报时读位置不会回退。
+ *
+ * @param id 记录编号
+ * @param messageId 新的最大已读消息编号
+ * @param readTime 已读时间
+ * @return 影响行数；0 表示新位置不大于已有位置，未更新
+ */
+ default int updateReadMessageIdToLarger(Long id, Long messageId, LocalDateTime readTime) {
+ return update(null, Wrappers.lambdaUpdate()
+ .set(ImConversationReadDO::getMessageId, messageId)
+ .set(ImConversationReadDO::getReadTime, readTime)
+ .set(ImConversationReadDO::getUpdateTime, readTime)
+ .eq(ImConversationReadDO::getId, id)
+ .lt(ImConversationReadDO::getMessageId, messageId));
+ }
+
+}
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/friend/ImFriendMapper.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/friend/ImFriendMapper.java
index 83752a9..8673100 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/friend/ImFriendMapper.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/friend/ImFriendMapper.java
@@ -1,5 +1,6 @@
package cn.iocoder.yudao.module.im.dal.mysql.friend;
+import cn.hutool.core.date.LocalDateTimeUtil;
import cn.iocoder.yudao.framework.common.pojo.PageResult;
import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
@@ -31,6 +32,28 @@ public interface ImFriendMapper extends BaseMapperX {
.eq(ImFriendDO::getUserId, userId)
.orderByDesc(ImFriendDO::getId));
}
+
+ /**
+ * 增量拉取当前用户的好友关系（含已删除，按 update_time + id 正向游标）
+ *
+ * @param userId 当前用户编号
+ * @param lastUpdateTime 上次拉取到的更新时间；首次拉取传 null
+ * @param lastId 上次拉取到的记录编号；首次拉取传 null
+ * @param limit 拉取数量
+ * @return 好友关系列表
+ */
+ default List selectPullListByUserId(Long userId, Long lastUpdateTime, Long lastId, Integer limit) {
+ LambdaQueryWrapperX query = new LambdaQueryWrapperX()
+ .eq(ImFriendDO::getUserId, userId);
+ if (lastUpdateTime != null && lastId != null) {
+ LocalDateTime lastTime = LocalDateTimeUtil.of(lastUpdateTime);
+ query.and(w -> w.gt(ImFriendDO::getUpdateTime, lastTime)
+ .or(n -> n.eq(ImFriendDO::getUpdateTime, lastTime).gt(ImFriendDO::getId, lastId)));
+ }
+ return selectList(query.orderByAsc(ImFriendDO::getUpdateTime).orderByAsc(ImFriendDO::getId)
+ .last("LIMIT " + limit));
+ }
+
default List selectListByUserIdAndStatus(Long userId, Integer status) {
return selectList(new LambdaQueryWrapperX()
.eq(ImFriendDO::getUserId, userId)
@@ -68,12 +91,14 @@ public interface ImFriendMapper extends BaseMapperX {
@SuppressWarnings("UnusedReturnValue")
default int updateReAddFields(Long id, Integer status, LocalDateTime addTime,
+ LocalDateTime updateTime,
Boolean silent, Boolean pinned, Boolean blocked,
String displayName, Integer addSource) {
return update(null, Wrappers.lambdaUpdate()
.eq(ImFriendDO::getId, id)
.set(ImFriendDO::getStatus, status)
.set(ImFriendDO::getAddTime, addTime)
+ .set(ImFriendDO::getUpdateTime, updateTime)
.set(ImFriendDO::getSilent, silent)
.set(ImFriendDO::getPinned, pinned)
.set(ImFriendDO::getBlocked, blocked)
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/friend/ImFriendRequestMapper.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/friend/ImFriendRequestMapper.java
index e81def0..6dffaa5 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/friend/ImFriendRequestMapper.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/friend/ImFriendRequestMapper.java
@@ -1,5 +1,6 @@
package cn.iocoder.yudao.module.im.dal.mysql.friend;
+import cn.hutool.core.date.LocalDateTimeUtil;
import cn.iocoder.yudao.framework.common.pojo.PageResult;
import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
@@ -28,6 +29,12 @@ public interface ImFriendRequestMapper extends BaseMapperX {
/**
* 拉取「我相关」的好友申请列表
+ *
+ * @param userId 当前用户编号
+ * @param maxRequestUpdateTime 起始更新时间；首次拉取传 null
+ * @param maxId 起始记录编号；首次拉取传 null
+ * @param limit 拉取数量
+ * @return 好友申请列表
*/
default List selectMyList(Long userId, LocalDateTime maxRequestUpdateTime,
Long maxId, int limit) {
@@ -45,6 +52,28 @@ public interface ImFriendRequestMapper extends BaseMapperX {
return selectList(wrapper);
}
+ /**
+ * 增量拉取「我相关」的好友申请（双向 OR，按 update_time + id 正向游标）
+ *
+ * @param userId 当前用户编号
+ * @param lastUpdateTime 上次拉取到的更新时间；首次拉取传 null
+ * @param lastId 上次拉取到的记录编号；首次拉取传 null
+ * @param limit 拉取数量
+ * @return 好友申请列表
+ */
+ default List selectPullListByUserId(Long userId, Long lastUpdateTime, Long lastId, Integer limit) {
+ LambdaQueryWrapperX query = new LambdaQueryWrapperX<>();
+ query.and(w -> w.eq(ImFriendRequestDO::getFromUserId, userId)
+ .or().eq(ImFriendRequestDO::getToUserId, userId));
+ if (lastUpdateTime != null && lastId != null) {
+ LocalDateTime lastTime = LocalDateTimeUtil.of(lastUpdateTime);
+ query.and(w -> w.gt(ImFriendRequestDO::getUpdateTime, lastTime)
+ .or(n -> n.eq(ImFriendRequestDO::getUpdateTime, lastTime).gt(ImFriendRequestDO::getId, lastId)));
+ }
+ return selectList(query.orderByAsc(ImFriendRequestDO::getUpdateTime).orderByAsc(ImFriendRequestDO::getId)
+ .last("LIMIT " + limit));
+ }
+
default int updateByIdAndHandleResult(Long id, Integer handleResult, ImFriendRequestDO updateObj) {
return update(updateObj, new LambdaUpdateWrapper()
.eq(ImFriendRequestDO::getId, id).eq(ImFriendRequestDO::getHandleResult, handleResult));
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/group/ImGroupMemberMapper.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/group/ImGroupMemberMapper.java
index 2ef54da..2afa559 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/group/ImGroupMemberMapper.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/group/ImGroupMemberMapper.java
@@ -1,7 +1,6 @@
package cn.iocoder.yudao.module.im.dal.mysql.group;
import cn.hutool.core.collection.CollUtil;
-import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupMemberDO;
@@ -59,20 +58,8 @@ public interface ImGroupMemberMapper extends BaseMapperX {
.eq(ImGroupMemberDO::getStatus, status));
}
- /**
- * 查询用户已退群的成员记录
- *
- * 当 {@code minQuitTime} 非空时额外按 {@code quitTime ≥ minQuitTime} 过滤。
- *
- * @param userId 用户编号
- * @param minQuitTime 最早退群时间（含），可空
- * @return 已退群成员记录列表
- */
- default List selectQuitListByUserId(Long userId, LocalDateTime minQuitTime) {
- return selectList(new LambdaQueryWrapperX()
- .eq(ImGroupMemberDO::getUserId, userId)
- .eq(ImGroupMemberDO::getStatus, CommonStatusEnum.DISABLE.getStatus())
- .geIfPresent(ImGroupMemberDO::getQuitTime, minQuitTime));
+ default List selectListByUserId(Long userId) {
+ return selectList(ImGroupMemberDO::getUserId, userId);
}
@SuppressWarnings("UnusedReturnValue")
@@ -143,7 +130,8 @@ public interface ImGroupMemberMapper extends BaseMapperX {
.set(ImGroupMemberDO::getAddSource, addSource)
.set(ImGroupMemberDO::getInviterUserId, inviterUserId)
.set(ImGroupMemberDO::getQuitTime, null)
- .set(ImGroupMemberDO::getMuteEndTime, null));
+ .set(ImGroupMemberDO::getMuteEndTime, null)
+ .set(ImGroupMemberDO::getUpdateTime, joinTime));
}
}
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/group/ImGroupRequestMapper.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/group/ImGroupRequestMapper.java
index 72fe756..c344493 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/group/ImGroupRequestMapper.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/group/ImGroupRequestMapper.java
@@ -1,5 +1,6 @@
package cn.iocoder.yudao.module.im.dal.mysql.group;
+import cn.hutool.core.date.LocalDateTimeUtil;
import cn.iocoder.yudao.framework.common.pojo.PageResult;
import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
@@ -43,6 +44,28 @@ public interface ImGroupRequestMapper extends BaseMapperX {
.orderByDesc(ImGroupRequestDO::getId));
}
+ /**
+ * 增量拉取「我管理的群」下的加群申请（含已处理，按 update_time + id 正向游标）
+ *
+ * @param groupIds 群编号集合
+ * @param lastUpdateTime 上次拉取到的更新时间；首次拉取传 null
+ * @param lastId 上次拉取到的记录编号；首次拉取传 null
+ * @param limit 拉取数量
+ * @return 加群申请列表
+ */
+ default List selectPullListByGroupIds(Collection groupIds, Long lastUpdateTime,
+ Long lastId, Integer limit) {
+ LambdaQueryWrapperX query = new LambdaQueryWrapperX()
+ .in(ImGroupRequestDO::getGroupId, groupIds);
+ if (lastUpdateTime != null && lastId != null) {
+ LocalDateTime lastTime = LocalDateTimeUtil.of(lastUpdateTime);
+ query.and(w -> w.gt(ImGroupRequestDO::getUpdateTime, lastTime)
+ .or(n -> n.eq(ImGroupRequestDO::getUpdateTime, lastTime).gt(ImGroupRequestDO::getId, lastId)));
+ }
+ return selectList(query.orderByAsc(ImGroupRequestDO::getUpdateTime).orderByAsc(ImGroupRequestDO::getId)
+ .last("LIMIT " + limit));
+ }
+
default int updateByIdAndHandleResult(Long id, Integer expectedHandleResult, ImGroupRequestDO updateObj) {
return update(updateObj, new LambdaUpdateWrapper()
.eq(ImGroupRequestDO::getId, id).eq(ImGroupRequestDO::getHandleResult, expectedHandleResult));
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImGroupMessageMapper.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImGroupMessageMapper.java
index 84736fd..d0d5584 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImGroupMessageMapper.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImGroupMessageMapper.java
@@ -4,13 +4,15 @@ import cn.iocoder.yudao.framework.common.pojo.PageResult;
import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
import cn.iocoder.yudao.framework.mybatis.core.query.QueryWrapperX;
+import cn.iocoder.yudao.framework.mybatis.core.util.MyBatisUtils;
import cn.iocoder.yudao.module.im.controller.admin.manager.message.vo.group.ImGroupMessageManagerPageReqVO;
import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO;
-import cn.iocoder.yudao.module.im.enums.message.ImGroupMessageReceiptStatusEnum;
+import cn.iocoder.yudao.module.im.enums.message.ImMessageReceiptStatusEnum;
import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
import org.apache.ibatis.annotations.Mapper;
import java.time.LocalDateTime;
+import java.util.Collection;
import java.util.List;
/**
@@ -22,48 +24,22 @@ import java.util.List;
public interface ImGroupMessageMapper extends BaseMapperX {
/**
- * 根据 minId + 时间窗口增量拉取群聊消息（在群成员使用）
+ * 根据 minId + 时间窗口增量拉取群聊消息
*
- * @param groupIds 用户当前仍在群内的群编号列表
+ * @param userId 当前用户编号
+ * @param groupIds 候选群编号集合（当前在群 ∪ 窗口内退群），不能为空
* @param minId 最小消息 id（不含）
* @param minSendTime 最早发送时间（不含），限制离线消息时间窗口
* @param size 拉取数量
* @return 消息列表（按 id 升序）
*/
- default List selectListByMinId(List groupIds, Long minId,
+ default List selectListByMinId(Long userId, Collection groupIds, Long minId,
LocalDateTime minSendTime, Integer size) {
QueryWrapperX wrapper = new QueryWrapperX<>();
wrapper.in("group_id", groupIds)
.gt("id", minId)
.gt("send_time", minSendTime)
- .orderByAsc("id");
- wrapper.limitN(size);
- return selectList(wrapper);
- }
-
- /**
- * 查询"退群前"的离线消息（退群成员使用）
- *
- * 语义：用户已退出某群，但仍需把 {@code minId} 之后、{@code minSendTime} 之后、不晚于退群时间的消息补齐到本地，便于前端看到完整上下文。
- *
- * 撤回消息（status=RECALL）保留返回，与在群成员的 {@link #selectListByMinId} 行为一致，由前端按撤回信号渲染「消息已撤回」气泡。
- *
- * @param groupId 群编号
- * @param minId 最小消息 id（不含）
- * @param minSendTime 最早发送时间（不含）
- * @param quitTime 退群时间（含），仅返回退群当时已存在的消息
- * @param size 拉取数量
- * @return 消息列表（按 id 升序）
- */
- default List selectListByGroupIdAndMinIdAndQuitTimeBefore(Long groupId, Long minId,
- LocalDateTime minSendTime,
- LocalDateTime quitTime,
- Integer size) {
- QueryWrapperX wrapper = new QueryWrapperX<>();
- wrapper.eq("group_id", groupId)
- .gt("id", minId)
- .gt("send_time", minSendTime)
- .le("send_time", quitTime)
+ .apply(MyBatisUtils.findInSet("receiver_user_ids"), userId)
.orderByAsc("id");
wrapper.limitN(size);
return selectList(wrapper);
@@ -72,17 +48,17 @@ public interface ImGroupMessageMapper extends BaseMapperX {
/**
* 查询群聊历史消息（游标拉取）
*
- * @param groupId 群编号
- * @param maxId 起始消息 id（不含），为空则从最新开始
- * @param limit 拉取数量
- * @param joinTime 入群时间，仅返回入群之后的消息
+ * @param userId 当前用户编号
+ * @param groupId 群编号
+ * @param maxId 起始消息 id（不含），为空则从最新开始
+ * @param limit 拉取数量
* @return 消息列表（按 id 倒序）
*/
- default List selectHistoryList(Long groupId, Long maxId, Integer limit, LocalDateTime joinTime) {
+ default List selectHistoryListByUser(Long userId, Long groupId, Long maxId, Integer limit) {
QueryWrapperX wrapper = new QueryWrapperX<>();
wrapper.eq("group_id", groupId)
.lt(maxId != null, "id", maxId)
- .ge(joinTime != null, "send_time", joinTime)
+ .apply(MyBatisUtils.findInSet("receiver_user_ids"), userId)
.orderByDesc("id");
wrapper.limitN(limit);
return selectList(wrapper);
@@ -108,7 +84,7 @@ public interface ImGroupMessageMapper extends BaseMapperX {
default List selectListByGroupIdAndPendingReceipt(Long groupId, Long minId, Long maxId) {
return selectList(new LambdaQueryWrapperX()
.eq(ImGroupMessageDO::getGroupId, groupId)
- .eq(ImGroupMessageDO::getReceiptStatus, ImGroupMessageReceiptStatusEnum.PENDING.getStatus())
+ .eq(ImGroupMessageDO::getReceiptStatus, ImMessageReceiptStatusEnum.PENDING.getStatus())
.gt(minId != null, ImGroupMessageDO::getId, minId)
.le(ImGroupMessageDO::getId, maxId)
.ne(ImGroupMessageDO::getStatus, ImMessageStatusEnum.RECALL.getStatus()));
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImPrivateMessageMapper.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImPrivateMessageMapper.java
index a33f29f..b666dfb 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImPrivateMessageMapper.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImPrivateMessageMapper.java
@@ -67,23 +67,13 @@ public interface ImPrivateMessageMapper extends BaseMapperX
.eq(ImPrivateMessageDO::getClientMessageId, clientMessageId));
}
- default Long selectMaxIdBySenderIdAndReceiverIdAndStatus(Long senderId, Long receiverId, Integer status) {
- ImPrivateMessageDO message = selectOne(new LambdaQueryWrapperX()
- .eq(ImPrivateMessageDO::getSenderId, senderId)
- .eq(ImPrivateMessageDO::getReceiverId, receiverId)
- .eq(ImPrivateMessageDO::getStatus, status)
- .orderByDesc(ImPrivateMessageDO::getId)
- .last("LIMIT 1"));
- return message != null ? message.getId() : null;
- }
-
- default int updateBySenderIdAndReceiverIdAndIdLeAndStatus(Long senderId, Long receiverId, Long maxMessageId,
- Integer whereStatus, ImPrivateMessageDO updateObj) {
+ default int updateBySenderIdAndReceiverIdAndIdLeAndReceiptStatus(Long senderId, Long receiverId, Long maxMessageId,
+ Integer whereReceiptStatus, ImPrivateMessageDO updateObj) {
return update(updateObj, new LambdaQueryWrapperX()
.eq(ImPrivateMessageDO::getSenderId, senderId)
.eq(ImPrivateMessageDO::getReceiverId, receiverId)
.le(ImPrivateMessageDO::getId, maxMessageId)
- .eq(ImPrivateMessageDO::getStatus, whereStatus));
+ .eq(ImPrivateMessageDO::getReceiptStatus, whereReceiptStatus));
}
default PageResult selectPage(ImPrivateMessageManagerPageReqVO reqVO) {
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/statistics/ImStatisticsManagerMapper.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/statistics/ImStatisticsManagerMapper.java
index 141f6b8..26e3d98 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/statistics/ImStatisticsManagerMapper.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/statistics/ImStatisticsManagerMapper.java
@@ -154,7 +154,7 @@ public interface ImStatisticsManagerMapper {
@Param("endTime") LocalDateTime endTime);
/**
- * 区间内消息类型分布（私聊+群聊合并）
+ * 区间内内容类型分布（私聊+群聊合并）
*
* @return [{type: 0, count: 123}, ...]
*/
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/redis/RedisKeyConstants.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/redis/RedisKeyConstants.java
index 247ad33..92d3890 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/redis/RedisKeyConstants.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/redis/RedisKeyConstants.java
@@ -9,20 +9,6 @@ package cn.iocoder.yudao.module.im.dal.redis;
public interface RedisKeyConstants {
/**
- * 群消息已读位置
- * KEY 格式： im:group:message:read:{groupId}
- * VALUE 数据类型： Hash (field: userId, value: maxReadMessageId)
- */
- String GROUP_MESSAGE_READ = "im:group:message:read:%s";
-
- /**
- * 频道消息已读位置
- * KEY 格式： im:channel:message:read:{channelId}
- * VALUE 数据类型： Hash (field: userId, value: maxReadMessageId)
- */
- String CHANNEL_MESSAGE_READ = "im:channel:message:read:%s";
-
- /**
* 好友关系状态缓存（合并「是否好友」+「是否拉黑」两态）
*
* KEY 格式：friend_state:{userId}_{friendUserId}
@@ -44,7 +30,7 @@ public interface RedisKeyConstants {
* KEY 格式：group_member_ids:{groupId}
* VALUE 数据类型：List
*
- * 说明：只缓存轻量的 userId 列表，适合"群消息推送目标"这类只关心 userId 的场景。
+ * 说明：只缓存轻量的 userId 列表，适合「群消息推送目标」这类只关心 userId 的场景
*/
String GROUP_MEMBER_IDS = "group_member_ids";
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/redis/message/ImChannelMessageReadRedisDAO.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/redis/message/ImChannelMessageReadRedisDAO.java
deleted file mode 100644
index bda4b33..0000000
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/redis/message/ImChannelMessageReadRedisDAO.java
+++ /dev/null
@@ -1,50 +0,0 @@
-package cn.iocoder.yudao.module.im.dal.redis.message;
-
-import cn.hutool.core.convert.Convert;
-import jakarta.annotation.Resource;
-import org.springframework.data.redis.core.StringRedisTemplate;
-import org.springframework.stereotype.Repository;
-
-import static cn.iocoder.yudao.module.im.dal.redis.RedisKeyConstants.CHANNEL_MESSAGE_READ;
-
-/**
- * IM 频道消息已读位置 Redis DAO
- *
- * @author 芋道源码
- */
-@Repository
-public class ImChannelMessageReadRedisDAO {
-
- @Resource
- private StringRedisTemplate stringRedisTemplate;
-
- /**
- * 更新用户在某频道的最大已读消息编号
- *
- * @param channelId 频道编号
- * @param userId 用户编号
- * @param maxMessageId 最大已读消息编号
- */
- public void updateReadMaxMessageId(Long channelId, Long userId, Long maxMessageId) {
- String key = formatKey(channelId);
- stringRedisTemplate.opsForHash().put(key, userId.toString(), maxMessageId.toString());
- }
-
- /**
- * 获取用户在某频道的最大已读消息编号
- *
- * @param channelId 频道编号
- * @param userId 用户编号
- * @return 最大已读消息编号；不存在则返回 null
- */
- public Long getReadMaxMessageId(Long channelId, Long userId) {
- String key = formatKey(channelId);
- Object val = stringRedisTemplate.opsForHash().get(key, userId.toString());
- return Convert.toLong(val);
- }
-
- private static String formatKey(Long channelId) {
- return String.format(CHANNEL_MESSAGE_READ, channelId);
- }
-
-}
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/redis/message/ImGroupMessageReadRedisDAO.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/redis/message/ImGroupMessageReadRedisDAO.java
deleted file mode 100644
index 5c76f79..0000000
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/redis/message/ImGroupMessageReadRedisDAO.java
+++ /dev/null
@@ -1,102 +0,0 @@
-package cn.iocoder.yudao.module.im.dal.redis.message;
-
-import cn.hutool.core.convert.Convert;
-import jakarta.annotation.Resource;
-import org.springframework.data.redis.core.StringRedisTemplate;
-import org.springframework.stereotype.Repository;
-
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Map;
-
-import static cn.iocoder.yudao.module.im.dal.redis.RedisKeyConstants.GROUP_MESSAGE_READ;
-
-/**
- * IM 群消息已读位置 Redis DAO
- *
- * @author 芋道源码
- */
-@Repository
-public class ImGroupMessageReadRedisDAO {
-
- @Resource
- private StringRedisTemplate stringRedisTemplate;
-
- /**
- * 更新用户在某群的最大已读消息编号
- *
- * @param groupId 群编号
- * @param userId 用户编号
- * @param maxMessageId 最大已读消息编号
- */
- public void updateReadMaxMessageId(Long groupId, Long userId, Long maxMessageId) {
- String key = formatKey(groupId);
- stringRedisTemplate.opsForHash().put(key, userId.toString(), maxMessageId.toString());
- }
-
- /**
- * 获取用户在某群的最大已读消息编号
- *
- * @param groupId 群编号
- * @param userId 用户编号
- * @return 最大已读消息编号，不存在则返回 null
- */
- public Long getReadMaxMessageId(Long groupId, Long userId) {
- String key = formatKey(groupId);
- Object val = stringRedisTemplate.opsForHash().get(key, userId.toString());
- return Convert.toLong(val);
- }
-
- /**
- * 获取某群所有用户的已读位置
- *
- * @param groupId 群编号
- * @return userId → maxReadMessageId 映射
- */
- public Map getReadMaxMessageIdMap(Long groupId) {
- String key = formatKey(groupId);
- Map entries = stringRedisTemplate.opsForHash().entries(key);
- // 转换为 Long → Long 的 Map
- Map result = new HashMap<>(entries.size());
- entries.forEach((k, v) -> result.put(Long.parseLong(k.toString()), Long.parseLong(v.toString())));
- return result;
- }
-
- /**
- * 删除用户在某群的已读位置
- *
- * @param groupId 群编号
- * @param userId 用户编号
- */
- public void deleteReadMaxMessageId(Long groupId, Long userId) {
- String key = formatKey(groupId);
- stringRedisTemplate.opsForHash().delete(key, userId.toString());
- }
-
- /**
- * 批量删除用户在某群的已读位置
- *
- * @param groupId 群编号
- * @param userIds 用户编号集合
- */
- public void deleteReadMaxMessageIds(Long groupId, Collection userIds) {
- String key = formatKey(groupId);
- Object[] hashKeys = userIds.stream().map(String::valueOf).toArray();
- stringRedisTemplate.opsForHash().delete(key, hashKeys);
- }
-
- /**
- * 删除某群所有用户的已读位置（整个 Hash Key）
- *
- * @param groupId 群编号
- */
- public void deleteReadMaxMessageIdMap(Long groupId) {
- String key = formatKey(groupId);
- stringRedisTemplate.delete(key);
- }
-
- private static String formatKey(Long groupId) {
- return String.format(GROUP_MESSAGE_READ, groupId);
- }
-
-}
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/message/ImMessageTypeEnum.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/ImContentTypeEnum.java
index d854306..2f68ad8 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/message/ImMessageTypeEnum.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/ImContentTypeEnum.java
@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.im.enums.message;
+package cn.iocoder.yudao.module.im.enums;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.Assert;
@@ -11,13 +11,13 @@ import java.util.Arrays;
import java.util.Set;
/**
- * IM 消息类型枚举
+ * IM 内容类型枚举
*
* @author 芋道源码
*/
@Getter
@RequiredArgsConstructor
-public enum ImMessageTypeEnum implements ArrayValuable {
+public enum ImContentTypeEnum implements ArrayValuable {
// ========== 用户聊天消息（101-105 直接复用 OpenIM 段位编号） ==========
/**
@@ -128,10 +128,6 @@ public enum ImMessageTypeEnum implements ArrayValuable {
RTC_CALL_END(1611, "通话结束", true, false),
// ========== 好友通知（1201-1210 直接复用 OpenIM 段位编号） ==========
- // TODO @芋艿：FRIEND_REQUEST_* 与 GROUP_REQUEST_* 都是 persistent=false 的 SysMsg，离线 pull 拉不到，
- // 目前可能丢失「实时 toast 提醒」体验（业务状态不丢，前端上线 fetch{Friend,Group}RequestList 能补回来）；
- // 未来思考下怎么优化：候选方案 1）改 persistent=true 入私聊消息流 + 让客户端按 type 自渲染；
- // 2）服务端补一个「未读通知拉取」接口给前端冷启动调用。
/**
* 对应 OpenIM：FriendApplicationApprovedNotification 1201
* 对应自己的类：FriendRequestApprovedNotification
@@ -285,9 +281,9 @@ public enum ImMessageTypeEnum implements ArrayValuable {
/**
* 对应 OpenIM：sdkws.GroupMemberInfoSetTips（GroupMemberInfoSetNotification 1516，窄化到 displayUserName）
* 对应自己的类：GroupMemberNicknameUpdateNotification
- * 场景：成员修改自己在群里的昵称，全员广播；前端按 displayUserName 局部更新对应 member
+ * 场景：成员修改自己在群里的昵称，在线成员同步对应 member
*/
- GROUP_MEMBER_NICKNAME_UPDATE(1516, "成员昵称变更", true, false),
+ GROUP_MEMBER_NICKNAME_UPDATE(1516, "成员昵称变更", false, false),
/**
* 对应 OpenIM：GroupMemberSetToAdminNotification 1517
* 对应自己的类：GroupAdminAddNotification
@@ -339,7 +335,7 @@ public enum ImMessageTypeEnum implements ArrayValuable {
*/
GROUP_BANNED(1533, "群封禁变更", true, false);
- public static final Integer[] ARRAYS = Arrays.stream(values()).map(ImMessageTypeEnum::getType).toArray(Integer[]::new);
+ public static final Integer[] ARRAYS = Arrays.stream(values()).map(ImContentTypeEnum::getType).toArray(Integer[]::new);
private static final Set FRIEND_NOTIFICATION_TYPES = CollUtil.newHashSet(
FRIEND_REQUEST_APPROVED.type,
@@ -396,8 +392,8 @@ public enum ImMessageTypeEnum implements ArrayValuable {
* @param type 消息类型
* @return 枚举实例
*/
- public static ImMessageTypeEnum validate(Integer type) {
- ImMessageTypeEnum result = ArrayUtil.firstMatch(item -> item.type.equals(type), values());
+ public static ImContentTypeEnum validate(Integer type) {
+ ImContentTypeEnum result = ArrayUtil.firstMatch(item -> item.type.equals(type), values());
Assert.notNull(result, "未注册的消息类型 type={}", type);
return result;
}
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/ImConversationTypeEnum.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/ImConversationTypeEnum.java
index 0f3f446..eb3c053 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/ImConversationTypeEnum.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/ImConversationTypeEnum.java
@@ -16,6 +16,7 @@ import java.util.Objects;
@Getter
public enum ImConversationTypeEnum implements ArrayValuable {
+ NONE(0, "无会话"), // 无会话
PRIVATE(1, "私聊"), // 私聊
GROUP(2, "群聊"), // 群聊
CHANNEL(3, "频道"); // 频道 / 公众号
@@ -36,6 +37,10 @@ public enum ImConversationTypeEnum implements ArrayValuable {
return ARRAYS;
}
+ public static boolean isNone(Integer type) {
+ return Objects.equals(NONE.type, type);
+ }
+
public static boolean isPrivate(Integer type) {
return Objects.equals(PRIVATE.type, type);
}
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/message/ImGroupMessageReceiptStatusEnum.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/message/ImMessageReceiptStatusEnum.java
index 256560c..41e3cf0 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/message/ImGroupMessageReceiptStatusEnum.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/message/ImMessageReceiptStatusEnum.java
@@ -7,19 +7,19 @@ import lombok.Getter;
import java.util.Arrays;
/**
- * IM 群消息回执状态枚举
+ * IM 消息回执状态枚举
*
* @author 芋道源码
*/
@Getter
@AllArgsConstructor
-public enum ImGroupMessageReceiptStatusEnum implements ArrayValuable {
+public enum ImMessageReceiptStatusEnum implements ArrayValuable {
NO_RECEIPT(0, "不需要回执"),
PENDING(1, "待完成"),
DONE(2, "已完成");
- public static final Integer[] ARRAYS = Arrays.stream(values()).map(ImGroupMessageReceiptStatusEnum::getStatus).toArray(Integer[]::new);
+ public static final Integer[] ARRAYS = Arrays.stream(values()).map(ImMessageReceiptStatusEnum::getStatus).toArray(Integer[]::new);
/**
* 状态
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/message/ImMessageStatusEnum.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/message/ImMessageStatusEnum.java
index 79cb8d8..537f95c 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/message/ImMessageStatusEnum.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/message/ImMessageStatusEnum.java
@@ -8,9 +8,6 @@ import java.util.Arrays;
/**
* IM 消息状态枚举
- *
- * 私聊：SENDING(-1, 仅客户端) / UNREAD(0) / RECALL(2) / READ(3)
- * 群聊：SENDING(-1, 仅客户端) / UNREAD(0, 作为正常状态) / RECALL(2)
*
* @author 芋道源码
*/
@@ -19,9 +16,8 @@ import java.util.Arrays;
public enum ImMessageStatusEnum implements ArrayValuable {
SENDING(-1, "发送中"), // 仅客户端使用
- UNREAD(0, "未读"), // 私聊=未读，群聊=正常（初始状态）
- RECALL(2, "已撤回"),
- READ(3, "已读"); // 仅私聊使用；群聊已读通过 Redis 已读位置实现
+ NORMAL(0, "正常"), // 私聊 / 群聊的正常（初始）状态
+ RECALL(2, "已撤回");
public static final Integer[] ARRAYS = Arrays.stream(values()).map(ImMessageStatusEnum::getStatus).toArray(Integer[]::new);
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/rtc/ImRtcParticipantStatusEnum.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/rtc/ImRtcParticipantStatusEnum.java
index 3f3acc4..c4af81b 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/rtc/ImRtcParticipantStatusEnum.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/rtc/ImRtcParticipantStatusEnum.java
@@ -24,7 +24,6 @@ import java.util.Objects;
@Getter
public enum ImRtcParticipantStatusEnum implements ArrayValuable {
- // TODO @芋艿：hand up 要不要也搞下。
INVITING(10, "邀请中"), // 已发出 invite，等被叫响应
JOINED(20, "已加入"), // 已 connect 进 LiveKit 房间
REJECTED(30, "已拒绝"), // 接通前点拒接
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/mq/consumer/friend/AdminUserProfileUpdateConsumer.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/mq/consumer/friend/AdminUserProfileUpdateConsumer.java
index b39ce13..13eeee6 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/mq/consumer/friend/AdminUserProfileUpdateConsumer.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/mq/consumer/friend/AdminUserProfileUpdateConsumer.java
@@ -2,11 +2,11 @@ package cn.iocoder.yudao.module.im.mq.consumer.friend;
import cn.hutool.core.collection.CollUtil;
import cn.iocoder.yudao.module.im.dal.dataobject.friend.ImFriendDO;
-import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
import cn.iocoder.yudao.module.im.service.friend.ImFriendService;
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.notification.friend.FriendInfoUpdatedNotification;
+import cn.iocoder.yudao.module.im.service.websocket.notification.friend.FriendInfoUpdatedNotification;
import cn.iocoder.yudao.module.system.api.message.user.AdminUserProfileUpdateMessage;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
@@ -52,8 +52,9 @@ public class AdminUserProfileUpdateConsumer {
try {
FriendInfoUpdatedNotification payload = (FriendInfoUpdatedNotification) new FriendInfoUpdatedNotification()
.setOperatorUserId(userId).setFriendUserId(userId);
- websocketService.sendPrivateMessageAsync(friend.getFriendUserId(), ImPrivateMessageDTO.ofFriendNotification(
- ImMessageTypeEnum.FRIEND_INFO_UPDATED.getType(), userId, friend.getFriendUserId(), payload));
+ websocketService.sendNotificationAsync(friend.getFriendUserId(),
+ ImConversationTypeEnum.NONE.getType(),
+ ImContentTypeEnum.FRIEND_INFO_UPDATED.getType(), payload);
successCount++;
} catch (Exception e) {
log.warn("[onMessage][userId({}) friendUserId({}) 推送失败]",
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/conversation/ImConversationReadService.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/conversation/ImConversationReadService.java
new file mode 100644
index 0000000..9c057ae
--- /dev/null
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/conversation/ImConversationReadService.java
@@ -0,0 +1,67 @@
+package cn.iocoder.yudao.module.im.service.conversation;
+
+import cn.iocoder.yudao.module.im.dal.dataobject.conversation.ImConversationReadDO;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * IM 会话读位置 Service 接口
+ *
+ * @author 芋道源码
+ */
+public interface ImConversationReadService {
+
+ /**
+ * 更新用户在某会话的最大已读位置（单调递增，乱序 / 并发上报不会回退）
+ *
+ * @param userId 用户编号
+ * @param conversationType 会话类型
+ * @param conversationId 会话编号
+ * @param readMessageId 已读到的最大消息编号
+ * @return 读位置是否前进；true 时调用方才需要下发已读 / 回执事件
+ */
+ boolean updateConversationReadPosition(Long userId, Integer conversationType, Long conversationId, Long readMessageId);
+
+ /**
+ * 获取用户在某会话的最大已读位置
+ *
+ * @param userId 用户编号
+ * @param conversationType 会话类型
+ * @param conversationId 会话编号
+ * @return 最大已读消息编号；不存在则返回 null
+ */
+ Long getConversationReadMessageId(Long userId, Integer conversationType, Long conversationId);
+
+ /**
+ * 获取某会话内所有用户的读位置（用于群回执人数聚合）
+ *
+ * @param conversationType 会话类型
+ * @param conversationId 会话编号
+ * @return userId → 最大已读消息编号
+ */
+ Map getUserReadMessageIdMap(Integer conversationType, Long conversationId);
+
+ /**
+ * 批量获取某用户在多个会话的读位置（用于频道批量读位置、重连后按活跃会话补偿）
+ *
+ * @param userId 用户编号
+ * @param conversationType 会话类型
+ * @param conversationIds 会话编号集合
+ * @return conversationId → 最大已读消息编号
+ */
+ Map getConversationReadMessageIdMap(Long userId, Integer conversationType, Collection conversationIds);
+
+ /**
+ * 增量拉取当前用户的会话读位置（重连 / 离线补偿：按 update_time + id 游标）
+ *
+ * @param userId 用户编号
+ * @param lastUpdateTime 上次拉取到的最新更新时间（毫秒时间戳）；首次拉取传 null
+ * @param lastId 上次拉取到的最后一条记录 id；首次拉取传 null
+ * @param limit 单次拉取条数
+ * @return 会话读位置列表，按更新时间、id 正序
+ */
+ List pullConversationReadList(Long userId, Long lastUpdateTime, Long lastId, Integer limit);
+
+}
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/conversation/ImConversationReadServiceImpl.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/conversation/ImConversationReadServiceImpl.java
new file mode 100644
index 0000000..a22d616
--- /dev/null
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/conversation/ImConversationReadServiceImpl.java
@@ -0,0 +1,86 @@
+package cn.iocoder.yudao.module.im.service.conversation;
+
+import cn.hutool.core.collection.CollUtil;
+import cn.iocoder.yudao.module.im.dal.dataobject.conversation.ImConversationReadDO;
+import cn.iocoder.yudao.module.im.dal.mysql.conversation.ImConversationReadMapper;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.stereotype.Service;
+
+import java.time.LocalDateTime;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertMap;
+
+/**
+ * IM 会话读位置 Service 实现类
+ *
+ * @author 芋道源码
+ */
+@Service
+@Slf4j
+public class ImConversationReadServiceImpl implements ImConversationReadService {
+
+ @Resource
+ private ImConversationReadMapper conversationReadMapper;
+
+ @Override
+ public boolean updateConversationReadPosition(Long userId, Integer conversationType, Long conversationId, Long readMessageId) {
+ LocalDateTime now = LocalDateTime.now();
+ // 1. 不存在则插入；并发下唯一键冲突，降级为回查 + CAS 更新
+ ImConversationReadDO existing = conversationReadMapper.selectByUserIdAndConversation(
+ userId, conversationType, conversationId);
+ if (existing == null) {
+ try {
+ conversationReadMapper.insert(new ImConversationReadDO().setUserId(userId)
+ .setConversationType(conversationType).setTargetId(conversationId)
+ .setMessageId(readMessageId).setReadTime(now));
+ return true;
+ } catch (DuplicateKeyException e) {
+ log.warn("[updateConversationReadPosition][userId({}) type({}) conversationId({}) 并发插入冲突，回查更新]",
+ userId, conversationType, conversationId);
+ existing = conversationReadMapper.selectByUserIdAndConversation(userId, conversationType, conversationId);
+ }
+ }
+ if (existing == null) {
+ return false;
+ }
+
+ // 2. CAS 单调更新：mapper 内 WHERE message_id < ? 保证乱序 / 并发不回退；影响行数 > 0 即读位置前进
+ return conversationReadMapper.updateReadMessageIdToLarger(existing.getId(), readMessageId, now) > 0;
+ }
+
+ @Override
+ public Long getConversationReadMessageId(Long userId, Integer conversationType, Long conversationId) {
+ ImConversationReadDO read = conversationReadMapper.selectByUserIdAndConversation(
+ userId, conversationType, conversationId);
+ return read != null ? read.getMessageId() : null;
+ }
+
+ @Override
+ public Map getUserReadMessageIdMap(Integer conversationType, Long conversationId) {
+ return convertMap(conversationReadMapper.selectListByConversation(conversationType, conversationId),
+ ImConversationReadDO::getUserId, ImConversationReadDO::getMessageId);
+ }
+
+ @Override
+ public Map getConversationReadMessageIdMap(Long userId, Integer conversationType,
+ Collection conversationIds) {
+ if (CollUtil.isEmpty(conversationIds)) {
+ return Collections.emptyMap();
+ }
+ List list = conversationReadMapper.selectListByUserIdAndConversations(
+ userId, conversationType, conversationIds);
+ return convertMap(list, ImConversationReadDO::getTargetId, ImConversationReadDO::getMessageId);
+ }
+
+ @Override
+ public List pullConversationReadList(Long userId, Long lastUpdateTime, Long lastId, Integer limit) {
+ return conversationReadMapper.selectPullListByUserId(userId, lastUpdateTime, lastId, limit);
+ }
+
+}
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/friend/ImFriendRequestService.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/friend/ImFriendRequestService.java
index a78bb38..8470206 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/friend/ImFriendRequestService.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/friend/ImFriendRequestService.java
@@ -51,6 +51,17 @@ public interface ImFriendRequestService {
List getMyFriendRequestList(Long userId, Long maxId, Integer limit);
/**
+ * 增量拉取「我相关」的好友申请（重连 / 离线补偿：双向 OR，按 update_time + id 游标）
+ *
+ * @param userId 用户编号
+ * @param lastUpdateTime 上次拉取到的最新更新时间（毫秒时间戳）；首次拉取传 null
+ * @param lastId 上次拉取到的最后一条记录 id；首次拉取传 null
+ * @param limit 单次拉取条数
+ * @return 申请记录列表，按更新时间、id 正序
+ */
+ List pullFriendRequestList(Long userId, Long lastUpdateTime, Long lastId, Integer limit);
+
+ /**
* 按 id 单查申请记录；通用读接口，调用方自行做越权过滤
*/
ImFriendRequestDO getFriendRequest(Long id);
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/friend/ImFriendRequestServiceImpl.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/friend/ImFriendRequestServiceImpl.java
index d024e9d..eeed457 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/friend/ImFriendRequestServiceImpl.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/friend/ImFriendRequestServiceImpl.java
@@ -14,13 +14,13 @@ import cn.iocoder.yudao.module.im.dal.dataobject.friend.ImFriendRequestDO;
import cn.iocoder.yudao.module.im.dal.mysql.friend.ImFriendRequestMapper;
import cn.iocoder.yudao.module.im.enums.friend.ImFriendRequestHandleResultEnum;
import cn.iocoder.yudao.module.im.enums.friend.ImFriendStateEnum;
-import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
import cn.iocoder.yudao.module.im.framework.config.ImProperties;
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.notification.friend.FriendRequestApprovedNotification;
-import cn.iocoder.yudao.module.im.service.websocket.dto.notification.friend.FriendRequestNotification;
-import cn.iocoder.yudao.module.im.service.websocket.dto.notification.friend.FriendRequestRejectedNotification;
+import cn.iocoder.yudao.module.im.service.websocket.notification.friend.FriendRequestApprovedNotification;
+import cn.iocoder.yudao.module.im.service.websocket.notification.friend.FriendRequestNotification;
+import cn.iocoder.yudao.module.im.service.websocket.notification.friend.FriendRequestRejectedNotification;
import cn.iocoder.yudao.module.system.api.user.AdminUserApi;
import cn.iocoder.yudao.module.system.api.user.dto.AdminUserRespDTO;
import jakarta.annotation.Resource;
@@ -105,8 +105,8 @@ public class ImFriendRequestServiceImpl implements ImFriendRequestService {
if (fromUser != null) {
payload.setFromNickname(fromUser.getNickname()).setFromAvatar(fromUser.getAvatar());
}
- websocketService.sendPrivateMessageAsync(toUserId, ImPrivateMessageDTO.ofFriendNotification(
- ImMessageTypeEnum.FRIEND_REQUEST_RECEIVED.getType(), fromUserId, toUserId, payload));
+ websocketService.sendNotificationAsync(toUserId, ImConversationTypeEnum.NONE.getType(),
+ ImContentTypeEnum.FRIEND_REQUEST_RECEIVED.getType(), payload);
// 4. 全局自动通过开关：注册 afterCommit 回调，事务提交后再走同意流程
// 回调内 try/catch 兜底 —— afterCommit 异常会被 Spring 静默吞掉，否则同意失败时申请方永远等不到 APPROVED
@@ -172,8 +172,6 @@ public class ImFriendRequestServiceImpl implements ImFriendRequestService {
public void agreeFriendRequest(Long userId, Long requestId) {
// 1.1 校验申请存在、未处理、操作人是接收方
ImFriendRequestDO request = validateRequestForHandle(userId, requestId);
- // 1.2 复验双方用户有效
- adminUserApi.validateUserList(ListUtil.of(request.getFromUserId(), request.getToUserId()));
// 2. 乐观锁更新申请处理结果
ImFriendRequestDO updateObj = new ImFriendRequestDO()
@@ -192,8 +190,8 @@ public class ImFriendRequestServiceImpl implements ImFriendRequestService {
FriendRequestApprovedNotification payload = (FriendRequestApprovedNotification)
new FriendRequestApprovedNotification().setRequestId(request.getId())
.setOperatorUserId(userId).setFriendUserId(userId);
- websocketService.sendPrivateMessageAsync(request.getFromUserId(), ImPrivateMessageDTO.ofFriendNotification(
- ImMessageTypeEnum.FRIEND_REQUEST_APPROVED.getType(), userId, request.getFromUserId(), payload));
+ websocketService.sendNotificationAsync(request.getFromUserId(), ImConversationTypeEnum.NONE.getType(),
+ ImContentTypeEnum.FRIEND_REQUEST_APPROVED.getType(), payload);
}
@Override
@@ -217,8 +215,8 @@ public class ImFriendRequestServiceImpl implements ImFriendRequestService {
new FriendRequestRejectedNotification().setRequestId(request.getId())
.setHandleContent(handleContent)
.setOperatorUserId(userId).setFriendUserId(userId);
- websocketService.sendPrivateMessageAsync(request.getFromUserId(), ImPrivateMessageDTO.ofFriendNotification(
- ImMessageTypeEnum.FRIEND_REQUEST_REJECTED.getType(), userId, request.getFromUserId(), payload));
+ websocketService.sendNotificationAsync(request.getFromUserId(), ImConversationTypeEnum.NONE.getType(),
+ ImContentTypeEnum.FRIEND_REQUEST_REJECTED.getType(), payload);
}
@Override
@@ -234,6 +232,11 @@ public class ImFriendRequestServiceImpl implements ImFriendRequestService {
}
@Override
+ public List pullFriendRequestList(Long userId, Long lastUpdateTime, Long lastId, Integer limit) {
+ return friendRequestMapper.selectPullListByUserId(userId, lastUpdateTime, lastId, limit);
+ }
+
+ @Override
public ImFriendRequestDO getFriendRequest(Long id) {
return friendRequestMapper.selectById(id);
}
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/friend/ImFriendService.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/friend/ImFriendService.java
index 84a6de9..1d1fc57 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/friend/ImFriendService.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/friend/ImFriendService.java
@@ -43,6 +43,11 @@ public interface ImFriendService {
List getFriendList(Long userId);
/**
+ * 增量拉取当前用户的好友关系（重连 / 离线补偿：含已删除，按 update_time + id 游标）
+ */
+ List pullFriendList(Long userId, Long lastUpdateTime, Long lastId, Integer limit);
+
+ /**
* 获得当前用户的有效好友列表（仅 ENABLE 状态）
*/
List getEnableFriendList(Long userId);
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/friend/ImFriendServiceImpl.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/friend/ImFriendServiceImpl.java
index 1919aae..f9dd452 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/friend/ImFriendServiceImpl.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/friend/ImFriendServiceImpl.java
@@ -11,12 +11,12 @@ import cn.iocoder.yudao.module.im.dal.dataobject.friend.ImFriendDO;
import cn.iocoder.yudao.module.im.dal.dataobject.friend.ImFriendRequestDO;
import cn.iocoder.yudao.module.im.dal.mysql.friend.ImFriendMapper;
import cn.iocoder.yudao.module.im.enums.friend.ImFriendStateEnum;
-import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
import cn.iocoder.yudao.module.im.service.message.ImPrivateMessageService;
import cn.iocoder.yudao.module.im.service.message.dto.ImPrivateMessageSendDTO;
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.notification.friend.*;
+import cn.iocoder.yudao.module.im.service.websocket.notification.friend.*;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cache.annotation.CacheEvict;
@@ -99,6 +99,11 @@ public class ImFriendServiceImpl implements ImFriendService {
}
@Override
+ public List pullFriendList(Long userId, Long lastUpdateTime, Long lastId, Integer limit) {
+ return friendMapper.selectPullListByUserId(userId, lastUpdateTime, lastId, limit);
+ }
+
+ @Override
public List getEnableFriendList(Long userId) {
return friendMapper.selectListByUserIdAndStatus(userId, CommonStatusEnum.ENABLE.getStatus());
}
@@ -155,8 +160,8 @@ public class ImFriendServiceImpl implements ImFriendService {
FriendUpdateNotification payload = (FriendUpdateNotification) new FriendUpdateNotification()
.setDisplayName(reqVO.getDisplayName()).setSilent(reqVO.getSilent()).setPinned(reqVO.getPinned())
.setOperatorUserId(userId).setFriendUserId(reqVO.getFriendUserId());
- websocketService.sendPrivateMessageAsync(userId, ImPrivateMessageDTO.ofFriendNotification(
- ImMessageTypeEnum.FRIEND_UPDATE.getType(), userId, userId, payload));
+ websocketService.sendNotificationAsync(userId, ImConversationTypeEnum.NONE.getType(),
+ ImContentTypeEnum.FRIEND_UPDATE.getType(), payload);
}
@Override
@@ -178,7 +183,7 @@ public class ImFriendServiceImpl implements ImFriendService {
FriendAddNotification payload = (FriendAddNotification) new FriendAddNotification()
.setOperatorUserId(fromUserId).setFriendUserId(toUserId);
privateMessageService.sendPrivateMessage(fromUserId, new ImPrivateMessageSendDTO()
- .setReceiverId(toUserId).setType(ImMessageTypeEnum.FRIEND_ADD.getType()).setContent(payload));
+ .setReceiverId(toUserId).setType(ImContentTypeEnum.FRIEND_ADD.getType()).setContent(payload));
}
@Override
@@ -197,7 +202,7 @@ public class ImFriendServiceImpl implements ImFriendService {
FriendAddNotification payload = (FriendAddNotification) new FriendAddNotification()
.setOperatorUserId(friendUserId).setFriendUserId(friendUserId);
privateMessageService.sendPrivateMessage(userId, new ImPrivateMessageSendDTO()
- .setReceiverId(friendUserId).setType(ImMessageTypeEnum.FRIEND_ADD.getType())
+ .setReceiverId(friendUserId).setType(ImContentTypeEnum.FRIEND_ADD.getType())
.setContent(payload).setPersistent(false));
}
@@ -219,7 +224,7 @@ public class ImFriendServiceImpl implements ImFriendService {
FriendDeleteNotification payload = ((FriendDeleteNotification) new FriendDeleteNotification()
.setOperatorUserId(userId).setFriendUserId(friendUserId)).setClear(clear);
privateMessageService.sendPrivateMessage(userId, new ImPrivateMessageSendDTO()
- .setReceiverId(friendUserId).setType(ImMessageTypeEnum.FRIEND_DELETE.getType())
+ .setReceiverId(friendUserId).setType(ImContentTypeEnum.FRIEND_DELETE.getType())
.setContent(payload).setPersistent(false));
}
@@ -245,8 +250,8 @@ public class ImFriendServiceImpl implements ImFriendService {
// 3. 推 FRIEND_BLOCK 给 A 多端
FriendBlockNotification payload = (FriendBlockNotification) new FriendBlockNotification()
.setOperatorUserId(userId).setFriendUserId(friendUserId);
- websocketService.sendPrivateMessageAsync(userId, ImPrivateMessageDTO.ofFriendNotification(
- ImMessageTypeEnum.FRIEND_BLOCK.getType(), userId, userId, payload));
+ websocketService.sendNotificationAsync(userId, ImConversationTypeEnum.NONE.getType(),
+ ImContentTypeEnum.FRIEND_BLOCK.getType(), payload);
}
@Override
@@ -271,8 +276,8 @@ public class ImFriendServiceImpl implements ImFriendService {
// 3. 推 FRIEND_UNBLOCK 给 A 多端
FriendUnblockNotification payload = (FriendUnblockNotification) new FriendUnblockNotification()
.setOperatorUserId(userId).setFriendUserId(friendUserId);
- websocketService.sendPrivateMessageAsync(userId, ImPrivateMessageDTO.ofFriendNotification(
- ImMessageTypeEnum.FRIEND_UNBLOCK.getType(), userId, userId, payload));
+ websocketService.sendNotificationAsync(userId, ImConversationTypeEnum.NONE.getType(),
+ ImContentTypeEnum.FRIEND_UNBLOCK.getType(), payload);
}
/**
@@ -294,7 +299,8 @@ public class ImFriendServiceImpl implements ImFriendService {
}
// 情况二：复用 DISABLE 旧记录 → 恢复 ENABLE + 重置 silent / pinned / blocked，对齐"重新加好友"语义
if (exists != null) {
- friendMapper.updateReAddFields(exists.getId(), CommonStatusEnum.ENABLE.getStatus(), LocalDateTime.now(),
+ LocalDateTime now = LocalDateTime.now();
+ friendMapper.updateReAddFields(exists.getId(), CommonStatusEnum.ENABLE.getStatus(), now, now,
false, false, false, displayName, addSource);
return;
}
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupMemberService.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupMemberService.java
index d9a8000..a99c0f3 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupMemberService.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupMemberService.java
@@ -87,13 +87,12 @@ public interface ImGroupMemberService {
List getActiveGroupMemberListByUserId(Long userId);
/**
- * 查询用户已退群的群成员记录（DISABLE 状态）
+ * 查询用户曾经加入的所有群成员记录（含已退群）
*
- * @param userId 用户编号，必传
- * @param minQuitTime 最早退群时间（含），可空
- * @return 已退群成员记录列表
+ * @param userId 用户编号
+ * @return 群成员记录列表
*/
- List getQuitGroupMemberListByUserId(Long userId, LocalDateTime minQuitTime);
+ List getGroupMemberListByUserId(Long userId);
/**
* 添加群成员（入群），角色默认 MEMBER
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupMemberServiceImpl.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupMemberServiceImpl.java
index ed9e1bb..c712a0b 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupMemberServiceImpl.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupMemberServiceImpl.java
@@ -77,7 +77,6 @@ public class ImGroupMemberServiceImpl implements ImGroupMemberService {
@Override
public List getGroupMemberListByOwnerAndAdmin(Long groupId) {
- // TODO DONE @AI：把条件往下传；这样减少加载数据量！
return groupMemberMapper.selectListByGroupIdAndStatusAndRoles(groupId, CommonStatusEnum.ENABLE.getStatus(),
ListUtil.of(ImGroupMemberRoleEnum.OWNER.getRole(), ImGroupMemberRoleEnum.ADMIN.getRole()));
}
@@ -106,8 +105,8 @@ public class ImGroupMemberServiceImpl implements ImGroupMemberService {
}
@Override
- public List getQuitGroupMemberListByUserId(Long userId, LocalDateTime minQuitTime) {
- return groupMemberMapper.selectQuitListByUserId(userId, minQuitTime);
+ public List getGroupMemberListByUserId(Long userId) {
+ return groupMemberMapper.selectListByUserId(userId);
}
@Override
@@ -259,7 +258,7 @@ public class ImGroupMemberServiceImpl implements ImGroupMemberService {
.setId(member.getId());
groupMemberMapper.updateById(updateObj);
- // 3.1 displayUserName 是公开字段，单独走 GROUP_MEMBER_NICKNAME_UPDATE 广播给全员；空串视为「清空昵称」也要广播；与旧值相同跳过
+ // 3.1 displayUserName 是公开字段，单独走 GROUP_MEMBER_NICKNAME_UPDATE 在线同步给全员；空串视为「清空昵称」也要同步；与旧值相同跳过
if (updateReqVO.getDisplayUserName() != null
&& ObjUtil.notEqual(updateReqVO.getDisplayUserName(), member.getDisplayUserName())) {
groupMessageService.sendGroupMessage(userId, ImGroupMessageSendDTO.ofGroupMemberNicknameUpdate(
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupRequestService.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupRequestService.java
index bd604f6..1c15d9b 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupRequestService.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupRequestService.java
@@ -64,6 +64,19 @@ public interface ImGroupRequestService {
List getUnhandledRequestListByOwnerOrAdmin(Long userId);
/**
+ * 增量拉取「我管理的群」下的加群申请（重连 / 离线补偿：含已处理，按 update_time + id 游标）
+ *
+ * 作用域与 {@link #getUnhandledRequestListByOwnerOrAdmin} 一致：按 ImGroupMember.role 取我作为 OWNER / ADMIN 的群
+ *
+ * @param userId 当前用户编号
+ * @param lastUpdateTime 游标：上次拉取的最后一条更新时间戳（毫秒）
+ * @param lastId 游标：上次拉取的最后一条申请编号
+ * @param limit 单次拉取条数
+ * @return 申请记录列表
+ */
+ List pullGroupRequestList(Long userId, Long lastUpdateTime, Long lastId, Integer limit);
+
+ /**
* 拉取指定群下的全部加群申请（含已处理）；仅群主 / 管理员可查
*
* 用于群「进群申请」子页：最新一条卡片化突出 + 历史申请按 id 倒序
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupRequestServiceImpl.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupRequestServiceImpl.java
index 71d3d76..40d1cad 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupRequestServiceImpl.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupRequestServiceImpl.java
@@ -13,15 +13,15 @@ import cn.iocoder.yudao.module.im.dal.mysql.group.ImGroupRequestMapper;
import cn.iocoder.yudao.module.im.enums.group.ImGroupAddSourceEnum;
import cn.iocoder.yudao.module.im.enums.group.ImGroupMemberRoleEnum;
import cn.iocoder.yudao.module.im.enums.group.ImGroupRequestHandleResultEnum;
-import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
import cn.iocoder.yudao.module.im.service.message.ImGroupMessageService;
import cn.iocoder.yudao.module.im.service.message.dto.ImGroupMessageSendDTO;
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.notification.group.BaseGroupNotification;
-import cn.iocoder.yudao.module.im.service.websocket.dto.notification.group.GroupRequestApprovedNotification;
-import cn.iocoder.yudao.module.im.service.websocket.dto.notification.group.GroupRequestReceivedNotification;
-import cn.iocoder.yudao.module.im.service.websocket.dto.notification.group.GroupRequestRejectedNotification;
+import cn.iocoder.yudao.module.im.service.websocket.notification.group.BaseGroupNotification;
+import cn.iocoder.yudao.module.im.service.websocket.notification.group.GroupRequestApprovedNotification;
+import cn.iocoder.yudao.module.im.service.websocket.notification.group.GroupRequestReceivedNotification;
+import cn.iocoder.yudao.module.im.service.websocket.notification.group.GroupRequestRejectedNotification;
import cn.iocoder.yudao.module.system.api.user.AdminUserApi;
import cn.iocoder.yudao.module.system.api.user.dto.AdminUserRespDTO;
import jakarta.annotation.Resource;
@@ -97,12 +97,12 @@ public class ImGroupRequestServiceImpl implements ImGroupRequestService {
// 3. 情况二：群开启了审批，创建或复用一条主动申请记录
ImGroupRequestDO request = createOrResetApplyRequest(groupId, userId, reqVO);
- // 4. 1503 私聊定向推群主 + 全部管理员（多端同步）；payload 携带申请方昵称 / 头像
+ // 4. 1503 定向推群主 + 全部管理员（多端同步）；payload 携带申请方昵称 / 头像
AdminUserRespDTO applyUser = adminUserApi.getUser(userId);
GroupRequestReceivedNotification payload = buildRequestNotification(group, request, applyUser);
for (Long receiverUserId : getGroupMemberListByOwnerAndAdminUserIds(group)) {
- websocketService.sendPrivateMessageAsync(receiverUserId, ImPrivateMessageDTO.ofGroupNotification(
- ImMessageTypeEnum.GROUP_REQUEST_RECEIVED.getType(), userId, receiverUserId, payload));
+ websocketService.sendNotificationAsync(receiverUserId, ImConversationTypeEnum.NONE.getType(),
+ ImContentTypeEnum.GROUP_REQUEST_RECEIVED.getType(), payload);
}
return request;
}
@@ -139,12 +139,12 @@ public class ImGroupRequestServiceImpl implements ImGroupRequestService {
groupMemberService.addGroupMember(request.getGroupId(), request.getUserId(),
ImGroupMemberRoleEnum.NORMAL.getRole(), request.getAddSource(), request.getInviterUserId());
- // 5.1 1505 私聊推送给申请人 + 群主 + 全部管理员（每端单推）
+ // 5.1 1505 定向推送给申请人 + 群主 + 全部管理员（每端单推）
GroupRequestApprovedNotification payload = (GroupRequestApprovedNotification) new GroupRequestApprovedNotification()
.setRequestId(request.getId()).setGroupId(request.getGroupId()).setUserId(request.getUserId())
.setOperatorUserId(userId);
broadcastToOwnerAdminsAndApplicant(request.getGroupId(), request.getUserId(), payload,
- ImMessageTypeEnum.GROUP_REQUEST_APPROVED.getType(), userId);
+ ImContentTypeEnum.GROUP_REQUEST_APPROVED.getType(), userId);
// 5.2 群事件：主动申请 → 1510 自由进群；被邀请 → 1509 成员加入
if (request.getInviterUserId() == null) {
groupMessageService.sendGroupMessage(userId,
@@ -174,12 +174,12 @@ public class ImGroupRequestServiceImpl implements ImGroupRequestService {
throw exception(GROUP_REQUEST_HANDLED);
}
- // 3. 1506 私聊推送给申请人 + 群主 + 全部管理员
+ // 3. 1506 定向推送给申请人 + 群主 + 全部管理员
GroupRequestRejectedNotification payload = (GroupRequestRejectedNotification) new GroupRequestRejectedNotification()
.setRequestId(request.getId()).setGroupId(request.getGroupId()).setUserId(request.getUserId())
.setHandleContent(handleContent).setOperatorUserId(userId);
broadcastToOwnerAdminsAndApplicant(request.getGroupId(), request.getUserId(), payload,
- ImMessageTypeEnum.GROUP_REQUEST_REJECTED.getType(), userId);
+ ImContentTypeEnum.GROUP_REQUEST_REJECTED.getType(), userId);
}
@Override
@@ -201,8 +201,8 @@ public class ImGroupRequestServiceImpl implements ImGroupRequestService {
AdminUserRespDTO applyUser = userMap.get(request.getUserId());
GroupRequestReceivedNotification payload = buildRequestNotification(group, request, applyUser);
for (Long receiverUserId : ownerAndAdmins) {
- websocketService.sendPrivateMessageAsync(receiverUserId, ImPrivateMessageDTO.ofGroupNotification(
- ImMessageTypeEnum.GROUP_REQUEST_RECEIVED.getType(), inviterUserId, receiverUserId, payload));
+ websocketService.sendNotificationAsync(receiverUserId, ImConversationTypeEnum.NONE.getType(),
+ ImContentTypeEnum.GROUP_REQUEST_RECEIVED.getType(), payload);
}
}
}
@@ -212,8 +212,7 @@ public class ImGroupRequestServiceImpl implements ImGroupRequestService {
// 1. 找出当前用户作为 OWNER / ADMIN 的所有群
List myMembers = groupMemberService.getActiveGroupMemberListByUserId(userId);
Set ownerOrAdminGroupIds = convertSet(myMembers,
- ImGroupMemberDO::getGroupId,
- m -> ImGroupMemberRoleEnum.isOwnerOrAdmin(m.getRole()));
+ ImGroupMemberDO::getGroupId, member -> ImGroupMemberRoleEnum.isOwnerOrAdmin(member.getRole()));
if (CollUtil.isEmpty(ownerOrAdminGroupIds)) {
return Collections.emptyList();
}
@@ -223,6 +222,19 @@ public class ImGroupRequestServiceImpl implements ImGroupRequestService {
}
@Override
+ public List pullGroupRequestList(Long userId, Long lastUpdateTime, Long lastId, Integer limit) {
+ // 1. 找出当前用户作为 OWNER / ADMIN 的所有群
+ List myMembers = groupMemberService.getActiveGroupMemberListByUserId(userId);
+ Set ownerOrAdminGroupIds = convertSet(myMembers,
+ ImGroupMemberDO::getGroupId, member -> ImGroupMemberRoleEnum.isOwnerOrAdmin(member.getRole()));
+ if (CollUtil.isEmpty(ownerOrAdminGroupIds)) {
+ return Collections.emptyList();
+ }
+ // 2. 按游标增量拉取这些群下的申请
+ return groupRequestMapper.selectPullListByGroupIds(ownerOrAdminGroupIds, lastUpdateTime, lastId, limit);
+ }
+
+ @Override
public List getGroupRequestListByGroupId(Long userId, Long groupId) {
// 1. 校验群存在 + 当前用户是群主 / 管理员
groupService.validateGroupExists(groupId);
@@ -383,7 +395,7 @@ public class ImGroupRequestServiceImpl implements ImGroupRequestService {
}
/**
- * 1505 / 1506 受众：申请人 + 群主 + 全部管理员；每端单独推一帧，前端按 receiver 是否申请人区分文案
+ * 1505 / 1506 受众：申请人 + 群主 + 全部管理员
*/
private void broadcastToOwnerAdminsAndApplicant(Long groupId, Long applicantUserId, BaseGroupNotification payload,
Integer messageType, Long operatorUserId) {
@@ -394,8 +406,8 @@ public class ImGroupRequestServiceImpl implements ImGroupRequestService {
Set receivers = new LinkedHashSet<>(getGroupMemberListByOwnerAndAdminUserIds(group));
receivers.add(applicantUserId);
for (Long receiverUserId : receivers) {
- websocketService.sendPrivateMessageAsync(receiverUserId, ImPrivateMessageDTO.ofGroupNotification(
- messageType, operatorUserId, receiverUserId, payload));
+ websocketService.sendNotificationAsync(receiverUserId, ImConversationTypeEnum.NONE.getType(),
+ messageType, payload);
}
}
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupServiceImpl.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupServiceImpl.java
index 61f0273..93fa996 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupServiceImpl.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupServiceImpl.java
@@ -30,7 +30,7 @@ import cn.iocoder.yudao.module.im.dal.mysql.group.ImGroupMapper;
import cn.iocoder.yudao.module.im.enums.group.ImGroupAddSourceEnum;
import cn.iocoder.yudao.module.im.enums.group.ImGroupMemberRoleEnum;
import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
-import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
import cn.iocoder.yudao.module.im.framework.config.ImProperties;
import cn.iocoder.yudao.module.im.service.friend.ImFriendService;
import cn.iocoder.yudao.module.im.service.message.ImGroupMessageService;
@@ -149,7 +149,9 @@ public class ImGroupServiceImpl implements ImGroupService {
boolean nameChanged = StrUtil.isNotEmpty(updateReqVO.getName());
boolean noticeChanged = updateReqVO.getNotice() != null;
boolean avatarChanged = StrUtil.isNotEmpty(updateReqVO.getAvatar());
- if (nameChanged || noticeChanged || avatarChanged) {
+ boolean joinApprovalChanged = updateReqVO.getJoinApproval() != null
+ && ObjUtil.notEqual(group.getJoinApproval(), updateReqVO.getJoinApproval());
+ if (nameChanged || noticeChanged || avatarChanged || joinApprovalChanged) {
List memberUserIds = groupMemberService.getActiveGroupMemberUserIdsByGroupId(groupId);
if (nameChanged) {
groupMessageService.sendGroupMessage(userId, memberUserIds, ImGroupMessageSendDTO.ofGroupNameUpdate(
@@ -161,7 +163,11 @@ public class ImGroupServiceImpl implements ImGroupService {
}
if (avatarChanged) {
groupMessageService.sendGroupMessage(userId, memberUserIds, ImGroupMessageSendDTO.ofGroupInfoUpdate(
- groupId, userId, group.getAvatar(), updateReqVO.getAvatar()));
+ groupId, userId, group.getAvatar(), updateReqVO.getAvatar(), null, null));
+ }
+ if (joinApprovalChanged) {
+ groupMessageService.sendGroupMessage(userId, memberUserIds, ImGroupMessageSendDTO.ofGroupInfoUpdate(
+ groupId, userId, null, null, group.getJoinApproval(), updateReqVO.getJoinApproval()));
}
}
@@ -193,8 +199,6 @@ public class ImGroupServiceImpl implements ImGroupService {
.setStatus(CommonStatusEnum.DISABLE.getStatus()).setDissolvedTime(LocalDateTime.now()));
// 2.2 移除全部群成员
groupMemberService.removeGroupMembersByGroupId(id);
- // 2.3 清理已读缓存
- groupMessageService.deleteReadMaxMessageIdMap(id);
}
// ==================== 群成员的写操作 ====================
@@ -267,10 +271,8 @@ public class ImGroupServiceImpl implements ImGroupService {
// 2. 先发广播，后移成员（见类 javadoc）
groupMessageService.sendGroupMessage(userId, ImGroupMessageSendDTO.ofGroupMemberQuit(groupId, userId));
- // 3.1 移除群成员
+ // 3. 移除群成员
groupMemberService.removeGroupMember(groupId, userId);
- // 3.2 清理已读缓存
- groupMessageService.deleteReadMaxMessageId(groupId, userId);
}
@Override
@@ -307,10 +309,8 @@ public class ImGroupServiceImpl implements ImGroupService {
groupMessageService.sendGroupMessage(userId,
ImGroupMessageSendDTO.ofGroupMemberKick(groupId, userId, validTargetUserIds));
- // 3.1 批量移除群成员
+ // 3. 批量移除群成员；不清理读位置（保留退群前历史已读，供离线补偿）
groupMemberService.removeGroupMembers(groupId, validTargetUserIds);
- // 3.2 批量清理已读缓存
- groupMessageService.deleteReadMaxMessageIds(groupId, validTargetUserIds);
}
@Override
@@ -442,7 +442,7 @@ public class ImGroupServiceImpl implements ImGroupService {
if (message == null || ObjUtil.notEqual(message.getGroupId(), groupId)) {
throw exception(MESSAGE_NOT_IN_GROUP);
}
- if (!ImMessageTypeEnum.validate(message.getType()).isNormal()
+ if (!ImContentTypeEnum.validate(message.getType()).isNormal()
|| ImMessageStatusEnum.RECALL.getStatus().equals(message.getStatus())) {
throw exception(MESSAGE_NOT_IN_GROUP);
}
@@ -601,11 +601,8 @@ public class ImGroupServiceImpl implements ImGroupService {
@Override
public List getMyGroupList(Long userId) {
- // 1.1 查用户所在的、仍有效的群成员记录（仅 ENABLE 状态）
- List members = groupMemberService.getActiveGroupMemberListByUserId(userId);
- // 1.2 再查最近 N 天（与群消息离线拉取窗口一致）内退群的成员记录（退群前可能有离线消息需要展示，一并返回作为前端缓存）
- LocalDateTime minQuitTime = LocalDateTime.now().minusDays(imProperties.getMessage().getGroupPullMaxDays());
- members.addAll(groupMemberService.getQuitGroupMemberListByUserId(userId, minQuitTime));
+ // 1. 查用户曾经加入的所有群（含退群，前端按需过滤）；退群前的离线消息也要能展示
+ List members = groupMemberService.getGroupMemberListByUserId(userId);
if (CollUtil.isEmpty(members)) {
return Collections.emptyList();
}
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImChannelMessageService.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImChannelMessageService.java
index 8b05161..b1e00ee 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImChannelMessageService.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImChannelMessageService.java
@@ -27,7 +27,7 @@ public interface ImChannelMessageService {
* @param size 返回条数
* @return 频道消息列表；按 id 升序
*/
- List getMessageListForPull(Long userId, Long minId, Integer size);
+ List pullChannelMessageList(Long userId, Long minId, Integer size);
/**
* 上报频道消息已读位置；同步推 READ 事件给自己多端
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImChannelMessageServiceImpl.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImChannelMessageServiceImpl.java
index cbdf710..fc9ad8e 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImChannelMessageServiceImpl.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImChannelMessageServiceImpl.java
@@ -2,6 +2,7 @@ package cn.iocoder.yudao.module.im.service.message;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.Assert;
+import cn.hutool.core.util.ObjUtil;
import cn.hutool.extra.spring.SpringUtil;
import cn.iocoder.yudao.framework.common.pojo.PageResult;
import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
@@ -11,12 +12,14 @@ import cn.iocoder.yudao.module.im.controller.admin.manager.message.vo.channel.Im
import cn.iocoder.yudao.module.im.dal.dataobject.channel.ImChannelMaterialDO;
import cn.iocoder.yudao.module.im.dal.dataobject.message.ImChannelMessageDO;
import cn.iocoder.yudao.module.im.dal.mysql.message.ImChannelMessageMapper;
-import cn.iocoder.yudao.module.im.dal.redis.message.ImChannelMessageReadRedisDAO;
-import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
import cn.iocoder.yudao.module.im.service.channel.ImChannelMaterialService;
+import cn.iocoder.yudao.module.im.service.conversation.ImConversationReadService;
import cn.iocoder.yudao.module.im.service.websocket.ImWebSocketService;
-import cn.iocoder.yudao.module.im.service.websocket.dto.ImChannelMessageDTO;
-import cn.iocoder.yudao.module.im.service.websocket.dto.message.MaterialMessage;
+import cn.iocoder.yudao.module.im.service.websocket.notification.message.ImChannelMessageNotification;
+import cn.iocoder.yudao.module.im.service.websocket.notification.message.ImMessageReadNotification;
+import cn.iocoder.yudao.module.im.dal.dataobject.message.content.MaterialMessage;
import jakarta.annotation.Resource;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
@@ -25,7 +28,6 @@ import org.springframework.validation.annotation.Validated;
import java.time.LocalDateTime;
import java.util.Collection;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -50,39 +52,41 @@ public class ImChannelMessageServiceImpl implements ImChannelMessageService {
private ImWebSocketService webSocketService;
@Resource
- private ImChannelMessageReadRedisDAO channelMessageReadRedisDAO;
+ private ImConversationReadService conversationReadService;
// ==================== 用户端 ====================
@Override
- public List getMessageListForPull(Long userId, Long minId, Integer size) {
+ public List pullChannelMessageList(Long userId, Long minId, Integer size) {
return channelMessageMapper.selectListByUserAndMinId(userId, minId, size);
}
@Override
public Map getChannelReadMaxMessageIdMap(Long userId, Collection channelIds) {
- Map result = new HashMap<>(channelIds.size());
- for (Long channelId : channelIds) {
- Long max = channelMessageReadRedisDAO.getReadMaxMessageId(channelId, userId);
- if (max != null) {
- result.put(channelId, max);
- }
- }
- return result;
+ return conversationReadService.getConversationReadMessageIdMap(
+ userId, ImConversationTypeEnum.CHANNEL.getType(), channelIds);
}
@Override
public void readChannelMessages(Long userId, Long channelId, Long messageId) {
Assert.notNull(channelId, "频道编号不能为空");
Assert.notNull(messageId, "已读消息编号不能为空");
- // 1. 已读位置未前进，直接返回
- Long prevMaxMessageId = channelMessageReadRedisDAO.getReadMaxMessageId(channelId, userId);
- if (prevMaxMessageId != null && prevMaxMessageId >= messageId) {
+ // 1.1 校验消息真实存在且属于该频道，避免未来 / 伪造 messageId 污染读位置
+ ImChannelMessageDO message = channelMessageMapper.selectById(messageId);
+ if (message == null || ObjUtil.notEqual(message.getChannelId(), channelId)) {
+ return;
+ }
+ // 1.2 定向消息校验对当前用户可见（receiver_user_ids 为空表示全员可见）
+ if (CollUtil.isNotEmpty(message.getReceiverUserIds()) && !message.getReceiverUserIds().contains(userId)) {
return;
}
- // 2. 更新 Redis 频道已读位置
- channelMessageReadRedisDAO.updateReadMaxMessageId(channelId, userId, messageId);
+ // 2. 更新频道已读位置；读位置未前进则不推
+ boolean advanced = conversationReadService.updateConversationReadPosition(
+ userId, ImConversationTypeEnum.CHANNEL.getType(), channelId, messageId);
+ if (!advanced) {
+ return;
+ }
// 3. 异步推 READ 事件给自己多端同步
getSelf().readChannelMessageEvent(userId, channelId, messageId);
@@ -93,7 +97,8 @@ public class ImChannelMessageServiceImpl implements ImChannelMessageService {
*/
@Async
public void readChannelMessageEvent(Long userId, Long channelId, Long readId) {
- webSocketService.sendChannelMessageAsync(userId, ImChannelMessageDTO.ofRead(channelId, readId));
+ webSocketService.sendNotificationAsync(userId, ImConversationTypeEnum.CHANNEL.getType(),
+ ImContentTypeEnum.READ.getType(), ImMessageReadNotification.ofChannel(channelId, readId));
}
private ImChannelMessageServiceImpl getSelf() {
@@ -113,15 +118,16 @@ public class ImChannelMessageServiceImpl implements ImChannelMessageService {
String payloadJson = JsonUtils.toJsonString(payload);
// 2.2 落库 1 行 message；reqVO 同名字段（materialId / receiverUserIds）自动拷贝，剩余字段补 set
ImChannelMessageDO message = BeanUtils.toBean(reqVO, ImChannelMessageDO.class).setChannelId(material.getChannelId())
- .setType(ImMessageTypeEnum.MATERIAL.getType()).setContent(payloadJson).setSendTime(LocalDateTime.now());
+ .setType(ImContentTypeEnum.MATERIAL.getType()).setContent(payloadJson).setSendTime(LocalDateTime.now());
channelMessageMapper.insert(message);
// 3. 异步推 WebSocket：指定用户走点对点；全员（receiverUserIds 为空）走广播
- ImChannelMessageDTO dto = ImChannelMessageDTO.ofSend(message);
+ ImChannelMessageNotification dto = ImChannelMessageNotification.ofSend(message);
if (CollUtil.isNotEmpty(reqVO.getReceiverUserIds())) {
- webSocketService.sendChannelMessageAsync(reqVO.getReceiverUserIds(), dto);
+ webSocketService.sendNotificationAsync(reqVO.getReceiverUserIds(), ImConversationTypeEnum.CHANNEL.getType(),
+ dto.getType(), dto);
} else {
- webSocketService.broadcastChannelMessageAsync(dto);
+ webSocketService.broadcastNotificationAsync(ImConversationTypeEnum.CHANNEL.getType(), dto.getType(), dto);
}
return message.getId();
}
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageService.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageService.java
index 51a28f1..77426d1 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageService.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageService.java
@@ -98,35 +98,6 @@ public interface ImGroupMessageService {
*/
List getGroupMessageList(Long userId, ImGroupMessageListReqVO reqVO);
- /**
- * 清理用户在某群的已读位置缓存
- *
- * 用于成员退群场景
- *
- * @param groupId 群编号
- * @param userId 用户编号
- */
- void deleteReadMaxMessageId(Long groupId, Long userId);
-
- /**
- * 批量清理用户在某群的已读位置缓存
- *
- * 用于批量踢出场景
- *
- * @param groupId 群编号
- * @param userIds 用户编号集合
- */
- void deleteReadMaxMessageIds(Long groupId, Collection userIds);
-
- /**
- * 清理某群所有用户的已读位置缓存
- *
- * 用于群解散场景
- *
- * @param groupId 群编号
- */
- void deleteReadMaxMessageIdMap(Long groupId);
-
// ==================== 管理后台 ====================
/**
diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java
index 2ed32cd..12914fb 100644
--- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java
+++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java
@@ -6,7 +6,6 @@ import cn.hutool.core.util.BooleanUtil;
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.pojo.PageResult;
import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
@@ -17,23 +16,27 @@ import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupDO;
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.mysql.message.ImGroupMessageMapper;
-import cn.iocoder.yudao.module.im.dal.redis.message.ImGroupMessageReadRedisDAO;
+import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
import cn.iocoder.yudao.module.im.enums.group.ImGroupMemberRoleEnum;
-import cn.iocoder.yudao.module.im.enums.message.ImGroupMessageReceiptStatusEnum;
+import cn.iocoder.yudao.module.im.enums.message.ImMessageReceiptStatusEnum;
import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
-import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
+import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
import cn.iocoder.yudao.module.im.framework.config.ImProperties;
+import cn.iocoder.yudao.module.im.service.conversation.ImConversationReadService;
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.message.dto.ImGroupMessageSendDTO;
import cn.iocoder.yudao.module.im.service.sensitiveword.ImSensitiveWordService;
import cn.iocoder.yudao.module.im.service.websocket.ImWebSocketService;
-import cn.iocoder.yudao.module.im.service.websocket.dto.ImGroupMessageDTO;
-import cn.iocoder.yudao.module.im.service.websocket.dto.message.QuoteMessage;
-import cn.iocoder.yudao.module.im.service.websocket.dto.message.RecallMessage;
+import cn.iocoder.yudao.module.im.service.websocket.notification.message.ImGroupMessageNotification;
+import cn.iocoder.yudao.module.im.service.websocket.notification.message.ImMessageReadNotification;
+import cn.iocoder.yudao.module.im.service.websocket.notification.message.ImMessageReceiptNotification;
+import cn.iocoder.yudao.module.im.dal.dataobject.message.content.QuoteMessage;
+import cn.iocoder.yudao.module.im.dal.dataobject.message.content.RecallMessage;
import cn.iocoder.yudao.module.im.util.ImMessageUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
+import org.springframework.dao.DuplicateKeyException;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@@ -41,7 +44,6 @@ import org.springframework.validation.annotation.Validated;
import java.time.LocalDateTime;
import java.util.*;
-import java.util.stream.Collectors;
import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.*;
@@ -57,17 +59,8 @@ import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.*;
@Slf4j
public class ImGroupMessageServiceImpl implements ImGroupMessageService {
- /**
- * 仅用于规避群消息 pull 的"假空页"：
- * 首批消息可能因入群时间或定向接收过滤后变成空列表，但后续更大的 id 仍然存在可见消息。
- * 因此仅在过滤结果为空时，按本轮消息最大 id 向后再试几次。
- */
- private static final int PULL_GROUP_MESSAGE_EMPTY_RETRY_TIMES = 3;
-
@Resource
private ImGroupMessageMapper groupMessageMapper;
- @Resource
- private ImGroupMessageReadRedisDAO groupMessageReadRedisDAO;
@Resource
private ImGroupService groupService;
@@ -75,6 +68,8 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
private ImGroupMemberService groupMemberService;
@Resource
private ImSensitiveWordService sensitiveWordService;
+ @Resource
+ private ImConversationReadService conversationReadService;
@Resource
private ImWebSocketService imWebSocketService;
@@ -100,34 +95,44 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
// 1.4 禁言校验
validateMuteStatus(group, senderMember);
// 1.5 文本消息敏感词过滤
- if (ImMessageTypeEnum.TEXT.getType().equals(reqVO.getType())) {
+ if (ImContentTypeEnum.TEXT.getType().equals(reqVO.getType())) {
sensitiveWordService.validateText(reqVO.getContent());
}
- // 2.1 引用 quote 消息规范化
- reqVO.setContent(normalizeQuoteContent(reqVO, senderMember));
- // 2.2 构建并保存消息
+ // 2.1 固化发送当时可见成员快照：用户发送均为全员广播；getReceiverUserIds 兜底纳入发送者，钉死发送者必可见
+ List memberUserIds = groupMemberService.getActiveGroupMemberUserIdsByGroupId(reqVO.getGroupId());
+ Set receiverUserIds = getReceiverUserIds(null, senderId, memberUserIds);
+ // 2.2 引用 quote 消息规范化（含可见性子集校验，防止定向消息内容被广播引用泄漏）
+ reqVO.setContent(normalizeQuoteContent(reqVO, senderMember, receiverUserIds));
+ // 2.3 构建并保存消息；唯一键冲突时回查已存在消息返回
ImGroupMessageDO message = BeanUtils.toBean(reqVO, ImGroupMessageDO.class, m -> m
- .setSenderId(senderId).setStatus(ImMessageStatusEnum.UNREAD.getStatus()).setSendTime(LocalDateTime.now())
+ .setSenderId(senderId).setStatus(ImMessageStatusEnum.NORMAL.getStatus()).setSendTime(LocalDateTime.now())
+ .setReceiverUserIds(new ArrayList<>(receiverUserIds))
.setReceiptStatus(resolveReceiptStatus(reqVO.getReceipt())));
- groupMessageMapper.insert(message);
+ try {
+ groupMessageMapper.insert(message);
+ } catch (DuplicateKeyException e) {
+ log.warn("[sendGroupMessage][senderId({}) clientMessageId({}) 并发插入冲突，回查返回]",
+ senderId, reqVO.getClientMessageId());
+ return groupMessageMapper.selectBySenderIdAndClientMessageId(senderId, reqVO.getClientMessageId());
+ }
- // 3. WebSocket 异步推送（群内可见成员 + 发送方多端同步）
- List memberUserIds = groupMemberService.getActiveGroupMemberUserIdsByGroupId(message.getGroupId());
- Set targetUserIds = getVisibleUserIds(message.getReceiverUserIds(), senderId, memberUserIds);
- imWebSocketService.sendGroupMessageAsync(targetUserIds, ImGroupMessageDTO.ofSend(message));
+ // 3. WebSocket 异步推送（快照内可见成员 + 发送方多端同步）
+ ImGroupMessageNotification notification = ImGroupMessageNotification.ofSend(message);
+ imWebSocketService.sendNotificationAsync(receiverUserIds, ImConversationTypeEnum.GROUP.getType(),
+ notification.getType(), notification);
return message;
}
@Override
public ImGroupMessageDO sendGroupMessage(Long senderId, ImGroupMessageSendDTO dto) {
List memberUserIds = groupMemberService.getActiveGroupMemberUserIdsByGroupId(dto.getGroupId());
- Set targetUserIds = getVisibleUserIds(dto.getReceiverUserIds(), senderId, memberUserIds);
- return sendGroupMessage(senderId, targetUserIds, dto);
+ Set receiverUserIds = getReceiverUserIds(dto.getReceiverUserIds(), senderId, memberUserIds);
+ return sendGroupMessage(senderId, receiverUserIds, dto);
}
@Override
- public ImGroupMessageDO sendGroupMessage(Long senderId, Collection targetUserIds, ImGroupMessageSendDTO dto) {
+ public ImGroupMessageDO sendGroupMessage(Long senderId, Collection receiverUserIds, ImGroupMessageSendDTO dto) {
// 1.1 content 序列化：null / String 透传，POJO 走 JSON
Object payload = dto.getContent();
String contentString = payload == null || payload instanceof String
@@ -137,16 +142,18 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
ImGroupMessageDO message = new ImGroupMessageDO().setClientMessageId(IdUtil.fastSimpleUUID())
.setSenderId(senderId).setGroupId(dto.getGroupId())
.setType(dto.getType()).setContent(contentString)
- .setStatus(ImMessageStatusEnum.UNREAD.getStatus()).setSendTime(LocalDateTime.now())
- .setAtUserIds(dto.getAtUserIds()).setReceiverUserIds(dto.getReceiverUserIds())
+ .setStatus(ImMessageStatusEnum.NORMAL.getStatus()).setSendTime(LocalDateTime.now())
+ .setAtUserIds(dto.getAtUserIds()).setReceiverUserIds(new ArrayList<>(receiverUserIds))
.setReceiptStatus(resolveReceiptStatus(dto.getReceipt()));
// 1.3 按 type.persistent 决定是否入库
- if (ImMessageTypeEnum.validate(dto.getType()).isPersistent()) {
+ if (ImContentTypeEnum.validate(dto.getType()).isPersistent()) {
groupMessageMapper.insert(message);
}
// 2. WebSocket 异步推送
- imWebSocketService.sendGroupMessageAsync(targetUserIds, ImGroupMessageDTO.ofSend(message));
+ ImGroupMessageNotification notification = ImGroupMessageNotification.ofSend(message);
+ imWebSocketService.sendNotificationAsync(receiverUserIds, ImConversationTypeEnum.GROUP.getType(),
+ notification.getType(), notification);
return message;
}
@@ -180,7 +187,7 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
// 3. 发送撤回事件
return sendGroupMessage(userId, new ImGroupMessageSendDTO().setGroupId(message.getGroupId())
- .setType(ImMessageTypeEnum.RECALL.getType())
+ .setType(ImContentTypeEnum.RECALL.getType())
.setReceiverUserIds(message.getReceiverUserIds())
.setContent(new RecallMessage().setMessageId(messageId)));
}
@@ -195,155 +202,53 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
// 0. 拉取时间窗；超过窗口的老消息不再通过离线通道推送
LocalDateTime minSendTime = LocalDateTime.now().minusDays(imProperties.getMessage().getGroupPullMaxDays());
- // 1.1 主查询：仅用"当前仍在群"的成员记录驱动；若首批消息过滤后为空，则允许内部重试
- List activeMembers = groupMemberService.getActiveGroupMemberListByUserId(userId);
- Map memberMap = convertMap(activeMembers, ImGroupMemberDO::getGroupId);
- List messages = new ArrayList<>();
- if (CollUtil.isNotEmpty(activeMembers)) {
- List groupIds = convertList(activeMembers, ImGroupMemberDO::getGroupId);
- messages.addAll(pullActiveGroupMessageList(userId, groupIds, minId, size, minSendTime, memberMap));
- }
- // 1.2 补齐"退群前"的消息：
- // - 继续基于用户原始 minId 单次查询，不能被主查询内部探测游标带着前进；
- // - 若 minId > 0 且能查到对应消息，则进一步把下限抬到该消息的 sendTime，避免把客户端已拥有的老消息再次推送。
- messages.addAll(pullQuitGroupMessageList(userId, minId, size, minSendTime, memberMap));
-
- // 2. 合并后统一过滤，得到当前用户可见的结果
- List result = filterGroupMessageList(messages, memberMap, userId, size);
-
- // 3. 按当前用户补齐：消息已读态（相对 Redis 已读游标）、本人发送的回执消息的已读人数
- appendMessageStatusAndReceipt(userId, result);
- log.info("[pullGroupMessageList][userId({}) minId({}) size({}) result({})]", userId, minId, size, result.size());
- return result;
- }
-
- /**
- * 拉取当前仍在群的主路径消息。
- *
- * 仅当首批消息过滤后仍无可见消息时，才按消息最大 id 继续向后探测，
- * 直到命中可见消息或确认该来源已耗尽。
- */
- private List pullActiveGroupMessageList(Long userId, List groupIds, Long minId,
- Integer size, LocalDateTime minSendTime,
- Map memberMap) {
- // 1. 主查询内部探测游标：仅用于向后探测，不代表客户端真实已送达边界
- Long activeMinId = minId;
- for (int retryCount = 0; retryCount <= PULL_GROUP_MESSAGE_EMPTY_RETRY_TIMES; retryCount++) {
- // 2. 查询本轮消息；若已无更多消息，则当前主路径直接结束
- List messages = groupMessageMapper.selectListByMinId(groupIds, activeMinId,
- minSendTime, size);
- if (CollUtil.isEmpty(messages)) {
- return Collections.emptyList();
- }
- boolean hasVisibleMessage = CollUtil.anyMatch(messages,
- message -> isMessageVisible(message, memberMap.get(message.getGroupId()), userId));
- boolean sourceExhausted = messages.size() < size;
- // 3. 本轮已命中可见消息，或主查询来源已耗尽，直接返回这一轮消息
- if (hasVisibleMessage || sourceExhausted) {
- return messages;
- }
-
- // 4. 按本轮消息最大 id 推进内部游标，跳过这段不可见区间；若游标未前进则直接停止
- Long maxMessageId = getMaxValue(messages, ImGroupMessageDO::getId);
- if (maxMessageId == null || maxMessageId <= activeMinId) {
- return Collections.emptyList();
- }
- activeMinId = maxMessageId;
+ // 1. 候选群 = 用户曾经加入的所有群（含退群）；可见性与时间窗都交给 SQL，退群群最多贡献 0 条
+ Set groupIds = convertSet(groupMemberService.getGroupMemberListByUserId(userId),
+ ImGroupMemberDO::getGroupId);
+ if (CollUtil.isEmpty(groupIds)) {
+ return Collections.emptyList();
}
- return Collections.emptyList();
- }
- /**
- * 拉取已离开群路径的消息（退群前消息）
- */
- private List pullQuitGroupMessageList(Long userId, Long minId, Integer size,
- LocalDateTime minSendTime,
- Map memberMap) {
- // 1. 退群补齐始终基于用户原始 minId 计算时间边界，避免被主查询内部重试游标误伤
- LocalDateTime minQuitTime = minSendTime;
- if (minId != null && minId > 0) {
- ImGroupMessageDO minMessage = groupMessageMapper.selectById(minId);
- if (minMessage != null && minMessage.getSendTime() != null
- && minMessage.getSendTime().isAfter(minSendTime)) {
- minQuitTime = minMessage.getSendTime();
- }
- }
- // 2. 查询用户离开的群记录；若原始 minId 对应消息仍在窗口内，则用它的发送时间抬升退群筛选下限
- List messages = new ArrayList<>();
- List quitMembers = groupMemberService.getQuitGroupMemberListByUserId(userId, minQuitTime);
- for (ImGroupMemberDO quitMember : quitMembers) {
- // 3. 按原始 minId + 退群时间补齐该群退群前消息，并把成员记录写回 memberMap 供统一可见性过滤使用
- List quitGroupMessages = groupMessageMapper.selectListByGroupIdAndMinIdAndQuitTimeBefore(
- quitMember.getGroupId(), minId, minSendTime, quitMember.getQuitTime(), size);
- if (CollUtil.isEmpty(quitGroupMessages)) {
- continue;
- }
- messages.addAll(quitGroupMessages);
- memberMap.put(quitMember.getGroupId(), quitMember);
- }
- return messages;
- }
+ // 2. 按 receiver_user_ids 快照过滤可见性，结果按 id 升序
+ List messages = groupMessageMapper.selectListByMinId(userId, groupIds, minId, minSendTime, size);
- /**
- * 过滤一批原始群消息，得到当前用户可见的返回结果
- */
- private List filterGroupMessageList(List messages,
- Map memberMap,
- Long userId, Integer size) {
- // 按可见性过滤（入群前不可见、定向消息排除），按 id 升序后仅取本页 size 条，
- // 避免「在群 + 退群前」多路合并时一次响应跨度过大、游标直接跳到全局最大 id 而漏拉中间消息
- return messages.stream()
- .filter(msg -> isMessageVisible(msg, memberMap.get(msg.getGroupId()), userId))
- .sorted(Comparator.comparing(ImGroupMessageDO::getId))
- .limit(size)
- .collect(Collectors.toList());
+ // 3. 补齐本人发送的回执消息的已读人数
+ appendMessageReceipt(userId, messages);
+ log.info("[pullGroupMessageList][userId({}) minId({}) size({}) result({})]", userId, minId, size, messages.size());
+ return messages;
}
/**
- * 补全消息已读态和回执已读人数
- *
- * 1. 消息已读态（status）：根据 Redis 已读游标判断 READ / UNREAD
- * 2. 回执已读人数（readCount）：仅对本人发送的回执消息，计算可见成员中的已读人数
+ * 补全本人发送消息的回执已读人数（readCount）
+ *
+ * 仅对本人发送、且需要回执（非 NO_RECEIPT）的消息，按 receiver_user_ids 快照 ∩ 当前有效成员 - 发送者 算已读人数
*/
- @SuppressWarnings({"StatementWithEmptyBody", "DataFlowIssue"})
- private void appendMessageStatusAndReceipt(Long userId, List messages) {
+ @SuppressWarnings("DataFlowIssue")
+ private void appendMessageReceipt(Long userId, List messages) {
if (CollUtil.isEmpty(messages)) {
return;
}
- // 群已读关闭：不查 Redis 已读游标，status 保持 DB 原值（含撤回），readCount 不补齐
+ // 群已读关闭：不补 readCount
if (BooleanUtil.isFalse(imProperties.getMessage().isGroupReadEnabled())) {
return;
}
- Map readMaxMessageIdsByGroup = new HashMap<>(); // 群 → 已读位置
Map> readPositionsByGroup = new HashMap<>(); // 群 → (用户 → 已读位置)
Map> membersByGroup = new HashMap<>(); // 群 → 全部成员列表
for (ImGroupMessageDO message : messages) {
- // 消息已读态（status）：撤回 > 已读 > 未读
- Long groupId = message.getGroupId();
- long readMaxMessageId = readMaxMessageIdsByGroup.computeIfAbsent(groupId, gid -> {
- Long readMaxMsgId = groupMessageReadRedisDAO.getReadMaxMessageId(gid, userId);
- return readMaxMsgId != null ? readMaxMsgId : -1L;
- });
- if (ImMessageStatusEnum.RECALL.getStatus().equals(message.getStatus())) {
- // 保持撤回态
- } else if (readMaxMessageId >= message.getId()) {
- message.setStatus(ImMessageStatusEnum.READ.getStatus());
- } else {
- message.setStatus(ImMessageStatusEnum.UNREAD.getStatus());
- }
-
- // 回执消息的已读人数（readCount）：仅补齐本人发送的，其他消息不处理（回执消息才关心已读人数，且只对发送者可见）
+ // 仅补本人发送、且需要回执（非 NO_RECEIPT）的消息
if (ObjUtil.notEqual(message.getSenderId(), userId)
- || ImGroupMessageReceiptStatusEnum.NO_RECEIPT.getStatus().equals(message.getReceiptStatus())) {
+ || ImMessageReceiptStatusEnum.NO_RECEIPT.getStatus().equals(message.getReceiptStatus())) {
continue;
}
+ Long groupId = message.getGroupId();
Map positions = readPositionsByGroup.computeIfAbsent(groupId,
- gid -> groupMessageReadRedisDAO.getReadMaxMessageIdMap(gid));
- List allMembers = membersByGroup.computeIfAbsent(groupId,
- groupMemberService::getGroupMemberListByGroupId);
- Set visibleUserIds = getVisibleUserIds(message, allMembers);
- visibleUserIds.remove(message.getSenderId());
- int readCount = getSumValue(visibleUserIds,
+ gid -> conversationReadService.getUserReadMessageIdMap(ImConversationTypeEnum.GROUP.getType(), gid));
+ // 应读分母取当前有效成员（剔除已退群），与异步回执刷新、前端已读弹层口径一致
+ List activeMembers = membersByGroup.computeIfAbsent(groupId,
+ groupMemberService::getActiveGroupMemberListByGroupId);
+ Set receiverUserIds = getReceiverUserIds(message, activeMembers);
+ receiverUserIds.remove(message.getSenderId());
+ int readCount = getSumValue(receiverUserIds,
uid -> positions.getOrDefault(uid, -1L) >= message.getId() ? 1 : null,
Integer::sum, 0);
message.setReadCount(readCount);
@@ -357,25 +262,24 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
throw exception(MESSAGE_GROUP_READ_DISABLED);
}
Assert.notNull(messageId, "已读消息编号不能为空");
- // 1.1 获取群成员记录
- ImGroupMemberDO member = groupMemberService.getGroupMember(groupId, userId);
- // 1.2 校验消息属于当前群，且对当前用户可见
+ // 1. 校验消息属于当前群，且对当前用户可见（按快照）
ImGroupMessageDO message = groupMessageMapper.selectById(messageId);
if (message == null
|| ObjUtil.notEqual(message.getGroupId(), groupId)
- || !isMessageVisible(message, member, userId)) {
+ || !isMessageReceived(message, userId)) {
throw exception(MESSAGE_NOT_IN_GROUP);
}
- // 2. 已读位置未前进，直接返回
- Long prevMaxMessageId = groupMessageReadRedisDAO.getReadMaxMessageId(groupId, userId);
- if (prevMaxMessageId != null && prevMaxMessageId >= messageId) {
+ // 2. 取旧读位置（用于回执刷新区间），再更新群已读位置；读位置未前进则不推
+ Long prevMaxMessageId = conversationReadService.getConversationReadMessageId(
+ userId, ImConversationTypeEnum.GROUP.getType(), groupId);
+ boolean advanced = conversationReadService.updateConversationReadPosition(
+ userId, ImConversationTypeEnum.GROUP.getType(), groupId, messageId);
+ if (!advanced) {
return;
}
- // 3. 更新 Redis 群已读位置
- groupMessageReadRedisDAO.updateReadMaxMessageId(groupId, userId, messageId);
- // 4. 异步发送 READ 事件 + 刷新范围内的群回执
+ // 3. 异步发送 READ 事件 + 刷新范围内的群回执
getSelf().readGroupMessageEvent(userId, groupId, prevMaxMessageId, messageId);
}
@@ -386,13 +290,13 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
throw exception(MESSAGE_GROUP_READ_DISABLED);
}
// 1.1 校验用户在群中（权限校验）
- ImGroupMemberDO operator = groupMemberService.validateMemberInGroup(groupId, userId);
+ groupMemberService.validateMemberInGroup(groupId, userId);
// 1.2 获取消息；并校验消息归属于该群、对调用者可见、调用者是发送方
ImGroupMessageDO message = groupMessageMapper.selectById(messageId);
if (message == null || ObjUtil.notEqual(message.getGroupId(), groupId)) {
return Collections.emptyList();
}
- if (!isMessageVisible(message, operator, userId)) {
+ if (!isMessageReceived(message, userId)) {
return Collections.emptyList();
}
// 1.3 仅消息发送方关心已读人数；非发送方查询直接返回空
@@ -400,18 +304,19 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
return Collections.emptyList();
}
- // 2. 获取所有成员和已读位置
- List allMembers = groupMemberService.getGroupMemberListByGroupId(groupId);
- Map allPositions = groupMessageReadRedisDAO.getReadMaxMessageIdMap(groupId);
+ // 2. 获取当前有效成员和已读位置（剔除已退群，与回执分母口径一致）
+ List activeMembers = groupMemberService.getActiveGroupMemberListByGroupId(groupId);
+ Map allPositions = conversationReadService.getUserReadMessageIdMap(
+ ImConversationTypeEnum.GROUP.getType(), groupId);
// 3. 计算该消息的可见成员集合（排除发送者自己）
- Set visibleUserIds = getVisibleUserIds(message, allMembers);
- visibleUserIds.remove(message.getSenderId());
+ Set receiverUserIds = getReceiverUserIds(message, activeMembers);
+ receiverUserIds.remove(message.getSenderId());
// 4. 只返回在可见范围内且已读位置 >= messageId 的用户
List readUserIds = new ArrayList<>();
allPositions.forEach((uid, readMaxMessageId) -> {
- if (visibleUserIds.contains(uid) && readMaxMessageId >= messageId) {
+ if (receiverUserIds.contains(uid) && readMaxMessageId >= messageId) {
readUserIds.add(uid);
}
});
@@ -420,15 +325,14 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
@Override
public List getGroupMessageList(Long userId, ImGroupMessageListReqVO reqVO) {
- // 1. 校验用户在群中
- ImGroupMemberDO member = groupMemberService.validateMemberInGroup(reqVO.getGroupId(), userId);
-
- // 2. 查询历史消息（仅入群之后）
- List messages = groupMessageMapper.selectHistoryList(
- reqVO.getGroupId(), reqVO.getMaxId(), reqVO.getLimit(), member.getJoinTime());
+ // 1. 校验用户曾经在群（当前在群或已退群），与 pull 退群窗口口径一致；内容仍由 receiver_user_ids 快照过滤
+ if (groupMemberService.getGroupMember(reqVO.getGroupId(), userId) == null) {
+ throw exception(GROUP_MEMBER_NOT_IN_GROUP);
+ }
- // 3. 过滤定向消息：仅保留当前用户可见的（receiverUserIds 为空 / 含当前用户 / 本人发送）
- return filterList(messages, message -> isMessageVisible(message, member, userId));
+ // 2. 查询历史消息：SQL 已按 receiver_user_ids 快照过滤当前用户可见
+ return groupMessageMapper.selectHistoryListByUser(
+ userId, reqVO.getGroupId(), reqVO.getMaxId(), reqVO.getLimit());
}
// ========== 异步 WebSocket 推送 ==========
@@ -445,8 +349,9 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
@SuppressWarnings("DataFlowIssue")
public void readGroupMessageEvent(Long userId, Long groupId, Long prevMaxMessageId, Long newMaxMessageId) {
// 1. 发送 READ 事件给自己的其他终端（多端同步）
- imWebSocketService.sendGroupMessageAsync(userId,
- ImGroupMessageDTO.ofRead(userId, groupId, newMaxMessageId));
+ imWebSocketService.sendNotificationAsync(userId, ImConversationTypeEnum.GROUP.getType(),
+ ImContentTypeEnum.READ.getType(),
+ ImMessageReadNotification.ofGroup(userId, groupId, newMaxMessageId));
// 2. 刷新 (prevMaxMessageId, newMaxMessageId] 区间内的待回执消息
List pendingMessages = groupMessageMapper.selectListByGroupIdAndPendingReceipt(
@@ -455,28 +360,30 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
return;
}
List activeMembers = groupMemberService.getActiveGroupMemberListByGroupId(groupId);
- Map allPositions = groupMessageReadRedisDAO.getReadMaxMessageIdMap(groupId);
+ Map allPositions = conversationReadService.getUserReadMessageIdMap(
+ ImConversationTypeEnum.GROUP.getType(), groupId);
for (ImGroupMessageDO message : pendingMessages) {
- // 2.1.1 统计可见成员中的已读人数
- Set visibleUserIds = getVisibleUserIds(message, activeMembers);
- visibleUserIds.remove(message.getSenderId()); // 发送者自己不算已读
- if (CollUtil.isEmpty(visibleUserIds)) {
+ // 2.1.1 统计可见成员中的已读人数（应读分母 = 快照 ∩ 当前有效成员 - sender）
+ Set
+ 
