// Package janitor 后台清理循环(安全审计 M5 / 26.9 过期回收): // 回收过期容量预留、超时未完成的上传会话(含其分片对象)、过期预签名会话 // (direct 模式残留对象一并删除),以及过期/超留存期的分享记录与存储对象。 package janitor import ( "context" "errors" "log" "time" "gorm.io/gorm" "fileshare/internal/model" "fileshare/internal/storage" ) // chunkSessionMaxAge 未完成分片会话的最大保留时长(预留 TTL 为 2h, // 会话保留 24h 以支持断点续传;超时后由本循环清理)。 const chunkSessionMaxAge = 24 * time.Hour // presignGrace 过期预签名会话的宽限时长(到点即删,避免与在途 confirm 竞争)。 const presignGrace = time.Hour // 回收批次上限:单轮每类最多处理 200 条,避免大清理阻塞下一 tick。 const recycleBatch = 200 // Recycler 回收配置(26.9):由 API 层注入(管理端 KV 动态读取)。 type Recycler struct { // Enabled 过期自动回收开关。 Enabled func() bool // RetentionDays 全局最长存储时长(天,0=不限制)。 RetentionDays func() int64 // OnRecycled 回收成功后的回调(审计可选),参数:码、文件名、字节数。 OnRecycled func(code, name string, size int64) } // Start 启动周期清理循环;ctx 取消时退出。 // interval 为兜底默认间隔;recycler 非 nil 时按 RecycleInterval 动态取间隔。 func Start(ctx context.Context, db *gorm.DB, store *storage.Manager, interval time.Duration, recycler *Recycler) { go func() { ticker := time.NewTicker(interval) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: Run(ctx, db, store) if recycler != nil && recycler.Enabled != nil && recycler.Enabled() { RecycleExpired(ctx, db, store, recycler) } } } }() } // Run 执行一轮基础设施清理;单项失败仅记日志,不影响其他项。 func Run(ctx context.Context, db *gorm.DB, store *storage.Manager) { now := time.Now() cleanExpiredReservations(ctx, db, now) cleanExpiredChunkSessions(ctx, db, store, now) cleanExpiredPresignSessions(ctx, db, store, now) } // cleanExpiredReservations 删除全部过期容量预留。 func cleanExpiredReservations(ctx context.Context, db *gorm.DB, now time.Time) { if err := db.WithContext(ctx). Where("expires_at <= ?", now). Delete(&model.StorageReservation{}).Error; err != nil { log.Printf("[janitor] 清理过期容量预留失败: %v", err) } } // engineFor 按归属引擎取回实例;空/未知引擎回落当前引擎(对齐 API 层 storeFor 语义)。 func engineFor(store *storage.Manager, name string) (storage.Storage, error) { if name != "" && storage.ValidEngine(name) { if s, err := store.EngineOf(name); err == nil { return s, nil } } return store.Current(), nil } // cleanExpiredChunkSessions 清理超时未完成的分片会话及其分片对象。 func cleanExpiredChunkSessions(ctx context.Context, db *gorm.DB, store *storage.Manager, now time.Time) { var sessions []model.UploadChunk if err := db.WithContext(ctx). Where("chunk_index = -1 AND created_at < ?", now.Add(-chunkSessionMaxAge)). Limit(200). Find(&sessions).Error; err != nil { log.Printf("[janitor] 查询过期分片会话失败: %v", err) return } for _, s := range sessions { engine, err := engineFor(store, s.Engine) if err == nil && s.SavePath != "" { if err := engine.CleanChunks(ctx, s.UploadID, s.SavePath); err != nil && !errors.Is(err, storage.ErrNotFound) && !errors.Is(err, storage.ErrInvalidPath) { log.Printf("[janitor] 清理分片对象失败 upload_id=%s: %v", s.UploadID, err) } } if err := db.WithContext(ctx). Where("upload_id = ?", s.UploadID). Delete(&model.UploadChunk{}).Error; err != nil { log.Printf("[janitor] 删除过期分片会话失败 upload_id=%s: %v", s.UploadID, err) continue } log.Printf("[janitor] 已清理超时分片会话 upload_id=%s file=%s", s.UploadID, s.FileName) } } // cleanExpiredPresignSessions 清理过期预签名会话;direct 模式残留对象一并删除。 func cleanExpiredPresignSessions(ctx context.Context, db *gorm.DB, store *storage.Manager, now time.Time) { var sessions []model.PresignUploadSession if err := db.WithContext(ctx). Where("expires_at < ?", now.Add(-presignGrace)). Limit(200). Find(&sessions).Error; err != nil { log.Printf("[janitor] 查询过期预签名会话失败: %v", err) return } for _, s := range sessions { if s.Mode == "direct" && s.SavePath != "" { if engine, err := engineFor(store, s.Engine); err == nil { if err := engine.DeleteFile(ctx, s.SavePath); err != nil && !errors.Is(err, storage.ErrNotFound) && !errors.Is(err, storage.ErrInvalidPath) { log.Printf("[janitor] 删除直传残留对象失败 upload_id=%s: %v", s.UploadID, err) } } } if err := db.WithContext(ctx). Where("upload_id = ?", s.UploadID). Delete(&model.PresignUploadSession{}).Error; err != nil { log.Printf("[janitor] 删除过期预签名会话失败 upload_id=%s: %v", s.UploadID, err) continue } log.Printf("[janitor] 已清理过期预签名会话 upload_id=%s mode=%s", s.UploadID, s.Mode) } } // ============ 26.9:过期分享回收 ============ // RecycleExpired 回收过期/超存储时长的分享记录与存储对象: // - 时间过期:expired_count<0 且 expired_at 已过; // - 次数耗尽:expired_count>=0 且 <=0; // - 超留存期:retentionDays>0 且 created_at 早于 now-retentionDays; // - 内容去重开启时同一存储对象可能被多条分享引用,删除前做引用计数 // (按 ContentHash/Engine/UUIDFileName 统计),仅删除最后一个引用。 // // 返回本轮回收的分享数。由 janitor 定时循环与管理端手动触发共用。 func RecycleExpired(ctx context.Context, db *gorm.DB, store *storage.Manager, r *Recycler) int { now := time.Now() q := db.WithContext(ctx).Model(&model.FileCodes{}). Where("(expired_count < 0 AND expired_at IS NOT NULL AND expired_at < ?)"+ " OR (expired_count >= 0 AND expired_count <= 0)", now) if r.RetentionDays != nil && r.RetentionDays() > 0 { cutoff := now.AddDate(0, 0, -int(r.RetentionDays())) q = q.Or("created_at < ?", cutoff) } var ids []int64 if err := q.Limit(recycleBatch).Pluck("id", &ids).Error; err != nil { log.Printf("[recycle] 查询过期分享失败: %v", err) return 0 } if len(ids) == 0 { return 0 } n := 0 for _, id := range ids { var fc model.FileCodes if err := db.WithContext(ctx).First(&fc, id).Error; err != nil { continue } // 复核:Expired 语义(避免查询窗口内被取件续期) if !fc.Expired(now) { if r.RetentionDays == nil || r.RetentionDays() <= 0 || fc.CreatedAt.After(now.AddDate(0, 0, -int(r.RetentionDays()))) { continue } } recycleOne(ctx, db, store, &fc, r) n++ } if n > 0 { log.Printf("[recycle] 本轮回收 %d 条过期分享", n) } return n } // RecycleRecord 回收单条分享(取件惰性回收入口):删除记录与存储对象(带引用计数)。 // 记录不存在时为幂等空操作。 func RecycleRecord(ctx context.Context, db *gorm.DB, store *storage.Manager, fc *model.FileCodes, r *Recycler) { // 存在性复核:可能已被定时循环/其他请求回收 var cur model.FileCodes if err := db.WithContext(ctx).Where("id = ?", fc.ID).First(&cur).Error; err != nil { return } recycleOne(ctx, db, store, &cur, r) } // recycleOne 删除单条分享记录及其存储对象(带去重引用计数)。 func recycleOne(ctx context.Context, db *gorm.DB, store *storage.Manager, fc *model.FileCodes, r *Recycler) { if fc.Text == nil && fc.UUIDFileName != nil { engine, err := engineFor(store, fc.Engine) if err != nil { log.Printf("[recycle] 引擎不可用 code=%s: %v", fc.Code, err) // 引擎不可用也删记录,避免永久堆积;对象留给对账巡检 } else { // 去重引用计数:同 ContentHash+Engine+UUIDFileName 的其他分享还在,则不删对象 if fc.ContentHash != nil && *fc.ContentHash != "" { var cnt int64 _ = db.WithContext(ctx).Model(&model.FileCodes{}). Where("content_hash = ? AND engine = ? AND uuid_file_name = ? AND id <> ?", *fc.ContentHash, fc.Engine, *fc.UUIDFileName, fc.ID). Count(&cnt).Error if cnt == 0 && fc.SavePath() != "" { delFile(ctx, engine, fc.SavePath(), fc.Code) } } else if fc.SavePath() != "" { delFile(ctx, engine, fc.SavePath(), fc.Code) } } } if err := db.WithContext(ctx).Delete(fc).Error; err != nil { log.Printf("[recycle] 删除分享记录失败 code=%s: %v", fc.Code, err) return } if r != nil && r.OnRecycled != nil { r.OnRecycled(fc.Code, fc.Prefix+fc.Suffix, fc.Size) } } // delFile 删除存储对象,NotFound 视为成功(幂等)。 func delFile(ctx context.Context, engine storage.Storage, savePath, code string) { if err := engine.DeleteFile(ctx, savePath); err != nil && !errors.Is(err, storage.ErrNotFound) && !errors.Is(err, storage.ErrInvalidPath) { log.Printf("[recycle] 删除存储对象失败 code=%s path=%s: %v", code, savePath, err) } }