Serve transcoded streams over HLS

Third step of ADR-008. Two routes inside the item family: the playlist starts
the session lazily and serves what ffmpeg has written so far, so playback can
begin while the rest still encodes; the segment route serves one .ts.

Segment names are validated against the exact shape ffmpeg writes before being
used as a path. Verified red-green: removing the check serves a file from
outside the session directory with a 200.

Two corrections made in review.

1. The manager was being built inside NewHandlers, which meant nothing could
   ever shut it down. Its reaper would run for the process lifetime and a
   running ffmpeg would outlive the server that started it. It is now created
   in main.go beside the job queue and deferred, and passed in - the same
   ownership the queue already has.

2. A playlist request could 404 before ffmpeg wrote the file, which a native
   HLS player treats as fatal rather than retrying. The playlist handler now
   waits up to four seconds for the first write. Segments are never waited for.

Verified end-to-end against a running server with real hardware: a 30-second
file produced a valid EVENT playlist in about a second (the wait doing its
job - no 404), a 658 KB segment with Content-Type video/mp2t, a 404 for a
traversal attempt, and a 401 unauthenticated.

Also confirmed at boot: all four metadata providers report their disabled
state and name their env var, including the new adult one.

cmd/vellum/main.go

@@ -35,6 +35,7 @@ import (
 	"github.com/mithral/vellum/internal/scanner"
 	"github.com/mithral/vellum/internal/search"
 	"github.com/mithral/vellum/internal/subsonic"
+	"github.com/mithral/vellum/internal/transcode"
 	"github.com/mithral/vellum/migrations"
 )
 
@@ -108,7 +109,13 @@ func run(cfg core.Config, log *slog.Logger) error {
 	queue := jobs.New(db, log)
 
 	library.NewHandlers(db, log, authenticator, queue).Register(api.Mux())
-	media.NewHandlers(db, log, authenticator, cfg.ConfigDir).Register(api.Mux())
+	// Transcode sessions. Hardware-gated per ADR-008, so a host with no usable
+	// encoder answers "unavailable" on these routes rather than pretending.
+	// Owned here rather than inside the media handlers so the deferred Shutdown
+	// stops a running ffmpeg instead of leaving it behind the server.
+	sessions := transcode.New()
+	defer sessions.Shutdown()
+	media.NewHandlers(db, log, authenticator, cfg.ConfigDir, sessions).Register(api.Mux())
 	// The landing screen is an aggregate over items, not an item resource, so it
 	// is mounted by its own package.
 	home.NewHandlers(db, log, authenticator).Register(api.Mux())

internal/media/fixtures_test.go

@@ -23,7 +23,10 @@ import (
 func mediaHandlers(t *testing.T, db *sql.DB) *Handlers {
 	t.Helper()
 	log := slog.New(slog.NewTextHandler(io.Discard, nil))
-	return NewHandlers(db, log, nil, t.TempDir())
+	// nil sessions: the constructor falls back to a fresh manager, which is what
+	// a test wants. Tests that exercise the transcode routes inject their own
+	// fake by replacing the field.
+	return NewHandlers(db, log, nil, t.TempDir(), nil)
 }
 
 func testDB(t *testing.T) *sql.DB {

internal/media/handlers.go

@@ -24,12 +24,23 @@ type Handlers struct {
 	log      *slog.Logger
 	auth     *auth.Authenticator
 	cacheDir string
+	sessions Sessions
 }
 
-// NewHandlers builds the media HTTP layer. cacheDir is where covers and
+// NewHandlers builds the media HTTP layer. cacheDir is where covers, extracted
-// extracted pages are written.
+// pages and transcode sessions are written.
-func NewHandlers(db *sql.DB, log *slog.Logger, a *auth.Authenticator, cacheDir string) *Handlers {
+//
-	return &Handlers{db: db, log: log, auth: a, cacheDir: cacheDir}
+// sessions is the transcode manager, owned by the caller so it can be shut
+// down. It is a parameter rather than being built here for exactly that reason:
+// a manager built in this constructor has an idle reaper running for the
+// process lifetime with nobody able to stop it, and a running ffmpeg would
+// outlive the server that started it. A nil value falls back to a fresh
+// manager, which is what a test wants.
+func NewHandlers(db *sql.DB, log *slog.Logger, a *auth.Authenticator, cacheDir string, sessions Sessions) *Handlers {
+	if sessions == nil {
+		sessions = newSessions()
+	}
+	return &Handlers{db: db, log: log, auth: a, cacheDir: cacheDir, sessions: sessions}
 }
 
 // Register mounts the routes.
@@ -49,6 +60,8 @@ func (h *Handlers) Register(mux *http.ServeMux) {
 	mux.Handle("GET /api/v1/items/{id}/cover", h.auth.Require(h.handleCover))
 	mux.Handle("GET /api/v1/items/{id}/subtitles", h.auth.Require(h.handleSubtitles))
 	mux.Handle("GET /api/v1/items/{id}/subtitles/{trackID}", h.auth.Require(h.handleSubtitleTrack))
+	mux.Handle("GET /api/v1/items/{id}/transcode/playlist.m3u8", h.auth.Require(h.handleTranscodePlaylist))
+	mux.Handle("GET /api/v1/items/{id}/transcode/{segment}", h.auth.Require(h.handleTranscodeSegment))
 	mux.Handle("GET /api/v1/items/{id}/progress", h.auth.Require(h.handleGetProgress))
 	mux.Handle("PUT /api/v1/items/{id}/progress", h.auth.Require(h.handlePutProgress))
 	mux.Handle("GET /api/v1/progress/recent", h.auth.Require(h.handleRecent))

internal/media/transcodes.go

@@ -0,0 +1,162 @@
+package media
+
+// HLS transcoding for an item: the playlist and its segments.
+//
+// The item is loaded through the catalog chokepoint exactly as handleCover does
+// (ADR-005), so a stream on an unreadable item answers with the same 404 an
+// absent item does.
+//
+// The playlist request starts the encoder lazily (Manager.Start), then serves
+// whatever ffmpeg has written so far, so a player can begin while the rest is
+// still encoding. Every request goes through Start, which touches the session
+// and keeps the idle reaper from stopping it in the middle of a watch.
+
+import (
+	"errors"
+	"net/http"
+	"os"
+	"path/filepath"
+	"regexp"
+	"time"
+
+	"github.com/mithral/vellum/internal/auth"
+	"github.com/mithral/vellum/internal/httpx"
+	"github.com/mithral/vellum/internal/transcode"
+)
+
+// Sessions starts (or reuses) the HLS transcode session for one item.
+//
+// *transcode.Manager is the production implementation. It is an interface so a
+// test can inject a fake and run these routes with no hardware and no ffmpeg.
+type Sessions interface {
+	Start(itemID int64, inputPath, cacheDir string) (transcode.Session, error)
+}
+
+// segmentName is the exact shape ffmpeg's -hls_segment_filename writes:
+// "seg" then five digits then ".ts". Anything else is refused, so a request can
+// never name a file outside the session directory.
+var segmentName = regexp.MustCompile(`^seg[0-9]{5}\.ts$`)
+
+// newSessions is the production manager, constructed at the wiring site when
+// none is injected.
+func newSessions() Sessions { return transcode.New() }
+
+// handleTranscodePlaylist starts the item's session if it is not already
+// running, then serves the HLS playlist as it stands.
+func (h *Handlers) handleTranscodePlaylist(w http.ResponseWriter, r *http.Request, u auth.User) {
+	it, ok := h.authorized(w, r, u)
+	if !ok {
+		return
+	}
+	s, ok := h.startSession(w, r, it)
+	if !ok {
+		return
+	}
+	h.serveTranscodeFile(w, r, s.PlaylistPath(), "application/vnd.apple.mpegurl")
+}
+
+// handleTranscodeSegment serves one .ts segment of the item's stream.
+func (h *Handlers) handleTranscodeSegment(w http.ResponseWriter, r *http.Request, u auth.User) {
+	it, ok := h.authorized(w, r, u)
+	if !ok {
+		return
+	}
+
+	name := r.PathValue("segment")
+	if !segmentName.MatchString(name) {
+		httpx.WriteError(w, http.StatusNotFound, "segment not found")
+		return
+	}
+
+	s, ok := h.startSession(w, r, it)
+	if !ok {
+		return
+	}
+	h.serveTranscodeFile(w, r, filepath.Join(s.Dir(), name), "video/mp2t")
+}
+
+// startSession starts or reuses the item's transcode session, translating the
+// manager's refusals into a status. On failure it has already written the
+// response.
+//
+// Reusing a live session is also the "touch": Start refreshes its idle timer,
+// so a request for a segment keeps the encoder alive while it is being watched.
+func (h *Handlers) startSession(w http.ResponseWriter, r *http.Request, it Item) (transcode.Session, bool) {
+	s, err := h.sessions.Start(it.ID, it.AbsPath(), h.cacheDir)
+	switch {
+	case errors.Is(err, transcode.ErrUnavailable):
+		httpx.WriteError(w, http.StatusServiceUnavailable, "transcoding is not available on this server")
+		return nil, false
+	case errors.Is(err, transcode.ErrAtCapacity):
+		httpx.WriteError(w, http.StatusServiceUnavailable, "transcode capacity reached, try again shortly")
+		return nil, false
+	case errors.Is(err, transcode.ErrShutdown):
+		httpx.WriteError(w, http.StatusServiceUnavailable, "transcoding is shutting down")
+		return nil, false
+	case err != nil:
+		h.fail(w, "start transcode", err)
+		return nil, false
+	}
+	return s, true
+}
+
+// serveTranscodeFile serves a file the encoder may still be appending to. It
+// sizes the response from the file as it is now — not from a finished transcode
+// — so a player gets the partial playlist immediately.
+func (h *Handlers) serveTranscodeFile(w http.ResponseWriter, r *http.Request, path, contentType string) {
+	// ffmpeg needs a moment to write the playlist after it is spawned. Waiting
+	// for it here is what makes a plain <video src="...m3u8"> work: without it
+	// the first manifest request is a 404, and a native HLS player treats that
+	// as fatal rather than retrying. Segments are never waited for - by the time
+	// the playlist exists the first segment is there or about to be.
+	if filepath.Base(path) == "playlist.m3u8" {
+		waitForFile(r, path, playlistWait)
+	}
+
+	f, err := os.Open(path)
+	if errors.Is(err, os.ErrNotExist) {
+		// The session is running but has not written this file yet.
+		httpx.WriteError(w, http.StatusNotFound, "transcode not ready")
+		return
+	}
+	if err != nil {
+		h.fail(w, "open transcode file", err)
+		return
+	}
+	defer f.Close()
+
+	info, err := f.Stat()
+	if err != nil || !info.Mode().IsRegular() {
+		httpx.WriteError(w, http.StatusNotFound, "transcode not ready")
+		return
+	}
+	// http.ServeContent would sniff the extension; .m3u8 and .ts are not in
+	// stream's MIME table, so the type is set explicitly here.
+	w.Header().Set("Content-Type", contentType)
+	http.ServeContent(w, r, info.Name(), info.ModTime(), f)
+}
+
+// playlistWait bounds how long a playlist request blocks on ffmpeg's first
+// write. Long enough for a hardware encoder to open the input, short enough
+// that a request for something that will never appear still answers.
+const playlistWait = 4 * time.Second
+
+// waitForFile blocks until path exists or the request is cancelled or the
+// deadline passes. It returns either way: the caller still has to handle the
+// file being absent.
+func waitForFile(r *http.Request, path string, limit time.Duration) {
+	deadline := time.Now().Add(limit)
+	for {
+		if _, err := os.Stat(path); err == nil {
+			return
+		}
+		if time.Now().After(deadline) {
+			return
+		}
+		select {
+		case <-r.Context().Done():
+			return
+		case <-time.After(50 * time.Millisecond):
+		}
+	}
+}

internal/media/transcodes_test.go

@@ -0,0 +1,234 @@
+package media
+
+import (
+	"database/sql"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/mithral/vellum/internal/auth"
+	"github.com/mithral/vellum/internal/transcode"
+)
+
+// fakeSessions stands in for *transcode.Manager: it records Start calls and
+// hands back a session rooted at a directory the test controls, with no
+// hardware and no ffmpeg anywhere.
+type fakeSessions struct {
+	dir      string
+	startErr error
+	starts   int
+}
+
+func (f *fakeSessions) Start(itemID int64, inputPath, cacheDir string) (transcode.Session, error) {
+	f.starts++
+	if f.startErr != nil {
+		return nil, f.startErr
+	}
+	return fakeSession{dir: f.dir}, nil
+}
+
+type fakeSession struct{ dir string }
+
+func (s fakeSession) PlaylistPath() string { return filepath.Join(s.dir, "playlist.m3u8") }
+func (s fakeSession) Dir() string          { return s.dir }
+func (s fakeSession) Close() error         { return nil }
+
+// transcodeFixture is a readable video item, an item in a library the caller may
+// not read, and a granted user.
+type transcodeFixture struct {
+	h         *Handlers
+	db        *sql.DB
+	base      string
+	fake      *fakeSessions
+	user      int64
+	item      int64
+	forbidden int64
+}
+
+func newTranscodeFixture(t *testing.T) transcodeFixture {
+	t.Helper()
+	db := testDB(t)
+	lib := newLibrary(t, db, "Video", "video", t.TempDir())
+	other := newLibrary(t, db, "Other", "video", t.TempDir())
+
+	// The session dir sits two levels below base, so "../../etc/passwd" from it
+	// resolves to base/etc/passwd — a file a traversal test can actually create.
+	base := t.TempDir()
+	dir := filepath.Join(base, "transcode", "1-1")
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		t.Fatalf("mkdir session: %v", err)
+	}
+
+	f := transcodeFixture{db: db, base: base, fake: &fakeSessions{dir: dir}}
+	f.user = newUser(t, db, "watcher")
+	grantRead(t, db, f.user, lib)
+
+	const at = "2026-01-01T00:00:00.000Z"
+	f.item = newItem(t, db, lib, "video", "movie.mkv", "Movie", "movie", false, at)
+	f.forbidden = newItem(t, db, other, "video", "secret.mkv", "Secret", "secret", false, at)
+
+	f.h = mediaHandlers(t, db)
+	f.h.sessions = f.fake
+	return f
+}
+
+func getPlaylist(t *testing.T, h *Handlers, u auth.User, itemID int64) *httptest.ResponseRecorder {
+	t.Helper()
+	req := httptest.NewRequest(http.MethodGet, "/api/v1/items/1/transcode/playlist.m3u8", nil)
+	req.SetPathValue("id", itoa(itemID))
+	rec := httptest.NewRecorder()
+	h.handleTranscodePlaylist(rec, req, u)
+	return rec
+}
+
+func getSegment(t *testing.T, h *Handlers, u auth.User, itemID int64, segment string) *httptest.ResponseRecorder {
+	t.Helper()
+	req := httptest.NewRequest(http.MethodGet, "/api/v1/items/1/transcode/"+segment, nil)
+	req.SetPathValue("id", itoa(itemID))
+	req.SetPathValue("segment", segment)
+	rec := httptest.NewRecorder()
+	h.handleTranscodeSegment(rec, req, u)
+	return rec
+}
+
+// TestTranscodePlaylistStartsAndServes pins the lazy start: the request begins
+// the session and returns the playlist bytes exactly as they are on disk.
+func TestTranscodePlaylistStartsAndServes(t *testing.T) {
+	f := newTranscodeFixture(t)
+	const playlist = "#EXTM3U\n#EXT-X-VERSION:3\n#EXTINF:6.0,\nseg00001.ts\n"
+	if err := os.WriteFile(filepath.Join(f.fake.dir, "playlist.m3u8"), []byte(playlist), 0o640); err != nil {
+		t.Fatalf("write playlist: %v", err)
+	}
+
+	rec := getPlaylist(t, f.h, auth.User{ID: f.user}, f.item)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String())
+	}
+	if ct := rec.Header().Get("Content-Type"); ct != "application/vnd.apple.mpegurl" {
+		t.Errorf("content-type = %q, want application/vnd.apple.mpegurl", ct)
+	}
+	if got := rec.Body.String(); got != playlist {
+		t.Errorf("body = %q, want %q", got, playlist)
+	}
+	if f.fake.starts != 1 {
+		t.Errorf("Start calls = %d, want 1", f.fake.starts)
+	}
+
+	// A second request touches the session again, so the idle reaper leaves a
+	// stream being watched alone.
+	_ = getPlaylist(t, f.h, auth.User{ID: f.user}, f.item)
+	if f.fake.starts != 2 {
+		t.Errorf("Start calls after second request = %d, want 2", f.fake.starts)
+	}
+}
+
+// TestTranscodeSegmentServedAndTraversalRefused pins the segment name check: the
+// exact shape ffmpeg writes is served, and everything else — a traversal above
+// all — is a 404.
+func TestTranscodeSegmentServedAndTraversalRefused(t *testing.T) {
+	f := newTranscodeFixture(t)
+	const segment = "\x47@\x00\x10payload"
+	if err := os.WriteFile(filepath.Join(f.fake.dir, "seg00001.ts"), []byte(segment), 0o640); err != nil {
+		t.Fatalf("write segment: %v", err)
+	}
+
+	rec := getSegment(t, f.h, auth.User{ID: f.user}, f.item, "seg00001.ts")
+	if rec.Code != http.StatusOK {
+		t.Fatalf("valid segment status = %d, body=%s", rec.Code, rec.Body.String())
+	}
+	if ct := rec.Header().Get("Content-Type"); ct != "video/mp2t" {
+		t.Errorf("content-type = %q, want video/mp2t", ct)
+	}
+	if got := rec.Body.String(); got != segment {
+		t.Errorf("body = %q, want %q", got, segment)
+	}
+
+	for _, name := range []string{
+		"../../etc/passwd",
+		"..%2f..%2fetc%2fpasswd",
+		"seg1.ts",
+		"seg0001.ts",
+		"seg000001.ts",
+		"seg00001.ts.bak",
+		"playlist.m3u8",
+		"",
+	} {
+		if rec := getSegment(t, f.h, auth.User{ID: f.user}, f.item, name); rec.Code != http.StatusNotFound {
+			t.Errorf("segment %q status = %d, want 404", name, rec.Code)
+		}
+	}
+}
+
+// TestTranscodeSegmentTraversalCannotEscapeOnce pins the traversal case with a
+// file that genuinely exists at the joined path: without the name check the
+// handler would serve it, so a green run proves the check is load-bearing.
+func TestTranscodeSegmentTraversalCannotEscapeOnce(t *testing.T) {
+	f := newTranscodeFixture(t)
+	if err := os.MkdirAll(filepath.Join(f.base, "etc"), 0o755); err != nil {
+		t.Fatalf("mkdir etc: %v", err)
+	}
+	if err := os.WriteFile(filepath.Join(f.base, "etc", "passwd"), []byte("SECRET"), 0o640); err != nil {
+		t.Fatalf("write passwd: %v", err)
+	}
+
+	rec := getSegment(t, f.h, auth.User{ID: f.user}, f.item, "../../etc/passwd")
+	if rec.Code != http.StatusNotFound {
+		t.Fatalf("traversal status = %d, want 404 (body=%s)", rec.Code, rec.Body.String())
+	}
+	if rec.Body.String() == "SECRET" {
+		t.Fatal("traversal served a file outside the session directory")
+	}
+}
+
+// TestTranscodeRoutesAreGrantScoped pins the access chokepoint on both routes:
+// an item in a library the caller cannot read answers with the same 404, byte
+// for byte, as an item that does not exist — and does not start an encoder.
+func TestTranscodeRoutesAreGrantScoped(t *testing.T) {
+	f := newTranscodeFixture(t)
+	u := auth.User{ID: f.user}
+
+	fb := getPlaylist(t, f.h, u, f.forbidden)
+	ab := getPlaylist(t, f.h, u, 999999)
+	if fb.Code != http.StatusNotFound || ab.Code != http.StatusNotFound {
+		t.Fatalf("playlist statuses = %d/%d, want 404/404", fb.Code, ab.Code)
+	}
+	if fb.Body.String() != ab.Body.String() {
+		t.Errorf("playlist forbidden body %q != absent body %q", fb.Body.String(), ab.Body.String())
+	}
+
+	fs := getSegment(t, f.h, u, f.forbidden, "seg00001.ts")
+	as := getSegment(t, f.h, u, 999999, "seg00001.ts")
+	if fs.Code != http.StatusNotFound || as.Code != http.StatusNotFound {
+		t.Fatalf("segment statuses = %d/%d, want 404/404", fs.Code, as.Code)
+	}
+	if fs.Body.String() != as.Body.String() {
+		t.Errorf("segment forbidden body %q != absent body %q", fs.Body.String(), as.Body.String())
+	}
+
+	// The item is rejected before Start, so a forbidden request never pins the
+	// GPU.
+	if f.fake.starts != 0 {
+		t.Errorf("Start calls on forbidden requests = %d, want 0", f.fake.starts)
+	}
+
+	none := getPlaylist(t, f.h, auth.User{ID: newUser(t, f.db, "nobody")}, f.item)
+	if none.Code != http.StatusNotFound {
+		t.Errorf("no-grant status = %d, want 404", none.Code)
+	}
+}
+
+// TestTranscodeRefusalsAreNot500 pins that a host with no encoder and a full
+// server both get a clear 503 rather than a 500.
+func TestTranscodeRefusalsAreNot500(t *testing.T) {
+	for _, err := range []error{transcode.ErrUnavailable, transcode.ErrAtCapacity, transcode.ErrShutdown} {
+		f := newTranscodeFixture(t)
+		f.fake.startErr = err
+
+		rec := getPlaylist(t, f.h, auth.User{ID: f.user}, f.item)
+		if rec.Code != http.StatusServiceUnavailable {
+			t.Errorf("start error %v status = %d, want 503", err, rec.Code)
+		}
+	}
+}