package markers import ( "context" "errors" "log/slog" "strconv" "strings" "time" "github.com/Silo-Server/silo-server/internal/models" ) // OutcomeStatus values for a contribution attempt, in addition to the provider // SubmissionStatus* values. const ( OutcomeStatusConflict = "conflict" OutcomeStatusError = "error" OutcomeStatusRateLimited = "rate_limited" OutcomeStatusSkipped = "skipped" contributionStatusClaim = "submitting" contributionSubmitLimit = 2 * time.Minute contributionClaimLease = 15 * time.Minute contributionRecordLimit = 5 * time.Second ) // ContributeOptions scopes a contribution run. type ContributeOptions struct { // Provider restricts the run to a single provider id ("" = all enabled). Provider string // Segments restricts the run to specific kinds (nil = every present segment). Segments []MarkerKind // Auto marks a background run: a segment is only contributed when the // provider has contribute_auto_local enabled and the segment is a local // (scanner) intro at or above the provider's confidence threshold. Auto bool } // ContributionOutcome is the per (provider, segment) result of a run. type ContributionOutcome struct { Provider string Segment MarkerKind Status string // pending | accepted | rejected | conflict | error | rate_limited | skipped SubmissionID string Reason string // skip reason or error message RetryAfter time.Duration } // providerConfigReader is the read surface ContributionService needs from the // provider config store (satisfied by *ProviderConfigStore). type providerConfigReader interface { Get(provider string) (ProviderConfig, bool) } // contributionRecorder is the audit surface ContributionService needs // (satisfied by *ContributionStore). type contributionRecorder interface { Claim(ctx context.Context, row ContributionRow, staleAfter time.Duration) (ContributionClaim, bool, error) Record(ctx context.Context, row ContributionRow) error } // ContributionService submits a file's eligible markers to enabled submitter // providers, idempotently and audited. Both the on-demand admin path and the // daily auto-submission task funnel through ContributeFile. type ContributionService struct { registry *Registry resolver ExternalIDResolver config providerConfigReader store contributionRecorder logger *slog.Logger } // NewContributionService constructs the service. func NewContributionService(registry *Registry, resolver ExternalIDResolver, config providerConfigReader, store contributionRecorder, logger *slog.Logger) *ContributionService { if logger == nil { logger = slog.Default() } return &ContributionService{registry: registry, resolver: resolver, config: config, store: store, logger: logger} } type segmentData struct { kind MarkerKind name string start *float64 end *float64 source *string confidence *float64 } func fileSegments(file *models.MediaFile) []segmentData { return []segmentData{ {MarkerKindIntro, "intro", file.IntroStart, file.IntroEnd, file.IntroMarkersSource, file.IntroMarkersConfidence}, {MarkerKindCredits, "credits", file.CreditsStart, file.CreditsEnd, file.CreditsMarkersSource, file.CreditsMarkersConfidence}, {MarkerKindRecap, "recap", file.RecapStart, file.RecapEnd, file.RecapMarkersSource, file.RecapMarkersConfidence}, {MarkerKindPreview, "preview", file.PreviewStart, file.PreviewEnd, file.PreviewMarkersSource, file.PreviewMarkersConfidence}, } } func (s *ContributionService) submitters() []Submitter { if s == nil || s.registry == nil { return nil } var out []Submitter for _, p := range s.registry.Providers() { if sub, ok := p.(Submitter); ok { out = append(out, sub) } } return out } // ContributeFile submits the file's eligible markers and returns one outcome // per (provider, segment) attempted (including explicit skips). func (s *ContributionService) ContributeFile(ctx context.Context, file *models.MediaFile, opts ContributeOptions) ([]ContributionOutcome, error) { if s == nil || file == nil || s.registry == nil || s.resolver == nil || s.config == nil || s.store == nil { return nil, nil } ids, err := s.resolver.ResolveForFile(ctx, file) if err != nil { return nil, err } if !ids.HasAnyID() { return nil, nil } want := func(k MarkerKind) bool { if len(opts.Segments) == 0 { return true } for _, kind := range opts.Segments { if kind == k { return true } } return false } var outcomes []ContributionOutcome for _, sub := range s.submitters() { providerID := sub.ID() if opts.Provider != "" && opts.Provider != providerID { continue } cfg, ok := s.config.Get(providerID) if !ok || !cfg.ContributeEnabled { continue } for _, seg := range fileSegments(file) { if !want(seg.kind) { continue } if outcome, attempted := s.contributeSegment(ctx, sub, providerID, cfg, file, ids, seg, opts.Auto); attempted { outcomes = append(outcomes, outcome) if outcome.Status == OutcomeStatusRateLimited { return outcomes, nil } } } } return outcomes, nil } func (s *ContributionService) contributeSegment( ctx context.Context, sub Submitter, providerID string, cfg ProviderConfig, file *models.MediaFile, ids ExternalIDs, seg segmentData, auto bool, ) (ContributionOutcome, bool) { if seg.start == nil || seg.end == nil { return ContributionOutcome{}, false } source := "" if seg.source != nil { source = *seg.source } // Only ever contribute our own data; never round-trip online markers. if source != models.MarkerSourceScanner && source != models.MarkerSourceManual { return ContributionOutcome{}, false } if auto { if !cfg.ContributeAutoLocal || source != models.MarkerSourceScanner || seg.kind != MarkerKindIntro { return ContributionOutcome{}, false } confidence := 0.0 if seg.confidence != nil { confidence = *seg.confidence } if confidence < cfg.ContributeMinConfidence { return ContributionOutcome{}, false } } if missing := missingRequiredExternalIDs(sub, ids); len(missing) > 0 { return ContributionOutcome{ Provider: providerID, Segment: seg.kind, Status: OutcomeStatusSkipped, Reason: strings.Join(missing, ",") + " id required", }, true } startMs := int64(*seg.start * 1000) endMs := int64(*seg.end * 1000) durMs := int64(file.Duration) * 1000 hash := ContentHash(seg.name, &startMs, &endMs, &durMs, contributionTargetParts(ids)...) row := ContributionRow{ MediaFileID: file.ID, Provider: providerID, SegmentKind: seg.name, Source: source, SubmittedStartMs: &startMs, SubmittedEndMs: &endMs, VideoDurationMs: &durMs, ContentHash: hash, Status: contributionStatusClaim, } claim, claimed, err := s.store.Claim(ctx, row, contributionClaimLease) if err != nil { return ContributionOutcome{Provider: providerID, Segment: seg.kind, Status: OutcomeStatusError, Reason: err.Error()}, true } if !claimed { return ContributionOutcome{Provider: providerID, Segment: seg.kind, Status: OutcomeStatusSkipped, Reason: "already submitted"}, true } row.ID = claim.ID row.ClaimToken = claim.Token startDur := time.Duration(*seg.start * float64(time.Second)) endDur := time.Duration(*seg.end * float64(time.Second)) req := SubmissionRequest{ Kind: ids.Kind, ExternalIDs: ids.AsRequestMap(), SeasonNumber: ids.SeasonNumber, EpisodeNumber: ids.EpisodeNumber, Segment: seg.kind, Start: &startDur, End: &endDur, Duration: time.Duration(file.Duration) * time.Second, } submitCtx, cancel := context.WithTimeout(ctx, contributionSubmitLimit) result, err := sub.SubmitMarker(submitCtx, req) cancel() if err != nil { msg := err.Error() row.Error = &msg var conflict *SubmissionConflictError if errors.As(err, &conflict) && conflict != nil { row.Status = OutcomeStatusConflict if conflict.HTTPStatus > 0 { status := conflict.HTTPStatus row.HTTPStatus = &status } } else { row.Status = OutcomeStatusError } s.recordContribution(ctx, row) if row.Status == OutcomeStatusConflict { return ContributionOutcome{Provider: providerID, Segment: seg.kind, Status: OutcomeStatusConflict, Reason: msg}, true } if after, ok := RetryAfter(err); ok { return ContributionOutcome{Provider: providerID, Segment: seg.kind, Status: OutcomeStatusRateLimited, Reason: msg, RetryAfter: after}, true } return ContributionOutcome{Provider: providerID, Segment: seg.kind, Status: OutcomeStatusError, Reason: msg}, true } row.Status = result.Status if row.Status == "" { row.Status = SubmissionStatusPending } if result.ID != "" { id := result.ID row.SubmissionID = &id } s.recordContribution(ctx, row) return ContributionOutcome{Provider: providerID, Segment: seg.kind, Status: row.Status, SubmissionID: result.ID}, true } func (s *ContributionService) recordContribution(ctx context.Context, row ContributionRow) { recordCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), contributionRecordLimit) defer cancel() if err := s.store.Record(recordCtx, row); err != nil { s.logger.WarnContext(recordCtx, "record contribution failed", "file_id", row.MediaFileID, "provider", row.Provider, "segment", row.SegmentKind, "status", row.Status, "error", err) } } func contributionTargetParts(ids ExternalIDs) []string { return []string{ itemTypeName(ids.Kind), ids.TmdbID, ids.ImdbID, ids.TvdbID, intPart(ids.SeasonNumber), intPart(ids.EpisodeNumber), } } func intPart(v int) string { if v == 0 { return "" } return strconv.Itoa(v) } func missingRequiredExternalIDs(sub Submitter, ids ExternalIDs) []string { reqProvider, ok := sub.(SubmissionRequirementProvider) if !ok { return nil } reqs := reqProvider.SubmissionRequirements() if len(reqs.RequiredExternalIDs) == 0 { return nil } values := ids.AsRequestMap() var missing []string for _, key := range reqs.RequiredExternalIDs { if strings.TrimSpace(values[key]) == "" { missing = append(missing, key) } } return missing }