Update version to v3.8.3-patch.6

.env

@@ -1,6 +1,6 @@
 MONGO_IMAGE=mongo:7.0
 REDIS_IMAGE=redis:7.0.0
-KAFKA_IMAGE=bitnamilegacy/kafka:3.5.1
+KAFKA_IMAGE=bitnami/kafka:3.5.1
 MINIO_IMAGE=minio/minio:RELEASE.2024-01-11T07-46-16Z
 ETCD_IMAGE=quay.io/coreos/etcd:v3.5.13
 PROMETHEUS_IMAGE=prom/prometheus:v2.45.6
@@ -22,5 +22,4 @@ MONGO_BACKUP_DIR=${DATA_DIR}components/backup/mongo/
 PROMETHEUS_PORT=19091
 ALERTMANAGER_PORT=19093
 GRAFANA_PORT=13000
-
 NODE_EXPORTER_PORT=19100

.github/workflows/go-build-test.yml

@@ -92,13 +92,12 @@ jobs:
       contents: write
     env:
       SDK_DIR: openim-sdk-core
-      NOTIFICATION_CONFIG_PATH: config/notification.yml
-      SHARE_CONFIG_PATH: config/share.yml
-
+      CONFIG_PATH: config/notification.yml
+    #   pull-requests: write
     strategy:
       matrix:
-        os: [ubuntu-latest]
-        go_version: ["1.22.x"]
+        os: [ ubuntu-latest ]
+        go_version: [ "1.22.x" ]
 
     steps:
       - name: Checkout Server repository
@@ -107,8 +106,7 @@ jobs:
       - name: Checkout SDK repository
         uses: actions/checkout@v4
         with:
-          repository: "openimsdk/openim-sdk-core"
-          ref: "main"
+          repository: 'openimsdk/openim-sdk-core'
           path: ${{ env.SDK_DIR }}
 
       - name: Set up Go ${{ matrix.go_version }}
@@ -121,11 +119,15 @@ jobs:
           go install github.com/magefile/mage@latest
           go mod download
 
+      - name: Install yq
+        run: |
+          sudo wget https://github.com/mikefarah/yq/releases/download/v4.34.1/yq_linux_amd64 -O /usr/bin/yq
+          sudo chmod +x /usr/bin/yq
+
       - name: Modify Server Configuration
         run: |
-          yq e '.groupCreated.isSendMsg = true' -i ${{ env.NOTIFICATION_CONFIG_PATH }}
-          yq e '.friendApplicationApproved.isSendMsg = true' -i ${{ env.NOTIFICATION_CONFIG_PATH }}
-          yq e '.secret = 123456' -i ${{ env.SHARE_CONFIG_PATH }}
+          yq e '.groupCreated.unreadCount = true' -i ${{ env.CONFIG_PATH }}
+          yq e '.friendApplicationApproved.unreadCount = true' -i ${{ env.CONFIG_PATH }}
 
       - name: Start Server Services
         run: |

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

@@ -4,80 +4,45 @@ 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.3"
-
-env:
-  GO_VERSION: "1.22"
-  IMAGE_NAME: "openim-server"
-  # IMAGE_NAME: ${{ github.event.repository.name }}
-  DOCKER_BUILDKIT: 1
+        default: "v3.8.0"
 
 jobs:
-  publish-docker-images:
+  build-and-test:
     runs-on: ubuntu-latest
-    if: ${{ !(github.event_name == 'pull_request' && github.event.pull_request.merged == false) }}
     steps:
-      - name: Checkout main repository
-        uses: actions/checkout@v4
+      - uses: actions/checkout@v4
         with:
           path: main-repo
 
       - name: Set up QEMU
-        uses: docker/setup-qemu-action@v3.3.0
+        uses: docker/setup-qemu-action@v3
 
       - name: Set up Docker Buildx
-        id: buildx
         uses: docker/setup-buildx-action@v3
+
+      - name: Build Docker image
+        id: build
+        uses: docker/build-push-action@v5
         with:
-          driver-opts: network=host
+          context: ./main-repo
+          load: true
+          tags: "openim/openim-server:local"
+          cache-from: type=gha
+          cache-to: type=gha,mode=max
 
-      - 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: Save Docker image to file
+        run: docker save -o image.tar openim/openim-server:local
 
       - name: Checkout compose repository
         uses: actions/checkout@v4
@@ -90,19 +55,18 @@ jobs:
         run: |
           IP=$(hostname -I | awk '{print $1}')
           echo "The IP Address is: $IP"
-          echo "ip=$IP" >> $GITHUB_OUTPUT
+          echo "::set-output name=ip::$IP"
 
       - name: Update .env to use the local image
         run: |
-          sed -i 's|OPENIM_SERVER_IMAGE=.*|OPENIM_SERVER_IMAGE=${{ secrets.DOCKER_USERNAME }}/${{ env.IMAGE_NAME }}:local|' ${{ github.workspace }}/compose-repo/.env
+          sed -i 's|OPENIM_SERVER_IMAGE=.*|OPENIM_SERVER_IMAGE=openim/openim-server: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
         run: |
           cd ${{ github.workspace }}/compose-repo
           docker compose up -d
-
-          docker compose ps
+          sleep 60
 
       # - name: Check openim-server health
       #   run: |
@@ -133,62 +97,54 @@ jobs:
       #         exit 0
       #       fi
 
+      - name: Load Docker image from file
+        run: docker load -i image.tar
+
+      - name: Extract metadata for Docker (tags, labels)
+        id: meta
+        uses: docker/metadata-action@v5.5.1
+        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={{major}}.{{minor}}
+            type=semver,pattern={{major}}
+            type=semver,pattern=release-{{raw}}
+            type=sha
+            type=raw,value=${{ github.event.inputs.tag }}
+
       - name: Log in to Docker Hub
-        uses: docker/login-action@v3.3.0
+        uses: docker/login-action@v2
         with:
           username: ${{ secrets.DOCKER_USERNAME }}
           password: ${{ secrets.DOCKER_PASSWORD }}
 
       - name: Log in to GitHub Container Registry
-        uses: docker/login-action@v3.3.0
+        uses: docker/login-action@v2
         with:
           registry: ghcr.io
           username: ${{ github.repository_owner }}
           password: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Log in to Aliyun Container Registry
-        uses: docker/login-action@v3.3.0
+        uses: docker/login-action@v2
         with:
           registry: registry.cn-hangzhou.aliyuncs.com
           username: ${{ secrets.ALIREGISTRY_USERNAME }}
           password: ${{ secrets.ALIREGISTRY_TOKEN }}
 
-      - 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=(
-            "${{ 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' | 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"
-                      exit 1
-                  fi
-                  amd64_found=$(echo "$manifest" | jq '.manifests[] | select(.platform.architecture == "amd64")')
-                  arm64_found=$(echo "$manifest" | jq '.manifests[] | select(.platform.architecture == "arm64")')
-                  if [[ -z "$amd64_found" ]]; then
-                      echo "Multi-platform support check failed for $image:$tag - missing amd64"
-                      exit 1
-                  fi
-                  if [[ -z "$arm64_found" ]]; then
-                      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
+      - 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 }}

config/redis.yml

@@ -1,8 +1,7 @@
 address: [ localhost:16379 ]
 username: 
 password: openIM123
-# redisMode can be "cluster", "sentinel", or "standalone"
-redisMode: "standalone"
+clusterMode: false
 db: 0
 maxRetry: 10
 poolSize: 100

docker-compose.yml

@@ -63,12 +63,7 @@ services:
     restart: always
     sysctls:
       net.core.somaxconn: 1024
-    command: >
-      redis-server
-      --requirepass openIM123
-      --appendonly yes
-      --aof-use-rdb-preamble yes
-      --save ""
+    command: redis-server /usr/local/redis/config/redis.conf --requirepass openIM123 --appendonly yes
     networks:
       - openim
 
@@ -104,62 +99,15 @@ services:
     environment:
       #KAFKA_HEAP_OPTS: "-Xms128m -Xmx256m"
       TZ: Asia/Shanghai
-      # Unique identifier for the Kafka node (required in controller mode)
       KAFKA_CFG_NODE_ID: 0
-      # Defines the roles this Kafka node plays: broker, controller, or both
       KAFKA_CFG_PROCESS_ROLES: controller,broker
-      # Specifies which nodes are controller nodes for quorum voting.
-      # The syntax follows the KRaft mode (no ZooKeeper): node.id@host:port
-      # The controller listener endpoint here is kafka:9093
       KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 0@kafka:9093
-      # Specifies which listener is used for controller-to-controller communication
+      KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094
+      KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,EXTERNAL://localhost:19094
+      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT
       KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
-      # Default number of partitions for new topics
       KAFKA_NUM_PARTITIONS: 8
-      # Whether to enable automatic topic creation
       KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: "true"
-      # Kafka internal listeners; Kafka supports multiple ports with different protocols
-      # Each port is used for a specific purpose: INTERNAL for internal broker communication,
-      # CONTROLLER for controller communication, EXTERNAL for external client connections.
-      # These logical listener names are mapped to actual protocols via KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP
-      # In short, Kafka is listening on three logical ports: 9092 for internal communication,
-      # 9093 for controller traffic, and 9094 for external access.
-      KAFKA_CFG_LISTENERS: "INTERNAL://:9092,CONTROLLER://:9093,EXTERNAL://:9094"
-      # Addresses advertised to clients. INTERNAL://kafka:9092 uses the internal Docker service name 'kafka',
-      # so other containers can access Kafka via kafka:9092.
-      # EXTERNAL://localhost:19094 is the address external clients (e.g., in the LAN) should use to connect.
-      # If Kafka is deployed on a different machine than IM, 'localhost' should be replaced with the LAN IP.
-      KAFKA_CFG_ADVERTISED_LISTENERS: "INTERNAL://kafka:9092,EXTERNAL://localhost:19094"
-      # Maps logical listener names to actual protocols.
-      # Supported protocols include: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL
-      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: "CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT"
-      # 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"
-    command: >
-      /bin/sh -c '
-        if [ -n "$${KAFKA_USERNAME}" ] && [ -n "$${KAFKA_PASSWORD}" ]; then
-          echo "=== Kafka SASL Authentication ENABLED ==="
-          echo "Username: $${KAFKA_USERNAME}"
-          
-          # Set environment variables for SASL authentication
-          export KAFKA_CFG_LISTENERS="SASL_PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094"
-          export KAFKA_CFG_ADVERTISED_LISTENERS="SASL_PLAINTEXT://kafka:9092,EXTERNAL://localhost:19094"
-          export KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP="CONTROLLER:PLAINTEXT,EXTERNAL:SASL_PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT"
-          export KAFKA_CFG_SASL_ENABLED_MECHANISMS="PLAIN"
-          export KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL="PLAIN"
-          export KAFKA_CFG_INTER_BROKER_LISTENER_NAME="SASL_PLAINTEXT"
-          export KAFKA_CLIENT_USERS="$${KAFKA_USERNAME}"
-          export KAFKA_CLIENT_PASSWORDS="$${KAFKA_PASSWORD}"
-        fi
-        
-        # Start Kafka with the configured environment
-        exec /opt/bitnami/scripts/kafka/entrypoint.sh /opt/bitnami/scripts/kafka/run.sh
-      '
     networks:
       - openim
 

go.mod

@@ -13,14 +13,14 @@ require (
 	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.12
-	github.com/openimsdk/tools v0.0.50-alpha.106
+	github.com/openimsdk/tools v0.0.50-alpha.84
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0
-	github.com/stretchr/testify v1.10.0
+	github.com/stretchr/testify v1.9.0
 	go.mongodb.org/mongo-driver v1.14.0
 	google.golang.org/api v0.170.0
-	google.golang.org/grpc v1.71.0
-	google.golang.org/protobuf v1.36.4
+	google.golang.org/grpc v1.68.0
+	google.golang.org/protobuf v1.35.1
 	gopkg.in/yaml.v3 v3.0.1
 )
 
@@ -40,13 +40,14 @@ require (
 	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.10.0
+	golang.org/x/sync v0.8.0
 )
 
 require (
 	cloud.google.com/go v0.112.1 // indirect
-	cloud.google.com/go/compute/metadata v0.6.0 // indirect
+	cloud.google.com/go/compute/metadata v0.5.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
@@ -88,7 +89,7 @@ require (
 	github.com/eapache/queue v1.1.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.9.0 // indirect
+	github.com/fsnotify/fsnotify v1.7.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
@@ -155,7 +156,6 @@ 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/sourcegraph/conc v0.3.0 // indirect
 	github.com/spf13/afero v1.11.0 // indirect
 	github.com/spf13/cast v1.6.0 // indirect
@@ -174,29 +174,27 @@ require (
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.13 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.13 // indirect
-	go.etcd.io/etcd/client/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.34.0 // indirect
-	go.opentelemetry.io/otel/metric v1.34.0 // indirect
-	go.opentelemetry.io/otel/trace v1.34.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.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.34.0 // indirect
-	golang.org/x/oauth2 v0.25.0 // indirect
-	golang.org/x/sys v0.29.0 // indirect
-	golang.org/x/term v0.28.0 // indirect
-	golang.org/x/text v0.21.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/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-20250106144421-5f5ef82da422 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20250115164207-1a7da9e5054f // 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
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gorm.io/gorm v1.25.8 // indirect
@@ -218,6 +216,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.32.0 // indirect
+	golang.org/x/crypto v0.27.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.6.0 h1:A6hENjEsCDtC1k8byVsgwvVcioamEHvZ4j01OwKxG9I=
-cloud.google.com/go/compute/metadata v0.6.0/go.mod h1:FjyFAW1MW0C203CEOMDTu3Dk1FlqW3Rga40jzHL4hfg=
+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/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=
@@ -117,8 +117,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.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k=
-github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0=
+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/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=
@@ -349,8 +349,8 @@ github.com/openimsdk/gomake v0.0.15-alpha.5 h1:eEZCEHm+NsmcO3onXZPIUbGFCYPYbsX5b
 github.com/openimsdk/gomake v0.0.15-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
 github.com/openimsdk/protocol v0.0.73-alpha.12 h1:2NYawXeHChYUeSme6QJ9pOLh+Empce2WmwEtbP4JvKk=
 github.com/openimsdk/protocol v0.0.73-alpha.12/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw=
-github.com/openimsdk/tools v0.0.50-alpha.106 h1:gaqU08IbRxOdL16ZEQNyYLhCTf7K1HNkrik8KZLA+BM=
-github.com/openimsdk/tools v0.0.50-alpha.106/go.mod h1:x9i/e+WJFW4tocy6RNJQ9NofQiP3KJ1Y576/06TqOG4=
+github.com/openimsdk/tools v0.0.50-alpha.84 h1:jN60Ys/0edZjL/TDmm/5VSJFP4pGYRipkWqhILJbq/8=
+github.com/openimsdk/tools v0.0.50-alpha.84/go.mod h1:n2poR3asX1e1XZce4O+MOWAp+X02QJRFvhcLCXZdzRo=
 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=
@@ -384,8 +384,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.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII=
-github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o=
+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/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 +393,6 @@ 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/sourcegraph/conc v0.3.0 h1:OQTbbt6P72L20UqAkXXuLOj79LfEanQ+YQFNpLA9ySo=
@@ -422,9 +420,8 @@ 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.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
-github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 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=
@@ -464,22 +461,18 @@ 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.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.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.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=
@@ -500,8 +493,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.32.0 h1:euUpcYgM8WcP71gNpTqQCn6rC2t6ULUPiOzfWaXVVfc=
-golang.org/x/crypto v0.32.0/go.mod h1:ZnnJkOaASj8g0AjIduWNlq2NRxL0PlBrbKVyZ6V/Ugc=
+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/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=
@@ -529,19 +522,19 @@ 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.34.0 h1:Mb7Mrk043xzHgnRM88suvJFwzVrRfHEHJEl5/71CKw0=
-golang.org/x/net v0.34.0/go.mod h1:di0qlW3YNM5oh6GqDGQr92MyTozJPmybPK4Ev/Gm31k=
+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/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
-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/oauth2 v0.23.0 h1:PbgcYx2W7i4LvjJWEbf0ngHV6qJYr86PkAV3bXdLEbs=
+golang.org/x/oauth2 v0.23.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.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
-golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sync v0.8.0 h1:3NFvSEYkUoMifnESzZl15y791HH1qU2xm6eCJU5ZPXQ=
+golang.org/x/sync v0.8.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=
@@ -556,14 +549,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.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU=
-golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+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/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.28.0 h1:/Ts8HFuMR2E6IP/jlo7QVLZHggjKQbhu/7H0LJFr3Gg=
-golang.org/x/term v0.28.0/go.mod h1:Sw/lC2IAUZ92udQNf3WodGtn4k/XoLyZoh8v/8uiwek=
+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/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=
@@ -572,8 +565,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.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
-golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
+golang.org/x/text v0.18.0 h1:XvMDiNzPAl0jr17s6W9lcaIhGUfUORdGCNsuLmPG224=
+golang.org/x/text v0.18.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
 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=
@@ -604,17 +597,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-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/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/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.71.0 h1:kF77BGdPTQ4/JZWMlb9VpJ5pa25aqvVqogsxNHHdeBg=
-google.golang.org/grpc v1.71.0/go.mod h1:H0GRtasmQOh9LkFoCPDu3ZrwUtD1YGE+b2vYBYd/8Ec=
+google.golang.org/grpc v1.68.0 h1:aHQeeJbo8zAkAa3pRzrVjZlbz6uSfeOXlJNQM0RAbz0=
+google.golang.org/grpc v1.68.0/go.mod h1:fmSPC5AsjSBCK54MyHRx48kpOti1/jRfOlwEWywNjWA=
 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=
@@ -624,8 +617,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.36.4 h1:6A3ZDJHn/eNqc1i+IdefRzy/9PokBTPvcqMySR7NNIM=
-google.golang.org/protobuf v1.36.4/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
+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=
 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/init.go

@@ -94,16 +94,9 @@ func Start(ctx context.Context, index int, cfg *Config) error {
 				return err
 			}
 
-			etcdClient, ok := client.(*etcd.SvcDiscoveryRegistryImpl)
-			if !ok {
-				return errs.New("only etcd support autoSetPorts").Wrap()
-			}
+			etcdClient := client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
 
-			target, err := jsonutil.JsonMarshal(prommetrics.BuildDefaultTarget(registerIP, prometheusPort))
-			if err != nil {
-				return errs.Wrap(err)
-			}
-			err = etcdClient.SetWithLease(ctx, prommetrics.BuildDiscoveryKey(prommetrics.APIKeyName, index), target, prommetrics.TTL)
+			_, err = etcdClient.Put(ctx, prommetrics.BuildDiscoveryKey(prommetrics.APIKeyName), jsonutil.StructToJsonString(prommetrics.BuildDefaultTarget(registerIP, prometheusPort)))
 			if err != nil {
 				return errs.WrapMsg(err, "etcd put err")
 			}

internal/api/jssdk/jssdk.go

@@ -2,14 +2,10 @@ package jssdk
 
 import (
 	"context"
+	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
 	"sort"
 
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
-	"github.com/openimsdk/protocol/constant"
-	"github.com/openimsdk/tools/log"
-
 	"github.com/gin-gonic/gin"
-
 	"github.com/openimsdk/protocol/conversation"
 	"github.com/openimsdk/protocol/jssdk"
 	"github.com/openimsdk/protocol/msg"
@@ -113,7 +109,10 @@ func (x *JSSdk) getActiveConversations(ctx context.Context, req *jssdk.GetActive
 	if len(conversationIDs) == 0 {
 		return &jssdk.GetActiveConversationsResp{}, nil
 	}
-
+	readSeq, err := x.msgClient.GetHasReadSeqs(ctx, conversationIDs, req.OwnerUserID)
+	if err != nil {
+		return nil, err
+	}
 	activeConversation, err := x.msgClient.GetActiveConversation(ctx, conversationIDs)
 	if err != nil {
 		return nil, err
@@ -121,10 +120,6 @@ func (x *JSSdk) getActiveConversations(ctx context.Context, req *jssdk.GetActive
 	if len(activeConversation) == 0 {
 		return &jssdk.GetActiveConversationsResp{}, nil
 	}
-	readSeq, err := x.msgClient.GetHasReadSeqs(ctx, conversationIDs, req.OwnerUserID)
-	if err != nil {
-		return nil, err
-	}
 	sortConversations := sortActiveConversations{
 		Conversation: activeConversation,
 	}
@@ -152,7 +147,6 @@ func (x *JSSdk) getActiveConversations(ctx context.Context, req *jssdk.GetActive
 	if err != nil {
 		return nil, err
 	}
-	x.checkMessagesAndGetLastMessage(ctx, req.OwnerUserID, msgs)
 	conversationMap := datautil.SliceToMap(conversations, func(c *conversation.Conversation) string {
 		return c.ConversationID
 	})
@@ -162,15 +156,16 @@ func (x *JSSdk) getActiveConversations(ctx context.Context, req *jssdk.GetActive
 		if !ok {
 			continue
 		}
+		var lastMsg *sdkws.MsgData
 		if msgList, ok := msgs[c.ConversationID]; ok && len(msgList.Msgs) > 0 {
-			resp = append(resp, &jssdk.ConversationMsg{
-				Conversation: conv,
-				LastMsg:      msgList.Msgs[0],
-				MaxSeq:       c.MaxSeq,
-				ReadSeq:      readSeq[c.ConversationID],
-			})
+			lastMsg = msgList.Msgs[0]
 		}
-
+		resp = append(resp, &jssdk.ConversationMsg{
+			Conversation: conv,
+			LastMsg:      lastMsg,
+			MaxSeq:       c.MaxSeq,
+			ReadSeq:      readSeq[c.ConversationID],
+		})
 	}
 	if err := x.fillConversations(ctx, resp); err != nil {
 		return nil, err
@@ -224,18 +219,18 @@ func (x *JSSdk) getConversations(ctx context.Context, req *jssdk.GetConversation
 			return nil, err
 		}
 	}
-	x.checkMessagesAndGetLastMessage(ctx, req.OwnerUserID, msgs)
 	resp := make([]*jssdk.ConversationMsg, 0, len(conversations))
 	for _, c := range conversations {
+		var lastMsg *sdkws.MsgData
 		if msgList, ok := msgs[c.ConversationID]; ok && len(msgList.Msgs) > 0 {
-			resp = append(resp, &jssdk.ConversationMsg{
-				Conversation: c,
-				LastMsg:      msgList.Msgs[0],
-				MaxSeq:       maxSeqs[c.ConversationID],
-				ReadSeq:      readSeqs[c.ConversationID],
-			})
+			lastMsg = msgList.Msgs[0]
 		}
-
+		resp = append(resp, &jssdk.ConversationMsg{
+			Conversation: c,
+			LastMsg:      lastMsg,
+			MaxSeq:       maxSeqs[c.ConversationID],
+			ReadSeq:      readSeqs[c.ConversationID],
+		})
 	}
 	if err := x.fillConversations(ctx, resp); err != nil {
 		return nil, err
@@ -252,38 +247,3 @@ func (x *JSSdk) getConversations(ctx context.Context, req *jssdk.GetConversation
 		UnreadCount:   unreadCount,
 	}, nil
 }
-
-// This function checks whether the latest MaxSeq message is valid.
-// If not, it needs to fetch a valid message again.
-func (x *JSSdk) checkMessagesAndGetLastMessage(ctx context.Context, userID string, messages map[string]*sdkws.PullMsgs) {
-	var conversationIDs []string
-
-	for conversationID, message := range messages {
-		allInValid := true
-		for _, data := range message.Msgs {
-			if data.Status < constant.MsgStatusHasDeleted {
-				allInValid = false
-				break
-			}
-		}
-
-		// when the conversation has been deleted by the user, the length of message.Msgs is empty
-		if allInValid && len(message.Msgs) > 0 {
-			conversationIDs = append(conversationIDs, conversationID)
-		}
-	}
-	if len(conversationIDs) > 0 {
-		resp, err := x.msgClient.GetLastMessage(ctx, &msg.GetLastMessageReq{
-			UserID:          userID,
-			ConversationIDs: conversationIDs,
-		})
-		if err != nil {
-			log.ZError(ctx, "fetchLatestValidMessages", err, "conversationIDs", conversationIDs)
-			return
-		}
-		for conversationID, message := range resp.Msgs {
-			messages[conversationID] = &sdkws.PullMsgs{Msgs: []*sdkws.MsgData{message}}
-		}
-	}
-
-}

internal/api/prometheus_discovery.go

@@ -2,7 +2,6 @@ package api
 
 import (
 	"encoding/json"
-	"errors"
 	"net/http"
 
 	"github.com/gin-gonic/gin"
@@ -12,11 +11,13 @@ import (
 	"github.com/openimsdk/tools/discovery"
 	"github.com/openimsdk/tools/discovery/etcd"
 	"github.com/openimsdk/tools/errs"
+	"github.com/openimsdk/tools/log"
+	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
 type PrometheusDiscoveryApi struct {
 	config *Config
-	client discovery.SvcDiscoveryRegistry
+	client *clientv3.Client
 }
 
 func NewPrometheusDiscoveryApi(cfg *Config, client discovery.SvcDiscoveryRegistry) *PrometheusDiscoveryApi {
@@ -24,7 +25,7 @@ func NewPrometheusDiscoveryApi(cfg *Config, client discovery.SvcDiscoveryRegistr
 		config: cfg,
 	}
 	if cfg.Discovery.Enable == config.ETCD {
-		api.client = client.(*etcd.SvcDiscoveryRegistryImpl)
+		api.client = client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
 	}
 	return api
 }
@@ -37,32 +38,35 @@ func (p *PrometheusDiscoveryApi) Enable(c *gin.Context) {
 }
 
 func (p *PrometheusDiscoveryApi) discovery(c *gin.Context, key string) {
-	value, err := p.client.GetKeyWithPrefix(c, prommetrics.BuildDiscoveryKeyPrefix(key))
+	eResp, err := p.client.Get(c, prommetrics.BuildDiscoveryKey(key))
 	if err != nil {
-		if errors.Is(err, discovery.ErrNotSupported) {
-			c.JSON(http.StatusOK, []struct{}{})
-			return
-		}
-		apiresp.GinError(c, errs.WrapMsg(err, "get key value"))
+		// Log and respond with an error if preparation fails.
+		apiresp.GinError(c, errs.WrapMsg(err, "etcd get err"))
 		return
 	}
-	if len(value) == 0 {
-		c.JSON(http.StatusOK, []*prommetrics.RespTarget{})
-		return
+	if len(eResp.Kvs) == 0 {
+		c.JSON(http.StatusOK, []*prommetrics.Target{})
 	}
-	var resp prommetrics.RespTarget
-	for i := range value {
-		var tmp prommetrics.Target
-		if err = json.Unmarshal(value[i], &tmp); err != nil {
-			apiresp.GinError(c, errs.WrapMsg(err, "json unmarshal err"))
-			return
+
+	var (
+		resp = &prommetrics.RespTarget{
+			Targets: make([]string, 0, len(eResp.Kvs)),
 		}
+	)
 
-		resp.Targets = append(resp.Targets, tmp.Target)
-		resp.Labels = tmp.Labels // default label is fixed. See prommetrics.BuildDefaultTarget
+	for i := range eResp.Kvs {
+		var target prommetrics.Target
+		err = json.Unmarshal(eResp.Kvs[i].Value, &target)
+		if err != nil {
+			log.ZError(c, "prometheus unmarshal err", errs.Wrap(err))
+		}
+		resp.Targets = append(resp.Targets, target.Target)
+		if resp.Labels == nil {
+			resp.Labels = target.Labels
+		}
 	}
 
-	c.JSON(http.StatusOK, []*prommetrics.RespTarget{&resp})
+	c.JSON(200, []*prommetrics.RespTarget{resp})
 }
 
 func (p *PrometheusDiscoveryApi) Api(c *gin.Context) {

internal/msggateway/client.go

@@ -16,6 +16,7 @@ package msggateway
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"sync"
 	"sync/atomic"
@@ -63,7 +64,7 @@ type PingPongHandler func(string) error
 
 type Client struct {
 	w              *sync.Mutex
-	conn           ClientConn
+	conn           LongConn
 	PlatformID     int    `json:"platformID"`
 	IsCompress     bool   `json:"isCompress"`
 	UserID         string `json:"userID"`
@@ -82,7 +83,7 @@ type Client struct {
 }
 
 // ResetClient updates the client's state with new connection and context information.
-func (c *Client) ResetClient(ctx *UserConnContext, conn ClientConn, longConnServer LongConnServer) {
+func (c *Client) ResetClient(ctx *UserConnContext, conn LongConn, longConnServer LongConnServer) {
 	c.w = new(sync.Mutex)
 	c.conn = conn
 	c.PlatformID = stringutil.StringToInt(ctx.GetPlatformID())
@@ -109,6 +110,22 @@ func (c *Client) ResetClient(ctx *UserConnContext, conn ClientConn, longConnServ
 	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() {
@@ -119,25 +136,52 @@ 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")
-		message, returnErr := c.conn.ReadMessage()
+		messageType, message, returnErr := c.conn.ReadMessage()
 		if returnErr != nil {
-			log.ZWarn(c.ctx, "readMessage", returnErr)
+			log.ZWarn(c.ctx, "readMessage", returnErr, "messageType", messageType)
 			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
 		}
 
-		parseDataErr := c.handleMessage(message)
-		if parseDataErr != nil {
-			c.closedErr = parseDataErr
+		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
 			return
+
+		default:
 		}
 	}
 }
@@ -312,13 +356,109 @@ 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(resultBuf)
+		return c.conn.WriteMessage(MessageBinary, resultBuf)
 	}
 
-	return c.conn.WriteMessage(encodedBuf)
+	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)
+	}
 }

internal/msggateway/client_conn.go

@@ -1,212 +0,0 @@
-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 {
-	_ = c.closeBy(fmt.Errorf("websocket connection closed"))
-	return nil
-}
-
-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))
-	_ = c.conn.Close()
-	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() {
-	var err error
-	for {
-		select {
-		case <-c.done:
-			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.ZWarn(context.Background(), "ping handler recv ping", nil, "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.ZWarn(context.Background(), "ping handler write pong success", nil, "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/hub_server.go

@@ -244,7 +244,6 @@ func (s *Server) MultiTerminalLoginCheck(ctx context.Context, req *msggateway.Mu
 		tempUserCtx.SetOperationID(mcontext.GetOperationID(ctx))
 		client := &Client{}
 		client.ctx = tempUserCtx
-		client.token = req.Token
 		client.UserID = req.UserID
 		client.PlatformID = int(req.PlatformID)
 		i := &kickHandler{

internal/msggateway/long_conn.go

@@ -0,0 +1,179 @@
+// 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,17 +2,13 @@ package msggateway
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"net/http"
-	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
 
-	"github.com/gorilla/websocket"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
-	"github.com/openimsdk/tools/apiresp"
 
 	"github.com/go-playground/validator/v10"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
@@ -46,7 +42,6 @@ type LongConnServer interface {
 }
 
 type WsServer struct {
-	websocket         *websocket.Upgrader
 	msgGatewayConfig  *Config
 	port              int
 	wsMaxConnNum      int64
@@ -136,13 +131,9 @@ func NewWsServer(msgGatewayConfig *Config, opts ...Option) *WsServer {
 		o(&config)
 	}
 	//userRpcClient := rpcclient.NewUserRpcClient(client, config.Share.RpcRegisterName.User, config.Share.IMAdminUserID)
-	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,
@@ -346,51 +337,17 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien
 		}
 	}
 
-	// If reconnect: When multiple msgGateway instances are deployed, a client may disconnect from instance A and reconnect to instance B.
-	// During this process, instance A might still be executing, resulting in two clients with the same token existing simultaneously.
-	// This situation needs to be filtered to prevent duplicate clients.
-	checkSameTokenFunc := func(oldClients []*Client) []*Client {
-		var clientsNeedToKick []*Client
-
-		for _, c := range oldClients {
-			if c.token == newClient.token {
-				log.ZDebug(newClient.ctx, "token is same, not kick",
-					"userID", newClient.UserID,
-					"platformID", newClient.PlatformID,
-					"token", newClient.token)
-				continue
-			}
-
-			clientsNeedToKick = append(clientsNeedToKick, c)
-		}
-
-		return clientsNeedToKick
-	}
-
 	switch ws.msgGatewayConfig.Share.MultiLogin.Policy {
 	case constant.DefalutNotKick:
 	case constant.PCAndOther:
 		if constant.PlatformIDToClass(newClient.PlatformID) == constant.TerminalPC {
 			return
 		}
-		clients, ok := ws.clients.GetAll(newClient.UserID)
-		clientOK = ok
-		oldClients = make([]*Client, 0, len(clients))
-		for _, c := range clients {
-			if constant.PlatformIDToClass(c.PlatformID) == constant.TerminalPC {
-				continue
-			}
-			oldClients = append(oldClients, c)
-		}
-
 		fallthrough
 	case constant.AllLoginButSameTermKick:
 		if !clientOK {
 			return
 		}
-
-		oldClients = checkSameTokenFunc(oldClients)
-
 		ws.clients.DeleteClients(newClient.UserID, oldClients)
 		for _, c := range oldClients {
 			err := c.KickOnlineMessage()
@@ -416,15 +373,14 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien
 		if !ok {
 			return
 		}
-
-		var kickClients []*Client
+		var (
+			kickClients []*Client
+		)
 		for _, client := range clients {
 			if constant.PlatformIDToClass(client.PlatformID) == constant.PlatformIDToClass(newClient.PlatformID) {
 				kickClients = append(kickClients, client)
 			}
 		}
-		kickClients = checkSameTokenFunc(kickClients)
-
 		kickTokenFunc(kickClients)
 	}
 }
@@ -458,29 +414,6 @@ 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)
@@ -488,7 +421,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
-		ws.handlerError(connContext, w, r, servererrs.ErrConnOverMaxNumLimit.WrapMsg("over max conn num limit"))
+		httpError(connContext, servererrs.ErrConnOverMaxNumLimit.WrapMsg("over max conn num limit"))
 		return
 	}
 
@@ -496,14 +429,26 @@ 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
-		ws.handlerError(connContext, w, r, err)
+
+		httpError(connContext, 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 {
-		ws.handlerError(connContext, w, r, err)
+		// 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)
 		return
 	}
 
@@ -511,24 +456,32 @@ 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
-		ws.handlerError(connContext, w, r, err)
+		httpError(connContext, err)
 		return
 	}
-	conn, err := ws.websocket.Upgrade(w, r, nil)
-	if err != nil {
-		log.ZWarn(connContext, "websocket upgrade failed", err)
-		return
-	}
-	log.ZDebug(connContext, "new conn", "token", connContext.GetToken())
 
-	var pingInterval time.Duration
-	if connContext.GetPlatformID() == strconv.Itoa(constant.WebPlatformID) {
-		pingInterval = pingPeriod
+	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)
+		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
+			}
+		}
 	}
 
 	// 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, NewWebSocketClientConn(conn, maxMessageSize, pongWait, pingInterval), ws)
+	client.ResetClient(connContext, wsLongConn, ws)
 
 	// Register the client with the server and start message processing
 	ws.registerChan <- client

internal/msgtransfer/init.go

@@ -176,16 +176,9 @@ func (m *MsgTransfer) Start(index int, cfg *Config) error {
 				return err
 			}
 
-			etcdClient, ok := client.(*etcd.SvcDiscoveryRegistryImpl)
-			if !ok {
-				return errs.New("only etcd support autoSetPorts").Wrap()
-			}
+			etcdClient := client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
 
-			target, err := jsonutil.JsonMarshal(prommetrics.BuildDefaultTarget(registerIP, prometheusPort))
-			if err != nil {
-				return errs.Wrap(err)
-			}
-			err = etcdClient.SetWithLease(context.TODO(), prommetrics.BuildDiscoveryKey(prommetrics.MessageTransferKeyName, index), target, prommetrics.TTL)
+			_, err = etcdClient.Put(context.TODO(), prommetrics.BuildDiscoveryKey(prommetrics.MessageTransferKeyName), jsonutil.StructToJsonString(prommetrics.BuildDefaultTarget(registerIP, prometheusPort)))
 			if err != nil {
 				return errs.WrapMsg(err, "etcd put err")
 			}

internal/msgtransfer/online_history_msg_handler.go

@@ -26,12 +26,10 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
 	"github.com/openimsdk/tools/discovery"
 
-	"github.com/go-redis/redis"
-	"google.golang.org/protobuf/proto"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 
 	"github.com/IBM/sarama"
+	"github.com/go-redis/redis"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/kafka"
 	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
@@ -43,6 +41,7 @@ import (
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/mcontext"
 	"github.com/openimsdk/tools/utils/stringutil"
+	"google.golang.org/protobuf/proto"
 )
 
 const (
@@ -141,48 +140,53 @@ func (och *OnlineHistoryRedisConsumerHandler) do(ctx context.Context, channelID
 
 func (och *OnlineHistoryRedisConsumerHandler) doSetReadSeq(ctx context.Context, msgs []*ContextMsg) {
 
-	// Outer map: conversationID -> (userID -> maxHasReadSeq)
-	conversationUserSeq := make(map[string]map[string]int64)
-
+	var conversationID string
+	var userSeqMap map[string]int64
 	for _, msg := range msgs {
 		if msg.message.ContentType != constant.HasReadReceipt {
 			continue
 		}
 		var elem sdkws.NotificationElem
 		if err := json.Unmarshal(msg.message.Content, &elem); err != nil {
-			log.ZWarn(ctx, "Unmarshal NotificationElem error", err, "msg", msg)
+			log.ZWarn(ctx, "handlerConversationRead Unmarshal NotificationElem msg err", err, "msg", msg)
 			continue
 		}
 		var tips sdkws.MarkAsReadTips
 		if err := json.Unmarshal([]byte(elem.Detail), &tips); err != nil {
-			log.ZWarn(ctx, "Unmarshal MarkAsReadTips error", err, "msg", msg)
+			log.ZWarn(ctx, "handlerConversationRead Unmarshal MarkAsReadTips msg err", err, "msg", msg)
 			continue
 		}
-		if len(tips.ConversationID) == 0 || tips.HasReadSeq < 0 {
-			continue
-		}
-
-		// Calculate the max seq from tips.Seqs
-		for _, seq := range tips.Seqs {
-			if tips.HasReadSeq < seq {
-				tips.HasReadSeq = seq
+		//The conversation ID for each batch of messages processed by the batcher is the same.
+		conversationID = tips.ConversationID
+		if len(tips.Seqs) > 0 {
+			for _, seq := range tips.Seqs {
+				if tips.HasReadSeq < seq {
+					tips.HasReadSeq = seq
+				}
 			}
+			clear(tips.Seqs)
+			tips.Seqs = nil
+		}
+		if tips.HasReadSeq < 0 {
+			continue
+		}
+		if userSeqMap == nil {
+			userSeqMap = make(map[string]int64)
 		}
 
-		if _, ok := conversationUserSeq[tips.ConversationID]; !ok {
-			conversationUserSeq[tips.ConversationID] = make(map[string]int64)
-		}
-		if conversationUserSeq[tips.ConversationID][tips.MarkAsReadUserID] < tips.HasReadSeq {
-			conversationUserSeq[tips.ConversationID][tips.MarkAsReadUserID] = tips.HasReadSeq
+		if userSeqMap[tips.MarkAsReadUserID] > tips.HasReadSeq {
+			continue
 		}
+		userSeqMap[tips.MarkAsReadUserID] = tips.HasReadSeq
 	}
-	log.ZInfo(ctx, "doSetReadSeq", "conversationUserSeq", conversationUserSeq)
-
-	// persist to db
-	for convID, userSeqMap := range conversationUserSeq {
-		if err := och.msgTransferDatabase.SetHasReadSeqToDB(ctx, convID, userSeqMap); err != nil {
-			log.ZWarn(ctx, "SetHasReadSeqToDB error", err, "conversationID", convID, "userSeqMap", userSeqMap)
-		}
+	if userSeqMap == nil {
+		return
+	}
+	if len(conversationID) == 0 {
+		log.ZWarn(ctx, "conversation err", nil, "conversationID", conversationID)
+	}
+	if err := och.msgTransferDatabase.SetHasReadSeqToDB(ctx, conversationID, userSeqMap); err != nil {
+		log.ZWarn(ctx, "set read seq to db error", err, "conversationID", conversationID, "userSeqMap", userSeqMap)
 	}
 
 }

internal/rpc/conversation/conversation.go

@@ -168,18 +168,7 @@ func (c *conversationServer) GetSortedConversationList(ctx context.Context, req
 	conversation_notPinTime := make(map[int64]string)
 	for _, v := range conversations {
 		conversationID := v.ConversationID
-		var time int64
-		if _, ok := conversationMsg[conversationID]; ok {
-			time = conversationMsg[conversationID].MsgInfo.LatestMsgRecvTime
-		} else {
-			conversationMsg[conversationID] = &pbconversation.ConversationElem{
-				ConversationID: conversationID,
-				IsPinned:       v.IsPinned,
-				MsgInfo:        nil,
-			}
-			time = v.CreateTime.UnixMilli()
-		}
-
+		time := conversationMsg[conversationID].MsgInfo.LatestMsgRecvTime
 		conversationMsg[conversationID].RecvMsgOpt = v.RecvMsgOpt
 		if v.IsPinned {
 			conversationMsg[conversationID].IsPinned = v.IsPinned
@@ -231,7 +220,6 @@ func (c *conversationServer) getConversations(ctx context.Context, ownerUserID s
 	return convert.ConversationsDB2Pb(conversations), nil
 }
 
-// Deprecated
 func (c *conversationServer) SetConversation(ctx context.Context, req *pbconversation.SetConversationReq) (*pbconversation.SetConversationResp, error) {
 	var conversation dbModel.Conversation
 	if err := datautil.CopyStructFields(&conversation, req.Conversation); err != nil {

internal/rpc/conversation/sync.go

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

internal/rpc/group/notification.go

@@ -476,15 +476,14 @@ func (g *NotificationSender) GroupApplicationAcceptedNotification(ctx context.Co
 	if err = g.fillOpUser(ctx, &opUser, group.GroupID); err != nil {
 		return
 	}
-	uid := g.uuid()
+	tips := &sdkws.GroupApplicationAcceptedTips{
+		Group:     group,
+		OpUser:    opUser,
+		HandleMsg: req.HandledMsg,
+		Uuid:      g.uuid(),
+		Request:   request,
+	}
 	for _, userID := range append(userIDs, req.FromUserID) {
-		tips := &sdkws.GroupApplicationAcceptedTips{
-			Group:     group,
-			OpUser:    opUser,
-			HandleMsg: req.HandledMsg,
-			Uuid:      uid,
-			Request:   request,
-		}
 		if userID == req.FromUserID {
 			tips.ReceiverAs = applicantReceiver
 		} else {
@@ -521,15 +520,14 @@ func (g *NotificationSender) GroupApplicationRejectedNotification(ctx context.Co
 	if err = g.fillOpUser(ctx, &opUser, group.GroupID); err != nil {
 		return
 	}
-	uid := g.uuid()
+	tips := &sdkws.GroupApplicationRejectedTips{
+		Group:     group,
+		OpUser:    opUser,
+		HandleMsg: req.HandledMsg,
+		Uuid:      g.uuid(),
+		Request:   request,
+	}
 	for _, userID := range append(userIDs, req.FromUserID) {
-		tips := &sdkws.GroupApplicationRejectedTips{
-			Group:     group,
-			OpUser:    opUser,
-			HandleMsg: req.HandledMsg,
-			Uuid:      uid,
-			Request:   request,
-		}
 		if userID == req.FromUserID {
 			tips.ReceiverAs = applicantReceiver
 		} else {

internal/rpc/group/sync.go

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

internal/rpc/relation/friend.go

@@ -194,7 +194,7 @@ func (s *friendServer) ImportFriends(ctx context.Context, req *relation.ImportFr
 			FromUserID:   req.OwnerUserID,
 			ToUserID:     userID,
 			HandleResult: constant.FriendResponseAgree,
-		}, false)
+		})
 	}
 
 	s.webhookAfterImportFriends(ctx, &s.config.WebhooksConfig.AfterImportFriends, req)
@@ -223,7 +223,7 @@ func (s *friendServer) RespondFriendApply(ctx context.Context, req *relation.Res
 			return nil, err
 		}
 		s.webhookAfterAddFriendAgree(ctx, &s.config.WebhooksConfig.AfterAddFriendAgree, req)
-		s.notificationSender.FriendApplicationAgreedNotification(ctx, req, true)
+		s.notificationSender.FriendApplicationAgreedNotification(ctx, req)
 		return resp, nil
 	}
 	if req.HandleResult == constant.FriendResponseRefuse {

internal/rpc/relation/notification.go

@@ -171,17 +171,11 @@ func (f *FriendNotificationSender) FriendApplicationAddNotification(ctx context.
 	f.Notification(ctx, req.FromUserID, req.ToUserID, constant.FriendApplicationNotification, &tips)
 }
 
-func (f *FriendNotificationSender) FriendApplicationAgreedNotification(ctx context.Context, req *relation.RespondFriendApplyReq, checkReq bool) {
-	var (
-		request *sdkws.FriendRequest
-		err     error
-	)
-	if checkReq {
-		request, err = f.getFriendRequests(ctx, req.FromUserID, req.ToUserID)
-		if err != nil {
-			log.ZError(ctx, "FriendApplicationAgreedNotification get friend request", err, "fromUserID", req.FromUserID, "toUserID", req.ToUserID)
-			return
-		}
+func (f *FriendNotificationSender) FriendApplicationAgreedNotification(ctx context.Context, req *relation.RespondFriendApplyReq) {
+	request, err := f.getFriendRequests(ctx, req.FromUserID, req.ToUserID)
+	if err != nil {
+		log.ZError(ctx, "FriendApplicationAgreedNotification get friend request", err, "fromUserID", req.FromUserID, "toUserID", req.ToUserID)
+		return
 	}
 	tips := sdkws.FriendApplicationApprovedTips{
 		FromToUserID: &sdkws.FromToUserID{

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:   uint64(vl.Version),
+		Version:   idHash,
 		VersionID: vl.ID.Hex(),
 		Equal:     req.IdHash == idHash,
 		UserIDs:   userIDs,

internal/rpc/third/third.go

@@ -23,8 +23,6 @@ import (
 	"github.com/openimsdk/tools/s3/aws"
 	"github.com/openimsdk/tools/s3/disable"
 
-	"google.golang.org/grpc"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller"
@@ -39,6 +37,7 @@ import (
 	"github.com/openimsdk/tools/s3/kodo"
 	"github.com/openimsdk/tools/s3/minio"
 	"github.com/openimsdk/tools/s3/oss"
+	"google.golang.org/grpc"
 )
 
 type thirdServer struct {

pkg/common/config/config.go

@@ -354,13 +354,13 @@ type User struct {
 }
 
 type Redis struct {
-	Address   []string `mapstructure:"address"`
-	Username  string   `mapstructure:"username"`
-	Password  string   `mapstructure:"password"`
-	RedisMode string   `mapstructure:"redisMode"`
-	DB        int      `mapstructure:"db"`
-	MaxRetry  int      `mapstructure:"maxRetry"`
-	PoolSize  int      `mapstructure:"poolSize"`
+	Address     []string `mapstructure:"address"`
+	Username    string   `mapstructure:"username"`
+	Password    string   `mapstructure:"password"`
+	ClusterMode bool     `mapstructure:"clusterMode"`
+	DB          int      `mapstructure:"storage"`
+	MaxRetry    int      `mapstructure:"maxRetry"`
+	PoolSize    int      `mapstructure:"poolSize"`
 }
 
 type BeforeConfig struct {
@@ -511,13 +511,13 @@ func (m *Mongo) Build() *mongoutil.Config {
 
 func (r *Redis) Build() *redisutil.Config {
 	return &redisutil.Config{
-		RedisMode: r.RedisMode,
-		Address:   r.Address,
-		Username:  r.Username,
-		Password:  r.Password,
-		DB:        r.DB,
-		MaxRetry:  r.MaxRetry,
-		PoolSize:  r.PoolSize,
+		ClusterMode: r.ClusterMode,
+		Address:     r.Address,
+		Username:    r.Username,
+		Password:    r.Password,
+		DB:          r.DB,
+		MaxRetry:    r.MaxRetry,
+		PoolSize:    r.PoolSize,
 	}
 }
 

pkg/common/discoveryregister/discoveryregister.go

@@ -29,7 +29,7 @@ import (
 func NewDiscoveryRegister(discovery *config.Discovery, share *config.Share, watchNames []string) (discovery.SvcDiscoveryRegistry, error) {
 	switch discovery.Enable {
 	case "k8s":
-		return kubernetes.NewConnManager("default", watchNames,
+		return kubernetes.NewKubernetesConnManager("default",
 			grpc.WithDefaultCallOptions(
 				grpc.MaxCallSendMsgSize(1024*1024*20),
 			),

pkg/common/prommetrics/discovery.go

@@ -5,7 +5,6 @@ import "fmt"
 const (
 	APIKeyName             = "api"
 	MessageTransferKeyName = "message-transfer"
-	TTL                    = 300
 )
 
 type Target struct {
@@ -18,14 +17,10 @@ type RespTarget struct {
 	Labels  map[string]string `json:"labels"`
 }
 
-func BuildDiscoveryKeyPrefix(name string) string {
+func BuildDiscoveryKey(name string) string {
 	return fmt.Sprintf("%s/%s/%s", "openim", "prometheus_discovery", name)
 }
 
-func BuildDiscoveryKey(name string, index int) string {
-	return fmt.Sprintf("%s/%s/%s/%d", "openim", "prometheus_discovery", name, index)
-}
-
 func BuildDefaultTarget(host string, ip int) Target {
 	return Target{
 		Target: fmt.Sprintf("%s:%d", host, ip),

pkg/common/startrpc/start.go

@@ -164,16 +164,9 @@ func Start[T any](ctx context.Context, discovery *conf.Discovery, prometheusConf
 				return err
 			}
 
-			etcdClient, ok := client.(*etcd.SvcDiscoveryRegistryImpl)
-			if !ok {
-				return errs.New("only etcd support autoSetPorts").Wrap()
-			}
+			etcdClient := client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
 
-			target, err := jsonutil.JsonMarshal(prommetrics.BuildDefaultTarget(registerIP, prometheusPort))
-			if err != nil {
-				return errs.Wrap(err)
-			}
-			err = etcdClient.SetWithLease(ctx, prommetrics.BuildDiscoveryKey(rpcRegisterName, index), target, prommetrics.TTL)
+			_, err = etcdClient.Put(ctx, prommetrics.BuildDiscoveryKey(rpcRegisterName), jsonutil.StructToJsonString(prommetrics.BuildDefaultTarget(registerIP, prometheusPort)))
 			if err != nil {
 				return errs.WrapMsg(err, "etcd put err")
 			}

pkg/common/storage/controller/msg.go

@@ -675,19 +675,12 @@ func (db *commonMsgDatabase) SearchMessage(ctx context.Context, req *pbmsg.Searc
 func (db *commonMsgDatabase) FindOneByDocIDs(ctx context.Context, conversationIDs []string, seqs map[string]int64) (map[string]*sdkws.MsgData, error) {
 	totalMsgs := make(map[string]*sdkws.MsgData)
 	for _, conversationID := range conversationIDs {
-		seq, ok := seqs[conversationID]
-		if !ok {
-			log.ZWarn(ctx, "seq not found for conversationID", errs.New("seq not found for conversation"), "conversationID", conversationID)
-			continue
-		}
-
+		seq := seqs[conversationID]
 		docID := db.msgTable.GetDocID(conversationID, seq)
 		msgs, err := db.msgDocDatabase.FindOneByDocID(ctx, docID)
 		if err != nil {
-			log.ZWarn(ctx, "FindOneByDocID failed", err, "conversationID", conversationID, "docID", docID, "seq", seq)
-			continue
+			return nil, err
 		}
-
 		index := db.msgTable.GetMsgIndex(seq)
 		totalMsgs[conversationID] = convert.MsgDB2Pb(msgs.Msg[index].Msg)
 	}

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.UnixMilli(ts)}
+		filter["create_time"] = bson.M{"$gt": time.Unix(ts, 0)}
 	}
 	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.UnixMilli(ts)}
+		filter["req_time"] = bson.M{"$gt": time.Unix(ts, 0)}
 	}
 	return mongoutil.Count(ctx, g.coll, filter)
 }

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

@@ -5,11 +5,6 @@ import (
 	"fmt"
 	"time"
 
-	"go.mongodb.org/mongo-driver/bson"
-	"go.mongodb.org/mongo-driver/bson/primitive"
-	"go.mongodb.org/mongo-driver/mongo"
-	"go.mongodb.org/mongo-driver/mongo/options"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 	"github.com/openimsdk/protocol/constant"
@@ -19,6 +14,10 @@ import (
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/utils/datautil"
 	"github.com/openimsdk/tools/utils/jsonutil"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
 )
 
 func NewMsgMongo(db *mongo.Database) (database.Msg, error) {
@@ -1150,7 +1149,7 @@ func (m *MsgMgo) findBeforeDocSendTime(ctx context.Context, docID string, limit
 	if err != nil {
 		return 0, 0, err
 	}
-	for i := len(res) - 1; i >= 0; i-- {
+	for i := len(res) - 1; i > 0; i-- {
 		v := res[i]
 		if v.Msgs != nil && v.Msgs.Msg != nil && v.Msgs.Msg.SendTime > 0 {
 			return v.Msgs.Msg.Seq, v.Msgs.Msg.SendTime, nil
@@ -1165,7 +1164,7 @@ func (m *MsgMgo) findBeforeSendTime(ctx context.Context, conversationID string,
 		limit := int64(-1)
 		if first {
 			first = false
-			limit = m.model.GetLimitForSingleDoc(seq)
+			limit = m.model.GetMsgIndex(seq)
 		}
 		docID := m.model.BuildDocIDByIndex(conversationID, i)
 		msgSeq, msgSendTime, err := m.findBeforeDocSendTime(ctx, docID, limit)

pkg/common/storage/model/msg.go

@@ -120,11 +120,15 @@ func (m *MsgDocModel) GetDocID(conversationID string, seq int64) string {
 
 func (m *MsgDocModel) GetDocIDSeqsMap(conversationID string, seqs []int64) map[string][]int64 {
 	t := make(map[string][]int64)
-	for _, seq := range seqs {
-		docID := m.GetDocID(conversationID, seq)
-		t[docID] = append(t[docID], seq)
+	for i := 0; i < len(seqs); i++ {
+		docID := m.GetDocID(conversationID, seqs[i])
+		if value, ok := t[docID]; !ok {
+			var temp []int64
+			t[docID] = append(temp, seqs[i])
+		} else {
+			t[docID] = append(value, seqs[i])
+		}
 	}
-
 	return t
 }
 
@@ -132,10 +136,6 @@ func (*MsgDocModel) GetMsgIndex(seq int64) int64 {
 	return (seq - 1) % singleGocMsgNum
 }
 
-func (*MsgDocModel) GetLimitForSingleDoc(seq int64) int64 {
-	return seq % singleGocMsgNum
-}
-
 func (*MsgDocModel) indexGen(conversationID string, seqSuffix int64) string {
 	return conversationID + ":" + strconv.FormatInt(seqSuffix, 10)
 }

pkg/localcache/cache.go

@@ -49,7 +49,7 @@ func New[V any](opts ...Option) Cache[V] {
 			if opt.expirationEvict {
 				return lru.NewExpirationLRU[string, V](opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
 			} else {
-				return lru.NewLazyLRU[string, V](opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
+				return lru.NewLayLRU[string, V](opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
 			}
 		}
 		if opt.localSlotNum == 1 {
@@ -72,18 +72,11 @@ type cache[V any] struct {
 
 func (c *cache[V]) onEvict(key string, value V) {
 	if c.link != nil {
-		// 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)
+		lks := c.link.Del(key)
+		for k := range lks {
+			if key != k { // prevent deadlock
+				c.local.Del(k)
+			}
 		}
 	}
 }
@@ -110,7 +103,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 && c.link != nil {
+			if len(link) > 0 {
 				c.link.Link(key, link...)
 			}
 			return fetch(ctx)

pkg/localcache/cache_test.go

@@ -22,8 +22,6 @@ import (
 	"sync/atomic"
 	"testing"
 	"time"
-
-	"github.com/openimsdk/open-im-server/v3/pkg/localcache/lru"
 )
 
 func TestName(t *testing.T) {
@@ -93,68 +91,3 @@ 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/lru/lru_expiration.go

@@ -52,53 +52,8 @@ 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) {
-	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
+	//TODO implement me
+	panic("implement me")
 }
 
 func (x *ExpirationLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) {

pkg/localcache/lru/lru_lazy.go

@@ -21,25 +21,25 @@ import (
 	"github.com/hashicorp/golang-lru/v2/simplelru"
 )
 
-type lazyLruItem[V any] struct {
+type layLruItem[V any] struct {
 	lock    sync.Mutex
 	expires int64
 	err     error
 	value   V
 }
 
-func NewLazyLRU[K comparable, V any](size int, successTTL, failedTTL time.Duration, target Target, onEvict EvictCallback[K, V]) *LazyLRU[K, V] {
-	var cb simplelru.EvictCallback[K, *lazyLruItem[V]]
+func NewLayLRU[K comparable, V any](size int, successTTL, failedTTL time.Duration, target Target, onEvict EvictCallback[K, V]) *LayLRU[K, V] {
+	var cb simplelru.EvictCallback[K, *layLruItem[V]]
 	if onEvict != nil {
-		cb = func(key K, value *lazyLruItem[V]) {
+		cb = func(key K, value *layLruItem[V]) {
 			onEvict(key, value.value)
 		}
 	}
-	core, err := simplelru.NewLRU[K, *lazyLruItem[V]](size, cb)
+	core, err := simplelru.NewLRU[K, *layLruItem[V]](size, cb)
 	if err != nil {
 		panic(err)
 	}
-	return &LazyLRU[K, V]{
+	return &LayLRU[K, V]{
 		core:       core,
 		successTTL: successTTL,
 		failedTTL:  failedTTL,
@@ -47,15 +47,15 @@ func NewLazyLRU[K comparable, V any](size int, successTTL, failedTTL time.Durati
 	}
 }
 
-type LazyLRU[K comparable, V any] struct {
+type LayLRU[K comparable, V any] struct {
 	lock       sync.Mutex
-	core       *simplelru.LRU[K, *lazyLruItem[V]]
+	core       *simplelru.LRU[K, *layLruItem[V]]
 	successTTL time.Duration
 	failedTTL  time.Duration
 	target     Target
 }
 
-func (x *LazyLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) {
+func (x *LayLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) {
 	x.lock.Lock()
 	v, ok := x.core.Get(key)
 	if ok {
@@ -68,7 +68,7 @@ func (x *LazyLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) {
 			return value, err
 		}
 	} else {
-		v = &lazyLruItem[V]{}
+		v = &layLruItem[V]{}
 		x.core.Add(key, v)
 		v.lock.Lock()
 		x.lock.Unlock()
@@ -88,15 +88,15 @@ func (x *LazyLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) {
 	return v.value, v.err
 }
 
-func (x *LazyLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) (map[K]V, error) {
+func (x *LayLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) (map[K]V, error) {
 	var (
 		err  error
 		once sync.Once
 	)
 
 	res := make(map[K]V)
-	queries := make([]K, 0, len(keys))
-
+	queries := make([]K, 0)
+	setVs := make(map[K]*layLruItem[V])
 	for _, key := range keys {
 		x.lock.Lock()
 		v, ok := x.core.Get(key)
@@ -118,20 +118,14 @@ func (x *LazyLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)
 		}
 		queries = append(queries, key)
 	}
-
-	if len(queries) == 0 {
-		return res, err
-	}
-
-	values, fetchErr := fetch(queries)
-	if fetchErr != nil {
+	values, err1 := fetch(queries)
+	if err1 != nil {
 		once.Do(func() {
-			err = fetchErr
+			err = err1
 		})
 	}
-
 	for key, val := range values {
-		v := &lazyLruItem[V]{}
+		v := &layLruItem[V]{}
 		v.value = val
 
 		if err == nil {
@@ -141,7 +135,7 @@ func (x *LazyLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)
 			v.expires = time.Now().Add(x.failedTTL).UnixMilli()
 			x.target.IncrGetFailed()
 		}
-
+		setVs[key] = v
 		x.lock.Lock()
 		x.core.Add(key, v)
 		x.lock.Unlock()
@@ -151,29 +145,29 @@ func (x *LazyLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)
 	return res, err
 }
 
-//func (x *LazyLRU[K, V]) Has(key K) bool {
+//func (x *LayLRU[K, V]) Has(key K) bool {
 //	x.lock.Lock()
 //	defer x.lock.Unlock()
 //	return x.core.Contains(key)
 //}
 
-func (x *LazyLRU[K, V]) Set(key K, value V) {
+func (x *LayLRU[K, V]) Set(key K, value V) {
 	x.lock.Lock()
 	defer x.lock.Unlock()
-	x.core.Add(key, &lazyLruItem[V]{value: value, expires: time.Now().Add(x.successTTL).UnixMilli()})
+	x.core.Add(key, &layLruItem[V]{value: value, expires: time.Now().Add(x.successTTL).UnixMilli()})
 }
 
-func (x *LazyLRU[K, V]) SetHas(key K, value V) bool {
+func (x *LayLRU[K, V]) SetHas(key K, value V) bool {
 	x.lock.Lock()
 	defer x.lock.Unlock()
 	if x.core.Contains(key) {
-		x.core.Add(key, &lazyLruItem[V]{value: value, expires: time.Now().Add(x.successTTL).UnixMilli()})
+		x.core.Add(key, &layLruItem[V]{value: value, expires: time.Now().Add(x.successTTL).UnixMilli()})
 		return true
 	}
 	return false
 }
 
-func (x *LazyLRU[K, V]) Del(key K) bool {
+func (x *LayLRU[K, V]) Del(key K) bool {
 	x.lock.Lock()
 	ok := x.core.Remove(key)
 	x.lock.Unlock()
@@ -185,6 +179,6 @@ func (x *LazyLRU[K, V]) Del(key K) bool {
 	return ok
 }
 
-func (x *LazyLRU[K, V]) Stop() {
+func (x *LayLRU[K, V]) Stop() {
 
 }

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_" + strings.Join(ids, "_") // server notification chat
+		return "sn_" + ids[0] // server notification chat
 	}
 	return ""
 }

pkg/rpccache/online.go

@@ -3,16 +3,15 @@ package rpccache
 import (
 	"context"
 	"fmt"
+	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
+	"github.com/openimsdk/protocol/constant"
+	"github.com/openimsdk/protocol/user"
 	"math/rand"
 	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
 
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
-	"github.com/openimsdk/protocol/constant"
-	"github.com/openimsdk/protocol/user"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
 	"github.com/openimsdk/open-im-server/v3/pkg/localcache"
 	"github.com/openimsdk/open-im-server/v3/pkg/localcache/lru"
@@ -47,7 +46,7 @@ func NewOnlineCache(client *rpcli.UserClient, group *GroupLocalCache, rdb redis.
 	case false:
 		log.ZDebug(ctx, "fullUserCache is false")
 		x.lruCache = lru.NewSlotLRU(1024, localcache.LRUStringHash, func() lru.LRU[string, []int32] {
-			return lru.NewLazyLRU[string, []int32](2048, cachekey.OnlineExpire/2, time.Second*3, localcache.EmptyTarget{}, func(key string, value []int32) {})
+			return lru.NewLayLRU[string, []int32](2048, cachekey.OnlineExpire/2, time.Second*3, localcache.EmptyTarget{}, func(key string, value []int32) {})
 		})
 		x.CurrentPhase.Store(DoSubscribeOver)
 		x.Cond.Broadcast()

pkg/rpcli/msg.go

@@ -2,11 +2,9 @@ package rpcli
 
 import (
 	"context"
-
-	"google.golang.org/grpc"
-
 	"github.com/openimsdk/protocol/msg"
 	"github.com/openimsdk/protocol/sdkws"
+	"google.golang.org/grpc"
 )
 
 func NewMsgClient(cc grpc.ClientConnInterface) *MsgClient {

version/version

@@ -1 +1 @@
-v3.8.3-patch.13
+v3.8.3-patch.6