// Package status tracks the live state of the single in-flight encode job. // The watcher processes one file at a time, so one mutex-guarded value is // enough — no per-job table, no concurrency design. package status import ( "bufio" "io" "strconv" "strings" "sync" "time" ) // Phase labels for the current job. const ( PhaseIdle = "idle" PhaseProbing = "probing" PhaseAudio = "audio" PhaseEncoding = "encoding" ) // JobMeta is the fetched metadata for the active job, shaped for display. type JobMeta struct { IsSeries bool `json:"is_series"` Title string `json:"title"` // movie title or episode title Collection string `json:"collection,omitempty"` // show name (series) Season string `json:"season,omitempty"` Episode string `json:"episode,omitempty"` DateReleased string `json:"date_released,omitempty"` // release date / airdate MediaType string `json:"media_type,omitempty"` } // Stream is one source audio or subtitle stream, for display. type Stream struct { Kind string `json:"kind"` // "audio" | "subtitle" Language string `json:"language,omitempty"` Codec string `json:"codec,omitempty"` Channels int `json:"channels,omitempty"` // audio only Title string `json:"title,omitempty"` } // Tracker holds live progress for the active job. Safe for concurrent use: // the encode goroutine writes, HTTP/log readers call Snapshot. type Tracker struct { mu sync.RWMutex file string phase string meta *JobMeta streams []Stream totalSec float64 // source duration; 0 until known outTime float64 // encoded position in seconds fps float64 speed float64 startedAt time.Time paused bool pausedAt time.Time // when the current pause began pausedTotal time.Duration // accumulated paused time this job } // Snapshot is an immutable view of the tracker for readers. type Snapshot struct { File string `json:"file"` Phase string `json:"phase"` Meta *JobMeta `json:"meta"` Streams []Stream `json:"streams"` Percent float64 `json:"percent"` FPS float64 `json:"fps"` Speed float64 `json:"speed"` Paused bool `json:"paused"` ElapsedSec int `json:"elapsed_sec"` ETASec int `json:"eta_sec"` StartedAt time.Time `json:"started_at"` } func New() *Tracker { return &Tracker{phase: PhaseIdle} } // Begin marks the start of a new job, resetting all progress fields. func (t *Tracker) Begin(file string) { t.mu.Lock() defer t.mu.Unlock() t.file = file t.phase = PhaseProbing t.meta = nil t.streams = nil t.totalSec, t.outTime, t.fps, t.speed = 0, 0, 0, 0 t.startedAt = time.Now() t.paused = false t.pausedAt = time.Time{} t.pausedTotal = 0 } // SetPaused records pause/resume transitions so elapsed time excludes the // paused span. Idempotent on repeated same-state calls. func (t *Tracker) SetPaused(p bool) { t.mu.Lock() defer t.mu.Unlock() if p == t.paused { return } if p { t.pausedAt = time.Now() } else if !t.pausedAt.IsZero() { t.pausedTotal += time.Since(t.pausedAt) t.pausedAt = time.Time{} } t.paused = p } func (t *Tracker) SetPhase(p string) { t.mu.Lock() defer t.mu.Unlock() t.phase = p } // SetMeta records the fetched metadata for the active job. func (t *Tracker) SetMeta(m JobMeta) { t.mu.Lock() defer t.mu.Unlock() t.meta = &m } // SetStreams records the source audio/subtitle streams for the active job. func (t *Tracker) SetStreams(s []Stream) { t.mu.Lock() defer t.mu.Unlock() t.streams = s } // SetTotal records the source duration in seconds (from ffprobe). func (t *Tracker) SetTotal(seconds float64) { t.mu.Lock() defer t.mu.Unlock() t.totalSec = seconds } // Update records one ffmpeg -progress sample. func (t *Tracker) Update(outTimeSec, fps, speed float64) { t.mu.Lock() defer t.mu.Unlock() t.outTime, t.fps, t.speed = outTimeSec, fps, speed } // Idle clears the tracker when no job is running. func (t *Tracker) Idle() { t.mu.Lock() defer t.mu.Unlock() t.file = "" t.phase = PhaseIdle t.meta = nil t.streams = nil t.totalSec, t.outTime, t.fps, t.speed = 0, 0, 0, 0 t.startedAt = time.Time{} t.paused = false t.pausedAt = time.Time{} t.pausedTotal = 0 } func (t *Tracker) Snapshot() Snapshot { t.mu.RLock() defer t.mu.RUnlock() s := Snapshot{ File: t.file, Phase: t.phase, Meta: t.meta, Streams: t.streams, FPS: t.fps, Speed: t.speed, Paused: t.paused, StartedAt: t.startedAt, } if !t.startedAt.IsZero() { elapsed := time.Since(t.startedAt) - t.pausedTotal if t.paused && !t.pausedAt.IsZero() { elapsed -= time.Since(t.pausedAt) } if elapsed < 0 { elapsed = 0 } s.ElapsedSec = int(elapsed.Seconds()) } if t.totalSec > 0 { s.Percent = t.outTime / t.totalSec * 100 if s.Percent > 100 { s.Percent = 100 } if t.speed > 0 { remaining := t.totalSec - t.outTime if remaining < 0 { remaining = 0 } s.ETASec = int(remaining / t.speed) } } return s } // ProgressSample is one completed ffmpeg -progress block. type ProgressSample struct { OutTimeSec float64 FPS float64 Speed float64 Done bool // progress=end } // ScanProgress reads ffmpeg `-progress` key=value output from r and calls // onSample once per block (each block is terminated by a "progress=" line). // Returns when r is exhausted. func ScanProgress(r io.Reader, onSample func(ProgressSample)) error { sc := bufio.NewScanner(r) var cur ProgressSample for sc.Scan() { if parseProgressLine(sc.Text(), &cur) { onSample(cur) cur = ProgressSample{} } } return sc.Err() } // parseProgressLine folds one "key=value" line into s. Returns true when the // line closes a block (key == "progress"). // // ponytail: ffmpeg field naming drifts between builds — out_time_us is the // modern microsecond field; out_time_ms is historically ALSO microseconds (a // known mislabel); out_time is the "HH:MM:SS.ffffff" string. Prefer out_time_us, // fall back to the others only if it hasn't set a value this block. Verify // against the ffmpeg in the Docker image if percentages look off. func parseProgressLine(line string, s *ProgressSample) (complete bool) { k, v, ok := strings.Cut(line, "=") if !ok { return false } k, v = strings.TrimSpace(k), strings.TrimSpace(v) switch k { case "out_time_us": if us, err := strconv.ParseFloat(v, 64); err == nil { s.OutTimeSec = us / 1e6 } case "out_time_ms": if s.OutTimeSec == 0 { if ms, err := strconv.ParseFloat(v, 64); err == nil { s.OutTimeSec = ms / 1e6 // really microseconds, see note above } } case "out_time": if s.OutTimeSec == 0 { s.OutTimeSec = parseTimecode(v) } case "fps": if f, err := strconv.ParseFloat(v, 64); err == nil { s.FPS = f } case "speed": if f, err := strconv.ParseFloat(strings.TrimSuffix(v, "x"), 64); err == nil { s.Speed = f // "N/A" leaves it 0 } case "progress": s.Done = v == "end" return true } return false } // parseTimecode parses "HH:MM:SS.ffffff" into seconds; 0 on bad input. func parseTimecode(tc string) float64 { parts := strings.Split(tc, ":") if len(parts) != 3 { return 0 } h, err1 := strconv.ParseFloat(parts[0], 64) m, err2 := strconv.ParseFloat(parts[1], 64) sec, err3 := strconv.ParseFloat(parts[2], 64) if err1 != nil || err2 != nil || err3 != nil { return 0 } return h*3600 + m*60 + sec }