Pete/internal/ingestion/poller.go

package ingestion

import (
	"context"
	"log/slog"
	"sync"
	"time"

	"pete/internal/config"
	"pete/internal/dedup"
	"pete/internal/storage"
)

// ProcessFunc is called for each new feed item that needs classification.
type ProcessFunc func(item *FeedItem)

// Poller manages per-source RSS polling goroutines.
type Poller struct {
	sources []config.SourceConfig
	process ProcessFunc
	wg      sync.WaitGroup
}

// NewPoller creates a poller for the given sources.
func NewPoller(sources []config.SourceConfig, process ProcessFunc) *Poller {
	return &Poller{
		sources: sources,
		process: process,
	}
}

// Start launches a goroutine per enabled source. Blocks until ctx is cancelled.
func (p *Poller) Start(ctx context.Context) {
	for _, src := range p.sources {
		if !src.Enabled {
			slog.Info("source disabled, skipping", "source", src.Name)
			continue
		}
		p.wg.Add(1)
		go p.pollSource(ctx, src)
	}
}

// Wait blocks until all polling goroutines have exited.
func (p *Poller) Wait() {
	p.wg.Wait()
}

func (p *Poller) pollSource(ctx context.Context, src config.SourceConfig) {
	defer p.wg.Done()

	interval := time.Duration(src.PollIntervalMinutes) * time.Minute
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	slog.Info("starting poller", "source", src.Name, "interval", interval)

	// Poll immediately on start, then on ticker
	p.pollOnce(src)

	consecutiveFailures := 0
	const adminWarningThreshold = 5

	for {
		select {
		case <-ctx.Done():
			slog.Info("poller stopping", "source", src.Name)
			return
		case <-ticker.C:
			if err := p.pollOnceWithErr(src); err != nil {
				consecutiveFailures++
				slog.Error("feed poll failed",
					"source", src.Name,
					"err", err,
					"consecutive_failures", consecutiveFailures,
				)
				if consecutiveFailures == adminWarningThreshold {
					slog.Warn("persistent feed failure",
						"source", src.Name,
						"failures", consecutiveFailures,
					)
					// TODO: send admin room warning via matrix client
				}
			} else {
				consecutiveFailures = 0
			}
		}
	}
}

func (p *Poller) pollOnce(src config.SourceConfig) {
	if err := p.pollOnceWithErr(src); err != nil {
		slog.Error("feed poll failed", "source", src.Name, "err", err)
	}
}

func (p *Poller) pollOnceWithErr(src config.SourceConfig) error {
	items, err := FetchFeed(src.FeedURL)
	if err != nil {
		return err
	}

	newCount := 0
	for i := range items {
		if storage.IsGUIDSeen(items[i].GUID) {
			continue
		}

		// Last-resort image fallback: feed gave us nothing, scrape og:image
		// from the article page. Only runs for genuinely new items.
		if items[i].ImageURL == "" {
			if og := FetchOGImage(items[i].ArticleURL); og != "" {
				items[i].ImageURL = og
				slog.Debug("og:image fallback used",
					"guid", items[i].GUID, "url", items[i].ArticleURL, "image", og)
			}
		}

		canonical := dedup.CanonicalURL(items[i].ArticleURL)
		headlineNorm := dedup.NormalizeHeadline(items[i].Headline)

		if storage.IsCanonicalSeen(canonical) {
			slog.Info("dropping duplicate story (canonical URL match)",
				"guid", items[i].GUID, "url_canonical", canonical, "source", src.Name)
			continue
		}
		if storage.IsHeadlineSeen(src.Name, headlineNorm) {
			slog.Info("dropping duplicate story (headline match)",
				"guid", items[i].GUID, "headline_norm", headlineNorm, "source", src.Name)
			continue
		}

		// Stamp source metadata onto the item
		items[i].Source = src.Name
		items[i].FeedHint = src.FeedHint
		items[i].Tier = src.Tier
		items[i].DirectRoute = src.DirectRoute

		// Store immediately (before classification) to prevent re-ingestion
		if err := storage.InsertStory(&storage.Story{
			GUID:         items[i].GUID,
			Headline:     items[i].Headline,
			Lede:         items[i].Lede,
			ImageURL:     items[i].ImageURL,
			ArticleURL:   items[i].ArticleURL,
			URLCanonical: canonical,
			HeadlineNorm: headlineNorm,
			Source:       items[i].Source,
			FeedHint:     items[i].FeedHint,
			SeenAt:       time.Now().Unix(),
		}); err != nil {
			slog.Error("failed to insert story", "guid", items[i].GUID, "err", err)
			continue
		}

		p.process(&items[i])
		newCount++
	}

	if newCount > 0 {
		slog.Info("ingested new stories", "source", src.Name, "count", newCount)
	}

	// Retry unclassified stories from this source only (cap at 20 to avoid runaway retries)
	unclassified, err := storage.GetUnclassifiedStories(src.Name)
	if err != nil {
		slog.Error("failed to get unclassified stories", "err", err)
		return nil
	}
	for _, s := range unclassified {
		// Skip stories we just ingested (they're already being processed above)
		alreadyProcessed := false
		for _, item := range items {
			if item.GUID == s.GUID {
				alreadyProcessed = true
				break
			}
		}
		if alreadyProcessed {
			continue
		}

		p.process(&FeedItem{
			GUID:        s.GUID,
			Headline:    s.Headline,
			Lede:        s.Lede,
			ImageURL:    s.ImageURL,
			ArticleURL:  s.ArticleURL,
			Source:       s.Source,
			FeedHint:    s.FeedHint,
			Tier:        src.Tier,
			DirectRoute: src.DirectRoute,
		})
	}

	return nil
}