@@ -8,8 +8,11 @@ import (
"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"
)
// 推送相关常量
@@ -24,6 +27,13 @@ const (
PushQueueSize = 1000
)
// ChatMessageSender 聊天消息发送者信息
type ChatMessageSender struct {
ID string
Name string
Avatar string
}
// PushPriority 推送优先级
type PushPriority int
@@ -40,6 +50,10 @@ 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
@@ -54,6 +68,7 @@ type PushService interface {
// 推送记录管理
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 )
@@ -66,8 +81,8 @@ type pushServiceImpl struct {
deviceRepo repository . DeviceTokenRepository
messageRepo repository . MessageRepository
wsHub * ws . Hub
jpushClient * jpush . Client
// 推送队列
pushQueue chan * pushTask
stopChan chan struct { }
}
@@ -85,12 +100,14 @@ func NewPushService(
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 { } ) ,
}
@@ -117,7 +134,56 @@ func (s *pushServiceImpl) PushToUser(ctx context.Context, userID string, message
return nil
}
// WebSocket推送失败, 加入推送队列等待移动端推 送
// 用户不在线,尝试通过极光推送直接发 送
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 ,
@@ -127,7 +193,7 @@ func (s *pushServiceImpl) PushToUser(ctx context.Context, userID string, message
return nil
default :
// 队列已满,直接创建待推送记录
_ , err := s . CreatePushRecord ( ctx , userID , message . ID , model . PushChannelFCM )
_ , err := s . CreatePushRecord ( ctx , userID , message . ID , model . PushChannelJPush )
if err != nil {
return fmt . Errorf ( "failed to create pending push record: %w" , err )
}
@@ -209,22 +275,49 @@ func (s *pushServiceImpl) pushViaWebSocket(ctx context.Context, userID string, m
return true
}
// pushViaFCM 通过FCM推送( 预留接口)
func ( s * pushServiceImpl ) pushViaFCM ( ctx context . Context , deviceToken * model . DeviceToken , message * model . Message ) error {
// TODO: 实现FCM推送
// 1. 构建FCM消息
// 2. 调用Firebase Admin SDK发送消息
// 3. 处理发送结果
return errors . New ( "FCM push not implemented" )
// 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
}
// pushViaAPNs 通过APNs推送( 预留接口)
func ( s * pushServiceImpl ) pushViaAPNs ( ctx context . Context , deviceToken * model . DeviceToken , message * model . Message ) error {
// TODO: 实现APNs推送
// 1. 构建APNs消息
// 2. 调用APNs SDK发送消息
// 3. 处理发送结果
return errors . New ( "APNs push not implemented" )
// 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 注册设备
@@ -283,6 +376,11 @@ func (s *pushServiceImpl) GetPendingPushes(ctx context.Context, userID string) (
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 ( )
@@ -315,34 +413,88 @@ func (s *pushServiceImpl) processPushTask(task *pushTask) {
devices , err := s . deviceRepo . GetActiveByUserID ( task . userID )
if err != nil || len ( devices ) == 0 {
// 没有可用设备,创建待推送记录
s . CreatePushRecord ( ctx , task . userID , task . message . ID , model . PushChannelFCM )
s . CreatePushRecord ( ctx , task . userID , task . message . ID , model . PushChannelJPush )
return
}
// 对每个设备创建推送记录并尝试推送
for _ , device := range devices {
record , err := s . CreatePushRecord ( ctx , task . userID , task . message . ID , s . getChannelForDevice ( device ) )
if err ! = nil {
continue
}
var pushErr error
switch {
case device . IsIOS ( ) :
pushErr = s . pushViaAPNs ( ctx , device , task . message )
case device . IsAndroid ( ) :
pushErr = s . pushViaFCM ( ctx , device , task . message )
default :
// Web设备只支持WebSocket
continue
}
if pushErr != nil {
record . MarkFailed ( pushErr . Error ( ) )
// 提取消息标题和内容
_ = 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 {
record . MarkPushed ( )
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 )
}
s . pushRepo . Update ( record )
}
}
@@ -350,9 +502,9 @@ func (s *pushServiceImpl) processPushTask(task *pushTask) {
func ( s * pushServiceImpl ) getChannelForDevice ( device * model . DeviceToken ) model . PushChannel {
switch device . DeviceType {
case model . DeviceTypeIOS :
return model . PushChannelAPNs
return model . PushChannelJPush
case model . DeviceTypeAndroid :
return model . PushChannelFCM
return model . PushChannelJPush
default :
return model . PushChannelWebSocket
}
@@ -416,11 +568,30 @@ func (s *pushServiceImpl) doRetry() {
}
var pushErr error
switch {
case device . IsIOS ( ) :
pushErr = s . pushViaAPNs ( ctx , device , message )
case device . IsAndroid ( ) :
pushErr = s . pushViaFCM ( ctx , device , message )
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 {
@@ -433,6 +604,99 @@ func (s *pushServiceImpl) doRetry() {
}
}
// 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推送
@@ -472,7 +736,57 @@ func (s *pushServiceImpl) PushSystemNotification(ctx context.Context, userID str
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
}