ffmpeg-gui/internal/task/manager.go

package task

import (
	"context"
	"ffmpeg-gui/internal/ffmpeg"
	"ffmpeg-gui/internal/media"
	"fmt"
	"sync"
	"time"

	"github.com/google/uuid"
)

// Event types for frontend updates.
const (
	EventTaskUpdated  = "task:updated"
	EventTaskProgress = "task:progress"
	EventTaskLog      = "task:log"
)

// EventCallback is called when a task state changes. The frontend will receive these.
type EventCallback func(eventType string, data any)

// Manager manages the task queue.
type Manager struct {
	mu      sync.Mutex
	tasks   []*Task
	exec    *ffmpeg.Executor
	hwAccel string // hardware accel method: "cuda", "d3d11va", "qsv", or ""
	onEvent EventCallback
	running bool
	ctx     context.Context
	cancel  context.CancelFunc
}

// NewManager creates a new task manager.
func NewManager(exec *ffmpeg.Executor) *Manager {
	ctx, cancel := context.WithCancel(context.Background())
	return &Manager{
		exec:   exec,
		ctx:    ctx,
		cancel: cancel,
	}
}

// SetEventCallback sets the function called on task events.
func (m *Manager) SetEventCallback(cb EventCallback) {
	m.onEvent = cb
}

// SetHWAccel sets the preferred hardware acceleration method.
func (m *Manager) SetHWAccel(accel string) {
	m.hwAccel = accel
}

// Add adds a task to the queue and returns its ID.
func (m *Manager) Add(t *Task) string {
	m.mu.Lock()
	defer m.mu.Unlock()

	t.ID = uuid.New().String()[:8]
	t.Status = StatusPending
	t.CreatedAt = time.Now()
	m.tasks = append(m.tasks, t)

	m.emit(EventTaskUpdated, t)
	return t.ID
}

// Start begins processing the queue.
func (m *Manager) Start(taskID string) error {
	m.mu.Lock()
	t := m.find(taskID)
	if t == nil {
		m.mu.Unlock()
		return fmt.Errorf("task %s not found", taskID)
	}
	if t.Status != StatusPending {
		m.mu.Unlock()
		return fmt.Errorf("task %s is not pending", taskID)
	}
	t.Status = StatusRunning
	m.emit(EventTaskUpdated, t)
	m.mu.Unlock()

	go m.runTask(t)
	return nil
}

// StartAll starts all pending tasks sequentially.
func (m *Manager) StartAll() {
	m.mu.Lock()
	defer m.mu.Unlock()

	m.running = true
	go m.processLoop()
}

// Stop stops processing tasks.
func (m *Manager) Stop() {
	m.mu.Lock()
	defer m.mu.Unlock()

	m.running = false
	if m.cancel != nil {
		m.cancel()
	}
}

// Cancel cancels a specific task.
func (m *Manager) Cancel(taskID string) error {
	m.mu.Lock()
	t := m.find(taskID)
	if t == nil {
		m.mu.Unlock()
		return fmt.Errorf("task %s not found", taskID)
	}
	if t.Status != StatusRunning && t.Status != StatusPending {
		m.mu.Unlock()
		return fmt.Errorf("cannot cancel task in status %s", t.Status)
	}

	if t.Cancel != nil {
		t.Cancel()
	}
	now := time.Now()
	t.Status = StatusCanceled
	t.CompletedAt = &now
	m.emit(EventTaskUpdated, t)
	m.mu.Unlock()
	return nil
}

// Remove removes a completed/failed/canceled task.
func (m *Manager) Remove(taskID string) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	for i, t := range m.tasks {
		if t.ID == taskID {
			if t.Status == StatusRunning {
				return fmt.Errorf("cannot remove running task")
			}
			if t.Cancel != nil {
				t.Cancel()
			}
			m.tasks = append(m.tasks[:i], m.tasks[i+1:]...)
			return nil
		}
	}
	return fmt.Errorf("task %s not found", taskID)
}

// List returns all tasks.
func (m *Manager) List() []*Task {
	m.mu.Lock()
	defer m.mu.Unlock()

	result := make([]*Task, len(m.tasks))
	copy(result, m.tasks)
	return result
}

// processLoop loops through pending tasks and runs them one at a time.
func (m *Manager) processLoop() {
	for m.running {
		m.mu.Lock()
		var next *Task
		for _, t := range m.tasks {
			if t.Status == StatusPending {
				next = t
				break
			}
		}
		if next == nil {
			m.running = false
			m.mu.Unlock()
			return
		}
		next.Status = StatusRunning
		m.emit(EventTaskUpdated, next)
		m.mu.Unlock()

		m.runTask(next)
	}
}

// runTask executes a single task.
func (m *Manager) runTask(t *Task) {
	info, err := media.GetInfo(m.exec, t.InputFile)
	if err != nil {
		m.completeTask(t, false, fmt.Sprintf("读取文件信息失败: %v", err))
		return
	}

	args := BuildArgs(t, m.hwAccel)
	t.Args = args

	cancel, progressCh, logCh, errCh := m.exec.Run(args, info.GetDurationSeconds())

	m.mu.Lock()
	t.Cancel = cancel
	m.mu.Unlock()

	done := make(chan struct{})
	go func() {
		for line := range logCh {
			m.mu.Lock()
			t.Logs = append(t.Logs, line)
			m.mu.Unlock()
			m.emit(EventTaskLog, map[string]any{
				"taskId": t.ID,
				"line":   line,
			})
		}
		close(done)
	}()

	var lastEmit time.Time
	for p := range progressCh {
		m.mu.Lock()
		t.Progress = p
		m.mu.Unlock()

		now := time.Now()
		if now.Sub(lastEmit) > 250*time.Millisecond {
			lastEmit = now
			m.emit(EventTaskProgress, map[string]any{
				"taskId":   t.ID,
				"progress": p,
			})
		}
	}

	<-done
	runErr := <-errCh
	if runErr != nil {
		m.completeTask(t, false, fmt.Sprintf("编码失败: %v", runErr))
	} else {
		m.completeTask(t, true, "")
	}
}

func (m *Manager) completeTask(t *Task, success bool, errMsg string) {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Don't overwrite canceled status
	if t.Status == StatusCanceled {
		return
	}

	now := time.Now()
	if success {
		t.Status = StatusDone
		t.Progress.Percent = 100
	} else {
		t.Status = StatusFailed
		t.Error = errMsg
	}
	t.CompletedAt = &now
	m.emit(EventTaskUpdated, t)
}

func (m *Manager) find(taskID string) *Task {
	for _, t := range m.tasks {
		if t.ID == taskID {
			return t
		}
	}
	return nil
}

func (m *Manager) emit(eventType string, data any) {
	if m.onEvent != nil {
		m.onEvent(eventType, data)
	}
}