From a2ef26bece73fb54dd09cc4b43219f60bd5576a9 Mon Sep 17 00:00:00 2001 From: CoffeeKnyte <67730400+CoffeeKnyte@users.noreply.github.com> Date: Thu, 9 Jul 2026 21:02:45 +0800 Subject: [PATCH] perf: root-cause fixes for endpoints still slow after #292 (NextUp, series badges, resume tail, subtitle fonts) (#350) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * docs(plans): root-cause analysis for endpoints still slow after PR #292 Five endpoint groups stayed slow after the home/Continue Watching/Latest latency work shipped: Resume (110s p95), NextUp (17s p95), Latest (17s), /Items, and the home sections routes. The caps and caches from PR #292 are live in the deployed binary; they bounded how many rows the loops touch but not what each underlying query costs. Documents the four confirmed root causes (4.3M stale completed-with-position progress rows + missing resume index, unbounded next-up anchor scan, per-episode series rollup fanout, two index-starved history/scanner paths) with live EXPLAIN ANALYZE measurements and the fix plan implemented by the follow-up commits. AI-use disclosure: analysis and doc produced with AI (Claude) assistance. * perf(catalog): bound the global next-up anchor scan to recent completions The completed_episodes CTE in buildListNextUpQuery derived per-series anchors from the profile's ENTIRE completed history — DISTINCT ON over 233k rows joined to episodes for the worst bulk-import profile, then a per-series LATERAL that scans every episode of a fully-watched series before yielding nothing. 648 slow executions in a 19h window, 44.7s worst; this drove /Shows/NextUp (17.1s p95) and the next-up injection on the native home sections aggregate. Global queries now derive anchors from the profile's nextUpAnchorMaxRows (500) most recent completed rows — an ordered index walk on idx_uwp_profile_completed, with the hidden-items exclusion and date cutoff applied inside the bounded scan so hidden/old rows never consume the anchor budget. A next-up rail surfaces ~24 series; the 500 most recent completions cover every series that can realistically rank on it. Series-scoped calls (the show-detail tile) keep the unbounded shape: they must anchor on the series' last completed episode no matter how long ago it was watched, and are naturally bounded by one series. Measured on the live worst-case profile with the exact generated SQL: 44.7s worst / ~2.6s avg before; 10ms after (together with the one-time stale-resume-point data repair applied directly to the deployment DB — see docs/superpowers/plans/2026-07-06-slow-endpoint-root-causes.md). AI-use disclosure: implemented with AI (Claude) assistance. * perf(jellycompat,userstore): aggregate series watch-state rollup in SQL The series Played/UnplayedItemCount badge on list rails (per-library Latest, library browse, search results) and series detail pages was computed by materializing EVERY episode of every series on the page (episodeRepo.ListBySeriesIDs) and then batching per-episode progress+history lookups in 500-id chunks. A 50-series page of an episode-heavy library (Sports) expanded to 32,467 episode rows and ~65 sequential queries — measured 17-18s per /Items/Latest request, and PR #292's cached Latest fast path pays it on every response for series libraries. The same fanout made /Items?searchTerm=... slow whenever the result set was mostly series (Meilisearch itself answers in milliseconds). New optional store capability userstore.SeriesEpisodeRollupStore, implemented by PostgresUserStore as one GROUP BY e.series_id aggregate with semantics identical to the chunked path (episode availability via episode_libraries, hidden-items visibility on progress rows, completed-history fold, in-progress = not watched with position > 0 — verified value-for-value against the old semantics on a real 1,586-episode series). enrichSeriesListUserData and enrichDetailUserData use it when present; SQLite-backed stores and rollup query failures keep the existing chunked path as fallback. catalog.SeasonUserDataFromCounts pins the counts-to-DTO mapping to EpisodeRollupUserData. Measured on the live worst-case profile against the real 50-series Sports Latest page: ~17s of chunked round-trips before, 119ms in one query after. Part of docs/superpowers/plans/2026-07-06-slow-endpoint-root-causes.md. AI-use disclosure: implemented with AI (Claude) assistance. * perf(catalog): bound superseded-episode completed walk to recent history The Resume / Continue Watching superseded-episode filter loaded a profile's *entire* completed history into memory on every request that contained an in-progress episode: CompletedProgressSnapshots paged user_watch_progress WHERE completed=TRUE with no upper bound. The 2026-07-06 slow-query comparison showed this surviving as a 60-116s Resume tail even after the in-progress index landed live, because the 4.3M zeroed Plex-import rows are still completed=TRUE and were re-walked every load. A completed episode can only supersede an in-progress one it was finished more recently than (the query gates on done_progress.updated_at > ip_progress.updated_at), so only completed rows newer than the oldest in-progress entry can matter. Compute that cutoff in SupersededEpisodeProgressIDs and pass it to CompletedProgressSnapshots, which — since the completed listing is ordered updated_at DESC — stops paging as soon as it crosses the cutoff. Import-heavy profiles whose back-catalogue predates their current in-progress items now stop on the first page instead of paging hundreds of thousands of irrelevant rows. Correctness is unchanged: no relevant superseding row is excluded. * perf(catalog): hard-cap superseded-episode completed walk at 5 pages The updated_at cutoff added in the previous commit bounds the completed walk on the relevance axis, but a very old in-progress entry sitting behind a large volume of newer completions could still page deep. Add a 5-page (2,500-row) hard backstop on top of the cutoff: normal profiles still stop on page one via the cutoff, and only the adversarial tail hits the cap. When it engages the tail of the completed set goes unscanned, so a superseded episode could momentarily survive on Continue Watching — we log a warning when that happens (with profile_id + rows scanned) rather than mis-filter silently, and it self-corrects once the stale in-progress entry ages out of the scanned window. * perf(playback): extract subtitle fonts in a single ffmpeg pass Embedded ASS/SSA font extraction spawned one ffmpeg process per font attachment, each re-opening the (usually CephFS-backed) media file. Anime releases carry 15-47 fonts, so the per-spawn file-open cost dominated and pushed GET /api/v1/stream/{sid}/subtitles/{track}/fonts to a 17-60 s plateau (p95 ~33 s in the live logs). Collapse the N spawns into one ffmpeg invocation that dumps every attachment to a temp dir (-dump_attachment:idx path ... -i file -map 0:t? -c copy), then read the files back. The file is opened once instead of N times, taking p95 from ~30 s to ~1-2 s with no change to output. Safety is preserved. The 32-attachment / 32 MiB caps still apply: attachment size is stat'd before read so an over-limit font never enters memory, and a watchdog polls the dump dir and kills ffmpeg if its on-disk output crosses the cap -- restoring the hard bound the old pipe-per-attachment reader enforced by killing at maxBytes+1, so a container with oversized "font" attachments can't fill the disk. Part of the slow-endpoint follow-up; see slow-query-analysis/subtitle-fonts-extraction-findings.md. * fix(review): report enforced font-byte cap; correct doc subtitle scope Address PR #350 review: - dumpFontAttachments reported the maxSubtitleFontBytes package constant in both over-limit errors instead of the maxBytes argument the caller passed, so the message misstated the enforced bound whenever a different cap was in effect (as the tests use). Interpolate maxBytes in both messages. - The root-cause plan claimed subtitle extraction was 'out of scope' while the branch actually optimizes /subtitles/{track}/fonts. Scope the out-of-scope note to subtitle *track* conversion and record the fonts single-pass work as deliverable 5. --- .../2026-07-06-slow-endpoint-root-causes.md | 219 ++++++++++++++++++ .../catalog/continue_watching_progress.go | 62 ++++- .../continue_watching_progress_test.go | 60 ++++- internal/catalog/nextup_repo.go | 62 ++++- internal/catalog/nextup_repo_test.go | 42 ++++ internal/catalog/user_data_rollup.go | 15 ++ internal/jellycompat/content_direct.go | 36 ++- internal/jellycompat/content_direct_test.go | 92 ++++++++ internal/playback/subtitle_fonts.go | 192 ++++++++++----- internal/playback/subtitle_fonts_test.go | 140 +++++++++-- internal/userstore/pgstore/progress.go | 80 +++++++ internal/userstore/progress_helpers.go | 23 ++ 12 files changed, 935 insertions(+), 88 deletions(-) create mode 100644 docs/superpowers/plans/2026-07-06-slow-endpoint-root-causes.md diff --git a/docs/superpowers/plans/2026-07-06-slow-endpoint-root-causes.md b/docs/superpowers/plans/2026-07-06-slow-endpoint-root-causes.md new file mode 100644 index 00000000..215dfb06 --- /dev/null +++ b/docs/superpowers/plans/2026-07-06-slow-endpoint-root-causes.md @@ -0,0 +1,219 @@ +# 2026-07-06 — Why the slow endpoints stayed slow after PR #292, and the fix plan + +Commands assume the repository root is the cwd. + +## Context + +PR #292 (commit `5606beed`) shipped the home/Continue Watching/Latest latency work: +scan caps in the jellycompat resume path, the shared resolved-list cache, the +per-library Latest fast path, batched presign, and the plugin-installation cache. +The image built from it has been serving production for a full day, and the +deployment's request logs still show the same endpoint groups breaching 1s +(analysis window: 19h post-deploy, `log_min_duration_statement = 500ms` on the +Postgres side): + +| Endpoint group | slow (≥1s) calls | p95 | worst | +|---|---|---|---| +| `/Shows/NextUp` | 301 | 17.1s | 44.8s | +| `/UserItems/Resume` + `Users/{userId}/Items/Resume` | 121 | 34.8s / 110.7s | 125.3s | +| `/Items/Latest` + `Users/{id}/Items/Latest` | 117 | 17.3s / 8.3s | 18.8s | +| `/Items` (incl. search-as-you-type) | 83 | 6.3s | 9.6s | +| native home sections + section items (`api/v1/home` routes) | 50 | 6.3s / 42.0s | 42.8s | + +The deployed binary provably contains the PR #292 symbols, so the caps and caches +are live. They bounded *how many rows the loops touch* — they did not touch *what +each underlying query costs*. The Postgres slow-statement log for the same window +shows where the time actually goes: + +| Statement | slow execs | total time | worst | +|---|---|---|---| +| `ListProgress(status="in_progress")` page query | 5,263 | 13,522s | 16.8s | +| Next-up `WITH completed_episodes …` CTE | 648 | 1,717s | 44.7s | +| `RemoveHistoryItems` history DELETE | 81 | 118s | 10.9s | +| Scanner `media_files` subtree lookup | 27 | 74s | 13.6s | +| `ListCompletedHistoryItems` (chunked rollup) | 65 | 64s | 1.6s | + +## Root cause 1 — 4.32M stale "completed but resumable" progress rows + no index for the resume ordering + +**Human-readable.** For accounts that bulk-imported their Plex watch history, every +imported "watched" row was stored as *finished AND parked at the very end of the +video*. The server's definition of "something you can resume" is "position greater +than zero" — deliberately, so a rewatch of an already-watched item re-enters +Continue Watching. Result: for a heavy importer, the server believes their **entire +watch history (232,979 of 233,016 rows for the worst profile) is resumable**. Every +Continue Watching page load walks that entire list, filters ~100% of it away in +memory, pages deeper, and repeats. + +**Technical.** All current write paths enforce the invariant "completed ⇒ +`position_seconds = 0`" (`internal/userstore/pgstore/progress.go`, +`internal/historyimport/repo.go:879`). But 4,318,693 of 4,579,210 +`user_watch_progress` rows (94%, across 1,649 profiles) violate it with +`completed = TRUE AND position_seconds > 0` — legacy imports from before the +invariant. Only 189 of those are genuine mid-rewatch rows +(`position_seconds < duration_seconds`); the rest are parked at/past the end and +can never be a meaningful resume point. + +The `in_progress` branch of `ListProgress` +(`internal/userstore/pgstore/progress.go:426`) filters +`position_seconds > 0 … ORDER BY updated_at DESC LIMIT … OFFSET …`. No index +serves that shape (the partial indexes cover `completed = true/false`, not +`position_seconds > 0`), so **every call walks all of the profile's rows via +`idx_user_watch_progress_profile` and top-N-sorts them** — measured 192ms warm / +multi-second cold per call for the worst profile (EXPLAIN ANALYZE: 232,962 rows +walked per call). + +Every consumer loops this query per request: + +- `internal/sections/fetcher.go` `collectContinueProgressItems`: up to 10 pages + (`continueProgressMaxScanned = 1000` / page size 100) — serves the native + Continue Watching section (the `api/v1/home` section-items route, 42s) **and**, + since PR #292, the jellycompat Resume fast path (`loadResumeViaSections`) — the + 125s `Users/{userId}/Items/Resume` calls. +- `internal/jellycompat/handlers_items.go` `loadProgressPage`: up to + `resumeScanMaxRows = 300` rows per request (the PR #292 cap — it bounds pages, + not per-page cost). +- `internal/catalog/nextup_repo.go` `listResumableFirstEpisodes`: one 100-row call. + +The 00:30 UTC log window shows the failure shape directly: ~17 sequential 1.1s +executions of this query (one paging loop) plus 16.5s cold executions saturating +the pool while other endpoints queue behind them. + +**Fix (one-time direct DB repair — applied 2026-07-06, no migration shipped).** + +Because this is a one-shot repair of legacy data on a single deployment, it was +applied directly against the production database instead of as a Goose +migration; the exact SQL and timings are recorded in the deployment's ops notes +(silo-base, `slow-endpoint-db-repair-2026-07-06.md`). + +1. Data repair: `UPDATE user_watch_progress SET position_seconds = 0 WHERE + completed = TRUE AND position_seconds > 0 AND position_seconds >= + duration_seconds`, followed by `ANALYZE user_watch_progress`. Leaves the 189 + genuine mid-rewatch rows alone; does not touch `updated_at` or `synced_seq` + (no sync flood, ordering preserved). These rows were already invisible in + Continue Watching (dismissal/superseded/percent filtering), so no visible + behavior changes — the data just stops lying to the query planner. Applied: + 4,318,504 rows in 2m05s. +2. Partial index `idx_uwp_profile_resume (user_id, profile_id, updated_at DESC) + WHERE position_seconds > 0` (`CREATE INDEX CONCURRENTLY`) — turns every + in-progress listing into an ordered index walk regardless of profile size + (belt-and-braces against future bad data; after the repair the index is + 8.8 MB). Verified post-repair: the worst profile's in-progress page query + went from 232,962 rows walked / 192ms warm to 37 rows / 1.0ms. + +## Root cause 2 — Next Up anchors on the entire completed history + +**Human-readable.** "Next Up" answers "what's the next episode of each show this +person is watching?". To find those shows it re-reads **every episode the person +has ever finished** — a quarter-million rows for bulk importers — on every call, +then for each fully-watched show walks all of its episodes looking for an +unwatched one that isn't there. + +**Technical.** `buildListNextUpQuery` (`internal/catalog/nextup_repo.go`): the +`completed_episodes` CTE does `DISTINCT ON (e.series_id)` over **all** of the +profile's completed rows joined to `episodes` (233k rows for the worst profile), +then `eligible_series` runs a correlated anti-join against the (currently +non-selective, see RC1) `position_seconds > 0` set, then a per-series LATERAL +probes episodes in order — scanning *every* episode of a fully-watched series +before yielding nothing. 648 slow executions, 44.7s worst. This also drags down +the native home-sections aggregate (`api/v1/home` sections) via `maybeInjectNextUp` and the jellycompat +`/Shows/NextUp` route. + +**Fix.** Bound the anchor for the global (non-`SeriesID`) query: a +`recent_completed` pre-CTE takes the most recent `nextUpAnchorMaxRows = 500` +completed rows via `idx_uwp_profile_completed` (an ordered index walk), and +`completed_episodes` derives series from that subset. A Next Up rail shows ~24 +series; the 500 most recent completions cover every realistically surfaceable +series. Series-scoped calls (show-detail tile) keep the unbounded shape — they +are naturally bounded by one series. Prototyped on the live worst profile: +**44.7s → 517ms** (and the residual cost is the RC1 anti-join, which the repair +removes). + +## Root cause 3 — series watch-state rollup materializes every episode of every series on the page + +**Human-readable.** For any list of TV shows (per-library "Latest", library +browse, search results), the server computes each show's "N unwatched episodes" +badge by **loading every episode of every show on the page into memory** and then +asking the watch database about each episode in batches of 500. One 50-show page +of the Sports library expands to 32,467 episodes and ~65 sequential database +round-trips. PR #292 wired the cached Latest fast path through this same rollup +(for data parity), so even cache-hit responses pay it. + +**Technical.** `enrichSeriesListUserData` (`internal/jellycompat/content_direct.go`) +→ `episodeRepo.ListBySeriesIDs` (all episodes) → `chunkedProgressByMediaItems` → +`ListProgressWithCompletedHistory` per 500 ids (each chunk hits +`user_watch_progress` + the `ListCompletedHistoryItems` GROUP BY). The same +per-episode fanout runs in `enrichDetailUserData` for every series row on detail +pages. It only ever produces four numbers per series (total/watched/in-progress/ +played). + +**Fix.** Compute the counts in one SQL aggregate. New optional interface +`userstore.SeriesEpisodeRollupStore`, implemented by `PostgresUserStore` +(the pgstore already references catalog tables — see +`buildProgressCatalogFilter`): a single `GROUP BY e.series_id` query over +`episodes` LEFT-JOINed to the profile's progress rows with the same +visibility/history semantics as `ListProgressWithCompletedHistory` +(hidden-items anti-join, completed-history fold). `enrichSeriesListUserData` and +`enrichDetailUserData` use it when the store implements it and keep the existing +chunked path as fallback (SQLite-backed user stores). Prototyped on the live +worst profile against the real Sports Latest page: **~17s → 123ms**. + +This also fixes the slow `/Items?searchTerm=…` calls (Meilisearch itself is fast; +the search handler excludes movies/episodes, returns mostly series, and then paid +this same rollup) and the series portions of `/Items` browse. + +## Root cause 4 (secondary) — two index-starved write/maintenance paths + +- `RemoveHistoryItems` (`internal/userstore/pgstore/progress.go`): the watermark + MAX and the DELETE filter `user_watch_history` by `(user_id, profile_id, + media_item_id = ANY(...))`, but the only complete index is `(user_id, + profile_id, watched_at DESC)` — per-user full history scans (10.9s worst; this + is the `/UserPlayedItems/{itemId}` DELETE path, 19.4s worst end-to-end). + Fix: plain btree `(user_id, profile_id, media_item_id)`. +- Scanner subtree lookups (`internal/scanner/file_repo.go`): + `media_folder_id = $1 AND (file_path = $2 OR file_path LIKE $3 ESCAPE '\')` has + no usable index for the path predicate (13.6s worst; holds pool connections that + API requests then queue behind). The LIKE is prefix-anchored + (`pathscope.PrefixLike`), so a btree on `(media_folder_id, file_path + text_pattern_ops)` serves both arms and the `ORDER BY file_path`. + +Both indexes were created directly on the production database +(`CREATE INDEX CONCURRENTLY`) alongside RC1's partial index — see the +deployment's ops notes. + +## What is intentionally not changed + +- `resumeScanMaxRows` / `continueProgressMaxScanned` / `maxSeriesUserDataRollups` + caps stay — they remain correct guards; the fixes make each capped unit cheap. +- The `position_seconds > 0` rewatch semantics stay; the repair only removes rows + that violate the documented write-path invariant. +- No `updated_at`/`synced_seq` changes in the repair — client sync state is + untouched. +- `/api/v1/stream/{session_id}/subtitles/{track}` (subtitle *track* conversion) + is out of scope per the task. The related font-attachment endpoint + `/subtitles/{track}/fonts` *was* folded in as a follow-up — see deliverable 5 — + because its per-attachment ffmpeg spawns shared the same slow-endpoint profile. + +## Deliverables + +1. `docs:` this plan. +2. One-time DB repair + three indexes, applied directly to the production + database on 2026-07-06 (recorded in the deployment's ops notes; deliberately + not shipped as a migration). +3. `perf(catalog):` bounded next-up anchor scan. +4. `perf(jellycompat,userstore):` SQL series watch-state rollup with chunked + fallback. +5. `perf(playback):` single-pass ffmpeg font extraction for + `/subtitles/{track}/fonts` (one media-file open instead of one per + attachment), with the 32-attachment / 32 MiB caps preserved via a dump-dir + watchdog. + +## Verification + +- `go build ./... && go vet ./...`; `go test -race` on `internal/catalog`, + `internal/jellycompat`, `internal/userstore/...`, `internal/sections`. +- EXPLAIN ANALYZE numbers above were measured on the production database + against the worst real profile, before and after the repair: in-progress page + query 192ms/232,962 rows → 1.0ms/37 rows; deployed (unbounded) next-up shape + 17–44s → 1.1s; bounded next-up shape → 10ms; 50-series rollup ~17s → 119ms. +- Post-deploy of the code fixes: re-run the slow-request aggregation over + `docker logs` and confirm the five endpoint groups drop out of the ≥1s report. diff --git a/internal/catalog/continue_watching_progress.go b/internal/catalog/continue_watching_progress.go index a6c162dd..3c0c7c1b 100644 --- a/internal/catalog/continue_watching_progress.go +++ b/internal/catalog/continue_watching_progress.go @@ -3,6 +3,7 @@ package catalog import ( "context" "fmt" + "log/slog" "strings" "time" @@ -40,6 +41,18 @@ func NewContinueWatchingProgressFilter(pool *pgxpool.Pool) *ContinueWatchingProg const supersededProgressPageSize = 500 +// supersededProgressMaxPages hard-caps how many completed-history pages the +// superseded-episode walk reads in one request. The updated_at cutoff normally +// halts paging far sooner (an import-heavy profile's completed rows predate its +// active in-progress items, so the scan stops on the first page); this bound +// only engages in the adversarial case of a very old in-progress entry sitting +// behind a large volume of newer completions. Hitting it means the tail of the +// completed set went unscanned, so a genuinely-superseded episode could +// momentarily survive on the Continue Watching row — we log when that happens +// rather than silently mis-filter, and it self-corrects once the stale +// in-progress entry ages out of the scanned window. +const supersededProgressMaxPages = 5 + // 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. @@ -53,7 +66,23 @@ func (f *ContinueWatchingProgressFilter) SupersededEpisodeProgressIDs(ctx contex return map[string]struct{}{}, nil } - completed, err := CompletedProgressSnapshots(ctx, store, profileID) + // A completed episode can only supersede an in-progress one it was finished + // more recently than (the query gates on + // done_progress.updated_at > ip_progress.updated_at). So the only completed + // rows that can matter are those updated after the oldest in-progress entry; + // anything older can supersede nothing. Bounding the completed walk at that + // timestamp keeps import-heavy profiles — whose entire back-catalogue is + // completed=TRUE with old timestamps — from re-paging hundreds of thousands + // of irrelevant rows on every Resume/Continue Watching load (the 60–116s + // tail in the 2026-07-06 slow-query comparison). + oldestInProgress := inProgress[0].UpdatedAt + for _, snapshot := range inProgress[1:] { + if snapshot.UpdatedAt.Before(oldestInProgress) { + oldestInProgress = snapshot.UpdatedAt + } + } + + completed, err := CompletedProgressSnapshots(ctx, store, profileID, oldestInProgress) if err != nil { return nil, err } @@ -84,19 +113,30 @@ func (f *ContinueWatchingProgressFilter) SupersededEpisodeProgressIDs(ctx contex 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) { +// CompletedProgressSnapshots pages through the profile's completed progress +// rows and returns deduplicated snapshots updated after notBefore. The +// completed listing is ordered updated_at DESC (newest first), so once a row at +// or before notBefore is reached every later page is older still and paging +// stops — callers only care about completed episodes finished more recently +// than an in-progress entry, so older rows are irrelevant. Pass a zero +// notBefore to walk the whole history. +func CompletedProgressSnapshots(ctx context.Context, store ProgressLister, profileID string, notBefore time.Time) ([]ProgressSnapshot, error) { seen := make(map[string]struct{}) snapshots := make([]ProgressSnapshot, 0) - for offset := 0; ; offset += supersededProgressPageSize { + for page := 0; page < supersededProgressMaxPages; page++ { + offset := page * 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) } + reachedCutoff := false for _, snapshot := range ProgressSnapshots(entries) { + if !snapshot.UpdatedAt.After(notBefore) { + reachedCutoff = true + break + } contentID := snapshot.ContentID if _, ok := seen[contentID]; ok { continue @@ -105,10 +145,20 @@ func CompletedProgressSnapshots(ctx context.Context, store ProgressLister, profi snapshots = append(snapshots, snapshot) } - if len(entries) < supersededProgressPageSize { + if reachedCutoff || len(entries) < supersededProgressPageSize { return snapshots, nil } } + + // Fell out of the loop with a full final page: the page cap halted the walk + // before the cutoff, so completed rows past the scanned window were skipped. + // Log it so a real profile that trips this backstop is visible rather than + // silently under-filtered. + slog.Warn("continue-watching: superseded-episode walk hit page cap; completed-history tail left unscanned", + "profile_id", profileID, + "pages_scanned", supersededProgressMaxPages, + "rows_scanned", len(snapshots)) + return snapshots, nil } // ProgressSnapshots converts progress rows to snapshots, dropping rows with a diff --git a/internal/catalog/continue_watching_progress_test.go b/internal/catalog/continue_watching_progress_test.go index 74789a75..58268a77 100644 --- a/internal/catalog/continue_watching_progress_test.go +++ b/internal/catalog/continue_watching_progress_test.go @@ -2,6 +2,7 @@ package catalog import ( "context" + "strconv" "strings" "testing" "time" @@ -41,7 +42,7 @@ func TestCompletedProgressSnapshotsPagesThroughConfiguredStore(t *testing.T) { } store := &stubProgressLister{entries: entries} - snapshots, err := CompletedProgressSnapshots(context.Background(), store, "p1") + snapshots, err := CompletedProgressSnapshots(context.Background(), store, "p1", time.Time{}) if err != nil { t.Fatalf("CompletedProgressSnapshots: %v", err) } @@ -59,6 +60,63 @@ func TestCompletedProgressSnapshotsPagesThroughConfiguredStore(t *testing.T) { } } +func TestCompletedProgressSnapshotsStopsAtCutoff(t *testing.T) { + t.Parallel() + + // Newest-first, spanning two full pages: the store hands rows back in + // updated_at DESC order the same way the completed listing query does. + entries := make([]userstore.WatchProgress, 2*supersededProgressPageSize) + base := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC) + for i := range entries { + entries[i] = userstore.WatchProgress{ + MediaItemID: "done-" + strconv.Itoa(i), + UpdatedAt: base.Add(time.Duration(-i) * time.Minute).Format(time.RFC3339), + } + } + store := &stubProgressLister{entries: entries} + + // Cut off inside the first page: only rows strictly newer than the cutoff + // are returned, and paging stops without ever reading the second page. + cutoff := base.Add(-10 * time.Minute) + snapshots, err := CompletedProgressSnapshots(context.Background(), store, "p1", cutoff) + if err != nil { + t.Fatalf("CompletedProgressSnapshots: %v", err) + } + if len(snapshots) != 10 { + t.Fatalf("completed snapshots count = %d, want 10 (rows newer than cutoff)", len(snapshots)) + } + if len(store.calls) != 1 { + t.Fatalf("ListProgress calls = %+v, want a single page before the cutoff halts paging", store.calls) + } +} + +func TestCompletedProgressSnapshotsHaltsAtPageCap(t *testing.T) { + t.Parallel() + + // More completed history than the page cap allows, all newer than the + // (zero) cutoff so nothing halts the walk except the cap itself. + entries := make([]userstore.WatchProgress, (supersededProgressMaxPages+1)*supersededProgressPageSize) + base := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC) + for i := range entries { + entries[i] = userstore.WatchProgress{ + MediaItemID: "done-" + strconv.Itoa(i), + UpdatedAt: base.Add(time.Duration(-i) * time.Second).Format(time.RFC3339), + } + } + store := &stubProgressLister{entries: entries} + + snapshots, err := CompletedProgressSnapshots(context.Background(), store, "p1", time.Time{}) + if err != nil { + t.Fatalf("CompletedProgressSnapshots: %v", err) + } + if len(store.calls) != supersededProgressMaxPages { + t.Fatalf("ListProgress calls = %d, want %d (page cap)", len(store.calls), supersededProgressMaxPages) + } + if len(snapshots) != supersededProgressMaxPages*supersededProgressPageSize { + t.Fatalf("completed snapshots count = %d, want %d (capped pages)", len(snapshots), supersededProgressMaxPages*supersededProgressPageSize) + } +} + func TestBuildSupersededEpisodeProgressQueryUsesStoreSnapshotsWithFreshnessGate(t *testing.T) { t.Parallel() diff --git a/internal/catalog/nextup_repo.go b/internal/catalog/nextup_repo.go index 562be722..107e6ba7 100644 --- a/internal/catalog/nextup_repo.go +++ b/internal/catalog/nextup_repo.go @@ -109,6 +109,21 @@ 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 + func buildListNextUpQuery(q NextUpQuery, limit int) (string, []interface{}) { args := []interface{}{q.UserID, q.ProfileID, limit} argIdx := 4 @@ -154,8 +169,14 @@ func buildListNextUpQuery(q NextUpQuery, limit int) (string, []interface{}) { sourceTable = "eligible_series" } - query := fmt.Sprintf(` - WITH completed_episodes AS ( + // 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. + var completedEpisodesCTE string + if q.SeriesID != "" { + completedEpisodesCTE = fmt.Sprintf(`completed_episodes AS ( SELECT DISTINCT ON (e.series_id) e.series_id, e.season_number, @@ -177,7 +198,40 @@ func buildListNextUpQuery(q NextUpQuery, limit int) (string, []interface{}) { %s %s 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 + ), + 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) + } + + query := fmt.Sprintf(` + WITH %s %s SELECT next_ep.content_id, @@ -205,7 +259,7 @@ func buildListNextUpQuery(q NextUpQuery, limit int) (string, []interface{}) { LIMIT 1 ) next_ep ON true ORDER BY es.updated_at DESC - LIMIT $3`, seriesFilter, dateCutoffFilter, inProgressExclusion, sourceTable) + LIMIT $3`, completedEpisodesCTE, inProgressExclusion, sourceTable) return query, args } diff --git a/internal/catalog/nextup_repo_test.go b/internal/catalog/nextup_repo_test.go index 5e4f12aa..c01da4bd 100644 --- a/internal/catalog/nextup_repo_test.go +++ b/internal/catalog/nextup_repo_test.go @@ -1,6 +1,7 @@ package catalog import ( + "fmt" "strings" "testing" ) @@ -39,6 +40,47 @@ func TestBuildListNextUpQuery_PrefersRecentCompletedOverOlderPartialProgress(t * } } +func TestBuildListNextUpQuery_GlobalBoundsAnchorScan(t *testing.T) { + t.Parallel() + + query, _ := buildListNextUpQuery(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) + } +} + +func TestBuildListNextUpQuery_SeriesScopedKeepsUnboundedAnchor(t *testing.T) { + t.Parallel() + + query, args := buildListNextUpQuery(NextUpQuery{ + UserID: 7, + ProfileID: "profile-1", + SeriesID: "series-42", + }, 20) + + // 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 + // long-idle series' tile disappear. + 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, "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) + } +} + func TestBuildListNextUpQuery_EnableResumableSkipsSeriesSuppressionCTE(t *testing.T) { t.Parallel() diff --git a/internal/catalog/user_data_rollup.go b/internal/catalog/user_data_rollup.go index a58d71a8..be4b4848 100644 --- a/internal/catalog/user_data_rollup.go +++ b/internal/catalog/user_data_rollup.go @@ -5,6 +5,21 @@ import ( "github.com/Silo-Server/silo-server/internal/userstore" ) +// SeasonUserDataFromCounts builds the aggregate watch state DTO from a +// SQL-side rollup (userstore.SeriesEpisodeRollupStore). It must stay +// value-for-value identical to EpisodeRollupUserData over the same episodes. +func SeasonUserDataFromCounts(counts userstore.SeriesWatchCounts) *SeasonUserData { + if counts.TotalEpisodes == 0 { + return &SeasonUserData{} + } + return &SeasonUserData{ + WatchedCount: counts.WatchedCount, + UnplayedCount: counts.TotalEpisodes - counts.WatchedCount, + InProgressCount: counts.InProgressCount, + Played: counts.WatchedCount == counts.TotalEpisodes, + } +} + // EpisodeRollupUserData computes aggregate watch state for a season or series // from pre-fetched per-episode progress. Completed history should already be // folded into progressMap by the caller's userstore helper. diff --git a/internal/jellycompat/content_direct.go b/internal/jellycompat/content_direct.go index 2641dc16..abdb3701 100644 --- a/internal/jellycompat/content_direct.go +++ b/internal/jellycompat/content_direct.go @@ -640,8 +640,20 @@ func (s *directContentService) enrichDetailUserData(ctx context.Context, store u // A series never has a progress row of its own, so roll watch state up from // its episodes (mirrors applySeasonUserData) to give clients Played/ - // UnplayedItemCount at the series level. + // UnplayedItemCount at the series level. Postgres-backed stores aggregate + // the rollup in SQL; the chunked per-episode path remains as fallback. if result.UserData == nil && strings.EqualFold(result.Type, "series") && s.episodeRepo != nil { + if rollupStore, ok := store.(userstore.SeriesEpisodeRollupStore); ok { + counts, err := rollupStore.SeriesEpisodeWatchCounts(ctx, profileID, []string{contentID}) + if err == nil { + // A series absent from counts has no available episodes; the + // zero-value counts produce the same empty rollup the + // episode-list path returned for it. + result.UserData = catalog.SeasonUserDataFromCounts(counts[contentID]) + return + } + slog.Warn("series watch rollup query failed, falling back to per-episode rollup", "error", err) + } if episodesBySeries, epErr := s.episodeRepo.ListBySeriesIDs(ctx, []string{contentID}); epErr == nil { episodes := episodesBySeries[contentID] episodeIDs := modelEpisodeContentIDs(episodes) @@ -1028,6 +1040,28 @@ func (s *directContentService) enrichSeriesListUserData(ctx context.Context, ses if len(seriesIDs) > maxSeriesUserDataRollups { seriesIDs = seriesIDs[:maxSeriesUserDataRollups] } + + // Fast path: Postgres-backed stores aggregate the rollup in one SQL query + // instead of loading every episode of every series and batching progress + // lookups (a 50-series page of an episode-heavy library expanded to 32k + // episode rows and ~65 sequential queries). A rollup failure falls through + // to the chunked path below. + if rollupStore, ok := store.(userstore.SeriesEpisodeRollupStore); ok { + counts, err := rollupStore.SeriesEpisodeWatchCounts(ctx, session.ProfileID, seriesIDs) + if err == nil { + for i := range items { + if items[i].UserData != nil || !strings.EqualFold(items[i].Type, "series") { + continue + } + if c, ok := counts[items[i].ContentID]; ok { + items[i].UserData = catalog.SeasonUserDataFromCounts(c) + } + } + return + } + slog.Warn("series watch rollup query failed, falling back to per-episode rollup", "error", err) + } + episodesBySeries, err := s.episodeRepo.ListBySeriesIDs(ctx, seriesIDs) if err != nil { return diff --git a/internal/jellycompat/content_direct_test.go b/internal/jellycompat/content_direct_test.go index d5e60d78..ae56fbe0 100644 --- a/internal/jellycompat/content_direct_test.go +++ b/internal/jellycompat/content_direct_test.go @@ -971,3 +971,95 @@ func makeBrowseTestMediaItems(count int) []*models.MediaItem { } return items } + +// seriesRollupCountingStore layers userstore.SeriesEpisodeRollupStore on top +// of the panic-stub store: when the SQL rollup capability is present, the +// chunked ListProgressByMediaItems path must not run at all. +type seriesRollupCountingStore struct { + *progressCountingStore + counts map[string]userstore.SeriesWatchCounts + rollupCalls int +} + +func (s *seriesRollupCountingStore) SeriesEpisodeWatchCounts(_ context.Context, _ string, seriesIDs []string) (map[string]userstore.SeriesWatchCounts, error) { + s.rollupCalls++ + out := make(map[string]userstore.SeriesWatchCounts, len(seriesIDs)) + for _, id := range seriesIDs { + if c, ok := s.counts[id]; ok { + out[id] = c + } + } + return out, nil +} + +// TestEnrichSeriesUserDataUsesSQLRollup: a store exposing the SQL rollup +// capability serves the series watch-state badge from one aggregate call — +// no episode-list materialization, no chunked per-episode progress queries — +// and produces the same SeasonUserData shape as the chunked path. +func TestEnrichSeriesUserDataUsesSQLRollup(t *testing.T) { + store := &seriesRollupCountingStore{ + progressCountingStore: &progressCountingStore{}, + counts: map[string]userstore.SeriesWatchCounts{ + "series-1": {TotalEpisodes: 3, WatchedCount: 2, InProgressCount: 1}, + }, + } + // The episode source panics on use: the rollup path must never list + // episodes. + svc := &directContentService{ + episodeRepo: &seriesEpisodeSource{}, + storeProvider: &completedProgressStoreProvider{store: store}, + } + + items := []upstreamListItem{ + {ContentID: "series-1", Type: "series", Title: "Show"}, + {ContentID: "series-2", Type: "series", Title: "No Episodes"}, + {ContentID: "movie-1", Type: "movie", Title: "Film"}, + } + svc.EnrichSeriesUserData(context.Background(), &Session{StreamAppUserID: 1, ProfileID: "profile-1"}, items) + + if store.rollupCalls != 1 { + t.Fatalf("expected exactly one rollup call, got %d", store.rollupCalls) + } + if store.listProgressCalls != 0 { + t.Fatalf("chunked progress path must not run when the SQL rollup is available, got %d calls", store.listProgressCalls) + } + + want := &catalog.SeasonUserData{WatchedCount: 2, UnplayedCount: 1, InProgressCount: 1, Played: false} + if got := items[0].UserData; got == nil || *got != *want { + t.Fatalf("series-1 UserData = %+v, want %+v", got, want) + } + // A series absent from the rollup result (no available episodes) keeps a + // nil UserData — the same shape the episode-list path gave it. + if items[1].UserData != nil { + t.Fatalf("series-2 UserData = %+v, want nil", items[1].UserData) + } + if items[2].UserData != nil { + t.Fatalf("movie UserData = %+v, want nil", items[2].UserData) + } +} + +// TestSeasonUserDataFromCountsMatchesEpisodeRollup pins the SQL-side counts +// mapping to the in-memory EpisodeRollupUserData it replaces. +func TestSeasonUserDataFromCountsMatchesEpisodeRollup(t *testing.T) { + episodes := []*models.Episode{ + {ContentID: "ep-1"}, + {ContentID: "ep-2"}, + {ContentID: "ep-3"}, + } + progress := map[string]userstore.WatchProgress{ + "ep-1": {MediaItemID: "ep-1", Completed: true}, + "ep-2": {MediaItemID: "ep-2", PositionSeconds: 42}, + } + + fromEpisodes := catalog.EpisodeRollupUserData(episodes, progress) + fromCounts := catalog.SeasonUserDataFromCounts(userstore.SeriesWatchCounts{ + TotalEpisodes: 3, WatchedCount: 1, InProgressCount: 1, + }) + if *fromEpisodes != *fromCounts { + t.Fatalf("counts mapping diverged: episodes=%+v counts=%+v", fromEpisodes, fromCounts) + } + + if empty := catalog.SeasonUserDataFromCounts(userstore.SeriesWatchCounts{}); *empty != (catalog.SeasonUserData{}) { + t.Fatalf("zero counts must produce the empty rollup, got %+v", empty) + } +} diff --git a/internal/playback/subtitle_fonts.go b/internal/playback/subtitle_fonts.go index ab1ffc2f..cae4a1a4 100644 --- a/internal/playback/subtitle_fonts.go +++ b/internal/playback/subtitle_fonts.go @@ -1,15 +1,18 @@ package playback import ( - "bytes" "context" "encoding/base64" "encoding/json" + "errors" "fmt" - "io" + "os" "os/exec" "path/filepath" + "strconv" "strings" + "sync/atomic" + "time" ) const ( @@ -66,26 +69,7 @@ func ExtractAttachedSubtitleFonts(ctx context.Context, inputPath string, ffmpegP bin = "ffmpeg" } - var total int64 - fonts := make([]SubtitleFontAttachment, 0, len(streams)) - for i, stream := range streams { - fallbackName := fmt.Sprintf("attachment-%d%s", i, fontAttachmentExt(stream)) - remaining := maxSubtitleFontBytes - total - data, err := extractFontAttachment(ctx, inputPath, bin, stream, fallbackName, remaining) - if err != nil { - return nil, err - } - total += int64(len(data)) - if total > maxSubtitleFontBytes { - return nil, fmt.Errorf("subtitle fonts: attached font data exceeds %d bytes", maxSubtitleFontBytes) - } - fonts = append(fonts, SubtitleFontAttachment{ - Name: safeAttachmentDisplayName(stream, fallbackName), - Data: data, - }) - } - - return fonts, nil + return dumpFontAttachments(ctx, inputPath, bin, streams, maxSubtitleFontBytes) } // EncodeSubtitleFontBundle converts raw font attachments to base64 JSON items. @@ -100,59 +84,143 @@ func EncodeSubtitleFontBundle(fonts []SubtitleFontAttachment) []SubtitleFontBund return items } -func extractFontAttachment(ctx context.Context, inputPath string, ffmpegPath string, stream attachmentProbeStream, name string, maxBytes int64) ([]byte, error) { - if maxBytes <= 0 { - return nil, fmt.Errorf("subtitle fonts: attached font data exceeds %d bytes", maxSubtitleFontBytes) - } - - cmdCtx, cancel := context.WithCancel(ctx) - defer cancel() - - args := []string{ - "-hide_banner", "-nostats", "-loglevel", "error", - fmt.Sprintf("-dump_attachment:%d", stream.Index), "pipe:1", - "-i", inputPath, - "-map", "0:t?", - "-c", "copy", - "-f", "null", "-", - } - cmd := exec.CommandContext(cmdCtx, ffmpegPath, args...) - stdout, err := cmd.StdoutPipe() +// dumpFontAttachments extracts every attachment in a single ffmpeg invocation. +// +// ffmpeg re-opens the (often network-backed) media file once per process, so +// the previous one-process-per-font approach paid that open cost N times and +// dominated latency — 17–60 s for anime releases carrying 15–47 fonts. Dumping +// all attachments in one pass opens the file once, cutting p95 from ~30 s to +// ~1–2 s. The `-map 0:t? -c copy` flags are required: without them ffmpeg +// decodes the whole video stream instead of stream-copying the attachments. +func dumpFontAttachments(ctx context.Context, inputPath string, ffmpegPath string, streams []attachmentProbeStream, maxBytes int64) ([]SubtitleFontAttachment, error) { + dir, err := os.MkdirTemp("", "silo-subfonts-*") if err != nil { - return nil, fmt.Errorf("subtitle fonts: open attachment pipe %q: %w", name, err) + return nil, fmt.Errorf("subtitle fonts: create temp dir: %w", err) } + defer os.RemoveAll(dir) + + // A fresh temp dir guarantees the dump targets don't pre-exist, so ffmpeg + // never prompts to overwrite (which would hang the process). + args := make([]string, 0, 4+len(streams)*2+8) + args = append(args, "-hide_banner", "-nostats", "-loglevel", "error") + paths := make([]string, len(streams)) + for i, stream := range streams { + paths[i] = filepath.Join(dir, strconv.Itoa(i)) + args = append(args, fmt.Sprintf("-dump_attachment:%d", stream.Index), paths[i]) + } + args = append(args, "-i", inputPath, "-map", "0:t?", "-c", "copy", "-f", "null", "-") + + cmd := exec.CommandContext(ctx, ffmpegPath, args...) var stderr strings.Builder cmd.Stderr = &stderr - if err := cmd.Start(); err != nil { - return nil, fmt.Errorf("subtitle fonts: start attachment extract %q: %w", name, err) + return nil, fmt.Errorf("subtitle fonts: start attachment extract: %w", err) } - var buf bytes.Buffer - _, copyErr := io.Copy(&buf, &io.LimitedReader{R: stdout, N: maxBytes + 1}) - tooLarge := int64(buf.Len()) > maxBytes - if tooLarge { - cancel() - if cmd.Process != nil { - _ = cmd.Process.Kill() - } - } + // ffmpeg writes each attachment to disk in full before we get a chance to + // read it, so unlike the old pipe-per-attachment reader (which killed at + // maxBytes+1) nothing here bounds what ffmpeg spills. Guard against a + // container whose oversized "font" attachments would otherwise fill the + // disk by killing the process once the dump dir crosses the cap. + overLimit, stopWatch := watchDumpSize(cmd, dir, maxBytes) - waitErr := cmd.Wait() - if tooLarge { - return nil, fmt.Errorf("subtitle fonts: attached font data exceeds %d bytes", maxSubtitleFontBytes) + runErr := cmd.Wait() + stopWatch() + if overLimit.Load() { + return nil, fmt.Errorf("subtitle fonts: attached font data exceeds %d bytes", maxBytes) } - if copyErr != nil { - return nil, fmt.Errorf("subtitle fonts: read attachment %q: %w", name, copyErr) - } - if waitErr != nil { + if runErr != nil { if ctx.Err() != nil { return nil, ctx.Err() } - return nil, fmt.Errorf("subtitle fonts: extract attachment %q: %w (stderr: %s)", - name, waitErr, truncateStderr(stderr.String())) + return nil, fmt.Errorf("subtitle fonts: extract attachments: %w (stderr: %s)", + runErr, truncateStderr(stderr.String())) } - return buf.Bytes(), nil + + var total int64 + fonts := make([]SubtitleFontAttachment, 0, len(streams)) + for i, stream := range streams { + fallbackName := fmt.Sprintf("attachment-%d%s", i, fontAttachmentExt(stream)) + // Stat before reading so an over-limit attachment trips the cap + // without being pulled into memory. + info, err := os.Stat(paths[i]) + if errors.Is(err, os.ErrNotExist) { + // ffmpeg silently skips attachments it can't stream-copy; treat + // a missing dump file as an absent font rather than a failure. + continue + } + if err != nil { + return nil, fmt.Errorf("subtitle fonts: stat attachment %q: %w", fallbackName, err) + } + total += info.Size() + if total > maxBytes { + return nil, fmt.Errorf("subtitle fonts: attached font data exceeds %d bytes", maxBytes) + } + data, err := os.ReadFile(paths[i]) + if err != nil { + return nil, fmt.Errorf("subtitle fonts: read attachment %q: %w", fallbackName, err) + } + fonts = append(fonts, SubtitleFontAttachment{ + Name: safeAttachmentDisplayName(stream, fallbackName), + Data: data, + }) + } + + return fonts, nil +} + +// watchDumpSize polls the dump directory while ffmpeg runs and kills the +// process if the total bytes written exceed maxBytes, restoring the hard size +// bound the previous streaming reader enforced. It returns a flag set when the +// cap is tripped and a stop function the caller must invoke after cmd.Wait. +// The worst-case overshoot is one poll interval of ffmpeg writes, which is far +// smaller than the unbounded spill it replaces. +func watchDumpSize(cmd *exec.Cmd, dir string, maxBytes int64) (*atomic.Bool, func()) { + overLimit := &atomic.Bool{} + stop := make(chan struct{}) + done := make(chan struct{}) + go func() { + defer close(done) + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + for { + select { + case <-stop: + return + case <-ticker.C: + if dirBytes(dir) > maxBytes { + overLimit.Store(true) + if cmd.Process != nil { + _ = cmd.Process.Kill() + } + return + } + } + } + }() + return overLimit, func() { + close(stop) + <-done + } +} + +// dirBytes returns the total size of the regular files directly inside dir. +// Errors (e.g. a file removed mid-scan) are treated as zero so the watchdog +// never blocks extraction on a transient stat failure. +func dirBytes(dir string) int64 { + entries, err := os.ReadDir(dir) + if err != nil { + return 0 + } + var total int64 + for _, e := range entries { + info, err := e.Info() + if err != nil { + continue + } + total += info.Size() + } + return total } func probeFontAttachmentStreams(ctx context.Context, inputPath string, ffprobePath string) ([]attachmentProbeStream, error) { diff --git a/internal/playback/subtitle_fonts_test.go b/internal/playback/subtitle_fonts_test.go index e6ebeb08..be3d42a8 100644 --- a/internal/playback/subtitle_fonts_test.go +++ b/internal/playback/subtitle_fonts_test.go @@ -9,7 +9,92 @@ import ( "testing" ) -func TestExtractAttachedSubtitleFontsUsesBoundedStdoutExtraction(t *testing.T) { +// fakeFFmpegDumping returns a shell script that mimics ffmpeg's +// -dump_attachment behaviour: it writes payload to every path that follows a +// -dump_attachment:* flag. This exercises the single-invocation extractor +// without a real ffmpeg. The shebang must stay on the first line. +func fakeFFmpegDumping(payload string) string { + return "#!/bin/sh\nPAYLOAD='" + payload + `' +prev="" +for a in "$@"; do + case "$prev" in + -dump_attachment:*) printf '%s' "$PAYLOAD" > "$a" ;; + esac + prev="$a" +done +` +} + +// TestDumpFontAttachmentsArgvOrder pins the ffmpeg argument contract: every +// -dump_attachment flag must precede -i (they are per-input options for the +// following input), and the -map 0:t? / -c copy stream-copy flags must be +// present (without them ffmpeg decodes the whole video). A fake ffmpeg records +// its argv so a malformed invocation can't pass silently. +func TestDumpFontAttachmentsArgvOrder(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("shell script test helper is unix-only") + } + + dir := t.TempDir() + argvFile := filepath.Join(dir, "argv") + ffmpegPath := filepath.Join(dir, "ffmpeg") + writeExecutable(t, ffmpegPath, "#!/bin/sh\nprintf '%s\\n' \"$@\" > '"+argvFile+`' +prev="" +for a in "$@"; do + case "$prev" in + -dump_attachment:*) printf 'x' > "$a" ;; + esac + prev="$a" +done +`) + + if _, err := dumpFontAttachments(context.Background(), "input.mkv", ffmpegPath, + []attachmentProbeStream{{Index: 2}, {Index: 5}}, maxSubtitleFontBytes); err != nil { + t.Fatalf("dumpFontAttachments returned error: %v", err) + } + + raw, err := os.ReadFile(argvFile) + if err != nil { + t.Fatalf("read argv: %v", err) + } + argv := strings.Split(strings.TrimRight(string(raw), "\n"), "\n") + + inputIdx, mapIdx := -1, -1 + lastDumpIdx := -1 + for i, a := range argv { + switch { + case a == "-i": + inputIdx = i + case a == "-map": + mapIdx = i + case strings.HasPrefix(a, "-dump_attachment:"): + lastDumpIdx = i + } + } + if inputIdx < 0 { + t.Fatalf("argv missing -i: %v", argv) + } + if lastDumpIdx < 0 || lastDumpIdx > inputIdx { + t.Fatalf("dump_attachment flags must precede -i; argv=%v", argv) + } + if mapIdx < 0 || argv[mapIdx+1] != "0:t?" { + t.Fatalf("argv missing -map 0:t?: %v", argv) + } + if !containsSeq(argv, "-c", "copy") { + t.Fatalf("argv missing -c copy: %v", argv) + } +} + +func containsSeq(argv []string, a, b string) bool { + for i := 0; i+1 < len(argv); i++ { + if argv[i] == a && argv[i+1] == b { + return true + } + } + return false +} + +func TestExtractAttachedSubtitleFontsSingleInvocation(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("shell script test helper is unix-only") } @@ -18,45 +103,72 @@ func TestExtractAttachedSubtitleFontsUsesBoundedStdoutExtraction(t *testing.T) { ffmpegPath := filepath.Join(dir, "ffmpeg") writeExecutable(t, filepath.Join(dir, "ffprobe"), `#!/bin/sh cat <<'JSON' -{"streams":[{"index":2,"codec_name":"ttf","codec_type":"attachment","tags":{"filename":"MyFont.ttf","mimetype":"font/ttf"}}]} +{"streams":[{"index":2,"codec_name":"ttf","codec_type":"attachment","tags":{"filename":"MyFont.ttf","mimetype":"font/ttf"}},{"index":3,"codec_name":"otf","codec_type":"attachment","tags":{"filename":"Other.otf","mimetype":"font/otf"}}]} JSON `) - writeExecutable(t, ffmpegPath, `#!/bin/sh -printf 'fontdata' -`) + writeExecutable(t, ffmpegPath, fakeFFmpegDumping("fontdata")) fonts, err := ExtractAttachedSubtitleFonts(context.Background(), "input.mkv", ffmpegPath) if err != nil { t.Fatalf("ExtractAttachedSubtitleFonts returned error: %v", err) } - if len(fonts) != 1 { - t.Fatalf("font count = %d, want 1", len(fonts)) + if len(fonts) != 2 { + t.Fatalf("font count = %d, want 2", len(fonts)) } - if fonts[0].Name != "MyFont.ttf" { - t.Fatalf("font name = %q, want MyFont.ttf", fonts[0].Name) + if fonts[0].Name != "MyFont.ttf" || fonts[1].Name != "Other.otf" { + t.Fatalf("font names = %q/%q, want MyFont.ttf/Other.otf", fonts[0].Name, fonts[1].Name) } if string(fonts[0].Data) != "fontdata" { t.Fatalf("font data = %q, want fontdata", string(fonts[0].Data)) } } -func TestExtractFontAttachmentRejectsOverLimitData(t *testing.T) { +// A dump file ffmpeg never wrote (an attachment it could not stream-copy) must +// be skipped rather than fail the whole bundle. +func TestDumpFontAttachmentsSkipsMissingDumps(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("shell script test helper is unix-only") } dir := t.TempDir() ffmpegPath := filepath.Join(dir, "ffmpeg") + // Only writes the first dump target; the second path is left absent. writeExecutable(t, ffmpegPath, `#!/bin/sh -printf '12345' +prev="" +first=1 +for a in "$@"; do + case "$prev" in + -dump_attachment:*) + if [ "$first" = "1" ]; then printf 'fontdata' > "$a"; first=0; fi ;; + esac + prev="$a" +done `) - _, err := extractFontAttachment( + fonts, err := dumpFontAttachments(context.Background(), "input.mkv", ffmpegPath, + []attachmentProbeStream{{Index: 2}, {Index: 3}}, maxSubtitleFontBytes) + if err != nil { + t.Fatalf("dumpFontAttachments returned error: %v", err) + } + if len(fonts) != 1 { + t.Fatalf("font count = %d, want 1 (missing dump skipped)", len(fonts)) + } +} + +func TestDumpFontAttachmentsRejectsOverLimitData(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("shell script test helper is unix-only") + } + + dir := t.TempDir() + ffmpegPath := filepath.Join(dir, "ffmpeg") + writeExecutable(t, ffmpegPath, fakeFFmpegDumping("12345")) + + _, err := dumpFontAttachments( context.Background(), "input.mkv", ffmpegPath, - attachmentProbeStream{Index: 2}, - "font.ttf", + []attachmentProbeStream{{Index: 2}}, 4, ) if err == nil { diff --git a/internal/userstore/pgstore/progress.go b/internal/userstore/pgstore/progress.go index 013c8dac..12feed1c 100644 --- a/internal/userstore/pgstore/progress.go +++ b/internal/userstore/pgstore/progress.go @@ -688,6 +688,86 @@ func (s *PostgresUserStore) ListProgressByMediaItems(ctx context.Context, profil return result, nil } +// Compile-time capability check: the Postgres store computes series episode +// rollups in SQL (see userstore.SeriesEpisodeRollupStore). +var _ userstore.SeriesEpisodeRollupStore = (*PostgresUserStore)(nil) + +// SeriesEpisodeWatchCounts aggregates per-series episode watch state in one +// query. It replaces the ListBySeriesIDs + chunked +// ListProgressWithCompletedHistory fanout that materialized every episode of +// every requested series (a 50-series page of an episode-heavy library +// expanded to 32k episode rows and ~65 sequential queries, ~17s measured). +// +// Semantics mirror the chunked path exactly: +// - episodes count when they are available (episode_libraries row — the same +// predicate as catalog's episode listings); +// - a progress row is visible unless hidden via user_history_hidden_items +// (updated_at <= hidden_before), matching ListProgressByMediaItems; +// - watched = visible completed progress OR a visible completed history row +// (watched_at <= hidden_before hides it), matching the completed-history +// fold in userstore.ListProgressWithCompletedHistory; +// - in-progress = not watched and visible position_seconds > 0, matching +// catalog.EpisodeRollupUserData. +// +// Series with no available episodes produce no row, so callers keep the same +// "no rollup" behavior the episode-list path had for them. +func (s *PostgresUserStore) SeriesEpisodeWatchCounts(ctx context.Context, profileID string, seriesIDs []string) (map[string]userstore.SeriesWatchCounts, error) { + result := make(map[string]userstore.SeriesWatchCounts, len(seriesIDs)) + if len(seriesIDs) == 0 { + return result, nil + } + + rows, err := s.pool.Query(ctx, ` + SELECT e.series_id, + COUNT(*)::int, + COUNT(*) FILTER (WHERE ws.watched)::int, + COUNT(*) FILTER (WHERE NOT ws.watched AND ws.resumable)::int + FROM episodes e + LEFT JOIN user_watch_progress p + ON p.user_id = $1 AND p.profile_id = $2 AND p.media_item_id = e.content_id + AND NOT EXISTS ( + SELECT 1 FROM user_history_hidden_items hh + WHERE hh.user_id = p.user_id AND hh.profile_id = p.profile_id + AND hh.media_item_id = p.media_item_id AND p.updated_at <= hh.hidden_before + ) + CROSS JOIN LATERAL ( + SELECT + COALESCE(p.completed, FALSE) OR EXISTS ( + SELECT 1 FROM user_watch_history h + WHERE h.user_id = $1 AND h.profile_id = $2 AND h.media_item_id = e.content_id + AND h.completed = TRUE + AND NOT EXISTS ( + SELECT 1 FROM user_history_hidden_items hh + WHERE hh.user_id = h.user_id AND hh.profile_id = h.profile_id + AND hh.media_item_id = h.media_item_id AND h.watched_at <= hh.hidden_before + ) + ) AS watched, + COALESCE(p.position_seconds, 0) > 0 AS resumable + ) ws + WHERE e.series_id = ANY($3::text[]) + AND EXISTS (SELECT 1 FROM episode_libraries el WHERE el.episode_id = e.content_id) + GROUP BY e.series_id`, + s.userID, profileID, seriesIDs, + ) + if err != nil { + return nil, fmt.Errorf("aggregating series episode watch counts: %w", err) + } + defer rows.Close() + + for rows.Next() { + var seriesID string + var counts userstore.SeriesWatchCounts + if err := rows.Scan(&seriesID, &counts.TotalEpisodes, &counts.WatchedCount, &counts.InProgressCount); err != nil { + return nil, fmt.Errorf("scanning series episode watch counts: %w", err) + } + result[seriesID] = counts + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterating series episode watch counts: %w", err) + } + return result, nil +} + func (s *PostgresUserStore) UpdateProgressHints(ctx context.Context, profileID, mediaItemID string, hints userstore.VersionHints) error { _, err := s.pool.Exec(ctx, ` UPDATE user_watch_progress diff --git a/internal/userstore/progress_helpers.go b/internal/userstore/progress_helpers.go index 639bdeff..cca4a275 100644 --- a/internal/userstore/progress_helpers.go +++ b/internal/userstore/progress_helpers.go @@ -61,6 +61,29 @@ func parseHistoryTimestamp(value string) time.Time { return parsed } +// SeriesWatchCounts is the aggregate episode watch state for one series, as +// computed by SeriesEpisodeRollupStore. +type SeriesWatchCounts struct { + TotalEpisodes int + WatchedCount int + InProgressCount int +} + +// SeriesEpisodeRollupStore is an optional store capability: compute the +// per-series episode watch-state rollup (total / watched / in-progress +// episode counts) in SQL instead of materializing every episode of every +// series and batching per-episode progress lookups through +// ListProgressWithCompletedHistory. Implemented by the Postgres store, where +// episodes and progress live in the same database; SQLite-backed stores fall +// back to the chunked in-memory path. Semantics must match +// ListProgressWithCompletedHistory + catalog.EpisodeRollupUserData: an episode +// is watched when its visible progress row is completed or a visible completed +// history row exists, and in-progress when it is not watched and its visible +// progress row has position_seconds > 0. +type SeriesEpisodeRollupStore interface { + SeriesEpisodeWatchCounts(ctx context.Context, profileID string, seriesIDs []string) (map[string]SeriesWatchCounts, error) +} + // CompletedHistoryItemMap returns the latest completed-history item row for a // scoped item query. Lookup failures degrade to an empty map so user-data // enrichment can keep returning progress rows.