feat: Support Streaming Messages in the Open-Source Server

go.mod

@@ -12,7 +12,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.73-alpha.19
+	github.com/openimsdk/protocol v0.0.73-alpha.20
 	github.com/openimsdk/tools v0.0.50-alpha.121
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0

go.sum

@@ -361,8 +361,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.17 h1:q8haP48VOH45WhJRiLj1YSBJyUFJqD8CTedH65i1YH8=
 github.com/openimsdk/gomake v0.0.17/go.mod h1:nnjS8yCtrPJAt1knMbyPiUwCH2gpyBzj/EZAONfUOXg=
-github.com/openimsdk/protocol v0.0.73-alpha.19 h1:CvXoDF2U73UcMhLnrtMFks2Aw+bXiDgH8AITEt783/s=
-github.com/openimsdk/protocol v0.0.73-alpha.19/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw=
+github.com/openimsdk/protocol v0.0.73-alpha.20 h1:9MnACSi6IKv2iqlxHYUJG9mgt9gyPRHxE7Lq8dxAoxI=
+github.com/openimsdk/protocol v0.0.73-alpha.20/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw=
 github.com/openimsdk/tools v0.0.50-alpha.121 h1:TXKKgtkeMeqIs0vpolbW8rIEngE9xlESq+0NV+FoLH0=
 github.com/openimsdk/tools v0.0.50-alpha.121/go.mod h1:I0WESSa7ghPIo9BL+ETlH/qEIbO6+KZioM1jwNuDwz0=
 github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=

internal/api/msg.go

@@ -79,12 +79,13 @@ func getMsgDataDescriptor() []protoreflect.FieldDescriptor {
 type MessageApi struct {
 	Client        msg.MsgClient
 	userClient    *rpcli.UserClient
+	authClient    *rpcli.AuthClient
 	imAdminUserID []string
 	validate      *validator.Validate
 }
 
-func NewMessageApi(client msg.MsgClient, userClient *rpcli.UserClient, imAdminUserID []string) MessageApi {
-	return MessageApi{Client: client, userClient: userClient, imAdminUserID: imAdminUserID, validate: validator.New()}
+func NewMessageApi(client msg.MsgClient, userClient *rpcli.UserClient, authClient *rpcli.AuthClient, imAdminUserID []string) MessageApi {
+	return MessageApi{Client: client, userClient: userClient, authClient: authClient, imAdminUserID: imAdminUserID, validate: validator.New()}
 }
 
 func (*MessageApi) SetOptions(options map[string]bool, value bool) {
@@ -219,6 +220,8 @@ func (m *MessageApi) getSendMsgReq(c *gin.Context, req apistruct.SendMsg) (sendM
 		data = &apistruct.CustomElem{}
 	case constant.MarkdownText:
 		data = &apistruct.MarkdownTextElem{}
+	case constant.Stream:
+		data = &apistruct.StreamMsgElem{}
 	case constant.Quote:
 		data = &apistruct.QuoteElem{}
 	case constant.OANotification:

internal/api/router.go

@@ -253,7 +253,7 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, cf
 		objectGroup.GET("/*name", t.ObjectRedirect)
 	}
 	// Message
-	m := NewMessageApi(msg.NewMsgClient(msgConn), rpcli.NewUserClient(userConn), cfg.Share.IMAdminUser.UserIDs)
+	m := NewMessageApi(msg.NewMsgClient(msgConn), rpcli.NewUserClient(userConn), rpcli.NewAuthClient(authConn), cfg.Share.IMAdminUser.UserIDs)
 	{
 		msgGroup := r.Group("/msg")
 		msgGroup.POST("/newest_seq", m.GetSeq)
@@ -277,6 +277,9 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, cf
 		msgGroup.POST("/send_simple_msg", m.SendSimpleMessage)
 		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)
+		msgGroup.PUT("/append_stream_msg", m.PutStreamMsg)
 	}
 	// Conversation
 	{

internal/api/stream_msg.go

@@ -0,0 +1,216 @@
+package api
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+	"unicode/utf8"
+
+	"github.com/gin-gonic/gin"
+
+	"github.com/openimsdk/protocol/constant"
+	"github.com/openimsdk/protocol/msg"
+	"github.com/openimsdk/tools/a2r"
+	"github.com/openimsdk/tools/apiresp"
+	"github.com/openimsdk/tools/errs"
+	"github.com/openimsdk/tools/log"
+)
+
+func (m *MessageApi) GetStreamMsg(c *gin.Context) {
+	a2r.Call(c, msg.MsgClient.GetStreamMsg, m.Client)
+}
+
+func (m *MessageApi) AppendStreamMsg(c *gin.Context) {
+	a2r.Call(c, msg.MsgClient.AppendStreamMsg, m.Client)
+}
+
+func (m *MessageApi) PutStreamMsg(c *gin.Context) {
+	var (
+		conversationID string
+		clientMsgID    string
+	)
+	{
+		operationID := c.GetHeader(constant.OperationID)
+		if operationID == "" {
+			operationID = c.Query(constant.OperationID)
+		}
+		if operationID == "" {
+			m.putErr(c, errs.ErrArgs.WrapMsg("operationID is empty"))
+			return
+		}
+		c.Set(constant.OperationID, operationID)
+		conversationID = c.Query("conversationID")
+		if conversationID == "" {
+			conversationID = c.GetHeader("conversationID")
+		}
+		if conversationID == "" {
+			m.putErr(c, errs.ErrArgs.WrapMsg("conversationID is empty"))
+			return
+		}
+		clientMsgID = c.Query("clientMsgID")
+		if clientMsgID == "" {
+			clientMsgID = c.GetHeader("clientMsgID")
+		}
+		if clientMsgID == "" {
+			m.putErr(c, errs.ErrArgs.WrapMsg("clientMsgID is empty"))
+			return
+		}
+		token := c.GetHeader("token")
+		if token == "" {
+			token = c.Query("token")
+		}
+		if token == "" {
+			m.putErr(c, errs.ErrTokenInvalid.WrapMsg("token is empty"))
+			return
+		}
+		resp, err := m.authClient.ParseToken(c, token)
+		if err != nil {
+			m.putErr(c, err)
+			return
+		}
+		c.Set(constant.OpUserPlatform, constant.PlatformIDToName(int(resp.PlatformID)))
+		c.Set(constant.OpUserID, resp.UserID)
+	}
+	done := make(chan struct{})
+	streamCh := make(chan string, 8)
+
+	go func() {
+		defer func() {
+			close(streamCh)
+			c.Request.Body.Close()
+		}()
+		buf := make([]byte, 256)
+		body := NewUTF8Reader(c.Request.Body)
+		for i := 1; ; i++ {
+			n, err := body.Read(buf)
+			if n > 0 {
+				select {
+				case streamCh <- string(buf[:n]):
+				case <-done:
+					return
+				}
+			}
+			if err != nil {
+				if err == io.EOF {
+					log.ZDebug(c, "read request body stream msg done", "clientMsgID", clientMsgID)
+				} else {
+					log.ZError(c, "read request body stream msg failed", err, "clientMsgID", clientMsgID, "error", err)
+				}
+				return
+			}
+			if n < 10 {
+				time.Sleep(time.Millisecond * 10)
+			}
+		}
+	}()
+
+	var (
+		packet   []string
+		end      bool
+		index    int
+		errCount int
+		lastErr  error
+	)
+	defer func() {
+		close(done)
+		if lastErr == nil {
+			apiresp.GinSuccess(c, nil)
+		} else {
+			m.putErr(c, lastErr)
+		}
+	}()
+	doAppend := func() {
+		if end == false && len(packet) == 0 {
+			return
+		}
+		ctx, cancel := context.WithTimeout(c, time.Second*10)
+		defer cancel()
+		req := &msg.AppendStreamMsgReq{
+			ConversationID: conversationID,
+			ClientMsgID:    clientMsgID,
+			StartIndex:     int64(index),
+			Packets:        packet,
+			End:            end,
+		}
+		_, lastErr = m.Client.AppendStreamMsg(ctx, req)
+		if lastErr == nil {
+			log.ZDebug(ctx, "AppendStreamMsg ok", "clientMsgID", clientMsgID)
+			index += len(packet)
+			packet = packet[:0]
+			errCount = 0
+			return
+		}
+		errCount++
+		if errs.ErrRecordNotFound.Is(lastErr) {
+			log.ZWarn(c, "msg not found", nil, "clientMsgID", clientMsgID)
+			return
+		} else if errs.ErrNoPermission.Is(lastErr) {
+			log.ZError(c, "msg permission error", nil, "clientMsgID", clientMsgID)
+			return
+		} else {
+			log.ZError(c, "append stream msg failed", lastErr, "clientMsgID", clientMsgID, "errCount", errCount)
+			time.Sleep(time.Millisecond * 50 * time.Duration(errCount))
+		}
+	}
+	for errCount < 10 {
+		select {
+		case s, ok := <-streamCh:
+			if ok {
+				packet = append(packet, s)
+			}
+			if !ok {
+				end = true
+			}
+			doAppend()
+			if end == true && lastErr == nil {
+				return
+			}
+		}
+	}
+}
+
+func NewUTF8Reader(r io.Reader) io.Reader {
+	return &UTF8Reader{
+		r: bufio.NewReaderSize(r, 512),
+	}
+}
+
+type UTF8Reader struct {
+	r   *bufio.Reader
+	buf bytes.Buffer
+}
+
+func (r *UTF8Reader) Read(b []byte) (int, error) {
+	for {
+		n, err := r.r.Read(b)
+		if err != nil {
+			return 0, err
+		}
+		r.buf.Write(b[:n])
+		data := r.buf.Bytes()
+		minIndex := min(len(b), len(data))
+		if minIndex == 0 {
+			continue
+		}
+		for i := minIndex; i > 0; i-- {
+			if utf8.Valid(data[:i]) {
+				n, err := r.buf.Read(b[:i])
+				if err != nil {
+					return 0, err
+				}
+				if n != i {
+					return 0, fmt.Errorf("invalid UTF-8 encoding")
+				}
+				return n, nil
+			}
+		}
+	}
+}
+
+func (m *MessageApi) putErr(c *gin.Context, err error) {
+	c.JSON(http.StatusOK, apiresp.ParseError(err))
+}

internal/rpc/msg/modify.go

@@ -0,0 +1,167 @@
+package msg
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
+	"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"
+	msgpb "github.com/openimsdk/protocol/msg"
+	"github.com/openimsdk/protocol/sdkws"
+	"github.com/openimsdk/tools/errs"
+	"github.com/openimsdk/tools/log"
+	"github.com/openimsdk/tools/mcontext"
+	"github.com/openimsdk/tools/utils/datautil"
+)
+
+func (m *msgServer) getModifyRawMessage(ctx context.Context, req *msgpb.ModifyMessageReq) (*model.MsgDataModel, error) {
+	opUserID := mcontext.GetOpUserID(ctx)
+	msgs, err := m.MsgDatabase.GetMessageBySeqsDB(ctx, req.ConversationID, opUserID, []int64{req.Seq})
+	if err != nil {
+		return nil, err
+	}
+	if len(msgs) == 0 {
+		return nil, errs.ErrRecordNotFound.WrapMsg("msg seq not found")
+	}
+	val := msgs[0]
+	if val == nil || val.Msg == nil || val.Msg.Status == constant.MsgStatusHasDeleted {
+		return nil, servererrs.ErrRecordNotFound.WrapMsg("msg already delete")
+	}
+	if val.Revoke != nil {
+		return nil, servererrs.ErrMsgAlreadyRevoke.WrapMsg("msg already revoke")
+	}
+	msgData := val.Msg
+	if req.OldContent != "" {
+		if req.OldContent != msgData.Content {
+			return nil, servererrs.ErrArgs.WrapMsg("old msg content not match")
+		}
+	}
+	if req.NewContent == msgData.Content {
+		return nil, errs.ErrArgs.WrapMsg("new content same as old content")
+	}
+	if datautil.Contain(opUserID, m.config.Share.IMAdminUser.UserIDs...) {
+		return msgData, nil
+	}
+	isGroup := msgprocessor.IsGroupConversationID(req.ConversationID)
+	if !isGroup {
+		if msgData.SendID != opUserID {
+			return nil, servererrs.ErrNoPermission.WrapMsg("no permission")
+		}
+		return msgData, nil
+	}
+	groupID := msgData.GroupID
+	if groupID == "" {
+		groupID = msgData.RecvID
+	}
+	groupInfo, err := m.GroupLocalCache.GetGroupInfo(ctx, groupID)
+	if err != nil {
+		return nil, err
+	}
+	if groupInfo.Status == constant.GroupStatusDismissed {
+		return nil, servererrs.ErrDismissedAlready.Wrap()
+	}
+	var memberUserIDs []string
+	if msgData.SendID == opUserID {
+		memberUserIDs = []string{opUserID}
+	} else {
+		memberUserIDs = []string{opUserID, msgData.SendID}
+	}
+	members, err := m.GroupLocalCache.GetGroupMemberInfoMap(ctx, groupID, memberUserIDs)
+	if err != nil {
+		return nil, err
+	}
+	opMember, ok := members[opUserID]
+	if !ok {
+		return nil, servererrs.ErrNoPermission.WrapMsg("opUser no in group")
+	}
+	if msgData.SendID == opUserID {
+		return msgData, nil
+	}
+	if opMember.RoleLevel <= constant.GroupOrdinaryUsers {
+		return nil, errs.ErrNoPermission.WrapMsg("no permission update other user msg")
+	}
+	var sendRoleLevel int32
+	if sendMember, ok := members[msgData.SendID]; ok {
+		sendRoleLevel = sendMember.RoleLevel
+	}
+	if sendRoleLevel >= opMember.RoleLevel {
+		return nil, errs.ErrNoPermission.WrapMsg("no permission update other user msg")
+	}
+	return msgData, nil
+}
+
+func (m *msgServer) ModifyMessage(ctx context.Context, req *msgpb.ModifyMessageReq) (*msgpb.ModifyMessageResp, error) {
+	lockKey := fmt.Sprintf("MODIFYMESSAGE:%s:%d", req.ConversationID, req.Seq)
+	lockValue, err := m.lock.Lock(ctx, lockKey, time.Second*30)
+	if err != nil {
+		return nil, err
+	}
+	defer m.lock.Unlock(ctx, lockKey, lockValue)
+	msg, err := m.getModifyRawMessage(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+	var attachedInfo map[string]json.RawMessage
+	if msg.AttachedInfo != "" && msg.AttachedInfo != "null" && msg.AttachedInfo != "{}" {
+		if err = json.Unmarshal([]byte(msg.AttachedInfo), &attachedInfo); err != nil {
+			log.ZWarn(ctx, "json.Unmarshal", err, "attachedInfo", msg.AttachedInfo)
+		}
+	}
+	if attachedInfo == nil {
+		attachedInfo = make(map[string]json.RawMessage)
+	}
+	const modifyAttachedKey = "lastModified"
+	type LastModified struct {
+		UserID        string `json:"userID"`        // last modified user ID
+		ModifiedTime  int64  `json:"modifiedTime"`  // last modified time
+		ModifiedCount int64  `json:"modifiedCount"` // last modified count
+	}
+	var modifyValue LastModified
+	if val := attachedInfo[modifyAttachedKey]; len(val) > 0 {
+		if err = json.Unmarshal(val, &modifyValue); err != nil {
+			return nil, errs.WrapMsg(err, "json.Unmarshal modifyValue", "val", val)
+		}
+		if modifyValue.ModifiedCount < 1 {
+			modifyValue.ModifiedCount = 1
+		}
+	}
+	modifyValue.ModifiedCount++
+	modifyValue.ModifiedTime = time.Now().UnixMilli()
+	modifyValue.UserID = mcontext.GetOpUserID(ctx)
+	modifyVal, err := json.Marshal(&modifyValue)
+	if err != nil {
+		return nil, err
+	}
+	attachedInfo[modifyAttachedKey] = modifyVal
+	attached, err := json.Marshal(attachedInfo)
+	if err != nil {
+		return nil, errs.ErrInternalServer.WrapMsg("json.Marshal attachedInfo", "attachedInfo", attachedInfo)
+	}
+	msg.Content = req.NewContent
+	msg.AttachedInfo = string(attached)
+	if err := m.MsgDatabase.UpdateMsg(ctx, req.ConversationID, msg); err != nil {
+		return nil, err
+	}
+	tips := &sdkws.ModifyMsgTips{
+		ConversationID: req.ConversationID,
+		Seq:            req.Seq,
+		ClientMsgID:    msg.ClientMsgID,
+		NewContent:     req.NewContent,
+		ModifiedTime:   modifyValue.ModifiedTime,
+		ModifiedCount:  modifyValue.ModifiedCount,
+		UserID:         modifyValue.UserID,
+	}
+	recvID := msg.GroupID
+	if recvID == "" {
+		recvID = msg.RecvID
+	}
+	m.notificationSender.NotificationWithSessionType(ctx, msg.SendID, recvID, constant.ModifyMessageNotification, msg.SessionType, tips)
+	return &msgpb.ModifyMessageResp{
+		ModifiedTime:  modifyValue.ModifiedTime,
+		ModifiedCount: modifyValue.ModifiedCount,
+	}, nil
+}

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

@@ -54,6 +54,11 @@ func (m *msgServer) SendMsg(ctx context.Context, req *pbmsg.SendMsgReq) (*pbmsg.
 
 func (m *msgServer) sendMsg(ctx context.Context, req *pbmsg.SendMsgReq, before **sdkws.MsgData) (*pbmsg.SendMsgResp, error) {
 	m.encapsulateMsgData(req.MsgData)
+	if req.MsgData.ContentType == constant.Stream {
+		if err := m.createStreamMsgHandler(ctx, req.MsgData); err != nil {
+			return nil, err
+		}
+	}
 	switch req.MsgData.SessionType {
 	case constant.SingleChatType:
 		return m.sendMsgSingleChat(ctx, req, before)

internal/rpc/msg/server.go

@@ -24,6 +24,7 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo"
@@ -69,6 +70,8 @@ type msgServer struct {
 	config                 *Config                          // Global configuration settings.
 	webhookClient          *webhook.Client
 	conversationClient     *rpcli.ConversationClient
+	lock                   cache.Lock
+	StreamMsgDatabase      controller.StreamMsgDatabase
 
 	adminUserIDs []string
 }
@@ -139,6 +142,8 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 		config:                 config,
 		webhookClient:          webhook.NewWebhookClient(config.WebhooksConfig.URL),
 		conversationClient:     conversationClient,
+		lock:                   redis.NewLock(rdb),
+		StreamMsgDatabase:      controller.NewStreamMsgDatabase(redis.NewStreamMsg(rdb)),
 		adminUserIDs:           config.Share.IMAdminUser.UserIDs,
 	}
 

internal/rpc/msg/stream_msg.go

@@ -0,0 +1,193 @@
+package msg
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/apistruct"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
+	"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"
+	"github.com/openimsdk/tools/log"
+	"github.com/openimsdk/tools/mcontext"
+)
+
+const (
+	StreamTimeoutEnd            = time.Minute * 10
+	StreamTimeoutEndMillisecond = int64(StreamTimeoutEnd / time.Millisecond)
+)
+
+func (m *msgServer) createStreamMsgHandler(ctx context.Context, msgData *sdkws.MsgData) error {
+	var elem apistruct.StreamMsgElem
+	if err := json.Unmarshal(msgData.Content, &elem); err != nil {
+		return errs.ErrArgs.WrapMsg("stream msg content is invalid", "content", string(msgData.Content))
+	}
+	conversationID := msgprocessor.GetConversationIDByMsg(msgData)
+	if _, err := m.StreamMsgDatabase.GetStreamMsg(ctx, conversationID, msgData.ClientMsgID); err != nil {
+		if !errs.ErrRecordNotFound.Is(err) {
+			return err
+		}
+	}
+	streamMsg := &cache.StreamMsg{
+		SendUserID:    msgData.SendID,
+		RecvID:        msgData.RecvID,
+		SessionType:   msgData.SessionType,
+		UpdateTime:    time.Now().UnixMilli(),
+		StreamType:    elem.Type,
+		StreamContent: elem.Content,
+	}
+	switch msgData.SessionType {
+	case constant.ReadGroupChatType, constant.WriteGroupChatType:
+		streamMsg.RecvID = msgData.GroupID
+	}
+	return m.StreamMsgDatabase.CreateStreamMsg(ctx, msgprocessor.GetConversationIDByMsg(msgData), msgData.ClientMsgID, streamMsg)
+}
+
+func (m *msgServer) AppendStreamMsg(ctx context.Context, req *msg.AppendStreamMsgReq) (*msg.AppendStreamMsgResp, error) {
+	end, err := m.StreamMsgDatabase.GetStreamMsgEnd(ctx, req.ConversationID, req.ClientMsgID)
+	if err != nil {
+		return nil, err
+	}
+	if end {
+		return nil, errs.ErrNoPermission.WrapMsg("stream msg is end")
+	}
+	res, err := m.StreamMsgDatabase.AppendStreamMsg(ctx, req.ConversationID, req.ClientMsgID, int(req.StartIndex), req.Packets, req.End, req.End)
+	if err != nil {
+		return nil, err
+	}
+	tips := &sdkws.StreamMsgTips{
+		ConversationID: req.ConversationID,
+		ClientMsgID:    req.ClientMsgID,
+		StartIndex:     req.StartIndex,
+		Packets:        req.Packets,
+		End:            req.End,
+	}
+	m.msgNotificationSender.StreamMsgNotification(ctx, res.SendUserID, res.RecvID, res.SessionType, tips)
+	if req.End {
+		m.modifyStreamMessage(ctx, req.ConversationID, req.ClientMsgID, res)
+	}
+	return &msg.AppendStreamMsgResp{}, nil
+}
+
+func (m *msgServer) modifyStreamMessage(ctx context.Context, conversationID string, clientMsgID string, res *cache.StreamMsg) {
+	packets := make([]string, 0, len(res.Packets))
+	for i := int64(0); ; i++ {
+		data, ok := res.Packets[i]
+		if !ok {
+			break
+		}
+		packets = append(packets, data)
+	}
+	content, err := json.Marshal(&apistruct.StreamMsgElem{
+		Type:     res.StreamType,
+		Content:  res.StreamContent,
+		Packets:  packets,
+		End:      res.End,
+		Deadline: res.UpdateTime,
+	})
+	if err != nil {
+		log.ZError(ctx, "modifyStreamMessage json.Marshal", err, "conversationID", conversationID, "clientMsgID", clientMsgID)
+		return
+	}
+	req := &msg.ModifyMessageReq{
+		ConversationID: conversationID,
+		NewContent:     string(content),
+	}
+	modifyMessage := func() error {
+		ctx, cancel := context.WithTimeout(ctx, time.Second*10)
+		defer cancel()
+		if req.Seq == 0 {
+			req.Seq, err = m.MsgDatabase.GetMessageSeq(ctx, conversationID, clientMsgID)
+			if err != nil {
+				return err
+			}
+		}
+		if _, err := m.ModifyMessage(ctx, req); err != nil {
+			return err
+		}
+		return nil
+	}
+
+	if err := modifyMessage(); err != nil {
+		log.ZError(ctx, "sync modifyStreamMessage", err, "conversationID", conversationID, "content", string(content))
+		ctx = context.WithoutCancel(ctx)
+		go func() {
+			for i := 1; i <= 10; i++ {
+				if err := modifyMessage(); err == nil {
+					log.ZDebug(ctx, "async modifyStreamMessage success", "conversationID", conversationID, "content", string(content), "count", i)
+					return
+				} else {
+					log.ZError(ctx, "modifyStreamMessage", err, "conversationID", conversationID, "content", string(content), "count", i)
+					time.Sleep(time.Second * time.Duration(i))
+				}
+			}
+		}()
+	}
+}
+
+func (m *msgServer) GetStreamMsg(ctx context.Context, req *msg.GetStreamMsgReq) (*msg.GetStreamMsgResp, error) {
+	value, err := m.StreamMsgDatabase.GetStreamMsg(ctx, req.ConversationID, req.ClientMsgID)
+	if err == nil {
+		resp := msg.GetStreamMsgResp{
+			UserID:  value.SendUserID,
+			Packets: make([]string, 0, len(value.Packets)),
+			End:     value.End,
+		}
+		for i := int64(0); ; i++ {
+			data, ok := value.Packets[i]
+			if !ok {
+				break
+			}
+			resp.Packets = append(resp.Packets, data)
+		}
+		if resp.End {
+			resp.DeadlineTime = value.UpdateTime
+		} else {
+			if now := time.Now().UnixMilli(); now-value.UpdateTime >= StreamTimeoutEndMillisecond {
+				resp.DeadlineTime = now + StreamTimeoutEndMillisecond
+				resp.End = true
+			}
+		}
+		return &resp, nil
+	} else if !errs.ErrRecordNotFound.Is(err) {
+		return nil, err
+	}
+	if req.Seq <= 0 || errs.ErrRecordNotFound.Is(err) == false {
+		return nil, err
+	}
+	msgs, err := m.MsgDatabase.GetMessageBySeqs(ctx, req.ConversationID, mcontext.GetOpUserID(ctx), []int64{req.Seq})
+	if err != nil {
+		return nil, err
+	}
+	if len(msgs) == 0 || msgs[0] == nil {
+		return nil, errs.ErrRecordNotFound.WrapMsg("stream message not found")
+	}
+	msgData := msgs[0]
+	if msgData.ClientMsgID != req.ClientMsgID {
+		return nil, errs.ErrRecordNotFound.WrapMsg("stream message id not match")
+	}
+	if msgData.ContentType != constant.Stream {
+		return nil, errs.ErrNoPermission.WrapMsg("stream message content type not match")
+	}
+	var elem apistruct.StreamMsgElem
+	if len(msgData.Content) > 0 {
+		if err := json.Unmarshal(msgData.Content, &elem); err != nil {
+			log.ZError(ctx, "stream msg unmarshal", err, "content", string(msgData.Content), "conversationID", req.ConversationID, "seq", req.Seq)
+		}
+	}
+	resp := &msg.GetStreamMsgResp{
+		UserID:       msgData.SendID,
+		Packets:      elem.Packets,
+		End:          elem.End,
+		DeadlineTime: elem.Deadline,
+	}
+	if !resp.End {
+		resp.End = true
+		resp.DeadlineTime = msgData.SendTime + StreamTimeoutEndMillisecond
+	}
+	return resp, nil
+}

pkg/apistruct/msg.go

@@ -90,8 +90,11 @@ type MarkdownTextElem struct {
 }
 
 type StreamMsgElem struct {
-	Type    string `mapstructure:"type" validate:"required"`
-	Content string `mapstructure:"content" validate:"required"`
+	Type     string   `mapstructure:"type" json:"type"`
+	Content  string   `mapstructure:"content" json:"content"`
+	Packets  []string `mapstructure:"packets" json:"packets"`
+	End      bool     `mapstructure:"end" json:"end"`
+	Deadline int64    `mapstructure:"deadline" json:"deadline"`
 }
 
 type RevokeElem struct {

pkg/common/storage/cache/cachekey/msg.go

@@ -21,6 +21,7 @@ import (
 const (
 	sendMsgFailedFlag = "SEND_MSG_FAILED_FLAG:"
 	messageCache      = "MSG_CACHE:"
+	messageSeq        = "MSG_SEQ:"
 )
 
 func GetMsgCacheKey(conversationID string, seq int64) string {
@@ -30,3 +31,7 @@ func GetMsgCacheKey(conversationID string, seq int64) string {
 func GetSendMsgKey(id string) string {
 	return sendMsgFailedFlag + id
 }
+
+func GetMsgSeqKey(conversationID string, clientMsgID string) string {
+	return messageSeq + conversationID + ":" + clientMsgID
+}

pkg/common/storage/cache/cachekey/stream_msg.go

@@ -0,0 +1,7 @@
+package cachekey
+
+const streamMessageCache = "STREAM_MSG:"
+
+func GetStreamMsgKey(conversationID string, clientMsgID string) string {
+	return streamMessageCache + conversationID + ":" + clientMsgID
+}

pkg/common/storage/cache/lock.go

@@ -0,0 +1,11 @@
+package cache
+
+import (
+	"context"
+	"time"
+)
+
+type Lock interface {
+	Lock(ctx context.Context, key string, timeout time.Duration) (string, error)
+	Unlock(ctx context.Context, key, value string)
+}

pkg/common/storage/cache/msg.go

@@ -16,6 +16,7 @@ package cache
 
 import (
 	"context"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 )
 
@@ -26,4 +27,5 @@ type MsgCache interface {
 	GetMessageBySeqs(ctx context.Context, conversationID string, seqs []int64) ([]*model.MsgInfoModel, error)
 	DelMessageBySeqs(ctx context.Context, conversationID string, seqs []int64) error
 	SetMessageBySeqs(ctx context.Context, conversationID string, msgs []*model.MsgInfoModel) error
+	GetMessageSeq(ctx context.Context, conversationID string, clientMsgID string) (int64, error)
 }

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

@@ -0,0 +1,60 @@
+package redis
+
+import (
+	"context"
+	"encoding/hex"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
+	"github.com/openimsdk/tools/errs"
+	"github.com/openimsdk/tools/log"
+	"github.com/redis/go-redis/v9"
+)
+
+const lockPrefix = "LOCK:"
+
+func NewLock(rdb redis.UniversalClient) cache.Lock {
+	return &redisLock{rdb: rdb}
+}
+
+type redisLock struct {
+	rdb redis.UniversalClient
+}
+
+func (x *redisLock) Lock(ctx context.Context, key string, timeout time.Duration) (string, error) {
+	uid, err := uuid.NewUUID()
+	if err != nil {
+		return "", err
+	}
+	if timeout < time.Second {
+		timeout = time.Minute * 2
+	}
+	value := hex.EncodeToString(uid[:])
+	key = lockPrefix + key
+	for {
+		ok, err := x.rdb.SetNX(ctx, key, value, timeout).Result()
+		if err != nil {
+			return "", errs.WrapMsg(err, "get redis lock", "key", key)
+		}
+		if ok {
+			return value, nil
+		}
+		timer := time.NewTimer(50 * time.Millisecond)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return "", context.Cause(ctx)
+		case <-timer.C:
+		}
+	}
+}
+
+func (x *redisLock) Unlock(ctx context.Context, key, value string) {
+	ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
+	defer cancel()
+	script := "\nlocal value = redis.call(\"GET\", KEYS[1])\nif value == ARGV[1] then\n  return redis.call(\"DEL\", KEYS[1])\nend\nreturn 0"
+	if err := x.rdb.Eval(ctx, script, []string{lockPrefix + key}, value).Err(); err != nil {
+		log.ZWarn(ctx, "unlock redis lock", err, "key", key)
+	}
+}

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

@@ -9,6 +9,7 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
 	"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/protocol/constant"
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/utils/datautil"
 	"github.com/redis/go-redis/v9"
@@ -89,6 +90,15 @@ func (c *msgCache) SetMessageBySeqs(ctx context.Context, conversationID string,
 		if err := c.rcClient.GetClient().RawSet(ctx, cachekey.GetMsgCacheKey(conversationID, msg.Msg.Seq), string(data), msgCacheTimeout); err != nil {
 			return err
 		}
+		if msg.Msg.ContentType == constant.Stream {
+			if err := c.rcClient.GetRedis().Set(ctx, cachekey.GetMsgSeqKey(conversationID, msg.Msg.ClientMsgID), msg.Msg.Seq, msgCacheTimeout).Err(); err != nil {
+				return err
+			}
+		}
 	}
 	return nil
 }
+
+func (c *msgCache) GetMessageSeq(ctx context.Context, conversationID string, clientMsgID string) (int64, error) {
+	return c.rcClient.GetRedis().Get(ctx, cachekey.GetMsgSeqKey(conversationID, clientMsgID)).Int64()
+}

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

@@ -0,0 +1,140 @@
+package redis
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
+	"github.com/openimsdk/tools/errs"
+	"github.com/redis/go-redis/v9"
+)
+
+func NewStreamMsg(rdb redis.UniversalClient) cache.StreamMsgCache {
+	return &streamMsg{rdb: rdb}
+}
+
+type streamMsg struct {
+	rdb redis.UniversalClient
+}
+
+func (x *streamMsg) getMsgKey(conversationID string, clientMsgID string) string {
+	return cachekey.GetStreamMsgKey(conversationID, clientMsgID)
+}
+
+func (x *streamMsg) CreateStreamMsg(ctx context.Context, conversationID string, clientMsgID string, msg *cache.StreamMsg) error {
+	key := x.getMsgKey(conversationID, clientMsgID)
+	pipeline := x.rdb.Pipeline()
+	pipeline.HSet(ctx, key, "sendUserID", msg.SendUserID)
+	pipeline.HSet(ctx, key, "recvID", msg.RecvID)
+	pipeline.HSet(ctx, key, "sessionType", strconv.Itoa(int(msg.SessionType)))
+	pipeline.HSet(ctx, key, "updateTime", time.Now().UnixMilli())
+	pipeline.HSet(ctx, key, "isEnd", false)
+	pipeline.HSet(ctx, key, "streamType", msg.StreamType)
+	pipeline.HSet(ctx, key, "streamContent", msg.StreamContent)
+	pipeline.Expire(ctx, key, 24*time.Hour)
+	_, err := pipeline.Exec(ctx)
+	return err
+}
+
+func (x *streamMsg) AppendStreamMsg(ctx context.Context, conversationID string, clientMsgID string, startIndex int, packets []string, end bool, retPacket bool) (*cache.StreamMsg, error) {
+	key := x.getMsgKey(conversationID, clientMsgID)
+	var mapCmd *redis.MapStringStringCmd
+	var sliceCmd *redis.SliceCmd
+	pipeline := x.rdb.Pipeline()
+	for i, packet := range packets {
+		pipeline.HSet(ctx, key, "i_"+strconv.Itoa(startIndex+i), packet)
+	}
+	pipeline.HSet(ctx, key, "isEnd", end)
+	pipeline.HSet(ctx, key, "updateTime", time.Now().UnixMilli())
+	pipeline.Expire(ctx, key, 24*time.Hour)
+	if retPacket {
+		mapCmd = pipeline.HGetAll(ctx, key)
+	} else {
+		sliceCmd = pipeline.HMGet(ctx, key, "sendUserID", "recvID", "sessionType")
+	}
+	if _, err := pipeline.Exec(ctx); err != nil {
+		return nil, err
+	}
+	var data map[string]string
+	var err error
+	if retPacket {
+		data, err = mapCmd.Result()
+	} else {
+		arr, resultErr := sliceCmd.Result()
+		if resultErr != nil {
+			return nil, resultErr
+		}
+		if len(arr) != 3 || arr[0] == nil || arr[1] == nil || arr[2] == nil {
+			return nil, errs.ErrRecordNotFound.WrapMsg("stream message not found")
+		}
+		data = map[string]string{
+			"sendUserID": fmt.Sprint(arr[0]), "recvID": fmt.Sprint(arr[1]), "sessionType": fmt.Sprint(arr[2]),
+		}
+	}
+	if err != nil {
+		return nil, err
+	}
+	return x.mapToStreamMsg(data, retPacket)
+}
+
+func (x *streamMsg) GetStreamMsg(ctx context.Context, conversationID string, clientMsgID string) (*cache.StreamMsg, error) {
+	data, err := x.rdb.HGetAll(ctx, x.getMsgKey(conversationID, clientMsgID)).Result()
+	if err != nil {
+		return nil, err
+	}
+	if len(data) == 0 {
+		return nil, errs.ErrRecordNotFound.WrapMsg("stream message not found")
+	}
+	return x.mapToStreamMsg(data, true)
+}
+
+func (x *streamMsg) mapToStreamMsg(data map[string]string, full bool) (*cache.StreamMsg, error) {
+	sessionType, err := strconv.ParseInt(data["sessionType"], 10, 32)
+	if err != nil {
+		return nil, err
+	}
+	if !full {
+		return &cache.StreamMsg{SendUserID: data["sendUserID"], RecvID: data["recvID"], SessionType: int32(sessionType)}, nil
+	}
+	end, err := strconv.ParseBool(data["isEnd"])
+	if err != nil {
+		return nil, err
+	}
+	updateTime, err := strconv.ParseInt(data["updateTime"], 10, 64)
+	if err != nil {
+		return nil, err
+	}
+	msg := cache.StreamMsg{
+		SendUserID: data["sendUserID"], RecvID: data["recvID"], SessionType: int32(sessionType),
+		StreamType: data["streamType"], StreamContent: data["streamContent"], UpdateTime: updateTime,
+		Packets: make(map[int64]string), End: end,
+	}
+	var maxIndex int64 = -1
+	for indexStr, value := range data {
+		if !strings.HasPrefix(indexStr, "i_") {
+			continue
+		}
+		index, err := strconv.ParseInt(strings.TrimPrefix(indexStr, "i_"), 10, 64)
+		if err != nil || index < 0 {
+			return nil, errs.ErrInternalServer.WrapMsg("packet index is invalid", "index", indexStr)
+		}
+		msg.Packets[index] = value
+		if maxIndex < index {
+			maxIndex = index
+		}
+	}
+	for i := int64(0); i <= maxIndex; i++ {
+		if _, ok := msg.Packets[i]; !ok {
+			return nil, errs.ErrInternalServer.WrapMsg("packet index is not continuous", "index", i)
+		}
+	}
+	return &msg, nil
+}
+
+func (x *streamMsg) GetStreamMsgEnd(ctx context.Context, conversationID string, clientMsgID string) (bool, error) {
+	return x.rdb.HGet(ctx, x.getMsgKey(conversationID, clientMsgID), "isEnd").Bool()
+}

pkg/common/storage/cache/stream_msg.go

@@ -0,0 +1,21 @@
+package cache
+
+import "context"
+
+type StreamMsg struct {
+	SendUserID    string
+	RecvID        string
+	SessionType   int32
+	StreamType    string
+	StreamContent string
+	Packets       map[int64]string
+	End           bool
+	UpdateTime    int64
+}
+
+type StreamMsgCache interface {
+	CreateStreamMsg(ctx context.Context, conversationID string, clientMsgID string, msg *StreamMsg) error
+	AppendStreamMsg(ctx context.Context, conversationID string, clientMsgID string, startIndex int, packets []string, end bool, retPacket bool) (*StreamMsg, error)
+	GetStreamMsg(ctx context.Context, conversationID string, clientMsgID string) (*StreamMsg, error)
+	GetStreamMsgEnd(ctx context.Context, conversationID string, clientMsgID string) (bool, error)
+}

pkg/common/storage/controller/msg.go

@@ -101,6 +101,11 @@ type CommonMsgDatabase interface {
 	GetLastMessageSeqByTime(ctx context.Context, conversationID string, time int64) (int64, error)
 
 	GetLastMessage(ctx context.Context, conversationIDS []string, userID string) (map[string]*sdkws.MsgData, error)
+
+	GetMessageBySeqs(ctx context.Context, conversationID string, userID string, seqs []int64) ([]*sdkws.MsgData, error)
+	GetMessageBySeqsDB(ctx context.Context, conversationID string, userID string, seqs []int64) ([]*model.MsgInfoModel, error)
+	GetMessageSeq(ctx context.Context, conversationID string, clientMsgID string) (int64, error)
+	UpdateMsg(ctx context.Context, conversationID string, msg *model.MsgDataModel) error
 }
 
 func NewCommonMsgDatabase(msgDocModel database.Msg, msg cache.MsgCache, seqUser cache.SeqUser, seqConversation cache.SeqConversationCache, producer mq.Producer) CommonMsgDatabase {
@@ -822,6 +827,29 @@ func (db *commonMsgDatabase) GetMessageBySeqs(ctx context.Context, conversationI
 	return res, nil
 }
 
+func (db *commonMsgDatabase) GetMessageBySeqsDB(ctx context.Context, conversationID string, userID string, seqs []int64) ([]*model.MsgInfoModel, error) {
+	msgs, err := db.msgCache.GetMessageBySeqs(ctx, conversationID, seqs)
+	if err != nil {
+		return nil, err
+	}
+	db.handlerDeleteAndRevoked(ctx, userID, msgs)
+	db.handlerQuote(ctx, userID, conversationID, msgs)
+	return msgs, nil
+}
+
+func (db *commonMsgDatabase) GetMessageSeq(ctx context.Context, conversationID string, clientMsgID string) (int64, error) {
+	return db.msgCache.GetMessageSeq(ctx, conversationID, clientMsgID)
+}
+
+func (db *commonMsgDatabase) UpdateMsg(ctx context.Context, conversationID string, msg *model.MsgDataModel) error {
+	docID := db.msgTable.GetDocID(conversationID, msg.Seq)
+	index := db.msgTable.GetMsgIndex(msg.Seq)
+	if _, err := db.msgDocDatabase.UpdateMsg(ctx, docID, index, "msg", msg); err != nil {
+		return err
+	}
+	return db.msgCache.DelMessageBySeqs(ctx, conversationID, []int64{msg.Seq})
+}
+
 func (db *commonMsgDatabase) GetLastMessage(ctx context.Context, conversationIDs []string, userID string) (map[string]*sdkws.MsgData, error) {
 	res := make(map[string]*sdkws.MsgData)
 	for _, conversationID := range conversationIDs {

pkg/common/storage/controller/stream_msg.go

@@ -0,0 +1,15 @@
+package controller
+
+import "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
+
+type StreamMsgDatabase interface {
+	cache.StreamMsgCache
+}
+
+func NewStreamMsgDatabase(db cache.StreamMsgCache) StreamMsgDatabase {
+	return &streamMsgDatabase{db}
+}
+
+type streamMsgDatabase struct {
+	cache.StreamMsgCache
+}

pkg/notification/msg.go

@@ -74,9 +74,10 @@ func newContentTypeConf(conf *config.Notification) map[int32]config.Notification
 		constant.ConversationUnreadNotification:      conf.ConversationChanged,
 		constant.ConversationPrivateChatNotification: conf.ConversationSetPrivate,
 		// msg
-		constant.MsgRevokeNotification:  {IsSendMsg: false, ReliabilityLevel: constant.ReliableNotificationNoMsg},
-		constant.HasReadReceipt:         {IsSendMsg: false, ReliabilityLevel: constant.ReliableNotificationNoMsg},
-		constant.DeleteMsgsNotification: {IsSendMsg: false, ReliabilityLevel: constant.ReliableNotificationNoMsg},
+		constant.MsgRevokeNotification:     {IsSendMsg: false, ReliabilityLevel: constant.ReliableNotificationNoMsg},
+		constant.HasReadReceipt:            {IsSendMsg: false, ReliabilityLevel: constant.ReliableNotificationNoMsg},
+		constant.DeleteMsgsNotification:    {IsSendMsg: false, ReliabilityLevel: constant.ReliableNotificationNoMsg},
+		constant.ModifyMessageNotification: {IsSendMsg: false, ReliabilityLevel: constant.ReliableNotificationNoMsg},
 	}
 }