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>

docker-compose.yml

@@ -24,6 +24,7 @@ services:
     restart: always
     networks:
       server:
+        # open-im-server_server
         ipv4_address: ${MYSQL_NETWORK_ADDRESS}
 
   mongodb:
@@ -142,68 +143,68 @@ services:
       server:
         ipv4_address: ${OPENIM_WEB_NETWORK_ADDRESS}
 
-  openim-admin:
-    image: ${IMAGE_REGISTRY}/openim-admin-front:v3.4.0
-    # image: ghcr.io/openimsdk/openim-admin-front:v3.4.0
-    # image: registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-admin-front:v3.4.0
-    # image: openim/openim-admin-front:v3.4.0
-    container_name: openim-admin
-    restart: always
-    ports:
-      - "${OPENIM_ADMIN_FRONT_PORT}:80"
-    networks:
-      server:
-        ipv4_address: ${OPENIM_ADMIN_FRONT_NETWORK_ADDRESS}
+  # openim-admin:
+  #   image: ${IMAGE_REGISTRY}/openim-admin-front:v3.4.0
+  #   # image: ghcr.io/openimsdk/openim-admin-front:v3.4.0
+  #   # image: registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-admin-front:v3.4.0
+  #   # image: openim/openim-admin-front:v3.4.0
+  #   container_name: openim-admin
+  #   restart: always
+  #   ports:
+  #     - "${OPENIM_ADMIN_FRONT_PORT}:80"
+  #   networks:
+  #     server:
+  #       ipv4_address: ${OPENIM_ADMIN_FRONT_NETWORK_ADDRESS}
 
-  prometheus:
-    image: prom/prometheus
-    container_name: prometheus
-    hostname: prometheus
-    restart: always
-    volumes:
-      - ./config/prometheus.yml:/etc/prometheus/prometheus.yml
-      - ./config/instance-down-rules.yml:/etc/prometheus/instance-down-rules.yml
-    ports:
-      - "${PROMETHEUS_PORT}:9090"
-    networks:
-      server:
-        ipv4_address: ${PROMETHEUS_NETWORK_ADDRESS}
+  # prometheus:
+  #   image: prom/prometheus
+  #   container_name: prometheus
+  #   hostname: prometheus
+  #   restart: always
+  #   volumes:
+  #     - ./config/prometheus.yml:/etc/prometheus/prometheus.yml
+  #     - ./config/instance-down-rules.yml:/etc/prometheus/instance-down-rules.yml
+  #   ports:
+  #     - "${PROMETHEUS_PORT}:9090"
+  #   networks:
+  #     server:
+  #       ipv4_address: ${PROMETHEUS_NETWORK_ADDRESS}
 
-  alertmanager:
-    image: prom/alertmanager
-    container_name: alertmanager
-    hostname: alertmanager
-    restart: always
-    volumes:
-      - ./config/alertmanager.yml:/etc/alertmanager/alertmanager.yml
-      - ./config/email.tmpl:/etc/alertmanager/email.tmpl
-    ports:
-      - "${ALERT_MANAGER_PORT}:9093"
-    networks:
-      server:
-        ipv4_address: ${ALERT_MANAGER_NETWORK_ADDRESS}
+  # alertmanager:
+  #   image: prom/alertmanager
+  #   container_name: alertmanager
+  #   hostname: alertmanager
+  #   restart: always
+  #   volumes:
+  #     - ./config/alertmanager.yml:/etc/alertmanager/alertmanager.yml
+  #     - ./config/email.tmpl:/etc/alertmanager/email.tmpl
+  #   ports:
+  #     - "${ALERT_MANAGER_PORT}:9093"
+  #   networks:
+  #     server:
+  #       ipv4_address: ${ALERT_MANAGER_NETWORK_ADDRESS}
 
-  grafana:
-    image: grafana/grafana
-    container_name: grafana
-    hostname: grafana
-    user: root
-    restart: always
-    ports:
-      - "${GRAFANA_PORT}:3000"
-    volumes:
-      - ${DATA_DIR}/components/grafana:/var/lib/grafana
-    networks:
-      server:
-        ipv4_address: ${GRAFANA_NETWORK_ADDRESS}
+  # grafana:
+  #   image: grafana/grafana
+  #   container_name: grafana
+  #   hostname: grafana
+  #   user: root
+  #   restart: always
+  #   ports:
+  #     - "${GRAFANA_PORT}:3000"
+  #   volumes:
+  #     - ${DATA_DIR}/components/grafana:/var/lib/grafana
+  #   networks:
+  #     server:
+  #       ipv4_address: ${GRAFANA_NETWORK_ADDRESS}
 
-  node-exporter:
-    image: quay.io/prometheus/node-exporter
-    container_name: node-exporter
-    hostname: node-exporter
-    restart: always
-    ports:
-      - "${NODE_EXPORTER_PORT}:9100"
-    networks:
-      server:
-        ipv4_address: ${NODE_EXPORTER_NETWORK_ADDRESS}
+  # node-exporter:
+  #   image: quay.io/prometheus/node-exporter
+  #   container_name: node-exporter
+  #   hostname: node-exporter
+  #   restart: always
+  #   ports:
+  #     - "${NODE_EXPORTER_PORT}:9100"
+  #   networks:
+  #     server:
+  #       ipv4_address: ${NODE_EXPORTER_NETWORK_ADDRESS}

go.mod

@@ -10,10 +10,8 @@ require (
 	github.com/go-playground/validator/v10 v10.15.5
 	github.com/gogo/protobuf v1.3.2
 	github.com/golang-jwt/jwt/v4 v4.5.0
-	github.com/golang/protobuf v1.5.3
 	github.com/gorilla/websocket v1.5.0
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
-	github.com/jinzhu/copier v0.4.0
 	github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible // indirect
 	github.com/minio/minio-go/v7 v7.0.63
 	github.com/mitchellh/mapstructure v1.5.0
@@ -29,19 +27,16 @@ require (
 	google.golang.org/grpc v1.59.0
 	google.golang.org/protobuf v1.31.0
 	gopkg.in/yaml.v3 v3.0.1
-	gorm.io/driver/mysql v1.5.2
-	gorm.io/gorm v1.25.5
 )
 
 require github.com/google/uuid v1.3.1
 
 require (
 	github.com/IBM/sarama v1.41.3
 	github.com/OpenIMSDK/protocol v0.0.31
-	github.com/OpenIMSDK/tools v0.0.17
+	github.com/OpenIMSDK/tools v0.0.18
 	github.com/aliyun/aliyun-oss-go-sdk v2.2.9+incompatible
 	github.com/go-redis/redis v6.15.9+incompatible
-	github.com/go-sql-driver/mysql v1.7.1
 	github.com/redis/go-redis/v9 v9.2.1
 	github.com/tencentyun/cos-go-sdk-v5 v0.7.45
 	go.uber.org/automaxprocs v1.5.3
@@ -93,6 +88,7 @@ require (
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/go-zookeeper/zk v1.0.3 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
+	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/go-querystring v1.1.0 // indirect
@@ -109,6 +105,7 @@ require (
 	github.com/jcmturner/gofork v1.7.6 // indirect
 	github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect
 	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
+	github.com/jinzhu/copier v0.3.5 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect
 	github.com/jinzhu/now v1.1.5 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
@@ -152,7 +149,7 @@ require (
 	golang.org/x/arch v0.3.0 // indirect
 	golang.org/x/net v0.17.0 // indirect
 	golang.org/x/oauth2 v0.13.0 // indirect
-	golang.org/x/sys v0.13.0 // indirect
+	golang.org/x/sys v0.14.0 // indirect
 	golang.org/x/text v0.13.0 // indirect
 	golang.org/x/time v0.3.0 // indirect
 	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
@@ -162,6 +159,7 @@ require (
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20231012201019-e917dd12ba7a // indirect
 	gopkg.in/src-d/go-billy.v4 v4.3.2 // indirect
 	gopkg.in/warnings.v0 v0.1.2 // indirect
+	gorm.io/gorm v1.23.8 // indirect
 )
 
 require (

go.sum

@@ -498,8 +498,8 @@ golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE=
-golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.14.0 h1:Vz7Qs629MkJkGyHxUlRHizWJRG2j8fbQKjELVSNhy7Q=
+golang.org/x/sys v0.14.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=

go.work

@@ -13,4 +13,5 @@ use (
 	./tools/component
 	./tools/url2im
 	./tools/data-conversion
+	./tools/up35
 )

internal/api/msg.go

@@ -150,7 +150,7 @@ func (m *MessageApi) DeleteMsgPhysical(c *gin.Context) {
 }
 
 func (m *MessageApi) getSendMsgReq(c *gin.Context, req apistruct.SendMsg) (sendMsgReq *msg.SendMsgReq, err error) {
-	var data interface{}
+	var data any
 	log.ZDebug(c, "getSendMsgReq", "req", req.Content)
 	switch req.ContentType {
 	case constant.Text:

internal/msggateway/encoder.go

@@ -22,8 +22,8 @@ import (
 )
 
 type Encoder interface {
-	Encode(data interface{}) ([]byte, error)
-	Decode(encodeData []byte, decodeData interface{}) error
+	Encode(data any) ([]byte, error)
+	Decode(encodeData []byte, decodeData any) error
 }
 
 type GobEncoder struct{}
@@ -32,7 +32,7 @@ func NewGobEncoder() *GobEncoder {
 	return &GobEncoder{}
 }
 
-func (g *GobEncoder) Encode(data interface{}) ([]byte, error) {
+func (g *GobEncoder) Encode(data any) ([]byte, error) {
 	buff := bytes.Buffer{}
 	enc := gob.NewEncoder(&buff)
 	err := enc.Encode(data)
@@ -42,7 +42,7 @@ func (g *GobEncoder) Encode(data interface{}) ([]byte, error) {
 	return buff.Bytes(), nil
 }
 
-func (g *GobEncoder) Decode(encodeData []byte, decodeData interface{}) error {
+func (g *GobEncoder) Decode(encodeData []byte, decodeData any) error {
 	buff := bytes.NewBuffer(encodeData)
 	dec := gob.NewDecoder(buff)
 	err := dec.Decode(decodeData)

internal/msggateway/n_ws_server.go

@@ -49,7 +49,7 @@ type LongConnServer interface {
 	wsHandler(w http.ResponseWriter, r *http.Request)
 	GetUserAllCons(userID string) ([]*Client, bool)
 	GetUserPlatformCons(userID string, platform int) ([]*Client, bool, bool)
-	Validate(s interface{}) error
+	Validate(s any) error
 	SetCacheHandler(cache cache.MsgModel)
 	SetDiscoveryRegistry(client discoveryregistry.SvcDiscoveryRegistry)
 	KickUserConn(client *Client) error
@@ -60,6 +60,12 @@ type LongConnServer interface {
 	MessageHandler
 }
 
+var bufferPool = sync.Pool{
+	New: func() any {
+		return make([]byte, 1024)
+	},
+}
+
 type WsServer struct {
 	port              int
 	wsMaxConnNum      int64
@@ -120,7 +126,7 @@ func (ws *WsServer) UnRegister(c *Client) {
 	ws.unregisterChan <- c
 }
 
-func (ws *WsServer) Validate(s interface{}) error {
+func (ws *WsServer) Validate(s any) error {
 	//?question?
 	return nil
 }
@@ -145,7 +151,7 @@ func NewWsServer(opts ...Option) (*WsServer, error) {
 		writeBufferSize:  config.writeBufferSize,
 		handshakeTimeout: config.handshakeTimeout,
 		clientPool: sync.Pool{
-			New: func() interface{} {
+			New: func() any {
 				return new(Client)
 			},
 		},

internal/msgtransfer/init.go

@@ -17,42 +17,33 @@ 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"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 )
 
 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