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