From a0e210feab384b33dcd844c24e9911fc1524d1d1 Mon Sep 17 00:00:00 2001 From: lafay <2021211506@stu.hit.edu.cn> Date: Wed, 17 Jun 2026 20:41:55 +0800 Subject: [PATCH] feat(server): implement chat file TTL cleanup and enhance JPush vendor channel support Add FileCleanupWorker for automatic expiration of chat files with configurable retention period and batch processing. Files uploaded via `/api/v1/uploads/files` are tracked in `uploaded_files` table with expiration timestamps, then deleted from S3 and database upon expiry. Expired file URLs are injected as `"expired": true` in message responses. Extend JPush push notification configuration with vendor-specific channel_id support (xiaomi, huawei, oppo, vivo, meizu, honor, fcm) for differentiating system vs chat notifications. Add iOS APNs thread-id grouping configuration for notification categorization. --- cmd/server/app.go | 55 ++-- cmd/server/wire_gen.go | 10 +- configs/config.yaml | 43 +++ internal/config/config.go | 59 ++++ internal/config/jpush.go | 65 ++++- internal/config/server.go | 8 + internal/database/database.go | 3 + internal/dto/message_converter.go | 29 ++ internal/handler/message_handler.go | 47 +++- internal/handler/upload_handler.go | 36 +++ internal/model/uploaded_file.go | 26 ++ internal/pkg/jpush/client.go | 78 +++--- internal/pkg/jpush/vendor/converters.go | 156 +++++++++++ internal/pkg/jpush/vendor/params.go | 151 ++++++++++ .../pkg/jpush/vendor/payload_compat_test.go | 135 +++++++++ internal/pkg/jpush/vendor/text_renderer.go | 61 ++++ internal/repository/uploaded_file_repo.go | 91 ++++++ internal/router/router.go | 1 + internal/service/file_cleanup_worker.go | 261 ++++++++++++++++++ internal/service/push_service.go | 130 +++++++-- internal/service/upload_service.go | 228 ++++++++++++++- internal/wire/handler.go | 3 +- internal/wire/repository.go | 3 + internal/wire/service.go | 21 +- 24 files changed, 1573 insertions(+), 127 deletions(-) create mode 100644 internal/model/uploaded_file.go create mode 100644 internal/pkg/jpush/vendor/converters.go create mode 100644 internal/pkg/jpush/vendor/params.go create mode 100644 internal/pkg/jpush/vendor/payload_compat_test.go create mode 100644 internal/pkg/jpush/vendor/text_renderer.go create mode 100644 internal/repository/uploaded_file_repo.go create mode 100644 internal/service/file_cleanup_worker.go diff --git a/cmd/server/app.go b/cmd/server/app.go index ce0825a..499229f 100644 --- a/cmd/server/app.go +++ b/cmd/server/app.go @@ -18,17 +18,18 @@ import ( // App 应用程序结构体 type App struct { - Config *config.Config - DB *gorm.DB - Router *router.Router - PushService service.PushService - HotRankWorker *service.HotRankWorker - GRPCServer *runner.Server - Server *http.Server - Publisher ws.MessagePublisher - TaskDispatcher runner.TaskDispatcher - Logger *zap.Logger - PushWorker *service.PushWorker + Config *config.Config + DB *gorm.DB + Router *router.Router + PushService service.PushService + HotRankWorker *service.HotRankWorker + FileCleanupWorker *service.FileCleanupWorker + GRPCServer *runner.Server + Server *http.Server + Publisher ws.MessagePublisher + TaskDispatcher runner.TaskDispatcher + Logger *zap.Logger + PushWorker *service.PushWorker } // NewApp 创建应用程序 @@ -38,6 +39,7 @@ func NewApp( r *router.Router, pushService service.PushService, hotRankWorker *service.HotRankWorker, + fileCleanupWorker *service.FileCleanupWorker, grpcServer *runner.Server, publisher ws.MessagePublisher, taskDispatcher runner.TaskDispatcher, @@ -45,16 +47,17 @@ func NewApp( pushWorker *service.PushWorker, ) *App { return &App{ - Config: cfg, - DB: db, - Router: r, - PushService: pushService, - HotRankWorker: hotRankWorker, - GRPCServer: grpcServer, - Publisher: publisher, - TaskDispatcher: taskDispatcher, - Logger: logger, - PushWorker: pushWorker, + Config: cfg, + DB: db, + Router: r, + PushService: pushService, + HotRankWorker: hotRankWorker, + FileCleanupWorker: fileCleanupWorker, + GRPCServer: grpcServer, + Publisher: publisher, + TaskDispatcher: taskDispatcher, + Logger: logger, + PushWorker: pushWorker, } } @@ -76,6 +79,11 @@ func (a *App) Start(ctx context.Context) error { a.HotRankWorker.Start(ctx) } + // 2.1.1 聊天文件过期清理 + if a.FileCleanupWorker != nil { + a.FileCleanupWorker.Start(ctx) + } + // 2.2 启动 PushWorker（Redis Stream 异步推送） if a.PushWorker != nil { a.PushWorker.Start(ctx) @@ -156,6 +164,11 @@ func (a *App) Stop(ctx context.Context) error { a.HotRankWorker.Stop() } + if a.FileCleanupWorker != nil { + a.Logger.Info("Stopping file cleanup worker") + a.FileCleanupWorker.Stop() + } + // 5.1 停止 PushWorker if a.PushWorker != nil { a.Logger.Info("Stopping push worker (stream)") diff --git a/cmd/server/wire_gen.go b/cmd/server/wire_gen.go index eb81a72..d326c26 100644 --- a/cmd/server/wire_gen.go +++ b/cmd/server/wire_gen.go @@ -35,7 +35,7 @@ func InitializeApp() (*App, error) { messagePublisher := wire.ProvideWSMessagePublisher(config, client) logger := wire.ProvideLogger() jpushClient := wire.ProvideJPushClient(config, logger) - pushService := wire.ProvidePushService(pushRecordRepository, deviceTokenRepository, messageRepository, messagePublisher, jpushClient) + pushService := wire.ProvidePushService(pushRecordRepository, deviceTokenRepository, messageRepository, messagePublisher, jpushClient, config) postRepository := repository.NewPostRepository(db) cache := wire.ProvideCache(config, client) systemMessageService := wire.ProvideSystemMessageService(systemNotificationRepository, pushService, userRepository, postRepository, cache) @@ -77,7 +77,8 @@ func InitializeApp() (*App, error) { if err != nil { return nil, err } - uploadService := wire.ProvideUploadService(s3Client, userService) + uploadedFileRepository := repository.NewUploadedFileRepository(db) + uploadService := wire.ProvideUploadService(s3Client, userService, uploadedFileRepository) seqBufferManager := wire.ProvideSeqBufferManager(config, client, cache) pushWorker := wire.ProvidePushWorker(config, client, messagePublisher, pushService, messageRepository, userRepository) conversationVersionLogRepository := repository.NewConversationVersionLogRepository(db) @@ -86,7 +87,7 @@ func InitializeApp() (*App, error) { groupRepository := repository.NewGroupRepository(db) groupJoinRequestRepository := repository.NewGroupJoinRequestRepository(db) groupService := wire.ProvideGroupService(groupRepository, userRepository, messageRepository, groupJoinRequestRepository, systemNotificationRepository, messagePublisher, cache) - messageHandler := wire.ProvideMessageHandler(chatService, messageService, userService, groupService, messagePublisher) + messageHandler := wire.ProvideMessageHandler(chatService, messageService, userService, groupService, uploadService, messagePublisher) notificationRepository := repository.NewNotificationRepository(db) notificationService := wire.ProvideNotificationService(notificationRepository, cache) notificationHandler := handler.NewNotificationHandler(notificationService) @@ -163,10 +164,11 @@ func InitializeApp() (*App, error) { liveKitHandler := wire.ProvideLiveKitHandler(liveKitService, callService, config, logger) router := ProvideRouter(userRepository, userHandler, postHandler, commentHandler, messageHandler, notificationHandler, uploadHandler, jwtService, pushHandler, systemMessageHandler, groupHandler, stickerHandler, voteHandler, channelHandler, scheduleHandler, gradeHandler, examHandler, emptyClassroomHandler, roleHandler, adminUserHandler, adminPostHandler, adminCommentHandler, adminGroupHandler, adminDashboardHandler, adminLogHandler, qrCodeHandler, materialHandler, callHandler, reportHandler, adminReportHandler, verificationHandler, adminVerificationHandler, adminProfileAuditHandler, setupHandler, tradeHandler, logService, userActivityService, casbinService, userService, wsHandler, liveKitHandler) hotRankWorker := wire.ProvideHotRankWorker(config, postRepository, channelRepository, cache, client) + fileCleanupWorker := wire.ProvideFileCleanupWorker(config, uploadedFileRepository, s3Client, client) server := wire.ProvideGRPCServer(config, logger, runnerHub) runnerRegistry := wire.ProvideRunnerRegistry(config, client) taskDispatcher := wire.ProvideTaskDispatcher(runnerHub, runnerRegistry, config, client, logger) - app := NewApp(config, db, router, pushService, hotRankWorker, server, messagePublisher, taskDispatcher, logger, pushWorker) + app := NewApp(config, db, router, pushService, hotRankWorker, fileCleanupWorker, server, messagePublisher, taskDispatcher, logger, pushWorker) return app, nil } diff --git a/configs/config.yaml b/configs/config.yaml index 2348a94..fb28bab 100644 --- a/configs/config.yaml +++ b/configs/config.yaml @@ -288,11 +288,54 @@ report: # 极光推送配置 # 环境变量: # APP_JPUSH_ENABLED, APP_JPUSH_APP_KEY, APP_JPUSH_MASTER_SECRET, APP_JPUSH_PRODUCTION +# 厂商通道 channel_id（区分系统消息与私聊消息，通过 options.third_party_channel 下发）: +# APP_JPUSH_CHANNEL_SYSTEM, APP_JPUSH_CHANNEL_CHAT +# APP_JPUSH_CHANNEL_XIAOMI_SYSTEM, APP_JPUSH_CHANNEL_XIAOMI_CHAT +# APP_JPUSH_CHANNEL_HUAWEI_SYSTEM, APP_JPUSH_CHANNEL_HUAWEI_CHAT +# APP_JPUSH_CHANNEL_OPPO_SYSTEM, APP_JPUSH_CHANNEL_OPPO_CHAT +# APP_JPUSH_CHANNEL_VIVO_SYSTEM, APP_JPUSH_CHANNEL_VIVO_CHAT +# APP_JPUSH_CHANNEL_MEIZU_SYSTEM, APP_JPUSH_CHANNEL_MEIZU_CHAT +# APP_JPUSH_CHANNEL_HONOR_SYSTEM, APP_JPUSH_CHANNEL_HONOR_CHAT +# APP_JPUSH_CHANNEL_FCM_SYSTEM, APP_JPUSH_CHANNEL_FCM_CHAT +# 小米消息模板（可选，配置后私信消息携带 channel_id 与 mi_template_id）: +# APP_JPUSH_CHANNEL_XIAOMI_MI_SYSTEM_TEMPLATE_ID, APP_JPUSH_CHANNEL_XIAOMI_MI_CHAT_TEMPLATE_ID +# OPPO 私信模板（可选，仅 OPPO，配置后下发私信时携带）: +# APP_JPUSH_CHANNEL_OPPO_SYSTEM_PRIVATE_TEMPLATE_ID, APP_JPUSH_CHANNEL_OPPO_CHAT_PRIVATE_TEMPLATE_ID +# iOS 通知分组 thread-id: +# APP_JPUSH_CHANNEL_IOS_CHAT_THREAD_ID, APP_JPUSH_CHANNEL_IOS_SYSTEM_THREAD_ID jpush: enabled: false app_key: "" master_secret: "" production: false # true: 生产环境, false: 开发环境 + # 厂商通道 channel_id 配置（全部可选，留空则不下发对应厂商字段） + # channel.system / channel.chat 为各厂商默认 channel_id + # channel.vendor.{vendor}.{system|chat} 可单独覆盖某厂商，留空则回退到默认值 + # 小米/OPPO 额外支持消息模板 id（按场景区分），配置后私信消息下发时携带 + # 其余厂商（华为/VIVO/魅族/荣耀/FCM）仅 channel_id + channel: + system: "" # 系统消息/通知默认 channel_id + chat: "" # 私聊/群聊消息默认 channel_id + vendor: + xiaomi: + system: "" + chat: "" + mi_system_template_id: "" # 小米系统消息模板 id + mi_chat_template_id: "" # 小米私聊消息模板 id + huawei: { system: "", chat: "" } + oppo: + system: "" + chat: "" + oppo_system_private_template_id: "" # OPPO 系统消息私信模板 id + oppo_chat_private_template_id: "" # OPPO 私聊消息私信模板 id + vivo: { system: "", chat: "" } + meizu: { system: "", chat: "" } + honor: { system: "", chat: "" } + fcm: { system: "", chat: "" } + # iOS APNs 通知分组（按 thread-id 区分系统/私聊通知，不配置则不分组） + ios: + chat_thread_id: "" # 私聊/群聊通知分组 thread-id + system_thread_id: "" # 系统消息/通知分组 thread-id # 初始化超级管理员密钥 # 环境变量: APP_SETUP_SECRET diff --git a/internal/config/config.go b/internal/config/config.go index e245bfa..3d2e8ac 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -19,6 +19,7 @@ type Config struct { Log LogConfig `mapstructure:"log"` RateLimit RateLimitConfig `mapstructure:"rate_limit"` Upload UploadConfig `mapstructure:"upload"` + FileCleanup FileCleanupConfig `mapstructure:"file_cleanup"` OpenAI OpenAIConfig `mapstructure:"openai"` TencentTMS TencentTMSConfig `mapstructure:"tencent_tms"` Email EmailConfig `mapstructure:"email"` @@ -95,6 +96,11 @@ func Load(configPath string) (*Config, error) { viper.SetDefault("rate_limit.requests_per_minute", 60) viper.SetDefault("upload.max_file_size", 10485760) viper.SetDefault("upload.allowed_types", []string{"image/jpeg", "image/png", "image/gif", "image/webp"}) + // 文件过期清理默认值 + viper.SetDefault("file_cleanup.enabled", true) + viper.SetDefault("file_cleanup.retention_days", 7) + viper.SetDefault("file_cleanup.interval_minutes", 360) + viper.SetDefault("file_cleanup.batch_size", 100) viper.SetDefault("s3.endpoint", "") viper.SetDefault("s3.access_key", "") viper.SetDefault("s3.secret_key", "") @@ -174,6 +180,59 @@ func Load(configPath string) (*Config, error) { viper.SetDefault("jpush.app_key", "") viper.SetDefault("jpush.master_secret", "") viper.SetDefault("jpush.production", false) + // JPush 厂商通道 channel_id 默认值 + // 各厂商 channel_id 通过环境变量配置（见下方 BindEnv） + viper.SetDefault("jpush.channel.system", "") + viper.SetDefault("jpush.channel.chat", "") + viper.SetDefault("jpush.channel.vendor.xiaomi.system", "") + viper.SetDefault("jpush.channel.vendor.xiaomi.chat", "") + viper.SetDefault("jpush.channel.vendor.huawei.system", "") + viper.SetDefault("jpush.channel.vendor.huawei.chat", "") + viper.SetDefault("jpush.channel.vendor.oppo.system", "") + viper.SetDefault("jpush.channel.vendor.oppo.chat", "") + viper.SetDefault("jpush.channel.vendor.vivo.system", "") + viper.SetDefault("jpush.channel.vendor.vivo.chat", "") + viper.SetDefault("jpush.channel.vendor.meizu.system", "") + viper.SetDefault("jpush.channel.vendor.meizu.chat", "") + viper.SetDefault("jpush.channel.vendor.honor.system", "") + viper.SetDefault("jpush.channel.vendor.honor.chat", "") + viper.SetDefault("jpush.channel.vendor.fcm.system", "") + viper.SetDefault("jpush.channel.vendor.fcm.chat", "") + // 小米/OPPO 模板 id 默认值 + viper.SetDefault("jpush.channel.vendor.xiaomi.mi_system_template_id", "") + viper.SetDefault("jpush.channel.vendor.xiaomi.mi_chat_template_id", "") + viper.SetDefault("jpush.channel.vendor.oppo.oppo_system_private_template_id", "") + viper.SetDefault("jpush.channel.vendor.oppo.oppo_chat_private_template_id", "") + // iOS 通知分组 thread-id 默认值 + viper.SetDefault("jpush.channel.ios.chat_thread_id", "") + viper.SetDefault("jpush.channel.ios.system_thread_id", "") + // JPush 厂商 channel_id 环境变量显式绑定 + // AutomaticEnv 对深层嵌套 key 不可靠，故逐个 BindEnv + // 命名规则：APP_JPUSH_CHANNEL_{VENDOR}_{SYSTEM|CHAT} + viper.BindEnv("jpush.channel.system", "APP_JPUSH_CHANNEL_SYSTEM") + viper.BindEnv("jpush.channel.chat", "APP_JPUSH_CHANNEL_CHAT") + viper.BindEnv("jpush.channel.vendor.xiaomi.system", "APP_JPUSH_CHANNEL_XIAOMI_SYSTEM") + viper.BindEnv("jpush.channel.vendor.xiaomi.chat", "APP_JPUSH_CHANNEL_XIAOMI_CHAT") + viper.BindEnv("jpush.channel.vendor.huawei.system", "APP_JPUSH_CHANNEL_HUAWEI_SYSTEM") + viper.BindEnv("jpush.channel.vendor.huawei.chat", "APP_JPUSH_CHANNEL_HUAWEI_CHAT") + viper.BindEnv("jpush.channel.vendor.oppo.system", "APP_JPUSH_CHANNEL_OPPO_SYSTEM") + viper.BindEnv("jpush.channel.vendor.oppo.chat", "APP_JPUSH_CHANNEL_OPPO_CHAT") + viper.BindEnv("jpush.channel.vendor.vivo.system", "APP_JPUSH_CHANNEL_VIVO_SYSTEM") + viper.BindEnv("jpush.channel.vendor.vivo.chat", "APP_JPUSH_CHANNEL_VIVO_CHAT") + viper.BindEnv("jpush.channel.vendor.meizu.system", "APP_JPUSH_CHANNEL_MEIZU_SYSTEM") + viper.BindEnv("jpush.channel.vendor.meizu.chat", "APP_JPUSH_CHANNEL_MEIZU_CHAT") + viper.BindEnv("jpush.channel.vendor.honor.system", "APP_JPUSH_CHANNEL_HONOR_SYSTEM") + viper.BindEnv("jpush.channel.vendor.honor.chat", "APP_JPUSH_CHANNEL_HONOR_CHAT") + viper.BindEnv("jpush.channel.vendor.fcm.system", "APP_JPUSH_CHANNEL_FCM_SYSTEM") + viper.BindEnv("jpush.channel.vendor.fcm.chat", "APP_JPUSH_CHANNEL_FCM_CHAT") + // 小米/OPPO 模板 id 环境变量绑定 + viper.BindEnv("jpush.channel.vendor.xiaomi.mi_system_template_id", "APP_JPUSH_CHANNEL_XIAOMI_MI_SYSTEM_TEMPLATE_ID") + viper.BindEnv("jpush.channel.vendor.xiaomi.mi_chat_template_id", "APP_JPUSH_CHANNEL_XIAOMI_MI_CHAT_TEMPLATE_ID") + viper.BindEnv("jpush.channel.vendor.oppo.oppo_system_private_template_id", "APP_JPUSH_CHANNEL_OPPO_SYSTEM_PRIVATE_TEMPLATE_ID") + viper.BindEnv("jpush.channel.vendor.oppo.oppo_chat_private_template_id", "APP_JPUSH_CHANNEL_OPPO_CHAT_PRIVATE_TEMPLATE_ID") + // iOS thread-id 环境变量绑定 + viper.BindEnv("jpush.channel.ios.chat_thread_id", "APP_JPUSH_CHANNEL_IOS_CHAT_THREAD_ID") + viper.BindEnv("jpush.channel.ios.system_thread_id", "APP_JPUSH_CHANNEL_IOS_SYSTEM_THREAD_ID") viper.SetDefault("setup_secret", "") // WebSocket 默认值 viper.SetDefault("websocket.mode", "standalone") diff --git a/internal/config/jpush.go b/internal/config/jpush.go index 012e418..d604881 100644 --- a/internal/config/jpush.go +++ b/internal/config/jpush.go @@ -1,8 +1,63 @@ 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 + Enabled bool `mapstructure:"enabled"` + AppKey string `mapstructure:"app_key"` + MasterSecret string `mapstructure:"master_secret"` + Production bool `mapstructure:"production"` + // Channel 厂商通道 channel_id 配置 + // 通过 options.third_party_channel 下发到各厂商（小米/华为/OPPO 等） + Channel ChannelConfig `mapstructure:"channel"` +} + +// ChannelConfig 厂商通道 channel_id 配置 +// 区分「系统消息」与「私聊/群聊消息」两类 channel +type ChannelConfig struct { + // System 系统消息/通知默认 channel_id，未单独配置 Vendor 时各厂商共用 + System string `mapstructure:"system"` + // Chat 私聊/群聊消息默认 channel_id，未单独配置 Vendor 时各厂商共用 + Chat string `mapstructure:"chat"` + // Vendor 各厂商 channel_id 覆盖配置（留空则回退到 System/Chat） + Vendor VendorChannel `mapstructure:"vendor"` + // iOS APNs 通知场景区分配置（按 thread-id 分组） + IOS IOSChannelConfig `mapstructure:"ios"` +} + +// IOSChannelConfig iOS APNs 通知场景区分配置 +// iOS 走 APNs 无 channel_id 概念，通过 thread-id 实现通知分组 +type IOSChannelConfig struct { + // ChatThreadID 私聊/群聊消息通知分组 thread-id + ChatThreadID string `mapstructure:"chat_thread_id"` + // SystemThreadID 系统消息/通知分组 thread-id + SystemThreadID string `mapstructure:"system_thread_id"` +} + +// VendorChannel 各厂商通道 channel_id 覆盖 +// 每个厂商区分 System（系统消息）与 Chat（私聊消息）两个 channel_id +type VendorChannel struct { + Xiaomi VendorChannelID `mapstructure:"xiaomi"` + Huawei VendorChannelID `mapstructure:"huawei"` + OPPO VendorChannelID `mapstructure:"oppo"` + VIVO VendorChannelID `mapstructure:"vivo"` + Meizu VendorChannelID `mapstructure:"meizu"` + Honor VendorChannelID `mapstructure:"honor"` + FCM VendorChannelID `mapstructure:"fcm"` +} + +// VendorChannelID 单个厂商的两类 channel 配置 +// 每个 channel 可携带厂商私有模板配置（如小米 mi_template_id、OPPO private_msg_template_id） +type VendorChannelID struct { + // channel_id + System string `mapstructure:"system"` + Chat string `mapstructure:"chat"` + + // 小米消息模板（可选，配置后私信消息下发时携带 channel_id 及 mi_template_id） + // 见：小米关于消息模板推送新规的更新通知 + MiSystemTemplateID string `mapstructure:"mi_system_template_id"` // 系统消息模板 id + MiChatTemplateID string `mapstructure:"mi_chat_template_id"` // 私聊消息模板 id + + // OPPO 私信模板（可选，仅 OPPO 厂商，配置后下发私信时携带） + // 见：OPUSH 私信模版校验能力接入说明 + OppoSystemPrivateTemplateID string `mapstructure:"oppo_system_private_template_id"` // 系统消息私信模板 id + OppoChatPrivateTemplateID string `mapstructure:"oppo_chat_private_template_id"` // 私聊消息私信模板 id +} diff --git a/internal/config/server.go b/internal/config/server.go index 4ceb2b6..52c3332 100644 --- a/internal/config/server.go +++ b/internal/config/server.go @@ -61,3 +61,11 @@ type UploadConfig struct { MaxFileSize int64 `mapstructure:"max_file_size"` AllowedTypes []string `mapstructure:"allowed_types"` } + +// FileCleanupConfig 聊天文件 TTL 过期清理配置 +type FileCleanupConfig struct { + Enabled bool `mapstructure:"enabled"` // 是否启用清理 worker + RetentionDays int `mapstructure:"retention_days"` // 保留天数，默认 7 + IntervalMinutes int `mapstructure:"interval_minutes"` // 扫描间隔（分钟），默认 360（6 小时） + BatchSize int `mapstructure:"batch_size"` // 单次扫描处理上限，默认 100 +} diff --git a/internal/database/database.go b/internal/database/database.go index bc51f02..5bd369e 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -231,6 +231,9 @@ func autoMigrate(db *gorm.DB) error { &model.TradeItem{}, &model.TradeImage{}, &model.TradeFavorite{}, + + // 文件上传记录（用于聊天文件 TTL 过期清理） + &model.UploadedFile{}, ) if err != nil { return err diff --git a/internal/dto/message_converter.go b/internal/dto/message_converter.go index aa303ee..f00e4cb 100644 --- a/internal/dto/message_converter.go +++ b/internal/dto/message_converter.go @@ -8,6 +8,13 @@ import ( // ConvertMessageToResponse 将Message转换为MessageResponse func ConvertMessageToResponse(message *model.Message) *MessageResponse { + return ConvertMessageToResponseWithExpiry(message, nil) +} + +// ConvertMessageToResponseWithExpiry 将Message转换为MessageResponse，并对已过期的 file segment 注入 expired 标记。 +// expiredURLSet 为已从 S3 清理的文件 URL 集合（可为 nil，表示不标记）。 +// 注意：注入时对 file segment 的 Data 做深拷贝，避免污染消息缓存。 +func ConvertMessageToResponseWithExpiry(message *model.Message, expiredURLSet map[string]struct{}) *MessageResponse { if message == nil { return nil } @@ -15,6 +22,19 @@ func ConvertMessageToResponse(message *model.Message) *MessageResponse { // 直接使用 segments，不需要解析 segments := make(model.MessageSegments, len(message.Segments)) for i, seg := range message.Segments { + // 对 file 类型且 URL 已过期的 segment，深拷贝 Data 并注入 expired 标记 + if seg.Type == string(model.ContentTypeFile) && len(expiredURLSet) > 0 { + url, _ := seg.Data["url"].(string) + if _, expired := expiredURLSet[url]; expired { + newData := make(map[string]any, len(seg.Data)+1) + for k, v := range seg.Data { + newData[k] = v + } + newData["expired"] = true + segments[i] = model.MessageSegment{Type: seg.Type, Data: newData} + continue + } + } segments[i] = model.MessageSegment{ Type: seg.Type, Data: seg.Data, @@ -107,6 +127,15 @@ func ConvertMessagesToResponse(messages []*model.Message) []*MessageResponse { return result } +// ConvertMessagesToResponseWithExpiry 将Message列表转换为响应列表，并标记过期文件 +func ConvertMessagesToResponseWithExpiry(messages []*model.Message, expiredURLSet map[string]struct{}) []*MessageResponse { + result := make([]*MessageResponse, 0, len(messages)) + for _, msg := range messages { + result = append(result, ConvertMessageToResponseWithExpiry(msg, expiredURLSet)) + } + return result +} + // ==================== PushRecord 转换 ==================== diff --git a/internal/handler/message_handler.go b/internal/handler/message_handler.go index 1ca83f8..5813540 100644 --- a/internal/handler/message_handler.go +++ b/internal/handler/message_handler.go @@ -90,20 +90,37 @@ type MessageHandler struct { messageService service.MessageService userService service.UserService groupService service.GroupService + uploadService service.UploadService wsPublisher ws.MessagePublisher } // NewMessageHandler 创建消息处理器 -func NewMessageHandler(chatService service.ChatService, messageService service.MessageService, userService service.UserService, groupService service.GroupService, wsPublisher ws.MessagePublisher) *MessageHandler { +func NewMessageHandler(chatService service.ChatService, messageService service.MessageService, userService service.UserService, groupService service.GroupService, uploadService service.UploadService, wsPublisher ws.MessagePublisher) *MessageHandler { return &MessageHandler{ chatService: chatService, messageService: messageService, userService: userService, groupService: groupService, + uploadService: uploadService, wsPublisher: wsPublisher, } } +// convertMessagesWithExpiry 将消息列表转换为响应，并对已过期的文件 segment 注入 expired 标记。 +// 若 uploadService 不可用或查询失败，降级为不带过期标记的普通转换。 +func (h *MessageHandler) convertMessagesWithExpiry(ctx context.Context, messages []*model.Message) []*dto.MessageResponse { + var expiredSet map[string]struct{} + if h.uploadService != nil { + if es, err := h.uploadService.GetExpiredURLSet(ctx); err == nil { + expiredSet = es + } + } + if len(expiredSet) == 0 { + return dto.ConvertMessagesToResponse(messages) + } + return dto.ConvertMessagesToResponseWithExpiry(messages, expiredSet) +} + // HandleTyping 输入状态上报 // POST /api/v1/conversations/typing func (h *MessageHandler) HandleTyping(c *gin.Context) { @@ -287,8 +304,8 @@ func (h *MessageHandler) GetMessages(c *gin.Context) { return } - // 转换为响应格式 - result := dto.ConvertMessagesToResponse(messages) + // 转换为响应格式（标记已过期文件） + result := h.convertMessagesWithExpiry(c.Request.Context(), messages) response.Success(c, &dto.MessageSyncResponse{ Messages: result, @@ -315,8 +332,8 @@ func (h *MessageHandler) GetMessages(c *gin.Context) { return } - // 转换为响应格式 - result := dto.ConvertMessagesToResponse(messages) + // 转换为响应格式（标记已过期文件） + result := h.convertMessagesWithExpiry(c.Request.Context(), messages) response.Success(c, &dto.MessageSyncResponse{ Messages: result, @@ -335,8 +352,8 @@ func (h *MessageHandler) GetMessages(c *gin.Context) { return } - // 转换为响应格式 - result := dto.ConvertMessagesToResponse(messages) + // 转换为响应格式（标记已过期文件） + result := h.convertMessagesWithExpiry(c.Request.Context(), messages) response.Paginated(c, result, total, page, pageSize) } @@ -857,8 +874,8 @@ func (h *MessageHandler) HandleGetMessages(c *gin.Context) { return } - // 转换为响应格式 - result := dto.ConvertMessagesToResponse(messages) + // 转换为响应格式（标记已过期文件） + result := h.convertMessagesWithExpiry(c.Request.Context(), messages) response.Success(c, &dto.MessageSyncResponse{ Messages: result, @@ -885,8 +902,8 @@ func (h *MessageHandler) HandleGetMessages(c *gin.Context) { return } - // 转换为响应格式 - result := dto.ConvertMessagesToResponse(messages) + // 转换为响应格式（标记已过期文件） + result := h.convertMessagesWithExpiry(c.Request.Context(), messages) response.Success(c, &dto.MessageSyncResponse{ Messages: result, @@ -905,8 +922,8 @@ func (h *MessageHandler) HandleGetMessages(c *gin.Context) { return } - // 转换为响应格式 - result := dto.ConvertMessagesToResponse(messages) + // 转换为响应格式（标记已过期文件） + result := h.convertMessagesWithExpiry(c.Request.Context(), messages) response.Paginated(c, result, total, page, pageSize) } @@ -1067,8 +1084,8 @@ func (h *MessageHandler) GetMessagesByCursor(c *gin.Context) { return } - // 转换为响应格式 - items := dto.ConvertMessagesToResponse(result.Items) + // 转换为响应格式（标记已过期文件） + items := h.convertMessagesWithExpiry(c.Request.Context(), result.Items) response.Success(c, &dto.MessageCursorPageResponse{ Items: items, diff --git a/internal/handler/upload_handler.go b/internal/handler/upload_handler.go index fdf22f4..656e627 100644 --- a/internal/handler/upload_handler.go +++ b/internal/handler/upload_handler.go @@ -92,3 +92,39 @@ func (h *UploadHandler) UploadCover(c *gin.Context) { response.Success(c, gin.H{"url": url}) } + +// UploadFile 上传聊天/通用文件（非图片类资源） +// form 字段：file（必填）、folder（可选，默认 chat） +func (h *UploadHandler) UploadFile(c *gin.Context) { + userID := c.GetString("user_id") + if userID == "" { + response.Unauthorized(c, "") + return + } + + file, err := c.FormFile("file") + if err != nil { + response.BadRequest(c, "file is required") + return + } + + folder := c.DefaultPostForm("folder", "chat") + + result, err := h.uploadService.UploadFile(c.Request.Context(), file, folder, userID) + if err != nil { + // 参数类错误（大小/类型）返回 400，其余返回 500 + if file.Size > 0 && file.Size <= 100<<20 { + response.BadRequest(c, err.Error()) + return + } + response.InternalServerError(c, "failed to upload file") + return + } + + response.Success(c, gin.H{ + "url": result.URL, + "name": result.Name, + "size": result.Size, + "mime_type": result.MimeType, + }) +} diff --git a/internal/model/uploaded_file.go b/internal/model/uploaded_file.go new file mode 100644 index 0000000..1dd1aca --- /dev/null +++ b/internal/model/uploaded_file.go @@ -0,0 +1,26 @@ +package model + +import "time" + +// UploadedFile 聊天/通用文件上传记录 +// 用于跟踪 files/ 前缀下的 S3 对象，配合 FileCleanupWorker 实现 TTL 过期清理。 +// 注意：images/avatars/covers 等内容寻址资源不记录本表，不会被自动清理。 +type UploadedFile struct { + ID uint `gorm:"primaryKey;autoIncrement" json:"id"` + URL string `gorm:"size:512;index;not null" json:"url"` // 完整访问 URL，DTO 标记失效用 + ObjectName string `gorm:"size:256;index;not null" json:"object_name"` // S3 对象名 files/{folder}/{hash}{ext} + Name string `gorm:"size:255" json:"name"` // 原始文件名 + Size int64 `json:"size"` // 字节数 + MimeType string `gorm:"size:128" json:"mime_type"` + Folder string `gorm:"size:32;index" json:"folder"` // chat / post 等来源 + UploaderID string `gorm:"size:50;index" json:"uploader_id"` // 上传者 UUID + Expired bool `gorm:"index;default:false" json:"expired"` // 是否已从 S3 清理 + ExpiredAt *time.Time `json:"expired_at,omitempty"` // 清理时间 + CreatedAt time.Time `gorm:"index" json:"created_at"` // 上传时间，TTL 计算基准 + UpdatedAt time.Time `json:"updated_at"` +} + +// TableName 表名 +func (UploadedFile) TableName() string { + return "uploaded_files" +} diff --git a/internal/pkg/jpush/client.go b/internal/pkg/jpush/client.go index bc401ac..d36049b 100644 --- a/internal/pkg/jpush/client.go +++ b/internal/pkg/jpush/client.go @@ -66,6 +66,18 @@ type Notification struct { IOS *IOSNotif `json:"ios,omitempty"` } +// ThirdPartyChannel 厂商通道配置，置于 push payload 的 options.third_party_channel +// 用于向各厂商（小米/华为/OPPO/VIVO/魅族/FCM/荣耀）下发 channel_id 等厂商私有参数 +type ThirdPartyChannel struct { + Xiaomi map[string]any `json:"xiaomi,omitempty"` + Huawei map[string]any `json:"huawei,omitempty"` + OPPO map[string]any `json:"oppo,omitempty"` + VIVO map[string]any `json:"vivo,omitempty"` + Meizu map[string]any `json:"meizu,omitempty"` + FCM map[string]any `json:"fcm,omitempty"` + Honor map[string]any `json:"honor,omitempty"` +} + type AndroidNotif struct { Alert string `json:"alert"` Title string `json:"title,omitempty"` @@ -244,7 +256,7 @@ func (c *Client) Push(payload map[string]any) (*PushResponse, *RateLimitInfo, er return result, rateLimit, parseErr } -func (c *Client) PushByRegistrationIDs(registrationIDs []string, notification *Notification) (*PushResponse, error) { +func (c *Client) PushByRegistrationIDs(registrationIDs []string, notification *Notification, tpc *ThirdPartyChannel) (*PushResponse, error) { if len(registrationIDs) == 0 { return nil, fmt.Errorf("registration IDs cannot be empty") } @@ -262,10 +274,7 @@ func (c *Client) PushByRegistrationIDs(registrationIDs []string, notification *N "platform": "all", "audience": map[string]any{"registration_id": registrationIDs}, "notification": notification, - "options": map[string]any{ - "time_to_live": 86400, - "apns_production": c.production, - }, + "options": c.buildOptions(tpc), } var result *PushResponse @@ -292,7 +301,7 @@ func (c *Client) PushByRegistrationIDs(registrationIDs []string, notification *N return result, nil } -func (c *Client) PushByAliases(aliases []string, notification *Notification) (*PushResponse, error) { +func (c *Client) PushByAliases(aliases []string, notification *Notification, tpc *ThirdPartyChannel) (*PushResponse, error) { if len(aliases) == 0 { return nil, fmt.Errorf("aliases cannot be empty") } @@ -310,10 +319,7 @@ func (c *Client) PushByAliases(aliases []string, notification *Notification) (*P "platform": "all", "audience": map[string]any{"alias": aliases}, "notification": notification, - "options": map[string]any{ - "time_to_live": 86400, - "apns_production": c.production, - }, + "options": c.buildOptions(tpc), } result, _, err := c.Push(payload) @@ -332,7 +338,7 @@ func (c *Client) PushByAliases(aliases []string, notification *Notification) (*P return result, nil } -func (c *Client) PushAll(notification *Notification) (*PushResponse, error) { +func (c *Client) PushAll(notification *Notification, tpc *ThirdPartyChannel) (*PushResponse, error) { c.logger.Info("jpush push all", zap.String("title", notification.Android.Title), zap.String("alert", notification.Alert), @@ -342,10 +348,7 @@ func (c *Client) PushAll(notification *Notification) (*PushResponse, error) { "platform": "all", "audience": "all", "notification": notification, - "options": map[string]any{ - "time_to_live": 86400, - "apns_production": c.production, - }, + "options": c.buildOptions(tpc), } result, _, err := c.Push(payload) @@ -362,41 +365,24 @@ func (c *Client) PushAll(notification *Notification) (*PushResponse, error) { return result, nil } -func BuildNotification(title, body string, notificationType string, extras map[string]any) *Notification { - androidExtras := map[string]any{ - "notification_type": notificationType, +// buildOptions 构造 push payload 的 options 部分 +// 当 tpc 非空时加入 third_party_channel 字段，用于向各厂商下发 channel_id +func (c *Client) buildOptions(tpc *ThirdPartyChannel) map[string]any { + opts := map[string]any{ + "time_to_live": 86400, + "apns_production": c.production, } - 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: map[string]any{ - "title": title, - "body": body, - }, - Sound: "default", - Badge: "+1", - Extras: iosExtras, - }, + if tpc != nil { + opts["third_party_channel"] = tpc } + return opts } +// 通知与厂商通道的构建逻辑已迁移至 internal/pkg/jpush/vendor 子包： +// - 通用通知（Android/iOS）由 vendor.TextRenderer 产出 +// - 各厂商私有字段（channel_id / 小米 mi_template / OPPO 私信模板）由 vendor.PrivateConverter 产出 +// - 组装入口为 vendor.Factory.Build，产物 *vendor.PushPayload 含 Notification 与 ThirdPartyChannel + type DeviceInfo struct { Tags []string `json:"tags"` Alias string `json:"alias"` diff --git a/internal/pkg/jpush/vendor/converters.go b/internal/pkg/jpush/vendor/converters.go new file mode 100644 index 0000000..8046051 --- /dev/null +++ b/internal/pkg/jpush/vendor/converters.go @@ -0,0 +1,156 @@ +package vendor + +import ( + "encoding/json" + + "with_you/internal/config" +) + +// resolveChannelID 按场景选择 channel_id：厂商专属值优先，空则回退到全局默认 +// scene 为 SceneSystem 时取 system 配置，其余取 chat 配置 +func resolveChannelID(scene, vendorSys, vendorChat, defaultSys, defaultChat string) string { + if scene == SceneSystem { + if vendorSys != "" { + return vendorSys + } + return defaultSys + } + if vendorChat != "" { + return vendorChat + } + return defaultChat +} + +// channelOnlyMap 仅 channel_id 厂商的通用转换逻辑（厂商空则回退全局默认） +// 最终 channel_id 为空时返回 nil（不下发该厂商） +func channelOnlyMap(scene, vendorSys, vendorChat, defaultSys, defaultChat string) map[string]any { + cid := resolveChannelID(scene, vendorSys, vendorChat, defaultSys, defaultChat) + if cid == "" { + return nil + } + return map[string]any{"channel_id": cid} +} + +// huaweiConverter 华为厂商转换器（仅 channel_id） +type huaweiConverter struct{} + +func (huaweiConverter) VendorName() string { return "huawei" } +func (huaweiConverter) Convert(params MessageParams, cfg config.ChannelConfig) map[string]any { + return channelOnlyMap(params.Scene, cfg.Vendor.Huawei.System, cfg.Vendor.Huawei.Chat, cfg.System, cfg.Chat) +} + +// vivoConverter VIVO 厂商转换器（仅 channel_id） +type vivoConverter struct{} + +func (vivoConverter) VendorName() string { return "vivo" } +func (vivoConverter) Convert(params MessageParams, cfg config.ChannelConfig) map[string]any { + return channelOnlyMap(params.Scene, cfg.Vendor.VIVO.System, cfg.Vendor.VIVO.Chat, cfg.System, cfg.Chat) +} + +// meizuConverter 魅族厂商转换器（仅 channel_id） +type meizuConverter struct{} + +func (meizuConverter) VendorName() string { return "meizu" } +func (meizuConverter) Convert(params MessageParams, cfg config.ChannelConfig) map[string]any { + return channelOnlyMap(params.Scene, cfg.Vendor.Meizu.System, cfg.Vendor.Meizu.Chat, cfg.System, cfg.Chat) +} + +// honorConverter 荣耀厂商转换器（仅 channel_id） +type honorConverter struct{} + +func (honorConverter) VendorName() string { return "honor" } +func (honorConverter) Convert(params MessageParams, cfg config.ChannelConfig) map[string]any { + return channelOnlyMap(params.Scene, cfg.Vendor.Honor.System, cfg.Vendor.Honor.Chat, cfg.System, cfg.Chat) +} + +// fcmConverter FCM 厂商转换器（仅 channel_id） +type fcmConverter struct{} + +func (fcmConverter) VendorName() string { return "fcm" } +func (fcmConverter) Convert(params MessageParams, cfg config.ChannelConfig) map[string]any { + return channelOnlyMap(params.Scene, cfg.Vendor.FCM.System, cfg.Vendor.FCM.Chat, cfg.System, cfg.Chat) +} + +// xiaomiConverter 小米厂商转换器 +// 配置 mi_template_id 时，私信消息下发时携带 channel_id 及 mi_template_id + mi_template_param +type xiaomiConverter struct{} + +func (xiaomiConverter) VendorName() string { return "xiaomi" } + +func (xiaomiConverter) Convert(params MessageParams, cfg config.ChannelConfig) map[string]any { + x := cfg.Vendor.Xiaomi + cid := resolveChannelID(params.Scene, x.System, x.Chat, cfg.System, cfg.Chat) + // 小米模板按场景选择 + templateID := x.MiChatTemplateID + if params.Scene == SceneSystem { + templateID = x.MiSystemTemplateID + } + + // 最终 channel_id 与模板均未配置则跳过 + if cid == "" && templateID == "" { + return nil + } + + m := map[string]any{} + if cid != "" { + m["channel_id"] = cid + } + if templateID != "" { + m["mi_template_id"] = templateID + // mi_template_param 为模板参数的 JSON 字符串 + m["mi_template_param"] = buildMiTemplateParam(params) + } + return m +} + +// buildMiTemplateParam 生成小米消息模板参数 JSON 字符串 +// 默认占位符：sender / title / content，可按实际小米模板占位符在集中处调整 +func buildMiTemplateParam(params MessageParams) string { + param := map[string]any{ + "sender": params.SenderName, + "title": params.Title, + "content": params.Body, + } + b, err := json.Marshal(param) + if err != nil { + return "{}" + } + return string(b) +} + +// oppoConverter OPPO 厂商转换器 +// 配置 private_msg_template_id 时，私信消息下发时携带 private_msg_template_id + private_title_parameters + private_content_parameters +type oppoConverter struct{} + +func (oppoConverter) VendorName() string { return "oppo" } + +func (oppoConverter) Convert(params MessageParams, cfg config.ChannelConfig) map[string]any { + o := cfg.Vendor.OPPO + cid := resolveChannelID(params.Scene, o.System, o.Chat, cfg.System, cfg.Chat) + // OPPO 私信模板按场景选择 + templateID := o.OppoChatPrivateTemplateID + if params.Scene == SceneSystem { + templateID = o.OppoSystemPrivateTemplateID + } + + if cid == "" && templateID == "" { + return nil + } + + m := map[string]any{} + if cid != "" { + m["channel_id"] = cid + } + if templateID != "" { + m["private_msg_template_id"] = templateID + // private_title_parameters / private_content_parameters 为 JSONObject + m["private_title_parameters"] = map[string]any{ + "title": params.Title, + } + m["private_content_parameters"] = map[string]any{ + "content": params.Body, + "sender": params.SenderName, + } + } + return m +} diff --git a/internal/pkg/jpush/vendor/params.go b/internal/pkg/jpush/vendor/params.go new file mode 100644 index 0000000..8075352 --- /dev/null +++ b/internal/pkg/jpush/vendor/params.go @@ -0,0 +1,151 @@ +// Package vendor 提供厂商无关的消息参数抽象与各厂商推送字段转换。 +// +// 设计要点： +// - 调用方只需提供厂商无关的 MessageParams，不感知厂商差异。 +// - 由于无法预知目标设备的具体厂商，JPush 要求所有厂商私有配置全数下发， +// 由 JPush 服务端按目标设备厂商路由。 +// - 双层架构：TextRenderer（基础层，产出 Android/iOS 通用通知）+ +// PrivateConverter（附加层，产出各厂商 third_party_channel 私有字段）。 +package vendor + +import ( + "with_you/internal/config" + "with_you/internal/pkg/jpush" +) + +// 场景常量 +const ( + // SceneSystem 系统消息/通知场景 + SceneSystem = "system" + // SceneChat 私聊/群聊消息场景 + SceneChat = "chat" +) + +// MessageParams 厂商无关的统一消息参数（超集，用不上的字段厂商可忽略） +type MessageParams struct { + // Scene 决定使用哪类 channel/模板：SceneSystem 或 SceneChat + Scene string + // Title 通知标题（如发送者昵称、群名:发送者） + Title string + // Body 通知正文（消息内容预览） + Body string + // SenderName 发送者昵称 + SenderName string + // SenderID 发送者 ID + SenderID string + // ConversationName 会话名称（群聊为群名） + ConversationName string + // ConversationID 会话 ID + ConversationID string + // MessageType 消息类型标识（chat_message / 系统通知类型），填充到 extras.notification_type + MessageType string + // ActionTime 操作时间（RFC3339） + ActionTime string + // LargeIcon 大图标 URL（群头像/发送者头像） + LargeIcon string + // Extra 兜底扩展字段，合并进 android/ios 的 extras + Extra map[string]any +} + +// PushPayload 工厂统一产物，client 据此组装最终推送 payload +type PushPayload struct { + // Notification 含 android + ios，由 TextRenderer 产出 + Notification *jpush.Notification + // TPC 各厂商私有字段（options.third_party_channel），由 PrivateConverter 产出 + TPC *jpush.ThirdPartyChannel +} + +// TextRenderer 通用文本渲染器（基础层） +// 产出所有厂商共享的通用通知：Android 通用通知 + iOS 通用通知 +type TextRenderer interface { + // RenderAndroid 渲染 Android 通用通知（所有 Android 厂商共享） + RenderAndroid(params MessageParams, cfg config.ChannelConfig) *jpush.AndroidNotif + // RenderIOS 渲染 iOS 通用通知（按场景设置 thread-id 分组） + RenderIOS(params MessageParams, cfg config.ChannelConfig) *jpush.IOSNotif +} + +// PrivateConverter 厂商私有转换器（附加层） +// 产出 third_party_channel.{vendor} 的私有字段（channel_id、模板等） +type PrivateConverter interface { + // VendorName 返回厂商标识，对应 third_party_channel 的 key（xiaomi/oppo/huawei/...） + VendorName() string + // Convert 根据消息参数 + 全局配置产出该厂商私有字段 map + // 厂商专属 channel_id 为空时回退到全局 cfg.System/Chat + // 返回 nil 表示该厂商无需下发（如最终 channel_id 与模板均未配置） + Convert(params MessageParams, cfg config.ChannelConfig) map[string]any +} + +// Factory 持有通用渲染器与全部厂商私有转换器 +type Factory struct { + renderer TextRenderer + privateConverters []PrivateConverter +} + +// NewFactory 创建工厂，注册默认渲染器与全部厂商转换器 +func NewFactory() *Factory { + return &Factory{ + renderer: newDefaultRenderer(), + privateConverters: []PrivateConverter{ + xiaomiConverter{}, + huaweiConverter{}, + oppoConverter{}, + vivoConverter{}, + meizuConverter{}, + honorConverter{}, + fcmConverter{}, + }, + } +} + +// NewFactoryWith 自定义渲染器与转换器创建工厂（便于测试） +func NewFactoryWith(renderer TextRenderer, converters []PrivateConverter) *Factory { + return &Factory{renderer: renderer, privateConverters: converters} +} + +// Build 根据统一参数与配置，组装完整的 PushPayload +// 调用通用渲染器产出 Notification，调用各私有转换器产出 ThirdPartyChannel +func (f *Factory) Build(params MessageParams, cfg config.ChannelConfig) *PushPayload { + android := f.renderer.RenderAndroid(params, cfg) + ios := f.renderer.RenderIOS(params, cfg) + + notification := &jpush.Notification{ + Alert: params.Body, + Android: android, + IOS: ios, + } + + tpc := &jpush.ThirdPartyChannel{} + for _, c := range f.privateConverters { + m := c.Convert(params, cfg) + if m == nil { + continue + } + switch c.VendorName() { + case "xiaomi": + tpc.Xiaomi = m + case "huawei": + tpc.Huawei = m + case "oppo": + tpc.OPPO = m + case "vivo": + tpc.VIVO = m + case "meizu": + tpc.Meizu = m + case "honor": + tpc.Honor = m + case "fcm": + tpc.FCM = m + } + } + + // 全部厂商字段为空则置 nil，避免序列化出空的 third_party_channel + if tpc.Xiaomi == nil && tpc.Huawei == nil && tpc.OPPO == nil && + tpc.VIVO == nil && tpc.Meizu == nil && tpc.Honor == nil && tpc.FCM == nil { + tpc = nil + } + + return &PushPayload{ + Notification: notification, + TPC: tpc, + } +} diff --git a/internal/pkg/jpush/vendor/payload_compat_test.go b/internal/pkg/jpush/vendor/payload_compat_test.go new file mode 100644 index 0000000..22bec37 --- /dev/null +++ b/internal/pkg/jpush/vendor/payload_compat_test.go @@ -0,0 +1,135 @@ +package vendor + +import ( + "encoding/json" + "strings" + "testing" + + "with_you/internal/config" +) + +// TestPayloadEmptyConfigMatchesLegacy 验证：未配置任何 channel/模板时， +// Factory 产出的推送 payload 与改动前 BuildNotification 行为一致。 +// +// 改动前 BuildNotification(title, body, "message", extras) 产出： +// notification: { alert, android:{alert,title,extras:{notification_type}+extras,priority:1}, ios:{alert:{title,body},sound:default,badge:+1,extras} } +// options: { time_to_live: 86400, apns_production } // 无 third_party_channel +func TestPayloadEmptyConfigMatchesLegacy(t *testing.T) { + factory := NewFactory() + cfg := config.ChannelConfig{} // 全空配置 + params := MessageParams{ + Scene: SceneChat, + Title: "张三", + Body: "你好", + MessageType: "message", + Extra: map[string]any{ + "message_id": "m1", + "conversation_id": "c1", + "sender_id": "u1", + "category": "chat", + }, + } + + payload := factory.Build(params, cfg) + + // 1. TPC 必须为 nil（无 third_party_channel 下发，与改动前一致） + if payload.TPC != nil { + t.Fatalf("TPC should be nil when all channel config empty, got: %+v", payload.TPC) + } + + notif := payload.Notification + if notif == nil || notif.Android == nil || notif.IOS == nil { + t.Fatalf("notification/android/ios must not be nil") + } + + // 2. Android 字段对比 + if notif.Android.Alert != "你好" || notif.Android.Title != "张三" || notif.Android.Priority != 1 { + t.Errorf("Android basic fields mismatch: %+v", notif.Android) + } + if notif.Android.ChannelID != "" { + t.Errorf("Android.ChannelID = %q, want empty", notif.Android.ChannelID) + } + if notif.Android.Extras["notification_type"] != "message" || notif.Android.Extras["message_id"] != "m1" { + t.Errorf("Android.Extras mismatch: %+v", notif.Android.Extras) + } + + // 3. iOS 字段对比 + if notif.IOS.Sound != "default" || notif.IOS.Badge != "+1" { + t.Errorf("IOS basic fields mismatch: %+v", notif.IOS) + } + if notif.IOS.ThreadID != "" { + t.Errorf("IOS.ThreadID = %q, want empty", notif.IOS.ThreadID) + } + if notif.IOS.Extras["notification_type"] != "message" { + t.Errorf("IOS.Extras[notification_type] mismatch: %+v", notif.IOS.Extras) + } + + // 4. 序列化 JSON 确认厂商私有字段不出现 + jsonBytes, err := json.Marshal(notif) + if err != nil { + t.Fatalf("marshal: %v", err) + } + jsonStr := string(jsonBytes) + for _, key := range []string{`"channel_id"`, `"thread-id"`, `"third_party_channel"`, `"mi_template"`, `"private_msg_template_id"`} { + if strings.Contains(jsonStr, key) { + t.Errorf("notification JSON should not contain %s when config empty, got: %s", key, jsonStr) + } + } + t.Logf("notification JSON: %s", jsonStr) +} + +// TestPayloadWithChannelConfig 验证：配置 channel_id 后正确下发，且不影响通用通知字段 +func TestPayloadWithChannelConfig(t *testing.T) { + factory := NewFactory() + cfg := config.ChannelConfig{ + System: "sys_channel", + Chat: "chat_channel", + } + params := MessageParams{Scene: SceneChat, Title: "T", Body: "B", MessageType: "chat_message"} + + payload := factory.Build(params, cfg) + + // 通用通知仍正常 + if payload.Notification.Android.Title != "T" { + t.Errorf("Android.Title = %q, want T", payload.Notification.Android.Title) + } + // Android channel_id 跟随场景（chat） + if payload.Notification.Android.ChannelID != "chat_channel" { + t.Errorf("Android.ChannelID = %q, want chat_channel", payload.Notification.Android.ChannelID) + } + // TPC 各厂商 channel_id 下发（华为/VIVO 等回退全局） + if payload.TPC == nil { + t.Fatal("TPC should not be nil when channel config present") + } + if payload.TPC.Huawei["channel_id"] != "chat_channel" { + t.Errorf("Huawei channel_id = %v, want chat_channel", payload.TPC.Huawei["channel_id"]) + } + t.Logf("TPC JSON: %s", mustJSON(payload.TPC)) +} + +// TestPayloadXiaomiTemplateOptional 验证：仅配 channel_id 不配模板时，小米不下发模板字段 +func TestPayloadXiaomiTemplateOptional(t *testing.T) { + factory := NewFactory() + cfg := config.ChannelConfig{ + Vendor: config.VendorChannel{ + Xiaomi: config.VendorChannelID{Chat: "mi_chat"}, + }, + } + params := MessageParams{Scene: SceneChat, Title: "T", Body: "B"} + + payload := factory.Build(params, cfg) + if payload.TPC == nil || payload.TPC.Xiaomi == nil { + t.Fatal("Xiaomi TPC should exist") + } + if _, ok := payload.TPC.Xiaomi["mi_template_id"]; ok { + t.Error("Xiaomi should NOT contain mi_template_id when template not configured") + } + if payload.TPC.Xiaomi["channel_id"] != "mi_chat" { + t.Errorf("Xiaomi channel_id = %v, want mi_chat", payload.TPC.Xiaomi["channel_id"]) + } +} + +func mustJSON(v any) string { + b, _ := json.Marshal(v) + return string(b) +} diff --git a/internal/pkg/jpush/vendor/text_renderer.go b/internal/pkg/jpush/vendor/text_renderer.go new file mode 100644 index 0000000..c09a2e9 --- /dev/null +++ b/internal/pkg/jpush/vendor/text_renderer.go @@ -0,0 +1,61 @@ +package vendor + +import ( + "with_you/internal/config" + "with_you/internal/pkg/jpush" +) + +// defaultRenderer 默认通用文本渲染器 +type defaultRenderer struct{} + +func newDefaultRenderer() *defaultRenderer { + return &defaultRenderer{} +} + +// mergeExtras 合并基础 extras 与额外 extras，返回新 map +func mergeExtras(messageType string, extra map[string]any) map[string]any { + out := make(map[string]any, len(extra)+1) + if messageType != "" { + out["notification_type"] = messageType + } + for k, v := range extra { + out[k] = v + } + return out +} + +// RenderAndroid 渲染 Android 通用通知 +// 所有 Android 厂商共享：alert/title/channel_id/extras/large_icon/priority +func (r *defaultRenderer) RenderAndroid(params MessageParams, cfg config.ChannelConfig) *jpush.AndroidNotif { + channelID := cfg.Chat + if params.Scene == SceneSystem { + channelID = cfg.System + } + return &jpush.AndroidNotif{ + Alert: params.Body, + Title: params.Title, + ChannelID: channelID, + LargeIcon: params.LargeIcon, + Extras: mergeExtras(params.MessageType, params.Extra), + Priority: 1, + } +} + +// RenderIOS 渲染 iOS 通用通知 +// iOS 走 APNs，通过 thread-id 实现按场景的通知分组 +func (r *defaultRenderer) RenderIOS(params MessageParams, cfg config.ChannelConfig) *jpush.IOSNotif { + threadID := cfg.IOS.ChatThreadID + if params.Scene == SceneSystem { + threadID = cfg.IOS.SystemThreadID + } + return &jpush.IOSNotif{ + Alert: map[string]any{ + "title": params.Title, + "body": params.Body, + }, + Sound: "default", + Badge: "+1", + ThreadID: threadID, + Extras: mergeExtras(params.MessageType, params.Extra), + } +} diff --git a/internal/repository/uploaded_file_repo.go b/internal/repository/uploaded_file_repo.go new file mode 100644 index 0000000..7f7ebcc --- /dev/null +++ b/internal/repository/uploaded_file_repo.go @@ -0,0 +1,91 @@ +package repository + +import ( + "context" + "time" + + "with_you/internal/model" + + "gorm.io/gorm" +) + +// UploadedFileRepository 文件上传记录仓储接口 +type UploadedFileRepository interface { + // Create 写入一条上传记录 + Create(ctx context.Context, file *model.UploadedFile) error + // GetByURL 按 URL 查询记录（DTO 失效标记用） + GetByURL(ctx context.Context, url string) (*model.UploadedFile, error) + // ListPendingCleanup 取 created_at < before 且 expired=false 的记录，分批 + ListPendingCleanup(ctx context.Context, before time.Time, limit int) ([]model.UploadedFile, error) + // MarkExpired 标记某条记录已清理 + MarkExpired(ctx context.Context, id uint, at time.Time) error + // GetExpiredURLSet 批量查询已标记过期的 URL 集合（DTO 注入用） + // 注意：这是全量扫描 expired=true 的记录，适用于上传量不大的场景； + // 若 uploaded_files 表增长过大，可改为按时间窗口分页。 + GetExpiredURLSet(ctx context.Context) (map[string]struct{}, error) +} + +// uploadedFileRepository 文件上传记录仓储实现 +type uploadedFileRepository struct { + db *gorm.DB +} + +// NewUploadedFileRepository 创建文件上传记录仓储 +func NewUploadedFileRepository(db *gorm.DB) UploadedFileRepository { + return &uploadedFileRepository{db: db} +} + +// Create 写入一条上传记录 +func (r *uploadedFileRepository) Create(ctx context.Context, file *model.UploadedFile) error { + return r.db.WithContext(ctx).Create(file).Error +} + +// GetByURL 按 URL 查询记录 +func (r *uploadedFileRepository) GetByURL(ctx context.Context, url string) (*model.UploadedFile, error) { + var file model.UploadedFile + if err := r.db.WithContext(ctx).Where("url = ?", url).First(&file).Error; err != nil { + return nil, err + } + return &file, nil +} + +// ListPendingCleanup 取待清理记录 +func (r *uploadedFileRepository) ListPendingCleanup(ctx context.Context, before time.Time, limit int) ([]model.UploadedFile, error) { + if limit <= 0 { + limit = 100 + } + var files []model.UploadedFile + err := r.db.WithContext(ctx). + Where("created_at < ? AND expired = ?", before, false). + Order("created_at ASC"). + Limit(limit). + Find(&files).Error + return files, err +} + +// MarkExpired 标记已清理 +func (r *uploadedFileRepository) MarkExpired(ctx context.Context, id uint, at time.Time) error { + return r.db.WithContext(ctx).Model(&model.UploadedFile{}). + Where("id = ?", id). + Updates(map[string]any{ + "expired": true, + "expired_at": at, + }).Error +} + +// GetExpiredURLSet 批量查询已过期 URL 集合 +func (r *uploadedFileRepository) GetExpiredURLSet(ctx context.Context) (map[string]struct{}, error) { + var urls []string + err := r.db.WithContext(ctx). + Model(&model.UploadedFile{}). + Where("expired = ?", true). + Pluck("url", &urls).Error + if err != nil { + return nil, err + } + set := make(map[string]struct{}, len(urls)) + for _, u := range urls { + set[u] = struct{}{} + } + return set, nil +} diff --git a/internal/router/router.go b/internal/router/router.go index d2a0d37..9e29332 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -436,6 +436,7 @@ func (r *Router) setupRoutes() { uploads := v1.Group("/uploads") { uploads.POST("/images", authMiddleware, r.UploadHandler.UploadImage) + uploads.POST("/files", authMiddleware, r.UploadHandler.UploadFile) } // 推送相关路由 diff --git a/internal/service/file_cleanup_worker.go b/internal/service/file_cleanup_worker.go new file mode 100644 index 0000000..e4e07aa --- /dev/null +++ b/internal/service/file_cleanup_worker.go @@ -0,0 +1,261 @@ +package service + +import ( + "context" + "errors" + "sync" + "time" + + "with_you/internal/config" + "with_you/internal/pkg/s3" + "with_you/internal/repository" + + redislib "github.com/redis/go-redis/v9" + "go.uber.org/zap" +) + +// FileCleanupWorker 聊天文件 TTL 过期清理 worker +// 周期性扫描 uploaded_files 表，将超过保留期的文件从 S3 删除并标记 expired。 +// 多实例部署时通过 Redis 分布式锁保证仅主实例执行清理。 +type FileCleanupWorker struct { + cfg *config.Config + uploadedRepo repository.UploadedFileRepository + s3Client *s3.Client + rdb *redislib.Client // 用于分布式锁（可为 nil，降级为本地运行） + stopOnce sync.Once + stopCh chan struct{} + doneCh chan struct{} +} + +const ( + // fileCleanupLockKey Redis 分布式锁键 + fileCleanupLockKey = "file_cleanup:leader_lock" + // fileCleanupLockTTL 锁持有时间（略大于扫描间隔，确保单实例持有） + fileCleanupLockTTL = 7 * time.Hour + // fileCleanupLockRetryInterval 竞争锁的重试间隔 + fileCleanupLockRetryInterval = 1 * time.Minute +) + +// NewFileCleanupWorker 创建文件清理 worker +func NewFileCleanupWorker( + cfg *config.Config, + uploadedRepo repository.UploadedFileRepository, + s3Client *s3.Client, + rdb *redislib.Client, +) *FileCleanupWorker { + return &FileCleanupWorker{ + cfg: cfg, + uploadedRepo: uploadedRepo, + s3Client: s3Client, + rdb: rdb, + stopCh: make(chan struct{}), + doneCh: make(chan struct{}), + } +} + +// tryAcquireLock 尝试获取分布式锁，返回是否成功 +func (w *FileCleanupWorker) tryAcquireLock(ctx context.Context) bool { + if w.rdb == nil { + return true // 无 Redis 时降级为本地运行 + } + ok, err := w.rdb.SetNX(ctx, fileCleanupLockKey, "1", fileCleanupLockTTL).Result() + if err != nil { + zap.L().Warn("file cleanup: failed to acquire leader lock, running locally", zap.Error(err)) + return true + } + return ok +} + +// keepLockAlive 周期性续期分布式锁 +func (w *FileCleanupWorker) keepLockAlive(ctx context.Context) { + if w.rdb == nil { + return + } + ticker := time.NewTicker(fileCleanupLockTTL / 2) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-w.stopCh: + return + case <-ticker.C: + w.rdb.Expire(ctx, fileCleanupLockKey, fileCleanupLockTTL) + } + } +} + +// releaseLock 释放分布式锁 +func (w *FileCleanupWorker) releaseLock() { + if w.rdb == nil { + return + } + w.rdb.Del(context.Background(), fileCleanupLockKey) +} + +// Start 启动清理 worker +func (w *FileCleanupWorker) Start(ctx context.Context) { + if w == nil || w.cfg == nil || !w.cfg.FileCleanup.Enabled { + close(w.doneCh) + return + } + if w.uploadedRepo == nil || w.s3Client == nil { + zap.L().Warn("file cleanup: missing dependencies, worker disabled") + close(w.doneCh) + return + } + + // 扫描间隔，默认 6 小时，最小 5 分钟 + interval := time.Duration(w.cfg.FileCleanup.IntervalMinutes) * time.Minute + if interval < 5*time.Minute { + interval = 6 * time.Hour + } + + go func() { + defer close(w.doneCh) + + isLeader := false + lockCtx, lockCancel := context.WithCancel(ctx) + + for { + select { + case <-w.stopCh: + if isLeader { + w.releaseLock() + } + lockCancel() + return + default: + } + + acquired := w.tryAcquireLock(ctx) + if acquired { + if !isLeader { + zap.L().Info("file cleanup: acquired leader lock, becoming primary") + isLeader = true + go w.keepLockAlive(lockCtx) + + // 主实例启动后立即执行一次清理 + w.runOnce(ctx) + + ticker := time.NewTicker(interval) + for { + select { + case <-w.stopCh: + ticker.Stop() + lockCancel() + w.releaseLock() + return + case <-ticker.C: + w.runOnce(ctx) + } + } + } + } else { + zap.L().Debug("file cleanup: another instance holds leader lock, retrying later") + } + + select { + case <-w.stopCh: + lockCancel() + return + case <-time.After(fileCleanupLockRetryInterval): + } + } + }() +} + +// Stop 停止 worker +func (w *FileCleanupWorker) Stop() { + if w == nil { + return + } + w.stopOnce.Do(func() { + close(w.stopCh) + }) + <-w.doneCh + w.releaseLock() +} + +// runOnce 执行一次清理扫描 +func (w *FileCleanupWorker) runOnce(ctx context.Context) { + cfg := w.cfg.FileCleanup + retentionDays := cfg.RetentionDays + if retentionDays <= 0 { + retentionDays = 7 + } + batchSize := cfg.BatchSize + if batchSize <= 0 { + batchSize = 100 + } + + before := time.Now().Add(-time.Duration(retentionDays) * 24 * time.Hour) + totalDeleted := 0 + totalFailed := 0 + + for { + select { + case <-w.stopCh: + return + default: + } + + files, err := w.uploadedRepo.ListPendingCleanup(ctx, before, batchSize) + if err != nil { + zap.L().Warn("file cleanup: list pending failed", zap.Error(err)) + break + } + if len(files) == 0 { + break + } + + for _, f := range files { + select { + case <-w.stopCh: + return + default: + } + + // 删除 S3 对象 + if derr := w.s3Client.Delete(ctx, f.ObjectName); derr != nil { + // 删除失败不标记，下次重试 + totalFailed++ + zap.L().Warn("file cleanup: delete object failed, will retry", + zap.String("object", f.ObjectName), + zap.String("url", f.URL), + zap.Error(derr), + ) + continue + } + + // 标记已清理 + now := time.Now() + if merr := w.uploadedRepo.MarkExpired(ctx, f.ID, now); merr != nil { + zap.L().Warn("file cleanup: mark expired failed", + zap.Uint("id", f.ID), + zap.String("url", f.URL), + zap.Error(merr), + ) + totalFailed++ + } else { + totalDeleted++ + } + } + + // 本批次不足 batchSize，说明已无待清理记录 + if len(files) < batchSize { + break + } + } + + if totalDeleted > 0 || totalFailed > 0 { + zap.L().Info("file cleanup sweep done", + zap.Int("deleted", totalDeleted), + zap.Int("failed", totalFailed), + zap.Int("retention_days", retentionDays), + ) + } +} + +// 兜底：保证 errors 包被引用（若未来 runOnce 增加错误判断） +var _ = errors.Is diff --git a/internal/service/push_service.go b/internal/service/push_service.go index 80f8008..be8f50d 100644 --- a/internal/service/push_service.go +++ b/internal/service/push_service.go @@ -6,9 +6,11 @@ import ( "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" @@ -85,6 +87,11 @@ type pushServiceImpl struct { wsHub ws.MessagePublisher jpushClient *jpush.Client + // channelConfig 厂商通道 channel_id 配置（系统消息/私聊消息分别注册） + channelConfig config.ChannelConfig + // vendorFactory 厂商转换器工厂，按统一参数产出各厂商推送字段 + vendorFactory *vendor.Factory + pushQueue chan *pushTask stopChan chan struct{} } @@ -104,15 +111,18 @@ func NewPushService( 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, - pushQueue: make(chan *pushTask, PushQueueSize), - stopChan: make(chan struct{}), + pushRepo: pushRepo, + deviceRepo: deviceRepo, + messageRepo: messageRepo, + wsHub: publisher, + jpushClient: jpushClient, + channelConfig: channelConfig, + vendorFactory: vendor.NewFactory(), + pushQueue: make(chan *pushTask, PushQueueSize), + stopChan: make(chan struct{}), } } @@ -143,8 +153,8 @@ func (s *pushServiceImpl) PushToUser(ctx context.Context, userID string, message if err == nil && len(devices) > 0 { mobileDevices := getMobileDevices(devices) if len(mobileDevices) > 0 { - payload := buildMessagePayload(message) - pushErr := s.pushViaJPushBatch(mobileDevices, payload.Title, payload.Content, payload.Extras) + params := s.messageToParams(message) + pushErr := s.pushViaJPushBatch(mobileDevices, params) if pushErr == nil { s.batchCreatePushedRecords(ctx, userID, message.ID, mobileDevices) return nil @@ -296,7 +306,8 @@ func buildMessagePayload(message *model.Message) messagePayload { } // pushViaJPush 通过极光推送发送通知 -func (s *pushServiceImpl) pushViaJPush(ctx context.Context, device *model.DeviceToken, title, content string, extras map[string]any) error { +// 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") } @@ -305,8 +316,8 @@ func (s *pushServiceImpl) pushViaJPush(ctx context.Context, device *model.Device return errors.New("device has no registration ID") } - notification := jpush.BuildNotification(title, content, "message", extras) - _, err := s.jpushClient.PushByRegistrationIDs([]string{device.PushToken}, notification) + 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) } @@ -343,7 +354,9 @@ func (s *pushServiceImpl) batchCreatePushedRecords(ctx context.Context, userID s } // pushViaJPushBatch 批量通过极光推送发送通知 -func (s *pushServiceImpl) pushViaJPushBatch(devices []*model.DeviceToken, title, content string, extras map[string]any) error { +// pushViaJPushBatch 批量通过极光推送发送通知 +// params 为厂商无关统一参数，由 vendor 工厂转换为各厂商推送字段 +func (s *pushServiceImpl) pushViaJPushBatch(devices []*model.DeviceToken, params vendor.MessageParams) error { if !s.isJPushAvailable() { return errors.New("jpush client not configured") } @@ -359,8 +372,8 @@ func (s *pushServiceImpl) pushViaJPushBatch(devices []*model.DeviceToken, title, return errors.New("no valid registration IDs") } - notification := jpush.BuildNotification(title, content, "message", extras) - _, err := s.jpushClient.PushByRegistrationIDs(regIDs, notification) + 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) } @@ -520,7 +533,7 @@ func (s *pushServiceImpl) processPushTask(task *pushTask) { return } - payload := buildMessagePayload(task.message) + params := s.messageToParams(task.message) // 尝试使用极光推送（统一推送到所有移动设备） mobileDevices := getMobileDevices(devices) @@ -528,7 +541,7 @@ func (s *pushServiceImpl) processPushTask(task *pushTask) { if len(mobileDevices) > 0 { // 优先使用极光推送 if s.isJPushAvailable() { - err := s.pushViaJPushBatch(mobileDevices, payload.Title, payload.Content, payload.Extras) + err := s.pushViaJPushBatch(mobileDevices, params) if err == nil { s.batchCreatePushedRecords(ctx, task.userID, task.message.ID, mobileDevices) // 原始 pending 记录标记为已推送（被批量记录替代） @@ -549,7 +562,7 @@ func (s *pushServiceImpl) processPushTask(task *pushTask) { var pushErr error if s.isJPushAvailable() { - pushErr = s.pushViaJPush(ctx, device, payload.Title, payload.Content, payload.Extras) + pushErr = s.pushViaJPush(ctx, device, params) } else { pushErr = errors.New("jpush not configured") } @@ -583,6 +596,46 @@ 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 等通用路径 +func (s *pushServiceImpl) messageToParams(message *model.Message) vendor.MessageParams { + _ = message.Decrypt() + payload := buildMessagePayload(message) + scene := s.resolveChannelScene(message) + + senderName := "" + if message.ExtraData != nil { + senderName = message.ExtraData.ActorName + } + return vendor.MessageParams{ + Scene: scene, + 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)) @@ -668,8 +721,8 @@ func (s *pushServiceImpl) doRetry() int { var pushErr error if s.isJPushAvailable() && device.SupportsMobilePush() && device.PushToken != "" { - payload := buildMessagePayload(message) - pushErr = s.pushViaJPush(ctx, device, payload.Title, payload.Content, payload.Extras) + params := s.messageToParams(message) + pushErr = s.pushViaJPush(ctx, device, params) } if pushErr != nil { @@ -760,14 +813,25 @@ func (s *pushServiceImpl) PushChatMessage(ctx context.Context, userID string, co extras["target_type"] = message.ExtraData.TargetType } - notification := jpush.BuildNotification(title, content, "chat_message", extras) // 群聊用群头像，私聊用发送者头像 + largeIcon := sender.Avatar if convType == model.ConversationTypeGroup && groupAvatar != "" { - notification.Android.LargeIcon = groupAvatar - } else if sender.Avatar != "" { - notification.Android.LargeIcon = sender.Avatar + largeIcon = groupAvatar } - if _, pushErr := s.jpushClient.PushByRegistrationIDs(regIDs, notification); pushErr != nil { + 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), @@ -847,10 +911,19 @@ func (s *pushServiceImpl) PushSystemNotification(ctx context.Context, userID str extras["target_type"] = notification.ExtraData.TargetType } notifType := string(notification.Type) - jpushNotif := jpush.BuildNotification(title, content, notifType, extras) + largeIcon := "" if notification.ExtraData != nil && notification.ExtraData.AvatarURL != "" { - jpushNotif.Android.LargeIcon = 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)) @@ -859,7 +932,8 @@ func (s *pushServiceImpl) PushSystemNotification(ctx context.Context, userID str } return ids }(), - jpushNotif, + payload.Notification, + payload.TPC, ) if pushErr == nil { return nil diff --git a/internal/service/upload_service.go b/internal/service/upload_service.go index edd99af..dc3df79 100644 --- a/internal/service/upload_service.go +++ b/internal/service/upload_service.go @@ -18,6 +18,7 @@ import ( "with_you/internal/model" "with_you/internal/pkg/s3" + "with_you/internal/repository" "github.com/disintegration/imaging" "go.uber.org/zap" @@ -26,22 +27,34 @@ import ( _ "golang.org/x/image/tiff" ) +// UploadFileResult 通用文件上传结果 +type UploadFileResult struct { + URL string `json:"url"` + Name string `json:"name"` + Size int64 `json:"size"` + MimeType string `json:"mime_type"` +} + // UploadService 上传服务接口 type UploadService interface { UploadImage(ctx context.Context, file *multipart.FileHeader) (string, string, string, error) UploadImageBytes(ctx context.Context, raw []byte, filename string) (url, previewURL, previewURLLarge string, err error) UploadAvatar(ctx context.Context, userID string, file *multipart.FileHeader) (string, error) UploadCover(ctx context.Context, userID string, file *multipart.FileHeader) (string, error) + UploadFile(ctx context.Context, file *multipart.FileHeader, folder string, uploaderID string) (*UploadFileResult, error) GetURL(ctx context.Context, objectName string) (string, error) Delete(ctx context.Context, objectName string) error GeneratePreviewImages(ctx context.Context, originalData []byte, hash string) (previewURL, previewURLLarge string, err error) ValidateChatMessageImageSegments(segments model.MessageSegments) error + // GetExpiredURLSet 返回所有已从 S3 清理的文件 URL 集合，供 DTO 层标记失效 + GetExpiredURLSet(ctx context.Context) (map[string]struct{}, error) } // uploadService 上传服务实现 type uploadService struct { - s3Client *s3.Client - userService UserService + s3Client *s3.Client + userService UserService + uploadedRepo repository.UploadedFileRepository } // 预览图配置 @@ -52,10 +65,11 @@ const ( ) // NewUploadService 创建上传服务 -func NewUploadService(s3Client *s3.Client, userService UserService) UploadService { +func NewUploadService(s3Client *s3.Client, userService UserService, uploadedRepo repository.UploadedFileRepository) UploadService { return &uploadService{ - s3Client: s3Client, - userService: userService, + s3Client: s3Client, + userService: userService, + uploadedRepo: uploadedRepo, } } @@ -216,6 +230,210 @@ func (s *uploadService) UploadCover(ctx context.Context, userID string, file *mu return url, nil } +// 聊天文件上传限制 +const ( + MaxChatFileSize = 50 << 20 // 50MB + chatFileFolderLimit = 32 // folder 名长度上限，避免路径注入 +) + +// chatFileAllowedMime 允许上传的文件 MIME 白名单（按类型前缀匹配）。 +// 不在白名单内的类型一律拒绝，防止上传可执行脚本等高危文件。 +var chatFileAllowedMime = map[string]bool{ + // 文档 + "application/pdf": true, + "application/msword": true, + "application/vnd.openxmlformats-officedocument.wordprocessingml.document": true, + "application/vnd.ms-excel": true, + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": true, + "application/vnd.ms-powerpoint": true, + "application/vnd.openxmlformats-officedocument.presentationml.presentation": true, + "application/rtf": true, + "text/plain": true, + "text/csv": true, + "text/markdown": true, + "application/json": true, + "application/xml": true, + "text/xml": true, + "application/zip": true, + "application/x-zip-compressed": true, + "application/x-rar-compressed": true, + "application/x-7z-compressed": true, + "application/gzip": true, + "application/x-tar": true, + // 音视频/图片（聊天文件入口） + "audio/mpeg": true, + "audio/mp3": true, + "audio/wav": true, + "audio/x-wav": true, + "audio/aac": true, + "audio/x-m4a": true, + "audio/flac": true, + "video/mp4": true, + "video/quicktime": true, + "video/x-msvideo": true, + "video/webm": true, + "image/jpeg": true, + "image/png": true, + "image/gif": true, + "image/webp": true, + "image/bmp": true, + // 其他常用类型 + "application/octet-stream": true, // 部分客户端（如 Windows）上传无明确类型，需结合扩展名二次校验 +} + +// chatFileExtFallback 当 MIME 为 octet-stream 时，按扩展名推断的真实类型 +var chatFileExtFallback = map[string]string{ + ".pdf": "application/pdf", + ".doc": "application/msword", + ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + ".xls": "application/vnd.ms-excel", + ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + ".ppt": "application/vnd.ms-powerpoint", + ".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation", + ".zip": "application/zip", + ".rar": "application/x-rar-compressed", + ".7z": "application/x-7z-compressed", + ".gz": "application/gzip", + ".tar": "application/x-tar", + ".mp3": "audio/mpeg", + ".mp4": "video/mp4", + ".mov": "video/quicktime", + ".avi": "video/x-msvideo", + ".json": "application/json", + ".csv": "text/csv", + ".md": "text/markdown", + ".txt": "text/plain", +} + +// normalizeChatFileMime 校正文件 MIME 类型： +// 1. 解析 multipart header 的 Content-Type； +// 2. 若为通用 octet-stream，依据文件名扩展名回退到更具体的类型； +// 3. 全部兜底为 application/octet-stream。 +func normalizeChatFileMime(headerType, filename string) string { + baseType, _, err := mime.ParseMediaType(strings.TrimSpace(headerType)) + if err != nil || baseType == "" { + baseType = "application/octet-stream" + } + baseType = strings.ToLower(baseType) + + if baseType == "application/octet-stream" { + ext := strings.ToLower(filepath.Ext(filename)) + if real, ok := chatFileExtFallback[ext]; ok { + return real + } + } + return baseType +} + +// validateChatFile 校验聊天文件大小与 MIME 白名单 +func validateChatFile(size int64, mimeType string) error { + if size <= 0 { + return fmt.Errorf("文件为空") + } + if size > MaxChatFileSize { + return fmt.Errorf("文件大小超过限制（最大 %d MB）", MaxChatFileSize>>20) + } + if !chatFileAllowedMime[mimeType] { + return fmt.Errorf("不支持的文件类型: %s", mimeType) + } + return nil +} + +// UploadFile 上传聊天文件（非图片，按文件类型原样存储到 S3） +// folder 用于隔离不同来源（如 chat/post），仅允许字母数字与下划线。 +// uploaderID 为上传者用户 ID，用于审计。 +func (s *uploadService) UploadFile(ctx context.Context, file *multipart.FileHeader, folder string, uploaderID string) (*UploadFileResult, error) { + if file == nil { + return nil, fmt.Errorf("文件为空") + } + if file.Size > MaxChatFileSize { + return nil, fmt.Errorf("文件大小超过限制（最大 %d MB）", MaxChatFileSize>>20) + } + + // 校验与规范化 folder，避免路径穿越 + folder = strings.Map(func(r rune) rune { + if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || r == '_' { + return r + } + return -1 + }, folder) + if folder == "" || len(folder) > chatFileFolderLimit { + folder = "chat" + } + + // 校验 MIME 类型 + mimeType := normalizeChatFileMime(file.Header.Get("Content-Type"), file.Filename) + if err := validateChatFile(file.Size, mimeType); err != nil { + return nil, err + } + + // 读取文件内容 + f, err := file.Open() + if err != nil { + return nil, fmt.Errorf("打开文件失败: %w", err) + } + defer f.Close() + + data, err := io.ReadAll(f) + if err != nil { + return nil, fmt.Errorf("读取文件失败: %w", err) + } + + if err := validateChatFile(int64(len(data)), mimeType); err != nil { + return nil, err + } + + // 以内容哈希命名，避免重复存储；保留原始扩展名（便于浏览器识别） + hash := sha256.Sum256(data) + hashStr := fmt.Sprintf("%x", hash) + ext := strings.ToLower(filepath.Ext(file.Filename)) + if ext == "" { + // 从 MIME 推断扩展名（mime.ExtensionsByType 在新版本返回 (exts, err)） + if es, _ := mime.ExtensionsByType(mimeType); len(es) > 0 { + ext = es[0] + } else { + ext = ".bin" + } + } + + objectName := fmt.Sprintf("files/%s/%s%s", folder, hashStr, ext) + url, err := s.s3Client.UploadData(ctx, objectName, data, mimeType) + if err != nil { + return nil, fmt.Errorf("上传到 S3 失败: %w", err) + } + + // 写入上传记录，用于 7 天 TTL 过期清理（失败仅记日志，不影响上传） + if s.uploadedRepo != nil { + record := &model.UploadedFile{ + URL: url, + ObjectName: objectName, + Name: file.Filename, + Size: file.Size, + MimeType: mimeType, + Folder: folder, + UploaderID: uploaderID, + } + if rerr := s.uploadedRepo.Create(ctx, record); rerr != nil { + zap.L().Warn("写入文件上传记录失败", zap.String("url", url), zap.Error(rerr)) + } + } + + return &UploadFileResult{ + URL: url, + Name: file.Filename, + Size: file.Size, + MimeType: mimeType, + }, nil +} + +// GetExpiredURLSet 返回所有已从 S3 清理的文件 URL 集合，供 DTO 层标记失效 +func (s *uploadService) GetExpiredURLSet(ctx context.Context) (map[string]struct{}, error) { + if s.uploadedRepo == nil { + return nil, nil + } + return s.uploadedRepo.GetExpiredURLSet(ctx) +} + // GetURL 获取文件URL func (s *uploadService) GetURL(ctx context.Context, objectName string) (string, error) { return s.s3Client.GetURL(ctx, objectName) diff --git a/internal/wire/handler.go b/internal/wire/handler.go index 1bd1ea4..988461c 100644 --- a/internal/wire/handler.go +++ b/internal/wire/handler.go @@ -68,9 +68,10 @@ func ProvideMessageHandler( messageService service.MessageService, userService service.UserService, groupService service.GroupService, + uploadService service.UploadService, publisher ws.MessagePublisher, ) *handler.MessageHandler { - return handler.NewMessageHandler(chatService, messageService, userService, groupService, publisher) + return handler.NewMessageHandler(chatService, messageService, userService, groupService, uploadService, publisher) } // ProvideSystemMessageHandler 提供系统消息处理器 diff --git a/internal/wire/repository.go b/internal/wire/repository.go index 11e77d4..63dd20a 100644 --- a/internal/wire/repository.go +++ b/internal/wire/repository.go @@ -55,6 +55,9 @@ var RepositorySet = wire.NewSet( // 版本日志同步 repository.NewConversationVersionLogRepository, + + // 文件上传记录（聊天文件 TTL 过期清理） + repository.NewUploadedFileRepository, ) // ProvideUserActivityRepository 提供用户活跃数据仓储 diff --git a/internal/wire/service.go b/internal/wire/service.go index 32a3013..c13e945 100644 --- a/internal/wire/service.go +++ b/internal/wire/service.go @@ -60,6 +60,7 @@ var ServiceSet = wire.NewSet( ProvideAdminDashboardService, ProvideQRCodeLoginService, ProvideHotRankWorker, + ProvideFileCleanupWorker, ProvideMaterialService, ProvideCallService, ProvideLiveKitService, @@ -112,8 +113,9 @@ func ProvidePushService( messageRepo repository.MessageRepository, publisher ws.MessagePublisher, jpushClient *jpush.Client, + cfg *config.Config, ) service.PushService { - return service.NewPushService(pushRepo, deviceTokenRepo, messageRepo, publisher, jpushClient) + return service.NewPushService(pushRepo, deviceTokenRepo, messageRepo, publisher, jpushClient, cfg.JPush.Channel) } // ProvideSystemMessageService 提供系统消息服务 @@ -291,8 +293,23 @@ func ProvideGroupService( func ProvideUploadService( s3Client *s3.Client, userService service.UserService, + uploadedRepo repository.UploadedFileRepository, ) service.UploadService { - return service.NewUploadService(s3Client, userService) + return service.NewUploadService(s3Client, userService, uploadedRepo) +} + +// ProvideFileCleanupWorker 提供聊天文件过期清理 worker +func ProvideFileCleanupWorker( + cfg *config.Config, + uploadedRepo repository.UploadedFileRepository, + s3Client *s3.Client, + redisClient *redis.Client, +) *service.FileCleanupWorker { + var rdb *redisPkg.Client + if redisClient != nil { + rdb = redisClient.GetClient() + } + return service.NewFileCleanupWorker(cfg, uploadedRepo, s3Client, rdb) } // ProvideUserActivityService 提供用户活跃统计服务