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]) }