silo-server/internal/metadata/match_queue_state_db_test.go

package metadata

import (
	"context"
	"fmt"
	"strings"
	"testing"
	"time"

	"github.com/Silo-Server/silo-server/internal/models"
	scannerrepo "github.com/Silo-Server/silo-server/internal/scanner"
)

const queueStatePending = "pending"

func TestBoundedMatchFailureMessageRedactsAndLimits(t *testing.T) {
	t.Parallel()
	message := "provider failed?api_key=secret&token=also-secret " + strings.Repeat("x", 1200)
	got := boundedMatchFailureMessage(message)
	if strings.Contains(got, "secret") {
		t.Fatalf("failure message leaked a secret: %q", got)
	}
	if len([]rune(got)) != 1000 {
		t.Fatalf("failure message length = %d, want 1000", len([]rune(got)))
	}
}

func TestBoundedMatchFailureMessageRedactsHeaderAndBearerForms(t *testing.T) {
	t.Parallel()
	got := boundedMatchFailureMessage("Authorization: Bearer secret-token; password: hunter2")
	for _, secret := range []string{"secret-token", "hunter2"} {
		if strings.Contains(got, secret) {
			t.Fatalf("failure message leaked %q: %q", secret, got)
		}
	}
}

func TestBoundedMatchDecisionLimitsProviderControlledFields(t *testing.T) {
	t.Parallel()
	decision := &MatchDecision{Outcome: MatchOutcome(strings.Repeat("x", 100)), CandidateCount: 99, Threshold: 55}
	for i := 0; i < 5; i++ {
		candidate := MatchDecisionCandidate{
			Title: strings.Repeat("t", 400), MatchedTitle: strings.Repeat("m", 400),
			ProviderIDs: make(map[string]string), Score: 42,
			Sources: make([]string, 12), Reasons: make([]string, 12),
		}
		for j := 0; j < 12; j++ {
			candidate.ProviderIDs[fmt.Sprintf("provider-%02d", j)] = strings.Repeat("i", 400)
			candidate.Sources[j] = strings.Repeat("s", 100)
			candidate.Reasons[j] = strings.Repeat("r", 200)
		}
		candidate.ProviderIDs["api_key"] = "must-not-persist"
		decision.TopCandidates = append(decision.TopCandidates, candidate)
	}

	got := boundedMatchDecision(decision)
	if len(got.TopCandidates) != 3 || len(got.TopCandidates[0].ProviderIDs) != 8 || len(got.TopCandidates[0].Sources) != 8 || len(got.TopCandidates[0].Reasons) != 8 {
		t.Fatalf("bounded decision sizes = candidates:%d ids:%d sources:%d reasons:%d", len(got.TopCandidates), len(got.TopCandidates[0].ProviderIDs), len(got.TopCandidates[0].Sources), len(got.TopCandidates[0].Reasons))
	}
	if len([]rune(got.TopCandidates[0].Title)) != 256 || len([]rune(got.TopCandidates[0].MatchedTitle)) != 256 {
		t.Fatalf("bounded title lengths = %d/%d", len([]rune(got.TopCandidates[0].Title)), len([]rune(got.TopCandidates[0].MatchedTitle)))
	}
	if _, exists := got.TopCandidates[0].ProviderIDs["api_key"]; exists {
		t.Fatal("bounded decision retained a credential-shaped provider ID")
	}
}

func TestNormalizeMatchFailureKindTreatsUnknownAsTransient(t *testing.T) {
	t.Parallel()
	if got := normalizeMatchFailureKind(MatchOutcomeCandidateRejected); got != MatchOutcomeCandidateRejected {
		t.Fatalf("known failure kind = %q", got)
	}
	if got := normalizeMatchFailureKind(MatchOutcome("unexpected-" + strings.Repeat("x", 500))); got != MatchOutcomeProviderTransient {
		t.Fatalf("unknown failure kind = %q, want provider_transient", got)
	}
}

func TestMatchQueueFingerprintIncludesMatcherAndProviderConfiguration(t *testing.T) {
	t.Parallel()
	expression := matchQueueInputFingerprintSQL("mf.file_path", "'movie'", "mf.media_folder_id", "folders.metadata_language", movieMatcherRevision)
	for _, required := range []string{
		"mf.file_path", "'movie'", "folders.metadata_language", "installation.version",
		"chain.priority", "chain.capability_id", "plugin_runtime_configs", "config.updated_at::text",
		fmt.Sprintf("|%d|", movieMatcherRevision),
	} {
		if !strings.Contains(expression, required) {
			t.Fatalf("fingerprint expression %q does not contain %q", expression, required)
		}
	}
}

func TestSeriesMatchQueueFingerprintIncludesEpisodePathShape(t *testing.T) {
	t.Parallel()
	expression := seriesMatchQueueInputFingerprintSQL("q.observed_root_path", "q.media_folder_id", "folders.metadata_language")
	for _, required := range []string{"shape_file.file_path", "shape_file.observed_root_path", "shape_file.missing_since", "shape_file.extra_id", fmt.Sprintf("|%d|", seriesMatcherRevision)} {
		if !strings.Contains(expression, required) {
			t.Fatalf("series fingerprint expression %q does not contain %q", expression, required)
		}
	}
	if movieMatcherRevision != 10 {
		t.Fatalf("movie matcher revision = %d, want shared title-normalization revision 10", movieMatcherRevision)
	}
	if seriesMatcherRevision != 10 {
		t.Fatalf("series matcher revision = %d, want consensus revision 10", seriesMatcherRevision)
	}
}

func TestMatchQueueSharedTitleRevisionWakesMoviesAndSeries(t *testing.T) {
	pool := chainBuiltinTestPool(t)
	ctx := context.Background()
	previousMatcherRevision := movieMatcherRevision - 1

	movieFolderID := insertTestFolder(t, pool, "movie")
	moviePath := fmt.Sprintf("/test/revision-isolation-%d/Movie.mkv", time.Now().UnixNano())
	var movieFileID int
	if err := pool.QueryRow(ctx, `
		INSERT INTO media_files (media_folder_id, file_path, base_type, file_size)
		VALUES ($1, $2, 'movie', 0) RETURNING id
	`, movieFolderID, moviePath).Scan(&movieFileID); err != nil {
		t.Fatalf("seed movie file: %v", err)
	}
	movieRepo := NewMovieMatchQueueRepository(pool, scannerrepo.NewFileRepository(pool))
	if err := movieRepo.EnqueueMovieFile(ctx, movieFileID); err != nil {
		t.Fatalf("EnqueueMovieFile(): %v", err)
	}
	if _, err := pool.Exec(ctx, `
		UPDATE movie_match_queue
		SET state = 'parked', available_at = NOW() + interval '24 hours', parked_at = NOW(),
			matcher_revision = $2
		WHERE media_file_id = $1
	`, movieFileID, previousMatcherRevision); err != nil {
		t.Fatalf("park movie row: %v", err)
	}

	seriesFolderID := insertTestFolder(t, pool, "series")
	seriesRoot := fmt.Sprintf("/test/revision-isolation-%d/Series", time.Now().UnixNano())
	if _, err := pool.Exec(ctx, `
		INSERT INTO media_files (
			media_folder_id, file_path, observed_root_path, base_type,
			season_number, episode_number, file_size
		) VALUES ($1, $2, $3, 'series', 1, 1, 0)
	`, seriesFolderID, seriesRoot+"/Season 01/Series S01E01.mkv", seriesRoot); err != nil {
		t.Fatalf("seed series file: %v", err)
	}
	seriesRepo := NewSeriesRootMatchQueueRepository(pool)
	if err := seriesRepo.EnqueueSeriesRoot(ctx, seriesFolderID, seriesRoot); err != nil {
		t.Fatalf("EnqueueSeriesRoot(): %v", err)
	}
	if _, err := pool.Exec(ctx, `
		UPDATE series_root_match_queue
		SET state = 'parked', available_at = NOW() + interval '24 hours', parked_at = NOW(),
			matcher_revision = $3
		WHERE media_folder_id = $1 AND observed_root_path = $2
	`, seriesFolderID, seriesRoot, previousMatcherRevision); err != nil {
		t.Fatalf("seed pre-change series revision: %v", err)
	}

	if _, err := movieRepo.WakeForChangedInputs(ctx); err != nil {
		t.Fatalf("movie WakeForChangedInputs(): %v", err)
	}
	var movieState string
	var movieRevision int
	if err := pool.QueryRow(ctx, `
		SELECT state, matcher_revision FROM movie_match_queue WHERE media_file_id = $1
	`, movieFileID).Scan(&movieState, &movieRevision); err != nil {
		t.Fatalf("load movie queue row: %v", err)
	}
	if movieState != queueStatePending || movieRevision != movieMatcherRevision {
		t.Fatalf("movie row was not awakened: state=%q revision=%d", movieState, movieRevision)
	}

	if _, err := seriesRepo.WakeForChangedInputs(ctx); err != nil {
		t.Fatalf("series WakeForChangedInputs(): %v", err)
	}
	var seriesState string
	var seriesRevision int
	if err := pool.QueryRow(ctx, `
		SELECT state, matcher_revision FROM series_root_match_queue
		WHERE media_folder_id = $1 AND observed_root_path = $2
	`, seriesFolderID, seriesRoot).Scan(&seriesState, &seriesRevision); err != nil {
		t.Fatalf("load series queue row: %v", err)
	}
	if seriesState != queueStatePending || seriesRevision != seriesMatcherRevision {
		t.Fatalf("series row was not awakened: state=%q revision=%d", seriesState, seriesRevision)
	}
}

func TestSeriesMatchQueueDeterministicFailuresParkAndRetryNowResets(t *testing.T) {
	pool := chainBuiltinTestPool(t)
	ctx := context.Background()
	folderID := insertTestFolder(t, pool, "series")
	root := fmt.Sprintf("/test/match-queue-%d", time.Now().UnixNano())
	if _, err := pool.Exec(ctx, `
		INSERT INTO series_root_match_queue (media_folder_id, observed_root_path, available_at)
		VALUES ($1, $2, NOW())`, folderID, root); err != nil {
		t.Fatalf("seed series match queue: %v", err)
	}

	repo := NewSeriesRootMatchQueueRepository(pool)
	for attempt, wantDelay := range []time.Duration{time.Hour, 24 * time.Hour, 24 * time.Hour} {
		leaseToken := fmt.Sprintf("deterministic-lease-%d", attempt)
		if _, err := pool.Exec(ctx, `UPDATE series_root_match_queue SET lease_token = $3 WHERE media_folder_id = $1 AND observed_root_path = $2`, folderID, root, leaseToken); err != nil {
			t.Fatalf("seed lease token: %v", err)
		}
		before := time.Now()
		if err := repo.UpdateFailure(ctx, folderID, root, leaseToken, MatchFailure{Kind: MatchOutcomeCandidateRejected, Message: "score below threshold"}); err != nil {
			t.Fatalf("UpdateFailure(%d): %v", attempt+1, err)
		}
		var state string
		var deterministicCount int
		var availableAt time.Time
		var parkedAt *time.Time
		if err := pool.QueryRow(ctx, `
			SELECT state, deterministic_attempt_count, available_at, parked_at
			FROM series_root_match_queue WHERE media_folder_id = $1 AND observed_root_path = $2`,
			folderID, root).Scan(&state, &deterministicCount, &availableAt, &parkedAt); err != nil {
			t.Fatalf("load queue state: %v", err)
		}
		if deterministicCount != attempt+1 {
			t.Fatalf("deterministic count = %d, want %d", deterministicCount, attempt+1)
		}
		wantState := queueStatePending
		if attempt == 2 {
			wantState = "parked"
		}
		if state != wantState {
			t.Fatalf("state = %q, want %q", state, wantState)
		}
		if attempt == 2 && parkedAt == nil {
			t.Fatal("parked_at is nil after third deterministic failure")
		}
		if availableAt.Before(before.Add(wantDelay - time.Minute)) {
			t.Fatalf("available_at = %v, want approximately %v later", availableAt, wantDelay)
		}
	}

	if _, err := repo.RetryNowByFolder(ctx, folderID); err != nil {
		t.Fatalf("RetryNowByFolder(): %v", err)
	}
	var state, failureKind string
	var deterministicCount int
	var availableAt time.Time
	var parkedAt *time.Time
	if err := pool.QueryRow(ctx, `
		SELECT state, failure_kind, deterministic_attempt_count, available_at, parked_at
		FROM series_root_match_queue WHERE media_folder_id = $1 AND observed_root_path = $2`,
		folderID, root).Scan(&state, &failureKind, &deterministicCount, &availableAt, &parkedAt); err != nil {
		t.Fatalf("load retried queue state: %v", err)
	}
	if state != queueStatePending || failureKind != "" || deterministicCount != 0 || parkedAt != nil {
		t.Fatalf("retry state = (%q, %q, %d, %v)", state, failureKind, deterministicCount, parkedAt)
	}
	if availableAt.After(time.Now().Add(time.Minute)) {
		t.Fatalf("RetryNow left available_at in the future: %v", availableAt)
	}
}

func TestSeriesMatchQueueTransientFailureDoesNotConsumeDeterministicBudget(t *testing.T) {
	pool := chainBuiltinTestPool(t)
	ctx := context.Background()
	folderID := insertTestFolder(t, pool, "series")
	root := fmt.Sprintf("/test/match-queue-transient-%d", time.Now().UnixNano())
	if _, err := pool.Exec(ctx, `
		INSERT INTO series_root_match_queue (media_folder_id, observed_root_path, available_at)
		VALUES ($1, $2, NOW())`, folderID, root); err != nil {
		t.Fatalf("seed series match queue: %v", err)
	}

	repo := NewSeriesRootMatchQueueRepository(pool)
	const leaseToken = "transient-lease"
	if _, err := pool.Exec(ctx, `UPDATE series_root_match_queue SET lease_token = $3 WHERE media_folder_id = $1 AND observed_root_path = $2`, folderID, root, leaseToken); err != nil {
		t.Fatalf("seed lease token: %v", err)
	}
	if err := repo.UpdateFailure(ctx, folderID, root, leaseToken, MatchFailure{Kind: MatchOutcomeProviderTransient, Message: "HTTP 429"}); err != nil {
		t.Fatalf("UpdateFailure(): %v", err)
	}
	var state string
	var deterministicCount int
	if err := pool.QueryRow(ctx, `
		SELECT state, deterministic_attempt_count FROM series_root_match_queue
		WHERE media_folder_id = $1 AND observed_root_path = $2`, folderID, root).Scan(&state, &deterministicCount); err != nil {
		t.Fatalf("load transient queue state: %v", err)
	}
	if state != queueStatePending || deterministicCount != 0 {
		t.Fatalf("transient state = (%q, %d), want pending with zero deterministic attempts", state, deterministicCount)
	}
}

func TestSeriesMatchQueueWakeForChangedInputsResetsOnlyChangedRows(t *testing.T) {
	pool := chainBuiltinTestPool(t)
	ctx := context.Background()
	folderID := insertTestFolder(t, pool, "series")
	root := fmt.Sprintf("/test/match-input-wake-%d", time.Now().UnixNano())
	if _, err := pool.Exec(ctx, `
		INSERT INTO series_root_match_queue (
			media_folder_id, observed_root_path, available_at, state,
			failure_kind, failure_detail, deterministic_attempt_count,
			input_fingerprint, matcher_revision, parked_at, last_error
		) VALUES ($1, $2, NOW() + interval '24 hours', 'parked',
			'candidate_rejected', '{"message":"old"}'::jsonb, 3,
			'old-fingerprint', 0, NOW(), 'old')
	`, folderID, root); err != nil {
		t.Fatalf("seed changed-input queue row: %v", err)
	}
	t.Cleanup(func() {
		_, _ = pool.Exec(ctx, `DELETE FROM series_root_match_queue WHERE media_folder_id = $1 AND observed_root_path = $2`, folderID, root)
	})

	repo := NewSeriesRootMatchQueueRepository(pool)
	const staleLeaseToken = "series-stale-lease"
	if _, err := pool.Exec(ctx, `
		UPDATE series_root_match_queue SET lease_token = $3
		WHERE media_folder_id = $1 AND observed_root_path = $2
	`, folderID, root, staleLeaseToken); err != nil {
		t.Fatalf("seed active lease: %v", err)
	}
	woken, err := repo.WakeForChangedInputs(ctx)
	if err != nil {
		t.Fatalf("WakeForChangedInputs(): %v", err)
	}
	if woken < 1 {
		t.Fatalf("woken = %d, want at least seeded row", woken)
	}
	var state, failureKind, lastError, fingerprint, leaseToken string
	var deterministicCount, revision int
	var availableAt time.Time
	var parkedAt *time.Time
	var rerunRequested bool
	if err := pool.QueryRow(ctx, `
		SELECT state, failure_kind, last_error, deterministic_attempt_count,
			input_fingerprint, matcher_revision, available_at, parked_at,
			lease_token, rerun_requested
		FROM series_root_match_queue
		WHERE media_folder_id = $1 AND observed_root_path = $2
	`, folderID, root).Scan(&state, &failureKind, &lastError, &deterministicCount, &fingerprint, &revision, &availableAt, &parkedAt, &leaseToken, &rerunRequested); err != nil {
		t.Fatalf("load woken row: %v", err)
	}
	if state != queueStatePending || failureKind != "" || lastError != "" || deterministicCount != 0 || fingerprint == "" || fingerprint == "old-fingerprint" || revision != seriesMatcherRevision || parkedAt != nil {
		t.Fatalf("woken row = state:%q failure:%q last:%q deterministic:%d fingerprint:%q revision:%d available:%v parked:%v", state, failureKind, lastError, deterministicCount, fingerprint, revision, availableAt, parkedAt)
	}
	if leaseToken != staleLeaseToken || !rerunRequested || availableAt.Before(time.Now().Add(time.Hour)) {
		t.Fatalf("woken lease ownership was not preserved: rerun:%v available:%v", rerunRequested, availableAt)
	}
	if woken, err := repo.WakeForChangedInputs(ctx); err != nil || woken != 0 {
		t.Fatalf("unchanged WakeForChangedInputs() = (%d, %v), want (0, nil)", woken, err)
	}
	if err := repo.UpdateFailure(ctx, folderID, root, staleLeaseToken, MatchFailure{
		Kind: MatchOutcomeCandidateRejected, Message: "stale worker result",
	}); err != nil {
		t.Fatalf("stale UpdateFailure(): %v", err)
	}
	if err := pool.QueryRow(ctx, `
		SELECT failure_kind FROM series_root_match_queue
		WHERE media_folder_id = $1 AND observed_root_path = $2
	`, folderID, root).Scan(&failureKind); err != nil {
		t.Fatalf("load row after stale series update: %v", err)
	}
	if failureKind != "" {
		t.Fatalf("stale series worker overwrote awakened row with failure %q", failureKind)
	}
	if err := pool.QueryRow(ctx, `
		SELECT lease_token, rerun_requested, available_at
		FROM series_root_match_queue
		WHERE media_folder_id = $1 AND observed_root_path = $2
	`, folderID, root).Scan(&leaseToken, &rerunRequested, &availableAt); err != nil {
		t.Fatalf("load released series rerun: %v", err)
	}
	if leaseToken != "" || !rerunRequested || availableAt.After(time.Now().Add(time.Minute)) {
		t.Fatalf("released rerun retained lease ownership: rerun:%v available:%v", rerunRequested, availableAt)
	}

	if _, err := pool.Exec(ctx, `UPDATE media_folders SET metadata_language = 'da' WHERE id = $1`, folderID); err != nil {
		t.Fatalf("change folder language: %v", err)
	}
	if _, err := pool.Exec(ctx, `
		UPDATE series_root_match_queue
		SET available_at = NOW() + interval '24 hours', deterministic_attempt_count = 2,
			failure_kind = 'candidate_rejected', last_error = 'old'
		WHERE media_folder_id = $1 AND observed_root_path = $2
	`, folderID, root); err != nil {
		t.Fatalf("back off queue row before language change wake: %v", err)
	}
	if woken, err := repo.WakeForChangedInputs(ctx); err != nil || woken < 1 {
		t.Fatalf("language-change WakeForChangedInputs() = (%d, %v), want seeded row", woken, err)
	}

	installationID := insertTestInstallation(t, pool, "plugin", true)
	insertTestCapability(t, pool, installationID, "config-fingerprint", `{}`)
	if _, err := pool.Exec(ctx, `
		INSERT INTO library_provider_chains (
			media_folder_id, plugin_installation_id, capability_id, capability_type,
			content_level, priority, enabled
		) VALUES ($1, $2, 'config-fingerprint', 'metadata_provider.v1', 'item', 1, true)
	`, folderID, installationID); err != nil {
		t.Fatalf("seed relevant provider chain: %v", err)
	}
	if woken, err := repo.WakeForChangedInputs(ctx); err != nil || woken < 1 {
		t.Fatalf("provider-chain WakeForChangedInputs() = (%d, %v), want seeded row", woken, err)
	}
	if woken, err := repo.WakeForChangedInputs(ctx); err != nil || woken != 0 {
		t.Fatalf("stable provider chain wake = (%d, %v), want (0, nil)", woken, err)
	}
	if _, err := pool.Exec(ctx, `
		INSERT INTO media_files (
			media_folder_id, file_path, observed_root_path, base_type,
			season_number, episode_number, file_size
		) VALUES ($1, $2, $3, 'series', 1, 8, 0)
	`, folderID, root+"/Season 01/Show S01E08.mkv", root); err != nil {
		t.Fatalf("add episode path to series shape: %v", err)
	}
	if woken, err := repo.WakeForChangedInputs(ctx); err != nil || woken < 1 {
		t.Fatalf("episode-shape WakeForChangedInputs() = (%d, %v), want seeded row", woken, err)
	}
	if woken, err := repo.WakeForChangedInputs(ctx); err != nil || woken != 0 {
		t.Fatalf("stable episode shape wake = (%d, %v), want (0, nil)", woken, err)
	}
	if _, err := pool.Exec(ctx, `
		INSERT INTO plugin_runtime_configs (plugin_installation_id, config_key, config_value)
		VALUES ($1, 'metadata', '{"api_key":"changed-but-never-persisted-in-the-queue"}'::jsonb)
	`, installationID); err != nil {
		t.Fatalf("change relevant provider config: %v", err)
	}
	if woken, err := repo.WakeForChangedInputs(ctx); err != nil || woken < 1 {
		t.Fatalf("provider-config WakeForChangedInputs() = (%d, %v), want seeded row", woken, err)
	}
	if err := pool.QueryRow(ctx, `
		SELECT input_fingerprint FROM series_root_match_queue
		WHERE media_folder_id = $1 AND observed_root_path = $2
	`, folderID, root).Scan(&fingerprint); err != nil {
		t.Fatalf("load provider-config fingerprint: %v", err)
	}
	if strings.Contains(fingerprint, "api_key") || strings.Contains(fingerprint, "changed-but-never") {
		t.Fatalf("queue fingerprint leaked provider configuration: %q", fingerprint)
	}
}

func TestMovieMatchQueueRetryDuringLeaseQueuesFencedRerun(t *testing.T) {
	pool := chainBuiltinTestPool(t)
	ctx := context.Background()
	folderID := insertTestFolder(t, pool, "movie")
	root := fmt.Sprintf("/test/claim-lease-%d", time.Now().UnixNano())
	var fileID int
	if err := pool.QueryRow(ctx, `
		INSERT INTO media_files (media_folder_id, file_path, base_type, file_size)
		VALUES ($1, $2, 'movie', 0) RETURNING id
	`, folderID, root+"/Movie.mkv").Scan(&fileID); err != nil {
		t.Fatalf("seed movie file: %v", err)
	}

	repo := NewMovieMatchQueueRepository(pool, scannerrepo.NewFileRepository(pool))
	if err := repo.EnqueueMovieFile(ctx, fileID); err != nil {
		t.Fatalf("EnqueueMovieFile(): %v", err)
	}
	claimTestMovie := func() ([]models.MovieMatchJob, error) {
		return repo.ClaimByFolderAndPathPrefix(ctx, folderID, root, 1, time.Time{})
	}
	claimed, err := claimTestMovie()
	if err != nil || len(claimed) != 1 || claimed[0].File == nil || claimed[0].File.ID != fileID || claimed[0].LeaseToken == "" {
		t.Fatalf("first Claim() = (%#v, %v), want file %d", claimed, err, fileID)
	}
	claimedAgain, err := claimTestMovie()
	if err != nil || len(claimedAgain) != 0 {
		t.Fatalf("second Claim() during lease = (%#v, %v), want empty", claimedAgain, err)
	}

	var leasedUntil time.Time
	if err := pool.QueryRow(ctx, `SELECT available_at FROM movie_match_queue WHERE media_file_id = $1`, fileID).Scan(&leasedUntil); err != nil {
		t.Fatalf("load claim lease: %v", err)
	}
	if leasedUntil.Before(time.Now().Add(time.Hour)) {
		t.Fatalf("claim lease = %v, want comfortably beyond one hour", leasedUntil)
	}

	if affected, err := repo.RetryNowByFolder(ctx, folderID); err != nil || affected != 1 {
		t.Fatalf("RetryNowByFolder() = (%d, %v), want (1, nil)", affected, err)
	}
	var state, leaseToken string
	var availableAt time.Time
	var rerunRequested bool
	if err := pool.QueryRow(ctx, `
		SELECT state, available_at, lease_token, rerun_requested
		FROM movie_match_queue
		WHERE media_file_id = $1
	`, fileID).Scan(&state, &availableAt, &leaseToken, &rerunRequested); err != nil {
		t.Fatalf("load retried movie row: %v", err)
	}
	if state != queueStatePending || availableAt.Before(time.Now().Add(time.Hour)) {
		t.Fatalf("retried movie row = state %q available %v, want active lease preserved", state, availableAt)
	}
	if leaseToken != claimed[0].LeaseToken || !rerunRequested {
		t.Fatalf("retried movie ownership was not preserved: rerun %v", rerunRequested)
	}
	if reclaimed, err := claimTestMovie(); err != nil || len(reclaimed) != 0 {
		t.Fatalf("Claim() while original worker runs = (%#v, %v), want empty", reclaimed, err)
	}
	if err := repo.Delete(ctx, fileID, claimed[0].LeaseToken); err != nil {
		t.Fatalf("original leased completion: %v", err)
	}
	var remaining int
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM movie_match_queue WHERE media_file_id = $1`, fileID).Scan(&remaining); err != nil {
		t.Fatalf("count retried movie row: %v", err)
	}
	if remaining != 1 {
		t.Fatalf("original completion deleted requested rerun; remaining = %d", remaining)
	}
	if err := repo.UpdateFailure(ctx, fileID, claimed[0].LeaseToken, MatchFailure{
		Kind: MatchOutcomeCandidateRejected, Message: "stale worker result",
	}); err != nil {
		t.Fatalf("stale leased failure: %v", err)
	}
	var failureKind string
	if err := pool.QueryRow(ctx, `SELECT failure_kind FROM movie_match_queue WHERE media_file_id = $1`, fileID).Scan(&failureKind); err != nil {
		t.Fatalf("load retried movie failure kind: %v", err)
	}
	if failureKind != "" {
		t.Fatalf("stale lease overwrote newly awakened row with failure %q", failureKind)
	}

	reclaimed, err := claimTestMovie()
	if err != nil || len(reclaimed) != 1 {
		t.Fatalf("Claim() after original completion = (%#v, %v), want one row", reclaimed, err)
	}
	if !reclaimed[0].RerunRequested {
		t.Fatal("reclaimed job did not carry the forced-rerun marker")
	}
	if _, err := pool.Exec(ctx, `
		UPDATE movie_match_queue SET available_at = NOW()
		WHERE media_file_id = $1
	`, fileID); err != nil {
		t.Fatalf("expire forced rerun lease: %v", err)
	}
	expiredReplacement, err := claimTestMovie()
	if err != nil || len(expiredReplacement) != 1 {
		t.Fatalf("Claim() after forced lease expiry = (%#v, %v), want one row", expiredReplacement, err)
	}
	if !expiredReplacement[0].RerunRequested {
		t.Fatal("expired forced rerun lost its durable intent")
	}
	if expiredReplacement[0].LeaseToken == reclaimed[0].LeaseToken {
		t.Fatal("expired forced rerun was not assigned fresh ownership")
	}
	if affected, err := repo.ReleaseLease(ctx, expiredReplacement[0].LeaseToken); err != nil || affected != 1 {
		t.Fatalf("ReleaseLease() = (%d, %v), want (1, nil)", affected, err)
	}
	reclaimedAgain, err := claimTestMovie()
	if err != nil || len(reclaimedAgain) != 1 {
		t.Fatalf("Claim() after ReleaseLease = (%#v, %v), want one immediately claimable row", reclaimedAgain, err)
	}
	if reclaimedAgain[0].LeaseToken == expiredReplacement[0].LeaseToken {
		t.Fatal("released claim was not assigned a fresh ownership token")
	}
	if !reclaimedAgain[0].RerunRequested {
		t.Fatal("released forced rerun lost its durable intent")
	}
	if err := repo.UpdateFailure(ctx, fileID, reclaimedAgain[0].LeaseToken, MatchFailure{
		Kind: "provider_transient", Message: "temporary provider outage",
	}); err != nil {
		t.Fatalf("forced rerun failure: %v", err)
	}
	var rerunAfterFailure, leaseRerunAfterFailure bool
	if err := pool.QueryRow(ctx, `
		SELECT rerun_requested, lease_forced_rerun
		FROM movie_match_queue
		WHERE media_file_id = $1
	`, fileID).Scan(&rerunAfterFailure, &leaseRerunAfterFailure); err != nil {
		t.Fatalf("load failed forced rerun: %v", err)
	}
	if !rerunAfterFailure || leaseRerunAfterFailure {
		t.Fatalf("failed forced rerun state = requested:%v leased:%v", rerunAfterFailure, leaseRerunAfterFailure)
	}
}

func TestMatchQueueSyncDeletesIneligibleReruns(t *testing.T) {
	pool := chainBuiltinTestPool(t)
	ctx := context.Background()

	t.Run("movie", func(t *testing.T) {
		folderID := insertTestFolder(t, pool, "movie")
		var fileID int
		if err := pool.QueryRow(ctx, `
			INSERT INTO media_files (media_folder_id, file_path, base_type, file_size)
			VALUES ($1, $2, 'movie', 0) RETURNING id
		`, folderID, fmt.Sprintf("/test/ineligible-rerun-%d/Movie.mkv", time.Now().UnixNano())).Scan(&fileID); err != nil {
			t.Fatalf("seed movie file: %v", err)
		}
		repo := NewMovieMatchQueueRepository(pool, scannerrepo.NewFileRepository(pool))
		if err := repo.EnqueueMovieFile(ctx, fileID); err != nil {
			t.Fatalf("EnqueueMovieFile(): %v", err)
		}
		if _, err := pool.Exec(ctx, `
			UPDATE movie_match_queue
			SET rerun_requested = true, lease_token = $2, available_at = NOW() + interval '24 hours'
			WHERE media_file_id = $1
		`, fileID, "cleanup-owner"); err != nil {
			t.Fatalf("seed movie rerun: %v", err)
		}
		if _, err := pool.Exec(ctx, `UPDATE media_files SET missing_since = NOW() WHERE id = $1`, fileID); err != nil {
			t.Fatalf("mark movie missing: %v", err)
		}
		if err := repo.SyncForFolder(ctx, folderID); err != nil {
			t.Fatalf("SyncForFolder(): %v", err)
		}
		var remaining int
		if err := pool.QueryRow(ctx, `SELECT count(*) FROM movie_match_queue WHERE media_file_id = $1`, fileID).Scan(&remaining); err != nil {
			t.Fatalf("count movie rerun: %v", err)
		}
		if remaining != 0 {
			t.Fatalf("ineligible movie reruns remaining = %d, want 0", remaining)
		}
	})

	t.Run("series", func(t *testing.T) {
		folderID := insertTestFolder(t, pool, "series")
		root := fmt.Sprintf("/test/ineligible-series-rerun-%d", time.Now().UnixNano())
		var fileID int
		if err := pool.QueryRow(ctx, `
			INSERT INTO media_files (
				media_folder_id, file_path, observed_root_path, base_type,
				season_number, episode_number, file_size
			) VALUES ($1, $2, $3, 'series', 1, 1, 0)
			RETURNING id
		`, folderID, root+"/Season 01/Show S01E01.mkv", root).Scan(&fileID); err != nil {
			t.Fatalf("seed series file: %v", err)
		}
		repo := NewSeriesRootMatchQueueRepository(pool)
		if err := repo.EnqueueSeriesRoot(ctx, folderID, root); err != nil {
			t.Fatalf("EnqueueSeriesRoot(): %v", err)
		}
		if _, err := pool.Exec(ctx, `
			UPDATE series_root_match_queue
			SET rerun_requested = true, lease_token = $3, available_at = NOW() + interval '24 hours'
			WHERE media_folder_id = $1 AND observed_root_path = $2
		`, folderID, root, "cleanup-owner"); err != nil {
			t.Fatalf("seed series rerun: %v", err)
		}
		if _, err := pool.Exec(ctx, `UPDATE media_files SET missing_since = NOW() WHERE id = $1`, fileID); err != nil {
			t.Fatalf("mark series file missing: %v", err)
		}
		if err := repo.SyncForFolder(ctx, folderID); err != nil {
			t.Fatalf("SyncForFolder(): %v", err)
		}
		var remaining int
		if err := pool.QueryRow(ctx, `
			SELECT count(*) FROM series_root_match_queue
			WHERE media_folder_id = $1 AND observed_root_path = $2
		`, folderID, root).Scan(&remaining); err != nil {
			t.Fatalf("count series rerun: %v", err)
		}
		if remaining != 0 {
			t.Fatalf("ineligible series reruns remaining = %d, want 0", remaining)
		}
	})
}