// 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) } } } }