Files
33a57ae672 fix(playback): route v3 direct play and remux through proxy nodes (#620)
* fix(playback): route v3 direct play and remux through proxy nodes

Protocol v3 consulted the node planner only for the HLS deliveries, so
`original_http` and `server_remux_progressive` sessions returned an
API-local `/stream/{session_id}` URL and the API node served the bytes —
ServeDirectPlay for direct play, a locally spawned ffmpeg for the remux.
An operator running dedicated proxy nodes still saw all of that egress on
the API node.

The capability already existed: the proxy implements /stream/direct and
/stream/remux, and the Jellyfin-compat transport already plans a proxy for
exactly these two methods. Native v3 was the only surface skipping it, so
Jellyfin clients routed correctly on a deployment where Silo's own clients
did not. This wires the same shape into the v3 identity transport rather
than inventing a second selection path.

The proxy serves from the stream token alone, so the token now carries the
media path, the file's Dolby Vision profile (a P7 remux must strip the
dangling RPU) and the audio-only flag (which picks audio/mp4 over
video/mp4, the MIME the plan promised). RecipeCard models none of the
three; a missing claim would not fail loudly, it would serve a subtly
different stream than the plan promised.

Two related fixes:

- Proxy direct play served via http.ServeFile, which sets no strong ETag.
  direct_stream_resume_v1 depends on the ETag ServeDirectPlay sets before
  ServeContent, so routing direct play to a proxy without this would have
  silently broken resumable direct streams: If-Range never validates and a
  resumed range restarts at 200. The proxy now uses the same serve path.

- playback.local_transcode_fallback was only checked in the HLS branch, so
  a progressive remux that converts audio still spawned ffmpeg locally on
  an API-only node with the setting disabled. Identity deliveries now
  honor the gate too — direct play still falls back locally, since moving
  bytes is not transcode work and single-node deployments must keep
  working.

Falling back to the API-local path when no proxy is eligible preserves
single-node behavior, and a planner reservation is released whenever the
session does not actually reach a proxy.

Closes #619

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(playback): validate proxy recipes and keep proxy sessions alive

Addresses three P1 findings on the proxy-transport change.

Proxies do run ffmpeg — /stream/remux converts audio and strips Dolby
Vision RPUs — but they exposed no capability endpoint, so unlike the HLS
offload path nothing checked that the selected proxy could execute the
transformations a plan froze. A pool whose proxies carry a different
ffmpeg build (rolling upgrade, custom image) would fail at stream time: a
missing aac encoder 500s, a missing dovi_rpu filter is refused outright by
the remux itself. Proxies now serve /hw-capabilities in the same shape and
at the same path as a transcode node, and identity planning validates the
frozen recipe against the selected proxy, falling back to a node that can
do the work. A proxy that does not answer is treated as incapable rather
than assumed good: an older proxy predating the endpoint is exactly the
mismatched build the check exists to catch. Direct play copies bytes and
needs no recipe, so it skips the probe entirely.

meteredResponseWriter implemented neither Unwrap nor SetWriteDeadline, so
RollingDeadlineWriter could not install its stall deadline on any proxy
stream. With the standalone proxy running WriteTimeout 0 there was no
server-level guard behind it, so a client that stopped reading without
closing its connection would block a write forever, holding the session,
the file, the goroutine and the connection.

A proxy-served session never produces a transport request on the API node,
so activeTransportCount — what protects a local stream from the idle
reaper — stays zero and a heartbeat gap longer than the active grace would
reap a healthy stream, after which progress, stop and replan all fail with
session-not-found while bytes still flow. Sessions are now marked as
remotely transported, which widens their idle windows rather than granting
immunity: this manager has no absolute session lifetime, so unconditional
immunity would leak a session forever when a client disappears without
stopping. The mark is always set on commit, so a re-plan that moves a
session back onto the API clears a stale one.

Also adopts the exported transformation constants in the tests and covers
the effective-recipe bitrate branch, per review.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(playback): pick capable sibling proxies and refresh transport locality

Narrow proxy selection by capability *before* selection rather than
rejecting a single round-robin pick afterwards. Abandoning the pool on one
mismatch meant a capable proxy with free capacity sat unused while
playback either ran ffmpeg on the API node or, with
playback.local_transcode_fallback disabled, was refused outright — the
exact api/proxy split this branch targets, during exactly the rolling
ffmpeg upgrade the capability check exists for. PlanSessionWith now
applies its eligibility predicate to the proxy on proxy-only plans (the
proxy is the executor there), mirroring how HLS filters transcode nodes,
and the planner grows ProxyNodeURLs to match TranscodeNodeURLs. Direct
play still skips the probe: it copies bytes and needs no recipe.

Every committed route now records transport locality, not just the
identity-proxy one. A session replanned from a proxy onto the integrated
transcoder previously kept a stale remote-transport mark, and the widened
idle grace it grants would hold that session's stream and transcode slots
for five minutes after the local stream disconnected without an explicit
stop. The remote HLS route sets it too — it also hands the client an
absolute proxy URL that never reaches this server.

The proxy's CORS config exposed no response headers, so cross-origin
JavaScript could send the If-Range/Range request headers it already allows
but never read the ETag, Accept-Ranges or Content-Range needed to build
them. direct_stream_resume_v1 silently degraded to a full restart whenever
the proxy was on a different origin than the web app, which is the normal
deployment.

Also regenerates internal/playback/testdata/protocol_v3 and the schema
fixtures, which were stale for output_change_v1 since #613/#617 and failed
CI on every branch.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(playback): restore the alternate-version fallback for burn-in refusals

#617 renamed the terminal a burn-in-forced adaptation reports: when the
subtitle burn requirement is the sole trigger, an HDR source that cannot
be re-encoded now returns subtitle_conversion_unsupported instead of
hdr_transcode_unsupported, so the refusal names the thing the viewer can
actually act on.

terminalAllowsAlternateFileV3 was not updated to match, and it gates the
alternate-version retry on the old reason strings. That silently retired
the fallback for exactly the case its own comment describes — a bitmap
subtitle needing burn-in that an HDR source cannot support while an SDR
alternate can. Playback was refused outright instead of switching to the
version that can serve it.

Adds the new reason to the gate and covers it directly, so a future
rename of a refusal reason fails on the gate rather than only on the
end-to-end replan test.

Also drops debug instrumentation that was committed by mistake in
TestHandleReplanPlaybackV3BitmapSubtitleFallsBackFromHDRToSDRVersion; the
assertion is back to its original form and now passes on the merits.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-13 10:21:27 -04:00

595 lines
23 KiB
Go

package proxy
import (
"context"
"encoding/json"
"fmt"
"io"
"log/slog"
"mime"
"net/http"
"os"
"path/filepath"
"strconv"
"strings"
"sync"
"time"
"github.com/go-chi/chi/v5"
"github.com/go-chi/cors"
"github.com/Silo-Server/silo-server/internal/downloadprepare"
"github.com/Silo-Server/silo-server/internal/downloads"
"github.com/Silo-Server/silo-server/internal/nodeconfig"
"github.com/Silo-Server/silo-server/internal/nodesessions"
"github.com/Silo-Server/silo-server/internal/playback"
"github.com/Silo-Server/silo-server/internal/streamtoken"
)
// Server is the HTTP handler for proxy mode.
type Server struct {
watcher *nodeconfig.Watcher
tracker *nodesessions.Tracker
httpClient *http.Client
artifactMissReporter remoteArtifactMissReporter
egress *egressMeter
// subCache stores full-track PGS (.sup) extracts under the transcode dir
// so repeat selections skip the whole-file ffmpeg demux.
subCache *playback.SubtitleCache
// Download limits are node-local once egress is delegated. Rebuild the
// manager when hot-reloaded settings change so existing transfers retain
// their original limiter while new transfers use the new values.
downloadBandwidthMu sync.Mutex
downloadBandwidth *downloads.BandwidthManager
downloadServerBPS int64
downloadUserBPS int64
}
type remoteArtifactMissReporter interface {
ReportRemoteArtifactMissing(ctx context.Context, artifactID, originNodeURL, originArtifactID string) error
}
// NewServer creates a new proxy server backed by a config watcher and session
// tracker.
func NewServer(watcher *nodeconfig.Watcher, tracker *nodesessions.Tracker) *Server {
return &Server{
watcher: watcher,
tracker: tracker,
// No overall timeout — stream bodies are long-lived. Hung nodes are
// bounded by the transport's response-header timeout instead.
httpClient: &http.Client{
Transport: newStreamTransport(),
CheckRedirect: func(_ *http.Request, _ []*http.Request) error {
return http.ErrUseLastResponse
},
},
egress: newEgressMeter(),
subCache: playback.NewSubtitleCache(func() string {
return watcher.Config().Playback.TranscodeDir
}),
}
}
// SetRemoteArtifactMissReporter wires the authoritative database transition
// used when an origin returns 404 after the API's proxy preflight. It must be
// called during construction, before the server begins handling requests.
func (s *Server) SetRemoteArtifactMissReporter(reporter remoteArtifactMissReporter) {
s.artifactMissReporter = reporter
}
// newStreamTransport tunes the proxy→transcode-node connection pool. Many
// concurrent viewers fan their segment fetches through one proxy→node pair,
// and Go's default of 2 idle connections per host causes constant connection
// churn (and TLS re-handshakes) under load. The response-header timeout
// bounds requests to a hung node; the longest legitimate server-side wait is
// the 30s manifest-readiness poll on the transcode node.
func newStreamTransport() *http.Transport {
t := http.DefaultTransport.(*http.Transport).Clone()
t.MaxIdleConns = 128
t.MaxIdleConnsPerHost = 32
t.ResponseHeaderTimeout = 60 * time.Second
return t
}
// Handler returns the chi.Router with all proxy routes mounted.
func (s *Server) Handler() http.Handler {
r := chi.NewRouter()
// hls.js uses XHR for manifest/segment fetches which are subject to
// CORS when the proxy runs on a different origin than the web app.
r.Use(cors.Handler(cors.Options{
AllowedOrigins: []string{"*"},
AllowedMethods: []string{"GET", "HEAD", "OPTIONS"},
AllowedHeaders: []string{
"Accept", "Authorization", "Content-Type", "Range",
"If-Match", "If-Modified-Since", "If-None-Match", "If-Range", "If-Unmodified-Since",
},
// direct_stream_resume_v1 has the client re-request a byte range with
// If-Range against the entity tag it stored. Cross-origin JavaScript
// can only read a response header that is explicitly exposed, so
// without these the client can send the conditional request headers
// above but never learn the values to put in them — the resume
// contract silently degrades to a full restart on a proxy that is on a
// different origin than the web app, which is the normal deployment.
ExposedHeaders: []string{
"Accept-Ranges", "Content-Encoding", "Content-Length", "Content-Range",
"ETag", "Last-Modified",
},
MaxAge: 86400,
}))
r.Get("/api/v1/health", s.handleHealth)
r.Group(func(r chi.Router) {
// Streaming and download bytes count toward the node's measured egress.
r.Use(s.meterEgress)
r.Head("/stream/direct/{token}", s.handleDirectPlay)
r.Get("/stream/direct/{token}", s.handleDirectPlay)
r.Head("/stream/remux/{token}", s.handleRemux)
r.Get("/stream/remux/{token}", s.handleRemux)
r.Head("/stream/transcode/{token}/master.m3u8", s.handleTranscodeManifest)
r.Get("/stream/transcode/{token}/master.m3u8", s.handleTranscodeManifest)
r.Get("/stream/transcode/{token}/segment/{name}", s.handleTranscodeSegment)
r.Get("/stream/subtitles/{token}/{track}/fonts", s.handleSubtitleFonts)
r.Get("/stream/subtitles/{token}/{track}", s.handleSubtitle)
r.Head("/downloads/file/{token}", s.handleDownloadFile)
r.Get("/downloads/file/{token}", s.handleDownloadFile)
})
// Admin routes — bearer-auth protected.
r.Group(func(r chi.Router) {
r.Use(s.requireBearer)
r.Get("/hw-capabilities", s.handleHWCapabilities)
r.Post("/admin/force-reload", s.handleForceReload)
r.Get("/status", s.handleStatus)
})
return r
}
// handleHWCapabilities advertises what this proxy's ffmpeg can actually do, in
// the same shape and at the same path as a transcode node.
//
// A proxy executes recipes too: /stream/remux runs ffmpeg to convert audio or
// strip a Dolby Vision RPU. Without this endpoint the API has no way to tell
// whether the proxy it just picked can run the transformations a plan froze, so
// a pool whose proxies carry a different ffmpeg build (a rolling upgrade, a
// custom image) would fail at stream time rather than at selection time.
func (s *Server) handleHWCapabilities(w http.ResponseWriter, r *http.Request) {
ffmpegPath := ""
if cfg := s.watcher.Config(); cfg != nil {
ffmpegPath = cfg.Playback.FFmpegPath
}
info := playback.DetectHWAccelWithFFmpeg(ffmpegPath)
info.Transformations = playback.ProbeTransformationRegistryV3(r.Context(), ffmpegPath).Advertised()
w.Header().Set("Content-Type", "application/json")
if err := json.NewEncoder(w).Encode(info); err != nil {
slog.WarnContext(r.Context(), "encode proxy capabilities", "component", "proxy", "error", err)
}
}
type healthResponse struct {
Status string `json:"status"`
ActiveJobs int `json:"active_jobs"`
EgressKbps int `json:"egress_kbps"`
}
func (s *Server) handleHealth(w http.ResponseWriter, _ *http.Request) {
activeJobs := 0
if s.tracker != nil {
activeJobs = s.tracker.ActiveCount()
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(healthResponse{
Status: "ok",
ActiveJobs: activeJobs,
EgressKbps: s.egress.RateKbps(),
})
}
// requireBearer checks Authorization: Bearer {secret} for admin endpoints.
func (s *Server) requireBearer(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
cfg := s.watcher.Config()
auth := r.Header.Get("Authorization")
if !strings.HasPrefix(auth, "Bearer ") || strings.TrimPrefix(auth, "Bearer ") != cfg.Auth.JWTSecret {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
next.ServeHTTP(w, r)
})
}
// verifyToken extracts and validates the stream token from the URL.
func (s *Server) verifyToken(w http.ResponseWriter, r *http.Request) *streamtoken.Claims {
cfg := s.watcher.Config()
tokenStr := chi.URLParam(r, "token")
claims, err := streamtoken.Verify(tokenStr, cfg.Auth.JWTSecret)
if err != nil {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return nil
}
return claims
}
func (s *Server) handleDirectPlay(w http.ResponseWriter, r *http.Request) {
claims := s.verifyToken(w, r)
if claims == nil {
return
}
info := sessionInfo(s.tracker, claims, "direct_play")
s.tracker.Track(r.Context(), info)
defer s.tracker.Remove(r.Context(), claims.SessionID)
// Serve through the same path the integrated server uses rather than a bare
// http.ServeFile: direct_stream_resume_v1 requires the strong ETag that
// ServeDirectPlay sets before ServeContent (ServeFile sets none, so
// If-Range never validates and a resumed range silently restarts at 200),
// and it carries the rolling write deadline and stream metrics with it.
_ = playback.ServeDirectPlay(w, r, claims.MediaPath)
}
func (s *Server) handleDownloadFile(w http.ResponseWriter, r *http.Request) {
claims := s.verifyToken(w, r)
if claims == nil {
return
}
remoteArtifact := claims.DownloadArtifactID != "" && strings.TrimSpace(claims.TranscodeNode) != ""
if claims.PlayMethod != streamtoken.PlayMethodDownload || (strings.TrimSpace(claims.MediaPath) == "" && !remoteArtifact) {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
// HEAD is a capability/path preflight, not an active transfer. Counting it
// would briefly consume job capacity and could make a health report retire
// the API's reservation before the client starts its GET.
if s.tracker != nil && r.Method != http.MethodHead {
info := sessionInfo(s.tracker, claims, "download")
s.tracker.Track(r.Context(), info)
defer s.tracker.Remove(context.WithoutCancel(r.Context()), claims.SessionID)
}
if remoteArtifact {
s.relayDownloadArtifact(w, r, claims)
return
}
f, err := os.Open(claims.MediaPath)
if err != nil {
http.NotFound(w, r)
return
}
defer func() { _ = f.Close() }()
stat, err := f.Stat()
if err != nil {
http.Error(w, "download unavailable", http.StatusInternalServerError)
return
}
filename := filepath.Base(claims.MediaPath)
if disposition := mime.FormatMediaType("attachment", map[string]string{"filename": filename}); disposition != "" {
w.Header().Set("Content-Disposition", disposition)
}
w.Header().Set("Content-Type", playback.MimeFromExtension(claims.MediaPath))
reader := io.ReadSeeker(f)
if bandwidth := s.downloadBandwidthManager(); bandwidth != nil {
reader = bandwidth.ThrottledReader(r.Context(), f, claims.UserID)
}
http.ServeContent(w, r, stat.Name(), stat.ModTime(), reader)
}
func (s *Server) relayDownloadArtifact(w http.ResponseWriter, r *http.Request, claims *streamtoken.Claims) {
if !downloadprepare.ValidArtifactID(claims.DownloadArtifactID) {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
cfg := s.watcher.Config()
if cfg == nil || strings.TrimSpace(cfg.Auth.JWTSecret) == "" {
http.Error(w, "download unavailable", http.StatusServiceUnavailable)
return
}
client := downloadprepare.HTTPPreparer{Client: s.httpClient}
resp, err := client.Open(r.Context(), claims.TranscodeNode, cfg.Auth.JWTSecret, claims.DownloadArtifactID, r.Method, r.Header)
if err != nil {
slog.WarnContext(r.Context(), "download artifact relay failed", "component", "proxy", "artifact_id", claims.DownloadArtifactID, "node", claims.TranscodeNode, "error", err)
http.Error(w, "download unavailable", http.StatusBadGateway)
return
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode == http.StatusNotFound {
if s.artifactMissReporter != nil && strings.TrimSpace(claims.DownloadArtifactRowID) != "" {
reportCtx, cancel := context.WithTimeout(context.WithoutCancel(r.Context()), 5*time.Second)
err := s.artifactMissReporter.ReportRemoteArtifactMissing(
reportCtx, claims.DownloadArtifactRowID, claims.TranscodeNode, claims.DownloadArtifactID,
)
cancel()
if err != nil {
slog.WarnContext(r.Context(), "report missing remote download artifact", "component", "proxy", "artifact_id", claims.DownloadArtifactRowID, "error", err)
}
}
http.NotFound(w, r)
return
}
if !downloadprepare.RelayStatusAllowed(resp.StatusCode) {
http.Error(w, "download unavailable", http.StatusBadGateway)
return
}
downloadprepare.CopyResponseHeaders(w.Header(), resp.Header)
if filename := filepath.Base(strings.TrimSpace(claims.DownloadFilename)); filename != "" && filename != "." {
if disposition := mime.FormatMediaType("attachment", map[string]string{"filename": filename}); disposition != "" {
w.Header().Set("Content-Disposition", disposition)
}
}
w.WriteHeader(resp.StatusCode)
if r.Method == http.MethodHead {
return
}
var reader io.Reader = resp.Body
if bandwidth := s.downloadBandwidthManager(); bandwidth != nil {
reader = bandwidth.ThrottledStreamReader(r.Context(), reader, claims.UserID)
}
if _, err := io.Copy(w, reader); err != nil && r.Context().Err() == nil {
slog.WarnContext(r.Context(), "download artifact relay interrupted", "component", "proxy", "artifact_id", claims.DownloadArtifactID, "error", err)
}
}
func (s *Server) downloadBandwidthManager() *downloads.BandwidthManager {
cfg := s.watcher.Config()
if cfg == nil {
return nil
}
serverBPS := cfg.Download.ServerBandwidthBPS
userBPS := cfg.Download.UserBandwidthBPS
s.downloadBandwidthMu.Lock()
defer s.downloadBandwidthMu.Unlock()
if s.downloadBandwidth == nil || serverBPS != s.downloadServerBPS || userBPS != s.downloadUserBPS {
s.downloadBandwidth = downloads.NewBandwidthManager(serverBPS, userBPS)
s.downloadServerBPS = serverBPS
s.downloadUserBPS = userBPS
}
return s.downloadBandwidth
}
func (s *Server) handleRemux(w http.ResponseWriter, r *http.Request) {
claims := s.verifyToken(w, r)
if claims == nil {
return
}
info := sessionInfo(s.tracker, claims, "remux")
s.tracker.Track(r.Context(), info)
defer s.tracker.Remove(r.Context(), claims.SessionID)
seekSeconds := 0.0
if seekStr := r.URL.Query().Get("seek"); seekStr != "" {
if v, err := strconv.ParseFloat(seekStr, 64); err == nil {
seekSeconds = v
}
}
// Honor the Dolby Vision mode frozen in the token (empty decodes as the
// legacy auto behavior for old tokens), mirroring how the integrated
// server's stream handler serves the same claims.
_ = playback.ServeRemuxWithOptions(w, r, claims.MediaPath, "mp4", seekSeconds, claims.TranscodeAudio, claims.AudioTrackIndex, claims.DVProfile, playback.RemuxServeOptions{
DVMode: playback.RemuxDVMode(claims.RemuxDVMode),
FFmpegPath: s.watcher.Config().Playback.FFmpegPath,
ContentType: playback.RemuxContentType(claims.AudioOnly),
AudioOnly: claims.AudioOnly,
TargetAudioChannels: claims.TargetAudioChannels,
TargetAudioBitrateKbps: claims.TargetAudioBitrateKbps,
})
}
func (s *Server) handleTranscodeManifest(w http.ResponseWriter, r *http.Request) {
claims := s.verifyToken(w, r)
if claims == nil {
return
}
s.touchTranscodeSession(r, claims)
s.proxyToTranscodeNode(w, r, claims, "/transcode/"+transcodeTransportIDFromClaims(claims)+"/master.m3u8")
}
func (s *Server) handleTranscodeSegment(w http.ResponseWriter, r *http.Request) {
claims := s.verifyToken(w, r)
if claims == nil {
return
}
s.touchTranscodeSession(r, claims)
name := chi.URLParam(r, "name")
s.proxyToTranscodeNode(w, r, claims, "/transcode/"+transcodeTransportIDFromClaims(claims)+"/segment/"+name)
}
func transcodeTransportIDFromClaims(claims *streamtoken.Claims) string {
if claims != nil && claims.TranscodeTransportID != "" {
return claims.TranscodeTransportID
}
if claims == nil {
return ""
}
return claims.SessionID
}
// touchTranscodeSession keeps HLS sessions visible in the active stream count.
// Unlike direct play and remux, transcode playback reaches the proxy as many
// short manifest/segment requests, so the session is tracked by recent
// activity instead of request lifetime.
func (s *Server) touchTranscodeSession(r *http.Request, claims *streamtoken.Claims) {
s.tracker.Touch(r.Context(), sessionInfo(s.tracker, claims, "transcode"))
}
// sessionInfo builds the node-session tracker record for a verified token,
// copying the numeric ownership keys the node-session tracker needs.
func sessionInfo(tr *nodesessions.Tracker, claims *streamtoken.Claims, kind string) nodesessions.SessionInfo {
return nodesessions.SessionInfo{
SessionID: claims.SessionID,
NodeURL: tr.NodeURL(),
NodeName: tr.NodeName(),
Type: kind,
StartedAt: time.Now().UTC().Format(time.RFC3339),
AuthUserID: claims.UserID,
ProfileID: claims.ProfileID,
MediaFileID: claims.MediaFileID,
}
}
func (s *Server) handleSubtitle(w http.ResponseWriter, r *http.Request) {
claims := s.verifyToken(w, r)
if claims == nil {
return
}
cfg := s.watcher.Config()
trackParam := chi.URLParam(r, "track")
trackIndex, requestedFormat, err := playback.ParseSubtitleTrackParam(trackParam)
if err != nil {
http.Error(w, "invalid subtitle index", http.StatusBadRequest)
return
}
// When the URL requests SUP format (e.g. /subtitles/{token}/2.sup),
// serve the PGS track as a raw .sup elementary stream for client-side
// bitmap rendering (libpgs). Unlike the buffered text paths below, this
// serves the cached full-track extract when present, and otherwise
// streams ffmpeg output directly (the client renders progressively as
// data arrives) while teeing it into the cache for the next request.
// Clients that manage their own sliding window opt in with ?windowed=1
// (+ ?position=/?duration=), mirroring the API stream handler; windowed
// requests extract only the requested slice — from the cached full
// track when one exists (warming it in the background when not).
if requestedFormat == "sup" {
allowWindow, seek, duration := playback.PGSWindowRequest(r.URL.Query())
err := s.subCache.ServeSUPExtract(w, r, playback.StreamExtractOpts{
InputPath: claims.MediaPath,
TrackIndex: trackIndex,
SourceCodec: "hdmv_pgs_subtitle", // .sup URLs are only generated for PGS tracks
SeekSeconds: seek,
DurationSeconds: duration,
AllowWindow: allowWindow,
FFmpegPath: cfg.Playback.FFmpegPath,
}, playback.StreamExtractSubtitle)
if err != nil && r.Context().Err() == nil {
// Headers already committed — log and let the client see a
// truncated response.
slog.ErrorContext(r.Context(), "stream subtitle (sup)", "component", "proxy", "error", err, "track", trackIndex,
"path", claims.MediaPath, "playback_session_id", claims.SessionID)
}
return
}
// When the URL requests ASS format (e.g. /subtitles/{token}/2.ass),
// extract as raw ASS to preserve styling for client-side rendering.
if requestedFormat == "ass" {
data, err := playback.ExtractSubtitleWithFormat(r.Context(), claims.MediaPath, trackIndex, "ass", cfg.Playback.FFmpegPath)
if err != nil {
slog.ErrorContext(r.Context(), "extract subtitle (ass)", "component", "proxy", "error", err, "track", trackIndex, "path", claims.MediaPath, "playback_session_id", claims.SessionID)
http.Error(w, "subtitle extraction failed", http.StatusInternalServerError)
return
}
playback.ServeSubtitle(w, data, "ass")
return
}
data, format, err := playback.ExtractSubtitle(r.Context(), claims.MediaPath, trackIndex, cfg.Playback.FFmpegPath)
if err != nil {
slog.ErrorContext(r.Context(), "extract subtitle", "component", "proxy", "error", err, "track", trackIndex, "path", claims.MediaPath, "playback_session_id", claims.SessionID)
http.Error(w, "subtitle extraction failed", http.StatusInternalServerError)
return
}
vtt, err := playback.ConvertToVTT(data, format)
if err != nil {
slog.ErrorContext(r.Context(), "convert to vtt", "component", "proxy", "error", err, "playback_session_id", claims.SessionID)
http.Error(w, "subtitle conversion failed", http.StatusInternalServerError)
return
}
playback.ServeSubtitle(w, vtt, "vtt")
}
func (s *Server) handleSubtitleFonts(w http.ResponseWriter, r *http.Request) {
claims := s.verifyToken(w, r)
if claims == nil {
return
}
cfg := s.watcher.Config()
trackParam := chi.URLParam(r, "track")
trackIndex, _, err := playback.ParseSubtitleTrackParam(trackParam)
if err != nil {
http.Error(w, "invalid subtitle index", http.StatusBadRequest)
return
}
fonts, err := playback.ExtractAttachedSubtitleFonts(r.Context(), claims.MediaPath, cfg.Playback.FFmpegPath)
if err != nil {
slog.ErrorContext(r.Context(), "extract subtitle fonts", "component", "proxy", "error", err, "track", trackIndex, "path", claims.MediaPath, "playback_session_id", claims.SessionID)
http.Error(w, "subtitle font extraction failed", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Cache-Control", "no-store")
if err := json.NewEncoder(w).Encode(playback.EncodeSubtitleFontBundle(fonts)); err != nil {
slog.WarnContext(r.Context(), "subtitle font response encode failed", "component", "proxy", "error", err, "playback_session_id", claims.SessionID)
}
}
// proxyToTranscodeNode forwards the request to the transcode node specified in the claims.
func (s *Server) proxyToTranscodeNode(w http.ResponseWriter, r *http.Request, claims *streamtoken.Claims, path string) {
cfg := s.watcher.Config()
if claims.TranscodeNode == "" {
http.Error(w, "no transcode node in token", http.StatusBadRequest)
return
}
targetURL := claims.TranscodeNode + path
if rawQuery := r.URL.RawQuery; rawQuery != "" {
targetURL = fmt.Sprintf("%s?%s", targetURL, rawQuery)
}
req, err := http.NewRequestWithContext(r.Context(), r.Method, targetURL, nil)
if err != nil {
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
req.Header.Set("Authorization", "Bearer "+cfg.Auth.JWTSecret)
// Forward the verified stream token so the transcode node can self-reconstruct
// a lost session after its OWN restart: the token carries the full byte-affecting
// recipe, so the node can re-spawn ffmpeg seeked to the requested segment instead
// of 404ing (the integrated server already does this from the same token). The
// node re-verifies the token independently before trusting it.
if token := chi.URLParam(r, "token"); token != "" {
req.Header.Set("X-Silo-Stream-Token", token)
}
resp, err := s.httpClient.Do(req)
if err != nil {
slog.ErrorContext(r.Context(), "proxy to transcode node", "component", "proxy", "error", err, "url", targetURL, "playback_session_id", claims.SessionID)
http.Error(w, "transcode node unavailable", http.StatusBadGateway)
return
}
defer resp.Body.Close()
// Copy response headers
for k, vv := range resp.Header {
for _, v := range vv {
w.Header().Add(k, v)
}
}
w.WriteHeader(resp.StatusCode)
io.Copy(w, resp.Body)
}
func (s *Server) handleForceReload(w http.ResponseWriter, r *http.Request) {
if err := s.watcher.ForceReload(r.Context()); err != nil {
http.Error(w, "reload failed: "+err.Error(), http.StatusInternalServerError)
return
}
slog.InfoContext(r.Context(), "proxy force reload completed", slog.String("component", "proxy"))
w.WriteHeader(http.StatusNoContent)
}
type statusResponse struct {
ActiveSessions int `json:"active_sessions"`
}
func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(statusResponse{
ActiveSessions: s.tracker.ActiveCount(),
})
}