Update CHANGELOG for release v3.8.2

* build: update Server version.

* update version.

.env

@@ -8,10 +8,10 @@ ALERTMANAGER_IMAGE=prom/alertmanager:v0.27.0
 GRAFANA_IMAGE=grafana/grafana:11.0.1
 
 OPENIM_WEB_FRONT_IMAGE=openim/openim-web-front:release-v3.8.1
-OPENIM_ADMIN_FRONT_IMAGE=openim/openim-admin-front:release-v1.8.3
+OPENIM_ADMIN_FRONT_IMAGE=openim/openim-admin-front:release-v1.8.2
 
 #FRONT_IMAGE: use aliyun images
 #OPENIM_WEB_FRONT_IMAGE=registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-web-front:release-v3.8.1
-#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.2
 
 DATA_DIR=./

CHANGELOG/CHANGELOG-3.8.md

@@ -0,0 +1,75 @@
+## [v3.8.2](https://github.com/openimsdk/open-im-server/releases/tag/v3.8.2) 	(2024-11-25)
+
+### New Features
+* feat: improve publish docker image workflows [#2697](https://github.com/openimsdk/open-im-server/pull/2697)
+* feat: Msg filter [#2703](https://github.com/openimsdk/open-im-server/pull/2703)
+* feat: provide the interface required [#2712](https://github.com/openimsdk/open-im-server/pull/2712)
+* feat: add webhooks of online status and remove zookeeper configuration. [#2716](https://github.com/openimsdk/open-im-server/pull/2716)
+* feat: Add More Multi Login Policy [#2770](https://github.com/openimsdk/open-im-server/pull/2770)
+* feat: Push configuration can ignore case sensitivity [#2775](https://github.com/openimsdk/open-im-server/pull/2775)
+* feat: support app update service [#2794](https://github.com/openimsdk/open-im-server/pull/2794)
+* feat: implement merge milestone PR to target-branch. [#2796](https://github.com/openimsdk/open-im-server/pull/2796)
+* feat: support app update service [#2811](https://github.com/openimsdk/open-im-server/pull/2811)
+* feat: ApplicationVersion move chat [#2813](https://github.com/openimsdk/open-im-server/pull/2813)
+* feat: Update login policy [#2822](https://github.com/openimsdk/open-im-server/pull/2822)
+* feat: merge js sdk [#2856](https://github.com/openimsdk/open-im-server/pull/2856)
+* feat: Print Panic Log [#2850](https://github.com/openimsdk/open-im-server/pull/2850)
+
+### Bug Fixes
+* fix: update load file logic. [#2700](https://github.com/openimsdk/open-im-server/pull/2700)
+* fix: the message I sent is not set to read seq in mongodb [#2718](https://github.com/openimsdk/open-im-server/pull/2718)
+* fix: cannot modify group member avatars [#2719](https://github.com/openimsdk/open-im-server/pull/2719)
+* fix: auth package import twice [#2724](https://github.com/openimsdk/open-im-server/pull/2724)
+* fix: join the group chat directly, notification type error [#2772](https://github.com/openimsdk/open-im-server/pull/2772)
+* fix: change update group member level logic [#2730](https://github.com/openimsdk/open-im-server/pull/2730)
+* fix: joinSource check args error. [#2773](https://github.com/openimsdk/open-im-server/pull/2773)
+* fix: Change group member roleLevel can`t send notification [#2777](https://github.com/openimsdk/open-im-server/pull/2777)
+* fix: client sends message status error to server [#2779](https://github.com/openimsdk/open-im-server/pull/2779)
+* fix: del UserB's conversation version cache when userA set conversati… [#2785](https://github.com/openimsdk/open-im-server/pull/2785)
+* fix: improve setConversationAtInfo logic. [#2782](https://github.com/openimsdk/open-im-server/pull/2782)
+* fix: improve transfer Owner logic when newOwner is mute. [#2790](https://github.com/openimsdk/open-im-server/pull/2790)
+* fix: improve getUserInfo logic. [#2792](https://github.com/openimsdk/open-im-server/pull/2792)
+* fix: improve time condition check mehtod. [#2804](https://github.com/openimsdk/open-im-server/pull/2804)
+* fix: webhook before online push [#2805](https://github.com/openimsdk/open-im-server/pull/2805)
+* fix: set own read seq in MongoDB when sender send a message. [#2808](https://github.com/openimsdk/open-im-server/pull/2808)
+* fix: solve err Notification when setGroupInfo. [#2806](https://github.com/openimsdk/open-im-server/pull/2806)
+* fix: improve condition check. [#2815](https://github.com/openimsdk/open-im-server/pull/2815)
+* fix: Write back message to Redis [#2836](https://github.com/openimsdk/open-im-server/pull/2836)
+* fix: get group return repeated result [#2842](https://github.com/openimsdk/open-im-server/pull/2842)
+* fix: SetConversations can update new conversation [#2838](https://github.com/openimsdk/open-im-server/pull/2838)
+* fix(push): push content with jpush [#2844](https://github.com/openimsdk/open-im-server/pull/2844)
+* fix #2860 migrate jpns to jpush [#2861](https://github.com/openimsdk/open-im-server/pull/2861)
+* fix: concurrent write to websocket connection [#2866](https://github.com/openimsdk/open-im-server/pull/2866)
+* fix: Remove admin token in redis [#2871](https://github.com/openimsdk/open-im-server/pull/2871)
+* Fix Push2User webhookBeforeOfflinePush [#2862](https://github.com/openimsdk/open-im-server/pull/2862)
+* fix: move workflow to correct path [#2837](https://github.com/openimsdk/open-im-server/pull/2837)
+* fix: del login Policy [#2825](https://github.com/openimsdk/open-im-server/pull/2825)
+* fix: Wrong Redis Error Check [#2876](https://github.com/openimsdk/open-im-server/pull/2876)
+* fix: minor log typo [#2881](https://github.com/openimsdk/open-im-server/pull/2881)
+* fix: webhookAfterSingleMsgRead will be called correctly [#2884](https://github.com/openimsdk/open-im-server/pull/2884)
+* fix: webhookBeforeSendSingleMsg will call before black and friend check [#2885](https://github.com/openimsdk/open-im-server/pull/2885)
+
+### Chores
+* chore: remove unused content [#2786](https://github.com/openimsdk/open-im-server/pull/2786)
+
+### Builds
+* build: improve workflows logic. [#2801](https://github.com/openimsdk/open-im-server/pull/2801)
+* build: implement version file update when release. [#2826](https://github.com/openimsdk/open-im-server/pull/2826)
+* build: update mongo and kafka start logic. [#2858](https://github.com/openimsdk/open-im-server/pull/2858)
+* build: create changelog tool and workflows. [#2869](https://github.com/openimsdk/open-im-server/pull/2869)
+* build(deps): bump github.com/golang-jwt/jwt/v4 from 4.5.0 to 4.5.1 [#2851](https://github.com/openimsdk/open-im-server/pull/2851)
+
+### Others
+* Revert: Change group member roleLevel can`t send notification [#2789](https://github.com/openimsdk/open-im-server/pull/2789)
+* Introducing OpenIM Guru on Gurubase.io [#2788](https://github.com/openimsdk/open-im-server/pull/2788)
+* revert: write msg to redis [#2883](https://github.com/openimsdk/open-im-server/pull/2883)
+* @lkzz made their first contribution in https://github.com/openimsdk/open-im-server/pull/2724
+ [#2724](https://github.com/openimsdk/open-im-server/pull/2724)
+* @alilestera made their first contribution in https://github.com/openimsdk/open-im-server/pull/2773
+ [#2773](https://github.com/openimsdk/open-im-server/pull/2773)
+* @kursataktas made their first contribution in https://github.com/openimsdk/open-im-server/pull/2788
+ [#2788](https://github.com/openimsdk/open-im-server/pull/2788)
+* @yoyo930021 made their first contribution in https://github.com/openimsdk/open-im-server/pull/2844
+ [#2844](https://github.com/openimsdk/open-im-server/pull/2844)
+* @wikylyu made their first contribution in https://github.com/openimsdk/open-im-server/pull/2861
+ [#2861](https://github.com/openimsdk/open-im-server/pull/2861)

CHANGELOG/README.md

@@ -0,0 +1,4 @@
+# CHANGELOGs
+
+- [CHANGELOG-3.8.md](./CHANGELOG-3.8.md)
+

Dockerfile

@@ -8,7 +8,7 @@ ENV SERVER_DIR=/openim-server
 WORKDIR $SERVER_DIR
 
 # Set the Go proxy to improve dependency resolution speed
-# ENV GOPROXY=https://goproxy.io,direct
+ENV GOPROXY=https://goproxy.io,direct
 
 # Copy all files from the current directory into the container
 COPY . .

README.md

@@ -16,6 +16,7 @@
 [![Best Practices](https://img.shields.io/badge/Best%20Practices-purple?style=for-the-badge)](https://www.bestpractices.dev/projects/8045)
 [![Good First Issues](https://img.shields.io/github/issues/openimsdk/open-im-server/good%20first%20issue?style=for-the-badge&logo=github)](https://github.com/openimsdk/open-im-server/issues?q=is%3Aissue+is%3Aopen+sort%3Aupdated-desc+label%3A%22good+first+issue%22)
 [![Language](https://img.shields.io/badge/Language-Go-blue.svg?style=for-the-badge&logo=go&logoColor=white)](https://golang.org/)
+[![Gurubase](https://img.shields.io/badge/Gurubase-Ask%20OpenIM%20Guru-006BFF?style=for-the-badge)](https://gurubase.io/g/openim)
 
      
 <p align="center">

config/openim-crontask.yml

@@ -1,4 +1,3 @@
 cronExecuteTime: 0 2 * * *
 retainChatRecords: 365
-fileExpireTime: 180
+fileExpireTime: 90
-deleteObjectType: ["msg-picture","msg-file", "msg-voice","msg-video","msg-video-snapshot","sdklog"]

config/openim-push.yml

@@ -13,8 +13,8 @@ prometheus:
   ports: [ 12170, 12171, 12172, 12173, 12174, 12175, 12176, 12177, 12178, 12179, 12180, 12182, 12183, 12184, 12185, 12186 ]
 
 maxConcurrentWorkers: 3
-#Use geTui for offline push notifications, or choose fcm or jpush; corresponding configuration settings must be specified.
+#Use geTui for offline push notifications, or choose fcm or jpns; corresponding configuration settings must be specified.
-enable: geTui
+enable:
 geTui:
   pushUrl: https://restapi.getui.com/v2/$appId
   masterSecret:

go.mod

@@ -1,6 +1,6 @@
 module github.com/openimsdk/open-im-server/v3
 
-go 1.22.7
+go 1.22.0
 
 toolchain go1.23.2
 
@@ -14,15 +14,15 @@ 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.63
+	github.com/openimsdk/protocol v0.0.72-alpha.55
-	github.com/openimsdk/tools v0.0.50-alpha.50
+	github.com/openimsdk/tools v0.0.50-alpha.32
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0
 	github.com/stretchr/testify v1.9.0
 	go.mongodb.org/mongo-driver v1.14.0
 	google.golang.org/api v0.170.0
-	google.golang.org/grpc v1.68.0
+	google.golang.org/grpc v1.66.2
-	google.golang.org/protobuf v1.35.1
+	google.golang.org/protobuf v1.34.2
 	gopkg.in/yaml.v3 v3.0.1
 )
 
@@ -50,31 +50,31 @@ require (
 
 require (
 	cloud.google.com/go v0.112.1 // indirect
-	cloud.google.com/go/compute/metadata v0.5.0 // indirect
+	cloud.google.com/go/compute/metadata v0.3.0 // indirect
 	cloud.google.com/go/firestore v1.15.0 // indirect
 	cloud.google.com/go/iam v1.1.7 // indirect
 	cloud.google.com/go/longrunning v0.5.5 // indirect
 	cloud.google.com/go/storage v1.40.0 // indirect
 	github.com/MicahParks/keyfunc v1.9.0 // indirect
 	github.com/aliyun/aliyun-oss-go-sdk v3.0.2+incompatible // indirect
-	github.com/aws/aws-sdk-go-v2 v1.32.5 // indirect
+	github.com/aws/aws-sdk-go-v2 v1.23.1 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.5.1 // indirect
-	github.com/aws/aws-sdk-go-v2/config v1.28.5 // indirect
+	github.com/aws/aws-sdk-go-v2/config v1.25.4 // indirect
-	github.com/aws/aws-sdk-go-v2/credentials v1.17.46 // indirect
+	github.com/aws/aws-sdk-go-v2/credentials v1.16.3 // indirect
-	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.20 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.14.5 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.24 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.2.4 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.24 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.5.4 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.1 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.7.1 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/v4a v1.2.4 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.10.1 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.2.4 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.5 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.10.4 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.16.4 // indirect
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.43.1 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sso v1.24.6 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.17.3 // indirect
-	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.5 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.20.1 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sts v1.33.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.25.4 // indirect
-	github.com/aws/smithy-go v1.22.1 // indirect
+	github.com/aws/smithy-go v1.17.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bytedance/sonic v1.11.6 // indirect
 	github.com/bytedance/sonic/loader v0.1.1 // indirect
@@ -177,13 +177,13 @@ require (
 	golang.org/x/arch v0.7.0 // indirect
 	golang.org/x/image v0.15.0 // indirect
 	golang.org/x/net v0.29.0 // indirect
-	golang.org/x/oauth2 v0.23.0 // indirect
+	golang.org/x/oauth2 v0.21.0 // indirect
 	golang.org/x/sys v0.25.0 // indirect
 	golang.org/x/text v0.18.0 // indirect
 	golang.org/x/time v0.5.0 // indirect
 	google.golang.org/appengine/v2 v2.0.2 // indirect
 	google.golang.org/genproto v0.0.0-20240213162025-012b6fc9bca9 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20240903143218-8af14fe29dc1 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20240604185151-ef581f913117 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 // indirect
 	gorm.io/gorm v1.25.8 // indirect
 	stathat.com/c/consistent v1.0.0 // indirect

go.sum

@@ -1,8 +1,8 @@
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cloud.google.com/go v0.112.1 h1:uJSeirPke5UNZHIb4SxfZklVSiWWVqW4oXlETwZziwM=
 cloud.google.com/go v0.112.1/go.mod h1:+Vbu+Y1UU+I1rjmzeMOb/8RfkKJK2Gyxi1X6jJCZLo4=
-cloud.google.com/go/compute/metadata v0.5.0 h1:Zr0eK8JbFv6+Wi4ilXAR8FJ3wyNdpxHKJNPos6LTZOY=
+cloud.google.com/go/compute/metadata v0.3.0 h1:Tz+eQXMEqDIKRsmY3cHTL6FVaynIjX2QxYC4trgAKZc=
-cloud.google.com/go/compute/metadata v0.5.0/go.mod h1:aHnloV2TPI38yx4s9+wAZhHykWvVCfu7hQbF+9CWoiY=
+cloud.google.com/go/compute/metadata v0.3.0/go.mod h1:zFmK7XCadkQkj6TtorcaGlCW1hT1fIilQDwofLpJ20k=
 cloud.google.com/go/firestore v1.15.0 h1:/k8ppuWOtNuDHt2tsRV42yI21uaGnKDEQnRFeBpbFF8=
 cloud.google.com/go/firestore v1.15.0/go.mod h1:GWOxFXcv8GZUtYpWHw/w6IuYNux/BtmeVTMmjrm4yhk=
 cloud.google.com/go/iam v1.1.7 h1:z4VHOhwKLF/+UYXAJDFwGtNF0b6gjsW1Pk9Ml0U/IoM=
@@ -21,42 +21,42 @@ github.com/MicahParks/keyfunc v1.9.0/go.mod h1:IdnCilugA0O/99dW+/MkvlyrsX8+L8+x9
 github.com/QcloudApi/qcloud_sign_golang v0.0.0-20141224014652-e4130a326409/go.mod h1:1pk82RBxDY/JZnPQrtqHlUFfCctgdorsd9M06fMynOM=
 github.com/aliyun/aliyun-oss-go-sdk v3.0.2+incompatible h1:8psS8a+wKfiLt1iVDX79F7Y6wUM49Lcha2FMXt4UM8g=
 github.com/aliyun/aliyun-oss-go-sdk v3.0.2+incompatible/go.mod h1:T/Aws4fEfogEE9v+HPhhw+CntffsBHJ8nXQCwKr0/g8=
-github.com/aws/aws-sdk-go-v2 v1.32.5 h1:U8vdWJuY7ruAkzaOdD7guwJjD06YSKmnKCJs7s3IkIo=
+github.com/aws/aws-sdk-go-v2 v1.23.1 h1:qXaFsOOMA+HsZtX8WoCa+gJnbyW7qyFFBlPqvTSzbaI=
-github.com/aws/aws-sdk-go-v2 v1.32.5/go.mod h1:P5WJBrYqqbWVaOxgH0X/FYYD47/nooaPOZPlQdmiN2U=
+github.com/aws/aws-sdk-go-v2 v1.23.1/go.mod h1:i1XDttT4rnf6vxc9AuskLc6s7XBee8rlLilKlc03uAA=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.5.1 h1:ZY3108YtBNq96jNZTICHxN1gSBSbnvIdYwwqnvCV4Mc=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.5.1/go.mod h1:t8PYl/6LzdAqsU4/9tz28V/kU+asFePvpOMkdul0gEQ=
-github.com/aws/aws-sdk-go-v2/config v1.28.5 h1:Za41twdCXbuyyWv9LndXxZZv3QhTG1DinqlFsSuvtI0=
+github.com/aws/aws-sdk-go-v2/config v1.25.4 h1:r+X1x8QI6FEPdJDWCNBDZHyAcyFwSjHN8q8uuus+Axs=
-github.com/aws/aws-sdk-go-v2/config v1.28.5/go.mod h1:4VsPbHP8JdcdUDmbTVgNL/8w9SqOkM5jyY8ljIxLO3o=
+github.com/aws/aws-sdk-go-v2/config v1.25.4/go.mod h1:8GTjImECskr7D88P/Nn9uM4M4rLY9i77hLJZgkZEWV8=
-github.com/aws/aws-sdk-go-v2/credentials v1.17.46 h1:AU7RcriIo2lXjUfHFnFKYsLCwgbz1E7Mm95ieIRDNUg=
+github.com/aws/aws-sdk-go-v2/credentials v1.16.3 h1:8PeI2krzzjDJ5etmgaMiD1JswsrLrWvKKu/uBUtNy1g=
-github.com/aws/aws-sdk-go-v2/credentials v1.17.46/go.mod h1:1FmYyLGL08KQXQ6mcTlifyFXfJVCNJTVGuQP4m0d/UA=
+github.com/aws/aws-sdk-go-v2/credentials v1.16.3/go.mod h1:Kdh/okh+//vQ/AjEt81CjvkTo64+/zIE4OewP7RpfXk=
-github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.20 h1:sDSXIrlsFSFJtWKLQS4PUWRvrT580rrnuLydJrCQ/yA=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.14.5 h1:KehRNiVzIfAcj6gw98zotVbb/K67taJE0fkfgM6vzqU=
-github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.20/go.mod h1:WZ/c+w0ofps+/OUqMwWgnfrgzZH1DZO1RIkktICsqnY=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.14.5/go.mod h1:VhnExhw6uXy9QzetvpXDolo1/hjhx4u9qukBGkuUwjs=
-github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.24 h1:4usbeaes3yJnCFC7kfeyhkdkPtoRYPa/hTmCqMpKpLI=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.2.4 h1:LAm3Ycm9HJfbSCd5I+wqC2S9Ej7FPrgr5CQoOljJZcE=
-github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.24/go.mod h1:5CI1JemjVwde8m2WG3cz23qHKPOxbpkq0HaoreEgLIY=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.2.4/go.mod h1:xEhvbJcyUf/31yfGSQBe01fukXwXJ0gxDp7rLfymWE0=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.24 h1:N1zsICrQglfzaBnrfM0Ys00860C+QFwu6u/5+LomP+o=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.5.4 h1:4GV0kKZzUxiWxSVpn/9gwR0g21NF1Jsyduzo9rHgC/Q=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.24/go.mod h1:dCn9HbJ8+K31i8IQ8EWmWj0EiIk0+vKiHNMxTTYveAg=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.5.4/go.mod h1:dYvTNAggxDZy6y1AF7YDwXsPuHFy/VNEpEI/2dWK9IU=
-github.com/aws/aws-sdk-go-v2/internal/ini v1.8.1 h1:VaRN3TlFdd6KxX1x3ILT5ynH6HvKgqdiXoTxAF4HQcQ=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.7.1 h1:uR9lXYjdPX0xY+NhvaJ4dD8rpSRz5VY81ccIIoNG+lw=
-github.com/aws/aws-sdk-go-v2/internal/ini v1.8.1/go.mod h1:FbtygfRFze9usAadmnGJNc8KsP346kEe+y2/oyhGAGc=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.7.1/go.mod h1:6fQQgfuGmw8Al/3M2IgIllycxV7ZW7WCdVSqfBeUiCY=
 github.com/aws/aws-sdk-go-v2/internal/v4a v1.2.4 h1:40Q4X5ebZruRtknEZH/bg91sT5pR853F7/1X9QRbI54=
 github.com/aws/aws-sdk-go-v2/internal/v4a v1.2.4/go.mod h1:u77N7eEECzUv7F0xl2gcfK/vzc8wcjWobpy+DcrLJ5E=
-github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.1 h1:iXtILhvDxB6kPvEXgsDhGaZCSC6LQET5ZHSdJozeI0Y=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.10.1 h1:rpkF4n0CyFcrJUG/rNNohoTmhtWlFTRI4BsZOh9PvLs=
-github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.1/go.mod h1:9nu0fVANtYiAePIBh2/pFUSwtJ402hLnp854CNoDOeE=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.10.1/go.mod h1:l9ymW25HOqymeU2m1gbUQ3rUIsTwKs8gYHXkqDQUhiI=
 github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.2.4 h1:6DRKQc+9cChgzL5gplRGusI5dBGeiEod4m/pmGbcX48=
 github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.2.4/go.mod h1:s8ORvrW4g4v7IvYKIAoBg17w3GQ+XuwXDXYrQ5SkzU0=
-github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.5 h1:wtpJ4zcwrSbwhECWQoI/g6WM9zqCcSpHDJIWSbMLOu4=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.10.4 h1:rdovz3rEu0vZKbzoMYPTehp0E8veoE9AyfzqCr5Eeao=
-github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.5/go.mod h1:qu/W9HXQbbQ4+1+JcZp0ZNPV31ym537ZJN+fiS7Ti8E=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.10.4/go.mod h1:aYCGNjyUCUelhofxlZyj63srdxWUSsBSGg5l6MCuXuE=
 github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.16.4 h1:o3DcfCxGDIT20pTbVKVhp3vWXOj/VvgazNJvumWeYW0=
 github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.16.4/go.mod h1:Uy0KVOxuTK2ne+/PKQ+VvEeWmjMMksE17k/2RK/r5oM=
 github.com/aws/aws-sdk-go-v2/service/s3 v1.43.1 h1:1w11lfXOa8HoHoSlNtt4mqv/N3HmDOa+OnUH3Y9DHm8=
 github.com/aws/aws-sdk-go-v2/service/s3 v1.43.1/go.mod h1:dqJ5JBL0clzgHriH35Amx3LRFY6wNIPUX7QO/BerSBo=
-github.com/aws/aws-sdk-go-v2/service/sso v1.24.6 h1:3zu537oLmsPfDMyjnUS2g+F2vITgy5pB74tHI+JBNoM=
+github.com/aws/aws-sdk-go-v2/service/sso v1.17.3 h1:CdsSOGlFF3Pn+koXOIpTtvX7st0IuGsZ8kJqcWMlX54=
-github.com/aws/aws-sdk-go-v2/service/sso v1.24.6/go.mod h1:WJSZH2ZvepM6t6jwu4w/Z45Eoi75lPN7DcydSRtJg6Y=
+github.com/aws/aws-sdk-go-v2/service/sso v1.17.3/go.mod h1:oA6VjNsLll2eVuUoF2D+CMyORgNzPEW/3PyUdq6WQjI=
-github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.5 h1:K0OQAsDywb0ltlFrZm0JHPY3yZp/S9OaoLU33S7vPS8=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.20.1 h1:cbRqFTVnJV+KRpwFl76GJdIZJKKCdTPnjUZ7uWh3pIU=
-github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.5/go.mod h1:ORITg+fyuMoeiQFiVGoqB3OydVTLkClw/ljbblMq6Cc=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.20.1/go.mod h1:hHL974p5auvXlZPIjJTblXJpbkfK4klBczlsEaMCGVY=
-github.com/aws/aws-sdk-go-v2/service/sts v1.33.1 h1:6SZUVRQNvExYlMLbHdlKB48x0fLbc2iVROyaNEwBHbU=
+github.com/aws/aws-sdk-go-v2/service/sts v1.25.4 h1:yEvZ4neOQ/KpUqyR+X0ycUTW/kVRNR4nDZ38wStHGAA=
-github.com/aws/aws-sdk-go-v2/service/sts v1.33.1/go.mod h1:GqWyYCwLXnlUB1lOAXQyNSPqPLQJvmo8J0DWBzp9mtg=
+github.com/aws/aws-sdk-go-v2/service/sts v1.25.4/go.mod h1:feTnm2Tk/pJxdX+eooEsxvlvTWBvDm6CasRZ+JOs2IY=
-github.com/aws/smithy-go v1.22.1 h1:/HPHZQ0g7f4eUeK6HKglFz8uwVfZKgoI25rb/J+dnro=
+github.com/aws/smithy-go v1.17.0 h1:wWJD7LX6PBV6etBUwO0zElG0nWN9rUhp0WdYeHSHAaI=
-github.com/aws/smithy-go v1.22.1/go.mod h1:irrKGvNn1InZwb2d7fkIRNucdfwR8R+Ts3wxYa/cJHg=
+github.com/aws/smithy-go v1.17.0/go.mod h1:NukqUGpCZIILqqiV0NIjeFh24kd/FAa4beRb6nbIUPE=
 github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
@@ -319,10 +319,10 @@ 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.14-alpha.5 h1:VY9c5x515lTfmdhhPjMvR3BBRrRquAUCFsz7t7vbv7Y=
 github.com/openimsdk/gomake v0.0.14-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
-github.com/openimsdk/protocol v0.0.72-alpha.63 h1:IyPBibEvwBtTmD8DSrlqcekfEXe74k4+KeeHsgdhGh0=
+github.com/openimsdk/protocol v0.0.72-alpha.55 h1:9PPWPHvkFk3neBSbNr+IoOdKIFjxTvEqUfMK/TEq1+8=
-github.com/openimsdk/protocol v0.0.72-alpha.63/go.mod h1:Iet+piS/jaS+kWWyj6EEr36mk4ISzIRYjoMSVA4dq2M=
+github.com/openimsdk/protocol v0.0.72-alpha.55/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
-github.com/openimsdk/tools v0.0.50-alpha.50 h1:+naDlvHcqJDj2NsCGnQd1LLQOET5IRPbrtmWbM/o7JQ=
+github.com/openimsdk/tools v0.0.50-alpha.32 h1:JEsUFHFnaYg230TG+Ke3SUnaA2h44t4kABAzEdv5VZw=
-github.com/openimsdk/tools v0.0.50-alpha.50/go.mod h1:muCtxguNJv8lFwLei27UASu2Nvg4ERSeN0R4K5tivk0=
+github.com/openimsdk/tools v0.0.50-alpha.32/go.mod h1:r5U6RbxcR4xhKb2fhTmKGC9Yt5LcErHBVt3lhXQIHSo=
 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=
@@ -497,8 +497,8 @@ golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.29.0 h1:5ORfpBpCs4HzDYoodCDBbwHzdR5UrLBZ3sOnUJmFoHo=
 golang.org/x/net v0.29.0/go.mod h1:gLkgy8jTGERgjzMic6DS9+SP0ajcu6Xu3Orq/SpETg0=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
-golang.org/x/oauth2 v0.23.0 h1:PbgcYx2W7i4LvjJWEbf0ngHV6qJYr86PkAV3bXdLEbs=
+golang.org/x/oauth2 v0.21.0 h1:tsimM75w1tF/uws5rbeHzIWxEqElMehnc+iW793zsZs=
-golang.org/x/oauth2 v0.23.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
+golang.org/x/oauth2 v0.21.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -565,8 +565,8 @@ google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98
 google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo=
 google.golang.org/genproto v0.0.0-20240213162025-012b6fc9bca9 h1:9+tzLLstTlPTRyJTh+ah5wIMsBW5c4tQwGTN3thOW9Y=
 google.golang.org/genproto v0.0.0-20240213162025-012b6fc9bca9/go.mod h1:mqHbVIp48Muh7Ywss/AD6I5kNVKZMmAa/QEW58Gxp2s=
-google.golang.org/genproto/googleapis/api v0.0.0-20240903143218-8af14fe29dc1 h1:hjSy6tcFQZ171igDaN5QHOw2n6vx40juYbC/x67CEhc=
+google.golang.org/genproto/googleapis/api v0.0.0-20240604185151-ef581f913117 h1:+rdxYoE3E5htTEWIe15GlN6IfvbURM//Jt0mmkmm6ZU=
-google.golang.org/genproto/googleapis/api v0.0.0-20240903143218-8af14fe29dc1/go.mod h1:qpvKtACPCQhAdu3PyQgV4l3LMXZEtft7y8QcarRsp9I=
+google.golang.org/genproto/googleapis/api v0.0.0-20240604185151-ef581f913117/go.mod h1:OimBR/bc1wPO9iV4NC2bpyjy3VnAwZh5EBPQdtaE5oo=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 h1:pPJltXNxVzT4pK9yD8vR9X75DaWYYmLGMsEvBfFQZzQ=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1/go.mod h1:UqMtugtsSgubUsoxbuAoiCXvqvErP7Gf0so0mK9tHxU=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
@@ -574,8 +574,8 @@ google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyac
 google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY=
 google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
 google.golang.org/grpc v1.33.2/go.mod h1:JMHMWHQWaTccqQQlmk3MJZS+GWXOdAesneDmEnv2fbc=
-google.golang.org/grpc v1.68.0 h1:aHQeeJbo8zAkAa3pRzrVjZlbz6uSfeOXlJNQM0RAbz0=
+google.golang.org/grpc v1.66.2 h1:3QdXkuq3Bkh7w+ywLdLvM56cmGvQHUMZpiCzt6Rqaoo=
-google.golang.org/grpc v1.68.0/go.mod h1:fmSPC5AsjSBCK54MyHRx48kpOti1/jRfOlwEWywNjWA=
+google.golang.org/grpc v1.66.2/go.mod h1:s3/l6xSSCURdVfAnL+TqCNMyTDAGN6+lZeVxnZR128Y=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
 google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
@@ -585,8 +585,8 @@ google.golang.org/protobuf v1.22.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2
 google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
 google.golang.org/protobuf v1.23.1-0.20200526195155-81db48ad09cc/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
 google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c=
-google.golang.org/protobuf v1.35.1 h1:m3LfL6/Ca+fqnjnlqQXNpFPABW1UD7mjh8KO2mKFytA=
+google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
-google.golang.org/protobuf v1.35.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
+google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=

internal/api/msg.go

@@ -173,6 +173,8 @@ func (m *MessageApi) getSendMsgReq(c *gin.Context, req apistruct.SendMsg) (sendM
 		data = apistruct.AtElem{}
 	case constant.Custom:
 		data = apistruct.CustomElem{}
+	case constant.Stream:
+		data = apistruct.StreamMsgElem{}
 	case constant.OANotification:
 		data = apistruct.OANotificationElem{}
 		req.SessionType = constant.NotificationChatType
@@ -373,3 +375,11 @@ func (m *MessageApi) SearchMsg(c *gin.Context) {
 func (m *MessageApi) GetServerTime(c *gin.Context) {
 	a2r.Call(msg.MsgClient.GetServerTime, m.Client, c)
 }
+
+func (m *MessageApi) GetStreamMsg(c *gin.Context) {
+	a2r.Call(msg.MsgClient.GetStreamMsg, m.Client, c)
+}
+
+func (m *MessageApi) AppendStreamMsg(c *gin.Context) {
+	a2r.Call(msg.MsgClient.AppendStreamMsg, m.Client, c)
+}

internal/api/router.go

@@ -222,6 +222,8 @@ func newGinRouter(disCov discovery.SvcDiscoveryRegistry, config *Config) *gin.En
 		msgGroup.POST("/batch_send_msg", m.BatchSendMsg)
 		msgGroup.POST("/check_msg_is_send_success", m.CheckMsgIsSendSuccess)
 		msgGroup.POST("/get_server_time", m.GetServerTime)
+		msgGroup.POST("/get_stream_msg", m.GetStreamMsg)
+		msgGroup.POST("/append_stream_msg", m.AppendStreamMsg)
 	}
 	// Conversation
 	conversationGroup := r.Group("/conversation")

internal/msggateway/client.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/openimsdk/tools/mw"
 	"runtime/debug"
 	"sync"
 	"sync/atomic"
@@ -377,7 +378,7 @@ func (c *Client) activeHeartbeat(ctx context.Context) {
 		go func() {
 			defer func() {
 				if r := recover(); r != nil {
-					log.ZPanic(ctx, "activeHeartbeat Panic", r)
+					mw.PanicStackToLog(ctx, r)
 				}
 			}()
 			log.ZDebug(ctx, "server initiative send heartbeat start.")

internal/msggateway/hub_server.go

@@ -56,7 +56,6 @@ func (s *Server) Start(ctx context.Context, index int, conf *Config) error {
 }
 
 type Server struct {
-	msggateway.UnimplementedMsgGatewayServer
 	rpcPort        int
 	LongConnServer LongConnServer
 	config         *Config

internal/msgtransfer/init.go

@@ -138,7 +138,7 @@ func (m *MsgTransfer) Start(index int, config *Config) error {
 		go func() {
 			defer func() {
 				if r := recover(); r != nil {
-					log.ZPanic(m.ctx, "MsgTransfer Start Panic", r)
+					mw.PanicStackToLog(m.ctx, r)
 				}
 			}()
 			prometheusPort, err := datautil.GetElemByIndex(config.MsgTransfer.Prometheus.Ports, index)

internal/msgtransfer/online_history_msg_handler.go

@@ -18,13 +18,13 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
+	"github.com/openimsdk/tools/mw"
 	"strconv"
 	"strings"
 	"sync"
 	"time"
 
-	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
-
 	"github.com/IBM/sarama"
 	"github.com/go-redis/redis"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
@@ -349,7 +349,7 @@ func (och *OnlineHistoryRedisConsumerHandler) handleNotification(ctx context.Con
 func (och *OnlineHistoryRedisConsumerHandler) HandleUserHasReadSeqMessages(ctx context.Context) {
 	defer func() {
 		if r := recover(); r != nil {
-			log.ZPanic(ctx, "HandleUserHasReadSeqMessages Panic", r)
+			mw.PanicStackToLog(ctx, r)
 		}
 	}()
 

internal/push/push.go

@@ -14,7 +14,6 @@ import (
 )
 
 type pushServer struct {
-	pbpush.UnimplementedPushMsgServiceServer
 	database      controller.PushDatabase
 	disCov        discovery.SvcDiscoveryRegistry
 	offlinePusher offlinepush.OfflinePusher

internal/push/push_handler.go

@@ -414,7 +414,6 @@ func (c *ConsumerHandler) DeleteMemberAndSetConversationSeq(ctx context.Context,
 	if err != nil {
 		return err
 	}
-	
 	return c.conversationRpcClient.SetConversationMaxSeq(ctx, userIDs, conversationID, maxSeq)
 }
 

internal/rpc/auth/auth.go

@@ -40,7 +40,6 @@ import (
 )
 
 type authServer struct {
-	pbauth.UnimplementedAuthServer
 	authDatabase   controller.AuthDatabase
 	userRpcClient  *rpcclient.UserRpcClient
 	RegisterCenter discovery.SvcDiscoveryRegistry
@@ -197,7 +196,7 @@ func (s *authServer) forceKickOff(ctx context.Context, userID string, platformID
 	}
 
 	m, err := s.authDatabase.GetTokensWithoutError(ctx, userID, int(platformID))
-	if err != nil && !errors.Is(err, redis.Nil) {
+	if err != nil && errors.Is(err, redis.Nil) {
 		return err
 	}
 	for k := range m {
@@ -215,7 +214,7 @@ func (s *authServer) forceKickOff(ctx context.Context, userID string, platformID
 
 func (s *authServer) InvalidateToken(ctx context.Context, req *pbauth.InvalidateTokenReq) (*pbauth.InvalidateTokenResp, error) {
 	m, err := s.authDatabase.GetTokensWithoutError(ctx, req.UserID, int(req.PlatformID))
-	if err != nil && !errors.Is(err, redis.Nil) {
+	if err != nil && errors.Is(err, redis.Nil) {
 		return nil, err
 	}
 	if m == nil {

internal/rpc/conversation/conversation.go

@@ -16,8 +16,6 @@ package conversation
 
 import (
 	"context"
-	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
-	pbmsg "github.com/openimsdk/protocol/msg"
 	"sort"
 	"time"
 
@@ -45,7 +43,6 @@ import (
 )
 
 type conversationServer struct {
-	pbconversation.UnimplementedConversationServer
 	msgRpcClient         *rpcclient.MessageRpcClient
 	user                 *rpcclient.UserRpcClient
 	groupRpcClient       *rpcclient.GroupRpcClient
@@ -435,38 +432,22 @@ func (c *conversationServer) CreateGroupChatConversations(ctx context.Context, r
 	if err != nil {
 		return nil, err
 	}
-	conversationID := msgprocessor.GetConversationIDBySessionType(constant.ReadGroupChatType, req.GroupID)
-	if _, err := c.msgRpcClient.Client.SetUserConversationMaxSeq(ctx, &pbmsg.SetUserConversationMaxSeqReq{ConversationID: conversationID, OwnerUserID: req.UserIDs, MaxSeq: 0}); err != nil {
-		return nil, err
-	}
 	return &pbconversation.CreateGroupChatConversationsResp{}, nil
 }
 
 func (c *conversationServer) SetConversationMaxSeq(ctx context.Context, req *pbconversation.SetConversationMaxSeqReq) (*pbconversation.SetConversationMaxSeqResp, error) {
-	if _, err := c.msgRpcClient.Client.SetUserConversationMaxSeq(ctx, &pbmsg.SetUserConversationMaxSeqReq{ConversationID: req.ConversationID, OwnerUserID: req.OwnerUserID, MaxSeq: req.MaxSeq}); err != nil {
-		return nil, err
-	}
 	if err := c.conversationDatabase.UpdateUsersConversationField(ctx, req.OwnerUserID, req.ConversationID,
 		map[string]any{"max_seq": req.MaxSeq}); err != nil {
 		return nil, err
 	}
-	for _, userID := range req.OwnerUserID {
-		c.conversationNotificationSender.ConversationChangeNotification(ctx, userID, []string{req.ConversationID})
-	}
 	return &pbconversation.SetConversationMaxSeqResp{}, nil
 }
 
 func (c *conversationServer) SetConversationMinSeq(ctx context.Context, req *pbconversation.SetConversationMinSeqReq) (*pbconversation.SetConversationMinSeqResp, error) {
-	if _, err := c.msgRpcClient.Client.SetUserConversationMinSeq(ctx, &pbmsg.SetUserConversationMinSeqReq{ConversationID: req.ConversationID, OwnerUserID: req.OwnerUserID, MinSeq: req.MinSeq}); err != nil {
-		return nil, err
-	}
 	if err := c.conversationDatabase.UpdateUsersConversationField(ctx, req.OwnerUserID, req.ConversationID,
 		map[string]any{"min_seq": req.MinSeq}); err != nil {
 		return nil, err
 	}
-	for _, userID := range req.OwnerUserID {
-		c.conversationNotificationSender.ConversationChangeNotification(ctx, userID, []string{req.ConversationID})
-	}
 	return &pbconversation.SetConversationMinSeqResp{}, nil
 }
 
@@ -688,7 +669,7 @@ func (c *conversationServer) GetOwnerConversation(ctx context.Context, req *pbco
 	}, nil
 }
 
-func (c *conversationServer) GetConversationsNeedClearMsg(ctx context.Context, _ *pbconversation.GetConversationsNeedClearMsgReq) (*pbconversation.GetConversationsNeedClearMsgResp, error) {
+func (c *conversationServer) GetConversationsNeedDestructMsgs(ctx context.Context, _ *pbconversation.GetConversationsNeedDestructMsgsReq) (*pbconversation.GetConversationsNeedDestructMsgsResp, error) {
 	num, err := c.conversationDatabase.GetAllConversationIDsNumber(ctx)
 	if err != nil {
 		log.ZError(ctx, "GetAllConversationIDsNumber failed", err)
@@ -712,7 +693,7 @@ func (c *conversationServer) GetConversationsNeedClearMsg(ctx context.Context, _
 
 		conversationIDs, err := c.conversationDatabase.PageConversationIDs(ctx, pagination)
 		if err != nil {
-			log.ZError(ctx, "PageConversationIDs failed", err, "pageNumber", pageNumber)
+			// log.ZError(ctx, "PageConversationIDs failed", err, "pageNumber", pageNumber)
 			continue
 		}
 
@@ -735,7 +716,7 @@ func (c *conversationServer) GetConversationsNeedClearMsg(ctx context.Context, _
 		}
 	}
 
-	return &pbconversation.GetConversationsNeedClearMsgResp{Conversations: convert.ConversationsDB2Pb(temp)}, nil
+	return &pbconversation.GetConversationsNeedDestructMsgsResp{Conversations: convert.ConversationsDB2Pb(temp)}, nil
 }
 
 func (c *conversationServer) GetNotNotifyConversationIDs(ctx context.Context, req *pbconversation.GetNotNotifyConversationIDsReq) (*pbconversation.GetNotNotifyConversationIDsResp, error) {

internal/rpc/group/db_map.go

@@ -110,7 +110,7 @@ func UpdateGroupMemberMap(req *pbgroup.SetGroupMemberInfo) map[string]any {
 		m["nickname"] = req.Nickname.Value
 	}
 	if req.FaceURL != nil {
-		m["face_url"] = req.FaceURL.Value
+		m["user_group_face_url"] = req.FaceURL.Value
 	}
 	if req.RoleLevel != nil {
 		m["role_level"] = req.RoleLevel.Value

internal/rpc/group/group.go

@@ -57,7 +57,6 @@ import (
 )
 
 type groupServer struct {
-	pbgroup.UnimplementedGroupServer
 	db                    controller.GroupDatabase
 	user                  rpcclient.UserRpcClient
 	notification          *GroupNotificationSender

internal/rpc/msg/as_read.go

@@ -137,7 +137,7 @@ func (m *msgServer) MarkConversationAsRead(ctx context.Context, req *msg.MarkCon
 		return nil, err
 	}
 	hasReadSeq, err := m.MsgDatabase.GetHasReadSeq(ctx, req.UserID, req.ConversationID)
-	if err != nil && !errors.Is(err, redis.Nil) {
+	if err != nil && errors.Is(err, redis.Nil) {
 		return nil, err
 	}
 	var seqs []int64

internal/rpc/msg/clear.go

@@ -12,14 +12,13 @@ import (
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/mcontext"
-	"github.com/openimsdk/tools/utils/datautil"
 	"github.com/openimsdk/tools/utils/idutil"
 	"github.com/openimsdk/tools/utils/stringutil"
 	"golang.org/x/sync/errgroup"
 )
 
 // hard delete in Database.
-func (m *msgServer) DestructMsgs(ctx context.Context, req *msg.DestructMsgsReq) (_ *msg.DestructMsgsResp, err error) {
+func (m *msgServer) ClearMsg(ctx context.Context, req *msg.ClearMsgReq) (_ *msg.ClearMsgResp, err error) {
 	if err := authverify.CheckAdmin(ctx, m.config.Share.IMAdminUserID); err != nil {
 		return nil, err
 	}
@@ -27,19 +26,18 @@ func (m *msgServer) DestructMsgs(ctx context.Context, req *msg.DestructMsgsReq)
 		return nil, errs.ErrArgs.WrapMsg("request millisecond timestamp error")
 	}
 	var (
-		docNum   int
+		docNum int
-		msgNum   int
+		msgNum int
-		start    = time.Now()
+		start  = time.Now()
-		getLimit = 5000
 	)
 
-	destructMsg := func(ctx context.Context) (bool, error) {
+	clearMsg := func(ctx context.Context) (bool, error) {
 		docIDs, err := m.MsgDatabase.GetDocIDs(ctx)
 		if err != nil {
 			return false, err
 		}
 
-		msgs, err := m.MsgDatabase.GetBeforeMsg(ctx, req.Timestamp, docIDs, getLimit)
+		msgs, err := m.MsgDatabase.GetBeforeMsg(ctx, req.Timestamp, docIDs, 5000)
 		if err != nil {
 			return false, err
 		}
@@ -63,7 +61,7 @@ func (m *msgServer) DestructMsgs(ctx context.Context, req *msg.DestructMsgsReq)
 		return true, nil
 	}
 
-	_, err = destructMsg(ctx)
+	_, err = clearMsg(ctx)
 	if err != nil {
 		log.ZError(ctx, "clear msg failed", err, "docNum", docNum, "msgNum", msgNum, "cost", time.Since(start))
 		return nil, err
@@ -71,11 +69,11 @@ func (m *msgServer) DestructMsgs(ctx context.Context, req *msg.DestructMsgsReq)
 
 	log.ZDebug(ctx, "clearing message", "docNum", docNum, "msgNum", msgNum, "cost", time.Since(start))
 
-	return &msg.DestructMsgsResp{}, nil
+	return &msg.ClearMsgResp{}, nil
 }
 
-// soft delete for user self
+// soft delete for self
-func (m *msgServer) ClearMsg(ctx context.Context, req *msg.ClearMsgReq) (_ *msg.ClearMsgResp, err error) {
+func (m *msgServer) DestructMsgs(ctx context.Context, req *msg.DestructMsgsReq) (_ *msg.DestructMsgsResp, err error) {
 	temp := convert.ConversationsPb2DB(req.Conversations)
 
 	batchNum := 100
@@ -95,31 +93,22 @@ func (m *msgServer) ClearMsg(ctx context.Context, req *msg.ClearMsgReq) (_ *msg.
 					"msgDestructTime", conversation.MsgDestructTime,
 					"lastMsgDestructTime", conversation.LatestMsgDestructTime)
 
-				seqs, err := m.MsgDatabase.ClearUserMsgs(handleCtx, conversation.OwnerUserID, conversation.ConversationID, conversation.MsgDestructTime, conversation.LatestMsgDestructTime)
+				seqs, err := m.MsgDatabase.UserMsgsDestruct(handleCtx, conversation.OwnerUserID, conversation.ConversationID, conversation.MsgDestructTime, conversation.LatestMsgDestructTime)
 				if err != nil {
 					log.ZError(handleCtx, "user msg destruct failed", err, "conversationID", conversation.ConversationID, "ownerUserID", conversation.OwnerUserID)
 					continue
 				}
 
 				if len(seqs) > 0 {
-					minseq := datautil.Max(seqs...)
-
-					// update
 					if err := m.Conversation.UpdateConversation(handleCtx,
 						&pbconversation.UpdateConversationReq{
 							UserIDs:               []string{conversation.OwnerUserID},
 							ConversationID:        conversation.ConversationID,
-							LatestMsgDestructTime: wrapperspb.Int64(time.Now().UnixMilli()),
+							LatestMsgDestructTime: wrapperspb.Int64(time.Now().UnixMilli())}); err != nil {
-							MinSeq:                wrapperspb.Int64(minseq),
-						}); err != nil {
 						log.ZError(handleCtx, "updateUsersConversationField failed", err, "conversationID", conversation.ConversationID, "ownerUserID", conversation.OwnerUserID)
 						continue
 					}
 
-					if err := m.Conversation.SetConversationMinSeq(handleCtx, []string{conversation.OwnerUserID}, conversation.ConversationID, minseq); err != nil {
-						return err
-					}
-
 					// if you need Notify SDK client userseq is update.
 					// m.msgNotificationSender.UserDeleteMsgsNotification(handleCtx, conversation.OwnerUserID, conversation.ConversationID, seqs)
 				}

internal/rpc/msg/delete.go

@@ -138,16 +138,9 @@ func (m *msgServer) clearConversation(ctx context.Context, conversationIDs []str
 	}
 	isSyncSelf, isSyncOther := m.validateDeleteSyncOpt(deleteSyncOpt)
 	if !isSyncOther {
-		setSeqs := m.getMinSeqs(maxSeqs)
+		if err := m.MsgDatabase.SetUserConversationsMinSeqs(ctx, userID, m.getMinSeqs(maxSeqs)); err != nil {
-		if err := m.MsgDatabase.SetUserConversationsMinSeqs(ctx, userID, setSeqs); err != nil {
 			return err
 		}
-		ownerUserIDs := []string{userID}
-		for conversationID, seq := range setSeqs {
-			if err := m.Conversation.SetConversationMinSeq(ctx, ownerUserIDs, conversationID, seq); err != nil {
-				return err
-			}
-		}
 		// notification 2 self
 		if isSyncSelf {
 			tips := &sdkws.ClearConversationTips{UserID: userID, ConversationIDs: existConversationIDs}

internal/rpc/msg/notification.go

@@ -48,3 +48,7 @@ func (m *MsgNotificationSender) MarkAsReadNotification(ctx context.Context, conv
 	}
 	m.NotificationWithSessionType(ctx, sendID, recvID, constant.HasReadReceipt, sessionType, tips)
 }
+
+func (m *MsgNotificationSender) StreamMsgNotification(ctx context.Context, sendID string, recvID string, sessionType int32, tips *sdkws.StreamMsgTips) {
+	m.NotificationWithSessionType(ctx, sendID, recvID, constant.StreamMsgNotification, sessionType, tips)
+}

internal/rpc/msg/send.go

@@ -16,6 +16,7 @@ package msg
 
 import (
 	"context"
+	"github.com/openimsdk/tools/mw"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
@@ -34,6 +35,11 @@ import (
 func (m *msgServer) SendMsg(ctx context.Context, req *pbmsg.SendMsgReq) (*pbmsg.SendMsgResp, error) {
 	if req.MsgData != nil {
 		m.encapsulateMsgData(req.MsgData)
+		if req.MsgData.ContentType == constant.Stream {
+			if err := m.handlerStreamMsg(ctx, req.MsgData); err != nil {
+				return nil, err
+			}
+		}
 		switch req.MsgData.SessionType {
 		case constant.SingleChatType:
 			return m.sendMsgSingleChat(ctx, req)
@@ -83,7 +89,7 @@ func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgDa
 
 	defer func() {
 		if r := recover(); r != nil {
-			log.ZPanic(nctx, "setConversationAtInfo Panic", r)
+			mw.PanicStackToLog(nctx, r)
 		}
 	}()
 

internal/rpc/msg/seq.go

@@ -84,21 +84,3 @@ func (m *msgServer) GetActiveConversation(ctx context.Context, req *pbmsg.GetAct
 	}
 	return &pbmsg.GetActiveConversationResp{Conversations: conversations}, nil
 }
-
-func (m *msgServer) SetUserConversationMaxSeq(ctx context.Context, req *pbmsg.SetUserConversationMaxSeqReq) (*pbmsg.SetUserConversationMaxSeqResp, error) {
-	for _, userID := range req.OwnerUserID {
-		if err := m.MsgDatabase.SetUserConversationsMaxSeq(ctx, req.ConversationID, userID, req.MaxSeq); err != nil {
-			return nil, err
-		}
-	}
-	return &pbmsg.SetUserConversationMaxSeqResp{}, nil
-}
-
-func (m *msgServer) SetUserConversationMinSeq(ctx context.Context, req *pbmsg.SetUserConversationMinSeqReq) (*pbmsg.SetUserConversationMinSeqResp, error) {
-	for _, userID := range req.OwnerUserID {
-		if err := m.MsgDatabase.SetUserConversationsMinSeq(ctx, req.ConversationID, userID, req.MinSeq); err != nil {
-			return nil, err
-		}
-	}
-	return &pbmsg.SetUserConversationMinSeqResp{}, nil
-}

internal/rpc/msg/server.go

@@ -16,6 +16,7 @@ package msg
 
 import (
 	"context"
+
 	"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/mgo"
@@ -41,8 +42,9 @@ type (
 
 	// MsgServer encapsulates dependencies required for message handling.
 	msgServer struct {
-		RegisterCenter         discovery.SvcDiscoveryRegistry   // Service discovery registry for service registration.
+		RegisterCenter         discovery.SvcDiscoveryRegistry // Service discovery registry for service registration.
-		MsgDatabase            controller.CommonMsgDatabase     // Interface for message database operations.
+		MsgDatabase            controller.CommonMsgDatabase   // Interface for message database operations.
+		StreamMsgDatabase      controller.StreamMsgDatabase
 		Conversation           *rpcclient.ConversationRpcClient // RPC client for conversation service.
 		UserLocalCache         *rpccache.UserLocalCache         // Local cache for user data.
 		FriendLocalCache       *rpccache.FriendLocalCache       // Local cache for friend data.
@@ -53,7 +55,6 @@ type (
 		msgNotificationSender  *MsgNotificationSender           // RPC client for sending msg notifications.
 		config                 *Config                          // Global configuration settings.
 		webhookClient          *webhook.Client
-		msg.UnimplementedMsgServer
 	}
 
 	Config struct {
@@ -101,6 +102,10 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 	if err != nil {
 		return err
 	}
+	streamMsg, err := mgo.NewStreamMsgMongo(mgocli.GetDB())
+	if err != nil {
+		return err
+	}
 	seqUserCache := redis.NewSeqUserCacheRedis(rdb, seqUser)
 	msgDatabase, err := controller.NewCommonMsgDatabase(msgDocModel, msgModel, seqUserCache, seqConversationCache, &config.KafkaConfig)
 	if err != nil {
@@ -109,6 +114,7 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 	s := &msgServer{
 		Conversation:           &conversationClient,
 		MsgDatabase:            msgDatabase,
+		StreamMsgDatabase:      controller.NewStreamMsgDatabase(streamMsg),
 		RegisterCenter:         client,
 		UserLocalCache:         rpccache.NewUserLocalCache(userRpcClient, &config.LocalCacheConfig, rdb),
 		GroupLocalCache:        rpccache.NewGroupLocalCache(groupRpcClient, &config.LocalCacheConfig, rdb),
@@ -139,11 +145,3 @@ func (m *msgServer) conversationAndGetRecvID(conversation *conversation.Conversa
 	}
 	return ""
 }
-
-func (m *msgServer) AppendStreamMsg(ctx context.Context, req *msg.AppendStreamMsgReq) (*msg.AppendStreamMsgResp, error) {
-	return nil, nil
-}
-
-func (m *msgServer) GetStreamMsg(ctx context.Context, req *msg.GetStreamMsgReq) (*msg.GetStreamMsgResp, error) {
-	return nil, nil
-}

internal/rpc/msg/stream_msg.go

@@ -0,0 +1,114 @@
+package msg
+
+import (
+	"context"
+	"fmt"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
+	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
+	"github.com/openimsdk/protocol/constant"
+	"github.com/openimsdk/protocol/msg"
+	"github.com/openimsdk/protocol/sdkws"
+	"github.com/openimsdk/tools/errs"
+	"time"
+)
+
+const StreamDeadlineTime = time.Second * 60 * 10
+
+func (m *msgServer) handlerStreamMsg(ctx context.Context, msgData *sdkws.MsgData) error {
+	now := time.Now()
+	val := &model.StreamMsg{
+		ClientMsgID:    msgData.ClientMsgID,
+		ConversationID: msgprocessor.GetConversationIDByMsg(msgData),
+		UserID:         msgData.SendID,
+		CreateTime:     now,
+		DeadlineTime:   now.Add(StreamDeadlineTime),
+	}
+	return m.StreamMsgDatabase.CreateStreamMsg(ctx, val)
+}
+
+func (m *msgServer) getStreamMsg(ctx context.Context, clientMsgID string) (*model.StreamMsg, error) {
+	res, err := m.StreamMsgDatabase.GetStreamMsg(ctx, clientMsgID)
+	if err != nil {
+		return nil, err
+	}
+	now := time.Now()
+	if !res.End && res.DeadlineTime.Before(now) {
+		res.End = true
+		res.DeadlineTime = now
+		_ = m.StreamMsgDatabase.AppendStreamMsg(ctx, res.ClientMsgID, 0, nil, true, now)
+	}
+	return res, nil
+}
+
+func (m *msgServer) AppendStreamMsg(ctx context.Context, req *msg.AppendStreamMsgReq) (*msg.AppendStreamMsgResp, error) {
+	res, err := m.getStreamMsg(ctx, req.ClientMsgID)
+	if err != nil {
+		return nil, err
+	}
+	if res.End {
+		return nil, errs.ErrNoPermission.WrapMsg("stream msg is end")
+	}
+	if len(res.Packets) < int(req.StartIndex) {
+		return nil, errs.ErrNoPermission.WrapMsg("start index is invalid")
+	}
+	if val := len(res.Packets) - int(req.StartIndex); val > 0 {
+		exist := res.Packets[int(req.StartIndex):]
+		for i, s := range exist {
+			if len(req.Packets) == 0 {
+				break
+			}
+			if s != req.Packets[i] {
+				return nil, errs.ErrNoPermission.WrapMsg(fmt.Sprintf("packet %d has been written and is inconsistent", i))
+			}
+			req.StartIndex++
+			req.Packets = req.Packets[1:]
+		}
+	}
+	if len(req.Packets) == 0 && res.End == req.End {
+		return &msg.AppendStreamMsgResp{}, nil
+	}
+	deadlineTime := time.Now().Add(StreamDeadlineTime)
+	if err := m.StreamMsgDatabase.AppendStreamMsg(ctx, req.ClientMsgID, int(req.StartIndex), req.Packets, req.End, deadlineTime); err != nil {
+		return nil, err
+	}
+	conversation, err := m.Conversation.GetConversation(ctx, res.UserID, res.ConversationID)
+	if err != nil {
+		return nil, err
+	}
+	tips := &sdkws.StreamMsgTips{
+		ConversationID: res.ConversationID,
+		ClientMsgID:    res.ClientMsgID,
+		StartIndex:     req.StartIndex,
+		Packets:        req.Packets,
+		End:            req.End,
+	}
+	var (
+		recvID      string
+		sessionType int32
+	)
+	if conversation.GroupID == "" {
+		sessionType = constant.SingleChatType
+		recvID = conversation.UserID
+	} else {
+		sessionType = constant.ReadGroupChatType
+		recvID = conversation.GroupID
+	}
+	m.msgNotificationSender.StreamMsgNotification(ctx, res.UserID, recvID, sessionType, tips)
+	return &msg.AppendStreamMsgResp{}, nil
+}
+
+func (m *msgServer) GetStreamMsg(ctx context.Context, req *msg.GetStreamMsgReq) (*msg.GetStreamMsgResp, error) {
+	res, err := m.getStreamMsg(ctx, req.ClientMsgID)
+	if err != nil {
+		return nil, err
+	}
+	return &msg.GetStreamMsgResp{
+		ClientMsgID:    res.ClientMsgID,
+		ConversationID: res.ConversationID,
+		UserID:         res.UserID,
+		Packets:        res.Packets,
+		End:            res.End,
+		CreateTime:     res.CreateTime.UnixMilli(),
+		DeadlineTime:   res.DeadlineTime.UnixMilli(),
+	}, nil
+}

internal/rpc/msg/sync_msg.go

@@ -92,7 +92,7 @@ func (m *msgServer) GetSeqMessage(ctx context.Context, req *msg.GetSeqMessageReq
 		NotificationMsgs: make(map[string]*sdkws.PullMsgs),
 	}
 	for _, conv := range req.Conversations {
-		isEnd, endSeq, msgs, err := m.MsgDatabase.GetMessagesBySeqWithBounds(ctx, req.UserID, conv.ConversationID, conv.Seqs, req.GetOrder())
+		_, _, msgs, err := m.MsgDatabase.GetMsgBySeqs(ctx, req.UserID, conv.ConversationID, conv.Seqs)
 		if err != nil {
 			return nil, err
 		}
@@ -111,8 +111,6 @@ func (m *msgServer) GetSeqMessage(ctx context.Context, req *msg.GetSeqMessageReq
 			}
 		}
 		pullMsgs.Msgs = append(pullMsgs.Msgs, msgs...)
-		pullMsgs.IsEnd = isEnd
-		pullMsgs.EndSeq = endSeq
 	}
 	return resp, nil
 }

internal/rpc/relation/friend.go

@@ -43,7 +43,6 @@ import (
 )
 
 type friendServer struct {
-	relation.UnimplementedFriendServer
 	db                    controller.FriendDatabase
 	blackDatabase         controller.BlackDatabase
 	userRpcClient         *rpcclient.UserRpcClient

internal/rpc/third/log.go

@@ -50,14 +50,14 @@ func (t *thirdServer) UploadLogs(ctx context.Context, req *third.UploadLogsReq)
 	platform := constant.PlatformID2Name[int(req.Platform)]
 	for _, fileURL := range req.FileURLs {
 		log := relationtb.Log{
-			Platform:     platform,
+			Platform:   platform,
-			UserID:       userID,
+			UserID:     userID,
-			CreateTime:   time.Now(),
+			CreateTime: time.Now(),
-			Url:          fileURL.URL,
+			Url:        fileURL.URL,
-			FileName:     fileURL.Filename,
+			FileName:   fileURL.Filename,
-			AppFramework: req.AppFramework,
+			SystemType: req.SystemType,
-			Version:      req.Version,
+			Version:    req.Version,
-			Ex:           req.Ex,
+			Ex:         req.Ex,
 		}
 		for i := 0; i < 20; i++ {
 			id := genLogID()

internal/rpc/third/s3.go

@@ -290,85 +290,48 @@ func (t *thirdServer) apiAddress(prefix, name string) string {
 func (t *thirdServer) DeleteOutdatedData(ctx context.Context, req *third.DeleteOutdatedDataReq) (*third.DeleteOutdatedDataResp, error) {
 	var conf config.Third
 	expireTime := time.UnixMilli(req.ExpireTime)
-
+	var deltotal int
 	findPagination := &sdkws.RequestPagination{
 		PageNumber: 1,
-		ShowNumber: 500,
+		ShowNumber: 1000,
 	}
-
+	for {
-	// Find all expired data in S3 database
+		total, models, err := t.s3dataBase.FindByExpires(ctx, expireTime, findPagination)
-	total, models, err := t.s3dataBase.FindNeedDeleteObjectByDB(ctx, expireTime, req.ObjectGroup, findPagination)
-	if err != nil && errs.Unwrap(err) != mongo.ErrNoDocuments {
-		return nil, errs.Wrap(err)
-	}
-
-	if total == 0 {
-		log.ZDebug(ctx, "Not have OutdatedData", "delete Total", total)
-		return &third.DeleteOutdatedDataResp{Count: int32(total)}, nil
-	}
-
-	needDelObjectKeys := make([]string, len(models))
-	for _, model := range models {
-		needDelObjectKeys = append(needDelObjectKeys, model.Key)
-	}
-
-	// Remove duplicate keys, have the same key use in different models
-	needDelObjectKeys = datautil.Distinct(needDelObjectKeys)
-
-	for _, key := range needDelObjectKeys {
-		// Find all models by key
-		keyModels, err := t.s3dataBase.FindModelsByKey(ctx, key)
 		if err != nil && errs.Unwrap(err) != mongo.ErrNoDocuments {
 			return nil, errs.Wrap(err)
 		}
-
+		needDelObjectKeys := make([]string, 0)
-		// check keyModels, if all keyModels.
+		for _, model := range models {
-		needDelKey := true // Default can delete
+			needDelObjectKeys = append(needDelObjectKeys, model.Key)
-		for _, keymodel := range keyModels {
-			// If group is empty or CreateTime is after expireTime, can't delete this key
-			if keymodel.Group == "" || keymodel.CreateTime.After(expireTime) {
-				needDelKey = false
-				break
-			}
 		}
 
-		// If this object is not referenced by not expire data, delete it
+		needDelObjectKeys = datautil.Distinct(needDelObjectKeys)
-		if needDelKey && t.minio != nil {
+		for _, key := range needDelObjectKeys {
-			// If have a thumbnail, delete it
+			count, err := t.s3dataBase.FindNotDelByS3(ctx, key, expireTime)
-			thumbnailKey, _ := t.getMinioImageThumbnailKey(ctx, key)
+			if err != nil && errs.Unwrap(err) != mongo.ErrNoDocuments {
-			if thumbnailKey != "" {
+				return nil, errs.Wrap(err)
-				err := t.s3dataBase.DeleteObject(ctx, thumbnailKey)
-				if err != nil {
-					log.ZWarn(ctx, "Delete thumbnail object is error:", errs.Wrap(err), "thumbnailKey", thumbnailKey)
-				}
 			}
+			if int(count) < 1 && t.minio != nil {
+				thumbnailKey, _ := t.getMinioImageThumbnailKey(ctx, key)
 
-			// Delete object
+				t.s3dataBase.DeleteObject(ctx, thumbnailKey)
-			err = t.s3dataBase.DeleteObject(ctx, key)
+				t.s3dataBase.DelS3Key(ctx, conf.Object.Enable, needDelObjectKeys...)
+				t.s3dataBase.DeleteObject(ctx, key)
+			}
+		}
+		for _, model := range models {
+			err := t.s3dataBase.DeleteSpecifiedData(ctx, model.Engine, model.Name)
 			if err != nil {
-				log.ZWarn(ctx, "Delete object is error", errs.Wrap(err), "object key", key)
+				return nil, errs.Wrap(err)
-			}
-
-			// Delete cache key
-			err = t.s3dataBase.DelS3Key(ctx, conf.Object.Enable, key)
-			if err != nil {
-				log.ZWarn(ctx, "Delete cache key is error:", errs.Wrap(err), "cache S3 key:", key)
 			}
 		}
-	}
+		if total < int64(findPagination.ShowNumber) {
-
+			break
-	// handle delete data in S3 database
-	for _, model := range models {
-		// Delete all expired data row in S3 database
-		err := t.s3dataBase.DeleteSpecifiedData(ctx, model.Engine, model.Name)
-		if err != nil {
-			return nil, errs.Wrap(err)
 		}
+		deltotal += int(total)
 	}
-
+	log.ZDebug(ctx, "DeleteOutdatedData", "delete Total", deltotal)
-	log.ZDebug(ctx, "DeleteOutdatedData", "delete Total", total)
+	return &third.DeleteOutdatedDataResp{}, nil
-
-	return &third.DeleteOutdatedDataResp{Count: int32(total)}, nil
 }
 
 type FormDataMate struct {

internal/rpc/third/third.go

@@ -38,7 +38,6 @@ import (
 )
 
 type thirdServer struct {
-	third.UnimplementedThirdServer
 	thirdDatabase controller.ThirdDatabase
 	s3dataBase    controller.S3Database
 	userRpcClient rpcclient.UserRpcClient

internal/rpc/user/user.go

@@ -52,7 +52,6 @@ import (
 )
 
 type userServer struct {
-	pbuser.UnimplementedUserServer
 	online                   cache.OnlineCache
 	db                       controller.UserDatabase
 	friendNotificationSender *relation.FriendNotificationSender

internal/tools/cron_task.go

@@ -24,7 +24,6 @@ import (
 	kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
 	pbconversation "github.com/openimsdk/protocol/conversation"
 	"github.com/openimsdk/protocol/msg"
-	"github.com/openimsdk/protocol/third"
 
 	"github.com/openimsdk/tools/mcontext"
 	"github.com/openimsdk/tools/mw"
@@ -59,10 +58,10 @@ func Start(ctx context.Context, config *CronTaskConfig) error {
 		return err
 	}
 
-	thirdConn, err := client.GetConn(ctx, config.Share.RpcRegisterName.Third)
+	// thirdConn, err := client.GetConn(ctx, config.Share.RpcRegisterName.Third)
-	if err != nil {
+	// if err != nil {
-		return err
+	// 	return err
-	}
+	// }
 
 	conversationConn, err := client.GetConn(ctx, config.Share.RpcRegisterName.Conversation)
 	if err != nil {
@@ -71,83 +70,66 @@ func Start(ctx context.Context, config *CronTaskConfig) error {
 
 	msgClient := msg.NewMsgClient(msgConn)
 	conversationClient := pbconversation.NewConversationClient(conversationConn)
-	thirdClient := third.NewThirdClient(thirdConn)
+	// thirdClient := third.NewThirdClient(thirdConn)
 
 	crontab := cron.New()
 
 	// scheduled hard delete outdated Msgs in specific time.
-	destructMsgsFunc := func() {
+	clearMsgFunc := func() {
 		now := time.Now()
 		deltime := now.Add(-time.Hour * 24 * time.Duration(config.CronTask.RetainChatRecords))
 		ctx := mcontext.SetOperationID(ctx, fmt.Sprintf("cron_%d_%d", os.Getpid(), deltime.UnixMilli()))
-		log.ZDebug(ctx, "Destruct chat records", "deltime", deltime, "timestamp", deltime.UnixMilli())
+		log.ZDebug(ctx, "clear chat records", "deltime", deltime, "timestamp", deltime.UnixMilli())
 
-		if _, err := msgClient.DestructMsgs(ctx, &msg.DestructMsgsReq{Timestamp: deltime.UnixMilli()}); err != nil {
+		if _, err := msgClient.ClearMsg(ctx, &msg.ClearMsgReq{Timestamp: deltime.UnixMilli()}); err != nil {
-			log.ZError(ctx, "cron destruct chat records failed", err, "deltime", deltime, "cont", time.Since(now))
+			log.ZError(ctx, "cron clear chat records failed", err, "deltime", deltime, "cont", time.Since(now))
 			return
 		}
-		log.ZDebug(ctx, "cron destruct chat records success", "deltime", deltime, "cont", time.Since(now))
+		log.ZDebug(ctx, "cron clear chat records success", "deltime", deltime, "cont", time.Since(now))
-	}
-	if _, err := crontab.AddFunc(config.CronTask.CronExecuteTime, destructMsgsFunc); err != nil {
-		return errs.Wrap(err)
-	}
-
-	// scheduled soft delete outdated Msgs in specific time when user set `is_msg_destruct` feature.
-	clearMsgFunc := func() {
-		now := time.Now()
-		ctx := mcontext.SetOperationID(ctx, fmt.Sprintf("cron_%d_%d", os.Getpid(), now.UnixMilli()))
-		log.ZDebug(ctx, "clear msg cron start", "now", now)
-
-		conversations, err := conversationClient.GetConversationsNeedClearMsg(ctx, &pbconversation.GetConversationsNeedClearMsgReq{})
-		if err != nil {
-			log.ZError(ctx, "Get conversation need Destruct msgs failed.", err)
-			return
-		}
-
-		_, err = msgClient.ClearMsg(ctx, &msg.ClearMsgReq{Conversations: conversations.Conversations})
-		if err != nil {
-			log.ZError(ctx, "Clear Msg failed.", err)
-			return
-		}
-
-		log.ZDebug(ctx, "clear msg cron task completed", "cont", time.Since(now))
 	}
 	if _, err := crontab.AddFunc(config.CronTask.CronExecuteTime, clearMsgFunc); err != nil {
 		return errs.Wrap(err)
 	}
 
-	// scheduled delete outdated file Objects and their datas in specific time.
+	// scheduled soft delete outdated Msgs in specific time when user set `is_msg_destruct` feature.
-	deleteObjectFunc := func() {
+	msgDestructFunc := func() {
 		now := time.Now()
-		executeNum := 5
+		ctx := mcontext.SetOperationID(ctx, fmt.Sprintf("cron_%d_%d", os.Getpid(), now.UnixMilli()))
-		// number of pagination. if need modify, need update value in third.DeleteOutdatedData
+		log.ZDebug(ctx, "msg destruct cron start", "now", now)
-		pageShowNumber := 500
-		deleteTime := now.Add(-time.Hour * 24 * time.Duration(config.CronTask.FileExpireTime))
-		ctx := mcontext.SetOperationID(ctx, fmt.Sprintf("cron_%d_%d", os.Getpid(), deleteTime.UnixMilli()))
-		log.ZDebug(ctx, "deleteoutDatedData", "deletetime", deleteTime, "timestamp", deleteTime.UnixMilli())
 
-		if len(config.CronTask.DeleteObjectType) == 0 {
+		conversations, err := conversationClient.GetConversationsNeedDestructMsgs(ctx, &pbconversation.GetConversationsNeedDestructMsgsReq{})
-			log.ZDebug(ctx, "cron deleteoutDatedData not type need delete", "deletetime", deleteTime, "DeleteObjectType", config.CronTask.DeleteObjectType, "cont", time.Since(now))
+		if err != nil {
+			log.ZError(ctx, "Get conversation need Destruct msgs failed.", err)
 			return
-		}
+		} else {
-
+			_, err := msgClient.DestructMsgs(ctx, &msg.DestructMsgsReq{Conversations: conversations.Conversations})
-		for i := 0; i < executeNum; i++ {
-			resp, err := thirdClient.DeleteOutdatedData(ctx, &third.DeleteOutdatedDataReq{ExpireTime: deleteTime.UnixMilli(), ObjectGroup: config.CronTask.DeleteObjectType})
 			if err != nil {
-				log.ZError(ctx, "cron deleteoutDatedData failed", err, "deleteTime", deleteTime, "cont", time.Since(now))
+				log.ZError(ctx, "Destruct Msgs failed.", err)
 				return
 			}
-			if resp.Count == 0 || resp.Count < int32(pageShowNumber) {
-				break
-			}
 		}
-
+		log.ZDebug(ctx, "msg destruct cron task completed", "cont", time.Since(now))
-		log.ZDebug(ctx, "cron deleteoutDatedData success", "deltime", deleteTime, "cont", time.Since(now))
 	}
-	if _, err := crontab.AddFunc(config.CronTask.CronExecuteTime, deleteObjectFunc); err != nil {
+	if _, err := crontab.AddFunc(config.CronTask.CronExecuteTime, msgDestructFunc); err != nil {
 		return errs.Wrap(err)
 	}
 
+	// // scheduled delete outdated file Objects and their datas in specific time.
+	// deleteObjectFunc := func() {
+	// 	now := time.Now()
+	// 	deleteTime := now.Add(-time.Hour * 24 * time.Duration(config.CronTask.FileExpireTime))
+	// 	ctx := mcontext.SetOperationID(ctx, fmt.Sprintf("cron_%d_%d", os.Getpid(), deleteTime.UnixMilli()))
+	// 	log.ZDebug(ctx, "deleteoutDatedData ", "deletetime", deleteTime, "timestamp", deleteTime.UnixMilli())
+	// 	if _, err := thirdClient.DeleteOutdatedData(ctx, &third.DeleteOutdatedDataReq{ExpireTime: deleteTime.UnixMilli()}); err != nil {
+	// 		log.ZError(ctx, "cron deleteoutDatedData failed", err, "deleteTime", deleteTime, "cont", time.Since(now))
+	// 		return
+	// 	}
+	// 	log.ZDebug(ctx, "cron deleteoutDatedData success", "deltime", deleteTime, "cont", time.Since(now))
+	// }
+	// if _, err := crontab.AddFunc(config.CronTask.CronExecuteTime, deleteObjectFunc); err != nil {
+	// 	return errs.Wrap(err)
+	// }
+
 	log.ZDebug(ctx, "start cron task", "CronExecuteTime", config.CronTask.CronExecuteTime)
 	crontab.Start()
 	<-ctx.Done()

pkg/apistruct/msg.go

@@ -81,6 +81,11 @@ type TextElem struct {
 	Content string `json:"content" validate:"required"`
 }
 
+type StreamMsgElem struct {
+	Type    string `mapstructure:"type" validate:"required"`
+	Content string `mapstructure:"content" validate:"required"`
+}
+
 type RevokeElem struct {
 	RevokeMsgClientID string `mapstructure:"revokeMsgClientID" validate:"required"`
 }

pkg/common/config/config.go

@@ -113,10 +113,9 @@ type API struct {
 }
 
 type CronTask struct {
-	CronExecuteTime   string   `mapstructure:"cronExecuteTime"`
+	CronExecuteTime   string `mapstructure:"cronExecuteTime"`
-	RetainChatRecords int      `mapstructure:"retainChatRecords"`
+	RetainChatRecords int    `mapstructure:"retainChatRecords"`
-	FileExpireTime    int      `mapstructure:"fileExpireTime"`
+	FileExpireTime    int    `mapstructure:"fileExpireTime"`
-	DeleteObjectType  []string `mapstructure:"deleteObjectType"`
 }
 
 type OfflinePushConfig struct {

pkg/common/storage/controller/auth.go

@@ -2,7 +2,6 @@ package controller
 
 import (
 	"context"
-
 	"github.com/golang-jwt/jwt/v4"
 	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
@@ -42,7 +41,8 @@ func NewAuthDatabase(cache cache.TokenModel, accessSecret string, accessExpire i
 	return &authDatabase{cache: cache, accessSecret: accessSecret, accessExpire: accessExpire, multiLogin: multiLoginConfig{
 		Policy:       multiLogin.Policy,
 		MaxNumOneEnd: multiLogin.MaxNumOneEnd,
-	}, adminUserIDs: adminUserIDs,
+	},
+		adminUserIDs: adminUserIDs,
 	}
 }
 

pkg/common/storage/controller/conversation.go

@@ -179,7 +179,7 @@ func (c *conversationDatabase) CreateConversation(ctx context.Context, conversat
 		if conversation.RecvMsgOpt == constant.ReceiveNotNotifyMessage {
 			notNotifyUserIDs = append(notNotifyUserIDs, conversation.OwnerUserID)
 		}
-		if conversation.IsPinned {
+		if conversation.IsPinned == true {
 			pinnedUserIDs = append(pinnedUserIDs, conversation.OwnerUserID)
 		}
 	}

pkg/common/storage/controller/msg.go

@@ -24,9 +24,6 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 
-	"github.com/redis/go-redis/v9"
-	"go.mongodb.org/mongo-driver/mongo"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/convert"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
@@ -38,6 +35,8 @@ import (
 	"github.com/openimsdk/tools/mq/kafka"
 	"github.com/openimsdk/tools/utils/datautil"
 	"github.com/openimsdk/tools/utils/timeutil"
+	"github.com/redis/go-redis/v9"
+	"go.mongodb.org/mongo-driver/mongo"
 )
 
 const (
@@ -57,10 +56,9 @@ type CommonMsgDatabase interface {
 	GetMsgBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) (minSeq int64, maxSeq int64, seqMsg []*sdkws.MsgData, err error)
 	// DeleteConversationMsgsAndSetMinSeq deletes conversation messages and resets the minimum sequence number. If `remainTime` is 0, all messages are deleted (this method does not delete Redis
 	// cache).
-	GetMessagesBySeqWithBounds(ctx context.Context, userID string, conversationID string, seqs []int64, pullOrder sdkws.PullOrder) (bool, int64, []*sdkws.MsgData, error)
 	DeleteConversationMsgsAndSetMinSeq(ctx context.Context, conversationID string, remainTime int64) error
-	// ClearUserMsgs marks messages for deletion based on clear time and returns a list of sequence numbers for marked messages.
+	// UserMsgsDestruct marks messages for deletion based on destruct time and returns a list of sequence numbers for marked messages.
-	ClearUserMsgs(ctx context.Context, userID string, conversationID string, clearTime int64, lastMsgClearTime time.Time) (seqs []int64, err error)
+	UserMsgsDestruct(ctx context.Context, userID string, conversationID string, destructTime int64, lastMsgDestructTime time.Time) (seqs []int64, err error)
 	// DeleteUserMsgsBySeqs allows a user to delete messages based on sequence numbers.
 	DeleteUserMsgsBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) error
 	// DeleteMsgsPhysicalBySeqs physically deletes messages by emptying them based on sequence numbers.
@@ -94,14 +92,11 @@ type CommonMsgDatabase interface {
 	RangeGroupSendCount(ctx context.Context, start time.Time, end time.Time, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, groups []*model.GroupCount, dateCount map[string]int64, err error)
 	ConvertMsgsDocLen(ctx context.Context, conversationIDs []string)
 
-	// get Msg when destruct msg before
+	// clear msg
 	GetBeforeMsg(ctx context.Context, ts int64, docIds []string, limit int) ([]*model.MsgDocModel, error)
 	DeleteDocMsgBefore(ctx context.Context, ts int64, doc *model.MsgDocModel) ([]int, error)
 
 	GetDocIDs(ctx context.Context) ([]string, error)
-
-	SetUserConversationsMaxSeq(ctx context.Context, conversationID string, userID string, seq int64) error
-	SetUserConversationsMinSeq(ctx context.Context, conversationID string, userID string, seq int64) error
 }
 
 func NewCommonMsgDatabase(msgDocModel database.Msg, msg cache.MsgCache, seqUser cache.SeqUser, seqConversation cache.SeqConversationCache, kafkaConf *config.Kafka) (CommonMsgDatabase, error) {
@@ -495,8 +490,8 @@ func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, co
 	}
 	successMsgs, failedSeqs, err := db.msg.GetMessagesBySeq(ctx, conversationID, newSeqs)
 	if err != nil {
-		if !errors.Is(err, redis.Nil) {
+		if errors.Is(err, redis.Nil) {
-			log.ZWarn(ctx, "get message from redis exception", err, "failedSeqs", failedSeqs, "conversationID", conversationID)
+			log.ZError(ctx, "get message from redis exception", err, "failedSeqs", failedSeqs, "conversationID", conversationID)
 		}
 	}
 	log.ZDebug(ctx, "db.seq.GetMessagesBySeq", "userID", userID, "conversationID", conversationID, "seqs",
@@ -519,81 +514,6 @@ func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, co
 	return minSeq, maxSeq, successMsgs, nil
 }
 
-func (db *commonMsgDatabase) GetMessagesBySeqWithBounds(ctx context.Context, userID string, conversationID string, seqs []int64, pullOrder sdkws.PullOrder) (bool, int64, []*sdkws.MsgData, error) {
-	var endSeq int64
-	var isEnd bool
-	userMinSeq, err := db.seqUser.GetUserMinSeq(ctx, conversationID, userID)
-	if err != nil {
-		return false, 0, nil, err
-	}
-	minSeq, err := db.seqConversation.GetMinSeq(ctx, conversationID)
-	if err != nil {
-		return false, 0, nil, err
-	}
-	maxSeq, err := db.seqConversation.GetMaxSeq(ctx, conversationID)
-	if err != nil {
-		return false, 0, nil, err
-	}
-	userMaxSeq, err := db.seqUser.GetUserMaxSeq(ctx, conversationID, userID)
-	if err != nil {
-		return false, 0, nil, err
-	}
-	if userMinSeq > minSeq {
-		minSeq = userMinSeq
-	}
-	if userMaxSeq > 0 && userMaxSeq < maxSeq {
-		maxSeq = userMaxSeq
-	}
-	newSeqs := make([]int64, 0, len(seqs))
-	for _, seq := range seqs {
-		if seq <= 0 {
-			continue
-		}
-		// The normal range and can fetch messages
-		if seq >= minSeq && seq <= maxSeq {
-			newSeqs = append(newSeqs, seq)
-			continue
-		}
-		// If the requested seq is smaller than the minimum seq and the pull order is descending (pulling older messages)
-		if seq < minSeq && pullOrder == sdkws.PullOrder_PullOrderDesc {
-			isEnd = true
-			endSeq = minSeq
-		}
-		// If the requested seq is larger than the maximum seq and the pull order is ascending (pulling newer messages)
-		if seq > maxSeq && pullOrder == sdkws.PullOrder_PullOrderAsc {
-			isEnd = true
-			endSeq = maxSeq
-		}
-	}
-	if len(newSeqs) == 0 {
-		return isEnd, endSeq, nil, nil
-	}
-	successMsgs, failedSeqs, err := db.msg.GetMessagesBySeq(ctx, conversationID, newSeqs)
-	if err != nil {
-		if !errors.Is(err, redis.Nil) {
-			log.ZWarn(ctx, "get message from redis exception", err, "failedSeqs", failedSeqs, "conversationID", conversationID)
-		}
-	}
-	log.ZDebug(ctx, "db.seq.GetMessagesBySeq", "userID", userID, "conversationID", conversationID, "seqs",
-		seqs, "len(successMsgs)", len(successMsgs), "failedSeqs", failedSeqs)
-
-	if len(failedSeqs) > 0 {
-		mongoMsgs, err := db.getMsgBySeqs(ctx, userID, conversationID, failedSeqs)
-		if err != nil {
-
-			return false, 0, nil, err
-		}
-
-		successMsgs = append(successMsgs, mongoMsgs...)
-
-		//_, err = db.msg.SetMessagesToCache(ctx, conversationID, mongoMsgs)
-		//if err != nil {
-		//	return 0, 0, nil, err
-		//}
-	}
-	return isEnd, endSeq, successMsgs, nil
-}
-
 func (db *commonMsgDatabase) DeleteConversationMsgsAndSetMinSeq(ctx context.Context, conversationID string, remainTime int64) error {
 	var delStruct delMsgRecursionStruct
 	var skip int64
@@ -608,10 +528,10 @@ func (db *commonMsgDatabase) DeleteConversationMsgsAndSetMinSeq(ctx context.Cont
 	return db.seqConversation.SetMinSeq(ctx, conversationID, minSeq)
 }
 
-func (db *commonMsgDatabase) ClearUserMsgs(ctx context.Context, userID string, conversationID string, clearTime int64, lastMsgClearTime time.Time) (seqs []int64, err error) {
+func (db *commonMsgDatabase) UserMsgsDestruct(ctx context.Context, userID string, conversationID string, destructTime int64, lastMsgDestructTime time.Time) (seqs []int64, err error) {
 	var index int64
 	for {
-		// from oldest 2 newest, ASC
+		// from oldest 2 newest
 		msgDocModel, err := db.msgDocDatabase.GetMsgDocModelByIndex(ctx, conversationID, index, 1)
 		if err != nil || msgDocModel.DocID == "" {
 			if err != nil {
@@ -624,19 +544,15 @@ func (db *commonMsgDatabase) ClearUserMsgs(ctx context.Context, userID string, c
 			// If an error is reported, or the error cannot be obtained, it is physically deleted and seq delMongoMsgsPhysical(delStruct.delDocIDList) is returned to end the recursion
 			break
 		}
-
 		index++
-
+		// && msgDocModel.Msg[0].Msg.SendTime > lastMsgDestructTime.UnixMilli()
-		// && msgDocModel.Msg[0].Msg.SendTime > lastMsgClearTime.UnixMilli()
 		if len(msgDocModel.Msg) > 0 {
 			i := 0
 			var over bool
 			for _, msg := range msgDocModel.Msg {
 				i++
-				// over clear time, need to clear
+				if msg != nil && msg.Msg != nil && msg.Msg.SendTime+destructTime*1000 <= time.Now().UnixMilli() {
-				if msg != nil && msg.Msg != nil && msg.Msg.SendTime+clearTime*1000 <= time.Now().UnixMilli() {
+					if msg.Msg.SendTime+destructTime*1000 > lastMsgDestructTime.UnixMilli() && !datautil.Contain(userID, msg.DelList...) {
-					// if msg is not in del list, add to del list
-					if msg.Msg.SendTime+clearTime*1000 > lastMsgClearTime.UnixMilli() && !datautil.Contain(userID, msg.DelList...) {
 						seqs = append(seqs, msg.Msg.Seq)
 					}
 				} else {
@@ -651,18 +567,13 @@ func (db *commonMsgDatabase) ClearUserMsgs(ctx context.Context, userID string, c
 		}
 	}
 
-	log.ZDebug(ctx, "ClearUserMsgs", "conversationID", conversationID, "userID", userID, "seqs", seqs)
+	log.ZDebug(ctx, "UserMsgsDestruct", "conversationID", conversationID, "userID", userID, "seqs", seqs)
-
-	// have msg need to destruct
 	if len(seqs) > 0 {
-		// update min seq to clear after
+		userMinSeq := seqs[len(seqs)-1] + 1
-		userMinSeq := seqs[len(seqs)-1] + 1                                             // user min seq when clear after
+		currentUserMinSeq, err := db.seqUser.GetUserMinSeq(ctx, conversationID, userID)
-		currentUserMinSeq, err := db.seqUser.GetUserMinSeq(ctx, conversationID, userID) // user min seq when clear before
 		if err != nil {
 			return nil, err
 		}
-
-		// if before < after, update min seq
 		if currentUserMinSeq < userMinSeq {
 			if err := db.seqUser.SetUserMinSeq(ctx, conversationID, userID, userMinSeq); err != nil {
 				return nil, err
@@ -782,14 +693,6 @@ func (db *commonMsgDatabase) SetUserConversationsMinSeqs(ctx context.Context, us
 	return db.seqUser.SetUserMinSeqs(ctx, userID, seqs)
 }
 
-func (db *commonMsgDatabase) SetUserConversationsMaxSeq(ctx context.Context, conversationID string, userID string, seq int64) error {
-	return db.seqUser.SetUserMaxSeq(ctx, conversationID, userID, seq)
-}
-
-func (db *commonMsgDatabase) SetUserConversationsMinSeq(ctx context.Context, conversationID string, userID string, seq int64) error {
-	return db.seqUser.SetUserMinSeq(ctx, conversationID, userID, seq)
-}
-
 func (db *commonMsgDatabase) UserSetHasReadSeqs(ctx context.Context, userID string, hasReadSeqs map[string]int64) error {
 	return db.seqUser.SetUserReadSeqs(ctx, userID, hasReadSeqs)
 }

pkg/common/storage/controller/s3.go

@@ -40,10 +40,10 @@ type S3Database interface {
 	SetObject(ctx context.Context, info *model.Object) error
 	StatObject(ctx context.Context, name string) (*s3.ObjectInfo, error)
 	FormData(ctx context.Context, name string, size int64, contentType string, duration time.Duration) (*s3.FormData, error)
-	FindNeedDeleteObjectByDB(ctx context.Context, duration time.Time, needDelType []string, pagination pagination.Pagination) (total int64, objects []*model.Object, err error)
+	FindByExpires(ctx context.Context, duration time.Time, pagination pagination.Pagination) (total int64, objects []*model.Object, err error)
 	DeleteObject(ctx context.Context, name string) error
 	DeleteSpecifiedData(ctx context.Context, engine string, name string) error
-	FindModelsByKey(ctx context.Context, key string) (objects []*model.Object, err error)
+	FindNotDelByS3(ctx context.Context, key string, duration time.Time) (int64, error)
 	DelS3Key(ctx context.Context, engine string, keys ...string) error
 }
 
@@ -120,8 +120,9 @@ func (s *s3Database) StatObject(ctx context.Context, name string) (*s3.ObjectInf
 func (s *s3Database) FormData(ctx context.Context, name string, size int64, contentType string, duration time.Duration) (*s3.FormData, error) {
 	return s.s3.FormData(ctx, name, size, contentType, duration)
 }
-func (s *s3Database) FindNeedDeleteObjectByDB(ctx context.Context, duration time.Time, needDelType []string, pagination pagination.Pagination) (total int64, objects []*model.Object, err error) {
+func (s *s3Database) FindByExpires(ctx context.Context, duration time.Time, pagination pagination.Pagination) (total int64, objects []*model.Object, err error) {
-	return s.db.FindNeedDeleteObjectByDB(ctx, duration, needDelType, pagination)
+
+	return s.db.FindByExpires(ctx, duration, pagination)
 }
 
 func (s *s3Database) DeleteObject(ctx context.Context, name string) error {
@@ -131,8 +132,8 @@ func (s *s3Database) DeleteSpecifiedData(ctx context.Context, engine string, nam
 	return s.db.Delete(ctx, engine, name)
 }
 
-func (s *s3Database) FindModelsByKey(ctx context.Context, key string) (objects []*model.Object, err error) {
+func (s *s3Database) FindNotDelByS3(ctx context.Context, key string, duration time.Time) (int64, error) {
-	return s.db.FindModelsByKey(ctx, key)
+	return s.db.FindNotDelByS3(ctx, key, duration)
 }
 
 func (s *s3Database) DelS3Key(ctx context.Context, engine string, keys ...string) error {

pkg/common/storage/controller/stream_msg.go

@@ -0,0 +1,34 @@
+package controller
+
+import (
+	"context"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
+	"time"
+)
+
+type StreamMsgDatabase interface {
+	CreateStreamMsg(ctx context.Context, model *model.StreamMsg) error
+	AppendStreamMsg(ctx context.Context, clientMsgID string, startIndex int, packets []string, end bool, deadlineTime time.Time) error
+	GetStreamMsg(ctx context.Context, clientMsgID string) (*model.StreamMsg, error)
+}
+
+func NewStreamMsgDatabase(db database.StreamMsg) StreamMsgDatabase {
+	return &streamMsgDatabase{db: db}
+}
+
+type streamMsgDatabase struct {
+	db database.StreamMsg
+}
+
+func (m *streamMsgDatabase) CreateStreamMsg(ctx context.Context, model *model.StreamMsg) error {
+	return m.db.CreateStreamMsg(ctx, model)
+}
+
+func (m *streamMsgDatabase) AppendStreamMsg(ctx context.Context, clientMsgID string, startIndex int, packets []string, end bool, deadlineTime time.Time) error {
+	return m.db.AppendStreamMsg(ctx, clientMsgID, startIndex, packets, end, deadlineTime)
+}
+
+func (m *streamMsgDatabase) GetStreamMsg(ctx context.Context, clientMsgID string) (*model.StreamMsg, error) {
+	return m.db.GetStreamMsg(ctx, clientMsgID)
+}

pkg/common/storage/database/conversation.go

@@ -16,7 +16,6 @@ package database
 
 import (
 	"context"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 	"github.com/openimsdk/tools/db/pagination"
 )

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

@@ -16,10 +16,9 @@ package mgo
 
 import (
 	"context"
-	"time"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
+	"time"
 
 	"github.com/openimsdk/protocol/constant"
 	"github.com/openimsdk/tools/db/mongoutil"

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

@@ -60,7 +60,7 @@ type GroupMemberMgo struct {
 }
 
 func (g *GroupMemberMgo) memberSort() any {
-	return bson.D{{Key: "role_level", Value: -1}, {Key: "create_time", Value: 1}}
+	return bson.D{{"role_level", -1}, {"create_time", 1}}
 }
 
 func (g *GroupMemberMgo) Create(ctx context.Context, groupMembers []*model.GroupMember) (err error) {

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

@@ -31,8 +31,6 @@ import (
 
 func NewS3Mongo(db *mongo.Database) (database.ObjectInfo, error) {
 	coll := db.Collection(database.ObjectName)
-
-	// Create index for name
 	_, err := coll.Indexes().CreateOne(context.Background(), mongo.IndexModel{
 		Keys: bson.D{
 			{Key: "name", Value: 1},
@@ -42,27 +40,6 @@ func NewS3Mongo(db *mongo.Database) (database.ObjectInfo, error) {
 	if err != nil {
 		return nil, errs.Wrap(err)
 	}
-
-	// Create index for create_time
-	_, err = coll.Indexes().CreateOne(context.Background(), mongo.IndexModel{
-		Keys: bson.D{
-			{Key: "create_time", Value: 1},
-		},
-	})
-	if err != nil {
-		return nil, errs.Wrap(err)
-	}
-
-	// Create index for key
-	_, err = coll.Indexes().CreateOne(context.Background(), mongo.IndexModel{
-		Keys: bson.D{
-			{Key: "key", Value: 1},
-		},
-	})
-	if err != nil {
-		return nil, errs.Wrap(err)
-	}
-
 	return &S3Mongo{coll: coll}, nil
 }
 
@@ -94,18 +71,14 @@ func (o *S3Mongo) Take(ctx context.Context, engine string, name string) (*model.
 func (o *S3Mongo) Delete(ctx context.Context, engine string, name string) error {
 	return mongoutil.DeleteOne(ctx, o.coll, bson.M{"name": name, "engine": engine})
 }
-
+func (o *S3Mongo) FindByExpires(ctx context.Context, duration time.Time, pagination pagination.Pagination) (total int64, objects []*model.Object, err error) {
-// Find Expires object
-func (o *S3Mongo) FindNeedDeleteObjectByDB(ctx context.Context, duration time.Time, needDelType []string, pagination pagination.Pagination) (total int64, objects []*model.Object, err error) {
 	return mongoutil.FindPage[*model.Object](ctx, o.coll, bson.M{
 		"create_time": bson.M{"$lt": duration},
-		"group":       bson.M{"$in": needDelType},
 	}, pagination)
 }
-
+func (o *S3Mongo) FindNotDelByS3(ctx context.Context, key string, duration time.Time) (int64, error) {
-// Find object by key
+	return mongoutil.Count(ctx, o.coll, bson.M{
-func (o *S3Mongo) FindModelsByKey(ctx context.Context, key string) (objects []*model.Object, err error) {
+		"key":         key,
-	return mongoutil.Find[*model.Object](ctx, o.coll, bson.M{
+		"create_time": bson.M{"$gt": duration},
-		"key": key,
 	})
 }

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

@@ -0,0 +1,60 @@
+package mgo
+
+import (
+	"context"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
+	"github.com/openimsdk/tools/db/mongoutil"
+	"github.com/openimsdk/tools/errs"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+	"time"
+)
+
+func NewStreamMsgMongo(db *mongo.Database) (*StreamMsgMongo, error) {
+	coll := db.Collection(database.StreamMsgName)
+	_, err := coll.Indexes().CreateOne(context.Background(), mongo.IndexModel{
+		Keys: bson.D{
+			{Key: "client_msg_id", Value: 1},
+		},
+		Options: options.Index().SetUnique(true),
+	})
+	if err != nil {
+		return nil, errs.Wrap(err)
+	}
+	return &StreamMsgMongo{coll: coll}, nil
+}
+
+type StreamMsgMongo struct {
+	coll *mongo.Collection
+}
+
+func (m *StreamMsgMongo) CreateStreamMsg(ctx context.Context, val *model.StreamMsg) error {
+	if val.Packets == nil {
+		val.Packets = []string{}
+	}
+	return mongoutil.InsertMany(ctx, m.coll, []*model.StreamMsg{val})
+}
+
+func (m *StreamMsgMongo) AppendStreamMsg(ctx context.Context, clientMsgID string, startIndex int, packets []string, end bool, deadlineTime time.Time) error {
+	update := bson.M{
+		"$set": bson.M{
+			"end":           end,
+			"deadline_time": deadlineTime,
+		},
+	}
+	if len(packets) > 0 {
+		update["$push"] = bson.M{
+			"packets": bson.M{
+				"$each":     packets,
+				"$position": startIndex,
+			},
+		}
+	}
+	return mongoutil.UpdateOne(ctx, m.coll, bson.M{"client_msg_id": clientMsgID, "end": false}, update, true)
+}
+
+func (m *StreamMsgMongo) GetStreamMsg(ctx context.Context, clientMsgID string) (*model.StreamMsg, error) {
+	return mongoutil.FindOne[*model.StreamMsg](ctx, m.coll, bson.M{"client_msg_id": clientMsgID})
+}

pkg/common/storage/database/name.go

@@ -17,4 +17,5 @@ const (
 	UserName                = "user"
 	SeqConversationName     = "seq"
 	SeqUserName             = "seq_user"
+	StreamMsgName           = "stream_msg"
 )

pkg/common/storage/database/object.go

@@ -26,6 +26,6 @@ type ObjectInfo interface {
 	SetObject(ctx context.Context, obj *model.Object) error
 	Take(ctx context.Context, engine string, name string) (*model.Object, error)
 	Delete(ctx context.Context, engine string, name string) error
-	FindNeedDeleteObjectByDB(ctx context.Context, duration time.Time, needDelType []string, pagination pagination.Pagination) (total int64, objects []*model.Object, err error)
+	FindByExpires(ctx context.Context, duration time.Time, pagination pagination.Pagination) (total int64, objects []*model.Object, err error)
-	FindModelsByKey(ctx context.Context, key string) (objects []*model.Object, err error)
+	FindNotDelByS3(ctx context.Context, key string, duration time.Time) (int64, error)
 }

pkg/common/storage/database/stream_msg.go

@@ -0,0 +1,13 @@
+package database
+
+import (
+	"context"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
+	"time"
+)
+
+type StreamMsg interface {
+	CreateStreamMsg(ctx context.Context, model *model.StreamMsg) error
+	AppendStreamMsg(ctx context.Context, clientMsgID string, startIndex int, packets []string, end bool, deadlineTime time.Time) error
+	GetStreamMsg(ctx context.Context, clientMsgID string) (*model.StreamMsg, error)
+}

pkg/common/storage/model/log.go

@@ -19,14 +19,13 @@ import (
 )
 
 type Log struct {
-	LogID        string    `bson:"log_id"`
+	LogID      string    `bson:"log_id"`
-	Platform     string    `bson:"platform"`
+	Platform   string    `bson:"platform"`
-	UserID       string    `bson:"user_id"`
+	UserID     string    `bson:"user_id"`
-	CreateTime   time.Time `bson:"create_time"`
+	CreateTime time.Time `bson:"create_time"`
-	Url          string    `bson:"url"`
+	Url        string    `bson:"url"`
-	FileName     string    `bson:"file_name"`
+	FileName   string    `bson:"file_name"`
-	SystemType   string    `bson:"system_type"`
+	SystemType string    `bson:"system_type"`
-	AppFramework string    `bson:"app_framework"`
+	Version    string    `bson:"version"`
-	Version      string    `bson:"version"`
+	Ex         string    `bson:"ex"`
-	Ex           string    `bson:"ex"`
 }

pkg/common/storage/model/stream_msg.go

@@ -0,0 +1,21 @@
+package model
+
+import (
+	"time"
+)
+
+const (
+	StreamMsgStatusWait = 0
+	StreamMsgStatusDone = 1
+	StreamMsgStatusFail = 2
+)
+
+type StreamMsg struct {
+	ClientMsgID    string    `bson:"client_msg_id"`
+	ConversationID string    `bson:"conversation_id"`
+	UserID         string    `bson:"user_id"`
+	Packets        []string  `bson:"packets"`
+	End            bool      `bson:"end"`
+	CreateTime     time.Time `bson:"create_time"`
+	DeadlineTime   time.Time `bson:"deadline_time"`
+}

pkg/rpccache/subscriber.go

@@ -17,6 +17,7 @@ package rpccache
 import (
 	"context"
 	"encoding/json"
+	"github.com/openimsdk/tools/mw"
 
 	"github.com/openimsdk/tools/log"
 	"github.com/redis/go-redis/v9"
@@ -25,7 +26,7 @@ import (
 func subscriberRedisDeleteCache(ctx context.Context, client redis.UniversalClient, channel string, del func(ctx context.Context, key ...string)) {
 	defer func() {
 		if r := recover(); r != nil {
-			log.ZPanic(ctx, "subscriberRedisDeleteCache Panic", r)
+			mw.PanicStackToLog(ctx, r)
 		}
 	}()
 	for message := range client.Subscribe(ctx, channel).Channel() {

pkg/rpcclient/conversation.go

@@ -152,8 +152,8 @@ func (c *ConversationRpcClient) GetConversationNotReceiveMessageUserIDs(ctx cont
 	return resp.UserIDs, nil
 }
 
-func (c *ConversationRpcClient) GetConversationsNeedClearMsg(ctx context.Context) ([]*pbconversation.Conversation, error) {
+func (c *ConversationRpcClient) GetConversationsNeedDestructMsgs(ctx context.Context) ([]*pbconversation.Conversation, error) {
-	resp, err := c.Client.GetConversationsNeedClearMsg(ctx, &pbconversation.GetConversationsNeedClearMsgReq{})
+	resp, err := c.Client.GetConversationsNeedDestructMsgs(ctx, &pbconversation.GetConversationsNeedDestructMsgsReq{})
 	if err != nil {
 		return nil, err
 	}

pkg/rpcclient/msg.go

@@ -244,8 +244,8 @@ func (m *MessageRpcClient) GetConversationMaxSeq(ctx context.Context, conversati
 	return resp.MaxSeq, nil
 }
 
-func (m *MessageRpcClient) DestructMsgs(ctx context.Context, ts int64) error {
+func (m *MessageRpcClient) ClearMsg(ctx context.Context, ts int64) error {
-	_, err := m.Client.DestructMsgs(ctx, &msg.DestructMsgsReq{Timestamp: ts})
+	_, err := m.Client.ClearMsg(ctx, &msg.ClearMsgReq{Timestamp: ts})
 	return err
 }
 

tools/streammsg/main.go

@@ -0,0 +1,161 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
+	"github.com/openimsdk/open-im-server/v3/pkg/apistruct"
+	cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
+	"github.com/openimsdk/protocol/auth"
+	"github.com/openimsdk/protocol/constant"
+	"github.com/openimsdk/protocol/msg"
+	"github.com/openimsdk/tools/apiresp"
+	"github.com/openimsdk/tools/errs"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+)
+
+const (
+	getAdminToken   = "/auth/get_admin_token"
+	sendMsgApi      = "/msg/send_msg"
+	appendStreamMsg = "/msg/append_stream_msg"
+)
+
+var (
+	ApiAddr = "http://127.0.0.1:10002"
+	Token   string
+)
+
+func ApiCall[R any](api string, req any) (*R, error) {
+	data, err := json.Marshal(req)
+	if err != nil {
+		return nil, err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
+	defer cancel()
+	request, err := http.NewRequestWithContext(ctx, http.MethodPost, ApiAddr+api, bytes.NewBuffer(data))
+	if err != nil {
+		return nil, err
+	}
+	if Token != "" {
+		request.Header.Set("token", Token)
+	}
+	request.Header.Set(constant.OperationID, uuid.New().String())
+	response, err := http.DefaultClient.Do(request)
+	if err != nil {
+		return nil, err
+	}
+	defer response.Body.Close()
+	var resp R
+	apiResponse := apiresp.ApiResponse{
+		Data: &resp,
+	}
+	if err := json.NewDecoder(response.Body).Decode(&apiResponse); err != nil {
+		return nil, err
+	}
+	if apiResponse.ErrCode != 0 {
+		return nil, errs.NewCodeError(apiResponse.ErrCode, apiResponse.ErrMsg)
+	}
+	return &resp, nil
+}
+
+func main() {
+	resp, err := ApiCall[auth.GetAdminTokenResp](getAdminToken, &auth.GetAdminTokenReq{
+		Secret: "openIM123",
+		UserID: "imAdmin",
+	})
+	if err != nil {
+		fmt.Println("get admin token failed", err)
+		return
+	}
+	Token = resp.Token
+	g := gin.Default()
+	g.POST("/callbackExample/callbackAfterSendSingleMsgCommand", toGin(handlerUserMsg))
+	if err := g.Run(":10006"); err != nil {
+		panic(err)
+	}
+}
+
+func toGin[R any](fn func(c *gin.Context, req *R) error) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		body, err := io.ReadAll(c.Request.Body)
+		if err != nil {
+			c.String(http.StatusInternalServerError, err.Error())
+			return
+		}
+		fmt.Printf("HTTP %s %s %s\n", c.Request.Method, c.Request.URL, body)
+		var req R
+		if err := json.Unmarshal(body, &req); err != nil {
+			c.String(http.StatusInternalServerError, err.Error())
+			return
+		}
+		if err := fn(c, &req); err != nil {
+			c.String(http.StatusInternalServerError, err.Error())
+			return
+		}
+		c.String(http.StatusOK, "{}")
+	}
+}
+
+func handlerUserMsg(c *gin.Context, req *cbapi.CallbackAfterSendSingleMsgReq) error {
+	if req.ContentType != constant.Text {
+		return nil
+	}
+	if !strings.Contains(req.Content, "stream") {
+		return nil
+	}
+	apiReq := apistruct.SendMsgReq{
+		RecvID: req.SendID,
+		SendMsg: apistruct.SendMsg{
+			SendID:           req.RecvID,
+			SenderNickname:   "xxx",
+			SenderFaceURL:    "",
+			SenderPlatformID: constant.AdminPlatformID,
+			ContentType:      constant.Stream,
+			SessionType:      req.SessionType,
+			SendTime:         time.Now().UnixMilli(),
+			Content: map[string]any{
+				"type":    "xxx",
+				"content": "server test stream msg",
+			},
+		},
+	}
+	go func() {
+		if err := doPushStreamMsg(&apiReq); err != nil {
+			fmt.Println("doPushStreamMsg failed", err)
+			return
+		}
+		fmt.Println("doPushStreamMsg success")
+	}()
+	return nil
+}
+
+func doPushStreamMsg(sendReq *apistruct.SendMsgReq) error {
+	resp, err := ApiCall[msg.SendMsgResp](sendMsgApi, sendReq)
+	if err != nil {
+		return err
+	}
+	const num = 5
+	for i := 1; i <= num; i++ {
+		_, err := ApiCall[msg.AppendStreamMsgResp](appendStreamMsg, &msg.AppendStreamMsgReq{
+			ClientMsgID: resp.ClientMsgID,
+			StartIndex:  int64(i - 1),
+			Packets: []string{
+				fmt.Sprintf("stream_msg_packet_%03d", i),
+			},
+			End: i == num,
+		})
+		if err != nil {
+			fmt.Println("append stream msg failed", "clientMsgID", resp.ClientMsgID, "index", fmt.Sprintf("%d/%d", i, num), "error", err)
+			return err
+		}
+		fmt.Println("append stream msg success", "clientMsgID", resp.ClientMsgID, "index", fmt.Sprintf("%d/%d", i, num))
+		time.Sleep(time.Second * 10)
+	}
+	return nil
+}

version/version

@@ -1 +1 @@
-v3.8.3-alpha.1
+main