Files
open-im-server/internal/rpc/msg/send_msg.go
T

572 lines
20 KiB
Go
Raw Normal View History

2021-12-23 17:19:57 +08:00
package msg
2021-05-26 19:40:38 +08:00
import (
"Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant"
2021-12-06 20:03:59 +08:00
"Open_IM/pkg/common/db"
2022-08-01 19:13:26 +08:00
rocksCache "Open_IM/pkg/common/db/rocks_cache"
"Open_IM/pkg/common/log"
2023-02-09 14:40:49 +08:00
"Open_IM/pkg/common/tokenverify"
2022-04-28 15:10:42 +08:00
cacheRpc "Open_IM/pkg/proto/cache"
2023-02-10 22:10:37 +08:00
"Open_IM/pkg/proto/msg"
2022-09-19 22:16:45 +08:00
pbPush "Open_IM/pkg/proto/push"
2022-05-11 18:33:48 +08:00
pbRelay "Open_IM/pkg/proto/relay"
2023-02-09 20:36:34 +08:00
sdkws "Open_IM/pkg/proto/sdkws"
"Open_IM/pkg/utils"
2021-05-26 19:40:38 +08:00
"context"
2022-05-11 18:33:48 +08:00
"errors"
2021-05-26 19:40:38 +08:00
"math/rand"
"strconv"
2021-06-28 15:35:18 +08:00
"strings"
2022-04-29 18:47:55 +08:00
"sync"
2021-05-26 19:40:38 +08:00
"time"
2022-08-17 14:57:57 +08:00
go_redis "github.com/go-redis/redis/v8"
2021-05-26 19:40:38 +08:00
)
2022-09-14 17:45:38 +08:00
var (
ExcludeContentType = []int{constant.HasReadReceipt, constant.GroupHasReadReceipt}
)
2022-08-05 19:10:08 +08:00
2022-09-29 17:57:26 +08:00
type Validator interface {
2023-02-10 22:10:37 +08:00
validate(pb *msg.SendMsgReq) (bool, int32, string)
2022-09-29 17:57:26 +08:00
}
type MessageRevoked struct {
RevokerID string `json:"revokerID"`
RevokerRole int32 `json:"revokerRole"`
ClientMsgID string `json:"clientMsgID"`
RevokerNickname string `json:"revokerNickname"`
RevokeTime int64 `json:"revokeTime"`
SourceMessageSendTime int64 `json:"sourceMessageSendTime"`
SourceMessageSendID string `json:"sourceMessageSendID"`
SourceMessageSenderNickname string `json:"sourceMessageSenderNickname"`
SessionType int32 `json:"sessionType"`
Seq uint32 `json:"seq"`
}
2021-06-28 15:35:18 +08:00
type MsgCallBackReq struct {
SendID string `json:"sendID"`
RecvID string `json:"recvID"`
Content string `json:"content"`
SendTime int64 `json:"sendTime"`
MsgFrom int32 `json:"msgFrom"`
ContentType int32 `json:"contentType"`
SessionType int32 `json:"sessionType"`
PlatformID int32 `json:"senderPlatformID"`
MsgID string `json:"msgID"`
IsOnlineOnly bool `json:"isOnlineOnly"`
2021-06-28 15:35:18 +08:00
}
type MsgCallBackResp struct {
ErrCode int32 `json:"errCode"`
ErrMsg string `json:"errMsg"`
ResponseErrCode int32 `json:"responseErrCode"`
ResponseResult struct {
ModifiedMsg string `json:"modifiedMsg"`
Ext string `json:"ext"`
}
}
2021-05-26 19:40:38 +08:00
2023-02-10 22:10:37 +08:00
func isMessageHasReadEnabled(pb *msg.SendMsgReq) (bool, int32, string) {
2022-06-09 12:39:28 +08:00
switch pb.MsgData.ContentType {
case constant.HasReadReceipt:
if config.Config.SingleMessageHasReadReceiptEnable {
return true, 0, ""
} else {
return false, constant.ErrMessageHasReadDisable.ErrCode, constant.ErrMessageHasReadDisable.ErrMsg
}
case constant.GroupHasReadReceipt:
if config.Config.GroupMessageHasReadReceiptEnable {
return true, 0, ""
} else {
return false, constant.ErrMessageHasReadDisable.ErrCode, constant.ErrMessageHasReadDisable.ErrMsg
}
}
return true, 0, ""
}
2023-01-11 11:15:46 +08:00
func userIsMuteAndIsAdminInGroup(ctx context.Context, groupID, userID string) (isMute bool, isAdmin bool, err error) {
groupMemberInfo, err := rocksCache.GetGroupMemberInfoFromCache(ctx, groupID, userID)
2022-11-11 18:06:15 +08:00
if err != nil {
2022-12-02 17:27:12 +08:00
return false, false, utils.Wrap(err, "")
2022-11-11 18:06:15 +08:00
}
2022-12-02 17:27:12 +08:00
2022-11-11 18:06:15 +08:00
if groupMemberInfo.MuteEndTime.Unix() >= time.Now().Unix() {
2022-12-02 17:27:12 +08:00
return true, groupMemberInfo.RoleLevel > constant.GroupOrdinaryUsers, nil
2022-11-11 18:06:15 +08:00
}
2022-12-02 17:27:12 +08:00
return false, groupMemberInfo.RoleLevel > constant.GroupOrdinaryUsers, nil
2022-11-11 18:06:15 +08:00
}
2023-01-11 11:15:46 +08:00
func groupIsMuted(ctx context.Context, groupID string) (bool, error) {
groupInfo, err := rocksCache.GetGroupInfoFromCache(ctx, groupID)
2022-11-29 14:41:20 +08:00
if err != nil {
return false, utils.Wrap(err, "GetGroupInfoFromCache failed")
}
if groupInfo.Status == constant.GroupStatusMuted {
return true, nil
}
return false, nil
}
2023-02-13 09:19:56 +08:00
func (rpc *msgServer) messageVerification(ctx context.Context, data *pbChat.SendMsgReq) (bool, int32, string, []string) {
2022-07-25 19:19:00 +08:00
switch data.MsgData.SessionType {
case constant.SingleChatType:
2022-08-05 15:51:34 +08:00
if utils.IsContain(data.MsgData.SendID, config.Config.Manager.AppManagerUid) {
return true, 0, "", nil
}
if data.MsgData.ContentType <= constant.NotificationEnd && data.MsgData.ContentType >= constant.NotificationBegin {
return true, 0, "", nil
}
2022-11-10 14:51:35 +08:00
log.NewDebug(data.OperationID, *config.Config.MessageVerify.FriendVerify)
2022-05-28 18:10:08 +08:00
reqGetBlackIDListFromCache := &cacheRpc.GetBlackIDListFromCacheReq{UserID: data.MsgData.RecvID, OperationID: data.OperationID}
2023-02-07 20:24:20 +08:00
etcdConn, err := rpc.GetConn(context.Background(), config.Config.RpcRegisterName.OpenImCacheName)
2023-01-11 11:15:46 +08:00
if err != nil {
2022-08-17 12:12:54 +08:00
errMsg := data.OperationID + "getcdv3.GetDefaultConn == nil"
2022-06-16 14:26:03 +08:00
log.NewError(data.OperationID, errMsg)
2022-07-25 19:19:00 +08:00
return true, 0, "", nil
2022-06-16 14:26:03 +08:00
}
2022-04-28 15:10:42 +08:00
cacheClient := cacheRpc.NewCacheClient(etcdConn)
2022-05-28 18:10:08 +08:00
cacheResp, err := cacheClient.GetBlackIDListFromCache(context.Background(), reqGetBlackIDListFromCache)
2022-03-15 19:03:22 +08:00
if err != nil {
2022-05-28 18:10:08 +08:00
log.NewError(data.OperationID, "GetBlackIDListFromCache rpc call failed ", err.Error())
2022-04-28 15:10:42 +08:00
} else {
if cacheResp.CommonResp.ErrCode != 0 {
2022-05-28 18:10:08 +08:00
log.NewError(data.OperationID, "GetBlackIDListFromCache rpc logic call failed ", cacheResp.String())
2022-04-28 15:10:42 +08:00
} else {
2022-05-28 18:10:08 +08:00
if utils.IsContain(data.MsgData.SendID, cacheResp.UserIDList) {
2022-07-25 19:19:00 +08:00
return false, 600, "in black list", nil
2022-04-28 15:10:42 +08:00
}
}
2022-03-15 19:03:22 +08:00
}
2022-11-10 14:51:35 +08:00
log.NewDebug(data.OperationID, *config.Config.MessageVerify.FriendVerify)
if *config.Config.MessageVerify.FriendVerify {
2022-05-28 18:10:08 +08:00
reqGetFriendIDListFromCache := &cacheRpc.GetFriendIDListFromCacheReq{UserID: data.MsgData.RecvID, OperationID: data.OperationID}
2023-02-07 20:24:20 +08:00
etcdConn, err := rpc.GetConn(context.Background(), config.Config.RpcRegisterName.OpenImCacheName)
2023-01-11 11:15:46 +08:00
if err != nil {
2022-08-17 12:12:54 +08:00
errMsg := data.OperationID + "getcdv3.GetDefaultConn == nil"
2022-06-16 14:26:03 +08:00
log.NewError(data.OperationID, errMsg)
2022-07-25 19:19:00 +08:00
return true, 0, "", nil
2022-06-16 14:26:03 +08:00
}
2022-05-28 18:10:08 +08:00
cacheClient := cacheRpc.NewCacheClient(etcdConn)
cacheResp, err := cacheClient.GetFriendIDListFromCache(context.Background(), reqGetFriendIDListFromCache)
if err != nil {
log.NewError(data.OperationID, "GetFriendIDListFromCache rpc call failed ", err.Error())
} else {
if cacheResp.CommonResp.ErrCode != 0 {
log.NewError(data.OperationID, "GetFriendIDListFromCache rpc logic call failed ", cacheResp.String())
} else {
if !utils.IsContain(data.MsgData.SendID, cacheResp.UserIDList) {
2022-07-25 19:19:00 +08:00
return false, 601, "not friend", nil
2022-05-28 18:10:08 +08:00
}
}
}
2022-07-25 19:19:00 +08:00
return true, 0, "", nil
2022-05-28 18:10:08 +08:00
} else {
2022-07-25 19:19:00 +08:00
return true, 0, "", nil
}
case constant.GroupChatType:
2023-01-30 15:28:46 +08:00
userIDList, err := utils.GetGroupMemberUserIDList(ctx, data.MsgData.GroupID, data.OperationID)
2022-10-10 17:51:57 +08:00
if err != nil {
errMsg := data.OperationID + err.Error()
log.NewError(data.OperationID, errMsg)
return false, 201, errMsg, nil
}
2023-02-09 14:40:49 +08:00
if tokenverify.IsManagerUserID(data.MsgData.SendID) {
2022-11-21 10:20:56 +08:00
return true, 0, "", userIDList
}
2022-11-22 14:18:57 +08:00
if data.MsgData.ContentType <= constant.NotificationEnd && data.MsgData.ContentType >= constant.NotificationBegin {
return true, 0, "", userIDList
} else {
if !utils.IsContain(data.MsgData.SendID, userIDList) {
//return returnMsg(&replay, pb, 202, "you are not in group", "", 0)
return false, 202, "you are not in group", nil
}
2022-10-10 17:51:57 +08:00
}
2023-01-11 11:15:46 +08:00
isMute, isAdmin, err := userIsMuteAndIsAdminInGroup(ctx, data.MsgData.GroupID, data.MsgData.SendID)
2022-11-29 14:41:20 +08:00
if err != nil {
errMsg := data.OperationID + err.Error()
return false, 223, errMsg, nil
}
if isMute {
2022-12-02 17:27:12 +08:00
return false, 224, "you are muted", nil
}
if isAdmin {
return true, 0, "", userIDList
2022-11-29 14:41:20 +08:00
}
2023-01-11 11:15:46 +08:00
isMute, err = groupIsMuted(ctx, data.MsgData.GroupID)
2022-11-11 18:06:15 +08:00
if err != nil {
errMsg := data.OperationID + err.Error()
return false, 223, errMsg, nil
}
if isMute {
2022-12-02 17:27:12 +08:00
return false, 225, "group id muted", nil
2022-11-11 18:06:15 +08:00
}
2022-10-10 17:51:57 +08:00
return true, 0, "", userIDList
2022-07-25 19:19:00 +08:00
case constant.SuperGroupChatType:
2023-01-11 11:15:46 +08:00
groupInfo, err := rocksCache.GetGroupInfoFromCache(ctx, data.MsgData.GroupID)
2022-07-25 19:19:00 +08:00
if err != nil {
return false, 201, err.Error(), nil
}
2022-10-10 17:51:57 +08:00
2022-09-29 17:57:26 +08:00
if data.MsgData.ContentType == constant.AdvancedRevoke {
2022-09-29 18:31:36 +08:00
revokeMessage := new(MessageRevoked)
err := utils.JsonStringToStruct(string(data.MsgData.Content), revokeMessage)
2022-09-29 17:57:26 +08:00
if err != nil {
log.Error(data.OperationID, "json unmarshal err:", err.Error())
return false, 201, err.Error(), nil
}
2022-09-29 18:35:47 +08:00
log.Debug(data.OperationID, "revoke message is", *revokeMessage)
2022-09-29 19:29:22 +08:00
if revokeMessage.RevokerID != revokeMessage.SourceMessageSendID {
req := pbChat.GetSuperGroupMsgReq{OperationID: data.OperationID, Seq: revokeMessage.Seq, GroupID: data.MsgData.GroupID}
resp, err := rpc.GetSuperGroupMsg(context.Background(), &req)
if err != nil {
log.Error(data.OperationID, "GetSuperGroupMsgReq err:", err.Error())
} else if resp.ErrCode != 0 {
2022-10-10 17:51:57 +08:00
log.Error(data.OperationID, "GetSuperGroupMsgReq err:", resp.ErrCode, resp.ErrMsg)
2022-09-29 17:57:26 +08:00
} else {
2022-09-29 19:29:22 +08:00
if resp.MsgData != nil && resp.MsgData.ClientMsgID == revokeMessage.ClientMsgID && resp.MsgData.Seq == revokeMessage.Seq {
revokeMessage.SourceMessageSendTime = resp.MsgData.SendTime
revokeMessage.SourceMessageSenderNickname = resp.MsgData.SenderNickname
revokeMessage.SourceMessageSendID = resp.MsgData.SendID
log.Debug(data.OperationID, "new revoke message is ", revokeMessage)
data.MsgData.Content = []byte(utils.StructToJsonString(revokeMessage))
} else {
return false, 201, errors.New("msg err").Error(), nil
}
2022-09-29 17:57:26 +08:00
}
}
}
2022-08-01 19:13:26 +08:00
if groupInfo.GroupType == constant.SuperGroup {
return true, 0, "", nil
} else {
2023-01-30 15:28:46 +08:00
userIDList, err := utils.GetGroupMemberUserIDList(ctx, data.MsgData.GroupID, data.OperationID)
2022-08-30 22:04:13 +08:00
if err != nil {
errMsg := data.OperationID + err.Error()
2022-08-01 19:13:26 +08:00
log.NewError(data.OperationID, errMsg)
return false, 201, errMsg, nil
2022-07-25 19:19:00 +08:00
}
2023-02-09 14:40:49 +08:00
if tokenverify.IsManagerUserID(data.MsgData.SendID) {
2022-11-21 10:20:56 +08:00
return true, 0, "", userIDList
}
2022-11-22 14:18:57 +08:00
if data.MsgData.ContentType <= constant.NotificationEnd && data.MsgData.ContentType >= constant.NotificationBegin {
return true, 0, "", userIDList
} else {
if !utils.IsContain(data.MsgData.SendID, userIDList) {
//return returnMsg(&replay, pb, 202, "you are not in group", "", 0)
return false, 202, "you are not in group", nil
}
2022-08-01 19:13:26 +08:00
}
2023-01-11 11:15:46 +08:00
isMute, isAdmin, err := userIsMuteAndIsAdminInGroup(ctx, data.MsgData.GroupID, data.MsgData.SendID)
2022-11-29 14:41:20 +08:00
if err != nil {
errMsg := data.OperationID + err.Error()
return false, 223, errMsg, nil
}
if isMute {
2022-12-02 17:27:12 +08:00
return false, 224, "you are muted", nil
2022-11-29 14:41:20 +08:00
}
2022-12-02 17:27:12 +08:00
if isAdmin {
return true, 0, "", userIDList
}
2023-01-11 11:15:46 +08:00
isMute, err = groupIsMuted(ctx, data.MsgData.GroupID)
2022-11-11 18:06:15 +08:00
if err != nil {
errMsg := data.OperationID + err.Error()
return false, 223, errMsg, nil
}
if isMute {
2022-12-02 17:27:12 +08:00
return false, 225, "group id muted", nil
2022-11-11 18:06:15 +08:00
}
2022-08-30 22:04:13 +08:00
return true, 0, "", userIDList
2022-07-25 19:19:00 +08:00
}
default:
return true, 0, "", nil
2022-03-15 19:03:22 +08:00
}
2022-07-25 19:19:00 +08:00
2022-02-21 18:18:53 +08:00
}
2023-02-13 09:19:56 +08:00
func (rpc *msgServer) encapsulateMsgData(msg *sdkws.MsgData) {
2021-12-23 17:19:57 +08:00
msg.ServerMsgID = GetMsgID(msg.SendID)
2022-02-22 16:01:32 +08:00
msg.SendTime = utils.GetCurrentTimestampByMill()
2021-12-23 17:19:57 +08:00
switch msg.ContentType {
case constant.Text:
fallthrough
case constant.Picture:
fallthrough
case constant.Voice:
fallthrough
case constant.Video:
fallthrough
case constant.File:
fallthrough
case constant.AtText:
fallthrough
case constant.Merger:
fallthrough
case constant.Card:
fallthrough
case constant.Location:
fallthrough
case constant.Custom:
fallthrough
case constant.Quote:
utils.SetSwitchFromOptions(msg.Options, constant.IsConversationUpdate, true)
utils.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, true)
utils.SetSwitchFromOptions(msg.Options, constant.IsSenderSync, true)
case constant.Revoke:
utils.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, false)
case constant.HasReadReceipt:
2022-02-14 16:59:26 +08:00
log.Info("", "this is a test start", msg, msg.Options)
2021-12-23 17:19:57 +08:00
utils.SetSwitchFromOptions(msg.Options, constant.IsConversationUpdate, false)
2022-03-28 18:30:12 +08:00
utils.SetSwitchFromOptions(msg.Options, constant.IsSenderConversationUpdate, false)
2021-12-23 17:19:57 +08:00
utils.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, false)
2022-02-14 16:59:26 +08:00
log.Info("", "this is a test end", msg, msg.Options)
2021-12-23 17:19:57 +08:00
case constant.Typing:
utils.SetSwitchFromOptions(msg.Options, constant.IsHistory, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsPersistent, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsSenderSync, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsConversationUpdate, false)
2022-03-28 18:30:12 +08:00
utils.SetSwitchFromOptions(msg.Options, constant.IsSenderConversationUpdate, false)
2021-12-23 17:19:57 +08:00
utils.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, false)
}
}
2021-12-10 17:30:11 +08:00
2023-02-13 09:19:56 +08:00
func (rpc *msgServer) sendMsgToWriter(ctx context.Context, m *pbChat.MsgDataToMQ, key string, status string) error {
2022-05-11 18:33:48 +08:00
switch status {
case constant.OnlineStatus:
2022-09-19 22:16:45 +08:00
if m.MsgData.ContentType == constant.SignalingNotification {
rpcPushMsg := pbPush.PushMsgReq{OperationID: m.OperationID, MsgData: m.MsgData, PushToUserID: key}
2023-02-07 20:24:20 +08:00
grpcConn, err := rpc.GetConn(ctx, config.Config.RpcRegisterName.OpenImPushName)
2023-01-11 11:15:46 +08:00
if err != nil {
return err
2022-09-19 22:16:45 +08:00
}
msgClient := pbPush.NewPushMsgServiceClient(grpcConn)
2023-01-11 11:15:46 +08:00
_, err = msgClient.PushMsg(context.Background(), &rpcPushMsg)
2022-09-19 22:16:45 +08:00
if err != nil {
log.Error(rpcPushMsg.OperationID, "rpc send failed", rpcPushMsg.OperationID, "push data", rpcPushMsg.String(), "err", err.Error())
return err
} else {
return nil
}
}
2022-09-29 17:57:26 +08:00
pid, offset, err := rpc.messageWriter.SendMessage(m, key, m.OperationID)
2022-05-11 18:33:48 +08:00
if err != nil {
log.Error(m.OperationID, "kafka send failed", "send data", m.String(), "pid", pid, "offset", offset, "err", err.Error(), "key", key, status)
2022-05-21 10:59:44 +08:00
} else {
2022-09-29 17:57:26 +08:00
// log.NewWarn(m.OperationID, "sendMsgToWriter client msgID ", m.MsgData.ClientMsgID)
2022-05-11 18:33:48 +08:00
}
return err
case constant.OfflineStatus:
2022-09-29 17:57:26 +08:00
pid, offset, err := rpc.messageWriter.SendMessage(m, key, m.OperationID)
2022-05-11 18:33:48 +08:00
if err != nil {
log.Error(m.OperationID, "kafka send failed", "send data", m.String(), "pid", pid, "offset", offset, "err", err.Error(), "key", key, status)
}
return err
2021-05-26 19:40:38 +08:00
}
2022-05-11 18:33:48 +08:00
return errors.New("status error")
2021-05-26 19:40:38 +08:00
}
func GetMsgID(sendID string) string {
t := time.Now().Format("2006-01-02 15:04:05")
2022-02-25 18:35:04 +08:00
return utils.Md5(t + "-" + sendID + "-" + strconv.Itoa(rand.Int()))
2021-05-26 19:40:38 +08:00
}
2021-12-21 21:40:50 +08:00
2021-12-23 17:19:57 +08:00
func returnMsg(replay *pbChat.SendMsgResp, pb *pbChat.SendMsgReq, errCode int32, errMsg, serverMsgID string, sendTime int64) (*pbChat.SendMsgResp, error) {
2021-06-28 15:35:18 +08:00
replay.ErrCode = errCode
replay.ErrMsg = errMsg
replay.ServerMsgID = serverMsgID
2021-12-23 17:19:57 +08:00
replay.ClientMsgID = pb.MsgData.ClientMsgID
2021-06-28 15:35:18 +08:00
replay.SendTime = sendTime
return replay, nil
}
2021-12-21 21:40:50 +08:00
2021-12-23 17:19:57 +08:00
func modifyMessageByUserMessageReceiveOpt(userID, sourceID string, sessionType int, pb *pbChat.SendMsgReq) bool {
2022-06-16 12:53:39 +08:00
opt, err := db.DB.GetUserGlobalMsgRecvOpt(userID)
if err != nil {
log.NewError(pb.OperationID, "GetUserGlobalMsgRecvOpt from redis err", userID, pb.String(), err.Error())
}
switch opt {
case constant.ReceiveMessage:
case constant.NotReceiveMessage:
return false
case constant.ReceiveNotNotifyMessage:
if pb.MsgData.Options == nil {
pb.MsgData.Options = make(map[string]bool, 10)
}
utils.SetSwitchFromOptions(pb.MsgData.Options, constant.IsOfflinePush, false)
return true
}
2021-12-06 20:03:59 +08:00
conversationID := utils.GetConversationIDBySessionType(sourceID, sessionType)
2022-06-16 12:53:39 +08:00
singleOpt, sErr := db.DB.GetSingleConversationRecvMsgOpt(userID, conversationID)
if sErr != nil && sErr != go_redis.Nil {
log.NewError(pb.OperationID, "GetSingleConversationMsgOpt from redis err", conversationID, pb.String(), sErr.Error())
2021-12-06 20:03:59 +08:00
return true
}
2022-06-16 12:53:39 +08:00
switch singleOpt {
2021-12-06 20:03:59 +08:00
case constant.ReceiveMessage:
return true
case constant.NotReceiveMessage:
2022-08-05 19:10:08 +08:00
if utils.IsContainInt(int(pb.MsgData.ContentType), ExcludeContentType) {
return true
}
2021-12-06 20:03:59 +08:00
return false
case constant.ReceiveNotNotifyMessage:
2021-12-23 17:19:57 +08:00
if pb.MsgData.Options == nil {
pb.MsgData.Options = make(map[string]bool, 10)
2021-12-08 18:30:23 +08:00
}
2021-12-23 17:19:57 +08:00
utils.SetSwitchFromOptions(pb.MsgData.Options, constant.IsOfflinePush, false)
2021-12-06 20:03:59 +08:00
return true
}
return true
}
2021-12-21 21:40:50 +08:00
2022-05-22 18:28:51 +08:00
func modifyMessageByUserMessageReceiveOptoptimization(userID, sourceID string, sessionType int, operationID string, options *map[string]bool) bool {
conversationID := utils.GetConversationIDBySessionType(sourceID, sessionType)
opt, err := db.DB.GetSingleConversationRecvMsgOpt(userID, conversationID)
2022-06-15 16:50:45 +08:00
if err != nil && err != go_redis.Nil {
2022-05-22 18:28:51 +08:00
log.NewError(operationID, "GetSingleConversationMsgOpt from redis err", userID, conversationID, err.Error())
return true
}
switch opt {
case constant.ReceiveMessage:
return true
case constant.NotReceiveMessage:
return false
case constant.ReceiveNotNotifyMessage:
if *options == nil {
*options = make(map[string]bool, 10)
}
utils.SetSwitchFromOptions(*options, constant.IsOfflinePush, false)
return true
}
return true
}
2022-05-13 21:05:39 +08:00
func getOnlineAndOfflineUserIDList(memberList []string, m map[string][]string, operationID string) {
var onllUserIDList, offlUserIDList []string
2022-05-11 18:33:48 +08:00
var wsResult []*pbRelay.GetUsersOnlineStatusResp_SuccessResult
req := &pbRelay.GetUsersOnlineStatusReq{}
req.UserIDList = memberList
req.OperationID = operationID
req.OpUserID = config.Config.Manager.AppManagerUid[0]
flag := false
2023-02-07 20:24:20 +08:00
grpcCons := rpc.GetDefaultGatewayConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), operationID)
2022-05-11 18:33:48 +08:00
for _, v := range grpcCons {
2022-07-20 21:15:30 +08:00
client := pbRelay.NewRelayClient(v)
2022-05-11 18:33:48 +08:00
reply, err := client.GetUsersOnlineStatus(context.Background(), req)
if err != nil {
log.NewError(operationID, "GetUsersOnlineStatus rpc err", req.String(), err.Error())
continue
} else {
if reply.ErrCode == 0 {
wsResult = append(wsResult, reply.SuccessResult...)
}
}
}
log.NewInfo(operationID, "call GetUsersOnlineStatus rpc server is success", wsResult)
//Online data merge of each node
for _, v1 := range memberList {
flag = false
for _, v2 := range wsResult {
if v2.UserID == v1 {
flag = true
onllUserIDList = append(onllUserIDList, v1)
}
}
if !flag {
offlUserIDList = append(offlUserIDList, v1)
}
}
2022-05-13 21:05:39 +08:00
m[constant.OnlineStatus] = onllUserIDList
m[constant.OfflineStatus] = offlUserIDList
2022-05-11 18:33:48 +08:00
}
2022-05-22 18:28:51 +08:00
func valueCopy(pb *pbChat.SendMsgReq) *pbChat.SendMsgReq {
2023-02-09 20:36:34 +08:00
offlinePushInfo := sdkws.OfflinePushInfo{}
2022-05-22 18:28:51 +08:00
if pb.MsgData.OfflinePushInfo != nil {
offlinePushInfo = *pb.MsgData.OfflinePushInfo
}
2023-02-09 20:36:34 +08:00
msgData := sdkws.MsgData{}
2022-05-22 18:28:51 +08:00
msgData = *pb.MsgData
msgData.OfflinePushInfo = &offlinePushInfo
options := make(map[string]bool, 10)
for key, value := range pb.MsgData.Options {
options[key] = value
}
msgData.Options = options
2022-05-23 09:54:43 +08:00
return &pbChat.SendMsgReq{Token: pb.Token, OperationID: pb.OperationID, MsgData: &msgData}
2022-05-22 18:28:51 +08:00
}
2023-02-13 09:19:56 +08:00
func (rpc *msgServer) sendMsgToGroup(ctx context.Context, list []string, pb pbChat.SendMsgReq, status string, sendTag *bool, wg *sync.WaitGroup) {
2022-05-11 18:33:48 +08:00
// log.Debug(pb.OperationID, "split userID ", list)
2023-02-09 20:36:34 +08:00
offlinePushInfo := sdkws.OfflinePushInfo{}
2022-05-11 18:33:48 +08:00
if pb.MsgData.OfflinePushInfo != nil {
2022-05-13 15:59:38 +08:00
offlinePushInfo = *pb.MsgData.OfflinePushInfo
2022-05-11 18:33:48 +08:00
}
2023-02-09 20:36:34 +08:00
msgData := sdkws.MsgData{}
2022-05-13 15:59:38 +08:00
msgData = *pb.MsgData
msgData.OfflinePushInfo = &offlinePushInfo
groupPB := pbChat.SendMsgReq{Token: pb.Token, OperationID: pb.OperationID, MsgData: &msgData}
msgToMQGroup := pbChat.MsgDataToMQ{Token: pb.Token, OperationID: pb.OperationID, MsgData: &msgData}
2022-05-11 18:33:48 +08:00
for _, v := range list {
2022-05-13 15:59:38 +08:00
options := make(map[string]bool, 10)
for key, value := range pb.MsgData.Options {
options[key] = value
}
2022-05-11 18:33:48 +08:00
groupPB.MsgData.RecvID = v
2022-05-13 15:59:38 +08:00
groupPB.MsgData.Options = options
isSend := modifyMessageByUserMessageReceiveOpt(v, msgData.GroupID, constant.GroupChatType, &groupPB)
2022-05-11 18:33:48 +08:00
if isSend {
msgToMQGroup.MsgData = groupPB.MsgData
2022-09-29 17:57:26 +08:00
// log.Debug(groupPB.OperationID, "sendMsgToWriter, ", v, groupID, msgToMQGroup.String())
2023-01-11 16:23:16 +08:00
err := rpc.sendMsgToWriter(ctx, &msgToMQGroup, v, status)
2022-05-11 18:33:48 +08:00
if err != nil {
log.NewError(msgToMQGroup.OperationID, "kafka send msg err:UserId", v, msgToMQGroup.String())
} else {
*sendTag = true
}
} else {
2022-09-29 17:57:26 +08:00
log.Debug(groupPB.OperationID, "not sendMsgToWriter, ", v)
2022-05-11 18:33:48 +08:00
}
}
wg.Done()
}
2022-05-22 18:28:51 +08:00
2023-02-13 09:19:56 +08:00
func (rpc *msgServer) sendMsgToGroupOptimization(ctx context.Context, list []string, groupPB *pbChat.SendMsgReq, status string, sendTag *bool, wg *sync.WaitGroup) {
2022-05-22 18:28:51 +08:00
msgToMQGroup := pbChat.MsgDataToMQ{Token: groupPB.Token, OperationID: groupPB.OperationID, MsgData: groupPB.MsgData}
2022-08-09 15:54:16 +08:00
tempOptions := make(map[string]bool, 1)
for k, v := range groupPB.MsgData.Options {
tempOptions[k] = v
}
2022-05-22 18:28:51 +08:00
for _, v := range list {
groupPB.MsgData.RecvID = v
2022-08-09 15:41:16 +08:00
options := make(map[string]bool, 1)
2022-08-09 15:54:16 +08:00
for k, v := range tempOptions {
2022-08-09 15:41:16 +08:00
options[k] = v
}
groupPB.MsgData.Options = options
2022-05-22 18:28:51 +08:00
isSend := modifyMessageByUserMessageReceiveOpt(v, groupPB.MsgData.GroupID, constant.GroupChatType, groupPB)
if isSend {
2022-06-07 13:44:52 +08:00
if v == "" || groupPB.MsgData.SendID == "" {
2022-06-07 14:02:11 +08:00
log.Error(msgToMQGroup.OperationID, "sendMsgToGroupOptimization userID nil ", msgToMQGroup.String())
continue
2022-06-07 13:44:52 +08:00
}
2023-01-11 16:23:16 +08:00
err := rpc.sendMsgToWriter(ctx, &msgToMQGroup, v, status)
2022-05-22 18:28:51 +08:00
if err != nil {
log.NewError(msgToMQGroup.OperationID, "kafka send msg err:UserId", v, msgToMQGroup.String())
} else {
*sendTag = true
}
} else {
2022-09-29 17:57:26 +08:00
log.Debug(groupPB.OperationID, "not sendMsgToWriter, ", v)
2022-05-22 18:28:51 +08:00
}
}
wg.Done()
}