rsspipe/internal/crawler/crawler.go

package crawler

import (
	"context"
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/haopengzhan/rsspipe/internal/db"
	"github.com/haopengzhan/rsspipe/internal/models"
	"github.com/haopengzhan/rsspipe/internal/parser"
)

type Crawler struct {
	db              *db.DB
	parser          *parser.Parser
	intervalMinutes int
	workerCount     int
	isRefreshing    bool
	mu              sync.Mutex
}

func NewCrawler(database *db.DB, intervalMinutes, workers int) *Crawler {
	return &Crawler{
		db:              database,
		parser:          parser.NewParser(),
		intervalMinutes: intervalMinutes,
		workerCount:     workers,
	}
}

func (c *Crawler) Start(ctx context.Context) {
	log.Printf("==> Starting RSS background crawler scheduler (Workers: %d)\n", c.workerCount)

	// Run initial fetch asynchronously after 2 seconds startup
	go func() {
		time.Sleep(2 * time.Second)
		c.RefreshAll(context.Background())
	}()

	// Check due feeds every 45 seconds
	ticker := time.NewTicker(45 * time.Second)
	go func() {
		for {
			select {
			case <-ctx.Done():
				ticker.Stop()
				return
			case <-ticker.C:
				c.RefreshDue(context.Background())
			}
		}
	}()
}

// RefreshDue checks all feeds and refreshes only those whose refresh interval has elapsed
func (c *Crawler) RefreshDue(ctx context.Context) {
	c.mu.Lock()
	if c.isRefreshing {
		c.mu.Unlock()
		return
	}
	c.isRefreshing = true
	c.mu.Unlock()

	defer func() {
		c.mu.Lock()
		c.isRefreshing = false
		c.mu.Unlock()
	}()

	feeds, err := c.db.ListFeeds()
	if err != nil {
		log.Printf("Crawler error listing feeds: %v\n", err)
		return
	}

	now := time.Now().UTC()
	var dueFeeds []models.Feed

	for _, f := range feeds {
		interval := f.RefreshIntervalMinutes
		if interval <= 0 {
			interval = 30
		}
		if f.LastFetchedAt == nil || now.Sub(*f.LastFetchedAt) >= time.Duration(interval)*time.Minute {
			dueFeeds = append(dueFeeds, f)
		}
	}

	if len(dueFeeds) == 0 {
		return
	}

	log.Printf("Crawler: Refreshing %d due feeds (out of %d total)...\n", len(dueFeeds), len(feeds))
	c.runWorkerPool(ctx, dueFeeds)
}

// RefreshAll refreshes all feeds unconditionally (e.g. on manual trigger)
func (c *Crawler) RefreshAll(ctx context.Context) {
	c.mu.Lock()
	if c.isRefreshing {
		c.mu.Unlock()
		log.Println("Crawler: RefreshAll already in progress, skipping")
		return
	}
	c.isRefreshing = true
	c.mu.Unlock()

	defer func() {
		c.mu.Lock()
		c.isRefreshing = false
		c.mu.Unlock()
	}()

	feeds, err := c.db.ListFeeds()
	if err != nil {
		log.Printf("Crawler error listing feeds: %v\n", err)
		return
	}

	if len(feeds) == 0 {
		return
	}

	log.Printf("Crawler: Refreshing all %d feeds...\n", len(feeds))
	c.runWorkerPool(ctx, feeds)
}

func (c *Crawler) runWorkerPool(ctx context.Context, feeds []models.Feed) {
	jobs := make(chan models.Feed, len(feeds))
	var wg sync.WaitGroup

	for i := 0; i < c.workerCount; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for f := range jobs {
				c.fetchSingleFeed(ctx, f)
			}
		}()
	}

	for _, f := range feeds {
		jobs <- f
	}
	close(jobs)

	wg.Wait()
	log.Printf("Crawler: Completed refresh of %d feeds.\n", len(feeds))
}

func (c *Crawler) RefreshSingle(ctx context.Context, feedID string) error {
	f, err := c.db.GetFeed(feedID)
	if err != nil {
		return fmt.Errorf("feed not found: %w", err)
	}
	return c.fetchSingleFeed(ctx, *f)
}

func (c *Crawler) fetchSingleFeed(ctx context.Context, f models.Feed) error {
	timeout := time.Duration(f.TimeoutSeconds) * time.Second
	if timeout <= 0 {
		timeout = 10 * time.Second
	}
	feedCtx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()

	res, err := c.parser.FetchFeed(feedCtx, f.FeedURL, f.ETag, f.LastModified)
	if err != nil {
		log.Printf("Crawler error fetching [%s]: %v\n", f.Title, err)
		_ = c.db.UpdateFeedFetchStatus(f.ID, f.ETag, f.LastModified, err.Error())
		return err
	}

	if res.NotModified {
		_ = c.db.UpdateFeedFetchStatus(f.ID, f.ETag, f.LastModified, "")
		return nil
	}

	// Upsert entries
	inserted, err := c.db.UpsertFeedEntries(f.ID, res.Entries)
	if err != nil {
		log.Printf("Crawler error upserting entries for [%s]: %v\n", f.Title, err)
	} else if inserted > 0 {
		log.Printf("Crawler [%s]: inserted %d new entries\n", f.Title, inserted)
	}

	_ = c.db.UpdateFeedFetchStatus(f.ID, res.ETag, res.LastModified, "")
	return nil
}