lan
d0852f0e78
Update the chat message push mechanism to include group ID and group avatar when sending messages in group conversations. This ensures that push notifications for group chats display the group's identity and icon rather than the sender's personal avatar. - Update `PushChatMessage` interface and implementation to accept `groupID` and `groupAvatar`. - Modify `chat_service.go` to extract group metadata and pass it to the push service. - Update `push_service.go` to include group details in the push payload extras and set the notification icon to the group avatar for group conversations.
918 lines
28 KiB
Go
918 lines
28 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"time"
|
||
|
||
"with_you/internal/dto"
|
||
"with_you/internal/model"
|
||
"with_you/internal/pkg/jpush"
|
||
"with_you/internal/pkg/ws"
|
||
"with_you/internal/repository"
|
||
|
||
"go.uber.org/zap"
|
||
)
|
||
|
||
// 推送相关常量
|
||
const (
|
||
// DefaultPushTimeout 默认推送超时时间
|
||
DefaultPushTimeout = 30 * time.Second
|
||
// MaxRetryCount 最大重试次数
|
||
MaxRetryCount = 3
|
||
// DefaultExpiredTime 默认消息过期时间（24小时）
|
||
DefaultExpiredTime = 24 * time.Hour
|
||
// PushQueueSize 推送队列大小
|
||
PushQueueSize = 1000
|
||
// MaxDevicesPerUser 每个用户最大设备数
|
||
MaxDevicesPerUser = 10
|
||
)
|
||
|
||
// ChatMessageSender 聊天消息发送者信息
|
||
type ChatMessageSender struct {
|
||
ID string
|
||
Name string
|
||
Avatar string
|
||
}
|
||
|
||
// PushPriority 推送优先级
|
||
type PushPriority int
|
||
|
||
const (
|
||
PriorityLow PushPriority = 1 // 低优先级（营销消息等）
|
||
PriorityNormal PushPriority = 5 // 普通优先级（系统通知）
|
||
PriorityHigh PushPriority = 8 // 高优先级（聊天消息）
|
||
PriorityCritical PushPriority = 10 // 最高优先级（重要系统通知）
|
||
)
|
||
|
||
// PushService 推送服务接口
|
||
type PushService interface {
|
||
// 推送核心方法
|
||
PushMessage(ctx context.Context, userID string, message *model.Message) error
|
||
PushToUser(ctx context.Context, userID string, message *model.Message, priority int) error
|
||
|
||
// 聊天消息推送（含免打扰检查）
|
||
// 该方法会：1. 检查用户是否在线（在线则跳过）；2. 检查会话免打扰（免打扰则跳过）；3. 通过 JPush 推送
|
||
PushChatMessage(ctx context.Context, userID string, conversationID string, sender *ChatMessageSender, convType model.ConversationType, convName string, message *model.Message, groupID string, groupAvatar string) error
|
||
|
||
// 系统消息推送
|
||
PushSystemMessage(ctx context.Context, userID string, msgType, title, content string, data map[string]any) error
|
||
|
||
// 系统通知推送（新接口，使用独立的 SystemNotification 模型）
|
||
PushSystemNotification(ctx context.Context, userID string, notification *model.SystemNotification) error
|
||
|
||
// 设备管理
|
||
RegisterDevice(ctx context.Context, userID string, deviceID string, deviceType model.DeviceType, pushToken string) error
|
||
UnregisterDevice(ctx context.Context, userID string, deviceID string) error
|
||
UpdateDeviceToken(ctx context.Context, userID string, deviceID string, newPushToken string) error
|
||
|
||
// 推送记录管理
|
||
CreatePushRecord(ctx context.Context, userID string, messageID string, channel model.PushChannel) (*model.PushRecord, error)
|
||
GetPendingPushes(ctx context.Context, userID string) ([]*model.PushRecord, error)
|
||
GetUserDevices(ctx context.Context, userID string) ([]*model.DeviceToken, error)
|
||
|
||
// 后台任务
|
||
StartPushWorker(ctx context.Context)
|
||
StopPushWorker()
|
||
}
|
||
|
||
// pushServiceImpl 推送服务实现
|
||
type pushServiceImpl struct {
|
||
pushRepo repository.PushRecordRepository
|
||
deviceRepo repository.DeviceTokenRepository
|
||
messageRepo repository.MessageRepository
|
||
wsHub ws.MessagePublisher
|
||
jpushClient *jpush.Client
|
||
|
||
pushQueue chan *pushTask
|
||
stopChan chan struct{}
|
||
}
|
||
|
||
// pushTask 推送任务
|
||
type pushTask struct {
|
||
userID string
|
||
message *model.Message
|
||
priority int
|
||
recordID int64 // DB record ID for crash recovery
|
||
}
|
||
|
||
// NewPushService 创建推送服务
|
||
func NewPushService(
|
||
pushRepo repository.PushRecordRepository,
|
||
deviceRepo repository.DeviceTokenRepository,
|
||
messageRepo repository.MessageRepository,
|
||
publisher ws.MessagePublisher,
|
||
jpushClient *jpush.Client,
|
||
) PushService {
|
||
return &pushServiceImpl{
|
||
pushRepo: pushRepo,
|
||
deviceRepo: deviceRepo,
|
||
messageRepo: messageRepo,
|
||
wsHub: publisher,
|
||
jpushClient: jpushClient,
|
||
pushQueue: make(chan *pushTask, PushQueueSize),
|
||
stopChan: make(chan struct{}),
|
||
}
|
||
}
|
||
|
||
// PushMessage 推送消息给用户
|
||
func (s *pushServiceImpl) PushMessage(ctx context.Context, userID string, message *model.Message) error {
|
||
return s.PushToUser(ctx, userID, message, int(PriorityNormal))
|
||
}
|
||
|
||
// PushToUser 带优先级的推送
|
||
func (s *pushServiceImpl) PushToUser(ctx context.Context, userID string, message *model.Message, priority int) error {
|
||
// 首先尝试WebSocket推送（实时推送）
|
||
if s.pushViaWebSocket(ctx, userID, message) {
|
||
// WebSocket推送成功，记录推送状态
|
||
record, err := s.CreatePushRecord(ctx, userID, message.ID, model.PushChannelWebSocket)
|
||
if err != nil {
|
||
return fmt.Errorf("failed to create push record: %w", err)
|
||
}
|
||
record.MarkPushed()
|
||
if err := s.pushRepo.Update(record); err != nil {
|
||
return fmt.Errorf("failed to update push record: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// 用户不在线，尝试通过极光推送直接发送
|
||
if s.isJPushAvailable() {
|
||
devices, err := s.deviceRepo.GetActiveByUserID(userID)
|
||
if err == nil && len(devices) > 0 {
|
||
mobileDevices := getMobileDevices(devices)
|
||
if len(mobileDevices) > 0 {
|
||
payload := buildMessagePayload(message)
|
||
pushErr := s.pushViaJPushBatch(mobileDevices, payload.Title, payload.Content, payload.Extras)
|
||
if pushErr == nil {
|
||
s.batchCreatePushedRecords(ctx, userID, message.ID, mobileDevices)
|
||
return nil
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// 极光推送也失败，先持久化推送记录再入内存队列
|
||
expiredAt := time.Now().Add(DefaultExpiredTime)
|
||
record := &model.PushRecord{
|
||
UserID: userID,
|
||
MessageID: message.ID,
|
||
PushChannel: model.PushChannelJPush,
|
||
PushStatus: model.PushStatusPending,
|
||
MaxRetry: MaxRetryCount,
|
||
ExpiredAt: &expiredAt,
|
||
}
|
||
if err := s.pushRepo.Create(record); err != nil {
|
||
return fmt.Errorf("failed to persist push record: %w", err)
|
||
}
|
||
|
||
select {
|
||
case s.pushQueue <- &pushTask{
|
||
userID: userID,
|
||
message: message,
|
||
priority: priority,
|
||
recordID: record.ID,
|
||
}:
|
||
return nil
|
||
default:
|
||
// 队列已满，记录已持久化，等待重试协程处理
|
||
return errors.New("push queue is full, message persisted for later delivery")
|
||
}
|
||
}
|
||
|
||
// pushViaWebSocket 通过WebSocket推送消息
|
||
// 返回true表示推送成功，false表示用户不在线
|
||
func (s *pushServiceImpl) pushViaWebSocket(ctx context.Context, userID string, message *model.Message) bool {
|
||
if s.wsHub == nil || !s.wsHub.HasClients(userID) {
|
||
return false
|
||
}
|
||
|
||
// 判断是否为系统消息/通知消息
|
||
if message.IsSystemMessage() || message.Category == model.CategoryNotification {
|
||
// 使用 NotificationMessage 格式推送系统通知
|
||
// 从 segments 中提取文本内容
|
||
content := dto.ExtractTextContentFromModel(message.Segments)
|
||
|
||
notification := map[string]any{
|
||
"id": fmt.Sprintf("%s", message.ID),
|
||
"type": "notification",
|
||
"system_type": string(message.SystemType),
|
||
"content": content,
|
||
"is_read": false,
|
||
"extra_data": map[string]any{},
|
||
"created_at": message.CreatedAt.Format("2006-01-02T15:04:05Z07:00"),
|
||
}
|
||
|
||
// 填充额外数据
|
||
if message.ExtraData != nil {
|
||
extra := notification["extra_data"].(map[string]any)
|
||
extra["actor_id"] = message.ExtraData.ActorID
|
||
extra["actor_name"] = message.ExtraData.ActorName
|
||
extra["avatar_url"] = message.ExtraData.AvatarURL
|
||
extra["target_id"] = message.ExtraData.TargetID
|
||
extra["target_type"] = message.ExtraData.TargetType
|
||
extra["action_url"] = message.ExtraData.ActionURL
|
||
extra["action_time"] = message.ExtraData.ActionTime
|
||
if message.ExtraData.ActorID > 0 {
|
||
notification["trigger_user"] = map[string]any{
|
||
"id": fmt.Sprintf("%d", message.ExtraData.ActorID),
|
||
"username": message.ExtraData.ActorName,
|
||
"avatar": message.ExtraData.AvatarURL,
|
||
}
|
||
}
|
||
}
|
||
s.wsHub.PublishToUserOnline(userID, "system_notification", notification)
|
||
return true
|
||
}
|
||
|
||
// 构建普通聊天消息的 WebSocket 消息 - 使用新的 WSEventResponse 格式
|
||
// 根据 Category 推断 detailType，群聊会话需要查询 Conversation 表获取准确类型
|
||
detailType := "private"
|
||
if message.Category == model.CategoryNotification || message.IsSystemMessage() {
|
||
detailType = "system"
|
||
}
|
||
|
||
// 直接使用 message.Segments
|
||
segments := message.Segments
|
||
|
||
event := &dto.WSEventResponse{
|
||
ID: fmt.Sprintf("%s", message.ID),
|
||
Time: message.CreatedAt.UnixMilli(),
|
||
Type: "message",
|
||
DetailType: detailType,
|
||
ConversationID: message.ConversationID,
|
||
Seq: fmt.Sprintf("%d", message.Seq),
|
||
Segments: segments,
|
||
SenderID: message.SenderID,
|
||
}
|
||
|
||
s.wsHub.PublishToUser(userID, "chat_message", map[string]any{
|
||
"detail_type": detailType,
|
||
"message": event,
|
||
})
|
||
return true
|
||
}
|
||
|
||
// messagePayload 推送消息的标题、内容和附加数据
|
||
type messagePayload struct {
|
||
Title string
|
||
Content string
|
||
Extras map[string]any
|
||
}
|
||
|
||
// buildMessagePayload 从消息中构建推送所需的标题、内容和附加数据
|
||
func buildMessagePayload(message *model.Message) messagePayload {
|
||
_ = message.Decrypt()
|
||
|
||
title := "新消息"
|
||
content := dto.ExtractTextContentFromModel(message.Segments)
|
||
if message.IsSystemMessage() || message.Category == model.CategoryNotification {
|
||
if message.ExtraData != nil && message.ExtraData.ActorName != "" {
|
||
title = message.ExtraData.ActorName
|
||
} else {
|
||
title = "系统通知"
|
||
}
|
||
}
|
||
|
||
extras := map[string]any{
|
||
"message_id": message.ID,
|
||
"conversation_id": message.ConversationID,
|
||
"sender_id": message.SenderID,
|
||
"category": string(message.Category),
|
||
}
|
||
if message.ExtraData != nil {
|
||
extras["actor_name"] = message.ExtraData.ActorName
|
||
extras["avatar_url"] = message.ExtraData.AvatarURL
|
||
extras["target_id"] = message.ExtraData.TargetID
|
||
extras["target_type"] = message.ExtraData.TargetType
|
||
}
|
||
|
||
return messagePayload{
|
||
Title: title,
|
||
Content: content,
|
||
Extras: extras,
|
||
}
|
||
}
|
||
|
||
// pushViaJPush 通过极光推送发送通知
|
||
func (s *pushServiceImpl) pushViaJPush(ctx context.Context, device *model.DeviceToken, title, content string, extras map[string]any) error {
|
||
if !s.isJPushAvailable() {
|
||
return errors.New("jpush client not configured")
|
||
}
|
||
|
||
if device.PushToken == "" {
|
||
return errors.New("device has no registration ID")
|
||
}
|
||
|
||
notification := jpush.BuildNotification(title, content, "message", extras)
|
||
_, err := s.jpushClient.PushByRegistrationIDs([]string{device.PushToken}, notification)
|
||
if err != nil {
|
||
return fmt.Errorf("jpush push failed: %w", err)
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// batchCreatePushedRecords 批量创建已推送成功的记录
|
||
func (s *pushServiceImpl) batchCreatePushedRecords(ctx context.Context, userID string, messageID string, devices []*model.DeviceToken) {
|
||
expiredAt := time.Now().Add(DefaultExpiredTime)
|
||
records := make([]*model.PushRecord, 0, len(devices))
|
||
for _, device := range devices {
|
||
now := time.Now()
|
||
record := &model.PushRecord{
|
||
UserID: userID,
|
||
MessageID: messageID,
|
||
PushChannel: model.PushChannelJPush,
|
||
PushStatus: model.PushStatusPushed,
|
||
DeviceToken: device.PushToken,
|
||
DeviceType: string(device.DeviceType),
|
||
MaxRetry: MaxRetryCount,
|
||
ExpiredAt: &expiredAt,
|
||
PushedAt: &now,
|
||
}
|
||
records = append(records, record)
|
||
}
|
||
if err := s.pushRepo.BatchCreate(records); err != nil {
|
||
zap.L().Error("batch create pushed records failed",
|
||
zap.String("userID", userID),
|
||
zap.String("messageID", messageID),
|
||
zap.Error(err),
|
||
)
|
||
}
|
||
}
|
||
|
||
// pushViaJPushBatch 批量通过极光推送发送通知
|
||
func (s *pushServiceImpl) pushViaJPushBatch(devices []*model.DeviceToken, title, content string, extras map[string]any) error {
|
||
if !s.isJPushAvailable() {
|
||
return errors.New("jpush client not configured")
|
||
}
|
||
|
||
var regIDs []string
|
||
for _, d := range devices {
|
||
if d.PushToken != "" {
|
||
regIDs = append(regIDs, d.PushToken)
|
||
}
|
||
}
|
||
|
||
if len(regIDs) == 0 {
|
||
return errors.New("no valid registration IDs")
|
||
}
|
||
|
||
notification := jpush.BuildNotification(title, content, "message", extras)
|
||
_, err := s.jpushClient.PushByRegistrationIDs(regIDs, notification)
|
||
if err != nil {
|
||
return fmt.Errorf("jpush batch push failed: %w", err)
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// RegisterDevice 注册设备
|
||
func (s *pushServiceImpl) RegisterDevice(ctx context.Context, userID string, deviceID string, deviceType model.DeviceType, pushToken string) error {
|
||
existing, err := s.deviceRepo.GetByUserID(userID)
|
||
if err == nil && len(existing) >= MaxDevicesPerUser {
|
||
return fmt.Errorf("exceeded maximum device limit (%d)", MaxDevicesPerUser)
|
||
}
|
||
|
||
deviceToken := &model.DeviceToken{
|
||
UserID: userID,
|
||
DeviceID: deviceID,
|
||
DeviceType: deviceType,
|
||
PushToken: pushToken,
|
||
IsActive: true,
|
||
}
|
||
deviceToken.UpdateLastUsed()
|
||
|
||
return s.deviceRepo.Upsert(deviceToken)
|
||
}
|
||
|
||
// UnregisterDevice 注销设备
|
||
func (s *pushServiceImpl) UnregisterDevice(ctx context.Context, userID string, deviceID string) error {
|
||
device, err := s.deviceRepo.GetByDeviceID(deviceID)
|
||
if err != nil {
|
||
return fmt.Errorf("device not found: %w", err)
|
||
}
|
||
if device.UserID != userID {
|
||
return fmt.Errorf("device does not belong to current user")
|
||
}
|
||
return s.deviceRepo.Deactivate(deviceID)
|
||
}
|
||
|
||
// UpdateDeviceToken 更新设备Token
|
||
func (s *pushServiceImpl) UpdateDeviceToken(ctx context.Context, userID string, deviceID string, newPushToken string) error {
|
||
deviceToken, err := s.deviceRepo.GetByDeviceID(deviceID)
|
||
if err != nil {
|
||
return fmt.Errorf("device not found: %w", err)
|
||
}
|
||
if deviceToken.UserID != userID {
|
||
return fmt.Errorf("device does not belong to current user")
|
||
}
|
||
|
||
deviceToken.PushToken = newPushToken
|
||
deviceToken.Activate()
|
||
|
||
return s.deviceRepo.Update(deviceToken)
|
||
}
|
||
|
||
// CreatePushRecord 创建推送记录
|
||
func (s *pushServiceImpl) CreatePushRecord(ctx context.Context, userID string, messageID string, channel model.PushChannel) (*model.PushRecord, error) {
|
||
expiredAt := time.Now().Add(DefaultExpiredTime)
|
||
record := &model.PushRecord{
|
||
UserID: userID,
|
||
MessageID: messageID,
|
||
PushChannel: channel,
|
||
PushStatus: model.PushStatusPending,
|
||
MaxRetry: MaxRetryCount,
|
||
ExpiredAt: &expiredAt,
|
||
}
|
||
|
||
if err := s.pushRepo.Create(record); err != nil {
|
||
return nil, fmt.Errorf("failed to create push record: %w", err)
|
||
}
|
||
|
||
return record, nil
|
||
}
|
||
|
||
// GetPendingPushes 获取待推送记录
|
||
func (s *pushServiceImpl) GetPendingPushes(ctx context.Context, userID string) ([]*model.PushRecord, error) {
|
||
return s.pushRepo.GetByUserID(userID, 100, 0)
|
||
}
|
||
|
||
// GetUserDevices 获取用户设备列表
|
||
func (s *pushServiceImpl) GetUserDevices(ctx context.Context, userID string) ([]*model.DeviceToken, error) {
|
||
return s.deviceRepo.GetByUserID(userID)
|
||
}
|
||
|
||
// StartPushWorker 启动推送工作协程
|
||
func (s *pushServiceImpl) StartPushWorker(ctx context.Context) {
|
||
s.recoverPendingPushes()
|
||
go s.processPushQueue()
|
||
go s.retryFailedPushes()
|
||
}
|
||
|
||
// recoverPendingPushes 从数据库恢复未处理的推送记录
|
||
func (s *pushServiceImpl) recoverPendingPushes() {
|
||
records, err := s.pushRepo.GetPendingPushes(100)
|
||
if err != nil {
|
||
zap.L().Error("failed to recover pending push records", zap.Error(err))
|
||
return
|
||
}
|
||
|
||
recovered := 0
|
||
for _, record := range records {
|
||
if record.IsExpired() {
|
||
s.pushRepo.UpdateStatus(record.ID, model.PushStatusExpired)
|
||
continue
|
||
}
|
||
|
||
message, err := s.messageRepo.GetMessageByID(record.MessageID)
|
||
if err != nil {
|
||
s.pushRepo.MarkAsFailed(record.ID, "message not found during recovery")
|
||
continue
|
||
}
|
||
|
||
select {
|
||
case s.pushQueue <- &pushTask{
|
||
userID: record.UserID,
|
||
message: message,
|
||
priority: int(PriorityNormal),
|
||
recordID: record.ID,
|
||
}:
|
||
recovered++
|
||
default:
|
||
// 队列已满，保留 pending 状态等待下次恢复
|
||
}
|
||
}
|
||
|
||
if recovered > 0 {
|
||
zap.L().Info("recovered pending push records", zap.Int("count", recovered))
|
||
}
|
||
}
|
||
|
||
// StopPushWorker 停止推送工作协程
|
||
func (s *pushServiceImpl) StopPushWorker() {
|
||
close(s.stopChan)
|
||
}
|
||
|
||
// processPushQueue 处理推送队列
|
||
func (s *pushServiceImpl) processPushQueue() {
|
||
for {
|
||
select {
|
||
case <-s.stopChan:
|
||
return
|
||
case task := <-s.pushQueue:
|
||
s.processPushTask(task)
|
||
}
|
||
}
|
||
}
|
||
|
||
// processPushTask 处理单个推送任务
|
||
func (s *pushServiceImpl) processPushTask(task *pushTask) {
|
||
ctx, cancel := context.WithTimeout(context.Background(), DefaultPushTimeout)
|
||
defer cancel()
|
||
|
||
// 获取用户活跃设备
|
||
devices, err := s.deviceRepo.GetActiveByUserID(task.userID)
|
||
if err != nil || len(devices) == 0 {
|
||
// 没有可用设备，保留记录为 pending 等待重试
|
||
s.pushRepo.UpdateStatus(task.recordID, model.PushStatusFailed)
|
||
return
|
||
}
|
||
|
||
payload := buildMessagePayload(task.message)
|
||
|
||
// 尝试使用极光推送（统一推送到所有移动设备）
|
||
mobileDevices := getMobileDevices(devices)
|
||
|
||
if len(mobileDevices) > 0 {
|
||
// 优先使用极光推送
|
||
if s.isJPushAvailable() {
|
||
err := s.pushViaJPushBatch(mobileDevices, payload.Title, payload.Content, payload.Extras)
|
||
if err == nil {
|
||
s.batchCreatePushedRecords(ctx, task.userID, task.message.ID, mobileDevices)
|
||
// 原始 pending 记录标记为已推送（被批量记录替代）
|
||
s.pushRepo.UpdateStatus(task.recordID, model.PushStatusPushed)
|
||
return
|
||
}
|
||
}
|
||
|
||
// 极光推送不可用或批量推送失败，逐个设备处理
|
||
for _, device := range mobileDevices {
|
||
channel := s.getChannelForDevice(device)
|
||
record, rerr := s.CreatePushRecord(ctx, task.userID, task.message.ID, channel)
|
||
if rerr != nil {
|
||
continue
|
||
}
|
||
record.DeviceToken = device.PushToken
|
||
record.DeviceType = string(device.DeviceType)
|
||
|
||
var pushErr error
|
||
if s.isJPushAvailable() {
|
||
pushErr = s.pushViaJPush(ctx, device, payload.Title, payload.Content, payload.Extras)
|
||
} else {
|
||
pushErr = errors.New("jpush not configured")
|
||
}
|
||
|
||
if pushErr != nil {
|
||
record.MarkFailed(pushErr.Error())
|
||
} else {
|
||
record.MarkPushed()
|
||
}
|
||
s.pushRepo.Update(record)
|
||
}
|
||
// 原始 pending 记录标记为已处理
|
||
s.pushRepo.UpdateStatus(task.recordID, model.PushStatusPushed)
|
||
}
|
||
}
|
||
|
||
// getChannelForDevice 根据设备类型获取推送通道
|
||
func (s *pushServiceImpl) getChannelForDevice(device *model.DeviceToken) model.PushChannel {
|
||
switch device.DeviceType {
|
||
case model.DeviceTypeIOS:
|
||
return model.PushChannelJPush
|
||
case model.DeviceTypeAndroid:
|
||
return model.PushChannelJPush
|
||
default:
|
||
return model.PushChannelWebSocket
|
||
}
|
||
}
|
||
|
||
// isJPushAvailable 检查极光推送是否可用
|
||
func (s *pushServiceImpl) isJPushAvailable() bool {
|
||
return s.jpushClient != nil
|
||
}
|
||
|
||
// getMobileDevices 从设备列表中筛选出支持手机推送的活跃设备
|
||
func getMobileDevices(devices []*model.DeviceToken) []*model.DeviceToken {
|
||
mobileDevices := make([]*model.DeviceToken, 0, len(devices))
|
||
for _, d := range devices {
|
||
if d.SupportsMobilePush() && d.PushToken != "" {
|
||
mobileDevices = append(mobileDevices, d)
|
||
}
|
||
}
|
||
return mobileDevices
|
||
}
|
||
|
||
// retryFailedPushes 重试失败的推送（使用指数退避）
|
||
func (s *pushServiceImpl) retryFailedPushes() {
|
||
baseInterval := 5 * time.Minute
|
||
maxInterval := 30 * time.Minute
|
||
currentInterval := baseInterval
|
||
|
||
ticker := time.NewTicker(currentInterval)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-s.stopChan:
|
||
return
|
||
case <-ticker.C:
|
||
retryCount := s.doRetry()
|
||
if retryCount > 0 {
|
||
currentInterval = baseInterval
|
||
} else {
|
||
currentInterval = currentInterval * 2
|
||
if currentInterval > maxInterval {
|
||
currentInterval = maxInterval
|
||
}
|
||
}
|
||
ticker.Reset(currentInterval)
|
||
}
|
||
}
|
||
}
|
||
|
||
// doRetry 执行重试，返回本次成功重试的数量
|
||
func (s *pushServiceImpl) doRetry() int {
|
||
ctx := context.Background()
|
||
|
||
// 获取失败待重试的推送
|
||
records, err := s.pushRepo.GetFailedPushesForRetry(100)
|
||
if err != nil {
|
||
return 0
|
||
}
|
||
|
||
retriedCount := 0
|
||
for _, record := range records {
|
||
// 检查是否过期
|
||
if record.IsExpired() {
|
||
record.MarkExpired()
|
||
s.pushRepo.Update(record)
|
||
continue
|
||
}
|
||
|
||
// 获取消息
|
||
message, err := s.messageRepo.GetMessageByID(record.MessageID)
|
||
if err != nil {
|
||
record.MarkFailed("message not found")
|
||
s.pushRepo.Update(record)
|
||
continue
|
||
}
|
||
|
||
// 尝试WebSocket推送
|
||
if s.pushViaWebSocket(ctx, record.UserID, message) {
|
||
record.MarkDelivered()
|
||
s.pushRepo.Update(record)
|
||
retriedCount++
|
||
continue
|
||
}
|
||
|
||
// 获取设备并尝试移动端推送
|
||
if record.DeviceToken != "" {
|
||
device, err := s.deviceRepo.GetByPushToken(record.DeviceToken)
|
||
if err != nil {
|
||
record.MarkFailed("device not found")
|
||
s.pushRepo.Update(record)
|
||
continue
|
||
}
|
||
|
||
var pushErr error
|
||
if s.isJPushAvailable() && device.SupportsMobilePush() && device.PushToken != "" {
|
||
payload := buildMessagePayload(message)
|
||
pushErr = s.pushViaJPush(ctx, device, payload.Title, payload.Content, payload.Extras)
|
||
}
|
||
|
||
if pushErr != nil {
|
||
record.MarkFailed(pushErr.Error())
|
||
} else {
|
||
record.MarkPushed()
|
||
retriedCount++
|
||
}
|
||
s.pushRepo.Update(record)
|
||
}
|
||
}
|
||
return retriedCount
|
||
}
|
||
|
||
// PushChatMessage 推送聊天消息给离线用户
|
||
// 检查用户在线状态和免打扰设置，仅对离线且未免打扰的用户通过 JPush 推送
|
||
func (s *pushServiceImpl) PushChatMessage(ctx context.Context, userID string, conversationID string, sender *ChatMessageSender, convType model.ConversationType, convName string, message *model.Message, groupID string, groupAvatar string) error {
|
||
// 1. 用户在线则跳过（已通过 WebSocket 收到消息）
|
||
if s.wsHub != nil && s.wsHub.HasClients(userID) {
|
||
return nil
|
||
}
|
||
|
||
// 2. 检查免打扰状态
|
||
participant, err := s.messageRepo.GetParticipant(conversationID, userID)
|
||
if err == nil && participant.NotificationMuted {
|
||
zap.L().Debug("push chat message skipped: notification muted",
|
||
zap.String("userID", userID),
|
||
zap.String("conversationID", conversationID),
|
||
)
|
||
return nil
|
||
}
|
||
|
||
// 3. 获取用户的移动设备
|
||
if !s.isJPushAvailable() {
|
||
return nil
|
||
}
|
||
devices, err := s.deviceRepo.GetActiveByUserID(userID)
|
||
if err != nil || len(devices) == 0 {
|
||
return nil
|
||
}
|
||
|
||
mobileDevices := getMobileDevices(devices)
|
||
if len(mobileDevices) == 0 {
|
||
return nil
|
||
}
|
||
|
||
var regIDs []string
|
||
for _, d := range mobileDevices {
|
||
regIDs = append(regIDs, d.PushToken)
|
||
}
|
||
|
||
// 4. 构建通知内容
|
||
title := sender.Name
|
||
if title == "" {
|
||
title = "新消息"
|
||
}
|
||
if convType == model.ConversationTypeGroup && convName != "" {
|
||
title = convName + ": " + sender.Name
|
||
}
|
||
content := dto.ExtractTextContentFromModel(message.Segments)
|
||
if content == "" {
|
||
content = "[非文本消息]"
|
||
}
|
||
extras := map[string]any{
|
||
"notification_type": "chat_message",
|
||
"message_id": message.ID,
|
||
"conversation_id": message.ConversationID,
|
||
"conversation_type": string(convType),
|
||
"sender_id": message.SenderID,
|
||
"sender_name": sender.Name,
|
||
"sender_avatar": sender.Avatar,
|
||
"category": string(message.Category),
|
||
}
|
||
if convType == model.ConversationTypeGroup {
|
||
extras["group_id"] = groupID
|
||
extras["group_name"] = convName
|
||
extras["group_avatar"] = groupAvatar
|
||
}
|
||
if message.Category == model.CategoryNotification && message.ExtraData != nil {
|
||
title = message.ExtraData.ActorName
|
||
if title == "" {
|
||
title = "系统通知"
|
||
}
|
||
extras["actor_name"] = message.ExtraData.ActorName
|
||
extras["actor_id"] = message.ExtraData.ActorID
|
||
extras["avatar_url"] = message.ExtraData.AvatarURL
|
||
extras["target_id"] = message.ExtraData.TargetID
|
||
extras["target_type"] = message.ExtraData.TargetType
|
||
}
|
||
|
||
notification := jpush.BuildNotification(title, content, "chat_message", extras)
|
||
// 群聊用群头像，私聊用发送者头像
|
||
if convType == model.ConversationTypeGroup && groupAvatar != "" {
|
||
notification.Android.LargeIcon = groupAvatar
|
||
} else if sender.Avatar != "" {
|
||
notification.Android.LargeIcon = sender.Avatar
|
||
}
|
||
if _, pushErr := s.jpushClient.PushByRegistrationIDs(regIDs, notification); pushErr != nil {
|
||
zap.L().Error("jpush push chat message failed",
|
||
zap.String("userID", userID),
|
||
zap.String("conversationID", conversationID),
|
||
zap.Error(pushErr),
|
||
)
|
||
return pushErr
|
||
}
|
||
|
||
zap.L().Debug("jpush push chat message sent",
|
||
zap.String("userID", userID),
|
||
zap.String("conversationID", conversationID),
|
||
zap.String("convType", string(convType)),
|
||
zap.Int("device_count", len(regIDs)),
|
||
)
|
||
return nil
|
||
}
|
||
|
||
// PushSystemMessage 推送系统消息
|
||
func (s *pushServiceImpl) PushSystemMessage(ctx context.Context, userID string, msgType, title, content string, data map[string]any) error {
|
||
// 首先尝试WebSocket推送
|
||
if s.pushSystemViaWebSocket(ctx, userID, msgType, title, content, data) {
|
||
return nil
|
||
}
|
||
|
||
// 用户不在线，创建待推送记录（移动端上线后可通过其他方式获取）
|
||
// 系统消息通常不需要离线推送，客户端上线后会主动拉取
|
||
return errors.New("user is offline, system message will be available on next sync")
|
||
}
|
||
|
||
// pushSystemViaWebSocket 通过WebSocket推送系统消息
|
||
// 使用 PublishToUserOnline 而非 PublishToUser，避免将系统消息存入断线重连历史回放
|
||
func (s *pushServiceImpl) pushSystemViaWebSocket(ctx context.Context, userID string, msgType, title, content string, data map[string]any) bool {
|
||
if s.wsHub == nil || !s.wsHub.HasClients(userID) {
|
||
return false
|
||
}
|
||
|
||
sysMsg := map[string]any{
|
||
"type": "notification",
|
||
"system_type": msgType,
|
||
"title": title,
|
||
"content": content,
|
||
"is_read": false,
|
||
"extra_data": data,
|
||
"created_at": time.Now().Format("2006-01-02T15:04:05Z07:00"),
|
||
}
|
||
s.wsHub.PublishToUserOnline(userID, "system_notification", sysMsg)
|
||
return true
|
||
}
|
||
|
||
// PushSystemNotification 推送系统通知（使用独立的 SystemNotification 模型）
|
||
func (s *pushServiceImpl) PushSystemNotification(ctx context.Context, userID string, notification *model.SystemNotification) error {
|
||
// 首先尝试WebSocket推送
|
||
if s.pushSystemNotificationViaWebSocket(ctx, userID, notification) {
|
||
return nil
|
||
}
|
||
|
||
// 用户不在线，尝试极光推送
|
||
if s.isJPushAvailable() {
|
||
devices, err := s.deviceRepo.GetActiveByUserID(userID)
|
||
if err == nil && len(devices) > 0 {
|
||
mobileDevices := getMobileDevices(devices)
|
||
if len(mobileDevices) > 0 {
|
||
content := notification.Content
|
||
title := notification.Title
|
||
if title == "" {
|
||
title = string(notification.Type)
|
||
}
|
||
extras := map[string]any{
|
||
"notification_id": fmt.Sprintf("%d", notification.ID),
|
||
"notification_type": string(notification.Type),
|
||
}
|
||
if notification.ExtraData != nil {
|
||
extras["actor_id"] = notification.ExtraData.ActorIDStr
|
||
extras["actor_name"] = notification.ExtraData.ActorName
|
||
extras["avatar_url"] = notification.ExtraData.AvatarURL
|
||
extras["target_id"] = notification.ExtraData.TargetID
|
||
extras["target_type"] = notification.ExtraData.TargetType
|
||
}
|
||
notifType := string(notification.Type)
|
||
jpushNotif := jpush.BuildNotification(title, content, notifType, extras)
|
||
if notification.ExtraData != nil && notification.ExtraData.AvatarURL != "" {
|
||
jpushNotif.Android.LargeIcon = notification.ExtraData.AvatarURL
|
||
}
|
||
_, pushErr := s.jpushClient.PushByRegistrationIDs(
|
||
func() []string {
|
||
ids := make([]string, 0, len(mobileDevices))
|
||
for _, d := range mobileDevices {
|
||
ids = append(ids, d.PushToken)
|
||
}
|
||
return ids
|
||
}(),
|
||
jpushNotif,
|
||
)
|
||
if pushErr == nil {
|
||
return nil
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// 系统通知已持久化到数据库中，用户上线后会主动拉取
|
||
return nil
|
||
}
|
||
|
||
// pushSystemNotificationViaWebSocket 通过WebSocket推送系统通知
|
||
// 使用 PublishToUserOnline 而非 PublishToUser，避免将通知存入断线重连历史回放
|
||
// 系统通知已持久化到数据库，客户端可通过 REST API 拉取，无需在重连时回放
|
||
func (s *pushServiceImpl) pushSystemNotificationViaWebSocket(ctx context.Context, userID string, notification *model.SystemNotification) bool {
|
||
if s.wsHub == nil || !s.wsHub.HasClients(userID) {
|
||
return false
|
||
}
|
||
|
||
wsNotification := map[string]any{
|
||
"id": fmt.Sprintf("%d", notification.ID),
|
||
"type": "notification",
|
||
"system_type": string(notification.Type),
|
||
"title": notification.Title,
|
||
"content": notification.Content,
|
||
"is_read": notification.IsRead,
|
||
"extra_data": map[string]any{},
|
||
"created_at": notification.CreatedAt.Format("2006-01-02T15:04:05Z07:00"),
|
||
}
|
||
|
||
// 填充额外数据
|
||
if notification.ExtraData != nil {
|
||
extra := wsNotification["extra_data"].(map[string]any)
|
||
extra["actor_id_str"] = notification.ExtraData.ActorIDStr
|
||
extra["actor_name"] = notification.ExtraData.ActorName
|
||
extra["avatar_url"] = notification.ExtraData.AvatarURL
|
||
extra["target_id"] = notification.ExtraData.TargetID
|
||
extra["target_type"] = notification.ExtraData.TargetType
|
||
extra["action_url"] = notification.ExtraData.ActionURL
|
||
extra["action_time"] = notification.ExtraData.ActionTime
|
||
|
||
// 设置触发用户信息
|
||
if notification.ExtraData.ActorIDStr != "" {
|
||
wsNotification["trigger_user"] = map[string]any{
|
||
"id": notification.ExtraData.ActorIDStr,
|
||
"username": notification.ExtraData.ActorName,
|
||
"avatar": notification.ExtraData.AvatarURL,
|
||
}
|
||
}
|
||
}
|
||
|
||
s.wsHub.PublishToUserOnline(userID, "system_notification", wsNotification)
|
||
return true
|
||
}
|