From 2dedf3ff26e488dce37d5ef81d7cdce2d9ec80e5 Mon Sep 17 00:00:00 2001 From: Quick <31828688+Quick104@users.noreply.github.com> Date: Sun, 2 Aug 2026 20:53:21 -0400 Subject: [PATCH] feat(metadata): user-triggered trailer refresh with weekly per-item cooldown (#531) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(metadata): user-triggered trailer refresh with weekly per-item cooldown Adds POST /api/v1/items/{id}/trailers/refresh so any viewer with access to a movie or series can ask the server to fetch its remote trailers, bounded by a one-week per-item cooldown enforced server-side. The cooldown lives in a new nullable media_items.trailers_refresh_requested_at column rather than the refresh debt queue, whose last_attempt_at evaporates on success (MarkTargetSuccess deletes the row when the reason mask clears). The gate is a single UPDATE that writes NOW() only when the stored timestamp is NULL or older than the window, so concurrent viewers cannot both win it; a losing caller reads the stored timestamp back to compute next_allowed_at. MetadataService.RequestTrailersRefresh resolves the per-library trailer_kinds allow-list first: a non-nil empty map means every containing library disabled remote videos, which answers "disabled" without consuming the cooldown slot (a nil map is allow-all and must not short-circuit). On winning the gate it reuses startOnDemandMetadataRefresh, whose scheduled mode merges fill-empty, so this non-admin trigger cannot clobber unlocked admin edits while found videos still persist. The handler checks item access before calling the service, so an unauthorized caller can never burn an item's slot, and rejects non movie/series types since those detail responses never carry videos. cooldown and disabled are expected client-rendered states and answer 200; 429 is reserved for the per-user in-memory limiter. Co-Authored-By: Claude Fable 5 * fix(metadata): release trailer-refresh slot on failed refresh; resolve episode ids to 400 Three review findings on the viewer-facing trailer fetch. The weekly per-item slot was consumed unconditionally on winning the gate, but the refresh it started ran detached and only logged on failure — nothing ever put the slot back. A brief TMDb outage therefore answered 202 queued, failed 30s later, and then answered cooldown for seven days over work that never happened. The repository gains an equality-guarded release (trailers_refresh_requested_at = NULL only while it still equals the timestamp this request wrote, so a later claim is never clobbered), and TryClaimTrailersRefresh now RETURNINGs the timestamp it stored so a winner holds the key to its own slot. startOnDemandMetadataRefresh splits into a claim step and runOnDemandMetadataRefresh, which takes an optional failure hook; only the trailer path passes one, so the existing callers are unchanged. A timeout counts as failure. A refresh that succeeds but finds nothing still keeps the slot — that semantics was chosen deliberately. The in-process dedup claim (shared with the item-detail view's stale nudge) silently dropped the start while the slot had already been consumed, so the caller was told queued for a refresh that never began. It is now taken before the durable slot: a request landing while an equivalent refresh is already in flight reports queued without consuming the slot, which is both honest and retryable if that refresh fails. Real episode and season content IDs answered 404 rather than the contracted 400, because neither is a media_items row and GetByID queries media_items alone. The handler now falls through to the same season/episode lookups HandleTranslateOnView uses, authorizing through the parent series, so a genuine episode ID reports unsupported-type and only unknown content 404s. The type-check test no longer fabricates a MediaItem{Type: "episode"} row that production never writes; it covers the types that do exist as media_items rows, with the episode and season paths tested through the lookups. Co-Authored-By: Claude Fable 5 * fix(metadata): address PR review on the trailer refresh action Six review findings on the viewer-facing trailer fetch, all verified against the current code before changing anything. Durable claim no longer rides the request context. A cancellation landing after Postgres commits the gate UPDATE but before pgx returns would consume the item's weekly slot with no refresh started and nothing holding the timestamp needed to release it. The claim now runs on context.WithoutCancel with its own deadline, mirroring the release. The cooldown gate retries once when the follow-up read finds the slot free. Classification spans two statements, so a concurrent failure-release can land between them; the old code reported that as a cooldown with no next_allowed_at while the slot was in fact free. A NULL read now retries the claim, and the doubly-lost case answers "queued" (an equivalent refresh is running) rather than an undateable cooldown. A failed item_videos write now releases the slot. mergeAndPersist logs and continues when the write fails, so the refresh reported success and the viewer was locked out for a week having stored nothing. A context-scoped observer, installed only by this action, surfaces that failure to the existing release hook. Winning the gate also records durable refresh debt, so a restart that kills the detached goroutine leaves work the refresh worker picks up instead of a consumed slot and no fetch. Uses a new reason bit rather than the generic failure reason: nothing is wrong with the item, so it must not sit in the failure band ahead of real debt or count as a failure in operator metrics. Any library lookup failure now degrades the video-kind scope to unknown. An item in two libraries where one resolved with trailers off and the other could not be read reported "disabled" — a guess made on behalf of a library that might be the one enabling trailers. A library that is genuinely gone is still skipped. Adds GET /api/v1/items/trailers/capability, following the existing per-subsystem probe convention. The action route is registered conditionally, so "this build has the feature" is not the same question as "this deployment serves it", and a 404 on the POST is indistinguishable from a missing item. The probe is registered unconditionally and answers refresh:false when unwired. Not changed: content-ID canonicalization mid-refresh stranding the cooldown on the old row. The re-anchor path is manual-refresh only and this action runs in scheduled mode, so only local-skeleton promotion can fire, and the rename carries the timestamp and the debt row to the new id along with everything else. Co-Authored-By: Claude Fable 5 * style(web): format overlays schema after merging main The line came in over-length from main's card_overlays merge and the Web CI format check runs prettier across all of src, not just changed files. Co-Authored-By: Claude Fable 5 * fix(metadata): address second review round — recovery-debt lease, locked-videos preflight, shared limiter The restart mitigation added in the first round reintroduced two of the problems it was closing, and the reviewer was right to push again. Lease the recovery debt behind the fast path. The row was enqueued due now, so refresh_metadata could claim it while the detached goroutine was still running the same refresh — RefreshScheduledTarget does not consult the in-process claim, so both would fetch the item at once. It is now due 5 minutes out, comfortably past the 2-minute on-demand timeout, and the goroutine settles the row on success so it fires only when the fast path really did not finish. Settling clears just the trailers-requested bit, keeping any real debt the item still carries. Release the cooldown after a failed recovery. A recovery runs in a worker that never saw the claim, so a failure left the viewer blocked for the week having stored nothing. RefreshScheduledTarget now adopts the claim when the debt row carries the trailers-requested reason, reading the stored timestamp so the release stays equality-guarded, and hands the slot back on the same failures the fast path's hook covers — including a videos write that failed and was only logged. Preflight the videos lock. locked_fields containing FieldVideos makes mergeAndPersist skip the item_videos write, so the refresh "succeeded" and kept the cooldown while never being able to save trailers. It now answers disabled before consuming the claim; reusing that status rather than adding one is deliberate, since clients treat an unknown status as a dead end and "trailers cannot be fetched for this item" is what disabled already means to a viewer. Use the shared limiter. A private MemoryLimiter gave every instance an independent per-user allowance on Redis deployments, and the per-item cooldown cannot compensate — it bounds one item, while this budget bounds how many distinct items a user can start refreshes for. The action now takes the middleware's configured limiter, with namespaced keys, and falls back to a private one only when rate limiting is off. Co-Authored-By: Claude Fable 5 --------- Co-authored-by: Claude Fable 5 --- internal/api/handlers/items.go | 306 +++++ internal/api/handlers/items_trailers_test.go | 571 +++++++++ internal/api/router.go | 27 + internal/catalog/item_repo.go | 121 ++ .../item_repo_trailers_refresh_db_test.go | 378 ++++++ internal/metadata/refresh_debt.go | 8 + internal/metadata/refresh_debt_repo.go | 1 + internal/metadata/service.go | 510 ++++++++- internal/metadata/service_test.go | 172 ++- internal/metadata/trailers_refresh_test.go | 1020 +++++++++++++++++ internal/ratelimit/middleware.go | 17 + ...ia_items_trailers_refresh_requested_at.sql | 12 + web/src/lib/overlays/schema.ts | 5 +- web/src/pages/AdminTaskDetail.tsx | 1 + web/src/pages/AdminTasks.tsx | 1 + 15 files changed, 3140 insertions(+), 10 deletions(-) create mode 100644 internal/api/handlers/items_trailers_test.go create mode 100644 internal/catalog/item_repo_trailers_refresh_db_test.go create mode 100644 internal/metadata/trailers_refresh_test.go create mode 100644 migrations/sql/20260802055513_media_items_trailers_refresh_requested_at.sql diff --git a/internal/api/handlers/items.go b/internal/api/handlers/items.go index 4c8047f7..c9d8fe4d 100644 --- a/internal/api/handlers/items.go +++ b/internal/api/handlers/items.go @@ -21,6 +21,7 @@ import ( "github.com/Silo-Server/silo-server/internal/metadata" "github.com/Silo-Server/silo-server/internal/models" "github.com/Silo-Server/silo-server/internal/overlays" + "github.com/Silo-Server/silo-server/internal/ratelimit" "github.com/Silo-Server/silo-server/internal/sections" "github.com/Silo-Server/silo-server/internal/userstore" "github.com/Silo-Server/silo-server/internal/watchstate" @@ -47,6 +48,33 @@ type MetadataRefreshRequester interface { RequestStaleMetadataRefresh(ctx context.Context, targetType, contentID string) error } +// TrailerRefreshRequester starts a viewer-triggered trailer fetch for one item +// and reports whether it was queued, in cooldown, or disabled by every +// containing library. Implemented by *metadata.MetadataService. +type TrailerRefreshRequester interface { + RequestTrailersRefresh(ctx context.Context, contentID string) (metadata.TrailerRefreshOutcome, error) +} + +// trailerItemAccess resolves and authorizes the item behind the trailer +// refresh route. The concrete *catalog.ItemRepository satisfies it; the +// interface keeps the handler testable without a database. +type trailerItemAccess interface { + GetByID(ctx context.Context, contentID string) (*models.MediaItem, error) + EnsureAccessible(ctx context.Context, contentID string, filter catalog.AccessFilter) error +} + +// trailerSeasonLookup and trailerEpisodeLookup resolve the content IDs that are +// not media_items rows. Season and episode detail pages carry their own IDs, so +// without these a client that asks for their trailers would get a 404 that +// looks like a missing item instead of the contracted "wrong type" answer. +type trailerSeasonLookup interface { + GetByID(ctx context.Context, contentID string) (*models.Season, error) +} + +type trailerEpisodeLookup interface { + GetByID(ctx context.Context, contentID string) (*models.Episode, error) +} + type LocalWatchEventDispatcher interface { HandleLocalWatchEvent(ctx context.Context, event watchsync.LocalWatchEvent) error } @@ -70,6 +98,11 @@ type ItemsHandler struct { profileStaler ProfileStaler profileRefreshRequester ProfileRefreshRequester metadataRefreshRequester MetadataRefreshRequester + trailerRefreshRequester TrailerRefreshRequester + trailerItemAccess trailerItemAccess + trailerSeasonLookup trailerSeasonLookup + trailerEpisodeLookup trailerEpisodeLookup + trailerRefreshLimiter ratelimit.RateLimiter localWatchDispatcher LocalWatchEventDispatcher ebookProgressStore EbookReaderProgressLister ebookReadStateStore EbookReadStateStore @@ -130,6 +163,47 @@ func (h *ItemsHandler) SetMetadataRefreshRequester(requester MetadataRefreshRequ h.metadataRefreshRequester = requester } +// SetTrailerRefreshLimiter wires the process's configured rate limiter into the +// trailer fetch action, so the per-user budget is shared across instances when +// the deployment runs the Redis backend. A private in-memory limiter would give +// each instance its own allowance for the same user, and the per-item database +// cooldown cannot make up the difference — it bounds one item, while this +// budget bounds how many distinct items a user can start refreshes for. +// +// Call before SetTrailerRefreshRequester, which falls back to a private +// in-memory limiter when none is set (single-instance deployments, and any +// deployment with rate limiting turned off entirely). +func (h *ItemsHandler) SetTrailerRefreshLimiter(limiter ratelimit.RateLimiter) { + if h == nil || limiter == nil { + return + } + h.trailerRefreshLimiter = limiter +} + +// SetTrailerRefreshRequester wires the viewer-facing trailer fetch action. +// Leaving it unset disables the route's behavior (503), so the router only +// registers it when the metadata service is available. +func (h *ItemsHandler) SetTrailerRefreshRequester(requester TrailerRefreshRequester) { + if h == nil { + return + } + h.trailerRefreshRequester = requester + if h.trailerRefreshLimiter == nil { + h.trailerRefreshLimiter = ratelimit.NewMemoryLimiter() + } + if h.trailerItemAccess == nil && h.itemRepo != nil { + h.trailerItemAccess = h.itemRepo + } + // Seasons and episodes are not media_items rows, so the route needs these + // to tell "this ID is an episode" from "no such content". + if h.trailerSeasonLookup == nil && h.seasonRepo != nil { + h.trailerSeasonLookup = h.seasonRepo + } + if h.trailerEpisodeLookup == nil && h.episodeRepo != nil { + h.trailerEpisodeLookup = h.episodeRepo + } +} + func (h *ItemsHandler) SetCatalogSearchProvider(provider catalog.CatalogSearchProvider) { if h == nil || h.catalogResolver == nil || provider == nil { return @@ -428,6 +502,238 @@ func (h *ItemsHandler) HandleGetWatchDetail(w http.ResponseWriter, r *http.Reque writeJSON(w, http.StatusOK, detail) } +// trailerRefreshRate bounds how often one user may trigger trailer fetches +// across all items. The per-item cooldown enforced by the metadata service is +// the real budget; this only keeps a misbehaving client from hammering the +// endpoint (same shape as personRefreshRate). +var trailerRefreshRate = ratelimit.Rate{ + RequestsPerSecond: 10, + RequestsPerMinute: 10, + Burst: 10, +} + +// trailerRefreshLimiterKey namespaces this action's per-user counter. The +// limiter behind it is normally the process-wide one shared with the rate-limit +// middleware, whose keys are namespaced the same way ("ip:", "key:"). +func trailerRefreshLimiterKey(userID int) string { + return "trailers:" + strconv.Itoa(userID) +} + +// trailerRefreshResponse is the body of the trailer refresh endpoint. +// NextAllowedAt is present only for the cooldown status. +type trailerRefreshResponse struct { + Status string `json:"status"` + NextAllowedAt string `json:"next_allowed_at,omitempty"` +} + +// trailerRefreshCapabilityResponse tells a client whether this server offers +// the viewer-facing trailer fetch, following the per-subsystem convention +// (/events/capability, /playback/capability, /ebooks/capability). +// +// Without it the only signal is a 404 from the POST, which a client cannot +// tell apart from a missing item — and the route is registered conditionally +// (it needs the metadata service to implement the optional interface), so +// "this build has the feature" is not the same question as "this deployment +// serves it". A client that finds refresh false should hide the action rather +// than offer a button that cannot work. +type trailerRefreshCapabilityResponse struct { + SchemaVersion int `json:"schema_version"` + // Refresh reports that POST /items/{id}/trailers/refresh is served here. + Refresh bool `json:"refresh"` + // CooldownSeconds is the per-item window between viewer-triggered + // refreshes, so a client can explain the wait without having received a + // cooldown response first. + CooldownSeconds int `json:"cooldown_seconds"` + // Statuses is every value the refresh endpoint's status field may take. + Statuses []string `json:"statuses"` + // SupportedTypes is the item types the action applies to; nothing else + // carries remote videos, so clients should not show the action elsewhere. + SupportedTypes []string `json:"supported_types"` +} + +// HandleTrailerRefreshCapability reports whether the trailer refresh action is +// available. GET /api/v1/items/trailers/capability. +// +// It answers even when the feature is unwired, because "refresh": false is the +// answer in that case; the router registers it unconditionally so a client +// never has to interpret a 404 on the probe itself. +func (h *ItemsHandler) HandleTrailerRefreshCapability(w http.ResponseWriter, _ *http.Request) { + enabled := h != nil && h.trailerRefreshRequester != nil && h.trailerItemAccess != nil + resp := trailerRefreshCapabilityResponse{ + SchemaVersion: 1, + Refresh: enabled, + Statuses: []string{}, + SupportedTypes: []string{}, + } + if enabled { + resp.CooldownSeconds = int(metadata.TrailerRefreshCooldown / time.Second) + resp.Statuses = []string{ + metadata.TrailerRefreshStatusQueued, + metadata.TrailerRefreshStatusCooldown, + metadata.TrailerRefreshStatusDisabled, + } + resp.SupportedTypes = []string{"movie", "series"} + } + writeJSON(w, http.StatusOK, resp) +} + +// HandleRequestTrailersRefresh handles POST /api/v1/items/{id}/trailers/refresh: +// any authenticated viewer with access to a movie or series may ask the server +// to fetch its remote trailers, at most once per item per cooldown window. +// +// "cooldown" and "disabled" are expected client-rendered states, not errors, +// so they answer 200; 429 stays reserved for the per-user limiter. The access +// check runs before the metadata service is called so a caller who cannot see +// the item can never consume its cooldown slot. A season or episode ID resolves +// through its own table to 400 unsupported-type; only genuinely unknown content +// answers 404. +func (h *ItemsHandler) HandleRequestTrailersRefresh(w http.ResponseWriter, r *http.Request) { + if h == nil || h.trailerRefreshRequester == nil || h.trailerItemAccess == nil { + writeError(w, http.StatusServiceUnavailable, "unavailable", "Trailer refresh is not configured") + return + } + + contentID := strings.TrimSpace(chi.URLParam(r, "id")) + if contentID == "" { + writeError(w, http.StatusBadRequest, "bad_request", "Item ID is required") + return + } + + userID := apimw.GetUserID(r.Context()) + if userID == 0 { + writeError(w, http.StatusUnauthorized, "unauthorized", "Authentication required") + return + } + + if h.trailerRefreshLimiter != nil { + // The limiter may be the process-wide one the middleware uses, so the + // key is namespaced: an unprefixed user id would share a counter with + // whatever else keys on the same string. + result := h.trailerRefreshLimiter.Allow(r.Context(), trailerRefreshLimiterKey(userID), trailerRefreshRate) + if !result.Allowed { + if result.RetryAfter > 0 { + w.Header().Set("Retry-After", strconv.Itoa(max(1, int(result.RetryAfter.Seconds())))) + } + writeError(w, http.StatusTooManyRequests, "rate_limited", "Too many trailer refresh requests") + return + } + } + + target, err := h.resolveTrailerRefreshTarget(r.Context(), contentID) + if err != nil { + if errors.Is(err, catalog.ErrItemNotFound) { + writeError(w, http.StatusNotFound, "not_found", "Item not found") + return + } + slog.ErrorContext(r.Context(), "trailers: failed to look up item", "component", "api", + "content_id", contentID, "error", err) + writeError(w, http.StatusInternalServerError, "internal_error", "Failed to authorize item") + return + } + // Authorize against the series for a season or episode ID, exactly as the + // on-view translation route does, so an unsupported-type answer never + // leaks the existence of content the caller cannot see. + if err := h.trailerItemAccess.EnsureAccessible(r.Context(), target.accessContentID, h.accessFilter(r)); err != nil { + if errors.Is(err, catalog.ErrItemNotFound) { + writeError(w, http.StatusNotFound, "not_found", "Item not found") + return + } + slog.ErrorContext(r.Context(), "trailers: failed to authorize item", "component", "api", + "content_id", contentID, "error", err) + writeError(w, http.StatusInternalServerError, "internal_error", "Failed to authorize item") + return + } + + // Only movie and series detail responses carry videos/extras, so anything + // else — another media_items type, or a season/episode ID, which is not a + // media_items row at all — is a client bug rather than an empty result. + if !target.supportsTrailers { + writeError(w, http.StatusBadRequest, "unsupported_type", "Trailers are only available for movies and series") + return + } + + outcome, err := h.trailerRefreshRequester.RequestTrailersRefresh(r.Context(), contentID) + if err != nil { + if errors.Is(err, catalog.ErrItemNotFound) { + writeError(w, http.StatusNotFound, "not_found", "Item not found") + return + } + slog.ErrorContext(r.Context(), "trailers: failed to request refresh", "component", "api", + "content_id", contentID, "error", err) + writeError(w, http.StatusInternalServerError, "internal_error", "Failed to request trailers") + return + } + + switch outcome.Status { + case metadata.TrailerRefreshStatusQueued: + writeJSON(w, http.StatusAccepted, trailerRefreshResponse{Status: outcome.Status}) + case metadata.TrailerRefreshStatusCooldown: + resp := trailerRefreshResponse{Status: outcome.Status} + if outcome.NextAllowedAt != nil { + resp.NextAllowedAt = outcome.NextAllowedAt.UTC().Format(time.RFC3339) + } + writeJSON(w, http.StatusOK, resp) + case metadata.TrailerRefreshStatusDisabled: + writeJSON(w, http.StatusOK, trailerRefreshResponse{Status: outcome.Status}) + default: + slog.ErrorContext(r.Context(), "trailers: unexpected refresh outcome", "component", "api", + "content_id", contentID, "status", outcome.Status) + writeError(w, http.StatusInternalServerError, "internal_error", "Failed to request trailers") + } +} + +// trailerRefreshTarget is what a content ID on the trailer refresh route turned +// out to be: whether trailers apply to it at all, and which item ID authorizes +// it (a season or episode is authorized through its series). +type trailerRefreshTarget struct { + supportsTrailers bool + accessContentID string +} + +// resolveTrailerRefreshTarget identifies the content behind an ID the same way +// the on-view translation route does. Seasons and episodes live in their own +// tables, so a media_items miss is not proof the content is absent — falling +// through to those lookups is what lets a real episode ID answer 400 +// unsupported-type instead of a misleading 404. +func (h *ItemsHandler) resolveTrailerRefreshTarget(ctx context.Context, contentID string) (trailerRefreshTarget, error) { + item, err := h.trailerItemAccess.GetByID(ctx, contentID) + switch { + case err == nil && item != nil: + return trailerRefreshTarget{ + supportsTrailers: item.Type == "movie" || item.Type == "series", + accessContentID: contentID, + }, nil + case err == nil, errors.Is(err, catalog.ErrItemNotFound): + // Fall through to the season and episode lookups. + default: + return trailerRefreshTarget{}, err + } + + if h.trailerSeasonLookup != nil { + season, err := h.trailerSeasonLookup.GetByID(ctx, contentID) + switch { + case err == nil && season != nil: + return trailerRefreshTarget{accessContentID: season.SeriesID}, nil + case err == nil, errors.Is(err, catalog.ErrSeasonNotFound): + default: + return trailerRefreshTarget{}, err + } + } + + if h.trailerEpisodeLookup != nil { + episode, err := h.trailerEpisodeLookup.GetByID(ctx, contentID) + switch { + case err == nil && episode != nil: + return trailerRefreshTarget{accessContentID: episode.SeriesID}, nil + case err == nil, errors.Is(err, catalog.ErrEpisodeNotFound): + default: + return trailerRefreshTarget{}, err + } + } + + return trailerRefreshTarget{}, catalog.ErrItemNotFound +} + // HandleMarkWatched handles POST /watched/{id}. func (h *ItemsHandler) HandleMarkWatched(w http.ResponseWriter, r *http.Request) { h.handleSetWatchedState(w, r, true) diff --git a/internal/api/handlers/items_trailers_test.go b/internal/api/handlers/items_trailers_test.go new file mode 100644 index 00000000..9b0e6a20 --- /dev/null +++ b/internal/api/handlers/items_trailers_test.go @@ -0,0 +1,571 @@ +package handlers + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/go-chi/chi/v5" + + "github.com/Silo-Server/silo-server/internal/access" + apimw "github.com/Silo-Server/silo-server/internal/api/middleware" + "github.com/Silo-Server/silo-server/internal/auth" + "github.com/Silo-Server/silo-server/internal/catalog" + "github.com/Silo-Server/silo-server/internal/metadata" + "github.com/Silo-Server/silo-server/internal/models" + "github.com/Silo-Server/silo-server/internal/ratelimit" +) + +type fakeTrailerItemAccess struct { + items map[string]*models.MediaItem + ensureErr map[string]error + getErr map[string]error + checked []string +} + +func (f *fakeTrailerItemAccess) GetByID(_ context.Context, contentID string) (*models.MediaItem, error) { + if err := f.getErr[contentID]; err != nil { + return nil, err + } + if item := f.items[contentID]; item != nil { + return item, nil + } + return nil, catalog.ErrItemNotFound +} + +func (f *fakeTrailerItemAccess) EnsureAccessible(_ context.Context, contentID string, _ catalog.AccessFilter) error { + f.checked = append(f.checked, contentID) + return f.ensureErr[contentID] +} + +type fakeTrailerRefreshRequester struct { + outcome metadata.TrailerRefreshOutcome + err error + requests []string +} + +func (f *fakeTrailerRefreshRequester) RequestTrailersRefresh(_ context.Context, contentID string) (metadata.TrailerRefreshOutcome, error) { + f.requests = append(f.requests, contentID) + if f.err != nil { + return metadata.TrailerRefreshOutcome{}, f.err + } + return f.outcome, nil +} + +// fakeTrailerSeasonLookup and fakeTrailerEpisodeLookup stand in for the season +// and episode tables. Their content IDs are real and resolvable, they are just +// not media_items rows — which is exactly why the route needs them. +type fakeTrailerSeasonLookup map[string]*models.Season + +func (f fakeTrailerSeasonLookup) GetByID(_ context.Context, contentID string) (*models.Season, error) { + if season := f[contentID]; season != nil { + return season, nil + } + return nil, catalog.ErrSeasonNotFound +} + +type fakeTrailerEpisodeLookup map[string]*models.Episode + +func (f fakeTrailerEpisodeLookup) GetByID(_ context.Context, contentID string) (*models.Episode, error) { + if episode := f[contentID]; episode != nil { + return episode, nil + } + return nil, catalog.ErrEpisodeNotFound +} + +func newTrailerRefreshHandler( + access *fakeTrailerItemAccess, + requester *fakeTrailerRefreshRequester, +) *ItemsHandler { + return &ItemsHandler{ + trailerItemAccess: access, + trailerRefreshRequester: requester, + trailerRefreshLimiter: ratelimit.NewMemoryLimiter(), + trailerSeasonLookup: fakeTrailerSeasonLookup{}, + trailerEpisodeLookup: fakeTrailerEpisodeLookup{}, + } +} + +func newTrailerRefreshRequest(contentID string, userID int) *http.Request { + req := httptest.NewRequest(http.MethodPost, "/api/v1/items/"+contentID+"/trailers/refresh", nil) + routeCtx := chi.NewRouteContext() + routeCtx.URLParams.Add("id", contentID) + ctx := context.WithValue(req.Context(), chi.RouteCtxKey, routeCtx) + ctx = apimw.SetClaims(ctx, &auth.Claims{UserID: userID, Role: "user", TokenType: auth.TokenTypeAccess}) + ctx = apimw.SetProfileID(ctx, "profile-1") + ctx = access.SetScope(ctx, access.Scope{UserID: userID, ProfileID: "profile-1"}) + return req.WithContext(ctx) +} + +func decodeTrailerResponse(t *testing.T, rr *httptest.ResponseRecorder) map[string]any { + t.Helper() + var body map[string]any + if err := json.Unmarshal(rr.Body.Bytes(), &body); err != nil { + t.Fatalf("decode body %q: %v", rr.Body.String(), err) + } + return body +} + +// The router discovers both seams by type assertion, so a signature drift +// would silently unregister the route rather than fail the build. +func TestTrailerRefreshWiringAssertionsHold(t *testing.T) { + var svc any = (*metadata.MetadataService)(nil) + if _, ok := svc.(TrailerRefreshRequester); !ok { + t.Fatal("*metadata.MetadataService must satisfy handlers.TrailerRefreshRequester") + } + var repo any = (*catalog.ItemRepository)(nil) + if _, ok := repo.(trailerItemAccess); !ok { + t.Fatal("*catalog.ItemRepository must satisfy trailerItemAccess") + } + // SetTrailerRefreshRequester adopts these from the handler's own repos, so + // drift here would silently downgrade every episode ID back to a 404. + var seasons any = (*catalog.SeasonRepository)(nil) + if _, ok := seasons.(trailerSeasonLookup); !ok { + t.Fatal("*catalog.SeasonRepository must satisfy trailerSeasonLookup") + } + var episodes any = (*catalog.EpisodeRepository)(nil) + if _, ok := episodes.(trailerEpisodeLookup); !ok { + t.Fatal("*catalog.EpisodeRepository must satisfy trailerEpisodeLookup") + } +} + +func TestTrailersRefreshReturnsQueued(t *testing.T) { + itemAccess := &fakeTrailerItemAccess{ + items: map[string]*models.MediaItem{"movie-1": {ContentID: "movie-1", Type: "movie"}}, + ensureErr: map[string]error{}, + } + requester := &fakeTrailerRefreshRequester{ + outcome: metadata.TrailerRefreshOutcome{Status: metadata.TrailerRefreshStatusQueued}, + } + handler := newTrailerRefreshHandler(itemAccess, requester) + + rr := httptest.NewRecorder() + handler.HandleRequestTrailersRefresh(rr, newTrailerRefreshRequest("movie-1", 7)) + + if rr.Code != http.StatusAccepted { + t.Fatalf("status = %d, want %d (%s)", rr.Code, http.StatusAccepted, rr.Body.String()) + } + body := decodeTrailerResponse(t, rr) + if body["status"] != "queued" { + t.Fatalf("status field = %v, want queued", body["status"]) + } + if _, ok := body["next_allowed_at"]; ok { + t.Fatalf("queued response must omit next_allowed_at, got %v", body) + } + if len(requester.requests) != 1 || requester.requests[0] != "movie-1" { + t.Fatalf("requests = %v, want [movie-1]", requester.requests) + } +} + +func TestTrailersRefreshReturnsCooldownWithNextAllowedAt(t *testing.T) { + next := time.Date(2026, 8, 9, 12, 0, 0, 0, time.UTC) + itemAccess := &fakeTrailerItemAccess{ + items: map[string]*models.MediaItem{"series-1": {ContentID: "series-1", Type: "series"}}, + ensureErr: map[string]error{}, + } + requester := &fakeTrailerRefreshRequester{ + outcome: metadata.TrailerRefreshOutcome{ + Status: metadata.TrailerRefreshStatusCooldown, + NextAllowedAt: &next, + }, + } + handler := newTrailerRefreshHandler(itemAccess, requester) + + rr := httptest.NewRecorder() + handler.HandleRequestTrailersRefresh(rr, newTrailerRefreshRequest("series-1", 7)) + + // Cooldown is an expected client-rendered state, not an error: 200, and + // 429 stays reserved for the per-user limiter. + if rr.Code != http.StatusOK { + t.Fatalf("status = %d, want %d (%s)", rr.Code, http.StatusOK, rr.Body.String()) + } + body := decodeTrailerResponse(t, rr) + if body["status"] != "cooldown" { + t.Fatalf("status field = %v, want cooldown", body["status"]) + } + if got := body["next_allowed_at"]; got != next.Format(time.RFC3339) { + t.Fatalf("next_allowed_at = %v, want %s", got, next.Format(time.RFC3339)) + } +} + +func TestTrailersRefreshReturnsDisabled(t *testing.T) { + itemAccess := &fakeTrailerItemAccess{ + items: map[string]*models.MediaItem{"movie-1": {ContentID: "movie-1", Type: "movie"}}, + ensureErr: map[string]error{}, + } + requester := &fakeTrailerRefreshRequester{ + outcome: metadata.TrailerRefreshOutcome{Status: metadata.TrailerRefreshStatusDisabled}, + } + handler := newTrailerRefreshHandler(itemAccess, requester) + + rr := httptest.NewRecorder() + handler.HandleRequestTrailersRefresh(rr, newTrailerRefreshRequest("movie-1", 7)) + + if rr.Code != http.StatusOK { + t.Fatalf("status = %d, want %d (%s)", rr.Code, http.StatusOK, rr.Body.String()) + } + body := decodeTrailerResponse(t, rr) + if body["status"] != "disabled" { + t.Fatalf("status field = %v, want disabled", body["status"]) + } + if _, ok := body["next_allowed_at"]; ok { + t.Fatalf("disabled response must omit next_allowed_at, got %v", body) + } +} + +// Only movie and series detail responses carry videos, so any other +// media_items type is a client bug rather than an empty result. These are the +// types that actually exist as media_items rows; episodes and seasons live in +// their own tables and are covered separately below. +func TestTrailersRefreshRejectsNonMovieSeriesTypes(t *testing.T) { + for _, itemType := range []string{"audiobook", "ebook", "manga"} { + t.Run(itemType, func(t *testing.T) { + itemAccess := &fakeTrailerItemAccess{ + items: map[string]*models.MediaItem{"item-1": {ContentID: "item-1", Type: itemType}}, + ensureErr: map[string]error{}, + } + requester := &fakeTrailerRefreshRequester{} + handler := newTrailerRefreshHandler(itemAccess, requester) + + rr := httptest.NewRecorder() + handler.HandleRequestTrailersRefresh(rr, newTrailerRefreshRequest("item-1", 7)) + + if rr.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want %d (%s)", rr.Code, http.StatusBadRequest, rr.Body.String()) + } + if len(requester.requests) != 0 { + t.Fatalf("unsupported type must not reach the service, got %v", requester.requests) + } + }) + } +} + +// Episodes and seasons are not media_items rows, so the item lookup misses on +// their real content IDs. Without the fallbacks the route would answer 404 +// "Item not found" for content that plainly exists; the contract is 400 +// unsupported-type. Authorization runs against the parent series, as on the +// on-view translation route. +func TestTrailersRefreshRejectsEpisodeAndSeasonIDsWith400(t *testing.T) { + tests := []struct { + name string + contentID string + wantAccess string + }{ + {name: "episode", contentID: "episode-1", wantAccess: "series-1"}, + {name: "season", contentID: "season-1", wantAccess: "series-1"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + itemAccess := &fakeTrailerItemAccess{ + items: map[string]*models.MediaItem{"series-1": {ContentID: "series-1", Type: "series"}}, + ensureErr: map[string]error{}, + } + requester := &fakeTrailerRefreshRequester{} + handler := newTrailerRefreshHandler(itemAccess, requester) + handler.trailerSeasonLookup = fakeTrailerSeasonLookup{ + "season-1": {ContentID: "season-1", SeriesID: "series-1"}, + } + handler.trailerEpisodeLookup = fakeTrailerEpisodeLookup{ + "episode-1": {ContentID: "episode-1", SeriesID: "series-1"}, + } + + rr := httptest.NewRecorder() + handler.HandleRequestTrailersRefresh(rr, newTrailerRefreshRequest(tc.contentID, 7)) + + if rr.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want %d (%s)", rr.Code, http.StatusBadRequest, rr.Body.String()) + } + body := decodeTrailerResponse(t, rr) + if code, _ := body["error"].(string); code != "unsupported_type" { + t.Fatalf("error code = %v, want unsupported_type (%s)", body["error"], rr.Body.String()) + } + if len(itemAccess.checked) != 1 || itemAccess.checked[0] != tc.wantAccess { + t.Fatalf("access checks = %v, want [%s]", itemAccess.checked, tc.wantAccess) + } + if len(requester.requests) != 0 { + t.Fatalf("unsupported type must not reach the service, got %v", requester.requests) + } + }) + } +} + +// An episode inside a series the caller cannot see must not be distinguishable +// from content that does not exist, so the access check runs before the type +// answer. +func TestTrailersRefreshEpisodeInInaccessibleSeriesReturns404(t *testing.T) { + itemAccess := &fakeTrailerItemAccess{ + items: map[string]*models.MediaItem{"series-1": {ContentID: "series-1", Type: "series"}}, + ensureErr: map[string]error{"series-1": catalog.ErrItemNotFound}, + } + requester := &fakeTrailerRefreshRequester{} + handler := newTrailerRefreshHandler(itemAccess, requester) + handler.trailerEpisodeLookup = fakeTrailerEpisodeLookup{ + "episode-1": {ContentID: "episode-1", SeriesID: "series-1"}, + } + + rr := httptest.NewRecorder() + handler.HandleRequestTrailersRefresh(rr, newTrailerRefreshRequest("episode-1", 7)) + + if rr.Code != http.StatusNotFound { + t.Fatalf("status = %d, want %d (%s)", rr.Code, http.StatusNotFound, rr.Body.String()) + } + if len(requester.requests) != 0 { + t.Fatalf("denied request must not reach the service, got %v", requester.requests) + } +} + +// An unauthorized caller must be turned away before the metadata service is +// asked, so it can never burn the item's cooldown slot. +func TestTrailersRefreshDeniedAccessReturns404WithoutConsumingCooldown(t *testing.T) { + itemAccess := &fakeTrailerItemAccess{ + items: map[string]*models.MediaItem{"movie-1": {ContentID: "movie-1", Type: "movie"}}, + ensureErr: map[string]error{"movie-1": catalog.ErrItemNotFound}, + } + requester := &fakeTrailerRefreshRequester{} + handler := newTrailerRefreshHandler(itemAccess, requester) + + rr := httptest.NewRecorder() + handler.HandleRequestTrailersRefresh(rr, newTrailerRefreshRequest("movie-1", 7)) + + if rr.Code != http.StatusNotFound { + t.Fatalf("status = %d, want %d (%s)", rr.Code, http.StatusNotFound, rr.Body.String()) + } + if len(itemAccess.checked) != 1 { + t.Fatalf("access checks = %v, want one check", itemAccess.checked) + } + if len(requester.requests) != 0 { + t.Fatalf("denied request must not reach the service, got %v", requester.requests) + } +} + +func TestTrailersRefreshMissingItemReturns404(t *testing.T) { + itemAccess := &fakeTrailerItemAccess{ + items: map[string]*models.MediaItem{}, + ensureErr: map[string]error{}, + } + requester := &fakeTrailerRefreshRequester{} + handler := newTrailerRefreshHandler(itemAccess, requester) + + rr := httptest.NewRecorder() + handler.HandleRequestTrailersRefresh(rr, newTrailerRefreshRequest("missing", 7)) + + if rr.Code != http.StatusNotFound { + t.Fatalf("status = %d, want %d (%s)", rr.Code, http.StatusNotFound, rr.Body.String()) + } + if len(requester.requests) != 0 { + t.Fatalf("missing item must not reach the service, got %v", requester.requests) + } +} + +func TestTrailersRefreshRequiresAuthentication(t *testing.T) { + itemAccess := &fakeTrailerItemAccess{ + items: map[string]*models.MediaItem{"movie-1": {ContentID: "movie-1", Type: "movie"}}, + ensureErr: map[string]error{}, + } + requester := &fakeTrailerRefreshRequester{} + handler := newTrailerRefreshHandler(itemAccess, requester) + + req := httptest.NewRequest(http.MethodPost, "/api/v1/items/movie-1/trailers/refresh", nil) + routeCtx := chi.NewRouteContext() + routeCtx.URLParams.Add("id", "movie-1") + req = req.WithContext(context.WithValue(req.Context(), chi.RouteCtxKey, routeCtx)) + + rr := httptest.NewRecorder() + handler.HandleRequestTrailersRefresh(rr, req) + + if rr.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want %d (%s)", rr.Code, http.StatusUnauthorized, rr.Body.String()) + } + if len(requester.requests) != 0 { + t.Fatalf("unauthenticated request must not reach the service, got %v", requester.requests) + } +} + +// The per-user limiter is the abuse guard in front of the per-item cooldown: +// once a user exhausts the burst it answers 429 with Retry-After. +func TestTrailersRefreshRateLimitsPerUser(t *testing.T) { + itemAccess := &fakeTrailerItemAccess{ + items: map[string]*models.MediaItem{"movie-1": {ContentID: "movie-1", Type: "movie"}}, + ensureErr: map[string]error{}, + } + requester := &fakeTrailerRefreshRequester{ + outcome: metadata.TrailerRefreshOutcome{Status: metadata.TrailerRefreshStatusQueued}, + } + handler := newTrailerRefreshHandler(itemAccess, requester) + + limited := false + for i := 0; i < int(trailerRefreshRate.RequestsPerMinute)+5; i++ { + rr := httptest.NewRecorder() + handler.HandleRequestTrailersRefresh(rr, newTrailerRefreshRequest("movie-1", 7)) + if rr.Code == http.StatusTooManyRequests { + limited = true + if rr.Header().Get("Retry-After") == "" { + t.Fatal("429 response must carry Retry-After") + } + break + } + } + if !limited { + t.Fatal("expected the per-user limiter to reject a burst of requests") + } + + // A different user is unaffected — the limiter keys on the user id. + rr := httptest.NewRecorder() + handler.HandleRequestTrailersRefresh(rr, newTrailerRefreshRequest("movie-1", 8)) + if rr.Code != http.StatusAccepted { + t.Fatalf("second user status = %d, want %d (%s)", rr.Code, http.StatusAccepted, rr.Body.String()) + } +} + +func TestTrailersRefreshUnconfiguredReturns503(t *testing.T) { + handler := &ItemsHandler{} + + rr := httptest.NewRecorder() + handler.HandleRequestTrailersRefresh(rr, newTrailerRefreshRequest("movie-1", 7)) + + if rr.Code != http.StatusServiceUnavailable { + t.Fatalf("status = %d, want %d (%s)", rr.Code, http.StatusServiceUnavailable, rr.Body.String()) + } +} + +func TestTrailersRefreshServiceErrorReturns500(t *testing.T) { + itemAccess := &fakeTrailerItemAccess{ + items: map[string]*models.MediaItem{"movie-1": {ContentID: "movie-1", Type: "movie"}}, + ensureErr: map[string]error{}, + } + requester := &fakeTrailerRefreshRequester{err: errors.New("database is down")} + handler := newTrailerRefreshHandler(itemAccess, requester) + + rr := httptest.NewRecorder() + handler.HandleRequestTrailersRefresh(rr, newTrailerRefreshRequest("movie-1", 7)) + + if rr.Code != http.StatusInternalServerError { + t.Fatalf("status = %d, want %d (%s)", rr.Code, http.StatusInternalServerError, rr.Body.String()) + } +} + +// The capability probe is what lets a client tell "this server does not have +// the trailer action" from "that item does not exist", so it must answer on +// both a wired and an unwired handler. +func TestTrailerRefreshCapability(t *testing.T) { + t.Run("wired", func(t *testing.T) { + h := newTrailerRefreshHandler(&fakeTrailerItemAccess{}, &fakeTrailerRefreshRequester{}) + rr := httptest.NewRecorder() + h.HandleTrailerRefreshCapability(rr, httptest.NewRequest(http.MethodGet, "/items/trailers/capability", nil)) + if rr.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rr.Code) + } + body := decodeTrailerResponse(t, rr) + if body["refresh"] != true { + t.Fatalf("refresh = %v, want true", body["refresh"]) + } + if got, want := body["cooldown_seconds"], float64(metadata.TrailerRefreshCooldown/time.Second); got != want { + t.Fatalf("cooldown_seconds = %v, want %v", got, want) + } + // The advertised statuses are the contract the client switches on, so + // they must be the service's constants rather than a stale copy. + statuses, _ := body["statuses"].([]any) + want := []string{ + metadata.TrailerRefreshStatusQueued, + metadata.TrailerRefreshStatusCooldown, + metadata.TrailerRefreshStatusDisabled, + } + if len(statuses) != len(want) { + t.Fatalf("statuses = %v, want %v", statuses, want) + } + for i, status := range want { + if statuses[i] != status { + t.Fatalf("statuses[%d] = %v, want %q", i, statuses[i], status) + } + } + }) + + t.Run("unwired", func(t *testing.T) { + h := &ItemsHandler{} + rr := httptest.NewRecorder() + h.HandleTrailerRefreshCapability(rr, httptest.NewRequest(http.MethodGet, "/items/trailers/capability", nil)) + if rr.Code != http.StatusOK { + t.Fatalf("status = %d, want 200 — the probe itself must never 404", rr.Code) + } + body := decodeTrailerResponse(t, rr) + if body["refresh"] != false { + t.Fatalf("refresh = %v, want false", body["refresh"]) + } + }) +} + +// recordingLimiter captures the keys an action limiter is called with. +type recordingLimiter struct { + keys []string + allowed bool +} + +func (l *recordingLimiter) Allow(_ context.Context, key string, _ ratelimit.Rate) ratelimit.AllowResult { + l.keys = append(l.keys, key) + return ratelimit.AllowResult{Allowed: l.allowed, RetryAfter: time.Second} +} + +func (l *recordingLimiter) Close() {} + +// The action's budget must be enforced by the process's configured limiter, or +// a Redis deployment gives every instance an independent allowance for the same +// user and multiplies the stated budget by the instance count. The per-item +// database cooldown cannot compensate: it bounds one item, while this bounds +// how many distinct items a user can start refreshes for. +func TestTrailersRefreshUsesTheInjectedSharedLimiter(t *testing.T) { + itemAccess := &fakeTrailerItemAccess{ + items: map[string]*models.MediaItem{"movie-1": {ContentID: "movie-1", Type: "movie"}}, + ensureErr: map[string]error{}, + } + requester := &fakeTrailerRefreshRequester{ + outcome: metadata.TrailerRefreshOutcome{Status: metadata.TrailerRefreshStatusQueued}, + } + handler := newTrailerRefreshHandler(itemAccess, requester) + shared := &recordingLimiter{allowed: false} + handler.SetTrailerRefreshLimiter(shared) + // The requester wiring must not replace an injected limiter with a private + // in-memory one, which is the whole point of injecting it. + handler.SetTrailerRefreshRequester(requester) + + rr := httptest.NewRecorder() + handler.HandleRequestTrailersRefresh(rr, newTrailerRefreshRequest("movie-1", 7)) + + if rr.Code != http.StatusTooManyRequests { + t.Fatalf("status = %d, want %d — the injected limiter's verdict was ignored (%s)", + rr.Code, http.StatusTooManyRequests, rr.Body.String()) + } + if len(shared.keys) != 1 { + t.Fatalf("shared limiter consulted %d times, want 1", len(shared.keys)) + } + // The limiter may be the process-wide one, whose keyspace is shared with + // the rate-limit middleware ("ip:", "key:"), so this action's keys have to + // be namespaced too. + if shared.keys[0] != trailerRefreshLimiterKey(7) { + t.Fatalf("limiter key = %q, want the namespaced %q", shared.keys[0], trailerRefreshLimiterKey(7)) + } + if shared.keys[0] == "7" { + t.Fatal("an unprefixed user id would collide with other keyspaces in a shared limiter") + } + if len(requester.requests) != 0 { + t.Fatalf("a rate-limited request must not reach the service, got %v", requester.requests) + } +} + +// Rate limiting can be disabled outright (or the database unavailable), in +// which case there is no shared limiter to inject. The action keeps its own +// in-memory guard rather than running unbounded. +func TestTrailersRefreshFallsBackToAPrivateLimiter(t *testing.T) { + handler := &ItemsHandler{} + handler.SetTrailerRefreshLimiter(nil) + handler.SetTrailerRefreshRequester(&fakeTrailerRefreshRequester{ + outcome: metadata.TrailerRefreshOutcome{Status: metadata.TrailerRefreshStatusQueued}, + }) + + if handler.trailerRefreshLimiter == nil { + t.Fatal("the action must keep a limiter even when no shared one is configured") + } +} diff --git a/internal/api/router.go b/internal/api/router.go index 9bba5311..81681c3c 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -2508,6 +2508,33 @@ func NewRouter(deps Dependencies) chi.Router { r.Get("/metadata/ai/status", handlers.WriteMetadataAIDisabledStatus) } + // Viewer-facing trailer fetch. Registered beside the on-view + // translation trigger because it is the same shape: a + // non-admin, item-scoped metadata action guarded by item + // access plus a per-user limiter, with the real budget being + // the per-item cooldown the metadata service enforces. + // + // The action route is conditional (it needs the metadata + // service to implement the optional interface), so the + // capability probe beside it is not: per the v1 rules a client + // feature-detects rather than version-sniffs, and a probe that + // itself 404s would leave it interpreting the same ambiguous + // status it was meant to replace. Unwired, the probe answers + // refresh:false. + if itemsHandler != nil && itemRepo != nil { + if requester, ok := deps.MetadataService.(handlers.TrailerRefreshRequester); ok { + // Share the process's configured limiter so the + // per-user budget is one budget on Redis deployments + // rather than one per instance. Nil when rate limiting + // is disabled; the handler then keeps its private + // in-memory fallback. + itemsHandler.SetTrailerRefreshLimiter(deps.RateLimitMW.SharedLimiter()) + itemsHandler.SetTrailerRefreshRequester(requester) + r.Post("/items/{id}/trailers/refresh", itemsHandler.HandleRequestTrailersRefresh) + } + r.Get("/items/trailers/capability", itemsHandler.HandleTrailerRefreshCapability) + } + // Subtitle search + AI translation routes. if subtitleSearchHandler != nil { if deps.FileRepo != nil && itemRepo != nil { diff --git a/internal/catalog/item_repo.go b/internal/catalog/item_repo.go index 3be5392a..6602ee19 100644 --- a/internal/catalog/item_repo.go +++ b/internal/catalog/item_repo.go @@ -7,6 +7,7 @@ import ( "log/slog" "strconv" "strings" + "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" @@ -1875,6 +1876,126 @@ func (r *ItemRepository) IncrementRefreshFailure(ctx context.Context, contentID return nil } +// TryClaimTrailersRefresh atomically consumes an item's trailer-refresh +// cooldown slot. The UPDATE is the gate: it writes NOW() only when the stored +// timestamp is NULL or older than the cooldown window, so concurrent callers +// cannot both win. +// +// Either way the returned timestamp is the value now stored in the column. A +// winner needs it to release its own claim later (see +// ReleaseTrailersRefreshClaim); a loser needs it to compute the next-allowed +// time. Losing the gate means either the item is in cooldown or it no longer +// exists, which the follow-up read distinguishes — a missing row yields +// ErrItemNotFound. +// +// The classification spans two statements, so a concurrent release can land +// between them: the UPDATE loses to another request's claim, that request's +// refresh fails and NULLs the column, and the follow-up SELECT then reads a +// free slot. Reporting that as cooldown would be a cooldown with no +// next-allowed time and no refresh actually running, so a NULL read retries +// the claim once — the slot is demonstrably free, and this caller may take it. +// +// Contract for the three outcomes, which callers rely on to avoid emitting an +// undateable cooldown: +// - (true, ts, nil): claimed; ts is the stored timestamp to release on. +// - (false, ts, nil): in cooldown until ts plus the window. +// - (false, nil, nil): lost the gate twice while the slot kept being freed, +// so another request is claiming it right now. Not a cooldown — the caller +// should treat it as "an equivalent refresh is already in flight", the same +// answer it gives when it loses the in-process dedup claim. +func (r *ItemRepository) TryClaimTrailersRefresh(ctx context.Context, contentID string, cooldown time.Duration) (bool, *time.Time, error) { + const maxAttempts = 2 + for attempt := 0; attempt < maxAttempts; attempt++ { + var claimedAt time.Time + err := r.pool.QueryRow(ctx, ` + UPDATE media_items + SET trailers_refresh_requested_at = NOW() + WHERE content_id = $1 + AND (trailers_refresh_requested_at IS NULL + OR trailers_refresh_requested_at < NOW() - $2::interval) + RETURNING trailers_refresh_requested_at`, + contentID, fmt.Sprintf("%d seconds", int64(cooldown.Seconds())), + ).Scan(&claimedAt) + switch { + case err == nil: + return true, &claimedAt, nil + case !errors.Is(err, pgx.ErrNoRows): + return false, nil, fmt.Errorf("claiming trailers refresh: %w", err) + } + + var requestedAt *time.Time + err = r.pool.QueryRow(ctx, ` + SELECT trailers_refresh_requested_at + FROM media_items + WHERE content_id = $1`, + contentID, + ).Scan(&requestedAt) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return false, nil, ErrItemNotFound + } + return false, nil, fmt.Errorf("reading trailers refresh timestamp: %w", err) + } + if requestedAt != nil { + return false, requestedAt, nil + } + // The slot was released underneath us; go around and try to take it. + } + // Both attempts lost the gate and both read a freed slot. Rather than + // report a cooldown we cannot date, treat it as contention lost to whoever + // is claiming and releasing in a tight loop: no timestamp, no claim. + return false, nil, nil +} + +// ReleaseTrailersRefreshClaim hands an item's trailer-refresh cooldown slot +// back after the refresh it started failed, so the viewer can retry instead of +// waiting out the whole window for work that never happened. +// +// claimedAt is the timestamp TryClaimTrailersRefresh wrote, and the equality +// guard is what makes this safe to run from a detached goroutine: if the window +// has since lapsed and another request claimed the slot, this UPDATE matches no +// row and the newer claim survives untouched. Zero rows affected is therefore a +// normal outcome, not an error. +func (r *ItemRepository) ReleaseTrailersRefreshClaim(ctx context.Context, contentID string, claimedAt time.Time) error { + _, err := r.pool.Exec(ctx, ` + UPDATE media_items + SET trailers_refresh_requested_at = NULL + WHERE content_id = $1 + AND trailers_refresh_requested_at = $2`, + contentID, claimedAt, + ) + if err != nil { + return fmt.Errorf("releasing trailers refresh claim: %w", err) + } + return nil +} + +// TrailersRefreshRequestedAt reads the timestamp of the trailer-refresh +// cooldown claim currently held for an item, or nil when the slot is free. +// +// The durable recovery path needs it: when the process that consumed a slot +// dies mid-refresh, the refresh-debt queue re-runs the work in a worker that +// never saw the claim and so has no timestamp to release it on. Reading the +// claim before the refresh starts gives that worker the same exact key the +// original request had, so ReleaseTrailersRefreshClaim's equality guard keeps +// working: a slot re-claimed in the meantime is left alone. +func (r *ItemRepository) TrailersRefreshRequestedAt(ctx context.Context, contentID string) (*time.Time, error) { + var requestedAt *time.Time + err := r.pool.QueryRow(ctx, ` + SELECT trailers_refresh_requested_at + FROM media_items + WHERE content_id = $1`, + contentID, + ).Scan(&requestedAt) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrItemNotFound + } + return nil, fmt.Errorf("reading trailers refresh timestamp: %w", err) + } + return requestedAt, nil +} + // MediaTMDBRow is a single result row from LookupTMDBIDs, containing the // fields needed by the pluginhost CatalogPresence adapter. type MediaTMDBRow struct { diff --git a/internal/catalog/item_repo_trailers_refresh_db_test.go b/internal/catalog/item_repo_trailers_refresh_db_test.go new file mode 100644 index 00000000..6c14f478 --- /dev/null +++ b/internal/catalog/item_repo_trailers_refresh_db_test.go @@ -0,0 +1,378 @@ +package catalog + +import ( + "context" + "errors" + "fmt" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// TestTryClaimTrailersRefresh exercises the cooldown gate against a real +// database: the check-and-set is a single UPDATE precisely so two concurrent +// viewers cannot both win it, and that guarantee lives entirely in SQL — a +// fake cannot verify it. +func TestTryClaimTrailersRefresh(t *testing.T) { + 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) + + repo := NewItemRepository(pool) + contentID := fmt.Sprintf("trailer-claim-%d", time.Now().UnixNano()) + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM media_items WHERE content_id = $1`, contentID) + }) + if _, err := pool.Exec(ctx, ` + INSERT INTO media_items (content_id, type, title, status, genres) + VALUES ($1, 'movie', 'Trailer Claim', 'matched', '{}'::text[]) + `, contentID); err != nil { + t.Fatalf("seed item: %v", err) + } + + const cooldown = 7 * 24 * time.Hour + + claimed, requestedAt, err := repo.TryClaimTrailersRefresh(ctx, contentID, cooldown) + if err != nil { + t.Fatalf("first claim: %v", err) + } + if !claimed { + t.Fatal("first claim on a NULL timestamp must win") + } + // The winner gets the timestamp it wrote; it is the key its own release + // is guarded on. + if requestedAt == nil { + t.Fatal("winning claim must report the timestamp it stored") + } + if time.Since(*requestedAt) > time.Minute { + t.Fatalf("claimed timestamp = %s, want approximately now", requestedAt) + } + + claimed, requestedAt, err = repo.TryClaimTrailersRefresh(ctx, contentID, cooldown) + if err != nil { + t.Fatalf("second claim: %v", err) + } + if claimed { + t.Fatal("second claim inside the window must lose") + } + if requestedAt == nil { + t.Fatal("losing claim must report the stored timestamp for next-allowed math") + } + if time.Since(*requestedAt) > time.Minute { + t.Fatalf("stored timestamp = %s, want approximately now", requestedAt) + } + + // Backdating past the window reopens the gate. + if _, err := pool.Exec(ctx, ` + UPDATE media_items SET trailers_refresh_requested_at = NOW() - INTERVAL '8 days' + WHERE content_id = $1`, contentID); err != nil { + t.Fatalf("backdate timestamp: %v", err) + } + claimed, _, err = repo.TryClaimTrailersRefresh(ctx, contentID, cooldown) + if err != nil { + t.Fatalf("claim after cooldown lapsed: %v", err) + } + if !claimed { + t.Fatal("claim must win once the stored timestamp predates the window") + } + + // A missing item is distinguishable from a cooldown: the follow-up read + // finds no row. + _, _, err = repo.TryClaimTrailersRefresh(ctx, contentID+"-missing", cooldown) + if !errors.Is(err, ErrItemNotFound) { + t.Fatalf("missing item err = %v, want ErrItemNotFound", err) + } +} + +// TestReleaseTrailersRefreshClaim covers the failure path's half of the gate: +// a refresh that failed hands its slot back, and the equality guard keeps a +// late release from clearing a slot someone else has since claimed. Both live +// in SQL, so a fake cannot verify them. +func TestReleaseTrailersRefreshClaim(t *testing.T) { + 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) + + repo := NewItemRepository(pool) + contentID := fmt.Sprintf("trailer-release-%d", time.Now().UnixNano()) + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM media_items WHERE content_id = $1`, contentID) + }) + if _, err := pool.Exec(ctx, ` + INSERT INTO media_items (content_id, type, title, status, genres) + VALUES ($1, 'movie', 'Trailer Release', 'matched', '{}'::text[]) + `, contentID); err != nil { + t.Fatalf("seed item: %v", err) + } + + const cooldown = 7 * 24 * time.Hour + + storedAt := func(t *testing.T) *time.Time { + t.Helper() + var stored *time.Time + if err := pool.QueryRow(ctx, ` + SELECT trailers_refresh_requested_at FROM media_items WHERE content_id = $1`, + contentID, + ).Scan(&stored); err != nil { + t.Fatalf("read stored timestamp: %v", err) + } + return stored + } + + claimed, claimedAt, err := repo.TryClaimTrailersRefresh(ctx, contentID, cooldown) + if err != nil || !claimed || claimedAt == nil { + t.Fatalf("claim = %v, at = %v, err = %v", claimed, claimedAt, err) + } + if err := repo.ReleaseTrailersRefreshClaim(ctx, contentID, *claimedAt); err != nil { + t.Fatalf("release own claim: %v", err) + } + if stored := storedAt(t); stored != nil { + t.Fatalf("released slot still holds %s", stored) + } + // With the slot free the next request wins immediately, no clock movement. + claimed, claimedAt, err = repo.TryClaimTrailersRefresh(ctx, contentID, cooldown) + if err != nil || !claimed || claimedAt == nil { + t.Fatalf("claim after release = %v, at = %v, err = %v", claimed, claimedAt, err) + } + + // A release naming a timestamp the column no longer holds — the shape of a + // late release arriving after a newer request re-claimed the slot — is a + // no-op, not an error. + stale := claimedAt.Add(-time.Hour) + if err := repo.ReleaseTrailersRefreshClaim(ctx, contentID, stale); err != nil { + t.Fatalf("release with a stale timestamp: %v", err) + } + stored := storedAt(t) + if stored == nil { + t.Fatal("a stale release cleared a slot it does not own") + } + if !stored.Equal(*claimedAt) { + t.Fatalf("stored timestamp = %s, want the current claim %s", stored, claimedAt) + } + + // Releasing a row that no longer exists is likewise a no-op. + if err := repo.ReleaseTrailersRefreshClaim(ctx, contentID+"-missing", *claimedAt); err != nil { + t.Fatalf("release for a missing item: %v", err) + } +} + +// TestTryClaimTrailersRefreshIsAtomic runs concurrent claims against one item; +// exactly one may win. +func TestTryClaimTrailersRefreshIsAtomic(t *testing.T) { + 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) + + repo := NewItemRepository(pool) + contentID := fmt.Sprintf("trailer-claim-race-%d", time.Now().UnixNano()) + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM media_items WHERE content_id = $1`, contentID) + }) + if _, err := pool.Exec(ctx, ` + INSERT INTO media_items (content_id, type, title, status, genres) + VALUES ($1, 'movie', 'Trailer Claim Race', 'matched', '{}'::text[]) + `, contentID); err != nil { + t.Fatalf("seed item: %v", err) + } + + const workers = 8 + results := make(chan bool, workers) + errs := make(chan error, workers) + start := make(chan struct{}) + for i := 0; i < workers; i++ { + go func() { + <-start + claimed, _, err := repo.TryClaimTrailersRefresh(ctx, contentID, 7*24*time.Hour) + if err != nil { + errs <- err + return + } + results <- claimed + }() + } + close(start) + + wins := 0 + for i := 0; i < workers; i++ { + select { + case err := <-errs: + t.Fatalf("concurrent claim: %v", err) + case claimed := <-results: + if claimed { + wins++ + } + case <-time.After(10 * time.Second): + t.Fatal("timed out waiting for concurrent claims") + } + } + if wins != 1 { + t.Fatalf("concurrent claims won %d times, want exactly 1", wins) + } +} + +// TestTryClaimTrailersRefreshRetriesWhenSlotIsFreedMidClassification covers the +// window between the conditional UPDATE and the follow-up SELECT: a caller can +// lose the gate to another request and then have that request's refresh fail +// and clear the timestamp before the read. Classifying that as a cooldown would +// report one with no next-allowed time while the slot is in fact free, so the +// claim is retried and this caller takes it. +func TestTryClaimTrailersRefreshRetriesWhenSlotIsFreedMidClassification(t *testing.T) { + 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) + + repo := NewItemRepository(pool) + contentID := fmt.Sprintf("trailer-claim-retry-%d", time.Now().UnixNano()) + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM media_items WHERE content_id = $1`, contentID) + }) + if _, err := pool.Exec(ctx, ` + INSERT INTO media_items (content_id, type, title, status, genres) + VALUES ($1, 'movie', 'Trailer Claim Retry', 'matched', '{}'::text[]) + `, contentID); err != nil { + t.Fatalf("seed item: %v", err) + } + + const cooldown = 7 * 24 * time.Hour + + // Another request owns the slot, so the first UPDATE below loses. + claimed, claimedAt, err := repo.TryClaimTrailersRefresh(ctx, contentID, cooldown) + if err != nil || !claimed || claimedAt == nil { + t.Fatalf("seed claim = %v, at = %v, err = %v", claimed, claimedAt, err) + } + + // That request's refresh fails and hands the slot back. Doing it here + // models the release landing between our lost UPDATE and our follow-up + // read: either way the read observes NULL, which is the state under test. + if err := repo.ReleaseTrailersRefreshClaim(ctx, contentID, *claimedAt); err != nil { + t.Fatalf("release the competing claim: %v", err) + } + + claimed, requestedAt, err := repo.TryClaimTrailersRefresh(ctx, contentID, cooldown) + if err != nil { + t.Fatalf("claim after the competing release: %v", err) + } + if !claimed { + t.Fatalf("a freed slot must be claimable, got claimed=false requestedAt=%v", requestedAt) + } + if requestedAt == nil { + t.Fatal("a winning claim must report the timestamp it stored") + } + + // And the state really is a claim, not a phantom: the next request is in + // cooldown against the timestamp we just wrote. + claimed, requestedAt, err = repo.TryClaimTrailersRefresh(ctx, contentID, cooldown) + if err != nil { + t.Fatalf("claim after the retry won: %v", err) + } + if claimed { + t.Fatal("the slot must be held after the retry won it") + } + if requestedAt == nil { + t.Fatal("a cooldown must be dateable") + } +} + +// TestTrailersRefreshRequestedAt covers the read the durable recovery path uses +// to inherit a claim: a worker recovering a request whose process died never +// saw the claim, so it reads back the stored timestamp and releases on that +// exact key, keeping ReleaseTrailersRefreshClaim's equality guard meaningful. +func TestTrailersRefreshRequestedAt(t *testing.T) { + 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) + + repo := NewItemRepository(pool) + contentID := fmt.Sprintf("trailer-read-%d", time.Now().UnixNano()) + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM media_items WHERE content_id = $1`, contentID) + }) + if _, err := pool.Exec(ctx, ` + INSERT INTO media_items (content_id, type, title, status, genres) + VALUES ($1, 'movie', 'Trailer Read', 'matched', '{}'::text[]) + `, contentID); err != nil { + t.Fatalf("seed item: %v", err) + } + + const cooldown = 7 * 24 * time.Hour + + // A free slot reads as nil rather than an error: the recovery path uses + // that to conclude it owes nobody a release. + requestedAt, err := repo.TrailersRefreshRequestedAt(ctx, contentID) + if err != nil { + t.Fatalf("read an unclaimed slot: %v", err) + } + if requestedAt != nil { + t.Fatalf("unclaimed slot read as %s, want nil", requestedAt) + } + + claimed, claimedAt, err := repo.TryClaimTrailersRefresh(ctx, contentID, cooldown) + if err != nil || !claimed || claimedAt == nil { + t.Fatalf("claim = %v, at = %v, err = %v", claimed, claimedAt, err) + } + + requestedAt, err = repo.TrailersRefreshRequestedAt(ctx, contentID) + if err != nil { + t.Fatalf("read a claimed slot: %v", err) + } + if requestedAt == nil { + t.Fatal("a claimed slot must read back its timestamp") + } + // The read must reproduce the claim exactly, or the equality-guarded + // release it feeds would silently match nothing. + if !requestedAt.Equal(*claimedAt) { + t.Fatalf("read %s, want the claimed timestamp %s", requestedAt, claimedAt) + } + if err := repo.ReleaseTrailersRefreshClaim(ctx, contentID, *requestedAt); err != nil { + t.Fatalf("release on the read timestamp: %v", err) + } + requestedAt, err = repo.TrailersRefreshRequestedAt(ctx, contentID) + if err != nil { + t.Fatalf("read after release: %v", err) + } + if requestedAt != nil { + t.Fatalf("released slot still holds %s", requestedAt) + } + + // A missing item is distinguishable from a free slot. + if _, err := repo.TrailersRefreshRequestedAt(ctx, contentID+"-missing"); !errors.Is(err, ErrItemNotFound) { + t.Fatalf("read for a missing item = %v, want ErrItemNotFound", err) + } +} diff --git a/internal/metadata/refresh_debt.go b/internal/metadata/refresh_debt.go index 4819da0c..8effad88 100644 --- a/internal/metadata/refresh_debt.go +++ b/internal/metadata/refresh_debt.go @@ -14,6 +14,14 @@ const ( RefreshDebtReasonRefreshFailure int64 = 4 RefreshDebtReasonCoreMetadataIncomplete int64 = 8 RefreshDebtReasonProviderIDIncomplete int64 = 16 + // RefreshDebtReasonTrailersRequested marks a refresh a viewer asked for + // through the trailer action. It exists so that request survives a restart + // that kills the detached goroutine actually doing the work, and it is + // deliberately not a "something is wrong with this item" reason: it carries + // no priority case (so it sits at the default band and never front-runs + // real debt) and nothing recomputes it, so the next successful refresh + // clears it like any resolved reason. + RefreshDebtReasonTrailersRequested int64 = 32 ) const ( diff --git a/internal/metadata/refresh_debt_repo.go b/internal/metadata/refresh_debt_repo.go index de19aa9b..4ad59817 100644 --- a/internal/metadata/refresh_debt_repo.go +++ b/internal/metadata/refresh_debt_repo.go @@ -451,6 +451,7 @@ func (r *RefreshDebtRepository) GetMetrics(ctx context.Context, sampleLimit int) {reason: "provider_id_incomplete", mask: RefreshDebtReasonProviderIDIncomplete}, {reason: "refresh_failure", mask: RefreshDebtReasonRefreshFailure}, {reason: "core_metadata_incomplete", mask: RefreshDebtReasonCoreMetadataIncomplete}, + {reason: "trailers_requested", mask: RefreshDebtReasonTrailersRequested}, } for _, def := range reasonDefs { var count int diff --git a/internal/metadata/service.go b/internal/metadata/service.go index 0c2bb629..270f54c7 100644 --- a/internal/metadata/service.go +++ b/internal/metadata/service.go @@ -80,6 +80,16 @@ type metadataItemDeleteRepo interface { Delete(ctx context.Context, contentID string) ([]string, error) } +// metadataTrailerRefreshRepo is the cooldown gate behind +// RequestTrailersRefresh. It is a separate optional interface (asserted on +// itemRepo) because only the viewer-facing trailer action needs it; the +// concrete *catalog.ItemRepository satisfies it. +type metadataTrailerRefreshRepo interface { + TryClaimTrailersRefresh(ctx context.Context, contentID string, cooldown time.Duration) (bool, *time.Time, error) + ReleaseTrailersRefreshClaim(ctx context.Context, contentID string, claimedAt time.Time) error + TrailersRefreshRequestedAt(ctx context.Context, contentID string) (*time.Time, error) +} + type metadataProviderIDRepo interface { GetByContentID(ctx context.Context, contentID string) ([]*models.MediaItemProviderID, error) ReplaceByContentID(ctx context.Context, contentID string, providerIDs map[string]string) error @@ -799,6 +809,16 @@ func (s *MetadataService) resolveFolderLanguage(ctx context.Context, folderID in // is provided. The union (most-permissive) mirrors the multi-library language // posture. A nil return means "allow all": unknown scope or a transient // lookup failure must never wipe stored trailers. +// +// The empty (non-nil) result is load-bearing in the other direction — it means +// every containing library turned remote videos off, which filters everything +// out and which RequestTrailersRefresh reports to the viewer as "disabled". So +// a partially-resolved union cannot be returned as if it were complete: an +// unreadable library might be the one that enables trailers, and answering +// "disabled" (or filtering everything away) on its behalf would be a guess. +// Any lookup failure therefore degrades the whole answer to unknown scope. A +// folder that is genuinely gone is not a failure and is simply skipped — a +// library that no longer exists cannot be the one enabling trailers. func (s *MetadataService) resolveAllowedVideoKinds(ctx context.Context, contentID string, folderID int) map[models.ExtraKind]bool { if s.folderRepo == nil { return nil @@ -822,7 +842,12 @@ func (s *MetadataService) resolveAllowedVideoKinds(ctx context.Context, contentI resolvedAny := false for _, id := range folderIDs { folder, err := s.folderRepo.GetByID(ctx, id) - if err != nil || folder == nil { + switch { + case err != nil && !errors.Is(err, catalog.ErrFolderNotFound): + slog.WarnContext(ctx, "metadata: reading library trailer kinds failed; treating video scope as unknown", + "component", "metadata", "content_id", contentID, "folder_id", id, "error", err) + return nil + case err != nil, folder == nil: continue } resolvedAny = true @@ -2406,6 +2431,11 @@ func (s *MetadataService) mergeAndPersist( if len(filtered) > 0 || mergeMode == MergeReplaceUnlocked { if err := s.videoRepo.ReplaceByContentID(ctx, contentID, itemVideosFromRemote(contentID, filtered)); err != nil { slog.WarnContext(ctx, "metadata: failed to replace item videos", "component", "metadata", "content_id", contentID, "error", err) + // A failed write is invisible in ProcessResult by design (the + // rest of the refresh still succeeded), so tell any observer + // that asked — today, the viewer trailer action, which must + // not charge a cooldown for trailers it did not store. + reportVideoPersistFailure(ctx, err) } } } @@ -2596,10 +2626,90 @@ func (s *MetadataService) RefreshScheduledItem(ctx context.Context, contentID st // RefreshScheduledTarget re-fetches metadata for a queued item, season, or // episode target using the background refresh merge policy. +// +// A queued item may be the durable recovery for a viewer's trailer request +// whose process died mid-refresh (see RequestTrailersRefresh). That request +// consumed a week-long cooldown slot and took its release hook down with the +// process, so this path adopts both: it carries the same failure semantics, and +// a recovery that fails hands the slot back instead of leaving the viewer +// blocked for a week over trailers nobody ever stored. func (s *MetadataService) RefreshScheduledTarget(ctx context.Context, targetType, contentID string) error { + if NormalizeRefreshTargetType(targetType) == RefreshTargetItem { + if claim := s.adoptTrailersRefreshClaim(ctx, contentID); claim != nil { + return claim.run(ctx) + } + } return s.refreshTarget(ctx, targetType, contentID, 0, ModeScheduledRefresh, false) } +// trailersRefreshRecovery is an inherited trailer-refresh cooldown claim, held +// across the scheduled refresh that is recovering the request which consumed +// it. +type trailersRefreshRecovery struct { + service *MetadataService + gate metadataTrailerRefreshRepo + contentID string + claimedAt time.Time +} + +// adoptTrailersRefreshClaim reports the cooldown claim a queued item's refresh +// is responsible for, or nil when the refresh owes nobody a release. +// +// The debt row's trailers-requested reason bit is what makes the claim +// identifiable: RequestTrailersRefresh sets it exactly when it consumes a slot, +// and the first refresh that resolves the row clears it. Reading the stored +// timestamp gives the same key the original request held, so the release stays +// equality-guarded — a slot re-claimed by a newer request in the meantime is +// that request's to release, not this one's. +func (s *MetadataService) adoptTrailersRefreshClaim(ctx context.Context, contentID string) *trailersRefreshRecovery { + if s == nil || strings.TrimSpace(contentID) == "" { + return nil + } + gate, ok := s.itemRepo.(metadataTrailerRefreshRepo) + if !ok || gate == nil { + return nil + } + reasonMask, err := s.currentRefreshDebtTargetReasonMask(ctx, RefreshTargetItem, contentID) + if err != nil || !hasRefreshDebtReason(reasonMask, RefreshDebtReasonTrailersRequested) { + return nil + } + claimedAt, err := gate.TrailersRefreshRequestedAt(ctx, contentID) + if err != nil { + slog.WarnContext(ctx, "metadata: failed to read the trailers refresh claim a queued refresh inherits", + "component", "metadata", "content_id", contentID, "error", err) + return nil + } + if claimedAt == nil { + // The slot was already handed back (or the window lapsed), so this + // refresh owes nothing. + return nil + } + return &trailersRefreshRecovery{service: s, gate: gate, contentID: contentID, claimedAt: *claimedAt} +} + +// run performs the recovery refresh under the inherited claim, releasing the +// slot on the same failures the original request's hook covered — including a +// videos write that failed and was only logged, which leaves the refresh +// "successful" while storing none of the trailers the cooldown was charged for. +func (r *trailersRefreshRecovery) run(ctx context.Context) error { + var videoPersistErr atomic.Pointer[error] + refreshCtx := withVideoPersistFailureObserver(ctx, func(persistErr error) { + videoPersistErr.CompareAndSwap(nil, &persistErr) + }) + + err := r.service.refreshTarget(refreshCtx, RefreshTargetItem, r.contentID, 0, ModeScheduledRefresh, false) + releaseErr := err + if releaseErr == nil { + if stored := videoPersistErr.Load(); stored != nil { + releaseErr = fmt.Errorf("persisting item videos: %w", *stored) + } + } + if releaseErr != nil { + r.service.releaseTrailersRefreshClaim(r.gate, r.contentID, r.claimedAt, releaseErr) + } + return err +} + // RefreshItemForLibrary re-fetches metadata for an item using a specific // library's provider chain and metadata language preferences. func (s *MetadataService) RefreshItemForLibrary(ctx context.Context, contentID string, folderID int) error { @@ -2674,6 +2784,355 @@ func (s *MetadataService) RequestStaleMetadataRefresh(ctx context.Context, targe return nil } +// Trailer refresh outcome statuses returned by RequestTrailersRefresh. +const ( + // TrailerRefreshStatusQueued means the request won the cooldown gate and a + // detached refresh was started. + TrailerRefreshStatusQueued = "queued" + // TrailerRefreshStatusCooldown means the item was refreshed within the + // cooldown window; NextAllowedAt says when the next request may win. + TrailerRefreshStatusCooldown = "cooldown" + // TrailerRefreshStatusDisabled means every library containing the item has + // remote videos turned off, so a refresh could not produce trailers. + TrailerRefreshStatusDisabled = "disabled" +) + +// TrailerRefreshCooldown is the per-item window between viewer-triggered +// trailer refreshes. A full single-item refresh is not cheap, and provider +// video sets change slowly, so the window is deliberately long. +const TrailerRefreshCooldown = 7 * 24 * time.Hour + +// TrailerRefreshOutcome reports what a viewer's "find trailers" request did. +// NextAllowedAt is set only for the cooldown status. +type TrailerRefreshOutcome struct { + Status string + NextAllowedAt *time.Time +} + +// trailerRefreshReleaseTimeout bounds the write that hands a cooldown slot back +// after a failed refresh. It runs on its own context because the refresh's +// context is frequently already expired — a timeout is one of the failures the +// release exists for. +const trailerRefreshReleaseTimeout = 15 * time.Second + +// trailerRefreshClaimTimeout bounds the durable claim. The claim runs on a +// context detached from the request (see RequestTrailersRefresh) and so needs +// a deadline of its own; it is a single indexed UPDATE, so this is generous. +const trailerRefreshClaimTimeout = 15 * time.Second + +// trailerRefreshRecoveryDelay holds the durable recovery row back until after +// the detached fast path can possibly still be running. +// +// The debt row exists only to survive a process that dies mid-refresh. Due +// immediately, it is claimable by the refresh_metadata task the moment it is +// written, and that task calls RefreshScheduledTarget without consulting the +// in-process claim — so the worker and the goroutine would run the same full +// provider refresh at once, burning provider quota and racing each other's +// writes. Delaying past metadataOnDemandRefreshTimeout means the row can only +// come due once the goroutine is guaranteed finished (or gone with its +// process); on the normal path the refresh's own debt sync resolves the row +// long before then. +const trailerRefreshRecoveryDelay = 5 * time.Minute + +// videoPersistFailureContextKey scopes a videos-persistence observer to one +// refresh. mergeAndPersist logs and continues when videoRepo.ReplaceByContentID +// fails, because a video write failure must not fail a whole metadata refresh +// that otherwise succeeded — but the viewer-triggered trailer action needs to +// know, since "refresh succeeded" is then not the same as "trailers were +// saved", and it would otherwise consume a week-long cooldown for nothing. +type videoPersistFailureContextKey struct{} + +// withVideoPersistFailureObserver returns a context that reports a failed +// item_videos write to the supplied callback. Refreshes that do not install +// one — every background and admin path — are unaffected. +func withVideoPersistFailureObserver(ctx context.Context, observe func(error)) context.Context { + if observe == nil { + return ctx + } + return context.WithValue(ctx, videoPersistFailureContextKey{}, observe) +} + +// reportVideoPersistFailure notifies an installed observer, if any. +func reportVideoPersistFailure(ctx context.Context, err error) { + observe, _ := ctx.Value(videoPersistFailureContextKey{}).(func(error)) + if observe != nil { + observe(err) + } +} + +// RequestTrailersRefresh is the viewer-facing trailer fetch: it starts a full +// single-item metadata refresh at most once per TrailerRefreshCooldown. +// +// The refresh runs in scheduled mode (MergeFillEmpty), so this non-admin +// trigger cannot overwrite unlocked admin edits, while found videos still +// persist — mergeAndPersist writes item_videos whenever providers returned +// any, and skips the write when they returned none, so a transient empty +// result cannot wipe stored trailers. +// +// Ordering matters, and each step is a way to answer without burning the +// item's weekly slot on work that will not happen: +// - the disabled check runs first, so an item whose libraries have remote +// videos turned off never consumes a slot; +// - the in-process dedup claim runs next, so a request that lands while an +// equivalent refresh is already in flight reports "queued" (truthfully — +// one is running) and leaves the slot for a real retry; +// - only then is the durable slot consumed, and it is handed back if the +// refresh it started fails. +// +// A refresh that succeeds but finds no videos keeps the slot: that is the +// accepted "nothing to find, come back next week" outcome. +func (s *MetadataService) RequestTrailersRefresh(ctx context.Context, contentID string) (TrailerRefreshOutcome, error) { + if s == nil { + return TrailerRefreshOutcome{}, ErrMetadataNotFound + } + contentID = strings.TrimSpace(contentID) + if contentID == "" { + return TrailerRefreshOutcome{}, catalog.ErrItemNotFound + } + + // An admin lock on the videos field makes mergeAndPersist skip the + // item_videos write entirely (its isFieldLocked(locked, FieldVideos) + // guard), so a refresh started here would report success and consume the + // week having saved nothing. From the viewer's side that is the same + // answer as a library with + // remote videos turned off — trailers cannot be fetched for this item — so + // it reuses "disabled" rather than inventing a status clients do not know: + // the Apple coordinator treats an unrecognized status as "stop, nothing + // found", which would be a worse answer than the one disabled already + // gives. + if s.trailerVideosLocked(ctx, contentID) { + return TrailerRefreshOutcome{Status: TrailerRefreshStatusDisabled}, nil + } + + // A non-nil empty allow-list means every containing library disabled + // remote videos. A nil map means allow-all (unknown scope or a transient + // lookup failure) and must not short-circuit. + if allowed := s.resolveAllowedVideoKinds(ctx, contentID, 0); allowed != nil && len(allowed) == 0 { + return TrailerRefreshOutcome{Status: TrailerRefreshStatusDisabled}, nil + } + + gate, ok := s.itemRepo.(metadataTrailerRefreshRepo) + if !ok || gate == nil { + return TrailerRefreshOutcome{}, ErrMetadataNotFound + } + + // Losing the in-process claim means an equivalent full refresh for this + // item is already running (this action or the detail view's stale nudge — + // they share the key). Report it as queued and leave the slot alone: if + // that refresh fails, the viewer can retry immediately. + if !s.claimOnDemandMetadataRefresh(RefreshTargetItem, contentID) { + return TrailerRefreshOutcome{Status: TrailerRefreshStatusQueued}, nil + } + // The claim is ours from here: either the detached refresh takes ownership + // of it, or it is released before this call returns. + startedRefresh := false + defer func() { + if !startedRefresh { + s.releaseOnDemandMetadataRefresh(RefreshTargetItem, contentID) + } + }() + + // The claim is a durable side effect, so it must not ride the request's + // context: a cancellation landing after Postgres commits the UPDATE but + // before pgx returns would consume the slot for the whole window with no + // refresh started and nothing left holding the information needed to + // release it. Detaching from cancellation (with a deadline of its own) + // keeps the claim and the goroutine that owns its release inseparable. + claimCtx, cancelClaim := context.WithTimeout(context.WithoutCancel(ctx), trailerRefreshClaimTimeout) + claimed, requestedAt, err := gate.TryClaimTrailersRefresh(claimCtx, contentID, TrailerRefreshCooldown) + cancelClaim() + if err != nil { + return TrailerRefreshOutcome{}, err + } + if !claimed { + // A nil timestamp on a lost claim means the repository saw the slot + // freed underneath it twice over: another request is claiming it right + // now, so the honest answer is the same one a lost in-process claim + // gets rather than a cooldown nobody can date. + if requestedAt == nil { + return TrailerRefreshOutcome{Status: TrailerRefreshStatusQueued}, nil + } + next := requestedAt.Add(TrailerRefreshCooldown).UTC() + return TrailerRefreshOutcome{ + Status: TrailerRefreshStatusCooldown, + NextAllowedAt: &next, + }, nil + } + + // Record the refresh in the durable debt queue as well. The goroutine below + // is the fast path and normally finishes in seconds, but it does not + // survive a restart; the debt row does, so a process that dies mid-refresh + // leaves behind work the refresh worker will pick up instead of an item + // that waits out the window having fetched nothing. The row is deliberately + // not due yet (trailerRefreshRecoveryDelay) so the worker cannot run the + // same refresh alongside the goroutine, and the goroutine clears it on + // success, so it fires only when the fast path really did not finish. The + // queue is idempotent (RequestDue merges into any existing row and never + // pulls a leased or recently-attempted target forward), so this is additive. + s.enqueueTrailersRefreshDebt(ctx, contentID) + + // Hand the slot back if the refresh this request started fails, including + // on timeout: otherwise a provider outage would lock the item for the whole + // cooldown window without ever having fetched anything. + hooks := onDemandRefreshHooks{} + if requestedAt != nil { + claimedAt := *requestedAt + // A refresh can report success while the item_videos write inside it + // failed and was logged — from this action's point of view that is a + // failure, because the cooldown is a budget for *fetching trailers*. + var videoPersistErr atomic.Pointer[error] + hooks.decorateContext = func(refreshCtx context.Context) context.Context { + return withVideoPersistFailureObserver(refreshCtx, func(persistErr error) { + videoPersistErr.CompareAndSwap(nil, &persistErr) + }) + } + hooks.onComplete = func(refreshErr error) { + if refreshErr == nil { + if stored := videoPersistErr.Load(); stored != nil { + refreshErr = fmt.Errorf("persisting item videos: %w", *stored) + } + } + if refreshErr == nil { + // The fast path did the work, so the recovery row has nothing + // left to recover. Clearing it keeps the worker from re-running + // a refresh that already happened; the refresh's own debt sync + // usually gets there first, and this is idempotent either way. + s.settleTrailersRefreshDebt(contentID) + return + } + s.releaseTrailersRefreshClaim(gate, contentID, claimedAt, refreshErr) + } + } + s.runOnDemandMetadataRefresh(RefreshTargetItem, contentID, hooks) + startedRefresh = true + return TrailerRefreshOutcome{Status: TrailerRefreshStatusQueued}, nil +} + +// enqueueTrailersRefreshDebt records the item in the durable refresh-debt queue +// so a restart that kills the detached goroutine does not leave the cooldown +// consumed with no refresh ever performed. Best effort by design: failing to +// write the safety net must not fail a request whose refresh is about to start. +func (s *MetadataService) enqueueTrailersRefreshDebt(ctx context.Context, contentID string) { + if s == nil || s.refreshDebtRepo == nil { + return + } + // RefreshDebtReasonTrailersRequested rather than the generic failure reason: + // nothing is wrong with this item, so it must not land in the failure band + // ahead of real debt, nor be counted as a failure in the operator metrics. + // Nothing recomputes the bit, so the next successful refresh clears it. + reasonMask := RefreshDebtReasonTrailersRequested + // Not due until the fast path cannot still be running: RequestDue keeps the + // earlier of the two timestamps when a row already exists, so genuinely due + // debt for this item is never pushed out by the delay. + dueAt := time.Now().UTC().Add(trailerRefreshRecoveryDelay) + dueCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), trailerRefreshClaimTimeout) + defer cancel() + if err := s.refreshDebtRepo.RequestDue( + dueCtx, + RefreshTargetItem, + contentID, + refreshDebtPriority(reasonMask), + reasonMask, + dueAt, + metadataRefreshNudgeCooldown, + ); err != nil { + slog.WarnContext(dueCtx, "metadata: failed to record durable debt for a trailers refresh", "component", "metadata", + "content_id", contentID, "error", err) + } +} + +// settleTrailersRefreshDebt resolves the recovery row after the fast path +// finished the work it was insurance for. +// +// It runs on its own context in the detached goroutine, after the refresh's own +// debt sync has normally already rewritten or deleted the row — so this is a +// no-op in the common case and matters only when that sync did not clear the +// trailers-requested bit. Clearing just that bit (rather than deleting the row) +// keeps any real debt the item still carries: another reason left in the mask +// means the item genuinely needs refreshing again, and the queue should keep +// saying so. +func (s *MetadataService) settleTrailersRefreshDebt(contentID string) { + if s == nil || s.refreshDebtRepo == nil { + return + } + ctx, cancel := context.WithTimeout(context.Background(), trailerRefreshClaimTimeout) + defer cancel() + + debt, err := s.refreshDebtRepo.GetTarget(ctx, RefreshTargetItem, contentID) + if err != nil { + if !errors.Is(err, ErrRefreshDebtNotFound) { + slog.WarnContext(ctx, "metadata: failed to read durable debt after a trailers refresh", "component", "metadata", + "content_id", contentID, "error", err) + } + return + } + if debt == nil || !hasRefreshDebtReason(debt.ReasonMask, RefreshDebtReasonTrailersRequested) { + return + } + remaining := debt.ReasonMask &^ RefreshDebtReasonTrailersRequested + if remaining == 0 { + if err := s.refreshDebtRepo.DeleteTargetDebt(ctx, RefreshTargetItem, contentID); err != nil { + slog.WarnContext(ctx, "metadata: failed to clear durable debt after a trailers refresh", "component", "metadata", + "content_id", contentID, "error", err) + } + return + } + if err := s.refreshDebtRepo.MarkTargetSuccess( + ctx, + RefreshTargetItem, + contentID, + effectiveRefreshDebtPriority(remaining, debt.AttemptCount), + remaining, + nextRefreshAtForDebt(remaining, debt.AttemptCount, time.Now().UTC()), + ); err != nil { + slog.WarnContext(ctx, "metadata: failed to settle durable debt after a trailers refresh", "component", "metadata", + "content_id", contentID, "error", err) + } +} + +// trailerVideosLocked reports that an admin has locked the item's videos field, +// which makes mergeAndPersist skip the item_videos write no matter what the +// providers return. A refresh started in that state would report success and +// charge the viewer a week for trailers it could never save. +// +// A lookup failure answers false: the preflight exists to avoid a pointless +// refresh, and refusing the action because the database blinked would be a +// worse failure than performing one. +func (s *MetadataService) trailerVideosLocked(ctx context.Context, contentID string) bool { + if s == nil || s.itemRepo == nil { + return false + } + item, err := s.itemRepo.GetByID(ctx, contentID) + if err != nil || item == nil { + return false + } + return isFieldLocked(intSliceToFields(item.LockedFields), FieldVideos) +} + +// releaseTrailersRefreshClaim clears the cooldown slot this request consumed. +// The repository's equality guard means a slot already re-claimed by a newer +// request is left alone, so this is safe to run long after the fact. +func (s *MetadataService) releaseTrailersRefreshClaim( + gate metadataTrailerRefreshRepo, + contentID string, + claimedAt time.Time, + refreshErr error, +) { + ctx, cancel := context.WithTimeout(context.Background(), trailerRefreshReleaseTimeout) + defer cancel() + if err := gate.ReleaseTrailersRefreshClaim(ctx, contentID, claimedAt); err != nil { + slog.WarnContext(ctx, "metadata: failed to release trailers refresh cooldown slot", "component", "metadata", + "content_id", contentID, + "refresh_error", refreshErr, + "error", err) + return + } + slog.InfoContext(ctx, "metadata: released trailers refresh cooldown slot after a failed refresh", + "component", "metadata", + "content_id", contentID, + "refresh_error", refreshErr) +} + func (s *MetadataService) refreshDebtTargetIsDue(ctx context.Context, targetType, contentID string, now time.Time) (bool, error) { if s == nil || s.refreshDebtRepo == nil { return false, nil @@ -2694,27 +3153,66 @@ func (s *MetadataService) refreshDebtTargetIsDue(ctx context.Context, targetType return !debt.NextRefreshAt.After(now), nil } +// startOnDemandMetadataRefresh takes the in-process claim for the target and, +// if it wins, runs a detached refresh. Losing the claim means an equivalent +// refresh is already in flight and this call is a no-op. func (s *MetadataService) startOnDemandMetadataRefresh(targetType, contentID string) { if !s.claimOnDemandMetadataRefresh(targetType, contentID) { return } + s.runOnDemandMetadataRefresh(targetType, contentID, onDemandRefreshHooks{}) +} + +// onDemandRefreshHooks lets a caller that consumed durable state to start a +// detached refresh observe how that refresh went, so it can put the state back. +// The zero value is the plain fire-and-forget refresh every background caller +// wants. +type onDemandRefreshHooks struct { + // decorateContext wraps the detached refresh's context before the refresh + // runs — the way a caller installs an observer scoped to just this refresh + // (see withVideoPersistFailureObserver). + decorateContext func(context.Context) context.Context + // onComplete runs in the detached goroutine once the refresh has finished, + // with the refresh error or nil on success. "Success" here is only the + // pipeline's own verdict: a caller that cares about a specific sub-result + // has to observe that separately, because a refresh can succeed overall + // while a single persistence step logged and continued. + onComplete func(error) +} + +// runOnDemandMetadataRefresh runs the detached refresh for a claim the caller +// already holds, and takes ownership of releasing it. +// +// hooks.onComplete runs *before* the in-process claim is released, which keeps +// a useful invariant for whoever picks the claim up next: by the time it is +// free, the durable state has already been put back. The alternative ordering +// leaves a window in which a concurrent request sees consumed state for a +// refresh that has already finished. +func (s *MetadataService) runOnDemandMetadataRefresh(targetType, contentID string, hooks onDemandRefreshHooks) { go func() { defer s.releaseOnDemandMetadataRefresh(targetType, contentID) ctx, cancel := context.WithTimeout(context.Background(), metadataOnDemandRefreshTimeout) defer cancel() + if hooks.decorateContext != nil { + ctx = hooks.decorateContext(ctx) + } slog.Info("metadata: starting on-demand stale refresh", "target_type", targetType, "content_id", contentID) - if err := s.refreshTarget(ctx, targetType, contentID, 0, ModeScheduledRefresh, false); err != nil { + err := s.refreshTarget(ctx, targetType, contentID, 0, ModeScheduledRefresh, false) + if err != nil { slog.Warn("metadata: on-demand stale refresh failed", "target_type", targetType, "content_id", contentID, "error", err) - return + } else { + slog.Info("metadata: completed on-demand stale refresh", + "target_type", targetType, + "content_id", contentID) + } + if hooks.onComplete != nil { + hooks.onComplete(err) } - slog.Info("metadata: completed on-demand stale refresh", - "target_type", targetType, - "content_id", contentID) }() } diff --git a/internal/metadata/service_test.go b/internal/metadata/service_test.go index 5e136b5d..21a251a3 100644 --- a/internal/metadata/service_test.go +++ b/internal/metadata/service_test.go @@ -32,6 +32,26 @@ const ( type fakeItemRepo struct { mu sync.Mutex items map[string]*models.MediaItem + + // Trailer refresh cooldown state (metadataTrailerRefreshRepo). + trailersRequestedAt map[string]time.Time + trailersRequestedAtErr error + trailersClaims int + trailersClaimErr error + trailersClaimResult *trailersClaimResult + trailersReleases int + trailersReleased chan struct{} + trailersReleaseGate chan struct{} + now func() time.Time +} + +// trailersClaimResult forces a fixed answer out of the cooldown gate, for the +// outcomes the in-memory model cannot reach on its own — notably the real +// repository's "lost the gate but the slot kept being freed" answer, which +// carries no timestamp. +type trailersClaimResult struct { + claimed bool + requestedAt *time.Time } func newFakeItemRepo() *fakeItemRepo { @@ -122,6 +142,118 @@ func (r *fakeItemRepo) ListUnmatchedByFolderAndPathPrefix(_ context.Context, _ i return nil, nil } +// TryClaimTrailersRefresh mirrors the SQL gate in *catalog.ItemRepository: the +// claim succeeds only when no timestamp is stored or the stored one predates +// the cooldown window, and either way the caller reads back the timestamp now +// stored in the column. +func (r *fakeItemRepo) TryClaimTrailersRefresh(_ context.Context, contentID string, cooldown time.Duration) (bool, *time.Time, error) { + r.mu.Lock() + defer r.mu.Unlock() + if _, ok := r.items[contentID]; !ok { + return false, nil, catalog.ErrItemNotFound + } + now := time.Now().UTC() + if r.now != nil { + now = r.now() + } + if r.trailersClaimErr != nil { + return false, nil, r.trailersClaimErr + } + if forced := r.trailersClaimResult; forced != nil { + return forced.claimed, forced.requestedAt, nil + } + stored, ok := r.trailersRequestedAt[contentID] + if !ok || stored.Before(now.Add(-cooldown)) { + if r.trailersRequestedAt == nil { + r.trailersRequestedAt = make(map[string]time.Time) + } + r.trailersRequestedAt[contentID] = now + r.trailersClaims++ + claimed := now + return true, &claimed, nil + } + blocked := stored + return false, &blocked, nil +} + +// ReleaseTrailersRefreshClaim mirrors the equality-guarded UPDATE: a slot that +// has since been re-claimed by a newer request is left alone. +// +// trailersReleaseGate, when set, holds the release until the test closes it, +// which lets a test interleave a newer claim with a late-arriving release. +func (r *fakeItemRepo) ReleaseTrailersRefreshClaim(_ context.Context, contentID string, claimedAt time.Time) error { + r.mu.Lock() + gate := r.trailersReleaseGate + r.mu.Unlock() + if gate != nil { + <-gate + } + + r.mu.Lock() + defer r.mu.Unlock() + r.trailersReleases++ + if stored, ok := r.trailersRequestedAt[contentID]; ok && stored.Equal(claimedAt) { + delete(r.trailersRequestedAt, contentID) + } + if r.trailersReleased != nil { + close(r.trailersReleased) + r.trailersReleased = nil + } + return nil +} + +// TrailersRefreshRequestedAt reads back the stored claim the way the durable +// recovery path does, so a refresh that inherits a claim can release it on the +// same key the original request wrote. +func (r *fakeItemRepo) TrailersRefreshRequestedAt(_ context.Context, contentID string) (*time.Time, error) { + r.mu.Lock() + defer r.mu.Unlock() + if _, ok := r.items[contentID]; !ok { + return nil, catalog.ErrItemNotFound + } + if r.trailersRequestedAtErr != nil { + return nil, r.trailersRequestedAtErr + } + if stored, ok := r.trailersRequestedAt[contentID]; ok { + return &stored, nil + } + return nil, nil +} + +// expectTrailersRelease arms a channel closed by the next +// ReleaseTrailersRefreshClaim, so a test can wait for the detached refresh's +// failure path instead of polling. +func (r *fakeItemRepo) expectTrailersRelease() chan struct{} { + r.mu.Lock() + defer r.mu.Unlock() + released := make(chan struct{}) + r.trailersReleased = released + return released +} + +func (r *fakeItemRepo) trailersClaimCount() int { + r.mu.Lock() + defer r.mu.Unlock() + return r.trailersClaims +} + +func (r *fakeItemRepo) trailersReleaseCount() int { + r.mu.Lock() + defer r.mu.Unlock() + return r.trailersReleases +} + +// trailersStoredAt reports the timestamp currently stored for the item, or nil +// when the slot is free. +func (r *fakeItemRepo) trailersStoredAt(contentID string) *time.Time { + r.mu.Lock() + defer r.mu.Unlock() + if stored, ok := r.trailersRequestedAt[contentID]; ok { + return &stored + } + return nil +} + type fakeRefreshDebtRepo struct { mu sync.Mutex debts map[string]*models.MetadataRefreshDebt @@ -179,8 +311,13 @@ func (r *fakeRefreshDebtRepo) UpsertTargetDebt(_ context.Context, targetType, co return nil } +// RequestDue mirrors the repository's merge semantics rather than overwriting: +// the real statement ORs the reason mask, keeps the greater priority and the +// earlier next_refresh_at. Callers reason about all three (a trailer request +// adds its reason to whatever debt an item already has, and must not push +// genuinely-due work out), so a fake that replaced the row would hide that. func (r *fakeRefreshDebtRepo) RequestDue( - ctx context.Context, + _ context.Context, targetType string, contentID string, priority int, @@ -188,7 +325,29 @@ func (r *fakeRefreshDebtRepo) RequestDue( nextRefreshAt time.Time, _ time.Duration, ) error { - return r.UpsertTargetDebt(ctx, targetType, contentID, priority, reasonMask, nextRefreshAt) + r.mu.Lock() + defer r.mu.Unlock() + targetType = NormalizeRefreshTargetType(targetType) + key := fakeRefreshDebtKey(targetType, contentID) + if key == "" || contentID == "" || reasonMask == 0 { + return nil + } + if existing, ok := r.debts[key]; ok { + existing.ReasonMask |= reasonMask + existing.Priority = max(existing.Priority, priority) + if nextRefreshAt.Before(existing.NextRefreshAt) { + existing.NextRefreshAt = nextRefreshAt + } + return nil + } + r.debts[key] = &models.MetadataRefreshDebt{ + TargetType: targetType, + ContentID: contentID, + Priority: priority, + ReasonMask: reasonMask, + NextRefreshAt: nextRefreshAt, + } + return nil } func (r *fakeRefreshDebtRepo) MarkFailure( @@ -729,14 +888,21 @@ func (r *fakeLibraryRepo) CountFoldersForItem(ctx context.Context, contentID str type fakeMetadataFolderRepo struct { folders map[int]*models.MediaFolder + // lookupErrs forces a transient failure for a folder that otherwise + // exists. Callers distinguish "this library is gone" from "this library + // could not be read", so the fake has to be able to produce both. + lookupErrs map[int]error } func (r *fakeMetadataFolderRepo) GetByID(_ context.Context, id int) (*models.MediaFolder, error) { + if err, ok := r.lookupErrs[id]; ok { + return nil, err + } if folder, ok := r.folders[id]; ok { cp := *folder return &cp, nil } - return nil, fmt.Errorf("folder not found: %d", id) + return nil, catalog.ErrFolderNotFound } // fakeRootClaimRepo implements metadataRootClaimRepo. diff --git a/internal/metadata/trailers_refresh_test.go b/internal/metadata/trailers_refresh_test.go new file mode 100644 index 00000000..a5a992c3 --- /dev/null +++ b/internal/metadata/trailers_refresh_test.go @@ -0,0 +1,1020 @@ +package metadata + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/Silo-Server/silo-server/internal/catalog" + "github.com/Silo-Server/silo-server/internal/models" +) + +// waitForProcess blocks until the detached on-demand refresh has called +// Process, so a queued test does not leak a goroutine into the next one. +func waitForProcess(t *testing.T, started <-chan struct{}) { + t.Helper() + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("timed out waiting for the on-demand refresh to start") + } +} + +// waitForOnDemandIdle blocks until no on-demand refresh holds an in-process +// claim. The claim is released in the detached goroutine's defer, slightly +// after Process returns, and a still-held claim silently drops the next +// refresh — so a test that queues twice has to wait for it. +func waitForOnDemandIdle(t *testing.T, s *MetadataService) { + t.Helper() + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + s.onDemandRefresh.mu.Lock() + running := len(s.onDemandRefresh.running) + s.onDemandRefresh.mu.Unlock() + if running == 0 { + return + } + time.Sleep(time.Millisecond) + } + t.Fatal("timed out waiting for on-demand refresh claims to clear") +} + +// RequestTrailersRefresh reaches the cooldown gate through a runtime type +// assertion on itemRepo, so a drift in the repository's signature would turn +// every request into an error instead of failing the build. +func TestItemRepositorySatisfiesTrailerRefreshGate(t *testing.T) { + var repo any = (*catalog.ItemRepository)(nil) + if _, ok := repo.(metadataTrailerRefreshRepo); !ok { + t.Fatal("*catalog.ItemRepository must satisfy metadataTrailerRefreshRepo") + } +} + +func TestRequestTrailersRefreshQueuesThenReportsCooldown(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC) + h.itemRepo.now = func() time.Time { return now } + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + + started := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + close(started) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + + outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1") + if err != nil { + t.Fatalf("RequestTrailersRefresh: %v", err) + } + if outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("first status = %q, want %q", outcome.Status, TrailerRefreshStatusQueued) + } + if outcome.NextAllowedAt != nil { + t.Fatalf("queued outcome must not carry next_allowed_at, got %v", outcome.NextAllowedAt) + } + waitForProcess(t, started) + waitForOnDemandIdle(t, h.service) + + // The second request inside the window loses the gate and reports when the + // next one may win: the stored timestamp plus the cooldown. + outcome, err = h.service.RequestTrailersRefresh(ctx, "movie-1") + if err != nil { + t.Fatalf("RequestTrailersRefresh second: %v", err) + } + if outcome.Status != TrailerRefreshStatusCooldown { + t.Fatalf("second status = %q, want %q", outcome.Status, TrailerRefreshStatusCooldown) + } + if got := h.itemRepo.trailersReleaseCount(); got != 0 { + t.Fatalf("a successful refresh must keep the slot, released %d times", got) + } + if outcome.NextAllowedAt == nil { + t.Fatal("cooldown outcome must carry next_allowed_at") + } + want := now.Add(TrailerRefreshCooldown) + if !outcome.NextAllowedAt.Equal(want) { + t.Fatalf("next_allowed_at = %s, want %s", outcome.NextAllowedAt, want) + } + if got := h.itemRepo.trailersClaimCount(); got != 1 { + t.Fatalf("cooldown slot consumed %d times, want 1", got) + } +} + +func TestRequestTrailersRefreshAllowsRetryAfterCooldownLapses(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC) + h.itemRepo.now = func() time.Time { return now } + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + + processed := make(chan string, 4) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + processed <- req.ContentID + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + + if outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1"); err != nil || + outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("first request = %+v, err = %v", outcome, err) + } + select { + case <-processed: + case <-time.After(time.Second): + t.Fatal("timed out waiting for the first on-demand refresh") + } + waitForOnDemandIdle(t, h.service) + + // One second past the window the gate opens again. + now = now.Add(TrailerRefreshCooldown + time.Second) + outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1") + if err != nil { + t.Fatalf("RequestTrailersRefresh after cooldown: %v", err) + } + if outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("status after cooldown lapsed = %q, want %q", outcome.Status, TrailerRefreshStatusQueued) + } + select { + case <-processed: + case <-time.After(time.Second): + t.Fatal("timed out waiting for the second on-demand refresh") + } + if got := h.itemRepo.trailersClaimCount(); got != 2 { + t.Fatalf("cooldown slot consumed %d times, want 2", got) + } +} + +// A library whose trailer_kinds allow-list is empty has remote videos turned +// off, so the request is answered "disabled" — and must not burn the item's +// weekly slot, or a user would be locked out for a week over a no-op. +func TestRequestTrailersRefreshDisabledDoesNotConsumeCooldownSlot(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + if err := h.libraryRepo.Upsert(ctx, "movie-1", 10, time.Now()); err != nil { + t.Fatalf("seed library membership: %v", err) + } + folder := &models.MediaFolder{ID: 10, Type: "movies", Enabled: true, TrailerKinds: nil} + h.service.folderRepo = &fakeMetadataFolderRepo{folders: map[int]*models.MediaFolder{10: folder}} + + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + t.Errorf("disabled request must not start a refresh (content_id %s)", req.ContentID) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + + outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1") + if err != nil { + t.Fatalf("RequestTrailersRefresh: %v", err) + } + if outcome.Status != TrailerRefreshStatusDisabled { + t.Fatalf("status = %q, want %q", outcome.Status, TrailerRefreshStatusDisabled) + } + if got := h.itemRepo.trailersClaimCount(); got != 0 { + t.Fatalf("disabled request consumed the cooldown slot %d times, want 0", got) + } + + // Re-enabling the library lets the very next request through, proving the + // slot really was untouched. + folder.TrailerKinds = []string{string(models.ExtraKindTrailer)} + started := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + close(started) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + outcome, err = h.service.RequestTrailersRefresh(ctx, "movie-1") + if err != nil { + t.Fatalf("RequestTrailersRefresh after re-enabling: %v", err) + } + if outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("status after re-enabling = %q, want %q", outcome.Status, TrailerRefreshStatusQueued) + } + waitForProcess(t, started) +} + +// A nil allow-list is "allow all" — an unknown scope or a transient library +// lookup failure. It must not be mistaken for "disabled". +func TestRequestTrailersRefreshTreatsUnknownLibraryScopeAsAllowed(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + // The item has no library membership, so resolveAllowedVideoKinds returns + // nil rather than an empty map. + h.service.folderRepo = &fakeMetadataFolderRepo{folders: map[int]*models.MediaFolder{}} + + started := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + close(started) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + + outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1") + if err != nil { + t.Fatalf("RequestTrailersRefresh: %v", err) + } + if outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("status = %q, want %q", outcome.Status, TrailerRefreshStatusQueued) + } + waitForProcess(t, started) +} + +func TestRequestTrailersRefreshPropagatesGateErrors(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + gateErr := errors.New("database is down") + h.itemRepo.trailersClaimErr = gateErr + + if _, err := h.service.RequestTrailersRefresh(ctx, "movie-1"); !errors.Is(err, gateErr) { + t.Fatalf("err = %v, want %v", err, gateErr) + } + // The failed gate call must not strand the in-process claim, or every + // later request for this item would be silently deduped away. + waitForOnDemandIdle(t, h.service) +} + +// The weekly slot pays for work actually done. When the refresh it started +// fails — a provider outage, a timeout — the slot goes back so the viewer can +// retry now instead of waiting out a window in which nothing was fetched. +func TestRequestTrailersRefreshReleasesSlotWhenRefreshFails(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC) + h.itemRepo.now = func() time.Time { return now } + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + + h.service.hooks.process = func(_ context.Context, _ ProcessRequest) (*ProcessResult, error) { + return nil, errors.New("tmdb is unreachable") + } + + released := h.itemRepo.expectTrailersRelease() + outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1") + if err != nil { + t.Fatalf("RequestTrailersRefresh: %v", err) + } + if outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("status = %q, want %q", outcome.Status, TrailerRefreshStatusQueued) + } + select { + case <-released: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the failed refresh to release the cooldown slot") + } + if stored := h.itemRepo.trailersStoredAt("movie-1"); stored != nil { + t.Fatalf("failed refresh left the slot consumed until %s", stored) + } + waitForOnDemandIdle(t, h.service) + + // The very next request wins the gate again, with no clock movement. + started := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + close(started) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + outcome, err = h.service.RequestTrailersRefresh(ctx, "movie-1") + if err != nil { + t.Fatalf("retry after a failed refresh: %v", err) + } + if outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("retry status = %q, want %q", outcome.Status, TrailerRefreshStatusQueued) + } + waitForProcess(t, started) +} + +// A refresh that succeeds but turns up nothing keeps the slot: "no trailers +// exist for this title" is an answer, and re-asking providers weekly is the +// accepted cost ceiling. +func TestRequestTrailersRefreshKeepsSlotWhenRefreshFindsNothing(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC) + h.itemRepo.now = func() time.Time { return now } + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + + // A successful refresh that produced no videos is indistinguishable here + // from any other success: Process returns Updated, and no video rows were + // written. + started := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + close(started) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + + if outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1"); err != nil || + outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("request = %+v, err = %v", outcome, err) + } + waitForProcess(t, started) + waitForOnDemandIdle(t, h.service) + + if got := h.itemRepo.trailersReleaseCount(); got != 0 { + t.Fatalf("successful refresh released the slot %d times, want 0", got) + } + if stored := h.itemRepo.trailersStoredAt("movie-1"); stored == nil { + t.Fatal("successful refresh must keep the slot consumed") + } + outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1") + if err != nil { + t.Fatalf("second request: %v", err) + } + if outcome.Status != TrailerRefreshStatusCooldown { + t.Fatalf("status after a successful empty refresh = %q, want %q", + outcome.Status, TrailerRefreshStatusCooldown) + } +} + +// The release is guarded on the timestamp the failing request wrote, so a +// release that lands after the window lapsed and a newer request claimed the +// slot must leave that newer claim alone — otherwise the late write would hand +// out a free extra refresh. +// +// The newer claim is taken against the gate directly rather than through +// RequestTrailersRefresh: the point under test is the timestamp guard, and +// driving it through the public API would only exercise the in-process dedup +// that TestRequestTrailersRefreshInFlightRefreshQueuesWithoutConsumingSlot +// already covers. +func TestRequestTrailersRefreshReleaseDoesNotClobberNewerClaim(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC) + h.itemRepo.now = func() time.Time { return now } + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + + // Hold the failing request's release until a newer claim is in place. + gate := make(chan struct{}) + h.itemRepo.trailersReleaseGate = gate + + failed := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, _ ProcessRequest) (*ProcessResult, error) { + close(failed) + return nil, errors.New("tmdb is unreachable") + } + + released := h.itemRepo.expectTrailersRelease() + if outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1"); err != nil || + outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("first request = %+v, err = %v", outcome, err) + } + waitForProcess(t, failed) + + // The window lapses and a later request wins the gate afresh. + now = now.Add(TrailerRefreshCooldown + time.Second) + newClaimAt := now + claimed, claimedAt, err := h.itemRepo.TryClaimTrailersRefresh(ctx, "movie-1", TrailerRefreshCooldown) + if err != nil || !claimed || claimedAt == nil { + t.Fatalf("newer claim = %v, at = %v, err = %v", claimed, claimedAt, err) + } + + // Only now does the first request's release land. + close(gate) + select { + case <-released: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the late release") + } + waitForOnDemandIdle(t, h.service) + + stored := h.itemRepo.trailersStoredAt("movie-1") + if stored == nil { + t.Fatal("the late release cleared a slot claimed by a newer request") + } + if !stored.Equal(newClaimAt) { + t.Fatalf("stored timestamp = %s, want the newer claim %s", stored, newClaimAt) + } +} + +// The in-process claim is shared with the detail view's stale-metadata nudge. +// A trailer request that arrives while an equivalent refresh is already running +// is answered "queued" — one really is running — without consuming the weekly +// slot, so a failure of that refresh still leaves the viewer able to retry. +func TestRequestTrailersRefreshInFlightRefreshQueuesWithoutConsumingSlot(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + + // Hold an in-flight refresh open for the duration of the request under + // test, exactly as the item-detail path's nudge would. + inFlight := make(chan struct{}) + entered := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + close(entered) + <-inFlight + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + h.service.startOnDemandMetadataRefresh(RefreshTargetItem, "movie-1") + waitForProcess(t, entered) + + outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1") + if err != nil { + t.Fatalf("RequestTrailersRefresh: %v", err) + } + if outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("status = %q, want %q", outcome.Status, TrailerRefreshStatusQueued) + } + if got := h.itemRepo.trailersClaimCount(); got != 0 { + t.Fatalf("in-flight refresh consumed the cooldown slot %d times, want 0", got) + } + + close(inFlight) + waitForOnDemandIdle(t, h.service) + + // The slot was untouched, so the next request still wins the gate. + started := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + close(started) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + outcome, err = h.service.RequestTrailersRefresh(ctx, "movie-1") + if err != nil { + t.Fatalf("request after the in-flight refresh finished: %v", err) + } + if outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("status = %q, want %q", outcome.Status, TrailerRefreshStatusQueued) + } + if got := h.itemRepo.trailersClaimCount(); got != 1 { + t.Fatalf("cooldown slot consumed %d times, want 1", got) + } + waitForProcess(t, started) +} + +// The disabled short-circuit returns before the in-process claim is taken, so +// it must not leave one behind either. +func TestRequestTrailersRefreshDisabledLeavesNoInProcessClaim(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + if err := h.libraryRepo.Upsert(ctx, "movie-1", 10, time.Now()); err != nil { + t.Fatalf("seed library membership: %v", err) + } + h.service.folderRepo = &fakeMetadataFolderRepo{folders: map[int]*models.MediaFolder{ + 10: {ID: 10, Type: "movies", Enabled: true, TrailerKinds: nil}, + }} + + if outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1"); err != nil || + outcome.Status != TrailerRefreshStatusDisabled { + t.Fatalf("request = %+v, err = %v", outcome, err) + } + waitForOnDemandIdle(t, h.service) +} + +// A cooldown answer takes and then hands back the in-process claim; leaking it +// would mute every subsequent refresh for the item until the process restarts. +func TestRequestTrailersRefreshCooldownLeavesNoInProcessClaim(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC) + h.itemRepo.now = func() time.Time { return now } + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + + started := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + close(started) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + if outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1"); err != nil || + outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("first request = %+v, err = %v", outcome, err) + } + waitForProcess(t, started) + waitForOnDemandIdle(t, h.service) + + if outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1"); err != nil || + outcome.Status != TrailerRefreshStatusCooldown { + t.Fatalf("second request = %+v, err = %v", outcome, err) + } + waitForOnDemandIdle(t, h.service) +} + +// A refresh whose item_videos write failed is a failure for this action even +// though the pipeline reports success: the cooldown is a budget for fetching +// trailers, and charging a week for trailers that were fetched but not stored +// would strand the viewer. +func TestRequestTrailersRefreshReleasesSlotWhenVideoPersistFails(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC) + h.itemRepo.now = func() time.Time { return now } + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + + // Stand in for mergeAndPersist: the pipeline succeeds overall while the + // videos write fails and is only logged, which is exactly the shape the + // observer exists to surface. + persistErr := errors.New("replace item videos: connection reset") + h.service.hooks.process = func(processCtx context.Context, req ProcessRequest) (*ProcessResult, error) { + reportVideoPersistFailure(processCtx, persistErr) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + + released := h.itemRepo.expectTrailersRelease() + if outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1"); err != nil || + outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("request = %+v, err = %v", outcome, err) + } + select { + case <-released: + case <-time.After(time.Second): + t.Fatal("timed out waiting for the slot to be released after a failed videos write") + } + waitForOnDemandIdle(t, h.service) + + if stored := h.itemRepo.trailersStoredAt("movie-1"); stored != nil { + t.Fatalf("slot must be free after a failed videos write, stored %s", stored) + } + // The viewer can retry immediately rather than waiting out the window. + started := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + close(started) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + if outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1"); err != nil || + outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("retry = %+v, want an immediately queued retry, err = %v", outcome, err) + } + waitForProcess(t, started) +} + +// The detached goroutine does not survive a restart, so winning the gate also +// records durable debt: a process that dies mid-refresh leaves work the refresh +// worker picks up instead of an item locked out for the window having fetched +// nothing. +func TestRequestTrailersRefreshRecordsDurableDebt(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + debts := newFakeRefreshDebtRepo() + h.service.refreshDebtRepo = debts + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + + started := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + close(started) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + if outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1"); err != nil || + outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("request = %+v, err = %v", outcome, err) + } + + debt, err := debts.GetTarget(ctx, RefreshTargetItem, "movie-1") + if err != nil { + t.Fatalf("queued request must leave durable debt behind: %v", err) + } + if !hasRefreshDebtReason(debt.ReasonMask, RefreshDebtReasonTrailersRequested) { + t.Fatalf("reason mask = %d, want the trailers-requested reason set", debt.ReasonMask) + } + // Nothing is wrong with the item, so the row must not sit in a band that + // front-runs genuine debt. + if debt.Priority != refreshDebtPriority(0) { + t.Fatalf("priority = %d, want the default band %d", debt.Priority, refreshDebtPriority(0)) + } + waitForProcess(t, started) + waitForOnDemandIdle(t, h.service) +} + +// A cooldown answer performs no work, so it must not enqueue debt either — +// otherwise repeated polling from a client would keep an item permanently due. +func TestRequestTrailersRefreshCooldownRecordsNoDebt(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC) + h.itemRepo.now = func() time.Time { return now } + debts := newFakeRefreshDebtRepo() + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + + started := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + close(started) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + if outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1"); err != nil || + outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("first request = %+v, err = %v", outcome, err) + } + waitForProcess(t, started) + waitForOnDemandIdle(t, h.service) + + // Wire the debt repo only now, so anything it records can only have come + // from the cooldown request below. + h.service.refreshDebtRepo = debts + if outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1"); err != nil || + outcome.Status != TrailerRefreshStatusCooldown { + t.Fatalf("second request = %+v, err = %v", outcome, err) + } + if _, err := debts.GetTarget(ctx, RefreshTargetItem, "movie-1"); !errors.Is(err, ErrRefreshDebtNotFound) { + t.Fatalf("a cooldown answer must not enqueue debt, got err = %v", err) + } +} + +// A claim lost to a slot that keeps being freed underneath the repository is +// not a cooldown — there is no timestamp to report one with. Reporting it as +// queued matches the in-process-dedup answer: an equivalent refresh is running. +func TestRequestTrailersRefreshUndateableLostClaimReportsQueued(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + h.itemRepo.trailersClaimResult = &trailersClaimResult{} + + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + t.Errorf("a lost claim must not start a refresh (content_id %s)", req.ContentID) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + + outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1") + if err != nil { + t.Fatalf("RequestTrailersRefresh: %v", err) + } + if outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("status = %q, want %q", outcome.Status, TrailerRefreshStatusQueued) + } + if outcome.NextAllowedAt != nil { + t.Fatalf("an undateable lost claim must not carry next_allowed_at, got %v", outcome.NextAllowedAt) + } + waitForOnDemandIdle(t, h.service) +} + +// "Disabled" means every containing library turned remote videos off. That +// claim cannot be made from a partially-resolved set: an unreadable library +// might be the one that enables trailers, so any lookup failure degrades the +// answer to unknown scope (allow-all) rather than a guess the viewer sees as +// "trailers are disabled for this library". +func TestRequestTrailersRefreshUnreadableLibraryIsNotDisabled(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + for _, folderID := range []int{10, 11} { + if err := h.libraryRepo.Upsert(ctx, "movie-1", folderID, time.Now()); err != nil { + t.Fatalf("seed library membership %d: %v", folderID, err) + } + } + // Folder 10 resolves with trailers off; folder 11 cannot be read at all. + h.service.folderRepo = &fakeMetadataFolderRepo{ + folders: map[int]*models.MediaFolder{ + 10: {ID: 10, Type: "movies", Enabled: true, TrailerKinds: nil}, + }, + lookupErrs: map[int]error{11: errors.New("connection reset")}, + } + + started := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + close(started) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + + outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1") + if err != nil { + t.Fatalf("RequestTrailersRefresh: %v", err) + } + if outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("status = %q, want %q — a failed library lookup is unknown scope, not disabled", + outcome.Status, TrailerRefreshStatusQueued) + } + waitForProcess(t, started) + waitForOnDemandIdle(t, h.service) +} + +// A library that no longer exists is not a failure: it cannot be the one +// enabling trailers, so it is skipped and the remaining libraries still decide. +func TestRequestTrailersRefreshMissingLibraryStillReportsDisabled(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + for _, folderID := range []int{10, 11} { + if err := h.libraryRepo.Upsert(ctx, "movie-1", folderID, time.Now()); err != nil { + t.Fatalf("seed library membership %d: %v", folderID, err) + } + } + // Folder 11 is absent from the repo entirely (deleted library). + h.service.folderRepo = &fakeMetadataFolderRepo{ + folders: map[int]*models.MediaFolder{ + 10: {ID: 10, Type: "movies", Enabled: true, TrailerKinds: nil}, + }, + } + + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + t.Errorf("disabled request must not start a refresh (content_id %s)", req.ContentID) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + + outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1") + if err != nil { + t.Fatalf("RequestTrailersRefresh: %v", err) + } + if outcome.Status != TrailerRefreshStatusDisabled { + t.Fatalf("status = %q, want %q", outcome.Status, TrailerRefreshStatusDisabled) + } + if got := h.itemRepo.trailersClaimCount(); got != 0 { + t.Fatalf("disabled request consumed the cooldown slot %d times, want 0", got) + } +} + +// An admin lock on the videos field makes mergeAndPersist skip the item_videos +// write, so a refresh started for one would "succeed" having saved nothing and +// charge the viewer a week for it. The preflight has to catch that before the +// slot is consumed. +func TestRequestTrailersRefreshLockedVideosDoesNotConsumeCooldownSlot(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + h.itemRepo.items["movie-1"] = &models.MediaItem{ + ContentID: "movie-1", + Type: "movie", + Status: "matched", + LockedFields: []int{int(FieldVideos)}, + } + + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + t.Errorf("a videos-locked item must not start a refresh (content_id %s)", req.ContentID) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + + outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1") + if err != nil { + t.Fatalf("RequestTrailersRefresh: %v", err) + } + // Reuses "disabled" rather than a new status: clients treat unknown + // statuses as a dead end, and "trailers cannot be fetched for this item" + // is exactly what disabled already means to a viewer. + if outcome.Status != TrailerRefreshStatusDisabled { + t.Fatalf("status = %q, want %q", outcome.Status, TrailerRefreshStatusDisabled) + } + if got := h.itemRepo.trailersClaimCount(); got != 0 { + t.Fatalf("a videos-locked item consumed the cooldown slot %d times, want 0", got) + } + waitForOnDemandIdle(t, h.service) + + // Unlocking lets the very next request through, proving the slot was never + // touched. + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + started := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + close(started) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + if outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1"); err != nil || + outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("request after unlocking = %+v, err = %v", outcome, err) + } + waitForProcess(t, started) +} + +// A lock on some *other* field says nothing about videos, so it must not block +// the action. +func TestRequestTrailersRefreshUnrelatedLockStillQueues(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + h.itemRepo.items["movie-1"] = &models.MediaItem{ + ContentID: "movie-1", + Type: "movie", + Status: "matched", + LockedFields: []int{int(FieldOverview), int(FieldImages)}, + } + + started := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + close(started) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + + outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1") + if err != nil { + t.Fatalf("RequestTrailersRefresh: %v", err) + } + if outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("status = %q, want %q", outcome.Status, TrailerRefreshStatusQueued) + } + waitForProcess(t, started) +} + +// The recovery row is insurance against a process that dies mid-refresh, so it +// must not be claimable while the fast path could still be running: the refresh +// task calls RefreshScheduledTarget without consulting the in-process claim, so +// a due-now row would have the worker and the goroutine fetching the same item +// at once. +func TestRequestTrailersRefreshRecoveryDebtIsNotDueDuringTheFastPath(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + debts := newFakeRefreshDebtRepo() + h.service.refreshDebtRepo = debts + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + + // Hold the refresh open so the debt row is observed exactly while the fast + // path is running — the window the reviewer's race lives in. + inFlight := make(chan struct{}) + entered := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + close(entered) + <-inFlight + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + if outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1"); err != nil || + outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("request = %+v, err = %v", outcome, err) + } + waitForProcess(t, entered) + + debt, err := debts.GetTarget(ctx, RefreshTargetItem, "movie-1") + if err != nil { + t.Fatalf("queued request must leave durable debt behind: %v", err) + } + if !debt.NextRefreshAt.After(time.Now().UTC().Add(metadataOnDemandRefreshTimeout)) { + t.Fatalf("recovery debt is due at %s, which is within the on-demand refresh window (%s) — "+ + "the refresh worker could claim it alongside the running goroutine", + debt.NextRefreshAt, metadataOnDemandRefreshTimeout) + } + + close(inFlight) + waitForOnDemandIdle(t, h.service) +} + +// Once the fast path has done the work, the recovery row has nothing left to +// recover: leaving it would have the worker re-run a refresh that already +// happened as soon as the delay lapsed. +func TestRequestTrailersRefreshClearsRecoveryDebtOnSuccess(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + debts := newFakeRefreshDebtRepo() + h.service.refreshDebtRepo = debts + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + + // Process here stands in for a refresh whose own debt sync did not run + // (hooks.process short-circuits processInternal), which is the case the + // settle step exists to cover. + started := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + close(started) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + if outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1"); err != nil || + outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("request = %+v, err = %v", outcome, err) + } + waitForProcess(t, started) + waitForOnDemandIdle(t, h.service) + + if _, err := debts.GetTarget(ctx, RefreshTargetItem, "movie-1"); !errors.Is(err, ErrRefreshDebtNotFound) { + t.Fatalf("a completed fast path must leave no recovery debt, got err = %v", err) + } +} + +// Settling the recovery reason must not discard debt the item genuinely has: +// another reason in the mask means it still needs refreshing, and the queue +// should keep saying so. +func TestRequestTrailersRefreshSettleKeepsOtherDebtReasons(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + debts := newFakeRefreshDebtRepo() + h.service.refreshDebtRepo = debts + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + if err := debts.UpsertTargetDebt(ctx, RefreshTargetItem, "movie-1", + refreshDebtPriority(RefreshDebtReasonCoreMetadataIncomplete), + RefreshDebtReasonCoreMetadataIncomplete, + time.Now().UTC()); err != nil { + t.Fatalf("seed existing debt: %v", err) + } + + started := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + close(started) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + if outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1"); err != nil || + outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("request = %+v, err = %v", outcome, err) + } + waitForProcess(t, started) + waitForOnDemandIdle(t, h.service) + + debt, err := debts.GetTarget(ctx, RefreshTargetItem, "movie-1") + if err != nil { + t.Fatalf("pre-existing debt must survive the settle: %v", err) + } + if hasRefreshDebtReason(debt.ReasonMask, RefreshDebtReasonTrailersRequested) { + t.Fatalf("reason mask = %d, want the trailers-requested bit cleared", debt.ReasonMask) + } + if !hasRefreshDebtReason(debt.ReasonMask, RefreshDebtReasonCoreMetadataIncomplete) { + t.Fatalf("reason mask = %d, want the pre-existing core-metadata reason kept", debt.ReasonMask) + } +} + +// The durable recovery is the path taken when the process that consumed a slot +// died mid-refresh. It runs in a worker that never saw the claim, so without an +// explicit adoption a failed recovery would leave the viewer blocked for the +// whole window having stored nothing — the exact gap the release hook closes on +// the fast path. +func TestRefreshScheduledTargetReleasesInheritedTrailerClaimOnFailure(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC) + h.itemRepo.now = func() time.Time { return now } + debts := newFakeRefreshDebtRepo() + h.service.refreshDebtRepo = debts + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + + // Stand in for the state a dead process left behind: the slot is consumed + // and the debt row carries the trailers-requested reason. + claimed, claimedAt, err := h.itemRepo.TryClaimTrailersRefresh(ctx, "movie-1", TrailerRefreshCooldown) + if err != nil || !claimed || claimedAt == nil { + t.Fatalf("seed claim = %v, at = %v, err = %v", claimed, claimedAt, err) + } + if err := debts.UpsertTargetDebt(ctx, RefreshTargetItem, "movie-1", + refreshDebtPriority(RefreshDebtReasonTrailersRequested), + RefreshDebtReasonTrailersRequested, + now); err != nil { + t.Fatalf("seed recovery debt: %v", err) + } + + h.service.hooks.process = func(_ context.Context, _ ProcessRequest) (*ProcessResult, error) { + return nil, errors.New("tmdb is unreachable") + } + + if err := h.service.RefreshScheduledTarget(ctx, RefreshTargetItem, "movie-1"); err == nil { + t.Fatal("the recovery refresh was expected to fail") + } + if stored := h.itemRepo.trailersStoredAt("movie-1"); stored != nil { + t.Fatalf("a failed recovery left the slot consumed until %s", stored) + } + + // The viewer can retry immediately rather than waiting out a window in + // which nothing was ever fetched. + started := make(chan struct{}) + h.service.hooks.process = func(_ context.Context, req ProcessRequest) (*ProcessResult, error) { + close(started) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + if outcome, err := h.service.RequestTrailersRefresh(ctx, "movie-1"); err != nil || + outcome.Status != TrailerRefreshStatusQueued { + t.Fatalf("retry after a failed recovery = %+v, err = %v", outcome, err) + } + waitForProcess(t, started) +} + +// A recovery whose videos write failed and was only logged is a failure for the +// cooldown's purposes too: the pipeline reports success while none of the +// trailers the week was charged for were stored. +func TestRefreshScheduledTargetReleasesInheritedClaimWhenVideoPersistFails(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC) + h.itemRepo.now = func() time.Time { return now } + debts := newFakeRefreshDebtRepo() + h.service.refreshDebtRepo = debts + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + + if claimed, _, err := h.itemRepo.TryClaimTrailersRefresh(ctx, "movie-1", TrailerRefreshCooldown); err != nil || !claimed { + t.Fatalf("seed claim = %v, err = %v", claimed, err) + } + if err := debts.UpsertTargetDebt(ctx, RefreshTargetItem, "movie-1", + refreshDebtPriority(RefreshDebtReasonTrailersRequested), + RefreshDebtReasonTrailersRequested, + now); err != nil { + t.Fatalf("seed recovery debt: %v", err) + } + + persistErr := errors.New("replace item videos: connection reset") + h.service.hooks.process = func(processCtx context.Context, req ProcessRequest) (*ProcessResult, error) { + reportVideoPersistFailure(processCtx, persistErr) + return &ProcessResult{ContentID: req.ContentID, Updated: true}, nil + } + + // The refresh itself succeeded, so the queue must still see a success. + if err := h.service.RefreshScheduledTarget(ctx, RefreshTargetItem, "movie-1"); err != nil { + t.Fatalf("RefreshScheduledTarget: %v", err) + } + if stored := h.itemRepo.trailersStoredAt("movie-1"); stored != nil { + t.Fatalf("slot must be free after a failed videos write, stored %s", stored) + } +} + +// A scheduled refresh for an item nobody asked trailers for owes no release: +// the item may hold a claim from an unrelated in-flight request, and clearing +// it would hand out a free extra refresh. +func TestRefreshScheduledTargetLeavesUnrelatedTrailerClaimsAlone(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC) + h.itemRepo.now = func() time.Time { return now } + debts := newFakeRefreshDebtRepo() + h.service.refreshDebtRepo = debts + h.itemRepo.items["movie-1"] = &models.MediaItem{ContentID: "movie-1", Type: "movie", Status: "matched"} + + if claimed, _, err := h.itemRepo.TryClaimTrailersRefresh(ctx, "movie-1", TrailerRefreshCooldown); err != nil || !claimed { + t.Fatalf("seed claim = %v, err = %v", claimed, err) + } + // Debt without the trailers-requested reason: ordinary scheduled work. + if err := debts.UpsertTargetDebt(ctx, RefreshTargetItem, "movie-1", + refreshDebtPriority(RefreshDebtReasonCoreMetadataIncomplete), + RefreshDebtReasonCoreMetadataIncomplete, + now); err != nil { + t.Fatalf("seed debt: %v", err) + } + + h.service.hooks.process = func(_ context.Context, _ ProcessRequest) (*ProcessResult, error) { + return nil, errors.New("tmdb is unreachable") + } + + if err := h.service.RefreshScheduledTarget(ctx, RefreshTargetItem, "movie-1"); err == nil { + t.Fatal("the scheduled refresh was expected to fail") + } + if h.itemRepo.trailersStoredAt("movie-1") == nil { + t.Fatal("an unrelated scheduled refresh released a cooldown slot it does not own") + } + if got := h.itemRepo.trailersReleaseCount(); got != 0 { + t.Fatalf("unrelated refresh released the slot %d times, want 0", got) + } +} diff --git a/internal/ratelimit/middleware.go b/internal/ratelimit/middleware.go index 0422f35f..fee47c58 100644 --- a/internal/ratelimit/middleware.go +++ b/internal/ratelimit/middleware.go @@ -46,6 +46,23 @@ func (mw *Middleware) ActiveBackend() string { return "redis" } +// SharedLimiter returns the process's configured per-key limiter, so an action +// handler that enforces its own budget (person refresh, trailer refresh) counts +// against the same backend the middleware uses rather than a private in-memory +// one. That distinction only matters on Redis deployments, where a private +// limiter would give every instance an independent allowance for the same user +// and multiply the stated budget by the instance count. +// +// Handlers must tolerate nil: rate limiting is disabled outright when +// rate_limit.enabled is false or the database is unavailable, and no limiter +// exists then. +func (mw *Middleware) SharedLimiter() RateLimiter { + if mw == nil { + return nil + } + return mw.perKey +} + // Init loads config and seeds defaults. Call once at startup. func (mw *Middleware) Init(ctx context.Context) error { if err := SeedDefaults(ctx, mw.store); err != nil { diff --git a/migrations/sql/20260802055513_media_items_trailers_refresh_requested_at.sql b/migrations/sql/20260802055513_media_items_trailers_refresh_requested_at.sql new file mode 100644 index 00000000..e7bd1a45 --- /dev/null +++ b/migrations/sql/20260802055513_media_items_trailers_refresh_requested_at.sql @@ -0,0 +1,12 @@ +-- +goose Up +-- Cooldown state for the viewer-facing "find trailers" action. The refresh +-- debt queue cannot hold it: MarkTargetSuccess deletes the row once the reason +-- mask clears, so its last_attempt_at evaporates exactly on success. NULL +-- means "never requested"; the request path's atomic check-and-set writes +-- NOW() only when the stored timestamp is older than the cooldown window. +ALTER TABLE media_items + ADD COLUMN trailers_refresh_requested_at TIMESTAMPTZ; + +-- +goose Down +ALTER TABLE media_items + DROP COLUMN trailers_refresh_requested_at; diff --git a/web/src/lib/overlays/schema.ts b/web/src/lib/overlays/schema.ts index 26b3376f..e1291959 100644 --- a/web/src/lib/overlays/schema.ts +++ b/web/src/lib/overlays/schema.ts @@ -19,7 +19,10 @@ export function buildDefaultPrefs(): CardOverlayPrefs { // the native clients' settings UIs can author them, and dropping them here // would erase another client's preference on the next web save. Their bases // mirror the native registries' defaults (ribbons: top-right, disabled). -const PASSTHROUGH_IDS = ["imdb_top_250", "rt_certified_fresh"] as const satisfies readonly OverlayId[]; +const PASSTHROUGH_IDS = [ + "imdb_top_250", + "rt_certified_fresh", +] as const satisfies readonly OverlayId[]; const PASSTHROUGH_BASE: OverlayItemConfig = { enabled: false, position: "top-right" }; function isKnownOverlayId(v: unknown): v is OverlayId { diff --git a/web/src/pages/AdminTaskDetail.tsx b/web/src/pages/AdminTaskDetail.tsx index 70db4f41..2913abcf 100644 --- a/web/src/pages/AdminTaskDetail.tsx +++ b/web/src/pages/AdminTaskDetail.tsx @@ -29,6 +29,7 @@ const REFRESH_REASON_LABELS: Record = { stale_provider_id: "Stale provider ID", refresh_failure: "Refresh failure", core_metadata_incomplete: "Core metadata incomplete", + trailers_requested: "Trailers requested", }; // --- Trigger display helpers --- diff --git a/web/src/pages/AdminTasks.tsx b/web/src/pages/AdminTasks.tsx index bde827de..6a36f935 100644 --- a/web/src/pages/AdminTasks.tsx +++ b/web/src/pages/AdminTasks.tsx @@ -31,6 +31,7 @@ const REFRESH_REASON_LABELS: Record = { stale_provider_id: "Stale provider ID", refresh_failure: "Refresh failure", core_metadata_incomplete: "Core metadata incomplete", + trailers_requested: "Trailers requested", }; function useTaskClock() {