fix(watchsync): repair MDBList scrobble lifecycle
This commit is contained in:
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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}
|
||||
|
||||
|
||||
@@ -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)}
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user