minstrel/internal/library/scheduler.go

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