fix: fix auto gen config

Signed-off-by: Xinwei Xiong(cubxxw) <3293172751nss@gmail.com>

.env

@@ -1,308 +0,0 @@
-# Copyright © 2023 OpenIM. All rights reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-# ======================================
-# ========= Basic Configuration ========
-# ======================================
-
-# The user for authentication or system operations.
-# Default: USER=root
-USER=root
-
-# Password associated with the specified user for authentication.
-# Default: PASSWORD=openIM123
-PASSWORD=openIM123
-
-# Endpoint for the MinIO object storage service.
-# Default: MINIO_ENDPOINT=http://172.28.0.1:10005
-MINIO_ENDPOINT=http://172.28.0.1:10005
-
-# Base URL for the application programming interface (API).
-# Default: API_URL=http://172.28.0.1:10002
-API_URL=http://172.28.0.1:10002
-
-# Directory path for storing data files or related information.
-# Default: DATA_DIR=./
-DATA_DIR=./
-
-# 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
-
-# ======================================
-# ========= Network Configuration ======
-# ======================================
-
-# Subnet for the Docker network.
-# Default: DOCKER_BRIDGE_SUBNET=172.28.0.0/16
-DOCKER_BRIDGE_SUBNET=172.28.0.0/16
-
-# Gateway for the Docker network.
-# Default: DOCKER_BRIDGE_GATEWAY=172.28.0.1
-DOCKER_BRIDGE_GATEWAY=172.28.0.1
-
-# Address or hostname for the MySQL network.
-# Default: MYSQL_NETWORK_ADDRESS=172.28.0.2
-MYSQL_NETWORK_ADDRESS=172.28.0.2
-
-# Address or hostname for the MongoDB network.
-# Default: MONGO_NETWORK_ADDRESS=172.28.0.3
-MONGO_NETWORK_ADDRESS=172.28.0.3
-
-# Address or hostname for the Redis network.
-# Default: REDIS_NETWORK_ADDRESS=172.28.0.4
-REDIS_NETWORK_ADDRESS=172.28.0.4
-
-# Address or hostname for the Kafka network.
-# Default: KAFKA_NETWORK_ADDRESS=172.28.0.5
-KAFKA_NETWORK_ADDRESS=172.28.0.5
-
-# Address or hostname for the ZooKeeper network.
-# Default: ZOOKEEPER_NETWORK_ADDRESS=172.28.0.6
-ZOOKEEPER_NETWORK_ADDRESS=172.28.0.6
-
-# Address or hostname for the MinIO network.
-# Default: MINIO_NETWORK_ADDRESS=172.28.0.7
-MINIO_NETWORK_ADDRESS=172.28.0.7
-
-# Address or hostname for the OpenIM web network.
-# Default: OPENIM_WEB_NETWORK_ADDRESS=172.28.0.8
-OPENIM_WEB_NETWORK_ADDRESS=172.28.0.8
-
-# Address or hostname for the OpenIM server network.
-# Default: OPENIM_SERVER_NETWORK_ADDRESS=172.28.0.9
-OPENIM_SERVER_NETWORK_ADDRESS=172.28.0.9
-
-# Address or hostname for the OpenIM chat network.
-# Default: OPENIM_CHAT_NETWORK_ADDRESS=172.28.0.10
-OPENIM_CHAT_NETWORK_ADDRESS=172.28.0.10
-
-# Address or hostname for the Prometheus network.
-# Default: PROMETHEUS_NETWORK_ADDRESS=172.28.0.11
-PROMETHEUS_NETWORK_ADDRESS=172.28.0.11
-
-# Address or hostname for the Grafana network.
-# Default: GRAFANA_NETWORK_ADDRESS=172.28.0.12
-GRAFANA_NETWORK_ADDRESS=172.28.0.12
-  
-# Address or hostname for the node_exporter network.
-# Default: NODE_EXPORTER_NETWORK_ADDRESS=172.28.0.13
-NODE_EXPORTER_NETWORK_ADDRESS=172.28.0.13
-  
-# Address or hostname for the OpenIM admin network.
-# Default: OPENIM_ADMIN_NETWORK_ADDRESS=172.28.0.14
-OPENIM_ADMIN_FRONT_NETWORK_ADDRESS=172.28.0.14
-
-# ===============================================
-# = Component Extension Configuration =
-# ===============================================
-
-# ============ Component Extension Configuration ==========
-# ----- ZooKeeper Configuration -----
-# Address or hostname for the ZooKeeper service.
-# Default: ZOOKEEPER_ADDRESS=172.28.0.1
-ZOOKEEPER_ADDRESS=172.28.0.6
-
-# Port for ZooKeeper service.
-# Default: ZOOKEEPER_PORT=12181
-ZOOKEEPER_PORT=12181
-
-# ----- MySQL Configuration -----
-
-# Address or hostname for the MySQL service.
-# Default: MYSQL_ADDRESS=172.28.0.1
-MYSQL_ADDRESS=172.28.0.2
-
-# Port on which MySQL database service is running.
-# Default: MYSQL_PORT=13306
-MYSQL_PORT=13306
-
-# Password to authenticate with the MySQL database service.
-# Default: MYSQL_PASSWORD=openIM123
-MYSQL_PASSWORD=openIM123
-
-# ----- MongoDB Configuration -----
-# Address or hostname for the MongoDB service.
-# Default: MONGO_ADDRESS=172.28.0.1
-MONGO_ADDRESS=172.28.0.3
-
-# Port on which MongoDB service is running.
-# Default: MONGO_PORT=37017
-MONGO_PORT=37017
-
-# 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
-
-# Name of the database in MongoDB to be used.
-# 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.4
-
-# Port on which Redis in-memory data structure store is running.
-# Default: REDIS_PORT=16379
-REDIS_PORT=16379
-
-# Password to authenticate with the Redis service.
-# 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.5
-
-# Port on which Kafka distributed streaming platform is running.
-# Default: KAFKA_PORT=19092
-KAFKA_PORT=19094
-
-# Topic in Kafka for storing the latest messages in Redis.
-# Default: KAFKA_LATESTMSG_REDIS_TOPIC=latestMsgToRedis
-KAFKA_LATESTMSG_REDIS_TOPIC=latestMsgToRedis
-
-# Topic in Kafka for pushing messages (e.g. notifications or updates).
-# Default: KAFKA_MSG_PUSH_TOPIC=msgToPush
-KAFKA_MSG_PUSH_TOPIC=msgToPush
-
-# Topic in Kafka for storing offline messages in MongoDB.
-# 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.7
-
-# Port on which MinIO object storage service is running.
-# Default: MINIO_PORT=10005
-MINIO_PORT=10005
-
-# Access key to authenticate with the MinIO service.
-# Default: MINIO_ACCESS_KEY=root
-MINIO_ACCESS_KEY=root
-
-# Secret key corresponding to the access key for MinIO authentication.
-# Default: MINIO_SECRET_KEY=openIM123
-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.11
-
-# 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.12
-
-# Port on which Grafana service is running.
-# Default: GRAFANA_PORT=3000
-GRAFANA_PORT=3000
-
-# ======================================
-# ============ OpenIM Web ===============
-# ======================================
-
-# Path to the OpenIM web distribution.
-# Default: OPENIM_WEB_DIST_PATH=/app/dist
-OPENIM_WEB_DIST_PATH=/app/dist
-
-# Port on which OpenIM web service is running.
-# Default: OPENIM_WEB_PORT=11001
-OPENIM_WEB_PORT=11001
-
-# Address or hostname for the OpenIM web service.
-# Default: OPENIM_WEB_ADDRESS=172.28.0.1
-OPENIM_WEB_ADDRESS=172.28.0.8
-
-# ======================================
-# ========= OpenIM Server ==============
-# ======================================
-
-# Address or hostname for the OpenIM server.
-# Default: OPENIM_SERVER_ADDRESS=172.28.0.1
-OPENIM_SERVER_ADDRESS=172.28.0.9
-
-# Port for the OpenIM WebSockets.
-# Default: OPENIM_WS_PORT=10001
-OPENIM_WS_PORT=10001
-
-# Port for the OpenIM API.
-# Default: API_OPENIM_PORT=10002
-API_OPENIM_PORT=10002
-
-
-# ======================================
-# ========== OpenIM Chat ===============
-# ======================================
-
-# Branch name for OpenIM chat.
-# Default: CHAT_BRANCH=main
-CHAT_BRANCH=main
-
-# Address or hostname for the OpenIM chat service.
-# Default: OPENIM_CHAT_ADDRESS=172.28.0.1
-OPENIM_CHAT_ADDRESS=172.28.0.10
-
-# Port for the OpenIM chat API.
-# Default: OPENIM_CHAT_API_PORT=10008
-OPENIM_CHAT_API_PORT=10008
-
-# Directory path for storing data files or related information for OpenIM chat.
-# Default: OPENIM_CHAT_DATA_DIR=./openim-chat/main
-OPENIM_CHAT_DATA_DIR=./openim-chat/main
-
-
-# ======================================
-# ========== OpenIM Admin ==============
-# ======================================
-
-# Branch name for OpenIM server.
-# Default: SERVER_BRANCH=main
-SERVER_BRANCH=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
-NODE_EXPORTER_PORT=19100
-  
-# Port for the prometheus.
-# Default: PROMETHEUS_PORT=19090
-PROMETHEUS_PORT=19090
-  
-# Port for the grafana.
-# Default: GRAFANA_PORT=3000
-GRAFANA_PORT=3000
-  
-# Port for the admin front.
-# Default: OPENIM_ADMIN_FRONT_PORT=11002
-OPENIM_ADMIN_FRONT_PORT=11002

Dockerfile

@@ -25,7 +25,8 @@ WORKDIR ${SERVER_WORKDIR}
 
 # Copy scripts and binary files to the production image
 COPY --from=builder ${OPENIM_SERVER_BINDIR} /openim/openim-server/_output/bin
-# COPY --from=builder ${OPENIM_SERVER_CMDDIR} /openim/openim-server/scripts
+COPY --from=builder ${OPENIM_SERVER_CMDDIR} /openim/openim-server/scripts
-# COPY --from=builder ${SERVER_WORKDIR}/config /openim/openim-server/config
+COPY --from=builder ${SERVER_WORKDIR}/config /openim/openim-server/config
+COPY --from=builder ${SERVER_WORKDIR}/deployments /openim/openim-server/deployments
 
 CMD ["/openim/openim-server/scripts/docker-start-all.sh"]

cmd/openim-api/main.go

@@ -18,11 +18,13 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"net/http"
 	_ "net/http/pprof"
+	"os"
+	"os/signal"
 	"strconv"
-
+	"syscall"
-	ginProm "github.com/openimsdk/open-im-server/v3/pkg/common/ginprometheus"
+	"time"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 
 	"github.com/OpenIMSDK/protocol/constant"
 	"github.com/OpenIMSDK/tools/discoveryregistry"
@@ -33,6 +35,8 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
 	kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
+	ginProm "github.com/openimsdk/open-im-server/v3/pkg/common/ginprometheus"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 )
 
 func main() {
@@ -51,13 +55,12 @@ func run(port int, proPort int) error {
 	if port == 0 || proPort == 0 {
 		err := "port or proPort is empty:" + strconv.Itoa(port) + "," + strconv.Itoa(proPort)
 		log.ZError(context.Background(), err, nil)
-
 		return fmt.Errorf(err)
 	}
+
 	rdb, err := cache.NewRedis()
 	if err != nil {
 		log.ZError(context.Background(), "Failed to initialize Redis", err)
-
 		return err
 	}
 	log.ZInfo(context.Background(), "api start init discov client")
@@ -68,30 +71,29 @@ func run(port int, proPort int) error {
 	client, err = kdisc.NewDiscoveryRegister(config.Config.Envs.Discovery)
 	if err != nil {
 		log.ZError(context.Background(), "Failed to initialize discovery register", err)
-
 		return err
 	}
+
 	if err = client.CreateRpcRootNodes(config.Config.GetServiceNames()); err != nil {
 		log.ZError(context.Background(), "Failed to create RPC root nodes", err)
-
 		return err
 	}
+
 	log.ZInfo(context.Background(), "api register public config to discov")
 	if err = client.RegisterConf2Registry(constant.OpenIMCommonConfigKey, config.Config.EncodeConfig()); err != nil {
 		log.ZError(context.Background(), "Failed to register public config to discov", err)
-
 		return err
 	}
+
 	log.ZInfo(context.Background(), "api register public config to discov success")
 	router := api.NewGinRouter(client, rdb)
-	//////////////////////////////
 	if config.Config.Prometheus.Enable {
 		p := ginProm.NewPrometheus("app", prommetrics.GetGinCusMetrics("Api"))
 		p.SetListenAddress(fmt.Sprintf(":%d", proPort))
 		p.Use(router)
 	}
-	/////////////////////////////////
 	log.ZInfo(context.Background(), "api init router success")
+
 	var address string
 	if config.Config.Api.ListenIP != "" {
 		address = net.JoinHostPort(config.Config.Api.ListenIP, strconv.Itoa(port))
@@ -100,10 +102,25 @@ func run(port int, proPort int) error {
 	}
 	log.ZInfo(context.Background(), "start api server", "address", address, "OpenIM version", config.Version)
 
-	err = router.Run(address)
+	server := http.Server{Addr: address, Handler: router}
-	if err != nil {
+	go func() {
-		log.ZError(context.Background(), "api run failed", err, "address", address)
+		err = server.ListenAndServe()
+		if err != nil && err != http.ErrServerClosed {
+			log.ZError(context.Background(), "api run failed", err, "address", address)
+			os.Exit(1)
+		}
+	}()
 
+	sigs := make(chan os.Signal, 1)
+	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
+	<-sigs
+
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+	defer cancel()
+
+	// graceful shutdown operation.
+	if err := server.Shutdown(ctx); err != nil {
+		log.ZError(context.Background(), "failed to api-server shutdown", err)
 		return err
 	}
 

cmd/openim-msggateway/main.go

@@ -23,6 +23,7 @@ func main() {
 	msgGatewayCmd.AddWsPortFlag()
 	msgGatewayCmd.AddPortFlag()
 	msgGatewayCmd.AddPrometheusPortFlag()
+
 	if err := msgGatewayCmd.Exec(); err != nil {
 		panic(err.Error())
 	}

config/alertmanager.yml

@@ -1,32 +0,0 @@
-###################### AlertManager Configuration ######################
-# AlertManager configuration using environment variables
-#
-# Resolve timeout
-# SMTP configuration for sending alerts
-# Templates for email notifications
-# Routing configurations for alerts
-# Receiver configurations
-global:
-  resolve_timeout: 5m
-  smtp_from: alert@openim.io
-  smtp_smarthost: smtp.163.com:465
-  smtp_auth_username: alert@openim.io
-  smtp_auth_password: YOURAUTHPASSWORD
-  smtp_require_tls: false
-  smtp_hello: xxx监控告警
-
-templates:
-  - /etc/alertmanager/email.tmpl
-
-route:
-  group_wait: 5s
-  group_interval: 5s
-  repeat_interval: 5m
-  receiver: email
-receivers:
-  - name: email
-    email_configs:
-      - to: {EMAIL_TO:-'alert@example.com'}
-        html: '{{ template "email.to.html" . }}'
-        headers: { Subject: "[OPENIM-SERVER]Alarm" }
-        send_resolved: true

config/config.yaml

@@ -1,398 +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.
-
-# -----------------------------------------------------------------
-# TODO: This config file is the template file
-# --| source: deployments/templates/openim.yaml
-# --| env: scripts/install/environment
-# --| target: config/config.yaml
-# -----------------------------------------------------------------
-
-envs:
-  discovery: zookeeper
-
-###################### Zookeeper ######################
-# Zookeeper configuration
-# It's not recommended to modify the schema
-# 
-# Zookeeper address
-# Zookeeper username
-# Zookeeper password
-zookeeper:
-  schema: openim
-  address: [ 172.28.0.1:12181 ]
-  username: ''
-  password: ''
-
-###################### Mysql ######################
-# MySQL configuration
-# Currently, only single machine setup is supported
-#
-# Maximum number of open connections
-# Maximum number of idle connections
-# Maximum lifetime in seconds a connection can be reused
-# Log level: 1=slient, 2=error, 3=warn, 4=info
-# Slow query threshold in milliseconds
-mysql:
-  address: [ 172.28.0.1:13306 ]
-  username: root
-  password: openIM123
-  database: openIM_v3
-  maxOpenConn: 1000
-  maxIdleConn: 100
-  maxLifeTime: 60
-  logLevel: 4
-  slowThreshold: 500
-
-###################### Mongo ######################
-# MongoDB configuration
-# If uri is not empty, it will be used directly
-#
-# MongoDB address for standalone setup, Mongos address for sharded cluster setup
-# Default MongoDB database name
-# Maximum connection pool size
-mongo:
-  uri: ''
-  address: [ 172.28.0.1:37017 ]
-  database: openIM_v3
-  username: root
-  password: openIM123
-  maxPoolSize: 100
-
-###################### Redis configuration information ######################
-# Redis configuration
-#
-# Username is required only for Redis version 6.0+
-redis:
-  address: [ 172.28.0.1:16379 ]
-  username: ''
-  password: openIM123
-
-###################### Kafka configuration information ######################
-# Kafka configuration
-#
-# Kafka username
-# Kafka password
-# It's not recommended to modify this topic name
-# Consumer group ID, it's not recommended to modify
-kafka:
-  username: ''
-  password: ''
-  addr: [ 172.28.0.1:19094 ]
-  latestMsgToRedis:
-    topic: "latestMsgToRedis"
-  offlineMsgToMongo:
-    topic: "offlineMsgToMongoMysql"
-  msgToPush:
-    topic: "msgToPush"
-  consumerGroupID:
-    msgToRedis: redis
-    msgToMongo: mongo
-    msgToMySql: mysql
-    msgToPush: push
-
-###################### RPC configuration information ######################
-# RPC configuration
-#
-# IP address to register with zookeeper when starting RPC, the IP and corresponding rpcPort should be accessible by api/gateway
-# Default listen IP is 0.0.0.0
-rpc:
-  registerIP: ''
-  listenIP: 0.0.0.0
-
-###################### API configuration information ######################
-# API configuration
-#
-# API service port
-# Default listen IP is 0.0.0.0
-api:
-  openImApiPort: [ 10002 ]
-  listenIP: 0.0.0.0
-
-###################### Object configuration information ######################
-# Object storage configuration
-#
-# Use minio for object storage
-# API URL should be accessible by the app
-# It's not recommended to modify the bucket name
-# Endpoint should be accessible by the app
-# Session token
-# Configuration for Tencent COS
-# Configuration for Aliyun OSS
-# apiURL is the address of the api, the access address of the app, use s3 must be configured
-# minio.endpoint can be configured as an intranet address,
-# minio.signEndpoint is minio public network address
-object:
-  enable: "minio"
-  apiURL: "http://127.0.0.1:10002"
-  minio:
-    bucket: "openim"
-    endpoint: "http://172.28.0.1:10005"
-    accessKeyID: "root"
-    secretAccessKey: "openIM123"
-    sessionToken: ''
-    signEndpoint: "http://127.0.0.1:10005"
-    publicRead: false
-  cos:
-    bucketURL: https://temp-1252357374.cos.ap-chengdu.myqcloud.com
-    secretID: ''
-    secretKey: ''
-    sessionToken: ''
-    publicRead: false
-  oss:
-    endpoint: "https://oss-cn-chengdu.aliyuncs.com"
-    bucket: "demo-9999999"
-    bucketURL: "https://demo-9999999.oss-cn-chengdu.aliyuncs.com"
-    accessKeyID: ''
-    accessKeySecret: ''
-    sessionToken: ''
-    publicRead: false
-
-###################### RPC Port Configuration ######################
-# RPC service ports
-# These ports are passed into the program by the script and are not recommended to modify
-# For launching multiple programs, just fill in multiple ports separated by commas
-# For example, [10110, 10111]
-rpcPort: 
-  openImUserPort: [ 10110 ]
-  openImFriendPort: [ 10120 ]
-  openImMessagePort: [ 10130 ]
-  openImGroupPort: [ 10150 ]
-  openImAuthPort: [ 10160 ]
-  openImPushPort: [ 10170 ]
-  openImConversationPort: [ 10180 ]
-  openImThirdPort: [ 10190 ]
-
-###################### RPC Register Name Configuration ######################
-# RPC service names for registration, it's not recommended to modify these
-rpcRegisterName:
-  openImUserName: User
-  openImFriendName: Friend
-  openImMsgName: Msg
-  openImPushName: Push
-  openImMessageGatewayName: MessageGateway
-  openImGroupName: Group
-  openImAuthName: Auth
-  openImConversationName: Conversation
-  openImThirdName: Third
-
-###################### Log Configuration ######################
-# Log configuration
-#
-# Storage directory
-# Log rotation time
-# Maximum number of logs to retain
-# Log level, 6 means all levels
-# Whether to output to stdout
-# Whether to output in json format
-# Whether to include stack trace in logs
-log:
-  storageLocation: ../logs/
-  rotationTime: 24
-  remainRotationCount: 2
-  remainLogLevel: 6
-  isStdout: false
-  isJson: false
-  withStack: false
-
-###################### Variables definition ######################
-# Long connection server configuration
-#
-# Websocket port for msg_gateway
-# Maximum number of websocket connections
-# Maximum length of websocket request package
-# Websocket connection handshake timeout
-longConnSvr:
-  openImWsPort: [ 10001 ]
-  websocketMaxConnNum: 100000
-  openImMessageGatewayPort: [ 10140 ]
-  websocketMaxMsgLen: 4096
-  websocketTimeout: 10
-
-# Push notification service configuration
-#
-# Use GeTui for push notifications
-# GeTui offline push configuration
-# FCM offline push configuration
-# Account file, place it in the config directory
-# JPush configuration, modify these after applying in JPush backend
-push:
-  enable: getui
-  geTui:
-    pushUrl: "https://restapi.getui.com/v2/$appId"
-    masterSecret: ''
-    appKey: ''
-    intent: ''
-    channelID: ''
-    channelName: ''
-  fcm:
-    serviceAccount: "x.json"
-  jpns:
-    appKey: ''
-    masterSecret: ''
-    pushUrl: ''
-    pushIntent: ''
-
-# App manager configuration
-#
-# Built-in app manager user IDs
-# Built-in app manager nicknames
-manager:
-  userID: [ "openIM123456", "openIM654321", "openIMAdmin" ]
-  nickname: [ "system1", "system2", "system3" ]
-
-# Multi-platform login policy
-# For each platform(Android, iOS, Windows, Mac, web), only one can be online at a time
-multiLoginPolicy: 1
-
-# Whether to store messages in MySQL, messages in MySQL are only used for management background
-chatPersistenceMysql: true
-
-# Message cache timeout in seconds, it's not recommended to modify
-msgCacheTimeout: 86400
-
-# Whether to enable read receipts for group chat
-groupMessageHasReadReceiptEnable: true
-
-# Whether to enable read receipts for single chat
-singleMessageHasReadReceiptEnable: true
-
-# MongoDB offline message retention period in days
-retainChatRecords: 365
-
-# Schedule to clear expired messages(older than retainChatRecords days) in MongoDB every Wednesday at 2am
-# This deletion is just for cleaning up disk usage according to previous configuration retainChatRecords, no notification will be sent
-chatRecordsClearTime: "0 2 * * 3"
-
-# Schedule to auto delete messages every day at 2am
-# This deletion is for messages that have been retained for more than msg_destruct_time (seconds) in the conversation field
-msgDestructTime: "0 2 * * *"
-
-# Secret key
-secret: openIM123
-
-# Token policy
-#
-# Token expiration period in days
-tokenPolicy:
-  expire: 90
-
-# Message verification policy
-#
-# Whether to verify friendship when sending messages
-messageVerify:
-  friendVerify: false
-
-# iOS push notification configuration
-#
-# iOS push notification sound
-# Whether to count badge
-# Whether it's production environment
-iosPush:
-  pushSound: "xxx"
-  badgeCount: true
-  production: false
-
-###################### Third-party service configuration ######################
-# Callback configuration
-#
-# Callback URL
-# Whether to enable this callback event
-# Timeout in seconds
-# Whether to continue execution if callback fails
-callback:
-  url:
-  beforeSendSingleMsg:
-    enable: false
-    timeout: 5
-    failedContinue: true
-  afterSendSingleMsg:
-    enable: false
-    timeout: 5
-  beforeSendGroupMsg:
-    enable: false
-    timeout: 5
-    failedContinue: true
-  afterSendGroupMsg:
-    enable: false
-    timeout: 5
-  msgModify:
-    enable: false
-    timeout: 5
-    failedContinue: true
-  userOnline:
-    enable: false
-    timeout: 5
-  userOffline:
-    enable: false
-    timeout: 5
-  userKickOff:
-    enable: false
-    timeout: 5
-  offlinePush:
-    enable: false
-    timeout: 5
-    failedContinue: true
-  onlinePush:
-    enable: false
-    timeout: 5
-    failedContinue: true
-  superGroupOnlinePush:
-    enable: false
-    timeout: 5
-    failedContinue: true
-  beforeAddFriend:
-    enable: false
-    timeout: 5
-    failedContinue: true
-  beforeUpdateUserInfo:
-    enable: false
-    timeout: 5
-    failedContinue: true
-  beforeCreateGroup:
-    enable: false
-    timeout: 5
-    failedContinue: true
-  beforeMemberJoinGroup:
-    enable: false
-    timeout: 5
-    failedContinue: true
-  beforeSetGroupMemberInfo:
-    enable: false
-    timeout: 5
-    failedContinue: true
-  setMessageReactionExtensions:
-    enable: false
-    timeout: 5
-    failedContinue: true
-
-###################### Prometheus ######################
-# Prometheus configuration for various services
-# The number of Prometheus ports per service needs to correspond to rpcPort
-# The number of ports needs to be consistent with msg_transfer_service_num in script/path_info.sh
-prometheus:
-  enable: false
-  prometheusUrl: "https://openim.prometheus"
-  apiPrometheusPort: [20100]
-  userPrometheusPort: [ 20110 ]
-  friendPrometheusPort: [ 20120 ]
-  messagePrometheusPort: [ 20130 ]
-  messageGatewayPrometheusPort: [ 20140 ]
-  groupPrometheusPort: [ 20150 ]
-  authPrometheusPort: [ 20160 ]
-  pushPrometheusPort: [ 20170 ]
-  conversationPrometheusPort: [ 20230 ]
-  rtcPrometheusPort: [ 21300 ]
-  thirdPrometheusPort: [ 21301 ]
-  messageTransferPrometheusPort: [ 21400, 21401, 21402, 21403 ] # List of ports

config/instance-down-rules.yml

@@ -9,3 +9,14 @@ groups:
         annotations:
           summary: "Instance {{ $labels.instance }} down"
           description: "{{ $labels.instance }} of job {{ $labels.job }} has been down for more than 1 minutes."
+
+  - name: database_insert_failure_alerts
+    rules:
+      - alert: DatabaseInsertFailed
+        expr: (increase(msg_insert_redis_failed_total[5m]) > 0) or (increase(msg_insert_mongo_failed_total[5m]) > 0)
+        for: 1m
+        labels:
+          severity: critical
+        annotations:
+          summary: "Increase in MsgInsertRedisFailedCounter or MsgInsertMongoFailedCounter detected"
+          description: "Either MsgInsertRedisFailedCounter or MsgInsertMongoFailedCounter has increased in the last 5 minutes, indicating failures in message insert operations to Redis or MongoDB,maybe the redis or mongodb is crash."

deployments/README.md

@@ -84,8 +84,8 @@ $ sudo sealos run labring/kubernetes:v1.25.0 labring/helm:v3.8.2 labring/calico:
 If you are local, you can also use Kind and Minikube to test, for example, using Kind:
 
 ```bash
-$ sGO111MODULE="on" go get sigs.k8s.io/kind@v0.11.1
+$ GO111MODULE="on" go get sigs.k8s.io/kind@v0.11.1
-$ skind create cluster
+$ kind create cluster
 ```
 
 ### Installing helm

deployments/templates/alertmanager.yml

@@ -27,7 +27,7 @@ route:
 receivers:
   - name: email
     email_configs:
-      - to: ${ALERTMANAGER_EMAIL_TO}
+      - to: '${ALERTMANAGER_EMAIL_TO}'
         html: '{{ template "email.to.html" . }}'
         headers: { Subject: "[OPENIM-SERVER]Alarm" }
         send_resolved: true

deployments/templates/openim.yaml

@@ -158,6 +158,14 @@ object:
     accessKeySecret: ${OSS_ACCESS_KEY_SECRET}
     sessionToken: ${OSS_SESSION_TOKEN}
     publicRead: ${OSS_PUBLIC_READ}
+  kodo:
+    endpoint: "${KODO_ENDPOINT}"
+    bucket: "${KODO_BUCKET}"
+    bucketURL: "${KODO_BUCKET_URL}"
+    accessKeyID: ${KODO_ACCESS_KEY_ID}
+    accessKeySecret: ${KODO_ACCESS_KEY_SECRET}
+    sessionToken: ${KODO_SESSION_TOKEN}
+    publicRead: ${KODO_PUBLIC_READ}
 
 ###################### RPC Port Configuration ######################
 # RPC service ports
@@ -312,71 +320,191 @@ iosPush:
 # Timeout in seconds
 # Whether to continue execution if callback fails
 callback:
-  url:
+  url: ""
   beforeSendSingleMsg:
-    enable: false
+    enable: ${CALLBACK_ENABLE}
-    timeout: 5
+    timeout: ${CALLBACK_TIMEOUT}
-    failedContinue: true
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
   afterSendSingleMsg:
-    enable: false
+    enable: ${CALLBACK_ENABLE}
-    timeout: 5
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
   beforeSendGroupMsg:
-    enable: false
+    enable: ${CALLBACK_ENABLE}
-    timeout: 5
+    timeout: ${CALLBACK_TIMEOUT}
-    failedContinue: true
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
   afterSendGroupMsg:
-    enable: false
+    enable: ${CALLBACK_ENABLE}
-    timeout: 5
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
   msgModify:
-    enable: false
+    enable: ${CALLBACK_ENABLE}
-    timeout: 5
+    timeout: ${CALLBACK_TIMEOUT}
-    failedContinue: true
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
   userOnline:
-    enable: false
+    enable: ${CALLBACK_ENABLE}
-    timeout: 5
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
   userOffline:
-    enable: false
+    enable: ${CALLBACK_ENABLE}
-    timeout: 5
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
   userKickOff:
-    enable: false
+    enable: ${CALLBACK_ENABLE}
-    timeout: 5
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
   offlinePush:
-    enable: false
+    enable: ${CALLBACK_ENABLE}
-    timeout: 5
+    timeout: ${CALLBACK_TIMEOUT}
-    failedContinue: true
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
   onlinePush:
-    enable: false
+    enable: ${CALLBACK_ENABLE}
-    timeout: 5
+    timeout: ${CALLBACK_TIMEOUT}
-    failedContinue: true
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
   superGroupOnlinePush:
-    enable: false
+    enable: ${CALLBACK_ENABLE}
-    timeout: 5
+    timeout: ${CALLBACK_TIMEOUT}
-    failedContinue: true
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
   beforeAddFriend:
-    enable: false
+    enable: ${CALLBACK_ENABLE}
-    timeout: 5
+    timeout: ${CALLBACK_TIMEOUT}
-    failedContinue: true
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
   beforeUpdateUserInfo:
-    enable: false
+    enable: ${CALLBACK_ENABLE}
-    timeout: 5
+    timeout: ${CALLBACK_TIMEOUT}
-    failedContinue: true
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
   beforeCreateGroup:
-    enable: false
+    enable: ${CALLBACK_ENABLE}
-    timeout: 5
+    timeout: ${CALLBACK_TIMEOUT}
-    failedContinue: true
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  afterCreateGroup:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
   beforeMemberJoinGroup:
-    enable: false
+    enable: ${CALLBACK_ENABLE}
-    timeout: 5
+    timeout: ${CALLBACK_TIMEOUT}
-    failedContinue: true
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
   beforeSetGroupMemberInfo:
-    enable: false
+    enable: ${CALLBACK_ENABLE}
-    timeout: 5
+    timeout: ${CALLBACK_TIMEOUT}
-    failedContinue: true
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  afterSetGroupMemberInfo:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
   setMessageReactionExtensions:
-    enable: false
+    enable: ${CALLBACK_ENABLE}
-    timeout: 5
+    timeout: ${CALLBACK_TIMEOUT}
-    failedContinue: true
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
-
+  quitGroup:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  killGroupMember:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  dismissGroup:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  joinGroup:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  groupMsgRead:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  singleMsgRead:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  updateUserInfo:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  beforeUserRegister:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  afterUserRegister:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  transferGroupOwner:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  beforeSetFriendRemark:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  afterSetFriendRemark:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  afterGroupMsgRead:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  afterGroupMsgRevoke:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  afterJoinGroup:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  beforeInviteUserToGroup:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  joinGroupAfter:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  setGroupInfoAfter:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  setGroupInfoBefore:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  revokeMsgAfter:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  addBlackBefore:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  addFriendAfter:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  addFriendAgreeBefore:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  deleteFriendAfter:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  importFriendsBefore:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  importFriendsAfter:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
+  removeBlackAfter:
+    enable: ${CALLBACK_ENABLE}
+    timeout: ${CALLBACK_TIMEOUT}
+    failedContinue: ${CALLBACK_FAILED_CONTINUE}
 ###################### Prometheus ######################
 # Prometheus configuration for various services
 # The number of Prometheus ports per service needs to correspond to rpcPort

docs/contrib/environment.md

@@ -37,6 +37,7 @@
 	* 2.20. [Prometheus Configuration](#PrometheusConfiguration-1)
 		* 2.20.1. [General Configuration](#GeneralConfiguration)
 		* 2.20.2. [Service-Specific Prometheus Ports](#Service-SpecificPrometheusPorts)
+	* 2.21. [Qiniu Cloud Kodo Configuration](#QiniuCloudKODOConfiguration)
 
 ## 0. <a name='TableofContents'></a>OpenIM Config File
 
@@ -466,7 +467,7 @@ This section involves configuring the log settings, including storage location,
 This section involves setting up additional configuration variables for Websocket, Push Notifications, and Chat.
 
 | Parameter               | Example Value     | Description                        |
-| ----------------------- | ----------------- | ---------------------------------- |
+|-------------------------|-------------------|------------------------------------|
 | WEBSOCKET_MAX_CONN_NUM  | "100000"          | Maximum Websocket connections      |
 | WEBSOCKET_MAX_MSG_LEN   | "4096"            | Maximum Websocket message length   |
 | WEBSOCKET_TIMEOUT       | "10"              | Websocket timeout                  |
@@ -500,9 +501,9 @@ This section involves setting up additional configuration variables for Websocke
 | TOKEN_EXPIRE            | "90"              | Token Expiry Time                  |
 | FRIEND_VERIFY           | "false"           | Friend Verification Enable         |
 | IOS_PUSH_SOUND          | "xxx"             | iOS                                |
-
+| CALLBACK_ENABLE         | "false"            | Enable callback                    | 
-
+| CALLBACK_TIMEOUT        | "5"               | Maximum timeout for callback call  |
-
+| CALLBACK_FAILED_CONTINUE| "true"            | fails to continue to the next step |
 ###  2.20. <a name='PrometheusConfiguration-1'></a>Prometheus Configuration
 
 This section involves configuring Prometheus, including enabling/disabling it and setting up ports for various services.
@@ -528,3 +529,18 @@ This section involves configuring Prometheus, including enabling/disabling it an
 | RTC Service              | `RTC_PROM_PORT`          | '21300'                      | Prometheus port for the RTC service.               |
 | Third Service            | `THIRD_PROM_PORT`        | '21301'                      | Prometheus port for the Third service.             |
 | Message Transfer Service | `MSG_TRANSFER_PROM_PORT` | '21400, 21401, 21402, 21403' | Prometheus ports for the Message Transfer service. |
+
+
+###  2.21. <a name='QiniuCloudKODOConfiguration'></a>Qiniu Cloud Kodo Configuration
+
+This section involves setting up Qiniu Cloud Kodo, including its endpoint, bucket name, and credentials.
+
+| Parameter             | Example Value                                                | Description                              |
+| --------------------- | ------------------------------------------------------------ | ---------------------------------------- |
+| KODO_ENDPOINT          | "[http://s3.cn-east-1.qiniucs.com](http://s3.cn-east-1.qiniucs.com)" | Endpoint URL for Qiniu Cloud Kodo.      |
+| KODO_BUCKET            | "demo-9999999"                                               | Bucket name for Qiniu Cloud Kodo.       |
+| KODO_BUCKET_URL        | "[http://your.domain.com](http://your.domain.com)" | Bucket URL for Qiniu Cloud Kodo.        |
+| KODO_ACCESS_KEY_ID     | [User Defined]                                               | Access key ID for Qiniu Cloud Kodo.     |
+| KODO_ACCESS_KEY_SECRET | [User Defined]                                               | Access key secret for Qiniu Cloud Kodo. |
+| KODO_SESSION_TOKEN     | [User Defined]                                               | Session token for Qiniu Cloud Kodo.     |
+| KODO_PUBLIC_READ       | "false"                                                      | Public read access.                      |

go.mod

@@ -38,7 +38,7 @@ require github.com/google/uuid v1.3.1
 require (
 	github.com/IBM/sarama v1.41.3
 	github.com/OpenIMSDK/protocol v0.0.31
-	github.com/OpenIMSDK/tools v0.0.16
+	github.com/OpenIMSDK/tools v0.0.17
 	github.com/aliyun/aliyun-oss-go-sdk v2.2.9+incompatible
 	github.com/go-redis/redis v6.15.9+incompatible
 	github.com/go-sql-driver/mysql v1.7.1
@@ -58,6 +58,24 @@ require (
 	cloud.google.com/go/iam v1.1.2 // indirect
 	cloud.google.com/go/longrunning v0.5.1 // indirect
 	cloud.google.com/go/storage v1.30.1 // indirect
+	github.com/aws/aws-sdk-go-v2 v1.23.1 // indirect
+	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.5.1 // indirect
+	github.com/aws/aws-sdk-go-v2/config v1.25.4 // indirect
+	github.com/aws/aws-sdk-go-v2/credentials v1.16.3 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.14.5 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.2.4 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.5.4 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.7.1 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/v4a v1.2.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.10.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.2.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.10.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.16.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.43.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.17.3 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.20.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.25.4 // indirect
+	github.com/aws/smithy-go v1.17.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bytedance/sonic v1.9.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
@@ -115,6 +133,7 @@ require (
 	github.com/prometheus/client_model v0.4.1-0.20230718164431-9a2bf3000d16 // indirect
 	github.com/prometheus/common v0.44.0 // indirect
 	github.com/prometheus/procfs v0.11.1 // indirect
+	github.com/qiniu/go-sdk/v7 v7.18.2 // indirect
 	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

go.sum

@@ -20,8 +20,8 @@ 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.31 h1:ax43x9aqA6EKNXNukS5MT5BSTqkUmwO4uTvbJLtzCgE=
 github.com/OpenIMSDK/protocol v0.0.31/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y=
-github.com/OpenIMSDK/tools v0.0.16 h1:te/GIq2imCMsrRPgU9OObYKbzZ3rT08Lih/o+3QFIz0=
+github.com/OpenIMSDK/tools v0.0.17 h1:1E1HUOL2W09YUHBb4wBwrXoTSZm5ONVwLxlEX1GhlKw=
-github.com/OpenIMSDK/tools v0.0.16/go.mod h1:eg+q4A34Qmu73xkY0mt37FHGMCMfC6CtmOnm0kFEGFI=
+github.com/OpenIMSDK/tools v0.0.17/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=
@@ -31,6 +31,42 @@ github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239 h1:kFOfPq6dUM1hTo
 github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239/go.mod h1:2FmKhYUyUczH0OGQWaF5ceTx0UBShxjsH6f8oGKYe2c=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPdPJAN/hZIm0C4OItdklCFmMRWYpio=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs=
+github.com/aws/aws-sdk-go-v2 v1.23.1 h1:qXaFsOOMA+HsZtX8WoCa+gJnbyW7qyFFBlPqvTSzbaI=
+github.com/aws/aws-sdk-go-v2 v1.23.1/go.mod h1:i1XDttT4rnf6vxc9AuskLc6s7XBee8rlLilKlc03uAA=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.5.1 h1:ZY3108YtBNq96jNZTICHxN1gSBSbnvIdYwwqnvCV4Mc=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.5.1/go.mod h1:t8PYl/6LzdAqsU4/9tz28V/kU+asFePvpOMkdul0gEQ=
+github.com/aws/aws-sdk-go-v2/config v1.25.4 h1:r+X1x8QI6FEPdJDWCNBDZHyAcyFwSjHN8q8uuus+Axs=
+github.com/aws/aws-sdk-go-v2/config v1.25.4/go.mod h1:8GTjImECskr7D88P/Nn9uM4M4rLY9i77hLJZgkZEWV8=
+github.com/aws/aws-sdk-go-v2/credentials v1.16.3 h1:8PeI2krzzjDJ5etmgaMiD1JswsrLrWvKKu/uBUtNy1g=
+github.com/aws/aws-sdk-go-v2/credentials v1.16.3/go.mod h1:Kdh/okh+//vQ/AjEt81CjvkTo64+/zIE4OewP7RpfXk=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.14.5 h1:KehRNiVzIfAcj6gw98zotVbb/K67taJE0fkfgM6vzqU=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.14.5/go.mod h1:VhnExhw6uXy9QzetvpXDolo1/hjhx4u9qukBGkuUwjs=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.2.4 h1:LAm3Ycm9HJfbSCd5I+wqC2S9Ej7FPrgr5CQoOljJZcE=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.2.4/go.mod h1:xEhvbJcyUf/31yfGSQBe01fukXwXJ0gxDp7rLfymWE0=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.5.4 h1:4GV0kKZzUxiWxSVpn/9gwR0g21NF1Jsyduzo9rHgC/Q=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.5.4/go.mod h1:dYvTNAggxDZy6y1AF7YDwXsPuHFy/VNEpEI/2dWK9IU=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.7.1 h1:uR9lXYjdPX0xY+NhvaJ4dD8rpSRz5VY81ccIIoNG+lw=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.7.1/go.mod h1:6fQQgfuGmw8Al/3M2IgIllycxV7ZW7WCdVSqfBeUiCY=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.2.4 h1:40Q4X5ebZruRtknEZH/bg91sT5pR853F7/1X9QRbI54=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.2.4/go.mod h1:u77N7eEECzUv7F0xl2gcfK/vzc8wcjWobpy+DcrLJ5E=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.10.1 h1:rpkF4n0CyFcrJUG/rNNohoTmhtWlFTRI4BsZOh9PvLs=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.10.1/go.mod h1:l9ymW25HOqymeU2m1gbUQ3rUIsTwKs8gYHXkqDQUhiI=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.2.4 h1:6DRKQc+9cChgzL5gplRGusI5dBGeiEod4m/pmGbcX48=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.2.4/go.mod h1:s8ORvrW4g4v7IvYKIAoBg17w3GQ+XuwXDXYrQ5SkzU0=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.10.4 h1:rdovz3rEu0vZKbzoMYPTehp0E8veoE9AyfzqCr5Eeao=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.10.4/go.mod h1:aYCGNjyUCUelhofxlZyj63srdxWUSsBSGg5l6MCuXuE=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.16.4 h1:o3DcfCxGDIT20pTbVKVhp3vWXOj/VvgazNJvumWeYW0=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.16.4/go.mod h1:Uy0KVOxuTK2ne+/PKQ+VvEeWmjMMksE17k/2RK/r5oM=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.43.1 h1:1w11lfXOa8HoHoSlNtt4mqv/N3HmDOa+OnUH3Y9DHm8=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.43.1/go.mod h1:dqJ5JBL0clzgHriH35Amx3LRFY6wNIPUX7QO/BerSBo=
+github.com/aws/aws-sdk-go-v2/service/sso v1.17.3 h1:CdsSOGlFF3Pn+koXOIpTtvX7st0IuGsZ8kJqcWMlX54=
+github.com/aws/aws-sdk-go-v2/service/sso v1.17.3/go.mod h1:oA6VjNsLll2eVuUoF2D+CMyORgNzPEW/3PyUdq6WQjI=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.20.1 h1:cbRqFTVnJV+KRpwFl76GJdIZJKKCdTPnjUZ7uWh3pIU=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.20.1/go.mod h1:hHL974p5auvXlZPIjJTblXJpbkfK4klBczlsEaMCGVY=
+github.com/aws/aws-sdk-go-v2/service/sts v1.25.4 h1:yEvZ4neOQ/KpUqyR+X0ycUTW/kVRNR4nDZ38wStHGAA=
+github.com/aws/aws-sdk-go-v2/service/sts v1.25.4/go.mod h1:feTnm2Tk/pJxdX+eooEsxvlvTWBvDm6CasRZ+JOs2IY=
+github.com/aws/smithy-go v1.17.0 h1:wWJD7LX6PBV6etBUwO0zElG0nWN9rUhp0WdYeHSHAaI=
+github.com/aws/smithy-go v1.17.0/go.mod h1:NukqUGpCZIILqqiV0NIjeFh24kd/FAa4beRb6nbIUPE=
 github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
@@ -56,6 +92,7 @@ github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDk
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
 github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7DoTY=
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -91,11 +128,17 @@ github.com/gin-gonic/gin v1.9.1 h1:4idEAncQnU5cB7BeOkPtxjfCSye0AAm1R0RVIqJ+Jmg=
 github.com/gin-gonic/gin v1.9.1/go.mod h1:hPrL7YrpYKXt5YId3A/Tnip5kqbEAP+KLuI3SUcPTeU=
 github.com/gliderlabs/ssh v0.2.2 h1:6zsha5zo/TWhRhwqCD3+EarCAgZ2yN28ipRnGPnwkI0=
 github.com/gliderlabs/ssh v0.2.2/go.mod h1:U7qILu1NlMHj9FlMhZLlkCdDnU1DBEAqr0aevW3Awn0=
+github.com/go-playground/assert/v2 v2.0.1/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
 github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s=
+github.com/go-playground/locales v0.13.0/go.mod h1:taPMhCMXrRLJO55olJkUXHZBHCxTMfnGwq/HNwmWNS8=
+github.com/go-playground/locales v0.14.0/go.mod h1:sawfccIbzZTqEDETgFXqTho0QybSa7l++s0DH+LDiLs=
 github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA=
 github.com/go-playground/locales v0.14.1/go.mod h1:hxrqLVvrK65+Rwrd5Fc6F2O76J/NuW9t0sjnWqG1slY=
+github.com/go-playground/universal-translator v0.17.0/go.mod h1:UkSxE5sNxxRwHyU+Scu5vgOQjsIJAF8j9muTVoKLVtA=
+github.com/go-playground/universal-translator v0.18.0/go.mod h1:UvRDBj+xPUEGrFYl+lu/H90nyDXpg0fqeB/AQUGNTVA=
 github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJnYK9S473LQFuzCbDbfSFY=
 github.com/go-playground/universal-translator v0.18.1/go.mod h1:xekY+UJKNuX9WP91TpwSH2VMlDf28Uj24BCp08ZFTUY=
+github.com/go-playground/validator/v10 v10.8.0/go.mod h1:9JhgTzTaE31GZDpH/HSvHiRJrJ3iKAgqqH0Bl/Ocjdk=
 github.com/go-playground/validator/v10 v10.15.5 h1:LEBecTWb/1j5TNY1YYG2RcOUN3R7NLylN+x8TTueE24=
 github.com/go-playground/validator/v10 v10.15.5/go.mod h1:9iXMNT7sEkjXb0I+enO7QXmzG6QCsPWY4zveKFVRSyU=
 github.com/go-redis/redis v6.15.9+incompatible h1:K0pv1D7EQUjfyoMql+r/jZqCLizCGKFlFgcHWWmHQjg=
@@ -214,11 +257,15 @@ github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa02
 github.com/klauspost/cpuid/v2 v2.2.5 h1:0E5MSMDEoAulmXNFquVs//DdoomxaoTY1kUhbc/qbZg=
 github.com/klauspost/cpuid/v2 v2.2.5/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/pty v1.1.8/go.mod h1:O1sed60cT9XZ5uDucP5qwvh+TE3NnUj51EiZO/lmSfw=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/leodido/go-urn v1.2.1/go.mod h1:zt4jvISO2HfUBqxjfIshjdMTYS56ZS/qv49ictyFfxY=
 github.com/leodido/go-urn v1.2.4 h1:XlAE/cm/ms7TE/VMVoduSpNBoyc2dOxHs5MZSwAN63Q=
 github.com/leodido/go-urn v1.2.4/go.mod h1:7ZrI8mTSeBSHl/UaRyKQW1qZeMgak41ANeCNaVckg+4=
 github.com/lestrrat-go/envload v0.0.0-20180220234015-a3eb8ddeffcc h1:RKf14vYWi2ttpEmkA4aQ3j4u9dStX2t4M8UM6qqNsG8=
@@ -275,6 +322,7 @@ github.com/pelletier/go-toml/v2 v2.0.8 h1:0ctb6s9mE31h0/lhu+J6OPmVeDxJn+kYnJc2jZ
 github.com/pelletier/go-toml/v2 v2.0.8/go.mod h1:vuYfssBdrU2XDZ9bYydBu6t+6a6PYNcZljzZR9VXg+4=
 github.com/pierrec/lz4/v4 v4.1.18 h1:xaKrnTkyoqfh1YItXl56+6KJNVYWlEEPuAQW9xsplYQ=
 github.com/pierrec/lz4/v4 v4.1.18/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
+github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -290,12 +338,18 @@ github.com/prometheus/common v0.44.0 h1:+5BrQJwiBB9xsMygAB3TNvpQKOwlkc25LbISbrdO
 github.com/prometheus/common v0.44.0/go.mod h1:ofAIvZbQ1e/nugmZGz4/qCb9Ap1VoSTIO7x0VV9VvuY=
 github.com/prometheus/procfs v0.11.1 h1:xRC8Iq1yyca5ypa9n1EZnWZkt7dwcoRPQwX/5gwaUuI=
 github.com/prometheus/procfs v0.11.1/go.mod h1:eesXgaPo1q7lBpVMoMy0ZOFTth9hBn4W/y0/p/ScXhY=
+github.com/qiniu/dyn v1.3.0/go.mod h1:E8oERcm8TtwJiZvkQPbcAh0RL8jO1G0VXJMW3FAWdkk=
+github.com/qiniu/go-sdk/v7 v7.18.2 h1:vk9eo5OO7aqgAOPF0Ytik/gt7CMKuNgzC/IPkhda6rk=
+github.com/qiniu/go-sdk/v7 v7.18.2/go.mod h1:nqoYCNo53ZlGA521RvRethvxUDvXKt4gtYXOwye868w=
+github.com/qiniu/x v1.10.5/go.mod h1:03Ni9tj+N2h2aKnAz+6N0Xfl8FwMEDRC2PAlxekASDs=
 github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 h1:N/ElC8H3+5XpJzTSTfLsJV/mx9Q9g7kxmchpfZyxgzM=
 github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/redis/go-redis/v9 v9.2.1 h1:WlYJg71ODF0dVspZZCpYmoF1+U1Jjk9Rwd7pq6QmlCg=
 github.com/redis/go-redis/v9 v9.2.1/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M=
 github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
 github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
+github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
+github.com/rogpeppe/go-internal v1.8.0/go.mod h1:WmiCO8CzOY8rg0OYDC4/i/2WRWAB6poM+XZ2dLUbcbE=
 github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
 github.com/rs/xid v1.5.0 h1:mKX4bl4iPYJtEIxp6CYiUuLQ/8DYMoz0PUdtGgMFRVc=
 github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
@@ -318,6 +372,7 @@ github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpE
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
+github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
@@ -368,8 +423,10 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk
 golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.1.0/go.mod h1:RecgLatLF4+eUMCP1PoPZQb+cVrJcOPbHkTkbkB9sbw=
 golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58=
 golang.org/x/crypto v0.14.0 h1:wBqGXzWJW6m1XrIKlAH0Hs1JJ7+9KBwnIO8v66Q9cHc=
 golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
@@ -400,6 +457,7 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v
 golang.org/x/net v0.0.0-20210428140749-89ef3d95e781/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
+golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco=
 golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
@@ -432,16 +490,19 @@ golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE=
 golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
 golang.org/x/term v0.13.0 h1:bb+I9cTfFazGW51MZqBVmZy7+JEJMouUHTUSKVQLBek=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -450,6 +511,7 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
+golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k=
 golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
@@ -510,6 +572,8 @@ google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqw
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
 gopkg.in/ini.v1 v1.67.0 h1:Dgnx+6+nfE+IfzjUEISNeydPJh9AXNNsWbGP9KzCsOA=
 gopkg.in/ini.v1 v1.67.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
@@ -529,6 +593,7 @@ gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gorm.io/driver/mysql v1.5.2 h1:QC2HRskSE75wBuOxe0+iCkyJZ+RqpudsQtqkp+IMuXs=

internal/api/msg.go

@@ -15,14 +15,6 @@
 package api
 
 import (
-	"github.com/OpenIMSDK/tools/mcontext"
-	"github.com/gin-gonic/gin"
-	"github.com/go-playground/validator/v10"
-	"github.com/mitchellh/mapstructure"
-
-	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
-
 	"github.com/OpenIMSDK/protocol/constant"
 	"github.com/OpenIMSDK/protocol/msg"
 	"github.com/OpenIMSDK/protocol/sdkws"
@@ -30,7 +22,13 @@ import (
 	"github.com/OpenIMSDK/tools/apiresp"
 	"github.com/OpenIMSDK/tools/errs"
 	"github.com/OpenIMSDK/tools/log"
+	"github.com/OpenIMSDK/tools/mcontext"
 	"github.com/OpenIMSDK/tools/utils"
+	"github.com/gin-gonic/gin"
+	"github.com/go-playground/validator/v10"
+	"github.com/mitchellh/mapstructure"
+	"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/apistruct"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"

internal/msggateway/callback.go

@@ -37,7 +37,7 @@ func CallbackUserOnline(ctx context.Context, userID string, platformID int, isAp
 	req := cbapi.CallbackUserOnlineReq{
 		UserStatusCallbackReq: cbapi.UserStatusCallbackReq{
 			UserStatusBaseCallback: cbapi.UserStatusBaseCallback{
-				CallbackCommand: constant.CallbackUserOnlineCommand,
+				CallbackCommand: cbapi.CallbackUserOnlineCommand,
 				OperationID:     mcontext.GetOperationID(ctx),
 				PlatformID:      platformID,
 				Platform:        constant.PlatformIDToName(platformID),
@@ -49,7 +49,10 @@ func CallbackUserOnline(ctx context.Context, userID string, platformID int, isAp
 		ConnID:          connID,
 	}
 	resp := cbapi.CommonCallbackResp{}
-	return http.CallBackPostReturn(ctx, callBackURL(), &req, &resp, config.Config.Callback.CallbackUserOnline)
+	if err := http.CallBackPostReturn(ctx, callBackURL(), &req, &resp, config.Config.Callback.CallbackUserOnline); err != nil {
+		return err
+	}
+	return nil
 }
 
 func CallbackUserOffline(ctx context.Context, userID string, platformID int, connID string) error {
@@ -59,7 +62,7 @@ func CallbackUserOffline(ctx context.Context, userID string, platformID int, con
 	req := &cbapi.CallbackUserOfflineReq{
 		UserStatusCallbackReq: cbapi.UserStatusCallbackReq{
 			UserStatusBaseCallback: cbapi.UserStatusBaseCallback{
-				CallbackCommand: constant.CallbackUserOfflineCommand,
+				CallbackCommand: cbapi.CallbackUserOfflineCommand,
 				OperationID:     mcontext.GetOperationID(ctx),
 				PlatformID:      platformID,
 				Platform:        constant.PlatformIDToName(platformID),
@@ -70,7 +73,10 @@ func CallbackUserOffline(ctx context.Context, userID string, platformID int, con
 		ConnID: connID,
 	}
 	resp := &cbapi.CallbackUserOfflineResp{}
-	return http.CallBackPostReturn(ctx, callBackURL(), req, resp, config.Config.Callback.CallbackUserOffline)
+	if err := http.CallBackPostReturn(ctx, callBackURL(), req, resp, config.Config.Callback.CallbackUserOffline); err != nil {
+		return err
+	}
+	return nil
 }
 
 func CallbackUserKickOff(ctx context.Context, userID string, platformID int) error {
@@ -80,7 +86,7 @@ func CallbackUserKickOff(ctx context.Context, userID string, platformID int) err
 	req := &cbapi.CallbackUserKickOffReq{
 		UserStatusCallbackReq: cbapi.UserStatusCallbackReq{
 			UserStatusBaseCallback: cbapi.UserStatusBaseCallback{
-				CallbackCommand: constant.CallbackUserKickOffCommand,
+				CallbackCommand: cbapi.CallbackUserKickOffCommand,
 				OperationID:     mcontext.GetOperationID(ctx),
 				PlatformID:      platformID,
 				Platform:        constant.PlatformIDToName(platformID),
@@ -90,7 +96,10 @@ func CallbackUserKickOff(ctx context.Context, userID string, platformID int) err
 		Seq: time.Now().UnixMilli(),
 	}
 	resp := &cbapi.CommonCallbackResp{}
-	return http.CallBackPostReturn(ctx, callBackURL(), req, resp, config.Config.Callback.CallbackUserOffline)
+	if err := http.CallBackPostReturn(ctx, callBackURL(), req, resp, config.Config.Callback.CallbackUserOffline); err != nil {
+		return err
+	}
+	return nil
 }
 
 // func callbackUserOnline(operationID, userID string, platformID int, token string, isAppBackground bool, connID

internal/msggateway/init.go

@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"github.com/OpenIMSDK/tools/utils"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 )
@@ -43,12 +44,22 @@ func RunWsAndServer(rpcPort, wsPort, prometheusPort int) error {
 	if err != nil {
 		return err
 	}
+
 	hubServer := NewServer(rpcPort, prometheusPort, longServer)
-	go func() {
+
-		err := hubServer.Start()
+	wg := errgroup.Group{}
+	wg.Go(func() error {
+		err = hubServer.Start()
 		if err != nil {
-			panic(utils.Wrap1(err))
+			return utils.Wrap1(err)
 		}
-	}()
+		return err
-	return hubServer.LongConnServer.Run()
+	})
+
+	wg.Go(func() error {
+		return hubServer.LongConnServer.Run()
+	})
+
+	err = wg.Wait()
+	return err
 }

internal/msggateway/n_ws_server.go

@@ -18,9 +18,12 @@ import (
 	"context"
 	"errors"
 	"net/http"
+	"os"
+	"os/signal"
 	"strconv"
 	"sync"
 	"sync/atomic"
+	"syscall"
 	"time"
 
 	"github.com/go-playground/validator/v10"
@@ -118,6 +121,7 @@ func (ws *WsServer) UnRegister(c *Client) {
 }
 
 func (ws *WsServer) Validate(s interface{}) error {
+	//?question?
 	return nil
 }
 
@@ -156,10 +160,22 @@ func NewWsServer(opts ...Option) (*WsServer, error) {
 }
 
 func (ws *WsServer) Run() error {
-	var client *Client
+	var (
-	go func() {
+		client *Client
+		wg     errgroup.Group
+
+		sigs = make(chan os.Signal, 1)
+		done = make(chan struct{}, 1)
+	)
+
+	server := http.Server{Addr: ":" + utils.IntToString(ws.port), Handler: nil}
+
+	wg.Go(func() error {
 		for {
 			select {
+			case <-done:
+				return nil
+
 			case client = <-ws.registerChan:
 				ws.registerClient(client)
 			case client = <-ws.unregisterChan:
@@ -168,10 +184,34 @@ func (ws *WsServer) Run() error {
 				ws.multiTerminalLoginChecker(onlineInfo.clientOK, onlineInfo.oldClients, onlineInfo.newClient)
 			}
 		}
+	})
+
+	wg.Go(func() error {
+		http.HandleFunc("/", ws.wsHandler)
+		return server.ListenAndServe()
+	})
+
+	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
+	<-sigs
+
+	go func() {
+		ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+		defer cancel()
+
+		// graceful exit operation for server
+		_ = server.Shutdown(ctx)
+		_ = wg.Wait()
+		close(done)
 	}()
-	http.HandleFunc("/", ws.wsHandler)
+
-	// http.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) {})
+	select {
-	return http.ListenAndServe(":"+utils.IntToString(ws.port), nil) // Start listening
+	case <-done:
+		return nil
+
+	case <-time.After(15 * time.Second):
+		return utils.Wrap1(errors.New("timeout exit"))
+	}
+
 }
 
 var concurrentRequest = 3

internal/msgtransfer/init.go

@@ -17,16 +17,15 @@ package msgtransfer
 import (
 	"errors"
 	"fmt"
-	"log"
-	"net/http"
-	"sync"
-
 	"github.com/OpenIMSDK/tools/mw"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/collectors"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	"log"
+	"net/http"
+	"sync"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"

internal/msgtransfer/online_history_msg_handler.go

@@ -252,7 +252,10 @@ func (och *OnlineHistoryRedisConsumerHandler) handleNotification(
 			return
 		}
 		log.ZDebug(ctx, "success to next topic", "conversationID", conversationID)
-		och.msgDatabase.MsgToMongoMQ(ctx, key, conversationID, storageList, lastSeq)
+		err = och.msgDatabase.MsgToMongoMQ(ctx, key, conversationID, storageList, lastSeq)
+		if err != nil {
+			log.ZError(ctx, "MsgToMongoMQ error", err)
+		}
 		och.toPushTopic(ctx, key, conversationID, storageList)
 	}
 }
@@ -277,9 +280,6 @@ func (och *OnlineHistoryRedisConsumerHandler) handleMsg(
 		lastSeq, isNewConversation, err := och.msgDatabase.BatchInsertChat2Cache(ctx, conversationID, storageList)
 		if err != nil && errs.Unwrap(err) != redis.Nil {
 			log.ZError(ctx, "batch data insert to redis err", err, "storageMsgList", storageList)
-			och.singleMsgFailedCountMutex.Lock()
-			och.singleMsgFailedCount += uint64(len(storageList))
-			och.singleMsgFailedCountMutex.Unlock()
 			return
 		}
 		if isNewConversation {
@@ -311,10 +311,10 @@ func (och *OnlineHistoryRedisConsumerHandler) handleMsg(
 		}
 
 		log.ZDebug(ctx, "success incr to next topic")
-		och.singleMsgSuccessCountMutex.Lock()
+		err = och.msgDatabase.MsgToMongoMQ(ctx, key, conversationID, storageList, lastSeq)
-		och.singleMsgSuccessCount += uint64(len(storageList))
+		if err != nil {
-		och.singleMsgSuccessCountMutex.Unlock()
+			log.ZError(ctx, "MsgToMongoMQ error", err)
-		och.msgDatabase.MsgToMongoMQ(ctx, key, conversationID, storageList, lastSeq)
+		}
 		och.toPushTopic(ctx, key, conversationID, storageList)
 	}
 }

internal/push/callback.go

@@ -19,7 +19,6 @@ import (
 
 	"github.com/OpenIMSDK/protocol/constant"
 	"github.com/OpenIMSDK/protocol/sdkws"
-	"github.com/OpenIMSDK/tools/errs"
 	"github.com/OpenIMSDK/tools/mcontext"
 	"github.com/OpenIMSDK/tools/utils"
 
@@ -44,7 +43,7 @@ func callbackOfflinePush(
 	req := &callbackstruct.CallbackBeforePushReq{
 		UserStatusBatchCallbackReq: callbackstruct.UserStatusBatchCallbackReq{
 			UserStatusBaseCallback: callbackstruct.UserStatusBaseCallback{
-				CallbackCommand: constant.CallbackOfflinePushCommand,
+				CallbackCommand: callbackstruct.CallbackOfflinePushCommand,
 				OperationID:     mcontext.GetOperationID(ctx),
 				PlatformID:      int(msg.SenderPlatformID),
 				Platform:        constant.PlatformIDToName(int(msg.SenderPlatformID)),
@@ -62,9 +61,6 @@ func callbackOfflinePush(
 	}
 	resp := &callbackstruct.CallbackBeforePushResp{}
 	if err := http.CallBackPostReturn(ctx, url(), req, resp, config.Config.Callback.CallbackOfflinePush); err != nil {
-		if err == errs.ErrCallbackContinue {
-			return nil
-		}
 		return err
 	}
 	if len(resp.UserIDs) != 0 {
@@ -83,7 +79,7 @@ func callbackOnlinePush(ctx context.Context, userIDs []string, msg *sdkws.MsgDat
 	req := callbackstruct.CallbackBeforePushReq{
 		UserStatusBatchCallbackReq: callbackstruct.UserStatusBatchCallbackReq{
 			UserStatusBaseCallback: callbackstruct.UserStatusBaseCallback{
-				CallbackCommand: constant.CallbackOnlinePushCommand,
+				CallbackCommand: callbackstruct.CallbackOnlinePushCommand,
 				OperationID:     mcontext.GetOperationID(ctx),
 				PlatformID:      int(msg.SenderPlatformID),
 				Platform:        constant.PlatformIDToName(int(msg.SenderPlatformID)),
@@ -99,7 +95,10 @@ func callbackOnlinePush(ctx context.Context, userIDs []string, msg *sdkws.MsgDat
 		Content:     GetContent(msg),
 	}
 	resp := &callbackstruct.CallbackBeforePushResp{}
-	return http.CallBackPostReturn(ctx, url(), req, resp, config.Config.Callback.CallbackOnlinePush)
+	if err := http.CallBackPostReturn(ctx, url(), req, resp, config.Config.Callback.CallbackOnlinePush); err != nil {
+		return err
+	}
+	return nil
 }
 
 func callbackBeforeSuperGroupOnlinePush(
@@ -113,7 +112,7 @@ func callbackBeforeSuperGroupOnlinePush(
 	}
 	req := callbackstruct.CallbackBeforeSuperGroupOnlinePushReq{
 		UserStatusBaseCallback: callbackstruct.UserStatusBaseCallback{
-			CallbackCommand: constant.CallbackSuperGroupOnlinePushCommand,
+			CallbackCommand: callbackstruct.CallbackSuperGroupOnlinePushCommand,
 			OperationID:     mcontext.GetOperationID(ctx),
 			PlatformID:      int(msg.SenderPlatformID),
 			Platform:        constant.PlatformIDToName(int(msg.SenderPlatformID)),
@@ -129,11 +128,9 @@ func callbackBeforeSuperGroupOnlinePush(
 	}
 	resp := &callbackstruct.CallbackBeforeSuperGroupOnlinePushResp{}
 	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, req, resp, config.Config.Callback.CallbackBeforeSuperGroupOnlinePush); err != nil {
-		if err == errs.ErrCallbackContinue {
-			return nil
-		}
 		return err
 	}
+	return nil
 	if len(resp.UserIDs) != 0 {
 		*pushToUserIDs = resp.UserIDs
 	}

internal/rpc/friend/black.go

@@ -74,6 +74,9 @@ func (s *friendServer) RemoveBlack(
 		return nil, err
 	}
 	s.notificationSender.BlackDeletedNotification(ctx, req)
+	if err := CallbackAfterRemoveBlack(ctx, req); err != nil {
+		return nil, err
+	}
 	return &pbfriend.RemoveBlackResp{}, nil
 }
 
@@ -85,6 +88,9 @@ func (s *friendServer) AddBlack(ctx context.Context, req *pbfriend.AddBlackReq)
 	if err != nil {
 		return nil, err
 	}
+	if err := CallbackBeforeAddBlack(ctx, req); err != nil {
+		return nil, err
+	}
 	black := relation.BlackModel{
 		OwnerUserID:    req.OwnerUserID,
 		BlockUserID:    req.BlackUserID,

internal/rpc/friend/callback.go

@@ -16,12 +16,8 @@ package friend
 
 import (
 	"context"
-
-	"github.com/OpenIMSDK/protocol/constant"
 	pbfriend "github.com/OpenIMSDK/protocol/friend"
-	"github.com/OpenIMSDK/tools/errs"
+	"github.com/OpenIMSDK/tools/utils"
-	"github.com/OpenIMSDK/tools/mcontext"
-
 	cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/http"
@@ -32,17 +28,162 @@ func CallbackBeforeAddFriend(ctx context.Context, req *pbfriend.ApplyToAddFriend
 		return nil
 	}
 	cbReq := &cbapi.CallbackBeforeAddFriendReq{
-		CallbackCommand: constant.CallbackBeforeAddFriendCommand,
+		CallbackCommand: cbapi.CallbackBeforeAddFriendCommand,
 		FromUserID:      req.FromUserID,
 		ToUserID:        req.ToUserID,
 		ReqMsg:          req.ReqMsg,
-		OperationID:     mcontext.GetOperationID(ctx),
+		Ex:              req.Ex,
 	}
 	resp := &cbapi.CallbackBeforeAddFriendResp{}
 	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeAddFriend); err != nil {
-		if err == errs.ErrCallbackContinue {
+		return err
-			return nil
+	}
-		}
+	return nil
+}
+
+func CallbackBeforeSetFriendRemark(ctx context.Context, req *pbfriend.SetFriendRemarkReq) error {
+	if !config.Config.Callback.CallbackBeforeSetFriendRemark.Enable {
+		return nil
+	}
+	cbReq := &cbapi.CallbackBeforeSetFriendRemarkReq{
+		CallbackCommand: cbapi.CallbackBeforeSetFriendRemark,
+		OwnerUserID:     req.OwnerUserID,
+		FriendUserID:    req.FriendUserID,
+		Remark:          req.Remark,
+	}
+	resp := &cbapi.CallbackBeforeSetFriendRemarkResp{}
+	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeAddFriend); err != nil {
+		return err
+	}
+	utils.NotNilReplace(&req.Remark, &resp.Remark)
+	return nil
+}
+
+func CallbackAfterSetFriendRemark(ctx context.Context, req *pbfriend.SetFriendRemarkReq) error {
+	if !config.Config.Callback.CallbackAfterSetFriendRemark.Enable {
+		return nil
+	}
+	cbReq := &cbapi.CallbackAfterSetFriendRemarkReq{
+		CallbackCommand: cbapi.CallbackAfterSetFriendRemark,
+		OwnerUserID:     req.OwnerUserID,
+		FriendUserID:    req.FriendUserID,
+		Remark:          req.Remark,
+	}
+	resp := &cbapi.CallbackAfterSetFriendRemarkResp{}
+	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeAddFriend); err != nil {
+		return err
+	}
+	return nil
+}
+func CallbackBeforeAddBlack(ctx context.Context, req *pbfriend.AddBlackReq) error {
+	if !config.Config.Callback.CallbackBeforeAddBlack.Enable {
+		return nil
+	}
+	cbReq := &cbapi.CallbackBeforeAddBlackReq{
+		CallbackCommand: cbapi.CallbackBeforeAddBlackCommand,
+		OwnerUserID:     req.OwnerUserID,
+		BlackUserID:     req.BlackUserID,
+	}
+	resp := &cbapi.CallbackBeforeAddBlackResp{}
+	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeAddBlack); err != nil {
+		return err
+	}
+	return nil
+}
+func CallbackAfterAddFriend(ctx context.Context, req *pbfriend.ApplyToAddFriendReq) error {
+	if !config.Config.Callback.CallbackAfterAddFriend.Enable {
+		return nil
+	}
+	cbReq := &cbapi.CallbackAfterAddFriendReq{
+		CallbackCommand: cbapi.CallbackAfterAddFriendCommand,
+		FromUserID:      req.FromUserID,
+		ToUserID:        req.ToUserID,
+		ReqMsg:          req.ReqMsg,
+	}
+	resp := &cbapi.CallbackAfterAddFriendResp{}
+	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackAfterAddFriend); err != nil {
+		return err
+	}
+
+	return nil
+}
+func CallbackBeforeAddFriendAgree(ctx context.Context, req *pbfriend.RespondFriendApplyReq) error {
+	if !config.Config.Callback.CallbackBeforeAddFriendAgree.Enable {
+		return nil
+	}
+	cbReq := &cbapi.CallbackBeforeAddFriendAgreeReq{
+		CallbackCommand: cbapi.CallbackBeforeAddFriendAgreeCommand,
+		FromUserID:      req.FromUserID,
+		ToUserID:        req.ToUserID,
+		HandleMsg:       req.HandleMsg,
+		HandleResult:    req.HandleResult,
+	}
+	resp := &cbapi.CallbackBeforeAddFriendAgreeResp{}
+	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeAddFriendAgree); err != nil {
+		return err
+	}
+	return nil
+}
+func CallbackAfterDeleteFriend(ctx context.Context, req *pbfriend.DeleteFriendReq) error {
+	if !config.Config.Callback.CallbackAfterDeleteFriend.Enable {
+		return nil
+	}
+	cbReq := &cbapi.CallbackAfterDeleteFriendReq{
+		CallbackCommand: cbapi.CallbackAfterDeleteFriendCommand,
+		OwnerUserID:     req.OwnerUserID,
+		FriendUserID:    req.FriendUserID,
+	}
+	resp := &cbapi.CallbackAfterDeleteFriendResp{}
+	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackAfterDeleteFriend); err != nil {
+		return err
+	}
+	return nil
+}
+func CallbackBeforeImportFriends(ctx context.Context, req *pbfriend.ImportFriendReq) error {
+	if !config.Config.Callback.CallbackBeforeImportFriends.Enable {
+		return nil
+	}
+	cbReq := &cbapi.CallbackBeforeImportFriendsReq{
+		CallbackCommand: cbapi.CallbackBeforeImportFriendsCommand,
+		OwnerUserID:     req.OwnerUserID,
+		FriendUserIDs:   req.FriendUserIDs,
+	}
+	resp := &cbapi.CallbackBeforeImportFriendsResp{}
+	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeImportFriends); err != nil {
+		return err
+	}
+	if len(resp.FriendUserIDs) != 0 {
+		req.FriendUserIDs = resp.FriendUserIDs
+	}
+	return nil
+}
+func CallbackAfterImportFriends(ctx context.Context, req *pbfriend.ImportFriendReq) error {
+	if !config.Config.Callback.CallbackAfterImportFriends.Enable {
+		return nil
+	}
+	cbReq := &cbapi.CallbackAfterImportFriendsReq{
+		CallbackCommand: cbapi.CallbackAfterImportFriendsCommand,
+		OwnerUserID:     req.OwnerUserID,
+		FriendUserIDs:   req.FriendUserIDs,
+	}
+	resp := &cbapi.CallbackAfterImportFriendsResp{}
+	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackAfterImportFriends); err != nil {
+		return err
+	}
+	return nil
+}
+
+func CallbackAfterRemoveBlack(ctx context.Context, req *pbfriend.RemoveBlackReq) error {
+	if !config.Config.Callback.CallbackAfterRemoveBlack.Enable {
+		return nil
+	}
+	cbReq := &cbapi.CallbackAfterRemoveBlackReq{
+		CallbackCommand: cbapi.CallbackAfterRemoveBlackCommand,
+		OwnerUserID:     req.OwnerUserID,
+		BlackUserID:     req.BlackUserID,
+	}
+	resp := &cbapi.CallbackAfterRemoveBlackResp{}
+	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackAfterRemoveBlack); err != nil {
 		return err
 	}
 	return nil

internal/rpc/friend/friend.go

@@ -103,7 +103,7 @@ func (s *friendServer) ApplyToAddFriend(
 	if req.ToUserID == req.FromUserID {
 		return nil, errs.ErrCanNotAddYourself.Wrap()
 	}
-	if err := CallbackBeforeAddFriend(ctx, req); err != nil && err != errs.ErrCallbackContinue {
+	if err = CallbackBeforeAddFriend(ctx, req); err != nil && err != errs.ErrCallbackContinue {
 		return nil, err
 	}
 	if _, err := s.userRpcClient.GetUsersInfoMap(ctx, []string{req.ToUserID, req.FromUserID}); err != nil {
@@ -120,6 +120,9 @@ func (s *friendServer) ApplyToAddFriend(
 		return nil, err
 	}
 	s.notificationSender.FriendApplicationAddNotification(ctx, req)
+	if err = CallbackAfterAddFriend(ctx, req); err != nil && err != errs.ErrCallbackContinue {
+		return nil, err
+	}
 	return resp, nil
 }
 
@@ -141,6 +144,10 @@ func (s *friendServer) ImportFriends(
 	if utils.Duplicate(req.FriendUserIDs) {
 		return nil, errs.ErrArgs.Wrap("friend userID repeated")
 	}
+	if err := CallbackBeforeImportFriends(ctx, req); err != nil {
+		return nil, err
+	}
+
 	if err := s.friendDatabase.BecomeFriends(ctx, req.OwnerUserID, req.FriendUserIDs, constant.BecomeFriendByImport); err != nil {
 		return nil, err
 	}
@@ -151,6 +158,9 @@ func (s *friendServer) ImportFriends(
 			HandleResult: constant.FriendResponseAgree,
 		})
 	}
+	if err := CallbackAfterImportFriends(ctx, req); err != nil {
+		return nil, err
+	}
 	return &pbfriend.ImportFriendResp{}, nil
 }
 
@@ -172,6 +182,9 @@ func (s *friendServer) RespondFriendApply(
 		HandleResult: req.HandleResult,
 	}
 	if req.HandleResult == constant.FriendResponseAgree {
+		if err := CallbackBeforeAddFriendAgree(ctx, req); err != nil && err != errs.ErrCallbackContinue {
+			return nil, err
+		}
 		err := s.friendDatabase.AgreeFriendRequest(ctx, &friendRequest)
 		if err != nil {
 			return nil, err
@@ -208,6 +221,9 @@ func (s *friendServer) DeleteFriend(
 		return nil, err
 	}
 	s.notificationSender.FriendDeletedNotification(ctx, req)
+	if err := CallbackAfterDeleteFriend(ctx, req); err != nil {
+		return nil, err
+	}
 	return resp, nil
 }
 
@@ -217,6 +233,10 @@ func (s *friendServer) SetFriendRemark(
 	req *pbfriend.SetFriendRemarkReq,
 ) (resp *pbfriend.SetFriendRemarkResp, err error) {
 	defer log.ZInfo(ctx, utils.GetFuncName()+" Return")
+
+	if err = CallbackBeforeSetFriendRemark(ctx, req); err != nil && err != errs.ErrCallbackContinue {
+		return nil, err
+	}
 	resp = &pbfriend.SetFriendRemarkResp{}
 	if err := s.userRpcClient.Access(ctx, req.OwnerUserID); err != nil {
 		return nil, err
@@ -228,6 +248,9 @@ func (s *friendServer) SetFriendRemark(
 	if err := s.friendDatabase.UpdateRemark(ctx, req.OwnerUserID, req.FriendUserID, req.Remark); err != nil {
 		return nil, err
 	}
+	if err := CallbackAfterSetFriendRemark(ctx, req); err != nil && err != errs.ErrCallbackContinue {
+		return nil, err
+	}
 	s.notificationSender.FriendRemarkSetNotification(ctx, req.OwnerUserID, req.FriendUserID)
 	return resp, nil
 }

internal/rpc/group/callback.go

@@ -16,15 +16,17 @@ package group
 
 import (
 	"context"
+	"github.com/OpenIMSDK/tools/log"
 	"time"
 
 	"github.com/OpenIMSDK/protocol/constant"
 	"github.com/OpenIMSDK/protocol/group"
 	"github.com/OpenIMSDK/protocol/wrapperspb"
-	"github.com/OpenIMSDK/tools/errs"
 	"github.com/OpenIMSDK/tools/mcontext"
 	"github.com/OpenIMSDK/tools/utils"
 
+	pbgroup "github.com/OpenIMSDK/protocol/group"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/apistruct"
 	"github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
@@ -37,7 +39,7 @@ func CallbackBeforeCreateGroup(ctx context.Context, req *group.CreateGroupReq) (
 		return nil
 	}
 	cbReq := &callbackstruct.CallbackBeforeCreateGroupReq{
-		CallbackCommand: constant.CallbackBeforeCreateGroupCommand,
+		CallbackCommand: callbackstruct.CallbackBeforeCreateGroupCommand,
 		OperationID:     mcontext.GetOperationID(ctx),
 		GroupInfo:       req.GroupInfo,
 	}
@@ -58,17 +60,7 @@ func CallbackBeforeCreateGroup(ctx context.Context, req *group.CreateGroupReq) (
 		})
 	}
 	resp := &callbackstruct.CallbackBeforeCreateGroupResp{}
-	err = http.CallBackPostReturn(
+	if err = http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeCreateGroup); err != nil {
-		ctx,
-		config.Config.Callback.CallbackUrl,
-		cbReq,
-		resp,
-		config.Config.Callback.CallbackBeforeCreateGroup,
-	)
-	if err != nil {
-		if err == errs.ErrCallbackContinue {
-			return nil
-		}
 		return err
 	}
 	utils.NotNilReplace(&req.GroupInfo.GroupID, resp.GroupID)
@@ -86,6 +78,37 @@ func CallbackBeforeCreateGroup(ctx context.Context, req *group.CreateGroupReq) (
 	return nil
 }
 
+func CallbackAfterCreateGroup(ctx context.Context, req *group.CreateGroupReq) (err error) {
+	if !config.Config.Callback.CallbackAfterCreateGroup.Enable {
+		return nil
+	}
+	cbReq := &callbackstruct.CallbackAfterCreateGroupReq{
+		CallbackCommand: callbackstruct.CallbackAfterCreateGroupCommand,
+		GroupInfo:       req.GroupInfo,
+	}
+	cbReq.InitMemberList = append(cbReq.InitMemberList, &apistruct.GroupAddMemberInfo{
+		UserID:    req.OwnerUserID,
+		RoleLevel: constant.GroupOwner,
+	})
+	for _, userID := range req.AdminUserIDs {
+		cbReq.InitMemberList = append(cbReq.InitMemberList, &apistruct.GroupAddMemberInfo{
+			UserID:    userID,
+			RoleLevel: constant.GroupAdmin,
+		})
+	}
+	for _, userID := range req.MemberUserIDs {
+		cbReq.InitMemberList = append(cbReq.InitMemberList, &apistruct.GroupAddMemberInfo{
+			UserID:    userID,
+			RoleLevel: constant.GroupOrdinaryUsers,
+		})
+	}
+	resp := &callbackstruct.CallbackAfterCreateGroupResp{}
+	if err = http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackAfterCreateGroup); err != nil {
+		return err
+	}
+	return nil
+}
+
 func CallbackBeforeMemberJoinGroup(
 	ctx context.Context,
 	groupMember *relation.GroupMemberModel,
@@ -95,8 +118,7 @@ func CallbackBeforeMemberJoinGroup(
 		return nil
 	}
 	callbackReq := &callbackstruct.CallbackBeforeMemberJoinGroupReq{
-		CallbackCommand: constant.CallbackBeforeMemberJoinGroupCommand,
+		CallbackCommand: callbackstruct.CallbackBeforeMemberJoinGroupCommand,
-		OperationID:     mcontext.GetOperationID(ctx),
 		GroupID:         groupMember.GroupID,
 		UserID:          groupMember.UserID,
 		Ex:              groupMember.Ex,
@@ -111,9 +133,6 @@ func CallbackBeforeMemberJoinGroup(
 		config.Config.Callback.CallbackBeforeMemberJoinGroup,
 	)
 	if err != nil {
-		if err == errs.ErrCallbackContinue {
-			return nil
-		}
 		return err
 	}
 	if resp.MuteEndTime != nil {
@@ -131,8 +150,7 @@ func CallbackBeforeSetGroupMemberInfo(ctx context.Context, req *group.SetGroupMe
 		return nil
 	}
 	callbackReq := callbackstruct.CallbackBeforeSetGroupMemberInfoReq{
-		CallbackCommand: constant.CallbackBeforeSetGroupMemberInfoCommand,
+		CallbackCommand: callbackstruct.CallbackBeforeSetGroupMemberInfoCommand,
-		OperationID:     mcontext.GetOperationID(ctx),
 		GroupID:         req.GroupID,
 		UserID:          req.UserID,
 	}
@@ -157,9 +175,6 @@ func CallbackBeforeSetGroupMemberInfo(ctx context.Context, req *group.SetGroupMe
 		config.Config.Callback.CallbackBeforeSetGroupMemberInfo,
 	)
 	if err != nil {
-		if err == errs.ErrCallbackContinue {
-			return nil
-		}
 		return err
 	}
 	if resp.FaceURL != nil {
@@ -176,3 +191,240 @@ func CallbackBeforeSetGroupMemberInfo(ctx context.Context, req *group.SetGroupMe
 	}
 	return nil
 }
+func CallbackAfterSetGroupMemberInfo(ctx context.Context, req *group.SetGroupMemberInfo) (err error) {
+	if !config.Config.Callback.CallbackBeforeSetGroupMemberInfo.Enable {
+		return nil
+	}
+	callbackReq := callbackstruct.CallbackAfterSetGroupMemberInfoReq{
+		CallbackCommand: callbackstruct.CallbackAfterSetGroupMemberInfoCommand,
+		GroupID:         req.GroupID,
+		UserID:          req.UserID,
+	}
+	if req.Nickname != nil {
+		callbackReq.Nickname = &req.Nickname.Value
+	}
+	if req.FaceURL != nil {
+		callbackReq.FaceURL = &req.FaceURL.Value
+	}
+	if req.RoleLevel != nil {
+		callbackReq.RoleLevel = &req.RoleLevel.Value
+	}
+	if req.Ex != nil {
+		callbackReq.Ex = &req.Ex.Value
+	}
+	resp := &callbackstruct.CallbackAfterSetGroupMemberInfoResp{}
+	if err = http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, callbackReq, resp, config.Config.Callback.CallbackAfterSetGroupMemberInfo); err != nil {
+		return err
+	}
+	return nil
+}
+
+func CallbackQuitGroup(ctx context.Context, req *group.QuitGroupReq) (err error) {
+	if !config.Config.Callback.CallbackQuitGroup.Enable {
+		return nil
+	}
+	cbReq := &callbackstruct.CallbackQuitGroupReq{
+		CallbackCommand: callbackstruct.CallbackQuitGroupCommand,
+		GroupID:         req.GroupID,
+		UserID:          req.UserID,
+	}
+	resp := &callbackstruct.CallbackQuitGroupResp{}
+	if err = http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackQuitGroup); err != nil {
+		return err
+	}
+	return nil
+}
+
+func CallbackKillGroupMember(ctx context.Context, req *pbgroup.KickGroupMemberReq) (err error) {
+	if !config.Config.Callback.CallbackKillGroupMember.Enable {
+		return nil
+	}
+	cbReq := &callbackstruct.CallbackKillGroupMemberReq{
+		CallbackCommand: callbackstruct.CallbackKillGroupCommand,
+		GroupID:         req.GroupID,
+		KickedUserIDs:   req.KickedUserIDs,
+	}
+	resp := &callbackstruct.CallbackKillGroupMemberResp{}
+	if err = http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackQuitGroup); err != nil {
+		return err
+	}
+	return nil
+}
+
+func CallbackDismissGroup(ctx context.Context, req *callbackstruct.CallbackDisMissGroupReq) (err error) {
+	if !config.Config.Callback.CallbackDismissGroup.Enable {
+		return nil
+	}
+	req.CallbackCommand = callbackstruct.CallbackDisMissGroupCommand
+	resp := &callbackstruct.CallbackDisMissGroupResp{}
+	if err = http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, req, resp, config.Config.Callback.CallbackQuitGroup); err != nil {
+		return err
+	}
+	return nil
+}
+
+func CallbackApplyJoinGroupBefore(ctx context.Context, req *callbackstruct.CallbackJoinGroupReq) (err error) {
+	if !config.Config.Callback.CallbackBeforeJoinGroup.Enable {
+		return nil
+	}
+
+	req.CallbackCommand = callbackstruct.CallbackBeforeJoinGroupCommand
+
+	resp := &callbackstruct.CallbackJoinGroupResp{}
+	if err = http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, req, resp, config.Config.Callback.CallbackBeforeJoinGroup); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func CallbackTransferGroupOwnerAfter(ctx context.Context, req *pbgroup.TransferGroupOwnerReq) (err error) {
+	if !config.Config.Callback.CallbackTransferGroupOwnerAfter.Enable {
+		return nil
+	}
+
+	cbReq := &callbackstruct.CallbackTransferGroupOwnerReq{
+		CallbackCommand: callbackstruct.CallbackTransferGroupOwnerAfter,
+		GroupID:         req.GroupID,
+		OldOwnerUserID:  req.OldOwnerUserID,
+		NewOwnerUserID:  req.NewOwnerUserID,
+	}
+
+	resp := &callbackstruct.CallbackTransferGroupOwnerResp{}
+	if err = http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeJoinGroup); err != nil {
+		return err
+	}
+	return nil
+}
+func CallbackBeforeInviteUserToGroup(ctx context.Context, req *group.InviteUserToGroupReq) (err error) {
+	if !config.Config.Callback.CallbackBeforeInviteUserToGroup.Enable {
+		return nil
+	}
+
+	callbackReq := &callbackstruct.CallbackBeforeInviteUserToGroupReq{
+		CallbackCommand: callbackstruct.CallbackBeforeInviteJoinGroupCommand,
+		OperationID:     mcontext.GetOperationID(ctx),
+		GroupID:         req.GroupID,
+		Reason:          req.Reason,
+		InvitedUserIDs:  req.InvitedUserIDs,
+	}
+
+	resp := &callbackstruct.CallbackBeforeInviteUserToGroupResp{}
+	err = http.CallBackPostReturn(
+		ctx,
+		config.Config.Callback.CallbackUrl,
+		callbackReq,
+		resp,
+		config.Config.Callback.CallbackBeforeInviteUserToGroup,
+	)
+
+	if err != nil {
+		return err
+	}
+
+	if len(resp.RefusedMembersAccount) > 0 {
+		// Handle the scenario where certain members are refused
+		// You might want to update the req.Members list or handle it as per your business logic
+	}
+	utils.StructFieldNotNilReplace(req, resp)
+
+	return nil
+}
+
+func CallbackAfterJoinGroup(ctx context.Context, req *group.JoinGroupReq) error {
+	if !config.Config.Callback.CallbackAfterJoinGroup.Enable {
+		return nil
+	}
+	callbackReq := &callbackstruct.CallbackAfterJoinGroupReq{
+		CallbackCommand: callbackstruct.CallbackAfterJoinGroupCommand,
+		OperationID:     mcontext.GetOperationID(ctx),
+		GroupID:         req.GroupID,
+		ReqMessage:      req.ReqMessage,
+		JoinSource:      req.JoinSource,
+		InviterUserID:   req.InviterUserID,
+	}
+	resp := &callbackstruct.CallbackAfterJoinGroupResp{}
+	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, callbackReq, resp, config.Config.Callback.CallbackAfterJoinGroup); err != nil {
+		return err
+	}
+	return nil
+}
+
+func CallbackBeforeSetGroupInfo(ctx context.Context, req *group.SetGroupInfoReq) error {
+	if !config.Config.Callback.CallbackBeforeSetGroupInfo.Enable {
+		return nil
+	}
+	callbackReq := &callbackstruct.CallbackBeforeSetGroupInfoReq{
+		CallbackCommand: callbackstruct.CallbackBeforeSetGroupInfoCommand,
+		GroupID:         req.GroupInfoForSet.GroupID,
+		Notification:    req.GroupInfoForSet.Notification,
+		Introduction:    req.GroupInfoForSet.Introduction,
+		FaceURL:         req.GroupInfoForSet.FaceURL,
+		GroupName:       req.GroupInfoForSet.GroupName,
+	}
+
+	if req.GroupInfoForSet.Ex != nil {
+		callbackReq.Ex = req.GroupInfoForSet.Ex.Value
+	}
+	log.ZDebug(ctx, "debug CallbackBeforeSetGroupInfo", callbackReq.Ex)
+	if req.GroupInfoForSet.NeedVerification != nil {
+		callbackReq.NeedVerification = req.GroupInfoForSet.NeedVerification.Value
+	}
+	if req.GroupInfoForSet.LookMemberInfo != nil {
+		callbackReq.LookMemberInfo = req.GroupInfoForSet.LookMemberInfo.Value
+	}
+	if req.GroupInfoForSet.ApplyMemberFriend != nil {
+		callbackReq.ApplyMemberFriend = req.GroupInfoForSet.ApplyMemberFriend.Value
+	}
+	resp := &callbackstruct.CallbackBeforeSetGroupInfoResp{}
+
+	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, callbackReq, resp, config.Config.Callback.CallbackBeforeSetGroupInfo); err != nil {
+		return err
+	}
+
+	if resp.Ex != nil {
+		req.GroupInfoForSet.Ex = wrapperspb.String(*resp.Ex)
+	}
+	if resp.NeedVerification != nil {
+		req.GroupInfoForSet.NeedVerification = wrapperspb.Int32(*resp.NeedVerification)
+	}
+	if resp.LookMemberInfo != nil {
+		req.GroupInfoForSet.LookMemberInfo = wrapperspb.Int32(*resp.LookMemberInfo)
+	}
+	if resp.ApplyMemberFriend != nil {
+		req.GroupInfoForSet.ApplyMemberFriend = wrapperspb.Int32(*resp.ApplyMemberFriend)
+	}
+	utils.StructFieldNotNilReplace(req, resp)
+	return nil
+}
+func CallbackAfterSetGroupInfo(ctx context.Context, req *group.SetGroupInfoReq) error {
+	if !config.Config.Callback.CallbackAfterSetGroupInfo.Enable {
+		return nil
+	}
+	callbackReq := &callbackstruct.CallbackAfterSetGroupInfoReq{
+		CallbackCommand: callbackstruct.CallbackAfterSetGroupInfoCommand,
+		GroupID:         req.GroupInfoForSet.GroupID,
+		Notification:    req.GroupInfoForSet.Notification,
+		Introduction:    req.GroupInfoForSet.Introduction,
+		FaceURL:         req.GroupInfoForSet.FaceURL,
+		GroupName:       req.GroupInfoForSet.GroupName,
+	}
+	if req.GroupInfoForSet.Ex != nil {
+		callbackReq.Ex = &req.GroupInfoForSet.Ex.Value
+	}
+	if req.GroupInfoForSet.NeedVerification != nil {
+		callbackReq.NeedVerification = &req.GroupInfoForSet.NeedVerification.Value
+	}
+	if req.GroupInfoForSet.LookMemberInfo != nil {
+		callbackReq.LookMemberInfo = &req.GroupInfoForSet.LookMemberInfo.Value
+	}
+	if req.GroupInfoForSet.ApplyMemberFriend != nil {
+		callbackReq.ApplyMemberFriend = &req.GroupInfoForSet.ApplyMemberFriend.Value
+	}
+	resp := &callbackstruct.CallbackAfterSetGroupInfoResp{}
+	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, callbackReq, resp, config.Config.Callback.CallbackAfterSetGroupInfo); err != nil {
+		return err
+	}
+	utils.StructFieldNotNilReplace(req, resp)
+	return nil
+}

internal/rpc/group/group.go

@@ -26,6 +26,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
 	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
 
@@ -225,6 +227,7 @@ func (s *groupServer) CreateGroup(ctx context.Context, req *pbgroup.CreateGroupR
 	if len(userMap) != len(userIDs) {
 		return nil, errs.ErrUserIDNotFound.Wrap("user not found")
 	}
+	// Callback Before create Group
 	if err := CallbackBeforeCreateGroup(ctx, req); err != nil {
 		return nil, err
 	}
@@ -298,6 +301,17 @@ func (s *groupServer) CreateGroup(ctx context.Context, req *pbgroup.CreateGroupR
 		}
 		s.Notification.GroupCreatedNotification(ctx, tips)
 	}
+	reqCallBackAfter := &pbgroup.CreateGroupReq{
+		MemberUserIDs: userIDs,
+		GroupInfo:     resp.GroupInfo,
+		OwnerUserID:   req.OwnerUserID,
+		AdminUserIDs:  req.AdminUserIDs,
+	}
+
+	if err := CallbackAfterCreateGroup(ctx, reqCallBackAfter); err != nil {
+		return nil, err
+	}
+
 	return resp, nil
 }
 
@@ -352,6 +366,7 @@ func (s *groupServer) GetJoinedGroupList(ctx context.Context, req *pbgroup.GetJo
 
 func (s *groupServer) InviteUserToGroup(ctx context.Context, req *pbgroup.InviteUserToGroupReq) (*pbgroup.InviteUserToGroupResp, error) {
 	resp := &pbgroup.InviteUserToGroupResp{}
+
 	if len(req.InvitedUserIDs) == 0 {
 		return nil, errs.ErrArgs.Wrap("user empty")
 	}
@@ -362,6 +377,7 @@ func (s *groupServer) InviteUserToGroup(ctx context.Context, req *pbgroup.Invite
 	if err != nil {
 		return nil, err
 	}
+
 	if group.Status == constant.GroupStatusDismissed {
 		return nil, errs.ErrDismissedAlready.Wrap()
 	}
@@ -385,6 +401,10 @@ func (s *groupServer) InviteUserToGroup(ctx context.Context, req *pbgroup.Invite
 		}
 		groupMember = groupMembers[0]
 	}
+
+	if err := CallbackBeforeInviteUserToGroup(ctx, req); err != nil {
+		return nil, err
+	}
 	if group.NeedVerification == constant.AllNeedVerification {
 		if !authverify.IsAppManagerUid(ctx) {
 			if !(groupMember.RoleLevel == constant.GroupOwner || groupMember.RoleLevel == constant.GroupAdmin) {
@@ -399,6 +419,7 @@ func (s *groupServer) InviteUserToGroup(ctx context.Context, req *pbgroup.Invite
 						HandledTime:   time.Unix(0, 0),
 					})
 				}
+
 				if err := s.GroupDatabase.CreateGroupRequest(ctx, requests); err != nil {
 					return nil, err
 				}
@@ -606,6 +627,10 @@ func (s *groupServer) KickGroupMember(ctx context.Context, req *pbgroup.KickGrou
 	if err := s.deleteMemberAndSetConversationSeq(ctx, req.GroupID, req.KickedUserIDs); err != nil {
 		return nil, err
 	}
+
+	if err := CallbackKillGroupMember(ctx, req); err != nil {
+		return nil, err
+	}
 	return resp, nil
 }
 
@@ -800,6 +825,7 @@ func (s *groupServer) GroupApplicationResponse(ctx context.Context, req *pbgroup
 	case constant.GroupResponseRefuse:
 		s.Notification.GroupApplicationRejectedNotification(ctx, req)
 	}
+
 	return &pbgroup.GroupApplicationResponseResp{}, nil
 }
 
@@ -816,6 +842,17 @@ func (s *groupServer) JoinGroup(ctx context.Context, req *pbgroup.JoinGroupReq)
 	if group.Status == constant.GroupStatusDismissed {
 		return nil, errs.ErrDismissedAlready.Wrap()
 	}
+
+	reqCall := &callbackstruct.CallbackJoinGroupReq{
+		GroupID:    req.GroupID,
+		GroupType:  string(group.GroupType),
+		ApplyID:    req.InviterUserID,
+		ReqMessage: req.ReqMessage,
+	}
+
+	if err = CallbackApplyJoinGroupBefore(ctx, reqCall); err != nil {
+		return nil, err
+	}
 	_, err = s.GroupDatabase.TakeGroupMember(ctx, req.GroupID, req.InviterUserID)
 	if err == nil {
 		return nil, errs.ErrArgs.Wrap("already in group")
@@ -843,10 +880,14 @@ func (s *groupServer) JoinGroup(ctx context.Context, req *pbgroup.JoinGroupReq)
 		if err := s.GroupDatabase.CreateGroup(ctx, nil, []*relationtb.GroupMemberModel{groupMember}); err != nil {
 			return nil, err
 		}
+
 		if err := s.conversationRpcClient.GroupChatFirstCreateConversation(ctx, req.GroupID, []string{req.InviterUserID}); err != nil {
 			return nil, err
 		}
 		s.Notification.MemberEnterNotification(ctx, req.GroupID, req.InviterUserID)
+		if err = CallbackAfterJoinGroup(ctx, req); err != nil {
+			return nil, err
+		}
 		return resp, nil
 	}
 	groupRequest := relationtb.GroupRequestModel{
@@ -900,6 +941,10 @@ func (s *groupServer) QuitGroup(ctx context.Context, req *pbgroup.QuitGroupReq)
 		return nil, err
 	}
 
+	// callback
+	if err := CallbackQuitGroup(ctx, req); err != nil {
+		return nil, err
+	}
 	return resp, nil
 }
 
@@ -924,6 +969,9 @@ func (s *groupServer) SetGroupInfo(ctx context.Context, req *pbgroup.SetGroupInf
 			return nil, errs.ErrNoPermission.Wrap("no group owner or admin")
 		}
 	}
+	if err := CallbackBeforeSetGroupInfo(ctx, req); err != nil {
+		return nil, err
+	}
 	group, err := s.GroupDatabase.TakeGroup(ctx, req.GroupInfoForSet.GroupID)
 	if err != nil {
 		return nil, err
@@ -992,6 +1040,9 @@ func (s *groupServer) SetGroupInfo(ctx context.Context, req *pbgroup.SetGroupInf
 	default:
 		s.Notification.GroupInfoSetNotification(ctx, tips)
 	}
+	if err := CallbackAfterSetGroupInfo(ctx, req); err != nil {
+		return nil, err
+	}
 	return resp, nil
 }
 
@@ -1031,6 +1082,10 @@ func (s *groupServer) TransferGroupOwner(ctx context.Context, req *pbgroup.Trans
 	if err := s.GroupDatabase.TransferGroupOwner(ctx, req.GroupID, req.OldOwnerUserID, req.NewOwnerUserID, newOwner.RoleLevel); err != nil {
 		return nil, err
 	}
+
+	if err := CallbackTransferGroupOwnerAfter(ctx, req); err != nil {
+		return nil, err
+	}
 	s.Notification.GroupOwnerTransferredNotification(ctx, req)
 	return resp, nil
 }
@@ -1201,6 +1256,20 @@ func (s *groupServer) DismissGroup(ctx context.Context, req *pbgroup.DismissGrou
 			s.Notification.GroupDismissedNotification(ctx, tips)
 		}
 	}
+	membersID, err := s.GroupDatabase.FindGroupMemberUserID(ctx, group.GroupID)
+	if err != nil {
+		return nil, err
+	}
+	reqCall := &callbackstruct.CallbackDisMissGroupReq{
+		GroupID:   req.GroupID,
+		OwnerID:   owner.UserID,
+		MembersID: membersID,
+		GroupType: string(group.GroupType),
+	}
+	if err := CallbackDismissGroup(ctx, reqCall); err != nil {
+		return nil, err
+	}
+
 	return resp, nil
 }
 
@@ -1439,6 +1508,12 @@ func (s *groupServer) SetGroupMemberInfo(ctx context.Context, req *pbgroup.SetGr
 			}
 		}
 	}
+	for i := 0; i < len(req.Members); i++ {
+		if err := CallbackAfterSetGroupMemberInfo(ctx, req.Members[i]); err != nil {
+			return nil, err
+		}
+	}
+
 	return resp, nil
 }
 

internal/rpc/msg/as_read.go

@@ -16,10 +16,14 @@ package msg
 
 import (
 	"context"
+	cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
+
+	utils2 "github.com/OpenIMSDK/tools/utils"
 
 	"github.com/redis/go-redis/v9"
 
 	"github.com/OpenIMSDK/protocol/constant"
+	"github.com/OpenIMSDK/protocol/conversation"
 	"github.com/OpenIMSDK/protocol/msg"
 	"github.com/OpenIMSDK/protocol/sdkws"
 	"github.com/OpenIMSDK/tools/errs"
@@ -88,10 +92,7 @@ func (m *msgServer) SetConversationHasReadSeq(
 	return &msg.SetConversationHasReadSeqResp{}, nil
 }
 
-func (m *msgServer) MarkMsgsAsRead(
+func (m *msgServer) MarkMsgsAsRead(ctx context.Context, req *msg.MarkMsgsAsReadReq) (resp *msg.MarkMsgsAsReadResp, err error) {
-	ctx context.Context,
-	req *msg.MarkMsgsAsReadReq,
-) (resp *msg.MarkMsgsAsReadResp, err error) {
 	if len(req.Seqs) < 1 {
 		return nil, errs.ErrArgs.Wrap("seqs must not be empty")
 	}
@@ -127,10 +128,7 @@ func (m *msgServer) MarkMsgsAsRead(
 	return &msg.MarkMsgsAsReadResp{}, nil
 }
 
-func (m *msgServer) MarkConversationAsRead(
+func (m *msgServer) MarkConversationAsRead(ctx context.Context, req *msg.MarkConversationAsReadReq) (resp *msg.MarkConversationAsReadResp, err error) {
-	ctx context.Context,
-	req *msg.MarkConversationAsReadReq,
-) (resp *msg.MarkConversationAsReadResp, err error) {
 	conversation, err := m.Conversation.GetConversation(ctx, req.UserID, req.ConversationID)
 	if err != nil {
 		return nil, err
@@ -139,49 +137,56 @@ func (m *msgServer) MarkConversationAsRead(
 	if err != nil && errs.Unwrap(err) != redis.Nil {
 		return nil, err
 	}
+	seqs := generateSeqs(hasReadSeq, req)
+
+	if len(seqs) > 0 || req.HasReadSeq > hasReadSeq {
+		err = m.updateReadStatus(ctx, req, conversation, seqs, hasReadSeq)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return &msg.MarkConversationAsReadResp{}, nil
+}
+
+func generateSeqs(hasReadSeq int64, req *msg.MarkConversationAsReadReq) []int64 {
 	var seqs []int64
+	for _, val := range req.Seqs {
+		if val > hasReadSeq && !utils2.Contain(val, seqs...) {
+			seqs = append(seqs, val)
+		}
+	}
+	return seqs
+}
 
-	log.ZDebug(ctx, "MarkConversationAsRead", "hasReadSeq", hasReadSeq,
+func (m *msgServer) updateReadStatus(ctx context.Context, req *msg.MarkConversationAsReadReq, conversation *conversation.Conversation, seqs []int64, hasReadSeq int64) error {
-		"req.HasReadSeq", req.HasReadSeq)
+	if conversation.ConversationType == constant.SingleChatType && len(seqs) > 0 {
-	if conversation.ConversationType == constant.SingleChatType {
+		log.ZDebug(ctx, "MarkConversationAsRead", "seqs", seqs, "conversationID", req.ConversationID)
-		for i := hasReadSeq + 1; i <= req.HasReadSeq; i++ {
+		if err := m.MsgDatabase.MarkSingleChatMsgsAsRead(ctx, req.UserID, req.ConversationID, seqs); err != nil {
-			seqs = append(seqs, i)
+			return err
 		}
-
+	}
-		if len(seqs) > 0 {
+	reqCall := &cbapi.CallbackGroupMsgReadReq{
-			log.ZDebug(ctx, "MarkConversationAsRead", "seqs", seqs, "conversationID", req.ConversationID)
+		SendID:       conversation.OwnerUserID,
-			if err = m.MsgDatabase.MarkSingleChatMsgsAsRead(ctx, req.UserID, req.ConversationID, seqs); err != nil {
+		ReceiveID:    req.UserID,
-				return nil, err
+		UnreadMsgNum: req.HasReadSeq,
-			}
+		ContentType:  int64(conversation.ConversationType),
-		}
+	}
-		if req.HasReadSeq > hasReadSeq {
+	if err := CallbackGroupMsgRead(ctx, reqCall); err != nil {
-			err = m.MsgDatabase.SetHasReadSeq(ctx, req.UserID, req.ConversationID, req.HasReadSeq)
+		return err
-			if err != nil {
-				return nil, err
-			}
-			hasReadSeq = req.HasReadSeq
-		}
-		if err = m.sendMarkAsReadNotification(ctx, req.ConversationID, conversation.ConversationType, req.UserID,
-			m.conversationAndGetRecvID(conversation, req.UserID), seqs, hasReadSeq); err != nil {
-			return nil, err
-		}
-	} else if conversation.ConversationType == constant.SuperGroupChatType ||
-		conversation.ConversationType == constant.NotificationChatType {
-		if req.HasReadSeq > hasReadSeq {
-			err = m.MsgDatabase.SetHasReadSeq(ctx, req.UserID, req.ConversationID, req.HasReadSeq)
-			if err != nil {
-				return nil, err
-			}
-			hasReadSeq = req.HasReadSeq
-		}
-		if err = m.sendMarkAsReadNotification(ctx, req.ConversationID, constant.SingleChatType, req.UserID,
-			req.UserID, seqs, hasReadSeq); err != nil {
-			return nil, err
-		}
-
 	}
 
-	return &msg.MarkConversationAsReadResp{}, nil
+	if req.HasReadSeq > hasReadSeq {
+		if err := m.MsgDatabase.SetHasReadSeq(ctx, req.UserID, req.ConversationID, req.HasReadSeq); err != nil {
+			return err
+		}
+	}
+
+	recvID := m.conversationAndGetRecvID(conversation, req.UserID)
+	if conversation.ConversationType == constant.SuperGroupChatType || conversation.ConversationType == constant.NotificationChatType {
+		recvID = req.UserID
+	}
+
+	return m.sendMarkAsReadNotification(ctx, req.ConversationID, conversation.ConversationType, req.UserID, recvID, seqs, req.HasReadSeq)
 }
 
 func (m *msgServer) sendMarkAsReadNotification(

internal/rpc/msg/callback.go

@@ -16,18 +16,16 @@ package msg
 
 import (
 	"context"
-
 	"github.com/OpenIMSDK/protocol/sdkws"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/OpenIMSDK/protocol/constant"
 	pbchat "github.com/OpenIMSDK/protocol/msg"
-	"github.com/OpenIMSDK/tools/errs"
 	"github.com/OpenIMSDK/tools/log"
 	"github.com/OpenIMSDK/tools/mcontext"
 	"github.com/OpenIMSDK/tools/utils"
-
 	cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/http"
 )
@@ -74,14 +72,11 @@ func callbackBeforeSendSingleMsg(ctx context.Context, msg *pbchat.SendMsgReq) er
 		return nil
 	}
 	req := &cbapi.CallbackBeforeSendSingleMsgReq{
-		CommonCallbackReq: toCommonCallback(ctx, msg, constant.CallbackBeforeSendSingleMsgCommand),
+		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackBeforeSendSingleMsgCommand),
 		RecvID:            msg.MsgData.RecvID,
 	}
 	resp := &cbapi.CallbackBeforeSendSingleMsgResp{}
 	if err := http.CallBackPostReturn(ctx, cbURL(), req, resp, config.Config.Callback.CallbackBeforeSendSingleMsg); err != nil {
-		if err == errs.ErrCallbackContinue {
-			return nil
-		}
 		return err
 	}
 	return nil
@@ -92,32 +87,26 @@ func callbackAfterSendSingleMsg(ctx context.Context, msg *pbchat.SendMsgReq) err
 		return nil
 	}
 	req := &cbapi.CallbackAfterSendSingleMsgReq{
-		CommonCallbackReq: toCommonCallback(ctx, msg, constant.CallbackAfterSendSingleMsgCommand),
+		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendSingleMsgCommand),
 		RecvID:            msg.MsgData.RecvID,
 	}
 	resp := &cbapi.CallbackAfterSendSingleMsgResp{}
 	if err := http.CallBackPostReturn(ctx, cbURL(), req, resp, config.Config.Callback.CallbackAfterSendSingleMsg); err != nil {
-		if err == errs.ErrCallbackContinue {
-			return nil
-		}
 		return err
 	}
 	return nil
 }
 
 func callbackBeforeSendGroupMsg(ctx context.Context, msg *pbchat.SendMsgReq) error {
-	if !config.Config.Callback.CallbackAfterSendSingleMsg.Enable {
+	if !config.Config.Callback.CallbackBeforeSendSingleMsg.Enable {
 		return nil
 	}
-	req := &cbapi.CallbackAfterSendGroupMsgReq{
+	req := &cbapi.CallbackBeforeSendGroupMsgReq{
-		CommonCallbackReq: toCommonCallback(ctx, msg, constant.CallbackBeforeSendGroupMsgCommand),
+		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackBeforeSendGroupMsgCommand),
 		GroupID:           msg.MsgData.GroupID,
 	}
 	resp := &cbapi.CallbackBeforeSendGroupMsgResp{}
 	if err := http.CallBackPostReturn(ctx, cbURL(), req, resp, config.Config.Callback.CallbackBeforeSendGroupMsg); err != nil {
-		if err == errs.ErrCallbackContinue {
-			return nil
-		}
 		return err
 	}
 	return nil
@@ -128,14 +117,11 @@ func callbackAfterSendGroupMsg(ctx context.Context, msg *pbchat.SendMsgReq) erro
 		return nil
 	}
 	req := &cbapi.CallbackAfterSendGroupMsgReq{
-		CommonCallbackReq: toCommonCallback(ctx, msg, constant.CallbackAfterSendGroupMsgCommand),
+		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendGroupMsgCommand),
 		GroupID:           msg.MsgData.GroupID,
 	}
 	resp := &cbapi.CallbackAfterSendGroupMsgResp{}
 	if err := http.CallBackPostReturn(ctx, cbURL(), req, resp, config.Config.Callback.CallbackAfterSendGroupMsg); err != nil {
-		if err == errs.ErrCallbackContinue {
-			return nil
-		}
 		return err
 	}
 	return nil
@@ -146,13 +132,10 @@ func callbackMsgModify(ctx context.Context, msg *pbchat.SendMsgReq) error {
 		return nil
 	}
 	req := &cbapi.CallbackMsgModifyCommandReq{
-		CommonCallbackReq: toCommonCallback(ctx, msg, constant.CallbackMsgModifyCommand),
+		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackMsgModifyCommand),
 	}
 	resp := &cbapi.CallbackMsgModifyCommandResp{}
 	if err := http.CallBackPostReturn(ctx, cbURL(), req, resp, config.Config.Callback.CallbackMsgModify); err != nil {
-		if err == errs.ErrCallbackContinue {
-			return nil
-		}
 		return err
 	}
 	if resp.Content != nil {
@@ -177,3 +160,46 @@ func callbackMsgModify(ctx context.Context, msg *pbchat.SendMsgReq) error {
 	log.ZDebug(ctx, "callbackMsgModify", "msg", msg.MsgData)
 	return nil
 }
+func CallbackGroupMsgRead(ctx context.Context, req *cbapi.CallbackGroupMsgReadReq) error {
+	if !config.Config.Callback.CallbackGroupMsgRead.Enable || req.ContentType != constant.Text {
+		return nil
+	}
+	req.CallbackCommand = cbapi.CallbackGroupMsgReadCommand
+
+	resp := &cbapi.CallbackGroupMsgReadResp{}
+	if err := http.CallBackPostReturn(ctx, cbURL(), req, resp, config.Config.Callback.CallbackMsgModify); err != nil {
+		return err
+	}
+	return nil
+}
+
+func CallbackSingleMsgRead(ctx context.Context, req *cbapi.CallbackSingleMsgReadReq) error {
+	if !config.Config.Callback.CallbackSingleMsgRead.Enable || req.ContentType != constant.Text {
+		return nil
+	}
+	req.CallbackCommand = cbapi.CallbackSingleMsgRead
+
+	resp := &cbapi.CallbackSingleMsgReadResp{}
+
+	if err := http.CallBackPostReturn(ctx, cbURL(), req, resp, config.Config.Callback.CallbackMsgModify); err != nil {
+		return err
+	}
+	return nil
+}
+func CallbackAfterRevokeMsg(ctx context.Context, req *pbchat.RevokeMsgReq) error {
+	if !config.Config.Callback.CallbackAfterRevokeMsg.Enable {
+		return nil
+	}
+	callbackReq := &cbapi.CallbackAfterRevokeMsgReq{
+		CallbackCommand: cbapi.CallbackAfterRevokeMsgCommand,
+		ConversationID:  req.ConversationID,
+		Seq:             req.Seq,
+		UserID:          req.UserID,
+	}
+	resp := &cbapi.CallbackAfterRevokeMsgResp{}
+	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, callbackReq, resp, config.Config.Callback.CallbackAfterRevokeMsg); err != nil {
+		return err
+	}
+	utils.StructFieldNotNilReplace(req, resp)
+	return nil
+}

internal/rpc/msg/revoke.go

@@ -61,6 +61,7 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
 	if msgs[0].ContentType == constant.MsgRevokeNotification {
 		return nil, errs.ErrMsgAlreadyRevoke.Wrap("msg already revoke")
 	}
+
 	data, _ := json.Marshal(msgs[0])
 	log.ZInfo(ctx, "GetMsgBySeqs", "conversationID", req.ConversationID, "seq", req.Seq, "msg", string(data))
 	var role int32
@@ -128,5 +129,8 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
 	if err := m.notificationSender.NotificationWithSesstionType(ctx, req.UserID, recvID, constant.MsgRevokeNotification, msgs[0].SessionType, &tips); err != nil {
 		return nil, err
 	}
+	if err = CallbackAfterRevokeMsg(ctx, req); err != nil {
+		return nil, err
+	}
 	return &msg.RevokeMsgResp{}, nil
 }

internal/rpc/msg/sync_msg.go

@@ -42,15 +42,8 @@ func (m *msgServer) PullMessageBySeqs(
 				log.ZError(ctx, "GetConversation error", err, "conversationID", seq.ConversationID)
 				continue
 			}
-			minSeq, maxSeq, msgs, err := m.MsgDatabase.GetMsgBySeqsRange(
+			minSeq, maxSeq, msgs, err := m.MsgDatabase.GetMsgBySeqsRange(ctx, req.UserID, seq.ConversationID,
-				ctx,
+				seq.Begin, seq.End, seq.Num, conversation.MaxSeq)
-				req.UserID,
-				seq.ConversationID,
-				seq.Begin,
-				seq.End,
-				seq.Num,
-				conversation.MaxSeq,
-			)
 			if err != nil {
 				log.ZWarn(ctx, "GetMsgBySeqsRange error", err, "conversationID", seq.ConversationID, "seq", seq)
 				continue
@@ -64,7 +57,6 @@ func (m *msgServer) PullMessageBySeqs(
 			}
 			if len(msgs) == 0 {
 				log.ZWarn(ctx, "not have msgs", nil, "conversationID", seq.ConversationID, "seq", seq)
-
 				continue
 			}
 			resp.Msgs[seq.ConversationID] = &sdkws.PullMsgs{Msgs: msgs, IsEnd: isEnd}

internal/rpc/third/third.go

@@ -17,14 +17,15 @@ package third
 import (
 	"context"
 	"fmt"
+
 	"net/url"
 	"time"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/s3"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/s3/cos"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/db/s3/kodo"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/s3/minio"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/s3/oss"
-
 	"google.golang.org/grpc"
 
 	"github.com/OpenIMSDK/protocol/third"
@@ -72,6 +73,8 @@ func Start(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) e
 		o, err = cos.NewCos()
 	case "oss":
 		o, err = oss.NewOSS()
+	case "kodo":
+		o, err = kodo.NewKodo()
 	default:
 		err = fmt.Errorf("invalid object enable: %s", enable)
 	}

internal/rpc/user/callback.go

@@ -16,11 +16,7 @@ package user
 
 import (
 	"context"
-
-	"github.com/OpenIMSDK/protocol/constant"
 	pbuser "github.com/OpenIMSDK/protocol/user"
-	"github.com/OpenIMSDK/tools/errs"
-	"github.com/OpenIMSDK/tools/mcontext"
 	"github.com/OpenIMSDK/tools/utils"
 
 	cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
@@ -33,17 +29,13 @@ func CallbackBeforeUpdateUserInfo(ctx context.Context, req *pbuser.UpdateUserInf
 		return nil
 	}
 	cbReq := &cbapi.CallbackBeforeUpdateUserInfoReq{
-		CallbackCommand: constant.CallbackBeforeUpdateUserInfoCommand,
+		CallbackCommand: cbapi.CallbackBeforeUpdateUserInfoCommand,
-		OperationID:     mcontext.GetOperationID(ctx),
 		UserID:          req.UserInfo.UserID,
 		FaceURL:         &req.UserInfo.FaceURL,
 		Nickname:        &req.UserInfo.Nickname,
 	}
 	resp := &cbapi.CallbackBeforeUpdateUserInfoResp{}
 	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeUpdateUserInfo); err != nil {
-		if err == errs.ErrCallbackContinue {
-			return nil
-		}
 		return err
 	}
 	utils.NotNilReplace(&req.UserInfo.FaceURL, resp.FaceURL)
@@ -51,3 +43,57 @@ func CallbackBeforeUpdateUserInfo(ctx context.Context, req *pbuser.UpdateUserInf
 	utils.NotNilReplace(&req.UserInfo.Nickname, resp.Nickname)
 	return nil
 }
+
+func CallbackAfterUpdateUserInfo(ctx context.Context, req *pbuser.UpdateUserInfoReq) error {
+	if !config.Config.Callback.CallbackAfterUpdateUserInfo.Enable {
+		return nil
+	}
+	cbReq := &cbapi.CallbackAfterUpdateUserInfoReq{
+		CallbackCommand: cbapi.CallbackAfterUpdateUserInfoCommand,
+		UserID:          req.UserInfo.UserID,
+		FaceURL:         req.UserInfo.FaceURL,
+		Nickname:        req.UserInfo.Nickname,
+	}
+	resp := &cbapi.CallbackAfterUpdateUserInfoResp{}
+	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeUpdateUserInfo); err != nil {
+		return err
+	}
+	return nil
+}
+
+func CallbackBeforeUserRegister(ctx context.Context, req *pbuser.UserRegisterReq) error {
+	if !config.Config.Callback.CallbackBeforeUserRegister.Enable {
+		return nil
+	}
+	cbReq := &cbapi.CallbackBeforeUserRegisterReq{
+		CallbackCommand: cbapi.CallbackBeforeUserRegisterCommand,
+		Secret:          req.Secret,
+		Users:           req.Users,
+	}
+
+	resp := &cbapi.CallbackBeforeUserRegisterResp{}
+	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeUpdateUserInfo); err != nil {
+		return err
+	}
+	if len(resp.Users) != 0 {
+		req.Users = resp.Users
+	}
+	return nil
+}
+
+func CallbackAfterUserRegister(ctx context.Context, req *pbuser.UserRegisterReq) error {
+	if !config.Config.Callback.CallbackAfterUserRegister.Enable {
+		return nil
+	}
+	cbReq := &cbapi.CallbackAfterUserRegisterReq{
+		CallbackCommand: cbapi.CallbackAfterUserRegisterCommand,
+		Secret:          req.Secret,
+		Users:           req.Users,
+	}
+
+	resp := &cbapi.CallbackAfterUserRegisterResp{}
+	if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackAfterUpdateUserInfo); err != nil {
+		return err
+	}
+	return nil
+}

internal/rpc/user/user.go

@@ -139,6 +139,9 @@ func (s *userServer) UpdateUserInfo(ctx context.Context, req *pbuser.UpdateUserI
 	for _, friendID := range friends {
 		s.friendNotificationSender.FriendInfoUpdatedNotification(ctx, req.UserInfo.UserID, friendID)
 	}
+	if err := CallbackAfterUpdateUserInfo(ctx, req); err != nil {
+		return nil, err
+	}
 	if err := s.groupRpcClient.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil {
 		log.ZError(ctx, "NotificationUserInfoUpdate", err, "userID", req.UserInfo.UserID)
 	}
@@ -230,6 +233,9 @@ func (s *userServer) UserRegister(ctx context.Context, req *pbuser.UserRegisterR
 	if exist {
 		return nil, errs.ErrRegisteredAlready.Wrap("userID registered already")
 	}
+	if err := CallbackBeforeUserRegister(ctx, req); err != nil {
+		return nil, err
+	}
 	now := time.Now()
 	users := make([]*tablerelation.UserModel, 0, len(req.Users))
 	for _, user := range req.Users {
@@ -246,6 +252,10 @@ func (s *userServer) UserRegister(ctx context.Context, req *pbuser.UserRegisterR
 	if err := s.Create(ctx, users); err != nil {
 		return nil, err
 	}
+
+	if err := CallbackAfterUserRegister(ctx, req); err != nil {
+		return nil, err
+	}
 	return resp, nil
 }
 

pkg/apistruct/msg.go

@@ -16,56 +16,56 @@ package apistruct
 
 type PictureBaseInfo struct {
 	UUID   string `mapstructure:"uuid"`
-	Type   string `mapstructure:"type"`
+	Type   string `mapstructure:"type" validate:"required"`
 	Size   int64  `mapstructure:"size"`
-	Width  int32  `mapstructure:"width"`
+	Width  int32  `mapstructure:"width" validate:"required"`
-	Height int32  `mapstructure:"height"`
+	Height int32  `mapstructure:"height" validate:"required"`
-	Url    string `mapstructure:"url"`
+	Url    string `mapstructure:"url" validate:"required"`
 }
 
 type PictureElem struct {
 	SourcePath      string          `mapstructure:"sourcePath"`
-	SourcePicture   PictureBaseInfo `mapstructure:"sourcePicture"`
+	SourcePicture   PictureBaseInfo `mapstructure:"sourcePicture"   validate:"required"`
-	BigPicture      PictureBaseInfo `mapstructure:"bigPicture"`
+	BigPicture      PictureBaseInfo `mapstructure:"bigPicture"      validate:"required"`
-	SnapshotPicture PictureBaseInfo `mapstructure:"snapshotPicture"`
+	SnapshotPicture PictureBaseInfo `mapstructure:"snapshotPicture" validate:"required"`
 }
 type SoundElem struct {
 	UUID      string `mapstructure:"uuid"`
 	SoundPath string `mapstructure:"soundPath"`
-	SourceURL string `mapstructure:"sourceUrl"`
+	SourceURL string `mapstructure:"sourceUrl" validate:"required"`
 	DataSize  int64  `mapstructure:"dataSize"`
-	Duration  int64  `mapstructure:"duration"`
+	Duration  int64  `mapstructure:"duration" validate:"required,min=1"`
 }
 type VideoElem struct {
-	VideoPath      string `mapstructure:"videoPath"`
+	VideoPath      string `mapstructure:"videoPath" `
 	VideoUUID      string `mapstructure:"videoUUID"`
-	VideoURL       string `mapstructure:"videoUrl"`
+	VideoURL       string `mapstructure:"videoUrl" validate:"required"`
-	VideoType      string `mapstructure:"videoType"`
+	VideoType      string `mapstructure:"videoType" validate:"required"`
-	VideoSize      int64  `mapstructure:"videoSize"`
+	VideoSize      int64  `mapstructure:"videoSize" validate:"required"`
-	Duration       int64  `mapstructure:"duration"`
+	Duration       int64  `mapstructure:"duration" validate:"required"`
 	SnapshotPath   string `mapstructure:"snapshotPath"`
 	SnapshotUUID   string `mapstructure:"snapshotUUID"`
 	SnapshotSize   int64  `mapstructure:"snapshotSize"`
-	SnapshotURL    string `mapstructure:"snapshotUrl"`
+	SnapshotURL    string `mapstructure:"snapshotUrl" validate:"required"`
-	SnapshotWidth  int32  `mapstructure:"snapshotWidth"`
+	SnapshotWidth  int32  `mapstructure:"snapshotWidth" validate:"required"`
-	SnapshotHeight int32  `mapstructure:"snapshotHeight"`
+	SnapshotHeight int32  `mapstructure:"snapshotHeight" validate:"required"`
 }
 type FileElem struct {
-	FilePath  string `mapstructure:"filePath"`
+	FilePath  string `mapstructure:"filePath" `
 	UUID      string `mapstructure:"uuid"`
-	SourceURL string `mapstructure:"sourceUrl"`
+	SourceURL string `mapstructure:"sourceUrl" validate:"required"`
-	FileName  string `mapstructure:"fileName"`
+	FileName  string `mapstructure:"fileName" validate:"required"`
-	FileSize  int64  `mapstructure:"fileSize"`
+	FileSize  int64  `mapstructure:"fileSize" validate:"required"`
 }
 type AtElem struct {
 	Text       string   `mapstructure:"text"`
-	AtUserList []string `mapstructure:"atUserList"`
+	AtUserList []string `mapstructure:"atUserList" validate:"required,max=1000"`
 	IsAtSelf   bool     `mapstructure:"isAtSelf"`
 }
 type LocationElem struct {
-	Description string  `mapstructure:"description"`
+	Description string  `mapstructure:"description" `
-	Longitude   float64 `mapstructure:"longitude"`
+	Longitude   float64 `mapstructure:"longitude" validate:"required"`
-	Latitude    float64 `mapstructure:"latitude"`
+	Latitude    float64 `mapstructure:"latitude" validate:"required"`
 }
 type CustomElem struct {
 	Data        string `mapstructure:"data"        validate:"required"`
@@ -80,18 +80,19 @@ type TextElem struct {
 type RevokeElem struct {
 	RevokeMsgClientID string `mapstructure:"revokeMsgClientID" validate:"required"`
 }
+
 type OANotificationElem struct {
-	NotificationName    string      `mapstructure:"notificationName"    json:"notificationName"    validate:"required"`
+	NotificationName    string       `mapstructure:"notificationName"    json:"notificationName"    validate:"required"`
-	NotificationFaceURL string      `mapstructure:"notificationFaceURL" json:"notificationFaceURL"`
+	NotificationFaceURL string       `mapstructure:"notificationFaceURL" json:"notificationFaceURL"`
-	NotificationType    int32       `mapstructure:"notificationType"    json:"notificationType"    validate:"required"`
+	NotificationType    int32        `mapstructure:"notificationType"    json:"notificationType"    validate:"required"`
-	Text                string      `mapstructure:"text"                json:"text"                validate:"required"`
+	Text                string       `mapstructure:"text"                json:"text"                validate:"required"`
-	Url                 string      `mapstructure:"url"                 json:"url"`
+	Url                 string       `mapstructure:"url"                 json:"url"`
-	MixType             int32       `mapstructure:"mixType"             json:"mixType"`
+	MixType             int32        `mapstructure:"mixType"             json:"mixType" validate:"required"`
-	PictureElem         PictureElem `mapstructure:"pictureElem"         json:"pictureElem"`
+	PictureElem         *PictureElem `mapstructure:"pictureElem"         json:"pictureElem"`
-	SoundElem           SoundElem   `mapstructure:"soundElem"           json:"soundElem"`
+	SoundElem           *SoundElem   `mapstructure:"soundElem"           json:"soundElem"`
-	VideoElem           VideoElem   `mapstructure:"videoElem"           json:"videoElem"`
+	VideoElem           *VideoElem   `mapstructure:"videoElem"           json:"videoElem"`
-	FileElem            FileElem    `mapstructure:"fileElem"            json:"fileElem"`
+	FileElem            *FileElem    `mapstructure:"fileElem"            json:"fileElem"`
-	Ex                  string      `mapstructure:"ex"                  json:"ex"`
+	Ex                  string       `mapstructure:"ex"                  json:"ex"`
 }
 type MessageRevoked struct {
 	RevokerID       string `mapstructure:"revokerID"       json:"revokerID"       validate:"required"`

pkg/callbackstruct/common.go

@@ -14,8 +14,10 @@
 
 package callbackstruct
 
-import (
+import "github.com/OpenIMSDK/tools/errs"
-	"github.com/OpenIMSDK/tools/errs"
+
+const (
+	Next = 1
 )
 
 type CommonCallbackReq struct {
@@ -51,14 +53,15 @@ type CallbackResp interface {
 }
 
 type CommonCallbackResp struct {
-	ActionCode int    `json:"actionCode"`
+	ActionCode int32  `json:"actionCode"`
 	ErrCode    int32  `json:"errCode"`
 	ErrMsg     string `json:"errMsg"`
 	ErrDlt     string `json:"errDlt"`
+	NextCode   int32  `json:"nextCode"`
 }
 
 func (c CommonCallbackResp) Parse() error {
-	if c.ActionCode != errs.NoError || c.ErrCode != errs.NoError {
+	if c.ActionCode != errs.NoError || c.NextCode == Next {
 		return errs.NewCodeError(int(c.ErrCode), c.ErrMsg).WithDetail(c.ErrDlt)
 	}
 	return nil

pkg/callbackstruct/constant.go

@@ -0,0 +1,49 @@
+package callbackstruct
+
+const CallbackBeforeInviteJoinGroupCommand = "callbackBeforeInviteJoinGroupCommand"
+const CallbackAfterJoinGroupCommand = "callbackAfterJoinGroupCommand"
+const CallbackAfterSetGroupInfoCommand = "callbackAfterSetGroupInfoCommand"
+const CallbackBeforeSetGroupInfoCommand = "callbackBeforeSetGroupInfoCommand"
+
+const CallbackAfterRevokeMsgCommand = "callbackBeforeAfterMsgCommand"
+const CallbackBeforeAddBlackCommand = "callbackBeforeAddBlackCommand"
+const CallbackAfterAddFriendCommand = "callbackAfterAddFriendCommand"
+const CallbackBeforeAddFriendAgreeCommand = "callbackBeforeAddFriendAgreeCommand"
+
+const CallbackAfterDeleteFriendCommand = "callbackAfterDeleteFriendCommand"
+const CallbackBeforeImportFriendsCommand = "callbackBeforeImportFriendsCommand"
+const CallbackAfterImportFriendsCommand = "callbackAfterImportFriendsCommand"
+const CallbackAfterRemoveBlackCommand = "callbackAfterRemoveBlackCommand"
+
+const (
+	CallbackQuitGroupCommand                = "callbackQuitGroupCommand"
+	CallbackKillGroupCommand                = "callbackKillGroupCommand"
+	CallbackDisMissGroupCommand             = "callbackDisMissGroupCommand"
+	CallbackBeforeJoinGroupCommand          = "callbackBeforeJoinGroupCommand"
+	CallbackGroupMsgReadCommand             = "callbackGroupMsgReadCommand"
+	CallbackMsgModifyCommand                = "callbackMsgModifyCommand"
+	CallbackAfterUpdateUserInfoCommand      = "callbackAfterUpdateUserInfoCommand"
+	CallbackBeforeUserRegisterCommand       = "callbackBeforeUserRegisterCommand"
+	CallbackAfterUserRegisterCommand        = "callbackAfterUserRegisterCommand"
+	CallbackTransferGroupOwnerAfter         = "callbackTransferGroupOwnerAfter"
+	CallbackBeforeSetFriendRemark           = "callbackBeforeSetFriendRemark"
+	CallbackAfterSetFriendRemark            = "callbackAfterSetFriendRemark"
+	CallbackSingleMsgRead                   = "callbackSingleMsgRead"
+	CallbackBeforeSendSingleMsgCommand      = "callbackBeforeSendSingleMsgCommand"
+	CallbackAfterSendSingleMsgCommand       = "callbackAfterSendSingleMsgCommand"
+	CallbackBeforeSendGroupMsgCommand       = "callbackBeforeSendGroupMsgCommand"
+	CallbackAfterSendGroupMsgCommand        = "callbackAfterSendGroupMsgCommand"
+	CallbackUserOnlineCommand               = "callbackUserOnlineCommand"
+	CallbackUserOfflineCommand              = "callbackUserOfflineCommand"
+	CallbackUserKickOffCommand              = "callbackUserKickOffCommand"
+	CallbackOfflinePushCommand              = "callbackOfflinePushCommand"
+	CallbackOnlinePushCommand               = "callbackOnlinePushCommand"
+	CallbackSuperGroupOnlinePushCommand     = "callbackSuperGroupOnlinePushCommand"
+	CallbackBeforeAddFriendCommand          = "callbackBeforeAddFriendCommand"
+	CallbackBeforeUpdateUserInfoCommand     = "callbackBeforeUpdateUserInfoCommand"
+	CallbackBeforeCreateGroupCommand        = "callbackBeforeCreateGroupCommand"
+	CallbackAfterCreateGroupCommand         = "callbackAfterCreateGroupCommand"
+	CallbackBeforeMemberJoinGroupCommand    = "callbackBeforeMemberJoinGroupCommand"
+	CallbackBeforeSetGroupMemberInfoCommand = "callbackBeforeSetGroupMemberInfoCommand"
+	CallbackAfterSetGroupMemberInfoCommand  = "callbackAfterSetGroupMemberInfoCommand"
+)

pkg/callbackstruct/friend.go

@@ -19,9 +19,109 @@ type CallbackBeforeAddFriendReq struct {
 	FromUserID      string `json:"fromUserID" `
 	ToUserID        string `json:"toUserID"`
 	ReqMsg          string `json:"reqMsg"`
-	OperationID     string `json:"operationID"`
+	Ex              string `json:"ex"`
 }
 
 type CallbackBeforeAddFriendResp struct {
 	CommonCallbackResp
 }
+
+type CallBackAddFriendReplyBeforeReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	FromUserID      string `json:"fromUserID" `
+	ToUserID        string `json:"toUserID"`
+}
+
+type CallBackAddFriendReplyBeforeResp struct {
+	CommonCallbackResp
+}
+
+type CallbackBeforeSetFriendRemarkReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	OwnerUserID     string `json:"ownerUserID"`
+	FriendUserID    string `json:"friendUserID"`
+	Remark          string `json:"remark"`
+}
+
+type CallbackBeforeSetFriendRemarkResp struct {
+	CommonCallbackResp
+	Remark string `json:"remark"`
+}
+
+type CallbackAfterSetFriendRemarkReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	OwnerUserID     string `json:"ownerUserID"`
+	FriendUserID    string `json:"friendUserID"`
+	Remark          string `json:"remark"`
+}
+
+type CallbackAfterSetFriendRemarkResp struct {
+	CommonCallbackResp
+}
+type CallbackAfterAddFriendReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	FromUserID      string `json:"fromUserID" `
+	ToUserID        string `json:"toUserID"`
+	ReqMsg          string `json:"reqMsg"`
+}
+
+type CallbackAfterAddFriendResp struct {
+	CommonCallbackResp
+}
+type CallbackBeforeAddBlackReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	OwnerUserID     string `json:"ownerUserID" `
+	BlackUserID     string `json:"blackUserID"`
+}
+
+type CallbackBeforeAddBlackResp struct {
+	CommonCallbackResp
+}
+
+type CallbackBeforeAddFriendAgreeReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	FromUserID      string `json:"fromUserID" `
+	ToUserID        string `json:"blackUserID"`
+	HandleResult    int32  `json:"HandleResult"`
+	HandleMsg       string `json:"HandleMsg"`
+}
+
+type CallbackBeforeAddFriendAgreeResp struct {
+	CommonCallbackResp
+}
+
+type CallbackAfterDeleteFriendReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	OwnerUserID     string `json:"ownerUserID" `
+	FriendUserID    string `json:"friendUserID"`
+}
+type CallbackAfterDeleteFriendResp struct {
+	CommonCallbackResp
+}
+
+type CallbackBeforeImportFriendsReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	OwnerUserID     string   `json:"ownerUserID" `
+	FriendUserIDs   []string `json:"friendUserIDs"`
+}
+type CallbackBeforeImportFriendsResp struct {
+	CommonCallbackResp
+	FriendUserIDs []string `json:"friendUserIDs"`
+}
+type CallbackAfterImportFriendsReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	OwnerUserID     string   `json:"ownerUserID" `
+	FriendUserIDs   []string `json:"friendUserIDs"`
+}
+type CallbackAfterImportFriendsResp struct {
+	CommonCallbackResp
+}
+
+type CallbackAfterRemoveBlackReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	OwnerUserID     string `json:"ownerUserID"`
+	BlackUserID     string `json:"blackUserID"`
+}
+type CallbackAfterRemoveBlackResp struct {
+	CommonCallbackResp
+}

pkg/callbackstruct/group.go

@@ -50,9 +50,18 @@ type CallbackBeforeCreateGroupResp struct {
 	ApplyMemberFriend *int32  `json:"applyMemberFriend"`
 }
 
+type CallbackAfterCreateGroupReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	*common.GroupInfo
+	InitMemberList []*apistruct.GroupAddMemberInfo `json:"initMemberList"`
+}
+
+type CallbackAfterCreateGroupResp struct {
+	CommonCallbackResp
+}
+
 type CallbackBeforeMemberJoinGroupReq struct {
 	CallbackCommand `json:"callbackCommand"`
-	OperationID     string `json:"operationID"`
 	GroupID         string `json:"groupID"`
 	UserID          string `json:"userID"`
 	Ex              string `json:"ex"`
@@ -70,7 +79,6 @@ type CallbackBeforeMemberJoinGroupResp struct {
 
 type CallbackBeforeSetGroupMemberInfoReq struct {
 	CallbackCommand `json:"callbackCommand"`
-	OperationID     string  `json:"operationID"`
 	GroupID         string  `json:"groupID"`
 	UserID          string  `json:"userID"`
 	Nickname        *string `json:"nickName"`
@@ -86,3 +94,142 @@ type CallbackBeforeSetGroupMemberInfoResp struct {
 	FaceURL   *string `json:"faceURL"`
 	RoleLevel *int32  `json:"roleLevel"`
 }
+
+type CallbackAfterSetGroupMemberInfoReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	GroupID         string  `json:"groupID"`
+	UserID          string  `json:"userID"`
+	Nickname        *string `json:"nickName"`
+	FaceURL         *string `json:"faceURL"`
+	RoleLevel       *int32  `json:"roleLevel"`
+	Ex              *string `json:"ex"`
+}
+
+type CallbackAfterSetGroupMemberInfoResp struct {
+	CommonCallbackResp
+}
+
+type CallbackQuitGroupReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	GroupID         string `json:"groupID"`
+	UserID          string `json:"userID"`
+}
+
+type CallbackQuitGroupResp struct {
+	CommonCallbackResp
+}
+
+type CallbackKillGroupMemberReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	GroupID         string   `json:"groupID"`
+	KickedUserIDs   []string `json:"kickedUserIDs"`
+	Reason          string   `json:"reason"`
+}
+
+type CallbackKillGroupMemberResp struct {
+	CommonCallbackResp
+}
+
+type CallbackDisMissGroupReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	GroupID         string   `json:"groupID"`
+	OwnerID         string   `json:"ownerID"`
+	GroupType       string   `json:"groupType"`
+	MembersID       []string `json:"membersID"`
+}
+
+type CallbackDisMissGroupResp struct {
+	CommonCallbackResp
+}
+
+type CallbackJoinGroupReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	GroupID         string `json:"groupID"`
+	GroupType       string `json:"groupType"`
+	ApplyID         string `json:"applyID"`
+	ReqMessage      string `json:"reqMessage"`
+}
+
+type CallbackJoinGroupResp struct {
+	CommonCallbackResp
+}
+
+type CallbackTransferGroupOwnerReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	GroupID         string `json:"groupID"`
+	OldOwnerUserID  string `json:"oldOwnerUserID"`
+	NewOwnerUserID  string `json:"newOwnerUserID"`
+}
+
+type CallbackTransferGroupOwnerResp struct {
+	CommonCallbackResp
+}
+
+type CallbackBeforeInviteUserToGroupReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	OperationID     string   `json:"operationID"`
+	GroupID         string   `json:"groupID"`
+	Reason          string   `json:"reason"`
+	InvitedUserIDs  []string `json:"invitedUserIDs"`
+}
+type CallbackBeforeInviteUserToGroupResp struct {
+	CommonCallbackResp
+	RefusedMembersAccount []string `json:"refusedMembersAccount,omitempty"` // Optional field to list members whose invitation is refused.
+}
+
+type CallbackAfterJoinGroupReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	OperationID     string `json:"operationID"`
+	GroupID         string `json:"groupID"`
+	ReqMessage      string `json:"reqMessage"`
+	JoinSource      int32  `json:"joinSource"`
+	InviterUserID   string `json:"inviterUserID"`
+}
+type CallbackAfterJoinGroupResp struct {
+	CommonCallbackResp
+}
+
+type CallbackBeforeSetGroupInfoReq struct {
+	CallbackCommand   `json:"callbackCommand"`
+	OperationID       string `json:"operationID"`
+	GroupID           string `json:"groupID"`
+	GroupName         string `json:"groupName"`
+	Notification      string `json:"notification"`
+	Introduction      string `json:"introduction"`
+	FaceURL           string `json:"faceURL"`
+	Ex                string `json:"ex"`
+	NeedVerification  int32  `json:"needVerification"`
+	LookMemberInfo    int32  `json:"lookMemberInfo"`
+	ApplyMemberFriend int32  `json:"applyMemberFriend"`
+}
+
+type CallbackBeforeSetGroupInfoResp struct {
+	CommonCallbackResp
+	GroupID           string  ` json:"groupID"`
+	GroupName         string  `json:"groupName"`
+	Notification      string  `json:"notification"`
+	Introduction      string  `json:"introduction"`
+	FaceURL           string  `json:"faceURL"`
+	Ex                *string `json:"ex"`
+	NeedVerification  *int32  `json:"needVerification"`
+	LookMemberInfo    *int32  `json:"lookMemberInfo"`
+	ApplyMemberFriend *int32  `json:"applyMemberFriend"`
+}
+
+type CallbackAfterSetGroupInfoReq struct {
+	CallbackCommand   `json:"callbackCommand"`
+	OperationID       string  `json:"operationID"`
+	GroupID           string  `json:"groupID"`
+	GroupName         string  `json:"groupName"`
+	Notification      string  `json:"notification"`
+	Introduction      string  `json:"introduction"`
+	FaceURL           string  `json:"faceURL"`
+	Ex                *string `json:"ex"`
+	NeedVerification  *int32  `json:"needVerification"`
+	LookMemberInfo    *int32  `json:"lookMemberInfo"`
+	ApplyMemberFriend *int32  `json:"applyMemberFriend"`
+}
+
+type CallbackAfterSetGroupInfoResp struct {
+	CommonCallbackResp
+}

pkg/callbackstruct/message.go

@@ -79,3 +79,26 @@ type CallbackMsgModifyCommandResp struct {
 	AttachedInfo     *string                `json:"attachedInfo"`
 	Ex               *string                `json:"ex"`
 }
+
+type CallbackGroupMsgReadReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	SendID          string `json:"sendID"`
+	ReceiveID       string `json:"receiveID"`
+	UnreadMsgNum    int64  `json:"unreadMsgNum"`
+	ContentType     int64  `json:"contentType"`
+}
+
+type CallbackGroupMsgReadResp struct {
+	CommonCallbackResp
+}
+
+type CallbackSingleMsgReadReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	SendID          string `json:"sendID"`
+	ReceiveID       string `json:"receiveID"`
+	ContentType     int64  `json:"contentType"`
+}
+
+type CallbackSingleMsgReadResp struct {
+	CommonCallbackResp
+}

pkg/callbackstruct/revoke.go

@@ -0,0 +1,11 @@
+package callbackstruct
+
+type CallbackAfterRevokeMsgReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	ConversationID  string `json:"conversationID"`
+	Seq             int64  `json:"seq"`
+	UserID          string `json:"userID"`
+}
+type CallbackAfterRevokeMsgResp struct {
+	CommonCallbackResp
+}

pkg/callbackstruct/user.go

@@ -14,9 +14,10 @@
 
 package callbackstruct
 
+import "github.com/OpenIMSDK/protocol/sdkws"
+
 type CallbackBeforeUpdateUserInfoReq struct {
 	CallbackCommand `json:"callbackCommand"`
-	OperationID     string  `json:"operationID"`
 	UserID          string  `json:"userID"`
 	Nickname        *string `json:"nickName"`
 	FaceURL         *string `json:"faceURL"`
@@ -28,3 +29,35 @@ type CallbackBeforeUpdateUserInfoResp struct {
 	FaceURL  *string `json:"faceURL"`
 	Ex       *string `json:"ex"`
 }
+
+type CallbackAfterUpdateUserInfoReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	UserID          string `json:"userID"`
+	Nickname        string `json:"nickName"`
+	FaceURL         string `json:"faceURL"`
+	Ex              string `json:"ex"`
+}
+type CallbackAfterUpdateUserInfoResp struct {
+	CommonCallbackResp
+}
+
+type CallbackBeforeUserRegisterReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	Secret          string            `json:"secret"`
+	Users           []*sdkws.UserInfo `json:"users"`
+}
+
+type CallbackBeforeUserRegisterResp struct {
+	CommonCallbackResp
+	Users []*sdkws.UserInfo `json:"users"`
+}
+
+type CallbackAfterUserRegisterReq struct {
+	CallbackCommand `json:"callbackCommand"`
+	Secret          string            `json:"secret"`
+	Users           []*sdkws.UserInfo `json:"users"`
+}
+
+type CallbackAfterUserRegisterResp struct {
+	CommonCallbackResp
+}

pkg/common/cmd/msg_gateway.go

@@ -17,12 +17,12 @@ package cmd
 import (
 	"log"
 
-	"github.com/openimsdk/open-im-server/v3/internal/msggateway"
-	v3config "github.com/openimsdk/open-im-server/v3/pkg/common/config"
-
 	"github.com/spf13/cobra"
 
 	"github.com/OpenIMSDK/protocol/constant"
+
+	"github.com/openimsdk/open-im-server/v3/internal/msggateway"
+	v3config "github.com/openimsdk/open-im-server/v3/pkg/common/config"
 )
 
 type MsgGatewayCmd struct {
@@ -60,14 +60,19 @@ func (m *MsgGatewayCmd) Exec() error {
 	m.addRunE()
 	return m.Execute()
 }
+
 func (m *MsgGatewayCmd) GetPortFromConfig(portType string) int {
-	if portType == constant.FlagWsPort {
+	switch portType {
+	case constant.FlagWsPort:
 		return v3config.Config.LongConnSvr.OpenImWsPort[0]
-	} else if portType == constant.FlagPort {
+
+	case constant.FlagPort:
 		return v3config.Config.LongConnSvr.OpenImMessageGatewayPort[0]
-	} else if portType == constant.FlagPrometheusPort {
+
+	case constant.FlagPrometheusPort:
 		return v3config.Config.Prometheus.MessageGatewayPrometheusPort[0]
-	} else {
+
+	default:
 		return 0
 	}
 }

pkg/common/config/config.go

@@ -16,7 +16,6 @@ package config
 
 import (
 	"bytes"
-
 	"github.com/OpenIMSDK/tools/discoveryregistry"
 	"gopkg.in/yaml.v3"
 )
@@ -153,6 +152,15 @@ type configStruct struct {
 			SessionToken    string `yaml:"sessionToken"`
 			PublicRead      bool   `yaml:"publicRead"`
 		} `yaml:"oss"`
+		Kodo struct {
+			Endpoint        string `yaml:"endpoint"`
+			Bucket          string `yaml:"bucket"`
+			BucketURL       string `yaml:"bucketURL"`
+			AccessKeyID     string `yaml:"accessKeyID"`
+			AccessKeySecret string `yaml:"accessKeySecret"`
+			SessionToken    string `yaml:"sessionToken"`
+			PublicRead      bool   `yaml:"publicRead"`
+		} `yaml:"kodo"`
 	} `yaml:"object"`
 
 	RpcPort struct {
@@ -254,6 +262,8 @@ type configStruct struct {
 		CallbackBeforeSendGroupMsg         CallBackConfig `yaml:"beforeSendGroupMsg"`
 		CallbackAfterSendGroupMsg          CallBackConfig `yaml:"afterSendGroupMsg"`
 		CallbackMsgModify                  CallBackConfig `yaml:"msgModify"`
+		CallbackSingleMsgRead              CallBackConfig `yaml:"singleMsgRead"`
+		CallbackGroupMsgRead               CallBackConfig `yaml:"groupMsgRead"`
 		CallbackUserOnline                 CallBackConfig `yaml:"userOnline"`
 		CallbackUserOffline                CallBackConfig `yaml:"userOffline"`
 		CallbackUserKickOff                CallBackConfig `yaml:"userKickOff"`
@@ -261,10 +271,35 @@ type configStruct struct {
 		CallbackOnlinePush                 CallBackConfig `yaml:"onlinePush"`
 		CallbackBeforeSuperGroupOnlinePush CallBackConfig `yaml:"superGroupOnlinePush"`
 		CallbackBeforeAddFriend            CallBackConfig `yaml:"beforeAddFriend"`
+		CallbackBeforeSetFriendRemark      CallBackConfig `yaml:"callbackBeforeSetFriendRemark"`
+		CallbackAfterSetFriendRemark       CallBackConfig `yaml:"callbackAfterSetFriendRemark"`
 		CallbackBeforeUpdateUserInfo       CallBackConfig `yaml:"beforeUpdateUserInfo"`
+		CallbackBeforeUserRegister         CallBackConfig `yaml:"beforeUserRegister"`
+		CallbackAfterUpdateUserInfo        CallBackConfig `yaml:"updateUserInfo"`
+		CallbackAfterUserRegister          CallBackConfig `yaml:"afterUserRegister"`
 		CallbackBeforeCreateGroup          CallBackConfig `yaml:"beforeCreateGroup"`
+		CallbackAfterCreateGroup           CallBackConfig `yaml:"afterCreateGroup"`
 		CallbackBeforeMemberJoinGroup      CallBackConfig `yaml:"beforeMemberJoinGroup"`
 		CallbackBeforeSetGroupMemberInfo   CallBackConfig `yaml:"beforeSetGroupMemberInfo"`
+		CallbackAfterSetGroupMemberInfo    CallBackConfig `yaml:"afterSetGroupMemberInfo"`
+		CallbackQuitGroup                  CallBackConfig `yaml:"quitGroup"`
+		CallbackKillGroupMember            CallBackConfig `yaml:"killGroupMember"`
+		CallbackDismissGroup               CallBackConfig `yaml:"dismissGroup"`
+		CallbackBeforeJoinGroup            CallBackConfig `yaml:"joinGroup"`
+		CallbackTransferGroupOwnerAfter    CallBackConfig `yaml:"transferGroupOwner"`
+		CallbackBeforeInviteUserToGroup    CallBackConfig `yaml:"beforeInviteUserToGroup"`
+		CallbackAfterJoinGroup             CallBackConfig `yaml:"joinGroupAfter"`
+		CallbackAfterSetGroupInfo          CallBackConfig `yaml:"setGroupInfoAfter"`
+		CallbackBeforeSetGroupInfo         CallBackConfig `yaml:"setGroupInfoBefore"`
+		CallbackAfterRevokeMsg             CallBackConfig `yaml:"revokeMsgAfter"`
+		CallbackBeforeAddBlack             CallBackConfig `yaml:"addBlackBefore"`
+		CallbackAfterAddFriend             CallBackConfig `yaml:"addFriendAfter"`
+		CallbackBeforeAddFriendAgree       CallBackConfig `yaml:"addFriendAgreeBefore"`
+
+		CallbackAfterDeleteFriend   CallBackConfig `yaml:"deleteFriendAfter"`
+		CallbackBeforeImportFriends CallBackConfig `yaml:"importFriendsBefore"`
+		CallbackAfterImportFriends  CallBackConfig `yaml:"importFriendsAfter"`
+		CallbackAfterRemoveBlack    CallBackConfig `yaml:"removeBlackAfter"`
 	} `yaml:"callback"`
 
 	Prometheus struct {

pkg/common/convert/user.go

@@ -15,8 +15,6 @@
 package convert
 
 import (
-	"time"
-
 	"github.com/OpenIMSDK/protocol/sdkws"
 
 	relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
@@ -43,7 +41,6 @@ func UserPb2DB(user *sdkws.UserInfo) *relationtb.UserModel {
 	userDB.Nickname = user.Nickname
 	userDB.FaceURL = user.FaceURL
 	userDB.Ex = user.Ex
-	userDB.CreateTime = time.UnixMilli(user.CreateTime)
 	userDB.AppMangerLevel = user.AppMangerLevel
 	userDB.GlobalRecvMsgOpt = user.GlobalRecvMsgOpt
 	return &userDB

pkg/common/db/cache/msg.go

@@ -173,7 +173,20 @@ func (c *msgCache) getSeqs(ctx context.Context, items []string, getkey func(s st
 }
 
 func (c *msgCache) SetMaxSeq(ctx context.Context, conversationID string, maxSeq int64) error {
-	return c.setSeq(ctx, conversationID, maxSeq, c.getMaxSeqKey)
+	var retErr error
+	for {
+		select {
+		case <-ctx.Done():
+			return errs.Wrap(retErr, "SetMaxSeq redis retry too many amount")
+		default:
+			retErr = c.setSeq(ctx, conversationID, maxSeq, c.getMaxSeqKey)
+			if retErr != nil {
+				time.Sleep(time.Second * 2)
+				continue
+			}
+			return nil
+		}
+	}
 }
 
 func (c *msgCache) GetMaxSeqs(ctx context.Context, conversationIDs []string) (m map[string]int64, err error) {
@@ -181,7 +194,21 @@ func (c *msgCache) GetMaxSeqs(ctx context.Context, conversationIDs []string) (m
 }
 
 func (c *msgCache) GetMaxSeq(ctx context.Context, conversationID string) (int64, error) {
-	return c.getSeq(ctx, conversationID, c.getMaxSeqKey)
+	var retErr error
+	var retData int64
+	for {
+		select {
+		case <-ctx.Done():
+			return -1, errs.Wrap(retErr, "GetMaxSeq redis retry too many amount")
+		default:
+			retData, retErr = c.getSeq(ctx, conversationID, c.getMaxSeqKey)
+			if retErr != nil && errs.Unwrap(retErr) != redis.Nil {
+				time.Sleep(time.Second * 2)
+				continue
+			}
+			return retData, retErr
+		}
+	}
 }
 
 func (c *msgCache) SetMinSeq(ctx context.Context, conversationID string, minSeq int64) error {
@@ -645,19 +672,35 @@ func (c *msgCache) PipeDeleteMessages(ctx context.Context, conversationID string
 }
 
 func (c *msgCache) CleanUpOneConversationAllMsg(ctx context.Context, conversationID string) error {
-	vals, err := c.rdb.Keys(ctx, c.allMessageCacheKey(conversationID)).Result()
+	var (
-	if errors.Is(err, redis.Nil) {
+		cursor uint64
-		return nil
+		keys   []string
-	}
+		err    error
-	if err != nil {
+
-		return errs.Wrap(err)
+		key = c.allMessageCacheKey(conversationID)
-	}
+	)
-	for _, v := range vals {
+
-		if err := c.rdb.Del(ctx, v).Err(); err != nil {
+	for {
+		// scan up to 10000 at a time, the count (10000) param refers to the number of scans on redis server.
+		// if the count is too small, needs to be run scan on redis frequently.
+		var limit int64 = 10000
+		keys, cursor, err = c.rdb.Scan(ctx, cursor, key, limit).Result()
+		if err != nil {
 			return errs.Wrap(err)
 		}
+
+		for _, key := range keys {
+			err := c.rdb.Del(ctx, key).Err()
+			if err != nil {
+				return errs.Wrap(err)
+			}
+		}
+
+		// scan end
+		if cursor == 0 {
+			return nil
+		}
 	}
-	return nil
 }
 
 func (c *msgCache) DelMsgFromCache(ctx context.Context, userID string, seqs []int64) error {

pkg/common/db/cache/msg_test.go

@@ -385,3 +385,50 @@ func testParallelDeleteMessagesMix(t *testing.T, cid string, seqs []int64, input
 		assert.EqualValues(t, 1, val) // exists
 	}
 }
+
+func TestCleanUpOneConversationAllMsg(t *testing.T) {
+	rdb := redis.NewClient(&redis.Options{})
+	defer rdb.Close()
+
+	cacher := msgCache{rdb: rdb}
+	count := 1000
+	prefix := fmt.Sprintf("%v", rand.Int63())
+
+	ids := []string{}
+	for i := 0; i < count; i++ {
+		id := fmt.Sprintf("%v-cid-%v", prefix, rand.Int63())
+		ids = append(ids, id)
+
+		key := cacher.allMessageCacheKey(id)
+		rdb.Set(context.Background(), key, "openim", 0)
+	}
+
+	// delete 100 keys with scan.
+	for i := 0; i < 100; i++ {
+		pickedKey := ids[i]
+		err := cacher.CleanUpOneConversationAllMsg(context.Background(), pickedKey)
+		assert.Nil(t, err)
+
+		ls, err := rdb.Keys(context.Background(), pickedKey).Result()
+		assert.Nil(t, err)
+		assert.Equal(t, 0, len(ls))
+
+		rcode, err := rdb.Exists(context.Background(), pickedKey).Result()
+		assert.Nil(t, err)
+		assert.EqualValues(t, 0, rcode) // non-exists
+	}
+
+	sid := fmt.Sprintf("%v-cid-*", prefix)
+	ls, err := rdb.Keys(context.Background(), cacher.allMessageCacheKey(sid)).Result()
+	assert.Nil(t, err)
+	assert.Equal(t, count-100, len(ls))
+
+	// delete fuzzy matching keys.
+	err = cacher.CleanUpOneConversationAllMsg(context.Background(), sid)
+	assert.Nil(t, err)
+
+	// don't contains keys matched `{prefix}-cid-{random}` on redis
+	ls, err = rdb.Keys(context.Background(), cacher.allMessageCacheKey(sid)).Result()
+	assert.Nil(t, err)
+	assert.Equal(t, 0, len(ls))
+}

pkg/common/db/controller/msg.go

@@ -357,7 +357,9 @@ func (db *commonMsgDatabase) DelUserDeleteMsgsList(ctx context.Context, conversa
 }
 
 func (db *commonMsgDatabase) BatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (seq int64, isNew bool, err error) {
-	currentMaxSeq, err := db.cache.GetMaxSeq(ctx, conversationID)
+	cancelCtx, cancel := context.WithTimeout(ctx, 1*time.Minute)
+	defer cancel()
+	currentMaxSeq, err := db.cache.GetMaxSeq(cancelCtx, conversationID)
 	if err != nil && errs.Unwrap(err) != redis.Nil {
 		log.ZError(ctx, "db.cache.GetMaxSeq", err)
 		return 0, false, err
@@ -384,19 +386,21 @@ func (db *commonMsgDatabase) BatchInsertChat2Cache(ctx context.Context, conversa
 		prommetrics.MsgInsertRedisFailedCounter.Add(float64(failedNum))
 		log.ZError(ctx, "setMessageToCache error", err, "len", len(msgs), "conversationID", conversationID)
 	} else {
-		prommetrics.MsgInsertRedisSuccessCounter.Inc()
+		prommetrics.MsgInsertRedisSuccessCounter.Add(float64(len(msgs)))
 	}
-	err = db.cache.SetMaxSeq(ctx, conversationID, currentMaxSeq)
+	cancelCtx, cancel = context.WithTimeout(ctx, 1*time.Minute)
+	defer cancel()
+	err = db.cache.SetMaxSeq(cancelCtx, conversationID, currentMaxSeq)
 	if err != nil {
 		log.ZError(ctx, "db.cache.SetMaxSeq error", err, "conversationID", conversationID)
 		prommetrics.SeqSetFailedCounter.Inc()
 	}
 	err2 := db.cache.SetHasReadSeqs(ctx, conversationID, userSeqMap)
-	if err != nil {
+	if err2 != nil {
 		log.ZError(ctx, "SetHasReadSeqs error", err2, "userSeqMap", userSeqMap, "conversationID", conversationID)
 		prommetrics.SeqSetFailedCounter.Inc()
 	}
-	return lastMaxSeq, isNew, utils.Wrap(err, "")
+	return lastMaxSeq, isNew, errs.Wrap(err, "redis SetMaxSeq error")
 }
 
 func (db *commonMsgDatabase) getMsgBySeqs(ctx context.Context, userID, conversationID string, seqs []int64) (totalMsgs []*sdkws.MsgData, err error) {
@@ -654,16 +658,26 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin
 
 func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) (int64, int64, []*sdkws.MsgData, error) {
 	userMinSeq, err := db.cache.GetConversationUserMinSeq(ctx, conversationID, userID)
-	if err != nil && errs.Unwrap(err) != redis.Nil {
+	if err != nil {
-		return 0, 0, nil, err
+		log.ZError(ctx, "cache.GetConversationUserMinSeq error", err)
+		if errs.Unwrap(err) != redis.Nil {
+			return 0, 0, nil, err
+		}
 	}
 	minSeq, err := db.cache.GetMinSeq(ctx, conversationID)
-	if err != nil && errs.Unwrap(err) != redis.Nil {
+	if err != nil {
-		return 0, 0, nil, err
+		log.ZError(ctx, "cache.GetMinSeq error", err)
+		if errs.Unwrap(err) != redis.Nil {
+			return 0, 0, nil, err
+		}
 	}
 	maxSeq, err := db.cache.GetMaxSeq(ctx, conversationID)
-	if err != nil && errs.Unwrap(err) != redis.Nil {
+	if err != nil {
-		return 0, 0, nil, err
+		log.ZError(ctx, "cache.GetMaxSeq error", err)
+		if errs.Unwrap(err) != redis.Nil {
+			return 0, 0, nil, err
+		}
+
 	}
 	if userMinSeq < minSeq {
 		minSeq = userMinSeq
@@ -676,34 +690,16 @@ func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, co
 	}
 	successMsgs, failedSeqs, err := db.cache.GetMessagesBySeq(ctx, conversationID, newSeqs)
 	if err != nil {
-		if err != redis.Nil {
+		log.ZError(ctx, "get message from redis exception", err, "failedSeqs", failedSeqs, "conversationID", conversationID)
-			log.ZError(ctx, "get message from redis exception", err, "failedSeqs", failedSeqs, "conversationID", conversationID)
-		}
 	}
-	log.ZInfo(
+	log.ZInfo(ctx, "db.cache.GetMessagesBySeq", "userID", userID, "conversationID", conversationID, "seqs", seqs, "successMsgs",
-		ctx,
+		len(successMsgs), "failedSeqs", failedSeqs, "conversationID", conversationID)
-		"db.cache.GetMessagesBySeq",
-		"userID",
-		userID,
-		"conversationID",
-		conversationID,
-		"seqs",
-		seqs,
-		"successMsgs",
-		len(successMsgs),
-		"failedSeqs",
-		failedSeqs,
-		"conversationID",
-		conversationID,
-	)
 
 	if len(failedSeqs) > 0 {
 		mongoMsgs, err := db.getMsgBySeqs(ctx, userID, conversationID, failedSeqs)
 		if err != nil {
-
 			return 0, 0, nil, err
 		}
-
 		successMsgs = append(successMsgs, mongoMsgs...)
 	}
 	return minSeq, maxSeq, successMsgs, nil

pkg/common/db/s3/kodo/internal.go

@@ -0,0 +1 @@
+package kodo

pkg/common/db/s3/kodo/kodo.go

@@ -0,0 +1,323 @@
+package kodo
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	awss3config "github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+	awss3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/db/s3"
+	"github.com/qiniu/go-sdk/v7/auth"
+	"github.com/qiniu/go-sdk/v7/storage"
+)
+
+const (
+	minPartSize = 1024 * 1024 * 1        // 1MB
+	maxPartSize = 1024 * 1024 * 1024 * 5 // 5GB
+	maxNumSize  = 10000
+)
+
+type Kodo struct {
+	AccessKey     string
+	SecretKey     string
+	Region        string
+	Token         string
+	Endpoint      string
+	BucketURL     string
+	Auth          *auth.Credentials
+	Client        *awss3.Client
+	PresignClient *awss3.PresignClient
+}
+
+func NewKodo() (s3.Interface, error) {
+	conf := config.Config.Object.Kodo
+	//init client
+	cfg, err := awss3config.LoadDefaultConfig(context.TODO(),
+		awss3config.WithRegion(conf.Bucket),
+		awss3config.WithEndpointResolverWithOptions(
+			aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {
+				return aws.Endpoint{URL: conf.Endpoint}, nil
+			})),
+		awss3config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(
+			conf.AccessKeyID,
+			conf.AccessKeySecret,
+			conf.SessionToken),
+		),
+	)
+	if err != nil {
+		panic(err)
+	}
+	client := awss3.NewFromConfig(cfg)
+	presignClient := awss3.NewPresignClient(client)
+
+	return &Kodo{
+		AccessKey:     conf.AccessKeyID,
+		SecretKey:     conf.AccessKeySecret,
+		Region:        conf.Bucket,
+		BucketURL:     conf.BucketURL,
+		Auth:          auth.New(conf.AccessKeyID, conf.AccessKeySecret),
+		Client:        client,
+		PresignClient: presignClient,
+	}, nil
+}
+
+func (k Kodo) Engine() string {
+	return "kodo"
+}
+
+func (k Kodo) PartLimit() *s3.PartLimit {
+	return &s3.PartLimit{
+		MinPartSize: minPartSize,
+		MaxPartSize: maxPartSize,
+		MaxNumSize:  maxNumSize,
+	}
+}
+
+func (k Kodo) InitiateMultipartUpload(ctx context.Context, name string) (*s3.InitiateMultipartUploadResult, error) {
+	result, err := k.Client.CreateMultipartUpload(ctx, &awss3.CreateMultipartUploadInput{
+		Bucket: aws.String(k.Region),
+		Key:    aws.String(name),
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &s3.InitiateMultipartUploadResult{
+		UploadID: aws.ToString(result.UploadId),
+		Bucket:   aws.ToString(result.Bucket),
+		Key:      aws.ToString(result.Key),
+	}, nil
+}
+
+func (k Kodo) CompleteMultipartUpload(ctx context.Context, uploadID string, name string, parts []s3.Part) (*s3.CompleteMultipartUploadResult, error) {
+	kodoParts := make([]awss3types.CompletedPart, len(parts))
+	for i, part := range parts {
+		kodoParts[i] = awss3types.CompletedPart{
+			PartNumber: aws.Int32(int32(part.PartNumber)),
+			ETag:       aws.String(part.ETag),
+		}
+	}
+	result, err := k.Client.CompleteMultipartUpload(ctx, &awss3.CompleteMultipartUploadInput{
+		Bucket:          aws.String(k.Region),
+		Key:             aws.String(name),
+		UploadId:        aws.String(uploadID),
+		MultipartUpload: &awss3types.CompletedMultipartUpload{Parts: kodoParts},
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &s3.CompleteMultipartUploadResult{
+		Location: aws.ToString(result.Location),
+		Bucket:   aws.ToString(result.Bucket),
+		Key:      aws.ToString(result.Key),
+		ETag:     strings.ToLower(strings.ReplaceAll(aws.ToString(result.ETag), `"`, ``)),
+	}, nil
+}
+
+func (k Kodo) PartSize(ctx context.Context, size int64) (int64, error) {
+	if size <= 0 {
+		return 0, errors.New("size must be greater than 0")
+	}
+	if size > maxPartSize*maxNumSize {
+		return 0, fmt.Errorf("size must be less than %db", maxPartSize*maxNumSize)
+	}
+	if size <= minPartSize*maxNumSize {
+		return minPartSize, nil
+	}
+	partSize := size / maxNumSize
+	if size%maxNumSize != 0 {
+		partSize++
+	}
+	return partSize, nil
+}
+
+func (k Kodo) AuthSign(ctx context.Context, uploadID string, name string, expire time.Duration, partNumbers []int) (*s3.AuthSignResult, error) {
+	result := s3.AuthSignResult{
+		URL:    k.BucketURL + "/" + name,
+		Query:  url.Values{"uploadId": {uploadID}},
+		Header: make(http.Header),
+		Parts:  make([]s3.SignPart, len(partNumbers)),
+	}
+	for i, partNumber := range partNumbers {
+		part, _ := k.PresignClient.PresignUploadPart(ctx, &awss3.UploadPartInput{
+			Bucket:     aws.String(k.Region),
+			UploadId:   aws.String(uploadID),
+			Key:        aws.String(name),
+			PartNumber: aws.Int32(int32(partNumber)),
+		})
+		result.Parts[i] = s3.SignPart{
+			PartNumber: partNumber,
+			URL:        part.URL,
+			Header:     part.SignedHeader,
+		}
+	}
+	return &result, nil
+
+}
+
+func (k Kodo) PresignedPutObject(ctx context.Context, name string, expire time.Duration) (string, error) {
+	object, err := k.PresignClient.PresignPutObject(ctx, &awss3.PutObjectInput{
+		Bucket: aws.String(k.Region),
+		Key:    aws.String(name),
+	}, func(po *awss3.PresignOptions) {
+		po.Expires = expire
+	})
+	return object.URL, err
+
+}
+
+func (k Kodo) DeleteObject(ctx context.Context, name string) error {
+	_, err := k.Client.DeleteObject(ctx, &awss3.DeleteObjectInput{
+		Bucket: aws.String(k.Region),
+		Key:    aws.String(name),
+	})
+	return err
+}
+
+func (k Kodo) CopyObject(ctx context.Context, src string, dst string) (*s3.CopyObjectInfo, error) {
+	result, err := k.Client.CopyObject(ctx, &awss3.CopyObjectInput{
+		Bucket:     aws.String(k.Region),
+		CopySource: aws.String(k.Region + "/" + src),
+		Key:        aws.String(dst),
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &s3.CopyObjectInfo{
+		Key:  dst,
+		ETag: strings.ToLower(strings.ReplaceAll(aws.ToString(result.CopyObjectResult.ETag), `"`, ``)),
+	}, nil
+}
+
+func (k Kodo) StatObject(ctx context.Context, name string) (*s3.ObjectInfo, error) {
+	info, err := k.Client.HeadObject(ctx, &awss3.HeadObjectInput{
+		Bucket: aws.String(k.Region),
+		Key:    aws.String(name),
+	})
+	if err != nil {
+		return nil, err
+	}
+	res := &s3.ObjectInfo{Key: name}
+	res.Size = aws.ToInt64(info.ContentLength)
+	res.ETag = strings.ToLower(strings.ReplaceAll(aws.ToString(info.ETag), `"`, ``))
+	return res, nil
+}
+
+func (k Kodo) IsNotFound(err error) bool {
+	return true
+}
+
+func (k Kodo) AbortMultipartUpload(ctx context.Context, uploadID string, name string) error {
+	_, err := k.Client.AbortMultipartUpload(ctx, &awss3.AbortMultipartUploadInput{
+		UploadId: aws.String(uploadID),
+		Bucket:   aws.String(k.Region),
+		Key:      aws.String(name),
+	})
+	return err
+}
+
+func (k Kodo) ListUploadedParts(ctx context.Context, uploadID string, name string, partNumberMarker int, maxParts int) (*s3.ListUploadedPartsResult, error) {
+	result, err := k.Client.ListParts(ctx, &awss3.ListPartsInput{
+		Key:              aws.String(name),
+		UploadId:         aws.String(uploadID),
+		Bucket:           aws.String(k.Region),
+		MaxParts:         aws.Int32(int32(maxParts)),
+		PartNumberMarker: aws.String(strconv.Itoa(partNumberMarker)),
+	})
+	if err != nil {
+		return nil, err
+	}
+	res := &s3.ListUploadedPartsResult{
+		Key:           aws.ToString(result.Key),
+		UploadID:      aws.ToString(result.UploadId),
+		MaxParts:      int(aws.ToInt32(result.MaxParts)),
+		UploadedParts: make([]s3.UploadedPart, len(result.Parts)),
+	}
+	// int to string
+	NextPartNumberMarker, err := strconv.Atoi(aws.ToString(result.NextPartNumberMarker))
+	if err != nil {
+		return nil, err
+	}
+	res.NextPartNumberMarker = NextPartNumberMarker
+	for i, part := range result.Parts {
+		res.UploadedParts[i] = s3.UploadedPart{
+			PartNumber:   int(aws.ToInt32(part.PartNumber)),
+			LastModified: aws.ToTime(part.LastModified),
+			ETag:         aws.ToString(part.ETag),
+			Size:         aws.ToInt64(part.Size),
+		}
+	}
+	return res, nil
+}
+
+func (k Kodo) AccessURL(ctx context.Context, name string, expire time.Duration, opt *s3.AccessURLOption) (string, error) {
+	//get object head
+	info, err := k.Client.HeadObject(ctx, &awss3.HeadObjectInput{
+		Bucket: aws.String(k.Region),
+		Key:    aws.String(name),
+	})
+	if err != nil {
+		return "", errors.New("AccessURL object not found")
+	}
+	if opt != nil {
+		if opt.ContentType != aws.ToString(info.ContentType) {
+			//修改文件类型
+			err := k.SetObjectContentType(ctx, name, opt.ContentType)
+			if err != nil {
+				return "", errors.New("AccessURL setContentType error")
+			}
+		}
+	}
+	imageMogr := ""
+	//image dispose
+	if opt != nil {
+		if opt.Image != nil {
+			//https://developer.qiniu.com/dora/8255/the-zoom
+			process := ""
+			if opt.Image.Width > 0 {
+				process += strconv.Itoa(opt.Image.Width) + "x"
+			}
+			if opt.Image.Height > 0 {
+				if opt.Image.Width > 0 {
+					process += strconv.Itoa(opt.Image.Height)
+				} else {
+					process += "x" + strconv.Itoa(opt.Image.Height)
+				}
+			}
+			imageMogr = "imageMogr2/thumbnail/" + process
+		}
+	}
+	//expire
+	deadline := time.Now().Add(time.Second * expire).Unix()
+	domain := k.BucketURL
+	query := url.Values{}
+	if opt != nil && opt.Filename != "" {
+		query.Add("attname", opt.Filename)
+	}
+	privateURL := storage.MakePrivateURLv2WithQuery(k.Auth, domain, name, query, deadline)
+	if imageMogr != "" {
+		privateURL += "&" + imageMogr
+	}
+	return privateURL, nil
+}
+
+func (k *Kodo) SetObjectContentType(ctx context.Context, name string, contentType string) error {
+	//set object content-type
+	_, err := k.Client.CopyObject(ctx, &awss3.CopyObjectInput{
+		Bucket:            aws.String(k.Region),
+		CopySource:        aws.String(k.Region + "/" + name),
+		Key:               aws.String(name),
+		ContentType:       aws.String(contentType),
+		MetadataDirective: awss3types.MetadataDirectiveReplace,
+	})
+	return err
+}

pkg/common/http/http_client.go

@@ -20,7 +20,6 @@ import (
 	"encoding/json"
 	"io"
 	"net/http"
-	urllib "net/url"
 	"time"
 
 	"github.com/OpenIMSDK/protocol/constant"
@@ -107,17 +106,18 @@ func PostReturn(ctx context.Context, url string, header map[string]string, input
 }
 
 func callBackPostReturn(ctx context.Context, url, command string, input interface{}, output callbackstruct.CallbackResp, callbackConfig config.CallBackConfig) error {
-	defer log.ZDebug(ctx, "callback", "url", url, "command", command, "input", input, "callbackConfig", callbackConfig)
+	defer log.ZDebug(ctx, "callback", "url", url, "command", command, "input", input, "output", output, "callbackConfig", callbackConfig)
-
+	//
-	v := urllib.Values{}
+	//v := urllib.Values{}
-	v.Set(constant.CallbackCommand, command)
+	//v.Set(constant.CallbackCommand, command)
-	url = url + "?" + v.Encode()
+	//url = url + "/" + v.Encode()
+	url = url + "/" + command
 
 	b, err := Post(ctx, url, nil, input, callbackConfig.CallbackTimeOut)
 	if err != nil {
 		if callbackConfig.CallbackFailedContinue != nil && *callbackConfig.CallbackFailedContinue {
 			log.ZWarn(ctx, "callback failed but continue", err, "url", url)
-			return errs.ErrCallbackContinue
+			return nil
 		}
 		return errs.ErrNetwork.Wrap(err.Error())
 	}
@@ -125,7 +125,7 @@ func callBackPostReturn(ctx context.Context, url, command string, input interfac
 	if err = json.Unmarshal(b, output); err != nil {
 		if callbackConfig.CallbackFailedContinue != nil && *callbackConfig.CallbackFailedContinue {
 			log.ZWarn(ctx, "callback failed but continue", err, "url", url)
-			return errs.ErrCallbackContinue
+			return nil
 		}
 		return errs.ErrData.Wrap(err.Error())
 	}

pkg/common/startrpc/start.go

@@ -15,14 +15,21 @@
 package startrpc
 
 import (
+	"errors"
 	"fmt"
 	"log"
 	"net"
 	"net/http"
+	"os"
+	"os/signal"
 	"strconv"
+	"sync"
+	"syscall"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
@@ -56,31 +63,37 @@ func Start(
 	if err != nil {
 		return err
 	}
+
 	defer listener.Close()
 	client, err := kdisc.NewDiscoveryRegister(config.Config.Envs.Discovery)
 	if err != nil {
 		return utils.Wrap1(err)
 	}
+
 	defer client.Close()
 	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()))
 	registerIP, err := network.GetRpcRegisterIP(config.Config.Rpc.RegisterIP)
 	if err != nil {
 		return err
 	}
+
 	var reg *prometheus.Registry
 	var metric *grpcprometheus.ServerMetrics
-	// ctx 中间件
 	if config.Config.Prometheus.Enable {
-		//////////////////////////
 		cusMetrics := prommetrics.GetGrpcCusMetrics(rpcRegisterName)
-		reg, metric, err = prommetrics.NewGrpcPromObj(cusMetrics)
+		reg, metric, _ = prommetrics.NewGrpcPromObj(cusMetrics)
 		options = append(options, mw.GrpcServer(), grpc.StreamInterceptor(metric.StreamServerInterceptor()),
 			grpc.UnaryInterceptor(metric.UnaryServerInterceptor()))
 	} else {
 		options = append(options, mw.GrpcServer())
 	}
+
 	srv := grpc.NewServer(options...)
-	defer srv.GracefulStop()
+	once := sync.Once{}
+	defer func() {
+		once.Do(srv.GracefulStop)
+	}()
+
 	err = rpcFn(client, srv)
 	if err != nil {
 		return utils.Wrap1(err)
@@ -94,7 +107,10 @@ func Start(
 	if err != nil {
 		return utils.Wrap1(err)
 	}
-	go func() {
+
+	var wg errgroup.Group
+
+	wg.Go(func() error {
 		if config.Config.Prometheus.Enable && prometheusPort != 0 {
 			metric.InitializeMetrics(srv)
 			// Create a HTTP server for prometheus.
@@ -103,7 +119,34 @@ func Start(
 				log.Fatal("Unable to start a http server.")
 			}
 		}
+		return nil
+	})
+
+	wg.Go(func() error {
+		return utils.Wrap1(srv.Serve(listener))
+	})
+
+	sigs := make(chan os.Signal, 1)
+	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
+	<-sigs
+
+	var (
+		done = make(chan struct{}, 1)
+		gerr error
+	)
+
+	go func() {
+		once.Do(srv.GracefulStop)
+		gerr = wg.Wait()
+		close(done)
 	}()
 
-	return utils.Wrap1(srv.Serve(listener))
+	select {
+	case <-done:
+		return gerr
+
+	case <-time.After(15 * time.Second):
+		return utils.Wrap1(errors.New("timeout exit"))
+	}
+
 }

scripts/install/environment.sh

@@ -223,6 +223,15 @@ def "OSS_ACCESS_KEY_SECRET"                                             # 阿里
 def "OSS_SESSION_TOKEN"                                                 # 阿里云OSS的会话令牌
 def "OSS_PUBLIC_READ" "false"                                           # 公有读
 
+#七牛云配置信息
+def "KODO_ENDPOINT" "http://s3.cn-east-1.qiniucs.com"                    # 七牛云OSS的端点URL
+def "KODO_BUCKET" "demo-9999999"                                         # 七牛云OSS的存储桶名称
+def "KODO_BUCKET_URL" "http://your.domain.com"                           # 七牛云OSS的存储桶URL
+def "KODO_ACCESS_KEY_ID"                                                 # 七牛云OSS的访问密钥ID
+def "KODO_ACCESS_KEY_SECRET"                                             # 七牛云OSS的密钥
+def "KODO_SESSION_TOKEN"                                                 # 七牛云OSS的会话令牌
+def "KODO_PUBLIC_READ" "false"                                           # 公有读
+
 ###################### Redis 配置信息 ######################
 def "REDIS_PORT" "16379"                                    # Redis的端口
 def "REDIS_ADDRESS" "${DOCKER_BRIDGE_GATEWAY}"              # Redis的地址
@@ -376,7 +385,10 @@ def "FRIEND_VERIFY" "false"     # 朋友验证
 def "IOS_PUSH_SOUND" "xxx"      # IOS推送声音
 def "IOS_BADGE_COUNT" "true"    # IOS徽章计数
 def "IOS_PRODUCTION" "false"    # IOS生产
-
+# callback 配置
+def "CALLBACK_ENABLE" "false"          # 是否开启 Callback
+def "CALLBACK_TIMEOUT" "5"            # 最长超时时间
+def "CALLBACK_FAILED_CONTINUE" "true" # 失败后是否继续
 ###################### Prometheus 配置信息 ######################
 # 是否启用 Prometheus
 readonly PROMETHEUS_ENABLE=${PROMETHEUS_ENABLE:-'false'}

scripts/install/test.sh

@@ -1168,6 +1168,40 @@ EOF
   openim::test::check_error "$response"
 }
 
+# Searches for messages.
+openim::test::search_msg() {
+  local sendID="${1}"
+  local recvID="${2}"
+  local msgType="${3}"
+  local sendTime="${4}"
+  local sessionType="${5}"
+  local pageNumber="${6}"
+  local showNumber="${7}"
+
+  # Construct the request body
+  local request_body=$(cat <<EOF
+{
+  "sendID": "${sendID}",
+  "recvID": "${recvID}",
+  "msgType": ${msgType},
+  "sendTime": "${sendTime}",
+  "sessionType": ${sessionType},
+  "pagination": {
+    "pageNumber": ${pageNumber},
+    "showNumber": ${showNumber}
+  }
+}
+EOF
+)
+  echo "$request_body"
+
+  # Send the request
+  local response=$(${CCURL} "${Token}" "${OperationID}" "${Header}" "${INSECURE_OPENIMAPI}/msg/search_msg" -d "${request_body}")
+
+  # Check the response for errors
+  openim::test::check_error "$response"
+}
+
 # Revokes a message.
 openim::test::revoke_msg() {
   local userID="${1}"
@@ -1221,14 +1255,22 @@ function openim::test::msg()
   # 0. Send a message.
   openim::test::send_msg "${SEND_USER_ID}" "${RECV_USER_ID}" "${GROUP_ID}"
 
-  # Assuming message sending was successful and returned a sequence number.
+  # Wait for a short duration to ensure message is sent
-  local SEQ_NUMBER=1 # This should be the actual sequence number of the message sent.
+  # 1. Search for the message
+  local SEARCH_TIME="2023-01-01T00:00:00Z" # You may need to adjust this
+  local MSG_TYPE=101
+  local SESSION_TYPE=1
+  local PAGE_NUMBER=1
+  local SHOW_NUMBER=20
 
-  # 1. Revoke a message.
+  echo "Searching for messages between ${SEND_USER_ID} and ${RECV_USER_ID}..."
+  openim::test::search_msg "${MANAGER_USERID_1}" "${RECV_USER_ID}" "${MSG_TYPE}" "${SEARCH_TIME}" "${SESSION_TYPE}" "${PAGE_NUMBER}" "${SHOW_NUMBER}"
+  
+  # 2. Revoke a message.
   # TODO：
   # openim::test::revoke_msg "${RECV_USER_ID}" "si_${SEND_USER_ID}_${RECV_USER_ID}" "${SEQ_NUMBER}"
 
-  # 2. Clear all messages for a user.
+  # 3. Clear all messages for a user.
   openim::test::user_clear_all_msg "${RECV_USER_ID}"
 
   # Log the completion of the message test suite.