Pete/internal/scheduler/roundrobin.go

// 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")
}