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 based on backup schedules.
16type Scheduler struct {
17 entries map[string]*entry
18 mu sync.RWMutex
19 stopCh chan struct{}
20 log zerowrap.Logger
21 nowFn func() time.Time
22 maxJobs int
23 started atomic.Bool
24}
25
26// Option configures a Scheduler.
27type Option func(*Scheduler)
28
29var ErrJobRunning = errors.New("schedule is currently running")
30
31// WithMaxJobs sets the maximum number of jobs run concurrently.
32func WithMaxJobs(n int) Option {
33 return func(s *Scheduler) {
34 if n > 0 {
35 s.maxJobs = n
36 }
37 }
38}
39
40type entry struct {
41 id string
42 name string
43 schedule domain.CronSchedule
44 job func(ctx context.Context) error
45 lastRun time.Time
46 nextRun time.Time
47 running atomic.Bool
48}
49
50// NewScheduler creates a scheduler instance.
51func NewScheduler(log zerowrap.Logger, opts ...Option) *Scheduler {
52 s := &Scheduler{
53 entries: make(map[string]*entry),
54 stopCh: make(chan struct{}),
55 log: log,
56 maxJobs: 4,
57 nowFn: func() time.Time {
58 return time.Now().UTC()
59 },
60 }
61
62 for _, opt := range opts {
63 if opt != nil {
64 opt(s)
65 }
66 }
67
68 return s
69}
70
71// Add registers a new scheduled job.
72func (s *Scheduler) Add(id, name string, sched domain.CronSchedule, job func(ctx context.Context) error) error {
73 if id == "" {
74 return fmt.Errorf("id is required")
75 }
76 if job == nil {
77 return fmt.Errorf("job is required")
78 }
79
80 nextRun, err := calculateNextRun(s.nowFn(), sched)
81 if err != nil {
82 return err
83 }
84
85 s.mu.Lock()
86 defer s.mu.Unlock()
87
88 if _, exists := s.entries[id]; exists {
89 return fmt.Errorf("schedule %q already exists", id)
90 }
91
92 s.entries[id] = &entry{
93 id: id,
94 name: name,
95 schedule: sched,
96 job: job,
97 nextRun: nextRun,
98 }
99
100 return nil
101}
102
103// Remove unregisters a scheduled job.
104func (s *Scheduler) Remove(id string) error {
105 s.mu.Lock()
106 defer s.mu.Unlock()
107 e := s.entries[id]
108 if e == nil {
109 return nil
110 }
111 if e.running.Load() {
112 return fmt.Errorf("%w: %s", ErrJobRunning, id)
113 }
114 delete(s.entries, id)
115 return nil
116}
117
118// Start begins the scheduler loop.
119func (s *Scheduler) Start(ctx context.Context) {
120 if !s.started.CompareAndSwap(false, true) {
121 return
122 }
123
124 select {
125 case <-s.stopCh:
126 s.started.Store(false)
127 return
128 default:
129 }
130
131 select {
132 case <-ctx.Done():
133 s.started.Store(false)
134 return
135 default:
136 }
137
138 ticker := time.NewTicker(time.Minute)
139 go func() {
140 defer ticker.Stop()
141 s.runDue(ctx)
142 for {
143 select {
144 case <-ctx.Done():
145 return
146 case <-s.stopCh:
147 return
148 case <-ticker.C:
149 s.runDue(ctx)
150 }
151 }
152 }()
153}
154
155// Stop stops Scheduler processing.
156//
157// Stop is idempotent and safe to call multiple times. Because Stop closes
158// Scheduler.stopCh, the scheduler instance is not restartable after Stop;
159// create a new Scheduler to start processing again.
160func (s *Scheduler) Stop() {
161 select {
162 case <-s.stopCh:
163 return
164 default:
165 close(s.stopCh)
166 }
167}
168
169// List returns current scheduler entries.
170func (s *Scheduler) List() []domain.CronEntry {
171 s.mu.RLock()
172 defer s.mu.RUnlock()
173
174 entries := make([]domain.CronEntry, 0, len(s.entries))
175 for _, e := range s.entries {
176 entries = append(entries, domain.CronEntry{
177 ID: e.id,
178 Name: e.name,
179 Schedule: e.schedule,
180 LastRun: e.lastRun,
181 NextRun: e.nextRun,
182 Running: e.running.Load(),
183 })
184 }
185
186 return entries
187}
188
189// RunNow triggers a registered job immediately with the provided context.
190func (s *Scheduler) RunNow(ctx context.Context, id string) error {
191 e := s.getEntry(id)
192 if e == nil {
193 return fmt.Errorf("schedule %q not found", id)
194 }
195
196 return s.executeEntry(ctx, e)
197}
198
199func (s *Scheduler) runDue(ctx context.Context) {
200 now := s.nowFn()
201 entries := s.dueEntries(now)
202 sem := make(chan struct{}, s.maxJobs)
203 var wg sync.WaitGroup
204 for _, e := range entries {
205 wg.Add(1)
206 sem <- struct{}{}
207
208 e := e
209 go func() {
210 defer wg.Done()
211 defer func() { <-sem }()
212 if err := s.executeEntry(ctx, e); err != nil {
213 s.log.Warn().Err(err).Str("schedule_id", e.id).Msg("scheduled job failed")
214 }
215 }()
216 }
217 wg.Wait()
218}
219
220func (s *Scheduler) executeEntry(ctx context.Context, e *entry) error {
221 if !e.running.CompareAndSwap(false, true) {
222 return fmt.Errorf("schedule %q is already running", e.id)
223 }
224 defer e.running.Store(false)
225
226 now := s.nowFn()
227 var jobErr error
228 func() {
229 defer func() {
230 if rec := recover(); rec != nil {
231 jobErr = fmt.Errorf("job panic: %v", rec)
232 }
233 }()
234 jobErr = e.job(ctx)
235 }()
236
237 completedAt := s.nowFn()
238 if completedAt.Before(now) {
239 completedAt = now
240 }
241
242 nextRun, nextErr := calculateNextRun(completedAt, e.schedule)
243 if nextErr != nil {
244 if jobErr != nil {
245 return fmt.Errorf("job failed: %w; also failed to calculate next run: %v", jobErr, nextErr)
246 }
247 return nextErr
248 }
249
250 s.mu.Lock()
251 e.lastRun = now
252 e.nextRun = nextRun
253 s.mu.Unlock()
254
255 return jobErr
256}
257
258func (s *Scheduler) getEntry(id string) *entry {
259 s.mu.RLock()
260 defer s.mu.RUnlock()
261 return s.entries[id]
262}
263
264func (s *Scheduler) dueEntries(now time.Time) []*entry {
265 s.mu.RLock()
266 defer s.mu.RUnlock()
267
268 entries := make([]*entry, 0, len(s.entries))
269 for _, e := range s.entries {
270 if now.Before(e.nextRun) {
271 continue
272 }
273 entries = append(entries, e)
274 }
275 return entries
276}
277
278// calculateNextRun computes the next execution time using UTC-only preset rules.
279//
280// Current MVP behavior is intentionally fixed to UTC because domain.CronSchedule
281// has no timezone field yet:
282// - daily: 02:00 UTC
283// - weekly: Sunday 03:00 UTC
284// - monthly: first day of month 04:00 UTC
285//
286// This means users in non-UTC timezones should account for UTC offsets.
287//
288// TODO: add timezone support on domain.CronSchedule once timezone-aware
289// scheduling is introduced. Default timezone should be the host system timezone
290// when no explicit timezone is configured.
291func calculateNextRun(now time.Time, schedule domain.CronSchedule) (time.Time, error) {
292 now = now.UTC()
293
294 switch schedule.Preset {
295 case domain.ScheduleHourly:
296 next := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, time.UTC)
297 if !next.After(now) {
298 next = next.Add(time.Hour)
299 }
300 return next, nil
301 case domain.ScheduleDaily:
302 next := time.Date(now.Year(), now.Month(), now.Day(), 2, 0, 0, 0, time.UTC)
303 if !next.After(now) {
304 next = next.AddDate(0, 0, 1)
305 }
306 return next, nil
307 case domain.ScheduleWeekly:
308 daysUntilSunday := (7 - int(now.Weekday())) % 7
309 next := time.Date(now.Year(), now.Month(), now.Day(), 3, 0, 0, 0, time.UTC).AddDate(0, 0, daysUntilSunday)
310 if !next.After(now) {
311 next = next.AddDate(0, 0, 7)
312 }
313 return next, nil
314 case domain.ScheduleMonthly:
315 next := time.Date(now.Year(), now.Month(), 1, 4, 0, 0, 0, time.UTC)
316 if !next.After(now) {
317 next = next.AddDate(0, 1, 0)
318 }
319 return next, nil
320 default:
321 return time.Time{}, fmt.Errorf("unsupported schedule preset: %q", schedule.Preset)
322 }
323}