URBM/internal/scheduler/scheduler.go

package scheduler

import (
	"context"
	"log/slog"
	"sync"
	"time"

	"git.casaderoll.de/michael/urbm/internal/model"
)

type ConfigProvider func() model.Config
type Enqueue func(model.Job) error
type EnqueueMaintenance func(string, string) (model.Run, error)
type LastRun func(string) time.Time

type Scheduler struct {
	config      ConfigProvider
	enqueue     Enqueue
	maintenance EnqueueMaintenance
	lastRun     LastRun
	log         *slog.Logger
	mu          sync.Mutex
	last        map[string]time.Time
}

func New(config ConfigProvider, enqueue Enqueue, maintenance EnqueueMaintenance, lastRun LastRun, log *slog.Logger) *Scheduler {
	return &Scheduler{config: config, enqueue: enqueue, maintenance: maintenance, lastRun: lastRun, log: log, last: map[string]time.Time{}}
}

func (s *Scheduler) Run(ctx context.Context) {
	s.catchUp(time.Now())
	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()
	s.tick(time.Now())
	for {
		select {
		case <-ctx.Done():
			return
		case now := <-ticker.C:
			s.tick(now)
		}
	}
}

func (s *Scheduler) tick(now time.Time) {
	config := s.config()
	minute := now.Truncate(time.Minute)
	for _, job := range config.Jobs {
		if !job.Enabled || job.Schedule.Cron == "" {
			continue
		}
		at, expression, ok := scheduledTime(job, minute, s.log)
		if !ok || !expression.Matches(at) {
			continue
		}
		job := job
		s.once("job:"+job.ID, minute, func() error { return s.enqueue(job) })
	}
	s.scheduleMaintenance(config, minute, "check", config.Settings.CheckCron)
	s.scheduleMaintenance(config, minute, "prune", config.Settings.PruneCron)
}

func (s *Scheduler) catchUp(now time.Time) {
	config := s.config()
	window := time.Duration(config.Settings.CatchUpWindowHrs) * time.Hour
	if window <= 0 {
		return
	}
	for _, job := range config.Jobs {
		if !job.Enabled || job.Schedule.Cron == "" {
			continue
		}
		loc := time.Local
		if job.Schedule.Timezone != "" {
			if loaded, err := time.LoadLocation(job.Schedule.Timezone); err == nil {
				loc = loaded
			}
		}
		expression, err := Parse(job.Schedule.Cron)
		if err != nil {
			continue
		}
		lastScheduled := previous(expression, now.In(loc), window)
		if !lastScheduled.IsZero() && s.lastRun(job.ID).Before(lastScheduled.UTC()) {
			if err := s.enqueue(job); err != nil {
				s.log.Warn("catch-up enqueue failed", "jobId", job.ID, "error", err)
			}
		}
	}
}

func (s *Scheduler) scheduleMaintenance(config model.Config, minute time.Time, action, spec string) {
	if spec == "" {
		return
	}
	expression, err := Parse(spec)
	if err != nil {
		s.log.Warn("invalid maintenance schedule", "action", action, "error", err)
		return
	}
	if !expression.Matches(minute) {
		return
	}
	for _, repo := range config.Repositories {
		repoID := repo.ID
		s.once(action+":"+repoID, minute, func() error {
			_, err := s.maintenance(repoID, action)
			return err
		})
	}
}

func (s *Scheduler) once(key string, minute time.Time, action func() error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.last[key].Equal(minute) {
		return
	}
	s.last[key] = minute
	if err := action(); err != nil {
		s.log.Warn("schedule enqueue failed", "key", key, "error", err)
	}
}

func scheduledTime(job model.Job, minute time.Time, log *slog.Logger) (time.Time, Expression, bool) {
	expression, err := Parse(job.Schedule.Cron)
	if err != nil {
		log.Warn("invalid job schedule", "jobId", job.ID, "error", err)
		return time.Time{}, Expression{}, false
	}
	loc := time.Local
	if job.Schedule.Timezone != "" {
		if loaded, err := time.LoadLocation(job.Schedule.Timezone); err == nil {
			loc = loaded
		} else {
			log.Warn("invalid job timezone", "jobId", job.ID, "error", err)
		}
	}
	return minute.In(loc), expression, true
}

func previous(expression Expression, now time.Time, window time.Duration) time.Time {
	current := now.Truncate(time.Minute)
	limit := current.Add(-window)
	for !current.Before(limit) {
		if expression.Matches(current) {
			return current
		}
		current = current.Add(-time.Minute)
	}
	return time.Time{}
}