fix: server can return isEnd to control fetch messages when sdk pull messages end normally. (#2949)

go.mod

@@ -15,7 +15,7 @@ require (
 	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.63
-	github.com/openimsdk/tools v0.0.50-alpha.50
+	github.com/openimsdk/tools v0.0.50-alpha.47
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0
 	github.com/stretchr/testify v1.9.0

go.sum

@@ -353,8 +353,8 @@ github.com/openimsdk/gomake v0.0.15-alpha.2 h1:5Q8yl8ezy2yx+q8/ucU/t4kJnDfCzNOrk
 github.com/openimsdk/gomake v0.0.15-alpha.2/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
 github.com/openimsdk/protocol v0.0.72-alpha.63 h1:IyPBibEvwBtTmD8DSrlqcekfEXe74k4+KeeHsgdhGh0=
 github.com/openimsdk/protocol v0.0.72-alpha.63/go.mod h1:Iet+piS/jaS+kWWyj6EEr36mk4ISzIRYjoMSVA4dq2M=
-github.com/openimsdk/tools v0.0.50-alpha.50 h1:+naDlvHcqJDj2NsCGnQd1LLQOET5IRPbrtmWbM/o7JQ=
+github.com/openimsdk/tools v0.0.50-alpha.47 h1:Cfe2va/g6WhLjOoQqZkjrdlEDq1dUsfcQsdUB5oADVA=
-github.com/openimsdk/tools v0.0.50-alpha.50/go.mod h1:muCtxguNJv8lFwLei27UASu2Nvg4ERSeN0R4K5tivk0=
+github.com/openimsdk/tools v0.0.50-alpha.47/go.mod h1:muCtxguNJv8lFwLei27UASu2Nvg4ERSeN0R4K5tivk0=
 github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=
 github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs=
 github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ=

internal/rpc/msg/sync_msg.go

@@ -92,11 +92,13 @@ func (m *msgServer) GetSeqMessage(ctx context.Context, req *msg.GetSeqMessageReq
 		NotificationMsgs: make(map[string]*sdkws.PullMsgs),
 	}
 	for _, conv := range req.Conversations {
-		_, _, msgs, err := m.MsgDatabase.GetMsgBySeqs(ctx, req.UserID, conv.ConversationID, conv.Seqs)
+		isEnd, endSeq, msgs, err := m.MsgDatabase.GetMessagesBySeqWithBounds(ctx, req.UserID, conv.ConversationID, conv.Seqs, req.GetOrder())
 		if err != nil {
 			return nil, err
 		}
 		var pullMsgs *sdkws.PullMsgs
+		pullMsgs.IsEnd = isEnd
+		pullMsgs.EndSeq = endSeq
 		if ok := false; conversationutil.IsNotificationConversationID(conv.ConversationID) {
 			pullMsgs, ok = resp.NotificationMsgs[conv.ConversationID]
 			if !ok {

pkg/common/storage/controller/msg.go

@@ -24,6 +24,9 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 
+	"github.com/redis/go-redis/v9"
+	"go.mongodb.org/mongo-driver/mongo"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/convert"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
@@ -35,8 +38,6 @@ import (
 	"github.com/openimsdk/tools/mq/kafka"
 	"github.com/openimsdk/tools/utils/datautil"
 	"github.com/openimsdk/tools/utils/timeutil"
-	"github.com/redis/go-redis/v9"
-	"go.mongodb.org/mongo-driver/mongo"
 )
 
 const (
@@ -56,6 +57,7 @@ type CommonMsgDatabase interface {
 	GetMsgBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) (minSeq int64, maxSeq int64, seqMsg []*sdkws.MsgData, err error)
 	// DeleteConversationMsgsAndSetMinSeq deletes conversation messages and resets the minimum sequence number. If `remainTime` is 0, all messages are deleted (this method does not delete Redis
 	// cache).
+	GetMessagesBySeqWithBounds(ctx context.Context, userID string, conversationID string, seqs []int64, pullOrder sdkws.PullOrder) (bool, int64, []*sdkws.MsgData, error)
 	DeleteConversationMsgsAndSetMinSeq(ctx context.Context, conversationID string, remainTime int64) error
 	// ClearUserMsgs marks messages for deletion based on clear time and returns a list of sequence numbers for marked messages.
 	ClearUserMsgs(ctx context.Context, userID string, conversationID string, clearTime int64, lastMsgClearTime time.Time) (seqs []int64, err error)
@@ -517,6 +519,81 @@ func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, co
 	return minSeq, maxSeq, successMsgs, nil
 }
 
+func (db *commonMsgDatabase) GetMessagesBySeqWithBounds(ctx context.Context, userID string, conversationID string, seqs []int64, pullOrder sdkws.PullOrder) (bool, int64, []*sdkws.MsgData, error) {
+	var endSeq int64
+	var isEnd bool
+	userMinSeq, err := db.seqUser.GetUserMinSeq(ctx, conversationID, userID)
+	if err != nil {
+		return false, 0, nil, err
+	}
+	minSeq, err := db.seqConversation.GetMinSeq(ctx, conversationID)
+	if err != nil {
+		return false, 0, nil, err
+	}
+	maxSeq, err := db.seqConversation.GetMaxSeq(ctx, conversationID)
+	if err != nil {
+		return false, 0, nil, err
+	}
+	userMaxSeq, err := db.seqUser.GetUserMaxSeq(ctx, conversationID, userID)
+	if err != nil {
+		return false, 0, nil, err
+	}
+	if userMinSeq > minSeq {
+		minSeq = userMinSeq
+	}
+	if userMaxSeq > 0 && userMaxSeq < maxSeq {
+		maxSeq = userMaxSeq
+	}
+	newSeqs := make([]int64, 0, len(seqs))
+	for _, seq := range seqs {
+		if seq <= 0 {
+			continue
+		}
+		// The normal range and can fetch messages
+		if seq >= minSeq && seq <= maxSeq {
+			newSeqs = append(newSeqs, seq)
+			continue
+		}
+		// If the requested seq is smaller than the minimum seq and the pull order is descending (pulling older messages)
+		if seq < minSeq && pullOrder == sdkws.PullOrder_PullOrderDesc {
+			isEnd = true
+			endSeq = minSeq
+		}
+		// If the requested seq is larger than the maximum seq and the pull order is ascending (pulling newer messages)
+		if seq > maxSeq && pullOrder == sdkws.PullOrder_PullOrderAsc {
+			isEnd = true
+			endSeq = maxSeq
+		}
+	}
+	if len(newSeqs) == 0 {
+		return isEnd, endSeq, nil, nil
+	}
+	successMsgs, failedSeqs, err := db.msg.GetMessagesBySeq(ctx, conversationID, newSeqs)
+	if err != nil {
+		if !errors.Is(err, redis.Nil) {
+			log.ZWarn(ctx, "get message from redis exception", err, "failedSeqs", failedSeqs, "conversationID", conversationID)
+		}
+	}
+	log.ZDebug(ctx, "db.seq.GetMessagesBySeq", "userID", userID, "conversationID", conversationID, "seqs",
+		seqs, "len(successMsgs)", len(successMsgs), "failedSeqs", failedSeqs)
+
+	if len(failedSeqs) > 0 {
+		mongoMsgs, err := db.getMsgBySeqs(ctx, userID, conversationID, failedSeqs)
+		if err != nil {
+
+			return false, 0, nil, err
+		}
+
+		successMsgs = append(successMsgs, mongoMsgs...)
+
+		//_, err = db.msg.SetMessagesToCache(ctx, conversationID, mongoMsgs)
+		//if err != nil {
+		//	return 0, 0, nil, err
+		//}
+	}
+	return isEnd, endSeq, successMsgs, nil
+}
+
 func (db *commonMsgDatabase) DeleteConversationMsgsAndSetMinSeq(ctx context.Context, conversationID string, remainTime int64) error {
 	var delStruct delMsgRecursionStruct
 	var skip int64