fix(jellycompat): hide dismissed and superseded entries from Resume (#114)

The jellycompat /UserItems/Resume endpoint served the raw in_progress
list, so Jellyfin clients showed every half-watched episode a profile
ever abandoned, including stale entries from earlier seasons of shows
the viewer had long moved past. The first-party Continue Watching row
already hid those via dismissal filtering and the superseded-episode
check, but that logic was private to internal/sections.

Extract the shared rules into internal/catalog
(ContinueWatchingProgressFilter, HomeDismissalIndex) and apply them to
the compat Resume path through a new UserDataService method,
FilterResumeProgress. The sections fetcher now delegates to the same
code, so both surfaces agree on what "still watching" means.

Resume pagination keeps advancing by raw batch counts so filtering
cannot terminate scans early, and the raw-offset fast path is limited
to StartIndex=0 for Resume because filtered lists make raw offsets
diverge from the visible list. The watched-items view ("completed"
status) stays unfiltered, and the per-series collapse deliberately
stays first-party only to preserve Jellyfin endpoint semantics.

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Quick
2026-06-09 18:01:36 -04:00
committed by GitHub
co-authored by Claude Fable 5
parent 0d0d7bab1e
commit b246f271cb
10 changed files with 726 additions and 275 deletions
@@ -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
}
@@ -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
}
@@ -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")
}
+5
View File
@@ -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
+24 -5
View File
@@ -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 {
+222
View File
@@ -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)
}
}
+6
View File
@@ -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,
)
+52
View File
@@ -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,
}
}
+12 -162
View File
@@ -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 {
@@ -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
}