Pete/main.go

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")
	backfillPaywall := flag.Bool("backfill-paywall", false, "re-check paywalled rows with current detection logic; clear false positives and fill missing thumbnails, then exit")
	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
	}

	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 !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
		}
		if !admins[sender] {
			slog.Warn("!post ignored: sender not in matrix.admins allowlist", "channel", channel, "sender", sender)
			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 <code>"+channel+"</code>",
		); 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, cfg.Sources)
		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, cfg.Sources)
	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, 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)
}