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.yaml", "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)")
	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
	}

	// 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 <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)
		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 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()
}