package maintenance import ( "context" "evobgp/internal/logging" "fmt" "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 { logging.Default().Info(fmt.Sprintf("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() } } }() logging.Default().Info(fmt.Sprintf("maintenance: policy scheduler started (tick=%s)", tick)) }