From 322aa9eebb90aaa46dee65ef875e004471b04451 Mon Sep 17 00:00:00 2001 From: lafay <2021211506@stu.hit.edu.cn> Date: Thu, 18 Jun 2026 20:34:22 +0800 Subject: [PATCH] feat(worker): add device registration_id cleanup worker Add DeviceCleanupWorker to periodically clean up stale device tokens that haven't been used within the retention period. This prevents accumulation of invalid registration_ids and maintains data hygiene. - New DeviceCleanupConfig with enabled, retention_days (3), and interval_minutes (360) - New repository methods DeleteStaleDevices and DeleteOldestActiveDevice - Updated RegisterDevice to evict oldest active device when reaching limit - Reduced MaxDevicesPerUser from 10 to 3 per user - Worker integrated into app startup/shutdown lifecycle --- .gitignore | 1 + cmd/server/app.go | 13 ++ cmd/server/wire_gen.go | 3 +- configs/config.yaml | 8 + internal/config/config.go | 5 + internal/config/server.go | 8 + internal/repository/device_token_repo.go | 34 ++++ internal/service/device_cleanup_worker.go | 193 ++++++++++++++++++++++ internal/service/push_service.go | 32 +++- internal/wire/service.go | 14 ++ 10 files changed, 303 insertions(+), 8 deletions(-) create mode 100644 internal/service/device_cleanup_worker.go diff --git a/.gitignore b/.gitignore index 7c9be17..46e906a 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ # Build artifacts /server +*.exe *.tar # Runtime files diff --git a/cmd/server/app.go b/cmd/server/app.go index 499229f..7c27638 100644 --- a/cmd/server/app.go +++ b/cmd/server/app.go @@ -24,6 +24,7 @@ type App struct { PushService service.PushService HotRankWorker *service.HotRankWorker FileCleanupWorker *service.FileCleanupWorker + DeviceCleanupWorker *service.DeviceCleanupWorker GRPCServer *runner.Server Server *http.Server Publisher ws.MessagePublisher @@ -40,6 +41,7 @@ func NewApp( pushService service.PushService, hotRankWorker *service.HotRankWorker, fileCleanupWorker *service.FileCleanupWorker, + deviceCleanupWorker *service.DeviceCleanupWorker, grpcServer *runner.Server, publisher ws.MessagePublisher, taskDispatcher runner.TaskDispatcher, @@ -53,6 +55,7 @@ func NewApp( PushService: pushService, HotRankWorker: hotRankWorker, FileCleanupWorker: fileCleanupWorker, + DeviceCleanupWorker: deviceCleanupWorker, GRPCServer: grpcServer, Publisher: publisher, TaskDispatcher: taskDispatcher, @@ -84,6 +87,11 @@ func (a *App) Start(ctx context.Context) error { a.FileCleanupWorker.Start(ctx) } + // 2.1.2 设备 registration_id 清理（清理不活跃设备 token） + if a.DeviceCleanupWorker != nil { + a.DeviceCleanupWorker.Start(ctx) + } + // 2.2 启动 PushWorker（Redis Stream 异步推送） if a.PushWorker != nil { a.PushWorker.Start(ctx) @@ -169,6 +177,11 @@ func (a *App) Stop(ctx context.Context) error { a.FileCleanupWorker.Stop() } + if a.DeviceCleanupWorker != nil { + a.Logger.Info("Stopping device cleanup worker") + a.DeviceCleanupWorker.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 ce9b043..348035f 100644 --- a/cmd/server/wire_gen.go +++ b/cmd/server/wire_gen.go @@ -165,10 +165,11 @@ func InitializeApp() (*App, error) { 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) + deviceCleanupWorker := wire.ProvideDeviceCleanupWorker(config, deviceTokenRepository, 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, fileCleanupWorker, server, messagePublisher, taskDispatcher, logger, pushWorker) + app := NewApp(config, db, router, pushService, hotRankWorker, fileCleanupWorker, deviceCleanupWorker, server, messagePublisher, taskDispatcher, logger, pushWorker) return app, nil } diff --git a/configs/config.yaml b/configs/config.yaml index ded28c0..b84ec25 100644 --- a/configs/config.yaml +++ b/configs/config.yaml @@ -159,6 +159,14 @@ upload: - image/gif - image/webp +# 设备 registration_id 清理配置 +# 周期性清理超过保留期未使用的设备 token，避免无效 registration_id 累积 +# 环境变量: APP_DEVICE_CLEANUP_ENABLED, APP_DEVICE_CLEANUP_RETENTION_DAYS, APP_DEVICE_CLEANUP_INTERVAL_MINUTES +device_cleanup: + enabled: true # 是否启用清理 worker + retention_days: 3 # 保留天数，3 天不活跃即清理（按 last_used_at 判断） + interval_minutes: 360 # 扫描间隔（分钟），默认 6 小时 + # 敏感词过滤配置 sensitive: enabled: true diff --git a/internal/config/config.go b/internal/config/config.go index c093d4f..b8d2994 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -20,6 +20,7 @@ type Config struct { RateLimit RateLimitConfig `mapstructure:"rate_limit"` Upload UploadConfig `mapstructure:"upload"` FileCleanup FileCleanupConfig `mapstructure:"file_cleanup"` + DeviceCleanup DeviceCleanupConfig `mapstructure:"device_cleanup"` OpenAI OpenAIConfig `mapstructure:"openai"` TencentTMS TencentTMSConfig `mapstructure:"tencent_tms"` Email EmailConfig `mapstructure:"email"` @@ -115,6 +116,10 @@ func Load(configPath string) (*Config, error) { viper.SetDefault("file_cleanup.retention_days", 7) viper.SetDefault("file_cleanup.interval_minutes", 360) viper.SetDefault("file_cleanup.batch_size", 100) + // 设备 registration_id 清理默认值（3 天不活跃即清理） + viper.SetDefault("device_cleanup.enabled", true) + viper.SetDefault("device_cleanup.retention_days", 3) + viper.SetDefault("device_cleanup.interval_minutes", 360) viper.SetDefault("s3.endpoint", "") viper.SetDefault("s3.access_key", "") viper.SetDefault("s3.secret_key", "") diff --git a/internal/config/server.go b/internal/config/server.go index 52c3332..2325206 100644 --- a/internal/config/server.go +++ b/internal/config/server.go @@ -69,3 +69,11 @@ type FileCleanupConfig struct { IntervalMinutes int `mapstructure:"interval_minutes"` // 扫描间隔（分钟），默认 360（6 小时） BatchSize int `mapstructure:"batch_size"` // 单次扫描处理上限，默认 100 } + +// DeviceCleanupConfig 设备 registration_id 清理配置 +// 周期性清理超过保留期未使用的设备 token，避免无效 registration_id 累积 +type DeviceCleanupConfig struct { + Enabled bool `mapstructure:"enabled"` // 是否启用清理 worker，默认 true + RetentionDays int `mapstructure:"retention_days"` // 保留天数，默认 3（3 天不活跃即清理） + IntervalMinutes int `mapstructure:"interval_minutes"` // 扫描间隔（分钟），默认 360（6 小时） +} diff --git a/internal/repository/device_token_repo.go b/internal/repository/device_token_repo.go index 47e084e..16d52f9 100644 --- a/internal/repository/device_token_repo.go +++ b/internal/repository/device_token_repo.go @@ -31,6 +31,12 @@ type DeviceTokenRepository interface { GetDeviceCountByUserID(userID int64) (int64, error) GetActiveDeviceCountByUserID(userID int64) (int64, error) DeleteInactiveDevices(before time.Time) error + // DeleteStaleDevices 删除超过指定时间未使用的设备（按 last_used_at 判断，不限 is_active） + // 用于周期性清理长期不活跃的 registration_id，last_used_at 为 NULL 的设备不清理 + DeleteStaleDevices(before time.Time) error + // TrimActiveDevices 将用户活跃设备数量收敛到 keep 以内，删除超额的最旧设备 + // 返回实际删除的记录数。用于设备数限制（如限制从 10 改为 3 后的存量收敛） + TrimActiveDevices(userID string, keep int) (int64, error) } // deviceTokenRepository 设备Token仓储实现 @@ -200,3 +206,31 @@ func (r *deviceTokenRepository) DeleteInactiveDevices(before time.Time) error { return r.db.Where("is_active = ? AND last_used_at < ?", false, before). Delete(&model.DeviceToken{}).Error } + +// DeleteStaleDevices 删除超过指定时间未使用的设备（按 last_used_at 判断，不限 is_active） +// last_used_at 为 NULL 的设备不清理（可能是刚注册尚未使用） +func (r *deviceTokenRepository) DeleteStaleDevices(before time.Time) error { + return r.db.Where("last_used_at IS NOT NULL AND last_used_at < ?", before). + Delete(&model.DeviceToken{}).Error +} + +// TrimActiveDevices 将用户活跃设备数量收敛到 keep 以内，删除超额的最旧设备 +// 按 last_used_at 降序保留最新的 keep 条活跃记录，删除其余活跃记录。 +// keep <= 0 时不删除任何记录。返回实际删除的记录数。 +// 用于设备数限制（新注册超限淘汰 + 历史存量超额收敛）。 +func (r *deviceTokenRepository) TrimActiveDevices(userID string, keep int) (int64, error) { + if keep <= 0 { + return 0, nil + } + // 子查询：取该用户最新的 keep 条活跃记录 ID + subQuery := r.db.Model(&model.DeviceToken{}). + Select("id"). + Where("user_id = ? AND is_active = ?", userID, true). + Order("last_used_at DESC NULLS LAST"). + Limit(keep) + + // 删除不在保留集合内的活跃记录 + result := r.db.Where("user_id = ? AND is_active = ? AND id NOT IN (?)", userID, true, subQuery). + Delete(&model.DeviceToken{}) + return result.RowsAffected, result.Error +} diff --git a/internal/service/device_cleanup_worker.go b/internal/service/device_cleanup_worker.go new file mode 100644 index 0000000..119c9a1 --- /dev/null +++ b/internal/service/device_cleanup_worker.go @@ -0,0 +1,193 @@ +package service + +import ( + "context" + "sync" + "time" + + "with_you/internal/config" + "with_you/internal/repository" + + redislib "github.com/redis/go-redis/v9" + "go.uber.org/zap" +) + +// DeviceCleanupWorker 设备 registration_id 清理 worker +// 周期性清理超过保留期未使用的设备 token（按 last_used_at 判断）。 +// 多实例部署时通过 Redis 分布式锁保证仅主实例执行清理。 +type DeviceCleanupWorker struct { + cfg *config.Config + deviceRepo repository.DeviceTokenRepository + rdb *redislib.Client // 用于分布式锁（可为 nil，降级为本地运行） + stopOnce sync.Once + stopCh chan struct{} + doneCh chan struct{} +} + +const ( + // deviceCleanupLockKey Redis 分布式锁键 + deviceCleanupLockKey = "device_cleanup:leader_lock" + // deviceCleanupLockTTL 锁持有时间（略大于扫描间隔，确保单实例持有） + deviceCleanupLockTTL = 7 * time.Hour + // deviceCleanupLockRetryInterval 竞争锁的重试间隔 + deviceCleanupLockRetryInterval = 1 * time.Minute +) + +// NewDeviceCleanupWorker 创建设备清理 worker +func NewDeviceCleanupWorker( + cfg *config.Config, + deviceRepo repository.DeviceTokenRepository, + rdb *redislib.Client, +) *DeviceCleanupWorker { + return &DeviceCleanupWorker{ + cfg: cfg, + deviceRepo: deviceRepo, + rdb: rdb, + stopCh: make(chan struct{}), + doneCh: make(chan struct{}), + } +} + +// tryAcquireLock 尝试获取分布式锁，返回是否成功 +func (w *DeviceCleanupWorker) tryAcquireLock(ctx context.Context) bool { + if w.rdb == nil { + return true // 无 Redis 时降级为本地运行 + } + ok, err := w.rdb.SetNX(ctx, deviceCleanupLockKey, "1", deviceCleanupLockTTL).Result() + if err != nil { + zap.L().Warn("device cleanup: failed to acquire leader lock, running locally", zap.Error(err)) + return true + } + return ok +} + +// keepLockAlive 周期性续期分布式锁 +func (w *DeviceCleanupWorker) keepLockAlive(ctx context.Context) { + if w.rdb == nil { + return + } + ticker := time.NewTicker(deviceCleanupLockTTL / 2) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-w.stopCh: + return + case <-ticker.C: + w.rdb.Expire(ctx, deviceCleanupLockKey, deviceCleanupLockTTL) + } + } +} + +// releaseLock 释放分布式锁 +func (w *DeviceCleanupWorker) releaseLock() { + if w.rdb == nil { + return + } + w.rdb.Del(context.Background(), deviceCleanupLockKey) +} + +// Start 启动清理 worker +func (w *DeviceCleanupWorker) Start(ctx context.Context) { + if w == nil || w.cfg == nil || !w.cfg.DeviceCleanup.Enabled { + close(w.doneCh) + return + } + if w.deviceRepo == nil { + zap.L().Warn("device cleanup: missing dependencies, worker disabled") + close(w.doneCh) + return + } + + // 扫描间隔，默认 6 小时，最小 5 分钟 + interval := time.Duration(w.cfg.DeviceCleanup.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("device 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("device cleanup: another instance holds leader lock, retrying later") + } + + select { + case <-w.stopCh: + lockCancel() + return + case <-time.After(deviceCleanupLockRetryInterval): + } + } + }() +} + +// Stop 停止 worker +func (w *DeviceCleanupWorker) Stop() { + if w == nil { + return + } + w.stopOnce.Do(func() { + close(w.stopCh) + }) + <-w.doneCh + w.releaseLock() +} + +// runOnce 执行一次清理扫描 +func (w *DeviceCleanupWorker) runOnce(ctx context.Context) { + cfg := w.cfg.DeviceCleanup + retentionDays := cfg.RetentionDays + if retentionDays <= 0 { + retentionDays = 3 + } + + before := time.Now().Add(-time.Duration(retentionDays) * 24 * time.Hour) + if err := w.deviceRepo.DeleteStaleDevices(before); err != nil { + zap.L().Warn("device cleanup: delete stale devices failed", zap.Error(err)) + return + } + + zap.L().Info("device cleanup sweep done", + zap.Int("retention_days", retentionDays), + zap.Time("before", before), + ) +} diff --git a/internal/service/push_service.go b/internal/service/push_service.go index 2c93851..86b89bd 100644 --- a/internal/service/push_service.go +++ b/internal/service/push_service.go @@ -28,7 +28,7 @@ const ( // PushQueueSize 推送队列大小 PushQueueSize = 1000 // MaxDevicesPerUser 每个用户最大设备数 - MaxDevicesPerUser = 10 + MaxDevicesPerUser = 3 ) // ChatMessageSender 聊天消息发送者信息 @@ -386,12 +386,9 @@ func (s *pushServiceImpl) pushViaJPushBatch(devices []*model.DeviceToken, params // 处理同设备多账号场景：registration_id（pushToken）是设备级标识， // 同一物理设备登录新账号时 pushToken 不变，需先失效其他用户对该 pushToken 的关联， // 避免给一个账号推送时同设备的其他账号也收到通知。 +// 每个用户最多保留 MaxDevicesPerUser 个活跃设备：注册后统一收敛到限额以内， +// 既能处理达上限淘汰，也能收敛历史超额数据（如限额从 10 调整为 3 后的存量清理）。 func (s *pushServiceImpl) RegisterDevice(ctx context.Context, userID string, deviceID string, deviceType model.DeviceType, pushToken string) error { - existing, err := s.deviceRepo.GetByUserID(userID) - if err == nil && len(existing) >= MaxDevicesPerUser { - return fmt.Errorf("exceeded maximum device limit (%d)", MaxDevicesPerUser) - } - // 失效其他用户/设备对同一 pushToken 的关联（防止跨账号误推） if pushToken != "" { if dErr := s.deviceRepo.DeactivateByPushToken(pushToken); dErr != nil { @@ -413,7 +410,28 @@ func (s *pushServiceImpl) RegisterDevice(ctx context.Context, userID string, dev } deviceToken.UpdateLastUsed() - return s.deviceRepo.Upsert(deviceToken) + if err := s.deviceRepo.Upsert(deviceToken); err != nil { + return fmt.Errorf("failed to upsert device: %w", err) + } + + // 收敛活跃设备数到 MaxDevicesPerUser 以内（保留最近使用的，淘汰超额的最旧设备） + // 覆盖两种场景：① 新设备注册导致超限；② 历史存量超额（限额调整后逐步收敛） + if evicted, err := s.deviceRepo.TrimActiveDevices(userID, MaxDevicesPerUser); err != nil { + zap.L().Warn("trim active devices failed", + zap.String("userID", userID), + zap.Int("limit", MaxDevicesPerUser), + zap.Error(err), + ) + } else if evicted > 0 { + zap.L().Info("evicted oldest devices for device limit", + zap.String("userID", userID), + zap.String("deviceID", deviceID), + zap.Int64("evicted", evicted), + zap.Int("limit", MaxDevicesPerUser), + ) + } + + return nil } // UnregisterDevice 注销设备 diff --git a/internal/wire/service.go b/internal/wire/service.go index c13e945..9bca189 100644 --- a/internal/wire/service.go +++ b/internal/wire/service.go @@ -61,6 +61,7 @@ var ServiceSet = wire.NewSet( ProvideQRCodeLoginService, ProvideHotRankWorker, ProvideFileCleanupWorker, + ProvideDeviceCleanupWorker, ProvideMaterialService, ProvideCallService, ProvideLiveKitService, @@ -312,6 +313,19 @@ func ProvideFileCleanupWorker( return service.NewFileCleanupWorker(cfg, uploadedRepo, s3Client, rdb) } +// ProvideDeviceCleanupWorker 提供设备 registration_id 清理 worker +func ProvideDeviceCleanupWorker( + cfg *config.Config, + deviceRepo repository.DeviceTokenRepository, + redisClient *redis.Client, +) *service.DeviceCleanupWorker { + var rdb *redisPkg.Client + if redisClient != nil { + rdb = redisClient.GetClient() + } + return service.NewDeviceCleanupWorker(cfg, deviceRepo, rdb) +} + // ProvideUserActivityService 提供用户活跃统计服务 func ProvideUserActivityService(repo repository.UserActivityRepository) service.UserActivityService { return service.NewUserActivityService(repo)