Pete/main.go

package main

import (
	"context"
	"flag"
	"log/slog"
	"os"
	"os/signal"
	"syscall"
	"time"

	"pete/internal/classifier"
	"pete/internal/config"
	"pete/internal/explainer"
	"pete/internal/ingestion"
	"pete/internal/matrix"
	"pete/internal/poster"
	"pete/internal/scheduler"
	"pete/internal/storage"
)

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),
		"ollama_model", cfg.Ollama.Model,
	)

	// 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 Ollama client and classifier
	ollamaClient := classifier.NewOllamaClient(cfg.Ollama)
	cls := classifier.NewClassifier(ollamaClient, mx)

	// Create post queue
	queue := poster.NewQueue(cfg.Posting, mx)

	// Wire reaction handler + ❓ → summary hook
	mx.SetReactionHandler(poster.HandleReaction)
	exp := explainer.New(mx, ollamaClient)
	poster.SetReactionCallback(exp.Handle)

	// 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

	// Build the pipeline callback: classify → enqueue (or just classify+store
	// when round-robin mode is enabled; the scheduler does the enqueueing).
	processItem := func(ctx context.Context, item *ingestion.FeedItem) {
		result, err := cls.Classify(ctx, item)
		if err != nil {
			slog.Error("classification failed, will retry next cycle",
				"guid", item.GUID,
				"err", err,
			)
			return // story stays unclassified for retry
		}

		// Add to recent headlines for future dedup
		storage.InsertRecentHeadline(item.GUID, item.Headline, item.Source)

		if !result.ShouldPost {
			// Mark classified with sentinel channel so it's not retried
			storage.MarkClassified(item.GUID, "_discarded", "[]")
			slog.Info("story gated out",
				"guid", item.GUID,
				"rationale", result.Rationale,
			)
			return
		}

		// Mark classified with actual channel
		platforms := storage.MarshalPlatforms(result.Platforms)
		storage.MarkClassified(item.GUID, result.Channel, platforms)

		if result.DuplicateOf != nil {
			// Overwrite with sentinel so the round-robin scheduler doesn't pick
			// this up later. Immediate-mode posting already returned above.
			storage.MarkClassified(item.GUID, "_duplicate", "[]")
			slog.Info("story deduplicated",
				"guid", item.GUID,
				"duplicate_of", *result.DuplicateOf,
			)
			return
		}

		if roundRobinMode {
			// Story stays in DB classified+postable; the scheduler will pick
			// it up on the next rotation tick.
			slog.Info("story classified, awaiting round-robin tick",
				"guid", item.GUID, "source", item.Source, "channel", result.Channel)
			return
		}

		// Validate image
		imageURL := ""
		if item.ImageURL != "" && ingestion.ValidateImageURL(item.ImageURL) {
			imageURL = item.ImageURL
		}

		// Enqueue for posting
		queue.Enqueue(poster.QueueItem{
			GUID:       item.GUID,
			Headline:   item.Headline,
			Lede:       item.Lede,
			ImageURL:   imageURL,
			ArticleURL: item.ArticleURL,
			Source:      item.Source,
			Channel:    result.Channel,
			Platforms:  result.Platforms,
		})
	}

	// Start pollers
	poller := ingestion.NewPoller(cfg.Sources, processItem)
	poller.Start(ctx)
	slog.Info("pollers started")

	if roundRobinMode {
		rr := scheduler.New(cfg.Sources, cfg.Posting.RoundRobin.IntervalHours, queue)
		if rr == nil {
			slog.Warn("round-robin enabled but no sources are enabled; nothing will post")
		} else {
			go rr.Start(ctx)
		}
	}

	// Run maintenance on startup
	storage.RunMaintenance(cfg.Storage.RecentWindowHours, cfg.Storage.ClassificationLogDays)

	// 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,
				FeedHint:   src.FeedHint,
				Classified: true,
				SeenAt:     timeNow(),
			}); err != nil {
				slog.Error("seed: insert failed", "guid", item.GUID, "err", err)
				continue
			}
			storage.InsertRecentHeadline(item.GUID, item.Headline, src.Name)
			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.FeedHint = s.FeedHint
			item.Tier = s.Tier
			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,
	)

	// Determine channel
	channel := "politics" // default
	if testItem.DirectRoute != nil {
		channel = *testItem.DirectRoute
	}

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