Merge branch 'main' into recovery

2025-10-26 21:22:16 +08:00 · 2024-11-22 14:54:40 +08:00 · 2024-11-22 14:54:40 +08:00 · f5c5396bbe
commit f5c5396bbe
parent cb3610ca68 bf88535800
30 changed files with 557 additions and 578 deletions
--- a/.env
+++ b/.env
@ -1,6 +1,5 @@
-MONGO_IMAGE=mongo:6.0.2
+MONGO_IMAGE=mongo:7.0
 REDIS_IMAGE=redis:7.0.0
-ZOOKEEPER_IMAGE=bitnami/zookeeper:3.8
 KAFKA_IMAGE=bitnami/kafka:3.5.1
 MINIO_IMAGE=minio/minio:RELEASE.2024-01-11T07-46-16Z
 ETCD_IMAGE=quay.io/coreos/etcd:v3.5.13
@ -16,4 +15,3 @@ OPENIM_ADMIN_FRONT_IMAGE=openim/openim-admin-front:release-v1.8.2
 #OPENIM_ADMIN_FRONT_IMAGE=registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-admin-front:release-v1.8.2

 DATA_DIR=./
-
--- a/.github/workflows/changelog.yml
+++ b/.github/workflows/changelog.yml
@ -0,0 +1,78 @@
+name: Release Changelog
+
+on:
+  release:
+    types: [released]
+
+permissions:
+  contents: write
+  pull-requests: write
+
+jobs:
+  update-changelog:
+    runs-on: ubuntu-latest
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v4
+
+    - name: Run Go Changelog Generator
+      run: |
+        # Run the Go changelog generator, passing the release tag if available
+        if [ "${{ github.event.release.tag_name }}" = "latest" ]; then
+          go run tools/changelog/changelog.go > "${{ github.event.release.tag_name }}-changelog.md"
+        else
+          go run tools/changelog/changelog.go "${{ github.event.release.tag_name }}" > "${{ github.event.release.tag_name }}-changelog.md"
+        fi
+
+    - name: Handle changelog files
+      run: |
+        # Ensure that the CHANGELOG directory exists
+        mkdir -p CHANGELOG
+
+        # Extract Major.Minor version by removing the 'v' prefix from the tag name
+        TAG_NAME=${{ github.event.release.tag_name }}
+        CHANGELOG_VERSION_NUMBER=$(echo "$TAG_NAME" | sed 's/^v//' | grep -oP '^\d+\.\d+')
+
+        # Define the new changelog file path
+        CHANGELOG_FILENAME="CHANGELOG-$CHANGELOG_VERSION_NUMBER.md"
+        CHANGELOG_PATH="CHANGELOG/$CHANGELOG_FILENAME"
+
+        # Check if the changelog file for the current release already exists
+        if [ -f "$CHANGELOG_PATH" ]; then
+          # If the file exists, append the new changelog to the existing one
+          cat "$CHANGELOG_PATH" >> "${TAG_NAME}-changelog.md"
+          # Overwrite the existing changelog with the updated content
+          mv "${TAG_NAME}-changelog.md" "$CHANGELOG_PATH"
+        else
+          # If the changelog file doesn't exist, rename the temp changelog file to the new changelog file
+          mv "${TAG_NAME}-changelog.md" "$CHANGELOG_PATH"
+
+          # Ensure that README.md exists
+          if [ ! -f "CHANGELOG/README.md" ]; then
+            echo -e "# CHANGELOGs\n\n" > CHANGELOG/README.md
+          fi
+          
+            # Add the new changelog entry at the top of the README.md
+            if ! grep -q "\[$CHANGELOG_FILENAME\]" CHANGELOG/README.md; then
+            sed -i "3i- [$CHANGELOG_FILENAME](./$CHANGELOG_FILENAME)" CHANGELOG/README.md
+            # Remove the extra newline character added by sed
+            # sed -i '4d' CHANGELOG/README.md
+            fi
+          fi
+
+    - name: Clean up
+      run: |
+        # Remove any temporary files that were created during the process
+        rm -f "${{ github.event.release.tag_name }}-changelog.md"
+
+    - name: Create Pull Request
+      uses: peter-evans/create-pull-request@v7.0.5
+      with:
+        token: ${{ secrets.GITHUB_TOKEN }}
+        commit-message: "Update CHANGELOG for release ${{ github.event.release.tag_name }}"
+        title: "Update CHANGELOG for release ${{ github.event.release.tag_name }}"
+        body: "This PR updates the CHANGELOG files for release ${{ github.event.release.tag_name }}"
+        branch: changelog-${{ github.event.release.tag_name }} 
+        base: main 
+        delete-branch: true
+        labels: changelog
--- a/config/mongodb.yml
+++ b/config/mongodb.yml
@ -8,6 +8,8 @@ database: openim_v3
 username: openIM
 # Password for database authentication
 password: openIM123
+# Authentication source for database authentication, if use root user, set it to admin
+authSource: openim_v3
 # Maximum number of connections in the connection pool
 maxPoolSize: 100
 # Maximum number of retry attempts for a failed database connection
--- a/config/openim-msggateway.yml
+++ b/config/openim-msggateway.yml
@ -22,5 +22,3 @@ longConnSvr:
  websocketMaxMsgLen: 4096
  # WebSocket connection handshake timeout in seconds
  websocketTimeout: 10
-
-
--- a/config/openim-push.yml
+++ b/config/openim-push.yml
@ -13,7 +13,7 @@ prometheus:
  ports: [ 12170, 12171, 12172, 12173, 12174, 12175, 12176, 12177, 12178, 12179, 12180, 12182, 12183, 12184, 12185, 12186 ]

 maxConcurrentWorkers: 3
-#Use geTui for offline push notifications, or choose fcm or jpns; corresponding configuration settings must be specified.
+#Use geTui for offline push notifications, or choose fcm or jpush; corresponding configuration settings must be specified.
 enable: geTui
 geTui:
  pushUrl: https://restapi.getui.com/v2/$appId
@ -26,7 +26,7 @@ fcm:
  # Prioritize using file paths. If the file path is empty, use URL
  filePath:   # File path is concatenated with the parameters passed in through - c(`mage` default pass in `config/`) and filePath.
  authURL:   #  Must start with https or http.
-jpns:
+jpush:
  appKey:
  masterSecret:
  pushURL:
@ -34,8 +34,8 @@ jpns:

 # iOS system push sound and badge count
 iosPush:
-      pushSound: xxx
-      badgeCount: true
-      production: false
+  pushSound: xxx
+  badgeCount: true
+  production: false

 fullUserCache: true
--- a/deployments/templates/config.yaml
+++ b/deployments/templates/config.yaml
@ -240,11 +240,11 @@ push:
    channelName: ${GETUI_CHANNEL_NAME}
  fcm:
    serviceAccount: "${FCM_SERVICE_ACCOUNT}"
-  jpns:
-    appKey: ${JPNS_APP_KEY}
-    masterSecret: ${JPNS_MASTER_SECRET}
-    pushUrl: ${JPNS_PUSH_URL}
-    pushIntent: ${JPNS_PUSH_INTENT}
+  jpush:
+    appKey: ${JPUSH_APP_KEY}
+    masterSecret: ${JPUSH_MASTER_SECRET}
+    pushUrl: ${JPUSH_PUSH_URL}
+    pushIntent: ${JPUSH_PUSH_INTENT}

 # App manager configuration
 #
--- a/docker-compose.yml
+++ b/docker-compose.yml
@ -8,12 +8,35 @@ services:
    ports:
      - "37017:27017"
    container_name: mongo
-    command: ["/bin/bash", "-c", "/docker-entrypoint-initdb.d/mongo-init.sh; docker-entrypoint.sh mongod --wiredTigerCacheSizeGB 1 --auth"]
+    command: >
+      bash -c '
+      docker-entrypoint.sh mongod --wiredTigerCacheSizeGB $$wiredTigerCacheSizeGB --auth &
+      until mongosh -u $$MONGO_INITDB_ROOT_USERNAME -p $$MONGO_INITDB_ROOT_PASSWORD --authenticationDatabase admin --eval "db.runCommand({ ping: 1 })" &>/dev/null; do
+        echo "Waiting for MongoDB to start..."
+        sleep 1
+      done &&
+      mongosh -u $$MONGO_INITDB_ROOT_USERNAME -p $$MONGO_INITDB_ROOT_PASSWORD --authenticationDatabase admin --eval "
+      db = db.getSiblingDB(\"$$MONGO_INITDB_DATABASE\");
+      if (!db.getUser(\"$$MONGO_OPENIM_USERNAME\")) {
+        db.createUser({
+          user: \"$$MONGO_OPENIM_USERNAME\",
+          pwd: \"$$MONGO_OPENIM_PASSWORD\",
+          roles: [{role: \"readWrite\", db: \"$$MONGO_INITDB_DATABASE\"}]
+        });
+        print(\"User created successfully: \");
+        print(\"Username: $$MONGO_OPENIM_USERNAME\");
+        print(\"Password: $$MONGO_OPENIM_PASSWORD\");
+        print(\"Database: $$MONGO_INITDB_DATABASE\");
+      } else {
+        print(\"User already exists in database: $$MONGO_INITDB_DATABASE, Username: $$MONGO_OPENIM_USERNAME\");
+      }
+      " &&
+      tail -f /dev/null
+      '
    volumes:
      - "${DATA_DIR}/components/mongodb/data/db:/data/db"
      - "${DATA_DIR}/components/mongodb/data/logs:/data/logs"
      - "${DATA_DIR}/components/mongodb/data/conf:/etc/mongo"
-      - "./scripts/mongo-init.sh:/docker-entrypoint-initdb.d/mongo-init.sh:ro"
    environment:
      - TZ=Asia/Shanghai
      - wiredTigerCacheSizeGB=1
@ -71,10 +94,7 @@ services:
    ports:
      - "19094:9094"
    volumes:
-      - ./scripts/create-topic.sh:/opt/bitnami/kafka/create-topic.sh
      - "${DATA_DIR}/components/kafka:/bitnami/kafka"
-    command: >
-      bash -c "/opt/bitnami/scripts/kafka/run.sh & /opt/bitnami/kafka/create-topic.sh; wait"
    environment:
      #KAFKA_HEAP_OPTS: "-Xms128m -Xmx256m"
      TZ: Asia/Shanghai
@ -85,10 +105,11 @@ services:
      KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,EXTERNAL://localhost:19094
      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      KAFKA_NUM_PARTITIONS: 8
+      KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: "true"
    networks:
      - openim

-
  minio:
    image: "${MINIO_IMAGE}"
    ports:
@ -171,4 +192,3 @@ services:
 #      - ${DATA_DIR:-./}/components/grafana:/var/lib/grafana
 #    networks:
 #      - openim
-
--- a/docs/contrib/environment.md
+++ b/docs/contrib/environment.md
@ -474,10 +474,10 @@ This section involves setting up additional configuration variables for Websocke
 | GETUI_CHANNEL_ID        | [User Defined]    | GeTui Channel ID                 |
 | GETUI_CHANNEL_NAME      | [User Defined]    | GeTui Channel Name               |
 | FCM_SERVICE_ACCOUNT     | "x.json"          | FCM Service Account              |
-| JPNS_APP_KEY            | [User Defined]    | JPNS Application Key             |
-| JPNS_MASTER_SECRET      | [User Defined]    | JPNS Master Secret               |
-| JPNS_PUSH_URL           | [User Defined]    | JPNS Push Notification URL       |
-| JPNS_PUSH_INTENT        | [User Defined]    | JPNS Push Intent                 |
+| JPUSH_APP_KEY            | [User Defined]    | JPUSH Application Key             |
+| JPUSH_MASTER_SECRET      | [User Defined]    | JPUSH Master Secret               |
+| JPUSH_PUSH_URL           | [User Defined]    | JPUSH Push Notification URL       |
+| JPUSH_PUSH_INTENT        | [User Defined]    | JPUSH Push Intent                 |
 | IM_ADMIN_USERID         | "imAdmin"         | IM Administrator ID              |
 | IM_ADMIN_NAME           | "imAdmin"         | IM Administrator Nickname        |
 | MULTILOGIN_POLICY       | "1"               | Multi-login Policy               |
--- a/go.mod
+++ b/go.mod
@ -2,7 +2,7 @@ module github.com/openimsdk/open-im-server/v3

 go 1.22.0

-toolchain go1.22.3
+toolchain go1.23.2

 require (
 	firebase.google.com/go/v4 v4.14.1
@ -15,7 +15,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/openimsdk/protocol v0.0.72-alpha.54
-	github.com/openimsdk/tools v0.0.50-alpha.26
+	github.com/openimsdk/tools v0.0.50-alpha.32
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0
 	github.com/stretchr/testify v1.9.0
--- a/go.sum
+++ b/go.sum
@ -321,8 +321,8 @@ github.com/openimsdk/gomake v0.0.14-alpha.5 h1:VY9c5x515lTfmdhhPjMvR3BBRrRquAUCF
 github.com/openimsdk/gomake v0.0.14-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
 github.com/openimsdk/protocol v0.0.72-alpha.54 h1:opato7N4QjjRq/SHD54bDSVBpOEEDp1VLWVk5Os2A9s=
 github.com/openimsdk/protocol v0.0.72-alpha.54/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
-github.com/openimsdk/tools v0.0.50-alpha.26 h1:pFNnL70oVR047TkMF0AhY0/R5ugMIASBBNvV5QsUAOA=
-github.com/openimsdk/tools v0.0.50-alpha.26/go.mod h1:r5U6RbxcR4xhKb2fhTmKGC9Yt5LcErHBVt3lhXQIHSo=
+github.com/openimsdk/tools v0.0.50-alpha.32 h1:JEsUFHFnaYg230TG+Ke3SUnaA2h44t4kABAzEdv5VZw=
+github.com/openimsdk/tools v0.0.50-alpha.32/go.mod h1:r5U6RbxcR4xhKb2fhTmKGC9Yt5LcErHBVt3lhXQIHSo=
 github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=
 github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs=
 github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ=
--- a/internal/msggateway/client.go
+++ b/internal/msggateway/client.go
@ -16,6 +16,7 @@ package msggateway

 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"github.com/openimsdk/tools/mw"
 	"runtime/debug"
@ -70,6 +71,8 @@ type Client struct {
 	IsCompress     bool   `json:"isCompress"`
 	UserID         string `json:"userID"`
 	IsBackground   bool   `json:"isBackground"`
+	SDKType        string `json:"sdkType"`
+	Encoder        Encoder
 	ctx            *UserConnContext
 	longConnServer LongConnServer
 	closed         atomic.Bool
@ -95,11 +98,17 @@ func (c *Client) ResetClient(ctx *UserConnContext, conn LongConn, longConnServer
 	c.closed.Store(false)
 	c.closedErr = nil
 	c.token = ctx.GetToken()
+	c.SDKType = ctx.GetSDKType()
 	c.hbCtx, c.hbCancel = context.WithCancel(c.ctx)
 	c.subLock = new(sync.Mutex)
 	if c.subUserIDs != nil {
 		clear(c.subUserIDs)
 	}
+	if c.SDKType == GoSDK {
+		c.Encoder = NewGobEncoder()
+	} else {
+		c.Encoder = NewJsonEncoder()
+	}
 	c.subUserIDs = make(map[string]struct{})
 }

@ -160,9 +169,12 @@ func (c *Client) readMessage() {
 				return
 			}
 		case MessageText:
-			c.closedErr = ErrNotSupportMessageProtocol
-			return
-
+			_ = c.conn.SetReadDeadline(pongWait)
+			parseDataErr := c.handlerTextMessage(message)
+			if parseDataErr != nil {
+				c.closedErr = parseDataErr
+				return
+			}
 		case PingMessage:
 			err := c.writePongMsg("")
 			log.ZError(c.ctx, "writePongMsg", err)
@ -189,7 +201,7 @@ func (c *Client) handleMessage(message []byte) error {
 	var binaryReq = getReq()
 	defer freeReq(binaryReq)

-	err := c.longConnServer.Decode(message, binaryReq)
+	err := c.Encoder.Decode(message, binaryReq)
 	if err != nil {
 		return err
 	}
@ -336,7 +348,7 @@ func (c *Client) writeBinaryMsg(resp Resp) error {
 		return nil
 	}

-	encodedBuf, err := c.longConnServer.Encode(resp)
+	encodedBuf, err := c.Encoder.Encode(resp)
 	if err != nil {
 		return err
 	}
@ -425,3 +437,28 @@ func (c *Client) writePongMsg(appData string) error {

 	return errs.Wrap(err)
 }
+
+func (c *Client) handlerTextMessage(b []byte) error {
+	var msg TextMessage
+	if err := json.Unmarshal(b, &msg); err != nil {
+		return err
+	}
+	switch msg.Type {
+	case TextPong:
+		return nil
+	case TextPing:
+		msg.Type = TextPong
+		msgData, err := json.Marshal(msg)
+		if err != nil {
+			return err
+		}
+		c.w.Lock()
+		defer c.w.Unlock()
+		if err := c.conn.SetWriteDeadline(writeWait); err != nil {
+			return err
+		}
+		return c.conn.WriteMessage(MessageText, msgData)
+	default:
+		return fmt.Errorf("not support message type %s", msg.Type)
+	}
+}
--- a/internal/msggateway/constant.go
+++ b/internal/msggateway/constant.go
@ -27,6 +27,12 @@ const (
 	GzipCompressionProtocol = "gzip"
 	BackgroundStatus        = "isBackground"
 	SendResponse            = "isMsgResp"
+	SDKType                 = "sdkType"
+)
+
+const (
+	GoSDK = "go"
+	JsSDK = "js"
 )

 const (
--- a/internal/msggateway/context.go
+++ b/internal/msggateway/context.go
@ -153,6 +153,14 @@ func (c *UserConnContext) GetCompression() bool {
 	return false
 }

+func (c *UserConnContext) GetSDKType() string {
+	sdkType := c.Req.URL.Query().Get(SDKType)
+	if sdkType == "" {
+		sdkType = GoSDK
+	}
+	return sdkType
+}
+
 func (c *UserConnContext) ShouldSendResp() bool {
 	errResp, exists := c.Query(SendResponse)
 	if exists {
@ -193,7 +201,11 @@ func (c *UserConnContext) ParseEssentialArgs() error {
 	_, err := strconv.Atoi(platformIDStr)
 	if err != nil {
 		return servererrs.ErrConnArgsErr.WrapMsg("platformID is not int")
-
+	}
+	switch sdkType, _ := c.Query(SDKType); sdkType {
+	case "", GoSDK, JsSDK:
+	default:
+		return servererrs.ErrConnArgsErr.WrapMsg("sdkType is not go or js")
 	}
 	return nil
 }
--- a/internal/msggateway/encoder.go
+++ b/internal/msggateway/encoder.go
@ -17,6 +17,7 @@ package msggateway
 import (
 	"bytes"
 	"encoding/gob"
+	"encoding/json"

 	"github.com/openimsdk/tools/errs"
 )
@ -28,12 +29,12 @@ type Encoder interface {

 type GobEncoder struct{}

-func NewGobEncoder() *GobEncoder {
-	return &GobEncoder{}
+func NewGobEncoder() Encoder {
+	return GobEncoder{}
 }

-func (g *GobEncoder) Encode(data any) ([]byte, error) {
-	buff := bytes.Buffer{}
+func (g GobEncoder) Encode(data any) ([]byte, error) {
+	var buff bytes.Buffer
 	enc := gob.NewEncoder(&buff)
 	if err := enc.Encode(data); err != nil {
 		return nil, errs.WrapMsg(err, "GobEncoder.Encode failed", "action", "encode")
@ -41,7 +42,7 @@ func (g *GobEncoder) Encode(data any) ([]byte, error) {
 	return buff.Bytes(), nil
 }

-func (g *GobEncoder) Decode(encodeData []byte, decodeData any) error {
+func (g GobEncoder) Decode(encodeData []byte, decodeData any) error {
 	buff := bytes.NewBuffer(encodeData)
 	dec := gob.NewDecoder(buff)
 	if err := dec.Decode(decodeData); err != nil {
@ -49,3 +50,25 @@ func (g *GobEncoder) Decode(encodeData []byte, decodeData any) error {
 	}
 	return nil
 }
+
+type JsonEncoder struct{}
+
+func NewJsonEncoder() Encoder {
+	return JsonEncoder{}
+}
+
+func (g JsonEncoder) Encode(data any) ([]byte, error) {
+	b, err := json.Marshal(data)
+	if err != nil {
+		return nil, errs.New("JsonEncoder.Encode failed", "action", "encode")
+	}
+	return b, nil
+}
+
+func (g JsonEncoder) Decode(encodeData []byte, decodeData any) error {
+	err := json.Unmarshal(encodeData, decodeData)
+	if err != nil {
+		return errs.New("JsonEncoder.Decode failed", "action", "decode")
+	}
+	return nil
+}
--- a/internal/msggateway/hub_server.go
+++ b/internal/msggateway/hub_server.go
@ -83,17 +83,11 @@ func NewServer(rpcPort int, longConnServer LongConnServer, conf *Config, ready f
 	return s
 }

-func (s *Server) OnlinePushMsg(
-	context context.Context,
-	req *msggateway.OnlinePushMsgReq,
-) (*msggateway.OnlinePushMsgResp, error) {
+func (s *Server) OnlinePushMsg(context context.Context, req *msggateway.OnlinePushMsgReq) (*msggateway.OnlinePushMsgResp, error) {
 	panic("implement me")
 }

-func (s *Server) GetUsersOnlineStatus(
-	ctx context.Context,
-	req *msggateway.GetUsersOnlineStatusReq,
-) (*msggateway.GetUsersOnlineStatusResp, error) {
+func (s *Server) GetUsersOnlineStatus(ctx context.Context, req *msggateway.GetUsersOnlineStatusReq) (*msggateway.GetUsersOnlineStatusResp, error) {
 	if !authverify.IsAppManagerUid(ctx, s.config.Share.IMAdminUserID) {
 		return nil, errs.ErrNoPermission.WrapMsg("only app manager")
 	}
@ -155,6 +149,7 @@ func (s *Server) pushToUser(ctx context.Context, userID string, msgData *sdkws.M
 			(client.IsBackground && client.PlatformID != constant.IOSPlatformID) {
 			err := client.PushMessage(ctx, msgData)
 			if err != nil {
+				log.ZWarn(ctx, "online push msg failed", err, "userID", userID, "platformID", client.PlatformID)
 				userPlatform.ResultCode = int64(servererrs.ErrPushMsgErr.Code())
 			} else {
 				if _, ok := s.pushTerminal[client.PlatformID]; ok {
@ -220,10 +215,7 @@ func (s *Server) SuperGroupOnlineBatchPushOneMsg(ctx context.Context, req *msgga
 	}
 }

-func (s *Server) KickUserOffline(
-	ctx context.Context,
-	req *msggateway.KickUserOfflineReq,
-) (*msggateway.KickUserOfflineResp, error) {
+func (s *Server) KickUserOffline(ctx context.Context, req *msggateway.KickUserOfflineReq) (*msggateway.KickUserOfflineResp, error) {
 	for _, v := range req.KickUserIDList {
 		clients, _, ok := s.LongConnServer.GetUserPlatformCons(v, int(req.PlatformID))
 		if !ok {
--- a/internal/msggateway/message_handler.go
+++ b/internal/msggateway/message_handler.go
@ -16,6 +16,7 @@ package msggateway

 import (
 	"context"
+	"encoding/json"
 	"sync"

 	"github.com/go-playground/validator/v10"
@ -31,6 +32,16 @@ import (
 	"github.com/openimsdk/tools/utils/jsonutil"
 )

+const (
+	TextPing = "ping"
+	TextPong = "pong"
+)
+
+type TextMessage struct {
+	Type string          `json:"type"`
+	Body json.RawMessage `json:"body"`
+}
+
 type Req struct {
 	ReqIdentifier int32  `json:"reqIdentifier" validate:"required"`
 	Token         string `json:"token"`
--- a/internal/msggateway/ws_server.go
+++ b/internal/msggateway/ws_server.go
@ -37,7 +37,6 @@ type LongConnServer interface {
 	SetKickHandlerInfo(i *kickHandler)
 	SubUserOnlineStatus(ctx context.Context, client *Client, data *Req) ([]byte, error)
 	Compressor
-	Encoder
 	MessageHandler
 }

@ -61,7 +60,7 @@ type WsServer struct {
 	authClient        *rpcclient.Auth
 	disCov            discovery.SvcDiscoveryRegistry
 	Compressor
-	Encoder
+	//Encoder
 	MessageHandler
 	webhookClient *webhook.Client
 }
@ -135,7 +134,6 @@ func NewWsServer(msgGatewayConfig *Config, opts ...Option) *WsServer {
 		clients:         newUserMap(),
 		subscription:    newSubscription(),
 		Compressor:      NewGzipCompressor(),
-		Encoder:         NewGobEncoder(),
 		webhookClient:   webhook.NewWebhookClient(msgGatewayConfig.WebhooksConfig.URL),
 	}
 }
@ -278,14 +276,7 @@ func (ws *WsServer) registerClient(client *Client) {

 	wg.Wait()

-	log.ZDebug(
-		client.ctx,
-		"user online",
-		"online user Num",
-		ws.onlineUserNum.Load(),
-		"online user conn Num",
-		ws.onlineUserConnNum.Load(),
-	)
+	log.ZDebug(client.ctx, "user online", "online user Num", ws.onlineUserNum.Load(), "online user conn Num", ws.onlineUserConnNum.Load())
 }

 func getRemoteAdders(client []*Client) string {
--- a/internal/push/offlinepush/jpush/body/notification.go
+++ b/internal/push/offlinepush/jpush/body/notification.go
@ -15,6 +15,7 @@
 package body

 import (
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 )

@ -26,38 +27,44 @@ type Notification struct {

 type Android struct {
 	Alert  string `json:"alert,omitempty"`
+	Title  string `json:"title,omitempty"`
 	Intent struct {
 		URL string `json:"url,omitempty"`
 	} `json:"intent,omitempty"`
-	Extras Extras `json:"extras"`
+	Extras map[string]string `json:"extras,omitempty"`
 }
 type Ios struct {
-	Alert          string `json:"alert,omitempty"`
-	Sound          string `json:"sound,omitempty"`
-	Badge          string `json:"badge,omitempty"`
-	Extras         Extras `json:"extras"`
-	MutableContent bool   `json:"mutable-content"`
+	Alert          IosAlert          `json:"alert,omitempty"`
+	Sound          string            `json:"sound,omitempty"`
+	Badge          string            `json:"badge,omitempty"`
+	Extras         map[string]string `json:"extras,omitempty"`
+	MutableContent bool              `json:"mutable-content"`
 }

-type Extras struct {
-	ClientMsgID string `json:"clientMsgID"`
+type IosAlert struct {
+	Title string `json:"title,omitempty"`
+	Body  string `json:"body,omitempty"`
 }

-func (n *Notification) SetAlert(alert string) {
+func (n *Notification) SetAlert(alert string, title string, opts *options.Opts) {
 	n.Alert = alert
 	n.Android.Alert = alert
-	n.IOS.Alert = alert
-	n.IOS.Sound = "default"
-	n.IOS.Badge = "+1"
+	n.Android.Title = title
+	n.IOS.Alert.Body = alert
+	n.IOS.Alert.Title = title
+	n.IOS.Sound = opts.IOSPushSound
+	if opts.IOSBadgeCount {
+		n.IOS.Badge = "+1"
+	}
 }

-func (n *Notification) SetExtras(extras Extras) {
+func (n *Notification) SetExtras(extras map[string]string) {
 	n.IOS.Extras = extras
 	n.Android.Extras = extras
 }

 func (n *Notification) SetAndroidIntent(pushConf *config.Push) {
-	n.Android.Intent.URL = pushConf.JPNS.PushIntent
+	n.Android.Intent.URL = pushConf.JPush.PushIntent
 }

 func (n *Notification) IOSEnableMutableContent() {
--- a/internal/push/offlinepush/jpush/push.go
+++ b/internal/push/offlinepush/jpush/push.go
@ -18,9 +18,9 @@ import (
 	"context"
 	"encoding/base64"
 	"fmt"
-	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"

 	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/jpush/body"
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/tools/utils/httputil"
 )
@ -57,17 +57,23 @@ func (j *JPush) Push(ctx context.Context, userIDs []string, title, content strin
 	var au body.Audience
 	au.SetAlias(userIDs)
 	var no body.Notification
-	var extras body.Extras
+	extras := make(map[string]string)
+	extras["ex"] = opts.Ex
 	if opts.Signal.ClientMsgID != "" {
-		extras.ClientMsgID = opts.Signal.ClientMsgID
+		extras["ClientMsgID"] = opts.Signal.ClientMsgID
 	}
 	no.IOSEnableMutableContent()
 	no.SetExtras(extras)
-	no.SetAlert(title)
+	no.SetAlert(content, title, opts)
 	no.SetAndroidIntent(j.pushConf)

 	var msg body.Message
 	msg.SetMsgContent(content)
+	msg.SetTitle(title)
+	if opts.Signal.ClientMsgID != "" {
+		msg.SetExtras("ClientMsgID", opts.Signal.ClientMsgID)
+	}
+	msg.SetExtras("ex", opts.Ex)
 	var opt body.Options
 	opt.SetApnsProduction(j.pushConf.IOSPush.Production)
 	var pushObj body.PushObj
@ -76,19 +82,26 @@ func (j *JPush) Push(ctx context.Context, userIDs []string, title, content strin
 	pushObj.SetNotification(&no)
 	pushObj.SetMessage(&msg)
 	pushObj.SetOptions(&opt)
-	var resp any
-	return j.request(ctx, pushObj, resp, 5)
+	var resp map[string]any
+	return j.request(ctx, pushObj, &resp, 5)
 }

-func (j *JPush) request(ctx context.Context, po body.PushObj, resp any, timeout int) error {
-	return j.httpClient.PostReturn(
+func (j *JPush) request(ctx context.Context, po body.PushObj, resp *map[string]any, timeout int) error {
+	err := j.httpClient.PostReturn(
 		ctx,
-		j.pushConf.JPNS.PushURL,
+		j.pushConf.JPush.PushURL,
 		map[string]string{
-			"Authorization": j.getAuthorization(j.pushConf.JPNS.AppKey, j.pushConf.JPNS.MasterSecret),
+			"Authorization": j.getAuthorization(j.pushConf.JPush.AppKey, j.pushConf.JPush.MasterSecret),
 		},
 		po,
 		resp,
 		timeout,
 	)
+	if err != nil {
+		return err
+	}
+	if (*resp)["sendno"] != "0" {
+		return fmt.Errorf("jpush push failed %v", resp)
+	}
+	return nil
 }
--- a/internal/push/offlinepush_handler.go
+++ b/internal/push/offlinepush_handler.go
@ -73,7 +73,7 @@ func (o *OfflinePushConsumerHandler) getOfflinePushInfos(msg *sdkws.MsgData) (ti
 		IsAtSelf   bool     `json:"isAtSelf"`
 	}

-	opts = &options.Opts{Signal: &options.Signal{}}
+	opts = &options.Opts{Signal: &options.Signal{ClientMsgID: msg.ClientMsgID}}
 	if msg.OfflinePushInfo != nil {
 		opts.IOSBadgeCount = msg.OfflinePushInfo.IOSBadgeCount
 		opts.IOSPushSound = msg.OfflinePushInfo.IOSPushSound
--- a/internal/push/push_handler.go
+++ b/internal/push/push_handler.go
@ -4,6 +4,10 @@ import (
 	"context"
 	"encoding/json"

+	"math/rand"
+	"strconv"
+	"time"
+
 	"github.com/IBM/sarama"
 	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
 	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
@ -27,9 +31,6 @@ import (
 	"github.com/openimsdk/tools/utils/timeutil"
 	"github.com/redis/go-redis/v9"
 	"google.golang.org/protobuf/proto"
-	"math/rand"
-	"strconv"
-	"time"
 )

 type ConsumerHandler struct {
@ -165,17 +166,21 @@ func (c *ConsumerHandler) Push2User(ctx context.Context, userIDs []string, msg *
 			return nil
 		}
 	}
-	offlinePushUserID := []string{msg.RecvID}
+	needOfflinePushUserID := []string{msg.RecvID}
+	var offlinePushUserID []string

 	//receiver offline push
-	if err = c.webhookBeforeOfflinePush(ctx, &c.config.WebhooksConfig.BeforeOfflinePush,
-		offlinePushUserID, msg, nil); err != nil {
+	if err = c.webhookBeforeOfflinePush(ctx, &c.config.WebhooksConfig.BeforeOfflinePush, needOfflinePushUserID, msg, &offlinePushUserID); err != nil {
 		return err
 	}
 	log.ZInfo(ctx, "webhookBeforeOfflinePush end")
-	err = c.offlinePushMsg(ctx, msg, offlinePushUserID)
+
+	if len(offlinePushUserID) > 0 {
+		needOfflinePushUserID = offlinePushUserID
+	}
+	err = c.offlinePushMsg(ctx, msg, needOfflinePushUserID)
 	if err != nil {
-		log.ZWarn(ctx, "offlinePushMsg failed", err, "offlinePushUserID", offlinePushUserID, "msg", msg)
+		log.ZWarn(ctx, "offlinePushMsg failed", err, "needOfflinePushUserID", needOfflinePushUserID, "msg", msg)
 		return nil
 	}

@ -335,6 +340,7 @@ func (c *ConsumerHandler) groupMessagesHandler(ctx context.Context, groupID stri
 func (c *ConsumerHandler) offlinePushMsg(ctx context.Context, msg *sdkws.MsgData, offlinePushUserIDs []string) error {
 	title, content, opts, err := c.getOfflinePushInfos(msg)
 	if err != nil {
+		log.ZError(ctx, "getOfflinePushInfos failed", err, "msg", msg)
 		return err
 	}
 	err = c.offlinePusher.Push(ctx, offlinePushUserIDs, title, content, opts)
@ -364,7 +370,7 @@ func (c *ConsumerHandler) getOfflinePushInfos(msg *sdkws.MsgData) (title, conten
 		IsAtSelf   bool     `json:"isAtSelf"`
 	}

-	opts = &options.Opts{Signal: &options.Signal{}}
+	opts = &options.Opts{Signal: &options.Signal{ClientMsgID: msg.ClientMsgID}}
 	if msg.OfflinePushInfo != nil {
 		opts.IOSBadgeCount = msg.OfflinePushInfo.IOSBadgeCount
 		opts.IOSPushSound = msg.OfflinePushInfo.IOSPushSound
--- a/internal/rpc/auth/auth.go
+++ b/internal/rpc/auth/auth.go
@ -16,6 +16,7 @@ package auth

 import (
 	"context"
+	"errors"

 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	redis2 "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
@ -66,6 +67,7 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 			config.Share.Secret,
 			config.RpcConfig.TokenPolicy.Expire,
 			config.Share.MultiLogin,
+			config.Share.IMAdminUserID,
 		),
 		config: config,
 	})
@ -129,6 +131,10 @@ func (s *authServer) parseToken(ctx context.Context, tokensString string) (claim
 	if err != nil {
 		return nil, errs.Wrap(err)
 	}
+	isAdmin := authverify.IsManagerUserID(claims.UserID, s.config.Share.IMAdminUserID)
+	if isAdmin {
+		return claims, nil
+	}
 	m, err := s.authDatabase.GetTokensWithoutError(ctx, claims.UserID, claims.PlatformID)
 	if err != nil {
 		return nil, err
@ -190,7 +196,7 @@ func (s *authServer) forceKickOff(ctx context.Context, userID string, platformID
 	}

 	m, err := s.authDatabase.GetTokensWithoutError(ctx, userID, int(platformID))
-	if err != nil && err != redis.Nil {
+	if err != nil && errors.Is(err, redis.Nil) {
 		return err
 	}
 	for k := range m {
@ -208,7 +214,7 @@ func (s *authServer) forceKickOff(ctx context.Context, userID string, platformID

 func (s *authServer) InvalidateToken(ctx context.Context, req *pbauth.InvalidateTokenReq) (*pbauth.InvalidateTokenResp, error) {
 	m, err := s.authDatabase.GetTokensWithoutError(ctx, req.UserID, int(req.PlatformID))
-	if err != nil && err != redis.Nil {
+	if err != nil && errors.Is(err, redis.Nil) {
 		return nil, err
 	}
 	if m == nil {
--- a/pkg/common/config/config.go
+++ b/pkg/common/config/config.go
@ -69,6 +69,7 @@ type Mongo struct {
 	Database    string   `mapstructure:"database"`
 	Username    string   `mapstructure:"username"`
 	Password    string   `mapstructure:"password"`
+	AuthSource  string   `mapstructure:"authSource"`
 	MaxPoolSize int      `mapstructure:"maxPoolSize"`
 	MaxRetry    int      `mapstructure:"maxRetry"`
 }
@ -212,12 +213,12 @@ type Push struct {
 		FilePath string `mapstructure:"filePath"`
 		AuthURL  string `mapstructure:"authURL"`
 	} `mapstructure:"fcm"`
-	JPNS struct {
+	JPush struct {
 		AppKey       string `mapstructure:"appKey"`
 		MasterSecret string `mapstructure:"masterSecret"`
 		PushURL      string `mapstructure:"pushURL"`
 		PushIntent   string `mapstructure:"pushIntent"`
-	} `mapstructure:"jpns"`
+	} `mapstructure:"jpush"`
 	IOSPush struct {
 		PushSound  string `mapstructure:"pushSound"`
 		BadgeCount bool   `mapstructure:"badgeCount"`
@ -490,6 +491,7 @@ func (m *Mongo) Build() *mongoutil.Config {
 		Database:    m.Database,
 		Username:    m.Username,
 		Password:    m.Password,
+		AuthSource:  m.AuthSource,
 		MaxPoolSize: m.MaxPoolSize,
 		MaxRetry:    m.MaxRetry,
 	}
--- a/pkg/common/storage/controller/auth.go
+++ b/pkg/common/storage/controller/auth.go
@ -2,15 +2,14 @@ package controller

 import (
 	"context"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
-	"github.com/openimsdk/tools/log"
-
 	"github.com/golang-jwt/jwt/v4"
 	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
 	"github.com/openimsdk/protocol/constant"
 	"github.com/openimsdk/tools/errs"
+	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/tokenverify"
 )

@ -36,9 +35,10 @@ type authDatabase struct {
 	accessSecret string
 	accessExpire int64
 	multiLogin   multiLoginConfig
+	adminUserIDs []string
 }

-func NewAuthDatabase(cache cache.TokenModel, accessSecret string, accessExpire int64, multiLogin config.MultiLogin) AuthDatabase {
+func NewAuthDatabase(cache cache.TokenModel, accessSecret string, accessExpire int64, multiLogin config.MultiLogin, adminUserIDs []string) AuthDatabase {
 	return &authDatabase{cache: cache, accessSecret: accessSecret, accessExpire: accessExpire, multiLogin: multiLoginConfig{
 		Policy:       multiLogin.Policy,
 		MaxNumOneEnd: multiLogin.MaxNumOneEnd,
@ -54,7 +54,8 @@ func NewAuthDatabase(cache cache.TokenModel, accessSecret string, accessExpire i
 			constant.IPadPlatformID:       multiLogin.CustomizeLoginNum.IPad,
 			constant.AdminPlatformID:      multiLogin.CustomizeLoginNum.Admin,
 		},
-	}}
+	}, adminUserIDs: adminUserIDs,
+	}
 }

 // If the result is empty.
@ -91,27 +92,31 @@ func (a *authDatabase) BatchSetTokenMapByUidPid(ctx context.Context, tokens []st

 // Create Token.
 func (a *authDatabase) CreateToken(ctx context.Context, userID string, platformID int) (string, error) {
-	tokens, err := a.cache.GetAllTokensWithoutError(ctx, userID)
-	if err != nil {
-		return "", err
-	}
-	deleteTokenKey, kickedTokenKey, err := a.checkToken(ctx, tokens, platformID)
-	if err != nil {
-		return "", err
-	}
-	if len(deleteTokenKey) != 0 {
-		err = a.cache.DeleteTokenByUidPid(ctx, userID, platformID, deleteTokenKey)
+	isAdmin := authverify.IsManagerUserID(userID, a.adminUserIDs)
+	if !isAdmin {
+		tokens, err := a.cache.GetAllTokensWithoutError(ctx, userID)
 		if err != nil {
 			return "", err
 		}
-	}
-	if len(kickedTokenKey) != 0 {
-		for _, k := range kickedTokenKey {
-			err := a.cache.SetTokenFlagEx(ctx, userID, platformID, k, constant.KickedToken)
+
+		deleteTokenKey, kickedTokenKey, err := a.checkToken(ctx, tokens, platformID)
+		if err != nil {
+			return "", err
+		}
+		if len(deleteTokenKey) != 0 {
+			err = a.cache.DeleteTokenByUidPid(ctx, userID, platformID, deleteTokenKey)
 			if err != nil {
 				return "", err
 			}
-			log.ZDebug(ctx, "kicked token in create token", "token", k)
+		}
+		if len(kickedTokenKey) != 0 {
+			for _, k := range kickedTokenKey {
+				err := a.cache.SetTokenFlagEx(ctx, userID, platformID, k, constant.KickedToken)
+				if err != nil {
+					return "", err
+				}
+				log.ZDebug(ctx, "kicked token in create token", "token", k)
+			}
 		}
 	}

@ -122,9 +127,12 @@ func (a *authDatabase) CreateToken(ctx context.Context, userID string, platformI
 		return "", errs.WrapMsg(err, "token.SignedString")
 	}

-	if err = a.cache.SetTokenFlagEx(ctx, userID, platformID, tokenString, constant.NormalToken); err != nil {
-		return "", err
+	if !isAdmin {
+		if err = a.cache.SetTokenFlagEx(ctx, userID, platformID, tokenString, constant.NormalToken); err != nil {
+			return "", err
+		}
 	}
+
 	return tokenString, nil
 }

@ -227,16 +235,16 @@ func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string
 		return nil, nil, errs.New("unknown multiLogin policy").Wrap()
 	}

-	var adminTokenMaxNum = a.multiLogin.MaxNumOneEnd
-	if a.multiLogin.Policy == constant.Customize {
-		adminTokenMaxNum = a.multiLogin.CustomizeLoginNum[constant.AdminPlatformID]
-	}
-	l := len(adminToken)
-	if platformID == constant.AdminPlatformID {
-		l++
-	}
-	if l > adminTokenMaxNum {
-		kickToken = append(kickToken, adminToken[:l-adminTokenMaxNum]...)
-	}
+	//var adminTokenMaxNum = a.multiLogin.MaxNumOneEnd
+	//if a.multiLogin.Policy == constant.Customize {
+	//	adminTokenMaxNum = a.multiLogin.CustomizeLoginNum[constant.AdminPlatformID]
+	//}
+	//l := len(adminToken)
+	//if platformID == constant.AdminPlatformID {
+	//	l++
+	//}
+	//if l > adminTokenMaxNum {
+	//	kickToken = append(kickToken, adminToken[:l-adminTokenMaxNum]...)
+	//}
 	return deleteToken, kickToken, nil
 }
--- a/pkg/common/storage/controller/msg.go
+++ b/pkg/common/storage/controller/msg.go
@ -490,7 +490,7 @@ func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, co
 	}
 	successMsgs, failedSeqs, err := db.msg.GetMessagesBySeq(ctx, conversationID, newSeqs)
 	if err != nil {
-		if err != redis.Nil {
+		if errors.Is(err, redis.Nil) {
 			log.ZError(ctx, "get message from redis exception", err, "failedSeqs", failedSeqs, "conversationID", conversationID)
 		}
 	}
--- a/scripts/create-topic.sh
+++ b/scripts/create-topic.sh
@ -1,55 +0,0 @@
-#!/usr/bin/env bash
-# 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.
-
-# Wait for Kafka to be ready
-
-KAFKA_SERVER=localhost:9092
-
-MAX_ATTEMPTS=300
-attempt_num=1
-
-echo "Waiting for Kafka to be ready..."
-
-until /opt/bitnami/kafka/bin/kafka-topics.sh --list --bootstrap-server $KAFKA_SERVER; do
-  echo "Attempt $attempt_num of $MAX_ATTEMPTS: Kafka not ready yet..."
-  if [ $attempt_num -eq $MAX_ATTEMPTS ]; then
-    echo "Kafka not ready after $MAX_ATTEMPTS attempts, exiting"
-    exit 1
-  fi
-  attempt_num=$((attempt_num+1))
-  sleep 1
-done
-
-echo "Kafka is ready. Creating topics..."
-
-
-topics=("toRedis" "toMongo" "toPush" "toOfflinePush")
-partitions=8
-replicationFactor=1
-
-for topic in "${topics[@]}"; do
-  if /opt/bitnami/kafka/bin/kafka-topics.sh --create \
-    --bootstrap-server $KAFKA_SERVER \
-    --replication-factor $replicationFactor \
-    --partitions $partitions \
-    --topic $topic
-  then
-    echo "Topic $topic created."
-  else
-    echo "Failed to create topic $topic."
-  fi
-done
-
-echo "All topics created."
--- a/scripts/mongo-init.sh
+++ b/scripts/mongo-init.sh
@ -1,66 +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.
-mongosh <<EOF
-var maxRetries = 300;
-var connected = false;
-var rootUsername = '$MONGO_INITDB_ROOT_USERNAME';
-var rootPassword = '$MONGO_INITDB_ROOT_PASSWORD';
-var dbName = '$MONGO_INITDB_DATABASE';
-var openimUsername = '$MONGO_OPENIM_USERNAME';
-var openimPassword = '$MONGO_OPENIM_PASSWORD';
-
-while (!connected && maxRetries > 0) {
-    try {
-        db = connect('mongodb://127.0.0.1:27017/admin');
-        var authResult = db.auth(rootUsername, rootPassword);
-        if (authResult) {
-            print('Authentication successful for root user: ' + rootUsername);
-            connected = true;
-        } else {
-            print('Authentication failed for root user: ' + rootUsername + ' with password: ' + rootPassword);
-            quit(1);
-        }
-    } catch (e) {
-        maxRetries--;
-        print('Connection failed, retrying... Remaining attempts: ' + maxRetries);
-        sleep(1000); // Sleep for 1 second
-    }
-}
-
-if (connected) {
-    db = db.getSiblingDB(dbName);
-    var createUserResult = db.createUser({
-        user: openimUsername,
-        pwd: openimPassword,
-        roles: [{
-            role: 'readWrite',
-            db: dbName
-        }]
-    });
-
-    if (createUserResult.ok == 1) {
-        print('User creation successful. User: ' + openimUsername + ', Database: ' + dbName);
-    } else {
-        print('User creation failed for user: ' + openimUsername + ' in database: ' + dbName);
-        quit(1);
-    }
-} else {
-    print('Failed to connect to MongoDB after 300 retries.');
-    quit(1);
-}
-EOF
-
-
-
-
--- a/tools/changelog/changelog.go
+++ b/tools/changelog/changelog.go
@ -0,0 +1,198 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"regexp"
+	"strings"
+)
+
+// You can specify a tag as a command line argument to generate the changelog for a specific version.
+// Example: go run tools/changelog/changelog.go v0.0.33
+// If no tag is provided, the latest release will be used.
+
+// Setting repo owner and repo name by generate changelog
+const (
+	repoOwner = "openimsdk"
+	repoName  = "open-im-server"
+)
+
+// GitHubRepo struct represents the repo details.
+type GitHubRepo struct {
+	Owner         string
+	Repo          string
+	FullChangelog string
+}
+
+// ReleaseData represents the JSON structure for release data.
+type ReleaseData struct {
+	TagName   string `json:"tag_name"`
+	Body      string `json:"body"`
+	HtmlUrl   string `json:"html_url"`
+	Published string `json:"published_at"`
+}
+
+// Method to classify and format release notes.
+func (g *GitHubRepo) classifyReleaseNotes(body string) map[string][]string {
+	result := map[string][]string{
+		"feat":     {},
+		"fix":      {},
+		"chore":    {},
+		"refactor": {},
+		"build":    {},
+		"other":    {},
+	}
+
+	// Regular expression to extract PR number and URL (case insensitive)
+	rePR := regexp.MustCompile(`(?i)in (https://github\.com/[^\s]+/pull/(\d+))`)
+
+	// Split the body into individual lines.
+	lines := strings.Split(body, "\n")
+
+	for _, line := range lines {
+		// Skip lines that contain "deps: Merge"
+		if strings.Contains(strings.ToLower(line), "deps: merge #") {
+			continue
+		}
+
+		// Use a regular expression to extract Full Changelog link and its title (case insensitive).
+		if strings.Contains(strings.ToLower(line), "**full changelog**") {
+			matches := regexp.MustCompile(`(?i)\*\*full changelog\*\*: (https://github\.com/[^\s]+/compare/([^\s]+))`).FindStringSubmatch(line)
+			if len(matches) > 2 {
+				// Format the Full Changelog link with title
+				g.FullChangelog = fmt.Sprintf("[%s](%s)", matches[2], matches[1])
+			}
+			continue // Skip further processing for this line.
+		}
+
+		if strings.HasPrefix(line, "*") {
+			var category string
+
+			// Use strings.ToLower to make the matching case insensitive
+			lowerLine := strings.ToLower(line)
+
+			// Determine the category based on the prefix (case insensitive).
+			if strings.HasPrefix(lowerLine, "* feat") {
+				category = "feat"
+			} else if strings.HasPrefix(lowerLine, "* fix") {
+				category = "fix"
+			} else if strings.HasPrefix(lowerLine, "* chore") {
+				category = "chore"
+			} else if strings.HasPrefix(lowerLine, "* refactor") {
+				category = "refactor"
+			} else if strings.HasPrefix(lowerLine, "* build") {
+				category = "build"
+			} else {
+				category = "other"
+			}
+
+			// Extract PR number and URL (case insensitive)
+			matches := rePR.FindStringSubmatch(line)
+			if len(matches) == 3 {
+				prURL := matches[1]
+				prNumber := matches[2]
+				// Format the line with the PR link and use original content for the final result
+				formattedLine := fmt.Sprintf("* %s [#%s](%s)", strings.Split(line, " by ")[0][2:], prNumber, prURL)
+				result[category] = append(result[category], formattedLine)
+			} else {
+				// If no PR link is found, just add the line as is
+				result[category] = append(result[category], line)
+			}
+		}
+	}
+
+	return result
+}
+
+// Method to generate the final changelog.
+func (g *GitHubRepo) generateChangelog(tag, date, htmlURL, body string) string {
+	sections := g.classifyReleaseNotes(body)
+
+	// Convert ISO 8601 date to simpler format (YYYY-MM-DD)
+	formattedDate := date[:10]
+
+	// Changelog header with tag, date, and links.
+	changelog := fmt.Sprintf("## [%s](%s) \t(%s)\n\n", tag, htmlURL, formattedDate)
+
+	if len(sections["feat"]) > 0 {
+		changelog += "### New Features\n" + strings.Join(sections["feat"], "\n") + "\n\n"
+	}
+	if len(sections["fix"]) > 0 {
+		changelog += "### Bug Fixes\n" + strings.Join(sections["fix"], "\n") + "\n\n"
+	}
+	if len(sections["chore"]) > 0 {
+		changelog += "### Chores\n" + strings.Join(sections["chore"], "\n") + "\n\n"
+	}
+	if len(sections["refactor"]) > 0 {
+		changelog += "### Refactors\n" + strings.Join(sections["refactor"], "\n") + "\n\n"
+	}
+	if len(sections["build"]) > 0 {
+		changelog += "### Builds\n" + strings.Join(sections["build"], "\n") + "\n\n"
+	}
+	if len(sections["other"]) > 0 {
+		changelog += "### Others\n" + strings.Join(sections["other"], "\n") + "\n\n"
+	}
+
+	if g.FullChangelog != "" {
+		changelog += fmt.Sprintf("**Full Changelog**: %s\n", g.FullChangelog)
+	}
+
+	return changelog
+}
+
+// Method to fetch release data from GitHub API.
+func (g *GitHubRepo) fetchReleaseData(version string) (*ReleaseData, error) {
+	var apiURL string
+
+	if version == "" {
+		// Fetch the latest release.
+		apiURL = fmt.Sprintf("https://api.github.com/repos/%s/%s/releases/latest", g.Owner, g.Repo)
+	} else {
+		// Fetch a specific version.
+		apiURL = fmt.Sprintf("https://api.github.com/repos/%s/%s/releases/tags/%s", g.Owner, g.Repo, version)
+	}
+
+	resp, err := http.Get(apiURL)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	var releaseData ReleaseData
+	err = json.Unmarshal(body, &releaseData)
+	if err != nil {
+		return nil, err
+	}
+
+	return &releaseData, nil
+}
+
+func main() {
+	repo := &GitHubRepo{Owner: repoOwner, Repo: repoName}
+
+	// Get the version from command line arguments, if provided
+	var version string // Default is use latest
+
+	if len(os.Args) > 1 {
+		version = os.Args[1] // Use the provided version
+	}
+
+	// Fetch release data (either for latest or specific version)
+	releaseData, err := repo.fetchReleaseData(version)
+	if err != nil {
+		fmt.Println("Error fetching release data:", err)
+		return
+	}
+
+	// Generate and print the formatted changelog
+	changelog := repo.generateChangelog(releaseData.TagName, releaseData.Published, releaseData.HtmlUrl, releaseData.Body)
+	fmt.Println(changelog)
+}
--- a/tools/changelog/main.go
+++ b/tools/changelog/main.go
@ -1,308 +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 main
-
-import (
-	"fmt"
-	"log"
-	"os"
-	"os/exec"
-	"regexp"
-	"sort"
-	"strings"
-)
-
-var (
-	mergeRequest   = regexp.MustCompile(`Merge pull request #([\d]+)`)
-	webconsoleBump = regexp.MustCompile(regexp.QuoteMeta("bump(github.com/openshift/origin-web-console): ") + `([\w]+)`)
-	upstreamKube   = regexp.MustCompile(`^UPSTREAM: (\d+)+:(.+)`)
-	upstreamRepo   = regexp.MustCompile(`^UPSTREAM: ([\w/-]+): (\d+)+:(.+)`)
-	prefix         = regexp.MustCompile(`^[\w-]: `)
-
-	assignments = []prefixAssignment{
-		{"cluster up", "cluster"},
-		{" pv ", "storage"},
-		{"haproxy", "router"},
-		{"router", "router"},
-		{"route", "route"},
-		{"authoriz", "auth"},
-		{"rbac", "auth"},
-		{"authent", "auth"},
-		{"reconcil", "auth"},
-		{"auth", "auth"},
-		{"role", "auth"},
-		{" dc ", "deploy"},
-		{"deployment", "deploy"},
-		{"rolling", "deploy"},
-		{"security context constr", "security"},
-		{"scc", "security"},
-		{"pipeline", "build"},
-		{"build", "build"},
-		{"registry", "registry"},
-		{"registries", "image"},
-		{"image", "image"},
-		{" arp ", "network"},
-		{" cni ", "network"},
-		{"egress", "network"},
-		{"network", "network"},
-		{"oc ", "cli"},
-		{"template", "template"},
-		{"etcd", "server"},
-		{"pod", "node"},
-		{"scripts/", "hack"},
-		{"e2e", "test"},
-		{"integration", "test"},
-		{"cluster", "cluster"},
-		{"master", "server"},
-		{"packages", "hack"},
-		{"api", "server"},
-	}
-)
-
-type prefixAssignment struct {
-	term   string
-	prefix string
-}
-
-type commit struct {
-	short   string
-	parents []string
-	message string
-}
-
-func contains(arr []string, value string) bool {
-	for _, s := range arr {
-		if s == value {
-			return true
-		}
-	}
-	return false
-}
-
-func main() {
-	log.SetFlags(0)
-	if len(os.Args) != 3 {
-		log.Fatalf("Must specify two arguments, FROM and TO")
-	}
-	from := os.Args[1]
-	to := os.Args[2]
-
-	out, err := exec.Command("git", "log", "--topo-order", "--pretty=tformat:%h %p|%s", "--reverse", fmt.Sprintf("%s..%s", from, to)).CombinedOutput()
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	hide := make(map[string]struct{})
-	var apiChanges []string
-	var webconsole []string
-	var commits []commit
-	var upstreams []commit
-	var bumps []commit
-	for _, line := range strings.Split(string(out), "\n") {
-		if len(strings.TrimSpace(line)) == 0 {
-			continue
-		}
-		parts := strings.SplitN(line, "|", 2)
-		hashes := strings.Split(parts[0], " ")
-		c := commit{short: hashes[0], parents: hashes[1:], message: parts[1]}
-
-		if strings.HasPrefix(c.message, "UPSTREAM: ") {
-			hide[c.short] = struct{}{}
-			upstreams = append(upstreams, c)
-		}
-		if strings.HasPrefix(c.message, "bump(") {
-			hide[c.short] = struct{}{}
-			bumps = append(bumps, c)
-		}
-
-		if len(c.parents) == 1 {
-			commits = append(commits, c)
-			continue
-		}
-
-		matches := mergeRequest.FindStringSubmatch(line)
-		if len(matches) == 0 {
-			// this may have been a human pressing the merge button, we'll just record this as a direct push
-			continue
-		}
-
-		// split the accumulated commits into any that are force merges (assumed to be the initial set due
-		// to --topo-order) from the PR commits as soon as we see any of our merge parents. Then print
-		// any of the force merges
-		var first int
-		for i := range commits {
-			first = i
-			if contains(c.parents, commits[i].short) {
-				first++
-				break
-			}
-		}
-		individual := commits[:first]
-		merged := commits[first:]
-		for _, commit := range individual {
-			if len(commit.parents) > 1 {
-				continue
-			}
-			if _, ok := hide[commit.short]; ok {
-				continue
-			}
-			fmt.Printf("force-merge: %s %s\n", commit.message, commit.short)
-		}
-
-		// try to find either the PR title or the first commit title from the merge commit
-		out, err := exec.Command("git", "show", "--pretty=tformat:%b", c.short).CombinedOutput()
-		if err != nil {
-			log.Fatal(err)
-		}
-		var message string
-		para := strings.Split(string(out), "\n\n")
-		if len(para) > 0 && strings.HasPrefix(para[0], "Automatic merge from submit-queue") {
-			para = para[1:]
-		}
-		// this is no longer necessary with the submit queue in place
-		if len(para) > 0 && strings.HasPrefix(para[0], "Merged by ") {
-			para = para[1:]
-		}
-		// post submit-queue, the merge bot will add the PR title, which is usually pretty good
-		if len(para) > 0 {
-			message = strings.Split(para[0], "\n")[0]
-		}
-		if len(message) == 0 && len(merged) > 0 {
-			message = merged[0].message
-		}
-		if len(message) > 0 && len(merged) == 1 && message == merged[0].message {
-			merged = nil
-		}
-
-		// try to calculate a prefix based on the diff
-		if len(message) > 0 && !prefix.MatchString(message) {
-			prefix, ok := findPrefixFor(message, merged)
-			if ok {
-				message = prefix + ": " + message
-			}
-		}
-
-		// github merge
-
-		// has api changes
-		display := fmt.Sprintf("%s [\\#%s](https://github.com/openimsdk/Open-IM-Server/pull/%s)", message, matches[1], matches[1])
-		if hasFileChanges(c.short, "pkg/apistruct/") {
-			apiChanges = append(apiChanges, display)
-		}
-
-		var filtered []commit
-		for _, commit := range merged {
-			if _, ok := hide[commit.short]; ok {
-				continue
-			}
-			filtered = append(filtered, commit)
-		}
-		if len(filtered) > 0 {
-			fmt.Printf("- %s\n", display)
-			for _, commit := range filtered {
-				fmt.Printf("  - %s (%s)\n", commit.message, commit.short)
-			}
-		}
-
-		// stick the merge commit in at the beginning of the next list so we can anchor the previous parent
-		commits = []commit{c}
-	}
-
-	// chunk the bumps
-	var lines []string
-	for _, commit := range bumps {
-		if m := webconsoleBump.FindStringSubmatch(commit.message); len(m) > 0 {
-			webconsole = append(webconsole, m[1])
-			continue
-		}
-		lines = append(lines, commit.message)
-	}
-	lines = sortAndUniq(lines)
-	for _, line := range lines {
-		fmt.Printf("- %s\n", line)
-	}
-
-	// chunk the upstreams
-	lines = nil
-	for _, commit := range upstreams {
-		lines = append(lines, commit.message)
-	}
-	lines = sortAndUniq(lines)
-	for _, line := range lines {
-		fmt.Printf("- %s\n", upstreamLinkify(line))
-	}
-
-	if len(webconsole) > 0 {
-		fmt.Printf("- web: from %s^..%s\n", webconsole[0], webconsole[len(webconsole)-1])
-	}
-
-	for _, apiChange := range apiChanges {
-		fmt.Printf("  - %s\n", apiChange)
-	}
-}
-
-func findPrefixFor(message string, commits []commit) (string, bool) {
-	message = strings.ToLower(message)
-	for _, m := range assignments {
-		if strings.Contains(message, m.term) {
-			return m.prefix, true
-		}
-	}
-	for _, c := range commits {
-		if prefix, ok := findPrefixFor(c.message, nil); ok {
-			return prefix, ok
-		}
-	}
-	return "", false
-}
-
-func hasFileChanges(commit string, prefixes ...string) bool {
-	out, err := exec.Command("git", "diff", "--name-only", fmt.Sprintf("%s^..%s", commit, commit)).CombinedOutput()
-	if err != nil {
-		log.Fatal(err)
-	}
-	for _, file := range strings.Split(string(out), "\n") {
-		for _, prefix := range prefixes {
-			if strings.HasPrefix(file, prefix) {
-				return true
-			}
-		}
-	}
-	return false
-}
-
-func sortAndUniq(lines []string) []string {
-	sort.Strings(lines)
-	out := make([]string, 0, len(lines))
-	last := ""
-	for _, s := range lines {
-		if last == s {
-			continue
-		}
-		last = s
-		out = append(out, s)
-	}
-	return out
-}
-
-func upstreamLinkify(line string) string {
-	if m := upstreamKube.FindStringSubmatch(line); len(m) > 0 {
-		return fmt.Sprintf("UPSTREAM: [#%s](https://github.com/openimsdk/open-im-server/pull/%s):%s", m[1], m[1], m[2])
-	}
-	if m := upstreamRepo.FindStringSubmatch(line); len(m) > 0 {
-		return fmt.Sprintf("UPSTREAM: [%s#%s](https://github.com/%s/pull/%s):%s", m[1], m[2], m[1], m[2], m[3])
-	}
-	return line
-}
--- a/tools/check-component/main.go
+++ b/tools/check-component/main.go
@ -66,7 +66,7 @@ func CheckMinIO(ctx context.Context, config *config.Minio) error {
 }

 func CheckKafka(ctx context.Context, conf *config.Kafka) error {
-	return kafka.Check(ctx, conf.Build(), []string{conf.ToMongoTopic, conf.ToRedisTopic, conf.ToPushTopic, conf.ToOfflinePushTopic})
+	return kafka.CheckHealth(ctx, conf.Build())
 }

 func initConfig(configDir string) (*config.Mongo, *config.Redis, *config.Kafka, *config.Minio, *config.Discovery, error) {