package main
import (
"context"
"flag"
"fmt"
"html/template"
"log/slog"
"os"
"os/signal"
"strconv"
"strings"
"syscall"
"time"
"pete/internal/config"
"pete/internal/ingestion"
"pete/internal/matrix"
"pete/internal/poster"
"pete/internal/scheduler"
"pete/internal/storage"
"pete/internal/web"
webpush "github.com/SherClockHolmes/webpush-go"
"maunium.net/go/mautrix/id"
)
// runGenVAPID prints a fresh VAPID keypair for the [web.push] config block.
func runGenVAPID() {
priv, pub, err := webpush.GenerateVAPIDKeys()
if err != nil {
fmt.Fprintln(os.Stderr, "genvapid: failed:", err)
os.Exit(1)
}
fmt.Println("# Add these under [web.push] in your config:")
fmt.Printf("vapid_public_key = %q\n", pub)
fmt.Printf("vapid_private_key = %q\n", priv)
}
func main() {
configPath := flag.String("config", "config.toml", "path to config file")
seed := flag.Bool("seed", false, "ingest current feed items as seen without posting, then exit")
test := flag.Bool("test", false, "post one story to verify the full pipeline, then exit")
testSource := flag.String("test-source", "", "source name to use for -test (default: first enabled)")
local := flag.Bool("local", false, "web/RSS-only mode: poll feeds and serve the web UI, no Matrix login or posting")
backfillPaywall := flag.Bool("backfill-paywall", false, "re-check paywalled rows with current detection logic; clear false positives and fill missing thumbnails, then exit")
genVAPID := flag.Bool("genvapid", false, "generate a VAPID keypair for web.push and print it, then exit")
flag.Parse()
// -genvapid needs neither config nor database; handle it before anything else.
if *genVAPID {
runGenVAPID()
return
}
slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{
Level: slog.LevelInfo,
})))
// Load config
cfg, err := config.Load(*configPath)
if err != nil {
slog.Error("failed to load config", "err", err)
os.Exit(1)
}
slog.Info("config loaded",
"sources", len(cfg.Sources),
"channels", len(cfg.Matrix.Channels),
)
// Initialize database
if err := storage.Init(cfg.Storage.DBPath); err != nil {
slog.Error("database init failed", "err", err)
os.Exit(1)
}
defer storage.Close()
// Seed mode: ingest all current feed items as seen, then exit
if *seed {
runSeed(cfg)
return
}
if *backfillPaywall {
runBackfillPaywall()
return
}
// Test mode: post one story to verify the full pipeline, then exit
if *test {
runTest(cfg, *testSource)
return
}
// Local mode: poll feeds + serve web UI only. No Matrix, no posting queue.
if *local {
runLocal(cfg)
return
}
// Create Matrix client
mx, err := matrix.New(cfg.Matrix)
if err != nil {
slog.Error("matrix client init failed", "err", err)
os.Exit(1)
}
// Create post queue
queue := poster.NewQueue(cfg.Posting, mx)
// Wire reaction handler
mx.SetReactionHandler(poster.HandleReaction)
// Allowlist of users permitted to run in-room commands. The channel rooms
// are public, so commands are gated to configured admins; an empty list
// disables them entirely (fail closed) rather than allowing anyone.
admins := make(map[id.UserID]bool, len(cfg.Matrix.Admins))
for _, a := range cfg.Matrix.Admins {
if a = strings.TrimSpace(a); a != "" {
admins[id.UserID(a)] = true
}
}
// Wire in-room commands. All are gated to the admin allowlist; the room
// rooms are public so anything not from an admin is silently ignored.
mx.SetMessageHandler(func(channel string, roomID id.RoomID, eventID id.EventID, sender id.UserID, body string) {
cmd := strings.TrimSpace(body)
if cmd != "!post" && cmd != "!petestats" {
return
}
if !admins[sender] {
slog.Warn("command ignored: sender not in matrix.admins allowlist", "cmd", cmd, "channel", channel, "sender", sender)
return
}
if cmd == "!petestats" {
slog.Info("!petestats requested", "channel", channel, "sender", sender)
plain, htmlBody := formatStats(storage.GetMetricsSummary())
if err := mx.PostThreadedReply(channel, eventID, plain, htmlBody); err != nil {
slog.Warn("!petestats: failed to send reply", "err", err)
}
return
}
// !post: force-publish the next queued story for the room's channel.
slog.Info("!post requested", "channel", channel, "sender", sender)
if queue.ForcePost(channel) {
return
}
// In-memory queue empty (common in round-robin mode). Fall back to the
// newest classified, not-yet-posted story for this channel.
story, err := storage.GetNewestPostableStoryByChannel(channel)
if err != nil {
slog.Error("!post: db lookup failed", "channel", channel, "err", err)
return
}
if story != nil {
imageURL := ""
if story.ImageURL != "" && ingestion.ValidateImageURL(story.ImageURL) {
imageURL = story.ImageURL
}
queue.PostNow(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),
})
return
}
if err := mx.PostThreadedReply(channel, eventID,
"nothing available for "+channel,
"nothing available for "+channel+"",
); err != nil {
slog.Warn("!post: failed to send empty reply", "err", err)
}
})
// Set up graceful shutdown
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
sigCh := make(chan os.Signal, 1)
signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
// Start Matrix sync loop
mx.Start(ctx)
slog.Info("matrix sync started")
// Start post queue
go queue.Start(ctx)
slog.Info("post queue started")
postingEnabled := *cfg.Posting.Enabled
roundRobinMode := cfg.Posting.RoundRobin.Enabled
if !postingEnabled {
slog.Info("automatic Matrix posting is disabled (posting.enabled=false); stories will still be ingested and served to the web UI, and !post still works")
}
// 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", "[]")
return
}
channel := item.DirectRoute
storage.MarkClassified(item.GUID, channel, "[]")
// Posting disabled: classify and keep for the web UI / manual !post,
// but never auto-enqueue to Matrix.
if !postingEnabled {
return
}
if roundRobinMode {
slog.Info("story routed, awaiting round-robin tick",
"guid", item.GUID, "source", item.Source, "channel", channel)
return
}
imageURL := ""
if item.ImageURL != "" && ingestion.ValidateImageURL(item.ImageURL) {
imageURL = item.ImageURL
}
queue.Enqueue(poster.QueueItem{
GUID: item.GUID,
Headline: item.Headline,
Lede: item.Lede,
ImageURL: imageURL,
ArticleURL: item.ArticleURL,
Source: item.Source,
Channel: channel,
})
}
// Start pollers
poller := ingestion.NewPoller(cfg.Sources, processItem)
poller.Start(ctx)
slog.Info("pollers started")
if postingEnabled && roundRobinMode {
channelNames := make([]string, 0, len(cfg.Matrix.Channels))
for name := range cfg.Matrix.Channels {
// The adventure channel drives its own posting: priority beats go
// live at ingest, bulletins wait for the daily digest. Leaving it in
// the rotation would let the scheduler post bulletins one-by-one
// (they're "classified, not yet posted") and steal them from the digest.
if cfg.Adventure.Enabled && name == cfg.Adventure.Channel {
continue
}
channelNames = append(channelNames, name)
}
rr := scheduler.New(channelNames, cfg.Posting.RoundRobin.IntervalHours, queue)
if rr == nil {
slog.Warn("round-robin enabled but no channels are configured; nothing will post")
} else {
go rr.Start(ctx)
}
}
// Adapter: the web ingest handler posts priority adventure beats live to
// Matrix through the same queue everything else uses (bypasses pacing).
//
// Adventure carries its own enable switch and is push-based: facts arrive
// from gogobee at ingest, they are not polled, classified or paced, and they
// never enter the round-robin rotation. So it is gated on [adventure] alone,
// NOT on posting.enabled — that flag governs the RSS pipeline's chatter, and
// an operator running "news on the web only, adventure live in the games
// room" is a legitimate configuration. A nil poster (adventure off, or no
// channel) still keeps both the live beats and the digest loop shut.
var advPost web.PriorityPoster
if cfg.Adventure.Enabled && cfg.Adventure.Channel != "" {
advPost = func(p web.AdvPost) {
queue.PostNow(poster.QueueItem{
GUID: p.GUID,
Headline: p.Headline,
Lede: p.Lede,
ImageURL: p.ImageURL,
ArticleURL: p.ArticleURL,
Source: p.Source,
Channel: p.Channel,
})
}
}
// Start the read-only web UI alongside the Matrix bot.
if cfg.Web.Enabled {
ws, err := web.New(cfg.Web, cfg.Sources, postingEnabled, cfg.Adventure, advPost)
if err != nil {
slog.Error("web server init failed", "err", err)
} else {
go ws.Start(ctx)
ws.StartPushSender(ctx)
ws.StartAdventureDigest(ctx)
ws.StartTriviaBank(ctx)
}
}
// Run maintenance on startup
storage.RunMaintenance()
// Wait for shutdown signal
sig := <-sigCh
slog.Info("shutdown signal received", "signal", sig)
cancel()
poller.Wait()
queue.Wait()
mx.Stop()
slog.Info("pete stopped")
}
// formatStats renders a usage summary for the !petestats reply, returning a
// plain-text and an HTML body. Daily uniques are reported per-day only: the
// privacy salt rotates each day, so they cannot be honestly summed across days.
func formatStats(m storage.MetricsSummary) (plain, htmlBody string) {
const topN = 8
var p, h strings.Builder
p.WriteString("📊 Pete usage\n")
h.WriteString("📊 Pete usage
")
fmt.Fprintf(&p, "Today: %s views · ~%s unique visitors\n",
comma(m.ViewsToday), comma(m.UniquesToday))
fmt.Fprintf(&h, "Today: %s views · ~%s unique visitors
",
comma(m.ViewsToday), comma(m.UniquesToday))
fmt.Fprintf(&p, "All time: %s views\n", comma(m.TotalViews))
fmt.Fprintf(&h, "All time: %s views
", comma(m.TotalViews))
if len(m.Pages) > 0 {
pages := m.Pages
if len(pages) > topN {
pages = pages[:topN]
}
parts := make([]string, 0, len(pages))
for _, pg := range pages {
parts = append(parts, fmt.Sprintf("%s %s", pg.Path, comma(pg.Total)))
}
fmt.Fprintf(&p, "Top pages (all time): %s\n", strings.Join(parts, " · "))
fmt.Fprintf(&h, "Top pages (all time): %s
",
template.HTMLEscapeString(strings.Join(parts, " · ")))
}
if len(m.Last7Days) > 0 {
parts := make([]string, 0, len(m.Last7Days))
for _, d := range m.Last7Days {
label := time.Unix(d.Day*86400, 0).UTC().Format("Mon")
parts = append(parts, fmt.Sprintf("%s %s", label, comma(d.Uniques)))
}
fmt.Fprintf(&p, "Daily uniques (7d): %s", strings.Join(parts, " · "))
fmt.Fprintf(&h, "Daily uniques (7d): %s",
template.HTMLEscapeString(strings.Join(parts, " · ")))
}
return p.String(), h.String()
}
// comma formats a non-negative int with thousands separators (e.g. 12503 →
// "12,503").
func comma(n int) string {
s := strconv.Itoa(n)
if n < 0 {
return s // not expected for counts; leave untouched
}
var b strings.Builder
for i, c := range s {
if i > 0 && (len(s)-i)%3 == 0 {
b.WriteByte(',')
}
b.WriteRune(c)
}
return b.String()
}
func runLocal(cfg *config.Config) {
cfg.Web.Enabled = true
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
sigCh := make(chan os.Signal, 1)
signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
processItem := func(_ context.Context, item *ingestion.FeedItem) {
if item.DirectRoute == "" {
slog.Info("local: item has no direct_route, discarding", "guid", item.GUID, "source", item.Source)
storage.MarkClassified(item.GUID, "_discarded", "[]")
return
}
storage.MarkClassified(item.GUID, item.DirectRoute, "[]")
slog.Info("local: story classified",
"guid", item.GUID, "source", item.Source, "channel", item.DirectRoute, "headline", item.Headline)
}
poller := ingestion.NewPoller(cfg.Sources, processItem)
poller.Start(ctx)
slog.Info("local: pollers started")
// Local mode never connects to Matrix, so it's always web-only: adventure
// stories still ingest and render, but nothing posts (nil priority poster).
ws, err := web.New(cfg.Web, cfg.Sources, false, cfg.Adventure, nil)
if err != nil {
slog.Error("local: web server init failed", "err", err)
os.Exit(1)
}
go ws.Start(ctx)
// Push only needs the web auth layer, not Matrix, so a web-only deployment
// still runs the digest sender — otherwise subscriptions accumulate but no
// digest ever fires.
ws.StartPushSender(ctx)
slog.Info("local: web UI listening", "addr", cfg.Web.ListenAddr)
storage.RunMaintenance()
sig := <-sigCh
slog.Info("local: shutdown signal received", "signal", sig)
cancel()
poller.Wait()
slog.Info("pete stopped")
}
func runSeed(cfg *config.Config) {
total := 0
for _, src := range cfg.Sources {
if !src.Enabled {
continue
}
items, err := ingestion.FetchFeed(src.FeedURL, src.UserAgent)
if err != nil {
slog.Error("seed: feed fetch failed", "source", src.Name, "err", err)
continue
}
for _, item := range items {
if storage.IsGUIDSeen(item.GUID) {
continue
}
if err := storage.InsertStory(&storage.Story{
GUID: item.GUID,
Headline: item.Headline,
Lede: item.Lede,
ImageURL: item.ImageURL,
ArticleURL: item.ArticleURL,
Source: src.Name,
Classified: true,
SeenAt: timeNow(),
}); err != nil {
slog.Error("seed: insert failed", "guid", item.GUID, "err", err)
continue
}
total++
}
slog.Info("seed: source done", "source", src.Name, "items", len(items))
}
slog.Info("seed complete — all current stories marked as seen", "total", total)
}
func runTest(cfg *config.Config, sourceName string) {
// Find the first story from the matching (or first enabled) feed
var testItem *ingestion.FeedItem
var src config.SourceConfig
for _, s := range cfg.Sources {
if !s.Enabled {
continue
}
if sourceName != "" && s.Name != sourceName {
continue
}
items, err := ingestion.FetchFeed(s.FeedURL, s.UserAgent)
if err != nil {
slog.Error("test: feed fetch failed", "source", s.Name, "err", err)
continue
}
if len(items) > 0 {
item := items[0]
item.Source = s.Name
item.DirectRoute = s.DirectRoute
testItem = &item
src = s
break
}
}
if testItem == nil {
slog.Error("test: no stories found in any feed")
os.Exit(1)
}
slog.Info("test: selected story",
"source", src.Name,
"headline", testItem.Headline,
"guid", testItem.GUID,
)
channel := testItem.DirectRoute
if channel == "" {
slog.Error("test: source has no direct_route", "source", src.Name)
os.Exit(1)
}
// Validate image
imageURL := ""
if testItem.ImageURL != "" && ingestion.ValidateImageURL(testItem.ImageURL) {
imageURL = testItem.ImageURL
}
// Connect to Matrix and post
mx, err := matrix.New(cfg.Matrix)
if err != nil {
slog.Error("test: matrix init failed", "err", err)
os.Exit(1)
}
defer mx.Stop()
story := &matrix.PostableStory{
ImageURL: imageURL,
Headline: testItem.Headline,
ArticleURL: testItem.ArticleURL,
Lede: testItem.Lede,
Source: testItem.Source,
Channel: channel,
}
eventID, _, err := mx.PostStory(channel, story)
if err != nil {
slog.Error("test: post failed", "err", err)
os.Exit(1)
}
slog.Info("test: story posted successfully",
"channel", channel,
"event_id", eventID,
"headline", testItem.Headline,
)
}
func timeNow() int64 {
return time.Now().Unix()
}
// runBackfillPaywall re-runs the current paywall detection logic against
// every row marked paywalled=1. For each row we re-fetch the stored article
// URL; if the new logic no longer considers it gated we clear the flag and,
// when the row has no thumbnail, populate it from the live og:image.
func runBackfillPaywall() {
rows, err := storage.Get().Query(`SELECT guid, article_url, image_url FROM stories WHERE paywalled = 1`)
if err != nil {
slog.Error("backfill: query failed", "err", err)
os.Exit(1)
}
type rec struct{ guid, url, img string }
var pending []rec
for rows.Next() {
var r rec
if err := rows.Scan(&r.guid, &r.url, &r.img); err != nil {
slog.Error("backfill: scan failed", "err", err)
continue
}
pending = append(pending, r)
}
rows.Close()
slog.Info("backfill: starting", "candidates", len(pending))
cleared, imaged, stillGated := 0, 0, 0
for _, r := range pending {
meta := ingestion.FetchArticleMeta(r.url)
gated := meta.Gated() || (meta.Fetched && meta.BodyChars < ingestion.PaywallBodyThreshold)
if gated {
stillGated++
slog.Info("backfill: still gated", "guid", r.guid, "status", meta.Status, "body_chars", meta.BodyChars)
continue
}
newImg := r.img
if newImg == "" && meta.ImageURL != "" {
newImg = meta.ImageURL
imaged++
}
if _, err := storage.Get().Exec(`UPDATE stories SET paywalled = 0, image_url = ? WHERE guid = ?`, newImg, r.guid); err != nil {
slog.Error("backfill: update failed", "guid", r.guid, "err", err)
continue
}
cleared++
slog.Info("backfill: cleared paywall flag", "guid", r.guid, "filled_image", newImg != r.img)
}
slog.Info("backfill complete", "cleared", cleared, "still_gated", stillGated, "thumbnails_filled", imaged)
}