`,
- wantHTML: "",
- },
- {
- name: "leading whitespace trimmed",
- in: " - first\n - second ",
- wantPlain: "• first\n• second",
- wantHTML: "",
- },
- }
- for _, c := range cases {
- t.Run(c.name, func(t *testing.T) {
- gotPlain, gotHTML := formatSummary(c.in)
- if gotPlain != c.wantPlain {
- t.Errorf("plain:\n got %q\n want %q", gotPlain, c.wantPlain)
- }
- if gotHTML != c.wantHTML {
- t.Errorf("html:\n got %q\n want %q", gotHTML, c.wantHTML)
- }
- })
- }
-}
-
-func TestFormatSummary_NoBulletsPassthrough(t *testing.T) {
- in := "Just a single paragraph with no bullets at all."
- plain, htmlBody := formatSummary(in)
- if plain != in {
- t.Errorf("plain should pass through raw, got %q", plain)
- }
- if !strings.HasPrefix(htmlBody, "") || !strings.HasSuffix(htmlBody, "
") {
- t.Errorf("html should wrap raw in , got %q", htmlBody)
- }
- if !strings.Contains(htmlBody, "single paragraph") {
- t.Errorf("html should contain the body text, got %q", htmlBody)
- }
-}
-
-func TestIsQuestionReaction(t *testing.T) {
- want := []string{"❓", "❓️", "❔", "❔️", "⁉", "⁉️", "🤔", "?", "？"}
- for _, k := range want {
- if !IsQuestionReaction(k) {
- t.Errorf("expected %q to trigger explain", k)
- }
- }
- notQuestion := []string{"👍", "👎", "🔥", "", "??", "❤️"}
- for _, k := range notQuestion {
- if IsQuestionReaction(k) {
- t.Errorf("expected %q to NOT trigger explain", k)
- }
- }
-}
diff --git a/internal/ingestion/parser.go b/internal/ingestion/parser.go
index 9e1cada..20c2043 100644
--- a/internal/ingestion/parser.go
+++ b/internal/ingestion/parser.go
@@ -16,7 +16,7 @@ const userAgent = "Pete/1.0 (newsbot; +https://github.com/reala-misaki/pete)"
var feedClient = &http.Client{Timeout: 30 * time.Second}
-// FeedItem represents a parsed RSS item ready for classification.
+// FeedItem represents a parsed RSS item ready for routing.
type FeedItem struct {
GUID string
Headline string
@@ -24,9 +24,7 @@ type FeedItem struct {
ImageURL string
ArticleURL string
Source string
- FeedHint string
- Tier int
- DirectRoute *string
+ DirectRoute string
}
var (
diff --git a/internal/ingestion/poller.go b/internal/ingestion/poller.go
index cc79e83..a6bc13e 100644
--- a/internal/ingestion/poller.go
+++ b/internal/ingestion/poller.go
@@ -11,9 +11,7 @@ import (
"pete/internal/storage"
)
-// ProcessFunc is called for each new feed item that needs classification.
-// ctx is the poller's parent context; callbacks should pass it into any
-// downstream LLM/HTTP work so shutdown aborts in-flight calls.
+// ProcessFunc is called for each new feed item that needs routing.
type ProcessFunc func(ctx context.Context, item *FeedItem)
// Poller manages per-source RSS polling goroutines.
@@ -81,7 +79,6 @@ func (p *Poller) pollSource(ctx context.Context, src config.SourceConfig) {
"source", src.Name,
"failures", consecutiveFailures,
)
- // TODO: send admin room warning via matrix client
}
} else {
consecutiveFailures = 0
@@ -163,11 +160,9 @@ func (p *Poller) pollOnceWithErr(ctx context.Context, src config.SourceConfig) e
// Stamp source metadata onto the item
items[i].Source = src.Name
- items[i].FeedHint = src.FeedHint
- items[i].Tier = src.Tier
items[i].DirectRoute = src.DirectRoute
- // Store immediately (before classification) to prevent re-ingestion
+ // Store immediately (before routing) to prevent re-ingestion
if err := storage.InsertStory(&storage.Story{
GUID: items[i].GUID,
Headline: items[i].Headline,
@@ -177,7 +172,6 @@ func (p *Poller) pollOnceWithErr(ctx context.Context, src config.SourceConfig) e
URLCanonical: canonical,
HeadlineNorm: headlineNorm,
Source: items[i].Source,
- FeedHint: items[i].FeedHint,
SeenAt: time.Now().Unix(),
}); err != nil {
slog.Error("failed to insert story", "guid", items[i].GUID, "err", err)
@@ -192,40 +186,5 @@ func (p *Poller) pollOnceWithErr(ctx context.Context, src config.SourceConfig) e
slog.Info("ingested new stories", "source", src.Name, "count", newCount)
}
- // Retry unclassified stories from this source only (cap at 20 to avoid runaway retries)
- unclassified, err := storage.GetUnclassifiedStories(src.Name)
- if err != nil {
- slog.Error("failed to get unclassified stories", "err", err)
- return nil
- }
- for _, s := range unclassified {
- if ctx.Err() != nil {
- return nil
- }
- // Skip stories we just ingested (they're already being processed above)
- alreadyProcessed := false
- for _, item := range items {
- if item.GUID == s.GUID {
- alreadyProcessed = true
- break
- }
- }
- if alreadyProcessed {
- continue
- }
-
- p.process(ctx, &FeedItem{
- GUID: s.GUID,
- Headline: s.Headline,
- Lede: s.Lede,
- ImageURL: s.ImageURL,
- ArticleURL: s.ArticleURL,
- Source: s.Source,
- FeedHint: s.FeedHint,
- Tier: src.Tier,
- DirectRoute: src.DirectRoute,
- })
- }
-
return nil
}
diff --git a/internal/scheduler/roundrobin_test.go b/internal/scheduler/roundrobin_test.go
index cc293d7..4f4e3e5 100644
--- a/internal/scheduler/roundrobin_test.go
+++ b/internal/scheduler/roundrobin_test.go
@@ -33,7 +33,6 @@ func insertPostable(t *testing.T, guid, source, channel string, seenAt int64) {
Lede: "Lede for " + guid,
ArticleURL: "https://example.com/" + guid,
Source: source,
- FeedHint: "tech",
SeenAt: seenAt,
}); err != nil {
t.Fatalf("InsertStory(%s): %v", guid, err)
diff --git a/internal/storage/db.go b/internal/storage/db.go
index 67fef8b..f7d49a3 100644
--- a/internal/storage/db.go
+++ b/internal/storage/db.go
@@ -101,15 +101,7 @@ func runMigrations(d *sql.DB) error {
}
// RunMaintenance prunes stale data. Called periodically.
-func RunMaintenance(recentWindowHours, classificationLogDays int) {
- recentCutoff := nowUnix() - int64(recentWindowHours*3600)
- exec("prune recent_headlines",
- `DELETE FROM recent_headlines WHERE seen_at < ?`, recentCutoff)
-
- logCutoff := nowUnix() - int64(classificationLogDays*86400)
- exec("prune classification_log",
- `DELETE FROM classification_log WHERE logged_at < ?`, logCutoff)
-
+func RunMaintenance() {
// Prune old stories (30 days) and their post logs / reactions
storyCutoff := nowUnix() - int64(30*86400)
exec("prune old stories",
diff --git a/internal/storage/queries.go b/internal/storage/queries.go
index d358c60..8c7254e 100644
--- a/internal/storage/queries.go
+++ b/internal/storage/queries.go
@@ -29,9 +29,9 @@ func InsertStory(s *Story) error {
classified = 1
}
_, err := Get().Exec(
- `INSERT INTO stories (guid, headline, lede, image_url, article_url, url_canonical, headline_norm, source, feed_hint, platforms, channel, classified, seen_at)
- VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
- s.GUID, s.Headline, s.Lede, s.ImageURL, s.ArticleURL, nullIfEmpty(s.URLCanonical), nullIfEmpty(s.HeadlineNorm), s.Source, s.FeedHint, s.Platforms, s.Channel, classified, s.SeenAt,
+ `INSERT INTO stories (guid, headline, lede, image_url, article_url, url_canonical, headline_norm, source, platforms, channel, classified, seen_at)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+ s.GUID, s.Headline, s.Lede, s.ImageURL, s.ArticleURL, nullIfEmpty(s.URLCanonical), nullIfEmpty(s.HeadlineNorm), s.Source, s.Platforms, s.Channel, classified, s.SeenAt,
)
return err
}
@@ -99,73 +99,18 @@ func MarkClassified(guid, channel, platforms string) {
channel, platforms, guid)
}
-// GetUnclassifiedStories returns stories from a specific source that were ingested but not yet classified.
-func GetUnclassifiedStories(source string) ([]Story, error) {
- rows, err := Get().Query(
- `SELECT guid, headline, lede, image_url, article_url, source, feed_hint, platforms, seen_at
- FROM stories WHERE classified = 0 AND source = ? ORDER BY seen_at ASC LIMIT 20`, source)
- if err != nil {
- return nil, err
- }
- defer rows.Close()
-
- var stories []Story
- for rows.Next() {
- var s Story
- if err := rows.Scan(&s.GUID, &s.Headline, &s.Lede, &s.ImageURL, &s.ArticleURL, &s.Source, &s.FeedHint, &s.Platforms, &s.SeenAt); err != nil {
- return nil, err
- }
- stories = append(stories, s)
- }
- return stories, rows.Err()
-}
-
// GetStoryByGUID returns the full story record for a GUID, or nil if not found.
func GetStoryByGUID(guid string) (*Story, error) {
row := Get().QueryRow(
- `SELECT guid, headline, lede, image_url, article_url, source, feed_hint, platforms, channel, seen_at
+ `SELECT guid, headline, lede, image_url, article_url, source, platforms, channel, seen_at
FROM stories WHERE guid = ?`, guid)
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 err := row.Scan(&s.GUID, &s.Headline, &s.Lede, &s.ImageURL, &s.ArticleURL, &s.Source, &s.Platforms, &s.Channel, &s.SeenAt); err != nil {
return nil, err
}
return &s, nil
}
-// InsertRecentHeadline adds a headline to the dedup context window.
-func InsertRecentHeadline(guid, headline, source string) {
- exec("insert recent_headline",
- `INSERT OR REPLACE INTO recent_headlines (guid, headline, source, seen_at) VALUES (?, ?, ?, ?)`,
- guid, headline, source, nowUnix())
-}
-
-// GetRecentHeadlines returns the most recent headlines for LLM dedup context, capped at limit.
-func GetRecentHeadlines(limit int) ([]RecentHeadline, error) {
- rows, err := Get().Query(
- `SELECT guid, headline, source, seen_at FROM recent_headlines ORDER BY seen_at DESC LIMIT ?`, limit)
- if err != nil {
- return nil, err
- }
- defer rows.Close()
-
- var headlines []RecentHeadline
- for rows.Next() {
- var h RecentHeadline
- if err := rows.Scan(&h.GUID, &h.Headline, &h.Source, &h.SeenAt); err != nil {
- return nil, err
- }
- headlines = append(headlines, h)
- }
- return headlines, rows.Err()
-}
-
-// InsertClassificationLog stores a classification result for tuning review.
-func InsertClassificationLog(guid, resultJSON string) {
- exec("insert classification_log",
- `INSERT OR REPLACE INTO classification_log (guid, result, logged_at) VALUES (?, ?, ?)`,
- guid, resultJSON, nowUnix())
-}
-
// InsertPostLog records that a story was posted to a channel.
// Uses OR IGNORE to prevent duplicate posts for the same story+channel.
func InsertPostLog(guid, channel, eventID, urlCanonical string) {
@@ -228,7 +173,7 @@ func InsertReaction(postGUID, channel, eventID, emoji, userID string, reactedAt
// 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
+ `SELECT guid, headline, lede, image_url, article_url, source, platforms, channel, seen_at
FROM stories
WHERE classified = 1
AND source = ?
@@ -238,7 +183,7 @@ func GetNewestPostableStory(source string) (*Story, error) {
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 err := row.Scan(&s.GUID, &s.Headline, &s.Lede, &s.ImageURL, &s.ArticleURL, &s.Source, &s.Platforms, &s.Channel, &s.SeenAt); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return nil, nil
}
@@ -253,7 +198,7 @@ func GetNewestPostableStory(source string) (*Story, error) {
// between ticks). Returns (nil, nil) when nothing qualifies.
func GetNewestPostableStoryByChannel(channel string) (*Story, error) {
row := Get().QueryRow(
- `SELECT guid, headline, lede, image_url, article_url, source, feed_hint, platforms, channel, seen_at
+ `SELECT guid, headline, lede, image_url, article_url, source, platforms, channel, seen_at
FROM stories
WHERE classified = 1
AND channel = ?
@@ -261,7 +206,7 @@ func GetNewestPostableStoryByChannel(channel string) (*Story, error) {
ORDER BY seen_at DESC
LIMIT 1`, channel)
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 err := row.Scan(&s.GUID, &s.Headline, &s.Lede, &s.ImageURL, &s.ArticleURL, &s.Source, &s.Platforms, &s.Channel, &s.SeenAt); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return nil, nil
}
@@ -275,7 +220,7 @@ func GetNewestPostableStoryByChannel(channel string) (*Story, error) {
// channels (_discarded, _duplicate) are excluded.
func ListClassifiedByChannel(channel string, limit, offset int) ([]Story, error) {
rows, err := Get().Query(
- `SELECT guid, headline, lede, image_url, article_url, source, feed_hint, platforms, channel, seen_at
+ `SELECT guid, headline, lede, image_url, article_url, source, platforms, channel, seen_at
FROM stories
WHERE classified = 1
AND channel = ?
@@ -288,7 +233,7 @@ func ListClassifiedByChannel(channel string, limit, offset int) ([]Story, error)
var out []Story
for rows.Next() {
var s Story
- if err := rows.Scan(&s.GUID, &s.Headline, &s.Lede, &s.ImageURL, &s.ArticleURL, &s.Source, &s.FeedHint, &s.Platforms, &s.Channel, &s.SeenAt); err != nil {
+ if err := rows.Scan(&s.GUID, &s.Headline, &s.Lede, &s.ImageURL, &s.ArticleURL, &s.Source, &s.Platforms, &s.Channel, &s.SeenAt); err != nil {
return nil, err
}
out = append(out, s)
diff --git a/internal/storage/schema.go b/internal/storage/schema.go
index e7ca475..cf1b513 100644
--- a/internal/storage/schema.go
+++ b/internal/storage/schema.go
@@ -11,26 +11,12 @@ CREATE TABLE IF NOT EXISTS stories (
url_canonical TEXT,
headline_norm TEXT,
source TEXT NOT NULL,
- feed_hint TEXT,
platforms TEXT,
channel TEXT,
classified INTEGER NOT NULL DEFAULT 0,
seen_at INTEGER NOT NULL
);
-CREATE TABLE IF NOT EXISTS recent_headlines (
- guid TEXT PRIMARY KEY,
- headline TEXT NOT NULL,
- source TEXT NOT NULL,
- seen_at INTEGER NOT NULL
-);
-
-CREATE TABLE IF NOT EXISTS classification_log (
- guid TEXT PRIMARY KEY,
- result TEXT NOT NULL,
- logged_at INTEGER NOT NULL
-);
-
CREATE TABLE IF NOT EXISTS post_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
guid TEXT NOT NULL,
diff --git a/internal/storage/storage_test.go b/internal/storage/storage_test.go
index 496becb..6b39448 100644
--- a/internal/storage/storage_test.go
+++ b/internal/storage/storage_test.go
@@ -1,7 +1,6 @@
package storage
import (
- "fmt"
"path/filepath"
"testing"
)
@@ -48,7 +47,6 @@ func TestInsertStoryAndGUIDSeen(t *testing.T) {
Lede: "Test lede sentence.",
ArticleURL: "https://example.com/article",
Source: "Test Source",
- FeedHint: "tech",
SeenAt: 1700000000,
}
if err := InsertStory(s); err != nil {
@@ -79,63 +77,21 @@ func TestInsertStoryDuplicateGUID(t *testing.T) {
}
}
-func TestMarkClassifiedAndGetUnclassified(t *testing.T) {
+func TestMarkClassified(t *testing.T) {
setupTestDB(t)
InsertStory(&Story{
GUID: "g1", Headline: "H1", ArticleURL: "https://a.com", Source: "S", SeenAt: 1,
})
- InsertStory(&Story{
- GUID: "g2", Headline: "H2", ArticleURL: "https://b.com", Source: "S", SeenAt: 2,
- })
-
- unclassified, err := GetUnclassifiedStories("S")
- if err != nil {
- t.Fatal(err)
- }
- if len(unclassified) != 2 {
- t.Fatalf("unclassified = %d, want 2", len(unclassified))
- }
MarkClassified("g1", "tech", "[]")
- unclassified, err = GetUnclassifiedStories("S")
- if err != nil {
- t.Fatal(err)
- }
- if len(unclassified) != 1 {
- t.Fatalf("unclassified = %d, want 1", len(unclassified))
- }
- if unclassified[0].GUID != "g2" {
- t.Errorf("remaining = %q, want g2", unclassified[0].GUID)
- }
-}
-
-func TestRecentHeadlines(t *testing.T) {
- setupTestDB(t)
-
- // Insert with explicit timestamps to ensure ordering
- db := Get()
- db.Exec(`INSERT OR REPLACE INTO recent_headlines (guid, headline, source, seen_at) VALUES (?, ?, ?, ?)`,
- "g1", "Headline One", "Source A", 1000)
- db.Exec(`INSERT OR REPLACE INTO recent_headlines (guid, headline, source, seen_at) VALUES (?, ?, ?, ?)`,
- "g2", "Headline Two", "Source B", 2000)
- db.Exec(`INSERT OR REPLACE INTO recent_headlines (guid, headline, source, seen_at) VALUES (?, ?, ?, ?)`,
- "g3", "Headline Three", "Source A", 3000)
-
- headlines, err := GetRecentHeadlines(2)
- if err != nil {
- t.Fatal(err)
- }
- if len(headlines) != 2 {
- t.Fatalf("headlines = %d, want 2", len(headlines))
- }
- // Should be most recent first
- if headlines[0].GUID != "g3" {
- t.Errorf("first = %q, want g3", headlines[0].GUID)
- }
- if headlines[1].GUID != "g2" {
- t.Errorf("second = %q, want g2", headlines[1].GUID)
+ var channel string
+ var classified int
+ Get().QueryRow(`SELECT channel, classified FROM stories WHERE guid = ?`, "g1").
+ Scan(&channel, &classified)
+ if channel != "tech" || classified != 1 {
+ t.Errorf("after MarkClassified: channel=%q classified=%d, want tech/1", channel, classified)
}
}
@@ -269,57 +225,6 @@ func TestInsertReaction_DuplicateIgnored(t *testing.T) {
}
}
-func TestGetUnclassifiedStories_ScopedBySource(t *testing.T) {
- setupTestDB(t)
-
- InsertStory(&Story{GUID: "g1", Headline: "H1", ArticleURL: "https://a.com", Source: "SourceA", SeenAt: 1})
- InsertStory(&Story{GUID: "g2", Headline: "H2", ArticleURL: "https://b.com", Source: "SourceB", SeenAt: 2})
- InsertStory(&Story{GUID: "g3", Headline: "H3", ArticleURL: "https://c.com", Source: "SourceA", SeenAt: 3})
-
- got, err := GetUnclassifiedStories("SourceA")
- if err != nil {
- t.Fatal(err)
- }
- if len(got) != 2 {
- t.Fatalf("expected 2 stories for SourceA, got %d", len(got))
- }
-
- got, err = GetUnclassifiedStories("SourceB")
- if err != nil {
- t.Fatal(err)
- }
- if len(got) != 1 {
- t.Fatalf("expected 1 story for SourceB, got %d", len(got))
- }
-
- got, err = GetUnclassifiedStories("SourceC")
- if err != nil {
- t.Fatal(err)
- }
- if len(got) != 0 {
- t.Fatalf("expected 0 stories for SourceC, got %d", len(got))
- }
-}
-
-func TestGetUnclassifiedStories_Limit20(t *testing.T) {
- setupTestDB(t)
-
- for i := 0; i < 30; i++ {
- InsertStory(&Story{
- GUID: fmt.Sprintf("g%d", i), Headline: fmt.Sprintf("H%d", i),
- ArticleURL: "https://a.com", Source: "S", SeenAt: int64(i),
- })
- }
-
- got, err := GetUnclassifiedStories("S")
- if err != nil {
- t.Fatal(err)
- }
- if len(got) != 20 {
- t.Fatalf("expected 20 (capped), got %d", len(got))
- }
-}
-
func TestRunMaintenance(t *testing.T) {
setupTestDB(t)
db := Get()
@@ -341,11 +246,7 @@ func TestRunMaintenance(t *testing.T) {
InsertReaction("old", "tech", "$r1", "👍", "@u:x", old)
InsertReaction("new", "tech", "$r2", "👍", "@u:x", now)
- // Insert old + recent headlines
- db.Exec(`INSERT OR REPLACE INTO recent_headlines (guid, headline, source, seen_at) VALUES (?, ?, ?, ?)`, "old", "Old", "S", old)
- db.Exec(`INSERT OR REPLACE INTO recent_headlines (guid, headline, source, seen_at) VALUES (?, ?, ?, ?)`, "new", "New", "S", now)
-
- RunMaintenance(24, 7)
+ RunMaintenance()
// Old stories should be pruned
var count int
@@ -363,11 +264,6 @@ func TestRunMaintenance(t *testing.T) {
if count != 1 {
t.Errorf("reactions: expected 1, got %d", count)
}
-
- db.QueryRow(`SELECT COUNT(*) FROM recent_headlines`).Scan(&count)
- if count != 1 {
- t.Errorf("recent_headlines: expected 1, got %d", count)
- }
}
func TestFTS5Search(t *testing.T) {
diff --git a/internal/storage/types.go b/internal/storage/types.go
index 05e97f3..88f1a22 100644
--- a/internal/storage/types.go
+++ b/internal/storage/types.go
@@ -2,26 +2,17 @@ package storage
// Story is the core record for an ingested news item.
type Story struct {
- ID int64
- GUID string
- Headline string
- Lede string
+ ID int64
+ GUID string
+ Headline string
+ Lede string
ImageURL string
ArticleURL string
URLCanonical string
HeadlineNorm string
Source string
- FeedHint string
- Platforms string // JSON array e.g. '["nintendo-switch","multi"]'
- Channel string
- Classified bool
- SeenAt int64
-}
-
-// RecentHeadline is a lightweight record for the LLM dedup context window.
-type RecentHeadline struct {
- GUID string
- Headline string
- Source string
- SeenAt int64
+ Platforms string // JSON array e.g. '["nintendo-switch","multi"]'
+ Channel string
+ Classified bool
+ SeenAt int64
}
diff --git a/main.go b/main.go
index 5837d28..35ae4e5 100644
--- a/main.go
+++ b/main.go
@@ -10,9 +10,7 @@ import (
"syscall"
"time"
- "pete/internal/classifier"
"pete/internal/config"
- "pete/internal/explainer"
"pete/internal/ingestion"
"pete/internal/matrix"
"pete/internal/poster"
@@ -43,7 +41,6 @@ func main() {
slog.Info("config loaded",
"sources", len(cfg.Sources),
"channels", len(cfg.Matrix.Channels),
- "ollama_model", cfg.Ollama.Model,
)
// Initialize database
@@ -72,17 +69,11 @@ func main() {
os.Exit(1)
}
- // Create Ollama client and classifier
- ollamaClient := classifier.NewOllamaClient(cfg.Ollama)
- cls := classifier.NewClassifier(ollamaClient, mx)
-
// Create post queue
queue := poster.NewQueue(cfg.Posting, mx)
- // Wire reaction handler + ❓ → summary hook
+ // Wire reaction handler
mx.SetReactionHandler(poster.HandleReaction)
- exp := explainer.New(mx, ollamaClient)
- poster.SetReactionCallback(exp.Handle)
// Wire !post command: force-publish next queued story for the room's channel.
mx.SetMessageHandler(func(channel string, roomID id.RoomID, eventID id.EventID, sender id.UserID, body string) {
@@ -142,70 +133,37 @@ func main() {
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 {
- slog.Error("classification failed, will retry next cycle",
- "guid", item.GUID,
- "err", err,
- )
- return // story stays unclassified for retry
- }
-
- // Add to recent headlines for future dedup
- storage.InsertRecentHeadline(item.GUID, item.Headline, item.Source)
-
- if !result.ShouldPost {
- // Mark classified with sentinel channel so it's not retried
+ // Pipeline callback: route the story to its direct_route channel, then
+ // either enqueue immediately or leave it for the round-robin scheduler.
+ processItem := func(_ context.Context, item *ingestion.FeedItem) {
+ if item.DirectRoute == "" {
+ slog.Error("item has no direct_route, skipping", "guid", item.GUID, "source", item.Source)
storage.MarkClassified(item.GUID, "_discarded", "[]")
- slog.Info("story gated out",
- "guid", item.GUID,
- "rationale", result.Rationale,
- )
return
}
- // Mark classified with actual channel
- platforms := storage.MarshalPlatforms(result.Platforms)
- 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,
- )
- return
- }
+ channel := item.DirectRoute
+ storage.MarkClassified(item.GUID, channel, "[]")
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)
+ slog.Info("story routed, awaiting round-robin tick",
+ "guid", item.GUID, "source", item.Source, "channel", channel)
return
}
- // Validate image
imageURL := ""
if item.ImageURL != "" && ingestion.ValidateImageURL(item.ImageURL) {
imageURL = item.ImageURL
}
- // Enqueue for posting
queue.Enqueue(poster.QueueItem{
GUID: item.GUID,
Headline: item.Headline,
Lede: item.Lede,
ImageURL: imageURL,
ArticleURL: item.ArticleURL,
- Source: item.Source,
- Channel: result.Channel,
- Platforms: result.Platforms,
+ Source: item.Source,
+ Channel: channel,
})
}
@@ -238,7 +196,7 @@ func main() {
}
// Run maintenance on startup
- storage.RunMaintenance(cfg.Storage.RecentWindowHours, cfg.Storage.ClassificationLogDays)
+ storage.RunMaintenance()
// Wait for shutdown signal
sig := <-sigCh
@@ -272,15 +230,13 @@ func runSeed(cfg *config.Config) {
Lede: item.Lede,
ImageURL: item.ImageURL,
ArticleURL: item.ArticleURL,
- Source: src.Name,
- FeedHint: src.FeedHint,
+ Source: src.Name,
Classified: true,
SeenAt: timeNow(),
}); err != nil {
slog.Error("seed: insert failed", "guid", item.GUID, "err", err)
continue
}
- storage.InsertRecentHeadline(item.GUID, item.Headline, src.Name)
total++
}
slog.Info("seed: source done", "source", src.Name, "items", len(items))
@@ -307,8 +263,6 @@ func runTest(cfg *config.Config, sourceName string) {
if len(items) > 0 {
item := items[0]
item.Source = s.Name
- item.FeedHint = s.FeedHint
- item.Tier = s.Tier
item.DirectRoute = s.DirectRoute
testItem = &item
src = s
@@ -326,10 +280,10 @@ func runTest(cfg *config.Config, sourceName string) {
"guid", testItem.GUID,
)
- // Determine channel
- channel := "politics" // default
- if testItem.DirectRoute != nil {
- channel = *testItem.DirectRoute
+ channel := testItem.DirectRoute
+ if channel == "" {
+ slog.Error("test: source has no direct_route", "source", src.Name)
+ os.Exit(1)
}
// Validate image
@@ -351,7 +305,7 @@ func runTest(cfg *config.Config, sourceName string) {
Headline: testItem.Headline,
ArticleURL: testItem.ArticleURL,
Lede: testItem.Lede,
- Source: testItem.Source,
+ Source: testItem.Source,
Channel: channel,
}
diff --git a/main_test.go b/main_test.go
index 7dbd418..7928575 100644
--- a/main_test.go
+++ b/main_test.go
@@ -58,7 +58,7 @@ func TestRunSeed_IngestsStories(t *testing.T) {
FeedURL: ts.URL,
Tier: 1,
PollIntervalMinutes: 20,
- FeedHint: "tech",
+ DirectRoute: "tech",
Enabled: true,
},
},
@@ -74,13 +74,11 @@ func TestRunSeed_IngestsStories(t *testing.T) {
}
}
- // Should have no unclassified stories (seed marks classified=true)
- unclassified, err := storage.GetUnclassifiedStories("Test Feed")
- if err != nil {
- t.Fatal(err)
- }
- if len(unclassified) != 0 {
- t.Errorf("expected 0 unclassified after seed, got %d", len(unclassified))
+ // Seed marks rows classified=1; confirm none remain unclassified.
+ var unclassified int
+ storage.Get().QueryRow(`SELECT COUNT(*) FROM stories WHERE classified = 0`).Scan(&unclassified)
+ if unclassified != 0 {
+ t.Errorf("expected 0 unclassified after seed, got %d", unclassified)
}
}
@@ -96,7 +94,7 @@ func TestRunSeed_Idempotent(t *testing.T) {
FeedURL: ts.URL,
Tier: 1,
PollIntervalMinutes: 20,
- FeedHint: "tech",
+ DirectRoute: "tech",
Enabled: true,
},
},
@@ -139,35 +137,6 @@ func TestRunSeed_SkipsDisabledSources(t *testing.T) {
}
}
-func TestRunSeed_RecentHeadlinesPopulated(t *testing.T) {
- setupTestDB(t)
- ts := serveFeed(t, 2)
- defer ts.Close()
-
- cfg := &config.Config{
- Sources: []config.SourceConfig{
- {
- Name: "Test Feed",
- FeedURL: ts.URL,
- Tier: 1,
- PollIntervalMinutes: 20,
- FeedHint: "tech",
- Enabled: true,
- },
- },
- }
-
- runSeed(cfg)
-
- headlines, err := storage.GetRecentHeadlines(10)
- if err != nil {
- t.Fatal(err)
- }
- if len(headlines) != 2 {
- t.Errorf("expected 2 recent headlines after seed, got %d", len(headlines))
- }
-}
-
func TestRunSeed_BadFeedURL(t *testing.T) {
setupTestDB(t)