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