opsx-vpn/backend/internal/taskmgr/taskmgr.go

// Package taskmgr 提供后台并发任务管理器
package taskmgr

import (
	"fmt"
	"sync"
	"time"
)

// TaskStatus 任务状态
type TaskStatus string

const (
	StatusPending  TaskStatus = "pending"
	StatusRunning  TaskStatus = "running"
	StatusDone     TaskStatus = "done"
	StatusFailed   TaskStatus = "failed"
	StatusCanceled TaskStatus = "canceled"
)

// TaskAction 任务动作
type TaskAction string

const (
	ActionStart  TaskAction = "start"
	ActionStop   TaskAction = "stop"
	ActionDelete TaskAction = "delete"
)

// SubTask 单个子任务（对应一个隧道）
type SubTask struct {
	TunnelID   uint       `json:"tunnel_id"`
	TunnelName string     `json:"tunnel_name"`
	Status     TaskStatus `json:"status"`
	Error      string     `json:"error,omitempty"`
	StartedAt  *time.Time `json:"started_at,omitempty"`
	FinishedAt *time.Time `json:"finished_at,omitempty"`
}

// Task 一个批量任务
type Task struct {
	ID        string     `json:"id"`
	Action    TaskAction `json:"action"`
	Status    TaskStatus `json:"status"`
	Total     int        `json:"total"`
	Success   int        `json:"success"`
	Failed    int        `json:"failed"`
	SubTasks  []SubTask  `json:"sub_tasks"`
	CreatedAt time.Time  `json:"created_at"`
	UpdatedAt time.Time  `json:"updated_at"`
	Error     string     `json:"error,omitempty"`
}

// Progress 返回进度百分比 0-100
func (t *Task) Progress() int {
	if t.Total == 0 {
		return 0
	}
	done := t.Success + t.Failed
	return done * 100 / t.Total
}

// Manager 内存任务管理器
type Manager struct {
	mu    sync.RWMutex
	tasks map[string]*Task
	seq   int
}

func NewManager() *Manager {
	return &Manager{tasks: make(map[string]*Task)}
}

// Create 创建任务并返回 ID
func (m *Manager) Create(action TaskAction, tunnelIDs []uint, names map[uint]string) string {
	m.mu.Lock()
	m.seq++
	id := fmt.Sprintf("task_%d_%d", time.Now().UnixMilli(), m.seq)
	subTasks := make([]SubTask, len(tunnelIDs))
	for i, tid := range tunnelIDs {
		subTasks[i] = SubTask{
			TunnelID:   tid,
			TunnelName: names[tid],
			Status:     StatusPending,
		}
	}
	t := &Task{
		ID:        id,
		Action:    action,
		Status:    StatusPending,
		Total:     len(tunnelIDs),
		SubTasks:  subTasks,
		CreatedAt: time.Now(),
		UpdatedAt: time.Now(),
	}
	m.tasks[id] = t
	m.mu.Unlock()

	// 自动清理 30 分钟前的任务
	go m.cleanup()

	return id
}

// Get 获取任务快照
func (m *Manager) Get(id string) (*Task, bool) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	t, ok := m.tasks[id]
	if !ok {
		return nil, false
	}
	// 返回副本
	cp := *t
	cp.SubTasks = make([]SubTask, len(t.SubTasks))
	copy(cp.SubTasks, t.SubTasks)
	return &cp, true
}

// List 列出最近的任务（最多 50 个）
func (m *Manager) List() []*Task {
	m.mu.RLock()
	defer m.mu.RUnlock()
	out := make([]*Task, 0, len(m.tasks))
	for _, t := range m.tasks {
		cp := *t
		cp.SubTasks = make([]SubTask, len(t.SubTasks))
		copy(cp.SubTasks, t.SubTasks)
		out = append(out, &cp)
	}
	// 按创建时间倒序
	for i := 0; i < len(out)-1; i++ {
		for j := i + 1; j < len(out); j++ {
			if out[j].CreatedAt.After(out[i].CreatedAt) {
				out[i], out[j] = out[j], out[i]
			}
		}
	}
	if len(out) > 50 {
		out = out[:50]
	}
	return out
}

// MarkRunning 标记子任务运行中
func (m *Manager) MarkRunning(taskID string, tunnelID uint) {
	m.mu.Lock()
	defer m.mu.Unlock()
	t, ok := m.tasks[taskID]
	if !ok {
		return
	}
	t.Status = StatusRunning
	now := time.Now()
	t.UpdatedAt = now
	for i := range t.SubTasks {
		if t.SubTasks[i].TunnelID == tunnelID {
			t.SubTasks[i].Status = StatusRunning
			t.SubTasks[i].StartedAt = &now
			break
		}
	}
}

// MarkDone 标记子任务成功
func (m *Manager) MarkDone(taskID string, tunnelID uint) {
	m.mu.Lock()
	defer m.mu.Unlock()
	t, ok := m.tasks[taskID]
	if !ok {
		return
	}
	now := time.Now()
	t.UpdatedAt = now
	for i := range t.SubTasks {
		if t.SubTasks[i].TunnelID == tunnelID {
			t.SubTasks[i].Status = StatusDone
			t.SubTasks[i].FinishedAt = &now
			t.Success++
			break
		}
	}
	t.maybeDone()
}

// MarkFailed 标记子任务失败
func (m *Manager) MarkFailed(taskID string, tunnelID uint, err error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	t, ok := m.tasks[taskID]
	if !ok {
		return
	}
	now := time.Now()
	t.UpdatedAt = now
	for i := range t.SubTasks {
		if t.SubTasks[i].TunnelID == tunnelID {
			t.SubTasks[i].Status = StatusFailed
			t.SubTasks[i].Error = err.Error()
			t.SubTasks[i].FinishedAt = &now
			t.Failed++
			break
		}
	}
	t.maybeDone()
}

// maybeDone 检查任务是否全部完成（调用者需持锁）
func (t *Task) maybeDone() {
	if t.Success+t.Failed >= t.Total {
		if t.Failed > 0 {
			t.Status = StatusDone // 部分失败也算完成
		} else {
			t.Status = StatusDone
		}
		now := time.Now()
		t.UpdatedAt = now
	}
}

func (m *Manager) cleanup() {
	m.mu.Lock()
	defer m.mu.Unlock()
	cutoff := time.Now().Add(-30 * time.Minute)
	for id, t := range m.tasks {
		if t.Status == StatusDone || t.Status == StatusFailed || t.Status == StatusCanceled {
			if t.UpdatedAt.Before(cutoff) {
				delete(m.tasks, id)
			}
		}
	}
}