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), ) }