3008 lines
136 KiB
Go
3008 lines
136 KiB
Go
package handlers
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"net/http"
|
|
"os"
|
|
"path/filepath"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/go-chi/chi/v5"
|
|
"github.com/google/uuid"
|
|
|
|
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/models"
|
|
"github.com/Silo-Server/silo-server/internal/nodepool"
|
|
"github.com/Silo-Server/silo-server/internal/playback"
|
|
"github.com/Silo-Server/silo-server/internal/subtitles"
|
|
"github.com/Silo-Server/silo-server/internal/transcodenode"
|
|
)
|
|
|
|
const (
|
|
maxPlaybackV3BodyBytes = 256 << 10
|
|
maxPlaybackV3EventBodyBytes = 32 << 10
|
|
replanLeaseDurationV3 = 15 * time.Second
|
|
replanReleaseTimeoutV3 = 3 * time.Second
|
|
v3NodeCapabilityTTL = time.Minute
|
|
playbackNodeIntegratedV3 = "integrated"
|
|
subtitleFormatVTTV3 = "vtt"
|
|
subtitleMIMEVTTV3 = "text/vtt"
|
|
subtitleUnavailableReasonV3 = "subtitle_artifact_unavailable"
|
|
seekRestorationPlayerV3 = "player_position"
|
|
// Failed capability fetches are memoized briefly so an unreachable node
|
|
// costs one timeout per window instead of one per planning request.
|
|
v3NodeCapabilityErrorTTL = 15 * time.Second
|
|
// Capability fetches on the planning path run under a deadline well below
|
|
// the fetch helper's own 10s timeout: planning happens on the start
|
|
// request path, where a slow node must degrade the union, not the user.
|
|
v3NodeCapabilityPlanTimeout = 3 * time.Second
|
|
)
|
|
|
|
var errSubtitleStoreUnavailableV3 = errors.New("subtitle store unavailable")
|
|
|
|
type v3NodeCapabilityCache struct {
|
|
transformations []playback.TransformationV3
|
|
err error
|
|
expiresAt time.Time
|
|
}
|
|
|
|
type preparedTransportV3 struct {
|
|
url string
|
|
nodeURL string
|
|
transportID string
|
|
commit func()
|
|
rollback func()
|
|
applySession func() (func() error, error)
|
|
afterDurableCommit func()
|
|
}
|
|
|
|
type transportErrorV3 struct {
|
|
reason string
|
|
message string
|
|
retryable bool
|
|
cause error
|
|
}
|
|
|
|
func subtitleArtifactErrorV3(message string, cause error) *transportErrorV3 {
|
|
return &transportErrorV3{
|
|
reason: subtitleUnavailableReasonV3,
|
|
message: message,
|
|
retryable: errors.Is(cause, errSubtitleStoreUnavailableV3),
|
|
cause: cause,
|
|
}
|
|
}
|
|
|
|
func wrapSubtitleStoreErrorV3(err error) error {
|
|
return fmt.Errorf("%w: %w", errSubtitleStoreUnavailableV3, err)
|
|
}
|
|
|
|
type v3ReplanLock struct {
|
|
mu sync.Mutex
|
|
refs int
|
|
}
|
|
|
|
type v3EventRate struct {
|
|
windowStart time.Time
|
|
count int
|
|
}
|
|
|
|
type replacementAdmissionCheckerV3 interface {
|
|
CheckReplacementAllowed(context.Context, string, playback.PlayMethod, bool) error
|
|
}
|
|
|
|
type replacementReservationCancellerV3 interface {
|
|
CancelReplacementReservation(string)
|
|
}
|
|
|
|
type replacementStateManagerV3 interface {
|
|
ApplyReplacement(string, playback.SessionReplacement) (playback.SessionReplacementRollback, error)
|
|
RollbackReplacement(string, playback.SessionReplacementRollback) error
|
|
}
|
|
|
|
type sessionReservationReleaserV3 interface {
|
|
ReleaseSession(string)
|
|
}
|
|
|
|
func (e *transportErrorV3) Error() string {
|
|
if e.cause != nil {
|
|
return e.reason + ": " + e.cause.Error()
|
|
}
|
|
return e.reason
|
|
}
|
|
|
|
func (h *PlaybackHandler) transformationRegistryV3(ctx context.Context) *playback.TransformationRegistryV3 {
|
|
h.v3RegistryOnce.Do(func() {
|
|
h.v3Registry = playback.ProbeTransformationRegistryV3(context.WithoutCancel(ctx), h.playbackConfig().FFmpegPath)
|
|
})
|
|
return h.v3Registry
|
|
}
|
|
|
|
// remoteTransformationsV3 is the transport-time capability lookup for a
|
|
// selected node. It never trusts memoized failures: those may be planning
|
|
// deadlines far shorter than this path's fetch budget, and rejecting the
|
|
// already-selected node on a stale planning timeout would fail a start the
|
|
// fetch could still validate.
|
|
func (h *PlaybackHandler) remoteTransformationsV3(ctx context.Context, nodeURL string) ([]playback.TransformationV3, error) {
|
|
return h.lookupRemoteTransformationsV3(ctx, nodeURL, false)
|
|
}
|
|
|
|
// remoteTransformationsPlanningV3 is the planning-time variant: it honors
|
|
// negatively-cached fetch failures so an unreachable node costs one timeout
|
|
// per error-TTL window instead of one per playback start.
|
|
func (h *PlaybackHandler) remoteTransformationsPlanningV3(ctx context.Context, nodeURL string) ([]playback.TransformationV3, error) {
|
|
return h.lookupRemoteTransformationsV3(ctx, nodeURL, true)
|
|
}
|
|
|
|
func (h *PlaybackHandler) lookupRemoteTransformationsV3(ctx context.Context, nodeURL string, honorCachedFailure bool) ([]playback.TransformationV3, error) {
|
|
now := time.Now()
|
|
h.v3NodeCapabilitiesMu.Lock()
|
|
entry, ok := h.v3NodeCapabilities[nodeURL]
|
|
h.v3NodeCapabilitiesMu.Unlock()
|
|
if ok && now.Before(entry.expiresAt) {
|
|
if entry.err == nil {
|
|
return append([]playback.TransformationV3(nil), entry.transformations...), nil
|
|
}
|
|
if honorCachedFailure {
|
|
return nil, entry.err
|
|
}
|
|
}
|
|
|
|
info, err := fetchRemoteTranscodeCapabilities(ctx, nodeURL, h.JWTSecret)
|
|
if err != nil {
|
|
h.v3NodeCapabilitiesMu.Lock()
|
|
if h.v3NodeCapabilities == nil {
|
|
h.v3NodeCapabilities = make(map[string]v3NodeCapabilityCache)
|
|
}
|
|
h.v3NodeCapabilities[nodeURL] = v3NodeCapabilityCache{err: err, expiresAt: now.Add(v3NodeCapabilityErrorTTL)}
|
|
h.v3NodeCapabilitiesMu.Unlock()
|
|
return nil, err
|
|
}
|
|
entry = v3NodeCapabilityCache{
|
|
transformations: append([]playback.TransformationV3(nil), info.Transformations...),
|
|
expiresAt: now.Add(v3NodeCapabilityTTL),
|
|
}
|
|
h.v3NodeCapabilitiesMu.Lock()
|
|
if h.v3NodeCapabilities == nil {
|
|
h.v3NodeCapabilities = make(map[string]v3NodeCapabilityCache)
|
|
}
|
|
h.v3NodeCapabilities[nodeURL] = entry
|
|
h.v3NodeCapabilitiesMu.Unlock()
|
|
return append([]playback.TransformationV3(nil), entry.transformations...), nil
|
|
}
|
|
|
|
// transcodeNodeEnumeratorV3 exposes the pooled transcode nodes whose
|
|
// advertised transformations widen HLS planning; *nodepool.Planner implements
|
|
// it.
|
|
type transcodeNodeEnumeratorV3 interface {
|
|
TranscodeNodeURLs() []string
|
|
}
|
|
|
|
// hlsPlanningRegistryV3 returns the registry HLS deliveries plan against: the
|
|
// local probe plus every pooled transcode node's advertised transformations.
|
|
// Only availability of locally-defined specs widens (name and recipe version
|
|
// pinned by this server), so any plan built from it passes the per-node
|
|
// advertisement validation when that node is selected, and the local-fallback
|
|
// validation in prepareTransportV3 rejects recipes only nodes can run.
|
|
// Without pooled nodes this is exactly the local registry.
|
|
func (h *PlaybackHandler) hlsPlanningRegistryV3(ctx context.Context) *playback.TransformationRegistryV3 {
|
|
local := h.transformationRegistryV3(ctx)
|
|
enumerator, ok := h.NodePlanner.(transcodeNodeEnumeratorV3)
|
|
if !ok {
|
|
return local
|
|
}
|
|
nodeURLs := enumerator.TranscodeNodeURLs()
|
|
if len(nodeURLs) == 0 {
|
|
return local
|
|
}
|
|
var merged []playback.TransformationV3
|
|
for _, transformations := range h.pooledNodeTransformationsV3(ctx, nodeURLs) {
|
|
merged = append(merged, transformations...)
|
|
}
|
|
return local.WithAdvertised(merged)
|
|
}
|
|
|
|
// lazyHLSPlanningRegistryV3 defers (and memoizes) the widened-registry build
|
|
// so the planner only pays for node capability lookups when a route decision
|
|
// actually depends on them; direct-play and other source-preserving starts
|
|
// never touch the pool.
|
|
func (h *PlaybackHandler) lazyHLSPlanningRegistryV3(ctx context.Context) func() *playback.TransformationRegistryV3 {
|
|
var once sync.Once
|
|
var registry *playback.TransformationRegistryV3
|
|
return func() *playback.TransformationRegistryV3 {
|
|
once.Do(func() { registry = h.hlsPlanningRegistryV3(ctx) })
|
|
return registry
|
|
}
|
|
}
|
|
|
|
// pooledNodeTransformationsV3 collects the advertised transformations of the
|
|
// given transcode nodes, keyed by node URL. Stale cache entries are refreshed
|
|
// concurrently under a short planning deadline; nodes that cannot be reached
|
|
// contribute nothing (their failures are negatively cached), so planning
|
|
// degrades toward the local registry instead of blocking the start path.
|
|
func (h *PlaybackHandler) pooledNodeTransformationsV3(ctx context.Context, nodeURLs []string) map[string][]playback.TransformationV3 {
|
|
fetchCtx, cancel := context.WithTimeout(ctx, v3NodeCapabilityPlanTimeout)
|
|
defer cancel()
|
|
results := make([][]playback.TransformationV3, len(nodeURLs))
|
|
var wg sync.WaitGroup
|
|
for i, nodeURL := range nodeURLs {
|
|
wg.Add(1)
|
|
go func(i int, nodeURL string) {
|
|
defer wg.Done()
|
|
transformations, err := h.remoteTransformationsPlanningV3(fetchCtx, nodeURL)
|
|
if err != nil {
|
|
slog.DebugContext(ctx, "protocol v3 node capability unavailable for planning", "component", "api", "node", nodeURL, "error", err)
|
|
return
|
|
}
|
|
results[i] = transformations
|
|
}(i, nodeURL)
|
|
}
|
|
wg.Wait()
|
|
byURL := make(map[string][]playback.TransformationV3, len(nodeURLs))
|
|
for i, transformations := range results {
|
|
if transformations != nil {
|
|
byURL[nodeURLs[i]] = transformations
|
|
}
|
|
}
|
|
return byURL
|
|
}
|
|
|
|
// capabilitySessionPlannerV3 is implemented by *nodepool.Planner; it lets the
|
|
// transport layer restrict node selection to nodes that can execute the
|
|
// plan's server transformations.
|
|
type capabilitySessionPlannerV3 interface {
|
|
PlanSessionWith(sessionID, currentTranscodeURL string, needsTranscode bool, estBitrateKbps int, eligible func(*nodepool.Node) bool) nodepool.Plan
|
|
}
|
|
|
|
// planNodeSessionV3 selects transcode/proxy nodes for the session. Plans that
|
|
// carry server transformations restrict selection to nodes whose advertised
|
|
// capabilities validate against the plan, so load balancing in a
|
|
// heterogeneous pool cannot land a recipe on a node that would reject it when
|
|
// a capable sibling exists. Capability-blind selection remains for
|
|
// transformation-free plans and non-enumerating planners.
|
|
func (h *PlaybackHandler) planNodeSessionV3(ctx context.Context, session *playback.Session, result playback.PlannerResultV3) nodepool.Plan {
|
|
selector, selectable := h.NodePlanner.(capabilitySessionPlannerV3)
|
|
enumerator, enumerable := h.NodePlanner.(transcodeNodeEnumeratorV3)
|
|
if !selectable || !enumerable || !planRequiresServerTransformationsV3(result.Plan) {
|
|
return h.NodePlanner.PlanSession(session.ID, session.TranscodeNodeURL, true, result.TargetBitrateKbps)
|
|
}
|
|
capable := make(map[string]struct{})
|
|
for nodeURL, advertised := range h.pooledNodeTransformationsV3(ctx, enumerator.TranscodeNodeURLs()) {
|
|
if validateAdvertisedTransformationsV3(result.Plan, advertised) == nil {
|
|
capable[nodeURL] = struct{}{}
|
|
}
|
|
}
|
|
// The predicate runs under the planner lock: a set lookup only.
|
|
return selector.PlanSessionWith(session.ID, session.TranscodeNodeURL, true, result.TargetBitrateKbps, func(node *nodepool.Node) bool {
|
|
if node == nil {
|
|
return false
|
|
}
|
|
_, ok := capable[node.URL]
|
|
return ok
|
|
})
|
|
}
|
|
|
|
// validateAdvertisedTransformationsV3 verifies that every server-executed
|
|
// transformation the plan requires is advertised — at the exact recipe
|
|
// version — by the executor under consideration (a pooled node's capability
|
|
// response or the local registry's Advertised set).
|
|
func validateAdvertisedTransformationsV3(plan *playback.PlanV3, advertised []playback.TransformationV3) error {
|
|
available := make(map[string]string, len(advertised))
|
|
for _, transformation := range advertised {
|
|
available[strings.ToLower(strings.TrimSpace(transformation.Name))] = strings.TrimSpace(transformation.RecipeVersion)
|
|
}
|
|
if plan == nil {
|
|
return errors.New("playback plan is unavailable")
|
|
}
|
|
for _, required := range plan.Transformations {
|
|
if strings.EqualFold(required.Executor, "client") {
|
|
continue
|
|
}
|
|
version, ok := available[strings.ToLower(strings.TrimSpace(required.Name))]
|
|
if !ok || version != strings.TrimSpace(required.RecipeVersion) {
|
|
return fmt.Errorf("executor lacks transformation %s@%s", required.Name, required.RecipeVersion)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// HandlePlaybackCapabilityV3 reports only transformations that the installed
|
|
// runtime has actually probed. Protocol v3 is the server's only playback
|
|
// protocol, so `enabled` is constant; it stays in the response because clients
|
|
// feature-detect against it and the field is part of the frozen contract.
|
|
func (h *PlaybackHandler) HandlePlaybackCapabilityV3(w http.ResponseWriter, r *http.Request) {
|
|
if apimw.GetUserID(r.Context()) == 0 {
|
|
writeError(w, http.StatusUnauthorized, "unauthorized", "Authentication required")
|
|
return
|
|
}
|
|
response := playback.CapabilityResponseV3{Enabled: true, ProtocolVersions: []int{playback.ProtocolV3}}
|
|
response.Features = playback.ServerFeaturesV3()
|
|
response.Deliveries = []playback.DeliveryV3{playback.DeliveryOriginalHTTPV3, playback.DeliveryRemuxProgressiveV3, playback.DeliveryRemuxHLSV3, playback.DeliveryTranscodeHLSV3}
|
|
response.Transformations = h.transformationRegistryV3(r.Context()).Advertised()
|
|
writeJSON(w, http.StatusOK, response)
|
|
}
|
|
|
|
func (h *PlaybackHandler) handleStartPlaybackV3(w http.ResponseWriter, r *http.Request, body []byte) {
|
|
var req playback.StartRequestV3
|
|
if err := json.Unmarshal(body, &req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "Invalid protocol v3 request body")
|
|
return
|
|
}
|
|
warnings, err := req.NormalizeAndValidate()
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
|
|
return
|
|
}
|
|
profileID := apimw.GetProfileID(r.Context())
|
|
if profileID == "" {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "X-Profile-Id header is required")
|
|
return
|
|
}
|
|
if req.ProfileID != profileID {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "profile_id must match X-Profile-Id")
|
|
return
|
|
}
|
|
userID := apimw.GetUserID(r.Context())
|
|
digestBytes := sha256.Sum256(body)
|
|
requestDigest := hex.EncodeToString(digestBytes[:])
|
|
if existing, lookupErr := h.PlanStoreV3.GetAttemptByPlaybackAttemptID(r.Context(), req.PlaybackAttemptID); lookupErr == nil {
|
|
if existing.UserID != userID || existing.ProfileID != profileID || existing.RequestedMediaFileID != req.FileID ||
|
|
(existing.RequestDigest != "" && existing.RequestDigest != requestDigest) {
|
|
writeError(w, http.StatusConflict, "playback_attempt_reused", "The playback attempt ID belongs to a different request")
|
|
return
|
|
}
|
|
response := decisionResponseFromAttemptV3(existing)
|
|
if response.Terminal != nil {
|
|
writeJSON(w, http.StatusCreated, response)
|
|
return
|
|
}
|
|
// The replayed plan is only usable while its session is alive; a dead
|
|
// session must surface as a retryable terminal so the client mints a
|
|
// fresh attempt instead of replaying a plan it can never stream.
|
|
if existing.SessionID == "" {
|
|
writeError(w, http.StatusInternalServerError, "internal_error", "Stored playback attempt has no replayable decision")
|
|
return
|
|
}
|
|
if _, sessionErr := h.sessionMgr.GetSession(existing.SessionID); sessionErr != nil {
|
|
writeJSON(w, http.StatusCreated, playback.NewTerminalResponseV3("session_expired", "The playback session for this attempt has ended.", true))
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusCreated, response)
|
|
return
|
|
} else if !errors.Is(lookupErr, playback.ErrSessionNotFound) {
|
|
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to check playback attempt idempotency")
|
|
return
|
|
}
|
|
requestedFile, err := h.loadAuthorizedFile(r, req.FileID)
|
|
if err != nil {
|
|
writeV3FileError(w, err)
|
|
return
|
|
}
|
|
requestedFile = h.ensurePlaybackProbe(r.Context(), requestedFile)
|
|
audioIndex, err := resolveV3AudioIndex(requestedFile, req.AudioTrackID, req.AudioTrackIndex)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
|
|
return
|
|
}
|
|
if req.AudioTrackID == "" && req.AudioTrackIndex == nil {
|
|
audioIndex, err = h.preferredAudioTrackIndexV3(r.Context(), userID, profileID, requestedFile)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to load the saved audio preference")
|
|
return
|
|
}
|
|
}
|
|
effectiveFile := requestedFile
|
|
settings := h.plannerSettingsV3(r.Context())
|
|
if err := preflightPlaybackFile(r.Context(), effectiveFile, h.MissingMarker, h.EventsHub); err != nil {
|
|
writePlaybackFilePreflightError(w, err)
|
|
return
|
|
}
|
|
if req.StartPosition == nil {
|
|
req.StartPosition, err = h.resumePositionV3(r.Context(), userID, profileID, effectiveFile)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to load saved playback progress")
|
|
return
|
|
}
|
|
}
|
|
result := playback.PlanPlaybackV3(playback.PlannerInputV3{
|
|
Request: req, RequestedFile: requestedFile, EffectiveFile: effectiveFile,
|
|
AudioTrackIndex: audioIndex, Settings: settings,
|
|
Registry: h.transformationRegistryV3(r.Context()), HLSRegistry: h.lazyHLSPlanningRegistryV3(r.Context()), DVRPUStrippable: h.lazyDVRPUStrippableV3(r.Context(), effectiveFile), Now: time.Now(),
|
|
AdditionalSubtitles: h.downloadedSubtitleInventoryV3(r.Context(), effectiveFile),
|
|
})
|
|
if result.Terminal != nil && result.Terminal.Reason == "no_alternate_version" && shouldTryAlternateFileV3(req.QualityPreference) {
|
|
if alternate, alternateErr := h.findAlternateFile(r.Context(), requestedFile); alternateErr == nil && alternate != nil {
|
|
effectiveFile = h.ensurePlaybackProbe(r.Context(), alternate)
|
|
audioIndex = remapAudioIndexV3(requestedFile, effectiveFile, audioIndex)
|
|
if err := h.remapSubtitleSelectionV3(r.Context(), requestedFile, effectiveFile, &req); err != nil {
|
|
response, persistErr := h.persistTerminalStartDecisionV3(r.Context(), userID, profileID, req, requestDigest, requestedFile.ID, effectiveFile.ID, playback.NewTerminalResponseV3("subtitle_unavailable_in_version", err.Error(), false))
|
|
if persistErr != nil {
|
|
writeStartAttemptPersistenceErrorV3(w, persistErr)
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusCreated, response)
|
|
return
|
|
}
|
|
if err := preflightPlaybackFile(r.Context(), effectiveFile, h.MissingMarker, h.EventsHub); err != nil {
|
|
writePlaybackFilePreflightError(w, err)
|
|
return
|
|
}
|
|
result = playback.PlanPlaybackV3(playback.PlannerInputV3{Request: req, RequestedFile: requestedFile, EffectiveFile: effectiveFile, AudioTrackIndex: audioIndex, Settings: settings, Registry: h.transformationRegistryV3(r.Context()), HLSRegistry: h.lazyHLSPlanningRegistryV3(r.Context()), DVRPUStrippable: h.lazyDVRPUStrippableV3(r.Context(), effectiveFile), Now: time.Now(), AdditionalSubtitles: h.downloadedSubtitleInventoryV3(r.Context(), effectiveFile)})
|
|
}
|
|
}
|
|
h.clarifyOriginalQuality4KTerminalV3(r.Context(), result.Terminal, requestedFile, !shouldTryAlternateFileV3(req.QualityPreference))
|
|
if result.Terminal != nil {
|
|
slog.InfoContext(r.Context(), "playback plan decided", "component", "playback",
|
|
"outcome", "terminal",
|
|
"reason", result.Terminal.Reason,
|
|
"file_id", effectiveFile.ID,
|
|
"quality_preference", req.QualityPreference,
|
|
)
|
|
response, persistErr := h.persistTerminalStartDecisionV3(r.Context(), userID, profileID, req, requestDigest, requestedFile.ID, effectiveFile.ID, playback.NewTerminalResponseV3(result.Terminal.Reason, result.Terminal.Message, result.Terminal.Retryable))
|
|
if persistErr != nil {
|
|
writeStartAttemptPersistenceErrorV3(w, persistErr)
|
|
return
|
|
}
|
|
if response.Terminal != nil {
|
|
h.enqueueRouteEventV3(playback.RouteEventRecordV3{RouteEventV3: playback.RouteEventV3{ProtocolVersion: playback.ProtocolV3, PlaybackAttemptID: req.PlaybackAttemptID, Event: playback.RouteEventTerminalV3, FallbackReason: response.Terminal.Reason, OutputContextID: req.ClientPlaybackContext.Output.OutputContextID}, UserID: userID, ProfileID: profileID, ClientName: playbackClientInfoFromRequest(r).Name, ClientVersion: playbackClientInfoFromRequest(r).Version, ClientModel: req.ClientPlaybackContext.Device.Model})
|
|
}
|
|
writeJSON(w, http.StatusCreated, response)
|
|
return
|
|
}
|
|
// One line per plan decision so route selection is reconstructible from
|
|
// server logs alone (finding a mis-planned route previously required
|
|
// correlating client logcat, ffmpeg commands, and session rows).
|
|
slog.InfoContext(r.Context(), "playback plan decided", "component", "playback",
|
|
"outcome", "plan",
|
|
"decision_reason", result.Plan.DecisionReason,
|
|
"delivery", result.Plan.Delivery,
|
|
"play_method", string(result.PlayMethod),
|
|
"requested_file_id", requestedFile.ID,
|
|
"effective_file_id", effectiveFile.ID,
|
|
"dv_profile", result.Plan.Source.DVProfile,
|
|
"dynamic_range", result.Plan.Source.DynamicRange,
|
|
"target_resolution", result.TargetResolution,
|
|
"target_bitrate_kbps", result.TargetBitrateKbps,
|
|
"quality_preference", req.QualityPreference,
|
|
"bandwidth_estimate_kbps", intOrZeroHandlerV3(req.BandwidthEstimateKbps),
|
|
)
|
|
result.Plan.DegradationWarnings = append(result.Plan.DegradationWarnings, warnings...)
|
|
response, statusErr := h.startPlannedPlaybackV3(r, userID, profileID, req, requestDigest, requestedFile, effectiveFile, audioIndex, result)
|
|
if statusErr != nil {
|
|
if statusErr.reason == "playback_attempt_reused" {
|
|
writeError(w, http.StatusConflict, "playback_attempt_reused", statusErr.message)
|
|
return
|
|
}
|
|
if statusErr.reason == "internal_error" {
|
|
slog.ErrorContext(r.Context(), "protocol v3 start failed", "component", "api", "reason", statusErr.reason, "error", statusErr.cause)
|
|
}
|
|
persistedResponse, persistErr := h.startFailureDecisionV3(r.Context(), userID, profileID, req, requestDigest, requestedFile.ID, effectiveFile.ID, statusErr)
|
|
if persistErr != nil {
|
|
writeStartAttemptPersistenceErrorV3(w, persistErr)
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusCreated, persistedResponse)
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusCreated, response)
|
|
}
|
|
|
|
func (h *PlaybackHandler) startPlannedPlaybackV3(r *http.Request, userID int, profileID string, req playback.StartRequestV3, requestDigest string, requestedFile, effectiveFile *models.MediaFile, audioIndex int, result playback.PlannerResultV3) (playback.DecisionResponseV3, *transportErrorV3) {
|
|
if result.Plan == nil {
|
|
return playback.DecisionResponseV3{}, &transportErrorV3{reason: "internal_error", message: "The server produced no playback plan."}
|
|
}
|
|
if checker, ok := h.sessionMgr.(transcodePermissionChecker); ok && (result.PlayMethod == playback.PlayTranscode || result.TranscodeAudio) {
|
|
if err := checker.CheckTranscodingAllowed(r.Context(), userID, result.PlayMethod == playback.PlayTranscode); err != nil {
|
|
reason := "transcoding_disabled"
|
|
if errors.Is(err, playback.ErrAudioTranscodingDisabled) {
|
|
reason = "audio_transcoding_disabled"
|
|
}
|
|
return playback.DecisionResponseV3{}, &transportErrorV3{reason: reason, message: "The selected server adaptation is disabled for this user."}
|
|
}
|
|
}
|
|
clientInfo := playbackClientInfoFromRequest(r)
|
|
ctx := playback.WithClientInfo(r.Context(), clientInfo)
|
|
var session *playback.Session
|
|
var err error
|
|
if starter, ok := h.sessionMgr.(sessionStarterWithFilesContext); ok {
|
|
session, err = starter.StartSessionWithFilesContext(ctx, userID, profileID, effectiveFile.ID, requestedFile.ID, result.PlayMethod, result.TranscodeAudio)
|
|
} else {
|
|
session, err = h.sessionMgr.StartSessionWithFiles(userID, profileID, effectiveFile.ID, requestedFile.ID, result.PlayMethod, result.TranscodeAudio)
|
|
}
|
|
if err != nil {
|
|
return playback.DecisionResponseV3{}, sessionStartErrorV3(err)
|
|
}
|
|
abort := func() { _ = h.stopPlaybackSessionByID(context.WithoutCancel(r.Context()), session.ID, false) }
|
|
if req.ProgressPersistence == playback.ProgressPersistenceClientV3 || !sessionOwnsResumeTimelineV3(effectiveFile) {
|
|
if err := h.sessionMgr.SetProgressPersistenceDisabled(session.ID, true); err != nil {
|
|
abort()
|
|
return playback.DecisionResponseV3{}, &transportErrorV3{
|
|
reason: "internal_error",
|
|
message: "Failed to establish the requested progress persistence policy.",
|
|
cause: err,
|
|
}
|
|
}
|
|
}
|
|
if err := h.sessionMgr.UpdateAudioTrack(session.ID, audioIndex, result.PlayMethod); err != nil {
|
|
abort()
|
|
return playback.DecisionResponseV3{}, &transportErrorV3{reason: "internal_error", message: "Failed to select the playback audio track.", cause: err}
|
|
}
|
|
position := floatOrZeroHandlerV3(req.StartPosition)
|
|
if err := h.sessionMgr.UpdateProgress(session.ID, position, false); err != nil {
|
|
abort()
|
|
return playback.DecisionResponseV3{}, &transportErrorV3{reason: "internal_error", message: "Failed to initialize the playback timeline.", cause: err}
|
|
}
|
|
session, err = h.sessionMgr.GetSession(session.ID)
|
|
if err != nil {
|
|
abort()
|
|
return playback.DecisionResponseV3{}, &transportErrorV3{reason: "internal_error", message: "Failed to load the initialized playback session.", cause: err}
|
|
}
|
|
result.Plan.SessionID = session.ID
|
|
frozenRecipe, frozenErr := h.freezeExecutableRecipeV3(r.Context(), effectiveFile, result)
|
|
if frozenErr != nil {
|
|
abort()
|
|
return playback.DecisionResponseV3{}, subtitleArtifactErrorV3("Failed to freeze the selected subtitle identity.", frozenErr)
|
|
}
|
|
transport, transportErr := h.prepareTransportV3(r, session, effectiveFile, result)
|
|
if transportErr != nil {
|
|
abort()
|
|
return playback.DecisionResponseV3{}, transportErr
|
|
}
|
|
result.Plan.Stream.URL = transport.url
|
|
if err := h.attachSubtitleArtifactV3(r.Context(), session.ID, effectiveFile, result.Plan, result.SubtitleTrackIndex, &frozenRecipe); err != nil {
|
|
transport.rollback()
|
|
abort()
|
|
return playback.DecisionResponseV3{}, subtitleArtifactErrorV3("Failed to prepare the selected subtitle artifact.", err)
|
|
}
|
|
response := playback.DecisionResponseV3{ProtocolVersion: playback.ProtocolV3, ServerFeatures: playback.ServerFeaturesV3(), Outcome: playback.OutcomePlayableV3, SessionID: session.ID, PlaybackPlan: result.Plan}
|
|
record := playback.AttemptRecordV3{PlaybackAttemptID: req.PlaybackAttemptID, SessionID: session.ID, UserID: userID, ProfileID: profileID, RequestedMediaFileID: requestedFile.ID, EffectiveMediaFileID: effectiveFile.ID, CurrentPlanID: result.Plan.PlanID, CurrentPlan: *result.Plan, FrozenRecipe: frozenRecipe, NormalizedRequest: req, StartResponse: response, RequestDigest: requestDigest, ExpiresAt: time.Now().Add(playback.MaxTokenTTL)}
|
|
if err := h.updateV3SessionState(r.Context(), session, effectiveFile, result, transport); err != nil {
|
|
transport.rollback()
|
|
abort()
|
|
return playback.DecisionResponseV3{}, &transportErrorV3{reason: "internal_error", message: "Failed to commit the live playback session.", cause: err}
|
|
}
|
|
if err := h.PlanStoreV3.SaveAttempt(r.Context(), record); err != nil {
|
|
transport.rollback()
|
|
abort()
|
|
if errors.Is(err, playback.ErrIdempotencyKeyReusedV3) {
|
|
return playback.DecisionResponseV3{}, &transportErrorV3{reason: "playback_attempt_reused", message: "The playback attempt ID was reused with different input."}
|
|
}
|
|
if errors.Is(err, playback.ErrPlaybackAttemptExistsV3) {
|
|
existing, lookupErr := h.PlanStoreV3.GetAttemptByPlaybackAttemptID(r.Context(), req.PlaybackAttemptID)
|
|
if lookupErr == nil && existing.UserID == userID && existing.ProfileID == profileID && existing.RequestedMediaFileID == req.FileID {
|
|
// Replaying a concurrent duplicate is only valid while its
|
|
// session is alive; otherwise tell the client to mint a new
|
|
// attempt rather than hand it a plan it can never stream.
|
|
if _, sessionErr := h.sessionMgr.GetSession(existing.SessionID); sessionErr != nil {
|
|
return playback.DecisionResponseV3{}, &transportErrorV3{reason: "session_expired", message: "The playback session for this attempt has ended.", retryable: true}
|
|
}
|
|
return decisionResponseFromAttemptV3(existing), nil
|
|
}
|
|
}
|
|
return playback.DecisionResponseV3{}, &transportErrorV3{reason: "internal_error", message: "Failed to persist the playback plan.", cause: err}
|
|
}
|
|
transport.commit()
|
|
// Start-side effects belong after both the attempt and transport commits:
|
|
// retries that lose the idempotency race must not emit duplicate provider
|
|
// scrobbles or analysis work for the short-lived session they roll back.
|
|
if !session.DisableProgressPersistence && h.WatchScrobbler != nil && effectiveFile != nil {
|
|
targetID := playbackProgressTarget(effectiveFile)
|
|
if targetID != "" {
|
|
event := h.scrobbleEventForSession(r.Context(), session, targetID, float64(effectiveFile.Duration), session.Position)
|
|
if err := h.WatchScrobbler.ScrobbleStart(r.Context(), event); err != nil {
|
|
slog.WarnContext(r.Context(), "failed to queue watch provider start scrobble", "component", "api", "session", session.ID, "error", err)
|
|
}
|
|
}
|
|
}
|
|
if h.ChapterThumbnailQueuer != nil && effectiveFile != nil {
|
|
slog.InfoContext(r.Context(),
|
|
"queueing chapter thumbnails", "component", "api",
|
|
"source", "playback_start",
|
|
"content_id", effectiveFile.ContentID,
|
|
"file_id", effectiveFile.ID,
|
|
"target_seconds", session.Position,
|
|
)
|
|
h.ChapterThumbnailQueuer.QueuePriorityFileAtPosition(r.Context(), effectiveFile.ID, session.Position)
|
|
}
|
|
h.maybeQueueLazyPlaybackMarkers(r.Context(), session, effectiveFile)
|
|
h.persistSeriesSelectionsV3(r.Context(), userID, profileID, effectiveFile, plannedAudioTrackIndexV3(result, audioIndex))
|
|
h.syncSessionsNow(r.Context(), "v3_start")
|
|
h.enqueueRouteEventV3(playback.RouteEventRecordV3{RouteEventV3: playback.RouteEventV3{ProtocolVersion: playback.ProtocolV3, PlaybackAttemptID: req.PlaybackAttemptID, SessionID: session.ID, PlanID: result.Plan.PlanID, Event: playback.RouteEventPlanSelectedV3, AppliedQuirkIDs: appliedQuirkIDsV3(result.Plan), QuirkRegistryRevision: appliedQuirkRevisionV3(result.Plan), OutputContextID: req.ClientPlaybackContext.Output.OutputContextID}, UserID: userID, ProfileID: profileID, ClientName: clientInfo.Name, ClientVersion: clientInfo.Version, ClientModel: req.ClientPlaybackContext.Device.Model})
|
|
return response, nil
|
|
}
|
|
|
|
// persistSeriesSelectionsV3 records the version and audio-track choices this
|
|
// plan settled on, so the next episode of the same series opens with them
|
|
// already applied. The catalog reads both preferences back
|
|
// (internal/catalog/detail.go), and v3 is now the only writer: the legacy start
|
|
// and audio-PATCH endpoints that used to record them are gone.
|
|
func (h *PlaybackHandler) persistSeriesSelectionsV3(ctx context.Context, userID int, profileID string, file *models.MediaFile, audioTrackIndex int) {
|
|
h.persistSeriesPlaybackPreference(ctx, userID, profileID, file)
|
|
h.persistAudioPreference(ctx, userID, profileID, file, audioTrackIndex)
|
|
}
|
|
|
|
func (h *PlaybackHandler) prepareTransportV3(r *http.Request, session *playback.Session, file *models.MediaFile, result playback.PlannerResultV3) (preparedTransportV3, *transportErrorV3) {
|
|
if result.Plan.Delivery != playback.DeliveryTranscodeHLSV3 && result.Plan.Delivery != playback.DeliveryRemuxHLSV3 {
|
|
return h.prepareIdentityTransportV3(session, result), nil
|
|
}
|
|
if h.NodePlanner != nil {
|
|
plan := h.planNodeSessionV3(r.Context(), session, result)
|
|
if plan.TranscodeNode != nil {
|
|
transformations, err := h.remoteTransformationsV3(r.Context(), plan.TranscodeNode.URL)
|
|
if err == nil {
|
|
err = validateAdvertisedTransformationsV3(result.Plan, transformations)
|
|
}
|
|
if err == nil {
|
|
transport, transportErr := h.prepareRemoteTransportV3(r, session, file, result, plan)
|
|
if transportErr != nil {
|
|
if releaser, ok := h.NodePlanner.(sessionReservationReleaserV3); ok {
|
|
releaser.ReleaseSession(session.ID)
|
|
}
|
|
}
|
|
return transport, transportErr
|
|
}
|
|
slog.WarnContext(r.Context(), "protocol v3 transcode node capability mismatch", "node", plan.TranscodeNode.URL, "error", err)
|
|
if releaser, ok := h.NodePlanner.(sessionReservationReleaserV3); ok {
|
|
releaser.ReleaseSession(session.ID)
|
|
}
|
|
if !nodepool.LocalTranscodeFallbackAllowed(r.Context(), h.SettingsRepo) {
|
|
return preparedTransportV3{}, &transportErrorV3{reason: "transcode_node_capability_unavailable", message: "No transcode node can execute the selected playback recipe.", retryable: true, cause: err}
|
|
}
|
|
}
|
|
if !nodepool.LocalTranscodeFallbackAllowed(r.Context(), h.SettingsRepo) {
|
|
return preparedTransportV3{}, &transportErrorV3{reason: "capacity_unavailable", message: "No transcode node is available and local fallback is disabled.", retryable: true}
|
|
}
|
|
}
|
|
// Capability-union planning may select transformations only pooled nodes
|
|
// can execute; the local binary must prove it carries the recipe before
|
|
// this fallback spawns an ffmpeg that would fail at runtime. Retryable:
|
|
// a capable node freeing up satisfies the same plan. Transformation-free
|
|
// plans skip the check (and the local probe behind it) entirely.
|
|
if planRequiresServerTransformationsV3(result.Plan) {
|
|
if err := validateAdvertisedTransformationsV3(result.Plan, h.transformationRegistryV3(r.Context()).Advertised()); err != nil {
|
|
return preparedTransportV3{}, &transportErrorV3{reason: "transcode_node_capability_unavailable", message: "No available transcode executor can run the selected playback recipe.", retryable: true, cause: err}
|
|
}
|
|
}
|
|
return h.prepareLocalTransportV3(r, session, file, result)
|
|
}
|
|
|
|
// planRequiresServerTransformationsV3 reports whether the plan carries any
|
|
// transformation the serving executor (local binary or transcode node) must
|
|
// perform, as opposed to client-executed ones.
|
|
func planRequiresServerTransformationsV3(plan *playback.PlanV3) bool {
|
|
if plan == nil {
|
|
return false
|
|
}
|
|
for _, transformation := range plan.Transformations {
|
|
if !strings.EqualFold(transformation.Executor, playback.ExecutorClientV3) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (h *PlaybackHandler) prepareIdentityTransportV3(session *playback.Session, result playback.PlannerResultV3) preparedTransportV3 {
|
|
routeSession := *session
|
|
routeSession.PlayMethod = result.PlayMethod
|
|
routeSession.BasePlayMethod = result.PlayMethod
|
|
routeSession.MediaFileID = result.Plan.EffectiveMediaFileID
|
|
routeSession.AudioTrackIndex = plannedAudioTrackIndexV3(result, session.AudioTrackIndex)
|
|
routeSession.TranscodeAudio = result.TranscodeAudio
|
|
routeSession.TargetAudioCodec = result.TargetAudioCodec
|
|
routeSession.TargetAudioChannels = result.TargetAudioChannels
|
|
routeSession.TargetAudioBitrateKbps = result.TargetAudioBitrateKbps
|
|
routeSession.RemuxDVMode = remuxDVModeForPlanV3(result.Plan)
|
|
previousNodeURL := session.TranscodeNodeURL
|
|
previousTransportID := remoteTransportID(session)
|
|
unlock := h.tm.LockSessionLifecycle(session.ID)
|
|
committed := false
|
|
streamURL := h.playbackStreamURL(&routeSession)
|
|
if result.Plan != nil && result.Plan.Delivery == playback.DeliveryRemuxProgressiveV3 {
|
|
configureProgressiveRemuxTimelineV3(result.Plan)
|
|
if seek := result.Plan.Timeline.StreamOriginSeconds; seek > 0 {
|
|
streamURL = appendPlaybackQueryV3(streamURL, "seek", strconv.FormatFloat(seek, 'f', -1, 64))
|
|
}
|
|
}
|
|
return preparedTransportV3{
|
|
url: streamURL,
|
|
commit: func() {
|
|
if committed {
|
|
return
|
|
}
|
|
committed = true
|
|
h.tm.CloseTranscodeSession(session.ID, "")
|
|
if previousNodeURL != "" {
|
|
h.tm.StopRemoteTranscode(previousTransportID, previousNodeURL)
|
|
}
|
|
unlock()
|
|
},
|
|
rollback: func() {
|
|
if committed {
|
|
return
|
|
}
|
|
committed = true
|
|
unlock()
|
|
},
|
|
}
|
|
}
|
|
|
|
// sessionOwnsResumeTimelineV3 reports whether the session's own position is a
|
|
// valid resume point for the item it belongs to.
|
|
//
|
|
// Resume state is keyed on the item (playbackProgressTarget resolves a file to
|
|
// its episode or content ID), but every part of a multipart presentation shares
|
|
// that key while carrying its own file-local clock. Persisting part 4's
|
|
// position would therefore store "12 minutes into the book" as the book's
|
|
// resume point. The client that stitches the parts into one timeline is the
|
|
// only party that knows the item-absolute position, and it reports that through
|
|
// the sync/progress surface instead.
|
|
//
|
|
// This is derived rather than requested: the mismatch is a property of the
|
|
// media, not of the client, so a client that forgot to ask would corrupt resume
|
|
// exactly the same way.
|
|
func sessionOwnsResumeTimelineV3(file *models.MediaFile) bool {
|
|
return file == nil || file.PresentationPartTotal <= 1
|
|
}
|
|
|
|
// preferredAudioTrackIndexV3 answers what an omitted audio track means: the
|
|
// language this profile has settled on for this series, this library, or
|
|
// generally — the same resolution the catalog performs when it publishes
|
|
// `effective_audio_track_index`, so the track a client sees on the detail page
|
|
// is the track that plays when it does not ask for one.
|
|
//
|
|
// The client sends a track identity only when the viewer picked one. Defaulting
|
|
// to ordinal zero instead would silently play the first track on the reel.
|
|
func (h *PlaybackHandler) preferredAudioTrackIndexV3(ctx context.Context, userID int, profileID string, file *models.MediaFile) (int, error) {
|
|
if file == nil || len(file.AudioTracks) == 0 || h.StoreProvider == nil {
|
|
return 0, nil
|
|
}
|
|
store, err := h.StoreProvider.ForUser(ctx, userID)
|
|
if err != nil {
|
|
slog.ErrorContext(ctx, "protocol v3 start: audio preference store lookup failed", "component", "api", "user_id", userID, "error", err)
|
|
return 0, err
|
|
}
|
|
var seriesPref *playback.AudioTrackPreference
|
|
if seriesID := h.resolveSeriesID(ctx, file); seriesID != "" {
|
|
stored, prefErr := store.GetAudioPreference(ctx, profileID, seriesID)
|
|
if prefErr != nil {
|
|
slog.ErrorContext(ctx, "protocol v3 start: series audio preference lookup failed", "component", "api", "profile_id", profileID, "series_id", seriesID, "error", prefErr)
|
|
return 0, prefErr
|
|
}
|
|
if stored != nil {
|
|
seriesPref = &playback.AudioTrackPreference{AudioTrackIndex: stored.AudioTrackIndex, AudioLanguage: stored.AudioLanguage, TrackSignature: stored.TrackSignature}
|
|
if seriesPref.AudioLanguage == playback.OriginalLanguageSentinel {
|
|
seriesPref.AudioLanguage = h.resolveOriginalLanguage(ctx, file)
|
|
}
|
|
}
|
|
}
|
|
preferredLang := resolvedProfileAudioLanguage(ctx, store, profileID)
|
|
// A library override applies only where no series-sticky choice exists;
|
|
// having watched this series in a language outranks the library default.
|
|
libraryLang := ""
|
|
if seriesPref == nil {
|
|
stored, prefErr := store.GetLibraryPlaybackPreference(ctx, profileID, file.MediaFolderID)
|
|
if prefErr != nil {
|
|
slog.ErrorContext(ctx, "protocol v3 start: library audio preference lookup failed", "component", "api", "profile_id", profileID, "library_id", file.MediaFolderID, "error", prefErr)
|
|
return 0, prefErr
|
|
}
|
|
if stored != nil {
|
|
libraryLang = stored.AudioLanguage
|
|
}
|
|
}
|
|
if preferredLang == playback.OriginalLanguageSentinel || libraryLang == playback.OriginalLanguageSentinel {
|
|
originalLang := h.resolveOriginalLanguage(ctx, file)
|
|
if preferredLang == playback.OriginalLanguageSentinel {
|
|
preferredLang = originalLang
|
|
}
|
|
if libraryLang == playback.OriginalLanguageSentinel {
|
|
libraryLang = originalLang
|
|
}
|
|
}
|
|
if libraryLang != "" {
|
|
preferredLang = libraryLang
|
|
}
|
|
return normalizeAudioTrackIndex(file, playback.SelectAudioTrack(file.AudioTracks, preferredLang, seriesPref)), nil
|
|
}
|
|
|
|
// resumePositionV3 answers what an omitted `start_position` means: resume where
|
|
// this profile left off. It runs before planning rather than after session
|
|
// creation because the plan's timeline is cut at the start position — a route
|
|
// chosen for zero and then seeked to 40 minutes is a different route.
|
|
//
|
|
// A client that wants to start over sends an explicit `start_position: 0`; only
|
|
// omission asks the server for its resume policy. Parts of a multipart item are
|
|
// skipped for the same reason their progress is not persisted: they share one
|
|
// resume point with the whole item, so a part-local seek to it is meaningless.
|
|
func (h *PlaybackHandler) resumePositionV3(ctx context.Context, userID int, profileID string, file *models.MediaFile) (*float64, error) {
|
|
if h.StoreProvider == nil || !sessionOwnsResumeTimelineV3(file) {
|
|
return nil, nil
|
|
}
|
|
targetID := playbackProgressTarget(file)
|
|
if targetID == "" {
|
|
return nil, nil
|
|
}
|
|
store, err := h.StoreProvider.ForUser(ctx, userID)
|
|
if err != nil {
|
|
slog.ErrorContext(ctx, "protocol v3 start: resume store lookup failed", "component", "api", "user_id", userID, "error", err)
|
|
return nil, err
|
|
}
|
|
progress, err := store.GetProgress(ctx, profileID, targetID)
|
|
if err != nil {
|
|
slog.ErrorContext(ctx, "protocol v3 start: resume progress lookup failed", "component", "api", "target", targetID, "error", err)
|
|
return nil, err
|
|
}
|
|
if progress == nil || progress.Completed || progress.PositionSeconds <= 0 {
|
|
return nil, nil
|
|
}
|
|
position := progress.PositionSeconds
|
|
return &position, nil
|
|
}
|
|
|
|
// A progressive remux is a freshly generated, chunked MP4 response and does
|
|
// not implement byte ranges. Its player clock therefore begins at zero at the
|
|
// requested source origin, and arbitrary seeks must request another server
|
|
// reanchor rather than issuing a Range request against the remux pipe.
|
|
func configureProgressiveRemuxTimelineV3(plan *playback.PlanV3) {
|
|
if plan == nil {
|
|
return
|
|
}
|
|
origin := plan.Timeline.SourceStartSeconds
|
|
plan.Timeline.PlayerStartSeconds = 0
|
|
plan.Timeline.StreamOriginSeconds = origin
|
|
plan.Timeline.TimelineOffsetSeconds = origin
|
|
plan.Timeline.SeekWindowStartSeconds = &origin
|
|
plan.Timeline.SeekWindowEndSeconds = nil
|
|
plan.Timeline.CanSeekAnywhere = false
|
|
plan.Timeline.SeekRestoration = "source_position"
|
|
}
|
|
|
|
func appendPlaybackQueryV3(rawURL, key, value string) string {
|
|
separator := "?"
|
|
if strings.ContainsRune(rawURL, '?') {
|
|
separator = "&"
|
|
}
|
|
return rawURL + separator + key + "=" + value
|
|
}
|
|
|
|
func (h *PlaybackHandler) prepareLocalTransportV3(r *http.Request, session *playback.Session, file *models.MediaFile, result playback.PlannerResultV3) (preparedTransportV3, *transportErrorV3) {
|
|
cfg := h.playbackConfig()
|
|
if err := os.MkdirAll(cfg.TranscodeDir, 0o755); err != nil {
|
|
return preparedTransportV3{}, &transportErrorV3{reason: "internal_error", message: "Failed to prepare the transcode directory.", cause: err}
|
|
}
|
|
outputSubdir := transportGenerationV3(session.ID, result.Plan.PlanID)
|
|
outputDir := filepath.Join(cfg.TranscodeDir, outputSubdir)
|
|
videoCodec := result.TargetVideoCodec
|
|
if result.Plan.Delivery == playback.DeliveryRemuxHLSV3 {
|
|
videoCodec = "copy"
|
|
}
|
|
sourceMetadata := sourceExecutionMetadataV3(file, result)
|
|
sourceProfile, sourceBitDepth := sourceVideoTranscodeFactsV3(file, result)
|
|
seekSeconds, startSegment := configureHLSTimelineV3(result.Plan, videoCodec, 2, sourceMetadata.DurationSeconds)
|
|
unlock := h.tm.LockSessionLifecycle(session.ID)
|
|
opts := playback.TranscodeOpts{InputPath: file.FilePath, OutputDir: outputDir, OutputSubdir: outputSubdir, SessionID: session.ID, SourceVideoCodec: sourceMetadata.VideoCodec, SourceVideoProfile: sourceProfile, SourceVideoBitDepth: sourceBitDepth, SoftwareVideoDecode: sourceMetadata.SoftwareVideoDecode, VideoBitstreamFilter: videoBitstreamFilterForPlanV3(result.Plan), SeekSeconds: seekSeconds, StartSegmentNumber: startSegment, TargetResolution: result.TargetResolution, TargetCodecVideo: videoCodec, TargetCodecAudio: result.TargetAudioCodec, TargetAudioChannels: result.TargetAudioChannels, TargetAudioBitrateKbps: result.TargetAudioBitrateKbps, TargetBitrateKbps: result.TargetBitrateKbps, SegmentDuration: 2, SegmentRetentionSeconds: cfg.SegmentRetentionSeconds, FFmpegPath: cfg.FFmpegPath, HWAccel: cfg.HWAccel, HWDevice: cfg.HWDevice, AudioTrackIndex: plannedAudioTrackIndexV3(result, session.AudioTrackIndex), SubtitleTrackIndex: result.SubtitleTransportTrackIndex, SubtitleBurnIn: result.SubtitleBurnIn, SubtitleCodec: result.SubtitleCodec, TotalDuration: sourceMetadata.DurationSeconds, FastStart: true, NodeType: playbackNodeIntegratedV3, ExecutionMode: playbackNodeIntegratedV3, FFmpegLogSink: h.FFmpegLogSink}
|
|
ts, err := h.startLocalPlaybackTransport(r.Context(), opts)
|
|
if err != nil {
|
|
unlock()
|
|
return preparedTransportV3{}, &transportErrorV3{reason: "transcode_start_failed", message: "Failed to start the playback transport.", retryable: true, cause: err}
|
|
}
|
|
if _, readyErr := ts.WaitForManifest(playback.ManifestStartupTimeout); readyErr != nil {
|
|
wasRunning := ts.IsRunning()
|
|
failedDevice := ts.Opts().HWDevice
|
|
transportErr := manifestStartupTransportErrorV3(wasRunning, readyErr)
|
|
_ = ts.Close()
|
|
if wasRunning {
|
|
unlock()
|
|
return preparedTransportV3{}, transportErr
|
|
}
|
|
|
|
// FFmpeg and GPU drivers can fail before producing their first segment
|
|
// even though the recipe is valid. Retry one clean generation, preferring
|
|
// another configured render device so a transient device failure does not
|
|
// become an immediate client-visible transport error.
|
|
retryOpts := opts
|
|
retryOpts.AvoidHWDevice = failedDevice
|
|
slog.WarnContext(r.Context(), "local transcode crashed during startup; retrying once",
|
|
"component", "playback",
|
|
"playback_session_id", session.ID,
|
|
"failed_device", failedDevice,
|
|
"configured_devices", retryOpts.HWDevice,
|
|
"error", readyErr)
|
|
ts, err = h.startLocalPlaybackTransport(r.Context(), retryOpts)
|
|
if err != nil {
|
|
unlock()
|
|
return preparedTransportV3{}, &transportErrorV3{reason: "transcode_start_failed", message: "Failed to start the playback transport.", retryable: true, cause: err}
|
|
}
|
|
if _, retryReadyErr := ts.WaitForManifest(playback.ManifestStartupTimeout); retryReadyErr != nil {
|
|
transportErr = manifestStartupTransportErrorV3(ts.IsRunning(), retryReadyErr)
|
|
_ = ts.Close()
|
|
unlock()
|
|
return preparedTransportV3{}, transportErr
|
|
}
|
|
}
|
|
card := playback.NewRecipeCard(session.UserID, session.ProfileID, file.ID, "", ts.Opts())
|
|
url := appendStreamToken(fmt.Sprintf("/playback/transcode/%s/master.m3u8", session.ID), h.signSessionToken(card))
|
|
committed := false
|
|
previousNodeURL := session.TranscodeNodeURL
|
|
previousTransportID := remoteTransportID(session)
|
|
return preparedTransportV3{
|
|
url: url,
|
|
commit: func() {
|
|
if committed {
|
|
return
|
|
}
|
|
committed = true
|
|
previous := h.tm.SwapTranscodeSession(session.ID, ts)
|
|
unlock()
|
|
if previous != nil && previous != ts {
|
|
_ = previous.Close()
|
|
}
|
|
if previousNodeURL != "" {
|
|
h.tm.StopRemoteTranscode(previousTransportID, previousNodeURL)
|
|
}
|
|
ts.SetRestartHook(func(ctx context.Context) {
|
|
h.maybeStartThrottler(ctx, ts)
|
|
h.tm.MonitorLocalTranscodeExit(session.ID, ts)
|
|
})
|
|
h.maybeStartThrottler(r.Context(), ts)
|
|
h.tm.MonitorLocalTranscodeExit(session.ID, ts)
|
|
},
|
|
rollback: func() {
|
|
if committed {
|
|
return
|
|
}
|
|
committed = true
|
|
_ = ts.Close()
|
|
unlock()
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
func manifestStartupTransportErrorV3(running bool, cause error) *transportErrorV3 {
|
|
message := "The playback transport failed before media became ready."
|
|
if running {
|
|
message = "The playback transport did not become ready in time."
|
|
}
|
|
return &transportErrorV3{reason: "transcode_start_failed", message: message, retryable: running, cause: cause}
|
|
}
|
|
|
|
func (h *PlaybackHandler) prepareRemoteTransportV3(r *http.Request, session *playback.Session, file *models.MediaFile, result playback.PlannerResultV3, nodePlan nodepool.Plan) (preparedTransportV3, *transportErrorV3) {
|
|
node := nodePlan.TranscodeNode
|
|
transportID := transportGenerationV3(session.ID, result.Plan.PlanID)
|
|
videoCodec := result.TargetVideoCodec
|
|
if result.Plan.Delivery == playback.DeliveryRemuxHLSV3 {
|
|
videoCodec = "copy"
|
|
}
|
|
sourceMetadata := sourceExecutionMetadataV3(file, result)
|
|
sourceProfile, sourceBitDepth := sourceVideoTranscodeFactsV3(file, result)
|
|
seekSeconds, startSegment := configureHLSTimelineV3(result.Plan, videoCodec, 2, sourceMetadata.DurationSeconds)
|
|
req := transcodenode.TranscodeStartRequest{SessionID: transportID, InputPath: file.FilePath, SourceVideoCodec: sourceMetadata.VideoCodec, SourceVideoProfile: sourceProfile, SourceVideoBitDepth: sourceBitDepth, SoftwareVideoDecode: sourceMetadata.SoftwareVideoDecode, VideoBitstreamFilter: videoBitstreamFilterForPlanV3(result.Plan), SeekSeconds: seekSeconds, StartSegmentNumber: startSegment, TargetResolution: result.TargetResolution, TargetCodecVideo: videoCodec, TargetCodecAudio: result.TargetAudioCodec, TargetAudioChannels: result.TargetAudioChannels, TargetAudioBitrateKbps: result.TargetAudioBitrateKbps, TargetBitrateKbps: result.TargetBitrateKbps, SegmentDuration: 2, HWAccel: h.playbackConfig().HWAccel, AudioTrackIndex: plannedAudioTrackIndexV3(result, session.AudioTrackIndex), SubtitleTrackIndex: result.SubtitleTransportTrackIndex, SubtitleBurnIn: result.SubtitleBurnIn, SubtitleCodec: result.SubtitleCodec, TotalDuration: sourceMetadata.DurationSeconds, RequireReady: true}
|
|
nodeResp, status, err := h.startRemotePlaybackTransport(r.Context(), node.URL, req)
|
|
if err != nil {
|
|
// A timeout can fire after the node actually started the job; the
|
|
// stop is a harmless 404 when it never did, and reaps an orphan
|
|
// full-length transcode when it did.
|
|
h.tm.StopRemoteTranscode(transportID, node.URL)
|
|
return preparedTransportV3{}, &transportErrorV3{reason: "transcode_node_unavailable", message: "The selected transcode node is unavailable.", retryable: true, cause: err}
|
|
}
|
|
if status != http.StatusAccepted {
|
|
h.tm.StopRemoteTranscode(transportID, node.URL)
|
|
return preparedTransportV3{}, &transportErrorV3{reason: "transcode_start_failed", message: "The selected transcode node rejected the playback transport.", retryable: true}
|
|
}
|
|
hw := firstNonEmptyHandlerV3(strings.TrimSpace(nodeResp.HWAccel), strings.TrimSpace(req.HWAccel))
|
|
card := playback.NewRecipeCard(session.UserID, session.ProfileID, file.ID, node.URL, playback.TranscodeOpts{InputPath: req.InputPath, SessionID: session.ID, TranscodeTransportID: transportID, SourceVideoCodec: req.SourceVideoCodec, SourceVideoProfile: req.SourceVideoProfile, SourceVideoBitDepth: req.SourceVideoBitDepth, SoftwareVideoDecode: req.SoftwareVideoDecode, VideoBitstreamFilter: req.VideoBitstreamFilter, SeekSeconds: req.SeekSeconds, StartSegmentNumber: req.StartSegmentNumber, TargetResolution: req.TargetResolution, TargetCodecVideo: req.TargetCodecVideo, TargetCodecAudio: req.TargetCodecAudio, TargetAudioChannels: req.TargetAudioChannels, TargetAudioBitrateKbps: req.TargetAudioBitrateKbps, TargetBitrateKbps: req.TargetBitrateKbps, SegmentDuration: req.SegmentDuration, HWAccel: hw, AudioTrackIndex: req.AudioTrackIndex, SubtitleTrackIndex: req.SubtitleTrackIndex, SubtitleBurnIn: req.SubtitleBurnIn, SubtitleCodec: req.SubtitleCodec, TotalDuration: req.TotalDuration})
|
|
url := h.buildProxyManifestURL(card, nodePlan.ProxyNode)
|
|
committed := false
|
|
previousNodeURL := session.TranscodeNodeURL
|
|
previousTransportID := remoteTransportID(session)
|
|
unlock := h.tm.LockSessionLifecycle(session.ID)
|
|
return preparedTransportV3{url: url, nodeURL: node.URL, transportID: transportID, commit: func() {
|
|
if committed {
|
|
return
|
|
}
|
|
committed = true
|
|
h.tm.CloseTranscodeSession(session.ID, "")
|
|
if previousNodeURL != "" {
|
|
h.tm.StopRemoteTranscode(previousTransportID, previousNodeURL)
|
|
}
|
|
unlock()
|
|
}, rollback: func() {
|
|
if committed {
|
|
return
|
|
}
|
|
committed = true
|
|
h.tm.StopRemoteTranscode(transportID, node.URL)
|
|
// The accepted node job is gone; drop the planner reservation too so
|
|
// repeated failed starts cannot pin the node's max-job or bandwidth
|
|
// budget until the reservation ages out.
|
|
if releaser, ok := h.NodePlanner.(sessionReservationReleaserV3); ok {
|
|
releaser.ReleaseSession(session.ID)
|
|
}
|
|
unlock()
|
|
}}, nil
|
|
}
|
|
|
|
func sourceExecutionMetadataV3(file *models.MediaFile, result playback.PlannerResultV3) playback.SourceExecutionMetadataV3 {
|
|
if result.FrozenSourceMetadata != nil {
|
|
return *result.FrozenSourceMetadata
|
|
}
|
|
if file == nil {
|
|
return playback.SourceExecutionMetadataV3{}
|
|
}
|
|
videoCodec, profile, bitDepth := playback.SourceVideoTranscodeFacts(file)
|
|
return playback.SourceExecutionMetadataV3{
|
|
VideoCodec: videoCodec,
|
|
SoftwareVideoDecode: playback.RequiresSoftwareVideoDecode(videoCodec, profile, bitDepth),
|
|
DurationSeconds: float64(file.Duration),
|
|
}
|
|
}
|
|
|
|
func sourceVideoTranscodeFactsV3(file *models.MediaFile, result playback.PlannerResultV3) (string, int) {
|
|
if result.FrozenSourceMetadata != nil {
|
|
return "", 0
|
|
}
|
|
_, profile, bitDepth := playback.SourceVideoTranscodeFacts(file)
|
|
return profile, bitDepth
|
|
}
|
|
|
|
func (h *PlaybackHandler) v3SessionStreamState(ctx context.Context, session *playback.Session, file *models.MediaFile, result playback.PlannerResultV3, transport preparedTransportV3) playback.SessionStreamState {
|
|
state := playback.SessionStreamState{PlayMethod: result.PlayMethod, BasePlayMethod: result.PlayMethod, AudioTrackIndex: plannedAudioTrackIndexV3(result, session.AudioTrackIndex), TranscodeAudio: result.TranscodeAudio, RemuxDVMode: remuxDVModeForPlanV3(result.Plan), TranscodeNodeURL: transport.nodeURL, TranscodeTransportID: transport.transportID, TranscodeRouteSet: true, ClientIP: clientip.FromContext(ctx), ClientName: session.ClientName, ClientVersion: session.ClientVersion, ClientUserAgent: session.ClientUserAgent, StreamBitrateKbps: result.TargetBitrateKbps, TargetVideoCodec: result.TargetVideoCodec, TargetAudioCodec: result.TargetAudioCodec, TargetAudioChannels: result.TargetAudioChannels, TargetAudioBitrateKbps: result.TargetAudioBitrateKbps, TargetResolution: result.TargetResolution, SubtitleTrackIndex: result.SubtitleTransportTrackIndex, SubtitleBurnIn: result.SubtitleBurnIn}
|
|
if result.Plan != nil && (result.Plan.Delivery == playback.DeliveryTranscodeHLSV3 || result.Plan.Delivery == playback.DeliveryRemuxHLSV3) {
|
|
state.SegmentDuration = 2
|
|
}
|
|
if state.StreamBitrateKbps <= 0 {
|
|
state.StreamBitrateKbps = result.TargetAudioBitrateKbps
|
|
}
|
|
if state.StreamBitrateKbps <= 0 {
|
|
state.StreamBitrateKbps = fileBitrateKbps(file)
|
|
}
|
|
return state
|
|
}
|
|
|
|
func (h *PlaybackHandler) updateV3SessionState(ctx context.Context, session *playback.Session, file *models.MediaFile, result playback.PlannerResultV3, transport preparedTransportV3) error {
|
|
return h.sessionMgr.UpdateStreamState(session.ID, h.v3SessionStreamState(ctx, session, file, result, transport))
|
|
}
|
|
|
|
func plannedAudioTrackIndexV3(result playback.PlannerResultV3, fallback int) int {
|
|
if result.Plan != nil && result.Plan.SelectedTracks.Audio != nil && result.Plan.SelectedTracks.Audio.Index != nil {
|
|
return *result.Plan.SelectedTracks.Audio.Index
|
|
}
|
|
return fallback
|
|
}
|
|
|
|
func transportGenerationV3(sessionID, planID string) string {
|
|
planSuffix := strings.TrimPrefix(planID, "plan:")
|
|
if len(planSuffix) > 12 {
|
|
planSuffix = planSuffix[:12]
|
|
}
|
|
return sessionID + "-" + planSuffix + "-" + uuid.NewString()[:8]
|
|
}
|
|
|
|
// attachSubtitleArtifactV3 republishes the plan's subtitle inventory with
|
|
// session-scoped URLs, then resolves the plan's selected ordinal against it and
|
|
// stamps that entry's URL onto the artifact. Publishing and resolution share one
|
|
// ordering implementation, so an artifact URL can never point at a different
|
|
// track than the inventory entry the client selected.
|
|
//
|
|
// The inventory is scoped unconditionally, not only when a track is selected:
|
|
// spec §8 makes it the authoritative track list and says a sidecar entry carries
|
|
// a `url` "once a session exists to scope it to" — which is true here for every
|
|
// entry, whatever the current selection is. Gating it on the selection published
|
|
// a URL-less menu whenever playback started with subtitles off, so a client
|
|
// building its picker from the inventory (the Cast receiver's text tracks, for
|
|
// one) had nothing fetchable to offer.
|
|
func (h *PlaybackHandler) attachSubtitleArtifactV3(ctx context.Context, sessionID string, file *models.MediaFile, plan *playback.PlanV3, selectedIndex int, recipe *playback.ExecutableRecipeV3) error {
|
|
if plan == nil || file == nil {
|
|
return nil
|
|
}
|
|
var frozenDownloaded *subtitles.DownloadedSubtitle
|
|
if recipe != nil && recipe.SubtitleSource == playback.SubtitleSourceDownloadedV3 {
|
|
if h == nil || h.SubtitleRepo == nil || recipe.DownloadedSubtitleID <= 0 {
|
|
return errors.New("the frozen downloaded subtitle is unavailable")
|
|
}
|
|
selected, err := h.SubtitleRepo.GetDownloadedSubtitle(ctx, recipe.DownloadedSubtitleID)
|
|
if err != nil {
|
|
return wrapSubtitleStoreErrorV3(err)
|
|
}
|
|
if selected == nil || selected.MediaFileID != file.ID {
|
|
return errors.New("the frozen downloaded subtitle is unavailable for the selected media file")
|
|
}
|
|
frozenDownloaded = selected
|
|
}
|
|
inventory := playback.ScopeSubtitleInventoryV3(sessionID, file, plan.Subtitle.Inventory)
|
|
// A plan restored from JSON no longer carries the server-only downloaded
|
|
// row IDs. Rebuild only in that case; a fresh plan stays on the exact
|
|
// planning snapshot instead of listing a mutable repository twice.
|
|
if playback.SubtitleInventoryNeedsDownloadedIdentityV3(plan.Subtitle.Inventory) {
|
|
if h == nil || h.SubtitleRepo == nil {
|
|
return errors.New("the downloaded subtitle inventory is unavailable")
|
|
}
|
|
downloaded, err := h.SubtitleRepo.ListDownloadedSubtitles(ctx, file.ID)
|
|
if err != nil {
|
|
return wrapSubtitleStoreErrorV3(err)
|
|
}
|
|
inventory = playback.SubtitleInventoryV3(sessionID, file, downloadedSubtitleEntriesV3(file, downloaded))
|
|
}
|
|
plan.Subtitle.Inventory = inventory
|
|
if selectedIndex < 0 || (plan.Subtitle.Mode != playback.SubtitleRenderV3 && plan.Subtitle.Mode != playback.SubtitleConvertV3) {
|
|
return nil
|
|
}
|
|
item, ok := playback.SubtitleInventoryItemAtV3(inventory, selectedIndex)
|
|
if !ok && frozenDownloaded == nil {
|
|
return errors.New("selected subtitle artifact is absent from the frozen inventory")
|
|
}
|
|
if frozenDownloaded == nil && item.URL == "" {
|
|
return fmt.Errorf("subtitle track %d is %s and has no fetchable artifact", selectedIndex, item.Delivery)
|
|
}
|
|
format := strings.ToLower(item.Codec)
|
|
mime := subtitleMIMEV3(format)
|
|
url := item.URL
|
|
if frozenDownloaded != nil {
|
|
format = strings.ToLower(string(frozenDownloaded.Format))
|
|
mime = subtitleMIMEV3(format)
|
|
url = playback.DownloadedSubtitleStreamURLV3(sessionID, selectedIndex, string(frozenDownloaded.Format), file.ID, frozenDownloaded.ID)
|
|
// The plan's selected ordinal must advertise the same opaque URL as the
|
|
// artifact even if another downloaded row was inserted before a seek.
|
|
for index := range plan.Subtitle.Inventory {
|
|
if plan.Subtitle.Inventory[index].CombinedIndex == selectedIndex {
|
|
plan.Subtitle.Inventory[index].URL = url
|
|
plan.Subtitle.Inventory[index].Codec = string(frozenDownloaded.Format)
|
|
break
|
|
}
|
|
}
|
|
}
|
|
if plan.Subtitle.Mode == playback.SubtitleConvertV3 {
|
|
format = playback.SubtitleFormatVTTV3
|
|
mime = playback.SubtitleMIMEVTTV3
|
|
url = forceSubtitleExtensionV3(url, playback.SubtitleExtVTTV3)
|
|
}
|
|
plan.Subtitle.Artifact = &playback.SubtitleArtifactV3{URL: url, MIMEType: mime, Format: format, TimingOriginSeconds: plan.Timeline.StreamOriginSeconds}
|
|
return nil
|
|
}
|
|
|
|
// downloadedSubtitleInventoryV3 lists the downloaded and AI-generated tracks
|
|
// that follow the file's own tracks in the combined-ordinal space. The
|
|
// repository orders by created_at, so the ordinals it produces are stable.
|
|
func (h *PlaybackHandler) downloadedSubtitleInventoryV3(ctx context.Context, file *models.MediaFile) []playback.SubtitleInventoryEntryV3 {
|
|
if h == nil || h.SubtitleRepo == nil || file == nil {
|
|
return nil
|
|
}
|
|
downloaded, err := h.SubtitleRepo.ListDownloadedSubtitles(ctx, file.ID)
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
return downloadedSubtitleEntriesV3(file, downloaded)
|
|
}
|
|
|
|
// downloadedSubtitleEntriesV3 converts downloaded rows into inventory entries
|
|
// at the ordinals that follow the file's external and embedded tracks.
|
|
func downloadedSubtitleEntriesV3(file *models.MediaFile, downloaded []subtitles.DownloadedSubtitle) []playback.SubtitleInventoryEntryV3 {
|
|
if file == nil {
|
|
return nil
|
|
}
|
|
base := len(file.ExternalSubtitles) + len(file.SubtitleTracks)
|
|
result := make([]playback.SubtitleInventoryEntryV3, 0, len(downloaded))
|
|
for index, value := range downloaded {
|
|
result = append(result, playback.SubtitleInventoryEntryV3{
|
|
CombinedIndex: base + index,
|
|
Codec: string(value.Format),
|
|
Source: playback.SubtitleSourceDownloadedV3,
|
|
Language: value.Language,
|
|
Label: downloadedSubtitleLabelV3(value),
|
|
HearingImpaired: value.HearingImpaired,
|
|
DownloadedSubtitleID: value.ID,
|
|
})
|
|
}
|
|
return result
|
|
}
|
|
|
|
func downloadedSubtitleLabelV3(value subtitles.DownloadedSubtitle) string {
|
|
if value.ReleaseName == "" && value.Provider == "" {
|
|
return ""
|
|
}
|
|
return value.ReleaseName + " (" + value.Provider + ")"
|
|
}
|
|
|
|
// HandleReplanPlaybackV3 provides persistent idempotency and preserves the old
|
|
// transport until a successor has entered its startup state and the new plan is
|
|
// durably committed.
|
|
func (h *PlaybackHandler) HandleReplanPlaybackV3(w http.ResponseWriter, r *http.Request) {
|
|
userID := apimw.GetUserID(r.Context())
|
|
profileID := apimw.GetProfileID(r.Context())
|
|
if userID == 0 || profileID == "" {
|
|
writeError(w, http.StatusUnauthorized, "unauthorized", "Authentication and profile are required")
|
|
return
|
|
}
|
|
body, err := readBoundedV3Body(w, r, maxPlaybackV3BodyBytes)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body")
|
|
return
|
|
}
|
|
var req playback.ReplanRequestV3
|
|
if err := json.Unmarshal(body, &req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "Invalid replan request")
|
|
return
|
|
}
|
|
// Reject malformed identity/bounds before doing any session lookup. When
|
|
// client_features is omitted, temporarily allow the only validation rule
|
|
// that depends on the durable start request; the authoritative merge and a
|
|
// second full validation happen after the attempt is loaded below.
|
|
preflightReq := req
|
|
if preflightReq.ClientFeatures == nil {
|
|
preflightReq.ClientFeatures = []string{playback.FeatureClientVideoTransforms}
|
|
}
|
|
if err := preflightReq.Validate(); err != nil {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "Invalid replan request")
|
|
return
|
|
}
|
|
sessionID := chiURLParamV3(r, "session_id")
|
|
releaseSlot, err := h.acquireReplanSlotV3(r.Context())
|
|
if err != nil {
|
|
writeError(w, http.StatusServiceUnavailable, "replan_capacity_exhausted", "The server is replanning too many sessions; retry shortly")
|
|
return
|
|
}
|
|
defer releaseSlot()
|
|
unlockReplan := h.lockReplanV3(sessionID)
|
|
defer unlockReplan()
|
|
unlockStore, err := h.PlanStoreV3.AcquireSessionLock(r.Context(), sessionID)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to serialize the replan request")
|
|
return
|
|
}
|
|
defer unlockStore()
|
|
record, err := h.PlanStoreV3.GetAttempt(r.Context(), sessionID)
|
|
if err != nil {
|
|
// A store outage must read as retryable, not as the session being
|
|
// gone: clients tear playback down on session_not_found.
|
|
if !errors.Is(err, playback.ErrSessionNotFound) {
|
|
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to load the playback attempt")
|
|
return
|
|
}
|
|
writePlaybackSessionNotFound(w)
|
|
return
|
|
}
|
|
if record.UserID != userID || record.ProfileID != profileID {
|
|
writeError(w, http.StatusForbidden, "forbidden", "Session belongs to another profile")
|
|
return
|
|
}
|
|
if record.PlaybackAttemptID != req.PlaybackAttemptID {
|
|
writeError(w, http.StatusConflict, "stale_playback_plan", "The failed plan is no longer current")
|
|
return
|
|
}
|
|
// Replan feature advertisement is optional. Validate transformations against
|
|
// the durable start-time features when the client omits the unchanged list;
|
|
// otherwise a valid replan can be rejected before executeReplanV3 gets the
|
|
// chance to perform the same merge.
|
|
if req.ClientFeatures == nil {
|
|
req.ClientFeatures = append([]string(nil), record.NormalizedRequest.ClientFeatures...)
|
|
}
|
|
if err := req.Validate(); err != nil {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "Invalid replan request")
|
|
return
|
|
}
|
|
if _, err := h.sessionMgr.GetSession(sessionID); err != nil {
|
|
writePlaybackSessionNotFound(w)
|
|
return
|
|
}
|
|
digestBytes := sha256.Sum256(body)
|
|
digest := hex.EncodeToString(digestBytes[:])
|
|
lease, err := h.PlanStoreV3.BeginReplan(
|
|
r.Context(),
|
|
sessionID,
|
|
req.ReplanRequestID,
|
|
digest,
|
|
record.CurrentReplanRequestID,
|
|
time.Now().Add(replanLeaseDurationV3),
|
|
)
|
|
if errors.Is(err, playback.ErrIdempotencyKeyReusedV3) {
|
|
writeError(w, http.StatusConflict, "idempotency_key_reused", "The replan request ID was reused with different input")
|
|
return
|
|
}
|
|
if errors.Is(err, playback.ErrStaleReplanLeaseV3) {
|
|
writeError(w, http.StatusConflict, "stale_playback_plan", "A newer replacement plan is already active")
|
|
return
|
|
}
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to reserve the replan request")
|
|
return
|
|
}
|
|
if lease.State == playback.ReplanLeaseInFlightV3 {
|
|
writeError(w, http.StatusConflict, "replan_in_progress", "An identical replan is still in progress")
|
|
return
|
|
}
|
|
if lease.State == playback.ReplanLeaseCompletedV3 {
|
|
if record.CurrentReplanRequestID != req.ReplanRequestID || !completedReplanResponseMatchesAttemptV3(lease.Response, record) {
|
|
writeError(w, http.StatusConflict, "stale_playback_plan", "A newer replacement plan is already active")
|
|
return
|
|
}
|
|
if _, err := h.sessionMgr.GetSession(sessionID); err != nil {
|
|
writePlaybackSessionNotFound(w)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.WriteHeader(http.StatusOK)
|
|
_, _ = w.Write(lease.Response)
|
|
return
|
|
}
|
|
leaseCompleted := false
|
|
defer func() {
|
|
if leaseCompleted {
|
|
return
|
|
}
|
|
releaseCtx, cancel := context.WithTimeout(context.WithoutCancel(r.Context()), replanReleaseTimeoutV3)
|
|
defer cancel()
|
|
if err := h.PlanStoreV3.ReleaseReplan(releaseCtx, sessionID, req.ReplanRequestID, lease.LeaseToken); err != nil {
|
|
slog.ErrorContext(r.Context(), "protocol v3 replan lease release failed", "component", "api", "session", sessionID, "replan_request_id", req.ReplanRequestID, "error", err)
|
|
}
|
|
}()
|
|
if record.CurrentPlanID != req.FailedPlanID {
|
|
writeError(w, http.StatusConflict, "stale_playback_plan", "The failed plan is no longer current")
|
|
return
|
|
}
|
|
response, updated, transport, replanErr := h.executeReplanV3(r, record, req)
|
|
if replanErr != nil {
|
|
if transport != nil {
|
|
transport.rollback()
|
|
}
|
|
response := playback.NewTerminalResponseV3(replanErr.reason, replanErr.message, replanErr.retryable)
|
|
encoded, _ := json.Marshal(response)
|
|
terminalRecord := *record
|
|
terminalRecord.CurrentReplanRequestID = req.ReplanRequestID
|
|
if err := h.PlanStoreV3.CompleteReplan(r.Context(), sessionID, req.ReplanRequestID, lease.LeaseToken, record.CurrentReplanRequestID, encoded, terminalRecord); err != nil {
|
|
if errors.Is(err, playback.ErrReplanSupersededV3) {
|
|
writeError(w, http.StatusConflict, "stale_playback_plan", "A newer replacement plan is already active")
|
|
return
|
|
}
|
|
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to persist the terminal replan decision")
|
|
return
|
|
}
|
|
leaseCompleted = true
|
|
writeJSON(w, http.StatusOK, response)
|
|
return
|
|
}
|
|
updated.CurrentReplanRequestID = req.ReplanRequestID
|
|
encoded, _ := json.Marshal(response)
|
|
var rollbackSession func() error
|
|
if transport != nil && transport.applySession != nil {
|
|
var err error
|
|
rollbackSession, err = transport.applySession()
|
|
if err != nil {
|
|
transport.rollback()
|
|
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to commit the live replacement session")
|
|
return
|
|
}
|
|
}
|
|
if err := h.PlanStoreV3.CompleteReplan(r.Context(), sessionID, req.ReplanRequestID, lease.LeaseToken, record.CurrentReplanRequestID, encoded, updated); err != nil {
|
|
rollbackFailed := false
|
|
if rollbackSession != nil {
|
|
if rollbackErr := rollbackSession(); rollbackErr != nil {
|
|
rollbackFailed = true
|
|
slog.ErrorContext(r.Context(), "protocol v3 replacement rollback failed", "session", sessionID, "error", rollbackErr)
|
|
}
|
|
}
|
|
if transport != nil {
|
|
transport.rollback()
|
|
}
|
|
if rollbackFailed {
|
|
_ = h.stopPlaybackSessionByID(context.WithoutCancel(r.Context()), sessionID, false)
|
|
}
|
|
if errors.Is(err, playback.ErrReplanSupersededV3) {
|
|
writeError(w, http.StatusConflict, "stale_playback_plan", "A newer replacement plan is already active")
|
|
return
|
|
}
|
|
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to commit the replacement plan")
|
|
return
|
|
}
|
|
leaseCompleted = true
|
|
if transport != nil {
|
|
transport.commit()
|
|
if transport.afterDurableCommit != nil {
|
|
transport.afterDurableCommit()
|
|
}
|
|
}
|
|
writeJSON(w, http.StatusOK, response)
|
|
}
|
|
|
|
func (h *PlaybackHandler) executeReplanV3(r *http.Request, record *playback.AttemptRecordV3, req playback.ReplanRequestV3) (playback.DecisionResponseV3, playback.AttemptRecordV3, *preparedTransportV3, *transportErrorV3) {
|
|
reservationHeld := false
|
|
reservationHandedOff := false
|
|
cancelReservation := func() {
|
|
if reservationHeld {
|
|
if canceller, ok := h.sessionMgr.(replacementReservationCancellerV3); ok {
|
|
canceller.CancelReplacementReservation(record.SessionID)
|
|
}
|
|
reservationHeld = false
|
|
}
|
|
}
|
|
defer func() {
|
|
if !reservationHandedOff {
|
|
cancelReservation()
|
|
}
|
|
}()
|
|
start := record.NormalizedRequest
|
|
operation := req.EffectiveOperation()
|
|
seekReanchor := operation == playback.ReplanOperationSeekReanchorV3
|
|
seekFailureRecovery := operation == playback.ReplanOperationSeekFailureRecoveryV3
|
|
seekScopedRecovery := seekReanchor || seekFailureRecovery
|
|
trackChange := operation == playback.ReplanOperationTrackChangeV3
|
|
qualityChange := operation == playback.ReplanOperationQualityChangeV3
|
|
// User-intent operations replace the legacy audio PATCH and client-recipe
|
|
// transcode start. Nothing failed, so their previous route stays eligible:
|
|
// neither attempted-key history nor the failed-plan exclusion applies.
|
|
userIntentOperation := trackChange || qualityChange
|
|
intentChange := false
|
|
if seekScopedRecovery {
|
|
if err := validateSeekRecoveryRequestV3(record, req); err != nil {
|
|
reason := "seek_reanchor_intent_mismatch"
|
|
if seekFailureRecovery {
|
|
reason = "seek_failure_recovery_intent_mismatch"
|
|
}
|
|
return playback.DecisionResponseV3{}, *record, nil, &transportErrorV3{
|
|
reason: reason,
|
|
message: err.Error(),
|
|
}
|
|
}
|
|
// Reconstruct the complete route intent from the durable current attempt.
|
|
// A seek request is not an authority boundary for replacing capability or
|
|
// device evidence: accepting those fields here could make the same file
|
|
// select a materially different route based on request-only claims.
|
|
start.FileID = record.EffectiveMediaFileID
|
|
start.StartPosition = &req.PositionSeconds
|
|
applySelectedTracksToStartV3(&start, record.CurrentPlan.SelectedTracks)
|
|
} else {
|
|
// Failure replans may omit unchanged tracks. The durable current plan
|
|
// holds the authoritative effective-file selections; the normalized
|
|
// request can still carry requested-edition identities after an
|
|
// alternate-version fallback, and validating those against the
|
|
// effective file would reject an otherwise valid replan. Seed from
|
|
// the plan first, then overlay the request's explicit changes.
|
|
applySelectedTracksToStartV3(&start, record.CurrentPlan.SelectedTracks)
|
|
switch {
|
|
case trackChange:
|
|
intentChange = audioSelectionDiffersFromStartV3(req.SelectedTracks, start) ||
|
|
subtitleSelectionDiffersFromStartV3(req.SelectedTracks, start)
|
|
case qualityChange:
|
|
nextQuality, _ := playback.NormalizeQualityV3(req.QualityPreference)
|
|
intentChange = nextQuality != start.QualityPreference
|
|
default:
|
|
switch req.Failure.Classification {
|
|
case "quality_changed":
|
|
nextQuality, _ := playback.NormalizeQualityV3(req.QualityPreference)
|
|
intentChange = nextQuality != start.QualityPreference
|
|
case "audio_track_changed":
|
|
intentChange = audioSelectionDiffersFromStartV3(req.SelectedTracks, start)
|
|
case "subtitle_track_changed":
|
|
intentChange = subtitleSelectionDiffersFromStartV3(req.SelectedTracks, start)
|
|
case "output_route_changed":
|
|
intentChange = req.ClientPlaybackContext.Output.OutputContextID != start.ClientPlaybackContext.Output.OutputContextID
|
|
}
|
|
}
|
|
// Failure replans use the current effective file. Quality/output intent may
|
|
// restart source selection from the requested edition, but a track change
|
|
// is expressed in the mounted alternate's inventory and must stay pinned to
|
|
// that file or its combined ordinals can select unrelated tracks.
|
|
start.FileID = record.EffectiveMediaFileID
|
|
if intentChange && !trackChange {
|
|
start.FileID = record.RequestedMediaFileID
|
|
}
|
|
if strings.TrimSpace(req.QualityPreference) != "" {
|
|
// Replans may omit unchanged intent. Normalizing an absent quality
|
|
// would silently reset "original" or a fixed rung to "auto".
|
|
start.QualityPreference = req.QualityPreference
|
|
}
|
|
start.StartPosition = &req.PositionSeconds
|
|
start.Metered = req.Metered
|
|
start.BandwidthEstimateKbps = copyOptionalIntV3(req.BandwidthEstimateKbps)
|
|
start.BandwidthCapKbps = copyOptionalIntV3(req.BandwidthCapKbps)
|
|
start.Capabilities = req.Capabilities
|
|
start.ClientPlaybackContext = req.ClientPlaybackContext
|
|
if req.ClientFeatures != nil {
|
|
// Feature advertisement is single-location (top-level); a replan
|
|
// that sends it refreshes the durable request's copy alongside the
|
|
// capability payloads. Omission keeps the start-time features.
|
|
start.ClientFeatures = req.ClientFeatures
|
|
}
|
|
if trackChange {
|
|
// A track_change is the only operation where an omitted subtitle
|
|
// means "subtitles off". Failure, seek, and quality replans may omit
|
|
// unchanged identities and must not erase the durable selection.
|
|
applySelectedTracksToStartV3(&start, req.SelectedTracks)
|
|
} else {
|
|
applySelectedTrackOverridesToStartV3(&start, req.SelectedTracks)
|
|
}
|
|
}
|
|
requestedFallbackID := record.EffectiveMediaFileID
|
|
effectiveFallbackID := record.RequestedMediaFileID
|
|
if seekScopedRecovery {
|
|
// Edition fallback is useful for ordinary failure replans, but never for
|
|
// a seek operation: the caller asked to move within the currently mounted
|
|
// source, not to select another version when that source disappears.
|
|
requestedFallbackID = 0
|
|
effectiveFallbackID = 0
|
|
}
|
|
requestedFile, err := h.loadFileByPreferredID(r.Context(), record.RequestedMediaFileID, requestedFallbackID)
|
|
requestedEditionResolved := err == nil && requestedFile != nil && requestedFile.ID == record.RequestedMediaFileID
|
|
if err != nil || requestedFile == nil {
|
|
if !seekScopedRecovery {
|
|
return playback.DecisionResponseV3{}, *record, nil, &transportErrorV3{reason: "source_unavailable", message: "The requested media source is unavailable."}
|
|
}
|
|
// The requested edition is identity-only once another effective edition
|
|
// is mounted. Seeking must depend on that effective file remaining
|
|
// available, not on an inactive original edition still resolving.
|
|
requestedFile = &models.MediaFile{ID: record.RequestedMediaFileID}
|
|
}
|
|
plannerRequestedFile := requestedFile
|
|
if requestedFile.ID != record.RequestedMediaFileID {
|
|
// The live loader may fall back to the current effective file when the
|
|
// original edition is gone. Keep that file for metadata/remapping while
|
|
// preserving the durable requested-edition identity in every new plan.
|
|
plannerRequestedFile = &models.MediaFile{ID: record.RequestedMediaFileID}
|
|
}
|
|
currentEffectiveFile, err := h.loadFileByPreferredID(r.Context(), record.EffectiveMediaFileID, effectiveFallbackID)
|
|
if err != nil || currentEffectiveFile == nil {
|
|
return playback.DecisionResponseV3{}, *record, nil, &transportErrorV3{reason: "source_unavailable", message: "The effective media source is unavailable."}
|
|
}
|
|
effectiveFile := currentEffectiveFile
|
|
if intentChange && !trackChange {
|
|
// Prefer returning to the requested edition, but a quality/output/track
|
|
// change must not abandon a healthy active alternate merely because the
|
|
// inactive original has gone missing since playback started.
|
|
if requestedEditionResolved && preflightPlaybackFile(r.Context(), requestedFile, h.MissingMarker, h.EventsHub) == nil {
|
|
effectiveFile = requestedFile
|
|
}
|
|
// Track identities only need remapping when the effective edition
|
|
// actually changes. Remapping within the same file would degrade an
|
|
// exact selection to a best-match lookup — e.g. moving a listener
|
|
// from an eng/ac3 commentary track to the identically-shaped main
|
|
// track on a quality change.
|
|
if currentEffectiveFile.ID != effectiveFile.ID {
|
|
if err := remapAudioSelectionV3(currentEffectiveFile, effectiveFile, &start); err != nil {
|
|
return playback.DecisionResponseV3{}, *record, nil, &transportErrorV3{reason: "track_unavailable", message: err.Error()}
|
|
}
|
|
if start.SubtitleTrackIndex != nil || start.SubtitleTrackID != "" {
|
|
if err := h.remapSubtitleSelectionV3(r.Context(), currentEffectiveFile, effectiveFile, &start); err != nil {
|
|
return playback.DecisionResponseV3{}, *record, nil, &transportErrorV3{reason: "track_unavailable", message: err.Error()}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
start.FileID = effectiveFile.ID
|
|
if err := preflightPlaybackFile(r.Context(), effectiveFile, h.MissingMarker, h.EventsHub); err != nil {
|
|
return playback.DecisionResponseV3{}, *record, nil, &transportErrorV3{
|
|
reason: "source_unavailable",
|
|
message: "The effective media source is unavailable.",
|
|
cause: err,
|
|
}
|
|
}
|
|
seekDuration := float64(effectiveFile.Duration)
|
|
if seekReanchor && record.FrozenRecipe.ValidFor(record.CurrentPlan) {
|
|
seekDuration = record.FrozenRecipe.SourceDurationSeconds
|
|
}
|
|
if seekScopedRecovery && seekDuration > 0 && req.PositionSeconds > seekDuration {
|
|
return playback.DecisionResponseV3{}, *record, nil, &transportErrorV3{
|
|
reason: "invalid_seek_position",
|
|
message: "The requested seek position is beyond the end of the selected media source.",
|
|
}
|
|
}
|
|
if _, err := start.NormalizeAndValidate(); err != nil {
|
|
return playback.DecisionResponseV3{}, *record, nil, &transportErrorV3{reason: "invalid_replan", message: err.Error()}
|
|
}
|
|
audioIndex := 0
|
|
if !seekReanchor {
|
|
audioIndex, err = resolveV3AudioIndex(effectiveFile, start.AudioTrackID, start.AudioTrackIndex)
|
|
if err != nil {
|
|
return playback.DecisionResponseV3{}, *record, nil, &transportErrorV3{reason: "track_unavailable", message: err.Error()}
|
|
}
|
|
}
|
|
attemptedKeys := []string(nil)
|
|
if !intentChange && !seekReanchor && !userIntentOperation {
|
|
attemptedKeys = append(attemptedKeys, req.AttemptedPlanKeys...)
|
|
if !containsStringExactV3(attemptedKeys, req.PlanAttemptKey) {
|
|
attemptedKeys = append(attemptedKeys, req.PlanAttemptKey)
|
|
}
|
|
}
|
|
if !seekReanchor && !userIntentOperation && (!intentChange || seekFailureRecovery) {
|
|
// Always exclude the durable server recipe so stale or malformed client
|
|
// history cannot immediately re-select the route that just failed and
|
|
// ping-pong the session. A client-reported local mutation (for example a
|
|
// PCM recovery route) is folded into the failed plan's key here — the
|
|
// server owns the hash; clients only echo opaque keys.
|
|
currentKey := playback.PlanAttemptKeyV3(record.CurrentPlan, record.NormalizedRequest.ClientPlaybackContext.Output.OutputContextID, req.LocalMutations)
|
|
if !containsStringExactV3(attemptedKeys, currentKey) {
|
|
attemptedKeys = append(attemptedKeys, currentKey)
|
|
}
|
|
if len(req.LocalMutations) > 0 {
|
|
// The unmutated recipe already failed before the client mutated it
|
|
// locally; exclude it as well.
|
|
unmutatedKey := playback.PlanAttemptKeyV3(record.CurrentPlan, record.NormalizedRequest.ClientPlaybackContext.Output.OutputContextID, nil)
|
|
if !containsStringExactV3(attemptedKeys, unmutatedKey) {
|
|
attemptedKeys = append(attemptedKeys, unmutatedKey)
|
|
}
|
|
}
|
|
}
|
|
var result playback.PlannerResultV3
|
|
if seekReanchor {
|
|
if err := h.validateFrozenSubtitleIdentityV3(r.Context(), effectiveFile, record.FrozenRecipe); err != nil {
|
|
return playback.DecisionResponseV3{}, *record, nil, subtitleArtifactErrorV3("The selected subtitle is no longer available at its frozen route.", err)
|
|
}
|
|
var frozenErr error
|
|
result, frozenErr = frozenSeekReanchorResultV3(record, req.PositionSeconds, time.Now())
|
|
if frozenErr != nil {
|
|
return playback.DecisionResponseV3{}, *record, nil, &transportErrorV3{
|
|
reason: "seek_reanchor_recipe_unavailable",
|
|
message: "The active playback recipe cannot be reopened; start a new playback attempt.",
|
|
retryable: true,
|
|
}
|
|
}
|
|
} else {
|
|
result = playback.PlanPlaybackV3(playback.PlannerInputV3{Request: start, RequestedFile: plannerRequestedFile, EffectiveFile: effectiveFile, AudioTrackIndex: audioIndex, Settings: h.plannerSettingsV3(r.Context()), Registry: h.transformationRegistryV3(r.Context()), HLSRegistry: h.lazyHLSPlanningRegistryV3(r.Context()), DVRPUStrippable: h.lazyDVRPUStrippableV3(r.Context(), effectiveFile), Now: time.Now(), AttemptedKeys: attemptedKeys, AdditionalSubtitles: h.downloadedSubtitleInventoryV3(r.Context(), effectiveFile)})
|
|
}
|
|
if result.Terminal != nil && result.Terminal.Reason == "no_alternate_version" && replanAllowsAlternateFileV3(operation, start.QualityPreference) {
|
|
if alternate, alternateErr := h.findAlternateFile(r.Context(), requestedFile); alternateErr == nil && alternate != nil {
|
|
alternate = h.ensurePlaybackProbe(r.Context(), alternate)
|
|
remappedAudio := remapAudioIndexV3(effectiveFile, alternate, audioIndex)
|
|
if err := h.remapSubtitleSelectionV3(r.Context(), effectiveFile, alternate, &start); err == nil {
|
|
start.FileID = alternate.ID
|
|
if err := preflightPlaybackFile(r.Context(), alternate, h.MissingMarker, h.EventsHub); err == nil {
|
|
effectiveFile = alternate
|
|
audioIndex = remappedAudio
|
|
result = playback.PlanPlaybackV3(playback.PlannerInputV3{Request: start, RequestedFile: plannerRequestedFile, EffectiveFile: effectiveFile, AudioTrackIndex: audioIndex, Settings: h.plannerSettingsV3(r.Context()), Registry: h.transformationRegistryV3(r.Context()), HLSRegistry: h.lazyHLSPlanningRegistryV3(r.Context()), DVRPUStrippable: h.lazyDVRPUStrippableV3(r.Context(), effectiveFile), Now: time.Now(), AttemptedKeys: attemptedKeys, AdditionalSubtitles: h.downloadedSubtitleInventoryV3(r.Context(), effectiveFile)})
|
|
}
|
|
} else if start.SubtitleTrackIndex != nil || start.SubtitleTrackID != "" {
|
|
result = playback.PlannerResultV3{Terminal: &playback.TerminalV3{
|
|
Reason: "subtitle_unavailable_in_version",
|
|
Message: "The selected subtitle track is unavailable in the fallback media version.",
|
|
Retryable: false,
|
|
}}
|
|
}
|
|
}
|
|
}
|
|
h.clarifyOriginalQuality4KTerminalV3(r.Context(), result.Terminal, requestedFile, replanAlternateFilePinnedByOriginalQualityV3(operation, start.QualityPreference))
|
|
if result.Terminal != nil {
|
|
return playback.NewTerminalResponseV3(result.Terminal.Reason, result.Terminal.Message, result.Terminal.Retryable), *record, nil, nil
|
|
}
|
|
session, err := h.sessionMgr.GetSession(record.SessionID)
|
|
if err != nil {
|
|
return playback.DecisionResponseV3{}, *record, nil, &transportErrorV3{reason: "session_expired", message: "The playback session has expired.", retryable: true}
|
|
}
|
|
replacementManager, ok := h.sessionMgr.(replacementStateManagerV3)
|
|
if !ok {
|
|
return playback.DecisionResponseV3{}, *record, nil, &transportErrorV3{reason: "internal_error", message: "The live session manager does not support atomic replacement."}
|
|
}
|
|
if checker, ok := h.sessionMgr.(replacementAdmissionCheckerV3); ok {
|
|
if err := checker.CheckReplacementAllowed(r.Context(), session.ID, result.PlayMethod, result.TranscodeAudio); err != nil {
|
|
mapped := sessionStartErrorV3(err)
|
|
return playback.DecisionResponseV3{}, *record, nil, mapped
|
|
}
|
|
_, reservationHeld = h.sessionMgr.(replacementReservationCancellerV3)
|
|
}
|
|
if checker, ok := h.sessionMgr.(transcodePermissionChecker); ok && (result.PlayMethod == playback.PlayTranscode || result.TranscodeAudio) {
|
|
if err := checker.CheckTranscodingAllowed(r.Context(), session.UserID, result.PlayMethod == playback.PlayTranscode); err != nil {
|
|
mapped := sessionStartErrorV3(err)
|
|
return playback.DecisionResponseV3{}, *record, nil, mapped
|
|
}
|
|
}
|
|
result.Plan.SessionID = session.ID
|
|
artifactRecipe := record.FrozenRecipe
|
|
if !seekReanchor {
|
|
frozenRecipe, frozenErr := h.freezeExecutableRecipeV3(r.Context(), effectiveFile, result)
|
|
if frozenErr != nil {
|
|
return playback.DecisionResponseV3{}, *record, nil, subtitleArtifactErrorV3("Failed to freeze the selected subtitle identity.", frozenErr)
|
|
}
|
|
artifactRecipe = frozenRecipe
|
|
}
|
|
transportReused := trackChange && h.hasActiveHLSTransportV3(session) && sidecarOnlyHLSReplanV3(record, result.Plan, artifactRecipe, req.ClientPlaybackContext.Output.OutputContextID)
|
|
var transport preparedTransportV3
|
|
if transportReused {
|
|
// A sidecar selection changes the plan and subtitle artifact, but it does
|
|
// not change the bytes FFmpeg produces. Keep the active HLS generation and
|
|
// its transport window so a client remount cannot strand itself between
|
|
// the killed old window and a replacement window that starts elsewhere.
|
|
// The requested source position still belongs to this replan: translate it
|
|
// onto the reused window instead of rewinding to the previous plan's start.
|
|
result.Plan.Stream = record.CurrentPlan.Stream
|
|
reusedTimeline := record.CurrentPlan.Timeline
|
|
reusedTimeline.SourceStartSeconds = result.Plan.Timeline.SourceStartSeconds
|
|
reusedTimeline.PlayerStartSeconds = max(0, reusedTimeline.SourceStartSeconds-reusedTimeline.StreamOriginSeconds)
|
|
result.Plan.Timeline = reusedTimeline
|
|
result.Plan.ExpiresAt = record.CurrentPlan.ExpiresAt
|
|
transport = reusedHLSTransportV3(session, record.CurrentPlan.Stream.URL)
|
|
slog.InfoContext(r.Context(), "protocol v3 replan reused active HLS A/V transport",
|
|
"component", "playback",
|
|
"playback_session_id", session.ID,
|
|
"previous_plan_id", record.CurrentPlanID,
|
|
"plan_id", result.Plan.PlanID,
|
|
)
|
|
} else {
|
|
var transportErr *transportErrorV3
|
|
transport, transportErr = h.prepareTransportV3(r, session, effectiveFile, result)
|
|
if transportErr != nil {
|
|
return playback.DecisionResponseV3{}, *record, nil, transportErr
|
|
}
|
|
}
|
|
result.Plan.Stream.URL = transport.url
|
|
if err := h.attachSubtitleArtifactV3(r.Context(), session.ID, effectiveFile, result.Plan, result.SubtitleTrackIndex, &artifactRecipe); err != nil {
|
|
transport.rollback()
|
|
return playback.DecisionResponseV3{}, *record, nil, subtitleArtifactErrorV3("Failed to prepare the selected subtitle artifact.", err)
|
|
}
|
|
if seekReanchor {
|
|
if err := validateSeekReanchorPlanV3(record, result.Plan); err != nil {
|
|
changedFields := seekReanchorIdentityChangesV3(record, result.Plan)
|
|
slog.ErrorContext(r.Context(), "protocol v3 seek reanchor changed route identity",
|
|
"session", record.SessionID,
|
|
"playback_attempt_id", record.PlaybackAttemptID,
|
|
"changed_fields", changedFields,
|
|
)
|
|
transport.rollback()
|
|
return playback.DecisionResponseV3{}, *record, nil, &transportErrorV3{
|
|
reason: "seek_reanchor_route_changed",
|
|
message: err.Error(),
|
|
}
|
|
}
|
|
}
|
|
response := playback.DecisionResponseV3{ProtocolVersion: playback.ProtocolV3, ServerFeatures: playback.ServerFeaturesV3(), Outcome: playback.OutcomePlayableV3, SessionID: session.ID, PlaybackPlan: result.Plan}
|
|
updated := *record
|
|
updated.CurrentPlanID = result.Plan.PlanID
|
|
updated.CurrentPlan = *result.Plan
|
|
// A seek reanchor replays the durable recipe verbatim (updated already
|
|
// carries it); re-freezing from live inventory could only re-introduce
|
|
// the drift this path exists to exclude. Every other replan just accepted
|
|
// a freshly planned route and must freeze its recipe — loudly, because a
|
|
// recipe with a silently missing subtitle identity would disable drift
|
|
// detection for every later seek on this attempt.
|
|
if !seekReanchor {
|
|
updated.FrozenRecipe = artifactRecipe
|
|
}
|
|
updated.NormalizedRequest = start
|
|
updated.EffectiveMediaFileID = effectiveFile.ID
|
|
updated.ExpiresAt = time.Now().Add(playback.MaxTokenTTL)
|
|
if transportReused {
|
|
if expiresAt, parseErr := time.Parse(time.RFC3339, result.Plan.ExpiresAt); parseErr == nil {
|
|
updated.ExpiresAt = expiresAt
|
|
}
|
|
}
|
|
originalRollback := transport.rollback
|
|
replacement := playback.SessionReplacement{
|
|
EffectiveMediaFileID: effectiveFile.ID,
|
|
StreamState: h.v3SessionStreamState(r.Context(), session, effectiveFile, result, transport),
|
|
}
|
|
if seekScopedRecovery {
|
|
replacement.PositionSeconds = &req.PositionSeconds
|
|
replacement.PreservePaused = true
|
|
}
|
|
transport.applySession = func() (func() error, error) {
|
|
rollback, err := replacementManager.ApplyReplacement(session.ID, replacement)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return func() error {
|
|
return replacementManager.RollbackReplacement(session.ID, rollback)
|
|
}, nil
|
|
}
|
|
transport.afterDurableCommit = func() {
|
|
cancelReservation()
|
|
if trackChange {
|
|
// A deliberate track switch is the same signal the legacy audio
|
|
// PATCH recorded; a failure recovery is not, so its forced audio
|
|
// route must not be written back as a user preference.
|
|
h.persistAudioPreference(r.Context(), session.UserID, session.ProfileID, effectiveFile, plannedAudioTrackIndexV3(result, session.AudioTrackIndex))
|
|
}
|
|
h.syncSessionsNow(r.Context(), "v3_replan")
|
|
event := playback.RouteEventPlanSelectedV3
|
|
clientModel := req.ClientPlaybackContext.Device.Model
|
|
if seekReanchor {
|
|
event = playback.RouteEventRuntimeCorrectionSucceededV3
|
|
clientModel = start.ClientPlaybackContext.Device.Model
|
|
}
|
|
h.enqueueRouteEventV3(playback.RouteEventRecordV3{RouteEventV3: playback.RouteEventV3{ProtocolVersion: playback.ProtocolV3, PlaybackAttemptID: req.PlaybackAttemptID, SessionID: session.ID, PlanID: result.Plan.PlanID, PlanAttemptID: req.PlanAttemptID, PlanAttemptKey: playback.PlanAttemptKeyV3(*result.Plan, start.ClientPlaybackContext.Output.OutputContextID, nil), Event: event, FallbackReason: req.Failure.Classification, AppliedQuirkIDs: appliedQuirkIDsV3(result.Plan), QuirkRegistryRevision: appliedQuirkRevisionV3(result.Plan), OutputContextID: start.ClientPlaybackContext.Output.OutputContextID}, UserID: session.UserID, ProfileID: session.ProfileID, ClientName: session.ClientName, ClientVersion: session.ClientVersion, ClientModel: clientModel})
|
|
}
|
|
transport.rollback = func() {
|
|
originalRollback()
|
|
cancelReservation()
|
|
}
|
|
reservationHandedOff = true
|
|
return response, updated, &transport, nil
|
|
}
|
|
|
|
func frozenSeekReanchorResultV3(record *playback.AttemptRecordV3, position float64, now time.Time) (playback.PlannerResultV3, error) {
|
|
if record == nil || !record.FrozenRecipe.ValidFor(record.CurrentPlan) {
|
|
return playback.PlannerResultV3{}, errors.New("the active playback recipe is unavailable")
|
|
}
|
|
plan := record.CurrentPlan
|
|
plan.ExpiresAt = playback.NewPlanExpiryV3(now)
|
|
plan.Timeline = playback.TimelineV3{
|
|
SourceStartSeconds: position,
|
|
PlayerStartSeconds: position,
|
|
CanSeekAnywhere: true,
|
|
SeekRestoration: seekRestorationPlayerV3,
|
|
}
|
|
return record.FrozenRecipe.PlannerResult(&plan), nil
|
|
}
|
|
|
|
type subtitleIndexLocationV3 struct {
|
|
source string
|
|
offset int
|
|
}
|
|
|
|
// classifySubtitleIndexV3 maps the combined subtitle index used by
|
|
// buildSubtitleURLs to its inventory segment and segment-local offset.
|
|
func classifySubtitleIndexV3(file *models.MediaFile, index int) (subtitleIndexLocationV3, bool) {
|
|
if file == nil || index < 0 {
|
|
return subtitleIndexLocationV3{}, false
|
|
}
|
|
externalCount := len(file.ExternalSubtitles)
|
|
if index < externalCount {
|
|
return subtitleIndexLocationV3{source: playback.SubtitleSourceExternalV3, offset: index}, true
|
|
}
|
|
embeddedOffset := index - externalCount
|
|
if embeddedOffset < len(file.SubtitleTracks) {
|
|
return subtitleIndexLocationV3{source: playback.SubtitleSourceEmbeddedV3, offset: embeddedOffset}, true
|
|
}
|
|
return subtitleIndexLocationV3{
|
|
source: playback.SubtitleSourceDownloadedV3,
|
|
offset: embeddedOffset - len(file.SubtitleTracks),
|
|
}, true
|
|
}
|
|
|
|
// freezeExecutableRecipeV3 extends the pure planner freeze with the identity
|
|
// of the selected sidecar subtitle. The combined subtitle index space
|
|
// (externals, then embedded, then downloaded — see buildSubtitleURLs) is not
|
|
// stable across inventory changes, so the index alone cannot anchor a durable
|
|
// selection. A downloaded selection whose identity cannot be established is
|
|
// an error: silently omitting it would disable drift detection for exactly
|
|
// the seeks this recipe exists to protect.
|
|
func (h *PlaybackHandler) freezeExecutableRecipeV3(_ context.Context, file *models.MediaFile, result playback.PlannerResultV3) (playback.ExecutableRecipeV3, error) {
|
|
recipe := playback.FreezeExecutableRecipeV3(result)
|
|
if file != nil {
|
|
sourceMetadata := sourceExecutionMetadataV3(file, playback.PlannerResultV3{})
|
|
recipe.SourceVideoCodec = sourceMetadata.VideoCodec
|
|
recipe.SoftwareVideoDecode = sourceMetadata.SoftwareVideoDecode
|
|
recipe.SourceDurationSeconds = sourceMetadata.DurationSeconds
|
|
}
|
|
if file == nil || result.SubtitleTrackIndex < 0 {
|
|
return recipe, nil
|
|
}
|
|
// A downloaded row ID was selected from the planner's inventory snapshot.
|
|
// Treat it as authoritative before consulting the mutable combined-index
|
|
// segments: an external or embedded subtitle added after planning must not
|
|
// make this downloaded selection look like a different source.
|
|
if recipe.DownloadedSubtitleID > 0 {
|
|
recipe.SubtitleSource = playback.SubtitleSourceDownloadedV3
|
|
return recipe, nil
|
|
}
|
|
location, ok := classifySubtitleIndexV3(file, result.SubtitleTrackIndex)
|
|
if !ok {
|
|
return recipe, nil
|
|
}
|
|
switch location.source {
|
|
case playback.SubtitleSourceExternalV3:
|
|
recipe.SubtitleSource = playback.SubtitleSourceExternalV3
|
|
recipe.ExternalSubtitlePath = file.ExternalSubtitles[location.offset].Path
|
|
case playback.SubtitleSourceEmbeddedV3:
|
|
recipe.SubtitleSource = playback.SubtitleSourceEmbeddedV3
|
|
recipe.EmbeddedStreamIndex = file.SubtitleTracks[location.offset].Index
|
|
case playback.SubtitleSourceDownloadedV3:
|
|
if recipe.DownloadedSubtitleID <= 0 {
|
|
return playback.ExecutableRecipeV3{}, errors.New("the selected downloaded subtitle has no stable identity")
|
|
}
|
|
}
|
|
return recipe, nil
|
|
}
|
|
|
|
// validateFrozenSubtitleIdentityV3 confirms the frozen combined subtitle
|
|
// index still resolves to the identical inventory entry it was frozen
|
|
// against. It mirrors the segment layout of buildSubtitleURLs so a change in
|
|
// any earlier segment's size — which shifts every later index — is detected
|
|
// as an identity mismatch rather than silently re-resolved.
|
|
func (h *PlaybackHandler) validateFrozenSubtitleIdentityV3(ctx context.Context, file *models.MediaFile, recipe playback.ExecutableRecipeV3) error {
|
|
if recipe.SubtitleSource == "" {
|
|
return nil
|
|
}
|
|
if file == nil || recipe.SubtitleTrackIndex < 0 {
|
|
return errors.New("the frozen subtitle selection is unavailable")
|
|
}
|
|
if recipe.SubtitleSource == playback.SubtitleSourceDownloadedV3 {
|
|
if h == nil || h.SubtitleRepo == nil || recipe.DownloadedSubtitleID <= 0 {
|
|
return errors.New("the downloaded subtitle inventory is unavailable")
|
|
}
|
|
downloaded, err := h.SubtitleRepo.GetDownloadedSubtitle(ctx, recipe.DownloadedSubtitleID)
|
|
if err != nil {
|
|
return wrapSubtitleStoreErrorV3(err)
|
|
}
|
|
if downloaded == nil || downloaded.MediaFileID != file.ID {
|
|
return errors.New("the frozen downloaded subtitle identity changed")
|
|
}
|
|
return nil
|
|
}
|
|
location, ok := classifySubtitleIndexV3(file, recipe.SubtitleTrackIndex)
|
|
if !ok || location.source != recipe.SubtitleSource {
|
|
return errors.New("the frozen subtitle inventory segment changed")
|
|
}
|
|
switch recipe.SubtitleSource {
|
|
case playback.SubtitleSourceExternalV3:
|
|
if file.ExternalSubtitles[location.offset].Path != recipe.ExternalSubtitlePath {
|
|
return errors.New("the frozen external subtitle identity changed")
|
|
}
|
|
case playback.SubtitleSourceEmbeddedV3:
|
|
if file.SubtitleTracks[location.offset].Index != recipe.EmbeddedStreamIndex {
|
|
return errors.New("the frozen embedded subtitle identity changed")
|
|
}
|
|
default:
|
|
return errors.New("the frozen subtitle identity is unrecognized")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func validateSeekRecoveryRequestV3(record *playback.AttemptRecordV3, req playback.ReplanRequestV3) error {
|
|
if record == nil {
|
|
return errors.New("the current playback attempt is unavailable")
|
|
}
|
|
wantedQuality, _ := playback.NormalizeQualityV3(record.NormalizedRequest.QualityPreference)
|
|
requestedQuality, _ := playback.NormalizeQualityV3(req.QualityPreference)
|
|
if requestedQuality != wantedQuality {
|
|
return errors.New("seek recovery cannot change playback quality")
|
|
}
|
|
if req.ClientPlaybackContext.Output.OutputContextID != record.NormalizedRequest.ClientPlaybackContext.Output.OutputContextID {
|
|
return errors.New("seek recovery cannot change the output route")
|
|
}
|
|
if !sameSelectedTracksV3(req.SelectedTracks, record.CurrentPlan.SelectedTracks) {
|
|
return errors.New("seek recovery cannot change selected tracks")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// seekReanchorIdentityChangesV3 returns only bounded, non-secret field names.
|
|
// It is safe for structured logs: values, URLs, headers, tokens, and subtitle
|
|
// artifact locations are deliberately excluded.
|
|
func seekReanchorIdentityChangesV3(record *playback.AttemptRecordV3, candidate *playback.PlanV3) []string {
|
|
if record == nil || candidate == nil {
|
|
return []string{"route"}
|
|
}
|
|
current := record.CurrentPlan
|
|
changed := make([]string, 0, 16)
|
|
add := func(name string, differs bool) {
|
|
if differs {
|
|
changed = append(changed, name)
|
|
}
|
|
}
|
|
add("plan_id", candidate.PlanID != record.CurrentPlanID || candidate.PlanID != current.PlanID)
|
|
add("requested_file_id", candidate.RequestedMediaFileID != record.RequestedMediaFileID)
|
|
add("effective_file_id", candidate.EffectiveMediaFileID != record.EffectiveMediaFileID)
|
|
add("delivery", candidate.Delivery != current.Delivery)
|
|
add("protocol", candidate.Stream.Protocol != current.Stream.Protocol)
|
|
add("container", candidate.Stream.Container != current.Stream.Container)
|
|
add("mime_type", candidate.Stream.MIMEType != current.Stream.MIMEType)
|
|
add("header_refresh", candidate.Stream.HeaderRefresh != current.Stream.HeaderRefresh)
|
|
add("video_codec", candidate.EffectiveRecipe.VideoCodec != current.EffectiveRecipe.VideoCodec)
|
|
add("audio_codec", candidate.EffectiveRecipe.AudioCodec != current.EffectiveRecipe.AudioCodec)
|
|
add("resolution", !optionalIntEqualV3(candidate.EffectiveRecipe.Width, current.EffectiveRecipe.Width) || !optionalIntEqualV3(candidate.EffectiveRecipe.Height, current.EffectiveRecipe.Height))
|
|
add("frame_rate", !optionalFloatEqualV3(candidate.EffectiveRecipe.FrameRate, current.EffectiveRecipe.FrameRate))
|
|
add("bitrate", !optionalIntEqualV3(candidate.EffectiveRecipe.BitrateKbps, current.EffectiveRecipe.BitrateKbps))
|
|
add("dynamic_range", candidate.EffectiveRecipe.DynamicRange != current.EffectiveRecipe.DynamicRange)
|
|
add("audio_channels", !optionalIntEqualV3(candidate.EffectiveRecipe.AudioChannels, current.EffectiveRecipe.AudioChannels) || candidate.EffectiveRecipe.AudioLayout != current.EffectiveRecipe.AudioLayout)
|
|
add("selected_audio", !sameTrackIdentityV3(candidate.SelectedTracks.Audio, current.SelectedTracks.Audio))
|
|
add("selected_subtitle", !sameTrackIdentityV3(candidate.SelectedTracks.Subtitle, current.SelectedTracks.Subtitle))
|
|
add("subtitle_mode", candidate.Subtitle.Mode != current.Subtitle.Mode || candidate.Subtitle.TrackID != current.Subtitle.TrackID)
|
|
add("subtitle_artifact_route", !sameSubtitleArtifactRouteV3(candidate.Subtitle.Artifact, current.Subtitle.Artifact))
|
|
add("subtitle_fidelity", candidate.SubtitleFidelityPolicy != current.SubtitleFidelityPolicy)
|
|
add("transformations", !sameTransformationsV3(candidate.Transformations, current.Transformations))
|
|
add("quirks", !sameAppliedQuirksV3(candidate.AppliedQuirks, current.AppliedQuirks))
|
|
add("runtime_corrections", !sameStringMultisetV3(candidate.RuntimeCorrections, current.RuntimeCorrections))
|
|
add("claims", candidate.Claims != current.Claims)
|
|
return changed
|
|
}
|
|
|
|
func validateSeekReanchorPlanV3(record *playback.AttemptRecordV3, candidate *playback.PlanV3) error {
|
|
if record == nil || candidate == nil {
|
|
return errors.New("seek reanchor produced no playback route")
|
|
}
|
|
changedFields := seekReanchorIdentityChangesV3(record, candidate)
|
|
if len(changedFields) == 0 {
|
|
return nil
|
|
}
|
|
if containsStringExactV3(changedFields, "plan_id") {
|
|
return errors.New("seek reanchor changed the playback plan identity")
|
|
}
|
|
if containsStringExactV3(changedFields, "requested_file_id") || containsStringExactV3(changedFields, "effective_file_id") {
|
|
return errors.New("seek reanchor changed the selected media version")
|
|
}
|
|
if containsStringExactV3(changedFields, "selected_audio") || containsStringExactV3(changedFields, "selected_subtitle") {
|
|
return errors.New("seek reanchor changed selected tracks")
|
|
}
|
|
return errors.New("seek reanchor changed the playback route semantics")
|
|
}
|
|
|
|
func sameSubtitleArtifactRouteV3(left, right *playback.SubtitleArtifactV3) bool {
|
|
if left == nil || right == nil {
|
|
return left == nil && right == nil
|
|
}
|
|
// Signed URLs and timing origins are allowed to rotate when a transport is
|
|
// reopened; the player-facing artifact representation is not.
|
|
return left.MIMEType == right.MIMEType && left.Format == right.Format
|
|
}
|
|
|
|
func sameTransformationsV3(left, right []playback.TransformationV3) bool {
|
|
if len(left) != len(right) {
|
|
return false
|
|
}
|
|
matched := make([]bool, len(right))
|
|
for _, candidate := range left {
|
|
found := false
|
|
for index, current := range right {
|
|
if !matched[index] && candidate.Name == current.Name && candidate.Executor == current.Executor &&
|
|
candidate.RecipeVersion == current.RecipeVersion && sameStringMultisetV3(candidate.ValidatedClaims, current.ValidatedClaims) {
|
|
matched[index] = true
|
|
found = true
|
|
break
|
|
}
|
|
}
|
|
if !found {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
func sameAppliedQuirksV3(left, right []playback.AppliedQuirkV3) bool {
|
|
if len(left) != len(right) {
|
|
return false
|
|
}
|
|
matched := make([]bool, len(right))
|
|
for _, candidate := range left {
|
|
found := false
|
|
for index, current := range right {
|
|
if !matched[index] && candidate == current {
|
|
matched[index] = true
|
|
found = true
|
|
break
|
|
}
|
|
}
|
|
if !found {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
func sameStringMultisetV3(left, right []string) bool {
|
|
if len(left) != len(right) {
|
|
return false
|
|
}
|
|
counts := make(map[string]int, len(left))
|
|
for _, value := range left {
|
|
counts[value]++
|
|
}
|
|
for _, value := range right {
|
|
counts[value]--
|
|
if counts[value] < 0 {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
// sidecarOnlyHLSReplanV3 proves that a track-change plan can keep the active
|
|
// HLS A/V generation. Subtitle identity and claims are intentionally excluded:
|
|
// those are the point of the replan and are delivered by the independently
|
|
// addressed sidecar artifact. Every field that can change FFmpeg's audio/video
|
|
// output remains part of the comparison.
|
|
func sidecarOnlyHLSReplanV3(record *playback.AttemptRecordV3, candidate *playback.PlanV3, candidateRecipe playback.ExecutableRecipeV3, outputContextID string) bool {
|
|
if record == nil || candidate == nil || record.CurrentPlan.Stream.URL == "" ||
|
|
!record.FrozenRecipe.ValidFor(record.CurrentPlan) || !candidateRecipe.ValidFor(*candidate) ||
|
|
record.NormalizedRequest.ClientPlaybackContext.Output.OutputContextID != outputContextID ||
|
|
record.EffectiveMediaFileID != candidate.EffectiveMediaFileID ||
|
|
record.CurrentPlan.RequestedMediaFileID != candidate.RequestedMediaFileID ||
|
|
record.CurrentPlan.EffectiveMediaFileID != candidate.EffectiveMediaFileID ||
|
|
!isHLSDeliveryV3(record.CurrentPlan.Delivery) || record.CurrentPlan.Delivery != candidate.Delivery ||
|
|
record.CurrentPlan.Subtitle.Mode == playback.SubtitleBurnInV3 || candidate.Subtitle.Mode == playback.SubtitleBurnInV3 ||
|
|
!sameTrackIdentityV3(record.CurrentPlan.SelectedTracks.Audio, candidate.SelectedTracks.Audio) {
|
|
return false
|
|
}
|
|
if record.CurrentPlan.Stream.Protocol != candidate.Stream.Protocol ||
|
|
record.CurrentPlan.Stream.Container != candidate.Stream.Container ||
|
|
record.CurrentPlan.Stream.MIMEType != candidate.Stream.MIMEType ||
|
|
record.CurrentPlan.Stream.HeaderRefresh != candidate.Stream.HeaderRefresh ||
|
|
!sameExecutableAVRecipeV3(record.FrozenRecipe, candidateRecipe) ||
|
|
!sameEffectiveAVRecipeV3(record.CurrentPlan.EffectiveRecipe, candidate.EffectiveRecipe) ||
|
|
record.CurrentPlan.Claims.Video != candidate.Claims.Video ||
|
|
record.CurrentPlan.Claims.Audio != candidate.Claims.Audio ||
|
|
!sameTransformationsV3(record.CurrentPlan.Transformations, candidate.Transformations) ||
|
|
!sameAppliedQuirksV3(record.CurrentPlan.AppliedQuirks, candidate.AppliedQuirks) ||
|
|
!sameStringMultisetV3(record.CurrentPlan.RuntimeCorrections, candidate.RuntimeCorrections) {
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
func isHLSDeliveryV3(delivery playback.DeliveryV3) bool {
|
|
return delivery == playback.DeliveryRemuxHLSV3 || delivery == playback.DeliveryTranscodeHLSV3
|
|
}
|
|
|
|
func sameExecutableAVRecipeV3(left, right playback.ExecutableRecipeV3) bool {
|
|
return left.PlayMethod == right.PlayMethod &&
|
|
left.TranscodeAudio == right.TranscodeAudio &&
|
|
left.TargetVideoCodec == right.TargetVideoCodec &&
|
|
left.TargetAudioCodec == right.TargetAudioCodec &&
|
|
left.TargetAudioChannels == right.TargetAudioChannels &&
|
|
left.TargetAudioBitrateKbps == right.TargetAudioBitrateKbps &&
|
|
left.TargetResolution == right.TargetResolution &&
|
|
left.TargetBitrateKbps == right.TargetBitrateKbps &&
|
|
left.SourceVideoCodec == right.SourceVideoCodec &&
|
|
left.SoftwareVideoDecode == right.SoftwareVideoDecode &&
|
|
left.SourceDurationSeconds == right.SourceDurationSeconds
|
|
}
|
|
|
|
func sameEffectiveAVRecipeV3(left, right playback.EffectiveRecipeV3) bool {
|
|
return left.VideoCodec == right.VideoCodec && left.AudioCodec == right.AudioCodec &&
|
|
optionalIntEqualV3(left.Width, right.Width) && optionalIntEqualV3(left.Height, right.Height) &&
|
|
optionalFloatEqualV3(left.FrameRate, right.FrameRate) &&
|
|
optionalIntEqualV3(left.BitrateKbps, right.BitrateKbps) &&
|
|
left.DynamicRange == right.DynamicRange &&
|
|
optionalIntEqualV3(left.AudioChannels, right.AudioChannels) && left.AudioLayout == right.AudioLayout
|
|
}
|
|
|
|
func reusedHLSTransportV3(session *playback.Session, streamURL string) preparedTransportV3 {
|
|
transport := preparedTransportV3{url: streamURL}
|
|
if session != nil {
|
|
transport.nodeURL = session.TranscodeNodeURL
|
|
transport.transportID = session.TranscodeTransportID
|
|
}
|
|
transport.commit = func() {}
|
|
transport.rollback = func() {}
|
|
return transport
|
|
}
|
|
|
|
func (h *PlaybackHandler) hasActiveHLSTransportV3(session *playback.Session) bool {
|
|
if h == nil || session == nil {
|
|
return false
|
|
}
|
|
if session.TranscodeNodeURL != "" {
|
|
return true
|
|
}
|
|
return h.tm.GetTranscodeSession(session.ID) != nil
|
|
}
|
|
|
|
func applySelectedTracksToStartV3(start *playback.StartRequestV3, selected playback.SelectedTracksV3) {
|
|
if start == nil {
|
|
return
|
|
}
|
|
if selected.Audio != nil {
|
|
start.AudioTrackID = selected.Audio.ID
|
|
start.AudioTrackIndex = copyOptionalIntV3(selected.Audio.Index)
|
|
}
|
|
if selected.Subtitle != nil {
|
|
start.SubtitleTrackID = selected.Subtitle.ID
|
|
start.SubtitleTrackIndex = copyOptionalIntV3(selected.Subtitle.Index)
|
|
} else {
|
|
start.SubtitleTrackID = ""
|
|
start.SubtitleTrackIndex = nil
|
|
}
|
|
}
|
|
|
|
// applySelectedTrackOverridesToStartV3 overlays only identities the caller
|
|
// actually sent. Replan bodies are intentionally sparse for every operation
|
|
// except track_change, so an omitted subtitle here means "unchanged", not
|
|
// "off". The exact-replacement helper above remains the authority for an
|
|
// explicit track_change and for reconstructing a durable plan selection.
|
|
func applySelectedTrackOverridesToStartV3(start *playback.StartRequestV3, selected playback.SelectedTracksV3) {
|
|
if start == nil {
|
|
return
|
|
}
|
|
if selected.Audio != nil {
|
|
start.AudioTrackID = selected.Audio.ID
|
|
start.AudioTrackIndex = copyOptionalIntV3(selected.Audio.Index)
|
|
}
|
|
if selected.Subtitle != nil {
|
|
start.SubtitleTrackID = selected.Subtitle.ID
|
|
start.SubtitleTrackIndex = copyOptionalIntV3(selected.Subtitle.Index)
|
|
}
|
|
}
|
|
|
|
// audioSelectionDiffersFromStartV3 reports whether the replan's audio
|
|
// selection names a track other than the start request's. An omitted audio
|
|
// identity means "unchanged" — clients may not resend the current track.
|
|
func audioSelectionDiffersFromStartV3(selected playback.SelectedTracksV3, start playback.StartRequestV3) bool {
|
|
return selected.Audio != nil &&
|
|
(selected.Audio.ID != start.AudioTrackID || !optionalIntEqualV3(selected.Audio.Index, start.AudioTrackIndex))
|
|
}
|
|
|
|
// subtitleSelectionDiffersFromStartV3 reports whether the replan's subtitle
|
|
// selection differs from the start request's. Unlike audio, a nil subtitle is
|
|
// an explicit "subtitles off" and counts as a change when one was selected.
|
|
func subtitleSelectionDiffersFromStartV3(selected playback.SelectedTracksV3, start playback.StartRequestV3) bool {
|
|
if selected.Subtitle == nil {
|
|
return start.SubtitleTrackIndex != nil
|
|
}
|
|
return selected.Subtitle.ID != start.SubtitleTrackID || !optionalIntEqualV3(selected.Subtitle.Index, start.SubtitleTrackIndex)
|
|
}
|
|
|
|
func sameSelectedTracksV3(left, right playback.SelectedTracksV3) bool {
|
|
return sameTrackIdentityV3(left.Audio, right.Audio) && sameTrackIdentityV3(left.Subtitle, right.Subtitle)
|
|
}
|
|
|
|
func sameTrackIdentityV3(left, right *playback.TrackIdentityV3) bool {
|
|
if left == nil || right == nil {
|
|
return left == nil && right == nil
|
|
}
|
|
return left.ID == right.ID && optionalIntEqualV3(left.Index, right.Index)
|
|
}
|
|
|
|
func copyOptionalIntV3(value *int) *int {
|
|
if value == nil {
|
|
return nil
|
|
}
|
|
copy := *value
|
|
return ©
|
|
}
|
|
|
|
func shouldTryAlternateFileV3(qualityPreference string) bool {
|
|
return !strings.EqualFold(strings.TrimSpace(qualityPreference), "original")
|
|
}
|
|
|
|
func replanAllowsAlternateFileV3(operation playback.ReplanOperationV3, qualityPreference string) bool {
|
|
switch operation {
|
|
case playback.ReplanOperationFailureRecoveryV3, playback.ReplanOperationQualityChangeV3:
|
|
// A quality change carries the same "another version may fit better"
|
|
// semantics its legacy quality_changed failure classification had; seek
|
|
// operations and track changes stay pinned to the mounted source.
|
|
return shouldTryAlternateFileV3(qualityPreference)
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func replanAlternateFilePinnedByOriginalQualityV3(operation playback.ReplanOperationV3, qualityPreference string) bool {
|
|
if shouldTryAlternateFileV3(qualityPreference) {
|
|
return false
|
|
}
|
|
return operation == playback.ReplanOperationFailureRecoveryV3 || operation == playback.ReplanOperationQualityChangeV3
|
|
}
|
|
|
|
func (h *PlaybackHandler) clarifyOriginalQuality4KTerminalV3(ctx context.Context, terminal *playback.TerminalV3, requestedFile *models.MediaFile, alternateFilePinned bool) {
|
|
if !alternateFilePinned || terminal == nil || terminal.Reason != "no_alternate_version" || terminal.Message != playback.TerminalMessage4KTranscodeDisabledV3 {
|
|
return
|
|
}
|
|
if alternate, err := h.findAlternateFile(ctx, requestedFile); err == nil && alternate != nil {
|
|
terminal.Message = "4K transcoding is disabled and quality 'original' pins the 4K version; a compatible lower-resolution version of this title is available."
|
|
}
|
|
}
|
|
|
|
func (h *PlaybackHandler) lockReplanV3(sessionID string) func() {
|
|
h.v3ReplanMu.Lock()
|
|
if h.v3ReplanLocks == nil {
|
|
h.v3ReplanLocks = make(map[string]*v3ReplanLock)
|
|
}
|
|
entry := h.v3ReplanLocks[sessionID]
|
|
if entry == nil {
|
|
entry = &v3ReplanLock{}
|
|
h.v3ReplanLocks[sessionID] = entry
|
|
}
|
|
entry.refs++
|
|
h.v3ReplanMu.Unlock()
|
|
entry.mu.Lock()
|
|
return func() {
|
|
entry.mu.Unlock()
|
|
h.v3ReplanMu.Lock()
|
|
entry.refs--
|
|
if entry.refs == 0 {
|
|
delete(h.v3ReplanLocks, sessionID)
|
|
}
|
|
h.v3ReplanMu.Unlock()
|
|
}
|
|
}
|
|
|
|
// maxConcurrentReplansV3 bounds simultaneous replan executions. Each replan
|
|
// pins one pooled DB connection for its advisory session lock while issuing
|
|
// further store queries from the same pool; without a bound, a recovery storm
|
|
// (a transcode node dying with dozens of active sessions) turns every pool
|
|
// connection into a lock holder and the inner queries deadlock against them.
|
|
const maxConcurrentReplansV3 = 8
|
|
|
|
// sessionLockCapacityAdvisorV3 lets a plan store cap replan concurrency below
|
|
// the fixed default when its own connection budget is smaller; a pool sized at
|
|
// or below the default would otherwise let lock holders starve the inner
|
|
// store queries that must complete before any lock is released.
|
|
type sessionLockCapacityAdvisorV3 interface {
|
|
SessionLockCapacity() int
|
|
}
|
|
|
|
// acquireReplanSlotV3 blocks until a replan slot frees or the request context
|
|
// is cancelled; excess replans queue here holding no DB resources at all.
|
|
func (h *PlaybackHandler) acquireReplanSlotV3(ctx context.Context) (func(), error) {
|
|
h.v3ReplanSlotsOnce.Do(func() {
|
|
capacity := maxConcurrentReplansV3
|
|
if advisor, ok := h.PlanStoreV3.(sessionLockCapacityAdvisorV3); ok {
|
|
if advised := advisor.SessionLockCapacity(); advised > 0 && advised < capacity {
|
|
capacity = advised
|
|
}
|
|
}
|
|
h.v3ReplanSlots = make(chan struct{}, capacity)
|
|
})
|
|
select {
|
|
case h.v3ReplanSlots <- struct{}{}:
|
|
var once sync.Once
|
|
return func() { once.Do(func() { <-h.v3ReplanSlots }) }, nil
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
}
|
|
}
|
|
|
|
func (h *PlaybackHandler) HandlePlaybackRouteEventV3(w http.ResponseWriter, r *http.Request) {
|
|
userID := apimw.GetUserID(r.Context())
|
|
profileID := apimw.GetProfileID(r.Context())
|
|
if userID == 0 || profileID == "" {
|
|
writeError(w, http.StatusUnauthorized, "unauthorized", "Authentication and profile are required")
|
|
return
|
|
}
|
|
body, err := readBoundedV3Body(w, r, maxPlaybackV3EventBodyBytes)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "Invalid event body")
|
|
return
|
|
}
|
|
var event playback.RouteEventV3
|
|
if err := json.Unmarshal(body, &event); err != nil || !validRouteEventV3(event) {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "Invalid route event")
|
|
return
|
|
}
|
|
// The rate limiter runs before the ownership lookup so the per-minute
|
|
// budget bounds the store reads as well as the writes.
|
|
if !h.allowRouteEventV3(userID, event.PlaybackAttemptID) {
|
|
writeError(w, http.StatusTooManyRequests, "event_rate_limited", "Playback route event rate exceeded")
|
|
return
|
|
}
|
|
var identity *playback.AttemptIdentityV3
|
|
var identityErr error
|
|
if event.SessionID != "" {
|
|
identity, identityErr = h.PlanStoreV3.GetAttemptIdentity(r.Context(), event.SessionID)
|
|
} else {
|
|
identity, identityErr = h.PlanStoreV3.GetAttemptIdentityByPlaybackAttemptID(r.Context(), event.PlaybackAttemptID)
|
|
}
|
|
if identityErr != nil {
|
|
// A store outage is not an ownership violation; keep 403 for genuine
|
|
// mismatches so clients stop sending events for foreign sessions.
|
|
if !errors.Is(identityErr, playback.ErrSessionNotFound) {
|
|
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to authorize the route event")
|
|
return
|
|
}
|
|
writeError(w, http.StatusForbidden, "forbidden", "Route event does not belong to this profile")
|
|
return
|
|
}
|
|
if identity.UserID != userID || identity.ProfileID != profileID ||
|
|
(event.SessionID != "" && identity.PlaybackAttemptID != event.PlaybackAttemptID) ||
|
|
(identity.SessionID == "" && !terminalStartRouteEventV3(event)) {
|
|
writeError(w, http.StatusForbidden, "forbidden", "Route event does not belong to this profile")
|
|
return
|
|
}
|
|
event.Diagnostics = sanitizeDiagnosticsV3(event.Diagnostics)
|
|
client := playbackClientInfoFromRequest(r)
|
|
h.enqueueRouteEventV3(playback.RouteEventRecordV3{RouteEventV3: event, UserID: userID, ProfileID: profileID, ClientName: client.Name, ClientVersion: client.Version, ClientModel: event.Diagnostics["device_model"]})
|
|
w.WriteHeader(http.StatusAccepted)
|
|
}
|
|
|
|
func terminalStartRouteEventV3(event playback.RouteEventV3) bool {
|
|
return event.Event == playback.RouteEventTerminalV3 &&
|
|
event.SessionID == "" && event.PlanID == "" &&
|
|
event.PlanAttemptID == "" && event.PlanAttemptKey == ""
|
|
}
|
|
|
|
// StartV3Maintenance expires cached signed responses and old telemetry on the
|
|
// application lifecycle rather than on latency-sensitive playback requests.
|
|
func (h *PlaybackHandler) StartV3Maintenance(ctx context.Context) {
|
|
if h == nil || h.PlanStoreV3 == nil || ctx == nil {
|
|
return
|
|
}
|
|
go func() {
|
|
ticker := time.NewTicker(time.Hour)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case now := <-ticker.C:
|
|
cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
if _, err := h.PlanStoreV3.CleanupExpired(cleanupCtx, now); err != nil {
|
|
slog.Warn("playback v3 cleanup failed", "error", err)
|
|
}
|
|
cancel()
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
func (h *PlaybackHandler) allowRouteEventV3(userID int, attemptID string) bool {
|
|
attemptKey := fmt.Sprintf("attempt:%d:%s", userID, attemptID)
|
|
userKey := fmt.Sprintf("user:%d", userID)
|
|
now := time.Now()
|
|
h.v3EventRateMu.Lock()
|
|
defer h.v3EventRateMu.Unlock()
|
|
if h.v3EventRates == nil {
|
|
h.v3EventRates = make(map[string]v3EventRate)
|
|
}
|
|
attemptEntry := h.v3EventRates[attemptKey]
|
|
if attemptEntry.windowStart.IsZero() || now.Sub(attemptEntry.windowStart) >= time.Minute {
|
|
attemptEntry = v3EventRate{windowStart: now}
|
|
}
|
|
userEntry := h.v3EventRates[userKey]
|
|
if userEntry.windowStart.IsZero() || now.Sub(userEntry.windowStart) >= time.Minute {
|
|
userEntry = v3EventRate{windowStart: now}
|
|
}
|
|
if attemptEntry.count >= 120 || userEntry.count >= 600 {
|
|
return false
|
|
}
|
|
attemptEntry.count++
|
|
userEntry.count++
|
|
h.v3EventRates[attemptKey] = attemptEntry
|
|
h.v3EventRates[userKey] = userEntry
|
|
if len(h.v3EventRates) > 10_000 {
|
|
for candidate, value := range h.v3EventRates {
|
|
if now.Sub(value.windowStart) > 2*time.Minute {
|
|
delete(h.v3EventRates, candidate)
|
|
}
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
func (h *PlaybackHandler) enqueueRouteEventV3(event playback.RouteEventRecordV3) {
|
|
if h == nil || h.PlanStoreV3 == nil {
|
|
return
|
|
}
|
|
h.v3EventOnce.Do(func() {
|
|
h.v3EventQueue = make(chan playback.RouteEventRecordV3, 512)
|
|
go func() {
|
|
for value := range h.v3EventQueue {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
|
if err := h.PlanStoreV3.RecordRouteEvent(ctx, value); err != nil {
|
|
slog.Warn("playback route event write failed", "error", err, "event", value.Event)
|
|
}
|
|
cancel()
|
|
}
|
|
}()
|
|
})
|
|
select {
|
|
case h.v3EventQueue <- event:
|
|
default:
|
|
slog.Warn("playback route event dropped", "event", event.Event, "playback_attempt_id", event.PlaybackAttemptID)
|
|
}
|
|
}
|
|
|
|
func (h *PlaybackHandler) plannerSettingsV3(ctx context.Context) playback.PlannerSettingsV3 {
|
|
settings := playback.PlannerSettingsV3{TranscodeEnabled: h.playbackConfig().TranscodeEnabled}
|
|
if h.SettingsRepo != nil {
|
|
value, _ := h.SettingsRepo.Get(ctx, "allow_4k_transcode")
|
|
settings.Allow4KTranscode = strings.EqualFold(value, "true")
|
|
}
|
|
return settings
|
|
}
|
|
|
|
func resolveV3AudioIndex(file *models.MediaFile, trackID string, fallback *int) (int, error) {
|
|
index := 0
|
|
if trackID != "" {
|
|
fileID, kind, ordinal, ok := playback.ParseTrackIDV3(trackID)
|
|
if !ok || kind != "audio" || file == nil || fileID != file.ID {
|
|
return 0, errors.New("selected audio track identity is invalid")
|
|
}
|
|
index = ordinal
|
|
} else if fallback != nil {
|
|
index = *fallback
|
|
}
|
|
if file == nil || len(file.AudioTracks) == 0 {
|
|
if index == 0 {
|
|
return 0, nil
|
|
}
|
|
return 0, errors.New("selected audio track is unavailable")
|
|
}
|
|
if index < 0 || index >= len(file.AudioTracks) {
|
|
return 0, errors.New("selected audio track is unavailable")
|
|
}
|
|
return index, nil
|
|
}
|
|
|
|
func remapAudioIndexV3(source, target *models.MediaFile, index int) int {
|
|
if source == nil || target == nil || index < 0 || index >= len(source.AudioTracks) {
|
|
return normalizeAudioTrackIndex(target, index)
|
|
}
|
|
wanted := source.AudioTracks[index]
|
|
for i, candidate := range target.AudioTracks {
|
|
if strings.EqualFold(candidate.Codec, wanted.Codec) && strings.EqualFold(candidate.Language, wanted.Language) && candidate.Channels == wanted.Channels {
|
|
return i
|
|
}
|
|
}
|
|
return normalizeAudioTrackIndex(target, index)
|
|
}
|
|
|
|
// remapAudioSelectionV3 rebinds the request's audio selection when the
|
|
// effective media file changes. ID-only selections are equally file-bound:
|
|
// the stale ID would be rejected against the new file's track list
|
|
// downstream, so derive the source index from it and remap like any other.
|
|
func remapAudioSelectionV3(source, target *models.MediaFile, request *playback.StartRequestV3) error {
|
|
if request == nil || source == nil || target == nil || source.ID == target.ID {
|
|
return nil
|
|
}
|
|
if request.AudioTrackIndex == nil {
|
|
if request.AudioTrackID == "" {
|
|
return nil
|
|
}
|
|
fileID, kind, ordinal, ok := playback.ParseTrackIDV3(request.AudioTrackID)
|
|
if !ok || kind != "audio" || fileID != source.ID {
|
|
return errors.New("The selected audio track identity is invalid for the source file.")
|
|
}
|
|
request.AudioTrackIndex = &ordinal
|
|
}
|
|
remapped := remapAudioIndexV3(source, target, *request.AudioTrackIndex)
|
|
request.AudioTrackIndex = &remapped
|
|
request.AudioTrackID = playback.TrackIDV3(target.ID, "audio", remapped)
|
|
return nil
|
|
}
|
|
|
|
func (h *PlaybackHandler) remapSubtitleSelectionV3(ctx context.Context, source, target *models.MediaFile, request *playback.StartRequestV3) error {
|
|
if request == nil || source == nil || target == nil || source.ID == target.ID {
|
|
return nil
|
|
}
|
|
if request.SubtitleTrackIndex == nil {
|
|
// ID-only selections are equally file-bound: the stale ID would be
|
|
// parsed against the alternate file's track list downstream, so
|
|
// derive the source index from it and remap like any other.
|
|
if request.SubtitleTrackID == "" {
|
|
return nil
|
|
}
|
|
fileID, kind, ordinal, ok := playback.ParseTrackIDV3(request.SubtitleTrackID)
|
|
if !ok || kind != "subtitle" || fileID != source.ID {
|
|
return errors.New("The selected subtitle track identity is invalid for the source file.")
|
|
}
|
|
request.SubtitleTrackIndex = &ordinal
|
|
}
|
|
index := *request.SubtitleTrackIndex
|
|
if index < 0 {
|
|
return errors.New("The selected subtitle track index is invalid.")
|
|
}
|
|
targetIndex := -1
|
|
switch {
|
|
case index < len(source.ExternalSubtitles):
|
|
wanted := source.ExternalSubtitles[index]
|
|
for candidateIndex, candidate := range target.ExternalSubtitles {
|
|
if strings.EqualFold(candidate.Language, wanted.Language) && strings.EqualFold(candidate.Format, wanted.Format) && candidate.Forced == wanted.Forced {
|
|
targetIndex = candidateIndex
|
|
break
|
|
}
|
|
}
|
|
case index < len(source.ExternalSubtitles)+len(source.SubtitleTracks):
|
|
wanted := source.SubtitleTracks[index-len(source.ExternalSubtitles)]
|
|
for candidateIndex, candidate := range target.SubtitleTracks {
|
|
if strings.EqualFold(candidate.Language, wanted.Language) && strings.EqualFold(candidate.Codec, wanted.Codec) && candidate.Forced == wanted.Forced {
|
|
targetIndex = len(target.ExternalSubtitles) + candidateIndex
|
|
break
|
|
}
|
|
}
|
|
default:
|
|
if h.SubtitleRepo != nil {
|
|
sourceDownloaded, sourceErr := h.SubtitleRepo.ListDownloadedSubtitles(ctx, source.ID)
|
|
targetDownloaded, targetErr := h.SubtitleRepo.ListDownloadedSubtitles(ctx, target.ID)
|
|
downloadedIndex := index - len(source.ExternalSubtitles) - len(source.SubtitleTracks)
|
|
if sourceErr == nil && targetErr == nil && downloadedIndex >= 0 && downloadedIndex < len(sourceDownloaded) {
|
|
wanted := sourceDownloaded[downloadedIndex]
|
|
for candidateIndex, candidate := range targetDownloaded {
|
|
if strings.EqualFold(candidate.Language, wanted.Language) && strings.EqualFold(string(candidate.Format), string(wanted.Format)) && strings.EqualFold(candidate.ReleaseName, wanted.ReleaseName) {
|
|
targetIndex = len(target.ExternalSubtitles) + len(target.SubtitleTracks) + candidateIndex
|
|
break
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if targetIndex < 0 {
|
|
return errors.New("The selected subtitle track is unavailable in the effective file version.")
|
|
}
|
|
request.SubtitleTrackIndex = &targetIndex
|
|
request.SubtitleTrackID = playback.TrackIDV3(target.ID, "subtitle", targetIndex)
|
|
return nil
|
|
}
|
|
|
|
func sessionStartErrorV3(err error) *transportErrorV3 {
|
|
switch {
|
|
case errors.Is(err, playback.ErrTooManyStreams), errors.Is(err, playback.ErrTooManyTranscodes):
|
|
return &transportErrorV3{reason: "capacity_unavailable", message: "Playback capacity is currently unavailable.", retryable: true}
|
|
case errors.Is(err, playback.ErrTranscodingDisabled), errors.Is(err, playback.ErrAudioTranscodingDisabled):
|
|
return &transportErrorV3{reason: "transcoding_disabled", message: "The selected server adaptation is disabled."}
|
|
case errors.Is(err, playback.ErrPlaybackNotAllowed):
|
|
return &transportErrorV3{reason: "policy_denied", message: "Playback is denied by server policy."}
|
|
default:
|
|
return &transportErrorV3{reason: "internal_error", message: "Failed to start the playback session.", cause: err}
|
|
}
|
|
}
|
|
|
|
func (h *PlaybackHandler) persistTerminalStartDecisionV3(ctx context.Context, userID int, profileID string, req playback.StartRequestV3, requestDigest string, requestedFileID, effectiveFileID int, response playback.DecisionResponseV3) (playback.DecisionResponseV3, error) {
|
|
record := playback.AttemptRecordV3{
|
|
PlaybackAttemptID: req.PlaybackAttemptID,
|
|
UserID: userID,
|
|
ProfileID: profileID,
|
|
RequestedMediaFileID: requestedFileID,
|
|
EffectiveMediaFileID: effectiveFileID,
|
|
NormalizedRequest: req,
|
|
StartResponse: response,
|
|
RequestDigest: requestDigest,
|
|
ExpiresAt: time.Now().Add(playback.MaxTokenTTL),
|
|
}
|
|
if err := h.PlanStoreV3.SaveAttempt(ctx, record); err == nil {
|
|
return response, nil
|
|
} else if !errors.Is(err, playback.ErrPlaybackAttemptExistsV3) {
|
|
return playback.DecisionResponseV3{}, err
|
|
}
|
|
|
|
existing, err := h.PlanStoreV3.GetAttemptByPlaybackAttemptID(ctx, req.PlaybackAttemptID)
|
|
if err != nil {
|
|
return playback.DecisionResponseV3{}, err
|
|
}
|
|
if existing.UserID != userID || existing.ProfileID != profileID ||
|
|
existing.RequestedMediaFileID != requestedFileID || existing.RequestDigest != requestDigest {
|
|
return playback.DecisionResponseV3{}, playback.ErrIdempotencyKeyReusedV3
|
|
}
|
|
return decisionResponseFromAttemptV3(existing), nil
|
|
}
|
|
|
|
func (h *PlaybackHandler) startFailureDecisionV3(ctx context.Context, userID int, profileID string, req playback.StartRequestV3, requestDigest string, requestedFileID, effectiveFileID int, failure *transportErrorV3) (playback.DecisionResponseV3, error) {
|
|
response := playback.NewTerminalResponseV3(failure.reason, failure.message, failure.retryable)
|
|
return h.persistTerminalStartDecisionV3(ctx, userID, profileID, req, requestDigest, requestedFileID, effectiveFileID, response)
|
|
}
|
|
|
|
func writeStartAttemptPersistenceErrorV3(w http.ResponseWriter, err error) {
|
|
if errors.Is(err, playback.ErrIdempotencyKeyReusedV3) {
|
|
writeError(w, http.StatusConflict, "playback_attempt_reused", "The playback attempt ID belongs to a different request")
|
|
return
|
|
}
|
|
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to persist the playback decision")
|
|
}
|
|
|
|
func decisionResponseFromAttemptV3(record *playback.AttemptRecordV3) playback.DecisionResponseV3 {
|
|
if record == nil {
|
|
return playback.DecisionResponseV3{}
|
|
}
|
|
if record.StartResponse.Outcome != "" || record.StartResponse.Terminal != nil || record.StartResponse.PlaybackPlan != nil {
|
|
return normalizeDecisionResponseV3(record.StartResponse)
|
|
}
|
|
plan := record.CurrentPlan
|
|
if plan.AppliedQuirks == nil {
|
|
plan.AppliedQuirks = []playback.AppliedQuirkV3{}
|
|
}
|
|
if plan.RuntimeCorrections == nil {
|
|
plan.RuntimeCorrections = []string{}
|
|
}
|
|
return normalizeDecisionResponseV3(playback.DecisionResponseV3{ProtocolVersion: playback.ProtocolV3, ServerFeatures: playback.ServerFeaturesV3(), Outcome: playback.OutcomePlayableV3, SessionID: record.SessionID, PlaybackPlan: &plan})
|
|
}
|
|
|
|
func normalizeDecisionResponseV3(response playback.DecisionResponseV3) playback.DecisionResponseV3 {
|
|
if response.ServerFeatures == nil {
|
|
response.ServerFeatures = playback.ServerFeaturesV3()
|
|
}
|
|
if response.PlaybackPlan == nil {
|
|
return response
|
|
}
|
|
plan := response.PlaybackPlan
|
|
if plan.Stream.Headers == nil {
|
|
plan.Stream.Headers = map[string]string{}
|
|
}
|
|
if plan.Transformations == nil {
|
|
plan.Transformations = []playback.TransformationV3{}
|
|
}
|
|
if plan.AppliedQuirks == nil {
|
|
plan.AppliedQuirks = []playback.AppliedQuirkV3{}
|
|
}
|
|
if plan.RuntimeCorrections == nil {
|
|
plan.RuntimeCorrections = []string{}
|
|
}
|
|
if plan.AvailableQualities == nil {
|
|
plan.AvailableQualities = []playback.AvailableQualityV3{}
|
|
}
|
|
if plan.DegradationWarnings == nil {
|
|
plan.DegradationWarnings = []playback.DegradationWarningV3{}
|
|
}
|
|
if plan.Subtitle.Inventory == nil {
|
|
plan.Subtitle.Inventory = []playback.SubtitleInventoryItemV3{}
|
|
}
|
|
return response
|
|
}
|
|
|
|
func completedReplanResponseMatchesAttemptV3(raw json.RawMessage, record *playback.AttemptRecordV3) bool {
|
|
if record == nil {
|
|
return false
|
|
}
|
|
var response playback.DecisionResponseV3
|
|
if len(raw) == 0 || json.Unmarshal(raw, &response) != nil {
|
|
return false
|
|
}
|
|
if response.PlaybackPlan == nil {
|
|
// Terminal responses deliberately leave the attempt plan untouched. Their
|
|
// freshness is carried by CurrentReplanRequestID (and its DB trigger).
|
|
return response.Terminal != nil
|
|
}
|
|
if response.SessionID != record.SessionID || response.PlaybackPlan.SessionID != record.SessionID {
|
|
return false
|
|
}
|
|
candidate, candidateErr := json.Marshal(response.PlaybackPlan)
|
|
current, currentErr := json.Marshal(record.CurrentPlan)
|
|
return candidateErr == nil && currentErr == nil && bytes.Equal(candidate, current)
|
|
}
|
|
|
|
func appliedQuirkIDsV3(plan *playback.PlanV3) []string {
|
|
if plan == nil {
|
|
return nil
|
|
}
|
|
result := make([]string, 0, len(plan.AppliedQuirks))
|
|
for _, quirk := range plan.AppliedQuirks {
|
|
result = append(result, quirk.ID)
|
|
}
|
|
return result
|
|
}
|
|
|
|
func appliedQuirkRevisionV3(plan *playback.PlanV3) string {
|
|
if plan == nil || len(plan.AppliedQuirks) == 0 {
|
|
return ""
|
|
}
|
|
return plan.AppliedQuirks[0].RegistryRevision
|
|
}
|
|
|
|
func writeV3FileError(w http.ResponseWriter, err error) {
|
|
if errors.Is(err, catalog.ErrItemNotFound) || errors.Is(err, catalog.ErrEpisodeNotFound) {
|
|
writeError(w, http.StatusNotFound, "not_found", "Media file not found")
|
|
return
|
|
}
|
|
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to authorize media file")
|
|
}
|
|
func readBoundedV3Body(w http.ResponseWriter, r *http.Request, limit int64) ([]byte, error) {
|
|
return ioReadAllV3(http.MaxBytesReader(w, r.Body, limit))
|
|
}
|
|
func ioReadAllV3(reader interface{ Read([]byte) (int, error) }) ([]byte, error) {
|
|
var buffer bytes.Buffer
|
|
_, err := buffer.ReadFrom(reader)
|
|
return buffer.Bytes(), err
|
|
}
|
|
func chiURLParamV3(r *http.Request, key string) string { return chi.URLParam(r, key) }
|
|
func floatOrZeroHandlerV3(v *float64) float64 {
|
|
if v == nil {
|
|
return 0
|
|
}
|
|
return *v
|
|
}
|
|
|
|
func intOrZeroHandlerV3(v *int) int {
|
|
if v == nil {
|
|
return 0
|
|
}
|
|
return *v
|
|
}
|
|
func firstNonEmptyHandlerV3(values ...string) string {
|
|
for _, value := range values {
|
|
if value != "" {
|
|
return value
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
func subtitleMIMEV3(format string) string {
|
|
switch strings.ToLower(format) {
|
|
case "ass", "ssa":
|
|
return "text/x-ssa"
|
|
case "srt", "subrip":
|
|
return "application/x-subrip"
|
|
case "pgs", "hdmv_pgs_subtitle":
|
|
return "application/octet-stream"
|
|
default:
|
|
return subtitleMIMEVTTV3
|
|
}
|
|
}
|
|
|
|
func forceSubtitleExtensionV3(rawURL, extension string) string {
|
|
pathPart, query, hasQuery := strings.Cut(rawURL, "?")
|
|
if slash := strings.LastIndex(pathPart, "/"); slash >= 0 {
|
|
if dot := strings.LastIndex(pathPart[slash+1:], "."); dot >= 0 {
|
|
pathPart = pathPart[:slash+1+dot] + extension
|
|
} else {
|
|
pathPart += extension
|
|
}
|
|
}
|
|
if hasQuery {
|
|
return pathPart + "?" + query
|
|
}
|
|
return pathPart
|
|
}
|
|
|
|
func remuxDVModeForPlanV3(plan *playback.PlanV3) playback.RemuxDVMode {
|
|
if plan == nil {
|
|
return ""
|
|
}
|
|
for _, transformation := range plan.Transformations {
|
|
if transformation.Name == playback.TransformationServerDV7HDR10V3 {
|
|
return playback.RemuxDVStripToHDR10V3
|
|
}
|
|
}
|
|
if plan.Source.DVProfile == 0 {
|
|
return ""
|
|
}
|
|
if plan.Source.DVProfile == 7 {
|
|
// Without the strip transformation a P7 remux would drop the
|
|
// enhancement layer and leave dangling RPUs. A P7 plan claiming Dolby
|
|
// Vision is a client-side transform of the original bytes, so any
|
|
// remux attempt against this session must still be rejected.
|
|
return playback.RemuxDVRejectP7V3
|
|
}
|
|
if plan.Claims.Video.DolbyVision {
|
|
return playback.RemuxDVPreserveV3
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func videoBitstreamFilterForPlanV3(plan *playback.PlanV3) string {
|
|
if plan == nil {
|
|
return ""
|
|
}
|
|
for _, transformation := range plan.Transformations {
|
|
if transformation.Executor == playback.ExecutorServerV3 && transformation.Name == playback.TransformationServerDV7HDR10V3 && transformation.RecipeVersion == "1" {
|
|
return playback.DV7ToHDR10BitstreamFilter
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// lazyDVRPUStrippableV3 defers (and memoizes) the per-source RPU probe so the
|
|
// planner only shells out to ffmpeg when a Dolby Vision strip route is
|
|
// genuinely on the table; every other start never touches it.
|
|
//
|
|
// The probe belongs to planning, not to the transport: the plan's HDR10 promise
|
|
// and the durable session's RemuxDVMode are both derived from the strip
|
|
// decision and are re-read by the restart and audio-switch paths, so
|
|
// suppressing the filter downstream would leave those claims describing a
|
|
// stream the server is no longer producing.
|
|
func (h *PlaybackHandler) lazyDVRPUStrippableV3(ctx context.Context, file *models.MediaFile) func() bool {
|
|
if file == nil || strings.TrimSpace(file.FilePath) == "" {
|
|
return nil
|
|
}
|
|
var once sync.Once
|
|
strippable := true
|
|
return func() bool {
|
|
once.Do(func() {
|
|
strippable = playback.DVRPUStrippable(ctx, h.playbackConfig().FFmpegPath, file.FilePath)
|
|
})
|
|
return strippable
|
|
}
|
|
}
|
|
|
|
func configureHLSTimelineV3(plan *playback.PlanV3, videoCodec string, segmentDuration int, durationSeconds float64) (float64, int) {
|
|
if plan == nil {
|
|
return 0, 0
|
|
}
|
|
requested := plan.Timeline.SourceStartSeconds
|
|
seek := alignedSeekSeconds(requested, segmentDuration, videoCodec)
|
|
startSegment := computeStartSegment(seek, segmentDuration)
|
|
plan.Timeline.SourceStartSeconds = requested
|
|
usesGrowingManifest := strings.EqualFold(videoCodec, "copy") ||
|
|
!playback.CanGenerateSyntheticManifest(durationSeconds, segmentDuration)
|
|
if usesGrowingManifest {
|
|
// Encoded streams seek to the preceding segment boundary. Preserve the
|
|
// requested sub-segment offset so playback still begins at the exact
|
|
// requested source position. Copy seeks are already exact, making this 0.
|
|
plan.Timeline.PlayerStartSeconds = max(0, requested-seek)
|
|
plan.Timeline.StreamOriginSeconds = seek
|
|
plan.Timeline.TimelineOffsetSeconds = seek
|
|
windowStart := seek
|
|
plan.Timeline.SeekWindowStartSeconds = &windowStart
|
|
// This transport is served from FFmpeg's live, still-growing playlist
|
|
// (see BuildPlaybackManifest), so the seekable extent is whatever has
|
|
// been produced so far — a value this plan cannot know and could not
|
|
// keep current if it did. Publishing the media runtime here instead
|
|
// made the window look *complete*, which clients read as proof that
|
|
// any target inside it is locally seekable; they then native-seek past
|
|
// the produced head instead of asking for a reanchor. Leaving the end
|
|
// open marks the window incomplete, which with can_seek_anywhere=false
|
|
// routes every seek back through the server.
|
|
//
|
|
// The media runtime is published on source.duration_seconds, which is
|
|
// a fact about the file rather than a claim about this transport.
|
|
plan.Timeline.SeekWindowEndSeconds = nil
|
|
plan.Timeline.CanSeekAnywhere = false
|
|
plan.Timeline.SeekRestoration = "source_position"
|
|
} else {
|
|
plan.Timeline.PlayerStartSeconds = requested
|
|
plan.Timeline.StreamOriginSeconds = 0
|
|
plan.Timeline.TimelineOffsetSeconds = 0
|
|
plan.Timeline.SeekWindowStartSeconds = nil
|
|
plan.Timeline.SeekWindowEndSeconds = nil
|
|
plan.Timeline.CanSeekAnywhere = durationSeconds > 0
|
|
plan.Timeline.SeekRestoration = seekRestorationPlayerV3
|
|
}
|
|
return seek, startSegment
|
|
}
|
|
|
|
var diagnosticKeysV3 = map[string]struct{}{
|
|
"decoder_name": {}, "decoder_init_ms": {}, "first_frame_ms": {},
|
|
"device_model": {}, "requested_quality": {}, "effective_quality": {},
|
|
"pcm_recovery": {}, "retry_outcome": {}, "replan_request_id": {},
|
|
"video_mime": {}, "video_codecs": {}, "video_width": {}, "video_height": {},
|
|
"color_transfer": {}, "color_range": {},
|
|
"error_code": {}, "error_code_name": {}, "error_cause": {},
|
|
"transformation_name": {}, "transformation_version": {}, "transformation_stage": {},
|
|
"input_dv_profile": {}, "output_dv_profile": {}, "rpu_converted_count": {},
|
|
"rpu_failed_count": {}, "el_nal_dropped_count": {}, "sample_count": {},
|
|
"transform_buffer_peak_bytes": {}, "requested_media_file_id": {}, "effective_media_file_id": {},
|
|
"audio_output_mode": {}, "audio_mime": {}, "audio_channels": {}, "audio_decoder_name": {},
|
|
"correction_id": {}, "correction_stage": {},
|
|
"network_transport": {}, "network_metered": {}, "network_validated": {},
|
|
"bandwidth_estimate_kbps": {}, "link_downstream_kbps": {},
|
|
"target_source_position_seconds": {}, "reason": {},
|
|
}
|
|
|
|
func validRouteEventV3(event playback.RouteEventV3) bool {
|
|
if event.ProtocolVersion != playback.ProtocolV3 || len(event.PlaybackAttemptID) < 8 || len(event.PlaybackAttemptID) > 128 || len(event.OutputContextID) > 128 || len(event.SessionID) > 128 || len(event.PlanID) > 128 || len(event.PlanAttemptID) > 128 || len(event.PlanAttemptKey) > 128 || len(event.FailureClassification) > 64 || len(event.FallbackReason) > 64 || len(event.AppliedQuirkIDs) > 16 || len(event.QuirkRegistryRevision) > 128 || len(event.Diagnostics) > 32 {
|
|
return false
|
|
}
|
|
for _, id := range event.AppliedQuirkIDs {
|
|
if len(id) == 0 || len(id) > 128 {
|
|
return false
|
|
}
|
|
}
|
|
return playback.ValidRouteEventNameV3(event.Event)
|
|
}
|
|
func sanitizeDiagnosticsV3(values map[string]string) map[string]string {
|
|
// Iterate the approved keys, not the client map: map iteration order is
|
|
// random, so a count-limited walk over client keys would keep an
|
|
// arbitrary subset and drop different diagnostics on identical retries.
|
|
result := make(map[string]string)
|
|
for key := range diagnosticKeysV3 {
|
|
value, ok := values[key]
|
|
if !ok {
|
|
continue
|
|
}
|
|
value = strings.TrimSpace(value)
|
|
if len(value) > 256 {
|
|
value = value[:256]
|
|
}
|
|
result[key] = value
|
|
}
|
|
return result
|
|
}
|
|
|
|
func containsStringFoldV3(values []string, wanted string) bool {
|
|
for _, value := range values {
|
|
if strings.EqualFold(value, wanted) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// containsStringExactV3 compares attempt keys byte-for-byte: they are
|
|
// case-sensitive FNV hex digests, so case-folding would treat distinct keys
|
|
// as equal.
|
|
func containsStringExactV3(values []string, wanted string) bool {
|
|
wanted = strings.TrimSpace(wanted)
|
|
for _, value := range values {
|
|
if strings.TrimSpace(value) == wanted {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func optionalIntEqualV3(left, right *int) bool {
|
|
if left == nil || right == nil {
|
|
return left == nil && right == nil
|
|
}
|
|
return *left == *right
|
|
}
|
|
|
|
func optionalFloatEqualV3(left, right *float64) bool {
|
|
if left == nil || right == nil {
|
|
return left == nil && right == nil
|
|
}
|
|
return *left == *right
|
|
}
|