From 4f249fda8f34fd1589cdbc6b6a9ca08c1c98b5e6 Mon Sep 17 00:00:00 2001 From: Quick104 <31828688+Quick104@users.noreply.github.com> Date: Mon, 20 Jul 2026 17:14:42 -0400 Subject: [PATCH] fix(watchsync): repair MDBList scrobble lifecycle --- internal/api/handlers/playback.go | 43 ++- internal/api/handlers/playback_test.go | 111 ++++++++ .../watchsync/providers/mdblist/provider.go | 264 ++++++++++++++++-- .../providers/mdblist/provider_test.go | 187 ++++++++++++- 4 files changed, 566 insertions(+), 39 deletions(-) diff --git a/internal/api/handlers/playback.go b/internal/api/handlers/playback.go index cee87d58..ecf883c5 100644 --- a/internal/api/handlers/playback.go +++ b/internal/api/handlers/playback.go @@ -1372,6 +1372,40 @@ func (h *PlaybackHandler) scrobbleEventForSession(ctx context.Context, session * return event } +func (h *PlaybackHandler) scrobbleEventForStoppedSession( + ctx context.Context, + session *playback.Session, + stopResult watchstate.PlaybackStopResult, +) (watchsync.ScrobbleEvent, bool) { + if session == nil || session.DisableProgressPersistence { + return watchsync.ScrobbleEvent{}, false + } + + mediaItemID := stopResult.MediaItemID + duration := stopResult.DurationSeconds + position := stopResult.FinalPositionSeconds + if mediaItemID == "" { + if h.fileResolver == nil { + return watchsync.ScrobbleEvent{}, false + } + file, err := h.loadFileByPreferredID(ctx, requestedMediaFileID(session), session.MediaFileID) + if err != nil || file == nil { + return watchsync.ScrobbleEvent{}, false + } + mediaItemID = playbackProgressTarget(file) + if mediaItemID == "" { + return watchsync.ScrobbleEvent{}, false + } + duration = float64(file.Duration) + position = session.Position + } + + event := h.scrobbleEventForSession(ctx, session, mediaItemID, duration, position) + event.HistoryID = stopResult.HistoryID + event.Completed = stopResult.Completed + return event, true +} + func intPtrValue(value *int) int { if value == nil { return 0 @@ -1463,15 +1497,12 @@ func (h *PlaybackHandler) finalizeSessionStop(ctx context.Context, session *play } stopResult := h.persistStopAndHistory(ctx, session) - if h.WatchScrobbler != nil && stopResult.MediaItemID != "" { - event := h.scrobbleEventForSession(ctx, session, stopResult.MediaItemID, stopResult.DurationSeconds, stopResult.FinalPositionSeconds) - event.HistoryID = stopResult.HistoryID - event.Completed = stopResult.Completed - if stopResult.Completed { + if h.WatchScrobbler != nil { + if event, ok := h.scrobbleEventForStoppedSession(ctx, session, stopResult); ok && (userInitiated || stopResult.Completed) { if err := h.WatchScrobbler.ScrobbleStop(ctx, event); err != nil { slog.WarnContext(ctx, "failed to queue watch provider stop scrobble", "component", "api", "session", session.ID, "error", err) } - } else if !stopResult.SkippedBelowMinResume { + } else if ok { if err := h.WatchScrobbler.ScrobblePause(ctx, event); err != nil { slog.WarnContext(ctx, "failed to queue watch provider pause scrobble", "component", "api", "session", session.ID, "error", err) } diff --git a/internal/api/handlers/playback_test.go b/internal/api/handlers/playback_test.go index 7a53a9ab..02297f8f 100644 --- a/internal/api/handlers/playback_test.go +++ b/internal/api/handlers/playback_test.go @@ -32,6 +32,7 @@ import ( "github.com/Silo-Server/silo-server/internal/transcodenode" "github.com/Silo-Server/silo-server/internal/userdb" "github.com/Silo-Server/silo-server/internal/userstore" + "github.com/Silo-Server/silo-server/internal/watchsync" ) type testUserStoreProvider struct { @@ -133,6 +134,27 @@ func (noopPlaybackAdminStore) RecordHistory(context.Context, AdminPlaybackHistor func (noopPlaybackAdminStore) DeleteSession(context.Context, string) error { return nil } +type recordingPlaybackWatchScrobbler struct { + starts []watchsync.ScrobbleEvent + pauses []watchsync.ScrobbleEvent + stops []watchsync.ScrobbleEvent +} + +func (s *recordingPlaybackWatchScrobbler) ScrobbleStart(_ context.Context, event watchsync.ScrobbleEvent) error { + s.starts = append(s.starts, event) + return nil +} + +func (s *recordingPlaybackWatchScrobbler) ScrobblePause(_ context.Context, event watchsync.ScrobbleEvent) error { + s.pauses = append(s.pauses, event) + return nil +} + +func (s *recordingPlaybackWatchScrobbler) ScrobbleStop(_ context.Context, event watchsync.ScrobbleEvent) error { + s.stops = append(s.stops, event) + return nil +} + type testEpisodeLookup struct { episode *models.Episode } @@ -447,6 +469,95 @@ func TestPlaybackSessionProgressPersistenceCanBeDisabled(t *testing.T) { } } +func TestFinalizeSessionStop_UsesProviderLifecycleSemantics(t *testing.T) { + tests := []struct { + name string + position float64 + userInitiated bool + disablePersistence bool + wantPauseCalls int + wantStopCalls int + wantEventPosition float64 + }{ + { + name: "user exit stops below-resume-threshold scrobble", + position: 120, + userInitiated: true, + wantStopCalls: 1, + wantEventPosition: 120, + }, + { + name: "system teardown pauses reconstructable scrobble", + position: 120, + wantPauseCalls: 1, + wantEventPosition: 120, + }, + { + name: "completed system teardown stops scrobble", + position: 3500, + wantStopCalls: 1, + wantEventPosition: 3500, + }, + { + name: "user exit at zero still stops scrobble", + userInitiated: true, + wantStopCalls: 1, + wantEventPosition: 0, + }, + { + name: "disabled persistence does not scrobble", + position: 120, + userInitiated: true, + disablePersistence: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + store := newPlaybackTestStore(t) + file := &models.MediaFile{ + ID: 42, + ContentID: "movie-1", + Duration: 3600, + } + scrobbler := &recordingPlaybackWatchScrobbler{} + handler := NewPlaybackHandler(playback.NewSessionManager(0, 0), testPlaybackFileResolver{file: file}) + handler.StoreProvider = testUserStoreProvider{store: store} + handler.WatchScrobbler = scrobbler + + session := &playback.Session{ + ID: "session-1", + UserID: 1, + ProfileID: "profile-1", + MediaFileID: file.ID, + Position: tt.position, + DisableProgressPersistence: tt.disablePersistence, + StartedAt: time.Now().Add(-time.Minute), + } + handler.finalizeSessionStop(context.Background(), session, false, "", tt.userInitiated) + + if got := len(scrobbler.pauses); got != tt.wantPauseCalls { + t.Fatalf("pause calls = %d, want %d", got, tt.wantPauseCalls) + } + if got := len(scrobbler.stops); got != tt.wantStopCalls { + t.Fatalf("stop calls = %d, want %d", got, tt.wantStopCalls) + } + if len(scrobbler.pauses)+len(scrobbler.stops) == 1 { + event := scrobbler.pauses + if len(event) == 0 { + event = scrobbler.stops + } + if event[0].MediaItemID != file.ContentID { + t.Fatalf("media item ID = %q, want %q", event[0].MediaItemID, file.ContentID) + } + if event[0].PositionSeconds != tt.wantEventPosition { + t.Fatalf("position = %v, want %v", event[0].PositionSeconds, tt.wantEventPosition) + } + } + }) + } +} + func TestBuildTranscodeStartResponse_UnifiedSeekAnywhere(t *testing.T) { file := &models.MediaFile{Duration: 3600} diff --git a/internal/watchsync/providers/mdblist/provider.go b/internal/watchsync/providers/mdblist/provider.go index d6499f7d..cf5ca598 100644 --- a/internal/watchsync/providers/mdblist/provider.go +++ b/internal/watchsync/providers/mdblist/provider.go @@ -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 } @@ -402,20 +434,44 @@ func (p *Provider) RemoveWatchlist(ctx context.Context, _ watchsync.ServerConfig func (p *Provider) sendWatchlist(ctx context.Context, conn watchsync.Connection, favorites []watchsync.LocalFavorite, path string) (watchsync.ExportResult, error) { payload := buildWatchlistPayload(favorites) if len(payload.Movies) == 0 && len(payload.Shows) == 0 { - return watchsync.ExportResult{}, nil + return watchlistExportResult(favorites, false), 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, response.NotFound > 0), nil +} + +func watchlistExportResult(favorites []watchsync.LocalFavorite, responseHasNotFound bool) watchsync.ExportResult { + result := watchsync.ExportResult{ + Sent: 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 { + // The API returns only an aggregate count, so a partial rejection + // cannot be attributed safely to one favorite. + 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 + } + return result } func (p *Provider) Start(ctx context.Context, _ watchsync.ServerConfig, conn watchsync.Connection, event watchsync.ScrobbleEvent) error { @@ -542,6 +598,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 +613,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 +751,100 @@ 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"` +} + +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 == "" { + return true, nil + } + if next == s.cursor { + return false, errors.New("provider returned the same cursor twice") + } + s.cursor = next + 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 int `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 +949,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 +1041,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 +1074,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 +1096,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 +1107,11 @@ 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": event.SeasonNumber, + "episode": event.EpisodeNumber, + } default: payload["movie"] = map[string]any{"ids": idsFromLocal(event.IMDbID, event.TMDBID, event.TVDBID)} } diff --git a/internal/watchsync/providers/mdblist/provider_test.go b/internal/watchsync/providers/mdblist/provider_test.go index 3b8240ab..537779e0 100644 --- a/internal/watchsync/providers/mdblist/provider_test.go +++ b/internal/watchsync/providers/mdblist/provider_test.go @@ -123,6 +123,31 @@ 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 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 +266,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 +348,47 @@ 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) + if season, _ := show["season"].(float64); int(season) != 3 { + t.Fatalf("expected show season 3, got %v", show["season"]) + } + if episode, _ := show["episode"].(float64); int(episode) != 7 { + t.Fatalf("expected show episode 7, got %v", show["episode"]) + } + if _, exists := gotBody["season"]; exists { + t.Fatalf("season must be nested under show: %#v", gotBody) + } 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 @@ -315,6 +424,63 @@ 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":1,"existing":0,"not_found":1}`)) + })) + 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 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 TestDoRetriesShortRateLimitInPlace(t *testing.T) { attempts := 0 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -411,6 +577,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 {