path: root/cmds/serve.go

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