diff --git a/Makefile b/Makefile index 32cff7f3..60a2eb4c 100644 --- a/Makefile +++ b/Makefile @@ -14,13 +14,20 @@ endif JELLYFIN_WEB_OUTPUT_DIR ?= third_party/jellyfin-web +# Build version stamping: inject the git revision so the admin Build panel shows a +# version even when Go's VCS metadata isn't embedded (mirrors the Dockerfile ldflags). +BUILDINFO_PKG := github.com/Silo-Server/silo-server/internal/buildinfo +BUILD_REVISION ?= $(shell git rev-parse HEAD 2>/dev/null) +BUILD_DIRTY ?= $(shell test -n "$$(git status --porcelain 2>/dev/null)" && echo true || echo false) +GO_LDFLAGS := -X $(BUILDINFO_PKG).revisionOverride=$(BUILD_REVISION) -X $(BUILDINFO_PKG).dirtyOverride=$(BUILD_DIRTY) + # Build the frontend (requires pnpm) frontend: cd web && pnpm install --frozen-lockfile && pnpm run build # Build the Go binary (depends on frontend) build: frontend - go build -o silo ./cmd/silo/ + go build -ldflags "$(GO_LDFLAGS)" -o silo ./cmd/silo/ # Run frontend dev server (proxies API to localhost:8080) dev-frontend: diff --git a/cmd/silo/main.go b/cmd/silo/main.go index 213c0e8e..1305fed6 100644 --- a/cmd/silo/main.go +++ b/cmd/silo/main.go @@ -15,6 +15,7 @@ import ( "os" "os/signal" "path/filepath" + "runtime/debug" "strconv" "strings" "syscall" @@ -724,6 +725,12 @@ func main() { pluginService.OnLifecycleChange(appCtx) } + // backgroundInit collects non-critical startup work (catalog-size-dependent + // seeding, network-bound reconciliation) that must not block the HTTP + // listener. The steps run sequentially in a background goroutine once the + // server is ready to serve. Failures are logged, never fatal. + var backgroundInit []func(context.Context) + // Step 4b: Create metadata service and match worker (if needed). var metadataService *metadata.MetadataService var personRefreshService *metadata.PersonRefreshService @@ -830,23 +837,28 @@ func main() { matchWorker.SetMovieFileClaimer(movieQueueRepo) } if seriesQueueRepo != nil { - if cleaned, err := seriesQueueRepo.CleanupLegacySeriesGroupQueue(appCtx); err != nil { - slog.Warn("failed to clean legacy series group queue rows", "error", err) - } else if cleaned > 0 { - slog.Info("cleaned legacy series group queue rows", "count", cleaned) - } matchWorker.SetSeriesRootClaimer(seriesQueueRepo, cfg.Matcher.TVSeriesRootQueueEnabled()) + backgroundInit = append(backgroundInit, func(ctx context.Context) { + if cleaned, err := seriesQueueRepo.CleanupLegacySeriesGroupQueue(ctx); err != nil { + slog.Warn("failed to clean legacy series group queue rows", "error", err) + } else if cleaned > 0 { + slog.Info("cleaned legacy series group queue rows", "count", cleaned) + } + }) } if deps.FolderRepo != nil { - enabledFolders, err := deps.FolderRepo.GetEnabled(appCtx) - if err != nil { - slog.Warn("failed to seed metadata queues", "error", err) - } else { + backgroundInit = append(backgroundInit, func(ctx context.Context) { + start := time.Now() + enabledFolders, err := deps.FolderRepo.GetEnabled(ctx) + if err != nil { + slog.Warn("failed to seed metadata queues", "error", err) + return + } seedMovieQueue := func(folderID int) { if movieQueueRepo == nil { return } - if err := movieQueueRepo.SyncForFolder(appCtx, folderID); err != nil { + if err := movieQueueRepo.SyncForFolder(ctx, folderID); err != nil { slog.Warn("failed to seed movie match queue", "folder_id", folderID, "error", err) } } @@ -854,7 +866,7 @@ func main() { if seriesQueueRepo == nil { return } - if err := seriesQueueRepo.SyncForFolder(appCtx, folderID); err != nil { + if err := seriesQueueRepo.SyncForFolder(ctx, folderID); err != nil { slog.Warn("failed to seed series root queue", "folder_id", folderID, "error", err) } } @@ -872,7 +884,8 @@ func main() { seedMovieQueue(folder.ID) } } - } + slog.Info("deferred init: metadata match queues seeded", "folders", len(enabledFolders), "duration", time.Since(start)) + }) } deps.SkippedRootRepo = skippedRootRepo @@ -1011,9 +1024,11 @@ func main() { WithMatcher(historyimport.NewMatcher(historyRepo)). WithWatchState(watchstate.NewService(userStoreProvider).WithStableIdentityResolver(historyIdentity)). WithUserStoreProvider(userStoreProvider) - if err := watchProviderService.SweepOpenScrobbles(appCtx); err != nil { - slog.Warn("failed to sweep open watch provider scrobbles", "error", err) - } + backgroundInit = append(backgroundInit, func(ctx context.Context) { + if err := watchProviderService.SweepOpenScrobbles(ctx); err != nil { + slog.Warn("failed to sweep open watch provider scrobbles", "error", err) + } + }) } deps.SessionMgr = sessionMgr deps.PlaybackRealtimeHub = playback.NewRealtimeHub() @@ -1653,6 +1668,30 @@ func main() { compatSrv.IdleTimeout = 120 * time.Second } + // Run non-critical startup work in the background so it doesn't delay the + // HTTP listener from accepting connections. Steps run sequentially and stop + // early if the app context is cancelled (shutdown). + if len(backgroundInit) > 0 { + go func() { + start := time.Now() + for _, step := range backgroundInit { + if appCtx.Err() != nil { + return + } + func() { + defer func() { + if p := recover(); p != nil { + slog.Error("deferred startup init step panicked; continuing", + "panic", p, "stack", string(debug.Stack())) + } + }() + step(appCtx) + }() + } + slog.Info("deferred startup init completed", "steps", len(backgroundInit), "duration", time.Since(start)) + }() + } + errCh := make(chan error, 2) go func() { slog.Info("HTTP server listening", "addr", cfg.Server.Listen) diff --git a/docker-compose.yml b/docker-compose.yml index 7bcf5f71..f995e672 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -56,6 +56,8 @@ services: env_file: - path: .env required: false + devices: + - /dev/dri:/dev/dri # Optional distributed-mode example. Most single-host installs should # just use the integrated service above. diff --git a/docs/superpowers/plans/2026-05-28-library-delete-resilience.md b/docs/superpowers/plans/2026-05-28-library-delete-resilience.md new file mode 100644 index 00000000..e8211bd2 --- /dev/null +++ b/docs/superpowers/plans/2026-05-28-library-delete-resilience.md @@ -0,0 +1,724 @@ +# Resilient Library Deletion Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Rewrite `delete_library` so it deletes a library in small, autocommitted, deadlock-retrying batches instead of one multi-minute transaction that deadlocks and rolls back on large libraries. + +**Architecture:** `FolderRepository.DeleteWithStats` becomes a phased, batched operation (orphan items → folder media_files → remaining shared junctions → folder row). Each batch is one autocommit statement wrapped in a `retryOnDeadlock` helper, so locks are held briefly and the job is resumable. Two small helpers (`retryOnDeadlock`, `deleteInBatches`) are unit-tested in pure Go; the full path is verified against the live dev database. + +**Tech Stack:** Go, pgx v5 (`pgxpool`, `pgconn`), PostgreSQL. Spec: `docs/superpowers/specs/2026-05-28-library-delete-resilience-design.md`. + +All commands assume the repository root is the cwd. + +--- + +## File Structure + +- **Modify** `internal/catalog/folder_repo.go` + - Add constants `orphanDeleteBatch`, `folderChildDeleteBatch`. + - Add package vars `deadlockMaxAttempts`, `deadlockBaseBackoff` and func `retryOnDeadlock`. + - Add func `deleteInBatches`. + - Add interface `rowQuerier`; change `collectImageDirs` and `filterUnreferencedImageDirs` to take `rowQuerier` instead of `pgx.Tx`; extract `collectRawImageDirs`. + - Rewrite `DeleteWithStats`; add helpers `collectOrphanBatch`, `dirSetToSlice`. + - Leave `collectOrphanIDs` and `Delete` unchanged. +- **Create** `internal/catalog/folder_delete_test.go` — pure-Go unit tests for `retryOnDeadlock` and `deleteInBatches`. +- **Do not modify** `internal/catalog/library_repo.go` — its `collectImageDirs`/`collectOrphanIDs` calls keep compiling because `pgx.Tx` satisfies `rowQuerier`. (Verified by build.) +- **Unchanged behavior** in `internal/adminjob/library_delete.go` (executor), `internal/adminjob/runner.go` (heartbeat). + +--- + +## Task 1: Deadlock-retry helper + +**Files:** +- Modify: `internal/catalog/folder_repo.go` +- Test: `internal/catalog/folder_delete_test.go` (create) + +- [ ] **Step 1: Write the failing tests** + +Create `internal/catalog/folder_delete_test.go`: + +```go +package catalog + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgconn" +) + +// withFastDeadlockRetry shrinks retry timing/attempts for tests and restores +// the originals on cleanup. Tests using it must not call t.Parallel(). +func withFastDeadlockRetry(t *testing.T, maxAttempts int) { + t.Helper() + oldMax, oldBackoff := deadlockMaxAttempts, deadlockBaseBackoff + deadlockMaxAttempts = maxAttempts + deadlockBaseBackoff = time.Millisecond + t.Cleanup(func() { + deadlockMaxAttempts = oldMax + deadlockBaseBackoff = oldBackoff + }) +} + +func TestRetryOnDeadlockRetriesThenSucceeds(t *testing.T) { + withFastDeadlockRetry(t, 5) + calls := 0 + err := retryOnDeadlock(context.Background(), func() error { + calls++ + if calls < 3 { + return &pgconn.PgError{Code: "40P01"} + } + return nil + }) + if err != nil { + t.Fatalf("expected success, got %v", err) + } + if calls != 3 { + t.Fatalf("expected 3 calls, got %d", calls) + } +} + +func TestRetryOnDeadlockReturnsNonRetryableImmediately(t *testing.T) { + withFastDeadlockRetry(t, 5) + sentinel := errors.New("boom") + calls := 0 + err := retryOnDeadlock(context.Background(), func() error { + calls++ + return sentinel + }) + if !errors.Is(err, sentinel) { + t.Fatalf("expected sentinel, got %v", err) + } + if calls != 1 { + t.Fatalf("expected 1 call, got %d", calls) + } +} + +func TestRetryOnDeadlockGivesUpAfterMaxAttempts(t *testing.T) { + withFastDeadlockRetry(t, 4) + calls := 0 + err := retryOnDeadlock(context.Background(), func() error { + calls++ + return &pgconn.PgError{Code: "40P01"} + }) + var pgErr *pgconn.PgError + if !errors.As(err, &pgErr) || pgErr.Code != "40P01" { + t.Fatalf("expected 40P01 pg error, got %v", err) + } + if calls != 4 { + t.Fatalf("expected 4 calls, got %d", calls) + } +} + +func TestRetryOnDeadlockStopsOnCanceledContext(t *testing.T) { + withFastDeadlockRetry(t, 5) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + calls := 0 + err := retryOnDeadlock(ctx, func() error { + calls++ + return &pgconn.PgError{Code: "40P01"} + }) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got %v", err) + } + if calls != 1 { + t.Fatalf("expected 1 call before cancel, got %d", calls) + } +} +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `go test ./internal/catalog/ -run TestRetryOnDeadlock -v` +Expected: compile failure / FAIL — `undefined: retryOnDeadlock`, `deadlockMaxAttempts`, `deadlockBaseBackoff`. + +- [ ] **Step 3: Implement the helper** + +In `internal/catalog/folder_repo.go`, add near the top of the file (after the `import` block, before the sentinel errors var is fine): + +```go +// Retry parameters for transient serialization/deadlock failures. They are +// package vars (not consts) only so tests can shrink them; production code +// never mutates them. +var ( + deadlockMaxAttempts = 5 + deadlockBaseBackoff = 50 * time.Millisecond +) + +// retryOnDeadlock runs op, retrying when Postgres reports a deadlock (40P01) or +// serialization failure (40001), with exponential backoff. It returns +// immediately for any other error, and honors context cancellation between +// attempts. +func retryOnDeadlock(ctx context.Context, op func() error) error { + backoff := deadlockBaseBackoff + for attempt := 1; ; attempt++ { + err := op() + if err == nil { + return nil + } + var pgErr *pgconn.PgError + if attempt < deadlockMaxAttempts && errors.As(err, &pgErr) && + (pgErr.Code == "40P01" || pgErr.Code == "40001") { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(backoff): + } + backoff *= 2 + continue + } + return err + } +} +``` + +(`context`, `errors`, `time`, and `github.com/jackc/pgx/v5/pgconn` are already imported in this file.) + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `go test ./internal/catalog/ -run TestRetryOnDeadlock -v` +Expected: PASS (4 tests). + +- [ ] **Step 5: Format and commit** + +```bash +gofmt -w internal/catalog/folder_repo.go internal/catalog/folder_delete_test.go +git add internal/catalog/folder_repo.go internal/catalog/folder_delete_test.go +git commit -m "feat(catalog): add deadlock-retry helper for batched deletes" +``` + +--- + +## Task 2: Batched-delete loop helper + +**Files:** +- Modify: `internal/catalog/folder_repo.go` +- Test: `internal/catalog/folder_delete_test.go` + +- [ ] **Step 1: Write the failing tests** + +Append to `internal/catalog/folder_delete_test.go`: + +```go +func TestDeleteInBatchesLoopsUntilUnderBatchSize(t *testing.T) { + withFastDeadlockRetry(t, 5) + counts := []int64{5, 5, 2} + idx := 0 + total, err := deleteInBatches(context.Background(), 5, func(context.Context) (int64, error) { + n := counts[idx] + idx++ + return n, nil + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if total != 12 { + t.Fatalf("expected total 12, got %d", total) + } + if idx != 3 { + t.Fatalf("expected 3 batches, got %d", idx) + } +} + +func TestDeleteInBatchesStopsImmediatelyWhenFirstBatchUnderSize(t *testing.T) { + withFastDeadlockRetry(t, 5) + calls := 0 + total, err := deleteInBatches(context.Background(), 5, func(context.Context) (int64, error) { + calls++ + return 0, nil + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if total != 0 || calls != 1 { + t.Fatalf("expected total 0 and 1 call, got total=%d calls=%d", total, calls) + } +} + +func TestDeleteInBatchesReturnsError(t *testing.T) { + withFastDeadlockRetry(t, 5) + sentinel := errors.New("delete failed") + _, err := deleteInBatches(context.Background(), 5, func(context.Context) (int64, error) { + return 0, sentinel + }) + if !errors.Is(err, sentinel) { + t.Fatalf("expected sentinel, got %v", err) + } +} +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `go test ./internal/catalog/ -run TestDeleteInBatches -v` +Expected: FAIL — `undefined: deleteInBatches`. + +- [ ] **Step 3: Implement the helper** + +In `internal/catalog/folder_repo.go`, add after `retryOnDeadlock`: + +```go +// deleteInBatches repeatedly runs deleteBatch (each a single autocommit +// statement) until a batch removes fewer than batchSize rows. Each batch is +// retried on deadlock. It returns the total number of rows deleted. +func deleteInBatches( + ctx context.Context, + batchSize int, + deleteBatch func(ctx context.Context) (int64, error), +) (int64, error) { + var total int64 + for { + var affected int64 + if err := retryOnDeadlock(ctx, func() error { + n, e := deleteBatch(ctx) + affected = n + return e + }); err != nil { + return total, err + } + total += affected + if affected < int64(batchSize) { + return total, nil + } + } +} +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `go test ./internal/catalog/ -run TestDeleteInBatches -v` +Expected: PASS (3 tests). + +- [ ] **Step 5: Format and commit** + +```bash +gofmt -w internal/catalog/folder_repo.go internal/catalog/folder_delete_test.go +git add internal/catalog/folder_repo.go internal/catalog/folder_delete_test.go +git commit -m "feat(catalog): add deleteInBatches loop helper" +``` + +--- + +## Task 3: Generalize image-dir helpers to a querier interface + +This lets the image-dir helpers run on either a `*pgxpool.Pool` (new batched path) or a `pgx.Tx` (existing `library_repo.go` path), and splits raw collection from filtering so the batched path can collect cheaply per batch and filter once. + +**Files:** +- Modify: `internal/catalog/folder_repo.go:547-635` (the `collectImageDirs` / `filterUnreferencedImageDirs` block) + +- [ ] **Step 1: Add the `rowQuerier` interface** + +In `internal/catalog/folder_repo.go`, add (near the other type declarations, e.g. just above `collectOrphanIDs`): + +```go +// rowQuerier is satisfied by both *pgxpool.Pool and pgx.Tx, letting read +// helpers run inside or outside an explicit transaction. +type rowQuerier interface { + Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) +} +``` + +- [ ] **Step 2: Split `collectImageDirs` into raw collection + filtering** + +Replace the existing `collectImageDirs` function (currently `internal/catalog/folder_repo.go:549-584`) with these two functions: + +```go +// collectImageDirs returns S3 directory prefixes for images belonging to the +// given content IDs that are not still referenced by other surviving content. +func collectImageDirs(ctx context.Context, q rowQuerier, contentIDs []string) ([]string, error) { + dirs, err := collectRawImageDirs(ctx, q, contentIDs) + if err != nil { + return nil, err + } + return filterUnreferencedImageDirs(ctx, q, dirs, contentIDs) +} + +// collectRawImageDirs returns the deduped S3 directory prefixes referenced by +// the given content IDs (items, their seasons, and their episodes), without +// filtering out dirs still used by other content. +func collectRawImageDirs(ctx context.Context, q rowQuerier, contentIDs []string) ([]string, error) { + imgRows, err := q.Query(ctx, ` + SELECT poster_path, backdrop_path, logo_path FROM media_items WHERE content_id = ANY($1) + UNION ALL + SELECT poster_path, '', '' FROM seasons WHERE series_id = ANY($1) + UNION ALL + SELECT still_path, '', '' FROM episodes WHERE series_id = ANY($1) + `, contentIDs) + if err != nil { + return nil, fmt.Errorf("collecting image paths: %w", err) + } + defer imgRows.Close() + dirSet := make(map[string]struct{}) + for imgRows.Next() { + var p1, p2, p3 string + if err := imgRows.Scan(&p1, &p2, &p3); err != nil { + return nil, fmt.Errorf("scanning image path: %w", err) + } + for _, p := range []string{p1, p2, p3} { + if p != "" && !strings.Contains(p, "://") { + if dir := pathDir(p); dir != "" { + dirSet[dir] = struct{}{} + } + } + } + } + if err := imgRows.Err(); err != nil { + return nil, fmt.Errorf("iterating image paths: %w", err) + } + dirs := make([]string, 0, len(dirSet)) + for dir := range dirSet { + dirs = append(dirs, dir) + } + return dirs, nil +} +``` + +- [ ] **Step 3: Change `filterUnreferencedImageDirs` to take `rowQuerier`** + +In `internal/catalog/folder_repo.go`, change the signature (currently `internal/catalog/folder_repo.go:586`) from: + +```go +func filterUnreferencedImageDirs(ctx context.Context, tx pgx.Tx, dirs, deletingContentIDs []string) ([]string, error) { +``` + +to: + +```go +func filterUnreferencedImageDirs(ctx context.Context, q rowQuerier, dirs, deletingContentIDs []string) ([]string, error) { +``` + +and change the single `tx.Query(ctx, ...)` call inside it to `q.Query(ctx, ...)`. Leave the SQL and the rest of the body unchanged. + +- [ ] **Step 4: Verify the build (proves `library_repo.go` still compiles with `pgx.Tx`)** + +Run: `go build ./... && go vet ./internal/catalog/` +Expected: no errors. (`library_repo.go` passes a `pgx.Tx` to `collectImageDirs`, which now accepts `rowQuerier`; `pgx.Tx` satisfies it.) + +- [ ] **Step 5: Run the catalog tests** + +Run: `go test ./internal/catalog/...` +Expected: PASS (existing tests plus Tasks 1–2 helpers). + +- [ ] **Step 6: Format and commit** + +```bash +gofmt -w internal/catalog/folder_repo.go +git add internal/catalog/folder_repo.go +git commit -m "refactor(catalog): make image-dir helpers querier-agnostic" +``` + +--- + +## Task 4: Rewrite `DeleteWithStats` as a batched, phased operation + +**Files:** +- Modify: `internal/catalog/folder_repo.go:415-511` (the `DeleteWithStats` body) and add two helpers + two constants. + +- [ ] **Step 1: Add the batch-size constants** + +In `internal/catalog/folder_repo.go`, add near the retry vars from Task 1: + +```go +const ( + // orphanDeleteBatch is small because each media_items row cascades across + // ~15 child tables. + orphanDeleteBatch = 1000 + // folderChildDeleteBatch covers the lighter folder-scoped media_files and + // junction deletes. + folderChildDeleteBatch = 5000 +) +``` + +- [ ] **Step 2: Replace the `DeleteWithStats` body** + +Replace the entire existing `DeleteWithStats` function (`internal/catalog/folder_repo.go:415-511`) with: + +```go +func (r *FolderRepository) DeleteWithStats( + ctx context.Context, + id int, + progress func(current, total int, message string), +) (*DeleteFolderStats, error) { + stats := &DeleteFolderStats{} + + // Phase 0: preflight reads (no long-lived transaction). + if err := r.pool.QueryRow(ctx, `SELECT name FROM media_folders WHERE id = $1`, id).Scan(&stats.LibraryName); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrFolderNotFound + } + return nil, fmt.Errorf("loading folder before delete: %w", err) + } + if err := r.pool.QueryRow(ctx, `SELECT COUNT(*) FROM media_files WHERE media_folder_id = $1`, id).Scan(&stats.MediaFiles); err != nil { + return nil, fmt.Errorf("counting media files: %w", err) + } + if err := r.pool.QueryRow(ctx, `SELECT COUNT(*) FROM media_item_libraries WHERE media_folder_id = $1`, id).Scan(&stats.MediaItemLinks); err != nil { + return nil, fmt.Errorf("counting media item links: %w", err) + } + var orphanTotal int + if err := r.pool.QueryRow(ctx, ` + SELECT COUNT(*) + FROM media_item_libraries mil + WHERE mil.media_folder_id = $1 + AND NOT EXISTS ( + SELECT 1 FROM media_item_libraries other + WHERE other.content_id = mil.content_id + AND other.media_folder_id <> $1 + )`, id).Scan(&orphanTotal); err != nil { + return nil, fmt.Errorf("counting orphaned items: %w", err) + } + + // Phase 1: delete orphaned media_items in detect-then-delete batches. + // Cascade removes their junctions, provider IDs, episodes, seasons, etc. + if progress != nil { + progress(0, orphanTotal, "Deleting orphaned items") + } + rawDirs := make(map[string]struct{}) + for { + ids, err := r.collectOrphanBatch(ctx, id, orphanDeleteBatch) + if err != nil { + return nil, err + } + if len(ids) == 0 { + break + } + dirs, err := collectRawImageDirs(ctx, r.pool, ids) + if err != nil { + return nil, err + } + for _, d := range dirs { + rawDirs[d] = struct{}{} + } + if err := retryOnDeadlock(ctx, func() error { + _, e := r.pool.Exec(ctx, `DELETE FROM media_items WHERE content_id = ANY($1)`, ids) + return e + }); err != nil { + return nil, fmt.Errorf("deleting orphaned items: %w", err) + } + stats.OrphanedItems += len(ids) + if progress != nil { + progress(stats.OrphanedItems, orphanTotal, "Deleting orphaned items") + } + } + + // Filter accumulated image dirs once, now that orphans are gone, against + // any surviving content. Empty deleting-set means "exclude nothing". + if len(rawDirs) > 0 { + filtered, err := filterUnreferencedImageDirs(ctx, r.pool, dirSetToSlice(rawDirs), []string{}) + if err != nil { + return nil, err + } + stats.OrphanedImageDirs = filtered + } + + // Phase 2: delete this folder's media_files (folder-tied, not item-tied). + if progress != nil { + progress(orphanTotal, orphanTotal, "Deleting media files") + } + if _, err := deleteInBatches(ctx, folderChildDeleteBatch, func(ctx context.Context) (int64, error) { + tag, e := r.pool.Exec(ctx, ` + DELETE FROM media_files + WHERE id IN ( + SELECT id FROM media_files WHERE media_folder_id = $1 LIMIT $2 + )`, id, folderChildDeleteBatch) + if e != nil { + return 0, e + } + return tag.RowsAffected(), nil + }); err != nil { + return nil, fmt.Errorf("deleting media files: %w", err) + } + + // Phase 3: delete remaining folder memberships (shared items kept; only the + // membership in this folder is removed). + if progress != nil { + progress(orphanTotal, orphanTotal, "Removing library memberships") + } + if _, err := deleteInBatches(ctx, folderChildDeleteBatch, func(ctx context.Context) (int64, error) { + tag, e := r.pool.Exec(ctx, ` + DELETE FROM media_item_libraries + WHERE ctid IN ( + SELECT ctid FROM media_item_libraries WHERE media_folder_id = $1 LIMIT $2 + )`, id, folderChildDeleteBatch) + if e != nil { + return 0, e + } + return tag.RowsAffected(), nil + }); err != nil { + return nil, fmt.Errorf("deleting media item links: %w", err) + } + + // Phase 4: delete the now-lightweight folder row. Tolerate 0 rows so a + // resumed run that already removed it still succeeds. + if err := retryOnDeadlock(ctx, func() error { + _, e := r.pool.Exec(ctx, `DELETE FROM media_folders WHERE id = $1`, id) + return e + }); err != nil { + return nil, fmt.Errorf("deleting folder: %w", err) + } + + if progress != nil { + progress(orphanTotal, orphanTotal, "Library deletion completed") + } + return stats, nil +} + +// collectOrphanBatch returns up to limit content IDs whose only library +// membership is the given folder. +func (r *FolderRepository) collectOrphanBatch(ctx context.Context, folderID, limit int) ([]string, error) { + rows, err := r.pool.Query(ctx, ` + SELECT mil.content_id + FROM media_item_libraries mil + WHERE mil.media_folder_id = $1 + AND NOT EXISTS ( + SELECT 1 FROM media_item_libraries other + WHERE other.content_id = mil.content_id + AND other.media_folder_id <> $1 + ) + LIMIT $2`, folderID, limit) + if err != nil { + return nil, fmt.Errorf("finding orphaned items: %w", err) + } + defer rows.Close() + var ids []string + for rows.Next() { + var contentID string + if err := rows.Scan(&contentID); err != nil { + return nil, fmt.Errorf("scanning orphan content_id: %w", err) + } + ids = append(ids, contentID) + } + return ids, rows.Err() +} + +// dirSetToSlice returns the keys of set as a slice (nil if empty). +func dirSetToSlice(set map[string]struct{}) []string { + if len(set) == 0 { + return nil + } + dirs := make([]string, 0, len(set)) + for d := range set { + dirs = append(dirs, d) + } + return dirs +} +``` + +Note: this removes the old single-transaction body (`tx, err := r.pool.Begin(...)`, the inline orphan-collection loop, the `collectImageDirs` call, the CASCADE folder drop, and `tx.Commit`). `collectOrphanIDs` and `Delete` are untouched. `collectImageDirs` remains (still used by `library_repo.go`). + +- [ ] **Step 3: Verify the build and vet** + +Run: `go build ./... && go vet ./internal/catalog/` +Expected: no errors. If `go vet` flags an unused function, confirm it is genuinely unused before removing — `collectImageDirs`, `collectOrphanIDs`, and `pathDir` must all remain (still referenced). + +- [ ] **Step 4: Run the full catalog + adminjob test suites** + +Run: `go test ./internal/catalog/... ./internal/adminjob/...` +Expected: PASS. (No DB-backed test exercises `DeleteWithStats`; the helper unit tests and existing tests must stay green.) + +- [ ] **Step 5: Lint** + +Run: `golangci-lint run ./internal/catalog/...` +Expected: no findings. (If `golangci-lint` is unavailable, run `gofmt -l internal/catalog/` and ensure it prints nothing.) + +- [ ] **Step 6: Format and commit** + +```bash +gofmt -w internal/catalog/folder_repo.go +git add internal/catalog/folder_repo.go +git commit -m "fix(catalog): delete libraries in deadlock-retrying batches + +Replaces the single multi-minute delete transaction with phased, batched +autocommit deletes (orphan items, media files, memberships, folder row), +each retried on deadlock. Holds only short locks, survives concurrent +writers, and is resumable on failure." +``` + +--- + +## Task 5: Verify on the dev server + +This is the integration test (no catalog DB harness exists). Follow the deployment-debugging runbook. The Audiobooks library (folder id 8) is currently stuck with ~249K items / ~339K files. + +- [ ] **Step 1: Deploy the branch to dev** + +Run: `make dev-deploy` +Expected: builds `silo:dev-local`, recreates the `silo` service. Then confirm health: +`ssh "$DEV_HOST" 'curl -s http://localhost:8090/api/v1/ready'` + +- [ ] **Step 2: Capture the starting counts** + +Run (psql via the dev compose `postgres` service, user/db `continuum`): + +```sql +SELECT + (SELECT COUNT(*) FROM media_files WHERE media_folder_id = 8) AS files, + (SELECT COUNT(*) FROM media_item_libraries WHERE media_folder_id = 8) AS links, + (SELECT EXISTS (SELECT 1 FROM media_folders WHERE id = 8)) AS folder_exists; +``` + +Expected: non-zero files/links, `folder_exists = t`. + +- [ ] **Step 3: Trigger the deletion** + +Re-trigger the library deletion through the normal path (admin UI "delete library" for Audiobooks, or the admin delete endpoint). Confirm a new job is queued: + +```sql +SELECT id, job_type, status, requested_at +FROM admin_jobs WHERE job_type = 'delete_library' +ORDER BY requested_at DESC LIMIT 3; +``` + +- [ ] **Step 4: Monitor to completion** + +Watch the job and the draining counts (re-run periodically): + +```sql +SELECT id, status, progress_current, progress_total, error_message, heartbeat_at +FROM admin_jobs WHERE job_type = 'delete_library' +ORDER BY requested_at DESC LIMIT 1; +``` + +```sql +SELECT COUNT(*) FROM media_item_libraries WHERE media_folder_id = 8; +``` + +Also confirm deadlock retries (if any) are recovered rather than fatal: +`ssh "$DEV_HOST" "$DEV_COMPOSE_CMD logs postgres 2>&1 | grep -i deadlock | tail -20"` +Expected: the job reaches `status = completed`; counts trend to 0; any deadlocks are transient (the job does not fail). + +- [ ] **Step 5: Confirm full removal** + +```sql +SELECT + (SELECT COUNT(*) FROM media_files WHERE media_folder_id = 8) AS files, + (SELECT COUNT(*) FROM media_item_libraries WHERE media_folder_id = 8) AS links, + (SELECT EXISTS (SELECT 1 FROM media_folders WHERE id = 8)) AS folder_exists; +``` + +Expected: `files = 0`, `links = 0`, `folder_exists = f`. + +- [ ] **Step 6: Report** + +Summarize: starting counts, time to completion, number of batches/retries observed, and final state. No commit (verification only). + +--- + +## Self-Review + +**Spec coverage:** +- Phased batched deletion (Phases 0–4) → Task 4. ✔ +- `retryOnDeadlock` (40P01/40001, 5 attempts, backoff, ctx-aware) → Task 1. ✔ +- Batch sizes `orphanDeleteBatch=1000`, `folderChildDeleteBatch=5000` → Task 4 Step 1. ✔ +- Autocommit per batch / no long transaction → Task 4 Step 2 (uses `r.pool.Exec`, no `Begin`). ✔ +- Stats + progress + heartbeat reliance → Task 4 Step 2; runner heartbeat unchanged. ✔ +- Resumability (folder row last, tolerate 0 rows) → Task 4 Phase 4. ✔ +- Unit tests for retry + batch loop → Tasks 1–2. ✔ +- `collectImageDirs` shared with `library_repo.go` preserved → Task 3 (interface widening), Task 4 Step 3 build check. ✔ +- Dev integration verification → Task 5. ✔ +- Out of scope (writer coordination, runner/executor changes, schema changes) → not present in any task. ✔ + +**Placeholder scan:** No TBD/TODO; every code step has complete code; every command has expected output. ✔ + +**Type consistency:** `retryOnDeadlock(ctx, func() error) error`, `deleteInBatches(ctx, int, func(ctx) (int64,error)) (int64,error)`, `rowQuerier{ Query(ctx,string,...any)(pgx.Rows,error) }`, `collectRawImageDirs`/`collectImageDirs`/`filterUnreferencedImageDirs(ctx, rowQuerier, ...)`, `collectOrphanBatch(ctx,int,int)([]string,error)`, `dirSetToSlice(map[string]struct{})[]string`, consts `orphanDeleteBatch`/`folderChildDeleteBatch`, vars `deadlockMaxAttempts`/`deadlockBaseBackoff` — names used consistently across tasks. ✔ diff --git a/docs/superpowers/specs/2026-05-28-library-delete-resilience-design.md b/docs/superpowers/specs/2026-05-28-library-delete-resilience-design.md new file mode 100644 index 00000000..1bf40e61 --- /dev/null +++ b/docs/superpowers/specs/2026-05-28-library-delete-resilience-design.md @@ -0,0 +1,143 @@ +# Resilient Library Deletion Design + +## Goal + +Make `delete_library` survive concurrent database writers and large libraries. Today it deletes an entire library inside a single multi-minute transaction; on a large library this deadlocks against ordinary background writes and rolls back, making zero progress. Rewrite the deletion to run in small, autocommitted batches with per-batch deadlock retry, so it completes reliably, holds only short locks, and resumes cleanly if interrupted. + +## Background + +A `delete_library` admin job repeatedly failed with: + +``` +deleting orphaned items: ERROR: deadlock detected (SQLSTATE 40P01) +``` + +Diagnosis on a library of ~249K items / ~339K media files: + +- The whole deletion runs in one transaction in `FolderRepository.DeleteWithStats` (`internal/catalog/folder_repo.go`): collect orphan IDs, drop the folder (CASCADE), then `DELETE FROM media_items WHERE content_id = ANY($1)`. +- `media_items` cascades to 15 child tables, so a bulk delete holds a very large lock set for minutes. +- The Postgres deadlock log shows the cycle: the delete cascade into `media_item_provider_ids` versus a concurrent `INSERT INTO media_item_provider_ids` from a routine background task. Each ~7-minute attempt was guaranteed to overlap an ordinary write and lose. +- Because it is one transaction, every failure rolls back completely: no progress, not resumable. + +The fix targets the deletion mechanism only. Coordinating or pausing background writers is explicitly out of scope (see Out of Scope) — a batched, retrying delete is robust against *any* concurrent writer, not just today's tasks. + +## Approach + +Replace the single transaction with a sequence of phases. Each phase deletes in bounded batches, and each batch is one autocommit statement wrapped in a deadlock-retry helper. Short lock windows make deadlocks rare; when one occurs, only that small batch retries. + +Scope of the rewrite: + +- Rewrite `FolderRepository.DeleteWithStats` in `internal/catalog/folder_repo.go`. +- Add an unexported `retryOnDeadlock` helper in the `catalog` package. +- `FolderRepository.Delete` (thin wrapper), the `DeleteFolderStats` struct, and the `delete_library` executor in `internal/adminjob/library_delete.go` keep their current signatures and behavior. S3 image-cleanup queueing and generated-home-section cleanup are untouched. + +## Deletion phases + +`DeleteWithStats(ctx, id, progress)` performs, in order: + +### Phase 0 — preflight (reads only) + +- Load the folder name; return `ErrFolderNotFound` if the row is gone. +- `COUNT` media files (`WHERE media_folder_id = $1`), item links (`WHERE media_folder_id = $1`), and orphans, to populate the result payload and the progress denominator. + +### Phase 1 — orphaned `media_items` (batched loop) + +Repeat until no rows are returned: + +```sql +SELECT mil.content_id +FROM media_item_libraries mil +WHERE mil.media_folder_id = $1 + AND NOT EXISTS ( + SELECT 1 FROM media_item_libraries other + WHERE other.content_id = mil.content_id + AND other.media_folder_id <> $1 + ) +LIMIT $2 -- orphanDeleteBatch +``` + +For each returned batch: collect S3 image dirs for those IDs (existing `collectImageDirs`, accumulated into a deduped set), then + +```sql +DELETE FROM media_items WHERE content_id = ANY($1) +``` + +The FK cascade removes each item's `media_item_libraries` (including this folder's), `media_item_provider_ids`, episodes, seasons, embeddings, and the rest of the 15 child tables. Deleting orphan items first also clears their junction rows, so the next iteration's detection query naturally advances. Orphan detection must run before Phase 3 touches shared junctions, so the "only in this folder" test stays valid. + +### Phase 2 — folder `media_files` (batched loop) + +`media_files` is folder-tied (`media_folder_id` FK, `ON DELETE CASCADE`), not item-tied, so it is deleted by folder. Repeat until a batch deletes fewer than the batch size: + +```sql +DELETE FROM media_files +WHERE id IN ( + SELECT id FROM media_files WHERE media_folder_id = $1 LIMIT $2 -- folderChildDeleteBatch +) +``` + +### Phase 3 — remaining shared-item junctions (batched loop) + +After Phase 1, the only `media_item_libraries` rows left for this folder belong to items that also live in other folders. Delete them in batches (dropping the membership only, never the shared item): + +```sql +DELETE FROM media_item_libraries +WHERE ctid IN ( + SELECT ctid FROM media_item_libraries WHERE media_folder_id = $1 LIMIT $2 -- folderChildDeleteBatch +) +``` + +### Phase 4 — folder row + +```sql +DELETE FROM media_folders WHERE id = $1 +``` + +Now lightweight. Tolerates 0 rows affected (a resumed re-run may already have removed it) — it does not re-raise `ErrFolderNotFound` here. The folder row is deleted **last**, so a partial deletion leaves the library present-but-disabled and a re-run continues. + +## Deadlock-retry helper + +New unexported helper in the `catalog` package: + +```go +func retryOnDeadlock(ctx context.Context, op func() error) error +``` + +- Retries when the error is a `*pgconn.PgError` with code `40P01` (deadlock) or `40001` (serialization failure). +- Up to 5 attempts, exponential backoff starting at 50ms, doubling each retry. +- Honors `ctx` cancellation between attempts (returns `ctx.Err()`). +- Any non-retryable error returns immediately; the final retryable error is returned after the cap. + +Every batch statement runs through `retryOnDeadlock`. It has a single caller today, so it stays package-local; it can be promoted to a shared DB utility if a second caller appears. + +## Batch sizes + +Named constants in `internal/catalog/folder_repo.go`, tunable: + +- `orphanDeleteBatch = 1000` — small, because each `media_items` row cascades across 15 tables. +- `folderChildDeleteBatch = 5000` — for the lighter folder-scoped `media_files` and junction deletes. + +## Transaction model, stats, progress + +- No surrounding transaction. Each batch is a single `pool.Exec` (its own implicit transaction), mirroring the existing batched-cleanup pattern in `internal/opslog/cleanup.go`. This is the core change that eliminates the multi-minute lock window. +- Stats: `MediaFiles` and `MediaItemLinks` come from the Phase 0 counts; `OrphanedItems` accumulates across Phase 1 batches; `OrphanedImageDirs` is the accumulated deduped dir set. Image-dir volume for the affected workloads is modest; the deduped set is the bounded-memory point of note. +- Progress: the callback reports orphan items processed against the Phase 0 orphan total, plus phase-transition messages. The admin-job runner's existing 10s heartbeat keeps the job alive across the drain. + +## Error handling & resumability + +- Because batches autocommit, a failure or cancellation persists partial progress. There is no rollback of hundreds of thousands of rows and no long lock hold. +- The job is resumable by re-running it: the folder still exists (deleted last), and Phase 1 re-detects whatever orphans remain. +- New items inserted concurrently during deletion are picked up by the loop's re-detection. + +## Testing + +The `catalog` package uses pure-Go unit tests; there is no Postgres integration harness, so tests avoid a live database. + +- **`retryOnDeadlock` unit test:** a fake `op` returns a `*pgconn.PgError{Code: "40P01"}` N times then `nil` — assert it retries, respects the attempt cap, backs off, and returns immediately on a non-retryable error; a separate case asserts `ctx` cancellation stops retries. +- **Batch-loop unit test:** extract the "loop until a batch is under the batch size" iteration behind a small injected exec function so the looping/chunking logic is testable without Postgres. +- **Integration verification on dev:** deploy and re-trigger the deletion against the real stuck library; confirm it drains to completion via application logs and database row counts. This is the integration test, given the absence of a catalog DB harness. (Commands assume the repository root is the cwd and follow the deployment-debugging runbook.) + +## Out of scope + +- Pausing or coordinating background writers (e.g., having metadata/scan tasks skip disabled or pending-deletion libraries). The batched, retrying delete is robust against any concurrent writer; writer coordination is a separate potential improvement. +- Changing the admin-job runner, the executor's image-cleanup queueing, or home-section cleanup. +- Schema or FK changes. diff --git a/internal/api/handlers/admin.go b/internal/api/handlers/admin.go index e22d1970..473e0dd2 100644 --- a/internal/api/handlers/admin.go +++ b/internal/api/handlers/admin.go @@ -966,6 +966,7 @@ type UpdateItemMetadataRequest struct { FirstAirDate *string `json:"first_air_date"` LastAirDate *string `json:"last_air_date"` AirTime *string `json:"air_time"` + AirTimezone *string `json:"air_timezone"` AirDate *string `json:"air_date"` Status *string `json:"status"` RatingIMDB *float64 `json:"rating_imdb"` @@ -993,6 +994,14 @@ func (h *AdminHandler) HandleUpdateItemMetadata(w http.ResponseWriter, r *http.R writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body") return } + if req.AirTimezone != nil { + trimmed := strings.TrimSpace(*req.AirTimezone) + req.AirTimezone = &trimmed + if !catalog.ValidateAirTimezone(trimmed) { + writeError(w, http.StatusBadRequest, "bad_request", "air_timezone must be a valid IANA timezone") + return + } + } upd := catalog.MetadataUpdate{ Title: req.Title, SortTitle: req.SortTitle, OriginalTitle: req.OriginalTitle, @@ -1000,7 +1009,7 @@ func (h *AdminHandler) HandleUpdateItemMetadata(w http.ResponseWriter, r *http.R Year: req.Year, Runtime: req.Runtime, Genres: req.Genres, Studios: req.Studios, Networks: req.Networks, Countries: req.Countries, ReleaseDate: req.ReleaseDate, FirstAirDate: req.FirstAirDate, LastAirDate: req.LastAirDate, - AirTime: req.AirTime, + AirTime: req.AirTime, AirTimezone: req.AirTimezone, AirDate: req.AirDate, Status: req.Status, RatingIMDB: req.RatingIMDB, RatingTMDB: req.RatingTMDB, RatingRTCritic: req.RatingRTCritic, RatingRTAudience: req.RatingRTAudience, diff --git a/internal/api/handlers/calendar.go b/internal/api/handlers/calendar.go index 47419ce7..3812c668 100644 --- a/internal/api/handlers/calendar.go +++ b/internal/api/handlers/calendar.go @@ -3,6 +3,7 @@ package handlers import ( "context" "net/http" + "sort" "strconv" "time" @@ -36,6 +37,9 @@ type calendarEventResponse struct { EpisodeNumber *int `json:"episode_number,omitempty"` AirDate string `json:"air_date"` AirTime *string `json:"air_time,omitempty"` + AirAt *string `json:"air_at,omitempty"` + AirTimezone *string `json:"air_timezone,omitempty"` + LocalAirDate string `json:"local_air_date"` PosterURL string `json:"poster_url,omitempty"` PosterThumbhash string `json:"poster_thumbhash,omitempty"` Badges []string `json:"badges"` @@ -91,9 +95,11 @@ func (h *CalendarHandler) HandleGetCalendar(w http.ResponseWriter, r *http.Reque return } + viewerLocation := catalog.CalendarLocation(q.Get("timezone")) + cf := catalog.CalendarFilter{ - Start: start, - End: end, + Start: start.AddDate(0, 0, -2), + End: end.AddDate(0, 0, 2), Filter: filter, AllowedLibraryIDs: af.AllowedLibraryIDs, DisabledLibraryIDs: af.DisabledLibraryIDs, @@ -118,11 +124,11 @@ func (h *CalendarHandler) HandleGetCalendar(w http.ResponseWriter, r *http.Reque } // Group events by date and build response. - days := groupEventsByDate(events, r, h.detailSvc) + days := groupEventsByDate(events, r, h.detailSvc, start, end, viewerLocation) writeJSON(w, http.StatusOK, calendarResponse{Events: days}) } -func groupEventsByDate(events []catalog.CalendarEvent, r *http.Request, detailSvc *catalog.DetailService) []calendarDayResponse { +func groupEventsByDate(events []catalog.CalendarEvent, r *http.Request, detailSvc *catalog.DetailService, start, end time.Time, viewerLocation *time.Location) []calendarDayResponse { if len(events) == 0 { return []calendarDayResponse{} } @@ -144,17 +150,75 @@ func groupEventsByDate(events []catalog.CalendarEvent, r *http.Request, detailSv posterURLs = detailSvc.PresignImageURLs(r.Context(), posterPaths, "poster", "small") } + type preparedCalendarEvent struct { + event catalog.CalendarEvent + localDate string + sourceDate string + airAt *time.Time + airAtString *string + } + + prepared := make([]preparedCalendarEvent, 0, len(events)) + for _, ev := range events { + airAt := catalog.CalendarEventAirAt(ev.AirDate, ev.AirTime, ev.AirTimezone) + localDateTime := ev.AirDate + if airAt != nil { + localDateTime = airAt.In(viewerLocation) + } + localDate := localDateTime.Format("2006-01-02") + if localDate < start.Format("2006-01-02") || localDate > end.Format("2006-01-02") { + continue + } + var airAtString *string + if airAt != nil { + formatted := airAt.Format(time.RFC3339) + airAtString = &formatted + } + prepared = append(prepared, preparedCalendarEvent{ + event: ev, + localDate: localDate, + sourceDate: ev.AirDate.Format("2006-01-02"), + airAt: airAt, + airAtString: airAtString, + }) + } + if len(prepared) == 0 { + return []calendarDayResponse{} + } + + sort.SliceStable(prepared, func(i, j int) bool { + left, right := prepared[i], prepared[j] + if left.localDate != right.localDate { + return left.localDate < right.localDate + } + if left.airAt != nil && right.airAt != nil && !left.airAt.Equal(*right.airAt) { + return left.airAt.Before(*right.airAt) + } + if (left.airAt != nil) != (right.airAt != nil) { + return left.airAt != nil + } + if left.event.Title != right.event.Title { + return left.event.Title < right.event.Title + } + if left.event.SeasonNumber != nil && right.event.SeasonNumber != nil && *left.event.SeasonNumber != *right.event.SeasonNumber { + return *left.event.SeasonNumber < *right.event.SeasonNumber + } + if left.event.EpisodeNumber != nil && right.event.EpisodeNumber != nil && *left.event.EpisodeNumber != *right.event.EpisodeNumber { + return *left.event.EpisodeNumber < *right.event.EpisodeNumber + } + return left.event.ContentID < right.event.ContentID + }) + var days []calendarDayResponse var currentDay *calendarDayResponse - for _, ev := range events { - dateStr := ev.AirDate.Format("2006-01-02") - - if currentDay == nil || currentDay.Date != dateStr { + for _, item := range prepared { + ev := item.event + if currentDay == nil || currentDay.Date != item.localDate { if currentDay != nil { days = append(days, *currentDay) } - currentDay = &calendarDayResponse{Date: dateStr} + currentDay = &calendarDayResponse{Date: item.localDate} } badges := buildBadges(ev) @@ -167,8 +231,11 @@ func groupEventsByDate(events []catalog.CalendarEvent, r *http.Request, detailSv SeriesID: ev.SeriesID, SeasonNumber: ev.SeasonNumber, EpisodeNumber: ev.EpisodeNumber, - AirDate: dateStr, + AirDate: item.sourceDate, AirTime: ev.AirTime, + AirAt: item.airAtString, + AirTimezone: ev.AirTimezone, + LocalAirDate: item.localDate, PosterURL: posterURLs[ev.PosterPath], PosterThumbhash: ev.PosterThumbhash, Badges: badges, diff --git a/internal/api/handlers/calendar_test.go b/internal/api/handlers/calendar_test.go index e9813b31..f9d2acd8 100644 --- a/internal/api/handlers/calendar_test.go +++ b/internal/api/handlers/calendar_test.go @@ -110,6 +110,62 @@ func TestHandleGetCalendar_ReturnsEmptyEvents(t *testing.T) { } } +func TestHandleGetCalendar_ExpandsRepoRangeAndGroupsByViewerLocalDate(t *testing.T) { + t.Parallel() + + airTime := "00:30" + airTimezone := "Asia/Tokyo" + repo := &stubCalendarRepo{ + events: []catalog.CalendarEvent{ + { + ContentID: "episode-1", + Type: "episode", + Title: "Series", + SeriesID: ptrString("series-1"), + AirDate: time.Date(2026, time.January, 2, 0, 0, 0, 0, time.UTC), + AirTime: &airTime, + AirTimezone: &airTimezone, + }, + }, + } + handler := &CalendarHandler{repo: repo} + req := httptest.NewRequest(http.MethodGet, "/calendar?start=2026-01-01&end=2026-01-01&timezone=America/New_York", nil) + rec := httptest.NewRecorder() + + handler.HandleGetCalendar(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want %d: %s", rec.Code, http.StatusOK, rec.Body.String()) + } + if got := repo.last.Start.Format("2006-01-02"); got != "2025-12-30" { + t.Fatalf("repo start = %s, want expanded 2025-12-30", got) + } + if got := repo.last.End.Format("2006-01-02"); got != "2026-01-03" { + t.Fatalf("repo end = %s, want expanded 2026-01-03", got) + } + + var resp calendarResponse + if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil { + t.Fatalf("decode response: %v", err) + } + if len(resp.Events) != 1 { + t.Fatalf("days len = %d, want 1", len(resp.Events)) + } + if resp.Events[0].Date != "2026-01-01" { + t.Fatalf("day = %q, want viewer-local 2026-01-01", resp.Events[0].Date) + } + item := resp.Events[0].Items[0] + if item.AirDate != "2026-01-02" { + t.Fatalf("air_date = %q, want source date 2026-01-02", item.AirDate) + } + if item.LocalAirDate != "2026-01-01" { + t.Fatalf("local_air_date = %q, want 2026-01-01", item.LocalAirDate) + } + if item.AirAt == nil || *item.AirAt != "2026-01-01T15:30:00Z" { + t.Fatalf("air_at = %v, want 2026-01-01T15:30:00Z", item.AirAt) + } +} + func TestHandleGetCalendar_GroupsEventsAndBatchResolvesCardPosters(t *testing.T) { t.Parallel() diff --git a/internal/api/handlers/items.go b/internal/api/handlers/items.go index ffa4ffe5..e99f0c51 100644 --- a/internal/api/handlers/items.go +++ b/internal/api/handlers/items.go @@ -996,16 +996,11 @@ func featuredPosterPath(path string) string { } // featuredBackdropPath converts an S3 backdrop path from original to w1920 for -// featured/hero contexts (displayed at full viewport width). -// Full URLs (TMDB/TVDB) and plugin-prefixed paths are returned as-is. +// featured/hero contexts (displayed at full viewport width). Episode stills +// used as backdrops lack a w1920 variant and clamp to their largest cached +// size. Full URLs (TMDB/TVDB) and plugin-prefixed paths are returned as-is. func featuredBackdropPath(path string) string { - if strings.HasPrefix(path, "http://") || strings.HasPrefix(path, "https://") { - return path - } - if strings.Contains(path, "://") { - return path - } - return strings.Replace(path, "/original.", "/w1920.", 1) + return catalog.BackdropVariantPath(path, "w1920") } // presignURL resolves an image path to a usable URL, delegating to the diff --git a/internal/api/handlers/sections.go b/internal/api/handlers/sections.go index 5ae1cb9e..3441b026 100644 --- a/internal/api/handlers/sections.go +++ b/internal/api/handlers/sections.go @@ -1359,15 +1359,7 @@ func (h *SectionHandler) toSectionItemResponse(sectionType sections.SectionType, // uses the cached w1280 backdrop for Continue Watching / Next Up rows. func sectionBackdropPath(sectionType sections.SectionType, path string) string { if sectionType == sections.SectionContinueWatching || sectionType == sections.SectionNextUp { - if strings.HasPrefix(path, "http://") || strings.HasPrefix(path, "https://") { - return path - } - // Plugin-prefixed paths pass through; the resolver still receives the - // existing "featured" semantic variant. - if strings.Contains(path, "://") { - return path - } - return strings.Replace(path, "/original.", "/w1280.", 1) + return catalog.BackdropVariantPath(path, "w1280") } return featuredBackdropPath(path) } diff --git a/internal/api/handlers/sections_test.go b/internal/api/handlers/sections_test.go index a46f5d2b..982e565c 100644 --- a/internal/api/handlers/sections_test.go +++ b/internal/api/handlers/sections_test.go @@ -52,6 +52,18 @@ func TestSectionBackdropPathUsesExpectedVariants(t *testing.T) { path: "/tmdb/shows/1399/backdrop/original.jpg", want: "/tmdb/shows/1399/backdrop/w1920.jpg", }, + { + name: "continue watching episode still clamps to w500", + sectionType: sections.SectionContinueWatching, + path: "tvdb/series/73141/seasons/22/episodes/9/still/original.webp", + want: "tvdb/series/73141/seasons/22/episodes/9/still/w500.webp", + }, + { + name: "featured section episode still clamps to w500", + sectionType: sections.SectionRecentlyAdded, + path: "tvdb/series/73141/seasons/22/episodes/9/still/original.webp", + want: "tvdb/series/73141/seasons/22/episodes/9/still/w500.webp", + }, { name: "http paths pass through", sectionType: sections.SectionContinueWatching, diff --git a/internal/catalog/air_schedule.go b/internal/catalog/air_schedule.go new file mode 100644 index 00000000..b4a9bade --- /dev/null +++ b/internal/catalog/air_schedule.go @@ -0,0 +1,114 @@ +package catalog + +import ( + "strings" + "time" +) + +var networkAirTimezones = map[string]string{ + "abc": "America/New_York", + "cbs": "America/New_York", + "nbc": "America/New_York", + "fox": "America/New_York", + "the cw": "America/New_York", + "cw": "America/New_York", + "hbo": "America/New_York", + "showtime": "America/New_York", + "fx": "America/New_York", + "amc": "America/New_York", + "bbc": "Europe/London", + "bbc one": "Europe/London", + "bbc two": "Europe/London", + "itv": "Europe/London", + "channel 4": "Europe/London", +} + +var countryAirTimezones = map[string]string{ + "france": "Europe/Paris", + "japan": "Asia/Tokyo", + "south korea": "Asia/Seoul", + "korea": "Asia/Seoul", + "united kingdom": "Europe/London", + "uk": "Europe/London", +} + +// ValidateAirTimezone reports whether tz is empty or a valid IANA timezone. +func ValidateAirTimezone(tz string) bool { + tz = strings.TrimSpace(tz) + if tz == "" { + return true + } + _, err := time.LoadLocation(tz) + return err == nil +} + +// InferAirTimezone returns a conservative source airing timezone for series metadata. +func InferAirTimezone(networks, countries []string) string { + for _, network := range networks { + if tz := networkAirTimezones[normalizeScheduleLookupKey(network)]; tz != "" { + return tz + } + } + for _, country := range countries { + if tz := countryAirTimezones[normalizeScheduleLookupKey(country)]; tz != "" { + return tz + } + } + return "" +} + +// CalendarEventAirAt combines a source date, source wall-clock time, and source timezone. +func CalendarEventAirAt(airDate time.Time, airTime, airTimezone *string) *time.Time { + if airTime == nil || strings.TrimSpace(*airTime) == "" { + return nil + } + if airTimezone == nil || strings.TrimSpace(*airTimezone) == "" { + return nil + } + loc, err := time.LoadLocation(strings.TrimSpace(*airTimezone)) + if err != nil { + return nil + } + parsed, ok := parseAirTime(*airTime) + if !ok { + return nil + } + local := time.Date( + airDate.Year(), + airDate.Month(), + airDate.Day(), + parsed.Hour(), + parsed.Minute(), + parsed.Second(), + 0, + loc, + ) + utc := local.UTC() + return &utc +} + +func CalendarLocation(name string) *time.Location { + if strings.TrimSpace(name) == "" { + return time.UTC + } + loc, err := time.LoadLocation(strings.TrimSpace(name)) + if err != nil { + return time.UTC + } + return loc +} + +func parseAirTime(value string) (time.Time, bool) { + value = strings.TrimSpace(value) + for _, layout := range []string{"15:04:05", "15:04"} { + parsed, err := time.Parse(layout, value) + if err == nil { + return parsed, true + } + } + return time.Time{}, false +} + +func normalizeScheduleLookupKey(value string) string { + return strings.ToLower(strings.TrimSpace(value)) +} diff --git a/internal/catalog/air_schedule_test.go b/internal/catalog/air_schedule_test.go new file mode 100644 index 00000000..65fa139b --- /dev/null +++ b/internal/catalog/air_schedule_test.go @@ -0,0 +1,71 @@ +package catalog + +import ( + "testing" + "time" +) + +func TestInferAirTimezone_UsesNetworkBeforeCountry(t *testing.T) { + t.Parallel() + + got := InferAirTimezone([]string{"BBC One"}, []string{"Japan"}) + if got != "Europe/London" { + t.Fatalf("timezone = %q, want Europe/London", got) + } +} + +func TestInferAirTimezone_UsesSingleTimezoneCountry(t *testing.T) { + t.Parallel() + + got := InferAirTimezone(nil, []string{"South Korea"}) + if got != "Asia/Seoul" { + t.Fatalf("timezone = %q, want Asia/Seoul", got) + } +} + +func TestValidateAirTimezone(t *testing.T) { + t.Parallel() + + if !ValidateAirTimezone("America/New_York") { + t.Fatal("expected America/New_York to be valid") + } + if ValidateAirTimezone("Eastern") { + t.Fatal("expected Eastern to be invalid") + } +} + +func TestCalendarEventAirAt_ConvertsSourceTimezoneToUTC(t *testing.T) { + t.Parallel() + + airDate := time.Date(2026, time.May, 28, 0, 0, 0, 0, time.UTC) + airTime := "23:30" + airTimezone := "Asia/Tokyo" + + got := CalendarEventAirAt(airDate, &airTime, &airTimezone) + if got == nil { + t.Fatal("expected air_at") + } + + want := time.Date(2026, time.May, 28, 14, 30, 0, 0, time.UTC) + if !got.Equal(want) { + t.Fatalf("air_at = %s, want %s", got.Format(time.RFC3339), want.Format(time.RFC3339)) + } +} + +func TestCalendarEventAirAt_UsesDSTForSourceDate(t *testing.T) { + t.Parallel() + + airTime := "20:00" + airTimezone := "America/New_York" + winter := CalendarEventAirAt(time.Date(2026, time.January, 15, 0, 0, 0, 0, time.UTC), &airTime, &airTimezone) + summer := CalendarEventAirAt(time.Date(2026, time.July, 15, 0, 0, 0, 0, time.UTC), &airTime, &airTimezone) + if winter == nil || summer == nil { + t.Fatal("expected air_at values") + } + if winter.Hour() != 1 { + t.Fatalf("winter UTC hour = %d, want 1", winter.Hour()) + } + if summer.Hour() != 0 { + t.Fatalf("summer UTC hour = %d, want 0", summer.Hour()) + } +} diff --git a/internal/catalog/calendar_repo.go b/internal/catalog/calendar_repo.go index 2247920a..46dd7069 100644 --- a/internal/catalog/calendar_repo.go +++ b/internal/catalog/calendar_repo.go @@ -21,6 +21,7 @@ type CalendarEvent struct { EpisodeNumber *int AirDate time.Time AirTime *string + AirTimezone *string PosterPath string PosterThumbhash string IsPremiere bool @@ -70,7 +71,7 @@ func (r *CalendarRepository) ListEvents(ctx context.Context, f CalendarFilter) ( var seasonNum, episodeNum *int if err := rows.Scan( &ev.ContentID, &ev.Type, &ev.Title, &episodeTitle, &seriesID, - &seasonNum, &episodeNum, &ev.AirDate, &ev.AirTime, + &seasonNum, &episodeNum, &ev.AirDate, &ev.AirTime, &ev.AirTimezone, &ev.PosterPath, &ev.PosterThumbhash, &ev.IsPremiere, &ev.IsFinale, ); err != nil { @@ -126,7 +127,7 @@ func (r *CalendarRepository) buildListEventsQuery(f CalendarFilter) (string, []a WHERE e.episode_number = 1 AND e.air_date IS NOT NULL ) SELECT content_id, type, title, episode_title, series_id, - season_number, episode_number, air_date, air_time, + season_number, episode_number, air_date, air_time, air_timezone, poster_path, poster_thumbhash, is_premiere, is_finale FROM ( @@ -150,7 +151,7 @@ func (r *CalendarRepository) buildMovieBranch(startArg, endArg int, f CalendarFi return fmt.Sprintf(`SELECT mi.content_id, 'movie'::text AS type, mi.title, NULL::text AS episode_title, NULL::text AS series_id, NULL::int AS season_number, NULL::int AS episode_number, - mi.release_date AS air_date, NULL::text AS air_time, + mi.release_date AS air_date, NULL::text AS air_time, NULL::text AS air_timezone, mi.poster_path, mi.poster_thumbhash, FALSE AS is_premiere, FALSE AS is_finale FROM media_items mi @@ -168,7 +169,7 @@ func (r *CalendarRepository) buildFilteredEpisodesCTE(startArg, endArg int, f Ca return fmt.Sprintf(`SELECT e.content_id, e.series_id, e.season_number, e.episode_number, e.title AS episode_title, e.air_date, - mi.title AS title, mi.air_time, + mi.title AS title, mi.air_time, mi.air_timezone, mi.poster_path, mi.poster_thumbhash FROM episodes e JOIN media_items mi ON mi.content_id = e.series_id @@ -179,7 +180,7 @@ func (r *CalendarRepository) buildEpisodeBranch() string { return `SELECT fe.content_id, 'episode'::text AS type, fe.title, fe.episode_title, fe.series_id, fe.season_number, fe.episode_number, - fe.air_date, fe.air_time, + fe.air_date, fe.air_time, fe.air_timezone, fe.poster_path, fe.poster_thumbhash, (fe.episode_number = 1) AS is_premiere, (fe.episode_number = sf.max_episode_number) AS is_finale @@ -197,7 +198,7 @@ func (r *CalendarRepository) buildFilteredSeasonsCTE(startArg, endArg int, f Cal r.appendPersonalFilterClause("s.series_id", f, &conditions, args, argIdx) return fmt.Sprintf(`SELECT s.content_id, s.series_id, s.season_number, - s.title AS episode_title, s.air_date, mi.title AS title, mi.air_time, + s.title AS episode_title, s.air_date, mi.title AS title, mi.air_time, mi.air_timezone, COALESCE(NULLIF(s.poster_path, ''), mi.poster_path) AS poster_path, COALESCE(NULLIF(s.poster_thumbhash, ''), mi.poster_thumbhash) AS poster_thumbhash FROM seasons s @@ -209,7 +210,7 @@ func (r *CalendarRepository) buildSeasonBranch() string { return `SELECT fs.content_id, 'season_premiere'::text AS type, fs.title, fs.episode_title, fs.series_id, fs.season_number, NULL::int AS episode_number, - fs.air_date, fs.air_time, + fs.air_date, fs.air_time, fs.air_timezone, fs.poster_path, fs.poster_thumbhash, TRUE AS is_premiere, FALSE AS is_finale FROM filtered_seasons fs diff --git a/internal/catalog/detail.go b/internal/catalog/detail.go index 99661385..5f628643 100644 --- a/internal/catalog/detail.go +++ b/internal/catalog/detail.go @@ -94,6 +94,7 @@ type ItemDetail struct { LastAirDate *string `json:"last_air_date,omitempty"` ReleaseDate *string `json:"release_date,omitempty"` AirTime *string `json:"air_time,omitempty"` + AirTimezone *string `json:"air_timezone,omitempty"` ShowStatus string `json:"show_status,omitempty"` // Presigned image URLs. @@ -806,6 +807,7 @@ func (s *DetailService) buildMediaItemDetail(ctx context.Context, item *models.M LastAirDate: item.LastAirDate, ReleaseDate: item.ReleaseDate, AirTime: item.AirTime, + AirTimezone: item.AirTimezone, ShowStatus: item.ShowStatus, PosterThumbhash: item.PosterThumbhash, BackdropThumbhash: item.BackdropThumbhash, @@ -2187,6 +2189,43 @@ func cachedImageVariantPath(path, imageType, size string) string { return path } +// imageTypeFromCachedPath returns the image type segment ("poster", "backdrop", +// "logo", "still") encoded in a cached S3 image path of the form +// ".../{imageType}/{variant}.{ext}". It returns "" for full URLs, +// plugin-prefixed paths, or paths with no directory segment. +func imageTypeFromCachedPath(path string) string { + if path == "" || strings.Contains(path, "://") { + return "" + } + lastSlash := strings.LastIndex(path, "/") + if lastSlash <= 0 { + return "" + } + dir := path[:lastSlash] + return dir[strings.LastIndex(dir, "/")+1:] +} + +// BackdropVariantPath rewrites a cached "/original." image path to the +// requested backdrop variant (e.g. "w1280" or "w1920"). Episode "backdrops" +// are frequently the episode still, which the cache only generates at +// w500/w300 — so requesting a backdrop width 404s. For still/poster/logo +// paths this clamps to that type's largest cached variant instead. Full URLs, +// plugin-prefixed paths, and non-"/original." paths pass through unchanged. +func BackdropVariantPath(path, desiredVariant string) string { + if path == "" || strings.Contains(path, "://") || !strings.Contains(path, "/original.") { + return path + } + variant := desiredVariant + switch imageType := imageTypeFromCachedPath(path); imageType { + case "still", "poster", "logo": + variant = cachedImageVariantKey(imageType, "") + } + if variant == "" { + return path + } + return strings.Replace(path, "/original.", "/"+variant+".", 1) +} + func cachedImageVariantKey(imageType, size string) string { if size == "original" { return "original" diff --git a/internal/catalog/folder_delete_test.go b/internal/catalog/folder_delete_test.go new file mode 100644 index 00000000..280a3e01 --- /dev/null +++ b/internal/catalog/folder_delete_test.go @@ -0,0 +1,138 @@ +package catalog + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgconn" +) + +// withFastDeadlockRetry shrinks retry timing/attempts for tests and restores +// the originals on cleanup. Tests using it must not call t.Parallel(). +func withFastDeadlockRetry(t *testing.T, maxAttempts int) { + t.Helper() + oldMax, oldBackoff := deadlockMaxAttempts, deadlockBaseBackoff + deadlockMaxAttempts = maxAttempts + deadlockBaseBackoff = time.Millisecond + t.Cleanup(func() { + deadlockMaxAttempts = oldMax + deadlockBaseBackoff = oldBackoff + }) +} + +func TestRetryOnDeadlockRetriesThenSucceeds(t *testing.T) { + withFastDeadlockRetry(t, 5) + calls := 0 + err := retryOnDeadlock(context.Background(), func() error { + calls++ + if calls < 3 { + return &pgconn.PgError{Code: "40P01"} + } + return nil + }) + if err != nil { + t.Fatalf("expected success, got %v", err) + } + if calls != 3 { + t.Fatalf("expected 3 calls, got %d", calls) + } +} + +func TestRetryOnDeadlockReturnsNonRetryableImmediately(t *testing.T) { + withFastDeadlockRetry(t, 5) + sentinel := errors.New("boom") + calls := 0 + err := retryOnDeadlock(context.Background(), func() error { + calls++ + return sentinel + }) + if !errors.Is(err, sentinel) { + t.Fatalf("expected sentinel, got %v", err) + } + if calls != 1 { + t.Fatalf("expected 1 call, got %d", calls) + } +} + +func TestRetryOnDeadlockGivesUpAfterMaxAttempts(t *testing.T) { + withFastDeadlockRetry(t, 4) + calls := 0 + err := retryOnDeadlock(context.Background(), func() error { + calls++ + return &pgconn.PgError{Code: "40P01"} + }) + var pgErr *pgconn.PgError + if !errors.As(err, &pgErr) || pgErr.Code != "40P01" { + t.Fatalf("expected 40P01 pg error, got %v", err) + } + if calls != 4 { + t.Fatalf("expected 4 calls, got %d", calls) + } +} + +func TestRetryOnDeadlockStopsOnCanceledContext(t *testing.T) { + withFastDeadlockRetry(t, 5) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + calls := 0 + // op always runs once at the top of the loop before the canceled ctx is + // observed in the select, so exactly one call is expected. + err := retryOnDeadlock(ctx, func() error { + calls++ + return &pgconn.PgError{Code: "40P01"} + }) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got %v", err) + } + if calls != 1 { + t.Fatalf("expected 1 call before cancel, got %d", calls) + } +} + +func TestDeleteInBatchesLoopsUntilUnderBatchSize(t *testing.T) { + withFastDeadlockRetry(t, 5) + counts := []int64{5, 5, 2} + idx := 0 + total, err := deleteInBatches(context.Background(), 5, func(context.Context) (int64, error) { + n := counts[idx] + idx++ + return n, nil + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if total != 12 { + t.Fatalf("expected total 12, got %d", total) + } + if idx != 3 { + t.Fatalf("expected 3 batches, got %d", idx) + } +} + +func TestDeleteInBatchesStopsImmediatelyWhenFirstBatchUnderSize(t *testing.T) { + withFastDeadlockRetry(t, 5) + calls := 0 + total, err := deleteInBatches(context.Background(), 5, func(context.Context) (int64, error) { + calls++ + return 0, nil + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if total != 0 || calls != 1 { + t.Fatalf("expected total 0 and 1 call, got total=%d calls=%d", total, calls) + } +} + +func TestDeleteInBatchesReturnsError(t *testing.T) { + withFastDeadlockRetry(t, 5) + sentinel := errors.New("delete failed") + _, err := deleteInBatches(context.Background(), 5, func(context.Context) (int64, error) { + return 0, sentinel + }) + if !errors.Is(err, sentinel) { + t.Fatalf("expected sentinel, got %v", err) + } +} diff --git a/internal/catalog/folder_repo.go b/internal/catalog/folder_repo.go index 1a40bca7..b1485308 100644 --- a/internal/catalog/folder_repo.go +++ b/internal/catalog/folder_repo.go @@ -14,6 +14,74 @@ import ( "github.com/Silo-Server/silo-server/internal/models" ) +// Retry parameters for transient serialization/deadlock failures. They are +// package vars (not consts) only so tests can shrink them; production code +// never mutates them. +var ( + deadlockMaxAttempts = 5 + deadlockBaseBackoff = 50 * time.Millisecond +) + +const ( + // orphanDeleteBatch is small because each media_items row cascades across + // ~15 child tables. + orphanDeleteBatch = 1000 + // folderChildDeleteBatch covers the lighter folder-scoped media_files and + // junction deletes. + folderChildDeleteBatch = 5000 +) + +// retryOnDeadlock runs op, retrying when Postgres reports a deadlock (40P01) or +// serialization failure (40001), with exponential backoff. It returns +// immediately for any other error, and honors context cancellation between +// attempts. +func retryOnDeadlock(ctx context.Context, op func() error) error { + backoff := deadlockBaseBackoff + for attempt := 1; ; attempt++ { + err := op() + if err == nil { + return nil + } + var pgErr *pgconn.PgError + if attempt < deadlockMaxAttempts && errors.As(err, &pgErr) && + (pgErr.Code == "40P01" || pgErr.Code == "40001") { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(backoff): + } + backoff *= 2 + continue + } + return err + } +} + +// deleteInBatches repeatedly runs deleteBatch (each a single autocommit +// statement) until a batch removes fewer than batchSize rows. Each batch is +// retried on deadlock. It returns the total number of rows deleted. +func deleteInBatches( + ctx context.Context, + batchSize int, + deleteBatch func(ctx context.Context) (int64, error), +) (int64, error) { + var total int64 + for { + var affected int64 + if err := retryOnDeadlock(ctx, func() error { + n, e := deleteBatch(ctx) + affected = n + return e + }); err != nil { + return total, err + } + total += affected + if affected < int64(batchSize) { + return total, nil + } + } +} + // Sentinel errors for folder repository operations. var ( ErrFolderNotFound = errors.New("folder not found") @@ -417,97 +485,187 @@ func (r *FolderRepository) DeleteWithStats( id int, progress func(current, total int, message string), ) (*DeleteFolderStats, error) { - tx, err := r.pool.Begin(ctx) - if err != nil { - return nil, fmt.Errorf("beginning delete transaction: %w", err) - } - defer func() { _ = tx.Rollback(ctx) }() - stats := &DeleteFolderStats{} - if err := tx.QueryRow(ctx, `SELECT name FROM media_folders WHERE id = $1`, id).Scan(&stats.LibraryName); err != nil { + + // Phase 0: preflight reads (no long-lived transaction). + if err := r.pool.QueryRow(ctx, `SELECT name FROM media_folders WHERE id = $1`, id).Scan(&stats.LibraryName); err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, ErrFolderNotFound } return nil, fmt.Errorf("loading folder before delete: %w", err) } - - if err := tx.QueryRow(ctx, `SELECT COUNT(*) FROM media_files WHERE media_folder_id = $1`, id).Scan(&stats.MediaFiles); err != nil { + if err := r.pool.QueryRow(ctx, `SELECT COUNT(*) FROM media_files WHERE media_folder_id = $1`, id).Scan(&stats.MediaFiles); err != nil { return nil, fmt.Errorf("counting media files: %w", err) } - if err := tx.QueryRow(ctx, `SELECT COUNT(*) FROM media_item_libraries WHERE media_folder_id = $1`, id).Scan(&stats.MediaItemLinks); err != nil { + if err := r.pool.QueryRow(ctx, `SELECT COUNT(*) FROM media_item_libraries WHERE media_folder_id = $1`, id).Scan(&stats.MediaItemLinks); err != nil { return nil, fmt.Errorf("counting media item links: %w", err) } + var orphanTotal int + if err := r.pool.QueryRow(ctx, ` + SELECT COUNT(*) + FROM media_item_libraries mil + WHERE mil.media_folder_id = $1 + AND NOT EXISTS ( + SELECT 1 FROM media_item_libraries other + WHERE other.content_id = mil.content_id + AND other.media_folder_id <> $1 + )`, id).Scan(&orphanTotal); err != nil { + return nil, fmt.Errorf("counting orphaned items: %w", err) + } - // Collect items that exist ONLY in this library before cascade removes the - // junction rows, so we know which media_items to clean up. - rows, err := tx.Query(ctx, ` + // Phase 1: delete orphaned media_items in detect-then-delete batches. + // Cascade removes their junctions, provider IDs, episodes, seasons, etc. + if progress != nil { + progress(0, orphanTotal, "Deleting orphaned items") + } + rawDirs := make(map[string]struct{}) + for { + ids, err := r.collectOrphanBatch(ctx, id, orphanDeleteBatch) + if err != nil { + return nil, err + } + if len(ids) == 0 { + break + } + dirs, err := collectRawImageDirs(ctx, r.pool, ids) + if err != nil { + return nil, err + } + for _, d := range dirs { + rawDirs[d] = struct{}{} + } + var deleted int64 + if err := retryOnDeadlock(ctx, func() error { + // Re-check the orphan invariant inside the delete. A concurrent + // scan/import may have attached one of these content IDs to another + // library after collectOrphanBatch returned; without this guard the + // cascade would delete the shared media_items row (and the + // newly-added membership), dropping the item from the other library. + tag, e := r.pool.Exec(ctx, ` + DELETE FROM media_items + WHERE content_id = ANY($1) + AND NOT EXISTS ( + SELECT 1 FROM media_item_libraries other + WHERE other.content_id = media_items.content_id + AND other.media_folder_id <> $2 + )`, ids, id) + if e != nil { + return e + } + deleted = tag.RowsAffected() + return nil + }); err != nil { + return nil, fmt.Errorf("deleting orphaned items: %w", err) + } + stats.OrphanedItems += int(deleted) + if progress != nil { + progress(stats.OrphanedItems, orphanTotal, "Deleting orphaned items") + } + } + + // Filter accumulated image dirs once, now that orphans are gone, against + // any surviving content. Empty deleting-set means "exclude nothing". + if len(rawDirs) > 0 { + filtered, err := filterUnreferencedImageDirs(ctx, r.pool, dirSetToSlice(rawDirs), []string{}) + if err != nil { + return nil, err + } + stats.OrphanedImageDirs = filtered + } + + // Phase 2: delete this folder's media_files (folder-tied, not item-tied). + if progress != nil { + progress(orphanTotal, orphanTotal, "Deleting media files") + } + if _, err := deleteInBatches(ctx, folderChildDeleteBatch, func(ctx context.Context) (int64, error) { + tag, e := r.pool.Exec(ctx, ` + DELETE FROM media_files + WHERE id IN ( + SELECT id FROM media_files WHERE media_folder_id = $1 LIMIT $2 + )`, id, folderChildDeleteBatch) + if e != nil { + return 0, e + } + return tag.RowsAffected(), nil + }); err != nil { + return nil, fmt.Errorf("deleting media files: %w", err) + } + + // Phase 3: delete remaining folder memberships (shared items kept; only the + // membership in this folder is removed). + if progress != nil { + progress(orphanTotal, orphanTotal, "Removing library memberships") + } + if _, err := deleteInBatches(ctx, folderChildDeleteBatch, func(ctx context.Context) (int64, error) { + tag, e := r.pool.Exec(ctx, ` + DELETE FROM media_item_libraries + WHERE ctid IN ( + SELECT ctid FROM media_item_libraries WHERE media_folder_id = $1 LIMIT $2 + )`, id, folderChildDeleteBatch) + if e != nil { + return 0, e + } + return tag.RowsAffected(), nil + }); err != nil { + return nil, fmt.Errorf("deleting media item links: %w", err) + } + + // Phase 4: delete the now-lightweight folder row. Tolerate 0 rows so a + // resumed run that already removed it still succeeds. + if err := retryOnDeadlock(ctx, func() error { + _, e := r.pool.Exec(ctx, `DELETE FROM media_folders WHERE id = $1`, id) + return e + }); err != nil { + return nil, fmt.Errorf("deleting folder: %w", err) + } + + if progress != nil { + progress(orphanTotal, orphanTotal, "Library deletion completed") + } + return stats, nil +} + +// collectOrphanBatch returns up to limit content IDs whose only library +// membership is the given folder. +func (r *FolderRepository) collectOrphanBatch(ctx context.Context, folderID, limit int) ([]string, error) { + rows, err := r.pool.Query(ctx, ` SELECT mil.content_id FROM media_item_libraries mil WHERE mil.media_folder_id = $1 AND NOT EXISTS ( SELECT 1 FROM media_item_libraries other WHERE other.content_id = mil.content_id - AND other.media_folder_id != $1 - )`, id) + AND other.media_folder_id <> $1 + ) + LIMIT $2`, folderID, limit) if err != nil { return nil, fmt.Errorf("finding orphaned items: %w", err) } - - var orphanIDs []string + defer rows.Close() + var ids []string for rows.Next() { var contentID string if err := rows.Scan(&contentID); err != nil { - rows.Close() return nil, fmt.Errorf("scanning orphan content_id: %w", err) } - orphanIDs = append(orphanIDs, contentID) + ids = append(ids, contentID) } - rows.Close() if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterating orphan rows: %w", err) } - stats.OrphanedItems = len(orphanIDs) + return ids, nil +} - // Collect S3 image paths from orphaned items (and their seasons/episodes) - // before deletion so the caller can clean up S3 objects afterward. - if len(orphanIDs) > 0 { - stats.OrphanedImageDirs, err = collectImageDirs(ctx, tx, orphanIDs) - if err != nil { - return nil, err - } +// dirSetToSlice returns the keys of set as a slice (nil if empty). +func dirSetToSlice(set map[string]struct{}) []string { + if len(set) == 0 { + return nil } - - // Delete the folder; CASCADE removes junction rows, media_files, and - // library_provider_chains automatically. - if progress != nil { - progress(1, 3, "Deleting library folder data") + dirs := make([]string, 0, len(set)) + for d := range set { + dirs = append(dirs, d) } - tag, err := tx.Exec(ctx, "DELETE FROM media_folders WHERE id = $1", id) - if err != nil { - return nil, fmt.Errorf("deleting folder: %w", err) - } - if tag.RowsAffected() == 0 { - return nil, ErrFolderNotFound - } - - // Remove orphaned media items (episodes CASCADE from media_items). - if len(orphanIDs) > 0 { - if progress != nil { - progress(2, 3, "Deleting orphaned items") - } - _, err = tx.Exec(ctx, "DELETE FROM media_items WHERE content_id = ANY($1)", orphanIDs) - if err != nil { - return nil, fmt.Errorf("deleting orphaned items: %w", err) - } - } - - if err := tx.Commit(ctx); err != nil { - return nil, fmt.Errorf("committing delete transaction: %w", err) - } - if progress != nil { - progress(3, 3, "Library deletion completed") - } - return stats, nil + return dirs } // pathDir extracts the directory portion of an S3 image path. @@ -520,6 +678,12 @@ func pathDir(path string) string { return path[:idx+1] } +// rowQuerier is satisfied by both *pgxpool.Pool and pgx.Tx, letting read +// helpers run inside or outside an explicit transaction. +type rowQuerier interface { + Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) +} + // collectOrphanIDs returns content IDs from the given set that have no // remaining library memberships. Must be called within a transaction. func collectOrphanIDs(ctx context.Context, tx pgx.Tx, contentIDs []string) ([]string, error) { @@ -544,10 +708,21 @@ func collectOrphanIDs(ctx context.Context, tx pgx.Tx, contentIDs []string) ([]st return ids, rows.Err() } -// collectImageDirs returns S3 directory prefixes for all images belonging to -// the given content IDs (items, seasons, and episodes). -func collectImageDirs(ctx context.Context, tx pgx.Tx, contentIDs []string) ([]string, error) { - imgRows, err := tx.Query(ctx, ` +// collectImageDirs returns S3 directory prefixes for images belonging to the +// given content IDs that are not still referenced by other surviving content. +func collectImageDirs(ctx context.Context, q rowQuerier, contentIDs []string) ([]string, error) { + dirs, err := collectRawImageDirs(ctx, q, contentIDs) + if err != nil { + return nil, err + } + return filterUnreferencedImageDirs(ctx, q, dirs, contentIDs) +} + +// collectRawImageDirs returns the deduped S3 directory prefixes referenced by +// the given content IDs (items, their seasons, and their episodes), without +// filtering out dirs still used by other content. +func collectRawImageDirs(ctx context.Context, q rowQuerier, contentIDs []string) ([]string, error) { + imgRows, err := q.Query(ctx, ` SELECT poster_path, backdrop_path, logo_path FROM media_items WHERE content_id = ANY($1) UNION ALL SELECT poster_path, '', '' FROM seasons WHERE series_id = ANY($1) @@ -575,20 +750,19 @@ func collectImageDirs(ctx context.Context, tx pgx.Tx, contentIDs []string) ([]st if err := imgRows.Err(); err != nil { return nil, fmt.Errorf("iterating image paths: %w", err) } - dirs := make([]string, 0, len(dirSet)) for dir := range dirSet { dirs = append(dirs, dir) } - return filterUnreferencedImageDirs(ctx, tx, dirs, contentIDs) + return dirs, nil } -func filterUnreferencedImageDirs(ctx context.Context, tx pgx.Tx, dirs, deletingContentIDs []string) ([]string, error) { +func filterUnreferencedImageDirs(ctx context.Context, q rowQuerier, dirs, deletingContentIDs []string) ([]string, error) { if len(dirs) == 0 { return nil, nil } - rows, err := tx.Query(ctx, ` + rows, err := q.Query(ctx, ` SELECT candidate.dir FROM unnest($1::text[]) AS candidate(dir) WHERE NOT EXISTS ( diff --git a/internal/catalog/image_variants_test.go b/internal/catalog/image_variants_test.go new file mode 100644 index 00000000..300ae0af --- /dev/null +++ b/internal/catalog/image_variants_test.go @@ -0,0 +1,85 @@ +package catalog + +import "testing" + +func TestImageTypeFromCachedPath(t *testing.T) { + tests := []struct { + name string + path string + want string + }{ + {"episode still", "tvdb/series/73141/seasons/22/episodes/9/still/original.webp", "still"}, + {"movie backdrop", "tmdb/movies/550/backdrop/original.webp", "backdrop"}, + {"leading slash poster", "/tmdb/movies/550/poster/original.jpg", "poster"}, + {"logo", "tmdb/movies/550/logo/original.png", "logo"}, + {"http url", "https://images.example.com/backdrop/original.jpg", ""}, + {"plugin path", "plugin://tmdb/backdrop/original.jpg", ""}, + {"no slash", "original.webp", ""}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := imageTypeFromCachedPath(tt.path); got != tt.want { + t.Fatalf("imageTypeFromCachedPath(%q) = %q, want %q", tt.path, got, tt.want) + } + }) + } +} + +func TestBackdropVariantPath(t *testing.T) { + tests := []struct { + name string + path string + desired string + want string + }{ + { + name: "real backdrop keeps requested w1280", + path: "tmdb/movies/550/backdrop/original.webp", + desired: "w1280", + want: "tmdb/movies/550/backdrop/w1280.webp", + }, + { + name: "real backdrop keeps requested w1920", + path: "/tmdb/shows/1399/backdrop/original.jpg", + desired: "w1920", + want: "/tmdb/shows/1399/backdrop/w1920.jpg", + }, + { + name: "episode still clamps to w500 (no large variant exists)", + path: "tvdb/series/73141/seasons/22/episodes/9/still/original.webp", + desired: "w1280", + want: "tvdb/series/73141/seasons/22/episodes/9/still/w500.webp", + }, + { + name: "episode still clamps w1920 to w500", + path: "tvdb/series/73141/seasons/22/episodes/9/still/original.webp", + desired: "w1920", + want: "tvdb/series/73141/seasons/22/episodes/9/still/w500.webp", + }, + { + name: "http url passes through", + path: "https://images.example.com/backdrop/original.jpg", + desired: "w1280", + want: "https://images.example.com/backdrop/original.jpg", + }, + { + name: "plugin path passes through", + path: "plugin://tmdb/backdrop/original.jpg", + desired: "w1280", + want: "plugin://tmdb/backdrop/original.jpg", + }, + { + name: "path without original segment passes through", + path: "tmdb/movies/550/backdrop/w300.webp", + desired: "w1280", + want: "tmdb/movies/550/backdrop/w300.webp", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := BackdropVariantPath(tt.path, tt.desired); got != tt.want { + t.Fatalf("BackdropVariantPath(%q, %q) = %q, want %q", tt.path, tt.desired, got, tt.want) + } + }) + } +} diff --git a/internal/catalog/item_repo.go b/internal/catalog/item_repo.go index 9788a91c..011b72db 100644 --- a/internal/catalog/item_repo.go +++ b/internal/catalog/item_repo.go @@ -52,7 +52,7 @@ const itemColumns = `content_id, type, title, sort_title, default_metadata_langu imdb_id, tmdb_id, tvdb_id, poster_path, poster_thumbhash, backdrop_path, backdrop_thumbhash, logo_path, metadata_s3_path, metadata_etag, season_count, - studios, networks, countries, keywords, original_language, release_date::text, first_air_date, last_air_date, air_time, + studios, networks, countries, keywords, original_language, release_date::text, first_air_date, last_air_date, air_time, air_timezone, show_status, matched_at, last_refreshed, refresh_failures, episode_metadata_incomplete, episode_metadata_last_checked_at, locked_fields, status, created_at, updated_at` @@ -65,7 +65,7 @@ func qualifiedItemColumns(alias string) string { "imdb_id", "tmdb_id", "tvdb_id", "poster_path", "poster_thumbhash", "backdrop_path", "backdrop_thumbhash", "logo_path", "metadata_s3_path", "metadata_etag", "season_count", - "studios", "networks", "countries", "keywords", "original_language", "release_date::text", "first_air_date", "last_air_date", "air_time", + "studios", "networks", "countries", "keywords", "original_language", "release_date::text", "first_air_date", "last_air_date", "air_time", "air_timezone", "show_status", "matched_at", "last_refreshed", "refresh_failures", "episode_metadata_incomplete", "episode_metadata_last_checked_at", "locked_fields", "status", "created_at", "updated_at", @@ -85,7 +85,7 @@ func qualifiedListItemColumns(alias string) string { "imdb_id", "tmdb_id", "tvdb_id", "poster_path", "poster_thumbhash", "backdrop_path", "backdrop_thumbhash", "logo_path", "metadata_s3_path", "metadata_etag", "season_count", - "studios", "networks", "countries", "keywords", "original_language", "release_date::text", "first_air_date", "last_air_date", "air_time", + "studios", "networks", "countries", "keywords", "original_language", "release_date::text", "first_air_date", "last_air_date", "air_time", "air_timezone", "show_status", "matched_at", "last_refreshed", "refresh_failures", "episode_metadata_incomplete", "episode_metadata_last_checked_at", "locked_fields", "status", "created_at", "updated_at", @@ -141,6 +141,7 @@ func scanItem(row pgx.Row) (*models.MediaItem, error) { &item.FirstAirDate, &item.LastAirDate, &item.AirTime, + &item.AirTimezone, &item.ShowStatus, &item.MatchedAt, &item.LastRefreshed, @@ -203,6 +204,7 @@ func scanItems(rows pgx.Rows) ([]*models.MediaItem, error) { &item.FirstAirDate, &item.LastAirDate, &item.AirTime, + &item.AirTimezone, &item.ShowStatus, &item.MatchedAt, &item.LastRefreshed, @@ -275,6 +277,7 @@ func scanItemsWithTotal(rows pgx.Rows) ([]*models.MediaItem, int, error) { &item.FirstAirDate, &item.LastAirDate, &item.AirTime, + &item.AirTimezone, &item.ShowStatus, &item.MatchedAt, &item.LastRefreshed, @@ -326,7 +329,7 @@ func (r *ItemRepository) upsert(ctx context.Context, execer itemExecer, item *mo imdb_id, tmdb_id, tvdb_id, poster_path, poster_thumbhash, backdrop_path, backdrop_thumbhash, logo_path, metadata_s3_path, metadata_etag, season_count, - studios, networks, countries, keywords, original_language, release_date, first_air_date, last_air_date, air_time, + studios, networks, countries, keywords, original_language, release_date, first_air_date, last_air_date, air_time, air_timezone, show_status, matched_at, last_refreshed, refresh_failures, episode_metadata_incomplete, episode_metadata_last_checked_at, status @@ -337,10 +340,10 @@ func (r *ItemRepository) upsert(ctx context.Context, execer itemExecer, item *mo $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, - $28, $29, $30, $31, $32, $33, $34, $35, $36, - $37, - $38, $39, $40, - $41, $42, $43 + $28, $29, $30, $31, $32, $33, $34, $35, $36, $37, + $38, + $39, $40, $41, + $42, $43, $44 ) ON CONFLICT (content_id) DO UPDATE SET type = EXCLUDED.type, @@ -378,6 +381,7 @@ func (r *ItemRepository) upsert(ctx context.Context, execer itemExecer, item *mo first_air_date = EXCLUDED.first_air_date, last_air_date = EXCLUDED.last_air_date, air_time = EXCLUDED.air_time, + air_timezone = EXCLUDED.air_timezone, show_status = EXCLUDED.show_status, matched_at = EXCLUDED.matched_at, last_refreshed = EXCLUDED.last_refreshed, @@ -424,6 +428,7 @@ func (r *ItemRepository) upsert(ctx context.Context, execer itemExecer, item *mo item.FirstAirDate, item.LastAirDate, item.AirTime, + item.AirTimezone, item.ShowStatus, item.MatchedAt, item.LastRefreshed, @@ -1236,6 +1241,15 @@ func (r *ItemRepository) UpdateMetadata(ctx context.Context, contentID string, u argIdx++ } } + // addNullableString behaves like addString but stores an empty string as SQL + // NULL (via NULLIF), so clearing a nullable column persists as NULL. + addNullableString := func(col string, val *string) { + if val != nil { + setClauses = append(setClauses, fmt.Sprintf("%s = NULLIF($%d, '')", col, argIdx)) + args = append(args, *val) + argIdx++ + } + } addInt := func(col string, val *int) { if val != nil { setClauses = append(setClauses, fmt.Sprintf("%s = $%d", col, argIdx)) @@ -1285,6 +1299,7 @@ func (r *ItemRepository) UpdateMetadata(ctx context.Context, contentID string, u addString("first_air_date", upd.FirstAirDate) addString("last_air_date", upd.LastAirDate) addString("air_time", upd.AirTime) + addNullableString("air_timezone", upd.AirTimezone) addString("status", upd.Status) addString("show_status", upd.ShowStatus) addString("imdb_id", upd.ImdbID) diff --git a/internal/catalog/update.go b/internal/catalog/update.go index 46cea8d6..b0057327 100644 --- a/internal/catalog/update.go +++ b/internal/catalog/update.go @@ -28,6 +28,7 @@ type MetadataUpdate struct { FirstAirDate *string LastAirDate *string AirTime *string + AirTimezone *string AirDate *string Status *string ShowStatus *string diff --git a/internal/catalogseed/export_stream.go b/internal/catalogseed/export_stream.go index f02734b3..7a4a9e88 100644 --- a/internal/catalogseed/export_stream.go +++ b/internal/catalogseed/export_stream.go @@ -372,6 +372,8 @@ func (s *Service) streamItemRecords(ctx context.Context, folderIDs []int, fn fun mi.release_date::text, mi.first_air_date, mi.last_air_date, + mi.air_time, + mi.air_timezone, mi.matched_at, mi.last_refreshed, COALESCE(mi.refresh_failures, 0), @@ -426,6 +428,8 @@ func (s *Service) streamItemRecords(ctx context.Context, folderIDs []int, fn fun &record.ReleaseDate, &record.FirstAirDate, &record.LastAirDate, + &record.AirTime, + &record.AirTimezone, &record.MatchedAt, &record.LastRefreshed, &record.RefreshFailures, diff --git a/internal/catalogseed/service.go b/internal/catalogseed/service.go index 0206895f..e3838dad 100644 --- a/internal/catalogseed/service.go +++ b/internal/catalogseed/service.go @@ -166,7 +166,7 @@ func (s *Service) ImportWithProgress(ctx context.Context, data []byte, opts Impo item.ImdbID, item.TmdbID, item.TvdbID, item.PosterPath, item.PosterThumbhash, item.BackdropPath, item.BackdropThumbhash, item.LogoPath, item.MetadataS3Path, item.MetadataEtag, item.SeasonCount, - studios, networks, countries, keywords, item.OriginalLanguage, item.ReleaseDate, item.FirstAirDate, item.LastAirDate, + studios, networks, countries, keywords, item.OriginalLanguage, item.ReleaseDate, item.FirstAirDate, item.LastAirDate, item.AirTime, item.AirTimezone, item.MatchedAt, item.LastRefreshed, item.RefreshFailures, item.LockedFields, item.Status, item.CreatedAt, item.UpdatedAt, }) @@ -179,7 +179,7 @@ func (s *Service) ImportWithProgress(ctx context.Context, data []byte, opts Impo "imdb_id", "tmdb_id", "tvdb_id", "poster_path", "poster_thumbhash", "backdrop_path", "backdrop_thumbhash", "logo_path", "metadata_s3_path", "metadata_etag", "season_count", - "studios", "networks", "countries", "keywords", "original_language", "release_date", "first_air_date", "last_air_date", + "studios", "networks", "countries", "keywords", "original_language", "release_date", "first_air_date", "last_air_date", "air_time", "air_timezone", "matched_at", "last_refreshed", "refresh_failures", "locked_fields", "status", "created_at", "updated_at", }, @@ -1044,7 +1044,7 @@ func bulkInsertItems(ctx context.Context, tx pgx.Tx, items []ItemRecord, onBatch item.ImdbID, item.TmdbID, item.TvdbID, item.PosterPath, item.PosterThumbhash, item.BackdropPath, item.BackdropThumbhash, item.LogoPath, item.MetadataS3Path, item.MetadataEtag, item.SeasonCount, - item.Studios, item.Networks, item.Countries, item.Keywords, item.OriginalLanguage, item.ReleaseDate, item.FirstAirDate, item.LastAirDate, + item.Studios, item.Networks, item.Countries, item.Keywords, item.OriginalLanguage, item.ReleaseDate, item.FirstAirDate, item.LastAirDate, item.AirTime, item.AirTimezone, item.MatchedAt, item.LastRefreshed, item.RefreshFailures, item.LockedFields, item.Status, item.CreatedAt, item.UpdatedAt, }) @@ -1058,12 +1058,12 @@ func bulkInsertItems(ctx context.Context, tx pgx.Tx, items []ItemRecord, onBatch imdb_id, tmdb_id, tvdb_id, poster_path, poster_thumbhash, backdrop_path, backdrop_thumbhash, logo_path, metadata_s3_path, metadata_etag, season_count, - studios, networks, countries, keywords, original_language, release_date, first_air_date, last_air_date, + studios, networks, countries, keywords, original_language, release_date, first_air_date, last_air_date, air_time, air_timezone, matched_at, last_refreshed, refresh_failures, locked_fields, status, created_at, updated_at ) VALUES `, rows, - 41, + 43, nil, "", onBatch, @@ -1904,13 +1904,13 @@ func batchImportItems(ctx context.Context, tx pgx.Tx, items []ItemRecord, mode C item.ImdbID, item.TmdbID, item.TvdbID, item.PosterPath, item.PosterThumbhash, item.BackdropPath, item.BackdropThumbhash, item.LogoPath, item.MetadataS3Path, item.MetadataEtag, item.SeasonCount, - studios, networks, countries, keywords, item.OriginalLanguage, item.ReleaseDate, item.FirstAirDate, item.LastAirDate, + studios, networks, countries, keywords, item.OriginalLanguage, item.ReleaseDate, item.FirstAirDate, item.LastAirDate, item.AirTime, item.AirTimezone, item.MatchedAt, item.LastRefreshed, item.RefreshFailures, item.LockedFields, item.Status, item.CreatedAt, item.UpdatedAt, }) } - const colCount = 41 + const colCount = 43 prefix := ` INSERT INTO media_items ( content_id, type, title, sort_title, original_title, year, genres, @@ -1919,7 +1919,7 @@ func batchImportItems(ctx context.Context, tx pgx.Tx, items []ItemRecord, mode C imdb_id, tmdb_id, tvdb_id, poster_path, poster_thumbhash, backdrop_path, backdrop_thumbhash, logo_path, metadata_s3_path, metadata_etag, season_count, - studios, networks, countries, keywords, original_language, release_date, first_air_date, last_air_date, + studios, networks, countries, keywords, original_language, release_date, first_air_date, last_air_date, air_time, air_timezone, matched_at, last_refreshed, refresh_failures, locked_fields, status, created_at, updated_at ) VALUES ` @@ -1963,6 +1963,8 @@ func batchImportItems(ctx context.Context, tx pgx.Tx, items []ItemRecord, mode C release_date = EXCLUDED.release_date, first_air_date = EXCLUDED.first_air_date, last_air_date = EXCLUDED.last_air_date, + air_time = EXCLUDED.air_time, + air_timezone = EXCLUDED.air_timezone, matched_at = EXCLUDED.matched_at, last_refreshed = EXCLUDED.last_refreshed, refresh_failures = EXCLUDED.refresh_failures, @@ -2055,7 +2057,7 @@ func importItem(ctx context.Context, tx pgx.Tx, item ItemRecord, mode ConflictMo imdb_id, tmdb_id, tvdb_id, poster_path, poster_thumbhash, backdrop_path, backdrop_thumbhash, logo_path, metadata_s3_path, metadata_etag, season_count, - studios, networks, countries, keywords, original_language, release_date, first_air_date, last_air_date, + studios, networks, countries, keywords, original_language, release_date, first_air_date, last_air_date, air_time, air_timezone, matched_at, last_refreshed, refresh_failures, locked_fields, status, created_at, updated_at ) VALUES ( @@ -2065,9 +2067,9 @@ func importItem(ctx context.Context, tx pgx.Tx, item ItemRecord, mode ConflictMo $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, - $27, $28, $29, $30, $31, $32, $33, $34, - $35, $36, $37, $38, $39, - $40, $41 + $27, $28, $29, $30, $31, $32, $33, $34, $35, $36, + $37, $38, $39, $40, $41, + $42, $43 ) ON CONFLICT (content_id) DO NOTHING`, item.ContentID, item.Type, item.Title, item.SortTitle, item.OriginalTitle, item.Year, item.Genres, @@ -2076,7 +2078,7 @@ func importItem(ctx context.Context, tx pgx.Tx, item ItemRecord, mode ConflictMo item.ImdbID, item.TmdbID, item.TvdbID, item.PosterPath, item.PosterThumbhash, item.BackdropPath, item.BackdropThumbhash, item.LogoPath, item.MetadataS3Path, item.MetadataEtag, item.SeasonCount, - item.Studios, item.Networks, item.Countries, item.Keywords, item.OriginalLanguage, item.ReleaseDate, item.FirstAirDate, item.LastAirDate, + item.Studios, item.Networks, item.Countries, item.Keywords, item.OriginalLanguage, item.ReleaseDate, item.FirstAirDate, item.LastAirDate, item.AirTime, item.AirTimezone, item.MatchedAt, item.LastRefreshed, item.RefreshFailures, item.LockedFields, item.Status, item.CreatedAt, item.UpdatedAt, ) @@ -2094,7 +2096,7 @@ func importItem(ctx context.Context, tx pgx.Tx, item ItemRecord, mode ConflictMo imdb_id, tmdb_id, tvdb_id, poster_path, poster_thumbhash, backdrop_path, backdrop_thumbhash, logo_path, metadata_s3_path, metadata_etag, season_count, - studios, networks, countries, keywords, original_language, release_date, first_air_date, last_air_date, + studios, networks, countries, keywords, original_language, release_date, first_air_date, last_air_date, air_time, air_timezone, matched_at, last_refreshed, refresh_failures, locked_fields, status, created_at, updated_at ) VALUES ( @@ -2104,9 +2106,9 @@ func importItem(ctx context.Context, tx pgx.Tx, item ItemRecord, mode ConflictMo $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, - $27, $28, $29, $30, $31, $32, $33, $34, - $35, $36, $37, $38, $39, - $40, $41 + $27, $28, $29, $30, $31, $32, $33, $34, $35, $36, + $37, $38, $39, $40, $41, + $42, $43 ) ON CONFLICT (content_id) DO UPDATE SET type = EXCLUDED.type, @@ -2142,6 +2144,8 @@ func importItem(ctx context.Context, tx pgx.Tx, item ItemRecord, mode ConflictMo release_date = EXCLUDED.release_date, first_air_date = EXCLUDED.first_air_date, last_air_date = EXCLUDED.last_air_date, + air_time = EXCLUDED.air_time, + air_timezone = EXCLUDED.air_timezone, matched_at = EXCLUDED.matched_at, last_refreshed = EXCLUDED.last_refreshed, refresh_failures = EXCLUDED.refresh_failures, @@ -2155,7 +2159,7 @@ func importItem(ctx context.Context, tx pgx.Tx, item ItemRecord, mode ConflictMo item.ImdbID, item.TmdbID, item.TvdbID, item.PosterPath, item.PosterThumbhash, item.BackdropPath, item.BackdropThumbhash, item.LogoPath, item.MetadataS3Path, item.MetadataEtag, item.SeasonCount, - item.Studios, item.Networks, item.Countries, item.Keywords, item.OriginalLanguage, item.ReleaseDate, item.FirstAirDate, item.LastAirDate, + item.Studios, item.Networks, item.Countries, item.Keywords, item.OriginalLanguage, item.ReleaseDate, item.FirstAirDate, item.LastAirDate, item.AirTime, item.AirTimezone, item.MatchedAt, item.LastRefreshed, item.RefreshFailures, item.LockedFields, item.Status, item.CreatedAt, item.UpdatedAt, ).Scan(&created) diff --git a/internal/catalogseed/types.go b/internal/catalogseed/types.go index d44b2a96..ff899a56 100644 --- a/internal/catalogseed/types.go +++ b/internal/catalogseed/types.go @@ -70,6 +70,8 @@ type ItemRecord struct { ReleaseDate *string `json:"release_date,omitempty"` FirstAirDate *string `json:"first_air_date,omitempty"` LastAirDate *string `json:"last_air_date,omitempty"` + AirTime *string `json:"air_time,omitempty"` + AirTimezone *string `json:"air_timezone,omitempty"` MatchedAt *time.Time `json:"matched_at,omitempty"` LastRefreshed *time.Time `json:"last_refreshed,omitempty"` RefreshFailures int `json:"refresh_failures"` diff --git a/internal/metadata/air_time_test.go b/internal/metadata/air_time_test.go index 89332a54..5348f0d8 100644 --- a/internal/metadata/air_time_test.go +++ b/internal/metadata/air_time_test.go @@ -49,18 +49,40 @@ func TestMetadataResultToItem_CarriesAirTime(t *testing.T) { } } +func TestMetadataResultToItem_InfersAirTimezone(t *testing.T) { + result := &MetadataResult{ + HasMetadata: true, + Title: "Series", + Networks: []string{"BBC One"}, + } + setMetadataAirTime(t, result, "20:00") + + item := metadataResultToItem(result, "series") + if item.AirTimezone == nil { + t.Fatal("expected item air_timezone to be inferred") + } + if got := *item.AirTimezone; got != "Europe/London" { + t.Fatalf("expected inferred air_timezone Europe/London, got %q", got) + } +} + func TestItemToMetadataResult_CarriesAirTime(t *testing.T) { airTime := "20:00" + airTimezone := "America/New_York" result := itemToMetadataResult(&models.MediaItem{ - ContentID: "series-1", - Type: "series", - Title: "Series", - AirTime: &airTime, + ContentID: "series-1", + Type: "series", + Title: "Series", + AirTime: &airTime, + AirTimezone: &airTimezone, }) if got := getMetadataAirTime(t, result); got != airTime { t.Fatalf("expected metadata air_time %q, got %q", airTime, got) } + if result.AirTimezone != airTimezone { + t.Fatalf("expected metadata air_timezone %q, got %q", airTimezone, result.AirTimezone) + } } func TestMergeMetadata_CarriesAirTime(t *testing.T) { @@ -75,6 +97,20 @@ func TestMergeMetadata_CarriesAirTime(t *testing.T) { } } +func TestMergeMetadata_RespectsAirScheduleLock(t *testing.T) { + source := &MetadataResult{AirTime: "20:00", AirTimezone: "Europe/London"} + target := &MetadataResult{AirTime: "21:00", AirTimezone: "America/New_York"} + + MergeMetadata(source, target, []MetadataField{FieldAirSchedule}, MergeReplaceUnlocked) + + if target.AirTime != "21:00" { + t.Fatalf("expected locked air_time to remain 21:00, got %q", target.AirTime) + } + if target.AirTimezone != "America/New_York" { + t.Fatalf("expected locked air_timezone to remain America/New_York, got %q", target.AirTimezone) + } +} + func TestMergeGlobalMetadata_CarriesAirTime(t *testing.T) { source := &MetadataResult{} target := &MetadataResult{} diff --git a/internal/metadata/merge.go b/internal/metadata/merge.go index f33c0f0b..fbeb65bc 100644 --- a/internal/metadata/merge.go +++ b/internal/metadata/merge.go @@ -49,7 +49,10 @@ func MergeMetadata(source, target *MetadataResult, locked []MetadataField, mode mergeInt(&target.SeasonCount, source.SeasonCount, mode) mergeScalar(&target.FirstAirDate, source.FirstAirDate, mode) mergeScalar(&target.LastAirDate, source.LastAirDate, mode) - mergeScalar(&target.AirTime, source.AirTime, mode) + if !isLocked(FieldAirSchedule) { + mergeScalar(&target.AirTime, source.AirTime, mode) + mergeScalar(&target.AirTimezone, source.AirTimezone, mode) + } // Genres follow provider priority during FillEmpty instead of unioning tags // from later providers, which can create noisy hybrid classifications. @@ -114,7 +117,10 @@ func MergeGlobalMetadata(source, target *MetadataResult, locked []MetadataField, mergeInt(&target.SeasonCount, source.SeasonCount, mode) mergeScalar(&target.FirstAirDate, source.FirstAirDate, mode) mergeScalar(&target.LastAirDate, source.LastAirDate, mode) - mergeScalar(&target.AirTime, source.AirTime, mode) + if !isLocked(FieldAirSchedule) { + mergeScalar(&target.AirTime, source.AirTime, mode) + mergeScalar(&target.AirTimezone, source.AirTimezone, mode) + } if !isLocked(FieldGenres) { mergePrioritizedStringSlice(&target.Genres, source.Genres, mode) diff --git a/internal/metadata/service.go b/internal/metadata/service.go index 4b699f87..3879ea8d 100644 --- a/internal/metadata/service.go +++ b/internal/metadata/service.go @@ -4762,6 +4762,9 @@ func itemToMetadataResult(item *models.MediaItem) *MetadataResult { if item.AirTime != nil { result.AirTime = *item.AirTime } + if item.AirTimezone != nil { + result.AirTimezone = *item.AirTimezone + } if item.ReleaseDate != nil { result.ReleaseDate = *item.ReleaseDate } @@ -4826,6 +4829,12 @@ func metadataResultToItem(r *MetadataResult, contentType string) *models.MediaIt if r.AirTime != "" { item.AirTime = &r.AirTime } + if r.AirTime != "" && r.AirTimezone == "" { + r.AirTimezone = catalog.InferAirTimezone(r.Networks, r.Countries) + } + if r.AirTimezone != "" { + item.AirTimezone = &r.AirTimezone + } if r.ReleaseDate != "" { item.ReleaseDate = &r.ReleaseDate } diff --git a/internal/metadata/types.go b/internal/metadata/types.go index b40f4f69..cb7e433b 100644 --- a/internal/metadata/types.go +++ b/internal/metadata/types.go @@ -39,6 +39,7 @@ const ( FieldTags FieldContentRating FieldImages + FieldAirSchedule ) // RefreshPriority controls queue ordering. @@ -178,6 +179,7 @@ type MetadataResult struct { FirstAirDate string LastAirDate string AirTime string + AirTimezone string } // Ratings holds ratings from multiple sources. diff --git a/internal/models/media.go b/internal/models/media.go index 54c64a82..30459504 100644 --- a/internal/models/media.go +++ b/internal/models/media.go @@ -314,6 +314,7 @@ type MediaItem struct { FirstAirDate *string // ISO date (series only), nullable LastAirDate *string // ISO date (series only), nullable AirTime *string // Series broadcast time (e.g. "20:00"), nullable + AirTimezone *string // Series broadcast timezone (IANA name, e.g. "America/New_York"), nullable ShowStatus string // Series lifecycle: "returning", "ended", "cancelled", "in_production", or "" if unknown (series only) People []ItemPerson MatchedAt *time.Time diff --git a/migrations/162_media_items_air_timezone.down.sql b/migrations/162_media_items_air_timezone.down.sql new file mode 100644 index 00000000..be2d1a6e --- /dev/null +++ b/migrations/162_media_items_air_timezone.down.sql @@ -0,0 +1 @@ +ALTER TABLE media_items DROP COLUMN IF EXISTS air_timezone; diff --git a/migrations/162_media_items_air_timezone.up.sql b/migrations/162_media_items_air_timezone.up.sql new file mode 100644 index 00000000..b4530669 --- /dev/null +++ b/migrations/162_media_items_air_timezone.up.sql @@ -0,0 +1 @@ +ALTER TABLE media_items ADD COLUMN IF NOT EXISTS air_timezone text; diff --git a/web/src/api/types.ts b/web/src/api/types.ts index 55c3a06e..e81ba3b0 100644 --- a/web/src/api/types.ts +++ b/web/src/api/types.ts @@ -831,6 +831,7 @@ export interface ItemDetail { first_air_date: string | null; last_air_date: string | null; air_time?: string | null; + air_timezone?: string | null; // Presigned image URLs. poster_url: string; @@ -2505,6 +2506,9 @@ export interface SectionItemUpcomingEvent { type: "movie" | "episode" | "season_premiere"; air_date: string; air_time?: string; + air_at?: string | null; + air_timezone?: string | null; + local_air_date?: string; episode_title?: string | null; season_number?: number | null; episode_number?: number | null; diff --git a/web/src/app.css b/web/src/app.css index 31b0696a..410149ec 100644 --- a/web/src/app.css +++ b/web/src/app.css @@ -343,7 +343,9 @@ --secondary: #232328; --secondary-foreground: #d0d0d6; --muted: #1c1c20; - --muted-foreground: #6e6e78; + /* Lightened from #6e6e78 (~3.4:1) to meet WCAG AA: ~6.3:1 on --background, + ~5.3:1 even on the lightest chip surface (#232328). */ + --muted-foreground: #9696a0; --accent: #232328; --accent-foreground: #e8e8ec; @@ -403,7 +405,9 @@ --secondary: #e8e8ec; --secondary-foreground: #3a3a42; --muted: #ebebef; - --muted-foreground: #78787f; + /* Darkened from #78787f (~4.0:1) to meet WCAG AA on light surfaces + (~4.9:1 on the #e8e8ec chip, ~5.4:1 on --background). */ + --muted-foreground: #63636b; --accent: #e8e8ec; --accent-foreground: #1a1a1e; diff --git a/web/src/components/AdminCatalogMaintenance.tsx b/web/src/components/AdminCatalogMaintenance.tsx index 16892fd5..7926105b 100644 --- a/web/src/components/AdminCatalogMaintenance.tsx +++ b/web/src/components/AdminCatalogMaintenance.tsx @@ -364,9 +364,10 @@ export default function AdminCatalogMaintenance() { variant="ghost" size="icon" className="shrink-0" + aria-label="Remove path rewrite" onClick={() => removeRewrite(index)} > - +