EvoBGP/internal/maintenance/scheduler.go

package maintenance

import (
	"context"
	"fmt"
	"log"
	"strings"
	"sync"
	"time"

	"github.com/robfig/cron/v3"
)

// StartScheduler enqueues maintenance_policy_run jobs when cron schedules match.
func StartScheduler(ctx context.Context, provider *ConfigProvider, enqueue func(policyID string, dryRun bool, idempotencyKey string), tick time.Duration) {
	if provider == nil || enqueue == nil {
		return
	}
	if tick <= 0 {
		tick = 30 * time.Second
	}
	go func() {
		parser := cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow)
		var mu sync.Mutex
		schedules := map[string]cron.Schedule{}
		lastFired := map[string]time.Time{}

		rebuild := func() {
			mu.Lock()
			defer mu.Unlock()
			schedules = map[string]cron.Schedule{}
			for _, p := range provider.Snapshot() {
				if p == nil || !p.Enabled || strings.TrimSpace(p.Schedule) == "" {
					continue
				}
				sched, err := parser.Parse(p.Schedule)
				if err != nil {
					log.Printf("maintenance: invalid cron for policy %s: %v", p.ID, err)
					continue
				}
				schedules[p.ID] = sched
			}
		}

		rebuild()
		t := time.NewTicker(tick)
		defer t.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-t.C:
				rebuild()
				now := time.Now().UTC()
				mu.Lock()
				for _, p := range provider.Snapshot() {
					if p == nil || !p.Enabled {
						continue
					}
					sched, ok := schedules[p.ID]
					if !ok {
						continue
					}
					prev := lastFired[p.ID]
					if prev.IsZero() {
						prev = now.Add(-time.Minute)
					}
					next := sched.Next(prev)
					if next.After(now) {
						continue
					}
					slot := next.Unix() / 60
					if lf, ok := lastFired[p.ID]; ok && lf.Unix()/60 == slot {
						continue
					}
					lastFired[p.ID] = next
					idem := fmt.Sprintf("maint-%s-%d", p.ID, slot)
					enqueue(p.ID, p.DryRunEnabled, idem)
				}
				mu.Unlock()
			}
		}
	}()
	log.Printf("maintenance: policy scheduler started (tick=%s)", tick)
}