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 ) // 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) 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, deviceID string) error UpdateDeviceToken(ctx context.Context, 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.Hub jpushClient *jpush.Client pushQueue chan *pushTask stopChan chan struct{} } // pushTask 推送任务 type pushTask struct { userID string message *model.Message priority int } // NewPushService 创建推送服务 func NewPushService( pushRepo repository.PushRecordRepository, deviceRepo repository.DeviceTokenRepository, messageRepo repository.MessageRepository, wsHub *ws.Hub, jpushClient *jpush.Client, ) PushService { return &pushServiceImpl{ pushRepo: pushRepo, deviceRepo: deviceRepo, messageRepo: messageRepo, wsHub: wsHub, 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.jpushClient != nil { devices, err := s.deviceRepo.GetActiveByUserID(userID) if err == nil && len(devices) > 0 { mobileDevices := make([]*model.DeviceToken, 0) for _, d := range devices { if d.SupportsMobilePush() && d.PushToken != "" { mobileDevices = append(mobileDevices, d) } } if len(mobileDevices) > 0 { 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 } pushErr := s.pushViaJPushBatch(mobileDevices, title, content, extras) if pushErr == nil { for _, device := range mobileDevices { record, rerr := s.CreatePushRecord(ctx, userID, message.ID, model.PushChannelJPush) if rerr == nil { record.DeviceToken = device.PushToken record.DeviceType = string(device.DeviceType) record.MarkPushed() s.pushRepo.Update(record) } } return nil } } } } // 极光推送也失败,加入推送队列等待重试 select { case s.pushQueue <- &pushTask{ userID: userID, message: message, priority: priority, }: return nil default: // 队列已满,直接创建待推送记录 _, err := s.CreatePushRecord(ctx, userID, message.ID, model.PushChannelJPush) if err != nil { return fmt.Errorf("failed to create pending push record: %w", err) } return errors.New("push queue is full, message queued 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 格式 // 获取会话类型 (private/group) detailType := "private" if message.ConversationID != "" { // 从会话中获取类型,需要查询数据库或从消息中判断 // 这里暂时默认为 private,group 类型需要额外逻辑 } // 直接使用 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 } // pushViaJPush 通过极光推送发送通知 func (s *pushServiceImpl) pushViaJPush(ctx context.Context, device *model.DeviceToken, title, content string, extras map[string]any) error { if s.jpushClient == nil { 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, extras) if err != nil { return fmt.Errorf("jpush push failed: %w", err) } return nil } // pushViaJPushBatch 批量通过极光推送发送通知 func (s *pushServiceImpl) pushViaJPushBatch(devices []*model.DeviceToken, title, content string, extras map[string]any) error { if s.jpushClient == nil { 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, extras) 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 { 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, deviceID string) error { return s.deviceRepo.Deactivate(deviceID) } // UpdateDeviceToken 更新设备Token func (s *pushServiceImpl) UpdateDeviceToken(ctx context.Context, deviceID string, newPushToken string) error { deviceToken, err := s.deviceRepo.GetByDeviceID(deviceID) if err != nil { return fmt.Errorf("device not found: %w", err) } 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) { go s.processPushQueue() go s.retryFailedPushes() } // 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 { // 没有可用设备,创建待推送记录 s.CreatePushRecord(ctx, task.userID, task.message.ID, model.PushChannelJPush) return } // 提取消息标题和内容 _ = task.message.Decrypt() title := "新消息" content := dto.ExtractTextContentFromModel(task.message.Segments) if task.message.IsSystemMessage() || task.message.Category == model.CategoryNotification { if task.message.ExtraData != nil && task.message.ExtraData.ActorName != "" { title = task.message.ExtraData.ActorName } else { title = "系统通知" } } extras := map[string]any{ "message_id": task.message.ID, "conversation_id": task.message.ConversationID, "sender_id": task.message.SenderID, "category": string(task.message.Category), } if task.message.ExtraData != nil { extras["actor_name"] = task.message.ExtraData.ActorName extras["avatar_url"] = task.message.ExtraData.AvatarURL extras["target_id"] = task.message.ExtraData.TargetID extras["target_type"] = task.message.ExtraData.TargetType } // 尝试使用极光推送(统一推送到所有移动设备) mobileDevices := make([]*model.DeviceToken, 0) for _, device := range devices { if device.SupportsMobilePush() && device.PushToken != "" { mobileDevices = append(mobileDevices, device) } } if len(mobileDevices) > 0 { // 优先使用极光推送 if s.jpushClient != nil { err := s.pushViaJPushBatch(mobileDevices, title, content, extras) if err == nil { // 极光推送成功,为每个设备创建推送记录 for _, device := range mobileDevices { record, rerr := s.CreatePushRecord(ctx, task.userID, task.message.ID, model.PushChannelJPush) if rerr != nil { continue } record.DeviceToken = device.PushToken record.DeviceType = string(device.DeviceType) record.MarkPushed() s.pushRepo.Update(record) } 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.jpushClient != nil { pushErr = s.pushViaJPush(ctx, device, title, content, extras) } else { pushErr = errors.New("jpush not configured") } if pushErr != nil { record.MarkFailed(pushErr.Error()) } else { record.MarkPushed() } s.pushRepo.Update(record) } } } // 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 } } // retryFailedPushes 重试失败的推送 func (s *pushServiceImpl) retryFailedPushes() { ticker := time.NewTicker(5 * time.Minute) defer ticker.Stop() for { select { case <-s.stopChan: return case <-ticker.C: s.doRetry() } } } // doRetry 执行重试 func (s *pushServiceImpl) doRetry() { ctx := context.Background() // 获取失败待重试的推送 records, err := s.pushRepo.GetFailedPushesForRetry(100) if err != nil { return } 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) 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.jpushClient != nil && device.SupportsMobilePush() && device.PushToken != "" { _ = 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 } pushErr = s.pushViaJPush(ctx, device, title, content, extras) } if pushErr != nil { record.MarkFailed(pushErr.Error()) } else { record.MarkPushed() } s.pushRepo.Update(record) } } } // PushChatMessage 推送聊天消息给离线用户 // 检查用户在线状态和免打扰设置,仅对离线且未免打扰的用户通过 JPush 推送 func (s *pushServiceImpl) PushChatMessage(ctx context.Context, userID string, conversationID string, sender *ChatMessageSender, convType model.ConversationType, convName string, message *model.Message) 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.jpushClient == nil { return nil } devices, err := s.deviceRepo.GetActiveByUserID(userID) if err != nil || len(devices) == 0 { return nil } var regIDs []string for _, d := range devices { if d.SupportsMobilePush() && d.PushToken != "" { regIDs = append(regIDs, d.PushToken) } } if len(regIDs) == 0 { return nil } // 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 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 sender.Avatar != "" { notification.Android.LargeIcon = sender.Avatar } if _, pushErr := s.jpushClient.PushByRegistrationIDs(regIDs, notification, extras); 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.jpushClient != nil { devices, err := s.deviceRepo.GetActiveByUserID(userID) if err == nil && len(devices) > 0 { mobileDevices := make([]*model.DeviceToken, 0) for _, d := range devices { if d.SupportsMobilePush() && d.PushToken != "" { mobileDevices = append(mobileDevices, d) } } 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, extras, ) 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 }