diff --git a/internal/ingestion/parser.go b/internal/ingestion/parser.go index bd3036b..d3bceb8 100644 --- a/internal/ingestion/parser.go +++ b/internal/ingestion/parser.go @@ -28,6 +28,7 @@ type FeedItem struct { ArticleURL string Source string DirectRoute string + PublishedAt int64 // unix seconds from RSS pubDate / Atom published (or updated as fallback); 0 if absent or unparseable } var ( @@ -53,11 +54,12 @@ func FetchFeed(feedURL string) ([]FeedItem, error) { items := make([]FeedItem, 0, len(feed.Items)) for _, item := range feed.Items { fi := FeedItem{ - GUID: itemGUID(item), - Headline: strings.TrimSpace(item.Title), - Lede: extractLede(item.Description), - ImageURL: NormalizeImageURL(extractImageURL(item)), - ArticleURL: strings.TrimSpace(item.Link), + GUID: itemGUID(item), + Headline: strings.TrimSpace(item.Title), + Lede: extractLede(item.Description), + ImageURL: NormalizeImageURL(extractImageURL(item)), + ArticleURL: strings.TrimSpace(item.Link), + PublishedAt: itemPublished(item), } if fi.GUID == "" || fi.ArticleURL == "" { continue @@ -76,6 +78,19 @@ func itemGUID(item *gofeed.Item) string { return item.Link } +// itemPublished returns the item's publish time as unix seconds, falling back +// to the updated time if pubDate was absent. Returns 0 when neither parses; +// the poller clamps future timestamps to ingest time. +func itemPublished(item *gofeed.Item) int64 { + if item.PublishedParsed != nil { + return item.PublishedParsed.Unix() + } + if item.UpdatedParsed != nil { + return item.UpdatedParsed.Unix() + } + return 0 +} + // extractLede returns the feed description verbatim, with HTML stripped. func extractLede(desc string) string { if desc == "" { diff --git a/internal/ingestion/poller.go b/internal/ingestion/poller.go index eafe4ab..926e7ef 100644 --- a/internal/ingestion/poller.go +++ b/internal/ingestion/poller.go @@ -188,7 +188,15 @@ func (p *Poller) pollOnceWithErr(ctx context.Context, src config.SourceConfig) e items[i].Source = src.Name items[i].DirectRoute = src.DirectRoute - // Store immediately (before routing) to prevent re-ingestion + // Store immediately (before routing) to prevent re-ingestion. + // Future-dated pubDates (publishers occasionally post-date drafts or + // have a misconfigured clock) get clamped to ingest time so they don't + // stick at the top of the queue indefinitely. + seenAt := time.Now().Unix() + publishedAt := items[i].PublishedAt + if publishedAt > seenAt { + publishedAt = seenAt + } if err := storage.InsertStory(&storage.Story{ GUID: items[i].GUID, Headline: items[i].Headline, @@ -199,7 +207,8 @@ func (p *Poller) pollOnceWithErr(ctx context.Context, src config.SourceConfig) e HeadlineNorm: headlineNorm, Source: items[i].Source, Paywalled: paywalled, - SeenAt: time.Now().Unix(), + SeenAt: seenAt, + PublishedAt: publishedAt, }); err != nil { slog.Error("failed to insert story", "guid", items[i].GUID, "err", err) continue diff --git a/internal/storage/db.go b/internal/storage/db.go index c7bb2e9..ba1a346 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, "stories", "paywalled", "INTEGER NOT NULL DEFAULT 0") + addColumnIfMissing(d, "stories", "published_at", "INTEGER") addColumnIfMissing(d, "post_log", "url_canonical", "TEXT") addColumnIfMissing(d, "round_robin_state", "last_channel", "TEXT") diff --git a/internal/storage/queries.go b/internal/storage/queries.go index f615368..7d29521 100644 --- a/internal/storage/queries.go +++ b/internal/storage/queries.go @@ -33,10 +33,14 @@ func InsertStory(s *Story) error { if s.Paywalled { paywalled = 1 } + var publishedAt any + if s.PublishedAt > 0 { + publishedAt = s.PublishedAt + } _, err := Get().Exec( - `INSERT INTO stories (guid, headline, lede, image_url, article_url, url_canonical, headline_norm, source, platforms, channel, classified, paywalled, 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, paywalled, s.SeenAt, + `INSERT INTO stories (guid, headline, lede, image_url, article_url, url_canonical, headline_norm, source, platforms, channel, classified, paywalled, seen_at, published_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, paywalled, s.SeenAt, publishedAt, ) return err } @@ -185,7 +189,7 @@ func GetNewestPostableStory(source string) (*Story, error) { 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 + ORDER BY COALESCE(published_at, 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.Platforms, &s.Channel, &s.SeenAt); err != nil { @@ -208,7 +212,7 @@ func GetNewestPostableStoryByChannel(channel string) (*Story, error) { WHERE classified = 1 AND channel = ? AND guid NOT IN (SELECT guid FROM post_log) - ORDER BY seen_at DESC + ORDER BY COALESCE(published_at, 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.Platforms, &s.Channel, &s.SeenAt); err != nil { @@ -230,7 +234,7 @@ func ListClassifiedByChannel(channel string, limit, offset int) ([]Story, error) FROM stories s WHERE s.classified = 1 AND s.channel = ? - ORDER BY s.seen_at DESC + ORDER BY COALESCE(s.published_at, s.seen_at) DESC LIMIT ? OFFSET ?`, channel, limit, offset) if err != nil { return nil, err @@ -257,7 +261,7 @@ func ListAllClassified(limit, offset int) ([]Story, error) { WHERE s.classified = 1 AND s.channel IS NOT NULL AND s.channel NOT IN ('_discarded', '_duplicate') - ORDER BY s.seen_at DESC + ORDER BY COALESCE(s.published_at, s.seen_at) DESC LIMIT ? OFFSET ?`, limit, offset) if err != nil { return nil, err diff --git a/internal/storage/schema.go b/internal/storage/schema.go index fb43339..360aa96 100644 --- a/internal/storage/schema.go +++ b/internal/storage/schema.go @@ -15,7 +15,8 @@ CREATE TABLE IF NOT EXISTS stories ( channel TEXT, classified INTEGER NOT NULL DEFAULT 0, paywalled INTEGER NOT NULL DEFAULT 0, - seen_at INTEGER NOT NULL + seen_at INTEGER NOT NULL, + published_at INTEGER ); CREATE TABLE IF NOT EXISTS post_log ( diff --git a/internal/storage/types.go b/internal/storage/types.go index 94d7b50..d81fab7 100644 --- a/internal/storage/types.go +++ b/internal/storage/types.go @@ -16,5 +16,6 @@ type Story struct { Classified bool Paywalled bool // source article is gated; ArticleURL may be an archive snapshot SeenAt int64 - Posted bool // true if this story has been posted to Matrix (joined from post_log) + PublishedAt int64 // from RSS pubDate/Atom published; 0 if missing. Future dates are clamped to SeenAt at ingest. + Posted bool // true if this story has been posted to Matrix (joined from post_log) }