fetch.go

package xrss

import (
	"context"
	"crypto/hmac"
	"crypto/rand"
	"crypto/sha512"
	"database/sql"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/url"
	"slices"
	"sync"
	"time"

	"github.com/mmcdole/gofeed"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

type hasher func([]byte) string

func pepperHasher(db *sql.DB, name string) func(data []byte) string {
	key := make([]byte, 32)
	n := try(rand.Read(key))("read crypto rand")
	if n < 32 {
		panic(fmt.Sprintf("failed to read enough randomness: n = %d", n))
	}
	row := db.QueryRow(`INSERT INTO keys (name,key) VALUES (?,?) ON CONFLICT DO UPDATE SET name=name RETURNING key;`, name, key)
	err := row.Scan(&key)
	if err != nil {
		panic(fmt.Sprintf("failed to read pepper key for '%s' from db: %v", name, err))
	}
	hasher := hmac.New(sha512.New384, key)
	return func(data []byte) string {
		return base64.URLEncoding.EncodeToString(hasher.Sum(data))
	}
}

func feedUrlHasher(db *sql.DB) hasher {
	return sync.OnceValue(func() hasher { return pepperHasher(db, "feed_url") })()
}

func fetchWorker(wdb, rdb *sql.DB, kv jetstream.KeyValue, conn *nats.Conn, log *slog.Logger) error {
	ch := make(chan *nats.Msg, 20)
	sub, err := conn.ChanQueueSubscribe("fetch.*", "fetch", ch)
	if err != nil {
		return fmt.Errorf("failed to subscribe to channel: %w", err)
	}
	defer sub.Unsubscribe()
	hasher := feedUrlHasher(wdb)
	for msg := range ch {
		err := fetch(msg, wdb, rdb, kv, hasher)
		if err != nil {
			log.Error("failed to fetch feed", slog.Any("error", err))
		}
	}
	return nil
}

func fetch(msg *nats.Msg, wdb, rdb *sql.DB, kv jetstream.KeyValue, hasher hasher) error {
	// basic url validation
	var url_, hash string
	{
		parsedUrl, err := url.Parse(string(msg.Data))
		if err != nil {
			return fmt.Errorf("failed to parse url: %w", err)
		}
		if parsedUrl.Host == "" {
			return fmt.Errorf("url has no host: %s", parsedUrl)

		}
		if !slices.Contains([]string{"http", "https"}, parsedUrl.Scheme) {
			return fmt.Errorf("url has invalid scheme: %s url: %s", parsedUrl.Scheme, parsedUrl)
		}
		url_ = parsedUrl.String()
		hash = hasher([]byte(url_))
	}

	// create or get current feed_id
	var feed_id int64
	{
		err := rdb.QueryRow(`SELECT id FROM feed WHERE hash=?;`, hash).Scan(&feed_id)
		if err == sql.ErrNoRows {
			err = wdb.QueryRow(`INSERT INTO feed(url,hash) VALUES (?, ?) ON CONFLICT DO UPDATE SET url=url RETURNING id;`, url_, hash).Scan(&feed_id)
		}
		if err != nil {
			return fmt.Errorf("failed get or create feed id: %w", err)
		}
	}

	// get previous fetch metadata
	var lastModified, etag string
	{
		err := rdb.QueryRow(`SELECT last_modified,etag FROM v_refetch WHERE id = ?`, feed_id).Scan(&lastModified, &etag)
		if err != nil && err != sql.ErrNoRows {
			return fmt.Errorf("failed to query fetch info for feed (feed_id:%d) (url:%s): %w", feed_id, url_, err)
		}
	}

	// fetch feed, store fetch result to db
	var fetch_id, total_items int64
	{
		// TODO(infogulch) choose "external" network adapter to help avoid
		// queries on internal network:
		// https://gist.github.com/2minchul/191716b3ca8799f53362746731d08e91
		result, err := FetchFeed(url_, etag, lastModified)
		if err != nil {
			result.FetchError = err.Error()
		}
		tx, err := wdb.BeginTx(context.Background(), nil)
		if err != nil {
			return fmt.Errorf("failed to begin transaction: %w", err)
		}
		err = tx.QueryRow(insertFetch, result.fields()...).Scan(&fetch_id)
		if err != nil {
			return fmt.Errorf("failed to insert fetch data: %w", err)
		}
		items, _ := json.Marshal(result.Items)
		sqlResult, err := tx.Exec(insertItems, feed_id, fetch_id, string(items))
		if err != nil {
			return fmt.Errorf("failed to insert item data: %w", err)
		}
		added_count, _ := sqlResult.RowsAffected()
		_, err = tx.Exec(`UPDATE fetch SET added_count = ? WHERE id = ?;`, added_count, fetch_id)
		if err != nil {
			return fmt.Errorf("failed to update fetch added: %w", err)
		}
		err = tx.QueryRow(`UPDATE feed SET total_items = total_items+? WHERE id = ? RETURNING total_items;`, added_count, feed_id).Scan(&total_items)
		if err != nil {
			return fmt.Errorf("failed to update feed total: %w", err)
		}
		err = tx.Commit()
		if err != nil {
			return fmt.Errorf("failed to commit tx: %w", err)
		}
	}

	// notify users
	{
		subs, err := rdb.Query(`SELECT user_id FROM subscription WHERE feed_id=?`, feed_id)
		if err != nil {
			return fmt.Errorf("failed to query feed subscribers: %w", err)
		}
		total_items_s := fmt.Sprintf("%d", total_items)
		for subs.Next() {
			var user_id int64
			err = subs.Scan(&user_id)
			if err != nil {
				return fmt.Errorf("failed to scan user: %w", err)
			}
			key := fmt.Sprintf("user.%d.feed.%d.total_items", user_id, feed_id)
			_, err = kv.PutString(context.Background(), key, total_items_s)
			if err != nil {
				return fmt.Errorf("failed to put key %s=%s: %w", key, total_items_s, err)
			}
		}
	}
	return nil
}

const (
	insertFetch = `INSERT INTO fetch (
feed_id, fetch_url, fetch_status, fetch_etag, fetch_lastmodified, fetch_error,
fetch_on, fetch_duration, fetch_bytes, fetch_count, added_count, data
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, JSONB(?)) RETURNING id;`

	insertItems = `INSERT INTO item (feed_id, fetch_id, data)
SELECT ? AS feed_id, ? AS fetch_id, JSONB(JSON_EACH.value) AS data
FROM JSON_EACH(?)
WHERE JSON_EACH.value ->> '$.guid' NOT IN (
	SELECT guid FROM item WHERE feed_id = NEW.feed_id
);`
)

type fetchResult struct {
	FeedId  int64
	FetchId int64

	FetchURL          string
	FetchStatus       int
	FetchEtag         string
	FetchLastModified string
	FetchError        string
	FetchOn           time.Time
	FetchDuration     time.Duration
	FetchBytes        int
	FetchCount        int

	Feed  *gofeed.Feed
	Items []*gofeed.Item
}

func (f *fetchResult) fields() []any {
	data, _ := json.Marshal(f.Feed)
	datastr := string(data)
	added_count := 0
	fs := [...]any{&f.FetchURL, &f.FetchStatus, &f.FetchEtag, &f.FetchLastModified, &f.FetchOn, &f.FetchDuration, &f.FetchBytes, &f.FetchCount, &added_count, &datastr}
	return fs[:]
}

var gmtTimeZoneLocation *time.Location = try(time.LoadLocation("GMT"))("load gmt time")

func FetchFeed(url string, etag string, lastModified string) (fetchResult, error) {
	start := time.Now().In(gmtTimeZoneLocation)

	req, err := http.NewRequest(http.MethodGet, url, nil)
	if err != nil {
		return fetchResult{}, err
	}
	req.Header.Set("User-Agent", "xrss/1.0")

	if etag != "" {
		req.Header.Set("If-None-Match", fmt.Sprintf(`"%s"`, etag))
	}

	if lastModified != "" {
		req.Header.Set("If-Modified-Since", lastModified)
	}

	resp, err := http.DefaultClient.Do(req)

	result := fetchResult{
		FetchStatus:   resp.StatusCode,
		FetchURL:      url,
		FetchOn:       start,
		FetchDuration: time.Since(start),
	}

	if err != nil {
		return result, err
	}

	if resp != nil {
		defer func() {
			ce := resp.Body.Close()
			if ce != nil {
				err = ce
			}
		}()
	}

	if resp.StatusCode == http.StatusNotModified {
		return result, nil
	}

	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return result, fmt.Errorf("feed fetch failed: status: %d, headers: %v", resp.StatusCode, resp.Header)
	}

	parser := gofeed.NewParser()
	// may want to use .AuthConfig or .Client

	counter := &CountingReader{reader: resp.Body}

	result.Feed, err = parser.Parse(counter)
	result.FetchBytes = counter.BytesRead
	result.FetchEtag = resp.Header.Get("Etag")
	result.FetchLastModified = resp.Header.Get("Last-Modified")
	if result.Feed != nil {
		result.Items = result.Feed.Items
		result.FetchCount = len(result.Items)
		result.Feed.Items = nil
	}
	return result, err
}

type CountingReader struct {
	reader    io.Reader
	BytesRead int
}

func (r *CountingReader) Read(p []byte) (n int, err error) {
	n, err = r.reader.Read(p)
	r.BytesRead += n
	return n, err
}

func try[T any](t T, err error) func(string) T {
	return func(desc string) T {
		try0(err, desc)
		return t
	}
}

func try0(err error, desc string) {
	if err != nil {
		panic(fmt.Errorf("failed to %s: %w", desc, err))
	}
}