feat(server/m7-recurring-scans): Scheduler goroutine + boot wiring

Adds ScheduleConfig + NextFire (per-mode logic for off/interval/ daily/weekly) and the Scheduler struct + loop goroutine. The loop reads scan_schedule on boot and on Refresh signals, computes next-fire, sleeps via time.Timer, and calls TryStartScan when the timer fires. Skip-on-conflict logging happens inside tryFire when TryStartScan reports started=false with a non-stale in-flight row. Boot wiring: cmd/minstrel/main.go constructs the scheduler after coverEnricher and calls Start with the server's long-lived ctx. The scheduler is not yet plumbed into the HTTP handlers (the next task does that — handlers gain a scheduler field, admin schedule endpoints land). Tests cover the NextFire truth table (off / interval / daily today + tomorrow / weekly today-before-time + today-after-time + upcoming day) plus invalid configs and parseHHMM edge cases. Refresh is verified non-blocking via the buffered-channel + default pattern. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-06 22:20:56 -04:00
parent 8784e938bd
commit 4607ce6db4
3 changed files with 371 additions and 0 deletions
@@ -163,6 +163,10 @@ func run() error {
 		ArtistEnrichCap: cfg.Library.CoverArtBackfillCap,
 		DataDir:         cfg.Storage.DataDir,
 	}
+	scheduler := library.NewScheduler(pool, logger.With("component", "scheduler"),
+		scanner, coverEnricher, scanCfg)
+	scheduler.Start(ctx)
+	_ = scheduler // keep reachable for future wiring (Task 4)
 	srv := server.New(logger, pool, scanner, subsonic.Config{
 		AllowPlaintextPassword: cfg.Subsonic.AllowPlaintextPassword,
 	}, cfg.Events, cfg.Recommendation, cfg.Storage.DataDir, cfg.Branding, coverEnricher, cfg.Library.CoverArtBackfillCap, coverSettings, scanner, scanCfg)
@@ -0,0 +1,238 @@
+package library
+
+import (
+	"context"
+	"log/slog"
+	"sync"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"git.fabledsword.com/bvandeusen/minstrel/internal/coverart"
+	"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
+)
+
+// ScheduleConfig represents the operator's recurring scan schedule.
+// Loaded from the scan_schedule singleton row on boot and on Refresh.
+type ScheduleConfig struct {
+	Mode          string // "off" | "interval" | "daily" | "weekly"
+	IntervalHours int
+	TimeOfDay     string // "HH:MM" (24h)
+	WeeklyDay     int    // 1..7 (Mon..Sun, ISO 8601)
+}
+
+// NextFire returns the next time the scan should fire given a reference
+// "now". Returns zero time when mode == "off" or config is invalid.
+//
+// DST semantics: "daily 02:30" on a spring-forward day will fire at the
+// computed time using the local zone's rules (Go time.Date silently
+// normalizes a non-existent local time). Single-operator scale; documented
+// as known and acceptable.
+func (c ScheduleConfig) NextFire(now time.Time) time.Time {
+	switch c.Mode {
+	case "off":
+		return time.Time{}
+	case "interval":
+		if c.IntervalHours <= 0 {
+			return time.Time{}
+		}
+		return now.Add(time.Duration(c.IntervalHours) * time.Hour)
+	case "daily":
+		h, m, ok := parseHHMM(c.TimeOfDay)
+		if !ok {
+			return time.Time{}
+		}
+		next := time.Date(now.Year(), now.Month(), now.Day(), h, m, 0, 0, now.Location())
+		if !next.After(now) {
+			next = next.AddDate(0, 0, 1)
+		}
+		return next
+	case "weekly":
+		h, m, ok := parseHHMM(c.TimeOfDay)
+		if !ok || c.WeeklyDay < 1 || c.WeeklyDay > 7 {
+			return time.Time{}
+		}
+		// Map ISO weekday (1=Mon..7=Sun) to Go's time.Weekday (0=Sun..6=Sat).
+		targetGoDow := c.WeeklyDay % 7
+		nowGoDow := int(now.Weekday())
+		daysAhead := (targetGoDow - nowGoDow + 7) % 7
+		next := time.Date(now.Year(), now.Month(), now.Day(), h, m, 0, 0, now.Location())
+		next = next.AddDate(0, 0, daysAhead)
+		if !next.After(now) {
+			next = next.AddDate(0, 0, 7)
+		}
+		return next
+	}
+	return time.Time{}
+}
+
+func parseHHMM(s string) (h, m int, ok bool) {
+	if len(s) != 5 || s[2] != ':' {
+		return 0, 0, false
+	}
+	for i, c := range s {
+		if i == 2 {
+			continue
+		}
+		if c < '0' || c > '9' {
+			return 0, 0, false
+		}
+	}
+	h = int(s[0]-'0')*10 + int(s[1]-'0')
+	m = int(s[3]-'0')*10 + int(s[4]-'0')
+	if h < 0 || h > 23 || m < 0 || m > 59 {
+		return 0, 0, false
+	}
+	return h, m, true
+}
+
+// Scheduler runs in a goroutine and triggers RunScan on the operator's
+// configured cadence via library.TryStartScan (which handles in-flight
+// conflicts and stale-row reaping).
+type Scheduler struct {
+	pool     *pgxpool.Pool
+	logger   *slog.Logger
+	scanner  *Scanner
+	enricher *coverart.Enricher
+	scanCfg  RunScanConfig
+
+	mu       sync.RWMutex
+	cfg      ScheduleConfig
+	nextFire time.Time
+
+	refresh chan struct{}
+	stop    chan struct{}
+}
+
+func NewScheduler(pool *pgxpool.Pool, logger *slog.Logger,
+	scanner *Scanner, enricher *coverart.Enricher,
+	scanCfg RunScanConfig,
+) *Scheduler {
+	return &Scheduler{
+		pool: pool, logger: logger,
+		scanner: scanner, enricher: enricher, scanCfg: scanCfg,
+		refresh: make(chan struct{}, 1),
+		stop:    make(chan struct{}),
+	}
+}
+
+// Start launches the scheduler's loop goroutine. The loop runs until
+// the passed ctx is cancelled or Stop is called.
+func (s *Scheduler) Start(ctx context.Context) {
+	go s.loop(ctx)
+}
+
+// Stop signals the loop to exit. Not idempotent — call only once
+// (typically at server shutdown).
+func (s *Scheduler) Stop() { close(s.stop) }
+
+// Refresh signals the loop to reload config and recompute next-fire.
+// Non-blocking: additional Refresh calls coalesce on the buffered channel.
+func (s *Scheduler) Refresh() {
+	select {
+	case s.refresh <- struct{}{}:
+	default:
+	}
+}
+
+// NextFire returns the cached next-fire time. Zero when mode=off.
+func (s *Scheduler) NextFire() time.Time {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.nextFire
+}
+
+func (s *Scheduler) setState(cfg ScheduleConfig, next time.Time) {
+	s.mu.Lock()
+	s.cfg = cfg
+	s.nextFire = next
+	s.mu.Unlock()
+}
+
+func (s *Scheduler) loadConfig(ctx context.Context) (ScheduleConfig, error) {
+	q := dbq.New(s.pool)
+	row, err := q.GetScanSchedule(ctx)
+	if err != nil {
+		return ScheduleConfig{}, err
+	}
+	cfg := ScheduleConfig{Mode: row.Mode}
+	if row.IntervalHours != nil {
+		cfg.IntervalHours = int(*row.IntervalHours)
+	}
+	if row.TimeOfDay != nil {
+		cfg.TimeOfDay = *row.TimeOfDay
+	}
+	if row.WeeklyDay != nil {
+		cfg.WeeklyDay = int(*row.WeeklyDay)
+	}
+	return cfg, nil
+}
+
+func (s *Scheduler) loop(ctx context.Context) {
+	for {
+		cfg, err := s.loadConfig(ctx)
+		if err != nil {
+			s.logger.Error("scheduler: load config failed", "err", err)
+			retry := time.NewTimer(1 * time.Minute)
+			select {
+			case <-retry.C:
+			case <-s.refresh:
+				retry.Stop()
+			case <-s.stop:
+				retry.Stop()
+				return
+			case <-ctx.Done():
+				retry.Stop()
+				return
+			}
+			continue
+		}
+
+		next := cfg.NextFire(time.Now())
+		s.setState(cfg, next)
+
+		if next.IsZero() {
+			s.logger.Info("scheduler: mode=off, waiting for config change")
+			select {
+			case <-s.refresh:
+				continue
+			case <-s.stop:
+				return
+			case <-ctx.Done():
+				return
+			}
+		}
+
+		wait := time.Until(next)
+		s.logger.Info("scheduler: next scan scheduled", "at", next, "in", wait)
+		timer := time.NewTimer(wait)
+		select {
+		case <-timer.C:
+			s.tryFire(ctx)
+		case <-s.refresh:
+			timer.Stop()
+		case <-s.stop:
+			timer.Stop()
+			return
+		case <-ctx.Done():
+			timer.Stop()
+			return
+		}
+	}
+}
+
+func (s *Scheduler) tryFire(ctx context.Context) {
+	started, existing, err := TryStartScan(
+		ctx, s.pool, s.scanner, s.enricher,
+		s.logger.With("source", "scheduler"), s.scanCfg,
+	)
+	if err != nil {
+		s.logger.Error("scheduler: try start failed", "err", err)
+		return
+	}
+	if !started && existing != nil {
+		s.logger.Info("scheduler: scan skipped — prior still in flight",
+			"in_flight_id", existing.ID,
+			"in_flight_started_at", existing.StartedAt.Time)
+	}
+}
@@ -0,0 +1,129 @@
+package library
+
+import (
+	"testing"
+	"time"
+)
+
+func TestNextFire_Off(t *testing.T) {
+	c := ScheduleConfig{Mode: "off"}
+	if got := c.NextFire(time.Now()); !got.IsZero() {
+		t.Errorf("NextFire = %v, want zero time", got)
+	}
+}
+
+func TestNextFire_Interval(t *testing.T) {
+	c := ScheduleConfig{Mode: "interval", IntervalHours: 6}
+	now := time.Date(2026, 5, 6, 14, 0, 0, 0, time.UTC)
+	want := time.Date(2026, 5, 6, 20, 0, 0, 0, time.UTC)
+	if got := c.NextFire(now); !got.Equal(want) {
+		t.Errorf("NextFire = %v, want %v", got, want)
+	}
+}
+
+func TestNextFire_Daily_Tomorrow(t *testing.T) {
+	c := ScheduleConfig{Mode: "daily", TimeOfDay: "03:00"}
+	now := time.Date(2026, 5, 6, 14, 0, 0, 0, time.UTC)
+	want := time.Date(2026, 5, 7, 3, 0, 0, 0, time.UTC)
+	if got := c.NextFire(now); !got.Equal(want) {
+		t.Errorf("NextFire = %v, want %v", got, want)
+	}
+}
+
+func TestNextFire_Daily_Today(t *testing.T) {
+	c := ScheduleConfig{Mode: "daily", TimeOfDay: "03:00"}
+	now := time.Date(2026, 5, 6, 2, 0, 0, 0, time.UTC)
+	want := time.Date(2026, 5, 6, 3, 0, 0, 0, time.UTC)
+	if got := c.NextFire(now); !got.Equal(want) {
+		t.Errorf("NextFire = %v, want %v", got, want)
+	}
+}
+
+func TestNextFire_Weekly_UpcomingDay(t *testing.T) {
+	// Tuesday 2pm, schedule = Sunday (ISO 7) 03:00 → upcoming Sunday at 03:00.
+	c := ScheduleConfig{Mode: "weekly", WeeklyDay: 7, TimeOfDay: "03:00"}
+	now := time.Date(2026, 5, 5, 14, 0, 0, 0, time.UTC)  // 2026-05-05 = Tuesday
+	want := time.Date(2026, 5, 10, 3, 0, 0, 0, time.UTC) // 2026-05-10 = Sunday
+	if got := c.NextFire(now); !got.Equal(want) {
+		t.Errorf("NextFire = %v, want %v", got, want)
+	}
+}
+
+func TestNextFire_Weekly_TodayBeforeTime(t *testing.T) {
+	c := ScheduleConfig{Mode: "weekly", WeeklyDay: 2, TimeOfDay: "03:00"} // ISO 2 = Tuesday
+	now := time.Date(2026, 5, 5, 2, 0, 0, 0, time.UTC)
+	want := time.Date(2026, 5, 5, 3, 0, 0, 0, time.UTC)
+	if got := c.NextFire(now); !got.Equal(want) {
+		t.Errorf("NextFire = %v, want %v", got, want)
+	}
+}
+
+func TestNextFire_Weekly_TodayAfterTime(t *testing.T) {
+	c := ScheduleConfig{Mode: "weekly", WeeklyDay: 2, TimeOfDay: "03:00"}
+	now := time.Date(2026, 5, 5, 14, 0, 0, 0, time.UTC)
+	want := time.Date(2026, 5, 12, 3, 0, 0, 0, time.UTC)
+	if got := c.NextFire(now); !got.Equal(want) {
+		t.Errorf("NextFire = %v, want %v", got, want)
+	}
+}
+
+func TestNextFire_InvalidConfigs(t *testing.T) {
+	cases := []ScheduleConfig{
+		{Mode: "interval", IntervalHours: 0},
+		{Mode: "interval", IntervalHours: -1},
+		{Mode: "daily", TimeOfDay: "bad"},
+		{Mode: "daily", TimeOfDay: "25:00"},
+		{Mode: "weekly", TimeOfDay: "03:00", WeeklyDay: 0},
+		{Mode: "weekly", TimeOfDay: "03:00", WeeklyDay: 8},
+		{Mode: "unknown"},
+	}
+	now := time.Now()
+	for _, c := range cases {
+		if got := c.NextFire(now); !got.IsZero() {
+			t.Errorf("NextFire(%+v) = %v, want zero", c, got)
+		}
+	}
+}
+
+func TestParseHHMM(t *testing.T) {
+	cases := []struct {
+		in     string
+		h, m   int
+		wantOK bool
+	}{
+		{"03:00", 3, 0, true},
+		{"23:59", 23, 59, true},
+		{"00:00", 0, 0, true},
+		{"24:00", 0, 0, false},
+		{"23:60", 0, 0, false},
+		{"3:00", 0, 0, false},
+		{"03-00", 0, 0, false},
+		{"ab:cd", 0, 0, false},
+	}
+	for _, c := range cases {
+		h, m, ok := parseHHMM(c.in)
+		if ok != c.wantOK || (ok && (h != c.h || m != c.m)) {
+			t.Errorf("parseHHMM(%q) = (%d, %d, %v), want (%d, %d, %v)",
+				c.in, h, m, ok, c.h, c.m, c.wantOK)
+		}
+	}
+}
+
+func TestScheduler_Refresh_NonBlocking(t *testing.T) {
+	s := &Scheduler{
+		refresh: make(chan struct{}, 1),
+		stop:    make(chan struct{}),
+	}
+	s.Refresh()
+	done := make(chan struct{})
+	go func() {
+		s.Refresh()
+		s.Refresh()
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(100 * time.Millisecond):
+		t.Errorf("Refresh blocked when channel was full")
+	}
+}