[onechat] / schedule.go

package main

import (
        "cmp"
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "log/slog"
        "os"
        "slices"
        "strconv"
        "strings"
        "sync"
        "time"
)

const (
        // scheduleFileName is the file of the schedules in the directory of the
        // agent.
        scheduleFileName = "schedules.json"

        // scheduleTextBytes is the most bytes of the text of a schedule.
        scheduleTextBytes = 4096

        // scheduleAtLayout is the layout of the time of a one-time schedule.
        scheduleAtLayout = "2006-01-02 15:04"

        // scheduleTimeLayout is the layout of the times in the answers of the
        // 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

        // scheduleEmptyReply is the last line of the message of a run.
        scheduleEmptyReply = "If nothing in this task needs the user, end your turn with an empty reply."

        // scheduleLateLine tells the agent that a run is late.
        scheduleLateLine = "This run is late: the daemon was stopped at that time."
)

// openSchedules reads the schedules of the file at path. A missing file holds
// 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,
                changed: make(chan struct{}, 1),
        }

        data, err := os.ReadFile(path)
        if errors.Is(err, os.ErrNotExist) {
                return s, nil
        }

        if err != nil {
                return nil, fmt.Errorf("read schedules: %w", err)
        }

        err = json.Unmarshal(data, &s.file)
        if err != nil {
                return nil, fmt.Errorf("%s: %w", path, err)
        }

        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
                }

                _, err = scheduleParseCron(schedule.Cron)
                if err != nil {
                        return nil, fmt.Errorf("%s: schedule %d: %w", path, schedule.ID, err)
                }
        }

        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
        owner   string
        clock   func() time.Time
        logger  *slog.Logger
        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 scheduleCheck 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, scheduleCheck))
                        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
// 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 "", 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
// 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)
                }
        }

        return due
}

// Done ends the run of schedule at now: a one-time schedule goes, and a
// repeating schedule gets the next time after now. So a repeating schedule that
// missed several times runs once. When a remove took the schedule first, Done
// changes nothing.
func (s *Schedules) Done(schedule Schedule, now time.Time) error {
        s.mutex.Lock()
        defer s.mutex.Unlock()

        index := slices.IndexFunc(s.file.Schedules, func(held Schedule) bool {
                return held.ID == schedule.ID
        })

        if index < 0 {
                return nil
        }

        if schedule.Cron == "" {
                s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1)

                return s.save()
        }

        cron, err := scheduleParseCron(schedule.Cron)
        if err != nil {
                return fmt.Errorf("schedule %d: %w", schedule.ID, err)
        }

        next, found := cron.Next(now)
        if !found {
                s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1)

                return s.save()
        }

        s.file.Schedules[index].Due = next

        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()

        slices.SortFunc(schedules, func(a, b Schedule) int {
                return cmp.Or(a.Due.Compare(b.Due), cmp.Compare(a.ID, b.ID))
        })

        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'))
        }

        if err != nil {
                return fmt.Errorf("save schedules: %w", err)
        }

        return 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"`
}

// 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"`
}

// Message returns the text of a run at now for the log, with the times in the
// location of now. The first line starts with [Schedule ID] and gives the due
// time and once or the cron expression. Then come the text, the line about a
// late run when now is more than scheduleLate after the due time, and the line
// about the empty reply.
func (s Schedule) Message(now time.Time) string {
        lines := []string{
                "[Schedule " + strconv.Itoa(s.ID) + "] " + s.Due.In(now.Location()).Format(scheduleTimeLayout) + ", " + s.repeats(),
                s.Text,
        }

        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 {
        if s.Cron == "" {
                return "once"
        }

        return "repeats " + s.Cron
}