fluxrec: personal Miniflux article recommender

1 files changed, 452 insertions, 0 deletions

diff --git a/cmds/serve.go b/cmds/serve.go
new file mode 100644
index 0000000..9d82d38
--- /dev/null
+++ b/cmds/serve.go
@@ -0,0 +1,452 @@
+// Serve command: the long-running server lane. Polls Miniflux for entries
+// newer than a persisted cursor, scores each one exactly once, assembles a
+// top-N + exploration batch per refresh into runs.jsonl, and serves the
+// recent batches as an RSS feed plus a JSON status endpoint. Strictly
+// read-only toward Miniflux (the client has no mutation methods).
+//
+// Scoring text is the same construction as training: title plus the
+// feed-identity token. Batch = top-N by raw score (no threshold cutoff)
+// with the exploration budget split at a fixed 1/3 runner-up + 2/3 random.
+package cmds
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"math"
+	"math/rand"
+	"net/http"
+	"os"
+	"os/signal"
+	"sort"
+	"strconv"
+	"strings"
+	"sync/atomic"
+	"syscall"
+	"time"
+
+	"fluxrec/core"
+	"fluxrec/miniflux"
+)
+
+type ServeCommand struct {
+	ModelPath   string
+	Cursor      string
+	Runs        string
+	Addr        string
+	RecFeedID   int64
+	TopN        int
+	ExploreFrac float64
+	RSSBatches  int
+	RefreshAt   string
+
+	pollsOK     atomic.Int64
+	pollsFailed atomic.Int64
+}
+
+func (c *ServeCommand) Name() string { return "serve" }
+
+func (c *ServeCommand) Init(args []string) error {
+	fs := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	fs.Usage = func() {
+		fmt.Fprint(fs.Output(), `Usage: fluxrec serve [flags]
+
+Poll Miniflux for new entries, score each once, keep a top-N + exploration
+batch per refresh in runs.jsonl, and serve the recent batches as RSS at
+/recommendations.xml plus JSON status at /api/status. Read-only toward
+Miniflux. Refresh times are wall-clock in the process timezone ($TZ,
+default UTC). Exits immediately if the model file is missing or corrupt.
+`)
+		fs.PrintDefaults()
+	}
+	fs.StringVar(&c.ModelPath, "model", "model.json", "Path to model.json")
+	fs.StringVar(&c.Cursor, "cursor", "cursor.json", "Path to the poll cursor file")
+	fs.StringVar(&c.Runs, "runs", "runs.jsonl", "Path to the run log (doubles as the RSS render source)")
+	fs.StringVar(&c.Addr, "addr", ":8080", "Listen address for RSS + status")
+	fs.Int64Var(&c.RecFeedID, "rec-feed-id", 0, "Miniflux feed ID of the recommendations feed itself; its entries are skipped so serve never re-recommends its own output (0 = not subscribed yet)")
+	fs.IntVar(&c.TopN, "top-n", 15, "Batch size: ranked picks plus exploration")
+	fs.Float64Var(&c.ExploreFrac, "explore-frac", 0.2, "Fraction of the batch spent on exploration (1/3 runner-up, 2/3 random)")
+	fs.IntVar(&c.RSSBatches, "rss-batches", 21, "How many recent batches the RSS feed renders")
+	fs.StringVar(&c.RefreshAt, "refresh-at", "06:00,12:00,18:00", "Comma-separated daily refresh times (HH:MM, local time)")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	if c.TopN < 1 {
+		return fmt.Errorf("--top-n must be >= 1, got %d", c.TopN)
+	}
+	if c.ExploreFrac < 0 || c.ExploreFrac >= 1 {
+		return fmt.Errorf("--explore-frac must be in [0,1), got %v", c.ExploreFrac)
+	}
+	if c.RSSBatches < 1 {
+		return fmt.Errorf("--rss-batches must be >= 1, got %d", c.RSSBatches)
+	}
+	if _, err := parseRefreshAt(c.RefreshAt); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (c *ServeCommand) Run(stdin io.Reader, stdout io.Writer) error {
+	cfg, err := miniflux.ConfigFromEnv()
+	if err != nil {
+		return err
+	}
+	client := miniflux.NewClient(cfg)
+
+	model, err := loadModel(c.ModelPath)
+	if err != nil {
+		return err
+	}
+	vec := core.CreateVectorizerFromModel(model)
+	mins, _ := parseRefreshAt(c.RefreshAt) // validated in Init
+
+	rng := rand.New(rand.NewSource(time.Now().UnixNano()))
+	poll := func() {
+		if err := c.pollOnce(client, vec, model, rng, stdout); err != nil {
+			c.pollsFailed.Add(1)
+			fmt.Fprintf(os.Stderr, "serve: poll failed (cursor not advanced, will retry next cycle): %v\n", err)
+		} else {
+			c.pollsOK.Add(1)
+		}
+	}
+
+	// Boot catch-up: if the last logged run is older than half the smallest
+	// gap between refreshes (e.g. the container was down across a slot),
+	// poll immediately rather than waiting for the next scheduled time.
+	due, err := catchUpDue(c.Runs, mins, time.Now())
+	if err != nil {
+		return err
+	}
+	if due {
+		poll()
+	}
+
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	srv := &http.Server{
+		Addr:         c.Addr,
+		Handler:      c.routes(mins, model.CreatedAt),
+		ReadTimeout:  core.DefaultReadTimeout,
+		WriteTimeout: core.DefaultWriteTimeout,
+		IdleTimeout:  core.DefaultIdleTimeout,
+	}
+	go func() {
+		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			fmt.Fprintf(os.Stderr, "serve: http listener failed: %v\n", err)
+			stop()
+		}
+	}()
+
+	go func() {
+		for {
+			timer := time.NewTimer(time.Until(nextRefresh(time.Now(), mins)))
+			select {
+			case <-ctx.Done():
+				timer.Stop()
+				return
+			case <-timer.C:
+				poll()
+			}
+		}
+	}()
+
+	fmt.Fprintf(stdout, "serve: listening on %s (refresh at %s, model %s)\n", c.Addr, c.RefreshAt, c.ModelPath)
+	<-ctx.Done()
+
+	shutdownCtx, cancel := context.WithTimeout(context.Background(), core.DefaultShutdownTimeout)
+	defer cancel()
+	return srv.Shutdown(shutdownCtx)
+}
+
+// pollCursor is the persisted serve cursor: the largest entry ID fully
+// processed so far.
+type pollCursor struct {
+	AfterEntryID int64 `json:"after_entry_id"`
+}
+
+func loadCursor(path string) (int64, error) {
+	body, err := os.ReadFile(path)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return 0, nil
+		}
+		return 0, fmt.Errorf("reading cursor %s: %w", path, err)
+	}
+	var c pollCursor
+	if err := json.Unmarshal(body, &c); err != nil {
+		return 0, fmt.Errorf("parsing cursor %s: %w", path, err)
+	}
+	return c.AfterEntryID, nil
+}
+
+func saveCursor(path string, id int64) error {
+	body, err := json.Marshal(pollCursor{AfterEntryID: id})
+	if err != nil {
+		return err
+	}
+	if err := os.WriteFile(path, body, 0o644); err != nil {
+		return fmt.Errorf("writing cursor %s: %w", path, err)
+	}
+	return nil
+}
+
+// pollOnce fetches every entry newer than the cursor, scores each one,
+// assembles the batch, appends one row to runs.jsonl, and only then
+// advances the cursor — process → append log → save cursor, so a crash at
+// any point re-processes at worst one batch. Nothing is written for an
+// empty poll (a run of empty batches would starve the count-based RSS
+// tail).
+func (c *ServeCommand) pollOnce(client *miniflux.Client, vec *core.TFIDFVectorizer, model *core.ModelEnvelope, rng *rand.Rand, stdout io.Writer) error {
+	started := time.Now()
+	cursor, err := loadCursor(c.Cursor)
+	if err != nil {
+		return err
+	}
+
+	var scored []scoredEntry
+	seen, recSkipped := 0, 0
+	maxID := cursor
+	f := miniflux.Filters{AfterEntryID: cursor, Limit: 200, Direction: "asc"}
+	for {
+		set, err := client.Entries(f)
+		if err != nil {
+			return err
+		}
+		if len(set.Entries) == 0 {
+			break
+		}
+		for _, e := range set.Entries {
+			seen++
+			if e.ID > maxID {
+				maxID = e.ID
+			}
+			// The recommendations feed's own items are our output, not
+			// candidates: skip them or the loop feeds itself.
+			if c.RecFeedID != 0 && e.Feed.ID == c.RecFeedID {
+				recSkipped++
+				continue
+			}
+			vecs := vec.Transform([]string{core.ScoringText(e.Title, e.Feed.Title)})
+			s, err := core.PredictScore(vecs[0], model.Weights)
+			if err != nil {
+				return fmt.Errorf("scoring entry %d: %w", e.ID, err)
+			}
+			scored = append(scored, scoredEntry{entry: e, score: s})
+		}
+		if len(set.Entries) < f.Limit || maxID == f.AfterEntryID {
+			// Short page: done. Unchanged max ID despite a full page means
+			// the server ignored the cursor — break rather than spin.
+			break
+		}
+		f.AfterEntryID = maxID
+	}
+
+	if seen == 0 {
+		fmt.Fprintln(stdout, "serve: no new entries")
+		return nil
+	}
+
+	row := core.RunRow{
+		RanAt:          started.UTC(),
+		EntriesSeen:    seen,
+		RecFeedSkipped: recSkipped,
+		Kept:           assembleBatch(scored, c.TopN, c.ExploreFrac, rng),
+		DurationMs:     time.Since(started).Milliseconds(),
+		ModelCreatedAt: model.CreatedAt,
+	}
+	if err := core.AppendRun(c.Runs, row); err != nil {
+		return err
+	}
+	if err := saveCursor(c.Cursor, maxID); err != nil {
+		return err
+	}
+	fmt.Fprintf(stdout, "serve: seen=%d kept=%d skipped-rec=%d cursor=%d\n",
+		seen, len(row.Kept), recSkipped, maxID)
+	return nil
+}
+
+// scoredEntry is an entry with its model score; batch assembly input.
+type scoredEntry struct {
+	entry miniflux.Entry
+	score float64
+}
+
+// assembleBatch ranks scored entries by score descending and selects the
+// batch: the top (topN minus the exploration budget) are ranked; the
+// budget splits at a fixed 1/3 : 2/3 into runner-ups (highest-scored just
+// below the cut — the model's borderline cases) and random picks from the
+// remaining tail (serendipity and a recall-ceiling estimator). With fewer
+// candidates than the ranked cut, everything is kept as ranked.
+func assembleBatch(entries []scoredEntry, topN int, exploreFrac float64, rng *rand.Rand) []core.KeptItem {
+	ranked := make([]scoredEntry, len(entries))
+	copy(ranked, entries)
+	sort.SliceStable(ranked, func(i, j int) bool {
+		if ranked[i].score == ranked[j].score {
+			return ranked[i].entry.ID < ranked[j].entry.ID
+		}
+		return ranked[i].score > ranked[j].score
+	})
+
+	explore := int(math.Round(float64(topN) * exploreFrac))
+	nRanked := topN - explore
+	if len(ranked) <= nRanked {
+		return keptFrom(ranked, core.KindRanked)
+	}
+	kept := keptFrom(ranked[:nRanked], core.KindRanked)
+
+	nRunnerUp := min(explore/3, len(ranked)-nRanked)
+	kept = append(kept, keptFrom(ranked[nRanked:nRanked+nRunnerUp], core.KindRunnerUp)...)
+
+	rest := ranked[nRanked+nRunnerUp:]
+	nRandom := min(explore-nRunnerUp, len(rest))
+	if nRandom > 0 {
+		picks := make([]scoredEntry, 0, nRandom)
+		for _, i := range rng.Perm(len(rest))[:nRandom] {
+			picks = append(picks, rest[i])
+		}
+		// Log random picks in ID order so runs.jsonl stays diff-friendly.
+		sort.Slice(picks, func(i, j int) bool { return picks[i].entry.ID < picks[j].entry.ID })
+		kept = append(kept, keptFrom(picks, core.KindRandom)...)
+	}
+	return kept
+}
+
+// keptFrom converts scored entries into run-log items, cleaning content on
+// the way in so RSS rendering never sees raw feed HTML.
+func keptFrom(entries []scoredEntry, kind string) []core.KeptItem {
+	out := make([]core.KeptItem, len(entries))
+	for i, se := range entries {
+		out[i] = core.KeptItem{
+			EntryID:     se.entry.ID,
+			URL:         se.entry.URL,
+			Title:       se.entry.Title,
+			FeedTitle:   se.entry.Feed.Title,
+			PublishedAt: se.entry.PublishedAt.UTC(),
+			Content:     core.CleanFeedContent(se.entry.Content),
+			Score:       se.score,
+			Kind:        kind,
+		}
+	}
+	return out
+}
+
+// parseRefreshAt parses "06:00,12:00,18:00" into ascending
+// minutes-after-midnight.
+func parseRefreshAt(spec string) ([]int, error) {
+	var mins []int
+	for _, part := range strings.Split(spec, ",") {
+		part = strings.TrimSpace(part)
+		hm := strings.Split(part, ":")
+		if len(hm) != 2 {
+			return nil, fmt.Errorf("--refresh-at: %q is not HH:MM", part)
+		}
+		h, errH := strconv.Atoi(hm[0])
+		m, errM := strconv.Atoi(hm[1])
+		if errH != nil || errM != nil || h < 0 || h > 23 || m < 0 || m > 59 {
+			return nil, fmt.Errorf("--refresh-at: %q is not a valid time", part)
+		}
+		mins = append(mins, h*60+m)
+	}
+	sort.Ints(mins)
+	return mins, nil
+}
+
+// nextRefresh returns the next occurrence of any refresh time strictly
+// after now, in now's location, wrapping to tomorrow past the last slot.
+func nextRefresh(now time.Time, mins []int) time.Time {
+	var best time.Time
+	for _, m := range mins {
+		t := time.Date(now.Year(), now.Month(), now.Day(), m/60, m%60, 0, 0, now.Location())
+		if !t.After(now) {
+			t = t.Add(24 * time.Hour)
+		}
+		if best.IsZero() || t.Before(best) {
+			best = t
+		}
+	}
+	return best
+}
+
+// catchUpDue reports whether startup should trigger an immediate poll:
+// nothing logged yet, or the last run is older than half the smallest gap
+// between refresh times (a container down across a slot catches up on boot
+// instead of waiting for the next one).
+func catchUpDue(runsPath string, mins []int, now time.Time) (bool, error) {
+	rows, err := core.ReadRunsTail(runsPath, 1)
+	if err != nil {
+		return false, err
+	}
+	if len(rows) == 0 {
+		return true, nil
+	}
+	half := time.Duration(minGapMinutes(mins)) * time.Minute / 2
+	return now.Sub(rows[len(rows)-1].RanAt) > half, nil
+}
+
+// minGapMinutes returns the smallest gap between consecutive refresh times,
+// including the wrap across midnight (a single time has a 24h gap).
+func minGapMinutes(mins []int) int {
+	if len(mins) < 2 {
+		return 24 * 60
+	}
+	sorted := make([]int, len(mins))
+	copy(sorted, mins)
+	sort.Ints(sorted)
+	gap := 24*60 - sorted[len(sorted)-1] + sorted[0]
+	for i := 1; i < len(sorted); i++ {
+		gap = min(gap, sorted[i]-sorted[i-1])
+	}
+	return gap
+}
+
+// routes wires the two surfaces: the RSS feed and the JSON status.
+func (c *ServeCommand) routes(mins []int, modelCreatedAt time.Time) http.Handler {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/recommendations.xml", func(w http.ResponseWriter, r *http.Request) {
+		runs, err := core.ReadRunsTail(c.Runs, c.RSSBatches)
+		if err != nil {
+			http.Error(w, "reading runs: "+err.Error(), http.StatusInternalServerError)
+			return
+		}
+		body, err := RenderRSS(runs)
+		if err != nil {
+			http.Error(w, "rendering RSS: "+err.Error(), http.StatusInternalServerError)
+			return
+		}
+		w.Header().Set("Content-Type", "application/rss+xml; charset=utf-8")
+		_, _ = w.Write(body)
+	})
+	mux.HandleFunc("/api/status", func(w http.ResponseWriter, r *http.Request) {
+		runs, err := core.ReadRunsTail(c.Runs, 1)
+		if err != nil {
+			http.Error(w, "reading runs: "+err.Error(), http.StatusInternalServerError)
+			return
+		}
+		status := map[string]any{
+			"model_created_at": modelCreatedAt.UTC(),
+			"model_age_hours":  math.Round(time.Since(modelCreatedAt).Hours()),
+			"next_refresh_at":  nextRefresh(time.Now(), mins),
+			"polls_ok":         c.pollsOK.Load(),
+			"polls_failed":     c.pollsFailed.Load(),
+			"last_run":         nil,
+		}
+		if len(runs) > 0 {
+			last := runs[len(runs)-1]
+			status["last_run"] = map[string]any{
+				"ran_at":           last.RanAt,
+				"entries_seen":     last.EntriesSeen,
+				"rec_feed_skipped": last.RecFeedSkipped,
+				"kept":             len(last.Kept),
+				"duration_ms":      last.DurationMs,
+				"model_created_at": last.ModelCreatedAt,
+			}
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(status)
+	})
+	return mux
+}