X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/396af2e18eb02d4ec1e35edbc62f01e84bae2b46..2daf91497c2cc70d7156f7c0d99b1451933a818a:/schedule.go?ds=inline

diff --git a/schedule.go b/schedule.go
index 3114ac7..9b95ce7 100644
--- a/schedule.go
+++ b/schedule.go
@@ -2,9 +2,11 @@ package main
 
 import (
 	"cmp"
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"log"
 	"os"
 	"slices"
 	"strconv"
@@ -28,6 +30,14 @@ const (
 	// 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
 
@@ -39,11 +49,15 @@ const (
 )
 
 // openSchedules reads the schedules of the file at path. A missing file holds
-// none. The clock gives the time and the time zone of the schedules.
-func openSchedules(path string, clock func() time.Time) (*Schedules, error) {
+// none. The clock gives the time and the time zone of the schedules, and logger
+// gets one line for each run.
+func openSchedules(path string, clock func() time.Time, logger *log.Logger) (*Schedules, error) {
 	s := &Schedules{
-		path:  path,
-		clock: clock,
+		path:    path,
+		clock:   clock,
+		logger:  logger,
+		check:   scheduleCheck,
+		changed: make(chan struct{}, 1),
 	}
 
 	data, err := os.ReadFile(path)
@@ -77,16 +91,51 @@ func openSchedules(path string, clock func() time.Time) (*Schedules, error) {
 // 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 the runs read
-// them.
+// 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
-	clock func() time.Time
+	path    string
+	clock   func() time.Time
+	logger  *log.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
+		var 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
@@ -141,6 +190,8 @@ func (s *Schedules) Set(text string, at, cron *string) (string, error) {
 		return "", err
 	}
 
+	s.change()
+
 	return fmt.Sprintf("Schedule %d set. Next: %s.", schedule.ID, schedule.Due.In(now.Location()).Format(scheduleTimeLayout)), nil
 }
 
@@ -162,6 +213,14 @@ func (s *Schedules) List() string {
 	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()
@@ -181,9 +240,56 @@ func (s *Schedules) Remove(id int) (string, error) {
 		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 {
+	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
+		}
+
+		if schedule.late(now) {
+			s.logger.Printf("schedule %d runs %s late, because the daemon was stopped at its time", schedule.ID, durationText(now.Sub(schedule.Due)))
+		} else {
+			s.logger.Printf("schedule %d runs", schedule.ID)
+		}
+
+		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
@@ -246,6 +352,14 @@ func (s *Schedules) Wait(now time.Time) (time.Duration, bool) {
 	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()
@@ -300,13 +414,19 @@ func (s Schedule) Message(now time.Time) string {
 		s.Text,
 	}
 
-	if now.Sub(s.Due) > scheduleLate {
+	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 {