Maven/internal/rss/poller.go

package rss

import (
	"context"
	"fmt"
	"log"
	"strings"
	"time"
)

// FeedConfig — one feed to read. A feed with no Name or no URL is ignored.
type FeedConfig struct {
	Name     string        // short id; the note source is "rss:<Name>"
	URL      string        // http(s) only, enforced by the fetcher
	Category string        // free text ("технологии"), used to answer "что по X?"
	Interval time.Duration // 0 ⇒ the poller's default
	Include  []string      // when non-empty, keep only items matching one of these
	Exclude  []string      // drop items matching any of these, even if included
}

// Fetcher is the guarded HTTP door (internal/webfetch). An interface so the
// poller is testable without a network and so it CANNOT fetch by any other
// means: no http.Client is constructed in this package.
type Fetcher interface {
	Get(ctx context.Context, url string) (*Body, error)
}

// Body is the minimum the poller needs from a response.
type Body struct{ Bytes []byte }

// Notes is core's note-writing half. Same shape as ipc.CoreAPI's method, so the
// daemon passes its API straight in.
type Notes interface {
	WriteNote(ctx context.Context, ts time.Time, text string, embedding []float32, source string) (int64, error)
}

// Marks remembers how far a feed was read. Durable, because the alternative is
// re-writing yesterday's headlines as fresh notes after every restart. The
// daemon backs this with config facts (key "rss:latest:<feed>").
type Marks interface {
	LastMark(ctx context.Context, feed string) (time.Time, error)
	SetMark(ctx context.Context, feed string, at time.Time) error
}

// Embedder embeds a note on its way into the store so recall can find it. nil ⇒
// notes are written without a vector (still readable by the recent-notes path).
type Embedder interface {
	Embed(ctx context.Context, text string) ([]float32, error)
}

// Ranker is the relevance seam. The plan called for scoring each item against
// an interest profile built from his notes; that profile does not exist yet, and
// a threshold over an embedder with no profile to compare to is a random filter
// with a confident name. So the seam is here, nil in the daemon, and the filter
// that actually runs is the per-feed keyword one — a rule he can read and
// predict. When there IS a profile, implement this and pass it.
//
// Note what a Ranker must NOT be: anything that sends his notes outward. The
// scoring happens locally against a local embedder; the feed item is the input,
// his memory is never the payload.
type Ranker interface {
	Relevant(ctx context.Context, text string) (bool, error)
}

// Config — poller-wide settings.
type Config struct {
	DefaultInterval time.Duration // 0 ⇒ DefaultPollInterval
	MaxItems        int           // most notes written per feed per poll; 0 ⇒ DefaultMaxItems
	MaxAge          time.Duration // ignore items older than this on a cold start; 0 ⇒ DefaultMaxAge
}

// Defaults chosen to be quiet: a feed read every half hour, at most a handful of
// items kept, and a cold start that does not import a month of history.
const (
	DefaultPollInterval = 30 * time.Minute
	DefaultMaxItems     = 5
	DefaultMaxAge       = 24 * time.Hour
)

// Poller reads feeds on a schedule and writes what survives filtering as notes.
type Poller struct {
	feeds   []FeedConfig
	fetch   Fetcher
	notes   Notes
	marks   Marks
	embed   Embedder
	ranker  Ranker
	cfg     Config
	nextDue map[string]time.Time
	seen    map[string]map[string]bool // feed → item ID, for items with no date
}

// NewPoller wires a poller. Returns nil when there is nothing to poll — a
// capability is off unless configured, and callers check for nil.
func NewPoller(feeds []FeedConfig, fetch Fetcher, notes Notes, marks Marks, embed Embedder, ranker Ranker, cfg Config) *Poller {
	var valid []FeedConfig
	for _, f := range feeds {
		if strings.TrimSpace(f.Name) == "" || strings.TrimSpace(f.URL) == "" {
			log.Printf("rss: skipping a feed with no name or no url")
			continue
		}
		valid = append(valid, f)
	}
	if len(valid) == 0 || fetch == nil || notes == nil {
		return nil
	}
	if cfg.DefaultInterval <= 0 {
		cfg.DefaultInterval = DefaultPollInterval
	}
	if cfg.MaxItems <= 0 {
		cfg.MaxItems = DefaultMaxItems
	}
	if cfg.MaxAge <= 0 {
		cfg.MaxAge = DefaultMaxAge
	}
	return &Poller{
		feeds: valid, fetch: fetch, notes: notes, marks: marks,
		embed: embed, ranker: ranker, cfg: cfg,
		nextDue: map[string]time.Time{},
		seen:    map[string]map[string]bool{},
	}
}

// Feeds returns the configured feeds (the answer path lists categories).
func (p *Poller) Feeds() []FeedConfig { return p.feeds }

// PollDue reads every feed whose interval has elapsed and returns how many
// notes were written. Errors are logged per feed, never returned: one dead feed
// must not stop the others, and there is nobody waiting on this.
func (p *Poller) PollDue(ctx context.Context, now time.Time) int {
	written := 0
	for _, f := range p.feeds {
		if due, ok := p.nextDue[f.Name]; ok && now.Before(due) {
			continue
		}
		interval := f.Interval
		if interval <= 0 {
			interval = p.cfg.DefaultInterval
		}
		p.nextDue[f.Name] = now.Add(interval)
		n, err := p.PollFeed(ctx, f, now)
		if err != nil {
			// The URL is configured by him and not a secret, so it is loggable;
			// item titles are not logged, only counts.
			log.Printf("rss: feed %s: %v", f.Name, err)
			continue
		}
		if n > 0 {
			log.Printf("rss: feed %s: %d new item(s) noted", f.Name, n)
		}
		written += n
	}
	return written
}

// PollFeed reads one feed now, regardless of its schedule.
func (p *Poller) PollFeed(ctx context.Context, f FeedConfig, now time.Time) (int, error) {
	body, err := p.fetch.Get(ctx, f.URL)
	if err != nil {
		return 0, err
	}
	feed, err := Parse(strings.NewReader(string(body.Bytes)))
	if err != nil {
		return 0, err
	}

	mark := p.mark(ctx, f.Name, now)
	newest := mark
	written := 0
	for _, it := range feed.Items {
		if written >= p.cfg.MaxItems {
			break
		}
		if !p.fresh(f, it, mark, now) {
			continue
		}
		if !Matches(f, it) {
			continue
		}
		if p.ranker != nil {
			ok, err := p.ranker.Relevant(ctx, it.Title+" "+it.Summary)
			if err != nil {
				log.Printf("rss: feed %s: relevance: %v", f.Name, err)
			} else if !ok {
				continue
			}
		}
		if err := p.write(ctx, f, it, now); err != nil {
			return written, err
		}
		written++
		if it.Published.After(newest) {
			newest = it.Published
		}
	}
	if p.marks != nil && newest.After(mark) {
		if err := p.marks.SetMark(ctx, f.Name, newest); err != nil {
			log.Printf("rss: feed %s: save mark: %v", f.Name, err)
		}
	}
	return written, nil
}

// mark — how far this feed was read. A feed with no mark starts MaxAge ago, so
// a first poll takes today's headlines instead of the whole archive.
func (p *Poller) mark(ctx context.Context, feed string, now time.Time) time.Time {
	cold := now.Add(-p.cfg.MaxAge)
	if p.marks == nil {
		return cold
	}
	at, err := p.marks.LastMark(ctx, feed)
	if err != nil || at.IsZero() {
		return cold
	}
	return at
}

// fresh — two dedup rules, because feeds are inconsistent about dates. A dated
// item must be newer than the mark; an undated one is kept once per process by
// ID. Both are needed: dates alone re-import undated feeds forever, IDs alone
// lose their memory on restart.
func (p *Poller) fresh(f FeedConfig, it Item, mark, now time.Time) bool {
	if !it.Published.IsZero() {
		if !it.Published.After(mark) {
			return false
		}
		// A feed that dates its items in the future (or a clock skew) must not
		// win the mark and mute everything after it.
		return !it.Published.After(now.Add(time.Hour))
	}
	id := it.ID
	if id == "" {
		id = it.Title
	}
	if p.seen[f.Name] == nil {
		p.seen[f.Name] = map[string]bool{}
	}
	if p.seen[f.Name][id] {
		return false
	}
	p.seen[f.Name][id] = true
	return true
}

// write stores one item as a note. Source "rss:<feed>" is what the answer path
// filters on, and what makes a feed note distinguishable from something he said.
func (p *Poller) write(ctx context.Context, f FeedConfig, it Item, now time.Time) error {
	text := NoteText(f, it)
	var vec []float32
	if p.embed != nil {
		v, err := p.embed.Embed(ctx, text)
		if err != nil {
			log.Printf("rss: feed %s: embed: %v", f.Name, err)
		} else {
			vec = v
		}
	}
	ts := it.Published
	if ts.IsZero() {
		ts = now
	}
	if _, err := p.notes.WriteNote(ctx, ts, text, vec, SourceFor(f.Name)); err != nil {
		return fmt.Errorf("write note: %w", err)
	}
	return nil
}

// SourceFor is the note source for a feed.
func SourceFor(feed string) string { return "rss:" + feed }

// SourcePrefix — what the answer path matches to find feed notes.
const SourcePrefix = "rss:"

// NoteText renders an item as the note body. The category is included because
// "что нового по технологиям?" is answered by reading notes, and a note has to
// carry enough to be recognised as belonging to that category.
func NoteText(f FeedConfig, it Item) string {
	var b strings.Builder
	b.WriteString(it.Title)
	if f.Category != "" {
		fmt.Fprintf(&b, " [%s]", f.Category)
	}
	if it.Summary != "" {
		b.WriteString("\n")
		b.WriteString(trimRunes(it.Summary, 500))
	}
	if it.Link != "" {
		b.WriteString("\n")
		b.WriteString(it.Link)
	}
	return b.String()
}

// trimRunes cuts on a rune boundary — a note is Russian as often as English and
// half a cyrillic letter is a broken note.
func trimRunes(s string, max int) string {
	r := []rune(s)
	if len(r) <= max {
		return s
	}
	return strings.TrimSpace(string(r[:max])) + "…"
}

// Matches applies the per-feed keyword filter: keep when Include is empty or one
// include matches, drop when any exclude matches. Case-insensitive substring,
// which for Russian is the honest choice — no stemmer here, so "выборы" does not
// match "выборах", and a filter he writes is a filter he can predict.
func Matches(f FeedConfig, it Item) bool {
	hay := strings.ToLower(it.Title + " " + it.Summary)
	for _, x := range f.Exclude {
		if x = strings.ToLower(strings.TrimSpace(x)); x != "" && strings.Contains(hay, x) {
			return false
		}
	}
	if len(f.Include) == 0 {
		return true
	}
	for _, in := range f.Include {
		if in = strings.ToLower(strings.TrimSpace(in)); in != "" && strings.Contains(hay, in) {
			return true
		}
	}
	return false
}