msg

pkg/common/db/cache/msg.go

@@ -22,48 +22,43 @@ import (
 )
 
 const (
-	userIncrSeq             = "REDIS_USER_INCR_SEQ:" // user incr seq
-	appleDeviceToken        = "DEVICE_TOKEN"
-	userMinSeq              = "REDIS_USER_MIN_SEQ:"
-	getuiToken              = "GETUI_TOKEN"
-	getuiTaskID             = "GETUI_TASK_ID"
-	messageCache            = "MESSAGE_CACHE:"
-	signalCache             = "SIGNAL_CACHE:"
-	signalListCache         = "SIGNAL_LIST_CACHE:"
-	fcmToken                = "FCM_TOKEN:"
-	groupUserMinSeq         = "GROUP_USER_MIN_SEQ:"
-	groupMaxSeq             = "GROUP_MAX_SEQ:"
-	groupMinSeq             = "GROUP_MIN_SEQ:"
+	maxSeq                 = "MAX_SEQ:"
+	minSeq                 = "MIN_SEQ:"
+	conversationUserMinSeq = "CON_USER_MIN_SEQ:"
+
+	appleDeviceToken = "DEVICE_TOKEN"
+	getuiToken       = "GETUI_TOKEN"
+	getuiTaskID      = "GETUI_TASK_ID"
+	messageCache     = "MESSAGE_CACHE:"
+	signalCache      = "SIGNAL_CACHE:"
+	signalListCache  = "SIGNAL_LIST_CACHE:"
+	fcmToken         = "FCM_TOKEN:"
+
 	sendMsgFailedFlag       = "SEND_MSG_FAILED_FLAG:"
 	userBadgeUnreadCountSum = "USER_BADGE_UNREAD_COUNT_SUM:"
 	exTypeKeyLocker         = "EX_LOCK:"
 	uidPidToken             = "UID_PID_TOKEN_STATUS:"
-	userNotificationSeq     = "USER_NOTIFICATION_SEQ:"
-	userMinNotificationSeq  = "USER_MIN_NOTIFICATION_SEQ:"
-	groupNotificationSeq    = "GROUP_NOTIFICATION_SEQ:"
-	groupMinNotificationSeq = "GROUP_MIN_NOTIFICATION_SEQ:"
 )
 
 type MsgModel interface {
-	IncrUserSeq(ctx context.Context, userID string) (int64, error)
-	GetUserMaxSeq(ctx context.Context, userID string) (int64, error)
-	SetUserMaxSeq(ctx context.Context, userID string, maxSeq int64) error
-	SetUserMinSeq(ctx context.Context, userID string, minSeq int64) (err error)
-	GetUserMinSeq(ctx context.Context, userID string) (int64, error)
-	SetGroupUserMinSeq(ctx context.Context, groupID, userID string, minSeq int64) (err error)
-	GetGroupUserMinSeq(ctx context.Context, groupID, userID string) (int64, error)
-	GetGroupMaxSeq(ctx context.Context, groupID string) (int64, error)
-	GetGroupMinSeq(ctx context.Context, groupID string) (int64, error)
-	IncrGroupMaxSeq(ctx context.Context, groupID string) (int64, error)
-	SetGroupMaxSeq(ctx context.Context, groupID string, maxSeq int64) error
-	SetGroupMinSeq(ctx context.Context, groupID string, minSeq int64) error
+	SetMaxSeq(ctx context.Context, conversationID string, maxSeq int64) error
+	GetMaxSeqs(ctx context.Context, conversationIDs []string) (map[string]int64, error)
+	GetMaxSeq(ctx context.Context, conversationID string) (int64, error)
+	SetMinSeq(ctx context.Context, conversationID string, minSeq int64) error
+	GetMinSeqs(ctx context.Context, conversationIDs []string) (map[string]int64, error)
+	GetMinSeq(ctx context.Context, conversationID string) (int64, error)
+	GetConversationUserMinSeq(ctx context.Context, conversationID string, userID string) (int64, error)
+	GetConversationUserMinSeqs(ctx context.Context, conversationID string, userIDs []string) (map[string]int64, error)
+	SetConversationUserMinSeq(ctx context.Context, conversationID string, userID string, minSeq int64) error
+	SetConversationUserMinSeqs(ctx context.Context, conversationID string, seqs map[string]int64) (err error)
+
 	AddTokenFlag(ctx context.Context, userID string, platformID int, token string, flag int) error
 	GetTokensWithoutError(ctx context.Context, userID, platformID string) (map[string]int, error)
 	SetTokenMapByUidPid(ctx context.Context, userID string, platform string, m map[string]int) error
 	DeleteTokenByUidPid(ctx context.Context, userID string, platform string, fields []string) error
-	GetMessagesBySeq(ctx context.Context, userID string, seqList []int64) (seqMsg []*sdkws.MsgData, failedSeqList []int64, err error)
-	SetMessageToCache(ctx context.Context, userID string, msgList []*sdkws.MsgData) (int, error)
-	DeleteMessageFromCache(ctx context.Context, userID string, msgList []*sdkws.MsgData) error
+	GetMessagesBySeq(ctx context.Context, conversationID string, seqList []int64) (seqMsg []*sdkws.MsgData, failedSeqList []int64, err error)
+	SetMessageToCache(ctx context.Context, conversationID string, msgList []*sdkws.MsgData) (int, error)
+	DeleteMessageFromCache(ctx context.Context, conversationID string, msgList []*sdkws.MsgData) error
 	CleanUpOneUserAllMsg(ctx context.Context, userID string) error
 	HandleSignalInvite(ctx context.Context, msg *sdkws.MsgData, pushToUserID string) (isSend bool, err error)
 	GetSignalInvitationInfoByClientMsgID(ctx context.Context, clientMsgID string) (invitationInfo *sdkws.SignalInviteReq, err error)
@@ -90,20 +85,6 @@ type MsgModel interface {
 	SetMessageTypeKeyValue(ctx context.Context, clientMsgID string, sessionType int32, typeKey, value string) error
 	LockMessageTypeKey(ctx context.Context, clientMsgID string, TypeKey string) error
 	UnLockMessageTypeKey(ctx context.Context, clientMsgID string, TypeKey string) error
-	// notificatio
-
-	// 	IncrUserNotificationSeq(ctx context.Context, userID string) (int64, error)
-	// 	GetUserNotificationMaxSeq(ctx context.Context, userID string) (int64, error)
-	// 	SetUserNotificationMaxSeq(ctx context.Context, userID string, maxSeq int64) error
-	// 	SetUserNotificationMinSeq(ctx context.Context, userID string, minSeq int64) (err error)
-	// 	GetUserNotificationMinSeq(ctx context.Context, userID string) (int64, error)
-	// 	SetGroupNotificationUserMinSeq(ctx context.Context, groupID, userID string, minSeq int64) (err error)
-	// 	GetGroupNotificationUserMinSeq(ctx context.Context, groupID, userID string) (int64, error)
-	// 	GetGroupNotificationMaxSeq(ctx context.Context, groupID string) (int64, error)
-	// 	GetGroupNotificationMinSeq(ctx context.Context, groupID string) (int64, error)
-	// 	IncrGroupNotificationMaxSeq(ctx context.Context, groupID string) (int64, error)
-	// 	SetGroupNotificationMaxSeq(ctx context.Context, groupID string, maxSeq int64) error
-	// 	SetGroupNotificationMinSeq(ctx context.Context, groupID string, minSeq int64) error
 }
 
 func NewMsgCacheModel(client redis.UniversalClient) MsgModel {
@@ -146,58 +127,91 @@ func (c *msgCache) DelKeys() {
 	}
 }
 
-func (c *msgCache) IncrUserSeq(ctx context.Context, userID string) (int64, error) {
-	return utils.Wrap2(c.rdb.Get(ctx, userIncrSeq+userID).Int64())
+func (c *msgCache) getMaxSeqKey(conversationID string) string {
+	return maxSeq + conversationID
 }
 
-func (c *msgCache) GetUserMaxSeq(ctx context.Context, userID string) (int64, error) {
-	return utils.Wrap2(c.rdb.Get(ctx, userIncrSeq+userID).Int64())
+func (c *msgCache) getMinSeqKey(conversationID string) string {
+	return minSeq + conversationID
 }
 
-func (c *msgCache) SetUserMaxSeq(ctx context.Context, userID string, maxSeq int64) error {
-	return errs.Wrap(c.rdb.Set(ctx, userIncrSeq+userID, maxSeq, 0).Err())
+func (c *msgCache) setSeq(ctx context.Context, conversationID string, seq int64, getkey func(conversationID string) string) error {
+	return utils.Wrap1(c.rdb.Set(ctx, getkey(conversationID), seq, 0).Err())
 }
 
-func (c *msgCache) SetUserMinSeq(ctx context.Context, userID string, minSeq int64) (err error) {
-	return errs.Wrap(c.rdb.Set(ctx, userMinSeq+userID, minSeq, 0).Err())
+func (c *msgCache) getSeq(ctx context.Context, conversationID string, getkey func(conversationID string) string) (int64, error) {
+	return utils.Wrap2(c.rdb.Get(ctx, getkey(conversationID)).Int64())
 }
 
-func (c *msgCache) GetUserMinSeq(ctx context.Context, userID string) (int64, error) {
-	return utils.Wrap2(c.rdb.Get(ctx, userMinSeq+userID).Int64())
+func (c *msgCache) getSeqs(ctx context.Context, items []string, getkey func(s string) string) (m map[string]int64, err error) {
+	pipe := c.rdb.Pipeline()
+	for _, v := range items {
+		if err := pipe.Get(ctx, getkey(v)).Err(); err != nil && err != redis.Nil {
+			return nil, err
+		}
+	}
+	result, err := pipe.Exec(ctx)
+	if err != nil {
+		return nil, errs.Wrap(err)
+	}
+	m = make(map[string]int64, len(items))
+	for i, v := range result {
+		if v.Err() != nil && err != redis.Nil {
+			return nil, errs.Wrap(v.Err())
+		}
+		m[items[i]] = utils.StringToInt64(v.String())
+	}
+	return m, nil
 }
 
-func (c *msgCache) SetGroupUserMinSeq(ctx context.Context, groupID, userID string, minSeq int64) (err error) {
-	key := groupUserMinSeq + "g:" + groupID + "u:" + userID
-	return errs.Wrap(c.rdb.Set(ctx, key, minSeq, 0).Err())
+func (c *msgCache) SetMaxSeq(ctx context.Context, conversationID string, maxSeq int64) error {
+	return c.setSeq(ctx, conversationID, maxSeq, c.getMaxSeqKey)
 }
 
-func (c *msgCache) GetGroupUserMinSeq(ctx context.Context, groupID, userID string) (int64, error) {
-	key := groupUserMinSeq + "g:" + groupID + "u:" + userID
-	return utils.Wrap2(c.rdb.Get(ctx, key).Int64())
+func (c *msgCache) GetMaxSeqs(ctx context.Context, conversationIDs []string) (m map[string]int64, err error) {
+	return c.getSeqs(ctx, conversationIDs, c.getMaxSeqKey)
 }
 
-func (c *msgCache) GetGroupMaxSeq(ctx context.Context, groupID string) (int64, error) {
-	return utils.Wrap2(c.rdb.Get(ctx, groupMaxSeq+groupID).Int64())
+func (c *msgCache) GetMaxSeq(ctx context.Context, conversationID string) (int64, error) {
+	return c.getSeq(ctx, conversationID, c.getMaxSeqKey)
+}
+func (c *msgCache) SetMinSeq(ctx context.Context, conversationID string, minSeq int64) error {
+	return c.setSeq(ctx, conversationID, minSeq, c.getMinSeqKey)
+}
+func (c *msgCache) GetMinSeqs(ctx context.Context, conversationIDs []string) (map[string]int64, error) {
+	return c.getSeqs(ctx, conversationIDs, c.getMinSeqKey)
+}
+func (c *msgCache) GetMinSeq(ctx context.Context, conversationID string) (int64, error) {
+	return c.getSeq(ctx, conversationID, c.getMinSeqKey)
 }
 
-func (c *msgCache) GetGroupMinSeq(ctx context.Context, groupID string) (int64, error) {
-	return utils.Wrap2(c.rdb.Get(ctx, groupMinSeq+groupID).Int64())
+func (c *msgCache) getConversationUserMinSeqKey(conversationID, userID string) string {
+	return conversationUserMinSeq + "g:" + conversationID + "u:" + userID
 }
 
-func (c *msgCache) IncrGroupMaxSeq(ctx context.Context, groupID string) (int64, error) {
-	key := groupMaxSeq + groupID
-	seq, err := c.rdb.Incr(ctx, key).Uint64()
-	return int64(seq), errs.Wrap(err)
+func (c *msgCache) GetConversationUserMinSeq(ctx context.Context, conversationID string, userID string) (int64, error) {
+	return utils.Wrap2(c.rdb.Get(ctx, c.getConversationUserMinSeqKey(conversationID, userID)).Int64())
 }
 
-func (c *msgCache) SetGroupMaxSeq(ctx context.Context, groupID string, maxSeq int64) error {
-	key := groupMaxSeq + groupID
-	return errs.Wrap(c.rdb.Set(ctx, key, maxSeq, 0).Err())
+func (c *msgCache) GetConversationUserMinSeqs(ctx context.Context, conversationID string, userIDs []string) (m map[string]int64, err error) {
+	return c.getSeqs(ctx, userIDs, func(userID string) string {
+		return c.getConversationUserMinSeqKey(conversationID, userID)
+	})
+}
+func (c *msgCache) SetConversationUserMinSeq(ctx context.Context, conversationID string, userID string, minSeq int64) error {
+	return utils.Wrap1(c.rdb.Set(ctx, c.getConversationUserMinSeqKey(conversationID, userID), minSeq, 0).Err())
 }
 
-func (c *msgCache) SetGroupMinSeq(ctx context.Context, groupID string, minSeq int64) error {
-	key := groupMinSeq + groupID
-	return errs.Wrap(c.rdb.Set(ctx, key, minSeq, 0).Err())
+func (c *msgCache) SetConversationUserMinSeqs(ctx context.Context, conversationID string, seqs map[string]int64) (err error) {
+	pipe := c.rdb.Pipeline()
+	for userID, minSeq := range seqs {
+		err = pipe.Set(ctx, c.getConversationUserMinSeqKey(conversationID, userID), minSeq, 0).Err()
+		if err != nil {
+			return errs.Wrap(err)
+		}
+	}
+	_, err = pipe.Exec(ctx)
+	return err
 }
 
 func (c *msgCache) AddTokenFlag(ctx context.Context, userID string, platformID int, token string, flag int) error {
@@ -266,11 +280,11 @@ func (c *msgCache) GetMessagesBySeq(ctx context.Context, userID string, seqs []i
 	return seqMsgs, failedSeqs, err
 }
 
-func (c *msgCache) SetMessageToCache(ctx context.Context, userID string, msgList []*sdkws.MsgData) (int, error) {
+func (c *msgCache) SetMessageToCache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (int, error) {
 	pipe := c.rdb.Pipeline()
 	var failedMsgs []sdkws.MsgData
-	for _, msg := range msgList {
-		key := c.getMessageCacheKey(userID, msg.Seq)
+	for _, msg := range msgs {
+		key := c.getMessageCacheKey(conversationID, msg.Seq)
 		s, err := utils.Pb2String(msg)
 		if err != nil {
 			return 0, errs.Wrap(err)
@@ -281,7 +295,7 @@ func (c *msgCache) SetMessageToCache(ctx context.Context, userID string, msgList
 		}
 	}
 	if len(failedMsgs) != 0 {
-		return len(failedMsgs), fmt.Errorf("set msg to msgCache failed, failed lists: %v, %s", failedMsgs, userID)
+		return len(failedMsgs), fmt.Errorf("set msg to msgCache failed, failed lists: %v, %s", failedMsgs, conversationID)
 	}
 	_, err := pipe.Exec(ctx)
 	return 0, err

pkg/common/db/controller/common_msg.go

@@ -6,26 +6,28 @@ import (
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/sdkws"
 )
 
-// sourceID 可以是通知也可以是conversation
+// conversationID 可以是通知也可以是conversation
 type commonMsgDatabase interface {
-	BatchInsertChat2DB(ctx context.Context, sourceID string, msgs []*sdkws.MsgData, currentMaxSeq int64) error
-	DeleteMessageFromCache(ctx context.Context, sourceID string, msgs []*sdkws.MsgData) error
+	BatchInsertChat2DB(ctx context.Context, conversationID string, msgs []*sdkws.MsgData, currentMaxSeq int64) error
+	DeleteMessageFromCache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) error
 
-	GetMsgBySeqs(ctx context.Context, sourceID string, seqs []int64) (seqMsg []*sdkws.MsgData, err error)
-	GetMsgBySeqsRange(ctx context.Context, sourceID string, begin, end, num int64) (seqMsg []*sdkws.MsgData, err error)
-	CleanUpUserMsg(ctx context.Context, sourceID string) error
-	DelMsgsBySeqs(ctx context.Context, sourceID string, seqs []int64) (totalUnExistSeqs []int64, err error)
-	DelMsgsAndResetMinSeq(ctx context.Context, sourceID string, userIDs []string, remainTime int64) error
+	GetMsgBySeqs(ctx context.Context, conversationID string, seqs []int64) (seqMsg []*sdkws.MsgData, err error)
+	GetMsgBySeqsRange(ctx context.Context, conversationID string, begin, end, num int64) (seqMsg []*sdkws.MsgData, err error)
+	CleanUpUserMsg(ctx context.Context, conversationID string) error
+	DelMsgsBySeqs(ctx context.Context, conversationID string, seqs []int64) (totalUnExistSeqs []int64, err error)
+	DelMsgsAndResetMinSeq(ctx context.Context, conversationID string, userIDs []string, remainTime int64) error
 
-	GetMinMaxSeqInMongoAndCache(ctx context.Context, sourceID string) (minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache int64, err error)
+	GetMinMaxSeqInMongoAndCache(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache int64, err error)
 
-	GetMaxSeq(ctx context.Context, sourceID string) (int64, error)
-	GetMinSeq(ctx context.Context, sourceID string) (int64, error)
-	SetMaxSeq(ctx context.Context, sourceID string, seq int64) error
-	SetMinSeq(ctx context.Context, sourceID string, seq int64) error
+	GetMaxSeq(ctx context.Context, conversationID string) (int64, error)
+	GetMinSeq(ctx context.Context, conversationID string) (int64, error)
+	SetMaxSeq(ctx context.Context, conversationID string, seq int64) error
+	SetMinSeq(ctx context.Context, conversationID string, seq int64) error
+	GetMaxSeqs(ctx context.Context, conversationIDs []string) (map[string]int64, error)
+	GetMinSeqs(ctx context.Context, conversationIDs []string) (map[string]int64, error)
 
-	MsgToMQ(ctx context.Context, sourceID string, msg2mq *sdkws.MsgData) error
-	MsgToModifyMQ(ctx context.Context, sourceID string, messages []*sdkws.MsgData) error
-	MsgToPushMQ(ctx context.Context, sourceID string, msg2mq *sdkws.MsgData) (int32, int64, error)
-	MsgToMongoMQ(ctx context.Context, sourceID string, messages []*sdkws.MsgData, lastSeq int64) error
+	MsgToMQ(ctx context.Context, conversationID string, msg2mq *sdkws.MsgData) error
+	MsgToModifyMQ(ctx context.Context, conversationID string, messages []*sdkws.MsgData) error
+	MsgToPushMQ(ctx context.Context, conversationID string, msg2mq *sdkws.MsgData) (int32, int64, error)
+	MsgToMongoMQ(ctx context.Context, conversationID string, messages []*sdkws.MsgData, lastSeq int64) error
 }

pkg/common/db/controller/conversation.go

@@ -29,6 +29,7 @@ type ConversationDatabase interface {
 	//SetUsersConversationFiledTx 设置多个用户会话关于某个字段的更新操作，如果会话不存在则创建，否则更新，内部保证事务操作
 	SetUsersConversationFiledTx(ctx context.Context, userIDs []string, conversation *relationTb.ConversationModel, filedMap map[string]interface{}) error
 	CreateGroupChatConversation(ctx context.Context, groupID string, userIDs []string) error
+	GetConversationIDs(ctx context.Context, userID string) ([]string, error)
 }
 
 func NewConversationDatabase(conversation relationTb.ConversationModelInterface, cache cache.ConversationCache, tx tx.Tx) ConversationDatabase {
@@ -216,3 +217,7 @@ func (c *ConversationDataBase) CreateGroupChatConversation(ctx context.Context,
 	})
 
 }
+
+func (c *ConversationDataBase) GetConversationIDs(ctx context.Context, userID string) ([]string, error) {
+	return c.cache.GetUserConversationIDs(ctx, userID)
+}

pkg/common/db/controller/msg.go

@@ -38,33 +38,26 @@ type MsgDatabase interface {
 	// incrSeq通知seq然后批量插入缓存
 	NotificationBatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (int64, error)
 	// 删除消息 返回不存在的seqList
-	DelMsgBySeqs(ctx context.Context, userID string, seqs []int64) (totalUnExistSeqs []int64, err error)
-	// 获取群ID或者UserID最新一条在mongo里面的消息
+	DelMsgBySeqs(ctx context.Context, conversationID string, seqs []int64) (totalUnExistSeqs []int64, err error)
 	//  通过seqList获取mongo中写扩散消息
 	GetMsgBySeqsRange(ctx context.Context, conversationID string, begin, end, num int64) (seqMsg []*sdkws.MsgData, err error)
 	// 通过seqList获取大群在 mongo里面的消息
-	GetMsgBySeqs(ctx context.Context, groupID string, seqs []int64) (seqMsg []*sdkws.MsgData, err error)
+	GetMsgBySeqs(ctx context.Context, conversationID string, seqs []int64) (seqMsg []*sdkws.MsgData, err error)
 	// 删除用户所有消息/redis/mongo然后重置seq
-	CleanUpUserMsg(ctx context.Context, userID string) error
+	CleanUpConversationMsgs(ctx context.Context, conversationID string) error
 	// 删除大群消息重置群成员最小群seq, remainTime为消息保留的时间单位秒,超时消息删除， 传0删除所有消息(此方法不删除 redis cache)
-	DeleteUserSuperGroupMsgsAndSetMinSeq(ctx context.Context, groupID string, userIDs []string, remainTime int64) error
-	// 删除用户消息重置最小seq， remainTime为消息保留的时间单位秒,超时消息删除， 传0删除所有消息(此方法不删除redis cache)
-	DeleteUserMsgsAndSetMinSeq(ctx context.Context, userID string, remainTime int64) error
-	// 获取用户 seq mongo和redis
-	GetUserMinMaxSeqInMongoAndCache(ctx context.Context, userID string) (minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache int64, err error)
-	// 获取群 seq mongo和redis
-	GetSuperGroupMinMaxSeqInMongoAndCache(ctx context.Context, groupID string) (minSeqMongo, maxSeqMongo, maxSeqCache int64, err error)
-	// 设置群用户最小seq 直接调用cache
-	SetGroupUserMinSeq(ctx context.Context, groupID, userID string, minSeq int64) (err error)
-	GetGroupUserMinSeq(ctx context.Context, groupID, userID string) (int64, error)
 
-	// 设置用户最小seq 直接调用cache
-	SetUserMinSeq(ctx context.Context, userID string, minSeq int64) (err error)
+	// DeleteUserSuperGroupMsgsAndSetMinSeq(ctx context.Context, groupID string, userIDs []string, remainTime int64) error
 
+	// 删除会话消息重置最小seq， remainTime为消息保留的时间单位秒,超时消息删除， 传0删除所有消息(此方法不删除redis cache)
+	DeleteConversationMsgsAndSetMinSeq(ctx context.Context, userID string, remainTime int64) error
+
+	// 获取会话 seq mongo和redis
+	GetConversationMinMaxSeqInMongoAndCache(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache int64, err error)
+
+	// msg modify
 	JudgeMessageReactionExist(ctx context.Context, clientMsgID string, sessionType int32) (bool, error)
-
 	SetMessageTypeKeyValue(ctx context.Context, clientMsgID string, sessionType int32, typeKey, value string) error
-
 	SetMessageReactionExpire(ctx context.Context, clientMsgID string, sessionType int32, expiration time.Duration) (bool, error)
 	GetExtendMsg(ctx context.Context, conversationID string, sessionType int32, clientMsgID string, maxMsgUpdateTime int64) (*pbMsg.ExtendMsg, error)
 	InsertOrUpdateReactionExtendMsgSet(ctx context.Context, conversationID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*sdkws.KeyValue) error
@@ -72,13 +65,20 @@ type MsgDatabase interface {
 	GetOneMessageAllReactionList(ctx context.Context, clientMsgID string, sessionType int32) (map[string]string, error)
 	DeleteOneMessageKey(ctx context.Context, clientMsgID string, sessionType int32, subKey string) error
 	DeleteReactionExtendMsgSet(ctx context.Context, conversationID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*sdkws.KeyValue) error
+	// msg send status
 	SetSendMsgStatus(ctx context.Context, id string, status int32) error
 	GetSendMsgStatus(ctx context.Context, id string) (int32, error)
-	GetUserMaxSeq(ctx context.Context, userID string) (int64, error)
-	GetUserMinSeq(ctx context.Context, userID string) (int64, error)
-	GetGroupMaxSeq(ctx context.Context, groupID string) (int64, error)
-	GetGroupMinSeq(ctx context.Context, groupID string) (int64, error)
 
+	SetMaxSeq(ctx context.Context, conversationID string, maxSeq int64) error
+	GetMaxSeqs(ctx context.Context, conversationIDs []string) (map[string]int64, error)
+	GetMaxSeq(ctx context.Context, conversationID string) (int64, error)
+	SetMinSeq(ctx context.Context, conversationID string, minSeq int64) error
+	GetMinSeqs(ctx context.Context, conversationIDs []string) (map[string]int64, error)
+	GetMinSeq(ctx context.Context, conversationID string) (int64, error)
+	GetUserMinSeq(ctx context.Context, conversationIDs []string) (map[string]int64, error)
+	SetUserMinSeq(ctx context.Context, seqs map[string]int64) (err error)
+
+	// to mq
 	MsgToMQ(ctx context.Context, key string, msg2mq *sdkws.MsgData) error
 	MsgToModifyMQ(ctx context.Context, conversationID string, messages []*sdkws.MsgData) error
 	MsgToPushMQ(ctx context.Context, conversationID string, msg2mq *sdkws.MsgData) (int32, int64, error)
@@ -213,24 +213,8 @@ func (db *msgDatabase) MsgToMongoMQ(ctx context.Context, conversationID string,
 	}
 	return nil
 }
-func (db *msgDatabase) GetUserMaxSeq(ctx context.Context, userID string) (int64, error) {
-	return db.cache.GetUserMaxSeq(ctx, userID)
-}
-
-func (db *msgDatabase) GetUserMinSeq(ctx context.Context, userID string) (int64, error) {
-	return db.cache.GetUserMinSeq(ctx, userID)
-}
-
-func (db *msgDatabase) GetGroupMaxSeq(ctx context.Context, groupID string) (int64, error) {
-	return db.cache.GetGroupMaxSeq(ctx, groupID)
-}
-
-func (db *msgDatabase) GetGroupMinSeq(ctx context.Context, groupID string) (int64, error) {
-	return db.cache.GetGroupMinSeq(ctx, groupID)
-}
 
 func (db *msgDatabase) BatchInsertChat2DB(ctx context.Context, conversationID string, msgList []*sdkws.MsgData, currentMaxSeq int64) error {
-	//newTime := utils.GetCurrentTimestampByMill()
 	if int64(len(msgList)) > db.msg.GetSingleGocMsgNum() {
 		return errors.New("too large")
 	}
@@ -309,21 +293,19 @@ func (db *msgDatabase) BatchInsertChat2DB(ctx context.Context, conversationID st
 		}
 		prome.Inc(prome.MsgInsertMongoSuccessCounter)
 	}
-	//log.Debug(operationID, "batch mgo  cost time ", mongo2.getCurrentTimestampByMill()-newTime, userID, len(msgList))
 	return nil
 }
 
-func (db *msgDatabase) DeleteMessageFromCache(ctx context.Context, userID string, msgs []*sdkws.MsgData) error {
-	return db.cache.DeleteMessageFromCache(ctx, userID, msgs)
+func (db *msgDatabase) DeleteMessageFromCache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) error {
+	return db.cache.DeleteMessageFromCache(ctx, conversationID, msgs)
 }
 
 func (db *msgDatabase) NotificationBatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (int64, error) {
 	return 0, nil
 }
 
-func (db *msgDatabase) BatchInsertChat2Cache(ctx context.Context, conversationID string, msgList []*sdkws.MsgData, currentMaxSeq int64) (int64, error) {
-	//newTime := utils.GetCurrentTimestampByMill()
-	lenList := len(msgList)
+func (db *msgDatabase) BatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData, currentMaxSeq int64) (int64, error) {
+	lenList := len(msgs)
 	if int64(lenList) > db.msg.GetSingleGocMsgNum() {
 		return 0, errors.New("too large")
 	}
@@ -332,25 +314,18 @@ func (db *msgDatabase) BatchInsertChat2Cache(ctx context.Context, conversationID
 	}
 	// judge sessionType to get seq
 	lastMaxSeq := currentMaxSeq
-	for _, m := range msgList {
+	for _, m := range msgs {
 		currentMaxSeq++
 		m.Seq = currentMaxSeq
-		//log.Debug(operationID, "cache msg node ", m.String(), m.MsgData.ClientMsgID, "userID: ", conversationID, "seq: ", currentMaxSeq)
 	}
-	//log.Debug(operationID, "SetMessageToCache ", conversationID, len(msgList))
-	failedNum, err := db.cache.SetMessageToCache(ctx, conversationID, msgList)
+	failedNum, err := db.cache.SetMessageToCache(ctx, conversationID, msgs)
 	if err != nil {
 		prome.Add(prome.MsgInsertRedisFailedCounter, failedNum)
-		//log.Error(operationID, "setMessageToCache failed, continue ", err.Error(), len(msgList), conversationID)
+		log.ZError(ctx, "setMessageToCache error", err, "len", len(msgs), "conversationID", conversationID)
 	} else {
 		prome.Inc(prome.MsgInsertRedisSuccessCounter)
 	}
-	//log.Debug(operationID, "batch to redis  cost time ", mongo2.getCurrentTimestampByMill()-newTime, conversationID, len(msgList))
-	if msgList[0].SessionType == constant.SuperGroupChatType {
-		err = db.cache.SetGroupMaxSeq(ctx, conversationID, currentMaxSeq)
-	} else {
-		err = db.cache.SetUserMaxSeq(ctx, conversationID, currentMaxSeq)
-	}
+	err = db.cache.SetMaxSeq(ctx, conversationID, currentMaxSeq)
 	if err != nil {
 		prome.Inc(prome.SeqSetFailedCounter)
 	} else {
@@ -359,9 +334,9 @@ func (db *msgDatabase) BatchInsertChat2Cache(ctx context.Context, conversationID
 	return lastMaxSeq, utils.Wrap(err, "")
 }
 
-func (db *msgDatabase) DelMsgBySeqs(ctx context.Context, userID string, seqs []int64) (totalUnExistSeqs []int64, err error) {
+func (db *msgDatabase) DelMsgBySeqs(ctx context.Context, conversationID string, seqs []int64) (totalUnExistSeqs []int64, err error) {
 	sortkeys.Int64s(seqs)
-	docIDSeqsMap := db.msg.GetDocIDSeqsMap(userID, seqs)
+	docIDSeqsMap := db.msg.GetDocIDSeqsMap(conversationID, seqs)
 	lock := sync.Mutex{}
 	var wg sync.WaitGroup
 	wg.Add(len(docIDSeqsMap))
@@ -456,14 +431,14 @@ func (db *msgDatabase) getMsgBySeqs(ctx context.Context, conversationID string,
 	for docID, value := range m {
 		doc, err := db.msgDocDatabase.FindOneByDocID(ctx, docID)
 		if err != nil {
-			//log.NewError(operationID, "not find seqUid", seqUid, value, uid, seqList, err.Error())
+			log.ZError(ctx, "get message from mongo exception", err, "docID", docID)
 			continue
 		}
 		singleCount = 0
 		for i := 0; i < len(doc.Msg); i++ {
 			msgPb, err := db.unmarshalMsg(&doc.Msg[i])
 			if err != nil {
-				//log.NewError(operationID, "Unmarshal err", seqUid, value, uid, seqList, err.Error())
+				log.ZError(ctx, "unmarshal message exception", err, "docID", docID, "msg", &doc.Msg[i])
 				return nil, err
 			}
 			if utils.Contain(msgPb.Seq, value...) {
@@ -488,12 +463,23 @@ func (db *msgDatabase) getMsgBySeqs(ctx context.Context, conversationID string,
 
 func (db *msgDatabase) getMsgBySeqsRange(ctx context.Context, conversationID string, seqs []int64, begin, end, num int64) (seqMsg []*sdkws.MsgData, err error) {
 	m := db.msg.GetDocIDSeqsMap(conversationID, seqs)
-	for {
+	for int64(len(seqMsg)) != num {
+		for docID, value := range m {
+			beginSeq, endSeq := db.msg.GetSeqsBeginEnd(value)
+			msgs, seqs, err := db.msgDocDatabase.GetMsgBySeqIndexIn1Doc(ctx, docID, beginSeq, endSeq)
+			if err != nil {
+				log.ZError(ctx, "GetMsgBySeqIndexIn1Doc error", err, "docID", docID, "beginSeq", beginSeq, "endSeq", endSeq)
+				continue
+			}
+			var newMsgs []*sdkws.MsgData
+			for _, msg := range msgs {
+				if msg.Status != constant.MsgDeleted {
+					newMsgs = append(newMsgs, msg)
+				}
+			}
+			if int64(len(newMsgs)) != num {
 
-		if int64(len(seqMsg)) != num {
-
-		} else {
-			break
+			}
 		}
 	}
 	return seqMsg, nil
@@ -550,53 +536,52 @@ func (db *msgDatabase) GetMsgBySeqs(ctx context.Context, conversationID string,
 	return successMsgs, nil
 }
 
-func (db *msgDatabase) CleanUpUserMsg(ctx context.Context, userID string) error {
-	err := db.DeleteUserMsgsAndSetMinSeq(ctx, userID, 0)
+func (db *msgDatabase) CleanUpConversationMsgs(ctx context.Context, conversationID string) error {
+	err := db.DeleteConversationMsgsAndSetMinSeq(ctx, conversationID, 0)
 	if err != nil {
 		return err
 	}
-	err = db.cache.CleanUpOneUserAllMsg(ctx, userID)
-	return utils.Wrap(err, "")
+	return db.cache.CleanUpOneUserAllMsg(ctx, conversationID)
 }
 
-func (db *msgDatabase) DeleteUserSuperGroupMsgsAndSetMinSeq(ctx context.Context, groupID string, userIDs []string, remainTime int64) error {
-	var delStruct delMsgRecursionStruct
-	minSeq, err := db.deleteMsgRecursion(ctx, groupID, unRelationTb.OldestList, &delStruct, remainTime)
-	if err != nil {
-		log.ZError(ctx, "deleteMsgRecursion failed", err)
-	}
-	if minSeq == 0 {
-		return nil
-	}
-	//log.NewDebug(operationID, utils.GetSelfFuncName(), "delMsgIDList:", delStruct, "minSeq", minSeq)
-	for _, userID := range userIDs {
-		userMinSeq, err := db.cache.GetGroupUserMinSeq(ctx, groupID, userID)
-		if err != nil && err != redis.Nil {
-			//log.NewError(operationID, utils.GetSelfFuncName(), "GetGroupUserMinSeq failed", groupID, userID, err.Error())
-			continue
-		}
-		if userMinSeq > minSeq {
-			err = db.cache.SetGroupUserMinSeq(ctx, groupID, userID, userMinSeq)
-		} else {
-			err = db.cache.SetGroupUserMinSeq(ctx, groupID, userID, minSeq)
-		}
-		if err != nil {
-			//log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), groupID, userID, userMinSeq, minSeq)
-		}
-	}
-	return nil
-}
+// func (db *msgDatabase) DeleteUserSuperGroupMsgsAndSetMinSeq(ctx context.Context, groupID string, userIDs []string, remainTime int64) error {
+// 	var delStruct delMsgRecursionStruct
+// 	minSeq, err := db.deleteMsgRecursion(ctx, groupID, unRelationTb.OldestList, &delStruct, remainTime)
+// 	if err != nil {
+// 		log.ZError(ctx, "deleteMsgRecursion failed", err)
+// 	}
+// 	if minSeq == 0 {
+// 		return nil
+// 	}
+// 	//log.NewDebug(operationID, utils.GetSelfFuncName(), "delMsgIDList:", delStruct, "minSeq", minSeq)
+// 	for _, userID := range userIDs {
+// 		userMinSeq, err := db.cache.GetGroupUserMinSeq(ctx, groupID, userID)
+// 		if err != nil && err != redis.Nil {
+// 			//log.NewError(operationID, utils.GetSelfFuncName(), "GetGroupUserMinSeq failed", groupID, userID, err.Error())
+// 			continue
+// 		}
+// 		if userMinSeq > minSeq {
+// 			err = db.cache.SetGroupUserMinSeq(ctx, groupID, userID, userMinSeq)
+// 		} else {
+// 			err = db.cache.SetGroupUserMinSeq(ctx, groupID, userID, minSeq)
+// 		}
+// 		if err != nil {
+// 			//log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), groupID, userID, userMinSeq, minSeq)
+// 		}
+// 	}
+// 	return nil
+// }
 
-func (db *msgDatabase) DeleteUserMsgsAndSetMinSeq(ctx context.Context, userID string, remainTime int64) error {
+func (db *msgDatabase) DeleteConversationMsgsAndSetMinSeq(ctx context.Context, conversationID string, remainTime int64) error {
 	var delStruct delMsgRecursionStruct
-	minSeq, err := db.deleteMsgRecursion(ctx, userID, unRelationTb.OldestList, &delStruct, remainTime)
+	minSeq, err := db.deleteMsgRecursion(ctx, conversationID, unRelationTb.OldestList, &delStruct, remainTime)
 	if err != nil {
 		return utils.Wrap(err, "")
 	}
 	if minSeq == 0 {
 		return nil
 	}
-	return db.cache.SetUserMinSeq(ctx, userID, minSeq)
+	return db.cache.SetUserMinSeq(ctx, map[string]int64{conversationID: minSeq})
 }
 
 // this is struct for recursion
@@ -678,35 +663,23 @@ func (db *msgDatabase) deleteMsgRecursion(ctx context.Context, conversationID st
 	return seq, utils.Wrap(err, "deleteMsg failed")
 }
 
-func (db *msgDatabase) GetUserMinMaxSeqInMongoAndCache(ctx context.Context, userID string) (minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache int64, err error) {
-	minSeqMongo, maxSeqMongo, err = db.GetMinMaxSeqMongo(ctx, userID)
+func (db *msgDatabase) GetConversationMinMaxSeqInMongoAndCache(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache int64, err error) {
+	minSeqMongo, maxSeqMongo, err = db.GetMinMaxSeqMongo(ctx, conversationID)
 	if err != nil {
 		return 0, 0, 0, 0, err
 	}
 	// from cache
-	minSeqCache, err = db.cache.GetUserMinSeq(ctx, userID)
+	minSeqCache, err = db.cache.GetUserMinSeq(ctx, conversationID)
 	if err != nil {
 		return 0, 0, 0, 0, err
 	}
-	maxSeqCache, err = db.cache.GetUserMaxSeq(ctx, userID)
+	maxSeqCache, err = db.cache.GetUserMaxSeq(ctx, conversationID)
 	if err != nil {
 		return 0, 0, 0, 0, err
 	}
 	return
 }
 
-func (db *msgDatabase) GetSuperGroupMinMaxSeqInMongoAndCache(ctx context.Context, groupID string) (minSeqMongo, maxSeqMongo, maxSeqCache int64, err error) {
-	minSeqMongo, maxSeqMongo, err = db.GetMinMaxSeqMongo(ctx, groupID)
-	if err != nil {
-		return 0, 0, 0, err
-	}
-	maxSeqCache, err = db.cache.GetGroupMaxSeq(ctx, groupID)
-	if err != nil {
-		return 0, 0, 0, err
-	}
-	return
-}
-
 func (db *msgDatabase) GetMinMaxSeqMongo(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo int64, err error) {
 	oldestMsgMongo, err := db.msgDocDatabase.GetOldestMsg(ctx, conversationID)
 	if err != nil {
@@ -729,14 +702,27 @@ func (db *msgDatabase) GetMinMaxSeqMongo(ctx context.Context, conversationID str
 	return
 }
 
-func (db *msgDatabase) SetGroupUserMinSeq(ctx context.Context, groupID, userID string, minSeq int64) (err error) {
-	return db.cache.SetGroupUserMinSeq(ctx, groupID, userID, minSeq)
+func (db *msgDatabase) SetMaxSeq(ctx context.Context, conversationID string, maxSeq int64) error {
+	return db.cache.SetMaxSeq(ctx, conversationID, maxSeq)
 }
-
-func (db *msgDatabase) SetUserMinSeq(ctx context.Context, userID string, minSeq int64) (err error) {
-	return db.cache.SetUserMinSeq(ctx, userID, minSeq)
+func (db *msgDatabase) GetMaxSeqs(ctx context.Context, conversationIDs []string) (map[string]int64, error) {
+	return db.cache.GetMaxSeqs(ctx, conversationIDs)
 }
-
-func (db *msgDatabase) GetGroupUserMinSeq(ctx context.Context, groupID, userID string) (int64, error) {
-	return db.cache.GetGroupUserMinSeq(ctx, groupID, userID)
+func (db *msgDatabase) GetMaxSeq(ctx context.Context, conversationID string) (int64, error) {
+	return db.cache.GetMaxSeq(ctx, conversationID)
+}
+func (db *msgDatabase) SetMinSeq(ctx context.Context, conversationID string, minSeq int64) error {
+	return db.cache.SetMinSeq(ctx, conversationID, minSeq)
+}
+func (db *msgDatabase) GetMinSeqs(ctx context.Context, conversationIDs []string) (map[string]int64, error) {
+	return db.cache.GetMinSeqs(ctx, conversationIDs)
+}
+func (db *msgDatabase) GetMinSeq(ctx context.Context, conversationID string) (int64, error) {
+	return db.cache.GetMinSeq(ctx, conversationID)
+}
+func (db *msgDatabase) GetUserMinSeq(ctx context.Context, conversationIDs []string) (map[string]int64, error) {
+	return db.cache.GetUserMinSeq(ctx, conversationIDs)
+}
+func (db *msgDatabase) SetUserMinSeq(ctx context.Context, seqs map[string]int64) (err error) {
+	return db.cache.SetUserMinSeq(ctx, seqs)
 }

pkg/common/db/table/unrelation/msg.go

@@ -31,7 +31,7 @@ type MsgDocModelInterface interface {
 	Create(ctx context.Context, model *MsgDocModel) error
 	UpdateMsgStatusByIndexInOneDoc(ctx context.Context, docID string, msg *sdkws.MsgData, seqIndex int, status int32) error
 	FindOneByDocID(ctx context.Context, docID string) (*MsgDocModel, error)
-	GetMsgBySeqIndexIn1Doc(ctx context.Context, conversationID string, begin, end int64) ([]*sdkws.MsgData, error)
+	GetMsgBySeqIndexIn1Doc(ctx context.Context, docID string, beginSeq, endSeq int64) ([]*sdkws.MsgData, []int64, error)
 	GetNewestMsg(ctx context.Context, conversationID string) (*MsgInfoModel, error)
 	GetOldestMsg(ctx context.Context, conversationID string) (*MsgInfoModel, error)
 	Delete(ctx context.Context, docIDs []string) error
@@ -76,14 +76,14 @@ func (m MsgDocModel) GetSeqDocIDList(userID string, maxSeq int64) []string {
 	return seqUserIDs
 }
 
-func (m MsgDocModel) getSeqSuperGroupID(groupID string, seq int64) string {
-	seqSuffix := seq / singleGocMsgNum
-	return m.superGroupIndexGen(groupID, seqSuffix)
-}
+// func (m MsgDocModel) getSeqSuperGroupID(groupID string, seq int64) string {
+// 	seqSuffix := seq / singleGocMsgNum
+// 	return m.superGroupIndexGen(groupID, seqSuffix)
+// }
 
-func (m MsgDocModel) superGroupIndexGen(groupID string, seqSuffix int64) string {
-	return "super_group_" + groupID + ":" + strconv.FormatInt(int64(seqSuffix), 10)
-}
+// func (m MsgDocModel) superGroupIndexGen(groupID string, seqSuffix int64) string {
+// 	return "super_group_" + groupID + ":" + strconv.FormatInt(int64(seqSuffix), 10)
+// }
 
 func (m MsgDocModel) GetDocIDSeqsMap(conversationID string, seqs []int64) map[string][]int64 {
 	t := make(map[string][]int64)
@@ -99,7 +99,14 @@ func (m MsgDocModel) GetDocIDSeqsMap(conversationID string, seqs []int64) map[st
 	return t
 }
 
-func (m MsgDocModel) getMsgIndex(seq int64) int64 {
+func (m MsgDocModel) GetSeqsBeginEnd(seqs []int64) (int64, int64) {
+	if len(seqs) == 0 {
+		return 0, 0
+	}
+	return seqs[0], seqs[len(seqs)-1]
+}
+
+func (m MsgDocModel) GetMsgIndex(seq int64) int64 {
 	seqSuffix := seq / singleGocMsgNum
 	var index int64
 	if seqSuffix == 0 {

pkg/common/db/unrelation/msg.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 
 	table "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/unrelation"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/sdkws"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
@@ -134,24 +135,31 @@ func (m *MsgMongoDriver) UpdateOneDoc(ctx context.Context, msg *table.MsgDocMode
 	return err
 }
 
-func (m *MsgMongoDriver) GetMsgBySeqIndexIn1Doc(ctx context.Context, docID string, begin, end int64) ([]*sdkws.MsgData, error) {
-	// uid = getSeqUid(uid, seq)
-	// seqIndex := getMsgIndex(seq)
-	// m.msg.GetSeqDocIDList()
-	result, err := m.MsgCollection.Find(ctx, bson.M{"doc_id": docID, "msg": bson.M{"$slice": []int{seqIndex, 1}}})
+func (m *MsgMongoDriver) GetMsgBySeqIndexIn1Doc(ctx context.Context, docID string, beginSeq, endSeq int64) (msgs []*sdkws.MsgData, seqs []int64, err error) {
+	beginIndex := m.msg.GetMsgIndex(beginSeq)
+	num := endSeq - beginSeq + 1
+	result, err := m.MsgCollection.Find(ctx, bson.M{"doc_id": docID, "msg": bson.M{"$slice": []int64{beginIndex, num}}})
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	var msgInfos []table.MsgInfoModel
 	if err := result.Decode(&msgInfos); err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	if len(msgInfos) < 1 {
-		return nil, errs.ErrRecordNotFound.Wrap("mongo GetMsgBySeqIndex failed, len is 0")
+		return nil, nil, errs.ErrRecordNotFound.Wrap("mongo GetMsgBySeqIndex failed, len is 0")
 	}
-	var msg sdkws.MsgData
-	if err := proto.Unmarshal(msgInfos[0].Msg, &msg); err != nil {
-		return nil, err
+	for _, v := range msgInfos {
+		var msg sdkws.MsgData
+		if err := proto.Unmarshal(v.Msg, &msg); err != nil {
+			return nil, nil, err
+		}
+		if msg.Seq >= beginSeq && msg.Seq <= endSeq {
+			msgs = append(msgs, &msg)
+			seqs = append(seqs, msg.Seq)
+		} else {
+			log.ZWarn(ctx, "this msg is at wrong position", nil, "msg", &msg)
+		}
 	}
-	return msg, nil
+	return msgs, seqs, nil
 }