fix: solve uncorrect outdated msg get. (#2513)

* refactor: refactor workflows contents.

* add tool workflows.

* update field.

* fix: remove chat error.

* Fix err.

* fix error.

* remove cn comment.

* update workflows files.

* update infra config.

* move workflows.

* feat: update bot.

* fix: solve uncorrect outdated msg get.

* update get docIDs logic.

* update

* update skip logic.

* fix

* update.

pkg/common/storage/controller/msg.go

@@ -18,11 +18,12 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 	"strings"
 	"time"
 
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
+
 	"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/prommetrics"
@@ -97,8 +98,10 @@ type CommonMsgDatabase interface {
 	ConvertMsgsDocLen(ctx context.Context, conversationIDs []string)
 
 	// clear msg
-	GetBeforeMsg(ctx context.Context, ts int64, limit int) ([]*model.MsgDocModel, error)
+	GetBeforeMsg(ctx context.Context, ts int64, docIds []string, limit int) ([]*model.MsgDocModel, error)
 	DeleteDocMsgBefore(ctx context.Context, ts int64, doc *model.MsgDocModel) ([]int, error)
+
+	GetDocIDs(ctx context.Context) ([]string, error)
 }
 
 func NewCommonMsgDatabase(msgDocModel database.Msg, msg cache.MsgCache, seqUser cache.SeqUser, seqConversation cache.SeqConversationCache, kafkaConf *config.Kafka) (CommonMsgDatabase, error) {
@@ -912,8 +915,25 @@ func (db *commonMsgDatabase) ConvertMsgsDocLen(ctx context.Context, conversation
 	db.msgDocDatabase.ConvertMsgsDocLen(ctx, conversationIDs)
 }
 
-func (db *commonMsgDatabase) GetBeforeMsg(ctx context.Context, ts int64, limit int) ([]*model.MsgDocModel, error) {
-	return db.msgDocDatabase.GetBeforeMsg(ctx, ts, limit)
+func (db *commonMsgDatabase) GetBeforeMsg(ctx context.Context, ts int64, docIDs []string, limit int) ([]*model.MsgDocModel, error) {
+	var msgs []*model.MsgDocModel
+	for i := 0; i < len(docIDs); i += 1000 {
+		end := i + 1000
+		if end > len(docIDs) {
+			end = len(docIDs)
+		}
+
+		res, err := db.msgDocDatabase.GetBeforeMsg(ctx, ts, docIDs[i:end], limit)
+		if err != nil {
+			return nil, err
+		}
+		msgs = append(msgs, res...)
+
+		if len(msgs) >= limit {
+			return msgs[:limit], nil
+		}
+	}
+	return msgs, nil
 }
 
 func (db *commonMsgDatabase) DeleteDocMsgBefore(ctx context.Context, ts int64, doc *model.MsgDocModel) ([]int, error) {
@@ -936,8 +956,10 @@ func (db *commonMsgDatabase) DeleteDocMsgBefore(ctx context.Context, ts int64, d
 		return index, err
 	}
 	if len(index) == notNull {
+		log.ZDebug(ctx, "Delete db in Doc", "DocID", doc.DocID, "index", index, "maxSeq", maxSeq)
 		return index, db.msgDocDatabase.DeleteDoc(ctx, doc.DocID)
 	} else {
+		log.ZDebug(ctx, "delete db in index", "DocID", doc.DocID, "index", index, "maxSeq", maxSeq)
 		return index, db.msgDocDatabase.DeleteMsgByIndex(ctx, doc.DocID, index)
 	}
 }
@@ -955,3 +977,7 @@ func (db *commonMsgDatabase) setMinSeq(ctx context.Context, conversationID strin
 	}
 	return db.seqConversation.SetMinSeq(ctx, conversationID, seq)
 }
+
+func (db *commonMsgDatabase) GetDocIDs(ctx context.Context) ([]string, error) {
+	return db.msgDocDatabase.GetDocIDs(ctx)
+}

pkg/common/storage/database/mgo/msg.go

@@ -8,6 +8,7 @@ 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/openimsdk/tools/utils/datautil"
+	"golang.org/x/exp/rand"
 
 	"github.com/openimsdk/protocol/constant"
 	"github.com/openimsdk/protocol/msg"
@@ -1226,10 +1227,53 @@ func (m *MsgMgo) ConvertMsgsDocLen(ctx context.Context, conversationIDs []string
 	}
 }
 
-func (m *MsgMgo) GetBeforeMsg(ctx context.Context, ts int64, limit int) ([]*model.MsgDocModel, error) {
+func (m *MsgMgo) GetDocIDs(ctx context.Context) ([]string, error) {
+	limit := 5000
+	var skip int
+	var docIDs []string
+	var offset int
+
+	count, err := m.coll.CountDocuments(ctx, bson.M{})
+	if err != nil {
+		return nil, err
+	}
+
+	if count < int64(limit) {
+		skip = 0
+	} else {
+		rand.Seed(uint64(time.Now().UnixMilli()))
+		skip = rand.Intn(int(count / int64(limit)))
+		offset = skip * limit
+	}
+	log.ZDebug(ctx, "offset", "skip", skip, "offset", offset)
+	res, err := mongoutil.Aggregate[*model.MsgDocModel](ctx, m.coll, []bson.M{
+		{
+			"$project": bson.M{
+				"doc_id": 1,
+			},
+		},
+		{
+			"$skip": offset,
+		},
+		{
+			"$limit": limit,
+		},
+	})
+
+	for _, doc := range res {
+		docIDs = append(docIDs, doc.DocID)
+	}
+
+	return docIDs, errs.Wrap(err)
+}
+
+func (m *MsgMgo) GetBeforeMsg(ctx context.Context, ts int64, docIDs []string, limit int) ([]*model.MsgDocModel, error) {
 	return mongoutil.Aggregate[*model.MsgDocModel](ctx, m.coll, []bson.M{
 		{
 			"$match": bson.M{
+				"doc_id": bson.M{
+					"$in": docIDs,
+				},
 				"msgs.msg.send_time": bson.M{
 					"$lt": ts,
 				},

pkg/common/storage/database/msg.go

@@ -16,10 +16,11 @@ package database
 
 import (
 	"context"
+	"time"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 	"github.com/openimsdk/protocol/msg"
 	"go.mongodb.org/mongo-driver/mongo"
-	"time"
 )
 
 type Msg interface {
@@ -44,5 +45,7 @@ type Msg interface {
 
 	DeleteDoc(ctx context.Context, docID string) error
 	DeleteMsgByIndex(ctx context.Context, docID string, index []int) error
-	GetBeforeMsg(ctx context.Context, ts int64, limit int) ([]*model.MsgDocModel, error)
+	GetBeforeMsg(ctx context.Context, ts int64, docIDs []string, limit int) ([]*model.MsgDocModel, error)
+
+	GetDocIDs(ctx context.Context) ([]string, error)
 }