merge

# Conflicts:
#	go.mod
#	go.sum
#	internal/push/push_to_client.go

.devcontainer/devcontainer.json

@@ -2,7 +2,6 @@
     // Reference Doc: https://code.visualstudio.com/remote/advancedcontainers/overview
     "name": "OpenIM Dev Environment",
     // Update the container version when you publish dev-container
-    "dockerComposeFile": "docker-compose.yml",
     "build": { "dockerfile": "Dockerfile" },
     // Replace with uncommented line below to build your own local copy of the image
     // "dockerFile": "../docker/Dockerfile-dev",

.github/workflows/help-comment-issue.yml

@@ -17,6 +17,7 @@ on:
   issues:
     types:
       - labeled
+
 jobs:
   add-comment:
     if: github.event.label.name == 'help wanted' || github.event.label.name == 'good first issue'

.github/workflows/openimci.yml

@@ -23,6 +23,8 @@ on:
         - "docs/**"
         - "README.md"
         - "README_zh-CN.md"
+        - "**.md"
+        - "docs/**"
         - "CONTRIBUTING.md"
   pull_request:
     branches:
@@ -31,7 +33,8 @@ on:
     paths-ignore:
         - "README.md"
         - "README_zh-CN.md"
-        - "CONTRIBUTING.md"
+        - "CONTRIBUTING/**"
+        - "**.md"
         - "docs/**"
 
 env:
@@ -67,6 +70,9 @@ jobs:
         version: '3.x' # If available, use the latest major version that's compatible
         repo-token: ${{ secrets.GITHUB_TOKEN }}
 
+    - name: OpenIM Scripts Verification(make verify)
+      run: sudo make verify
+
     - name: Module Operations
       run: |
         sudo make tidy
@@ -91,13 +97,6 @@ jobs:
     - name: Cleanup Build
       run: sudo make clean
 
-    - name: Push Changes to Main
-      uses: stefanzweifel/git-auto-commit-action@v5
-      with:
-        commit_message: "cicd: robot automated Change"
-        branch: main
-      continue-on-error: true
-
     - name: Set Current Directory
       id: set_directory
       run: echo "::set-output name=directory::$(pwd)"

.github/workflows/pull-request.yml

@@ -78,9 +78,6 @@ jobs:
           echo "Run unit test and get test coverage successfully" 
         continue-on-error: true
 
-      - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@v2
-
       - name: OpenIM verify copyright
         run: |
           sudo make verify-copyright

.github/workflows/sync-release.yml

@@ -3,7 +3,7 @@
 # you may not use this file except in compliance with the License.
 
 # https://github.com/BetaHuhn/repo-file-sync-action
-name: Synchronize kubecub public code to other repositories
+name: Synchronize OpenIM Release Branch Public Code To Other Repositories
 on:
     push:
         paths:
@@ -41,3 +41,4 @@ jobs:
             automerge
           ASSIGNEES: |
             kubbot
+        continue-on-error: true

.github/workflows/sync.yml

@@ -3,7 +3,7 @@
 # you may not use this file except in compliance with the License.
 
 # https://github.com/BetaHuhn/repo-file-sync-action
-name: Synchronize kubecub public code to other repositories
+name: Synchronize OpenIM Main Branch Public Code To Other Repositories
 on:
     push:
         branches:
@@ -36,4 +36,5 @@ jobs:
             file-sync
             automerge
           ASSIGNEES: |
-            kubbot
+            kubbot
+        continue-on-error: true

.gitignore

@@ -34,6 +34,7 @@ deployments/charts/generated-configs/
 ### OpenIM Config ###
 .env
 config/config.yaml
+config/openim.yaml
 config/alertmanager.yml
 config/prometheus.yml
 config/email.tmpl

CHANGELOG/CHANGELOG-3.5.0.md

@@ -1,15 +0,0 @@
-# Version logging for OpenIM
-
-<!-- BEGIN MUNGE: GENERATED_TOC -->
-
-<!-- END MUNGE: GENERATED_TOC -->
-
-<a name="unreleased"></a>
-## [Unreleased]
-
-
-<a name="v3.5.0+3.97baaac"></a>
-## [v3.5.0+3.97baaac] - 2024-01-12
-
-[Unreleased]: https://github.com/openimsdk/open-im-server/compare/v3.5.0+3.97baaac...HEAD
-[v3.5.0+3.97baaac]: https://github.com/openimsdk/open-im-server/compare/v3.5.0+5.950e970...v3.5.0+3.97baaac

README.md

@@ -42,7 +42,7 @@ OpenIM is a service platform specifically designed for integrating chat, audio-v
   - 🛡️ API wrapping
   - 🌐 Connection management
 
-  ## 📚 Main Modules:
++ 📚 Main Modules:
 
   1. 🚀 Initialization and Login
   2. 👤 User Management
@@ -70,11 +70,16 @@ It is built using Golang and supports cross-platform deployment, ensuring a cons
 
 ## :rocket: Quick Start
 
+We support many platforms. Here are the addresses for quick experience on the web side：
+
+👉 **[OpenIM online web demo](https://web-enterprise.rentsoft.cn/)**
+
 🤲 To facilitate user experience, we offer various deployment solutions. You can choose your deployment method from the list below:
 
 + **[Source Code Deployment Guide](https://docs.openim.io/guides/gettingStarted/imSourceCodeDeployment)**
 + **[Docker Deployment Guide](https://docs.openim.io/guides/gettingStarted/dockerCompose)**
 + **[Kubernetes Deployment Guide](https://docs.openim.io/guides/gettingStarted/k8s-deployment)**
++ **[Mac Developer Deployment Guide](https://docs.openim.io/guides/gettingstarted/mac-deployment-guide)**
 
 ## :hammer_and_wrench: To Start Developing OpenIM
 

config/templates/config.yaml.template

@@ -14,7 +14,7 @@
 
 # -----------------------------------------------------------------
 # TODO: This config file is the template file
-# --| source: deployments/templates/openim.yaml
+# --| source: deployments/templates/config.yaml
 # --| env: scripts/install/environment
 # --| target: config/config.yaml
 # -----------------------------------------------------------------
@@ -53,7 +53,7 @@ mongo:
 # Maximum connection pool size
   address: [ 172.28.0.1:37017 ]
   database: openim_v3
-  username: root
+  username: openIM
   password: openIM123
   maxPoolSize: 100
 
@@ -122,14 +122,14 @@ api:
 # minio.signEndpoint is minio public network address
 object:
   enable: "minio"
-  apiURL: "http://14.155.64.202:10002"
+  apiURL: "http://172.28.0.1:10002"
   minio:
     bucket: "openim"
     endpoint: "http://172.28.0.1:10005"
     accessKeyID: "root"
     secretAccessKey: "openIM123"
     sessionToken: ''
-    signEndpoint: "http://14.155.64.202:10005"
+    signEndpoint: "http://172.28.0.1:10005"
     publicRead: false
   cos:
     bucketURL: https://temp-1252357374.cos.ap-chengdu.myqcloud.com
@@ -193,7 +193,7 @@ rpcRegisterName:
 # Whether to output in json format
 # Whether to include stack trace in logs
 log:
-  storageLocation: /data/workspaces/open-im-server/logs/
+  storageLocation: /workspaces/open-im-server/logs/
   rotationTime: 24
   remainRotationCount: 2
   remainLogLevel: 6
@@ -247,6 +247,14 @@ manager:
   userID: [ "openIM123456", "openIM654321", "openIMAdmin" ]
   nickname: [ "system1", "system2", "system3" ]
 
+# chatAdmin, use for send notification
+#
+# Built-in app system notification account ID
+# Built-in app system notification account nickname
+im-admin:
+  userID: [ "imAdmin" ]
+  nickname: [ "imAdmin" ]
+
 # Multi-platform login policy
 # For each platform(Android, iOS, Windows, Mac, web), only one can be online at a time
 multiLoginPolicy: 1
@@ -307,21 +315,21 @@ iosPush:
 # Timeout in seconds
 # Whether to continue execution if callback fails
 callback:
-  url: ""
+  url: "http://127.0.0.1:10008/callbackExample"
   beforeSendSingleMsg:
     enable: false
     timeout: 5
     failedContinue: true
   beforeUpdateUserInfoEx:
-    enable: false
+    enable:  false
     timeout: 5
     failedContinue: true
   afterUpdateUserInfoEx:
-    enable: false
+    enable:  false
     timeout: 5
     failedContinue: true
   afterSendSingleMsg:
-    enable: false
+    enable: true
     timeout: 5
     failedContinue: true
   beforeSendGroupMsg:
@@ -505,8 +513,8 @@ callback:
 # The number of Prometheus ports per service needs to correspond to rpcPort
 # The number of ports needs to be consistent with msg_transfer_service_num in script/path_info.sh
 prometheus:
-  enable: false
+  enable: true
-  grafanaUrl: 172.28.0.1:13000
+  grafanaUrl: http://172.28.0.1:13000/
   apiPrometheusPort: [20100]
   userPrometheusPort: [ 20110 ]
   friendPrometheusPort: [ 20120 ]

config/templates/env.template

@@ -1,4 +1,4 @@
-# Copyright © 2023 OpenIM. All rights reserved.
+# Copyright © 2024 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.
@@ -12,31 +12,26 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-# ======================================
+# -----------------------------------------------------------------------------
-# ========= Basic Configuration ========
+# General Configuration
-# ======================================
+# This section contains general configuration options for the entire environment.
+# These options can be set via environment variables. If both environment variables 
+# and settings in this .env file exist, the environment variables take precedence.
+# -----------------------------------------------------------------------------
+# ==========================
+# General Configuration
+# ==========================
+# These settings apply to the overall environment.
 
-# The user for authentication or system operations.
+# Data storage directory for persistent data.
-# Default: OPENIM_USER=root
+# Example: DATA_DIR=/path/to/data
-USER=root
+DATA_DIR=/workspaces/open-im-server
 
-# Password associated with the specified user for authentication.
+# Docker image registry. Uncomment the preferred one.
-# Default: PASSWORD=openIM123
+# Options: ghcr.io/openimsdk, openim, registry.cn-hangzhou.aliyuncs.com/openimsdk
-PASSWORD=openIM123
+# IMAGE_REGISTRY="ghcr.io/openimsdk"
-
+# IMAGE_REGISTRY="openim"
-# Base URL for the application programming interface (API).
+# IMAGE_REGISTRY="registry.cn-hangzhou.aliyuncs.com/openimsdk"
-# Default: API_URL=http://172.28.0.1:10002
-API_URL=http://14.155.64.202:10002
-
-# Directory path for storing data files or related information.
-# Default: DATA_DIR=./
-DATA_DIR=/data/workspaces/open-im-server
-
-# Choose the appropriate image address, the default is GITHUB image, 
-# you can choose docker hub, for Chinese users can choose Ali Cloud
-# export IMAGE_REGISTRY="ghcr.io/openimsdk"
-# export IMAGE_REGISTRY="openim"
-# export IMAGE_REGISTRY="registry.cn-hangzhou.aliyuncs.com/openimsdk"
 IMAGE_REGISTRY=ghcr.io/openimsdk
 
 # ======================================
@@ -47,10 +42,9 @@ IMAGE_REGISTRY=ghcr.io/openimsdk
 # Default: DOCKER_BRIDGE_SUBNET=172.28.0.0/16
 DOCKER_BRIDGE_SUBNET=172.28.0.0/16
 
-# Gateway for the Docker network.
+# Set and specify the IP addresses of some containers. Generally speaking, 
-# Default: DOCKER_BRIDGE_GATEWAY=172.28.0.1
+# you do not need to modify these configurations to facilitate debugging
 DOCKER_BRIDGE_GATEWAY=172.28.0.1
-
 MONGO_NETWORK_ADDRESS=172.28.0.2
 REDIS_NETWORK_ADDRESS=172.28.0.3
 KAFKA_NETWORK_ADDRESS=172.28.0.4
@@ -65,45 +59,66 @@ NODE_EXPORTER_NETWORK_ADDRESS=172.28.0.12
 OPENIM_ADMIN_FRONT_NETWORK_ADDRESS=172.28.0.13
 ALERT_MANAGER_NETWORK_ADDRESS=172.28.0.14
 
-# ===============================================
+# ==============================================================================
-# = Component Extension Configuration =
+# Configuration Update Instructions
-# ===============================================
+# ==============================================================================
+# This header outlines the methods to update common variables in config.yaml and .env files.
+# These instructions are vital for maintaining the OpenIM environment's configuration.
+#
+# METHOD 1: Regenerate All Configurations
+# ----------------------------------------
+# Use this method to regenerate all configurations.
+# Steps:
+#   1. Delete existing config files:
+#      - openim-server/config/config.yaml
+#      - openim-chat/config/config.yaml
+#   2. Modify the .env file as required.
+#   3. Run 'docker compose up -d'. This will regenerate:
+#       - config/config.yaml
+#
+# METHOD 2: Modify Individual Configuration Files
+# -----------------------------------------------
+# Use this method to update specific configuration files.
+# Steps:
+#   1. Modify the .env file as necessary.
+#   2. Update the corresponding entries in:
+#      - config/config.yaml
+#   3. Restart the services with 'docker compose up -d'.
+#   4. Special Note: If you modify OPENIM_IP, API_OPENIM_PORT, or MINIO_PORT in .env,
+#      ensure to update the corresponding services and configurations accordingly.
+#
+# It is essential to follow these methods to ensure consistent and correct application behavior.
+# ==============================================================================
+# Local IP address of the service. Modify if necessary.
+# Example: OPENIM_IP=172.28.0.1，
+OPENIM_IP=172.28.0.1
 
-# ============ Component Extension Configuration ==========
 # ----- ZooKeeper Configuration -----
-# Address or hostname for the ZooKeeper service.
-# Default: ZOOKEEPER_ADDRESS=172.28.0.1
-ZOOKEEPER_ADDRESS=172.28.0.5
-
 # Port for ZooKeeper service.
 # Default: ZOOKEEPER_PORT=12181
 ZOOKEEPER_PORT=12181
 
-# ----- MongoDB Configuration -----
+# MongoDB service port configuration.
-# Address or hostname for the MongoDB service.
-# Default: MONGO_ADDRESS=172.28.0.1
-MONGO_ADDRESS=172.28.0.2
-
-# Port on which MongoDB service is running.
 # Default: MONGO_PORT=37017
 # MONGO_PORT=37017
 
-# Username to authenticate with the MongoDB service.
+# Password for MongoDB admin user. Used for service authentication.
-# Default: MONGO_USERNAME=root
-# MONGO_USERNAME=root
-
-# Password to authenticate with the MongoDB service.
 # Default: MONGO_PASSWORD=openIM123
 MONGO_PASSWORD=openIM123
 
-# Name of the database in MongoDB to be used.
+# Username for a regular OpenIM user in MongoDB.
+# Default: MONGO_OPENIM_USERNAME=openIM
+MONGO_OPENIM_USERNAME=openIM
+
+# Password for a regular OpenIM user in MongoDB.
+# Default: MONGO_OPENIM_PASSWORD=openIM123456
+MONGO_OPENIM_PASSWORD=openIM123
+
+# Specifies the database name to be used within MongoDB.
 # Default: MONGO_DATABASE=openim_v3
 MONGO_DATABASE=openim_v3
 
 # ----- Redis Configuration -----
-# Address or hostname for the Redis service.
-# Default: REDIS_ADDRESS=172.28.0.1
-REDIS_ADDRESS=172.28.0.3
 
 # Port on which Redis in-memory data structure store is running.
 # Default: REDIS_PORT=16379
@@ -113,11 +128,6 @@ REDIS_PORT=16379
 # Default: REDIS_PASSWORD=openIM123
 REDIS_PASSWORD=openIM123
 
-# ----- Kafka Configuration -----
-# Address or hostname for the Kafka service.
-# Default: KAFKA_ADDRESS=172.28.0.1
-KAFKA_ADDRESS=172.28.0.4
-
 # Kakfa username to authenticate with the Kafka service.
 # KAFKA_USERNAME=''
 
@@ -129,20 +139,13 @@ KAFKA_PORT=19094
 # Default: KAFKA_LATESTMSG_REDIS_TOPIC=latestMsgToRedis
 KAFKA_LATESTMSG_REDIS_TOPIC=latestMsgToRedis
 
-# Topic in Kafka for pushing messages (e.g. notifications or updates).
+# MINIO_PORT
-# Default: KAFKA_MSG_PUSH_TOPIC=msgToPush
+# ----------
-KAFKA_MSG_PUSH_TOPIC=msgToPush
+# MINIO_PORT sets the port for the MinIO object storage service.
-
+# Upon changing this port, the MinIO endpoint URLs in the  file must be updated
-# Topic in Kafka for storing offline messages in MongoDB.
+# to reflect this change. The endpoints include both the 'endpoint' and 'signEndpoint' 
-# Default: KAFKA_OFFLINEMSG_MONGO_TOPIC=offlineMsgToMongoMysql
+# under the MinIO configuration.
-KAFKA_OFFLINEMSG_MONGO_TOPIC=offlineMsgToMongoMysql
+#
-
-# ----- MinIO Configuration ----
-# Address or hostname for the MinIO object storage service.
-# Default: MINIO_ADDRESS=172.28.0.1
-MINIO_ADDRESS=172.28.0.6
-
-# Port on which MinIO object storage service is running.
 # Default: MINIO_PORT=10005
 MINIO_PORT=10005
 
@@ -155,19 +158,11 @@ MINIO_PORT=10005
 MINIO_SECRET_KEY=openIM123
 
 # ----- Prometheus Configuration -----
-# Address or hostname for the Prometheus service.
-# Default: PROMETHEUS_ADDRESS=172.28.0.1
-PROMETHEUS_ADDRESS=172.28.0.10
-
 # Port on which Prometheus service is running.
 # Default: PROMETHEUS_PORT=19090
 PROMETHEUS_PORT=19090
 
 # ----- Grafana Configuration -----
-# Address or hostname for the Grafana service.
-# Default: GRAFANA_ADDRESS=172.28.0.1
-GRAFANA_ADDRESS=172.28.0.11
-
 # Port on which Grafana service is running.
 # Default: GRAFANA_PORT=13000
 GRAFANA_PORT=13000
@@ -183,41 +178,34 @@ OPENIM_WEB_PORT=11001
 # ======================================
 # ========= OpenIM Server ==============
 # ======================================
-
-# Address or hostname for the OpenIM server.
-# Default: OPENIM_SERVER_ADDRESS=172.28.0.1
-OPENIM_SERVER_ADDRESS=172.28.0.8
-
 # Port for the OpenIM WebSockets.
 # Default: OPENIM_WS_PORT=10001
 OPENIM_WS_PORT=10001
 
-# Port for the OpenIM API.
+# API_OPENIM_PORT
+# ---------------
+# This variable defines the port on which the OpenIM API service will listen.
+# When changing this port, it's essential to update the apiURL in the config.yaml file
+# to ensure the API service is accessible at the new port.
+#
 # Default: API_OPENIM_PORT=10002
 API_OPENIM_PORT=10002
 
-
 # ======================================
 # ========== OpenIM Chat ===============
 # ======================================
 
 # Branch name for OpenIM chat.
 # Default: CHAT_IMAGE_VERSION=main
-# https://github.com/openimsdk/open-im-server/blob/main/docs/contrib/version.md
 CHAT_IMAGE_VERSION=main
 
-# Address or hostname for the OpenIM chat service.
-# Default: OPENIM_CHAT_ADDRESS=172.28.0.1
-OPENIM_CHAT_ADDRESS=172.28.0.9
-
 # Port for the OpenIM chat API.
 # Default: OPENIM_CHAT_API_PORT=10008
 OPENIM_CHAT_API_PORT=10008
 
-# Directory path for storing data files or related information for OpenIM chat.
+# Port for the OpenIM admin API.
-# Default: OPENIM_CHAT_DATA_DIR=./openim-chat/main
+# Default: OPENIM_ADMIN_API_PORT=10009
-OPENIM_CHAT_DATA_DIR=./openim-chat/main
+OPENIM_ADMIN_API_PORT=10009
-
 
 # ======================================
 # ========== OpenIM Admin ==============
@@ -226,10 +214,6 @@ OPENIM_CHAT_DATA_DIR=./openim-chat/main
 # Branch name for OpenIM server.
 # Default: SERVER_IMAGE_VERSION=main
 SERVER_IMAGE_VERSION=main
-
-# Port for the OpenIM admin API.
-# Default: OPENIM_ADMIN_API_PORT=10009
-OPENIM_ADMIN_API_PORT=10009
   
 # Port for the node exporter.
 # Default: NODE_EXPORTER_PORT=19100

config/templates/open-im-ng-example.conf

@@ -169,4 +169,4 @@ server {
 
         proxy_pass http://minio_console_2; # This uses the upstream directive definition to load balance
     }
-}
+}

config/templates/prometheus-dashboard.yaml

@@ -1471,4 +1471,4 @@
   "uid": "f5f5de9a-6ec5-499a-841e-6e901c33b1f7",
   "version": 16,
   "weekStart": ""
-}
+}

deployments/Readme.md

@@ -165,7 +165,7 @@ export MINIO_ENDPOINT="http://im-minio:9000"
 export MINIO_SIGN_ENDPOINT="https://openim.server.com/im-minio-api"
 
 mkdir ./charts/generated-configs
-../scripts/genconfig.sh ../scripts/install/environment.sh ./templates/openim.yaml > ./charts/generated-configs/config.yaml
+../scripts/genconfig.sh ../scripts/install/environment.sh ./templates/config.yaml > ./charts/generated-configs/config.yaml
 cp ../config/notification.yaml ./charts/generated-configs/notification.yaml
 ../scripts/genconfig.sh ../scripts/install/environment.sh ./templates/helm-image.yaml > ./charts/generated-configs/helm-image.yaml
 ```

deployments/templates/config.yaml

@@ -14,7 +14,7 @@
 
 # -----------------------------------------------------------------
 # TODO: This config file is the template file
-# --| source: deployments/templates/openim.yaml
+# --| source: deployments/templates/config.yaml
 # --| env: scripts/install/environment
 # --| target: config/config.yaml
 # -----------------------------------------------------------------
@@ -527,3 +527,39 @@ prometheus:
   rtcPrometheusPort: [ ${RTC_PROM_PORT} ]
   thirdPrometheusPort: [ ${THIRD_PROM_PORT} ]
   messageTransferPrometheusPort: [ ${MSG_TRANSFER_PROM_PORT} ] # List of ports
+
+###################### LocalCache configuration information ######################
+# topic: redis subscriber channel
+# slotNum: number of slots, multiple slots can prevent too many keys from competing for a lock
+# slotSize: number of slots, the number of cached keys per slot, the overall cache quantity is slotNum * slotSize
+# successExpire: successful cache time seconds
+# failedExpire: failed cache time seconds
+# disable local caching and annotate topic, slotNum, and slotSize
+localCache:
+  user:
+    topic: DELETE_CACHE_USER
+    slotNum: 100
+    slotSize: 2000
+    successExpire: 300
+    failedExpire: 5
+
+  group:
+    topic: DELETE_CACHE_GROUP
+    slotNum: 100
+    slotSize: 2000
+    successExpire: 300
+    failedExpire: 5
+
+  friend:
+    topic: DELETE_CACHE_FRIEND
+    slotNum: 100
+    slotSize: 2000
+    successExpire: 300
+    failedExpire: 5
+
+  conversation:
+    topic: DELETE_CACHE_CONVERSATION
+    slotNum: 100
+    slotSize: 2000
+    successExpire: 300
+    failedExpire: 5

docker-compose.yml

@@ -214,6 +214,7 @@ services:
   #     - "${OPENIM_ADMIN_API_PORT:-10009}:10009"
   #   volumes:
   #     - "${DATA_DIR:-./}/components/openim-chat/logs:/openim/openim-chat/logs"
+  #     - "${DATA_DIR:-./}/components/openim-chat/_output/logs:/openim/openim-chat/_output/logs"
   #     - "${DATA_DIR:-./}/components/openim-chat/config:/openim/openim-chat/config"
   #   restart: always
   #   # user: root:root

docs/.generated_docs

@@ -31,7 +31,7 @@ docs/guide/en-US/cmd/openim/openim-rpc-user_list.md
 docs/guide/en-US/cmd/openim/openim-rpc-user_update.md
 docs/guide/en-US/cmd/openim/openim_validate.md
 docs/guide/en-US/cmd/openim/openim_version.md
-docs/guide/en-US/yaml/openim/openim.yaml
+docs/guide/en-US/yaml/openim/config.yaml
 docs/guide/en-US/yaml/openim/openim_color.yaml
 docs/guide/en-US/yaml/openim/openim_completion.yaml
 docs/guide/en-US/yaml/openim/openim_info.yaml

docs/contrib/environment.md

@@ -89,7 +89,7 @@ While the first two methods will be our main focus, it's worth noting that the t
 
 ###  1.2. <a name='SourceCodeDeployment'></a>Source Code Deployment
 
-In the source code deployment method, the configuration generation process involves executing `make init`, which fundamentally runs the script `./scripts/init-config.sh`. This script utilizes variables defined in the [`environment.sh`](https://github.com/openimsdk/open-im-server/blob/main/scripts/install/environment.sh) script to render the [`openim.yaml`](https://github.com/openimsdk/open-im-server/blob/main/deployments/templates/openim.yaml) template file, subsequently generating the [`config.yaml`](https://github.com/openimsdk/open-im-server/blob/main/config/config.yaml) configuration file.
+In the source code deployment method, the configuration generation process involves executing `make init`, which fundamentally runs the script `./scripts/init-config.sh`. This script utilizes variables defined in the [`environment.sh`](https://github.com/openimsdk/open-im-server/blob/main/scripts/install/environment.sh) script to render the [`config.yaml`](https://github.com/openimsdk/open-im-server/blob/main/deployments/templates/config.yaml) template file, subsequently generating the [`config.yaml`](https://github.com/openimsdk/open-im-server/blob/main/config/config.yaml) configuration file.
 
 ###  1.3. <a name='DockerComposeDeployment'></a>Docker Compose Deployment
 

docs/contrib/mac-developer-deployment-guide.md

@@ -45,63 +45,6 @@ Homebrew is an essential package manager for macOS. Install it using:
    git config --global user.email "your.email@example.com"
    ```
 
-<<<<<<< HEAD
-=======
-### Forking and Cloning the Repository
-
-To optimize and add logic to your instructions, particularly regarding deploying on a Mac, you can modify them as follows:
-
-1. **Fork the OpenIM Repository**: Fork the OpenIM repository on GitHub to your account.
-
-2. **Clone Your Fork to Your Local Machine**:
-   Open your terminal and execute the following commands:
-
-   ```sh
-   # Clone the repository
-   git clone https://github.com/<your-username>/open-im-server.git
-
-   # Set Docker bridging network mode for Mac
-   export DOCKER_BRIDGE_SUBNET=127.0.0.0/16
-
-   # Set OpenIM IP
-   export OPENIM_IP=<your-ip>
-
-   # Initialize configuration
-   make init
-
-   # Start components using Docker
-   docker compose up -d
-
-   # Start OpenIM Server
-   make start
-   ```
-
-3. **Additional Steps for Mac Deployment**:
-   If you are deploying on a Mac and wish to use the chat feature, you need to modify the `docker-compose` file. Specifically, you'll need to uncomment the sections related to `openim-chat` and `mysql`. 
-
-   Here's how to do it:
-
-   - Open the `docker-compose.yml` file in a text editor.
-   - Find the sections for `openim-chat` and `mysql`.
-   - Remove the comment marks (`#`) at the beginning of the lines in these sections to enable them.
-   - Save the file after making these changes.
-
-4. **Update and Restart Services**:
-   After modifying the `docker-compose` file, you need to update and restart the services to apply these changes. Run the following command in your terminal:
-
-   ```sh
-   # Update and restart services
-   docker compose up -d
-
-   # Check openim-chat start
-   docker compose logs openim-chat
-   ```
-
-   This command will re-read the `docker-compose.yml` file, apply the new configuration, and restart the necessary containers.
-
-Remember, replacing `<your-username>` and `<your-ip>` with your actual GitHub username and desired IP address for OpenIM is crucial. These steps should streamline the setup process, particularly for Mac users wishing to use the chat feature.
-
->>>>>>> origin/main
 ### Setting Up the Devcontainer
 
 `Devcontainers` provide a Docker-based isolated development environment. 
@@ -242,12 +185,9 @@ make start
 To check, ensure the following four processes start successfully:
 
 ```bash
-➜  chat git:(main) ✗ ps -ef | grep chat      
+make check 
-# [Output showing four running chat processes]
 ```
 
-Access the web interface: http://ip:11001
-
 ### Docker Deployment
 
 Refer to https://github.com/openimsdk/openim-docker for Docker deployment instructions, which can be followed similarly on Linux.

go.mod

@@ -4,8 +4,8 @@ go 1.19
 
 require (
 	firebase.google.com/go v3.13.0+incompatible
-	github.com/OpenIMSDK/protocol v0.0.48
+	github.com/OpenIMSDK/protocol v0.0.49
-	github.com/OpenIMSDK/tools v0.0.23
+	github.com/OpenIMSDK/tools v0.0.28
 	github.com/bwmarrin/snowflake v0.3.0 // indirect
 	github.com/dtm-labs/rockscache v0.1.1
 	github.com/gin-gonic/gin v1.9.1
@@ -18,6 +18,7 @@ require (
 	github.com/minio/minio-go/v7 v7.0.63
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646 // indirect
+	github.com/openimsdk/localcache v0.0.1
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.17.0
 	github.com/robfig/cron/v3 v3.0.1
@@ -38,6 +39,7 @@ require (
 	github.com/aliyun/aliyun-oss-go-sdk v2.2.9+incompatible
 	github.com/go-redis/redis v6.15.9+incompatible
 	github.com/redis/go-redis/v9 v9.2.1
+	github.com/spf13/pflag v1.0.5
 	github.com/stathat/consistent v1.0.0
 	github.com/tencentyun/cos-go-sdk-v5 v0.7.45
 	go.uber.org/automaxprocs v1.5.3
@@ -81,6 +83,7 @@ require (
 	github.com/hashicorp/errwrap v1.0.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/go-uuid v1.0.3 // indirect
+	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 // indirect
 	github.com/jcmturner/aescts/v2 v2.0.0 // indirect
@@ -116,7 +119,6 @@ require (
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
 	github.com/rs/xid v1.5.0 // indirect
 	github.com/sergi/go-diff v1.0.0 // indirect
-	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/src-d/gcfg v1.4.0 // indirect
 	github.com/stretchr/objx v0.5.0 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
@@ -157,4 +159,4 @@ require (
 	gopkg.in/ini.v1 v1.67.0 // indirect
 )
 
-replace github.com/OpenIMSDK/protocol v0.0.47 => github.com/AndrewZuo01/protocol v0.0.0-20240112093520-fd9c53e27b94
+replace github.com/openimsdk/localcache => ./pkg/localcache

go.sum

@@ -15,13 +15,13 @@ cloud.google.com/go/storage v1.30.1 h1:uOdMxAs8HExqBlnLtnQyP0YkvbiDpdGShGKtx6U/o
 cloud.google.com/go/storage v1.30.1/go.mod h1:NfxhC0UJE1aXSx7CIIbCf7y9HKT7BiccwkR7+P7gN8E=
 firebase.google.com/go v3.13.0+incompatible h1:3TdYC3DDi6aHn20qoRkxwGqNgdjtblwVAyRLQwGn/+4=
 firebase.google.com/go v3.13.0+incompatible/go.mod h1:xlah6XbEyW6tbfSklcfe5FHJIwjt8toICdV5Wh9ptHs=
-github.com/AndrewZuo01/protocol v0.0.0-20240112093520-fd9c53e27b94 h1:o86vkek41ZrQqoBGqyKvS0z6N0uJj64mpzK72OkDZVM=
-github.com/AndrewZuo01/protocol v0.0.0-20240112093520-fd9c53e27b94/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/IBM/sarama v1.41.3 h1:MWBEJ12vHC8coMjdEXFq/6ftO6DUZnQlFYcxtOJFa7c=
 github.com/IBM/sarama v1.41.3/go.mod h1:Xxho9HkHd4K/MDUo/T/sOqwtX/17D33++E9Wib6hUdQ=
-github.com/OpenIMSDK/tools v0.0.23 h1:xozfrGzhbpNPlDTap5DLVPk+JfgZ/ZyIj4Cuu3/bm9w=
+github.com/OpenIMSDK/protocol v0.0.49 h1:wcqJOMBis7f153zNI7V82Fc4WyqA1GanMgXUQgL618k=
-github.com/OpenIMSDK/tools v0.0.23/go.mod h1:eg+q4A34Qmu73xkY0mt37FHGMCMfC6CtmOnm0kFEGFI=
+github.com/OpenIMSDK/protocol v0.0.49/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y=
+github.com/OpenIMSDK/tools v0.0.27 h1:J/kSRqM+y9U4XK/pQ9RkEB31oQ5BTYD1oA5r1PITPRA=
+github.com/OpenIMSDK/tools v0.0.27/go.mod h1:eg+q4A34Qmu73xkY0mt37FHGMCMfC6CtmOnm0kFEGFI=
 github.com/QcloudApi/qcloud_sign_golang v0.0.0-20141224014652-e4130a326409/go.mod h1:1pk82RBxDY/JZnPQrtqHlUFfCctgdorsd9M06fMynOM=
 github.com/alcortesm/tgz v0.0.0-20161220082320-9c5fe88206d7 h1:uSoVVbwJiQipAclBbw+8quDsfcvFjOpI5iCf4p/cqCs=
 github.com/alcortesm/tgz v0.0.0-20161220082320-9c5fe88206d7/go.mod h1:6zEj6s6u/ghQa61ZWa/C2Aw3RkjiTBOix7dkqa1VLIs=
@@ -171,6 +171,8 @@ github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9
 github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
 github.com/hashicorp/go-uuid v1.0.3 h1:2gKiV6YVmrJ1i2CKKa9obLvRieoRGviZFL26PcT/Co8=
 github.com/hashicorp/go-uuid v1.0.3/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
+github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
+github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=

internal/api/route.go

@@ -16,6 +16,7 @@ package api
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 
 	"github.com/OpenIMSDK/protocol/constant"
@@ -43,7 +44,7 @@ import (
 )
 
 func NewGinRouter(discov discoveryregistry.SvcDiscoveryRegistry, rdb redis.UniversalClient) *gin.Engine {
-	discov.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials())) // 默认RPC中间件
+	discov.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin"))) // 默认RPC中间件
 	gin.SetMode(gin.ReleaseMode)
 	r := gin.New()
 	if v, ok := binding.Validator.Engine().(*validator.Validate); ok {

internal/api/user.go

@@ -201,27 +201,27 @@ func (u *UserApi) GetSubscribeUsersStatus(c *gin.Context) {
 	a2r.Call(user.UserClient.GetSubscribeUsersStatus, u.Client, c)
 }
 
-// ProcessUserCommandAdd user general function add
+// ProcessUserCommandAdd user general function add.
 func (u *UserApi) ProcessUserCommandAdd(c *gin.Context) {
 	a2r.Call(user.UserClient.ProcessUserCommandAdd, u.Client, c)
 }
 
-// ProcessUserCommandDelete user general function delete
+// ProcessUserCommandDelete user general function delete.
 func (u *UserApi) ProcessUserCommandDelete(c *gin.Context) {
 	a2r.Call(user.UserClient.ProcessUserCommandDelete, u.Client, c)
 }
 
-// ProcessUserCommandUpdate  user general function update
+// ProcessUserCommandUpdate  user general function update.
 func (u *UserApi) ProcessUserCommandUpdate(c *gin.Context) {
 	a2r.Call(user.UserClient.ProcessUserCommandUpdate, u.Client, c)
 }
 
-// ProcessUserCommandGet user general function get
+// ProcessUserCommandGet user general function get.
 func (u *UserApi) ProcessUserCommandGet(c *gin.Context) {
 	a2r.Call(user.UserClient.ProcessUserCommandGet, u.Client, c)
 }
 
-// ProcessUserCommandGet user general function get all
+// ProcessUserCommandGet user general function get all.
 func (u *UserApi) ProcessUserCommandGetAll(c *gin.Context) {
 	a2r.Call(user.UserClient.ProcessUserCommandGetAll, u.Client, c)
 }

internal/msggateway/callback.go

@@ -105,7 +105,7 @@ func CallbackUserKickOff(ctx context.Context, userID string, platformID int) err
 // func callbackUserOnline(operationID, userID string, platformID int, token string, isAppBackground bool, connID
 // string) cbApi.CommonCallbackResp {
 //	callbackResp := cbApi.CommonCallbackResp{OperationID: operationID}
-//	if !config.Config.Callback.CallbackUserOnline.Enable {
+//	if !config.Config.Callback.CallbackUserOnline.WithEnable {
 //		return callbackResp
 //	}
 //	callbackUserOnlineReq := cbApi.CallbackUserOnlineReq{
@@ -134,7 +134,7 @@ func CallbackUserKickOff(ctx context.Context, userID string, platformID int) err
 //}
 //func callbackUserOffline(operationID, userID string, platformID int, connID string) cbApi.CommonCallbackResp {
 //	callbackResp := cbApi.CommonCallbackResp{OperationID: operationID}
-//	if !config.Config.Callback.CallbackUserOffline.Enable {
+//	if !config.Config.Callback.CallbackUserOffline.WithEnable {
 //		return callbackResp
 //	}
 //	callbackOfflineReq := cbApi.CallbackUserOfflineReq{
@@ -161,7 +161,7 @@ func CallbackUserKickOff(ctx context.Context, userID string, platformID int) err
 //}
 //func callbackUserKickOff(operationID string, userID string, platformID int) cbApi.CommonCallbackResp {
 //	callbackResp := cbApi.CommonCallbackResp{OperationID: operationID}
-//	if !config.Config.Callback.CallbackUserKickOff.Enable {
+//	if !config.Config.Callback.CallbackUserKickOff.WithEnable {
 //		return callbackResp
 //	}
 //	callbackUserKickOffReq := cbApi.CallbackUserKickOffReq{

internal/msggateway/hub_server.go

@@ -25,8 +25,6 @@ import (
 	"github.com/OpenIMSDK/tools/errs"
 	"github.com/OpenIMSDK/tools/log"
 	"github.com/OpenIMSDK/tools/mcontext"
-	"github.com/OpenIMSDK/tools/utils"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
@@ -59,7 +57,7 @@ type Server struct {
 	rpcPort        int
 	prometheusPort int
 	LongConnServer LongConnServer
-	pushTerminal   []int
+	pushTerminal   map[int]struct{}
 }
 
 func (s *Server) SetLongConnServer(LongConnServer LongConnServer) {
@@ -67,12 +65,15 @@ func (s *Server) SetLongConnServer(LongConnServer LongConnServer) {
 }
 
 func NewServer(rpcPort int, proPort int, longConnServer LongConnServer) *Server {
-	return &Server{
+	s := &Server{
 		rpcPort:        rpcPort,
 		prometheusPort: proPort,
 		LongConnServer: longConnServer,
-		pushTerminal:   []int{constant.IOSPlatformID, constant.AndroidPlatformID},
+		pushTerminal:   make(map[int]struct{}),
 	}
+	s.pushTerminal[constant.IOSPlatformID] = struct{}{}
+	s.pushTerminal[constant.AndroidPlatformID] = struct{}{}
+	return s
 }
 
 func (s *Server) OnlinePushMsg(
@@ -126,13 +127,9 @@ func (s *Server) OnlineBatchPushOneMsg(
 	panic("implement me")
 }
 
-func (s *Server) SuperGroupOnlineBatchPushOneMsg(
+func (s *Server) SuperGroupOnlineBatchPushOneMsg(ctx context.Context, req *msggateway.OnlineBatchPushOneMsgReq,
-	ctx context.Context,
-	req *msggateway.OnlineBatchPushOneMsgReq,
 ) (*msggateway.OnlineBatchPushOneMsgResp, error) {
-
 	var singleUserResults []*msggateway.SingleMsgToUserResults
-
 	for _, v := range req.PushToUserIDs {
 		var resp []*msggateway.SingleMsgToUserPlatform
 		results := &msggateway.SingleMsgToUserResults{
@@ -153,23 +150,22 @@ func (s *Server) SuperGroupOnlineBatchPushOneMsg(
 			}
 
 			userPlatform := &msggateway.SingleMsgToUserPlatform{
-				RecvID:         v,
+				PlatFormID: int32(client.PlatformID),
-				RecvPlatFormID: int32(client.PlatformID),
 			}
 			if !client.IsBackground ||
 				(client.IsBackground && client.PlatformID != constant.IOSPlatformID) {
 				err := client.PushMessage(ctx, req.MsgData)
 				if err != nil {
-					userPlatform.ResultCode = -2
+					userPlatform.ResultCode = int64(errs.ErrPushMsgErr.Code())
 					resp = append(resp, userPlatform)
 				} else {
-					if utils.IsContainInt(client.PlatformID, s.pushTerminal) {
+					if _, ok := s.pushTerminal[client.PlatformID]; ok {
 						results.OnlinePush = true
 						resp = append(resp, userPlatform)
 					}
 				}
 			} else {
-				userPlatform.ResultCode = -3
+				userPlatform.ResultCode = int64(errs.ErrIOSBackgroundPushErr.Code())
 				resp = append(resp, userPlatform)
 			}
 		}

internal/msggateway/n_ws_server.go

@@ -19,7 +19,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"github.com/OpenIMSDK/tools/apiresp"
 	"net/http"
 	"os"
 	"os/signal"
@@ -29,6 +28,8 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/OpenIMSDK/tools/apiresp"
+
 	"github.com/go-playground/validator/v10"
 	"github.com/redis/go-redis/v9"
 	"golang.org/x/sync/errgroup"

internal/msgtransfer/init.go

@@ -66,7 +66,7 @@ func StartTransfer(prometheusPort int) error {
 	if err := client.CreateRpcRootNodes(config.Config.GetServiceNames()); err != nil {
 		return err
 	}
-	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
 	msgModel := cache.NewMsgCacheModel(rdb)
 	msgDocModel := unrelation.NewMsgMongoDriver(mongo.GetDatabase())
 	msgDatabase := controller.NewCommonMsgDatabase(msgDocModel, msgModel)

internal/push/callback.go

@@ -16,6 +16,7 @@ package push
 
 import (
 	"context"
+	"encoding/json"
 
 	"github.com/OpenIMSDK/protocol/constant"
 	"github.com/OpenIMSDK/protocol/sdkws"
@@ -136,3 +137,14 @@ func callbackBeforeSuperGroupOnlinePush(
 	}
 	return nil
 }
+func GetContent(msg *sdkws.MsgData) string {
+	if msg.ContentType >= constant.NotificationBegin && msg.ContentType <= constant.NotificationEnd {
+		var notification sdkws.NotificationElem
+		if err := json.Unmarshal(msg.Content, &notification); err != nil {
+			return ""
+		}
+		return notification.Detail
+	} else {
+		return string(msg.Content)
+	}
+}

internal/push/consumer_init.go

@@ -1,32 +0,0 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package push
-
-type Consumer struct {
-	pushCh       ConsumerHandler
-	successCount uint64
-}
-
-func NewConsumer(pusher *Pusher) *Consumer {
-	return &Consumer{
-		pushCh: *NewConsumerHandler(pusher),
-	}
-}
-
-func (c *Consumer) Start() {
-	// statistics.NewStatistics(&c.successCount, config.Config.ModuleName.PushName, fmt.Sprintf("%d second push to
-	// msg_gateway count", constant.StatisticsTimeInterval), constant.StatisticsTimeInterval)
-	go c.pushCh.pushConsumerGroup.RegisterHandleAndConsumer(&c.pushCh)
-}

internal/push/offlinepush/dummy/push.go

@@ -16,17 +16,16 @@ package dummy
 
 import (
 	"context"
-
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
-	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
 )
 
-func NewClient() *Dummy {
+func NewDummy() *Dummy {
 	return &Dummy{}
 }
 
 type Dummy struct {
 }
 
-func (d *Dummy) Push(ctx context.Context, userIDs []string, title, content string, opts *offlinepush.Opts) error {
+func (d *Dummy) Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error {
 	return nil
 }

internal/push/offlinepush/fcm/push.go

@@ -16,6 +16,7 @@ package fcm
 
 import (
 	"context"
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
 	"path/filepath"
 
 	firebase "firebase.google.com/go"
@@ -25,7 +26,6 @@ import (
 
 	"github.com/OpenIMSDK/protocol/constant"
 
-	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
 )
@@ -39,7 +39,7 @@ type Fcm struct {
 	cache     cache.MsgModel
 }
 
-func NewClient(cache cache.MsgModel) *Fcm {
+func NewFcm(cache cache.MsgModel) *Fcm {
 	projectRoot := config.GetProjectRoot()
 	credentialsFilePath := filepath.Join(projectRoot, "config", config.Config.Push.Fcm.ServiceAccount)
 	opt := option.WithCredentialsFile(credentialsFilePath)
@@ -56,7 +56,7 @@ func NewClient(cache cache.MsgModel) *Fcm {
 	return &Fcm{fcmMsgCli: fcmMsgClient, cache: cache}
 }
 
-func (f *Fcm) Push(ctx context.Context, userIDs []string, title, content string, opts *offlinepush.Opts) error {
+func (f *Fcm) Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error {
 	// accounts->registrationToken
 	allTokens := make(map[string][]string, 0)
 	for _, account := range userIDs {

internal/push/offlinepush/getui/push.go

@@ -19,6 +19,7 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"errors"
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
 	"strconv"
 	"sync"
 	"time"
@@ -30,7 +31,6 @@ import (
 	"github.com/OpenIMSDK/tools/mcontext"
 	"github.com/OpenIMSDK/tools/utils/splitter"
 
-	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
 	http2 "github.com/openimsdk/open-im-server/v3/pkg/common/http"
@@ -55,17 +55,17 @@ const (
 	taskIDTTL       = 1000 * 60 * 60 * 24
 )
 
-type Client struct {
+type GeTui struct {
 	cache           cache.MsgModel
 	tokenExpireTime int64
 	taskIDTTL       int64
 }
 
-func NewClient(cache cache.MsgModel) *Client {
+func NewGeTui(cache cache.MsgModel) *GeTui {
-	return &Client{cache: cache, tokenExpireTime: tokenExpireTime, taskIDTTL: taskIDTTL}
+	return &GeTui{cache: cache, tokenExpireTime: tokenExpireTime, taskIDTTL: taskIDTTL}
 }
 
-func (g *Client) Push(ctx context.Context, userIDs []string, title, content string, opts *offlinepush.Opts) error {
+func (g *GeTui) Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error {
 	token, err := g.cache.GetGetuiToken(ctx)
 	if err != nil {
 		if errs.Unwrap(err) == redis.Nil {
@@ -111,7 +111,7 @@ func (g *Client) Push(ctx context.Context, userIDs []string, title, content stri
 	return err
 }
 
-func (g *Client) Auth(ctx context.Context, timeStamp int64) (token string, expireTime int64, err error) {
+func (g *GeTui) Auth(ctx context.Context, timeStamp int64) (token string, expireTime int64, err error) {
 	h := sha256.New()
 	h.Write(
 		[]byte(config.Config.Push.GeTui.AppKey + strconv.Itoa(int(timeStamp)) + config.Config.Push.GeTui.MasterSecret),
@@ -131,7 +131,7 @@ func (g *Client) Auth(ctx context.Context, timeStamp int64) (token string, expir
 	return respAuth.Token, int64(expire), err
 }
 
-func (g *Client) GetTaskID(ctx context.Context, token string, pushReq PushReq) (string, error) {
+func (g *GeTui) GetTaskID(ctx context.Context, token string, pushReq PushReq) (string, error) {
 	respTask := TaskResp{}
 	ttl := int64(1000 * 60 * 5)
 	pushReq.Settings = &Settings{TTL: &ttl}
@@ -143,7 +143,7 @@ func (g *Client) GetTaskID(ctx context.Context, token string, pushReq PushReq) (
 }
 
 // max num is 999.
-func (g *Client) batchPush(ctx context.Context, token string, userIDs []string, pushReq PushReq) error {
+func (g *GeTui) batchPush(ctx context.Context, token string, userIDs []string, pushReq PushReq) error {
 	taskID, err := g.GetTaskID(ctx, token, pushReq)
 	if err != nil {
 		return err
@@ -152,21 +152,21 @@ func (g *Client) batchPush(ctx context.Context, token string, userIDs []string,
 	return g.request(ctx, batchPushURL, pushReq, token, nil)
 }
 
-func (g *Client) singlePush(ctx context.Context, token, userID string, pushReq PushReq) error {
+func (g *GeTui) singlePush(ctx context.Context, token, userID string, pushReq PushReq) error {
 	operationID := mcontext.GetOperationID(ctx)
 	pushReq.RequestID = &operationID
 	pushReq.Audience = &Audience{Alias: []string{userID}}
 	return g.request(ctx, pushURL, pushReq, token, nil)
 }
 
-func (g *Client) request(ctx context.Context, url string, input any, token string, output any) error {
+func (g *GeTui) request(ctx context.Context, url string, input any, token string, output any) error {
 	header := map[string]string{"token": token}
 	resp := &Resp{}
 	resp.Data = output
 	return g.postReturn(ctx, config.Config.Push.GeTui.PushUrl+url, header, input, resp, 3)
 }
 
-func (g *Client) postReturn(
+func (g *GeTui) postReturn(
 	ctx context.Context,
 	url string,
 	header map[string]string,
@@ -181,7 +181,7 @@ func (g *Client) postReturn(
 	return output.parseError()
 }
 
-func (g *Client) getTokenAndSave2Redis(ctx context.Context) (token string, err error) {
+func (g *GeTui) getTokenAndSave2Redis(ctx context.Context) (token string, err error) {
 	token, _, err = g.Auth(ctx, time.Now().UnixNano()/1e6)
 	if err != nil {
 		return
@@ -193,7 +193,7 @@ func (g *Client) getTokenAndSave2Redis(ctx context.Context) (token string, err e
 	return token, nil
 }
 
-func (g *Client) GetTaskIDAndSave2Redis(ctx context.Context, token string, pushReq PushReq) (taskID string, err error) {
+func (g *GeTui) GetTaskIDAndSave2Redis(ctx context.Context, token string, pushReq PushReq) (taskID string, err error) {
 	pushReq.Settings = &Settings{TTL: &g.taskIDTTL}
 	taskID, err = g.GetTaskID(ctx, token, pushReq)
 	if err != nil {

internal/push/offlinepush/jpush/push.go

@@ -18,8 +18,8 @@ import (
 	"context"
 	"encoding/base64"
 	"fmt"
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
 
-	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
 	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/jpush/body"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	http2 "github.com/openimsdk/open-im-server/v3/pkg/common/http"
@@ -27,7 +27,7 @@ import (
 
 type JPush struct{}
 
-func NewClient() *JPush {
+func NewJPush() *JPush {
 	return &JPush{}
 }
 
@@ -46,7 +46,7 @@ func (j *JPush) getAuthorization(appKey string, masterSecret string) string {
 	return Authorization
 }
 
-func (j *JPush) Push(ctx context.Context, userIDs []string, title, content string, opts *offlinepush.Opts) error {
+func (j *JPush) Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error {
 	var pf body.Platform
 	pf.SetAll()
 	var au body.Audience

internal/push/offlinepush/offlinepush_interface.go

@@ -1,37 +0,0 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package offlinepush
-
-import (
-	"context"
-)
-
-// OfflinePusher Offline Pusher.
-type OfflinePusher interface {
-	Push(ctx context.Context, userIDs []string, title, content string, opts *Opts) error
-}
-
-// Opts opts.
-type Opts struct {
-	Signal        *Signal
-	IOSPushSound  string
-	IOSBadgeCount bool
-	Ex            string
-}
-
-// Signal message id.
-type Signal struct {
-	ClientMsgID string
-}

internal/push/offlinepush/offlinepusher.go

@@ -0,0 +1,52 @@
+// 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 offlinepush
+
+import (
+	"context"
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/dummy"
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/fcm"
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/getui"
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/jpush"
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
+)
+
+const (
+	GETUI    = "getui"
+	FIREBASE = "fcm"
+	JPUSH    = "jpush"
+)
+
+// OfflinePusher Offline Pusher.
+type OfflinePusher interface {
+	Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error
+}
+
+func NewOfflinePusher(cache cache.MsgModel) OfflinePusher {
+	var offlinePusher OfflinePusher
+	switch config.Config.Push.Enable {
+	case GETUI:
+		offlinePusher = getui.NewGeTui(cache)
+	case FIREBASE:
+		offlinePusher = fcm.NewFcm(cache)
+	case JPUSH:
+		offlinePusher = jpush.NewJPush()
+	default:
+		offlinePusher = dummy.NewDummy()
+	}
+	return offlinePusher
+}

internal/push/offlinepush/options/options.go

@@ -0,0 +1,14 @@
+package options
+
+// Opts opts.
+type Opts struct {
+	Signal        *Signal
+	IOSPushSound  string
+	IOSBadgeCount bool
+	Ex            string
+}
+
+// Signal message id.
+type Signal struct {
+	ClientMsgID string
+}

internal/push/onlinepusher.go

@@ -0,0 +1,211 @@
+package push
+
+import (
+	"context"
+	"github.com/OpenIMSDK/protocol/msggateway"
+	"github.com/OpenIMSDK/protocol/sdkws"
+	"github.com/OpenIMSDK/tools/discoveryregistry"
+	"github.com/OpenIMSDK/tools/log"
+	"github.com/OpenIMSDK/tools/utils"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"golang.org/x/sync/errgroup"
+	"google.golang.org/grpc"
+	"os"
+	"sync"
+)
+
+const (
+	ENVNAME    = "ENVS_DISCOVERY"
+	KUBERNETES = "k8s"
+	ZOOKEEPER  = "zookeeper"
+)
+
+type OnlinePusher interface {
+	GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData,
+		pushToUserIDs []string) (wsResults []*msggateway.SingleMsgToUserResults, err error)
+	GetOnlinePushFailedUserIDs(ctx context.Context, msg *sdkws.MsgData, wsResults []*msggateway.SingleMsgToUserResults,
+		pushToUserIDs *[]string) []string
+}
+
+type emptyOnlinePUsher struct{}
+
+func newEmptyOnlinePUsher() *emptyOnlinePUsher {
+	return &emptyOnlinePUsher{}
+}
+
+func (emptyOnlinePUsher) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData,
+	pushToUserIDs []string) (wsResults []*msggateway.SingleMsgToUserResults, err error) {
+	log.ZWarn(ctx, "emptyOnlinePUsher GetConnsAndOnlinePush", nil)
+	return nil, nil
+}
+func (u emptyOnlinePUsher) GetOnlinePushFailedUserIDs(ctx context.Context, msg *sdkws.MsgData,
+	wsResults []*msggateway.SingleMsgToUserResults, pushToUserIDs *[]string) []string {
+	log.ZWarn(ctx, "emptyOnlinePUsher GetOnlinePushFailedUserIDs", nil)
+	return nil
+}
+
+func NewOnlinePusher(disCov discoveryregistry.SvcDiscoveryRegistry) OnlinePusher {
+	var envType string
+	if value := os.Getenv(ENVNAME); value != "" {
+		envType = os.Getenv(ENVNAME)
+	} else {
+		envType = config.Config.Envs.Discovery
+	}
+	switch envType {
+	case KUBERNETES:
+		return NewK8sStaticConsistentHash(disCov)
+	case ZOOKEEPER:
+		return NewDefaultAllNode(disCov)
+	default:
+		return newEmptyOnlinePUsher()
+	}
+}
+
+type DefaultAllNode struct {
+	disCov discoveryregistry.SvcDiscoveryRegistry
+}
+
+func NewDefaultAllNode(disCov discoveryregistry.SvcDiscoveryRegistry) *DefaultAllNode {
+	return &DefaultAllNode{disCov: disCov}
+}
+
+func (d *DefaultAllNode) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData,
+	pushToUserIDs []string) (wsResults []*msggateway.SingleMsgToUserResults, err error) {
+	conns, err := d.disCov.GetConns(ctx, config.Config.RpcRegisterName.OpenImMessageGatewayName)
+	log.ZDebug(ctx, "get gateway conn", "conn length", len(conns))
+	if err != nil {
+		return nil, err
+	}
+
+	var (
+		mu         sync.Mutex
+		wg         = errgroup.Group{}
+		input      = &msggateway.OnlineBatchPushOneMsgReq{MsgData: msg, PushToUserIDs: pushToUserIDs}
+		maxWorkers = config.Config.Push.MaxConcurrentWorkers
+	)
+
+	if maxWorkers < 3 {
+		maxWorkers = 3
+	}
+
+	wg.SetLimit(maxWorkers)
+
+	// Online push message
+	for _, conn := range conns {
+		conn := conn // loop var safe
+		wg.Go(func() error {
+			msgClient := msggateway.NewMsgGatewayClient(conn)
+			reply, err := msgClient.SuperGroupOnlineBatchPushOneMsg(ctx, input)
+			if err != nil {
+				return nil
+			}
+
+			log.ZDebug(ctx, "push result", "reply", reply)
+			if reply != nil && reply.SinglePushResult != nil {
+				mu.Lock()
+				wsResults = append(wsResults, reply.SinglePushResult...)
+				mu.Unlock()
+			}
+
+			return nil
+		})
+	}
+
+	_ = wg.Wait()
+
+	// always return nil
+	return wsResults, nil
+}
+
+func (d *DefaultAllNode) GetOnlinePushFailedUserIDs(_ context.Context, msg *sdkws.MsgData,
+	wsResults []*msggateway.SingleMsgToUserResults, pushToUserIDs *[]string) []string {
+
+	onlineSuccessUserIDs := []string{msg.SendID}
+	for _, v := range wsResults {
+		//message sender do not need offline push
+		if msg.SendID == v.UserID {
+			continue
+		}
+		// mobile online push success
+		if v.OnlinePush {
+			onlineSuccessUserIDs = append(onlineSuccessUserIDs, v.UserID)
+		}
+
+	}
+
+	return utils.SliceSub(*pushToUserIDs, onlineSuccessUserIDs)
+}
+
+type K8sStaticConsistentHash struct {
+	disCov discoveryregistry.SvcDiscoveryRegistry
+}
+
+func NewK8sStaticConsistentHash(disCov discoveryregistry.SvcDiscoveryRegistry) *K8sStaticConsistentHash {
+	return &K8sStaticConsistentHash{disCov: disCov}
+}
+
+func (k *K8sStaticConsistentHash) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData,
+	pushToUserIDs []string) (wsResults []*msggateway.SingleMsgToUserResults, err error) {
+
+	var usersHost = make(map[string][]string)
+	for _, v := range pushToUserIDs {
+		tHost, err := k.disCov.GetUserIdHashGatewayHost(ctx, v)
+		if err != nil {
+			log.ZError(ctx, "get msg gateway hash error", err)
+			return nil, err
+		}
+		tUsers, tbl := usersHost[tHost]
+		if tbl {
+			tUsers = append(tUsers, v)
+			usersHost[tHost] = tUsers
+		} else {
+			usersHost[tHost] = []string{v}
+		}
+	}
+	log.ZDebug(ctx, "genUsers send hosts struct:", "usersHost", usersHost)
+	var usersConns = make(map[*grpc.ClientConn][]string)
+	for host, userIds := range usersHost {
+		tconn, _ := k.disCov.GetConn(ctx, host)
+		usersConns[tconn] = userIds
+	}
+	var (
+		mu         sync.Mutex
+		wg         = errgroup.Group{}
+		maxWorkers = config.Config.Push.MaxConcurrentWorkers
+	)
+	if maxWorkers < 3 {
+		maxWorkers = 3
+	}
+	wg.SetLimit(maxWorkers)
+	for conn, userIds := range usersConns {
+		tcon := conn
+		tuserIds := userIds
+		wg.Go(func() error {
+			input := &msggateway.OnlineBatchPushOneMsgReq{MsgData: msg, PushToUserIDs: tuserIds}
+			msgClient := msggateway.NewMsgGatewayClient(tcon)
+			reply, err := msgClient.SuperGroupOnlineBatchPushOneMsg(ctx, input)
+			if err != nil {
+				return nil
+			}
+			log.ZDebug(ctx, "push result", "reply", reply)
+			if reply != nil && reply.SinglePushResult != nil {
+				mu.Lock()
+				wsResults = append(wsResults, reply.SinglePushResult...)
+				mu.Unlock()
+			}
+			return nil
+		})
+	}
+	_ = wg.Wait()
+	return wsResults, nil
+}
+func (k *K8sStaticConsistentHash) GetOnlinePushFailedUserIDs(_ context.Context, _ *sdkws.MsgData,
+	wsResults []*msggateway.SingleMsgToUserResults, _ *[]string) []string {
+	var needOfflinePushUserIDs []string
+	for _, v := range wsResults {
+		if !v.OnlinePush {
+			needOfflinePushUserIDs = append(needOfflinePushUserIDs, v.UserID)
+		}
+	}
+	return needOfflinePushUserIDs
+}

internal/push/push.go

@@ -0,0 +1,51 @@
+package push
+
+import (
+	"context"
+	pbpush "github.com/OpenIMSDK/protocol/push"
+	"github.com/OpenIMSDK/tools/discoveryregistry"
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/db/controller"
+	"google.golang.org/grpc"
+)
+
+type pushServer struct {
+	database      controller.PushDatabase
+	disCov        discoveryregistry.SvcDiscoveryRegistry
+	offlinePusher offlinepush.OfflinePusher
+	pushCh        *ConsumerHandler
+}
+
+func (p pushServer) PushMsg(ctx context.Context, req *pbpush.PushMsgReq) (*pbpush.PushMsgResp, error) {
+	//todo reserved Interface
+	return nil, nil
+}
+
+func (p pushServer) DelUserPushToken(ctx context.Context,
+	req *pbpush.DelUserPushTokenReq) (resp *pbpush.DelUserPushTokenResp, err error) {
+	if err = p.database.DelFcmToken(ctx, req.UserID, int(req.PlatformID)); err != nil {
+		return nil, err
+	}
+	return &pbpush.DelUserPushTokenResp{}, nil
+}
+
+func Start(disCov discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) error {
+	rdb, err := cache.NewRedis()
+	if err != nil {
+		return err
+	}
+	cacheModel := cache.NewMsgCacheModel(rdb)
+	offlinePusher := offlinepush.NewOfflinePusher(cacheModel)
+	database := controller.NewPushDatabase(cacheModel)
+
+	consumer := NewConsumerHandler(offlinePusher, rdb, disCov)
+	pbpush.RegisterPushMsgServiceServer(server, &pushServer{
+		database:      database,
+		disCov:        disCov,
+		offlinePusher: offlinePusher,
+		pushCh:        consumer,
+	})
+	go consumer.pushConsumerGroup.RegisterHandleAndConsumer(consumer)
+	return nil
+}

internal/push/push_handler.go

@@ -16,6 +16,17 @@ package push
 
 import (
 	"context"
+	"encoding/json"
+	"github.com/OpenIMSDK/protocol/sdkws"
+	"github.com/OpenIMSDK/tools/discoveryregistry"
+	"github.com/OpenIMSDK/tools/mcontext"
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
+	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
+	"github.com/openimsdk/open-im-server/v3/pkg/rpccache"
+	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
+	"github.com/redis/go-redis/v9"
 
 	"github.com/IBM/sarama"
 	"google.golang.org/protobuf/proto"
@@ -31,18 +42,31 @@ import (
 )
 
 type ConsumerHandler struct {
-	pushConsumerGroup *kfk.MConsumerGroup
+	pushConsumerGroup      *kfk.MConsumerGroup
-	pusher            *Pusher
+	offlinePusher          offlinepush.OfflinePusher
+	onlinePusher           OnlinePusher
+	groupLocalCache        *rpccache.GroupLocalCache
+	conversationLocalCache *rpccache.ConversationLocalCache
+	msgRpcClient           rpcclient.MessageRpcClient
+	conversationRpcClient  rpcclient.ConversationRpcClient
+	groupRpcClient         rpcclient.GroupRpcClient
 }
 
-func NewConsumerHandler(pusher *Pusher) *ConsumerHandler {
+func NewConsumerHandler(offlinePusher offlinepush.OfflinePusher,
+	rdb redis.UniversalClient, disCov discoveryregistry.SvcDiscoveryRegistry) *ConsumerHandler {
 	var consumerHandler ConsumerHandler
-	consumerHandler.pusher = pusher
 	consumerHandler.pushConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{
 		KafkaVersion:   sarama.V2_0_0_0,
 		OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false,
 	}, []string{config.Config.Kafka.MsgToPush.Topic}, config.Config.Kafka.Addr,
 		config.Config.Kafka.ConsumerGroupID.MsgToPush)
+	consumerHandler.offlinePusher = offlinePusher
+	consumerHandler.onlinePusher = NewOnlinePusher(disCov)
+	consumerHandler.groupRpcClient = rpcclient.NewGroupRpcClient(disCov)
+	consumerHandler.groupLocalCache = rpccache.NewGroupLocalCache(consumerHandler.groupRpcClient, rdb)
+	consumerHandler.msgRpcClient = rpcclient.NewMessageRpcClient(disCov)
+	consumerHandler.conversationRpcClient = rpcclient.NewConversationRpcClient(disCov)
+	consumerHandler.conversationLocalCache = rpccache.NewConversationLocalCache(consumerHandler.conversationRpcClient, rdb)
 	return &consumerHandler
 }
 
@@ -65,7 +89,7 @@ func (c *ConsumerHandler) handleMs2PsChat(ctx context.Context, msg []byte) {
 	var err error
 	switch msgFromMQ.MsgData.SessionType {
 	case constant.SuperGroupChatType:
-		err = c.pusher.Push2SuperGroup(ctx, pbData.MsgData.GroupID, pbData.MsgData)
+		err = c.Push2SuperGroup(ctx, pbData.MsgData.GroupID, pbData.MsgData)
 	default:
 		var pushUserIDList []string
 		isSenderSync := utils.GetSwitchFromOptions(pbData.MsgData.Options, constant.IsSenderSync)
@@ -74,18 +98,14 @@ func (c *ConsumerHandler) handleMs2PsChat(ctx context.Context, msg []byte) {
 		} else {
 			pushUserIDList = append(pushUserIDList, pbData.MsgData.RecvID, pbData.MsgData.SendID)
 		}
-		err = c.pusher.Push2User(ctx, pushUserIDList, pbData.MsgData)
+		err = c.Push2User(ctx, pushUserIDList, pbData.MsgData)
 	}
 	if err != nil {
-		if err == errNoOfflinePusher {
+		log.ZError(ctx, "push failed", err, "msg", pbData.String())
-			log.ZWarn(ctx, "offline push failed", err, "msg", pbData.String())
-		} else {
-			log.ZError(ctx, "push failed", err, "msg", pbData.String())
-		}
 	}
 }
-func (ConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
+func (*ConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
-func (ConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
+func (*ConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
 func (c *ConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
 	claim sarama.ConsumerGroupClaim,
 ) error {
@@ -96,3 +116,243 @@ func (c *ConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
 	}
 	return nil
 }
+
+// Push2User Suitable for two types of conversations, one is SingleChatType and the other is NotificationChatType.
+func (c *ConsumerHandler) Push2User(ctx context.Context, userIDs []string, msg *sdkws.MsgData) error {
+	log.ZDebug(ctx, "Get msg from msg_transfer And push msg", "userIDs", userIDs, "msg", msg.String())
+	if err := callbackOnlinePush(ctx, userIDs, msg); err != nil {
+		return err
+	}
+
+	wsResults, err := c.onlinePusher.GetConnsAndOnlinePush(ctx, msg, userIDs)
+	if err != nil {
+		return err
+	}
+
+	log.ZDebug(ctx, "single and notification push result", "result", wsResults, "msg", msg, "push_to_userID", userIDs)
+
+	if !c.shouldPushOffline(ctx, msg) {
+		return nil
+	}
+
+	for _, v := range wsResults {
+		//message sender do not need offline push
+		if msg.SendID == v.UserID {
+			continue
+		}
+		//receiver online push success
+		if v.OnlinePush {
+			return nil
+		}
+	}
+	offlinePUshUserID := []string{msg.RecvID}
+	//receiver offline push
+	if err = callbackOfflinePush(ctx, offlinePUshUserID, msg, nil); err != nil {
+		return err
+	}
+
+	err = c.offlinePushMsg(ctx, msg, offlinePUshUserID)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (c *ConsumerHandler) Push2SuperGroup(ctx context.Context, groupID string, msg *sdkws.MsgData) (err error) {
+	log.ZDebug(ctx, "Get super group msg from msg_transfer and push msg", "msg", msg.String(), "groupID", groupID)
+	var pushToUserIDs []string
+	if err = callbackBeforeSuperGroupOnlinePush(ctx, groupID, msg, &pushToUserIDs); err != nil {
+		return err
+	}
+
+	err = c.groupMessagesHandler(ctx, groupID, &pushToUserIDs, msg)
+	if err != nil {
+		return err
+	}
+
+	wsResults, err := c.onlinePusher.GetConnsAndOnlinePush(ctx, msg, pushToUserIDs)
+	if err != nil {
+		return err
+	}
+
+	log.ZDebug(ctx, "group push result", "result", wsResults, "msg", msg)
+
+	if !c.shouldPushOffline(ctx, msg) {
+		return nil
+	}
+	needOfflinePushUserIDs := c.onlinePusher.GetOnlinePushFailedUserIDs(ctx, msg, wsResults, &pushToUserIDs)
+
+	//filter some user, like don not disturb or don't need offline push etc.
+	needOfflinePushUserIDs, err = c.filterGroupMessageOfflinePush(ctx, groupID, msg, needOfflinePushUserIDs)
+	if err != nil {
+		return err
+	}
+	// Use offline push messaging
+	if len(needOfflinePushUserIDs) > 0 {
+		var offlinePushUserIDs []string
+		err = callbackOfflinePush(ctx, needOfflinePushUserIDs, msg, &offlinePushUserIDs)
+		if err != nil {
+			return err
+		}
+
+		if len(offlinePushUserIDs) > 0 {
+			needOfflinePushUserIDs = offlinePushUserIDs
+		}
+
+		err = c.offlinePushMsg(ctx, msg, needOfflinePushUserIDs)
+		if err != nil {
+			log.ZError(ctx, "offlinePushMsg failed", err, "groupID", groupID, "msg", msg)
+			return err
+		}
+
+	}
+
+	return nil
+}
+
+func (c *ConsumerHandler) offlinePushMsg(ctx context.Context, msg *sdkws.MsgData, offlinePushUserIDs []string) error {
+	title, content, opts, err := c.getOfflinePushInfos(msg)
+	if err != nil {
+		return err
+	}
+	err = c.offlinePusher.Push(ctx, offlinePushUserIDs, title, content, opts)
+	if err != nil {
+		prommetrics.MsgOfflinePushFailedCounter.Inc()
+		return err
+	}
+	return nil
+}
+
+func (c *ConsumerHandler) filterGroupMessageOfflinePush(ctx context.Context, groupID string, msg *sdkws.MsgData,
+	offlinePushUserIDs []string) (userIDs []string, err error) {
+
+	//todo local cache Obtain the difference set through local comparison.
+	needOfflinePushUserIDs, err := c.conversationRpcClient.GetConversationOfflinePushUserIDs(
+		ctx, utils.GenGroupConversationID(groupID), offlinePushUserIDs)
+	if err != nil {
+		return nil, err
+	}
+	return needOfflinePushUserIDs, nil
+}
+
+func (c *ConsumerHandler) getOfflinePushInfos(msg *sdkws.MsgData) (title, content string, opts *options.Opts, err error) {
+	type AtTextElem struct {
+		Text       string   `json:"text,omitempty"`
+		AtUserList []string `json:"atUserList,omitempty"`
+		IsAtSelf   bool     `json:"isAtSelf"`
+	}
+
+	opts = &options.Opts{Signal: &options.Signal{}}
+	if msg.OfflinePushInfo != nil {
+		opts.IOSBadgeCount = msg.OfflinePushInfo.IOSBadgeCount
+		opts.IOSPushSound = msg.OfflinePushInfo.IOSPushSound
+		opts.Ex = msg.OfflinePushInfo.Ex
+	}
+
+	if msg.OfflinePushInfo != nil {
+		title = msg.OfflinePushInfo.Title
+		content = msg.OfflinePushInfo.Desc
+	}
+	if title == "" {
+		switch msg.ContentType {
+		case constant.Text:
+			fallthrough
+		case constant.Picture:
+			fallthrough
+		case constant.Voice:
+			fallthrough
+		case constant.Video:
+			fallthrough
+		case constant.File:
+			title = constant.ContentType2PushContent[int64(msg.ContentType)]
+		case constant.AtText:
+			ac := AtTextElem{}
+			_ = utils.JsonStringToStruct(string(msg.Content), &ac)
+		case constant.SignalingNotification:
+			title = constant.ContentType2PushContent[constant.SignalMsg]
+		default:
+			title = constant.ContentType2PushContent[constant.Common]
+		}
+	}
+	if content == "" {
+		content = title
+	}
+	return
+}
+func (c *ConsumerHandler) groupMessagesHandler(ctx context.Context, groupID string, pushToUserIDs *[]string, msg *sdkws.MsgData) (err error) {
+	if len(*pushToUserIDs) == 0 {
+		*pushToUserIDs, err = c.groupLocalCache.GetGroupMemberIDs(ctx, groupID)
+		if err != nil {
+			return err
+		}
+		switch msg.ContentType {
+		case constant.MemberQuitNotification:
+			var tips sdkws.MemberQuitTips
+			if unmarshalNotificationElem(msg.Content, &tips) != nil {
+				return err
+			}
+			if err = c.DeleteMemberAndSetConversationSeq(ctx, groupID, []string{tips.QuitUser.UserID}); err != nil {
+				log.ZError(ctx, "MemberQuitNotification DeleteMemberAndSetConversationSeq", err, "groupID", groupID, "userID", tips.QuitUser.UserID)
+			}
+			*pushToUserIDs = append(*pushToUserIDs, tips.QuitUser.UserID)
+		case constant.MemberKickedNotification:
+			var tips sdkws.MemberKickedTips
+			if unmarshalNotificationElem(msg.Content, &tips) != nil {
+				return err
+			}
+			kickedUsers := utils.Slice(tips.KickedUserList, func(e *sdkws.GroupMemberFullInfo) string { return e.UserID })
+			if err = c.DeleteMemberAndSetConversationSeq(ctx, groupID, kickedUsers); err != nil {
+				log.ZError(ctx, "MemberKickedNotification DeleteMemberAndSetConversationSeq", err, "groupID", groupID, "userIDs", kickedUsers)
+			}
+
+			*pushToUserIDs = append(*pushToUserIDs, kickedUsers...)
+		case constant.GroupDismissedNotification:
+			if msgprocessor.IsNotification(msgprocessor.GetConversationIDByMsg(msg)) { // 消息先到,通知后到
+				var tips sdkws.GroupDismissedTips
+				if unmarshalNotificationElem(msg.Content, &tips) != nil {
+					return err
+				}
+				log.ZInfo(ctx, "GroupDismissedNotificationInfo****", "groupID", groupID, "num", len(*pushToUserIDs), "list", pushToUserIDs)
+				if len(config.Config.Manager.UserID) > 0 {
+					ctx = mcontext.WithOpUserIDContext(ctx, config.Config.Manager.UserID[0])
+				}
+				defer func(groupID string) {
+					if err = c.groupRpcClient.DismissGroup(ctx, groupID); err != nil {
+						log.ZError(ctx, "DismissGroup Notification clear members", err, "groupID", groupID)
+					}
+				}(groupID)
+			}
+		}
+	}
+	return err
+}
+
+func (c *ConsumerHandler) DeleteMemberAndSetConversationSeq(ctx context.Context, groupID string, userIDs []string) error {
+	conversationID := msgprocessor.GetConversationIDBySessionType(constant.SuperGroupChatType, groupID)
+	maxSeq, err := c.msgRpcClient.GetConversationMaxSeq(ctx, conversationID)
+	if err != nil {
+		return err
+	}
+	return c.conversationRpcClient.SetConversationMaxSeq(ctx, userIDs, conversationID, maxSeq)
+}
+
+func unmarshalNotificationElem(bytes []byte, t any) error {
+	var notification sdkws.NotificationElem
+	if err := json.Unmarshal(bytes, &notification); err != nil {
+		return err
+	}
+
+	return json.Unmarshal([]byte(notification.Detail), t)
+}
+
+func (c *ConsumerHandler) shouldPushOffline(_ context.Context, msg *sdkws.MsgData) bool {
+	isOfflinePush := utils.GetSwitchFromOptions(msg.Options, constant.IsOfflinePush)
+	if !isOfflinePush {
+		return false
+	}
+	if msg.ContentType == constant.SignalingNotification {
+		return false
+	}
+	return true
+}

internal/push/push_rpc_server.go

@@ -1,109 +0,0 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package push
-
-import (
-	"context"
-	"github.com/OpenIMSDK/tools/utils"
-	"sync"
-
-	"google.golang.org/grpc"
-
-	"github.com/OpenIMSDK/protocol/constant"
-	pbpush "github.com/OpenIMSDK/protocol/push"
-	"github.com/OpenIMSDK/tools/discoveryregistry"
-	"github.com/OpenIMSDK/tools/log"
-
-	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/db/controller"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/db/localcache"
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
-)
-
-type pushServer struct {
-	pusher *Pusher
-}
-
-func Start(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) error {
-	rdb, err := cache.NewRedis()
-	if err != nil {
-		return err
-	}
-	cacheModel := cache.NewMsgCacheModel(rdb)
-	offlinePusher := NewOfflinePusher(cacheModel)
-	database := controller.NewPushDatabase(cacheModel)
-	groupRpcClient := rpcclient.NewGroupRpcClient(client)
-	conversationRpcClient := rpcclient.NewConversationRpcClient(client)
-	msgRpcClient := rpcclient.NewMessageRpcClient(client)
-	pusher := NewPusher(
-		client,
-		offlinePusher,
-		database,
-		localcache.NewGroupLocalCache(&groupRpcClient),
-		localcache.NewConversationLocalCache(&conversationRpcClient),
-		&conversationRpcClient,
-		&groupRpcClient,
-		&msgRpcClient,
-	)
-	var wg sync.WaitGroup
-	wg.Add(2)
-	go func() {
-		defer wg.Done()
-		pbpush.RegisterPushMsgServiceServer(server, &pushServer{
-			pusher: pusher,
-		})
-	}()
-	go func() {
-		defer wg.Done()
-		consumer := NewConsumer(pusher)
-		consumer.Start()
-	}()
-	wg.Wait()
-	return nil
-}
-
-func (r *pushServer) PushMsg(ctx context.Context, pbData *pbpush.PushMsgReq) (resp *pbpush.PushMsgResp, err error) {
-	switch pbData.MsgData.SessionType {
-	case constant.SuperGroupChatType:
-		err = r.pusher.Push2SuperGroup(ctx, pbData.MsgData.GroupID, pbData.MsgData)
-	default:
-		var pushUserIDList []string
-		isSenderSync := utils.GetSwitchFromOptions(pbData.MsgData.Options, constant.IsSenderSync)
-		if !isSenderSync {
-			pushUserIDList = append(pushUserIDList, pbData.MsgData.RecvID)
-		} else {
-			pushUserIDList = append(pushUserIDList, pbData.MsgData.RecvID, pbData.MsgData.SendID)
-		}
-		err = r.pusher.Push2User(ctx, pushUserIDList, pbData.MsgData)
-	}
-	if err != nil {
-		if err != errNoOfflinePusher {
-			return nil, err
-		} else {
-			log.ZWarn(ctx, "offline push failed", err, "msg", pbData.String())
-		}
-	}
-	return &pbpush.PushMsgResp{}, nil
-}
-
-func (r *pushServer) DelUserPushToken(
-	ctx context.Context,
-	req *pbpush.DelUserPushTokenReq,
-) (resp *pbpush.DelUserPushTokenResp, err error) {
-	if err = r.pusher.database.DelFcmToken(ctx, req.UserID, int(req.PlatformID)); err != nil {
-		return nil, err
-	}
-	return &pbpush.DelUserPushTokenResp{}, nil
-}

internal/push/push_to_client.go

@@ -1,511 +0,0 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package push
-
-import (
-	"context"
-	"encoding/json"
-	"errors"
-	"google.golang.org/grpc"
-	"sync"
-
-	"golang.org/x/sync/errgroup"
-
-	"github.com/OpenIMSDK/protocol/constant"
-	"github.com/OpenIMSDK/protocol/conversation"
-	"github.com/OpenIMSDK/protocol/msggateway"
-	"github.com/OpenIMSDK/protocol/sdkws"
-	"github.com/OpenIMSDK/tools/discoveryregistry"
-	"github.com/OpenIMSDK/tools/log"
-	"github.com/OpenIMSDK/tools/mcontext"
-	"github.com/OpenIMSDK/tools/utils"
-
-	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
-	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/dummy"
-	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/fcm"
-	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/getui"
-	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/jpush"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/db/controller"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/db/localcache"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
-	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
-)
-
-type Pusher struct {
-	database               controller.PushDatabase
-	discov                 discoveryregistry.SvcDiscoveryRegistry
-	offlinePusher          offlinepush.OfflinePusher
-	groupLocalCache        *localcache.GroupLocalCache
-	conversationLocalCache *localcache.ConversationLocalCache
-	msgRpcClient           *rpcclient.MessageRpcClient
-	conversationRpcClient  *rpcclient.ConversationRpcClient
-	groupRpcClient         *rpcclient.GroupRpcClient
-}
-
-var errNoOfflinePusher = errors.New("no offlinePusher is configured")
-
-func NewPusher(discov discoveryregistry.SvcDiscoveryRegistry, offlinePusher offlinepush.OfflinePusher, database controller.PushDatabase,
-	groupLocalCache *localcache.GroupLocalCache, conversationLocalCache *localcache.ConversationLocalCache,
-	conversationRpcClient *rpcclient.ConversationRpcClient, groupRpcClient *rpcclient.GroupRpcClient, msgRpcClient *rpcclient.MessageRpcClient,
-) *Pusher {
-	return &Pusher{
-		discov:                 discov,
-		database:               database,
-		offlinePusher:          offlinePusher,
-		groupLocalCache:        groupLocalCache,
-		conversationLocalCache: conversationLocalCache,
-		msgRpcClient:           msgRpcClient,
-		conversationRpcClient:  conversationRpcClient,
-		groupRpcClient:         groupRpcClient,
-	}
-}
-
-func NewOfflinePusher(cache cache.MsgModel) offlinepush.OfflinePusher {
-	var offlinePusher offlinepush.OfflinePusher
-	switch config.Config.Push.Enable {
-	case "getui":
-		offlinePusher = getui.NewClient(cache)
-	case "fcm":
-		offlinePusher = fcm.NewClient(cache)
-	case "jpush":
-		offlinePusher = jpush.NewClient()
-	default:
-		offlinePusher = dummy.NewClient()
-	}
-	return offlinePusher
-}
-
-func (p *Pusher) DeleteMemberAndSetConversationSeq(ctx context.Context, groupID string, userIDs []string) error {
-	conevrsationID := msgprocessor.GetConversationIDBySessionType(constant.SuperGroupChatType, groupID)
-	maxSeq, err := p.msgRpcClient.GetConversationMaxSeq(ctx, conevrsationID)
-	if err != nil {
-		return err
-	}
-	return p.conversationRpcClient.SetConversationMaxSeq(ctx, userIDs, conevrsationID, maxSeq)
-}
-
-func (p *Pusher) Push2User(ctx context.Context, userIDs []string, msg *sdkws.MsgData) error {
-	log.ZDebug(ctx, "Get msg from msg_transfer And push msg", "userIDs", userIDs, "msg", msg.String())
-	if err := callbackOnlinePush(ctx, userIDs, msg); err != nil {
-		return err
-	}
-	// push
-	wsResults, err := p.GetConnsAndOnlinePush(ctx, msg, userIDs)
-	if err != nil {
-		return err
-	}
-
-	isOfflinePush := utils.GetSwitchFromOptions(msg.Options, constant.IsOfflinePush)
-	log.ZDebug(ctx, "push_result", "ws push result", wsResults, "sendData", msg, "isOfflinePush", isOfflinePush, "push_to_userID", userIDs)
-
-	if !isOfflinePush {
-		return nil
-	}
-
-	for _, v := range wsResults {
-		if !v.OnlinePush && msg.SendID == v.UserID {
-			if err = callbackOfflinePush(ctx, userIDs, msg, &[]string{}); err != nil {
-				return err
-			}
-
-			err = p.offlinePushMsg(ctx, msg.SendID, msg, []string{v.UserID})
-			if err != nil {
-				return err
-			}
-		}
-
-	}
-	return nil
-}
-
-func (p *Pusher) UnmarshalNotificationElem(bytes []byte, t any) error {
-	var notification sdkws.NotificationElem
-	if err := json.Unmarshal(bytes, &notification); err != nil {
-		return err
-	}
-
-	return json.Unmarshal([]byte(notification.Detail), t)
-}
-
-/*
-k8s deployment,offline push group messages function
-*/
-func (p *Pusher) k8sOfflinePush2SuperGroup(ctx context.Context, groupID string, msg *sdkws.MsgData, wsResults []*msggateway.SingleMsgToUserResults) error {
-
-	var needOfflinePushUserIDs []string
-	for _, v := range wsResults {
-		if !v.OnlinePush {
-			needOfflinePushUserIDs = append(needOfflinePushUserIDs, v.UserID)
-		}
-	}
-	if len(needOfflinePushUserIDs) > 0 {
-		var offlinePushUserIDs []string
-		err := callbackOfflinePush(ctx, needOfflinePushUserIDs, msg, &offlinePushUserIDs)
-		if err != nil {
-			return err
-		}
-
-		if len(offlinePushUserIDs) > 0 {
-			needOfflinePushUserIDs = offlinePushUserIDs
-		}
-		if msg.ContentType != constant.SignalingNotification {
-			resp, err := p.conversationRpcClient.Client.GetConversationOfflinePushUserIDs(
-				ctx,
-				&conversation.GetConversationOfflinePushUserIDsReq{ConversationID: utils.GenGroupConversationID(groupID), UserIDs: needOfflinePushUserIDs},
-			)
-			if err != nil {
-				return err
-			}
-			if len(resp.UserIDs) > 0 {
-				err = p.offlinePushMsg(ctx, groupID, msg, resp.UserIDs)
-				if err != nil {
-					log.ZError(ctx, "offlinePushMsg failed", err, "groupID", groupID, "msg", msg)
-					return err
-				}
-			}
-		}
-
-	}
-	return nil
-}
-func (p *Pusher) Push2SuperGroup(ctx context.Context, groupID string, msg *sdkws.MsgData) (err error) {
-	log.ZDebug(ctx, "Get super group msg from msg_transfer and push msg", "msg", msg.String(), "groupID", groupID)
-	var pushToUserIDs []string
-	if err = callbackBeforeSuperGroupOnlinePush(ctx, groupID, msg, &pushToUserIDs); err != nil {
-		return err
-	}
-
-	if len(pushToUserIDs) == 0 {
-		pushToUserIDs, err = p.groupLocalCache.GetGroupMemberIDs(ctx, groupID)
-		if err != nil {
-			return err
-		}
-
-		switch msg.ContentType {
-		case constant.MemberQuitNotification:
-			var tips sdkws.MemberQuitTips
-			if p.UnmarshalNotificationElem(msg.Content, &tips) != nil {
-				return err
-			}
-			defer func(groupID string, userIDs []string) {
-				if err = p.DeleteMemberAndSetConversationSeq(ctx, groupID, userIDs); err != nil {
-					log.ZError(ctx, "MemberQuitNotification DeleteMemberAndSetConversationSeq", err, "groupID", groupID, "userIDs", userIDs)
-				}
-			}(groupID, []string{tips.QuitUser.UserID})
-			pushToUserIDs = append(pushToUserIDs, tips.QuitUser.UserID)
-		case constant.MemberKickedNotification:
-			var tips sdkws.MemberKickedTips
-			if p.UnmarshalNotificationElem(msg.Content, &tips) != nil {
-				return err
-			}
-			kickedUsers := utils.Slice(tips.KickedUserList, func(e *sdkws.GroupMemberFullInfo) string { return e.UserID })
-			defer func(groupID string, userIDs []string) {
-				if err = p.DeleteMemberAndSetConversationSeq(ctx, groupID, userIDs); err != nil {
-					log.ZError(ctx, "MemberKickedNotification DeleteMemberAndSetConversationSeq", err, "groupID", groupID, "userIDs", userIDs)
-				}
-			}(groupID, kickedUsers)
-			pushToUserIDs = append(pushToUserIDs, kickedUsers...)
-		case constant.GroupDismissedNotification:
-			if msgprocessor.IsNotification(msgprocessor.GetConversationIDByMsg(msg)) { // 消息先到,通知后到
-				var tips sdkws.GroupDismissedTips
-				if p.UnmarshalNotificationElem(msg.Content, &tips) != nil {
-					return err
-				}
-				log.ZInfo(ctx, "GroupDismissedNotificationInfo****", "groupID", groupID, "num", len(pushToUserIDs), "list", pushToUserIDs)
-				if len(config.Config.Manager.UserID) > 0 {
-					ctx = mcontext.WithOpUserIDContext(ctx, config.Config.Manager.UserID[0])
-				}
-				defer func(groupID string) {
-					if err = p.groupRpcClient.DismissGroup(ctx, groupID); err != nil {
-						log.ZError(ctx, "DismissGroup Notification clear members", err, "groupID", groupID)
-					}
-				}(groupID)
-			}
-		}
-	}
-
-	wsResults, err := p.GetConnsAndOnlinePush(ctx, msg, pushToUserIDs)
-	if err != nil {
-		return err
-	}
-
-	log.ZDebug(ctx, "get conn and online push success", "result", wsResults, "msg", msg)
-	isOfflinePush := utils.GetSwitchFromOptions(msg.Options, constant.IsOfflinePush)
-	if isOfflinePush && config.Config.Envs.Discovery == "k8s" {
-		return p.k8sOfflinePush2SuperGroup(ctx, groupID, msg, wsResults)
-	}
-	if isOfflinePush && config.Config.Envs.Discovery == "zookeeper" {
-		var (
-			onlineSuccessUserIDs      = []string{msg.SendID}
-			webAndPcBackgroundUserIDs []string
-		)
-
-		for _, v := range wsResults {
-			if v.OnlinePush && v.UserID != msg.SendID {
-				onlineSuccessUserIDs = append(onlineSuccessUserIDs, v.UserID)
-			}
-
-			if v.OnlinePush {
-				continue
-			}
-
-			if len(v.Resp) == 0 {
-				continue
-			}
-
-			for _, singleResult := range v.Resp {
-				if singleResult.ResultCode != -2 {
-					continue
-				}
-
-				isPC := constant.PlatformIDToName(int(singleResult.RecvPlatFormID)) == constant.TerminalPC
-				isWebID := singleResult.RecvPlatFormID == constant.WebPlatformID
-
-				if isPC || isWebID {
-					webAndPcBackgroundUserIDs = append(webAndPcBackgroundUserIDs, v.UserID)
-				}
-			}
-		}
-
-		needOfflinePushUserIDs := utils.DifferenceString(onlineSuccessUserIDs, pushToUserIDs)
-
-		// Use offline push messaging
-		if len(needOfflinePushUserIDs) > 0 {
-			var offlinePushUserIDs []string
-			err = callbackOfflinePush(ctx, needOfflinePushUserIDs, msg, &offlinePushUserIDs)
-			if err != nil {
-				return err
-			}
-
-			if len(offlinePushUserIDs) > 0 {
-				needOfflinePushUserIDs = offlinePushUserIDs
-			}
-			if msg.ContentType != constant.SignalingNotification {
-				resp, err := p.conversationRpcClient.Client.GetConversationOfflinePushUserIDs(
-					ctx,
-					&conversation.GetConversationOfflinePushUserIDsReq{ConversationID: utils.GenGroupConversationID(groupID), UserIDs: needOfflinePushUserIDs},
-				)
-				if err != nil {
-					return err
-				}
-				if len(resp.UserIDs) > 0 {
-					err = p.offlinePushMsg(ctx, groupID, msg, resp.UserIDs)
-					if err != nil {
-						log.ZError(ctx, "offlinePushMsg failed", err, "groupID", groupID, "msg", msg)
-						return err
-					}
-					if _, err := p.GetConnsAndOnlinePush(ctx, msg, utils.IntersectString(resp.UserIDs, webAndPcBackgroundUserIDs)); err != nil {
-						log.ZError(ctx, "offlinePushMsg failed", err, "groupID", groupID, "msg", msg, "userIDs", utils.IntersectString(needOfflinePushUserIDs, webAndPcBackgroundUserIDs))
-						return err
-					}
-				}
-			}
-
-		}
-	}
-	return nil
-}
-
-func (p *Pusher) k8sOnlinePush(ctx context.Context, msg *sdkws.MsgData, pushToUserIDs []string) (wsResults []*msggateway.SingleMsgToUserResults, err error) {
-	var usersHost = make(map[string][]string)
-	for _, v := range pushToUserIDs {
-		tHost, err := p.discov.GetUserIdHashGatewayHost(ctx, v)
-		if err != nil {
-			log.ZError(ctx, "get msggateway hash error", err)
-			return nil, err
-		}
-		tUsers, tbl := usersHost[tHost]
-		if tbl {
-			tUsers = append(tUsers, v)
-			usersHost[tHost] = tUsers
-		} else {
-			usersHost[tHost] = []string{v}
-		}
-	}
-	log.ZDebug(ctx, "genUsers send hosts struct:", "usersHost", usersHost)
-	var usersConns = make(map[*grpc.ClientConn][]string)
-	for host, userIds := range usersHost {
-		tconn, _ := p.discov.GetConn(ctx, host)
-		usersConns[tconn] = userIds
-	}
-	var (
-		mu         sync.Mutex
-		wg         = errgroup.Group{}
-		maxWorkers = config.Config.Push.MaxConcurrentWorkers
-	)
-	if maxWorkers < 3 {
-		maxWorkers = 3
-	}
-	wg.SetLimit(maxWorkers)
-	for conn, userIds := range usersConns {
-		tcon := conn
-		tuserIds := userIds
-		wg.Go(func() error {
-			input := &msggateway.OnlineBatchPushOneMsgReq{MsgData: msg, PushToUserIDs: tuserIds}
-			msgClient := msggateway.NewMsgGatewayClient(tcon)
-			reply, err := msgClient.SuperGroupOnlineBatchPushOneMsg(ctx, input)
-			if err != nil {
-				return nil
-			}
-			log.ZDebug(ctx, "push result", "reply", reply)
-			if reply != nil && reply.SinglePushResult != nil {
-				mu.Lock()
-				wsResults = append(wsResults, reply.SinglePushResult...)
-				mu.Unlock()
-			}
-			return nil
-		})
-	}
-	_ = wg.Wait()
-	return wsResults, nil
-}
-func (p *Pusher) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData, pushToUserIDs []string) (wsResults []*msggateway.SingleMsgToUserResults, err error) {
-	if config.Config.Envs.Discovery == "k8s" {
-		return p.k8sOnlinePush(ctx, msg, pushToUserIDs)
-	}
-	conns, err := p.discov.GetConns(ctx, config.Config.RpcRegisterName.OpenImMessageGatewayName)
-	log.ZDebug(ctx, "get gateway conn", "conn length", len(conns))
-	if err != nil {
-		return nil, err
-	}
-
-	var (
-		mu         sync.Mutex
-		wg         = errgroup.Group{}
-		input      = &msggateway.OnlineBatchPushOneMsgReq{MsgData: msg, PushToUserIDs: pushToUserIDs}
-		maxWorkers = config.Config.Push.MaxConcurrentWorkers
-	)
-
-	if maxWorkers < 3 {
-		maxWorkers = 3
-	}
-
-	wg.SetLimit(maxWorkers)
-
-	// Online push message
-	for _, conn := range conns {
-		conn := conn // loop var safe
-		wg.Go(func() error {
-			msgClient := msggateway.NewMsgGatewayClient(conn)
-			reply, err := msgClient.SuperGroupOnlineBatchPushOneMsg(ctx, input)
-			if err != nil {
-				return nil
-			}
-
-			log.ZDebug(ctx, "push result", "reply", reply)
-			if reply != nil && reply.SinglePushResult != nil {
-				mu.Lock()
-				wsResults = append(wsResults, reply.SinglePushResult...)
-				mu.Unlock()
-			}
-
-			return nil
-		})
-	}
-
-	_ = wg.Wait()
-
-	// always return nil
-	return wsResults, nil
-}
-
-func (p *Pusher) offlinePushMsg(ctx context.Context, conversationID string, msg *sdkws.MsgData, offlinePushUserIDs []string) error {
-	title, content, opts, err := p.getOfflinePushInfos(conversationID, msg)
-	if err != nil {
-		return err
-	}
-	err = p.offlinePusher.Push(ctx, offlinePushUserIDs, title, content, opts)
-	if err != nil {
-		prommetrics.MsgOfflinePushFailedCounter.Inc()
-		return err
-	}
-	return nil
-}
-
-func (p *Pusher) GetOfflinePushOpts(msg *sdkws.MsgData) (opts *offlinepush.Opts, err error) {
-	opts = &offlinepush.Opts{Signal: &offlinepush.Signal{}}
-	// if msg.ContentType > constant.SignalingNotificationBegin && msg.ContentType < constant.SignalingNotificationEnd {
-	// 	req := &sdkws.SignalReq{}
-	// 	if err := proto.Unmarshal(msg.Content, req); err != nil {
-	// 		return nil, utils.Wrap(err, "")
-	// 	}
-	// 	switch req.Payload.(type) {
-	// 	case *sdkws.SignalReq_Invite, *sdkws.SignalReq_InviteInGroup:
-	// 		opts.Signal = &offlinepush.Signal{ClientMsgID: msg.ClientMsgID}
-	// 	}
-	// }
-	if msg.OfflinePushInfo != nil {
-		opts.IOSBadgeCount = msg.OfflinePushInfo.IOSBadgeCount
-		opts.IOSPushSound = msg.OfflinePushInfo.IOSPushSound
-		opts.Ex = msg.OfflinePushInfo.Ex
-	}
-	return opts, nil
-}
-
-func (p *Pusher) getOfflinePushInfos(conversationID string, msg *sdkws.MsgData) (title, content string, opts *offlinepush.Opts, err error) {
-	if p.offlinePusher == nil {
-		err = errNoOfflinePusher
-		return
-	}
-
-	type atContent struct {
-		Text       string   `json:"text"`
-		AtUserList []string `json:"atUserList"`
-		IsAtSelf   bool     `json:"isAtSelf"`
-	}
-
-	opts, err = p.GetOfflinePushOpts(msg)
-	if err != nil {
-		return
-	}
-
-	if msg.OfflinePushInfo != nil {
-		title = msg.OfflinePushInfo.Title
-		content = msg.OfflinePushInfo.Desc
-	}
-	if title == "" {
-		switch msg.ContentType {
-		case constant.Text:
-			fallthrough
-		case constant.Picture:
-			fallthrough
-		case constant.Voice:
-			fallthrough
-		case constant.Video:
-			fallthrough
-		case constant.File:
-			title = constant.ContentType2PushContent[int64(msg.ContentType)]
-		case constant.AtText:
-			ac := atContent{}
-			_ = utils.JsonStringToStruct(string(msg.Content), &ac)
-			if utils.IsContain(conversationID, ac.AtUserList) {
-				title = constant.ContentType2PushContent[constant.AtText] + constant.ContentType2PushContent[constant.Common]
-			} else {
-				title = constant.ContentType2PushContent[constant.GroupMsg]
-			}
-		case constant.SignalingNotification:
-			title = constant.ContentType2PushContent[constant.SignalMsg]
-		default:
-			title = constant.ContentType2PushContent[constant.Common]
-		}
-	}
-	if content == "" {
-		content = title
-	}
-	return
-}

internal/push/tools.go

@@ -1,32 +0,0 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package push
-
-import (
-	"github.com/OpenIMSDK/protocol/constant"
-	"github.com/OpenIMSDK/protocol/sdkws"
-	"google.golang.org/protobuf/proto"
-)
-
-func GetContent(msg *sdkws.MsgData) string {
-	if msg.ContentType >= constant.NotificationBegin && msg.ContentType <= constant.NotificationEnd {
-		var tips sdkws.TipsComm
-		_ = proto.Unmarshal(msg.Content, &tips)
-		content := tips.JsonDetail
-		return content
-	} else {
-		return string(msg.Content)
-	}
-}

internal/rpc/conversation/conversaion.go

@@ -17,9 +17,10 @@ package conversation
 import (
 	"context"
 	"errors"
-	"github.com/OpenIMSDK/protocol/sdkws"
 	"sort"
 
+	"github.com/OpenIMSDK/protocol/sdkws"
+
 	"github.com/OpenIMSDK/tools/tx"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/mgo"

internal/rpc/friend/friend.go

@@ -16,7 +16,6 @@ package friend
 
 import (
 	"context"
-
 	"github.com/OpenIMSDK/tools/tx"
 
 	"github.com/OpenIMSDK/protocol/sdkws"

internal/rpc/msg/as_read.go

@@ -18,6 +18,7 @@ import (
 	"context"
 
 	utils2 "github.com/OpenIMSDK/tools/utils"
+
 	cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
 
 	"github.com/redis/go-redis/v9"
@@ -43,7 +44,7 @@ func (m *msgServer) GetConversationsHasReadAndMaxSeq(ctx context.Context, req *m
 	if err != nil {
 		return nil, err
 	}
-	conversations, err := m.Conversation.GetConversations(ctx, req.UserID, conversationIDs)
+	conversations, err := m.ConversationLocalCache.GetConversations(ctx, req.UserID, conversationIDs)
 	if err != nil {
 		return nil, err
 	}
@@ -106,7 +107,7 @@ func (m *msgServer) MarkMsgsAsRead(
 	if hasReadSeq > maxSeq {
 		return nil, errs.ErrArgs.Wrap("hasReadSeq must not be bigger than maxSeq")
 	}
-	conversation, err := m.Conversation.GetConversation(ctx, req.UserID, req.ConversationID)
+	conversation, err := m.ConversationLocalCache.GetConversation(ctx, req.UserID, req.ConversationID)
 	if err != nil {
 		return
 	}
@@ -146,7 +147,7 @@ func (m *msgServer) MarkConversationAsRead(
 	ctx context.Context,
 	req *msg.MarkConversationAsReadReq,
 ) (resp *msg.MarkConversationAsReadResp, err error) {
-	conversation, err := m.Conversation.GetConversation(ctx, req.UserID, req.ConversationID)
+	conversation, err := m.ConversationLocalCache.GetConversation(ctx, req.UserID, req.ConversationID)
 	if err != nil {
 		return nil, err
 	}

internal/rpc/msg/revoke.go

@@ -47,7 +47,7 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
 	if err := authverify.CheckAccessV3(ctx, req.UserID); err != nil {
 		return nil, err
 	}
-	user, err := m.User.GetUserInfo(ctx, req.UserID)
+	user, err := m.UserLocalCache.GetUserInfo(ctx, req.UserID)
 	if err != nil {
 		return nil, err
 	}
@@ -73,12 +73,7 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
 			}
 			role = user.AppMangerLevel
 		case constant.SuperGroupChatType:
-			members, err := m.Group.GetGroupMemberInfoMap(
+			members, err := m.GroupLocalCache.GetGroupMemberInfoMap(ctx, msgs[0].GroupID, utils.Distinct([]string{req.UserID, msgs[0].SendID}))
-				ctx,
-				msgs[0].GroupID,
-				utils.Distinct([]string{req.UserID, msgs[0].SendID}),
-				true,
-			)
 			if err != nil {
 				return nil, err
 			}

internal/rpc/msg/send.go

@@ -98,7 +98,7 @@ func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgDa
 	}
 	tagAll := utils.IsContain(constant.AtAllString, msg.AtUserIDList)
 	if tagAll {
-		memberUserIDList, err := m.Group.GetGroupMemberIDs(ctx, msg.GroupID)
+		memberUserIDList, err := m.GroupLocalCache.GetGroupMemberIDs(ctx, msg.GroupID)
 		if err != nil {
 			log.ZWarn(ctx, "GetGroupMemberIDs", err)
 			return
@@ -144,6 +144,7 @@ func (m *msgServer) sendMsgNotification(
 }
 
 func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq) (resp *pbmsg.SendMsgResp, err error) {
+	log.ZDebug(ctx, "sendMsgSingleChat return")
 	if err := m.messageVerification(ctx, req); err != nil {
 		return nil, err
 	}

internal/rpc/msg/seq.go

@@ -16,6 +16,7 @@ package msg
 
 import (
 	"context"
+
 	pbmsg "github.com/OpenIMSDK/protocol/msg"
 )
 

internal/rpc/msg/server.go

@@ -16,6 +16,7 @@ package msg
 
 import (
 	"context"
+	"github.com/openimsdk/open-im-server/v3/pkg/rpccache"
 
 	"google.golang.org/grpc"
 
@@ -26,7 +27,6 @@ import (
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/controller"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/db/localcache"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/unrelation"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 )
@@ -36,12 +36,11 @@ type (
 	msgServer               struct {
 		RegisterCenter         discoveryregistry.SvcDiscoveryRegistry
 		MsgDatabase            controller.CommonMsgDatabase
-		Group                  *rpcclient.GroupRpcClient
-		User                   *rpcclient.UserRpcClient
 		Conversation           *rpcclient.ConversationRpcClient
-		friend                 *rpcclient.FriendRpcClient
+		UserLocalCache         *rpccache.UserLocalCache
-		GroupLocalCache        *localcache.GroupLocalCache
+		FriendLocalCache       *rpccache.FriendLocalCache
-		ConversationLocalCache *localcache.ConversationLocalCache
+		GroupLocalCache        *rpccache.GroupLocalCache
+		ConversationLocalCache *rpccache.ConversationLocalCache
 		Handlers               MessageInterceptorChain
 		notificationSender     *rpcclient.NotificationSender
 	}
@@ -81,15 +80,15 @@ func Start(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) e
 	groupRpcClient := rpcclient.NewGroupRpcClient(client)
 	friendRpcClient := rpcclient.NewFriendRpcClient(client)
 	msgDatabase := controller.NewCommonMsgDatabase(msgDocModel, cacheModel)
+
 	s := &msgServer{
 		Conversation:           &conversationClient,
-		User:                   &userRpcClient,
-		Group:                  &groupRpcClient,
 		MsgDatabase:            msgDatabase,
 		RegisterCenter:         client,
-		GroupLocalCache:        localcache.NewGroupLocalCache(&groupRpcClient),
+		UserLocalCache:         rpccache.NewUserLocalCache(userRpcClient, rdb),
-		ConversationLocalCache: localcache.NewConversationLocalCache(&conversationClient),
+		GroupLocalCache:        rpccache.NewGroupLocalCache(groupRpcClient, rdb),
-		friend:                 &friendRpcClient,
+		ConversationLocalCache: rpccache.NewConversationLocalCache(conversationClient, rdb),
+		FriendLocalCache:       rpccache.NewFriendLocalCache(friendRpcClient, rdb),
 	}
 	s.notificationSender = rpcclient.NewNotificationSender(rpcclient.WithLocalSendMsg(s.SendMsg))
 	s.addInterceptorHandler(MessageHasReadEnabled)

internal/rpc/msg/statistics.go

@@ -41,7 +41,7 @@ func (m *msgServer) GetActiveUser(ctx context.Context, req *msg.GetActiveUserReq
 	var pbUsers []*msg.ActiveUser
 	if len(users) > 0 {
 		userIDs := utils.Slice(users, func(e *unrelation.UserCount) string { return e.UserID })
-		userMap, err := m.User.GetUsersInfoMap(ctx, userIDs)
+		userMap, err := m.UserLocalCache.GetUsersInfoMap(ctx, userIDs)
 		if err != nil {
 			return nil, err
 		}
@@ -83,7 +83,7 @@ func (m *msgServer) GetActiveGroup(ctx context.Context, req *msg.GetActiveGroupR
 	var pbgroups []*msg.ActiveGroup
 	if len(groups) > 0 {
 		groupIDs := utils.Slice(groups, func(e *unrelation.GroupCount) string { return e.GroupID })
-		resp, err := m.Group.GetGroupInfos(ctx, groupIDs, false)
+		resp, err := m.GroupLocalCache.GetGroupInfos(ctx, groupIDs)
 		if err != nil {
 			return nil, err
 		}

internal/rpc/msg/sync_msg.go

@@ -37,7 +37,7 @@ func (m *msgServer) PullMessageBySeqs(
 	resp.NotificationMsgs = make(map[string]*sdkws.PullMsgs)
 	for _, seq := range req.SeqRanges {
 		if !msgprocessor.IsNotification(seq.ConversationID) {
-			conversation, err := m.Conversation.GetConversation(ctx, req.UserID, seq.ConversationID)
+			conversation, err := m.ConversationLocalCache.GetConversation(ctx, req.UserID, seq.ConversationID)
 			if err != nil {
 				log.ZError(ctx, "GetConversation error", err, "conversationID", seq.ConversationID)
 				continue
@@ -140,7 +140,7 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq
 		}
 	}
 	if len(sendIDs) != 0 {
-		sendInfos, err := m.User.GetUsersInfo(ctx, sendIDs)
+		sendInfos, err := m.UserLocalCache.GetUsersInfo(ctx, sendIDs)
 		if err != nil {
 			return nil, err
 		}
@@ -149,7 +149,7 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq
 		}
 	}
 	if len(recvIDs) != 0 {
-		recvInfos, err := m.User.GetUsersInfo(ctx, recvIDs)
+		recvInfos, err := m.UserLocalCache.GetUsersInfo(ctx, recvIDs)
 		if err != nil {
 			return nil, err
 		}
@@ -158,7 +158,7 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq
 		}
 	}
 	if len(groupIDs) != 0 {
-		groupInfos, err := m.Group.GetGroupInfos(ctx, groupIDs, true)
+		groupInfos, err := m.GroupLocalCache.GetGroupInfos(ctx, groupIDs)
 		if err != nil {
 			return nil, err
 		}

internal/rpc/msg/verify.go

@@ -16,6 +16,7 @@ package msg
 
 import (
 	"context"
+	"github.com/OpenIMSDK/tools/log"
 	"math/rand"
 	"strconv"
 	"time"
@@ -58,7 +59,7 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
 			data.MsgData.ContentType >= constant.NotificationBegin {
 			return nil
 		}
-		black, err := m.friend.IsBlocked(ctx, data.MsgData.SendID, data.MsgData.RecvID)
+		black, err := m.FriendLocalCache.IsBlack(ctx, data.MsgData.SendID, data.MsgData.RecvID)
 		if err != nil {
 			return err
 		}
@@ -66,7 +67,7 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
 			return errs.ErrBlockedByPeer.Wrap()
 		}
 		if *config.Config.MessageVerify.FriendVerify {
-			friend, err := m.friend.IsFriend(ctx, data.MsgData.SendID, data.MsgData.RecvID)
+			friend, err := m.FriendLocalCache.IsFriend(ctx, data.MsgData.SendID, data.MsgData.RecvID)
 			if err != nil {
 				return err
 			}
@@ -77,7 +78,7 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
 		}
 		return nil
 	case constant.SuperGroupChatType:
-		groupInfo, err := m.Group.GetGroupInfoCache(ctx, data.MsgData.GroupID)
+		groupInfo, err := m.GroupLocalCache.GetGroupInfo(ctx, data.MsgData.GroupID)
 		if err != nil {
 			return err
 		}
@@ -95,17 +96,17 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
 			data.MsgData.ContentType >= constant.NotificationBegin {
 			return nil
 		}
-		// memberIDs, err := m.GroupLocalCache.GetGroupMemberIDs(ctx, data.MsgData.GroupID)
+		memberIDs, err := m.GroupLocalCache.GetGroupMemberIDMap(ctx, data.MsgData.GroupID)
-		// if err != nil {
-		// 	return err
-		// }
-		// if !utils.IsContain(data.MsgData.SendID, memberIDs) {
-		// 	return errs.ErrNotInGroupYet.Wrap()
-		// }
-
-		groupMemberInfo, err := m.Group.GetGroupMemberCache(ctx, data.MsgData.GroupID, data.MsgData.SendID)
 		if err != nil {
-			if err == errs.ErrRecordNotFound {
+			return err
+		}
+		if _, ok := memberIDs[data.MsgData.SendID]; !ok {
+			return errs.ErrNotInGroupYet.Wrap()
+		}
+
+		groupMemberInfo, err := m.GroupLocalCache.GetGroupMember(ctx, data.MsgData.GroupID, data.MsgData.SendID)
+		if err != nil {
+			if errs.ErrRecordNotFound.Is(err) {
 				return errs.ErrNotInGroupYet.Wrap(err.Error())
 			}
 			return err
@@ -186,7 +187,8 @@ func (m *msgServer) modifyMessageByUserMessageReceiveOpt(
 	sessionType int,
 	pb *msg.SendMsgReq,
 ) (bool, error) {
-	opt, err := m.User.GetUserGlobalMsgRecvOpt(ctx, userID)
+	defer log.ZDebug(ctx, "modifyMessageByUserMessageReceiveOpt return")
+	opt, err := m.UserLocalCache.GetUserGlobalMsgRecvOpt(ctx, userID)
 	if err != nil {
 		return false, err
 	}
@@ -202,7 +204,7 @@ func (m *msgServer) modifyMessageByUserMessageReceiveOpt(
 		return true, nil
 	}
 	// conversationID := utils.GetConversationIDBySessionType(conversationID, sessionType)
-	singleOpt, err := m.Conversation.GetSingleConversationRecvMsgOpt(ctx, userID, conversationID)
+	singleOpt, err := m.ConversationLocalCache.GetSingleConversationRecvMsgOpt(ctx, userID, conversationID)
 	if errs.ErrRecordNotFound.Is(err) {
 		return true, nil
 	} else if err != nil {

internal/rpc/third/s3.go

@@ -19,12 +19,14 @@ import (
 	"encoding/base64"
 	"encoding/hex"
 	"encoding/json"
-	"github.com/google/uuid"
-	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
 	"path"
 	"strconv"
 	"time"
 
+	"github.com/google/uuid"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/s3"
 
 	"github.com/OpenIMSDK/protocol/third"

internal/rpc/user/callback.go

@@ -16,6 +16,7 @@ package user
 
 import (
 	"context"
+
 	pbuser "github.com/OpenIMSDK/protocol/user"
 	"github.com/OpenIMSDK/tools/utils"
 

internal/rpc/user/user.go

@@ -17,12 +17,14 @@ package user
 import (
 	"context"
 	"errors"
-	"github.com/OpenIMSDK/tools/pagination"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
 	"math/rand"
 	"strings"
 	"time"
 
+	"github.com/OpenIMSDK/tools/pagination"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
+
 	"github.com/OpenIMSDK/tools/tx"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/mgo"
@@ -59,6 +61,7 @@ type userServer struct {
 	RegisterCenter           registry.SvcDiscoveryRegistry
 }
 
+
 func Start(client registry.SvcDiscoveryRegistry, server *grpc.Server) error {
 	rdb, err := cache.NewRedis()
 	if err != nil {
@@ -241,6 +244,7 @@ func (s *userServer) GetPaginationUsers(ctx context.Context, req *pbuser.GetPagi
 			return nil, err
 		}
 		return &pbuser.GetPaginationUsersResp{Total: int32(total), Users: convert.UsersDB2Pb(users)}, err
+
 	}
 
 }
@@ -387,7 +391,7 @@ func (s *userServer) GetSubscribeUsersStatus(ctx context.Context,
 	return &pbuser.GetSubscribeUsersStatusResp{StatusList: onlineStatusList}, nil
 }
 
-// ProcessUserCommandAdd user general function add
+// ProcessUserCommandAdd user general function add.
 func (s *userServer) ProcessUserCommandAdd(ctx context.Context, req *pbuser.ProcessUserCommandAddReq) (*pbuser.ProcessUserCommandAddResp, error) {
 	err := authverify.CheckAccessV3(ctx, req.UserID)
 	if err != nil {
@@ -418,7 +422,7 @@ func (s *userServer) ProcessUserCommandAdd(ctx context.Context, req *pbuser.Proc
 	return &pbuser.ProcessUserCommandAddResp{}, nil
 }
 
-// ProcessUserCommandDelete user general function delete
+// ProcessUserCommandDelete user general function delete.
 func (s *userServer) ProcessUserCommandDelete(ctx context.Context, req *pbuser.ProcessUserCommandDeleteReq) (*pbuser.ProcessUserCommandDeleteResp, error) {
 	err := authverify.CheckAccessV3(ctx, req.UserID)
 	if err != nil {
@@ -437,10 +441,11 @@ func (s *userServer) ProcessUserCommandDelete(ctx context.Context, req *pbuser.P
 	if err != nil {
 		return nil, err
 	}
+
 	return &pbuser.ProcessUserCommandDeleteResp{}, nil
 }
 
-// ProcessUserCommandUpdate user general function update
+// ProcessUserCommandUpdate user general function update.
 func (s *userServer) ProcessUserCommandUpdate(ctx context.Context, req *pbuser.ProcessUserCommandUpdateReq) (*pbuser.ProcessUserCommandUpdateResp, error) {
 	err := authverify.CheckAccessV3(ctx, req.UserID)
 	if err != nil {
@@ -473,6 +478,7 @@ func (s *userServer) ProcessUserCommandUpdate(ctx context.Context, req *pbuser.P
 }
 
 func (s *userServer) ProcessUserCommandGet(ctx context.Context, req *pbuser.ProcessUserCommandGetReq) (*pbuser.ProcessUserCommandGetResp, error) {
+
 	err := authverify.CheckAccessV3(ctx, req.UserID)
 	if err != nil {
 		return nil, err

internal/tools/msg.go

@@ -79,7 +79,7 @@ func InitMsgTool() (*MsgTool, error) {
 	if err != nil {
 		return nil, err
 	}
-	discov.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	discov.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
 	userDB, err := mgo.NewUserMongo(mongo.GetDatabase())
 	if err != nil {
 		return nil, err

pkg/common/cachekey/black.go

@@ -0,0 +1,15 @@
+package cachekey
+
+const (
+	BlackIDsKey = "BLACK_IDS:"
+	IsBlackKey  = "IS_BLACK:" // local cache
+)
+
+func GetBlackIDsKey(ownerUserID string) string {
+	return BlackIDsKey + ownerUserID
+
+}
+
+func GetIsBlackIDsKey(possibleBlackUserID, userID string) string {
+	return IsBlackKey + userID + "-" + possibleBlackUserID
+}

pkg/common/cachekey/conversation.go

@@ -0,0 +1,44 @@
+package cachekey
+
+const (
+	ConversationKey                          = "CONVERSATION:"
+	ConversationIDsKey                       = "CONVERSATION_IDS:"
+	ConversationIDsHashKey                   = "CONVERSATION_IDS_HASH:"
+	ConversationHasReadSeqKey                = "CONVERSATION_HAS_READ_SEQ:"
+	RecvMsgOptKey                            = "RECV_MSG_OPT:"
+	SuperGroupRecvMsgNotNotifyUserIDsKey     = "SUPER_GROUP_RECV_MSG_NOT_NOTIFY_USER_IDS:"
+	SuperGroupRecvMsgNotNotifyUserIDsHashKey = "SUPER_GROUP_RECV_MSG_NOT_NOTIFY_USER_IDS_HASH:"
+	ConversationNotReceiveMessageUserIDsKey  = "CONVERSATION_NOT_RECEIVE_MESSAGE_USER_IDS:"
+)
+
+func GetConversationKey(ownerUserID, conversationID string) string {
+	return ConversationKey + ownerUserID + ":" + conversationID
+}
+
+func GetConversationIDsKey(ownerUserID string) string {
+	return ConversationIDsKey + ownerUserID
+}
+
+func GetSuperGroupRecvNotNotifyUserIDsKey(groupID string) string {
+	return SuperGroupRecvMsgNotNotifyUserIDsKey + groupID
+}
+
+func GetRecvMsgOptKey(ownerUserID, conversationID string) string {
+	return RecvMsgOptKey + ownerUserID + ":" + conversationID
+}
+
+func GetSuperGroupRecvNotNotifyUserIDsHashKey(groupID string) string {
+	return SuperGroupRecvMsgNotNotifyUserIDsHashKey + groupID
+}
+
+func GetConversationHasReadSeqKey(ownerUserID, conversationID string) string {
+	return ConversationHasReadSeqKey + ownerUserID + ":" + conversationID
+}
+
+func GetConversationNotReceiveMessageUserIDsKey(conversationID string) string {
+	return ConversationNotReceiveMessageUserIDsKey + conversationID
+}
+
+func GetUserConversationIDsHashKey(ownerUserID string) string {
+	return ConversationIDsHashKey + ownerUserID
+}

pkg/common/cachekey/friend.go

@@ -0,0 +1,24 @@
+package cachekey
+
+const (
+	FriendIDsKey        = "FRIEND_IDS:"
+	TwoWayFriendsIDsKey = "COMMON_FRIENDS_IDS:"
+	FriendKey           = "FRIEND_INFO:"
+	IsFriendKey         = "IS_FRIEND:" // local cache key
+)
+
+func GetFriendIDsKey(ownerUserID string) string {
+	return FriendIDsKey + ownerUserID
+}
+
+func GetTwoWayFriendsIDsKey(ownerUserID string) string {
+	return TwoWayFriendsIDsKey + ownerUserID
+}
+
+func GetFriendKey(ownerUserID, friendUserID string) string {
+	return FriendKey + ownerUserID + "-" + friendUserID
+}
+
+func GetIsFriendKey(possibleFriendUserID, userID string) string {
+	return IsFriendKey + possibleFriendUserID + "-" + userID
+}

pkg/common/cachekey/group.go

@@ -0,0 +1,45 @@
+package cachekey
+
+import (
+	"strconv"
+	"time"
+)
+
+const (
+	groupExpireTime            = time.Second * 60 * 60 * 12
+	GroupInfoKey               = "GROUP_INFO:"
+	GroupMemberIDsKey          = "GROUP_MEMBER_IDS:"
+	GroupMembersHashKey        = "GROUP_MEMBERS_HASH2:"
+	GroupMemberInfoKey         = "GROUP_MEMBER_INFO:"
+	JoinedGroupsKey            = "JOIN_GROUPS_KEY:"
+	GroupMemberNumKey          = "GROUP_MEMBER_NUM_CACHE:"
+	GroupRoleLevelMemberIDsKey = "GROUP_ROLE_LEVEL_MEMBER_IDS:"
+)
+
+func GetGroupInfoKey(groupID string) string {
+	return GroupInfoKey + groupID
+}
+
+func GetJoinedGroupsKey(userID string) string {
+	return JoinedGroupsKey + userID
+}
+
+func GetGroupMembersHashKey(groupID string) string {
+	return GroupMembersHashKey + groupID
+}
+
+func GetGroupMemberIDsKey(groupID string) string {
+	return GroupMemberIDsKey + groupID
+}
+
+func GetGroupMemberInfoKey(groupID, userID string) string {
+	return GroupMemberInfoKey + groupID + "-" + userID
+}
+
+func GetGroupMemberNumKey(groupID string) string {
+	return GroupMemberNumKey + groupID
+}
+
+func GetGroupRoleLevelMemberIDsKey(groupID string, roleLevel int32) string {
+	return GroupRoleLevelMemberIDsKey + groupID + "-" + strconv.Itoa(int(roleLevel))
+}

pkg/common/cachekey/user.go

@@ -0,0 +1,14 @@
+package cachekey
+
+const (
+	UserInfoKey             = "USER_INFO:"
+	UserGlobalRecvMsgOptKey = "USER_GLOBAL_RECV_MSG_OPT_KEY:"
+)
+
+func GetUserInfoKey(userID string) string {
+	return UserInfoKey + userID
+}
+
+func GetUserGlobalRecvMsgOptKey(userID string) string {
+	return UserGlobalRecvMsgOptKey + userID
+}

pkg/common/config/config.go

@@ -16,6 +16,7 @@ package config
 
 import (
 	"bytes"
+	"time"
 
 	"github.com/OpenIMSDK/tools/discoveryregistry"
 	"gopkg.in/yaml.v3"
@@ -258,6 +259,8 @@ type configStruct struct {
 		FriendVerify *bool `yaml:"friendVerify"`
 	} `yaml:"messageVerify"`
 
+	LocalCache localCache `yaml:"localCache"`
+
 	IOSPush struct {
 		PushSound  string `yaml:"pushSound"`
 		BadgeCount bool   `yaml:"badgeCount"`
@@ -370,6 +373,33 @@ type notification struct {
 	ConversationSetPrivate NotificationConf `yaml:"conversationSetPrivate"`
 }
 
+type LocalCache struct {
+	Topic         string `yaml:"topic"`
+	SlotNum       int    `yaml:"slotNum"`
+	SlotSize      int    `yaml:"slotSize"`
+	SuccessExpire int    `yaml:"successExpire"` // second
+	FailedExpire  int    `yaml:"failedExpire"`  // second
+}
+
+func (l LocalCache) Failed() time.Duration {
+	return time.Second * time.Duration(l.FailedExpire)
+}
+
+func (l LocalCache) Success() time.Duration {
+	return time.Second * time.Duration(l.SuccessExpire)
+}
+
+func (l LocalCache) Enable() bool {
+	return l.Topic != "" && l.SlotNum > 0 && l.SlotSize > 0
+}
+
+type localCache struct {
+	User         LocalCache `yaml:"user"`
+	Group        LocalCache `yaml:"group"`
+	Friend       LocalCache `yaml:"friend"`
+	Conversation LocalCache `yaml:"conversation"`
+}
+
 func (c *configStruct) GetServiceNames() []string {
 	return []string{
 		c.RpcRegisterName.OpenImUserName,

pkg/common/config/parse_test.go

@@ -16,6 +16,8 @@ package config
 
 import (
 	_ "embed"
+	"fmt"
+	"gopkg.in/yaml.v3"
 	"reflect"
 	"testing"
 
@@ -115,3 +117,13 @@ func TestInitConfig(t *testing.T) {
 		})
 	}
 }
+
+func TestName(t *testing.T) {
+	Config.LocalCache.Friend.Topic = "friend"
+	Config.LocalCache.Friend.SlotNum = 500
+	Config.LocalCache.Friend.SlotSize = 20000
+
+	data, _ := yaml.Marshal(&Config)
+
+	fmt.Println(string(data))
+}

pkg/common/convert/friend.go

@@ -17,6 +17,7 @@ package convert
 import (
 	"context"
 	"fmt"
+
 	"github.com/OpenIMSDK/protocol/sdkws"
 	"github.com/OpenIMSDK/tools/utils"
 

pkg/common/db/cache/black.go

@@ -16,6 +16,9 @@ package cache
 
 import (
 	"context"
+	"github.com/OpenIMSDK/tools/log"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"time"
 
 	"github.com/dtm-labs/rockscache"
@@ -52,11 +55,15 @@ func NewBlackCacheRedis(
 	options rockscache.Options,
 ) BlackCache {
 	rcClient := rockscache.NewClient(rdb, options)
-
+	mc := NewMetaCacheRedis(rcClient)
+	b := config.Config.LocalCache.Friend
+	log.ZDebug(context.Background(), "black local cache init", "Topic", b.Topic, "SlotNum", b.SlotNum, "SlotSize", b.SlotSize, "enable", b.Enable())
+	mc.SetTopic(b.Topic)
+	mc.SetRawRedisClient(rdb)
 	return &BlackCacheRedis{
 		expireTime: blackExpireTime,
 		rcClient:   rcClient,
-		metaCache:  NewMetaCacheRedis(rcClient),
+		metaCache:  mc,
 		blackDB:    blackDB,
 	}
 }
@@ -66,12 +73,12 @@ func (b *BlackCacheRedis) NewCache() BlackCache {
 		expireTime: b.expireTime,
 		rcClient:   b.rcClient,
 		blackDB:    b.blackDB,
-		metaCache:  NewMetaCacheRedis(b.rcClient, b.metaCache.GetPreDelKeys()...),
+		metaCache:  b.Copy(),
 	}
 }
 
 func (b *BlackCacheRedis) getBlackIDsKey(ownerUserID string) string {
-	return blackIDsKey + ownerUserID
+	return cachekey.GetBlackIDsKey(ownerUserID)
 }
 
 func (b *BlackCacheRedis) GetBlackIDs(ctx context.Context, userID string) (blackIDs []string, err error) {

pkg/common/db/cache/config.go

@@ -0,0 +1,66 @@
+package cache
+
+import (
+	"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"strings"
+	"sync"
+)
+
+var (
+	once      sync.Once
+	subscribe map[string][]string
+)
+
+func getPublishKey(topic string, key []string) []string {
+	if topic == "" || len(key) == 0 {
+		return nil
+	}
+	once.Do(func() {
+		list := []struct {
+			Local config.LocalCache
+			Keys  []string
+		}{
+			{
+				Local: config.Config.LocalCache.User,
+				Keys:  []string{cachekey.UserInfoKey, cachekey.UserGlobalRecvMsgOptKey},
+			},
+			{
+				Local: config.Config.LocalCache.Group,
+				Keys:  []string{cachekey.GroupMemberIDsKey, cachekey.GroupInfoKey, cachekey.GroupMemberInfoKey},
+			},
+			{
+				Local: config.Config.LocalCache.Friend,
+				Keys:  []string{cachekey.FriendIDsKey, cachekey.BlackIDsKey},
+			},
+			{
+				Local: config.Config.LocalCache.Conversation,
+				Keys:  []string{cachekey.ConversationIDsKey, cachekey.ConversationKey},
+			},
+		}
+		subscribe = make(map[string][]string)
+		for _, v := range list {
+			if v.Local.Enable() {
+				subscribe[v.Local.Topic] = v.Keys
+			}
+		}
+	})
+	prefix, ok := subscribe[topic]
+	if !ok {
+		return nil
+	}
+	res := make([]string, 0, len(key))
+	for _, k := range key {
+		var exist bool
+		for _, p := range prefix {
+			if strings.HasPrefix(k, p) {
+				exist = true
+				break
+			}
+		}
+		if exist {
+			res = append(res, k)
+		}
+	}
+	return res
+}

pkg/common/db/cache/conversation.go

@@ -17,6 +17,9 @@ package cache
 import (
 	"context"
 	"errors"
+	"github.com/OpenIMSDK/tools/log"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"math/big"
 	"strings"
 	"time"
@@ -30,14 +33,14 @@ import (
 )
 
 const (
-	conversationKey                          = "CONVERSATION:"
+	//conversationKey                          = "CONVERSATION:"
-	conversationIDsKey                       = "CONVERSATION_IDS:"
+	//conversationIDsKey                       = "CONVERSATION_IDS:"
-	conversationIDsHashKey                   = "CONVERSATION_IDS_HASH:"
+	//conversationIDsHashKey                   = "CONVERSATION_IDS_HASH:"
-	conversationHasReadSeqKey                = "CONVERSATION_HAS_READ_SEQ:"
+	//conversationHasReadSeqKey                = "CONVERSATION_HAS_READ_SEQ:"
-	recvMsgOptKey                            = "RECV_MSG_OPT:"
+	//recvMsgOptKey                            = "RECV_MSG_OPT:"
-	superGroupRecvMsgNotNotifyUserIDsKey     = "SUPER_GROUP_RECV_MSG_NOT_NOTIFY_USER_IDS:"
+	//superGroupRecvMsgNotNotifyUserIDsKey     = "SUPER_GROUP_RECV_MSG_NOT_NOTIFY_USER_IDS:"
-	superGroupRecvMsgNotNotifyUserIDsHashKey = "SUPER_GROUP_RECV_MSG_NOT_NOTIFY_USER_IDS_HASH:"
+	//superGroupRecvMsgNotNotifyUserIDsHashKey = "SUPER_GROUP_RECV_MSG_NOT_NOTIFY_USER_IDS_HASH:"
-	conversationNotReceiveMessageUserIDsKey  = "CONVERSATION_NOT_RECEIVE_MESSAGE_USER_IDS:"
+	//conversationNotReceiveMessageUserIDsKey  = "CONVERSATION_NOT_RECEIVE_MESSAGE_USER_IDS:"
 
 	conversationExpireTime = time.Second * 60 * 60 * 12
 )
@@ -84,10 +87,14 @@ type ConversationCache interface {
 
 func NewConversationRedis(rdb redis.UniversalClient, opts rockscache.Options, db relationtb.ConversationModelInterface) ConversationCache {
 	rcClient := rockscache.NewClient(rdb, opts)
-
+	mc := NewMetaCacheRedis(rcClient)
+	c := config.Config.LocalCache.Conversation
+	log.ZDebug(context.Background(), "black local cache init", "Topic", c.Topic, "SlotNum", c.SlotNum, "SlotSize", c.SlotSize, "enable", c.Enable())
+	mc.SetTopic(c.Topic)
+	mc.SetRawRedisClient(rdb)
 	return &ConversationRedisCache{
 		rcClient:       rcClient,
-		metaCache:      NewMetaCacheRedis(rcClient),
+		metaCache:      mc,
 		conversationDB: db,
 		expireTime:     conversationExpireTime,
 	}
@@ -118,38 +125,42 @@ type ConversationRedisCache struct {
 func (c *ConversationRedisCache) NewCache() ConversationCache {
 	return &ConversationRedisCache{
 		rcClient:       c.rcClient,
-		metaCache:      NewMetaCacheRedis(c.rcClient, c.metaCache.GetPreDelKeys()...),
+		metaCache:      c.Copy(),
 		conversationDB: c.conversationDB,
 		expireTime:     c.expireTime,
 	}
 }
 
 func (c *ConversationRedisCache) getConversationKey(ownerUserID, conversationID string) string {
-	return conversationKey + ownerUserID + ":" + conversationID
+	return cachekey.GetConversationKey(ownerUserID, conversationID)
 }
 
 func (c *ConversationRedisCache) getConversationIDsKey(ownerUserID string) string {
-	return conversationIDsKey + ownerUserID
+	return cachekey.GetConversationIDsKey(ownerUserID)
 }
 
 func (c *ConversationRedisCache) getSuperGroupRecvNotNotifyUserIDsKey(groupID string) string {
-	return superGroupRecvMsgNotNotifyUserIDsKey + groupID
+	return cachekey.GetSuperGroupRecvNotNotifyUserIDsKey(groupID)
 }
 
 func (c *ConversationRedisCache) getRecvMsgOptKey(ownerUserID, conversationID string) string {
-	return recvMsgOptKey + ownerUserID + ":" + conversationID
+	return cachekey.GetRecvMsgOptKey(ownerUserID, conversationID)
 }
 
 func (c *ConversationRedisCache) getSuperGroupRecvNotNotifyUserIDsHashKey(groupID string) string {
-	return superGroupRecvMsgNotNotifyUserIDsHashKey + groupID
+	return cachekey.GetSuperGroupRecvNotNotifyUserIDsHashKey(groupID)
 }
 
 func (c *ConversationRedisCache) getConversationHasReadSeqKey(ownerUserID, conversationID string) string {
-	return conversationHasReadSeqKey + ownerUserID + ":" + conversationID
+	return cachekey.GetConversationHasReadSeqKey(ownerUserID, conversationID)
 }
 
 func (c *ConversationRedisCache) getConversationNotReceiveMessageUserIDsKey(conversationID string) string {
-	return conversationNotReceiveMessageUserIDsKey + conversationID
+	return cachekey.GetConversationNotReceiveMessageUserIDsKey(conversationID)
+}
+
+func (c *ConversationRedisCache) getUserConversationIDsHashKey(ownerUserID string) string {
+	return cachekey.GetUserConversationIDsHashKey(ownerUserID)
 }
 
 func (c *ConversationRedisCache) GetUserConversationIDs(ctx context.Context, ownerUserID string) ([]string, error) {
@@ -169,10 +180,6 @@ func (c *ConversationRedisCache) DelConversationIDs(userIDs ...string) Conversat
 	return cache
 }
 
-func (c *ConversationRedisCache) getUserConversationIDsHashKey(ownerUserID string) string {
-	return conversationIDsHashKey + ownerUserID
-}
-
 func (c *ConversationRedisCache) GetUserConversationIDsHash(ctx context.Context, ownerUserID string) (hash uint64, err error) {
 	return getCache(
 		ctx,

pkg/common/db/cache/friend.go

@@ -16,6 +16,9 @@ package cache
 
 import (
 	"context"
+	"github.com/OpenIMSDK/tools/log"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"time"
 
 	"github.com/dtm-labs/rockscache"
@@ -27,10 +30,10 @@ import (
 )
 
 const (
-	friendExpireTime    = time.Second * 60 * 60 * 12
+	friendExpireTime = time.Second * 60 * 60 * 12
-	friendIDsKey        = "FRIEND_IDS:"
+	//friendIDsKey        = "FRIEND_IDS:"
-	TwoWayFriendsIDsKey = "COMMON_FRIENDS_IDS:"
+	//TwoWayFriendsIDsKey = "COMMON_FRIENDS_IDS:"
-	friendKey           = "FRIEND_INFO:"
+	//friendKey           = "FRIEND_INFO:"
 )
 
 // FriendCache is an interface for caching friend-related data.
@@ -60,8 +63,13 @@ type FriendCacheRedis struct {
 func NewFriendCacheRedis(rdb redis.UniversalClient, friendDB relationtb.FriendModelInterface,
 	options rockscache.Options) FriendCache {
 	rcClient := rockscache.NewClient(rdb, options)
+	mc := NewMetaCacheRedis(rcClient)
+	f := config.Config.LocalCache.Friend
+	log.ZDebug(context.Background(), "friend local cache init", "Topic", f.Topic, "SlotNum", f.SlotNum, "SlotSize", f.SlotSize, "enable", f.Enable())
+	mc.SetTopic(f.Topic)
+	mc.SetRawRedisClient(rdb)
 	return &FriendCacheRedis{
-		metaCache:  NewMetaCacheRedis(rcClient),
+		metaCache:  mc,
 		friendDB:   friendDB,
 		expireTime: friendExpireTime,
 		rcClient:   rcClient,
@@ -72,7 +80,7 @@ func NewFriendCacheRedis(rdb redis.UniversalClient, friendDB relationtb.FriendMo
 func (f *FriendCacheRedis) NewCache() FriendCache {
 	return &FriendCacheRedis{
 		rcClient:   f.rcClient,
-		metaCache:  NewMetaCacheRedis(f.rcClient, f.metaCache.GetPreDelKeys()...),
+		metaCache:  f.Copy(),
 		friendDB:   f.friendDB,
 		expireTime: f.expireTime,
 	}
@@ -80,17 +88,17 @@ func (f *FriendCacheRedis) NewCache() FriendCache {
 
 // getFriendIDsKey returns the key for storing friend IDs in the cache.
 func (f *FriendCacheRedis) getFriendIDsKey(ownerUserID string) string {
-	return friendIDsKey + ownerUserID
+	return cachekey.GetFriendIDsKey(ownerUserID)
 }
 
 // getTwoWayFriendsIDsKey returns the key for storing two-way friend IDs in the cache.
 func (f *FriendCacheRedis) getTwoWayFriendsIDsKey(ownerUserID string) string {
-	return TwoWayFriendsIDsKey + ownerUserID
+	return cachekey.GetTwoWayFriendsIDsKey(ownerUserID)
 }
 
 // getFriendKey returns the key for storing friend info in the cache.
 func (f *FriendCacheRedis) getFriendKey(ownerUserID, friendUserID string) string {
-	return friendKey + ownerUserID + "-" + friendUserID
+	return cachekey.GetFriendKey(ownerUserID, friendUserID)
 }
 
 // GetFriendIDs retrieves friend IDs from the cache or the database if not found.

pkg/common/db/cache/group.go

@@ -17,7 +17,8 @@ package cache
 import (
 	"context"
 	"fmt"
-	"strconv"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"time"
 
 	"github.com/OpenIMSDK/protocol/constant"
@@ -34,15 +35,14 @@ import (
 )
 
 const (
-	groupExpireTime     = time.Second * 60 * 60 * 12
+	groupExpireTime = time.Second * 60 * 60 * 12
-	groupInfoKey        = "GROUP_INFO:"
+	//groupInfoKey        = "GROUP_INFO:"
-	groupMemberIDsKey   = "GROUP_MEMBER_IDS:"
+	//groupMemberIDsKey   = "GROUP_MEMBER_IDS:"
-	groupMembersHashKey = "GROUP_MEMBERS_HASH2:"
+	//groupMembersHashKey = "GROUP_MEMBERS_HASH2:"
-	groupMemberInfoKey  = "GROUP_MEMBER_INFO:"
+	//groupMemberInfoKey  = "GROUP_MEMBER_INFO:"
-	//groupOwnerInfoKey   = "GROUP_OWNER_INFO:".
+	//joinedGroupsKey            = "JOIN_GROUPS_KEY:"
-	joinedGroupsKey            = "JOIN_GROUPS_KEY:"
+	//groupMemberNumKey          = "GROUP_MEMBER_NUM_CACHE:"
-	groupMemberNumKey          = "GROUP_MEMBER_NUM_CACHE:"
+	//groupRoleLevelMemberIDsKey = "GROUP_ROLE_LEVEL_MEMBER_IDS:"
-	groupRoleLevelMemberIDsKey = "GROUP_ROLE_LEVEL_MEMBER_IDS:"
 )
 
 type GroupHash interface {
@@ -105,12 +105,16 @@ func NewGroupCacheRedis(
 	opts rockscache.Options,
 ) GroupCache {
 	rcClient := rockscache.NewClient(rdb, opts)
-
+	mc := NewMetaCacheRedis(rcClient)
+	g := config.Config.LocalCache.Group
+	mc.SetTopic(g.Topic)
+	log.ZDebug(context.Background(), "group local cache init", "Topic", g.Topic, "SlotNum", g.SlotNum, "SlotSize", g.SlotSize, "enable", g.Enable())
+	mc.SetRawRedisClient(rdb)
 	return &GroupCacheRedis{
 		rcClient: rcClient, expireTime: groupExpireTime,
 		groupDB: groupDB, groupMemberDB: groupMemberDB, groupRequestDB: groupRequestDB,
 		groupHash: hashCode,
-		metaCache: NewMetaCacheRedis(rcClient),
+		metaCache: mc,
 	}
 }
 
@@ -121,36 +125,36 @@ func (g *GroupCacheRedis) NewCache() GroupCache {
 		groupDB:        g.groupDB,
 		groupMemberDB:  g.groupMemberDB,
 		groupRequestDB: g.groupRequestDB,
-		metaCache:      NewMetaCacheRedis(g.rcClient, g.metaCache.GetPreDelKeys()...),
+		metaCache:      g.Copy(),
 	}
 }
 
 func (g *GroupCacheRedis) getGroupInfoKey(groupID string) string {
-	return groupInfoKey + groupID
+	return cachekey.GetGroupInfoKey(groupID)
 }
 
 func (g *GroupCacheRedis) getJoinedGroupsKey(userID string) string {
-	return joinedGroupsKey + userID
+	return cachekey.GetJoinedGroupsKey(userID)
 }
 
 func (g *GroupCacheRedis) getGroupMembersHashKey(groupID string) string {
-	return groupMembersHashKey + groupID
+	return cachekey.GetGroupMembersHashKey(groupID)
 }
 
 func (g *GroupCacheRedis) getGroupMemberIDsKey(groupID string) string {
-	return groupMemberIDsKey + groupID
+	return cachekey.GetGroupMemberIDsKey(groupID)
 }
 
 func (g *GroupCacheRedis) getGroupMemberInfoKey(groupID, userID string) string {
-	return groupMemberInfoKey + groupID + "-" + userID
+	return cachekey.GetGroupMemberInfoKey(groupID, userID)
 }
 
 func (g *GroupCacheRedis) getGroupMemberNumKey(groupID string) string {
-	return groupMemberNumKey + groupID
+	return cachekey.GetGroupMemberNumKey(groupID)
 }
 
 func (g *GroupCacheRedis) getGroupRoleLevelMemberIDsKey(groupID string, roleLevel int32) string {
-	return groupRoleLevelMemberIDsKey + groupID + "-" + strconv.Itoa(int(roleLevel))
+	return cachekey.GetGroupRoleLevelMemberIDsKey(groupID, roleLevel)
 }
 
 func (g *GroupCacheRedis) GetGroupIndex(group *relationtb.GroupModel, keys []string) (int, error) {

pkg/common/db/cache/meta_cache.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"github.com/redis/go-redis/v9"
 	"time"
 
 	"github.com/OpenIMSDK/tools/mw/specialerror"
@@ -44,6 +45,9 @@ type metaCache interface {
 	AddKeys(keys ...string)
 	ClearKeys()
 	GetPreDelKeys() []string
+	SetTopic(topic string)
+	SetRawRedisClient(cli redis.UniversalClient)
+	Copy() metaCache
 }
 
 func NewMetaCacheRedis(rcClient *rockscache.Client, keys ...string) metaCache {
@@ -51,10 +55,36 @@ func NewMetaCacheRedis(rcClient *rockscache.Client, keys ...string) metaCache {
 }
 
 type metaCacheRedis struct {
+	topic         string
 	rcClient      *rockscache.Client
 	keys          []string
 	maxRetryTimes int
 	retryInterval time.Duration
+	redisClient   redis.UniversalClient
+}
+
+func (m *metaCacheRedis) Copy() metaCache {
+	var keys []string
+	if len(m.keys) > 0 {
+		keys = make([]string, 0, len(m.keys)*2)
+		keys = append(keys, m.keys...)
+	}
+	return &metaCacheRedis{
+		topic:         m.topic,
+		rcClient:      m.rcClient,
+		keys:          keys,
+		maxRetryTimes: m.maxRetryTimes,
+		retryInterval: m.retryInterval,
+		redisClient:   redisClient,
+	}
+}
+
+func (m *metaCacheRedis) SetTopic(topic string) {
+	m.topic = topic
+}
+
+func (m *metaCacheRedis) SetRawRedisClient(cli redis.UniversalClient) {
+	m.redisClient = cli
 }
 
 func (m *metaCacheRedis) ExecDel(ctx context.Context, distinct ...bool) error {
@@ -62,7 +92,7 @@ func (m *metaCacheRedis) ExecDel(ctx context.Context, distinct ...bool) error {
 		m.keys = utils.Distinct(m.keys)
 	}
 	if len(m.keys) > 0 {
-		log.ZDebug(ctx, "delete cache", "keys", m.keys)
+		log.ZDebug(ctx, "delete cache", "topic", m.topic, "keys", m.keys)
 		for _, key := range m.keys {
 			for i := 0; i < m.maxRetryTimes; i++ {
 				if err := m.rcClient.TagAsDeleted(key); err != nil {
@@ -72,31 +102,18 @@ func (m *metaCacheRedis) ExecDel(ctx context.Context, distinct ...bool) error {
 				}
 				break
 			}
-
+		}
-			//retryTimes := 0
+		if pk := getPublishKey(m.topic, m.keys); len(pk) > 0 {
-			//for {
+			data, err := json.Marshal(pk)
-			//	m.rcClient.TagAsDeleted()
+			if err != nil {
-			//	if err := m.rcClient.TagAsDeletedBatch2(ctx, []string{key}); err != nil {
+				log.ZError(ctx, "keys json marshal failed", err, "topic", m.topic, "keys", pk)
-			//		if retryTimes >= m.maxRetryTimes {
+			} else {
-			//			err = errs.ErrInternalServer.Wrap(
+				if err := m.redisClient.Publish(ctx, m.topic, string(data)).Err(); err != nil {
-			//				fmt.Sprintf(
+					log.ZError(ctx, "redis publish cache delete error", err, "topic", m.topic, "keys", pk)
-			//					"delete cache error: %v, keys: %v, retry times %d, please check redis server",
+				}
-			//					err,
+			}
-			//					key,
-			//					retryTimes,
-			//				),
-			//			)
-			//			log.ZWarn(ctx, "delete cache failed, please handle keys", err, "keys", key)
-			//			return err
-			//		}
-			//		retryTimes++
-			//	} else {
-			//		break
-			//	}
-			//}
 		}
 	}
-
 	return nil
 }
 

pkg/common/db/cache/msg.go

@@ -17,6 +17,7 @@ package cache
 import (
 	"context"
 	"errors"
+	"github.com/dtm-labs/rockscache"
 	"strconv"
 	"time"
 
@@ -44,12 +45,12 @@ const (
 	conversationUserMinSeq = "CON_USER_MIN_SEQ:"
 	hasReadSeq             = "HAS_READ_SEQ:"
 
-	appleDeviceToken = "DEVICE_TOKEN"
+	//appleDeviceToken = "DEVICE_TOKEN"
-	getuiToken       = "GETUI_TOKEN"
+	getuiToken  = "GETUI_TOKEN"
-	getuiTaskID      = "GETUI_TASK_ID"
+	getuiTaskID = "GETUI_TASK_ID"
-	signalCache      = "SIGNAL_CACHE:"
+	//signalCache      = "SIGNAL_CACHE:"
-	signalListCache  = "SIGNAL_LIST_CACHE:"
+	//signalListCache  = "SIGNAL_LIST_CACHE:"
-	FCM_TOKEN        = "FCM_TOKEN:"
+	FCM_TOKEN = "FCM_TOKEN:"
 
 	messageCache            = "MESSAGE_CACHE:"
 	messageDelUserList      = "MESSAGE_DEL_USER_LIST:"
@@ -128,7 +129,8 @@ type MsgModel interface {
 }
 
 func NewMsgCacheModel(client redis.UniversalClient) MsgModel {
-	return &msgCache{rdb: client}
+	rcClient := rockscache.NewClient(client, rockscache.NewDefaultOptions())
+	return &msgCache{metaCache: NewMetaCacheRedis(rcClient), rdb: client}
 }
 
 type msgCache struct {
@@ -148,6 +150,10 @@ func (c *msgCache) getHasReadSeqKey(conversationID string, userID string) string
 	return hasReadSeq + userID + ":" + conversationID
 }
 
+func (c *msgCache) getConversationUserMinSeqKey(conversationID, userID string) string {
+	return conversationUserMinSeq + conversationID + "u:" + userID
+}
+
 func (c *msgCache) setSeq(ctx context.Context, conversationID string, seq int64, getkey func(conversationID string) string) error {
 	return utils.Wrap1(c.rdb.Set(ctx, getkey(conversationID), seq, 0).Err())
 }
@@ -209,10 +215,6 @@ func (c *msgCache) GetMinSeq(ctx context.Context, conversationID string) (int64,
 	return c.getSeq(ctx, conversationID, c.getMinSeqKey)
 }
 
-func (c *msgCache) getConversationUserMinSeqKey(conversationID, userID string) string {
-	return conversationUserMinSeq + conversationID + "u:" + userID
-}
-
 func (c *msgCache) GetConversationUserMinSeq(ctx context.Context, conversationID string, userID string) (int64, error) {
 	return utils.Wrap2(c.rdb.Get(ctx, c.getConversationUserMinSeqKey(conversationID, userID)).Int64())
 }

pkg/common/db/cache/user.go

@@ -18,6 +18,8 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"hash/crc32"
 	"strconv"
 	"time"
@@ -36,8 +38,8 @@ import (
 )
 
 const (
-	userExpireTime            = time.Second * 60 * 60 * 12
+	userExpireTime = time.Second * 60 * 60 * 12
-	userInfoKey               = "USER_INFO:"
+	//userInfoKey               = "USER_INFO:"
 	userGlobalRecvMsgOptKey   = "USER_GLOBAL_RECV_MSG_OPT_KEY:"
 	olineStatusKey            = "ONLINE_STATUS:"
 	userOlineStatusExpireTime = time.Second * 60 * 60 * 24
@@ -72,7 +74,11 @@ func NewUserCacheRedis(
 	options rockscache.Options,
 ) UserCache {
 	rcClient := rockscache.NewClient(rdb, options)
-
+	mc := NewMetaCacheRedis(rcClient)
+	u := config.Config.LocalCache.User
+	log.ZDebug(context.Background(), "user local cache init", "Topic", u.Topic, "SlotNum", u.SlotNum, "SlotSize", u.SlotSize, "enable", u.Enable())
+	mc.SetTopic(u.Topic)
+	mc.SetRawRedisClient(rdb)
 	return &UserCacheRedis{
 		rdb:        rdb,
 		metaCache:  NewMetaCacheRedis(rcClient),
@@ -85,7 +91,7 @@ func NewUserCacheRedis(
 func (u *UserCacheRedis) NewCache() UserCache {
 	return &UserCacheRedis{
 		rdb:        u.rdb,
-		metaCache:  NewMetaCacheRedis(u.rcClient, u.metaCache.GetPreDelKeys()...),
+		metaCache:  u.Copy(),
 		userDB:     u.userDB,
 		expireTime: u.expireTime,
 		rcClient:   u.rcClient,
@@ -93,18 +99,17 @@ func (u *UserCacheRedis) NewCache() UserCache {
 }
 
 func (u *UserCacheRedis) getUserInfoKey(userID string) string {
-	return userInfoKey + userID
+	return cachekey.GetUserInfoKey(userID)
 }
 
 func (u *UserCacheRedis) getUserGlobalRecvMsgOptKey(userID string) string {
-	return userGlobalRecvMsgOptKey + userID
+	return cachekey.GetUserGlobalRecvMsgOptKey(userID)
 }
 
 func (u *UserCacheRedis) GetUserInfo(ctx context.Context, userID string) (userInfo *relationtb.UserModel, err error) {
 	return getCache(ctx, u.rcClient, u.getUserInfoKey(userID), u.expireTime, func(ctx context.Context) (*relationtb.UserModel, error) {
 		return u.userDB.Take(ctx, userID)
-	},
+	})
-	)
 }
 
 func (u *UserCacheRedis) GetUsersInfo(ctx context.Context, userIDs []string) ([]*relationtb.UserModel, error) {

pkg/common/db/controller/user.go

@@ -142,12 +142,12 @@ func (u *userDatabase) Find(ctx context.Context, userIDs []string) (users []*rel
 	return u.cache.GetUsersInfo(ctx, userIDs)
 }
 
-// Find userInfo By Nickname
+// Find userInfo By Nickname.
 func (u *userDatabase) FindByNickname(ctx context.Context, nickname string) (users []*relation.UserModel, err error) {
 	return u.userDB.TakeByNickname(ctx, nickname)
 }
 
-// Find notificationAccouts
+// Find notificationAccouts.
 func (u *userDatabase) FindNotification(ctx context.Context, level int64) (users []*relation.UserModel, err error) {
 	return u.userDB.TakeNotification(ctx, level)
 }
@@ -190,7 +190,14 @@ func (u *userDatabase) Page(ctx context.Context, pagination pagination.Paginatio
 func (u *userDatabase) PageFindUser(ctx context.Context, level1 int64, level2 int64, pagination pagination.Pagination) (count int64, users []*relation.UserModel, err error) {
 	return u.userDB.PageFindUser(ctx, level1, level2, pagination)
 }
-func (u *userDatabase) PageFindUserWithKeyword(ctx context.Context, level1 int64, level2 int64, userID, nickName string, pagination pagination.Pagination) (count int64, users []*relation.UserModel, err error) {
+
+func (u *userDatabase) PageFindUserWithKeyword(
+	ctx context.Context,
+	level1 int64,
+	level2 int64,
+	userID, nickName string,
+	pagination pagination.Pagination,
+) (count int64, users []*relation.UserModel, err error) {
 	return u.userDB.PageFindUserWithKeyword(ctx, level1, level2, userID, nickName, pagination)
 }
 

pkg/common/db/localcache/conversation.go

@@ -1,87 +0,0 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package localcache
-
-import (
-	"context"
-	"sync"
-
-	"github.com/OpenIMSDK/protocol/conversation"
-
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
-)
-
-type ConversationLocalCache struct {
-	lock                              sync.Mutex
-	superGroupRecvMsgNotNotifyUserIDs map[string]Hash
-	conversationIDs                   map[string]Hash
-	client                            *rpcclient.ConversationRpcClient
-}
-
-type Hash struct {
-	hash uint64
-	ids  []string
-}
-
-func NewConversationLocalCache(client *rpcclient.ConversationRpcClient) *ConversationLocalCache {
-	return &ConversationLocalCache{
-		superGroupRecvMsgNotNotifyUserIDs: make(map[string]Hash),
-		conversationIDs:                   make(map[string]Hash),
-		client:                            client,
-	}
-}
-
-func (g *ConversationLocalCache) GetRecvMsgNotNotifyUserIDs(ctx context.Context, groupID string) ([]string, error) {
-	resp, err := g.client.Client.GetRecvMsgNotNotifyUserIDs(ctx, &conversation.GetRecvMsgNotNotifyUserIDsReq{
-		GroupID: groupID,
-	})
-	if err != nil {
-		return nil, err
-	}
-	return resp.UserIDs, nil
-}
-
-func (g *ConversationLocalCache) GetConversationIDs(ctx context.Context, userID string) ([]string, error) {
-	resp, err := g.client.Client.GetUserConversationIDsHash(ctx, &conversation.GetUserConversationIDsHashReq{
-		OwnerUserID: userID,
-	})
-	if err != nil {
-		return nil, err
-	}
-
-	g.lock.Lock()
-	hash, ok := g.conversationIDs[userID]
-	g.lock.Unlock()
-
-	if !ok || hash.hash != resp.Hash {
-		conversationIDsResp, err := g.client.Client.GetConversationIDs(ctx, &conversation.GetConversationIDsReq{
-			UserID: userID,
-		})
-		if err != nil {
-			return nil, err
-		}
-
-		g.lock.Lock()
-		defer g.lock.Unlock()
-		g.conversationIDs[userID] = Hash{
-			hash: resp.Hash,
-			ids:  conversationIDsResp.ConversationIDs,
-		}
-
-		return conversationIDsResp.ConversationIDs, nil
-	}
-
-	return hash.ids, nil
-}

pkg/common/db/localcache/doc.go

@@ -1,15 +0,0 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package localcache // import "github.com/openimsdk/open-im-server/v3/pkg/common/db/localcache"

pkg/common/db/localcache/group.go

@@ -1,78 +0,0 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package localcache
-
-import (
-	"context"
-	"sync"
-
-	"github.com/OpenIMSDK/protocol/group"
-	"github.com/OpenIMSDK/tools/errs"
-
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
-)
-
-type GroupLocalCache struct {
-	lock   sync.Mutex
-	cache  map[string]GroupMemberIDsHash
-	client *rpcclient.GroupRpcClient
-}
-
-type GroupMemberIDsHash struct {
-	memberListHash uint64
-	userIDs        []string
-}
-
-func NewGroupLocalCache(client *rpcclient.GroupRpcClient) *GroupLocalCache {
-	return &GroupLocalCache{
-		cache:  make(map[string]GroupMemberIDsHash, 0),
-		client: client,
-	}
-}
-
-func (g *GroupLocalCache) GetGroupMemberIDs(ctx context.Context, groupID string) ([]string, error) {
-	resp, err := g.client.Client.GetGroupAbstractInfo(ctx, &group.GetGroupAbstractInfoReq{
-		GroupIDs: []string{groupID},
-	})
-	if err != nil {
-		return nil, err
-	}
-	if len(resp.GroupAbstractInfos) < 1 {
-		return nil, errs.ErrGroupIDNotFound
-	}
-
-	g.lock.Lock()
-	localHashInfo, ok := g.cache[groupID]
-	if ok && localHashInfo.memberListHash == resp.GroupAbstractInfos[0].GroupMemberListHash {
-		g.lock.Unlock()
-		return localHashInfo.userIDs, nil
-	}
-	g.lock.Unlock()
-
-	groupMembersResp, err := g.client.Client.GetGroupMemberUserIDs(ctx, &group.GetGroupMemberUserIDsReq{
-		GroupID: groupID,
-	})
-	if err != nil {
-		return nil, err
-	}
-
-	g.lock.Lock()
-	defer g.lock.Unlock()
-	g.cache[groupID] = GroupMemberIDsHash{
-		memberListHash: resp.GroupAbstractInfos[0].GroupMemberListHash,
-		userIDs:        groupMembersResp.UserIDs,
-	}
-	return g.cache[groupID].userIDs, nil
-}

pkg/common/db/localcache/meta_local_cache.go

@@ -1,15 +0,0 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package localcache

pkg/common/db/mgo/friend.go

@@ -16,6 +16,7 @@ package mgo
 
 import (
 	"context"
+
 	"github.com/OpenIMSDK/tools/mgoutil"
 	"github.com/OpenIMSDK/tools/pagination"
 	"go.mongodb.org/mongo-driver/mongo/options"

pkg/common/db/mgo/user.go

@@ -16,10 +16,11 @@ package mgo
 
 import (
 	"context"
+	"time"
+
 	"github.com/OpenIMSDK/protocol/user"
 	"github.com/OpenIMSDK/tools/errs"
 	"go.mongodb.org/mongo-driver/bson/primitive"
-	"time"
 
 	"github.com/OpenIMSDK/tools/mgoutil"
 	"github.com/OpenIMSDK/tools/pagination"
@@ -89,7 +90,15 @@ func (u *UserMgo) PageFindUser(ctx context.Context, level1 int64, level2 int64,
 
 	return mgoutil.FindPage[*relation.UserModel](ctx, u.coll, query, pagination)
 }
-func (u *UserMgo) PageFindUserWithKeyword(ctx context.Context, level1 int64, level2 int64, userID string, nickName string, pagination pagination.Pagination) (count int64, users []*relation.UserModel, err error) {
+
+func (u *UserMgo) PageFindUserWithKeyword(
+	ctx context.Context,
+	level1 int64,
+	level2 int64,
+	userID string,
+	nickName string,
+	pagination pagination.Pagination,
+) (count int64, users []*relation.UserModel, err error) {
 	// Initialize the base query with level conditions
 	query := bson.M{
 		"$and": []bson.M{

pkg/common/db/s3/cos/cos.go

@@ -36,9 +36,9 @@ import (
 )
 
 const (
-	minPartSize = 1024 * 1024 * 1        // 1MB
+	minPartSize int64 = 1024 * 1024 * 1        // 1MB
-	maxPartSize = 1024 * 1024 * 1024 * 5 // 5GB
+	maxPartSize int64 = 1024 * 1024 * 1024 * 5 // 5GB
-	maxNumSize  = 1000
+	maxNumSize  int64 = 1000
 )
 
 const (
@@ -133,7 +133,7 @@ func (c *Cos) PartSize(ctx context.Context, size int64) (int64, error) {
 		return 0, errors.New("size must be greater than 0")
 	}
 	if size > maxPartSize*maxNumSize {
-		return 0, fmt.Errorf("size must be less than %db", maxPartSize*maxNumSize)
+		return 0, fmt.Errorf("COS size must be less than the maximum allowed limit")
 	}
 	if size <= minPartSize*maxNumSize {
 		return minPartSize, nil

pkg/common/db/s3/minio/minio.go

@@ -45,9 +45,9 @@ const (
 )
 
 const (
-	minPartSize = 1024 * 1024 * 5        // 1MB
+	minPartSize int64 = 1024 * 1024 * 5        // 1MB
-	maxPartSize = 1024 * 1024 * 1024 * 5 // 5GB
+	maxPartSize int64 = 1024 * 1024 * 1024 * 5 // 5GB
-	maxNumSize  = 10000
+	maxNumSize  int64 = 10000
 )
 
 const (
@@ -240,7 +240,7 @@ func (m *Minio) PartSize(ctx context.Context, size int64) (int64, error) {
 		return 0, errors.New("size must be greater than 0")
 	}
 	if size > maxPartSize*maxNumSize {
-		return 0, fmt.Errorf("size must be less than %db", maxPartSize*maxNumSize)
+		return 0, fmt.Errorf("MINIO size must be less than the maximum allowed limit")
 	}
 	if size <= minPartSize*maxNumSize {
 		return minPartSize, nil

pkg/common/db/s3/oss/oss.go

@@ -37,9 +37,9 @@ import (
 )
 
 const (
-	minPartSize = 1024 * 1024 * 1        // 1MB
+	minPartSize int64 = 1024 * 1024 * 1        // 1MB
-	maxPartSize = 1024 * 1024 * 1024 * 5 // 5GB
+	maxPartSize int64 = 1024 * 1024 * 1024 * 5 // 5GB
-	maxNumSize  = 10000
+	maxNumSize  int64 = 10000
 )
 
 const (
@@ -141,7 +141,7 @@ func (o *OSS) PartSize(ctx context.Context, size int64) (int64, error) {
 		return 0, errors.New("size must be greater than 0")
 	}
 	if size > maxPartSize*maxNumSize {
-		return 0, fmt.Errorf("size must be less than %db", maxPartSize*maxNumSize)
+		return 0, fmt.Errorf("OSS size must be less than the maximum allowed limit")
 	}
 	if size <= minPartSize*maxNumSize {
 		return minPartSize, nil

pkg/common/db/s3/s3.go

@@ -24,7 +24,7 @@ import (
 type PartLimit struct {
 	MinPartSize int64 `json:"minPartSize"`
 	MaxPartSize int64 `json:"maxPartSize"`
-	MaxNumSize  int   `json:"maxNumSize"`
+	MaxNumSize  int64 `json:"maxNumSize"`
 }
 
 type InitiateMultipartUploadResult struct {

pkg/common/db/table/relation/user.go

@@ -16,9 +16,10 @@ package relation
 
 import (
 	"context"
-	"github.com/OpenIMSDK/protocol/user"
 	"time"
 
+	"github.com/OpenIMSDK/protocol/user"
+
 	"github.com/OpenIMSDK/tools/pagination"
 )
 

pkg/common/discoveryregister/direct/directResolver.go

@@ -0,0 +1,81 @@
+package direct
+
+import (
+	"context"
+	"github.com/OpenIMSDK/tools/log"
+	"google.golang.org/grpc/resolver"
+	"math/rand"
+	"strings"
+)
+
+const (
+	slashSeparator = "/"
+	// EndpointSepChar is the separator char in endpoints.
+	EndpointSepChar = ','
+
+	subsetSize = 32
+	scheme     = "direct"
+)
+
+type ResolverDirect struct {
+}
+
+func NewResolverDirect() *ResolverDirect {
+	return &ResolverDirect{}
+}
+
+func (rd *ResolverDirect) Build(target resolver.Target, cc resolver.ClientConn, _ resolver.BuildOptions) (
+	resolver.Resolver, error) {
+	log.ZDebug(context.Background(), "Build", "target", target)
+	endpoints := strings.FieldsFunc(GetEndpoints(target), func(r rune) bool {
+		return r == EndpointSepChar
+	})
+	endpoints = subset(endpoints, subsetSize)
+	addrs := make([]resolver.Address, 0, len(endpoints))
+
+	for _, val := range endpoints {
+		addrs = append(addrs, resolver.Address{
+			Addr: val,
+		})
+	}
+	if err := cc.UpdateState(resolver.State{
+		Addresses: addrs,
+	}); err != nil {
+		return nil, err
+	}
+
+	return &nopResolver{cc: cc}, nil
+}
+func init() {
+	resolver.Register(&ResolverDirect{})
+}
+func (rd *ResolverDirect) Scheme() string {
+	return scheme // return your custom scheme name
+}
+
+// GetEndpoints returns the endpoints from the given target.
+func GetEndpoints(target resolver.Target) string {
+	return strings.Trim(target.URL.Path, slashSeparator)
+}
+func subset(set []string, sub int) []string {
+	rand.Shuffle(len(set), func(i, j int) {
+		set[i], set[j] = set[j], set[i]
+	})
+	if len(set) <= sub {
+		return set
+	}
+
+	return set[:sub]
+}
+
+type nopResolver struct {
+	cc resolver.ClientConn
+}
+
+func (n nopResolver) ResolveNow(options resolver.ResolveNowOptions) {
+
+}
+
+func (n nopResolver) Close() {
+
+}

pkg/common/discoveryregister/direct/directconn.go

@@ -0,0 +1,154 @@
+package direct
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"github.com/OpenIMSDK/tools/errs"
+	config2 "github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+type ServiceAddresses map[string][]int
+
+func getServiceAddresses() ServiceAddresses {
+	return ServiceAddresses{
+		config2.Config.RpcRegisterName.OpenImUserName:           config2.Config.RpcPort.OpenImUserPort,
+		config2.Config.RpcRegisterName.OpenImFriendName:         config2.Config.RpcPort.OpenImFriendPort,
+		config2.Config.RpcRegisterName.OpenImMsgName:            config2.Config.RpcPort.OpenImMessagePort,
+		config2.Config.RpcRegisterName.OpenImMessageGatewayName: config2.Config.LongConnSvr.OpenImMessageGatewayPort,
+		config2.Config.RpcRegisterName.OpenImGroupName:          config2.Config.RpcPort.OpenImGroupPort,
+		config2.Config.RpcRegisterName.OpenImAuthName:           config2.Config.RpcPort.OpenImAuthPort,
+		config2.Config.RpcRegisterName.OpenImPushName:           config2.Config.RpcPort.OpenImPushPort,
+		config2.Config.RpcRegisterName.OpenImConversationName:   config2.Config.RpcPort.OpenImConversationPort,
+		config2.Config.RpcRegisterName.OpenImThirdName:          config2.Config.RpcPort.OpenImThirdPort,
+	}
+}
+
+type ConnDirect struct {
+	additionalOpts        []grpc.DialOption
+	currentServiceAddress string
+	conns                 map[string][]*grpc.ClientConn
+	resolverDirect        *ResolverDirect
+}
+
+func (cd *ConnDirect) GetClientLocalConns() map[string][]*grpc.ClientConn {
+	return nil
+}
+
+func (cd *ConnDirect) GetUserIdHashGatewayHost(ctx context.Context, userId string) (string, error) {
+	return "", nil
+}
+
+func (cd *ConnDirect) Register(serviceName, host string, port int, opts ...grpc.DialOption) error {
+	return nil
+}
+
+func (cd *ConnDirect) UnRegister() error {
+	return nil
+}
+
+func (cd *ConnDirect) CreateRpcRootNodes(serviceNames []string) error {
+	return nil
+}
+
+func (cd *ConnDirect) RegisterConf2Registry(key string, conf []byte) error {
+	return nil
+}
+
+func (cd *ConnDirect) GetConfFromRegistry(key string) ([]byte, error) {
+	return nil, nil
+}
+
+func (cd *ConnDirect) Close() {
+
+}
+
+func NewConnDirect() (*ConnDirect, error) {
+	return &ConnDirect{
+		conns:          make(map[string][]*grpc.ClientConn),
+		resolverDirect: NewResolverDirect(),
+	}, nil
+}
+
+func (cd *ConnDirect) GetConns(ctx context.Context,
+	serviceName string, opts ...grpc.DialOption) ([]*grpc.ClientConn, error) {
+
+	if conns, exists := cd.conns[serviceName]; exists {
+		return conns, nil
+	}
+	ports := getServiceAddresses()[serviceName]
+	var connections []*grpc.ClientConn
+	for _, port := range ports {
+		conn, err := cd.dialServiceWithoutResolver(ctx, fmt.Sprintf(config2.Config.Rpc.ListenIP+":%d", port), append(cd.additionalOpts, opts...)...)
+		if err != nil {
+			fmt.Printf("connect to port %d failed,serviceName %s, IP %s\n", port, serviceName, config2.Config.Rpc.ListenIP)
+		}
+		connections = append(connections, conn)
+	}
+
+	if len(connections) == 0 {
+		return nil, fmt.Errorf("no connections found for service: %s", serviceName)
+	}
+	return connections, nil
+}
+
+func (cd *ConnDirect) GetConn(ctx context.Context, serviceName string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
+	// Get service addresses
+	addresses := getServiceAddresses()
+	address, ok := addresses[serviceName]
+	if !ok {
+		return nil, errs.Wrap(errors.New("unknown service name"), "serviceName", serviceName)
+	}
+	var result string
+	for _, addr := range address {
+		if result != "" {
+			result = result + "," + fmt.Sprintf(config2.Config.Rpc.ListenIP+":%d", addr)
+		} else {
+			result = fmt.Sprintf(config2.Config.Rpc.ListenIP+":%d", addr)
+		}
+	}
+	// Try to dial a new connection
+	conn, err := cd.dialService(ctx, result, append(cd.additionalOpts, opts...)...)
+	if err != nil {
+		return nil, errs.Wrap(err, "address", result)
+	}
+
+	// Store the new connection
+	cd.conns[serviceName] = append(cd.conns[serviceName], conn)
+	return conn, nil
+}
+
+func (cd *ConnDirect) GetSelfConnTarget() string {
+	return cd.currentServiceAddress
+}
+
+func (cd *ConnDirect) AddOption(opts ...grpc.DialOption) {
+	cd.additionalOpts = append(cd.additionalOpts, opts...)
+}
+
+func (cd *ConnDirect) CloseConn(conn *grpc.ClientConn) {
+	if conn != nil {
+		conn.Close()
+	}
+}
+
+func (cd *ConnDirect) dialService(ctx context.Context, address string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
+	options := append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	conn, err := grpc.DialContext(ctx, cd.resolverDirect.Scheme()+":///"+address, options...)
+
+	if err != nil {
+		return nil, err
+	}
+	return conn, nil
+}
+func (cd *ConnDirect) dialServiceWithoutResolver(ctx context.Context, address string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
+	options := append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	conn, err := grpc.DialContext(ctx, address, options...)
+
+	if err != nil {
+		return nil, err
+	}
+	return conn, nil
+}

pkg/common/discoveryregister/discoveryregister.go

@@ -16,6 +16,7 @@ package discoveryregister
 
 import (
 	"errors"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister/direct"
 	"os"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister/kubernetes"
@@ -36,6 +37,8 @@ func NewDiscoveryRegister(envType string) (discoveryregistry.SvcDiscoveryRegistr
 		return zookeeper.NewZookeeperDiscoveryRegister()
 	case "k8s":
 		return kubernetes.NewK8sDiscoveryRegister()
+	case "direct":
+		return direct.NewConnDirect()
 	default:
 		return nil, errors.New("envType not correct")
 	}

pkg/common/discoveryregister/discoveryregister_test.go

@@ -40,6 +40,7 @@ func TestNewDiscoveryRegister(t *testing.T) {
 	}{
 		{"zookeeper", false, true},
 		{"k8s", false, true}, // 假设 k8s 配置也已正确设置
+		{"direct", false, true},
 		{"invalid", true, false},
 	}
 

pkg/common/discoveryregister/kubernetes/kubernetes.go

@@ -18,15 +18,17 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/stathat/consistent"
 	"os"
 	"strconv"
 	"strings"
 
+	"github.com/stathat/consistent"
+
 	"google.golang.org/grpc"
 
 	"github.com/OpenIMSDK/tools/discoveryregistry"
 	"github.com/OpenIMSDK/tools/log"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 )
 
@@ -102,7 +104,7 @@ func getSelfHost(ctx context.Context) string {
 	return host
 }
 
-// like openimserver-openim-msggateway-0.openimserver-openim-msggateway-headless.openim-lin.svc.cluster.local:88
+// like openimserver-openim-msggateway-0.openimserver-openim-msggateway-headless.openim-lin.svc.cluster.local:88.
 func getMsgGatewayHost(ctx context.Context) []string {
 	port := 88
 	instance := "openimserver"
@@ -130,20 +132,40 @@ func getMsgGatewayHost(ctx context.Context) []string {
 // GetConns returns the gRPC client connections to the specified service.
 func (cli *K8sDR) GetConns(ctx context.Context, serviceName string, opts ...grpc.DialOption) ([]*grpc.ClientConn, error) {
 
+	// This conditional checks if the serviceName is not the OpenImMessageGatewayName.
+	// It seems to handle a special case for the OpenImMessageGateway.
 	if serviceName != config.Config.RpcRegisterName.OpenImMessageGatewayName {
+		// DialContext creates a client connection to the given target (serviceName) using the specified context.
+		// 'cli.options' are likely default or common options for all connections in this struct.
+		// 'opts...' allows for additional gRPC dial options to be passed and used.
 		conn, err := grpc.DialContext(ctx, serviceName, append(cli.options, opts...)...)
+
+		// The function returns a slice of client connections with the new connection, or an error if occurred.
 		return []*grpc.ClientConn{conn}, err
 	} else {
+		// This block is executed if the serviceName is OpenImMessageGatewayName.
+		// 'ret' will accumulate the connections to return.
 		var ret []*grpc.ClientConn
+
+		// getMsgGatewayHost presumably retrieves hosts for the message gateway service.
+		// The context is passed, likely for cancellation and timeout control.
 		gatewayHosts := getMsgGatewayHost(ctx)
+
+		// Iterating over the retrieved gateway hosts.
 		for _, host := range gatewayHosts {
+			// Establishes a connection to each host.
+			// Again, appending cli.options with any additional opts provided.
 			conn, err := grpc.DialContext(ctx, host, append(cli.options, opts...)...)
+
+			// If there's an error while dialing any host, the function returns immediately with the error.
 			if err != nil {
 				return nil, err
 			} else {
+				// If the connection is successful, it is added to the 'ret' slice.
 				ret = append(ret, conn)
 			}
 		}
+		// After all hosts are processed, the slice of connections is returned.
 		return ret, nil
 	}
 }
@@ -166,7 +188,7 @@ func (cli *K8sDR) CloseConn(conn *grpc.ClientConn) {
 	conn.Close()
 }
 
-// do not use this method for call rpc
+// do not use this method for call rpc.
 func (cli *K8sDR) GetClientLocalConns() map[string][]*grpc.ClientConn {
 	fmt.Println("should not call this function!!!!!!!!!!!!!!!!!!!!!!!!!")
 	return nil

pkg/common/redispubsub/redispubliser.go

@@ -0,0 +1,16 @@
+package redispubsub
+
+import "github.com/redis/go-redis/v9"
+
+type Publisher struct {
+	client  redis.UniversalClient
+	channel string
+}
+
+func NewPublisher(client redis.UniversalClient, channel string) *Publisher {
+	return &Publisher{client: client, channel: channel}
+}
+
+func (p *Publisher) Publish(message string) error {
+	return p.client.Publish(ctx, p.channel, message).Err()
+}

pkg/common/redispubsub/redissubscriber.go

@@ -0,0 +1,34 @@
+package redispubsub
+
+import (
+	"context"
+	"github.com/redis/go-redis/v9"
+)
+
+var ctx = context.Background()
+
+type Subscriber struct {
+	client  redis.UniversalClient
+	channel string
+}
+
+func NewSubscriber(client redis.UniversalClient, channel string) *Subscriber {
+	return &Subscriber{client: client, channel: channel}
+}
+
+func (s *Subscriber) OnMessage(ctx context.Context, callback func(string)) error {
+	messageChannel := s.client.Subscribe(ctx, s.channel).Channel()
+
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case msg := <-messageChannel:
+				callback(msg.Payload)
+			}
+		}
+	}()
+
+	return nil
+}

pkg/common/startrpc/start.go

@@ -71,7 +71,7 @@ func Start(
 	}
 
 	defer client.Close()
-	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
 	registerIP, err := network.GetRpcRegisterIP(config.Config.Rpc.RegisterIP)
 	if err != nil {
 		return err

pkg/localcache/cache.go

@@ -0,0 +1,112 @@
+package localcache
+
+import (
+	"context"
+	"github.com/openimsdk/localcache/link"
+	"github.com/openimsdk/localcache/lru"
+	"hash/fnv"
+	"unsafe"
+)
+
+type Cache[V any] interface {
+	Get(ctx context.Context, key string, fetch func(ctx context.Context) (V, error)) (V, error)
+	GetLink(ctx context.Context, key string, fetch func(ctx context.Context) (V, error), link ...string) (V, error)
+	Del(ctx context.Context, key ...string)
+	DelLocal(ctx context.Context, key ...string)
+	Stop()
+}
+
+func New[V any](opts ...Option) Cache[V] {
+	opt := defaultOption()
+	for _, o := range opts {
+		o(opt)
+	}
+
+	c := cache[V]{opt: opt}
+	if opt.localSlotNum > 0 && opt.localSlotSize > 0 {
+		createSimpleLRU := func() lru.LRU[string, V] {
+			if opt.expirationEvict {
+				return lru.NewExpirationLRU[string, V](opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
+			} else {
+				return lru.NewLayLRU[string, V](opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
+			}
+		}
+		if opt.localSlotNum == 1 {
+			c.local = createSimpleLRU()
+		} else {
+			c.local = lru.NewSlotLRU[string, V](opt.localSlotNum, func(key string) uint64 {
+				h := fnv.New64a()
+				h.Write(*(*[]byte)(unsafe.Pointer(&key)))
+				return h.Sum64()
+			}, createSimpleLRU)
+		}
+		if opt.linkSlotNum > 0 {
+			c.link = link.New(opt.linkSlotNum)
+		}
+	}
+	return &c
+}
+
+type cache[V any] struct {
+	opt   *option
+	link  link.Link
+	local lru.LRU[string, V]
+}
+
+func (c *cache[V]) onEvict(key string, value V) {
+	if c.link != nil {
+		lks := c.link.Del(key)
+		for k := range lks {
+			if key != k { // prevent deadlock
+				c.local.Del(k)
+			}
+		}
+	}
+}
+
+func (c *cache[V]) del(key ...string) {
+	if c.local == nil {
+		return
+	}
+	for _, k := range key {
+		c.local.Del(k)
+		if c.link != nil {
+			lks := c.link.Del(k)
+			for k := range lks {
+				c.local.Del(k)
+			}
+		}
+	}
+}
+
+func (c *cache[V]) Get(ctx context.Context, key string, fetch func(ctx context.Context) (V, error)) (V, error) {
+	return c.GetLink(ctx, key, fetch)
+}
+
+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.Link(key, link...)
+			}
+			return fetch(ctx)
+		})
+	} else {
+		return fetch(ctx)
+	}
+}
+
+func (c *cache[V]) Del(ctx context.Context, key ...string) {
+	for _, fn := range c.opt.delFn {
+		fn(ctx, key...)
+	}
+	c.del(key...)
+}
+
+func (c *cache[V]) DelLocal(ctx context.Context, key ...string) {
+	c.del(key...)
+}
+
+func (c *cache[V]) Stop() {
+	c.local.Stop()
+}

pkg/localcache/cache_test.go

@@ -0,0 +1,79 @@
+package localcache
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestName(t *testing.T) {
+	c := New[string](WithExpirationEvict())
+	//c := New[string]()
+	ctx := context.Background()
+
+	const (
+		num  = 10000
+		tNum = 10000
+		kNum = 100000
+		pNum = 100
+	)
+
+	getKey := func(v uint64) string {
+		return fmt.Sprintf("key_%d", v%kNum)
+	}
+
+	start := time.Now()
+	t.Log("start", start)
+
+	var (
+		get atomic.Int64
+		del atomic.Int64
+	)
+
+	incrGet := func() {
+		if v := get.Add(1); v%pNum == 0 {
+			//t.Log("#get count", v/pNum)
+		}
+	}
+	incrDel := func() {
+		if v := del.Add(1); v%pNum == 0 {
+			//t.Log("@del count", v/pNum)
+		}
+	}
+
+	var wg sync.WaitGroup
+
+	for i := 0; i < tNum; i++ {
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			for i := 0; i < num; i++ {
+				c.Get(ctx, getKey(rand.Uint64()), func(ctx context.Context) (string, error) {
+					return fmt.Sprintf("index_%d", i), nil
+				})
+				incrGet()
+			}
+		}()
+
+		go func() {
+			defer wg.Done()
+			time.Sleep(time.Second / 10)
+			for i := 0; i < num; i++ {
+				c.Del(ctx, getKey(rand.Uint64()))
+				incrDel()
+			}
+		}()
+	}
+
+	wg.Wait()
+	end := time.Now()
+	t.Log("end", end)
+	t.Log("time", end.Sub(start))
+	t.Log("get", get.Load())
+	t.Log("del", del.Load())
+	// 137.35s
+}

pkg/localcache/go.mod

@@ -0,0 +1,5 @@
+module github.com/openimsdk/localcache
+
+go 1.19
+
+require github.com/hashicorp/golang-lru/v2 v2.0.7

pkg/localcache/link/link.go

@@ -0,0 +1,109 @@
+package link
+
+import (
+	"hash/fnv"
+	"sync"
+	"unsafe"
+)
+
+type Link interface {
+	Link(key string, link ...string)
+	Del(key string) map[string]struct{}
+}
+
+func newLinkKey() *linkKey {
+	return &linkKey{
+		data: make(map[string]map[string]struct{}),
+	}
+}
+
+type linkKey struct {
+	lock sync.Mutex
+	data map[string]map[string]struct{}
+}
+
+func (x *linkKey) link(key string, link ...string) {
+	x.lock.Lock()
+	defer x.lock.Unlock()
+	v, ok := x.data[key]
+	if !ok {
+		v = make(map[string]struct{})
+		x.data[key] = v
+	}
+	for _, k := range link {
+		v[k] = struct{}{}
+	}
+}
+
+func (x *linkKey) del(key string) map[string]struct{} {
+	x.lock.Lock()
+	defer x.lock.Unlock()
+	ks, ok := x.data[key]
+	if !ok {
+		return nil
+	}
+	delete(x.data, key)
+	return ks
+}
+
+func New(n int) Link {
+	if n <= 0 {
+		panic("must be greater than 0")
+	}
+	slots := make([]*linkKey, n)
+	for i := 0; i < len(slots); i++ {
+		slots[i] = newLinkKey()
+	}
+	return &slot{
+		n:     uint64(n),
+		slots: slots,
+	}
+}
+
+type slot struct {
+	n     uint64
+	slots []*linkKey
+}
+
+func (x *slot) index(s string) uint64 {
+	h := fnv.New64a()
+	_, _ = h.Write(*(*[]byte)(unsafe.Pointer(&s)))
+	return h.Sum64() % x.n
+}
+
+func (x *slot) Link(key string, link ...string) {
+	if len(link) == 0 {
+		return
+	}
+	mk := key
+	lks := make([]string, len(link))
+	for i, k := range link {
+		lks[i] = k
+	}
+	x.slots[x.index(mk)].link(mk, lks...)
+	for _, lk := range lks {
+		x.slots[x.index(lk)].link(lk, mk)
+	}
+}
+
+func (x *slot) Del(key string) map[string]struct{} {
+	return x.delKey(key)
+}
+
+func (x *slot) delKey(k string) map[string]struct{} {
+	del := make(map[string]struct{})
+	stack := []string{k}
+	for len(stack) > 0 {
+		curr := stack[len(stack)-1]
+		stack = stack[:len(stack)-1]
+		if _, ok := del[curr]; ok {
+			continue
+		}
+		del[curr] = struct{}{}
+		childKeys := x.slots[x.index(curr)].del(curr)
+		for ck := range childKeys {
+			stack = append(stack, ck)
+		}
+	}
+	return del
+}

pkg/localcache/link/link_test.go

@@ -0,0 +1,20 @@
+package link
+
+import (
+	"testing"
+)
+
+func TestName(t *testing.T) {
+
+	v := New(1)
+
+	//v.Link("a:1", "b:1", "c:1", "d:1")
+	v.Link("a:1", "b:1", "c:1")
+	v.Link("z:1", "b:1")
+
+	//v.DelKey("a:1")
+	v.Del("z:1")
+
+	t.Log(v)
+
+}

pkg/localcache/lru/lru.go

@@ -0,0 +1,20 @@
+package lru
+
+import "github.com/hashicorp/golang-lru/v2/simplelru"
+
+type EvictCallback[K comparable, V any] simplelru.EvictCallback[K, V]
+
+type LRU[K comparable, V any] interface {
+	Get(key K, fetch func() (V, error)) (V, error)
+	Del(key K) bool
+	Stop()
+}
+
+type Target interface {
+	IncrGetHit()
+	IncrGetSuccess()
+	IncrGetFailed()
+
+	IncrDelHit()
+	IncrDelNotFound()
+}

pkg/localcache/lru/lru_expiration.go

@@ -0,0 +1,78 @@
+package lru
+
+import (
+	"github.com/hashicorp/golang-lru/v2/expirable"
+	"sync"
+	"time"
+)
+
+func NewExpirationLRU[K comparable, V any](size int, successTTL, failedTTL time.Duration, target Target, onEvict EvictCallback[K, V]) LRU[K, V] {
+	var cb expirable.EvictCallback[K, *expirationLruItem[V]]
+	if onEvict != nil {
+		cb = func(key K, value *expirationLruItem[V]) {
+			onEvict(key, value.value)
+		}
+	}
+	core := expirable.NewLRU[K, *expirationLruItem[V]](size, cb, successTTL)
+	return &ExpirationLRU[K, V]{
+		core:       core,
+		successTTL: successTTL,
+		failedTTL:  failedTTL,
+		target:     target,
+	}
+}
+
+type expirationLruItem[V any] struct {
+	lock  sync.RWMutex
+	err   error
+	value V
+}
+
+type ExpirationLRU[K comparable, V any] struct {
+	lock       sync.Mutex
+	core       *expirable.LRU[K, *expirationLruItem[V]]
+	successTTL time.Duration
+	failedTTL  time.Duration
+	target     Target
+}
+
+func (x *ExpirationLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) {
+	x.lock.Lock()
+	v, ok := x.core.Get(key)
+	if ok {
+		x.lock.Unlock()
+		x.target.IncrGetSuccess()
+		v.lock.RLock()
+		defer v.lock.RUnlock()
+		return v.value, v.err
+	} else {
+		v = &expirationLruItem[V]{}
+		x.core.Add(key, v)
+		v.lock.Lock()
+		x.lock.Unlock()
+		defer v.lock.Unlock()
+		v.value, v.err = fetch()
+		if v.err == nil {
+			x.target.IncrGetSuccess()
+		} else {
+			x.target.IncrGetFailed()
+			x.core.Remove(key)
+		}
+		return v.value, v.err
+	}
+}
+
+func (x *ExpirationLRU[K, V]) Del(key K) bool {
+	x.lock.Lock()
+	ok := x.core.Remove(key)
+	x.lock.Unlock()
+	if ok {
+		x.target.IncrDelHit()
+	} else {
+		x.target.IncrDelNotFound()
+	}
+	return ok
+}
+
+func (x *ExpirationLRU[K, V]) Stop() {
+}