From d91572b887cb9b2d0631fa192dac391f05540514 Mon Sep 17 00:00:00 2001 From: Xinwei Xiong <3293172751@qq.com> Date: Sun, 26 Nov 2023 20:36:06 +0800 Subject: [PATCH] Update as_read.go --- internal/rpc/msg/as_read.go | 87 ++++++++++++++++++++++++++++--------- 1 file changed, 66 insertions(+), 21 deletions(-) diff --git a/internal/rpc/msg/as_read.go b/internal/rpc/msg/as_read.go index 9096fe06e..fca182fe8 100644 --- a/internal/rpc/msg/as_read.go +++ b/internal/rpc/msg/as_read.go @@ -1,34 +1,25 @@ -// Copyright © 2023 OpenIM. All rights reserved. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - +// Package msg provides message handling functionalities for the messaging server. package msg import ( "context" - utils2 "github.com/OpenIMSDK/tools/utils" "github.com/redis/go-redis/v9" + // Importing necessary components from OpenIMSDK. "github.com/OpenIMSDK/protocol/constant" + "github.com/OpenIMSDK/protocol/conversation" "github.com/OpenIMSDK/protocol/msg" "github.com/OpenIMSDK/protocol/sdkws" "github.com/OpenIMSDK/tools/errs" "github.com/OpenIMSDK/tools/log" ) +// GetConversationsHasReadAndMaxSeq retrieves the read status and maximum sequence number of specified conversations. func (m *msgServer) GetConversationsHasReadAndMaxSeq(ctx context.Context, req *msg.GetConversationsHasReadAndMaxSeqReq) (resp *msg.GetConversationsHasReadAndMaxSeqResp, err error) { var conversationIDs []string + + // If no conversation IDs are provided, retrieve them from the local cache. if len(req.ConversationIDs) == 0 { conversationIDs, err = m.ConversationLocalCache.GetConversationIDs(ctx, req.UserID) if err != nil { @@ -37,51 +28,71 @@ func (m *msgServer) GetConversationsHasReadAndMaxSeq(ctx context.Context, req *m } else { conversationIDs = req.ConversationIDs } + + // Retrieve the read sequence numbers for the conversations. hasReadSeqs, err := m.MsgDatabase.GetHasReadSeqs(ctx, req.UserID, conversationIDs) if err != nil { return nil, err } + + // Retrieve conversation details. conversations, err := m.Conversation.GetConversations(ctx, req.UserID, conversationIDs) if err != nil { return nil, err } + + // Prepare a map to store the maximum sequence numbers. conversationMaxSeqMap := make(map[string]int64) for _, conversation := range conversations { if conversation.MaxSeq != 0 { conversationMaxSeqMap[conversation.ConversationID] = conversation.MaxSeq } } + + // Retrieve the maximum sequence numbers for the conversations. maxSeqs, err := m.MsgDatabase.GetMaxSeqs(ctx, conversationIDs) if err != nil { return nil, err } + + // Prepare the response with the sequence information. resp = &msg.GetConversationsHasReadAndMaxSeqResp{Seqs: make(map[string]*msg.Seqs)} - for conversarionID, maxSeq := range maxSeqs { - resp.Seqs[conversarionID] = &msg.Seqs{ - HasReadSeq: hasReadSeqs[conversarionID], + for conversationID, maxSeq := range maxSeqs { + resp.Seqs[conversationID] = &msg.Seqs{ + HasReadSeq: hasReadSeqs[conversationID], MaxSeq: maxSeq, } - if v, ok := conversationMaxSeqMap[conversarionID]; ok { - resp.Seqs[conversarionID].MaxSeq = v + + // Override the maximum sequence number if available in the map. + if v, ok := conversationMaxSeqMap[conversationID]; ok { + resp.Seqs[conversationID].MaxSeq = v } } return resp, nil } +// SetConversationHasReadSeq updates the read sequence number for a specific conversation. func (m *msgServer) SetConversationHasReadSeq( ctx context.Context, req *msg.SetConversationHasReadSeqReq, ) (resp *msg.SetConversationHasReadSeqResp, err error) { + // Retrieve the maximum sequence number for the conversation. maxSeq, err := m.MsgDatabase.GetMaxSeq(ctx, req.ConversationID) if err != nil { return } + + // Validate the provided read sequence number. if req.HasReadSeq > maxSeq { return nil, errs.ErrArgs.Wrap("hasReadSeq must not be bigger than maxSeq") } + + // Update the read sequence number in the database. if err := m.MsgDatabase.SetHasReadSeq(ctx, req.UserID, req.ConversationID, req.HasReadSeq); err != nil { return nil, err } + + // Send a notification for the read status update. if err = m.sendMarkAsReadNotification(ctx, req.ConversationID, constant.SingleChatType, req.UserID, req.UserID, nil, req.HasReadSeq); err != nil { return @@ -89,35 +100,52 @@ func (m *msgServer) SetConversationHasReadSeq( return &msg.SetConversationHasReadSeqResp{}, nil } +// MarkMsgsAsRead marks specific messages in a conversation as read. func (m *msgServer) MarkMsgsAsRead(ctx context.Context, req *msg.MarkMsgsAsReadReq) (resp *msg.MarkMsgsAsReadResp, err error) { + // Ensure that the sequence numbers are provided. + // Ensure that the sequence numbers are provided. if len(req.Seqs) < 1 { return nil, errs.ErrArgs.Wrap("seqs must not be empty") } + + // Retrieve the maximum sequence number for the conversation. maxSeq, err := m.MsgDatabase.GetMaxSeq(ctx, req.ConversationID) if err != nil { return } + + // Determine the highest sequence number from the request. hasReadSeq := req.Seqs[len(req.Seqs)-1] if hasReadSeq > maxSeq { return nil, errs.ErrArgs.Wrap("hasReadSeq must not be bigger than maxSeq") } + + // Retrieve conversation details. conversation, err := m.Conversation.GetConversation(ctx, req.UserID, req.ConversationID) if err != nil { return } + + // Mark the specified messages as read in the database. if err = m.MsgDatabase.MarkSingleChatMsgsAsRead(ctx, req.UserID, req.ConversationID, req.Seqs); err != nil { return } + + // Get the current read sequence number. currentHasReadSeq, err := m.MsgDatabase.GetHasReadSeq(ctx, req.UserID, req.ConversationID) if err != nil && errs.Unwrap(err) != redis.Nil { return } + + // Update the read sequence number if the new value is greater. if hasReadSeq > currentHasReadSeq { err = m.MsgDatabase.SetHasReadSeq(ctx, req.UserID, req.ConversationID, hasReadSeq) if err != nil { return } } + + // Send a notification to indicate that messages have been marked as read. if err = m.sendMarkAsReadNotification(ctx, req.ConversationID, conversation.ConversationType, req.UserID, m.conversationAndGetRecvID(conversation, req.UserID), req.Seqs, hasReadSeq); err != nil { return @@ -125,17 +153,24 @@ func (m *msgServer) MarkMsgsAsRead(ctx context.Context, req *msg.MarkMsgsAsReadR return &msg.MarkMsgsAsReadResp{}, nil } +// MarkConversationAsRead marks an entire conversation as read. func (m *msgServer) MarkConversationAsRead(ctx context.Context, req *msg.MarkConversationAsReadReq) (resp *msg.MarkConversationAsReadResp, err error) { + // Retrieve conversation details. conversation, err := m.Conversation.GetConversation(ctx, req.UserID, req.ConversationID) if err != nil { return nil, err } + + // Get the current read sequence number. hasReadSeq, err := m.MsgDatabase.GetHasReadSeq(ctx, req.UserID, req.ConversationID) if err != nil && errs.Unwrap(err) != redis.Nil { return nil, err } + + // Generate the sequence numbers to be marked as read. seqs := generateSeqs(hasReadSeq, req) + // Update the read status if there are new sequences to mark or if the hasReadSeq is greater. if len(seqs) > 0 || req.HasReadSeq > hasReadSeq { err = m.updateReadStatus(ctx, req, conversation, seqs, hasReadSeq) if err != nil { @@ -145,6 +180,7 @@ func (m *msgServer) MarkConversationAsRead(ctx context.Context, req *msg.MarkCon return &msg.MarkConversationAsReadResp{}, nil } +// generateSeqs creates a slice of sequence numbers that are greater than the provided hasReadSeq. func generateSeqs(hasReadSeq int64, req *msg.MarkConversationAsReadReq) []int64 { var seqs []int64 for _, val := range req.Seqs { @@ -155,7 +191,9 @@ func generateSeqs(hasReadSeq int64, req *msg.MarkConversationAsReadReq) []int64 return seqs } -func (m *msgServer) updateReadStatus(ctx context.Context, req *msg.MarkConversationAsReadReq, conversation *Conversation, seqs []int64, hasReadSeq int64) error { +// updateReadStatus updates the read status for messages in a conversation. +func (m *msgServer) updateReadStatus(ctx context.Context, req *msg.MarkConversationAsReadReq, conversation *conversation.Conversation, seqs []int64, hasReadSeq int64) error { + // Special handling for single chat type conversations. if conversation.ConversationType == constant.SingleChatType && len(seqs) > 0 { log.ZDebug(ctx, "MarkConversationAsRead", "seqs", seqs, "conversationID", req.ConversationID) if err := m.MsgDatabase.MarkSingleChatMsgsAsRead(ctx, req.UserID, req.ConversationID, seqs); err != nil { @@ -163,20 +201,25 @@ func (m *msgServer) updateReadStatus(ctx context.Context, req *msg.MarkConversat } } + // Update the hasReadSeq if the new value is greater. if req.HasReadSeq > hasReadSeq { if err := m.MsgDatabase.SetHasReadSeq(ctx, req.UserID, req.ConversationID, req.HasReadSeq); err != nil { return err } } + // Determine the receiver ID for the read receipt. recvID := m.conversationAndGetRecvID(conversation, req.UserID) + // Adjust the receiver ID for specific conversation types. if conversation.ConversationType == constant.SuperGroupChatType || conversation.ConversationType == constant.NotificationChatType { recvID = req.UserID } + // Send a notification to indicate the read status update. return m.sendMarkAsReadNotification(ctx, req.ConversationID, conversation.ConversationType, req.UserID, recvID, seqs, req.HasReadSeq) } +// sendMarkAsReadNotification sends a notification about the read status update. func (m *msgServer) sendMarkAsReadNotification( ctx context.Context, conversationID string, @@ -185,12 +228,14 @@ func (m *msgServer) sendMarkAsReadNotification( seqs []int64, hasReadSeq int64, ) error { + // Construct the read receipt notification. tips := &sdkws.MarkAsReadTips{ MarkAsReadUserID: sendID, ConversationID: conversationID, Seqs: seqs, HasReadSeq: hasReadSeq, } + // Send the notification with session type information. err := m.notificationSender.NotificationWithSesstionType(ctx, sendID, recvID, constant.HasReadReceipt, sessionType, tips) if err != nil { log.ZWarn(ctx, "send has read Receipt err", err)