Update version to v3.8.3-patch.2

fix: Offline push does not have a badge && Android offline push (#3146)

.env

@@ -6,12 +6,18 @@ ETCD_IMAGE=quay.io/coreos/etcd:v3.5.13
 PROMETHEUS_IMAGE=prom/prometheus:v2.45.6
 ALERTMANAGER_IMAGE=prom/alertmanager:v0.27.0
 GRAFANA_IMAGE=grafana/grafana:11.0.1
+NODE_EXPORTER_IMAGE=prom/node-exporter:v1.7.0
 
-OPENIM_WEB_FRONT_IMAGE=openim/openim-web-front:release-v3.8.1
+OPENIM_WEB_FRONT_IMAGE=openim/openim-web-front:release-v3.8.3
-OPENIM_ADMIN_FRONT_IMAGE=openim/openim-admin-front:release-v1.8.3
+OPENIM_ADMIN_FRONT_IMAGE=openim/openim-admin-front:release-v1.8.4
 
 #FRONT_IMAGE: use aliyun images
-#OPENIM_WEB_FRONT_IMAGE=registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-web-front:release-v3.8.1
+#OPENIM_WEB_FRONT_IMAGE=registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-web-front:release-v3.8.3
-#OPENIM_ADMIN_FRONT_IMAGE=registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-admin-front:release-v1.8.3
+#OPENIM_ADMIN_FRONT_IMAGE=registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-admin-front:release-v1.8.4
 
 DATA_DIR=./
+
+PROMETHEUS_PORT=19091
+ALERTMANAGER_PORT=19093
+GRAFANA_PORT=13000
+NODE_EXPORTER_PORT=19100

config/openim-api.yml

@@ -10,7 +10,10 @@ api:
 prometheus:
   # Whether to enable prometheus
   enable: true
+  # autoSetPorts indicates whether to automatically set the ports
+  autoSetPorts: true
   # Prometheus listening ports, must match the number of api.ports
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 12002 ]
   # This address can be accessed via a browser
   grafanaURL: http://127.0.0.1:13000/

config/openim-msggateway.yml

@@ -1,13 +1,17 @@
 rpc:
   # The IP address where this RPC service registers itself; if left blank, it defaults to the internal network IP
   registerIP:
+  # autoSetPorts indicates whether to automatically set the ports
+  autoSetPorts: true
   # List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 10140, 10141, 10142, 10143, 10144, 10145, 10146, 10147, 10148, 10149, 10150, 10151, 10152, 10153, 10154, 10155 ]
 
 prometheus:
   # Enable or disable Prometheus monitoring
   enable: true
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 12140, 12141, 12142, 12143, 12144, 12145, 12146, 12147, 12148, 12149, 12150, 12151, 12152, 12153, 12154, 12155 ]
 
 # IP address that the RPC/WebSocket service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP

config/openim-msgtransfer.yml

@@ -1,6 +1,8 @@
 prometheus:
   # Enable or disable Prometheus monitoring
   enable: true
+  # autoSetPorts indicates whether to automatically set the ports
+  autoSetPorts: true
   # List of ports that Prometheus listens on; each port corresponds to an instance of monitoring. Ensure these are managed accordingly
-  # Because four instances have been launched, four ports need to be specified
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 12020, 12021, 12022, 12023, 12024, 12025, 12026, 12027, 12028, 12029, 12030, 12031, 12032, 12033, 12034, 12035 ]

config/openim-push.yml

@@ -3,13 +3,17 @@ rpc:
   registerIP:
   # IP address that the RPC service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
   listenIP: 0.0.0.0
+  # autoSetPorts indicates whether to automatically set the ports
+  autoSetPorts: true
   # List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 10170, 10171, 10172, 10173, 10174, 10175, 10176, 10177, 10178, 10179, 10180, 10181, 10182, 10183, 10184, 10185 ]
 
 prometheus:
   # Enable or disable Prometheus monitoring
   enable: true
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 12170, 12171, 12172, 12173, 12174, 12175, 12176, 12177, 12178, 12179, 12180, 12182, 12183, 12184, 12185, 12186 ]
 
 maxConcurrentWorkers: 3

config/openim-rpc-auth.yml

@@ -3,13 +3,17 @@ rpc:
   registerIP: 
   # IP address that the RPC service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
   listenIP: 0.0.0.0
+  # autoSetPorts indicates whether to automatically set the ports
+  autoSetPorts: true
   # List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 10200 ]
 
 prometheus:
   # Enable or disable Prometheus monitoring
   enable: true
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 12200 ]
 
 tokenPolicy:

config/openim-rpc-conversation.yml

@@ -3,11 +3,15 @@ rpc:
   registerIP: 
   # IP address that the RPC service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
   listenIP: 0.0.0.0
+  # autoSetPorts indicates whether to automatically set the ports
+  autoSetPorts: true
   # List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 10220 ]
 
 prometheus:
   # Enable or disable Prometheus monitoring
   enable: true
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 12220 ]

config/openim-rpc-friend.yml

@@ -3,11 +3,15 @@ rpc:
   registerIP: 
   # IP address that the RPC service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
   listenIP: 0.0.0.0
+  # autoSetPorts indicates whether to automatically set the ports
+  autoSetPorts: true
   # List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 10240 ]
 
 prometheus:
   # Enable or disable Prometheus monitoring
   enable: true
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 12240 ]

config/openim-rpc-group.yml

@@ -3,13 +3,17 @@ rpc:
   registerIP:
   # IP address that the RPC service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
   listenIP: 0.0.0.0
+  # autoSetPorts indicates whether to automatically set the ports
+  autoSetPorts: true
   # List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 10260 ]
 
 prometheus:
   # Enable or disable Prometheus monitoring
   enable: true
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 12260 ]
 
 

config/openim-rpc-msg.yml

@@ -3,13 +3,17 @@ rpc:
   registerIP: 
   # IP address that the RPC service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
   listenIP: 0.0.0.0
+  # autoSetPorts indicates whether to automatically set the ports
+  autoSetPorts: true
   # List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 10280 ]
 
 prometheus:
   # Enable or disable Prometheus monitoring
   enable: true
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 12280 ]
 
 

config/openim-rpc-third.yml

@@ -3,13 +3,17 @@ rpc:
   registerIP: 
   # IP address that the RPC service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
   listenIP: 0.0.0.0
+  # autoSetPorts indicates whether to automatically set the ports
+  autoSetPorts: true
   # List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 10300 ]
 
 prometheus:
   # Enable or disable Prometheus monitoring
   enable: true
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 12300 ]
 
 
@@ -38,3 +42,10 @@ object:
     accessKeySecret: 
     sessionToken: 
     publicRead: false
+  aws:
+    region: ap-southeast-2
+    bucket: testdemo832234
+    accessKeyID:
+    secretAccessKey:
+    sessionToken:
+    publicRead: false

config/openim-rpc-user.yml

@@ -3,11 +3,15 @@ rpc:
   registerIP: 
   # Listening IP; 0.0.0.0 means both internal and external IPs are listened to, if blank, the internal network IP is automatically obtained by default
   listenIP: 0.0.0.0
-  # Listening ports; if multiple are configured, multiple instances will be launched, and must be consistent with the number of prometheus.ports
+  # autoSetPorts indicates whether to automatically set the ports
+  autoSetPorts: true
+  # List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 10320 ]
 
 prometheus:
   # Whether to enable prometheus
   enable: true
   # Prometheus listening ports, must be consistent with the number of rpc.ports
+  # It will only take effect when autoSetPorts is set to false.
   ports: [ 12320 ]

config/prometheus.yml

@@ -8,7 +8,7 @@ global:
 alerting:
   alertmanagers:
     - static_configs:
-        - targets: [internal_ip:19093]
+        - targets: [127.0.0.1:19093]
 
 # Load rules once and periodically evaluate them according to the global evaluation_interval.
 rule_files:
@@ -25,62 +25,95 @@ scrape_configs:
   # prometheus fetches application services
   - job_name: node_exporter
     static_configs:
-      - targets: [ internal_ip:20500 ]
+      - targets: [ 127.0.0.1:19100 ]
+
   - job_name: openimserver-openim-api
-    static_configs:
+    http_sd_configs:
-      - targets: [ internal_ip:12002 ]
+      - url: "http://127.0.0.1:10002/prometheus_discovery/api"
-        labels:
+#    static_configs:
-          namespace: default
+#      - targets: [ 127.0.0.1:12002 ]
+#        labels:
+#          namespace: default
+
   - job_name: openimserver-openim-msggateway
-    static_configs:
+    http_sd_configs:
-      - targets: [ internal_ip:12140 ]
+      - url: "http://127.0.0.1:10002/prometheus_discovery/msg_gateway"
-#      - targets: [ internal_ip:12140, internal_ip:12141, internal_ip:12142, internal_ip:12143, internal_ip:12144, internal_ip:12145, internal_ip:12146, internal_ip:12147, internal_ip:12148, internal_ip:12149, internal_ip:12150, internal_ip:12151, internal_ip:12152, internal_ip:12153, internal_ip:12154, internal_ip:12155 ]
+#    static_configs:
-        labels:
+#      - targets: [ 127.0.0.1:12140 ]
-          namespace: default
+#        #      - targets: [ 127.0.0.1:12140, 127.0.0.1:12141, 127.0.0.1:12142, 127.0.0.1:12143, 127.0.0.1:12144, 127.0.0.1:12145, 127.0.0.1:12146, 127.0.0.1:12147, 127.0.0.1:12148, 127.0.0.1:12149, 127.0.0.1:12150, 127.0.0.1:12151, 127.0.0.1:12152, 127.0.0.1:12153, 127.0.0.1:12154, 127.0.0.1:12155 ]
+#        labels:
+#          namespace: default
+
   - job_name: openimserver-openim-msgtransfer
-    static_configs:
+    http_sd_configs:
-      - targets: [ internal_ip:12020, internal_ip:12021, internal_ip:12022, internal_ip:12023, internal_ip:12024, internal_ip:12025, internal_ip:12026, internal_ip:12027 ]
+      - url: "http://127.0.0.1:10002/prometheus_discovery/msg_transfer"
-#      - targets: [ internal_ip:12020, internal_ip:12021, internal_ip:12022, internal_ip:12023, internal_ip:12024, internal_ip:12025, internal_ip:12026, internal_ip:12027, internal_ip:12028, internal_ip:12029, internal_ip:12030, internal_ip:12031, internal_ip:12032, internal_ip:12033, internal_ip:12034, internal_ip:12035 ]
+#    static_configs:
-        labels:
+#      - targets: [ 127.0.0.1:12020, 127.0.0.1:12021, 127.0.0.1:12022, 127.0.0.1:12023, 127.0.0.1:12024, 127.0.0.1:12025, 127.0.0.1:12026, 127.0.0.1:12027 ]
-          namespace: default
+#        #      - targets: [ 127.0.0.1:12020, 127.0.0.1:12021, 127.0.0.1:12022, 127.0.0.1:12023, 127.0.0.1:12024, 127.0.0.1:12025, 127.0.0.1:12026, 127.0.0.1:12027, 127.0.0.1:12028, 127.0.0.1:12029, 127.0.0.1:12030, 127.0.0.1:12031, 127.0.0.1:12032, 127.0.0.1:12033, 127.0.0.1:12034, 127.0.0.1:12035 ]
+#        labels:
+#          namespace: default
+
   - job_name: openimserver-openim-push
-    static_configs:
+    http_sd_configs:
-      - targets: [ internal_ip:12170, internal_ip:12171, internal_ip:12172, internal_ip:12173, internal_ip:12174, internal_ip:12175, internal_ip:12176, internal_ip:12177 ]
+      - url: "http://127.0.0.1:10002/prometheus_discovery/push"
-#      - targets: [ internal_ip:12170, internal_ip:12171, internal_ip:12172, internal_ip:12173, internal_ip:12174, internal_ip:12175, internal_ip:12176, internal_ip:12177, internal_ip:12178, internal_ip:12179, internal_ip:12180,  internal_ip:12182, internal_ip:12183, internal_ip:12184, internal_ip:12185, internal_ip:12186 ]
+#    static_configs:
-        labels:
+#      - targets: [ 127.0.0.1:12170, 127.0.0.1:12171, 127.0.0.1:12172, 127.0.0.1:12173, 127.0.0.1:12174, 127.0.0.1:12175, 127.0.0.1:12176, 127.0.0.1:12177 ]
-          namespace: default
+##      - targets: [ 127.0.0.1:12170, 127.0.0.1:12171, 127.0.0.1:12172, 127.0.0.1:12173, 127.0.0.1:12174, 127.0.0.1:12175, 127.0.0.1:12176, 127.0.0.1:12177, 127.0.0.1:12178, 127.0.0.1:12179, 127.0.0.1:12180,  127.0.0.1:12182, 127.0.0.1:12183, 127.0.0.1:12184, 127.0.0.1:12185, 127.0.0.1:12186 ]
+#        labels:
+#          namespace: default
+
   - job_name: openimserver-openim-rpc-auth
-    static_configs:
+    http_sd_configs:
-      - targets: [ internal_ip:12200 ]
+      - url: "http://127.0.0.1:10002/prometheus_discovery/auth"
-        labels:
+#    static_configs:
-          namespace: default
+#      - targets: [ 127.0.0.1:12200 ]
+#        labels:
+#          namespace: default
+
   - job_name: openimserver-openim-rpc-conversation
-    static_configs:
+    http_sd_configs:
-      - targets: [ internal_ip:12220 ]
+      - url: "http://127.0.0.1:10002/prometheus_discovery/conversation"
-        labels:
+#    static_configs:
-          namespace: default
+#      - targets: [ 127.0.0.1:12220 ]
+#        labels:
+#          namespace: default
+
   - job_name: openimserver-openim-rpc-friend
-    static_configs:
+    http_sd_configs:
-      - targets: [ internal_ip:12240 ]
+      - url: "http://127.0.0.1:10002/prometheus_discovery/friend"
-        labels:
+#    static_configs:
-          namespace: default
+#      - targets: [ 127.0.0.1:12240 ]
+#        labels:
+#          namespace: default
+
   - job_name: openimserver-openim-rpc-group
-    static_configs:
+    http_sd_configs:
-      - targets: [ internal_ip:12260 ]
+      - url: "http://127.0.0.1:10002/prometheus_discovery/group"
-        labels:
+#    static_configs:
-          namespace: default
+#      - targets: [ 127.0.0.1:12260 ]
+#        labels:
+#          namespace: default.
+
   - job_name: openimserver-openim-rpc-msg
-    static_configs:
+    http_sd_configs:
-      - targets: [ internal_ip:12280 ]
+      - url: "http://127.0.0.1:10002/prometheus_discovery/msg"
-        labels:
+#    static_configs:
-          namespace: default
+#      - targets: [ 127.0.0.1:12280 ]
+#        labels:
+#          namespace: default
+
   - job_name: openimserver-openim-rpc-third
-    static_configs:
+    http_sd_configs:
-      - targets: [ internal_ip:12300 ]
+      - url: "http://127.0.0.1:10002/prometheus_discovery/third"
-        labels:
+#    static_configs:
-          namespace: default
+#      - targets: [ 127.0.0.1:12300 ]
+#        labels:
+#          namespace: default
+
   - job_name: openimserver-openim-rpc-user
-    static_configs:
+    http_sd_configs:
-      - targets: [ internal_ip:12320 ]
+      - url: "http://127.0.0.1:10002/prometheus_discovery/user"
-        labels:
+#    static_configs:
-          namespace: default
+#      - targets: [ 127.0.0.1:12320 ]
+#        labels:
+#          namespace: default

docker-compose.yml

@@ -146,49 +146,68 @@ services:
     networks:
       - openim
       
-#  prometheus:
+  prometheus:
-#    image: ${PROMETHEUS_IMAGE}
+    image: ${PROMETHEUS_IMAGE}
-#    container_name: prometheus
+    container_name: prometheus
-#    restart: always
+    restart: always
-#    user: root
+    user: root
-#    volumes:
+    profiles:
-#      - ./config/prometheus.yml:/etc/prometheus/prometheus.yml
+      - m
-#      - ./config/instance-down-rules.yml:/etc/prometheus/instance-down-rules.yml
+    volumes:
-#      - ${DATA_DIR}/components/prometheus/data:/prometheus
+      - ./config/prometheus.yml:/etc/prometheus/prometheus.yml
-#    command:
+      - ./config/instance-down-rules.yml:/etc/prometheus/instance-down-rules.yml
-#      - '--config.file=/etc/prometheus/prometheus.yml'
+      - ${DATA_DIR}/components/prometheus/data:/prometheus
-#      - '--storage.tsdb.path=/prometheus'
+    command:
-#    ports:
+      - '--config.file=/etc/prometheus/prometheus.yml'
-#      - "19091:9090"
+      - '--storage.tsdb.path=/prometheus'
-#    networks:
+      - '--web.listen-address=:${PROMETHEUS_PORT}'
-#      - openim
+    network_mode: host
-#
+
-#  alertmanager:
+  alertmanager:
-#    image: ${ALERTMANAGER_IMAGE}
+    image: ${ALERTMANAGER_IMAGE}
-#    container_name: alertmanager
+    container_name: alertmanager
-#    restart: always
+    restart: always
-#    volumes:
+    profiles:
-#      - ./config/alertmanager.yml:/etc/alertmanager/alertmanager.yml
+      - m
-#      - ./config/email.tmpl:/etc/alertmanager/email.tmpl
+    volumes:
-#    ports:
+      - ./config/alertmanager.yml:/etc/alertmanager/alertmanager.yml
-#      - "19093:9093"
+      - ./config/email.tmpl:/etc/alertmanager/email.tmpl
-#    networks:
+    command:
-#      - openim
+      - '--config.file=/etc/alertmanager/alertmanager.yml'
-#
+      - '--web.listen-address=:${ALERTMANAGER_PORT}'
-#  grafana:
+    network_mode: host
-#    image: ${GRAFANA_IMAGE}
+
-#    container_name: grafana
+  grafana:
-#    user: root
+    image: ${GRAFANA_IMAGE}
-#    restart: always
+    container_name: grafana
-#    environment:
+    user: root
-#      - GF_SECURITY_ALLOW_EMBEDDING=true
+    restart: always
-#      - GF_SESSION_COOKIE_SAMESITE=none
+    profiles:
-#      - GF_SESSION_COOKIE_SECURE=true
+      - m
-#      - GF_AUTH_ANONYMOUS_ENABLED=true
+    environment:
-#      - GF_AUTH_ANONYMOUS_ORG_ROLE=Admin
+      - GF_SECURITY_ALLOW_EMBEDDING=true
-#    ports:
+      - GF_SESSION_COOKIE_SAMESITE=none
-#      - "13000:3000"
+      - GF_SESSION_COOKIE_SECURE=true
-#    volumes:
+      - GF_AUTH_ANONYMOUS_ENABLED=true
-#      - ${DATA_DIR:-./}/components/grafana:/var/lib/grafana
+      - GF_AUTH_ANONYMOUS_ORG_ROLE=Admin
-#    networks:
+      - GF_SERVER_HTTP_PORT=${GRAFANA_PORT}
-#      - openim
+    volumes:
+      - ${DATA_DIR:-./}/components/grafana:/var/lib/grafana
+    network_mode: host
+
+  node-exporter:
+    image: ${NODE_EXPORTER_IMAGE}
+    container_name: node-exporter
+    restart: always
+    profiles:
+      - m
+    volumes:
+      - /proc:/host/proc:ro
+      - /sys:/host/sys:ro
+      - /:/rootfs:ro
+    command:
+      - '--path.procfs=/host/proc'
+      - '--path.sysfs=/host/sys'
+      - '--path.rootfs=/rootfs'
+      - '--web.listen-address=:${NODE_EXPORTER_PORT}'
+    network_mode: host

go.mod

@@ -12,8 +12,8 @@ require (
 	github.com/gorilla/websocket v1.5.1
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
 	github.com/mitchellh/mapstructure v1.5.0
-	github.com/openimsdk/protocol v0.0.72-alpha.69
+	github.com/openimsdk/protocol v0.0.72-alpha.71
-	github.com/openimsdk/tools v0.0.50-alpha.62
+	github.com/openimsdk/tools v0.0.50-alpha.67
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0
 	github.com/stretchr/testify v1.9.0
@@ -41,6 +41,7 @@ require (
 	github.com/shirou/gopsutil v3.21.11+incompatible
 	github.com/spf13/viper v1.18.2
 	github.com/stathat/consistent v1.0.0
+	go.etcd.io/etcd/client/v3 v3.5.13
 	go.uber.org/automaxprocs v1.5.3
 	golang.org/x/sync v0.8.0
 )
@@ -162,7 +163,6 @@ require (
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.13 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.13 // indirect
-	go.etcd.io/etcd/client/v3 v3.5.13 // indirect
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.49.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0 // indirect

go.sum

@@ -317,12 +317,12 @@ github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
 github.com/onsi/ginkgo v1.16.5/go.mod h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU=
 github.com/onsi/gomega v1.25.0 h1:Vw7br2PCDYijJHSfBOWhov+8cAnUf8MfMaIOV323l6Y=
 github.com/onsi/gomega v1.25.0/go.mod h1:r+zV744Re+DiYCIPRlYOTxn0YkOLcAnW8k1xXdMPGhM=
-github.com/openimsdk/gomake v0.0.15-alpha.2 h1:5Q8yl8ezy2yx+q8/ucU/t4kJnDfCzNOrkXcDACCqtyM=
+github.com/openimsdk/gomake v0.0.14-alpha.5 h1:VY9c5x515lTfmdhhPjMvR3BBRrRquAUCFsz7t7vbv7Y=
-github.com/openimsdk/gomake v0.0.15-alpha.2/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
+github.com/openimsdk/gomake v0.0.14-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
-github.com/openimsdk/protocol v0.0.72-alpha.69 h1:b22oY2XTdBR/BePqA73KsrM3GDF3Vk8YcBEXZU4ArJc=
+github.com/openimsdk/protocol v0.0.72-alpha.71 h1:R3utzOlqepaJWTAmnfJi4ccUM/XIoFasSyjQMOipM70=
-github.com/openimsdk/protocol v0.0.72-alpha.69/go.mod h1:Iet+piS/jaS+kWWyj6EEr36mk4ISzIRYjoMSVA4dq2M=
+github.com/openimsdk/protocol v0.0.72-alpha.71/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw=
-github.com/openimsdk/tools v0.0.50-alpha.62 h1:e/m1XL7+EXbkOoxr/En/612WcOPKOUHPBj0++gG6MuQ=
+github.com/openimsdk/tools v0.0.50-alpha.67 h1:K7kguqvPbjldHAi7pGhcG2ERkctCqG9ZFlteT7UKaxM=
-github.com/openimsdk/tools v0.0.50-alpha.62/go.mod h1:JowL2jYr8tu4vcQe+5hJh4v3BtSx1T0CIS3pgU/Mw+U=
+github.com/openimsdk/tools v0.0.50-alpha.67/go.mod h1:B+oqV0zdewN7OiEHYJm+hW+8/Te7B8tHHgD8rK5ZLZk=
 github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=
 github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs=
 github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ=

internal/api/config_manager.go

@@ -0,0 +1,313 @@
+package api
+
+//
+//import (
+//	"encoding/json"
+//	"reflect"
+//	"strconv"
+//	"time"
+//
+//	"github.com/gin-gonic/gin"
+//	"github.com/openimsdk/open-im-server/v3/pkg/apistruct"
+//	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
+//	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+//	"github.com/openimsdk/open-im-server/v3/pkg/common/discovery/etcd"
+//	"github.com/openimsdk/open-im-server/v3/version"
+//	"github.com/openimsdk/tools/apiresp"
+//	"github.com/openimsdk/tools/errs"
+//	"github.com/openimsdk/tools/log"
+//	"github.com/openimsdk/tools/utils/runtimeenv"
+//	clientv3 "go.etcd.io/etcd/client/v3"
+//)
+//
+//const (
+//	// wait for Restart http call return
+//	waitHttp = time.Millisecond * 200
+//)
+//
+//type ConfigManager struct {
+//	imAdminUserID []string
+//	config        *config.AllConfig
+//	client        *clientv3.Client
+//
+//	configPath string
+//	runtimeEnv string
+//}
+//
+//func NewConfigManager(IMAdminUserID []string, cfg *config.AllConfig, client *clientv3.Client, configPath string, runtimeEnv string) *ConfigManager {
+//	cm := &ConfigManager{
+//		imAdminUserID: IMAdminUserID,
+//		config:        cfg,
+//		client:        client,
+//		configPath:    configPath,
+//		runtimeEnv:    runtimeEnv,
+//	}
+//	return cm
+//}
+//
+//func (cm *ConfigManager) CheckAdmin(c *gin.Context) {
+//	if err := authverify.CheckAdmin(c, cm.imAdminUserID); err != nil {
+//		apiresp.GinError(c, err)
+//		c.Abort()
+//	}
+//}
+//
+//func (cm *ConfigManager) GetConfig(c *gin.Context) {
+//	var req apistruct.GetConfigReq
+//	if err := c.BindJSON(&req); err != nil {
+//		apiresp.GinError(c, errs.ErrArgs.WithDetail(err.Error()).Wrap())
+//		return
+//	}
+//	conf := cm.config.Name2Config(req.ConfigName)
+//	if conf == nil {
+//		apiresp.GinError(c, errs.ErrArgs.WithDetail("config name not found").Wrap())
+//		return
+//	}
+//	b, err := json.Marshal(conf)
+//	if err != nil {
+//		apiresp.GinError(c, err)
+//		return
+//	}
+//	apiresp.GinSuccess(c, string(b))
+//}
+//
+//func (cm *ConfigManager) GetConfigList(c *gin.Context) {
+//	var resp apistruct.GetConfigListResp
+//	resp.ConfigNames = cm.config.GetConfigNames()
+//	resp.Environment = runtimeenv.PrintRuntimeEnvironment()
+//	resp.Version = version.Version
+//
+//	apiresp.GinSuccess(c, resp)
+//}
+//
+//func (cm *ConfigManager) SetConfig(c *gin.Context) {
+//	if cm.config.Discovery.Enable != config.ETCD {
+//		apiresp.GinError(c, errs.New("only etcd support set config").Wrap())
+//		return
+//	}
+//	var req apistruct.SetConfigReq
+//	if err := c.BindJSON(&req); err != nil {
+//		apiresp.GinError(c, errs.ErrArgs.WithDetail(err.Error()).Wrap())
+//		return
+//	}
+//	var err error
+//	switch req.ConfigName {
+//	case cm.config.Discovery.GetConfigFileName():
+//		err = compareAndSave[config.Discovery](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.Kafka.GetConfigFileName():
+//		err = compareAndSave[config.Kafka](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.LocalCache.GetConfigFileName():
+//		err = compareAndSave[config.LocalCache](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.Log.GetConfigFileName():
+//		err = compareAndSave[config.Log](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.Minio.GetConfigFileName():
+//		err = compareAndSave[config.Minio](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.Mongo.GetConfigFileName():
+//		err = compareAndSave[config.Mongo](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.Notification.GetConfigFileName():
+//		err = compareAndSave[config.Notification](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.API.GetConfigFileName():
+//		err = compareAndSave[config.API](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.CronTask.GetConfigFileName():
+//		err = compareAndSave[config.CronTask](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.MsgGateway.GetConfigFileName():
+//		err = compareAndSave[config.MsgGateway](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.MsgTransfer.GetConfigFileName():
+//		err = compareAndSave[config.MsgTransfer](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.Push.GetConfigFileName():
+//		err = compareAndSave[config.Push](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.Auth.GetConfigFileName():
+//		err = compareAndSave[config.Auth](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.Conversation.GetConfigFileName():
+//		err = compareAndSave[config.Conversation](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.Friend.GetConfigFileName():
+//		err = compareAndSave[config.Friend](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.Group.GetConfigFileName():
+//		err = compareAndSave[config.Group](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.Msg.GetConfigFileName():
+//		err = compareAndSave[config.Msg](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.Third.GetConfigFileName():
+//		err = compareAndSave[config.Third](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.User.GetConfigFileName():
+//		err = compareAndSave[config.User](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.Redis.GetConfigFileName():
+//		err = compareAndSave[config.Redis](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.Share.GetConfigFileName():
+//		err = compareAndSave[config.Share](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	case cm.config.Webhooks.GetConfigFileName():
+//		err = compareAndSave[config.Webhooks](c, cm.config.Name2Config(req.ConfigName), &req, cm)
+//	default:
+//		apiresp.GinError(c, errs.ErrArgs.Wrap())
+//		return
+//	}
+//	if err != nil {
+//		apiresp.GinError(c, errs.ErrArgs.WithDetail(err.Error()).Wrap())
+//		return
+//	}
+//	apiresp.GinSuccess(c, nil)
+//}
+//
+//func compareAndSave[T any](c *gin.Context, old any, req *apistruct.SetConfigReq, cm *ConfigManager) error {
+//	conf := new(T)
+//	err := json.Unmarshal([]byte(req.Data), &conf)
+//	if err != nil {
+//		return errs.ErrArgs.WithDetail(err.Error()).Wrap()
+//	}
+//	eq := reflect.DeepEqual(old, conf)
+//	if eq {
+//		return nil
+//	}
+//	data, err := json.Marshal(conf)
+//	if err != nil {
+//		return errs.ErrArgs.WithDetail(err.Error()).Wrap()
+//	}
+//	_, err = cm.client.Put(c, etcd.BuildKey(req.ConfigName), string(data))
+//	if err != nil {
+//		return errs.WrapMsg(err, "save to etcd failed")
+//	}
+//	return nil
+//}
+//
+//func (cm *ConfigManager) ResetConfig(c *gin.Context) {
+//	go func() {
+//		if err := cm.resetConfig(c, true); err != nil {
+//			log.ZError(c, "reset config err", err)
+//		}
+//	}()
+//	apiresp.GinSuccess(c, nil)
+//}
+//
+//func (cm *ConfigManager) resetConfig(c *gin.Context, checkChange bool, ops ...clientv3.Op) error {
+//	txn := cm.client.Txn(c)
+//	type initConf struct {
+//		old any
+//		new any
+//	}
+//	configMap := map[string]*initConf{
+//		cm.config.Discovery.GetConfigFileName():    {old: &cm.config.Discovery, new: new(config.Discovery)},
+//		cm.config.Kafka.GetConfigFileName():        {old: &cm.config.Kafka, new: new(config.Kafka)},
+//		cm.config.LocalCache.GetConfigFileName():   {old: &cm.config.LocalCache, new: new(config.LocalCache)},
+//		cm.config.Log.GetConfigFileName():          {old: &cm.config.Log, new: new(config.Log)},
+//		cm.config.Minio.GetConfigFileName():        {old: &cm.config.Minio, new: new(config.Minio)},
+//		cm.config.Mongo.GetConfigFileName():        {old: &cm.config.Mongo, new: new(config.Mongo)},
+//		cm.config.Notification.GetConfigFileName(): {old: &cm.config.Notification, new: new(config.Notification)},
+//		cm.config.API.GetConfigFileName():          {old: &cm.config.API, new: new(config.API)},
+//		cm.config.CronTask.GetConfigFileName():     {old: &cm.config.CronTask, new: new(config.CronTask)},
+//		cm.config.MsgGateway.GetConfigFileName():   {old: &cm.config.MsgGateway, new: new(config.MsgGateway)},
+//		cm.config.MsgTransfer.GetConfigFileName():  {old: &cm.config.MsgTransfer, new: new(config.MsgTransfer)},
+//		cm.config.Push.GetConfigFileName():         {old: &cm.config.Push, new: new(config.Push)},
+//		cm.config.Auth.GetConfigFileName():         {old: &cm.config.Auth, new: new(config.Auth)},
+//		cm.config.Conversation.GetConfigFileName(): {old: &cm.config.Conversation, new: new(config.Conversation)},
+//		cm.config.Friend.GetConfigFileName():       {old: &cm.config.Friend, new: new(config.Friend)},
+//		cm.config.Group.GetConfigFileName():        {old: &cm.config.Group, new: new(config.Group)},
+//		cm.config.Msg.GetConfigFileName():          {old: &cm.config.Msg, new: new(config.Msg)},
+//		cm.config.Third.GetConfigFileName():        {old: &cm.config.Third, new: new(config.Third)},
+//		cm.config.User.GetConfigFileName():         {old: &cm.config.User, new: new(config.User)},
+//		cm.config.Redis.GetConfigFileName():        {old: &cm.config.Redis, new: new(config.Redis)},
+//		cm.config.Share.GetConfigFileName():        {old: &cm.config.Share, new: new(config.Share)},
+//		cm.config.Webhooks.GetConfigFileName():     {old: &cm.config.Webhooks, new: new(config.Webhooks)},
+//	}
+//
+//	changedKeys := make([]string, 0, len(configMap))
+//	for k, v := range configMap {
+//		err := config.Load(
+//			cm.configPath,
+//			k,
+//			config.EnvPrefixMap[k],
+//			cm.runtimeEnv,
+//			v.new,
+//		)
+//		if err != nil {
+//			log.ZError(c, "load config failed", err)
+//			continue
+//		}
+//		equal := reflect.DeepEqual(v.old, v.new)
+//		if !checkChange || !equal {
+//			changedKeys = append(changedKeys, k)
+//		}
+//	}
+//
+//	for _, k := range changedKeys {
+//		data, err := json.Marshal(configMap[k].new)
+//		if err != nil {
+//			log.ZError(c, "marshal config failed", err)
+//			continue
+//		}
+//		ops = append(ops, clientv3.OpPut(etcd.BuildKey(k), string(data)))
+//	}
+//	if len(ops) > 0 {
+//		txn.Then(ops...)
+//		_, err := txn.Commit()
+//		if err != nil {
+//			return errs.WrapMsg(err, "commit etcd txn failed")
+//		}
+//	}
+//	return nil
+//}
+//
+//func (cm *ConfigManager) Restart(c *gin.Context) {
+//	go cm.restart(c)
+//	apiresp.GinSuccess(c, nil)
+//}
+//
+//func (cm *ConfigManager) restart(c *gin.Context) {
+//	time.Sleep(waitHttp) // wait for Restart http call return
+//	t := time.Now().Unix()
+//	_, err := cm.client.Put(c, etcd.BuildKey(etcd.RestartKey), strconv.Itoa(int(t)))
+//	if err != nil {
+//		log.ZError(c, "restart etcd put key failed", err)
+//	}
+//}
+//
+//func (cm *ConfigManager) SetEnableConfigManager(c *gin.Context) {
+//	if cm.config.Discovery.Enable != config.ETCD {
+//		apiresp.GinError(c, errs.New("only etcd support config manager").Wrap())
+//		return
+//	}
+//	var req apistruct.SetEnableConfigManagerReq
+//	if err := c.BindJSON(&req); err != nil {
+//		apiresp.GinError(c, errs.ErrArgs.WithDetail(err.Error()).Wrap())
+//		return
+//	}
+//	var enableStr string
+//	if req.Enable {
+//		enableStr = etcd.Enable
+//	} else {
+//		enableStr = etcd.Disable
+//	}
+//	resp, err := cm.client.Get(c, etcd.BuildKey(etcd.EnableConfigCenterKey))
+//	if err != nil {
+//		apiresp.GinError(c, errs.WrapMsg(err, "getEnableConfigManager failed"))
+//		return
+//	}
+//	if !(resp.Count > 0 && string(resp.Kvs[0].Value) == etcd.Enable) && req.Enable {
+//		go func() {
+//			time.Sleep(waitHttp) // wait for Restart http call return
+//			err := cm.resetConfig(c, false, clientv3.OpPut(etcd.BuildKey(etcd.EnableConfigCenterKey), enableStr))
+//			if err != nil {
+//				log.ZError(c, "resetConfig failed", err)
+//			}
+//		}()
+//	} else {
+//		_, err = cm.client.Put(c, etcd.BuildKey(etcd.EnableConfigCenterKey), enableStr)
+//		if err != nil {
+//			apiresp.GinError(c, errs.WrapMsg(err, "setEnableConfigManager failed"))
+//			return
+//		}
+//	}
+//
+//	apiresp.GinSuccess(c, nil)
+//}
+//
+//func (cm *ConfigManager) GetEnableConfigManager(c *gin.Context) {
+//	resp, err := cm.client.Get(c, etcd.BuildKey(etcd.EnableConfigCenterKey))
+//	if err != nil {
+//		apiresp.GinError(c, errs.WrapMsg(err, "getEnableConfigManager failed"))
+//		return
+//	}
+//	var enable bool
+//	if resp.Count > 0 && string(resp.Kvs[0].Value) == etcd.Enable {
+//		enable = true
+//	}
+//	apiresp.GinSuccess(c, &apistruct.GetEnableConfigManagerResp{Enable: enable})
+//}

internal/api/init.go

@@ -1,25 +1,9 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
 package api
 
 import (
 	"context"
+	"errors"
 	"fmt"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
-	"github.com/openimsdk/tools/utils/datautil"
-	"github.com/openimsdk/tools/utils/network"
 	"net"
 	"net/http"
 	"os"
@@ -28,12 +12,21 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/tools/mw"
+	"github.com/openimsdk/tools/utils/datautil"
+	"github.com/openimsdk/tools/utils/network"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
 	kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/tools/discovery"
+	"github.com/openimsdk/tools/discovery/etcd"
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/system/program"
+	"github.com/openimsdk/tools/utils/jsonutil"
 )
 
 type Config struct {
@@ -42,8 +35,8 @@ type Config struct {
 	Discovery config.Discovery
 }
 
-func Start(ctx context.Context, index int, config *Config) error {
+func Start(ctx context.Context, index int, cfg *Config) error {
-	apiPort, err := datautil.GetElemByIndex(config.API.Api.Ports, index)
+	apiPort, err := datautil.GetElemByIndex(cfg.API.Api.Ports, index)
 	if err != nil {
 		return err
 	}
@@ -51,10 +44,14 @@ func Start(ctx context.Context, index int, config *Config) error {
 	var client discovery.SvcDiscoveryRegistry
 
 	// Determine whether zk is passed according to whether it is a clustered deployment
-	client, err = kdisc.NewDiscoveryRegister(&config.Discovery, &config.Share)
+	client, err = kdisc.NewDiscoveryRegister(&cfg.Discovery, &cfg.Share, []string{
+		cfg.Share.RpcRegisterName.MessageGateway,
+	})
 	if err != nil {
 		return errs.WrapMsg(err, "failed to register discovery service")
 	}
+	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
 
 	var (
 		netDone        = make(chan struct{}, 1)
@@ -62,32 +59,73 @@ func Start(ctx context.Context, index int, config *Config) error {
 		prometheusPort int
 	)
 
-	router, err := newGinRouter(ctx, client, config)
+	router, err := newGinRouter(ctx, client, cfg)
 	if err != nil {
 		return err
 	}
-	if config.API.Prometheus.Enable {
+	registerIP, err := network.GetRpcRegisterIP("")
-		go func() {
-			prometheusPort, err = datautil.GetElemByIndex(config.API.Prometheus.Ports, index)
 	if err != nil {
-				netErr = err
+		return err
-				netDone <- struct{}{}
-				return
 	}
-			if err := prommetrics.ApiInit(prometheusPort); err != nil && err != http.ErrServerClosed {
+
+	getAutoPort := func() (net.Listener, int, error) {
+		registerAddr := net.JoinHostPort(registerIP, "0")
+		listener, err := net.Listen("tcp", registerAddr)
+		if err != nil {
+			return nil, 0, errs.WrapMsg(err, "listen err", "registerAddr", registerAddr)
+		}
+		_, portStr, _ := net.SplitHostPort(listener.Addr().String())
+		port, _ := strconv.Atoi(portStr)
+		return listener, port, nil
+	}
+
+	if cfg.API.Prometheus.AutoSetPorts && cfg.Discovery.Enable != config.ETCD {
+		return errs.New("only etcd support autoSetPorts", "RegisterName", "api").Wrap()
+	}
+
+	if cfg.API.Prometheus.Enable {
+		var (
+			listener net.Listener
+		)
+
+		if cfg.API.Prometheus.AutoSetPorts {
+			listener, prometheusPort, err = getAutoPort()
+			if err != nil {
+				return err
+			}
+
+			etcdClient := client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
+
+			_, err = etcdClient.Put(ctx, prommetrics.BuildDiscoveryKey(prommetrics.APIKeyName), jsonutil.StructToJsonString(prommetrics.BuildDefaultTarget(registerIP, prometheusPort)))
+			if err != nil {
+				return errs.WrapMsg(err, "etcd put err")
+			}
+		} else {
+			prometheusPort, err = datautil.GetElemByIndex(cfg.API.Prometheus.Ports, index)
+			if err != nil {
+				return err
+			}
+			listener, err = net.Listen("tcp", fmt.Sprintf(":%d", prometheusPort))
+			if err != nil {
+				return errs.WrapMsg(err, "listen err", "addr", fmt.Sprintf(":%d", prometheusPort))
+			}
+		}
+
+		go func() {
+			if err := prommetrics.ApiInit(listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
 				netErr = errs.WrapMsg(err, fmt.Sprintf("api prometheus start err: %d", prometheusPort))
 				netDone <- struct{}{}
 			}
 		}()
 
 	}
-	address := net.JoinHostPort(network.GetListenIP(config.API.Api.ListenIP), strconv.Itoa(apiPort))
+	address := net.JoinHostPort(network.GetListenIP(cfg.API.Api.ListenIP), strconv.Itoa(apiPort))
 
 	server := http.Server{Addr: address, Handler: router}
 	log.CInfo(ctx, "API server is initializing", "address", address, "apiPort", apiPort, "prometheusPort", prometheusPort)
 	go func() {
 		err = server.ListenAndServe()
-		if err != nil && err != http.ErrServerClosed {
+		if err != nil && !errors.Is(err, http.ErrServerClosed) {
 			netErr = errs.WrapMsg(err, fmt.Sprintf("api start err: %s", server.Addr))
 			netDone <- struct{}{}
 

internal/api/prometheus_discovery.go

@@ -0,0 +1,114 @@
+package api
+
+import (
+	"encoding/json"
+	"net/http"
+
+	"github.com/gin-gonic/gin"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
+	"github.com/openimsdk/tools/apiresp"
+	"github.com/openimsdk/tools/discovery"
+	"github.com/openimsdk/tools/discovery/etcd"
+	"github.com/openimsdk/tools/errs"
+	"github.com/openimsdk/tools/log"
+	clientv3 "go.etcd.io/etcd/client/v3"
+)
+
+type PrometheusDiscoveryApi struct {
+	config *Config
+	client *clientv3.Client
+}
+
+func NewPrometheusDiscoveryApi(cfg *Config, client discovery.SvcDiscoveryRegistry) *PrometheusDiscoveryApi {
+	api := &PrometheusDiscoveryApi{
+		config: cfg,
+	}
+	if cfg.Discovery.Enable == config.ETCD {
+		api.client = client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
+	}
+	return api
+}
+
+func (p *PrometheusDiscoveryApi) Enable(c *gin.Context) {
+	if p.config.Discovery.Enable != config.ETCD {
+		c.JSON(http.StatusOK, []struct{}{})
+		c.Abort()
+	}
+}
+
+func (p *PrometheusDiscoveryApi) discovery(c *gin.Context, key string) {
+	eResp, err := p.client.Get(c, prommetrics.BuildDiscoveryKey(key))
+	if err != nil {
+		// Log and respond with an error if preparation fails.
+		apiresp.GinError(c, errs.WrapMsg(err, "etcd get err"))
+		return
+	}
+	if len(eResp.Kvs) == 0 {
+		c.JSON(http.StatusOK, []*prommetrics.Target{})
+	}
+
+	var (
+		resp = &prommetrics.RespTarget{
+			Targets: make([]string, 0, len(eResp.Kvs)),
+		}
+	)
+
+	for i := range eResp.Kvs {
+		var target prommetrics.Target
+		err = json.Unmarshal(eResp.Kvs[i].Value, &target)
+		if err != nil {
+			log.ZError(c, "prometheus unmarshal err", errs.Wrap(err))
+		}
+		resp.Targets = append(resp.Targets, target.Target)
+		if resp.Labels == nil {
+			resp.Labels = target.Labels
+		}
+	}
+
+	c.JSON(200, []*prommetrics.RespTarget{resp})
+}
+
+func (p *PrometheusDiscoveryApi) Api(c *gin.Context) {
+	p.discovery(c, prommetrics.APIKeyName)
+}
+
+func (p *PrometheusDiscoveryApi) User(c *gin.Context) {
+	p.discovery(c, p.config.Share.RpcRegisterName.User)
+}
+
+func (p *PrometheusDiscoveryApi) Group(c *gin.Context) {
+	p.discovery(c, p.config.Share.RpcRegisterName.Group)
+}
+
+func (p *PrometheusDiscoveryApi) Msg(c *gin.Context) {
+	p.discovery(c, p.config.Share.RpcRegisterName.Msg)
+}
+
+func (p *PrometheusDiscoveryApi) Friend(c *gin.Context) {
+	p.discovery(c, p.config.Share.RpcRegisterName.Friend)
+}
+
+func (p *PrometheusDiscoveryApi) Conversation(c *gin.Context) {
+	p.discovery(c, p.config.Share.RpcRegisterName.Conversation)
+}
+
+func (p *PrometheusDiscoveryApi) Third(c *gin.Context) {
+	p.discovery(c, p.config.Share.RpcRegisterName.Third)
+}
+
+func (p *PrometheusDiscoveryApi) Auth(c *gin.Context) {
+	p.discovery(c, p.config.Share.RpcRegisterName.Auth)
+}
+
+func (p *PrometheusDiscoveryApi) Push(c *gin.Context) {
+	p.discovery(c, p.config.Share.RpcRegisterName.Push)
+}
+
+func (p *PrometheusDiscoveryApi) MessageGateway(c *gin.Context) {
+	p.discovery(c, p.config.Share.RpcRegisterName.MessageGateway)
+}
+
+func (p *PrometheusDiscoveryApi) MessageTransfer(c *gin.Context) {
+	p.discovery(c, prommetrics.MessageTransferKeyName)
+}

internal/api/router.go

@@ -2,7 +2,9 @@ package api
 
 import (
 	"context"
-	"fmt"
+	"net/http"
+	"strings"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
 	pbAuth "github.com/openimsdk/protocol/auth"
 	"github.com/openimsdk/protocol/conversation"
@@ -11,8 +13,6 @@ import (
 	"github.com/openimsdk/protocol/relation"
 	"github.com/openimsdk/protocol/third"
 	"github.com/openimsdk/protocol/user"
-	"net/http"
-	"strings"
 
 	"github.com/openimsdk/open-im-server/v3/internal/api/jssdk"
 
@@ -21,9 +21,6 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/gin-gonic/gin/binding"
 	"github.com/go-playground/validator/v10"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
 	"github.com/openimsdk/protocol/constant"
@@ -56,8 +53,6 @@ func prommetricsGin() gin.HandlerFunc {
 }
 
 func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, config *Config) (*gin.Engine, error) {
-	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
 	authConn, err := client.GetConn(ctx, config.Share.RpcRegisterName.Auth)
 	if err != nil {
 		return nil, err
@@ -100,12 +95,11 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, co
 	case BestSpeed:
 		r.Use(gzip.Gzip(gzip.BestSpeed))
 	}
-	r.Use(prommetricsGin(), gin.RecoveryWithWriter(gin.DefaultErrorWriter, mw.GinPanicErr), mw.CorsHandler(),
+	r.Use(prommetricsGin(), gin.RecoveryWithWriter(gin.DefaultErrorWriter, mw.GinPanicErr), mw.CorsHandler(), mw.GinParseOperationID(), GinParseToken(rpcli.NewAuthClient(authConn)))
-		mw.GinParseOperationID(), GinParseToken(rpcli.NewAuthClient(authConn)))
-
 	u := NewUserApi(user.NewUserClient(userConn), client, config.Share.RpcRegisterName)
-	{
+	m := NewMessageApi(msg.NewMsgClient(msgConn), rpcli.NewUserClient(userConn), config.Share.IMAdminUserID)
 	userRouterGroup := r.Group("/user")
+	{
 		userRouterGroup.POST("/user_register", u.UserRegister)
 		userRouterGroup.POST("/update_user_info", u.UpdateUserInfo)
 		userRouterGroup.POST("/update_user_info_ex", u.UpdateUserInfoEx)
@@ -228,7 +222,6 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, co
 		objectGroup.GET("/*name", t.ObjectRedirect)
 	}
 	// Message
-	m := NewMessageApi(msg.NewMsgClient(msgConn), rpcli.NewUserClient(userConn), config.Share.IMAdminUserID)
 	{
 		msgGroup := r.Group("/msg")
 		msgGroup.POST("/newest_seq", m.GetSeq)
@@ -284,7 +277,21 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, co
 		jssdk.POST("/get_conversations", j.GetConversations)
 		jssdk.POST("/get_active_conversations", j.GetActiveConversations)
 	}
-
+	{
+		pd := NewPrometheusDiscoveryApi(config, client)
+		proDiscoveryGroup := r.Group("/prometheus_discovery", pd.Enable)
+		proDiscoveryGroup.GET("/api", pd.Api)
+		proDiscoveryGroup.GET("/user", pd.User)
+		proDiscoveryGroup.GET("/group", pd.Group)
+		proDiscoveryGroup.GET("/msg", pd.Msg)
+		proDiscoveryGroup.GET("/friend", pd.Friend)
+		proDiscoveryGroup.GET("/conversation", pd.Conversation)
+		proDiscoveryGroup.GET("/third", pd.Third)
+		proDiscoveryGroup.GET("/auth", pd.Auth)
+		proDiscoveryGroup.GET("/push", pd.Push)
+		proDiscoveryGroup.GET("/msg_gateway", pd.MessageGateway)
+		proDiscoveryGroup.GET("/msg_transfer", pd.MessageTransfer)
+	}
 	return r, nil
 }
 

internal/msggateway/client.go

@@ -236,6 +236,8 @@ func (c *Client) handleMessage(message []byte) error {
 		resp, messageErr = c.longConnServer.GetSeqMessage(ctx, binaryReq)
 	case WSGetConvMaxReadSeq:
 		resp, messageErr = c.longConnServer.GetConversationsHasReadAndMaxSeq(ctx, binaryReq)
+	case WsPullConvLastMessage:
+		resp, messageErr = c.longConnServer.GetLastMessage(ctx, binaryReq)
 	case WsLogoutMsg:
 		resp, messageErr = c.longConnServer.UserLogout(ctx, binaryReq)
 	case WsSetBackgroundStatus:

internal/msggateway/constant.go

@@ -47,6 +47,7 @@ const (
 	WSSendSignalMsg       = 1004
 	WSPullMsg             = 1005
 	WSGetConvMaxReadSeq   = 1006
+	WsPullConvLastMessage = 1007
 	WSPushMsg             = 2001
 	WSKickOnlineMsg       = 2002
 	WsLogoutMsg           = 2003

internal/msggateway/hub_server.go

@@ -16,9 +16,10 @@ package msggateway
 
 import (
 	"context"
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
 	"sync/atomic"
 
+	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/startrpc"
@@ -53,10 +54,14 @@ func (s *Server) InitServer(ctx context.Context, config *Config, disCov discover
 func (s *Server) Start(ctx context.Context, index int, conf *Config) error {
 	return startrpc.Start(ctx, &conf.Discovery, &conf.MsgGateway.Prometheus, conf.MsgGateway.ListenIP,
 		conf.MsgGateway.RPC.RegisterIP,
+		conf.MsgGateway.RPC.AutoSetPorts,
 		conf.MsgGateway.RPC.Ports, index,
 		conf.Share.RpcRegisterName.MessageGateway,
 		&conf.Share,
 		conf,
+		[]string{
+			conf.Share.RpcRegisterName.MessageGateway,
+		},
 		s.InitServer,
 	)
 }
@@ -89,10 +94,6 @@ func NewServer(longConnServer LongConnServer, conf *Config, ready func(srv *Serv
 	return s
 }
 
-func (s *Server) OnlinePushMsg(context context.Context, req *msggateway.OnlinePushMsgReq) (*msggateway.OnlinePushMsgResp, error) {
-	panic("implement me")
-}
-
 func (s *Server) GetUsersOnlineStatus(ctx context.Context, req *msggateway.GetUsersOnlineStatusReq) (*msggateway.GetUsersOnlineStatusResp, error) {
 	if !authverify.IsAppManagerUid(ctx, s.config.Share.IMAdminUserID) {
 		return nil, errs.ErrNoPermission.WrapMsg("only app manager")
@@ -126,11 +127,6 @@ func (s *Server) GetUsersOnlineStatus(ctx context.Context, req *msggateway.GetUs
 	return &resp, nil
 }
 
-func (s *Server) OnlineBatchPushOneMsg(ctx context.Context, req *msggateway.OnlineBatchPushOneMsgReq) (*msggateway.OnlineBatchPushOneMsgResp, error) {
-	// todo implement
-	return nil, nil
-}
-
 func (s *Server) pushToUser(ctx context.Context, userID string, msgData *sdkws.MsgData) *msggateway.SingleMsgToUserResults {
 	clients, ok := s.LongConnServer.GetUserAllCons(userID)
 	if !ok {

internal/msggateway/init.go

@@ -16,13 +16,13 @@ package msggateway
 
 import (
 	"context"
+	"time"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpccache"
 	"github.com/openimsdk/tools/db/redisutil"
-	"github.com/openimsdk/tools/utils/datautil"
-	"time"
-
 	"github.com/openimsdk/tools/log"
+	"github.com/openimsdk/tools/utils/datautil"
 )
 
 type Config struct {
@@ -35,7 +35,8 @@ type Config struct {
 
 // Start run ws server.
 func Start(ctx context.Context, index int, conf *Config) error {
-	log.CInfo(ctx, "MSG-GATEWAY server is initializing", "rpcPorts", conf.MsgGateway.RPC.Ports,
+	log.CInfo(ctx, "MSG-GATEWAY server is initializing", "autoSetPorts", conf.MsgGateway.RPC.AutoSetPorts,
+		"rpcPorts", conf.MsgGateway.RPC.Ports,
 		"wsPort", conf.MsgGateway.LongConnSvr.Ports, "prometheusPorts", conf.MsgGateway.Prometheus.Ports)
 	wsPort, err := datautil.GetElemByIndex(conf.MsgGateway.LongConnSvr.Ports, index)
 	if err != nil {

internal/msggateway/message_handler.go

@@ -108,6 +108,7 @@ type MessageHandler interface {
 	GetSeqMessage(ctx context.Context, data *Req) ([]byte, error)
 	UserLogout(ctx context.Context, data *Req) ([]byte, error)
 	SetUserDeviceBackground(ctx context.Context, data *Req) ([]byte, bool, error)
+	GetLastMessage(ctx context.Context, data *Req) ([]byte, error)
 }
 
 var _ MessageHandler = (*GrpcHandler)(nil)
@@ -266,3 +267,15 @@ func (g *GrpcHandler) SetUserDeviceBackground(ctx context.Context, data *Req) ([
 	}
 	return nil, req.IsBackground, nil
 }
+
+func (g *GrpcHandler) GetLastMessage(ctx context.Context, data *Req) ([]byte, error) {
+	var req msg.GetLastMessageReq
+	if err := proto.Unmarshal(data.Data, &req); err != nil {
+		return nil, err
+	}
+	resp, err := g.msgClient.GetLastMessage(ctx, &req)
+	if err != nil {
+		return nil, err
+	}
+	return proto.Marshal(resp)
+}

internal/msgtransfer/init.go

@@ -18,11 +18,17 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net"
 	"net/http"
 	"os"
 	"os/signal"
+	"strconv"
 	"syscall"
 
+	"github.com/openimsdk/tools/discovery/etcd"
+	"github.com/openimsdk/tools/utils/jsonutil"
+	"github.com/openimsdk/tools/utils/network"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo"
@@ -30,8 +36,9 @@ import (
 	"github.com/openimsdk/tools/db/redisutil"
 	"github.com/openimsdk/tools/utils/datautil"
 
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	conf "github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	discRegister "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
+	kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller"
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
@@ -53,16 +60,17 @@ type MsgTransfer struct {
 }
 
 type Config struct {
-	MsgTransfer    config.MsgTransfer
+	MsgTransfer    conf.MsgTransfer
-	RedisConfig    config.Redis
+	RedisConfig    conf.Redis
-	MongodbConfig  config.Mongo
+	MongodbConfig  conf.Mongo
-	KafkaConfig    config.Kafka
+	KafkaConfig    conf.Kafka
-	Share          config.Share
+	Share          conf.Share
-	WebhooksConfig config.Webhooks
+	WebhooksConfig conf.Webhooks
-	Discovery      config.Discovery
+	Discovery      conf.Discovery
 }
 
 func Start(ctx context.Context, index int, config *Config) error {
+
 	log.CInfo(ctx, "MSG-TRANSFER server is initializing", "prometheusPorts",
 		config.MsgTransfer.Prometheus.Ports, "index", index)
 
@@ -74,7 +82,7 @@ func Start(ctx context.Context, index int, config *Config) error {
 	if err != nil {
 		return err
 	}
-	client, err := discRegister.NewDiscoveryRegister(&config.Discovery, &config.Share)
+	client, err := discRegister.NewDiscoveryRegister(&config.Discovery, &config.Share, nil)
 	if err != nil {
 		return err
 	}
@@ -116,7 +124,7 @@ func Start(ctx context.Context, index int, config *Config) error {
 	return msgTransfer.Start(index, config)
 }
 
-func (m *MsgTransfer) Start(index int, config *Config) error {
+func (m *MsgTransfer) Start(index int, cfg *Config) error {
 	m.ctx, m.cancel = context.WithCancel(context.Background())
 	var (
 		netDone = make(chan struct{}, 1)
@@ -131,16 +139,67 @@ func (m *MsgTransfer) Start(index int, config *Config) error {
 		return err
 	}
 
-	if config.MsgTransfer.Prometheus.Enable {
+	client, err := kdisc.NewDiscoveryRegister(&cfg.Discovery, &cfg.Share, nil)
-		go func() {
-			prometheusPort, err := datautil.GetElemByIndex(config.MsgTransfer.Prometheus.Ports, index)
 	if err != nil {
-				netErr = err
+		return errs.WrapMsg(err, "failed to register discovery service")
-				netDone <- struct{}{}
-				return
 	}
 
-			if err := prommetrics.TransferInit(prometheusPort); err != nil && !errors.Is(err, http.ErrServerClosed) {
+	registerIP, err := network.GetRpcRegisterIP("")
+	if err != nil {
+		return err
+	}
+
+	getAutoPort := func() (net.Listener, int, error) {
+		registerAddr := net.JoinHostPort(registerIP, "0")
+		listener, err := net.Listen("tcp", registerAddr)
+		if err != nil {
+			return nil, 0, errs.WrapMsg(err, "listen err", "registerAddr", registerAddr)
+		}
+		_, portStr, _ := net.SplitHostPort(listener.Addr().String())
+		port, _ := strconv.Atoi(portStr)
+		return listener, port, nil
+	}
+
+	if cfg.MsgTransfer.Prometheus.AutoSetPorts && cfg.Discovery.Enable != conf.ETCD {
+		return errs.New("only etcd support autoSetPorts", "RegisterName", "api").Wrap()
+	}
+
+	if cfg.MsgTransfer.Prometheus.Enable {
+		var (
+			listener       net.Listener
+			prometheusPort int
+		)
+
+		if cfg.MsgTransfer.Prometheus.AutoSetPorts {
+			listener, prometheusPort, err = getAutoPort()
+			if err != nil {
+				return err
+			}
+
+			etcdClient := client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
+
+			_, err = etcdClient.Put(context.TODO(), prommetrics.BuildDiscoveryKey(prommetrics.MessageTransferKeyName), jsonutil.StructToJsonString(prommetrics.BuildDefaultTarget(registerIP, prometheusPort)))
+			if err != nil {
+				return errs.WrapMsg(err, "etcd put err")
+			}
+		} else {
+			prometheusPort, err = datautil.GetElemByIndex(cfg.MsgTransfer.Prometheus.Ports, index)
+			if err != nil {
+				return err
+			}
+			listener, err = net.Listen("tcp", fmt.Sprintf(":%d", prometheusPort))
+			if err != nil {
+				return errs.WrapMsg(err, "listen err", "addr", fmt.Sprintf(":%d", prometheusPort))
+			}
+		}
+
+		go func() {
+			defer func() {
+				if r := recover(); r != nil {
+					log.ZPanic(m.ctx, "MsgTransfer Start Panic", errs.ErrPanic(r))
+				}
+			}()
+			if err := prommetrics.TransferInit(listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
 				netErr = errs.WrapMsg(err, "prometheus start error", "prometheusPort", prometheusPort)
 				netDone <- struct{}{}
 			}

internal/push/offlinepush/dummy/push.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
 	"github.com/openimsdk/tools/log"
+	"sync/atomic"
 )
 
 func NewClient() *Dummy {
@@ -25,10 +26,12 @@ func NewClient() *Dummy {
 }
 
 type Dummy struct {
+	v atomic.Bool
 }
 
 func (d *Dummy) Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error {
-	log.ZDebug(ctx, "dummy push")
+	if d.v.CompareAndSwap(false, true) {
-	log.ZWarn(ctx, "Dummy push", nil, "ps", "The offline push is not configured. To configure it, please go to config/openim-push.yml.")
+		log.ZWarn(ctx, "dummy push", nil, "ps", "the offline push is not configured. to configure it, please go to config/openim-push.yml")
+	}
 	return nil
 }

internal/push/offlinepush/fcm/push.go

@@ -16,12 +16,14 @@ package fcm
 
 import (
 	"context"
+	"errors"
 	"fmt"
-	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
-	"github.com/openimsdk/tools/utils/httputil"
 	"path/filepath"
 	"strings"
 
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
+	"github.com/openimsdk/tools/utils/httputil"
+
 	firebase "firebase.google.com/go/v4"
 	"firebase.google.com/go/v4/messaging"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
@@ -133,7 +135,7 @@ func (f *Fcm) Push(ctx context.Context, userIDs []string, title, content string,
 			unreadCountSum, err := f.cache.GetUserBadgeUnreadCountSum(ctx, userID)
 			if err == nil && unreadCountSum != 0 {
 				apns.Payload.Aps.Badge = &unreadCountSum
-			} else if err == redis.Nil || unreadCountSum == 0 {
+			} else if errors.Is(err, redis.Nil) || unreadCountSum == 0 {
 				zero := 1
 				apns.Payload.Aps.Badge = &zero
 			} else {

internal/push/offlinepush/getui/body.go

@@ -18,6 +18,16 @@ import (
 	"fmt"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/tools/utils/datautil"
+)
+
+var (
+	incOne          = datautil.ToPtr("+1")
+	addNum          = "1"
+	defaultStrategy = strategy{
+		Default: 1,
+	}
+	msgCategory = "CATEGORY_MESSAGE"
 )
 
 type Resp struct {
@@ -59,6 +69,23 @@ type TaskResp struct {
 
 type Settings struct {
 	TTL      *int64   `json:"ttl"`
+	Strategy strategy `json:"strategy"`
+}
+
+type strategy struct {
+	Default int64 `json:"default"`
+	//IOS     int64 `json:"ios"`
+	//St      int64 `json:"st"`
+	//Hw      int64 `json:"hw"`
+	//Ho      int64 `json:"ho"`
+	//XM      int64 `json:"xm"`
+	//XMG     int64 `json:"xmg"`
+	//VV      int64 `json:"vv"`
+	//Op      int64 `json:"op"`
+	//OpG     int64 `json:"opg"`
+	//MZ      int64 `json:"mz"`
+	//HosHw   int64 `json:"hoshw"`
+	//WX      int64 `json:"wx"`
 }
 
 type Audience struct {
@@ -112,6 +139,8 @@ type Notification struct {
 	ChannelID   string `json:"channelID"`
 	ChannelName string `json:"ChannelName"`
 	ClickType   string `json:"click_type"`
+	BadgeAddNum string `json:"badge_add_num"`
+	Category    string `json:"category"`
 }
 
 type Options struct {
@@ -120,6 +149,7 @@ type Options struct {
 		ChannelID    string `json:"/message/android/notification/channel_id"`
 		Sound        string `json:"/message/android/notification/sound"`
 		Importance   string `json:"/message/android/notification/importance"`
+		Category     string `json:"/message/android/category"`
 	} `json:"HW"`
 	XM struct {
 		ChannelID string `json:"/extra.channel_id"`
@@ -140,6 +170,8 @@ func newPushReq(pushConf *config.Push, title, content string) PushReq {
 		ClickType:   "startapp",
 		ChannelID:   pushConf.GeTui.ChannelID,
 		ChannelName: pushConf.GeTui.ChannelName,
+		BadgeAddNum: addNum,
+		Category:    msgCategory,
 	}}}
 	return pushReq
 }
@@ -156,6 +188,7 @@ func (pushReq *PushReq) setPushChannel(title string, body string) {
 	notify := "notify"
 	pushReq.PushChannel.Ios.NotificationType = &notify
 	pushReq.PushChannel.Ios.Aps.Sound = "default"
+	pushReq.PushChannel.Ios.AutoBadge = incOne
 	pushReq.PushChannel.Ios.Aps.Alert = Alert{
 		Title: title,
 		Body:  body,
@@ -172,7 +205,8 @@ func (pushReq *PushReq) setPushChannel(title string, body string) {
 			ChannelID    string `json:"/message/android/notification/channel_id"`
 			Sound        string `json:"/message/android/notification/sound"`
 			Importance   string `json:"/message/android/notification/importance"`
-		}{ChannelID: "RingRing4", Sound: "/raw/ring001", Importance: "NORMAL"},
+			Category     string `json:"/message/android/category"`
+		}{ChannelID: "RingRing4", Sound: "/raw/ring001", Importance: "NORMAL", Category: "IM"},
 		XM: struct {
 			ChannelID string `json:"/extra.channel_id"`
 		}{ChannelID: "high_system"},

internal/push/offlinepush/getui/push.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"errors"
 	"strconv"
 	"sync"
 	"time"
@@ -70,7 +71,7 @@ func NewClient(pushConf *config.Push, cache cache.ThirdCache) *Client {
 func (g *Client) Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error {
 	token, err := g.cache.GetGetuiToken(ctx)
 	if err != nil {
-		if errs.Unwrap(err) == redis.Nil {
+		if errors.Is(err, redis.Nil) {
 			log.ZDebug(ctx, "getui token not exist in redis")
 			token, err = g.getTokenAndSave2Redis(ctx)
 			if err != nil {
@@ -144,7 +145,7 @@ func (g *Client) Auth(ctx context.Context, timeStamp int64) (token string, expir
 func (g *Client) GetTaskID(ctx context.Context, token string, pushReq PushReq) (string, error) {
 	respTask := TaskResp{}
 	ttl := int64(1000 * 60 * 5)
-	pushReq.Settings = &Settings{TTL: &ttl}
+	pushReq.Settings = &Settings{TTL: &ttl, Strategy: defaultStrategy}
 	err := g.request(ctx, taskURL, pushReq, token, &respTask)
 	if err != nil {
 		return "", errs.Wrap(err)
@@ -188,6 +189,7 @@ func (g *Client) postReturn(
 	if err != nil {
 		return err
 	}
+	log.ZDebug(ctx, "postReturn", "url", url, "header", header, "input", input, "timeout", timeout, "output", output)
 	return output.parseError()
 }
 
@@ -204,7 +206,7 @@ func (g *Client) getTokenAndSave2Redis(ctx context.Context) (token string, err e
 }
 
 func (g *Client) GetTaskIDAndSave2Redis(ctx context.Context, token string, pushReq PushReq) (taskID string, err error) {
-	pushReq.Settings = &Settings{TTL: &g.taskIDTTL}
+	pushReq.Settings = &Settings{TTL: &g.taskIDTTL, Strategy: defaultStrategy}
 	taskID, err = g.GetTaskID(ctx, token, pushReq)
 	if err != nil {
 		return

internal/rpc/auth/auth.go

@@ -17,6 +17,7 @@ package auth
 import (
 	"context"
 	"errors"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
@@ -118,9 +119,13 @@ func (s *authServer) GetUserToken(ctx context.Context, req *pbauth.GetUserTokenR
 	if authverify.IsManagerUserID(req.UserID, s.config.Share.IMAdminUserID) {
 		return nil, errs.ErrNoPermission.WrapMsg("don't get Admin token")
 	}
-	if err := s.userClient.CheckUser(ctx, []string{req.UserID}); err != nil {
+	user, err := s.userClient.GetUserInfo(ctx, req.UserID)
+	if err != nil {
 		return nil, err
 	}
+	if user.AppMangerLevel >= constant.AppNotificationAdmin {
+		return nil, errs.ErrArgs.WrapMsg("app account can`t get token")
+	}
 	token, err := s.authDatabase.CreateToken(ctx, req.UserID, int(req.PlatformID))
 	if err != nil {
 		return nil, err

internal/rpc/msg/sync_msg.go

@@ -245,3 +245,11 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq
 func (m *msgServer) GetServerTime(ctx context.Context, _ *msg.GetServerTimeReq) (*msg.GetServerTimeResp, error) {
 	return &msg.GetServerTimeResp{ServerTime: timeutil.GetCurrentTimestampByMill()}, nil
 }
+
+func (m *msgServer) GetLastMessage(ctx context.Context, req *msg.GetLastMessageReq) (*msg.GetLastMessageResp, error) {
+	msgs, err := m.MsgDatabase.GetLastMessage(ctx, req.ConversationIDs, req.UserID)
+	if err != nil {
+		return nil, err
+	}
+	return &msg.GetLastMessageResp{Msgs: msgs}, nil
+}

internal/rpc/user/user.go

@@ -17,7 +17,6 @@ package user
 import (
 	"context"
 	"errors"
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
 	"math/rand"
 	"strings"
 	"sync"
@@ -32,6 +31,7 @@ import (
 	tablerelation "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/webhook"
 	"github.com/openimsdk/open-im-server/v3/pkg/localcache"
+	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
 	"github.com/openimsdk/protocol/group"
 	friendpb "github.com/openimsdk/protocol/relation"
 	"github.com/openimsdk/tools/db/redisutil"
@@ -480,7 +480,9 @@ func (s *userServer) AddNotificationAccount(ctx context.Context, req *pbuser.Add
 	if err := authverify.CheckAdmin(ctx, s.config.Share.IMAdminUserID); err != nil {
 		return nil, err
 	}
-
+	if req.AppMangerLevel < constant.AppNotificationAdmin {
+		return nil, errs.ErrArgs.WithDetail("app level not supported")
+	}
 	if req.UserID == "" {
 		for i := 0; i < 20; i++ {
 			userId := s.genUserID()
@@ -506,7 +508,7 @@ func (s *userServer) AddNotificationAccount(ctx context.Context, req *pbuser.Add
 		Nickname:       req.NickName,
 		FaceURL:        req.FaceURL,
 		CreateTime:     time.Now(),
-		AppMangerLevel: constant.AppNotificationAdmin,
+		AppMangerLevel: req.AppMangerLevel,
 	}
 	if err := s.db.Create(ctx, []*tablerelation.User{user}); err != nil {
 		return nil, err
@@ -516,6 +518,7 @@ func (s *userServer) AddNotificationAccount(ctx context.Context, req *pbuser.Add
 		UserID:         req.UserID,
 		NickName:       req.NickName,
 		FaceURL:        req.FaceURL,
+		AppMangerLevel: req.AppMangerLevel,
 	}, nil
 }
 
@@ -595,8 +598,13 @@ func (s *userServer) GetNotificationAccount(ctx context.Context, req *pbuser.Get
 	if err != nil {
 		return nil, servererrs.ErrUserIDNotFound.Wrap()
 	}
-	if user.AppMangerLevel == constant.AppAdmin || user.AppMangerLevel == constant.AppNotificationAdmin {
+	if user.AppMangerLevel == constant.AppAdmin || user.AppMangerLevel >= constant.AppNotificationAdmin {
-		return &pbuser.GetNotificationAccountResp{}, nil
+		return &pbuser.GetNotificationAccountResp{Account: &pbuser.NotificationAccountInfo{
+			UserID:         user.UserID,
+			FaceURL:        user.FaceURL,
+			NickName:       user.Nickname,
+			AppMangerLevel: user.AppMangerLevel,
+		}}, nil
 	}
 
 	return nil, errs.ErrNoPermission.WrapMsg("notification messages cannot be sent for this ID")
@@ -621,11 +629,12 @@ func (s *userServer) userModelToResp(users []*tablerelation.User, pagination pag
 	accounts := make([]*pbuser.NotificationAccountInfo, 0)
 	var total int64
 	for _, v := range users {
-		if v.AppMangerLevel == constant.AppNotificationAdmin && !datautil.Contain(v.UserID, s.config.Share.IMAdminUserID...) {
+		if v.AppMangerLevel >= constant.AppNotificationAdmin && !datautil.Contain(v.UserID, s.config.Share.IMAdminUserID...) {
 			temp := &pbuser.NotificationAccountInfo{
 				UserID:         v.UserID,
 				FaceURL:        v.FaceURL,
 				NickName:       v.Nickname,
+				AppMangerLevel: v.AppMangerLevel,
 			}
 			accounts = append(accounts, temp)
 			total += 1

internal/tools/cron_task.go

@@ -16,6 +16,7 @@ package tools
 
 import (
 	"context"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
 	pbconversation "github.com/openimsdk/protocol/conversation"
@@ -43,7 +44,7 @@ func Start(ctx context.Context, config *CronTaskConfig) error {
 	if config.CronTask.RetainChatRecords < 1 {
 		return errs.New("msg destruct time must be greater than 1").Wrap()
 	}
-	client, err := kdisc.NewDiscoveryRegister(&config.Discovery, &config.Share)
+	client, err := kdisc.NewDiscoveryRegister(&config.Discovery, &config.Share, nil)
 	if err != nil {
 		return errs.WrapMsg(err, "failed to register discovery service")
 	}

pkg/common/cmd/auth.go

@@ -55,6 +55,10 @@ func (a *AuthRpcCmd) Exec() error {
 
 func (a *AuthRpcCmd) runE() error {
 	return startrpc.Start(a.ctx, &a.authConfig.Discovery, &a.authConfig.RpcConfig.Prometheus, a.authConfig.RpcConfig.RPC.ListenIP,
-		a.authConfig.RpcConfig.RPC.RegisterIP, a.authConfig.RpcConfig.RPC.Ports,
+		a.authConfig.RpcConfig.RPC.RegisterIP, a.authConfig.RpcConfig.RPC.AutoSetPorts, a.authConfig.RpcConfig.RPC.Ports,
-		a.Index(), a.authConfig.Share.RpcRegisterName.Auth, &a.authConfig.Share, a.authConfig, auth.Start)
+		a.Index(), a.authConfig.Share.RpcRegisterName.Auth, &a.authConfig.Share, a.authConfig,
+		[]string{
+			a.authConfig.Share.RpcRegisterName.MessageGateway,
+		},
+		auth.Start)
 }

pkg/common/cmd/conversation.go

@@ -57,6 +57,8 @@ func (a *ConversationRpcCmd) Exec() error {
 
 func (a *ConversationRpcCmd) runE() error {
 	return startrpc.Start(a.ctx, &a.conversationConfig.Discovery, &a.conversationConfig.RpcConfig.Prometheus, a.conversationConfig.RpcConfig.RPC.ListenIP,
-		a.conversationConfig.RpcConfig.RPC.RegisterIP, a.conversationConfig.RpcConfig.RPC.Ports,
+		a.conversationConfig.RpcConfig.RPC.RegisterIP, a.conversationConfig.RpcConfig.RPC.AutoSetPorts, a.conversationConfig.RpcConfig.RPC.Ports,
-		a.Index(), a.conversationConfig.Share.RpcRegisterName.Conversation, &a.conversationConfig.Share, a.conversationConfig, conversation.Start)
+		a.Index(), a.conversationConfig.Share.RpcRegisterName.Conversation, &a.conversationConfig.Share, a.conversationConfig,
+		nil,
+		conversation.Start)
 }

pkg/common/cmd/friend.go

@@ -58,6 +58,8 @@ func (a *FriendRpcCmd) Exec() error {
 
 func (a *FriendRpcCmd) runE() error {
 	return startrpc.Start(a.ctx, &a.relationConfig.Discovery, &a.relationConfig.RpcConfig.Prometheus, a.relationConfig.RpcConfig.RPC.ListenIP,
-		a.relationConfig.RpcConfig.RPC.RegisterIP, a.relationConfig.RpcConfig.RPC.Ports,
+		a.relationConfig.RpcConfig.RPC.RegisterIP, a.relationConfig.RpcConfig.RPC.AutoSetPorts, a.relationConfig.RpcConfig.RPC.Ports,
-		a.Index(), a.relationConfig.Share.RpcRegisterName.Friend, &a.relationConfig.Share, a.relationConfig, relation.Start)
+		a.Index(), a.relationConfig.Share.RpcRegisterName.Friend, &a.relationConfig.Share, a.relationConfig,
+		nil,
+		relation.Start)
 }

pkg/common/cmd/group.go

@@ -59,6 +59,8 @@ func (a *GroupRpcCmd) Exec() error {
 
 func (a *GroupRpcCmd) runE() error {
 	return startrpc.Start(a.ctx, &a.groupConfig.Discovery, &a.groupConfig.RpcConfig.Prometheus, a.groupConfig.RpcConfig.RPC.ListenIP,
-		a.groupConfig.RpcConfig.RPC.RegisterIP, a.groupConfig.RpcConfig.RPC.Ports,
+		a.groupConfig.RpcConfig.RPC.RegisterIP, a.groupConfig.RpcConfig.RPC.AutoSetPorts, a.groupConfig.RpcConfig.RPC.Ports,
-		a.Index(), a.groupConfig.Share.RpcRegisterName.Group, &a.groupConfig.Share, a.groupConfig, group.Start, versionctx.EnableVersionCtx())
+		a.Index(), a.groupConfig.Share.RpcRegisterName.Group, &a.groupConfig.Share, a.groupConfig,
+		nil,
+		group.Start, versionctx.EnableVersionCtx())
 }

pkg/common/cmd/msg.go

@@ -59,6 +59,8 @@ func (a *MsgRpcCmd) Exec() error {
 
 func (a *MsgRpcCmd) runE() error {
 	return startrpc.Start(a.ctx, &a.msgConfig.Discovery, &a.msgConfig.RpcConfig.Prometheus, a.msgConfig.RpcConfig.RPC.ListenIP,
-		a.msgConfig.RpcConfig.RPC.RegisterIP, a.msgConfig.RpcConfig.RPC.Ports,
+		a.msgConfig.RpcConfig.RPC.RegisterIP, a.msgConfig.RpcConfig.RPC.AutoSetPorts, a.msgConfig.RpcConfig.RPC.Ports,
-		a.Index(), a.msgConfig.Share.RpcRegisterName.Msg, &a.msgConfig.Share, a.msgConfig, msg.Start)
+		a.Index(), a.msgConfig.Share.RpcRegisterName.Msg, &a.msgConfig.Share, a.msgConfig,
+		nil,
+		msg.Start)
 }

pkg/common/cmd/push.go

@@ -59,6 +59,10 @@ func (a *PushRpcCmd) Exec() error {
 
 func (a *PushRpcCmd) runE() error {
 	return startrpc.Start(a.ctx, &a.pushConfig.Discovery, &a.pushConfig.RpcConfig.Prometheus, a.pushConfig.RpcConfig.RPC.ListenIP,
-		a.pushConfig.RpcConfig.RPC.RegisterIP, a.pushConfig.RpcConfig.RPC.Ports,
+		a.pushConfig.RpcConfig.RPC.RegisterIP, a.pushConfig.RpcConfig.RPC.AutoSetPorts, a.pushConfig.RpcConfig.RPC.Ports,
-		a.Index(), a.pushConfig.Share.RpcRegisterName.Push, &a.pushConfig.Share, a.pushConfig, push.Start)
+		a.Index(), a.pushConfig.Share.RpcRegisterName.Push, &a.pushConfig.Share, a.pushConfig,
+		[]string{
+			a.pushConfig.Share.RpcRegisterName.MessageGateway,
+		},
+		push.Start)
 }

pkg/common/cmd/third.go

@@ -58,6 +58,8 @@ func (a *ThirdRpcCmd) Exec() error {
 
 func (a *ThirdRpcCmd) runE() error {
 	return startrpc.Start(a.ctx, &a.thirdConfig.Discovery, &a.thirdConfig.RpcConfig.Prometheus, a.thirdConfig.RpcConfig.RPC.ListenIP,
-		a.thirdConfig.RpcConfig.RPC.RegisterIP, a.thirdConfig.RpcConfig.RPC.Ports,
+		a.thirdConfig.RpcConfig.RPC.RegisterIP, a.thirdConfig.RpcConfig.RPC.AutoSetPorts, a.thirdConfig.RpcConfig.RPC.Ports,
-		a.Index(), a.thirdConfig.Share.RpcRegisterName.Third, &a.thirdConfig.Share, a.thirdConfig, third.Start)
+		a.Index(), a.thirdConfig.Share.RpcRegisterName.Third, &a.thirdConfig.Share, a.thirdConfig,
+		nil,
+		third.Start)
 }

pkg/common/cmd/user.go

@@ -59,6 +59,8 @@ func (a *UserRpcCmd) Exec() error {
 
 func (a *UserRpcCmd) runE() error {
 	return startrpc.Start(a.ctx, &a.userConfig.Discovery, &a.userConfig.RpcConfig.Prometheus, a.userConfig.RpcConfig.RPC.ListenIP,
-		a.userConfig.RpcConfig.RPC.RegisterIP, a.userConfig.RpcConfig.RPC.Ports,
+		a.userConfig.RpcConfig.RPC.RegisterIP, a.userConfig.RpcConfig.RPC.AutoSetPorts, a.userConfig.RpcConfig.RPC.Ports,
-		a.Index(), a.userConfig.Share.RpcRegisterName.User, &a.userConfig.Share, a.userConfig, user.Start)
+		a.Index(), a.userConfig.Share.RpcRegisterName.User, &a.userConfig.Share, a.userConfig,
+		nil,
+		user.Start)
 }

pkg/common/config/config.go

@@ -21,6 +21,7 @@ import (
 	"github.com/openimsdk/tools/db/mongoutil"
 	"github.com/openimsdk/tools/db/redisutil"
 	"github.com/openimsdk/tools/mq/kafka"
+	"github.com/openimsdk/tools/s3/aws"
 	"github.com/openimsdk/tools/s3/cos"
 	"github.com/openimsdk/tools/s3/kodo"
 	"github.com/openimsdk/tools/s3/minio"
@@ -107,6 +108,7 @@ type API struct {
 	} `mapstructure:"api"`
 	Prometheus struct {
 		Enable       bool   `mapstructure:"enable"`
+		AutoSetPorts bool   `mapstructure:"autoSetPorts"`
 		Ports        []int  `mapstructure:"ports"`
 		GrafanaURL   string `mapstructure:"grafanaURL"`
 	} `mapstructure:"prometheus"`
@@ -177,6 +179,7 @@ type Prometheus struct {
 type MsgGateway struct {
 	RPC struct {
 		RegisterIP   string `mapstructure:"registerIP"`
+		AutoSetPorts bool   `mapstructure:"autoSetPorts"`
 		Ports        []int  `mapstructure:"ports"`
 	} `mapstructure:"rpc"`
 	Prometheus  Prometheus `mapstructure:"prometheus"`
@@ -190,13 +193,18 @@ type MsgGateway struct {
 }
 
 type MsgTransfer struct {
-	Prometheus Prometheus `mapstructure:"prometheus"`
+	Prometheus struct {
+		Enable       bool  `mapstructure:"enable"`
+		AutoSetPorts bool  `mapstructure:"autoSetPorts"`
+		Ports        []int `mapstructure:"ports"`
+	} `mapstructure:"prometheus"`
 }
 
 type Push struct {
 	RPC struct {
 		RegisterIP   string `mapstructure:"registerIP"`
 		ListenIP     string `mapstructure:"listenIP"`
+		AutoSetPorts bool   `mapstructure:"autoSetPorts"`
 		Ports        []int  `mapstructure:"ports"`
 	} `mapstructure:"rpc"`
 	Prometheus           Prometheus `mapstructure:"prometheus"`
@@ -232,6 +240,7 @@ type Auth struct {
 	RPC struct {
 		RegisterIP   string `mapstructure:"registerIP"`
 		ListenIP     string `mapstructure:"listenIP"`
+		AutoSetPorts bool   `mapstructure:"autoSetPorts"`
 		Ports        []int  `mapstructure:"ports"`
 	} `mapstructure:"rpc"`
 	Prometheus  Prometheus `mapstructure:"prometheus"`
@@ -244,6 +253,7 @@ type Conversation struct {
 	RPC struct {
 		RegisterIP   string `mapstructure:"registerIP"`
 		ListenIP     string `mapstructure:"listenIP"`
+		AutoSetPorts bool   `mapstructure:"autoSetPorts"`
 		Ports        []int  `mapstructure:"ports"`
 	} `mapstructure:"rpc"`
 	Prometheus Prometheus `mapstructure:"prometheus"`
@@ -253,6 +263,7 @@ type Friend struct {
 	RPC struct {
 		RegisterIP   string `mapstructure:"registerIP"`
 		ListenIP     string `mapstructure:"listenIP"`
+		AutoSetPorts bool   `mapstructure:"autoSetPorts"`
 		Ports        []int  `mapstructure:"ports"`
 	} `mapstructure:"rpc"`
 	Prometheus Prometheus `mapstructure:"prometheus"`
@@ -262,6 +273,7 @@ type Group struct {
 	RPC struct {
 		RegisterIP   string `mapstructure:"registerIP"`
 		ListenIP     string `mapstructure:"listenIP"`
+		AutoSetPorts bool   `mapstructure:"autoSetPorts"`
 		Ports        []int  `mapstructure:"ports"`
 	} `mapstructure:"rpc"`
 	Prometheus                 Prometheus `mapstructure:"prometheus"`
@@ -272,6 +284,7 @@ type Msg struct {
 	RPC struct {
 		RegisterIP   string `mapstructure:"registerIP"`
 		ListenIP     string `mapstructure:"listenIP"`
+		AutoSetPorts bool   `mapstructure:"autoSetPorts"`
 		Ports        []int  `mapstructure:"ports"`
 	} `mapstructure:"rpc"`
 	Prometheus   Prometheus `mapstructure:"prometheus"`
@@ -282,6 +295,7 @@ type Third struct {
 	RPC struct {
 		RegisterIP   string `mapstructure:"registerIP"`
 		ListenIP     string `mapstructure:"listenIP"`
+		AutoSetPorts bool   `mapstructure:"autoSetPorts"`
 		Ports        []int  `mapstructure:"ports"`
 	} `mapstructure:"rpc"`
 	Prometheus Prometheus `mapstructure:"prometheus"`
@@ -290,14 +304,7 @@ type Third struct {
 		Cos    Cos    `mapstructure:"cos"`
 		Oss    Oss    `mapstructure:"oss"`
 		Kodo   Kodo   `mapstructure:"kodo"`
-		Aws    struct {
+		Aws    Aws    `mapstructure:"aws"`
-			Endpoint        string `mapstructure:"endpoint"`
-			Region          string `mapstructure:"region"`
-			Bucket          string `mapstructure:"bucket"`
-			AccessKeyID     string `mapstructure:"accessKeyID"`
-			AccessKeySecret string `mapstructure:"accessKeySecret"`
-			PublicRead      bool   `mapstructure:"publicRead"`
-		} `mapstructure:"aws"`
 	} `mapstructure:"object"`
 }
 type Cos struct {
@@ -327,10 +334,20 @@ type Kodo struct {
 	PublicRead      bool   `mapstructure:"publicRead"`
 }
 
+type Aws struct {
+	Endpoint        string `mapstructure:"endpoint"`
+	Region          string `mapstructure:"region"`
+	Bucket          string `mapstructure:"bucket"`
+	AccessKeyID     string `mapstructure:"accessKeyID"`
+	SecretAccessKey string `mapstructure:"secretAccessKey"`
+	SessionToken    string `mapstructure:"sessionToken"`
+}
+
 type User struct {
 	RPC struct {
 		RegisterIP   string `mapstructure:"registerIP"`
 		ListenIP     string `mapstructure:"listenIP"`
+		AutoSetPorts bool   `mapstructure:"autoSetPorts"`
 		Ports        []int  `mapstructure:"ports"`
 	} `mapstructure:"rpc"`
 	Prometheus Prometheus `mapstructure:"prometheus"`
@@ -567,6 +584,16 @@ func (o *Kodo) Build() *kodo.Config {
 	}
 }
 
+func (o *Aws) Build() *aws.Config {
+	return &aws.Config{
+		Region:          o.Region,
+		Bucket:          o.Bucket,
+		AccessKeyID:     o.AccessKeyID,
+		SecretAccessKey: o.SecretAccessKey,
+		SessionToken:    o.SessionToken,
+	}
+}
+
 func (l *CacheConfig) Failed() time.Duration {
 	return time.Second * time.Duration(l.FailedExpire)
 }
@@ -578,3 +605,115 @@ func (l *CacheConfig) Success() time.Duration {
 func (l *CacheConfig) Enable() bool {
 	return l.Topic != "" && l.SlotNum > 0 && l.SlotSize > 0
 }
+
+var (
+	DiscoveryConfigFilename          = "discovery.yml"
+	KafkaConfigFileName              = "kafka.yml"
+	LocalCacheConfigFileName         = "local-cache.yml"
+	LogConfigFileName                = "log.yml"
+	MinioConfigFileName              = "minio.yml"
+	MongodbConfigFileName            = "mongodb.yml"
+	OpenIMAPICfgFileName             = "openim-api.yml"
+	OpenIMCronTaskCfgFileName        = "openim-crontask.yml"
+	OpenIMMsgGatewayCfgFileName      = "openim-msggateway.yml"
+	OpenIMMsgTransferCfgFileName     = "openim-msgtransfer.yml"
+	OpenIMPushCfgFileName            = "openim-push.yml"
+	OpenIMRPCAuthCfgFileName         = "openim-rpc-auth.yml"
+	OpenIMRPCConversationCfgFileName = "openim-rpc-conversation.yml"
+	OpenIMRPCFriendCfgFileName       = "openim-rpc-friend.yml"
+	OpenIMRPCGroupCfgFileName        = "openim-rpc-group.yml"
+	OpenIMRPCMsgCfgFileName          = "openim-rpc-msg.yml"
+	OpenIMRPCThirdCfgFileName        = "openim-rpc-third.yml"
+	OpenIMRPCUserCfgFileName         = "openim-rpc-user.yml"
+	RedisConfigFileName              = "redis.yml"
+	ShareFileName                    = "share.yml"
+	WebhooksConfigFileName           = "webhooks.yml"
+)
+
+func (d *Discovery) GetConfigFileName() string {
+	return DiscoveryConfigFilename
+}
+
+func (k *Kafka) GetConfigFileName() string {
+	return KafkaConfigFileName
+}
+
+func (lc *LocalCache) GetConfigFileName() string {
+	return LocalCacheConfigFileName
+}
+
+func (l *Log) GetConfigFileName() string {
+	return LogConfigFileName
+}
+
+func (m *Minio) GetConfigFileName() string {
+	return MinioConfigFileName
+}
+
+func (m *Mongo) GetConfigFileName() string {
+	return MongodbConfigFileName
+}
+
+func (n *Notification) GetConfigFileName() string {
+	return NotificationFileName
+}
+
+func (a *API) GetConfigFileName() string {
+	return OpenIMAPICfgFileName
+}
+
+func (ct *CronTask) GetConfigFileName() string {
+	return OpenIMCronTaskCfgFileName
+}
+
+func (mg *MsgGateway) GetConfigFileName() string {
+	return OpenIMMsgGatewayCfgFileName
+}
+
+func (mt *MsgTransfer) GetConfigFileName() string {
+	return OpenIMMsgTransferCfgFileName
+}
+
+func (p *Push) GetConfigFileName() string {
+	return OpenIMPushCfgFileName
+}
+
+func (a *Auth) GetConfigFileName() string {
+	return OpenIMRPCAuthCfgFileName
+}
+
+func (c *Conversation) GetConfigFileName() string {
+	return OpenIMRPCConversationCfgFileName
+}
+
+func (f *Friend) GetConfigFileName() string {
+	return OpenIMRPCFriendCfgFileName
+}
+
+func (g *Group) GetConfigFileName() string {
+	return OpenIMRPCGroupCfgFileName
+}
+
+func (m *Msg) GetConfigFileName() string {
+	return OpenIMRPCMsgCfgFileName
+}
+
+func (t *Third) GetConfigFileName() string {
+	return OpenIMRPCThirdCfgFileName
+}
+
+func (u *User) GetConfigFileName() string {
+	return OpenIMRPCUserCfgFileName
+}
+
+func (r *Redis) GetConfigFileName() string {
+	return RedisConfigFileName
+}
+
+func (s *Share) GetConfigFileName() string {
+	return ShareFileName
+}
+
+func (w *Webhooks) GetConfigFileName() string {
+	return WebhooksConfigFileName
+}

pkg/common/config/constant.go

@@ -14,7 +14,10 @@
 
 package config
 
-const ConfKey = "conf"
+const (
+	ConfKey = "conf"
+	ETCD    = "etcd"
+)
 
 const (
 	// DefaultDirPerm is used for creating general directories, allowing the owner to read, write, and execute,

pkg/common/config/load_config_test.go

@@ -59,3 +59,11 @@ func TestLoadOpenIMThirdConfig(t *testing.T) {
 
 	// Environment: IMENV_OPENIM_RPC_THIRD_OBJECT_ENABLE=enabled;IMENV_OPENIM_RPC_THIRD_OBJECT_OSS_ENDPOINT=https://oss-cn-chengdu.aliyuncs.com;IMENV_OPENIM_RPC_THIRD_OBJECT_OSS_BUCKET=my_bucket_name;IMENV_OPENIM_RPC_THIRD_OBJECT_OSS_BUCKETURL=https://my_bucket_name.oss-cn-chengdu.aliyuncs.com;IMENV_OPENIM_RPC_THIRD_OBJECT_OSS_ACCESSKEYID=AKID1234567890;IMENV_OPENIM_RPC_THIRD_OBJECT_OSS_ACCESSKEYSECRET=abc123xyz789;IMENV_OPENIM_RPC_THIRD_OBJECT_OSS_SESSIONTOKEN=session_token_value;IMENV_OPENIM_RPC_THIRD_OBJECT_OSS_PUBLICREAD=true
 }
+
+func TestTransferConfig(t *testing.T) {
+	var tran MsgTransfer
+	err := LoadConfig("../../../config/openim-msgtransfer.yml", "IMENV_OPENIM-MSGTRANSFER", &tran)
+	assert.Nil(t, err)
+	assert.Equal(t, true, tran.Prometheus.Enable)
+	assert.Equal(t, true, tran.Prometheus.AutoSetPorts)
+}

pkg/common/discoveryregister/discoveryregister.go

@@ -15,33 +15,25 @@
 package discoveryregister
 
 import (
+	"time"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister/kubernetes"
 	"github.com/openimsdk/tools/discovery"
 	"github.com/openimsdk/tools/discovery/etcd"
-	"github.com/openimsdk/tools/discovery/zookeeper"
 	"github.com/openimsdk/tools/errs"
-	"time"
 )
 
 // NewDiscoveryRegister creates a new service discovery and registry client based on the provided environment type.
-func NewDiscoveryRegister(discovery *config.Discovery, share *config.Share) (discovery.SvcDiscoveryRegistry, error) {
+func NewDiscoveryRegister(discovery *config.Discovery, share *config.Share, watchNames []string) (discovery.SvcDiscoveryRegistry, error) {
 	switch discovery.Enable {
-	case "zookeeper":
-		return zookeeper.NewZkClient(
-			discovery.ZooKeeper.Address,
-			discovery.ZooKeeper.Schema,
-			zookeeper.WithFreq(time.Hour),
-			zookeeper.WithUserNameAndPassword(discovery.ZooKeeper.Username, discovery.ZooKeeper.Password),
-			zookeeper.WithRoundRobin(),
-			zookeeper.WithTimeout(10),
-		)
 	case "k8s":
 		return kubernetes.NewK8sDiscoveryRegister(share.RpcRegisterName.MessageGateway)
 	case "etcd":
 		return etcd.NewSvcDiscoveryRegistry(
 			discovery.Etcd.RootDirectory,
 			discovery.Etcd.Address,
+			watchNames,
 			etcd.WithDialTimeout(10*time.Second),
 			etcd.WithMaxCallSendMsgSize(20*1024*1024),
 			etcd.WithUsernameAndPassword(discovery.Etcd.Username, discovery.Etcd.Password))

pkg/common/prommetrics/api.go

@@ -3,6 +3,7 @@ package prommetrics
 import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"net"
 	"strconv"
 )
 
@@ -23,14 +24,14 @@ var (
 	)
 )
 
-func ApiInit(prometheusPort int) error {
+func ApiInit(listener net.Listener) error {
 	apiRegistry := prometheus.NewRegistry()
 	cs := append(
 		baseCollector,
 		apiCounter,
 		httpCounter,
 	)
-	return Init(apiRegistry, prometheusPort, commonPath, promhttp.HandlerFor(apiRegistry, promhttp.HandlerOpts{}), cs...)
+	return Init(apiRegistry, listener, commonPath, promhttp.HandlerFor(apiRegistry, promhttp.HandlerOpts{}), cs...)
 }
 
 func APICall(path string, method string, apiCode int) {

pkg/common/prommetrics/discovery.go

@@ -0,0 +1,31 @@
+package prommetrics
+
+import "fmt"
+
+const (
+	APIKeyName             = "api"
+	MessageTransferKeyName = "message-transfer"
+)
+
+type Target struct {
+	Target string            `json:"target"`
+	Labels map[string]string `json:"labels"`
+}
+
+type RespTarget struct {
+	Targets []string          `json:"targets"`
+	Labels  map[string]string `json:"labels"`
+}
+
+func BuildDiscoveryKey(name string) string {
+	return fmt.Sprintf("%s/%s/%s", "openim", "prometheus_discovery", name)
+}
+
+func BuildDefaultTarget(host string, ip int) Target {
+	return Target{
+		Target: fmt.Sprintf("%s:%d", host, ip),
+		Labels: map[string]string{
+			"namespace": "default",
+		},
+	}
+}

pkg/common/prommetrics/prommetrics.go

@@ -15,9 +15,9 @@
 package prommetrics
 
 import (
-	"fmt"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/collectors"
+	"net"
 	"net/http"
 )
 
@@ -30,9 +30,9 @@ var (
 	}
 )
 
-func Init(registry *prometheus.Registry, prometheusPort int, path string, handler http.Handler, cs ...prometheus.Collector) error {
+func Init(registry *prometheus.Registry, listener net.Listener, path string, handler http.Handler, cs ...prometheus.Collector) error {
 	registry.MustRegister(cs...)
 	srv := http.NewServeMux()
 	srv.Handle(path, handler)
-	return http.ListenAndServe(fmt.Sprintf(":%d", prometheusPort), srv)
+	return http.Serve(listener, srv)
 }

pkg/common/prommetrics/rpc.go

@@ -5,6 +5,7 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"net"
 	"strconv"
 )
 
@@ -21,13 +22,13 @@ var (
 	)
 )
 
-func RpcInit(cs []prometheus.Collector, prometheusPort int) error {
+func RpcInit(cs []prometheus.Collector, listener net.Listener) error {
 	reg := prometheus.NewRegistry()
 	cs = append(append(
 		baseCollector,
 		rpcCounter,
 	), cs...)
-	return Init(reg, prometheusPort, rpcPath, promhttp.HandlerFor(reg, promhttp.HandlerOpts{Registry: reg}), cs...)
+	return Init(reg, listener, rpcPath, promhttp.HandlerFor(reg, promhttp.HandlerOpts{Registry: reg}), cs...)
 }
 
 func RPCCall(name string, path string, code int) {

pkg/common/prommetrics/transfer.go

@@ -17,6 +17,7 @@ package prommetrics
 import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"net"
 )
 
 var (
@@ -42,7 +43,7 @@ var (
 	})
 )
 
-func TransferInit(prometheusPort int) error {
+func TransferInit(listener net.Listener) error {
 	reg := prometheus.NewRegistry()
 	cs := append(
 		baseCollector,
@@ -52,5 +53,5 @@ func TransferInit(prometheusPort int) error {
 		MsgInsertMongoFailedCounter,
 		SeqSetFailedCounter,
 	)
-	return Init(reg, prometheusPort, commonPath, promhttp.HandlerFor(reg, promhttp.HandlerOpts{Registry: reg}), cs...)
+	return Init(reg, listener, commonPath, promhttp.HandlerFor(reg, promhttp.HandlerOpts{Registry: reg}), cs...)
 }

pkg/common/startrpc/start.go

@@ -16,10 +16,8 @@ package startrpc
 
 import (
 	"context"
+	"errors"
 	"fmt"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
-	"github.com/openimsdk/tools/utils/datautil"
-	"google.golang.org/grpc/status"
 	"net"
 	"net/http"
 	"os"
@@ -28,6 +26,12 @@ import (
 	"syscall"
 	"time"
 
+	conf "github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/tools/discovery/etcd"
+	"github.com/openimsdk/tools/utils/datautil"
+	"github.com/openimsdk/tools/utils/jsonutil"
+	"google.golang.org/grpc/status"
+
 	kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/tools/discovery"
@@ -40,28 +44,48 @@ import (
 )
 
 // Start rpc server.
-func Start[T any](ctx context.Context, discovery *config.Discovery, prometheusConfig *config.Prometheus, listenIP,
+func Start[T any](ctx context.Context, discovery *conf.Discovery, prometheusConfig *conf.Prometheus, listenIP,
-	registerIP string, rpcPorts []int, index int, rpcRegisterName string, share *config.Share, config T, rpcFn func(ctx context.Context,
+	registerIP string, autoSetPorts bool, rpcPorts []int, index int, rpcRegisterName string, share *conf.Share, config T,
-	config T, client discovery.SvcDiscoveryRegistry, server *grpc.Server) error, options ...grpc.ServerOption) error {
+	watchServiceNames []string,
+	rpcFn func(ctx context.Context, config T, client discovery.SvcDiscoveryRegistry, server *grpc.Server) error,
+	options ...grpc.ServerOption) error {
 
-	rpcPort, err := datautil.GetElemByIndex(rpcPorts, index)
+	var (
+		rpcTcpAddr     string
+		netDone        = make(chan struct{}, 2)
+		netErr         error
+		prometheusPort int
+	)
+
+	registerIP, err := network.GetRpcRegisterIP(registerIP)
 	if err != nil {
 		return err
 	}
 
-	log.CInfo(ctx, "RPC server is initializing", "rpcRegisterName", rpcRegisterName, "rpcPort", rpcPort,
+	if !autoSetPorts {
-		"prometheusPorts", prometheusConfig.Ports)
+		rpcPort, err := datautil.GetElemByIndex(rpcPorts, index)
-	rpcTcpAddr := net.JoinHostPort(network.GetListenIP(listenIP), strconv.Itoa(rpcPort))
-
-	listener, err := net.Listen(
-		"tcp",
-		rpcTcpAddr,
-	)
 		if err != nil {
-		return errs.WrapMsg(err, "listen err", "rpcTcpAddr", rpcTcpAddr)
+			return err
 		}
-	defer listener.Close()
+		rpcTcpAddr = net.JoinHostPort(network.GetListenIP(listenIP), strconv.Itoa(rpcPort))
-	client, err := kdisc.NewDiscoveryRegister(discovery, share)
+	} else {
+		rpcTcpAddr = net.JoinHostPort(network.GetListenIP(listenIP), "0")
+	}
+
+	getAutoPort := func() (net.Listener, int, error) {
+		listener, err := net.Listen("tcp", rpcTcpAddr)
+		if err != nil {
+			return nil, 0, errs.WrapMsg(err, "listen err", "rpcTcpAddr", rpcTcpAddr)
+		}
+		_, portStr, _ := net.SplitHostPort(listener.Addr().String())
+		port, _ := strconv.Atoi(portStr)
+		return listener, port, nil
+	}
+
+	if autoSetPorts && discovery.Enable != conf.ETCD {
+		return errs.New("only etcd support autoSetPorts", "rpcRegisterName", rpcRegisterName).Wrap()
+	}
+	client, err := kdisc.NewDiscoveryRegister(discovery, share, watchServiceNames)
 	if err != nil {
 		return err
 	}
@@ -81,41 +105,36 @@ func Start[T any](ctx context.Context, discovery *config.Discovery, prometheusCo
 			prommetricsUnaryInterceptor(rpcRegisterName),
 			prommetricsStreamInterceptor(rpcRegisterName),
 		)
-	} else {
-		options = append(options, mw.GrpcServer())
-	}
-
-	srv := grpc.NewServer(options...)
-
-	err = rpcFn(ctx, config, client, srv)
-	if err != nil {
-		return err
-	}
-
-	err = client.Register(
-		rpcRegisterName,
-		registerIP,
-		rpcPort,
-		grpc.WithTransportCredentials(insecure.NewCredentials()),
-	)
-	if err != nil {
-		return err
-	}
 
 		var (
-		netDone = make(chan struct{}, 2)
+			listener net.Listener
-		netErr  error
 		)
-	if prometheusConfig.Enable {
+
-		go func() {
+		if autoSetPorts {
-			prometheusPort, err := datautil.GetElemByIndex(prometheusConfig.Ports, index)
+			listener, prometheusPort, err = getAutoPort()
 			if err != nil {
-				netErr = err
+				return err
-				netDone <- struct{}{}
+			}
-				return
+
+			etcdClient := client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
+
+			_, err = etcdClient.Put(ctx, prommetrics.BuildDiscoveryKey(rpcRegisterName), jsonutil.StructToJsonString(prommetrics.BuildDefaultTarget(registerIP, prometheusPort)))
+			if err != nil {
+				return errs.WrapMsg(err, "etcd put err")
+			}
+		} else {
+			prometheusPort, err = datautil.GetElemByIndex(prometheusConfig.Ports, index)
+			if err != nil {
+				return err
+			}
+			listener, err = net.Listen("tcp", fmt.Sprintf(":%d", prometheusPort))
+			if err != nil {
+				return errs.WrapMsg(err, "listen err", "rpcTcpAddr", rpcTcpAddr)
+			}
 		}
 		cs := prommetrics.GetGrpcCusMetrics(rpcRegisterName, share)
-			if err := prommetrics.RpcInit(cs, prometheusPort); err != nil && err != http.ErrServerClosed {
+		go func() {
+			if err := prommetrics.RpcInit(cs, listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
 				netErr = errs.WrapMsg(err, fmt.Sprintf("rpc %s prometheus start err: %d", rpcRegisterName, prometheusPort))
 				netDone <- struct{}{}
 			}
@@ -127,11 +146,39 @@ func Start[T any](ctx context.Context, discovery *config.Discovery, prometheusCo
 			//	netDone <- struct{}{}
 			// }
 		}()
+	} else {
+		options = append(options, mw.GrpcServer())
+	}
+
+	listener, port, err := getAutoPort()
+	if err != nil {
+		return err
+	}
+
+	log.CInfo(ctx, "RPC server is initializing", "rpcRegisterName", rpcRegisterName, "rpcPort", port,
+		"prometheusPort", prometheusPort)
+
+	defer listener.Close()
+	srv := grpc.NewServer(options...)
+
+	err = rpcFn(ctx, config, client, srv)
+	if err != nil {
+		return err
+	}
+
+	err = client.Register(
+		rpcRegisterName,
+		registerIP,
+		port,
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	if err != nil {
+		return err
 	}
 
 	go func() {
 		err := srv.Serve(listener)
-		if err != nil {
+		if err != nil && !errors.Is(err, http.ErrServerClosed) {
 			netErr = errs.WrapMsg(err, "rpc start err: ", rpcTcpAddr)
 			netDone <- struct{}{}
 		}

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

@@ -2,7 +2,9 @@ package redis
 
 import (
 	"context"
+	"errors"
 	"fmt"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
@@ -56,7 +58,7 @@ func callLua(ctx context.Context, rdb redis.Scripter, script *redis.Script, keys
 		}
 	}
 	v, err := r.Result()
-	if err == redis.Nil {
+	if errors.Is(err, redis.Nil) {
 		err = nil
 	}
 	return v, errs.WrapMsg(err, "call lua err", "scriptHash", script.Hash(), "keys", keys, "args", args)

pkg/common/storage/controller/msg.go

@@ -97,6 +97,8 @@ type CommonMsgDatabase interface {
 	DeleteDoc(ctx context.Context, docID string) error
 
 	GetLastMessageSeqByTime(ctx context.Context, conversationID string, time int64) (int64, error)
+
+	GetLastMessage(ctx context.Context, conversationIDS []string, userID string) (map[string]*sdkws.MsgData, error)
 }
 
 func NewCommonMsgDatabase(msgDocModel database.Msg, msg cache.MsgCache, seqUser cache.SeqUser, seqConversation cache.SeqConversationCache, kafkaConf *config.Kafka) (CommonMsgDatabase, error) {
@@ -811,8 +813,29 @@ func (db *commonMsgDatabase) GetMessageBySeqs(ctx context.Context, conversationI
 		if v, ok := seqMsgs[seq]; ok {
 			res = append(res, convert.MsgDB2Pb(v.Msg))
 		} else {
-			res = append(res, &sdkws.MsgData{Seq: seq})
+			res = append(res, &sdkws.MsgData{Seq: seq, Status: constant.MsgStatusHasDeleted})
 		}
 	}
 	return res, nil
 }
+
+func (db *commonMsgDatabase) GetLastMessage(ctx context.Context, conversationIDs []string, userID string) (map[string]*sdkws.MsgData, error) {
+	res := make(map[string]*sdkws.MsgData)
+	for _, conversationID := range conversationIDs {
+		if _, ok := res[conversationID]; ok {
+			continue
+		}
+		msg, err := db.msgDocDatabase.GetLastMessage(ctx, conversationID)
+		if err != nil {
+			if errs.Unwrap(err) == mongo.ErrNoDocuments {
+				continue
+			}
+			return nil, err
+		}
+		tmp := []*model.MsgInfoModel{msg}
+		db.handlerDeleteAndRevoked(ctx, userID, tmp)
+		db.handlerQuote(ctx, userID, conversationID, tmp)
+		res[conversationID] = convert.MsgDB2Pb(msg.Msg)
+	}
+	return res, nil
+}

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

@@ -997,6 +997,68 @@ func (m *MsgMgo) GetLastMessageSeqByTime(ctx context.Context, conversationID str
 	return seq, nil
 }
 
+func (m *MsgMgo) GetLastMessage(ctx context.Context, conversationID string) (*model.MsgInfoModel, error) {
+	pipeline := []bson.M{
+		{
+			"$match": bson.M{
+				"doc_id": bson.M{
+					"$regex": fmt.Sprintf("^%s", conversationID),
+				},
+			},
+		},
+		{
+			"$match": bson.M{
+				"msgs.msg.status": bson.M{
+					"$lt": constant.MsgStatusHasDeleted,
+				},
+			},
+		},
+		{
+			"$sort": bson.M{
+				"_id": -1,
+			},
+		},
+		{
+			"$limit": 1,
+		},
+		{
+			"$project": bson.M{
+				"_id":    0,
+				"doc_id": 0,
+			},
+		},
+		{
+			"$unwind": "$msgs",
+		},
+		{
+			"$match": bson.M{
+				"msgs.msg.status": bson.M{
+					"$lt": constant.MsgStatusHasDeleted,
+				},
+			},
+		},
+		{
+			"$sort": bson.M{
+				"msgs.msg.seq": -1,
+			},
+		},
+		{
+			"$limit": 1,
+		},
+	}
+	type Result struct {
+		Msgs *model.MsgInfoModel `bson:"msgs"`
+	}
+	res, err := mongoutil.Aggregate[*Result](ctx, m.coll, pipeline)
+	if err != nil {
+		return nil, err
+	}
+	if len(res) == 0 {
+		return nil, errs.Wrap(mongo.ErrNoDocuments)
+	}
+	return res[0].Msgs, nil
+}
+
 func (m *MsgMgo) onlyFindDocIndex(ctx context.Context, docID string, indexes []int64) ([]*model.MsgInfoModel, error) {
 	if len(indexes) == 0 {
 		return nil, nil

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

@@ -112,3 +112,15 @@ func (o *S3Mongo) FindExpirationObject(ctx context.Context, engine string, expir
 func (o *S3Mongo) GetKeyCount(ctx context.Context, engine string, key string) (int64, error) {
 	return mongoutil.Count(ctx, o.coll, bson.M{"engine": engine, "key": key})
 }
+
+func (o *S3Mongo) GetEngineCount(ctx context.Context, engine string) (int64, error) {
+	return mongoutil.Count(ctx, o.coll, bson.M{"engine": engine})
+}
+
+func (o *S3Mongo) GetEngineInfo(ctx context.Context, engine string, limit int, skip int) ([]*model.Object, error) {
+	return mongoutil.Find[*model.Object](ctx, o.coll, bson.M{"engine": engine}, options.Find().SetLimit(int64(limit)).SetSkip(int64(skip)))
+}
+
+func (o *S3Mongo) UpdateEngine(ctx context.Context, oldEngine, oldName string, newEngine string) error {
+	return mongoutil.UpdateOne(ctx, o.coll, bson.M{"engine": oldEngine, "name": oldName}, bson.M{"$set": bson.M{"engine": newEngine}}, false)
+}

pkg/common/storage/database/msg.go

@@ -39,5 +39,6 @@ type Msg interface {
 	DeleteDoc(ctx context.Context, docID string) error
 	GetRandBeforeMsg(ctx context.Context, ts int64, limit int) ([]*model.MsgDocModel, error)
 	GetLastMessageSeqByTime(ctx context.Context, conversationID string, time int64) (int64, error)
+	GetLastMessage(ctx context.Context, conversationID string) (*model.MsgInfoModel, error)
 	FindSeqs(ctx context.Context, conversationID string, seqs []int64) ([]*model.MsgInfoModel, error)
 }

pkg/common/storage/database/object.go

@@ -27,4 +27,8 @@ type ObjectInfo interface {
 	Delete(ctx context.Context, engine string, name []string) error
 	FindExpirationObject(ctx context.Context, engine string, expiration time.Time, needDelType []string, count int64) ([]*model.Object, error)
 	GetKeyCount(ctx context.Context, engine string, key string) (int64, error)
+
+	GetEngineCount(ctx context.Context, engine string) (int64, error)
+	GetEngineInfo(ctx context.Context, engine string, limit int, skip int) ([]*model.Object, error)
+	UpdateEngine(ctx context.Context, oldEngine, oldName string, newEngine string) error
 }

pkg/rpccache/online.go

@@ -270,8 +270,8 @@ func (o *OnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]s
 		}
 	}
 
-	log.ZInfo(ctx, "get users online", "online users length", len(userIDs), "offline users length", len(offlineUserIDs), "cost", time.Since(t))
+	log.ZInfo(ctx, "get users online", "online users length", len(onlineUserIDs), "offline users length", len(offlineUserIDs), "cost", time.Since(t))
-	return userIDs, offlineUserIDs, nil
+	return onlineUserIDs, offlineUserIDs, nil
 }
 
 func (o *OnlineCache) setUserOnline(userID string, platformIDs []int32) {

tools/s3/README.md

@@ -0,0 +1,12 @@
+# After s3 switches the storage engine, convert the data
+
+- build
+```shell
+go build -o s3convert main.go
+```
+
+- start
+```shell
+./s3convert -config <config dir path> -name <old s3 name>
+# ./s3convert -config ./../../config -name minio
+```

tools/s3/internal/conversion.go

@@ -0,0 +1,202 @@
+package internal
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"github.com/mitchellh/mapstructure"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo"
+	"github.com/openimsdk/tools/db/mongoutil"
+	"github.com/openimsdk/tools/db/redisutil"
+	"github.com/openimsdk/tools/s3"
+	"github.com/openimsdk/tools/s3/aws"
+	"github.com/openimsdk/tools/s3/cos"
+	"github.com/openimsdk/tools/s3/kodo"
+	"github.com/openimsdk/tools/s3/minio"
+	"github.com/openimsdk/tools/s3/oss"
+	"github.com/spf13/viper"
+	"go.mongodb.org/mongo-driver/mongo"
+	"log"
+	"net/http"
+	"path/filepath"
+	"time"
+)
+
+const defaultTimeout = time.Second * 10
+
+func readConf(path string, val any) error {
+	v := viper.New()
+	v.SetConfigFile(path)
+	if err := v.ReadInConfig(); err != nil {
+		return err
+	}
+	fn := func(config *mapstructure.DecoderConfig) {
+		config.TagName = "mapstructure"
+	}
+	return v.Unmarshal(val, fn)
+}
+
+func getS3(path string, name string, thirdConf *config.Third) (s3.Interface, error) {
+	switch name {
+	case "minio":
+		ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
+		defer cancel()
+		var minioConf config.Minio
+		if err := readConf(filepath.Join(path, minioConf.GetConfigFileName()), &minioConf); err != nil {
+			return nil, err
+		}
+		var redisConf config.Redis
+		if err := readConf(filepath.Join(path, redisConf.GetConfigFileName()), &redisConf); err != nil {
+			return nil, err
+		}
+		rdb, err := redisutil.NewRedisClient(ctx, redisConf.Build())
+		if err != nil {
+			return nil, err
+		}
+		return minio.NewMinio(ctx, redis.NewMinioCache(rdb), *minioConf.Build())
+	case "cos":
+		return cos.NewCos(*thirdConf.Object.Cos.Build())
+	case "oss":
+		return oss.NewOSS(*thirdConf.Object.Oss.Build())
+	case "kodo":
+		return kodo.NewKodo(*thirdConf.Object.Kodo.Build())
+	case "aws":
+		return aws.NewAws(*thirdConf.Object.Aws.Build())
+	default:
+		return nil, fmt.Errorf("invalid object enable: %s", name)
+	}
+}
+
+func getMongo(path string) (database.ObjectInfo, error) {
+	var mongoConf config.Mongo
+	if err := readConf(filepath.Join(path, mongoConf.GetConfigFileName()), &mongoConf); err != nil {
+		return nil, err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
+	defer cancel()
+	mgocli, err := mongoutil.NewMongoDB(ctx, mongoConf.Build())
+	if err != nil {
+		return nil, err
+	}
+	return mgo.NewS3Mongo(mgocli.GetDB())
+}
+
+func Main(path string, engine string) error {
+	var thirdConf config.Third
+	if err := readConf(filepath.Join(path, thirdConf.GetConfigFileName()), &thirdConf); err != nil {
+		return err
+	}
+	if thirdConf.Object.Enable == engine {
+		return errors.New("same s3 storage")
+	}
+	s3db, err := getMongo(path)
+	if err != nil {
+		return err
+	}
+	oldS3, err := getS3(path, engine, &thirdConf)
+	if err != nil {
+		return err
+	}
+	newS3, err := getS3(path, thirdConf.Object.Enable, &thirdConf)
+	if err != nil {
+		return err
+	}
+	count, err := getEngineCount(s3db, oldS3.Engine())
+	if err != nil {
+		return err
+	}
+	log.Printf("engine %s count: %d", oldS3.Engine(), count)
+	var skip int
+	for i := 1; i <= count+1; i++ {
+		log.Printf("start %d/%d", i, count)
+		start := time.Now()
+		res, err := doObject(s3db, newS3, oldS3, skip)
+		if err != nil {
+			log.Printf("end [%s] %d/%d error %s", time.Since(start), i, count, err)
+			return err
+		}
+		log.Printf("end [%s] %d/%d result %+v", time.Since(start), i, count, *res)
+		if res.Skip {
+			skip++
+		}
+		if res.End {
+			break
+		}
+	}
+	return nil
+}
+
+func getEngineCount(db database.ObjectInfo, name string) (int, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
+	defer cancel()
+	count, err := db.GetEngineCount(ctx, name)
+	if err != nil {
+		return 0, err
+	}
+	return int(count), nil
+}
+
+func doObject(db database.ObjectInfo, newS3, oldS3 s3.Interface, skip int) (*Result, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
+	defer cancel()
+	infos, err := db.GetEngineInfo(ctx, oldS3.Engine(), 1, skip)
+	if err != nil {
+		return nil, err
+	}
+	if len(infos) == 0 {
+		return &Result{End: true}, nil
+	}
+	obj := infos[0]
+	if _, err := db.Take(ctx, newS3.Engine(), obj.Name); err == nil {
+		return &Result{Skip: true}, nil
+	} else if !errors.Is(err, mongo.ErrNoDocuments) {
+		return nil, err
+	}
+	downloadURL, err := oldS3.AccessURL(ctx, obj.Key, time.Hour, &s3.AccessURLOption{})
+	if err != nil {
+		return nil, err
+	}
+	putURL, err := newS3.PresignedPutObject(ctx, obj.Key, time.Hour)
+	if err != nil {
+		return nil, err
+	}
+	downloadResp, err := http.Get(downloadURL)
+	if err != nil {
+		return nil, err
+	}
+	defer downloadResp.Body.Close()
+	switch downloadResp.StatusCode {
+	case http.StatusNotFound:
+		return &Result{Skip: true}, nil
+	case http.StatusOK:
+	default:
+		return nil, fmt.Errorf("download object failed %s", downloadResp.Status)
+	}
+	log.Printf("file size %d", obj.Size)
+	request, err := http.NewRequest(http.MethodPut, putURL, downloadResp.Body)
+	if err != nil {
+		return nil, err
+	}
+	putResp, err := http.DefaultClient.Do(request)
+	if err != nil {
+		return nil, err
+	}
+	defer putResp.Body.Close()
+	if putResp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("put object failed %s", putResp.Status)
+	}
+	ctx, cancel = context.WithTimeout(context.Background(), defaultTimeout)
+	defer cancel()
+	if err := db.UpdateEngine(ctx, obj.Engine, obj.Name, newS3.Engine()); err != nil {
+		return nil, err
+	}
+	return &Result{}, nil
+}
+
+type Result struct {
+	Skip bool
+	End  bool
+}

tools/s3/main.go

@@ -0,0 +1,23 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"github.com/openimsdk/open-im-server/v3/tools/s3/internal"
+	"os"
+)
+
+func main() {
+	var (
+		name   string
+		config string
+	)
+	flag.StringVar(&name, "name", "", "old previous storage name")
+	flag.StringVar(&config, "config", "", "config directory")
+	flag.Parse()
+	if err := internal.Main(config, name); err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
+	fmt.Fprintln(os.Stdout, "success")
+}

version/version

@@ -1 +1 @@
-3.8.2
+v3.8.3-patch.2