Merge remote-tracking branch 'origin/errcode' into errcode

internal/msgtransfer/init.go

@@ -87,7 +87,7 @@ func (m *MsgTransfer) Start(prometheusPort int) error {
 	wg.Add(1)
 	fmt.Println("start msg transfer", "prometheusPort:", prometheusPort)
 	if config.Config.ChatPersistenceMysql {
-		go m.persistentCH.persistentConsumerGroup.RegisterHandleAndConsumer(m.persistentCH)
+		// go m.persistentCH.persistentConsumerGroup.RegisterHandleAndConsumer(m.persistentCH)
 	} else {
 		fmt.Println("msg transfer not start mysql consumer")
 	}

internal/msgtransfer/modify_msg_handler.go

@@ -53,15 +53,14 @@ func (mmc *ModifyMsgConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSessi
 }
 
 func (mmc *ModifyMsgConsumerHandler) ModifyMsg(ctx context.Context, cMsg *sarama.ConsumerMessage, msgKey string, _ sarama.ConsumerGroupSession) {
-	log.NewInfo("msg come here ModifyMsg!!!", "", "msg", string(cMsg.Value), msgKey)
 	msgFromMQ := pbMsg.MsgDataToModifyByMQ{}
 	operationID := mcontext.GetOperationID(ctx)
 	err := proto.Unmarshal(cMsg.Value, &msgFromMQ)
 	if err != nil {
-		log.NewError(msgFromMQ.TriggerID, "msg_transfer Unmarshal msg err", "msg", string(cMsg.Value), "err", err.Error())
+		log.ZError(ctx, "msg_transfer Unmarshal msg err", err, "msg", string(cMsg.Value))
 		return
 	}
-	log.Debug(msgFromMQ.TriggerID, "proto.Unmarshal MsgDataToMQ", msgFromMQ.String())
+	log.ZDebug(ctx, "proto.Unmarshal MsgDataToMQ", "msgs", msgFromMQ.String())
 	for _, msg := range msgFromMQ.Messages {
 		isReactionFromCache := utils.GetSwitchFromOptions(msg.Options, constant.IsReactionFromCache)
 		if !isReactionFromCache {
@@ -74,7 +73,6 @@ func (mmc *ModifyMsgConsumerHandler) ModifyMsg(ctx context.Context, cMsg *sarama
 				continue
 			}
 			if notification.IsExternalExtensions {
-				log.NewInfo(operationID, "msg:", notification, "this is external extensions")
 				continue
 			}
 			if !notification.IsReact {
@@ -94,12 +92,12 @@ func (mmc *ModifyMsgConsumerHandler) ModifyMsg(ctx context.Context, cMsg *sarama
 				}
 
 				if err := mmc.extendMsgDatabase.InsertExtendMsg(ctx, notification.ConversationID, notification.SessionType, &extendMsg); err != nil {
-					log.NewError(operationID, "MsgFirstModify InsertExtendMsg failed", notification.ConversationID, notification.SessionType, extendMsg, err.Error())
+					// log.ZError(ctx, "MsgFirstModify InsertExtendMsg failed",  notification.ConversationID, notification.SessionType, extendMsg, err.Error())
 					continue
 				}
 			} else {
 				if err := mmc.extendMsgDatabase.InsertOrUpdateReactionExtendMsgSet(ctx, notification.ConversationID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, mmc.extendSetMsgModel.Pb2Model(notification.SuccessReactionExtensions)); err != nil {
-					log.NewError(operationID, "InsertOrUpdateReactionExtendMsgSet failed")
+					// log.NewError(operationID, "InsertOrUpdateReactionExtendMsgSet failed")
 				}
 			}
 		} else if msg.ContentType == constant.ReactionMessageDeleter {
@@ -108,9 +106,8 @@ func (mmc *ModifyMsgConsumerHandler) ModifyMsg(ctx context.Context, cMsg *sarama
 				continue
 			}
 			if err := mmc.extendMsgDatabase.DeleteReactionExtendMsgSet(ctx, notification.ConversationID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, mmc.extendSetMsgModel.Pb2Model(notification.SuccessReactionExtensions)); err != nil {
-				log.NewError(operationID, "InsertOrUpdateReactionExtendMsgSet failed")
+				// log.NewError(operationID, "InsertOrUpdateReactionExtendMsgSet failed")
 			}
 		}
 	}
-
 }

internal/msgtransfer/online_history_msg_handler.go

@@ -29,9 +29,9 @@ const MongoMessages = 5
 const ChannelNum = 100
 
 type MsgChannelValue struct {
-	conversationID string //maybe userID or super groupID
-	ctx            context.Context
-	ctxMsgList     []*ContextMsg
+	uniqueKey  string
+	ctx        context.Context
+	ctxMsgList []*ContextMsg
 }
 
 type TriggerChannelValue struct {
@@ -90,14 +90,16 @@ func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
 				msgChannelValue := cmd.Value.(MsgChannelValue)
 				ctxMsgList := msgChannelValue.ctxMsgList
 				ctx := msgChannelValue.ctx
-				log.ZDebug(withAggregationCtx(ctx, ctxMsgList), "msg arrived channel", "channel id", channelID, "msgList length", len(ctxMsgList), "conversationID", msgChannelValue.conversationID)
+				log.ZDebug(withAggregationCtx(ctx, ctxMsgList), "msg arrived channel", "channel id", channelID, "msgList length", len(ctxMsgList), "uniqueKey", msgChannelValue.uniqueKey)
 				storageMsgList, notStorageMsgList, storageNotificationList, notStorageNotificationList, modifyMsgList := och.getPushStorageMsgList(ctxMsgList)
 				log.ZDebug(ctx, "msg lens", "storageMsgList", len(storageMsgList), "notStorageMsgList", len(notStorageMsgList),
 					"storageNotificationList", len(storageNotificationList), "notStorageNotificationList", len(notStorageNotificationList), "modifyMsgList", len(modifyMsgList))
-				och.handleMsg(ctx, utils.GetChatConversationIDByMsg(ctxMsgList[0].message), storageMsgList, notStorageMsgList)
-				och.handleNotification(ctx, utils.GetNotificationConversationID(ctxMsgList[0].message), storageNotificationList, notStorageNotificationList)
-				if err := och.msgDatabase.MsgToModifyMQ(ctx, msgChannelValue.conversationID, modifyMsgList); err != nil {
-					log.ZError(ctx, "msg to modify mq error", err, "conversationID", msgChannelValue.conversationID, "modifyMsgList", modifyMsgList)
+				conversationIDMsg := utils.GetChatConversationIDByMsg(ctxMsgList[0].message)
+				conversationIDNotification := utils.GetNotificationConversationID(ctxMsgList[0].message)
+				och.handleMsg(ctx, msgChannelValue.uniqueKey, conversationIDMsg, storageMsgList, notStorageMsgList)
+				och.handleNotification(ctx, msgChannelValue.uniqueKey, conversationIDNotification, storageNotificationList, notStorageNotificationList)
+				if err := och.msgDatabase.MsgToModifyMQ(ctx, msgChannelValue.uniqueKey, conversationIDNotification, modifyMsgList); err != nil {
+					log.ZError(ctx, "msg to modify mq error", err, "uniqueKey", msgChannelValue.uniqueKey, "modifyMsgList", modifyMsgList)
 				}
 			}
 		}
@@ -156,8 +158,8 @@ func (och *OnlineHistoryRedisConsumerHandler) getPushStorageMsgList(totalMsgs []
 	return
 }
 
-func (och *OnlineHistoryRedisConsumerHandler) handleNotification(ctx context.Context, conversationID string, storageList, notStorageList []*sdkws.MsgData) {
-	och.toPushTopic(ctx, conversationID, notStorageList)
+func (och *OnlineHistoryRedisConsumerHandler) handleNotification(ctx context.Context, key, conversationID string, storageList, notStorageList []*sdkws.MsgData) {
+	och.toPushTopic(ctx, key, conversationID, notStorageList)
 	if len(storageList) > 0 {
 		lastSeq, _, err := och.msgDatabase.BatchInsertChat2Cache(ctx, conversationID, storageList)
 		if err != nil {
@@ -165,19 +167,19 @@ func (och *OnlineHistoryRedisConsumerHandler) handleNotification(ctx context.Con
 			return
 		}
 		log.ZDebug(ctx, "success to next topic", "conversationID", conversationID)
-		och.msgDatabase.MsgToMongoMQ(ctx, conversationID, storageList, lastSeq)
-		och.toPushTopic(ctx, conversationID, storageList)
+		och.msgDatabase.MsgToMongoMQ(ctx, key, conversationID, storageList, lastSeq)
+		och.toPushTopic(ctx, key, conversationID, storageList)
 	}
 }
 
-func (och *OnlineHistoryRedisConsumerHandler) toPushTopic(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) {
+func (och *OnlineHistoryRedisConsumerHandler) toPushTopic(ctx context.Context, key, conversationID string, msgs []*sdkws.MsgData) {
 	for _, v := range msgs {
-		och.msgDatabase.MsgToPushMQ(ctx, conversationID, v)
+		och.msgDatabase.MsgToPushMQ(ctx, key, conversationID, v)
 	}
 }
 
-func (och *OnlineHistoryRedisConsumerHandler) handleMsg(ctx context.Context, conversationID string, storageList, notStorageList []*sdkws.MsgData) {
-	och.toPushTopic(ctx, conversationID, notStorageList)
+func (och *OnlineHistoryRedisConsumerHandler) handleMsg(ctx context.Context, key, conversationID string, storageList, notStorageList []*sdkws.MsgData) {
+	och.toPushTopic(ctx, key, conversationID, notStorageList)
 	if len(storageList) > 0 {
 		lastSeq, isNewConversation, err := och.msgDatabase.BatchInsertChat2Cache(ctx, conversationID, storageList)
 		if err != nil && errs.Unwrap(err) != redis.Nil {
@@ -209,8 +211,8 @@ func (och *OnlineHistoryRedisConsumerHandler) handleMsg(ctx context.Context, con
 		och.singleMsgSuccessCountMutex.Lock()
 		och.singleMsgSuccessCount += uint64(len(storageList))
 		och.singleMsgSuccessCountMutex.Unlock()
-		och.msgDatabase.MsgToMongoMQ(ctx, conversationID, storageList, lastSeq)
-		och.toPushTopic(ctx, conversationID, storageList)
+		och.msgDatabase.MsgToMongoMQ(ctx, key, conversationID, storageList, lastSeq)
+		och.toPushTopic(ctx, key, conversationID, storageList)
 	}
 }
 
@@ -228,8 +230,8 @@ func (och *OnlineHistoryRedisConsumerHandler) MessagesDistributionHandle() {
 				log.ZDebug(ctx, "batch messages come to distribution center", "length", len(consumerMessages))
 				for i := 0; i < len(consumerMessages); i++ {
 					ctxMsg := &ContextMsg{}
-					var msgFromMQ sdkws.MsgData
-					err := proto.Unmarshal(consumerMessages[i].Value, &msgFromMQ)
+					msgFromMQ := &sdkws.MsgData{}
+					err := proto.Unmarshal(consumerMessages[i].Value, msgFromMQ)
 					if err != nil {
 						log.ZError(ctx, "msg_transfer Unmarshal msg err", err, string(consumerMessages[i].Value))
 						continue
@@ -240,7 +242,7 @@ func (och *OnlineHistoryRedisConsumerHandler) MessagesDistributionHandle() {
 					}
 					log.ZInfo(ctx, "consumer.kafka.GetContextWithMQHeader", "len", len(consumerMessages[i].Headers), "header", strings.Join(arr, ", "))
 					ctxMsg.ctx = kafka.GetContextWithMQHeader(consumerMessages[i].Headers)
-					ctxMsg.message = &msgFromMQ
+					ctxMsg.message = msgFromMQ
 					log.ZDebug(ctx, "single msg come to distribution center", "message", msgFromMQ, "key", string(consumerMessages[i].Key))
 					//aggregationMsgs[string(consumerMessages[i].Key)] = append(aggregationMsgs[string(consumerMessages[i].Key)], ctxMsg)
 					if oldM, ok := aggregationMsgs[string(consumerMessages[i].Key)]; ok {
@@ -253,12 +255,12 @@ func (och *OnlineHistoryRedisConsumerHandler) MessagesDistributionHandle() {
 					}
 				}
 				log.ZDebug(ctx, "generate map list users len", "length", len(aggregationMsgs))
-				for conversationID, v := range aggregationMsgs {
+				for uniqueKey, v := range aggregationMsgs {
 					if len(v) >= 0 {
-						hashCode := utils.GetHashCode(conversationID)
+						hashCode := utils.GetHashCode(uniqueKey)
 						channelID := hashCode % ChannelNum
-						log.ZDebug(withAggregationCtx(ctx, v), "generate channelID", "hashCode", hashCode, "channelID", channelID, "conversationID", conversationID)
-						och.chArrays[channelID] <- Cmd2Value{Cmd: SourceMessages, Value: MsgChannelValue{conversationID: conversationID, ctxMsgList: v, ctx: ctx}}
+						log.ZDebug(withAggregationCtx(ctx, v), "generate channelID", "hashCode", hashCode, "channelID", channelID, "uniqueKey", uniqueKey)
+						och.chArrays[channelID] <- Cmd2Value{Cmd: SourceMessages, Value: MsgChannelValue{uniqueKey: uniqueKey, ctxMsgList: v, ctx: ctx}}
 					}
 				}
 			}

internal/msgtransfer/online_msg_to_mongo_handler.go

@@ -45,11 +45,11 @@ func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(ctx context.Cont
 	log.ZInfo(ctx, "mongo consumer recv msg", "msgs", msgFromMQ.MsgData)
 	err = mc.msgDatabase.BatchInsertChat2DB(ctx, msgFromMQ.ConversationID, msgFromMQ.MsgData, msgFromMQ.LastSeq)
 	if err != nil {
-		log.ZError(ctx, "single data insert to mongo err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID, "triggerID", msgFromMQ.TriggerID)
+		log.ZError(ctx, "single data insert to mongo err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID)
 	}
 	err = mc.msgDatabase.DeleteMessageFromCache(ctx, msgFromMQ.ConversationID, msgFromMQ.MsgData)
 	if err != nil {
-		log.ZError(ctx, "remove cache msg from redis err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID, "triggerID", msgFromMQ.TriggerID)
+		log.ZError(ctx, "remove cache msg from redis err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID)
 	}
 	for _, v := range msgFromMQ.MsgData {
 		if v.ContentType == constant.DeleteMessageNotification {

internal/rpc/msg/send_msg.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"math/rand"
 	"strconv"
-	"sync"
 	"time"
 
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
@@ -310,34 +309,3 @@ func valueCopy(pb *msg.SendMsgReq) *msg.SendMsgReq {
 	msgData.Options = options
 	return &msg.SendMsgReq{MsgData: &msgData}
 }
-
-func (m *msgServer) sendMsgToGroupOptimization(ctx context.Context, list []string, req *msg.SendMsgReq, wg *sync.WaitGroup) (err error) {
-	defer wg.Done()
-	tempOptions := make(map[string]bool, 1)
-	for k, v := range req.MsgData.Options {
-		tempOptions[k] = v
-	}
-	for _, v := range list {
-		req.MsgData.RecvID = v
-		options := make(map[string]bool, 1)
-		for k, v := range tempOptions {
-			options[k] = v
-		}
-		req.MsgData.Options = options
-		conversationID := utils.GetConversationIDBySessionType(constant.GroupChatType, req.MsgData.GroupID)
-		isSend, err := m.modifyMessageByUserMessageReceiveOpt(ctx, v, conversationID, constant.GroupChatType, req)
-		if err != nil {
-			return err
-		}
-		if isSend {
-			if v == "" || req.MsgData.SendID == "" {
-				return errs.ErrArgs.Wrap("userID or groupPB.MsgData.SendID is empty")
-			}
-			err := m.MsgDatabase.MsgToMQ(ctx, v, req.MsgData)
-			if err != nil {
-				return err
-			}
-		}
-	}
-	return nil
-}

internal/rpc/msg/send_pull.go

@@ -2,34 +2,30 @@ package msg
 
 import (
 	"context"
-	"sync"
 
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
 	promePkg "github.com/OpenIMSDK/Open-IM-Server/pkg/common/prome"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
-	pbConversation "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/conversation"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg"
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/sdkws"
+	pbMsg "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
-	"google.golang.org/protobuf/proto"
 )
 
-func (m *msgServer) sendMsgSuperGroupChat(ctx context.Context, req *msg.SendMsgReq) (resp *msg.SendMsgResp, err error) {
-	resp = &msg.SendMsgResp{}
+func (m *msgServer) sendMsgSuperGroupChat(ctx context.Context, req *pbMsg.SendMsgReq) (resp *pbMsg.SendMsgResp, err error) {
+	resp = &pbMsg.SendMsgResp{}
 	promePkg.Inc(promePkg.WorkSuperGroupChatMsgRecvSuccessCounter)
 	if _, err = m.messageVerification(ctx, req); err != nil {
 		promePkg.Inc(promePkg.WorkSuperGroupChatMsgProcessFailedCounter)
 		return nil, err
 	}
-	err = m.MsgDatabase.MsgToMQ(ctx, utils.GetConversationIDByMsg(req.MsgData), req.MsgData)
+	err = m.MsgDatabase.MsgToMQ(ctx, utils.GenConversationUniqueKeyForGroup(req.MsgData.GroupID), req.MsgData)
 	if err != nil {
 		return nil, err
 	}
-	// callback
-	if err = CallbackAfterSendGroupMsg(ctx, req); err != nil && err != errs.ErrCallbackContinue {
-		return nil, err
+	if err = CallbackAfterSendGroupMsg(ctx, req); err != nil {
+		log.ZError(ctx, "CallbackAfterSendGroupMsg", err)
 	}
-
 	promePkg.Inc(promePkg.WorkSuperGroupChatMsgProcessSuccessCounter)
 	resp.SendTime = req.MsgData.SendTime
 	resp.ServerMsgID = req.MsgData.ServerMsgID
@@ -37,18 +33,13 @@ func (m *msgServer) sendMsgSuperGroupChat(ctx context.Context, req *msg.SendMsgR
 	return resp, nil
 }
 
-func (m *msgServer) sendMsgNotification(ctx context.Context, req *msg.SendMsgReq) (resp *msg.SendMsgResp, err error) {
-	err = m.MsgDatabase.MsgToMQ(ctx, req.MsgData.RecvID, req.MsgData)
-	if err != nil {
+func (m *msgServer) sendMsgNotification(ctx context.Context, req *pbMsg.SendMsgReq) (resp *pbMsg.SendMsgResp, err error) {
+	promePkg.Inc(promePkg.SingleChatMsgRecvSuccessCounter)
+	if err := m.MsgDatabase.MsgToMQ(ctx, utils.GenConversationUniqueKeyForSingle(req.MsgData.SendID, req.MsgData.RecvID), req.MsgData); err != nil {
+		promePkg.Inc(promePkg.SingleChatMsgProcessFailedCounter)
 		return nil, err
 	}
-	if req.MsgData.SendID != req.MsgData.RecvID { //Filter messages sent to yourself
-		err = m.MsgDatabase.MsgToMQ(ctx, req.MsgData.SendID, req.MsgData)
-		if err != nil {
-			return nil, err
-		}
-	}
-	resp = &msg.SendMsgResp{
+	resp = &pbMsg.SendMsgResp{
 		ServerMsgID: req.MsgData.ServerMsgID,
 		ClientMsgID: req.MsgData.ClientMsgID,
 		SendTime:    req.MsgData.SendTime,
@@ -56,164 +47,38 @@ func (m *msgServer) sendMsgNotification(ctx context.Context, req *msg.SendMsgReq
 	return resp, nil
 }
 
-func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *msg.SendMsgReq) (resp *msg.SendMsgResp, err error) {
+func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbMsg.SendMsgReq) (resp *pbMsg.SendMsgResp, err error) {
 	promePkg.Inc(promePkg.SingleChatMsgRecvSuccessCounter)
 	_, err = m.messageVerification(ctx, req)
 	if err != nil {
 		return nil, err
 	}
-
 	var isSend bool = true
-	conversationID := utils.GetConversationIDByMsg(req.MsgData)
-	if !utils.IsNotification(conversationID) {
-		isSend, err = m.modifyMessageByUserMessageReceiveOpt(ctx, req.MsgData.RecvID, conversationID, constant.SingleChatType, req)
+	isNotification := utils.IsNotificationByMsg(req.MsgData)
+	if !isNotification {
+		isSend, err = m.modifyMessageByUserMessageReceiveOpt(ctx, req.MsgData.RecvID, utils.GenConversationIDForSingle(req.MsgData.SendID, req.MsgData.RecvID), constant.SingleChatType, req)
 		if err != nil {
 			return nil, err
 		}
 	}
-	if isSend {
-		err = m.MsgDatabase.MsgToMQ(ctx, conversationID, req.MsgData)
-		if err != nil {
-			return nil, errs.ErrInternalServer.Wrap("insert to mq")
+	if !isSend {
+		promePkg.Inc(promePkg.SingleChatMsgProcessFailedCounter)
+		return nil, errs.ErrUserNotRecvMsg
+	} else {
+		if err := m.MsgDatabase.MsgToMQ(ctx, utils.GenConversationUniqueKeyForSingle(req.MsgData.SendID, req.MsgData.RecvID), req.MsgData); err != nil {
+			promePkg.Inc(promePkg.SingleChatMsgProcessFailedCounter)
+			return nil, err
 		}
+		err = CallbackAfterSendSingleMsg(ctx, req)
+		if err != nil && err != errs.ErrCallbackContinue {
+			return nil, err
+		}
+		resp = &msg.SendMsgResp{
+			ServerMsgID: req.MsgData.ServerMsgID,
+			ClientMsgID: req.MsgData.ClientMsgID,
+			SendTime:    req.MsgData.SendTime,
+		}
+		promePkg.Inc(promePkg.SingleChatMsgProcessSuccessCounter)
+		return resp, nil
 	}
-	err = CallbackAfterSendSingleMsg(ctx, req)
-	if err != nil && err != errs.ErrCallbackContinue {
-		return nil, err
-	}
-	promePkg.Inc(promePkg.SingleChatMsgProcessSuccessCounter)
-	resp = &msg.SendMsgResp{
-		ServerMsgID: req.MsgData.ServerMsgID,
-		ClientMsgID: req.MsgData.ClientMsgID,
-		SendTime:    req.MsgData.SendTime,
-	}
-	return resp, nil
-}
-
-func (m *msgServer) sendMsgGroupChat(ctx context.Context, req *msg.SendMsgReq) (resp *msg.SendMsgResp, err error) {
-	// callback
-	promePkg.Inc(promePkg.GroupChatMsgRecvSuccessCounter)
-
-	var memberUserIDList []string
-	if memberUserIDList, err = m.messageVerification(ctx, req); err != nil {
-		promePkg.Inc(promePkg.GroupChatMsgProcessFailedCounter)
-		return nil, err
-	}
-
-	var addUidList []string
-	switch req.MsgData.ContentType {
-	case constant.MemberKickedNotification:
-		var tips sdkws.TipsComm
-		var memberKickedTips sdkws.MemberKickedTips
-		err := proto.Unmarshal(req.MsgData.Content, &tips)
-		if err != nil {
-			return nil, err
-		}
-		err = proto.Unmarshal(tips.Detail, &memberKickedTips)
-		if err != nil {
-			return nil, err
-		}
-		for _, v := range memberKickedTips.KickedUserList {
-			addUidList = append(addUidList, v.UserID)
-		}
-	case constant.MemberQuitNotification:
-		addUidList = append(addUidList, req.MsgData.SendID)
-
-	default:
-	}
-	if len(addUidList) > 0 {
-		memberUserIDList = append(memberUserIDList, addUidList...)
-	}
-
-	//split  parallel send
-	var wg sync.WaitGroup
-	var split = 20
-	msgToMQSingle := msg.MsgDataToMQ{MsgData: req.MsgData}
-	mErr := make([]error, 0)
-	var mutex sync.RWMutex
-	remain := len(memberUserIDList) % split
-	for i := 0; i < len(memberUserIDList)/split; i++ {
-		wg.Add(1)
-		tmp := valueCopy(req)
-		go func(i int) {
-			err := m.sendMsgToGroupOptimization(ctx, memberUserIDList[i*split:(i+1)*split], tmp, &wg)
-			if err != nil {
-				mutex.Lock()
-				mErr = append(mErr, err)
-				mutex.Unlock()
-			}
-
-		}(i)
-	}
-	if remain > 0 {
-		wg.Add(1)
-		tmp := valueCopy(req)
-		go m.sendMsgToGroupOptimization(ctx, memberUserIDList[split*(len(memberUserIDList)/split):], tmp, &wg)
-	}
-
-	wg.Wait()
-
-	// callback
-	err = CallbackAfterSendGroupMsg(ctx, req)
-	if err != nil && err != errs.ErrCallbackContinue {
-		return nil, err
-	}
-
-	for _, v := range mErr {
-		if v != nil {
-			return nil, v
-		}
-	}
-
-	if req.MsgData.ContentType == constant.AtText {
-		go func() {
-			var conversationReq pbConversation.ModifyConversationFieldReq
-			var tag bool
-			var atUserID []string
-			conversation := pbConversation.Conversation{
-				OwnerUserID:      req.MsgData.SendID,
-				ConversationID:   utils.GetConversationIDBySessionType(constant.GroupChatType, req.MsgData.GroupID),
-				ConversationType: constant.GroupChatType,
-				GroupID:          req.MsgData.GroupID,
-			}
-			conversationReq.Conversation = &conversation
-			conversationReq.FieldType = constant.FieldGroupAtType
-			tagAll := utils.IsContain(constant.AtAllString, req.MsgData.AtUserIDList)
-			if tagAll {
-				atUserID = utils.DifferenceString([]string{constant.AtAllString}, req.MsgData.AtUserIDList)
-				if len(atUserID) == 0 { //just @everyone
-					conversationReq.UserIDList = memberUserIDList
-					conversation.GroupAtType = constant.AtAll
-				} else { //@Everyone and @other people
-					conversationReq.UserIDList = atUserID
-					conversation.GroupAtType = constant.AtAllAtMe
-					tag = true
-				}
-			} else {
-				conversationReq.UserIDList = req.MsgData.AtUserIDList
-				conversation.GroupAtType = constant.AtMe
-			}
-
-			err := m.Conversation.ModifyConversationField(ctx, &conversationReq)
-			if err != nil {
-				return
-			}
-
-			if tag {
-				conversationReq.UserIDList = utils.DifferenceString(atUserID, memberUserIDList)
-				conversation.GroupAtType = constant.AtAll
-				err := m.Conversation.ModifyConversationField(ctx, &conversationReq)
-				if err != nil {
-					return
-				}
-			}
-		}()
-	}
-	//
-
-	promePkg.Inc(promePkg.GroupChatMsgProcessSuccessCounter)
-	resp.SendTime = msgToMQSingle.MsgData.SendTime
-	resp.ServerMsgID = msgToMQSingle.MsgData.ServerMsgID
-	resp.ClientMsgID = msgToMQSingle.MsgData.ClientMsgID
-	return resp, nil
 }

internal/rpc/msg/server.go

@@ -116,8 +116,6 @@ func (m *msgServer) SendMsg(ctx context.Context, req *msg.SendMsgReq) (resp *msg
 	switch req.MsgData.SessionType {
 	case constant.SingleChatType:
 		return m.sendMsgSingleChat(ctx, req)
-	case constant.GroupChatType:
-		return m.sendMsgGroupChat(ctx, req)
 	case constant.NotificationChatType:
 		return m.sendMsgNotification(ctx, req)
 	case constant.SuperGroupChatType: