vdi-cloud/platform/internal/server/task_history.go

package server

import (
	"bytes"
	"database/sql"
	"encoding/csv"
	"fmt"
	"strings"
	"time"
	"unicode/utf8"

	"github.com/gin-gonic/gin"
	"vdi-cloud/platform/internal/model"
	"vdi-cloud/platform/internal/store"
)

// The same filter serves SQL pagination, pending-operation polling and CSV.
func (s *Server) taskFilter(c *gin.Context) (string, []any, bool) {
	where, args := " FROM tasks WHERE true", []any{}
	add := func(expr string, value any) {
		args = append(args, value)
		where += fmt.Sprintf(" AND "+expr, len(args))
	}
	var from, to time.Time
	for _, key := range []string{"from", "to"} {
		if raw := c.Query(key); raw != "" {
			v, err := time.Parse(time.RFC3339Nano, raw)
			if err != nil {
				s.fail(c, 400, "invalid_range", "时间格式无效，请使用含时区的时间")
				return "", nil, false
			}
			if key == "from" {
				from = v
				add("created_at >= $%d", v)
			} else {
				to = v
				add("created_at <= $%d", v)
			}
		}
	}
	if !from.IsZero() && !to.IsZero() && !to.After(from) {
		s.fail(c, 400, "invalid_range", "结束时间必须晚于开始时间")
		return "", nil, false
	}
	if active := c.Query("active"); active != "" {
		if active != "true" && active != "false" {
			s.fail(c, 400, "invalid_request", "active 只能为 true 或 false")
			return "", nil, false
		}
		if active == "true" {
			where += " AND status IN ('queued','running','timeout')"
		}
	}
	for _, key := range []string{"keyword", "actor", "action", "kind", "status", "targetId"} {
		value := c.Query(key)
		if utf8.RuneCountInString(value) > 200 || strings.IndexByte(value, 0) >= 0 {
			s.fail(c, 400, "invalid_request", "筛选条件最长 200 字，且不能包含空字符")
			return "", nil, false
		}
		if value == "" {
			continue
		}
		if key == "status" && value != "queued" && value != "running" && value != "success" && value != "failed" && value != "timeout" {
			s.fail(c, 400, "invalid_request", "任务状态无效")
			return "", nil, false
		}
		if key != "keyword" && key != "actor" {
			column := key
			if key == "targetId" {
				column = "target_id"
			}
			add(column+" = $%d", value)
			continue
		}
		pattern := "%" + strings.NewReplacer(`\`, `\\`, "%", `\%`, "_", `\_`).Replace(value) + "%"
		if key == "actor" {
			add("actor ILIKE $%d", pattern)
		} else {
			args = append(args, pattern)
			p := fmt.Sprintf("$%d", len(args))
			where += " AND (id ILIKE " + p + " OR target_id ILIKE " + p + " OR target_name ILIKE " + p + " OR actor ILIKE " + p + " OR action ILIKE " + p + " OR message ILIKE " + p + ")"
		}
	}
	return where, args, true
}

func (s *Server) tasks(c *gin.Context) {
	where, args, ok := s.taskFilter(c)
	if !ok {
		return
	}
	page, size := pageParams(c)
	tx, err := s.DB.BeginTxx(c.Request.Context(), &sql.TxOptions{Isolation: sql.LevelRepeatableRead, ReadOnly: true})
	if err != nil {
		s.dbError(c, err)
		return
	}
	defer tx.Rollback()
	var total int
	if err = tx.GetContext(c.Request.Context(), &total, "SELECT count(*)"+where, args...); err != nil {
		s.dbError(c, err)
		return
	}
	items := []model.Task{}
	query := "SELECT *" + where + fmt.Sprintf(" ORDER BY created_at DESC,id DESC LIMIT $%d OFFSET $%d", len(args)+1, len(args)+2)
	args = append(args, size, (page-1)*size)
	if err = tx.SelectContext(c.Request.Context(), &items, query, args...); err != nil {
		s.dbError(c, err)
		return
	}
	c.JSON(200, gin.H{"items": items, "total": total, "page": page, "pageSize": size, "requestId": c.GetString("requestId")})
}

func (s *Server) exportTasks(c *gin.Context) {
	where, args, ok := s.taskFilter(c)
	if !ok {
		return
	}
	items := []model.Task{}
	if err := s.DB.SelectContext(c.Request.Context(), &items, "SELECT *"+where+" ORDER BY created_at DESC,id DESC LIMIT 10001", args...); err != nil {
		s.dbError(c, err)
		return
	}
	if len(items) > 10000 {
		s.fail(c, 413, "export_too_large", "匹配任务超过 10000 条，请缩小时间范围或增加筛选条件后导出")
		return
	}
	var content bytes.Buffer
	content.WriteString("\ufeff")
	w := csv.NewWriter(&content)
	w.UseCRLF = true
	_ = w.Write([]string{"提交时间（UTC）", "更新时间（UTC）", "任务标识", "任务类型", "动作", "操作者", "目标标识", "目标名称", "状态", "执行结果"})
	for _, t := range items {
		cells := []string{t.CreatedAt.UTC().Format(time.RFC3339Nano), t.UpdatedAt.UTC().Format(time.RFC3339Nano), t.ID, t.Kind, t.Action, t.Actor, t.TargetID, t.TargetName, t.Status, t.Message}
		for i := range cells {
			cells[i] = safeCSVCell(cells[i])
		}
		_ = w.Write(cells)
	}
	w.Flush()
	if err := w.Error(); err != nil {
		s.fail(c, 500, "export_failed", "导出生成失败，请重试")
		return
	}
	if _, err := s.DB.ExecContext(c.Request.Context(), "INSERT INTO audit_logs(id,actor,action,target,result,request_id) VALUES($1,$2,'export_tasks',$3,'success',$4)", store.ID("audit_"), s.admin(c).Username, fmt.Sprintf("rows=%d", len(items)), c.GetString("requestId")); err != nil {
		s.dbError(c, err)
		return
	}
	c.Header("Content-Disposition", `attachment; filename="tasks-`+time.Now().UTC().Format("20060102-150405")+`.csv"`)
	c.Header("X-Export-Count", fmt.Sprint(len(items)))
	c.Header("X-Content-Type-Options", "nosniff")
	c.Data(200, "text/csv; charset=utf-8", content.Bytes())
}