git.giacomolli.uk - onechat/blob - schedule.go

   1 package main
   2
   3 import (
   4         "cmp"
   5         "context"
   6         "encoding/json"
   7         "errors"
   8         "fmt"
   9         "os"
  10         "slices"
  11         "strconv"
  12         "strings"
  13         "sync"
  14         "time"
  15 )
  16
  17 const (
  18         // scheduleFileName is the file of the schedules in the directory of the
  19         // agent.
  20         scheduleFileName = "schedules.json"
  21
  22         // scheduleTextBytes is the most bytes of the text of a schedule.
  23         scheduleTextBytes = 4096
  24
  25         // scheduleAtLayout is the layout of the time of a one-time schedule.
  26         scheduleAtLayout = "2006-01-02 15:04"
  27
  28         // scheduleTimeLayout is the layout of the times in the answers of the
  29         // tools and in the log.
  30         scheduleTimeLayout = "Monday 2006-01-02 15:04 -07:00"
  31
  32         // scheduleSource is the source of the work events of the runs.
  33         scheduleSource = "schedule"
  34
  35         // scheduleCheck is the longest wait of Run before it looks at the times
  36         // again, so that it sees a suspend of the machine or a change of the
  37         // clock.
  38         scheduleCheck = time.Minute
  39
  40         // scheduleLate is how long after its time a run is late.
  41         scheduleLate = time.Minute
  42
  43         // scheduleEmptyReply is the last line of the message of a run.
  44         scheduleEmptyReply = "If nothing in this task needs the user, end your turn with an empty reply."
  45
  46         // scheduleLateLine tells the agent that a run is late.
  47         scheduleLateLine = "This run is late: the daemon was stopped at that time."
  48 )
  49
  50 // openSchedules reads the schedules of the file at path. A missing file holds
  51 // none. The clock gives the time and the time zone of the schedules.
  52 func openSchedules(path string, clock func() time.Time) (*Schedules, error) {
  53         s := &Schedules{
  54                 path:    path,
  55                 clock:   clock,
  56                 check:   scheduleCheck,
  57                 changed: make(chan struct{}, 1),
  58         }
  59
  60         data, err := os.ReadFile(path)
  61         if errors.Is(err, os.ErrNotExist) {
  62                 return s, nil
  63         }
  64
  65         if err != nil {
  66                 return nil, fmt.Errorf("read schedules: %w", err)
  67         }
  68
  69         err = json.Unmarshal(data, &s.file)
  70         if err != nil {
  71                 return nil, fmt.Errorf("%s: %w", path, err)
  72         }
  73
  74         for _, schedule := range s.file.Schedules {
  75                 if schedule.Cron == "" {
  76                         continue
  77                 }
  78
  79                 _, err = scheduleParseCron(schedule.Cron)
  80                 if err != nil {
  81                         return nil, fmt.Errorf("%s: schedule %d: %w", path, schedule.ID, err)
  82                 }
  83         }
  84
  85         return s, nil
  86 }
  87
  88 // Schedules holds the schedules of the agent: tasks that start a turn at one
  89 // time or at the times of a cron expression. Each change writes the file whole
  90 // before it returns, so a schedule survives a restart. The mutex protects the
  91 // schedules, because the tools change them from a turn while Run reads them.
  92 // changed tells Run that a set or a remove changed the times.
  93 type Schedules struct {
  94         path    string
  95         clock   func() time.Time
  96         check   time.Duration
  97         changed chan struct{}
  98
  99         mutex sync.Mutex
 100         file  ScheduleFile
 101 }
 102
 103 // Run starts the runs of the due schedules in l until ctx ends. It waits until
 104 // the first due time, at most check at a time, and also wakes at a set or a
 105 // remove. So a schedule that came due while the daemon was stopped runs at the
 106 // start. It returns the error of a write of the log or of the file.
 107 func (s *Schedules) Run(ctx context.Context, l *TurnLoop) error {
 108         for {
 109                 err := s.deliver(ctx, l)
 110                 if err != nil {
 111                         return err
 112                 }
 113
 114                 var timer *time.Timer
 115                 var timeout <-chan time.Time
 116                 wait, found := s.Wait(s.clock())
 117                 if found {
 118                         timer = time.NewTimer(min(wait, s.check))
 119                         timeout = timer.C
 120                 }
 121
 122                 select {
 123                 case <-ctx.Done():
 124                         return nil
 125                 case <-s.changed:
 126                 case <-timeout:
 127                 }
 128
 129                 if timer != nil {
 130                         timer.Stop()
 131                 }
 132         }
 133 }
 134
 135 // Set adds a schedule with text and returns the answer of the tool. Exactly one
 136 // of at and cron is not nil. at is a local time in scheduleAtLayout after now,
 137 // and it goes through scheduleWallTime, so a time in the hour that the clock
 138 // repeats is its first occurrence. cron is an expression of scheduleParseCron.
 139 func (s *Schedules) Set(text string, at, cron *string) (string, error) {
 140         text = strings.TrimSpace(text)
 141         if text == "" || len(text) > scheduleTextBytes {
 142                 return "", fmt.Errorf("the text must have 1 to %d bytes", scheduleTextBytes)
 143         }
 144
 145         now := s.clock()
 146         schedule := Schedule{
 147                 Text: text,
 148         }
 149
 150         switch {
 151         case (at == nil) == (cron == nil):
 152                 return "", errors.New("give either at or cron, and null for the other")
 153         case at != nil:
 154                 parsed, err := time.Parse(scheduleAtLayout, *at)
 155                 if err != nil {
 156                         return "", fmt.Errorf("at must be a local time such as %s", now.Format(scheduleAtLayout))
 157                 }
 158
 159                 schedule.Due = scheduleWallTime(parsed.Year(), parsed.Month(), parsed.Day(), parsed.Hour(), parsed.Minute(), now.Location())
 160                 if !schedule.Due.After(now) {
 161                         return "", fmt.Errorf("the time %s is not after now, %s", schedule.Due.Format(scheduleTimeLayout), now.Format(scheduleTimeLayout))
 162                 }
 163         default:
 164                 parsed, err := scheduleParseCron(*cron)
 165                 if err != nil {
 166                         return "", err
 167                 }
 168
 169                 next, found := parsed.Next(now)
 170                 if !found {
 171                         return "", fmt.Errorf("the cron expression %q matches no time in %d years", *cron, scheduleCronYears)
 172                 }
 173
 174                 schedule.Cron = parsed.String()
 175                 schedule.Due = next
 176         }
 177
 178         s.mutex.Lock()
 179         defer s.mutex.Unlock()
 180
 181         s.file.Next++
 182         schedule.ID = s.file.Next
 183         s.file.Schedules = append(s.file.Schedules, schedule)
 184         err := s.save()
 185         if err != nil {
 186                 return "", err
 187         }
 188
 189         s.change()
 190
 191         return fmt.Sprintf("Schedule %d set. Next: %s.", schedule.ID, schedule.Due.In(now.Location()).Format(scheduleTimeLayout)), nil
 192 }
 193
 194 // List returns the answer of the tool: one line for each schedule, by due time,
 195 // with its ID, its next time, its cron expression or once, and its text on one
 196 // line.
 197 func (s *Schedules) List() string {
 198         location := s.clock().Location()
 199         schedules := s.sorted()
 200         if len(schedules) == 0 {
 201                 return "No schedules."
 202         }
 203
 204         lines := make([]string, 0, len(schedules))
 205         for _, schedule := range schedules {
 206                 lines = append(lines, strconv.Itoa(schedule.ID)+" | "+schedule.Due.In(location).Format(scheduleTimeLayout)+" | "+cmp.Or(schedule.Cron, "once")+" | "+strings.Join(strings.Fields(schedule.Text), " "))
 207         }
 208
 209         return strings.Join(lines, "\n")
 210 }
 211
 212 // Len returns the number of schedules.
 213 func (s *Schedules) Len() int {
 214         s.mutex.Lock()
 215         defer s.mutex.Unlock()
 216
 217         return len(s.file.Schedules)
 218 }
 219
 220 // Remove removes the schedule id and returns the answer of the tool.
 221 func (s *Schedules) Remove(id int) (string, error) {
 222         s.mutex.Lock()
 223         defer s.mutex.Unlock()
 224
 225         index := slices.IndexFunc(s.file.Schedules, func(schedule Schedule) bool {
 226                 return schedule.ID == id
 227         })
 228
 229         if index < 0 {
 230                 return "", fmt.Errorf("no schedule %d", id)
 231         }
 232
 233         s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1)
 234         err := s.save()
 235         if err != nil {
 236                 return "", err
 237         }
 238
 239         s.change()
 240
 241         return fmt.Sprintf("Schedule %d removed.", id), nil
 242 }
 243
 244 // deliver logs the message of each due schedule in l, in due order, as a work
 245 // message of scheduleSource, and wakes the loop once with all of them. A
 246 // schedule ends its run with Done after its message is durable in the log. A
 247 // crash between the two makes the run repeat once.
 248 func (s *Schedules) deliver(ctx context.Context, l *TurnLoop) error {
 249         now := s.clock()
 250         var turns []Message
 251         for _, schedule := range s.Due(now) {
 252                 message, err := l.Work(scheduleSource, schedule.Message(now), "")
 253                 if err != nil {
 254                         return fmt.Errorf("log a schedule: %w", err)
 255                 }
 256
 257                 // A stop of the daemon during the flush is no failure: the
 258                 // schedule stays due and runs at the next start.
 259                 err = l.Flush(ctx)
 260                 if ctx.Err() != nil {
 261                         return nil
 262                 }
 263
 264                 if err != nil {
 265                         return fmt.Errorf("log a schedule: %w", err)
 266                 }
 267
 268                 err = s.Done(schedule, now)
 269                 if err != nil {
 270                         return err
 271                 }
 272
 273                 turns = append(turns, message)
 274         }
 275
 276         l.Wake(turns...)
 277
 278         return nil
 279 }
 280
 281 // Due returns the schedules whose time is not after now, by due time.
 282 func (s *Schedules) Due(now time.Time) []Schedule {
 283         var due []Schedule
 284         for _, schedule := range s.sorted() {
 285                 if !schedule.Due.After(now) {
 286                         due = append(due, schedule)
 287                 }
 288         }
 289
 290         return due
 291 }
 292
 293 // Done ends the run of schedule at now: a one-time schedule goes, and a
 294 // repeating schedule gets the next time after now. So a repeating schedule that
 295 // missed several times runs once. When a remove took the schedule first, Done
 296 // changes nothing.
 297 func (s *Schedules) Done(schedule Schedule, now time.Time) error {
 298         s.mutex.Lock()
 299         defer s.mutex.Unlock()
 300
 301         index := slices.IndexFunc(s.file.Schedules, func(held Schedule) bool {
 302                 return held.ID == schedule.ID
 303         })
 304
 305         if index < 0 {
 306                 return nil
 307         }
 308
 309         if schedule.Cron == "" {
 310                 s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1)
 311
 312                 return s.save()
 313         }
 314
 315         cron, err := scheduleParseCron(schedule.Cron)
 316         if err != nil {
 317                 return fmt.Errorf("schedule %d: %w", schedule.ID, err)
 318         }
 319
 320         next, found := cron.Next(now)
 321         if !found {
 322                 s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1)
 323
 324                 return s.save()
 325         }
 326
 327         s.file.Schedules[index].Due = next
 328
 329         return s.save()
 330 }
 331
 332 // Wait returns how long after now the first schedule is due, or false when
 333 // there is no schedule.
 334 func (s *Schedules) Wait(now time.Time) (time.Duration, bool) {
 335         schedules := s.sorted()
 336         if len(schedules) == 0 {
 337                 return 0, false
 338         }
 339
 340         return max(schedules[0].Due.Sub(now), 0), true
 341 }
 342
 343 // change wakes Run after a set or a remove.
 344 func (s *Schedules) change() {
 345         select {
 346         case s.changed <- struct{}{}:
 347         default:
 348         }
 349 }
 350
 351 // sorted returns a copy of the schedules by due time, then by ID.
 352 func (s *Schedules) sorted() []Schedule {
 353         s.mutex.Lock()
 354         schedules := slices.Clone(s.file.Schedules)
 355         s.mutex.Unlock()
 356
 357         slices.SortFunc(schedules, func(a, b Schedule) int {
 358                 return cmp.Or(a.Due.Compare(b.Due), cmp.Compare(a.ID, b.ID))
 359         })
 360
 361         return schedules
 362 }
 363
 364 // save writes the file whole, with mode 0600. The caller holds the mutex.
 365 func (s *Schedules) save() error {
 366         data, err := json.Marshal(s.file)
 367         if err == nil {
 368                 err = writeFileAtomic(s.path, append(data, '\n'))
 369         }
 370
 371         if err != nil {
 372                 return fmt.Errorf("save schedules: %w", err)
 373         }
 374
 375         return nil
 376 }
 377
 378 // ScheduleFile is the content of the file of the schedules: the ID of the last
 379 // schedule, so that an ID is never used again, and the schedules.
 380 type ScheduleFile struct {
 381         Next      int        `json:"next"`
 382         Schedules []Schedule `json:"schedules"`
 383 }
 384
 385 // Schedule is one schedule: its ID, the task for the agent, the cron expression
 386 // of a repeating schedule, empty for a one-time schedule, and its next time.
 387 type Schedule struct {
 388         ID   int       `json:"id"`
 389         Text string    `json:"text"`
 390         Cron string    `json:"cron"`
 391         Due  time.Time `json:"due"`
 392 }
 393
 394 // Message returns the text of a run at now for the log, with the times in the
 395 // location of now. The first line starts with [Schedule ID] and gives the due
 396 // time and once or the cron expression. Then come the text, the line about a
 397 // late run when now is more than scheduleLate after the due time, and the line
 398 // about the empty reply.
 399 func (s Schedule) Message(now time.Time) string {
 400         lines := []string{
 401                 "[Schedule " + strconv.Itoa(s.ID) + "] " + s.Due.In(now.Location()).Format(scheduleTimeLayout) + ", " + s.repeats(),
 402                 s.Text,
 403         }
 404
 405         if now.Sub(s.Due) > scheduleLate {
 406                 lines = append(lines, scheduleLateLine)
 407         }
 408
 409         return strings.Join(append(lines, scheduleEmptyReply), "\n")
 410 }
 411
 412 // repeats returns once for a one-time schedule, or else repeats and the cron
 413 // expression.
 414 func (s Schedule) repeats() string {
 415         if s.Cron == "" {
 416                 return "once"
 417         }
 418
 419         return "repeats " + s.Cron
 420 }