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 routing.
type ProcessFunc func(ctx context.Context, 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(ctx, 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(ctx, 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,
					)
				}
			} else {
				consecutiveFailures = 0
			}
		}
	}
}

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

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

	newCount := 0
	for i := range items {
		if ctx.Err() != nil {
			slog.Info("poller: cancellation observed mid-loop", "source", src.Name, "processed", newCount)
			return nil
		}
		if storage.IsGUIDSeen(items[i].GUID) {
			continue
		}

		// Dedup checks first — canonical and headline use the ORIGINAL URL so
		// they stay stable whether we end up swapping to a Wayback snapshot.
		originalURL := items[i].ArticleURL
		canonical := dedup.CanonicalURL(originalURL)
		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
		}

		// Genuinely new story — fetch the article page once. We use it for:
		//   1. og:image fallback when the feed didn't give us one.
		//   2. Paywall detection: visible body text below the threshold means
		//      the article is gated, so swap in a Wayback snapshot for both
		//      Pete (image) and the reader (posted link).
		meta := FetchArticleMeta(originalURL)
		paywalled := !meta.Fetched || meta.Paywalled || meta.BodyChars < PaywallBodyThreshold
		if paywalled {
			if snap := ResolveWayback(originalURL); snap != "" {
				snapMeta := FetchArticleMeta(snap)
				if snapMeta.Fetched {
					items[i].ArticleURL = snap
					if items[i].ImageURL == "" && snapMeta.ImageURL != "" {
						items[i].ImageURL = snapMeta.ImageURL
					}
					slog.Info("paywall detected, using wayback snapshot",
						"guid", items[i].GUID, "original", originalURL,
						"snapshot", snap, "body_chars", meta.BodyChars)
				} else {
					slog.Debug("paywall detected but snapshot fetch failed",
						"guid", items[i].GUID, "original", originalURL, "snapshot", snap)
				}
			} else {
				slog.Debug("paywall detected but no wayback snapshot available",
					"guid", items[i].GUID, "url", originalURL, "body_chars", meta.BodyChars)
			}
		} else if items[i].ImageURL == "" && meta.ImageURL != "" {
			items[i].ImageURL = meta.ImageURL
			slog.Debug("og:image fallback used",
				"guid", items[i].GUID, "url", originalURL, "image", meta.ImageURL)
		}

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

		// Store immediately (before routing) 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,
			SeenAt:       time.Now().Unix(),
		}); err != nil {
			slog.Error("failed to insert story", "guid", items[i].GUID, "err", err)
			continue
		}

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

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

	return nil
}