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