diff --git a/cmd/minstrel/main.go b/cmd/minstrel/main.go index ea83344f..b5ff5480 100644 --- a/cmd/minstrel/main.go +++ b/cmd/minstrel/main.go @@ -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) diff --git a/internal/library/scheduler.go b/internal/library/scheduler.go new file mode 100644 index 00000000..378543c7 --- /dev/null +++ b/internal/library/scheduler.go @@ -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) + } +} diff --git a/internal/library/scheduler_test.go b/internal/library/scheduler_test.go new file mode 100644 index 00000000..0e76d68a --- /dev/null +++ b/internal/library/scheduler_test.go @@ -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") + } +}