diff --git a/PROJECT.md b/PROJECT.md index 91cb7d4..d1ff2e8 100644 --- a/PROJECT.md +++ b/PROJECT.md @@ -3,7 +3,7 @@ title: tver の現状 tags: - homelab - tver -status: worker のダウンロード本体を実装中 +status: worker の登録番組確認を実装中 --- # tver の現状 @@ -14,7 +14,7 @@ status: worker のダウンロード本体を実装中 ## 今の状態 - Go の HTTP サーバー、TVer API クライアント、ドラマ・バラエティの新着表示、テスト、web・worker コンテナの定義がある -- **worker のダウンロード本体を実装中。** 待ち行列から yt-dlp を起動し、進捗・heartbeat・取消・出力検査・保存・失敗分類を扱う。登録番組の定期確認はまだ行わない +- **worker のダウンロード本体と登録番組の定期確認を実装中。** 待ち行列からの取得に加え、有効な登録番組の新しい回を確認して自動登録する - リポジトリ: `https://gitea.kazu634.com/kazu634/tver.git` ## 目的 @@ -41,6 +41,9 @@ TVer(`https://tver.jp/`)で公開中の動画情報を取得し、選んだ | 保持期間 | **削除は手動のみ**。自動削除はしない | | 待ち行列 | **Postgres のテーブルだけで実装する**。専用のキュー基盤は持たない | | 定期確認の間隔 | 登録番組の新しい回を **24 時間ごと**に確認する | +| 定期確認の走査 | worker のスケジューラが起動直後と1時間ごとに、最終確認から24時間以上経過した有効な番組を探す。番組は1件ずつ確認し、間に数秒空ける | +| 定期確認の予告 | **180秒以下**の回は自動登録せず、既知としてだけ記録する。新着画面の予告判定と同じしきい値を使う | +| 定期確認の失敗 | `last_checked_at` を更新せず、失敗日時と秘匿値を伏せた短いメッセージを保存する。次の1時間走査で再試行し、成功時にエラーを消す | | 登録時の既存回 | **登録の都度**「公開中の回もダウンロードする」かを選ぶ。選ばない場合も公開中の回を既知として記録する | | 番組の一時停止 | 登録は残し、自動ダウンロードだけ止める。再開後は停止中に公開された未確認の回も対象にする | | 動画のマウント先 | web と worker は `VIDEO_DIR`(既定 `/videos`)に同じ NAS 共有をマウントする。`download_jobs.file_path` は `VIDEO_DIR` からの相対パスを保存する | @@ -86,7 +89,9 @@ TVer(`https://tver.jp/`)で公開中の動画情報を取得し、選んだ - 現在の `web` は `/healthz`、新着(`/`)、待ち行列(`/queue`)、登録番組(`/subscriptions`)、完了済みライブラリ(`/library`)、再生ページ(`/library/{id}`)、Range 対応の MP4 配信(`/videos/{id}`)を提供する - 動画配信と削除はクライアントからジョブ ID だけを受け取り、DB の相対 `file_path` を使う。絶対パス・`..`・`VIDEO_DIR` 外を指すシンボリックリンクを拒否する。削除後は新着から手動で再登録できるが、登録番組の既知記録は残るため自動再登録しない - 新着の「番組を登録」から、公開中の回をダウンロードするか選ぶ。web はサーバー側の新着キャッシュから番組 ID と名前を引き、TVer の season ごとの公開中エピソードを取得する。取得失敗時は登録しない -- 登録と公開中の回の既知記録・任意の待ち行列投入は1トランザクションで行い、登録時刻を最終確認日時に設定する。`known_episodes` は番組解除時に連動して削除する。worker 用の未確認回の抽出と、未確認回だけの既知記録・投入も `internal/store` に用意した。一時停止中は投入を拒否する +- 登録と公開中の回の既知記録・任意の待ち行列投入は1トランザクションで行い、登録時刻を最終確認日時に設定する。`known_episodes` は番組解除時に連動して削除する。一時停止中は投入を拒否する +- worker のスケジューラは起動直後と1時間ごとに、`enabled` かつ最終確認から24時間以上の番組を抽出する。TVer から公開中の回を順次取得し、未確認の180秒超の回だけを `source=subscription` で投入する。180秒以下も既知にし、成功時の最終確認更新・エラー消去まで1トランザクションで行う +- TVer の番組取得に失敗した場合は最終確認日時を変えず、失敗日時と伏字済みの短いメッセージを `subscriptions` に記録して画面に表示する。番組間には3秒空け、SIGTERM では取得中または待機中の確認を中断する - 新着の登録ボタンはエピソード ID とジャンルだけを送る。web が有効な新着キャッシュから番組情報を引き、`episode_id` の一意制約で重複登録を防ぐ。新着一覧はジョブ状態を一括取得し、失敗・取り消しには再試行を表示する - 待ち行列は実行中・待機中・失敗・取り消し・完了を状態別に直近50件ずつ表示する。段階別の進捗と失敗理由を表示し、待機中の取り消しと失敗・取り消しの再試行ができる。実行中または待機中のジョブがある間だけ5秒ごとに一覧を更新する - 実行中のジョブにも取り消しボタンを表示する。worker は DB の `canceled` を定期確認し、yt-dlp をプロセスグループごと停止する @@ -97,7 +102,7 @@ TVer(`https://tver.jp/`)で公開中の動画情報を取得し、選んだ - 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 を外部コマンドとして使い、映像・音声・結合・移動の段階別進捗を1秒間隔に間引いて保存する。独立した heartbeat を5秒ごとに更新し、正確な全体パーセントは保存しない。`after_move:filepath` で実パスを取り、ffprobe で H.264 と AAC の MP4 を確認して成功とする。失敗は取り消し、DRM、地域制限、ネットワーク、配信不可、抽出失敗、出力不正、その他の順に扱い、自動再試行しない -- worker は起動時にマイグレーションと停止中ジョブの `RequeueStale` を行う。SIGTERM では yt-dlp を止め、担当中ジョブを `queued` に戻してから終了する。`Dockerfile.worker` は yt-dlp 2026.08.19 を ARG で固定し、ffmpeg・ffprobe とともに非 root で実行する +- worker は起動時にマイグレーションと停止中ジョブの `RequeueStale` を行い、登録番組の確認も直ちに1回行う。SIGTERM では yt-dlp と確認処理を止め、担当中ジョブを `queued` に戻してから終了する。`Dockerfile.worker` は yt-dlp 2026.08.19 を ARG で固定し、ffmpeg・ffprobe とともに非 root で実行する - 保存先の NAS 共有フォルダは dockage VM に NFS でマウントし、`worker`(書き込み)と `web`(読み取り・削除)に同じ `VIDEO_DIR` でバインドマウントする ## 外部とのつながり @@ -128,7 +133,6 @@ TVer(`https://tver.jp/`)で公開中の動画情報を取得し、選んだ - 新着 API が 100 件で打ち切りか、続きを取得できるか [要確認] - 新着のうち DRM 付きの作品の割合 [要確認] - 非公開 API と yt-dlp の TVer 対応は、TVer 側の変更で予告なく壊れうる。壊れたことを検知する手段が要る -- 登録番組の24時間ごとの確認は未実装 - NAS の準備(NFS 共有フォルダ、tver 専用 Postgres 17 コンテナ、DSM ファイアウォール)は DSM での手作業になる。DB の公開ポート(候補 15434)は本番前に実機で確認する [要確認] - DRM・地域制限の yt-dlp 失敗時の文言と終了コードは未実測 [要確認] @@ -142,5 +146,6 @@ TVer(`https://tver.jp/`)で公開中の動画情報を取得し、選んだ - 新着画面のテストは偽物の TVer クライアント・store と差し替え可能な時計を使い、表示項目、絞り込みの境界、10 分キャッシュ、取得失敗、JST 表示、登録・状態表示・待ち行列の操作を確認する - ライブラリのテストは一時ディレクトリの小さな動画を使い、Range の 206・範囲外の 416、相対パスとシンボリックリンクの検査、404、ファイル削除後の記録削除を確認する。共有のファイル名規則は `internal/media` でテストする - worker のテストは `internal/worker/testdata` の偽 yt-dlp・ffprobe を使い、本物の TVer へ接続せずに正常完了、失敗分類、取消、終了時の待機復帰、進捗の間引き、heartbeat、出力検査、移動先、作業場所の片付けを確認する +- 登録番組の定期確認は偽の TVer クライアントと差し替え可能な時計・タイマーを使い、24時間の境界、順次確認、1時間走査、失敗記録、予告の既知化、終了時の中断を確認する。store の抽出・既知化・投入・日時更新・エラー消去は embedded Postgres 17 で確認する - `env GOPATH=/private/tmp/tver-gopath GOCACHE=/private/tmp/tver-go-cache GOMODCACHE=/private/tmp/tver-go-modcache GOPROXY=off go vet ./...` で静的解析する - `DATABASE_URL`、`VIDEO_DIR`、`WEB_ADDR=127.0.0.1:18080` を設定して `go run ./cmd/web` で web を起動できる。worker は同じ `DATABASE_URL` と `VIDEO_DIR` を設定して `go run ./cmd/worker` で起動する。worker の実行は TVer への実アクセスを伴うため手動確認は少数回に留める diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 9b6dbb3..6b2e8df 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -13,6 +13,7 @@ import ( "time" "gitea.kazu634.com/kazu634/tver/internal/store" + "gitea.kazu634.com/kazu634/tver/internal/tver" workerpkg "gitea.kazu634.com/kazu634/tver/internal/worker" "github.com/jackc/pgx/v5/pgxpool" ) @@ -34,14 +35,30 @@ func main() { log.Fatal(err) } defer pool.Close() - runner, err := workerpkg.New(store.New(pool), config, log.Default()) + jobStore := store.New(pool) + runner, err := workerpkg.New(jobStore, config, log.Default()) if err != nil { log.Fatal(err) } + tverClient, err := tver.NewClient("", nil) + if err != nil { + log.Fatal(err) + } + scheduler, err := workerpkg.NewSubscriptionScheduler(jobStore, tverClient, + workerpkg.SubscriptionSchedulerConfig{PreviewMaxSeconds: tver.PreviewMaxSeconds}, log.Default()) + if err != nil { + log.Fatal(err) + } + schedulerDone := make(chan struct{}) + go func() { + defer close(schedulerDone) + scheduler.Run(ctx) + }() log.Printf("worker started (concurrency=%d)", config.Concurrency) if err := runner.Run(ctx); err != nil { log.Fatal(err) } + <-schedulerDone log.Print("worker stopped") } diff --git a/internal/store/migrations/003_subscription_check_errors.sql b/internal/store/migrations/003_subscription_check_errors.sql new file mode 100644 index 0000000..4c31825 --- /dev/null +++ b/internal/store/migrations/003_subscription_check_errors.sql @@ -0,0 +1,6 @@ +ALTER TABLE subscriptions + ADD COLUMN last_check_error_at TIMESTAMPTZ, + ADD COLUMN last_check_error TEXT; + +CREATE INDEX subscriptions_due_idx + ON subscriptions (last_checked_at) WHERE enabled; diff --git a/internal/store/store.go b/internal/store/store.go index 3c8667c..3f49f3f 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -59,6 +59,8 @@ type Subscription struct { Enabled bool CreatedAt time.Time LastCheckedAt *time.Time + LastCheckErrorAt *time.Time + LastCheckError *string JobCount int64 } @@ -301,6 +303,7 @@ func (s *Store) DeleteSubscription(ctx context.Context, seriesID string) (bool, func (s *Store) ListSubscriptions(ctx context.Context) ([]Subscription, error) { rows, err := s.pool.Query(ctx, `SELECT s.series_id, s.series_title, s.enabled, s.created_at, s.last_checked_at, + s.last_check_error_at, s.last_check_error, (SELECT count(*) FROM download_jobs j WHERE j.series_id=s.series_id) FROM subscriptions s ORDER BY s.created_at DESC, s.series_id`) if err != nil { @@ -310,7 +313,8 @@ func (s *Store) ListSubscriptions(ctx context.Context) ([]Subscription, error) { var subs []Subscription for rows.Next() { var sub Subscription - if err := rows.Scan(&sub.SeriesID, &sub.SeriesTitle, &sub.Enabled, &sub.CreatedAt, &sub.LastCheckedAt, &sub.JobCount); err != nil { + if err := rows.Scan(&sub.SeriesID, &sub.SeriesTitle, &sub.Enabled, &sub.CreatedAt, &sub.LastCheckedAt, + &sub.LastCheckErrorAt, &sub.LastCheckError, &sub.JobCount); err != nil { return nil, err } subs = append(subs, sub) diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 401cfc6..0b10b29 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -113,7 +113,7 @@ func TestMigrate(t *testing.T) { if err := pool.QueryRow(ctx, `SELECT count(*), min(applied_at) FROM schema_migrations`).Scan(&count, &second); err != nil { t.Fatal(err) } - if count != 2 || !first.Equal(second) { + if count != 3 || !first.Equal(second) { t.Fatalf("再適用で履歴が変わった: %d %v %v", count, first, second) } } @@ -138,7 +138,7 @@ func TestMigrateConcurrent(t *testing.T) { if err := pool.QueryRow(context.Background(), `SELECT count(*) FROM schema_migrations`).Scan(&count); err != nil { t.Fatal(err) } - if count != 2 { + if count != 3 { t.Fatalf("適用件数: %d", count) } } @@ -166,7 +166,7 @@ func TestMigrateAddsKnownEpisodesToExistingDatabase(t *testing.T) { t.Fatal(err) } var count int - if err := pool.QueryRow(ctx, `SELECT count(*) FROM schema_migrations`).Scan(&count); err != nil || count != 2 { + if err := pool.QueryRow(ctx, `SELECT count(*) FROM schema_migrations`).Scan(&count); err != nil || count != 3 { t.Fatalf("移行履歴: %d %v", count, err) } if _, err := pool.Exec(ctx, `INSERT INTO known_episodes (series_id, episode_id) VALUES ('existing', 'ep1')`); err != nil { @@ -180,6 +180,47 @@ func TestMigrateAddsKnownEpisodesToExistingDatabase(t *testing.T) { } } +func TestMigrateAddsSubscriptionCheckErrorsToVersionTwoDatabase(t *testing.T) { + pool := testPool(t, false) + ctx := context.Background() + for _, name := range []string{"migrations/001_initial.sql", "migrations/002_known_episodes.sql"} { + sql, err := migrations.ReadFile(name) + if err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, string(sql), pgx.QueryExecModeSimpleProtocol); err != nil { + t.Fatal(err) + } + } + if _, err := pool.Exec(ctx, `CREATE TABLE schema_migrations (version TEXT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now())`); err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, `INSERT INTO schema_migrations (version) VALUES ('migrations/001_initial.sql'), ('migrations/002_known_episodes.sql')`); err != nil { + t.Fatal(err) + } + checked := time.Date(2026, 10, 1, 2, 3, 4, 0, time.UTC) + if _, err := pool.Exec(ctx, `INSERT INTO subscriptions (series_id, series_title, last_checked_at) VALUES ('existing-v2', '既存番組', $1)`, checked); err != nil { + t.Fatal(err) + } + if err := Migrate(ctx, pool); err != nil { + t.Fatal(err) + } + var title string + var last time.Time + var errorAt, errorMessage any + if err := pool.QueryRow(ctx, `SELECT series_title, last_checked_at, last_check_error_at, last_check_error FROM subscriptions WHERE series_id='existing-v2'`).Scan(&title, &last, &errorAt, &errorMessage); err != nil || title != "既存番組" || !last.Equal(checked) || errorAt != nil || errorMessage != nil { + t.Fatalf("v2データ保持と追加列: title=%q last=%v errorAt=%v error=%v err=%v", title, last, errorAt, errorMessage, err) + } + var indexDef string + if err := pool.QueryRow(ctx, `SELECT indexdef FROM pg_indexes WHERE schemaname=current_schema() AND indexname='subscriptions_due_idx'`).Scan(&indexDef); err != nil || !strings.Contains(indexDef, "last_checked_at") || !strings.Contains(indexDef, "WHERE enabled") { + t.Fatalf("期限抽出索引: %q %v", indexDef, err) + } + var versions int + if err := pool.QueryRow(ctx, `SELECT count(*) FROM schema_migrations`).Scan(&versions); err != nil || versions != 3 { + t.Fatalf("移行履歴: %d %v", versions, err) + } +} + func TestDownloadJobIndexes(t *testing.T) { pool := testPool(t, true) rows, err := pool.Query(context.Background(), `SELECT indexname, indexdef FROM pg_indexes WHERE schemaname=current_schema() AND tablename='download_jobs'`) diff --git a/internal/store/subscriptions.go b/internal/store/subscriptions.go index f4c670e..c1e912c 100644 --- a/internal/store/subscriptions.go +++ b/internal/store/subscriptions.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "time" "github.com/jackc/pgx/v5" ) @@ -66,6 +67,95 @@ func (s *Store) QueueUnseenEpisodes(ctx context.Context, seriesID string, episod return count, nil } +// ListDueSubscriptions は有効で、cutoff 以前に確認された(または未確認の)番組を返す。 +func (s *Store) ListDueSubscriptions(ctx context.Context, cutoff time.Time) ([]Subscription, error) { + rows, err := s.pool.Query(ctx, `SELECT series_id, series_title, enabled, created_at, last_checked_at, + last_check_error_at, last_check_error, 0::bigint + FROM subscriptions + WHERE enabled AND (last_checked_at IS NULL OR last_checked_at <= $1) + ORDER BY last_checked_at ASC NULLS FIRST, created_at, series_id`, cutoff) + if err != nil { + return nil, err + } + defer rows.Close() + var result []Subscription + for rows.Next() { + var sub Subscription + if err := rows.Scan(&sub.SeriesID, &sub.SeriesTitle, &sub.Enabled, &sub.CreatedAt, + &sub.LastCheckedAt, &sub.LastCheckErrorAt, &sub.LastCheckError, &sub.JobCount); err != nil { + return nil, err + } + result = append(result, sub) + } + return result, rows.Err() +} + +// CompleteSubscriptionCheck は全公開回を既知にし、しきい値を超える未確認回だけを投入する。 +// 最終確認日時の更新と前回エラーの消去も同じ取引で行う。 +func (s *Store) CompleteSubscriptionCheck(ctx context.Context, seriesID string, episodes []Episode, previewMaxSeconds int, checkedAt time.Time) (int, error) { + if previewMaxSeconds < 0 { + return 0, errors.New("preview threshold must not be negative") + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return 0, err + } + defer tx.Rollback(ctx) + var enabled bool + err = tx.QueryRow(ctx, `SELECT enabled FROM subscriptions WHERE series_id=$1 FOR UPDATE`, seriesID).Scan(&enabled) + if errors.Is(err, pgx.ErrNoRows) { + return 0, ErrInvalidTransition + } + if err != nil { + return 0, err + } + if !enabled { + return 0, ErrSubscriptionPaused + } + queued := 0 + for _, ep := range episodes { + if ep.ID == "" || ep.SeriesID != seriesID || ep.DurationSeconds < 0 { + return 0, fmt.Errorf("invalid episode identity for series %s", seriesID) + } + var inserted string + err := tx.QueryRow(ctx, `INSERT INTO known_episodes (series_id, episode_id) + VALUES ($1,$2) ON CONFLICT DO NOTHING RETURNING episode_id`, seriesID, ep.ID).Scan(&inserted) + if errors.Is(err, pgx.ErrNoRows) { + continue + } + if err != nil { + return 0, err + } + if ep.DurationSeconds <= previewMaxSeconds { + continue + } + tag, err := tx.Exec(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,'subscription') + ON CONFLICT (episode_id) DO NOTHING`, ep.ID, seriesID, ep.SeriesTitle, ep.Title, + ep.BroadcasterName, ep.DurationSeconds, ep.EndAt, ep.ThumbnailPath) + if err != nil { + return 0, err + } + queued += int(tag.RowsAffected()) + } + if _, err := tx.Exec(ctx, `UPDATE subscriptions SET last_checked_at=$2, + last_check_error_at=NULL, last_check_error=NULL WHERE series_id=$1`, seriesID, checkedAt); err != nil { + return 0, err + } + if err := tx.Commit(ctx); err != nil { + return 0, err + } + return queued, nil +} + +// RecordSubscriptionCheckFailure は取得失敗を記録し、最終確認日時は変更しない。 +func (s *Store) RecordSubscriptionCheckFailure(ctx context.Context, seriesID string, failedAt time.Time, message string) error { + tag, err := s.pool.Exec(ctx, `UPDATE subscriptions SET last_check_error_at=$2, last_check_error=$3 + WHERE series_id=$1 AND enabled`, seriesID, failedAt, message) + return transitionResult(tag.RowsAffected(), err) +} + func addKnownAndMaybeQueue(ctx context.Context, tx pgx.Tx, seriesID string, episodes []Episode, queue bool) (int, error) { count := 0 for _, ep := range episodes { diff --git a/internal/store/subscriptions_test.go b/internal/store/subscriptions_test.go index 8b24c64..e362d49 100644 --- a/internal/store/subscriptions_test.go +++ b/internal/store/subscriptions_test.go @@ -5,8 +5,144 @@ import ( "errors" "sync" "testing" + "time" ) +func TestListDueSubscriptionsBoundaryAndPaused(t *testing.T) { + pool := testPool(t, true) + s := New(pool) + ctx := context.Background() + now := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC) + for _, id := range []string{"null", "old", "boundary", "fresh", "paused"} { + if created, err := s.RegisterSubscription(ctx, id, "番組 "+id, nil, false); err != nil || !created { + t.Fatalf("登録 %s: %t %v", id, created, err) + } + } + updates := map[string]*time.Time{ + "null": nil, "old": timePointer(now.Add(-25 * time.Hour)), + "boundary": timePointer(now.Add(-24 * time.Hour)), "fresh": timePointer(now.Add(-24*time.Hour + time.Microsecond)), + "paused": timePointer(now.Add(-48 * time.Hour)), + } + for id, checked := range updates { + if _, err := pool.Exec(ctx, `UPDATE subscriptions SET last_checked_at=$2 WHERE series_id=$1`, id, checked); err != nil { + t.Fatal(err) + } + } + if err := s.SetSubscriptionEnabled(ctx, "paused", false); err != nil { + t.Fatal(err) + } + due, err := s.ListDueSubscriptions(ctx, now.Add(-24*time.Hour)) + if err != nil { + t.Fatal(err) + } + if len(due) != 3 || due[0].SeriesID != "null" || due[1].SeriesID != "old" || due[2].SeriesID != "boundary" { + t.Fatalf("24時間境界と一時停止: %+v", due) + } +} + +func TestCompleteSubscriptionCheckKnownPreviewErrorAndAtomicity(t *testing.T) { + pool := testPool(t, true) + s := New(pool) + ctx := context.Background() + if created, err := s.RegisterSubscription(ctx, "series-check", "番組", []Episode{seriesEpisode("known", "series-check")}, false); err != nil || !created { + t.Fatalf("登録: %t %v", created, err) + } + before := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC) + failedAt := time.Date(2026, 10, 3, 1, 2, 3, 0, time.UTC) + if _, err := pool.Exec(ctx, `UPDATE subscriptions SET last_checked_at=$2 WHERE series_id=$1`, "series-check", before); err != nil { + t.Fatal(err) + } + if err := s.RecordSubscriptionCheckFailure(ctx, "series-check", failedAt, "network token=[REDACTED]"); err != nil { + t.Fatal(err) + } + var unchanged time.Time + if err := pool.QueryRow(ctx, `SELECT last_checked_at FROM subscriptions WHERE series_id='series-check'`).Scan(&unchanged); err != nil || !unchanged.Equal(before) { + t.Fatalf("失敗時に最終確認を更新: %v %v", unchanged, err) + } + checkedAt := time.Date(2026, 10, 3, 4, 5, 6, 0, time.UTC) + preview := seriesEpisode("preview", "series-check") + preview.DurationSeconds = 180 + normal := seriesEpisode("normal", "series-check") + normal.DurationSeconds = 181 + count, err := s.CompleteSubscriptionCheck(ctx, "series-check", []Episode{ + seriesEpisode("known", "series-check"), preview, normal, + }, 180, checkedAt) + if err != nil || count != 1 { + t.Fatalf("確認完了: count=%d err=%v", count, err) + } + unknown, err := s.UnknownEpisodeIDs(ctx, "series-check", []string{"known", "preview", "normal"}) + if err != nil || len(unknown) != 0 { + t.Fatalf("予告を含む既知化: %v %v", unknown, err) + } + var jobs int + if err := pool.QueryRow(ctx, `SELECT count(*) FROM download_jobs WHERE series_id='series-check'`).Scan(&jobs); err != nil || jobs != 1 { + t.Fatalf("予告の投入除外: %d %v", jobs, err) + } + var last time.Time + var errorAt, errorMessage any + if err := pool.QueryRow(ctx, `SELECT last_checked_at, last_check_error_at, last_check_error FROM subscriptions WHERE series_id='series-check'`).Scan(&last, &errorAt, &errorMessage); err != nil || !last.Equal(checkedAt) || errorAt != nil || errorMessage != nil { + t.Fatalf("成功時の日時・エラー消去: %v %v %v %v", last, errorAt, errorMessage, err) + } + + bad := seriesEpisode("bad", "series-check") + bad.DurationSeconds = -1 + if _, err := s.CompleteSubscriptionCheck(ctx, "series-check", []Episode{seriesEpisode("rollback", "series-check"), bad}, 180, checkedAt.Add(time.Hour)); err == nil { + t.Fatal("不正な回を含む確認が成功した") + } + unknown, err = s.UnknownEpisodeIDs(ctx, "series-check", []string{"rollback"}) + if err != nil || len(unknown) != 1 { + t.Fatalf("失敗時の既知記録を巻き戻さない: %v %v", unknown, err) + } + if err := pool.QueryRow(ctx, `SELECT last_checked_at FROM subscriptions WHERE series_id='series-check'`).Scan(&last); err != nil || !last.Equal(checkedAt) { + t.Fatalf("失敗時の最終確認を巻き戻さない: %v %v", last, err) + } + if err := s.SetSubscriptionEnabled(ctx, "series-check", false); err != nil { + t.Fatal(err) + } + if _, err := s.CompleteSubscriptionCheck(ctx, "series-check", nil, 180, checkedAt.Add(time.Hour)); !errors.Is(err, ErrSubscriptionPaused) { + t.Fatalf("一時停止中の確認完了: %v", err) + } + if err := s.RecordSubscriptionCheckFailure(ctx, "series-check", failedAt, "ignored"); !errors.Is(err, ErrInvalidTransition) { + t.Fatalf("一時停止中のエラー記録: %v", err) + } +} + +func TestCompleteSubscriptionCheckKeepsExistingManualJob(t *testing.T) { + pool := testPool(t, true) + s := New(pool) + ctx := context.Background() + if created, err := s.RegisterSubscription(ctx, "series-manual", "番組", nil, false); err != nil || !created { + t.Fatalf("登録: %t %v", created, err) + } + episode := seriesEpisode("manual-existing", "series-manual") + manual, _, err := s.Enqueue(ctx, episode, "manual") + if err != nil { + t.Fatal(err) + } + checkedAt := time.Date(2026, 10, 3, 4, 5, 6, 0, time.UTC) + queued, err := s.CompleteSubscriptionCheck(ctx, "series-manual", []Episode{episode, episode}, 180, checkedAt) + if err != nil || queued != 0 { + t.Fatalf("既存手動ジョブの確認: queued=%d err=%v", queued, err) + } + got, err := s.GetJob(ctx, manual.ID) + if err != nil || got.Source != "manual" { + t.Fatalf("既存手動ジョブを変更した: %+v %v", got, err) + } + var jobs, known int + if err := pool.QueryRow(ctx, `SELECT count(*) FROM download_jobs WHERE episode_id=$1`, episode.ID).Scan(&jobs); err != nil || jobs != 1 { + t.Fatalf("ジョブが二重化した: %d %v", jobs, err) + } + if err := pool.QueryRow(ctx, `SELECT count(*) FROM known_episodes WHERE series_id=$1 AND episode_id=$2`, episode.SeriesID, episode.ID).Scan(&known); err != nil || known != 1 { + t.Fatalf("既知記録: %d %v", known, err) + } + var last time.Time + if err := pool.QueryRow(ctx, `SELECT last_checked_at FROM subscriptions WHERE series_id=$1`, episode.SeriesID).Scan(&last); err != nil || !last.Equal(checkedAt) { + t.Fatalf("最終確認日時: %v %v", last, err) + } +} + +func timePointer(value time.Time) *time.Time { return &value } + func seriesEpisode(id, seriesID string) Episode { ep := sampleEpisode(id) ep.SeriesID = seriesID diff --git a/internal/tver/client.go b/internal/tver/client.go index aedf647..6e52fba 100644 --- a/internal/tver/client.go +++ b/internal/tver/client.go @@ -12,7 +12,10 @@ import ( "time" ) -const DefaultBaseURL = "https://platform-api.tver.jp" +const ( + DefaultBaseURL = "https://platform-api.tver.jp" + PreviewMaxSeconds = 180 +) type Session struct { PlatformUID string `json:"platform_uid"` diff --git a/internal/web/newer.go b/internal/web/newer.go index a1b1bc7..bd0603b 100644 --- a/internal/web/newer.go +++ b/internal/web/newer.go @@ -14,8 +14,7 @@ import ( ) const ( - previewMaxSeconds = 180 - newerCacheTTL = 10 * time.Minute + newerCacheTTL = 10 * time.Minute ) var platformTokenPattern = regexp.MustCompile(`(?i)(platform_token=)[^&\s]+`) @@ -137,7 +136,7 @@ func (s *server) newerView(ctx context.Context, options newerOptions) newerView } visible := make([]tver.Episode, 0, len(episodes)) for _, episode := range episodes { - if options.hidePreviews && episode.Duration <= previewMaxSeconds { + if options.hidePreviews && episode.Duration <= tver.PreviewMaxSeconds { continue } visible = append(visible, episode) diff --git a/internal/web/subscriptions.go b/internal/web/subscriptions.go index 20673f1..c3f3c26 100644 --- a/internal/web/subscriptions.go +++ b/internal/web/subscriptions.go @@ -15,6 +15,7 @@ type subscriptionActionView struct { type subscriptionView struct { ID, Title, CreatedAtText, LastCheckedText string + CheckErrorAtText, CheckErrorText string Enabled bool Jobs int64 } @@ -136,6 +137,10 @@ func (s *server) loadSubscriptions(ctx context.Context) (subscriptionsView, erro if sub.LastCheckedAt != nil { item.LastCheckedText = sub.LastCheckedAt.In(s.zone).Format("2006-01-02 15:04 MST") } + if sub.LastCheckErrorAt != nil && sub.LastCheckError != nil { + item.CheckErrorAtText = sub.LastCheckErrorAt.In(s.zone).Format("2006-01-02 15:04 MST") + item.CheckErrorText = *sub.LastCheckError + } view.Items = append(view.Items, item) } return view, nil diff --git a/internal/web/subscriptions_test.go b/internal/web/subscriptions_test.go index 7b6d7e9..b81c83a 100644 --- a/internal/web/subscriptions_test.go +++ b/internal/web/subscriptions_test.go @@ -159,9 +159,12 @@ func TestSubscriptionMutationsWithoutHtmxRedirect(t *testing.T) { func TestSubscriptionsPagePauseResumeDeleteAndEscaping(t *testing.T) { checked := time.Date(2026, 10, 4, 2, 30, 0, 0, time.UTC) + failedAt := time.Date(2026, 10, 4, 3, 45, 0, 0, time.UTC) + checkError := ` token=[REDACTED]` jobs := &fakeJobStore{subscriptions: map[string]store.Subscription{ "sr1": {SeriesID: "sr1", SeriesTitle: ``, Enabled: true, - CreatedAt: time.Date(2026, 10, 3, 0, 0, 0, 0, time.UTC), LastCheckedAt: &checked, JobCount: 3}, + CreatedAt: time.Date(2026, 10, 3, 0, 0, 0, 0, time.UTC), LastCheckedAt: &checked, + LastCheckErrorAt: &failedAt, LastCheckError: &checkError, JobCount: 3}, "sr2": {SeriesID: "sr2", SeriesTitle: "別番組", Enabled: false, JobCount: 0}, }} handler := newTestHandlerWithStore(&fakeFetcher{}, jobs, nil) @@ -170,13 +173,13 @@ func TestSubscriptionsPagePauseResumeDeleteAndEscaping(t *testing.T) { t.Fatalf("登録番組ページ: %d", page.Code) } body := page.Body.String() - for _, want := range []string{"<script>", "登録日時: 2026-10-03 09:00 JST", "最終確認: 2026-10-04 11:30 JST", "最終確認: 未確認", "関連ジョブ: 3件", "有効", "一時停止", `hx-confirm="番組の登録を解除しますか?"`} { + for _, want := range []string{"<script>", "登録日時: 2026-10-03 09:00 JST", "最終確認: 2026-10-04 11:30 JST", "最終確認: 未確認", "最後の確認エラー(2026-10-04 12:45 JST)", `<script>alert("error")</script> token=[REDACTED]`, "関連ジョブ: 3件", "有効", "一時停止", `hx-confirm="番組の登録を解除しますか?"`} { if !strings.Contains(body, want) { t.Errorf("登録番組の情報 %q がない", want) } } - if strings.Contains(body, ``) { - t.Fatal("番組名がエスケープされていない") + if strings.Contains(body, `