haeaeae/silo-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(watchsync): import current MDBList watched history (#628)

Browse Source 
Parse current MDBList watched payloads and expand aggregate show and season markers to local episode leaves without duplicating overlapping items.
This commit is contained in:
Quick
2026-08-13 11:10:36 -04:00
committed by GitHub
parent c4f38b50a6
commit ebc99ac09a
8 changed files with 520 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/historyimport/matcher.go
+50
View File
@@ -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 {
internal/historyimport/matcher_test.go
+69
View File
@@ -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()
internal/historyimport/repo.go
+39
View File
@@ -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(
internal/historyimport/types.go
+1
View File
@@ -21,6 +21,7 @@ const (
KindMovie = "movie"
KindSeries = "series"
KindSeason = "season"
KindEpisode = "episode"
)
internal/watchsync/providers/mdblist/provider.go
+120 -24
View File
@@ -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"`
}
internal/watchsync/providers/mdblist/provider_test.go
+106 -17
View File
@@ -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) {
internal/watchsync/service.go
+50 -6
View File
@@ -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,
internal/watchsync/service_test.go
+85
View File
@@ -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{