diff --git a/internal/ebooks/enrichment.go b/internal/ebooks/enrichment.go index 02130497..791d9067 100644 --- a/internal/ebooks/enrichment.go +++ b/internal/ebooks/enrichment.go @@ -33,6 +33,8 @@ const ( defaultEnrichBatchSize = 50 defaultEnrichWorkers = 4 + + defaultEnrichmentItemTimeout = 2 * time.Minute ) // errEnrichmentSkipped preserves the direct helper contract for an item that @@ -58,33 +60,35 @@ func ebookEnrichWorkers() int { } type enrichmentItemRow struct { - ContentID string - Title string - Year int - FolderID int - Language string - Author string - ProviderIDs map[string]string - Status string - Overview string - ReleaseDate string - Genres []string - Studios []string - PosterPath string + ContentID string + Title string + Year int + FolderID int + Language string + Author string + ProviderIDs map[string]string + Status string + Overview string + Tagline string + ContentRating string + Runtime int + ReleaseDate string + Genres []string + Studios []string + PosterPath string + BackdropPath string + LogoPath string + LockedFields []int + ProtectedFields []string } type enrichmentQueue interface { MaterializeCandidates(ctx context.Context) error ClaimBatch(ctx context.Context, limit int, leaseDuration time.Duration) ([]EnrichmentJob, error) - Complete(ctx context.Context, contentID string, outcome EnrichmentOutcome, refreshAfter time.Duration) error - Fail(ctx context.Context, contentID string, errorClass EnrichmentErrorClass, message string, retryAfter time.Duration) error - Release(ctx context.Context, contentID string) error -} - -type claimAwareEnrichmentQueue interface { - CompleteClaim(ctx context.Context, job EnrichmentJob, outcome EnrichmentOutcome, refreshAfter time.Duration) error - FailClaim(ctx context.Context, job EnrichmentJob, errorClass EnrichmentErrorClass, message string, retryAfter time.Duration) error - ReleaseClaim(ctx context.Context, job EnrichmentJob) error + Complete(ctx context.Context, job EnrichmentJob, outcome EnrichmentOutcome, refreshAfter time.Duration) error + Fail(ctx context.Context, job EnrichmentJob, errorClass EnrichmentErrorClass, message string, retryAfter time.Duration) error + Release(ctx context.Context, job EnrichmentJob) error + Discard(ctx context.Context, job EnrichmentJob) error } // Enricher drives the ebook metadata enrichment sweep. @@ -100,6 +104,7 @@ type Enricher struct { workLinker literaryWorkLinker batchSize int workers int + itemTimeout time.Duration queue enrichmentQueue loadClaimedItemsFn func(context.Context, []EnrichmentJob) ([]enrichmentItemRow, error) @@ -214,13 +219,16 @@ func (e *Enricher) runQueueBatch( for _, job := range jobs { claimedJobs[job.ContentID] = job } + var transitionErrs []error loaded := make(map[string]struct{}, len(items)) for i := range items { loaded[items[i].ContentID] = struct{}{} } for _, job := range jobs { if _, ok := loaded[job.ContentID]; !ok { - _ = e.releaseJob(queue, job) + if err := e.discardJob(queue, job); err != nil && !errors.Is(err, ErrEnrichmentLeaseLost) { + transitionErrs = append(transitionErrs, fmt.Errorf("%s: %w", job.ContentID, err)) + } } } @@ -232,15 +240,18 @@ func (e *Enricher) runQueueBatch( workers = len(items) } if workers == 0 { - return 0, nil + return 0, errors.Join(transitionErrs...) + } + itemTimeout := e.itemTimeout + if itemTimeout <= 0 || itemTimeout >= defaultEnrichmentLease { + itemTimeout = defaultEnrichmentItemTimeout } ch := make(chan enrichmentItemRow, workers) var ( - wg sync.WaitGroup - enriched int64 - transitionMu sync.Mutex - transitionErrs []error + wg sync.WaitGroup + enriched int64 + transitionMu sync.Mutex ) recordTransitionError := func(contentID string, err error) { if err == nil || errors.Is(err, ErrEnrichmentLeaseLost) { @@ -262,8 +273,12 @@ func (e *Enricher) runQueueBatch( continue } - outcome, enrichErr := enrichFn(ctx, item) - if ctx.Err() != nil || errors.Is(enrichErr, context.Canceled) || errors.Is(enrichErr, context.DeadlineExceeded) { + itemCtx, cancelItem := context.WithTimeout(ctx, itemTimeout) + outcome, enrichErr := enrichFn(itemCtx, item) + itemCtxErr := itemCtx.Err() + cancelItem() + if ctx.Err() != nil || itemCtxErr != nil || + errors.Is(enrichErr, context.Canceled) || errors.Is(enrichErr, context.DeadlineExceeded) { recordTransitionError(item.ContentID, e.releaseJob(queue, job)) continue } @@ -330,10 +345,7 @@ func (e *Enricher) completeJob( job EnrichmentJob, outcome EnrichmentOutcome, ) error { - if claimQueue, ok := queue.(claimAwareEnrichmentQueue); ok { - return claimQueue.CompleteClaim(ctx, job, outcome, enrichmentRefreshHorizon(outcome)) - } - return queue.Complete(ctx, job.ContentID, outcome, enrichmentRefreshHorizon(outcome)) + return queue.Complete(ctx, job, outcome, enrichmentRefreshHorizon(outcome)) } func (e *Enricher) failJob( @@ -344,19 +356,19 @@ func (e *Enricher) failJob( message string, retryAfter time.Duration, ) error { - if claimQueue, ok := queue.(claimAwareEnrichmentQueue); ok { - return claimQueue.FailClaim(ctx, job, errorClass, message, retryAfter) - } - return queue.Fail(ctx, job.ContentID, errorClass, message, retryAfter) + return queue.Fail(ctx, job, errorClass, message, retryAfter) } func (e *Enricher) releaseJob(queue enrichmentQueue, job EnrichmentJob) error { releaseCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - if claimQueue, ok := queue.(claimAwareEnrichmentQueue); ok { - return claimQueue.ReleaseClaim(releaseCtx, job) - } - return queue.Release(releaseCtx, job.ContentID) + return queue.Release(releaseCtx, job) +} + +func (e *Enricher) discardJob(queue enrichmentQueue, job EnrichmentJob) error { + discardCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + return queue.Discard(discardCtx, job) } func (e *Enricher) runBatch( @@ -441,10 +453,16 @@ var loadEnrichmentItemsQuery = ` ) AS author, COALESCE(mi.status, ''), COALESCE(mi.overview, ''), - COALESCE(mi.release_date, ''), + COALESCE(mi.tagline, ''), + COALESCE(mi.content_rating, ''), + COALESCE(mi.runtime, 0), + COALESCE(mi.release_date::text, ''), COALESCE(mi.genres, '{}'), COALESCE(mi.studios, '{}'), - COALESCE(mi.poster_path, '') + COALESCE(mi.poster_path, ''), + COALESCE(mi.backdrop_path, ''), + COALESCE(mi.logo_path, ''), + COALESCE(mi.locked_fields, '{}'::integer[]) FROM unnest($1::text[]) WITH ORDINALITY AS claimed(content_id, position) JOIN media_items mi ON mi.content_id = claimed.content_id LEFT JOIN LATERAL ( @@ -462,8 +480,10 @@ var loadEnrichmentItemsQuery = ` func (e *Enricher) loadClaimedItems(ctx context.Context, jobs []EnrichmentJob) ([]enrichmentItemRow, error) { contentIDs := make([]string, 0, len(jobs)) + claimedJobs := make(map[string]EnrichmentJob, len(jobs)) for _, job := range jobs { contentIDs = append(contentIDs, job.ContentID) + claimedJobs[job.ContentID] = job } rows, err := e.pool.Query(ctx, loadEnrichmentItemsQuery, contentIDs) if err != nil { @@ -483,13 +503,20 @@ func (e *Enricher) loadClaimedItems(ctx context.Context, jobs []EnrichmentJob) ( &item.Author, &item.Status, &item.Overview, + &item.Tagline, + &item.ContentRating, + &item.Runtime, &item.ReleaseDate, &item.Genres, &item.Studios, &item.PosterPath, + &item.BackdropPath, + &item.LogoPath, + &item.LockedFields, ); err != nil { return nil, fmt.Errorf("scanning ebook enrichment row: %w", err) } + item.ProtectedFields = append([]string(nil), claimedJobs[item.ContentID].ProtectedFields...) items = append(items, item) } if err := rows.Err(); err != nil { @@ -604,35 +631,75 @@ func preserveEbookLocalMetadata(item enrichmentItemRow, result *metadata.Metadat if result == nil { return } - if item.PosterPath != "" && !ebookPosterOwnedByRemoteProvider(item.PosterPath) { + protected := make(map[string]struct{}, len(item.ProtectedFields)) + for _, field := range item.ProtectedFields { + protected[strings.ToLower(strings.TrimSpace(field))] = struct{}{} + } + isProtected := func(field string) bool { + _, ok := protected[field] + return ok + } + isLocked := func(field metadata.MetadataField) bool { + for _, locked := range item.LockedFields { + if locked == int(field) { + return true + } + } + return false + } + + if isProtected("title") || isLocked(metadata.FieldName) { + result.Title = "" + result.OriginalTitle = "" + result.SortTitle = "" + } + if isProtected("year") || isLocked(metadata.FieldReleaseDates) { + result.Year = 0 + } + if isProtected("overview") || isLocked(metadata.FieldOverview) { + result.Overview = "" + } + if isProtected("tagline") { + result.Tagline = "" + } + if isProtected("content_rating") || isLocked(metadata.FieldContentRating) { + result.ContentRating = "" + } + if isProtected("runtime") || isLocked(metadata.FieldRuntime) { + result.Runtime = 0 + } + if isProtected("release_date") || isLocked(metadata.FieldReleaseDates) { + result.ReleaseDate = "" + } + if isProtected("genres") || isLocked(metadata.FieldGenres) { + result.Genres = nil + } + if isProtected("studios") || isLocked(metadata.FieldStudios) { + result.Studios = nil + } + if isProtected("authors") || isLocked(metadata.FieldCrew) || isLocked(metadata.FieldCast) { + result.People = nil + } + + imagesLocked := isLocked(metadata.FieldImages) + if imagesLocked || isProtected("poster_path") || + (item.PosterPath != "" && !ebookArtworkOwnedByRemoteProvider(item.PosterPath)) { result.PosterPath = "" result.PosterThumbhash = "" } - if !strings.EqualFold(strings.TrimSpace(item.Status), "pending") { - return + if imagesLocked || isProtected("backdrop_path") || + (item.BackdropPath != "" && !ebookArtworkOwnedByRemoteProvider(item.BackdropPath)) { + result.BackdropPath = "" + result.BackdropThumbhash = "" } - if item.Year > 0 { - result.Year = 0 - } - if item.Overview != "" { - result.Overview = "" - } - if item.ReleaseDate != "" { - result.ReleaseDate = "" - } - if len(item.Genres) > 0 { - result.Genres = nil - } - if len(item.Studios) > 0 { - result.Studios = nil - } - if item.Author != "" { - result.People = nil + if imagesLocked || isProtected("logo_path") || + (item.LogoPath != "" && !ebookArtworkOwnedByRemoteProvider(item.LogoPath)) { + result.LogoPath = "" } } -func ebookPosterOwnedByRemoteProvider(path string) bool { - path = strings.TrimSpace(path) +func ebookArtworkOwnedByRemoteProvider(path string) bool { + path = strings.ToLower(strings.TrimSpace(path)) return isRemoteHTTPImage(path) || strings.HasPrefix(path, ebookMetadataImageProviderID+"/ebooks/") } diff --git a/internal/ebooks/enrichment_queue.go b/internal/ebooks/enrichment_queue.go index d25624b3..742c7c94 100644 --- a/internal/ebooks/enrichment_queue.go +++ b/internal/ebooks/enrichment_queue.go @@ -5,7 +5,6 @@ import ( "errors" "fmt" "strings" - "sync" "time" "github.com/jackc/pgx/v5/pgxpool" @@ -39,29 +38,92 @@ const ( var ErrEnrichmentLeaseLost = errors.New("ebook enrichment lease lost") type EnrichmentJob struct { - ContentID string - Attempts int - LastAttemptAt time.Time + ContentID string + Token string + Attempts int + LastAttemptAt time.Time + ProtectedFields []string } type EnrichmentQueue struct { pool *pgxpool.Pool - - claimsMu sync.Mutex - claims map[string]EnrichmentJob } func NewEnrichmentQueue(pool *pgxpool.Pool) *EnrichmentQueue { return &EnrichmentQueue{pool: pool} } +const ebookProtectedFieldsSQL = ` + ARRAY_REMOVE(ARRAY[ + CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.title, '')) <> '' THEN 'title' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND COALESCE(mi.year, 0) > 0 THEN 'year' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.overview, '')) <> '' THEN 'overview' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.tagline, '')) <> '' THEN 'tagline' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.content_rating, '')) <> '' THEN 'content_rating' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND COALESCE(mi.runtime, 0) > 0 THEN 'runtime' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.release_date::text, '')) <> '' THEN 'release_date' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND cardinality(COALESCE(mi.genres, '{}'::text[])) > 0 THEN 'genres' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND cardinality(COALESCE(mi.studios, '{}'::text[])) > 0 THEN 'studios' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND EXISTS ( + SELECT 1 FROM item_people ip WHERE ip.content_id = mi.content_id AND ip.kind = 7 + ) THEN 'authors' END, + CASE WHEN trim(COALESCE(mi.poster_path, '')) <> '' + AND lower(trim(mi.poster_path)) NOT LIKE 'http://%' + AND lower(trim(mi.poster_path)) NOT LIKE 'https://%' + AND lower(trim(mi.poster_path)) NOT LIKE 'ebook-metadata/ebooks/%' + THEN 'poster_path' END, + CASE WHEN trim(COALESCE(mi.backdrop_path, '')) <> '' + AND lower(trim(mi.backdrop_path)) NOT LIKE 'http://%' + AND lower(trim(mi.backdrop_path)) NOT LIKE 'https://%' + AND lower(trim(mi.backdrop_path)) NOT LIKE 'ebook-metadata/ebooks/%' + THEN 'backdrop_path' END, + CASE WHEN trim(COALESCE(mi.logo_path, '')) <> '' + AND lower(trim(mi.logo_path)) NOT LIKE 'http://%' + AND lower(trim(mi.logo_path)) NOT LIKE 'https://%' + AND lower(trim(mi.logo_path)) NOT LIKE 'ebook-metadata/ebooks/%' + THEN 'logo_path' END + ]::text[], NULL) +` + +const mergeEbookProtectedFieldsSQL = ` + ARRAY( + SELECT DISTINCT field + FROM unnest( + ebook_enrichment_state.protected_fields || + ARRAY( + SELECT candidate + FROM unnest(EXCLUDED.protected_fields) AS candidate + WHERE candidate IN ('poster_path', 'backdrop_path', 'logo_path') + ) + ) AS field + ORDER BY field + ) +` + var enqueueEnrichmentJobQuery = ` INSERT INTO ebook_enrichment_state ( - content_id, status, priority, next_attempt_at, updated_at + content_id, status, priority, next_attempt_at, protected_fields, updated_at ) - VALUES ($1, 'pending', $2, now(), now()) + SELECT mi.content_id, 'pending', $2, now(), ` + ebookProtectedFieldsSQL + `, now() + FROM media_items mi + WHERE mi.content_id = $1 + AND mi.type = 'ebook' + AND ` + catalog.MangaChapterExclusionWhere("mi") + ` ON CONFLICT (content_id) DO UPDATE SET - priority = GREATEST(ebook_enrichment_state.priority, EXCLUDED.priority), + status = CASE + WHEN ebook_enrichment_state.status = 'running' THEN ebook_enrichment_state.status + ELSE 'pending' + END, + priority = CASE + WHEN ebook_enrichment_state.status = 'running' THEN ebook_enrichment_state.priority + ELSE GREATEST(ebook_enrichment_state.priority, EXCLUDED.priority) + END, + next_attempt_at = CASE + WHEN ebook_enrichment_state.status = 'running' THEN ebook_enrichment_state.next_attempt_at + ELSE now() + END, + requeue_requested = ebook_enrichment_state.requeue_requested OR ebook_enrichment_state.status = 'running', + protected_fields = ` + mergeEbookProtectedFieldsSQL + `, updated_at = now() ` @@ -78,14 +140,54 @@ func (q *EnrichmentQueue) Enqueue(ctx context.Context, contentID string, priorit var materializeEnrichmentJobsQuery = ` INSERT INTO ebook_enrichment_state ( - content_id, status, priority, next_attempt_at, updated_at + content_id, status, priority, next_attempt_at, completed_at, protected_fields, updated_at ) - SELECT mi.content_id, 'pending', 100, now(), now() + SELECT + mi.content_id, + 'pending', + CASE WHEN mi.last_refreshed IS NULL THEN 100 ELSE 0 END, + CASE + WHEN mi.last_refreshed IS NULL THEN now() + ELSE GREATEST(mi.last_refreshed + interval '90 days', now()) + END, + mi.last_refreshed, + ` + ebookProtectedFieldsSQL + `, + now() FROM media_items mi WHERE mi.type = 'ebook' AND ` + catalog.MangaChapterExclusionWhere("mi") + ` - AND mi.last_refreshed IS NULL - ON CONFLICT (content_id) DO NOTHING + ON CONFLICT (content_id) DO UPDATE SET + status = CASE + WHEN ebook_enrichment_state.status = 'running' THEN ebook_enrichment_state.status + WHEN ebook_enrichment_state.status = 'discarded' THEN 'pending' + ELSE ebook_enrichment_state.status + END, + priority = CASE + WHEN ebook_enrichment_state.status = 'running' THEN ebook_enrichment_state.priority + WHEN ebook_enrichment_state.status = 'discarded' THEN EXCLUDED.priority + WHEN EXCLUDED.completed_at IS NULL AND ebook_enrichment_state.completed_at IS NOT NULL THEN 100 + ELSE ebook_enrichment_state.priority + END, + next_attempt_at = CASE + WHEN ebook_enrichment_state.status = 'running' THEN ebook_enrichment_state.next_attempt_at + WHEN ebook_enrichment_state.status = 'discarded' THEN EXCLUDED.next_attempt_at + WHEN EXCLUDED.completed_at IS NULL AND ebook_enrichment_state.completed_at IS NOT NULL THEN now() + ELSE ebook_enrichment_state.next_attempt_at + END, + completed_at = CASE + WHEN ebook_enrichment_state.status = 'running' THEN ebook_enrichment_state.completed_at + WHEN ebook_enrichment_state.status = 'discarded' THEN EXCLUDED.completed_at + WHEN EXCLUDED.completed_at IS NULL AND ebook_enrichment_state.completed_at IS NOT NULL THEN NULL + ELSE ebook_enrichment_state.completed_at + END, + outcome = CASE + WHEN ebook_enrichment_state.status = 'discarded' + OR (EXCLUDED.completed_at IS NULL AND ebook_enrichment_state.completed_at IS NOT NULL) + THEN NULL + ELSE ebook_enrichment_state.outcome + END, + protected_fields = ` + mergeEbookProtectedFieldsSQL + `, + updated_at = now() ` func (q *EnrichmentQueue) MaterializeCandidates(ctx context.Context) error { @@ -102,19 +204,24 @@ var claimEnrichmentJobsQuery = ` FROM ebook_enrichment_state WHERE next_attempt_at <= now() AND (status = 'pending' OR (status = 'running' AND lease_until < now())) - ORDER BY priority DESC, next_attempt_at, updated_at + ORDER BY + (priority + FLOOR(EXTRACT(EPOCH FROM (now() - next_attempt_at)) / 3600)::integer) DESC, + priority DESC, + next_attempt_at, + updated_at FOR UPDATE SKIP LOCKED LIMIT $1 ) UPDATE ebook_enrichment_state state SET status = 'running', lease_until = now() + $2::interval, + claim_token = gen_random_uuid()::text, last_attempt_at = now(), attempts = attempts + 1, updated_at = now() FROM candidates WHERE state.content_id = candidates.content_id - RETURNING state.content_id, state.attempts, state.last_attempt_at + RETURNING state.content_id, state.claim_token, state.attempts, state.last_attempt_at, state.protected_fields ` func (q *EnrichmentQueue) ClaimBatch(ctx context.Context, limit int, leaseDuration time.Duration) ([]EnrichmentJob, error) { @@ -137,11 +244,10 @@ func (q *EnrichmentQueue) ClaimBatch(ctx context.Context, limit int, leaseDurati jobs := make([]EnrichmentJob, 0, limit) for rows.Next() { var job EnrichmentJob - if err := rows.Scan(&job.ContentID, &job.Attempts, &job.LastAttemptAt); err != nil { + if err := rows.Scan(&job.ContentID, &job.Token, &job.Attempts, &job.LastAttemptAt, &job.ProtectedFields); err != nil { return nil, err } jobs = append(jobs, job) - q.rememberClaim(job) } if err := rows.Err(); err != nil { return nil, err @@ -153,36 +259,25 @@ var completeEnrichmentJobQuery = ` UPDATE ebook_enrichment_state SET status = 'pending', lease_until = NULL, + claim_token = NULL, completed_at = now(), - next_attempt_at = now() + $3::interval, + next_attempt_at = CASE + WHEN requeue_requested THEN now() + ELSE now() + $3::interval + END, outcome = $2, attempts = 0, - priority = GREATEST(priority, 0), + priority = CASE WHEN requeue_requested THEN 100 ELSE 0 END, + requeue_requested = false, last_error_class = NULL, last_error = NULL, updated_at = now() WHERE content_id = $1 AND status = 'running' - AND last_attempt_at = $4 + AND claim_token = $4 ` func (q *EnrichmentQueue) Complete( - ctx context.Context, - contentID string, - outcome EnrichmentOutcome, - refreshAfter time.Duration, -) error { - if q == nil || q.pool == nil { - return errors.New("ebook enrichment queue is not configured") - } - job, ok := q.claimedJob(contentID) - if !ok { - return ErrEnrichmentLeaseLost - } - return q.CompleteClaim(ctx, job, outcome, refreshAfter) -} - -func (q *EnrichmentQueue) CompleteClaim( ctx context.Context, job EnrichmentJob, outcome EnrichmentOutcome, @@ -191,6 +286,9 @@ func (q *EnrichmentQueue) CompleteClaim( if q == nil || q.pool == nil { return errors.New("ebook enrichment queue is not configured") } + if job.ContentID == "" || job.Token == "" { + return ErrEnrichmentLeaseLost + } if refreshAfter <= 0 { refreshAfter = enrichmentRefreshHorizon(outcome) } @@ -204,16 +302,14 @@ func (q *EnrichmentQueue) CompleteClaim( job.ContentID, string(outcome), postgresInterval(refreshAfter), - job.LastAttemptAt, + job.Token, ) if err != nil { return err } if tag.RowsAffected() == 0 { - q.forgetClaim(job) return ErrEnrichmentLeaseLost } - q.forgetClaim(job) return nil } @@ -221,34 +317,23 @@ var failEnrichmentJobQuery = ` UPDATE ebook_enrichment_state SET status = 'pending', lease_until = NULL, - next_attempt_at = now() + $4::interval, + claim_token = NULL, + next_attempt_at = CASE + WHEN requeue_requested THEN now() + ELSE now() + $4::interval + END, + priority = CASE WHEN requeue_requested THEN 100 ELSE priority END, + requeue_requested = false, outcome = 'failed', last_error_class = $2, last_error = $3, updated_at = now() WHERE content_id = $1 AND status = 'running' - AND last_attempt_at = $5 + AND claim_token = $5 ` func (q *EnrichmentQueue) Fail( - ctx context.Context, - contentID string, - errorClass EnrichmentErrorClass, - message string, - retryAfter time.Duration, -) error { - if q == nil || q.pool == nil { - return errors.New("ebook enrichment queue is not configured") - } - job, ok := q.claimedJob(contentID) - if !ok { - return ErrEnrichmentLeaseLost - } - return q.FailClaim(ctx, job, errorClass, message, retryAfter) -} - -func (q *EnrichmentQueue) FailClaim( ctx context.Context, job EnrichmentJob, errorClass EnrichmentErrorClass, @@ -258,6 +343,9 @@ func (q *EnrichmentQueue) FailClaim( if q == nil || q.pool == nil { return errors.New("ebook enrichment queue is not configured") } + if job.ContentID == "" || job.Token == "" { + return ErrEnrichmentLeaseLost + } delay := enrichmentRetryDelay(errorClass, job.Attempts, retryAfter) tag, err := q.pool.Exec( ctx, @@ -266,16 +354,14 @@ func (q *EnrichmentQueue) FailClaim( string(errorClass), message, postgresInterval(delay), - job.LastAttemptAt, + job.Token, ) if err != nil { return err } if tag.RowsAffected() == 0 { - q.forgetClaim(job) return ErrEnrichmentLeaseLost } - q.forgetClaim(job) return nil } @@ -283,62 +369,67 @@ var releaseEnrichmentJobQuery = ` UPDATE ebook_enrichment_state SET status = 'pending', lease_until = NULL, + claim_token = NULL, attempts = GREATEST(attempts - 1, 0), + next_attempt_at = CASE WHEN requeue_requested THEN now() ELSE next_attempt_at END, + priority = CASE WHEN requeue_requested THEN 100 ELSE priority END, + requeue_requested = false, updated_at = now() WHERE content_id = $1 AND status = 'running' - AND last_attempt_at = $2 + AND claim_token = $2 ` -func (q *EnrichmentQueue) Release(ctx context.Context, contentID string) error { +func (q *EnrichmentQueue) Release(ctx context.Context, job EnrichmentJob) error { if q == nil || q.pool == nil { return errors.New("ebook enrichment queue is not configured") } - job, ok := q.claimedJob(contentID) - if !ok { + if job.ContentID == "" || job.Token == "" { return ErrEnrichmentLeaseLost } - return q.ReleaseClaim(ctx, job) -} - -func (q *EnrichmentQueue) ReleaseClaim(ctx context.Context, job EnrichmentJob) error { - if q == nil || q.pool == nil { - return errors.New("ebook enrichment queue is not configured") - } - tag, err := q.pool.Exec(ctx, releaseEnrichmentJobQuery, job.ContentID, job.LastAttemptAt) + tag, err := q.pool.Exec(ctx, releaseEnrichmentJobQuery, job.ContentID, job.Token) if err != nil { return err } - q.forgetClaim(job) if tag.RowsAffected() == 0 { return ErrEnrichmentLeaseLost } return nil } -func (q *EnrichmentQueue) rememberClaim(job EnrichmentJob) { - q.claimsMu.Lock() - defer q.claimsMu.Unlock() - if q.claims == nil { - q.claims = make(map[string]EnrichmentJob) - } - q.claims[job.ContentID] = job -} +var discardEnrichmentJobQuery = ` + UPDATE ebook_enrichment_state + SET status = 'discarded', + lease_until = NULL, + claim_token = NULL, + completed_at = now(), + outcome = 'discarded', + attempts = 0, + priority = 0, + requeue_requested = false, + last_error_class = NULL, + last_error = NULL, + updated_at = now() + WHERE content_id = $1 + AND status = 'running' + AND claim_token = $2 +` -func (q *EnrichmentQueue) claimedJob(contentID string) (EnrichmentJob, bool) { - q.claimsMu.Lock() - defer q.claimsMu.Unlock() - job, ok := q.claims[contentID] - return job, ok -} - -func (q *EnrichmentQueue) forgetClaim(job EnrichmentJob) { - q.claimsMu.Lock() - defer q.claimsMu.Unlock() - current, ok := q.claims[job.ContentID] - if ok && current.LastAttemptAt.Equal(job.LastAttemptAt) { - delete(q.claims, job.ContentID) +func (q *EnrichmentQueue) Discard(ctx context.Context, job EnrichmentJob) error { + if q == nil || q.pool == nil { + return errors.New("ebook enrichment queue is not configured") } + if job.ContentID == "" || job.Token == "" { + return ErrEnrichmentLeaseLost + } + tag, err := q.pool.Exec(ctx, discardEnrichmentJobQuery, job.ContentID, job.Token) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrEnrichmentLeaseLost + } + return nil } func enrichmentRefreshHorizon(outcome EnrichmentOutcome) time.Duration { diff --git a/internal/ebooks/enrichment_queue_test.go b/internal/ebooks/enrichment_queue_test.go index c161c848..9e0618ea 100644 --- a/internal/ebooks/enrichment_queue_test.go +++ b/internal/ebooks/enrichment_queue_test.go @@ -22,11 +22,13 @@ func TestEnrichmentQueueClaimQueryUsesAtomicLeasedClaims(t *testing.T) { "FOR UPDATE SKIP LOCKED", "status = 'pending' OR (status = 'running' AND lease_until < now())", "next_attempt_at <= now()", - "ORDER BY priority DESC, next_attempt_at, updated_at", + "priority + FLOOR(EXTRACT(EPOCH FROM (now() - next_attempt_at)) / 3600)::integer", "SET status = 'running'", "lease_until = now() + $2::interval", + "claim_token = gen_random_uuid()::text", "attempts = attempts + 1", - "RETURNING state.content_id, state.attempts", + "RETURNING state.content_id, state.claim_token, state.attempts", + "state.protected_fields", } { if !strings.Contains(query, fragment) { t.Fatalf("claim query missing %q:\n%s", fragment, claimEnrichmentJobsQuery) @@ -34,21 +36,76 @@ func TestEnrichmentQueueClaimQueryUsesAtomicLeasedClaims(t *testing.T) { } } -func TestEnrichmentQueueMaterializesUnrefreshedEbooksRegardlessOfPoster(t *testing.T) { +func TestEnrichmentQueueMaterializesEveryStandaloneEbookAndReactivatesResets(t *testing.T) { query := strings.Join(strings.Fields(materializeEnrichmentJobsQuery), " ") - if !strings.Contains(query, "SELECT mi.content_id, 'pending', 100, now(), now()") { - t.Fatalf("newly discovered ebooks must enter ahead of legacy backfill:\n%s", materializeEnrichmentJobsQuery) + for _, fragment := range []string{ + "CASE WHEN mi.last_refreshed IS NULL THEN 100 ELSE 0 END", + "GREATEST(mi.last_refreshed + interval '90 days', now())", + "mi.type = 'ebook'", + "manga_chapters", + "ON CONFLICT (content_id) DO UPDATE SET", + "ebook_enrichment_state.status = 'discarded'", + "EXCLUDED.completed_at IS NULL AND ebook_enrichment_state.completed_at IS NOT NULL", + "protected_fields", + "lower(trim(mi.status)) = 'pending'", + "ip.kind = 7", + "poster_path", + "ebook_enrichment_state.protected_fields ||", + "WHERE candidate IN ('poster_path', 'backdrop_path', 'logo_path')", + } { + if !strings.Contains(query, fragment) { + t.Fatalf("materialization query missing %q:\n%s", fragment, materializeEnrichmentJobsQuery) + } } - if !strings.Contains(query, "mi.last_refreshed IS NULL") { - t.Fatalf("materialization must use refresh state as its eligibility gate:\n%s", materializeEnrichmentJobsQuery) - } - if strings.Contains(query, "poster_path") { - t.Fatalf("local or embedded covers must not make ebooks ineligible for metadata enrichment:\n%s", materializeEnrichmentJobsQuery) + if strings.Contains(query, "AND mi.last_refreshed IS NULL") { + t.Fatalf("last_refreshed must schedule work, not exclude refreshed ebooks:\n%s", materializeEnrichmentJobsQuery) } } -func TestEnrichmentQueueMigrationSeedsLegacyBacklogAtLowPriority(t *testing.T) { +func TestEnrichmentQueueCapturesEveryProviderWritablePendingField(t *testing.T) { + query := strings.Join(strings.Fields(ebookProtectedFieldsSQL), " ") + for _, field := range []string{ + "title", + "year", + "overview", + "tagline", + "content_rating", + "runtime", + "release_date", + "genres", + "studios", + "authors", + "poster_path", + "backdrop_path", + "logo_path", + } { + if !strings.Contains(query, "'"+field+"'") { + t.Fatalf("protected-field capture missing %q:\n%s", field, ebookProtectedFieldsSQL) + } + } +} + +func TestEnrichmentQueueEnqueueMakesPendingRowsDueAndDefersRunningRequeue(t *testing.T) { + query := strings.Join(strings.Fields(enqueueEnrichmentJobQuery), " ") + for _, fragment := range []string{ + "INSERT INTO ebook_enrichment_state", + "FROM media_items mi", + "ON CONFLICT (content_id) DO UPDATE SET", + "WHEN ebook_enrichment_state.status = 'running' THEN ebook_enrichment_state.next_attempt_at", + "ELSE now()", + "requeue_requested = ebook_enrichment_state.requeue_requested OR ebook_enrichment_state.status = 'running'", + } { + if !strings.Contains(query, fragment) { + t.Fatalf("enqueue query missing %q:\n%s", fragment, enqueueEnrichmentJobQuery) + } + } + if strings.Contains(query, "lease_until =") || strings.Contains(query, "claim_token =") { + t.Fatalf("enqueue must not mutate an active lease:\n%s", enqueueEnrichmentJobQuery) + } +} + +func TestEnrichmentQueueMigrationIsCrashSafeAndSeedsAllLegacyEbooks(t *testing.T) { body, err := os.ReadFile("../../migrations/sql/20260719090000_ebook_enrichment_jobs.sql") if err != nil { t.Fatalf("read enrichment queue migration: %v", err) @@ -56,20 +113,53 @@ func TestEnrichmentQueueMigrationSeedsLegacyBacklogAtLowPriority(t *testing.T) { migration := strings.Join(strings.Fields(string(body)), " ") for _, fragment := range []string{ + "-- +goose NO TRANSACTION", + "ADD COLUMN IF NOT EXISTS claim_token text", + "ADD COLUMN IF NOT EXISTS requeue_requested boolean NOT NULL DEFAULT false", + "ADD COLUMN IF NOT EXISTS protected_fields text[] NOT NULL DEFAULT '{}'::text[]", "INSERT INTO ebook_enrichment_state", - "SELECT mi.content_id, 'pending', -100, 0, now(), now()", + "CASE WHEN mi.last_refreshed IS NULL THEN -100 ELSE 0 END", + "GREATEST(mi.last_refreshed + interval '90 days', now())", "mi.type = 'ebook'", - "mi.last_refreshed IS NULL", "NOT EXISTS", "manga_chapters", "ON CONFLICT (content_id) DO NOTHING", + "NOT i.indisvalid", + "DROP INDEX public.ebook_enrichment_state_claim_idx", + "CREATE INDEX CONCURRENTLY IF NOT EXISTS ebook_enrichment_state_claim_idx", + "DROP INDEX CONCURRENTLY IF EXISTS ebook_enrichment_state_claim_idx", } { if !strings.Contains(migration, fragment) { t.Fatalf("legacy backlog migration missing %q:\n%s", fragment, body) } } - if strings.Contains(migration, "poster_path") { - t.Fatalf("legacy backlog must include ebooks with local or embedded covers:\n%s", body) + if strings.Contains(migration, "AND mi.last_refreshed IS NULL") { + t.Fatalf("migration must seed refreshed ebooks too:\n%s", body) + } +} + +func TestEnrichmentQueueMigrationDownPreservesCurrentFailureHistory(t *testing.T) { + body, err := os.ReadFile("../../migrations/sql/20260719090000_ebook_enrichment_jobs.sql") + if err != nil { + t.Fatalf("read enrichment queue migration: %v", err) + } + parts := strings.SplitN(strings.Join(strings.Fields(string(body)), " "), "-- +goose Down", 2) + if len(parts) != 2 { + t.Fatal("migration missing Down section") + } + down := parts[1] + for _, fragment := range []string{ + "SET failures = attempts", + "WHERE outcome = 'failed'", + "DELETE FROM ebook_enrichment_state WHERE outcome IN ('success', 'no_match')", + "IF EXISTS", + } { + if !strings.Contains(down, fragment) { + t.Fatalf("down migration missing %q:\n%s", fragment, body) + } + } + if strings.Contains(down, "WHERE failures = 0") { + t.Fatalf("down migration must not erase post-migration failures based on the stale legacy counter:\n%s", body) } } @@ -80,11 +170,12 @@ func TestEnrichmentQueueTransitionsKeepDurableRowsAndReleaseLeases(t *testing.T) "status = 'pending'", "lease_until = NULL", "completed_at = now()", - "next_attempt_at = now() + $3::interval", + "ELSE now() + $3::interval", "outcome = $2", "attempts = 0", - "priority = GREATEST(priority, 0)", - "AND last_attempt_at = $4", + "WHEN requeue_requested THEN 100 ELSE 0 END", + "requeue_requested = false", + "AND claim_token = $4", } { if !strings.Contains(complete, fragment) { t.Fatalf("complete query missing %q:\n%s", fragment, completeEnrichmentJobQuery) @@ -101,7 +192,7 @@ func TestEnrichmentQueueTransitionsKeepDurableRowsAndReleaseLeases(t *testing.T) "attempts = GREATEST(attempts - 1, 0)", "WHERE content_id = $1", "AND status = 'running'", - "AND last_attempt_at = $2", + "AND claim_token = $2", } { if !strings.Contains(release, fragment) { t.Fatalf("release query missing %q:\n%s", fragment, releaseEnrichmentJobQuery) @@ -109,8 +200,28 @@ func TestEnrichmentQueueTransitionsKeepDurableRowsAndReleaseLeases(t *testing.T) } failure := strings.Join(strings.Fields(failEnrichmentJobQuery), " ") - if !strings.Contains(failure, "AND last_attempt_at = $5") { - t.Fatalf("failure transition must reject stale leases:\n%s", failEnrichmentJobQuery) + for _, fragment := range []string{ + "WHEN requeue_requested THEN now()", + "WHEN requeue_requested THEN 100 ELSE priority END", + "AND claim_token = $5", + } { + if !strings.Contains(failure, fragment) { + t.Fatalf("failure transition missing %q:\n%s", fragment, failEnrichmentJobQuery) + } + } + + discard := strings.Join(strings.Fields(discardEnrichmentJobQuery), " ") + for _, fragment := range []string{ + "status = 'discarded'", + "lease_until = NULL", + "claim_token = NULL", + "outcome = 'discarded'", + "WHERE content_id = $1", + "AND claim_token = $2", + } { + if !strings.Contains(discard, fragment) { + t.Fatalf("discard transition missing %q:\n%s", fragment, discardEnrichmentJobQuery) + } } } diff --git a/internal/ebooks/enrichment_test.go b/internal/ebooks/enrichment_test.go index ddba0cb0..924160af 100644 --- a/internal/ebooks/enrichment_test.go +++ b/internal/ebooks/enrichment_test.go @@ -184,15 +184,20 @@ func TestEnrichmentQueriesKeepEbookAndAudiobookMetadataSeparate(t *testing.T) { if !strings.Contains(loadEnrichmentItemsQuery, "ip.kind = 7") { t.Fatalf("ebook load query must load author credits only:\n%s", loadEnrichmentItemsQuery) } + for _, field := range []string{"locked_fields", "backdrop_path", "logo_path"} { + if !strings.Contains(loadEnrichmentItemsQuery, field) { + t.Fatalf("ebook load query must load %s for protection decisions:\n%s", field, loadEnrichmentItemsQuery) + } + } } func TestEnricherRunTransitionsClaimedJobsByOutcome(t *testing.T) { queue := &fakeEnrichmentQueue{ jobs: []EnrichmentJob{ - {ContentID: "success", Attempts: 1}, - {ContentID: "no-match", Attempts: 1}, - {ContentID: "skipped", Attempts: 1}, - {ContentID: "failed", Attempts: 3}, + {ContentID: "success", Token: "success-token", Attempts: 1}, + {ContentID: "no-match", Token: "no-match-token", Attempts: 1}, + {ContentID: "skipped", Token: "skipped-token", Attempts: 1}, + {ContentID: "failed", Token: "failed-token", Attempts: 3}, }, } items := []enrichmentItemRow{ @@ -247,14 +252,19 @@ func TestEnricherRunTransitionsClaimedJobsByOutcome(t *testing.T) { if got := queue.failed["failed"]; got != EnrichmentErrorTransient { t.Fatalf("failed class = %q, want transient", got) } + for contentID, job := range queue.transitionedJobs { + if job.Token != contentID+"-token" { + t.Fatalf("transition for %q used token %q", contentID, job.Token) + } + } } func TestEnricherRunReleasesEveryLeaseOnCancellation(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) queue := &fakeEnrichmentQueue{ jobs: []EnrichmentJob{ - {ContentID: "first", Attempts: 1}, - {ContentID: "second", Attempts: 1}, + {ContentID: "first", Token: "first-token", Attempts: 1}, + {ContentID: "second", Token: "second-token", Attempts: 1}, }, } items := []enrichmentItemRow{{ContentID: "first"}, {ContentID: "second"}} @@ -290,7 +300,7 @@ func TestEnricherRunReleasesEveryLeaseOnCancellation(t *testing.T) { func TestEnricherRunReleasesLeaseWhenCompletionLosesContext(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) queue := &fakeEnrichmentQueue{ - jobs: []EnrichmentJob{{ContentID: "success", Attempts: 1}}, + jobs: []EnrichmentJob{{ContentID: "success", Token: "success-token", Attempts: 1}}, completeCancel: cancel, completeErr: context.Canceled, } @@ -318,6 +328,72 @@ func TestEnricherRunReleasesLeaseWhenCompletionLosesContext(t *testing.T) { } } +func TestEnricherRunDiscardsClaimedRowsThatAreNoLongerEligible(t *testing.T) { + queue := &fakeEnrichmentQueue{ + jobs: []EnrichmentJob{ + {ContentID: "became-manga", Token: "manga-token", Attempts: 1}, + {ContentID: "folderless", Token: "folderless-token", Attempts: 1}, + }, + } + e := &Enricher{ + queue: queue, + batchSize: 2, + workers: 1, + loadClaimedItemsFn: func(context.Context, []EnrichmentJob) ([]enrichmentItemRow, error) { + return []enrichmentItemRow{{ContentID: "folderless", FolderID: 0}}, nil + }, + enrichClaimedItemFn: func(context.Context, enrichmentItemRow) (EnrichmentOutcome, error) { + return EnrichmentOutcomeSkipped, nil + }, + } + + if _, err := e.Run(context.Background()); err != nil { + t.Fatalf("Run() error = %v", err) + } + if got := strings.Join(queue.discarded, ","); got != "became-manga" { + t.Fatalf("discarded = %q, want became-manga", got) + } + if got := queue.completed["folderless"]; got != EnrichmentOutcomeSkipped { + t.Fatalf("folderless outcome = %q, want skipped", got) + } + if len(queue.released) != 0 { + t.Fatalf("ineligible rows were released back into immediate churn: %v", queue.released) + } +} + +func TestEnricherRunBoundsEachItemBelowTheLease(t *testing.T) { + queue := &fakeEnrichmentQueue{ + jobs: []EnrichmentJob{{ContentID: "slow", Token: "slow-token", Attempts: 1}}, + } + e := &Enricher{ + queue: queue, + batchSize: 1, + workers: 1, + itemTimeout: 20 * time.Millisecond, + loadClaimedItemsFn: func(context.Context, []EnrichmentJob) ([]enrichmentItemRow, error) { + return []enrichmentItemRow{{ContentID: "slow"}}, nil + }, + enrichClaimedItemFn: func(ctx context.Context, _ enrichmentItemRow) (EnrichmentOutcome, error) { + <-ctx.Done() + return "", ctx.Err() + }, + } + + started := time.Now() + if _, err := e.Run(context.Background()); err != nil { + t.Fatalf("Run() error = %v", err) + } + if elapsed := time.Since(started); elapsed > time.Second { + t.Fatalf("per-item timeout took %s", elapsed) + } + if got := strings.Join(queue.released, ","); got != "slow" { + t.Fatalf("released = %q, want slow", got) + } + if defaultEnrichmentItemTimeout >= defaultEnrichmentLease/2 { + t.Fatalf("default item timeout %s is not materially shorter than lease %s", defaultEnrichmentItemTimeout, defaultEnrichmentLease) + } +} + func TestEnrichItemSkipsItemWithoutLibraryFolder(t *testing.T) { // Membership rows are inserted after the item upsert, so a scan-window // race can claim an item before its folder link exists. The item must be @@ -718,16 +794,12 @@ func TestBuildEbookMetadataRequestCarriesAccumulatedISBN(t *testing.T) { } } -func TestPreservePendingEbookLocalMetadata(t *testing.T) { +func TestPreserveDurableEbookLocalMetadataAcrossRefreshes(t *testing.T) { item := enrichmentItemRow{ - Status: "pending", - Year: 2021, - Overview: "Embedded description", - ReleaseDate: "2021-03-04", - Genres: []string{"Local genre"}, - Studios: []string{"Local publisher"}, - PosterPath: "local/ebooks/book/poster/original.webp", - Author: "Embedded Author", + Status: "matched", + ProtectedFields: []string{ + "year", "overview", "release_date", "genres", "studios", "poster_path", "authors", + }, } result := &metadata.MetadataResult{ HasMetadata: true, @@ -759,7 +831,7 @@ func TestPreservePendingEbookLocalMetadata(t *testing.T) { } } -func TestPreservePendingEbookLocalMetadataAllowsRefreshReplacement(t *testing.T) { +func TestPreserveEbookMetadataAllowsProviderOwnedRefreshReplacement(t *testing.T) { result := &metadata.MetadataResult{ HasMetadata: true, Year: 2022, @@ -767,9 +839,7 @@ func TestPreservePendingEbookLocalMetadataAllowsRefreshReplacement(t *testing.T) } preserveEbookLocalMetadata(enrichmentItemRow{ - Status: "matched", - Year: 2021, - Overview: "Old remote description", + Status: "matched", }, result) if result.Year != 2022 || result.Overview != "Corrected remote description" { @@ -782,15 +852,20 @@ func TestPreserveEbookLocalPosterDuringControlledRefresh(t *testing.T) { HasMetadata: true, PosterPath: "https://example.test/replacement.jpg", PosterThumbhash: "remote-thumb", + BackdropPath: "https://example.test/backdrop.jpg", + LogoPath: "https://example.test/logo.png", Overview: "Corrected remote description", } preserveEbookLocalMetadata(enrichmentItemRow{ - Status: "matched", - PosterPath: "local/ebooks/book/poster/original.webp", + Status: "matched", + PosterPath: "local/ebooks/book/poster/original.webp", + BackdropPath: "/books/book/backdrop.jpg", + LogoPath: "embedded/book/logo.png", }, result) - if result.PosterPath != "" || result.PosterThumbhash != "" { + if result.PosterPath != "" || result.PosterThumbhash != "" || + result.BackdropPath != "" || result.LogoPath != "" { t.Fatalf("controlled refresh would replace local poster: %+v", result) } if result.Overview != "Corrected remote description" { @@ -798,6 +873,46 @@ func TestPreserveEbookLocalPosterDuringControlledRefresh(t *testing.T) { } } +func TestPreserveEbookMetadataHonorsGenericLockedFields(t *testing.T) { + result := &metadata.MetadataResult{ + HasMetadata: true, + Title: "Remote title", + Overview: "Remote overview", + Year: 2024, + ReleaseDate: "2024-01-02", + Runtime: 500, + Genres: []string{"Remote genre"}, + Studios: []string{"Remote publisher"}, + ContentRating: "Teen", + PosterPath: "https://example.test/poster.jpg", + BackdropPath: "https://example.test/backdrop.jpg", + LogoPath: "https://example.test/logo.png", + People: []models.ItemPerson{ + {Person: models.Person{Name: "Remote Author"}, Kind: models.PersonKindAuthor}, + }, + } + item := enrichmentItemRow{LockedFields: []int{ + int(metadata.FieldName), + int(metadata.FieldOverview), + int(metadata.FieldGenres), + int(metadata.FieldStudios), + int(metadata.FieldCrew), + int(metadata.FieldRuntime), + int(metadata.FieldContentRating), + int(metadata.FieldImages), + int(metadata.FieldReleaseDates), + }} + + preserveEbookLocalMetadata(item, result) + + if result.Title != "" || result.Overview != "" || result.Year != 0 || result.ReleaseDate != "" || + result.Runtime != 0 || len(result.Genres) != 0 || len(result.Studios) != 0 || + result.ContentRating != "" || result.PosterPath != "" || result.BackdropPath != "" || + result.LogoPath != "" || len(result.People) != 0 { + t.Fatalf("locked metadata was not protected: %+v", result) + } +} + func TestPreserveEbookLocalPosterAllowsProviderOwnedReplacement(t *testing.T) { for _, current := range []string{ "", @@ -872,6 +987,8 @@ type fakeEnrichmentQueue struct { completed map[string]EnrichmentOutcome failed map[string]EnrichmentErrorClass released []string + discarded []string + transitionedJobs map[string]EnrichmentJob releaseSawCanceledContext bool completeCancel context.CancelFunc completeErr error @@ -893,37 +1010,55 @@ func (f *fakeEnrichmentQueue) ClaimBatch(_ context.Context, limit int, leaseDura return append([]EnrichmentJob(nil), f.jobs...), nil } -func (f *fakeEnrichmentQueue) Complete(_ context.Context, contentID string, outcome EnrichmentOutcome, _ time.Duration) error { +func (f *fakeEnrichmentQueue) Complete(_ context.Context, job EnrichmentJob, outcome EnrichmentOutcome, _ time.Duration) error { f.mu.Lock() defer f.mu.Unlock() if f.completed == nil { f.completed = make(map[string]EnrichmentOutcome) } - f.completed[contentID] = outcome + f.completed[job.ContentID] = outcome + f.recordTransition(job) if f.completeCancel != nil { f.completeCancel() } return f.completeErr } -func (f *fakeEnrichmentQueue) Fail(_ context.Context, contentID string, errorClass EnrichmentErrorClass, _ string, _ time.Duration) error { +func (f *fakeEnrichmentQueue) Fail(_ context.Context, job EnrichmentJob, errorClass EnrichmentErrorClass, _ string, _ time.Duration) error { f.mu.Lock() defer f.mu.Unlock() if f.failed == nil { f.failed = make(map[string]EnrichmentErrorClass) } - f.failed[contentID] = errorClass + f.failed[job.ContentID] = errorClass + f.recordTransition(job) return nil } -func (f *fakeEnrichmentQueue) Release(ctx context.Context, contentID string) error { +func (f *fakeEnrichmentQueue) Release(ctx context.Context, job EnrichmentJob) error { f.mu.Lock() defer f.mu.Unlock() f.releaseSawCanceledContext = f.releaseSawCanceledContext || ctx.Err() != nil - f.released = append(f.released, contentID) + f.released = append(f.released, job.ContentID) + f.recordTransition(job) return nil } +func (f *fakeEnrichmentQueue) Discard(_ context.Context, job EnrichmentJob) error { + f.mu.Lock() + defer f.mu.Unlock() + f.discarded = append(f.discarded, job.ContentID) + f.recordTransition(job) + return nil +} + +func (f *fakeEnrichmentQueue) recordTransition(job EnrichmentJob) { + if f.transitionedJobs == nil { + f.transitionedJobs = make(map[string]EnrichmentJob) + } + f.transitionedJobs[job.ContentID] = job +} + func TestCleanEbookSearchTitle(t *testing.T) { cases := []struct { title, author, want string diff --git a/migrations/sql/20260719090000_ebook_enrichment_jobs.sql b/migrations/sql/20260719090000_ebook_enrichment_jobs.sql index a749232a..092593df 100644 --- a/migrations/sql/20260719090000_ebook_enrichment_jobs.sql +++ b/migrations/sql/20260719090000_ebook_enrichment_jobs.sql @@ -1,38 +1,130 @@ +-- +goose NO TRANSACTION + -- +goose Up +-- This ALTER is intentionally its own autocommitted statement so its table +-- lock is released before the potentially large backfill runs separately. +ALTER TABLE ebook_enrichment_state + ADD COLUMN IF NOT EXISTS status text NOT NULL DEFAULT 'pending', + ADD COLUMN IF NOT EXISTS priority integer NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS attempts integer NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS next_attempt_at timestamptz NOT NULL DEFAULT now(), + ADD COLUMN IF NOT EXISTS lease_until timestamptz, + ADD COLUMN IF NOT EXISTS claim_token text, + ADD COLUMN IF NOT EXISTS requeue_requested boolean NOT NULL DEFAULT false, + ADD COLUMN IF NOT EXISTS protected_fields text[] NOT NULL DEFAULT '{}'::text[], + ADD COLUMN IF NOT EXISTS last_attempt_at timestamptz, + ADD COLUMN IF NOT EXISTS completed_at timestamptz, + ADD COLUMN IF NOT EXISTS outcome text, + ADD COLUMN IF NOT EXISTS last_error_class text, + ADD COLUMN IF NOT EXISTS last_error text; + +ALTER TABLE ebook_enrichment_state + DROP CONSTRAINT IF EXISTS ebook_enrichment_state_status_check, + DROP CONSTRAINT IF EXISTS ebook_enrichment_state_attempts_check; + ALTER TABLE ebook_enrichment_state - ADD COLUMN status text NOT NULL DEFAULT 'pending', - ADD COLUMN priority integer NOT NULL DEFAULT 0, - ADD COLUMN attempts integer NOT NULL DEFAULT 0, - ADD COLUMN next_attempt_at timestamptz NOT NULL DEFAULT now(), - ADD COLUMN lease_until timestamptz, - ADD COLUMN last_attempt_at timestamptz, - ADD COLUMN completed_at timestamptz, - ADD COLUMN outcome text, - ADD COLUMN last_error_class text, - ADD COLUMN last_error text, ADD CONSTRAINT ebook_enrichment_state_status_check - CHECK (status IN ('pending', 'running')), + CHECK (status IN ('pending', 'running', 'discarded')), ADD CONSTRAINT ebook_enrichment_state_attempts_check CHECK (attempts >= 0); --- Rows already tracked by the former failure counter belong to the legacy --- backlog. Preserve their history but keep them behind incremental work. -UPDATE ebook_enrichment_state -SET attempts = failures, - priority = -100; +-- Existing failure rows are part of the legacy backlog. Refreshed rows retain +-- their refresh date as the start of the 90-day schedule; unrefreshed rows keep +-- their old failure count and enter at legacy priority -100. +UPDATE ebook_enrichment_state state +SET status = 'pending', + priority = CASE WHEN mi.last_refreshed IS NULL THEN -100 ELSE 0 END, + attempts = state.failures, + next_attempt_at = CASE + WHEN mi.last_refreshed IS NULL THEN now() + ELSE GREATEST(mi.last_refreshed + interval '90 days', now()) + END, + completed_at = mi.last_refreshed, + protected_fields = ARRAY_REMOVE(ARRAY[ + CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.title, '')) <> '' THEN 'title' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND COALESCE(mi.year, 0) > 0 THEN 'year' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.overview, '')) <> '' THEN 'overview' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.tagline, '')) <> '' THEN 'tagline' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.content_rating, '')) <> '' THEN 'content_rating' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND COALESCE(mi.runtime, 0) > 0 THEN 'runtime' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.release_date::text, '')) <> '' THEN 'release_date' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND cardinality(COALESCE(mi.genres, '{}'::text[])) > 0 THEN 'genres' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND cardinality(COALESCE(mi.studios, '{}'::text[])) > 0 THEN 'studios' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND EXISTS ( + SELECT 1 FROM item_people ip WHERE ip.content_id = mi.content_id AND ip.kind = 7 + ) THEN 'authors' END, + CASE WHEN trim(COALESCE(mi.poster_path, '')) <> '' + AND lower(trim(mi.poster_path)) NOT LIKE 'http://%' + AND lower(trim(mi.poster_path)) NOT LIKE 'https://%' + AND lower(trim(mi.poster_path)) NOT LIKE 'ebook-metadata/ebooks/%' + THEN 'poster_path' END, + CASE WHEN trim(COALESCE(mi.backdrop_path, '')) <> '' + AND lower(trim(mi.backdrop_path)) NOT LIKE 'http://%' + AND lower(trim(mi.backdrop_path)) NOT LIKE 'https://%' + AND lower(trim(mi.backdrop_path)) NOT LIKE 'ebook-metadata/ebooks/%' + THEN 'backdrop_path' END, + CASE WHEN trim(COALESCE(mi.logo_path, '')) <> '' + AND lower(trim(mi.logo_path)) NOT LIKE 'http://%' + AND lower(trim(mi.logo_path)) NOT LIKE 'https://%' + AND lower(trim(mi.logo_path)) NOT LIKE 'ebook-metadata/ebooks/%' + THEN 'logo_path' END + ]::text[], NULL), + updated_at = now() +FROM media_items mi +WHERE mi.content_id = state.content_id; --- Snapshot the pre-migration library as durable low-priority backfill. Runtime --- materialization assigns new discoveries priority 100, so this backlog cannot --- occupy claim slots ahead of incremental or due refresh work. +-- Snapshot every pre-migration standalone ebook. Refreshed rows become normal +-- 90-day refresh work; only the unrefreshed legacy backlog starts at -100. INSERT INTO ebook_enrichment_state ( content_id, status, priority, attempts, next_attempt_at, + completed_at, + protected_fields, updated_at ) -SELECT mi.content_id, 'pending', -100, 0, now(), now() +SELECT + mi.content_id, + 'pending', + CASE WHEN mi.last_refreshed IS NULL THEN -100 ELSE 0 END, + 0, + CASE + WHEN mi.last_refreshed IS NULL THEN now() + ELSE GREATEST(mi.last_refreshed + interval '90 days', now()) + END, + mi.last_refreshed, + ARRAY_REMOVE(ARRAY[ + CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.title, '')) <> '' THEN 'title' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND COALESCE(mi.year, 0) > 0 THEN 'year' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.overview, '')) <> '' THEN 'overview' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.tagline, '')) <> '' THEN 'tagline' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.content_rating, '')) <> '' THEN 'content_rating' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND COALESCE(mi.runtime, 0) > 0 THEN 'runtime' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.release_date::text, '')) <> '' THEN 'release_date' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND cardinality(COALESCE(mi.genres, '{}'::text[])) > 0 THEN 'genres' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND cardinality(COALESCE(mi.studios, '{}'::text[])) > 0 THEN 'studios' END, + CASE WHEN lower(trim(mi.status)) = 'pending' AND EXISTS ( + SELECT 1 FROM item_people ip WHERE ip.content_id = mi.content_id AND ip.kind = 7 + ) THEN 'authors' END, + CASE WHEN trim(COALESCE(mi.poster_path, '')) <> '' + AND lower(trim(mi.poster_path)) NOT LIKE 'http://%' + AND lower(trim(mi.poster_path)) NOT LIKE 'https://%' + AND lower(trim(mi.poster_path)) NOT LIKE 'ebook-metadata/ebooks/%' + THEN 'poster_path' END, + CASE WHEN trim(COALESCE(mi.backdrop_path, '')) <> '' + AND lower(trim(mi.backdrop_path)) NOT LIKE 'http://%' + AND lower(trim(mi.backdrop_path)) NOT LIKE 'https://%' + AND lower(trim(mi.backdrop_path)) NOT LIKE 'ebook-metadata/ebooks/%' + THEN 'backdrop_path' END, + CASE WHEN trim(COALESCE(mi.logo_path, '')) <> '' + AND lower(trim(mi.logo_path)) NOT LIKE 'http://%' + AND lower(trim(mi.logo_path)) NOT LIKE 'https://%' + AND lower(trim(mi.logo_path)) NOT LIKE 'ebook-metadata/ebooks/%' + THEN 'logo_path' END + ]::text[], NULL), + now() FROM media_items mi WHERE mi.type = 'ebook' AND NOT EXISTS ( @@ -40,20 +132,58 @@ WHERE mi.type = 'ebook' FROM manga_chapters mc WHERE mc.chapter_content_id = mi.content_id ) - AND mi.last_refreshed IS NULL ON CONFLICT (content_id) DO NOTHING; -CREATE INDEX ebook_enrichment_state_claim_idx - ON ebook_enrichment_state (priority DESC, next_attempt_at, updated_at) +-- A crashed CREATE INDEX CONCURRENTLY can leave an INVALID index which blocks +-- an IF NOT EXISTS retry. Match the repository's crash-safe index pattern. +-- +goose StatementBegin +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 + FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + JOIN pg_index i ON i.indexrelid = c.oid + WHERE n.nspname = 'public' + AND c.relname = 'ebook_enrichment_state_claim_idx' + AND NOT i.indisvalid + ) THEN + DROP INDEX public.ebook_enrichment_state_claim_idx; + END IF; +END; +$$; +-- +goose StatementEnd + +CREATE INDEX CONCURRENTLY IF NOT EXISTS ebook_enrichment_state_claim_idx + ON public.ebook_enrichment_state (next_attempt_at, priority DESC, updated_at) WHERE status IN ('pending', 'running'); -- +goose Down -DROP INDEX IF EXISTS ebook_enrichment_state_claim_idx; +DROP INDEX CONCURRENTLY IF EXISTS ebook_enrichment_state_claim_idx; --- The former state table only retained positive failure rows. Remove queue rows --- created by this migration/runtime before restoring that representation. -DELETE FROM ebook_enrichment_state -WHERE failures = 0; +-- +goose StatementBegin +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 + FROM information_schema.columns + WHERE table_schema = 'public' + AND table_name = 'ebook_enrichment_state' + AND column_name = 'attempts' + ) THEN + UPDATE ebook_enrichment_state + SET failures = attempts + WHERE outcome = 'failed'; + + -- Success and no-match rows have no representation in the old + -- failure-only schema. Keep pending, skipped, discarded, and failed + -- rows; old code can faithfully interpret their copied failure count. + DELETE FROM ebook_enrichment_state + WHERE outcome IN ('success', 'no_match'); + END IF; +END; +$$; +-- +goose StatementEnd ALTER TABLE ebook_enrichment_state DROP CONSTRAINT IF EXISTS ebook_enrichment_state_attempts_check, @@ -63,6 +193,9 @@ ALTER TABLE ebook_enrichment_state DROP COLUMN IF EXISTS outcome, DROP COLUMN IF EXISTS completed_at, DROP COLUMN IF EXISTS last_attempt_at, + DROP COLUMN IF EXISTS protected_fields, + DROP COLUMN IF EXISTS requeue_requested, + DROP COLUMN IF EXISTS claim_token, DROP COLUMN IF EXISTS lease_until, DROP COLUMN IF EXISTS next_attempt_at, DROP COLUMN IF EXISTS attempts,