Merge branch 'OpenIMSDK:main' into main

cmd/openim-crontask/main.go

@@ -21,7 +21,7 @@ import (
 
 func main() {
 	cronTaskCmd := cmd.NewCronTaskCmd()
-	if err := cronTaskCmd.Exec(tools.StartCronTask); err != nil {
+	if err := cronTaskCmd.Exec(tools.StartTask); err != nil {
 		panic(err.Error())
 	}
 }

go.work.sum

@@ -1,9 +1,6 @@
-github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
-github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prashantv/gostub v1.1.0/go.mod h1:A5zLQHz7ieHGG7is6LLXLz7I8+3LZzsrV0P1IAHhP5U=
-github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+go.uber.org/automaxprocs v1.5.3/go.mod h1:eRbA25aqJrxAbsLO0xy5jVwPt7FQnRgjW+efnwa1WM0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

internal/msgtransfer/online_history_msg_handler.go

@@ -145,7 +145,7 @@ func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
 					len(modifyMsgList),
 				)
 				conversationIDMsg := msgprocessor.GetChatConversationIDByMsg(ctxMsgList[0].message)
-				conversationIDNotification := msgprocessor.GetNotificationConversationID(ctxMsgList[0].message)
+				conversationIDNotification := msgprocessor.GetNotificationConversationIDByMsg(ctxMsgList[0].message)
 				och.handleMsg(ctx, msgChannelValue.uniqueKey, conversationIDMsg, storageMsgList, notStorageMsgList)
 				och.handleNotification(
 					ctx,

internal/rpc/msg/sync_msg.go

@@ -116,41 +116,75 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq
 	if total, chatLogs, err = m.MsgDatabase.SearchMessage(ctx, req); err != nil {
 		return nil, err
 	}
+
+	var (
+		sendIDs  []string
+		recvIDs  []string
+		groupIDs []string
+		sendMap  = make(map[string]string)
+		recvMap  = make(map[string]string)
+		groupMap = make(map[string]*sdkws.GroupInfo)
+	)
+	for _, chatLog := range chatLogs {
+		if chatLog.SenderNickname == "" {
+			sendIDs = append(sendIDs, chatLog.SendID)
+		}
+		switch chatLog.SessionType {
+		case constant.SingleChatType:
+			recvIDs = append(recvIDs, chatLog.RecvID)
+		case constant.GroupChatType, constant.SuperGroupChatType:
+			groupIDs = append(groupIDs, chatLog.GroupID)
+		}
+	}
+	if len(sendIDs) != 0 {
+		sendInfos, err := m.User.GetUsersInfo(ctx, sendIDs)
+		if err != nil {
+			return nil, err
+		}
+		for _, sendInfo := range sendInfos {
+			sendMap[sendInfo.UserID] = sendInfo.Nickname
+		}
+	}
+	if len(recvIDs) != 0 {
+		recvInfos, err := m.User.GetUsersInfo(ctx, recvIDs)
+		if err != nil {
+			return nil, err
+		}
+		for _, recvInfo := range recvInfos {
+			recvMap[recvInfo.UserID] = recvInfo.Nickname
+		}
+	}
+	if len(groupIDs) != 0 {
+		groupInfos, err := m.Group.GetGroupInfos(ctx, groupIDs, true)
+		if err != nil {
+			return nil, err
+		}
+		for _, groupInfo := range groupInfos {
+			groupMap[groupInfo.GroupID] = groupInfo
+		}
+	}
 	for _, chatLog := range chatLogs {
 		pbChatLog := &msg.ChatLog{}
 		utils.CopyStructFields(pbChatLog, chatLog)
 		pbChatLog.SendTime = chatLog.SendTime
 		pbChatLog.CreateTime = chatLog.CreateTime
 		if chatLog.SenderNickname == "" {
-			sendUser, err := m.User.GetUserInfo(ctx, chatLog.SendID)
+			pbChatLog.SenderNickname = sendMap[chatLog.SendID]
-			if err != nil {
-				return nil, err
-			}
-			pbChatLog.SenderNickname = sendUser.Nickname
 		}
 		switch chatLog.SessionType {
 		case constant.SingleChatType:
-			recvUser, err := m.User.GetUserInfo(ctx, chatLog.RecvID)
+			pbChatLog.RecvNickname = recvMap[chatLog.RecvID]
-			if err != nil {
-				return nil, err
-			}
-			pbChatLog.RecvNickname = recvUser.Nickname
 
 		case constant.GroupChatType, constant.SuperGroupChatType:
-			group, err := m.Group.GetGroupInfo(ctx, chatLog.GroupID)
+			pbChatLog.SenderFaceURL = groupMap[chatLog.GroupID].FaceURL
-			if err != nil {
+			pbChatLog.GroupMemberCount = groupMap[chatLog.GroupID].MemberCount
-				return nil, err
+			pbChatLog.RecvID = groupMap[chatLog.GroupID].GroupID
-			}
+			pbChatLog.GroupName = groupMap[chatLog.GroupID].GroupName
-			pbChatLog.SenderFaceURL = group.FaceURL
+			pbChatLog.GroupOwner = groupMap[chatLog.GroupID].OwnerUserID
-			pbChatLog.GroupMemberCount = group.MemberCount
+			pbChatLog.GroupType = groupMap[chatLog.GroupID].GroupType
-			pbChatLog.RecvID = group.GroupID
-			pbChatLog.GroupName = group.GroupName
-			pbChatLog.GroupOwner = group.OwnerUserID
-			pbChatLog.GroupType = group.GroupType
 		}
 		resp.ChatLogs = append(resp.ChatLogs, pbChatLog)
 	}
-
 	resp.ChatLogsNum = total
 	return resp, nil
 }

internal/rpc/user/user.go

@@ -20,16 +20,12 @@ import (
 	"strings"
 	"time"
 
-	"github.com/OpenIMSDK/tools/log"
-
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/authverify"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/unrelation"
-
 	"github.com/OpenIMSDK/protocol/constant"
 	"github.com/OpenIMSDK/protocol/sdkws"
-	pbuser "github.com/OpenIMSDK/protocol/user"
-	registry "github.com/OpenIMSDK/tools/discoveryregistry"
 	"github.com/OpenIMSDK/tools/errs"
+	"github.com/OpenIMSDK/tools/log"
 	"github.com/OpenIMSDK/tools/tx"
 
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
@@ -40,10 +36,11 @@ import (
 	tablerelation "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/rpcclient"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/rpcclient/notification"
+	registry "github.com/OpenIMSDK/tools/discoveryregistry"
 
-	"google.golang.org/grpc"
+	pbuser "github.com/OpenIMSDK/protocol/user"
-
 	"github.com/OpenIMSDK/tools/utils"
+	"google.golang.org/grpc"
 )
 
 type userServer struct {

internal/tools/cron_task.go

@@ -26,12 +26,13 @@ import (
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
 )
 
-func StartCronTask() error {
+func StartTask() error {
 	fmt.Println("cron task start, config", config.Config.ChatRecordsClearTime)
 	msgTool, err := InitMsgTool()
 	if err != nil {
 		return err
 	}
+	msgTool.ConvertTools()
 	c := cron.New()
 	var wg sync.WaitGroup
 	wg.Add(1)

internal/tools/msg_doc_convert.go

@@ -0,0 +1,32 @@
+package tools
+
+import (
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/msgprocessor"
+	"github.com/OpenIMSDK/protocol/constant"
+	"github.com/OpenIMSDK/tools/log"
+	"github.com/OpenIMSDK/tools/mcontext"
+)
+
+func (c *MsgTool) ConvertTools() {
+	ctx := mcontext.NewCtx("convert")
+	conversationIDs, err := c.conversationDatabase.GetAllConversationIDs(ctx)
+	if err != nil {
+		log.ZError(ctx, "get all conversation ids failed", err)
+		return
+	}
+	for _, conversationID := range conversationIDs {
+		conversationIDs = append(conversationIDs, msgprocessor.GetNotificationConversationIDByConversationID(conversationID))
+	}
+	userIDs, err := c.userDatabase.GetAllUserID(ctx, 0, 0)
+	if err != nil {
+		log.ZError(ctx, "get all user ids failed", err)
+		return
+	}
+	log.ZDebug(ctx, "all userIDs", "len userIDs", len(userIDs))
+	for _, userID := range userIDs {
+		conversationIDs = append(conversationIDs, msgprocessor.GetConversationIDBySessionType(constant.SingleChatType, userID, userID))
+		conversationIDs = append(conversationIDs, msgprocessor.GetNotificationConversationID(constant.SingleChatType, userID, userID))
+	}
+	log.ZDebug(ctx, "all conversationIDs", "len userIDs", len(conversationIDs))
+	c.msgDatabase.ConvertMsgsDocLen(ctx, conversationIDs)
+}

pkg/common/db/controller/msg.go

@@ -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 {
@@ -961,7 +962,14 @@ func (db *commonMsgDatabase) SearchMessage(ctx context.Context, req *pbMsg.Searc
 		return 0, nil, err
 	}
 	for _, msg := range msgs {
+		if msg.IsRead {
+			msg.Msg.IsRead = true
+		}
 		totalMsgs = append(totalMsgs, convert.MsgDB2Pb(msg.Msg))
 	}
 	return total, totalMsgs, nil
 }
+
+func (db *commonMsgDatabase) ConvertMsgsDocLen(ctx context.Context, conversationIDs []string) {
+	db.msgDocDatabase.ConvertMsgsDocLen(ctx, conversationIDs)
+}

pkg/common/db/relation/user_model.go

@@ -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) {

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

@@ -27,10 +27,11 @@ import (
 )
 
 const (
-	singleGocMsgNum = 5000
+	singleGocMsgNum     = 100
-	Msg             = "msg"
+	singleGocMsgNum5000 = 5000
-	OldestList      = 0
+	Msg                 = "msg"
-	NewestList      = -1
+	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
 }

pkg/common/db/unrelation/msg.go

@@ -1073,11 +1073,6 @@ func (m *MsgMongoDriver) SearchMessage(ctx context.Context, req *msg.SearchMessa
 	if err != nil {
 		return 0, nil, err
 	}
-	for _, msg1 := range msgs {
-		if msg1.IsRead {
-			msg1.Msg.IsRead = true
-		}
-	}
 	return total, msgs, nil
 }
 
@@ -1151,13 +1146,22 @@ func (m *MsgMongoDriver) searchMessage(ctx context.Context, req *msg.SearchMessa
 				{"doc_id", 1},
 			}},
 		},
+		{
+			{"$unwind", bson.M{"path": "$msgs"}},
+		},
+		{
+			{"$sort", bson.M{"msgs.msg.send_time": -1}},
+		},
 	}
 	cursor, err := m.MsgCollection.Aggregate(ctx, pipe)
 	if err != nil {
 		return 0, nil, err
 	}
-
+	type docModel struct {
-	var msgsDocs []table.MsgDocModel
+		DocID string              `bson:"doc_id"`
+		Msg   *table.MsgInfoModel `bson:"msgs"`
+	}
+	var msgsDocs []docModel
 	err = cursor.All(ctx, &msgsDocs)
 	if err != nil {
 		return 0, nil, err
@@ -1167,41 +1171,39 @@ func (m *MsgMongoDriver) searchMessage(ctx context.Context, req *msg.SearchMessa
 	}
 	msgs := make([]*table.MsgInfoModel, 0)
 	for index := range msgsDocs {
-		for i := range msgsDocs[index].Msg {
+		msgInfo := msgsDocs[index].Msg
-			msg := msgsDocs[index].Msg[i]
+		if msgInfo == nil || msgInfo.Msg == nil {
-			if msg == nil || msg.Msg == nil {
+			continue
-				continue
-			}
-			if msg.Revoke != nil {
-				revokeContent := sdkws.MessageRevokedContent{
-					RevokerID:                   msg.Revoke.UserID,
-					RevokerRole:                 msg.Revoke.Role,
-					ClientMsgID:                 msg.Msg.ClientMsgID,
-					RevokerNickname:             msg.Revoke.Nickname,
-					RevokeTime:                  msg.Revoke.Time,
-					SourceMessageSendTime:       msg.Msg.SendTime,
-					SourceMessageSendID:         msg.Msg.SendID,
-					SourceMessageSenderNickname: msg.Msg.SenderNickname,
-					SessionType:                 msg.Msg.SessionType,
-					Seq:                         msg.Msg.Seq,
-					Ex:                          msg.Msg.Ex,
-				}
-				data, err := json.Marshal(&revokeContent)
-				if err != nil {
-					return 0, nil, err
-				}
-				elem := sdkws.NotificationElem{
-					Detail: string(data),
-				}
-				content, err := json.Marshal(&elem)
-				if err != nil {
-					return 0, nil, err
-				}
-				msg.Msg.ContentType = constant.MsgRevokeNotification
-				msg.Msg.Content = string(content)
-			}
-			msgs = append(msgs, msg)
 		}
+		if msgInfo.Revoke != nil {
+			revokeContent := sdkws.MessageRevokedContent{
+				RevokerID:                   msgInfo.Revoke.UserID,
+				RevokerRole:                 msgInfo.Revoke.Role,
+				ClientMsgID:                 msgInfo.Msg.ClientMsgID,
+				RevokerNickname:             msgInfo.Revoke.Nickname,
+				RevokeTime:                  msgInfo.Revoke.Time,
+				SourceMessageSendTime:       msgInfo.Msg.SendTime,
+				SourceMessageSendID:         msgInfo.Msg.SendID,
+				SourceMessageSenderNickname: msgInfo.Msg.SenderNickname,
+				SessionType:                 msgInfo.Msg.SessionType,
+				Seq:                         msgInfo.Msg.Seq,
+				Ex:                          msgInfo.Msg.Ex,
+			}
+			data, err := json.Marshal(&revokeContent)
+			if err != nil {
+				return 0, nil, err
+			}
+			elem := sdkws.NotificationElem{
+				Detail: string(data),
+			}
+			content, err := json.Marshal(&elem)
+			if err != nil {
+				return 0, nil, err
+			}
+			msgInfo.Msg.ContentType = constant.MsgRevokeNotification
+			msgInfo.Msg.Content = string(content)
+		}
+		msgs = append(msgs, msgInfo)
 	}
 	start := (req.Pagination.PageNumber - 1) * req.Pagination.ShowNumber
 	n := int32(len(msgs))

pkg/common/db/unrelation/msg_convert.go

@@ -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))
+			}
+		}
+	}
+}

pkg/msgprocessor/conversation.go

@@ -23,7 +23,7 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
-func GetNotificationConversationID(msg *sdkws.MsgData) string {
+func GetNotificationConversationIDByMsg(msg *sdkws.MsgData) string {
 	switch msg.SessionType {
 	case constant.SingleChatType:
 		l := []string{msg.SendID, msg.RecvID}
@@ -114,6 +114,30 @@ func GetConversationIDBySessionType(sessionType int, ids ...string) string {
 	return ""
 }
 
+func GetNotificationConversationIDByConversationID(conversationID string) string {
+	l := strings.Split(conversationID, "_")
+	if len(l) > 1 {
+		l[0] = "n"
+		return strings.Join(l, "_")
+	} else {
+		return ""
+	}
+}
+
+func GetNotificationConversationID(sessionType int, ids ...string) string {
+	sort.Strings(ids)
+	if len(ids) > 2 || len(ids) < 1 {
+		return ""
+	}
+	switch sessionType {
+	case constant.SingleChatType:
+		return "n_" + strings.Join(ids, "_") // single chat
+	case constant.SuperGroupChatType:
+		return "n_" + ids[0] // super group chat
+	}
+	return ""
+}
+
 func IsNotification(conversationID string) bool {
 	return strings.HasPrefix(conversationID, "n_")
 }

scripts/start.bat

@@ -0,0 +1,12 @@
+cd %~p0../_output/bin/platforms/windows
+start api.exe -p 10002
+start auth.exe -p 10060
+start conversation.exe -p 10080
+start friend.exe -p 10020
+start group.exe -p 10050
+start msg.exe -p 10030
+start msggateway.exe -p 10040 -w 10001
+start msgtransfer.exe
+start third.exe -p 10090
+start push.exe -p 10070
+start user.exe -p 10010