diff --git a/.dockerignore b/.dockerignore index 4e0696c..03c8a24 100644 --- a/.dockerignore +++ b/.dockerignore @@ -2,3 +2,5 @@ .gitignore PROJECT.md AGENTS.md +.env +.env.* diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..ce81892 --- /dev/null +++ b/.env.example @@ -0,0 +1 @@ +DATABASE_URL= diff --git a/.gitignore b/.gitignore index 03fabd9..d2629fd 100644 --- a/.gitignore +++ b/.gitignore @@ -4,6 +4,7 @@ # 秘匿情報(homelab 規約: .env に分離し、コミットしない) .env .env.* +!.env.example # エージェントのローカル設定(端末ごとの権限・agmsg のフック) .claude/settings.local.json diff --git a/PROJECT.md b/PROJECT.md index 3023a08..2282109 100644 --- a/PROJECT.md +++ b/PROJECT.md @@ -3,7 +3,7 @@ title: tver の現状 tags: - homelab - tver -status: DB の土台を実装中 +status: 待ち行列の画面を実装中 --- # tver の現状 @@ -14,7 +14,7 @@ status: DB の土台を実装中 ## 今の状態 - Go の HTTP サーバー、TVer API クライアント、ドラマ・バラエティの新着表示、テスト、web コンテナの定義がある -- **DB の土台を実装中。** SQL マイグレーションと待ち行列・番組登録の操作を追加した。web・worker への接続と画面は今後実装する +- **待ち行列の画面を実装中。** web は起動時に DB へ接続してマイグレーションを適用し、新着からの登録・状態表示と待ち行列ページを提供する。worker がまだ無いため、登録したジョブは待機中のまま残る - リポジトリ: `https://gitea.kazu634.com/kazu634/tver.git` ## 目的 @@ -73,12 +73,14 @@ TVer(`https://tver.jp/`)で公開中の動画情報を取得し、選んだ | `web` | UI・API。新着一覧の取得、待ち行列への登録、番組登録、動画の配信(再生) | | `worker` | Go + yt-dlp + ffmpeg。待ち行列からダウンロードし、登録番組の新しい回を定期的に確認する | -- 現在の `web` は `/healthz`、ドラマ・バラエティの新着ページ(`/`)、新着一覧の HTML 断片(`/newer`)を提供する。「待ち行列/登録番組/ライブラリ」のページは空のまま +- 現在の `web` は `/healthz`、ドラマ・バラエティの新着ページ(`/`)、新着一覧の HTML 断片(`/newer`)、待ち行列ページ(`/queue`)と更新用断片(`/queue/list`)を提供する。「登録番組/ライブラリ」のページは空のまま +- 新着の登録ボタンはエピソード ID とジャンルだけを送る。web が有効な新着キャッシュから番組情報を引き、`episode_id` の一意制約で重複登録を防ぐ。新着一覧はジョブ状態を一括取得し、失敗・取り消しには再試行を表示する +- 待ち行列は実行中・待機中・失敗・取り消し・完了を状態別に直近50件ずつ表示する。段階別の進捗と失敗理由を表示し、待機中の取り消しと失敗・取り消しの再試行ができる。実行中または待機中のジョブがある間だけ5秒ごとに一覧を更新する - 新着はジャンルごとにメモリへ 10 分間キャッシュする。キャッシュ中は TVer セッションの再発行と一覧の再取得を行わない。TVer の取得に失敗した場合、一覧の場所にエラーを表示して詳細をログへ記録する - 「予告を隠す」は既定でオンとし、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 上に tver 専用 Postgres 17 コンテナを置く。`internal/store` の埋め込み SQL を `schema_migrations` で管理し、`pg_advisory_lock` で複数プロセスからの同時適用を直列化する。起動処理への組み込みは今後行う +- `compose.yaml` は web のみを定義し、dockage VM の LAN アドレス `192.168.10.155:8080` で公開する。待ち受けは `WEB_ADDR` で変更でき、既定は `:8080`。DB 接続先は必須の `DATABASE_URL` を `.env` から渡す(値の無い雛形は `.env.example`) +- DB は NAS 上に tver 専用 Postgres 17 コンテナを置く。`internal/store` の埋め込み SQL を `schema_migrations` で管理し、`pg_advisory_lock` で複数プロセスからの同時適用を直列化する。web は起動時に `pgxpool` で接続してマイグレーションを適用する - `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`(読み取り・削除)にバインドマウントする @@ -115,8 +117,9 @@ TVer(`https://tver.jp/`)で公開中の動画情報を取得し、選んだ ## 確かめ方 - Codex では `go test -count=1 ./...` を環境変数の前置なしで実行する。`go test` を許可する rules に一致させ、Postgres の共有メモリ作成に必要な権限で実行するため。初回のモジュール取得にはネットワーク接続が必要 +- web を起動するには `.env.example` を参考に、Git 管理外の `.env` に実際の `DATABASE_URL` を設定する。未設定なら起動時にエラー終了する。DB 接続がある環境では起動時にマイグレーションが適用され、`/queue` で待ち行列を確認できる - 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 表示を確認する +- 新着画面のテストは偽物の TVer クライアント・store と差し替え可能な時計を使い、表示項目、絞り込みの境界、10 分キャッシュ、取得失敗、JST 表示、登録・状態表示・待ち行列の操作を確認する - `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 への実アクセスを伴うため手動確認は少数回に留める +- `DATABASE_URL` と `WEB_ADDR=127.0.0.1:18080` を設定して `go run ./cmd/web` で起動し、`/healthz`、`/`、`/newer?genre=variety`、`/queue` を確認できる。TVer への実アクセスを伴うため手動確認は少数回に留める diff --git a/cmd/web/main.go b/cmd/web/main.go index a02fed3..7297bee 100644 --- a/cmd/web/main.go +++ b/cmd/web/main.go @@ -1,17 +1,28 @@ package main import ( + "context" + "errors" + "fmt" "log" "net/http" "os" + "strings" "time" _ "time/tzdata" + "gitea.kazu634.com/kazu634/tver/internal/store" "gitea.kazu634.com/kazu634/tver/internal/tver" "gitea.kazu634.com/kazu634/tver/internal/web" + "github.com/jackc/pgx/v5/pgxpool" ) func main() { + pool, err := openDatabase(context.Background(), os.Getenv("DATABASE_URL"), pgxpool.New, store.Migrate) + if err != nil { + log.Fatal(err) + } + defer pool.Close() address := os.Getenv("WEB_ADDR") if address == "" { address = ":8080" @@ -22,9 +33,26 @@ func main() { } server := &http.Server{ Addr: address, - Handler: web.NewHandler(client, time.Now, log.Default()), + Handler: web.NewHandler(client, store.New(pool), time.Now, log.Default()), ReadHeaderTimeout: 5 * time.Second, } log.Printf("web listening on %s", address) log.Fatal(server.ListenAndServe()) } + +func openDatabase(ctx context.Context, databaseURL string, + connect func(context.Context, string) (*pgxpool.Pool, error), + migrate func(context.Context, *pgxpool.Pool) error) (*pgxpool.Pool, error) { + if strings.TrimSpace(databaseURL) == "" { + return nil, errors.New("DATABASE_URL が設定されていません") + } + pool, err := connect(ctx, databaseURL) + if err != nil { + return nil, errors.New("DATABASE_URL から接続プールを作成できません") + } + if err := migrate(ctx, pool); err != nil { + pool.Close() + return nil, fmt.Errorf("DB マイグレーションに失敗: %w", err) + } + return pool, nil +} diff --git a/cmd/web/main_test.go b/cmd/web/main_test.go new file mode 100644 index 0000000..af02705 --- /dev/null +++ b/cmd/web/main_test.go @@ -0,0 +1,72 @@ +package main + +import ( + "context" + "errors" + "strings" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" +) + +func TestOpenDatabaseRequiresURLAndRunsMigration(t *testing.T) { + ctx := context.Background() + connectCalls, migrateCalls := 0, 0 + connect := func(_ context.Context, url string) (*pgxpool.Pool, error) { + connectCalls++ + if url != "postgres://test:secret@localhost/test" { + t.Errorf("URL = %q", url) + } + return pgxpool.New(ctx, url) + } + migrate := func(_ context.Context, _ *pgxpool.Pool) error { + migrateCalls++ + return nil + } + if _, err := openDatabase(ctx, " ", connect, migrate); err == nil || !strings.Contains(err.Error(), "DATABASE_URL") { + t.Fatalf("未設定 URL のエラー: %v", err) + } + if connectCalls != 0 || migrateCalls != 0 { + t.Fatalf("未設定でも接続・移行された: %d %d", connectCalls, migrateCalls) + } + pool, err := openDatabase(ctx, "postgres://test:secret@localhost/test", connect, migrate) + if err != nil || pool == nil { + t.Fatalf("DB 初期化: %v", err) + } + pool.Close() + if connectCalls != 1 || migrateCalls != 1 { + t.Fatalf("接続・移行回数: %d %d", connectCalls, migrateCalls) + } +} + +func TestOpenDatabaseReportsMigrationFailure(t *testing.T) { + ctx := context.Background() + pool, err := openDatabase(ctx, "postgres://test:secret@localhost/test", pgxpool.New, + func(context.Context, *pgxpool.Pool) error { return errors.New("migration failed") }) + if pool != nil || err == nil || !strings.Contains(err.Error(), "migration failed") { + t.Fatalf("移行失敗: pool=%v err=%v", pool, err) + } +} + +func TestOpenDatabaseReportsConnectionFailureWithoutLeakingURL(t *testing.T) { + migrateCalled := false + pool, err := openDatabase(context.Background(), "postgres://user:secret@localhost/test", + func(context.Context, string) (*pgxpool.Pool, error) { + return nil, errors.New("postgres://user:secret@localhost/test: connection refused") + }, + func(context.Context, *pgxpool.Pool) error { + migrateCalled = true + return nil + }) + if pool != nil || err == nil || strings.Contains(err.Error(), "secret") || migrateCalled { + t.Fatalf("接続失敗: pool=%v err=%v migrateCalled=%v", pool, err, migrateCalled) + } +} + +func TestOpenDatabaseDoesNotExposeInvalidURL(t *testing.T) { + _, err := openDatabase(context.Background(), "postgres://user:secret@%", pgxpool.New, + func(context.Context, *pgxpool.Pool) error { t.Fatal("移行してはいけない"); return nil }) + if err == nil || strings.Contains(err.Error(), "secret") { + t.Fatalf("接続 URL の秘匿: %v", err) + } +} diff --git a/compose.yaml b/compose.yaml index f00af23..d8255ae 100644 --- a/compose.yaml +++ b/compose.yaml @@ -1,6 +1,7 @@ services: web: build: . + env_file: .env environment: TZ: Asia/Tokyo WEB_ADDR: ":8080" diff --git a/internal/store/store.go b/internal/store/store.go index 9189849..1037732 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -121,6 +121,17 @@ func (s *Store) Retry(ctx context.Context, id int64) error { return transitionResult(tag.RowsAffected(), err) } +// RetryEpisode はエピソード ID で失敗・中断されたジョブを再投入する。 +func (s *Store) RetryEpisode(ctx context.Context, episodeID string) 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 episode_id=$1 AND status IN ('failed','canceled')`, episodeID) + 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 ( @@ -162,6 +173,14 @@ func (s *Store) Cancel(ctx context.Context, id int64) error { return transitionResult(tag.RowsAffected(), err) } +// CancelQueued は worker 未着手のジョブだけを取り消す。 +func (s *Store) CancelQueued(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='queued'`, 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, diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 3d737dc..10a6522 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -407,6 +407,89 @@ func TestRejectedTransitions(t *testing.T) { } } +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() diff --git a/internal/web/actions.go b/internal/web/actions.go new file mode 100644 index 0000000..d13401f --- /dev/null +++ b/internal/web/actions.go @@ -0,0 +1,90 @@ +package web + +import ( + "errors" + "net/http" + + "gitea.kazu634.com/kazu634/tver/internal/store" +) + +type episodeActionView struct { + ID, Genre, Status, StatusText string +} + +func statusText(status string) string { + switch status { + case "queued": + return "待機中" + case "running": + return "ダウンロード中" + case "succeeded": + return "完了" + case "failed": + return "失敗" + case "canceled": + return "取り消し" + default: + return "" + } +} + +func (s *server) enqueue(w http.ResponseWriter, r *http.Request) { + genre, id, ok := episodeForm(w, r) + if !ok { + return + } + ep, found := s.cachedEpisode(genre, id) + if !found { + http.Error(w, "新着キャッシュにエピソードがありません。新着を再表示してください", http.StatusNotFound) + return + } + job, _, err := s.jobs.Enqueue(r.Context(), storeEpisode(ep), "manual") + if err != nil { + s.logger.Printf("待ち行列への登録に失敗: %v", err) + http.Error(w, "待ち行列に登録できませんでした", http.StatusInternalServerError) + return + } + s.episodeActionResponse(w, r, episodeActionView{ID: id, Genre: genre, Status: job.Status, StatusText: statusText(job.Status)}) +} + +func (s *server) retryEpisode(w http.ResponseWriter, r *http.Request) { + genre, id, ok := episodeForm(w, r) + if !ok { + return + } + if _, found := s.cachedEpisode(genre, id); !found { + http.Error(w, "新着キャッシュにエピソードがありません。新着を再表示してください", http.StatusNotFound) + return + } + if err := s.jobs.RetryEpisode(r.Context(), id); err != nil { + if errors.Is(err, store.ErrInvalidTransition) { + http.Error(w, "再試行できる状態ではありません", http.StatusConflict) + } else { + s.logger.Printf("ジョブの再試行に失敗: %v", err) + http.Error(w, "再試行できませんでした", http.StatusInternalServerError) + } + return + } + s.episodeActionResponse(w, r, episodeActionView{ID: id, Genre: genre, Status: "queued", StatusText: statusText("queued")}) +} + +func episodeForm(w http.ResponseWriter, r *http.Request) (string, string, bool) { + if err := r.ParseForm(); err != nil { + http.Error(w, "フォームを読み取れませんでした", http.StatusBadRequest) + return "", "", false + } + genre, id := r.Form.Get("genre"), r.Form.Get("episode_id") + if (genre != "drama" && genre != "variety") || id == "" { + http.Error(w, "ジャンルまたはエピソード ID が不正です", http.StatusBadRequest) + return "", "", false + } + return genre, id, true +} + +func (s *server) episodeActionResponse(w http.ResponseWriter, r *http.Request, view episodeActionView) { + if r.Header.Get("HX-Request") != "true" { + http.Redirect(w, r, "/?genre="+view.Genre, http.StatusSeeOther) + return + } + s.render(w, "episode_action", view) +} diff --git a/internal/web/handler.go b/internal/web/handler.go index 3fd1abe..e5f4070 100644 --- a/internal/web/handler.go +++ b/internal/web/handler.go @@ -1,6 +1,7 @@ package web import ( + "context" "embed" "html/template" "io/fs" @@ -8,6 +9,8 @@ import ( "net/http" "sync" "time" + + "gitea.kazu634.com/kazu634/tver/internal/store" ) //go:embed templates/*.html static/* @@ -16,10 +19,22 @@ var assets embed.FS type page struct { Title string Newer *newerView + Queue *queueView +} + +type JobStore interface { + Enqueue(context.Context, store.Episode, string) (store.Job, bool, error) + EpisodeStatuses(context.Context, []string) (map[string]string, error) + ListJobs(context.Context, string, int) ([]store.Job, error) + GetJob(context.Context, int64) (store.Job, error) + Retry(context.Context, int64) error + RetryEpisode(context.Context, string) error + CancelQueued(context.Context, int64) error } type server struct { fetcher NewerFetcher + jobs JobStore now func() time.Time logger *log.Logger zone *time.Location @@ -28,7 +43,7 @@ type server struct { cache map[string]cacheEntry } -func NewHandler(fetcher NewerFetcher, now func() time.Time, logger *log.Logger) http.Handler { +func NewHandler(fetcher NewerFetcher, jobs JobStore, now func() time.Time, logger *log.Logger) http.Handler { tmpl := template.Must(template.ParseFS(assets, "templates/*.html")) staticFiles, err := fs.Sub(assets, "static") if err != nil { @@ -38,7 +53,7 @@ func NewHandler(fetcher NewerFetcher, now func() time.Time, logger *log.Logger) if err != nil { panic(err) } - s := &server{fetcher: fetcher, now: now, logger: logger, zone: zone, tmpl: tmpl, cache: make(map[string]cacheEntry)} + s := &server{fetcher: fetcher, jobs: jobs, now: now, logger: logger, zone: zone, tmpl: tmpl, cache: make(map[string]cacheEntry)} mux := http.NewServeMux() mux.Handle("GET /static/", http.StripPrefix("/static/", http.FileServerFS(staticFiles))) mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) { @@ -47,8 +62,13 @@ func NewHandler(fetcher NewerFetcher, now func() time.Time, logger *log.Logger) }) mux.HandleFunc("GET /{$}", s.home) mux.HandleFunc("GET /newer", s.newerFragment) + mux.HandleFunc("POST /episodes/enqueue", s.enqueue) + mux.HandleFunc("POST /episodes/retry", s.retryEpisode) + mux.HandleFunc("GET /queue", s.queuePage) + mux.HandleFunc("GET /queue/list", s.queueFragment) + mux.HandleFunc("POST /jobs/{id}/retry", s.retryJob) + mux.HandleFunc("POST /jobs/{id}/cancel", s.cancelJob) for path, title := range map[string]string{ - "GET /queue": "待ち行列", "GET /subscriptions": "登録番組", "GET /library": "ライブラリ", } { diff --git a/internal/web/newer.go b/internal/web/newer.go index e5c4836..100b543 100644 --- a/internal/web/newer.go +++ b/internal/web/newer.go @@ -9,6 +9,7 @@ import ( "regexp" "time" + "gitea.kazu634.com/kazu634/tver/internal/store" "gitea.kazu634.com/kazu634/tver/internal/tver" ) @@ -34,6 +35,10 @@ type newerOptions struct { } type episodeView struct { + ID string + Genre string + Status string + StatusText string URL string ThumbnailURL string SeriesTitle string @@ -50,6 +55,7 @@ type newerView struct { HidePreviews bool Episodes []episodeView Failed bool + ErrorText string } func parseNewerOptions(r *http.Request) (newerOptions, error) { @@ -123,14 +129,34 @@ func (s *server) newerView(ctx context.Context, options newerOptions) newerView if err != nil { s.logger.Printf("TVer の新着取得に失敗(genre=%s): %s", options.genre, platformTokenPattern.ReplaceAllString(err.Error(), "${1}[REDACTED]")) view.Failed = true + view.ErrorText = "新着を取得できませんでした" return view } + visible := make([]tver.Episode, 0, len(episodes)) for _, episode := range episodes { if options.hidePreviews && episode.Duration <= previewMaxSeconds { continue } + visible = append(visible, episode) + } + ids := make([]string, 0, len(visible)) + for _, episode := range visible { + ids = append(ids, episode.ID) + } + statuses, err := s.jobs.EpisodeStatuses(ctx, ids) + if err != nil { + s.logger.Printf("新着のジョブ状態取得に失敗: %v", err) + view.Failed = true + view.ErrorText = "状態を取得できませんでした" + return view + } + for _, episode := range visible { endAt := time.Unix(episode.EndAt, 0).In(s.zone) view.Episodes = append(view.Episodes, episodeView{ + ID: episode.ID, + Genre: options.genre, + Status: statuses[episode.ID], + StatusText: statusText(statuses[episode.ID]), URL: "https://tver.jp/episodes/" + url.PathEscape(episode.ID), ThumbnailURL: "https://statics.tver.jp" + episode.ThumbnailPath, SeriesTitle: episode.SeriesTitle, @@ -145,6 +171,32 @@ func (s *server) newerView(ctx context.Context, options newerOptions) newerView return view } +func (s *server) cachedEpisode(genre, id string) (tver.Episode, bool) { + s.cacheMu.Lock() + defer s.cacheMu.Unlock() + entry, ok := s.cache[genre] + if !ok || !s.now().Before(entry.expiresAt) { + return tver.Episode{}, false + } + for _, ep := range entry.episodes { + if ep.ID == id { + return ep, true + } + } + return tver.Episode{}, false +} + +func storeEpisode(ep tver.Episode) store.Episode { + var endAt *time.Time + if ep.EndAt > 0 { + end := time.Unix(ep.EndAt, 0) + endAt = &end + } + return store.Episode{ID: ep.ID, SeriesID: ep.SeriesID, SeriesTitle: ep.SeriesTitle, + Title: ep.Title, BroadcasterName: ep.BroadcasterName, DurationSeconds: int(ep.Duration), + EndAt: endAt, ThumbnailPath: ep.ThumbnailPath} +} + func durationMinutes(seconds int64) int64 { minutes := seconds / 60 if seconds%60 != 0 { diff --git a/internal/web/newer_test.go b/internal/web/newer_test.go index 7a22a6a..1a089c9 100644 --- a/internal/web/newer_test.go +++ b/internal/web/newer_test.go @@ -31,10 +31,14 @@ func (f *fakeFetcher) FetchNewer(_ context.Context, genre string) ([]tver.Episod } func newTestHandler(fetcher *fakeFetcher, now func() time.Time) http.Handler { + return newTestHandlerWithStore(fetcher, &fakeJobStore{}, now) +} + +func newTestHandlerWithStore(fetcher *fakeFetcher, jobs *fakeJobStore, now func() time.Time) http.Handler { if now == nil { now = func() time.Time { return time.Date(2026, 10, 3, 0, 0, 0, 0, time.UTC) } } - return NewHandler(fetcher, now, log.New(io.Discard, "", 0)) + return NewHandler(fetcher, jobs, now, log.New(io.Discard, "", 0)) } func getPage(handler http.Handler, path string) *httptest.ResponseRecorder { @@ -226,7 +230,7 @@ func TestNewerPageEscapesUntrustedEpisodeFields(t *testing.T) { func TestNewerFailureShowsFriendlyMessageAndRedactsToken(t *testing.T) { fetcher := &fakeFetcher{err: errors.New("GET https://platform-api.tver.jp/?platform_token=secret-123&x=1: upstream unavailable")} var logs bytes.Buffer - handler := NewHandler(fetcher, time.Now, log.New(&logs, "", 0)) + handler := NewHandler(fetcher, &fakeJobStore{}, time.Now, log.New(&logs, "", 0)) response := getPage(handler, "/") if response.Code != http.StatusOK || !strings.Contains(response.Body.String(), "新着を取得できませんでした") { t.Errorf("error page: status=%d body=%s", response.Code, response.Body.String()) diff --git a/internal/web/queue.go b/internal/web/queue.go new file mode 100644 index 0000000..e81b270 --- /dev/null +++ b/internal/web/queue.go @@ -0,0 +1,191 @@ +package web + +import ( + "context" + "errors" + "fmt" + "net/http" + "strconv" + + "gitea.kazu634.com/kazu634/tver/internal/store" + "github.com/jackc/pgx/v5" +) + +const queueListLimit = 50 + +type queueJobView struct { + store.Job + SourceText, CreatedAtText, PhaseText string + PercentText, SpeedText, ETAText string + FragmentText, ErrorKindText string +} + +type queueView struct { + Running, Queued, Failed, Canceled, Succeeded []queueJobView + Poll bool +} + +func (s *server) queuePage(w http.ResponseWriter, r *http.Request) { + view, err := s.loadQueue(r.Context()) + if err != nil { + s.logger.Printf("待ち行列の取得に失敗: %v", err) + http.Error(w, "待ち行列を取得できませんでした", http.StatusInternalServerError) + return + } + s.render(w, "layout", page{Title: "待ち行列", Queue: &view}) +} + +func (s *server) queueFragment(w http.ResponseWriter, r *http.Request) { + view, err := s.loadQueue(r.Context()) + if err != nil { + s.logger.Printf("待ち行列の取得に失敗: %v", err) + http.Error(w, "待ち行列を取得できませんでした", http.StatusInternalServerError) + return + } + s.render(w, "queue_list", view) +} + +func (s *server) loadQueue(ctx context.Context) (queueView, error) { + var view queueView + for _, item := range []struct { + status string + assign func([]queueJobView) + }{ + {"running", func(j []queueJobView) { view.Running = j }}, + {"queued", func(j []queueJobView) { view.Queued = j }}, + {"failed", func(j []queueJobView) { view.Failed = j }}, + {"canceled", func(j []queueJobView) { view.Canceled = j }}, + {"succeeded", func(j []queueJobView) { view.Succeeded = j }}, + } { + jobs, err := s.jobs.ListJobs(ctx, item.status, queueListLimit) + if err != nil { + return queueView{}, err + } + views := make([]queueJobView, 0, len(jobs)) + for _, job := range jobs { + views = append(views, s.queueJob(job)) + } + item.assign(views) + } + view.Poll = len(view.Running) > 0 || len(view.Queued) > 0 + return view, nil +} + +func (s *server) queueJob(job store.Job) queueJobView { + view := queueJobView{Job: job, CreatedAtText: job.CreatedAt.In(s.zone).Format("2006-01-02 15:04 MST")} + if job.Source == "subscription" { + view.SourceText = "自動" + } else { + view.SourceText = "手動" + } + if job.Phase != nil { + switch *job.Phase { + case "video": + view.PhaseText = "映像" + case "audio": + view.PhaseText = "音声" + case "merging": + view.PhaseText = "結合" + case "moving": + view.PhaseText = "移動" + } + } + if job.DownloadedBytes != nil && job.TotalBytes != nil && *job.TotalBytes > 0 { + percent := 100 * float64(*job.DownloadedBytes) / float64(*job.TotalBytes) + percent = max(0, min(percent, 100)) + if job.TotalIsEstimate { + view.PercentText = fmt.Sprintf("約%.0f%%", percent) + } else { + view.PercentText = fmt.Sprintf("%.0f%%", percent) + } + } + if job.SpeedBPS != nil && *job.SpeedBPS > 0 { + view.SpeedText = formatBytes(*job.SpeedBPS) + "/秒" + } + if job.ETASeconds != nil && *job.ETASeconds >= 0 { + seconds := int64(*job.ETASeconds) + view.ETAText = fmt.Sprintf("残り%d分%d秒", seconds/60, seconds%60) + } + if job.FragmentIndex != nil && job.FragmentCount != nil && *job.FragmentCount > 0 { + view.FragmentText = fmt.Sprintf("断片 %d/%d", *job.FragmentIndex, *job.FragmentCount) + } + if job.ErrorKind != nil { + view.ErrorKindText = errorKindText(*job.ErrorKind) + } + return view +} + +func formatBytes(n float64) string { + for _, unit := range []string{"B", "KiB", "MiB", "GiB"} { + if n < 1024 || unit == "GiB" { + return fmt.Sprintf("%.1f %s", n, unit) + } + n /= 1024 + } + return "" +} + +func errorKindText(kind string) string { + switch kind { + case "drm": + return "DRM のため取得不可" + case "geo": + return "地域制限" + case "network": + return "ネットワーク" + case "unavailable": + return "配信終了・存在しない" + case "extract": + return "抽出失敗" + case "invalid_output": + return "出力形式の不正" + default: + return "不明" + } +} + +func (s *server) retryJob(w http.ResponseWriter, r *http.Request) { + s.queueMutation(w, r, s.jobs.Retry, false) +} + +func (s *server) cancelJob(w http.ResponseWriter, r *http.Request) { + s.queueMutation(w, r, s.jobs.CancelQueued, true) +} + +func (s *server) queueMutation(w http.ResponseWriter, r *http.Request, change func(context.Context, int64) error, queuedOnly bool) { + id, err := strconv.ParseInt(r.PathValue("id"), 10, 64) + if err != nil || id <= 0 { + http.Error(w, "ジョブ ID が不正です", http.StatusBadRequest) + return + } + if queuedOnly { + job, err := s.jobs.GetJob(r.Context(), id) + if errors.Is(err, pgx.ErrNoRows) { + http.Error(w, "ジョブが見つかりません", http.StatusNotFound) + return + } + if err != nil { + s.logger.Printf("ジョブ状態の取得に失敗: %v", err) + http.Error(w, "操作できませんでした", http.StatusInternalServerError) + return + } + if job.Status != "queued" { + http.Error(w, "待機中のジョブだけ取り消せます", http.StatusConflict) + return + } + } + if err := change(r.Context(), id); err != nil { + if errors.Is(err, store.ErrInvalidTransition) { + http.Error(w, "操作できる状態ではありません", http.StatusConflict) + } else { + s.logger.Printf("待ち行列の操作に失敗: %v", err) + http.Error(w, "操作できませんでした", http.StatusInternalServerError) + } + return + } + if r.Header.Get("HX-Request") != "true" { + http.Redirect(w, r, "/queue", http.StatusSeeOther) + return + } + s.queueFragment(w, r) +} diff --git a/internal/web/queue_test.go b/internal/web/queue_test.go new file mode 100644 index 0000000..584bc5d --- /dev/null +++ b/internal/web/queue_test.go @@ -0,0 +1,346 @@ +package web + +import ( + "errors" + "net/http" + "net/http/httptest" + "net/url" + "regexp" + "slices" + "strings" + "testing" + "time" + + "gitea.kazu634.com/kazu634/tver/internal/store" + "gitea.kazu634.com/kazu634/tver/internal/tver" +) + +func postForm(handler http.Handler, path string, values url.Values) *httptest.ResponseRecorder { + r := httptest.NewRequest(http.MethodPost, path, strings.NewReader(values.Encode())) + r.Header.Set("Content-Type", "application/x-www-form-urlencoded") + r.Header.Set("HX-Request", "true") + w := httptest.NewRecorder() + handler.ServeHTTP(w, r) + return w +} + +func TestEnqueueUsesCachedEpisodeAndDuplicateStatus(t *testing.T) { + fetcher := &fakeFetcher{episodes: []tver.Episode{{ID: "ep1", SeriesID: "series1", SeriesTitle: "正しい番組", Title: "正しい回", BroadcasterName: "正しい局", Duration: 1234, EndAt: 1791061200, ThumbnailPath: "/correct.jpg"}}} + jobs := &fakeJobStore{} + handler := newTestHandlerWithStore(fetcher, jobs, nil) + getPage(handler, "/") + if jobs.statusCalls != 1 || len(jobs.statusIDs) != 1 || jobs.statusIDs[0] != "ep1" { + t.Fatalf("状態の一括取得: calls=%d ids=%v", jobs.statusCalls, jobs.statusIDs) + } + values := url.Values{"genre": {"drama"}, "episode_id": {"ep1"}, "series_title": {"偽番組"}, "title": {"偽回"}} + response := postForm(handler, "/episodes/enqueue", values) + if response.Code != http.StatusOK || !strings.Contains(response.Body.String(), "待機中") || !strings.Contains(response.Body.String(), `class="episode-action"`) || strings.Contains(response.Body.String(), "ダウンロード") { + t.Fatalf("登録後の一覧: %s", page.Body.String()) + } +} + +func TestEnqueueRejectsInvalidOrExpiredCache(t *testing.T) { + now := time.Date(2026, 10, 3, 0, 0, 0, 0, time.UTC) + fetcher := &fakeFetcher{episodes: []tver.Episode{{ID: "ep1", Duration: 200}}} + jobs := &fakeJobStore{} + handler := newTestHandlerWithStore(fetcher, jobs, func() time.Time { return now }) + valid := url.Values{"genre": {"drama"}, "episode_id": {"ep1"}} + if got := postForm(handler, "/episodes/enqueue", valid).Code; got != http.StatusNotFound { + t.Fatalf("未取得キャッシュ: %d", got) + } + getPage(handler, "/") + for _, values := range []url.Values{ + {"genre": {"news"}, "episode_id": {"ep1"}}, + {"genre": {"drama"}}, + } { + if got := postForm(handler, "/episodes/enqueue", values).Code; got != http.StatusBadRequest { + t.Errorf("不正なフォーム: %d", got) + } + } + if got := postForm(handler, "/episodes/enqueue", url.Values{"genre": {"drama"}, "episode_id": {"unknown"}}).Code; got != http.StatusNotFound { + t.Fatalf("同じジャンルの未収録 ID: %d", got) + } + if got := postForm(handler, "/episodes/enqueue", url.Values{"genre": {"variety"}, "episode_id": {"ep1"}}).Code; got != http.StatusNotFound { + t.Fatalf("別ジャンルのキャッシュ: %d", got) + } + now = now.Add(newerCacheTTL) + if got := postForm(handler, "/episodes/enqueue", valid).Code; got != http.StatusNotFound { + t.Fatalf("期限切れキャッシュ: %d", got) + } + if jobs.enqueueCalls != 0 || len(fetcher.calls) != 1 { + t.Fatalf("無効な入力で登録・再取得: enqueue=%d fetch=%v", jobs.enqueueCalls, fetcher.calls) + } +} + +func TestEnqueueUsesSelectedGenreCacheForSameEpisodeID(t *testing.T) { + fetcher := &fakeFetcher{episodesByGenre: map[string][]tver.Episode{ + "drama": {{ID: "shared", SeriesTitle: "ドラマの番組", Duration: 200}}, + "variety": {{ID: "shared", SeriesTitle: "バラエティの番組", Duration: 200}}, + }} + jobs := &fakeJobStore{} + handler := newTestHandlerWithStore(fetcher, jobs, nil) + getPage(handler, "/") + getPage(handler, "/?genre=variety") + response := postForm(handler, "/episodes/enqueue", url.Values{ + "genre": {"variety"}, "episode_id": {"shared"}, "series_title": {"偽タイトル"}, + }) + if response.Code != http.StatusOK || jobs.lastEpisode.SeriesTitle != "バラエティの番組" || jobs.enqueueCalls != 1 || !slices.Equal(fetcher.calls, []string{"drama", "variety"}) { + t.Fatalf("選択ジャンルのキャッシュ: status=%d episode=%+v calls=%v", response.Code, jobs.lastEpisode, fetcher.calls) + } +} + +func TestNewerStatusesAndRetry(t *testing.T) { + fetcher := &fakeFetcher{episodes: []tver.Episode{{ID: "failed", Duration: 200}, {ID: "canceled", Duration: 200}, {ID: "running", Duration: 200}, {ID: "done", Duration: 200}}} + jobs := &fakeJobStore{jobs: map[string]store.Job{ + "failed": {ID: 1, Status: "failed"}, "canceled": {ID: 2, Status: "canceled"}, + "running": {ID: 3, Status: "running"}, "done": {ID: 4, Status: "succeeded"}, + }} + handler := newTestHandlerWithStore(fetcher, jobs, nil) + body := getPage(handler, "/").Body.String() + for _, want := range []string{"失敗", "取り消し", "ダウンロード中", "完了"} { + if !strings.Contains(body, want) { + t.Errorf("状態表示がない: %s", want) + } + } + if got := strings.Count(body, ">再試行"); got != 2 { + t.Fatalf("再試行ボタン数: %d", got) + } + response := postForm(handler, "/episodes/retry", url.Values{"genre": {"drama"}, "episode_id": {"failed"}}) + if response.Code != http.StatusOK || !strings.Contains(response.Body.String(), "待機中") || jobs.jobs["failed"].Status != "queued" { + t.Fatalf("新着から再試行: %d %s", response.Code, response.Body.String()) + } + if got := postForm(handler, "/episodes/retry", url.Values{"genre": {"drama"}, "episode_id": {"running"}}).Code; got != http.StatusConflict { + t.Fatalf("実行中の再試行: %d", got) + } + if got := postForm(handler, "/episodes/retry", url.Values{"genre": {"drama"}, "episode_id": {"missing"}}).Code; got != http.StatusNotFound { + t.Fatalf("キャッシュにない ID の再試行: %d", got) + } +} + +func TestQueueGroupsProgressAndPolling(t *testing.T) { + phase, errorKind, errorMessage := "video", "drm", `` + downloaded, total := int64(25), int64(100) + speed, eta := float64(2048), float64(63) + index, count := int32(2), int32(10) + created := time.Date(2026, 10, 3, 0, 0, 0, 0, time.UTC) + jobs := &fakeJobStore{jobs: map[string]store.Job{ + "running": {ID: 1, Status: "running", Source: "subscription", SeriesTitle: "番組A", Title: "第1話", BroadcasterName: "局A", CreatedAt: created, Phase: &phase, DownloadedBytes: &downloaded, TotalBytes: &total, TotalIsEstimate: true, SpeedBPS: &speed, ETASeconds: &eta, FragmentIndex: &index, FragmentCount: &count}, + "queued": {ID: 2, Status: "queued", Source: "manual", SeriesTitle: "番組B", CreatedAt: created}, + "failed": {ID: 3, Status: "failed", ErrorKind: &errorKind, ErrorMessage: &errorMessage, CreatedAt: created}, + "canceled": {ID: 4, Status: "canceled", CreatedAt: created}, + "done": {ID: 5, Status: "succeeded", CreatedAt: created}, + }} + handler := newTestHandlerWithStore(&fakeFetcher{}, jobs, nil) + response := getPage(handler, "/queue") + if response.Code != http.StatusOK { + t.Fatalf("待ち行列: %d", response.Code) + } + body := response.Body.String() + for _, want := range []string{"ダウンロード中", "待機中", "失敗", "取り消し", "完了(直近50件)", "番組A", "第1話", "局A", "自動", "手動", "2026-10-03 09:00 JST", "映像", "約25%", "2.0 KiB/秒", "残り1分3秒", "断片 2/10", "DRM のため取得不可", "
", "<script>", `hx-get="/queue/list"`, `hx-trigger="every 5s"`} { + if !strings.Contains(body, want) { + t.Errorf("待ち行列に %q がない", want) + } + } + if strings.Contains(body, errorMessage) || strings.Contains(body, `action="/jobs/1/cancel"`) || strings.Contains(body, "全体") { + t.Errorf("不正な詳細・操作・全体進捗: %s", body) + } + fragment := getPage(handler, "/queue/list") + if fragment.Code != http.StatusOK || strings.Contains(fragment.Body.String(), "(.*?)`).FindStringSubmatch(response.Body.String()) + if len(section) != 2 || !strings.Contains(section[1], "番組-"+statuses[i]) { + t.Errorf("%s に %s がない: %v", label, statuses[i], section) + continue + } + for j, other := range statuses { + if j != i && strings.Contains(section[1], "番組-"+other) { + t.Errorf("%s に %s が混入", label, other) + } + } + cancelAction := `action="/jobs/` + string(rune('1'+i)) + `/cancel"` + retryAction := `action="/jobs/` + string(rune('1'+i)) + `/retry"` + switch statuses[i] { + case "queued": + if !strings.Contains(section[1], cancelAction) || strings.Contains(section[1], retryAction) { + t.Errorf("%s の操作が不正: %s", statuses[i], section[1]) + } + case "failed", "canceled": + if !strings.Contains(section[1], retryAction) || strings.Contains(section[1], cancelAction) { + t.Errorf("%s の操作が不正: %s", statuses[i], section[1]) + } + default: + if strings.Contains(section[1], "/cancel") || strings.Contains(section[1], "/retry") { + t.Errorf("%s に操作がある: %s", statuses[i], section[1]) + } + } + } +} + +func TestQueueEscapesEpisodeFields(t *testing.T) { + attack := `` + message := `` + kind := "extract" + jobs := &fakeJobStore{jobs: map[string]store.Job{ + "bad": {ID: 1, Status: "failed", SeriesTitle: attack, Title: attack, + BroadcasterName: attack, ErrorKind: &kind, ErrorMessage: &message}, + }} + response := getPage(newTestHandlerWithStore(&fakeFetcher{}, jobs, nil), "/queue") + body := response.Body.String() + if response.Code != http.StatusOK || strings.Contains(body, attack) || strings.Contains(body, message) || strings.Count(body, "<script>") != 3 || !strings.Contains(body, "<img src=x onerror=alert(1)>") { + t.Fatalf("待ち行列の HTML エスケープ: %d %s", response.Code, body) + } +} + +func TestQueueCancelRetryAndStopPolling(t *testing.T) { + jobs := &fakeJobStore{jobs: map[string]store.Job{"ep": {ID: 1, Status: "queued"}}} + handler := newTestHandlerWithStore(&fakeFetcher{}, jobs, nil) + if !strings.Contains(getPage(handler, "/queue").Body.String(), `hx-get="/queue/list"`) { + t.Fatal("待機中なのに更新がない") + } + canceled := postForm(handler, "/jobs/1/cancel", nil) + if canceled.Code != http.StatusOK || !strings.Contains(canceled.Body.String(), `action="/jobs/1/retry"`) || strings.Contains(canceled.Body.String(), `hx-get="/queue/list"`) { + t.Fatalf("取り消し後: %d %s", canceled.Code, canceled.Body.String()) + } + retried := postForm(handler, "/jobs/1/retry", nil) + if retried.Code != http.StatusOK || !strings.Contains(retried.Body.String(), `action="/jobs/1/cancel"`) || !strings.Contains(retried.Body.String(), `hx-get="/queue/list"`) { + t.Fatalf("再試行後: %d %s", retried.Code, retried.Body.String()) + } + if got := postForm(handler, "/jobs/1/retry", nil).Code; got != http.StatusConflict { + t.Fatalf("待機中の再試行: %d", got) + } + if got := postForm(handler, "/jobs/bad/cancel", nil).Code; got != http.StatusBadRequest { + t.Fatalf("不正な ID: %d", got) + } + jobs.jobs["ep"] = store.Job{ID: 1, Status: "running"} + if got := postForm(handler, "/jobs/1/cancel", nil).Code; got != http.StatusConflict { + t.Fatalf("実行中の取り消し: %d", got) + } +} + +func TestStateChangingGETIsRejected(t *testing.T) { + jobs := &fakeJobStore{jobs: map[string]store.Job{"ep": {ID: 1, Status: "queued"}}} + handler := newTestHandlerWithStore(&fakeFetcher{episodes: []tver.Episode{{ID: "ep", Duration: 200}}}, jobs, nil) + getPage(handler, "/") + for _, path := range []string{ + "/episodes/enqueue?episode_id=ep&genre=drama", + "/episodes/retry?episode_id=ep&genre=drama", + "/jobs/1/cancel", "/jobs/1/retry", + } { + response := getPage(handler, path) + if response.Code != http.StatusMethodNotAllowed || !strings.Contains(response.Header().Get("Allow"), "POST") { + t.Errorf("GET %s: status=%d Allow=%q", path, response.Code, response.Header().Get("Allow")) + } + } + if jobs.jobs["ep"].Status != "queued" || jobs.enqueueCalls != 0 { + t.Fatalf("GET が状態を変更した: jobs=%+v enqueue=%d", jobs.jobs, jobs.enqueueCalls) + } +} + +func TestStateChangingPOSTWithoutHtmxRedirects(t *testing.T) { + jobs := &fakeJobStore{} + handler := newTestHandlerWithStore(&fakeFetcher{episodes: []tver.Episode{{ID: "ep", Duration: 200}}}, jobs, nil) + getPage(handler, "/") + form := url.Values{"episode_id": {"ep"}, "genre": {"drama"}} + request := httptest.NewRequest(http.MethodPost, "/episodes/enqueue", strings.NewReader(form.Encode())) + request.Header.Set("Content-Type", "application/x-www-form-urlencoded") + response := httptest.NewRecorder() + handler.ServeHTTP(response, request) + if response.Code != http.StatusSeeOther || response.Header().Get("Location") != "/?genre=drama" || jobs.jobs["ep"].Status != "queued" { + t.Fatalf("通常 POST 後の遷移: status=%d Location=%q jobs=%+v", response.Code, response.Header().Get("Location"), jobs.jobs) + } + request = httptest.NewRequest(http.MethodPost, "/jobs/1/cancel", nil) + response = httptest.NewRecorder() + handler.ServeHTTP(response, request) + if response.Code != http.StatusSeeOther || response.Header().Get("Location") != "/queue" || jobs.jobs["ep"].Status != "canceled" { + t.Fatalf("待ち行列操作後の遷移: status=%d Location=%q jobs=%+v", response.Code, response.Header().Get("Location"), jobs.jobs) + } +} + +func TestNewerStoreFailureDoesNotOfferDownload(t *testing.T) { + jobs := &fakeJobStore{err: errors.New("db unavailable")} + handler := newTestHandlerWithStore(&fakeFetcher{episodes: []tver.Episode{{ID: "ep", Duration: 200}}}, jobs, nil) + response := getPage(handler, "/") + if response.Code != http.StatusOK || !strings.Contains(response.Body.String(), "状態を取得できませんでした") || strings.Contains(response.Body.String(), ">ダウンロード") { + t.Errorf("状態取得失敗: %d %s", response.Code, response.Body.String()) + } +} + +func TestEnqueueStoreFailureShowsShortError(t *testing.T) { + jobs := &fakeJobStore{} + handler := newTestHandlerWithStore(&fakeFetcher{episodes: []tver.Episode{{ID: "ep", Duration: 200}}}, jobs, nil) + getPage(handler, "/") + jobs.err = errors.New("secret database detail") + response := postForm(handler, "/episodes/enqueue", url.Values{"genre": {"drama"}, "episode_id": {"ep"}}) + if response.Code != http.StatusInternalServerError || !strings.Contains(response.Body.String(), "待ち行列に登録できませんでした") || strings.Contains(response.Body.String(), "secret") { + t.Errorf("DB エラーの表示: %d %s", response.Code, response.Body.String()) + } +} + +func TestQueueProgressAndFailureLabels(t *testing.T) { + s := &server{zone: time.FixedZone("JST", 9*60*60)} + for _, tt := range []struct{ kind, label string }{ + {"drm", "DRM のため取得不可"}, {"geo", "地域制限"}, + {"network", "ネットワーク"}, {"unavailable", "配信終了・存在しない"}, + {"extract", "抽出失敗"}, {"invalid_output", "出力形式の不正"}, + {"unknown", "不明"}, + } { + if got := errorKindText(tt.kind); got != tt.label { + t.Errorf("%s: %q", tt.kind, got) + } + } + for _, tt := range []struct{ phase, label string }{ + {"video", "映像"}, {"audio", "音声"}, {"merging", "結合"}, {"moving", "移動"}, + } { + if got := s.queueJob(store.Job{Phase: &tt.phase}).PhaseText; got != tt.label { + t.Errorf("%s: %q", tt.phase, got) + } + } + downloaded, total := int64(50), int64(100) + job := store.Job{DownloadedBytes: &downloaded, TotalBytes: &total} + if got := s.queueJob(job).PercentText; got != "50%" { + t.Errorf("確定した段階の割合: %q", got) + } + job.TotalIsEstimate = true + if got := s.queueJob(job).PercentText; got != "約50%" { + t.Errorf("推定した段階の割合: %q", got) + } + total = 0 + if got := s.queueJob(job).PercentText; got != "" { + t.Errorf("総量ゼロでの割合: %q", got) + } +} diff --git a/internal/web/static/app.css b/internal/web/static/app.css index 0460a6c..3c2ed79 100644 --- a/internal/web/static/app.css +++ b/internal/web/static/app.css @@ -13,3 +13,10 @@ main { max-width: 72rem; margin: 2rem auto; padding: 0 2rem; } .episode-copy { padding: .8rem 1rem; } .episode-copy h2 { font-size: 1.1rem; margin: 0 0 .4rem; } .episode-copy p { margin: .3rem 0; } +.episode-action { padding: 0 .8rem .8rem; } +.episode-action form { display: inline; } +.episode-action button, .queue-job button { font: inherit; padding: .35rem .7rem; } +.queue-job { border: 1px solid #d6dee4; border-radius: .5rem; margin: .75rem 0; padding: .8rem 1rem; } +.queue-job h3 { margin: 0 0 .4rem; } +.queue-job p { margin: .4rem 0; } +.queue-job pre { white-space: pre-wrap; overflow-wrap: anywhere; } diff --git a/internal/web/store_test.go b/internal/web/store_test.go new file mode 100644 index 0000000..ea70b0e --- /dev/null +++ b/internal/web/store_test.go @@ -0,0 +1,132 @@ +package web + +import ( + "context" + "sort" + "time" + + "gitea.kazu634.com/kazu634/tver/internal/store" + "github.com/jackc/pgx/v5" +) + +type fakeJobStore struct { + jobs map[string]store.Job + nextID int64 + lastEpisode store.Episode + enqueueCalls, statusCalls int + statusIDs []string + listStatuses []string + listLimits []int + err error +} + +func (f *fakeJobStore) ensure() { + if f.jobs == nil { + f.jobs = make(map[string]store.Job) + } +} + +func (f *fakeJobStore) Enqueue(_ context.Context, ep store.Episode, source string) (store.Job, bool, error) { + f.enqueueCalls++ + f.lastEpisode = ep + if f.err != nil { + return store.Job{}, false, f.err + } + f.ensure() + if job, ok := f.jobs[ep.ID]; ok { + return job, false, nil + } + f.nextID++ + job := store.Job{ID: f.nextID, EpisodeID: ep.ID, SeriesTitle: ep.SeriesTitle, + Title: ep.Title, BroadcasterName: ep.BroadcasterName, Source: source, + Status: "queued", CreatedAt: time.Date(2026, 10, 3, 0, 0, 0, 0, time.UTC)} + f.jobs[ep.ID] = job + return job, true, nil +} + +func (f *fakeJobStore) EpisodeStatuses(_ context.Context, ids []string) (map[string]string, error) { + f.statusCalls++ + f.statusIDs = append([]string(nil), ids...) + if f.err != nil { + return nil, f.err + } + result := make(map[string]string) + for _, id := range ids { + if job, ok := f.jobs[id]; ok { + result[id] = job.Status + } + } + return result, nil +} + +func (f *fakeJobStore) ListJobs(_ context.Context, status string, limit int) ([]store.Job, error) { + f.listStatuses = append(f.listStatuses, status) + f.listLimits = append(f.listLimits, limit) + if f.err != nil { + return nil, f.err + } + var result []store.Job + for _, job := range f.jobs { + if job.Status == status { + result = append(result, job) + } + } + sort.Slice(result, func(i, j int) bool { return result[i].ID > result[j].ID }) + if len(result) > limit { + result = result[:limit] + } + return result, nil +} + +func (f *fakeJobStore) GetJob(_ context.Context, id int64) (store.Job, error) { + if f.err != nil { + return store.Job{}, f.err + } + for _, job := range f.jobs { + if job.ID == id { + return job, nil + } + } + return store.Job{}, pgx.ErrNoRows +} + +func (f *fakeJobStore) Retry(_ context.Context, id int64) error { + for ep, job := range f.jobs { + if job.ID == id && (job.Status == "failed" || job.Status == "canceled") { + job.Status = "queued" + f.jobs[ep] = job + return nil + } + } + return store.ErrInvalidTransition +} + +func (f *fakeJobStore) RetryEpisode(_ context.Context, id string) error { + job, ok := f.jobs[id] + if !ok || (job.Status != "failed" && job.Status != "canceled") { + return store.ErrInvalidTransition + } + job.Status = "queued" + f.jobs[id] = job + return nil +} + +func (f *fakeJobStore) Cancel(_ context.Context, id int64) error { + if f.err != nil { + return f.err + } + for ep, job := range f.jobs { + if job.ID == id && job.Status == "queued" { + job.Status = "canceled" + f.jobs[ep] = job + return nil + } + } + return store.ErrInvalidTransition +} + +func (f *fakeJobStore) CancelQueued(ctx context.Context, id int64) error { + return f.Cancel(ctx, id) +} + +var _ JobStore = (*fakeJobStore)(nil) diff --git a/internal/web/templates/layout.html b/internal/web/templates/layout.html index 316107d..f92f3e4 100644 --- a/internal/web/templates/layout.html +++ b/internal/web/templates/layout.html @@ -20,6 +20,8 @@
{{if .Newer}} {{template "newer_page" .Newer}} + {{else if .Queue}} + {{template "queue_page" .Queue}} {{else}}

{{.Title}}

{{end}} diff --git a/internal/web/templates/newer.html b/internal/web/templates/newer.html index d325576..080c7a4 100644 --- a/internal/web/templates/newer.html +++ b/internal/web/templates/newer.html @@ -21,7 +21,7 @@ {{define "newer_results"}}
{{if .Failed}} -

新着を取得できませんでした

+

{{.ErrorText}}

{{else if not .Episodes}}

表示する新着はありません

{{else}} @@ -39,9 +39,31 @@

配信終了:

+ {{template "episode_action" .}} {{end}} {{end}}
{{end}} + +{{define "episode_action"}} +
+ {{if not .Status}} +
+ + + +
+ {{else}} + {{.StatusText}} + {{if or (eq .Status "failed") (eq .Status "canceled")}} +
+ + + +
+ {{end}} + {{end}} +
+{{end}} diff --git a/internal/web/templates/queue.html b/internal/web/templates/queue.html new file mode 100644 index 0000000..8055a1f --- /dev/null +++ b/internal/web/templates/queue.html @@ -0,0 +1,56 @@ +{{define "queue_page"}} +

待ち行列

+{{template "queue_list" .}} +{{end}} + +{{define "queue_list"}} +
+
+

ダウンロード中

+ {{range .Running}}{{template "queue_job" .}}{{else}}

ありません

{{end}} +
+
+

待機中

+ {{range .Queued}}{{template "queue_job" .}}{{else}}

ありません

{{end}} +
+
+

失敗

+ {{range .Failed}}{{template "queue_job" .}}{{else}}

ありません

{{end}} +
+
+

取り消し

+ {{range .Canceled}}{{template "queue_job" .}}{{else}}

ありません

{{end}} +
+
+

完了(直近50件)

+ {{range .Succeeded}}{{template "queue_job" .}}{{else}}

ありません

{{end}} +
+
+{{end}} + +{{define "queue_job"}} +
+

{{.SeriesTitle}} — {{.Title}}

+

放送局: {{.BroadcasterName}}/登録元: {{.SourceText}}/登録日時: {{.CreatedAtText}}

+ {{if eq .Status "running"}} +

段階: {{if .PhaseText}}{{.PhaseText}}{{else}}準備中{{end}} + {{if .PercentText}}/{{.PercentText}}{{end}} + {{if .SpeedText}}/{{.SpeedText}}{{end}} + {{if .ETAText}}/{{.ETAText}}{{end}} + {{if .FragmentText}}/{{.FragmentText}}{{end}}

+ {{end}} + {{if eq .Status "failed"}} +

失敗理由: {{.ErrorKindText}}

+ {{if .ErrorMessage}}
詳細
{{.ErrorMessage}}
{{end}} + {{end}} + {{if eq .Status "queued"}} +
+ +
+ {{else if or (eq .Status "failed") (eq .Status "canceled")}} +
+ +
+ {{end}} +
+{{end}}