tver/internal/worker/subscriptions.go

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