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

diff --git a/schedule.go b/schedule.go
index 3114ac7..b3a65ea 100644
--- a/schedule.go
+++ b/schedule.go
@@ -2,9 +2,11 @@ package main
 
 import (
 	"cmp"
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"log/slog"
 	"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,17 @@ 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 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,
-		clock: clock,
+		path:    path,
+		owner:   owner,
+		clock:   clock,
+		logger:  logger,
+		check:   scheduleCheck,
+		changed: make(chan struct{}, 1),
 	}
 
 	data, err := os.ReadFile(path)
@@ -60,7 +76,18 @@ func openSchedules(path string, clock func() time.Time) (*Schedules, error) {
 		return nil, fmt.Errorf("%s: %w", path, err)
 	}
 
-	for _, schedule := range s.file.Schedules {
+	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
 		}
@@ -71,22 +98,68 @@ func openSchedules(path string, clock func() time.Time) (*Schedules, error) {
 		}
 	}
 
+	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 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
+	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
@@ -99,7 +172,8 @@ func (s *Schedules) Set(text string, at, cron *string) (string, error) {
 
 	now := s.clock()
 	schedule := Schedule{
-		Text: text,
+		Text:  text,
+		Owner: s.owner,
 	}
 
 	switch {
@@ -141,6 +215,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 +238,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,12 +265,67 @@ 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 {
+	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)
@@ -246,9 +385,18 @@ 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()
+
 	schedules := slices.Clone(s.file.Schedules)
 	s.mutex.Unlock()
 
@@ -280,13 +428,15 @@ type ScheduleFile struct {
 	Schedules []Schedule `json:"schedules"`
 }
 
-// Schedule is one schedule: its ID, the task for the agent, the cron expression
-// of a repeating schedule, empty for a one-time schedule, and its next time.
+// 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 {
-	ID   int       `json:"id"`
-	Text string    `json:"text"`
-	Cron string    `json:"cron"`
-	Due  time.Time `json:"due"`
+	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
@@ -300,13 +450,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 {