Pete/internal/ingestion/poller.go

package ingestion

import (
	"context"
	"log/slog"
	"strings"
	"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, src.UserAgent)
	if err != nil {
		// Persist the failure for the owner-facing source-health dashboard. The
		// in-memory consecutiveFailures in pollSource drives log warnings; this
		// row is the authoritative record the dashboard reads.
		storage.RecordPollResult(src.Name, false, 0, err)
		return err
	}
	// A successful fetch: record health now (before per-item processing) with
	// the number of items the feed returned, so the dashboard reflects feed
	// reachability even if individual items are later dropped as duplicates.
	storage.RecordPollResult(src.Name, true, len(items), nil)

	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
		}

		// An item older than the retention window is one Pete has either
		// already shown and since pruned, or never needs to: maintenance would
		// delete it on the next pass. Without this, a slow feed that still
		// lists month-old items re-ingests them as new the day after their row
		// is pruned, since that row was the only memory of them. Items with no
		// date can't be judged and pass through.
		if items[i].PublishedAt > 0 && time.Since(time.Unix(items[i].PublishedAt, 0)) > storage.StoryRetention {
			continue
		}

		// Drop items in languages this source isn't configured to surface.
		// Items without a language tag pass through — only filter when the
		// feed explicitly marks an item with a non-matching language.
		if src.Language != "" && items[i].Language != "" &&
			!strings.HasPrefix(items[i].Language, strings.ToLower(src.Language)) {
			slog.Info("dropping story (language filter)",
				"guid", items[i].GUID, "source", src.Name,
				"item_lang", items[i].Language, "want", src.Language)
			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 (with an internal
		// bypass-UA retry on gating signals). We use it for:
		//   1. og:image fallback when the feed didn't give us one.
		//   2. Paywall detection: explicit gating signals or a body too short
		//      to be a real article means we swap in an archive snapshot for
		//      both Pete (image) and the reader (posted link).
		//
		// Pure transport failures (FetchError) are NOT treated as gating —
		// swapping in a stale snapshot for a transient blip is worse than
		// posting the live link.
		meta := FetchArticleMeta(originalURL)
		// Some sources (e.g. LWN's subscriber-only articles) have no public
		// version reachable by any workaround. Drop these from view entirely
		// rather than stamping them as paywalled.
		if meta.SubscriberOnly {
			// No DB row written, so the next poll will re-fetch and re-detect
			// this URL — fine in practice since LWN's feed is tiny and the
			// alternative (a tombstone row) leaks into every story query.
			slog.Info("dropping subscriber-only story",
				"guid", items[i].GUID, "url", originalURL, "source", src.Name)
			continue
		}
		// bodyText is the article text we may store for reader mode. It starts
		// as the body scraped during the fetch above and is upgraded to the
		// snapshot's body if we end up swapping in an archive snapshot.
		bodyText := meta.BodyText
		gated := meta.Gated() || (meta.Fetched && meta.BodyChars < PaywallBodyThreshold)
		// paywalled tracks whether the link the user will click is still
		// gated — i.e. gating was detected AND no archive workaround worked.
		// When a snapshot swap succeeds, the reader gets a readable page, so
		// we don't stamp it.
		paywalled := false
		if gated {
			workedAround := false
			if snap := resolveArchive(originalURL); snap != "" {
				snapMeta := FetchArticleMeta(snap)
				if snapMeta.Fetched {
					items[i].ArticleURL = snap
					if items[i].ImageURL == "" && snapMeta.ImageURL != "" {
						items[i].ImageURL = snapMeta.ImageURL
					}
					if snapMeta.BodyText != "" {
						bodyText = snapMeta.BodyText
					}
					workedAround = true
					slog.Info("paywall detected, using archive snapshot",
						"guid", items[i].GUID, "original", originalURL,
						"snapshot", snap, "body_chars", meta.BodyChars,
						"status", meta.Status)
				} else {
					slog.Debug("paywall detected but snapshot fetch failed",
						"guid", items[i].GUID, "original", originalURL, "snapshot", snap)
				}
			} else {
				slog.Debug("paywall detected but no archive snapshot available",
					"guid", items[i].GUID, "url", originalURL,
					"body_chars", meta.BodyChars, "status", meta.Status)
			}
			paywalled = !workedAround
		} 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.
		// Future-dated pubDates (publishers occasionally post-date drafts or
		// have a misconfigured clock) get clamped to ingest time so they don't
		// stick at the top of the queue indefinitely.
		seenAt := time.Now().Unix()
		publishedAt := items[i].PublishedAt
		if publishedAt > seenAt {
			publishedAt = seenAt
		}
		// Reader-mode body: prefer whichever source gave us more complete text.
		// Feeds that ship full content:encoded usually win; feeds that only
		// syndicate a snippet fall back to the scraped article body.
		content := items[i].Content
		if len(bodyText) > len(content) {
			content = bodyText
		}
		if err := storage.InsertStory(&storage.Story{
			GUID:         items[i].GUID,
			Headline:     items[i].Headline,
			Lede:         items[i].Lede,
			Content:      content,
			ImageURL:     items[i].ImageURL,
			ArticleURL:   items[i].ArticleURL,
			URLCanonical: canonical,
			HeadlineNorm: headlineNorm,
			Source:       items[i].Source,
			Paywalled:    paywalled,
			SeenAt:       seenAt,
			PublishedAt:  publishedAt,
		}); 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
}

// resolveArchive tries Wayback first (well-behaved availability API, fresh
// snapshots filtered) and falls back to archive.today, which often has
// captures Wayback doesn't — especially for hard-walled publishers.
func resolveArchive(articleURL string) string {
	if snap := ResolveWayback(articleURL); snap != "" {
		return snap
	}
	return ResolveArchiveToday(articleURL)
}