Add support for interval triggers

pkg/automation/scheduled.go

@@ -2,6 +2,8 @@ package automation
 
 import (
 	"time"
+
+	"github.com/DestinyWang/cronexpr"
 )
 
 type (
@@ -12,7 +14,7 @@ type (
 	schedule struct {
 		scriptID   uint64
 		timestamps []time.Time
-		// intervals []interface{}
+		intervals  []cronexpr.Expression
 	}
 )
 
@@ -20,9 +22,14 @@ var (
 	now = func() time.Time { return time.Now() }
 )
 
+const (
+	pickInterval = time.Minute
+)
+
 // scans runnables and builds a set of scheduled scripts
 func buildScheduleList(runables ScriptSet) scheduledSet {
 	set := scheduledSet{}
+	n := now()
 
 	_ = runables.Walk(func(s *Script) error {
 		sch := schedule{scriptID: s.ID}
@@ -38,7 +45,7 @@ func buildScheduleList(runables ScriptSet) scheduledSet {
 
 			if ts, err := time.Parse(time.RFC3339, t.Condition); err == nil {
 				ts = ts.Truncate(time.Minute)
-				if ts.Before(now()) {
+				if ts.Before(n) {
 					// in the past...
 					continue
 				}
@@ -46,13 +53,26 @@ func buildScheduleList(runables ScriptSet) scheduledSet {
 				sch.timestamps = append(sch.timestamps, ts)
 				continue
 			}
+		}
 
-			// @todo parse cron format and fill intervals
+		for _, t := range s.triggers {
+			if !t.IsInterval() {
+				// only interested in interval scripts
+				continue
+			}
+
+			if t.Condition == "" {
+				continue
+			}
+
+			if itv, err := cronexpr.Parse(t.Condition); err == nil {
+				sch.intervals = append(sch.intervals, *itv)
+			}
 		}
 
 		// If there is anything useful in the schedule,
 		// add it to the list
-		if len(sch.timestamps) > 0 {
+		if len(sch.timestamps)+len(sch.intervals) > 0 {
 			set = append(set, sch)
 		}
 
@@ -69,16 +89,25 @@ func buildScheduleList(runables ScriptSet) scheduledSet {
 // is scheduled multiple times,
 func (set scheduledSet) pick() []uint64 {
 	uu := []uint64{}
-	thisMinute := now().Truncate(time.Minute)
+	n := now()
+	thisMinute := n.Truncate(pickInterval)
 	for _, s := range set {
 		for _, t := range s.timestamps {
-			if thisMinute.Equal(t) {
+			if thisMinute.Equal(t.Truncate(pickInterval)) {
 				uu = append(uu, s.scriptID)
 				break
 			}
 		}
 
-		// @todo pick from intervals
+		for _, i := range s.intervals {
+			// go back 1 interval step to check if script is ran in the next step;
+			// since cronexpr can't assert if given time is in the interval.
+			nn := thisMinute.Add(-pickInterval)
+			if thisMinute.Equal(i.Next(nn).Truncate(pickInterval)) {
+				uu = append(uu, s.scriptID)
+				break
+			}
+		}
 	}
 
 	return uu

pkg/automation/service.go

@@ -111,14 +111,14 @@ func (svc *service) Watch(ctx context.Context) {
 		}
 	}()
 
-	// @todo enable when deferred scripts can be execured
-	// go svc.runScheduled(ctx)
+	go svc.runScheduled(ctx)
 
 	svc.logger.Debug("watcher initialized")
 }
 
 // Runs scheduled scripts every minute
 func (svc *service) runScheduled(ctx context.Context) {
+	// @todo make time.Minute
 	var ticker = time.NewTicker(time.Second)
 	defer ticker.Stop()
 	for {
@@ -207,8 +207,9 @@ func (svc *service) reload(ctx context.Context) {
 
 		// update scheduled list
 		// @todo enable when deferred scripts can be execured
-		// svc.scheduled = buildScheduleList(svc.runnables)
-		// svc.logger.Info("deferred scripts scheduled", zap.Int("count", len(svc.scheduled)))
+
+		svc.scheduled = buildScheduleList(svc.runnables)
+		svc.logger.Info("deferred scripts scheduled", zap.Int("count", len(svc.scheduled)))
 
 		return
 	})