feat: gomake upgrade (#3702)

* fix: performance issues with Kafka caused by encapsulating the MQ interface

* fix: admin token in standalone mode

* fix: full id version

* fix: resolve deadlock in cache eviction and improve GetBatch implementation

* refactor: replace LongConn with ClientConn interface and simplify message handling

* refactor: replace LongConn with ClientConn interface and simplify message handling

* fix: seq use $setOnInsert for min_seq in conversation update

* feat: add error code for handled friend requests and improve error handling in friend operations

.env

@@ -1,8 +1,8 @@
 MONGO_IMAGE=mongo:7.0
 REDIS_IMAGE=redis:7.0.0
-KAFKA_IMAGE=bitnami/kafka:3.5.1
+KAFKA_IMAGE=bitnamilegacy/kafka:3.5.1
 MINIO_IMAGE=minio/minio:RELEASE.2024-01-11T07-46-16Z
-ETCD_IMAGE=bitnami/etcd:3.5.13
+ETCD_IMAGE=bitnamilegacy/etcd:3.5.13
 PROMETHEUS_IMAGE=prom/prometheus:v2.45.6
 ALERTMANAGER_IMAGE=prom/alertmanager:v0.27.0
 GRAFANA_IMAGE=grafana/grafana:11.0.1

.github/workflows/publish-docker-image.yml

@@ -4,42 +4,80 @@ on:
   push:
     branches:
       - release-*
-    # tags:
-    #   - 'v*'
-
   release:
     types: [published]
-
   workflow_dispatch:
     inputs:
       tag:
         description: "Tag version to be used for Docker image"
         required: true
-        default: "v3.8.0"
+        default: "v3.8.3"
+
+env:
+  GO_VERSION: "1.22"
+  IMAGE_NAME: "openim-server"
+  # IMAGE_NAME: ${{ github.event.repository.name }}
+  DOCKER_BUILDKIT: 1
 
 jobs:
-  build-and-test:
+  publish-docker-images:
     runs-on: ubuntu-latest
+    if: ${{ !(github.event_name == 'pull_request' && github.event.pull_request.merged == false) }}
     steps:
-      - uses: actions/checkout@v4
+      - name: Checkout main repository
+        uses: actions/checkout@v4
         with:
           path: main-repo
 
-      # - name: Set up QEMU
-      #   uses: docker/setup-qemu-action@v3.3.0
+      - name: Set up QEMU
+        uses: docker/setup-qemu-action@v3.3.0
 
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3.8.0
-
-      - name: Build Docker image
-        id: build
-        uses: docker/build-push-action@v5
+        id: buildx
+        uses: docker/setup-buildx-action@v3
         with:
-          context: ./main-repo
-          load: true
-          tags: "openim/openim-server:local"
-          cache-from: type=gha,scope=build
-          cache-to: type=gha,mode=max,scope=build
+          driver-opts: network=host
+
+      - name: Extract metadata for Docker
+        id: meta
+        uses: docker/metadata-action@v5.6.0
+        with:
+          images: |
+            ${{ secrets.DOCKER_USERNAME }}/${{ env.IMAGE_NAME }}
+            ghcr.io/${{ github.repository_owner }}/${{ env.IMAGE_NAME }}
+            registry.cn-hangzhou.aliyuncs.com/openimsdk/${{ env.IMAGE_NAME }}
+          tags: |
+            type=ref,event=tag
+            type=schedule
+            type=ref,event=branch
+            type=ref,event=pr
+            type=semver,pattern={{version}}
+            type=semver,pattern=v{{version}}
+            type=semver,pattern={{major}}.{{minor}}
+            type=semver,pattern={{major}}
+            type=sha
+
+      - name: Install skopeo
+        run: |
+          sudo apt-get update && sudo apt-get install -y skopeo
+
+      - name: Build multi-arch images as OCI
+        run: |
+          mkdir -p /tmp/oci-image /tmp/docker-cache
+
+          # Build multi-architecture image and save in OCI format        
+          docker buildx build \
+            --platform linux/amd64,linux/arm64 \
+            --output type=oci,dest=/tmp/oci-image/multi-arch.tar \
+            --cache-to type=local,dest=/tmp/docker-cache \
+            --cache-from type=gha \
+            ./main-repo
+
+            # Use skopeo to convert the amd64 image from OCI format to Docker format and load it
+          skopeo copy --override-arch amd64 oci-archive:/tmp/oci-image/multi-arch.tar docker-daemon:${{ secrets.DOCKER_USERNAME }}/${{ env.IMAGE_NAME }}:local
+
+          # check image
+          docker image ls | grep openim
 
       - name: Checkout compose repository
         uses: actions/checkout@v4
@@ -52,11 +90,11 @@ jobs:
         run: |
           IP=$(hostname -I | awk '{print $1}')
           echo "The IP Address is: $IP"
-          echo "::set-output name=ip::$IP"
+          echo "ip=$IP" >> $GITHUB_OUTPUT
 
       - name: Update .env to use the local image
         run: |
-          sed -i 's|OPENIM_SERVER_IMAGE=.*|OPENIM_SERVER_IMAGE=openim/openim-server:local|' ${{ github.workspace }}/compose-repo/.env
+          sed -i 's|OPENIM_SERVER_IMAGE=.*|OPENIM_SERVER_IMAGE=${{ secrets.DOCKER_USERNAME }}/${{ env.IMAGE_NAME }}:local|' ${{ github.workspace }}/compose-repo/.env
           sed -i 's|MINIO_EXTERNAL_ADDRESS=.*|MINIO_EXTERNAL_ADDRESS=http://${{ steps.get-ip.outputs.ip }}:10005|' ${{ github.workspace }}/compose-repo/.env
 
       - name: Start services using Docker Compose
@@ -66,23 +104,34 @@ jobs:
 
           docker compose ps
 
-      - name: Extract metadata for Docker (tags, labels)
-        id: meta
-        uses: docker/metadata-action@v5.6.0
-        with:
-          images: |
-            openim/openim-server
-            ghcr.io/openimsdk/openim-server
-            registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-server
-          tags: |
-            type=ref,event=tag
-            type=schedule
-            type=ref,event=branch
-            # type=semver,pattern={{version}}
-            type=semver,pattern=v{{version}}
-            type=semver,pattern=release-{{raw}}
-            type=sha
-            type=raw,value=${{ github.event.inputs.tag }}
+      # - name: Check openim-server health
+      #   run: |
+      #     timeout=300
+      #     interval=30
+      #     elapsed=0
+      #     while [[ $elapsed -le $timeout ]]; do
+      #       if ! docker exec openim-server mage check; then
+      #         echo "openim-server is not ready, waiting..."
+      #         sleep $interval
+      #         elapsed=$(($elapsed + $interval))
+      #       else
+      #         echo "Health check successful"
+      #         exit 0
+      #       fi
+      #     done
+      #     echo "Health check failed after 5 minutes"
+      #     exit 1
+
+      # - name: Check openim-chat health
+      #   if: success()
+      #   run: |
+      #     if ! docker exec openim-chat mage check; then
+      #         echo "openim-chat check failed"
+      #         exit 1
+      #       else
+      #         echo "Health check successful"
+      #         exit 0
+      #       fi
 
       - name: Log in to Docker Hub
         uses: docker/login-action@v3.3.0
@@ -104,22 +153,27 @@ jobs:
           username: ${{ secrets.ALIREGISTRY_USERNAME }}
           password: ${{ secrets.ALIREGISTRY_TOKEN }}
 
-      - name: Push Docker images
-        uses: docker/build-push-action@v5
-        with:
-          context: ./main-repo
-          push: true
-          platforms: linux/amd64,linux/arm64
-          tags: ${{ steps.meta.outputs.tags }}
-          labels: ${{ steps.meta.outputs.labels }}
-          cache-from: type=gha,scope=build
-          cache-to: type=gha,mode=max,scope=build
+      - name: Push multi-architecture images
+        if: success()
+        run: |
+          docker buildx build \
+            --platform linux/amd64,linux/arm64 \
+            $(echo "${{ steps.meta.outputs.tags }}" | sed 's/,/ --tag /g' | sed 's/^/--tag /') \
+            --cache-from type=local,src=/tmp/docker-cache \
+            --push \
+            ./main-repo
 
       - name: Verify multi-platform support
         run: |
-          images=("openim/openim-server" "ghcr.io/openimsdk/openim-server" "registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-server")
+          images=(
+            "${{ secrets.DOCKER_USERNAME }}/${{ env.IMAGE_NAME }}"
+            "ghcr.io/${{ github.repository_owner }}/${{ env.IMAGE_NAME }}"
+            "registry.cn-hangzhou.aliyuncs.com/openimsdk/${{ env.IMAGE_NAME }}"
+          )
+
           for image in "${images[@]}"; do
-              for tag in $(echo "${{ steps.meta.outputs.tags }}" | tr ',' '\n'); do
+              for tag in $(echo "${{ steps.meta.outputs.tags }}" | tr ',' '\n' | cut -d':' -f2); do
+                  echo "Verifying multi-arch support for $image:$tag"
                   manifest=$(docker manifest inspect "$image:$tag" || echo "error")
                   if [[ "$manifest" == "error" ]]; then
                       echo "Manifest not found for $image:$tag"
@@ -135,5 +189,6 @@ jobs:
                       echo "Multi-platform support check failed for $image:$tag - missing arm64"
                       exit 1
                   fi
+                  echo "✅ $image:$tag supports both amd64 and arm64 architectures"
               done
           done

.github/workflows/update-version-file-on-release.yml

@@ -96,13 +96,13 @@ jobs:
                   repo,
                   per_page: 100
                 });
-                
+            
                 release = releases.data.find(r => r.draft && r.tag_name === tagName);            
                 if (!release) {
                   throw new Error(`No release found with tag ${tagName}`);
                 }
               }
-              
+            
               await github.rest.repos.updateRelease({
                 owner,
                 repo,
@@ -110,10 +110,10 @@ jobs:
                 draft: false,
                 prerelease: release.prerelease
               });
-              
+            
               const status = release.draft ? "was draft" : "was already published";
               core.info(`Release ${tagName} ensured to be published (${status}).`);
-              
+            
             } catch (error) {
               core.warning(`Could not find or update release for tag ${tagName}: ${error.message}`);
             }

config/openim-api.yml

@@ -17,3 +17,13 @@ prometheus:
   ports:
   # This address can be accessed via a browser
   grafanaURL:
+
+ratelimiter:
+  # Whether to enable rate limiting
+  enable: false
+  # WindowSize defines time duration per window
+  window: 20s
+  # BucketNum defines bucket number for each window
+  bucket: 500
+  # CPU threshold; valid range 0–1000 (1000 = 100%)
+  cpuThreshold: 850

config/openim-msggateway.yml

@@ -26,3 +26,20 @@ longConnSvr:
   websocketMaxMsgLen: 4096
   # WebSocket connection handshake timeout in seconds
   websocketTimeout: 10
+
+ratelimiter:
+  # Whether to enable rate limiting
+  enable: false
+  # WindowSize defines time duration per window
+  window: 20s
+  # BucketNum defines bucket number for each window
+  bucket: 500
+  # CPU threshold; valid range 0–1000 (1000 = 100%)
+  cpuThreshold: 850
+
+circuitBreaker:
+  enable: false
+  window: 5s            # Time window size (seconds)
+  bucket: 100            # Number of buckets
+  success: 0.6          # Success rate threshold (0.6 means 60%)
+  request: 500 # Request threshold; circuit breaker evaluation occurs when reached

config/openim-msgtransfer.yml

@@ -6,3 +6,20 @@ prometheus:
   # List of ports that Prometheus listens on; each port corresponds to an instance of monitoring. Ensure these are managed accordingly
   # It will only take effect when autoSetPorts is set to false.
   ports:
+
+ratelimiter:
+  # Whether to enable rate limiting
+  enable: false
+  # WindowSize defines time duration per window
+  window: 20s
+  # BucketNum defines bucket number for each window
+  bucket: 500
+  # CPU threshold; valid range 0–1000 (1000 = 100%)
+  cpuThreshold: 850
+
+circuitBreaker:
+  enable: false
+  window: 5s            # Time window size (seconds)
+  bucket: 100            # Number of buckets
+  success: 0.6          # Success rate threshold (0.6 means 60%)
+  request: 500 # Request threshold; circuit breaker evaluation occurs when reached

config/openim-push.yml

@@ -10,10 +10,26 @@ rpc:
   # It will only take effect when autoSetPorts is set to false.
   ports:
 
+ratelimiter:
+  # Whether to enable rate limiting
+  enable: false
+  # WindowSize defines time duration per window
+  window: 20s
+  # BucketNum defines bucket number for each window
+  bucket: 500
+  # CPU threshold; valid range 0–1000 (1000 = 100%)
+  cpuThreshold: 850
+
+circuitBreaker:
+  enable: false
+  window: 5s            # Time window size (seconds)
+  bucket: 100            # Number of buckets
+  success: 0.6          # Success rate threshold (0.6 means 60%)
+  request: 500 # Request threshold; circuit breaker evaluation occurs when reached
 
 prometheus:
   # Enable or disable Prometheus monitoring
-  enable: true
+  enable: false
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
   # It will only take effect when autoSetPorts is set to false.
   ports:

config/openim-rpc-auth.yml

@@ -20,3 +20,20 @@ prometheus:
 tokenPolicy:
   # Token validity period, in days
   expire: 90
+
+ratelimiter:
+  # Whether to enable rate limiting
+  enable: false
+  # WindowSize defines time duration per window
+  window: 20s
+  # BucketNum defines bucket number for each window
+  bucket: 500
+  # CPU threshold; valid range 0–1000 (1000 = 100%)
+  cpuThreshold: 850
+
+circuitBreaker:
+  enable: false
+  window: 5s            # Time window size (seconds)
+  bucket: 100            # Number of buckets
+  success: 0.6          # Success rate threshold (0.6 means 60%)
+  request: 500 # Request threshold; circuit breaker evaluation occurs when reached

config/openim-rpc-conversation.yml

@@ -16,3 +16,20 @@ prometheus:
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
   # It will only take effect when autoSetPorts is set to false.
   ports:
+
+ratelimiter:
+  # Whether to enable rate limiting
+  enable: false
+  # WindowSize defines time duration per window
+  window: 20s
+  # BucketNum defines bucket number for each window
+  bucket: 500
+  # CPU threshold; valid range 0–1000 (1000 = 100%)
+  cpuThreshold: 850
+
+circuitBreaker:
+  enable: false
+  window: 5s            # Time window size (seconds)
+  bucket: 100            # Number of buckets
+  success: 0.6          # Success rate threshold (0.6 means 60%)
+  request: 500 # Request threshold; circuit breaker evaluation occurs when reached

config/openim-rpc-friend.yml

@@ -16,3 +16,20 @@ prometheus:
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
   # It will only take effect when autoSetPorts is set to false.
   ports:
+
+ratelimiter:
+  # Whether to enable rate limiting
+  enable: false
+  # WindowSize defines time duration per window
+  window: 20s
+  # BucketNum defines bucket number for each window
+  bucket: 500
+  # CPU threshold; valid range 0–1000 (1000 = 100%)
+  cpuThreshold: 850
+
+circuitBreaker:
+  enable: false
+  window: 5s            # Time window size (seconds)
+  bucket: 100            # Number of buckets
+  success: 0.6          # Success rate threshold (0.6 means 60%)
+  request: 500 # Request threshold; circuit breaker evaluation occurs when reached

config/openim-rpc-group.yml

@@ -19,3 +19,20 @@ prometheus:
 
 
 enableHistoryForNewMembers: true
+
+ratelimiter:
+  # Whether to enable rate limiting
+  enable: false
+  # WindowSize defines time duration per window
+  window: 20s
+  # BucketNum defines bucket number for each window
+  bucket: 500
+  # CPU threshold; valid range 0–1000 (1000 = 100%)
+  cpuThreshold: 850
+
+circuitBreaker:
+  enable: false
+  window: 5s            # Time window size (seconds)
+  bucket: 100            # Number of buckets
+  success: 0.6          # Success rate threshold (0.6 means 60%)
+  request: 500 # Request threshold; circuit breaker evaluation occurs when reached

config/openim-rpc-msg.yml

@@ -1,6 +1,6 @@
 rpc:
   # The IP address where this RPC service registers itself; if left blank, it defaults to the internal network IP
-  registerIP: 
+  registerIP:
   # IP address that the RPC service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
   listenIP: 0.0.0.0
   # autoSetPorts indicates whether to automatically set the ports
@@ -20,3 +20,20 @@ prometheus:
 
 # Does sending messages require friend verification
 friendVerify: false
+
+ratelimiter:
+  # Whether to enable rate limiting
+  enable: false
+  # WindowSize defines time duration per window
+  window: 20s
+  # BucketNum defines bucket number for each window
+  bucket: 500
+  # CPU threshold; valid range 0–1000 (1000 = 100%)
+  cpuThreshold: 850
+
+circuitBreaker:
+  enable: false
+  window: 5s            # Time window size (seconds)
+  bucket: 100            # Number of buckets
+  success: 0.6          # Success rate threshold (0.6 means 60%)
+  request: 500 # Request threshold; circuit breaker evaluation occurs when reached

config/openim-rpc-third.yml

@@ -17,6 +17,22 @@ prometheus:
   # It will only take effect when autoSetPorts is set to false.
   ports:
 
+ratelimiter:
+  # Whether to enable rate limiting
+  enable: false
+  # WindowSize defines time duration per window
+  window: 20s
+  # BucketNum defines bucket number for each window
+  bucket: 500
+  # CPU threshold; valid range 0–1000 (1000 = 100%)
+  cpuThreshold: 850
+
+circuitBreaker:
+  enable: false
+  window: 5s            # Time window size (seconds)
+  bucket: 100            # Number of buckets
+  success: 0.6          # Success rate threshold (0.6 means 60%)
+  request: 500 # Request threshold; circuit breaker evaluation occurs when reached
 
 object:
   # Use MinIO as object storage, or set to "cos", "oss", "kodo", "aws", while also configuring the corresponding settings

config/openim-rpc-user.yml

@@ -16,3 +16,20 @@ prometheus:
   # Prometheus listening ports, must be consistent with the number of rpc.ports
   # It will only take effect when autoSetPorts is set to false.
   ports:
+
+ratelimiter:
+  # Whether to enable rate limiting
+  enable: false
+  # WindowSize defines time duration per window
+  window: 20s
+  # BucketNum defines bucket number for each window
+  bucket: 500
+  # CPU threshold; valid range 0–1000 (1000 = 100%)
+  cpuThreshold: 850
+
+circuitBreaker:
+  enable: false
+  window: 5s            # Time window size (seconds)
+  bucket: 100            # Number of buckets
+  success: 0.6          # Success rate threshold (0.6 means 60%)
+  request: 500 # Request threshold; circuit breaker evaluation occurs when reached

config/webhooks.yml

@@ -41,6 +41,9 @@ afterSendGroupMsg:
   attentionIds: []
   # See beforeSendSingleMsg comment.
   deniedTypes: []
+afterMsgSaveDB:
+  enable: false
+  timeout: 5
 afterUserOnline:
   enable: false
   timeout: 5

docker-compose.yml

@@ -213,7 +213,6 @@ services:
       # Defines which listener is used for inter-broker communication within the Kafka cluster
       KAFKA_CFG_INTER_BROKER_LISTENER_NAME: "INTERNAL"
 
-
       # Authentication configuration variables - comment out to disable auth
       # KAFKA_USERNAME: "openIM"
       # KAFKA_PASSWORD: "openIM123"
@@ -267,14 +266,14 @@ services:
     networks:
       - openim
 
-  openim-admin-front:
-    image: ${OPENIM_ADMIN_FRONT_IMAGE}
-    container_name: openim-admin-front
-    restart: always
-    ports:
-      - "11002:80"
-    networks:
-      - openim
+  # openim-admin-front:
+  #   image: ${OPENIM_ADMIN_FRONT_IMAGE}
+  #   container_name: openim-admin-front
+  #   restart: always
+  #   ports:
+  #     - "11002:80"
+  #   networks:
+  #     - openim
 
   prometheus:
     image: ${PROMETHEUS_IMAGE}

go.mod

@@ -1,6 +1,6 @@
 module github.com/openimsdk/open-im-server/v3
 
-go 1.22.7
+go 1.25.0
 
 require (
 	firebase.google.com/go/v4 v4.14.1
@@ -12,21 +12,22 @@ require (
 	github.com/gorilla/websocket v1.5.1
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
 	github.com/mitchellh/mapstructure v1.5.0
-	github.com/openimsdk/protocol v0.0.73-alpha.14
-	github.com/openimsdk/tools v0.0.50-alpha.97
+	github.com/openimsdk/protocol v0.0.73-alpha.19
+	github.com/openimsdk/tools v0.0.50-alpha.113
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0
-	github.com/stretchr/testify v1.9.0
+	github.com/stretchr/testify v1.11.1
 	go.mongodb.org/mongo-driver v1.14.0
 	google.golang.org/api v0.170.0
-	google.golang.org/grpc v1.68.0
-	google.golang.org/protobuf v1.35.1
+	google.golang.org/grpc v1.71.0
+	google.golang.org/protobuf v1.36.4
 	gopkg.in/yaml.v3 v3.0.1
 )
 
 require github.com/google/uuid v1.6.0
 
 require (
+	github.com/IBM/sarama v1.43.0
 	github.com/fatih/color v1.14.1
 	github.com/gin-contrib/gzip v1.0.1
 	github.com/go-redis/redis v6.15.9+incompatible
@@ -34,14 +35,14 @@ require (
 	github.com/hashicorp/golang-lru/v2 v2.0.7
 	github.com/kelindar/bitmap v1.5.2
 	github.com/likexian/gokit v0.25.13
-	github.com/openimsdk/gomake v0.0.15-alpha.11
+	github.com/openimsdk/gomake v0.0.17
 	github.com/redis/go-redis/v9 v9.4.0
 	github.com/robfig/cron/v3 v3.0.1
 	github.com/shirou/gopsutil v3.21.11+incompatible
 	github.com/spf13/viper v1.18.2
 	go.etcd.io/etcd/client/v3 v3.5.13
 	go.uber.org/automaxprocs v1.5.3
-	golang.org/x/sync v0.8.0
+	golang.org/x/sync v0.10.0
 	k8s.io/api v0.31.2
 	k8s.io/apimachinery v0.31.2
 	k8s.io/client-go v0.31.2
@@ -49,12 +50,11 @@ require (
 
 require (
 	cloud.google.com/go v0.112.1 // indirect
-	cloud.google.com/go/compute/metadata v0.5.0 // indirect
+	cloud.google.com/go/compute/metadata v0.6.0 // indirect
 	cloud.google.com/go/firestore v1.15.0 // indirect
 	cloud.google.com/go/iam v1.1.7 // indirect
 	cloud.google.com/go/longrunning v0.5.5 // indirect
 	cloud.google.com/go/storage v1.40.0 // indirect
-	github.com/IBM/sarama v1.43.0 // indirect
 	github.com/MicahParks/keyfunc v1.9.0 // indirect
 	github.com/aliyun/aliyun-oss-go-sdk v3.0.2+incompatible // indirect
 	github.com/aws/aws-sdk-go-v2 v1.32.5 // indirect
@@ -76,6 +76,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/sts v1.33.1 // indirect
 	github.com/aws/smithy-go v1.22.1 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/bmatcuk/doublestar/v4 v4.10.0 // indirect
 	github.com/bytedance/sonic v1.11.6 // indirect
 	github.com/bytedance/sonic/loader v0.1.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
@@ -90,15 +91,16 @@ require (
 	github.com/eapache/go-resiliency v1.6.0 // indirect
 	github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect
 	github.com/eapache/queue v1.1.0 // indirect
+	github.com/ebitengine/purego v0.10.0 // indirect
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
-	github.com/fsnotify/fsnotify v1.7.0 // indirect
+	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/fxamacker/cbor/v2 v2.7.0 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.3 // indirect
 	github.com/gin-contrib/sse v0.1.0 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
-	github.com/go-ole/go-ole v1.2.6 // indirect
+	github.com/go-ole/go-ole v1.3.0 // indirect
 	github.com/go-openapi/jsonpointer v0.19.6 // indirect
 	github.com/go-openapi/jsonreference v0.20.2 // indirect
 	github.com/go-openapi/swag v0.22.4 // indirect
@@ -108,7 +110,7 @@ require (
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/gnostic-models v0.6.8 // indirect
-	github.com/google/go-cmp v0.6.0 // indirect
+	github.com/google/go-cmp v0.7.0 // indirect
 	github.com/google/go-querystring v1.1.0 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/google/s2a-go v0.1.7 // indirect
@@ -135,6 +137,7 @@ require (
 	github.com/leodido/go-urn v1.4.0 // indirect
 	github.com/lestrrat-go/strftime v1.0.6 // indirect
 	github.com/lithammer/shortuuid v3.0.0+incompatible // indirect
+	github.com/lufia/plan9stats v0.0.0-20260216142805-b3301c5f2a88 // indirect
 	github.com/magefile/mage v1.15.0 // indirect
 	github.com/magiconair/properties v1.8.7 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
@@ -151,6 +154,7 @@ require (
 	github.com/pelletier/go-toml/v2 v2.2.2 // indirect
 	github.com/pierrec/lz4/v4 v4.1.21 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
+	github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
 	github.com/prometheus/client_model v0.5.0 // indirect
 	github.com/prometheus/common v0.45.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
@@ -159,6 +163,10 @@ require (
 	github.com/rs/xid v1.5.0 // indirect
 	github.com/sagikazarmark/locafero v0.4.0 // indirect
 	github.com/sagikazarmark/slog-shim v0.1.0 // indirect
+	github.com/sercand/kuberesolver/v6 v6.0.1 // indirect
+	github.com/shirou/gopsutil/v3 v3.24.5 // indirect
+	github.com/shirou/gopsutil/v4 v4.26.2 // indirect
+	github.com/shoenig/go-m1cpu v0.1.6 // indirect
 	github.com/sourcegraph/conc v0.3.0 // indirect
 	github.com/spf13/afero v1.11.0 // indirect
 	github.com/spf13/cast v1.6.0 // indirect
@@ -166,8 +174,8 @@ require (
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/tencentyun/cos-go-sdk-v5 v0.7.47 // indirect
-	github.com/tklauser/go-sysconf v0.3.13 // indirect
-	github.com/tklauser/numcpus v0.7.0 // indirect
+	github.com/tklauser/go-sysconf v0.3.16 // indirect
+	github.com/tklauser/numcpus v0.11.0 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
@@ -178,26 +186,27 @@ require (
 	go.etcd.io/etcd/api/v3 v3.5.13 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.13 // indirect
 	go.opencensus.io v0.24.0 // indirect
+	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.49.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0 // indirect
-	go.opentelemetry.io/otel v1.24.0 // indirect
-	go.opentelemetry.io/otel/metric v1.24.0 // indirect
-	go.opentelemetry.io/otel/trace v1.24.0 // indirect
+	go.opentelemetry.io/otel v1.34.0 // indirect
+	go.opentelemetry.io/otel/metric v1.34.0 // indirect
+	go.opentelemetry.io/otel/trace v1.34.0 // indirect
 	go.uber.org/atomic v1.9.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/arch v0.7.0 // indirect
 	golang.org/x/exp v0.0.0-20230905200255-921286631fa9 // indirect
 	golang.org/x/image v0.15.0 // indirect
-	golang.org/x/net v0.29.0 // indirect
-	golang.org/x/oauth2 v0.23.0 // indirect
-	golang.org/x/sys v0.25.0 // indirect
-	golang.org/x/term v0.24.0 // indirect
-	golang.org/x/text v0.18.0 // indirect
+	golang.org/x/net v0.34.0 // indirect
+	golang.org/x/oauth2 v0.25.0 // indirect
+	golang.org/x/sys v0.42.0 // indirect
+	golang.org/x/term v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
 	golang.org/x/time v0.5.0 // indirect
 	google.golang.org/appengine/v2 v2.0.2 // indirect
 	google.golang.org/genproto v0.0.0-20240213162025-012b6fc9bca9 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20240903143218-8af14fe29dc1 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20250106144421-5f5ef82da422 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250115164207-1a7da9e5054f // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gorm.io/gorm v1.25.8 // indirect
@@ -216,6 +225,6 @@ require (
 	github.com/spf13/cobra v1.8.0
 	github.com/ugorji/go/codec v1.2.12 // indirect
 	go.uber.org/zap v1.24.0 // indirect
-	golang.org/x/crypto v0.27.0 // indirect
+	golang.org/x/crypto v0.32.0 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 )

go.sum

@@ -1,8 +1,8 @@
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cloud.google.com/go v0.112.1 h1:uJSeirPke5UNZHIb4SxfZklVSiWWVqW4oXlETwZziwM=
 cloud.google.com/go v0.112.1/go.mod h1:+Vbu+Y1UU+I1rjmzeMOb/8RfkKJK2Gyxi1X6jJCZLo4=
-cloud.google.com/go/compute/metadata v0.5.0 h1:Zr0eK8JbFv6+Wi4ilXAR8FJ3wyNdpxHKJNPos6LTZOY=
-cloud.google.com/go/compute/metadata v0.5.0/go.mod h1:aHnloV2TPI38yx4s9+wAZhHykWvVCfu7hQbF+9CWoiY=
+cloud.google.com/go/compute/metadata v0.6.0 h1:A6hENjEsCDtC1k8byVsgwvVcioamEHvZ4j01OwKxG9I=
+cloud.google.com/go/compute/metadata v0.6.0/go.mod h1:FjyFAW1MW0C203CEOMDTu3Dk1FlqW3Rga40jzHL4hfg=
 cloud.google.com/go/firestore v1.15.0 h1:/k8ppuWOtNuDHt2tsRV42yI21uaGnKDEQnRFeBpbFF8=
 cloud.google.com/go/firestore v1.15.0/go.mod h1:GWOxFXcv8GZUtYpWHw/w6IuYNux/BtmeVTMmjrm4yhk=
 cloud.google.com/go/iam v1.1.7 h1:z4VHOhwKLF/+UYXAJDFwGtNF0b6gjsW1Pk9Ml0U/IoM=
@@ -61,6 +61,8 @@ github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLj
 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
+github.com/bmatcuk/doublestar/v4 v4.10.0 h1:zU9WiOla1YA122oLM6i4EXvGW62DvKZVxIe6TYWexEs=
+github.com/bmatcuk/doublestar/v4 v4.10.0/go.mod h1:xBQ8jztBU6kakFMg+8WGxn0c6z1fTSPVIjEY1Wr7jzc=
 github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
 github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
 github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
@@ -103,6 +105,8 @@ github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 h1:Oy0F4A
 github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3/go.mod h1:YvSRo5mw33fLEx1+DlK6L2VV43tJt5Eyel9n9XBcR+0=
 github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc=
 github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
+github.com/ebitengine/purego v0.10.0 h1:QIw4xfpWT6GWTzaW5XEKy3HXoqrJGx1ijYHzTF0/ISU=
+github.com/ebitengine/purego v0.10.0/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ=
 github.com/emicklei/go-restful/v3 v3.11.0 h1:rAQeMHw1c7zTmncogyy8VvRZwtkmkZ4FxERmMY4rD+g=
 github.com/emicklei/go-restful/v3 v3.11.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
@@ -117,8 +121,8 @@ github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8
 github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
 github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
 github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
-github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA=
-github.com/fsnotify/fsnotify v1.7.0/go.mod h1:40Bi/Hjc2AVfZrqy+aj+yEI+/bRxZnMJyTJwOpGvigM=
+github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k=
+github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0=
 github.com/fxamacker/cbor/v2 v2.7.0 h1:iM5WgngdRBanHcxugY4JySA0nk1wZorNOpTgCMedv5E=
 github.com/fxamacker/cbor/v2 v2.7.0/go.mod h1:pxXPTn3joSm21Gbwsv0w9OSA2y1HFR9qXEeXQVeNoDQ=
 github.com/gabriel-vasile/mimetype v1.4.3 h1:in2uUcidCuFcDKtdcBxlR0rJ1+fsokWf+uqxgUFjbI0=
@@ -134,8 +138,9 @@ github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
 github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
-github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY=
 github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
+github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE=
+github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78=
 github.com/go-openapi/jsonpointer v0.19.6 h1:eCs3fxoIi3Wh6vtgmLTOjdhSpiqphQ+DaPn38N2ZdrE=
 github.com/go-openapi/jsonpointer v0.19.6/go.mod h1:osyAmYz/mB/C3I+WsTTSgw1ONzaLJoLCyoi6/zppojs=
 github.com/go-openapi/jsonreference v0.20.2 h1:3sVjiK66+uXK/6oQ8xgcRKcFgQ5KXa2KvnJRumpMGbE=
@@ -202,8 +207,8 @@ github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.3/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
-github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
-github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/google/go-querystring v1.0.0/go.mod h1:odCYkC5MyYFN7vkCjXpyrEuKhc/BUO6wN/zVPAxq5ck=
 github.com/google/go-querystring v1.1.0 h1:AnCroh3fv4ZBgVIf1Iwtovgjaw/GiKJo8M8yD/fhyJ8=
 github.com/google/go-querystring v1.1.0/go.mod h1:Kcdr2DB4koayq7X8pmAG4sNG59So17icRSOU623lUBU=
@@ -303,6 +308,8 @@ github.com/likexian/gokit v0.25.13 h1:p2Uw3+6fGG53CwdU2Dz0T6bOycdb2+bAFAa3ymwWVk
 github.com/likexian/gokit v0.25.13/go.mod h1:qQhEWFBEfqLCO3/vOEo2EDKd+EycekVtUK4tex+l2H4=
 github.com/lithammer/shortuuid v3.0.0+incompatible h1:NcD0xWW/MZYXEHa6ITy6kaXN5nwm/V115vj2YXfhS0w=
 github.com/lithammer/shortuuid v3.0.0+incompatible/go.mod h1:FR74pbAuElzOUuenUHTK2Tciko1/vKuIKS9dSkDrA4w=
+github.com/lufia/plan9stats v0.0.0-20260216142805-b3301c5f2a88 h1:PTw+yKnXcOFCR6+8hHTyWBeQ/P4Nb7dd4/0ohEcWQuM=
+github.com/lufia/plan9stats v0.0.0-20260216142805-b3301c5f2a88/go.mod h1:autxFIvghDt3jPTLoqZ9OZ7s9qTGNAWmYCjVFWPX/zg=
 github.com/magefile/mage v1.15.0 h1:BvGheCMAsG3bWUDbZ8AyXXpCNwU9u5CB6sM+HNb9HYg=
 github.com/magefile/mage v1.15.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A=
 github.com/magiconair/properties v1.8.7 h1:IeQXZAiQcpL9mgcAe1Nu6cX9LLw6ExEHKjN0VQdvPDY=
@@ -345,12 +352,12 @@ github.com/onsi/ginkgo/v2 v2.19.0 h1:9Cnnf7UHo57Hy3k6/m5k3dRfGTMXGvxhHFvkDTCTpvA
 github.com/onsi/ginkgo/v2 v2.19.0/go.mod h1:rlwLi9PilAFJ8jCg9UE1QP6VBpd6/xj3SRC0d6TU0To=
 github.com/onsi/gomega v1.25.0 h1:Vw7br2PCDYijJHSfBOWhov+8cAnUf8MfMaIOV323l6Y=
 github.com/onsi/gomega v1.25.0/go.mod h1:r+zV744Re+DiYCIPRlYOTxn0YkOLcAnW8k1xXdMPGhM=
-github.com/openimsdk/gomake v0.0.15-alpha.11 h1:PQudYDRESYeYlUYrrLLJhYIlUPO5x7FAx+o5El9U/Bw=
-github.com/openimsdk/gomake v0.0.15-alpha.11/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
-github.com/openimsdk/protocol v0.0.73-alpha.14 h1:lv9wNiPRm6G7q74TfpMobKrSfeTaBlZ+Ps3O6UFPmaE=
-github.com/openimsdk/protocol v0.0.73-alpha.14/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw=
-github.com/openimsdk/tools v0.0.50-alpha.97 h1:6ik5w3PpgDG6VjSo3nb3FT/fxN3JX7iIARVxVu9g7VY=
-github.com/openimsdk/tools v0.0.50-alpha.97/go.mod h1:n2poR3asX1e1XZce4O+MOWAp+X02QJRFvhcLCXZdzRo=
+github.com/openimsdk/gomake v0.0.17 h1:q8haP48VOH45WhJRiLj1YSBJyUFJqD8CTedH65i1YH8=
+github.com/openimsdk/gomake v0.0.17/go.mod h1:nnjS8yCtrPJAt1knMbyPiUwCH2gpyBzj/EZAONfUOXg=
+github.com/openimsdk/protocol v0.0.73-alpha.19 h1:CvXoDF2U73UcMhLnrtMFks2Aw+bXiDgH8AITEt783/s=
+github.com/openimsdk/protocol v0.0.73-alpha.19/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw=
+github.com/openimsdk/tools v0.0.50-alpha.113 h1:rhLWaSJuhjgJFNVzmpChLCG7dPXS0+bte+CPI0008Us=
+github.com/openimsdk/tools v0.0.50-alpha.113/go.mod h1:x9i/e+WJFW4tocy6RNJQ9NofQiP3KJ1Y576/06TqOG4=
 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=
@@ -361,6 +368,8 @@ github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 h1:o4JXh1EVt9k/+g42oCprj/FisM4qX9L3sZB3upGN2ZU=
+github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE=
 github.com/prashantv/gostub v1.1.0 h1:BTyx3RfQjRHnUWaGF9oQos79AlQ5k8WNktv7VGvVH4g=
 github.com/prashantv/gostub v1.1.0/go.mod h1:A5zLQHz7ieHGG7is6LLXLz7I8+3LZzsrV0P1IAHhP5U=
 github.com/prometheus/client_golang v1.18.0 h1:HzFfmkOzH5Q8L8G+kSJKUx5dtG87sewO+FoDDqP5Tbk=
@@ -384,8 +393,8 @@ github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
 github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
 github.com/rogpeppe/go-internal v1.8.0/go.mod h1:WmiCO8CzOY8rg0OYDC4/i/2WRWAB6poM+XZ2dLUbcbE=
-github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
-github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
+github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII=
+github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o=
 github.com/rs/xid v1.5.0 h1:mKX4bl4iPYJtEIxp6CYiUuLQ/8DYMoz0PUdtGgMFRVc=
 github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
@@ -393,8 +402,18 @@ github.com/sagikazarmark/locafero v0.4.0 h1:HApY1R9zGo4DBgr7dqsTH/JJxLTTsOt7u6ke
 github.com/sagikazarmark/locafero v0.4.0/go.mod h1:Pe1W6UlPYUk/+wc/6KFhbORCfqzgYEpgQ3O5fPuL3H4=
 github.com/sagikazarmark/slog-shim v0.1.0 h1:diDBnUNK9N/354PgrxMywXnAwEr1QZcOr6gto+ugjYE=
 github.com/sagikazarmark/slog-shim v0.1.0/go.mod h1:SrcSrq8aKtyuqEI1uvTDTK1arOWRIczQRv+GVI1AkeQ=
+github.com/sercand/kuberesolver/v6 v6.0.1 h1:XZUTA0gy/lgDYp/UhEwv7Js24F1j8NJ833QrWv0Xux4=
+github.com/sercand/kuberesolver/v6 v6.0.1/go.mod h1:C0tsTuRMONSY+Xf7pv7RMW1/JlewY1+wS8SZE+1lf1s=
 github.com/shirou/gopsutil v3.21.11+incompatible h1:+1+c1VGhc88SSonWP6foOcLhvnKlUeu/erjjvaPEYiI=
 github.com/shirou/gopsutil v3.21.11+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=
+github.com/shirou/gopsutil/v3 v3.24.5 h1:i0t8kL+kQTvpAYToeuiVk3TgDeKOFioZO3Ztz/iZ9pI=
+github.com/shirou/gopsutil/v3 v3.24.5/go.mod h1:bsoOS1aStSs9ErQ1WWfxllSeS1K5D+U30r2NfcubMVk=
+github.com/shirou/gopsutil/v4 v4.26.2 h1:X8i6sicvUFih4BmYIGT1m2wwgw2VG9YgrDTi7cIRGUI=
+github.com/shirou/gopsutil/v4 v4.26.2/go.mod h1:LZ6ewCSkBqUpvSOf+LsTGnRinC6iaNUNMGBtDkJBaLQ=
+github.com/shoenig/go-m1cpu v0.1.6 h1:nxdKQNcEB6vzgA2E2bvzKIYRuNj7XNJ4S/aRSwKzFtM=
+github.com/shoenig/go-m1cpu v0.1.6/go.mod h1:1JJMcUBvfNwpq05QDQVAnx3gUHr9IYF7GNg9SUEw2VQ=
+github.com/shoenig/test v0.6.4 h1:kVTaSd7WLz5WZ2IaoM0RSzRsUD+m8wRR+5qvntpn4LU=
+github.com/shoenig/test v0.6.4/go.mod h1:byHiCGXqrVaflBLAMq/srcZIHynQPQgeyvkvXnjqq0k=
 github.com/sourcegraph/conc v0.3.0 h1:OQTbbt6P72L20UqAkXXuLOj79LfEanQ+YQFNpLA9ySo=
 github.com/sourcegraph/conc v0.3.0/go.mod h1:Sdozi7LEKbFPqYX2/J+iBAM6HpqSLTASQIKqDmF7Mt0=
 github.com/spf13/afero v1.11.0 h1:WJQKhtpdm3v2IzqG8VMqrr6Rf3UYpEF239Jy9wNepM8=
@@ -420,18 +439,19 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
-github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8=
 github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU=
 github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/common v1.0.563/go.mod h1:7sCQWVkxcsR38nffDW057DRGk8mUjK1Ing/EFOK8s8Y=
 github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/kms v1.0.563/go.mod h1:uom4Nvi9W+Qkom0exYiJ9VWJjXwyxtPYTkKkaLMlfE0=
 github.com/tencentyun/cos-go-sdk-v5 v0.7.47 h1:uoS4Sob16qEYoapkqJq1D1Vnsy9ira9BfNUMtoFYTI4=
 github.com/tencentyun/cos-go-sdk-v5 v0.7.47/go.mod h1:DH9US8nB+AJXqwu/AMOrCFN1COv3dpytXuJWHgdg7kE=
-github.com/tklauser/go-sysconf v0.3.13 h1:GBUpcahXSpR2xN01jhkNAbTLRk2Yzgggk8IM08lq3r4=
-github.com/tklauser/go-sysconf v0.3.13/go.mod h1:zwleP4Q4OehZHGn4CYZDipCgg9usW5IJePewFCGVEa0=
-github.com/tklauser/numcpus v0.7.0 h1:yjuerZP127QG9m5Zh/mSO4wqurYil27tHrqwRoRjpr4=
-github.com/tklauser/numcpus v0.7.0/go.mod h1:bb6dMVcj8A42tSE7i32fsIUCbQNllK5iDguyOZRUzAY=
+github.com/tklauser/go-sysconf v0.3.16 h1:frioLaCQSsF5Cy1jgRBrzr6t502KIIwQ0MArYICU0nA=
+github.com/tklauser/go-sysconf v0.3.16/go.mod h1:/qNL9xxDhc7tx3HSRsLWNnuzbVfh3e7gh/BmM179nYI=
+github.com/tklauser/numcpus v0.11.0 h1:nSTwhKH5e1dMNsCdVBukSZrURJRoHbSEQjdEbY+9RXw=
+github.com/tklauser/numcpus v0.11.0/go.mod h1:z+LwcLq54uWZTX0u/bGobaV34u6V7KNlTZejzM6/3MQ=
 github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI=
 github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
 github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE=
@@ -461,18 +481,22 @@ go.mongodb.org/mongo-driver v1.14.0 h1:P98w8egYRjYe3XDjxhYJagTokP/H6HzlsnojRgZRd
 go.mongodb.org/mongo-driver v1.14.0/go.mod h1:Vzb0Mk/pa7e6cWw85R4F/endUC3u0U9jGcNU603k65c=
 go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0=
 go.opencensus.io v0.24.0/go.mod h1:vNK8G9p7aAivkbmorf4v+7Hgx+Zs0yY+0fOtgBfjQKo=
+go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
+go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.49.0 h1:4Pp6oUg3+e/6M4C0A/3kJ2VYa++dsWVTtGgLVj5xtHg=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.49.0/go.mod h1:Mjt1i1INqiaoZOMGR1RIUJN+i3ChKoFRqzrRQhlkbs0=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0 h1:jq9TW8u3so/bN+JPT166wjOI6/vQPF6Xe7nMNIltagk=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0/go.mod h1:p8pYQP+m5XfbZm9fxtSKAbM6oIllS7s2AfxrChvc7iw=
-go.opentelemetry.io/otel v1.24.0 h1:0LAOdjNmQeSTzGBzduGe/rU4tZhMwL5rWgtp9Ku5Jfo=
-go.opentelemetry.io/otel v1.24.0/go.mod h1:W7b9Ozg4nkF5tWI5zsXkaKKDjdVjpD4oAt9Qi/MArHo=
-go.opentelemetry.io/otel/metric v1.24.0 h1:6EhoGWWK28x1fbpA4tYTOWBkPefTDQnb8WSGXlc88kI=
-go.opentelemetry.io/otel/metric v1.24.0/go.mod h1:VYhLe1rFfxuTXLgj4CBiyz+9WYBA8pNGJgDcSFRKBco=
-go.opentelemetry.io/otel/sdk v1.22.0 h1:6coWHw9xw7EfClIC/+O31R8IY3/+EiRFHevmHafB2Gw=
-go.opentelemetry.io/otel/sdk v1.22.0/go.mod h1:iu7luyVGYovrRpe2fmj3CVKouQNdTOkxtLzPvPz1DOc=
-go.opentelemetry.io/otel/trace v1.24.0 h1:CsKnnL4dUAr/0llH9FKuc698G04IrpWV0MQA/Y1YELI=
-go.opentelemetry.io/otel/trace v1.24.0/go.mod h1:HPc3Xr/cOApsBI154IU0OI0HJexz+aw5uPdbs3UCjNU=
+go.opentelemetry.io/otel v1.34.0 h1:zRLXxLCgL1WyKsPVrgbSdMN4c0FMkDAskSTQP+0hdUY=
+go.opentelemetry.io/otel v1.34.0/go.mod h1:OWFPOQ+h4G8xpyjgqo4SxJYdDQ/qmRH+wivy7zzx9oI=
+go.opentelemetry.io/otel/metric v1.34.0 h1:+eTR3U0MyfWjRDhmFMxe2SsW64QrZ84AOhvqS7Y+PoQ=
+go.opentelemetry.io/otel/metric v1.34.0/go.mod h1:CEDrp0fy2D0MvkXE+dPV7cMi8tWZwX3dmaIhwPOaqHE=
+go.opentelemetry.io/otel/sdk v1.34.0 h1:95zS4k/2GOy069d321O8jWgYsW3MzVV+KuSPKp7Wr1A=
+go.opentelemetry.io/otel/sdk v1.34.0/go.mod h1:0e/pNiaMAqaykJGKbi+tSjWfNNHMTxoC9qANsCzbyxU=
+go.opentelemetry.io/otel/sdk/metric v1.34.0 h1:5CeK9ujjbFVL5c1PhLuStg1wxA7vQv7ce1EK0Gyvahk=
+go.opentelemetry.io/otel/sdk/metric v1.34.0/go.mod h1:jQ/r8Ze28zRKoNRdkjCZxfs6YvBTG1+YIqyFVFYec5w=
+go.opentelemetry.io/otel/trace v1.34.0 h1:+ouXS2V8Rd4hp4580a8q23bg0azF2nI8cqLYnC8mh/k=
+go.opentelemetry.io/otel/trace v1.34.0/go.mod h1:Svm7lSjQD7kG7KJ/MUHPVXSDGz2OX4h0M2jHBhmSfRE=
 go.uber.org/atomic v1.9.0 h1:ECmE8Bn/WFTYwEW/bpKD3M8VtR/zQVbavAoalC1PYyE=
 go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
 go.uber.org/automaxprocs v1.5.3 h1:kWazyxZUrS3Gs4qUpbwo5kEIMGe/DAvi5Z4tl2NW4j8=
@@ -493,8 +517,8 @@ golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5y
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.1.0/go.mod h1:RecgLatLF4+eUMCP1PoPZQb+cVrJcOPbHkTkbkB9sbw=
 golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58=
-golang.org/x/crypto v0.27.0 h1:GXm2NjJrPaiv/h1tb2UH8QfgC/hOf/+z0p6PT8o1w7A=
-golang.org/x/crypto v0.27.0/go.mod h1:1Xngt8kV6Dvbssa53Ziq6Eqn0HqbZi5Z6R0ZpwQzt70=
+golang.org/x/crypto v0.32.0 h1:euUpcYgM8WcP71gNpTqQCn6rC2t6ULUPiOzfWaXVVfc=
+golang.org/x/crypto v0.32.0/go.mod h1:ZnnJkOaASj8g0AjIduWNlq2NRxL0PlBrbKVyZ6V/Ugc=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20230905200255-921286631fa9 h1:GoHiUyI/Tp2nVkLI2mCxVkOjsbSXD66ic0XW0js0R9g=
 golang.org/x/exp v0.0.0-20230905200255-921286631fa9/go.mod h1:S2oDrQGGwySpoQPVqRShND87VCbxmc6bL1Yd2oYrm6k=
@@ -522,25 +546,26 @@ golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug
 golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco=
 golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
-golang.org/x/net v0.29.0 h1:5ORfpBpCs4HzDYoodCDBbwHzdR5UrLBZ3sOnUJmFoHo=
-golang.org/x/net v0.29.0/go.mod h1:gLkgy8jTGERgjzMic6DS9+SP0ajcu6Xu3Orq/SpETg0=
+golang.org/x/net v0.34.0 h1:Mb7Mrk043xzHgnRM88suvJFwzVrRfHEHJEl5/71CKw0=
+golang.org/x/net v0.34.0/go.mod h1:di0qlW3YNM5oh6GqDGQr92MyTozJPmybPK4Ev/Gm31k=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
-golang.org/x/oauth2 v0.23.0 h1:PbgcYx2W7i4LvjJWEbf0ngHV6qJYr86PkAV3bXdLEbs=
-golang.org/x/oauth2 v0.23.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
+golang.org/x/oauth2 v0.25.0 h1:CY4y7XT9v0cRI9oupztF8AgiIu99L/ksR/Xp/6jrZ70=
+golang.org/x/oauth2 v0.25.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.8.0 h1:3NFvSEYkUoMifnESzZl15y791HH1qU2xm6eCJU5ZPXQ=
-golang.org/x/sync v0.8.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
+golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -549,14 +574,14 @@ golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/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.25.0 h1:r+8e+loiHxRqhXVl6ML1nO3l1+oFoWbnlu2Ehimmi34=
-golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
+golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 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=
 golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
-golang.org/x/term v0.24.0 h1:Mh5cbb+Zk2hqqXNO7S1iTjEphVL+jb8ZWaqh/g+JWkM=
-golang.org/x/term v0.24.0/go.mod h1:lOBK/LVxemqiMij05LGJ0tzNr8xlmwBRJ81PX6wVLH8=
+golang.org/x/term v0.28.0 h1:/Ts8HFuMR2E6IP/jlo7QVLZHggjKQbhu/7H0LJFr3Gg=
+golang.org/x/term v0.28.0/go.mod h1:Sw/lC2IAUZ92udQNf3WodGtn4k/XoLyZoh8v/8uiwek=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
@@ -565,8 +590,8 @@ golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
 golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
-golang.org/x/text v0.18.0 h1:XvMDiNzPAl0jr17s6W9lcaIhGUfUORdGCNsuLmPG224=
-golang.org/x/text v0.18.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
+golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
+golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
 golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
 golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
@@ -597,17 +622,17 @@ google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98
 google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo=
 google.golang.org/genproto v0.0.0-20240213162025-012b6fc9bca9 h1:9+tzLLstTlPTRyJTh+ah5wIMsBW5c4tQwGTN3thOW9Y=
 google.golang.org/genproto v0.0.0-20240213162025-012b6fc9bca9/go.mod h1:mqHbVIp48Muh7Ywss/AD6I5kNVKZMmAa/QEW58Gxp2s=
-google.golang.org/genproto/googleapis/api v0.0.0-20240903143218-8af14fe29dc1 h1:hjSy6tcFQZ171igDaN5QHOw2n6vx40juYbC/x67CEhc=
-google.golang.org/genproto/googleapis/api v0.0.0-20240903143218-8af14fe29dc1/go.mod h1:qpvKtACPCQhAdu3PyQgV4l3LMXZEtft7y8QcarRsp9I=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 h1:pPJltXNxVzT4pK9yD8vR9X75DaWYYmLGMsEvBfFQZzQ=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1/go.mod h1:UqMtugtsSgubUsoxbuAoiCXvqvErP7Gf0so0mK9tHxU=
+google.golang.org/genproto/googleapis/api v0.0.0-20250106144421-5f5ef82da422 h1:GVIKPyP/kLIyVOgOnTwFOrvQaQUzOzGMCxgFUOEmm24=
+google.golang.org/genproto/googleapis/api v0.0.0-20250106144421-5f5ef82da422/go.mod h1:b6h1vNKhxaSoEI+5jc3PJUCustfli/mRab7295pY7rw=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250115164207-1a7da9e5054f h1:OxYkA3wjPsZyBylwymxSHa7ViiW1Sml4ToBrncvFehI=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250115164207-1a7da9e5054f/go.mod h1:+2Yz8+CLJbIfL9z73EW45avw8Lmge3xVElCP9zEKi50=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg=
 google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY=
 google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
 google.golang.org/grpc v1.33.2/go.mod h1:JMHMWHQWaTccqQQlmk3MJZS+GWXOdAesneDmEnv2fbc=
-google.golang.org/grpc v1.68.0 h1:aHQeeJbo8zAkAa3pRzrVjZlbz6uSfeOXlJNQM0RAbz0=
-google.golang.org/grpc v1.68.0/go.mod h1:fmSPC5AsjSBCK54MyHRx48kpOti1/jRfOlwEWywNjWA=
+google.golang.org/grpc v1.71.0 h1:kF77BGdPTQ4/JZWMlb9VpJ5pa25aqvVqogsxNHHdeBg=
+google.golang.org/grpc v1.71.0/go.mod h1:H0GRtasmQOh9LkFoCPDu3ZrwUtD1YGE+b2vYBYd/8Ec=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
 google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
@@ -617,8 +642,8 @@ google.golang.org/protobuf v1.22.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2
 google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
 google.golang.org/protobuf v1.23.1-0.20200526195155-81db48ad09cc/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
 google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c=
-google.golang.org/protobuf v1.35.1 h1:m3LfL6/Ca+fqnjnlqQXNpFPABW1UD7mjh8KO2mKFytA=
-google.golang.org/protobuf v1.35.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
+google.golang.org/protobuf v1.36.4 h1:6A3ZDJHn/eNqc1i+IdefRzy/9PokBTPvcqMySR7NNIM=
+google.golang.org/protobuf v1.36.4/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=

internal/api/conversation.go

@@ -76,3 +76,7 @@ func (o *ConversationApi) GetPinnedConversationIDs(c *gin.Context) {
 func (o *ConversationApi) UpdateConversationsByUser(c *gin.Context) {
 	a2r.Call(c, conversation.ConversationClient.UpdateConversationsByUser, o.Client)
 }
+
+func (o *ConversationApi) DeleteConversations(c *gin.Context) {
+	a2r.Call(c, conversation.ConversationClient.DeleteConversations, o.Client)
+}

internal/api/ratelimit.go

@@ -0,0 +1,83 @@
+package api
+
+import (
+	"fmt"
+	"math"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/gin-gonic/gin"
+
+	"github.com/openimsdk/tools/apiresp"
+	"github.com/openimsdk/tools/errs"
+	"github.com/openimsdk/tools/log"
+	"github.com/openimsdk/tools/stability/ratelimit"
+	"github.com/openimsdk/tools/stability/ratelimit/bbr"
+)
+
+type RateLimiter struct {
+	Enable       bool          `yaml:"enable"`
+	Window       time.Duration `yaml:"window"`       // time duration per window
+	Bucket       int           `yaml:"bucket"`       // bucket number for each window
+	CPUThreshold int64         `yaml:"cpuThreshold"` // CPU threshold; valid range 0–1000 (1000 = 100%)
+}
+
+func RateLimitMiddleware(config *RateLimiter) gin.HandlerFunc {
+	if !config.Enable {
+		return func(c *gin.Context) {
+			c.Next()
+		}
+	}
+
+	limiter := bbr.NewBBRLimiter(
+		bbr.WithWindow(config.Window),
+		bbr.WithBucket(config.Bucket),
+		bbr.WithCPUThreshold(config.CPUThreshold),
+	)
+
+	return func(c *gin.Context) {
+		status := limiter.Stat()
+
+		c.Header("X-BBR-CPU", strconv.FormatInt(status.CPU, 10))
+		c.Header("X-BBR-MinRT", strconv.FormatInt(status.MinRt, 10))
+		c.Header("X-BBR-MaxPass", strconv.FormatInt(status.MaxPass, 10))
+		c.Header("X-BBR-MaxInFlight", strconv.FormatInt(status.MaxInFlight, 10))
+		c.Header("X-BBR-InFlight", strconv.FormatInt(status.InFlight, 10))
+
+		done, err := limiter.Allow()
+		if err != nil {
+
+			c.Header("X-RateLimit-Policy", "BBR")
+			c.Header("Retry-After", calculateBBRRetryAfter(status))
+			c.Header("X-RateLimit-Limit", strconv.FormatInt(status.MaxInFlight, 10))
+			c.Header("X-RateLimit-Remaining", "0") // There is no concept of remaining quota in BBR.
+
+			fmt.Println("rate limited:", err, "path:", c.Request.URL.Path)
+			log.ZWarn(c, "rate limited", err, "path", c.Request.URL.Path)
+			c.AbortWithStatus(http.StatusTooManyRequests)
+			apiresp.GinError(c, errs.NewCodeError(http.StatusTooManyRequests, "too many requests, please try again later"))
+			return
+		}
+
+		c.Next()
+		done(ratelimit.DoneInfo{})
+	}
+}
+
+func calculateBBRRetryAfter(status bbr.Stat) string {
+	loadRatio := float64(status.CPU) / float64(status.CPU)
+
+	if loadRatio < 0.8 {
+		return "1"
+	}
+	if loadRatio < 0.95 {
+		return "2"
+	}
+
+	backoff := 1 + int64(math.Pow(loadRatio-0.95, 2)*50)
+	if backoff > 5 {
+		backoff = 5
+	}
+	return strconv.FormatInt(backoff, 10)
+}

internal/api/router.go

@@ -97,6 +97,18 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, cf
 	case BestSpeed:
 		r.Use(gzip.Gzip(gzip.BestSpeed))
 	}
+
+	// Use rate limiter middleware
+	if cfg.API.RateLimiter.Enable {
+		rl := &RateLimiter{
+			Enable:       cfg.API.RateLimiter.Enable,
+			Window:       cfg.API.RateLimiter.Window,
+			Bucket:       cfg.API.RateLimiter.Bucket,
+			CPUThreshold: cfg.API.RateLimiter.CPUThreshold,
+		}
+		r.Use(RateLimitMiddleware(rl))
+	}
+
 	if config.Standalone() {
 		r.Use(func(c *gin.Context) {
 			c.Set(authverify.CtxAdminUserIDsKey, cfg.Share.IMAdminUser.UserIDs)
@@ -277,6 +289,8 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, cf
 		conversationGroup.POST("/get_owner_conversation", c.GetOwnerConversation)
 		conversationGroup.POST("/get_not_notify_conversation_ids", c.GetNotNotifyConversationIDs)
 		conversationGroup.POST("/get_pinned_conversation_ids", c.GetPinnedConversationIDs)
+		conversationGroup.POST("/delete_conversations", c.DeleteConversations)
+		conversationGroup.POST("/update_conversations_by_user", c.UpdateConversationsByUser)
 	}
 
 	{

internal/msggateway/client.go

@@ -16,7 +16,6 @@ package msggateway
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"sync"
 	"sync/atomic"
@@ -31,7 +30,6 @@ import (
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/mcontext"
-	"github.com/openimsdk/tools/utils/stringutil"
 )
 
 var (
@@ -64,12 +62,13 @@ type PingPongHandler func(string) error
 
 type Client struct {
 	w              *sync.Mutex
-	conn           LongConn
+	conn           ClientConn
 	PlatformID     int    `json:"platformID"`
 	IsCompress     bool   `json:"isCompress"`
 	UserID         string `json:"userID"`
 	IsBackground   bool   `json:"isBackground"`
 	SDKType        string `json:"sdkType"`
+	SDKVersion     string `json:"sdkVersion"`
 	Encoder        Encoder
 	ctx            *UserConnContext
 	longConnServer LongConnServer
@@ -83,10 +82,10 @@ type Client struct {
 }
 
 // ResetClient updates the client's state with new connection and context information.
-func (c *Client) ResetClient(ctx *UserConnContext, conn LongConn, longConnServer LongConnServer) {
+func (c *Client) ResetClient(ctx *UserConnContext, conn ClientConn, longConnServer LongConnServer) {
 	c.w = new(sync.Mutex)
 	c.conn = conn
-	c.PlatformID = stringutil.StringToInt(ctx.GetPlatformID())
+	c.PlatformID = ctx.GetPlatformID()
 	c.IsCompress = ctx.GetCompression()
 	c.IsBackground = ctx.GetBackground()
 	c.UserID = ctx.GetUserID()
@@ -97,6 +96,7 @@ func (c *Client) ResetClient(ctx *UserConnContext, conn LongConn, longConnServer
 	c.closedErr = nil
 	c.token = ctx.GetToken()
 	c.SDKType = ctx.GetSDKType()
+	c.SDKVersion = ctx.GetSDKVersion()
 	c.hbCtx, c.hbCancel = context.WithCancel(c.ctx)
 	c.subLock = new(sync.Mutex)
 	if c.subUserIDs != nil {
@@ -110,22 +110,6 @@ func (c *Client) ResetClient(ctx *UserConnContext, conn LongConn, longConnServer
 	c.subUserIDs = make(map[string]struct{})
 }
 
-func (c *Client) pingHandler(appData string) error {
-	if err := c.conn.SetReadDeadline(pongWait); err != nil {
-		return err
-	}
-
-	log.ZDebug(c.ctx, "ping Handler Success.", "appData", appData)
-	return c.writePongMsg(appData)
-}
-
-func (c *Client) pongHandler(_ string) error {
-	if err := c.conn.SetReadDeadline(pongWait); err != nil {
-		return err
-	}
-	return nil
-}
-
 // readMessage continuously reads messages from the connection.
 func (c *Client) readMessage() {
 	defer func() {
@@ -136,52 +120,25 @@ func (c *Client) readMessage() {
 		c.close()
 	}()
 
-	c.conn.SetReadLimit(maxMessageSize)
-	_ = c.conn.SetReadDeadline(pongWait)
-	c.conn.SetPongHandler(c.pongHandler)
-	c.conn.SetPingHandler(c.pingHandler)
-	c.activeHeartbeat(c.hbCtx)
-
 	for {
 		log.ZDebug(c.ctx, "readMessage")
-		messageType, message, returnErr := c.conn.ReadMessage()
+		message, returnErr := c.conn.ReadMessage()
 		if returnErr != nil {
-			log.ZWarn(c.ctx, "readMessage", returnErr, "messageType", messageType)
+			log.ZWarn(c.ctx, "readMessage", returnErr)
 			c.closedErr = returnErr
 			return
 		}
 
-		log.ZDebug(c.ctx, "readMessage", "messageType", messageType)
 		if c.closed.Load() {
 			// The scenario where the connection has just been closed, but the coroutine has not exited
 			c.closedErr = ErrConnClosed
 			return
 		}
 
-		switch messageType {
-		case MessageBinary:
-			_ = c.conn.SetReadDeadline(pongWait)
-			parseDataErr := c.handleMessage(message)
-			if parseDataErr != nil {
-				c.closedErr = parseDataErr
-				return
-			}
-		case MessageText:
-			_ = 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)
-
-		case CloseMessage:
-			c.closedErr = ErrClientClosed
+		parseDataErr := c.handleMessage(message)
+		if parseDataErr != nil {
+			c.closedErr = parseDataErr
 			return
-
-		default:
 		}
 	}
 }
@@ -356,109 +313,13 @@ func (c *Client) writeBinaryMsg(resp Resp) error {
 	c.w.Lock()
 	defer c.w.Unlock()
 
-	err = c.conn.SetWriteDeadline(writeWait)
-	if err != nil {
-		return err
-	}
-
 	if c.IsCompress {
 		resultBuf, compressErr := c.longConnServer.CompressWithPool(encodedBuf)
 		if compressErr != nil {
 			return compressErr
 		}
-		return c.conn.WriteMessage(MessageBinary, resultBuf)
+		return c.conn.WriteMessage(resultBuf)
 	}
 
-	return c.conn.WriteMessage(MessageBinary, encodedBuf)
-}
-
-// Actively initiate Heartbeat when platform in Web.
-func (c *Client) activeHeartbeat(ctx context.Context) {
-	if c.PlatformID == constant.WebPlatformID {
-		go func() {
-			defer func() {
-				if r := recover(); r != nil {
-					log.ZPanic(ctx, "activeHeartbeat Panic", errs.ErrPanic(r))
-				}
-			}()
-			log.ZDebug(ctx, "server initiative send heartbeat start.")
-			ticker := time.NewTicker(pingPeriod)
-			defer ticker.Stop()
-
-			for {
-				select {
-				case <-ticker.C:
-					if err := c.writePingMsg(); err != nil {
-						log.ZWarn(c.ctx, "send Ping Message error.", err)
-						return
-					}
-				case <-c.hbCtx.Done():
-					return
-				}
-			}
-		}()
-	}
-}
-func (c *Client) writePingMsg() error {
-	if c.closed.Load() {
-		return nil
-	}
-
-	c.w.Lock()
-	defer c.w.Unlock()
-
-	err := c.conn.SetWriteDeadline(writeWait)
-	if err != nil {
-		return err
-	}
-
-	return c.conn.WriteMessage(PingMessage, nil)
-}
-
-func (c *Client) writePongMsg(appData string) error {
-	log.ZDebug(c.ctx, "write Pong Msg in Server", "appData", appData)
-	if c.closed.Load() {
-		log.ZWarn(c.ctx, "is closed in server", nil, "appdata", appData, "closed err", c.closedErr)
-		return nil
-	}
-
-	c.w.Lock()
-	defer c.w.Unlock()
-
-	err := c.conn.SetWriteDeadline(writeWait)
-	if err != nil {
-		log.ZWarn(c.ctx, "SetWriteDeadline in Server have error", errs.Wrap(err), "writeWait", writeWait, "appData", appData)
-		return errs.Wrap(err)
-	}
-	err = c.conn.WriteMessage(PongMessage, []byte(appData))
-	if err != nil {
-		log.ZWarn(c.ctx, "Write Message have error", errs.Wrap(err), "Pong msg", PongMessage)
-	}
-
-	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)
-	}
+	return c.conn.WriteMessage(encodedBuf)
 }

internal/msggateway/client_conn.go

@@ -0,0 +1,229 @@
+package msggateway
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sync/atomic"
+	"time"
+
+	"github.com/gorilla/websocket"
+
+	"github.com/openimsdk/tools/log"
+)
+
+var ErrWriteFull = fmt.Errorf("websocket write buffer full,close connection")
+
+type ClientConn interface {
+	ReadMessage() ([]byte, error)
+	WriteMessage(message []byte) error
+	Close() error
+}
+
+type websocketMessage struct {
+	MessageType int
+	Data        []byte
+}
+
+func NewWebSocketClientConn(conn *websocket.Conn, readLimit int64, readTimeout time.Duration, pingInterval time.Duration) ClientConn {
+	c := &websocketClientConn{
+		readTimeout: readTimeout,
+		conn:        conn,
+		writer:      make(chan *websocketMessage, 256),
+		done:        make(chan struct{}),
+	}
+	if readLimit > 0 {
+		c.conn.SetReadLimit(readLimit)
+	}
+	c.conn.SetPingHandler(c.pingHandler)
+	c.conn.SetPongHandler(c.pongHandler)
+
+	go c.loopSend()
+	if pingInterval > 0 {
+		go c.doPing(pingInterval)
+	}
+	return c
+}
+
+type websocketClientConn struct {
+	readTimeout time.Duration
+	conn        *websocket.Conn
+	writer      chan *websocketMessage
+	done        chan struct{}
+	err         atomic.Pointer[error]
+}
+
+func (c *websocketClientConn) ReadMessage() ([]byte, error) {
+	buf, err := c.readMessage()
+	if err != nil {
+		return nil, c.closeBy(fmt.Errorf("read message %w", err))
+	}
+	return buf, nil
+}
+
+func (c *websocketClientConn) WriteMessage(message []byte) error {
+	return c.writeMessage(websocket.BinaryMessage, message)
+}
+
+func (c *websocketClientConn) Close() error {
+	return c.closeBy(fmt.Errorf("websocket connection closed"))
+}
+
+func (c *websocketClientConn) closeBy(err error) error {
+	if !c.err.CompareAndSwap(nil, &err) {
+		return *c.err.Load()
+	}
+	close(c.done)
+	log.ZWarn(context.Background(), "websocket connection closed", err, "remoteAddr", c.conn.RemoteAddr(),
+		"chan length", len(c.writer))
+	return err
+}
+
+func (c *websocketClientConn) writeMessage(messageType int, data []byte) error {
+	if errPtr := c.err.Load(); errPtr != nil {
+		return *errPtr
+	}
+	select {
+	case c.writer <- &websocketMessage{MessageType: messageType, Data: data}:
+		return nil
+	default:
+		return c.closeBy(ErrWriteFull)
+	}
+}
+
+func (c *websocketClientConn) loopSend() {
+	defer func() {
+		_ = c.conn.Close()
+	}()
+	var err error
+	for {
+		select {
+		case <-c.done:
+			for {
+				select {
+				case msg := <-c.writer:
+					switch msg.MessageType {
+					case websocket.TextMessage, websocket.BinaryMessage:
+						err = c.conn.WriteMessage(msg.MessageType, msg.Data)
+					default:
+						err = c.conn.WriteControl(msg.MessageType, msg.Data, time.Time{})
+					}
+					if err != nil {
+						_ = c.closeBy(err)
+						return
+					}
+				default:
+					return
+				}
+			}
+		case msg := <-c.writer:
+			switch msg.MessageType {
+			case websocket.TextMessage, websocket.BinaryMessage:
+				err = c.conn.WriteMessage(msg.MessageType, msg.Data)
+			default:
+				err = c.conn.WriteControl(msg.MessageType, msg.Data, time.Time{})
+			}
+			if err != nil {
+				_ = c.closeBy(err)
+				return
+			}
+		}
+	}
+}
+
+func (c *websocketClientConn) setReadDeadline() error {
+	deadline := time.Now().Add(c.readTimeout)
+	return c.conn.SetReadDeadline(deadline)
+}
+
+func (c *websocketClientConn) readMessage() ([]byte, error) {
+	for {
+		if err := c.setReadDeadline(); err != nil {
+			return nil, err
+		}
+		messageType, buf, err := c.conn.ReadMessage()
+		if err != nil {
+			return nil, err
+		}
+		switch messageType {
+		case websocket.BinaryMessage:
+			return buf, nil
+		case websocket.TextMessage:
+			if err := c.onReadTextMessage(buf); err != nil {
+				return nil, err
+			}
+		case websocket.PingMessage:
+			if err := c.pingHandler(string(buf)); err != nil {
+				return nil, err
+			}
+		case websocket.PongMessage:
+			if err := c.pongHandler(string(buf)); err != nil {
+				return nil, err
+			}
+		case websocket.CloseMessage:
+			if len(buf) == 0 {
+				return nil, errors.New("websocket connection closed by peer")
+			}
+			return nil, fmt.Errorf("websocket connection closed by peer, data %s", string(buf))
+		default:
+			return nil, fmt.Errorf("unknown websocket message type %d", messageType)
+		}
+	}
+}
+
+func (c *websocketClientConn) onReadTextMessage(buf []byte) error {
+	var msg struct {
+		Type string          `json:"type"`
+		Body json.RawMessage `json:"body"`
+	}
+	if err := json.Unmarshal(buf, &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
+		}
+		return c.writeMessage(websocket.TextMessage, msgData)
+	default:
+		return fmt.Errorf("not support text message type %s", msg.Type)
+	}
+}
+
+func (c *websocketClientConn) pingHandler(appData string) error {
+	log.ZDebug(context.Background(), "ping handler recv ping", "remoteAddr", c.conn.RemoteAddr(), "appData", appData)
+	if err := c.setReadDeadline(); err != nil {
+		return err
+	}
+	err := c.conn.WriteControl(websocket.PongMessage, []byte(appData), time.Now().Add(time.Second*1))
+	if err != nil {
+		log.ZWarn(context.Background(), "ping handler write pong error", err, "remoteAddr", c.conn.RemoteAddr(), "appData", appData)
+	}
+	log.ZDebug(context.Background(), "ping handler write pong success", "remoteAddr", c.conn.RemoteAddr(), "appData", appData)
+	return nil
+}
+
+func (c *websocketClientConn) pongHandler(string) error {
+	return nil
+}
+
+func (c *websocketClientConn) doPing(d time.Duration) {
+	ticker := time.NewTicker(d)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-c.done:
+			return
+		case <-ticker.C:
+			if err := c.writeMessage(websocket.PingMessage, nil); err != nil {
+				_ = c.closeBy(fmt.Errorf("send ping %w", err))
+				return
+			}
+		}
+	}
+}

internal/msggateway/constant.go

@@ -28,6 +28,7 @@ const (
 	BackgroundStatus        = "isBackground"
 	SendResponse            = "isMsgResp"
 	SDKType                 = "sdkType"
+	SDKVersion              = "sdkVersion"
 )
 
 const (

internal/msggateway/context.go

@@ -15,18 +15,32 @@
 package msggateway
 
 import (
-	"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
+	"encoding/base64"
+	"encoding/json"
 	"net/http"
 	"net/url"
 	"strconv"
 	"time"
 
+	"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
+
 	"github.com/openimsdk/protocol/constant"
 	"github.com/openimsdk/tools/utils/encrypt"
-	"github.com/openimsdk/tools/utils/stringutil"
 	"github.com/openimsdk/tools/utils/timeutil"
 )
 
+type UserConnContextInfo struct {
+	Token        string `json:"token"`
+	UserID       string `json:"userID"`
+	PlatformID   int    `json:"platformID"`
+	OperationID  string `json:"operationID"`
+	Compression  string `json:"compression"`
+	SDKType      string `json:"sdkType"`
+	SendResponse bool   `json:"sendResponse"`
+	Background   bool   `json:"background"`
+	SDKVersion   string `json:"sdkVersion"`
+}
+
 type UserConnContext struct {
 	RespWriter http.ResponseWriter
 	Req        *http.Request
@@ -34,6 +48,7 @@ type UserConnContext struct {
 	Method     string
 	RemoteAddr string
 	ConnID     string
+	info       *UserConnContextInfo
 }
 
 func (c *UserConnContext) Deadline() (deadline time.Time, ok bool) {
@@ -57,9 +72,11 @@ func (c *UserConnContext) Value(key any) any {
 	case constant.ConnID:
 		return c.GetConnID()
 	case constant.OpUserPlatform:
-		return constant.PlatformIDToName(stringutil.StringToInt(c.GetPlatformID()))
+		return c.GetPlatformID()
 	case constant.RemoteAddr:
 		return c.RemoteAddr
+	case SDKVersion:
+		return c.info.SDKVersion
 	default:
 		return ""
 	}
@@ -82,30 +99,92 @@ func newContext(respWriter http.ResponseWriter, req *http.Request) *UserConnCont
 
 func newTempContext() *UserConnContext {
 	return &UserConnContext{
-		Req: &http.Request{URL: &url.URL{}},
+		Req:  &http.Request{URL: &url.URL{}},
+		info: &UserConnContextInfo{},
 	}
 }
 
+func (c *UserConnContext) ParseEssentialArgs() error {
+	query := c.Req.URL.Query()
+	if data := query.Get("v"); data != "" {
+		return c.parseByJson(data)
+	} else {
+		return c.parseByQuery(query, c.Req.Header)
+	}
+}
+
+func (c *UserConnContext) parseByQuery(query url.Values, header http.Header) error {
+	info := UserConnContextInfo{
+		Token:       query.Get(Token),
+		UserID:      query.Get(WsUserID),
+		OperationID: query.Get(OperationID),
+		Compression: query.Get(Compression),
+		SDKType:     query.Get(SDKType),
+		SDKVersion:  query.Get(SDKVersion),
+	}
+	platformID, err := strconv.Atoi(query.Get(PlatformID))
+	if err != nil {
+		return servererrs.ErrConnArgsErr.WrapMsg("platformID is not int")
+	}
+	info.PlatformID = platformID
+	if val := query.Get(SendResponse); val != "" {
+		ok, err := strconv.ParseBool(val)
+		if err != nil {
+			return servererrs.ErrConnArgsErr.WrapMsg("isMsgResp is not bool")
+		}
+		info.SendResponse = ok
+	}
+	if info.Compression == "" {
+		info.Compression = header.Get(Compression)
+	}
+	background, err := strconv.ParseBool(query.Get(BackgroundStatus))
+	if err != nil {
+		return err
+	}
+	info.Background = background
+	return c.checkInfo(&info)
+}
+
+func (c *UserConnContext) parseByJson(data string) error {
+	reqInfo, err := base64.RawURLEncoding.DecodeString(data)
+	if err != nil {
+		return servererrs.ErrConnArgsErr.WrapMsg("data is not base64")
+	}
+	var info UserConnContextInfo
+	if err := json.Unmarshal(reqInfo, &info); err != nil {
+		return servererrs.ErrConnArgsErr.WrapMsg("data is not json", "info", err.Error())
+	}
+	return c.checkInfo(&info)
+}
+
+func (c *UserConnContext) checkInfo(info *UserConnContextInfo) error {
+	if info.OperationID == "" {
+		return servererrs.ErrConnArgsErr.WrapMsg("operationID is empty")
+	}
+	if info.Token == "" {
+		return servererrs.ErrConnArgsErr.WrapMsg("token is empty")
+	}
+	if info.UserID == "" {
+		return servererrs.ErrConnArgsErr.WrapMsg("sendID is empty")
+	}
+	if _, ok := constant.PlatformID2Name[info.PlatformID]; !ok {
+		return servererrs.ErrConnArgsErr.WrapMsg("platformID is invalid")
+	}
+	switch info.SDKType {
+	case "":
+		info.SDKType = GoSDK
+	case GoSDK, JsSDK:
+	default:
+		return servererrs.ErrConnArgsErr.WrapMsg("sdkType is invalid")
+	}
+	c.info = info
+	return nil
+}
+
 func (c *UserConnContext) GetRemoteAddr() string {
 	return c.RemoteAddr
 }
 
-func (c *UserConnContext) Query(key string) (string, bool) {
-	var value string
-	if value = c.Req.URL.Query().Get(key); value == "" {
-		return value, false
-	}
-	return value, true
-}
-
-func (c *UserConnContext) GetHeader(key string) (string, bool) {
-	var value string
-	if value = c.Req.Header.Get(key); value == "" {
-		return value, false
-	}
-	return value, true
-}
-
 func (c *UserConnContext) SetHeader(key, value string) {
 	c.RespWriter.Header().Set(key, value)
 }
@@ -119,93 +198,76 @@ func (c *UserConnContext) GetConnID() string {
 }
 
 func (c *UserConnContext) GetUserID() string {
-	return c.Req.URL.Query().Get(WsUserID)
+	if c == nil || c.info == nil {
+		return ""
+	}
+	return c.info.UserID
 }
 
-func (c *UserConnContext) GetPlatformID() string {
-	return c.Req.URL.Query().Get(PlatformID)
+func (c *UserConnContext) GetPlatformID() int {
+	if c == nil || c.info == nil {
+		return 0
+	}
+	return c.info.PlatformID
 }
 
 func (c *UserConnContext) GetOperationID() string {
-	return c.Req.URL.Query().Get(OperationID)
+	if c == nil || c.info == nil {
+		return ""
+	}
+	return c.info.OperationID
 }
 
 func (c *UserConnContext) SetOperationID(operationID string) {
-	values := c.Req.URL.Query()
-	values.Set(OperationID, operationID)
-	c.Req.URL.RawQuery = values.Encode()
+	if c.info == nil {
+		c.info = &UserConnContextInfo{}
+	}
+	c.info.OperationID = operationID
 }
 
 func (c *UserConnContext) GetToken() string {
-	return c.Req.URL.Query().Get(Token)
+	if c == nil || c.info == nil {
+		return ""
+	}
+	return c.info.Token
 }
 
 func (c *UserConnContext) GetCompression() bool {
-	compression, exists := c.Query(Compression)
-	if exists && compression == GzipCompressionProtocol {
-		return true
-	} else {
-		compression, exists := c.GetHeader(Compression)
-		if exists && compression == GzipCompressionProtocol {
-			return true
-		}
-	}
-	return false
+	return c != nil && c.info != nil && c.info.Compression == GzipCompressionProtocol
 }
 
 func (c *UserConnContext) GetSDKType() string {
-	sdkType := c.Req.URL.Query().Get(SDKType)
-	if sdkType == "" {
-		sdkType = GoSDK
+	if c == nil || c.info == nil {
+		return GoSDK
 	}
-	return sdkType
+	switch c.info.SDKType {
+	case "", GoSDK:
+		return GoSDK
+	case JsSDK:
+		return JsSDK
+	default:
+		return ""
+	}
+}
+
+func (c *UserConnContext) GetSDKVersion() string {
+	if c == nil || c.info == nil {
+		return ""
+	}
+	return c.info.SDKVersion
 }
 
 func (c *UserConnContext) ShouldSendResp() bool {
-	errResp, exists := c.Query(SendResponse)
-	if exists {
-		b, err := strconv.ParseBool(errResp)
-		if err != nil {
-			return false
-		} else {
-			return b
-		}
-	}
-	return false
+	return c != nil && c.info != nil && c.info.SendResponse
 }
 
 func (c *UserConnContext) SetToken(token string) {
-	c.Req.URL.RawQuery = Token + "=" + token
+	if c.info == nil {
+		c.info = &UserConnContextInfo{}
+	}
+	c.info.Token = token
 }
 
 func (c *UserConnContext) GetBackground() bool {
-	b, err := strconv.ParseBool(c.Req.URL.Query().Get(BackgroundStatus))
-	if err != nil {
-		return false
-	}
-	return b
-}
-func (c *UserConnContext) ParseEssentialArgs() error {
-	_, exists := c.Query(Token)
-	if !exists {
-		return servererrs.ErrConnArgsErr.WrapMsg("token is empty")
-	}
-	_, exists = c.Query(WsUserID)
-	if !exists {
-		return servererrs.ErrConnArgsErr.WrapMsg("sendID is empty")
-	}
-	platformIDStr, exists := c.Query(PlatformID)
-	if !exists {
-		return servererrs.ErrConnArgsErr.WrapMsg("platformID is empty")
-	}
-	_, 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
+	return c != nil && c.info != nil && c.info.Background
 }

internal/msggateway/hub_server.go

@@ -152,19 +152,16 @@ func (s *Server) pushToUser(ctx context.Context, userID string, msgData *sdkws.M
 		userPlatform := &msggateway.SingleMsgToUserPlatform{
 			RecvPlatFormID: int32(client.PlatformID),
 		}
-		if !client.IsBackground ||
-			(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 {
-					result.OnlinePush = true
-				}
-			}
-		} else {
+		if client.IsBackground && client.PlatformID == constant.IOSPlatformID {
 			userPlatform.ResultCode = int64(servererrs.ErrIOSBackgroundPushErr.Code())
+			result.Resp = append(result.Resp, userPlatform)
+			continue
+		}
+		if err := client.PushMessage(ctx, msgData); 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 {
+			result.OnlinePush = true
 		}
 		result.Resp = append(result.Resp, userPlatform)
 	}

internal/msggateway/long_conn.go

@@ -1,179 +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 msggateway
-
-import (
-	"encoding/json"
-	"net/http"
-	"time"
-
-	"github.com/openimsdk/tools/apiresp"
-
-	"github.com/gorilla/websocket"
-	"github.com/openimsdk/tools/errs"
-)
-
-type LongConn interface {
-	// Close this connection
-	Close() error
-	// WriteMessage Write message to connection,messageType means data type,can be set binary(2) and text(1).
-	WriteMessage(messageType int, message []byte) error
-	// ReadMessage Read message from connection.
-	ReadMessage() (int, []byte, error)
-	// SetReadDeadline sets the read deadline on the underlying network connection,
-	// after a read has timed out, will return an error.
-	SetReadDeadline(timeout time.Duration) error
-	// SetWriteDeadline sets to write deadline when send message,when read has timed out,will return error.
-	SetWriteDeadline(timeout time.Duration) error
-	// Dial Try to dial a connection,url must set auth args,header can control compress data
-	Dial(urlStr string, requestHeader http.Header) (*http.Response, error)
-	// IsNil Whether the connection of the current long connection is nil
-	IsNil() bool
-	// SetConnNil Set the connection of the current long connection to nil
-	SetConnNil()
-	// SetReadLimit sets the maximum size for a message read from the peer.bytes
-	SetReadLimit(limit int64)
-	SetPongHandler(handler PingPongHandler)
-	SetPingHandler(handler PingPongHandler)
-	// GenerateLongConn Check the connection of the current and when it was sent are the same
-	GenerateLongConn(w http.ResponseWriter, r *http.Request) error
-}
-type GWebSocket struct {
-	protocolType     int
-	conn             *websocket.Conn
-	handshakeTimeout time.Duration
-	writeBufferSize  int
-}
-
-func newGWebSocket(protocolType int, handshakeTimeout time.Duration, wbs int) *GWebSocket {
-	return &GWebSocket{protocolType: protocolType, handshakeTimeout: handshakeTimeout, writeBufferSize: wbs}
-}
-
-func (d *GWebSocket) Close() error {
-	return d.conn.Close()
-}
-
-func (d *GWebSocket) GenerateLongConn(w http.ResponseWriter, r *http.Request) error {
-	upgrader := &websocket.Upgrader{
-		HandshakeTimeout: d.handshakeTimeout,
-		CheckOrigin:      func(r *http.Request) bool { return true },
-	}
-	if d.writeBufferSize > 0 { // default is 4kb.
-		upgrader.WriteBufferSize = d.writeBufferSize
-	}
-
-	conn, err := upgrader.Upgrade(w, r, nil)
-	if err != nil {
-		// The upgrader.Upgrade method usually returns enough error messages to diagnose problems that may occur during the upgrade
-		return errs.WrapMsg(err, "GenerateLongConn: WebSocket upgrade failed")
-	}
-	d.conn = conn
-	return nil
-}
-
-func (d *GWebSocket) WriteMessage(messageType int, message []byte) error {
-	// d.setSendConn(d.conn)
-	return d.conn.WriteMessage(messageType, message)
-}
-
-// func (d *GWebSocket) setSendConn(sendConn *websocket.Conn) {
-//	d.sendConn = sendConn
-//}
-
-func (d *GWebSocket) ReadMessage() (int, []byte, error) {
-	return d.conn.ReadMessage()
-}
-
-func (d *GWebSocket) SetReadDeadline(timeout time.Duration) error {
-	return d.conn.SetReadDeadline(time.Now().Add(timeout))
-}
-
-func (d *GWebSocket) SetWriteDeadline(timeout time.Duration) error {
-	if timeout <= 0 {
-		return errs.New("timeout must be greater than 0")
-	}
-
-	// TODO SetWriteDeadline Future add error handling
-	if err := d.conn.SetWriteDeadline(time.Now().Add(timeout)); err != nil {
-		return errs.WrapMsg(err, "GWebSocket.SetWriteDeadline failed")
-	}
-	return nil
-}
-
-func (d *GWebSocket) Dial(urlStr string, requestHeader http.Header) (*http.Response, error) {
-	conn, httpResp, err := websocket.DefaultDialer.Dial(urlStr, requestHeader)
-	if err != nil {
-		return httpResp, errs.WrapMsg(err, "GWebSocket.Dial failed", "url", urlStr)
-	}
-	d.conn = conn
-	return httpResp, nil
-}
-
-func (d *GWebSocket) IsNil() bool {
-	return d.conn == nil
-	//
-	// if d.conn != nil {
-	// 	return false
-	// }
-	// return true
-}
-
-func (d *GWebSocket) SetConnNil() {
-	d.conn = nil
-}
-
-func (d *GWebSocket) SetReadLimit(limit int64) {
-	d.conn.SetReadLimit(limit)
-}
-
-func (d *GWebSocket) SetPongHandler(handler PingPongHandler) {
-	d.conn.SetPongHandler(handler)
-}
-
-func (d *GWebSocket) SetPingHandler(handler PingPongHandler) {
-	d.conn.SetPingHandler(handler)
-}
-
-func (d *GWebSocket) RespondWithError(err error, w http.ResponseWriter, r *http.Request) error {
-	if err := d.GenerateLongConn(w, r); err != nil {
-		return err
-	}
-	data, err := json.Marshal(apiresp.ParseError(err))
-	if err != nil {
-		_ = d.Close()
-		return errs.WrapMsg(err, "json marshal failed")
-	}
-
-	if err := d.WriteMessage(MessageText, data); err != nil {
-		_ = d.Close()
-		return errs.WrapMsg(err, "WriteMessage failed")
-	}
-	_ = d.Close()
-	return nil
-}
-
-func (d *GWebSocket) RespondWithSuccess() error {
-	data, err := json.Marshal(apiresp.ParseError(nil))
-	if err != nil {
-		_ = d.Close()
-		return errs.WrapMsg(err, "json marshal failed")
-	}
-
-	if err := d.WriteMessage(MessageText, data); err != nil {
-		_ = d.Close()
-		return errs.WrapMsg(err, "WriteMessage failed")
-	}
-	return nil
-}

internal/msggateway/ws_server.go

@@ -2,13 +2,16 @@ package msggateway
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"net/http"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	"github.com/gorilla/websocket"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
+	"github.com/openimsdk/tools/apiresp"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/webhook"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpccache"
@@ -22,10 +25,11 @@ import (
 	"github.com/openimsdk/protocol/msggateway"
 	"github.com/openimsdk/tools/discovery"
 	"github.com/openimsdk/tools/log"
-	"github.com/openimsdk/tools/utils/stringutil"
 	"golang.org/x/sync/errgroup"
 )
 
+var wsSuccessResponse, _ = json.Marshal(&apiresp.ApiResponse{})
+
 type LongConnServer interface {
 	Run(ctx context.Context) error
 	wsHandler(w http.ResponseWriter, r *http.Request)
@@ -42,6 +46,7 @@ type LongConnServer interface {
 }
 
 type WsServer struct {
+	websocket         *websocket.Upgrader
 	msgGatewayConfig  *Config
 	port              int
 	wsMaxConnNum      int64
@@ -64,6 +69,8 @@ type WsServer struct {
 	webhookClient *webhook.Client
 	userClient    *rpcli.UserClient
 	authClient    *rpcli.AuthClient
+
+	ready atomic.Bool
 }
 
 type kickHandler struct {
@@ -93,6 +100,8 @@ func (ws *WsServer) SetDiscoveryRegistry(ctx context.Context, disCov discovery.C
 	ws.authClient = rpcli.NewAuthClient(authConn)
 	ws.MessageHandler = NewGrpcHandler(ws.validate, rpcli.NewMsgClient(msgConn), rpcli.NewPushMsgServiceClient(pushConn))
 	ws.disCov = disCov
+
+	ws.ready.Store(true)
 	return nil
 }
 
@@ -131,9 +140,13 @@ func NewWsServer(msgGatewayConfig *Config, opts ...Option) *WsServer {
 		o(&config)
 	}
 	//userRpcClient := rpcclient.NewUserRpcClient(client, config.Discovery.RpcService.User, config.Share.IMAdminUser)
-
+	upgrader := &websocket.Upgrader{
+		HandshakeTimeout: config.handshakeTimeout,
+		CheckOrigin:      func(r *http.Request) bool { return true },
+	}
 	v := validator.New()
 	return &WsServer{
+		websocket:        upgrader,
 		msgGatewayConfig: msgGatewayConfig,
 		port:             config.port,
 		wsMaxConnNum:     config.maxConnNum,
@@ -254,6 +267,9 @@ func (ws *WsServer) registerClient(client *Client) {
 		oldClients []*Client
 	)
 	oldClients, userOK, clientOK = ws.clients.Get(client.UserID, client.PlatformID)
+
+	log.ZInfo(client.ctx, "registerClient", "userID", client.UserID, "platformID", client.PlatformID)
+
 	if !userOK {
 		ws.clients.Set(client.UserID, client)
 		log.ZDebug(client.ctx, "user not exist", "userID", client.UserID, "platformID", client.PlatformID)
@@ -439,7 +455,7 @@ func (ws *WsServer) unregisterClient(client *Client) {
 // validateRespWithRequest checks if the response matches the expected userID and platformID.
 func (ws *WsServer) validateRespWithRequest(ctx *UserConnContext, resp *pbAuth.ParseTokenResp) error {
 	userID := ctx.GetUserID()
-	platformID := stringutil.StringToInt32(ctx.GetPlatformID())
+	platformID := int32(ctx.GetPlatformID())
 	if resp.UserID != userID {
 		return servererrs.ErrTokenInvalid.WrapMsg(fmt.Sprintf("token uid %s != userID %s", resp.UserID, userID))
 	}
@@ -449,6 +465,29 @@ func (ws *WsServer) validateRespWithRequest(ctx *UserConnContext, resp *pbAuth.P
 	return nil
 }
 
+func (ws *WsServer) handlerError(ctx *UserConnContext, w http.ResponseWriter, r *http.Request, err error) {
+	if !ctx.ShouldSendResp() {
+		httpError(ctx, err)
+		return
+	}
+	// the browser cannot get the response of upgrade failure
+	data, err := json.Marshal(apiresp.ParseError(err))
+	if err != nil {
+		log.ZError(ctx, "json marshal failed", err)
+		return
+	}
+	conn, upgradeErr := ws.websocket.Upgrade(w, r, nil)
+	if upgradeErr != nil {
+		log.ZWarn(ctx, "websocket upgrade failed", upgradeErr, "respErr", err, "resp", string(data))
+		return
+	}
+	defer conn.Close()
+	if err := conn.WriteMessage(websocket.TextMessage, data); err != nil {
+		log.ZWarn(ctx, "WriteMessage failed", err, "respErr", err, "resp", string(data))
+		return
+	}
+}
+
 func (ws *WsServer) wsHandler(w http.ResponseWriter, r *http.Request) {
 	// Create a new connection context
 	connContext := newContext(w, r)
@@ -456,7 +495,7 @@ func (ws *WsServer) wsHandler(w http.ResponseWriter, r *http.Request) {
 	// Check if the current number of online user connections exceeds the maximum limit
 	if ws.onlineUserConnNum.Load() >= ws.wsMaxConnNum {
 		// If it exceeds the maximum connection number, return an error via HTTP and stop processing
-		httpError(connContext, servererrs.ErrConnOverMaxNumLimit.WrapMsg("over max conn num limit"))
+		ws.handlerError(connContext, w, r, servererrs.ErrConnOverMaxNumLimit.WrapMsg("over max conn num limit"))
 		return
 	}
 
@@ -464,26 +503,14 @@ func (ws *WsServer) wsHandler(w http.ResponseWriter, r *http.Request) {
 	err := connContext.ParseEssentialArgs()
 	if err != nil {
 		// If there's an error during parsing, return an error via HTTP and stop processing
-
-		httpError(connContext, err)
+		ws.handlerError(connContext, w, r, err)
 		return
 	}
 
 	// Call the authentication client to parse the Token obtained from the context
 	resp, err := ws.authClient.ParseToken(connContext, connContext.GetToken())
 	if err != nil {
-		// If there's an error parsing the Token, decide whether to send the error message via WebSocket based on the context flag
-		shouldSendError := connContext.ShouldSendResp()
-		if shouldSendError {
-			// Create a WebSocket connection object and attempt to send the error message via WebSocket
-			wsLongConn := newGWebSocket(WebSocket, ws.handshakeTimeout, ws.writeBufferSize)
-			if err := wsLongConn.RespondWithError(err, w, r); err == nil {
-				// If the error message is successfully sent via WebSocket, stop processing
-				return
-			}
-		}
-		// If sending via WebSocket is not required or fails, return the error via HTTP and stop processing
-		httpError(connContext, err)
+		ws.handlerError(connContext, w, r, err)
 		return
 	}
 
@@ -491,32 +518,30 @@ func (ws *WsServer) wsHandler(w http.ResponseWriter, r *http.Request) {
 	err = ws.validateRespWithRequest(connContext, resp)
 	if err != nil {
 		// If validation fails, return an error via HTTP and stop processing
-		httpError(connContext, err)
+		ws.handlerError(connContext, w, r, err)
 		return
 	}
-
-	log.ZDebug(connContext, "new conn", "token", connContext.GetToken())
-	// Create a WebSocket long connection object
-	wsLongConn := newGWebSocket(WebSocket, ws.handshakeTimeout, ws.writeBufferSize)
-	if err := wsLongConn.GenerateLongConn(w, r); err != nil {
-		//If the creation of the long connection fails, the error is handled internally during the handshake process.
-		log.ZWarn(connContext, "long connection fails", err)
+	conn, err := ws.websocket.Upgrade(w, r, nil)
+	if err != nil {
+		log.ZWarn(connContext, "websocket upgrade failed", err)
 		return
-	} else {
-		// Check if a normal response should be sent via WebSocket
-		shouldSendSuccessResp := connContext.ShouldSendResp()
-		if shouldSendSuccessResp {
-			// Attempt to send a success message through WebSocket
-			if err := wsLongConn.RespondWithSuccess(); err != nil {
-				// If the success message is successfully sent, end further processing
-				return
-			}
+	}
+	if connContext.ShouldSendResp() {
+		if err := conn.WriteMessage(websocket.TextMessage, wsSuccessResponse); err != nil {
+			log.ZWarn(connContext, "WriteMessage first response", err)
+			return
 		}
 	}
 
-	// Retrieve a client object from the client pool, reset its state, and associate it with the current WebSocket long connection
-	client := ws.clientPool.Get().(*Client)
-	client.ResetClient(connContext, wsLongConn, ws)
+	log.ZDebug(connContext, "new conn", "token", connContext.GetToken())
+
+	var pingInterval time.Duration
+	if connContext.GetPlatformID() == constant.WebPlatformID {
+		pingInterval = pingPeriod
+	}
+
+	client := new(Client)
+	client.ResetClient(connContext, NewWebSocketClientConn(conn, maxMessageSize, pongWait, pingInterval), ws)
 
 	// Register the client with the server and start message processing
 	ws.registerChan <- client

internal/msgtransfer/callback.go

@@ -51,37 +51,24 @@ func GetContent(msg *sdkws.MsgData) string {
 	}
 }
 
-func (mc *OnlineHistoryMongoConsumerHandler) webhookAfterSendSingleMsg(ctx context.Context, after *config.AfterConfig, msg *sdkws.MsgData) {
-	if msg.ContentType == constant.Typing {
-		return
-	}
-
+func (mc *OnlineHistoryMongoConsumerHandler) webhookAfterMsgSaveDB(ctx context.Context, after *config.AfterConfig, msg *sdkws.MsgData) {
 	if !filterAfterMsg(msg, after) {
 		return
 	}
 
-	cbReq := &cbapi.CallbackAfterSendSingleMsgReq{
-		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendSingleMsgCommand),
-		RecvID:            msg.RecvID,
-	}
-	mc.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendSingleMsgResp{}, after, buildKeyMsgDataQuery(msg))
-}
-
-func (mc *OnlineHistoryMongoConsumerHandler) webhookAfterSendGroupMsg(ctx context.Context, after *config.AfterConfig, msg *sdkws.MsgData) {
-	if msg.ContentType == constant.Typing {
-		return
+	cbReq := &cbapi.CallbackAfterMsgSaveDBReq{
+		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterMsgSaveDBCommand),
 	}
 
-	if !filterAfterMsg(msg, after) {
-		return
+	switch msg.SessionType {
+	case constant.SingleChatType, constant.NotificationChatType:
+		cbReq.RecvID = msg.RecvID
+	case constant.ReadGroupChatType:
+		cbReq.GroupID = msg.GroupID
+	default:
 	}
 
-	cbReq := &cbapi.CallbackAfterSendGroupMsgReq{
-		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendGroupMsgCommand),
-		GroupID:           msg.GroupID,
-	}
-
-	mc.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendGroupMsgResp{}, after, buildKeyMsgDataQuery(msg))
+	mc.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterMsgSaveDBResp{}, after, buildKeyMsgDataQuery(msg))
 }
 
 func buildKeyMsgDataQuery(msg *sdkws.MsgData) map[string]string {

internal/msgtransfer/online_history_msg_handler.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+
 	"github.com/openimsdk/tools/mq"
 
 	"sync"

internal/msgtransfer/online_msg_to_mongo_handler.go

@@ -15,7 +15,6 @@
 package msgtransfer
 
 import (
-	"github.com/openimsdk/protocol/constant"
 	"github.com/openimsdk/tools/mq"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
@@ -57,7 +56,7 @@ func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(val mq.Message)
 	log.ZDebug(ctx, "mongo consumer recv msg", "msgs", msgFromMQ.String())
 	err = mc.msgTransferDatabase.BatchInsertChat2DB(ctx, msgFromMQ.ConversationID, msgFromMQ.MsgData, msgFromMQ.LastSeq)
 	if err != nil {
-		log.ZError(ctx, "single data insert to mongo err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID)
+		log.ZError(ctx, "batch data insert to mongo err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID)
 		prommetrics.MsgInsertMongoFailedCounter.Inc()
 	} else {
 		prommetrics.MsgInsertMongoSuccessCounter.Inc()
@@ -65,12 +64,7 @@ func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(val mq.Message)
 	}
 
 	for _, msgData := range msgFromMQ.MsgData {
-		switch msgData.SessionType {
-		case constant.SingleChatType:
-			mc.webhookAfterSendSingleMsg(ctx, &mc.config.WebhooksConfig.AfterSendSingleMsg, msgData)
-		case constant.ReadGroupChatType:
-			mc.webhookAfterSendGroupMsg(ctx, &mc.config.WebhooksConfig.AfterSendGroupMsg, msgData)
-		}
+		mc.webhookAfterMsgSaveDB(ctx, &mc.config.WebhooksConfig.AfterMsgSaveDB, msgData)
 	}
 
 	//var seqs []int64

internal/rpc/conversation/conversation.go

@@ -37,6 +37,7 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
 	"github.com/openimsdk/protocol/constant"
 	pbconversation "github.com/openimsdk/protocol/conversation"
+	"github.com/openimsdk/protocol/msg"
 	"github.com/openimsdk/protocol/sdkws"
 	"github.com/openimsdk/tools/discovery"
 	"github.com/openimsdk/tools/errs"
@@ -512,14 +513,6 @@ func (c *conversationServer) GetUserConversationIDsHash(ctx context.Context, req
 	return &pbconversation.GetUserConversationIDsHashResp{Hash: hash}, nil
 }
 
-func (c *conversationServer) GetConversationsByConversationID(ctx context.Context, req *pbconversation.GetConversationsByConversationIDReq) (*pbconversation.GetConversationsByConversationIDResp, error) {
-	conversations, err := c.conversationDatabase.GetConversationsByConversationID(ctx, req.ConversationIDs)
-	if err != nil {
-		return nil, err
-	}
-	return &pbconversation.GetConversationsByConversationIDResp{Conversations: convert.ConversationsDB2Pb(conversations)}, nil
-}
-
 func (c *conversationServer) GetConversationOfflinePushUserIDs(ctx context.Context, req *pbconversation.GetConversationOfflinePushUserIDsReq) (*pbconversation.GetConversationOfflinePushUserIDsResp, error) {
 	if req.ConversationID == "" {
 		return nil, errs.ErrArgs.WrapMsg("conversationID is empty")
@@ -716,56 +709,6 @@ func (c *conversationServer) GetOwnerConversation(ctx context.Context, req *pbco
 	}, nil
 }
 
-func (c *conversationServer) GetConversationsNeedClearMsg(ctx context.Context, _ *pbconversation.GetConversationsNeedClearMsgReq) (*pbconversation.GetConversationsNeedClearMsgResp, error) {
-	num, err := c.conversationDatabase.GetAllConversationIDsNumber(ctx)
-	if err != nil {
-		log.ZError(ctx, "GetAllConversationIDsNumber failed", err)
-		return nil, err
-	}
-	const batchNum = 100
-
-	if num == 0 {
-		return nil, errs.New("Need Destruct Msg is nil").Wrap()
-	}
-
-	maxPage := (num + batchNum - 1) / batchNum
-
-	temp := make([]*dbModel.Conversation, 0, maxPage*batchNum)
-
-	for pageNumber := 0; pageNumber < int(maxPage); pageNumber++ {
-		pagination := &sdkws.RequestPagination{
-			PageNumber: int32(pageNumber),
-			ShowNumber: batchNum,
-		}
-
-		conversationIDs, err := c.conversationDatabase.PageConversationIDs(ctx, pagination)
-		if err != nil {
-			log.ZError(ctx, "PageConversationIDs failed", err, "pageNumber", pageNumber)
-			continue
-		}
-
-		// log.ZDebug(ctx, "PageConversationIDs success", "pageNumber", pageNumber, "conversationIDsNum", len(conversationIDs), "conversationIDs", conversationIDs)
-		if len(conversationIDs) == 0 {
-			continue
-		}
-
-		conversations, err := c.conversationDatabase.GetConversationsByConversationID(ctx, conversationIDs)
-		if err != nil {
-			log.ZError(ctx, "GetConversationsByConversationID failed", err, "conversationIDs", conversationIDs)
-			continue
-		}
-
-		for _, conversation := range conversations {
-			if conversation.IsMsgDestruct && conversation.MsgDestructTime != 0 && ((time.Now().UnixMilli() > (conversation.MsgDestructTime + conversation.LatestMsgDestructTime.UnixMilli() + 8*60*60)) || // 8*60*60 is UTC+8
-				conversation.LatestMsgDestructTime.IsZero()) {
-				temp = append(temp, conversation)
-			}
-		}
-	}
-
-	return &pbconversation.GetConversationsNeedClearMsgResp{Conversations: convert.ConversationsDB2Pb(temp)}, nil
-}
-
 func (c *conversationServer) GetNotNotifyConversationIDs(ctx context.Context, req *pbconversation.GetNotNotifyConversationIDsReq) (*pbconversation.GetNotNotifyConversationIDsResp, error) {
 	if err := authverify.CheckAccess(ctx, req.UserID); err != nil {
 		return nil, err
@@ -795,7 +738,7 @@ func (c *conversationServer) ClearUserConversationMsg(ctx context.Context, req *
 	}
 	latestMsgDestructTime := time.UnixMilli(req.Timestamp)
 	for i, conversation := range conversations {
-		if conversation.IsMsgDestruct == false || conversation.MsgDestructTime == 0 {
+		if !conversation.IsMsgDestruct || conversation.MsgDestructTime == 0 {
 			continue
 		}
 		seq, err := c.msgClient.GetLastMessageSeqByTime(ctx, conversation.ConversationID, req.Timestamp-(conversation.MsgDestructTime*1000))
@@ -835,3 +778,53 @@ func (c *conversationServer) setConversationMinSeqAndLatestMsgDestructTime(ctx c
 	c.conversationNotificationSender.ConversationChangeNotification(ctx, ownerUserID, []string{conversationID})
 	return nil
 }
+
+func (c *conversationServer) DeleteConversations(ctx context.Context, req *pbconversation.DeleteConversationsReq) (resp *pbconversation.DeleteConversationsResp, err error) {
+	if err := authverify.CheckAccess(ctx, req.OwnerUserID); err != nil {
+		return nil, err
+	}
+	if req.NeedDeleteTime == 0 && len(req.ConversationIDs) == 0 {
+		return nil, errs.ErrArgs.WrapMsg("need_delete_time or conversationIDs need be set")
+	}
+
+	if req.NeedDeleteTime != 0 && len(req.ConversationIDs) != 0 {
+		return nil, errs.ErrArgs.WrapMsg("need_delete_time and conversationIDs cannot both be set")
+	}
+
+	var needDeleteConversationIDs []string
+
+	if len(req.ConversationIDs) == 0 {
+		deleteTimeThreshold := time.Now().AddDate(0, 0, -int(req.NeedDeleteTime)).UnixMilli()
+		conversationIDs, err := c.conversationDatabase.GetConversationIDs(ctx, req.OwnerUserID)
+		if err != nil {
+			return nil, err
+		}
+		latestMsgs, err := c.msgClient.GetLastMessage(ctx, &msg.GetLastMessageReq{
+			UserID:          req.OwnerUserID,
+			ConversationIDs: conversationIDs,
+		})
+		if err != nil {
+			return nil, err
+		}
+
+		for conversationID, msg := range latestMsgs.Msgs {
+			if msg.SendTime < deleteTimeThreshold {
+				needDeleteConversationIDs = append(needDeleteConversationIDs, conversationID)
+			}
+		}
+
+		if len(needDeleteConversationIDs) == 0 {
+			return &pbconversation.DeleteConversationsResp{}, nil
+		}
+	} else {
+		needDeleteConversationIDs = req.ConversationIDs
+	}
+
+	if err := c.conversationDatabase.DeleteUsersConversations(ctx, req.OwnerUserID, needDeleteConversationIDs); err != nil {
+		return nil, err
+	}
+
+	// c.conversationNotificationSender.ConversationDeleteNotification(ctx, req.OwnerUserID, needDeleteConversationIDs)
+
+	return &pbconversation.DeleteConversationsResp{}, nil
+}

internal/rpc/conversation/notification.go

@@ -73,3 +73,12 @@ func (c *ConversationNotificationSender) ConversationUnreadChangeNotification(
 
 	c.Notification(ctx, userID, userID, constant.ConversationUnreadNotification, tips)
 }
+
+func (c *ConversationNotificationSender) ConversationDeleteNotification(ctx context.Context, userID string, conversationIDs []string) {
+	tips := &sdkws.ConversationDeleteTips{
+		UserID:          userID,
+		ConversationIDs: conversationIDs,
+	}
+
+	c.Notification(ctx, userID, userID, constant.ConversationDeleteNotification, tips)
+}

internal/rpc/conversation/sync.go

@@ -27,7 +27,7 @@ func (c *conversationServer) GetFullOwnerConversationIDs(ctx context.Context, re
 		conversationIDs = nil
 	}
 	return &conversation.GetFullOwnerConversationIDsResp{
-		Version:         idHash,
+		Version:         uint64(vl.Version),
 		VersionID:       vl.ID.Hex(),
 		Equal:           req.IdHash == idHash,
 		ConversationIDs: conversationIDs,

internal/rpc/group/group.go

@@ -472,6 +472,9 @@ func (g *groupServer) InviteUserToGroup(ctx context.Context, req *pbgroup.Invite
 			g.notification.GroupApplicationAgreeMemberEnterNotification(ctx, req.GroupID, req.SendMessage, opUserID, userIDs...)
 		}
 	}
+	if err := g.setMemberJoinSeq(ctx, req.GroupID, req.InvitedUserIDs); err != nil {
+		return nil, err
+	}
 	return &pbgroup.InviteUserToGroupResp{}, nil
 }
 
@@ -602,10 +605,6 @@ func (g *groupServer) KickGroupMember(ctx context.Context, req *pbgroup.KickGrou
 			}
 		}
 	}
-	num, err := g.db.FindGroupMemberNum(ctx, req.GroupID)
-	if err != nil {
-		return nil, err
-	}
 	ownerUserIDs, err := g.db.GetGroupRoleLevelMemberIDs(ctx, req.GroupID, constant.GroupOwner)
 	if err != nil {
 		return nil, err
@@ -617,6 +616,10 @@ func (g *groupServer) KickGroupMember(ctx context.Context, req *pbgroup.KickGrou
 	if err := g.db.DeleteGroupMember(ctx, group.GroupID, req.KickedUserIDs); err != nil {
 		return nil, err
 	}
+	num, err := g.db.FindGroupMemberNum(ctx, req.GroupID)
+	if err != nil {
+		return nil, err
+	}
 	tips := &sdkws.MemberKickedTips{
 		Group: &sdkws.GroupInfo{
 			GroupID:                group.GroupID,
@@ -626,7 +629,7 @@ func (g *groupServer) KickGroupMember(ctx context.Context, req *pbgroup.KickGrou
 			FaceURL:                group.FaceURL,
 			OwnerUserID:            ownerUserID,
 			CreateTime:             group.CreateTime.UnixMilli(),
-			MemberCount:            num - uint32(len(req.KickedUserIDs)),
+			MemberCount:            num,
 			Ex:                     group.Ex,
 			Status:                 group.Status,
 			CreatorUserID:          group.CreatorUserID,
@@ -905,6 +908,9 @@ func (g *groupServer) GroupApplicationResponse(ctx context.Context, req *pbgroup
 					return nil, err
 				}
 			}
+			if err := g.setMemberJoinSeq(ctx, req.GroupID, []string{req.FromUserID}); err != nil {
+				return nil, err
+			}
 		}
 	case constant.GroupResponseRefuse:
 		g.notification.GroupApplicationRejectedNotification(ctx, req)
@@ -967,6 +973,9 @@ func (g *groupServer) JoinGroup(ctx context.Context, req *pbgroup.JoinGroupReq)
 		if err = g.notification.MemberEnterNotification(ctx, req.GroupID, req.InviterUserID); err != nil {
 			return nil, err
 		}
+		if err := g.setMemberJoinSeq(ctx, req.GroupID, []string{req.InviterUserID}); err != nil {
+			return nil, err
+		}
 		g.webhookAfterJoinGroup(ctx, &g.config.WebhooksConfig.AfterJoinGroup, req)
 
 		return &pbgroup.JoinGroupResp{}, nil
@@ -1028,6 +1037,11 @@ func (g *groupServer) deleteMemberAndSetConversationSeq(ctx context.Context, gro
 	return g.conversationClient.SetConversationMaxSeq(ctx, conversationID, userIDs, maxSeq)
 }
 
+func (g *groupServer) setMemberJoinSeq(ctx context.Context, groupID string, userIDs []string) error {
+	conversationID := msgprocessor.GetConversationIDBySessionType(constant.ReadGroupChatType, groupID)
+	return g.conversationClient.SetConversationMaxSeq(ctx, conversationID, userIDs, 0)
+}
+
 func (g *groupServer) SetGroupInfo(ctx context.Context, req *pbgroup.SetGroupInfoReq) (*pbgroup.SetGroupInfoResp, error) {
 	var opMember *model.GroupMember
 	if !authverify.IsAdmin(ctx) {

internal/rpc/group/sync.go

@@ -34,7 +34,7 @@ func (g *groupServer) GetFullGroupMemberUserIDs(ctx context.Context, req *pbgrou
 		userIDs = nil
 	}
 	return &pbgroup.GetFullGroupMemberUserIDsResp{
-		Version:   idHash,
+		Version:   uint64(vl.Version),
 		VersionID: vl.ID.Hex(),
 		Equal:     req.IdHash == idHash,
 		UserIDs:   userIDs,
@@ -58,7 +58,7 @@ func (g *groupServer) GetFullJoinGroupIDs(ctx context.Context, req *pbgroup.GetF
 		groupIDs = nil
 	}
 	return &pbgroup.GetFullJoinGroupIDsResp{
-		Version:   idHash,
+		Version:   uint64(vl.Version),
 		VersionID: vl.ID.Hex(),
 		Equal:     req.IdHash == idHash,
 		GroupIDs:  groupIDs,

internal/rpc/msg/callback.go

@@ -16,8 +16,10 @@ package msg
 
 import (
 	"context"
+	"encoding/base64"
 	"encoding/json"
 
+	"github.com/openimsdk/open-im-server/v3/pkg/apistruct"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/webhook"
 	"github.com/openimsdk/tools/errs"
 
@@ -28,6 +30,7 @@ import (
 	"github.com/openimsdk/protocol/sdkws"
 	"github.com/openimsdk/tools/mcontext"
 	"github.com/openimsdk/tools/utils/datautil"
+	"github.com/openimsdk/tools/utils/stringutil"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -87,19 +90,19 @@ func (m *msgServer) webhookBeforeSendSingleMsg(ctx context.Context, before *conf
 }
 
 // Move to msgtransfer
-// func (m *msgServer) webhookAfterSendSingleMsg(ctx context.Context, after *config.AfterConfig, msg *pbchat.SendMsgReq) {
-// 	if msg.MsgData.ContentType == constant.Typing {
-// 		return
-// 	}
-// 	if !filterAfterMsg(msg, after) {
-// 		return
-// 	}
-// 	cbReq := &cbapi.CallbackAfterSendSingleMsgReq{
-// 		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendSingleMsgCommand),
-// 		RecvID:            msg.MsgData.RecvID,
-// 	}
-// 	m.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendSingleMsgResp{}, after, buildKeyMsgDataQuery(msg.MsgData))
-// }
+func (m *msgServer) webhookAfterSendSingleMsg(ctx context.Context, after *config.AfterConfig, msg *pbchat.SendMsgReq) {
+	if msg.MsgData.ContentType == constant.Typing {
+		return
+	}
+	if !filterAfterMsg(msg, after) {
+		return
+	}
+	cbReq := &cbapi.CallbackAfterSendSingleMsgReq{
+		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendSingleMsgCommand),
+		RecvID:            msg.MsgData.RecvID,
+	}
+	m.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendSingleMsgResp{}, after, buildKeyMsgDataQuery(msg.MsgData))
+}
 
 func (m *msgServer) webhookBeforeSendGroupMsg(ctx context.Context, before *config.BeforeConfig, msg *pbchat.SendMsgReq) error {
 	return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
@@ -121,21 +124,20 @@ func (m *msgServer) webhookBeforeSendGroupMsg(ctx context.Context, before *confi
 	})
 }
 
-// Move to msgtransfer
-// func (m *msgServer) webhookAfterSendGroupMsg(ctx context.Context, after *config.AfterConfig, msg *pbchat.SendMsgReq) {
-// 	if msg.MsgData.ContentType == constant.Typing {
-// 		return
-// 	}
-// 	if !filterAfterMsg(msg, after) {
-// 		return
-// 	}
-// 	cbReq := &cbapi.CallbackAfterSendGroupMsgReq{
-// 		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendGroupMsgCommand),
-// 		GroupID:           msg.MsgData.GroupID,
-// 	}
+func (m *msgServer) webhookAfterSendGroupMsg(ctx context.Context, after *config.AfterConfig, msg *pbchat.SendMsgReq) {
+	if msg.MsgData.ContentType == constant.Typing {
+		return
+	}
+	if !filterAfterMsg(msg, after) {
+		return
+	}
+	cbReq := &cbapi.CallbackAfterSendGroupMsgReq{
+		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendGroupMsgCommand),
+		GroupID:           msg.MsgData.GroupID,
+	}
 
-// 	m.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendGroupMsgResp{}, after, buildKeyMsgDataQuery(msg.MsgData))
-// }
+	m.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendGroupMsgResp{}, after, buildKeyMsgDataQuery(msg.MsgData))
+}
 
 func (m *msgServer) webhookBeforeMsgModify(ctx context.Context, before *config.BeforeConfig, msg *pbchat.SendMsgReq, beforeMsgData **sdkws.MsgData) error {
 	return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
@@ -204,14 +206,14 @@ func (m *msgServer) webhookAfterRevokeMsg(ctx context.Context, after *config.Aft
 	m.webhookClient.AsyncPost(ctx, callbackReq.GetCallbackCommand(), callbackReq, &cbapi.CallbackAfterRevokeMsgResp{}, after)
 }
 
-// func buildKeyMsgDataQuery(msg *sdkws.MsgData) map[string]string {
-// 	keyMsgData := apistruct.KeyMsgData{
-// 		SendID:  msg.SendID,
-// 		RecvID:  msg.RecvID,
-// 		GroupID: msg.GroupID,
-// 	}
+func buildKeyMsgDataQuery(msg *sdkws.MsgData) map[string]string {
+	keyMsgData := apistruct.KeyMsgData{
+		SendID:  msg.SendID,
+		RecvID:  msg.RecvID,
+		GroupID: msg.GroupID,
+	}
 
-// 	return map[string]string{
-// 		webhook.Key: base64.StdEncoding.EncodeToString(stringutil.StructToJsonBytes(keyMsgData)),
-// 	}
-// }
+	return map[string]string{
+		webhook.Key: base64.StdEncoding.EncodeToString(stringutil.StructToJsonBytes(keyMsgData)),
+	}
+}

internal/rpc/msg/delete.go

@@ -19,7 +19,6 @@ import (
 
 	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
 	"github.com/openimsdk/protocol/constant"
-	"github.com/openimsdk/protocol/conversation"
 	"github.com/openimsdk/protocol/msg"
 	"github.com/openimsdk/protocol/sdkws"
 	"github.com/openimsdk/tools/log"
@@ -74,7 +73,7 @@ func (m *msgServer) DeleteMsgs(ctx context.Context, req *msg.DeleteMsgsReq) (*ms
 		if err := m.MsgDatabase.DeleteMsgsPhysicalBySeqs(ctx, req.ConversationID, req.Seqs); err != nil {
 			return nil, err
 		}
-		conv, err := m.conversationClient.GetConversationsByConversationID(ctx, req.ConversationID)
+		conv, err := m.conversationClient.GetConversation(ctx, req.ConversationID, req.UserID)
 		if err != nil {
 			return nil, err
 		}
@@ -116,14 +115,12 @@ func (m *msgServer) DeleteMsgPhysical(ctx context.Context, req *msg.DeleteMsgPhy
 }
 
 func (m *msgServer) clearConversation(ctx context.Context, conversationIDs []string, userID string, deleteSyncOpt *msg.DeleteSyncOpt) error {
-	conversations, err := m.conversationClient.GetConversationsByConversationIDs(ctx, conversationIDs)
+	conversations, err := m.conversationClient.GetConversations(ctx, conversationIDs, userID)
 	if err != nil {
 		return err
 	}
-	var existConversations []*conversation.Conversation
 	var existConversationIDs []string
 	for _, conversation := range conversations {
-		existConversations = append(existConversations, conversation)
 		existConversationIDs = append(existConversationIDs, conversation.ConversationID)
 	}
 	log.ZDebug(ctx, "ClearConversationsMsg", "existConversationIDs", existConversationIDs)
@@ -152,7 +149,7 @@ func (m *msgServer) clearConversation(ctx context.Context, conversationIDs []str
 		if err := m.MsgDatabase.SetMinSeqs(ctx, m.getMinSeqs(maxSeqs)); err != nil {
 			return err
 		}
-		for _, conversation := range existConversations {
+		for _, conversation := range conversations {
 			tips := &sdkws.ClearConversationTips{UserID: userID, ConversationIDs: []string{conversation.ConversationID}}
 			m.notificationSender.NotificationWithSessionType(ctx, userID, m.conversationAndGetRecvID(conversation, userID), constant.ClearConversationNotification, conversation.ConversationType, tips)
 		}

internal/rpc/msg/send.go

@@ -86,7 +86,7 @@ func (m *msgServer) sendMsgGroupChat(ctx context.Context, req *pbmsg.SendMsgReq,
 		go m.setConversationAtInfo(ctx, req.MsgData)
 	}
 
-	// m.webhookAfterSendGroupMsg(ctx, &m.config.WebhooksConfig.AfterSendGroupMsg, req)
+	m.webhookAfterSendGroupMsg(ctx, &m.config.WebhooksConfig.AfterSendGroupMsg, req)
 
 	prommetrics.GroupChatMsgProcessSuccessCounter.Inc()
 	resp = &pbmsg.SendMsgResp{}
@@ -194,7 +194,7 @@ func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq
 			return nil, err
 		}
 
-		// m.webhookAfterSendSingleMsg(ctx, &m.config.WebhooksConfig.AfterSendSingleMsg, req)
+		m.webhookAfterSendSingleMsg(ctx, &m.config.WebhooksConfig.AfterSendSingleMsg, req)
 		prommetrics.SingleChatMsgProcessSuccessCounter.Inc()
 		return &pbmsg.SendMsgResp{
 			ServerMsgID: req.MsgData.ServerMsgID,

internal/rpc/relation/sync.go

@@ -56,7 +56,7 @@ func (s *friendServer) GetFullFriendUserIDs(ctx context.Context, req *relation.G
 		userIDs = nil
 	}
 	return &relation.GetFullFriendUserIDsResp{
-		Version:   idHash,
+		Version:   uint64(vl.Version),
 		VersionID: vl.ID.Hex(),
 		Equal:     req.IdHash == idHash,
 		UserIDs:   userIDs,

magefile.go

@@ -5,9 +5,12 @@ package main
 
 import (
 	"flag"
+	"fmt"
 	"os"
 
 	"github.com/openimsdk/gomake/mageutil"
+	"github.com/openimsdk/open-im-server/v3/version"
+	"github.com/openimsdk/tools/utils/datautil"
 )
 
 var Default = Build
@@ -25,7 +28,6 @@ var (
 	customToolsDir  = "tools"   // tools source code directory, default is "tools"
 )
 
-
 // Build support specifical binary build.
 //
 // Example: `mage build openim-api openim-rpc-user seq`
@@ -35,8 +37,7 @@ func Build() {
 	if len(bin) != 0 {
 		bin = bin[1:]
 	}
-
-	mageutil.Build(bin, nil)
+	mageutil.WithSpinner("Building binaries...", func() { mageutil.Build(bin, nil, nil) })
 }
 
 func BuildWithCustomConfig() {
@@ -53,7 +54,9 @@ func BuildWithCustomConfig() {
 		ToolsDir:  &customToolsDir,
 	}
 
-	mageutil.Build(bin, config)
+	mageutil.WithSpinner("Building binaries with custom config...", func() {
+		mageutil.Build(bin, config, nil)
+	})
 }
 
 func Start() {
@@ -70,7 +73,9 @@ func Start() {
 		bin = bin[1:]
 	}
 
-	mageutil.StartToolsAndServices(bin, nil)
+	mageutil.WithSpinner("Starting...", func() {
+		mageutil.StartToolsAndServices(bin, nil)
+	})
 }
 
 func StartWithCustomConfig() {
@@ -93,13 +98,42 @@ func StartWithCustomConfig() {
 		ConfigDir: &customConfigDir,
 	}
 
-	mageutil.StartToolsAndServices(bin, config)
+	mageutil.WithSpinner("Starting with custom config...", func() {
+		mageutil.StartToolsAndServices(bin, config)
+	})
 }
 
 func Stop() {
-	mageutil.StopAndCheckBinaries()
+	mageutil.WithSpinner("Stopping...", mageutil.StopAndCheckBinaries)
 }
 
 func Check() {
-	mageutil.CheckAndReportBinariesStatus()
+	mageutil.WithSpinner("Checking binaries...", mageutil.CheckAndReportBinariesStatus)
+}
+
+func Export() {
+	mappingPaths, err := mageutil.GetDefaultExportMappingPaths([]string{
+		"cmd",
+		"internal",
+		"pkg",
+		"test",
+		"tools",
+		"**/*.go",
+		"go.mod",
+		"go.work",
+	})
+	if err != nil {
+		mageutil.PrintRed("GetDefaultExportMappingPaths failed " + err.Error())
+		os.Exit(1)
+	}
+
+	mageutil.WithSpinner("Exporting...", func() {
+		mageutil.ExportMageLauncherArchived(mappingPaths, &mageutil.ExportOptions{
+			ProjectName: datautil.ToPtr(fmt.Sprintf("open-im-server_%s", version.Version)),
+			BuildOpt: &mageutil.BuildOptions{
+				Release:  datautil.ToPtr(true),
+				Compress: datautil.ToPtr(true),
+			},
+		})
+	})
 }

pkg/callbackstruct/constant.go

@@ -66,4 +66,5 @@ const (
 	CallbackAfterCreateSingleChatConversationsCommand  = "callbackAfterCreateSingleChatConversationsCommand"
 	CallbackBeforeCreateGroupChatConversationsCommand  = "callbackBeforeCreateGroupChatConversationsCommand"
 	CallbackAfterCreateGroupChatConversationsCommand   = "callbackAfterCreateGroupChatConversationsCommand"
+	CallbackAfterMsgSaveDBCommand                      = "callbackAfterMsgSaveDBCommand"
 )

pkg/callbackstruct/conversation.go

@@ -2,42 +2,42 @@ package callbackstruct
 
 type CallbackBeforeCreateSingleChatConversationsReq struct {
 	CallbackCommand  `json:"callbackCommand"`
-	OwnerUserID      string `json:"owner_user_id"`
-	ConversationID   string `json:"conversation_id"`
-	ConversationType int32  `json:"conversation_type"`
-	UserID           string `json:"user_id"`
-	RecvMsgOpt       int32  `json:"recv_msg_opt"`
-	IsPinned         bool   `json:"is_pinned"`
-	IsPrivateChat    bool   `json:"is_private_chat"`
-	BurnDuration     int32  `json:"burn_duration"`
-	GroupAtType      int32  `json:"group_at_type"`
-	AttachedInfo     string `json:"attached_info"`
+	OwnerUserID      string `json:"ownerUserId"`
+	ConversationID   string `json:"conversationId"`
+	ConversationType int32  `json:"conversationType"`
+	UserID           string `json:"userId"`
+	RecvMsgOpt       int32  `json:"recvMsgOpt"`
+	IsPinned         bool   `json:"isPinned"`
+	IsPrivateChat    bool   `json:"isPrivateChat"`
+	BurnDuration     int32  `json:"burnDuration"`
+	GroupAtType      int32  `json:"groupAtType"`
+	AttachedInfo     string `json:"attachedInfo"`
 	Ex               string `json:"ex"`
 }
 
 type CallbackBeforeCreateSingleChatConversationsResp struct {
 	CommonCallbackResp
-	RecvMsgOpt    *int32  `json:"recv_msg_opt"`
-	IsPinned      *bool   `json:"is_pinned"`
-	IsPrivateChat *bool   `json:"is_private_chat"`
-	BurnDuration  *int32  `json:"burn_duration"`
-	GroupAtType   *int32  `json:"group_at_type"`
-	AttachedInfo  *string `json:"attached_info"`
+	RecvMsgOpt    *int32  `json:"recvMsgOpt"`
+	IsPinned      *bool   `json:"isPinned"`
+	IsPrivateChat *bool   `json:"isPrivateChat"`
+	BurnDuration  *int32  `json:"burnDuration"`
+	GroupAtType   *int32  `json:"groupAtType"`
+	AttachedInfo  *string `json:"attachedInfo"`
 	Ex            *string `json:"ex"`
 }
 
 type CallbackAfterCreateSingleChatConversationsReq struct {
 	CallbackCommand  `json:"callbackCommand"`
-	OwnerUserID      string `json:"owner_user_id"`
-	ConversationID   string `json:"conversation_id"`
-	ConversationType int32  `json:"conversation_type"`
-	UserID           string `json:"user_id"`
-	RecvMsgOpt       int32  `json:"recv_msg_opt"`
-	IsPinned         bool   `json:"is_pinned"`
-	IsPrivateChat    bool   `json:"is_private_chat"`
-	BurnDuration     int32  `json:"burn_duration"`
-	GroupAtType      int32  `json:"group_at_type"`
-	AttachedInfo     string `json:"attached_info"`
+	OwnerUserID      string `json:"ownerUserId"`
+	ConversationID   string `json:"conversationId"`
+	ConversationType int32  `json:"conversationType"`
+	UserID           string `json:"userId"`
+	RecvMsgOpt       int32  `json:"recvMsgOpt"`
+	IsPinned         bool   `json:"isPinned"`
+	IsPrivateChat    bool   `json:"isPrivateChat"`
+	BurnDuration     int32  `json:"burnDuration"`
+	GroupAtType      int32  `json:"groupAtType"`
+	AttachedInfo     string `json:"attachedInfo"`
 	Ex               string `json:"ex"`
 }
 
@@ -47,42 +47,42 @@ type CallbackAfterCreateSingleChatConversationsResp struct {
 
 type CallbackBeforeCreateGroupChatConversationsReq struct {
 	CallbackCommand  `json:"callbackCommand"`
-	OwnerUserID      string `json:"owner_user_id"`
-	ConversationID   string `json:"conversation_id"`
-	ConversationType int32  `json:"conversation_type"`
-	GroupID          string `json:"group_id"`
-	RecvMsgOpt       int32  `json:"recv_msg_opt"`
-	IsPinned         bool   `json:"is_pinned"`
-	IsPrivateChat    bool   `json:"is_private_chat"`
-	BurnDuration     int32  `json:"burn_duration"`
-	GroupAtType      int32  `json:"group_at_type"`
-	AttachedInfo     string `json:"attached_info"`
+	OwnerUserID      string `json:"ownerUserId"`
+	ConversationID   string `json:"conversationId"`
+	ConversationType int32  `json:"conversationType"`
+	GroupID          string `json:"groupId"`
+	RecvMsgOpt       int32  `json:"recvMsgOpt"`
+	IsPinned         bool   `json:"isPinned"`
+	IsPrivateChat    bool   `json:"isPrivateChat"`
+	BurnDuration     int32  `json:"burnDuration"`
+	GroupAtType      int32  `json:"groupAtType"`
+	AttachedInfo     string `json:"attachedInfo"`
 	Ex               string `json:"ex"`
 }
 
 type CallbackBeforeCreateGroupChatConversationsResp struct {
 	CommonCallbackResp
-	RecvMsgOpt    *int32  `json:"recv_msg_opt"`
-	IsPinned      *bool   `json:"is_pinned"`
-	IsPrivateChat *bool   `json:"is_private_chat"`
-	BurnDuration  *int32  `json:"burn_duration"`
-	GroupAtType   *int32  `json:"group_at_type"`
-	AttachedInfo  *string `json:"attached_info"`
+	RecvMsgOpt    *int32  `json:"recvMsgOpt"`
+	IsPinned      *bool   `json:"isPinned"`
+	IsPrivateChat *bool   `json:"isPrivateChat"`
+	BurnDuration  *int32  `json:"burnDuration"`
+	GroupAtType   *int32  `json:"groupAtType"`
+	AttachedInfo  *string `json:"attachedInfo"`
 	Ex            *string `json:"ex"`
 }
 
 type CallbackAfterCreateGroupChatConversationsReq struct {
 	CallbackCommand  `json:"callbackCommand"`
-	OwnerUserID      string `json:"owner_user_id"`
-	ConversationID   string `json:"conversation_id"`
-	ConversationType int32  `json:"conversation_type"`
-	GroupID          string `json:"group_id"`
-	RecvMsgOpt       int32  `json:"recv_msg_opt"`
-	IsPinned         bool   `json:"is_pinned"`
-	IsPrivateChat    bool   `json:"is_private_chat"`
-	BurnDuration     int32  `json:"burn_duration"`
-	GroupAtType      int32  `json:"group_at_type"`
-	AttachedInfo     string `json:"attached_info"`
+	OwnerUserID      string `json:"ownerUserId"`
+	ConversationID   string `json:"conversationId"`
+	ConversationType int32  `json:"conversationType"`
+	GroupID          string `json:"groupId"`
+	RecvMsgOpt       int32  `json:"recvMsgOpt"`
+	IsPinned         bool   `json:"isPinned"`
+	IsPrivateChat    bool   `json:"isPrivateChat"`
+	BurnDuration     int32  `json:"burnDuration"`
+	GroupAtType      int32  `json:"groupAtType"`
+	AttachedInfo     string `json:"attachedInfo"`
 	Ex               string `json:"ex"`
 }
 

pkg/callbackstruct/message.go

@@ -103,3 +103,13 @@ type CallbackSingleMsgReadReq struct {
 type CallbackSingleMsgReadResp struct {
 	CommonCallbackResp
 }
+
+type CallbackAfterMsgSaveDBReq struct {
+	CommonCallbackReq
+	RecvID  string `json:"recvID"`
+	GroupID string `json:"groupID"`
+}
+
+type CallbackAfterMsgSaveDBResp struct {
+	CommonCallbackResp
+}

pkg/common/cmd/api.go

@@ -81,6 +81,9 @@ func (a *ApiCmd) runE() error {
 	}
 	return startrpc.Start(
 		a.ctx, &a.apiConfig.Discovery,
+		nil,
+		nil,
+		// &a.apiConfig.API.RateLimiter,
 		&prometheus,
 		a.apiConfig.API.Api.ListenIP, "",
 		a.apiConfig.API.Prometheus.AutoSetPorts,

pkg/common/cmd/auth.go

@@ -57,7 +57,7 @@ func (a *AuthRpcCmd) Exec() error {
 }
 
 func (a *AuthRpcCmd) runE() error {
-	return startrpc.Start(a.ctx, &a.authConfig.Discovery, &a.authConfig.RpcConfig.Prometheus, a.authConfig.RpcConfig.RPC.ListenIP,
+	return startrpc.Start(a.ctx, &a.authConfig.Discovery, &a.authConfig.RpcConfig.CircuitBreaker, &a.authConfig.RpcConfig.RateLimiter, &a.authConfig.RpcConfig.Prometheus, a.authConfig.RpcConfig.RPC.ListenIP,
 		a.authConfig.RpcConfig.RPC.RegisterIP, a.authConfig.RpcConfig.RPC.AutoSetPorts, a.authConfig.RpcConfig.RPC.Ports,
 		a.Index(), a.authConfig.Discovery.RpcService.Auth, nil, a.authConfig,
 		[]string{

pkg/common/cmd/conversation.go

@@ -58,7 +58,7 @@ func (a *ConversationRpcCmd) Exec() error {
 }
 
 func (a *ConversationRpcCmd) runE() error {
-	return startrpc.Start(a.ctx, &a.conversationConfig.Discovery, &a.conversationConfig.RpcConfig.Prometheus, a.conversationConfig.RpcConfig.RPC.ListenIP,
+	return startrpc.Start(a.ctx, &a.conversationConfig.Discovery, &a.conversationConfig.RpcConfig.CircuitBreaker, &a.conversationConfig.RpcConfig.RateLimiter, &a.conversationConfig.RpcConfig.Prometheus, a.conversationConfig.RpcConfig.RPC.ListenIP,
 		a.conversationConfig.RpcConfig.RPC.RegisterIP, a.conversationConfig.RpcConfig.RPC.AutoSetPorts, a.conversationConfig.RpcConfig.RPC.Ports,
 		a.Index(), a.conversationConfig.Discovery.RpcService.Conversation, &a.conversationConfig.NotificationConfig, a.conversationConfig,
 		[]string{

pkg/common/cmd/cron_task.go

@@ -56,6 +56,8 @@ func (a *CronTaskCmd) runE() error {
 	var prometheus config.Prometheus
 	return startrpc.Start(
 		a.ctx, &a.cronTaskConfig.Discovery,
+		nil,
+		nil,
 		&prometheus,
 		"", "",
 		true,

pkg/common/cmd/friend.go

@@ -58,7 +58,7 @@ func (a *FriendRpcCmd) Exec() error {
 }
 
 func (a *FriendRpcCmd) runE() error {
-	return startrpc.Start(a.ctx, &a.relationConfig.Discovery, &a.relationConfig.RpcConfig.Prometheus, a.relationConfig.RpcConfig.RPC.ListenIP,
+	return startrpc.Start(a.ctx, &a.relationConfig.Discovery, &a.relationConfig.RpcConfig.CircuitBreaker, &a.relationConfig.RpcConfig.RateLimiter, &a.relationConfig.RpcConfig.Prometheus, a.relationConfig.RpcConfig.RPC.ListenIP,
 		a.relationConfig.RpcConfig.RPC.RegisterIP, a.relationConfig.RpcConfig.RPC.AutoSetPorts, a.relationConfig.RpcConfig.RPC.Ports,
 		a.Index(), a.relationConfig.Discovery.RpcService.Friend, &a.relationConfig.NotificationConfig, a.relationConfig,
 		[]string{

pkg/common/cmd/group.go

@@ -59,7 +59,7 @@ func (a *GroupRpcCmd) Exec() error {
 }
 
 func (a *GroupRpcCmd) runE() error {
-	return startrpc.Start(a.ctx, &a.groupConfig.Discovery, &a.groupConfig.RpcConfig.Prometheus, a.groupConfig.RpcConfig.RPC.ListenIP,
+	return startrpc.Start(a.ctx, &a.groupConfig.Discovery, &a.groupConfig.RpcConfig.CircuitBreaker, &a.groupConfig.RpcConfig.RateLimiter, &a.groupConfig.RpcConfig.Prometheus, a.groupConfig.RpcConfig.RPC.ListenIP,
 		a.groupConfig.RpcConfig.RPC.RegisterIP, a.groupConfig.RpcConfig.RPC.AutoSetPorts, a.groupConfig.RpcConfig.RPC.Ports,
 		a.Index(), a.groupConfig.Discovery.RpcService.Group, &a.groupConfig.NotificationConfig, a.groupConfig,
 		[]string{

pkg/common/cmd/msg.go

@@ -59,7 +59,7 @@ func (a *MsgRpcCmd) Exec() error {
 }
 
 func (a *MsgRpcCmd) runE() error {
-	return startrpc.Start(a.ctx, &a.msgConfig.Discovery, &a.msgConfig.RpcConfig.Prometheus, a.msgConfig.RpcConfig.RPC.ListenIP,
+	return startrpc.Start(a.ctx, &a.msgConfig.Discovery, &a.msgConfig.RpcConfig.CircuitBreaker, &a.msgConfig.RpcConfig.RateLimiter, &a.msgConfig.RpcConfig.Prometheus, a.msgConfig.RpcConfig.RPC.ListenIP,
 		a.msgConfig.RpcConfig.RPC.RegisterIP, a.msgConfig.RpcConfig.RPC.AutoSetPorts, a.msgConfig.RpcConfig.RPC.Ports,
 		a.Index(), a.msgConfig.Discovery.RpcService.Msg, &a.msgConfig.NotificationConfig, a.msgConfig,
 		[]string{

pkg/common/cmd/msg_gateway.go

@@ -61,6 +61,8 @@ func (m *MsgGatewayCmd) runE() error {
 	var prometheus config.Prometheus
 	return startrpc.Start(
 		m.ctx, &m.msgGatewayConfig.Discovery,
+		&m.msgGatewayConfig.MsgGateway.CircuitBreaker,
+		&m.msgGatewayConfig.MsgGateway.RateLimiter,
 		&prometheus,
 		rpc.ListenIP, rpc.RegisterIP,
 		rpc.AutoSetPorts,

pkg/common/cmd/msg_transfer.go

@@ -62,6 +62,8 @@ func (m *MsgTransferCmd) runE() error {
 	var prometheus config.Prometheus
 	return startrpc.Start(
 		m.ctx, &m.msgTransferConfig.Discovery,
+		&m.msgTransferConfig.MsgTransfer.CircuitBreaker,
+		&m.msgTransferConfig.MsgTransfer.RateLimiter,
 		&prometheus,
 		"", "",
 		true,

pkg/common/cmd/push.go

@@ -60,7 +60,7 @@ func (a *PushRpcCmd) Exec() error {
 }
 
 func (a *PushRpcCmd) runE() error {
-	return startrpc.Start(a.ctx, &a.pushConfig.Discovery, &a.pushConfig.RpcConfig.Prometheus, a.pushConfig.RpcConfig.RPC.ListenIP,
+	return startrpc.Start(a.ctx, &a.pushConfig.Discovery, &a.pushConfig.RpcConfig.CircuitBreaker, &a.pushConfig.RpcConfig.RateLimiter, &a.pushConfig.RpcConfig.Prometheus, a.pushConfig.RpcConfig.RPC.ListenIP,
 		a.pushConfig.RpcConfig.RPC.RegisterIP, a.pushConfig.RpcConfig.RPC.AutoSetPorts, a.pushConfig.RpcConfig.RPC.Ports,
 		a.Index(), a.pushConfig.Discovery.RpcService.Push, &a.pushConfig.NotificationConfig, a.pushConfig,
 		[]string{

pkg/common/cmd/third.go

@@ -58,7 +58,7 @@ func (a *ThirdRpcCmd) Exec() error {
 }
 
 func (a *ThirdRpcCmd) runE() error {
-	return startrpc.Start(a.ctx, &a.thirdConfig.Discovery, &a.thirdConfig.RpcConfig.Prometheus, a.thirdConfig.RpcConfig.RPC.ListenIP,
+	return startrpc.Start(a.ctx, &a.thirdConfig.Discovery, &a.thirdConfig.RpcConfig.CircuitBreaker, &a.thirdConfig.RpcConfig.RateLimiter, &a.thirdConfig.RpcConfig.Prometheus, a.thirdConfig.RpcConfig.RPC.ListenIP,
 		a.thirdConfig.RpcConfig.RPC.RegisterIP, a.thirdConfig.RpcConfig.RPC.AutoSetPorts, a.thirdConfig.RpcConfig.RPC.Ports,
 		a.Index(), a.thirdConfig.Discovery.RpcService.Third, &a.thirdConfig.NotificationConfig, a.thirdConfig,
 		[]string{

pkg/common/cmd/user.go

@@ -59,7 +59,7 @@ func (a *UserRpcCmd) Exec() error {
 }
 
 func (a *UserRpcCmd) runE() error {
-	return startrpc.Start(a.ctx, &a.userConfig.Discovery, &a.userConfig.RpcConfig.Prometheus, a.userConfig.RpcConfig.RPC.ListenIP,
+	return startrpc.Start(a.ctx, &a.userConfig.Discovery, &a.userConfig.RpcConfig.CircuitBreaker, &a.userConfig.RpcConfig.RateLimiter, &a.userConfig.RpcConfig.Prometheus, a.userConfig.RpcConfig.RPC.ListenIP,
 		a.userConfig.RpcConfig.RPC.RegisterIP, a.userConfig.RpcConfig.RPC.AutoSetPorts, a.userConfig.RpcConfig.RPC.Ports,
 		a.Index(), a.userConfig.Discovery.RpcService.User, &a.userConfig.NotificationConfig, a.userConfig,
 		[]string{

pkg/common/config/config.go

@@ -143,6 +143,23 @@ type API struct {
 		Ports        []int  `yaml:"ports"`
 		GrafanaURL   string `yaml:"grafanaURL"`
 	} `yaml:"prometheus"`
+
+	RateLimiter RateLimiter `yaml:"rateLimiter"`
+}
+
+type RateLimiter struct {
+	Enable       bool          `yaml:"enable"`
+	Window       time.Duration `yaml:"window"`
+	Bucket       int           `yaml:"bucket"`
+	CPUThreshold int64         `yaml:"cpuThreshold"`
+}
+
+type CircuitBreaker struct {
+	Enable  bool          `yaml:"enable"`
+	Window  time.Duration `yaml:"window"`
+	Bucket  int           `yaml:"bucket"`
+	Success float64       `yaml:"success"`
+	Request int64         `yaml:"request"`
 }
 
 type CronTask struct {
@@ -217,6 +234,8 @@ type MsgGateway struct {
 		WebsocketMaxMsgLen  int   `yaml:"websocketMaxMsgLen"`
 		WebsocketTimeout    int   `yaml:"websocketTimeout"`
 	} `yaml:"longConnSvr"`
+	RateLimiter    RateLimiter    `yaml:"rateLimiter"`
+	CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"`
 }
 
 type MsgTransfer struct {
@@ -225,6 +244,8 @@ type MsgTransfer struct {
 		AutoSetPorts bool  `yaml:"autoSetPorts"`
 		Ports        []int `yaml:"ports"`
 	} `yaml:"prometheus"`
+	RateLimiter    RateLimiter    `yaml:"rateLimiter"`
+	CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"`
 }
 
 type Push struct {
@@ -255,7 +276,9 @@ type Push struct {
 		BadgeCount bool   `yaml:"badgeCount"`
 		Production bool   `yaml:"production"`
 	} `yaml:"iosPush"`
-	FullUserCache bool `yaml:"fullUserCache"`
+	FullUserCache  bool           `yaml:"fullUserCache"`
+	RateLimiter    RateLimiter    `yaml:"rateLimiter"`
+	CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"`
 }
 
 type Auth struct {
@@ -264,28 +287,38 @@ type Auth struct {
 	TokenPolicy struct {
 		Expire int64 `yaml:"expire"`
 	} `yaml:"tokenPolicy"`
+	RateLimiter    RateLimiter    `yaml:"rateLimiter"`
+	CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"`
 }
 
 type Conversation struct {
-	RPC        RPC        `yaml:"rpc"`
-	Prometheus Prometheus `yaml:"prometheus"`
+	RPC            RPC            `yaml:"rpc"`
+	Prometheus     Prometheus     `yaml:"prometheus"`
+	RateLimiter    RateLimiter    `yaml:"rateLimiter"`
+	CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"`
 }
 
 type Friend struct {
-	RPC        RPC        `yaml:"rpc"`
-	Prometheus Prometheus `yaml:"prometheus"`
+	RPC            RPC            `yaml:"rpc"`
+	Prometheus     Prometheus     `yaml:"prometheus"`
+	RateLimiter    RateLimiter    `yaml:"rateLimiter"`
+	CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"`
 }
 
 type Group struct {
-	RPC                        RPC        `yaml:"rpc"`
-	Prometheus                 Prometheus `yaml:"prometheus"`
-	EnableHistoryForNewMembers bool       `yaml:"enableHistoryForNewMembers"`
+	RPC                        RPC            `yaml:"rpc"`
+	Prometheus                 Prometheus     `yaml:"prometheus"`
+	EnableHistoryForNewMembers bool           `yaml:"enableHistoryForNewMembers"`
+	RateLimiter                RateLimiter    `yaml:"rateLimiter"`
+	CircuitBreaker             CircuitBreaker `yaml:"circuitBreaker"`
 }
 
 type Msg struct {
-	RPC          RPC        `yaml:"rpc"`
-	Prometheus   Prometheus `yaml:"prometheus"`
-	FriendVerify bool       `yaml:"friendVerify"`
+	RPC            RPC            `yaml:"rpc"`
+	Prometheus     Prometheus     `yaml:"prometheus"`
+	FriendVerify   bool           `yaml:"friendVerify"`
+	RateLimiter    RateLimiter    `yaml:"rateLimiter"`
+	CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"`
 }
 
 type Third struct {
@@ -298,6 +331,8 @@ type Third struct {
 		Kodo   Kodo   `yaml:"kodo"`
 		Aws    Aws    `yaml:"aws"`
 	} `yaml:"object"`
+	RateLimiter    RateLimiter    `yaml:"rateLimiter"`
+	CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"`
 }
 type Cos struct {
 	BucketURL    string `yaml:"bucketURL"`
@@ -336,8 +371,10 @@ type Aws struct {
 }
 
 type User struct {
-	RPC        RPC        `yaml:"rpc"`
-	Prometheus Prometheus `yaml:"prometheus"`
+	RPC            RPC            `yaml:"rpc"`
+	Prometheus     Prometheus     `yaml:"prometheus"`
+	RateLimiter    RateLimiter    `yaml:"rateLimiter"`
+	CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"`
 }
 
 type RPC struct {
@@ -436,6 +473,7 @@ type Webhooks struct {
 	BeforeSendGroupMsg                  BeforeConfig `yaml:"beforeSendGroupMsg"`
 	BeforeMsgModify                     BeforeConfig `yaml:"beforeMsgModify"`
 	AfterSendGroupMsg                   AfterConfig  `yaml:"afterSendGroupMsg"`
+	AfterMsgSaveDB                      AfterConfig  `yaml:"afterMsgSaveDB"`
 	AfterUserOnline                     AfterConfig  `yaml:"afterUserOnline"`
 	AfterUserOffline                    AfterConfig  `yaml:"afterUserOffline"`
 	AfterUserKickOff                    AfterConfig  `yaml:"afterUserKickOff"`

pkg/common/servererrs/code.go

@@ -70,6 +70,7 @@ const (
 	BlockedByPeer            = 1302 // Blocked by the peer
 	NotPeersFriend           = 1303 // Not the peer's friend
 	RelationshipAlreadyError = 1304 // Already in a friend relationship
+	FriendRequestHandled     = 1305 // Friend request has already been handled
 
 	// Message error codes.
 	MessageHasReadDisable = 1401

pkg/common/servererrs/predefine.go

@@ -51,10 +51,11 @@ var (
 
 	ErrMessageHasReadDisable = errs.NewCodeError(MessageHasReadDisable, "MessageHasReadDisable")
 
-	ErrCanNotAddYourself   = errs.NewCodeError(CanNotAddYourselfError, "CanNotAddYourselfError")
-	ErrBlockedByPeer       = errs.NewCodeError(BlockedByPeer, "BlockedByPeer")
-	ErrNotPeersFriend      = errs.NewCodeError(NotPeersFriend, "NotPeersFriend")
-	ErrRelationshipAlready = errs.NewCodeError(RelationshipAlreadyError, "RelationshipAlreadyError")
+	ErrCanNotAddYourself    = errs.NewCodeError(CanNotAddYourselfError, "CanNotAddYourselfError")
+	ErrBlockedByPeer        = errs.NewCodeError(BlockedByPeer, "BlockedByPeer")
+	ErrNotPeersFriend       = errs.NewCodeError(NotPeersFriend, "NotPeersFriend")
+	ErrRelationshipAlready  = errs.NewCodeError(RelationshipAlreadyError, "RelationshipAlreadyError")
+	ErrFriendRequestHandled = errs.NewCodeError(FriendRequestHandled, "FriendRequestHandled")
 
 	ErrMutedInGroup     = errs.NewCodeError(MutedInGroup, "MutedInGroup")
 	ErrMutedGroup       = errs.NewCodeError(MutedGroup, "MutedGroup")

pkg/common/startrpc/circuitbreaker.go

@@ -0,0 +1,107 @@
+package startrpc
+
+import (
+	"context"
+	"time"
+
+	"github.com/openimsdk/tools/log"
+	"github.com/openimsdk/tools/stability/circuitbreaker"
+	"github.com/openimsdk/tools/stability/circuitbreaker/sre"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+type CircuitBreaker struct {
+	Enable  bool          `yaml:"enable"`
+	Success float64       `yaml:"success"` // success rate threshold (0.0-1.0)
+	Request int64         `yaml:"request"` // request threshold
+	Bucket  int           `yaml:"bucket"`  // number of buckets
+	Window  time.Duration `yaml:"window"`  // time window for statistics
+}
+
+func NewCircuitBreaker(config *CircuitBreaker) circuitbreaker.CircuitBreaker {
+	if !config.Enable {
+		return nil
+	}
+
+	return sre.NewSREBraker(
+		sre.WithWindow(config.Window),
+		sre.WithBucket(config.Bucket),
+		sre.WithSuccess(config.Success),
+		sre.WithRequest(config.Request),
+	)
+}
+
+func UnaryCircuitBreakerInterceptor(breaker circuitbreaker.CircuitBreaker) grpc.ServerOption {
+	if breaker == nil {
+		return grpc.ChainUnaryInterceptor(func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp any, err error) {
+			return handler(ctx, req)
+		})
+	}
+
+	return grpc.ChainUnaryInterceptor(func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp any, err error) {
+		if err := breaker.Allow(); err != nil {
+			log.ZWarn(ctx, "rpc circuit breaker open", err, "method", info.FullMethod)
+			return nil, status.Error(codes.Unavailable, "service unavailable due to circuit breaker")
+		}
+
+		resp, err = handler(ctx, req)
+
+		if err != nil {
+			if st, ok := status.FromError(err); ok {
+				switch st.Code() {
+				case codes.OK:
+					breaker.MarkSuccess()
+				case codes.InvalidArgument, codes.NotFound, codes.AlreadyExists, codes.PermissionDenied:
+					breaker.MarkSuccess()
+				default:
+					breaker.MarkFailed()
+				}
+			} else {
+				breaker.MarkFailed()
+			}
+		} else {
+			breaker.MarkSuccess()
+		}
+
+		return resp, err
+
+	})
+}
+
+func StreamCircuitBreakerInterceptor(breaker circuitbreaker.CircuitBreaker) grpc.ServerOption {
+	if breaker == nil {
+		return grpc.ChainStreamInterceptor(func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+			return handler(srv, ss)
+		})
+	}
+
+	return grpc.ChainStreamInterceptor(func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+		if err := breaker.Allow(); err != nil {
+			log.ZWarn(ss.Context(), "rpc circuit breaker open", err, "method", info.FullMethod)
+			return status.Error(codes.Unavailable, "service unavailable due to circuit breaker")
+		}
+
+		err := handler(srv, ss)
+
+		if err != nil {
+			if st, ok := status.FromError(err); ok {
+				switch st.Code() {
+				case codes.OK:
+					breaker.MarkSuccess()
+				case codes.InvalidArgument, codes.NotFound, codes.AlreadyExists, codes.PermissionDenied:
+					breaker.MarkSuccess()
+				default:
+					breaker.MarkFailed()
+				}
+			} else {
+				breaker.MarkFailed()
+			}
+		} else {
+			breaker.MarkSuccess()
+		}
+
+		return err
+	})
+}

pkg/common/startrpc/ratelimit.go

@@ -0,0 +1,70 @@
+package startrpc
+
+import (
+	"context"
+	"time"
+
+	"github.com/openimsdk/tools/log"
+	"github.com/openimsdk/tools/stability/ratelimit"
+	"github.com/openimsdk/tools/stability/ratelimit/bbr"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+type RateLimiter struct {
+	Enable       bool
+	Window       time.Duration
+	Bucket       int
+	CPUThreshold int64
+}
+
+func NewRateLimiter(config *RateLimiter) ratelimit.Limiter {
+	if !config.Enable {
+		return nil
+	}
+
+	return bbr.NewBBRLimiter(
+		bbr.WithWindow(config.Window),
+		bbr.WithBucket(config.Bucket),
+		bbr.WithCPUThreshold(config.CPUThreshold),
+	)
+}
+
+func UnaryRateLimitInterceptor(limiter ratelimit.Limiter) grpc.ServerOption {
+	if limiter == nil {
+		return grpc.ChainUnaryInterceptor(func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp any, err error) {
+			return handler(ctx, req)
+		})
+	}
+
+	return grpc.ChainUnaryInterceptor(func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp any, err error) {
+		done, err := limiter.Allow()
+		if err != nil {
+			log.ZWarn(ctx, "rpc rate limited", err, "method", info.FullMethod)
+			return nil, status.Errorf(codes.ResourceExhausted, "rpc request rate limit exceeded: %v, please try again later", err)
+		}
+
+		defer done(ratelimit.DoneInfo{})
+		return handler(ctx, req)
+	})
+}
+
+func StreamRateLimitInterceptor(limiter ratelimit.Limiter) grpc.ServerOption {
+	if limiter == nil {
+		return grpc.ChainStreamInterceptor(func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+			return handler(srv, ss)
+		})
+	}
+
+	return grpc.ChainStreamInterceptor(func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+		done, err := limiter.Allow()
+		if err != nil {
+			log.ZWarn(ss.Context(), "rpc rate limited", err, "method", info.FullMethod)
+			return status.Errorf(codes.ResourceExhausted, "rpc request rate limit exceeded: %v, please try again later", err)
+		}
+		defer done(ratelimit.DoneInfo{})
+
+		return handler(srv, ss)
+	})
+}

pkg/common/startrpc/start.go

@@ -47,7 +47,7 @@ func init() {
 	prommetrics.RegistryAll()
 }
 
-func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *conf.Prometheus, listenIP,
+func Start[T any](ctx context.Context, disc *conf.Discovery, circuitBreakerConfig *conf.CircuitBreaker, rateLimiterConfig *conf.RateLimiter, prometheusConfig *conf.Prometheus, listenIP,
 	registerIP string, autoSetPorts bool, rpcPorts []int, index int, rpcRegisterName string, notification *conf.Notification, config T,
 	watchConfigNames []string, watchServiceNames []string,
 	rpcFn func(ctx context.Context, config T, client discovery.SvcDiscoveryRegistry, server grpc.ServiceRegistrar) error,
@@ -84,6 +84,45 @@ func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *c
 		}
 	}
 
+	if circuitBreakerConfig != nil && circuitBreakerConfig.Enable {
+		cb := &CircuitBreaker{
+			Enable:  circuitBreakerConfig.Enable,
+			Success: circuitBreakerConfig.Success,
+			Request: circuitBreakerConfig.Request,
+			Bucket:  circuitBreakerConfig.Bucket,
+			Window:  circuitBreakerConfig.Window,
+		}
+
+		breaker := NewCircuitBreaker(cb)
+
+		options = append(options,
+			UnaryCircuitBreakerInterceptor(breaker),
+			StreamCircuitBreakerInterceptor(breaker),
+		)
+
+		log.ZInfo(ctx, "RPC circuit breaker enabled",
+			"service", rpcRegisterName,
+			"window", circuitBreakerConfig.Window,
+			"bucket", circuitBreakerConfig.Bucket,
+			"success", circuitBreakerConfig.Success,
+			"requestThreshold", circuitBreakerConfig.Request)
+	}
+
+	if rateLimiterConfig != nil && rateLimiterConfig.Enable {
+		limiter := NewRateLimiter((*RateLimiter)(rateLimiterConfig))
+
+		options = append(options,
+			UnaryRateLimitInterceptor(limiter),
+			StreamRateLimitInterceptor(limiter),
+		)
+
+		log.ZInfo(ctx, "RPC rate limiter enabled",
+			"service", rpcRegisterName,
+			"window", rateLimiterConfig.Window,
+			"bucket", rateLimiterConfig.Bucket,
+			"cpuThreshold", rateLimiterConfig.CPUThreshold)
+	}
+
 	registerIP, err := network.GetRpcRegisterIP(registerIP)
 	if err != nil {
 		return err
@@ -123,7 +162,7 @@ func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *c
 
 	go func() {
 		sigs := make(chan os.Signal, 1)
-		signal.Notify(sigs, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL)
+		signal.Notify(sigs, syscall.SIGTERM, syscall.SIGINT)
 		select {
 		case <-ctx.Done():
 			return

pkg/common/storage/cache/conversation.go

@@ -16,6 +16,7 @@ package cache
 
 import (
 	"context"
+
 	relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 )
 
@@ -57,7 +58,7 @@ type ConversationCache interface {
 	GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error)
 	DelConversationNotReceiveMessageUserIDs(conversationIDs ...string) ConversationCache
 	DelConversationNotNotifyMessageUserIDs(userIDs ...string) ConversationCache
-	DelConversationPinnedMessageUserIDs(userIDs ...string) ConversationCache
+	DelUserPinnedConversations(userIDs ...string) ConversationCache
 	DelConversationVersionUserIDs(userIDs ...string) ConversationCache
 
 	FindMaxConversationUserVersion(ctx context.Context, userID string) (*relationtb.VersionLog, error)

pkg/common/storage/cache/redis/conversation.go

@@ -253,7 +253,7 @@ func (c *ConversationRedisCache) DelConversationNotNotifyMessageUserIDs(userIDs
 	return cache
 }
 
-func (c *ConversationRedisCache) DelConversationPinnedMessageUserIDs(userIDs ...string) cache.ConversationCache {
+func (c *ConversationRedisCache) DelUserPinnedConversations(userIDs ...string) cache.ConversationCache {
 	cache := c.CloneConversationCache()
 	for _, userID := range userIDs {
 		cache.AddKeys(c.getPinnedConversationIDsKey(userID))

pkg/common/storage/cache/redis/token.go

@@ -220,7 +220,6 @@ func (c *tokenCache) DeleteAndSetTemporary(ctx context.Context, userID string, p
 	if err := c.rdb.HDel(ctx, key, fields...).Err(); err != nil {
 		return errs.Wrap(err)
 	}
-
 	if c.localCache != nil {
 		c.removeLocalTokenCache(ctx, key)
 	}

pkg/common/storage/controller/conversation.go

@@ -61,8 +61,6 @@ type ConversationDatabase interface {
 	GetAllConversationIDsNumber(ctx context.Context) (int64, error)
 	// PageConversationIDs paginates through conversation IDs based on the specified pagination settings.
 	PageConversationIDs(ctx context.Context, pagination pagination.Pagination) (conversationIDs []string, err error)
-	// GetConversationsByConversationID retrieves conversations by their IDs.
-	GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*relationtb.Conversation, error)
 	// GetConversationIDsNeedDestruct fetches conversations that need to be destructed based on specific criteria.
 	GetConversationIDsNeedDestruct(ctx context.Context) ([]*relationtb.Conversation, error)
 	// GetConversationNotReceiveMessageUserIDs gets user IDs for users in a conversation who have not received messages.
@@ -78,6 +76,8 @@ type ConversationDatabase interface {
 	GetPinnedConversationIDs(ctx context.Context, userID string) ([]string, error)
 	// FindRandConversation finds random conversations based on the specified timestamp and limit.
 	FindRandConversation(ctx context.Context, ts int64, limit int) ([]*relationtb.Conversation, error)
+
+	DeleteUsersConversations(ctx context.Context, userID string, conversationIDs []string) (err error)
 }
 
 func NewConversationDatabase(conversation database.Conversation, cache cache.ConversationCache, tx tx.Tx) ConversationDatabase {
@@ -120,7 +120,7 @@ func (c *conversationDatabase) SetUsersConversationFieldTx(ctx context.Context,
 				cache = cache.DelConversationNotNotifyMessageUserIDs(userIDs...)
 			}
 			if _, ok := fieldMap["is_pinned"]; ok {
-				cache = cache.DelConversationPinnedMessageUserIDs(userIDs...)
+				cache = cache.DelUserPinnedConversations(userIDs...)
 			}
 			cache = cache.DelConversationVersionUserIDs(haveUserIDs...)
 		}
@@ -172,7 +172,7 @@ func (c *conversationDatabase) UpdateUsersConversationField(ctx context.Context,
 		cache = cache.DelConversationNotNotifyMessageUserIDs(userIDs...)
 	}
 	if _, ok := args["is_pinned"]; ok {
-		cache = cache.DelConversationPinnedMessageUserIDs(userIDs...)
+		cache = cache.DelUserPinnedConversations(userIDs...)
 	}
 	return cache.ChainExecDel(ctx)
 }
@@ -203,7 +203,7 @@ func (c *conversationDatabase) CreateConversation(ctx context.Context, conversat
 		DelUserConversationIDsHash(userIDs...).
 		DelConversationVersionUserIDs(userIDs...).
 		DelConversationNotNotifyMessageUserIDs(notNotifyUserIDs...).
-		DelConversationPinnedMessageUserIDs(pinnedUserIDs...).
+		DelUserPinnedConversations(pinnedUserIDs...).
 		ChainExecDel(ctx)
 }
 
@@ -259,7 +259,7 @@ func (c *conversationDatabase) SetUserConversations(ctx context.Context, ownerUs
 		cache := c.cache.CloneConversationCache()
 		cache = cache.DelConversationVersionUserIDs(ownerUserID).
 			DelConversationNotNotifyMessageUserIDs(ownerUserID).
-			DelConversationPinnedMessageUserIDs(ownerUserID)
+			DelUserPinnedConversations(ownerUserID)
 
 		groupIDs := datautil.Distinct(datautil.Filter(conversations, func(e *relationtb.Conversation) (string, bool) {
 			return e.GroupID, e.GroupID != ""
@@ -373,10 +373,6 @@ func (c *conversationDatabase) PageConversationIDs(ctx context.Context, paginati
 	return c.conversationDB.PageConversationIDs(ctx, pagination)
 }
 
-func (c *conversationDatabase) GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*relationtb.Conversation, error) {
-	return c.conversationDB.GetConversationsByConversationID(ctx, conversationIDs)
-}
-
 func (c *conversationDatabase) GetConversationIDsNeedDestruct(ctx context.Context) ([]*relationtb.Conversation, error) {
 	return c.conversationDB.GetConversationIDsNeedDestruct(ctx)
 }
@@ -429,3 +425,21 @@ func (c *conversationDatabase) GetPinnedConversationIDs(ctx context.Context, use
 func (c *conversationDatabase) FindRandConversation(ctx context.Context, ts int64, limit int) ([]*relationtb.Conversation, error) {
 	return c.conversationDB.FindRandConversation(ctx, ts, limit)
 }
+
+func (c *conversationDatabase) DeleteUsersConversations(ctx context.Context, userID string, conversationIDs []string) (err error) {
+	return c.tx.Transaction(ctx, func(ctx context.Context) error {
+		err = c.conversationDB.DeleteUsersConversations(ctx, userID, conversationIDs)
+		if err != nil {
+			return err
+		}
+		cache := c.cache.CloneConversationCache()
+		cache = cache.DelConversations(userID, conversationIDs...).
+			DelConversationVersionUserIDs(userID).
+			DelConversationIDs(userID).
+			DelUserConversationIDsHash(userID).
+			DelConversationNotNotifyMessageUserIDs(userID).
+			DelUserPinnedConversations(userID)
+
+		return cache.ChainExecDel(ctx)
+	})
+}

pkg/common/storage/controller/friend.go

@@ -16,9 +16,9 @@ package controller
 
 import (
 	"context"
-	"fmt"
 	"time"
 
+	"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
@@ -109,15 +109,13 @@ func (f *friendDatabase) CheckIn(ctx context.Context, userID1, userID2 string) (
 	// Retrieve friend IDs of userID1 from the cache
 	userID1FriendIDs, err := f.cache.GetFriendIDs(ctx, userID1)
 	if err != nil {
-		err = fmt.Errorf("error retrieving friend IDs for user %s: %w", userID1, err)
-		return
+		return false, false, err
 	}
 
 	// Retrieve friend IDs of userID2 from the cache
 	userID2FriendIDs, err := f.cache.GetFriendIDs(ctx, userID2)
 	if err != nil {
-		err = fmt.Errorf("error retrieving friend IDs for user %s: %w", userID2, err)
-		return
+		return false, false, err
 	}
 
 	// Check if userID2 is in userID1's friend list and vice versa
@@ -214,12 +212,12 @@ func (f *friendDatabase) RefuseFriendRequest(ctx context.Context, friendRequest
 	// Attempt to retrieve the friend request from the database.
 	fr, err := f.friendRequest.Take(ctx, friendRequest.FromUserID, friendRequest.ToUserID)
 	if err != nil {
-		return fmt.Errorf("failed to retrieve friend request from %s to %s: %w", friendRequest.FromUserID, friendRequest.ToUserID, err)
+		return err
 	}
 
 	// Check if the friend request has already been handled.
 	if fr.HandleResult != 0 {
-		return fmt.Errorf("friend request from %s to %s has already been processed", friendRequest.FromUserID, friendRequest.ToUserID)
+		return servererrs.ErrFriendRequestHandled.WrapMsg("friend request has already been processed", "from", friendRequest.FromUserID, "to", friendRequest.ToUserID)
 	}
 
 	// Log the action of refusing the friend request for debugging and auditing purposes.
@@ -232,7 +230,7 @@ func (f *friendDatabase) RefuseFriendRequest(ctx context.Context, friendRequest
 	friendRequest.HandleResult = constant.FriendResponseRefuse
 	friendRequest.HandleTime = time.Now()
 	if err := f.friendRequest.Update(ctx, friendRequest); err != nil {
-		return fmt.Errorf("failed to update friend request from %s to %s as refused: %w", friendRequest.FromUserID, friendRequest.ToUserID, err)
+		return err
 	}
 
 	return nil
@@ -350,9 +348,9 @@ func (f *friendDatabase) PageFriendRequestToMe(ctx context.Context, userID strin
 func (f *friendDatabase) FindFriendsWithError(ctx context.Context, ownerUserID string, friendUserIDs []string) (friends []*model.Friend, err error) {
 	friends, err = f.friend.FindFriends(ctx, ownerUserID, friendUserIDs)
 	if err != nil {
-		return
+		return nil, err
 	}
-	return
+	return friends, nil
 }
 
 func (f *friendDatabase) FindFriendUserIDs(ctx context.Context, ownerUserID string) (friendUserIDs []string, err error) {

pkg/common/storage/database/conversation.go

@@ -39,9 +39,9 @@ type Conversation interface {
 	GetAllConversationIDs(ctx context.Context) ([]string, error)
 	GetAllConversationIDsNumber(ctx context.Context) (int64, error)
 	PageConversationIDs(ctx context.Context, pagination pagination.Pagination) (conversationIDs []string, err error)
-	GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*model.Conversation, error)
 	GetConversationIDsNeedDestruct(ctx context.Context) ([]*model.Conversation, error)
 	GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error)
 	FindConversationUserVersion(ctx context.Context, userID string, version uint, limit int) (*model.VersionLog, error)
 	FindRandConversation(ctx context.Context, ts int64, limit int) ([]*model.Conversation, error)
+	DeleteUsersConversations(ctx context.Context, userID string, conversationIDs []string) (err error)
 }

pkg/common/storage/database/mgo/conversation.go

@@ -47,6 +47,12 @@ func NewConversationMongo(db *mongo.Database) (*ConversationMgo, error) {
 			},
 			Options: options.Index(),
 		},
+		{
+			Keys: bson.D{
+				{Key: "conversation_id", Value: 1},
+			},
+			Options: options.Index().SetUnique(true),
+		},
 	})
 	if err != nil {
 		return nil, errs.Wrap(err)
@@ -232,10 +238,6 @@ func (c *ConversationMgo) PageConversationIDs(ctx context.Context, pagination pa
 	return mongoutil.FindPageOnly[string](ctx, c.coll, bson.M{}, pagination, options.Find().SetProjection(bson.M{"conversation_id": 1}))
 }
 
-func (c *ConversationMgo) GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*model.Conversation, error) {
-	return mongoutil.Find[*model.Conversation](ctx, c.coll, bson.M{"conversation_id": bson.M{"$in": conversationIDs}})
-}
-
 func (c *ConversationMgo) GetConversationIDsNeedDestruct(ctx context.Context) ([]*model.Conversation, error) {
 	// "is_msg_destruct = 1 && msg_destruct_time != 0 && (UNIX_TIMESTAMP(NOW()) > (msg_destruct_time + UNIX_TIMESTAMP(latest_msg_destruct_time)) || latest_msg_destruct_time is NULL)"
 	return mongoutil.Find[*model.Conversation](ctx, c.coll, bson.M{
@@ -308,3 +310,20 @@ func (c *ConversationMgo) FindRandConversation(ctx context.Context, ts int64, li
 	}
 	return mongoutil.Aggregate[*model.Conversation](ctx, c.coll, pipeline)
 }
+
+func (c *ConversationMgo) DeleteUsersConversations(ctx context.Context, userID string, conversationIDs []string) (err error) {
+	if len(conversationIDs) == 0 {
+		return nil
+	}
+	return mongoutil.IncrVersion(func() error {
+		err := mongoutil.DeleteMany(ctx, c.coll, bson.M{"owner_user_id": userID, "conversation_id": bson.M{"$in": conversationIDs}})
+		return err
+	}, func() error {
+		for _, conversationID := range conversationIDs {
+			if err := c.version.IncrVersion(ctx, userID, []string{conversationID}, model.VersionStateDelete); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+}

pkg/common/storage/database/mgo/friend_request.go

@@ -137,7 +137,7 @@ func (f *FriendRequestMgo) Take(ctx context.Context, fromUserID, toUserID string
 func (f *FriendRequestMgo) GetUnhandledCount(ctx context.Context, userID string, ts int64) (int64, error) {
 	filter := bson.M{"to_user_id": userID, "handle_result": 0}
 	if ts != 0 {
-		filter["create_time"] = bson.M{"$gt": time.Unix(ts, 0)}
+		filter["create_time"] = bson.M{"$gt": time.UnixMilli(ts)}
 	}
 	return mongoutil.Count(ctx, f.coll, filter)
 }

pkg/common/storage/database/mgo/group_request.go

@@ -109,7 +109,7 @@ func (g *GroupRequestMgo) GetUnhandledCount(ctx context.Context, groupIDs []stri
 	}
 	filter := bson.M{"group_id": bson.M{"$in": groupIDs}, "handle_result": 0}
 	if ts != 0 {
-		filter["req_time"] = bson.M{"$gt": time.Unix(ts, 0)}
+		filter["req_time"] = bson.M{"$gt": time.UnixMilli(ts)}
 	}
 	return mongoutil.Count(ctx, g.coll, filter)
 }

pkg/common/storage/database/mgo/seq_conversation.go

@@ -57,8 +57,8 @@ func (s *seqConversationMongo) Malloc(ctx context.Context, conversationID string
 	}
 	filter := map[string]any{"conversation_id": conversationID}
 	update := map[string]any{
-		"$inc": map[string]any{"max_seq": size},
-		"$set": map[string]any{"min_seq": int64(0)},
+		"$inc":         map[string]any{"max_seq": size},
+		"$setOnInsert": map[string]any{"min_seq": int64(0)},
 	}
 	opt := options.FindOneAndUpdate().SetUpsert(true).SetReturnDocument(options.After).SetProjection(map[string]any{"_id": 0, "max_seq": 1})
 	lastSeq, err := mongoutil.FindOneAndUpdate[int64](ctx, s.coll, filter, update, opt)

pkg/common/storage/database/mgo/user.go

@@ -73,7 +73,7 @@ func (u *UserMgo) TakeNotification(ctx context.Context, level int64) (user []*mo
 }
 
 func (u *UserMgo) TakeGTEAppManagerLevel(ctx context.Context, level int64) (user []*model.User, err error) {
-	return mongoutil.Find[*model.User](ctx, u.coll, bson.M{"app_manager_level": bson.M{"$gte": level}})
+	return mongoutil.Find[*model.User](ctx, u.coll, bson.M{"app_manger_level": bson.M{"$gte": level}})
 }
 
 func (u *UserMgo) TakeByNickname(ctx context.Context, nickname string) (user []*model.User, err error) {

pkg/common/storage/kafka/config.go

@@ -0,0 +1,33 @@
+// Copyright © 2024 OpenIM open source community. 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 kafka
+
+type TLSConfig struct {
+	EnableTLS          bool   `yaml:"enableTLS"`
+	CACrt              string `yaml:"caCrt"`
+	ClientCrt          string `yaml:"clientCrt"`
+	ClientKey          string `yaml:"clientKey"`
+	ClientKeyPwd       string `yaml:"clientKeyPwd"`
+	InsecureSkipVerify bool   `yaml:"insecureSkipVerify"`
+}
+
+type Config struct {
+	Username     string    `yaml:"username"`
+	Password     string    `yaml:"password"`
+	ProducerAck  string    `yaml:"producerAck"`
+	CompressType string    `yaml:"compressType"`
+	Addr         []string  `yaml:"addr"`
+	TLS          TLSConfig `yaml:"tls"`
+}

pkg/common/storage/kafka/consumer_group.go

@@ -0,0 +1,68 @@
+// 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 kafka
+
+import (
+	"context"
+	"errors"
+
+	"github.com/IBM/sarama"
+	"github.com/openimsdk/tools/log"
+)
+
+type MConsumerGroup struct {
+	sarama.ConsumerGroup
+	groupID string
+	topics  []string
+}
+
+func NewMConsumerGroup(conf *Config, groupID string, topics []string, autoCommitEnable bool) (*MConsumerGroup, error) {
+	config, err := BuildConsumerGroupConfig(conf, sarama.OffsetNewest, autoCommitEnable)
+	if err != nil {
+		return nil, err
+	}
+	group, err := NewConsumerGroup(config, conf.Addr, groupID)
+	if err != nil {
+		return nil, err
+	}
+	return &MConsumerGroup{
+		ConsumerGroup: group,
+		groupID:       groupID,
+		topics:        topics,
+	}, nil
+}
+
+func (mc *MConsumerGroup) GetContextFromMsg(cMsg *sarama.ConsumerMessage) context.Context {
+	return GetContextWithMQHeader(cMsg.Headers)
+}
+
+func (mc *MConsumerGroup) RegisterHandleAndConsumer(ctx context.Context, handler sarama.ConsumerGroupHandler) {
+	for {
+		err := mc.ConsumerGroup.Consume(ctx, mc.topics, handler)
+		if errors.Is(err, sarama.ErrClosedConsumerGroup) {
+			return
+		}
+		if errors.Is(err, context.Canceled) {
+			return
+		}
+		if err != nil {
+			log.ZWarn(ctx, "consume err", err, "topic", mc.topics, "groupID", mc.groupID)
+		}
+	}
+}
+
+func (mc *MConsumerGroup) Close() error {
+	return mc.ConsumerGroup.Close()
+}

pkg/common/storage/kafka/producer.go

@@ -0,0 +1,82 @@
+// 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 kafka
+
+import (
+	"context"
+	"github.com/IBM/sarama"
+	"github.com/openimsdk/tools/errs"
+	"google.golang.org/protobuf/proto"
+)
+
+// Producer represents a Kafka producer.
+type Producer struct {
+	addr     []string
+	topic    string
+	config   *sarama.Config
+	producer sarama.SyncProducer
+}
+
+func NewKafkaProducer(config *sarama.Config, addr []string, topic string) (*Producer, error) {
+	producer, err := NewProducer(config, addr)
+	if err != nil {
+		return nil, err
+	}
+	return &Producer{
+		addr:     addr,
+		topic:    topic,
+		config:   config,
+		producer: producer,
+	}, nil
+}
+
+// SendMessage sends a message to the Kafka topic configured in the Producer.
+func (p *Producer) SendMessage(ctx context.Context, key string, msg proto.Message) (int32, int64, error) {
+	// Marshal the protobuf message
+	bMsg, err := proto.Marshal(msg)
+	if err != nil {
+		return 0, 0, errs.WrapMsg(err, "kafka proto Marshal err")
+	}
+	if len(bMsg) == 0 {
+		return 0, 0, errs.WrapMsg(errEmptyMsg, "kafka proto Marshal err")
+	}
+
+	// Prepare Kafka message
+	kMsg := &sarama.ProducerMessage{
+		Topic: p.topic,
+		Key:   sarama.StringEncoder(key),
+		Value: sarama.ByteEncoder(bMsg),
+	}
+
+	// Validate message key and value
+	if kMsg.Key.Length() == 0 || kMsg.Value.Length() == 0 {
+		return 0, 0, errs.Wrap(errEmptyMsg)
+	}
+
+	// Attach context metadata as headers
+	header, err := GetMQHeaderWithContext(ctx)
+	if err != nil {
+		return 0, 0, err
+	}
+	kMsg.Headers = header
+
+	// Send the message
+	partition, offset, err := p.producer.SendMessage(kMsg)
+	if err != nil {
+		return 0, 0, errs.WrapMsg(err, "p.producer.SendMessage error")
+	}
+
+	return partition, offset, nil
+}

pkg/common/storage/kafka/sarama.go

@@ -0,0 +1,85 @@
+package kafka
+
+import (
+	"bytes"
+	"strings"
+
+	"github.com/IBM/sarama"
+	"github.com/openimsdk/tools/errs"
+)
+
+func BuildConsumerGroupConfig(conf *Config, initial int64, autoCommitEnable bool) (*sarama.Config, error) {
+	kfk := sarama.NewConfig()
+	kfk.Version = sarama.V2_0_0_0
+	kfk.Consumer.Offsets.Initial = initial
+	kfk.Consumer.Offsets.AutoCommit.Enable = autoCommitEnable
+	kfk.Consumer.Return.Errors = false
+	if conf.Username != "" || conf.Password != "" {
+		kfk.Net.SASL.Enable = true
+		kfk.Net.SASL.User = conf.Username
+		kfk.Net.SASL.Password = conf.Password
+	}
+	if conf.TLS.EnableTLS {
+		tls, err := newTLSConfig(conf.TLS.ClientCrt, conf.TLS.ClientKey, conf.TLS.CACrt, []byte(conf.TLS.ClientKeyPwd), conf.TLS.InsecureSkipVerify)
+		if err != nil {
+			return nil, err
+		}
+		kfk.Net.TLS.Config = tls
+		kfk.Net.TLS.Enable = true
+	}
+	return kfk, nil
+}
+
+func NewConsumerGroup(conf *sarama.Config, addr []string, groupID string) (sarama.ConsumerGroup, error) {
+	cg, err := sarama.NewConsumerGroup(addr, groupID, conf)
+	if err != nil {
+		return nil, errs.WrapMsg(err, "NewConsumerGroup failed", "addr", addr, "groupID", groupID, "conf", *conf)
+	}
+	return cg, nil
+}
+
+func BuildProducerConfig(conf Config) (*sarama.Config, error) {
+	kfk := sarama.NewConfig()
+	kfk.Producer.Return.Successes = true
+	kfk.Producer.Return.Errors = true
+	kfk.Producer.Partitioner = sarama.NewHashPartitioner
+	if conf.Username != "" || conf.Password != "" {
+		kfk.Net.SASL.Enable = true
+		kfk.Net.SASL.User = conf.Username
+		kfk.Net.SASL.Password = conf.Password
+	}
+	switch strings.ToLower(conf.ProducerAck) {
+	case "no_response":
+		kfk.Producer.RequiredAcks = sarama.NoResponse
+	case "wait_for_local":
+		kfk.Producer.RequiredAcks = sarama.WaitForLocal
+	case "wait_for_all":
+		kfk.Producer.RequiredAcks = sarama.WaitForAll
+	default:
+		kfk.Producer.RequiredAcks = sarama.WaitForAll
+	}
+	if conf.CompressType == "" {
+		kfk.Producer.Compression = sarama.CompressionNone
+	} else {
+		if err := kfk.Producer.Compression.UnmarshalText(bytes.ToLower([]byte(conf.CompressType))); err != nil {
+			return nil, errs.WrapMsg(err, "UnmarshalText failed", "compressType", conf.CompressType)
+		}
+	}
+	if conf.TLS.EnableTLS {
+		tls, err := newTLSConfig(conf.TLS.ClientCrt, conf.TLS.ClientKey, conf.TLS.CACrt, []byte(conf.TLS.ClientKeyPwd), conf.TLS.InsecureSkipVerify)
+		if err != nil {
+			return nil, err
+		}
+		kfk.Net.TLS.Config = tls
+		kfk.Net.TLS.Enable = true
+	}
+	return kfk, nil
+}
+
+func NewProducer(conf *sarama.Config, addr []string) (sarama.SyncProducer, error) {
+	producer, err := sarama.NewSyncProducer(addr, conf)
+	if err != nil {
+		return nil, errs.WrapMsg(err, "NewSyncProducer failed", "addr", addr, "conf", *conf)
+	}
+	return producer, nil
+}

pkg/common/storage/kafka/tls.go

@@ -0,0 +1,83 @@
+// Copyright © 2024 OpenIM open source community. 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 kafka
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/pem"
+	"os"
+
+	"github.com/openimsdk/tools/errs"
+)
+
+// decryptPEM decrypts a PEM block using a password.
+func decryptPEM(data []byte, passphrase []byte) ([]byte, error) {
+	if len(passphrase) == 0 {
+		return data, nil
+	}
+	b, _ := pem.Decode(data)
+	d, err := x509.DecryptPEMBlock(b, passphrase)
+	if err != nil {
+		return nil, errs.WrapMsg(err, "DecryptPEMBlock failed")
+	}
+	return pem.EncodeToMemory(&pem.Block{
+		Type:  b.Type,
+		Bytes: d,
+	}), nil
+}
+
+func readEncryptablePEMBlock(path string, pwd []byte) ([]byte, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return nil, errs.WrapMsg(err, "ReadFile failed", "path", path)
+	}
+	return decryptPEM(data, pwd)
+}
+
+// newTLSConfig setup the TLS config from general config file.
+func newTLSConfig(clientCertFile, clientKeyFile, caCertFile string, keyPwd []byte, insecureSkipVerify bool) (*tls.Config, error) {
+	var tlsConfig tls.Config
+	if clientCertFile != "" && clientKeyFile != "" {
+		certPEMBlock, err := os.ReadFile(clientCertFile)
+		if err != nil {
+			return nil, errs.WrapMsg(err, "ReadFile failed", "clientCertFile", clientCertFile)
+		}
+		keyPEMBlock, err := readEncryptablePEMBlock(clientKeyFile, keyPwd)
+		if err != nil {
+			return nil, err
+		}
+
+		cert, err := tls.X509KeyPair(certPEMBlock, keyPEMBlock)
+		if err != nil {
+			return nil, errs.WrapMsg(err, "X509KeyPair failed")
+		}
+		tlsConfig.Certificates = []tls.Certificate{cert}
+	}
+
+	if caCertFile != "" {
+		caCert, err := os.ReadFile(caCertFile)
+		if err != nil {
+			return nil, errs.WrapMsg(err, "ReadFile failed", "caCertFile", caCertFile)
+		}
+		caCertPool := x509.NewCertPool()
+		if ok := caCertPool.AppendCertsFromPEM(caCert); !ok {
+			return nil, errs.New("AppendCertsFromPEM failed")
+		}
+		tlsConfig.RootCAs = caCertPool
+	}
+	tlsConfig.InsecureSkipVerify = insecureSkipVerify
+	return &tlsConfig, nil
+}

pkg/common/storage/kafka/util.go

@@ -0,0 +1,34 @@
+package kafka
+
+import (
+	"context"
+	"errors"
+	"github.com/IBM/sarama"
+	"github.com/openimsdk/protocol/constant"
+	"github.com/openimsdk/tools/mcontext"
+)
+
+var errEmptyMsg = errors.New("kafka binary msg is empty")
+
+// GetMQHeaderWithContext extracts message queue headers from the context.
+func GetMQHeaderWithContext(ctx context.Context) ([]sarama.RecordHeader, error) {
+	operationID, opUserID, platform, connID, err := mcontext.GetCtxInfos(ctx)
+	if err != nil {
+		return nil, err
+	}
+	return []sarama.RecordHeader{
+		{Key: []byte(constant.OperationID), Value: []byte(operationID)},
+		{Key: []byte(constant.OpUserID), Value: []byte(opUserID)},
+		{Key: []byte(constant.OpUserPlatform), Value: []byte(platform)},
+		{Key: []byte(constant.ConnID), Value: []byte(connID)},
+	}, nil
+}
+
+// GetContextWithMQHeader creates a context from message queue headers.
+func GetContextWithMQHeader(header []*sarama.RecordHeader) context.Context {
+	var values []string
+	for _, recordHeader := range header {
+		values = append(values, string(recordHeader.Value))
+	}
+	return mcontext.WithMustInfoCtx(values) // Attach extracted values to context
+}

pkg/common/storage/kafka/verify.go

@@ -0,0 +1,79 @@
+// Copyright © 2024 OpenIM open source community. 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 kafka
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/IBM/sarama"
+	"github.com/openimsdk/tools/errs"
+)
+
+func CheckTopics(ctx context.Context, conf *Config, topics []string) error {
+	kfk, err := BuildConsumerGroupConfig(conf, sarama.OffsetNewest, false)
+	if err != nil {
+		return err
+	}
+	cli, err := sarama.NewClient(conf.Addr, kfk)
+	if err != nil {
+		return errs.WrapMsg(err, "NewClient failed", "config: ", fmt.Sprintf("%+v", conf))
+	}
+	defer cli.Close()
+
+	existingTopics, err := cli.Topics()
+	if err != nil {
+		return errs.WrapMsg(err, "Failed to list topics")
+	}
+
+	existingTopicsMap := make(map[string]bool)
+	for _, t := range existingTopics {
+		existingTopicsMap[t] = true
+	}
+
+	for _, topic := range topics {
+		if !existingTopicsMap[topic] {
+			return errs.New("topic not exist", "topic", topic).Wrap()
+		}
+	}
+	return nil
+}
+
+func CheckHealth(ctx context.Context, conf *Config) error {
+	kfk, err := BuildConsumerGroupConfig(conf, sarama.OffsetNewest, false)
+	if err != nil {
+		return err
+	}
+	cli, err := sarama.NewClient(conf.Addr, kfk)
+	if err != nil {
+		return errs.WrapMsg(err, "NewClient failed", "config: ", fmt.Sprintf("%+v", conf))
+	}
+	defer cli.Close()
+
+	// Get broker list
+	brokers := cli.Brokers()
+	if len(brokers) == 0 {
+		return errs.New("no brokers found").Wrap()
+	}
+
+	// Check if all brokers are reachable
+	for _, broker := range brokers {
+		if err := broker.Open(kfk); err != nil {
+			return errs.WrapMsg(err, "failed to open broker", "broker", broker.Addr())
+		}
+	}
+
+	return nil
+}

pkg/localcache/cache.go

@@ -47,15 +47,15 @@ func New[V any](opts ...Option) Cache[V] {
 	if opt.localSlotNum > 0 && opt.localSlotSize > 0 {
 		createSimpleLRU := func() lru.LRU[string, V] {
 			if opt.expirationEvict {
-				return lru.NewExpirationLRU(opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
+				return lru.NewExpirationLRU[string, V](opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
 			} else {
-				return lru.NewLazyLRU(opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
+				return lru.NewLazyLRU[string, V](opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
 			}
 		}
 		if opt.localSlotNum == 1 {
 			c.local = createSimpleLRU()
 		} else {
-			c.local = lru.NewSlotLRU(opt.localSlotNum, LRUStringHash, createSimpleLRU)
+			c.local = lru.NewSlotLRU[string, V](opt.localSlotNum, LRUStringHash, createSimpleLRU)
 		}
 		if opt.linkSlotNum > 0 {
 			c.link = link.New(opt.linkSlotNum)
@@ -71,14 +71,19 @@ type cache[V any] struct {
 }
 
 func (c *cache[V]) onEvict(key string, value V) {
-	_ = value
-
 	if c.link != nil {
-		lks := c.link.Del(key)
-		for k := range lks {
-			if key != k { // prevent deadlock
-				c.local.Del(k)
-			}
+		// Do not delete other keys while the underlying LRU still holds its lock;
+		// defer linked deletions to avoid re-entering the same slot and deadlocking.
+		if lks := c.link.Del(key); len(lks) > 0 {
+			go c.delLinked(key, lks)
+		}
+	}
+}
+
+func (c *cache[V]) delLinked(src string, keys map[string]struct{}) {
+	for k := range keys {
+		if src != k {
+			c.local.Del(k)
 		}
 	}
 }
@@ -105,7 +110,7 @@ func (c *cache[V]) Get(ctx context.Context, key string, fetch func(ctx context.C
 func (c *cache[V]) GetLink(ctx context.Context, key string, fetch func(ctx context.Context) (V, error), link ...string) (V, error) {
 	if c.local != nil {
 		return c.local.Get(key, func() (V, error) {
-			if len(link) > 0 {
+			if len(link) > 0 && c.link != nil {
 				c.link.Link(key, link...)
 			}
 			return fetch(ctx)

pkg/localcache/cache_test.go

@@ -22,6 +22,8 @@ import (
 	"sync/atomic"
 	"testing"
 	"time"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/localcache/lru"
 )
 
 func TestName(t *testing.T) {
@@ -91,3 +93,68 @@ func TestName(t *testing.T) {
 	t.Log("del", del.Load())
 	// 137.35s
 }
+
+// Test deadlock scenario when eviction callback deletes a linked key that hashes to the same slot.
+func TestCacheEvictDeadlock(t *testing.T) {
+	ctx := context.Background()
+	c := New[string](WithLocalSlotNum(1), WithLocalSlotSize(1), WithLazy())
+
+	if _, err := c.GetLink(ctx, "k1", func(ctx context.Context) (string, error) {
+		return "v1", nil
+	}, "k2"); err != nil {
+		t.Fatalf("seed cache failed: %v", err)
+	}
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		_, _ = c.GetLink(ctx, "k2", func(ctx context.Context) (string, error) {
+			return "v2", nil
+		}, "k1")
+	}()
+
+	select {
+	case <-done:
+		// expected to finish quickly; current implementation deadlocks here.
+	case <-time.After(time.Second):
+		t.Fatal("GetLink deadlocked during eviction of linked key")
+	}
+}
+
+func TestExpirationLRUGetBatch(t *testing.T) {
+	l := lru.NewExpirationLRU[string, string](2, time.Minute, time.Second*5, EmptyTarget{}, nil)
+
+	keys := []string{"a", "b"}
+	values, err := l.GetBatch(keys, func(keys []string) (map[string]string, error) {
+		res := make(map[string]string)
+		for _, k := range keys {
+			res[k] = k + "_v"
+		}
+		return res, nil
+	})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(values) != len(keys) {
+		t.Fatalf("expected %d values, got %d", len(keys), len(values))
+	}
+	for _, k := range keys {
+		if v, ok := values[k]; !ok || v != k+"_v" {
+			t.Fatalf("unexpected value for %s: %q, ok=%v", k, v, ok)
+		}
+	}
+
+	// second batch should hit cache
+	values, err = l.GetBatch(keys, func(keys []string) (map[string]string, error) {
+		t.Fatalf("should not fetch on cache hit")
+		return nil, nil
+	})
+	if err != nil {
+		t.Fatalf("unexpected error on cache hit: %v", err)
+	}
+	for _, k := range keys {
+		if v, ok := values[k]; !ok || v != k+"_v" {
+			t.Fatalf("unexpected cached value for %s: %q, ok=%v", k, v, ok)
+		}
+	}
+}

pkg/localcache/init.go

@@ -33,10 +33,6 @@ func InitLocalCache(localCache *config.LocalCache) {
 			Local config.CacheConfig
 			Keys  []string
 		}{
-			{
-				Local: localCache.Auth,
-				Keys:  []string{cachekey.UidPidToken},
-			},
 			{
 				Local: localCache.User,
 				Keys:  []string{cachekey.UserInfoKey, cachekey.UserGlobalRecvMsgOptKey},

pkg/localcache/lru/lru_expiration.go

@@ -52,8 +52,53 @@ type ExpirationLRU[K comparable, V any] struct {
 }
 
 func (x *ExpirationLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) (map[K]V, error) {
-	//TODO implement me
-	panic("implement me")
+	var (
+		err     error
+		results = make(map[K]V)
+		misses  = make([]K, 0, len(keys))
+	)
+
+	for _, key := range keys {
+		x.lock.Lock()
+		v, ok := x.core.Get(key)
+		x.lock.Unlock()
+		if ok {
+			x.target.IncrGetHit()
+			v.lock.RLock()
+			results[key] = v.value
+			if v.err != nil && err == nil {
+				err = v.err
+			}
+			v.lock.RUnlock()
+			continue
+		}
+		misses = append(misses, key)
+	}
+
+	if len(misses) == 0 {
+		return results, err
+	}
+
+	fetchValues, fetchErr := fetch(misses)
+	if fetchErr != nil && err == nil {
+		err = fetchErr
+	}
+
+	for key, val := range fetchValues {
+		results[key] = val
+		if fetchErr != nil {
+			x.target.IncrGetFailed()
+			continue
+		}
+		x.target.IncrGetSuccess()
+		item := &expirationLruItem[V]{value: val}
+		x.lock.Lock()
+		x.core.Add(key, item)
+		x.lock.Unlock()
+	}
+
+	// any keys not returned from fetch remain absent (no cache write)
+	return results, err
 }
 
 func (x *ExpirationLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) {

pkg/localcache/lru/lru_slot.go

@@ -35,7 +35,7 @@ type slotLRU[K comparable, V any] struct {
 func (x *slotLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) (map[K]V, error) {
 	var (
 		slotKeys = make(map[uint64][]K)
-		kVs       = make(map[K]V)
+		vs       = make(map[K]V)
 	)
 
 	for _, k := range keys {
@@ -49,10 +49,10 @@ func (x *slotLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)
 			return nil, err
 		}
 		for key, value := range batches {
-			kVs[key] = value
+			vs[key] = value
 		}
 	}
-	return kVs, nil
+	return vs, nil
 }
 
 func (x *slotLRU[K, V]) getIndex(k K) uint64 {

pkg/msgprocessor/conversation.go

@@ -109,7 +109,7 @@ func GetConversationIDBySessionType(sessionType int, ids ...string) string {
 	case constant.ReadGroupChatType:
 		return "sg_" + ids[0] // super group chat
 	case constant.NotificationChatType:
-		return "sn_" + ids[0] // server notification chat
+		return "sn_" + strings.Join(ids, "_") // server notification chat
 	}
 	return ""
 }

pkg/rpcli/conversation.go

@@ -2,6 +2,7 @@ package rpcli
 
 import (
 	"context"
+
 	"github.com/openimsdk/protocol/conversation"
 	"google.golang.org/grpc"
 )
@@ -30,18 +31,6 @@ func (x *ConversationClient) SetConversations(ctx context.Context, ownerUserIDs
 	return ignoreResp(x.ConversationClient.SetConversations(ctx, req))
 }
 
-func (x *ConversationClient) GetConversationsByConversationIDs(ctx context.Context, conversationIDs []string) ([]*conversation.Conversation, error) {
-	if len(conversationIDs) == 0 {
-		return nil, nil
-	}
-	req := &conversation.GetConversationsByConversationIDReq{ConversationIDs: conversationIDs}
-	return extractField(ctx, x.ConversationClient.GetConversationsByConversationID, req, (*conversation.GetConversationsByConversationIDResp).GetConversations)
-}
-
-func (x *ConversationClient) GetConversationsByConversationID(ctx context.Context, conversationID string) (*conversation.Conversation, error) {
-	return firstValue(x.GetConversationsByConversationIDs(ctx, []string{conversationID}))
-}
-
 func (x *ConversationClient) SetConversationMinSeq(ctx context.Context, conversationID string, ownerUserIDs []string, minSeq int64) error {
 	if len(ownerUserIDs) == 0 {
 		return nil

tools/seq/internal/seq.go

@@ -72,7 +72,7 @@ func Main(conf string, del time.Duration) error {
 	if err != nil {
 		return err
 	}
-	
+
 	mongodbConfig, err := readConfig[config.Mongo](conf, config.MongodbConfigFileName)
 	if err != nil {
 		return err

tools/stress-test-v2/main.go

@@ -0,0 +1,735 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"os/signal"
+	"sync"
+	"syscall"
+	"time"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/apistruct"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/protocol/auth"
+	"github.com/openimsdk/protocol/constant"
+	"github.com/openimsdk/protocol/group"
+	"github.com/openimsdk/protocol/sdkws"
+	pbuser "github.com/openimsdk/protocol/user"
+	"github.com/openimsdk/tools/log"
+	"github.com/openimsdk/tools/system/program"
+)
+
+// 1. Create 100K New Users
+// 2. Create 100 100K Groups
+// 3. Create 1000 999 Groups
+// 4. Send message to 100K Groups every second
+// 5. Send message to 999 Groups every minute
+
+var (
+	//  Use default userIDs List for testing, need to be created.
+	TestTargetUserList = []string{
+		// "<need-update-it>",
+	}
+	// DefaultGroupID = "<need-update-it>" // Use default group ID for testing, need to be created.
+)
+
+var (
+	ApiAddress string
+
+	// API method
+	GetAdminToken      = "/auth/get_admin_token"
+	UserCheck          = "/user/account_check"
+	CreateUser         = "/user/user_register"
+	ImportFriend       = "/friend/import_friend"
+	InviteToGroup      = "/group/invite_user_to_group"
+	GetGroupMemberInfo = "/group/get_group_members_info"
+	SendMsg            = "/msg/send_msg"
+	CreateGroup        = "/group/create_group"
+	GetUserToken       = "/auth/user_token"
+)
+
+const (
+	MaxUser            = 100000
+	Max100KGroup       = 100
+	Max999Group        = 1000
+	MaxInviteUserLimit = 999
+
+	CreateUserTicker         = 1 * time.Second
+	CreateGroupTicker        = 1 * time.Second
+	Create100KGroupTicker    = 1 * time.Second
+	Create999GroupTicker     = 1 * time.Second
+	SendMsgTo100KGroupTicker = 1 * time.Second
+	SendMsgTo999GroupTicker  = 1 * time.Minute
+)
+
+type BaseResp struct {
+	ErrCode int             `json:"errCode"`
+	ErrMsg  string          `json:"errMsg"`
+	Data    json.RawMessage `json:"data"`
+}
+
+type StressTest struct {
+	Conf                   *conf
+	AdminUserID            string
+	AdminToken             string
+	DefaultGroupID         string
+	DefaultUserID          string
+	UserCounter            int
+	CreateUserCounter      int
+	Create100kGroupCounter int
+	Create999GroupCounter  int
+	MsgCounter             int
+	CreatedUsers           []string
+	CreatedGroups          []string
+	Mutex                  sync.Mutex
+	Ctx                    context.Context
+	Cancel                 context.CancelFunc
+	HttpClient             *http.Client
+	Wg                     sync.WaitGroup
+	Once                   sync.Once
+}
+
+type conf struct {
+	Share config.Share
+	Api   config.API
+}
+
+func initConfig(configDir string) (*config.Share, *config.API, error) {
+	var (
+		share     = &config.Share{}
+		apiConfig = &config.API{}
+	)
+
+	err := config.Load(configDir, config.ShareFileName, config.EnvPrefixMap[config.ShareFileName], share)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	err = config.Load(configDir, config.OpenIMAPICfgFileName, config.EnvPrefixMap[config.OpenIMAPICfgFileName], apiConfig)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return share, apiConfig, nil
+}
+
+// Post Request
+func (st *StressTest) PostRequest(ctx context.Context, url string, reqbody any) ([]byte, error) {
+	// Marshal body
+	jsonBody, err := json.Marshal(reqbody)
+	if err != nil {
+		log.ZError(ctx, "Failed to marshal request body", err, "url", url, "reqbody", reqbody)
+		return nil, err
+	}
+
+	req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(jsonBody))
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("operationID", st.AdminUserID)
+	if st.AdminToken != "" {
+		req.Header.Set("token", st.AdminToken)
+	}
+
+	// log.ZInfo(ctx, "Header info is ", "Content-Type", "application/json", "operationID", st.AdminUserID, "token", st.AdminToken)
+
+	resp, err := st.HttpClient.Do(req)
+	if err != nil {
+		log.ZError(ctx, "Failed to send request", err, "url", url, "reqbody", reqbody)
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	respBody, err := io.ReadAll(resp.Body)
+	if err != nil {
+		log.ZError(ctx, "Failed to read response body", err, "url", url)
+		return nil, err
+	}
+
+	var baseResp BaseResp
+	if err := json.Unmarshal(respBody, &baseResp); err != nil {
+		log.ZError(ctx, "Failed to unmarshal response body", err, "url", url, "respBody", string(respBody))
+		return nil, err
+	}
+
+	if baseResp.ErrCode != 0 {
+		err = fmt.Errorf(baseResp.ErrMsg)
+		log.ZError(ctx, "Failed to send request", err, "url", url, "reqbody", reqbody, "resp", baseResp)
+		return nil, err
+	}
+
+	return baseResp.Data, nil
+}
+
+func (st *StressTest) GetAdminToken(ctx context.Context) (string, error) {
+	req := auth.GetAdminTokenReq{
+		Secret: st.Conf.Share.Secret,
+		UserID: st.AdminUserID,
+	}
+
+	resp, err := st.PostRequest(ctx, ApiAddress+GetAdminToken, &req)
+	if err != nil {
+		return "", err
+	}
+
+	data := &auth.GetAdminTokenResp{}
+	if err := json.Unmarshal(resp, &data); err != nil {
+		return "", err
+	}
+
+	return data.Token, nil
+}
+
+func (st *StressTest) CheckUser(ctx context.Context, userIDs []string) ([]string, error) {
+	req := pbuser.AccountCheckReq{
+		CheckUserIDs: userIDs,
+	}
+
+	resp, err := st.PostRequest(ctx, ApiAddress+UserCheck, &req)
+	if err != nil {
+		return nil, err
+	}
+
+	data := &pbuser.AccountCheckResp{}
+	if err := json.Unmarshal(resp, &data); err != nil {
+		return nil, err
+	}
+
+	unRegisteredUserIDs := make([]string, 0)
+
+	for _, res := range data.Results {
+		if res.AccountStatus == constant.UnRegistered {
+			unRegisteredUserIDs = append(unRegisteredUserIDs, res.UserID)
+		}
+	}
+
+	return unRegisteredUserIDs, nil
+}
+
+func (st *StressTest) CreateUser(ctx context.Context, userID string) (string, error) {
+	user := &sdkws.UserInfo{
+		UserID:   userID,
+		Nickname: userID,
+	}
+
+	req := pbuser.UserRegisterReq{
+		Users: []*sdkws.UserInfo{user},
+	}
+
+	_, err := st.PostRequest(ctx, ApiAddress+CreateUser, &req)
+	if err != nil {
+		return "", err
+	}
+
+	st.UserCounter++
+	return userID, nil
+}
+
+func (st *StressTest) CreateUserBatch(ctx context.Context, userIDs []string) error {
+	// The method can import a large number of users at once.
+	var userList []*sdkws.UserInfo
+
+	defer st.Once.Do(
+		func() {
+			st.DefaultUserID = userIDs[0]
+			fmt.Println("Default Send User Created ID:", st.DefaultUserID)
+		})
+
+	needUserIDs, err := st.CheckUser(ctx, userIDs)
+	if err != nil {
+		return err
+	}
+
+	for _, userID := range needUserIDs {
+		user := &sdkws.UserInfo{
+			UserID:   userID,
+			Nickname: userID,
+		}
+		userList = append(userList, user)
+	}
+
+	req := pbuser.UserRegisterReq{
+		Users: userList,
+	}
+
+	_, err = st.PostRequest(ctx, ApiAddress+CreateUser, &req)
+	if err != nil {
+		return err
+	}
+
+	st.UserCounter += len(userList)
+	return nil
+}
+
+func (st *StressTest) GetGroupMembersInfo(ctx context.Context, groupID string, userIDs []string) ([]string, error) {
+	needInviteUserIDs := make([]string, 0)
+
+	const maxBatchSize = 500
+	if len(userIDs) > maxBatchSize {
+		for i := 0; i < len(userIDs); i += maxBatchSize {
+			end := min(i+maxBatchSize, len(userIDs))
+			batchUserIDs := userIDs[i:end]
+
+			// log.ZInfo(ctx, "Processing group members batch", "groupID", groupID, "batch", i/maxBatchSize+1,
+			// 	"batchUserCount", len(batchUserIDs))
+
+			// Process a single batch
+			batchReq := group.GetGroupMembersInfoReq{
+				GroupID: groupID,
+				UserIDs: batchUserIDs,
+			}
+
+			resp, err := st.PostRequest(ctx, ApiAddress+GetGroupMemberInfo, &batchReq)
+			if err != nil {
+				log.ZError(ctx, "Batch query failed", err, "batch", i/maxBatchSize+1)
+				continue
+			}
+
+			data := &group.GetGroupMembersInfoResp{}
+			if err := json.Unmarshal(resp, &data); err != nil {
+				log.ZError(ctx, "Failed to parse batch response", err, "batch", i/maxBatchSize+1)
+				continue
+			}
+
+			// Process the batch results
+			existingMembers := make(map[string]bool)
+			for _, member := range data.Members {
+				existingMembers[member.UserID] = true
+			}
+
+			for _, userID := range batchUserIDs {
+				if !existingMembers[userID] {
+					needInviteUserIDs = append(needInviteUserIDs, userID)
+				}
+			}
+		}
+
+		return needInviteUserIDs, nil
+	}
+
+	req := group.GetGroupMembersInfoReq{
+		GroupID: groupID,
+		UserIDs: userIDs,
+	}
+
+	resp, err := st.PostRequest(ctx, ApiAddress+GetGroupMemberInfo, &req)
+	if err != nil {
+		return nil, err
+	}
+
+	data := &group.GetGroupMembersInfoResp{}
+	if err := json.Unmarshal(resp, &data); err != nil {
+		return nil, err
+	}
+
+	existingMembers := make(map[string]bool)
+	for _, member := range data.Members {
+		existingMembers[member.UserID] = true
+	}
+
+	for _, userID := range userIDs {
+		if !existingMembers[userID] {
+			needInviteUserIDs = append(needInviteUserIDs, userID)
+		}
+	}
+
+	return needInviteUserIDs, nil
+}
+
+func (st *StressTest) InviteToGroup(ctx context.Context, groupID string, userIDs []string) error {
+	req := group.InviteUserToGroupReq{
+		GroupID:        groupID,
+		InvitedUserIDs: userIDs,
+	}
+	_, err := st.PostRequest(ctx, ApiAddress+InviteToGroup, &req)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (st *StressTest) SendMsg(ctx context.Context, userID string, groupID string) error {
+	contentObj := map[string]any{
+		// "content": fmt.Sprintf("index %d. The current time is %s", st.MsgCounter, time.Now().Format("2006-01-02 15:04:05.000")),
+		"content": fmt.Sprintf("The current time is %s", time.Now().Format("2006-01-02 15:04:05.000")),
+	}
+
+	req := &apistruct.SendMsgReq{
+		SendMsg: apistruct.SendMsg{
+			SendID:         userID,
+			SenderNickname: userID,
+			GroupID:        groupID,
+			ContentType:    constant.Text,
+			SessionType:    constant.ReadGroupChatType,
+			Content:        contentObj,
+		},
+	}
+
+	_, err := st.PostRequest(ctx, ApiAddress+SendMsg, &req)
+	if err != nil {
+		log.ZError(ctx, "Failed to send message", err, "userID", userID, "req", &req)
+		return err
+	}
+
+	st.MsgCounter++
+
+	return nil
+}
+
+// Max userIDs number is 1000
+func (st *StressTest) CreateGroup(ctx context.Context, groupID string, userID string, userIDsList []string) (string, error) {
+	groupInfo := &sdkws.GroupInfo{
+		GroupID:   groupID,
+		GroupName: groupID,
+		GroupType: constant.WorkingGroup,
+	}
+
+	req := group.CreateGroupReq{
+		OwnerUserID:   userID,
+		MemberUserIDs: userIDsList,
+		GroupInfo:     groupInfo,
+	}
+
+	resp := group.CreateGroupResp{}
+
+	response, err := st.PostRequest(ctx, ApiAddress+CreateGroup, &req)
+	if err != nil {
+		return "", err
+	}
+
+	if err := json.Unmarshal(response, &resp); err != nil {
+		return "", err
+	}
+
+	// st.GroupCounter++
+
+	return resp.GroupInfo.GroupID, nil
+}
+
+func main() {
+	var configPath string
+	// defaultConfigDir := filepath.Join("..", "..", "..", "..", "..", "config")
+	// flag.StringVar(&configPath, "c", defaultConfigDir, "config path")
+	flag.StringVar(&configPath, "c", "", "config path")
+	flag.Parse()
+
+	if configPath == "" {
+		_, _ = fmt.Fprintln(os.Stderr, "config path is empty")
+		os.Exit(1)
+		return
+	}
+
+	fmt.Printf(" Config Path: %s\n", configPath)
+
+	share, apiConfig, err := initConfig(configPath)
+	if err != nil {
+		program.ExitWithError(err)
+		return
+	}
+
+	ApiAddress = fmt.Sprintf("http://%s:%s", "127.0.0.1", fmt.Sprint(apiConfig.Api.Ports[0]))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	// ch := make(chan struct{})
+
+	st := &StressTest{
+		Conf: &conf{
+			Share: *share,
+			Api:   *apiConfig,
+		},
+		AdminUserID: share.IMAdminUser.UserIDs[0],
+		Ctx:         ctx,
+		Cancel:      cancel,
+		HttpClient: &http.Client{
+			Timeout: 50 * time.Second,
+		},
+	}
+
+	c := make(chan os.Signal, 1)
+	signal.Notify(c, os.Interrupt, syscall.SIGTERM)
+	go func() {
+		<-c
+		fmt.Println("\nReceived stop signal, stopping...")
+
+		go func() {
+			// time.Sleep(5 * time.Second)
+			fmt.Println("Force exit")
+			os.Exit(0)
+		}()
+
+		st.Cancel()
+	}()
+
+	token, err := st.GetAdminToken(st.Ctx)
+	if err != nil {
+		log.ZError(ctx, "Get Admin Token failed.", err, "AdminUserID", st.AdminUserID)
+	}
+
+	st.AdminToken = token
+	fmt.Println("Admin Token:", st.AdminToken)
+	fmt.Println("ApiAddress:", ApiAddress)
+	for i := 0; i < MaxUser; i++ {
+		userID := fmt.Sprintf("v2_StressTest_User_%d", i)
+		st.CreatedUsers = append(st.CreatedUsers, userID)
+		st.CreateUserCounter++
+	}
+
+	// err = st.CreateUserBatch(st.Ctx, st.CreatedUsers)
+	// if err != nil {
+	// 	log.ZError(ctx, "Create user failed.", err)
+	// }
+
+	const batchSize = 1000
+	totalUsers := len(st.CreatedUsers)
+	successCount := 0
+
+	if st.DefaultUserID == "" && len(st.CreatedUsers) > 0 {
+		st.DefaultUserID = st.CreatedUsers[0]
+	}
+
+	for i := 0; i < totalUsers; i += batchSize {
+		end := min(i+batchSize, totalUsers)
+
+		userBatch := st.CreatedUsers[i:end]
+		log.ZInfo(st.Ctx, "Creating user batch", "batch", i/batchSize+1, "count", len(userBatch))
+
+		err = st.CreateUserBatch(st.Ctx, userBatch)
+		if err != nil {
+			log.ZError(st.Ctx, "Batch user creation failed", err, "batch", i/batchSize+1)
+		} else {
+			successCount += len(userBatch)
+			log.ZInfo(st.Ctx, "Batch user creation succeeded", "batch", i/batchSize+1,
+				"progress", fmt.Sprintf("%d/%d", successCount, totalUsers))
+		}
+	}
+
+	// Execute create 100k group
+	st.Wg.Add(1)
+	go func() {
+		defer st.Wg.Done()
+
+		create100kGroupTicker := time.NewTicker(Create100KGroupTicker)
+		defer create100kGroupTicker.Stop()
+
+		for i := 0; i < Max100KGroup; i++ {
+			select {
+			case <-st.Ctx.Done():
+				log.ZInfo(st.Ctx, "Stop Create 100K Group")
+				return
+
+			case <-create100kGroupTicker.C:
+				// Create 100K groups
+				st.Wg.Add(1)
+				go func(idx int) {
+					defer st.Wg.Done()
+					defer func() {
+						st.Create100kGroupCounter++
+					}()
+
+					groupID := fmt.Sprintf("v2_StressTest_Group_100K_%d", idx)
+
+					if _, err = st.CreateGroup(st.Ctx, groupID, st.DefaultUserID, TestTargetUserList); err != nil {
+						log.ZError(st.Ctx, "Create group failed.", err)
+						// continue
+					}
+
+					for i := 0; i < MaxUser/MaxInviteUserLimit; i++ {
+						InviteUserIDs := make([]string, 0)
+						// ensure TargetUserList is in group
+						InviteUserIDs = append(InviteUserIDs, TestTargetUserList...)
+
+						startIdx := max(i*MaxInviteUserLimit, 1)
+						endIdx := min((i+1)*MaxInviteUserLimit, MaxUser)
+
+						for j := startIdx; j < endIdx; j++ {
+							userCreatedID := fmt.Sprintf("v2_StressTest_User_%d", j)
+							InviteUserIDs = append(InviteUserIDs, userCreatedID)
+						}
+
+						if len(InviteUserIDs) == 0 {
+							log.ZWarn(st.Ctx, "InviteUserIDs is empty", nil, "groupID", groupID)
+							continue
+						}
+
+						InviteUserIDs, err := st.GetGroupMembersInfo(ctx, groupID, InviteUserIDs)
+						if err != nil {
+							log.ZError(st.Ctx, "GetGroupMembersInfo failed.", err, "groupID", groupID)
+							continue
+						}
+
+						if len(InviteUserIDs) == 0 {
+							log.ZWarn(st.Ctx, "InviteUserIDs is empty", nil, "groupID", groupID)
+							continue
+						}
+
+						// Invite To Group
+						if err = st.InviteToGroup(st.Ctx, groupID, InviteUserIDs); err != nil {
+							log.ZError(st.Ctx, "Invite To Group failed.", err, "UserID", InviteUserIDs)
+							continue
+							// os.Exit(1)
+							// return
+						}
+					}
+				}(i)
+			}
+		}
+	}()
+
+	// create 999 groups
+	st.Wg.Add(1)
+	go func() {
+		defer st.Wg.Done()
+
+		create999GroupTicker := time.NewTicker(Create999GroupTicker)
+		defer create999GroupTicker.Stop()
+
+		for i := 0; i < Max999Group; i++ {
+			select {
+			case <-st.Ctx.Done():
+				log.ZInfo(st.Ctx, "Stop Create 999 Group")
+				return
+
+			case <-create999GroupTicker.C:
+				// Create 999 groups
+				st.Wg.Add(1)
+				go func(idx int) {
+					defer st.Wg.Done()
+					defer func() {
+						st.Create999GroupCounter++
+					}()
+
+					groupID := fmt.Sprintf("v2_StressTest_Group_1K_%d", idx)
+
+					if _, err = st.CreateGroup(st.Ctx, groupID, st.DefaultUserID, TestTargetUserList); err != nil {
+						log.ZError(st.Ctx, "Create group failed.", err)
+						// continue
+					}
+					for i := 0; i < MaxUser/MaxInviteUserLimit; i++ {
+						InviteUserIDs := make([]string, 0)
+						// ensure TargetUserList is in group
+						InviteUserIDs = append(InviteUserIDs, TestTargetUserList...)
+
+						startIdx := max(i*MaxInviteUserLimit, 1)
+						endIdx := min((i+1)*MaxInviteUserLimit, MaxUser)
+
+						for j := startIdx; j < endIdx; j++ {
+							userCreatedID := fmt.Sprintf("v2_StressTest_User_%d", j)
+							InviteUserIDs = append(InviteUserIDs, userCreatedID)
+						}
+
+						if len(InviteUserIDs) == 0 {
+							log.ZWarn(st.Ctx, "InviteUserIDs is empty", nil, "groupID", groupID)
+							continue
+						}
+
+						InviteUserIDs, err := st.GetGroupMembersInfo(ctx, groupID, InviteUserIDs)
+						if err != nil {
+							log.ZError(st.Ctx, "GetGroupMembersInfo failed.", err, "groupID", groupID)
+							continue
+						}
+
+						if len(InviteUserIDs) == 0 {
+							log.ZWarn(st.Ctx, "InviteUserIDs is empty", nil, "groupID", groupID)
+							continue
+						}
+
+						// Invite To Group
+						if err = st.InviteToGroup(st.Ctx, groupID, InviteUserIDs); err != nil {
+							log.ZError(st.Ctx, "Invite To Group failed.", err, "UserID", InviteUserIDs)
+							continue
+							// os.Exit(1)
+							// return
+						}
+					}
+				}(i)
+			}
+		}
+	}()
+
+	// Send message to 100K groups
+	st.Wg.Wait()
+	fmt.Println("All groups created successfully, starting to send messages...")
+	log.ZInfo(ctx, "All groups created successfully, starting to send messages...")
+
+	var groups100K []string
+	var groups999 []string
+
+	for i := 0; i < Max100KGroup; i++ {
+		groupID := fmt.Sprintf("v2_StressTest_Group_100K_%d", i)
+		groups100K = append(groups100K, groupID)
+	}
+
+	for i := 0; i < Max999Group; i++ {
+		groupID := fmt.Sprintf("v2_StressTest_Group_1K_%d", i)
+		groups999 = append(groups999, groupID)
+	}
+
+	send100kGroupLimiter := make(chan struct{}, 20)
+	send999GroupLimiter := make(chan struct{}, 100)
+
+	// execute Send message to 100K groups
+	go func() {
+		ticker := time.NewTicker(SendMsgTo100KGroupTicker)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-st.Ctx.Done():
+				log.ZInfo(st.Ctx, "Stop Send Message to 100K Group")
+				return
+
+			case <-ticker.C:
+				// Send message to 100K groups
+				for _, groupID := range groups100K {
+					send100kGroupLimiter <- struct{}{}
+					go func(groupID string) {
+						defer func() { <-send100kGroupLimiter }()
+						if err := st.SendMsg(st.Ctx, st.DefaultUserID, groupID); err != nil {
+							log.ZError(st.Ctx, "Send message to 100K group failed.", err)
+						}
+					}(groupID)
+				}
+				// log.ZInfo(st.Ctx, "Send message to 100K groups successfully.")
+			}
+		}
+	}()
+
+	// execute Send message to 999 groups
+	go func() {
+		ticker := time.NewTicker(SendMsgTo999GroupTicker)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-st.Ctx.Done():
+				log.ZInfo(st.Ctx, "Stop Send Message to 999 Group")
+				return
+
+			case <-ticker.C:
+				// Send message to 999 groups
+				for _, groupID := range groups999 {
+					send999GroupLimiter <- struct{}{}
+					go func(groupID string) {
+						defer func() { <-send999GroupLimiter }()
+
+						if err := st.SendMsg(st.Ctx, st.DefaultUserID, groupID); err != nil {
+							log.ZError(st.Ctx, "Send message to 999 group failed.", err)
+						}
+					}(groupID)
+				}
+				// log.ZInfo(st.Ctx, "Send message to 999 groups successfully.")
+			}
+		}
+	}()
+
+	<-st.Ctx.Done()
+	fmt.Println("Received signal to exit, shutting down...")
+}