Update CHANGELOG for release v3.8.2

* build: update Server version.

* update version.

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)
+

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-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.
-enable: geTui
+#Use geTui for offline push notifications, or choose fcm or jpns; corresponding configuration settings must be specified.
+enable:
 geTui:
   pushUrl: https://restapi.getui.com/v2/$appId
   masterSecret:

go.mod

@@ -14,7 +14,7 @@ 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.54
+	github.com/openimsdk/protocol v0.0.72-alpha.55
 	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

go.sum

@@ -319,8 +319,8 @@ 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.54 h1:opato7N4QjjRq/SHD54bDSVBpOEEDp1VLWVk5Os2A9s=
-github.com/openimsdk/protocol v0.0.72-alpha.54/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
+github.com/openimsdk/protocol v0.0.72-alpha.55 h1:9PPWPHvkFk3neBSbNr+IoOdKIFjxTvEqUfMK/TEq1+8=
+github.com/openimsdk/protocol v0.0.72-alpha.55/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
 github.com/openimsdk/tools v0.0.50-alpha.32 h1:JEsUFHFnaYg230TG+Ke3SUnaA2h44t4kABAzEdv5VZw=
 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=

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/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

@@ -35,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)

internal/rpc/msg/server.go

@@ -42,8 +42,9 @@ type (
 
 	// MsgServer encapsulates dependencies required for message handling.
 	msgServer struct {
-		RegisterCenter         discovery.SvcDiscoveryRegistry   // Service discovery registry for service registration.
-		MsgDatabase            controller.CommonMsgDatabase     // Interface for message database operations.
+		RegisterCenter         discovery.SvcDiscoveryRegistry // Service discovery registry for service registration.
+		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.
@@ -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
+}

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/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/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/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/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/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"`
+}

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 @@
-3.8.2
+main