Maven/internal/rss/poller.go

package rss

import (
	"bytes"
	"context"
	"fmt"
	"log"
	"sort"
	"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]*seenIDs // feed → item IDs, for items with no date
	polled  map[string]bool     // feed → polled at least once in THIS process
}

// 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) == "" {
			// Name the offender. A silent skip in a list of six feeds is a
			// config typo nobody finds.
			log.Printf("rss: skipping feed %q (%q): a feed needs both a name and a url", f.Name, f.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]*seenIDs{},
		polled:  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
	}
	// bytes.NewReader and not strings.NewReader(string(…)): the latter copied a
	// feed document that can run to a megabyte, for nothing.
	feed, err := Parse(bytes.NewReader(body.Bytes))
	if err != nil {
		return 0, err
	}

	mark, durable := p.mark(ctx, f.Name, now)
	// resync — the first poll of this feed since the process started, on a feed
	// we have read before. Undated items are deduped by an in-memory ID set that
	// dies with the process, so on this poll they are all "unseen" again and
	// would all be re-noted. See fresh.
	resync := durable && !p.polled[f.Name]
	p.polled[f.Name] = true

	// Gather first, cap second, and write OLDEST first.
	//
	// The old loop walked the feed newest-first and stopped at MaxItems, then
	// marked the newest item it had written. Feeds are newest-first, so with
	// twenty new items and a cap of five it wrote the five newest and moved the
	// mark past all twenty: items six through twenty were older than the mark on
	// the next poll and were dropped for good. max_items reads as a pacing knob
	// in the config doc, and that made it a silent loss. Writing the oldest five
	// and marking the newest of THOSE is pacing: the rest arrive over the polls
	// that follow, in order, each one exactly once.
	var cands []Item
	sawUndated := false
	for _, it := range feed.Items {
		if it.Published.IsZero() {
			sawUndated = true
		}
		if !p.fresh(f, it, mark, now, resync) {
			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
			}
		}
		cands = append(cands, it)
	}
	sort.SliceStable(cands, func(i, j int) bool {
		return itemTime(cands[i], now).Before(itemTime(cands[j], now))
	})
	if len(cands) > p.cfg.MaxItems {
		cands = cands[:p.cfg.MaxItems]
	}

	newest := time.Time{}
	written := 0
	for _, it := range cands {
		if err := p.write(ctx, f, it, now); err != nil {
			return written, err
		}
		written++
		if it.Published.After(newest) {
			newest = it.Published
		}
	}
	p.advance(ctx, f.Name, mark, newest, sawUndated, now)
	return written, nil
}

// itemTime — an item's own date, or now when the feed did not give one. Undated
// items sort last, which is the only defensible guess: they were seen now.
func itemTime(it Item, now time.Time) time.Time {
	if it.Published.IsZero() {
		return now
	}
	return it.Published
}

// advance moves the durable mark to the newest item actually WRITTEN. Because
// the cap is applied to the oldest candidates (see PollFeed), that is never
// ahead of an item still waiting to be read.
//
// A feed whose items carry no dates gets the mark set to now instead. Nothing
// else would ever set it, and the mark's existence is what tells the next
// process that this feed has been read before.
func (p *Poller) advance(ctx context.Context, feed string, mark, newest time.Time, sawUndated bool, now time.Time) {
	if p.marks == nil {
		return
	}
	at := newest
	if at.IsZero() && sawUndated {
		at = now
	}
	if at.IsZero() || !at.After(mark) {
		return
	}
	if err := p.marks.SetMark(ctx, feed, at); err != nil {
		log.Printf("rss: feed %s: save mark: %v", feed, err)
	}
}

// mark — how far this feed was read, and whether that came from the durable
// store. A feed with no mark starts MaxAge ago, so a first poll takes today's
// headlines instead of the whole archive; durable is false in that case, and it
// is what tells PollFeed the difference between "never read" and "read by an
// earlier process".
func (p *Poller) mark(ctx context.Context, feed string, now time.Time) (time.Time, bool) {
	cold := now.Add(-p.cfg.MaxAge)
	if p.marks == nil {
		return cold, false
	}
	at, err := p.marks.LastMark(ctx, feed)
	if err != nil || at.IsZero() {
		return cold, false
	}
	return at, true
}

// maxSeenPerFeed bounds the undated-item set. It has to stay comfortably above
// any one feed's front page, or an item still listed there would fall out of the
// set and be written a second time. A few hundred entries covers the largest
// page anyone publishes, and the set only has to span one poll window plus the
// resync guard, not all of history.
const maxSeenPerFeed = 512

// seenIDs is a bounded insertion-ordered set. The map answers the lookup, the
// slice remembers what to drop first, so an undated feed cannot grow the poller
// for as long as mavend runs.
type seenIDs struct {
	ids   map[string]bool
	order []string
}

// add records id and reports whether it was new.
func (s *seenIDs) add(id string) bool {
	if s.ids == nil {
		s.ids = make(map[string]bool, maxSeenPerFeed)
	}
	if s.ids[id] {
		return false
	}
	s.ids[id] = true
	s.order = append(s.order, id)
	if len(s.order) > maxSeenPerFeed {
		delete(s.ids, s.order[0])
		s.order = s.order[1:]
	}
	return true
}

// 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.
//
// The ID set does not survive a restart, and on its own that re-notes an undated
// feed's whole front page on every boot — five notes, then five more, all stamped
// `now`, sitting at the top of the recent-notes window and crowding out the notes
// he actually made. A crash loop turns it into a flood. So on the first poll after
// a restart of a feed we have read before (resync), undated items are recorded as
// seen and NOT written. The cost is the undated items that appeared while the
// daemon was down. That is a bounded loss, and the alternative is an unbounded one.
func (p *Poller) fresh(f FeedConfig, it Item, mark, now time.Time, resync bool) 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] = &seenIDs{}
	}
	if !p.seen[f.Name].add(id) {
		return false
	}
	return !resync
}

// 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()
}

// NoteHeadline is the part of a feed note she reads out: the first line with the
// category tag taken off. The tag is bookkeeping for the answer path, and piper
// says brackets out loud — "Заголовок [технологии]" is what he heard before.
func NoteHeadline(text string) string {
	line := text
	if i := strings.IndexByte(line, '\n'); i >= 0 {
		line = line[:i]
	}
	line = strings.TrimSpace(line)
	if head, _, ok := splitTag(line); ok {
		return head
	}
	return line
}

// NoteCategory is the category tag a feed note carries, empty when it has none.
// Matching a topic against THIS rather than against the whole note is what keeps
// "что нового про погоду" from matching a tech headline whose link happens to
// contain "pogod".
func NoteCategory(text string) string {
	line := text
	if i := strings.IndexByte(line, '\n'); i >= 0 {
		line = line[:i]
	}
	_, tag, _ := splitTag(strings.TrimSpace(line))
	return tag
}

// splitTag pulls a trailing "[...]" off a headline. Only a trailing one: a title
// that opens with "[перевод]" is the feed's own word, not ours.
func splitTag(line string) (head, tag string, ok bool) {
	if !strings.HasSuffix(line, "]") {
		return line, "", false
	}
	i := strings.LastIndexByte(line, '[')
	if i < 0 {
		return line, "", false
	}
	return strings.TrimSpace(line[:i]), strings.TrimSpace(line[i+1 : len(line)-1]), true
}

// 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
}