av1dae/internal/watcher/watcher.go

package watcher

import (
	"context"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"sync"
	"time"

	"av1dae/pkg/types"
)

// failureBackoff is how long we skip a file after processFn returned an error
// for it, provided the file has not been touched since (mtime unchanged). Chosen
// to be much larger than the polling interval so a permanently-broken input
// (e.g. a video-only file producing "no audio streams found") does not spam the
// log every tick, but short enough that an operator who fixes the underlying
// problem by replacing the file sees it picked up promptly on the next stable
// scan.
const failureBackoff = 5 * time.Minute

// inputExtensions lists the source container extensions the watcher will pick
// up. Globbed case-sensitively (matching prior .mkv behavior).
var inputExtensions = []string{
	"mkv", "mp4", "m4v", "mov", "avi",
	"ts", "m2ts", "mts",
	"mpg", "mpeg", "vob",
	"webm", "wmv", "flv",
}

// InputFiles returns the source files currently in dir that the watcher would
// consider, sorted. The status server uses this to report the pending queue, so
// it stays in sync with inputExtensions.
func InputFiles(dir string) []string {
	var files []string
	for _, ext := range inputExtensions {
		matches, _ := filepath.Glob(filepath.Join(dir, "*."+ext))
		files = append(files, matches...)
	}
	sort.Strings(files)
	return files
}

// fileStat is the (mtime, size) pair used to decide whether a file has settled
// between two consecutive ticks.
type fileStat struct {
	mtime time.Time
	size  int64
}

// failureRecord remembers that processFn failed for a given file, so we can
// back off rather than retrying on every tick.
type failureRecord struct {
	failedAt  time.Time
	fileMtime time.Time
}

type Watcher struct {
	inputDir string
	interval time.Duration
	// runMu guards running: the start/hold gate. While held, settled files are
	// tracked (so the queue is reported) but not handed to processFn.
	runMu   sync.RWMutex
	running bool
	// seen tracks the last-observed (mtime, size) for every file currently in
	// the input dir. A file is only handed to processFn once two consecutive
	// ticks agree on both fields (partial-write protection).
	seen map[string]fileStat
	// failed tracks files that recently errored from processFn so we can skip
	// them until either failureBackoff elapses or their mtime changes.
	failed map[string]failureRecord
}

func New(inputDir string, intervalSeconds int, autostart bool) *Watcher {
	interval := time.Duration(intervalSeconds) * time.Second
	if interval < 10*time.Second {
		interval = 10 * time.Second
	}
	if interval > 30*time.Second {
		interval = 30 * time.Second
	}
	return &Watcher{
		inputDir: inputDir,
		interval: interval,
		running:  autostart,
		seen:     make(map[string]fileStat),
		failed:   make(map[string]failureRecord),
	}
}

// SetRunning toggles the start/hold gate.
func (w *Watcher) SetRunning(v bool) {
	w.runMu.Lock()
	w.running = v
	w.runMu.Unlock()
}

// Running reports whether the queue is being processed.
func (w *Watcher) Running() bool {
	w.runMu.RLock()
	defer w.runMu.RUnlock()
	return w.running
}

func (w *Watcher) Start(ctx context.Context, processFn func(context.Context, string) error) {
	ticker := time.NewTicker(w.interval)
	defer ticker.Stop()

	w.scanAndProcess(ctx, processFn)

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			w.scanAndProcess(ctx, processFn)
		}
	}
}

func (w *Watcher) scanAndProcess(ctx context.Context, processFn func(context.Context, string) error) {
	var files []string
	for _, ext := range inputExtensions {
		matches, err := filepath.Glob(filepath.Join(w.inputDir, "*."+ext))
		if err != nil {
			fmt.Printf("Error scanning input directory: %v\n", err)
			return
		}
		files = append(files, matches...)
	}

	now := time.Now()
	nextSeen := make(map[string]fileStat, len(files))
	nextFailed := make(map[string]failureRecord, len(w.failed))

	for _, file := range files {
		info, err := os.Stat(file)
		if err != nil {
			// File vanished between glob and stat, or unreadable. Drop any
			// state for it by not carrying it forward.
			fmt.Printf("Error stating %s: %v\n", file, err)
			continue
		}

		current := fileStat{mtime: info.ModTime(), size: info.Size()}
		nextSeen[file] = current

		// Carry the failure record forward only if the file hasn't been
		// touched since it failed; a changed mtime means the user replaced
		// or modified the file and we should give it another chance.
		if rec, ok := w.failed[file]; ok && rec.fileMtime.Equal(current.mtime) {
			if now.Sub(rec.failedAt) < failureBackoff {
				nextFailed[file] = rec
				continue
			}
			// Backoff expired; clear the record and let the file be
			// re-processed if it is otherwise stable.
		}

		prev, ok := w.seen[file]
		if !ok || prev.mtime != current.mtime || prev.size != current.size {
			// First time we've seen this (mtime, size); wait one more tick
			// to make sure the file isn't still being written.
			continue
		}

		if !w.Running() {
			// Held: the file stays in the queue (already in nextSeen) and is
			// picked up once the user starts processing.
			continue
		}

		if err := processFn(ctx, file); err != nil {
			fmt.Printf("Error processing %s: %v\n", file, err)
			nextFailed[file] = failureRecord{
				failedAt:  now,
				fileMtime: current.mtime,
			}
		}
	}

	w.seen = nextSeen
	w.failed = nextFailed
}

func DetectMediaType(width, height int) types.MediaType {
	pixels := width * height
	if pixels < 600000 {
		return types.MediaTypeDVD
	}
	return types.MediaTypeBluRay
}