diff --git a/internal/api/handlers/playback.go b/internal/api/handlers/playback.go index bc7be4f0..6bf5bb40 100644 --- a/internal/api/handlers/playback.go +++ b/internal/api/handlers/playback.go @@ -956,6 +956,13 @@ func (h *PlaybackHandler) persistProgress(ctx context.Context, session *playback if session == nil || session.DisableProgressPersistence { return } + // Position 0 carries no resume information (mirrors persistStopAndHistory + // and the jellycompat report path). Progress is last-write-wins, so an + // early zero heartbeat — e.g. before a client finishes seeking to its + // resume point — must not wipe the stored resume position. + if session.Position <= 0 { + return + } file, err := h.loadFileByPreferredID(ctx, requestedMediaFileID(session), session.MediaFileID) targetID := playbackProgressTarget(file) @@ -1600,6 +1607,7 @@ func (h *PlaybackHandler) HandleStartPlayback(w http.ResponseWriter, r *http.Req // Resolve media path if possible. if effectiveFile != nil { tokenClaims.MediaPath = effectiveFile.FilePath + tokenClaims.DVProfile = effectiveFile.PrimaryDVProfile() } tokenClaims.TranscodeAudio = session.TranscodeAudio @@ -2160,6 +2168,7 @@ func (h *PlaybackHandler) HandleChangeAudioTrack(w http.ResponseWriter, r *http. MediaPath: file.FilePath, TranscodeAudio: updatedSession.TranscodeAudio, AudioTrackIndex: req.AudioTrackIndex, + DVProfile: file.PrimaryDVProfile(), UserID: updatedSession.UserID, ProfileID: updatedSession.ProfileID, MediaFileID: updatedSession.MediaFileID, @@ -2174,8 +2183,7 @@ func (h *PlaybackHandler) HandleChangeAudioTrack(w http.ResponseWriter, r *http. resp.StreamURL = proxyNode.URL + "/stream/remux/" + token case playback.PlayTranscode: resp.StreamURL = proxyNode.URL + "/stream/transcode/" + token + "/master.m3u8" - } - } + } } } } } diff --git a/internal/api/handlers/stream.go b/internal/api/handlers/stream.go index a82f6613..b8b9a9f2 100644 --- a/internal/api/handlers/stream.go +++ b/internal/api/handlers/stream.go @@ -154,7 +154,7 @@ func (h *StreamHandler) HandleStream(w http.ResponseWriter, r *http.Request) { seekSeconds = s } } - if err := playback.ServeRemux(w, r, file.FilePath, "mp4", seekSeconds, session.TranscodeAudio, session.AudioTrackIndex); err != nil { + if err := playback.ServeRemux(w, r, file.FilePath, "mp4", seekSeconds, session.TranscodeAudio, session.AudioTrackIndex, file.PrimaryDVProfile()); err != nil { h.handleTransportStartFailure(r.Context(), session, file, err) } diff --git a/internal/jellycompat/handlers_playback.go b/internal/jellycompat/handlers_playback.go index 35fe49fb..b2ed8707 100644 --- a/internal/jellycompat/handlers_playback.go +++ b/internal/jellycompat/handlers_playback.go @@ -313,6 +313,7 @@ func (h *PlaybackHandler) buildProxyRedirectURL( TranscodeAudio: source.TranscodeAudio, AudioTrackIndex: audioTrackIndex, TranscodeNode: transcodeNodeURL, + DVProfile: file.PrimaryDVProfile(), } token, err := streamtoken.Sign(claims, h.JWTSecret, 24*time.Hour) if err != nil { diff --git a/internal/jellycompat/streams.go b/internal/jellycompat/streams.go index fa7c6e51..5cd14319 100644 --- a/internal/jellycompat/streams.go +++ b/internal/jellycompat/streams.go @@ -132,7 +132,7 @@ func (h *PlaybackHandler) HandleVideoStream(w http.ResponseWriter, r *http.Reque if resolvedAudioTrackIndex, ok := compatAudioTrackIndex(*source); ok { audioTrackIndex = resolvedAudioTrackIndex } - _ = playback.ServeRemux(w, r, file.FilePath, "mp4", seekSeconds, source.TranscodeAudio, audioTrackIndex) + _ = playback.ServeRemux(w, r, file.FilePath, "mp4", seekSeconds, source.TranscodeAudio, audioTrackIndex, file.PrimaryDVProfile()) default: _ = playback.ServeDirectPlay(w, r, file.FilePath) } diff --git a/internal/models/media.go b/internal/models/media.go index eaf3d013..98041cb6 100644 --- a/internal/models/media.go +++ b/internal/models/media.go @@ -144,6 +144,15 @@ type OverlaySummary struct { MultiSub bool `json:"multi_sub,omitempty"` // ≥1 subtitle track (embedded or external) } +// PrimaryDVProfile returns the Dolby Vision profile of the first video +// track (0 when none/unprobed). +func (f *MediaFile) PrimaryDVProfile() int { + if f == nil || len(f.VideoTracks) == 0 { + return 0 + } + return f.VideoTracks[0].DVProfile +} + // VideoTrack represents a probed video stream stored as JSONB. type VideoTrack struct { Title string `json:"title,omitempty"` diff --git a/internal/playback/remux.go b/internal/playback/remux.go index a0223523..eaabae1e 100644 --- a/internal/playback/remux.go +++ b/internal/playback/remux.go @@ -1,9 +1,11 @@ package playback import ( + "bytes" "context" "fmt" "io" + "log/slog" "net/http" "os" "os/exec" @@ -14,6 +16,9 @@ import ( var ( resolvedFFmpegPath string ffmpegOnce sync.Once + + doviRPUAvailable bool + doviRPUOnce sync.Once ) // ffmpegBinary returns the path to the ffmpeg binary. @@ -30,6 +35,32 @@ func ffmpegBinary() string { return resolvedFFmpegPath } +// supportsDoviRPUFilter reports whether the resolved ffmpeg binary ships the +// dovi_rpu bitstream filter (FFmpeg 7.1+). Probed once per process. +func supportsDoviRPUFilter() bool { + doviRPUOnce.Do(func() { + out, err := exec.Command(ffmpegBinary(), "-hide_banner", "-bsfs").Output() + doviRPUAvailable = err == nil && bytes.Contains(out, []byte("dovi_rpu")) + if !doviRPUAvailable { + slog.Warn("ffmpeg lacks the dovi_rpu bitstream filter (needs FFmpeg 7.1+); " + + "Dolby Vision profile 7 remuxes will keep their dangling dual-layer RPUs") + } + }) + return doviRPUAvailable +} + +// remuxDVProfile neutralizes a Dolby Vision profile the local ffmpeg cannot +// handle. Profile 7 is the only profile that triggers an RPU strip in +// buildRemuxArgs; when the dovi_rpu filter is unavailable the remux must +// still start (an unknown bitstream filter aborts ffmpeg immediately), so +// fall back to the pre-strip behavior instead of failing playback. +func remuxDVProfile(dvProfile int, canStripRPU bool) int { + if dvProfile == 7 && !canStripRPU { + return 0 + } + return dvProfile +} + // RemuxSession represents a running ffmpeg remux process that copies // codecs to a new container format without re-encoding. type RemuxSession struct { @@ -44,7 +75,12 @@ type RemuxSession struct { // pipe:1 for stdout output. // When transcodeAudio is true, video is copied but audio is transcoded to // stereo AAC (handles cases like DTS/TrueHD that browsers cannot decode). -func buildRemuxArgs(filePath, outputFormat string, seekSeconds float64, transcodeAudio bool, audioTrackIndex int) []string { +// dvProfile is the file's Dolby Vision profile (0 = none). Profile 7 remuxes +// strip DV RPUs: the enhancement layer is dropped by the video map below, so +// the RPUs would dangle — stripping yields a clean HDR10 base layer (the +// Apple-parity fallback for devices without a P7 decoder). Profile 8 RPUs +// stay: the base layer is self-contained and DV clients can render it. +func buildRemuxArgs(filePath, outputFormat string, seekSeconds float64, transcodeAudio bool, audioTrackIndex int, dvProfile int) []string { args := []string{ "-nostdin", "-hide_banner", @@ -88,6 +124,10 @@ func buildRemuxArgs(filePath, outputFormat string, seekSeconds float64, transcod } args = append(args, "-sn", "-dn") + if dvProfile == 7 { + args = append(args, "-bsf:v", "dovi_rpu=strip=1") + } + if transcodeAudio { // Video copy + stereo AAC encode is effectively single-threaded work. // ffmpeg's default auto-threading spawns one filter thread per CPU @@ -122,11 +162,12 @@ func buildRemuxArgs(filePath, outputFormat string, seekSeconds float64, transcod // // When transcodeAudio is true video is copied but audio is transcoded to AAC. // The caller must call Close() when done to clean up resources. -func StartRemux(ctx context.Context, filePath, outputFormat string, seekSeconds float64, transcodeAudio bool, audioTrackIndex int) (*RemuxSession, error) { +func StartRemux(ctx context.Context, filePath, outputFormat string, seekSeconds float64, transcodeAudio bool, audioTrackIndex int, dvProfile int) (*RemuxSession, error) { ctx, cancel := context.WithCancel(ctx) bin := ffmpegBinary() - args := buildRemuxArgs(filePath, outputFormat, seekSeconds, transcodeAudio, audioTrackIndex) + args := buildRemuxArgs(filePath, outputFormat, seekSeconds, transcodeAudio, audioTrackIndex, + remuxDVProfile(dvProfile, supportsDoviRPUFilter())) cmd := exec.CommandContext(ctx, bin, args...) stdout, err := cmd.StdoutPipe() @@ -182,7 +223,7 @@ func containerMIME(format string) string { // response writer. The response is streamed (chunked transfer) since the // total size is not known in advance. // When transcodeAudio is true, audio is transcoded to AAC while video is copied. -func ServeRemux(w http.ResponseWriter, r *http.Request, filePath, outputFormat string, seekSeconds float64, transcodeAudio bool, audioTrackIndex int) error { +func ServeRemux(w http.ResponseWriter, r *http.Request, filePath, outputFormat string, seekSeconds float64, transcodeAudio bool, audioTrackIndex int, dvProfile int) error { // Check file exists before starting ffmpeg to return a proper 404. // Headers must be written before streaming begins, so we can't detect // ffmpeg errors after WriteHeader(200) has been sent. @@ -195,7 +236,7 @@ func ServeRemux(w http.ResponseWriter, r *http.Request, filePath, outputFormat s return err } - session, err := StartRemux(r.Context(), filePath, outputFormat, seekSeconds, transcodeAudio, audioTrackIndex) + session, err := StartRemux(r.Context(), filePath, outputFormat, seekSeconds, transcodeAudio, audioTrackIndex, dvProfile) if err != nil { http.Error(w, "failed to start remux", http.StatusInternalServerError) return err diff --git a/internal/playback/remux_dv_test.go b/internal/playback/remux_dv_test.go new file mode 100644 index 00000000..afa61008 --- /dev/null +++ b/internal/playback/remux_dv_test.go @@ -0,0 +1,56 @@ +package playback + +import ( + "strings" + "testing" +) + +func argsContainPair(args []string, a, b string) bool { + for i := 0; i < len(args)-1; i++ { + if args[i] == a && args[i+1] == b { + return true + } + } + return false +} + +// Profile 7 remuxes drop the enhancement-layer track (-map 0:v:0 keeps only +// the base layer), which leaves dangling dual-layer RPU metadata on the BL. +// Stripping the RPUs yields a clean HDR10 stream — both a correctness fix and +// the Apple-parity fallback presentation for devices without a P7 decoder. +func TestBuildRemuxArgsStripsDolbyVisionRPUForProfile7(t *testing.T) { + args := buildRemuxArgs("/x.mkv", "mp4", 0, false, -1, 7) + if !argsContainPair(args, "-bsf:v", "dovi_rpu=strip=1") { + t.Fatalf("profile 7 remux must strip DV RPUs, args=%v", strings.Join(args, " ")) + } +} + +// An ffmpeg without the dovi_rpu bitstream filter (pre-7.1) would abort on +// the unknown filter before producing any output. The profile must be +// neutralized so the remux still starts, keeping the pre-strip behavior. +func TestRemuxDVProfileFallsBackWithoutFilterSupport(t *testing.T) { + if got := remuxDVProfile(7, false); got != 0 { + t.Errorf("remuxDVProfile(7, false) = %d, want 0 (no strip without filter support)", got) + } + if got := remuxDVProfile(7, true); got != 7 { + t.Errorf("remuxDVProfile(7, true) = %d, want 7", got) + } + for _, profile := range []int{0, 5, 8} { + for _, canStrip := range []bool{false, true} { + if got := remuxDVProfile(profile, canStrip); got != profile { + t.Errorf("remuxDVProfile(%d, %t) = %d, want %d (pass through)", profile, canStrip, got, profile) + } + } + } +} + +// Profile 8 base layers are self-contained: the RPU stays valid without an +// enhancement layer and DV-capable clients can render it. Never strip. +func TestBuildRemuxArgsKeepsRPUForProfile8AndPlainFiles(t *testing.T) { + for _, profile := range []int{0, 5, 8} { + args := buildRemuxArgs("/x.mkv", "mp4", 0, false, -1, profile) + if argsContainPair(args, "-bsf:v", "dovi_rpu=strip=1") { + t.Fatalf("profile %d remux must not strip DV RPUs, args=%v", profile, strings.Join(args, " ")) + } + } +} diff --git a/internal/proxy/server.go b/internal/proxy/server.go index 01a49a41..f53bab95 100644 --- a/internal/proxy/server.go +++ b/internal/proxy/server.go @@ -163,7 +163,7 @@ func (s *Server) handleRemux(w http.ResponseWriter, r *http.Request) { seekSeconds = v } } - _ = playback.ServeRemux(w, r, claims.MediaPath, "mp4", seekSeconds, claims.TranscodeAudio, claims.AudioTrackIndex) + _ = playback.ServeRemux(w, r, claims.MediaPath, "mp4", seekSeconds, claims.TranscodeAudio, claims.AudioTrackIndex, claims.DVProfile) } func (s *Server) handleTranscodeManifest(w http.ResponseWriter, r *http.Request) { diff --git a/internal/streamtoken/token.go b/internal/streamtoken/token.go index fdef16fe..1d5d4881 100644 --- a/internal/streamtoken/token.go +++ b/internal/streamtoken/token.go @@ -28,6 +28,10 @@ type Claims struct { AudioCodec string `json:"ac,omitempty"` AudioChannels int `json:"ach,omitempty"` AudioTrackIndex int `json:"ati,omitempty"` + // DVProfile is the file's Dolby Vision profile (0 = none); remux nodes + // use it to strip dangling profile 7 RPUs. Absent in older tokens, which + // decodes as 0 (no strip — the pre-existing behavior). + DVProfile int `json:"dvp,omitempty"` // Ownership / authorization lookup keys (re-resolved at reconstruct). // Not trust assertions. diff --git a/internal/userdb/progress.go b/internal/userdb/progress.go index e1b0cc0d..4fa23d40 100644 --- a/internal/userdb/progress.go +++ b/internal/userdb/progress.go @@ -26,9 +26,13 @@ func UpdateProgress(db *sql.DB, profileID, mediaItemID string, position, duratio } now := nowUTC() // Mirrors the Postgres pgstore UpdateProgress: `completed` is a one-way - // watched latch; position resets to 0 on completion so a rewatch - // heartbeat on a completed row re-enters Continue Watching through plain - // MAX while the watched flag survives. + // watched latch; position resets to 0 on completion so `position_seconds + // > 0` means "has a resume point". Position itself is last-write-wins — + // a deliberate backward seek is a legitimate resume point (the old MAX + // clamp made "rewind and stop" resume at the stale, later position). + // Rewatching a completed row still re-enters Continue Watching (stored + // position is 0, any heartbeat replaces it) while the watched flag + // survives. query := ` INSERT INTO watch_progress (profile_id, media_item_id, position_seconds, duration_seconds, completed, updated_at) SELECT ?, ?, ?, ?, ?, ` + visibleTimestampSQL + ` @@ -39,7 +43,7 @@ func UpdateProgress(db *sql.DB, profileID, mediaItemID string, position, duratio WHERE true ON CONFLICT(profile_id, media_item_id) DO UPDATE SET position_seconds = CASE WHEN excluded.completed = 1 THEN 0 - ELSE MAX(excluded.position_seconds, watch_progress.position_seconds) END, + ELSE excluded.position_seconds END, duration_seconds = excluded.duration_seconds, completed = CASE WHEN excluded.completed = 1 THEN 1 ELSE watch_progress.completed END, diff --git a/internal/userdb/progress_test.go b/internal/userdb/progress_test.go index 774b89cb..2cfdcb01 100644 --- a/internal/userdb/progress_test.go +++ b/internal/userdb/progress_test.go @@ -441,3 +441,33 @@ func TestClearProgressBatch_CompactsDirtyInput(t *testing.T) { t.Fatalf("watch_progress row count = %d, want 1", count) } } + +// Backward seeks are legitimate resume points: UpdateProgress is +// last-write-wins on position (the old MAX clamp made "rewind and stop" +// resume at the stale later position). Mirrors the pgstore expectation. +func TestUpdateProgressBackwardSeekPersists(t *testing.T) { + db, err := sql.Open("sqlite3", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + defer db.Close() + if err := InitSchema(db); err != nil { + t.Fatalf("InitSchema: %v", err) + } + + noThreshold := userstore.ProgressThresholds{} + if err := UpdateProgress(db, "p1", "movie-1", 2400, 7200, noThreshold); err != nil { + t.Fatalf("UpdateProgress(forward): %v", err) + } + if err := UpdateProgress(db, "p1", "movie-1", 600, 7200, noThreshold); err != nil { + t.Fatalf("UpdateProgress(backward): %v", err) + } + + wp, err := GetProgress(db, "p1", "movie-1") + if err != nil { + t.Fatalf("GetProgress: %v", err) + } + if wp.PositionSeconds != 600 { + t.Errorf("PositionSeconds = %v, want 600 (last write wins)", wp.PositionSeconds) + } +} diff --git a/internal/userstore/pgstore/progress.go b/internal/userstore/pgstore/progress.go index 5ea0bcf4..013c8dac 100644 --- a/internal/userstore/pgstore/progress.go +++ b/internal/userstore/pgstore/progress.go @@ -59,10 +59,12 @@ func (s *PostgresUserStore) UpdateProgress(ctx context.Context, profileID, media } now := time.Now().UTC() // `completed` is a one-way watched latch; position resets to 0 on - // completion so `position_seconds > 0` means "has a resume point". A - // rewatch heartbeat on a completed row therefore re-enters Continue - // Watching through plain GREATEST (stored position is 0) while the - // watched flag survives. + // completion so `position_seconds > 0` means "has a resume point". + // Position itself is last-write-wins: a deliberate backward seek is a + // legitimate resume point (the old GREATEST clamp made "rewind and stop" + // resume at the stale, later position on every client). Rewatching a + // completed row still re-enters Continue Watching (stored position is 0, + // any heartbeat replaces it) while the watched flag survives. _, err := s.pool.Exec(ctx, ` WITH visible AS ( SELECT @@ -82,7 +84,7 @@ func (s *PostgresUserStore) UpdateProgress(ctx context.Context, profileID, media FROM visible ON CONFLICT(user_id, profile_id, media_item_id) DO UPDATE SET position_seconds = CASE WHEN excluded.completed THEN 0 - ELSE GREATEST(excluded.position_seconds, user_watch_progress.position_seconds) END, + ELSE excluded.position_seconds END, duration_seconds = excluded.duration_seconds, completed = CASE WHEN excluded.completed THEN TRUE ELSE user_watch_progress.completed END, diff --git a/internal/userstore/storetest/suite.go b/internal/userstore/storetest/suite.go index a48db67a..7f9b37c4 100644 --- a/internal/userstore/storetest/suite.go +++ b/internal/userstore/storetest/suite.go @@ -254,13 +254,15 @@ func testProgress(t *testing.T, newStore func(t *testing.T) userstore.UserStore) t.Errorf("PositionSeconds after forward = %v, want 1000", wp.PositionSeconds) } - // UpdateProgress backward: should NOT go backward + // UpdateProgress backward: last write wins. A deliberate backward seek is + // a legitimate resume point — clamping to the furthest position made + // "rewind and stop" resume at the old, later spot on every client. if err := store.UpdateProgress(ctx, "p1", "movie-1", 600, 7200, noThreshold); err != nil { t.Fatalf("UpdateProgress(backward): %v", err) } wp, _ = store.GetProgress(ctx, "p1", "movie-1") - if wp.PositionSeconds != 1000 { - t.Errorf("PositionSeconds after backward = %v, want 1000 (unchanged)", wp.PositionSeconds) + if wp.PositionSeconds != 600 { + t.Errorf("PositionSeconds after backward = %v, want 600 (last write wins)", wp.PositionSeconds) } // SetProgress backward: should work (unconditional)