// Package server exposes a read-only HTTP status endpoint for the daemon: // the live encode progress, the pending input queue, and recent log events. package server import ( _ "embed" "encoding/json" "net/http" "os" "path/filepath" "sync" "time" "av1dae/internal/logger" "av1dae/internal/settings" "av1dae/internal/status" "av1dae/internal/watcher" ) //go:embed index.html var indexHTML []byte //go:embed settings.html var settingsHTML []byte // Controls is the runtime-control surface the UI drives, wired in main from the // watcher (start/hold gate), encoder (pause/resume), and mover (retry). type Controls struct { Running func() bool SetRunning func(bool) Pause func() error Resume func() error RetryFailed func(name string) error } type Server struct { tracker *status.Tracker log *logger.Logger store *settings.Store controls Controls durCache *durationCache inputDir string failedDir string } func New(tracker *status.Tracker, log *logger.Logger, store *settings.Store, controls Controls, probeDuration func(string) (float64, error), inputDir, failedDir string) *Server { return &Server{ tracker: tracker, log: log, store: store, controls: controls, durCache: newDurationCache(probeDuration), inputDir: inputDir, failedDir: failedDir, } } type durEntry struct { mtime time.Time sec float64 } // durationCache memoizes source durations (keyed by path+mtime) so the queue // ETA doesn't re-probe every file on every 1s poll. A miss kicks a background // ffprobe and resolves on a later poll; the file shows as "estimating" until. type durationCache struct { probe func(string) (float64, error) mu sync.Mutex m map[string]durEntry inflight map[string]bool } func newDurationCache(probe func(string) (float64, error)) *durationCache { return &durationCache{probe: probe, m: map[string]durEntry{}, inflight: map[string]bool{}} } func (c *durationCache) Get(path string, mtime time.Time) (float64, bool) { c.mu.Lock() if e, ok := c.m[path]; ok && e.mtime.Equal(mtime) { c.mu.Unlock() return e.sec, true } if c.inflight[path] { c.mu.Unlock() return 0, false } c.inflight[path] = true c.mu.Unlock() go func() { sec, err := c.probe(path) c.mu.Lock() delete(c.inflight, path) if err == nil { c.m[path] = durEntry{mtime, sec} } c.mu.Unlock() }() return 0, false } // Retain drops cached entries for paths no longer present, bounding the map on // a long-running daemon. func (c *durationCache) Retain(paths []string) { keep := make(map[string]bool, len(paths)) for _, p := range paths { keep[p] = true } c.mu.Lock() for p := range c.m { if !keep[p] { delete(c.m, p) } } c.mu.Unlock() } // queueETASeconds estimates wall-clock time to clear the queue: the current // job's remaining time plus each pending file's duration / current speed. // 0 when there's no speed to extrapolate from (idle/held). func queueETASeconds(currentETASec int, sumQueuedSec, speed float64) int { if speed <= 0 { return 0 } return currentETASec + int(sumQueuedSec/speed) } // Handler returns the mux for all status routes. Phase 3 adds "/" (the HTML // dashboard) to this same mux. func (s *Server) Handler() http.Handler { mux := http.NewServeMux() mux.HandleFunc("/status", s.handleStatus) mux.HandleFunc("/settings", s.handleSettingsPage) mux.HandleFunc("/api/settings", s.handleAPISettings) mux.HandleFunc("/api/start", s.gateHandler(true)) mux.HandleFunc("/api/hold", s.gateHandler(false)) mux.HandleFunc("/api/pause", s.actionHandler(func() error { return s.controls.Pause() }, "Encode paused")) mux.HandleFunc("/api/resume", s.actionHandler(func() error { return s.controls.Resume() }, "Encode resumed")) mux.HandleFunc("/api/retry", s.handleRetry) mux.HandleFunc("/", s.handleIndex) return mux } // gateHandler flips the start/hold gate. POST only. func (s *Server) gateHandler(run bool) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { methodNotAllowed(w) return } s.controls.SetRunning(run) if run { s.log.Info("Queue started via web UI") } else { s.log.Info("Queue held via web UI") } w.WriteHeader(http.StatusNoContent) } } // actionHandler wraps a no-arg control action (pause/resume). POST only. func (s *Server) actionHandler(fn func() error, logMsg string) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { methodNotAllowed(w) return } if err := fn(); err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } s.log.Info(logMsg + " via web UI") w.WriteHeader(http.StatusNoContent) } } // handleRetry moves a named file from failed/ back to input/. POST ?file=NAME. func (s *Server) handleRetry(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { methodNotAllowed(w) return } name := r.URL.Query().Get("file") if name == "" { http.Error(w, "missing file", http.StatusBadRequest) return } if err := s.controls.RetryFailed(name); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } s.log.Info("Retry requested via web UI: " + name) w.WriteHeader(http.StatusNoContent) } func methodNotAllowed(w http.ResponseWriter) { w.Header().Set("Allow", "POST") http.Error(w, "method not allowed", http.StatusMethodNotAllowed) } func (s *Server) handleSettingsPage(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/html; charset=utf-8") _, _ = w.Write(settingsHTML) } // handleAPISettings serves the current settings (GET) and saves new ones (PUT). // Validation lives in settings.Store.Set; a bad payload returns 400. func (s *Server) handleAPISettings(w http.ResponseWriter, r *http.Request) { switch r.Method { case http.MethodGet: w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(s.store.Get()) case http.MethodPut: var in settings.Settings if err := json.NewDecoder(r.Body).Decode(&in); err != nil { http.Error(w, "invalid JSON: "+err.Error(), http.StatusBadRequest) return } if err := s.store.Set(in); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } s.log.SetRetention(in.LogRetentionDays) s.log.Info("Settings updated via web UI") w.WriteHeader(http.StatusNoContent) default: w.Header().Set("Allow", "GET, PUT") http.Error(w, "method not allowed", http.StatusMethodNotAllowed) } } func (s *Server) handleIndex(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/" { http.NotFound(w, r) return } w.Header().Set("Content-Type", "text/html; charset=utf-8") _, _ = w.Write(indexHTML) } type statusResponse struct { Running bool `json:"running"` Current status.Snapshot `json:"current"` Queue []string `json:"queue"` Failed []string `json:"failed"` QueueETASec int `json:"queue_eta_sec"` QueueETAPartial bool `json:"queue_eta_partial"` Recent []logger.LogEntry `json:"recent"` } func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) { recent, err := s.log.RecentLogs(50) if err != nil { http.Error(w, "reading logs", http.StatusInternalServerError) return } snap := s.tracker.Snapshot() files := watcher.InputFiles(s.inputDir) // Sum durations of pending files (excluding the current job, whose remaining // time is already in snap.ETASec). Unprobed files mark the estimate partial. var sumQueued float64 partial := false for _, f := range files { if f == snap.File { continue } info, statErr := os.Stat(f) if statErr != nil { partial = true continue } if sec, ok := s.durCache.Get(f, info.ModTime()); ok { sumQueued += sec } else { partial = true } } s.durCache.Retain(files) w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(statusResponse{ Running: s.controls.Running(), Current: snap, Queue: baseNames(files), Failed: baseNames(watcher.InputFiles(s.failedDir)), QueueETASec: queueETASeconds(snap.ETASec, sumQueued, snap.Speed), QueueETAPartial: partial, Recent: recent, }) } func baseNames(paths []string) []string { out := []string{} for _, p := range paths { out = append(out, filepath.Base(p)) } return out }