2026-05-22 20:26:11 -04:00
package handlers
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log/slog"
"net/http"
"os"
"path/filepath"
"sort"
"strconv"
"strings"
"sync"
"time"
"github.com/go-chi/chi/v5"
apimw "github.com/Silo-Server/silo-server/internal/api/middleware"
"github.com/Silo-Server/silo-server/internal/catalog"
"github.com/Silo-Server/silo-server/internal/clientip"
2026-06-10 19:25:07 -04:00
"github.com/Silo-Server/silo-server/internal/config"
2026-05-22 20:26:11 -04:00
evt "github.com/Silo-Server/silo-server/internal/events"
2026-07-09 23:07:16 -04:00
"github.com/Silo-Server/silo-server/internal/httpstream"
2026-05-22 20:26:11 -04:00
"github.com/Silo-Server/silo-server/internal/markers"
"github.com/Silo-Server/silo-server/internal/models"
"github.com/Silo-Server/silo-server/internal/nodepool"
"github.com/Silo-Server/silo-server/internal/playback"
2026-07-30 10:52:41 -04:00
"github.com/Silo-Server/silo-server/internal/settingscontract"
"github.com/Silo-Server/silo-server/internal/settingskeys"
"github.com/Silo-Server/silo-server/internal/settingsresolve"
2026-05-22 20:26:11 -04:00
"github.com/Silo-Server/silo-server/internal/streamtoken"
"github.com/Silo-Server/silo-server/internal/subtitles"
2026-08-11 14:30:55 -04:00
"github.com/Silo-Server/silo-server/internal/transcodeproxy"
2026-05-22 20:26:11 -04:00
"github.com/Silo-Server/silo-server/internal/userstore"
"github.com/Silo-Server/silo-server/internal/watchstate"
"github.com/Silo-Server/silo-server/internal/watchsync"
)
// SessionManagerInterface defines the operations the PlaybackHandler needs
// on the session manager.
type SessionManagerInterface interface {
StartSession ( userID int , profileID string , fileID int , method playback . PlayMethod , transcodeAudio bool ) ( * playback . Session , error )
StartSessionWithFiles ( userID int , profileID string , effectiveFileID int , requestedFileID int , method playback . PlayMethod , transcodeAudio bool ) ( * playback . Session , error )
UpdateProgress ( sessionID string , position float64 , isPaused bool ) error
UpdateAudioTrack ( sessionID string , audioTrackIndex int , method playback . PlayMethod ) error
UpdateStreamState ( sessionID string , state playback . SessionStreamState ) error
TouchActivity ( sessionID string ) error
BeginTransport ( sessionID string ) error
EndTransport ( sessionID string ) error
SetEffectiveMediaFileID ( sessionID string , fileID int ) error
SetTranscodeNodeURL ( sessionID , url string ) error
2026-07-20 19:25:46 +02:00
SetTranscodeRoute ( sessionID string , route playback . TranscodeRoute ) error
ApplyReplacement ( sessionID string , replacement playback . SessionReplacement ) ( playback . SessionReplacementRollback , error )
ApplyReplacementIfRoute ( sessionID string , expected playback . TranscodeRoute , replacement playback . SessionReplacement ) ( playback . SessionReplacementRollback , bool , error )
RollbackReplacement ( sessionID string , rollback playback . SessionReplacementRollback ) error
2026-05-22 20:26:11 -04:00
SetWebSocket ( sessionID string , connected bool ) error
2026-06-09 03:55:35 +02:00
SetRealtimeConnection ( sessionID string , connected bool ) error
SetProgressPersistenceDisabled ( sessionID string , disabled bool ) error
2026-05-22 20:26:11 -04:00
StopSession ( sessionID string ) error
GetSession ( sessionID string ) ( * playback . Session , error )
}
type sessionStarterWithFilesContext interface {
StartSessionWithFilesContext ( ctx context . Context , userID int , profileID string , effectiveFileID int , requestedFileID int , method playback . PlayMethod , transcodeAudio bool ) ( * playback . Session , error )
}
2026-07-11 04:30:04 +02:00
type transcodePermissionChecker interface {
CheckTranscodingAllowed ( ctx context . Context , userID int , requiresVideoTranscode bool ) error
}
func ( h * PlaybackHandler ) ensureUserTranscodingAllowed ( w http . ResponseWriter , r * http . Request , userID int , requiresVideoTranscode bool ) bool {
checker , ok := h . sessionMgr .( transcodePermissionChecker )
if ! ok {
return true
}
if err := checker . CheckTranscodingAllowed ( r . Context (), userID , requiresVideoTranscode ); err != nil {
if errors . Is ( err , playback . ErrTranscodingDisabled ) {
writeError ( w , http . StatusForbidden , "transcoding_disabled" , "Transcoding is disabled for your user" )
return false
}
if errors . Is ( err , playback . ErrAudioTranscodingDisabled ) {
writeError ( w , http . StatusForbidden , "audio_transcoding_disabled" , "Audio transcoding is disabled for your user" )
return false
}
writeError ( w , http . StatusInternalServerError , "internal_error" , "Failed to verify transcoding access" )
return false
}
return true
}
2026-05-22 20:26:11 -04:00
type PlaybackItemAccessChecker interface {
EnsureAccessible ( ctx context . Context , contentID string , filter catalog . AccessFilter ) error
}
type PlaybackEpisodeLookup interface {
GetByID ( ctx context . Context , contentID string ) ( * models . Episode , error )
}
2026-07-06 18:52:44 -04:00
// PlaybackExtraLookup resolves local extras (media_extras) so their files
// authorize through the parent item, like episodes authorize through their
// series.
type PlaybackExtraLookup interface {
GetByID ( ctx context . Context , contentID string ) ( * models . MediaExtra , error )
}
2026-05-22 20:26:11 -04:00
type PlaybackSessionSyncer interface {
SyncNow ( ctx context . Context ) error
}
// PlaybackSettingsReader reads server settings for playback decisions.
type PlaybackSettingsReader interface {
Get ( ctx context . Context , key string ) ( string , error )
}
// PlaybackFileVersionFetcher retrieves alternate file versions for a content item.
type PlaybackFileVersionFetcher interface {
GetByContentID ( ctx context . Context , contentID string ) ([] * models . MediaFile , error )
GetByEpisodeID ( ctx context . Context , episodeID string ) ([] * models . MediaFile , error )
}
type PlaybackProbeEnsurer interface {
Ensure ( ctx context . Context , file * models . MediaFile ) ( * models . MediaFile , error )
}
type PlaybackChapterThumbnailQueuer interface {
QueuePriorityFileAtPosition ( ctx context . Context , fileID int , targetSeconds float64 )
}
// PlaybackOriginalLanguageLookup fetches the original language for a content item.
type PlaybackOriginalLanguageLookup interface {
GetOriginalLanguage ( ctx context . Context , contentID string ) ( string , error )
}
// PlaybackHandler handles playback session HTTP endpoints.
type PlaybackHandler struct {
sessionMgr SessionManagerInterface
fileResolver FilePathResolver // optional; enables stream_url in responses
StoreProvider userstore . UserStoreProvider // optional; enables progress/history persistence
WatchScrobbler PlaybackWatchScrobbler
StableIdentityResolver * watchstate . StableIdentityResolver
2026-06-26 16:05:53 -04:00
CompletionObserver watchstate . CompletionObserver // optional; auto-removes watched items from the watchlist
2026-05-22 20:26:11 -04:00
profileStaler ProfileStaler
profileRefreshRequester ProfileRefreshRequester
AdminStore PlaybackAdminStore // optional; enables admin playback history/live session cleanup
SessionSyncer PlaybackSessionSyncer // optional; enables immediate session sync to shared admin view
EventsHub * evt . Hub
MissingMarker MissingFileMarker
2026-06-10 17:18:18 -04:00
NodePlanner nodepool . SessionPlanner // optional; enables proxy/transcode node selection
2026-05-22 20:26:11 -04:00
JWTSecret string // needed for signing stream tokens
ItemAccess PlaybackItemAccessChecker // optional; enables file authorization checks
EpisodeLookup PlaybackEpisodeLookup // optional; resolves episode files to their series
2026-07-06 18:52:44 -04:00
ExtraLookup PlaybackExtraLookup // optional; resolves extras files to their parent item
2026-05-22 20:26:11 -04:00
OriginalLangLookup PlaybackOriginalLanguageLookup
SettingsRepo PlaybackSettingsReader // optional; reads server settings (e.g., allow_4k_transcode)
FileVersionFetcher PlaybackFileVersionFetcher // optional; queries sibling file versions for 4K guard
ProbeEnsurer PlaybackProbeEnsurer // optional; repairs missing probe metadata on demand
ChapterThumbnailQueuer PlaybackChapterThumbnailQueuer
IntroAnalyzer IntroEpisodeAnalyzer
IntroRepository PlaybackIntroEligibilityChecker
MarkerRegistry * markers . Registry
2026-05-23 23:48:25 -04:00
MarkerResolver markers . ExternalIDResolver
MarkerUpserter PlaybackMarkerUpserter
2026-05-22 20:26:11 -04:00
MarkerUpdateNotifier PlaybackMarkerUpdateNotifier
MarkerLazyContext context . Context
MarkerLazyInFlight sync . Map
SubtitleRepo subtitles . Repository // optional; enables downloaded subtitles in playback
RealtimeHub * playback . RealtimeHub
CommandTracker * playback . CommandTracker
CommandDispatcher * playback . CommandDispatcher
2026-06-10 19:25:07 -04:00
// PlaybackConfig returns the current playback config (ffmpeg path,
// hwaccel, transcode dir). Wired to the live config in integrated mode
// so admin changes apply to newly started transcodes. Read it through
// playbackConfig(), which falls back to defaults when unset.
2026-06-23 14:26:14 -04:00
PlaybackConfig func () config . PlaybackConfig
FFmpegLogSink playback . FFmpegLogSink
realtimeCommandMu sync . Mutex
realtimeCommands map [ string ] playbackCommandRecord
2026-07-03 02:23:14 +08:00
// tm owns the transcode-session lifecycle (live map, recipe cards, and
// restart reconstruct) shared with the jellycompat handler. The handler
// delegates all transcode-session and recipe operations to it.
tm * playback . TranscodeManager
2026-07-14 11:51:27 -04:00
// PlanStoreV3 owns the short-lived protocol-v3 control-plane state. Router
// wiring replaces the in-memory default with PostgreSQL in integrated mode.
PlanStoreV3 playback . PlanStoreV3
v3RegistryOnce sync . Once
v3Registry * playback . TransformationRegistryV3
v3NodeCapabilitiesMu sync . Mutex
v3NodeCapabilities map [ string ] v3NodeCapabilityCache
v3EventOnce sync . Once
v3EventQueue chan playback . RouteEventRecordV3
v3ReplanMu sync . Mutex
v3ReplanLocks map [ string ] * v3ReplanLock
v3ReplanSlotsOnce sync . Once
v3ReplanSlots chan struct {}
v3EventRateMu sync . Mutex
v3EventRates map [ string ] v3EventRate
2026-05-22 20:26:11 -04:00
}
type PlaybackWatchScrobbler interface {
ScrobbleStart ( ctx context . Context , event watchsync . ScrobbleEvent ) error
ScrobblePause ( ctx context . Context , event watchsync . ScrobbleEvent ) error
ScrobbleStop ( ctx context . Context , event watchsync . ScrobbleEvent ) error
}
type sessionExpirationHookSetter interface {
SetExpirationHook ( func ( * playback . Session ))
}
// NewPlaybackHandler creates a new PlaybackHandler backed by the given
// session manager. Pass optional FilePathResolver to enable stream_url
// and subtitle_urls in start playback responses.
func NewPlaybackHandler ( sessionMgr SessionManagerInterface , opts ... FilePathResolver ) * PlaybackHandler {
h := & PlaybackHandler {
sessionMgr : sessionMgr ,
realtimeCommands : make ( map [ string ] playbackCommandRecord ),
2026-07-03 02:23:14 +08:00
tm : playback . NewTranscodeManager (),
2026-07-14 11:51:27 -04:00
PlanStoreV3 : playback . NewMemoryPlanStoreV3 (),
2026-05-22 20:26:11 -04:00
}
if len ( opts ) > 0 {
h . fileResolver = opts [ 0 ]
}
2026-07-03 02:23:14 +08:00
// Wire the shared transcode manager with closures so it reads the handler's
// (often late-set) config/store/secret fields lazily at call time, avoiding a
// field-ordering hazard during router setup.
h . tm . JWTSecretFn = func () string { return h . JWTSecret }
h . tm . LogSinkFn = func () playback . FFmpegLogSink { return h . FFmpegLogSink }
h . tm . Config = func () playback . TranscodeRuntimeConfig {
c := h . playbackConfig ()
return playback . TranscodeRuntimeConfig {
2026-08-11 13:42:43 -04:00
TranscodeDir : c . TranscodeDir ,
FFmpegPath : c . FFmpegPath ,
HWAccel : c . HWAccel ,
HWDevice : c . HWDevice ,
SegmentRetentionSeconds : c . SegmentRetentionSeconds ,
2026-07-03 02:23:14 +08:00
}
}
h . tm . StartThrottler = func ( ctx context . Context , ts * playback . TranscodeSession ) {
h . maybeStartThrottler ( ctx , ts )
}
h . tm . OnFFmpegCrash = func ( ctx context . Context , sessionID string , dead * playback . TranscodeSession ) {
// ffmpeg crash — tear the session down; a client holding a valid stream
// token can reconstruct it on the next request.
//
// Compare-and-delete the dead transcode first: between ffmpeg's error exit
// and this teardown a reconstruct may have registered a fresh successor
// under the same id. CloseTranscodeSessionIf only removes (and Close()s, which
// reaps the shared output dir) the entry when it is still the dead session;
// if a successor won, it leaves the live one untouched and we must NOT tear
// down the reconstructed playback session that now backs it.
var nodeURL string
if s , err := h . sessionMgr . GetSession ( sessionID ); err == nil {
nodeURL = s . TranscodeNodeURL
}
if successor := h . tm . GetTranscodeSession ( sessionID ); successor != nil && successor != dead {
// A reconstruct already replaced the crashed process; the live successor
// and its session stand. Cheap fast-path only — the authoritative gate is
// the compare-and-delete result below.
return
}
// CloseTranscodeSessionIf is the authoritative gate: a successor may register
// under the same id between the pre-check above and here. We only tear down the
// upstream playback session when the compare-and-delete actually matched the
// dead transcode. When it returns false a successor owns the session — do
// nothing further, or finalizeSessionStop's unconditional CloseTranscodeSession
// would reap the live successor's output dir mid-serve.
if ! h . tm . CloseTranscodeSessionIf ( sessionID , dead , nodeURL ) {
return
}
if err := h . stopPlaybackSessionByID ( ctx , sessionID , false ); err != nil && ! errors . Is ( err , playback . ErrSessionNotFound ) {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( ctx , "failed to stop playback after local transcode exit" , "component" , "api" , "session" , sessionID , "error" , err , "playback_session_id" , sessionID )
2026-07-03 02:23:14 +08:00
}
}
if reg , ok := sessionMgr .( interface {
RegisterReconstructed ( s * playback . Session ) * playback . Session
RegisterReconstructedWithLimits ( ctx context . Context , s * playback . Session ) ( * playback . Session , error )
}); ok {
h . tm . Sessions = reg
}
2026-05-22 20:26:11 -04:00
if setter , ok := sessionMgr .( sessionExpirationHookSetter ); ok {
setter . SetExpirationHook ( h . handleExpiredSession )
}
return h
}
2026-07-03 02:23:14 +08:00
// TranscodeManager returns the shared transcode/reconstruct manager so sibling
// handlers (e.g. StreamHandler) can reuse the same recipe-card store, live
// transcode map, and reconstruct front door rather than wiring a second one.
func ( h * PlaybackHandler ) TranscodeManager () * playback . TranscodeManager {
return h . tm
}
2026-05-22 20:26:11 -04:00
// SetProfileStaler configures an optional staleness trigger for taste profiles.
func ( h * PlaybackHandler ) SetProfileStaler ( ps ProfileStaler ) {
h . profileStaler = ps
}
// SetProfileRefreshRequester configures an optional background refresh queue for taste profiles.
func ( h * PlaybackHandler ) SetProfileRefreshRequester ( requester ProfileRefreshRequester ) {
h . profileRefreshRequester = requester
}
2026-06-10 19:25:07 -04:00
// playbackConfig returns the current playback config, falling back to the
// same defaults as config loading (transcode enabled, temp transcode dir)
// when no provider is wired (tests, minimal setups).
func ( h * PlaybackHandler ) playbackConfig () config . PlaybackConfig {
if h . PlaybackConfig != nil {
return h . PlaybackConfig ()
}
return config . PlaybackConfig {
TranscodeEnabled : true ,
TranscodeDir : filepath . Join ( os . TempDir (), "silo-transcode" ),
}
}
2026-05-22 20:26:11 -04:00
// CleanupOrphanedTranscodes removes stale per-session temp directories for
2026-07-03 02:23:14 +08:00
// transcodes that are no longer tracked in memory, sparing dirs whose recipe
// card still exists. Delegates to the shared transcode manager.
2026-05-22 20:26:11 -04:00
func ( h * PlaybackHandler ) CleanupOrphanedTranscodes () ( int , error ) {
2026-07-03 02:23:14 +08:00
return h . tm . CleanupOrphanedTranscodes ()
2026-05-22 20:26:11 -04:00
}
// playbackThresholds reads the playback.watched_threshold and
// playback.min_resume_threshold settings. Zero values mean "use defaults".
func ( h * PlaybackHandler ) playbackThresholds ( ctx context . Context ) userstore . ProgressThresholds {
if h . SettingsRepo == nil {
return userstore . ProgressThresholds {}
}
var t userstore . ProgressThresholds
if v , _ := h . SettingsRepo . Get ( ctx , "playback.watched_threshold" ); v != "" {
if pct , err := strconv . Atoi ( v ); err == nil && pct > 0 {
t . WatchedPct = pct
}
}
if v , _ := h . SettingsRepo . Get ( ctx , "playback.min_resume_threshold" ); v != "" {
if pct , err := strconv . Atoi ( v ); err == nil && pct > 0 {
t . MinResumePct = pct
}
}
return t
}
// --- Request/Response types ---
// progressRequest represents the JSON body for POST /playback/{session_id}/progress.
type progressRequest struct {
Position float64 `json:"position"`
IsPaused bool `json:"is_paused"`
}
func semanticPlayMethod ( s * playback . Session ) playback . PlayMethod {
if s == nil {
return ""
}
if s . BasePlayMethod != "" {
return s . BasePlayMethod
}
return s . PlayMethod
}
func ( h * PlaybackHandler ) ensurePlaybackProbe ( ctx context . Context , file * models . MediaFile ) * models . MediaFile {
if h == nil || h . ProbeEnsurer == nil || file == nil {
return file
}
repaired , err := h . ProbeEnsurer . Ensure ( ctx , file )
if err != nil {
2026-07-09 20:53:52 +08:00
slog . WarnContext ( ctx , "playback probe repair failed" , "component" , "api" , "file_id" , file . ID , "path" , file . FilePath , "error" , err )
2026-05-22 20:26:11 -04:00
return file
}
if repaired != nil {
return repaired
}
return file
}
2026-07-03 02:23:14 +08:00
// streamTokenParam is the query parameter that carries the signed stream token
// on the native integrated serve path. The token is the durable reconstruction
// descriptor: a front-end that lost its in-memory session rebuilds from it. It
// rides a query parameter (not a path segment) because the integrated server is
// hit directly by the client — there is no query-stripping proxy hop in between,
// and the transcode manifest rewriter already appends the request RawQuery to
// every segment URI, so segment requests inherit the token for free. The
// proxy/node path keeps the token in the URL path (see the proxy server).
const streamTokenParam = "st"
// signSessionToken mints a stream token carrying the session's full
// reconstruction recipe. Returns "" when no signing secret is configured
// (reconstruct effectively disabled, e.g. in tests).
func ( h * PlaybackHandler ) signSessionToken ( card playback . RecipeCard ) string {
if h . JWTSecret == "" {
return ""
}
token , err := streamtoken . Sign ( card . ToClaims (), h . JWTSecret , playback . MaxTokenTTL )
if err != nil {
slog . Warn ( "sign stream token failed" , "error" , err , "session" , card . SessionID , "playback_session_id" , card . SessionID )
return ""
}
return token
}
// streamCardFromQuery verifies the stream token in the request's ?st= parameter
// and returns the decoded reconstruction recipe, or nil when the token is
// absent, invalid/expired, or bound to a different session. A live session needs
// no token (the result is simply nil); the recipe is consumed only on
// reconstruct.
func ( h * PlaybackHandler ) streamCardFromQuery ( r * http . Request , sessionID string ) * playback . RecipeCard {
return streamCardFromToken ( r . URL . Query (). Get ( streamTokenParam ), sessionID , h . JWTSecret )
}
// loadTranscodeServeSession resolves the playback Session for the transcode
// manifest/segment serve routes while keeping stream-token verification off the
// hot path. The overwhelmingly common case is a live in-memory session, which
// needs no token at all, so the cheap GetSession lookup runs first and the
// (HMAC + JSON) token decode is performed only on a not-found miss where a
// reconstruct is actually required. On that miss it delegates to the shared
// LoadOrReconstructSession front door so reconstruct/ownership semantics stay
// identical. The returned card (nil on the live-session path) is the decoded
// recipe the caller's own reconstruct branch consumes.
func ( h * PlaybackHandler ) loadTranscodeServeSession ( r * http . Request , sessionID string ) ( * playback . Session , playback . SessionLoadStatus , * playback . RecipeCard ) {
requestUserID := apimw . GetUserID ( r . Context ())
session , err := h . sessionMgr . GetSession ( sessionID )
if err == nil {
// Live session: enforce the same ownership rule as LoadOrReconstructSession
// (a zero caller is allowed; a non-zero mismatch is refused). No token
// verification on this hot path.
if requestUserID != 0 && session . UserID != requestUserID {
return nil , playback . SessionForbidden , nil
}
return session , playback . SessionLoaded , nil
}
if ! errors . Is ( err , playback . ErrSessionNotFound ) {
return nil , playback . SessionLoadFailed , nil
}
// Genuine miss (e.g. after a restart): now — and only now — pay for the token
// decode so the recipe is available for reconstruction.
card := h . streamCardFromQuery ( r , sessionID )
session , status := h . tm . LoadOrReconstructSession ( r . Context (), h . sessionMgr . GetSession , sessionID , requestUserID , card )
return session , status , card
}
// streamCardFromToken verifies a stream token and decodes its reconstruction
// recipe, returning nil when the token is absent, unparseable/expired, or bound
// to a different session id. Shared by the native serve handlers (PlaybackHandler
// and StreamHandler).
func streamCardFromToken ( tokenStr , sessionID , secret string ) * playback . RecipeCard {
if tokenStr == "" || secret == "" {
return nil
}
claims , err := streamtoken . Verify ( tokenStr , secret )
if err != nil || claims . SessionID != sessionID {
return nil
}
card := playback . RecipeCardFromClaims ( claims )
return & card
}
// appendStreamToken adds the ?st=<token> parameter to a native serve URL.
func appendStreamToken ( rawURL , token string ) string {
if token == "" {
return rawURL
}
sep := "?"
if strings . ContainsRune ( rawURL , '?' ) {
sep = "&"
}
return rawURL + sep + streamTokenParam + "=" + token
}
// playbackStreamURL builds the native serve URL for a session and appends an
// identity stream token so a direct-play/remux session survives a restart (the
2026-08-10 18:14:49 -04:00
// client re-supplies its byte position). Transcode sessions are told which URL
// to play by their v3 plan; the URL here is an informational placeholder that
// the plan's delivery URL supersedes.
2026-07-03 02:23:14 +08:00
func ( h * PlaybackHandler ) playbackStreamURL ( s * playback . Session ) string {
2026-05-22 20:26:11 -04:00
if s == nil {
return ""
}
if s . PlayMethod == playback . PlayTranscode {
return fmt . Sprintf ( "/playback/transcode/%s/master.m3u8" , s . ID )
}
2026-07-03 02:23:14 +08:00
card := identityRecipeCard ( s )
return appendStreamToken ( fmt . Sprintf ( "/stream/%s" , s . ID ), h . signSessionToken ( card ))
}
// identityRecipeCard builds the identity-only recipe for a direct-play or remux
// session: reconstruction needs only ownership plus the audio selection, since
// the bytes are served by HTTP Range / a re-spawned remux pipe at the
// client-supplied position.
func identityRecipeCard ( s * playback . Session ) playback . RecipeCard {
switch s . PlayMethod {
case playback . PlayRemux :
2026-08-10 18:14:49 -04:00
card := playback . NewRemuxRecipeCard ( s . ID , s . UserID , s . ProfileID , s . MediaFileID , s . TranscodeAudio , s . AudioTrackIndex , s . RemuxDVMode )
card . TargetCodecAudio = s . TargetAudioCodec
card . TargetAudioChannels = s . TargetAudioChannels
card . TargetAudioBitrateKbps = s . TargetAudioBitrateKbps
return card
2026-07-03 02:23:14 +08:00
default :
return playback . NewDirectRecipeCard ( s . ID , s . UserID , s . ProfileID , s . MediaFileID )
}
2026-05-22 20:26:11 -04:00
}
2026-06-10 17:18:18 -04:00
func fileBitrateKbps ( file * models . MediaFile ) int {
if file == nil || file . Bitrate <= 0 {
return 0
}
return file . Bitrate
}
2026-05-22 20:26:11 -04:00
func requestedMediaFileID ( session * playback . Session ) int {
if session == nil {
return 0
}
if session . RequestedMediaFileID > 0 {
return session . RequestedMediaFileID
}
return session . MediaFileID
}
2026-07-14 11:51:27 -04:00
func remoteTransportID ( session * playback . Session ) string {
if session != nil && session . TranscodeTransportID != "" {
return session . TranscodeTransportID
}
if session == nil {
return ""
}
return session . ID
}
func ( h * PlaybackHandler ) closeTranscodeForSession ( session * playback . Session ) {
if session == nil {
return
}
// Local sessions remain keyed by the public playback session. Remote v3
// processes use a plan-scoped transport identity so a prepared successor can
// coexist with its predecessor until commit.
h . tm . CloseTranscodeSession ( session . ID , "" )
if session . TranscodeNodeURL != "" {
h . tm . StopRemoteTranscode ( remoteTransportID ( session ), session . TranscodeNodeURL )
}
}
2026-05-22 20:26:11 -04:00
func ( h * PlaybackHandler ) loadFileByPreferredID (
ctx context . Context ,
preferredID int ,
fallbackID int ,
) ( * models . MediaFile , error ) {
if h . fileResolver == nil {
return nil , fmt . Errorf ( "file resolver not configured" )
}
if preferredID > 0 {
file , err := h . fileResolver . GetByID ( ctx , preferredID )
if err == nil && file != nil {
return file , nil
}
if err != nil && ( fallbackID == 0 || fallbackID == preferredID ) {
return nil , err
}
}
if fallbackID > 0 && fallbackID != preferredID {
return h . fileResolver . GetByID ( ctx , fallbackID )
}
return nil , nil
}
func directPlayAudioTrackIndex ( file * models . MediaFile ) int {
if file == nil || len ( file . AudioTracks ) == 0 {
return 0
}
for i , track := range file . AudioTracks {
if track . Default {
return i
}
}
return 0
}
func normalizeAudioTrackIndex ( file * models . MediaFile , audioTrackIndex int ) int {
if file == nil || len ( file . AudioTracks ) == 0 {
return 0
}
if audioTrackIndex >= 0 && audioTrackIndex < len ( file . AudioTracks ) {
return audioTrackIndex
}
return directPlayAudioTrackIndex ( file )
}
func ( h * PlaybackHandler ) resolveSeriesID ( ctx context . Context , file * models . MediaFile ) string {
if file . EpisodeID == "" || h . EpisodeLookup == nil {
return ""
}
ep , err := h . EpisodeLookup . GetByID ( ctx , file . EpisodeID )
if err != nil || ep == nil {
return ""
}
return ep . SeriesID
}
// resolveOriginalLanguage fetches the original language for a media file's content item.
// For episodes, it looks up the parent series. Returns empty string if unavailable.
func ( h * PlaybackHandler ) resolveOriginalLanguage ( ctx context . Context , file * models . MediaFile ) string {
if h . OriginalLangLookup == nil {
return ""
}
contentID := file . ContentID
if file . EpisodeID != "" {
contentID = h . resolveSeriesID ( ctx , file )
}
if contentID == "" {
return ""
}
lang , err := h . OriginalLangLookup . GetOriginalLanguage ( ctx , contentID )
if err != nil {
return ""
}
return lang
}
2026-07-30 10:52:41 -04:00
// resolvedProfileAudioLanguage returns the effective playback.audio_language
// for the profile with no content context, resolved through the settings
// contract — the canonical replacement for reading the legacy
// user_profiles.language column, matching catalog's detail resolution. It may
// return playback.OriginalLanguageSentinel, which the caller resolves to a
// concrete language. Returns "" when nothing is stored: the contract default
// is null, "no preference".
func resolvedProfileAudioLanguage ( ctx context . Context , store userstore . UserStore , profileID string ) string {
if store == nil || profileID == "" {
return ""
}
contract , err := settingscontract . Load ()
if err != nil {
return ""
}
resolved , err := settingsresolve . New ( contract ). Resolve ( ctx , store ,
settingsresolve . Context { ProfileID : profileID },
[] string { settingskeys . PlaybackAudioLanguage }, nil )
if err != nil || len ( resolved ) == 0 {
return ""
}
var language string
if json . Unmarshal ( resolved [ 0 ]. Value , & language ) != nil {
return ""
}
return strings . TrimSpace ( language )
}
2026-05-22 20:26:11 -04:00
// --- Persistence helpers ---
// persistProgress saves the current playback position to the UserStore.
// It resolves the mediaFileID to a mediaItemID via the file resolver.
// Errors are logged but do not fail the HTTP request.
func ( h * PlaybackHandler ) persistProgress ( ctx context . Context , session * playback . Session ) {
if h . StoreProvider == nil || h . fileResolver == nil {
return
}
2026-06-09 03:55:35 +02:00
if session == nil || session . DisableProgressPersistence {
return
}
2026-07-07 20:38:44 +02:00
// Position 0 carries no resume information (mirrors persistStopAndHistory
// and the jellycompat report path). Progress is last-write-wins, so an
// early zero heartbeat — e.g. before a client finishes seeking to its
// resume point — must not wipe the stored resume position.
if session . Position <= 0 {
return
}
2026-05-22 20:26:11 -04:00
file , err := h . loadFileByPreferredID ( ctx , requestedMediaFileID ( session ), session . MediaFileID )
targetID := playbackProgressTarget ( file )
if err != nil || targetID == "" {
return // file not found or not yet matched to a media item
}
store , err := h . StoreProvider . ForUser ( ctx , session . UserID )
if err != nil {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( ctx , "failed to get user store" , "component" , "api" , "user_id" , session . UserID , "error" , err )
2026-05-22 20:26:11 -04:00
return
}
duration := float64 ( file . Duration )
if err := store . UpdateProgress ( ctx , session . ProfileID , targetID , session . Position , duration , h . playbackThresholds ( ctx )); err != nil {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( ctx , "failed to persist progress" , "component" , "api" , "session" , session . ID , "error" , err )
2026-05-22 20:26:11 -04:00
} else {
triggerProfileRefresh ( ctx , h . profileStaler , h . profileRefreshRequester , session . UserID , session . ProfileID )
}
if err := store . UpdateProgressHints ( ctx , session . ProfileID , targetID , userstore . VersionHints {
FileID : file . ID ,
Resolution : file . Resolution ,
HDR : file . HDR ,
CodecVideo : file . CodecVideo ,
EditionKey : file . EditionKey ,
}); err != nil {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( ctx , "failed to persist version hints" , "component" , "api" , "session" , session . ID , "error" , err )
2026-05-22 20:26:11 -04:00
}
}
// persistStopAndHistory saves the final position and adds a watch history entry
// when a playback session is stopped. Errors are logged but do not fail the
// HTTP request.
func ( h * PlaybackHandler ) persistStopAndHistory ( ctx context . Context , session * playback . Session ) watchstate . PlaybackStopResult {
if h . StoreProvider == nil || h . fileResolver == nil {
return watchstate . PlaybackStopResult {}
}
2026-06-09 03:55:35 +02:00
if session == nil || session . DisableProgressPersistence || session . Position <= 0 {
2026-05-22 20:26:11 -04:00
return watchstate . PlaybackStopResult {}
}
file , err := h . loadFileByPreferredID ( ctx , requestedMediaFileID ( session ), session . MediaFileID )
targetID := playbackProgressTarget ( file )
if err != nil || targetID == "" {
return watchstate . PlaybackStopResult {}
}
duration := float64 ( file . Duration )
thresholds := h . playbackThresholds ( ctx )
2026-06-26 16:05:53 -04:00
watchSvc := watchstate . NewService ( h . StoreProvider ).
WithStableIdentityResolver ( h . StableIdentityResolver ).
WithCompletionObserver ( h . CompletionObserver )
2026-05-22 20:26:11 -04:00
stoppedAt := time . Now (). UTC ()
result , err := watchSvc . RecordPlaybackStop ( ctx , session . UserID , session . ProfileID , targetID , duration , session . Position , stoppedAt , userstore . VersionHints {
FileID : file . ID ,
Resolution : file . Resolution ,
HDR : file . HDR ,
CodecVideo : file . CodecVideo ,
EditionKey : file . EditionKey ,
}, thresholds )
if err != nil {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( ctx , "failed to persist playback stop" , "component" , "api" , "session" , session . ID , "error" , err )
2026-05-22 20:26:11 -04:00
} else {
triggerProfileRefresh ( ctx , h . profileStaler , h . profileRefreshRequester , session . UserID , session . ProfileID )
}
return result
}
func ( h * PlaybackHandler ) scrobbleEventForSession ( ctx context . Context , session * playback . Session , mediaItemID string , duration , position float64 ) watchsync . ScrobbleEvent {
event := watchsync . ScrobbleEvent {
PlaybackSessionID : session . ID ,
UserID : session . UserID ,
ProfileID : session . ProfileID ,
MediaItemID : mediaItemID ,
PositionSeconds : position ,
DurationSeconds : duration ,
OccurredAt : time . Now (). UTC (),
}
2026-07-22 21:41:05 -04:00
return watchsync . ResolveScrobbleIdentity ( ctx , h . StableIdentityResolver , event )
2026-05-22 20:26:11 -04:00
}
2026-07-20 17:14:42 -04:00
func ( h * PlaybackHandler ) scrobbleEventForStoppedSession (
ctx context . Context ,
session * playback . Session ,
stopResult watchstate . PlaybackStopResult ,
) ( watchsync . ScrobbleEvent , bool ) {
if session == nil || session . DisableProgressPersistence {
return watchsync . ScrobbleEvent {}, false
}
mediaItemID := stopResult . MediaItemID
duration := stopResult . DurationSeconds
position := stopResult . FinalPositionSeconds
if mediaItemID == "" {
if h . fileResolver == nil {
return watchsync . ScrobbleEvent {}, false
}
file , err := h . loadFileByPreferredID ( ctx , requestedMediaFileID ( session ), session . MediaFileID )
if err != nil || file == nil {
return watchsync . ScrobbleEvent {}, false
}
mediaItemID = playbackProgressTarget ( file )
if mediaItemID == "" {
return watchsync . ScrobbleEvent {}, false
}
duration = float64 ( file . Duration )
position = session . Position
}
event := h . scrobbleEventForSession ( ctx , session , mediaItemID , duration , position )
event . HistoryID = stopResult . HistoryID
event . Completed = stopResult . Completed
return event , true
}
2026-05-22 20:26:11 -04:00
func ( h * PlaybackHandler ) buildAdminHistoryEntry (
ctx context . Context ,
session * playback . Session ,
) ( * AdminPlaybackHistoryEntry , error ) {
if h . AdminStore == nil || h . fileResolver == nil || session == nil {
return nil , nil
}
file , err := h . loadFileByPreferredID ( ctx , requestedMediaFileID ( session ), session . MediaFileID )
if err != nil {
return nil , fmt . Errorf ( "loading media file: %w" , err )
}
targetID := playbackProgressTarget ( file )
profileName := session . ProfileID
if h . StoreProvider != nil {
store , storeErr := h . StoreProvider . ForUser ( ctx , session . UserID )
if storeErr != nil {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( ctx , "failed to get user store for admin history" , "component" , "api" , "session" , session . ID , "error" , storeErr )
2026-05-22 20:26:11 -04:00
} else if store != nil {
profile , profileErr := store . GetProfile ( ctx , session . ProfileID )
if profileErr != nil {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( ctx , "failed to load profile for admin history" , "component" , "api" , "session" , session . ID , "error" , profileErr )
2026-05-22 20:26:11 -04:00
} else if profile != nil && strings . TrimSpace ( profile . Name ) != "" {
profileName = profile . Name
}
}
}
var durationPtr * float64
completed := false
if file != nil {
duration := float64 ( file . Duration )
durationPtr = & duration
if duration > 0 && session . Position / duration > userstore . WatchedFraction ( h . playbackThresholds ( ctx ). WatchedPct ) {
completed = true
}
}
entry := & AdminPlaybackHistoryEntry {
SessionID : session . ID ,
UserID : session . UserID ,
ProfileID : session . ProfileID ,
ProfileName : profileName ,
MediaItemID : targetID ,
MediaFileID : requestedMediaFileID ( session ),
PlayMethod : string ( semanticPlayMethod ( session )),
StartedAt : session . StartedAt . UTC (). Format ( time . RFC3339Nano ),
EndedAt : time . Now (). UTC (). Format ( time . RFC3339Nano ),
WatchedSeconds : session . Position ,
DurationSeconds : durationPtr ,
Completed : completed ,
ClientIP : clientip . FromContext ( ctx ),
}
return entry , nil
}
func ( h * PlaybackHandler ) syncSessionsNow ( ctx context . Context , reason string ) {
if h . SessionSyncer == nil {
return
}
if err := h . SessionSyncer . SyncNow ( ctx ); err != nil {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( ctx , "failed to sync sessions" , "component" , "api" , "reason" , reason , "error" , err )
2026-05-22 20:26:11 -04:00
}
}
func ( h * PlaybackHandler ) touchSessionActivity ( sessionID string ) {
if h == nil || sessionID == "" {
return
}
if err := h . sessionMgr . TouchActivity ( sessionID ); err != nil && ! errors . Is ( err , playback . ErrSessionNotFound ) {
slog . Warn ( "failed to refresh playback activity" , "session" , sessionID , "error" , err , "playback_session_id" , sessionID )
}
}
2026-07-03 02:23:14 +08:00
func ( h * PlaybackHandler ) finalizeSessionStop ( ctx context . Context , session * playback . Session , syncNow bool , syncReason string , userInitiated bool ) {
2026-05-22 20:26:11 -04:00
if h == nil || session == nil || session . ID == "" {
return
}
if ctx == nil {
ctx = context . Background ()
}
stopResult := h . persistStopAndHistory ( ctx , session )
2026-07-20 17:14:42 -04:00
if h . WatchScrobbler != nil {
if event , ok := h . scrobbleEventForStoppedSession ( ctx , session , stopResult ); ok && ( userInitiated || stopResult . Completed ) {
2026-05-22 20:26:11 -04:00
if err := h . WatchScrobbler . ScrobbleStop ( ctx , event ); err != nil {
2026-07-09 20:53:52 +08:00
slog . WarnContext ( ctx , "failed to queue watch provider stop scrobble" , "component" , "api" , "session" , session . ID , "error" , err )
2026-05-22 20:26:11 -04:00
}
2026-07-20 17:14:42 -04:00
} else if ok {
2026-05-22 20:26:11 -04:00
if err := h . WatchScrobbler . ScrobblePause ( ctx , event ); err != nil {
2026-07-09 20:53:52 +08:00
slog . WarnContext ( ctx , "failed to queue watch provider pause scrobble" , "component" , "api" , "session" , session . ID , "error" , err )
2026-05-22 20:26:11 -04:00
}
}
}
if entry , buildErr := h . buildAdminHistoryEntry ( ctx , session ); buildErr != nil {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( ctx , "failed to build admin history" , "component" , "api" , "session" , session . ID , "error" , buildErr )
2026-05-22 20:26:11 -04:00
} else if entry != nil && h . AdminStore != nil {
if err := h . AdminStore . RecordHistory ( ctx , * entry ); err != nil {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( ctx , "failed to record admin history" , "component" , "api" , "session" , session . ID , "error" , err )
2026-05-22 20:26:11 -04:00
}
}
if h . AdminStore != nil {
if err := h . AdminStore . DeleteSession ( ctx , session . ID ); err != nil {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( ctx , "failed to delete synced session" , "component" , "api" , "session" , session . ID , "error" , err )
2026-05-22 20:26:11 -04:00
}
}
2026-07-14 11:51:27 -04:00
h . closeTranscodeForSession ( session )
2026-05-22 20:26:11 -04:00
if syncNow {
h . syncSessionsNow ( ctx , syncReason )
}
}
func ( h * PlaybackHandler ) finalizeSessionAbort ( ctx context . Context , session * playback . Session , syncNow bool , syncReason string ) {
if h == nil || session == nil || session . ID == "" {
return
}
if ctx == nil {
ctx = context . Background ()
}
if h . WatchScrobbler != nil && h . fileResolver != nil {
if file , err := h . loadFileByPreferredID ( ctx , requestedMediaFileID ( session ), session . MediaFileID ); err == nil && file != nil {
targetID := playbackProgressTarget ( file )
if targetID != "" {
event := h . scrobbleEventForSession ( ctx , session , targetID , float64 ( file . Duration ), session . Position )
if err := h . WatchScrobbler . ScrobblePause ( ctx , event ); err != nil {
2026-07-09 20:53:52 +08:00
slog . WarnContext ( ctx , "failed to queue watch provider abort scrobble" , "component" , "api" , "session" , session . ID , "error" , err )
2026-05-22 20:26:11 -04:00
}
}
}
}
if h . AdminStore != nil {
if err := h . AdminStore . DeleteSession ( ctx , session . ID ); err != nil {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( ctx , "failed to delete synced session" , "component" , "api" , "session" , session . ID , "error" , err )
2026-05-22 20:26:11 -04:00
}
}
2026-07-03 02:23:14 +08:00
// Abort is a connection drop / non-terminal teardown — keep the recipe card
// so the client can reconstruct on reconnect.
2026-07-14 11:51:27 -04:00
h . closeTranscodeForSession ( session )
2026-05-22 20:26:11 -04:00
if syncNow {
h . syncSessionsNow ( ctx , syncReason )
}
}
func ( h * PlaybackHandler ) handleExpiredSession ( session * playback . Session ) {
if h == nil || session == nil {
return
}
sessionCopy := * session
go func () {
slog . Info ( "expired inactive playback session" , "session" , sessionCopy . ID , "playback_session_id" , sessionCopy . ID )
2026-07-03 02:23:14 +08:00
// Expiry is a liveness reap, not a user stop — keep the recipe card so a
// resume reconstructs under the same id (the card's own TTL reaps it if
// the session is truly abandoned).
h . finalizeSessionStop ( context . Background (), & sessionCopy , false , "" , false )
2026-05-22 20:26:11 -04:00
}()
}
func playbackProgressTarget ( file * models . MediaFile ) string {
if file == nil {
return ""
}
if file . EpisodeID != "" {
return file . EpisodeID
}
return file . ContentID
}
func ( h * PlaybackHandler ) persistSeriesPlaybackPreference (
ctx context . Context ,
userID int ,
profileID string ,
file * models . MediaFile ,
) {
if h . StoreProvider == nil || file == nil {
return
}
seriesID := h . resolveSeriesID ( ctx , file )
if seriesID == "" {
return
}
store , err := h . StoreProvider . ForUser ( ctx , userID )
if err != nil {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( ctx , "failed to access user store for series playback preference" , "component" , "api" , "user_id" , userID , "error" , err )
2026-05-22 20:26:11 -04:00
return
}
if err := store . SetSeriesPlaybackPreference ( ctx , userstore . SeriesPlaybackPreference {
ProfileID : profileID ,
SeriesID : seriesID ,
Resolution : file . Resolution ,
HDR : file . HDR ,
CodecVideo : file . CodecVideo ,
}); err != nil {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( ctx , "failed to persist series playback preference" , "component" , "api" , "series_id" , seriesID , "profile_id" , profileID , "error" , err )
2026-05-22 20:26:11 -04:00
}
}
func ( h * PlaybackHandler ) persistAudioPreference (
ctx context . Context ,
userID int ,
profileID string ,
file * models . MediaFile ,
trackIndex int ,
) {
if h . StoreProvider == nil || file == nil || trackIndex < 0 || trackIndex >= len ( file . AudioTracks ) {
return
}
seriesID := h . resolveSeriesID ( ctx , file )
if seriesID == "" {
return
}
store , err := h . StoreProvider . ForUser ( ctx , userID )
if err != nil {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( ctx , "failed to access user store for audio preference" , "component" , "api" , "user_id" , userID , "error" , err )
2026-05-22 20:26:11 -04:00
return
}
track := file . AudioTracks [ trackIndex ]
if err := store . SetAudioPreference ( ctx , userstore . AudioPreference {
ProfileID : profileID ,
SeriesID : seriesID ,
AudioTrackIndex : trackIndex ,
AudioLanguage : track . Language ,
TrackSignature : playback . AudioTrackSignatureFromTrack ( track ),
}); err != nil {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( ctx , "failed to persist audio preference" , "component" , "api" , "series_id" , seriesID , "profile_id" , profileID , "error" , err )
2026-05-22 20:26:11 -04:00
}
}
// --- Handler methods ---
2026-08-10 18:14:49 -04:00
// HandleStartPlayback starts playback. Protocol v3 is the only protocol this
// endpoint speaks: a start that does not declare it comes from a build that
// predates the contract and cannot interpret a plan, so it is refused with
// 426 rather than served something it would misread.
2026-05-22 20:26:11 -04:00
func ( h * PlaybackHandler ) HandleStartPlayback ( w http . ResponseWriter , r * http . Request ) {
2026-07-14 11:51:27 -04:00
if apimw . GetUserID ( r . Context ()) == 0 {
writeError ( w , http . StatusUnauthorized , "unauthorized" , "Authentication required" )
return
}
body , err := io . ReadAll ( http . MaxBytesReader ( w , r . Body , maxPlaybackV3BodyBytes ))
if err != nil {
writeError ( w , http . StatusBadRequest , "bad_request" , "Invalid request body" )
return
}
var envelope struct {
ProtocolVersion * int `json:"protocol_version"`
2026-08-10 18:14:49 -04:00
Capabilities * struct {
VideoEvidence * string `json:"video_evidence"`
AudioEvidence * string `json:"audio_evidence"`
} `json:"client_capabilities"`
2026-07-14 11:51:27 -04:00
}
2026-08-10 18:14:49 -04:00
if err := json . Unmarshal ( body , & envelope ); err != nil {
2026-07-14 11:51:27 -04:00
writeError ( w , http . StatusBadRequest , "bad_request" , "Invalid request body" )
return
}
2026-08-10 18:14:49 -04:00
if envelope . ProtocolVersion == nil || * envelope . ProtocolVersion != playback . ProtocolV3 ||
envelope . Capabilities == nil || envelope . Capabilities . VideoEvidence == nil || envelope . Capabilities . AudioEvidence == nil {
upgrade := playback . LegacyUpgradeErrorV3 ()
writeError ( w , http . StatusUpgradeRequired , upgrade . Error , upgrade . Message )
2026-07-14 11:51:27 -04:00
return
}
2026-08-10 18:14:49 -04:00
h . handleStartPlaybackV3 ( w , r , body )
}
func playbackClientInfoFromRequest ( r * http . Request ) playback . ClientInfo {
if r == nil {
return playback . ClientInfo {}
}
return playback . ClientInfo {
Name : strings . TrimSpace ( r . Header . Get ( "X-Silo-Client" )),
Version : strings . TrimSpace ( r . Header . Get ( "X-Silo-Client-Version" )),
UserAgent : r . UserAgent (),
}
2026-07-14 11:51:27 -04:00
}
2026-08-10 18:14:49 -04:00
// HandleUpdateProgress handles POST /playback/{session_id}/progress.
func ( h * PlaybackHandler ) HandleUpdateProgress ( w http . ResponseWriter , r * http . Request ) {
2026-05-22 20:26:11 -04:00
userID := apimw . GetUserID ( r . Context ())
if userID == 0 {
writeError ( w , http . StatusUnauthorized , "unauthorized" , "Authentication required" )
return
}
2026-08-10 18:14:49 -04:00
sessionID := chi . URLParam ( r , "session_id" )
if sessionID == "" {
writeError ( w , http . StatusBadRequest , "bad_request" , "Session ID is required" )
2026-05-22 20:26:11 -04:00
return
}
2026-08-10 18:14:49 -04:00
setPlaybackSessionLogContext ( r , sessionID )
session , err := h . sessionMgr . GetSession ( sessionID )
if err != nil {
if errors . Is ( err , playback . ErrSessionNotFound ) {
writePlaybackSessionNotFound ( w )
return
}
writeError ( w , http . StatusInternalServerError , "internal_error" , "Failed to load playback session" )
2026-05-22 20:26:11 -04:00
return
}
2026-08-10 18:14:49 -04:00
if session . UserID != userID {
writeError ( w , http . StatusForbidden , "forbidden" , "Session belongs to another user" )
2026-05-22 20:26:11 -04:00
return
}
2026-08-10 18:14:49 -04:00
wasPaused := session . IsPaused
var req progressRequest
if err := json . NewDecoder ( r . Body ). Decode ( & req ); err != nil {
writeError ( w , http . StatusBadRequest , "bad_request" , "Invalid request body" )
2026-05-22 20:26:11 -04:00
return
}
2026-08-10 18:14:49 -04:00
err = h . sessionMgr . UpdateProgress ( sessionID , req . Position , req . IsPaused )
2026-05-22 20:26:11 -04:00
if err != nil {
2026-08-10 18:14:49 -04:00
if errors . Is ( err , playback . ErrSessionNotFound ) {
writePlaybackSessionNotFound ( w )
return
2026-05-22 20:26:11 -04:00
}
2026-08-10 18:14:49 -04:00
writeError ( w , http . StatusInternalServerError , "internal_error" , "Failed to update progress" )
2026-05-22 20:26:11 -04:00
return
}
2026-08-10 18:14:49 -04:00
h . syncSessionsNow ( r . Context (), "progress" )
2026-05-22 20:26:11 -04:00
2026-08-10 18:14:49 -04:00
// Persist progress to UserStore (best-effort).
if sess , getErr := h . sessionMgr . GetSession ( sessionID ); getErr == nil {
h . persistProgress ( r . Context (), sess )
if ! sess . DisableProgressPersistence && h . WatchScrobbler != nil && wasPaused != sess . IsPaused {
if file , loadErr := h . loadFileByPreferredID ( r . Context (), requestedMediaFileID ( sess ), sess . MediaFileID ); loadErr == nil && file != nil {
targetID := playbackProgressTarget ( file )
if targetID != "" {
event := h . scrobbleEventForSession ( r . Context (), sess , targetID , float64 ( file . Duration ), sess . Position )
if sess . IsPaused {
if err := h . WatchScrobbler . ScrobblePause ( r . Context (), event ); err != nil {
slog . WarnContext ( r . Context (), "failed to queue watch provider pause scrobble" , "component" , "api" , "session" , sessionID , "error" , err )
}
} else if err := h . WatchScrobbler . ScrobbleStart ( r . Context (), event ); err != nil {
slog . WarnContext ( r . Context (), "failed to queue watch provider resume scrobble" , "component" , "api" , "session" , sessionID , "error" , err )
}
2026-05-22 20:26:11 -04:00
}
}
}
}
w . WriteHeader ( http . StatusNoContent )
}
// HandleStopPlayback handles DELETE /playback/{session_id}.
func ( h * PlaybackHandler ) HandleStopPlayback ( w http . ResponseWriter , r * http . Request ) {
userID := apimw . GetUserID ( r . Context ())
if userID == 0 {
writeError ( w , http . StatusUnauthorized , "unauthorized" , "Authentication required" )
return
}
sessionID := chi . URLParam ( r , "session_id" )
if sessionID == "" {
writeError ( w , http . StatusBadRequest , "bad_request" , "Session ID is required" )
return
}
setPlaybackSessionLogContext ( r , sessionID )
session , err := h . sessionMgr . GetSession ( sessionID )
if err != nil {
if errors . Is ( err , playback . ErrSessionNotFound ) {
writePlaybackSessionNotFound ( w )
return
}
writeError ( w , http . StatusInternalServerError , "internal_error" , "Failed to load playback session" )
return
}
if session . UserID != userID {
writeError ( w , http . StatusForbidden , "forbidden" , "Session belongs to another user" )
return
}
2026-07-03 02:23:14 +08:00
err = h . stopPlaybackSession ( r . Context (), session , true )
2026-05-22 20:26:11 -04:00
if err != nil {
if errors . Is ( err , playback . ErrSessionNotFound ) {
writePlaybackSessionNotFound ( w )
return
}
writeError ( w , http . StatusInternalServerError , "internal_error" , "Failed to stop playback session" )
return
}
w . WriteHeader ( http . StatusNoContent )
}
func ( h * PlaybackHandler ) loadAuthorizedFile ( r * http . Request , fileID int ) ( * models . MediaFile , error ) {
if h . fileResolver == nil || h . ItemAccess == nil {
return nil , fmt . Errorf ( "playback authorization dependencies not configured" )
}
file , err := h . fileResolver . GetByID ( r . Context (), fileID )
if err != nil {
2026-05-26 17:57:26 -04:00
return nil , mapMediaFileLookupError ( err )
2026-05-22 20:26:11 -04:00
}
if file == nil || file . MissingSince != nil {
return nil , catalog . ErrItemNotFound
}
filter := requestAccessFilter ( r )
switch {
case file . EpisodeID != "" :
if h . EpisodeLookup == nil {
return nil , fmt . Errorf ( "episode lookup not configured" )
}
episode , err := h . EpisodeLookup . GetByID ( r . Context (), file . EpisodeID )
if err != nil {
return nil , err
}
if episode == nil {
return nil , catalog . ErrEpisodeNotFound
}
if err := h . ItemAccess . EnsureAccessible ( r . Context (), episode . SeriesID , filter ); err != nil {
return nil , err
}
case file . ContentID != "" :
if err := h . ItemAccess . EnsureAccessible ( r . Context (), file . ContentID , filter ); err != nil {
return nil , err
}
2026-07-06 18:52:44 -04:00
case file . ExtraID != "" :
if h . ExtraLookup == nil {
return nil , fmt . Errorf ( "extra lookup not configured" )
}
extra , err := h . ExtraLookup . GetByID ( r . Context (), file . ExtraID )
if err != nil {
if errors . Is ( err , catalog . ErrExtraNotFound ) {
return nil , catalog . ErrItemNotFound
}
return nil , err
}
if extra == nil {
return nil , catalog . ErrItemNotFound
}
if err := h . ItemAccess . EnsureAccessible ( r . Context (), extra . ParentID , filter ); err != nil {
return nil , err
}
2026-05-22 20:26:11 -04:00
default :
return nil , catalog . ErrItemNotFound
}
if ! catalog . FileAllowedByAccess ( file , filter ) {
return nil , catalog . ErrItemNotFound
}
return file , nil
}
// computeStartSegment returns the HLS segment number corresponding to a seek
// position given the segment duration. Both remote and local transcode paths
// use this to align ffmpeg output filenames with the VOD manifest.
func computeStartSegment ( seekSeconds float64 , segmentDuration int ) int {
if segmentDuration <= 0 {
segmentDuration = 2
}
if seekSeconds <= 0 {
return 0
}
return int ( seekSeconds / float64 ( segmentDuration ))
}
2026-07-10 05:21:26 -07:00
// alignedSeekSeconds snaps an encoded transcode's ffmpeg start position down
// to the boundary of the segment computeStartSegment assigns it. The synthetic
// VOD manifest declares segment N to begin at exactly N× segmentDuration;
// spawning ffmpeg at the raw seek position makes segment N actually begin up
// to one segment later, and hls.js aligns that content to the declared
// position — shifting the session's entire timeline (audio, video, and every
// out-of-band subtitle cue) late by seek mod segmentDuration. Copy-mode
// sessions serve ffmpeg's real manifest, whose declared timings match the
// fragments it produces, so they keep the raw seek.
func alignedSeekSeconds ( seekSeconds float64 , segmentDuration int , targetVideoCodec string ) float64 {
if strings . EqualFold ( targetVideoCodec , "copy" ) || seekSeconds <= 0 {
return seekSeconds
}
if segmentDuration <= 0 {
segmentDuration = 2
}
return float64 ( computeStartSegment ( seekSeconds , segmentDuration ) * segmentDuration )
}
2026-05-22 20:26:11 -04:00
// HandleGetTranscodeManifest handles GET /playback/transcode/{session_id}/master.m3u8.
// Auth is optional — the session UUID serves as an access token (same pattern
// as /stream/{session_id}). When auth context is present, ownership is verified.
//
2026-07-20 19:25:46 +02:00
// Known-duration encoded sessions expose a synthetic full VOD manifest so the
// player can seek immediately. Copy-video sessions expose FFmpeg's real
2026-08-10 18:14:49 -04:00
// keyframe-aligned manifest and use the resolved stream origin the v3 plan
// reports as the timeline's stream_origin_seconds.
2026-05-22 20:26:11 -04:00
func ( h * PlaybackHandler ) HandleGetTranscodeManifest ( w http . ResponseWriter , r * http . Request ) {
sessionID := chi . URLParam ( r , "session_id" )
2026-07-03 02:23:14 +08:00
session , status , card := h . loadTranscodeServeSession ( r , sessionID )
switch status {
case playback . SessionMissing :
writePlaybackSessionNotFound ( w )
return
case playback . SessionLoadFailed :
2026-05-22 20:26:11 -04:00
writeError ( w , http . StatusInternalServerError , "internal_error" , "Failed to load playback session" )
return
2026-07-03 02:23:14 +08:00
case playback . SessionForbidden :
2026-05-22 20:26:11 -04:00
writeError ( w , http . StatusForbidden , "forbidden" , "Session belongs to another user" )
return
}
2026-07-03 02:23:14 +08:00
transcodeSession := h . tm . GetTranscodeSession ( sessionID )
2026-05-22 20:26:11 -04:00
if transcodeSession == nil {
// No local session — try proxying to remote transcode node.
if session . TranscodeNodeURL != "" {
h . touchSessionActivity ( sessionID )
h . proxyToTranscodeNode ( w , r , session . TranscodeNodeURL ,
2026-07-14 11:51:27 -04:00
"/transcode/" + remoteTransportID ( session ) + "/master.m3u8" )
2026-05-22 20:26:11 -04:00
return
}
2026-07-03 02:23:14 +08:00
// Local transcode whose process state was lost: reconstruct it from the
// token recipe. The manifest path has no segment context, so pass -1 (use
// the token's seek position).
if card == nil {
writeError ( w , http . StatusNotFound , "not_found" , "Transcode session not found" )
return
}
transcodeSession = h . tm . ReconstructTranscode ( r . Context (), sessionID , - 1 , * card )
if transcodeSession == nil {
writeError ( w , http . StatusNotFound , "not_found" , "Transcode session not found" )
return
}
2026-05-22 20:26:11 -04:00
}
h . touchSessionActivity ( sessionID )
manifest , err := transcodeSession . BuildPlaybackManifest ( "segment/" , r . URL . RawQuery )
if err != nil {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( r . Context (), "build transcode manifest" , "component" , "api" , "error" , err , "session" , sessionID , "playback_session_id" , sessionID )
2026-05-22 20:26:11 -04:00
writeError ( w , http . StatusServiceUnavailable , "unavailable" , "Transcode manifest not ready" )
return
}
w . Header (). Set ( "Content-Type" , "application/vnd.apple.mpegurl" )
w . Header (). Set ( "Cache-Control" , "no-store, max-age=0" )
w . Header (). Set ( "Pragma" , "no-cache" )
w . WriteHeader ( http . StatusOK )
_ , _ = w . Write ( manifest )
}
// HandleGetTranscodeSegment handles GET /playback/transcode/{session_id}/segment/{name}.
// Auth is optional — the session UUID serves as an access token.
func ( h * PlaybackHandler ) HandleGetTranscodeSegment ( w http . ResponseWriter , r * http . Request ) {
sessionID := chi . URLParam ( r , "session_id" )
2026-07-03 02:23:14 +08:00
session , status , card := h . loadTranscodeServeSession ( r , sessionID )
switch status {
case playback . SessionMissing :
writePlaybackSessionNotFound ( w )
return
case playback . SessionLoadFailed :
2026-05-22 20:26:11 -04:00
writeError ( w , http . StatusInternalServerError , "internal_error" , "Failed to load playback session" )
return
2026-07-03 02:23:14 +08:00
case playback . SessionForbidden :
2026-05-22 20:26:11 -04:00
writeError ( w , http . StatusForbidden , "forbidden" , "Session belongs to another user" )
return
}
2026-07-03 02:23:14 +08:00
transcodeSession := h . tm . GetTranscodeSession ( sessionID )
2026-05-22 20:26:11 -04:00
if transcodeSession == nil {
if session . TranscodeNodeURL != "" {
h . touchSessionActivity ( sessionID )
segmentName := chi . URLParam ( r , "name" )
h . proxyToTranscodeNode ( w , r , session . TranscodeNodeURL ,
2026-07-14 11:51:27 -04:00
"/transcode/" + remoteTransportID ( session ) + "/segment/" + segmentName )
2026-05-22 20:26:11 -04:00
return
}
2026-07-03 02:23:14 +08:00
// Resume near the segment the client is fetching so reconstruct does not
// restart from the original seek point and stall. A non-segment name
// (e.g. init.mp4) parses as negative and falls back to the token position.
requestedSegment := - 1
if segNum , parseErr := playback . ParseSegmentNumber ( chi . URLParam ( r , "name" )); parseErr == nil {
requestedSegment = segNum
}
if card == nil {
writeError ( w , http . StatusNotFound , "not_found" , "Transcode session not found" )
return
}
transcodeSession = h . tm . ReconstructTranscode ( r . Context (), sessionID , requestedSegment , * card )
if transcodeSession == nil {
writeError ( w , http . StatusNotFound , "not_found" , "Transcode session not found" )
return
}
2026-05-22 20:26:11 -04:00
}
h . touchSessionActivity ( sessionID )
segmentName := chi . URLParam ( r , "name" )
2026-08-11 12:59:25 -04:00
segmentLease , err := transcodeSession . OpenSegment ( segmentName )
2026-05-22 20:26:11 -04:00
if err != nil && errors . Is ( err , playback . ErrSegmentNotFound ) {
segNum , parseErr := playback . ParseSegmentNumber ( segmentName )
if parseErr == nil {
now := time . Now ()
decision := transcodeSession . SegmentRecoveryDecision ( segNum , now )
lastProducedAgeMS := int64 ( - 1 )
if ! decision . Progress . LastProducedAt . IsZero () {
lastProducedAgeMS = now . Sub ( decision . Progress . LastProducedAt ). Milliseconds ()
}
2026-07-09 20:53:52 +08:00
slog . InfoContext ( r . Context (), "transcode segment missing" , "component" , "api" ,
2026-05-22 20:26:11 -04:00
"segment" , segmentName ,
"requested_segment" , segNum ,
"produced_head" , decision . Progress . ProducedHead ,
"last_requested_segment" , decision . Progress . LastRequestedSegment ,
"start_segment_number" , decision . Progress . StartSegmentNumber ,
"last_produced_age_ms" , lastProducedAgeMS ,
"wait_timeout_ms" , decision . WaitTimeout . Milliseconds (),
2026-06-07 04:30:24 +02:00
"restart_on_timeout" , decision . RestartOnTimeout ,
2026-05-22 20:26:11 -04:00
"reason" , decision . Reason ,
"session" , sessionID ,
"playback_session_id" , sessionID ,
)
if decision . Wait {
2026-07-09 20:53:52 +08:00
slog . InfoContext ( r . Context (), "transcode segment wait" , "component" , "api" ,
2026-05-22 20:26:11 -04:00
"segment" , segmentName ,
"requested_segment" , segNum ,
"produced_head" , decision . Progress . ProducedHead ,
"last_requested_segment" , decision . Progress . LastRequestedSegment ,
"start_segment_number" , decision . Progress . StartSegmentNumber ,
"last_produced_age_ms" , lastProducedAgeMS ,
"wait_timeout_ms" , decision . WaitTimeout . Milliseconds (),
2026-06-07 04:30:24 +02:00
"restart_on_timeout" , decision . RestartOnTimeout ,
2026-05-22 20:26:11 -04:00
"reason" , decision . Reason ,
"session" , sessionID ,
"playback_session_id" , sessionID ,
)
2026-08-11 12:59:25 -04:00
segmentLease , err = transcodeSession . WaitForOpenSegment ( segmentName , decision . WaitTimeout )
2026-05-22 20:26:11 -04:00
if err != nil && errors . Is ( err , playback . ErrSegmentNotFound ) {
2026-07-09 20:53:52 +08:00
slog . InfoContext ( r . Context (), "transcode segment wait timeout" , "component" , "api" ,
2026-05-22 20:26:11 -04:00
"segment" , segmentName ,
"requested_segment" , segNum ,
"produced_head" , decision . Progress . ProducedHead ,
"last_requested_segment" , decision . Progress . LastRequestedSegment ,
"start_segment_number" , decision . Progress . StartSegmentNumber ,
"last_produced_age_ms" , lastProducedAgeMS ,
"wait_timeout_ms" , decision . WaitTimeout . Milliseconds (),
2026-06-07 04:30:24 +02:00
"restart_on_timeout" , decision . RestartOnTimeout ,
2026-05-22 20:26:11 -04:00
"reason" , decision . Reason ,
"session" , sessionID ,
"playback_session_id" , sessionID ,
)
}
}
// If the segment is still missing (timed out, or outside the
// active encode range), either restart at the exact manifest-derived
// timeline position or return 404 for copy-mode segments outside the
// current manifest window.
2026-06-07 04:30:24 +02:00
if err != nil && errors . Is ( err , playback . ErrSegmentNotFound ) && decision . RestartOnTimeout {
2026-05-22 20:26:11 -04:00
seekSeconds , ok , seekErr := transcodeSession . RestartSeekTarget ( segNum )
if seekErr != nil && ! errors . Is ( seekErr , playback . ErrManifestNotReady ) {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( r . Context (), "resolve transcode seek target" , "component" , "api" , "error" , seekErr , "segment" , segmentName , "session" , sessionID , "playback_session_id" , sessionID )
2026-05-22 20:26:11 -04:00
}
2026-07-03 02:23:14 +08:00
// Copy-mode with an unresolved seek target (ok=false, no error)
// means the manifest can't place this segment yet. Don't restart
// at a fabricated position; surface ErrSegmentNotFound so the
// client retries while the session keeps producing manifest.
// Mirrors the transcode-node guard in
// internal/transcodenode/server.go.
if ! ok && seekErr == nil && transcodeSession . IsCopyVideo () {
err = playback . ErrSegmentNotFound
}
2026-05-22 20:26:11 -04:00
if ok {
2026-07-09 20:53:52 +08:00
slog . InfoContext ( r . Context (), "transcode seek restart" , "component" , "api" ,
2026-05-22 20:26:11 -04:00
"segment" , segmentName ,
"requested_segment" , segNum ,
"produced_head" , decision . Progress . ProducedHead ,
"last_requested_segment" , decision . Progress . LastRequestedSegment ,
"start_segment_number" , decision . Progress . StartSegmentNumber ,
"last_produced_age_ms" , lastProducedAgeMS ,
"wait_timeout_ms" , decision . WaitTimeout . Milliseconds (),
2026-06-07 04:30:24 +02:00
"restart_on_timeout" , decision . RestartOnTimeout ,
2026-05-22 20:26:11 -04:00
"reason" , decision . Reason ,
"seek_seconds" , seekSeconds ,
"session" , sessionID ,
"playback_session_id" , sessionID ,
)
2026-07-03 02:23:14 +08:00
if restartErr := h . tm . RestartSessionLocked (
2026-05-22 20:26:11 -04:00
context . WithoutCancel ( r . Context ()),
2026-07-03 02:23:14 +08:00
sessionID ,
transcodeSession ,
2026-05-22 20:26:11 -04:00
seekSeconds ,
segNum ,
); restartErr == nil {
2026-07-02 14:06:25 -04:00
// Throttler + exit monitor re-arm via the session's
// restart hook.
2026-08-11 12:59:25 -04:00
segmentLease , err = transcodeSession . WaitForOpenSegment ( segmentName , 30 * time . Second )
2026-05-22 20:26:11 -04:00
if err == nil && strings . EqualFold ( transcodeSession . Opts (). TargetCodecVideo , "copy" ) {
// Copy-mode seeks can resume as soon as the target segment
// exists, but that sometimes leaves the player one segment
// away from stalling while FFmpeg catches up. Briefly wait
// for a single lookahead fragment when available so the
// first resumed playback window is less brittle.
nextSegmentName := fmt . Sprintf ( "seg_%05d.m4s" , segNum + 1 )
2026-08-11 12:59:25 -04:00
if nextSegment , nextErr := transcodeSession . WaitForOpenSegment ( nextSegmentName , 1200 * time . Millisecond ); nextErr == nil {
2026-08-08 15:23:16 -04:00
_ = nextSegment . Close ()
}
2026-05-22 20:26:11 -04:00
}
}
}
}
} else if transcodeSession . IsRunning () {
// Non-numbered segment (e.g., init.mp4 for fMP4 HLS).
// Wait briefly — the init segment is written almost immediately.
2026-08-11 12:59:25 -04:00
segmentLease , err = transcodeSession . WaitForOpenSegment ( segmentName , 10 * time . Second )
2026-05-22 20:26:11 -04:00
}
}
if err != nil {
if errors . Is ( err , playback . ErrSegmentNotFound ) {
writeError ( w , http . StatusNotFound , "not_found" , "Segment not found" )
return
}
writeError ( w , http . StatusInternalServerError , "internal_error" , "Failed to load segment" )
return
}
w . Header (). Set ( "Cache-Control" , "no-store, max-age=0" )
w . Header (). Set ( "Pragma" , "no-cache" )
2026-08-11 12:59:25 -04:00
defer func () { _ = segmentLease . Close () }()
2026-08-08 15:23:16 -04:00
sw := httpstream . NewRollingDeadlineWriter ( w )
2026-08-11 12:59:25 -04:00
http . ServeContent ( sw , r , segmentLease . Info . Name (), segmentLease . Info . ModTime (), segmentLease . File )
2026-08-08 15:23:16 -04:00
if r . Method == http . MethodGet &&
2026-08-11 15:02:38 -04:00
sw . CompletedFullResponse ( segmentLease . Info . Size ()) {
2026-08-08 15:23:16 -04:00
if segNum , parseErr := playback . ParseSegmentNumber ( segmentName ); parseErr == nil {
2026-08-11 12:59:25 -04:00
transcodeSession . ReportSegmentDownloadedForGeneration ( segNum , segmentLease . Generation )
2026-08-08 15:23:16 -04:00
}
}
2026-05-22 20:26:11 -04:00
}
2026-07-03 02:23:14 +08:00
// buildProxyManifestURL signs a stream token carrying the session's full
// reconstruction recipe and builds the manifest URL. proxyNode is the planner's
// pick; when nil the URL falls back to the API-local path, where the token rides
// the ?st= query parameter so the integrated server can reconstruct from it.
func ( h * PlaybackHandler ) buildProxyManifestURL ( card playback . RecipeCard , proxyNode * nodepool . Node ) string {
token := h . signSessionToken ( card )
localURL := fmt . Sprintf ( "/playback/transcode/%s/master.m3u8" , card . SessionID )
2026-05-22 20:26:11 -04:00
if proxyNode == nil {
2026-07-03 02:23:14 +08:00
return appendStreamToken ( localURL , token )
2026-05-22 20:26:11 -04:00
}
2026-07-03 02:23:14 +08:00
if token == "" {
return localURL
2026-05-22 20:26:11 -04:00
}
return proxyNode . URL + "/stream/transcode/" + token + "/master.m3u8"
}
// proxyToTranscodeNode forwards a request to the remote transcode node.
func ( h * PlaybackHandler ) proxyToTranscodeNode ( w http . ResponseWriter , r * http . Request , transcodeNodeURL , path string ) {
sessionID := chi . URLParam ( r , "session_id" )
targetURL := transcodeNodeURL + path
2026-08-11 14:21:33 -04:00
isSegmentRoute := strings . Contains ( path , "/segment/" )
_ , segmentParseErr := playback . ParseSegmentNumber ( filepath . Base ( path ))
isMediaSegment := segmentParseErr == nil
2026-07-03 02:23:14 +08:00
// Capture the signed stream token ("st") before stripping it from the URL.
// We forward it out-of-band as a header so the node can reconstruct after a
// self-restart, while keeping it out of the forwarded/logged URL.
stToken := r . URL . Query (). Get ( "st" )
// Strip the signed stream token ("st") before forwarding/logging: it is a
// 24h bearer reconstruction descriptor exposing media path + recipe claims.
// Other query params are preserved.
query := r . URL . Query ()
query . Del ( "st" )
if encoded := query . Encode (); encoded != "" {
targetURL += "?" + encoded
2026-05-22 20:26:11 -04:00
}
req , err := http . NewRequestWithContext ( r . Context (), http . MethodGet , targetURL , nil )
if err != nil {
http . Error ( w , "internal error" , http . StatusInternalServerError )
return
}
req . Header . Set ( "Authorization" , "Bearer " + h . JWTSecret )
2026-08-11 14:21:33 -04:00
if isSegmentRoute {
// The node's immediate transport peer is this API process, so receiving a
// complete response there does not prove that the browser received it.
// Suppress node-local accounting and acknowledge only after the downstream
// writer completes. Forward range validators so both hops serve the same
// representation.
2026-08-11 14:30:55 -04:00
transcodeproxy . PrepareRequest ( req , r )
2026-08-11 14:21:33 -04:00
}
2026-07-03 02:23:14 +08:00
// Best-effort forward of the stream token as a header so the node's
// reconstruct path (X-Silo-Stream-Token) can rebuild after a self-restart.
// Verify at the API boundary and confirm it belongs to this session; an
// invalid or missing token never blocks the live proxy. validToken is kept so
// the same verified token can be re-injected into the node's manifest segment
// URIs below.
var validToken string
if stToken != "" && h . JWTSecret != "" {
claims , verifyErr := streamtoken . Verify ( stToken , h . JWTSecret )
if verifyErr == nil && claims . SessionID == sessionID {
req . Header . Set ( "X-Silo-Stream-Token" , stToken )
validToken = stToken
} else if verifyErr != nil {
2026-07-09 20:53:52 +08:00
slog . WarnContext ( r . Context (), "stream token not forwarded to transcode node" , "component" , "api" , "error" , verifyErr , "playback_session_id" , sessionID )
2026-07-03 02:23:14 +08:00
}
}
2026-05-22 20:26:11 -04:00
resp , err := http . DefaultClient . Do ( req )
if err != nil {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( r . Context (), "proxy to transcode node" , "component" , "api" , "error" , err , "url" , targetURL , "playback_session_id" , sessionID )
2026-05-22 20:26:11 -04:00
http . Error ( w , "transcode node unavailable" , http . StatusBadGateway )
return
}
2026-08-11 14:21:33 -04:00
defer func () { _ = resp . Body . Close () }()
2026-05-22 20:26:11 -04:00
2026-07-03 02:23:14 +08:00
// The node strips "st" from the request query (kept out of node URLs/logs),
// so the segment/init URIs in the manifest it builds carry no token. Without
// it, a segment fetched after a node or API restart cannot reconstruct the
// session and 404s. Re-inject the client-facing token into every URI at this
// boundary so the client's later segment requests carry "st" again. Only the
// manifest body is rewritten; segments stream through untouched.
if validToken != "" && resp . StatusCode == http . StatusOK && strings . HasSuffix ( path , ".m3u8" ) {
body , readErr := io . ReadAll ( resp . Body )
if readErr != nil {
2026-07-09 20:53:52 +08:00
slog . ErrorContext ( r . Context (), "read transcode node manifest" , "component" , "api" , "error" , readErr , "url" , targetURL , "playback_session_id" , sessionID )
2026-07-03 02:23:14 +08:00
http . Error ( w , "transcode node unavailable" , http . StatusBadGateway )
return
}
rewritten := playback . AppendManifestQueryParam ( body , streamTokenParam , validToken )
for k , vv := range resp . Header {
if http . CanonicalHeaderKey ( k ) == "Content-Length" {
continue
}
for _ , v := range vv {
w . Header (). Add ( k , v )
}
}
w . Header (). Set ( "Content-Length" , strconv . Itoa ( len ( rewritten )))
w . WriteHeader ( resp . StatusCode )
_ , _ = w . Write ( rewritten )
return
}
2026-08-11 14:30:55 -04:00
generation := resp . Header . Get ( transcodeproxy . GenerationHeader )
transcodeproxy . CopyResponseHeaders ( w . Header (), resp . Header )
2026-07-09 23:07:16 -04:00
// Proxied transcode output can stream past the server's absolute
// WriteTimeout; roll the write deadline with progress instead.
sw := httpstream . NewRollingDeadlineWriter ( w )
sw . WriteHeader ( resp . StatusCode )
2026-08-11 14:21:33 -04:00
if _ , copyErr := io . Copy ( sw , resp . Body ); copyErr != nil {
return
}
2026-08-11 14:30:55 -04:00
fullSize := transcodeproxy . FullRepresentationSize ( resp )
2026-08-11 14:21:33 -04:00
if isMediaSegment && generation != "" && r . Method == http . MethodGet &&
2026-08-11 15:02:38 -04:00
sw . CompletedFullResponse ( fullSize ) {
2026-08-11 14:30:55 -04:00
if ackErr := transcodeproxy . Acknowledge ( r . Context (), http . DefaultClient , transcodeNodeURL + path , h . JWTSecret , generation ); ackErr != nil {
slog . WarnContext ( r . Context (), "acknowledge transcode segment completion" , "component" , "api" , "error" , ackErr , "playback_session_id" , sessionID )
}
2026-08-11 14:21:33 -04:00
}
2026-05-22 20:26:11 -04:00
}
// maybeStartThrottler reads throttle settings and starts the throttler if enabled.
func ( h * PlaybackHandler ) maybeStartThrottler ( ctx context . Context , session * playback . TranscodeSession ) {
if h . SettingsRepo == nil {
return
}
enableThrottle , _ := h . SettingsRepo . Get ( ctx , "enable_transcode_throttle" )
if enableThrottle != "true" {
return
}
thresholdStr , _ := h . SettingsRepo . Get ( ctx , "transcode_throttle_seconds" )
threshold := 300 // default
if v , err := strconv . Atoi ( thresholdStr ); err == nil && v > 0 {
threshold = v
}
session . StartThrottler ( threshold )
}
// findAlternateFile finds a non-4K file version for the same content.
// Prefers SDR over HDR, then highest resolution, then highest bitrate.
func ( h * PlaybackHandler ) findAlternateFile ( ctx context . Context , source * models . MediaFile ) ( * models . MediaFile , error ) {
if h . FileVersionFetcher == nil {
return nil , fmt . Errorf ( "file version fetcher not configured" )
}
var files [] * models . MediaFile
var err error
if source . EpisodeID != "" {
files , err = h . FileVersionFetcher . GetByEpisodeID ( ctx , source . EpisodeID )
} else {
files , err = h . FileVersionFetcher . GetByContentID ( ctx , source . ContentID )
}
if err != nil {
return nil , err
}
// Filter to non-4K candidates.
candidates := make ([] * models . MediaFile , 0 , len ( files ))
for _ , f := range files {
if f . ID == source . ID || f . Resolution == "2160p" {
continue
}
if source . EditionKey != "" && f . EditionKey != source . EditionKey {
continue
}
if source . EditionKey == "" && f . EditionKey != "" {
continue
}
if source . PresentationGroupKey != "" && f . PresentationGroupKey != "" && f . PresentationGroupKey != source . PresentationGroupKey {
continue
}
if source . PresentationKind != "" && f . PresentationKind != "" && f . PresentationKind != source . PresentationKind {
continue
}
candidates = append ( candidates , f )
}
if len ( candidates ) == 0 {
return nil , nil
}
// Sort: SDR before HDR, then highest resolution, then highest bitrate.
sort . Slice ( candidates , func ( i , j int ) bool {
a , b := candidates [ i ], candidates [ j ]
// Prefer SDR over HDR (SDR = !HDR, so !HDR < HDR means SDR first).
if a . HDR != b . HDR {
return ! a . HDR
}
aRes := resolutionRank ( a . Resolution )
bRes := resolutionRank ( b . Resolution )
if aRes != bRes {
return aRes > bRes
}
return a . Bitrate > b . Bitrate
})
return candidates [ 0 ], nil
}
2026-07-28 19:07:05 +02:00
const (
transcodeResolution2160p = "2160p"
transcodeResolution1080p = "1080p"
transcodeResolution720p = "720p"
transcodeResolution480p = "480p"
transcodeResolution420p = "420p"
transcodeResolution328p = "328p"
)
2026-05-22 20:26:11 -04:00
// resolutionRank returns a numeric rank for resolution sorting.
func resolutionRank ( res string ) int {
2026-07-28 19:07:05 +02:00
height , known := transcodeResolutionHeight ( res )
if ! known {
return 0
}
switch {
case height >= 2160 :
2026-05-22 20:26:11 -04:00
return 4
2026-07-28 19:07:05 +02:00
case height >= 1080 :
2026-05-22 20:26:11 -04:00
return 3
2026-07-28 19:07:05 +02:00
case height >= 720 :
2026-05-22 20:26:11 -04:00
return 2
2026-07-28 19:07:05 +02:00
case height >= 480 :
2026-05-22 20:26:11 -04:00
return 1
default :
return 0
}
}
2026-07-28 19:07:05 +02:00
func transcodeResolutionHeight ( resolution string ) ( int , bool ) {
switch resolution {
case transcodeResolution2160p :
return 2160 , true
case transcodeResolution1080p :
return 1080 , true
case transcodeResolution720p :
return 720 , true
case transcodeResolution480p :
return 480 , true
case transcodeResolution420p :
return 420 , true
case transcodeResolution328p :
return 328 , true
default :
return 0 , false
}
}