new feature data convert (#819)

* new feature: add batch send msg * new feature: add batch send msg * new feature: add batch send msg * new feature: add batch send msg * new feature: add batch send msg * new feature: add batch send msg * fix bug: multiple gateway kick user * fix bug: multiple gateway kick user * fix bug: multiple gateway kick user * fix bug: multiple gateway kick user * fix bug: multiple gateway kick user * MsgDestructTime * fix bug: msg destruct sql * fix bug: msg destruct * fix bug: msg destruct * fix bug: msg destruct sql * fix bug: msg destruct sql * fix bug: msg destruct sql * fix bug: msg destruct sql * debug: print stack * debug: print stack * debug: print stack * fix bug: msg destruct sql Signed-off-by: wangchuxiao <wangchuxiao97@outlook.com> * fix bug: msg notification self 2 self push twice * fix bug: heartbeat get self notification * fix bug: init grpc conn in one process * fix bug: grpc conn Signed-off-by: wangchuxiao <wangchuxiao97@outlook.com> * fix bug: zk client recreate node when reconn * fix bug: set friend mark args error * fix bug: rpc client intercepter called twice Signed-off-by: wangchuxiao <wangchuxiao97@outlook.com> * cicd: robot automated Change Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> * test: document msg num set 100 * new feat: sync designated model * new feat: sync designated model * new feat: sync designated model * new feat: sync designated model * new feat: sync designated model * new feat: sync designated model * new feat: sync designated model * merge code * merge code * fix bug: repeat add friend not effect * feat: msg doc len changed * feat: msg doc len changed * feat: msg doc len changed * feat: msg doc len changed * feat: msg doc len changed * feat: msg doc len changed * feat: msg doc len changed * feat: msg doc len changed * feat: msg doc len changed --------- Signed-off-by: wangchuxiao <wangchuxiao97@outlook.com> Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Co-authored-by: wangchuxiao-dev <wangchuxiao-dev@users.noreply.github.com>
2026-05-09 11:35:59 +08:00 · 2023-08-09 10:37:56 +08:00
parent 04a97ac154
commit 5615e47d99
11 changed files with 153 additions and 20 deletions
@@ -118,6 +118,7 @@ type CommonMsgDatabase interface {
 		pageNumber int32,
 		showNumber int32,
 	) (msgCount int64, userCount int64, groups []*unRelationTb.GroupCount, dateCount map[string]int64, err error)
+	ConvertMsgsDocLen(ctx context.Context, conversationIDs []string)
 }

 func NewCommonMsgDatabase(msgDocModel unRelationTb.MsgDocModelInterface, cacheModel cache.MsgModel) CommonMsgDatabase {
@@ -965,3 +966,7 @@ func (db *commonMsgDatabase) SearchMessage(ctx context.Context, req *pbMsg.Searc
 	}
 	return total, totalMsgs, nil
 }
+
+func (db *commonMsgDatabase) ConvertMsgsDocLen(ctx context.Context, conversationIDs []string) {
+	db.msgDocDatabase.ConvertMsgsDocLen(ctx, conversationIDs)
+}
@@ -86,7 +86,11 @@ func (u *UserGorm) Page(

 // 获取所有用户ID.
 func (u *UserGorm) GetAllUserID(ctx context.Context, pageNumber, showNumber int32) (userIDs []string, err error) {
-	return userIDs, errs.Wrap(u.db(ctx).Limit(int(showNumber)).Offset(int((pageNumber-1)*showNumber)).Pluck("user_id", &userIDs).Error)
+	if pageNumber == 0 || showNumber == 0 {
+		return userIDs, errs.Wrap(u.db(ctx).Pluck("user_id", &userIDs).Error)
+	} else {
+		return userIDs, errs.Wrap(u.db(ctx).Limit(int(showNumber)).Offset(int((pageNumber-1)*showNumber)).Pluck("user_id", &userIDs).Error)
+	}
 }

 func (u *UserGorm) GetUserGlobalRecvMsgOpt(ctx context.Context, userID string) (opt int, err error) {
@@ -27,10 +27,11 @@ import (
 )

 const (
-	singleGocMsgNum = 5000
-	Msg             = "msg"
-	OldestList      = 0
-	NewestList      = -1
+	singleGocMsgNum     = 100
+	singleGocMsgNum5000 = 5000
+	Msg                 = "msg"
+	OldestList          = 0
+	NewestList          = -1
 )

 type MsgDocModel struct {
@@ -128,6 +129,7 @@ type MsgDocModelInterface interface {
 		pageNumber int32,
 		showNumber int32,
 	) (msgCount int64, userCount int64, groups []*GroupCount, dateCount map[string]int64, err error)
+	ConvertMsgsDocLen(ctx context.Context, conversationIDs []string)
 }

 func (MsgDocModel) TableName() string {
@@ -138,6 +140,10 @@ func (MsgDocModel) GetSingleGocMsgNum() int64 {
 	return singleGocMsgNum
 }

+func (MsgDocModel) GetSingleGocMsgNum5000() int64 {
+	return singleGocMsgNum5000
+}
+
 func (m *MsgDocModel) IsFull() bool {
 	return m.Msg[len(m.Msg)-1].Msg != nil
 }
@@ -0,0 +1,67 @@
+package unrelation
+
+import (
+	"context"
+	"fmt"
+
+	table "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/unrelation"
+	"github.com/OpenIMSDK/tools/log"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+)
+
+func (m *MsgMongoDriver) ConvertMsgsDocLen(ctx context.Context, conversationIDs []string) {
+	for _, conversationID := range conversationIDs {
+		regex := primitive.Regex{Pattern: fmt.Sprintf("^%s:", conversationID)}
+		cursor, err := m.MsgCollection.Find(ctx, bson.M{"doc_id": regex})
+		if err != nil {
+			log.ZError(ctx, "convertAll find msg doc failed", err, "conversationID", conversationID)
+			continue
+		}
+		var msgDocs []table.MsgDocModel
+		err = cursor.All(ctx, &msgDocs)
+		if err != nil {
+			log.ZError(ctx, "convertAll cursor all failed", err, "conversationID", conversationID)
+			continue
+		}
+		if len(msgDocs) < 1 {
+			continue
+		}
+		log.ZInfo(ctx, "msg doc convert", "conversationID", conversationID, "len(msgDocs)", len(msgDocs))
+		if len(msgDocs[0].Msg) == int(m.model.GetSingleGocMsgNum5000()) {
+			if _, err := m.MsgCollection.DeleteMany(ctx, bson.M{"doc_id": regex}); err != nil {
+				log.ZError(ctx, "convertAll delete many failed", err, "conversationID", conversationID)
+				continue
+			}
+			var newMsgDocs []interface{}
+			for _, msgDoc := range msgDocs {
+				if int64(len(msgDoc.Msg)) == m.model.GetSingleGocMsgNum() {
+					continue
+				}
+				var index int64
+				for index < int64(len(msgDoc.Msg)) {
+					msg := msgDoc.Msg[index]
+					if msg != nil && msg.Msg != nil {
+						msgDocModel := table.MsgDocModel{DocID: m.model.GetDocID(conversationID, msg.Msg.Seq)}
+						end := index + m.model.GetSingleGocMsgNum()
+						if int(end) >= len(msgDoc.Msg) {
+							msgDocModel.Msg = msgDoc.Msg[index:]
+						} else {
+							msgDocModel.Msg = msgDoc.Msg[index:end]
+						}
+						newMsgDocs = append(newMsgDocs, msgDocModel)
+						index = end
+					} else {
+						break
+					}
+				}
+			}
+			_, err = m.MsgCollection.InsertMany(ctx, newMsgDocs)
+			if err != nil {
+				log.ZError(ctx, "convertAll insert many failed", err, "conversationID", conversationID, "len(newMsgDocs)", len(newMsgDocs))
+			} else {
+				log.ZInfo(ctx, "msg doc convert", "conversationID", conversationID, "len(newMsgDocs)", len(newMsgDocs))
+			}
+		}
+	}
+}