rsspipe/internal/crawler/crawler.go

package crawler

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

	"git.pengzhan.dev/haopengzhan/rsspipe/internal/db"
	"git.pengzhan.dev/haopengzhan/rsspipe/internal/events"
	"git.pengzhan.dev/haopengzhan/rsspipe/internal/logger"
	"git.pengzhan.dev/haopengzhan/rsspipe/internal/models"
	"git.pengzhan.dev/haopengzhan/rsspipe/internal/parser"
	"go.uber.org/zap"
)

type Crawler struct {
	db              *db.DB
	parser          *parser.Parser
	bus             *events.EventBus
	intervalMinutes int
	workerCount     int
	isRefreshing    bool
	mu              sync.Mutex
	log             *zap.Logger
}

func NewCrawler(database *db.DB, bus *events.EventBus, intervalMinutes, workers int) *Crawler {
	return &Crawler{
		db:              database,
		parser:          parser.NewParser(),
		bus:             bus,
		intervalMinutes: intervalMinutes,
		workerCount:     workers,
		log:             logger.Named("crawler"),
	}
}

func (c *Crawler) Start(ctx context.Context) {
	c.log.Info("Starting RSS background crawler scheduler", zap.Int("worker_count", 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 {
		c.log.Error("Crawler error listing feeds", zap.Error(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
	}

	c.log.Info("Crawler refreshing due feeds", zap.Int("due_count", len(dueFeeds)), zap.Int("total_count", 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()
		c.log.Debug("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 {
		c.log.Error("Crawler error listing feeds", zap.Error(err))
		return
	}

	if len(feeds) == 0 {
		return
	}

	c.log.Info("Crawler refreshing all feeds", zap.Int("count", 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()
	c.log.Info("Crawler completed refresh batch", zap.Int("feed_count", 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 {
		c.log.Warn("Crawler error fetching feed",
			zap.String("feed_title", f.Title),
			zap.String("feed_url", f.FeedURL),
			zap.Error(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
	newEntries, err := c.db.UpsertFeedEntries(f.ID, res.Entries)
	if err != nil {
		c.log.Error("Crawler error upserting entries",
			zap.String("feed_title", f.Title),
			zap.Error(err),
		)
	} else if len(newEntries) > 0 {
		c.log.Info("Crawler inserted new entries",
			zap.String("feed_title", f.Title),
			zap.Int("new_entries", len(newEntries)),
		)

		// Publish on_post_created and on_entry_transition events for newly inserted entries
		if c.bus != nil {
			for _, entry := range newEntries {
				entry.FeedTitle = f.Title
				entry.FeedSiteURL = f.SiteURL
				diff := &models.EntryDiff{
					FieldsChanged: []string{"id", "title", "content", "url"},
				}
				c.bus.Publish(ctx, events.Event{
					Type:     events.EventPostCreated,
					Entry:    entry,
					NewEntry: &entry,
					Diff:     diff,
					Depth:    0,
				})
				c.bus.Publish(ctx, events.Event{
					Type:     events.EventEntryTransition,
					Entry:    entry,
					NewEntry: &entry,
					Diff:     diff,
					Depth:    0,
				})
			}
		}
	}

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