package main import ( "cmp" "context" "encoding/json" "errors" "fmt" "log/slog" "os" "slices" "strconv" "strings" "sync" "time" ) const ( // scheduleFileName is the file of the schedules in the directory of the // agent. scheduleFileName = "schedules.json" // scheduleTextBytes is the most bytes of the text of a schedule. scheduleTextBytes = 4096 // scheduleAtLayout is the layout of the time of a one-time schedule. scheduleAtLayout = "2006-01-02 15:04" // scheduleTimeLayout is the layout of the times in the answers of the // tools and in the log. scheduleTimeLayout = "Monday 2006-01-02 15:04 -07:00" // scheduleSource is the source of the work events of the runs. scheduleSource = "schedule" // scheduleCheck is the longest wait of Run before it looks at the times // again, so that it sees a suspend of the machine or a change of the // clock. scheduleCheck = time.Minute // scheduleLate is how long after its time a run is late. scheduleLate = time.Minute // scheduleEmptyReply is the last line of the message of a run. scheduleEmptyReply = "If nothing in this task needs the user, end your turn with an empty reply." // scheduleLateLine tells the agent that a run is late. scheduleLateLine = "This run is late: the daemon was stopped at that time." ) // openSchedules reads the schedules of the file at path. A missing file holds // none. The owner names its agent. Schedules without an owner take this name. A // different stored owner is an error. The clock gives the time and the time // zone of the schedules, and logger gets one line for each run. func openSchedules(path, owner string, clock func() time.Time, logger *slog.Logger) (*Schedules, error) { s := &Schedules{ path: path, owner: owner, clock: clock, logger: logger, check: scheduleCheck, changed: make(chan struct{}, 1), } data, err := os.ReadFile(path) if errors.Is(err, os.ErrNotExist) { return s, nil } if err != nil { return nil, fmt.Errorf("read schedules: %w", err) } err = json.Unmarshal(data, &s.file) if err != nil { return nil, fmt.Errorf("%s: %w", path, err) } changed := false for i := range s.file.Schedules { schedule := &s.file.Schedules[i] if schedule.Owner == "" { schedule.Owner = owner changed = true } if schedule.Owner != owner { return nil, fmt.Errorf("schedule %d belongs to %q, not %q", schedule.ID, schedule.Owner, owner) } if schedule.Cron == "" { continue } _, err = scheduleParseCron(schedule.Cron) if err != nil { return nil, fmt.Errorf("%s: schedule %d: %w", path, schedule.ID, err) } } if changed { err = s.save() if err != nil { return nil, fmt.Errorf("record schedule owner: %w", err) } } return s, nil } // Schedules holds the schedules of the agent: tasks that start a turn at one // time or at the times of a cron expression. Each change writes the file whole // before it returns, so a schedule survives a restart. The mutex protects the // schedules, because the tools change them from a turn while Run reads them. // changed tells Run that a set or a remove changed the times. type Schedules struct { path string owner string clock func() time.Time logger *slog.Logger check time.Duration changed chan struct{} mutex sync.Mutex file ScheduleFile } // Run starts the runs of the due schedules in l until ctx ends. It waits until // the first due time, at most check at a time, and also wakes at a set or a // remove. So a schedule that came due while the daemon was stopped runs at the // start. It returns the error of a write of the log or of the file. func (s *Schedules) Run(ctx context.Context, l *TurnLoop) error { for { err := s.deliver(ctx, l) if err != nil { return err } var ( timer *time.Timer timeout <-chan time.Time ) wait, found := s.Wait(s.clock()) if found { timer = time.NewTimer(min(wait, s.check)) timeout = timer.C } select { case <-ctx.Done(): return nil case <-s.changed: case <-timeout: } if timer != nil { timer.Stop() } } } // Set adds a schedule with text and returns the answer of the tool. Exactly one // of at and cron is not nil. at is a local time in scheduleAtLayout after now, // and it goes through scheduleWallTime, so a time in the hour that the clock // repeats is its first occurrence. cron is an expression of scheduleParseCron. func (s *Schedules) Set(text string, at, cron *string) (string, error) { text = strings.TrimSpace(text) if text == "" || len(text) > scheduleTextBytes { return "", fmt.Errorf("the text must have 1 to %d bytes", scheduleTextBytes) } now := s.clock() schedule := Schedule{ Text: text, Owner: s.owner, } switch { case (at == nil) == (cron == nil): return "", errors.New("give either at or cron, and null for the other") case at != nil: parsed, err := time.Parse(scheduleAtLayout, *at) if err != nil { return "", fmt.Errorf("at must be a local time such as %s", now.Format(scheduleAtLayout)) } schedule.Due = scheduleWallTime(parsed.Year(), parsed.Month(), parsed.Day(), parsed.Hour(), parsed.Minute(), now.Location()) if !schedule.Due.After(now) { return "", fmt.Errorf("the time %s is not after now, %s", schedule.Due.Format(scheduleTimeLayout), now.Format(scheduleTimeLayout)) } default: parsed, err := scheduleParseCron(*cron) if err != nil { return "", err } next, found := parsed.Next(now) if !found { return "", fmt.Errorf("the cron expression %q matches no time in %d years", *cron, scheduleCronYears) } schedule.Cron = parsed.String() schedule.Due = next } s.mutex.Lock() defer s.mutex.Unlock() s.file.Next++ schedule.ID = s.file.Next s.file.Schedules = append(s.file.Schedules, schedule) err := s.save() if err != nil { return "", err } s.change() return fmt.Sprintf("Schedule %d set. Next: %s.", schedule.ID, schedule.Due.In(now.Location()).Format(scheduleTimeLayout)), nil } // List returns the answer of the tool: one line for each schedule, by due time, // with its ID, its next time, its cron expression or once, and its text on one // line. func (s *Schedules) List() string { location := s.clock().Location() schedules := s.sorted() if len(schedules) == 0 { return "No schedules." } lines := make([]string, 0, len(schedules)) for _, schedule := range schedules { lines = append(lines, strconv.Itoa(schedule.ID)+" | "+schedule.Due.In(location).Format(scheduleTimeLayout)+" | "+cmp.Or(schedule.Cron, "once")+" | "+strings.Join(strings.Fields(schedule.Text), " ")) } return strings.Join(lines, "\n") } // Len returns the number of schedules. func (s *Schedules) Len() int { s.mutex.Lock() defer s.mutex.Unlock() return len(s.file.Schedules) } // Remove removes the schedule id and returns the answer of the tool. func (s *Schedules) Remove(id int) (string, error) { s.mutex.Lock() defer s.mutex.Unlock() index := slices.IndexFunc(s.file.Schedules, func(schedule Schedule) bool { return schedule.ID == id }) if index < 0 { return "", fmt.Errorf("no schedule %d", id) } s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1) err := s.save() if err != nil { return "", err } s.change() return fmt.Sprintf("Schedule %d removed.", id), nil } // deliver logs the message of each due schedule in l, in due order, as a work // message of scheduleSource, and wakes the loop once with all of them. A // schedule ends its run with Done after its message is durable in the log. A // crash between the two makes the run repeat once. Each run logs one line with // the ID of the schedule and, for a late run, how late it is. The line holds no // text of the schedule. func (s *Schedules) deliver(ctx context.Context, l *TurnLoop) error { if l.agent.config.Name != s.owner { return fmt.Errorf("schedule owner %q does not match agent %q", s.owner, l.agent.config.Name) } now := s.clock() var turns []Message for _, schedule := range s.Due(now) { message, err := l.Work(scheduleSource, schedule.Message(now), "") if err != nil { return fmt.Errorf("log a schedule: %w", err) } // A stop of the daemon during the flush is no failure: the // schedule stays due and runs at the next start. err = l.Flush(ctx) if ctx.Err() != nil { return nil } if err != nil { return fmt.Errorf("log a schedule: %w", err) } err = s.Done(schedule, now) if err != nil { return err } attrs := []any{"id", schedule.ID} if schedule.late(now) { attrs = append(attrs, "late", now.Sub(schedule.Due)) } s.logger.Info("schedule ran", attrs...) turns = append(turns, message) } l.Wake(turns...) return nil } // Due returns the schedules whose time is not after now, by due time. func (s *Schedules) Due(now time.Time) []Schedule { var due []Schedule for _, schedule := range s.sorted() { if !schedule.Due.After(now) { due = append(due, schedule) } } return due } // Done ends the run of schedule at now: a one-time schedule goes, and a // repeating schedule gets the next time after now. So a repeating schedule that // missed several times runs once. When a remove took the schedule first, Done // changes nothing. func (s *Schedules) Done(schedule Schedule, now time.Time) error { s.mutex.Lock() defer s.mutex.Unlock() index := slices.IndexFunc(s.file.Schedules, func(held Schedule) bool { return held.ID == schedule.ID }) if index < 0 { return nil } if schedule.Cron == "" { s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1) return s.save() } cron, err := scheduleParseCron(schedule.Cron) if err != nil { return fmt.Errorf("schedule %d: %w", schedule.ID, err) } next, found := cron.Next(now) if !found { s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1) return s.save() } s.file.Schedules[index].Due = next return s.save() } // Wait returns how long after now the first schedule is due, or false when // there is no schedule. func (s *Schedules) Wait(now time.Time) (time.Duration, bool) { schedules := s.sorted() if len(schedules) == 0 { return 0, false } return max(schedules[0].Due.Sub(now), 0), true } // change wakes Run after a set or a remove. func (s *Schedules) change() { select { case s.changed <- struct{}{}: default: } } // sorted returns a copy of the schedules by due time, then by ID. func (s *Schedules) sorted() []Schedule { s.mutex.Lock() schedules := slices.Clone(s.file.Schedules) s.mutex.Unlock() slices.SortFunc(schedules, func(a, b Schedule) int { return cmp.Or(a.Due.Compare(b.Due), cmp.Compare(a.ID, b.ID)) }) return schedules } // save writes the file whole, with mode 0600. The caller holds the mutex. func (s *Schedules) save() error { data, err := json.Marshal(s.file) if err == nil { err = writeFileAtomic(s.path, append(data, '\n')) } if err != nil { return fmt.Errorf("save schedules: %w", err) } return nil } // ScheduleFile is the content of the file of the schedules: the ID of the last // schedule, so that an ID is never used again, and the schedules. type ScheduleFile struct { Next int `json:"next"` Schedules []Schedule `json:"schedules"` } // Schedule is one schedule: its owner, its ID, the task for the agent, the cron // expression of a repeating schedule, empty for a one-time schedule, and its // next time. type Schedule struct { Owner string `json:"owner"` ID int `json:"id"` Text string `json:"text"` Cron string `json:"cron"` Due time.Time `json:"due"` } // Message returns the text of a run at now for the log, with the times in the // location of now. The first line starts with [Schedule ID] and gives the due // time and once or the cron expression. Then come the text, the line about a // late run when now is more than scheduleLate after the due time, and the line // about the empty reply. func (s Schedule) Message(now time.Time) string { lines := []string{ "[Schedule " + strconv.Itoa(s.ID) + "] " + s.Due.In(now.Location()).Format(scheduleTimeLayout) + ", " + s.repeats(), s.Text, } if s.late(now) { lines = append(lines, scheduleLateLine) } return strings.Join(append(lines, scheduleEmptyReply), "\n") } // late tells if a run at now is late: more than scheduleLate after the due // time. func (s Schedule) late(now time.Time) bool { return now.Sub(s.Due) > scheduleLate } // repeats returns once for a one-time schedule, or else repeats and the cron // expression. func (s Schedule) repeats() string { if s.Cron == "" { return "once" } return "repeats " + s.Cron }