diff --git a/config.example.yaml b/config.example.yaml index 7059583..63bf43c 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -21,6 +21,9 @@ posting: burst_cap_count: 3 burst_cap_window_seconds: 1800 daily_cap_total: 5 # hard global cap across ALL channels (rolling 24h); 0 disables + round_robin: + enabled: false # when true, replaces immediate posting with paced rotation + interval_hours: 4 # one story per N hours, cycling through sources in config order storage: db_path: "./data/pete.db" diff --git a/internal/config/config.go b/internal/config/config.go index cb17ba4..4452356 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -44,7 +44,16 @@ type PostingConfig struct { DedupCooldownHours int `yaml:"dedup_cooldown_hours"` // DailyCapTotal is the hard global cap on posts across ALL channels in a // rolling 24-hour window. 0 disables the cap. - DailyCapTotal int `yaml:"daily_cap_total"` + DailyCapTotal int `yaml:"daily_cap_total"` + RoundRobin RoundRobinConfig `yaml:"round_robin"` +} + +// RoundRobinConfig switches Pete from immediate-on-classify posting to a +// paced rotation: one story per IntervalHours, picking the next source in +// config order that has a postable story (skip-and-advance, newest-first). +type RoundRobinConfig struct { + Enabled bool `yaml:"enabled"` + IntervalHours int `yaml:"interval_hours"` } type StorageConfig struct { @@ -160,6 +169,9 @@ func (c *Config) applyDefaults() { if c.Posting.DedupCooldownHours == 0 { c.Posting.DedupCooldownHours = 48 } + if c.Posting.RoundRobin.IntervalHours == 0 { + c.Posting.RoundRobin.IntervalHours = 4 + } if c.Storage.RecentWindowHours == 0 { c.Storage.RecentWindowHours = 24 } diff --git a/internal/scheduler/roundrobin.go b/internal/scheduler/roundrobin.go new file mode 100644 index 0000000..0f41a09 --- /dev/null +++ b/internal/scheduler/roundrobin.go @@ -0,0 +1,150 @@ +// 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") +} diff --git a/internal/scheduler/roundrobin_test.go b/internal/scheduler/roundrobin_test.go new file mode 100644 index 0000000..5090278 --- /dev/null +++ b/internal/scheduler/roundrobin_test.go @@ -0,0 +1,203 @@ +package scheduler + +import ( + "path/filepath" + "testing" + + "pete/internal/config" + "pete/internal/poster" + "pete/internal/storage" +) + +type fakeQueue struct { + items []poster.QueueItem +} + +func (f *fakeQueue) Enqueue(item poster.QueueItem) { + f.items = append(f.items, item) +} + +func setupTestDB(t *testing.T) { + t.Helper() + dbPath := filepath.Join(t.TempDir(), "test.db") + if err := storage.Init(dbPath); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { storage.Close() }) +} + +func insertPostable(t *testing.T, guid, source, channel string, seenAt int64) { + t.Helper() + if err := storage.InsertStory(&storage.Story{ + GUID: guid, + Headline: "Headline for " + guid, + Lede: "Lede for " + guid, + ArticleURL: "https://example.com/" + guid, + Source: source, + FeedHint: "tech", + SeenAt: seenAt, + }); err != nil { + t.Fatalf("InsertStory(%s): %v", guid, err) + } + 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, + } + } + 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{}) + 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) + } +} + +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) { + setupTestDB(t) + insertPostable(t, "a-old", "A", "tech", 100) + insertPostable(t, "a-new", "A", "tech", 200) + insertPostable(t, "b-1", "B", "politics", 150) + + fq := &fakeQueue{} + rr := New(sources("A", "B"), 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) + } + lastSource, _, _ := storage.GetRoundRobinState() + if lastSource != "A" { + t.Errorf("expected last_source=A, got %s", lastSource) + } +} + +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) + + fq := &fakeQueue{} + rr := New(sources("A", "B", "C"), 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) + } + lastSource, _, _ := storage.GetRoundRobinState() + if lastSource != "C" { + t.Errorf("expected last_source=C after skipping B, got %s", lastSource) + } +} + +func TestTick_NothingPostable_LeavesPointerUnchanged(t *testing.T) { + setupTestDB(t) + storage.SetRoundRobinState("A", 1000) + + fq := &fakeQueue{} + rr := New(sources("A", "B"), 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) + } + if lastTickAt <= 1000 { + t.Errorf("expected last_tick_at to advance past 1000, got %d", lastTickAt) + } +} + +func TestTick_WrapsAroundFromLastSource(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 + + fq := &fakeQueue{} + rr := New(sources("A", "B", "C"), 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) + } +} + +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 + + fq := &fakeQueue{} + rr := New(sources("A"), 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) + } +} + +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") + + fq := &fakeQueue{} + rr := New(sources("A"), 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) + } +} + +func TestTick_UnknownLastSource_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) + + fq := &fakeQueue{} + rr := New(sources("A", "B"), 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) + } +} diff --git a/internal/storage/queries.go b/internal/storage/queries.go index d59de59..fe0f2f4 100644 --- a/internal/storage/queries.go +++ b/internal/storage/queries.go @@ -1,7 +1,9 @@ package storage import ( + "database/sql" "encoding/json" + "errors" "log/slog" "time" ) @@ -221,6 +223,55 @@ func InsertReaction(postGUID, channel, eventID, emoji, userID string, reactedAt postGUID, channel, eventID, emoji, userID, reactedAt) } +// GetNewestPostableStory returns the newest classified story from a source +// that has a real (non-sentinel) channel and has not yet been posted (no +// row in post_log). Returns (nil, nil) when nothing qualifies. +func GetNewestPostableStory(source string) (*Story, error) { + row := Get().QueryRow( + `SELECT guid, headline, lede, image_url, article_url, source, feed_hint, platforms, channel, seen_at + FROM stories + WHERE classified = 1 + AND source = ? + AND channel IS NOT NULL + AND channel NOT IN ('_discarded', '_duplicate') + AND guid NOT IN (SELECT guid FROM post_log) + ORDER BY seen_at DESC + LIMIT 1`, source) + var s Story + if err := row.Scan(&s.GUID, &s.Headline, &s.Lede, &s.ImageURL, &s.ArticleURL, &s.Source, &s.FeedHint, &s.Platforms, &s.Channel, &s.SeenAt); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + return nil, err + } + return &s, nil +} + +// GetRoundRobinState returns the last source 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 { + if errors.Is(scanErr, sql.ErrNoRows) { + return "", 0, nil + } + return "", 0, scanErr + } + if ls != nil { + lastSource = *ls + } + return lastSource, lastTickAt, nil +} + +// SetRoundRobinState persists the last-posted source and tick timestamp. +func SetRoundRobinState(lastSource 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) +} + // MarshalPlatforms converts a string slice to a JSON array string for storage. func MarshalPlatforms(platforms []string) string { if len(platforms) == 0 { diff --git a/internal/storage/schema.go b/internal/storage/schema.go index 168e6f2..9843dac 100644 --- a/internal/storage/schema.go +++ b/internal/storage/schema.go @@ -40,6 +40,12 @@ CREATE TABLE IF NOT EXISTS post_log ( posted_at INTEGER NOT NULL ); +CREATE TABLE IF NOT EXISTS round_robin_state ( + id INTEGER PRIMARY KEY CHECK (id = 1), + last_source TEXT, + last_tick_at INTEGER NOT NULL DEFAULT 0 +); + CREATE TABLE IF NOT EXISTS reactions ( id INTEGER PRIMARY KEY AUTOINCREMENT, post_guid TEXT NOT NULL, diff --git a/main.go b/main.go index 293d2bb..ab1ed3d 100644 --- a/main.go +++ b/main.go @@ -15,6 +15,7 @@ import ( "pete/internal/ingestion" "pete/internal/matrix" "pete/internal/poster" + "pete/internal/scheduler" "pete/internal/storage" ) @@ -94,7 +95,10 @@ func main() { go queue.Start(ctx) slog.Info("post queue started") - // Build the pipeline callback: classify → enqueue + roundRobinMode := cfg.Posting.RoundRobin.Enabled + + // Build the pipeline callback: classify → enqueue (or just classify+store + // when round-robin mode is enabled; the scheduler does the enqueueing). processItem := func(ctx context.Context, item *ingestion.FeedItem) { result, err := cls.Classify(ctx, item) if err != nil { @@ -123,6 +127,9 @@ func main() { storage.MarkClassified(item.GUID, result.Channel, platforms) if result.DuplicateOf != nil { + // Overwrite with sentinel so the round-robin scheduler doesn't pick + // this up later. Immediate-mode posting already returned above. + storage.MarkClassified(item.GUID, "_duplicate", "[]") slog.Info("story deduplicated", "guid", item.GUID, "duplicate_of", *result.DuplicateOf, @@ -130,6 +137,14 @@ func main() { return } + if roundRobinMode { + // Story stays in DB classified+postable; the scheduler will pick + // it up on the next rotation tick. + slog.Info("story classified, awaiting round-robin tick", + "guid", item.GUID, "source", item.Source, "channel", result.Channel) + return + } + // Validate image imageURL := "" if item.ImageURL != "" && ingestion.ValidateImageURL(item.ImageURL) { @@ -154,6 +169,15 @@ func main() { poller.Start(ctx) slog.Info("pollers started") + if roundRobinMode { + rr := scheduler.New(cfg.Sources, cfg.Posting.RoundRobin.IntervalHours, queue) + if rr == nil { + slog.Warn("round-robin enabled but no sources are enabled; nothing will post") + } else { + go rr.Start(ctx) + } + } + // Run maintenance on startup storage.RunMaintenance(cfg.Storage.RecentWindowHours, cfg.Storage.ClassificationLogDays)