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) ws.StartTableClock(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) }