msg

internal/msgtransfer/online_history_msg_handler.go

@@ -93,8 +93,8 @@ func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
 				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, storageMsgList, notStorageMsgList)
-				och.handleNotification(ctx, storageNotificationList, notStorageNotificationList)
+				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)
 				}
@@ -147,8 +147,7 @@ func (och *OnlineHistoryRedisConsumerHandler) getPushStorageMsgList(totalMsgs []
 	return
 }
 
-func (och *OnlineHistoryRedisConsumerHandler) handleNotification(ctx context.Context, storageList, notStorageList []*sdkws.MsgData) {
-	conversationID := utils.GetConversationIDByMsg(storageList[0])
+func (och *OnlineHistoryRedisConsumerHandler) handleNotification(ctx context.Context, conversationID string, storageList, notStorageList []*sdkws.MsgData) {
 	och.toPushTopic(ctx, conversationID, notStorageList)
 	if len(storageList) > 0 {
 		lastSeq, _, err := och.msgDatabase.BatchInsertChat2Cache(ctx, conversationID, storageList)
@@ -168,8 +167,7 @@ func (och *OnlineHistoryRedisConsumerHandler) toPushTopic(ctx context.Context, c
 	}
 }
 
-func (och *OnlineHistoryRedisConsumerHandler) handleMsg(ctx context.Context, storageList, notStorageList []*sdkws.MsgData) {
-	conversationID := utils.GetConversationIDByMsg(storageList[0])
+func (och *OnlineHistoryRedisConsumerHandler) handleMsg(ctx context.Context, conversationID string, storageList, notStorageList []*sdkws.MsgData) {
 	och.toPushTopic(ctx, conversationID, notStorageList)
 	if len(storageList) > 0 {
 		lastSeq, isNewConversation, err := och.msgDatabase.BatchInsertChat2Cache(ctx, conversationID, storageList)

pkg/utils/utils.go

@@ -175,6 +175,38 @@ func MsgIsNotification(msg *sdkws.MsgData) bool {
 	return !options.IsNotNotification()
 }
 
+func GetNotificationConversationID(msg *sdkws.MsgData) string {
+	switch msg.SessionType {
+	case constant.SingleChatType:
+		l := []string{msg.SendID, msg.RecvID}
+		sort.Strings(l)
+		return "n_" + strings.Join(l, "_")
+	case constant.GroupChatType:
+		return "n_" + msg.GroupID
+	case constant.SuperGroupChatType:
+		return "n_" + msg.GroupID
+	case constant.NotificationChatType:
+		return "n_" + msg.SendID + "_" + msg.RecvID
+	}
+	return ""
+}
+
+func GetChatConversationIDByMsg(msg *sdkws.MsgData) string {
+	switch msg.SessionType {
+	case constant.SingleChatType:
+		l := []string{msg.SendID, msg.RecvID}
+		sort.Strings(l)
+		return "si_" + strings.Join(l, "_")
+	case constant.GroupChatType:
+		return "g_" + msg.GroupID
+	case constant.SuperGroupChatType:
+		return "sg_" + msg.GroupID
+	case constant.NotificationChatType:
+		return "sn_" + msg.SendID + "_" + msg.RecvID
+	}
+	return ""
+}
+
 func GetConversationIDByMsg(msg *sdkws.MsgData) string {
 	options := Options(msg.Options)
 	switch msg.SessionType {