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