diff --git a/internal/catalog/nextup_repo.go b/internal/catalog/nextup_repo.go index 107e6ba7..ce77f48c 100644 --- a/internal/catalog/nextup_repo.go +++ b/internal/catalog/nextup_repo.go @@ -3,6 +3,7 @@ package catalog import ( "context" "fmt" + "log/slog" "time" "github.com/jackc/pgx/v5/pgxpool" @@ -57,27 +58,63 @@ func (r *NextUpRepository) ListNextUp(ctx context.Context, q NextUpQuery) ([]Nex limit = 20 } - query, args := buildListNextUpQuery(q, limit) - - rows, err := r.pool.Query(ctx, query, args...) - if err != nil { - return nil, fmt.Errorf("querying next-up episodes: %w", err) - } - defer rows.Close() - var results []NextUpResult - for rows.Next() { - var res NextUpResult - if err := rows.Scan( - &res.ContentID, &res.SeriesID, &res.SeriesTitle, - &res.SeasonNumber, &res.EpisodeNumber, &res.CompletedAt, - ); err != nil { - return nil, fmt.Errorf("scanning next-up row: %w", err) + if q.SeriesID != "" { + query, args := buildListNextUpQuery(q, limit, nil) + rows, err := r.pool.Query(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("querying next-up episodes: %w", err) + } + defer rows.Close() + + for rows.Next() { + var res NextUpResult + if err := rows.Scan( + &res.ContentID, &res.SeriesID, &res.SeriesTitle, + &res.SeasonNumber, &res.EpisodeNumber, &res.CompletedAt, + ); err != nil { + return nil, fmt.Errorf("scanning next-up row: %w", err) + } + results = append(results, res) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterating next-up rows: %w", err) + } + } else { + remaining := limit + var cursor *nextUpWalkCursor + batchesRun := 0 + exhausted := false + + // Every batch resumes below the previous compound frontier, so appending + // batches preserves strictly descending (updated_at, media_item_id) order. + for batch := 0; batch < nextUpAnchorMaxBatches; batch++ { + batchResults, frontier, err := r.listNextUpBatch(ctx, q, remaining, cursor) + if err != nil { + return nil, err + } + batchesRun++ + results = append(results, batchResults...) + remaining -= len(batchResults) + + if remaining <= 0 { + exhausted = true + break + } + if frontier == nil || frontier.n < nextUpAnchorMaxSeries { + exhausted = true + break + } + cursor = &frontier.nextUpWalkCursor + } + + if !exhausted && remaining > 0 { + slog.WarnContext(ctx, "next-up anchor walk hit batch cap; eligible series tail left unscanned", + "component", "catalog", + "profile_id", q.ProfileID, + "batches", batchesRun, + "results_found", len(results)) } - results = append(results, res) - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("iterating next-up rows: %w", err) } if q.EnableResumable { @@ -109,22 +146,82 @@ func (r *NextUpRepository) ListNextUp(ctx context.Context, q NextUpQuery) ([]Nex return results, nil } -// nextUpAnchorMaxRows bounds how many of the profile's most recent completed -// rows the global next-up query considers when deriving per-series anchors. -// Without a bound the completed_episodes CTE re-reads the profile's entire -// completed history per call — 233k rows joined to episodes for the worst -// bulk-import profile, measured at up to 44.7s. A next-up rail surfaces ~24 -// series; the 500 most recently completed episodes cover every series that can -// realistically rank on it (a series whose last completed episode is older -// than 500 completions of other content sorts far past the rail's limit). -// Progress rows are recency-ordered on idx_uwp_profile_completed, so the -// bounded anchor scan is an ordered index walk. Series-scoped calls (the -// show-detail tile) stay unbounded: they must anchor on the series' last -// completed episode no matter how long ago it was watched, and are naturally -// bounded by one series. -const nextUpAnchorMaxRows = 500 +type nextUpWalkCursor struct { + updatedAt time.Time + mediaItemID string + seen []string +} -func buildListNextUpQuery(q NextUpQuery, limit int) (string, []interface{}) { +type nextUpWalkFrontier struct { + nextUpWalkCursor + n int +} + +// nextUpAnchorMaxSeries bounds the number of distinct series anchors visited by +// one global next-up batch. A row-based cap let one series consume the window; +// counting distinct series prevents that, while batching continues past a full +// batch whose anchors are all ineligible. Series-scoped calls stay unbounded so +// they can anchor on the series' last completed episode regardless of age. +const nextUpAnchorMaxSeries = 96 + +// nextUpAnchorMaxBatches bounds a global call to 10 × 96 = 960 visited series +// as a runaway guard. +const nextUpAnchorMaxBatches = 10 + +func (r *NextUpRepository) listNextUpBatch( + ctx context.Context, + q NextUpQuery, + limit int, + cursor *nextUpWalkCursor, +) ([]NextUpResult, *nextUpWalkFrontier, error) { + query, args := buildListNextUpQuery(q, limit, cursor) + rows, err := r.pool.Query(ctx, query, args...) + if err != nil { + return nil, nil, fmt.Errorf("querying next-up episodes: %w", err) + } + defer rows.Close() + + var results []NextUpResult + var frontier *nextUpWalkFrontier + for rows.Next() { + currentFrontier := nextUpWalkFrontier{} + var contentID, seriesID, seriesTitle *string + var seasonNumber, episodeNumber *int + var completedAt *time.Time + if err := rows.Scan( + ¤tFrontier.updatedAt, + ¤tFrontier.mediaItemID, + ¤tFrontier.seen, + ¤tFrontier.n, + &contentID, + &seriesID, + &seriesTitle, + &seasonNumber, + &episodeNumber, + &completedAt, + ); err != nil { + return nil, nil, fmt.Errorf("scanning next-up row: %w", err) + } + frontier = ¤tFrontier + if contentID == nil { + continue + } + results = append(results, NextUpResult{ + ContentID: *contentID, + SeriesID: *seriesID, + SeriesTitle: *seriesTitle, + SeasonNumber: *seasonNumber, + EpisodeNumber: *episodeNumber, + CompletedAt: *completedAt, + }) + } + if err := rows.Err(); err != nil { + return nil, nil, fmt.Errorf("iterating next-up rows: %w", err) + } + return results, frontier, nil +} + +func buildListNextUpQuery(q NextUpQuery, limit int, cursor *nextUpWalkCursor) (string, []interface{}) { args := []interface{}{q.UserID, q.ProfileID, limit} argIdx := 4 @@ -142,6 +239,21 @@ func buildListNextUpQuery(q NextUpQuery, limit int) (string, []interface{}) { argIdx++ } + // seedSeen is projected alongside the seed's picked series, which the global + // walk exposes as `pick`; the series-scoped branch has no walk and no seed. + cursorFilter := "" + seedSeen := "ARRAY[pick.series_id]" + if q.SeriesID == "" && cursor != nil { + cursorUpdatedAtArg := argIdx + cursorMediaItemIDArg := argIdx + 1 + cursorSeenArg := argIdx + 2 + cursorFilter = fmt.Sprintf(` + AND (uwp.updated_at, uwp.media_item_id) < ($%d, $%d) + AND NOT (e.series_id = ANY($%d))`, cursorUpdatedAtArg, cursorMediaItemIDArg, cursorSeenArg) + seedSeen = fmt.Sprintf("$%d::text[] || pick.series_id", cursorSeenArg) + args = append(args, cursor.updatedAt, cursor.mediaItemID, cursor.seen) + } + // When resumable items are disabled, suppress next-up only if the series has // newer in-progress activity than the most recent completed episode. Older // partial watches should not block the user's current progression. @@ -169,11 +281,16 @@ func buildListNextUpQuery(q NextUpQuery, limit int) (string, []interface{}) { sourceTable = "eligible_series" } - // Global queries derive series anchors from a bounded recent-completions - // scan (see nextUpAnchorMaxRows); series-scoped queries keep the unbounded - // shape so the anchor is the series' last completed episode regardless of - // age. The hidden-items exclusion and date cutoff apply inside the bounded - // scan so hidden/old rows never consume the anchor budget. + // Global queries use a recursive loose-index scan to find at most + // nextUpAnchorMaxSeries distinct series without letting one series' rows + // consume the budget. The compound cursor is the total order required when + // bulk updates give many rows the same timestamp; because media_item_id is + // unique it can only order the walk, never decide which episode of a series + // anchors it — see the anchor lateral below. + // Series-scoped queries keep the unbounded shape so the anchor is the + // series' last completed episode regardless of age. Hidden rows and the date + // cutoff are excluded inside every walk step so they neither become anchors + // nor advance the cursor. var completedEpisodesCTE string if q.SeriesID != "" { completedEpisodesCTE = fmt.Sprintf(`completed_episodes AS ( @@ -200,37 +317,121 @@ func buildListNextUpQuery(q NextUpQuery, limit int) (string, []interface{}) { ORDER BY e.series_id, uwp.updated_at DESC, e.season_number DESC, e.episode_number DESC )`, seriesFilter, dateCutoffFilter) } else { - completedEpisodesCTE = fmt.Sprintf(`recent_completed AS ( - SELECT uwp.media_item_id, uwp.updated_at - FROM user_watch_progress uwp - WHERE uwp.user_id = $1 - AND uwp.profile_id = $2 - AND uwp.completed = TRUE - AND NOT EXISTS ( - SELECT 1 - FROM user_history_hidden_items hhi - WHERE hhi.user_id = uwp.user_id - AND hhi.profile_id = uwp.profile_id - AND hhi.media_item_id = uwp.media_item_id - AND uwp.updated_at <= hhi.hidden_before - ) - %s - ORDER BY uwp.updated_at DESC - LIMIT %d + // Picking the series and picking which of its episodes to anchor on are + // two different orderings, so they are two different steps. + // + // The walk advances on (updated_at, media_item_id), the total order that + // guarantees forward progress. media_item_id is unique, so no two rows + // ever tie on that pair and any season/episode clause appended behind it + // is unreachable. A bulk mark-watched series — every row written with one + // timestamp — would then anchor on whichever content_id sorts highest, + // which is an arbitrary season once IDs are not scan-ordered (season 2 + // scanned before a season-1 backfill), and the rail would surface an + // already-watched episode's successor. + // + // So `pick` chooses the series and the cursor position, and the anchor + // lateral then chooses the episode by (season, episode) among that + // series' rows sharing pick.updated_at — its newest completed timestamp. + // Pinning updated_at rather than re-sorting keeps this an index probe, + // and both rows carry the same updated_at, so cursor order and the + // reported CompletedAt are unchanged by the split. + anchorLateral := `JOIN LATERAL ( + SELECT e_a.season_number, e_a.episode_number + FROM episodes e_a + JOIN user_watch_progress uwp_a + ON uwp_a.media_item_id = e_a.content_id + AND uwp_a.user_id = $1 + AND uwp_a.profile_id = $2 + AND uwp_a.completed = TRUE + AND uwp_a.updated_at = pick.updated_at + WHERE e_a.series_id = pick.series_id + AND NOT EXISTS ( + SELECT 1 + FROM user_history_hidden_items hhi + WHERE hhi.user_id = uwp_a.user_id + AND hhi.profile_id = uwp_a.profile_id + AND hhi.media_item_id = uwp_a.media_item_id + AND uwp_a.updated_at <= hhi.hidden_before + ) + ORDER BY e_a.season_number DESC, e_a.episode_number DESC + LIMIT 1 + ) anchor ON TRUE` + + completedEpisodesCTE = fmt.Sprintf(`RECURSIVE walk AS ( + ( + SELECT + pick.series_id, + anchor.season_number, + anchor.episode_number, + pick.updated_at, + pick.media_item_id, + %s AS seen, + 1 AS n + FROM ( + SELECT e.series_id, uwp.updated_at, uwp.media_item_id + FROM user_watch_progress uwp + JOIN episodes e ON e.content_id = uwp.media_item_id + WHERE uwp.user_id = $1 + AND uwp.profile_id = $2 + AND uwp.completed = TRUE + AND NOT EXISTS ( + SELECT 1 + FROM user_history_hidden_items hhi + WHERE hhi.user_id = uwp.user_id + AND hhi.profile_id = uwp.profile_id + AND hhi.media_item_id = uwp.media_item_id + AND uwp.updated_at <= hhi.hidden_before + ) + %s + %s + ORDER BY uwp.updated_at DESC, uwp.media_item_id DESC + LIMIT 1 + ) pick + %s + ) + UNION ALL + SELECT + pick.series_id, + anchor.season_number, + anchor.episode_number, + pick.updated_at, + pick.media_item_id, + w.seen || pick.series_id, + w.n + 1 + FROM walk w + JOIN LATERAL ( + SELECT e.series_id, uwp.updated_at, uwp.media_item_id + FROM user_watch_progress uwp + JOIN episodes e ON e.content_id = uwp.media_item_id + WHERE uwp.user_id = $1 + AND uwp.profile_id = $2 + AND uwp.completed = TRUE + AND (uwp.updated_at, uwp.media_item_id) < (w.updated_at, w.media_item_id) + AND NOT (e.series_id = ANY(w.seen)) + AND NOT EXISTS ( + SELECT 1 + FROM user_history_hidden_items hhi + WHERE hhi.user_id = uwp.user_id + AND hhi.profile_id = uwp.profile_id + AND hhi.media_item_id = uwp.media_item_id + AND uwp.updated_at <= hhi.hidden_before + ) + %s + ORDER BY uwp.updated_at DESC, uwp.media_item_id DESC + LIMIT 1 + ) pick ON true + %s + WHERE w.n < %d ), completed_episodes AS ( - SELECT DISTINCT ON (e.series_id) - e.series_id, - e.season_number, - e.episode_number, - uwp.updated_at - FROM recent_completed uwp - JOIN episodes e ON e.content_id = uwp.media_item_id - ORDER BY e.series_id, uwp.updated_at DESC, e.season_number DESC, e.episode_number DESC - )`, dateCutoffFilter, nextUpAnchorMaxRows) + SELECT series_id, season_number, episode_number, updated_at + FROM walk + )`, seedSeen, cursorFilter, dateCutoffFilter, anchorLateral, + dateCutoffFilter, anchorLateral, nextUpAnchorMaxSeries) } - query := fmt.Sprintf(` + if q.SeriesID != "" { + query := fmt.Sprintf(` WITH %s %s SELECT @@ -260,6 +461,64 @@ func buildListNextUpQuery(q NextUpQuery, limit int) (string, []interface{}) { ) next_ep ON true ORDER BY es.updated_at DESC LIMIT $3`, completedEpisodesCTE, inProgressExclusion, sourceTable) + return query, args + } + + resultQuery := fmt.Sprintf(` + SELECT + next_ep.content_id, + next_ep.series_id, + si.title, + next_ep.season_number, + next_ep.episode_number, + es.updated_at AS completed_at + FROM %s es + JOIN media_items si ON si.content_id = es.series_id + JOIN LATERAL ( + SELECT e2.content_id, e2.series_id, e2.season_number, e2.episode_number + FROM episodes e2 + WHERE e2.series_id = es.series_id + AND (e2.season_number, e2.episode_number) > (es.season_number, es.episode_number) + AND EXISTS (SELECT 1 FROM media_files mf WHERE mf.episode_id = e2.content_id AND mf.missing_since IS NULL) + AND NOT EXISTS ( + SELECT 1 FROM user_watch_progress uwp2 + WHERE uwp2.user_id = $1 + AND uwp2.profile_id = $2 + AND uwp2.media_item_id = e2.content_id + AND (uwp2.completed = TRUE OR uwp2.position_seconds > 0) + ) + ORDER BY e2.season_number, e2.episode_number + LIMIT 1 + ) next_ep ON true + ORDER BY es.updated_at DESC + LIMIT $3`, sourceTable) + + query := fmt.Sprintf(` + WITH %s + %s + , + frontier AS ( + SELECT w.updated_at, w.media_item_id, w.seen, w.n + FROM walk w + ORDER BY w.n DESC + LIMIT 1 + ) + SELECT + f.updated_at AS frontier_updated_at, + f.media_item_id AS frontier_media_item_id, + f.seen AS frontier_seen, + f.n AS frontier_n, + r.content_id, + r.series_id, + r.title, + r.season_number, + r.episode_number, + r.completed_at + FROM frontier f + LEFT JOIN ( + %s + ) r ON true + ORDER BY r.completed_at DESC NULLS LAST`, completedEpisodesCTE, inProgressExclusion, resultQuery) return query, args } diff --git a/internal/catalog/nextup_repo_test.go b/internal/catalog/nextup_repo_test.go index c01da4bd..16b7a596 100644 --- a/internal/catalog/nextup_repo_test.go +++ b/internal/catalog/nextup_repo_test.go @@ -1,18 +1,58 @@ package catalog import ( + "context" "fmt" + "os" "strings" "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/Silo-Server/silo-server/internal/userstore" ) +type nextUpTestStoreProvider struct{} + +func (nextUpTestStoreProvider) ForUser(context.Context, int) (userstore.UserStore, error) { + return nil, nil +} + +func (nextUpTestStoreProvider) Close() error { return nil } + +func newNextUpTestPool(t *testing.T) *pgxpool.Pool { + t.Helper() + + dsn := os.Getenv("SILO_TEST_DATABASE_URL") + if dsn == "" { + t.Skip("SILO_TEST_DATABASE_URL is not set") + } + ctx := context.Background() + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatalf("connect test database: %v", err) + } + t.Cleanup(pool.Close) + + var tableName *string + if err := pool.QueryRow(ctx, `SELECT to_regclass('public.media_items')::text`).Scan(&tableName); err != nil { + t.Fatalf("check media_items table: %v", err) + } + if tableName == nil || *tableName == "" { + t.Skip("test database has not applied base schema") + } + + return pool +} + func TestBuildListNextUpQuery_PrefersRecentCompletedOverOlderPartialProgress(t *testing.T) { t.Parallel() query, args := buildListNextUpQuery(NextUpQuery{ UserID: 7, ProfileID: "profile-1", - }, 20) + }, 20, nil) expectedFragments := []string{ "eligible_series AS (", @@ -22,7 +62,7 @@ func TestBuildListNextUpQuery_PrefersRecentCompletedOverOlderPartialProgress(t * "FROM user_history_hidden_items hhi", "uwp.updated_at <= hhi.hidden_before", "AND (uwp2.completed = TRUE OR uwp2.position_seconds > 0)", - "ORDER BY e.series_id, uwp.updated_at DESC, e.season_number DESC, e.episode_number DESC", + "ORDER BY uwp.updated_at DESC, uwp.media_item_id DESC", "LIMIT $3", } for _, fragment := range expectedFragments { @@ -43,29 +83,100 @@ func TestBuildListNextUpQuery_PrefersRecentCompletedOverOlderPartialProgress(t * func TestBuildListNextUpQuery_GlobalBoundsAnchorScan(t *testing.T) { t.Parallel() - query, _ := buildListNextUpQuery(NextUpQuery{ + q := NextUpQuery{ UserID: 7, ProfileID: "profile-1", - }, 20) - - // The global query must derive series anchors from a bounded - // recent-completions scan, not the profile's entire completed history. - if !strings.Contains(query, "recent_completed AS (") { - t.Fatalf("expected global query to scan recent completions, got:\n%s", query) } - if !strings.Contains(query, fmt.Sprintf("LIMIT %d", nextUpAnchorMaxRows)) { - t.Fatalf("expected global anchor scan bounded at %d rows, got:\n%s", nextUpAnchorMaxRows, query) + query, args := buildListNextUpQuery(q, 20, nil) + + expectedFragments := []string{ + "WITH RECURSIVE", + "(uwp.updated_at, uwp.media_item_id) < (w.updated_at, w.media_item_id)", + "NOT (e.series_id = ANY(w.seen))", + fmt.Sprintf("w.n < %d", nextUpAnchorMaxSeries), + "frontier AS (", + "LEFT JOIN (", + } + for _, fragment := range expectedFragments { + if !strings.Contains(query, fragment) { + t.Fatalf("expected global anchor walk to contain %q, got:\n%s", fragment, query) + } + } + // Seed step, recursive step, and the anchor lateral attached to each. + if got := strings.Count(query, "FROM user_history_hidden_items hhi"); got != 4 { + t.Fatalf("expected hidden-items filter in every walk step and anchor, got %d occurrences:\n%s", got, query) + } + seed := strings.SplitN(query, "UNION ALL", 2)[0] + if strings.Contains(seed, "(uwp.updated_at, uwp.media_item_id) < ($") { + t.Fatalf("first batch seed must not contain a continuation cursor, got:\n%s", seed) + } + if strings.Contains(seed, "e.series_id = ANY($") { + t.Fatalf("first batch seed must not contain a seen-series exclusion, got:\n%s", seed) + } + if len(args) != 3 { + t.Fatalf("expected first batch args only, got %v", args) + } + + cursorAt := time.Now().UTC() + cursor := &nextUpWalkCursor{ + updatedAt: cursorAt, + mediaItemID: "episode-96", + seen: []string{"series-1", "series-2"}, + } + continuedQuery, continuedArgs := buildListNextUpQuery(q, 7, cursor) + continuedSeed := strings.SplitN(continuedQuery, "UNION ALL", 2)[0] + continuedFragments := []string{ + "(uwp.updated_at, uwp.media_item_id) < ($4, $5)", + "NOT (e.series_id = ANY($6))", + "$6::text[] || pick.series_id AS seen", + "frontier AS (", + "LEFT JOIN (", + } + for _, fragment := range continuedFragments { + if !strings.Contains(continuedQuery, fragment) { + t.Fatalf("expected continued query to contain %q, got:\n%s", fragment, continuedQuery) + } + } + if !strings.Contains(continuedSeed, "(uwp.updated_at, uwp.media_item_id) < ($4, $5)") || + !strings.Contains(continuedSeed, "NOT (e.series_id = ANY($6))") { + t.Fatalf("continued batch seed must contain cursor and seen-series predicates, got:\n%s", continuedSeed) + } + if len(continuedArgs) != 6 || continuedArgs[3] != cursorAt || continuedArgs[4] != cursor.mediaItemID { + t.Fatalf("unexpected continuation args: %v", continuedArgs) + } + seenArg, ok := continuedArgs[5].([]string) + if !ok || len(seenArg) != len(cursor.seen) || seenArg[0] != cursor.seen[0] || seenArg[1] != cursor.seen[1] { + t.Fatalf("expected cursor seen array verbatim, got %#v", continuedArgs[5]) + } +} + +func TestBuildListNextUpQuery_GlobalDateCutoffAppliesToEveryWalkStep(t *testing.T) { + t.Parallel() + + cutoff := time.Now().UTC().Add(-30 * 24 * time.Hour) + query, args := buildListNextUpQuery(NextUpQuery{ + UserID: 7, + ProfileID: "profile-1", + DateCutoff: &cutoff, + }, 20, nil) + + if got := strings.Count(query, "AND uwp.updated_at >= $4"); got != 2 { + t.Fatalf("expected date cutoff in seed and recursive step, got %d occurrences:\n%s", got, query) + } + if len(args) != 4 || args[3] != cutoff { + t.Fatalf("expected one shared date-cutoff arg, got %v", args) } } func TestBuildListNextUpQuery_SeriesScopedKeepsUnboundedAnchor(t *testing.T) { t.Parallel() - query, args := buildListNextUpQuery(NextUpQuery{ + q := NextUpQuery{ UserID: 7, ProfileID: "profile-1", SeriesID: "series-42", - }, 20) + } + query, args := buildListNextUpQuery(q, 20, nil) // The show-detail tile must anchor on the series' last completed episode // no matter how long ago it was watched: the recency bound would make a @@ -73,12 +184,461 @@ func TestBuildListNextUpQuery_SeriesScopedKeepsUnboundedAnchor(t *testing.T) { if strings.Contains(query, "recent_completed AS (") { t.Fatalf("series-scoped query must not bound the anchor scan, got:\n%s", query) } + if strings.Contains(query, "WITH RECURSIVE") { + t.Fatalf("series-scoped query must keep its non-recursive SQL shape, got:\n%s", query) + } if !strings.Contains(query, "AND e.series_id = $4") { t.Fatalf("series-scoped query must filter by series_id, got:\n%s", query) } if len(args) != 4 { t.Fatalf("expected 4 args with SeriesID, got %d (%v)", len(args), args) } + + queryWithCursor, argsWithCursor := buildListNextUpQuery(q, 20, &nextUpWalkCursor{ + updatedAt: time.Now().UTC(), + mediaItemID: "ignored-episode", + seen: []string{"ignored-series"}, + }) + if queryWithCursor != query || fmt.Sprint(argsWithCursor) != fmt.Sprint(args) { + t.Fatalf("series-scoped SQL and args must be byte-identical when a walk cursor is supplied") + } +} + +func TestBuildListNextUpQuery_GlobalAnchorOrdersByEpisodeNotMediaItemID(t *testing.T) { + t.Parallel() + + // media_item_id is unique, so a season/episode tie-break appended behind it + // can never be reached. Choosing the series and choosing which of its + // episodes anchors the rail have to be separate orderings. + query, _ := buildListNextUpQuery(NextUpQuery{ + UserID: 7, + ProfileID: "profile-1", + }, 20, nil) + + if strings.Contains(query, `ORDER BY uwp.updated_at DESC, uwp.media_item_id DESC, + e.season_number DESC, e.episode_number DESC`) { + t.Fatalf("season/episode must not sit behind the unique media_item_id tie-break, got:\n%s", query) + } + if got := strings.Count(query, "ORDER BY e_a.season_number DESC, e_a.episode_number DESC"); got != 2 { + t.Fatalf("expected episode-ordered anchor in seed and recursive step, got %d occurrences:\n%s", got, query) + } + if got := strings.Count(query, "AND uwp_a.updated_at = pick.updated_at"); got != 2 { + t.Fatalf("expected anchor pinned to the series' newest completed timestamp, got %d:\n%s", got, query) + } +} + +func TestNextUpRepository_BulkMarkWatchedAnchorsOnHighestEpisode(t *testing.T) { + // A bulk mark-watched series writes every row with one timestamp. The anchor + // must then be the highest (season, episode), not the highest content_id: + // with production-shape IDs where season 2 was scanned before a season-1 + // backfill, season 1 sorts higher and the rail would surface s01e04 — an + // episode after one the user already watched — instead of s02e04. + pool := newNextUpTestPool(t) + ctx := context.Background() + prefix := fmt.Sprintf("nextup-order-%d", time.Now().UnixNano()) + seriesID := prefix + "-series" + + userID, profileID, folderID := seedNextUpTestOwner(t, ctx, pool, prefix) + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM media_items WHERE content_id = $1`, seriesID) + }) + seedNextUpSeries(t, ctx, pool, seriesID, prefix+" Ordering") + + // Season 2 scanned first (lower IDs), season 1 backfilled later (higher IDs). + const wantContentID = "1" + "00000000000000004" // s02e04 + const trapContentID = "9" + "00000000000000004" // s01e04 + episodeIDs := []string{ + "100000000000000001", "100000000000000002", "100000000000000003", wantContentID, + "900000000000000001", "900000000000000002", "900000000000000003", trapContentID, + } + seasons := []int{2, 2, 2, 2, 1, 1, 1, 1} + numbers := []int{1, 2, 3, 4, 1, 2, 3, 4} + for i, episodeID := range episodeIDs { + if _, err := pool.Exec(ctx, ` + INSERT INTO episodes (content_id, series_id, season_number, episode_number, title) + VALUES ($1, $2, $3, $4, 'Ordering Episode') + `, episodeID, seriesID, seasons[i], numbers[i]); err != nil { + t.Fatalf("seed episode %s: %v", episodeID, err) + } + } + seedNextUpFiles(t, ctx, pool, folderID, episodeIDs) + + // Every completed row shares one timestamp, as a bulk mark-watched write does. + bulkAt := time.Now().UTC().Truncate(time.Microsecond) + if _, err := pool.Exec(ctx, ` + INSERT INTO user_watch_progress ( + user_id, profile_id, media_item_id, position_seconds, + duration_seconds, completed, updated_at + ) + SELECT $1, $2, unnest($3::text[]), 0, 1800, TRUE, $4 + `, userID, profileID, []string{ + "100000000000000001", "100000000000000002", "100000000000000003", + "900000000000000001", "900000000000000002", "900000000000000003", + }, bulkAt); err != nil { + t.Fatalf("seed bulk progress: %v", err) + } + + repo := NewNextUpRepository(pool, nextUpTestStoreProvider{}) + results, err := repo.ListNextUp(ctx, NextUpQuery{ + UserID: userID, + ProfileID: profileID, + Limit: 20, + EnableResumable: false, + }) + if err != nil { + t.Fatalf("ListNextUp: %v", err) + } + assertNextUpContentIDs(t, results, wantContentID) +} + +func TestNextUpRepository_GlobalAnchorWalkSurvivesFloodedSeries(t *testing.T) { + pool := newNextUpTestPool(t) + ctx := context.Background() + suffix := time.Now().UnixNano() + prefix := fmt.Sprintf("nextup-flood-%d", suffix) + seriesA := prefix + "-series-a" + seriesB := prefix + "-series-b" + seriesC := prefix + "-series-c" + seriesIDs := []string{seriesA, seriesB, seriesC} + episodeAPrefix := prefix + "-episode-a-" + episodeB1 := prefix + "-episode-b-1" + episodeB2 := prefix + "-episode-b-2" + episodeC1 := prefix + "-episode-c-1" + episodeC2 := prefix + "-episode-c-2" + + userID, profileID, folderID := seedNextUpTestOwner(t, ctx, pool, prefix) + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM media_items WHERE content_id = ANY($1)`, seriesIDs) + }) + + seedNextUpSeries(t, ctx, pool, seriesA, prefix+" Series A") + seedNextUpSeries(t, ctx, pool, seriesB, prefix+" Series B") + seedNextUpSeries(t, ctx, pool, seriesC, prefix+" Series C") + if _, err := pool.Exec(ctx, ` + INSERT INTO episodes (content_id, series_id, season_number, episode_number, title) + SELECT $1 || gs::text, $2, 1, gs, 'Flood Episode ' || gs::text + FROM generate_series(1, 600) gs + `, episodeAPrefix, seriesA); err != nil { + t.Fatalf("seed flooded episodes: %v", err) + } + seedNextUpEpisodes(t, ctx, pool, + []string{episodeB1, episodeB2, episodeC1, episodeC2}, + []string{seriesB, seriesB, seriesC, seriesC}, + []int{1, 2, 1, 2}, + ) + seedNextUpFiles(t, ctx, pool, folderID, []string{episodeB2, episodeC2}) + + floodedAt := time.Now().UTC().Truncate(time.Microsecond) + if _, err := pool.Exec(ctx, ` + INSERT INTO user_watch_progress ( + user_id, profile_id, media_item_id, position_seconds, + duration_seconds, completed, updated_at + ) + SELECT $1, $2, $3 || gs::text, 0, 1800, TRUE, $4 + FROM generate_series(1, 600) gs + `, userID, profileID, episodeAPrefix, floodedAt); err != nil { + t.Fatalf("seed flooded progress: %v", err) + } + if _, err := pool.Exec(ctx, ` + INSERT INTO user_watch_progress ( + user_id, profile_id, media_item_id, position_seconds, + duration_seconds, completed, updated_at + ) + VALUES ($1, $2, $3, 0, 1800, TRUE, $5), + ($1, $2, $4, 0, 1800, TRUE, $6) + `, userID, profileID, episodeB1, episodeC1, floodedAt.Add(-time.Hour), floodedAt.Add(-2*time.Hour)); err != nil { + t.Fatalf("seed older series progress: %v", err) + } + + repo := NewNextUpRepository(pool, nextUpTestStoreProvider{}) + results, err := repo.ListNextUp(ctx, NextUpQuery{ + UserID: userID, + ProfileID: profileID, + Limit: 20, + EnableResumable: false, + }) + if err != nil { + t.Fatalf("ListNextUp: %v", err) + } + assertNextUpContentIDs(t, results, episodeB2, episodeC2) +} + +func TestNextUpRepository_GlobalAnchorWalkKeepsSameTimestampSeries(t *testing.T) { + pool := newNextUpTestPool(t) + ctx := context.Background() + suffix := time.Now().UnixNano() + prefix := fmt.Sprintf("nextup-cursor-%d", suffix) + seriesD := prefix + "-series-d" + seriesE := prefix + "-series-e" + seriesIDs := []string{seriesD, seriesE} + episodeD1 := prefix + "-episode-d-1" + episodeD2 := prefix + "-episode-d-2" + episodeE1 := prefix + "-episode-e-1" + episodeE2 := prefix + "-episode-e-2" + + userID, profileID, folderID := seedNextUpTestOwner(t, ctx, pool, prefix) + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM media_items WHERE content_id = ANY($1)`, seriesIDs) + }) + + seedNextUpSeries(t, ctx, pool, seriesD, prefix+" Series D") + seedNextUpSeries(t, ctx, pool, seriesE, prefix+" Series E") + seedNextUpEpisodes(t, ctx, pool, + []string{episodeD1, episodeD2, episodeE1, episodeE2}, + []string{seriesD, seriesD, seriesE, seriesE}, + []int{1, 2, 1, 2}, + ) + seedNextUpFiles(t, ctx, pool, folderID, []string{episodeD2, episodeE2}) + + completedAt := time.Now().UTC().Truncate(time.Microsecond) + if _, err := pool.Exec(ctx, ` + INSERT INTO user_watch_progress ( + user_id, profile_id, media_item_id, position_seconds, + duration_seconds, completed, updated_at + ) + VALUES ($1, $2, $3, 0, 1800, TRUE, $5), + ($1, $2, $4, 0, 1800, TRUE, $5) + `, userID, profileID, episodeD1, episodeE1, completedAt); err != nil { + t.Fatalf("seed same-timestamp progress: %v", err) + } + + repo := NewNextUpRepository(pool, nextUpTestStoreProvider{}) + results, err := repo.ListNextUp(ctx, NextUpQuery{ + UserID: userID, + ProfileID: profileID, + Limit: 20, + EnableResumable: false, + }) + if err != nil { + t.Fatalf("ListNextUp: %v", err) + } + assertNextUpContentIDs(t, results, episodeD2, episodeE2) +} + +func TestNextUpRepository_GlobalAnchorWalkContinuesPastIneligibleSeries(t *testing.T) { + pool := newNextUpTestPool(t) + ctx := context.Background() + prefix := fmt.Sprintf("nextup-ineligible-%d", time.Now().UnixNano()) + + const ineligibleCount = 100 + seriesIDs := make([]string, 0, ineligibleCount+2) + episodeIDs := make([]string, 0, ineligibleCount+4) + episodeSeriesIDs := make([]string, 0, ineligibleCount+4) + episodeNumbers := make([]int, 0, ineligibleCount+4) + completedEpisodeIDs := make([]string, 0, ineligibleCount+2) + completedTimes := make([]time.Time, 0, ineligibleCount+2) + + userID, profileID, folderID := seedNextUpTestOwner(t, ctx, pool, prefix) + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM media_items WHERE content_id = ANY($1)`, seriesIDs) + }) + + newest := time.Now().UTC().Truncate(time.Microsecond) + for i := 0; i < ineligibleCount; i++ { + seriesID := fmt.Sprintf("%s-series-%03d", prefix, i) + episodeID := seriesID + "-episode-1" + seriesIDs = append(seriesIDs, seriesID) + episodeIDs = append(episodeIDs, episodeID) + episodeSeriesIDs = append(episodeSeriesIDs, seriesID) + episodeNumbers = append(episodeNumbers, 1) + completedEpisodeIDs = append(completedEpisodeIDs, episodeID) + completedTimes = append(completedTimes, newest.Add(-time.Duration(i)*time.Minute)) + seedNextUpSeries(t, ctx, pool, seriesID, fmt.Sprintf("%s Ineligible %03d", prefix, i)) + } + + olderSeriesA := prefix + "-older-series-a" + olderSeriesB := prefix + "-older-series-b" + olderA1 := olderSeriesA + "-episode-1" + olderA2 := olderSeriesA + "-episode-2" + olderB1 := olderSeriesB + "-episode-1" + olderB2 := olderSeriesB + "-episode-2" + seriesIDs = append(seriesIDs, olderSeriesA, olderSeriesB) + seedNextUpSeries(t, ctx, pool, olderSeriesA, prefix+" Older A") + seedNextUpSeries(t, ctx, pool, olderSeriesB, prefix+" Older B") + episodeIDs = append(episodeIDs, olderA1, olderA2, olderB1, olderB2) + episodeSeriesIDs = append(episodeSeriesIDs, olderSeriesA, olderSeriesA, olderSeriesB, olderSeriesB) + episodeNumbers = append(episodeNumbers, 1, 2, 1, 2) + completedEpisodeIDs = append(completedEpisodeIDs, olderA1, olderB1) + completedTimes = append(completedTimes, + newest.Add(-(ineligibleCount+1)*time.Minute), + newest.Add(-(ineligibleCount+2)*time.Minute), + ) + + seedNextUpEpisodes(t, ctx, pool, episodeIDs, episodeSeriesIDs, episodeNumbers) + seedNextUpFiles(t, ctx, pool, folderID, []string{olderA2, olderB2}) + if _, err := pool.Exec(ctx, ` + INSERT INTO user_watch_progress ( + user_id, profile_id, media_item_id, position_seconds, + duration_seconds, completed, updated_at + ) + SELECT $1, $2, episode_id, 0, 1800, TRUE, completed_at + FROM unnest($3::text[], $4::timestamptz[]) AS fixture(episode_id, completed_at) + `, userID, profileID, completedEpisodeIDs, completedTimes); err != nil { + t.Fatalf("seed ineligible-flood progress: %v", err) + } + + repo := NewNextUpRepository(pool, nextUpTestStoreProvider{}) + results, err := repo.ListNextUp(ctx, NextUpQuery{ + UserID: userID, + ProfileID: profileID, + Limit: 20, + EnableResumable: false, + }) + if err != nil { + t.Fatalf("ListNextUp: %v", err) + } + assertNextUpContentIDs(t, results, olderA2, olderB2) +} + +func TestNextUpRepository_GlobalAnchorWalkStopsWhenHistoryExhausted(t *testing.T) { + pool := newNextUpTestPool(t) + ctx := context.Background() + prefix := fmt.Sprintf("nextup-exhausted-%d", time.Now().UnixNano()) + seriesIDs := []string{ + prefix + "-series-a", + prefix + "-series-b", + prefix + "-series-c", + } + episodeIDs := []string{ + seriesIDs[0] + "-episode-1", + seriesIDs[1] + "-episode-1", + seriesIDs[2] + "-episode-1", + } + + userID, profileID, _ := seedNextUpTestOwner(t, ctx, pool, prefix) + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM media_items WHERE content_id = ANY($1)`, seriesIDs) + }) + for i, seriesID := range seriesIDs { + seedNextUpSeries(t, ctx, pool, seriesID, fmt.Sprintf("%s Finished %d", prefix, i)) + } + seedNextUpEpisodes(t, ctx, pool, episodeIDs, seriesIDs, []int{1, 1, 1}) + + completedAt := time.Now().UTC().Truncate(time.Microsecond) + if _, err := pool.Exec(ctx, ` + INSERT INTO user_watch_progress ( + user_id, profile_id, media_item_id, position_seconds, + duration_seconds, completed, updated_at + ) + SELECT $1, $2, episode_id, 0, 1800, TRUE, $4::timestamptz - (ordinality * INTERVAL '1 minute') + FROM unnest($3::text[]) WITH ORDINALITY AS fixture(episode_id, ordinality) + `, userID, profileID, episodeIDs, completedAt); err != nil { + t.Fatalf("seed exhausted progress: %v", err) + } + + repo := NewNextUpRepository(pool, nextUpTestStoreProvider{}) + results, err := repo.ListNextUp(ctx, NextUpQuery{ + UserID: userID, + ProfileID: profileID, + Limit: 20, + EnableResumable: false, + }) + if err != nil { + t.Fatalf("ListNextUp: %v", err) + } + if len(results) != 0 { + t.Fatalf("ListNextUp returned %d rows after exhausting finished history: %+v", len(results), results) + } +} + +func seedNextUpTestOwner(t *testing.T, ctx context.Context, pool *pgxpool.Pool, prefix string) (int, string, int) { + t.Helper() + + var folderID int + if err := pool.QueryRow(ctx, ` + INSERT INTO media_folders (type, name, enabled) + VALUES ('series', $1, TRUE) + RETURNING id + `, prefix+" Library").Scan(&folderID); err != nil { + t.Fatalf("seed media folder: %v", err) + } + + var userID int + if err := pool.QueryRow(ctx, ` + INSERT INTO users (username, role) + VALUES ($1, 'user') + RETURNING id + `, prefix+"-user").Scan(&userID); err != nil { + _, _ = pool.Exec(ctx, `DELETE FROM media_folders WHERE id = $1`, folderID) + t.Fatalf("seed user: %v", err) + } + + profileID := fmt.Sprintf("00000000-0000-4000-8000-%012d", time.Now().UnixNano()%1_000_000_000_000) + if _, err := pool.Exec(ctx, ` + INSERT INTO user_profiles (id, user_id, name) + VALUES ($1, $2, 'Next Up Regression') + `, profileID, userID); err != nil { + _, _ = pool.Exec(ctx, `DELETE FROM users WHERE id = $1`, userID) + _, _ = pool.Exec(ctx, `DELETE FROM media_folders WHERE id = $1`, folderID) + t.Fatalf("seed profile: %v", err) + } + + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM media_files WHERE media_folder_id = $1`, folderID) + _, _ = pool.Exec(ctx, `DELETE FROM users WHERE id = $1`, userID) + _, _ = pool.Exec(ctx, `DELETE FROM media_folders WHERE id = $1`, folderID) + }) + + return userID, profileID, folderID +} + +func seedNextUpSeries(t *testing.T, ctx context.Context, pool *pgxpool.Pool, seriesID, title string) { + t.Helper() + + if _, err := pool.Exec(ctx, ` + INSERT INTO media_items (content_id, type, title, status, genres) + VALUES ($1, 'series', $2, 'matched', '{}'::text[]) + `, seriesID, title); err != nil { + t.Fatalf("seed series %s: %v", seriesID, err) + } +} + +func seedNextUpEpisodes( + t *testing.T, + ctx context.Context, + pool *pgxpool.Pool, + episodeIDs []string, + seriesIDs []string, + episodeNumbers []int, +) { + t.Helper() + + if _, err := pool.Exec(ctx, ` + INSERT INTO episodes (content_id, series_id, season_number, episode_number, title) + SELECT episode_id, series_id, 1, episode_number, 'Next Up Episode ' || episode_number::text + FROM unnest($1::text[], $2::text[], $3::int[]) AS fixture(episode_id, series_id, episode_number) + `, episodeIDs, seriesIDs, episodeNumbers); err != nil { + t.Fatalf("seed episodes: %v", err) + } +} + +func seedNextUpFiles(t *testing.T, ctx context.Context, pool *pgxpool.Pool, folderID int, episodeIDs []string) { + t.Helper() + + if _, err := pool.Exec(ctx, ` + INSERT INTO media_files (episode_id, media_folder_id, file_path, duration) + SELECT episode_id, $2, '/nextup-regression/' || episode_id || '.mkv', 1800 + FROM unnest($1::text[]) AS fixture(episode_id) + `, episodeIDs, folderID); err != nil { + t.Fatalf("seed media files: %v", err) + } +} + +func assertNextUpContentIDs(t *testing.T, results []NextUpResult, want ...string) { + t.Helper() + + if len(results) != len(want) { + t.Fatalf("ListNextUp returned %d rows, want %d: %+v", len(results), len(want), results) + } + got := make(map[string]bool, len(results)) + for _, result := range results { + got[result.ContentID] = true + } + for _, contentID := range want { + if !got[contentID] { + t.Errorf("ListNextUp missing %s: %+v", contentID, results) + } + } } func TestBuildListNextUpQuery_EnableResumableSkipsSeriesSuppressionCTE(t *testing.T) { @@ -88,7 +648,7 @@ func TestBuildListNextUpQuery_EnableResumableSkipsSeriesSuppressionCTE(t *testin UserID: 7, ProfileID: "profile-1", EnableResumable: true, - }, 20) + }, 20, nil) if strings.Contains(query, "eligible_series AS (") { t.Fatalf("expected resumable query to skip eligible_series suppression CTE, got:\n%s", query) diff --git a/migrations/sql/20260811145848_add_nextup_anchor_walk_cursor_index.sql b/migrations/sql/20260811145848_add_nextup_anchor_walk_cursor_index.sql new file mode 100644 index 00000000..d6c6b3c0 --- /dev/null +++ b/migrations/sql/20260811145848_add_nextup_anchor_walk_cursor_index.sql @@ -0,0 +1,25 @@ +-- +goose NO TRANSACTION +-- +goose Up +-- The global Next Up anchor walk advances on a compound +-- (updated_at, media_item_id) cursor: each step asks for the newest completed +-- row strictly before the previous anchor's pair. idx_uwp_profile_completed is +-- (user_id, profile_id, updated_at DESC) only, so the media_item_id half of +-- that comparison is a filter rather than part of the seek, and every step +-- re-reads the rows tied on updated_at — exactly the shape a bulk +-- mark-watched profile has, where hundreds of rows share one timestamp. +-- +-- Adding media_item_id to the index makes each step a single ordered seek. +-- Measured by dropping and recreating this index around the shipped query on a +-- synthetic profile with a 40k-row bulk-marked series: 341ms -> 224ms (~34%), +-- same plan shape otherwise. The partial predicate matches the existing +-- completed-only index so the two stay interchangeable for planning; this one +-- supersedes it for the walk. +-- +-- CONCURRENTLY avoids locking user_watch_progress writes during the build; it +-- cannot run inside a transaction, hence the NO TRANSACTION annotation above. +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_uwp_profile_completed_cursor +ON public.user_watch_progress USING btree (user_id, profile_id, updated_at DESC, media_item_id DESC) +WHERE completed = TRUE; + +-- +goose Down +DROP INDEX CONCURRENTLY IF EXISTS public.idx_uwp_profile_completed_cursor;