X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/150d707b16a5495272dfb02d690d99bfae50c14e..d8f3785bb69288663f4cc46bf4d609c6770452d5:/schedule.go

diff --git a/schedule.go b/schedule.go
index 9b95ce7..7eb747f 100644
--- a/schedule.go
+++ b/schedule.go
@@ -6,7 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"log"
+	"log/slog"
 	"os"
 	"slices"
 	"strconv"
@@ -33,10 +33,10 @@ const (
 	// 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
+	// scheduleClockWaitTimeout 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.
+	scheduleClockWaitTimeout = time.Minute
 
 	// scheduleLate is how long after its time a run is late.
 	scheduleLate = time.Minute
@@ -48,15 +48,50 @@ const (
 	scheduleLateLine = "This run is late: the daemon was stopped at that time."
 )
 
+// 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 Run reads them.
+// changed tells Run that a set or a remove changed the times.
+type Schedules struct {
+	path    string
+	owner   string
+	clock   func() time.Time
+	logger  *slog.Logger
+	changed chan struct{}
+
+	mutex sync.Mutex
+	file  ScheduleFile
+}
+
+// ScheduleFile is the content of the file of the schedules: the ID of the last
+// schedule, so that an ID is never used again, and the schedules.
+type ScheduleFile struct {
+	Next      int        `json:"next"`
+	Schedules []Schedule `json:"schedules"`
+}
+
+// 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 {
+	Owner string    `json:"owner"`
+	ID    int       `json:"id"`
+	Text  string    `json:"text"`
+	Cron  string    `json:"cron"`
+	Due   time.Time `json:"due"`
+}
+
 // 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, and logger
-// gets one line for each run.
-func openSchedules(path string, clock func() time.Time, logger *log.Logger) (*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,
+		owner:   owner,
 		clock:   clock,
 		logger:  logger,
-		check:   scheduleCheck,
 		changed: make(chan struct{}, 1),
 	}
 
@@ -74,7 +109,18 @@ func openSchedules(path string, clock func() time.Time, logger *log.Logger) (*Sc
 		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
 		}
@@ -85,29 +131,35 @@ func openSchedules(path string, clock func() time.Time, logger *log.Logger) (*Sc
 		}
 	}
 
+	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 Run reads them.
-// changed tells Run that a set or a remove changed the times.
-type Schedules struct {
-	path    string
-	clock   func() time.Time
-	logger  *log.Logger
-	check   time.Duration
-	changed chan struct{}
+// save writes the file whole, with mode 0600. The caller holds the mutex.
+func (s *Schedules) save() error {
+	data, err := json.Marshal(s.file)
+	if err == nil {
+		err = writeFileAtomic(s.path, append(data, '\n'))
+	}
 
-	mutex sync.Mutex
-	file  ScheduleFile
+	if err != nil {
+		return fmt.Errorf("save schedules: %w", err)
+	}
+
+	return nil
 }
 
 // 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.
+// the first due time, at most scheduleClockWaitTimeout 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)
@@ -115,11 +167,14 @@ func (s *Schedules) Run(ctx context.Context, l *TurnLoop) error {
 			return err
 		}
 
-		var timer *time.Timer
-		var timeout <-chan time.Time
+		var (
+			timer   *time.Timer
+			timeout <-chan time.Time
+		)
+
 		wait, found := s.Wait(s.clock())
 		if found {
-			timer = time.NewTimer(min(wait, s.check))
+			timer = time.NewTimer(min(wait, scheduleClockWaitTimeout))
 			timeout = timer.C
 		}
 
@@ -136,115 +191,6 @@ func (s *Schedules) Run(ctx context.Context, l *TurnLoop) error {
 	}
 }
 
-// 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
-// repeats is its first occurrence. cron is an expression of scheduleParseCron.
-func (s *Schedules) Set(text string, at, cron *string) (string, error) {
-	text = strings.TrimSpace(text)
-	if text == "" || len(text) > scheduleTextBytes {
-		return "", fmt.Errorf("the text must have 1 to %d bytes", scheduleTextBytes)
-	}
-
-	now := s.clock()
-	schedule := Schedule{
-		Text: text,
-	}
-
-	switch {
-	case (at == nil) == (cron == nil):
-		return "", errors.New("give either at or cron, and null for the other")
-	case at != nil:
-		parsed, err := time.Parse(scheduleAtLayout, *at)
-		if err != nil {
-			return "", fmt.Errorf("at must be a local time such as %s", now.Format(scheduleAtLayout))
-		}
-
-		schedule.Due = scheduleWallTime(parsed.Year(), parsed.Month(), parsed.Day(), parsed.Hour(), parsed.Minute(), now.Location())
-		if !schedule.Due.After(now) {
-			return "", fmt.Errorf("the time %s is not after now, %s", schedule.Due.Format(scheduleTimeLayout), now.Format(scheduleTimeLayout))
-		}
-	default:
-		parsed, err := scheduleParseCron(*cron)
-		if err != nil {
-			return "", err
-		}
-
-		next, found := parsed.Next(now)
-		if !found {
-			return "", fmt.Errorf("the cron expression %q matches no time in %d years", *cron, scheduleCronYears)
-		}
-
-		schedule.Cron = parsed.String()
-		schedule.Due = next
-	}
-
-	s.mutex.Lock()
-	defer s.mutex.Unlock()
-
-	s.file.Next++
-	schedule.ID = s.file.Next
-	s.file.Schedules = append(s.file.Schedules, schedule)
-	err := s.save()
-	if err != nil {
-		return "", err
-	}
-
-	s.change()
-
-	return fmt.Sprintf("Schedule %d set. Next: %s.", schedule.ID, schedule.Due.In(now.Location()).Format(scheduleTimeLayout)), nil
-}
-
-// List returns the answer of the tool: one line for each schedule, by due time,
-// with its ID, its next time, its cron expression or once, and its text on one
-// line.
-func (s *Schedules) List() string {
-	location := s.clock().Location()
-	schedules := s.sorted()
-	if len(schedules) == 0 {
-		return "No schedules."
-	}
-
-	lines := make([]string, 0, len(schedules))
-	for _, schedule := range schedules {
-		lines = append(lines, strconv.Itoa(schedule.ID)+" | "+schedule.Due.In(location).Format(scheduleTimeLayout)+" | "+cmp.Or(schedule.Cron, "once")+" | "+strings.Join(strings.Fields(schedule.Text), " "))
-	}
-
-	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()
-	defer s.mutex.Unlock()
-
-	index := slices.IndexFunc(s.file.Schedules, func(schedule Schedule) bool {
-		return schedule.ID == id
-	})
-
-	if index < 0 {
-		return "", fmt.Errorf("no schedule %d", id)
-	}
-
-	s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1)
-	err := s.save()
-	if err != nil {
-		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
@@ -252,8 +198,14 @@ func (s *Schedules) Remove(id int) (string, error) {
 // 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 {
@@ -276,12 +228,13 @@ func (s *Schedules) deliver(ctx context.Context, l *TurnLoop) error {
 			return err
 		}
 
+		attrs := []any{"id", schedule.ID}
 		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)
+			attrs = append(attrs, "late", now.Sub(schedule.Due))
 		}
 
+		s.logger.Info("schedule ran", attrs...)
+
 		turns = append(turns, message)
 	}
 
@@ -290,9 +243,21 @@ func (s *Schedules) deliver(ctx context.Context, l *TurnLoop) error {
 	return nil
 }
 
+// Wait returns how long after now the first schedule is due, or false when
+// there is no schedule.
+func (s *Schedules) Wait(now time.Time) (time.Duration, bool) {
+	schedules := s.sorted()
+	if len(schedules) == 0 {
+		return 0, false
+	}
+
+	return max(schedules[0].Due.Sub(now), 0), true
+}
+
 // 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)
@@ -341,28 +306,10 @@ func (s *Schedules) Done(schedule Schedule, now time.Time) error {
 	return s.save()
 }
 
-// Wait returns how long after now the first schedule is due, or false when
-// there is no schedule.
-func (s *Schedules) Wait(now time.Time) (time.Duration, bool) {
-	schedules := s.sorted()
-	if len(schedules) == 0 {
-		return 0, false
-	}
-
-	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()
 
@@ -373,34 +320,122 @@ func (s *Schedules) sorted() []Schedule {
 	return schedules
 }
 
-// save writes the file whole, with mode 0600. The caller holds the mutex.
-func (s *Schedules) save() error {
-	data, err := json.Marshal(s.file)
-	if err == nil {
-		err = writeFileAtomic(s.path, append(data, '\n'))
+// 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
+// repeats is its first occurrence. cron is an expression of scheduleParseCron.
+func (s *Schedules) Set(text string, at, cron *string) (string, error) {
+	text = strings.TrimSpace(text)
+	if text == "" || len(text) > scheduleTextBytes {
+		return "", fmt.Errorf("the text must have 1 to %d bytes", scheduleTextBytes)
+	}
+
+	now := s.clock()
+	schedule := Schedule{
+		Text:  text,
+		Owner: s.owner,
 	}
 
+	switch {
+	case (at == nil) == (cron == nil):
+		return "", errors.New("give either at or cron, and null for the other")
+	case at != nil:
+		parsed, err := time.Parse(scheduleAtLayout, *at)
+		if err != nil {
+			return "", fmt.Errorf("at must be a local time such as %s", now.Format(scheduleAtLayout))
+		}
+
+		schedule.Due = scheduleWallTime(parsed.Year(), parsed.Month(), parsed.Day(), parsed.Hour(), parsed.Minute(), now.Location())
+		if !schedule.Due.After(now) {
+			return "", fmt.Errorf("the time %s is not after now, %s", schedule.Due.Format(scheduleTimeLayout), now.Format(scheduleTimeLayout))
+		}
+	default:
+		parsed, err := scheduleParseCron(*cron)
+		if err != nil {
+			return "", err
+		}
+
+		next, found := parsed.Next(now)
+		if !found {
+			return "", fmt.Errorf("the cron expression %q matches no time in %d years", *cron, scheduleCronYears)
+		}
+
+		schedule.Cron = parsed.String()
+		schedule.Due = next
+	}
+
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	s.file.Next++
+	schedule.ID = s.file.Next
+	s.file.Schedules = append(s.file.Schedules, schedule)
+	err := s.save()
 	if err != nil {
-		return fmt.Errorf("save schedules: %w", err)
+		return "", err
 	}
 
-	return nil
+	s.change()
+
+	return fmt.Sprintf("Schedule %d set. Next: %s.", schedule.ID, schedule.Due.In(now.Location()).Format(scheduleTimeLayout)), nil
 }
 
-// ScheduleFile is the content of the file of the schedules: the ID of the last
-// schedule, so that an ID is never used again, and the schedules.
-type ScheduleFile struct {
-	Next      int        `json:"next"`
-	Schedules []Schedule `json:"schedules"`
+// change wakes Run after a set or a remove.
+func (s *Schedules) change() {
+	select {
+	case s.changed <- struct{}{}:
+	default:
+	}
 }
 
-// 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.
-type Schedule struct {
-	ID   int       `json:"id"`
-	Text string    `json:"text"`
-	Cron string    `json:"cron"`
-	Due  time.Time `json:"due"`
+// List returns the answer of the tool: one line for each schedule, by due time,
+// with its ID, its next time, its cron expression or once, and its text on one
+// line.
+func (s *Schedules) List() string {
+	location := s.clock().Location()
+	schedules := s.sorted()
+	if len(schedules) == 0 {
+		return "No schedules."
+	}
+
+	lines := make([]string, 0, len(schedules))
+	for _, schedule := range schedules {
+		lines = append(lines, strconv.Itoa(schedule.ID)+" | "+schedule.Due.In(location).Format(scheduleTimeLayout)+" | "+cmp.Or(schedule.Cron, "once")+" | "+strings.Join(strings.Fields(schedule.Text), " "))
+	}
+
+	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()
+	defer s.mutex.Unlock()
+
+	index := slices.IndexFunc(s.file.Schedules, func(schedule Schedule) bool {
+		return schedule.ID == id
+	})
+
+	if index < 0 {
+		return "", fmt.Errorf("no schedule %d", id)
+	}
+
+	s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1)
+	err := s.save()
+	if err != nil {
+		return "", err
+	}
+
+	s.change()
+
+	return fmt.Sprintf("Schedule %d removed.", id), nil
 }
 
 // Message returns the text of a run at now for the log, with the times in the