登録番組の定期確認を追加（24 時間ごと、予告の除外、確認エラーの記録）

2026-10-03 23:42:25 +09:00
parent 82536c6398
commit 944086c8ae
14 changed files with 739 additions and 18 deletions
@@ -0,0 +1,156 @@
+package worker
+
+import (
+	"context"
+	"errors"
+	"log"
+	"strings"
+	"time"
+	"unicode/utf8"
+
+	"gitea.kazu634.com/kazu634/tver/internal/store"
+	"gitea.kazu634.com/kazu634/tver/internal/tver"
+)
+
+type SubscriptionStore interface {
+	ListDueSubscriptions(context.Context, time.Time) ([]store.Subscription, error)
+	CompleteSubscriptionCheck(context.Context, string, []store.Episode, int, time.Time) (int, error)
+	RecordSubscriptionCheckFailure(context.Context, string, time.Time, string) error
+}
+
+type SeriesFetcher interface {
+	FetchSeriesEpisodes(context.Context, string) ([]tver.Episode, error)
+}
+
+type SubscriptionSchedulerConfig struct {
+	ScanInterval      time.Duration
+	DueAfter          time.Duration
+	SeriesInterval    time.Duration
+	PreviewMaxSeconds int
+	Now               func() time.Time
+	After             func(time.Duration) <-chan time.Time
+}
+
+type SubscriptionScheduler struct {
+	store   SubscriptionStore
+	fetcher SeriesFetcher
+	config  SubscriptionSchedulerConfig
+	logger  *log.Logger
+}
+
+func NewSubscriptionScheduler(subscriptions SubscriptionStore, fetcher SeriesFetcher, config SubscriptionSchedulerConfig, logger *log.Logger) (*SubscriptionScheduler, error) {
+	if subscriptions == nil || fetcher == nil {
+		return nil, errors.New("subscription store and fetcher are required")
+	}
+	if config.ScanInterval <= 0 {
+		config.ScanInterval = time.Hour
+	}
+	if config.DueAfter <= 0 {
+		config.DueAfter = 24 * time.Hour
+	}
+	if config.SeriesInterval <= 0 {
+		config.SeriesInterval = 3 * time.Second
+	}
+	if config.PreviewMaxSeconds == 0 {
+		config.PreviewMaxSeconds = tver.PreviewMaxSeconds
+	}
+	if config.PreviewMaxSeconds < 0 {
+		return nil, errors.New("preview threshold must not be negative")
+	}
+	if config.Now == nil {
+		config.Now = time.Now
+	}
+	if config.After == nil {
+		config.After = time.After
+	}
+	if logger == nil {
+		logger = log.Default()
+	}
+	return &SubscriptionScheduler{store: subscriptions, fetcher: fetcher, config: config, logger: logger}, nil
+}
+
+// Run は起動直後に確認し、その後は ScanInterval ごとに繰り返す。
+func (s *SubscriptionScheduler) Run(ctx context.Context) {
+	for {
+		s.checkDue(ctx)
+		if !s.wait(ctx, s.config.ScanInterval) {
+			return
+		}
+	}
+}
+
+func (s *SubscriptionScheduler) checkDue(ctx context.Context) {
+	now := s.config.Now()
+	subscriptions, err := s.store.ListDueSubscriptions(ctx, now.Add(-s.config.DueAfter))
+	if err != nil {
+		if ctx.Err() == nil {
+			s.logger.Printf("確認対象の登録番組を取得できません: %v", err)
+		}
+		return
+	}
+	for index, subscription := range subscriptions {
+		if index > 0 && !s.wait(ctx, s.config.SeriesInterval) {
+			return
+		}
+		episodes, err := s.fetcher.FetchSeriesEpisodes(ctx, subscription.SeriesID)
+		if err != nil {
+			if ctx.Err() != nil {
+				return
+			}
+			message := shortError(err)
+			if recordErr := s.store.RecordSubscriptionCheckFailure(ctx, subscription.SeriesID, s.config.Now(), message); recordErr != nil &&
+				!errors.Is(recordErr, store.ErrInvalidTransition) {
+				s.logger.Printf("登録番組 %s の確認エラーを記録できません: %v", subscription.SeriesID, recordErr)
+			}
+			s.logger.Printf("登録番組 %s の公開中エピソードを取得できません: %s", subscription.SeriesID, message)
+			continue
+		}
+		items := make([]store.Episode, 0, len(episodes))
+		for _, episode := range episodes {
+			items = append(items, subscriptionEpisode(episode))
+		}
+		queued, err := s.store.CompleteSubscriptionCheck(ctx, subscription.SeriesID, items,
+			s.config.PreviewMaxSeconds, s.config.Now())
+		if err != nil {
+			if ctx.Err() != nil {
+				return
+			}
+			if !errors.Is(err, store.ErrSubscriptionPaused) && !errors.Is(err, store.ErrInvalidTransition) {
+				s.logger.Printf("登録番組 %s の確認結果を保存できません: %v", subscription.SeriesID, err)
+			}
+			continue
+		}
+		s.logger.Printf("登録番組 %s を確認しました（新規ジョブ=%d）", subscription.SeriesID, queued)
+	}
+}
+
+func (s *SubscriptionScheduler) wait(ctx context.Context, duration time.Duration) bool {
+	select {
+	case <-ctx.Done():
+		return false
+	case <-s.config.After(duration):
+		return true
+	}
+}
+
+func subscriptionEpisode(episode tver.Episode) store.Episode {
+	item := store.Episode{ID: episode.ID, SeriesID: episode.SeriesID, SeriesTitle: episode.SeriesTitle,
+		Title: episode.Title, BroadcasterName: episode.BroadcasterName,
+		DurationSeconds: int(episode.Duration), ThumbnailPath: episode.ThumbnailPath}
+	if episode.EndAt > 0 {
+		end := time.Unix(episode.EndAt, 0)
+		item.EndAt = &end
+	}
+	return item
+}
+
+func shortError(err error) string {
+	message := redact(err.Error())
+	message = strings.Join(strings.Fields(message), " ")
+	const maxRunes = 512
+	if utf8.RuneCountInString(message) <= maxRunes {
+		return message
+	}
+	runes := []rune(message)
+	return string(runes[:maxRunes])
+}
@@ -0,0 +1,253 @@
+package worker
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"log"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"gitea.kazu634.com/kazu634/tver/internal/store"
+	"gitea.kazu634.com/kazu634/tver/internal/tver"
+)
+
+type schedulerCompletion struct {
+	seriesID string
+	episodes []store.Episode
+	limit    int
+	at       time.Time
+}
+
+type schedulerFailure struct {
+	seriesID, message string
+	at                time.Time
+}
+
+type fakeSubscriptionStore struct {
+	mu          sync.Mutex
+	due         []store.Subscription
+	cutoffs     []time.Time
+	completions []schedulerCompletion
+	failures    []schedulerFailure
+	listErr     error
+	completeErr error
+}
+
+func (f *fakeSubscriptionStore) ListDueSubscriptions(_ context.Context, cutoff time.Time) ([]store.Subscription, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.cutoffs = append(f.cutoffs, cutoff)
+	return append([]store.Subscription(nil), f.due...), f.listErr
+}
+
+func (f *fakeSubscriptionStore) CompleteSubscriptionCheck(_ context.Context, seriesID string, episodes []store.Episode, limit int, at time.Time) (int, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.completions = append(f.completions, schedulerCompletion{seriesID, append([]store.Episode(nil), episodes...), limit, at})
+	return len(episodes), f.completeErr
+}
+
+func (f *fakeSubscriptionStore) RecordSubscriptionCheckFailure(_ context.Context, seriesID string, at time.Time, message string) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.failures = append(f.failures, schedulerFailure{seriesID, message, at})
+	return nil
+}
+
+type fakeSeriesFetcher struct {
+	mu       sync.Mutex
+	calls    []string
+	episodes map[string][]tver.Episode
+	errors   map[string]error
+	block    bool
+	started  chan string
+}
+
+func (f *fakeSeriesFetcher) FetchSeriesEpisodes(ctx context.Context, seriesID string) ([]tver.Episode, error) {
+	f.mu.Lock()
+	f.calls = append(f.calls, seriesID)
+	block := f.block
+	started := f.started
+	episodes := append([]tver.Episode(nil), f.episodes[seriesID]...)
+	err := f.errors[seriesID]
+	f.mu.Unlock()
+	if started != nil {
+		started <- seriesID
+	}
+	if block {
+		<-ctx.Done()
+		return nil, ctx.Err()
+	}
+	return episodes, err
+}
+
+func TestSubscriptionSchedulerChecksImmediatelyAndMapsEpisodes(t *testing.T) {
+	now := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC)
+	subscriptions := &fakeSubscriptionStore{due: []store.Subscription{{SeriesID: "one"}, {SeriesID: "two"}}}
+	fetcher := &fakeSeriesFetcher{episodes: map[string][]tver.Episode{
+		"one": {{ID: "ep1", SeriesID: "one", SeriesTitle: "番組", Title: "回", BroadcasterName: "局", Duration: 180, EndAt: 1_800_000_000, ThumbnailPath: "/image"}},
+		"two": {{ID: "ep2", SeriesID: "two", Duration: 181}},
+	}, errors: map[string]error{}}
+	var waits []time.Duration
+	after := func(duration time.Duration) <-chan time.Time {
+		waits = append(waits, duration)
+		ch := make(chan time.Time, 1)
+		ch <- now
+		return ch
+	}
+	scheduler, err := NewSubscriptionScheduler(subscriptions, fetcher, SubscriptionSchedulerConfig{
+		DueAfter: 24 * time.Hour, SeriesInterval: 4 * time.Second,
+		Now: func() time.Time { return now }, After: after,
+	}, log.New(&bytes.Buffer{}, "", 0))
+	if err != nil {
+		t.Fatal(err)
+	}
+	scheduler.checkDue(context.Background())
+	if len(subscriptions.cutoffs) != 1 || !subscriptions.cutoffs[0].Equal(now.Add(-24*time.Hour)) {
+		t.Fatalf("確認対象日時: %v", subscriptions.cutoffs)
+	}
+	if len(fetcher.calls) != 2 || fetcher.calls[0] != "one" || fetcher.calls[1] != "two" || len(waits) != 1 || waits[0] != 4*time.Second {
+		t.Fatalf("順次確認と間隔: calls=%v waits=%v", fetcher.calls, waits)
+	}
+	if len(subscriptions.completions) != 2 || subscriptions.completions[0].limit != tver.PreviewMaxSeconds || !subscriptions.completions[0].at.Equal(now) {
+		t.Fatalf("確認完了: %+v", subscriptions.completions)
+	}
+	got := subscriptions.completions[0].episodes[0]
+	if got.ID != "ep1" || got.DurationSeconds != 180 || got.EndAt == nil || got.EndAt.Unix() != 1_800_000_000 {
+		t.Fatalf("エピソード変換: %+v", got)
+	}
+}
+
+func TestShortErrorLimitsUnicodeByCharacters(t *testing.T) {
+	message := shortError(errors.New(strings.Repeat("界", 260) + "\n\t" + strings.Repeat("語", 300)))
+	if got := len([]rune(message)); got != 512 {
+		t.Fatalf("エラー文字数=%d want=512", got)
+	}
+	if strings.ContainsAny(message, "\r\n\t") || strings.Contains(message, "  ") {
+		t.Fatalf("空白を1行へ正規化していない: %q", message)
+	}
+	if !strings.HasSuffix(message, strings.Repeat("語", 251)) {
+		t.Fatalf("UTF-8文字の途中で切断または境界が不正: %q", message)
+	}
+}
+
+func TestSubscriptionSchedulerRecordsRedactedFailureWithoutCompleting(t *testing.T) {
+	now := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC)
+	subscriptions := &fakeSubscriptionStore{due: []store.Subscription{{SeriesID: "failed"}}}
+	fetcher := &fakeSeriesFetcher{errors: map[string]error{
+		"failed": errors.New("request failed\nplatform_token=secret-token Cookie: session-secret"),
+	}}
+	var logs bytes.Buffer
+	scheduler, err := NewSubscriptionScheduler(subscriptions, fetcher, SubscriptionSchedulerConfig{Now: func() time.Time { return now }}, log.New(&logs, "", 0))
+	if err != nil {
+		t.Fatal(err)
+	}
+	scheduler.checkDue(context.Background())
+	if len(subscriptions.completions) != 0 || len(subscriptions.failures) != 1 || !subscriptions.failures[0].at.Equal(now) {
+		t.Fatalf("失敗の記録: completions=%v failures=%v", subscriptions.completions, subscriptions.failures)
+	}
+	message := subscriptions.failures[0].message
+	if strings.Contains(message, "secret-token") || strings.Contains(message, "session-secret") || strings.Contains(logs.String(), "secret") || strings.Contains(message, "\n") {
+		t.Fatalf("秘密値または改行が残った: message=%q logs=%q", message, logs.String())
+	}
+}
+
+func TestSubscriptionSchedulerRunStartsImmediatelyThenUsesScanInterval(t *testing.T) {
+	now := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC)
+	subscriptions := &fakeSubscriptionStore{}
+	fetcher := &fakeSeriesFetcher{}
+	requested := make(chan time.Duration, 2)
+	tick := make(chan time.Time, 1)
+	after := func(duration time.Duration) <-chan time.Time {
+		requested <- duration
+		return tick
+	}
+	scheduler, err := NewSubscriptionScheduler(subscriptions, fetcher, SubscriptionSchedulerConfig{
+		ScanInterval: time.Hour, Now: func() time.Time { return now }, After: after,
+	}, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() { scheduler.Run(ctx); close(done) }()
+	select {
+	case duration := <-requested:
+		if duration != time.Hour {
+			t.Fatalf("走査間隔: %v", duration)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("起動直後の確認が行われなかった")
+	}
+	subscriptions.mu.Lock()
+	initialCalls := len(subscriptions.cutoffs)
+	subscriptions.mu.Unlock()
+	if initialCalls != 1 {
+		t.Fatalf("起動直後の確認回数: %d", initialCalls)
+	}
+	tick <- now
+	select {
+	case <-requested:
+	case <-time.After(time.Second):
+		t.Fatal("1時間後の確認が行われなかった")
+	}
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("終了できなかった")
+	}
+	subscriptions.mu.Lock()
+	defer subscriptions.mu.Unlock()
+	if len(subscriptions.cutoffs) != 2 {
+		t.Fatalf("確認回数: %d", len(subscriptions.cutoffs))
+	}
+}
+
+func TestSubscriptionSchedulerCancellationInterruptsFetchAndSeriesWait(t *testing.T) {
+	now := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC)
+	t.Run("fetch", func(t *testing.T) {
+		subscriptions := &fakeSubscriptionStore{due: []store.Subscription{{SeriesID: "one"}}}
+		fetcher := &fakeSeriesFetcher{block: true, started: make(chan string, 1)}
+		scheduler, _ := NewSubscriptionScheduler(subscriptions, fetcher, SubscriptionSchedulerConfig{Now: func() time.Time { return now }}, nil)
+		ctx, cancel := context.WithCancel(context.Background())
+		done := make(chan struct{})
+		go func() { scheduler.checkDue(ctx); close(done) }()
+		select {
+		case <-fetcher.started:
+		case <-time.After(time.Second):
+			t.Fatal("取得が開始されなかった")
+		}
+		cancel()
+		select {
+		case <-done:
+		case <-time.After(time.Second):
+			t.Fatal("取得中断で終了しなかった")
+		}
+		if len(subscriptions.failures) != 0 || len(subscriptions.completions) != 0 {
+			t.Fatalf("中断した番組を記録した: failures=%v completions=%v", subscriptions.failures, subscriptions.completions)
+		}
+	})
+
+	t.Run("between series", func(t *testing.T) {
+		subscriptions := &fakeSubscriptionStore{due: []store.Subscription{{SeriesID: "one"}, {SeriesID: "two"}}}
+		fetcher := &fakeSeriesFetcher{episodes: map[string][]tver.Episode{}, errors: map[string]error{}}
+		waiting := make(chan struct{})
+		never := make(chan time.Time)
+		scheduler, _ := NewSubscriptionScheduler(subscriptions, fetcher, SubscriptionSchedulerConfig{
+			Now: func() time.Time { return now }, After: func(time.Duration) <-chan time.Time { close(waiting); return never },
+		}, nil)
+		ctx, cancel := context.WithCancel(context.Background())
+		done := make(chan struct{})
+		go func() { scheduler.checkDue(ctx); close(done) }()
+		<-waiting
+		cancel()
+		<-done
+		if len(fetcher.calls) != 1 || fetcher.calls[0] != "one" {
+			t.Fatalf("番組間の中断: %v", fetcher.calls)
+		}
+	})
+}