feat: Remove MySQL and implement it all with Mongo (#1508)

* fix: GetUserReqApplicationList error when there is a disbanded group chat

* fix: error when querying some information about disbanded group

* fix: GetUserReqApplicationList dismissed group error

* fix: the original message referenced by the pull message processing is withdrawn

* fix: the original message referenced by the pull message processing is withdrawn

* fix: the original message referenced by the pull message processing is withdrawn

* fix: the original message referenced by the pull message processing is withdrawn

* fix: the original message referenced by the pull message processing is withdrawn

* fix: the original message referenced by the pull message processing is withdrawn

* fix: the original message referenced by the pull message processing is withdrawn

* fix: the original message referenced by the pull message processing is withdrawn

* fix: the original message referenced by the pull message processing is withdrawn

* merge

* cicd: robot automated Change

* sdkws.MsgData

* user

* interface{} -> any

* user

* third

* group

* group

* group

* group

* group

* group

* conversation

* standalone mysql db model

* tx

* s3

* group

* mongo

* group

* group

* group

* group

* group

* group

* refactor: add openim mysql to mongo refactor

Signed-off-by: Xinwei Xiong(cubxxw) <3293172751nss@gmail.com>

* refactor: add openim mysql to mongo refactor

Signed-off-by: Xinwei Xiong(cubxxw) <3293172751nss@gmail.com>

* remove mysql

* remove mysql

* friend

* friend

* friend

* friend

* friend

* friend

* group

* convert

* index

* index

* all

* all

* mysql2mongo

* data conversion

* up35

* up35

* feat: add format set

Signed-off-by: Xinwei Xiong(cubxxw) <3293172751nss@gmail.com>

* fix: fix scripts

Signed-off-by: Xinwei Xiong(cubxxw) <3293172751nss@gmail.com>

* merge main

* merge main

* Update init-config.sh

* fix: user args check

---------

Signed-off-by: Xinwei Xiong(cubxxw) <3293172751nss@gmail.com>
Co-authored-by: withchao <withchao@users.noreply.github.com>
Co-authored-by: Xinwei Xiong(cubxxw) <3293172751nss@gmail.com>
Co-authored-by: Xinwei Xiong <3293172751@qq.com>

internal/msgtransfer/init.go

@@ -17,21 +17,20 @@ package msgtransfer
 import (
 	"errors"
 	"fmt"
+	"log"
+	"net/http"
+	"sync"
+
 	"github.com/OpenIMSDK/tools/mw"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/collectors"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
-	"log"
-	"net/http"
-	"sync"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/controller"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/db/relation"
-	relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/unrelation"
 	kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
@@ -39,20 +38,12 @@ import (
 )
 
 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
 }
 
 func StartTransfer(prometheusPort int) error {
-	db, err := relation.NewGormDB()
-	if err != nil {
-		return err
-	}
-	if err := db.AutoMigrate(&relationtb.ChatLogModel{}); err != nil {
-		fmt.Printf("gorm: AutoMigrate ChatLogModel err: %v\n", err)
-	}
 	rdb, err := cache.NewRedis()
 	if err != nil {
 		return err
@@ -78,21 +69,16 @@ func StartTransfer(prometheusPort int) error {
 	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()))
 	msgModel := cache.NewMsgCacheModel(rdb)
 	msgDocModel := unrelation.NewMsgMongoDriver(mongo.GetDatabase())
-	msgMysModel := relation.NewChatLogGorm(db)
-	chatLogDatabase := controller.NewChatLogDatabase(msgMysModel)
 	msgDatabase := controller.NewCommonMsgDatabase(msgDocModel, msgModel)
 	conversationRpcClient := rpcclient.NewConversationRpcClient(client)
 	groupRpcClient := rpcclient.NewGroupRpcClient(client)
-	msgTransfer := NewMsgTransfer(chatLogDatabase, msgDatabase, &conversationRpcClient, &groupRpcClient)
+	msgTransfer := NewMsgTransfer(msgDatabase, &conversationRpcClient, &groupRpcClient)
 	return msgTransfer.Start(prometheusPort)
 }
 
-func NewMsgTransfer(chatLogDatabase controller.ChatLogDatabase,
-	msgDatabase controller.CommonMsgDatabase,
-	conversationRpcClient *rpcclient.ConversationRpcClient, groupRpcClient *rpcclient.GroupRpcClient,
-) *MsgTransfer {
+func NewMsgTransfer(msgDatabase controller.CommonMsgDatabase, conversationRpcClient *rpcclient.ConversationRpcClient, groupRpcClient *rpcclient.GroupRpcClient) *MsgTransfer {
 	return &MsgTransfer{
-		persistentCH: NewPersistentConsumerHandler(chatLogDatabase), historyCH: NewOnlineHistoryRedisConsumerHandler(msgDatabase, conversationRpcClient, groupRpcClient),
+		historyCH:      NewOnlineHistoryRedisConsumerHandler(msgDatabase, conversationRpcClient, groupRpcClient),
 		historyMongoCH: NewOnlineHistoryMongoConsumerHandler(msgDatabase),
 	}
 }

internal/msgtransfer/online_history_msg_handler.go

@@ -61,7 +61,7 @@ type TriggerChannelValue struct {
 
 type Cmd2Value struct {
 	Cmd   int
-	Value interface{}
+	Value any
 }
 type ContextMsg struct {
 	message *sdkws.MsgData

internal/msgtransfer/persistent_msg_handler.go

@@ -1,119 +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"
-
-	"github.com/OpenIMSDK/protocol/constant"
-	pbmsg "github.com/OpenIMSDK/protocol/msg"
-	"github.com/OpenIMSDK/tools/log"
-	"github.com/OpenIMSDK/tools/utils"
-
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/db/controller"
-	kfk "github.com/openimsdk/open-im-server/v3/pkg/common/kafka"
-
-	"github.com/IBM/sarama"
-	"google.golang.org/protobuf/proto"
-)
-
-type PersistentConsumerHandler struct {
-	persistentConsumerGroup *kfk.MConsumerGroup
-	chatLogDatabase         controller.ChatLogDatabase
-}
-
-func NewPersistentConsumerHandler(database controller.ChatLogDatabase) *PersistentConsumerHandler {
-	return &PersistentConsumerHandler{
-		persistentConsumerGroup: kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{
-			KafkaVersion:   sarama.V2_0_0_0,
-			OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false,
-		}, []string{config.Config.Kafka.LatestMsgToRedis.Topic},
-			config.Config.Kafka.Addr, config.Config.Kafka.ConsumerGroupID.MsgToMySql),
-		chatLogDatabase: database,
-	}
-}
-
-func (pc *PersistentConsumerHandler) handleChatWs2Mysql(
-	ctx context.Context,
-	cMsg *sarama.ConsumerMessage,
-	msgKey string,
-	_ sarama.ConsumerGroupSession,
-) {
-	msg := cMsg.Value
-	var tag bool
-	msgFromMQ := pbmsg.MsgDataToMQ{}
-	err := proto.Unmarshal(msg, &msgFromMQ)
-	if err != nil {
-		log.ZError(ctx, "msg_transfer Unmarshal msg err", err)
-		return
-	}
-
-	log.ZDebug(ctx, "handleChatWs2Mysql", "msg", msgFromMQ.MsgData)
-	// Control whether to store history messages (mysql)
-	isPersist := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsPersistent)
-	// Only process receiver data
-	if isPersist {
-		switch msgFromMQ.MsgData.SessionType {
-		case constant.SingleChatType, constant.NotificationChatType:
-			if msgKey == msgFromMQ.MsgData.RecvID {
-				tag = true
-			}
-		case constant.GroupChatType:
-			if msgKey == msgFromMQ.MsgData.SendID {
-				tag = true
-			}
-		case constant.SuperGroupChatType:
-			tag = true
-		}
-		if tag {
-			log.ZInfo(ctx, "msg_transfer msg persisting", "msg", string(msg))
-			if err = pc.chatLogDatabase.CreateChatLog(&msgFromMQ); err != nil {
-				log.ZError(ctx, "Message insert failed", err, "msg", msgFromMQ.String())
-				return
-			}
-		}
-	}
-}
-func (PersistentConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
-func (PersistentConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
-
-func (pc *PersistentConsumerHandler) ConsumeClaim(
-	sess sarama.ConsumerGroupSession,
-	claim sarama.ConsumerGroupClaim,
-) error {
-	for msg := range claim.Messages() {
-		ctx := pc.persistentConsumerGroup.GetContextFromMsg(msg)
-		log.ZDebug(
-			ctx,
-			"kafka get info to mysql",
-			"msgTopic",
-			msg.Topic,
-			"msgPartition",
-			msg.Partition,
-			"msg",
-			string(msg.Value),
-			"key",
-			string(msg.Key),
-		)
-		if len(msg.Value) != 0 {
-			pc.handleChatWs2Mysql(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
-}