package main import ( "context" "flag" "log/slog" "os" "os/signal" "strings" "syscall" "time" "pete/internal/config" "pete/internal/ingestion" "pete/internal/matrix" "pete/internal/poster" "pete/internal/scheduler" "pete/internal/storage" "pete/internal/web" "maunium.net/go/mautrix/id" ) 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") flag.Parse() 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 } // 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) // 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) { if strings.TrimSpace(body) != "!post" { return } 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") roundRobinMode := cfg.Posting.RoundRobin.Enabled // 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, "[]") 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 roundRobinMode { channelNames := make([]string, 0, len(cfg.Matrix.Channels)) for name := range cfg.Matrix.Channels { 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) } } // Start the read-only web UI alongside the Matrix bot. if cfg.Web.Enabled { ws, err := web.New(cfg.Web) if err != nil { slog.Error("web server init failed", "err", err) } else { go ws.Start(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") } 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") ws, err := web.New(cfg.Web) if err != nil { slog.Error("local: web server init failed", "err", err) os.Exit(1) } go ws.Start(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) 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) 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() }