From ebc99ac09a010509b6aa3f61792f1a1feade490e Mon Sep 17 00:00:00 2001 From: Quick <31828688+Quick104@users.noreply.github.com> Date: Thu, 13 Aug 2026 11:10:36 -0400 Subject: [PATCH] fix(watchsync): import current MDBList watched history (#628) Parse current MDBList watched payloads and expand aggregate show and season markers to local episode leaves without duplicating overlapping items. --- internal/historyimport/matcher.go | 50 ++++++ internal/historyimport/matcher_test.go | 69 +++++++++ internal/historyimport/repo.go | 39 +++++ internal/historyimport/types.go | 1 + .../watchsync/providers/mdblist/provider.go | 144 +++++++++++++++--- .../providers/mdblist/provider_test.go | 123 ++++++++++++--- internal/watchsync/service.go | 56 ++++++- internal/watchsync/service_test.go | 85 +++++++++++ 8 files changed, 520 insertions(+), 47 deletions(-) diff --git a/internal/historyimport/matcher.go b/internal/historyimport/matcher.go index d7361233..ae74a469 100644 --- a/internal/historyimport/matcher.go +++ b/internal/historyimport/matcher.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "strings" + "time" ) type matcherRepository interface { @@ -11,6 +12,7 @@ type matcherRepository interface { MatchMediaByTitleYear(ctx context.Context, kind, title string, year int) ([]mediaLookupRow, error) MatchEpisodeByExternalID(ctx context.Context, column, value string) ([]mediaLookupRow, error) MatchEpisodeBySeries(ctx context.Context, seriesID string, seasonNumber, episodeNumber int) (*Match, error) + MatchEpisodesBySeries(ctx context.Context, seriesID string, seasonNumber *int, watchedAt *time.Time) ([]Match, error) } type Matcher struct { @@ -34,6 +36,54 @@ func (m *Matcher) Match(ctx context.Context, record Record) (*Match, string, err } } +// MatchLeaves resolves a provider watched marker to concrete playable items. +// Movies and episodes have one leaf. Whole-show and whole-season markers must +// expand to the local episodes that existed when the provider marker was +// recorded; Silo derives series/season completion from those episode leaves. +func (m *Matcher) MatchLeaves(ctx context.Context, record Record) ([]Match, string, error) { + switch record.Kind { + case KindMovie, KindEpisode: + match, reason, err := m.Match(ctx, record) + if err != nil || match == nil { + return nil, reason, err + } + return []Match{*match}, "", nil + case KindSeries, KindSeason: + seriesRecord := record + var seasonNumber *int + if record.Kind == KindSeason { + seriesRecord = Record{ + Kind: KindSeries, + Title: record.SeriesTitle, + Year: record.SeriesYear, + IMDbID: record.SeriesIMDbID, + TMDBID: record.SeriesTMDBID, + TVDBID: record.SeriesTVDBID, + } + season := record.SeasonNumber + seasonNumber = &season + } + seriesRecord.Kind = KindSeries + series, reason, err := m.matchSeries(ctx, seriesRecord) + if err != nil || series == nil { + return nil, reason, err + } + matches, err := m.repo.MatchEpisodesBySeries(ctx, series.MediaItemID, seasonNumber, record.LastPlayedAt) + if err != nil { + return nil, "", err + } + if len(matches) == 0 { + if seasonNumber != nil { + return nil, fmt.Sprintf("no available episodes matched for season %d", *seasonNumber), nil + } + return nil, "no available episodes matched for series", nil + } + return matches, "", nil + default: + return nil, "unsupported item kind", nil + } +} + func (m *Matcher) matchMovie(ctx context.Context, record Record) (*Match, string, error) { match, reason, err := m.matchMedia(ctx, KindMovie, record.TMDBID, record.IMDbID, "", record.Title, record.Year, false) if err != nil { diff --git a/internal/historyimport/matcher_test.go b/internal/historyimport/matcher_test.go index 07831607..109d528b 100644 --- a/internal/historyimport/matcher_test.go +++ b/internal/historyimport/matcher_test.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "testing" + "time" ) type matcherRepoStub struct { @@ -11,11 +12,15 @@ type matcherRepoStub struct { mediaByExternal map[string][]mediaLookupRow mediaByTitleYear map[string][]mediaLookupRow episodeBySeries *Match + episodesBySeries []Match episodeBySeriesCalls int episodeBySeriesID string episodeBySeriesSeason int episodeBySeriesEpisode int + episodesBySeriesID string + episodesBySeriesSeason *int + episodesBySeriesAt *time.Time mediaByExternalCalls int } @@ -40,6 +45,70 @@ func (s *matcherRepoStub) MatchEpisodeBySeries(_ context.Context, seriesID strin return s.episodeBySeries, nil } +func (s *matcherRepoStub) MatchEpisodesBySeries(_ context.Context, seriesID string, seasonNumber *int, watchedAt *time.Time) ([]Match, error) { + s.episodesBySeriesID = seriesID + if seasonNumber != nil { + season := *seasonNumber + s.episodesBySeriesSeason = &season + } + if watchedAt != nil { + at := *watchedAt + s.episodesBySeriesAt = &at + } + return s.episodesBySeries, nil +} + +func TestMatcherMatchLeavesExpandsWholeShowToEpisodes(t *testing.T) { + t.Parallel() + + watchedAt := time.Date(2025, time.October, 15, 20, 0, 0, 0, time.UTC) + repo := &matcherRepoStub{ + mediaByExternal: map[string][]mediaLookupRow{ + "series:tmdb_id:1396": {{ContentID: "series-1", Title: "Breaking Bad", Year: 2008}}, + }, + episodesBySeries: []Match{ + {MediaItemID: "episode-1", Kind: KindEpisode}, + {MediaItemID: "episode-2", Kind: KindEpisode}, + }, + } + + matches, reason, err := NewMatcher(repo).MatchLeaves(context.Background(), Record{ + Kind: KindSeries, Title: "Breaking Bad", Year: 2008, TMDBID: "1396", LastPlayedAt: &watchedAt, + }) + if err != nil || reason != "" { + t.Fatalf("MatchLeaves error = %v, reason = %q", err, reason) + } + if len(matches) != 2 || matches[0].MediaItemID != "episode-1" || matches[1].MediaItemID != "episode-2" { + t.Fatalf("matches = %#v", matches) + } + if repo.episodesBySeriesID != "series-1" || repo.episodesBySeriesSeason != nil || + repo.episodesBySeriesAt == nil || !repo.episodesBySeriesAt.Equal(watchedAt) { + t.Fatalf("unexpected expansion scope: id=%q season=%v watched_at=%v", + repo.episodesBySeriesID, repo.episodesBySeriesSeason, repo.episodesBySeriesAt) + } +} + +func TestMatcherMatchLeavesPreservesSpecialsSeasonZero(t *testing.T) { + t.Parallel() + + repo := &matcherRepoStub{ + mediaByExternal: map[string][]mediaLookupRow{ + "series:tvdb_id:75978": {{ContentID: "series-1", Title: "Family Guy", Year: 1999}}, + }, + episodesBySeries: []Match{{MediaItemID: "special-1", Kind: KindEpisode}}, + } + matches, reason, err := NewMatcher(repo).MatchLeaves(context.Background(), Record{ + Kind: KindSeason, SeriesTitle: "Family Guy", SeriesYear: 1999, + SeriesTVDBID: "75978", SeasonNumber: 0, + }) + if err != nil || reason != "" || len(matches) != 1 { + t.Fatalf("matches = %#v, reason = %q, err = %v", matches, reason, err) + } + if repo.episodesBySeriesSeason == nil || *repo.episodesBySeriesSeason != 0 { + t.Fatalf("season scope = %v, want explicit season 0", repo.episodesBySeriesSeason) + } +} + func TestMatcherMatchEpisode_UsesExternalIDsWithoutSeasonEpisodeNumbers(t *testing.T) { t.Parallel() diff --git a/internal/historyimport/repo.go b/internal/historyimport/repo.go index 1177cf33..782e040b 100644 --- a/internal/historyimport/repo.go +++ b/internal/historyimport/repo.go @@ -1050,6 +1050,45 @@ func (r *Repository) MatchEpisodeBySeries(ctx context.Context, seriesID string, return &match, nil } +func (r *Repository) MatchEpisodesBySeries( + ctx context.Context, + seriesID string, + seasonNumber *int, + watchedAt *time.Time, +) ([]Match, error) { + rows, err := r.pool.Query(ctx, ` + SELECT e.content_id, COALESCE(e.title, ''), COALESCE(series.year, 0) + FROM episodes e + LEFT JOIN media_items series ON series.content_id = e.series_id + WHERE e.series_id = $1 + AND ($2::integer IS NULL OR e.season_number = $2) + AND ($3::timestamptz IS NULL OR e.air_date IS NULL OR e.air_date <= ($3::timestamptz AT TIME ZONE 'UTC')::date) + AND EXISTS ( + SELECT 1 FROM episode_libraries el WHERE el.episode_id = e.content_id + ) + ORDER BY e.season_number, e.episode_number, e.content_id`, + seriesID, seasonNumber, watchedAt, + ) + if err != nil { + return nil, fmt.Errorf("matching episodes by series: %w", err) + } + defer rows.Close() + + matches := make([]Match, 0) + for rows.Next() { + var match Match + if err := rows.Scan(&match.MediaItemID, &match.Title, &match.Year); err != nil { + return nil, fmt.Errorf("scanning series episode match: %w", err) + } + match.Kind = KindEpisode + matches = append(matches, match) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterating series episode matches: %w", err) + } + return matches, nil +} + func scanSource(scanner interface{ Scan(dest ...any) error }) (*Source, error) { var source Source if err := scanner.Scan( diff --git a/internal/historyimport/types.go b/internal/historyimport/types.go index 539c8e96..95637967 100644 --- a/internal/historyimport/types.go +++ b/internal/historyimport/types.go @@ -21,6 +21,7 @@ const ( KindMovie = "movie" KindSeries = "series" + KindSeason = "season" KindEpisode = "episode" ) diff --git a/internal/watchsync/providers/mdblist/provider.go b/internal/watchsync/providers/mdblist/provider.go index bdc4f24c..841c23c3 100644 --- a/internal/watchsync/providers/mdblist/provider.go +++ b/internal/watchsync/providers/mdblist/provider.go @@ -136,7 +136,7 @@ func (p *Provider) FetchWatched(ctx context.Context, _ watchsync.ServerConfig, c } pageRows := watchedRowsFromPayload(p.Key(), payload) rows = append(rows, pageRows...) - fetched := len(payload.Movies) + len(payload.Episodes) + fetched := len(payload.Movies) + len(payload.Shows) + len(payload.Seasons) + len(payload.Episodes) done, err := page.advance(payload.Pagination, fetched) if err != nil { return nil, fmt.Errorf("mdblist watched pagination: %w", err) @@ -149,9 +149,9 @@ func (p *Provider) FetchWatched(ctx context.Context, _ watchsync.ServerConfig, c } func watchedRowsFromPayload(providerKey string, payload mdblistWatchedResponse) []watchsync.RemoteWatch { - rows := make([]watchsync.RemoteWatch, 0, len(payload.Movies)+len(payload.Episodes)) + rows := make([]watchsync.RemoteWatch, 0, len(payload.Movies)+len(payload.Shows)+len(payload.Seasons)+len(payload.Episodes)) for _, movie := range payload.Movies { - watchedAt := movie.WatchedAt + watchedAt := movie.LastWatchedAt if watchedAt == nil { continue } @@ -172,8 +172,55 @@ func watchedRowsFromPayload(providerKey string, payload mdblistWatchedResponse) LastWatchedAt: watchedAt, }) } + for _, show := range payload.Shows { + watchedAt := show.LastWatchedAt + if watchedAt == nil { + continue + } + key := showKey(show.Show.IDs) + if key == "" { + continue + } + rows = append(rows, watchsync.RemoteWatch{ + Provider: providerKey, + ProviderItemKey: "show:" + key, + Kind: historyimport.KindSeries, + Title: show.Show.Title, + Year: show.Show.Year, + IMDbID: show.Show.IDs.IMDb, + TMDBID: intString(show.Show.IDs.TMDB), + TVDBID: intString(show.Show.IDs.TVDB), + PlayCount: 1, + LastWatchedAt: watchedAt, + }) + } + for _, season := range payload.Seasons { + watchedAt := season.LastWatchedAt + if watchedAt == nil { + continue + } + key := showKey(season.Season.Show.IDs) + if key == "" { + continue + } + rows = append(rows, watchsync.RemoteWatch{ + Provider: providerKey, + ProviderItemKey: fmt.Sprintf("show:%s:s%d", key, season.Season.Number), + Kind: historyimport.KindSeason, + Title: season.Season.Name, + TMDBID: intString(season.Season.IDs.TMDB), + SeriesTitle: season.Season.Show.Title, + SeriesYear: season.Season.Show.Year, + SeriesIMDbID: season.Season.Show.IDs.IMDb, + SeriesTMDBID: intString(season.Season.Show.IDs.TMDB), + SeriesTVDBID: intString(season.Season.Show.IDs.TVDB), + SeasonNumber: season.Season.Number, + PlayCount: 1, + LastWatchedAt: watchedAt, + }) + } for _, episode := range payload.Episodes { - watchedAt := episode.WatchedAt + watchedAt := episode.LastWatchedAt if watchedAt == nil { continue } @@ -703,42 +750,76 @@ type mdblistShow struct { } type mdblistEpisode struct { - Title string `json:"title"` - Season int `json:"season"` - Number int `json:"number"` - IDs mdblistIDs `json:"ids"` + Title string `json:"title"` + Name string `json:"name"` + Season int `json:"season"` + Number int `json:"number"` + IDs mdblistIDs `json:"ids"` + Show mdblistShow `json:"show"` } type mdblistWatchedMovie struct { - WatchedAt *time.Time `json:"watched_at"` - Movie mdblistMovie `json:"movie"` + LastWatchedAt *time.Time `json:"last_watched_at"` + Movie mdblistMovie `json:"movie"` +} + +func (m *mdblistWatchedMovie) UnmarshalJSON(data []byte) error { + var raw struct { + LastWatchedAt *time.Time `json:"last_watched_at"` + WatchedAt *time.Time `json:"watched_at"` + Movie mdblistMovie `json:"movie"` + } + if err := json.Unmarshal(data, &raw); err != nil { + return err + } + m.LastWatchedAt = firstTimestamp(raw.LastWatchedAt, raw.WatchedAt) + m.Movie = raw.Movie + return nil +} + +type mdblistWatchedShow struct { + LastWatchedAt *time.Time `json:"last_watched_at"` + Show mdblistShow `json:"show"` +} + +type mdblistSeason struct { + Number int `json:"number"` + Name string `json:"name"` + IDs mdblistIDs `json:"ids"` + Show mdblistShow `json:"show"` +} + +type mdblistWatchedSeason struct { + LastWatchedAt *time.Time `json:"last_watched_at"` + Season mdblistSeason `json:"season"` } // mdblistWatchedEpisode tolerates both shapes MDBList uses for episode rows: // season/number inlined on the row, or nested under an `episode` object. type mdblistWatchedEpisode struct { - WatchedAt *time.Time `json:"watched_at"` - Season int `json:"season"` - Number int `json:"number"` - Title string `json:"title"` - IDs mdblistIDs `json:"ids"` - Show mdblistShow `json:"show"` + LastWatchedAt *time.Time + Season int + Number int + Title string + IDs mdblistIDs + Show mdblistShow } func (e *mdblistWatchedEpisode) UnmarshalJSON(data []byte) error { var raw struct { - WatchedAt *time.Time `json:"watched_at"` - Season int `json:"season"` - Number int `json:"number"` - Title string `json:"title"` - IDs mdblistIDs `json:"ids"` - Show mdblistShow `json:"show"` - Episode *mdblistEpisode `json:"episode"` + LastWatchedAt *time.Time `json:"last_watched_at"` + WatchedAt *time.Time `json:"watched_at"` + Season int `json:"season"` + Number int `json:"number"` + Title string `json:"title"` + IDs mdblistIDs `json:"ids"` + Show mdblistShow `json:"show"` + Episode *mdblistEpisode `json:"episode"` } if err := json.Unmarshal(data, &raw); err != nil { return err } - e.WatchedAt = raw.WatchedAt + e.LastWatchedAt = firstTimestamp(raw.LastWatchedAt, raw.WatchedAt) e.Season = raw.Season e.Number = raw.Number e.Title = raw.Title @@ -753,16 +834,31 @@ func (e *mdblistWatchedEpisode) UnmarshalJSON(data []byte) error { } if e.Title == "" { e.Title = raw.Episode.Title + if e.Title == "" { + e.Title = raw.Episode.Name + } } if e.IDs == (mdblistIDs{}) { e.IDs = raw.Episode.IDs } + if e.Show == (mdblistShow{}) { + e.Show = raw.Episode.Show + } } return nil } +func firstTimestamp(preferred, fallback *time.Time) *time.Time { + if preferred != nil { + return preferred + } + return fallback +} + type mdblistWatchedResponse struct { Movies []mdblistWatchedMovie `json:"movies"` + Shows []mdblistWatchedShow `json:"shows"` + Seasons []mdblistWatchedSeason `json:"seasons"` Episodes []mdblistWatchedEpisode `json:"episodes"` Pagination *mdblistPagination `json:"pagination"` } diff --git a/internal/watchsync/providers/mdblist/provider_test.go b/internal/watchsync/providers/mdblist/provider_test.go index 9316e515..03a8fc13 100644 --- a/internal/watchsync/providers/mdblist/provider_test.go +++ b/internal/watchsync/providers/mdblist/provider_test.go @@ -75,31 +75,55 @@ func TestConnectWithAPIKeyRejectsEmpty(t *testing.T) { } } -func TestFetchWatchedParsesMoviesAndEpisodes(t *testing.T) { +func TestFetchWatchedParsesCurrentMoviesShowsSeasonsAndEpisodes(t *testing.T) { watched := time.Date(2025, time.October, 21, 12, 0, 0, 0, time.UTC) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]any{ "movies": []map[string]any{{ - "watched_at": watched.Format(time.RFC3339), + "last_watched_at": watched.Format(time.RFC3339), "movie": map[string]any{ "title": "Beetlejuice Beetlejuice", "year": 2024, "ids": map[string]any{"imdb": "tt2049403", "tmdb": 917496}, }, }}, - "episodes": []map[string]any{{ - "watched_at": watched.Format(time.RFC3339), - "season": 1, - "number": 2, - "title": "Cat's in the Bag...", - "ids": map[string]any{"tvdb": 349231}, + "shows": []map[string]any{{ + "last_watched_at": watched.Format(time.RFC3339), "show": map[string]any{ - "title": "Breaking Bad", - "year": 2008, - "ids": map[string]any{"tvdb": 81189, "imdb": "tt0903747"}, + "title": "The Expanse", + "year": 2015, + "ids": map[string]any{"tvdb": 280619, "tmdb": 63639}, }, }}, + "seasons": []map[string]any{{ + "last_watched_at": watched.Format(time.RFC3339), + "season": map[string]any{ + "number": 2, + "name": "Season 2", + "ids": map[string]any{"tmdb": 8160}, + "show": map[string]any{ + "title": "Breaking Bad", + "year": 2008, + "ids": map[string]any{"imdb": "tt0903747", "tmdb": 1396}, + }, + }, + }}, + "episodes": []map[string]any{{ + "last_watched_at": watched.Format(time.RFC3339), + "episode": map[string]any{ + "season": 1, + "number": 2, + "name": "Cat's in the Bag...", + "ids": map[string]any{"tmdb": 62086}, + "show": map[string]any{ + "title": "Breaking Bad", + "year": 2008, + "ids": map[string]any{"tvdb": 81189, "imdb": "tt0903747"}, + }, + }, + }}, + "pagination": map[string]any{"has_more": false}, }) })) defer server.Close() @@ -109,17 +133,51 @@ func TestFetchWatchedParsesMoviesAndEpisodes(t *testing.T) { if err != nil { t.Fatalf("fetch watched: %v", err) } - if len(rows) != 2 { - t.Fatalf("got %d rows, want 2", len(rows)) + if len(rows) != 4 { + t.Fatalf("got %d rows, want 4", len(rows)) } if rows[0].Kind != historyimport.KindMovie || rows[0].ProviderItemKey != "imdb:tt2049403" { t.Fatalf("unexpected movie row: %#v", rows[0]) } - if rows[1].Kind != historyimport.KindEpisode || rows[1].SeasonNumber != 1 || rows[1].EpisodeNumber != 2 { - t.Fatalf("unexpected episode row: %#v", rows[1]) + if rows[1].Kind != historyimport.KindSeries || rows[1].ProviderItemKey != "show:tvdb:280619" { + t.Fatalf("unexpected show row: %#v", rows[1]) } - if rows[1].SeriesIMDbID != "tt0903747" { - t.Fatalf("expected series imdb propagated, got %q", rows[1].SeriesIMDbID) + if rows[2].Kind != historyimport.KindSeason || rows[2].SeasonNumber != 2 || rows[2].SeriesTMDBID != "1396" { + t.Fatalf("unexpected season row: %#v", rows[2]) + } + if rows[3].Kind != historyimport.KindEpisode || rows[3].SeasonNumber != 1 || rows[3].EpisodeNumber != 2 { + t.Fatalf("unexpected episode row: %#v", rows[3]) + } + if rows[3].Title != "Cat's in the Bag..." || rows[3].SeriesIMDbID != "tt0903747" { + t.Fatalf("expected nested episode/show fields propagated, got %#v", rows[3]) + } + for i, row := range rows { + if row.LastWatchedAt == nil || !row.LastWatchedAt.Equal(watched) { + t.Fatalf("row %d watched timestamp = %v, want %v", i, row.LastWatchedAt, watched) + } + } +} + +func TestFetchWatchedAcceptsLegacyWatchedAtAndFlatEpisode(t *testing.T) { + watched := time.Date(2025, time.November, 1, 8, 30, 0, 0, time.UTC) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{ + "movies":[{"watched_at":"2025-11-01T08:30:00Z","movie":{"ids":{"imdb":"tt0111161"}}}], + "episodes":[{"watched_at":"2025-11-01T08:30:00Z","season":1,"number":2,"show":{"ids":{"tvdb":81189}}}] + }`)) + })) + defer server.Close() + + rows, err := NewProvider(server.Client(), server.URL).FetchWatched( + context.Background(), watchsync.ServerConfig{}, watchsync.Connection{AccessToken: "k"}, + ) + if err != nil { + t.Fatalf("fetch watched: %v", err) + } + if len(rows) != 2 || rows[0].LastWatchedAt == nil || !rows[0].LastWatchedAt.Equal(watched) || + rows[1].LastWatchedAt == nil || !rows[1].LastWatchedAt.Equal(watched) { + t.Fatalf("unexpected legacy rows: %#v", rows) } } @@ -174,6 +232,37 @@ func TestFetchWatchedContinuesOffsetPaginationWhenHasMore(t *testing.T) { } } +func TestFetchWatchedOffsetCountsAggregateRows(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") == "2" { + _, _ = w.Write([]byte(`{"movies":[],"shows":[],"seasons":[],"episodes":[],"pagination":{"has_more":false}}`)) + return + } + _, _ = w.Write([]byte(`{ + "shows":[{"last_watched_at":"2025-11-01T08:30:00Z","show":{"ids":{"tmdb":1396}}}], + "seasons":[{"last_watched_at":"2025-11-01T08:30:00Z","season":{"number":1,"show":{"ids":{"tmdb":1396}}}}], + "pagination":{"has_more":true} + }`)) + })) + defer server.Close() + + rows, err := NewProvider(server.Client(), server.URL).FetchWatched( + context.Background(), watchsync.ServerConfig{}, watchsync.Connection{AccessToken: "k"}, + ) + if err != nil { + t.Fatalf("fetch watched: %v", err) + } + if len(rows) != 2 { + t.Fatalf("got %d rows, want 2", len(rows)) + } + if len(offsets) != 2 || offsets[0] != "" || offsets[1] != "2" { + 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) { diff --git a/internal/watchsync/service.go b/internal/watchsync/service.go index bdae327e..5221ec6f 100644 --- a/internal/watchsync/service.go +++ b/internal/watchsync/service.go @@ -42,6 +42,10 @@ type mediaMatcher interface { Match(ctx context.Context, record historyimport.Record) (*historyimport.Match, string, error) } +type watchedLeafMatcher interface { + MatchLeaves(ctx context.Context, record historyimport.Record) ([]historyimport.Match, string, error) +} + type watchStateImporter interface { RecordImportedWatchIfNewerWithSource(ctx context.Context, userID int, profileID, targetID string, duration, position float64, completed bool, updatedAt time.Time, watchedAt *time.Time, source userstore.WatchHistorySource) (bool, error) } @@ -1020,6 +1024,11 @@ type ImportWatchedResult struct { Warnings []string } +type matchedWatchedLeaf struct { + match historyimport.Match + watchedAt time.Time +} + func (s *Service) ImportWatched( ctx context.Context, conn Connection, @@ -1038,12 +1047,13 @@ func (s *Service) ImportWatched( } rows := batch.Rows result := ImportWatchedResult{Found: len(rows), Warnings: append([]string{}, batch.Warnings...)} + matchedLeaves := make(map[string]matchedWatchedLeaf) for _, row := range rows { - match, reason, err := s.matcher.Match(ctx, row.HistoryRecord()) + matches, reason, err := s.matchWatchedLeaves(ctx, row) if err != nil { return result, err } - if match == nil { + if len(matches) == 0 { result.Unmatched++ if reason != "" { result.Warnings = append(result.Warnings, reason) @@ -1053,17 +1063,35 @@ func (s *Service) ImportWatched( if row.LastWatchedAt == nil { continue } - duration, _ := s.mediaDuration(ctx, match.MediaItemID) + for _, match := range matches { + existing, ok := matchedLeaves[match.MediaItemID] + if !ok || row.LastWatchedAt.After(existing.watchedAt) { + matchedLeaves[match.MediaItemID] = matchedWatchedLeaf{ + match: match, + watchedAt: *row.LastWatchedAt, + } + } + } + } + mediaItemIDs := make([]string, 0, len(matchedLeaves)) + for mediaItemID := range matchedLeaves { + mediaItemIDs = append(mediaItemIDs, mediaItemID) + } + sort.Strings(mediaItemIDs) + for _, mediaItemID := range mediaItemIDs { + leaf := matchedLeaves[mediaItemID] + duration, _ := s.mediaDuration(ctx, leaf.match.MediaItemID) + watchedAt := leaf.watchedAt created, err := s.watchState.RecordImportedWatchIfNewerWithSource( ctx, conn.UserID, conn.ProfileID, - match.MediaItemID, + leaf.match.MediaItemID, duration, 0, true, - *row.LastWatchedAt, - row.LastWatchedAt, + watchedAt, + &watchedAt, historySourceForProvider(importer), ) if err != nil { @@ -1083,6 +1111,22 @@ func (s *Service) ImportWatched( return result, nil } +func (s *Service) matchWatchedLeaves(ctx context.Context, row RemoteWatch) ([]historyimport.Match, string, error) { + record := row.HistoryRecord() + if row.Kind == historyimport.KindSeries || row.Kind == historyimport.KindSeason { + matcher, ok := s.matcher.(watchedLeafMatcher) + if !ok { + return nil, "watch provider matcher cannot expand aggregate watched state", nil + } + return matcher.MatchLeaves(ctx, record) + } + match, reason, err := s.matcher.Match(ctx, record) + if err != nil || match == nil { + return nil, reason, err + } + return []historyimport.Match{*match}, "", nil +} + func fetchWatchedImportBatch( ctx context.Context, cfg ServerConfig, diff --git a/internal/watchsync/service_test.go b/internal/watchsync/service_test.go index 0f7f6e11..6a7aabbb 100644 --- a/internal/watchsync/service_test.go +++ b/internal/watchsync/service_test.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "errors" + "reflect" "strconv" "strings" "sync" @@ -1060,6 +1061,18 @@ func (m matchedMatcherStub) Match(context.Context, historyimport.Record) (*histo return &historyimport.Match{MediaItemID: m.mediaItemID}, "", nil } +type watchedLeafMatcherStub struct { + matches []historyimport.Match +} + +func (m watchedLeafMatcherStub) Match(context.Context, historyimport.Record) (*historyimport.Match, string, error) { + return nil, "unexpected single-item match", nil +} + +func (m watchedLeafMatcherStub) MatchLeaves(context.Context, historyimport.Record) ([]historyimport.Match, string, error) { + return m.matches, "", nil +} + type noOpWatchState struct{} func (noOpWatchState) RecordImportedHistoryWithSource( @@ -1682,6 +1695,78 @@ func TestServiceImportWatchedUsesProviderHistorySource(t *testing.T) { } } +func TestServiceImportWatchedExpandsAggregateMarkerToEpisodeLeaves(t *testing.T) { + repo := newServiceFakeRepo() + watchedAt := time.Date(2026, 5, 4, 12, 0, 0, 0, time.UTC) + provider := watchedImporterStub{ + key: "mdblist", + source: userstore.WatchHistorySourceMDBList, + rows: []RemoteWatch{{ + Provider: "mdblist", Kind: historyimport.KindSeries, Title: "Breaking Bad", LastWatchedAt: &watchedAt, + }}, + } + watchState := &recordingWatchState{} + service := NewService(repo, NewRegistry()). + WithMatcher(watchedLeafMatcherStub{matches: []historyimport.Match{ + {MediaItemID: "episode-1", Kind: historyimport.KindEpisode}, + {MediaItemID: "episode-2", Kind: historyimport.KindEpisode}, + }}). + WithWatchState(watchState) + + result, err := service.ImportWatched(context.Background(), Connection{ + ID: "conn-1", Provider: "mdblist", UserID: 7, ProfileID: "profile-1", + }, ServerConfig{}, provider) + if err != nil { + t.Fatalf("ImportWatched: %v", err) + } + if result.Found != 1 || result.Imported != 2 || result.Unmatched != 0 { + t.Fatalf("result = %+v, want one aggregate found and two leaves imported", result) + } + if !reflect.DeepEqual(watchState.targetIDs, []string{"episode-1", "episode-2"}) { + t.Fatalf("recorded target ids = %+v", watchState.targetIDs) + } + if !reflect.DeepEqual(watchState.sources, []userstore.WatchHistorySource{ + userstore.WatchHistorySourceMDBList, + userstore.WatchHistorySourceMDBList, + }) { + t.Fatalf("recorded sources = %+v", watchState.sources) + } +} + +func TestServiceImportWatchedDeduplicatesResolvedLeavesAtNewestTimestamp(t *testing.T) { + repo := newServiceFakeRepo() + olderWatchedAt := time.Date(2026, 5, 3, 12, 0, 0, 0, time.UTC) + newerWatchedAt := olderWatchedAt.Add(24 * time.Hour) + provider := watchedImporterStub{ + key: "mdblist", + source: userstore.WatchHistorySourceMDBList, + rows: []RemoteWatch{ + {Provider: "mdblist", Kind: historyimport.KindMovie, Title: "Inception", LastWatchedAt: &olderWatchedAt}, + {Provider: "mdblist", Kind: historyimport.KindMovie, Title: "Inception", LastWatchedAt: &newerWatchedAt}, + }, + } + watchState := &recordingWatchState{} + service := NewService(repo, NewRegistry()). + WithMatcher(matchedMatcherStub{mediaItemID: testMovieMediaID}). + WithWatchState(watchState) + + result, err := service.ImportWatched(context.Background(), Connection{ + ID: "conn-1", Provider: "mdblist", UserID: 7, ProfileID: "profile-1", + }, ServerConfig{}, provider) + if err != nil { + t.Fatalf("ImportWatched: %v", err) + } + if result.Found != 2 || result.Imported != 1 || result.Unmatched != 0 { + t.Fatalf("result = %+v, want two provider markers and one imported leaf", result) + } + if !reflect.DeepEqual(watchState.targetIDs, []string{testMovieMediaID}) { + t.Fatalf("recorded target ids = %+v", watchState.targetIDs) + } + if len(watchState.watchedAt) != 1 || watchState.watchedAt[0] == nil || !watchState.watchedAt[0].Equal(newerWatchedAt) { + t.Fatalf("recorded watched_at = %+v, want %v", watchState.watchedAt, newerWatchedAt) + } +} + func TestServiceImportWatchedSkipsRowsWithoutLastWatchedAt(t *testing.T) { repo := newServiceFakeRepo() provider := watchedImporterStub{