ffmpeg-gui/internal/task/manager.go

package task

import (
	"context"
	"ffmpeg-gui/internal/ffmpeg"
	"ffmpeg-gui/internal/media"
	"fmt"
	"os"
	"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
	}

	// For burn tasks: extract internal subtitles to temp files first.
	// The subtitles filter (libass) hangs on Windows when re-opening the
	// input via si= — extracting avoids the file re-open entirely.
	var tempFiles []string
	if t.Type == TypeBurn {
		for i, sub := range t.Subtitle.Subtitles {
			if sub.Source == "internal" {
				tmpPath := t.OutputFile + fmt.Sprintf(".sub_%d_tmp.srt", i)
				if _, err := m.exec.RunSync("-y", "-i", t.InputFile,
					"-map", fmt.Sprintf("0:%d", sub.Index), "-c:s", "srt", tmpPath); err != nil {
					m.completeTask(t, false, fmt.Sprintf("提取字幕轨道 #%d 失败: %v", sub.Index, err))
					return
				}
				t.Subtitle.Subtitles[i].Source = "external"
				t.Subtitle.Subtitles[i].FilePath = tmpPath
				tempFiles = append(tempFiles, tmpPath)
			}
		}
	}

	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

	// Clean up temp subtitle files
	for _, f := range tempFiles {
		os.Remove(f)
	}

	if runErr != nil {
		m.completeTask(t, false, fmt.Sprintf("%s失败: %v", taskTypeLabel(t.Type), 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)
	// Emit final progress so frontend bar reaches 100%
	m.emit(EventTaskProgress, map[string]any{
		"taskId":   t.ID,
		"progress": t.Progress,
	})
}

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

func taskTypeLabel(typ Type) string {
	switch typ {
	case TypeRemux:
		return "封装"
	case TypeEncode:
		return "转码"
	case TypeBurn:
		return "字幕烧录"
	default:
		return "任务"
	}
}