616 lines
25 KiB
Go
616 lines
25 KiB
Go
package push
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"time"
|
||
|
||
"git.imall.cloud/openim/open-im-server-deploy/internal/push/offlinepush"
|
||
"git.imall.cloud/openim/open-im-server-deploy/internal/push/offlinepush/options"
|
||
"git.imall.cloud/openim/open-im-server-deploy/pkg/common/prommetrics"
|
||
"git.imall.cloud/openim/open-im-server-deploy/pkg/common/storage/controller"
|
||
"git.imall.cloud/openim/open-im-server-deploy/pkg/common/webhook"
|
||
"git.imall.cloud/openim/open-im-server-deploy/pkg/msgprocessor"
|
||
"git.imall.cloud/openim/open-im-server-deploy/pkg/rpccache"
|
||
"git.imall.cloud/openim/open-im-server-deploy/pkg/rpcli"
|
||
"git.imall.cloud/openim/open-im-server-deploy/pkg/util/conversationutil"
|
||
"git.imall.cloud/openim/protocol/constant"
|
||
"git.imall.cloud/openim/protocol/msggateway"
|
||
pbpush "git.imall.cloud/openim/protocol/push"
|
||
"git.imall.cloud/openim/protocol/sdkws"
|
||
"github.com/openimsdk/tools/discovery"
|
||
"github.com/openimsdk/tools/log"
|
||
"github.com/openimsdk/tools/mcontext"
|
||
"github.com/openimsdk/tools/utils/datautil"
|
||
"github.com/openimsdk/tools/utils/jsonutil"
|
||
"github.com/openimsdk/tools/utils/timeutil"
|
||
"github.com/redis/go-redis/v9"
|
||
"google.golang.org/protobuf/proto"
|
||
)
|
||
|
||
type ConsumerHandler struct {
|
||
//pushConsumerGroup mq.Consumer
|
||
offlinePusher offlinepush.OfflinePusher
|
||
onlinePusher OnlinePusher
|
||
pushDatabase controller.PushDatabase
|
||
onlineCache rpccache.OnlineCache
|
||
groupLocalCache *rpccache.GroupLocalCache
|
||
conversationLocalCache *rpccache.ConversationLocalCache
|
||
webhookClient *webhook.Client
|
||
config *Config
|
||
userClient *rpcli.UserClient
|
||
groupClient *rpcli.GroupClient
|
||
msgClient *rpcli.MsgClient
|
||
conversationClient *rpcli.ConversationClient
|
||
}
|
||
|
||
func NewConsumerHandler(ctx context.Context, config *Config, database controller.PushDatabase, offlinePusher offlinepush.OfflinePusher, rdb redis.UniversalClient, client discovery.Conn) (*ConsumerHandler, error) {
|
||
userConn, err := client.GetConn(ctx, config.Discovery.RpcService.User)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
groupConn, err := client.GetConn(ctx, config.Discovery.RpcService.Group)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
msgConn, err := client.GetConn(ctx, config.Discovery.RpcService.Msg)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
conversationConn, err := client.GetConn(ctx, config.Discovery.RpcService.Conversation)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
onlinePusher, err := NewOnlinePusher(client, config)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var consumerHandler ConsumerHandler
|
||
consumerHandler.userClient = rpcli.NewUserClient(userConn)
|
||
consumerHandler.groupClient = rpcli.NewGroupClient(groupConn)
|
||
consumerHandler.msgClient = rpcli.NewMsgClient(msgConn)
|
||
consumerHandler.conversationClient = rpcli.NewConversationClient(conversationConn)
|
||
|
||
consumerHandler.offlinePusher = offlinePusher
|
||
consumerHandler.onlinePusher = onlinePusher
|
||
consumerHandler.groupLocalCache = rpccache.NewGroupLocalCache(consumerHandler.groupClient, &config.LocalCacheConfig, rdb)
|
||
consumerHandler.conversationLocalCache = rpccache.NewConversationLocalCache(consumerHandler.conversationClient, &config.LocalCacheConfig, rdb)
|
||
consumerHandler.webhookClient = webhook.NewWebhookClient(config.WebhooksConfig.URL)
|
||
consumerHandler.config = config
|
||
consumerHandler.pushDatabase = database
|
||
consumerHandler.onlineCache, err = rpccache.NewOnlineCache(consumerHandler.userClient, consumerHandler.groupLocalCache, rdb, config.RpcConfig.FullUserCache, nil)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &consumerHandler, nil
|
||
}
|
||
|
||
func (c *ConsumerHandler) HandleMs2PsChat(ctx context.Context, msg []byte) {
|
||
msgFromMQ := pbpush.PushMsgReq{}
|
||
if err := proto.Unmarshal(msg, &msgFromMQ); err != nil {
|
||
log.ZError(ctx, "push Unmarshal msg err", err, "msg", string(msg))
|
||
return
|
||
}
|
||
|
||
sec := msgFromMQ.MsgData.SendTime / 1000
|
||
nowSec := timeutil.GetCurrentTimestampBySecond()
|
||
|
||
if nowSec-sec > 10 {
|
||
prommetrics.MsgLoneTimePushCounter.Inc()
|
||
log.ZWarn(ctx, "it’s been a while since the message was sent", nil, "msg", msgFromMQ.String(), "sec", sec, "nowSec", nowSec, "nowSec-sec", nowSec-sec)
|
||
}
|
||
var err error
|
||
|
||
switch msgFromMQ.MsgData.SessionType {
|
||
case constant.ReadGroupChatType:
|
||
err = c.Push2Group(ctx, msgFromMQ.MsgData.GroupID, msgFromMQ.MsgData)
|
||
default:
|
||
var pushUserIDList []string
|
||
isSenderSync := datautil.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsSenderSync)
|
||
if !isSenderSync || msgFromMQ.MsgData.SendID == msgFromMQ.MsgData.RecvID {
|
||
pushUserIDList = append(pushUserIDList, msgFromMQ.MsgData.RecvID)
|
||
} else {
|
||
pushUserIDList = append(pushUserIDList, msgFromMQ.MsgData.RecvID, msgFromMQ.MsgData.SendID)
|
||
}
|
||
err = c.Push2User(ctx, pushUserIDList, msgFromMQ.MsgData)
|
||
}
|
||
if err != nil {
|
||
log.ZWarn(ctx, "push failed", err, "msg", msgFromMQ.String())
|
||
}
|
||
}
|
||
|
||
func (c *ConsumerHandler) WaitCache() {
|
||
c.onlineCache.WaitCache()
|
||
}
|
||
|
||
// Push2User Suitable for two types of conversations, one is SingleChatType and the other is NotificationChatType.
|
||
func (c *ConsumerHandler) Push2User(ctx context.Context, userIDs []string, msg *sdkws.MsgData) (err error) {
|
||
log.ZInfo(ctx, "Get msg from msg_transfer And push msg", "userIDs", userIDs, "msg", msg.String())
|
||
defer func(duration time.Time) {
|
||
t := time.Since(duration)
|
||
log.ZInfo(ctx, "Get msg from msg_transfer And push msg end", "msg", msg.String(), "time cost", t)
|
||
}(time.Now())
|
||
if err := c.webhookBeforeOnlinePush(ctx, &c.config.WebhooksConfig.BeforeOnlinePush, userIDs, msg); err != nil {
|
||
return err
|
||
}
|
||
|
||
wsResults, err := c.GetConnsAndOnlinePush(ctx, msg, userIDs)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
log.ZDebug(ctx, "single and notification push result", "result", wsResults, "msg", msg, "push_to_userID", userIDs)
|
||
log.ZInfo(ctx, "single and notification push end")
|
||
|
||
if !c.shouldPushOffline(ctx, msg) {
|
||
return nil
|
||
}
|
||
log.ZInfo(ctx, "pushOffline start")
|
||
|
||
for _, v := range wsResults {
|
||
//message sender do not need offline push
|
||
if msg.SendID == v.UserID {
|
||
continue
|
||
}
|
||
//receiver online push success
|
||
if v.OnlinePush {
|
||
return nil
|
||
}
|
||
}
|
||
needOfflinePushUserID := []string{msg.RecvID}
|
||
var offlinePushUserID []string
|
||
|
||
//receiver offline push
|
||
if err = c.webhookBeforeOfflinePush(ctx, &c.config.WebhooksConfig.BeforeOfflinePush, needOfflinePushUserID, msg, &offlinePushUserID); err != nil {
|
||
return err
|
||
}
|
||
|
||
if len(offlinePushUserID) > 0 {
|
||
needOfflinePushUserID = offlinePushUserID
|
||
}
|
||
err = c.offlinePushMsg(ctx, msg, needOfflinePushUserID)
|
||
if err != nil {
|
||
log.ZDebug(ctx, "offlinePushMsg failed", err, "needOfflinePushUserID", needOfflinePushUserID, "msg", msg)
|
||
log.ZWarn(ctx, "offlinePushMsg failed", err, "needOfflinePushUserID length", len(needOfflinePushUserID), "msg", msg)
|
||
return nil
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func (c *ConsumerHandler) shouldPushOffline(_ context.Context, msg *sdkws.MsgData) bool {
|
||
isOfflinePush := datautil.GetSwitchFromOptions(msg.Options, constant.IsOfflinePush)
|
||
if !isOfflinePush {
|
||
return false
|
||
}
|
||
switch msg.ContentType {
|
||
case constant.RoomParticipantsConnectedNotification:
|
||
return false
|
||
case constant.RoomParticipantsDisconnectedNotification:
|
||
return false
|
||
}
|
||
return true
|
||
}
|
||
|
||
func (c *ConsumerHandler) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData, pushToUserIDs []string) ([]*msggateway.SingleMsgToUserResults, error) {
|
||
if msg != nil && msg.Status == constant.MsgStatusSending {
|
||
msg.Status = constant.MsgStatusSendSuccess
|
||
}
|
||
onlineUserIDs, offlineUserIDs, err := c.onlineCache.GetUsersOnline(ctx, pushToUserIDs)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
log.ZDebug(ctx, "GetConnsAndOnlinePush online cache", "sendID", msg.SendID, "recvID", msg.RecvID, "groupID", msg.GroupID, "sessionType", msg.SessionType, "clientMsgID", msg.ClientMsgID, "serverMsgID", msg.ServerMsgID, "offlineUserIDs", offlineUserIDs, "onlineUserIDs", onlineUserIDs)
|
||
var result []*msggateway.SingleMsgToUserResults
|
||
if len(onlineUserIDs) > 0 {
|
||
var err error
|
||
result, err = c.onlinePusher.GetConnsAndOnlinePush(ctx, msg, onlineUserIDs)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
for _, userID := range offlineUserIDs {
|
||
result = append(result, &msggateway.SingleMsgToUserResults{
|
||
UserID: userID,
|
||
})
|
||
}
|
||
return result, nil
|
||
}
|
||
|
||
func (c *ConsumerHandler) Push2Group(ctx context.Context, groupID string, msg *sdkws.MsgData) (err error) {
|
||
log.ZInfo(ctx, "Get group msg from msg_transfer and push msg", "msg", msg.String(), "groupID", groupID, "contentType", msg.ContentType)
|
||
defer func(duration time.Time) {
|
||
t := time.Since(duration)
|
||
log.ZInfo(ctx, "Get group msg from msg_transfer and push msg end", "msg", msg.String(), "groupID", groupID, "time cost", t)
|
||
}(time.Now())
|
||
var pushToUserIDs []string
|
||
if err = c.webhookBeforeGroupOnlinePush(ctx, &c.config.WebhooksConfig.BeforeGroupOnlinePush, groupID, msg,
|
||
&pushToUserIDs); err != nil {
|
||
log.ZWarn(ctx, "Push2Group webhookBeforeGroupOnlinePush failed", err, "groupID", groupID)
|
||
return err
|
||
}
|
||
log.ZDebug(ctx, "Push2Group after webhook", "groupID", groupID, "pushToUserIDsFromWebhook", pushToUserIDs, "count", len(pushToUserIDs))
|
||
|
||
err = c.groupMessagesHandler(ctx, groupID, &pushToUserIDs, msg)
|
||
if err != nil {
|
||
log.ZWarn(ctx, "Push2Group groupMessagesHandler failed", err, "groupID", groupID)
|
||
return err
|
||
}
|
||
log.ZDebug(ctx, "Push2Group after groupMessagesHandler", "groupID", groupID, "pushToUserIDs", pushToUserIDs, "count", len(pushToUserIDs))
|
||
|
||
wsResults, err := c.GetConnsAndOnlinePush(ctx, msg, pushToUserIDs)
|
||
if err != nil {
|
||
log.ZWarn(ctx, "Push2Group GetConnsAndOnlinePush failed", err, "groupID", groupID)
|
||
return err
|
||
}
|
||
log.ZDebug(ctx, "Push2Group online push completed", "groupID", groupID, "wsResultsCount", len(wsResults))
|
||
|
||
log.ZDebug(ctx, "group push result", "result", wsResults, "msg", msg)
|
||
log.ZInfo(ctx, "online group push end")
|
||
|
||
if !c.shouldPushOffline(ctx, msg) {
|
||
return nil
|
||
}
|
||
needOfflinePushUserIDs := c.onlinePusher.GetOnlinePushFailedUserIDs(ctx, msg, wsResults, &pushToUserIDs)
|
||
//filter some user, like don not disturb or don't need offline push etc.
|
||
needOfflinePushUserIDs, err = c.filterGroupMessageOfflinePush(ctx, groupID, msg, needOfflinePushUserIDs)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
log.ZInfo(ctx, "filterGroupMessageOfflinePush end")
|
||
|
||
// Use offline push messaging
|
||
if len(needOfflinePushUserIDs) > 0 {
|
||
c.asyncOfflinePush(ctx, needOfflinePushUserIDs, msg)
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func (c *ConsumerHandler) asyncOfflinePush(ctx context.Context, needOfflinePushUserIDs []string, msg *sdkws.MsgData) {
|
||
var offlinePushUserIDs []string
|
||
err := c.webhookBeforeOfflinePush(ctx, &c.config.WebhooksConfig.BeforeOfflinePush, needOfflinePushUserIDs, msg, &offlinePushUserIDs)
|
||
if err != nil {
|
||
log.ZWarn(ctx, "webhookBeforeOfflinePush failed", err, "msg", msg)
|
||
return
|
||
}
|
||
|
||
if len(offlinePushUserIDs) > 0 {
|
||
needOfflinePushUserIDs = offlinePushUserIDs
|
||
}
|
||
if err := c.pushDatabase.MsgToOfflinePushMQ(ctx, conversationutil.GenConversationUniqueKeyForSingle(msg.SendID, msg.RecvID), needOfflinePushUserIDs, msg); err != nil {
|
||
log.ZDebug(ctx, "Msg To OfflinePush MQ error", err, "needOfflinePushUserIDs",
|
||
needOfflinePushUserIDs, "msg", msg)
|
||
log.ZWarn(ctx, "Msg To OfflinePush MQ error", err, "needOfflinePushUserIDs length",
|
||
len(needOfflinePushUserIDs), "msg", msg)
|
||
prommetrics.GroupChatMsgProcessFailedCounter.Inc()
|
||
return
|
||
}
|
||
}
|
||
|
||
func (c *ConsumerHandler) groupMessagesHandler(ctx context.Context, groupID string, pushToUserIDs *[]string, msg *sdkws.MsgData) (err error) {
|
||
if len(*pushToUserIDs) == 0 {
|
||
*pushToUserIDs, err = c.groupLocalCache.GetGroupMemberIDs(ctx, groupID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
switch msg.ContentType {
|
||
case constant.MemberQuitNotification:
|
||
var tips sdkws.MemberQuitTips
|
||
if unmarshalNotificationElem(msg.Content, &tips) != nil {
|
||
return err
|
||
}
|
||
if err = c.DeleteMemberAndSetConversationSeq(ctx, groupID, []string{tips.QuitUser.UserID}); err != nil {
|
||
log.ZError(ctx, "MemberQuitNotification DeleteMemberAndSetConversationSeq", err, "groupID", groupID, "userID", tips.QuitUser.UserID)
|
||
}
|
||
// 退出群聊通知只通知群主和管理员，不通知退出者本人
|
||
case constant.MemberKickedNotification:
|
||
var tips sdkws.MemberKickedTips
|
||
if unmarshalNotificationElem(msg.Content, &tips) != nil {
|
||
return err
|
||
}
|
||
kickedUsers := datautil.Slice(tips.KickedUserList, func(e *sdkws.GroupMemberFullInfo) string { return e.UserID })
|
||
if err = c.DeleteMemberAndSetConversationSeq(ctx, groupID, kickedUsers); err != nil {
|
||
log.ZError(ctx, "MemberKickedNotification DeleteMemberAndSetConversationSeq", err, "groupID", groupID, "userIDs", kickedUsers)
|
||
}
|
||
// 被踢出群聊通知只通知群主和管理员，不通知被踢出的用户本人
|
||
case constant.GroupDismissedNotification:
|
||
if msgprocessor.IsNotification(msgprocessor.GetConversationIDByMsg(msg)) {
|
||
var tips sdkws.GroupDismissedTips
|
||
if unmarshalNotificationElem(msg.Content, &tips) != nil {
|
||
return err
|
||
}
|
||
log.ZDebug(ctx, "GroupDismissedNotificationInfo****", "groupID", groupID, "num", len(*pushToUserIDs), "list", pushToUserIDs)
|
||
if len(c.config.Share.IMAdminUser.UserIDs) > 0 {
|
||
ctx = mcontext.WithOpUserIDContext(ctx, c.config.Share.IMAdminUser.UserIDs[0])
|
||
}
|
||
defer func(groupID string) {
|
||
if err := c.groupClient.DismissGroup(ctx, groupID, true); err != nil {
|
||
log.ZError(ctx, "DismissGroup Notification clear members", err, "groupID", groupID)
|
||
}
|
||
}(groupID)
|
||
}
|
||
}
|
||
}
|
||
// 过滤通知消息，只通知群主、管理员和相关成员本人
|
||
switch msg.ContentType {
|
||
case constant.GroupMemberMutedNotification, constant.GroupMemberCancelMutedNotification:
|
||
// 禁言和取消禁言：通知群主、管理员和本人
|
||
if err := c.filterNotificationWithUser(ctx, groupID, pushToUserIDs, msg, true); err != nil {
|
||
return err
|
||
}
|
||
case constant.MemberQuitNotification, constant.MemberEnterNotification, constant.GroupMemberInfoSetNotification:
|
||
// 退出、进入、成员信息设置：只通知群主和管理员
|
||
if err := c.filterNotificationWithUser(ctx, groupID, pushToUserIDs, msg, false); err != nil {
|
||
return err
|
||
}
|
||
case constant.MemberKickedNotification:
|
||
// 被踢出：通知群主、管理员和本人
|
||
if err := c.filterNotificationWithUser(ctx, groupID, pushToUserIDs, msg, true); err != nil {
|
||
return err
|
||
}
|
||
case constant.MemberInvitedNotification:
|
||
// 被邀请：通知群主、管理员和被邀请的本人
|
||
if err := c.filterNotificationWithUser(ctx, groupID, pushToUserIDs, msg, true); err != nil {
|
||
return err
|
||
}
|
||
case constant.GroupMemberSetToAdminNotification, constant.GroupMemberSetToOrdinaryUserNotification:
|
||
// 设置为管理员/普通用户：通知群主、管理员和本人
|
||
if err := c.filterNotificationWithUser(ctx, groupID, pushToUserIDs, msg, true); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return err
|
||
}
|
||
|
||
// filterNotificationWithUser 过滤通知消息，只通知群主、管理员，可选是否包含相关用户本人
|
||
// includeUser: true 表示包含相关用户本人，false 表示排除相关用户
|
||
func (c *ConsumerHandler) filterNotificationWithUser(ctx context.Context, groupID string, pushToUserIDs *[]string, msg *sdkws.MsgData, includeUser bool) error {
|
||
notificationName := getNotificationName(msg.ContentType)
|
||
log.ZInfo(ctx, notificationName+" push filter start", "groupID", groupID, "originalPushToUserIDs", *pushToUserIDs, "originalCount", len(*pushToUserIDs), "includeUser", includeUser)
|
||
|
||
// 解析通知内容，提取相关用户ID
|
||
var relatedUserIDs []string
|
||
var excludeUserIDs []string
|
||
switch msg.ContentType {
|
||
case constant.GroupMemberMutedNotification:
|
||
var tips sdkws.GroupMemberMutedTips
|
||
if err := unmarshalNotificationElem(msg.Content, &tips); err != nil {
|
||
log.ZWarn(ctx, notificationName+" unmarshalNotificationElem failed", err, "groupID", groupID)
|
||
return err
|
||
}
|
||
if includeUser {
|
||
relatedUserIDs = append(relatedUserIDs, tips.MutedUser.UserID)
|
||
}
|
||
log.ZDebug(ctx, notificationName+" parsed tips", "mutedUserID", tips.MutedUser.UserID, "opUserID", tips.OpUser.UserID)
|
||
case constant.GroupMemberCancelMutedNotification:
|
||
var tips sdkws.GroupMemberCancelMutedTips
|
||
if err := unmarshalNotificationElem(msg.Content, &tips); err != nil {
|
||
log.ZWarn(ctx, notificationName+" unmarshalNotificationElem failed", err, "groupID", groupID)
|
||
return err
|
||
}
|
||
if includeUser {
|
||
relatedUserIDs = append(relatedUserIDs, tips.MutedUser.UserID)
|
||
}
|
||
log.ZDebug(ctx, notificationName+" parsed tips", "cancelMutedUserID", tips.MutedUser.UserID, "opUserID", tips.OpUser.UserID)
|
||
case constant.MemberQuitNotification:
|
||
var tips sdkws.MemberQuitTips
|
||
if err := unmarshalNotificationElem(msg.Content, &tips); err != nil {
|
||
log.ZWarn(ctx, notificationName+" unmarshalNotificationElem failed", err, "groupID", groupID)
|
||
return err
|
||
}
|
||
excludeUserIDs = append(excludeUserIDs, tips.QuitUser.UserID)
|
||
log.ZDebug(ctx, notificationName+" parsed tips", "quitUserID", tips.QuitUser.UserID)
|
||
case constant.MemberInvitedNotification:
|
||
var tips sdkws.MemberInvitedTips
|
||
if err := unmarshalNotificationElem(msg.Content, &tips); err != nil {
|
||
log.ZWarn(ctx, notificationName+" unmarshalNotificationElem failed", err, "groupID", groupID)
|
||
return err
|
||
}
|
||
invitedUserIDs := datautil.Slice(tips.InvitedUserList, func(e *sdkws.GroupMemberFullInfo) string { return e.UserID })
|
||
if includeUser {
|
||
relatedUserIDs = append(relatedUserIDs, invitedUserIDs...)
|
||
} else {
|
||
excludeUserIDs = append(excludeUserIDs, invitedUserIDs...)
|
||
}
|
||
log.ZDebug(ctx, notificationName+" parsed tips", "invitedUserIDs", invitedUserIDs, "opUserID", tips.OpUser.UserID)
|
||
case constant.MemberEnterNotification:
|
||
var tips sdkws.MemberEnterTips
|
||
if err := unmarshalNotificationElem(msg.Content, &tips); err != nil {
|
||
log.ZWarn(ctx, notificationName+" unmarshalNotificationElem failed", err, "groupID", groupID)
|
||
return err
|
||
}
|
||
excludeUserIDs = append(excludeUserIDs, tips.EntrantUser.UserID)
|
||
log.ZDebug(ctx, notificationName+" parsed tips", "entrantUserID", tips.EntrantUser.UserID)
|
||
case constant.MemberKickedNotification:
|
||
var tips sdkws.MemberKickedTips
|
||
if err := unmarshalNotificationElem(msg.Content, &tips); err != nil {
|
||
log.ZWarn(ctx, notificationName+" unmarshalNotificationElem failed", err, "groupID", groupID)
|
||
return err
|
||
}
|
||
kickedUserIDs := datautil.Slice(tips.KickedUserList, func(e *sdkws.GroupMemberFullInfo) string { return e.UserID })
|
||
if includeUser {
|
||
relatedUserIDs = append(relatedUserIDs, kickedUserIDs...)
|
||
} else {
|
||
excludeUserIDs = append(excludeUserIDs, kickedUserIDs...)
|
||
}
|
||
log.ZDebug(ctx, notificationName+" parsed tips", "kickedUserIDs", kickedUserIDs, "opUserID", tips.OpUser.UserID)
|
||
case constant.GroupMemberInfoSetNotification, constant.GroupMemberSetToAdminNotification, constant.GroupMemberSetToOrdinaryUserNotification:
|
||
var tips sdkws.GroupMemberInfoSetTips
|
||
if err := unmarshalNotificationElem(msg.Content, &tips); err != nil {
|
||
log.ZWarn(ctx, notificationName+" unmarshalNotificationElem failed", err, "groupID", groupID)
|
||
return err
|
||
}
|
||
if includeUser {
|
||
relatedUserIDs = append(relatedUserIDs, tips.ChangedUser.UserID)
|
||
} else {
|
||
excludeUserIDs = append(excludeUserIDs, tips.ChangedUser.UserID)
|
||
}
|
||
log.ZDebug(ctx, notificationName+" parsed tips", "changedUserID", tips.ChangedUser.UserID, "opUserID", tips.OpUser.UserID)
|
||
default:
|
||
log.ZWarn(ctx, notificationName+" unsupported notification type", nil, "contentType", msg.ContentType)
|
||
return nil
|
||
}
|
||
|
||
// 获取所有群成员信息（如果还没有获取）
|
||
allMemberIDs := *pushToUserIDs
|
||
if len(allMemberIDs) == 0 {
|
||
var err error
|
||
allMemberIDs, err = c.groupLocalCache.GetGroupMemberIDs(ctx, groupID)
|
||
if err != nil {
|
||
log.ZWarn(ctx, notificationName+" GetGroupMemberIDs failed", err, "groupID", groupID)
|
||
return err
|
||
}
|
||
log.ZDebug(ctx, notificationName+" fetched all member IDs", "groupID", groupID, "memberCount", len(allMemberIDs))
|
||
}
|
||
|
||
members, err := c.groupLocalCache.GetGroupMembers(ctx, groupID, allMemberIDs)
|
||
if err != nil {
|
||
log.ZWarn(ctx, notificationName+" GetGroupMembers failed", err, "groupID", groupID)
|
||
return err
|
||
}
|
||
log.ZDebug(ctx, notificationName+" got members", "groupID", groupID, "memberCount", len(members))
|
||
|
||
// 筛选群主和管理员
|
||
var targetUserIDs []string
|
||
var ownerUserIDs []string
|
||
var adminUserIDs []string
|
||
for _, member := range members {
|
||
// 排除相关用户
|
||
if len(excludeUserIDs) > 0 && datautil.Contain(member.UserID, excludeUserIDs...) {
|
||
continue
|
||
}
|
||
if member.RoleLevel == constant.GroupOwner {
|
||
targetUserIDs = append(targetUserIDs, member.UserID)
|
||
ownerUserIDs = append(ownerUserIDs, member.UserID)
|
||
} else if member.RoleLevel == constant.GroupAdmin {
|
||
targetUserIDs = append(targetUserIDs, member.UserID)
|
||
adminUserIDs = append(adminUserIDs, member.UserID)
|
||
}
|
||
}
|
||
log.ZDebug(ctx, notificationName+" filtered owners and admins", "ownerCount", len(ownerUserIDs), "ownerIDs", ownerUserIDs, "adminCount", len(adminUserIDs), "adminIDs", adminUserIDs)
|
||
|
||
// 添加相关用户本人（如果需要）
|
||
if includeUser && len(relatedUserIDs) > 0 {
|
||
targetUserIDs = append(targetUserIDs, relatedUserIDs...)
|
||
log.ZDebug(ctx, notificationName+" added related users", "relatedUserIDs", relatedUserIDs, "targetCountBeforeDistinct", len(targetUserIDs))
|
||
}
|
||
|
||
// 去重并更新推送列表
|
||
*pushToUserIDs = datautil.Distinct(targetUserIDs)
|
||
log.ZInfo(ctx, notificationName+" push filter completed", "groupID", groupID, "finalPushToUserIDs", *pushToUserIDs, "finalCount", len(*pushToUserIDs), "filteredFrom", len(allMemberIDs))
|
||
return nil
|
||
}
|
||
|
||
// getNotificationName 根据通知类型获取通知名称
|
||
func getNotificationName(contentType int32) string {
|
||
switch contentType {
|
||
case constant.GroupMemberMutedNotification:
|
||
return "GroupMemberMutedNotification"
|
||
case constant.GroupMemberCancelMutedNotification:
|
||
return "GroupMemberCancelMutedNotification"
|
||
case constant.MemberQuitNotification:
|
||
return "MemberQuitNotification"
|
||
case constant.MemberInvitedNotification:
|
||
return "MemberInvitedNotification"
|
||
case constant.MemberEnterNotification:
|
||
return "MemberEnterNotification"
|
||
case constant.MemberKickedNotification:
|
||
return "MemberKickedNotification"
|
||
case constant.GroupMemberInfoSetNotification:
|
||
return "GroupMemberInfoSetNotification"
|
||
case constant.GroupMemberSetToAdminNotification:
|
||
return "GroupMemberSetToAdminNotification"
|
||
case constant.GroupMemberSetToOrdinaryUserNotification:
|
||
return "GroupMemberSetToOrdinaryUserNotification"
|
||
default:
|
||
return "UnknownNotification"
|
||
}
|
||
}
|
||
|
||
func (c *ConsumerHandler) offlinePushMsg(ctx context.Context, msg *sdkws.MsgData, offlinePushUserIDs []string) error {
|
||
title, content, opts, err := c.getOfflinePushInfos(msg)
|
||
if err != nil {
|
||
log.ZError(ctx, "getOfflinePushInfos failed", err, "msg", msg)
|
||
return err
|
||
}
|
||
err = c.offlinePusher.Push(ctx, offlinePushUserIDs, title, content, opts)
|
||
if err != nil {
|
||
prommetrics.MsgOfflinePushFailedCounter.Inc()
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (c *ConsumerHandler) filterGroupMessageOfflinePush(ctx context.Context, groupID string, msg *sdkws.MsgData,
|
||
offlinePushUserIDs []string) (userIDs []string, err error) {
|
||
needOfflinePushUserIDs, err := c.conversationClient.GetConversationOfflinePushUserIDs(ctx, conversationutil.GenGroupConversationID(groupID), offlinePushUserIDs)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return needOfflinePushUserIDs, nil
|
||
}
|
||
|
||
func (c *ConsumerHandler) getOfflinePushInfos(msg *sdkws.MsgData) (title, content string, opts *options.Opts, err error) {
|
||
type AtTextElem struct {
|
||
Text string `json:"text,omitempty"`
|
||
AtUserList []string `json:"atUserList,omitempty"`
|
||
IsAtSelf bool `json:"isAtSelf"`
|
||
}
|
||
|
||
opts = &options.Opts{Signal: &options.Signal{ClientMsgID: msg.ClientMsgID}}
|
||
if msg.OfflinePushInfo != nil {
|
||
opts.IOSBadgeCount = msg.OfflinePushInfo.IOSBadgeCount
|
||
opts.IOSPushSound = msg.OfflinePushInfo.IOSPushSound
|
||
opts.Ex = msg.OfflinePushInfo.Ex
|
||
}
|
||
|
||
if msg.OfflinePushInfo != nil {
|
||
title = msg.OfflinePushInfo.Title
|
||
content = msg.OfflinePushInfo.Desc
|
||
}
|
||
if title == "" {
|
||
switch msg.ContentType {
|
||
case constant.Text:
|
||
fallthrough
|
||
case constant.Picture:
|
||
fallthrough
|
||
case constant.Voice:
|
||
fallthrough
|
||
case constant.Video:
|
||
fallthrough
|
||
case constant.File:
|
||
title = constant.ContentType2PushContent[int64(msg.ContentType)]
|
||
case constant.AtText:
|
||
ac := AtTextElem{}
|
||
_ = jsonutil.JsonStringToStruct(string(msg.Content), &ac)
|
||
case constant.SignalingNotification:
|
||
title = constant.ContentType2PushContent[constant.SignalMsg]
|
||
default:
|
||
title = constant.ContentType2PushContent[constant.Common]
|
||
}
|
||
}
|
||
if content == "" {
|
||
content = title
|
||
}
|
||
return
|
||
}
|
||
|
||
func (c *ConsumerHandler) DeleteMemberAndSetConversationSeq(ctx context.Context, groupID string, userIDs []string) error {
|
||
conversationID := msgprocessor.GetConversationIDBySessionType(constant.ReadGroupChatType, groupID)
|
||
maxSeq, err := c.msgClient.GetConversationMaxSeq(ctx, conversationID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return c.conversationClient.SetConversationMaxSeq(ctx, conversationID, userIDs, maxSeq)
|
||
}
|
||
|
||
func unmarshalNotificationElem(bytes []byte, t any) error {
|
||
var notification sdkws.NotificationElem
|
||
if err := json.Unmarshal(bytes, ¬ification); err != nil {
|
||
return err
|
||
}
|
||
return json.Unmarshal([]byte(notification.Detail), t)
|
||
}
|