// Package scheduler provides the round-robin posting scheduler: when // enabled, Pete posts one classified story per interval, cycling through // sources in config order. Empty sources are skipped and the rotation // pointer advances to the source that actually posted. package scheduler import ( "context" "log/slog" "time" "pete/internal/config" "pete/internal/ingestion" "pete/internal/poster" "pete/internal/storage" ) // Enqueuer is the subset of *poster.Queue the scheduler depends on. Defined // as an interface so tests can substitute a capturing fake without standing // up a Matrix client. type Enqueuer interface { Enqueue(item poster.QueueItem) } // RoundRobin paces posts across sources. One tick = at most one Enqueue. type RoundRobin struct { sources []config.SourceConfig // only enabled sources, in config order interval time.Duration queue Enqueuer } // New filters to enabled sources and returns nil if the rotation has nothing // to work with (callers should fall back to immediate posting in that case). func New(sources []config.SourceConfig, intervalHours int, queue Enqueuer) *RoundRobin { var enabled []config.SourceConfig for _, s := range sources { if s.Enabled { enabled = append(enabled, s) } } if len(enabled) == 0 { return nil } return &RoundRobin{ sources: enabled, interval: time.Duration(intervalHours) * time.Hour, queue: queue, } } // Start runs the scheduler until ctx is cancelled. Honors persisted // last_tick_at so a restart partway through an interval waits the // remainder rather than firing immediately. func (r *RoundRobin) Start(ctx context.Context) { _, lastTickAt, err := storage.GetRoundRobinState() if err != nil { slog.Error("round-robin: failed to load state, starting fresh", "err", err) } // Compute time until the next scheduled tick. var initialDelay time.Duration if lastTickAt == 0 { initialDelay = r.interval } else { nextTick := time.Unix(lastTickAt, 0).Add(r.interval) initialDelay = time.Until(nextTick) if initialDelay < 0 { initialDelay = 0 } } slog.Info("round-robin scheduler started", "sources", len(r.sources), "interval", r.interval, "first_tick_in", initialDelay, ) timer := time.NewTimer(initialDelay) defer timer.Stop() for { select { case <-ctx.Done(): slog.Info("round-robin scheduler stopping") return case <-timer.C: r.tick() timer.Reset(r.interval) } } } // tick scans sources starting after the last-posted one and enqueues the // first postable story it finds. func (r *RoundRobin) tick() { lastSource, _, err := storage.GetRoundRobinState() if err != nil { slog.Error("round-robin: failed to load state", "err", err) // continue with empty lastSource — start from index 0 } startIdx := 0 if lastSource != "" { for i, s := range r.sources { if s.Name == lastSource { startIdx = (i + 1) % len(r.sources) break } } } for offset := 0; offset < len(r.sources); offset++ { idx := (startIdx + offset) % len(r.sources) src := r.sources[idx] story, err := storage.GetNewestPostableStory(src.Name) if err != nil { slog.Error("round-robin: query failed, skipping source", "source", src.Name, "err", err) continue } if story == nil { slog.Debug("round-robin: source empty, advancing", "source", src.Name) continue } imageURL := "" if story.ImageURL != "" && ingestion.ValidateImageURL(story.ImageURL) { imageURL = story.ImageURL } r.queue.Enqueue(poster.QueueItem{ GUID: story.GUID, Headline: story.Headline, Lede: story.Lede, ImageURL: imageURL, ArticleURL: story.ArticleURL, Source: story.Source, Channel: story.Channel, Platforms: storage.UnmarshalPlatforms(story.Platforms), }) storage.SetRoundRobinState(src.Name, time.Now().Unix()) slog.Info("round-robin: enqueued", "source", src.Name, "guid", story.GUID, "channel", story.Channel) return } // Nothing to post this tick — record the tick timestamp but leave // last_source untouched so the next tick resumes from the same offset. storage.SetRoundRobinState(lastSource, time.Now().Unix()) slog.Info("round-robin: no postable stories this tick") }