1package cron
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "sync"
8 "sync/atomic"
9 "time"
10
11 "github.com/bnema/gordon/internal/domain"
12 "github.com/bnema/zerowrap"
13)
14
15// Scheduler runs recurring jobs on preset schedules. Times are wall-clock
16// times in each schedule's Location; nowFn returns an absolute instant.
17type Scheduler struct {
18 entries map[string]*entry
19 mu sync.RWMutex
20 stopCh chan struct{}
21 log zerowrap.Logger
22 nowFn func() time.Time
23 maxJobs int
24 started atomic.Bool
25}
26
27// Option configures a Scheduler.
28type Option func(*Scheduler)
29
30var ErrJobRunning = errors.New("schedule is currently running")
31
32// WithMaxJobs sets the maximum number of jobs run concurrently.
33func WithMaxJobs(n int) Option {
34 return func(s *Scheduler) {
35 if n > 0 {
36 s.maxJobs = n
37 }
38 }
39}
40
41type entry struct {
42 id string
43 name string
44 schedule domain.CronSchedule
45 job func(ctx context.Context) error
46 lastRun time.Time
47 nextRun time.Time
48 running atomic.Bool
49}
50
51// NewScheduler creates a scheduler instance.
52func NewScheduler(log zerowrap.Logger, opts ...Option) *Scheduler {
53 s := &Scheduler{
54 entries: make(map[string]*entry),
55 stopCh: make(chan struct{}),
56 log: log,
57 maxJobs: 4,
58 nowFn: func() time.Time {
59 return time.Now().UTC()
60 },
61 }
62
63 for _, opt := range opts {
64 if opt != nil {
65 opt(s)
66 }
67 }
68
69 return s
70}
71
72// Add registers a new scheduled job.
73func (s *Scheduler) Add(id, name string, sched domain.CronSchedule, job func(ctx context.Context) error) error {
74 if id == "" {
75 return fmt.Errorf("id is required")
76 }
77 if job == nil {
78 return fmt.Errorf("job is required")
79 }
80
81 nextRun, err := calculateNextRun(s.nowFn(), sched)
82 if err != nil {
83 return err
84 }
85
86 s.mu.Lock()
87 defer s.mu.Unlock()
88
89 if _, exists := s.entries[id]; exists {
90 return fmt.Errorf("schedule %q already exists", id)
91 }
92
93 s.entries[id] = &entry{
94 id: id,
95 name: name,
96 schedule: sched,
97 job: job,
98 nextRun: nextRun,
99 }
100
101 return nil
102}
103
104// Remove unregisters a scheduled job.
105func (s *Scheduler) Remove(id string) error {
106 s.mu.Lock()
107 defer s.mu.Unlock()
108 e := s.entries[id]
109 if e == nil {
110 return nil
111 }
112 if e.running.Load() {
113 return fmt.Errorf("%w: %s", ErrJobRunning, id)
114 }
115 delete(s.entries, id)
116 return nil
117}
118
119// Start begins the scheduler loop.
120func (s *Scheduler) Start(ctx context.Context) {
121 if !s.started.CompareAndSwap(false, true) {
122 return
123 }
124
125 select {
126 case <-s.stopCh:
127 s.started.Store(false)
128 return
129 default:
130 }
131
132 select {
133 case <-ctx.Done():
134 s.started.Store(false)
135 return
136 default:
137 }
138
139 ticker := time.NewTicker(time.Minute)
140 go func() {
141 defer ticker.Stop()
142 s.runDue(ctx)
143 for {
144 select {
145 case <-ctx.Done():
146 return
147 case <-s.stopCh:
148 return
149 case <-ticker.C:
150 s.runDue(ctx)
151 }
152 }
153 }()
154}
155
156// Stop stops Scheduler processing.
157//
158// Stop is idempotent and safe to call multiple times. Because Stop closes
159// Scheduler.stopCh, the scheduler instance is not restartable after Stop;
160// create a new Scheduler to start processing again.
161func (s *Scheduler) Stop() {
162 select {
163 case <-s.stopCh:
164 return
165 default:
166 close(s.stopCh)
167 }
168}
169
170// List returns current scheduler entries.
171func (s *Scheduler) List() []domain.CronEntry {
172 s.mu.RLock()
173 defer s.mu.RUnlock()
174
175 entries := make([]domain.CronEntry, 0, len(s.entries))
176 for _, e := range s.entries {
177 entries = append(entries, domain.CronEntry{
178 ID: e.id,
179 Name: e.name,
180 Schedule: e.schedule,
181 LastRun: e.lastRun,
182 NextRun: e.nextRun,
183 Running: e.running.Load(),
184 })
185 }
186
187 return entries
188}
189
190// RunNow triggers a registered job immediately with the provided context.
191func (s *Scheduler) RunNow(ctx context.Context, id string) error {
192 e := s.getEntry(id)
193 if e == nil {
194 return fmt.Errorf("schedule %q not found", id)
195 }
196
197 return s.executeEntry(ctx, e)
198}
199
200func (s *Scheduler) runDue(ctx context.Context) {
201 now := s.nowFn()
202 entries := s.dueEntries(now)
203 sem := make(chan struct{}, s.maxJobs)
204 var wg sync.WaitGroup
205 for _, e := range entries {
206 wg.Add(1)
207 sem <- struct{}{}
208
209 e := e
210 go func() {
211 defer wg.Done()
212 defer func() { <-sem }()
213 if err := s.executeEntry(ctx, e); err != nil {
214 s.log.Warn().Err(err).Str("schedule_id", e.id).Msg("scheduled job failed")
215 }
216 }()
217 }
218 wg.Wait()
219}
220
221func (s *Scheduler) executeEntry(ctx context.Context, e *entry) error {
222 if !e.running.CompareAndSwap(false, true) {
223 return fmt.Errorf("schedule %q is already running", e.id)
224 }
225 defer e.running.Store(false)
226
227 now := s.nowFn()
228 var jobErr error
229 func() {
230 defer func() {
231 if rec := recover(); rec != nil {
232 jobErr = fmt.Errorf("job panic: %v", rec)
233 }
234 }()
235 jobErr = e.job(ctx)
236 }()
237
238 completedAt := s.nowFn()
239 if completedAt.Before(now) {
240 completedAt = now
241 }
242
243 nextRun, nextErr := calculateNextRun(completedAt, e.schedule)
244 if nextErr != nil {
245 if jobErr != nil {
246 return fmt.Errorf("job failed: %w; also failed to calculate next run: %v", jobErr, nextErr)
247 }
248 return nextErr
249 }
250
251 s.mu.Lock()
252 e.lastRun = now
253 e.nextRun = nextRun
254 s.mu.Unlock()
255
256 return jobErr
257}
258
259func (s *Scheduler) getEntry(id string) *entry {
260 s.mu.RLock()
261 defer s.mu.RUnlock()
262 return s.entries[id]
263}
264
265func (s *Scheduler) dueEntries(now time.Time) []*entry {
266 s.mu.RLock()
267 defer s.mu.RUnlock()
268
269 entries := make([]*entry, 0, len(s.entries))
270 for _, e := range s.entries {
271 if now.Before(e.nextRun) {
272 continue
273 }
274 entries = append(entries, e)
275 }
276 return entries
277}
278
279// presetDefaultTimes are the built-in times of day used when a schedule
280// sets no explicit time.
281var presetDefaultTimes = map[domain.BackupSchedule]domain.TimeOfDay{
282 domain.ScheduleHourly: {Hour: 0, Minute: 0},
283 domain.ScheduleDaily: {Hour: 2, Minute: 0},
284 domain.ScheduleWeekly: {Hour: 3, Minute: 0},
285 domain.ScheduleMonthly: {Hour: 4, Minute: 0},
286}
287
288// calculateNextRun returns the first scheduled instant strictly after now.
289//
290// Times are wall-clock times in schedule.Location (UTC when nil), so a
291// daily 03:30 stays 03:30 across daylight-saving changes. A wall time that
292// does not exist on a given day (spring-forward gap) runs at the end of the
293// gap. A wall time that happens twice (fall-back) runs once, at its first
294// occurrence; hourly schedules follow real elapsed hours.
295//
296// The result only ever depends on now: a run missed while the process was
297// down is skipped, never caught up.
298// - hourly: every hour at At.Minute
299// - daily: every day at At
300// - weekly: every Sunday at At
301// - monthly: the first day of every month at At
302func calculateNextRun(now time.Time, schedule domain.CronSchedule) (time.Time, error) {
303 at, ok := presetDefaultTimes[schedule.Preset]
304 if !ok {
305 return time.Time{}, fmt.Errorf("unsupported schedule preset: %q", schedule.Preset)
306 }
307 if schedule.At != nil {
308 at = *schedule.At
309 }
310 loc := schedule.Location
311 if loc == nil {
312 loc = time.UTC
313 }
314 local := now.In(loc)
315 y, m, d := local.Date()
316
317 switch schedule.Preset {
318 case domain.ScheduleHourly:
319 return nextMinuteOfHour(now, at.Minute, loc), nil
320 case domain.ScheduleDaily:
321 return firstAfter(now, func(i int) time.Time { return wallTime(y, m, d+i, at, loc) }), nil
322 case domain.ScheduleWeekly:
323 sunday := d + (7-int(local.Weekday()))%7
324 return firstAfter(now, func(i int) time.Time { return wallTime(y, m, sunday+7*i, at, loc) }), nil
325 default: // domain.ScheduleMonthly
326 return firstAfter(now, func(i int) time.Time { return wallTime(y, m+time.Month(i), 1, at, loc) }), nil
327 }
328}
329
330// firstAfter returns candidate(0) or candidate(1), whichever is the first
331// strictly after now.
332func firstAfter(now time.Time, candidate func(i int) time.Time) time.Time {
333 if next := candidate(0); next.After(now) {
334 return next
335 }
336 return candidate(1)
337}
338
339// nextMinuteOfHour returns the first whole minute strictly after now whose
340// wall-clock minute in loc equals minute. Walking real minutes keeps hourly
341// runs one hour apart across DST changes.
342func nextMinuteOfHour(now time.Time, minute int, loc *time.Location) time.Time {
343 t := now.Truncate(time.Minute).Add(time.Minute)
344 for i := 0; i < 120 && t.In(loc).Minute() != minute; i++ {
345 t = t.Add(time.Minute)
346 }
347 return t
348}
349
350// wallTime returns the instant of the given wall-clock time in loc. Day and
351// month overflow are normalized like time.Date. When the wall time falls in
352// a spring-forward gap, the end of the gap is returned.
353func wallTime(year int, month time.Month, day int, at domain.TimeOfDay, loc *time.Location) time.Time {
354 t := time.Date(year, month, day, at.Hour, at.Minute, 0, 0, loc)
355 if t.Hour() == at.Hour && t.Minute() == at.Minute {
356 return t
357 }
358 // Nonexistent wall time: interpret it with the offset in force before
359 // the gap, which lands after the transition, then snap back to it.
360 naive := time.Date(year, month, day, at.Hour, at.Minute, 0, 0, time.UTC)
361 _, offsetBefore := naive.Add(-12 * time.Hour).In(loc).Zone()
362 start, _ := naive.Add(-time.Duration(offsetBefore) * time.Second).In(loc).ZoneBounds()
363 return start.In(loc)
364}