feat: Optimizing RPC call (#2993)

* pb

* fix: Modifying other fields while setting IsPrivateChat does not take effect

* fix: quote message error revoke

* refactoring scheduled tasks

* refactoring scheduled tasks

* refactoring scheduled tasks

* refactoring scheduled tasks

* refactoring scheduled tasks

* refactoring scheduled tasks

* rpc client

* rpc client

* rpc client

* rpc client

* rpc client

* rpc client

* rpc client

* rpc client

internal/rpc/msg/clear.go

@@ -2,23 +2,13 @@ package msg
 
 import (
 	"context"
-	"strings"
-	"time"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/convert"
-	pbconversation "github.com/openimsdk/protocol/conversation"
 	"github.com/openimsdk/protocol/msg"
-	"github.com/openimsdk/protocol/wrapperspb"
 	"github.com/openimsdk/tools/log"
-	"github.com/openimsdk/tools/mcontext"
-	"github.com/openimsdk/tools/utils/datautil"
-	"github.com/openimsdk/tools/utils/idutil"
-	"github.com/openimsdk/tools/utils/stringutil"
-	"golang.org/x/sync/errgroup"
+	"strings"
 )
 
-// hard delete in Database.
+// DestructMsgs hard delete in Database.
 func (m *msgServer) DestructMsgs(ctx context.Context, req *msg.DestructMsgsReq) (*msg.DestructMsgsResp, error) {
 	if err := authverify.CheckAdmin(ctx, m.config.Share.IMAdminUserID); err != nil {
 		return nil, err
@@ -61,67 +51,6 @@ func (m *msgServer) DestructMsgs(ctx context.Context, req *msg.DestructMsgsReq)
 	return &msg.DestructMsgsResp{Count: int32(len(docs))}, nil
 }
 
-// soft delete for user self
-func (m *msgServer) ClearMsg(ctx context.Context, req *msg.ClearMsgReq) (*msg.ClearMsgResp, error) {
-	temp := convert.ConversationsPb2DB(req.Conversations)
-
-	batchNum := 100
-
-	errg, _ := errgroup.WithContext(ctx)
-	errg.SetLimit(100)
-
-	for i := 0; i < len(temp); i += batchNum {
-		batch := temp[i:min(i+batchNum, len(temp))]
-
-		errg.Go(func() error {
-			for _, conversation := range batch {
-				handleCtx := mcontext.NewCtx(stringutil.GetSelfFuncName() + "-" + idutil.OperationIDGenerator() + "-" + conversation.ConversationID + "-" + conversation.OwnerUserID)
-				log.ZDebug(handleCtx, "User MsgsDestruct",
-					"conversationID", conversation.ConversationID,
-					"ownerUserID", conversation.OwnerUserID,
-					"msgDestructTime", conversation.MsgDestructTime,
-					"lastMsgDestructTime", conversation.LatestMsgDestructTime)
-
-				seqs, err := m.MsgDatabase.ClearUserMsgs(handleCtx, conversation.OwnerUserID, conversation.ConversationID, conversation.MsgDestructTime, conversation.LatestMsgDestructTime)
-				if err != nil {
-					log.ZError(handleCtx, "user msg destruct failed", err, "conversationID", conversation.ConversationID, "ownerUserID", conversation.OwnerUserID)
-					continue
-				}
-
-				if len(seqs) > 0 {
-					minseq := datautil.Max(seqs...)
-
-					// update
-					if err := m.Conversation.UpdateConversation(handleCtx,
-						&pbconversation.UpdateConversationReq{
-							UserIDs:               []string{conversation.OwnerUserID},
-							ConversationID:        conversation.ConversationID,
-							LatestMsgDestructTime: wrapperspb.Int64(time.Now().UnixMilli()),
-							MinSeq:                wrapperspb.Int64(minseq),
-						}); err != nil {
-						log.ZError(handleCtx, "updateUsersConversationField failed", err, "conversationID", conversation.ConversationID, "ownerUserID", conversation.OwnerUserID)
-						continue
-					}
-
-					if err := m.Conversation.SetConversationMinSeq(handleCtx, []string{conversation.OwnerUserID}, conversation.ConversationID, minseq); err != nil {
-						return err
-					}
-
-					// if you need Notify SDK client userseq is update.
-					// m.msgNotificationSender.UserDeleteMsgsNotification(handleCtx, conversation.OwnerUserID, conversation.ConversationID, seqs)
-				}
-			}
-			return nil
-		})
-	}
-
-	if err := errg.Wait(); err != nil {
-		return nil, err
-	}
-
-	return nil, nil
-}
-
 func (m *msgServer) GetLastMessageSeqByTime(ctx context.Context, req *msg.GetLastMessageSeqByTimeReq) (*msg.GetLastMessageSeqByTimeResp, error) {
 	seq, err := m.MsgDatabase.GetLastMessageSeqByTime(ctx, req.ConversationID, req.Time)
 	if err != nil {

internal/rpc/msg/delete.go

@@ -74,19 +74,13 @@ func (m *msgServer) DeleteMsgs(ctx context.Context, req *msg.DeleteMsgsReq) (*ms
 		if err := m.MsgDatabase.DeleteMsgsPhysicalBySeqs(ctx, req.ConversationID, req.Seqs); err != nil {
 			return nil, err
 		}
-		conversations, err := m.Conversation.GetConversationsByConversationID(ctx, []string{req.ConversationID})
+		conv, err := m.conversationClient.GetConversationsByConversationID(ctx, req.ConversationID)
 		if err != nil {
 			return nil, err
 		}
 		tips := &sdkws.DeleteMsgsTips{UserID: req.UserID, ConversationID: req.ConversationID, Seqs: req.Seqs}
-		m.notificationSender.NotificationWithSessionType(
-			ctx,
-			req.UserID,
-			m.conversationAndGetRecvID(conversations[0], req.UserID),
-			constant.DeleteMsgsNotification,
-			conversations[0].ConversationType,
-			tips,
-		)
+		m.notificationSender.NotificationWithSessionType(ctx, req.UserID, m.conversationAndGetRecvID(conv, req.UserID),
+			constant.DeleteMsgsNotification, conv.ConversationType, tips)
 	} else {
 		if err := m.MsgDatabase.DeleteUserMsgsBySeqs(ctx, req.UserID, req.ConversationID, req.Seqs); err != nil {
 			return nil, err
@@ -121,7 +115,7 @@ func (m *msgServer) DeleteMsgPhysical(ctx context.Context, req *msg.DeleteMsgPhy
 }
 
 func (m *msgServer) clearConversation(ctx context.Context, conversationIDs []string, userID string, deleteSyncOpt *msg.DeleteSyncOpt) error {
-	conversations, err := m.Conversation.GetConversationsByConversationID(ctx, conversationIDs)
+	conversations, err := m.conversationClient.GetConversationsByConversationIDs(ctx, conversationIDs)
 	if err != nil {
 		return err
 	}
@@ -144,7 +138,7 @@ func (m *msgServer) clearConversation(ctx context.Context, conversationIDs []str
 		}
 		ownerUserIDs := []string{userID}
 		for conversationID, seq := range setSeqs {
-			if err := m.Conversation.SetConversationMinSeq(ctx, ownerUserIDs, conversationID, seq); err != nil {
+			if err := m.conversationClient.SetConversationMinSeq(ctx, conversationID, ownerUserIDs, seq); err != nil {
 				return err
 			}
 		}

internal/rpc/msg/notification.go

@@ -17,7 +17,7 @@ package msg
 import (
 	"context"
 
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
+	"github.com/openimsdk/open-im-server/v3/pkg/notification"
 	"github.com/openimsdk/protocol/constant"
 	"github.com/openimsdk/protocol/sdkws"
 )

internal/rpc/msg/send.go

@@ -113,19 +113,14 @@ func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgDa
 			conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAll}
 		} else { // @Everyone and @other people
 			conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAllAtMe}
-
-			err = m.Conversation.SetConversations(ctx, atUserID, conversation)
-			if err != nil {
+			if err := m.conversationClient.SetConversations(ctx, atUserID, conversation); err != nil {
 				log.ZWarn(ctx, "SetConversations", err, "userID", atUserID, "conversation", conversation)
 			}
-
 			memberUserIDList = datautil.Single(atUserID, memberUserIDList)
 		}
 
 		conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAll}
-
-		err = m.Conversation.SetConversations(ctx, memberUserIDList, conversation)
-		if err != nil {
+		if err := m.conversationClient.SetConversations(ctx, memberUserIDList, conversation); err != nil {
 			log.ZWarn(ctx, "SetConversations", err, "userID", memberUserIDList, "conversation", conversation)
 		}
 
@@ -133,8 +128,7 @@ func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgDa
 	}
 	conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtMe}
 
-	err := m.Conversation.SetConversations(ctx, msg.AtUserIDList, conversation)
-	if err != nil {
+	if err := m.conversationClient.SetConversations(ctx, msg.AtUserIDList, conversation); err != nil {
 		log.ZWarn(ctx, "SetConversations", err, msg.AtUserIDList, conversation)
 	}
 }

internal/rpc/msg/server.go

@@ -16,6 +16,8 @@ package msg
 
 import (
 	"context"
+	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo"
@@ -25,8 +27,8 @@ import (
 	"github.com/openimsdk/tools/db/redisutil"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller"
+	"github.com/openimsdk/open-im-server/v3/pkg/notification"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpccache"
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 	"github.com/openimsdk/protocol/constant"
 	"github.com/openimsdk/protocol/conversation"
 	"github.com/openimsdk/protocol/msg"
@@ -35,39 +37,39 @@ import (
 )
 
 type MessageInterceptorFunc func(ctx context.Context, globalConfig *Config, req *msg.SendMsgReq) (*sdkws.MsgData, error)
-type (
-	// MessageInterceptorChain defines a chain of message interceptor functions.
-	MessageInterceptorChain []MessageInterceptorFunc
 
-	// MsgServer encapsulates dependencies required for message handling.
-	msgServer struct {
-		RegisterCenter         discovery.SvcDiscoveryRegistry   // Service discovery registry for service registration.
-		MsgDatabase            controller.CommonMsgDatabase     // Interface for message database operations.
-		Conversation           *rpcclient.ConversationRpcClient // RPC client for conversation service.
-		UserLocalCache         *rpccache.UserLocalCache         // Local cache for user data.
-		FriendLocalCache       *rpccache.FriendLocalCache       // Local cache for friend data.
-		GroupLocalCache        *rpccache.GroupLocalCache        // Local cache for group data.
-		ConversationLocalCache *rpccache.ConversationLocalCache // Local cache for conversation data.
-		Handlers               MessageInterceptorChain          // Chain of handlers for processing messages.
-		notificationSender     *rpcclient.NotificationSender    // RPC client for sending notifications.
-		msgNotificationSender  *MsgNotificationSender           // RPC client for sending msg notifications.
-		config                 *Config                          // Global configuration settings.
-		webhookClient          *webhook.Client
-		msg.UnimplementedMsgServer
-	}
+// MessageInterceptorChain defines a chain of message interceptor functions.
+type MessageInterceptorChain []MessageInterceptorFunc
 
-	Config struct {
-		RpcConfig          config.Msg
-		RedisConfig        config.Redis
-		MongodbConfig      config.Mongo
-		KafkaConfig        config.Kafka
-		NotificationConfig config.Notification
-		Share              config.Share
-		WebhooksConfig     config.Webhooks
-		LocalCacheConfig   config.LocalCache
-		Discovery          config.Discovery
-	}
-)
+type Config struct {
+	RpcConfig          config.Msg
+	RedisConfig        config.Redis
+	MongodbConfig      config.Mongo
+	KafkaConfig        config.Kafka
+	NotificationConfig config.Notification
+	Share              config.Share
+	WebhooksConfig     config.Webhooks
+	LocalCacheConfig   config.LocalCache
+	Discovery          config.Discovery
+}
+
+// MsgServer encapsulates dependencies required for message handling.
+type msgServer struct {
+	msg.UnimplementedMsgServer
+	RegisterCenter         discovery.SvcDiscoveryRegistry // Service discovery registry for service registration.
+	MsgDatabase            controller.CommonMsgDatabase   // Interface for message database operations.
+	StreamMsgDatabase      controller.StreamMsgDatabase
+	UserLocalCache         *rpccache.UserLocalCache         // Local cache for user data.
+	FriendLocalCache       *rpccache.FriendLocalCache       // Local cache for friend data.
+	GroupLocalCache        *rpccache.GroupLocalCache        // Local cache for group data.
+	ConversationLocalCache *rpccache.ConversationLocalCache // Local cache for conversation data.
+	Handlers               MessageInterceptorChain          // Chain of handlers for processing messages.
+	notificationSender     *rpcclient.NotificationSender    // RPC client for sending notifications.
+	msgNotificationSender  *MsgNotificationSender           // RPC client for sending msg notifications.
+	config                 *Config                          // Global configuration settings.
+	webhookClient          *webhook.Client
+	conversationClient     *rpcli.ConversationClient
+}
 
 func (m *msgServer) addInterceptorHandler(interceptorFunc ...MessageInterceptorFunc) {
 	m.Handlers = append(m.Handlers, interceptorFunc...)
@@ -106,16 +108,34 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 	if err != nil {
 		return err
 	}
+	userConn, err := client.GetConn(ctx, config.Discovery.RpcService.User)
+	if err != nil {
+		return err
+	}
+	groupConn, err := client.GetConn(ctx, config.Discovery.RpcService.Group)
+	if err != nil {
+		return err
+	}
+	friendConn, err := client.GetConn(ctx, config.Discovery.RpcService.Friend)
+	if err != nil {
+		return err
+	}
+	conversationConn, err := client.GetConn(ctx, config.Discovery.RpcService.Conversation)
+	if err != nil {
+		return err
+	}
+	conversationClient := rpcli.NewConversationClient(conversationConn)
 	s := &msgServer{
 		Conversation:           &conversationClient,
 		MsgDatabase:            msgDatabase,
 		RegisterCenter:         client,
-		UserLocalCache:         rpccache.NewUserLocalCache(userRpcClient, &config.LocalCacheConfig, rdb),
-		GroupLocalCache:        rpccache.NewGroupLocalCache(groupRpcClient, &config.LocalCacheConfig, rdb),
+		UserLocalCache:         rpccache.NewUserLocalCache(rpcli.NewUserClient(userConn), &config.LocalCacheConfig, rdb),
+		GroupLocalCache:        rpccache.NewGroupLocalCache(rpcli.NewGroupClient(groupConn), &config.LocalCacheConfig, rdb),
 		ConversationLocalCache: rpccache.NewConversationLocalCache(conversationClient, &config.LocalCacheConfig, rdb),
-		FriendLocalCache:       rpccache.NewFriendLocalCache(friendRpcClient, &config.LocalCacheConfig, rdb),
+		FriendLocalCache:       rpccache.NewFriendLocalCache(rpcli.NewRelationClient(friendConn), &config.LocalCacheConfig, rdb),
 		config:                 config,
 		webhookClient:          webhook.NewWebhookClient(config.WebhooksConfig.URL),
+		conversationClient:     conversationClient,
 	}
 
 	s.notificationSender = rpcclient.NewNotificationSender(&config.NotificationConfig, rpcclient.WithLocalSendMsg(s.SendMsg))

internal/rpc/msg/stream_msg.go

@@ -0,0 +1,115 @@
+package msg
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
+	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
+	"github.com/openimsdk/protocol/constant"
+	"github.com/openimsdk/protocol/msg"
+	"github.com/openimsdk/protocol/sdkws"
+	"github.com/openimsdk/tools/errs"
+)
+
+const StreamDeadlineTime = time.Second * 60 * 10
+
+func (m *msgServer) handlerStreamMsg(ctx context.Context, msgData *sdkws.MsgData) error {
+	now := time.Now()
+	val := &model.StreamMsg{
+		ClientMsgID:    msgData.ClientMsgID,
+		ConversationID: msgprocessor.GetConversationIDByMsg(msgData),
+		UserID:         msgData.SendID,
+		CreateTime:     now,
+		DeadlineTime:   now.Add(StreamDeadlineTime),
+	}
+	return m.StreamMsgDatabase.CreateStreamMsg(ctx, val)
+}
+
+func (m *msgServer) getStreamMsg(ctx context.Context, clientMsgID string) (*model.StreamMsg, error) {
+	res, err := m.StreamMsgDatabase.GetStreamMsg(ctx, clientMsgID)
+	if err != nil {
+		return nil, err
+	}
+	now := time.Now()
+	if !res.End && res.DeadlineTime.Before(now) {
+		res.End = true
+		res.DeadlineTime = now
+		_ = m.StreamMsgDatabase.AppendStreamMsg(ctx, res.ClientMsgID, 0, nil, true, now)
+	}
+	return res, nil
+}
+
+func (m *msgServer) AppendStreamMsg(ctx context.Context, req *msg.AppendStreamMsgReq) (*msg.AppendStreamMsgResp, error) {
+	res, err := m.getStreamMsg(ctx, req.ClientMsgID)
+	if err != nil {
+		return nil, err
+	}
+	if res.End {
+		return nil, errs.ErrNoPermission.WrapMsg("stream msg is end")
+	}
+	if len(res.Packets) < int(req.StartIndex) {
+		return nil, errs.ErrNoPermission.WrapMsg("start index is invalid")
+	}
+	if val := len(res.Packets) - int(req.StartIndex); val > 0 {
+		exist := res.Packets[int(req.StartIndex):]
+		for i, s := range exist {
+			if len(req.Packets) == 0 {
+				break
+			}
+			if s != req.Packets[i] {
+				return nil, errs.ErrNoPermission.WrapMsg(fmt.Sprintf("packet %d has been written and is inconsistent", i))
+			}
+			req.StartIndex++
+			req.Packets = req.Packets[1:]
+		}
+	}
+	if len(req.Packets) == 0 && res.End == req.End {
+		return &msg.AppendStreamMsgResp{}, nil
+	}
+	deadlineTime := time.Now().Add(StreamDeadlineTime)
+	if err := m.StreamMsgDatabase.AppendStreamMsg(ctx, req.ClientMsgID, int(req.StartIndex), req.Packets, req.End, deadlineTime); err != nil {
+		return nil, err
+	}
+	conversation, err := m.conversationClient.GetConversation(ctx, res.ConversationID, res.UserID)
+	if err != nil {
+		return nil, err
+	}
+	tips := &sdkws.StreamMsgTips{
+		ConversationID: res.ConversationID,
+		ClientMsgID:    res.ClientMsgID,
+		StartIndex:     req.StartIndex,
+		Packets:        req.Packets,
+		End:            req.End,
+	}
+	var (
+		recvID      string
+		sessionType int32
+	)
+	if conversation.GroupID == "" {
+		sessionType = constant.SingleChatType
+		recvID = conversation.UserID
+	} else {
+		sessionType = constant.ReadGroupChatType
+		recvID = conversation.GroupID
+	}
+	m.msgNotificationSender.StreamMsgNotification(ctx, res.UserID, recvID, sessionType, tips)
+	return &msg.AppendStreamMsgResp{}, nil
+}
+
+func (m *msgServer) GetStreamMsg(ctx context.Context, req *msg.GetStreamMsgReq) (*msg.GetStreamMsgResp, error) {
+	res, err := m.getStreamMsg(ctx, req.ClientMsgID)
+	if err != nil {
+		return nil, err
+	}
+	return &msg.GetStreamMsgResp{
+		ClientMsgID:    res.ClientMsgID,
+		ConversationID: res.ConversationID,
+		UserID:         res.UserID,
+		Packets:        res.Packets,
+		End:            res.End,
+		CreateTime:     res.CreateTime.UnixMilli(),
+		DeadlineTime:   res.DeadlineTime.UnixMilli(),
+	}, nil
+}