// Copyright © 2023 OpenIM. All rights reserved. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package msg import ( "context" "github.com/openimsdk/open-im-server/v3/pkg/msgprocessor" "github.com/OpenIMSDK/protocol/constant" pbconversation "github.com/OpenIMSDK/protocol/conversation" pbmsg "github.com/OpenIMSDK/protocol/msg" "github.com/OpenIMSDK/protocol/sdkws" "github.com/OpenIMSDK/protocol/wrapperspb" "github.com/OpenIMSDK/tools/errs" "github.com/OpenIMSDK/tools/log" "github.com/OpenIMSDK/tools/mcontext" "github.com/OpenIMSDK/tools/utils" promepkg "github.com/openimsdk/open-im-server/v3/pkg/common/prome" ) func (m *msgServer) SendMsg(ctx context.Context, req *pbmsg.SendMsgReq) (resp *pbmsg.SendMsgResp, error error) { resp = &pbmsg.SendMsgResp{} if req.MsgData != nil { flag := isMessageHasReadEnabled(req.MsgData) if !flag { return nil, errs.ErrMessageHasReadDisable.Wrap() } m.encapsulateMsgData(req.MsgData) switch req.MsgData.SessionType { case constant.SingleChatType: return m.sendMsgSingleChat(ctx, req) case constant.NotificationChatType: return m.sendMsgNotification(ctx, req) case constant.SuperGroupChatType: return m.sendMsgSuperGroupChat(ctx, req) default: return nil, errs.ErrArgs.Wrap("unknown sessionType") } } else { return nil, errs.ErrArgs.Wrap("msgData is nil") } } func (m *msgServer) sendMsgSuperGroupChat( ctx context.Context, req *pbmsg.SendMsgReq, ) (resp *pbmsg.SendMsgResp, err error) { promepkg.Inc(promepkg.WorkSuperGroupChatMsgRecvSuccessCounter) if err = m.messageVerification(ctx, req); err != nil { promepkg.Inc(promepkg.WorkSuperGroupChatMsgProcessFailedCounter) return nil, err } if err = callbackBeforeSendGroupMsg(ctx, req); err != nil { return nil, err } if err := callbackMsgModify(ctx, req); err != nil { return nil, err } err = m.MsgDatabase.MsgToMQ(ctx, utils.GenConversationUniqueKeyForGroup(req.MsgData.GroupID), req.MsgData) if err != nil { return nil, err } if req.MsgData.ContentType == constant.AtText { go m.setConversationAtInfo(ctx, req.MsgData) } if err = callbackAfterSendGroupMsg(ctx, req); err != nil { log.ZWarn(ctx, "CallbackAfterSendGroupMsg", err) } promepkg.Inc(promepkg.WorkSuperGroupChatMsgProcessSuccessCounter) resp = &pbmsg.SendMsgResp{} resp.SendTime = req.MsgData.SendTime resp.ServerMsgID = req.MsgData.ServerMsgID resp.ClientMsgID = req.MsgData.ClientMsgID return resp, nil } func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgData) { log.ZDebug(nctx, "setConversationAtInfo", "msg", msg) ctx := mcontext.NewCtx("@@@" + mcontext.GetOperationID(nctx)) var atUserID []string conversation := &pbconversation.ConversationReq{ ConversationID: msgprocessor.GetConversationIDByMsg(msg), ConversationType: msg.SessionType, GroupID: msg.GroupID, } tagAll := utils.IsContain(constant.AtAllString, msg.AtUserIDList) if tagAll { memberUserIDList, err := m.Group.GetGroupMemberIDs(ctx, msg.GroupID) if err != nil { log.ZWarn(ctx, "GetGroupMemberIDs", err) return } atUserID = utils.DifferenceString([]string{constant.AtAllString}, msg.AtUserIDList) if len(atUserID) == 0 { // just @everyone conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAll} } else { //@Everyone and @other people conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAllAtMe} err := m.Conversation.SetConversations(ctx, atUserID, conversation) if err != nil { log.ZWarn(ctx, "SetConversations", err, "userID", atUserID, "conversation", conversation) } memberUserIDList = utils.DifferenceString(atUserID, memberUserIDList) } conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAll} err = m.Conversation.SetConversations(ctx, memberUserIDList, conversation) if err != nil { log.ZWarn(ctx, "SetConversations", err, "userID", memberUserIDList, "conversation", conversation) } } else { conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtMe} err := m.Conversation.SetConversations(ctx, msg.AtUserIDList, conversation) if err != nil { log.ZWarn(ctx, "SetConversations", err, msg.AtUserIDList, conversation) } } } func (m *msgServer) sendMsgNotification( ctx context.Context, req *pbmsg.SendMsgReq, ) (resp *pbmsg.SendMsgResp, err error) { promepkg.Inc(promepkg.SingleChatMsgRecvSuccessCounter) if err := m.MsgDatabase.MsgToMQ(ctx, utils.GenConversationUniqueKeyForSingle(req.MsgData.SendID, req.MsgData.RecvID), req.MsgData); err != nil { promepkg.Inc(promepkg.SingleChatMsgProcessFailedCounter) return nil, err } resp = &pbmsg.SendMsgResp{ ServerMsgID: req.MsgData.ServerMsgID, ClientMsgID: req.MsgData.ClientMsgID, SendTime: req.MsgData.SendTime, } return resp, nil } func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq) (resp *pbmsg.SendMsgResp, err error) { promepkg.Inc(promepkg.SingleChatMsgRecvSuccessCounter) if err := m.messageVerification(ctx, req); err != nil { return nil, err } isSend := true isNotification := msgprocessor.IsNotificationByMsg(req.MsgData) if !isNotification { isSend, err = m.modifyMessageByUserMessageReceiveOpt( ctx, req.MsgData.RecvID, utils.GenConversationIDForSingle(req.MsgData.SendID, req.MsgData.RecvID), constant.SingleChatType, req, ) if err != nil { return nil, err } } if !isSend { promepkg.Inc(promepkg.SingleChatMsgProcessFailedCounter) return nil, nil } else { if err = callbackBeforeSendSingleMsg(ctx, req); err != nil { return nil, err } if err := callbackMsgModify(ctx, req); err != nil { return nil, err } if err := m.MsgDatabase.MsgToMQ(ctx, utils.GenConversationUniqueKeyForSingle(req.MsgData.SendID, req.MsgData.RecvID), req.MsgData); err != nil { promepkg.Inc(promepkg.SingleChatMsgProcessFailedCounter) return nil, err } err = callbackAfterSendSingleMsg(ctx, req) if err != nil { log.ZWarn(ctx, "CallbackAfterSendSingleMsg", err, "req", req) } resp = &pbmsg.SendMsgResp{ ServerMsgID: req.MsgData.ServerMsgID, ClientMsgID: req.MsgData.ClientMsgID, SendTime: req.MsgData.SendTime, } promepkg.Inc(promepkg.SingleChatMsgProcessSuccessCounter) return resp, nil } } func (m *msgServer) BatchSendMsg(ctx context.Context, in *pbmsg.BatchSendMessageReq) (*pbmsg.BatchSendMessageResp, error) { return nil, nil }