Round-robin: rotate by channel instead of by source

Source-keyed rotation skewed toward whichever channel had the most feeds (4 of 7 sources routed to politics, so politics dominated the rotation). Channel-keyed rotation guarantees variety regardless of feed counts. Schema: round_robin_state.last_source -> last_channel, added via addColumnIfMissing so existing DBs migrate in place.
2026-05-24 19:31:27 -07:00
parent afe2ef996b
commit 87906719fa
7 changed files with 121 additions and 137 deletions
--- a/README.md
+++ b/README.md
@@ -8,7 +8,7 @@ A Matrix news bot that ingests RSS feeds from curated sources, classifies storie
 - **Two-tier classification** — Tier 1 sources use direct routing or lightweight LLM routing; Tier 2 (wire services) use a keyword gating pipeline before LLM
 - **Semantic deduplication** — GUID exact match + LLM-based cross-source duplicate detection
 - **Metered release** — per-channel post queues with minimum interval (5min) and burst cap (3/30min)
- **Round-robin mode** — opt-in pacing: one story per N hours (default 4), cycling through sources in config order, skip-and-advance over empty sources, state persists across restarts
+- **Round-robin mode** — opt-in pacing: one story per N hours (default 4), cycling through channels in sorted order, skip-and-advance over empty channels, state persists across restarts
 - **Reaction tracking** — records emoji reactions on posts for classifier tuning
 - **`!explain` via ❓ reaction** — react with `❓` (or `❔ ⁉ 🤔 ?`) on any post; Pete fetches the article body and replies in-thread with a 3-bullet Ollama-generated TL;DR
 - **`!post` on demand** — type `!post` in any configured channel room and Pete force-publishes the next queued story for that channel, bypassing min-interval, burst cap, and daily cap (canonical-URL dedup still applies); replies in-thread if the queue is empty
@@ -67,7 +67,7 @@ Environment variables can be referenced with `${VAR}` syntax in the YAML.

 ### Round-robin mode

-Set `posting.round_robin.enabled: true` to switch Pete from "post on classify" to a paced rotation. On each tick (`interval_hours`, default 4) Pete picks the newest unposted classified story from the next source in config order, posting through the existing queue. Empty sources are skipped; the rotation pointer advances to whichever source actually posted, and `last_source` / `last_tick_at` are persisted so restarts don't reset the cycle.
+Set `posting.round_robin.enabled: true` to switch Pete from "post on classify" to a paced rotation. On each tick (`interval_hours`, default 4) Pete picks the newest unposted classified story routed to the next channel in sorted order, posting through the existing queue. Empty channels are skipped; the rotation pointer advances to whichever channel actually posted, and `last_channel` / `last_tick_at` are persisted so restarts don't reset the cycle. Rotating by channel (not by source) guarantees variety even when one channel has many more feeds than the others.

 With a 4-hour cadence Pete will post at most 6 stories/day, so set `posting.daily_cap_total` to 0 (or ≥6) when enabling this — otherwise the daily cap can silently swallow a tick.

@@ -112,7 +112,7 @@ Posted message ← ❓ reaction → Explainer → Article Fetch → Ollama summa
 | `internal/matrix` | Password auth with device persistence, posting, threaded replies, reaction + message listener |
 | `internal/poster` | Per-channel metered release queue, reaction tracking, callback hook |
 | `internal/explainer` | ❓-reaction → article fetch → Ollama summary → threaded reply |
-| `internal/scheduler` | Round-robin posting scheduler: paced rotation across sources when enabled |
+| `internal/scheduler` | Round-robin posting scheduler: paced rotation across channels when enabled |

 ## Post Format

--- a/internal/scheduler/roundrobin.go
+++ b/internal/scheduler/roundrobin.go
@@ -1,15 +1,15 @@
 // 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.
+// 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/config"
 	"pete/internal/ingestion"
 	"pete/internal/poster"
 	"pete/internal/storage"
@@ -22,27 +22,26 @@ type Enqueuer interface {
 	Enqueue(item poster.QueueItem)
 }

-// RoundRobin paces posts across sources. One tick = at most one Enqueue.
+// RoundRobin paces posts across channels. One tick = at most one Enqueue.
 type RoundRobin struct {
-	sources  []config.SourceConfig // only enabled sources, in config order
+	channels []string // routable channel names, in stable (sorted) 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 {
+// 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{
-		sources:  enabled,
+		channels: sorted,
 		interval: time.Duration(intervalHours) * time.Hour,
 		queue:    queue,
 	}
@@ -57,7 +56,6 @@ func (r *RoundRobin) Start(ctx context.Context) {
 		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
@@ -69,7 +67,7 @@ func (r *RoundRobin) Start(ctx context.Context) {
 		}
 	}
 	slog.Info("round-robin scheduler started",
-		"sources", len(r.sources),
+		"channels", len(r.channels),
 		"interval", r.interval,
 		"first_tick_in", initialDelay,
 	)
@@ -89,36 +87,35 @@ func (r *RoundRobin) Start(ctx context.Context) {
 	}
 }

-// tick scans sources starting after the last-posted one and enqueues the
-// first postable story it finds.
+// 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() {
-	lastSource, _, err := storage.GetRoundRobinState()
+	lastChannel, _, 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)
+	if lastChannel != "" {
+		for i, ch := range r.channels {
+			if ch == lastChannel {
+				startIdx = (i + 1) % len(r.channels)
 				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)
+	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 source",
-				"source", src.Name, "err", err)
+			slog.Error("round-robin: query failed, skipping channel",
+				"channel", ch, "err", err)
 			continue
 		}
 		if story == nil {
-			slog.Debug("round-robin: source empty, advancing", "source", src.Name)
+			slog.Debug("round-robin: channel empty, advancing", "channel", ch)
 			continue
 		}

@@ -137,14 +134,14 @@ func (r *RoundRobin) tick() {
 			Channel:    story.Channel,
 			Platforms:  storage.UnmarshalPlatforms(story.Platforms),
 		})
-		storage.SetRoundRobinState(src.Name, time.Now().Unix())
+		storage.SetRoundRobinState(ch, time.Now().Unix())
 		slog.Info("round-robin: enqueued",
-			"source", src.Name, "guid", story.GUID, "channel", story.Channel)
+			"channel", ch, "guid", story.GUID, "source", story.Source)
 		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())
+	// 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")
 }
--- a/internal/scheduler/roundrobin_test.go
+++ b/internal/scheduler/roundrobin_test.go
@@ -4,7 +4,6 @@ import (
 	"path/filepath"
 	"testing"

-	"pete/internal/config"
 	"pete/internal/poster"
 	"pete/internal/storage"
 )
@@ -42,162 +41,145 @@ func insertPostable(t *testing.T, guid, source, channel string, seenAt int64) {
 	storage.MarkClassified(guid, channel, "[]")
 }

-func sources(names ...string) []config.SourceConfig {
-	out := make([]config.SourceConfig, len(names))
-	for i, n := range names {
-		out[i] = config.SourceConfig{
-			Name:                n,
-			FeedURL:             "https://example.com/" + n,
-			Tier:                1,
-			PollIntervalMinutes: 20,
-			Enabled:             true,
-		}
+func TestNew_EmptyChannels_ReturnsNil(t *testing.T) {
+	if New(nil, 4, &fakeQueue{}) != nil {
+		t.Error("expected nil when no channels are given")
 	}
-	return out
 }

-func TestNew_FiltersDisabledSources(t *testing.T) {
-	srcs := []config.SourceConfig{
-		{Name: "A", Enabled: true},
-		{Name: "B", Enabled: false},
-		{Name: "C", Enabled: true},
-	}
-	rr := New(srcs, 4, &fakeQueue{})
+func TestNew_SortsChannels(t *testing.T) {
+	rr := New([]string{"tech", "gaming", "politics"}, 4, &fakeQueue{})
 	if rr == nil {
 		t.Fatal("expected non-nil RoundRobin")
 	}
-	if len(rr.sources) != 2 || rr.sources[0].Name != "A" || rr.sources[1].Name != "C" {
-		t.Errorf("expected [A,C], got %v", rr.sources)
+	want := []string{"gaming", "politics", "tech"}
+	for i, ch := range rr.channels {
+		if ch != want[i] {
+			t.Errorf("channel[%d]=%s, want %s", i, ch, want[i])
+		}
 	}
 }

-func TestNew_NoEnabledSources_ReturnsNil(t *testing.T) {
-	srcs := []config.SourceConfig{{Name: "A", Enabled: false}}
-	if New(srcs, 4, &fakeQueue{}) != nil {
-		t.Error("expected nil when no sources are enabled")
-	}
-}
-
-func TestTick_PicksNewestPostableFromFirstSource(t *testing.T) {
+func TestTick_PicksNewestPostableFromFirstChannel(t *testing.T) {
 	setupTestDB(t)
-	insertPostable(t, "a-old", "A", "tech", 100)
-	insertPostable(t, "a-new", "A", "tech", 200)
-	insertPostable(t, "b-1", "B", "politics", 150)
+	insertPostable(t, "g-old", "TimeExt", "gaming", 100)
+	insertPostable(t, "g-new", "TimeExt", "gaming", 200)
+	insertPostable(t, "p-1", "GuardianPolitics", "politics", 150)

 	fq := &fakeQueue{}
-	rr := New(sources("A", "B"), 4, fq)
+	rr := New([]string{"gaming", "politics"}, 4, fq)
 	rr.tick()

 	if len(fq.items) != 1 {
 		t.Fatalf("expected 1 enqueued, got %d", len(fq.items))
 	}
-	if fq.items[0].GUID != "a-new" {
-		t.Errorf("expected newest from A (a-new), got %s", fq.items[0].GUID)
+	if fq.items[0].GUID != "g-new" {
+		t.Errorf("expected newest from gaming (g-new), got %s", fq.items[0].GUID)
 	}
-	lastSource, _, _ := storage.GetRoundRobinState()
-	if lastSource != "A" {
-		t.Errorf("expected last_source=A, got %s", lastSource)
+	lastChannel, _, _ := storage.GetRoundRobinState()
+	if lastChannel != "gaming" {
+		t.Errorf("expected last_channel=gaming, got %s", lastChannel)
 	}
 }

 func TestTick_SkipsEmptyAndAdvancesPointer(t *testing.T) {
 	setupTestDB(t)
-	// Pretend we just posted from A; next tick should start at B, which is
-	// empty, so it should skip to C.
-	storage.SetRoundRobinState("A", 0)
-	insertPostable(t, "c-1", "C", "tech", 100)
+	// Pretend we just posted in gaming; next tick should start at politics, which
+	// is empty, so it should skip to tech.
+	storage.SetRoundRobinState("gaming", 0)
+	insertPostable(t, "t-1", "GuardianTech", "tech", 100)

 	fq := &fakeQueue{}
-	rr := New(sources("A", "B", "C"), 4, fq)
+	rr := New([]string{"gaming", "politics", "tech"}, 4, fq)
 	rr.tick()

-	if len(fq.items) != 1 || fq.items[0].GUID != "c-1" {
-		t.Fatalf("expected c-1 to be enqueued, got %+v", fq.items)
+	if len(fq.items) != 1 || fq.items[0].GUID != "t-1" {
+		t.Fatalf("expected t-1 to be enqueued, got %+v", fq.items)
 	}
-	lastSource, _, _ := storage.GetRoundRobinState()
-	if lastSource != "C" {
-		t.Errorf("expected last_source=C after skipping B, got %s", lastSource)
+	lastChannel, _, _ := storage.GetRoundRobinState()
+	if lastChannel != "tech" {
+		t.Errorf("expected last_channel=tech after skipping politics, got %s", lastChannel)
 	}
 }

 func TestTick_NothingPostable_LeavesPointerUnchanged(t *testing.T) {
 	setupTestDB(t)
-	storage.SetRoundRobinState("A", 1000)
+	storage.SetRoundRobinState("gaming", 1000)

 	fq := &fakeQueue{}
-	rr := New(sources("A", "B"), 4, fq)
+	rr := New([]string{"gaming", "tech"}, 4, fq)
 	rr.tick()

 	if len(fq.items) != 0 {
 		t.Fatalf("expected 0 enqueued, got %d", len(fq.items))
 	}
-	lastSource, lastTickAt, _ := storage.GetRoundRobinState()
-	if lastSource != "A" {
-		t.Errorf("expected last_source unchanged=A, got %s", lastSource)
+	lastChannel, lastTickAt, _ := storage.GetRoundRobinState()
+	if lastChannel != "gaming" {
+		t.Errorf("expected last_channel unchanged=gaming, got %s", lastChannel)
 	}
 	if lastTickAt <= 1000 {
 		t.Errorf("expected last_tick_at to advance past 1000, got %d", lastTickAt)
 	}
 }

-func TestTick_WrapsAroundFromLastSource(t *testing.T) {
+func TestTick_WrapsAroundFromLastChannel(t *testing.T) {
 	setupTestDB(t)
-	// last_source = C means next tick starts at A (wraparound).
-	storage.SetRoundRobinState("C", 0)
-	insertPostable(t, "a-1", "A", "tech", 100)
-	insertPostable(t, "c-1", "C", "tech", 200) // C is newer, but rotation must pick A
+	// Sorted order is [gaming, politics, tech]. last_channel=tech wraps to gaming.
+	storage.SetRoundRobinState("tech", 0)
+	insertPostable(t, "g-1", "TimeExt", "gaming", 100)
+	insertPostable(t, "t-1", "GuardianTech", "tech", 200) // tech is newer, but rotation must pick gaming

 	fq := &fakeQueue{}
-	rr := New(sources("A", "B", "C"), 4, fq)
+	rr := New([]string{"gaming", "politics", "tech"}, 4, fq)
 	rr.tick()

-	if len(fq.items) != 1 || fq.items[0].GUID != "a-1" {
-		t.Fatalf("expected a-1 (wraparound), got %+v", fq.items)
+	if len(fq.items) != 1 || fq.items[0].GUID != "g-1" {
+		t.Fatalf("expected g-1 (wraparound), got %+v", fq.items)
 	}
 }

 func TestTick_DiscardedAndDuplicateExcluded(t *testing.T) {
 	setupTestDB(t)
-	insertPostable(t, "a-discard", "A", "_discarded", 200) // sentinel
-	insertPostable(t, "a-dup", "A", "_duplicate", 150)     // sentinel
-	insertPostable(t, "a-real", "A", "tech", 100)          // older but only real one
+	insertPostable(t, "t-discard", "GuardianTech", "_discarded", 200)
+	insertPostable(t, "t-dup", "GuardianTech", "_duplicate", 150)
+	insertPostable(t, "t-real", "GuardianTech", "tech", 100)

 	fq := &fakeQueue{}
-	rr := New(sources("A"), 4, fq)
+	rr := New([]string{"tech"}, 4, fq)
 	rr.tick()

-	if len(fq.items) != 1 || fq.items[0].GUID != "a-real" {
-		t.Fatalf("expected a-real (sentinels excluded), got %+v", fq.items)
+	if len(fq.items) != 1 || fq.items[0].GUID != "t-real" {
+		t.Fatalf("expected t-real (sentinels excluded), got %+v", fq.items)
 	}
 }

 func TestTick_AlreadyPostedExcluded(t *testing.T) {
 	setupTestDB(t)
-	insertPostable(t, "a-posted", "A", "tech", 200)
-	insertPostable(t, "a-fresh", "A", "tech", 100)
-	storage.InsertPostLog("a-posted", "tech", "$evt1", "https://example.com/a-posted")
+	insertPostable(t, "t-posted", "GuardianTech", "tech", 200)
+	insertPostable(t, "t-fresh", "GuardianTech", "tech", 100)
+	storage.InsertPostLog("t-posted", "tech", "$evt1", "https://example.com/t-posted")

 	fq := &fakeQueue{}
-	rr := New(sources("A"), 4, fq)
+	rr := New([]string{"tech"}, 4, fq)
 	rr.tick()

-	if len(fq.items) != 1 || fq.items[0].GUID != "a-fresh" {
-		t.Fatalf("expected a-fresh (a-posted already in post_log), got %+v", fq.items)
+	if len(fq.items) != 1 || fq.items[0].GUID != "t-fresh" {
+		t.Fatalf("expected t-fresh (t-posted already in post_log), got %+v", fq.items)
 	}
 }

-func TestTick_UnknownLastSource_StartsAtIndexZero(t *testing.T) {
+func TestTick_UnknownLastChannel_StartsAtIndexZero(t *testing.T) {
 	setupTestDB(t)
-	// Stale state pointing at a source no longer in config.
-	storage.SetRoundRobinState("RetiredSource", 0)
-	insertPostable(t, "a-1", "A", "tech", 100)
-	insertPostable(t, "b-1", "B", "tech", 200)
+	// Stale state pointing at a channel no longer in config.
+	storage.SetRoundRobinState("retired", 0)
+	insertPostable(t, "g-1", "TimeExt", "gaming", 100)
+	insertPostable(t, "t-1", "GuardianTech", "tech", 200)

 	fq := &fakeQueue{}
-	rr := New(sources("A", "B"), 4, fq)
+	rr := New([]string{"gaming", "tech"}, 4, fq)
 	rr.tick()

-	if len(fq.items) != 1 || fq.items[0].GUID != "a-1" {
-		t.Fatalf("expected a-1 (start at idx 0 when last_source unknown), got %+v", fq.items)
+	if len(fq.items) != 1 || fq.items[0].GUID != "g-1" {
+		t.Fatalf("expected g-1 (start at idx 0 when last_channel unknown), got %+v", fq.items)
 	}
 }
--- a/internal/storage/db.go
+++ b/internal/storage/db.go
@@ -81,6 +81,7 @@ func runMigrations(d *sql.DB) error {
 	addColumnIfMissing(d, "stories", "url_canonical", "TEXT")
 	addColumnIfMissing(d, "stories", "headline_norm", "TEXT")
 	addColumnIfMissing(d, "post_log", "url_canonical", "TEXT")
+	addColumnIfMissing(d, "round_robin_state", "last_channel", "TEXT")

 	// FTS5 virtual tables don't support IF NOT EXISTS reliably.
 	// Check sqlite_master before creating.
--- a/internal/storage/queries.go
+++ b/internal/storage/queries.go
@@ -270,29 +270,29 @@ func GetNewestPostableStoryByChannel(channel string) (*Story, error) {
 	return &s, nil
 }

-// GetRoundRobinState returns the last source posted by the round-robin
+// GetRoundRobinState returns the last channel posted by the round-robin
 // scheduler and the timestamp of that tick. Empty string + 0 if no state yet.
-func GetRoundRobinState() (lastSource string, lastTickAt int64, err error) {
-	row := Get().QueryRow(`SELECT last_source, last_tick_at FROM round_robin_state WHERE id = 1`)
-	var ls *string
-	if scanErr := row.Scan(&ls, &lastTickAt); scanErr != nil {
+func GetRoundRobinState() (lastChannel string, lastTickAt int64, err error) {
+	row := Get().QueryRow(`SELECT last_channel, last_tick_at FROM round_robin_state WHERE id = 1`)
+	var lc *string
+	if scanErr := row.Scan(&lc, &lastTickAt); scanErr != nil {
 		if errors.Is(scanErr, sql.ErrNoRows) {
 			return "", 0, nil
 		}
 		return "", 0, scanErr
 	}
-	if ls != nil {
-		lastSource = *ls
+	if lc != nil {
+		lastChannel = *lc
 	}
-	return lastSource, lastTickAt, nil
+	return lastChannel, lastTickAt, nil
 }

-// SetRoundRobinState persists the last-posted source and tick timestamp.
-func SetRoundRobinState(lastSource string, tickAt int64) {
+// SetRoundRobinState persists the last-posted channel and tick timestamp.
+func SetRoundRobinState(lastChannel string, tickAt int64) {
 	exec("set round_robin_state",
-		`INSERT INTO round_robin_state (id, last_source, last_tick_at) VALUES (1, ?, ?)
-		 ON CONFLICT(id) DO UPDATE SET last_source = excluded.last_source, last_tick_at = excluded.last_tick_at`,
-		lastSource, tickAt)
+		`INSERT INTO round_robin_state (id, last_channel, last_tick_at) VALUES (1, ?, ?)
+		 ON CONFLICT(id) DO UPDATE SET last_channel = excluded.last_channel, last_tick_at = excluded.last_tick_at`,
+		lastChannel, tickAt)
 }

 // MarshalPlatforms converts a string slice to a JSON array string for storage.
--- a/internal/storage/schema.go
+++ b/internal/storage/schema.go
@@ -42,7 +42,7 @@ CREATE TABLE IF NOT EXISTS post_log (

 CREATE TABLE IF NOT EXISTS round_robin_state (
    id           INTEGER PRIMARY KEY CHECK (id = 1),
-    last_source  TEXT,
+    last_channel TEXT,
    last_tick_at INTEGER NOT NULL DEFAULT 0
 );

--- a/main.go
+++ b/main.go
@@ -214,9 +214,13 @@ func main() {
 	slog.Info("pollers started")

 	if roundRobinMode {
-		rr := scheduler.New(cfg.Sources, cfg.Posting.RoundRobin.IntervalHours, queue)
+		channelNames := make([]string, 0, len(cfg.Matrix.Channels))
+		for name := range cfg.Matrix.Channels {
+			channelNames = append(channelNames, name)
+		}
+		rr := scheduler.New(channelNames, cfg.Posting.RoundRobin.IntervalHours, queue)
 		if rr == nil {
-			slog.Warn("round-robin enabled but no sources are enabled; nothing will post")
+			slog.Warn("round-robin enabled but no channels are configured; nothing will post")
 		} else {
 			go rr.Start(ctx)
 		}