fix(playback): last-write-wins progress and DV P7 RPU strip on remux (#334)

* fix(playback): last-write-wins progress and DV P7 RPU strip on remux

Progress: UpdateProgress (the live playback-session path) clamped
position_seconds to GREATEST(new, old), so a deliberate backward seek
could never persist — "rewind and stop" resumed at the stale later
position on every client. Position is now last-write-wins, matching the
/sync/progress path that was always unconditional. The completed latch
and rewatch re-entry semantics are unchanged.

Remux: profile 7 Dolby Vision remuxes drop the enhancement-layer track
(-map 0:v:0 keeps only the base layer) but previously left the dangling
dual-layer RPUs on the BL — broken metadata that a DV-honoring display
can mis-render. Remuxes of P7 files now strip DV RPUs via the dovi_rpu
bitstream filter, yielding a clean HDR10 stream (the same fallback
presentation the Apple client's P7 HDR10 toggle produces). Profile 8
RPUs are kept: the BL is self-contained and DV clients render it.
Adds MediaFile.PrimaryDVProfile() and threads the profile through
ServeRemux callers; the proxy path (no track metadata in claims) keeps
prior behavior.

True P7->8.1 DV conversion needs dovi_tool alongside FFmpeg (the
dovi_rpu bsf only strips/recompresses); the remux plumbing now carries
the DV profile so that can slot in later.

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

* fix(userdb): apply last-write-wins progress to the SQLite backend too

Review follow-up (P2): the LWW change only covered pgstore; the SQLite
userdb UpdateProgress kept the MAX clamp, so rewind-and-stop still
resumed at the stale later position for sqlite-backed installs. The
conflict clause now matches Postgres (position last-write-wins,
completed latch and rewatch re-entry unchanged), with a backward-seek
regression test.

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

* fix(playback): harden LWW progress and DV RPU strip from review

- Probe ffmpeg for the dovi_rpu bitstream filter once per process and fall
  back to a no-strip remux (the pre-existing behavior) when it is missing:
  on pre-7.1 ffmpeg the unknown filter aborted the process, turning every
  Dolby Vision profile 7 remux into a hard playback failure.
- Skip zero-position heartbeats in persistProgress, mirroring the stop path
  and the jellycompat report path. Under last-write-wins an early zero
  heartbeat (e.g. before the client seeks to its resume point) would wipe
  the stored resume position; GREATEST previously masked this.
- Carry the DV profile in stream token claims (dvp, omitempty) so standalone
  proxy nodes strip profile 7 RPUs the same way integrated mode does. Old
  tokens decode as 0 and keep prior behavior.

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

---------

Co-authored-by: rxwatcher <rxwatcher@users.noreply.github.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: Quick <31828688+Quick104@users.noreply.github.com>

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"
-					}
-				}
+					}				}
 			}
 		}
 	}

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)
 		}
 

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 {

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)
 	}

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"`

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

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, " "))
+		}
+	}
+}

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) {

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.

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,

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)
+	}
+}

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,

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)