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, `