tver/internal/store/store_test.go

package store

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	embeddedpostgres "github.com/fergusstrange/embedded-postgres"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

var adminPool *pgxpool.Pool
var testPort int
var dbSequence atomic.Uint64

func TestMain(m *testing.M) {
	root, err := os.MkdirTemp("/private/tmp", "tver-pg17-")
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
	listener, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
	testPort = listener.Addr().(*net.TCPAddr).Port
	listener.Close()
	config := embeddedpostgres.DefaultConfig().Version(embeddedpostgres.V17).
		Port(uint32(testPort)).CachePath("/private/tmp/tver-embedded-pg-cache").
		RuntimePath(filepath.Join(root, "runtime")).DataPath(filepath.Join(root, "data")).
		Logger(io.Discard).StartTimeout(60 * time.Second)
	db := embeddedpostgres.NewDatabase(config)
	if err := db.Start(); err != nil {
		fmt.Fprintf(os.Stderr, "embedded Postgres 17 start: %v\n", err)
		_ = os.RemoveAll(root)
		os.Exit(1)
	}
	adminPool, err = pgxpool.New(context.Background(), config.GetConnectionURL())
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		_ = db.Stop()
		_ = os.RemoveAll(root)
		os.Exit(1)
	}
	code := m.Run()
	adminPool.Close()
	if err := db.Stop(); err != nil {
		fmt.Fprintf(os.Stderr, "embedded Postgres 17 stop: %v\n", err)
		code = 1
	}
	_ = os.RemoveAll(root)
	os.Exit(code)
}

func testPool(t *testing.T, migrate bool) *pgxpool.Pool {
	t.Helper()
	name := fmt.Sprintf("tver_test_%d", dbSequence.Add(1))
	ctx := context.Background()
	if _, err := adminPool.Exec(ctx, `CREATE DATABASE `+name); err != nil {
		t.Fatal(err)
	}
	url := fmt.Sprintf("postgres://postgres:postgres@localhost:%d/%s", testPort, name)
	pool, err := pgxpool.New(ctx, url)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() {
		pool.Close()
		if _, err := adminPool.Exec(ctx, `DROP DATABASE `+name); err != nil {
			t.Error(err)
		}
	})
	if migrate {
		if err := Migrate(ctx, pool); err != nil {
			t.Fatal(err)
		}
	}
	return pool
}

func sampleEpisode(id string) Episode {
	end := time.Date(2026, 10, 4, 12, 0, 0, 0, time.UTC)
	return Episode{ID: id, SeriesID: "series-1", SeriesTitle: "番組", Title: "第1話",
		BroadcasterName: "放送局", DurationSeconds: 300, EndAt: &end, ThumbnailPath: "/thumb.jpg"}
}

func TestMigrate(t *testing.T) {
	pool := testPool(t, false)
	ctx := context.Background()
	if err := Migrate(ctx, pool); err != nil {
		t.Fatal(err)
	}
	var first time.Time
	if err := pool.QueryRow(ctx, `SELECT applied_at FROM schema_migrations WHERE version='migrations/001_initial.sql'`).Scan(&first); err != nil {
		t.Fatal(err)
	}
	if err := Migrate(ctx, pool); err != nil {
		t.Fatal(err)
	}
	var count int
	var second time.Time
	if err := pool.QueryRow(ctx, `SELECT count(*), min(applied_at) FROM schema_migrations`).Scan(&count, &second); err != nil {
		t.Fatal(err)
	}
	if count != 3 || !first.Equal(second) {
		t.Fatalf("再適用で履歴が変わった: %d %v %v", count, first, second)
	}
}

func TestMigrateConcurrent(t *testing.T) {
	pool := testPool(t, false)
	const workers = 8
	errCh := make(chan error, workers)
	var wg sync.WaitGroup
	for range workers {
		wg.Add(1)
		go func() { defer wg.Done(); errCh <- Migrate(context.Background(), pool) }()
	}
	wg.Wait()
	close(errCh)
	for err := range errCh {
		if err != nil {
			t.Fatal(err)
		}
	}
	var count int
	if err := pool.QueryRow(context.Background(), `SELECT count(*) FROM schema_migrations`).Scan(&count); err != nil {
		t.Fatal(err)
	}
	if count != 3 {
		t.Fatalf("適用件数: %d", count)
	}
}

func TestMigrateAddsKnownEpisodesToExistingDatabase(t *testing.T) {
	pool := testPool(t, false)
	ctx := context.Background()
	initial, err := migrations.ReadFile("migrations/001_initial.sql")
	if err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, string(initial), pgx.QueryExecModeSimpleProtocol); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `CREATE TABLE schema_migrations (version TEXT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now())`); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `INSERT INTO schema_migrations (version) VALUES ('migrations/001_initial.sql')`); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `INSERT INTO subscriptions (series_id, series_title) VALUES ('existing', '既存番組')`); err != nil {
		t.Fatal(err)
	}
	if err := Migrate(ctx, pool); err != nil {
		t.Fatal(err)
	}
	var count int
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM schema_migrations`).Scan(&count); err != nil || count != 3 {
		t.Fatalf("移行履歴: %d %v", count, err)
	}
	if _, err := pool.Exec(ctx, `INSERT INTO known_episodes (series_id, episode_id) VALUES ('existing', 'ep1')`); err != nil {
		t.Fatalf("既存番組への既知記録: %v", err)
	}
	if _, err := pool.Exec(ctx, `DELETE FROM subscriptions WHERE series_id='existing'`); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM known_episodes WHERE series_id='existing'`).Scan(&count); err != nil || count != 0 {
		t.Fatalf("アップグレード後の cascade: %d %v", count, err)
	}
}

func TestMigrateAddsSubscriptionCheckErrorsToVersionTwoDatabase(t *testing.T) {
	pool := testPool(t, false)
	ctx := context.Background()
	for _, name := range []string{"migrations/001_initial.sql", "migrations/002_known_episodes.sql"} {
		sql, err := migrations.ReadFile(name)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := pool.Exec(ctx, string(sql), pgx.QueryExecModeSimpleProtocol); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := pool.Exec(ctx, `CREATE TABLE schema_migrations (version TEXT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now())`); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `INSERT INTO schema_migrations (version) VALUES ('migrations/001_initial.sql'), ('migrations/002_known_episodes.sql')`); err != nil {
		t.Fatal(err)
	}
	checked := time.Date(2026, 10, 1, 2, 3, 4, 0, time.UTC)
	if _, err := pool.Exec(ctx, `INSERT INTO subscriptions (series_id, series_title, last_checked_at) VALUES ('existing-v2', '既存番組', $1)`, checked); err != nil {
		t.Fatal(err)
	}
	if err := Migrate(ctx, pool); err != nil {
		t.Fatal(err)
	}
	var title string
	var last time.Time
	var errorAt, errorMessage any
	if err := pool.QueryRow(ctx, `SELECT series_title, last_checked_at, last_check_error_at, last_check_error FROM subscriptions WHERE series_id='existing-v2'`).Scan(&title, &last, &errorAt, &errorMessage); err != nil || title != "既存番組" || !last.Equal(checked) || errorAt != nil || errorMessage != nil {
		t.Fatalf("v2データ保持と追加列: title=%q last=%v errorAt=%v error=%v err=%v", title, last, errorAt, errorMessage, err)
	}
	var indexDef string
	if err := pool.QueryRow(ctx, `SELECT indexdef FROM pg_indexes WHERE schemaname=current_schema() AND indexname='subscriptions_due_idx'`).Scan(&indexDef); err != nil || !strings.Contains(indexDef, "last_checked_at") || !strings.Contains(indexDef, "WHERE enabled") {
		t.Fatalf("期限抽出索引: %q %v", indexDef, err)
	}
	var versions int
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM schema_migrations`).Scan(&versions); err != nil || versions != 3 {
		t.Fatalf("移行履歴: %d %v", versions, err)
	}
}

func TestDownloadJobIndexes(t *testing.T) {
	pool := testPool(t, true)
	rows, err := pool.Query(context.Background(), `SELECT indexname, indexdef FROM pg_indexes WHERE schemaname=current_schema() AND tablename='download_jobs'`)
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()
	indexes := make(map[string]string)
	for rows.Next() {
		var name, definition string
		if err := rows.Scan(&name, &definition); err != nil {
			t.Fatal(err)
		}
		indexes[name] = definition
	}
	if err := rows.Err(); err != nil {
		t.Fatal(err)
	}
	checks := map[string][]string{
		"download_jobs_episode_id_key":    {"UNIQUE", "(episode_id)"},
		"download_jobs_claim_idx":         {"(created_at, id)", "status = 'queued'"},
		"download_jobs_status_newest_idx": {"(status, created_at DESC, id DESC)"},
		"download_jobs_stale_idx":         {"(heartbeat_at)", "status = 'running'"},
	}
	for name, parts := range checks {
		definition := indexes[name]
		for _, part := range parts {
			if !strings.Contains(definition, part) {
				t.Errorf("索引 %s に %q がない: %q", name, part, definition)
			}
		}
	}
}

func TestEnqueueAndListing(t *testing.T) {
	pool := testPool(t, true)
	s := New(pool)
	ctx := context.Background()
	first, created, err := s.Enqueue(ctx, sampleEpisode("ep1"), "manual")
	if err != nil || !created || first.Status != "queued" || first.Source != "manual" || first.DurationSeconds != 300 || first.EndAt == nil {
		t.Fatalf("enqueue: %+v %v %v", first, created, err)
	}
	again, created, err := s.Enqueue(ctx, sampleEpisode("ep1"), "subscription")
	if err != nil || created || again.ID != first.ID || again.Source != "manual" {
		t.Fatalf("重複: %+v %v %v", again, created, err)
	}
	second, _, err := s.Enqueue(ctx, sampleEpisode("ep2"), "subscription")
	if err != nil {
		t.Fatal(err)
	}
	jobs, err := s.ListJobs(ctx, "queued", 10)
	if err != nil || len(jobs) != 2 || jobs[0].ID != second.ID {
		t.Fatalf("list: %+v %v", jobs, err)
	}
	statuses, err := s.EpisodeStatuses(ctx, []string{"ep1", "ep2", "absent"})
	if err != nil || len(statuses) != 2 || statuses["ep1"] != "queued" {
		t.Fatalf("statuses: %+v %v", statuses, err)
	}
	if _, err := s.ListJobs(ctx, "bad", 10); err == nil {
		t.Fatal("不正な状態を許した")
	}
	if _, _, err := s.Enqueue(ctx, sampleEpisode("bad"), "invalid"); err == nil {
		t.Fatal("source CHECK を許した")
	}
	if _, err := pool.Exec(ctx, `UPDATE download_jobs SET phase='bad' WHERE id=$1`, first.ID); err == nil {
		t.Fatal("phase CHECK を許した")
	}
	if _, err := pool.Exec(ctx, `UPDATE download_jobs SET status='bad' WHERE id=$1`, first.ID); err == nil {
		t.Fatal("status CHECK を許した")
	}
	if _, err := pool.Exec(ctx, `UPDATE download_jobs SET error_kind='bad' WHERE id=$1`, first.ID); err == nil {
		t.Fatal("error CHECK を許した")
	}
}

func TestTransitions(t *testing.T) {
	pool := testPool(t, true)
	s := New(pool)
	ctx := context.Background()
	job, _, err := s.Enqueue(ctx, sampleEpisode("ep1"), "manual")
	if err != nil {
		t.Fatal(err)
	}
	if err := s.UpdateProgress(ctx, job.ID, Progress{Phase: "video"}); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("queued progress: %v", err)
	}
	if err := s.Complete(ctx, job.ID, "/file.mp4", 100); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("queued complete: %v", err)
	}
	if err := s.Fail(ctx, job.ID, "network", "failed"); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("queued fail: %v", err)
	}
	if err := s.Retry(ctx, job.ID); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("queued retry: %v", err)
	}
	claimed, err := s.ClaimNext(ctx)
	if err != nil || claimed.ID != job.ID || claimed.Status != "running" || claimed.Attempts != 1 || claimed.HeartbeatAt == nil || claimed.StartedAt == nil {
		t.Fatalf("claim: %+v %v", claimed, err)
	}
	oldHeartbeat := time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC)
	if _, err := pool.Exec(ctx, `UPDATE download_jobs SET heartbeat_at=$2 WHERE id=$1`, job.ID, oldHeartbeat); err != nil {
		t.Fatal(err)
	}
	bytes := int64(123)
	total := int64(1000)
	speed := 40.5
	eta := 12.0
	fragment := int32(2)
	fragments := int32(10)
	format := "video-1"
	if err := s.UpdateProgress(ctx, job.ID, Progress{Phase: "video", DownloadedBytes: &bytes,
		TotalBytes: &total, TotalIsEstimate: true, SpeedBPS: &speed, ETASeconds: &eta,
		FragmentIndex: &fragment, FragmentCount: &fragments, FormatID: &format}); err != nil {
		t.Fatal(err)
	}
	progressed, err := s.GetJob(ctx, job.ID)
	if err != nil || progressed.Phase == nil || *progressed.Phase != "video" || progressed.DownloadedBytes == nil || *progressed.DownloadedBytes != bytes || progressed.TotalBytes == nil || *progressed.TotalBytes != total || !progressed.TotalIsEstimate || progressed.SpeedBPS == nil || *progressed.SpeedBPS != speed || progressed.ETASeconds == nil || *progressed.ETASeconds != eta || progressed.FragmentIndex == nil || *progressed.FragmentIndex != fragment || progressed.FragmentCount == nil || *progressed.FragmentCount != fragments || progressed.FormatID == nil || *progressed.FormatID != format || progressed.HeartbeatAt == nil || !progressed.HeartbeatAt.After(oldHeartbeat) {
		t.Fatalf("progress: %+v %v", progressed, err)
	}
	if err := s.Fail(ctx, job.ID, "network", "connection lost"); err != nil {
		t.Fatal(err)
	}
	failed, err := s.GetJob(ctx, job.ID)
	if err != nil || failed.Status != "failed" || failed.ErrorKind == nil || *failed.ErrorKind != "network" || failed.ErrorMessage == nil || *failed.ErrorMessage != "connection lost" || failed.HeartbeatAt != nil || failed.FinishedAt == nil {
		t.Fatalf("fail: %+v %v", failed, err)
	}
	if err := s.UpdateProgress(ctx, job.ID, Progress{Phase: "audio"}); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("failed progress: %v", err)
	}
	if err := s.Retry(ctx, job.ID); err != nil {
		t.Fatal(err)
	}
	retried, err := s.GetJob(ctx, job.ID)
	if err != nil || retried.Status != "queued" || retried.Phase != nil || retried.ErrorKind != nil || retried.ErrorMessage != nil || retried.DownloadedBytes != nil || retried.TotalBytes != nil || retried.TotalIsEstimate || retried.SpeedBPS != nil || retried.ETASeconds != nil || retried.FragmentIndex != nil || retried.FragmentCount != nil || retried.FormatID != nil || retried.HeartbeatAt != nil || retried.StartedAt != nil || retried.FinishedAt != nil || retried.Attempts != 1 {
		t.Fatalf("retry: %+v %v", retried, err)
	}
	claimed, err = s.ClaimNext(ctx)
	if err != nil || claimed.Attempts != 2 {
		t.Fatalf("reclaim: %+v %v", claimed, err)
	}
	if err := s.Complete(ctx, job.ID, "/file.mp4", 100); err != nil {
		t.Fatal(err)
	}
	completed, err := s.GetJob(ctx, job.ID)
	if err != nil || completed.Status != "succeeded" || completed.Phase != nil || completed.HeartbeatAt != nil || completed.FinishedAt == nil || completed.FilePath == nil || *completed.FilePath != "/file.mp4" || completed.FileSizeBytes == nil || *completed.FileSizeBytes != 100 {
		t.Fatalf("complete: %+v %v", completed, err)
	}
	if err := s.Cancel(ctx, job.ID); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("succeeded cancel: %v", err)
	}
	if err := s.Retry(ctx, job.ID); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("succeeded retry: %v", err)
	}
	if _, err := s.ClaimNext(ctx); !errors.Is(err, pgx.ErrNoRows) {
		t.Fatalf("empty queue: %v", err)
	}
}

func TestCancelAndStaleBoundary(t *testing.T) {
	pool := testPool(t, true)
	s := New(pool)
	ctx := context.Background()
	queued, _, err := s.Enqueue(ctx, sampleEpisode("queued"), "manual")
	if err != nil {
		t.Fatal(err)
	}
	if err := s.Cancel(ctx, queued.ID); err != nil {
		t.Fatal(err)
	}
	canceled, err := s.GetJob(ctx, queued.ID)
	if err != nil || canceled.Status != "canceled" || canceled.ErrorKind == nil || *canceled.ErrorKind != "canceled" {
		t.Fatalf("cancel: %+v %v", canceled, err)
	}
	if err := s.Retry(ctx, queued.ID); err != nil {
		t.Fatal(err)
	}
	claimed, err := s.ClaimNext(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if err := s.Cancel(ctx, claimed.ID); err != nil {
		t.Fatal(err)
	}
	if err := s.Complete(ctx, claimed.ID, "/late.mp4", 1); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("canceled complete: %v", err)
	}
	if err := s.Retry(ctx, claimed.ID); err != nil {
		t.Fatal(err)
	}
	claimed, err = s.ClaimNext(ctx)
	if err != nil || claimed.Attempts != 2 {
		t.Fatalf("cancel retry claim: %+v %v", claimed, err)
	}
	cutoff := time.Date(2026, 10, 3, 0, 0, 0, 0, time.UTC)
	if _, err := pool.Exec(ctx, `UPDATE download_jobs SET heartbeat_at=$2 WHERE id=$1`, claimed.ID, cutoff); err != nil {
		t.Fatal(err)
	}
	fresh, _, err := s.Enqueue(ctx, sampleEpisode("fresh"), "manual")
	if err != nil {
		t.Fatal(err)
	}
	if _, err := s.ClaimNext(ctx); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `UPDATE download_jobs SET heartbeat_at=$2 WHERE id=$1`, fresh.ID, cutoff.Add(time.Hour)); err != nil {
		t.Fatal(err)
	}
	n, err := s.RequeueStale(ctx, cutoff)
	if err != nil || n != 0 {
		t.Fatalf("境界上: %d %v", n, err)
	}
	n, err = s.RequeueStale(ctx, cutoff.Add(time.Microsecond))
	if err != nil || n != 1 {
		t.Fatalf("境界後: %d %v", n, err)
	}
	job, err := s.GetJob(ctx, claimed.ID)
	if err != nil || job.Status != "queued" || job.HeartbeatAt != nil || job.Attempts != 2 {
		t.Fatalf("requeue: %+v %v", job, err)
	}
	stillRunning, err := s.GetJob(ctx, fresh.ID)
	if err != nil || stillRunning.Status != "running" || stillRunning.HeartbeatAt == nil || !stillRunning.HeartbeatAt.Equal(cutoff.Add(time.Hour)) {
		t.Fatalf("新しい heartbeat を持つジョブ: %+v %v", stillRunning, err)
	}
}

func TestHeartbeatAndGracefulRequeue(t *testing.T) {
	s := New(testPool(t, true))
	ctx := context.Background()
	job, _, err := s.Enqueue(ctx, sampleEpisode("heartbeat"), "manual")
	if err != nil {
		t.Fatal(err)
	}
	if err := s.Heartbeat(ctx, job.ID); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("待機中の heartbeat: %v", err)
	}
	claimed, err := s.ClaimNext(ctx)
	if err != nil {
		t.Fatal(err)
	}
	old := time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC)
	if _, err := s.pool.Exec(ctx, `UPDATE download_jobs SET heartbeat_at=$2 WHERE id=$1`, claimed.ID, old); err != nil {
		t.Fatal(err)
	}
	if err := s.Heartbeat(ctx, claimed.ID); err != nil {
		t.Fatal(err)
	}
	heartbeated, err := s.GetJob(ctx, claimed.ID)
	if err != nil || heartbeated.HeartbeatAt == nil || !heartbeated.HeartbeatAt.After(old) {
		t.Fatalf("heartbeat: %+v %v", heartbeated, err)
	}
	bytes := int64(10)
	if err := s.UpdateProgress(ctx, claimed.ID, Progress{Phase: "video", DownloadedBytes: &bytes}); err != nil {
		t.Fatal(err)
	}
	if err := s.RequeueRunning(ctx, claimed.ID); err != nil {
		t.Fatal(err)
	}
	requeued, err := s.GetJob(ctx, claimed.ID)
	if err != nil || requeued.Status != "queued" || requeued.StartedAt != nil || requeued.HeartbeatAt != nil || requeued.Phase != nil || requeued.DownloadedBytes != nil || requeued.Attempts != 1 {
		t.Fatalf("graceful requeue: %+v %v", requeued, err)
	}
	if err := s.RequeueRunning(ctx, claimed.ID); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("待機中の再復帰: %v", err)
	}
}

func TestRejectedTransitions(t *testing.T) {
	pool := testPool(t, true)
	s := New(pool)
	ctx := context.Background()
	operations := map[string]func(int64) error{
		"progress": func(id int64) error { return s.UpdateProgress(ctx, id, Progress{Phase: "video"}) },
		"complete": func(id int64) error { return s.Complete(ctx, id, "/file.mp4", 1) },
		"fail":     func(id int64) error { return s.Fail(ctx, id, "network", "error") },
		"cancel":   func(id int64) error { return s.Cancel(ctx, id) },
		"retry":    func(id int64) error { return s.Retry(ctx, id) },
	}
	for status, rejected := range map[string][]string{
		"queued":    {"progress", "complete", "fail", "retry"},
		"running":   {"retry"},
		"succeeded": {"progress", "complete", "fail", "cancel", "retry"},
		"failed":    {"progress", "complete", "fail", "cancel"},
		"canceled":  {"progress", "complete", "fail", "cancel"},
	} {
		t.Run(status, func(t *testing.T) {
			job, _, err := s.Enqueue(ctx, sampleEpisode("rejected-"+status), "manual")
			if err != nil {
				t.Fatal(err)
			}
			if _, err := pool.Exec(ctx, `UPDATE download_jobs SET status=$2 WHERE id=$1`, job.ID, status); err != nil {
				t.Fatal(err)
			}
			for _, name := range rejected {
				if err := operations[name](job.ID); !errors.Is(err, ErrInvalidTransition) {
					t.Errorf("%s on %s: %v", name, status, err)
				}
			}
			got, err := s.GetJob(ctx, job.ID)
			if err != nil || got.Status != status {
				t.Fatalf("拒否後の状態: %+v %v", got, err)
			}
		})
	}
}

func TestRetryEpisode(t *testing.T) {
	s := New(testPool(t, true))
	ctx := context.Background()
	job, _, err := s.Enqueue(ctx, sampleEpisode("retry-by-episode"), "manual")
	if err != nil {
		t.Fatal(err)
	}
	if err := s.RetryEpisode(ctx, job.EpisodeID); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("待機中の再試行: %v", err)
	}
	if err := s.Cancel(ctx, job.ID); err != nil {
		t.Fatal(err)
	}
	if err := s.RetryEpisode(ctx, job.EpisodeID); err != nil {
		t.Fatal(err)
	}
	got, err := s.GetJob(ctx, job.ID)
	if err != nil || got.Status != "queued" || got.ErrorKind != nil || got.FinishedAt != nil {
		t.Fatalf("エピソード ID による再試行: %+v %v", got, err)
	}
	claimed, err := s.ClaimNext(ctx)
	if err != nil {
		t.Fatal(err)
	}
	downloaded, total := int64(10), int64(20)
	format := "video-1"
	if err := s.UpdateProgress(ctx, claimed.ID, Progress{
		Phase: "video", DownloadedBytes: &downloaded, TotalBytes: &total,
		TotalIsEstimate: true, FormatID: &format,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.Fail(ctx, claimed.ID, "network", "通信失敗"); err != nil {
		t.Fatal(err)
	}
	if err := s.RetryEpisode(ctx, job.EpisodeID); err != nil {
		t.Fatalf("失敗からの再試行: %v", err)
	}
	got, err = s.GetJob(ctx, job.ID)
	if err != nil || got.Status != "queued" || got.Phase != nil || got.DownloadedBytes != nil || got.TotalBytes != nil || got.TotalIsEstimate || got.FormatID != nil || got.ErrorKind != nil || got.ErrorMessage != nil || got.HeartbeatAt != nil || got.StartedAt != nil || got.FinishedAt != nil || got.Attempts != 1 {
		t.Fatalf("失敗情報の消去: %+v %v", got, err)
	}
	if err := s.RetryEpisode(ctx, "unknown"); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("存在しないエピソード: %v", err)
	}
}

func TestCancelQueuedDoesNotCancelRunning(t *testing.T) {
	s := New(testPool(t, true))
	ctx := context.Background()
	queued, _, err := s.Enqueue(ctx, sampleEpisode("queue-cancel"), "manual")
	if err != nil {
		t.Fatal(err)
	}
	if err := s.CancelQueued(ctx, queued.ID); err != nil {
		t.Fatal(err)
	}
	canceled, err := s.GetJob(ctx, queued.ID)
	if err != nil || canceled.Status != "canceled" || canceled.ErrorKind == nil || *canceled.ErrorKind != "canceled" || canceled.FinishedAt == nil {
		t.Fatalf("待機中の取り消し: %+v %v", canceled, err)
	}
	if err := s.CancelQueued(ctx, queued.ID); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("再取り消し: %v", err)
	}
	if err := s.CancelQueued(ctx, -1); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("存在しないジョブの取り消し: %v", err)
	}
	if err := s.Retry(ctx, queued.ID); err != nil {
		t.Fatal(err)
	}
	claimed, err := s.ClaimNext(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if err := s.CancelQueued(ctx, claimed.ID); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("実行中の取り消し: %v", err)
	}
	job, err := s.GetJob(ctx, claimed.ID)
	if err != nil || job.Status != "running" {
		t.Fatalf("実行中の状態: %+v %v", job, err)
	}
}

func TestClaimNextConcurrent(t *testing.T) {
	s := New(testPool(t, true))
	ctx := context.Background()
	const jobs = 12
	for i := range jobs {
		if _, _, err := s.Enqueue(ctx, sampleEpisode(fmt.Sprintf("ep%d", i)), "manual"); err != nil {
			t.Fatal(err)
		}
	}
	results := make(chan int64, jobs)
	errCh := make(chan error, jobs)
	var wg sync.WaitGroup
	for range jobs {
		wg.Add(1)
		go func() {
			defer wg.Done()
			job, err := s.ClaimNext(ctx)
			if err != nil {
				errCh <- err
				return
			}
			results <- job.ID
		}()
	}
	wg.Wait()
	close(results)
	close(errCh)
	for err := range errCh {
		t.Error(err)
	}
	seen := make(map[int64]bool)
	for id := range results {
		if seen[id] {
			t.Errorf("二重 claim: %d", id)
		}
		seen[id] = true
	}
	if len(seen) != jobs {
		t.Fatalf("claim 数: %d", len(seen))
	}
}

func TestClaimNextSkipsLockedOldest(t *testing.T) {
	pool := testPool(t, true)
	s := New(pool)
	ctx := context.Background()
	oldest, _, err := s.Enqueue(ctx, sampleEpisode("oldest"), "manual")
	if err != nil {
		t.Fatal(err)
	}
	newer, _, err := s.Enqueue(ctx, sampleEpisode("newer"), "manual")
	if err != nil {
		t.Fatal(err)
	}
	base := time.Date(2026, 10, 3, 0, 0, 0, 0, time.UTC)
	if _, err := pool.Exec(ctx, `UPDATE download_jobs SET created_at=$2 WHERE id=$1`, oldest.ID, base); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `UPDATE download_jobs SET created_at=$2 WHERE id=$1`, newer.ID, base.Add(time.Hour)); err != nil {
		t.Fatal(err)
	}
	tx, err := pool.Begin(ctx)
	if err != nil {
		t.Fatal(err)
	}
	defer tx.Rollback(ctx)
	if _, err := tx.Exec(ctx, `SELECT id FROM download_jobs WHERE id=$1 FOR UPDATE`, oldest.ID); err != nil {
		t.Fatal(err)
	}
	claimCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()
	claimed, err := s.ClaimNext(claimCtx)
	if err != nil || claimed.ID != newer.ID {
		t.Fatalf("ロック中の最古ジョブを飛ばせない: %+v %v", claimed, err)
	}
	if err := tx.Rollback(ctx); err != nil {
		t.Fatal(err)
	}
	claimed, err = s.ClaimNext(ctx)
	if err != nil || claimed.ID != oldest.ID {
		t.Fatalf("ロック解放後の最古ジョブ: %+v %v", claimed, err)
	}
}

func TestListJobsFiltersAndLimit(t *testing.T) {
	pool := testPool(t, true)
	s := New(pool)
	ctx := context.Background()
	ids := make([]int64, 3)
	for i := range ids {
		job, _, err := s.Enqueue(ctx, sampleEpisode(fmt.Sprintf("list-%d", i)), "manual")
		if err != nil {
			t.Fatal(err)
		}
		ids[i] = job.ID
		created := time.Date(2026, 10, 3, i, 0, 0, 0, time.UTC)
		if _, err := pool.Exec(ctx, `UPDATE download_jobs SET created_at=$2 WHERE id=$1`, job.ID, created); err != nil {
			t.Fatal(err)
		}
	}
	if err := s.Cancel(ctx, ids[1]); err != nil {
		t.Fatal(err)
	}
	all, err := s.ListJobs(ctx, "", 2)
	if err != nil || len(all) != 2 || all[0].ID != ids[2] || all[1].ID != ids[1] {
		t.Fatalf("全件の新しい順と上限: %+v %v", all, err)
	}
	queued, err := s.ListJobs(ctx, "queued", 10)
	if err != nil || len(queued) != 2 || queued[0].ID != ids[2] || queued[1].ID != ids[0] {
		t.Fatalf("状態別一覧: %+v %v", queued, err)
	}
	canceled, err := s.ListJobs(ctx, "canceled", 10)
	if err != nil || len(canceled) != 1 || canceled[0].ID != ids[1] {
		t.Fatalf("中断済み一覧: %+v %v", canceled, err)
	}
	if _, err := s.ListJobs(ctx, "", 0); err == nil {
		t.Fatal("0 件の上限を許した")
	}
	statuses, err := s.EpisodeStatuses(ctx, []string{"list-0", "list-1", "absent"})
	if err != nil || len(statuses) != 2 || statuses["list-0"] != "queued" || statuses["list-1"] != "canceled" {
		t.Fatalf("状態の一括取得: %+v %v", statuses, err)
	}
}

func TestSubscriptions(t *testing.T) {
	s := New(testPool(t, true))
	ctx := context.Background()
	created, err := s.RegisterSubscription(ctx, "series", "番組", nil, false)
	if err != nil || !created {
		t.Fatalf("add: %v %v", created, err)
	}
	created, err = s.RegisterSubscription(ctx, "series", "別タイトル", nil, false)
	if err != nil || created {
		t.Fatalf("duplicate: %v %v", created, err)
	}
	when := time.Date(2026, 10, 3, 0, 0, 0, 0, time.UTC)
	if err := s.SetSubscriptionChecked(ctx, "series", when); err != nil {
		t.Fatal(err)
	}
	subs, err := s.ListSubscriptions(ctx)
	if err != nil || len(subs) != 1 || subs[0].SeriesTitle != "番組" || !subs[0].Enabled || subs[0].LastCheckedAt == nil || !subs[0].LastCheckedAt.Equal(when) {
		t.Fatalf("list: %+v %v", subs, err)
	}
	deleted, err := s.DeleteSubscription(ctx, "series")
	if err != nil || !deleted {
		t.Fatalf("delete: %v %v", deleted, err)
	}
	deleted, err = s.DeleteSubscription(ctx, "series")
	if err != nil || deleted {
		t.Fatalf("duplicate delete: %v %v", deleted, err)
	}
	if err := s.SetSubscriptionChecked(ctx, "series", when); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("missing: %v", err)
	}
}