Files
silo-server/internal/api/handlers/playback.go
T
881c96864b feat(playback): finalize platform-neutral protocol v3 (#567)
* docs(playback): add v3 neutral-contract finalization plan

Supersedes the wire-contract sections of the 2026-07-12 v3 plan: server-owned
attempt keys, delivery-keyed negotiation without Media3 engine names, tiered
capability evidence, neutral device/output context, track/quality replan
operations, audio-only planning, and coordinated no-back-compat rollout
across server, Android, Apple, and web.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* feat(playback): make v3 attempt keys server-owned and replace engines with deliveries

Contract core of the platform-neutral v3 finalization (plan sections 3.1
and 3.2), breaking on purpose — v3 is dark and all clients move together:

- Every PlanV3 now carries plan_attempt_key, an opaque server-computed
  token clients store and echo in attempted_plan_keys; ReplanRequestV3
  gains bounded local_mutations that the replan handler folds into the
  failed plan's key. Clients never hash anything.
- KotlinName() is deleted from DeliveryV3, StreamProtocolV3 and
  SubtitleModeV3; the attempt-key canonical string now uses lowercase
  wire tokens, and PlanRecipeVersionV3 bumps to v3.3 so no key or plan
  ID computed under the old canonicalization can collide.
- EngineV3 leaves the wire: ClientPlaybackContextV3.Engines (media3_*)
  becomes Deliveries keyed original_http|progressive|hls, with
  EngineCapabilityV3 renamed DeliveryCapabilityV3. PlanV3.Engine is
  removed; the planner, subtitle policy and quirk registry re-key on
  delivery class, and the media3_only feature token is deleted.
- Validated-claim strings drop the prefix: media3_h264_decode ->
  h264_decode, media3_audio_decode -> audio_decode.
- Golden fixtures in testdata/protocol_v3 are regenerated by Go and are
  now the cross-repo source of truth.

Part of the playback protocol v3 neutral-contract train (steps 2-3 of
docs/superpowers/plans/2026-07-30-playback-protocol-v3-neutral-contract.md).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* feat(playback): add v3 evidence tiers and neutral device/output context

Implement plan sections 3.3 and 3.4 of the v3 neutral-contract pass:

- ClientCodecCapabilitiesV3 gains required video_evidence and
  audio_evidence closed enums (exact | platform_attested | declared).
  Planner strictness follows the tier: exact keeps the strict decode-entry
  validation, platform_attested validates codec/resolution/bit-depth/
  frame-rate but skips profile/level matching, declared grants copy routes
  from the flat codec lists. Only exact audio evidence earns passthrough
  claims. The detailed_decode_capabilities feature token is deleted
  (subsumed by video_evidence=exact), and evidence-blocked direct routes
  carry the new evidence_insufficient_for_direct reason/warning.

- DeviceContextV3 is now platform/os_version/manufacturer/model plus a
  bounded platform_details map (<=16 entries, <=128 chars); the Android
  Build dump fields are gone. Fire TV quirks keep matching on
  manufacturer/model (brand fallback removed with the field).

- output_route_generation (int64, dual-location) becomes an optional
  opaque output_context_id string on the output context; the dual-location
  consistency validation is deleted. Attempt keys, plan invalidation,
  route events, and the planstore column follow (new Goose migration).

- Feature advertisement collapses to the top-level client_features list
  only; ClientPlaybackContextV3.Features is deleted and ReplanRequestV3
  gains an optional client_features refresh.

- PlanRecipeVersionV3 bumped v3.3 -> v3.4; fixtures re-keyed.

Part of the playback protocol v3 neutral-contract finalization plan
(docs/superpowers/plans/2026-07-30-playback-protocol-v3-neutral-contract.md).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* feat(playback): add v3 intent replans, quality menu, and audio-only routes

Protocol v3 could only replan after a failure, so changing the audio track
or the quality still required the legacy audio PATCH and the client-recipe
transcode start — the two endpoints v3 is meant to replace. Clients also had
to own a resolution ladder to render a quality menu, and a source with no
video track was terminaled by the video/HDR gates, keeping audiobooks on the
legacy path.

Add track_change and quality_change replan operations. They carry no failure
classification and route through the existing replan transaction, so they
inherit its idempotency, capacity reservation, and staged-successor commit
for free. Because nothing failed, the previous route stays eligible: neither
the attempted-key history nor the failed-plan exclusion applies to them.

Publish the server ladder on the plan as available_qualities so the quality
menu is server-owned; the rungs come from the same resolutionLabelV3 and
ladderBitrateKbpsV3 helpers the planner itself uses, not a parallel table.

Plan audio-only sources through their own reduced route family: original_http
when the client decodes the codec, otherwise a progressive AAC conversion.
The plan advertises audio/mp4 for that remux and the transport now serves the
same value, because a declared-tier client probes the advertised MIME with
isTypeSupported before attaching a source buffer, and "video/mp4" on a stream
with no video track is exactly the mismatch that makes the probe lie.

Name the protocol's string vocabulary (dynamic ranges, transformations,
executors, validated claims, terminal reasons) as constants while touching
these lines, so the wire values have one definition.

Part of #135

* docs(playback): publish the v3 protocol contract and fix subtitle ordinals

Protocol v3 exists only as Go code today, so the Android and Apple ports have
no authority to implement against other than reading this repository. Publish
the contract as a normative document, machine-checkable schemas, and generated
golden fixtures, and fix the one place where the server's own wire output
disagreed with the ordinal space it publishes.

- docs/architecture/playback-protocol-v3.md is self-contained enough for a
  third-party client: endpoints and status codes, evidence tiers and their
  bound-matching rules, delivery classes, the timeline model, replan
  semantics, registries, track identity, plan identity, quality, and
  transformations.
- docs/design/schemas/playback-v3/ carries JSON Schemas for the five wire
  shapes plus valid and invalid fixtures, following the client-diagnostics
  layout. internal/playback/contract validates every fixture against its
  schema, so a schema that drifts from the Go types fails the Go suite.
- cmd/playbackfixtures generates internal/playback/testdata/protocol_v3 from
  the production planner. `make playback-fixtures` writes them and
  `make verify-playback-fixtures` (wired into CI) fails when they are stale.
  These files are what the client ports consume, so drift would otherwise
  surface as a playback bug on three platforms at once.

The subtitle fix: combined ordinals are one dense space over externals, then
embedded tracks, then downloaded ones, but the legacy URL builder skipped
burn-in-only tracks while assigning indices, so every track after a DVD/DVB
track was numbered one too low and resolved to its neighbour. Ordinal
assignment now lives in playback.BuildSubtitleInventoryV3 and both the plan
inventory and the legacy `subtitle_urls` shape project from it; the legacy
shape still filters burn-in-only entries but keeps each track's real index.

Part of #135

* feat(web): migrate the players to the neutral playback v3 contract

The web player was the last client still speaking the legacy start
protocol: it picked its own file version from a codec probe, posted an
ffmpeg recipe to start a transcode, PATCHed an endpoint to change audio
tracks, and derived its own quality ladder. None of that survives a
server-owned plan, and none of it produced telemetry the apps could be
compared against.

Video player: starts with a v3 request that advertises `declared`
evidence from `isTypeSupported` probes and the three delivery classes,
then consumes the returned plan for its URL, timeline, tracks and
warnings. Quality and track changes become replans (`quality_change`,
`track_change`), the quality menu renders `available_qualities` instead
of computing rungs, and playback failures emit `route-events` so web
failures land in the same diagnostics as Android and Apple. The
duration comes from `source.duration_seconds` rather than the playback
engine, and the "how was this delivered" overlay reads the plan's
delivery and server transformations instead of comparing codec strings.

Audiobook player: starts against the audio-only planner path with a
single `original` rung, and takes its seek anchor from
`timeline.player_start_seconds` so the progressive-remux route (which
anchors the stream and restarts the player clock at zero) does not seek
twice.

Server side, `disable_progress_persistence` left the wire, so the rule
it encoded is now derived. Resume state is keyed on the item, but every
part of a multipart presentation shares that key while carrying its own
file-local clock — persisting part 4's position would store "12 minutes
in" as the book's resume point. `PresentationPartTotal > 1` expresses
that directly and generalizes to multipart movies and split episodes,
and a client can no longer forget to ask or lie about it.

`useTranscodeQuality` and the legacy response types are deleted, and
`WEBTEST_KNOWN_FAILURES` loses the audiobook entry along with its fix.

Part of #135

* feat(playback)!: make v3 the only playback protocol

Protocol v3 shipped behind a flag, alongside the legacy start path it was
designed to replace. Running both meant every planner change had to be made
twice, in two shapes that disagree about who decides the route: the legacy
body carried a decision the client had already made, while v3 asks the server
to make it. This deletes the legacy half.

Removed:

- `handleStartPlaybackLegacy` and its request/response bodies. The
  `POST /playback/start` route stays, but the protocol-version dispatch
  envelope is now a strict v3 decode — a body that does not declare
  `protocol_version: 3` gets `426 client_upgrade_required` so an outdated app
  can render a clear "update required" state instead of misreading a plan.
  Deliberately not a `400`: the request may be well-formed for the protocol it
  was written against.
- `POST /playback/transcode/start`, superseded by the `quality_change` replan
  operation, and `PATCH /playback/{session_id}/audio`, superseded by
  `track_change`. Both mutated a session without re-planning.
- The shadow planner and both rollout settings rows. With v3 the only
  protocol, `playback.protocol_v3_enabled` would mean "no playback at all";
  `playback.protocol_v3_shadow_enabled` gated a comparison against a path that
  no longer exists. `409 protocol_disabled` on route-events goes with them, and
  capability `enabled` is now constant `true` (the field stays — clients
  feature-detect against it).
- Version-selection helpers in `internal/playback/resolver.go` that only legacy
  start reached. `Resolve`/`ClientCapabilities`/`PlayDecision` stay: downloads
  consumes them. `internal/jellycompat` has its own resolution surface and is
  untouched.

Behaviour the legacy handlers owned and v3 now owns explicitly: series version
and audio-track preferences are persisted on start and on a `track_change`
replan (not on failure recovery, whose forced route is not a user choice); an
omitted `start_position` resolves to the profile's saved resume point; and an
omitted audio track resolves through the series preference, the profile audio
language, then the library override. Both are settled before planning, because
the plan's timeline is cut at the start position. Spec §2.2 documents this as
"omission is a request, not a default".

The encode-target clamp that lived in the deleted transcode handler is already
enforced in the planner, twice — `availableQualitiesV3` omits rungs at or above
the source height, and the encode path clamps `targetHeight` to it.

Unchanged: progress, stop, HLS manifest and segment delivery, the realtime
control socket, stream tokens and restart reconstruction, watch together,
downloads, jellycompat.

Every removal is recorded in the pre-lock removals table in
docs/architecture/v1-scope.md.

Part of #135

* fix(scanner): stop recording embedded cover art as a video track

ffprobe reports embedded cover art as a video stream carrying
disposition.attached_pic. convertProbeData appended every "video" stream
to VideoTracks without consulting isMainVideoStream, the predicate that
already existed for duration decisions, so the picture was persisted as a
playable track. That misreports the file twice:

  - An audio file with a cover picks up a video track, so it no longer
    satisfies MediaFile.IsAudioOnly and the v3 planner routes an
    audiobook through the video path instead of planAudioOnlyV3.
  - When the picture is ordered ahead of the real stream, the flat
    codec_video/resolution/hdr columns describe the poster: a 954x720
    h264 episode was stored as mjpeg 480x480.

Filter attached_pic streams out of the track loop. The guard is the
disposition flag, not the codec name, so a genuine MJPEG video is still
probed as video — the library has one.

Already-probed rows self-heal on the next playback: NeedsCriticalProbeRepair
already reprobes tracks missing color_range, which covers 21 of the 23
affected rows, and applyProbeData overwrites VideoTracks wholesale. The
remaining two need a rescan; nothing persisted records attached_pic, and
keying repair off still-image codec names would reprobe the genuine MJPEG
file on every playback forever.

Part of the playback v3 neutral-contract work: it is what lets Android
drop AUDIOBOOK_COVER_ART_CODECS, which fabricated decode support the
client cannot honestly claim under video_evidence: "exact".

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(playback): publish subtitle URLs even when playback starts with subtitles off

The v3 plan's subtitle inventory is the authoritative track list a client builds
its subtitle menu from, but the handler only rewrote it with session-scoped URLs
when a track was actually selected. A start or replan that resolved to
`subtitle.mode: "off"` therefore returned the planner's URL-less inventory, so a
client whose picker reads the inventory had a menu it could not fetch anything
from. The Cast path hits this every time: it starts with subtitles off and needs
the receiver's text tracks up front.

attachSubtitleArtifactV3 now scopes and publishes the inventory unconditionally
and gates only the artifact stamping on the selection. Spec §8 records that the
`url` on a sidecar entry does not depend on the current selection.

Part of the v3 neutral-contract finalization.

* chore(playback): reconcile neutral v3 with main

* fix(playback): preserve subtitle intent across replans

* fix(playback): retain subtitle inventory on adapted routes

* fix(playback): software-decode High10 AVC for QSV

* fix(playback): scale High10 frames before QSV upload

* fix(playback): preserve empty subtitle inventories

* fix(playback): freeze terminal attempt contract

* chore(playback): name fixture contract tokens

* fix(playback): close v3 conformance review gaps

* chore(playback): name conformance category

* fix(playback): complete v3 conformance contract

* fix(playback): keep schema fixtures generated

* fix(playback): emit schema-valid conformance arrays

* fix(playback): omit empty replan failures

* fix(web): omit empty replan failures

* fix(playback): close neutral v3 contract gaps

* fix(playback): harden v3 replan, transcode, and quality-ladder edge cases

Review remediation for the neutral v3 cutover, server side:

- A failed replan no longer overwrites the durable StartResponse with a
  terminal or advances the replan request ID; an idempotent start replay
  of a still-healthy session returns the original plan.
- SoftwareVideoDecode is now derived inside the transcode layer from
  source facts (codec/profile/bit depth) carried on TranscodeOpts, so
  jellycompat, downloads, recipe-card reconstruction, and transcode
  nodes get the High10 software-decode fix, not just the v3 handler.
  video_to_h264 recipe version bumps to 2 so mixed-version node pools
  that would silently drop the flag fail validation instead.
- Local transport startup shares the 30s ManifestStartupTimeout; a
  timeout with the process still running stays retryable and is no
  longer persisted as a durable terminal against the attempt.
- Sparse replan bodies (failure_recovery et al) no longer reset a
  user-selected quality preference to auto; the empty-value guard now
  covers every operation.
- availableQualitiesV3 publishes no fixed rungs when the source height
  is unknown, keeping the no-upscaling ladder contract.
- The proxy remux path serves audio-only fMP4 as audio/mp4 via a new
  additive AudioOnly token claim, matching the integrated path.
- Plain text subtitle sidecars accept any requested extension again
  (served as VTT), restoring the permissive v1 behavior; ASS and bitmap
  handling is unchanged.
- The 4K-disallowed terminal message discloses when a lower-resolution
  alternate exists but was pinned away by quality "original".

Part of #135.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(web): keep playback alive through failed replans and honest audio claims

Review remediation for the neutral v3 cutover, web player:

- A failed or refused replan no longer unmounts the player: the fatal
  error screen is reserved for loads with no adopted plan, and replan
  failures surface through the existing non-fatal replanError path.
- changeQuality rolls its optimistic preference back when the replan is
  refused or errors, so a failed switch is not silently applied by the
  next unrelated replan and the menu shows the real active rung.
- The capability probe now tests mp3/vorbis codecs and mp3/flac/ogg
  containers (MediaSource with a canPlayType fallback), restoring
  direct play for mp3 audiobooks instead of per-part AAC re-encodes.
- Reanchor seeks issued while a replan is in flight coalesce and run
  when it settles instead of being silently dropped with the scrubber
  pinned to a phantom position.
- Subtitle refresh/translation replans use the resume anchor while the
  media element has no metadata, so a subtitle_ready broadcast during
  startup no longer restarts a resumed stream at 0:00.
- An exhausted failure-recovery chain sets a visible error instead of
  returning silently.

Part of #135.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(playback): accept video-only and VP9 probe metadata

Treat audio and video probe completeness independently so legitimate video-only assets converge without repeated ffprobe repair. Allow unknown codec profile/level metadata to fall through to server adaptation while preserving exact direct-decode constraints.

Fixes #574

* fix(playback): address protocol v3 review findings

* fix(playback): harden lease and probe repair decisions

* fix(playback): close remaining v3 review gaps

* fix(playback): recover failed transcode starts

* fix(playback): address remaining review-bot findings on v3 replan and audio planning

Server:
- The deferred replan lease release is bounded by a 3s timeout so a
  saturated pool or DB outage cannot wedge a handler goroutine that
  holds the per-session store lock on an uncancellable context.
- planAudioOnlyV3 honors the request bandwidth cap: an over-cap source
  skips the original_http direct route and converts to AAC with the
  same bandwidth_cap_applied warning and decision reason the video
  ladder uses. Unknown source bitrate never triggers the cap.
- A copy-audio progressive plan rejected only by a per-delivery
  audio_decode_codecs subset retries as an AAC conversion instead of
  returning adaptation_unavailable, and the AAC recipe respects the
  delivery's max_channels.

Web:
- failure_recovery replans issued while another replan is in flight
  queue (superseding a pending seek reanchor) instead of being
  silently dropped with the fatal overlay already suppressed.
- A terminal response to a fresh non-preserving start clears the
  previous plan and stops its session, so episode navigation cannot
  keep rendering the prior item under the new title.
- A refused recovery replan for a transport-dead plan surfaces the
  error and re-arms the plan failure key, so transient recovery
  failures no longer strand an endless spinner; the audiobook player
  gets the same guard reset.
- A track-less subtitle_translation_completed hands off to the
  refreshed persisted track once the inventory settles, clearing the
  live overlay, instead of pinning the synthetic live track forever.

Part of #135.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(playback): reuse HLS transport for sidecar replans

* fix(playback): stabilize copy HLS remount timeline

* fix(playback): address v3 review findings

* fix(playback): satisfy player contract types

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-08-10 18:14:49 -04:00

1743 lines
66 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
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"
"github.com/Silo-Server/silo-server/internal/config"
evt "github.com/Silo-Server/silo-server/internal/events"
"github.com/Silo-Server/silo-server/internal/httpstream"
"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"
"github.com/Silo-Server/silo-server/internal/settingscontract"
"github.com/Silo-Server/silo-server/internal/settingskeys"
"github.com/Silo-Server/silo-server/internal/settingsresolve"
"github.com/Silo-Server/silo-server/internal/streamtoken"
"github.com/Silo-Server/silo-server/internal/subtitles"
"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
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
SetWebSocket(sessionID string, connected bool) error
SetRealtimeConnection(sessionID string, connected bool) error
SetProgressPersistenceDisabled(sessionID string, disabled bool) error
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)
}
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
}
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)
}
// 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)
}
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
CompletionObserver watchstate.CompletionObserver // optional; auto-removes watched items from the watchlist
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
NodePlanner nodepool.SessionPlanner // optional; enables proxy/transcode node selection
JWTSecret string // needed for signing stream tokens
ItemAccess PlaybackItemAccessChecker // optional; enables file authorization checks
EpisodeLookup PlaybackEpisodeLookup // optional; resolves episode files to their series
ExtraLookup PlaybackExtraLookup // optional; resolves extras files to their parent item
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
MarkerResolver markers.ExternalIDResolver
MarkerUpserter PlaybackMarkerUpserter
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
// 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.
PlaybackConfig func() config.PlaybackConfig
FFmpegLogSink playback.FFmpegLogSink
realtimeCommandMu sync.Mutex
realtimeCommands map[string]playbackCommandRecord
// 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
// 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
}
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),
tm: playback.NewTranscodeManager(),
PlanStoreV3: playback.NewMemoryPlanStoreV3(),
}
if len(opts) > 0 {
h.fileResolver = opts[0]
}
// 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{
TranscodeDir: c.TranscodeDir,
FFmpegPath: c.FFmpegPath,
HWAccel: c.HWAccel,
HWDevice: c.HWDevice,
}
}
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) {
slog.ErrorContext(ctx, "failed to stop playback after local transcode exit", "component", "api", "session", sessionID, "error", err, "playback_session_id", sessionID)
}
}
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
}
if setter, ok := sessionMgr.(sessionExpirationHookSetter); ok {
setter.SetExpirationHook(h.handleExpiredSession)
}
return h
}
// 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
}
// 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
}
// 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"),
}
}
// CleanupOrphanedTranscodes removes stale per-session temp directories for
// transcodes that are no longer tracked in memory, sparing dirs whose recipe
// card still exists. Delegates to the shared transcode manager.
func (h *PlaybackHandler) CleanupOrphanedTranscodes() (int, error) {
return h.tm.CleanupOrphanedTranscodes()
}
// 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 {
slog.WarnContext(ctx, "playback probe repair failed", "component", "api", "file_id", file.ID, "path", file.FilePath, "error", err)
return file
}
if repaired != nil {
return repaired
}
return file
}
// 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
// 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.
func (h *PlaybackHandler) playbackStreamURL(s *playback.Session) string {
if s == nil {
return ""
}
if s.PlayMethod == playback.PlayTranscode {
return fmt.Sprintf("/playback/transcode/%s/master.m3u8", s.ID)
}
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:
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
default:
return playback.NewDirectRecipeCard(s.ID, s.UserID, s.ProfileID, s.MediaFileID)
}
}
func fileBitrateKbps(file *models.MediaFile) int {
if file == nil || file.Bitrate <= 0 {
return 0
}
return file.Bitrate
}
func requestedMediaFileID(session *playback.Session) int {
if session == nil {
return 0
}
if session.RequestedMediaFileID > 0 {
return session.RequestedMediaFileID
}
return session.MediaFileID
}
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)
}
}
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
}
// 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)
}
// --- 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
}
if session == nil || session.DisableProgressPersistence {
return
}
// 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
}
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 {
slog.ErrorContext(ctx, "failed to get user store", "component", "api", "user_id", session.UserID, "error", err)
return
}
duration := float64(file.Duration)
if err := store.UpdateProgress(ctx, session.ProfileID, targetID, session.Position, duration, h.playbackThresholds(ctx)); err != nil {
slog.ErrorContext(ctx, "failed to persist progress", "component", "api", "session", session.ID, "error", err)
} 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 {
slog.ErrorContext(ctx, "failed to persist version hints", "component", "api", "session", session.ID, "error", err)
}
}
// 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{}
}
if session == nil || session.DisableProgressPersistence || session.Position <= 0 {
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)
watchSvc := watchstate.NewService(h.StoreProvider).
WithStableIdentityResolver(h.StableIdentityResolver).
WithCompletionObserver(h.CompletionObserver)
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 {
slog.ErrorContext(ctx, "failed to persist playback stop", "component", "api", "session", session.ID, "error", err)
} 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(),
}
return watchsync.ResolveScrobbleIdentity(ctx, h.StableIdentityResolver, event)
}
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
}
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 {
slog.ErrorContext(ctx, "failed to get user store for admin history", "component", "api", "session", session.ID, "error", storeErr)
} else if store != nil {
profile, profileErr := store.GetProfile(ctx, session.ProfileID)
if profileErr != nil {
slog.ErrorContext(ctx, "failed to load profile for admin history", "component", "api", "session", session.ID, "error", profileErr)
} 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 {
slog.ErrorContext(ctx, "failed to sync sessions", "component", "api", "reason", reason, "error", err)
}
}
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)
}
}
func (h *PlaybackHandler) finalizeSessionStop(ctx context.Context, session *playback.Session, syncNow bool, syncReason string, userInitiated bool) {
if h == nil || session == nil || session.ID == "" {
return
}
if ctx == nil {
ctx = context.Background()
}
stopResult := h.persistStopAndHistory(ctx, session)
if h.WatchScrobbler != nil {
if event, ok := h.scrobbleEventForStoppedSession(ctx, session, stopResult); ok && (userInitiated || stopResult.Completed) {
if err := h.WatchScrobbler.ScrobbleStop(ctx, event); err != nil {
slog.WarnContext(ctx, "failed to queue watch provider stop scrobble", "component", "api", "session", session.ID, "error", err)
}
} else if ok {
if err := h.WatchScrobbler.ScrobblePause(ctx, event); err != nil {
slog.WarnContext(ctx, "failed to queue watch provider pause scrobble", "component", "api", "session", session.ID, "error", err)
}
}
}
if entry, buildErr := h.buildAdminHistoryEntry(ctx, session); buildErr != nil {
slog.ErrorContext(ctx, "failed to build admin history", "component", "api", "session", session.ID, "error", buildErr)
} else if entry != nil && h.AdminStore != nil {
if err := h.AdminStore.RecordHistory(ctx, *entry); err != nil {
slog.ErrorContext(ctx, "failed to record admin history", "component", "api", "session", session.ID, "error", err)
}
}
if h.AdminStore != nil {
if err := h.AdminStore.DeleteSession(ctx, session.ID); err != nil {
slog.ErrorContext(ctx, "failed to delete synced session", "component", "api", "session", session.ID, "error", err)
}
}
h.closeTranscodeForSession(session)
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 {
slog.WarnContext(ctx, "failed to queue watch provider abort scrobble", "component", "api", "session", session.ID, "error", err)
}
}
}
}
if h.AdminStore != nil {
if err := h.AdminStore.DeleteSession(ctx, session.ID); err != nil {
slog.ErrorContext(ctx, "failed to delete synced session", "component", "api", "session", session.ID, "error", err)
}
}
// Abort is a connection drop / non-terminal teardown — keep the recipe card
// so the client can reconstruct on reconnect.
h.closeTranscodeForSession(session)
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)
// 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)
}()
}
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 {
slog.ErrorContext(ctx, "failed to access user store for series playback preference", "component", "api", "user_id", userID, "error", err)
return
}
if err := store.SetSeriesPlaybackPreference(ctx, userstore.SeriesPlaybackPreference{
ProfileID: profileID,
SeriesID: seriesID,
Resolution: file.Resolution,
HDR: file.HDR,
CodecVideo: file.CodecVideo,
}); err != nil {
slog.ErrorContext(ctx, "failed to persist series playback preference", "component", "api", "series_id", seriesID, "profile_id", profileID, "error", err)
}
}
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 {
slog.ErrorContext(ctx, "failed to access user store for audio preference", "component", "api", "user_id", userID, "error", err)
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 {
slog.ErrorContext(ctx, "failed to persist audio preference", "component", "api", "series_id", seriesID, "profile_id", profileID, "error", err)
}
}
// --- Handler methods ---
// 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.
func (h *PlaybackHandler) HandleStartPlayback(w http.ResponseWriter, r *http.Request) {
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"`
Capabilities *struct {
VideoEvidence *string `json:"video_evidence"`
AudioEvidence *string `json:"audio_evidence"`
} `json:"client_capabilities"`
}
if err := json.Unmarshal(body, &envelope); err != nil {
writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body")
return
}
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)
return
}
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(),
}
}
// HandleUpdateProgress handles POST /playback/{session_id}/progress.
func (h *PlaybackHandler) HandleUpdateProgress(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
}
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")
return
}
err = h.sessionMgr.UpdateProgress(sessionID, req.Position, req.IsPaused)
if err != nil {
if errors.Is(err, playback.ErrSessionNotFound) {
writePlaybackSessionNotFound(w)
return
}
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to update progress")
return
}
h.syncSessionsNow(r.Context(), "progress")
// 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)
}
}
}
}
}
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
}
err = h.stopPlaybackSession(r.Context(), session, true)
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 {
return nil, mapMediaFileLookupError(err)
}
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
}
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
}
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))
}
// 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)
}
// 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.
//
// Known-duration encoded sessions expose a synthetic full VOD manifest so the
// player can seek immediately. Copy-video sessions expose FFmpeg's real
// keyframe-aligned manifest and use the resolved stream origin the v3 plan
// reports as the timeline's stream_origin_seconds.
func (h *PlaybackHandler) HandleGetTranscodeManifest(w http.ResponseWriter, r *http.Request) {
sessionID := chi.URLParam(r, "session_id")
session, status, card := h.loadTranscodeServeSession(r, sessionID)
switch status {
case playback.SessionMissing:
writePlaybackSessionNotFound(w)
return
case playback.SessionLoadFailed:
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to load playback session")
return
case playback.SessionForbidden:
writeError(w, http.StatusForbidden, "forbidden", "Session belongs to another user")
return
}
transcodeSession := h.tm.GetTranscodeSession(sessionID)
if transcodeSession == nil {
// No local session — try proxying to remote transcode node.
if session.TranscodeNodeURL != "" {
h.touchSessionActivity(sessionID)
h.proxyToTranscodeNode(w, r, session.TranscodeNodeURL,
"/transcode/"+remoteTransportID(session)+"/master.m3u8")
return
}
// 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
}
}
h.touchSessionActivity(sessionID)
manifest, err := transcodeSession.BuildPlaybackManifest("segment/", r.URL.RawQuery)
if err != nil {
slog.ErrorContext(r.Context(), "build transcode manifest", "component", "api", "error", err, "session", sessionID, "playback_session_id", sessionID)
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")
session, status, card := h.loadTranscodeServeSession(r, sessionID)
switch status {
case playback.SessionMissing:
writePlaybackSessionNotFound(w)
return
case playback.SessionLoadFailed:
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to load playback session")
return
case playback.SessionForbidden:
writeError(w, http.StatusForbidden, "forbidden", "Session belongs to another user")
return
}
transcodeSession := h.tm.GetTranscodeSession(sessionID)
if transcodeSession == nil {
if session.TranscodeNodeURL != "" {
h.touchSessionActivity(sessionID)
segmentName := chi.URLParam(r, "name")
h.proxyToTranscodeNode(w, r, session.TranscodeNodeURL,
"/transcode/"+remoteTransportID(session)+"/segment/"+segmentName)
return
}
// 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
}
}
h.touchSessionActivity(sessionID)
segmentName := chi.URLParam(r, "name")
segmentPath, err := transcodeSession.GetSegment(segmentName)
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()
}
slog.InfoContext(r.Context(), "transcode segment missing", "component", "api",
"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(),
"restart_on_timeout", decision.RestartOnTimeout,
"reason", decision.Reason,
"session", sessionID,
"playback_session_id", sessionID,
)
if decision.Wait {
slog.InfoContext(r.Context(), "transcode segment wait", "component", "api",
"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(),
"restart_on_timeout", decision.RestartOnTimeout,
"reason", decision.Reason,
"session", sessionID,
"playback_session_id", sessionID,
)
segmentPath, err = transcodeSession.WaitForSegment(segmentName, decision.WaitTimeout)
if err != nil && errors.Is(err, playback.ErrSegmentNotFound) {
slog.InfoContext(r.Context(), "transcode segment wait timeout", "component", "api",
"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(),
"restart_on_timeout", decision.RestartOnTimeout,
"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.
if err != nil && errors.Is(err, playback.ErrSegmentNotFound) && decision.RestartOnTimeout {
seekSeconds, ok, seekErr := transcodeSession.RestartSeekTarget(segNum)
if seekErr != nil && !errors.Is(seekErr, playback.ErrManifestNotReady) {
slog.ErrorContext(r.Context(), "resolve transcode seek target", "component", "api", "error", seekErr, "segment", segmentName, "session", sessionID, "playback_session_id", sessionID)
}
// 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
}
if ok {
slog.InfoContext(r.Context(), "transcode seek restart", "component", "api",
"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(),
"restart_on_timeout", decision.RestartOnTimeout,
"reason", decision.Reason,
"seek_seconds", seekSeconds,
"session", sessionID,
"playback_session_id", sessionID,
)
if restartErr := h.tm.RestartSessionLocked(
context.WithoutCancel(r.Context()),
sessionID,
transcodeSession,
seekSeconds,
segNum,
); restartErr == nil {
// Throttler + exit monitor re-arm via the session's
// restart hook.
segmentPath, err = transcodeSession.WaitForSegment(segmentName, 30*time.Second)
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)
_, _ = transcodeSession.WaitForSegment(nextSegmentName, 1200*time.Millisecond)
}
}
}
}
} else if transcodeSession.IsRunning() {
// Non-numbered segment (e.g., init.mp4 for fMP4 HLS).
// Wait briefly — the init segment is written almost immediately.
segmentPath, err = transcodeSession.WaitForSegment(segmentName, 10*time.Second)
}
}
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
}
// Report segment download for throttle tracking.
if segNum, parseErr := playback.ParseSegmentNumber(segmentName); parseErr == nil {
transcodeSession.ReportSegmentDownloaded(segNum)
}
w.Header().Set("Cache-Control", "no-store, max-age=0")
w.Header().Set("Pragma", "no-cache")
http.ServeFile(w, r, segmentPath)
}
// 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)
if proxyNode == nil {
return appendStreamToken(localURL, token)
}
if token == "" {
return localURL
}
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
// 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
}
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)
// 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 {
slog.WarnContext(r.Context(), "stream token not forwarded to transcode node", "component", "api", "error", verifyErr, "playback_session_id", sessionID)
}
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
slog.ErrorContext(r.Context(), "proxy to transcode node", "component", "api", "error", err, "url", targetURL, "playback_session_id", sessionID)
http.Error(w, "transcode node unavailable", http.StatusBadGateway)
return
}
defer resp.Body.Close()
// 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 {
slog.ErrorContext(r.Context(), "read transcode node manifest", "component", "api", "error", readErr, "url", targetURL, "playback_session_id", sessionID)
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
}
for k, vv := range resp.Header {
for _, v := range vv {
w.Header().Add(k, v)
}
}
// 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)
io.Copy(sw, resp.Body)
}
// 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
}
const (
transcodeResolution2160p = "2160p"
transcodeResolution1080p = "1080p"
transcodeResolution720p = "720p"
transcodeResolution480p = "480p"
transcodeResolution420p = "420p"
transcodeResolution328p = "328p"
)
// resolutionRank returns a numeric rank for resolution sorting.
func resolutionRank(res string) int {
height, known := transcodeResolutionHeight(res)
if !known {
return 0
}
switch {
case height >= 2160:
return 4
case height >= 1080:
return 3
case height >= 720:
return 2
case height >= 480:
return 1
default:
return 0
}
}
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
}
}