diff --git a/README.md b/README.md index a7e4d85..53ec50f 100644 --- 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 diff --git a/internal/scheduler/roundrobin.go b/internal/scheduler/roundrobin.go index 0f41a09..9395ed3 100644 --- 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") } diff --git a/internal/scheduler/roundrobin_test.go b/internal/scheduler/roundrobin_test.go index 5090278..cc293d7 100644 --- 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) } } diff --git a/internal/storage/db.go b/internal/storage/db.go index 1fe3fd8..67fef8b 100644 --- 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. diff --git a/internal/storage/queries.go b/internal/storage/queries.go index 5459bf0..976c03c 100644 --- 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. diff --git a/internal/storage/schema.go b/internal/storage/schema.go index 9843dac..e7ca475 100644 --- 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 ); diff --git a/main.go b/main.go index 1f61530..99dbc1f 100644 --- 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) }