DB の土台を追加（マイグレーション、待ち行列・番組登録の操作、embedded-postgres によるテスト）

Dockerfile

@@ -1,7 +1,8 @@
 FROM golang:1.27.1-alpine AS build
 
 WORKDIR /src
-COPY go.mod ./
+COPY go.mod go.sum ./
+RUN go mod download
 COPY cmd ./cmd
 COPY internal ./internal
 RUN CGO_ENABLED=0 go build -trimpath -ldflags="-s -w" -o /web ./cmd/web

PROJECT.md

@@ -3,7 +3,7 @@ title: tver の現状
 tags:
   - homelab
   - tver
-status: 新着一覧実装中
+status: DB の土台を実装中
 ---
 
 # tver の現状
@@ -13,8 +13,8 @@ status: 新着一覧実装中
 
 ## 今の状態
 
-- **新着一覧の画面を実装中。** Go の HTTP サーバー、TVer API クライアント、ドラマ・バラエティの新着表示、テスト、web コンテナの定義がある
-- DB、worker は未実装。ダウンロード済み・待ち行列中の区別は DB 導入後に実装する
+- Go の HTTP サーバー、TVer API クライアント、ドラマ・バラエティの新着表示、テスト、web コンテナの定義がある
+- **DB の土台を実装中。** SQL マイグレーションと待ち行列・番組登録の操作を追加した。web・worker への接続と画面は今後実装する
 - リポジトリ: `https://gitea.kazu634.com/kazu634/tver.git`
 
 ## 目的
@@ -33,7 +33,9 @@ TVer（`https://tver.jp/`）で公開中の動画情報を取得し、選んだ
 | ダウンロード | yt-dlp を使う |
 | デプロイ | Dockge で管理する複数コンテナ構成。dockage VM の `/opt/stacks/tver` |
 | 開発言語 | **Go**（`web`・`worker` とも）。yt-dlp は外部コマンドとして呼び、進捗・結果は yt-dlp の出力から読み取る |
-| DB | **NAS の共有 Postgres**。バックアップは規約どおり `pg_dump` → NAS |
+| DB | **NAS 上の tver 専用 Postgres 17 コンテナ**（Gitea と同じ方式）。バックアップは規約どおり `pg_dump` → NAS。公開ポートは未定（候補 15434、実機で確認する） |
+| DB ドライバ | `github.com/jackc/pgx/v5` の `pgxpool` |
+| DB テスト | `github.com/fergusstrange/embedded-postgres` v1.34.0 の Postgres 17 を起動する。Apple Silicon では Rosetta 2 が必要 |
 | 定期実行 | **`worker` 内のスケジューラ** |
 | NAS のマウント | **NFS**。DSM で `192.168.10.155` のみ許可する |
 | 保持期間 | **削除は手動のみ**。自動削除はしない |
@@ -76,7 +78,9 @@ TVer（`https://tver.jp/`）で公開中の動画情報を取得し、選んだ
 - 「予告を隠す」は既定でオンとし、180 秒以下の件を非表示にする。長さは秒から分へ切り上げて表示し、配信終了日時は Asia/Tokyo で表示する。ジャンルと予告表示の状態は URL に残す
 - 画面は `html/template` と htmx 2.0.11 を使い、htmx は Go の `embed` から `/static/` でローカル配信する
 - `compose.yaml` は web のみを定義し、dockage VM の LAN アドレス `192.168.10.155:8080` で公開する。待ち受けは `WEB_ADDR` で変更でき、既定は `:8080`
-- DB はコンテナを持たず、NAS の共有 Postgres に接続する（番組登録・ダウンロード履歴・待ち行列）
+- DB は NAS 上に tver 専用 Postgres 17 コンテナを置く。`internal/store` の埋め込み SQL を `schema_migrations` で管理し、`pg_advisory_lock` で複数プロセスからの同時適用を直列化する。起動処理への組み込みは今後行う
+- `download_jobs` は `episode_id` を一意にして二重登録を防ぎ、ジョブの状態・段階別進捗・結果を持つ。`subscriptions` はシリーズ単位の登録と最終確認日時を持つ。worker からの操作と画面用の一覧取得を `internal/store` にまとめる
+- worker は `yt-dlp[default]` と ffmpeg を外部コマンドとして使い、映像・音声・結合・移動の段階別進捗を扱う。正確な全体パーセントは保存しない。`after_move:filepath` で実パスを取り、ffprobe で H.264 と AAC の MP4 を確認して成功とする。失敗は DRM、地域制限、ネットワーク、配信不可、抽出失敗、出力不正、その他の順に分類する
 - 保存先の NAS 共有フォルダは dockage VM に NFS でマウントし、`worker`（書き込み）と `web`（読み取り・削除）にバインドマウントする
 
 ## 外部とのつながり
@@ -86,13 +90,17 @@ TVer（`https://tver.jp/`）で公開中の動画情報を取得し、選んだ
 | TVer `platform-api.tver.jp` | 新着一覧（`/service/api/v1/callNewerDetail/{drama\|variety}`）・番組のエピソード一覧 | **非公開 API**。セッションは `/v2/api/platform_users/browser/create` で発行する（yt-dlp と同じ方式） |
 | yt-dlp | ダウンロード | TVer extractor を使う |
 | NAS ds923p（NFS） | 動画の保存先 | DSM で NFS を有効化し、共有フォルダを `192.168.10.155` のみに許可する |
-| NAS ds923p（Postgres） | DB | DSM 内蔵の Postgres が `5432` を使うため、共有 Postgres コンテナの公開ポートを使う。DSM ファイアウォールで `192.168.10.155` のみ許可する |
+| NAS ds923p（Postgres 17） | DB | tver 専用コンテナを使う。DSM 内蔵の Postgres が `5432` を使うため別ポートで公開し、DSM ファイアウォールで `192.168.10.155` のみ許可する。公開ポートは実機で決める |
 
 ## 確認済みの事実
 
 - 新着 API は `drama` / `variety` それぞれ 100 件を返す。予告動画も含まれる。各件に `id`・`seriesID`・`seriesTitle`・`title`・`broadcasterName`・`duration`・`endAt`・`thumbnailPath` がある [実機] 2026-10-03 確認
 - dockage VM から TVer API に到達でき（HTTP 200）、送信元は日本と判定される [実機] 2026-10-03 確認
 - yt-dlp 2026.08.19 で TVer のエピソードを抽出できる（`--simulate`）。映像（H.264 / `avc1`）と音声が別ストリームで、結合に ffmpeg が要る。1080p・約 50 分で約 1.2GiB [実機] 2026-10-03 確認
+- 5 分の `epq432nee4` を yt-dlp 2026.08.19 で取得した。`--newline` と `--progress-template` の 1 行 1 JSON は Go の外部プロセスから逐次読める。ただし `uvx yt-dlp` のみで `pycryptodomex` が無いと AES-128 HLS を ffmpeg に委譲し、途中のダウンロード進捗が出なかった。`yt-dlp[default]` に含まれる `pycryptodomex` を加えた試行では途中の進捗が出た [実機] 2026-10-03 確認
+- 別ストリームの映像と音声は進捗が個別に始まり、`format_id` で区別できる。結合処理は `postprocess:` の出力で検知でき、`--print after_move:filepath` で実ファイルのパスを取得した。完成した MP4 は 24,613,787 バイトで、ffprobe では H.264 640×360 と AAC LC だった。初回取得は約 10 秒だった [実機] 2026-10-03 確認
+- SIGTERM で止めると `.part` と `.ytdl` が残り、再実行で 437,066 バイトから 438,090 バイトへ続きから取得できた [実機] 2026-10-03 確認
+- 到達不能のプロキシ経由の通信失敗と存在しないエピソードは、ともに yt-dlp の終了コード 1 だった。stderr の文言で分類が必要。DRM・地域制限の失敗文言と終了コードは実測していない（yt-dlp の extractor ソースのみ確認） [実機] 2026-10-03 確認
 - yt-dlp の TVer extractor は日本のみ（`_GEO_COUNTRIES = ['JP']`）。DRM 付きの作品は取得できない [公式] https://github.com/yt-dlp/yt-dlp/blob/master/yt_dlp/extractor/tver.py (取得日: 2026-10-03)
 - dockage VM に NAS のマウントは無い。ルートの空きは約 20GB [実機] 2026-10-03 確認
 
@@ -100,14 +108,15 @@ TVer（`https://tver.jp/`）で公開中の動画情報を取得し、選んだ
 
 - 新着 API が 100 件で打ち切りか、続きを取得できるか [要確認]
 - 新着のうち DRM 付きの作品の割合 [要確認]
-- 実ダウンロードの所要時間とファイルサイズ（`--simulate` の推定値のみ） [要確認]
 - 非公開 API と yt-dlp の TVer 対応は、TVer 側の変更で予告なく壊れうる。壊れたことを検知する手段が要る
-- NAS の準備（NFS 共有フォルダ、共有 Postgres へのデータベース・ユーザー作成、DSM ファイアウォール）は DSM での手作業になる。共有 Postgres の現行の公開ポートは [[homelab/docs/inventory|インフラ詳細]] に記載が無い [要確認]
-- Go から yt-dlp の進捗を読み取る方式（`--progress-template` と `--print` で 1 行 1 JSON を出す等）を実機で確かめる [要確認]
+- NAS の準備（NFS 共有フォルダ、tver 専用 Postgres 17 コンテナ、DSM ファイアウォール）は DSM での手作業になる。DB の公開ポート（候補 15434）は本番前に実機で確認する [要確認]
+- DRM・地域制限の yt-dlp 失敗時の文言と終了コードは未実測 [要確認]
 
 ## 確かめ方
 
-- `env GOCACHE=/private/tmp/tver-go-cache GOPROXY=off go test -count=1 ./...` で Go のテストを実行する。テストの TVer 応答は `internal/tver/testdata/newer_drama.json`（2026-10-03 の実応答を 3 件に縮小）と `httptest` を使い、テストから本物の TVer にはアクセスしない
+- Codex では `go test -count=1 ./...` を環境変数の前置なしで実行する。`go test` を許可する rules に一致させ、Postgres の共有メモリ作成に必要な権限で実行するため。初回のモジュール取得にはネットワーク接続が必要
+- DB テストは `embedded-postgres` v1.34.0 が Postgres 17 を一度起動し、テストごとに独立した DB を作る。バイナリキャッシュ、実行ファイル、データは `/private/tmp` 配下に置く。Apple Silicon での実行には Rosetta 2 の導入が必要
+- TVer 応答のテストは `internal/tver/testdata/newer_drama.json`（2026-10-03 の実応答を 3 件に縮小）と `httptest` を使い、テストから本物の TVer にはアクセスしない
 - 新着画面のテストは偽物の TVer クライアントと差し替え可能な時計を使い、表示項目、絞り込みの境界、10 分キャッシュ、取得失敗、JST 表示を確認する
-- `env GOCACHE=/private/tmp/tver-go-cache GOPROXY=off go vet ./...` で静的解析する
+- `env GOPATH=/private/tmp/tver-gopath GOCACHE=/private/tmp/tver-go-cache GOMODCACHE=/private/tmp/tver-go-modcache GOPROXY=off go vet ./...` で静的解析する
 - `WEB_ADDR=127.0.0.1:18080 go run ./cmd/web` で起動し、`/healthz`、`/`、`/newer?genre=variety` を確認できる。TVer への実アクセスを伴うため手動確認は少数回に留める

go.mod

@@ -1,3 +1,18 @@
 module gitea.kazu634.com/kazu634/tver
 
 go 1.27.1
+
+require (
+	github.com/fergusstrange/embedded-postgres v1.34.0
+	github.com/jackc/pgx/v5 v5.11.0
+)
+
+require (
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
+	github.com/lib/pq v1.10.9 // indirect
+	github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 // indirect
+	golang.org/x/sync v0.17.0 // indirect
+	golang.org/x/text v0.29.0 // indirect
+)

go.sum

@@ -0,0 +1,34 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/fergusstrange/embedded-postgres v1.34.0 h1:c6RKhPKFsLVU+Tdxsx8q0UxCHsvZZ/iShAnljRBXs6s=
+github.com/fergusstrange/embedded-postgres v1.34.0/go.mod h1:w0YvnCgf19o6tskInrOOACtnqfVlOvluz3hlNLY7tRk=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.11.0 h1:IzBBtyK9AHqf98cctWFifYSci2hgQR/cd56wB4p+ogg=
+github.com/jackc/pgx/v5 v5.11.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
+github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
+github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 h1:nIPpBwaJSVYIxUFsDv3M8ofmx9yWTog9BfvIu0q41lo=
+github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8/go.mod h1:HUYIGzjTL3rfEspMxjDjgmT5uz5wzYJKVo23qUhYTos=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug=
+golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
+golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk=
+golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

internal/store/migrate.go

@@ -0,0 +1,73 @@
+package store
+
+import (
+	"context"
+	"embed"
+	"fmt"
+	"io/fs"
+	"sort"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+//go:embed migrations/*.sql
+var migrations embed.FS
+
+const migrationLockID int64 = 0x747665725f646231
+
+// Migrate は同じ接続で排他ロックを保持し、未適用の SQL を順に実行する。
+func Migrate(ctx context.Context, pool *pgxpool.Pool) error {
+	conn, err := pool.Acquire(ctx)
+	if err != nil {
+		return fmt.Errorf("migration connection: %w", err)
+	}
+	defer conn.Release()
+	if _, err := conn.Exec(ctx, "SELECT pg_advisory_lock($1)", migrationLockID); err != nil {
+		return fmt.Errorf("migration lock: %w", err)
+	}
+	defer func() {
+		// 接続を返す前にロックを解放する。解放に失敗した接続は閉じる。
+		if _, err := conn.Exec(context.Background(), "SELECT pg_advisory_unlock($1)", migrationLockID); err != nil {
+			conn.Conn().Close(context.Background())
+		}
+	}()
+
+	if _, err := conn.Exec(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations (
+		version TEXT PRIMARY KEY,
+		applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
+	)`); err != nil {
+		return fmt.Errorf("migration metadata: %w", err)
+	}
+	files, err := fs.Glob(migrations, "migrations/*.sql")
+	if err != nil {
+		return err
+	}
+	sort.Strings(files)
+	for _, name := range files {
+		tx, err := conn.Begin(ctx)
+		if err != nil {
+			return err
+		}
+		var applied bool
+		err = tx.QueryRow(ctx, "SELECT EXISTS (SELECT 1 FROM schema_migrations WHERE version=$1)", name).Scan(&applied)
+		if err == nil && !applied {
+			var sql []byte
+			sql, err = migrations.ReadFile(name)
+			if err == nil {
+				_, err = tx.Exec(ctx, string(sql), pgx.QueryExecModeSimpleProtocol)
+			}
+			if err == nil {
+				_, err = tx.Exec(ctx, "INSERT INTO schema_migrations (version) VALUES ($1)", name)
+			}
+		}
+		if err != nil {
+			_ = tx.Rollback(ctx)
+			return fmt.Errorf("migration %s: %w", name, err)
+		}
+		if err := tx.Commit(ctx); err != nil {
+			return fmt.Errorf("migration %s commit: %w", name, err)
+		}
+	}
+	return nil
+}

internal/store/migrations/001_initial.sql

@@ -0,0 +1,44 @@
+CREATE TABLE download_jobs (
+    id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+    episode_id TEXT NOT NULL UNIQUE,
+    series_id TEXT NOT NULL,
+    series_title TEXT NOT NULL,
+    title TEXT NOT NULL,
+    broadcaster_name TEXT NOT NULL,
+    duration_seconds INTEGER NOT NULL CHECK (duration_seconds >= 0),
+    end_at TIMESTAMPTZ,
+    thumbnail_path TEXT NOT NULL,
+    source TEXT NOT NULL CHECK (source IN ('manual', 'subscription')),
+    status TEXT NOT NULL DEFAULT 'queued' CHECK (status IN ('queued', 'running', 'succeeded', 'failed', 'canceled')),
+    phase TEXT CHECK (phase IN ('video', 'audio', 'merging', 'moving')),
+    downloaded_bytes BIGINT CHECK (downloaded_bytes >= 0),
+    total_bytes BIGINT CHECK (total_bytes >= 0),
+    total_is_estimate BOOLEAN NOT NULL DEFAULT FALSE,
+    speed_bps DOUBLE PRECISION CHECK (speed_bps >= 0),
+    eta_seconds DOUBLE PRECISION CHECK (eta_seconds >= 0),
+    fragment_index INTEGER CHECK (fragment_index >= 0),
+    fragment_count INTEGER CHECK (fragment_count >= 0),
+    format_id TEXT,
+    error_kind TEXT CHECK (error_kind IN ('drm', 'geo', 'network', 'unavailable', 'extract', 'canceled', 'invalid_output', 'unknown')),
+    error_message TEXT,
+    file_path TEXT,
+    file_size_bytes BIGINT CHECK (file_size_bytes >= 0),
+    attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0),
+    heartbeat_at TIMESTAMPTZ,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    started_at TIMESTAMPTZ,
+    finished_at TIMESTAMPTZ,
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX download_jobs_claim_idx ON download_jobs (created_at, id) WHERE status = 'queued';
+CREATE INDEX download_jobs_status_newest_idx ON download_jobs (status, created_at DESC, id DESC);
+CREATE INDEX download_jobs_stale_idx ON download_jobs (heartbeat_at) WHERE status = 'running';
+
+CREATE TABLE subscriptions (
+    series_id TEXT PRIMARY KEY,
+    series_title TEXT NOT NULL,
+    enabled BOOLEAN NOT NULL DEFAULT TRUE,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    last_checked_at TIMESTAMPTZ
+);

internal/store/store.go

@@ -0,0 +1,263 @@
+package store
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+var ErrInvalidTransition = errors.New("invalid job status transition")
+
+type Store struct{ pool *pgxpool.Pool }
+
+func New(pool *pgxpool.Pool) *Store { return &Store{pool: pool} }
+
+type Episode struct {
+	ID, SeriesID, SeriesTitle, Title, BroadcasterName, ThumbnailPath string
+	DurationSeconds                                                  int
+	EndAt                                                            *time.Time
+}
+
+type Job struct {
+	ID                                 int64
+	EpisodeID, SeriesID, SeriesTitle   string
+	Title, BroadcasterName             string
+	DurationSeconds                    int
+	EndAt                              *time.Time
+	ThumbnailPath, Source, Status      string
+	Phase, FormatID                    *string
+	DownloadedBytes, TotalBytes        *int64
+	TotalIsEstimate                    bool
+	SpeedBPS, ETASeconds               *float64
+	FragmentIndex, FragmentCount       *int32
+	ErrorKind, ErrorMessage, FilePath  *string
+	FileSizeBytes                      *int64
+	Attempts                           int
+	HeartbeatAt, StartedAt, FinishedAt *time.Time
+	CreatedAt, UpdatedAt               time.Time
+}
+
+type Progress struct {
+	Phase           string
+	DownloadedBytes *int64
+	TotalBytes      *int64
+	TotalIsEstimate bool
+	SpeedBPS        *float64
+	ETASeconds      *float64
+	FragmentIndex   *int32
+	FragmentCount   *int32
+	FormatID        *string
+}
+
+type Subscription struct {
+	SeriesID, SeriesTitle string
+	Enabled               bool
+	CreatedAt             time.Time
+	LastCheckedAt         *time.Time
+}
+
+const jobColumns = `id, episode_id, series_id, series_title, title, broadcaster_name,
+	duration_seconds, end_at, thumbnail_path, source, status, phase,
+	downloaded_bytes, total_bytes, total_is_estimate, speed_bps, eta_seconds,
+	fragment_index, fragment_count, format_id, error_kind, error_message,
+	file_path, file_size_bytes, attempts, heartbeat_at, created_at,
+	started_at, finished_at, updated_at`
+
+var claimedJobColumns = func() string {
+	columns := strings.Split(jobColumns, ",")
+	for i, column := range columns {
+		columns[i] = "j." + strings.TrimSpace(column)
+	}
+	return strings.Join(columns, ", ")
+}()
+
+func scanJob(row pgx.Row) (Job, error) {
+	var j Job
+	err := row.Scan(&j.ID, &j.EpisodeID, &j.SeriesID, &j.SeriesTitle, &j.Title,
+		&j.BroadcasterName, &j.DurationSeconds, &j.EndAt, &j.ThumbnailPath,
+		&j.Source, &j.Status, &j.Phase, &j.DownloadedBytes, &j.TotalBytes,
+		&j.TotalIsEstimate, &j.SpeedBPS, &j.ETASeconds, &j.FragmentIndex,
+		&j.FragmentCount, &j.FormatID, &j.ErrorKind, &j.ErrorMessage,
+		&j.FilePath, &j.FileSizeBytes, &j.Attempts, &j.HeartbeatAt,
+		&j.CreatedAt, &j.StartedAt, &j.FinishedAt, &j.UpdatedAt)
+	return j, err
+}
+
+// Enqueue は同じエピソードの既存ジョブを再利用する。
+func (s *Store) Enqueue(ctx context.Context, ep Episode, source string) (Job, bool, error) {
+	row := s.pool.QueryRow(ctx, `INSERT INTO download_jobs
+		(episode_id, series_id, series_title, title, broadcaster_name, duration_seconds, end_at, thumbnail_path, source)
+		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9) ON CONFLICT (episode_id) DO NOTHING RETURNING `+jobColumns,
+		ep.ID, ep.SeriesID, ep.SeriesTitle, ep.Title, ep.BroadcasterName,
+		ep.DurationSeconds, ep.EndAt, ep.ThumbnailPath, source)
+	job, err := scanJob(row)
+	if err == nil {
+		return job, true, nil
+	}
+	if !errors.Is(err, pgx.ErrNoRows) {
+		return Job{}, false, err
+	}
+	job, err = scanJob(s.pool.QueryRow(ctx, `SELECT `+jobColumns+` FROM download_jobs WHERE episode_id=$1`, ep.ID))
+	return job, false, err
+}
+
+func (s *Store) GetJob(ctx context.Context, id int64) (Job, error) {
+	return scanJob(s.pool.QueryRow(ctx, `SELECT `+jobColumns+` FROM download_jobs WHERE id=$1`, id))
+}
+
+// Retry は失敗・中断されたジョブの実行時情報を消して再投入する。
+func (s *Store) Retry(ctx context.Context, id int64) error {
+	tag, err := s.pool.Exec(ctx, `UPDATE download_jobs SET status='queued', phase=NULL,
+		downloaded_bytes=NULL, total_bytes=NULL, total_is_estimate=false, speed_bps=NULL,
+		eta_seconds=NULL, fragment_index=NULL, fragment_count=NULL, format_id=NULL,
+		error_kind=NULL, error_message=NULL, heartbeat_at=NULL, started_at=NULL,
+		finished_at=NULL, updated_at=now()
+		WHERE id=$1 AND status IN ('failed','canceled')`, id)
+	return transitionResult(tag.RowsAffected(), err)
+}
+
+// ClaimNext は最も古い未実行ジョブを同時 worker と重複せず取得する。
+func (s *Store) ClaimNext(ctx context.Context) (Job, error) {
+	return scanJob(s.pool.QueryRow(ctx, `WITH next AS (
+		SELECT id FROM download_jobs WHERE status='queued'
+		ORDER BY created_at, id LIMIT 1 FOR UPDATE SKIP LOCKED
+	) UPDATE download_jobs AS j SET status='running', attempts=j.attempts+1,
+		started_at=now(), heartbeat_at=now(), updated_at=now()
+		FROM next WHERE j.id=next.id RETURNING `+claimedJobColumns))
+}
+
+func (s *Store) UpdateProgress(ctx context.Context, id int64, p Progress) error {
+	tag, err := s.pool.Exec(ctx, `UPDATE download_jobs SET phase=$2, downloaded_bytes=$3,
+		total_bytes=$4, total_is_estimate=$5, speed_bps=$6, eta_seconds=$7,
+		fragment_index=$8, fragment_count=$9, format_id=$10,
+		heartbeat_at=now(), updated_at=now() WHERE id=$1 AND status='running'`,
+		id, p.Phase, p.DownloadedBytes, p.TotalBytes, p.TotalIsEstimate,
+		p.SpeedBPS, p.ETASeconds, p.FragmentIndex, p.FragmentCount, p.FormatID)
+	return transitionResult(tag.RowsAffected(), err)
+}
+
+func (s *Store) Complete(ctx context.Context, id int64, filePath string, fileSizeBytes int64) error {
+	tag, err := s.pool.Exec(ctx, `UPDATE download_jobs SET status='succeeded', phase=NULL,
+		file_path=$2, file_size_bytes=$3, heartbeat_at=NULL, finished_at=now(), updated_at=now()
+		WHERE id=$1 AND status='running'`, id, filePath, fileSizeBytes)
+	return transitionResult(tag.RowsAffected(), err)
+}
+
+func (s *Store) Fail(ctx context.Context, id int64, kind, message string) error {
+	tag, err := s.pool.Exec(ctx, `UPDATE download_jobs SET status='failed', phase=NULL,
+		error_kind=$2, error_message=$3, heartbeat_at=NULL, finished_at=now(), updated_at=now()
+		WHERE id=$1 AND status='running'`, id, kind, message)
+	return transitionResult(tag.RowsAffected(), err)
+}
+
+func (s *Store) Cancel(ctx context.Context, id int64) error {
+	tag, err := s.pool.Exec(ctx, `UPDATE download_jobs SET status='canceled', phase=NULL,
+		error_kind='canceled', heartbeat_at=NULL, finished_at=now(), updated_at=now()
+		WHERE id=$1 AND status IN ('queued','running')`, id)
+	return transitionResult(tag.RowsAffected(), err)
+}
+
+// RequeueStale は cutoff より前に heartbeat が止まった実行中ジョブを戻す。
+func (s *Store) RequeueStale(ctx context.Context, cutoff time.Time) (int64, error) {
+	tag, err := s.pool.Exec(ctx, `UPDATE download_jobs SET status='queued', phase=NULL,
+		downloaded_bytes=NULL, total_bytes=NULL, total_is_estimate=false, speed_bps=NULL,
+		eta_seconds=NULL, fragment_index=NULL, fragment_count=NULL, format_id=NULL,
+		heartbeat_at=NULL, started_at=NULL, updated_at=now()
+		WHERE status='running' AND heartbeat_at < $1`, cutoff)
+	return tag.RowsAffected(), err
+}
+
+func transitionResult(count int64, err error) error {
+	if err != nil {
+		return err
+	}
+	if count == 0 {
+		return ErrInvalidTransition
+	}
+	return nil
+}
+
+// ListJobs は指定状態のジョブを新しい順に返す。空の状態は全件を意味する。
+func (s *Store) ListJobs(ctx context.Context, status string, limit int) ([]Job, error) {
+	if status != "" && status != "queued" && status != "running" && status != "succeeded" && status != "failed" && status != "canceled" {
+		return nil, fmt.Errorf("invalid status: %s", status)
+	}
+	if limit <= 0 {
+		return nil, errors.New("limit must be positive")
+	}
+	rows, err := s.pool.Query(ctx, `SELECT `+jobColumns+` FROM download_jobs
+		WHERE ($1='' OR status=$1) ORDER BY created_at DESC, id DESC LIMIT $2`, status, limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var jobs []Job
+	for rows.Next() {
+		job, err := scanJob(rows)
+		if err != nil {
+			return nil, err
+		}
+		jobs = append(jobs, job)
+	}
+	return jobs, rows.Err()
+}
+
+func (s *Store) EpisodeStatuses(ctx context.Context, ids []string) (map[string]string, error) {
+	result := make(map[string]string)
+	if len(ids) == 0 {
+		return result, nil
+	}
+	rows, err := s.pool.Query(ctx, `SELECT episode_id, status FROM download_jobs WHERE episode_id = ANY($1)`, ids)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var id, status string
+		if err := rows.Scan(&id, &status); err != nil {
+			return nil, err
+		}
+		result[id] = status
+	}
+	return result, rows.Err()
+}
+
+// AddSubscription は登録済みシリーズのタイトルと有効状態を変更しない。
+func (s *Store) AddSubscription(ctx context.Context, seriesID, title string) (bool, error) {
+	tag, err := s.pool.Exec(ctx, `INSERT INTO subscriptions (series_id, series_title)
+		VALUES ($1,$2) ON CONFLICT (series_id) DO NOTHING`, seriesID, title)
+	return tag.RowsAffected() == 1, err
+}
+
+func (s *Store) DeleteSubscription(ctx context.Context, seriesID string) (bool, error) {
+	tag, err := s.pool.Exec(ctx, `DELETE FROM subscriptions WHERE series_id=$1`, seriesID)
+	return tag.RowsAffected() == 1, err
+}
+
+func (s *Store) ListSubscriptions(ctx context.Context) ([]Subscription, error) {
+	rows, err := s.pool.Query(ctx, `SELECT series_id, series_title, enabled, created_at, last_checked_at
+		FROM subscriptions ORDER BY created_at DESC, series_id`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var subs []Subscription
+	for rows.Next() {
+		var sub Subscription
+		if err := rows.Scan(&sub.SeriesID, &sub.SeriesTitle, &sub.Enabled, &sub.CreatedAt, &sub.LastCheckedAt); err != nil {
+			return nil, err
+		}
+		subs = append(subs, sub)
+	}
+	return subs, rows.Err()
+}
+
+func (s *Store) SetSubscriptionChecked(ctx context.Context, seriesID string, checkedAt time.Time) error {
+	tag, err := s.pool.Exec(ctx, `UPDATE subscriptions SET last_checked_at=$2 WHERE series_id=$1`, seriesID, checkedAt)
+	return transitionResult(tag.RowsAffected(), err)
+}

internal/store/store_test.go

@@ -0,0 +1,564 @@
+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 != 1 || !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 != 1 {
+		t.Fatalf("適用件数: %d", count)
+	}
+}
+
+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 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 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.AddSubscription(ctx, "series", "番組")
+	if err != nil || !created {
+		t.Fatalf("add: %v %v", created, err)
+	}
+	created, err = s.AddSubscription(ctx, "series", "別タイトル")
+	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)
+	}
+}