package scheduler import ( "context" "fmt" "strings" "sync" "time" "gitea.mixdep.ru/mix/gosentry/src/domain" "gitea.mixdep.ru/mix/gosentry/src/storage" "gitea.mixdep.ru/mix/gosentry/src/runner" ) // Scheduler owns the timing loop for jobs that are currently loaded in the GUI. // It receives a pointer to the jobs slice because the GUI edits the same slice; // this keeps the early architecture simple while storage and scheduling are // still in one desktop process. type Scheduler struct { store *storage.Store jobs *[]domain.Job onChange func(domain.RunRecord) mu sync.Mutex ctx context.Context cancel context.CancelFunc paused bool schedules map[int]domain.Schedule // parsed once per job on load/edit } func NewScheduler(store *storage.Store, jobs *[]domain.Job, onChange func(domain.RunRecord)) *Scheduler { ctx, cancel := context.WithCancel(context.Background()) s := &Scheduler{ store: store, jobs: jobs, onChange: onChange, ctx: ctx, cancel: cancel, schedules: make(map[int]domain.Schedule), } s.resetNextRuns(time.Now()) return s } func (s *Scheduler) Start() { // A one-second ticker is accurate enough for cron-style desktop automation // and avoids the complexity of maintaining one timer per job. Five-field cron // expressions have minute precision, while @every values may be shorter for // testing and lightweight local tasks. ticker := time.NewTicker(time.Second) go func() { defer ticker.Stop() for { select { case <-s.ctx.Done(): return case now := <-ticker.C: s.tick(now) } } }() } func (s *Scheduler) Stop() { s.cancel() } func (s *Scheduler) SetPaused(paused bool) { s.mu.Lock() defer s.mu.Unlock() s.paused = paused now := time.Now() // Pause state is reflected into each job's display string so the list view is // understandable even before the next scheduler tick. for index := range *s.jobs { job := &(*s.jobs)[index] if !job.Enabled { job.NextRun = "Paused" continue } if paused { job.NextRun = "Scheduler paused" continue } s.prepareNextRun(job, now) } _ = s.store.SaveJobs(*s.jobs) } func (s *Scheduler) RunNow(index int) bool { s.mu.Lock() defer s.mu.Unlock() if index < 0 || index >= len(*s.jobs) { return false } // Manual runs share the same runner and log writer as scheduled runs. The // Trigger field is the only difference, which keeps History comparable and // prevents "Run now" from becoming a separate behavior path. return s.startRunLocked(index, "Manual") } func (s *Scheduler) RefreshSchedule(index int) { s.mu.Lock() defer s.mu.Unlock() if index < 0 || index >= len(*s.jobs) { return } job := &(*s.jobs)[index] s.parseJobSchedule(job) // re-parse in case the schedule string changed if !job.Enabled { job.NextRun = "Paused" return } if s.paused { job.NextRun = "Scheduler paused" return } s.prepareNextRun(job, time.Now()) } func (s *Scheduler) tick(now time.Time) { var changed bool s.mu.Lock() if !s.paused { for index := range *s.jobs { job := &(*s.jobs)[index] if !job.Enabled || job.NextDue.IsZero() || now.Before(job.NextDue) { continue } // Run only one due job per tick for now. That avoids overlapping shell // commands in the GUI process and keeps the first version predictable; // a future worker pool can add concurrency once cancellation and status // reporting are more explicit. changed = s.startRunLocked(index, "Schedule") break } } s.mu.Unlock() _ = changed } func (s *Scheduler) startRunLocked(index int, trigger string) bool { job := &(*s.jobs)[index] if job.LastState == "Running" { return false } jobCopy := *job job.LastState = "Running" job.NextRun = "Running" job.Output = runningOutput(jobCopy, trigger, time.Now()) job.NextDue = time.Time{} _ = s.store.SaveJobs(*s.jobs) go func() { record := runner.RunJob(s.ctx, &jobCopy, trigger, s.store.Paths.LogsDir) s.mu.Lock() if current := s.findJobByIDLocked(jobCopy.ID); current != nil { current.LastRun = record.Time current.LastState = record.State current.Output = record.Output current.Logs = append([]domain.RunRecord{record}, current.Logs...) if len(current.Logs) > 50 { current.Logs = current.Logs[:50] } s.prepareNextRun(current, time.Now()) _ = runner.CleanupLogs(s.store.Paths.LogsDir, s.store.Config.MaxLogFiles, s.store.Config.MaxLogAgeDays) _ = s.store.SaveJobs(*s.jobs) } s.mu.Unlock() if s.onChange != nil { s.onChange(record) } }() return true } func (s *Scheduler) findJobByIDLocked(id int) *domain.Job { for index := range *s.jobs { if (*s.jobs)[index].ID == id { return &(*s.jobs)[index] } } return nil } func runningOutput(job domain.Job, trigger string, started time.Time) string { var builder strings.Builder builder.WriteString("status:\n") builder.WriteString("Running since " + started.Format("2006-01-02 15:04:05") + "\n\n") builder.WriteString("trigger:\n") builder.WriteString(trigger + "\n\n") builder.WriteString("command:\n") builder.WriteString(job.Command + "\n\n") builder.WriteString("arguments:\n") builder.WriteString(runner.LogArguments(job.Arguments)) builder.WriteString("\n\nsuccess_exit_codes:\n") builder.WriteString(runner.SuccessExitCodesText(job)) builder.WriteString("\n\nstart_only:\n") builder.WriteString(fmt.Sprintf("%t", job.StartOnly)) return builder.String() } func (s *Scheduler) resetNextRuns(now time.Time) { for index := range *s.jobs { job := &(*s.jobs)[index] s.parseJobSchedule(job) // parse once on load if !job.Enabled { job.NextRun = "Paused" continue } s.prepareNextRun(job, now) } _ = s.store.SaveJobs(*s.jobs) } // parseJobSchedule caches a parsed domain.Schedule for the job. Invalid // schedule strings are silently dropped from the cache so prepareNextRun can // distinguish them from valid ones. func (s *Scheduler) parseJobSchedule(job *domain.Job) { sched, err := domain.Parse(job.Schedule) if err != nil { delete(s.schedules, job.ID) return } s.schedules[job.ID] = sched } func (s *Scheduler) prepareNextRun(job *domain.Job, from time.Time) { sched, ok := s.schedules[job.ID] if !ok { job.NextRun = "Invalid schedule" job.NextDue = time.Time{} return } job.NextDue = sched.Next(from) job.NextRun = job.NextDue.Format("2006-01-02 15:04:05") }