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() {