package watcher import ( "context" "fmt" "os" "path/filepath" "sort" "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. 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 // 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) { 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 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 }