opsxsql/backend/internal/handler/cleanup.go

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