blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2026-06-30 23:08:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

群阅后即焚

Browse Source
This commit is contained in:
hawklin2017 2026-05-20 17:54:49 +08:00
parent 35152a2832
commit 6f5575f583
4 changed files with 28 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
internal/rpc/msg/as_read.go
View File

@ -189,7 +189,7 @@ func (m *msgServer) MarkConversationAsRead(ctx context.Context, req *msg.MarkCon
for i := oldHasReadSeq + 1; i <= req.HasReadSeq; i++ {
groupSeqs = append(groupSeqs, i)
}
m.recordGroupBurnReadCount(ctx, conversation, groupSeqs)
m.recordGroupBurnReadCount(ctx, conversation, req.UserID, groupSeqs)
}
m.sendMarkAsReadNotification(ctx, req.ConversationID, constant.SingleChatType, req.UserID,
req.UserID, seqs, hasReadSeq)
@ -276,9 +276,9 @@ func (m *msgServer) recordBurnDeadlines(ctx context.Context, conv *conversation.
}
// recordGroupBurnReadCount 在群聊阅读时记录「阅后即焚」进度。
// 每次已读触发 $inc read_count；首次写入时记录 member_count 与 burn_end_time。
// 每次已读触发 $inc read_count；首次写入时记录 member_count、burn_end_time、send_id（发送者）。
// 仅在群的 MsgBurnDuration > 0 时生效；失败只记日志，不影响主流程。
func (m *msgServer) recordGroupBurnReadCount(ctx context.Context, conv *conversation.Conversation, seqs []int64) {
func (m *msgServer) recordGroupBurnReadCount(ctx context.Context, conv *conversation.Conversation, readerUserID string, seqs []int64) {
if len(seqs) == 0 || m.groupMsgBurnRecordDB == nil {
return
}
@ -290,10 +290,22 @@ func (m *msgServer) recordGroupBurnReadCount(ctx context.Context, conv *conversa
if groupInfo.MsgBurnDuration <= 0 {
return
}
seqSenderID := make(map[int64]string, len(seqs))
_, _, msgs, err := m.MsgDatabase.GetMsgBySeqs(ctx, readerUserID, conv.ConversationID, seqs)
if err != nil {
log.ZWarn(ctx, "recordGroupBurnReadCount GetMsgBySeqs failed", err,
"groupID", conv.GroupID, "conversationID", conv.ConversationID, "readerUserID", readerUserID, "seqs", seqs)
} else {
for _, md := range msgs {
if md != nil && md.Seq > 0 {
seqSenderID[md.Seq] = md.SendID
}
}
}
now := time.Now().UnixMilli()
burnEndTimeMs := now + int64(groupInfo.MsgBurnDuration)*1000
memberCount := int32(groupInfo.MemberCount)
if err := m.groupMsgBurnRecordDB.UpsertOnRead(ctx, conv.GroupID, seqs, memberCount, burnEndTimeMs); err != nil {
if err := m.groupMsgBurnRecordDB.UpsertOnRead(ctx, conv.GroupID, seqs, seqSenderID, memberCount, burnEndTimeMs); err != nil {
log.ZError(ctx, "recordGroupBurnReadCount UpsertOnRead failed", err,
"groupID", conv.GroupID, "seqs", seqs)
}

6
pkg/common/storage/database/group_msg_burn_record.go
View File

@ -34,9 +34,9 @@ type ExpiredGroupBurn struct {
// 消费：conversation 服务 ClearGroupBurnExpiredMsgs cron 入口。
type GroupMsgBurnRecord interface {
// UpsertOnRead 批量原子更新阅读记录：
// - 若 (group_id, seq) 不存在：插入 {member_count, burn_end_time, create_time, read_count=1}
// - 若已存在：仅对 read_count 执行 $inc，不覆盖首次写入的 burn_end_time
UpsertOnRead(ctx context.Context, groupID string, seqs []int64, memberCount int32, burnEndTimeMs int64) error
// - 若 (group_id, seq) 不存在：插入 {member_count, burn_end_time, create_time, send_id, read_count=1}；send_id 来自 seqSenderID[seq]，可为空。
// - 若已存在：仅对 read_count 执行 $inc，不覆盖首次写入的 burn_end_time、send_id
UpsertOnRead(ctx context.Context, groupID string, seqs []int64, seqSenderID map[int64]string, memberCount int32, burnEndTimeMs int64) error
// FindExpired 查询满足以下条件的记录并按 group_id 聚合：
// burn_end_time <= nowMs AND read_count >= member_count

9
pkg/common/storage/database/mgo/group_msg_burn_record.go
View File

@ -52,15 +52,19 @@ type groupMsgBurnRecordMgo struct {
}
// UpsertOnRead 对每条 seq 执行 upsert：
// - 首次插入（$setOnInsert）写入 member_count、burn_end_time、create_time，read_count 初始化为 1。
// - 首次插入（$setOnInsert）写入 member_count、burn_end_time、create_time、send_id，read_count 初始化为 1。
// - 已存在时仅对 read_count 执行 $inc 1。
func (m *groupMsgBurnRecordMgo) UpsertOnRead(ctx context.Context, groupID string, seqs []int64, memberCount int32, burnEndTimeMs int64) error {
func (m *groupMsgBurnRecordMgo) UpsertOnRead(ctx context.Context, groupID string, seqs []int64, seqSenderID map[int64]string, memberCount int32, burnEndTimeMs int64) error {
if len(seqs) == 0 {
return nil
}
now := time.Now().UnixMilli()
models := make([]mongo.WriteModel, 0, len(seqs))
for _, seq := range seqs {
senderID := ""
if seqSenderID != nil {
senderID = seqSenderID[seq]
}
filter := bson.M{
"group_id": groupID,
"seq": seq,
@ -70,6 +74,7 @@ func (m *groupMsgBurnRecordMgo) UpsertOnRead(ctx context.Context, groupID string
"$setOnInsert": bson.M{
"group_id": groupID,
"seq": seq,
"send_id": senderID,
"member_count": memberCount,
"burn_end_time": burnEndTimeMs,
"create_time": now,

2
pkg/common/storage/model/group_msg_burn_record.go
View File

@ -26,6 +26,8 @@ type GroupMsgBurnRecord struct {
GroupID string `bson:"group_id"`
// Seq 消息序列号
Seq int64 `bson:"seq"`
// SendID 发送该条群消息的用户 ID（首次有成员已读时写入，$setOnInsert）
SendID string `bson:"send_id"`
// ReadCount 已阅读该消息的成员数量（原子累加）
ReadCount int32 `bson:"read_count"`
// MemberCount 创建记录时的群成员总数；用于判断是否全员已读