handler_aggregate.go

package main

import (
	"context"
	"database/sql"
	"encoding/xml"
	"fmt"
	"html"
	"io"
	"log"
	"net/http"
	"os"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/josephus-git/gator/internal/database"
)

// RSSItem represents a single item within an RSS feed.
type RSSItem struct {
	Title       string `xml:"title"`
	Link        string `xml:"link"`
	Description string `xml:"description"`
	PubDate     string `xml:"pubDate"`
}

// RSSFeed represents the structure of an RSS 2.0 feed, containing channel information and items.
type RSSFeed struct {
	Channel struct {
		Title       string    `xml:"title"`
		Link        string    `xml:"link"`
		Description string    `xml:"description"`
		Item        []RSSItem `xml:"item"`
	} `xml:"channel"`
}

// fetchFeed fetches and parses an RSS feed from the given URL.
func fetchFeed(ctx context.Context, feedURL string) (*RSSFeed, error) {
	// Create a new HTTP request with context
	req, err := http.NewRequestWithContext(ctx, "GET", feedURL, nil)
	if err != nil {
		return nil, fmt.Errorf("failed to create request: %w", err)
	}

	// Set the User-Agent header
	req.Header.Set("User-Agent", "gator")

	// Create a HTTP client and perform the request
	client := &http.Client{
		Timeout: 10 * time.Second, // set a timeout for the request
	}
	resp, err := client.Do(req)
	if err != nil {
		return nil, fmt.Errorf("failed to perform request: %w", err)
	}
	defer resp.Body.Close()

	// Check for a successful HTTP status code
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("non-ok HTTP status: %d", resp.StatusCode)
	}

	// Read the response body
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, fmt.Errorf("failed to read respoonse body: %w", err)
	}

	// Unmarshal the XML into an RSSFeed struct
	var rssFeed RSSFeed
	err = xml.Unmarshal(body, &rssFeed)
	if err != nil {
		return nil, fmt.Errorf("failed to unmarshal XML: %w", err)
	}

	// Decode HTML entities for channel title and description
	rssFeed.Channel.Title = html.UnescapeString(rssFeed.Channel.Title)
	rssFeed.Channel.Description = html.UnescapeString(rssFeed.Channel.Description)

	// Decode HTML entities for item titles and descriptions
	for i := range rssFeed.Channel.Item {
		rssFeed.Channel.Item[i].Title = html.UnescapeString(rssFeed.Channel.Item[i].Title)
		rssFeed.Channel.Item[i].Description = html.UnescapeString(rssFeed.Channel.Item[i].Description)
	}
	return &rssFeed, nil
}

// scrapeFeeds fetches the next available RSS feed, marks it as fetched, and saves its posts to the database.
func scrapeFeeds(s *state) error {
	// fetch next feed id
	feed, err := s.db.GetNextFeedToFetch(context.Background())
	if err != nil {
		return fmt.Errorf("error fetching nextfeed: %v", err)
	}

	//mark feed as fetched
	markParams := database.MarkFeedFetchedParams{
		LastFetchedAt: sql.NullTime{Time: time.Now(), Valid: true},
		UpdatedAt:     time.Now(),
		ID:            feed.ID,
	}
	err = s.db.MarkFeedFetched(context.Background(), markParams)
	if err != nil {
		return fmt.Errorf("error marking fetched feed: %v", err)
	}

	// fetch feed
	fetchedFeed, err := fetchFeed(context.Background(), feed.Url)
	if err != nil {
		return fmt.Errorf("error in getting feed: %v", err)
	}
	for _, item := range fetchedFeed.Channel.Item {
		publishedAt := sql.NullTime{}
		if t, err := time.Parse(time.RFC1123Z, item.PubDate); err == nil {
			publishedAt = sql.NullTime{
				Time:  t,
				Valid: true,
			}
		}

		_, err = s.db.CreatePost(context.Background(), database.CreatePostParams{
			ID:        uuid.New(),
			CreatedAt: time.Now().UTC(),
			UpdatedAt: time.Now().UTC(),
			FeedID:    feed.ID,
			Title:     item.Title,
			Description: sql.NullString{
				String: item.Description,
				Valid:  true,
			},
			Url:         item.Link,
			PublishedAt: publishedAt,
		})
	}

	if err != nil {
		if strings.Contains(err.Error(), "duplicate key value violates unique constraint") {
		}
		log.Printf("Couldn't create post: %v", err)
	}

	log.Printf("Feed %s collected, %v posts found", feed.Name, len(fetchedFeed.Channel.Item))

	return nil
}

// aggregate continuously scrapes and aggregates new posts from feeds at a specified interval.
func aggregate(s *state, cmd command) error {
	if len(cmd.Handler) < 2 {
		fmt.Println("Usage: ./gator agg <time_in_seconds(numbersonly)>")
		os.Exit(1)
	}
	duration := fmt.Sprintf("%ss", cmd.Handler[1])
	time_between_reqs, err := time.ParseDuration(duration)
	if err != nil {
		return fmt.Errorf("error creating time between requests: %v", err)
	}

	fmt.Printf("Collecting feeds every %v\n", time_between_reqs)

	ticker := time.NewTicker(time_between_reqs)
	for ; ; <-ticker.C {
		scrapeFeeds(s)
		fmt.Println("feeds completely printed")
		fmt.Println("----------------")
		fmt.Println("")
	}
}