feat: add a function for business info change to update related conve… (#3225)

* feat: add a function for business info change to update related conversation's ex info.

* feat: add a function for business info change to update related conversation's ex info.

* feat: add a function for business info change to update related conversation's ex info.

* feat: add a function for business info change to update related conversation's ex info.

internal/rpc/msg/notification.go

@@ -48,7 +48,3 @@ func (m *MsgNotificationSender) MarkAsReadNotification(ctx context.Context, conv
 	}
 	m.NotificationWithSessionType(ctx, sendID, recvID, constant.HasReadReceipt, sessionType, tips)
 }
-
-func (m *MsgNotificationSender) StreamMsgNotification(ctx context.Context, sendID string, recvID string, sessionType int32, tips *sdkws.StreamMsgTips) {
-	m.NotificationWithSessionType(ctx, sendID, recvID, constant.StreamMsgNotification, sessionType, tips)
-}

internal/rpc/msg/send.go

@@ -17,6 +17,8 @@ package msg
 import (
 	"context"
 
+	"google.golang.org/protobuf/proto"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
 	"github.com/openimsdk/open-im-server/v3/pkg/util/conversationutil"
@@ -29,7 +31,6 @@ import (
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/mcontext"
 	"github.com/openimsdk/tools/utils/datautil"
-	"google.golang.org/protobuf/proto"
 )
 
 func (m *msgServer) SendMsg(ctx context.Context, req *pbmsg.SendMsgReq) (*pbmsg.SendMsgResp, error) {
@@ -49,11 +50,6 @@ func (m *msgServer) SendMsg(ctx context.Context, req *pbmsg.SendMsgReq) (*pbmsg.
 
 func (m *msgServer) sendMsg(ctx context.Context, req *pbmsg.SendMsgReq, before **sdkws.MsgData) (*pbmsg.SendMsgResp, error) {
 	m.encapsulateMsgData(req.MsgData)
-	if req.MsgData.ContentType == constant.Stream {
-		if err := m.handlerStreamMsg(ctx, req.MsgData); err != nil {
-			return nil, err
-		}
-	}
 	switch req.MsgData.SessionType {
 	case constant.SingleChatType:
 		return m.sendMsgSingleChat(ctx, req, before)

internal/rpc/msg/server.go

@@ -59,9 +59,8 @@ type Config struct {
 // MsgServer encapsulates dependencies required for message handling.
 type msgServer struct {
 	msg.UnimplementedMsgServer
-	RegisterCenter         discovery.Conn               // Service discovery registry for service registration.
-	MsgDatabase            controller.CommonMsgDatabase // Interface for message database operations.
-	StreamMsgDatabase      controller.StreamMsgDatabase
+	RegisterCenter         discovery.Conn                   // Service discovery registry for service registration.
+	MsgDatabase            controller.CommonMsgDatabase     // Interface for message database operations.
 	UserLocalCache         *rpccache.UserLocalCache         // Local cache for user data.
 	FriendLocalCache       *rpccache.FriendLocalCache       // Local cache for friend data.
 	GroupLocalCache        *rpccache.GroupLocalCache        // Local cache for group data.
@@ -117,10 +116,6 @@ func Start(ctx context.Context, config *Config, client discovery.Conn, server gr
 	if err != nil {
 		return err
 	}
-	streamMsg, err := mgo.NewStreamMsgMongo(mgocli.GetDB())
-	if err != nil {
-		return err
-	}
 	seqUserCache := redis.NewSeqUserCacheRedis(rdb, seqUser)
 	userConn, err := client.GetConn(ctx, config.Discovery.RpcService.User)
 	if err != nil {
@@ -142,7 +137,6 @@ func Start(ctx context.Context, config *Config, client discovery.Conn, server gr
 	msgDatabase := controller.NewCommonMsgDatabase(msgDocModel, msgModel, seqUserCache, seqConversationCache, redisProducer)
 	s := &msgServer{
 		MsgDatabase:            msgDatabase,
-		StreamMsgDatabase:      controller.NewStreamMsgDatabase(streamMsg),
 		RegisterCenter:         client,
 		UserLocalCache:         rpccache.NewUserLocalCache(rpcli.NewUserClient(userConn), &config.LocalCacheConfig, rdb),
 		GroupLocalCache:        rpccache.NewGroupLocalCache(rpcli.NewGroupClient(groupConn), &config.LocalCacheConfig, rdb),

internal/rpc/msg/stream_msg.go

@@ -1,115 +0,0 @@
-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
-}