diff --git a/cmd/vellum/main.go b/cmd/vellum/main.go index a3b4f5d..8a8933f 100644 --- a/cmd/vellum/main.go +++ b/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()) diff --git a/internal/media/fixtures_test.go b/internal/media/fixtures_test.go index d599ec6..6a44eb7 100644 --- a/internal/media/fixtures_test.go +++ b/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 { diff --git a/internal/media/handlers.go b/internal/media/handlers.go index f6317b0..769f75f 100644 --- a/internal/media/handlers.go +++ b/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 -// extracted pages 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} +// NewHandlers builds the media HTTP layer. cacheDir is where covers, extracted +// pages and transcode sessions are written. +// +// 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)) diff --git a/internal/media/transcodes.go b/internal/media/transcodes.go new file mode 100644 index 0000000..49161f4 --- /dev/null +++ b/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