755392e999
Move the vendor switch logic into the ThirdPartyChannel struct as Set/IsEmpty/Normalize methods to improve encapsulation and eliminate code duplication in the vendor params builder. Also simplify messageToParams to reuse buildMessagePayload results and remove unused notification_type extras from push notifications.
991 lines
31 KiB
Go
991 lines
31 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"time"
|
||
|
||
"with_you/internal/config"
|
||
"with_you/internal/dto"
|
||
"with_you/internal/model"
|
||
"with_you/internal/pkg/jpush"
|
||
"with_you/internal/pkg/jpush/vendor"
|
||
"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
|
||
|
||
// channelConfig 厂商通道 channel_id 配置（系统消息/私聊消息分别注册）
|
||
channelConfig config.ChannelConfig
|
||
// vendorFactory 厂商转换器工厂，按统一参数产出各厂商推送字段
|
||
vendorFactory *vendor.Factory
|
||
|
||
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,
|
||
channelConfig config.ChannelConfig,
|
||
) PushService {
|
||
return &pushServiceImpl{
|
||
pushRepo: pushRepo,
|
||
deviceRepo: deviceRepo,
|
||
messageRepo: messageRepo,
|
||
wsHub: publisher,
|
||
jpushClient: jpushClient,
|
||
channelConfig: channelConfig,
|
||
vendorFactory: vendor.NewFactory(),
|
||
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 {
|
||
params := s.messageToParams(message)
|
||
pushErr := s.pushViaJPushBatch(mobileDevices, params)
|
||
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: 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 通过极光推送发送通知
|
||
// params 为厂商无关统一参数，由 vendor 工厂转换为各厂商推送字段
|
||
func (s *pushServiceImpl) pushViaJPush(ctx context.Context, device *model.DeviceToken, params vendor.MessageParams) error {
|
||
if !s.isJPushAvailable() {
|
||
return errors.New("jpush client not configured")
|
||
}
|
||
|
||
if device.PushToken == "" {
|
||
return errors.New("device has no registration ID")
|
||
}
|
||
|
||
payload := s.buildPayload(params)
|
||
_, err := s.jpushClient.PushByRegistrationIDs([]string{device.PushToken}, payload.Notification, payload.TPC)
|
||
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 批量通过极光推送发送通知
|
||
// pushViaJPushBatch 批量通过极光推送发送通知
|
||
// params 为厂商无关统一参数，由 vendor 工厂转换为各厂商推送字段
|
||
func (s *pushServiceImpl) pushViaJPushBatch(devices []*model.DeviceToken, params vendor.MessageParams) 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")
|
||
}
|
||
|
||
payload := s.buildPayload(params)
|
||
_, err := s.jpushClient.PushByRegistrationIDs(regIDs, payload.Notification, payload.TPC)
|
||
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
|
||
}
|
||
|
||
params := s.messageToParams(task.message)
|
||
|
||
// 尝试使用极光推送（统一推送到所有移动设备）
|
||
mobileDevices := getMobileDevices(devices)
|
||
|
||
if len(mobileDevices) > 0 {
|
||
// 优先使用极光推送
|
||
if s.isJPushAvailable() {
|
||
err := s.pushViaJPushBatch(mobileDevices, params)
|
||
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, params)
|
||
} 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
|
||
}
|
||
|
||
// resolveChannelScene 根据消息类型判断 channel 场景
|
||
// 系统消息/通知/公告 -> vendor.SceneSystem，其余（普通聊天）-> vendor.SceneChat
|
||
func (s *pushServiceImpl) resolveChannelScene(message *model.Message) string {
|
||
if message == nil {
|
||
return vendor.SceneChat
|
||
}
|
||
if message.IsSystemMessage() ||
|
||
message.Category == model.CategoryNotification ||
|
||
message.Category == model.CategoryAnnouncement {
|
||
return vendor.SceneSystem
|
||
}
|
||
return vendor.SceneChat
|
||
}
|
||
|
||
// buildPayload 根据统一参数，通过厂商工厂产出完整推送片段（Notification + ThirdPartyChannel）
|
||
func (s *pushServiceImpl) buildPayload(params vendor.MessageParams) *vendor.PushPayload {
|
||
return s.vendorFactory.Build(params, s.channelConfig)
|
||
}
|
||
|
||
// messageToParams 从 model.Message 提取厂商无关的统一参数
|
||
// 用于 PushToUser / worker / retry 等通用路径
|
||
// 复用 buildMessagePayload 的计算结果，避免重复解析 message
|
||
func (s *pushServiceImpl) messageToParams(message *model.Message) vendor.MessageParams {
|
||
payload := buildMessagePayload(message)
|
||
// senderName 复用 payload.Extras 中已提取的 actor_name，无需再次访问 ExtraData
|
||
senderName := ""
|
||
if name, ok := payload.Extras["actor_name"]; ok {
|
||
if n, ok := name.(string); ok {
|
||
senderName = n
|
||
}
|
||
}
|
||
return vendor.MessageParams{
|
||
Scene: s.resolveChannelScene(message),
|
||
Title: payload.Title,
|
||
Body: payload.Content,
|
||
SenderName: senderName,
|
||
MessageType: "message",
|
||
Extra: payload.Extras,
|
||
}
|
||
}
|
||
|
||
// 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 != "" {
|
||
params := s.messageToParams(message)
|
||
pushErr = s.pushViaJPush(ctx, device, params)
|
||
}
|
||
|
||
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{
|
||
"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
|
||
}
|
||
|
||
// 群聊用群头像，私聊用发送者头像
|
||
largeIcon := sender.Avatar
|
||
if convType == model.ConversationTypeGroup && groupAvatar != "" {
|
||
largeIcon = groupAvatar
|
||
}
|
||
params := vendor.MessageParams{
|
||
Scene: s.resolveChannelScene(message),
|
||
Title: title,
|
||
Body: content,
|
||
SenderName: sender.Name,
|
||
SenderID: message.SenderID,
|
||
ConversationName: convName,
|
||
ConversationID: message.ConversationID,
|
||
MessageType: "chat_message",
|
||
LargeIcon: largeIcon,
|
||
Extra: extras,
|
||
}
|
||
payload := s.buildPayload(params)
|
||
if _, pushErr := s.jpushClient.PushByRegistrationIDs(regIDs, payload.Notification, payload.TPC); 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),
|
||
}
|
||
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)
|
||
largeIcon := ""
|
||
if notification.ExtraData != nil && notification.ExtraData.AvatarURL != "" {
|
||
largeIcon = notification.ExtraData.AvatarURL
|
||
}
|
||
params := vendor.MessageParams{
|
||
Scene: vendor.SceneSystem,
|
||
Title: title,
|
||
Body: content,
|
||
MessageType: notifType,
|
||
LargeIcon: largeIcon,
|
||
Extra: extras,
|
||
}
|
||
payload := s.buildPayload(params)
|
||
_, pushErr := s.jpushClient.PushByRegistrationIDs(
|
||
func() []string {
|
||
ids := make([]string, 0, len(mobileDevices))
|
||
for _, d := range mobileDevices {
|
||
ids = append(ids, d.PushToken)
|
||
}
|
||
return ids
|
||
}(),
|
||
payload.Notification,
|
||
payload.TPC,
|
||
)
|
||
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
|
||
}
|