Start of Merge Conflict Resolution

.devcontainer/devcontainer.json

@@ -2,6 +2,7 @@
     // 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,7 +17,6 @@ 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,8 +23,6 @@ on:
         - "docs/**"
         - "README.md"
         - "README_zh-CN.md"
-        - "**.md"
-        - "docs/**"
         - "CONTRIBUTING.md"
   pull_request:
     branches:
@@ -33,8 +31,7 @@ on:
     paths-ignore:
         - "README.md"
         - "README_zh-CN.md"
-        - "CONTRIBUTING/**"
+        - "CONTRIBUTING.md"
-        - "**.md"
         - "docs/**"
 
 env:
@@ -70,9 +67,6 @@ 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
@@ -97,6 +91,13 @@ 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,6 +78,9 @@ 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 OpenIM Release Branch Public Code To Other Repositories
+name: Synchronize kubecub public code to other repositories
 on:
     push:
         paths:
@@ -41,4 +41,3 @@ 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 OpenIM Main Branch Public Code To Other Repositories
+name: Synchronize kubecub public code to other repositories
 on:
     push:
         branches:
@@ -36,5 +36,4 @@ jobs:
             file-sync
             automerge
           ASSIGNEES: |
-            kubbot
+            kubbot
-        continue-on-error: true

.gitignore

@@ -34,7 +34,6 @@ 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

@@ -0,0 +1,15 @@
+# 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,16 +70,11 @@ 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/config.yaml
+# --| source: deployments/templates/openim.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: openIM
+  username: root
   password: openIM123
   maxPoolSize: 100
 
@@ -122,14 +122,14 @@ api:
 # minio.signEndpoint is minio public network address
 object:
   enable: "minio"
-  apiURL: "http://172.28.0.1:10002"
+  apiURL: "http://14.155.64.202:10002"
   minio:
     bucket: "openim"
     endpoint: "http://172.28.0.1:10005"
     accessKeyID: "root"
     secretAccessKey: "openIM123"
     sessionToken: ''
-    signEndpoint: "http://172.28.0.1:10005"
+    signEndpoint: "http://14.155.64.202: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: /workspaces/open-im-server/logs/
+  storageLocation: /data/workspaces/open-im-server/logs/
   rotationTime: 24
   remainRotationCount: 2
   remainLogLevel: 6
@@ -247,14 +247,6 @@ 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
@@ -315,21 +307,21 @@ iosPush:
 # Timeout in seconds
 # Whether to continue execution if callback fails
 callback:
-  url: "http://127.0.0.1:10008/callbackExample"
+  url: ""
   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: true
+    enable: false
     timeout: 5
     failedContinue: true
   beforeSendGroupMsg:
@@ -513,8 +505,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: true
+  enable: false
-  grafanaUrl: http://172.28.0.1:13000/
+  grafanaUrl: 172.28.0.1:13000
   apiPrometheusPort: [20100]
   userPrometheusPort: [ 20110 ]
   friendPrometheusPort: [ 20120 ]

config/templates/env.template

@@ -1,4 +1,4 @@
-# Copyright © 2024 OpenIM. All rights reserved.
+# 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.
@@ -12,26 +12,31 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-# -----------------------------------------------------------------------------
+# ======================================
-# General Configuration
+# ========= Basic 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.
 
-# Data storage directory for persistent data.
+# The user for authentication or system operations.
-# Example: DATA_DIR=/path/to/data
+# Default: OPENIM_USER=root
-DATA_DIR=/workspaces/open-im-server
+USER=root
 
-# Docker image registry. Uncomment the preferred one.
+# Password associated with the specified user for authentication.
-# Options: ghcr.io/openimsdk, openim, registry.cn-hangzhou.aliyuncs.com/openimsdk
+# Default: PASSWORD=openIM123
-# IMAGE_REGISTRY="ghcr.io/openimsdk"
+PASSWORD=openIM123
-# IMAGE_REGISTRY="openim"
+
-# IMAGE_REGISTRY="registry.cn-hangzhou.aliyuncs.com/openimsdk"
+# Base URL for the application programming interface (API).
+# 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
 
 # ======================================
@@ -42,9 +47,10 @@ IMAGE_REGISTRY=ghcr.io/openimsdk
 # Default: DOCKER_BRIDGE_SUBNET=172.28.0.0/16
 DOCKER_BRIDGE_SUBNET=172.28.0.0/16
 
-# Set and specify the IP addresses of some containers. Generally speaking, 
+# Gateway for the Docker network.
-# you do not need to modify these configurations to facilitate debugging
+# Default: DOCKER_BRIDGE_GATEWAY=172.28.0.1
 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
@@ -59,66 +65,45 @@ 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
 
-# ==============================================================================
+# ===============================================
-# Configuration Update Instructions
+# = Component Extension Configuration =
-# ==============================================================================
+# ===============================================
-# 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 service port configuration.
+# ----- MongoDB 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
 
-# Password for MongoDB admin user. Used for service authentication.
+# Username to authenticate with the MongoDB service.
+# Default: MONGO_USERNAME=root
+# MONGO_USERNAME=root
+
+# Password to authenticate with the MongoDB service.
 # Default: MONGO_PASSWORD=openIM123
 MONGO_PASSWORD=openIM123
 
-# Username for a regular OpenIM user in MongoDB.
+# Name of the database in MongoDB to be used.
-# 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
@@ -128,6 +113,11 @@ 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=''
 
@@ -139,13 +129,20 @@ KAFKA_PORT=19094
 # Default: KAFKA_LATESTMSG_REDIS_TOPIC=latestMsgToRedis
 KAFKA_LATESTMSG_REDIS_TOPIC=latestMsgToRedis
 
-# MINIO_PORT
+# Topic in Kafka for pushing messages (e.g. notifications or updates).
-# ----------
+# Default: KAFKA_MSG_PUSH_TOPIC=msgToPush
-# MINIO_PORT sets the port for the MinIO object storage service.
+KAFKA_MSG_PUSH_TOPIC=msgToPush
-# Upon changing this port, the MinIO endpoint URLs in the  file must be updated
+
-# to reflect this change. The endpoints include both the 'endpoint' and 'signEndpoint' 
+# Topic in Kafka for storing offline messages in MongoDB.
-# under the MinIO configuration.
+# Default: KAFKA_OFFLINEMSG_MONGO_TOPIC=offlineMsgToMongoMysql
-#
+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
 
@@ -158,11 +155,19 @@ 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
@@ -178,34 +183,41 @@ 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
 
-# API_OPENIM_PORT
+# Port for the OpenIM API.
-# ---------------
-# 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
 
-# Port for the OpenIM admin API.
+# Directory path for storing data files or related information for OpenIM chat.
-# Default: OPENIM_ADMIN_API_PORT=10009
+# Default: OPENIM_CHAT_DATA_DIR=./openim-chat/main
-OPENIM_ADMIN_API_PORT=10009
+OPENIM_CHAT_DATA_DIR=./openim-chat/main
+
 
 # ======================================
 # ========== OpenIM Admin ==============
@@ -214,6 +226,10 @@ OPENIM_ADMIN_API_PORT=10009
 # 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/config.yaml > ./charts/generated-configs/config.yaml
+../scripts/genconfig.sh ../scripts/install/environment.sh ./templates/openim.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/openim.yaml

@@ -14,7 +14,7 @@
 
 # -----------------------------------------------------------------
 # TODO: This config file is the template file
-# --| source: deployments/templates/config.yaml
+# --| source: deployments/templates/openim.yaml
 # --| env: scripts/install/environment
 # --| target: config/config.yaml
 # -----------------------------------------------------------------
@@ -527,39 +527,3 @@ 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,7 +214,6 @@ 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/config.yaml
+docs/guide/en-US/yaml/openim/openim.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 [`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.
+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.
 
 ###  1.3. <a name='DockerComposeDeployment'></a>Docker Compose Deployment
 

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

@@ -45,6 +45,63 @@ 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. 
@@ -185,9 +242,12 @@ make start
 To check, ensure the following four processes start successfully:
 
 ```bash
-make check 
+➜  chat git:(main) ✗ ps -ef | grep chat      
+# [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.49
+	github.com/OpenIMSDK/protocol v0.0.48
-	github.com/OpenIMSDK/tools v0.0.28
+	github.com/OpenIMSDK/tools v0.0.23
 	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,7 +18,6 @@ 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
@@ -39,7 +38,6 @@ 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
@@ -83,7 +81,6 @@ 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
@@ -119,6 +116,7 @@ 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
@@ -159,4 +157,4 @@ require (
 	gopkg.in/ini.v1 v1.67.0 // indirect
 )
 
-replace github.com/openimsdk/localcache => ./pkg/localcache
+replace github.com/OpenIMSDK/protocol v0.0.47 => github.com/AndrewZuo01/protocol v0.0.0-20240112093520-fd9c53e27b94

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/protocol v0.0.49 h1:wcqJOMBis7f153zNI7V82Fc4WyqA1GanMgXUQgL618k=
+github.com/OpenIMSDK/tools v0.0.23 h1:xozfrGzhbpNPlDTap5DLVPk+JfgZ/ZyIj4Cuu3/bm9w=
-github.com/OpenIMSDK/protocol v0.0.49/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y=
+github.com/OpenIMSDK/tools v0.0.23/go.mod h1:eg+q4A34Qmu73xkY0mt37FHGMCMfC6CtmOnm0kFEGFI=
-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,8 +171,6 @@ 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,7 +16,6 @@ package api
 
 import (
 	"context"
-	"fmt"
 	"net/http"
 
 	"github.com/OpenIMSDK/protocol/constant"
@@ -44,7 +43,7 @@ import (
 )
 
 func NewGinRouter(discov discoveryregistry.SvcDiscoveryRegistry, rdb redis.UniversalClient) *gin.Engine {
-	discov.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin"))) // 默认RPC中间件
+	discov.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials())) // 默认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.WithEnable {
+//	if !config.Config.Callback.CallbackUserOnline.Enable {
 //		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.WithEnable {
+//	if !config.Config.Callback.CallbackUserOffline.Enable {
 //		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.WithEnable {
+//	if !config.Config.Callback.CallbackUserKickOff.Enable {
 //		return callbackResp
 //	}
 //	callbackUserKickOffReq := cbApi.CallbackUserKickOffReq{

internal/msggateway/hub_server.go

@@ -25,6 +25,8 @@ 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"
@@ -57,7 +59,7 @@ type Server struct {
 	rpcPort        int
 	prometheusPort int
 	LongConnServer LongConnServer
-	pushTerminal   map[int]struct{}
+	pushTerminal   []int
 }
 
 func (s *Server) SetLongConnServer(LongConnServer LongConnServer) {
@@ -65,15 +67,12 @@ func (s *Server) SetLongConnServer(LongConnServer LongConnServer) {
 }
 
 func NewServer(rpcPort int, proPort int, longConnServer LongConnServer) *Server {
-	s := &Server{
+	return &Server{
 		rpcPort:        rpcPort,
 		prometheusPort: proPort,
 		LongConnServer: longConnServer,
-		pushTerminal:   make(map[int]struct{}),
+		pushTerminal:   []int{constant.IOSPlatformID, constant.AndroidPlatformID},
 	}
-	s.pushTerminal[constant.IOSPlatformID] = struct{}{}
-	s.pushTerminal[constant.AndroidPlatformID] = struct{}{}
-	return s
 }
 
 func (s *Server) OnlinePushMsg(
@@ -127,9 +126,13 @@ func (s *Server) OnlineBatchPushOneMsg(
 	panic("implement me")
 }
 
-func (s *Server) SuperGroupOnlineBatchPushOneMsg(ctx context.Context, req *msggateway.OnlineBatchPushOneMsgReq,
+func (s *Server) SuperGroupOnlineBatchPushOneMsg(
+	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{
@@ -150,22 +153,23 @@ func (s *Server) SuperGroupOnlineBatchPushOneMsg(ctx context.Context, req *msgga
 			}
 
 			userPlatform := &msggateway.SingleMsgToUserPlatform{
-				PlatFormID: int32(client.PlatformID),
+				RecvID:         v,
+				RecvPlatFormID: int32(client.PlatformID),
 			}
 			if !client.IsBackground ||
 				(client.IsBackground && client.PlatformID != constant.IOSPlatformID) {
 				err := client.PushMessage(ctx, req.MsgData)
 				if err != nil {
-					userPlatform.ResultCode = int64(errs.ErrPushMsgErr.Code())
+					userPlatform.ResultCode = -2
 					resp = append(resp, userPlatform)
 				} else {
-					if _, ok := s.pushTerminal[client.PlatformID]; ok {
+					if utils.IsContainInt(client.PlatformID, s.pushTerminal) {
 						results.OnlinePush = true
 						resp = append(resp, userPlatform)
 					}
 				}
 			} else {
-				userPlatform.ResultCode = int64(errs.ErrIOSBackgroundPushErr.Code())
+				userPlatform.ResultCode = -3
 				resp = append(resp, userPlatform)
 			}
 		}

internal/msggateway/n_ws_server.go

@@ -19,6 +19,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/OpenIMSDK/tools/apiresp"
 	"net/http"
 	"os"
 	"os/signal"
@@ -28,8 +29,6 @@ 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()), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
+	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()))
 	msgModel := cache.NewMsgCacheModel(rdb)
 	msgDocModel := unrelation.NewMsgMongoDriver(mongo.GetDatabase())
 	msgDatabase := controller.NewCommonMsgDatabase(msgDocModel, msgModel)

internal/push/callback.go

@@ -16,7 +16,6 @@ package push
 
 import (
 	"context"
-	"encoding/json"
 
 	"github.com/OpenIMSDK/protocol/constant"
 	"github.com/OpenIMSDK/protocol/sdkws"
@@ -137,14 +136,3 @@ 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

@@ -0,0 +1,32 @@
+// 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,16 +16,17 @@ 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 NewDummy() *Dummy {
+func NewClient() *Dummy {
 	return &Dummy{}
 }
 
 type Dummy struct {
 }
 
-func (d *Dummy) Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error {
+func (d *Dummy) Push(ctx context.Context, userIDs []string, title, content string, opts *offlinepush.Opts) error {
 	return nil
 }

internal/push/offlinepush/fcm/push.go

@@ -16,7 +16,6 @@ package fcm
 
 import (
 	"context"
-	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
 	"path/filepath"
 
 	firebase "firebase.google.com/go"
@@ -26,6 +25,7 @@ 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 NewFcm(cache cache.MsgModel) *Fcm {
+func NewClient(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 NewFcm(cache cache.MsgModel) *Fcm {
 	return &Fcm{fcmMsgCli: fcmMsgClient, cache: cache}
 }
 
-func (f *Fcm) Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error {
+func (f *Fcm) Push(ctx context.Context, userIDs []string, title, content string, opts *offlinepush.Opts) error {
 	// accounts->registrationToken
 	allTokens := make(map[string][]string, 0)
 	for _, account := range userIDs {

internal/push/offlinepush/getui/push.go

@@ -19,7 +19,6 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"errors"
-	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
 	"strconv"
 	"sync"
 	"time"
@@ -31,6 +30,7 @@ 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 GeTui struct {
+type Client struct {
 	cache           cache.MsgModel
 	tokenExpireTime int64
 	taskIDTTL       int64
 }
 
-func NewGeTui(cache cache.MsgModel) *GeTui {
+func NewClient(cache cache.MsgModel) *Client {
-	return &GeTui{cache: cache, tokenExpireTime: tokenExpireTime, taskIDTTL: taskIDTTL}
+	return &Client{cache: cache, tokenExpireTime: tokenExpireTime, taskIDTTL: taskIDTTL}
 }
 
-func (g *GeTui) Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error {
+func (g *Client) Push(ctx context.Context, userIDs []string, title, content string, opts *offlinepush.Opts) error {
 	token, err := g.cache.GetGetuiToken(ctx)
 	if err != nil {
 		if errs.Unwrap(err) == redis.Nil {
@@ -111,7 +111,7 @@ func (g *GeTui) Push(ctx context.Context, userIDs []string, title, content strin
 	return err
 }
 
-func (g *GeTui) Auth(ctx context.Context, timeStamp int64) (token string, expireTime int64, err error) {
+func (g *Client) 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 *GeTui) Auth(ctx context.Context, timeStamp int64) (token string, expire
 	return respAuth.Token, int64(expire), err
 }
 
-func (g *GeTui) GetTaskID(ctx context.Context, token string, pushReq PushReq) (string, error) {
+func (g *Client) 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 *GeTui) GetTaskID(ctx context.Context, token string, pushReq PushReq) (s
 }
 
 // max num is 999.
-func (g *GeTui) batchPush(ctx context.Context, token string, userIDs []string, pushReq PushReq) error {
+func (g *Client) 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 *GeTui) batchPush(ctx context.Context, token string, userIDs []string, p
 	return g.request(ctx, batchPushURL, pushReq, token, nil)
 }
 
-func (g *GeTui) singlePush(ctx context.Context, token, userID string, pushReq PushReq) error {
+func (g *Client) 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 *GeTui) request(ctx context.Context, url string, input any, token string, output any) error {
+func (g *Client) 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 *GeTui) postReturn(
+func (g *Client) postReturn(
 	ctx context.Context,
 	url string,
 	header map[string]string,
@@ -181,7 +181,7 @@ func (g *GeTui) postReturn(
 	return output.parseError()
 }
 
-func (g *GeTui) getTokenAndSave2Redis(ctx context.Context) (token string, err error) {
+func (g *Client) 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 *GeTui) getTokenAndSave2Redis(ctx context.Context) (token string, err er
 	return token, nil
 }
 
-func (g *GeTui) GetTaskIDAndSave2Redis(ctx context.Context, token string, pushReq PushReq) (taskID string, err error) {
+func (g *Client) 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 NewJPush() *JPush {
+func NewClient() *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 *options.Opts) error {
+func (j *JPush) Push(ctx context.Context, userIDs []string, title, content string, opts *offlinepush.Opts) error {
 	var pf body.Platform
 	pf.SetAll()
 	var au body.Audience

internal/push/offlinepush/offlinepush_interface.go

@@ -0,0 +1,37 @@
+// 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

@@ -1,52 +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"
-	"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

@@ -1,14 +0,0 @@
-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

@@ -1,211 +0,0 @@
-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

@@ -1,51 +0,0 @@
-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,17 +16,6 @@ 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"
@@ -42,31 +31,18 @@ import (
 )
 
 type ConsumerHandler struct {
-	pushConsumerGroup      *kfk.MConsumerGroup
+	pushConsumerGroup *kfk.MConsumerGroup
-	offlinePusher          offlinepush.OfflinePusher
+	pusher            *Pusher
-	onlinePusher           OnlinePusher
-	groupLocalCache        *rpccache.GroupLocalCache
-	conversationLocalCache *rpccache.ConversationLocalCache
-	msgRpcClient           rpcclient.MessageRpcClient
-	conversationRpcClient  rpcclient.ConversationRpcClient
-	groupRpcClient         rpcclient.GroupRpcClient
 }
 
-func NewConsumerHandler(offlinePusher offlinepush.OfflinePusher,
+func NewConsumerHandler(pusher *Pusher) *ConsumerHandler {
-	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
 }
 
@@ -89,7 +65,7 @@ func (c *ConsumerHandler) handleMs2PsChat(ctx context.Context, msg []byte) {
 	var err error
 	switch msgFromMQ.MsgData.SessionType {
 	case constant.SuperGroupChatType:
-		err = c.Push2SuperGroup(ctx, pbData.MsgData.GroupID, pbData.MsgData)
+		err = c.pusher.Push2SuperGroup(ctx, pbData.MsgData.GroupID, pbData.MsgData)
 	default:
 		var pushUserIDList []string
 		isSenderSync := utils.GetSwitchFromOptions(pbData.MsgData.Options, constant.IsSenderSync)
@@ -98,14 +74,18 @@ func (c *ConsumerHandler) handleMs2PsChat(ctx context.Context, msg []byte) {
 		} else {
 			pushUserIDList = append(pushUserIDList, pbData.MsgData.RecvID, pbData.MsgData.SendID)
 		}
-		err = c.Push2User(ctx, pushUserIDList, pbData.MsgData)
+		err = c.pusher.Push2User(ctx, pushUserIDList, pbData.MsgData)
 	}
 	if err != nil {
-		log.ZError(ctx, "push failed", err, "msg", pbData.String())
+		if err == errNoOfflinePusher {
+			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 {
@@ -116,243 +96,3 @@ 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

@@ -0,0 +1,109 @@
+// 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

@@ -0,0 +1,511 @@
+// 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

@@ -0,0 +1,32 @@
+// 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,8 @@ package conversation
 import (
 	"context"
 	"errors"
-	"sort"
-
 	"github.com/OpenIMSDK/protocol/sdkws"
+	"sort"
 
 	"github.com/OpenIMSDK/tools/tx"
 

internal/rpc/friend/friend.go

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

internal/rpc/msg/as_read.go

@@ -18,7 +18,6 @@ import (
 	"context"
 
 	utils2 "github.com/OpenIMSDK/tools/utils"
-
 	cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
 
 	"github.com/redis/go-redis/v9"
@@ -44,7 +43,7 @@ func (m *msgServer) GetConversationsHasReadAndMaxSeq(ctx context.Context, req *m
 	if err != nil {
 		return nil, err
 	}
-	conversations, err := m.ConversationLocalCache.GetConversations(ctx, req.UserID, conversationIDs)
+	conversations, err := m.Conversation.GetConversations(ctx, req.UserID, conversationIDs)
 	if err != nil {
 		return nil, err
 	}
@@ -107,7 +106,7 @@ func (m *msgServer) MarkMsgsAsRead(
 	if hasReadSeq > maxSeq {
 		return nil, errs.ErrArgs.Wrap("hasReadSeq must not be bigger than maxSeq")
 	}
-	conversation, err := m.ConversationLocalCache.GetConversation(ctx, req.UserID, req.ConversationID)
+	conversation, err := m.Conversation.GetConversation(ctx, req.UserID, req.ConversationID)
 	if err != nil {
 		return
 	}
@@ -147,7 +146,7 @@ func (m *msgServer) MarkConversationAsRead(
 	ctx context.Context,
 	req *msg.MarkConversationAsReadReq,
 ) (resp *msg.MarkConversationAsReadResp, err error) {
-	conversation, err := m.ConversationLocalCache.GetConversation(ctx, req.UserID, req.ConversationID)
+	conversation, err := m.Conversation.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.UserLocalCache.GetUserInfo(ctx, req.UserID)
+	user, err := m.User.GetUserInfo(ctx, req.UserID)
 	if err != nil {
 		return nil, err
 	}
@@ -73,7 +73,12 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
 			}
 			role = user.AppMangerLevel
 		case constant.SuperGroupChatType:
-			members, err := m.GroupLocalCache.GetGroupMemberInfoMap(ctx, msgs[0].GroupID, utils.Distinct([]string{req.UserID, msgs[0].SendID}))
+			members, err := m.Group.GetGroupMemberInfoMap(
+				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.GroupLocalCache.GetGroupMemberIDs(ctx, msg.GroupID)
+		memberUserIDList, err := m.Group.GetGroupMemberIDs(ctx, msg.GroupID)
 		if err != nil {
 			log.ZWarn(ctx, "GetGroupMemberIDs", err)
 			return
@@ -144,7 +144,6 @@ 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,7 +16,6 @@ package msg
 
 import (
 	"context"
-
 	pbmsg "github.com/OpenIMSDK/protocol/msg"
 )
 

internal/rpc/msg/server.go

@@ -16,7 +16,6 @@ package msg
 
 import (
 	"context"
-	"github.com/openimsdk/open-im-server/v3/pkg/rpccache"
 
 	"google.golang.org/grpc"
 
@@ -27,6 +26,7 @@ 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,11 +36,12 @@ type (
 	msgServer               struct {
 		RegisterCenter         discoveryregistry.SvcDiscoveryRegistry
 		MsgDatabase            controller.CommonMsgDatabase
+		Group                  *rpcclient.GroupRpcClient
+		User                   *rpcclient.UserRpcClient
 		Conversation           *rpcclient.ConversationRpcClient
-		UserLocalCache         *rpccache.UserLocalCache
+		friend                 *rpcclient.FriendRpcClient
-		FriendLocalCache       *rpccache.FriendLocalCache
+		GroupLocalCache        *localcache.GroupLocalCache
-		GroupLocalCache        *rpccache.GroupLocalCache
+		ConversationLocalCache *localcache.ConversationLocalCache
-		ConversationLocalCache *rpccache.ConversationLocalCache
 		Handlers               MessageInterceptorChain
 		notificationSender     *rpcclient.NotificationSender
 	}
@@ -80,15 +81,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,
-		UserLocalCache:         rpccache.NewUserLocalCache(userRpcClient, rdb),
+		GroupLocalCache:        localcache.NewGroupLocalCache(&groupRpcClient),
-		GroupLocalCache:        rpccache.NewGroupLocalCache(groupRpcClient, rdb),
+		ConversationLocalCache: localcache.NewConversationLocalCache(&conversationClient),
-		ConversationLocalCache: rpccache.NewConversationLocalCache(conversationClient, rdb),
+		friend:                 &friendRpcClient,
-		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.UserLocalCache.GetUsersInfoMap(ctx, userIDs)
+		userMap, err := m.User.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.GroupLocalCache.GetGroupInfos(ctx, groupIDs)
+		resp, err := m.Group.GetGroupInfos(ctx, groupIDs, false)
 		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.ConversationLocalCache.GetConversation(ctx, req.UserID, seq.ConversationID)
+			conversation, err := m.Conversation.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.UserLocalCache.GetUsersInfo(ctx, sendIDs)
+		sendInfos, err := m.User.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.UserLocalCache.GetUsersInfo(ctx, recvIDs)
+		recvInfos, err := m.User.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.GroupLocalCache.GetGroupInfos(ctx, groupIDs)
+		groupInfos, err := m.Group.GetGroupInfos(ctx, groupIDs, true)
 		if err != nil {
 			return nil, err
 		}

internal/rpc/msg/verify.go

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

internal/rpc/third/s3.go

@@ -19,14 +19,12 @@ 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,7 +16,6 @@ package user
 
 import (
 	"context"
-
 	pbuser "github.com/OpenIMSDK/protocol/user"
 	"github.com/OpenIMSDK/tools/utils"
 

internal/rpc/user/user.go

@@ -17,14 +17,12 @@ 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"
@@ -61,7 +59,6 @@ type userServer struct {
 	RegisterCenter           registry.SvcDiscoveryRegistry
 }
 
-
 func Start(client registry.SvcDiscoveryRegistry, server *grpc.Server) error {
 	rdb, err := cache.NewRedis()
 	if err != nil {
@@ -244,7 +241,6 @@ func (s *userServer) GetPaginationUsers(ctx context.Context, req *pbuser.GetPagi
 			return nil, err
 		}
 		return &pbuser.GetPaginationUsersResp{Total: int32(total), Users: convert.UsersDB2Pb(users)}, err
-
 	}
 
 }
@@ -391,7 +387,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 {
@@ -422,7 +418,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 {
@@ -441,11 +437,10 @@ 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 {
@@ -478,7 +473,6 @@ 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()), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
+	discov.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()))
 	userDB, err := mgo.NewUserMongo(mongo.GetDatabase())
 	if err != nil {
 		return nil, err

pkg/common/cachekey/black.go

@@ -1,15 +0,0 @@
-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

@@ -1,44 +0,0 @@
-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

@@ -1,24 +0,0 @@
-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

@@ -1,45 +0,0 @@
-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

@@ -1,14 +0,0 @@
-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,7 +16,6 @@ package config
 
 import (
 	"bytes"
-	"time"
 
 	"github.com/OpenIMSDK/tools/discoveryregistry"
 	"gopkg.in/yaml.v3"
@@ -259,8 +258,6 @@ type configStruct struct {
 		FriendVerify *bool `yaml:"friendVerify"`
 	} `yaml:"messageVerify"`
 
-	LocalCache localCache `yaml:"localCache"`
-
 	IOSPush struct {
 		PushSound  string `yaml:"pushSound"`
 		BadgeCount bool   `yaml:"badgeCount"`
@@ -373,33 +370,6 @@ 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,8 +16,6 @@ package config
 
 import (
 	_ "embed"
-	"fmt"
-	"gopkg.in/yaml.v3"
 	"reflect"
 	"testing"
 
@@ -117,13 +115,3 @@ 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,7 +17,6 @@ package convert
 import (
 	"context"
 	"fmt"
-
 	"github.com/OpenIMSDK/protocol/sdkws"
 	"github.com/OpenIMSDK/tools/utils"
 

pkg/common/db/cache/black.go

@@ -16,9 +16,6 @@ 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"
@@ -55,15 +52,11 @@ 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:  mc,
+		metaCache:  NewMetaCacheRedis(rcClient),
 		blackDB:    blackDB,
 	}
 }
@@ -73,12 +66,12 @@ func (b *BlackCacheRedis) NewCache() BlackCache {
 		expireTime: b.expireTime,
 		rcClient:   b.rcClient,
 		blackDB:    b.blackDB,
-		metaCache:  b.Copy(),
+		metaCache:  NewMetaCacheRedis(b.rcClient, b.metaCache.GetPreDelKeys()...),
 	}
 }
 
 func (b *BlackCacheRedis) getBlackIDsKey(ownerUserID string) string {
-	return cachekey.GetBlackIDsKey(ownerUserID)
+	return blackIDsKey + ownerUserID
 }
 
 func (b *BlackCacheRedis) GetBlackIDs(ctx context.Context, userID string) (blackIDs []string, err error) {

pkg/common/db/cache/config.go

@@ -1,66 +0,0 @@
-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,9 +17,6 @@ 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"
@@ -33,14 +30,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
 )
@@ -87,14 +84,10 @@ 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:      mc,
+		metaCache:      NewMetaCacheRedis(rcClient),
 		conversationDB: db,
 		expireTime:     conversationExpireTime,
 	}
@@ -125,42 +118,38 @@ type ConversationRedisCache struct {
 func (c *ConversationRedisCache) NewCache() ConversationCache {
 	return &ConversationRedisCache{
 		rcClient:       c.rcClient,
-		metaCache:      c.Copy(),
+		metaCache:      NewMetaCacheRedis(c.rcClient, c.metaCache.GetPreDelKeys()...),
 		conversationDB: c.conversationDB,
 		expireTime:     c.expireTime,
 	}
 }
 
 func (c *ConversationRedisCache) getConversationKey(ownerUserID, conversationID string) string {
-	return cachekey.GetConversationKey(ownerUserID, conversationID)
+	return conversationKey + ownerUserID + ":" + conversationID
 }
 
 func (c *ConversationRedisCache) getConversationIDsKey(ownerUserID string) string {
-	return cachekey.GetConversationIDsKey(ownerUserID)
+	return conversationIDsKey + ownerUserID
 }
 
 func (c *ConversationRedisCache) getSuperGroupRecvNotNotifyUserIDsKey(groupID string) string {
-	return cachekey.GetSuperGroupRecvNotNotifyUserIDsKey(groupID)
+	return superGroupRecvMsgNotNotifyUserIDsKey + groupID
 }
 
 func (c *ConversationRedisCache) getRecvMsgOptKey(ownerUserID, conversationID string) string {
-	return cachekey.GetRecvMsgOptKey(ownerUserID, conversationID)
+	return recvMsgOptKey + ownerUserID + ":" + conversationID
 }
 
 func (c *ConversationRedisCache) getSuperGroupRecvNotNotifyUserIDsHashKey(groupID string) string {
-	return cachekey.GetSuperGroupRecvNotNotifyUserIDsHashKey(groupID)
+	return superGroupRecvMsgNotNotifyUserIDsHashKey + groupID
 }
 
 func (c *ConversationRedisCache) getConversationHasReadSeqKey(ownerUserID, conversationID string) string {
-	return cachekey.GetConversationHasReadSeqKey(ownerUserID, conversationID)
+	return conversationHasReadSeqKey + ownerUserID + ":" + conversationID
 }
 
 func (c *ConversationRedisCache) getConversationNotReceiveMessageUserIDsKey(conversationID string) string {
-	return cachekey.GetConversationNotReceiveMessageUserIDsKey(conversationID)
+	return conversationNotReceiveMessageUserIDsKey + conversationID
-}
-
-func (c *ConversationRedisCache) getUserConversationIDsHashKey(ownerUserID string) string {
-	return cachekey.GetUserConversationIDsHashKey(ownerUserID)
 }
 
 func (c *ConversationRedisCache) GetUserConversationIDs(ctx context.Context, ownerUserID string) ([]string, error) {
@@ -180,6 +169,10 @@ 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,9 +16,6 @@ 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"
@@ -30,10 +27,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.
@@ -63,13 +60,8 @@ 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:  mc,
+		metaCache:  NewMetaCacheRedis(rcClient),
 		friendDB:   friendDB,
 		expireTime: friendExpireTime,
 		rcClient:   rcClient,
@@ -80,7 +72,7 @@ func NewFriendCacheRedis(rdb redis.UniversalClient, friendDB relationtb.FriendMo
 func (f *FriendCacheRedis) NewCache() FriendCache {
 	return &FriendCacheRedis{
 		rcClient:   f.rcClient,
-		metaCache:  f.Copy(),
+		metaCache:  NewMetaCacheRedis(f.rcClient, f.metaCache.GetPreDelKeys()...),
 		friendDB:   f.friendDB,
 		expireTime: f.expireTime,
 	}
@@ -88,17 +80,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 cachekey.GetFriendIDsKey(ownerUserID)
+	return friendIDsKey + ownerUserID
 }
 
 // getTwoWayFriendsIDsKey returns the key for storing two-way friend IDs in the cache.
 func (f *FriendCacheRedis) getTwoWayFriendsIDsKey(ownerUserID string) string {
-	return cachekey.GetTwoWayFriendsIDsKey(ownerUserID)
+	return TwoWayFriendsIDsKey + ownerUserID
 }
 
 // getFriendKey returns the key for storing friend info in the cache.
 func (f *FriendCacheRedis) getFriendKey(ownerUserID, friendUserID string) string {
-	return cachekey.GetFriendKey(ownerUserID, friendUserID)
+	return friendKey + ownerUserID + "-" + friendUserID
 }
 
 // GetFriendIDs retrieves friend IDs from the cache or the database if not found.

pkg/common/db/cache/group.go

@@ -17,8 +17,7 @@ package cache
 import (
 	"context"
 	"fmt"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
+	"strconv"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"time"
 
 	"github.com/OpenIMSDK/protocol/constant"
@@ -35,14 +34,15 @@ 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:"
-	//joinedGroupsKey            = "JOIN_GROUPS_KEY:"
+	//groupOwnerInfoKey   = "GROUP_OWNER_INFO:".
-	//groupMemberNumKey          = "GROUP_MEMBER_NUM_CACHE:"
+	joinedGroupsKey            = "JOIN_GROUPS_KEY:"
-	//groupRoleLevelMemberIDsKey = "GROUP_ROLE_LEVEL_MEMBER_IDS:"
+	groupMemberNumKey          = "GROUP_MEMBER_NUM_CACHE:"
+	groupRoleLevelMemberIDsKey = "GROUP_ROLE_LEVEL_MEMBER_IDS:"
 )
 
 type GroupHash interface {
@@ -105,16 +105,12 @@ 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: mc,
+		metaCache: NewMetaCacheRedis(rcClient),
 	}
 }
 
@@ -125,36 +121,36 @@ func (g *GroupCacheRedis) NewCache() GroupCache {
 		groupDB:        g.groupDB,
 		groupMemberDB:  g.groupMemberDB,
 		groupRequestDB: g.groupRequestDB,
-		metaCache:      g.Copy(),
+		metaCache:      NewMetaCacheRedis(g.rcClient, g.metaCache.GetPreDelKeys()...),
 	}
 }
 
 func (g *GroupCacheRedis) getGroupInfoKey(groupID string) string {
-	return cachekey.GetGroupInfoKey(groupID)
+	return groupInfoKey + groupID
 }
 
 func (g *GroupCacheRedis) getJoinedGroupsKey(userID string) string {
-	return cachekey.GetJoinedGroupsKey(userID)
+	return joinedGroupsKey + userID
 }
 
 func (g *GroupCacheRedis) getGroupMembersHashKey(groupID string) string {
-	return cachekey.GetGroupMembersHashKey(groupID)
+	return groupMembersHashKey + groupID
 }
 
 func (g *GroupCacheRedis) getGroupMemberIDsKey(groupID string) string {
-	return cachekey.GetGroupMemberIDsKey(groupID)
+	return groupMemberIDsKey + groupID
 }
 
 func (g *GroupCacheRedis) getGroupMemberInfoKey(groupID, userID string) string {
-	return cachekey.GetGroupMemberInfoKey(groupID, userID)
+	return groupMemberInfoKey + groupID + "-" + userID
 }
 
 func (g *GroupCacheRedis) getGroupMemberNumKey(groupID string) string {
-	return cachekey.GetGroupMemberNumKey(groupID)
+	return groupMemberNumKey + groupID
 }
 
 func (g *GroupCacheRedis) getGroupRoleLevelMemberIDsKey(groupID string, roleLevel int32) string {
-	return cachekey.GetGroupRoleLevelMemberIDsKey(groupID, roleLevel)
+	return groupRoleLevelMemberIDsKey + groupID + "-" + strconv.Itoa(int(roleLevel))
 }
 
 func (g *GroupCacheRedis) GetGroupIndex(group *relationtb.GroupModel, keys []string) (int, error) {

pkg/common/db/cache/meta_cache.go

@@ -18,7 +18,6 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"github.com/redis/go-redis/v9"
 	"time"
 
 	"github.com/OpenIMSDK/tools/mw/specialerror"
@@ -45,9 +44,6 @@ 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 {
@@ -55,36 +51,10 @@ 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 {
@@ -92,7 +62,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", "topic", m.topic, "keys", m.keys)
+		log.ZDebug(ctx, "delete cache", "keys", m.keys)
 		for _, key := range m.keys {
 			for i := 0; i < m.maxRetryTimes; i++ {
 				if err := m.rcClient.TagAsDeleted(key); err != nil {
@@ -102,18 +72,31 @@ func (m *metaCacheRedis) ExecDel(ctx context.Context, distinct ...bool) error {
 				}
 				break
 			}
-		}
+
-		if pk := getPublishKey(m.topic, m.keys); len(pk) > 0 {
+			//retryTimes := 0
-			data, err := json.Marshal(pk)
+			//for {
-			if err != nil {
+			//	m.rcClient.TagAsDeleted()
-				log.ZError(ctx, "keys json marshal failed", err, "topic", m.topic, "keys", pk)
+			//	if err := m.rcClient.TagAsDeletedBatch2(ctx, []string{key}); err != nil {
-			} else {
+			//		if retryTimes >= m.maxRetryTimes {
-				if err := m.redisClient.Publish(ctx, m.topic, string(data)).Err(); err != nil {
+			//			err = errs.ErrInternalServer.Wrap(
-					log.ZError(ctx, "redis publish cache delete error", err, "topic", m.topic, "keys", pk)
+			//				fmt.Sprintf(
-				}
+			//					"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,7 +17,6 @@ package cache
 import (
 	"context"
 	"errors"
-	"github.com/dtm-labs/rockscache"
 	"strconv"
 	"time"
 
@@ -45,12 +44,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:"
@@ -129,8 +128,7 @@ type MsgModel interface {
 }
 
 func NewMsgCacheModel(client redis.UniversalClient) MsgModel {
-	rcClient := rockscache.NewClient(client, rockscache.NewDefaultOptions())
+	return &msgCache{rdb: client}
-	return &msgCache{metaCache: NewMetaCacheRedis(rcClient), rdb: client}
 }
 
 type msgCache struct {
@@ -150,10 +148,6 @@ 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())
 }
@@ -215,6 +209,10 @@ 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,8 +18,6 @@ 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"
@@ -38,8 +36,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
@@ -74,11 +72,7 @@ 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),
@@ -91,7 +85,7 @@ func NewUserCacheRedis(
 func (u *UserCacheRedis) NewCache() UserCache {
 	return &UserCacheRedis{
 		rdb:        u.rdb,
-		metaCache:  u.Copy(),
+		metaCache:  NewMetaCacheRedis(u.rcClient, u.metaCache.GetPreDelKeys()...),
 		userDB:     u.userDB,
 		expireTime: u.expireTime,
 		rcClient:   u.rcClient,
@@ -99,17 +93,18 @@ func (u *UserCacheRedis) NewCache() UserCache {
 }
 
 func (u *UserCacheRedis) getUserInfoKey(userID string) string {
-	return cachekey.GetUserInfoKey(userID)
+	return userInfoKey + userID
 }
 
 func (u *UserCacheRedis) getUserGlobalRecvMsgOptKey(userID string) string {
-	return cachekey.GetUserGlobalRecvMsgOptKey(userID)
+	return userGlobalRecvMsgOptKey + 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,14 +190,7 @@ 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

@@ -0,0 +1,87 @@
+// 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

@@ -0,0 +1,15 @@
+// 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

@@ -0,0 +1,78 @@
+// 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

@@ -0,0 +1,15 @@
+// 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,7 +16,6 @@ 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,11 +16,10 @@ 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"
@@ -90,15 +89,7 @@ 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 int64 = 1024 * 1024 * 1        // 1MB
+	minPartSize = 1024 * 1024 * 1        // 1MB
-	maxPartSize int64 = 1024 * 1024 * 1024 * 5 // 5GB
+	maxPartSize = 1024 * 1024 * 1024 * 5 // 5GB
-	maxNumSize  int64 = 1000
+	maxNumSize  = 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("COS size must be less than the maximum allowed limit")
+		return 0, fmt.Errorf("size must be less than %db", maxPartSize*maxNumSize)
 	}
 	if size <= minPartSize*maxNumSize {
 		return minPartSize, nil

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

@@ -45,9 +45,9 @@ const (
 )
 
 const (
-	minPartSize int64 = 1024 * 1024 * 5        // 1MB
+	minPartSize = 1024 * 1024 * 5        // 1MB
-	maxPartSize int64 = 1024 * 1024 * 1024 * 5 // 5GB
+	maxPartSize = 1024 * 1024 * 1024 * 5 // 5GB
-	maxNumSize  int64 = 10000
+	maxNumSize  = 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("MINIO size must be less than the maximum allowed limit")
+		return 0, fmt.Errorf("size must be less than %db", maxPartSize*maxNumSize)
 	}
 	if size <= minPartSize*maxNumSize {
 		return minPartSize, nil

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

@@ -37,9 +37,9 @@ import (
 )
 
 const (
-	minPartSize int64 = 1024 * 1024 * 1        // 1MB
+	minPartSize = 1024 * 1024 * 1        // 1MB
-	maxPartSize int64 = 1024 * 1024 * 1024 * 5 // 5GB
+	maxPartSize = 1024 * 1024 * 1024 * 5 // 5GB
-	maxNumSize  int64 = 10000
+	maxNumSize  = 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("OSS size must be less than the maximum allowed limit")
+		return 0, fmt.Errorf("size must be less than %db", maxPartSize*maxNumSize)
 	}
 	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  int64 `json:"maxNumSize"`
+	MaxNumSize  int   `json:"maxNumSize"`
 }
 
 type InitiateMultipartUploadResult struct {

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

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

pkg/common/discoveryregister/direct/directResolver.go

@@ -1,81 +0,0 @@
-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

@@ -1,154 +0,0 @@
-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,7 +16,6 @@ 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"
@@ -37,8 +36,6 @@ 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,7 +40,6 @@ 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,17 +18,15 @@ 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"
 )
 
@@ -104,7 +102,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"
@@ -132,40 +130,20 @@ 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
 	}
 }
@@ -188,7 +166,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

@@ -1,16 +0,0 @@
-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

@@ -1,34 +0,0 @@
-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()), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
+	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()))
 	registerIP, err := network.GetRpcRegisterIP(config.Config.Rpc.RegisterIP)
 	if err != nil {
 		return err

pkg/localcache/cache.go

@@ -1,112 +0,0 @@
-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

@@ -1,79 +0,0 @@
-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

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

pkg/localcache/link/link.go

@@ -1,109 +0,0 @@
-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

@@ -1,20 +0,0 @@
-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

@@ -1,20 +0,0 @@
-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

@@ -1,78 +0,0 @@
-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() {
-}