package handler import ( "context" "fmt" "time" "sql-platform/internal/middleware" "sql-platform/internal/model" ) // 保留期清理任务的调参。设计取舍: // - 分批删(每批 5000)而不是一条 DELETE 全删:SQLite 是单写者, // 大事务会把整个库锁住,前端所有请求都会排队等锁。 // - 单表单轮最多删 20 万行:极端情况下(比如首次清理积压半年的审计日志), // 剩下的留给下一轮或手动跑,避免一次任务跑太久。 // - 删完看 freelist 比例决定要不要 VACUUM:SQLite 的 DELETE 只把页标记为空闲, // 文件不会自动变小,必须 VACUUM 才真正归还磁盘。 const ( cleanupBatchSize = 5000 cleanupMaxPerTable = 200000 cleanupInterval = 24 * time.Hour cleanupFirstDelay = 2 * time.Minute vacuumFreeRatio = 0.25 ) // retentionRule 一条「按时间列保留 N 天」的清理规则 type retentionRule struct { Table string TimeCol string DaysKey string // system_settings 里对应的保留天数 key Default int } // 会无限增长的表清单。tickets 不在这里(要按状态过滤,单独处理)。 var retentionRules = []retentionRule{ {"audit_logs", "created_at", "audit_log_retention_days", 365}, {"query_histories", "created_at", "query_history_retention_days", 90}, {"slow_logs", "created_at", "slow_log_retention_days", 180}, {"operation_logs", "created_at", "operation_log_retention_days", 365}, {"workflow_execution_logs", "created_at", "workflow_log_retention_days", 180}, {"scheduled_task_logs", "started_at", "scheduled_log_retention_days", 180}, } // deleteInBatches 分批删除,返回实际删除行数。 // // 用 `DELETE ... WHERE id IN (SELECT id ... LIMIT n)` 而不是 GORM 的 Delete: // GORM 不支持 Limit,一次性删会生成超大事务。 func deleteInBatches(table, where string, args []any, maxRows int64) int64 { var total int64 for total < maxRows { params := make([]any, 0, len(args)+1) params = append(params, args...) params = append(params, cleanupBatchSize) res := model.DB.Exec( fmt.Sprintf("DELETE FROM %s WHERE id IN (SELECT id FROM %s WHERE %s LIMIT ?)", table, table, where), params..., ) if res.Error != nil { middleware.LogJSON(map[string]any{ "ts": time.Now().Format(time.RFC3339Nano), "level": "error", "type": "cleanup", "msg": "删除失败", "table": table, "error": res.Error.Error(), }) break } if res.RowsAffected == 0 { break } total += res.RowsAffected time.Sleep(50 * time.Millisecond) // 让出写锁,避免饿死业务请求 } return total } // RunRetentionCleanup 执行一轮清理,返回 {表名: 删除行数} func RunRetentionCleanup() map[string]int64 { now := time.Now() deleted := map[string]int64{} for _, r := range retentionRules { days := getSettingInt(r.DaysKey, r.Default) if days <= 0 { continue // <=0 视为不清理,避免误配把数据全删 } cutoff := now.AddDate(0, 0, -days) n := deleteInBatches(r.Table, r.TimeCol+" < ?", []any{cutoff}, cleanupMaxPerTable) if n > 0 { deleted[r.Table] = n } } // 工单:只删终态(success/failed/closed/rejected),进行中的不能动 ticketDays := getSettingInt("ticket_retention_days", 180) if ticketDays > 0 { n := deleteInBatches("tickets", "status IN ('success','failed','closed','rejected') AND create_time < ?", []any{now.AddDate(0, 0, -ticketDays)}, cleanupMaxPerTable) if n > 0 { deleted["tickets"] = n } } // 刷新令牌:过期/已吊销的直接清掉(按时间点,不是保留天数) // 不清的话每次登录都会新增一行,永远不减少。 n := deleteInBatches("refresh_tokens", "expires_at < ?", []any{now}, 100000) if n > 0 { deleted["refresh_tokens"] = n } return deleted } // CleanupPreview 返回各表「按当前保留策略会被删掉多少行」 func CleanupPreview() map[string]any { now := time.Now() out := map[string]any{} count := func(table, where string, args ...any) int64 { var n int64 model.DB.Raw(fmt.Sprintf("SELECT COUNT(*) FROM %s WHERE %s", table, where), args...).Scan(&n) return n } for _, r := range retentionRules { days := getSettingInt(r.DaysKey, r.Default) out[r.Table] = map[string]any{ "days": days, "count": count(r.Table, r.TimeCol+" < ?", now.AddDate(0, 0, -days)), } } ticketDays := getSettingInt("ticket_retention_days", 180) out["tickets"] = map[string]any{ "days": ticketDays, "count": count("tickets", "status IN ('success','failed','closed','rejected') AND create_time < ?", now.AddDate(0, 0, -ticketDays)), } out["refresh_tokens"] = map[string]any{ "days": 0, "count": count("refresh_tokens", "expires_at < ?", now), } return out } // VacuumIfNeeded 空闲页占比超过阈值时执行 VACUUM 回收磁盘,返回 (是否执行, 释放字节数) func VacuumIfNeeded() (bool, int64) { var pageCount, freeList, pageSize int64 model.DB.Raw("PRAGMA page_count").Scan(&pageCount) model.DB.Raw("PRAGMA freelist_count").Scan(&freeList) model.DB.Raw("PRAGMA page_size").Scan(&pageSize) if pageCount <= 0 || pageSize <= 0 { return false, 0 } if float64(freeList)/float64(pageCount) < vacuumFreeRatio { return false, 0 } before := pageCount * pageSize if err := model.DB.Exec("VACUUM").Error; err != nil { middleware.LogJSON(map[string]any{ "ts": time.Now().Format(time.RFC3339Nano), "level": "error", "type": "cleanup", "msg": "VACUUM 失败", "error": err.Error(), }) return false, 0 } model.DB.Raw("PRAGMA page_count").Scan(&pageCount) after := pageCount * pageSize return true, before - after } // StartCleanupScheduler 启动后台清理:启动 2 分钟后跑一次,之后每 24h 一次。 // 可通过系统设置 cleanup_auto_enabled=false 关掉(关掉后仍可手动点「立即清理」)。 func StartCleanupScheduler(ctx context.Context) { go func() { select { case <-time.After(cleanupFirstDelay): case <-ctx.Done(): return } runScheduledCleanup() ticker := time.NewTicker(cleanupInterval) defer ticker.Stop() for { select { case <-ticker.C: runScheduledCleanup() case <-ctx.Done(): return } } }() } func runScheduledCleanup() { if getSetting("cleanup_auto_enabled", "true") != "true" { return } start := time.Now() deleted := RunRetentionCleanup() vacuumed, freed := VacuumIfNeeded() var total int64 for _, n := range deleted { total += n } if total == 0 && !vacuumed { return // 没什么可清的,不刷日志 } middleware.LogJSON(map[string]any{ "ts": time.Now().Format(time.RFC3339Nano), "level": "info", "type": "cleanup", "msg": "保留期清理完成", "deleted": deleted, "total": total, "vacuumed": vacuumed, "freed_bytes": freed, "cost_ms": time.Since(start).Milliseconds(), }) }