feat: add local cache for high frequency reads (#2036)

* feat: msg local cache * feat: msg local cache * feat: msg local cache * feat: msg local cache * feat: msg local cache * feat: msg local cache * fix: mongo * fix: mongo * fix: mongo * openim.yaml * localcache * localcache * localcache * localcache * localcache * localcache * localcache * localcache * localcache * local cache * local cache * local cache * local cache * fix: GroupApplicationAcceptedNotification * fix: GroupApplicationAcceptedNotification * fix: NotificationUserInfoUpdate * feat: cache add single-flight and timing-wheel. * feat: local cache * feat: local cache * feat: local cache * feat: cache add single-flight and timing-wheel. * feat: local cache * feat: local cache * feat: local cache * feat: local cache * feat: local cache * feat: local cache * feat: local cache * feat: local cache * feat: local cache * feat: local cache * feat: local cache * feat: local cache * feat: local cache * feat: local cache * feat: local cache * feat: local cache * feat: local cache * feat: msg rpc local cache * feat: msg rpc local cache * feat: msg rpc local cache * feat: msg rpc local cache * feat: msg rpc local cache * feat: msg rpc local cache * refactor: refactor the code of push and optimization. * cicd: robot automated Change * refactor: rename cache. * merge * fix: refactor project dir avoid import cycle. * update tools * merge * feat: conversation FindRecvMsgNotNotifyUserIDs * feat: conversation FindRecvMsgNotNotifyUserIDs * feat: conversation FindRecvMsgNotNotifyUserIDs * merge * merge the latest main --------- Co-authored-by: Gordon <46924906+FGadvancer@users.noreply.github.com> Co-authored-by: withchao <withchao@users.noreply.github.com>
2026-05-15 06:25:58 +08:00 · 2024-03-08 16:30:47 +08:00
parent 291443dd6b
commit b9cf40034c
60 changed files with 1860 additions and 390 deletions
@@ -41,7 +41,7 @@ func (m *msgServer) GetConversationsHasReadAndMaxSeq(ctx context.Context, req *m
 	if err != nil {
 		return nil, err
 	}
-	conversations, err := m.Conversation.GetConversations(ctx, req.UserID, conversationIDs)
+	conversations, err := m.ConversationLocalCache.GetConversations(ctx, req.UserID, conversationIDs)
 	if err != nil {
 		return nil, err
 	}
@@ -104,7 +104,7 @@ func (m *msgServer) MarkMsgsAsRead(
 	if hasReadSeq > maxSeq {
 		return nil, errs.ErrArgs.Wrap("hasReadSeq must not be bigger than maxSeq")
 	}
-	conversation, err := m.Conversation.GetConversation(ctx, req.UserID, req.ConversationID)
+	conversation, err := m.ConversationLocalCache.GetConversation(ctx, req.UserID, req.ConversationID)
 	if err != nil {
 		return
 	}
@@ -144,7 +144,7 @@ func (m *msgServer) MarkConversationAsRead(
 	ctx context.Context,
 	req *msg.MarkConversationAsReadReq,
 ) (resp *msg.MarkConversationAsReadResp, err error) {
-	conversation, err := m.Conversation.GetConversation(ctx, req.UserID, req.ConversationID)
+	conversation, err := m.ConversationLocalCache.GetConversation(ctx, req.UserID, req.ConversationID)
 	if err != nil {
 		return nil, err
 	}
@@ -44,7 +44,7 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
 	if err := authverify.CheckAccessV3(ctx, req.UserID, m.config); err != nil {
 		return nil, err
 	}
-	user, err := m.User.GetUserInfo(ctx, req.UserID)
+	user, err := m.UserLocalCache.GetUserInfo(ctx, req.UserID)
 	if err != nil {
 		return nil, err
 	}
@@ -70,12 +70,7 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
 			}
 			role = user.AppMangerLevel
 		case constant.SuperGroupChatType:
-			members, err := m.Group.GetGroupMemberInfoMap(
-				ctx,
-				msgs[0].GroupID,
-				utils.Distinct([]string{req.UserID, msgs[0].SendID}),
-				true,
-			)
+			members, err := m.GroupLocalCache.GetGroupMemberInfoMap(ctx, msgs[0].GroupID, utils.Distinct([]string{req.UserID, msgs[0].SendID}))
 			if err != nil {
 				return nil, err
 			}
@@ -97,7 +97,7 @@ func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgDa
 	}
 	tagAll := utils.IsContain(constant.AtAllString, msg.AtUserIDList)
 	if tagAll {
-		memberUserIDList, err := m.Group.GetGroupMemberIDs(ctx, msg.GroupID)
+		memberUserIDList, err := m.GroupLocalCache.GetGroupMemberIDs(ctx, msg.GroupID)
 		if err != nil {
 			log.ZWarn(ctx, "GetGroupMemberIDs", err)
 			return
@@ -143,6 +143,7 @@ func (m *msgServer) sendMsgNotification(
 }

 func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq) (resp *pbmsg.SendMsgResp, err error) {
+	log.ZDebug(ctx, "sendMsgSingleChat return")
 	if err := m.messageVerification(ctx, req); err != nil {
 		return nil, err
 	}
@@ -15,6 +15,8 @@
 package msg

 import (
+	"github.com/openimsdk/open-im-server/v3/pkg/rpccache"
+
 	"github.com/OpenIMSDK/protocol/constant"
 	"github.com/OpenIMSDK/protocol/conversation"
 	"github.com/OpenIMSDK/protocol/msg"
@@ -22,7 +24,6 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/controller"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/db/localcache"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/unrelation"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 	"google.golang.org/grpc"
@@ -33,12 +34,11 @@ type (
 	msgServer               struct {
 		RegisterCenter         discoveryregistry.SvcDiscoveryRegistry
 		MsgDatabase            controller.CommonMsgDatabase
-		Group                  *rpcclient.GroupRpcClient
-		User                   *rpcclient.UserRpcClient
 		Conversation           *rpcclient.ConversationRpcClient
-		friend                 *rpcclient.FriendRpcClient
-		GroupLocalCache        *localcache.GroupLocalCache
-		ConversationLocalCache *localcache.ConversationLocalCache
+		UserLocalCache         *rpccache.UserLocalCache
+		FriendLocalCache       *rpccache.FriendLocalCache
+		GroupLocalCache        *rpccache.GroupLocalCache
+		ConversationLocalCache *rpccache.ConversationLocalCache
 		Handlers               MessageInterceptorChain
 		notificationSender     *rpcclient.NotificationSender
 		config                 *config.GlobalConfig
@@ -84,13 +84,12 @@ func Start(config *config.GlobalConfig, client discoveryregistry.SvcDiscoveryReg
 	}
 	s := &msgServer{
 		Conversation:           &conversationClient,
-		User:                   &userRpcClient,
-		Group:                  &groupRpcClient,
 		MsgDatabase:            msgDatabase,
 		RegisterCenter:         client,
-		GroupLocalCache:        localcache.NewGroupLocalCache(&groupRpcClient),
-		ConversationLocalCache: localcache.NewConversationLocalCache(&conversationClient),
-		friend:                 &friendRpcClient,
+		UserLocalCache:         rpccache.NewUserLocalCache(userRpcClient, rdb),
+		GroupLocalCache:        rpccache.NewGroupLocalCache(groupRpcClient, rdb),
+		ConversationLocalCache: rpccache.NewConversationLocalCache(conversationClient, rdb),
+		FriendLocalCache:       rpccache.NewFriendLocalCache(friendRpcClient, rdb),
 		config:                 config,
 	}
 	s.notificationSender = rpcclient.NewNotificationSender(config, rpcclient.WithLocalSendMsg(s.SendMsg))
@@ -40,7 +40,7 @@ func (m *msgServer) GetActiveUser(ctx context.Context, req *msg.GetActiveUserReq
 	var pbUsers []*msg.ActiveUser
 	if len(users) > 0 {
 		userIDs := utils.Slice(users, func(e *unrelation.UserCount) string { return e.UserID })
-		userMap, err := m.User.GetUsersInfoMap(ctx, userIDs)
+		userMap, err := m.UserLocalCache.GetUsersInfoMap(ctx, userIDs)
 		if err != nil {
 			return nil, err
 		}
@@ -82,7 +82,7 @@ func (m *msgServer) GetActiveGroup(ctx context.Context, req *msg.GetActiveGroupR
 	var pbgroups []*msg.ActiveGroup
 	if len(groups) > 0 {
 		groupIDs := utils.Slice(groups, func(e *unrelation.GroupCount) string { return e.GroupID })
-		resp, err := m.Group.GetGroupInfos(ctx, groupIDs, false)
+		resp, err := m.GroupLocalCache.GetGroupInfos(ctx, groupIDs)
 		if err != nil {
 			return nil, err
 		}
@@ -35,7 +35,7 @@ func (m *msgServer) PullMessageBySeqs(
 	resp.NotificationMsgs = make(map[string]*sdkws.PullMsgs)
 	for _, seq := range req.SeqRanges {
 		if !msgprocessor.IsNotification(seq.ConversationID) {
-			conversation, err := m.Conversation.GetConversation(ctx, req.UserID, seq.ConversationID)
+			conversation, err := m.ConversationLocalCache.GetConversation(ctx, req.UserID, seq.ConversationID)
 			if err != nil {
 				log.ZError(ctx, "GetConversation error", err, "conversationID", seq.ConversationID)
 				continue
@@ -138,7 +138,7 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq
 		}
 	}
 	if len(sendIDs) != 0 {
-		sendInfos, err := m.User.GetUsersInfo(ctx, sendIDs)
+		sendInfos, err := m.UserLocalCache.GetUsersInfo(ctx, sendIDs)
 		if err != nil {
 			return nil, err
 		}
@@ -147,7 +147,7 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq
 		}
 	}
 	if len(recvIDs) != 0 {
-		recvInfos, err := m.User.GetUsersInfo(ctx, recvIDs)
+		recvInfos, err := m.UserLocalCache.GetUsersInfo(ctx, recvIDs)
 		if err != nil {
 			return nil, err
 		}
@@ -156,7 +156,7 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq
 		}
 	}
 	if len(groupIDs) != 0 {
-		groupInfos, err := m.Group.GetGroupInfos(ctx, groupIDs, true)
+		groupInfos, err := m.GroupLocalCache.GetGroupInfos(ctx, groupIDs)
 		if err != nil {
 			return nil, err
 		}
@@ -16,6 +16,7 @@ package msg

 import (
 	"context"
+	"github.com/OpenIMSDK/tools/log"
 	"math/rand"
 	"strconv"
 	"time"
@@ -59,15 +60,15 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
 			data.MsgData.ContentType >= constant.NotificationBegin {
 			return nil
 		}
-		black, err := m.friend.IsBlocked(ctx, data.MsgData.SendID, data.MsgData.RecvID)
+		black, err := m.FriendLocalCache.IsBlack(ctx, data.MsgData.SendID, data.MsgData.RecvID)
 		if err != nil {
 			return err
 		}
 		if black {
 			return errs.ErrBlockedByPeer.Wrap()
 		}
-		if *m.config.MessageVerify.FriendVerify {
-			friend, err := m.friend.IsFriend(ctx, data.MsgData.SendID, data.MsgData.RecvID)
+		if m.config.MessageVerify.FriendVerify != nil && *m.config.MessageVerify.FriendVerify {
+			friend, err := m.FriendLocalCache.IsFriend(ctx, data.MsgData.SendID, data.MsgData.RecvID)
 			if err != nil {
 				return err
 			}
@@ -78,7 +79,7 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
 		}
 		return nil
 	case constant.SuperGroupChatType:
-		groupInfo, err := m.Group.GetGroupInfoCache(ctx, data.MsgData.GroupID)
+		groupInfo, err := m.GroupLocalCache.GetGroupInfo(ctx, data.MsgData.GroupID)
 		if err != nil {
 			return err
 		}
@@ -99,17 +100,17 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
 			data.MsgData.ContentType >= constant.NotificationBegin {
 			return nil
 		}
-		// memberIDs, err := m.GroupLocalCache.GetGroupMemberIDs(ctx, data.MsgData.GroupID)
-		// if err != nil {
-		// 	return err
-		// }
-		// if !utils.IsContain(data.MsgData.SendID, memberIDs) {
-		// 	return errs.ErrNotInGroupYet.Wrap()
-		// }
-
-		groupMemberInfo, err := m.Group.GetGroupMemberCache(ctx, data.MsgData.GroupID, data.MsgData.SendID)
+		memberIDs, err := m.GroupLocalCache.GetGroupMemberIDMap(ctx, data.MsgData.GroupID)
 		if err != nil {
-			if err == errs.ErrRecordNotFound {
+			return err
+		}
+		if _, ok := memberIDs[data.MsgData.SendID]; !ok {
+			return errs.ErrNotInGroupYet.Wrap()
+		}
+
+		groupMemberInfo, err := m.GroupLocalCache.GetGroupMember(ctx, data.MsgData.GroupID, data.MsgData.SendID)
+		if err != nil {
+			if errs.ErrRecordNotFound.Is(err) {
 				return errs.ErrNotInGroupYet.Wrap(err.Error())
 			}
 			return err
@@ -157,6 +158,9 @@ func (m *msgServer) encapsulateMsgData(msg *sdkws.MsgData) {
 	case constant.Custom:
 		fallthrough
 	case constant.Quote:
+		utils.SetSwitchFromOptions(msg.Options, constant.IsConversationUpdate, true)
+		utils.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, true)
+		utils.SetSwitchFromOptions(msg.Options, constant.IsSenderSync, true)
 	case constant.Revoke:
 		utils.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, false)
 		utils.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, false)
@@ -187,7 +191,8 @@ func (m *msgServer) modifyMessageByUserMessageReceiveOpt(
 	sessionType int,
 	pb *msg.SendMsgReq,
 ) (bool, error) {
-	opt, err := m.User.GetUserGlobalMsgRecvOpt(ctx, userID)
+	defer log.ZDebug(ctx, "modifyMessageByUserMessageReceiveOpt return")
+	opt, err := m.UserLocalCache.GetUserGlobalMsgRecvOpt(ctx, userID)
 	if err != nil {
 		return false, err
 	}
@@ -203,7 +208,7 @@ func (m *msgServer) modifyMessageByUserMessageReceiveOpt(
 		return true, nil
 	}
 	// conversationID := utils.GetConversationIDBySessionType(conversationID, sessionType)
-	singleOpt, err := m.Conversation.GetSingleConversationRecvMsgOpt(ctx, userID, conversationID)
+	singleOpt, err := m.ConversationLocalCache.GetSingleConversationRecvMsgOpt(ctx, userID, conversationID)
 	if errs.ErrRecordNotFound.Is(err) {
 		return true, nil
 	} else if err != nil {