agentic-coding-workflow/internal/jobs/scheduler.go

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