package rss import ( "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:" 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:"). 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 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]map[string]bool{}, 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 } feed, err := Parse(strings.NewReader(string(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 } // 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] = map[string]bool{} } if p.seen[f.Name][id] { return false } p.seen[f.Name][id] = true return !resync } // write stores one item as a note. Source "rss:" 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 }