diff --git a/cmd/av1dae/main.go b/cmd/av1dae/main.go
index 520b8d6..33bb9f3 100644
--- a/cmd/av1dae/main.go
+++ b/cmd/av1dae/main.go
@@ -91,8 +91,29 @@ func main() {
ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer cancel()
+ // Start/hold gate: held by default so the user clicks Start; AV1DAE_AUTOSTART
+ // (truthy) restores start-on-boot.
+ autostart := envTruthy(os.Getenv("AV1DAE_AUTOSTART"))
+ w := watcher.New(cfg.Paths.Input, 15, autostart)
+ if !autostart {
+ log.Info("Queue held on startup — click Start (set AV1DAE_AUTOSTART=1 to auto-start)")
+ }
+
+ controls := server.Controls{
+ Running: w.Running,
+ SetRunning: w.SetRunning,
+ Pause: enc.Pause,
+ Resume: enc.Resume,
+ RetryFailed: func(name string) error {
+ if name == "" || name != filepath.Base(name) {
+ return fmt.Errorf("invalid file name")
+ }
+ return mover.Rename(filepath.Join(cfg.Paths.Failed, name), filepath.Join(cfg.Paths.Input, name))
+ },
+ }
+
if addr := *cfg.HTTPAddr; addr != "" {
- srv := &http.Server{Addr: addr, Handler: server.New(tracker, log, store, cfg.Paths.Input).Handler()}
+ srv := &http.Server{Addr: addr, Handler: server.New(tracker, log, store, controls, cfg.Paths.Input, cfg.Paths.Failed).Handler()}
go func() {
log.Info(fmt.Sprintf("Status server listening on %s", addr))
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
@@ -107,7 +128,6 @@ func main() {
}()
}
- w := watcher.New(cfg.Paths.Input, 15)
w.Start(ctx, func(ctx context.Context, inputPath string) error {
return processFile(ctx, inputPath, cfg, enc, metaClient, log, tracker, store)
})
@@ -269,6 +289,15 @@ func failToFailed(path, failedDir string, log *logger.Logger) {
}
}
+// envTruthy reports whether an env var is set to a truthy value.
+func envTruthy(s string) bool {
+ switch strings.ToLower(strings.TrimSpace(s)) {
+ case "1", "true", "yes", "on":
+ return true
+ }
+ return false
+}
+
// toStatusStreams maps probed source streams to the display shape, keeping only
// audio and subtitle streams (the video stream isn't shown).
func toStatusStreams(streams []encoder.StreamMetadata) []status.Stream {
diff --git a/docker-compose.yml b/docker-compose.yml
index c61e110..0ddd0bd 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -12,7 +12,10 @@ services:
cpus: "8.0"
memory: 4g
ports:
- - "8080:8080" # status server: GET http://host:8080/status (needs http_addr: ":8080" in config)
+ - "8080:8080" # status server + dashboard at http://host:8080 (needs http_addr: ":8080" in config)
+ # By default the queue is HELD on boot — click Start in the UI. Uncomment to auto-start:
+ # environment:
+ # - AV1DAE_AUTOSTART=1
volumes:
- ./data/config.yaml:/config/config.yaml:ro # your config — paths inside must point at /data/*
- ./data/media:/data # holds input/ output/ originals/ failed/ work/ + logs
diff --git a/internal/encoder/encoder.go b/internal/encoder/encoder.go
index 9ce04a7..17d1de4 100644
--- a/internal/encoder/encoder.go
+++ b/internal/encoder/encoder.go
@@ -5,12 +5,15 @@ import (
"context"
"encoding/json"
"fmt"
+ "os"
"os/exec"
"path/filepath"
"regexp"
"sort"
"strconv"
"strings"
+ "sync"
+ "syscall"
"time"
"av1dae/internal/logger"
@@ -26,6 +29,59 @@ type Encoder struct {
opusencPath string
log *logger.Logger
tracker *status.Tracker
+
+ procMu sync.Mutex
+ cur *os.Process // the in-flight video encode, for pause/resume
+}
+
+// Pause freezes the active video encode in place via SIGSTOP. libsvtav1 runs in
+// the ffmpeg process (no forked children), so one signal suspends all its
+// threads. No-op if nothing is encoding. The process keeps its memory and
+// partial output and resumes exactly where it left off.
+func (e *Encoder) Pause() error {
+ e.procMu.Lock()
+ defer e.procMu.Unlock()
+ if e.cur == nil {
+ return nil
+ }
+ if err := e.cur.Signal(syscall.SIGSTOP); err != nil {
+ return err
+ }
+ if e.tracker != nil {
+ e.tracker.SetPaused(true)
+ }
+ return nil
+}
+
+// Resume thaws a paused encode via SIGCONT. No-op if nothing is encoding.
+func (e *Encoder) Resume() error {
+ e.procMu.Lock()
+ defer e.procMu.Unlock()
+ if e.cur == nil {
+ return nil
+ }
+ if err := e.cur.Signal(syscall.SIGCONT); err != nil {
+ return err
+ }
+ if e.tracker != nil {
+ e.tracker.SetPaused(false)
+ }
+ return nil
+}
+
+func (e *Encoder) setProc(p *os.Process) {
+ e.procMu.Lock()
+ e.cur = p
+ e.procMu.Unlock()
+}
+
+func (e *Encoder) clearProc() {
+ e.procMu.Lock()
+ e.cur = nil
+ e.procMu.Unlock()
+ if e.tracker != nil {
+ e.tracker.SetPaused(false)
+ }
}
type StreamInfo struct {
@@ -94,6 +150,9 @@ func (e *Encoder) runCmdProgress(ctx context.Context, label, file, name string,
if err := cmd.Start(); err != nil {
return nil, err
}
+ // Publish the process so Pause/Resume can signal it; clear on exit.
+ e.setProc(cmd.Process)
+ defer e.clearProc()
// Reads stdout to EOF (when ffmpeg exits), so Wait below is safe afterwards.
var lastLog time.Time
diff --git a/internal/server/index.html b/internal/server/index.html
index f688440..302ff6c 100644
--- a/internal/server/index.html
+++ b/internal/server/index.html
@@ -114,6 +114,22 @@
.day:first-child { margin-top:0; }
@media (prefers-reduced-motion:reduce){ .dot.ok{animation:none;} .bar.indet>i{animation:none; width:100%; opacity:.4;} }
+
+ /* control bar */
+ .controls { display:flex; align-items:center; gap:.6rem; flex-wrap:wrap; margin-bottom:1.25rem; min-height:1px; }
+ .ctl { font-family:var(--mono); font-size:.82rem; cursor:pointer; border-radius:7px; padding:.45rem .9rem;
+ background:var(--panel-2); color:var(--text); border:1px solid var(--line); }
+ .ctl:hover { border-color:var(--amber); color:var(--amber); }
+ .ctl.primary { background:var(--amber); color:#1a1200; border-color:var(--amber); font-weight:600; }
+ .ctl.primary:hover { background:var(--amber-soft); color:#1a1200; }
+ .ctl.small { font-size:.72rem; padding:.25rem .65rem; }
+ .ctl-state { font-family:var(--mono); font-size:.72rem; color:var(--muted); margin-left:.3rem; text-transform:uppercase; letter-spacing:.08em; }
+ .ctl-state.paused { color:var(--amber); }
+ .count { font-family:var(--mono); color:var(--red); }
+ /* failed list */
+ #failed .frow { display:flex; align-items:center; gap:.6rem; padding:.35rem 0; border-bottom:1px solid var(--line); font-family:var(--mono); font-size:.82rem; }
+ #failed .frow:last-child { border-bottom:0; }
+ #failed .frow .fn { flex:1; color:var(--text); word-break:break-word; }
@@ -124,10 +140,17 @@
connecting
+
+
+
+
Queue
@@ -356,17 +379,60 @@
txt.textContent = state === "ok" ? "live" : state === "idle" ? "idle" : state === "err" ? "offline" : "connecting";
}
+ // ---- runtime controls: start/hold, pause/resume, retry ----
+ async function post(path) {
+ try { await fetch(path, { method: "POST" }); } catch (e) {}
+ poll(); // reflect the new state immediately
+ }
+
+ function renderControls(d) {
+ const c = d.current || {};
+ const encoding = c.file && c.phase !== "idle";
+ let html = d.running
+ ? ``
+ : ``;
+ if (encoding) {
+ html += c.paused
+ ? ``
+ : ``;
+ }
+ const state = !d.running ? "held" : c.paused ? "running · encode paused" : "running";
+ html += `${state}`;
+ $("controls").innerHTML = html;
+ }
+
+ function renderFailed(failed) {
+ const card = $("failedCard");
+ if (!failed || !failed.length) { card.hidden = true; $("failed").innerHTML = ""; return; }
+ card.hidden = false;
+ $("fcount").textContent = "(" + failed.length + ")";
+ $("failed").innerHTML = failed.map(f =>
+ `${esc(f)}
`
+ ).join("");
+ }
+
+ $("controls").addEventListener("click", e => {
+ const b = e.target.closest("[data-act]");
+ if (b) post(b.getAttribute("data-act"));
+ });
+ $("failed").addEventListener("click", e => {
+ const b = e.target.closest("[data-retry]");
+ if (b) post("/api/retry?file=" + encodeURIComponent(b.getAttribute("data-retry")));
+ });
+
async function poll() {
try {
const r = await fetch("status", { cache: "no-store" });
if (!r.ok) throw new Error(r.status);
const d = await r.json();
pushSpeed(d.current);
+ renderControls(d);
renderJob(d.current);
+ renderFailed(d.failed);
renderQueue(d.queue, d.current);
renderEvents(d.recent);
const active = d.current && d.current.file && d.current.phase !== "idle";
- setLive(active ? "ok" : "idle");
+ setLive(d.current && d.current.paused ? "idle" : active ? "ok" : "idle");
} catch (e) {
setLive("err");
} finally {
diff --git a/internal/server/server.go b/internal/server/server.go
index 30a9963..58db645 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -20,15 +20,27 @@ var indexHTML []byte
//go:embed settings.html
var settingsHTML []byte
-type Server struct {
- tracker *status.Tracker
- log *logger.Logger
- store *settings.Store
- inputDir string
+// 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
}
-func New(tracker *status.Tracker, log *logger.Logger, store *settings.Store, inputDir string) *Server {
- return &Server{tracker: tracker, log: log, store: store, inputDir: inputDir}
+type Server struct {
+ tracker *status.Tracker
+ log *logger.Logger
+ store *settings.Store
+ controls Controls
+ inputDir string
+ failedDir string
+}
+
+func New(tracker *status.Tracker, log *logger.Logger, store *settings.Store, controls Controls, inputDir, failedDir string) *Server {
+ return &Server{tracker: tracker, log: log, store: store, controls: controls, inputDir: inputDir, failedDir: failedDir}
}
// Handler returns the mux for all status routes. Phase 3 adds "/" (the HTML
@@ -38,10 +50,72 @@ func (s *Server) Handler() http.Handler {
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)
@@ -83,17 +157,14 @@ func (s *Server) handleIndex(w http.ResponseWriter, r *http.Request) {
}
type statusResponse struct {
+ Running bool `json:"running"`
Current status.Snapshot `json:"current"`
Queue []string `json:"queue"`
+ Failed []string `json:"failed"`
Recent []logger.LogEntry `json:"recent"`
}
func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) {
- queue := []string{}
- for _, f := range watcher.InputFiles(s.inputDir) {
- queue = append(queue, filepath.Base(f))
- }
-
recent, err := s.log.RecentLogs(50)
if err != nil {
http.Error(w, "reading logs", http.StatusInternalServerError)
@@ -102,8 +173,18 @@ func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(statusResponse{
+ Running: s.controls.Running(),
Current: s.tracker.Snapshot(),
- Queue: queue,
+ Queue: baseNames(watcher.InputFiles(s.inputDir)),
+ Failed: baseNames(watcher.InputFiles(s.failedDir)),
Recent: recent,
})
}
+
+func baseNames(paths []string) []string {
+ out := []string{}
+ for _, p := range paths {
+ out = append(out, filepath.Base(p))
+ }
+ return out
+}
diff --git a/internal/status/status.go b/internal/status/status.go
index fd579eb..10177cf 100644
--- a/internal/status/status.go
+++ b/internal/status/status.go
@@ -43,16 +43,19 @@ type Stream struct {
// 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
+ 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.
@@ -64,6 +67,7 @@ type Snapshot struct {
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"`
@@ -83,6 +87,26 @@ func (t *Tracker) Begin(file string) {
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) {
@@ -129,6 +153,9 @@ func (t *Tracker) Idle() {
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 {
@@ -141,10 +168,18 @@ func (t *Tracker) Snapshot() Snapshot {
Streams: t.streams,
FPS: t.fps,
Speed: t.speed,
+ Paused: t.paused,
StartedAt: t.startedAt,
}
if !t.startedAt.IsZero() {
- s.ElapsedSec = int(time.Since(t.startedAt).Seconds())
+ 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
diff --git a/internal/status/status_test.go b/internal/status/status_test.go
index fbcc4cd..db89dbe 100644
--- a/internal/status/status_test.go
+++ b/internal/status/status_test.go
@@ -81,6 +81,27 @@ func TestSnapshotPercentAndETA(t *testing.T) {
}
}
+func TestSetPausedFlag(t *testing.T) {
+ tr := New()
+ tr.Begin("x.mkv")
+ if tr.Snapshot().Paused {
+ t.Fatal("should not start paused")
+ }
+ tr.SetPaused(true)
+ if !tr.Snapshot().Paused {
+ t.Error("expected paused after SetPaused(true)")
+ }
+ tr.SetPaused(true) // idempotent — must not double-count
+ tr.SetPaused(false)
+ if tr.Snapshot().Paused {
+ t.Error("expected not paused after SetPaused(false)")
+ }
+ tr.Idle()
+ if tr.Snapshot().Paused {
+ t.Error("Idle should clear paused")
+ }
+}
+
func TestSnapshotIdleNoDivByZero(t *testing.T) {
s := New().Snapshot() // no Begin, totalSec 0
if s.Percent != 0 || s.ETASec != 0 || s.Phase != PhaseIdle {
diff --git a/internal/watcher/watcher.go b/internal/watcher/watcher.go
index 472e439..96dd1c8 100644
--- a/internal/watcher/watcher.go
+++ b/internal/watcher/watcher.go
@@ -6,6 +6,7 @@ import (
"os"
"path/filepath"
"sort"
+ "sync"
"time"
"av1dae/pkg/types"
@@ -59,6 +60,10 @@ type failureRecord struct {
type Watcher struct {
inputDir string
interval time.Duration
+ // runMu guards running: the start/hold gate. While held, settled files are
+ // tracked (so the queue is reported) but not handed to processFn.
+ runMu sync.RWMutex
+ running bool
// 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).
@@ -68,7 +73,7 @@ type Watcher struct {
failed map[string]failureRecord
}
-func New(inputDir string, intervalSeconds int) *Watcher {
+func New(inputDir string, intervalSeconds int, autostart bool) *Watcher {
interval := time.Duration(intervalSeconds) * time.Second
if interval < 10*time.Second {
interval = 10 * time.Second
@@ -79,11 +84,26 @@ func New(inputDir string, intervalSeconds int) *Watcher {
return &Watcher{
inputDir: inputDir,
interval: interval,
+ running: autostart,
seen: make(map[string]fileStat),
failed: make(map[string]failureRecord),
}
}
+// SetRunning toggles the start/hold gate.
+func (w *Watcher) SetRunning(v bool) {
+ w.runMu.Lock()
+ w.running = v
+ w.runMu.Unlock()
+}
+
+// Running reports whether the queue is being processed.
+func (w *Watcher) Running() bool {
+ w.runMu.RLock()
+ defer w.runMu.RUnlock()
+ return w.running
+}
+
func (w *Watcher) Start(ctx context.Context, processFn func(context.Context, string) error) {
ticker := time.NewTicker(w.interval)
defer ticker.Stop()
@@ -146,6 +166,12 @@ func (w *Watcher) scanAndProcess(ctx context.Context, processFn func(context.Con
continue
}
+ if !w.Running() {
+ // Held: the file stays in the queue (already in nextSeen) and is
+ // picked up once the user starts processing.
+ continue
+ }
+
if err := processFn(ctx, file); err != nil {
fmt.Printf("Error processing %s: %v\n", file, err)
nextFailed[file] = failureRecord{