Maven/cmd/mavend/feeds.go

// mavend/feeds.go — the driver for RSS/Atom reading (Vikunja #258,
// docs/plans/13-rss-news-feeds.md). The reader itself is pure and lives in
// internal/rss; this is the impure half: a ticker, the guarded fetcher, and the
// two adapters that let a pure package talk to the store.
//
// Why in-core rather than its own daemon like mavmaild and mavpoll: those two
// hold a CREDENTIAL (an IMAP password, a zenmoney token), and the reason they
// are separate processes is that core must never see it. A feed URL is public,
// there is no secret to isolate, and a whole extra binary and compose service
// would buy nothing. The other half of the mavpoll precedent — off unless
// configured — is kept: no `feeds` block, no poller, no outbound request.
//
// It is its own goroutine, not a step on the tick: the tick has a delivery
// deadline behind it, and a feed read is a network round-trip that nobody is
// waiting on.
//
// Nothing here dispatches. A feed that announced itself would be a nag, so the
// only output is notes with source "rss:<feed>", which the answer path reads
// when he asks ("что нового в лентах?" — see queryFeeds in actions_query.go).
package main

import (
	"context"
	"log"
	"net/url"
	"time"

	"github.com/kami/maven/internal/config"
	"github.com/kami/maven/internal/ipc"
	"github.com/kami/maven/internal/router"
	"github.com/kami/maven/internal/rss"
	"github.com/kami/maven/internal/webfetch"
)

// feedWorker — ticker + poller.
type feedWorker struct {
	poller   *rss.Poller
	interval time.Duration
}

// feedTickInterval — how often the worker asks the poller what is due. Per-feed
// cadence is the poller's business; this is just the granularity.
const feedTickInterval = 5 * time.Minute

// newFeedWorker wires feed reading, or returns nil when it must not run:
// no `feeds` block (the normal case), or nothing valid in it. Every caller
// checks for nil.
func newFeedWorker(api ipc.CoreAPI, emb router.Embedder, cfg *config.Config) *feedWorker {
	if cfg.Feeds == nil {
		return nil
	}
	fc := cfg.Feeds

	feeds := make([]rss.FeedConfig, 0, len(fc.Sources))
	hosts := append([]string(nil), fc.AllowHosts...)
	for _, s := range fc.Sources {
		feeds = append(feeds, rss.FeedConfig{
			Name:     s.Name,
			URL:      s.URL,
			Category: s.Category,
			Interval: time.Duration(s.Interval),
			Include:  s.Include,
			Exclude:  s.Exclude,
		})
		// Each configured feed's own host is allowed. The allowlist is then
		// exactly "the feeds he asked for", so a redirect off to somewhere else
		// is refused by the fetcher rather than followed.
		if u, err := url.Parse(s.URL); err == nil && u.Hostname() != "" {
			hosts = append(hosts, u.Hostname())
		}
	}

	fetcher := webfetch.New(webfetch.Config{
		AllowHosts: hosts,
		Timeout:    time.Duration(fc.Timeout),
		MaxBytes:   fc.MaxBytes,
	})
	poller := rss.NewPoller(feeds, &feedFetcher{f: fetcher}, api, &factMarks{api: api},
		embedderFor(emb), nil, rss.Config{
			DefaultInterval: time.Duration(fc.PollInterval),
			MaxItems:        fc.MaxItems,
			MaxAge:          time.Duration(fc.MaxAge),
		})
	if poller == nil {
		log.Printf("feeds: configured but nothing pollable — feed reading disabled")
		return nil
	}
	log.Printf("feeds: reading %d feed(s), checking what is due every %s", len(feeds), feedTickInterval)
	return &feedWorker{poller: poller, interval: feedTickInterval}
}

// run polls what is due until ctx is canceled. The first round runs immediately
// so a restart does not blind her for the first interval; it writes notes only,
// so an early round cannot startle anyone.
func (w *feedWorker) run(ctx context.Context) {
	w.poller.PollDue(ctx, time.Now())
	t := time.NewTicker(w.interval)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case now := <-t.C:
			w.poller.PollDue(ctx, now)
		}
	}
}

// embedderOf — the voice wiring's embedder, or nil when voice is not wired.
// Feed notes are embedded with the SAME model the rest of the store uses, or not
// at all; a second embedder would write vectors nothing can search.
func embedderOf(w *voiceWiring) router.Embedder {
	if w == nil {
		return nil
	}
	return w.embedder
}

// feedFetcher adapts webfetch to rss.Fetcher — the pure package names the two
// fields it needs and stays free of net/http.
type feedFetcher struct{ f *webfetch.Fetcher }

func (a *feedFetcher) Get(ctx context.Context, u string) (*rss.Body, error) {
	resp, err := a.f.Get(ctx, u)
	if err != nil {
		return nil, err
	}
	return &rss.Body{Bytes: resp.Body}, nil
}

// factMarks stores "how far this feed was read" as a config fact, the same
// mechanism the plan named and the same one the pattern tick uses for its own
// bookkeeping. Durable, inspectable on /dash, and cheap.
type factMarks struct{ api ipc.CoreAPI }

func markKey(feed string) string { return "rss:latest:" + feed }

func (m *factMarks) LastMark(ctx context.Context, feed string) (time.Time, error) {
	f, err := m.api.LatestFact(ctx, markKey(feed))
	if err != nil {
		// No mark yet is not an error worth propagating: the poller treats a
		// zero time as a cold start.
		return time.Time{}, nil
	}
	t, err := time.Parse(time.RFC3339, f.Value)
	if err != nil {
		return time.Time{}, nil
	}
	return t, nil
}

func (m *factMarks) SetMark(ctx context.Context, feed string, at time.Time) error {
	_, err := m.api.WriteFact(ctx, ipc.WriteFactReq{
		Ts:         time.Now(),
		Kind:       "config",
		Key:        markKey(feed),
		Value:      at.UTC().Format(time.RFC3339),
		Source:     "poll:rss",
		Confidence: 1.0,
	})
	return err
}

// embedderFor adapts router.Embedder to rss.Embedder, and returns nil when
// there is none — a note without a vector is still a note the recent-notes path
// can read.
//
// EmbedPassage, not Embed: a feed item is text being searched FOR, and the e5
// embedder is asymmetric. Getting this backwards makes the item unfindable by
// the question that should have matched it.
func embedderFor(emb router.Embedder) rss.Embedder {
	if emb == nil {
		return nil
	}
	return passageEmbedder{emb}
}

type passageEmbedder struct{ e router.Embedder }

func (p passageEmbedder) Embed(ctx context.Context, text string) ([]float32, error) {
	return router.EmbedPassage(ctx, p.e, text)
}