From e77d9e933ca8d5814719c58c55af9d39818dd2cb Mon Sep 17 00:00:00 2001 From: Quick <31828688+Quick104@users.noreply.github.com> Date: Tue, 11 Aug 2026 12:08:45 -0400 Subject: [PATCH] fix(catalog): bound Next Up anchors by distinct series, not rows (#593) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(catalog): bound Next Up anchors by distinct series, not rows The global Next Up query capped its anchor scan at the 500 most recently completed rows (nextUpAnchorMaxRows, #350). Bulk mark-watched writes hundreds of completed rows with the newest timestamps, so one series could flood the whole window and evict every other series from the rail - observed in production wiping a user's Next Up row entirely. Replace the row-capped CTE with a recursive skip-scan over idx_uwp_profile_completed that emits the newest completed row of each not-yet-seen series and stops after nextUpAnchorMaxSeries (96) distinct series. A compound (updated_at, media_item_id) cursor keeps the walk total when bulk writes share one timestamp across series. The series-scoped branch (show-detail tile) keeps its unbounded shape. DB-backed regression tests cover the flood shape (600 same-timestamp rows of one series must not evict others) and same-timestamp anchors across series; the flood test fails against the previous query. Fixes #592 Co-Authored-By: Claude Fable 5 * fix(catalog): continue Next Up anchor walk until enough eligible series found The 96-series walk budget counted visited series, but eligibility (next episode exists with a present file, no newer partial progress) is only decided downstream, so 96 consecutive ineligible anchors - all caught up, unavailable, or blocked - still emptied the rail and silently capped /Shows/NextUp pagination at whatever survived one walk. Run the walk in batches: each batch keeps the 96-series budget, reports its frontier (compound cursor position, seen-series array, rows walked), and ListNextUp resumes the next batch below that frontier until the requested limit is filled, history is exhausted (frontier short of the budget), or a 10-batch runaway guard trips (logged, never silent). Batches emit anchors in strictly descending (updated_at, media_item_id) order, so appending preserves rail order. Series-scoped queries keep their single-shot shape. DB-backed regressions: 100 fully-watched series must not hide older series with eligible next episodes (fails on the single-batch code), and an all-caught-up profile returns empty without spinning to the batch cap. Co-Authored-By: Claude Fable 5 * fix(catalog): anchor Next Up on the highest episode, not the highest ID Review follow-up to the anchor walk. Two issues remained after the batching fix. The walk ordered anchors by (updated_at, media_item_id) and then by (season, episode). media_item_id is unique, so no two rows ever tie on that pair and the season/episode clause was unreachable. For a bulk mark-watched series — every row written with one timestamp — the anchor was therefore whichever content_id sorted highest. With production-shape 18-digit IDs where season 2 was scanned before a season-1 backfill, season 1 sorts higher, so the rail surfaced s01e04: an episode following one the user had already watched. Choosing the series and choosing its anchor episode are now separate steps. The walk still advances on the compound cursor, which is the total order it needs; a lateral then picks the highest (season, episode) among that series' rows sharing pick.updated_at, matching what the series-scoped branch already did. Pinning updated_at instead of re-sorting keeps it an index probe, and both rows carry the same updated_at, so cursor order and the reported CompletedAt are unchanged. Also adds the index the compound cursor needs. idx_uwp_profile_completed stops at updated_at, so the media_item_id half of the seek was a filter and every step re-read the rows tied on one timestamp — the exact shape a bulk mark-watched profile has. Measured by dropping and recreating the index around the shipped query on a synthetic 40k-row profile: 341ms -> 224ms, same plan shape otherwise. Verified against Postgres 18.3. The new DB regression seeds the scanned-out-of-order ID shape and fails on the previous ordering (returns s01e04, wants s02e04); the SQL-shape test pins that media_item_id can no longer gate the episode choice. Part of #592 Co-Authored-By: Claude Opus 5 (1M context) --------- Co-authored-by: Claude Fable 5 --- internal/catalog/nextup_repo.go | 391 ++++++++++-- internal/catalog/nextup_repo_test.go | 588 +++++++++++++++++- ...48_add_nextup_anchor_walk_cursor_index.sql | 25 + 3 files changed, 924 insertions(+), 80 deletions(-) create mode 100644 migrations/sql/20260811145848_add_nextup_anchor_walk_cursor_index.sql 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;