fbd4b84eaf
Serves Pete's classified-story archive over HTTP alongside the Matrix
bot. Three sections (gaming/tech/politics), Animal-Crossing-vibe Tailwind
templates, day/night palette driven by the visitor's browser clock.
Web port configurable via web.listen_addr and ${PETE_WEB_PORT} in
docker-compose. Tailwind built in a node stage in the Dockerfile so
deployments don't need node at runtime.
358 lines
12 KiB
Go
358 lines
12 KiB
Go
package storage
|
|
|
|
import (
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"log/slog"
|
|
"time"
|
|
)
|
|
|
|
func nowUnix() int64 {
|
|
return time.Now().Unix()
|
|
}
|
|
|
|
// IsGUIDSeen checks if a GUID has already been ingested.
|
|
func IsGUIDSeen(guid string) bool {
|
|
var count int
|
|
if err := Get().QueryRow(`SELECT COUNT(*) FROM stories WHERE guid = ?`, guid).Scan(&count); err != nil {
|
|
slog.Error("IsGUIDSeen query failed", "guid", guid, "err", err)
|
|
return true // fail closed: assume seen to prevent duplicates
|
|
}
|
|
return count > 0
|
|
}
|
|
|
|
// InsertStory inserts a new story record.
|
|
func InsertStory(s *Story) error {
|
|
classified := 0
|
|
if s.Classified {
|
|
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,
|
|
)
|
|
return err
|
|
}
|
|
|
|
// nullIfEmpty returns a SQL NULL for empty strings so partial unique indexes
|
|
// (WHERE col IS NOT NULL) can hold multiple "unknown" rows without conflict.
|
|
func nullIfEmpty(s string) any {
|
|
if s == "" {
|
|
return nil
|
|
}
|
|
return s
|
|
}
|
|
|
|
// IsCanonicalSeen reports whether any story with this canonical URL exists.
|
|
// Empty input always returns false (no canonical = no dedup possible).
|
|
func IsCanonicalSeen(canonical string) bool {
|
|
if canonical == "" {
|
|
return false
|
|
}
|
|
var n int
|
|
if err := Get().QueryRow(`SELECT COUNT(*) FROM stories WHERE url_canonical = ?`, canonical).Scan(&n); err != nil {
|
|
slog.Error("IsCanonicalSeen query failed", "err", err)
|
|
return true // fail closed
|
|
}
|
|
return n > 0
|
|
}
|
|
|
|
// IsHeadlineSeen reports whether a same-source story with this normalized
|
|
// headline already exists. Empty inputs return false.
|
|
func IsHeadlineSeen(source, headlineNorm string) bool {
|
|
if source == "" || headlineNorm == "" {
|
|
return false
|
|
}
|
|
var n int
|
|
if err := Get().QueryRow(
|
|
`SELECT COUNT(*) FROM stories WHERE source = ? AND headline_norm = ?`,
|
|
source, headlineNorm).Scan(&n); err != nil {
|
|
slog.Error("IsHeadlineSeen query failed", "err", err)
|
|
return true
|
|
}
|
|
return n > 0
|
|
}
|
|
|
|
// WasCanonicalPostedRecently reports whether the canonical URL was posted to
|
|
// the given channel within `cooldownSeconds`. Empty canonical returns false.
|
|
func WasCanonicalPostedRecently(canonical, channel string, cooldownSeconds int64) bool {
|
|
if canonical == "" || cooldownSeconds <= 0 {
|
|
return false
|
|
}
|
|
cutoff := nowUnix() - cooldownSeconds
|
|
var n int
|
|
if err := Get().QueryRow(
|
|
`SELECT COUNT(*) FROM post_log WHERE url_canonical = ? AND channel = ? AND posted_at >= ?`,
|
|
canonical, channel, cutoff).Scan(&n); err != nil {
|
|
slog.Error("WasCanonicalPostedRecently query failed", "err", err)
|
|
return true // fail closed
|
|
}
|
|
return n > 0
|
|
}
|
|
|
|
// MarkClassified marks a story as successfully classified and sets its channel.
|
|
func MarkClassified(guid, channel, platforms string) {
|
|
exec("mark classified",
|
|
`UPDATE stories SET classified = 1, channel = ?, platforms = ? WHERE guid = ?`,
|
|
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
|
|
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 {
|
|
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) {
|
|
exec("insert post_log",
|
|
`INSERT OR IGNORE INTO post_log (guid, channel, event_id, url_canonical, posted_at) VALUES (?, ?, ?, ?, ?)`,
|
|
guid, channel, eventID, nullIfEmpty(urlCanonical), nowUnix())
|
|
}
|
|
|
|
// GetLastPostTime returns the unix timestamp of the most recent post to a channel.
|
|
func GetLastPostTime(channel string) int64 {
|
|
var t int64
|
|
if err := Get().QueryRow(`SELECT COALESCE(MAX(posted_at), 0) FROM post_log WHERE channel = ?`, channel).Scan(&t); err != nil {
|
|
slog.Error("GetLastPostTime query failed", "channel", channel, "err", err)
|
|
return nowUnix() // fail closed: pretend we just posted to prevent flooding
|
|
}
|
|
return t
|
|
}
|
|
|
|
// CountAllPostsInWindow counts posts across ALL channels within a time window.
|
|
// Used for the global daily cap.
|
|
func CountAllPostsInWindow(windowStart int64) int {
|
|
var count int
|
|
if err := Get().QueryRow(`SELECT COUNT(*) FROM post_log WHERE posted_at >= ?`, windowStart).Scan(&count); err != nil {
|
|
slog.Error("CountAllPostsInWindow query failed", "err", err)
|
|
return 1<<31 - 1 // fail closed: huge number prevents posting
|
|
}
|
|
return count
|
|
}
|
|
|
|
// CountPostsInWindow counts posts to a channel within a time window.
|
|
func CountPostsInWindow(channel string, windowStart int64) int {
|
|
var count int
|
|
if err := Get().QueryRow(`SELECT COUNT(*) FROM post_log WHERE channel = ? AND posted_at >= ?`, channel, windowStart).Scan(&count); err != nil {
|
|
slog.Error("CountPostsInWindow query failed", "channel", channel, "err", err)
|
|
return 999 // fail closed: pretend burst cap reached to prevent flooding
|
|
}
|
|
return count
|
|
}
|
|
|
|
// LookupPostGUID finds the story GUID for a given Matrix event ID.
|
|
func LookupPostGUID(eventID string) (guid, channel string, found bool) {
|
|
err := Get().QueryRow(
|
|
`SELECT guid, channel FROM post_log WHERE event_id = ? LIMIT 1`, eventID).Scan(&guid, &channel)
|
|
if err != nil {
|
|
return "", "", false
|
|
}
|
|
return guid, channel, true
|
|
}
|
|
|
|
// InsertReaction records a reaction on a posted story.
|
|
// Uses OR IGNORE to deduplicate if Matrix delivers the same reaction twice.
|
|
func InsertReaction(postGUID, channel, eventID, emoji, userID string, reactedAt int64) {
|
|
exec("insert reaction",
|
|
`INSERT OR IGNORE INTO reactions (post_guid, channel, event_id, emoji, user_id, reacted_at) VALUES (?, ?, ?, ?, ?, ?)`,
|
|
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
|
|
}
|
|
|
|
// GetNewestPostableStoryByChannel returns the newest classified story routed
|
|
// to the given channel that has not yet been posted. Used by !post to satisfy
|
|
// on-demand requests in round-robin mode (where the in-memory queue is empty
|
|
// 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
|
|
FROM stories
|
|
WHERE classified = 1
|
|
AND channel = ?
|
|
AND guid NOT IN (SELECT guid FROM post_log)
|
|
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 errors.Is(err, sql.ErrNoRows) {
|
|
return nil, nil
|
|
}
|
|
return nil, err
|
|
}
|
|
return &s, nil
|
|
}
|
|
|
|
// ListClassifiedByChannel returns up to `limit` classified stories routed to a
|
|
// real channel, newest first, with optional offset for pagination. Sentinel
|
|
// 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
|
|
FROM stories
|
|
WHERE classified = 1
|
|
AND channel = ?
|
|
ORDER BY seen_at DESC
|
|
LIMIT ? OFFSET ?`, channel, limit, offset)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
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 {
|
|
return nil, err
|
|
}
|
|
out = append(out, s)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// CountClassifiedByChannel returns how many classified stories exist for a channel.
|
|
func CountClassifiedByChannel(channel string) (int, error) {
|
|
var n int
|
|
err := Get().QueryRow(
|
|
`SELECT COUNT(*) FROM stories WHERE classified = 1 AND channel = ?`,
|
|
channel).Scan(&n)
|
|
return n, err
|
|
}
|
|
|
|
// 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() (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 lc != nil {
|
|
lastChannel = *lc
|
|
}
|
|
return lastChannel, lastTickAt, nil
|
|
}
|
|
|
|
// 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_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.
|
|
func MarshalPlatforms(platforms []string) string {
|
|
if len(platforms) == 0 {
|
|
return "[]"
|
|
}
|
|
data, err := json.Marshal(platforms)
|
|
if err != nil {
|
|
slog.Error("marshal platforms failed", "err", err)
|
|
return "[]"
|
|
}
|
|
return string(data)
|
|
}
|
|
|
|
// UnmarshalPlatforms converts a JSON array string back to a string slice.
|
|
func UnmarshalPlatforms(raw string) []string {
|
|
if raw == "" || raw == "[]" {
|
|
return nil
|
|
}
|
|
var platforms []string
|
|
if err := json.Unmarshal([]byte(raw), &platforms); err != nil {
|
|
slog.Error("unmarshal platforms failed", "err", err, "raw", raw)
|
|
return nil
|
|
}
|
|
return platforms
|
|
}
|