diff --git a/go.mod b/go.mod index 8a8dae98fc..e3f52757a5 100644 --- a/go.mod +++ b/go.mod @@ -12,8 +12,8 @@ require ( github.com/gorilla/websocket v1.5.1 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 github.com/mitchellh/mapstructure v1.5.0 - github.com/openimsdk/protocol v0.0.73-alpha.19 - github.com/openimsdk/tools v0.0.50-alpha.121 + github.com/openimsdk/protocol v0.0.73-alpha.20 + github.com/openimsdk/tools v0.0.50-alpha.122 github.com/pkg/errors v0.9.1 // indirect github.com/prometheus/client_golang v1.18.0 github.com/stretchr/testify v1.11.1 diff --git a/go.sum b/go.sum index 5d0c9e0120..8ae6bab5b9 100644 --- a/go.sum +++ b/go.sum @@ -361,10 +361,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.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/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/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.122 h1:bKP6hrJ6kGyGUqTGuyqUqPdZxY7mZSsh86aRQJ/6Jt8= +github.com/openimsdk/tools v0.0.50-alpha.122/go.mod h1:I0WESSa7ghPIo9BL+ETlH/qEIbO6+KZioM1jwNuDwz0= 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= diff --git a/internal/api/msg.go b/internal/api/msg.go index 06fd149362..f93163a651 100644 --- a/internal/api/msg.go +++ b/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: diff --git a/internal/api/router.go b/internal/api/router.go index c5f93c6c0f..0a7f743940 100644 --- a/internal/api/router.go +++ b/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 { diff --git a/internal/api/stream_msg.go b/internal/api/stream_msg.go new file mode 100644 index 0000000000..5df4c40353 --- /dev/null +++ b/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)) +} diff --git a/internal/rpc/msg/modify.go b/internal/rpc/msg/modify.go new file mode 100644 index 0000000000..205375dbb7 --- /dev/null +++ b/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 +} diff --git a/internal/rpc/msg/notification.go b/internal/rpc/msg/notification.go index 0418823d63..0daafbe6cb 100644 --- a/internal/rpc/msg/notification.go +++ b/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) +} diff --git a/internal/rpc/msg/send.go b/internal/rpc/msg/send.go index 18ad9cc563..60d0fe9968 100644 --- a/internal/rpc/msg/send.go +++ b/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) diff --git a/internal/rpc/msg/server.go b/internal/rpc/msg/server.go index e7c589b8ca..cb67ae4bbf 100644 --- a/internal/rpc/msg/server.go +++ b/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, } diff --git a/internal/rpc/msg/stream_msg.go b/internal/rpc/msg/stream_msg.go new file mode 100644 index 0000000000..ee237e0677 --- /dev/null +++ b/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 +} diff --git a/pkg/apistruct/msg.go b/pkg/apistruct/msg.go index 0e1b356a1c..2333ca2dd2 100644 --- a/pkg/apistruct/msg.go +++ b/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 { diff --git a/pkg/common/storage/cache/cachekey/msg.go b/pkg/common/storage/cache/cachekey/msg.go index ac449df38e..6f0eb90d98 100644 --- a/pkg/common/storage/cache/cachekey/msg.go +++ b/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 +} diff --git a/pkg/common/storage/cache/cachekey/stream_msg.go b/pkg/common/storage/cache/cachekey/stream_msg.go new file mode 100644 index 0000000000..613296676a --- /dev/null +++ b/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 +} diff --git a/pkg/common/storage/cache/lock.go b/pkg/common/storage/cache/lock.go new file mode 100644 index 0000000000..b867dc70a3 --- /dev/null +++ b/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) +} diff --git a/pkg/common/storage/cache/msg.go b/pkg/common/storage/cache/msg.go index 271ed19fe3..09e6889567 100644 --- a/pkg/common/storage/cache/msg.go +++ b/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) } diff --git a/pkg/common/storage/cache/redis/lock.go b/pkg/common/storage/cache/redis/lock.go new file mode 100644 index 0000000000..ad4a887d58 --- /dev/null +++ b/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) + } +} diff --git a/pkg/common/storage/cache/redis/msg.go b/pkg/common/storage/cache/redis/msg.go index dfe6ca04df..64e9820995 100644 --- a/pkg/common/storage/cache/redis/msg.go +++ b/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() +} diff --git a/pkg/common/storage/cache/redis/stream_msg.go b/pkg/common/storage/cache/redis/stream_msg.go new file mode 100644 index 0000000000..7f40623603 --- /dev/null +++ b/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() +} diff --git a/pkg/common/storage/cache/stream_msg.go b/pkg/common/storage/cache/stream_msg.go new file mode 100644 index 0000000000..eec55df6bd --- /dev/null +++ b/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) +} diff --git a/pkg/common/storage/controller/msg.go b/pkg/common/storage/controller/msg.go index f833008e8d..26742fefa7 100644 --- a/pkg/common/storage/controller/msg.go +++ b/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 { diff --git a/pkg/common/storage/controller/stream_msg.go b/pkg/common/storage/controller/stream_msg.go new file mode 100644 index 0000000000..96c1c2af56 --- /dev/null +++ b/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 +} diff --git a/pkg/notification/msg.go b/pkg/notification/msg.go index ba8a9185a9..797b68b29a 100644 --- a/pkg/notification/msg.go +++ b/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}, } }