Merge: main

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/gorilla/websocket v1.5.1
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
 	github.com/mitchellh/mapstructure v1.5.0
-	github.com/openimsdk/protocol v0.0.72-alpha.60
+	github.com/openimsdk/protocol v0.0.72-alpha.61
 	github.com/openimsdk/tools v0.0.50-alpha.38
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.18.0

go.sum

@@ -319,8 +319,8 @@ github.com/onsi/gomega v1.25.0 h1:Vw7br2PCDYijJHSfBOWhov+8cAnUf8MfMaIOV323l6Y=
 github.com/onsi/gomega v1.25.0/go.mod h1:r+zV744Re+DiYCIPRlYOTxn0YkOLcAnW8k1xXdMPGhM=
 github.com/openimsdk/gomake v0.0.14-alpha.5 h1:VY9c5x515lTfmdhhPjMvR3BBRrRquAUCFsz7t7vbv7Y=
 github.com/openimsdk/gomake v0.0.14-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
-github.com/openimsdk/protocol v0.0.72-alpha.60 h1:4ghyVsMqvmnaSHUkg1SiHIor2aa8Yjp8wHxSRkf4zBg=
-github.com/openimsdk/protocol v0.0.72-alpha.60/go.mod h1:Iet+piS/jaS+kWWyj6EEr36mk4ISzIRYjoMSVA4dq2M=
+github.com/openimsdk/protocol v0.0.72-alpha.61 h1:RuZR9/Sg3p6Bpb2CKPjPoA2AUmTvHITmhZ3PT/RbWMs=
+github.com/openimsdk/protocol v0.0.72-alpha.61/go.mod h1:Iet+piS/jaS+kWWyj6EEr36mk4ISzIRYjoMSVA4dq2M=
 github.com/openimsdk/tools v0.0.50-alpha.38 h1:AU6/cvDfN4ciIOwAj8IWEwze3DeEp2cHYPgW3y0OlbU=
 github.com/openimsdk/tools v0.0.50-alpha.38/go.mod h1:/Em/fQH46CuWf60+hcmvZyboGCQpSDEb2MdQ4nmQRAk=
 github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=

internal/rpc/conversation/conversation.go

@@ -16,6 +16,8 @@ package conversation
 
 import (
 	"context"
+	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
+	pbmsg "github.com/openimsdk/protocol/msg"
 	"sort"
 	"time"
 
@@ -433,23 +435,38 @@ func (c *conversationServer) CreateGroupChatConversations(ctx context.Context, r
 	if err != nil {
 		return nil, err
 	}
+	conversationID := msgprocessor.GetConversationIDBySessionType(constant.ReadGroupChatType, req.GroupID)
+	if _, err := c.msgRpcClient.Client.SetUserConversationMaxSeq(ctx, &pbmsg.SetUserConversationMaxSeqReq{ConversationID: conversationID, OwnerUserID: req.UserIDs, MaxSeq: 0}); err != nil {
+		return nil, err
+	}
 	return &pbconversation.CreateGroupChatConversationsResp{}, nil
 }
 
 func (c *conversationServer) SetConversationMaxSeq(ctx context.Context, req *pbconversation.SetConversationMaxSeqReq) (*pbconversation.SetConversationMaxSeqResp, error) {
+	if _, err := c.msgRpcClient.Client.SetUserConversationMaxSeq(ctx, &pbmsg.SetUserConversationMaxSeqReq{ConversationID: req.ConversationID, OwnerUserID: req.OwnerUserID, MaxSeq: req.MaxSeq}); err != nil {
+		return nil, err
+	}
 	if err := c.conversationDatabase.UpdateUsersConversationField(ctx, req.OwnerUserID, req.ConversationID,
 		map[string]any{"max_seq": req.MaxSeq}); err != nil {
 		return nil, err
 	}
-
+	for _, userID := range req.OwnerUserID {
+		c.conversationNotificationSender.ConversationChangeNotification(ctx, userID, []string{req.ConversationID})
+	}
 	return &pbconversation.SetConversationMaxSeqResp{}, nil
 }
 
 func (c *conversationServer) SetConversationMinSeq(ctx context.Context, req *pbconversation.SetConversationMinSeqReq) (*pbconversation.SetConversationMinSeqResp, error) {
+	if _, err := c.msgRpcClient.Client.SetUserConversationMinSeq(ctx, &pbmsg.SetUserConversationMinSeqReq{ConversationID: req.ConversationID, OwnerUserID: req.OwnerUserID, MinSeq: req.MinSeq}); err != nil {
+		return nil, err
+	}
 	if err := c.conversationDatabase.UpdateUsersConversationField(ctx, req.OwnerUserID, req.ConversationID,
 		map[string]any{"min_seq": req.MinSeq}); err != nil {
 		return nil, err
 	}
+	for _, userID := range req.OwnerUserID {
+		c.conversationNotificationSender.ConversationChangeNotification(ctx, userID, []string{req.ConversationID})
+	}
 	return &pbconversation.SetConversationMinSeqResp{}, nil
 }
 

internal/rpc/group/group.go

@@ -964,7 +964,6 @@ func (g *groupServer) deleteMemberAndSetConversationSeq(ctx context.Context, gro
 	if err != nil {
 		return err
 	}
-
 	return g.conversationRpcClient.SetConversationMaxSeq(ctx, userIDs, conevrsationID, maxSeq)
 }
 

internal/rpc/msg/seq.go

@@ -84,3 +84,21 @@ func (m *msgServer) GetActiveConversation(ctx context.Context, req *pbmsg.GetAct
 	}
 	return &pbmsg.GetActiveConversationResp{Conversations: conversations}, nil
 }
+
+func (m *msgServer) SetUserConversationMaxSeq(ctx context.Context, req *pbmsg.SetUserConversationMaxSeqReq) (*pbmsg.SetUserConversationMaxSeqResp, error) {
+	for _, userID := range req.OwnerUserID {
+		if err := m.MsgDatabase.SetUserConversationsMaxSeq(ctx, req.ConversationID, userID, req.MaxSeq); err != nil {
+			return nil, err
+		}
+	}
+	return &pbmsg.SetUserConversationMaxSeqResp{}, nil
+}
+
+func (m *msgServer) SetUserConversationMinSeq(ctx context.Context, req *pbmsg.SetUserConversationMinSeqReq) (*pbmsg.SetUserConversationMinSeqResp, error) {
+	for _, userID := range req.OwnerUserID {
+		if err := m.MsgDatabase.SetUserConversationsMinSeq(ctx, req.ConversationID, userID, req.MinSeq); err != nil {
+			return nil, err
+		}
+	}
+	return &pbmsg.SetUserConversationMinSeqResp{}, nil
+}

internal/rpc/msg/server.go

@@ -16,7 +16,6 @@ package msg
 
 import (
 	"context"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo"

pkg/common/storage/controller/msg.go

@@ -97,6 +97,9 @@ type CommonMsgDatabase interface {
 	DeleteDocMsgBefore(ctx context.Context, ts int64, doc *model.MsgDocModel) ([]int, error)
 
 	GetDocIDs(ctx context.Context) ([]string, error)
+
+	SetUserConversationsMaxSeq(ctx context.Context, conversationID string, userID string, seq int64) error
+	SetUserConversationsMinSeq(ctx context.Context, conversationID string, userID string, seq int64) error
 }
 
 func NewCommonMsgDatabase(msgDocModel database.Msg, msg cache.MsgCache, seqUser cache.SeqUser, seqConversation cache.SeqConversationCache, kafkaConf *config.Kafka) (CommonMsgDatabase, error) {
@@ -702,6 +705,14 @@ func (db *commonMsgDatabase) SetUserConversationsMinSeqs(ctx context.Context, us
 	return db.seqUser.SetUserMinSeqs(ctx, userID, seqs)
 }
 
+func (db *commonMsgDatabase) SetUserConversationsMaxSeq(ctx context.Context, conversationID string, userID string, seq int64) error {
+	return db.seqUser.SetUserMaxSeq(ctx, conversationID, userID, seq)
+}
+
+func (db *commonMsgDatabase) SetUserConversationsMinSeq(ctx context.Context, conversationID string, userID string, seq int64) error {
+	return db.seqUser.SetUserMinSeq(ctx, conversationID, userID, seq)
+}
+
 func (db *commonMsgDatabase) UserSetHasReadSeqs(ctx context.Context, userID string, hasReadSeqs map[string]int64) error {
 	return db.seqUser.SetUserReadSeqs(ctx, userID, hasReadSeqs)
 }