feat: Remove MySQL and implement it all with Mongo

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

@@ -16,8 +16,8 @@
 
 import (
  "github.com/OpenIMSDK/protocol/constant"
  "github.com/OpenIMSDK/protocol/msg"
  "github.com/OpenIMSDK/protocol/sdkws"
  "github.com/OpenIMSDK/tools/a2r"
  "github.com/OpenIMSDK/tools/apiresp"
  "github.com/OpenIMSDK/tools/errs"
@@ -150,7 +150,7 @@
 }
 
 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 @@
  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 @@
  MessageHandler
 }
 
+var bufferPool = sync.Pool{
+ New: func() any {
+ return make([]byte, 1024)
+ },
+}
+
 type WsServer struct {
  port int
  wsMaxConnNum int64
@@ -120,7 +126,7 @@
  ws.unregisterChan <- c
 }
 
-func (ws *WsServer) Validate(s interface{}) error {
+func (ws *WsServer) Validate(s any) error {
  //?question?
  return nil
 }
@@ -145,7 +151,7 @@
  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