diff --git a/internal/catalog/continue_watching_progress.go b/internal/catalog/continue_watching_progress.go new file mode 100644 index 00000000..a6c162dd --- /dev/null +++ b/internal/catalog/continue_watching_progress.go @@ -0,0 +1,212 @@ +package catalog + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/Silo-Server/silo-server/internal/userstore" +) + +// ProgressLister pages watch-progress rows for one profile. +// userstore.UserStore satisfies it. +type ProgressLister interface { + ListProgress(ctx context.Context, profileID, status string, limit, offset int) ([]userstore.WatchProgress, error) +} + +// ProgressSnapshot pairs a media item with the time its progress row last changed. +type ProgressSnapshot struct { + ContentID string + UpdatedAt time.Time +} + +// ContinueWatchingProgressFilter identifies in-progress entries that Continue +// Watching surfaces should hide: episodes superseded by a later-completed +// episode in the same series. The first-party sections fetcher and the +// jellycompat Resume endpoint share it so both surfaces agree on what "still +// watching" means. +type ContinueWatchingProgressFilter struct { + pool *pgxpool.Pool +} + +// NewContinueWatchingProgressFilter creates a filter. A nil pool disables the +// superseded-episode check, leaving entries unfiltered. +func NewContinueWatchingProgressFilter(pool *pgxpool.Pool) *ContinueWatchingProgressFilter { + return &ContinueWatchingProgressFilter{pool: pool} +} + +const supersededProgressPageSize = 500 + +// SupersededEpisodeProgressIDs returns the content IDs of in-progress entries +// whose series has a later episode completed more recently than the entry's +// own progress. Those entries are stale — the viewer already moved past them. +// Non-episode entries never match. +func (f *ContinueWatchingProgressFilter) SupersededEpisodeProgressIDs(ctx context.Context, store ProgressLister, profileID string, entries []userstore.WatchProgress) (map[string]struct{}, error) { + if f == nil || f.pool == nil { + return map[string]struct{}{}, nil + } + inProgress := ProgressSnapshots(entries) + if len(inProgress) == 0 { + return map[string]struct{}{}, nil + } + + completed, err := CompletedProgressSnapshots(ctx, store, profileID) + if err != nil { + return nil, err + } + if len(completed) == 0 { + return map[string]struct{}{}, nil + } + + inProgressIDs, inProgressUpdatedAts := splitProgressSnapshots(inProgress) + completedIDs, completedUpdatedAts := splitProgressSnapshots(completed) + query := buildSupersededEpisodeProgressQuery() + rows, err := f.pool.Query(ctx, query, inProgressIDs, inProgressUpdatedAts, completedIDs, completedUpdatedAts) + if err != nil { + return nil, fmt.Errorf("querying superseded episode progress: %w", err) + } + defer rows.Close() + + superseded := make(map[string]struct{}) + for rows.Next() { + var mediaItemID string + if err := rows.Scan(&mediaItemID); err != nil { + return nil, fmt.Errorf("scanning superseded episode progress: %w", err) + } + superseded[mediaItemID] = struct{}{} + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterating superseded episode progress: %w", err) + } + return superseded, nil +} + +// CompletedProgressSnapshots pages through all completed progress rows for the +// profile and returns deduplicated snapshots. +func CompletedProgressSnapshots(ctx context.Context, store ProgressLister, profileID string) ([]ProgressSnapshot, error) { + seen := make(map[string]struct{}) + snapshots := make([]ProgressSnapshot, 0) + + for offset := 0; ; offset += supersededProgressPageSize { + entries, err := store.ListProgress(ctx, profileID, "completed", supersededProgressPageSize, offset) + if err != nil { + return nil, fmt.Errorf("listing completed progress for superseded episodes: %w", err) + } + + for _, snapshot := range ProgressSnapshots(entries) { + contentID := snapshot.ContentID + if _, ok := seen[contentID]; ok { + continue + } + seen[contentID] = struct{}{} + snapshots = append(snapshots, snapshot) + } + + if len(entries) < supersededProgressPageSize { + return snapshots, nil + } + } +} + +// ProgressSnapshots converts progress rows to snapshots, dropping rows with a +// blank media item ID or an unparseable timestamp. +func ProgressSnapshots(entries []userstore.WatchProgress) []ProgressSnapshot { + snapshots := make([]ProgressSnapshot, 0, len(entries)) + for _, entry := range entries { + contentID := strings.TrimSpace(entry.MediaItemID) + if contentID == "" { + continue + } + updatedAt, err := time.Parse(time.RFC3339, entry.UpdatedAt) + if err != nil || updatedAt.IsZero() { + continue + } + snapshots = append(snapshots, ProgressSnapshot{ + ContentID: contentID, + UpdatedAt: updatedAt.UTC(), + }) + } + return snapshots +} + +func splitProgressSnapshots(snapshots []ProgressSnapshot) ([]string, []time.Time) { + contentIDs := make([]string, len(snapshots)) + updatedAts := make([]time.Time, len(snapshots)) + for i, snapshot := range snapshots { + contentIDs[i] = snapshot.ContentID + updatedAts[i] = snapshot.UpdatedAt + } + return contentIDs, updatedAts +} + +// The snapshots arrive as unnest arrays instead of joins against +// user_watch_progress because per-user progress may live in a SQLite store +// rather than this Postgres database. +func buildSupersededEpisodeProgressQuery() string { + return ` + WITH in_progress(content_id, updated_at) AS ( + SELECT * FROM unnest($1::text[], $2::timestamptz[]) + ), + completed(content_id, updated_at) AS ( + SELECT * FROM unnest($3::text[], $4::timestamptz[]) + ) + SELECT DISTINCT ip.content_id + FROM in_progress ip_progress + JOIN episodes ip ON ip.content_id = ip_progress.content_id + JOIN episodes done + ON done.series_id = ip.series_id + AND (done.season_number, done.episode_number) > (ip.season_number, ip.episode_number) + JOIN completed done_progress + ON done_progress.content_id = done.content_id + WHERE done_progress.updated_at > ip_progress.updated_at` +} + +// FilterSupersededProgress drops entries whose media item ID is in the +// superseded set. +func FilterSupersededProgress(entries []userstore.WatchProgress, superseded map[string]struct{}) []userstore.WatchProgress { + if len(entries) == 0 || len(superseded) == 0 { + return entries + } + + filtered := make([]userstore.WatchProgress, 0, len(entries)) + for _, entry := range entries { + if _, ok := superseded[entry.MediaItemID]; ok { + continue + } + filtered = append(filtered, entry) + } + return filtered +} + +// HomeDismissalIndex maps media item ID to its dismissal row for one home surface. +type HomeDismissalIndex map[string]userstore.HomeItemDismissal + +// NewHomeDismissalIndex builds an index from dismissal rows. +func NewHomeDismissalIndex(dismissals []userstore.HomeItemDismissal) HomeDismissalIndex { + index := make(HomeDismissalIndex, len(dismissals)) + for _, dismissal := range dismissals { + index[dismissal.MediaItemID] = dismissal + } + return index +} + +// FilterProgress drops entries still covered by a dismissal. A dismissal only +// holds while the entry's progress timestamp matches the one captured when the +// user dismissed it; resuming playback re-surfaces the item. +func (idx HomeDismissalIndex) FilterProgress(entries []userstore.WatchProgress) []userstore.WatchProgress { + if len(entries) == 0 || len(idx) == 0 { + return entries + } + + filtered := make([]userstore.WatchProgress, 0, len(entries)) + for _, entry := range entries { + dismissal, ok := idx[entry.MediaItemID] + if !ok || dismissal.ProgressUpdatedAt == nil || *dismissal.ProgressUpdatedAt != entry.UpdatedAt { + filtered = append(filtered, entry) + } + } + return filtered +} diff --git a/internal/catalog/continue_watching_progress_test.go b/internal/catalog/continue_watching_progress_test.go new file mode 100644 index 00000000..74789a75 --- /dev/null +++ b/internal/catalog/continue_watching_progress_test.go @@ -0,0 +1,189 @@ +package catalog + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/Silo-Server/silo-server/internal/userstore" +) + +func TestFilterSupersededProgressDropsOlderPartialsAfterLaterCompletedEpisode(t *testing.T) { + t.Parallel() + + entries := []userstore.WatchProgress{ + {MediaItemID: "boys-s1e1"}, + {MediaItemID: "boys-s5e3"}, + {MediaItemID: "movie-1"}, + } + superseded := map[string]struct{}{ + "boys-s1e1": {}, + "boys-s5e3": {}, + } + + filtered := FilterSupersededProgress(entries, superseded) + + if len(filtered) != 1 || filtered[0].MediaItemID != "movie-1" { + t.Fatalf("filtered entries = %+v, want only movie-1", filtered) + } +} + +func TestCompletedProgressSnapshotsPagesThroughConfiguredStore(t *testing.T) { + t.Parallel() + + entries := make([]userstore.WatchProgress, supersededProgressPageSize+1) + for i := range entries { + entries[i] = userstore.WatchProgress{ + MediaItemID: "done-" + time.Unix(int64(i), 0).Format("150405"), + UpdatedAt: time.Date(2025, 1, 1, 0, 0, i, 0, time.UTC).Format(time.RFC3339), + } + } + store := &stubProgressLister{entries: entries} + + snapshots, err := CompletedProgressSnapshots(context.Background(), store, "p1") + if err != nil { + t.Fatalf("CompletedProgressSnapshots: %v", err) + } + if len(snapshots) != len(entries) { + t.Fatalf("completed snapshots count = %d, want %d", len(snapshots), len(entries)) + } + if len(store.calls) != 2 { + t.Fatalf("ListProgress calls = %+v, want 2 paged calls", store.calls) + } + if store.calls[0] != (progressListCall{profileID: "p1", status: "completed", limit: supersededProgressPageSize, offset: 0}) { + t.Fatalf("first ListProgress call = %+v", store.calls[0]) + } + if store.calls[1] != (progressListCall{profileID: "p1", status: "completed", limit: supersededProgressPageSize, offset: supersededProgressPageSize}) { + t.Fatalf("second ListProgress call = %+v", store.calls[1]) + } +} + +func TestBuildSupersededEpisodeProgressQueryUsesStoreSnapshotsWithFreshnessGate(t *testing.T) { + t.Parallel() + + query := buildSupersededEpisodeProgressQuery() + expectedFragments := []string{ + "unnest($1::text[], $2::timestamptz[])", + "unnest($3::text[], $4::timestamptz[])", + "FROM in_progress ip_progress", + "done_progress.updated_at > ip_progress.updated_at", + } + for _, fragment := range expectedFragments { + if !strings.Contains(query, fragment) { + t.Fatalf("expected superseded progress query to contain %q, got:\n%s", fragment, query) + } + } + unexpectedFragments := []string{ + "user_watch_progress", + "user_history_hidden_items", + } + for _, fragment := range unexpectedFragments { + if strings.Contains(query, fragment) { + t.Fatalf("superseded progress query contains %q, got:\n%s", fragment, query) + } + } +} + +func TestSupersededEpisodeProgressIDsWithoutPoolReturnsEmptySet(t *testing.T) { + t.Parallel() + + filter := NewContinueWatchingProgressFilter(nil) + entries := []userstore.WatchProgress{{ + MediaItemID: "ep-1", + UpdatedAt: time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC).Format(time.RFC3339), + }} + store := &stubProgressLister{} + + superseded, err := filter.SupersededEpisodeProgressIDs(context.Background(), store, "p1", entries) + if err != nil { + t.Fatalf("SupersededEpisodeProgressIDs: %v", err) + } + if len(superseded) != 0 { + t.Fatalf("superseded = %v, want empty set", superseded) + } + if len(store.calls) != 0 { + t.Fatalf("ListProgress calls = %+v, want none without a pool", store.calls) + } +} + +func TestHomeDismissalIndexFilterProgressDropsOnlyMatchingTimestamps(t *testing.T) { + t.Parallel() + + dismissedAt := "2025-01-01T00:00:00Z" + resumedAt := "2025-01-02T00:00:00Z" + idx := NewHomeDismissalIndex([]userstore.HomeItemDismissal{ + {MediaItemID: "still-dismissed", ProgressUpdatedAt: &dismissedAt}, + {MediaItemID: "resumed-since", ProgressUpdatedAt: &dismissedAt}, + {MediaItemID: "no-timestamp"}, + }) + + entries := []userstore.WatchProgress{ + {MediaItemID: "still-dismissed", UpdatedAt: dismissedAt}, + {MediaItemID: "resumed-since", UpdatedAt: resumedAt}, + {MediaItemID: "no-timestamp", UpdatedAt: dismissedAt}, + {MediaItemID: "never-dismissed", UpdatedAt: dismissedAt}, + } + + filtered := idx.FilterProgress(entries) + + got := make([]string, 0, len(filtered)) + for _, entry := range filtered { + got = append(got, entry.MediaItemID) + } + want := []string{"resumed-since", "no-timestamp", "never-dismissed"} + if len(got) != len(want) { + t.Fatalf("filtered = %v, want %v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("filtered = %v, want %v", got, want) + } + } +} + +func TestProgressSnapshotsSkipsBlankIDsAndBadTimestamps(t *testing.T) { + t.Parallel() + + valid := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC) + entries := []userstore.WatchProgress{ + {MediaItemID: "ok", UpdatedAt: valid.Format(time.RFC3339)}, + {MediaItemID: " ", UpdatedAt: valid.Format(time.RFC3339)}, + {MediaItemID: "bad-time", UpdatedAt: "not-a-time"}, + } + + snapshots := ProgressSnapshots(entries) + + if len(snapshots) != 1 || snapshots[0].ContentID != "ok" || !snapshots[0].UpdatedAt.Equal(valid) { + t.Fatalf("snapshots = %+v, want single valid snapshot for %q", snapshots, "ok") + } +} + +type progressListCall struct { + profileID string + status string + limit int + offset int +} + +type stubProgressLister struct { + entries []userstore.WatchProgress + calls []progressListCall +} + +func (s *stubProgressLister) ListProgress(_ context.Context, profileID, status string, limit, offset int) ([]userstore.WatchProgress, error) { + s.calls = append(s.calls, progressListCall{ + profileID: profileID, + status: status, + limit: limit, + offset: offset, + }) + if offset >= len(s.entries) { + return nil, nil + } + end := offset + limit + if end > len(s.entries) { + end = len(s.entries) + } + return s.entries[offset:end], nil +} diff --git a/internal/jellycompat/batch_progress_test.go b/internal/jellycompat/batch_progress_test.go index 33fddd5c..44f7696c 100644 --- a/internal/jellycompat/batch_progress_test.go +++ b/internal/jellycompat/batch_progress_test.go @@ -68,6 +68,10 @@ func (m *mockUserDataService) ListProgress(context.Context, *Session, string, in panic("unused") } +func (m *mockUserDataService) FilterResumeProgress(_ context.Context, _ *Session, entries []upstreamProgress) ([]upstreamProgress, error) { + return entries, nil +} + func (m *mockUserDataService) MarkPlayed(context.Context, *Session, string) error { panic("unused") } diff --git a/internal/jellycompat/content_service.go b/internal/jellycompat/content_service.go index da3b9d63..1537e100 100644 --- a/internal/jellycompat/content_service.go +++ b/internal/jellycompat/content_service.go @@ -26,6 +26,11 @@ type UserDataService interface { AddFavorite(ctx context.Context, session *Session, contentID string) error RemoveFavorite(ctx context.Context, session *Session, contentID string) error ListProgress(ctx context.Context, session *Session, status string, limit, offset int) ([]upstreamProgress, error) + // FilterResumeProgress drops in-progress entries that Continue Watching + // surfaces should hide: entries the user dismissed from the row and + // episodes superseded by a later-completed episode in the same series. + // It mirrors the first-party sections fetcher so both surfaces agree. + FilterResumeProgress(ctx context.Context, session *Session, entries []upstreamProgress) ([]upstreamProgress, error) ListProgressByMediaItems(ctx context.Context, session *Session, mediaItemIDs []string) (map[string]*upstreamProgress, error) GetProgress(ctx context.Context, session *Session, contentID string) (*upstreamProgress, error) MarkPlayed(ctx context.Context, session *Session, contentID string) error diff --git a/internal/jellycompat/handlers_items.go b/internal/jellycompat/handlers_items.go index cfeb2d0b..6f517341 100644 --- a/internal/jellycompat/handlers_items.go +++ b/internal/jellycompat/handlers_items.go @@ -1819,7 +1819,12 @@ type progressHydratedItem struct { const maxDetailUpgrades = 100 func (h *ItemsHandler) loadProgressPage(ctx context.Context, session *Session, status string, query itemsQuery, typeSet map[string]bool, libraryID *int) ([]baseItemDTO, int, error) { - if len(typeSet) == 0 && libraryID == nil && !query.enableTotalRecordCount { + // Resume views hide dismissed and superseded entries, so the visible list + // is sparser than the raw store list. The raw-offset fast path below is + // only safe for the first page; deeper StartIndex values must go through + // the scan-from-zero branch, which paginates over visible entries. + resumeFiltered := status == "in_progress" + if len(typeSet) == 0 && libraryID == nil && !query.enableTotalRecordCount && (!resumeFiltered || query.startIndex == 0) { batchSize := min(max(query.limit*2, 48), 200) if batchSize <= 0 { batchSize = 48 @@ -1835,6 +1840,13 @@ func (h *ItemsHandler) loadProgressPage(ctx context.Context, session *Session, s if len(progressEntries) == 0 { break } + rawCount := len(progressEntries) + if resumeFiltered { + progressEntries, err = h.userData.FilterResumeProgress(ctx, session, progressEntries) + if err != nil { + return nil, 0, err + } + } items, err := h.hydrateProgressItems(ctx, session, progressEntries, query.requestedFields, libraryID) if err != nil { @@ -1846,10 +1858,10 @@ func (h *ItemsHandler) loadProgressPage(ctx context.Context, session *Session, s } result = append(result, item) } - if len(result) >= query.limit || len(progressEntries) < batchSize { + if len(result) >= query.limit || rawCount < batchSize { break } - offset += len(progressEntries) + offset += rawCount } return h.finishProgressPage(ctx, session, result, query, libraryID), 0, nil } @@ -1871,6 +1883,13 @@ func (h *ItemsHandler) loadProgressPage(ctx context.Context, session *Session, s if len(progressEntries) == 0 { break } + rawCount := len(progressEntries) + if resumeFiltered { + progressEntries, err = h.userData.FilterResumeProgress(ctx, session, progressEntries) + if err != nil { + return nil, 0, err + } + } hydrated, err := h.hydrateProgressItems(ctx, session, progressEntries, query.requestedFields, libraryID) if err != nil { @@ -1886,8 +1905,8 @@ func (h *ItemsHandler) loadProgressPage(ctx context.Context, session *Session, s matchedCount++ } - offset += len(progressEntries) - if len(progressEntries) < batchSize { + offset += rawCount + if rawCount < batchSize { break } if !query.enableTotalRecordCount && len(items) >= query.limit { diff --git a/internal/jellycompat/resume_filter_test.go b/internal/jellycompat/resume_filter_test.go new file mode 100644 index 00000000..c022bf3b --- /dev/null +++ b/internal/jellycompat/resume_filter_test.go @@ -0,0 +1,222 @@ +package jellycompat + +import ( + "context" + "testing" + + "github.com/Silo-Server/silo-server/internal/catalog" + "github.com/Silo-Server/silo-server/internal/config" + "github.com/Silo-Server/silo-server/internal/models" + "github.com/Silo-Server/silo-server/internal/userstore" +) + +// resumeFilteringUserData serves a fixed in-progress list and hides a chosen +// set of media item IDs from FilterResumeProgress, recording which statuses +// were listed and how often the filter ran. +type resumeFilteringUserData struct { + mockUserDataService + entries []upstreamProgress + hidden map[string]bool + listedStatuses []string + filterCalls int + filteredBatchSize []int +} + +func (s *resumeFilteringUserData) ListProgress(_ context.Context, _ *Session, status string, limit, offset int) ([]upstreamProgress, error) { + s.listedStatuses = append(s.listedStatuses, status) + if offset >= len(s.entries) { + return nil, nil + } + end := min(offset+limit, len(s.entries)) + return s.entries[offset:end], nil +} + +func (s *resumeFilteringUserData) FilterResumeProgress(_ context.Context, _ *Session, entries []upstreamProgress) ([]upstreamProgress, error) { + s.filterCalls++ + s.filteredBatchSize = append(s.filteredBatchSize, len(entries)) + kept := make([]upstreamProgress, 0, len(entries)) + for _, entry := range entries { + if s.hidden[entry.MediaItemID] { + continue + } + kept = append(kept, entry) + } + return kept, nil +} + +func resumeTestHandler(userData UserDataService, itemsByID map[string]*models.MediaItem) *ItemsHandler { + codec := NewResourceIDCodec() + return &ItemsHandler{ + content: &stubContentService{detail: &upstreamItemDetail{}}, + userData: userData, + itemRepo: &countingItemRepo{itemsByID: itemsByID}, + codec: codec, + mapper: newMapper(codec, &config.Config{}), + } +} + +func dtoNames(dtos []baseItemDTO) []string { + names := make([]string, 0, len(dtos)) + for _, dto := range dtos { + names = append(names, dto.Name) + } + return names +} + +// TestLoadProgressPage_ResumeHidesFilteredEntries pins that the Resume path +// runs every raw batch through FilterResumeProgress so dismissed and +// superseded entries never reach Jellyfin clients, mirroring the first-party +// Continue Watching row. +func TestLoadProgressPage_ResumeHidesFilteredEntries(t *testing.T) { + items := map[string]*models.MediaItem{ + "movie-1": {ContentID: "movie-1", Type: "movie", Title: "Movie One"}, + "movie-2": {ContentID: "movie-2", Type: "movie", Title: "Movie Two"}, + "movie-3": {ContentID: "movie-3", Type: "movie", Title: "Movie Three"}, + } + userData := &resumeFilteringUserData{ + entries: []upstreamProgress{ + {MediaItemID: "movie-1", PositionSeconds: 10, DurationSeconds: 100}, + {MediaItemID: "movie-2", PositionSeconds: 20, DurationSeconds: 100}, + {MediaItemID: "movie-3", PositionSeconds: 30, DurationSeconds: 100}, + }, + hidden: map[string]bool{"movie-2": true}, + } + h := resumeTestHandler(userData, items) + + session := &Session{StreamAppUserID: 1, ProfileID: "profile-1"} + dtos, _, err := h.loadProgressPage(context.Background(), session, "in_progress", itemsQuery{limit: 10}, nil, nil) + if err != nil { + t.Fatalf("loadProgressPage: %v", err) + } + + got := dtoNames(dtos) + want := []string{"Movie One", "Movie Three"} + if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] { + t.Fatalf("resume names = %v, want %v", got, want) + } + if userData.filterCalls == 0 { + t.Fatal("expected FilterResumeProgress to be called for in_progress status") + } +} + +// TestLoadProgressPage_ResumeFilterAppliesWithTypeFilter covers the +// scan-from-zero branch (IncludeItemTypes present): filtering must apply +// there too, and visible pagination must skip filtered entries. +func TestLoadProgressPage_ResumeFilterAppliesWithTypeFilter(t *testing.T) { + items := map[string]*models.MediaItem{ + "movie-1": {ContentID: "movie-1", Type: "movie", Title: "Movie One"}, + "movie-2": {ContentID: "movie-2", Type: "movie", Title: "Movie Two"}, + "movie-3": {ContentID: "movie-3", Type: "movie", Title: "Movie Three"}, + } + userData := &resumeFilteringUserData{ + entries: []upstreamProgress{ + {MediaItemID: "movie-1", PositionSeconds: 10, DurationSeconds: 100}, + {MediaItemID: "movie-2", PositionSeconds: 20, DurationSeconds: 100}, + {MediaItemID: "movie-3", PositionSeconds: 30, DurationSeconds: 100}, + }, + hidden: map[string]bool{"movie-1": true}, + } + h := resumeTestHandler(userData, items) + + session := &Session{StreamAppUserID: 1, ProfileID: "profile-1"} + query := itemsQuery{limit: 10, enableTotalRecordCount: true} + typeSet := map[string]bool{"movie": true} + dtos, total, err := h.loadProgressPage(context.Background(), session, "in_progress", query, typeSet, nil) + if err != nil { + t.Fatalf("loadProgressPage: %v", err) + } + + got := dtoNames(dtos) + want := []string{"Movie Two", "Movie Three"} + if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] { + t.Fatalf("resume names = %v, want %v", got, want) + } + if total != 2 { + t.Fatalf("TotalRecordCount = %d, want 2 (hidden entries excluded)", total) + } +} + +// TestLoadProgressPage_CompletedSkipsResumeFilter pins that the watched-items +// view ("completed" status) is served unfiltered: dismissals and superseded +// hiding are Continue Watching semantics only. +func TestLoadProgressPage_CompletedSkipsResumeFilter(t *testing.T) { + items := map[string]*models.MediaItem{ + "movie-1": {ContentID: "movie-1", Type: "movie", Title: "Movie One"}, + } + userData := &resumeFilteringUserData{ + entries: []upstreamProgress{ + {MediaItemID: "movie-1", PositionSeconds: 100, DurationSeconds: 100, Completed: true}, + }, + hidden: map[string]bool{"movie-1": true}, + } + h := resumeTestHandler(userData, items) + + session := &Session{StreamAppUserID: 1, ProfileID: "profile-1"} + dtos, _, err := h.loadProgressPage(context.Background(), session, "completed", itemsQuery{limit: 10}, nil, nil) + if err != nil { + t.Fatalf("loadProgressPage: %v", err) + } + + if userData.filterCalls != 0 { + t.Fatalf("FilterResumeProgress calls = %d, want 0 for completed status", userData.filterCalls) + } + if len(dtos) != 1 || dtos[0].Name != "Movie One" { + t.Fatalf("completed names = %v, want [Movie One]", dtoNames(dtos)) + } +} + +// fakeDismissalStore embeds userstore.UserStore so only the methods +// FilterResumeProgress touches need real implementations. +type fakeDismissalStore struct { + userstore.UserStore + dismissals []userstore.HomeItemDismissal +} + +func (s *fakeDismissalStore) ListHomeDismissals(_ context.Context, _ string, _ string) ([]userstore.HomeItemDismissal, error) { + return s.dismissals, nil +} + +func (s *fakeDismissalStore) ListProgress(_ context.Context, _ string, _ string, _ int, _ int) ([]userstore.WatchProgress, error) { + return nil, nil +} + +type fakeDismissalStoreProvider struct { + store userstore.UserStore +} + +func (p *fakeDismissalStoreProvider) ForUser(context.Context, int) (userstore.UserStore, error) { + return p.store, nil +} + +func (p *fakeDismissalStoreProvider) Close() error { return nil } + +// TestDirectUserDataServiceFilterResumeProgressDropsDismissedEntries covers +// the glue in directUserDataService: dismissals from the store hide matching +// entries (same timestamp rule as the first-party row), everything else +// passes through. The superseded check is disabled via a nil pool — its +// behavior is pinned by the catalog package tests. +func TestDirectUserDataServiceFilterResumeProgressDropsDismissedEntries(t *testing.T) { + dismissedAt := "2025-01-01T00:00:00Z" + svc := &directUserDataService{ + storeProvider: &fakeDismissalStoreProvider{store: &fakeDismissalStore{ + dismissals: []userstore.HomeItemDismissal{ + {MediaItemID: "movie-1", ProgressUpdatedAt: &dismissedAt}, + }, + }}, + resumeFilter: catalog.NewContinueWatchingProgressFilter(nil), + } + + entries := []upstreamProgress{ + {MediaItemID: "movie-1", PositionSeconds: 10, DurationSeconds: 100, UpdatedAt: dismissedAt}, + {MediaItemID: "movie-2", PositionSeconds: 20, DurationSeconds: 100, UpdatedAt: dismissedAt}, + } + + session := &Session{StreamAppUserID: 1, ProfileID: "profile-1"} + got, err := svc.FilterResumeProgress(context.Background(), session, entries) + if err != nil { + t.Fatalf("FilterResumeProgress: %v", err) + } + if len(got) != 1 || got[0].MediaItemID != "movie-2" { + t.Fatalf("filtered = %+v, want only movie-2", got) + } +} diff --git a/internal/jellycompat/router.go b/internal/jellycompat/router.go index 74e87647..063a8211 100644 --- a/internal/jellycompat/router.go +++ b/internal/jellycompat/router.go @@ -10,6 +10,7 @@ import ( "github.com/go-chi/chi/v5" "github.com/go-chi/chi/v5/middleware" "github.com/go-chi/cors" + "github.com/jackc/pgx/v5/pgxpool" "gopkg.in/natefinch/lumberjack.v2" "github.com/Silo-Server/silo-server/internal/catalog" @@ -297,12 +298,17 @@ func withDefaults(deps Dependencies) Dependencies { if deps.DB != nil { staler = recommendations.NewRepo(deps.DB) } + var pool *pgxpool.Pool + if deps.BrowseRepo != nil { + pool = deps.BrowseRepo.Pool() + } deps.UserDataService = newDirectUserDataService( deps.UserStoreProvider, deps.ItemRepo, deps.EpisodeRepo, deps.ProviderIDRepo, deps.DetailSvc, + catalog.NewContinueWatchingProgressFilter(pool), staler, deps.RecWorker, ) diff --git a/internal/jellycompat/userdata_direct.go b/internal/jellycompat/userdata_direct.go index 5fde5abb..a2e82811 100644 --- a/internal/jellycompat/userdata_direct.go +++ b/internal/jellycompat/userdata_direct.go @@ -3,6 +3,7 @@ package jellycompat import ( "context" "fmt" + "log/slog" "time" "github.com/Silo-Server/silo-server/internal/catalog" @@ -16,6 +17,7 @@ type directUserDataService struct { itemRepo *catalog.ItemRepository detailSvc *catalog.DetailService watchState *watchstate.Service + resumeFilter *catalog.ContinueWatchingProgressFilter profileStaler profileStaler profileRefreshRequester profileRefreshRequester } @@ -26,6 +28,7 @@ func newDirectUserDataService( episodeRepo *catalog.EpisodeRepository, providerIDRepo *catalog.ProviderIDRepository, detailSvc *catalog.DetailService, + resumeFilter *catalog.ContinueWatchingProgressFilter, staler profileStaler, requester profileRefreshRequester, ) *directUserDataService { @@ -36,6 +39,7 @@ func newDirectUserDataService( watchState: watchstate.NewService(storeProvider).WithStableIdentityResolver( watchstate.NewStableIdentityResolver(itemRepo, episodeRepo, providerIDRepo), ), + resumeFilter: resumeFilter, profileStaler: staler, profileRefreshRequester: requester, } @@ -156,6 +160,44 @@ func (s *directUserDataService) ListProgress(ctx context.Context, session *Sessi return result, nil } +// FilterResumeProgress applies the same hiding rules as the first-party +// Continue Watching fetcher: dismissed entries and episodes superseded by a +// later-completed episode in the same series. +func (s *directUserDataService) FilterResumeProgress(ctx context.Context, session *Session, entries []upstreamProgress) ([]upstreamProgress, error) { + if len(entries) == 0 { + return entries, nil + } + store, err := s.storeProvider.ForUser(ctx, session.StreamAppUserID) + if err != nil { + return nil, fmt.Errorf("open user store: %w", err) + } + + progress := make([]userstore.WatchProgress, 0, len(entries)) + for _, entry := range entries { + progress = append(progress, fromUpstreamProgress(entry)) + } + + // Dismissal lookup failures degrade to showing the entries, matching the + // first-party fetcher. + if dismissals, err := store.ListHomeDismissals(ctx, session.ProfileID, userstore.HomeSurfaceContinueWatching); err != nil { + slog.Error("listing continue watching dismissals", "profile_id", session.ProfileID, "error", err) + } else { + progress = catalog.NewHomeDismissalIndex(dismissals).FilterProgress(progress) + } + + superseded, err := s.resumeFilter.SupersededEpisodeProgressIDs(ctx, store, session.ProfileID, progress) + if err != nil { + return nil, fmt.Errorf("filter superseded progress: %w", err) + } + progress = catalog.FilterSupersededProgress(progress, superseded) + + result := make([]upstreamProgress, 0, len(progress)) + for _, entry := range progress { + result = append(result, toUpstreamProgress(entry)) + } + return result, nil +} + func (s *directUserDataService) ListProgressByMediaItems(ctx context.Context, session *Session, mediaItemIDs []string) (map[string]*upstreamProgress, error) { store, err := s.storeProvider.ForUser(ctx, session.StreamAppUserID) if err != nil { @@ -252,3 +294,13 @@ func toUpstreamProgress(entry userstore.WatchProgress) upstreamProgress { UpdatedAt: entry.UpdatedAt, } } + +func fromUpstreamProgress(entry upstreamProgress) userstore.WatchProgress { + return userstore.WatchProgress{ + MediaItemID: entry.MediaItemID, + PositionSeconds: entry.PositionSeconds, + DurationSeconds: entry.DurationSeconds, + Completed: entry.Completed, + UpdatedAt: entry.UpdatedAt, + } +} diff --git a/internal/sections/fetcher.go b/internal/sections/fetcher.go index d08f7b89..fa5a14e7 100644 --- a/internal/sections/fetcher.go +++ b/internal/sections/fetcher.go @@ -68,6 +68,7 @@ type trendingSnapshotGetter interface { // Fetcher runs section queries against the database. type Fetcher struct { pool *pgxpool.Pool + progressFilter *catalog.ContinueWatchingProgressFilter StoreProvider userstore.UserStoreProvider CollectionRepo *catalog.LibraryCollectionRepository RecommendationRepo *recommendations.Repo // retained for non-reader call sites @@ -91,7 +92,11 @@ type Fetcher struct { // NewFetcher creates a new section Fetcher. func NewFetcher(pool *pgxpool.Pool) *Fetcher { - return &Fetcher{pool: pool, Clock: recipes.RealClock{}} + return &Fetcher{ + pool: pool, + progressFilter: catalog.NewContinueWatchingProgressFilter(pool), + Clock: recipes.RealClock{}, + } } type editorialCandidateLoader func(context.Context, string, *int, []int, catalog.AccessFilter) ([]string, error) @@ -385,7 +390,7 @@ func (f *Fetcher) fetchContinueWatchingSection(ctx context.Context, resolved Res break } rawProgressCount := len(progressEntries) - progressEntries = filterContinueWatchingDismissals(progressEntries, dismissals) + progressEntries = dismissals.FilterProgress(progressEntries) pageItems, pageMeta, err := f.fetchContinueProgressItems(ctx, store, profileID, progressEntries, continueType, effectiveLibID, effectiveLibraryIDs, filter) if err != nil { @@ -490,11 +495,11 @@ func (f *Fetcher) fetchContinueProgressItems(ctx context.Context, store userstor matchingEntries = append(matchingEntries, entry) } if ContinueTypeAllowsNextUp(continueType) && hasEpisodeEntries { - supersededEpisodeProgress, err := f.fetchSupersededEpisodeProgressIDs(ctx, store, profileID, matchingEntries) + supersededEpisodeProgress, err := f.progressFilter.SupersededEpisodeProgressIDs(ctx, store, profileID, matchingEntries) if err != nil { return nil, nil, err } - matchingEntries = filterSupersededEpisodeProgressEntries(matchingEntries, supersededEpisodeProgress) + matchingEntries = catalog.FilterSupersededProgress(matchingEntries, supersededEpisodeProgress) } itemMeta := make(map[string]SectionItemMeta, len(matchingEntries)) @@ -633,142 +638,6 @@ func (f *Fetcher) FetchNextUpItems(ctx context.Context, userID int, profileID st return orderedItems, meta, nil } -type progressLister interface { - ListProgress(ctx context.Context, profileID, status string, limit, offset int) ([]userstore.WatchProgress, error) -} - -const supersededProgressPageSize = 500 - -type progressSnapshot struct { - ContentID string - UpdatedAt time.Time -} - -func (f *Fetcher) fetchSupersededEpisodeProgressIDs(ctx context.Context, store progressLister, profileID string, entries []userstore.WatchProgress) (map[string]struct{}, error) { - inProgress := progressSnapshots(entries) - if len(inProgress) == 0 { - return map[string]struct{}{}, nil - } - - completed, err := completedProgressSnapshots(ctx, store, profileID) - if err != nil { - return nil, err - } - if len(completed) == 0 { - return map[string]struct{}{}, nil - } - - inProgressIDs, inProgressUpdatedAts := splitProgressSnapshots(inProgress) - completedIDs, completedUpdatedAts := splitProgressSnapshots(completed) - query := buildSupersededEpisodeProgressQuery() - rows, err := f.pool.Query(ctx, query, inProgressIDs, inProgressUpdatedAts, completedIDs, completedUpdatedAts) - if err != nil { - return nil, fmt.Errorf("querying superseded episode progress: %w", err) - } - defer rows.Close() - - superseded := make(map[string]struct{}) - for rows.Next() { - var mediaItemID string - if err := rows.Scan(&mediaItemID); err != nil { - return nil, fmt.Errorf("scanning superseded episode progress: %w", err) - } - superseded[mediaItemID] = struct{}{} - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("iterating superseded episode progress: %w", err) - } - return superseded, nil -} - -func completedProgressSnapshots(ctx context.Context, store progressLister, profileID string) ([]progressSnapshot, error) { - seen := make(map[string]struct{}) - snapshots := make([]progressSnapshot, 0) - - for offset := 0; ; offset += supersededProgressPageSize { - entries, err := store.ListProgress(ctx, profileID, "completed", supersededProgressPageSize, offset) - if err != nil { - return nil, fmt.Errorf("listing completed progress for superseded episodes: %w", err) - } - - for _, snapshot := range progressSnapshots(entries) { - contentID := snapshot.ContentID - if _, ok := seen[contentID]; ok { - continue - } - seen[contentID] = struct{}{} - snapshots = append(snapshots, snapshot) - } - - if len(entries) < supersededProgressPageSize { - return snapshots, nil - } - } -} - -func progressSnapshots(entries []userstore.WatchProgress) []progressSnapshot { - snapshots := make([]progressSnapshot, 0, len(entries)) - for _, entry := range entries { - contentID := strings.TrimSpace(entry.MediaItemID) - if contentID == "" { - continue - } - updatedAt, err := time.Parse(time.RFC3339, entry.UpdatedAt) - if err != nil || updatedAt.IsZero() { - continue - } - snapshots = append(snapshots, progressSnapshot{ - ContentID: contentID, - UpdatedAt: updatedAt.UTC(), - }) - } - return snapshots -} - -func splitProgressSnapshots(snapshots []progressSnapshot) ([]string, []time.Time) { - contentIDs := make([]string, len(snapshots)) - updatedAts := make([]time.Time, len(snapshots)) - for i, snapshot := range snapshots { - contentIDs[i] = snapshot.ContentID - updatedAts[i] = snapshot.UpdatedAt - } - return contentIDs, updatedAts -} - -func buildSupersededEpisodeProgressQuery() string { - return ` - WITH in_progress(content_id, updated_at) AS ( - SELECT * FROM unnest($1::text[], $2::timestamptz[]) - ), - completed(content_id, updated_at) AS ( - SELECT * FROM unnest($3::text[], $4::timestamptz[]) - ) - SELECT DISTINCT ip.content_id - FROM in_progress ip_progress - JOIN episodes ip ON ip.content_id = ip_progress.content_id - JOIN episodes done - ON done.series_id = ip.series_id - AND (done.season_number, done.episode_number) > (ip.season_number, ip.episode_number) - JOIN completed done_progress - ON done_progress.content_id = done.content_id - WHERE done_progress.updated_at > ip_progress.updated_at` -} - -func filterSupersededEpisodeProgressEntries(entries []userstore.WatchProgress, superseded map[string]struct{}) []userstore.WatchProgress { - if len(entries) == 0 || len(superseded) == 0 { - return entries - } - - filtered := make([]userstore.WatchProgress, 0, len(entries)) - for _, entry := range entries { - if _, ok := superseded[entry.MediaItemID]; ok { - continue - } - filtered = append(filtered, entry) - } - return filtered -} - func collapseContinueWatchingSeriesCandidates(items []*models.MediaItem, meta map[string]SectionItemMeta) []*models.MediaItem { selectedBySeries := make(map[string]int) result := make([]*models.MediaItem, 0, len(items)) @@ -830,32 +699,13 @@ func episodeOrdinal(meta SectionItemMeta) int { return season*100000 + episode } -func (f *Fetcher) listContinueWatchingDismissals(ctx context.Context, store userstore.UserStore, profileID string) map[string]userstore.HomeItemDismissal { +func (f *Fetcher) listContinueWatchingDismissals(ctx context.Context, store userstore.UserStore, profileID string) catalog.HomeDismissalIndex { dismissals, err := store.ListHomeDismissals(ctx, profileID, userstore.HomeSurfaceContinueWatching) if err != nil { slog.Error("listing continue watching dismissals", "profile_id", profileID, "error", err) - return map[string]userstore.HomeItemDismissal{} + return catalog.HomeDismissalIndex{} } - dismissalByItemID := make(map[string]userstore.HomeItemDismissal, len(dismissals)) - for _, dismissal := range dismissals { - dismissalByItemID[dismissal.MediaItemID] = dismissal - } - return dismissalByItemID -} - -func filterContinueWatchingDismissals(entries []userstore.WatchProgress, dismissalByItemID map[string]userstore.HomeItemDismissal) []userstore.WatchProgress { - if len(entries) == 0 || len(dismissalByItemID) == 0 { - return entries - } - - filtered := make([]userstore.WatchProgress, 0, len(entries)) - for _, entry := range entries { - dismissal, ok := dismissalByItemID[entry.MediaItemID] - if !ok || dismissal.ProgressUpdatedAt == nil || *dismissal.ProgressUpdatedAt != entry.UpdatedAt { - filtered = append(filtered, entry) - } - } - return filtered + return catalog.NewHomeDismissalIndex(dismissals) } func (f *Fetcher) filterNextUpDismissals(ctx context.Context, userID int, profileID string, results []catalog.NextUpResult) []catalog.NextUpResult { diff --git a/internal/sections/fetcher_continue_watching_test.go b/internal/sections/fetcher_continue_watching_test.go index 53d74b81..cd8a217d 100644 --- a/internal/sections/fetcher_continue_watching_test.go +++ b/internal/sections/fetcher_continue_watching_test.go @@ -1,13 +1,10 @@ package sections import ( - "context" - "strings" "testing" "time" "github.com/Silo-Server/silo-server/internal/models" - "github.com/Silo-Server/silo-server/internal/userstore" ) func TestCollapseContinueWatchingSeriesCandidatesPrefersNewestInProgressEpisode(t *testing.T) { @@ -87,26 +84,6 @@ func TestCollapseContinueWatchingSeriesCandidatesKeepsNextUpWhenNoInProgress(t * } } -func TestFilterSupersededEpisodeProgressEntriesDropsOlderPartialsAfterLaterCompletedEpisode(t *testing.T) { - t.Parallel() - - entries := []userstore.WatchProgress{ - {MediaItemID: "boys-s1e1"}, - {MediaItemID: "boys-s5e3"}, - {MediaItemID: "movie-1"}, - } - superseded := map[string]struct{}{ - "boys-s1e1": {}, - "boys-s5e3": {}, - } - - filtered := filterSupersededEpisodeProgressEntries(entries, superseded) - - if len(filtered) != 1 || filtered[0].MediaItemID != "movie-1" { - t.Fatalf("filtered entries = %+v, want only movie-1", filtered) - } -} - func TestMatchesContinueWatchingFilterIncludesAudiobooks(t *testing.T) { t.Parallel() @@ -171,62 +148,6 @@ func TestParseContinueTypeRejectsUnknownExplicitType(t *testing.T) { } } -func TestCompletedProgressSnapshotsPagesThroughConfiguredStore(t *testing.T) { - t.Parallel() - - entries := make([]userstore.WatchProgress, supersededProgressPageSize+1) - for i := range entries { - entries[i] = userstore.WatchProgress{ - MediaItemID: "done-" + time.Unix(int64(i), 0).Format("150405"), - UpdatedAt: time.Date(2025, 1, 1, 0, 0, i, 0, time.UTC).Format(time.RFC3339), - } - } - store := &stubProgressLister{entries: entries} - - snapshots, err := completedProgressSnapshots(context.Background(), store, "p1") - if err != nil { - t.Fatalf("completedProgressSnapshots: %v", err) - } - if len(snapshots) != len(entries) { - t.Fatalf("completed snapshots count = %d, want %d", len(snapshots), len(entries)) - } - if len(store.calls) != 2 { - t.Fatalf("ListProgress calls = %+v, want 2 paged calls", store.calls) - } - if store.calls[0] != (progressListCall{profileID: "p1", status: "completed", limit: supersededProgressPageSize, offset: 0}) { - t.Fatalf("first ListProgress call = %+v", store.calls[0]) - } - if store.calls[1] != (progressListCall{profileID: "p1", status: "completed", limit: supersededProgressPageSize, offset: supersededProgressPageSize}) { - t.Fatalf("second ListProgress call = %+v", store.calls[1]) - } -} - -func TestBuildSupersededEpisodeProgressQueryUsesStoreSnapshotsWithFreshnessGate(t *testing.T) { - t.Parallel() - - query := buildSupersededEpisodeProgressQuery() - expectedFragments := []string{ - "unnest($1::text[], $2::timestamptz[])", - "unnest($3::text[], $4::timestamptz[])", - "FROM in_progress ip_progress", - "done_progress.updated_at > ip_progress.updated_at", - } - for _, fragment := range expectedFragments { - if !strings.Contains(query, fragment) { - t.Fatalf("expected superseded progress query to contain %q, got:\n%s", fragment, query) - } - } - unexpectedFragments := []string{ - "user_watch_progress", - "user_history_hidden_items", - } - for _, fragment := range unexpectedFragments { - if strings.Contains(query, fragment) { - t.Fatalf("superseded progress query contains %q, got:\n%s", fragment, query) - } - } -} - func contentIDs(items []*models.MediaItem) []string { ids := make([]string, 0, len(items)) for _, item := range items { @@ -238,32 +159,3 @@ func contentIDs(items []*models.MediaItem) []string { func intPtr(v int) *int { return &v } - -type progressListCall struct { - profileID string - status string - limit int - offset int -} - -type stubProgressLister struct { - entries []userstore.WatchProgress - calls []progressListCall -} - -func (s *stubProgressLister) ListProgress(_ context.Context, profileID, status string, limit, offset int) ([]userstore.WatchProgress, error) { - s.calls = append(s.calls, progressListCall{ - profileID: profileID, - status: status, - limit: limit, - offset: offset, - }) - if offset >= len(s.entries) { - return nil, nil - } - end := offset + limit - if end > len(s.entries) { - end = len(s.entries) - } - return s.entries[offset:end], nil -}