convert/internal/convert/runner.go

package convert

import (
	"bufio"
	"bytes"
	"context"
	"fmt"
	"io"
	"os"
	"os/exec"
	"strings"
	"sync"

	"convert/internal/media"
)

type Runner struct {
	mu     sync.Mutex
	cancel context.CancelFunc
	stdin  io.WriteCloser
	active bool
}

func NewRunner() *Runner {
	return &Runner{}
}

func (r *Runner) IsRunning() bool {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.active
}

func (r *Runner) Cancel() bool {
	r.mu.Lock()
	defer r.mu.Unlock()
	if !r.active {
		return false
	}
	if r.stdin != nil {
		_, _ = io.WriteString(r.stdin, "q\n")
	}
	if r.cancel != nil {
		r.cancel()
	}
	return true
}

func (r *Runner) Run(
	parentCtx context.Context,
	ffmpegBin string,
	built *BuiltCommand,
	onProgress func(Progress),
) error {
	if ffmpegBin == "" {
		return fmt.Errorf("ffmpeg binary not found")
	}
	if built == nil {
		return fmt.Errorf("built command is nil")
	}
	if err := EnsureUniqueOutputDir(built.OutputPath); err != nil {
		return fmt.Errorf("failed to create output directory: %w", err)
	}

	r.mu.Lock()
	if r.active {
		r.mu.Unlock()
		return fmt.Errorf("conversion already in progress")
	}
	ctx, cancel := context.WithCancel(parentCtx)
	r.cancel = cancel
	r.active = true
	r.mu.Unlock()

	defer func() {
		r.mu.Lock()
		r.active = false
		r.cancel = nil
		r.stdin = nil
		r.mu.Unlock()
		cancel()
	}()

	args := append([]string{"-progress", "pipe:1", "-nostats"}, built.Args...)
	cmd := exec.CommandContext(ctx, ffmpegBin, args...)
	media.HideCmdWindow(cmd)

	stdout, err := cmd.StdoutPipe()
	if err != nil {
		return err
	}
	stdin, err := cmd.StdinPipe()
	if err == nil {
		r.mu.Lock()
		r.stdin = stdin
		r.mu.Unlock()
	}

	var stderrBuf bytes.Buffer
	cmd.Stderr = &limitedBuffer{buf: &stderrBuf, max: 32 * 1024}

	_, statErr := os.Stat(built.OutputPath)
	existedBefore := statErr == nil
	hasNoClobber := false
	for _, a := range built.Args {
		if a == "-n" {
			hasNoClobber = true
			break
		}
	}
	if existedBefore && hasNoClobber {
		return fmt.Errorf("output file already exists: %s", built.OutputPath)
	}

	if err := cmd.Start(); err != nil {
		return fmt.Errorf("failed to start ffmpeg: %w", err)
	}

	parser := newProgressParser(built.ExpectedDuration)
	scanner := bufio.NewScanner(stdout)
	for scanner.Scan() {
		if prog, ok := parser.feed(scanner.Text()); ok && onProgress != nil {
			prog.OutputPath = built.OutputPath
			onProgress(prog)
		}
	}

	waitErr := cmd.Wait()
	if ctx.Err() != nil {
		if !(existedBefore && hasNoClobber) {
			_ = os.Remove(built.OutputPath)
		}
		return fmt.Errorf("conversion cancelled")
	}
	if waitErr != nil {
		if !(existedBefore && hasNoClobber) {
			_ = os.Remove(built.OutputPath)
		}
		errTail := strings.TrimSpace(stderrBuf.String())
		if len(errTail) > 800 {
			errTail = errTail[len(errTail)-800:]
		}
		if errTail != "" {
			return fmt.Errorf("ffmpeg error: %s", errTail)
		}
		return fmt.Errorf("ffmpeg failed: %w", waitErr)
	}

	var outSize int64
	if st, err := os.Stat(built.OutputPath); err == nil {
		outSize = st.Size()
	}
	if onProgress != nil {
		onProgress(Progress{
			Percent:     100,
			OutTimeSec:  built.ExpectedDuration,
			TotalSec:    built.ExpectedDuration,
			TotalSize:   outSize,
			OutputBytes: outSize,
			OutputPath:  built.OutputPath,
			Done:        true,
		})
	}
	return nil
}

type limitedBuffer struct {
	buf *bytes.Buffer
	max int
}

func (l *limitedBuffer) Write(p []byte) (int, error) {
	if l.buf.Len()+len(p) > l.max {
		// keep latest tail
		excess := l.buf.Len() + len(p) - l.max
		if excess >= l.buf.Len() {
			l.buf.Reset()
		} else {
			b := l.buf.Bytes()[excess:]
			l.buf.Reset()
			l.buf.Write(b)
		}
	}
	return l.buf.Write(p)
}