av1dae/internal/watcher/watcher.go

package watcher

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

	"videnc-vibe/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. video-only mkv 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

// 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
	// 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) *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,
		seen:     make(map[string]fileStat),
		failed:   make(map[string]failureRecord),
	}
}

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) {
	files, err := filepath.Glob(filepath.Join(w.inputDir, "*.mkv"))
	if err != nil {
		fmt.Printf("Error scanning input directory: %v\n", err)
		return
	}

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