Merge remote-tracking branch 'origin/main' into feat/client-diagnostics-server

This commit is contained in:
Quick104
2026-07-21 08:17:24 -04:00
41 changed files with 4754 additions and 884 deletions
+84 -10
View File
@@ -19,7 +19,15 @@ env:
jobs:
build:
runs-on: [self-hosted, linux]
strategy:
fail-fast: false
matrix:
include:
- platform: linux/amd64
runner: ubuntu-latest
- platform: linux/arm64
runner: ubuntu-24.04-arm
runs-on: ${{ matrix.runner }}
permissions:
contents: read
packages: write
@@ -28,6 +36,13 @@ jobs:
- name: Checkout
uses: actions/checkout@v5
- name: Prepare platform slug and lowercase image name
run: |
echo "PLATFORM_PAIR=${PLATFORM//\//-}" >> "$GITHUB_ENV"
echo "IMAGE_LC=${IMAGE_NAME,,}" >> "$GITHUB_ENV"
env:
PLATFORM: ${{ matrix.platform }}
- name: Set up pnpm
uses: pnpm/action-setup@v4
with:
@@ -76,9 +91,6 @@ jobs:
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
with:
name: silo-builder
keep-state: true
- name: Log in to GitHub Container Registry
uses: docker/login-action@v3
@@ -92,19 +104,81 @@ jobs:
uses: docker/metadata-action@v5
with:
images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
tags: |
type=raw,value=latest,enable={{is_default_branch}}
type=sha,prefix=,format=short
- name: Build and push
- name: Build and push by digest
id: build
uses: docker/build-push-action@v6
with:
context: .
push: true
tags: ${{ steps.meta.outputs.tags }}
platforms: ${{ matrix.platform }}
labels: ${{ steps.meta.outputs.labels }}
build-contexts: |
frontend_dist=./web/dist
build-args: |
BUILD_REVISION=${{ github.sha }}
BUILD_DIRTY=false
cache-from: type=gha,scope=docker-${{ env.PLATFORM_PAIR }}
cache-to: type=gha,scope=docker-${{ env.PLATFORM_PAIR }},mode=max
outputs: type=image,name=${{ env.REGISTRY }}/${{ env.IMAGE_LC }},push-by-digest=true,name-canonical=true,push=true
- name: Export digest
run: |
mkdir -p "${RUNNER_TEMP}/digests"
digest="${{ steps.build.outputs.digest }}"
touch "${RUNNER_TEMP}/digests/${digest#sha256:}"
- name: Upload digest
uses: actions/upload-artifact@v4
with:
name: digests-${{ env.PLATFORM_PAIR }}
path: ${{ runner.temp }}/digests/*
if-no-files-found: error
retention-days: 1
merge:
runs-on: ubuntu-latest
needs: build
permissions:
packages: write
steps:
- name: Prepare lowercase image name
run: echo "IMAGE_LC=${IMAGE_NAME,,}" >> "$GITHUB_ENV"
- name: Download digests
uses: actions/download-artifact@v4
with:
path: ${{ runner.temp }}/digests
pattern: digests-*
merge-multiple: true
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Log in to GitHub Container Registry
uses: docker/login-action@v3
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Extract metadata
id: meta
uses: docker/metadata-action@v5
with:
images: ${{ env.REGISTRY }}/${{ env.IMAGE_LC }}
tags: |
type=raw,value=latest,enable={{is_default_branch}}
type=sha,prefix=,format=short
- name: Create and push manifest list
working-directory: ${{ runner.temp }}/digests
run: |
# shellcheck disable=SC2046
docker buildx imagetools create \
$(jq -cr '.tags | map("-t " + .) | join(" ")' <<< "$DOCKER_METADATA_OUTPUT_JSON") \
$(printf "${REGISTRY}/${IMAGE_LC}@sha256:%s " *)
- name: Inspect manifest list
run: |
docker buildx imagetools inspect "${REGISTRY}/${IMAGE_LC}:${GITHUB_SHA::7}"
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+4 -4
View File
@@ -47,6 +47,8 @@ type AuthMiddleware struct {
sessionValidator SessionValidator
apiKeyValidator APIKeyValidator // nil if API keys not configured
apiKeyUserLoader APIKeyUserLoader // nil if API keys not configured
apiKeyLastUsed *auth.APIKeyLastUsedTracker
}
// NewAuthMiddleware creates a new AuthMiddleware with the given token validator
@@ -57,6 +59,7 @@ func NewAuthMiddleware(tv TokenValidator, sv SessionValidator, akv APIKeyValidat
sessionValidator: sv,
apiKeyValidator: akv,
apiKeyUserLoader: akul,
apiKeyLastUsed: auth.NewAPIKeyLastUsedTracker(akv, nil),
}
}
@@ -98,10 +101,7 @@ func (am *AuthMiddleware) RequireAuth(next http.Handler) http.Handler {
return
}
// Update last_used_at asynchronously.
go func(id int64) {
_ = am.apiKeyValidator.UpdateLastUsed(context.Background(), id)
}(apiKey.ID)
am.apiKeyLastUsed.Touch(apiKey.ID)
claims = &auth.Claims{
UserID: user.ID,
+80
View File
@@ -0,0 +1,80 @@
package auth
import (
"context"
"log/slog"
"sync"
"time"
)
const (
apiKeyLastUsedInterval = time.Minute
apiKeyLastUsedTimeout = 5 * time.Second
)
// APIKeyLastUsedUpdater persists an API key's last-used timestamp.
type APIKeyLastUsedUpdater interface {
UpdateLastUsed(ctx context.Context, id int64) error
}
// APIKeyLastUsedTracker coalesces last-used writes per key without blocking
// authentication requests. Stale entries are pruned inline once per interval,
// avoiding both unbounded historical-key retention and another goroutine.
type APIKeyLastUsedTracker struct {
updater APIKeyLastUsedUpdater
now func() time.Time
mu sync.Mutex
lastUsedAt map[int64]time.Time
nextPrune time.Time
}
// NewAPIKeyLastUsedTracker creates a tracker. A nil clock uses time.Now.
func NewAPIKeyLastUsedTracker(updater APIKeyLastUsedUpdater, now func() time.Time) *APIKeyLastUsedTracker {
if now == nil {
now = time.Now
}
return &APIKeyLastUsedTracker{
updater: updater,
now: now,
lastUsedAt: make(map[int64]time.Time),
}
}
// Touch records key usage asynchronously, at most once per interval per key.
func (t *APIKeyLastUsedTracker) Touch(id int64) {
if t == nil || t.updater == nil || !t.shouldUpdate(id) {
return
}
go func() {
ctx, cancel := context.WithTimeout(context.Background(), apiKeyLastUsedTimeout)
defer cancel()
if err := t.updater.UpdateLastUsed(ctx, id); err != nil {
slog.DebugContext(ctx, "api key last-used update failed", "component", "auth", "id", id, "error", err)
}
}()
}
func (t *APIKeyLastUsedTracker) shouldUpdate(id int64) bool {
now := t.now()
t.mu.Lock()
defer t.mu.Unlock()
if t.nextPrune.IsZero() || !now.Before(t.nextPrune) {
cutoff := now.Add(-apiKeyLastUsedInterval)
for keyID, lastUsed := range t.lastUsedAt {
if !lastUsed.After(cutoff) {
delete(t.lastUsedAt, keyID)
}
}
t.nextPrune = now.Add(apiKeyLastUsedInterval)
}
if lastUsed, ok := t.lastUsedAt[id]; ok && now.Sub(lastUsed) < apiKeyLastUsedInterval {
return false
}
t.lastUsedAt[id] = now
return true
}
+82
View File
@@ -0,0 +1,82 @@
package auth
import (
"context"
"testing"
"time"
)
type recordingLastUsedUpdater struct {
calls chan lastUsedCall
}
type lastUsedCall struct {
id int64
deadline time.Time
}
func (u *recordingLastUsedUpdater) UpdateLastUsed(ctx context.Context, id int64) error {
deadline, _ := ctx.Deadline()
u.calls <- lastUsedCall{id: id, deadline: deadline}
return nil
}
func TestAPIKeyLastUsedTrackerThrottlesPerKey(t *testing.T) {
now := time.Date(2026, time.July, 20, 12, 0, 0, 0, time.UTC)
tracker := NewAPIKeyLastUsedTracker(nil, func() time.Time { return now })
if !tracker.shouldUpdate(1) {
t.Fatal("first use should update")
}
if tracker.shouldUpdate(1) {
t.Fatal("repeat use inside the interval should be throttled")
}
if !tracker.shouldUpdate(2) {
t.Fatal("a different key should update independently")
}
now = now.Add(apiKeyLastUsedInterval)
if !tracker.shouldUpdate(1) {
t.Fatal("use at the next interval should update")
}
}
func TestAPIKeyLastUsedTrackerPrunesExpiredKeys(t *testing.T) {
now := time.Date(2026, time.July, 20, 12, 0, 0, 0, time.UTC)
tracker := NewAPIKeyLastUsedTracker(nil, func() time.Time { return now })
tracker.shouldUpdate(1)
tracker.shouldUpdate(2)
now = now.Add(apiKeyLastUsedInterval)
tracker.shouldUpdate(3)
if len(tracker.lastUsedAt) != 1 {
t.Fatalf("retained entries = %d, want 1", len(tracker.lastUsedAt))
}
if _, ok := tracker.lastUsedAt[3]; !ok {
t.Fatal("current key was not retained")
}
}
func TestAPIKeyLastUsedTrackerAddsWriteDeadline(t *testing.T) {
updater := &recordingLastUsedUpdater{calls: make(chan lastUsedCall, 1)}
tracker := NewAPIKeyLastUsedTracker(updater, nil)
tracker.Touch(42)
select {
case call := <-updater.calls:
if call.id != 42 {
t.Fatalf("key id = %d, want 42", call.id)
}
if call.deadline.IsZero() {
t.Fatal("last-used update context has no deadline")
}
remaining := time.Until(call.deadline)
if remaining <= 0 || remaining > apiKeyLastUsedTimeout {
t.Fatalf("deadline after %s, want within (0, %s]", remaining, apiKeyLastUsedTimeout)
}
case <-time.After(time.Second):
t.Fatal("timed out waiting for last-used update")
}
}
+10 -35
View File
@@ -9,6 +9,7 @@ import (
"sync"
"time"
"github.com/Silo-Server/silo-server/internal/auth"
"github.com/Silo-Server/silo-server/internal/models"
"github.com/Silo-Server/silo-server/internal/userstore"
)
@@ -36,8 +37,7 @@ type AdminAPIKeyAuthenticator struct {
// authorization decision, which is re-checked on every request.
profiles *apiKeyProfileCache
lastUsedMu sync.Mutex
lastUsedAt map[int64]time.Time
apiKeyLastUsed *auth.APIKeyLastUsedTracker
}
type adminAPIKeyAuthResult struct {
@@ -61,12 +61,12 @@ func NewAdminAPIKeyAuthenticator(keys apiKeyValidator, users apiKeyUserLoader, p
now = time.Now
}
return &AdminAPIKeyAuthenticator{
keys: keys,
users: users,
provider: provider,
now: now,
profiles: newAPIKeyProfileCache(),
lastUsedAt: make(map[int64]time.Time),
keys: keys,
users: users,
provider: provider,
now: now,
profiles: newAPIKeyProfileCache(),
apiKeyLastUsed: auth.NewAPIKeyLastUsedTracker(keys, now),
}
}
@@ -138,7 +138,7 @@ func (a *AdminAPIKeyAuthenticator) authenticate(r *http.Request) adminAPIKeyAuth
if !res.ok {
return res
}
a.touchLastUsed(apiKey.ID)
a.apiKeyLastUsed.Touch(apiKey.ID)
return adminAPIKeyAuthResult{
ctx: context.WithValue(r.Context(), adminAPIKeyKey, true),
status: http.StatusOK,
@@ -182,28 +182,6 @@ func (a *AdminAPIKeyAuthenticator) validate(ctx context.Context, token string) (
return apiKey, user, adminAPIKeyAuthResult{ok: true}
}
// touchLastUsed records key usage without blocking the request, throttled to at
// most once per apiKeyLastUsedInterval per key so per-request validation on the
// hot path (e.g. HLS segments) does not issue a DB write each time.
func (a *AdminAPIKeyAuthenticator) touchLastUsed(id int64) {
now := a.now()
a.lastUsedMu.Lock()
if last, ok := a.lastUsedAt[id]; ok && now.Sub(last) < apiKeyLastUsedInterval {
a.lastUsedMu.Unlock()
return
}
a.lastUsedAt[id] = now
a.lastUsedMu.Unlock()
go func(id int64) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := a.keys.UpdateLastUsed(ctx, id); err != nil {
slog.Debug("jellycompat api key last-used update failed", "id", id, "error", err)
}
}(id)
}
// resolveSession authenticates an sa_ API key and returns a compat session
// synthesized for the key user's primary profile, so ordinary browse/stream
// handlers (which read SessionFromContext) work unchanged for an API key.
@@ -224,7 +202,7 @@ func (a *AdminAPIKeyAuthenticator) resolveSession(ctx context.Context, token str
if !res.ok {
return nil, res, true
}
a.touchLastUsed(apiKey.ID)
a.apiKeyLastUsed.Touch(apiKey.ID)
profile, err := a.primaryProfile(ctx, user.ID)
if err != nil {
@@ -298,9 +276,6 @@ const (
// reassigned/renamed primary profile is picked up, never authorization
// (the key and user are re-validated on every request).
apiKeyProfileCacheTTL = 5 * time.Minute
// apiKeyLastUsedInterval throttles api_keys.last_used_at writes so that
// per-request validation on the hot path does not write each time.
apiKeyLastUsedInterval = time.Minute
)
type apiKeyProfileCacheEntry struct {
+149
View File
@@ -0,0 +1,149 @@
package playback
import (
"bytes"
"context"
"encoding/json"
"fmt"
"math"
"os/exec"
"strconv"
"strings"
"time"
"golang.org/x/sync/singleflight"
)
const (
maxConcurrentCopySeekProbes = 4
copySeekProbeTimeout = 15 * time.Second
)
var (
copySeekProbeGroup singleflight.Group
copySeekProbeSlots = make(chan struct{}, maxConcurrentCopySeekProbes)
)
type copySeekProbePacket struct {
PTSSeconds string `json:"pts_time"`
DTSSeconds string `json:"dts_time"`
Flags string `json:"flags"`
}
type copySeekProbeOutput struct {
Packets []copySeekProbePacket `json:"packets"`
}
type copySeekAnchor struct {
seconds float64
segment int
}
// ResolveCopySeekAnchor returns the keyframe timestamp FFmpeg's input seek will
// actually use for a copy-video restart. FFmpeg cannot discard the pre-roll
// between that keyframe and requestedSeekSeconds while preserving -c:v copy,
// so callers need both timestamps: requestedSeekSeconds remains the -ss input,
// while the returned anchor defines the stream's real timeline origin.
//
// The tiny read interval is intentional. ffprobe asks the demuxer to seek to
// requestedSeekSeconds, which lands on the same preceding key packet FFmpeg
// uses, then emits only the packets at that seek point instead of scanning the
// media from the beginning.
func ResolveCopySeekAnchor(
ctx context.Context,
ffmpegPath string,
inputPath string,
requestedSeekSeconds float64,
segmentDuration int,
) (float64, int, error) {
if requestedSeekSeconds <= 0 {
return 0, 0, nil
}
if strings.TrimSpace(inputPath) == "" {
return 0, 0, fmt.Errorf("resolve copy seek anchor: empty input path")
}
if segmentDuration <= 0 {
segmentDuration = DefaultSegmentDuration
}
key := strings.Join([]string{
ffprobePathFromFFmpeg(ffmpegPath),
inputPath,
strconv.FormatFloat(requestedSeekSeconds, 'f', 6, 64),
strconv.Itoa(segmentDuration),
}, "\x00")
resultCh := copySeekProbeGroup.DoChan(key, func() (any, error) {
probeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), copySeekProbeTimeout)
defer cancel()
select {
case copySeekProbeSlots <- struct{}{}:
defer func() { <-copySeekProbeSlots }()
case <-probeCtx.Done():
return copySeekAnchor{}, probeCtx.Err()
}
seconds, segment, err := resolveCopySeekAnchor(probeCtx, ffmpegPath, inputPath, requestedSeekSeconds, segmentDuration)
return copySeekAnchor{seconds: seconds, segment: segment}, err
})
select {
case <-ctx.Done():
return 0, 0, ctx.Err()
case result := <-resultCh:
if result.Err != nil {
return 0, 0, result.Err
}
anchor := result.Val.(copySeekAnchor)
return anchor.seconds, anchor.segment, nil
}
}
func resolveCopySeekAnchor(
ctx context.Context,
ffmpegPath string,
inputPath string,
requestedSeekSeconds float64,
segmentDuration int,
) (float64, int, error) {
interval := strconv.FormatFloat(requestedSeekSeconds, 'f', 6, 64) + "%+0.001"
cmd := exec.CommandContext(ctx, ffprobePathFromFFmpeg(ffmpegPath),
"-v", "error",
"-select_streams", "v:0",
"-read_intervals", interval,
"-show_entries", "packet=pts_time,dts_time,flags",
"-of", "json",
inputPath,
)
var stdout bytes.Buffer
stderr := newBoundedTailBuffer(stderrTailMaxBytes)
cmd.Stdout = &stdout
cmd.Stderr = stderr
if err := cmd.Run(); err != nil {
if tail := truncateStderr(stderr.String()); tail != "" {
return 0, 0, fmt.Errorf("resolve copy seek anchor: ffprobe failed: %w (stderr: %s)", err, tail)
}
return 0, 0, fmt.Errorf("resolve copy seek anchor: ffprobe failed: %w", err)
}
var output copySeekProbeOutput
if err := json.Unmarshal(stdout.Bytes(), &output); err != nil {
return 0, 0, fmt.Errorf("resolve copy seek anchor: decode ffprobe output: %w", err)
}
for _, packet := range output.Packets {
if !strings.Contains(packet.Flags, "K") {
continue
}
timestamp := packet.PTSSeconds
if timestamp == "" || strings.EqualFold(timestamp, "N/A") {
timestamp = packet.DTSSeconds
}
anchor, err := strconv.ParseFloat(timestamp, 64)
if err != nil || math.IsNaN(anchor) || math.IsInf(anchor, 0) {
continue
}
anchor = math.Max(0, anchor)
return anchor, int(anchor / float64(segmentDuration)), nil
}
return 0, 0, fmt.Errorf("resolve copy seek anchor: ffprobe returned no key packet")
}
+174
View File
@@ -0,0 +1,174 @@
package playback
import (
"context"
"fmt"
"math"
"os"
"os/exec"
"path/filepath"
"strings"
"sync"
"testing"
"time"
)
func TestResolveCopySeekAnchorUsesKeyPacketTimestamp(t *testing.T) {
dir := t.TempDir()
ffmpegPath := filepath.Join(dir, "ffmpeg")
ffprobePath := filepath.Join(dir, "ffprobe")
if err := os.WriteFile(ffmpegPath, []byte("#!/bin/sh\nexit 0\n"), 0o755); err != nil {
t.Fatalf("write fake ffmpeg: %v", err)
}
probe := `#!/bin/sh
printf '%s' '{"packets":[{"pts_time":"N/A","dts_time":"14.500000","flags":"K__"}]}'
`
if err := os.WriteFile(ffprobePath, []byte(probe), 0o755); err != nil {
t.Fatalf("write fake ffprobe: %v", err)
}
anchor, segment, err := ResolveCopySeekAnchor(context.Background(), ffmpegPath, "/media/movie.mkv", 18.261, 2)
if err != nil {
t.Fatalf("ResolveCopySeekAnchor: %v", err)
}
if anchor != 14.5 || segment != 7 {
t.Fatalf("resolved anchor = %v, segment = %d; want 14.5, 7", anchor, segment)
}
}
func TestResolveCopySeekAnchorCoalescesMatchingConcurrentProbes(t *testing.T) {
dir := t.TempDir()
ffmpegPath := filepath.Join(dir, "ffmpeg")
ffprobePath := filepath.Join(dir, "ffprobe")
countPath := filepath.Join(dir, "probe-count")
if err := os.WriteFile(ffmpegPath, []byte("#!/bin/sh\nexit 0\n"), 0o755); err != nil {
t.Fatalf("write fake ffmpeg: %v", err)
}
probe := "#!/bin/sh\n" +
"printf x >> \"" + countPath + "\"\n" +
"sleep 0.1\n" +
"printf '%s' '{\"packets\":[{\"pts_time\":\"16.000000\",\"flags\":\"K__\"}]}'\n"
if err := os.WriteFile(ffprobePath, []byte(probe), 0o755); err != nil {
t.Fatalf("write fake ffprobe: %v", err)
}
start := make(chan struct{})
errs := make(chan error, 2)
var ready sync.WaitGroup
ready.Add(2)
for range 2 {
go func() {
ready.Done()
<-start
anchor, segment, err := ResolveCopySeekAnchor(context.Background(), ffmpegPath, "/media/movie.mkv", 18, 2)
if err == nil && (anchor != 16 || segment != 8) {
err = fmt.Errorf("resolved anchor = %v, segment = %d", anchor, segment)
}
errs <- err
}()
}
ready.Wait()
close(start)
for range 2 {
if err := <-errs; err != nil {
t.Fatal(err)
}
}
count, err := os.ReadFile(countPath)
if err != nil {
t.Fatalf("read probe count: %v", err)
}
if string(count) != "x" {
t.Fatalf("ffprobe executions = %d, want 1", len(count))
}
}
func TestResolveCopySeekAnchorMatchesRealLongGOPHEVC(t *testing.T) {
if testing.Short() {
t.Skip("real FFmpeg integration test")
}
ffmpegPath, err := exec.LookPath("ffmpeg")
if err != nil {
t.Skip("ffmpeg is not installed")
}
if _, err := exec.LookPath(ffprobePathFromFFmpeg(ffmpegPath)); err != nil {
t.Skip("ffprobe is not installed beside ffmpeg")
}
encoders, err := exec.Command(ffmpegPath, "-hide_banner", "-encoders").CombinedOutput()
if err != nil || !strings.Contains(string(encoders), "libx265") {
t.Skip("ffmpeg does not provide libx265")
}
sourcePath := filepath.Join(t.TempDir(), "long-gop-hevc.mkv")
encodeCtx, cancelEncode := context.WithTimeout(context.Background(), 20*time.Second)
defer cancelEncode()
encode := exec.CommandContext(encodeCtx, ffmpegPath,
"-hide_banner", "-loglevel", "error",
"-f", "lavfi", "-i", "testsrc2=size=320x180:rate=24",
"-t", "22",
"-c:v", "libx265", "-preset", "ultrafast",
"-x265-params", "keyint=240:min-keyint=240:scenecut=0:log-level=error:pools=1:frame-threads=1",
"-an", "-y", sourcePath,
)
if output, err := encode.CombinedOutput(); err != nil {
t.Fatalf("generate long-GOP HEVC fixture: %v\n%s", err, output)
}
probeCtx, cancelProbe := context.WithTimeout(context.Background(), 5*time.Second)
defer cancelProbe()
anchor, segment, err := ResolveCopySeekAnchor(probeCtx, ffmpegPath, sourcePath, 18.261, 2)
if err != nil {
t.Fatalf("ResolveCopySeekAnchor: %v", err)
}
if math.Abs(anchor-10) > 0.001 || segment != 5 {
t.Fatalf("resolved anchor = %v, segment = %d; want 10, 5", anchor, segment)
}
outputDir := filepath.Join(t.TempDir(), "hls")
if err := os.MkdirAll(outputDir, 0o755); err != nil {
t.Fatalf("create HLS output: %v", err)
}
opts := TranscodeOpts{
InputPath: sourcePath,
OutputDir: outputDir,
SessionID: "copy-anchor-integration",
SourceVideoCodec: "hevc",
SeekSeconds: 18.261,
StreamOriginSeconds: anchor,
CopySeekAnchorResolved: true,
TargetCodecVideo: "copy",
TargetCodecAudio: "copy",
SegmentDuration: 2,
StartSegmentNumber: segment,
FFmpegPath: ffmpegPath,
}
packageCtx, cancelPackage := context.WithTimeout(context.Background(), 10*time.Second)
defer cancelPackage()
packageHLS := exec.CommandContext(packageCtx, ffmpegPath, buildFFmpegArgs(opts)...)
if output, err := packageHLS.CombinedOutput(); err != nil {
t.Fatalf("package copy HLS: %v\n%s", err, output)
}
manifestPath := filepath.Join(outputDir, "stream.m3u8")
manifest, err := os.ReadFile(manifestPath)
if err != nil {
t.Fatalf("read copy HLS manifest: %v", err)
}
if !strings.Contains(string(manifest), "#EXT-X-MEDIA-SEQUENCE:5") {
t.Fatalf("copy HLS media sequence does not use resolved anchor:\n%s", manifest)
}
firstFrame, err := exec.Command(ffprobePathFromFFmpeg(ffmpegPath),
"-v", "error",
"-select_streams", "v:0",
"-read_intervals", "%+0.1",
"-show_entries", "frame=best_effort_timestamp_time,key_frame",
"-of", "csv=p=0",
manifestPath,
).CombinedOutput()
if err != nil {
t.Fatalf("probe copy HLS first frame: %v\n%s", err, firstFrame)
}
if !strings.Contains(string(firstFrame), "10.000000") {
t.Fatalf("copy HLS first frame = %q, want resolved 10-second keyframe", firstFrame)
}
}
+134 -124
View File
@@ -44,25 +44,27 @@ type RecipeCard struct {
// Encode parameters — mirror of the byte-affecting TranscodeOpts fields.
// Unused (zero) for direct/remux cards, which carry no segment-based encode.
InputPath string `json:"input_path"`
OutputSubdir string `json:"output_subdir,omitempty"`
SourceVideoCodec string `json:"source_video_codec,omitempty"`
VideoBitstreamFilter string `json:"video_bitstream_filter,omitempty"`
SeekSeconds float64 `json:"seek_seconds"`
TargetResolution string `json:"target_resolution,omitempty"`
TargetCodecVideo string `json:"target_codec_video,omitempty"`
TargetCodecAudio string `json:"target_codec_audio,omitempty"`
SegmentDuration int `json:"segment_duration"`
StartSegmentNumber int `json:"start_segment_number"`
HWAccel string `json:"hw_accel,omitempty"`
HWDevice string `json:"hw_device,omitempty"`
SubtitleTrackIndex int `json:"subtitle_track_index"`
SubtitleBurnIn bool `json:"subtitle_burn_in,omitempty"`
SubtitleCodec string `json:"subtitle_codec,omitempty"`
AudioTrackIndex int `json:"audio_track_index"`
TargetBitrateKbps int `json:"target_bitrate_kbps,omitempty"`
TotalDuration float64 `json:"total_duration"`
FastStart bool `json:"fast_start,omitempty"`
InputPath string `json:"input_path"`
OutputSubdir string `json:"output_subdir,omitempty"`
SourceVideoCodec string `json:"source_video_codec,omitempty"`
VideoBitstreamFilter string `json:"video_bitstream_filter,omitempty"`
SeekSeconds float64 `json:"seek_seconds"`
StreamOriginSeconds float64 `json:"stream_origin_seconds,omitempty"`
CopySeekAnchorResolved bool `json:"copy_seek_anchor_resolved,omitempty"`
TargetResolution string `json:"target_resolution,omitempty"`
TargetCodecVideo string `json:"target_codec_video,omitempty"`
TargetCodecAudio string `json:"target_codec_audio,omitempty"`
SegmentDuration int `json:"segment_duration"`
StartSegmentNumber int `json:"start_segment_number"`
HWAccel string `json:"hw_accel,omitempty"`
HWDevice string `json:"hw_device,omitempty"`
SubtitleTrackIndex int `json:"subtitle_track_index"`
SubtitleBurnIn bool `json:"subtitle_burn_in,omitempty"`
SubtitleCodec string `json:"subtitle_codec,omitempty"`
AudioTrackIndex int `json:"audio_track_index"`
TargetBitrateKbps int `json:"target_bitrate_kbps,omitempty"`
TotalDuration float64 `json:"total_duration"`
FastStart bool `json:"fast_start,omitempty"`
}
// NewRecipeCard builds a RecipeCard from the durable identity fields plus the
@@ -72,33 +74,35 @@ type RecipeCard struct {
// operator's config change applies to reconstructed sessions too.
func NewRecipeCard(userID int, profileID string, mediaFileID int, transcodeNodeURL string, opts TranscodeOpts) RecipeCard {
return RecipeCard{
SessionID: opts.SessionID,
UserID: userID,
ProfileID: profileID,
MediaFileID: mediaFileID,
TranscodeNodeURL: transcodeNodeURL,
TranscodeTransportID: opts.TranscodeTransportID,
PlayMethod: PlayTranscode,
TranscodeAudio: TranscodesAudio(opts.TargetCodecAudio),
InputPath: opts.InputPath,
OutputSubdir: opts.OutputSubdir,
SourceVideoCodec: opts.SourceVideoCodec,
VideoBitstreamFilter: opts.VideoBitstreamFilter,
SeekSeconds: opts.SeekSeconds,
TargetResolution: opts.TargetResolution,
TargetCodecVideo: opts.TargetCodecVideo,
TargetCodecAudio: opts.TargetCodecAudio,
SegmentDuration: opts.SegmentDuration,
StartSegmentNumber: opts.StartSegmentNumber,
HWAccel: opts.HWAccel,
HWDevice: opts.HWDevice,
SubtitleTrackIndex: opts.SubtitleTrackIndex,
SubtitleBurnIn: opts.SubtitleBurnIn,
SubtitleCodec: opts.SubtitleCodec,
AudioTrackIndex: opts.AudioTrackIndex,
TargetBitrateKbps: opts.TargetBitrateKbps,
TotalDuration: opts.TotalDuration,
FastStart: opts.FastStart,
SessionID: opts.SessionID,
UserID: userID,
ProfileID: profileID,
MediaFileID: mediaFileID,
TranscodeNodeURL: transcodeNodeURL,
TranscodeTransportID: opts.TranscodeTransportID,
PlayMethod: PlayTranscode,
TranscodeAudio: TranscodesAudio(opts.TargetCodecAudio),
InputPath: opts.InputPath,
OutputSubdir: opts.OutputSubdir,
SourceVideoCodec: opts.SourceVideoCodec,
VideoBitstreamFilter: opts.VideoBitstreamFilter,
SeekSeconds: opts.SeekSeconds,
StreamOriginSeconds: opts.StreamOriginSeconds,
CopySeekAnchorResolved: opts.CopySeekAnchorResolved,
TargetResolution: opts.TargetResolution,
TargetCodecVideo: opts.TargetCodecVideo,
TargetCodecAudio: opts.TargetCodecAudio,
SegmentDuration: opts.SegmentDuration,
StartSegmentNumber: opts.StartSegmentNumber,
HWAccel: opts.HWAccel,
HWDevice: opts.HWDevice,
SubtitleTrackIndex: opts.SubtitleTrackIndex,
SubtitleBurnIn: opts.SubtitleBurnIn,
SubtitleCodec: opts.SubtitleCodec,
AudioTrackIndex: opts.AudioTrackIndex,
TargetBitrateKbps: opts.TargetBitrateKbps,
TotalDuration: opts.TotalDuration,
FastStart: opts.FastStart,
}
}
@@ -141,32 +145,34 @@ func NewRemuxRecipeCard(sessionID string, userID int, profileID string, mediaFil
// they are environment-specific and not pinned in the card.
func (c RecipeCard) TranscodeOpts(outputDir, ffmpegPath string, logSink FFmpegLogSink) TranscodeOpts {
return TranscodeOpts{
InputPath: c.InputPath,
OutputSubdir: c.OutputSubdir,
OutputDir: outputDir,
SessionID: c.SessionID,
TranscodeTransportID: c.TranscodeTransportID,
SourceVideoCodec: c.SourceVideoCodec,
VideoBitstreamFilter: c.VideoBitstreamFilter,
SeekSeconds: c.SeekSeconds,
TargetResolution: c.TargetResolution,
TargetCodecVideo: c.TargetCodecVideo,
TargetCodecAudio: c.TargetCodecAudio,
SegmentDuration: c.SegmentDuration,
StartSegmentNumber: c.StartSegmentNumber,
FFmpegPath: ffmpegPath,
HWAccel: c.HWAccel,
HWDevice: c.HWDevice,
SubtitleTrackIndex: c.SubtitleTrackIndex,
SubtitleBurnIn: c.SubtitleBurnIn,
SubtitleCodec: c.SubtitleCodec,
AudioTrackIndex: c.AudioTrackIndex,
TargetBitrateKbps: c.TargetBitrateKbps,
TotalDuration: c.TotalDuration,
FastStart: c.FastStart,
NodeType: "integrated",
ExecutionMode: "integrated",
FFmpegLogSink: logSink,
InputPath: c.InputPath,
OutputSubdir: c.OutputSubdir,
OutputDir: outputDir,
SessionID: c.SessionID,
TranscodeTransportID: c.TranscodeTransportID,
SourceVideoCodec: c.SourceVideoCodec,
VideoBitstreamFilter: c.VideoBitstreamFilter,
SeekSeconds: c.SeekSeconds,
StreamOriginSeconds: c.StreamOriginSeconds,
CopySeekAnchorResolved: c.CopySeekAnchorResolved,
TargetResolution: c.TargetResolution,
TargetCodecVideo: c.TargetCodecVideo,
TargetCodecAudio: c.TargetCodecAudio,
SegmentDuration: c.SegmentDuration,
StartSegmentNumber: c.StartSegmentNumber,
FFmpegPath: ffmpegPath,
HWAccel: c.HWAccel,
HWDevice: c.HWDevice,
SubtitleTrackIndex: c.SubtitleTrackIndex,
SubtitleBurnIn: c.SubtitleBurnIn,
SubtitleCodec: c.SubtitleCodec,
AudioTrackIndex: c.AudioTrackIndex,
TargetBitrateKbps: c.TargetBitrateKbps,
TotalDuration: c.TotalDuration,
FastStart: c.FastStart,
NodeType: "integrated",
ExecutionMode: "integrated",
FFmpegLogSink: logSink,
}
}
@@ -185,32 +191,34 @@ const MaxTokenTTL = 24 * time.Hour
// change applies to reconstructed sessions too.
func (c RecipeCard) ToClaims() streamtoken.Claims {
return streamtoken.Claims{
SessionID: c.SessionID,
MediaPath: c.InputPath,
OutputSubdir: c.OutputSubdir,
PlayMethod: string(c.PlayMethod),
TranscodeAudio: c.TranscodeAudio,
RemuxDVMode: string(c.RemuxDVMode),
TranscodeNode: c.TranscodeNodeURL,
TranscodeTransportID: c.TranscodeTransportID,
TargetCodec: c.TargetCodecVideo,
TargetRes: c.TargetResolution,
AudioTrackIndex: c.AudioTrackIndex,
UserID: c.UserID,
ProfileID: c.ProfileID,
MediaFileID: c.MediaFileID,
SourceVideoCodec: c.SourceVideoCodec,
VideoBitstreamFilter: c.VideoBitstreamFilter,
SeekSeconds: c.SeekSeconds,
SegmentDuration: c.SegmentDuration,
StartSegmentNumber: c.StartSegmentNumber,
SubtitleTrackIndex: c.SubtitleTrackIndex,
SubtitleBurnIn: c.SubtitleBurnIn,
SubtitleCodec: c.SubtitleCodec,
TargetBitrateKbps: c.TargetBitrateKbps,
TotalDuration: c.TotalDuration,
FastStart: c.FastStart,
TargetCodecAudio: c.TargetCodecAudio,
SessionID: c.SessionID,
MediaPath: c.InputPath,
OutputSubdir: c.OutputSubdir,
PlayMethod: string(c.PlayMethod),
TranscodeAudio: c.TranscodeAudio,
RemuxDVMode: string(c.RemuxDVMode),
TranscodeNode: c.TranscodeNodeURL,
TranscodeTransportID: c.TranscodeTransportID,
TargetCodec: c.TargetCodecVideo,
TargetRes: c.TargetResolution,
AudioTrackIndex: c.AudioTrackIndex,
UserID: c.UserID,
ProfileID: c.ProfileID,
MediaFileID: c.MediaFileID,
SourceVideoCodec: c.SourceVideoCodec,
VideoBitstreamFilter: c.VideoBitstreamFilter,
SeekSeconds: c.SeekSeconds,
StreamOriginSeconds: c.StreamOriginSeconds,
CopySeekAnchorResolved: c.CopySeekAnchorResolved,
SegmentDuration: c.SegmentDuration,
StartSegmentNumber: c.StartSegmentNumber,
SubtitleTrackIndex: c.SubtitleTrackIndex,
SubtitleBurnIn: c.SubtitleBurnIn,
SubtitleCodec: c.SubtitleCodec,
TargetBitrateKbps: c.TargetBitrateKbps,
TotalDuration: c.TotalDuration,
FastStart: c.FastStart,
TargetCodecAudio: c.TargetCodecAudio,
}
}
@@ -227,31 +235,33 @@ func RecipeCardFromClaims(c *streamtoken.Claims) RecipeCard {
method = PlayTranscode
}
return RecipeCard{
SessionID: c.SessionID,
UserID: c.UserID,
ProfileID: c.ProfileID,
MediaFileID: c.MediaFileID,
TranscodeNodeURL: c.TranscodeNode,
TranscodeTransportID: c.TranscodeTransportID,
PlayMethod: method,
TranscodeAudio: c.TranscodeAudio,
RemuxDVMode: RemuxDVMode(c.RemuxDVMode),
InputPath: c.MediaPath,
OutputSubdir: c.OutputSubdir,
SourceVideoCodec: c.SourceVideoCodec,
VideoBitstreamFilter: c.VideoBitstreamFilter,
SeekSeconds: c.SeekSeconds,
TargetResolution: c.TargetRes,
TargetCodecVideo: c.TargetCodec,
TargetCodecAudio: c.TargetCodecAudio,
SegmentDuration: c.SegmentDuration,
StartSegmentNumber: c.StartSegmentNumber,
SubtitleTrackIndex: c.SubtitleTrackIndex,
SubtitleBurnIn: c.SubtitleBurnIn,
SubtitleCodec: c.SubtitleCodec,
AudioTrackIndex: c.AudioTrackIndex,
TargetBitrateKbps: c.TargetBitrateKbps,
TotalDuration: c.TotalDuration,
FastStart: c.FastStart,
SessionID: c.SessionID,
UserID: c.UserID,
ProfileID: c.ProfileID,
MediaFileID: c.MediaFileID,
TranscodeNodeURL: c.TranscodeNode,
TranscodeTransportID: c.TranscodeTransportID,
PlayMethod: method,
TranscodeAudio: c.TranscodeAudio,
RemuxDVMode: RemuxDVMode(c.RemuxDVMode),
InputPath: c.MediaPath,
OutputSubdir: c.OutputSubdir,
SourceVideoCodec: c.SourceVideoCodec,
VideoBitstreamFilter: c.VideoBitstreamFilter,
SeekSeconds: c.SeekSeconds,
StreamOriginSeconds: c.StreamOriginSeconds,
CopySeekAnchorResolved: c.CopySeekAnchorResolved,
TargetResolution: c.TargetRes,
TargetCodecVideo: c.TargetCodec,
TargetCodecAudio: c.TargetCodecAudio,
SegmentDuration: c.SegmentDuration,
StartSegmentNumber: c.StartSegmentNumber,
SubtitleTrackIndex: c.SubtitleTrackIndex,
SubtitleBurnIn: c.SubtitleBurnIn,
SubtitleCodec: c.SubtitleCodec,
AudioTrackIndex: c.AudioTrackIndex,
TargetBitrateKbps: c.TargetBitrateKbps,
TotalDuration: c.TotalDuration,
FastStart: c.FastStart,
}
}
+46 -38
View File
@@ -9,26 +9,28 @@ import (
func TestRecipeCardRoundTripOpts(t *testing.T) {
opts := TranscodeOpts{
InputPath: "/media/movie.mkv",
OutputDir: "/tmp/silo-transcode/abc",
SessionID: "abc",
SourceVideoCodec: "hevc",
VideoBitstreamFilter: "dovi_rpu=strip=1",
SeekSeconds: 900,
TargetResolution: "1080p",
TargetCodecVideo: "h264",
TargetCodecAudio: "aac",
SegmentDuration: 2,
StartSegmentNumber: 450,
HWAccel: "qsv",
HWDevice: "/dev/dri/renderD128",
SubtitleTrackIndex: 3,
SubtitleBurnIn: true,
SubtitleCodec: "hdmv_pgs_subtitle",
AudioTrackIndex: 1,
TargetBitrateKbps: 8000,
TotalDuration: 7200,
FastStart: true,
InputPath: "/media/movie.mkv",
OutputDir: "/tmp/silo-transcode/abc",
SessionID: "abc",
SourceVideoCodec: "hevc",
VideoBitstreamFilter: "dovi_rpu=strip=1",
SeekSeconds: 900,
StreamOriginSeconds: 896,
CopySeekAnchorResolved: true,
TargetResolution: "1080p",
TargetCodecVideo: "h264",
TargetCodecAudio: "aac",
SegmentDuration: 2,
StartSegmentNumber: 450,
HWAccel: "qsv",
HWDevice: "/dev/dri/renderD128",
SubtitleTrackIndex: 3,
SubtitleBurnIn: true,
SubtitleCodec: "hdmv_pgs_subtitle",
AudioTrackIndex: 1,
TargetBitrateKbps: 8000,
TotalDuration: 7200,
FastStart: true,
}
card := NewRecipeCard(42, "profile-1", 77, "", opts)
@@ -44,6 +46,9 @@ func TestRecipeCardRoundTripOpts(t *testing.T) {
if got.SeekSeconds != 900 {
t.Errorf("SeekSeconds = %v, want 900", got.SeekSeconds)
}
if got.StreamOriginSeconds != 896 || !got.CopySeekAnchorResolved {
t.Errorf("copy seek anchor lost: origin=%v resolved=%v", got.StreamOriginSeconds, got.CopySeekAnchorResolved)
}
if !got.SubtitleBurnIn {
t.Errorf("SubtitleBurnIn lost in round trip")
}
@@ -196,23 +201,25 @@ func TestRecipeCardLegacyDecodeHasEmptyPlayMethod(t *testing.T) {
// not asserted here.
func TestRecipeCardClaimsRoundTrip(t *testing.T) {
card := NewRecipeCard(42, "profile-1", 77, "http://node:9000", TranscodeOpts{
InputPath: "/media/movie.mkv",
SessionID: "abc",
SourceVideoCodec: "hevc",
VideoBitstreamFilter: "dovi_rpu=strip=1",
SeekSeconds: 900,
TargetResolution: "1080p",
TargetCodecVideo: "h264",
TargetCodecAudio: "aac",
SegmentDuration: 2,
StartSegmentNumber: 450,
SubtitleTrackIndex: 3,
SubtitleBurnIn: true,
SubtitleCodec: "hdmv_pgs_subtitle",
AudioTrackIndex: 1,
TargetBitrateKbps: 8000,
TotalDuration: 7200,
FastStart: true,
InputPath: "/media/movie.mkv",
SessionID: "abc",
SourceVideoCodec: "hevc",
VideoBitstreamFilter: "dovi_rpu=strip=1",
SeekSeconds: 900,
StreamOriginSeconds: 896,
CopySeekAnchorResolved: true,
TargetResolution: "1080p",
TargetCodecVideo: "h264",
TargetCodecAudio: "aac",
SegmentDuration: 2,
StartSegmentNumber: 450,
SubtitleTrackIndex: 3,
SubtitleBurnIn: true,
SubtitleCodec: "hdmv_pgs_subtitle",
AudioTrackIndex: 1,
TargetBitrateKbps: 8000,
TotalDuration: 7200,
FastStart: true,
})
claims := card.ToClaims()
@@ -227,7 +234,8 @@ func TestRecipeCardClaimsRoundTrip(t *testing.T) {
// Byte-affecting encode parameters.
if got.InputPath != card.InputPath || got.SourceVideoCodec != card.SourceVideoCodec ||
got.VideoBitstreamFilter != card.VideoBitstreamFilter ||
got.SeekSeconds != card.SeekSeconds || got.TargetResolution != card.TargetResolution ||
got.SeekSeconds != card.SeekSeconds || got.StreamOriginSeconds != card.StreamOriginSeconds ||
got.CopySeekAnchorResolved != card.CopySeekAnchorResolved || got.TargetResolution != card.TargetResolution ||
got.TargetCodecVideo != card.TargetCodecVideo || got.TargetCodecAudio != card.TargetCodecAudio ||
got.SegmentDuration != card.SegmentDuration || got.StartSegmentNumber != card.StartSegmentNumber ||
got.SubtitleTrackIndex != card.SubtitleTrackIndex || got.SubtitleBurnIn != card.SubtitleBurnIn ||
+56
View File
@@ -93,6 +93,14 @@ type SessionStreamState struct {
SegmentDuration int
}
// TranscodeRoute identifies the process serving a playback session. An empty
// NodeURL means the integrated server owns the process; an empty TransportID
// means a remote process uses the public playback session ID.
type TranscodeRoute struct {
NodeURL string
TransportID string
}
// SessionReplacement is the complete mutable session state associated with a
// protocol-v3 replacement plan. Position is optional because ordinary failure
// recovery must preserve the player's latest progress while seek recovery
@@ -834,6 +842,36 @@ func (m *SessionManager) ApplyReplacement(sessionID string, replacement SessionR
if !ok {
return SessionReplacementRollback{}, ErrSessionNotFound
}
return m.applyReplacementLocked(s, sessionID, replacement)
}
// ApplyReplacementIfRoute applies a complete replacement only while the
// session still routes to expected. It publishes route and stream state in one
// critical section so callers never expose a successor with predecessor state.
func (m *SessionManager) ApplyReplacementIfRoute(
sessionID string,
expected TranscodeRoute,
replacement SessionReplacement,
) (SessionReplacementRollback, bool, error) {
m.mu.Lock()
defer m.mu.Unlock()
s, ok := m.sessions[sessionID]
if !ok {
return SessionReplacementRollback{}, false, ErrSessionNotFound
}
if s.TranscodeNodeURL != expected.NodeURL || s.TranscodeTransportID != expected.TransportID {
return SessionReplacementRollback{}, false, nil
}
rollback, err := m.applyReplacementLocked(s, sessionID, replacement)
return rollback, err == nil, err
}
func (m *SessionManager) applyReplacementLocked(
s *Session,
sessionID string,
replacement SessionReplacement,
) (SessionReplacementRollback, error) {
if replacement.EffectiveMediaFileID <= 0 {
return SessionReplacementRollback{}, errors.New("replacement effective media file id is invalid")
}
@@ -926,6 +964,24 @@ func (m *SessionManager) SetTranscodeNodeURL(sessionID, url string) error {
return nil
}
// SetTranscodeRoute atomically assigns the node and process identity used to
// serve a transcode.
func (m *SessionManager) SetTranscodeRoute(sessionID string, route TranscodeRoute) error {
m.mu.Lock()
defer m.mu.Unlock()
s, ok := m.sessions[sessionID]
if !ok {
return ErrSessionNotFound
}
s.TranscodeNodeURL = route.NodeURL
s.TranscodeTransportID = route.TransportID
s.streamRevision++
m.touchSessionLocked(s)
return nil
}
// SetEffectiveMediaFileID updates the currently delivered source file while
// preserving the originally requested file selection.
func (m *SessionManager) SetEffectiveMediaFileID(sessionID string, fileID int) error {
+73
View File
@@ -661,6 +661,79 @@ func TestSetTranscodeNodeURL_NotFound(t *testing.T) {
}
}
func TestSetTranscodeRoutePublishesNodeAndTransportTogether(t *testing.T) {
mgr := playback.NewSessionManager(0, 0)
session, err := mgr.StartSession(1, "profile-1", 42, playback.PlayRemux, true)
if err != nil {
t.Fatalf("StartSession: %v", err)
}
if err := mgr.SetTranscodeRoute(session.ID, playback.TranscodeRoute{NodeURL: "http://node:8070", TransportID: session.ID + "-legacy-next"}); err != nil {
t.Fatalf("SetTranscodeRoute: %v", err)
}
got, err := mgr.GetSession(session.ID)
if err != nil {
t.Fatalf("GetSession: %v", err)
}
if got.TranscodeNodeURL != "http://node:8070" || got.TranscodeTransportID != session.ID+"-legacy-next" {
t.Fatalf("transcode route = %q/%q", got.TranscodeNodeURL, got.TranscodeTransportID)
}
}
func TestApplyReplacementIfRouteRejectsStalePredecessor(t *testing.T) {
mgr := playback.NewSessionManager(0, 0)
session, err := mgr.StartSession(1, "profile-1", 42, playback.PlayRemux, true)
if err != nil {
t.Fatalf("StartSession: %v", err)
}
if err := mgr.SetTranscodeRoute(session.ID, playback.TranscodeRoute{NodeURL: "http://old-node", TransportID: "old-transport"}); err != nil {
t.Fatalf("SetTranscodeRoute: %v", err)
}
replacement := playback.SessionReplacement{
EffectiveMediaFileID: 42,
StreamState: playback.SessionStreamState{
PlayMethod: playback.PlayTranscode,
BasePlayMethod: playback.PlayRemux,
TranscodeAudio: true,
TranscodeRouteSet: true,
SubtitleTrackIndex: -1,
TranscodeNodeURL: "http://new-node",
TranscodeTransportID: "new-transport",
},
}
_, published, err := mgr.ApplyReplacementIfRoute(
session.ID,
playback.TranscodeRoute{NodeURL: "http://stale-node", TransportID: "stale-transport"},
replacement,
)
if err != nil {
t.Fatalf("ApplyReplacementIfRoute stale: %v", err)
}
if published {
t.Fatal("stale predecessor unexpectedly published a replacement")
}
_, published, err = mgr.ApplyReplacementIfRoute(
session.ID,
playback.TranscodeRoute{NodeURL: "http://old-node", TransportID: "old-transport"},
replacement,
)
if err != nil {
t.Fatalf("ApplyReplacementIfRoute current: %v", err)
}
if !published {
t.Fatal("current predecessor did not publish its replacement")
}
got, err := mgr.GetSession(session.ID)
if err != nil {
t.Fatalf("GetSession: %v", err)
}
if got.TranscodeNodeURL != "http://new-node" || got.TranscodeTransportID != "new-transport" {
t.Fatalf("transcode route = %q/%q", got.TranscodeNodeURL, got.TranscodeTransportID)
}
}
func TestSetEffectiveMediaFileID(t *testing.T) {
mgr := playback.NewSessionManager(0, 0)
session, err := mgr.StartSession(1, "profile-1", 100, playback.PlayDirect, false)
+51 -10
View File
@@ -38,16 +38,24 @@ type TranscodeOpts struct {
SourceVideoCodec string
VideoBitstreamFilter string // validated copy-mode BSF, e.g. dovi_rpu=strip=1
SeekSeconds float64
TargetResolution string // e.g., 1080p, 720p
TargetCodecVideo string // e.g., h264 (or hevc if allowed)
TargetCodecAudio string // e.g., aac
SegmentDuration int // seconds, default 6
StartSegmentNumber int // -hls_segment_start_number, default 0
FFmpegPath string // optional explicit ffmpeg binary path
HWAccel string // auto, qsv, vaapi, nvenc, none
HWDevice string // e.g., /dev/dri/renderD128 (default if empty)
SubtitleTrackIndex int // -1 = no subtitles
SubtitleBurnIn bool
// StreamOriginSeconds is the keyframe timestamp at which a copy-video
// stream actually begins. SeekSeconds remains the client-requested -ss so
// FFmpeg performs exactly one demuxer seek; this origin keeps response and
// reconstruction timelines aligned with the resulting media pre-roll.
StreamOriginSeconds float64
// CopySeekAnchorResolved distinguishes a valid zero-second origin from
// older/shared recipes that never resolved a copy seek anchor.
CopySeekAnchorResolved bool
TargetResolution string // e.g., 1080p, 720p
TargetCodecVideo string // e.g., h264 (or hevc if allowed)
TargetCodecAudio string // e.g., aac
SegmentDuration int // seconds, default 6
StartSegmentNumber int // -hls_segment_start_number, default 0
FFmpegPath string // optional explicit ffmpeg binary path
HWAccel string // auto, qsv, vaapi, nvenc, none
HWDevice string // e.g., /dev/dri/renderD128 (default if empty)
SubtitleTrackIndex int // -1 = no subtitles
SubtitleBurnIn bool
// SubtitleCodec is the probed codec of the burn-in track (e.g. "subrip",
// "hdmv_pgs_subtitle"). Bitmap codecs (PGS/DVD/DVB) select the overlay
// filter_complex pipeline; text codecs use the libass subtitles filter.
@@ -1533,6 +1541,28 @@ func (s *TranscodeSession) cleanStaleSegments(startSegment int) {
// copy-mode sessions, stale segments at or after the restart point are
// cleaned to prevent serving data from the wrong timeline position.
func (s *TranscodeSession) Restart(ctx context.Context, seekSeconds float64, startSegment int) error {
return s.restart(ctx, seekSeconds, startSegment, 0, false)
}
// RestartWithCopySeekAnchor restarts a copy-video stream with the keyframe
// origin resolved for this specific seek. Keeping this metadata explicit
// prevents a prior seek's origin from being reused after an audio switch.
func (s *TranscodeSession) RestartWithCopySeekAnchor(
ctx context.Context,
seekSeconds float64,
startSegment int,
streamOriginSeconds float64,
) error {
return s.restart(ctx, seekSeconds, startSegment, streamOriginSeconds, true)
}
func (s *TranscodeSession) restart(
ctx context.Context,
seekSeconds float64,
startSegment int,
streamOriginSeconds float64,
copySeekAnchorResolved bool,
) error {
s.mu.Lock()
// Single-flight: a second caller arriving while a restart is in
// progress must not kill the process the first restart just started.
@@ -1574,6 +1604,14 @@ func (s *TranscodeSession) Restart(ctx context.Context, seekSeconds float64, sta
opts.SeekSeconds = seekSeconds
opts.StartSegmentNumber = startSegment
if strings.EqualFold(opts.TargetCodecVideo, "copy") {
// A seek restart describes a new emitted timeline. Never retain the
// previous copy seek's keyframe origin when the caller has not resolved
// a replacement; falling back to SeekSeconds is conservative and matches
// the behavior of recipes created before copy anchors were introduced.
opts.StreamOriginSeconds = streamOriginSeconds
opts.CopySeekAnchorResolved = copySeekAnchorResolved
}
opts.FastStart = false // seek-restarts use veryfast for better quality
args := buildFFmpegArgs(opts)
@@ -1871,6 +1909,9 @@ func (s *TranscodeSession) SegmentStartTime(segNum int) (float64, bool, error) {
s.mu.Lock()
manifestPath := filepath.Join(s.outputDir, "stream.m3u8")
baseSeekSeconds := s.opts.SeekSeconds
if strings.EqualFold(s.opts.TargetCodecVideo, "copy") && s.opts.CopySeekAnchorResolved {
baseSeekSeconds = s.opts.StreamOriginSeconds
}
s.mu.Unlock()
manifest, err := os.ReadFile(manifestPath)
+39 -1
View File
@@ -198,6 +198,23 @@ func (m *TranscodeManager) SwapTranscodeSession(sessionID string, successor *Tra
return predecessor
}
// SwapTranscodeSessionIf publishes successor only while the live map still
// contains expected. Callers use it after staging a process in a distinct
// output directory so a stale replacement cannot overwrite a newer process.
func (m *TranscodeManager) SwapTranscodeSessionIf(
sessionID string,
expected *TranscodeSession,
successor *TranscodeSession,
) bool {
m.transcodeMu.Lock()
defer m.transcodeMu.Unlock()
if m.transcodes[sessionID] != expected {
return false
}
m.transcodes[sessionID] = successor
return true
}
// StopRemoteTranscode removes only the remote node process. It deliberately
// leaves the local live-map entry untouched for an atomic remote-to-local v3
// replacement.
@@ -246,12 +263,33 @@ func (m *TranscodeManager) LockSessionLifecycle(sessionID string) func() {
// replaced it, the stale handle is not re-spawned and ErrSessionSuperseded is
// returned.
func (m *TranscodeManager) RestartSessionLocked(ctx context.Context, sessionID string, ts *TranscodeSession, seekSeconds float64, startSegment int) error {
return m.restartSessionLocked(sessionID, ts, func() error {
return ts.Restart(ctx, seekSeconds, startSegment)
})
}
func (m *TranscodeManager) restartSessionLocked(sessionID string, ts *TranscodeSession, restart func() error) error {
unlock := m.LockSessionLifecycle(sessionID)
defer unlock()
if live := m.GetTranscodeSession(sessionID); live != ts {
return ErrSessionSuperseded
}
return ts.Restart(ctx, seekSeconds, startSegment)
return restart()
}
// RestartSessionLockedWithCopySeekAnchor is RestartSessionLocked with the
// resolved keyframe origin for a legacy copy-video seek restart.
func (m *TranscodeManager) RestartSessionLockedWithCopySeekAnchor(
ctx context.Context,
sessionID string,
ts *TranscodeSession,
seekSeconds float64,
startSegment int,
streamOriginSeconds float64,
) error {
return m.restartSessionLocked(sessionID, ts, func() error {
return ts.RestartWithCopySeekAnchor(ctx, seekSeconds, startSegment, streamOriginSeconds)
})
}
// markReconstructing records that sessionID's ffmpeg is mid-reconstruct and
+8 -5
View File
@@ -300,9 +300,12 @@ func TestRestartSeekTarget_CopyModeUsesManifestTimelineWhenAvailable(t *testing.
session := &TranscodeSession{
outputDir: tempDir,
opts: TranscodeOpts{
SeekSeconds: 18.261,
TargetCodecVideo: "copy",
SegmentDuration: 2,
SeekSeconds: 18.261,
StreamOriginSeconds: 18,
CopySeekAnchorResolved: true,
TargetCodecVideo: "copy",
SegmentDuration: 2,
StartSegmentNumber: 9,
},
}
@@ -313,8 +316,8 @@ func TestRestartSeekTarget_CopyModeUsesManifestTimelineWhenAvailable(t *testing.
if !ok {
t.Fatal("RestartSeekTarget returned ok=false")
}
if math.Abs(got-20.93) > 0.0001 {
t.Fatalf("RestartSeekTarget(10) = %.6f, want 20.93", got)
if math.Abs(got-20.669) > 0.0001 {
t.Fatalf("RestartSeekTarget(10) = %.6f, want 20.669", got)
}
}
@@ -74,6 +74,47 @@ func TestRestartInvokesRestartHook(t *testing.T) {
}
}
func TestRestartCopySeekOriginIsReplacedOrCleared(t *testing.T) {
truePath, err := exec.LookPath("true")
if err != nil {
t.Skipf("`true` not found in PATH: %v", err)
}
newSession := func() *TranscodeSession {
return &TranscodeSession{
outputDir: t.TempDir(),
opts: TranscodeOpts{
TargetCodecVideo: "copy",
SegmentDuration: 2,
SeekSeconds: 18,
StreamOriginSeconds: 10,
CopySeekAnchorResolved: true,
StartSegmentNumber: 5,
FFmpegPath: truePath,
},
}
}
resolved := newSession()
if err := resolved.RestartWithCopySeekAnchor(context.Background(), 100, 48, 96); err != nil {
t.Fatalf("RestartWithCopySeekAnchor: %v", err)
}
resolvedOpts := resolved.Opts()
if resolvedOpts.SeekSeconds != 100 || resolvedOpts.StreamOriginSeconds != 96 ||
!resolvedOpts.CopySeekAnchorResolved || resolvedOpts.StartSegmentNumber != 48 {
t.Fatalf("resolved restart opts = %+v", resolvedOpts)
}
unresolved := newSession()
if err := unresolved.Restart(context.Background(), 100, 50); err != nil {
t.Fatalf("Restart: %v", err)
}
unresolvedOpts := unresolved.Opts()
if unresolvedOpts.StreamOriginSeconds != 0 || unresolvedOpts.CopySeekAnchorResolved {
t.Fatalf("generic restart retained stale copy origin: %+v", unresolvedOpts)
}
}
// TestRestartIsSingleFlight covers the other half: Restart must be
// single-flight per session. A second caller arriving while a restart is in
// progress must return immediately without killing the process the first
+15 -13
View File
@@ -46,19 +46,21 @@ type Claims struct {
// Reconstruction recipe — the byte-affecting encode parameters, mirroring the
// former playback.RecipeCard. Zero for direct/remux tokens, which reconstruct
// from identity alone plus the client-supplied position.
SourceVideoCodec string `json:"svc,omitempty"`
VideoBitstreamFilter string `json:"vbsf,omitempty"`
OutputSubdir string `json:"osd,omitempty"`
SeekSeconds float64 `json:"seek,omitempty"`
SegmentDuration int `json:"segd,omitempty"`
StartSegmentNumber int `json:"ssn,omitempty"`
SubtitleTrackIndex int `json:"sti,omitempty"`
SubtitleBurnIn bool `json:"sbi,omitempty"`
SubtitleCodec string `json:"sbc,omitempty"`
TargetBitrateKbps int `json:"tbr,omitempty"`
TotalDuration float64 `json:"dur,omitempty"`
FastStart bool `json:"fs,omitempty"`
TargetCodecAudio string `json:"tca,omitempty"`
SourceVideoCodec string `json:"svc,omitempty"`
VideoBitstreamFilter string `json:"vbsf,omitempty"`
OutputSubdir string `json:"osd,omitempty"`
SeekSeconds float64 `json:"seek,omitempty"`
StreamOriginSeconds float64 `json:"origin,omitempty"`
CopySeekAnchorResolved bool `json:"origin_ok,omitempty"`
SegmentDuration int `json:"segd,omitempty"`
StartSegmentNumber int `json:"ssn,omitempty"`
SubtitleTrackIndex int `json:"sti,omitempty"`
SubtitleBurnIn bool `json:"sbi,omitempty"`
SubtitleCodec string `json:"sbc,omitempty"`
TargetBitrateKbps int `json:"tbr,omitempty"`
TotalDuration float64 `json:"dur,omitempty"`
FastStart bool `json:"fs,omitempty"`
TargetCodecAudio string `json:"tca,omitempty"`
// Recipe staleness hint, bumped on each re-mint after a recipe mutation
// (audio/quality/seek switch). An optional client-side hint only.
+48 -44
View File
@@ -27,25 +27,27 @@ import (
// TranscodeStartRequest is the JSON body for POST /transcode/start.
type TranscodeStartRequest struct {
SessionID string `json:"session_id"`
InputPath string `json:"input_path"`
SourceVideoCodec string `json:"source_video_codec"`
VideoBitstreamFilter string `json:"video_bitstream_filter,omitempty"`
SeekSeconds float64 `json:"seek_seconds"`
StartSegmentNumber int `json:"start_segment_number"`
TargetResolution string `json:"target_resolution"`
TargetCodecVideo string `json:"target_codec_video"`
TargetCodecAudio string `json:"target_codec_audio"`
TargetAudioChannels int `json:"target_audio_channels,omitempty"`
TargetBitrateKbps int `json:"target_bitrate_kbps"`
SegmentDuration int `json:"segment_duration"`
HWAccel string `json:"hw_accel"`
AudioTrackIndex int `json:"audio_track_index"`
SubtitleTrackIndex int `json:"subtitle_track_index"`
SubtitleBurnIn bool `json:"subtitle_burn_in"`
SubtitleCodec string `json:"subtitle_codec,omitempty"`
TotalDuration float64 `json:"total_duration"`
RequireReady bool `json:"require_ready,omitempty"`
SessionID string `json:"session_id"`
InputPath string `json:"input_path"`
SourceVideoCodec string `json:"source_video_codec"`
VideoBitstreamFilter string `json:"video_bitstream_filter,omitempty"`
SeekSeconds float64 `json:"seek_seconds"`
StreamOriginSeconds float64 `json:"stream_origin_seconds,omitempty"`
CopySeekAnchorResolved bool `json:"copy_seek_anchor_resolved,omitempty"`
StartSegmentNumber int `json:"start_segment_number"`
TargetResolution string `json:"target_resolution"`
TargetCodecVideo string `json:"target_codec_video"`
TargetCodecAudio string `json:"target_codec_audio"`
TargetAudioChannels int `json:"target_audio_channels,omitempty"`
TargetBitrateKbps int `json:"target_bitrate_kbps"`
SegmentDuration int `json:"segment_duration"`
HWAccel string `json:"hw_accel"`
AudioTrackIndex int `json:"audio_track_index"`
SubtitleTrackIndex int `json:"subtitle_track_index"`
SubtitleBurnIn bool `json:"subtitle_burn_in"`
SubtitleCodec string `json:"subtitle_codec,omitempty"`
TotalDuration float64 `json:"total_duration"`
RequireReady bool `json:"require_ready,omitempty"`
}
// TranscodeStartResponse is the JSON response for POST /transcode/start.
@@ -474,31 +476,33 @@ func (s *Server) handleStart(w http.ResponseWriter, r *http.Request) {
outputDir := filepath.Join(cfg.Playback.TranscodeDir, req.SessionID)
opts := playback.TranscodeOpts{
InputPath: req.InputPath,
OutputDir: outputDir,
SessionID: req.SessionID,
SourceVideoCodec: req.SourceVideoCodec,
VideoBitstreamFilter: req.VideoBitstreamFilter,
SeekSeconds: req.SeekSeconds,
StartSegmentNumber: req.StartSegmentNumber,
TargetResolution: req.TargetResolution,
TargetCodecVideo: req.TargetCodecVideo,
TargetCodecAudio: req.TargetCodecAudio,
TargetAudioChannels: req.TargetAudioChannels,
TargetBitrateKbps: req.TargetBitrateKbps,
SegmentDuration: req.SegmentDuration,
FFmpegPath: cfg.Playback.FFmpegPath,
HWAccel: req.HWAccel,
HWDevice: "",
AudioTrackIndex: req.AudioTrackIndex,
SubtitleTrackIndex: req.SubtitleTrackIndex,
SubtitleBurnIn: req.SubtitleBurnIn,
SubtitleCodec: req.SubtitleCodec,
TotalDuration: req.TotalDuration,
FastStart: true,
NodeType: "transcode",
ExecutionMode: "transcode_node",
FFmpegLogSink: s.ffmpegSink,
InputPath: req.InputPath,
OutputDir: outputDir,
SessionID: req.SessionID,
SourceVideoCodec: req.SourceVideoCodec,
VideoBitstreamFilter: req.VideoBitstreamFilter,
SeekSeconds: req.SeekSeconds,
StreamOriginSeconds: req.StreamOriginSeconds,
CopySeekAnchorResolved: req.CopySeekAnchorResolved,
StartSegmentNumber: req.StartSegmentNumber,
TargetResolution: req.TargetResolution,
TargetCodecVideo: req.TargetCodecVideo,
TargetCodecAudio: req.TargetCodecAudio,
TargetAudioChannels: req.TargetAudioChannels,
TargetBitrateKbps: req.TargetBitrateKbps,
SegmentDuration: req.SegmentDuration,
FFmpegPath: cfg.Playback.FFmpegPath,
HWAccel: req.HWAccel,
HWDevice: "",
AudioTrackIndex: req.AudioTrackIndex,
SubtitleTrackIndex: req.SubtitleTrackIndex,
SubtitleBurnIn: req.SubtitleBurnIn,
SubtitleCodec: req.SubtitleCodec,
TotalDuration: req.TotalDuration,
FastStart: true,
NodeType: "transcode",
ExecutionMode: "transcode_node",
FFmpegLogSink: s.ffmpegSink,
}
if opts.HWAccel == "" && cfg.Playback.HWAccel != "" {
+43
View File
@@ -71,6 +71,49 @@ func TestHandleStartRequireReadyRejectsExitedFFmpeg(t *testing.T) {
}
}
func TestHandleStartDistinctReplacementFailurePreservesPredecessor(t *testing.T) {
server := newTestServer(t)
server.sessions["public-session"] = &playback.TranscodeSession{}
server.activeJobs.Store(1)
ffmpegPath := filepath.Join(t.TempDir(), "failing-ffmpeg.sh")
if err := os.WriteFile(ffmpegPath, []byte("#!/bin/sh\nexit 1\n"), 0o755); err != nil {
t.Fatal(err)
}
server.watcher.Config().Playback.FFmpegPath = ffmpegPath
requestBody, err := json.Marshal(TranscodeStartRequest{
SessionID: "public-session-legacy-replacement",
InputPath: "/media/movie.mkv",
TargetCodecVideo: "copy",
TargetCodecAudio: "aac",
SegmentDuration: 2,
RequireReady: true,
})
if err != nil {
t.Fatal(err)
}
req := httptest.NewRequest(http.MethodPost, "/transcode/start", bytes.NewReader(requestBody))
rr := httptest.NewRecorder()
server.handleStart(rr, req)
if rr.Code != http.StatusInternalServerError {
t.Fatalf("status = %d, body = %s", rr.Code, rr.Body.String())
}
server.mu.RLock()
predecessor := server.sessions["public-session"]
_, replacementRegistered := server.sessions["public-session-legacy-replacement"]
server.mu.RUnlock()
if predecessor == nil {
t.Fatal("failed distinct replacement removed the active predecessor")
}
if replacementRegistered {
t.Fatal("failed distinct replacement was registered")
}
if got := server.activeJobs.Load(); got != 1 {
t.Fatalf("active jobs = %d, want predecessor only", got)
}
}
func signCard(t *testing.T, card playback.RecipeCard) string {
t.Helper()
tok, err := streamtoken.Sign(card.ToClaims(), testSecret, time.Hour)
+45 -5
View File
@@ -114,13 +114,53 @@ func ListCollections(db *sql.DB, profileID string) ([]Collection, error) {
if err := rows.Scan(&c.ID, &c.ProfileID, &c.CreatorProfileID, &c.Name, &c.CollectionType, &c.IsShared, &c.QueryDefinition, &c.SortConfig, &c.CreatedAt, &c.UpdatedAt); err != nil {
return nil, err
}
c.AllowedProfileIDs, err = listCollectionProfiles(db, c.ID)
if err != nil {
return nil, err
}
collections = append(collections, c)
}
return collections, rows.Err()
if err := rows.Err(); err != nil {
return nil, err
}
if err := attachCollectionProfiles(db, profileID, collections); err != nil {
return nil, err
}
return collections, nil
}
// attachCollectionProfiles fills AllowedProfileIDs for every collection with a
// single batched query, avoiding the N+1 round trip a per-collection lookup
// would create.
func attachCollectionProfiles(db *sql.DB, profileID string, collections []Collection) error {
if len(collections) == 0 {
return nil
}
rows, err := db.Query(
`SELECT allowed.collection_id, allowed.profile_id
FROM personal_collection_profiles AS visible
JOIN personal_collection_profiles AS allowed
ON allowed.collection_id = visible.collection_id
WHERE visible.profile_id = ?
ORDER BY allowed.profile_id ASC`,
profileID,
)
if err != nil {
return err
}
defer rows.Close()
byCollection := make(map[string][]string, len(collections))
for rows.Next() {
var collectionID, profileID string
if err := rows.Scan(&collectionID, &profileID); err != nil {
return err
}
byCollection[collectionID] = append(byCollection[collectionID], profileID)
}
if err := rows.Err(); err != nil {
return err
}
for i := range collections {
collections[i].AllowedProfileIDs = byCollection[collections[i].ID]
}
return nil
}
// UpdateCollection renames a collection and updates its updated_at timestamp.
+92
View File
@@ -0,0 +1,92 @@
package userdb
import (
"database/sql"
"reflect"
"testing"
"github.com/Silo-Server/silo-server/internal/userstore"
)
func newListingTestDB(t *testing.T) *sql.DB {
t.Helper()
db, err := sql.Open("sqlite3", ":memory:")
if err != nil {
t.Fatalf("open sqlite: %v", err)
}
db.SetMaxOpenConns(1)
t.Cleanup(func() { _ = db.Close() })
if err := InitSchema(db); err != nil {
t.Fatalf("InitSchema: %v", err)
}
return db
}
func TestListProfilesAttachesAllowedLibrariesInOrder(t *testing.T) {
db := newListingTestDB(t)
for _, profile := range []Profile{
{ID: "p1", Name: "One", AllowedLibraryIDs: []int{5, 2}, CreatedAt: "2026-01-01T00:00:00Z"},
{ID: "p2", Name: "Two", AllowedLibraryIDs: []int{3}, CreatedAt: "2026-01-02T00:00:00Z"},
{ID: "p3", Name: "Three", CreatedAt: "2026-01-03T00:00:00Z"},
} {
if err := CreateProfile(db, profile); err != nil {
t.Fatalf("CreateProfile(%s): %v", profile.ID, err)
}
}
profiles, err := ListProfiles(db)
if err != nil {
t.Fatalf("ListProfiles: %v", err)
}
if len(profiles) != 3 {
t.Fatalf("ListProfiles returned %d profiles, want 3", len(profiles))
}
if got, want := profiles[0].AllowedLibraryIDs, []int{2, 5}; !reflect.DeepEqual(got, want) {
t.Fatalf("p1 AllowedLibraryIDs = %v, want %v", got, want)
}
if got, want := profiles[1].AllowedLibraryIDs, []int{3}; !reflect.DeepEqual(got, want) {
t.Fatalf("p2 AllowedLibraryIDs = %v, want %v", got, want)
}
if profiles[2].AllowedLibraryIDs != nil {
t.Fatalf("p3 AllowedLibraryIDs = %v, want nil", profiles[2].AllowedLibraryIDs)
}
}
func TestListCollectionsAttachesAllowedProfilesInOrder(t *testing.T) {
db := newListingTestDB(t)
collection, err := CreateCollection(db, userstore.CreateCollectionInput{
CreatorProfileID: "p1",
Name: "Shared",
IsShared: true,
AllowedProfileIDs: []string{"p3", "p2"},
})
if err != nil {
t.Fatalf("CreateCollection: %v", err)
}
collections, err := ListCollections(db, "p2")
if err != nil {
t.Fatalf("ListCollections: %v", err)
}
if len(collections) != 1 || collections[0].ID != collection.ID {
t.Fatalf("ListCollections = %+v, want collection %s", collections, collection.ID)
}
if got, want := collections[0].AllowedProfileIDs, []string{"p1", "p2", "p3"}; !reflect.DeepEqual(got, want) {
t.Fatalf("AllowedProfileIDs = %v, want %v", got, want)
}
}
func TestBatchAttachDoesNotScaleBindParametersWithRows(t *testing.T) {
db := newListingTestDB(t)
const aboveSQLiteVariableLimit = 32767
profiles := make([]Profile, aboveSQLiteVariableLimit)
if err := attachAllowedLibraries(db, profiles); err != nil {
t.Fatalf("attachAllowedLibraries: %v", err)
}
collections := make([]Collection, aboveSQLiteVariableLimit)
if err := attachCollectionProfiles(db, "viewer", collections); err != nil {
t.Fatalf("attachCollectionProfiles: %v", err)
}
}
+34
View File
@@ -3,6 +3,8 @@ package userdb
import (
"database/sql"
"fmt"
"github.com/Silo-Server/silo-server/internal/userstore"
)
func listProfileAllowedLibraries(db *sql.DB, profileID string) ([]int, error) {
@@ -29,6 +31,38 @@ func listProfileAllowedLibraries(db *sql.DB, profileID string) ([]int, error) {
return libraryIDs, nil
}
// attachAllowedLibraries fills AllowedLibraryIDs for every profile with a
// single batched query, avoiding the N+1 round trip a per-profile lookup
// would create.
func attachAllowedLibraries(db *sql.DB, profiles []Profile) error {
if len(profiles) == 0 {
return nil
}
rows, err := db.Query(
`SELECT profile_id, library_id
FROM profile_allowed_libraries
ORDER BY library_id ASC`,
)
if err != nil {
return fmt.Errorf("listing allowed libraries: %w", err)
}
defer rows.Close()
var allowedLibraries []userstore.ProfileAllowedLibrary
for rows.Next() {
var allowedLibrary userstore.ProfileAllowedLibrary
if err := rows.Scan(&allowedLibrary.ProfileID, &allowedLibrary.LibraryID); err != nil {
return fmt.Errorf("scanning allowed library: %w", err)
}
allowedLibraries = append(allowedLibraries, allowedLibrary)
}
if err := rows.Err(); err != nil {
return fmt.Errorf("iterating allowed libraries: %w", err)
}
userstore.AttachAllowedLibraries(profiles, allowedLibraries)
return nil
}
func replaceProfileAllowedLibrariesTx(tx *sql.Tx, profileID string, libraryIDs []int) error {
if _, err := tx.Exec("DELETE FROM profile_allowed_libraries WHERE profile_id = ?", profileID); err != nil {
return fmt.Errorf("clearing allowed libraries for profile %s: %w", profileID, err)
+3 -4
View File
@@ -128,15 +128,14 @@ func ListProfiles(db *sql.DB) ([]Profile, error) {
); err != nil {
return nil, fmt.Errorf("scanning profile row: %w", err)
}
p.AllowedLibraryIDs, err = listProfileAllowedLibraries(db, p.ID)
if err != nil {
return nil, err
}
profiles = append(profiles, p)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("iterating profile rows: %w", err)
}
if err := attachAllowedLibraries(db, profiles); err != nil {
return nil, err
}
return profiles, nil
}
@@ -6,6 +6,8 @@ import (
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgconn"
"github.com/Silo-Server/silo-server/internal/userstore"
)
type profileLibraryQuerier interface {
@@ -41,6 +43,45 @@ func listProfileAllowedLibraries(ctx context.Context, q profileLibraryQuerier, u
return libraryIDs, nil
}
// attachAllowedLibraries fills AllowedLibraryIDs for every profile with a
// single batched query, avoiding the N+1 round trip a per-profile lookup
// would create.
func (s *PostgresUserStore) attachAllowedLibraries(ctx context.Context, profiles []userstore.Profile) error {
if len(profiles) == 0 {
return nil
}
ids := make([]string, len(profiles))
for i := range profiles {
ids[i] = profiles[i].ID
}
rows, err := s.pool.Query(ctx,
`SELECT profile_id, library_id
FROM user_profile_allowed_libraries
WHERE user_id = $1 AND profile_id = ANY($2)
ORDER BY library_id ASC`,
s.userID,
ids,
)
if err != nil {
return fmt.Errorf("listing allowed libraries for user %d: %w", s.userID, err)
}
defer rows.Close()
var allowedLibraries []userstore.ProfileAllowedLibrary
for rows.Next() {
var allowedLibrary userstore.ProfileAllowedLibrary
if err := rows.Scan(&allowedLibrary.ProfileID, &allowedLibrary.LibraryID); err != nil {
return fmt.Errorf("scanning allowed library for user %d: %w", s.userID, err)
}
allowedLibraries = append(allowedLibraries, allowedLibrary)
}
if err := rows.Err(); err != nil {
return fmt.Errorf("iterating allowed libraries for user %d: %w", s.userID, err)
}
userstore.AttachAllowedLibraries(profiles, allowedLibraries)
return nil
}
func replaceProfileAllowedLibraries(ctx context.Context, q profileLibraryQuerier, userID int, profileID string, libraryIDs []int) error {
if _, err := q.Exec(ctx,
"DELETE FROM user_profile_allowed_libraries WHERE user_id = $1 AND profile_id = $2",
+3 -4
View File
@@ -126,15 +126,14 @@ func (s *PostgresUserStore) ListProfiles(ctx context.Context) ([]userstore.Profi
if err != nil {
return nil, fmt.Errorf("scanning profile row: %w", err)
}
p.AllowedLibraryIDs, err = listProfileAllowedLibraries(ctx, s.pool, s.userID, p.ID)
if err != nil {
return nil, err
}
profiles = append(profiles, *p)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("iterating profile rows: %w", err)
}
if err := s.attachAllowedLibraries(ctx, profiles); err != nil {
return nil, err
}
return profiles, nil
}
+22
View File
@@ -0,0 +1,22 @@
package userstore
// ProfileAllowedLibrary associates one profile with one allowed library.
type ProfileAllowedLibrary struct {
ProfileID string
LibraryID int
}
// AttachAllowedLibraries replaces each profile's AllowedLibraryIDs with the
// matching associations, preserving their input order.
func AttachAllowedLibraries(profiles []Profile, allowedLibraries []ProfileAllowedLibrary) {
byProfile := make(map[string][]int, len(profiles))
for _, allowedLibrary := range allowedLibraries {
byProfile[allowedLibrary.ProfileID] = append(
byProfile[allowedLibrary.ProfileID],
allowedLibrary.LibraryID,
)
}
for i := range profiles {
profiles[i].AllowedLibraryIDs = byProfile[profiles[i].ID]
}
}
@@ -0,0 +1,31 @@
package userstore
import (
"reflect"
"testing"
)
func TestAttachAllowedLibraries(t *testing.T) {
profiles := []Profile{
{ID: "p1", AllowedLibraryIDs: []int{99}},
{ID: "p2", AllowedLibraryIDs: []int{99}},
{ID: "p3", AllowedLibraryIDs: []int{99}},
}
allowedLibraries := []ProfileAllowedLibrary{
{ProfileID: "p2", LibraryID: 1},
{ProfileID: "p1", LibraryID: 2},
{ProfileID: "p1", LibraryID: 5},
}
AttachAllowedLibraries(profiles, allowedLibraries)
if got, want := profiles[0].AllowedLibraryIDs, []int{2, 5}; !reflect.DeepEqual(got, want) {
t.Fatalf("p1 AllowedLibraryIDs = %v, want %v", got, want)
}
if got, want := profiles[1].AllowedLibraryIDs, []int{1}; !reflect.DeepEqual(got, want) {
t.Fatalf("p2 AllowedLibraryIDs = %v, want %v", got, want)
}
if profiles[2].AllowedLibraryIDs != nil {
t.Fatalf("p3 AllowedLibraryIDs = %v, want nil", profiles[2].AllowedLibraryIDs)
}
}
+262 -33
View File
@@ -9,6 +9,7 @@ import (
"errors"
"fmt"
"io"
"math"
"net/http"
"net/url"
"strconv"
@@ -38,6 +39,7 @@ const (
// survives burst-limit blips; anything longer defers the sync run.
maxInPlaceRetryWait = 15 * time.Second
maxRetryAttempts = 2
maxErrorBodyBytes = 4 << 10
// Without a Retry-After header we cannot tell a burst limit from an
// exhausted daily quota, so default to backing off until the next
@@ -125,19 +127,23 @@ func (p *Provider) RefreshToken(_ context.Context, _ watchsync.ServerConfig, con
func (p *Provider) FetchWatched(ctx context.Context, _ watchsync.ServerConfig, conn watchsync.Connection) ([]watchsync.RemoteWatch, error) {
var rows []watchsync.RemoteWatch
for offset := 0; ; {
page := mdblistPageState{}
for {
var payload mdblistWatchedResponse
path := fmt.Sprintf("/sync/watched?limit=%d&offset=%d", syncPageLimit, offset)
path := page.path("/sync/watched")
if err := p.do(ctx, http.MethodGet, path, conn.AccessToken, nil, &payload); err != nil {
return nil, err
}
page := watchedRowsFromPayload(p.Key(), payload)
rows = append(rows, page...)
pageRows := watchedRowsFromPayload(p.Key(), payload)
rows = append(rows, pageRows...)
fetched := len(payload.Movies) + len(payload.Episodes)
if fetched < syncPageLimit {
done, err := page.advance(payload.Pagination, fetched)
if err != nil {
return nil, fmt.Errorf("mdblist watched pagination: %w", err)
}
if done {
break
}
offset += fetched
}
return rows, nil
}
@@ -302,50 +308,76 @@ func (p *Provider) FetchHistory(ctx context.Context, cfg watchsync.ServerConfig,
func (p *Provider) ExportHistory(ctx context.Context, _ watchsync.ServerConfig, conn watchsync.Connection, plays []watchsync.LocalPlay) (watchsync.ExportResult, error) {
payload := buildWatchedPayload(plays, true)
if payload.empty() {
return watchsync.ExportResult{}, nil
}
return p.sendWatched(ctx, conn, plays, "/sync/watched", payload)
}
func (p *Provider) RemoveHistory(ctx context.Context, _ watchsync.ServerConfig, conn watchsync.Connection, plays []watchsync.LocalPlay) (watchsync.ExportResult, error) {
payload := buildWatchedPayload(plays, false)
if payload.empty() {
return watchsync.ExportResult{}, nil
}
return p.sendWatched(ctx, conn, plays, "/sync/watched/remove", payload)
}
func (p *Provider) sendWatched(ctx context.Context, conn watchsync.Connection, plays []watchsync.LocalPlay, path string, payload mdblistWatchedPayload) (watchsync.ExportResult, error) {
if payload.empty() {
return watchedExportResult(plays, false), nil
}
var body bytes.Buffer
if err := json.NewEncoder(&body).Encode(payload); err != nil {
return watchsync.ExportResult{}, fmt.Errorf("encode mdblist watched payload: %w", err)
}
if err := p.do(ctx, http.MethodPost, path, conn.AccessToken, &body, nil); err != nil {
var response mdblistWatchedWriteResponse
if err := p.do(ctx, http.MethodPost, path, conn.AccessToken, &body, &response); err != nil {
return watchsync.ExportResult{}, err
}
result := watchsync.ExportResult{Sent: make([]string, 0, len(plays))}
return watchedExportResult(plays, !emptyJSONValue(response.NotFound)), nil
}
func watchedExportResult(plays []watchsync.LocalPlay, responseHasNotFound bool) watchsync.ExportResult {
result := watchsync.ExportResult{
Sent: make([]string, 0, len(plays)),
Failed: make(map[string]string),
}
for _, play := range plays {
if play.HistoryID == "" {
continue
}
if !watchedPlaySupported(play) {
result.Failed[play.HistoryID] = "MDBList watched export requires a supported media kind and external ID"
continue
}
if responseHasNotFound {
// MDBList documents not_found only as an unstructured object. It does
// not guarantee enough identity information to safely attribute a
// partial rejection, so never claim that any item in this batch sent.
result.Failed[play.HistoryID] = "MDBList did not accept one or more watched items in the batch"
continue
}
result.Sent = append(result.Sent, play.HistoryID)
}
return result, nil
if len(result.Failed) == 0 {
result.Failed = nil
}
return result
}
func (p *Provider) FetchWatchlist(ctx context.Context, _ watchsync.ServerConfig, conn watchsync.Connection) ([]watchsync.RemoteFavorite, error) {
now := time.Now().UTC()
var rows []watchsync.RemoteFavorite
for offset := 0; ; {
page := mdblistPageState{}
for {
var payload mdblistWatchlistResponse
path := fmt.Sprintf("/watchlist/items?limit=%d&offset=%d", syncPageLimit, offset)
path := page.path("/watchlist/items")
if err := p.do(ctx, http.MethodGet, path, conn.AccessToken, nil, &payload); err != nil {
return nil, err
}
rows = append(rows, watchlistRowsFromPayload(p.Key(), payload, now)...)
fetched := len(payload.Movies) + len(payload.Shows)
if fetched < syncPageLimit {
done, err := page.advance(payload.Pagination, fetched)
if err != nil {
return nil, fmt.Errorf("mdblist watchlist pagination: %w", err)
}
if done {
break
}
offset += fetched
}
return rows, nil
}
@@ -392,30 +424,65 @@ func watchlistRowsFromPayload(providerKey string, payload mdblistWatchlistRespon
}
func (p *Provider) ExportWatchlist(ctx context.Context, _ watchsync.ServerConfig, conn watchsync.Connection, items []watchsync.LocalFavorite) (watchsync.ExportResult, error) {
return p.sendWatchlist(ctx, conn, items, "/watchlist/items/add")
return p.sendWatchlist(ctx, conn, items, "/watchlist/items/add", false)
}
func (p *Provider) RemoveWatchlist(ctx context.Context, _ watchsync.ServerConfig, conn watchsync.Connection, items []watchsync.LocalFavorite) (watchsync.ExportResult, error) {
return p.sendWatchlist(ctx, conn, items, "/watchlist/items/remove")
return p.sendWatchlist(ctx, conn, items, "/watchlist/items/remove", true)
}
func (p *Provider) sendWatchlist(ctx context.Context, conn watchsync.Connection, favorites []watchsync.LocalFavorite, path string) (watchsync.ExportResult, error) {
func (p *Provider) sendWatchlist(ctx context.Context, conn watchsync.Connection, favorites []watchsync.LocalFavorite, path string, removing bool) (watchsync.ExportResult, error) {
payload := buildWatchlistPayload(favorites)
if len(payload.Movies) == 0 && len(payload.Shows) == 0 {
return watchsync.ExportResult{}, nil
return watchlistExportResult(favorites, false, removing), nil
}
var body bytes.Buffer
if err := json.NewEncoder(&body).Encode(payload); err != nil {
return watchsync.ExportResult{}, fmt.Errorf("encode mdblist watchlist payload: %w", err)
}
if err := p.do(ctx, http.MethodPost, path, conn.AccessToken, &body, nil); err != nil {
var response mdblistWatchlistWriteResponse
if err := p.do(ctx, http.MethodPost, path, conn.AccessToken, &body, &response); err != nil {
return watchsync.ExportResult{}, err
}
result := watchsync.ExportResult{Sent: make([]string, 0, len(favorites)*2)}
return watchlistExportResult(favorites, !emptyJSONValue(response.NotFound), removing), nil
}
func watchlistExportResult(favorites []watchsync.LocalFavorite, responseHasNotFound, removing bool) watchsync.ExportResult {
result := watchsync.ExportResult{
Sent: make([]string, 0, len(favorites)),
NotFound: make([]string, 0, len(favorites)),
Failed: make(map[string]string),
}
for _, fav := range favorites {
if fav.MediaItemID == "" {
continue
}
if !watchlistFavoriteSupported(fav) {
result.Failed[fav.MediaItemID] = "MDBList watchlist export requires a movie or series with an external ID"
continue
}
if responseHasNotFound {
if removing {
// A successful removal leaves both removed and already-absent items
// absent remotely. MDBList returns only aggregate counts, so mark
// the batch reconciled without trying to attribute individual rows.
result.NotFound = append(result.NotFound, fav.MediaItemID)
continue
}
// Adds still need item-level identity that the aggregate response
// does not provide, so never claim that any item in the batch sent.
result.Failed[fav.MediaItemID] = "MDBList did not accept one or more watchlist items in the batch"
continue
}
result.Sent = append(result.Sent, fav.MediaItemID)
}
return result, nil
if len(result.Failed) == 0 {
result.Failed = nil
}
if len(result.NotFound) == 0 {
result.NotFound = nil
}
return result
}
func (p *Provider) Start(ctx context.Context, _ watchsync.ServerConfig, conn watchsync.Connection, event watchsync.ScrobbleEvent) error {
@@ -542,6 +609,10 @@ func (p *Provider) doOnce(ctx context.Context, method, path, target string, payl
return -1, fmt.Errorf("mdblist request %s %s rejected: status %d (check api key)", method, path, resp.StatusCode)
}
if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
detail := responseErrorDetail(resp.Body)
if detail != "" {
return -1, fmt.Errorf("mdblist request %s %s failed: status %d: %s", method, path, resp.StatusCode, detail)
}
return -1, fmt.Errorf("mdblist request %s %s failed: status %d", method, path, resp.StatusCode)
}
if out == nil || resp.StatusCode == http.StatusNoContent {
@@ -553,6 +624,28 @@ func (p *Provider) doOnce(ctx context.Context, method, path, target string, payl
return -1, nil
}
func responseErrorDetail(body io.Reader) string {
raw, err := io.ReadAll(io.LimitReader(body, maxErrorBodyBytes))
if err != nil {
return ""
}
raw = bytes.TrimSpace(raw)
if len(raw) == 0 {
return ""
}
var envelope struct {
Error json.RawMessage `json:"error"`
}
if json.Unmarshal(raw, &envelope) == nil && len(envelope.Error) > 0 {
raw = bytes.TrimSpace(envelope.Error)
}
var compact bytes.Buffer
if json.Compact(&compact, raw) == nil {
return compact.String()
}
return strings.Join(strings.Fields(string(raw)), " ")
}
// parseRetryAfter reads an RFC 7231 Retry-After value (delay-seconds or
// HTTP-date). It returns 0 when the header is absent or unparseable.
func parseRetryAfter(value string, now time.Time) time.Duration {
@@ -669,8 +762,110 @@ func (e *mdblistWatchedEpisode) UnmarshalJSON(data []byte) error {
}
type mdblistWatchedResponse struct {
Movies []mdblistWatchedMovie `json:"movies"`
Episodes []mdblistWatchedEpisode `json:"episodes"`
Movies []mdblistWatchedMovie `json:"movies"`
Episodes []mdblistWatchedEpisode `json:"episodes"`
Pagination *mdblistPagination `json:"pagination"`
}
type mdblistPagination struct {
NextCursor string `json:"next_cursor"`
HasMore bool `json:"has_more"`
}
type mdblistPageState struct {
cursor string
offset int
legacyOffset bool
}
func (s mdblistPageState) path(base string) string {
query := url.Values{"limit": {strconv.Itoa(syncPageLimit)}}
if s.cursor != "" {
query.Set("cursor", s.cursor)
} else if s.legacyOffset {
query.Set("offset", strconv.Itoa(s.offset))
}
return base + "?" + query.Encode()
}
func (s *mdblistPageState) advance(pagination *mdblistPagination, fetched int) (bool, error) {
if pagination != nil {
next := strings.TrimSpace(pagination.NextCursor)
if next != "" {
if next == s.cursor {
return false, errors.New("provider returned the same cursor twice")
}
s.cursor = next
return false, nil
}
if !pagination.HasMore {
return true, nil
}
if fetched == 0 {
return false, errors.New("provider reported more pages without returning items")
}
s.cursor = ""
s.legacyOffset = true
s.offset += fetched
return false, nil
}
if fetched < syncPageLimit {
return true, nil
}
// Older MDBList deployments omitted pagination metadata. Preserve the
// deprecated offset fallback for those responses while preferring cursors.
s.cursor = ""
s.legacyOffset = true
s.offset += fetched
return false, nil
}
type mdblistWatchedWriteResponse struct {
NotFound json.RawMessage `json:"not_found"`
}
type mdblistWatchlistWriteResponse struct {
NotFound json.RawMessage `json:"not_found"`
}
func emptyJSONValue(raw json.RawMessage) bool {
if len(bytes.TrimSpace(raw)) == 0 {
return true
}
var value any
if err := json.Unmarshal(raw, &value); err != nil {
return false
}
return emptyJSONTree(value)
}
func emptyJSONTree(value any) bool {
switch typed := value.(type) {
case nil:
return true
case bool:
return !typed
case float64:
return typed == 0
case string:
return strings.TrimSpace(typed) == ""
case []any:
for _, child := range typed {
if !emptyJSONTree(child) {
return false
}
}
return true
case map[string]any:
for _, child := range typed {
if !emptyJSONTree(child) {
return false
}
}
return true
default:
return false
}
}
type mdblistPlaybackItem struct {
@@ -775,8 +970,9 @@ func (i mdblistWatchlistItem) preferredIDs() mdblistIDs {
}
type mdblistWatchlistResponse struct {
Movies []mdblistWatchlistItem `json:"movies"`
Shows []mdblistWatchlistItem `json:"shows"`
Movies []mdblistWatchlistItem `json:"movies"`
Shows []mdblistWatchlistItem `json:"shows"`
Pagination *mdblistPagination `json:"pagination"`
}
type mdblistUser struct {
@@ -866,6 +1062,18 @@ func buildWatchedPayload(plays []watchsync.LocalPlay, includeWatchedAt bool) mdb
return payload
}
func watchedPlaySupported(play watchsync.LocalPlay) bool {
switch play.Kind {
case historyimport.KindMovie:
return idsFromLocal(play.IMDbID, play.TMDBID, play.TVDBID) != (mdblistIDs{})
case historyimport.KindEpisode:
return idsFromLocal(play.IMDbID, play.TMDBID, play.TVDBID) != (mdblistIDs{}) ||
idsFromLocal(play.SeriesIMDbID, play.SeriesTMDBID, play.SeriesTVDBID) != (mdblistIDs{})
default:
return false
}
}
func buildWatchlistPayload(favorites []watchsync.LocalFavorite) mdblistWatchlistPayload {
var payload mdblistWatchlistPayload
for _, fav := range favorites {
@@ -887,6 +1095,17 @@ func buildWatchlistPayload(favorites []watchsync.LocalFavorite) mdblistWatchlist
return payload
}
func watchlistFavoriteSupported(fav watchsync.LocalFavorite) bool {
if fav.Kind != historyimport.KindMovie && fav.Kind != historyimport.KindSeries {
return false
}
ids := idsFromLocal(fav.IMDbID, fav.TMDBID, fav.TVDBID)
if ids == (mdblistIDs{}) {
ids = idsFromProviderItemKey(fav.ProviderItemKey)
}
return ids != (mdblistIDs{})
}
func buildScrobblePayload(event watchsync.ScrobbleEvent) map[string]any {
progress := 0.0
if event.DurationSeconds > 0 {
@@ -898,6 +1117,10 @@ func buildScrobblePayload(event watchsync.ScrobbleEvent) map[string]any {
if progress < 0 {
progress = 0
}
// MDBList validates progress as a decimal with at most five total digits.
// Two decimal places keeps every value in the 0-100 range within that
// contract and avoids raw floating-point expansion on the wire.
progress = math.Round(progress*100) / 100
payload := map[string]any{"progress": progress}
switch event.Kind {
case historyimport.KindEpisode:
@@ -905,9 +1128,15 @@ func buildScrobblePayload(event watchsync.ScrobbleEvent) map[string]any {
if showIDs == (mdblistIDs{}) {
showIDs = idsFromLocal(event.IMDbID, event.TMDBID, event.TVDBID)
}
payload["show"] = map[string]any{"ids": showIDs}
payload["season"] = event.SeasonNumber
payload["episode"] = event.EpisodeNumber
payload["show"] = map[string]any{
"ids": showIDs,
"season": map[string]any{
"number": event.SeasonNumber,
"episode": map[string]any{
"number": event.EpisodeNumber,
},
},
}
default:
payload["movie"] = map[string]any{"ids": idsFromLocal(event.IMDbID, event.TMDBID, event.TVDBID)}
}
@@ -123,6 +123,57 @@ func TestFetchWatchedParsesMoviesAndEpisodes(t *testing.T) {
}
}
func TestFetchWatchedUsesCursorPagination(t *testing.T) {
var queries []string
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
queries = append(queries, r.URL.RawQuery)
w.Header().Set("Content-Type", "application/json")
if r.URL.Query().Get("cursor") == "next-page" {
_, _ = w.Write([]byte(`{"movies":[],"episodes":[],"pagination":{"next_cursor":null}}`))
return
}
_, _ = w.Write([]byte(`{"movies":[],"episodes":[],"pagination":{"next_cursor":"next-page"}}`))
}))
defer server.Close()
p := NewProvider(server.Client(), server.URL)
if _, err := p.FetchWatched(context.Background(), watchsync.ServerConfig{}, watchsync.Connection{AccessToken: "k"}); err != nil {
t.Fatalf("fetch watched: %v", err)
}
if len(queries) != 2 {
t.Fatalf("got %d requests, want 2", len(queries))
}
if strings.Contains(queries[0], "offset=") || !strings.Contains(queries[1], "cursor=next-page") {
t.Fatalf("unexpected pagination queries: %#v", queries)
}
}
func TestFetchWatchedContinuesOffsetPaginationWhenHasMore(t *testing.T) {
var offsets []string
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
offsets = append(offsets, r.URL.Query().Get("offset"))
w.Header().Set("Content-Type", "application/json")
if r.URL.Query().Get("offset") == "1" {
_, _ = w.Write([]byte(`{"movies":[],"episodes":[],"pagination":{"has_more":false}}`))
return
}
_, _ = w.Write([]byte(`{"movies":[{"watched_at":"2025-11-01T08:30:00Z","movie":{"ids":{"imdb":"tt0111161"}}}],"episodes":[],"pagination":{"has_more":true}}`))
}))
defer server.Close()
p := NewProvider(server.Client(), server.URL)
rows, err := p.FetchWatched(context.Background(), watchsync.ServerConfig{}, watchsync.Connection{AccessToken: "k"})
if err != nil {
t.Fatalf("fetch watched: %v", err)
}
if len(rows) != 1 {
t.Fatalf("got %d rows, want 1", len(rows))
}
if len(offsets) != 2 || offsets[0] != "" || offsets[1] != "1" {
t.Fatalf("unexpected offsets: %#v", offsets)
}
}
func TestFetchProgressParsesFlatArray(t *testing.T) {
pausedAt := time.Date(2025, time.November, 1, 8, 30, 0, 0, time.UTC)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -241,6 +292,59 @@ func TestExportHistorySendsBulkPayload(t *testing.T) {
}
}
func TestExportHistoryDoesNotClaimNotFoundBatchWasSent(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"added":{"movies":[]},"updated":{},"not_found":{"movies":[{"ids":{"imdb":"tt-missing"}}]}}`))
}))
defer server.Close()
p := NewProvider(server.Client(), server.URL)
plays := []watchsync.LocalPlay{{HistoryID: "h1", Kind: historyimport.KindMovie, IMDbID: "tt-missing"}}
result, err := p.ExportHistory(context.Background(), watchsync.ServerConfig{}, watchsync.Connection{AccessToken: "k"}, plays)
if err != nil {
t.Fatalf("export: %v", err)
}
if len(result.Sent) != 0 || result.Failed["h1"] == "" {
t.Fatalf("unexpected result: %#v", result)
}
}
func TestExportHistoryTreatsEmptyNotFoundBucketsAsSuccess(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"added":{"movies":1},"not_found":{"movies":[],"episodes":[]}}`))
}))
defer server.Close()
p := NewProvider(server.Client(), server.URL)
plays := []watchsync.LocalPlay{{HistoryID: "h1", Kind: historyimport.KindMovie, IMDbID: "tt0111161"}}
result, err := p.ExportHistory(context.Background(), watchsync.ServerConfig{}, watchsync.Connection{AccessToken: "k"}, plays)
if err != nil {
t.Fatalf("export: %v", err)
}
if len(result.Sent) != 1 || len(result.Failed) != 0 {
t.Fatalf("unexpected result: %#v", result)
}
}
func TestExportHistoryReportsUnsupportedItemWithoutCallingMDBList(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
t.Fatal("MDBList should not be called for an unsupported item")
}))
defer server.Close()
p := NewProvider(server.Client(), server.URL)
plays := []watchsync.LocalPlay{{HistoryID: "h1", Kind: historyimport.KindMovie}}
result, err := p.ExportHistory(context.Background(), watchsync.ServerConfig{}, watchsync.Connection{AccessToken: "k"}, plays)
if err != nil {
t.Fatalf("export: %v", err)
}
if len(result.Sent) != 0 || result.Failed["h1"] == "" {
t.Fatalf("unexpected result: %#v", result)
}
}
func TestScrobbleStartUsesEpisodeShape(t *testing.T) {
var gotPath string
var gotBody map[string]any
@@ -270,16 +374,52 @@ func TestScrobbleStartUsesEpisodeShape(t *testing.T) {
if progress, _ := gotBody["progress"].(float64); progress != 25 {
t.Fatalf("expected progress 25, got %v", gotBody["progress"])
}
if season, _ := gotBody["season"].(float64); int(season) != 3 {
t.Fatalf("expected season 3, got %v", gotBody["season"])
}
show, _ := gotBody["show"].(map[string]any)
season, _ := show["season"].(map[string]any)
if number, _ := season["number"].(float64); int(number) != 3 {
t.Fatalf("expected show season number 3, got %v", season["number"])
}
episode, _ := season["episode"].(map[string]any)
if number, _ := episode["number"].(float64); int(number) != 7 {
t.Fatalf("expected show episode number 7, got %v", episode["number"])
}
if _, exists := gotBody["season"]; exists {
t.Fatalf("season must be nested under show: %#v", gotBody)
}
if _, exists := show["episode"]; exists {
t.Fatalf("episode must be nested under show season: %#v", show)
}
ids, _ := show["ids"].(map[string]any)
if ids["imdb"] != "tt0903747" {
t.Fatalf("expected show imdb tt0903747, got %v", ids["imdb"])
}
}
func TestScrobbleRoundsProgressToMDBListPrecision(t *testing.T) {
var gotBody map[string]any
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if err := json.NewDecoder(r.Body).Decode(&gotBody); err != nil {
t.Fatalf("decode request: %v", err)
}
w.WriteHeader(http.StatusCreated)
}))
defer server.Close()
p := NewProvider(server.Client(), server.URL)
err := p.Start(context.Background(), watchsync.ServerConfig{}, watchsync.Connection{AccessToken: "k"}, watchsync.ScrobbleEvent{
Kind: historyimport.KindMovie,
TMDBID: "950387",
PositionSeconds: 611.8,
DurationSeconds: 2163.4,
})
if err != nil {
t.Fatalf("scrobble start: %v", err)
}
if got := gotBody["progress"]; got != 28.28 {
t.Fatalf("progress = %v, want 28.28", got)
}
}
func TestExportWatchlistSendsToWatchlistAdd(t *testing.T) {
var gotPath string
var gotBody map[string]any
@@ -288,7 +428,7 @@ func TestExportWatchlistSendsToWatchlistAdd(t *testing.T) {
body, _ := io.ReadAll(r.Body)
_ = json.Unmarshal(body, &gotBody)
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte(`{"added":1,"existing":0,"not_found":0}`))
_, _ = w.Write([]byte(`{"added":{"movies":1,"shows":1},"existing":{"movies":0,"shows":0},"not_found":{"movies":0,"shows":0}}`))
}))
defer server.Close()
@@ -315,6 +455,110 @@ func TestExportWatchlistSendsToWatchlistAdd(t *testing.T) {
}
}
func TestExportWatchlistDoesNotClaimNotFoundBatchWasSent(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"added":{"movies":0,"shows":0},"existing":{"movies":0,"shows":0},"not_found":{"movies":1,"shows":0}}`))
}))
defer server.Close()
p := NewProvider(server.Client(), server.URL)
favorites := []watchsync.LocalFavorite{{MediaItemID: "m1", Kind: historyimport.KindMovie, IMDbID: "tt0111161"}}
result, err := p.ExportWatchlist(context.Background(), watchsync.ServerConfig{}, watchsync.Connection{AccessToken: "k"}, favorites)
if err != nil {
t.Fatalf("export watchlist: %v", err)
}
if len(result.Sent) != 0 || result.Failed["m1"] == "" {
t.Fatalf("unexpected result: %#v", result)
}
}
func TestRemoveWatchlistTreatsNotFoundBatchAsReconciled(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/watchlist/items/remove" {
t.Fatalf("path = %q, want /watchlist/items/remove", r.URL.Path)
}
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"added":{"movies":0,"shows":0},"existing":{"movies":0,"shows":0},"not_found":{"movies":1,"shows":0}}`))
}))
defer server.Close()
p := NewProvider(server.Client(), server.URL)
favorites := []watchsync.LocalFavorite{{MediaItemID: "m1", Kind: historyimport.KindMovie, IMDbID: "tt0111161"}}
result, err := p.RemoveWatchlist(context.Background(), watchsync.ServerConfig{}, watchsync.Connection{AccessToken: "k"}, favorites)
if err != nil {
t.Fatalf("remove watchlist: %v", err)
}
if len(result.NotFound) != 1 || result.NotFound[0] != "m1" || len(result.Sent) != 0 || len(result.Failed) != 0 {
t.Fatalf("unexpected result: %#v", result)
}
}
func TestExportWatchlistReportsUnsupportedItemWithoutCallingMDBList(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
t.Fatal("MDBList should not be called for an unsupported item")
}))
defer server.Close()
p := NewProvider(server.Client(), server.URL)
favorites := []watchsync.LocalFavorite{{MediaItemID: "m1", Kind: historyimport.KindMovie}}
result, err := p.ExportWatchlist(context.Background(), watchsync.ServerConfig{}, watchsync.Connection{AccessToken: "k"}, favorites)
if err != nil {
t.Fatalf("export watchlist: %v", err)
}
if len(result.Sent) != 0 || result.Failed["m1"] == "" {
t.Fatalf("unexpected result: %#v", result)
}
}
func TestFetchWatchlistUsesCursorPagination(t *testing.T) {
var cursors []string
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
cursors = append(cursors, r.URL.Query().Get("cursor"))
w.Header().Set("Content-Type", "application/json")
if r.URL.Query().Get("cursor") == "watch-next" {
_, _ = w.Write([]byte(`{"movies":[],"shows":[],"pagination":{"next_cursor":null}}`))
return
}
_, _ = w.Write([]byte(`{"movies":[],"shows":[],"pagination":{"next_cursor":"watch-next"}}`))
}))
defer server.Close()
p := NewProvider(server.Client(), server.URL)
if _, err := p.FetchWatchlist(context.Background(), watchsync.ServerConfig{}, watchsync.Connection{AccessToken: "k"}); err != nil {
t.Fatalf("fetch watchlist: %v", err)
}
if len(cursors) != 2 || cursors[0] != "" || cursors[1] != "watch-next" {
t.Fatalf("unexpected cursors: %#v", cursors)
}
}
func TestFetchWatchlistContinuesOffsetPaginationWhenHasMore(t *testing.T) {
var offsets []string
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
offsets = append(offsets, r.URL.Query().Get("offset"))
w.Header().Set("Content-Type", "application/json")
if r.URL.Query().Get("offset") == "1" {
_, _ = w.Write([]byte(`{"movies":[],"shows":[],"pagination":{"has_more":false}}`))
return
}
_, _ = w.Write([]byte(`{"movies":[{"title":"The Shawshank Redemption","release_year":1994,"ids":{"imdb":"tt0111161"}}],"shows":[],"pagination":{"has_more":true}}`))
}))
defer server.Close()
p := NewProvider(server.Client(), server.URL)
rows, err := p.FetchWatchlist(context.Background(), watchsync.ServerConfig{}, watchsync.Connection{AccessToken: "k"})
if err != nil {
t.Fatalf("fetch watchlist: %v", err)
}
if len(rows) != 1 {
t.Fatalf("got %d rows, want 1", len(rows))
}
if len(offsets) != 2 || offsets[0] != "" || offsets[1] != "1" {
t.Fatalf("unexpected offsets: %#v", offsets)
}
}
func TestDoRetriesShortRateLimitInPlace(t *testing.T) {
attempts := 0
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -411,6 +655,21 @@ func TestDoReplaysBodyOnRateLimitRetry(t *testing.T) {
}
}
func TestDoIncludesMDBListValidationError(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusBadRequest)
_, _ = w.Write([]byte(`{"error":{"progress":["Ensure that there are no more than 5 digits in total."]}}`))
}))
defer server.Close()
p := NewProvider(server.Client(), server.URL)
err := p.do(context.Background(), http.MethodPost, "/scrobble/start", "key", strings.NewReader(`{}`), nil)
if err == nil || !strings.Contains(err.Error(), "no more than 5 digits") {
t.Fatalf("expected validation detail, got %v", err)
}
}
func TestParseRetryAfter(t *testing.T) {
now := time.Date(2026, time.July, 5, 12, 0, 0, 0, time.UTC)
cases := []struct {
+8
View File
@@ -26,6 +26,14 @@ export function formatActiveScanMode(scan: Pick<ScanRun, "mode">) {
}
}
export function formatActiveScanTarget(scan: Pick<ScanRun, "path">) {
if (!scan.path) {
return "Entire library";
}
const segments = scan.path.replace(/\/+$/, "").split("/");
return segments[segments.length - 1] || scan.path;
}
export function formatActiveScanTrigger(trigger: string) {
switch (trigger) {
case "autoscan":
+3 -1
View File
@@ -299,7 +299,9 @@ describe("AdminLibraries", () => {
const markup = renderPage();
expect(markup).toContain("Processing files · 10 / 20 (50%)");
expect(markup).toContain("Full library scan · Entire library");
// Mode and target render as separate compact-row elements.
expect(markup).toContain("Full library scan");
expect(markup).toContain("Entire library");
});
it("renders the collapsed Ambiguous Roots section when no roots exist", () => {
+232 -112
View File
@@ -95,6 +95,7 @@ import {
compareActiveScans,
formatActiveScanMode,
formatActiveScanProgress,
formatActiveScanTarget,
formatActiveScanTime,
formatActiveScanTrigger,
} from "@/lib/scanRuns";
@@ -789,89 +790,14 @@ function ScanQueuePopover({
{/* Library groups */}
<div className="max-h-[360px] overflow-y-auto px-3 py-2">
{groups.map((group, gi) => (
<div key={group.libraryID}>
{/* Library header row */}
<div className="flex items-center gap-2 py-1.5">
<span className="text-muted-foreground/60 text-[10px] font-semibold tracking-widest uppercase">
{getLibraryScanGroupName(group.library, group.libraryID)}
</span>
<div className="bg-border/25 h-px flex-1" />
<Button
variant="ghost"
size="sm"
className="text-muted-foreground hover:text-destructive h-6 gap-1 px-2 text-[10px]"
disabled={cancellingLibraryID === group.libraryID}
onClick={() => onCancel(group.libraryID)}
>
<Square className="h-2.5 w-2.5" />
Cancel
</Button>
</div>
{/* Scan rows */}
{group.scans.map((scan, si) => (
<div
key={scan.id}
className={cn(
"flex items-start gap-2.5 rounded-lg px-2.5 py-2 transition-colors",
scan.status === "running" ? "bg-primary/[0.04]" : "",
)}
style={{
animation: "fade-in 0.25s ease-out backwards",
animationDelay: `${(gi * 4 + si) * 40}ms`,
}}
>
{/* Status dot */}
<div className="relative mt-[5px] flex h-2.5 w-2.5 shrink-0 items-center justify-center">
{scan.status === "running" ? (
<>
<span className="bg-success/20 absolute h-2.5 w-2.5 animate-ping rounded-full" />
<span className="bg-success shadow-success/30 relative h-[7px] w-[7px] rounded-full shadow-[0_0_5px_1px]" />
</>
) : (
<span className="bg-muted-foreground/20 ring-muted-foreground/15 h-[5px] w-[5px] rounded-full ring-[1.5px]" />
)}
</div>
{/* Details */}
<div className="min-w-0 flex-1">
<div className="flex items-baseline gap-1.5">
<span className="text-xs leading-snug font-medium">
{formatActiveScanMode(scan)}
</span>
{scan.trigger && (
<span className="bg-muted/50 text-muted-foreground rounded px-1 py-px text-[9px] font-medium">
{formatActiveScanTrigger(scan.trigger)}
</span>
)}
</div>
<div className="text-muted-foreground mt-px flex items-center gap-1 text-[10px] leading-relaxed">
{scan.path ? (
<code className="text-muted-foreground/70 max-w-[200px] truncate font-mono">
{scan.path}
</code>
) : (
<span>Entire library</span>
)}
<span className="text-border/50 shrink-0">·</span>
<span className="shrink-0">
{scan.status === "running"
? formatActiveScanTime(scan.started_at, "Started")
: "Waiting for capacity"}
</span>
</div>
{formatActiveScanProgress(scan) && (
<div className="text-muted-foreground/80 mt-1 truncate text-[10px] leading-relaxed">
{formatActiveScanProgress(scan)}
</div>
)}
</div>
</div>
))}
{/* Separator between groups */}
{gi < groups.length - 1 && <div className="bg-border/20 my-1 h-px" />}
</div>
<ScanQueueGroup
key={group.libraryID}
group={group}
groupIndex={gi}
isLast={gi === groups.length - 1}
cancelling={cancellingLibraryID === group.libraryID}
onCancel={onCancel}
/>
))}
</div>
</PopoverContent>
@@ -879,6 +805,146 @@ function ScanQueuePopover({
);
}
// How many scan rows a group shows before the rest collapse behind an expander.
const COLLAPSED_SCAN_ROW_LIMIT = 4;
function useCollapsedScans(scans: ScanRun[]) {
const [expanded, setExpanded] = useState(false);
const visibleScans = expanded ? scans : scans.slice(0, COLLAPSED_SCAN_ROW_LIMIT);
return {
expanded,
setExpanded,
visibleScans,
hiddenCount: scans.length - visibleScans.length,
collapsible: scans.length > COLLAPSED_SCAN_ROW_LIMIT,
};
}
function ScanQueueGroup({
group,
groupIndex,
isLast,
cancelling,
onCancel,
}: {
group: {
libraryID: number;
library: Library | null;
scans: ScanRun[];
runningCount: number;
queuedCount: number;
};
groupIndex: number;
isLast: boolean;
cancelling: boolean;
onCancel: (libraryID: number) => void;
}) {
const { expanded, setExpanded, visibleScans, hiddenCount, collapsible } = useCollapsedScans(
group.scans,
);
return (
<div>
{/* Library header row */}
<div className="flex items-center gap-2 py-1.5">
<span className="text-muted-foreground/60 text-[10px] font-semibold tracking-widest uppercase">
{getLibraryScanGroupName(group.library, group.libraryID)}
</span>
<div className="bg-border/25 h-px flex-1" />
<Button
variant="ghost"
size="sm"
className="text-muted-foreground hover:text-destructive h-6 gap-1 px-2 text-[10px]"
disabled={cancelling}
onClick={() => onCancel(group.libraryID)}
>
<Square className="h-2.5 w-2.5" />
Cancel
</Button>
</div>
{/* Scan rows */}
{visibleScans.map((scan, si) => (
<div
key={scan.id}
className={cn(
"flex items-start gap-2.5 rounded-lg px-2.5 py-2 transition-colors",
scan.status === "running" ? "bg-primary/[0.04]" : "",
)}
style={{
animation: "fade-in 0.25s ease-out backwards",
animationDelay: `${Math.min(groupIndex * 4 + si, 12) * 40}ms`,
}}
>
{/* Status dot */}
<div className="relative mt-[5px] flex h-2.5 w-2.5 shrink-0 items-center justify-center">
{scan.status === "running" ? (
<>
<span className="bg-success/20 absolute h-2.5 w-2.5 animate-ping rounded-full" />
<span className="bg-success shadow-success/30 relative h-[7px] w-[7px] rounded-full shadow-[0_0_5px_1px]" />
</>
) : (
<span className="bg-muted-foreground/20 ring-muted-foreground/15 h-[5px] w-[5px] rounded-full ring-[1.5px]" />
)}
</div>
{/* Details */}
<div className="min-w-0 flex-1">
<div className="flex items-baseline gap-1.5">
<span className="text-xs leading-snug font-medium">{formatActiveScanMode(scan)}</span>
{scan.trigger && (
<span className="bg-muted/50 text-muted-foreground rounded px-1 py-px text-[9px] font-medium">
{formatActiveScanTrigger(scan.trigger)}
</span>
)}
</div>
<div className="text-muted-foreground mt-px flex items-center gap-1 text-[10px] leading-relaxed">
<code
className="text-muted-foreground/70 max-w-[200px] truncate font-mono"
title={scan.path}
>
{formatActiveScanTarget(scan)}
</code>
<span className="text-border/50 shrink-0">·</span>
<span className="shrink-0">
{scan.status === "running"
? formatActiveScanTime(scan.started_at, "Started")
: "Waiting for capacity"}
</span>
</div>
{formatActiveScanProgress(scan) && (
<div className="text-muted-foreground/80 mt-1 truncate text-[10px] leading-relaxed">
{formatActiveScanProgress(scan)}
</div>
)}
</div>
</div>
))}
{collapsible ? (
<button
type="button"
className="text-muted-foreground hover:text-foreground flex items-center gap-1 rounded-lg px-2.5 py-1.5 text-[10px] transition-colors"
onClick={() => setExpanded((current) => !current)}
>
{expanded ? (
<>
Show less <ChevronUp className="h-3 w-3" />
</>
) : (
<>
+ {hiddenCount} more{hiddenCount <= group.queuedCount ? " queued" : ""}{" "}
<ChevronDown className="h-3 w-3" />
</>
)}
</button>
) : null}
{/* Separator between groups */}
{!isLast && <div className="bg-border/20 my-1 h-px" />}
</div>
);
}
function getLibraryScanGroupName(library: Library | null, libraryID: number) {
return library?.name ?? `Library #${libraryID}`;
}
@@ -954,57 +1020,111 @@ function LibraryActiveWorkRow({
</div>
</div>
) : null}
{activeLibraryScans.map((scan) => (
<LibraryScanTaskRow
key={scan.id}
scan={scan}
{activeLibraryScans.length > 0 ? (
<LibraryScanTasks
scans={activeLibraryScans}
cancelling={cancellingLibraryID === libraryID}
libraryID={libraryID}
onCancelScans={onCancelScans}
/>
))}
) : null}
</div>
</TableCell>
</TableRow>
);
}
function LibraryScanTaskRow({
scan,
function LibraryScanTasks({
scans,
cancelling,
libraryID,
onCancelScans,
}: {
scan: ScanRun;
scans: ScanRun[];
cancelling: boolean;
libraryID: number;
onCancelScans: (libraryID: number) => void;
}) {
const progress = formatActiveScanProgress(scan);
const { expanded, setExpanded, visibleScans, collapsible } = useCollapsedScans(scans);
const runningCount = scans.filter((scan) => scan.status === "running").length;
const queuedCount = scans.length - runningCount;
return (
<div className="flex min-w-0 items-start gap-1.5">
<StopTaskButton
disabled={cancelling}
label="Cancel library scans"
onClick={() => onCancelScans(libraryID)}
/>
<RefreshCw
className={cn("mt-0.5 h-3 w-3 shrink-0", scan.status === "running" && "animate-spin")}
/>
<div className="min-w-0 flex-1 space-y-0.5">
<div className="flex min-w-0 items-center gap-1.5">
<span className="text-foreground/80 font-medium">Scan</span>
<span className="shrink-0">{scan.status === "running" ? "Running" : "Queued"}</span>
</div>
<div className="text-muted-foreground/80 truncate text-[10px]">
{formatActiveScanMode(scan)}
{scan.path ? ` · ${scan.path}` : " · Entire library"}
</div>
{progress ? (
<div className="text-muted-foreground/80 truncate text-[10px]">{progress}</div>
<div className="flex min-w-0 flex-col gap-1">
<div className="flex min-w-0 items-center gap-1.5">
<StopTaskButton
disabled={cancelling}
label="Cancel library scans"
onClick={() => onCancelScans(libraryID)}
/>
<RefreshCw
className={cn("h-3 w-3 shrink-0", runningCount > 0 && "animate-spin")}
style={{ animationDuration: "2s" }}
/>
<span className="text-foreground/80 font-medium">Scan</span>
<span className="tabular-nums">
{runningCount > 0 ? `${runningCount} running` : null}
{runningCount > 0 && queuedCount > 0 ? " · " : null}
{queuedCount > 0 ? `${queuedCount} queued` : null}
</span>
{collapsible ? (
<button
type="button"
className="text-muted-foreground hover:text-foreground flex items-center gap-0.5 rounded px-1 text-[10px] transition-colors"
onClick={() => setExpanded((current) => !current)}
>
{expanded ? (
<>
Show less <ChevronUp className="h-3 w-3" />
</>
) : (
<>
Show all {scans.length} <ChevronDown className="h-3 w-3" />
</>
)}
</button>
) : null}
</div>
<div
className={cn(
"flex min-w-0 flex-col gap-0.5 pl-[3.375rem]",
expanded && "max-h-56 overflow-y-auto",
)}
>
{visibleScans.map((scan) => (
<CompactScanRow key={scan.id} scan={scan} />
))}
</div>
</div>
);
}
function CompactScanRow({ scan }: { scan: ScanRun }) {
const progress = scan.status === "running" ? formatActiveScanProgress(scan) : "";
return (
<div className="flex min-w-0 items-center gap-1.5" title={scan.path}>
<span
className={cn(
"h-1.5 w-1.5 shrink-0 rounded-full",
scan.status === "running" ? "bg-success" : "bg-muted-foreground/30",
)}
/>
{scan.mode !== "file" ? (
<span className="shrink-0 text-[10px]">{formatActiveScanMode(scan)}</span>
) : null}
{scan.path ? (
<code className="text-muted-foreground/80 truncate font-mono text-[10px]">
{formatActiveScanTarget(scan)}
</code>
) : (
<span className="text-muted-foreground/80 truncate text-[10px]">
{formatActiveScanTarget(scan)}
</span>
)}
{progress ? (
<span className="text-muted-foreground/60 truncate text-[10px]">· {progress}</span>
) : null}
</div>
);
}
+56 -8
View File
@@ -34,6 +34,7 @@ import type {
} from "../realtime-protocol";
import { resolvePendingSeekTime } from "../utils/pendingSeek";
import { resolveVersionAudioLanguage } from "../utils/effectiveAudioLanguage";
import { HlsStartupGuard } from "../utils/hlsStartupGuard";
import { normalizeSubtitleMode } from "../utils/subtitleMode";
import type {
PlaybackExitState,
@@ -42,6 +43,7 @@ import type {
PlayerPlaybackStateChange,
PlayerPlaybackTransport,
PlaybackSessionPlaybackInfo,
PlaybackTransportRestart,
PlayerAudioTrack,
PlayerChapter,
PlayMethod,
@@ -83,6 +85,7 @@ interface VideoPlayerProps {
onSwitchVersion?: (fileId: number, currentPosition: number) => void;
subtitleUrls: PlayerSubtitleInfo[];
initialPosition: number;
transportRestart?: PlaybackTransportRestart | null;
preferredSubtitleLanguage?: string | null;
preferredSubtitleTrackSignature?: PlayerSubtitleTrackSignature | null;
subtitleMode?: SubtitleMode;
@@ -173,6 +176,7 @@ export function VideoPlayer({
onSwitchVersion,
subtitleUrls,
initialPosition,
transportRestart,
preferredSubtitleLanguage,
preferredSubtitleTrackSignature,
subtitleMode,
@@ -218,6 +222,7 @@ export function VideoPlayer({
const containerRef = useRef<HTMLDivElement>(null);
const isMountedRef = useRef(true);
const hlsRef = useRef<HlsType | null>(null);
const hlsStartupGuardRef = useRef<HlsStartupGuard | null>(null);
const mediaRecoveryAttemptsRef = useRef(0);
const lastRecoveryRef = useRef(0);
const streamOriginRef = useRef(0);
@@ -335,7 +340,41 @@ export function VideoPlayer({
playMethod,
initialPosition,
qualityPreference,
transportRestart,
});
const { cancelPendingTranscodeStart, startupGeneration } = transcodeQuality;
const hlsStartupExpected =
transcodeQuality.isTranscoding || transcodeQuality.transcodeStreamUrl !== null;
const failHlsStartup = useCallback(() => {
console.error("[hls.js] Playback startup timed out or exhausted recovery attempts");
cancelPendingTranscodeStart();
const activeHls = hlsRef.current;
hlsRef.current = null;
activeHls?.destroy();
const video = videoRef.current;
if (video) {
video.removeAttribute("src");
video.load();
}
setError("Playback failed. The media could not be loaded.");
}, [cancelPendingTranscodeStart]);
useEffect(() => {
if (!hlsStartupExpected || startupGeneration === 0) return;
const guard = new HlsStartupGuard(failHlsStartup);
hlsStartupGuardRef.current = guard;
return () => {
guard.dispose();
if (hlsStartupGuardRef.current === guard) {
hlsStartupGuardRef.current = null;
}
};
}, [failHlsStartup, hlsStartupExpected, startupGeneration]);
// Derive effective stream URL and play method.
// Both transcode and remux go through HLS, so treat them as "transcode" for the player.
@@ -1238,7 +1277,7 @@ export function VideoPlayer({
// -- hls.js lifecycle --
useEffect(() => {
const video = videoRef.current;
if (!video || !isPlayerReady) return;
if (!video || !isPlayerReady || hlsStartupGuardRef.current?.hasFailed()) return;
let hls: HlsType | null = null;
let destroyed = false;
@@ -1251,10 +1290,11 @@ export function VideoPlayer({
const cleanupStartupListeners = () => {
video.removeEventListener("loadeddata", attemptAutoplayWhenReady);
video.removeEventListener("canplay", attemptAutoplayWhenReady);
video.removeEventListener("loadedmetadata", attemptAutoplayWhenReady);
};
const attemptAutoplayWhenReady = () => {
if (destroyed || autoplayStarted) return;
if (destroyed || autoplayStarted || hlsStartupGuardRef.current?.hasFailed()) return;
// HAVE_FUTURE_DATA means the browser has enough media to advance beyond
// the current frame. Starting earlier can produce a visible first-frame
// freeze where audio advances before video begins moving.
@@ -1273,7 +1313,7 @@ export function VideoPlayer({
if (effectivePlayMethod === "transcode") {
try {
const Hls = await hlsPromise;
if (destroyed) return;
if (destroyed || hlsStartupGuardRef.current?.hasFailed()) return;
if (Hls.isSupported()) {
const maxBufferLength = selectedVersionBitrate >= 25000 ? 60 : 120;
@@ -1317,8 +1357,12 @@ export function VideoPlayer({
lastRecoveryRef.current = now;
if (data.type === Hls.ErrorTypes.NETWORK_ERROR) {
console.warn("[hls.js] Fatal network error, attempting recovery...");
hls?.startLoad();
if (hlsStartupGuardRef.current?.handleFatalNetworkError() ?? true) {
console.warn("[hls.js] Fatal network error, attempting recovery...");
hls?.startLoad();
} else {
console.error("[hls.js] Fatal startup network error, giving up");
}
} else if (data.type === Hls.ErrorTypes.MEDIA_ERROR) {
if (mediaRecoveryAttemptsRef.current === 0) {
console.warn("[hls.js] Fatal media error, attempting recovery...");
@@ -1411,6 +1455,10 @@ export function VideoPlayer({
}
setBuffering(false);
};
const markPlaybackStarted = () => {
hlsStartupGuardRef.current?.markPlaybackStarted();
setAwaitingFirstFrame(false);
};
const onTimeUpdate = () => {
const nextTime = toMediaTime(video.currentTime, streamOriginRef.current);
const resolved = resolvePendingSeekTime(nextTime, pendingSeekTime);
@@ -1421,13 +1469,13 @@ export function VideoPlayer({
// timeupdate is the most reliable signal that frames are rendering.
// Also clears any stale buffering state from HLS segment transitions
// where `waiting` fired but `canplay`/`playing` never followed.
setAwaitingFirstFrame(false);
markPlaybackStarted();
clearBuffering();
};
const onSeeked = () => {
setPendingSeekTime(null);
setCurrentTime(toMediaTime(video.currentTime, streamOriginRef.current));
setAwaitingFirstFrame(false);
markPlaybackStarted();
clearBuffering();
if (roomSyncWaiting && watchTogetherSync.attachedSessionId === sessionId) {
watchTogetherSync.reportReady();
@@ -1469,7 +1517,7 @@ export function VideoPlayer({
};
const onPlaying = () => {
clearBuffering();
setAwaitingFirstFrame(false);
markPlaybackStarted();
};
const onStalled = () => {
if (watchTogetherRoomActive && watchTogetherSync.attachedSessionId === sessionId) {
+1
View File
@@ -460,6 +460,7 @@ export function WatchPage({
onSwitchVersion={handleSwitchVersion}
subtitleUrls={playableSubtitles}
initialPosition={session.initialPosition}
transportRestart={session.transportRestart}
preferredSubtitleLanguage={preferredSubtitleLanguage}
preferredSubtitleTrackSignature={preferredSubtitleTrackSignature}
subtitleMode={subtitleMode}
+21 -7
View File
@@ -14,6 +14,7 @@ import type {
ChangeAudioResponse,
PlaybackSessionPlaybackInfo,
PlaybackSessionResponse,
PlaybackTransportRestart,
PlayMethod,
PlayerFileVersion,
PlayerPlaybackVariant,
@@ -31,6 +32,7 @@ interface PlaybackSessionState {
durationSeconds: number | null;
subtitleUrls: PlayerSubtitleInfo[];
playbackInfo: PlaybackSessionPlaybackInfo | null;
transportRestart: PlaybackTransportRestart | null;
loading: boolean;
replacing: boolean;
errorTitle: string | null;
@@ -215,6 +217,7 @@ export function usePlaybackSession(
durationSeconds: null,
subtitleUrls: [],
playbackInfo: null,
transportRestart: null,
loading: true,
replacing: false,
errorTitle: null,
@@ -222,6 +225,7 @@ export function usePlaybackSession(
});
const sessionIdRef = useRef<string | null>(null);
const audioSwitchRevisionRef = useRef(0);
const stateRef = useRef(state);
const activeRequestKeyRef = useRef<string | null>(null);
const switchingRef = useRef(false);
@@ -304,6 +308,7 @@ export function usePlaybackSession(
: undefined,
})),
playbackInfo: session.playback_info ?? null,
transportRestart: null,
loading: false,
replacing: false,
errorTitle: null,
@@ -538,19 +543,28 @@ export function usePlaybackSession(
});
const token = config.getAccessToken();
const nextStreamUrl = buildPlayerStreamUrl(
config.apiBaseUrl,
resp.stream_url,
token,
resp.play_method,
currentPosition,
);
audioSwitchRevisionRef.current += 1;
setState((prev) => ({
...prev,
streamUrl: buildPlayerStreamUrl(
config.apiBaseUrl,
resp.stream_url,
token,
resp.play_method,
currentPosition,
),
streamUrl: nextStreamUrl,
playMethod: resp.play_method,
audioTrackIndex: resp.audio_track_index,
playbackInfo: resp.playback_info ?? prev.playbackInfo,
initialPosition: currentPosition,
transportRestart: {
revision: audioSwitchRevisionRef.current,
streamUrl: nextStreamUrl,
playerStartSeconds: resp.player_start_seconds ?? currentPosition,
streamOriginSeconds: resp.stream_origin_seconds ?? resp.timeline_offset_seconds ?? 0,
canSeekAnywhere: resp.can_seek_anywhere ?? true,
},
}));
} catch (err) {
console.error("Failed to switch audio track:", err);
@@ -28,7 +28,7 @@ function wrapper({ children }: { children: ReactNode }) {
return <PlayerConfigProvider config={config}>{children}</PlayerConfigProvider>;
}
function transcodeStartResponse() {
function transcodeStartResponse(overrides: Record<string, unknown> = {}) {
return {
ok: true,
status: 200,
@@ -40,6 +40,7 @@ function transcodeStartResponse() {
player_start_seconds: 0,
timeline_offset_seconds: 0,
can_seek_anywhere: true,
...overrides,
}),
};
}
@@ -75,6 +76,70 @@ function renderQuality() {
}
describe("useTranscodeQuality", () => {
it("adopts an audio-switch transport without starting another transcode", async () => {
const { result } = renderHook(
() =>
useTranscodeQuality({
sessionId: "sess-1",
selectedVersion: version,
versions: [version],
playMethod: "remux",
initialPosition: 100,
transportRestart: {
revision: 1,
streamUrl: "/api/v1/playback/transcode/sess-1/master.m3u8?st=token",
playerStartSeconds: 4,
streamOriginSeconds: 96,
canSeekAnywhere: false,
},
}),
{ wrapper },
);
await waitFor(() =>
expect(result.current.transcodeStreamUrl).toContain("master.m3u8?st=token"),
);
expect(result.current.playerStartSeconds).toBe(4);
expect(result.current.streamOriginSeconds).toBe(96);
expect(result.current.canSeekAnywhere).toBe(false);
await new Promise((resolve) => setTimeout(resolve, 20));
expect(fetchMock).not.toHaveBeenCalled();
});
it("preserves video copy when auto-starting a resumed remux", async () => {
fetchMock.mockImplementationOnce(() =>
Promise.resolve(
transcodeStartResponse({
player_start_seconds: 2.261,
stream_origin_seconds: 16,
timeline_offset_seconds: 16,
can_seek_anywhere: false,
}),
),
);
const { result } = renderHook(
() =>
useTranscodeQuality({
sessionId: "sess-1",
selectedVersion: version,
versions: [version],
playMethod: "remux",
initialPosition: 478.25,
}),
{ wrapper },
);
await waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(1));
const body = sentBodies()[0]!;
expect(body.seek_seconds).toBe(478.25);
expect(body.target_codec_video).toBe("copy");
expect(body.target_resolution).toBe("");
expect(body.target_bitrate_kbps).toBe(0);
await waitFor(() => expect(result.current.streamOriginSeconds).toBe(16));
expect(result.current.playerStartSeconds).toBe(2.261);
expect(result.current.canSeekAnywhere).toBe(false);
});
it("coalesces same-tick restarts into a single start with the final params", async () => {
const { result } = renderQuality();
@@ -150,6 +215,34 @@ describe("useTranscodeQuality", () => {
expect(fetchMock).not.toHaveBeenCalled();
});
it("cancels an in-flight startup before a manifest arrives", async () => {
let requestSignal: AbortSignal | undefined;
fetchMock.mockImplementationOnce((_, init: RequestInit) => {
requestSignal = init.signal as AbortSignal;
return new Promise((_, reject) => {
requestSignal?.addEventListener(
"abort",
() => reject(new DOMException("Aborted", "AbortError")),
{ once: true },
);
});
});
const { result } = renderQuality();
await waitFor(() => expect(fetchMock).toHaveBeenCalledOnce());
expect(result.current.startupGeneration).toBe(1);
expect(result.current.isTranscoding).toBe(true);
act(() => {
result.current.cancelPendingTranscodeStart();
});
expect(requestSignal?.aborted).toBe(true);
await waitFor(() => expect(result.current.isTranscoding).toBe(false));
expect(result.current.error).toBeNull();
});
it("rolls back a failed burn-in selection so the same track can be retried", async () => {
const { result } = renderQuality();
+46 -2
View File
@@ -2,7 +2,13 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react";
import { usePlayerConfig } from "../context/PlayerConfigContext";
import { PlayerFetchError, playerFetch } from "../player-fetch";
import { describeTranscodingPolicyError } from "../playback-errors";
import type { PlayMethod, PlayerFileVersion, QualityOption, TranscodeStartRequest } from "../types";
import type {
PlaybackTransportRestart,
PlayMethod,
PlayerFileVersion,
QualityOption,
TranscodeStartRequest,
} from "../types";
import { QUALITY_TO_RESOLUTION } from "./useCodecDetection";
/** Quality tier definition. ID is frontend-only; backend receives resolution + bitrate separately. */
@@ -51,6 +57,7 @@ interface UseTranscodeQualityParams {
playMethod: PlayMethod | null;
initialPosition: number;
qualityPreference?: string | null;
transportRestart?: PlaybackTransportRestart | null;
}
interface UseTranscodeQualityResult {
@@ -76,6 +83,10 @@ interface UseTranscodeQualityResult {
canSeekAnywhere: boolean;
durationSeconds: number | null;
isTranscoding: boolean;
/** Increments when a user-visible HLS startup attempt begins. */
startupGeneration: number;
/** Cancels a transcode start that has not produced playable media yet. */
cancelPendingTranscodeStart: () => void;
error: string | null;
switchedFileId: number | null;
effectiveVersion: PlayerFileVersion | undefined;
@@ -179,6 +190,7 @@ export function useTranscodeQuality({
playMethod,
initialPosition,
qualityPreference,
transportRestart,
}: UseTranscodeQualityParams): UseTranscodeQualityResult {
const config = usePlayerConfig();
const [activeQualityId, setActiveQualityId] = useState("original");
@@ -188,6 +200,7 @@ export function useTranscodeQuality({
const [canSeekAnywhere, setCanSeekAnywhere] = useState(true);
const [durationSeconds, setDurationSeconds] = useState<number | null>(null);
const [isTranscoding, setIsTranscoding] = useState(false);
const [startupGeneration, setStartupGeneration] = useState(0);
const [error, setError] = useState<string | null>(null);
const [switchedFileId, setSwitchedFileId] = useState<number | null>(null);
const switchAbortRef = useRef<AbortController | null>(null);
@@ -260,6 +273,23 @@ export function useTranscodeQuality({
[],
);
useEffect(() => {
if (!sessionId || !transportRestart) return;
switchAbortRef.current?.abort();
switchAbortRef.current = null;
if (dispatchTimerRef.current != null) {
clearTimeout(dispatchTimerRef.current);
dispatchTimerRef.current = null;
}
setTranscodeStreamUrl(transportRestart.streamUrl);
setPlayerStartSeconds(transportRestart.playerStartSeconds);
setStreamOriginSeconds(transportRestart.streamOriginSeconds);
setCanSeekAnywhere(transportRestart.canSeekAnywhere);
setIsTranscoding(false);
setError(null);
}, [sessionId, transportRestart]);
const startTranscode = useCallback(
(qualityId: string, currentPosition: number, forceRestart = false) => {
if (!sessionId) return;
@@ -336,6 +366,7 @@ export function useTranscodeQuality({
const abortController = new AbortController();
switchAbortRef.current = abortController;
setStartupGeneration((generation) => generation + 1);
setIsTranscoding(true);
setActiveQualityId(qualityId);
setError(null);
@@ -469,6 +500,16 @@ export function useTranscodeQuality({
[sessionId, activeQualityId, qualityOptions, config, effectiveVersion, playMethod],
);
const cancelPendingTranscodeStart = useCallback(() => {
if (dispatchTimerRef.current != null) {
clearTimeout(dispatchTimerRef.current);
dispatchTimerRef.current = null;
}
switchAbortRef.current?.abort();
switchAbortRef.current = null;
setIsTranscoding(false);
}, []);
const switchQuality = useCallback(
(qualityId: string, currentPosition: number, forceRestart?: boolean) => {
requestedQualityIdRef.current = qualityId;
@@ -520,7 +561,7 @@ export function useTranscodeQuality({
);
useEffect(() => {
if (!sessionId || (playMethod !== "transcode" && playMethod !== "remux")) {
if (!sessionId || transportRestart || (playMethod !== "transcode" && playMethod !== "remux")) {
return;
}
@@ -565,6 +606,7 @@ export function useTranscodeQuality({
selectedVersion?.file_id,
sessionId,
startTranscode,
transportRestart,
]);
return {
@@ -579,6 +621,8 @@ export function useTranscodeQuality({
canSeekAnywhere,
durationSeconds,
isTranscoding,
startupGeneration,
cancelPendingTranscodeStart,
error,
switchedFileId,
effectiveVersion,
+13
View File
@@ -210,9 +210,22 @@ export interface ChangeAudioResponse {
play_method: PlayMethod;
stream_url: string;
switch_mode: "reload";
player_start_seconds?: number;
stream_origin_seconds?: number;
timeline_offset_seconds?: number;
can_seek_anywhere?: boolean;
playback_info?: PlaybackSessionPlaybackInfo;
}
/** A server-prepared stream replacement that the player should adopt without restarting it. */
export interface PlaybackTransportRestart {
revision: number;
streamUrl: string;
playerStartSeconds: number;
streamOriginSeconds: number;
canSeekAnywhere: boolean;
}
/** Client codec capabilities sent to the server. */
export interface ClientCodecCapabilities {
codecs_video: string[];
@@ -0,0 +1,62 @@
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { HLS_STARTUP_TIMEOUT_MS, HlsStartupGuard } from "./hlsStartupGuard";
describe("HlsStartupGuard", () => {
beforeEach(() => {
vi.useFakeTimers();
});
afterEach(() => {
vi.useRealTimers();
});
it("fails startup when no playable frame arrives before the deadline", () => {
const onFailure = vi.fn();
const guard = new HlsStartupGuard(onFailure);
vi.advanceTimersByTime(HLS_STARTUP_TIMEOUT_MS - 1);
expect(onFailure).not.toHaveBeenCalled();
expect(guard.hasFailed()).toBe(false);
vi.advanceTimersByTime(1);
expect(onFailure).toHaveBeenCalledOnce();
expect(guard.hasFailed()).toBe(true);
guard.dispose();
});
it("allows one fatal network recovery before failing startup", () => {
const onFailure = vi.fn();
const guard = new HlsStartupGuard(onFailure);
expect(guard.handleFatalNetworkError()).toBe(true);
expect(onFailure).not.toHaveBeenCalled();
expect(guard.handleFatalNetworkError()).toBe(false);
expect(onFailure).toHaveBeenCalledOnce();
expect(guard.hasFailed()).toBe(true);
});
it("disarms startup limits after playable media arrives", () => {
const onFailure = vi.fn();
const guard = new HlsStartupGuard(onFailure);
guard.markPlaybackStarted();
vi.advanceTimersByTime(HLS_STARTUP_TIMEOUT_MS);
expect(onFailure).not.toHaveBeenCalled();
expect(guard.hasFailed()).toBe(false);
expect(guard.handleFatalNetworkError()).toBe(true);
expect(guard.handleFatalNetworkError()).toBe(true);
});
it("does not fail after disposal", () => {
const onFailure = vi.fn();
const guard = new HlsStartupGuard(onFailure);
guard.dispose();
vi.advanceTimersByTime(HLS_STARTUP_TIMEOUT_MS);
expect(onFailure).not.toHaveBeenCalled();
});
});
+59
View File
@@ -0,0 +1,59 @@
export const HLS_STARTUP_TIMEOUT_MS = 60_000;
const MAX_FATAL_NETWORK_RECOVERIES = 1;
type StartupState = "starting" | "playable" | "failed" | "disposed";
export class HlsStartupGuard {
private state: StartupState = "starting";
private fatalNetworkRecoveries = 0;
private timeoutId: ReturnType<typeof setTimeout> | null;
constructor(private readonly onFailure: () => void) {
this.timeoutId = setTimeout(() => this.fail(), HLS_STARTUP_TIMEOUT_MS);
}
handleFatalNetworkError(): boolean {
if (this.state === "playable") return true;
if (this.state !== "starting") return false;
if (this.fatalNetworkRecoveries < MAX_FATAL_NETWORK_RECOVERIES) {
this.fatalNetworkRecoveries++;
return true;
}
this.fail();
return false;
}
markPlaybackStarted() {
if (this.state !== "starting") return;
this.state = "playable";
this.clearTimeout();
}
hasFailed() {
return this.state === "failed";
}
dispose() {
this.state = "disposed";
this.clearTimeout();
}
private fail() {
if (this.state !== "starting") return;
this.state = "failed";
this.clearTimeout();
this.onFailure();
}
private clearTimeout() {
if (this.timeoutId === null) return;
clearTimeout(this.timeoutId);
this.timeoutId = null;
}
}