feat: merge v3dev into main

2026-05-07 02:26:00 +08:00 · 2023-07-12 15:12:23 +08:00
parent e000236eca 2e6ea7b193
commit 524f63ab35
123 changed files with 7377 additions and 10288 deletions
@@ -1,17 +1,3 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
 package msgtransfer

 import (
@@ -19,28 +5,26 @@ import (
 	"sync"
 	"time"

-	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
-
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/controller"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/relation"
 	relationTb "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation"
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/tx"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/unrelation"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/mw"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/prome"
 	openKeeper "github.com/OpenIMSDK/Open-IM-Server/pkg/discoveryregistry/zookeeper"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/rpcclient"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 )

 type MsgTransfer struct {
 	persistentCH   *PersistentConsumerHandler         // 聊天记录持久化到mysql的消费者 订阅的topic: ws2ms_chat
 	historyCH      *OnlineHistoryRedisConsumerHandler // 这个消费者聚合消息, 订阅的topic：ws2ms_chat, 修改通知发往msg_to_modify topic, 消息存入redis后Incr Redis, 再发消息到ms2pschat topic推送， 发消息到msg_to_mongo topic持久化
 	historyMongoCH *OnlineHistoryMongoConsumerHandler // mongoDB批量插入, 成功后删除redis中消息，以及处理删除通知消息删除的 订阅的topic: msg_to_mongo
-	modifyCH       *ModifyMsgConsumerHandler          // 负责消费修改消息通知的consumer, 订阅的topic: msg_to_modify
+	// modifyCH       *ModifyMsgConsumerHandler          // 负责消费修改消息通知的consumer, 订阅的topic: msg_to_modify
 }

 func StartTransfer(prometheusPort int) error {
@@ -62,18 +46,9 @@ func StartTransfer(prometheusPort int) error {
 	if err := mongo.CreateMsgIndex(); err != nil {
 		return err
 	}
-	client, err := openKeeper.NewClient(
-		config.Config.Zookeeper.ZkAddr,
-		config.Config.Zookeeper.Schema,
-		openKeeper.WithFreq(
-			time.Hour,
-		),
-		openKeeper.WithRoundRobin(),
-		openKeeper.WithUserNameAndPassword(config.Config.Zookeeper.Username,
-			config.Config.Zookeeper.Password),
-		openKeeper.WithTimeout(10),
-		openKeeper.WithLogger(log.NewZkLogger()),
-	)
+	client, err := openKeeper.NewClient(config.Config.Zookeeper.ZkAddr, config.Config.Zookeeper.Schema,
+		openKeeper.WithFreq(time.Hour), openKeeper.WithRoundRobin(), openKeeper.WithUserNameAndPassword(config.Config.Zookeeper.Username,
+			config.Config.Zookeeper.Password), openKeeper.WithTimeout(10), openKeeper.WithLogger(log.NewZkLogger()))
 	if err != nil {
 		return err
 	}
@@ -83,35 +58,20 @@ func StartTransfer(prometheusPort int) error {
 	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()))
 	msgModel := cache.NewMsgCacheModel(rdb)
 	msgDocModel := unrelation.NewMsgMongoDriver(mongo.GetDatabase())
-	extendMsgModel := unrelation.NewExtendMsgSetMongoDriver(mongo.GetDatabase())
-	extendMsgCache := cache.NewExtendMsgSetCacheRedis(rdb, extendMsgModel, cache.GetDefaultOpt())
 	chatLogDatabase := controller.NewChatLogDatabase(relation.NewChatLogGorm(db))
-	extendMsgDatabase := controller.NewExtendMsgDatabase(extendMsgModel, extendMsgCache, tx.NewMongo(mongo.GetClient()))
 	msgDatabase := controller.NewCommonMsgDatabase(msgDocModel, msgModel)
 	conversationRpcClient := rpcclient.NewConversationRpcClient(client)
 	groupRpcClient := rpcclient.NewGroupRpcClient(client)
-	msgTransfer := NewMsgTransfer(
-		chatLogDatabase,
-		extendMsgDatabase,
-		msgDatabase,
-		&conversationRpcClient,
-		&groupRpcClient,
-	)
+	msgTransfer := NewMsgTransfer(chatLogDatabase, msgDatabase, &conversationRpcClient, &groupRpcClient)
 	msgTransfer.initPrometheus()
 	return msgTransfer.Start(prometheusPort)
 }

 func NewMsgTransfer(chatLogDatabase controller.ChatLogDatabase,
-	extendMsgDatabase controller.ExtendMsgDatabase, msgDatabase controller.CommonMsgDatabase,
+	msgDatabase controller.CommonMsgDatabase,
 	conversationRpcClient *rpcclient.ConversationRpcClient, groupRpcClient *rpcclient.GroupRpcClient) *MsgTransfer {
-	return &MsgTransfer{
-		persistentCH: NewPersistentConsumerHandler(chatLogDatabase),
-		historyCH:    NewOnlineHistoryRedisConsumerHandler(msgDatabase, conversationRpcClient, groupRpcClient),
-		historyMongoCH: NewOnlineHistoryMongoConsumerHandler(
-			msgDatabase,
-		),
-		modifyCH: NewModifyMsgConsumerHandler(extendMsgDatabase),
-	}
+	return &MsgTransfer{persistentCH: NewPersistentConsumerHandler(chatLogDatabase), historyCH: NewOnlineHistoryRedisConsumerHandler(msgDatabase, conversationRpcClient, groupRpcClient),
+		historyMongoCH: NewOnlineHistoryMongoConsumerHandler(msgDatabase)}
 }

 func (m *MsgTransfer) initPrometheus() {
@@ -136,7 +96,7 @@ func (m *MsgTransfer) Start(prometheusPort int) error {
 	}
 	go m.historyCH.historyConsumerGroup.RegisterHandleAndConsumer(m.historyCH)
 	go m.historyMongoCH.historyConsumerGroup.RegisterHandleAndConsumer(m.historyMongoCH)
-	go m.modifyCH.modifyMsgConsumerGroup.RegisterHandleAndConsumer(m.modifyCH)
+	// go m.modifyCH.modifyMsgConsumerGroup.RegisterHandleAndConsumer(m.modifyCH)
 	err := prome.StartPrometheusSrv(prometheusPort)
 	if err != nil {
 		return err
@@ -1,145 +0,0 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package msgtransfer
-
-import (
-	"context"
-	"encoding/json"
-
-	"github.com/Shopify/sarama"
-
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/controller"
-	unRelationTb "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/unrelation"
-	kfk "github.com/OpenIMSDK/Open-IM-Server/pkg/common/kafka"
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/mcontext"
-	pbMsg "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg"
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/sdkws"
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
-
-	"google.golang.org/protobuf/proto"
-)
-
-type ModifyMsgConsumerHandler struct {
-	modifyMsgConsumerGroup *kfk.MConsumerGroup
-
-	extendMsgDatabase controller.ExtendMsgDatabase
-	extendSetMsgModel unRelationTb.ExtendMsgSetModel
-}
-
-func NewModifyMsgConsumerHandler(database controller.ExtendMsgDatabase) *ModifyMsgConsumerHandler {
-	return &ModifyMsgConsumerHandler{
-		modifyMsgConsumerGroup: kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0,
-			OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.MsgToModify.Topic},
-			config.Config.Kafka.Addr, config.Config.Kafka.ConsumerGroupID.MsgToModify),
-		extendMsgDatabase: database,
-	}
-}
-
-func (ModifyMsgConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
-func (ModifyMsgConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
-func (mmc *ModifyMsgConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
-	claim sarama.ConsumerGroupClaim) error {
-	for msg := range claim.Messages() {
-		ctx := mmc.modifyMsgConsumerGroup.GetContextFromMsg(msg)
-		log.ZDebug(
-			ctx,
-			"kafka get info to mysql",
-			"ModifyMsgConsumerHandler",
-			msg.Topic,
-			"msgPartition",
-			msg.Partition,
-			"msg",
-			string(msg.Value),
-			"key",
-			string(msg.Key),
-		)
-		if len(msg.Value) != 0 {
-			mmc.ModifyMsg(ctx, msg, string(msg.Key), sess)
-		} else {
-			log.ZError(ctx, "msg get from kafka but is nil", nil, "key", msg.Key)
-		}
-		sess.MarkMessage(msg, "")
-	}
-	return nil
-}
-
-func (mmc *ModifyMsgConsumerHandler) ModifyMsg(
-	ctx context.Context,
-	cMsg *sarama.ConsumerMessage,
-	msgKey string,
-	_ sarama.ConsumerGroupSession,
-) {
-	msgFromMQ := pbMsg.MsgDataToModifyByMQ{}
-	operationID := mcontext.GetOperationID(ctx)
-	err := proto.Unmarshal(cMsg.Value, &msgFromMQ)
-	if err != nil {
-		log.ZError(ctx, "msg_transfer Unmarshal msg err", err, "msg", string(cMsg.Value))
-		return
-	}
-	log.ZDebug(ctx, "proto.Unmarshal MsgDataToMQ", "msgs", msgFromMQ.String())
-	for _, msg := range msgFromMQ.Messages {
-		isReactionFromCache := utils.GetSwitchFromOptions(msg.Options, constant.IsReactionFromCache)
-		if !isReactionFromCache {
-			continue
-		}
-		ctx = mcontext.SetOperationID(ctx, operationID)
-		if msg.ContentType == constant.ReactionMessageModifier {
-			notification := &sdkws.ReactionMessageModifierNotification{}
-			if err := json.Unmarshal(msg.Content, notification); err != nil {
-				continue
-			}
-			if notification.IsExternalExtensions {
-				continue
-			}
-			if !notification.IsReact {
-				// first time to modify
-				var reactionExtensionList = make(map[string]unRelationTb.KeyValueModel)
-				extendMsg := unRelationTb.ExtendMsgModel{
-					ReactionExtensionList: reactionExtensionList,
-					ClientMsgID:           notification.ClientMsgID,
-					MsgFirstModifyTime:    notification.MsgFirstModifyTime,
-				}
-				for _, v := range notification.SuccessReactionExtensions {
-					reactionExtensionList[v.TypeKey] = unRelationTb.KeyValueModel{
-						TypeKey:          v.TypeKey,
-						Value:            v.Value,
-						LatestUpdateTime: v.LatestUpdateTime,
-					}
-				}
-
-				if err := mmc.extendMsgDatabase.InsertExtendMsg(ctx, notification.ConversationID, notification.SessionType, &extendMsg); err != nil {
-					// 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")
-				}
-			}
-		} else if msg.ContentType == constant.ReactionMessageDeleter {
-			notification := &sdkws.ReactionMessageDeleteNotification{}
-			if err := json.Unmarshal(msg.Content, notification); err != nil {
-				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")
-			}
-		}
-	}
-}