diff --git a/.gitea/workflows/deploy.yml b/.gitea/workflows/deploy.yml index 95a36a9..9c74001 100644 --- a/.gitea/workflows/deploy.yml +++ b/.gitea/workflows/deploy.yml @@ -110,6 +110,17 @@ jobs: -e APP_GRPC_ENABLED=true \ -e APP_GRPC_PORT=50051 \ -e APP_SERVER_MODE=release \ + -e APP_TENCENT_TMS_ENABLED=false \ + -e "APP_TENCENT_TMS_SECRET_ID=${{ secrets.TENCENT_TMS_SECRET_ID }}" \ + -e "APP_TENCENT_TMS_SECRET_KEY=${{ secrets.TENCENT_TMS_SECRET_KEY }}" \ + -e APP_TENCENT_TMS_REGION=ap-guangzhou \ + -e APP_TENCENT_TMS_TIMEOUT=30 \ + -e APP_JPUSH_ENABLED=true \ + -e "APP_JPUSH_APP_KEY=${{ secrets.JPUSH_APP_KEY }}" \ + -e "APP_JPUSH_MASTER_SECRET=${{ secrets.JPUSH_MASTER_SECRET }}" \ + -e APP_JPUSH_PRODUCTION=true \ + -e APP_SENSITIVE_ENABLED=true \ + -e APP_REPORT_AUTO_HIDE_THRESHOLD=3 \ -e "APP_SETUP_SECRET=${{ secrets.SETUP_SECRET }}" \ ${IMAGE} diff --git a/cmd/server/wire_gen.go b/cmd/server/wire_gen.go index 42423ec..160703b 100644 --- a/cmd/server/wire_gen.go +++ b/cmd/server/wire_gen.go @@ -32,7 +32,9 @@ func InitializeApp() (*App, error) { deviceTokenRepository := repository.NewDeviceTokenRepository(db) messageRepository := repository.NewMessageRepository(db) hub := wire.ProvideWSHub() - pushService := wire.ProvidePushService(pushRecordRepository, deviceTokenRepository, messageRepository, hub) + logger := wire.ProvideLogger() + jpushClient := wire.ProvideJPushClient(config, logger) + pushService := wire.ProvidePushService(pushRecordRepository, deviceTokenRepository, messageRepository, hub, jpushClient) postRepository := repository.NewPostRepository(db) client := wire.ProvideRedisClient(config) cache := wire.ProvideCache(config, client) @@ -42,7 +44,6 @@ func InitializeApp() (*App, error) { operationLogRepository := repository.NewOperationLogRepository(db) loginLogRepository := repository.NewLoginLogRepository(db) dataChangeLogRepository := repository.NewDataChangeLogRepository(db) - logger := wire.ProvideLogger() asyncLogManager := wire.ProvideAsyncLogManager(operationLogRepository, loginLogRepository, dataChangeLogRepository, logger, config) operationLogService := wire.ProvideOperationLogService(asyncLogManager, operationLogRepository) loginLogService := wire.ProvideLoginLogService(asyncLogManager, loginLogRepository) @@ -77,7 +78,7 @@ func InitializeApp() (*App, error) { return nil, err } uploadService := wire.ProvideUploadService(s3Client, userService) - chatService := wire.ProvideChatService(messageRepository, userRepository, hub, cache, uploadService) + chatService := wire.ProvideChatService(messageRepository, userRepository, hub, cache, uploadService, pushService) messageService := wire.ProvideMessageService(messageRepository, cache, uploadService) groupRepository := repository.NewGroupRepository(db) groupJoinRequestRepository := repository.NewGroupJoinRequestRepository(db) diff --git a/configs/config.yaml b/configs/config.yaml index 4fbd1db..9d4f8b3 100644 --- a/configs/config.yaml +++ b/configs/config.yaml @@ -263,6 +263,15 @@ casbin: report: auto_hide_threshold: 3 # 自动隐藏阈值，举报次数达到此数值时自动隐藏内容 +# 极光推送配置 +# 环境变量: +# APP_JPUSH_ENABLED, APP_JPUSH_APP_KEY, APP_JPUSH_MASTER_SECRET, APP_JPUSH_PRODUCTION +jpush: + enabled: false + app_key: "" + master_secret: "" + production: false # true: 生产环境, false: 开发环境 + # 初始化超级管理员密钥 # 环境变量: APP_SETUP_SECRET # 设置后可使用 POST /api/v1/admin/setup-super-admin 接口初始化第一位超级管理员 diff --git a/internal/config/config.go b/internal/config/config.go index 13e2d40..846816d 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1,4 +1,4 @@ -package config +package config import ( "cmp" @@ -33,6 +33,7 @@ type Config struct { WebRTC WebRTCConfig `mapstructure:"webrtc"` Report ReportConfig `mapstructure:"report"` Sensitive SensitiveConfig `mapstructure:"sensitive"` + JPush JPushConfig `mapstructure:"jpush"` SetupSecret string `mapstructure:"setup_secret"` } @@ -155,6 +156,13 @@ func Load(configPath string) (*Config, error) { viper.SetDefault("casbin.enable_cache", true) viper.SetDefault("casbin.cache_ttl", 300) viper.SetDefault("casbin.skip_routes", []string{"/health", "/api/v1/auth/login", "/api/v1/auth/register"}) + // Casbin 默认值 + viper.SetDefault("casbin.model_path", "./configs/casbin/model.conf") + viper.SetDefault("casbin.auto_save", true) + viper.SetDefault("casbin.auto_load", true) + viper.SetDefault("casbin.enable_cache", true) + viper.SetDefault("casbin.cache_ttl", 300) + viper.SetDefault("casbin.skip_routes", []string{"/health", "/api/v1/auth/login", "/api/v1/auth/register"}) // Encryption 默认值 viper.SetDefault("encryption.enabled", false) viper.SetDefault("encryption.key", "") @@ -171,6 +179,10 @@ func Load(configPath string) (*Config, error) { }) // Report 默认值 viper.SetDefault("report.auto_hide_threshold", 3) + viper.SetDefault("jpush.enabled", false) + viper.SetDefault("jpush.app_key", "") + viper.SetDefault("jpush.master_secret", "") + viper.SetDefault("jpush.production", false) viper.SetDefault("setup_secret", "") if err := viper.ReadInConfig(); err != nil { @@ -270,6 +282,10 @@ func Load(configPath string) (*Config, error) { cfg.Sensitive.LoadFromDB, _ = strconv.ParseBool(getEnvOrDefault("APP_SENSITIVE_LOAD_FROM_DB", fmt.Sprintf("%t", cfg.Sensitive.LoadFromDB))) cfg.Sensitive.LoadFromRedis, _ = strconv.ParseBool(getEnvOrDefault("APP_SENSITIVE_LOAD_FROM_REDIS", fmt.Sprintf("%t", cfg.Sensitive.LoadFromRedis))) cfg.Sensitive.RedisKeyPrefix = getEnvOrDefault("APP_SENSITIVE_REDIS_KEY_PREFIX", cfg.Sensitive.RedisKeyPrefix) + cfg.JPush.Enabled, _ = strconv.ParseBool(getEnvOrDefault("APP_JPUSH_ENABLED", fmt.Sprintf("%t", cfg.JPush.Enabled))) + cfg.JPush.AppKey = getEnvOrDefault("APP_JPUSH_APP_KEY", cfg.JPush.AppKey) + cfg.JPush.MasterSecret = getEnvOrDefault("APP_JPUSH_MASTER_SECRET", cfg.JPush.MasterSecret) + cfg.JPush.Production, _ = strconv.ParseBool(getEnvOrDefault("APP_JPUSH_PRODUCTION", fmt.Sprintf("%t", cfg.JPush.Production))) cfg.SetupSecret = getEnvOrDefault("APP_SETUP_SECRET", cfg.SetupSecret) diff --git a/internal/config/jpush.go b/internal/config/jpush.go new file mode 100644 index 0000000..012e418 --- /dev/null +++ b/internal/config/jpush.go @@ -0,0 +1,8 @@ +package config + +type JPushConfig struct { + Enabled bool `mapstructure:"enabled"` + AppKey string `mapstructure:"app_key"` + MasterSecret string `mapstructure:"master_secret"` + Production bool `mapstructure:"production"` +} \ No newline at end of file diff --git a/internal/handler/push_handler.go b/internal/handler/push_handler.go index 395692f..ba3ea7c 100644 --- a/internal/handler/push_handler.go +++ b/internal/handler/push_handler.go @@ -85,12 +85,13 @@ func (h *PushHandler) GetMyDevices(c *gin.Context) { return } - // 这里需要从DeviceTokenRepository获取设备列表 - // 由于PushService接口没有提供获取设备列表的方法，我们暂时返回空列表 - // TODO: 在PushService接口中添加GetUserDevices方法 - _ = userID // 避免未使用变量警告 + devices, err := h.pushService.GetUserDevices(c.Request.Context(), userID) + if err != nil { + response.InternalServerError(c, "failed to get devices") + return + } - response.Success(c, []*dto.DeviceTokenResponse{}) + response.Success(c, dto.DeviceTokensToResponse(devices)) } // GetPushRecords 获取推送记录 @@ -140,7 +141,7 @@ func (h *PushHandler) UpdateDeviceToken(c *gin.Context) { } var req struct { - PushToken string `json:"push_token" binding:"required"` + PushToken string `json:"push_token"` } if err := c.ShouldBindJSON(&req); err != nil { response.BadRequest(c, err.Error()) diff --git a/internal/middleware/casbin.go b/internal/middleware/casbin.go index 2a87844..91e3ab0 100644 --- a/internal/middleware/casbin.go +++ b/internal/middleware/casbin.go @@ -1,4 +1,4 @@ -package middleware +package middleware import ( "slices" diff --git a/internal/model/device_token.go b/internal/model/device_token.go index 0785671..884640c 100644 --- a/internal/model/device_token.go +++ b/internal/model/device_token.go @@ -24,7 +24,7 @@ type DeviceToken struct { UserID string `gorm:"column:user_id;type:varchar(50);index;not null" json:"user_id"` // 用户ID (UUID格式) DeviceID string `gorm:"type:varchar(100);not null" json:"device_id"` // 设备唯一标识 DeviceType DeviceType `gorm:"type:varchar(20);not null" json:"device_type"` // 设备类型 - PushToken string `gorm:"type:varchar(256);not null" json:"push_token"` // 推送Token（FCM/APNs等） + PushToken string `gorm:"type:varchar(256)" json:"push_token,omitempty"` // 推送Token（JPush RegistrationID等，Web端可为空） IsActive bool `gorm:"default:true" json:"is_active"` // 是否活跃 DeviceName string `gorm:"type:varchar(100)" json:"device_name,omitempty"` // 设备名称（可选） diff --git a/internal/model/push_record.go b/internal/model/push_record.go index 3ce3db5..b25f8b0 100644 --- a/internal/model/push_record.go +++ b/internal/model/push_record.go @@ -16,6 +16,7 @@ const ( PushChannelFCM PushChannel = "fcm" // Firebase Cloud Messaging PushChannelAPNs PushChannel = "apns" // Apple Push Notification service PushChannelHuawei PushChannel = "huawei" // 华为推送 + PushChannelJPush PushChannel = "jpush" // 极光推送 ) // PushStatus 推送状态 diff --git a/internal/pkg/jpush/client.go b/internal/pkg/jpush/client.go new file mode 100644 index 0000000..cf84f06 --- /dev/null +++ b/internal/pkg/jpush/client.go @@ -0,0 +1,437 @@ +package jpush + +import ( + "bytes" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strconv" + "time" + + "go.uber.org/zap" +) + +const ( + PushAPIBaseURL = "https://api.jpush.cn/v3" + DeviceAPIBaseURL = "https://device.jpush.cn/v3" + ReportAPIBaseURL = "https://report.jpush.cn/v3" +) + +type Client struct { + appKey string + masterSecret string + production bool + httpClient *http.Client + logger *zap.Logger +} + +func NewClient(appKey, masterSecret string, production bool, logger *zap.Logger) *Client { + return &Client{ + appKey: appKey, + masterSecret: masterSecret, + production: production, + httpClient: &http.Client{Timeout: 30 * time.Second}, + logger: logger, + } +} + +type Notification struct { + Alert string `json:"alert,omitempty"` + Android *AndroidNotif `json:"android,omitempty"` + IOS *IOSNotif `json:"ios,omitempty"` +} + +type AndroidNotif struct { + Alert string `json:"alert"` + Title string `json:"title,omitempty"` + BuilderID int `json:"builder_id,omitempty"` + ChannelID string `json:"channel_id,omitempty"` + Category string `json:"category,omitempty"` + Priority int `json:"priority,omitempty"` + Style int `json:"style,omitempty"` + AlertType int `json:"alert_type,omitempty"` + BigText string `json:"big_text,omitempty"` + BigPicPath string `json:"big_pic_path,omitempty"` + LargeIcon string `json:"large_icon,omitempty"` + Sound string `json:"sound,omitempty"` + Intent *IntentConfig `json:"intent,omitempty"` + BadgeAddNum int `json:"badge_add_num,omitempty"` + BadgeSetNum int `json:"badge_set_num,omitempty"` + Extras map[string]any `json:"extras,omitempty"` +} + +type IntentConfig struct { + URL string `json:"url"` +} + +type IOSNotif struct { + Alert any `json:"alert,omitempty"` + Sound any `json:"sound,omitempty"` + Badge string `json:"badge,omitempty"` + MutableContent bool `json:"mutable-content,omitempty"` + ContentAvailable bool `json:"content-available,omitempty"` + Category string `json:"category,omitempty"` + ThreadID string `json:"thread-id,omitempty"` + Extras map[string]any `json:"extras,omitempty"` +} + +type PushOptions struct { + TimeToLive int `json:"time_to_live,omitempty"` + ApnsProduction bool `json:"apns_production"` + ApnsCollapseID string `json:"apns_collapse_id,omitempty"` +} + +type PushResponse struct { + MsgID string `json:"msg_id"` + SendNo int64 `json:"sendno"` +} + +type RateLimitInfo struct { + Limit int + Remaining int + Reset int +} + +type APIError struct { + ErrCode int `json:"code"` + ErrMessage string `json:"message"` + ErrDetail struct { + Code int `json:"code"` + Message string `json:"message"` + } `json:"error"` +} + +func (e *APIError) Error() string { + if e.ErrDetail.Code != 0 { + return fmt.Sprintf("jpush error %d: %s", e.ErrDetail.Code, e.ErrDetail.Message) + } + return fmt.Sprintf("jpush error %d: %s", e.ErrCode, e.ErrMessage) +} + +func (e *APIError) Code() int { + if e.ErrDetail.Code != 0 { + return e.ErrDetail.Code + } + return e.ErrCode +} + +func (c *Client) generateAuth() string { + auth := base64.StdEncoding.EncodeToString([]byte(c.appKey + ":" + c.masterSecret)) + return "Basic " + auth +} + +func (c *Client) doRequest(method, fullPath string, payload any) (*http.Response, []byte, *RateLimitInfo, error) { + var bodyBytes []byte + var err error + if payload != nil { + bodyBytes, err = json.Marshal(payload) + if err != nil { + return nil, nil, nil, fmt.Errorf("marshal payload: %w", err) + } + } + + req, err := http.NewRequest(method, fullPath, bytes.NewReader(bodyBytes)) + if err != nil { + return nil, nil, nil, fmt.Errorf("create request: %w", err) + } + + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", c.generateAuth()) + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, nil, nil, fmt.Errorf("send request: %w", err) + } + + respBody, err := io.ReadAll(resp.Body) + resp.Body.Close() + if err != nil { + return nil, nil, nil, fmt.Errorf("read response: %w", err) + } + + rateLimit := &RateLimitInfo{} + if v := resp.Header.Get("X-Rate-Limit-Limit"); v != "" { + rateLimit.Limit, _ = strconv.Atoi(v) + } + if v := resp.Header.Get("X-Rate-Limit-Remaining"); v != "" { + rateLimit.Remaining, _ = strconv.Atoi(v) + } + if v := resp.Header.Get("X-Rate-Limit-Reset"); v != "" { + rateLimit.Reset, _ = strconv.Atoi(v) + } + + return resp, respBody, rateLimit, nil +} + +func (c *Client) parseResponse(resp *http.Response, body []byte) (*PushResponse, error) { + if resp.StatusCode == http.StatusOK { + var result PushResponse + if err := json.Unmarshal(body, &result); err != nil { + return nil, fmt.Errorf("unmarshal response: %w", err) + } + return &result, nil + } + + var apiErr APIError + if json.Unmarshal(body, &apiErr) == nil && apiErr.Code() != 0 { + c.logger.Error("jpush api error", + zap.Int("status", resp.StatusCode), + zap.Int("code", apiErr.Code()), + ) + return nil, &apiErr + } + + return nil, fmt.Errorf("jpush HTTP %d: %s", resp.StatusCode, string(body)) +} + +func (c *Client) Push(payload map[string]any) (*PushResponse, *RateLimitInfo, error) { + body, _ := json.Marshal(payload) + c.logger.Debug("jpush push request", zap.Int("body_size", len(body))) + + resp, respBody, rateLimit, err := c.doRequest("POST", PushAPIBaseURL+"/push", payload) + if err != nil { + return nil, rateLimit, err + } + + result, parseErr := c.parseResponse(resp, respBody) + return result, rateLimit, parseErr +} + +func (c *Client) PushByRegistrationIDs(registrationIDs []string, notification *Notification, extras map[string]any) (*PushResponse, error) { + if len(registrationIDs) == 0 { + return nil, fmt.Errorf("registration IDs cannot be empty") + } + if len(registrationIDs) > 1000 { + return nil, fmt.Errorf("registration IDs exceed 1000 limit") + } + + mergeExtras(notification, extras) + + payload := map[string]any{ + "platform": "all", + "audience": map[string]any{"registration_id": registrationIDs}, + "notification": notification, + "options": map[string]any{ + "time_to_live": 86400, + "apns_production": c.production, + }, + } + + result, _, err := c.Push(payload) + return result, err +} + +func (c *Client) PushByAliases(aliases []string, notification *Notification, extras map[string]any) (*PushResponse, error) { + if len(aliases) == 0 { + return nil, fmt.Errorf("aliases cannot be empty") + } + if len(aliases) > 1000 { + return nil, fmt.Errorf("aliases exceed 1000 limit") + } + + mergeExtras(notification, extras) + + payload := map[string]any{ + "platform": "all", + "audience": map[string]any{"alias": aliases}, + "notification": notification, + "options": map[string]any{ + "time_to_live": 86400, + "apns_production": c.production, + }, + } + + result, _, err := c.Push(payload) + return result, err +} + +func (c *Client) PushAll(notification *Notification) (*PushResponse, error) { + payload := map[string]any{ + "platform": "all", + "audience": "all", + "notification": notification, + "options": map[string]any{ + "time_to_live": 86400, + "apns_production": c.production, + }, + } + + result, _, err := c.Push(payload) + return result, err +} + +func mergeExtras(notification *Notification, extras map[string]any) { + if extras == nil { + return + } + if notification.Android != nil { + if notification.Android.Extras == nil { + notification.Android.Extras = make(map[string]any) + } + for k, v := range extras { + notification.Android.Extras[k] = v + } + } + if notification.IOS != nil { + if notification.IOS.Extras == nil { + notification.IOS.Extras = make(map[string]any) + } + for k, v := range extras { + notification.IOS.Extras[k] = v + } + } +} + +func BuildNotification(title, body string, notificationType string, extras map[string]any) *Notification { + androidExtras := map[string]any{ + "notification_type": notificationType, + } + for k, v := range extras { + androidExtras[k] = v + } + + iosExtras := map[string]any{ + "notification_type": notificationType, + } + for k, v := range extras { + iosExtras[k] = v + } + + return &Notification{ + Alert: body, + Android: &AndroidNotif{ + Alert: body, + Title: title, + Extras: androidExtras, + Priority: 1, + }, + IOS: &IOSNotif{ + Alert: body, + Sound: "default", + Badge: "+1", + MutableContent: true, + Extras: iosExtras, + }, + } +} + +type DeviceInfo struct { + Tags []string `json:"tags"` + Alias string `json:"alias"` + Mobile string `json:"mobile"` +} + +func (c *Client) GetDeviceInfo(registrationID string) (*DeviceInfo, error) { + path := DeviceAPIBaseURL + "/v3/devices/" + url.PathEscape(registrationID) + resp, body, _, err := c.doRequest("GET", path, nil) + if err != nil { + return nil, err + } + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("jpush get device info failed: HTTP %d", resp.StatusCode) + } + + var info DeviceInfo + if err := json.Unmarshal(body, &info); err != nil { + return nil, fmt.Errorf("unmarshal device info: %w", err) + } + return &info, nil +} + +func (c *Client) SetDeviceAlias(registrationID string, alias string) error { + path := DeviceAPIBaseURL + "/v3/devices/" + url.PathEscape(registrationID) + payload := map[string]any{ + "alias": alias, + } + + resp, body, _, err := c.doRequest("POST", path, payload) + if err != nil { + return err + } + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("jpush set device alias failed: HTTP %d: %s", resp.StatusCode, string(body)) + } + return nil +} + +func (c *Client) SetDeviceTags(registrationID string, addTags, removeTags []string) error { + path := DeviceAPIBaseURL + "/v3/devices/" + url.PathEscape(registrationID) + tags := map[string]any{} + if len(addTags) > 0 { + tags["add"] = addTags + } + if len(removeTags) > 0 { + tags["remove"] = removeTags + } + + payload := map[string]any{ + "tags": tags, + } + + resp, body, _, err := c.doRequest("POST", path, payload) + if err != nil { + return err + } + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("jpush set device tags failed: HTTP %d: %s", resp.StatusCode, string(body)) + } + return nil +} + +func (c *Client) GetAliasRegistrationIDs(alias string) ([]string, error) { + path := DeviceAPIBaseURL + "/v3/aliases/" + url.PathEscape(alias) + resp, body, _, err := c.doRequest("GET", path, nil) + if err != nil { + return nil, err + } + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("jpush get alias failed: HTTP %d", resp.StatusCode) + } + + var result struct { + RegistrationIDs []string `json:"registration_ids"` + } + if err := json.Unmarshal(body, &result); err != nil { + return nil, fmt.Errorf("unmarshal alias response: %w", err) + } + return result.RegistrationIDs, nil +} + +func (c *Client) DeleteAlias(alias string) error { + path := DeviceAPIBaseURL + "/v3/aliases/" + url.PathEscape(alias) + resp, body, _, err := c.doRequest("DELETE", path, nil) + if err != nil { + return err + } + + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent { + return fmt.Errorf("jpush delete alias failed: HTTP %d: %s", resp.StatusCode, string(body)) + } + return nil +} + +func (c *Client) RemoveAliasDevices(alias string, registrationIDs []string) error { + path := DeviceAPIBaseURL + "/v3/aliases/" + url.PathEscape(alias) + payload := map[string]any{ + "registration_ids": map[string]any{ + "remove": registrationIDs, + }, + } + + resp, body, _, err := c.doRequest("POST", path, payload) + if err != nil { + return err + } + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("jpush remove alias devices failed: HTTP %d: %s", resp.StatusCode, string(body)) + } + return nil +} diff --git a/internal/repository/device_token_repo.go b/internal/repository/device_token_repo.go index 10dfd73..0391121 100644 --- a/internal/repository/device_token_repo.go +++ b/internal/repository/device_token_repo.go @@ -124,9 +124,11 @@ func (r *deviceTokenRepository) Upsert(token *model.DeviceToken) error { return err } - // 更新现有记录 + // 更新现有记录（包括 user_id，设备可能在用户间转移） return r.db.Model(&existing).Updates(map[string]any{ + "user_id": token.UserID, "push_token": token.PushToken, + "device_type": token.DeviceType, "is_active": true, "device_name": token.DeviceName, "last_used_at": time.Now(), diff --git a/internal/service/casbin_service.go b/internal/service/casbin_service.go index 370f8d2..3a3f3f4 100644 --- a/internal/service/casbin_service.go +++ b/internal/service/casbin_service.go @@ -1,4 +1,4 @@ -package service +package service import ( "context" diff --git a/internal/service/chat_service.go b/internal/service/chat_service.go index 5afe8ca..22084dd 100644 --- a/internal/service/chat_service.go +++ b/internal/service/chat_service.go @@ -61,6 +61,7 @@ type chatServiceImpl struct { userRepo repository.UserRepository sensitive SensitiveService wsHub *ws.Hub + pushSvc PushService // 缓存相关字段 conversationCache *cache.ConversationCache @@ -75,6 +76,7 @@ func NewChatService( wsHub *ws.Hub, cacheBackend cache.Cache, uploadService *UploadService, + pushSvc PushService, ) ChatService { // 创建适配器 convRepoAdapter := cache.NewConversationRepositoryAdapter(repo) @@ -93,6 +95,7 @@ func NewChatService( userRepo: userRepo, sensitive: sensitive, wsHub: wsHub, + pushSvc: pushSvc, conversationCache: conversationCache, uploadService: uploadService, } @@ -384,6 +387,34 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv } } + // 对离线用户通过 JPush 推送聊天消息通知 + if s.pushSvc != nil && len(participants) > 0 { + sender := ChatMessageSender{ID: senderID} + if senderUser, sErr := s.userRepo.GetByID(senderID); sErr == nil { + sender.Name = senderUser.Nickname + sender.Avatar = senderUser.Avatar + } + convType := conv.Type + convName := "" + if conv.Type == model.ConversationTypeGroup && conv.Group != nil { + convName = conv.Group.Name + } + for _, p := range participants { + if p.UserID == senderID { + continue + } + go func(userID string, sender ChatMessageSender) { + if pushErr := s.pushSvc.PushChatMessage(context.Background(), userID, conversationID, &sender, convType, convName, message); pushErr != nil { + zap.L().Debug("push chat message skipped or failed", + zap.String("userID", userID), + zap.String("conversationID", conversationID), + zap.Error(pushErr), + ) + } + }(p.UserID, sender) + } + } + return message, nil } diff --git a/internal/service/push_service.go b/internal/service/push_service.go index bbc0d0e..c0d0434 100644 --- a/internal/service/push_service.go +++ b/internal/service/push_service.go @@ -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 } diff --git a/internal/service/setup_service.go b/internal/service/setup_service.go index b12d74f..ae9df63 100644 --- a/internal/service/setup_service.go +++ b/internal/service/setup_service.go @@ -60,4 +60,4 @@ func (s *setupService) SetupSuperAdmin(ctx context.Context, userID, setupSecret } return s.casbinSvc.AddRoleForUser(ctx, user.ID, model.RoleSuperAdmin) -} +} \ No newline at end of file diff --git a/internal/wire/casbin.go b/internal/wire/casbin.go index 5f6a2db..6eb9039 100644 --- a/internal/wire/casbin.go +++ b/internal/wire/casbin.go @@ -1,4 +1,4 @@ -package wire +package wire import ( "with_you/internal/cache" diff --git a/internal/wire/infrastructure.go b/internal/wire/infrastructure.go index b6e8975..85afe30 100644 --- a/internal/wire/infrastructure.go +++ b/internal/wire/infrastructure.go @@ -9,6 +9,7 @@ import ( "with_you/internal/pkg/crypto" "with_you/internal/pkg/email" "with_you/internal/pkg/hook" + "with_you/internal/pkg/jpush" "with_you/internal/pkg/openai" "with_you/internal/pkg/redis" "with_you/internal/pkg/s3" @@ -53,6 +54,7 @@ var InfrastructureSet = wire.NewSet( ProvideTencentClient, ProvideEmailClient, ProvideS3Client, + ProvideJPushClient, // 消息加密器 ProvideMessageEncryptor, @@ -193,6 +195,23 @@ func ProvideS3Client(cfg *config.Config) (*s3.Client, error) { return s3.New(&cfg.S3) } +// ProvideJPushClient 提供极光推送客户端 +func ProvideJPushClient(cfg *config.Config, logger *zap.Logger) *jpush.Client { + if !cfg.JPush.Enabled { + logger.Info("JPush is disabled") + return nil + } + if cfg.JPush.AppKey == "" || cfg.JPush.MasterSecret == "" { + logger.Warn("JPush is enabled but app_key or master_secret is not configured") + return nil + } + client := jpush.NewClient(cfg.JPush.AppKey, cfg.JPush.MasterSecret, cfg.JPush.Production, logger) + logger.Info("JPush client initialized", + zap.Bool("production", cfg.JPush.Production), + ) + return client +} + // ProvideTransactionManager 提供事务管理器 func ProvideTransactionManager(db *gorm.DB) repository.TransactionManager { return repository.NewTransactionManager(db) diff --git a/internal/wire/service.go b/internal/wire/service.go index 8799494..7462551 100644 --- a/internal/wire/service.go +++ b/internal/wire/service.go @@ -8,6 +8,7 @@ import ( "with_you/internal/grpc/runner" "with_you/internal/pkg/email" "with_you/internal/pkg/hook" + "with_you/internal/pkg/jpush" "with_you/internal/pkg/openai" "with_you/internal/pkg/redis" "with_you/internal/pkg/s3" @@ -105,8 +106,9 @@ func ProvidePushService( deviceTokenRepo repository.DeviceTokenRepository, messageRepo repository.MessageRepository, wsHub *ws.Hub, + jpushClient *jpush.Client, ) service.PushService { - return service.NewPushService(pushRepo, deviceTokenRepo, messageRepo, wsHub) + return service.NewPushService(pushRepo, deviceTokenRepo, messageRepo, wsHub, jpushClient) } // ProvideSystemMessageService 提供系统消息服务 @@ -206,8 +208,9 @@ func ProvideChatService( wsHub *ws.Hub, cacheBackend cache.Cache, uploadService *service.UploadService, + pushService service.PushService, ) service.ChatService { - return service.NewChatService(messageRepo, userRepo, nil, wsHub, cacheBackend, uploadService) + return service.NewChatService(messageRepo, userRepo, nil, wsHub, cacheBackend, uploadService, pushService) } // ProvideScheduleService 提供日程服务