Create crond scheduler and dependencies and store to disk

scheduler/scheduler.go

@@ -0,0 +1,121 @@
+package scheduler
+
+import (
+	"fmt"
+	"local/firestormy/config"
+	"local/firestormy/logger"
+	"time"
+
+	cron "github.com/robfig/cron/v3"
+)
+
+type Scheduler struct {
+	cron    *cron.Cron
+	running map[string]cron.EntryID
+}
+
+func New() *Scheduler {
+	l := logger.New()
+	c := cron.New(
+		cron.WithLocation(time.Local),
+		cron.WithLogger(l),
+		cron.WithChain(
+			cron.SkipIfStillRunning(l),
+			cron.Recover(l),
+		),
+	)
+	return &Scheduler{
+		cron:    c,
+		running: make(map[string]cron.EntryID),
+	}
+}
+
+func (s *Scheduler) Start() error {
+	jobs, err := s.List()
+	if err != nil {
+		return err
+	}
+	for i := range jobs {
+		if err := s.Add(jobs[i]); err != nil && err != ErrDuplicateJob {
+			return err
+		}
+	}
+	s.cron.Start()
+	return nil
+}
+
+func (s *Scheduler) Stop() error {
+	ctx := s.cron.Stop()
+	<-ctx.Done()
+	return nil
+}
+
+func (s *Scheduler) List() ([]*Job, error) {
+	entries, err := config.Store.List(nil)
+	if err != nil {
+		return nil, err
+	}
+	errors := []error{}
+	jobs := []*Job{}
+	for _, k := range entries {
+		j, err := s.loadJobFromStore(k)
+		if err != nil {
+			errors = append(errors, err)
+		} else {
+			jobs = append(jobs, j)
+		}
+	}
+	err = nil
+	if len(errors) > 0 {
+		err = fmt.Errorf("errors listing jobs: %v", errors)
+	}
+	return jobs, err
+}
+
+func (s *Scheduler) loadJobFromStore(k string) (*Job, error) {
+	b, err := config.Store.Get(k)
+	if err != nil {
+		return nil, err
+	}
+	j := &Job{}
+	err = j.Decode(b)
+	return j, err
+}
+
+func (s *Scheduler) Add(j *Job) error {
+	if _, ok := s.getEntry(j); ok {
+		return ErrDuplicateJob
+	}
+	b, err := j.Encode()
+	if err != nil {
+		return err
+	}
+	if err := config.Store.Set(j.Name, b); err != nil {
+		return err
+	}
+	entryID, err := s.cron.AddJob(j.Schedule, j)
+	if err != nil {
+		return err
+	}
+	s.running[j.Name] = entryID
+	return nil
+}
+
+func (s *Scheduler) Remove(j *Job) error {
+	entryID, ok := s.getEntry(j)
+	if !ok {
+		return ErrJobNotFound
+	}
+	was := len(s.cron.Entries())
+	s.cron.Remove(entryID)
+	is := len(s.cron.Entries())
+	if was == is {
+		return ErrJobNotFound
+	}
+	return config.Store.Set(j.Name, nil)
+}
+
+func (s *Scheduler) getEntry(j *Job) (cron.EntryID, bool) {
+	entryID, ok := s.running[j.Name]
+	return entryID, ok
+}