package jobs import ( "context" "log/slog" "sync" "time" ) // Scheduler drives a set of named periodic runners. Each runner is a closure // invoked at a fixed interval; panics are recovered and logged. type Scheduler struct { log *slog.Logger entries []schedulerEntry wg sync.WaitGroup } type schedulerEntry struct { name string interval time.Duration enabled bool fn func(context.Context) } // NewScheduler builds a Scheduler. Add entries before calling Start. func NewScheduler(log *slog.Logger) *Scheduler { return &Scheduler{log: log} } // Add registers a periodic runner. enabled=false skips it at Start. func (s *Scheduler) Add(name string, interval time.Duration, enabled bool, fn func(context.Context)) { s.entries = append(s.entries, schedulerEntry{name: name, interval: interval, enabled: enabled, fn: fn}) } // Start launches one goroutine per enabled entry. Each goroutine exits when // ctx is canceled. Wait blocks until all goroutines exit. func (s *Scheduler) Start(ctx context.Context) { for _, e := range s.entries { if !e.enabled || e.interval <= 0 { continue } s.wg.Add(1) go s.run(ctx, e) } } // Wait blocks until Start's goroutines exit (typically after ctx cancel). func (s *Scheduler) Wait() { s.wg.Wait() } func (s *Scheduler) run(ctx context.Context, e schedulerEntry) { defer s.wg.Done() t := time.NewTicker(e.interval) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: s.invoke(ctx, e) } } } func (s *Scheduler) invoke(ctx context.Context, e schedulerEntry) { defer func() { if r := recover(); r != nil { s.log.Error("runner.panic", "name", e.name, "panic", r) } }() start := time.Now() e.fn(ctx) s.log.Info("runner.tick", "name", e.name, "duration_ms", time.Since(start).Milliseconds(), ) }