* feat(requests): enforce per-profile rating limits in discovery - Resolve each profile's max content rating and filter discovery, detail, and browse results against it, failing closed on missing ratings - Reject request submissions for titles above the viewer's ceiling - Add TMDB GetCertification backed by release_dates/content_ratings with a long-lived cache and singleflight - Push certification.lte to TMDB for studio/network/genre browse as a cost pre-filter - Backfill restricted section pages from a fixed window of TMDB pages to keep carousels populated and pagination stable * fix(requests): address discovery rating review findings - Preserve backfill overflow: sections use plain TMDB cursor semantics plus an additive next_page field instead of fixed windows, so an early stop never drops allowed titles from unconsumed pages (bit hardest at permissive R/TV-MA ceilings). - Bound cold-path cost: DiscoverAll backfills at most 2 TMDB pages per section (vs 5 for a direct section request), capping worst-case cold certification hydration at 240 lookups instead of 600. - Keep the TMDB prefilter a superset: rank-3 ceilings now push down certification.lte=NC-17/TV-MA rather than R, so titles the local ladder allows can't vanish upstream unrecoverably. - Fail closed on foreign certifications: enforcement-path lookups use new US-only pickers (a Canadian PG no longer reads as US PG), while the display path keeps its any-country fallback. US multi-entry disagreements prefer the theatrical/real rating over festival NR. - Detach shared certification fetches from the first caller's context (WithoutCancel + 30s bound) so one disconnecting client can't fail the singleflight result for concurrent waiters. - Advertise enforcement via rating_restrictions_enforced on /requests/status so clients can feature-detect instead of version-sniffing. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(requests): harden rating enforcement per second review pass - GetDetail gates on the US-only enforcement certification (cached GetCertification) instead of the display rating, whose any-country fallback let a foreign "PG" pass the US ladder. - pickUSMovieCertification takes the strictest recognized US rating when multiple release entries disagree ([PG, R] -> R); entry order is not meaningful and enforcement must not admit a title on its most lenient certificate. - Certification singleflight uses DoChan so a canceled caller returns ctx.Err() immediately instead of blocking up to 30s on the detached shared fetch (which still completes for surviving waiters). - Viewer rating ceiling resolves once per request and threads through discover/browse/detail enrichment (enrichPageWithCeiling); DiscoverAll drops from 12 scope resolutions per load to 1. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2228 lines
72 KiB
Go
2228 lines
72 KiB
Go
package requests
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/Silo-Server/silo-server/internal/access"
|
|
"github.com/Silo-Server/silo-server/internal/idgen"
|
|
"github.com/Silo-Server/silo-server/internal/metadata/tmdb"
|
|
"golang.org/x/sync/errgroup"
|
|
)
|
|
|
|
type TMDBClient interface {
|
|
SearchMedia(ctx context.Context, mediaType, query string, page int) (*tmdb.MediaPage, error)
|
|
DiscoverSection(ctx context.Context, section string, page int) (*tmdb.MediaPage, error)
|
|
GetMediaDetail(ctx context.Context, mediaType string, id int) (*tmdb.MediaDetail, error)
|
|
DiscoverPage(ctx context.Context, mediaType string, params tmdb.DiscoverParams, page int) (*tmdb.MediaPage, error)
|
|
}
|
|
|
|
type TMDBExternalIDClient interface {
|
|
GetExternalIDs(ctx context.Context, mediaType string, id int) (*tmdb.ExternalIDs, error)
|
|
}
|
|
|
|
// TMDBCertificationClient resolves a title's content rating. Detected by type
|
|
// assertion on the service's TMDBClient, like TMDBExternalIDClient.
|
|
type TMDBCertificationClient interface {
|
|
GetCertification(ctx context.Context, mediaType string, id int) (string, error)
|
|
}
|
|
|
|
const externalIDHydrationConcurrency = 4
|
|
|
|
const certificationHydrationConcurrency = 8
|
|
|
|
type EntitlementResolver interface {
|
|
// MaxPlaybackQuality returns the requester's effective playback-quality
|
|
// ceiling (already combining account- and profile-level caps). Empty string
|
|
// means "no cap".
|
|
MaxPlaybackQuality(ctx context.Context, userID int, profileID string) (string, error)
|
|
}
|
|
|
|
// ContentRatingResolver resolves the viewer's effective parental rating
|
|
// ceiling. Detected by type assertion on the service's EntitlementResolver so
|
|
// existing EntitlementResolver fakes keep compiling.
|
|
type ContentRatingResolver interface {
|
|
// MaxContentRating returns the profile's content-rating ceiling. Empty
|
|
// string means "no ceiling".
|
|
MaxContentRating(ctx context.Context, userID int, profileID string) (string, error)
|
|
}
|
|
|
|
// RequesterIdentityResolver resolves a requesting user id into the identity a
|
|
// per-user request_router plugin needs (e.g. Seerr attribution by email).
|
|
type RequesterIdentityResolver interface {
|
|
ResolveRequester(ctx context.Context, userID int) (email, username string, err error)
|
|
}
|
|
|
|
type Service struct {
|
|
store Store
|
|
tmdb TMDBClient
|
|
presence PresenceResolver
|
|
router RequestRouterProvider
|
|
entitlements EntitlementResolver
|
|
groupProvider access.GroupPolicyProvider
|
|
requesterIdentity RequesterIdentityResolver
|
|
notifier FulfillmentNotifier
|
|
lifecycle LifecycleNotifier
|
|
Now func() time.Time
|
|
}
|
|
|
|
type DiscoverySection struct {
|
|
Key string `json:"key"`
|
|
Title string `json:"title"`
|
|
Page int `json:"page"`
|
|
TotalPages int `json:"total_pages"`
|
|
TotalResults int `json:"total_results"`
|
|
Results []MediaResult `json:"results"`
|
|
// NextPage is the cursor to request for the following page, needed when
|
|
// rating-filter backfill consumes more than one TMDB page per request
|
|
// (page+1 would repeat consumed pages). 0 when there are no more pages.
|
|
// Additive v1 field; absent (0) also when the viewer is unrestricted and
|
|
// plain page+1 semantics apply.
|
|
NextPage int `json:"next_page,omitempty"`
|
|
}
|
|
|
|
func NewService(store Store, tmdbClient TMDBClient, presence PresenceResolver) *Service {
|
|
return &Service{
|
|
store: store,
|
|
tmdb: tmdbClient,
|
|
presence: presence,
|
|
Now: func() time.Time { return time.Now().UTC() },
|
|
}
|
|
}
|
|
|
|
func (s *Service) SetRouterProvider(p RequestRouterProvider) { s.router = p }
|
|
|
|
func (s *Service) SetEntitlementResolver(r EntitlementResolver) { s.entitlements = r }
|
|
|
|
func (s *Service) SetGroupPolicyProvider(p access.GroupPolicyProvider) { s.groupProvider = p }
|
|
|
|
func (s *Service) SetRequesterIdentityResolver(r RequesterIdentityResolver) {
|
|
s.requesterIdentity = r
|
|
}
|
|
|
|
// populateRequesterIdentity fills req.RequesterEmail/Username from the resolver.
|
|
// Nil resolver or any error leaves them empty (the plugin then behaves as admin).
|
|
func (s *Service) populateRequesterIdentity(ctx context.Context, req *Request) {
|
|
if s.requesterIdentity == nil || req.RequestedByUserID <= 0 {
|
|
return
|
|
}
|
|
email, username, err := s.requesterIdentity.ResolveRequester(ctx, req.RequestedByUserID)
|
|
if err != nil {
|
|
slog.WarnContext(ctx, "requests: requester identity resolve failed; attributing to admin", "component", "requests", "user_id", req.RequestedByUserID, "error", err)
|
|
return
|
|
}
|
|
req.RequesterEmail, req.RequesterUsername = email, username
|
|
}
|
|
|
|
func (s *Service) requesterCeiling(ctx context.Context, userID int, profileID string) string {
|
|
if s.entitlements == nil {
|
|
return "" // no resolver -> unlimited (1080p baseline still applies)
|
|
}
|
|
q, err := s.entitlements.MaxPlaybackQuality(ctx, userID, profileID)
|
|
if err != nil {
|
|
return access.PlaybackQualityStandard // fail safe: HD only
|
|
}
|
|
return q
|
|
}
|
|
|
|
// viewerContentCeiling resolves the viewer's parental rating ceiling. Empty
|
|
// string means unrestricted. Unlike the quality ceiling this is a safety
|
|
// filter, so a resolver error propagates instead of degrading: silently
|
|
// treating a failed lookup as "unrestricted" would leak adult content to a
|
|
// kid profile, and treating it as "restricted" would render every carousel
|
|
// empty with no visible cause.
|
|
func (s *Service) viewerContentCeiling(ctx context.Context, viewer Viewer) (string, error) {
|
|
resolver, ok := s.entitlements.(ContentRatingResolver)
|
|
if !ok {
|
|
return "", nil
|
|
}
|
|
return resolver.MaxContentRating(ctx, viewer.UserID, viewer.ProfileID)
|
|
}
|
|
|
|
type certKey struct {
|
|
mediaType MediaType
|
|
id int
|
|
}
|
|
|
|
// hydrateCertifications resolves content ratings for every unique
|
|
// (mediaType, id) pair on the page. The TMDB client caches certifications
|
|
// title-keyed with a long TTL, so in steady state this issues no requests.
|
|
func (s *Service) hydrateCertifications(ctx context.Context, raw *tmdb.MediaPage) (map[certKey]string, error) {
|
|
client, ok := s.tmdb.(TMDBCertificationClient)
|
|
if !ok {
|
|
return nil, fmt.Errorf("requests: tmdb client cannot resolve certifications")
|
|
}
|
|
|
|
keys := make([]certKey, 0, len(raw.Results))
|
|
seen := map[certKey]bool{}
|
|
for _, item := range raw.Results {
|
|
mediaType, err := normalizeMediaType(MediaType(item.MediaType))
|
|
if err != nil || item.ID <= 0 {
|
|
continue
|
|
}
|
|
key := certKey{mediaType: mediaType, id: item.ID}
|
|
if !seen[key] {
|
|
seen[key] = true
|
|
keys = append(keys, key)
|
|
}
|
|
}
|
|
|
|
certs := make([]string, len(keys))
|
|
group, gctx := errgroup.WithContext(ctx)
|
|
group.SetLimit(certificationHydrationConcurrency)
|
|
for i, key := range keys {
|
|
i, key := i, key
|
|
group.Go(func() error {
|
|
cert, err := client.GetCertification(gctx, tmdbMediaType(key.mediaType), key.id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
certs[i] = cert
|
|
return nil
|
|
})
|
|
}
|
|
if err := group.Wait(); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
out := make(map[certKey]string, len(keys))
|
|
for i, key := range keys {
|
|
out[key] = certs[i]
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// Section backfill: fail-closed filtering hides most of a TMDB section page
|
|
// for a restricted profile (typically 15+ of 20, since unrated titles are
|
|
// dropped), which renders as a nearly empty carousel. To compensate, a
|
|
// restricted viewer's section page consumes consecutive TMDB pages, starting
|
|
// at the requested page, until a page's worth of titles survives filtering or
|
|
// the per-request budget (sectionBackfillMaxPagesPerRequest) is spent.
|
|
//
|
|
// Pagination stays honest through two properties: `page` keeps plain TMDB
|
|
// cursor semantics (same as for unrestricted viewers), and the response's
|
|
// next_page reports the cursor after the last TMDB page actually consumed.
|
|
// Every survivor from a consumed page is returned — nothing is trimmed and no
|
|
// consumed page is partially dropped — so resuming at next_page never skips
|
|
// or repeats an allowed title. The budget also bounds the cold-cache cost: a
|
|
// permissive ceiling (R/TV-MA) fills from one TMDB page and stops immediately,
|
|
// while a strict ceiling spends at most the budget per request.
|
|
const (
|
|
// Per-request TMDB page budgets. A single-section request can afford a
|
|
// deeper scan than the six-section DiscoverAll aggregate: on a cold
|
|
// certification cache each consumed page costs up to 20 cert lookups
|
|
// against the client's rate limiter, so DiscoverAll's worst case is
|
|
// 6 sections x sectionBackfillBudgetAggregate pages x 20. Keeping the
|
|
// aggregate budget small bounds the first-paint latency for a restricted
|
|
// profile; the per-section endpoint (carousel "load more") gets the
|
|
// deeper budget. Steady state is unaffected — certifications are cached
|
|
// for 7 days, shared across all profiles.
|
|
sectionBackfillBudgetSingle = 5
|
|
sectionBackfillBudgetAggregate = 2
|
|
sectionResultsPerPage = 20
|
|
sectionBackfillMaxPage = 500 // TMDB hard-caps page at 500
|
|
)
|
|
|
|
func (s *Service) backfillSectionPage(ctx context.Context, section string, page, pageBudget int, ceiling string) (result *tmdb.MediaPage, nextPage int, err error) {
|
|
if page <= 0 {
|
|
page = 1
|
|
}
|
|
if pageBudget <= 0 {
|
|
pageBudget = 1
|
|
}
|
|
|
|
out := &tmdb.MediaPage{Page: page}
|
|
var results []tmdb.MediaResult
|
|
// Trending/popular orderings shift between fetches, so consecutive TMDB
|
|
// pages can overlap; dedupe within the request.
|
|
seen := map[certKey]bool{}
|
|
totalPages := page // until TMDB tells us the real count, assume the current page exists
|
|
consumed := 0
|
|
for next := page; consumed < pageBudget && next <= totalPages && next <= sectionBackfillMaxPage; next++ {
|
|
// Enough survived — stop before spending another TMDB page. Titles on
|
|
// unconsumed pages are not lost: next_page points at the first page
|
|
// this request did not consume.
|
|
if len(results) >= sectionResultsPerPage {
|
|
break
|
|
}
|
|
raw, err := s.tmdb.DiscoverSection(ctx, section, next)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
if raw == nil {
|
|
break
|
|
}
|
|
consumed++
|
|
nextPage = next + 1
|
|
if raw.TotalPages > 0 {
|
|
totalPages = raw.TotalPages
|
|
out.TotalResults = raw.TotalResults
|
|
}
|
|
filtered, err := s.filterPageByCeiling(ctx, raw, ceiling)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
for _, item := range filtered.Results {
|
|
key := certKey{mediaType: MediaType(item.MediaType), id: item.ID}
|
|
if !seen[key] {
|
|
seen[key] = true
|
|
results = append(results, item)
|
|
}
|
|
}
|
|
}
|
|
out.Results = results
|
|
// TotalPages/TotalResults stay TMDB's unfiltered counts — page keeps TMDB
|
|
// cursor semantics, and the filtered totals are unknowable without a full
|
|
// scan. next_page is the honest resume cursor.
|
|
out.TotalPages = totalPages
|
|
if nextPage > totalPages || nextPage > sectionBackfillMaxPage {
|
|
nextPage = 0 // exhausted
|
|
}
|
|
return out, nextPage, nil
|
|
}
|
|
|
|
// ensureCreateAllowedByCeiling rejects request submissions for titles above
|
|
// the viewer's rating ceiling. List filtering alone is cosmetic — the create
|
|
// endpoint is directly callable with a guessable TMDB id.
|
|
func (s *Service) ensureCreateAllowedByCeiling(ctx context.Context, viewer Viewer, input CreateRequestInput) error {
|
|
ceiling, err := s.viewerContentCeiling(ctx, viewer)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if ceiling == "" {
|
|
return nil
|
|
}
|
|
client, ok := s.tmdb.(TMDBCertificationClient)
|
|
if !ok {
|
|
return fmt.Errorf("requests: tmdb client cannot resolve certifications")
|
|
}
|
|
cert, err := client.GetCertification(ctx, tmdbMediaType(input.MediaType), input.TMDBID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !access.RatingAllowed(cert, ceiling) {
|
|
return ErrForbidden
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// filterPageByCeiling drops results whose certification exceeds the ceiling,
|
|
// failing closed on missing or unrecognized certifications. TMDB's own
|
|
// certification.lte pre-filter (applied on browse paths) is not trusted for
|
|
// this: it ranks "NR" below "G" and matches titles when any one of several
|
|
// US cert entries qualifies, both of which leak over-ceiling titles.
|
|
// TotalPages/TotalResults are intentionally left as TMDB reported them —
|
|
// recomputing them would require scanning every page, and short pages are
|
|
// benign for the carousel/browse UIs.
|
|
func (s *Service) filterPageByCeiling(ctx context.Context, raw *tmdb.MediaPage, ceiling string) (*tmdb.MediaPage, error) {
|
|
certs, err := s.hydrateCertifications(ctx, raw)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
filtered := *raw
|
|
filtered.Results = make([]tmdb.MediaResult, 0, len(raw.Results))
|
|
for _, item := range raw.Results {
|
|
mediaType, err := normalizeMediaType(MediaType(item.MediaType))
|
|
if err != nil || item.ID <= 0 {
|
|
continue
|
|
}
|
|
if access.RatingAllowed(certs[certKey{mediaType: mediaType, id: item.ID}], ceiling) {
|
|
filtered.Results = append(filtered.Results, item)
|
|
}
|
|
}
|
|
return &filtered, nil
|
|
}
|
|
|
|
// allowedQualities returns the qualities a request may receive: 1080p always,
|
|
// plus 2160p when force-dual is on or the requester's entitlement ceiling allows 4K.
|
|
func (s *Service) allowedQualities(ctx context.Context, req Request, settings Settings) []Quality {
|
|
out := []Quality{Quality1080p}
|
|
ceiling := s.requesterCeiling(ctx, req.RequestedByUserID, req.RequestedByProfileID)
|
|
// QualityAllowed treats an empty ceiling as "no cap" (the "Any" preset), so a
|
|
// requester with unlimited playback quality correctly gets 4K. A raw
|
|
// CompareQuality would rank "" as the LOWEST quality and wrongly drop 4K.
|
|
if settings.ForceDualQuality || access.QualityAllowed(access.PlaybackQuality4K, ceiling) {
|
|
out = append(out, Quality2160p)
|
|
}
|
|
return out
|
|
}
|
|
|
|
// fulfillContext caches the global fulfillment inputs for one reconcile cycle
|
|
// (or a single Approve/Retry) so integrations and settings are fetched once
|
|
// instead of per request. API keys need no cache here: the repository decrypts
|
|
// api_key_ref on read, so Integration.APIKeyRef already holds the literal key.
|
|
type fulfillContext struct {
|
|
integrations []Integration
|
|
settings Settings
|
|
}
|
|
|
|
func (s *Service) newFulfillContext(ctx context.Context) (*fulfillContext, error) {
|
|
integrations, err := s.store.ListIntegrations(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
settings, err := s.store.GetSettings(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &fulfillContext{integrations: integrations, settings: settings}, nil
|
|
}
|
|
|
|
// resolveRouterConnections turns enabled request_router integrations that serve
|
|
// the given media type into ResolvedRouterConnections (api key resolved to
|
|
// plaintext, plugin_config attached), and returns the installation+capability to
|
|
// dispatch to.
|
|
//
|
|
// It filters by media type to match the integrationConfigured auto-approve gate
|
|
// (so a series-only connection is never used for a movie request). Multi-
|
|
// installation routing isn't supported yet: it picks the first eligible
|
|
// connection's installation and includes ONLY connections belonging to it, so a
|
|
// second installation's resolved plaintext credentials are never handed to the
|
|
// first plugin. A connection whose api key cannot be resolved (or resolves empty)
|
|
// is skipped rather than aborting the whole request — a sibling healthy
|
|
// connection can still fulfill it, and an unauthenticated request is never sent.
|
|
func (s *Service) resolveRouterConnections(ctx context.Context, fc *fulfillContext, mediaType MediaType) ([]ResolvedRouterConnection, int, string, error) {
|
|
var conns []ResolvedRouterConnection
|
|
installationID, capabilityID := 0, ""
|
|
chosen := false
|
|
for _, in := range fc.integrations {
|
|
if !eligibleRouterConnection(in, mediaType) {
|
|
continue
|
|
}
|
|
// Contain to the first chosen (installation, capability): a plugin may
|
|
// expose more than one request_router capability, and a connection of a
|
|
// different capability must never be handed to the chosen one.
|
|
if chosen && (*in.InstallationID != installationID || in.CapabilityID != capabilityID) {
|
|
continue
|
|
}
|
|
// in.APIKeyRef was decrypted by the repo on read; empty means unconfigured.
|
|
apiKey := strings.TrimSpace(in.APIKeyRef)
|
|
if apiKey == "" {
|
|
slog.WarnContext(ctx, "requests: skipping router connection with no api key", "component", "requests", "connection_id", in.ID)
|
|
continue
|
|
}
|
|
// Lock on the first SUCCESSFULLY resolved connection so a skipped
|
|
// bad-key connection never pins the installation/capability.
|
|
if !chosen {
|
|
installationID, capabilityID, chosen = *in.InstallationID, in.CapabilityID, true
|
|
}
|
|
conns = append(conns, ResolvedRouterConnection{ID: in.ID, BaseURL: in.BaseURL, APIKey: apiKey, Config: in.PluginConfig})
|
|
}
|
|
return conns, installationID, capabilityID, nil
|
|
}
|
|
|
|
// eligibleRouterConnection reports whether a connection is a candidate fulfillment
|
|
// backend for the media type: enabled, bound to an installation, and naming a
|
|
// capability sub-id that serves the media type. resolveRouterConnections (which
|
|
// then resolves credentials) and integrationConfigured (the auto-approval gate)
|
|
// share this predicate so the two cannot drift.
|
|
func eligibleRouterConnection(in Integration, mediaType MediaType) bool {
|
|
return in.Enabled && in.CapabilityID != "" && in.InstallationID != nil &&
|
|
integrationSupportsMediaType(in, mediaType)
|
|
}
|
|
|
|
func (s *Service) Search(ctx context.Context, viewer Viewer, query string, mediaType MediaType, page int) (*MediaPage, error) {
|
|
if s == nil || s.store == nil || s.tmdb == nil {
|
|
return nil, fmt.Errorf("request service is not configured")
|
|
}
|
|
if err := s.ensureRequestsEnabled(ctx); err != nil {
|
|
return nil, err
|
|
}
|
|
mediaType, err := normalizeSearchMediaType(mediaType)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
query = strings.TrimSpace(query)
|
|
if query == "" {
|
|
return nil, fmt.Errorf("%w: query is required", ErrInvalidInput)
|
|
}
|
|
raw, err := s.tmdb.SearchMedia(ctx, string(mediaType), query, page)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return s.enrichPage(ctx, viewer, raw)
|
|
}
|
|
|
|
func (s *Service) Discover(ctx context.Context, viewer Viewer, section string, page int) (*DiscoverySection, error) {
|
|
if s == nil || s.store == nil || s.tmdb == nil {
|
|
return nil, fmt.Errorf("request service is not configured")
|
|
}
|
|
ceiling, err := s.viewerContentCeiling(ctx, viewer)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return s.discover(ctx, viewer, section, page, sectionBackfillBudgetSingle, ceiling)
|
|
}
|
|
|
|
// discover renders one section for an already-resolved ceiling. Callers own
|
|
// the ceiling lookup so a DiscoverAll fan-out resolves the viewer scope once,
|
|
// not once per section per page.
|
|
func (s *Service) discover(ctx context.Context, viewer Viewer, section string, page, backfillBudget int, ceiling string) (*DiscoverySection, error) {
|
|
if s == nil || s.store == nil || s.tmdb == nil {
|
|
return nil, fmt.Errorf("request service is not configured")
|
|
}
|
|
if err := s.ensureRequestsEnabled(ctx); err != nil {
|
|
return nil, err
|
|
}
|
|
section = strings.TrimSpace(section)
|
|
if _, ok := discoverySectionTitles[section]; !ok {
|
|
return nil, fmt.Errorf("%w: invalid discovery section", ErrInvalidInput)
|
|
}
|
|
var raw *tmdb.MediaPage
|
|
var err error
|
|
var nextPage int
|
|
if ceiling != "" {
|
|
raw, nextPage, err = s.backfillSectionPage(ctx, section, page, backfillBudget, ceiling)
|
|
} else {
|
|
raw, err = s.tmdb.DiscoverSection(ctx, section, page)
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
enriched, err := s.enrichPageWithCeiling(ctx, viewer, raw, ceiling)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &DiscoverySection{
|
|
Key: section,
|
|
Title: discoverySectionTitles[section],
|
|
Page: enriched.Page,
|
|
TotalPages: enriched.TotalPages,
|
|
TotalResults: enriched.TotalResults,
|
|
Results: enriched.Results,
|
|
NextPage: nextPage,
|
|
}, nil
|
|
}
|
|
|
|
func (s *Service) DiscoverAll(ctx context.Context, viewer Viewer) ([]DiscoverySection, error) {
|
|
if s == nil || s.store == nil || s.tmdb == nil {
|
|
return nil, fmt.Errorf("request service is not configured")
|
|
}
|
|
if err := s.ensureRequestsEnabled(ctx); err != nil {
|
|
return nil, err
|
|
}
|
|
ceiling, err := s.viewerContentCeiling(ctx, viewer)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
sections := make([]DiscoverySection, len(discoverySectionOrder))
|
|
group, gctx := errgroup.WithContext(ctx)
|
|
group.SetLimit(externalIDHydrationConcurrency)
|
|
for i, key := range discoverySectionOrder {
|
|
i, key := i, key
|
|
group.Go(func() error {
|
|
section, err := s.discover(gctx, viewer, key, 1, sectionBackfillBudgetAggregate, ceiling)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
sections[i] = *section
|
|
return nil
|
|
})
|
|
}
|
|
if err := group.Wait(); err != nil {
|
|
return nil, err
|
|
}
|
|
return sections, nil
|
|
}
|
|
|
|
// GetDetail fetches a TMDB detail payload and overlays the same availability /
|
|
// request-state signals used by search and discovery. Recommendations carry
|
|
// their own per-item state so the detail page can render them as request cards.
|
|
func (s *Service) GetDetail(ctx context.Context, viewer Viewer, mediaType MediaType, tmdbID int) (*MediaDetail, error) {
|
|
if s == nil || s.store == nil || s.tmdb == nil {
|
|
return nil, fmt.Errorf("request service is not configured")
|
|
}
|
|
if err := s.ensureRequestsEnabled(ctx); err != nil {
|
|
return nil, err
|
|
}
|
|
mediaType, err := normalizeMediaType(mediaType)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if tmdbID <= 0 {
|
|
return nil, fmt.Errorf("%w: tmdb id is required", ErrInvalidInput)
|
|
}
|
|
|
|
raw, err := s.tmdb.GetMediaDetail(ctx, string(mediaType), tmdbID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if raw == nil {
|
|
return nil, ErrNotFound
|
|
}
|
|
|
|
// Deep-linking a detail page must not bypass the discovery rating filter.
|
|
// The guard uses the US-only enforcement certification (GetCertification,
|
|
// cached), NOT raw.ContentRating: the display rating falls back to any
|
|
// country's cert, and a foreign "PG"/"G" is the same string as the US
|
|
// rating, so it would pass the US ladder. ErrNotFound rather than
|
|
// ErrForbidden: a restricted profile shouldn't learn the title exists.
|
|
ceiling, err := s.viewerContentCeiling(ctx, viewer)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if ceiling != "" {
|
|
client, ok := s.tmdb.(TMDBCertificationClient)
|
|
if !ok {
|
|
return nil, fmt.Errorf("requests: tmdb client cannot resolve certifications")
|
|
}
|
|
cert, err := client.GetCertification(ctx, tmdbMediaType(mediaType), tmdbID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !access.RatingAllowed(cert, ceiling) {
|
|
return nil, ErrNotFound
|
|
}
|
|
}
|
|
|
|
policy, err := s.EffectivePolicy(ctx, viewer.UserID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
primaryPresence, err := s.lookupAvailable(ctx, mediaType, []int{raw.ID})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
primaryMatch := primaryPresence[raw.ID]
|
|
primaryRequests, err := s.store.ListActiveByTMDB(ctx, mediaType, []int{raw.ID})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
detail := &MediaDetail{
|
|
MediaType: mediaType,
|
|
TMDBID: raw.ID,
|
|
IMDbID: raw.IMDbID,
|
|
Title: raw.Title,
|
|
OriginalTitle: raw.OriginalTitle,
|
|
Tagline: raw.Tagline,
|
|
Overview: raw.Overview,
|
|
PosterPath: raw.PosterPath,
|
|
BackdropPath: raw.BackdropPath,
|
|
ReleaseDate: raw.ReleaseDate,
|
|
Year: raw.Year,
|
|
Runtime: raw.Runtime,
|
|
Genres: raw.Genres,
|
|
VoteAverage: raw.VoteAverage,
|
|
VoteCount: raw.VoteCount,
|
|
Status: raw.Status,
|
|
Homepage: raw.Homepage,
|
|
ContentRating: raw.ContentRating,
|
|
ProductionCompanies: raw.ProductionCompanies,
|
|
NumberOfSeasons: raw.NumberOfSeasons,
|
|
NumberOfEpisodes: raw.NumberOfEpisodes,
|
|
FirstAirDate: raw.FirstAirDate,
|
|
LastAirDate: raw.LastAirDate,
|
|
Networks: raw.Networks,
|
|
Director: raw.Director,
|
|
Creators: raw.Creators,
|
|
Availability: availabilityValue(primaryMatch.Available),
|
|
LibraryContentID: primaryMatch.ContentID,
|
|
Request: requestStateFor(viewer, policy, primaryMatch.Available, primaryRequests[raw.ID]),
|
|
}
|
|
if raw.TVDBID > 0 {
|
|
tvdb := raw.TVDBID
|
|
detail.TVDBID = &tvdb
|
|
}
|
|
if len(raw.Cast) > 0 {
|
|
detail.Cast = make([]MediaCastMember, 0, len(raw.Cast))
|
|
for _, member := range raw.Cast {
|
|
detail.Cast = append(detail.Cast, MediaCastMember{
|
|
Name: member.Name,
|
|
Character: member.Character,
|
|
ProfilePath: member.ProfilePath,
|
|
Order: member.Order,
|
|
})
|
|
}
|
|
}
|
|
|
|
if len(raw.Recommendations) > 0 {
|
|
recPage := &tmdb.MediaPage{Results: raw.Recommendations}
|
|
enriched, err := s.enrichPageWithCeiling(ctx, viewer, recPage, ceiling)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
detail.Recommendations = enriched.Results
|
|
}
|
|
|
|
return detail, nil
|
|
}
|
|
|
|
func (s *Service) CreateRequest(ctx context.Context, viewer Viewer, input CreateRequestInput) (*Request, error) {
|
|
if err := validateViewer(viewer); err != nil {
|
|
return nil, err
|
|
}
|
|
if s == nil || s.store == nil {
|
|
return nil, fmt.Errorf("request service is not configured")
|
|
}
|
|
if err := s.ensureRequestsEnabled(ctx); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := s.ensureViewerRequestsAllowed(ctx, viewer.UserID); err != nil {
|
|
return nil, err
|
|
}
|
|
normalized, err := normalizeCreateInput(input)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := s.ensureCreateAllowedByCeiling(ctx, viewer, normalized); err != nil {
|
|
return nil, err
|
|
}
|
|
s.enrichExternalIDs(ctx, &normalized)
|
|
isAnime := s.detectRequestAnime(ctx, normalized.MediaType, normalized.TMDBID)
|
|
|
|
matches, err := s.lookupPresence(ctx, normalized.MediaType, []PresenceCandidate{createPresenceCandidate(normalized)})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if matches[normalized.TMDBID].Available {
|
|
return nil, ErrAlreadyAvailable
|
|
}
|
|
|
|
active, err := s.store.ListActiveByTMDB(ctx, normalized.MediaType, []int{normalized.TMDBID})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if active[normalized.TMDBID] != nil {
|
|
return nil, ErrAlreadyRequested
|
|
}
|
|
|
|
// Re-requesting media that previously failed (e.g., transient integration
|
|
// error) should not leave stale failed rows behind in user/admin lists.
|
|
if _, err := s.store.DeleteFailedByTMDB(ctx, normalized.MediaType, normalized.TMDBID); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
policy, err := s.EffectivePolicy(ctx, viewer.UserID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := validateCreatePolicy(policy); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
id, err := idgen.NextID()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
status := StatusPending
|
|
if policy.AutoApprove {
|
|
configured, err := s.integrationConfigured(ctx, normalized.MediaType)
|
|
if err == nil && configured {
|
|
status = StatusApproved
|
|
}
|
|
}
|
|
record := CreateRequestRecord{
|
|
ID: id,
|
|
Input: normalized,
|
|
Status: status,
|
|
Outcome: OutcomeActive,
|
|
IsAnime: isAnime,
|
|
Requester: viewer,
|
|
Now: s.now(),
|
|
}
|
|
if !policy.Unlimited {
|
|
record.Quota = &QuotaCheck{
|
|
UserID: viewer.UserID,
|
|
WindowStart: policy.WindowStart,
|
|
MaxRequests: policy.MaxRequests,
|
|
}
|
|
}
|
|
req, err := s.store.CreateRequest(ctx, record)
|
|
if err != nil {
|
|
if errors.Is(err, ErrAlreadyRequested) {
|
|
return nil, ErrAlreadyRequested
|
|
}
|
|
if errors.Is(err, ErrQuotaExceeded) {
|
|
return nil, QuotaError{
|
|
Used: policy.MaxRequests,
|
|
Limit: policy.MaxRequests,
|
|
WindowDays: policy.WindowDays,
|
|
}
|
|
}
|
|
return nil, err
|
|
}
|
|
s.notifyLifecycle(ctx, *req, LifecycleNotifier.RequestSubmitted)
|
|
if req.Status == StatusApproved {
|
|
// Auto-approval is a real approval transition; channels subscribed to
|
|
// approvals see it alongside the submission.
|
|
s.notifyLifecycle(ctx, *req, LifecycleNotifier.RequestApproved)
|
|
return s.submitApprovedRequest(ctx, *req, viewer, nil)
|
|
}
|
|
return req, nil
|
|
}
|
|
|
|
func (s *Service) ListMine(ctx context.Context, viewer Viewer, filter ListFilter) ([]*Request, error) {
|
|
if viewer.UserID == 0 {
|
|
return nil, ErrForbidden
|
|
}
|
|
if err := s.ensureRequestsEnabled(ctx); err != nil {
|
|
return nil, err
|
|
}
|
|
reqs, err := s.store.ListMine(ctx, viewer.UserID, normalizeListFilter(filter))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := s.attachTargets(ctx, reqs...); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := s.attachLibraryContent(ctx, reqs...); err != nil {
|
|
return nil, err
|
|
}
|
|
return reqs, nil
|
|
}
|
|
|
|
func (s *Service) ListAdmin(ctx context.Context, viewer Viewer, filter ListFilter) ([]*Request, error) {
|
|
if !viewer.IsAdmin {
|
|
return nil, ErrForbidden
|
|
}
|
|
reqs, err := s.store.ListAdmin(ctx, normalizeListFilter(filter))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := s.attachTargets(ctx, reqs...); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := s.attachLibraryContent(ctx, reqs...); err != nil {
|
|
return nil, err
|
|
}
|
|
return reqs, nil
|
|
}
|
|
|
|
// attachTargets loads and attaches the per-instance fulfillment targets for each
|
|
// request so callers (admin queue, detail view) can surface multi-target status.
|
|
func (s *Service) attachTargets(ctx context.Context, reqs ...*Request) error {
|
|
for _, r := range reqs {
|
|
if r == nil {
|
|
continue
|
|
}
|
|
targets, err := s.store.ListTargets(ctx, r.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
r.Targets = targets
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) attachLibraryContent(ctx context.Context, reqs ...*Request) error {
|
|
if s == nil || s.presence == nil || len(reqs) == 0 {
|
|
return nil
|
|
}
|
|
|
|
type requestKey struct {
|
|
mediaType MediaType
|
|
tmdbID int
|
|
}
|
|
|
|
candidatesByType := map[MediaType][]PresenceCandidate{}
|
|
requestsByKey := map[requestKey][]*Request{}
|
|
seen := map[requestKey]bool{}
|
|
for _, req := range reqs {
|
|
if req == nil || req.TMDBID <= 0 {
|
|
continue
|
|
}
|
|
key := requestKey{mediaType: req.MediaType, tmdbID: req.TMDBID}
|
|
requestsByKey[key] = append(requestsByKey[key], req)
|
|
if seen[key] {
|
|
continue
|
|
}
|
|
seen[key] = true
|
|
candidatesByType[req.MediaType] = append(candidatesByType[req.MediaType], requestPresenceCandidate(*req))
|
|
}
|
|
|
|
for mediaType, candidates := range candidatesByType {
|
|
matches, err := s.lookupPresence(ctx, mediaType, candidates)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for tmdbID, match := range matches {
|
|
if !match.Available || strings.TrimSpace(match.ContentID) == "" {
|
|
continue
|
|
}
|
|
for _, req := range requestsByKey[requestKey{mediaType: mediaType, tmdbID: tmdbID}] {
|
|
req.LibraryContentID = match.ContentID
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) GetRequest(ctx context.Context, viewer Viewer, id string) (*Request, error) {
|
|
if err := s.ensureRequestsEnabled(ctx); err != nil {
|
|
return nil, err
|
|
}
|
|
req, err := s.store.GetRequest(ctx, strings.TrimSpace(id))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !viewer.IsAdmin && req.RequestedByUserID != viewer.UserID {
|
|
return nil, ErrForbidden
|
|
}
|
|
if err := s.attachTargets(ctx, req); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := s.attachLibraryContent(ctx, req); err != nil {
|
|
return nil, err
|
|
}
|
|
return req, nil
|
|
}
|
|
|
|
func (s *Service) Approve(ctx context.Context, viewer Viewer, id string) (*Request, error) {
|
|
if !viewer.IsAdmin {
|
|
return nil, ErrForbidden
|
|
}
|
|
req, err := s.store.GetRequest(ctx, strings.TrimSpace(id))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if req.Outcome != OutcomeActive || req.Status != StatusPending {
|
|
return nil, ErrInvalidState
|
|
}
|
|
approved, err := s.store.SetStatus(ctx, req.ID, StatusApproved, viewer)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
s.notifyLifecycle(ctx, *approved, LifecycleNotifier.RequestApproved)
|
|
return s.submitApprovedRequest(ctx, *approved, viewer, nil)
|
|
}
|
|
|
|
func (s *Service) Decline(ctx context.Context, viewer Viewer, id, reason string) (*Request, error) {
|
|
if !viewer.IsAdmin {
|
|
return nil, ErrForbidden
|
|
}
|
|
req, err := s.store.GetRequest(ctx, strings.TrimSpace(id))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
// Approved requests are pending submission by the reconciler; declining
|
|
// while submission may be in flight risks a divergent external state.
|
|
if req.Outcome != OutcomeActive ||
|
|
req.Status == StatusApproved ||
|
|
req.Status == StatusCompleted ||
|
|
req.Status == StatusQueued ||
|
|
req.Status == StatusDownloading ||
|
|
strings.TrimSpace(req.ExternalID) != "" ||
|
|
strings.TrimSpace(req.IntegrationKind) != "" {
|
|
return nil, ErrInvalidState
|
|
}
|
|
declined, err := s.store.SetOutcome(ctx, req.ID, OutcomeDeclined, viewer, reason)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
declined.DeclineReason = strings.TrimSpace(reason)
|
|
s.notifyLifecycle(ctx, *declined, LifecycleNotifier.RequestDeclined)
|
|
return declined, nil
|
|
}
|
|
|
|
// Cancel withdraws a request that has not yet been submitted to a downstream
|
|
// integration. Owners can cancel their own pending requests; admins can cancel
|
|
// any active request that has not entered the fulfillment pipeline. Requests
|
|
// already approved, queued, downloading, or completed cannot be cancelled —
|
|
// callers should decline (admin) or wait for completion in those cases.
|
|
func (s *Service) Cancel(ctx context.Context, viewer Viewer, id, reason string) (*Request, error) {
|
|
if viewer.UserID == 0 {
|
|
return nil, ErrForbidden
|
|
}
|
|
if !viewer.IsAdmin {
|
|
if err := s.ensureRequestsEnabled(ctx); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
req, err := s.store.GetRequest(ctx, strings.TrimSpace(id))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !viewer.IsAdmin && req.RequestedByUserID != viewer.UserID {
|
|
return nil, ErrForbidden
|
|
}
|
|
if req.Outcome != OutcomeActive ||
|
|
req.Status == StatusApproved ||
|
|
req.Status == StatusCompleted ||
|
|
req.Status == StatusQueued ||
|
|
req.Status == StatusDownloading ||
|
|
strings.TrimSpace(req.ExternalID) != "" ||
|
|
strings.TrimSpace(req.IntegrationKind) != "" {
|
|
return nil, ErrInvalidState
|
|
}
|
|
return s.store.SetOutcome(ctx, req.ID, OutcomeCancelled, viewer, reason)
|
|
}
|
|
|
|
func (s *Service) Retry(ctx context.Context, viewer Viewer, id string) (*Request, error) {
|
|
if !viewer.IsAdmin {
|
|
return nil, ErrForbidden
|
|
}
|
|
req, err := s.store.GetRequest(ctx, strings.TrimSpace(id))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if req.Outcome != OutcomeFailed {
|
|
return nil, ErrInvalidState
|
|
}
|
|
if _, err := s.store.SetOutcome(ctx, req.ID, OutcomeActive, viewer, "retry requested"); err != nil {
|
|
return nil, err
|
|
}
|
|
// submitApprovedRequest only re-submits qualities lacking a healthy target, so
|
|
// it is idempotent; gate it on the approved status it expects.
|
|
active, err := s.store.SetStatus(ctx, req.ID, StatusApproved, viewer)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return s.submitApprovedRequest(ctx, *active, viewer, nil)
|
|
}
|
|
|
|
func (s *Service) ReconcileRequests(ctx context.Context, limit int) (ReconcileResult, error) {
|
|
if s == nil || s.store == nil {
|
|
return ReconcileResult{}, fmt.Errorf("request service is not configured")
|
|
}
|
|
if limit <= 0 || limit > 500 {
|
|
limit = 100
|
|
}
|
|
candidates, err := s.store.ListReconciliationCandidates(ctx, limit)
|
|
if err != nil {
|
|
return ReconcileResult{}, err
|
|
}
|
|
fc, err := s.newFulfillContext(ctx)
|
|
if err != nil {
|
|
return ReconcileResult{}, err
|
|
}
|
|
result := ReconcileResult{Checked: len(candidates)}
|
|
for _, req := range candidates {
|
|
if err := ctx.Err(); err != nil {
|
|
return result, err
|
|
}
|
|
change, err := s.reconcileRequest(ctx, *req, fc)
|
|
if err != nil {
|
|
slog.WarnContext(ctx, "request reconcile failed", "component", "requests",
|
|
"request_id", req.ID,
|
|
"media_type", req.MediaType,
|
|
"tmdb_id", req.TMDBID,
|
|
"status", req.Status,
|
|
"integration_kind", req.IntegrationKind,
|
|
"err", err,
|
|
)
|
|
result.Errors++
|
|
continue
|
|
}
|
|
switch change {
|
|
case reconcileSubmitted:
|
|
result.Submitted++
|
|
case reconcileDownloading:
|
|
result.Downloading++
|
|
case reconcileCompleted:
|
|
result.Completed++
|
|
case reconcileFailed:
|
|
result.Failed++
|
|
case reconcileSkipped:
|
|
result.Skipped++
|
|
}
|
|
}
|
|
// Presence-gated fulfillment notifications: completion above (and via the
|
|
// per-target aggregate path) only marks status; the notification fires
|
|
// once the media is confirmed present in the catalog.
|
|
if s.notifier != nil {
|
|
s.notifyFulfilledPending(ctx)
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Service) GetSettings(ctx context.Context, viewer Viewer) (Settings, error) {
|
|
if !viewer.IsAdmin {
|
|
return Settings{}, ErrForbidden
|
|
}
|
|
return s.store.GetSettings(ctx)
|
|
}
|
|
|
|
func (s *Service) GetFeatureStatus(ctx context.Context, _ Viewer) (FeatureStatus, error) {
|
|
settings, err := s.store.GetSettings(ctx)
|
|
if err != nil {
|
|
return FeatureStatus{}, err
|
|
}
|
|
// Rating enforcement is active when the wiring can resolve both a
|
|
// profile ceiling and per-title certifications; with either missing the
|
|
// server behaves like an older version, and clients should know that.
|
|
_, hasRatings := s.entitlements.(ContentRatingResolver)
|
|
_, hasCerts := s.tmdb.(TMDBCertificationClient)
|
|
return FeatureStatus{
|
|
RequestsEnabled: settings.RequestsEnabled,
|
|
RatingRestrictionsEnforced: hasRatings && hasCerts,
|
|
}, nil
|
|
}
|
|
|
|
func (s *Service) ensureRequestsEnabled(ctx context.Context) error {
|
|
settings, err := s.store.GetSettings(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !settings.RequestsEnabled {
|
|
return ErrRequestsDisabled
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) ensureViewerRequestsAllowed(ctx context.Context, userID int) error {
|
|
if s.groupProvider == nil {
|
|
return nil
|
|
}
|
|
group, err := s.groupProvider.GetPolicyForUser(ctx, userID)
|
|
if err != nil {
|
|
return ErrForbidden
|
|
}
|
|
if group != nil && !group.RequestsAllowed {
|
|
return ErrForbidden
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) UpdateSettings(ctx context.Context, viewer Viewer, settings Settings) (Settings, error) {
|
|
if !viewer.IsAdmin {
|
|
return Settings{}, ErrForbidden
|
|
}
|
|
if settings.GlobalMaxRequests < 0 || settings.GlobalWindowDays <= 0 {
|
|
return Settings{}, fmt.Errorf("%w: invalid request settings", ErrInvalidInput)
|
|
}
|
|
return s.store.UpdateSettings(ctx, settings)
|
|
}
|
|
|
|
func (s *Service) GetUserLimit(ctx context.Context, viewer Viewer, userID int) (*UserLimit, error) {
|
|
if !viewer.IsAdmin {
|
|
return nil, ErrForbidden
|
|
}
|
|
if userID <= 0 {
|
|
return nil, fmt.Errorf("%w: invalid user id", ErrInvalidInput)
|
|
}
|
|
limit, err := s.store.GetUserLimit(ctx, userID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if limit != nil {
|
|
return limit, nil
|
|
}
|
|
return &UserLimit{
|
|
UserID: userID,
|
|
LimitMode: LimitModeInherit,
|
|
ApprovalMode: ApprovalModeInherit,
|
|
}, nil
|
|
}
|
|
|
|
func (s *Service) UpsertUserLimit(ctx context.Context, viewer Viewer, limit UserLimit) (*UserLimit, error) {
|
|
if !viewer.IsAdmin {
|
|
return nil, ErrForbidden
|
|
}
|
|
normalized, err := normalizeUserLimit(limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return s.store.UpsertUserLimit(ctx, normalized)
|
|
}
|
|
|
|
func (s *Service) ListIntegrations(ctx context.Context, viewer Viewer) ([]Integration, error) {
|
|
if !viewer.IsAdmin {
|
|
return nil, ErrForbidden
|
|
}
|
|
return s.store.ListIntegrations(ctx)
|
|
}
|
|
|
|
func (s *Service) CreateIntegration(ctx context.Context, viewer Viewer, in Integration) (*Integration, error) {
|
|
if !viewer.IsAdmin {
|
|
return nil, ErrForbidden
|
|
}
|
|
id, err := idgen.NextID()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
in.ID = id
|
|
if err := validateInstance(&in); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := s.validateViaPlugin(ctx, in); err != nil {
|
|
return nil, err
|
|
}
|
|
return s.store.SaveIntegrationWithDefaults(ctx, in, true)
|
|
}
|
|
|
|
func (s *Service) UpdateIntegration(ctx context.Context, viewer Viewer, in Integration) (*Integration, error) {
|
|
if !viewer.IsAdmin {
|
|
return nil, ErrForbidden
|
|
}
|
|
if strings.TrimSpace(in.ID) == "" {
|
|
return nil, fmt.Errorf("%w: integration id required", ErrInvalidInput)
|
|
}
|
|
if err := validateInstance(&in); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := s.validateViaPlugin(ctx, in); err != nil {
|
|
return nil, err
|
|
}
|
|
return s.store.SaveIntegrationWithDefaults(ctx, in, false)
|
|
}
|
|
|
|
// validateViaPlugin asks the bound request_router plugin to validate the
|
|
// connection config on save. Field/form errors are surfaced as *ValidationError
|
|
// so the API layer can render them inline.
|
|
func (s *Service) validateViaPlugin(ctx context.Context, in Integration) error {
|
|
if s.router == nil || in.InstallationID == nil {
|
|
return nil
|
|
}
|
|
// On UPDATE the client omits api_key_ref ("leave blank to keep saved key"),
|
|
// so we would otherwise validate against an empty credential. Mirror
|
|
// LoadIntegrationOptions's backfill: load the stored row by id and reuse the
|
|
// saved (already-decrypted) api key (and BaseURL/PluginConfig if also blank).
|
|
// Nil-safe — a brand-new id has no stored row, so just proceed with what the
|
|
// body carries.
|
|
if strings.TrimSpace(in.APIKeyRef) == "" && strings.TrimSpace(in.ID) != "" {
|
|
stored, err := s.store.GetIntegration(ctx, in.ID)
|
|
if err != nil && !errors.Is(err, ErrNotFound) {
|
|
return err
|
|
}
|
|
if stored != nil {
|
|
// Don't pair a stored API key with a caller-changed base URL: require the
|
|
// key to be re-entered when the server URL changes (defense against
|
|
// exfiltrating a stored, API-unreadable key to an attacker-supplied URL).
|
|
if strings.TrimSpace(in.BaseURL) != "" && strings.TrimSpace(in.BaseURL) != strings.TrimSpace(stored.BaseURL) {
|
|
return &ValidationError{FieldErrors: map[string]string{"api_key_ref": "re-enter the API key when changing the base URL"}}
|
|
}
|
|
in.APIKeyRef = stored.APIKeyRef
|
|
if strings.TrimSpace(in.BaseURL) == "" {
|
|
in.BaseURL = stored.BaseURL
|
|
}
|
|
if in.PluginConfig == nil {
|
|
in.PluginConfig = stored.PluginConfig
|
|
}
|
|
}
|
|
}
|
|
// in.APIKeyRef is the decrypted literal (from the body, or backfilled from the
|
|
// stored row above).
|
|
apiKey := strings.TrimSpace(in.APIKeyRef)
|
|
conn := ResolvedRouterConnection{ID: in.ID, BaseURL: in.BaseURL, APIKey: apiKey, Config: in.PluginConfig}
|
|
siblings, err := s.siblingConnections(ctx, in)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
fe, form, err := s.router.Validate(ctx, *in.InstallationID, in.CapabilityID, conn, siblings)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(fe) > 0 || form != "" {
|
|
return &ValidationError{FieldErrors: fe, FormError: form}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// siblingConnections returns the other connections bound to the same plugin
|
|
// installation as `in` (self excluded), carrying only id + config so a plugin
|
|
// can enforce cross-connection rules without the host resolving sibling
|
|
// credentials.
|
|
func (s *Service) siblingConnections(ctx context.Context, in Integration) ([]ResolvedRouterConnection, error) {
|
|
if in.InstallationID == nil {
|
|
return nil, nil
|
|
}
|
|
all, err := s.store.ListIntegrations(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var out []ResolvedRouterConnection
|
|
for _, other := range all {
|
|
if other.ID == in.ID || other.InstallationID == nil || *other.InstallationID != *in.InstallationID {
|
|
continue
|
|
}
|
|
out = append(out, ResolvedRouterConnection{ID: other.ID, Config: other.PluginConfig})
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (s *Service) DeleteIntegration(ctx context.Context, viewer Viewer, id string) error {
|
|
if !viewer.IsAdmin {
|
|
return ErrForbidden
|
|
}
|
|
return s.store.DeleteIntegration(ctx, strings.TrimSpace(id))
|
|
}
|
|
|
|
func validateInstance(in *Integration) error {
|
|
if strings.TrimSpace(in.Name) == "" {
|
|
return fmt.Errorf("%w: name is required", ErrInvalidInput)
|
|
}
|
|
// capability_id carries the capability SUB-ID ("arr"/"seerr"), not the type:
|
|
// the host resolves the plugin via requireCapability("request_router.v1", id),
|
|
// which keys on (type, id), so storing the type "request_router.v1" here
|
|
// resolves nothing. Matches the scan_source/metadata convention
|
|
// (autoscan_sources.capability_id = "arr"). The bound plugin's Validate RPC is
|
|
// the authority on whether the sub-id names a real capability.
|
|
in.CapabilityID = strings.TrimSpace(in.CapabilityID)
|
|
if in.CapabilityID == "" {
|
|
return fmt.Errorf("%w: capability_id is required", ErrInvalidInput)
|
|
}
|
|
if in.InstallationID == nil {
|
|
return fmt.Errorf("%w: installation_id is required", ErrInvalidInput)
|
|
}
|
|
// The is_default/is_4k/is_default_4k cross-field consistency check is owned by
|
|
// the request_router plugin's Validate RPC, which surfaces it as an inline
|
|
// field error (better UX than a generic host 400). See validateViaPlugin.
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) LoadIntegrationOptions(ctx context.Context, viewer Viewer, integration Integration) (map[string][]RouterOption, error) {
|
|
if !viewer.IsAdmin {
|
|
return nil, ErrForbidden
|
|
}
|
|
// For a saved instance the request body carries only the path id (no creds and
|
|
// often no plugin wiring), so resolve the saved row by id and backfill what the
|
|
// body omitted. This makes "Test connection" reuse the correct per-instance key
|
|
// (each plugin can have multiple connections) instead of borrowing a sibling's.
|
|
if id := strings.TrimSpace(integration.ID); id != "" && id != "new" {
|
|
stored, err := s.store.GetIntegration(ctx, id)
|
|
if err != nil && !errors.Is(err, ErrNotFound) {
|
|
return nil, err
|
|
}
|
|
if stored != nil {
|
|
submittedBaseURL := strings.TrimSpace(integration.BaseURL)
|
|
storedBaseURL := strings.TrimSpace(stored.BaseURL)
|
|
if strings.TrimSpace(integration.BaseURL) == "" {
|
|
integration.BaseURL = stored.BaseURL
|
|
}
|
|
if strings.TrimSpace(integration.APIKeyRef) == "" && (submittedBaseURL == "" || submittedBaseURL == storedBaseURL) {
|
|
integration.APIKeyRef = stored.APIKeyRef
|
|
}
|
|
if strings.TrimSpace(integration.CapabilityID) == "" {
|
|
integration.CapabilityID = stored.CapabilityID
|
|
}
|
|
if integration.InstallationID == nil {
|
|
integration.InstallationID = stored.InstallationID
|
|
}
|
|
if integration.PluginConfig == nil {
|
|
integration.PluginConfig = stored.PluginConfig
|
|
}
|
|
}
|
|
}
|
|
|
|
apiKey := strings.TrimSpace(integration.APIKeyRef)
|
|
if s.router == nil || integration.InstallationID == nil {
|
|
return nil, fmt.Errorf("no fulfillment backend configured")
|
|
}
|
|
conn := ResolvedRouterConnection{ID: integration.ID, BaseURL: integration.BaseURL, APIKey: apiKey, Config: integration.PluginConfig}
|
|
return s.router.ListConfigOptions(ctx, *integration.InstallationID, integration.CapabilityID, conn)
|
|
}
|
|
|
|
func (s *Service) EffectivePolicy(ctx context.Context, userID int) (EffectivePolicy, error) {
|
|
settings, err := s.store.GetSettings(ctx)
|
|
if err != nil {
|
|
return EffectivePolicy{}, err
|
|
}
|
|
limit, err := s.store.GetUserLimit(ctx, userID)
|
|
if err != nil {
|
|
return EffectivePolicy{}, err
|
|
}
|
|
|
|
policy := EffectivePolicy{
|
|
RequestsEnabled: settings.RequestsEnabled,
|
|
MaxRequests: settings.GlobalMaxRequests,
|
|
WindowDays: settings.GlobalWindowDays,
|
|
AutoApprove: settings.GlobalAutoApprovalEnabled,
|
|
}
|
|
if policy.WindowDays <= 0 {
|
|
policy.WindowDays = 7
|
|
}
|
|
if limit != nil {
|
|
switch limit.LimitMode {
|
|
case LimitModeBlocked:
|
|
policy.Blocked = true
|
|
case LimitModeUnlimited:
|
|
policy.Unlimited = true
|
|
case LimitModeCustom:
|
|
if limit.MaxRequests != nil {
|
|
policy.MaxRequests = *limit.MaxRequests
|
|
}
|
|
if limit.WindowDays != nil && *limit.WindowDays > 0 {
|
|
policy.WindowDays = *limit.WindowDays
|
|
}
|
|
}
|
|
switch limit.ApprovalMode {
|
|
case ApprovalModeBlocked:
|
|
policy.Blocked = true
|
|
case ApprovalModeManual:
|
|
policy.AutoApprove = false
|
|
case ApprovalModeAuto:
|
|
policy.AutoApprove = true
|
|
}
|
|
}
|
|
|
|
policy.WindowStart = s.now().AddDate(0, 0, -policy.WindowDays)
|
|
if !policy.Unlimited {
|
|
used, err := s.store.CountUserRequestsSince(ctx, userID, policy.WindowStart)
|
|
if err != nil {
|
|
return EffectivePolicy{}, err
|
|
}
|
|
policy.Used = used
|
|
policy.Remaining = policy.MaxRequests - used
|
|
if policy.Remaining < 0 {
|
|
policy.Remaining = 0
|
|
}
|
|
}
|
|
return policy, nil
|
|
}
|
|
|
|
func (s *Service) enrichPage(ctx context.Context, viewer Viewer, raw *tmdb.MediaPage) (*MediaPage, error) {
|
|
ceiling, err := s.viewerContentCeiling(ctx, viewer)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return s.enrichPageWithCeiling(ctx, viewer, raw, ceiling)
|
|
}
|
|
|
|
// enrichPageWithCeiling is enrichPage for callers that already resolved the
|
|
// viewer's rating ceiling (discovery, browse, detail). The production
|
|
// resolver loads the user, profile, and policy on every call, so resolving
|
|
// once per request instead of again per page matters — DiscoverAll otherwise
|
|
// doubles to 12 resolutions per load.
|
|
func (s *Service) enrichPageWithCeiling(ctx context.Context, viewer Viewer, raw *tmdb.MediaPage, ceiling string) (*MediaPage, error) {
|
|
if raw == nil {
|
|
return &MediaPage{Results: []MediaResult{}}, nil
|
|
}
|
|
var err error
|
|
if ceiling != "" {
|
|
// Filtering before the presence/active-request lookups below means
|
|
// those (and their external-ID hydration) only pay for surviving items.
|
|
raw, err = s.filterPageByCeiling(ctx, raw, ceiling)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
policy, err := s.EffectivePolicy(ctx, viewer.UserID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
idsByType := map[MediaType][]int{}
|
|
for _, item := range raw.Results {
|
|
mediaType, err := normalizeMediaType(MediaType(item.MediaType))
|
|
if err != nil || item.ID <= 0 {
|
|
continue
|
|
}
|
|
idsByType[mediaType] = append(idsByType[mediaType], item.ID)
|
|
}
|
|
|
|
available := map[MediaType]map[int]PresenceMatch{}
|
|
active := map[MediaType]map[int]*Request{}
|
|
for mediaType, ids := range idsByType {
|
|
presence, err := s.lookupAvailable(ctx, mediaType, ids)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
available[mediaType] = presence
|
|
requests, err := s.store.ListActiveByTMDB(ctx, mediaType, ids)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
active[mediaType] = requests
|
|
}
|
|
|
|
out := &MediaPage{
|
|
Page: raw.Page,
|
|
TotalPages: raw.TotalPages,
|
|
TotalResults: raw.TotalResults,
|
|
Results: make([]MediaResult, 0, len(raw.Results)),
|
|
}
|
|
for _, item := range raw.Results {
|
|
mediaType, err := normalizeMediaType(MediaType(item.MediaType))
|
|
if err != nil || item.ID <= 0 {
|
|
continue
|
|
}
|
|
match := available[mediaType][item.ID]
|
|
activeRequest := active[mediaType][item.ID]
|
|
out.Results = append(out.Results, MediaResult{
|
|
MediaType: mediaType,
|
|
TMDBID: item.ID,
|
|
Title: item.Title,
|
|
Year: item.Year,
|
|
Overview: item.Overview,
|
|
PosterPath: item.PosterPath,
|
|
BackdropPath: item.BackdropPath,
|
|
ReleaseDate: item.ReleaseDate,
|
|
Popularity: item.Popularity,
|
|
VoteAverage: item.VoteAverage,
|
|
Availability: availabilityValue(match.Available),
|
|
LibraryContentID: match.ContentID,
|
|
Request: requestStateFor(viewer, policy, match.Available, activeRequest),
|
|
})
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (s *Service) lookupPresence(ctx context.Context, mediaType MediaType, candidates []PresenceCandidate) (map[int]PresenceMatch, error) {
|
|
if s.presence == nil {
|
|
return map[int]PresenceMatch{}, nil
|
|
}
|
|
return s.presence.Lookup(ctx, mediaType, candidates)
|
|
}
|
|
|
|
func requestPresenceCandidate(req Request) PresenceCandidate {
|
|
candidate := PresenceCandidate{
|
|
TMDBID: req.TMDBID,
|
|
IMDbID: strings.TrimSpace(req.IMDbID),
|
|
}
|
|
if req.TVDBID != nil && *req.TVDBID > 0 {
|
|
tvdbID := *req.TVDBID
|
|
candidate.TVDBID = &tvdbID
|
|
}
|
|
return candidate
|
|
}
|
|
|
|
func createPresenceCandidate(input CreateRequestInput) PresenceCandidate {
|
|
candidate := PresenceCandidate{
|
|
TMDBID: input.TMDBID,
|
|
IMDbID: strings.TrimSpace(input.IMDbID),
|
|
}
|
|
if input.TVDBID != nil && *input.TVDBID > 0 {
|
|
tvdbID := *input.TVDBID
|
|
candidate.TVDBID = &tvdbID
|
|
}
|
|
return candidate
|
|
}
|
|
|
|
func (s *Service) hydratePresenceCandidate(ctx context.Context, mediaType MediaType, candidate PresenceCandidate) PresenceCandidate {
|
|
if candidate.TMDBID <= 0 {
|
|
return candidate
|
|
}
|
|
client, ok := s.tmdb.(TMDBExternalIDClient)
|
|
if !ok {
|
|
return candidate
|
|
}
|
|
externalIDs, err := client.GetExternalIDs(ctx, tmdbMediaType(mediaType), candidate.TMDBID)
|
|
if err != nil || externalIDs == nil {
|
|
return candidate
|
|
}
|
|
if candidate.IMDbID == "" {
|
|
candidate.IMDbID = strings.TrimSpace(externalIDs.IMDbID)
|
|
}
|
|
if candidate.TVDBID == nil && externalIDs.TVDBID > 0 {
|
|
tvdbID := externalIDs.TVDBID
|
|
candidate.TVDBID = &tvdbID
|
|
}
|
|
return candidate
|
|
}
|
|
|
|
func (s *Service) hydratePresenceCandidates(ctx context.Context, mediaType MediaType, candidates []PresenceCandidate) []PresenceCandidate {
|
|
if len(candidates) == 0 {
|
|
return candidates
|
|
}
|
|
if _, ok := s.tmdb.(TMDBExternalIDClient); !ok {
|
|
return candidates
|
|
}
|
|
|
|
hydrated := append([]PresenceCandidate(nil), candidates...)
|
|
if externalIDHydrationConcurrency <= 1 {
|
|
for i := range hydrated {
|
|
if ctx.Err() != nil {
|
|
return hydrated
|
|
}
|
|
hydrated[i] = s.hydratePresenceCandidate(ctx, mediaType, hydrated[i])
|
|
}
|
|
return hydrated
|
|
}
|
|
|
|
group, groupCtx := errgroup.WithContext(ctx)
|
|
group.SetLimit(externalIDHydrationConcurrency)
|
|
for i := range hydrated {
|
|
if groupCtx.Err() != nil {
|
|
break
|
|
}
|
|
i := i
|
|
group.Go(func() error {
|
|
if err := groupCtx.Err(); err != nil {
|
|
return err
|
|
}
|
|
hydrated[i] = s.hydratePresenceCandidate(groupCtx, mediaType, hydrated[i])
|
|
return nil
|
|
})
|
|
}
|
|
_ = group.Wait()
|
|
return hydrated
|
|
}
|
|
|
|
func tmdbMediaType(mediaType MediaType) string {
|
|
if mediaType == MediaTypeSeries {
|
|
return "tv"
|
|
}
|
|
return "movie"
|
|
}
|
|
|
|
func (s *Service) lookupAvailable(ctx context.Context, mediaType MediaType, ids []int) (map[int]PresenceMatch, error) {
|
|
if s.presence == nil {
|
|
return map[int]PresenceMatch{}, nil
|
|
}
|
|
candidates := make([]PresenceCandidate, 0, len(ids))
|
|
for _, id := range ids {
|
|
if id > 0 {
|
|
candidates = append(candidates, PresenceCandidate{TMDBID: id})
|
|
}
|
|
}
|
|
candidates = s.hydratePresenceCandidates(ctx, mediaType, candidates)
|
|
matches, err := s.lookupPresence(ctx, mediaType, candidates)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return matches, nil
|
|
}
|
|
|
|
func (s *Service) enrichExternalIDs(ctx context.Context, input *CreateRequestInput) {
|
|
if input == nil {
|
|
return
|
|
}
|
|
client, ok := s.tmdb.(TMDBExternalIDClient)
|
|
if !ok {
|
|
return
|
|
}
|
|
externalIDs, err := client.GetExternalIDs(ctx, tmdbMediaType(input.MediaType), input.TMDBID)
|
|
if err != nil || externalIDs == nil {
|
|
return
|
|
}
|
|
if input.IMDbID == "" {
|
|
input.IMDbID = strings.TrimSpace(externalIDs.IMDbID)
|
|
}
|
|
if input.TVDBID == nil && externalIDs.TVDBID > 0 {
|
|
tvdbID := externalIDs.TVDBID
|
|
input.TVDBID = &tvdbID
|
|
}
|
|
}
|
|
|
|
func (s *Service) detectRequestAnime(ctx context.Context, mediaType MediaType, tmdbID int) bool {
|
|
detail, err := s.tmdb.GetMediaDetail(ctx, tmdbMediaType(mediaType), tmdbID)
|
|
if err != nil || detail == nil {
|
|
return false
|
|
}
|
|
return detectAnime(detail.KeywordIDs)
|
|
}
|
|
|
|
// integrationConfigured reports whether a fulfillment backend exists for the
|
|
// media type, gating auto-approval (pending vs approved). It uses the same
|
|
// router-connection selection as resolveRouterConnections — an enabled
|
|
// request_router.v1 connection with an installation — and additionally honors a
|
|
// connection's declared media-type support so a movie request only auto-approves
|
|
// when a router connection supporting "movie" exists.
|
|
func (s *Service) integrationConfigured(ctx context.Context, mediaType MediaType) (bool, error) {
|
|
instances, err := s.store.ListIntegrations(ctx)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
for _, in := range instances {
|
|
if eligibleRouterConnection(in, mediaType) &&
|
|
strings.TrimSpace(in.BaseURL) != "" && strings.TrimSpace(in.APIKeyRef) != "" {
|
|
return true, nil
|
|
}
|
|
}
|
|
return false, nil
|
|
}
|
|
|
|
// integrationSupportsMediaType reports whether a router connection serves the
|
|
// given media type. An empty SupportedMediaTypes is treated as "supports all".
|
|
func integrationSupportsMediaType(in Integration, mediaType MediaType) bool {
|
|
if len(in.SupportedMediaTypes) == 0 {
|
|
return true
|
|
}
|
|
for _, mt := range in.SupportedMediaTypes {
|
|
if mt == string(mediaType) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (s *Service) submitApprovedRequest(ctx context.Context, req Request, actor Viewer, fc *fulfillContext) (*Request, error) {
|
|
if req.Outcome != OutcomeActive || req.Status != StatusApproved {
|
|
return &req, nil
|
|
}
|
|
if s.router == nil {
|
|
return s.markSubmissionFailed(ctx, req.ID, actor, fmt.Errorf("no fulfillment backend configured"))
|
|
}
|
|
if fc == nil {
|
|
built, err := s.newFulfillContext(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
fc = built
|
|
}
|
|
conns, installationID, capabilityID, err := s.resolveRouterConnections(ctx, fc, req.MediaType)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(conns) == 0 {
|
|
// Distinguish "no backend at all" from the migration breakage where an
|
|
// existing connection row exists but its installation_id is NULL (the row
|
|
// predates the plugin install and was never re-bound).
|
|
msg := "no fulfillment backend configured"
|
|
for _, in := range fc.integrations {
|
|
if in.Enabled && in.CapabilityID != "" && in.InstallationID == nil {
|
|
msg = "request backend connection is not bound to a plugin installation; re-save it in admin"
|
|
break
|
|
}
|
|
}
|
|
return s.markSubmissionFailed(ctx, req.ID, actor, errors.New(msg))
|
|
}
|
|
existing, err := s.store.ListTargets(ctx, req.ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
healthy := map[Quality]bool{}
|
|
for _, t := range existing {
|
|
if t.Status != StatusFailed {
|
|
healthy[t.Quality] = true
|
|
}
|
|
}
|
|
allowed := s.allowedQualities(ctx, req, fc.settings)
|
|
if !fc.settings.ForceDualQuality {
|
|
allowed = filterUnconfiguredOptionalQualities(allowed, conns)
|
|
}
|
|
var want []Quality
|
|
for _, q := range allowed {
|
|
if !healthy[q] {
|
|
want = append(want, q)
|
|
}
|
|
}
|
|
if len(want) == 0 {
|
|
return &req, nil
|
|
}
|
|
for _, t := range existing { // drop stale failed targets for the qualities we re-submit
|
|
if t.Status == StatusFailed {
|
|
for _, q := range want {
|
|
if t.Quality == q {
|
|
if err := s.store.DeleteTarget(ctx, t.ID); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
s.populateRequesterIdentity(ctx, &req)
|
|
targets, msg, err := s.router.Fulfill(ctx, installationID, capabilityID, req, want, conns)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(targets) == 0 {
|
|
if msg == "" {
|
|
msg = "fulfillment backend created no targets"
|
|
}
|
|
return s.markSubmissionFailed(ctx, req.ID, actor, errors.New(msg))
|
|
}
|
|
connKind := connectionKindByID(conns)
|
|
latest := &req
|
|
// The plugin is an out-of-process trust boundary: validate every returned
|
|
// target against the DB CHECK constraints (quality, status) and skip any
|
|
// quality that is duplicated in the batch or already has a healthy target, so
|
|
// a misbehaving plugin can't violate UNIQUE(request_id, quality) and wedge the
|
|
// request.
|
|
validQuality := map[Quality]bool{Quality1080p: true, Quality2160p: true}
|
|
validStatus := map[Status]bool{StatusQueued: true, StatusDownloading: true, StatusCompleted: true, StatusFailed: true}
|
|
returned := map[Quality]bool{}
|
|
for _, rt := range targets {
|
|
if !validQuality[rt.Quality] {
|
|
slog.WarnContext(ctx, "requests: plugin returned unknown quality; skipping", "component", "requests", "request_id", req.ID, "quality", string(rt.Quality))
|
|
continue
|
|
}
|
|
if returned[rt.Quality] || healthy[rt.Quality] {
|
|
continue // dup-in-batch, or a healthy target already exists for this quality
|
|
}
|
|
if rt.ConnectionID != "" {
|
|
if _, ok := connKind[rt.ConnectionID]; !ok {
|
|
slog.WarnContext(ctx, "requests: plugin returned unknown connection id; skipping target", "component", "requests", "request_id", req.ID, "connection_id", rt.ConnectionID)
|
|
continue
|
|
}
|
|
}
|
|
returned[rt.Quality] = true
|
|
created, err := s.store.CreateTarget(ctx, Target{
|
|
RequestID: req.ID, IntegrationID: rt.ConnectionID, IntegrationKind: connKind[rt.ConnectionID],
|
|
Quality: rt.Quality, IsAnime: req.IsAnime, Status: StatusQueued,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
status := rt.Status
|
|
if status == "" || !validStatus[status] {
|
|
status = StatusQueued // coerce unknown/empty status to the DB-valid default
|
|
}
|
|
updated, err := s.store.UpdateTargetStatus(ctx, created.ID, status, rt.ExternalID, rt.ExternalStatus, rt.Message, actor)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if updated != nil {
|
|
latest = updated
|
|
}
|
|
}
|
|
// Any wanted quality the plugin did not fulfill is recorded as a failed target
|
|
// rather than silently dropped, so it stays visible and Retry re-attempts it
|
|
// (a failed target is not "healthy").
|
|
const noTargetMsg = "fulfillment backend returned no target for this quality"
|
|
for _, q := range want {
|
|
if returned[q] {
|
|
continue
|
|
}
|
|
created, err := s.store.CreateTarget(ctx, Target{
|
|
RequestID: req.ID, Quality: q, IsAnime: req.IsAnime, Status: StatusFailed, LastError: noTargetMsg,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
updated, err := s.store.UpdateTargetStatus(ctx, created.ID, StatusFailed, "", "", noTargetMsg, actor)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if updated != nil {
|
|
latest = updated
|
|
}
|
|
}
|
|
return latest, nil
|
|
}
|
|
|
|
// connectionKindByID maps each connection id to its plugin-declared service kind
|
|
// (e.g. "radarr"/"sonarr") from PluginConfig["service_kind"], for the
|
|
// integration_kind column on persisted targets. Missing kinds map to "".
|
|
func connectionKindByID(conns []ResolvedRouterConnection) map[string]string {
|
|
out := make(map[string]string, len(conns))
|
|
for _, c := range conns {
|
|
out[c.ID] = ""
|
|
if c.Config != nil {
|
|
if kind, ok := c.Config["service_kind"].(string); ok {
|
|
out[c.ID] = kind
|
|
}
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
func filterUnconfiguredOptionalQualities(qualities []Quality, conns []ResolvedRouterConnection) []Quality {
|
|
out := make([]Quality, 0, len(qualities))
|
|
for _, q := range qualities {
|
|
if q == Quality2160p && !routerQualityConfigured(q, conns) {
|
|
continue
|
|
}
|
|
out = append(out, q)
|
|
}
|
|
return out
|
|
}
|
|
|
|
func routerQualityConfigured(q Quality, conns []ResolvedRouterConnection) bool {
|
|
usesTieredDefaults := false
|
|
for _, conn := range conns {
|
|
if conn.Config == nil {
|
|
continue
|
|
}
|
|
if hasRouterQualityKey(conn.Config) {
|
|
usesTieredDefaults = true
|
|
}
|
|
if q == Quality2160p && boolConfig(conn.Config, "is_default_4k") {
|
|
return true
|
|
}
|
|
}
|
|
// Generic request_router implementations may not expose arr-style HD/4K
|
|
// default flags. In that case, preserve the host's requested qualities and
|
|
// let the plugin decide what it can fulfill.
|
|
return !usesTieredDefaults
|
|
}
|
|
|
|
func hasRouterQualityKey(config map[string]any) bool {
|
|
for _, key := range []string{"is_default", "is_default_4k", "is_4k"} {
|
|
if _, ok := config[key]; ok {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func boolConfig(config map[string]any, key string) bool {
|
|
v, ok := config[key]
|
|
if !ok {
|
|
return false
|
|
}
|
|
b, ok := v.(bool)
|
|
return ok && b
|
|
}
|
|
|
|
func (s *Service) markSubmissionFailed(ctx context.Context, requestID string, actor Viewer, submitErr error) (*Request, error) {
|
|
failed, err := s.store.SetOutcome(ctx, requestID, OutcomeFailed, actor, submitErr.Error())
|
|
if err != nil {
|
|
return nil, fmt.Errorf("submit request failed: %w; mark failed: %v", submitErr, err)
|
|
}
|
|
return failed, nil
|
|
}
|
|
|
|
type reconcileChange string
|
|
|
|
const (
|
|
reconcileUnchanged reconcileChange = "unchanged"
|
|
reconcileSkipped reconcileChange = "skipped"
|
|
reconcileSubmitted reconcileChange = "submitted"
|
|
reconcileDownloading reconcileChange = "downloading"
|
|
reconcileCompleted reconcileChange = "completed"
|
|
reconcileFailed reconcileChange = "failed"
|
|
)
|
|
|
|
func (s *Service) reconcileRequest(ctx context.Context, req Request, fc *fulfillContext) (reconcileChange, error) {
|
|
completed, err := s.requestAvailable(ctx, req)
|
|
if err != nil {
|
|
return reconcileUnchanged, err
|
|
}
|
|
if completed {
|
|
// The presence check is quality-agnostic (TMDB id only), so it must not
|
|
// force-complete a request whose targets are still in flight — that would
|
|
// orphan in-progress downloads. Only take the shortcut for legacy/no-live
|
|
// -target requests; otherwise let per-target reconcile + aggregate drive
|
|
// completion.
|
|
live, err := s.liveTargets(ctx, req.ID)
|
|
if err != nil {
|
|
return reconcileUnchanged, err
|
|
}
|
|
if len(live) == 0 {
|
|
if req.Status == StatusCompleted {
|
|
return reconcileUnchanged, nil
|
|
}
|
|
if _, err := s.store.SetStatus(ctx, req.ID, StatusCompleted, Viewer{}); err != nil {
|
|
return reconcileUnchanged, err
|
|
}
|
|
return reconcileCompleted, nil
|
|
}
|
|
updated, retired, err := s.retireStalledTargets(ctx, req, live)
|
|
if err != nil {
|
|
return reconcileUnchanged, err
|
|
}
|
|
if retired && updated != nil && updated.Status == StatusCompleted {
|
|
return reconcileCompleted, nil
|
|
}
|
|
}
|
|
|
|
if req.Status == StatusApproved {
|
|
updated, err := s.submitApprovedRequest(ctx, req, Viewer{}, fc)
|
|
if err != nil {
|
|
return reconcileUnchanged, err
|
|
}
|
|
switch {
|
|
case updated.Outcome == OutcomeFailed:
|
|
return reconcileFailed, nil
|
|
case updated.Status == StatusQueued:
|
|
return reconcileSubmitted, nil
|
|
default:
|
|
return reconcileSkipped, nil
|
|
}
|
|
}
|
|
|
|
targets, err := s.store.ListTargets(ctx, req.ID)
|
|
if err != nil {
|
|
return reconcileUnchanged, err
|
|
}
|
|
if s.router == nil {
|
|
return reconcileUnchanged, nil
|
|
}
|
|
conns, installationID, capabilityID, err := s.resolveRouterConnections(ctx, fc, req.MediaType)
|
|
if err != nil {
|
|
return reconcileUnchanged, err
|
|
}
|
|
if len(conns) == 0 {
|
|
return reconcileUnchanged, nil
|
|
}
|
|
|
|
var refs []RouterTargetRef
|
|
for _, t := range targets {
|
|
if t.Status == StatusCompleted || t.Status == StatusFailed {
|
|
continue
|
|
}
|
|
refs = append(refs, RouterTargetRef{Quality: t.Quality, ConnectionID: t.IntegrationID, ExternalID: t.ExternalID})
|
|
}
|
|
if len(refs) == 0 {
|
|
return reconcileUnchanged, nil
|
|
}
|
|
|
|
statuses, err := s.router.CheckStatus(ctx, installationID, capabilityID, req, refs, conns)
|
|
if err != nil {
|
|
return reconcileUnchanged, err
|
|
}
|
|
|
|
change := reconcileUnchanged
|
|
for _, st := range statuses {
|
|
// Match the returned status to the live target by (quality, connection).
|
|
var target *Target
|
|
for i := range targets {
|
|
if targets[i].Quality == st.Quality && targets[i].IntegrationID == st.ConnectionID {
|
|
target = &targets[i]
|
|
break
|
|
}
|
|
}
|
|
if target == nil || target.Status == StatusCompleted || target.Status == StatusFailed {
|
|
continue
|
|
}
|
|
newStatus := st.Status
|
|
if newStatus == "" || newStatus == target.Status {
|
|
continue
|
|
}
|
|
if _, err := s.store.UpdateTargetStatus(ctx, target.ID, newStatus, "", st.ExternalStatus, st.Message, Viewer{}); err != nil {
|
|
return reconcileUnchanged, err
|
|
}
|
|
switch newStatus {
|
|
case StatusCompleted:
|
|
change = reconcileCompleted
|
|
case StatusDownloading:
|
|
if change == reconcileUnchanged {
|
|
change = reconcileDownloading
|
|
}
|
|
case StatusFailed:
|
|
if change == reconcileUnchanged {
|
|
change = reconcileFailed
|
|
}
|
|
}
|
|
}
|
|
return change, nil
|
|
}
|
|
|
|
// liveTargets returns the request's non-terminal (queued or downloading)
|
|
// fulfillment targets.
|
|
func (s *Service) liveTargets(ctx context.Context, requestID string) ([]Target, error) {
|
|
targets, err := s.store.ListTargets(ctx, requestID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var live []Target
|
|
for _, t := range targets {
|
|
if t.Status == StatusQueued || t.Status == StatusDownloading {
|
|
live = append(live, t)
|
|
}
|
|
}
|
|
return live, nil
|
|
}
|
|
|
|
// stalledTargetHorizon is how long a queued target may go without a single
|
|
// status transition before presence-confirmed media is allowed to retire it.
|
|
// Reconciliation runs on a schedule measured in minutes and only writes on a
|
|
// real status change, so a target older than this has had many chances to move
|
|
// and has not.
|
|
const stalledTargetHorizon = 24 * time.Hour
|
|
|
|
// ExternalStatusPresenceConfirmed marks a target closed out by the presence
|
|
// backstop rather than by a router-reported completion. Exported so clients can
|
|
// distinguish "the arr said it finished" from "we found the media ourselves".
|
|
const ExternalStatusPresenceConfirmed = "presence_confirmed"
|
|
|
|
// retireStalledTargets is the backstop for a router that never reports
|
|
// completion. The presence shortcut in reconcileRequest stays disabled for as
|
|
// long as any target looks live, so a router that reports "queued" forever — a
|
|
// buggy plugin, a connection pointing at an instance that no longer tracks the
|
|
// item — pins the request open permanently even though the media is sitting in
|
|
// the library.
|
|
//
|
|
// Targets that are actively downloading are never retired: those are exactly
|
|
// the in-flight downloads the quality-agnostic presence check must not orphan.
|
|
// Only targets stuck in queued past the horizon are closed out, and the request
|
|
// status follows from the usual target aggregate. It returns the request as of
|
|
// the last retirement, if any.
|
|
func (s *Service) retireStalledTargets(ctx context.Context, req Request, live []Target) (*Request, bool, error) {
|
|
cutoff := s.now().Add(-stalledTargetHorizon)
|
|
var updated *Request
|
|
retired := false
|
|
for _, t := range live {
|
|
if t.Status != StatusQueued || t.UpdatedAt.After(cutoff) {
|
|
continue
|
|
}
|
|
next, err := s.store.UpdateTargetStatus(ctx, t.ID, StatusCompleted, "", ExternalStatusPresenceConfirmed, "", Viewer{})
|
|
if err != nil {
|
|
return updated, retired, err
|
|
}
|
|
updated, retired = next, true
|
|
slog.WarnContext(ctx, "requests: retired stalled target on presence", "component", "requests",
|
|
"request_id", req.ID,
|
|
"target_id", t.ID,
|
|
"media_type", req.MediaType,
|
|
"tmdb_id", req.TMDBID,
|
|
"quality", t.Quality,
|
|
"integration_kind", t.IntegrationKind,
|
|
"external_id", t.ExternalID,
|
|
"external_status", t.ExternalStatus,
|
|
"target_updated_at", t.UpdatedAt,
|
|
)
|
|
}
|
|
return updated, retired, nil
|
|
}
|
|
|
|
func (s *Service) requestAvailable(ctx context.Context, req Request) (bool, error) {
|
|
matches, err := s.lookupPresence(ctx, req.MediaType, []PresenceCandidate{requestPresenceCandidate(req)})
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
return matches[req.TMDBID].Available, nil
|
|
}
|
|
|
|
func (s *Service) now() time.Time {
|
|
if s.Now != nil {
|
|
return s.Now()
|
|
}
|
|
return time.Now().UTC()
|
|
}
|
|
|
|
func requestStateFor(viewer Viewer, policy EffectivePolicy, available bool, req *Request) RequestState {
|
|
if req != nil {
|
|
state := RequestState{
|
|
Status: req.Status,
|
|
Requestable: false,
|
|
Reason: "already_requested",
|
|
}
|
|
if viewer.IsAdmin || req.RequestedByUserID == viewer.UserID {
|
|
state.RequestID = req.ID
|
|
}
|
|
return state
|
|
}
|
|
switch {
|
|
case available:
|
|
return RequestState{Requestable: false, Reason: "already_available"}
|
|
case !policy.RequestsEnabled:
|
|
return RequestState{Requestable: false, Reason: "requests_disabled"}
|
|
case policy.Blocked:
|
|
return RequestState{Requestable: false, Reason: "blocked"}
|
|
case !policy.Unlimited && policy.Used >= policy.MaxRequests:
|
|
return RequestState{Requestable: false, Reason: "quota_exceeded"}
|
|
default:
|
|
return RequestState{Requestable: true}
|
|
}
|
|
}
|
|
|
|
func validateCreatePolicy(policy EffectivePolicy) error {
|
|
switch {
|
|
case !policy.RequestsEnabled:
|
|
return ErrRequestsDisabled
|
|
case policy.Blocked:
|
|
return ErrUserBlocked
|
|
case !policy.Unlimited && policy.Used >= policy.MaxRequests:
|
|
return QuotaError{Used: policy.Used, Limit: policy.MaxRequests, WindowDays: policy.WindowDays}
|
|
default:
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func validateViewer(viewer Viewer) error {
|
|
if viewer.UserID == 0 {
|
|
return ErrForbidden
|
|
}
|
|
if strings.TrimSpace(viewer.ProfileID) == "" {
|
|
return fmt.Errorf("%w: profile is required", ErrInvalidInput)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func normalizeCreateInput(input CreateRequestInput) (CreateRequestInput, error) {
|
|
mediaType, err := normalizeMediaType(input.MediaType)
|
|
if err != nil {
|
|
return CreateRequestInput{}, err
|
|
}
|
|
input.MediaType = mediaType
|
|
input.Title = strings.TrimSpace(input.Title)
|
|
input.IMDbID = strings.TrimSpace(input.IMDbID)
|
|
input.Overview = strings.TrimSpace(input.Overview)
|
|
input.PosterPath = strings.TrimSpace(input.PosterPath)
|
|
input.BackdropPath = strings.TrimSpace(input.BackdropPath)
|
|
if input.TMDBID <= 0 {
|
|
return CreateRequestInput{}, fmt.Errorf("%w: tmdb_id is required", ErrInvalidInput)
|
|
}
|
|
if input.Title == "" {
|
|
return CreateRequestInput{}, fmt.Errorf("%w: title is required", ErrInvalidInput)
|
|
}
|
|
return input, nil
|
|
}
|
|
|
|
func normalizeUserLimit(limit UserLimit) (UserLimit, error) {
|
|
if limit.UserID <= 0 {
|
|
return UserLimit{}, fmt.Errorf("%w: invalid user id", ErrInvalidInput)
|
|
}
|
|
switch limit.LimitMode {
|
|
case "", LimitModeInherit:
|
|
limit.LimitMode = LimitModeInherit
|
|
limit.MaxRequests = nil
|
|
limit.WindowDays = nil
|
|
case LimitModeCustom:
|
|
if limit.MaxRequests == nil || limit.WindowDays == nil || *limit.MaxRequests < 0 || *limit.WindowDays <= 0 {
|
|
return UserLimit{}, fmt.Errorf("%w: custom limits require max_requests >= 0 and window_days > 0", ErrInvalidInput)
|
|
}
|
|
case LimitModeUnlimited:
|
|
limit.MaxRequests = nil
|
|
limit.WindowDays = nil
|
|
case LimitModeBlocked:
|
|
limit.MaxRequests = nil
|
|
limit.WindowDays = nil
|
|
default:
|
|
return UserLimit{}, fmt.Errorf("%w: invalid limit mode", ErrInvalidInput)
|
|
}
|
|
switch limit.ApprovalMode {
|
|
case "", ApprovalModeInherit:
|
|
limit.ApprovalMode = ApprovalModeInherit
|
|
case ApprovalModeManual, ApprovalModeAuto, ApprovalModeBlocked:
|
|
default:
|
|
return UserLimit{}, fmt.Errorf("%w: invalid approval mode", ErrInvalidInput)
|
|
}
|
|
return limit, nil
|
|
}
|
|
|
|
func normalizeMediaType(mediaType MediaType) (MediaType, error) {
|
|
switch MediaType(strings.ToLower(strings.TrimSpace(string(mediaType)))) {
|
|
case MediaTypeMovie:
|
|
return MediaTypeMovie, nil
|
|
case MediaTypeSeries, "tv":
|
|
return MediaTypeSeries, nil
|
|
default:
|
|
return "", ErrInvalidMediaType
|
|
}
|
|
}
|
|
|
|
func normalizeSearchMediaType(mediaType MediaType) (MediaType, error) {
|
|
switch MediaType(strings.ToLower(strings.TrimSpace(string(mediaType)))) {
|
|
case "", MediaTypeAll:
|
|
return MediaTypeAll, nil
|
|
case MediaTypeMovie:
|
|
return MediaTypeMovie, nil
|
|
case MediaTypeSeries, "tv":
|
|
return MediaTypeSeries, nil
|
|
default:
|
|
return "", ErrInvalidMediaType
|
|
}
|
|
}
|
|
|
|
const (
|
|
defaultRequestListLimit = 50
|
|
maxRequestListLimit = 100
|
|
)
|
|
|
|
func normalizeListFilter(filter ListFilter) ListFilter {
|
|
if filter.Limit <= 0 {
|
|
filter.Limit = defaultRequestListLimit
|
|
}
|
|
if filter.Limit > maxRequestListLimit {
|
|
filter.Limit = maxRequestListLimit
|
|
}
|
|
if filter.Offset < 0 {
|
|
filter.Offset = 0
|
|
}
|
|
return filter
|
|
}
|
|
|
|
func availabilityValue(available bool) Availability {
|
|
if available {
|
|
return AvailabilityAvailable
|
|
}
|
|
return AvailabilityMissing
|
|
}
|
|
|
|
var discoverySectionOrder = []string{
|
|
"trending_movies",
|
|
"trending_series",
|
|
"popular_movies",
|
|
"popular_series",
|
|
"upcoming_movies",
|
|
"on_air_series",
|
|
}
|
|
|
|
var discoverySectionTitles = map[string]string{
|
|
"trending_movies": "Trending Movies",
|
|
"trending_series": "Trending Series",
|
|
"popular_movies": "Popular Movies",
|
|
"popular_series": "Popular Series",
|
|
"upcoming_movies": "Upcoming Movies",
|
|
"on_air_series": "On Air Series",
|
|
}
|