Pete/internal/scheduler/roundrobin.go

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