av1dae/internal/server/server.go

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