// Package scheduler provides the round-robin posting scheduler: when // enabled, Pete posts one classified story per interval, cycling through // channels in sorted order. Empty channels are skipped and the rotation // pointer advances to the channel that actually posted. package scheduler import ( "context" "log/slog" "sort" "time" "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 channels. One tick = at most one Enqueue. type RoundRobin struct { channels []string // routable channel names, in stable (sorted) order interval time.Duration queue Enqueuer } // New returns a scheduler that rotates across the given channel names. // Returns nil if the channel list is empty (callers should fall back to // immediate posting in that case). Channel order is normalized so behavior // is independent of map iteration order. func New(channels []string, intervalHours int, queue Enqueuer) *RoundRobin { if len(channels) == 0 { return nil } sorted := make([]string, len(channels)) copy(sorted, channels) sort.Strings(sorted) return &RoundRobin{ channels: sorted, 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) } 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", "channels", len(r.channels), "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 channels starting after the last-posted one and enqueues the // newest unposted story routed to the first channel with anything available. func (r *RoundRobin) tick() { lastChannel, _, err := storage.GetRoundRobinState() if err != nil { slog.Error("round-robin: failed to load state", "err", err) } startIdx := 0 if lastChannel != "" { for i, ch := range r.channels { if ch == lastChannel { startIdx = (i + 1) % len(r.channels) break } } } for offset := 0; offset < len(r.channels); offset++ { idx := (startIdx + offset) % len(r.channels) ch := r.channels[idx] story, err := storage.GetNewestPostableStoryByChannel(ch) if err != nil { slog.Error("round-robin: query failed, skipping channel", "channel", ch, "err", err) continue } if story == nil { slog.Debug("round-robin: channel empty, advancing", "channel", ch) 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(ch, time.Now().Unix()) slog.Info("round-robin: enqueued", "channel", ch, "guid", story.GUID, "source", story.Source) return } // Nothing to post this tick — record the tick timestamp but leave // last_channel untouched so the next tick resumes from the same offset. storage.SetRoundRobinState(lastChannel, time.Now().Unix()) slog.Info("round-robin: no postable stories this tick") }