Add scheduler package, service

corteza/corteza.go

@@ -11,10 +11,12 @@ import (
 	"github.com/cortezaproject/corteza-server/pkg/auth"
 	"github.com/cortezaproject/corteza-server/pkg/corredor"
 	"github.com/cortezaproject/corteza-server/pkg/db"
+	"github.com/cortezaproject/corteza-server/pkg/eventbus"
 	"github.com/cortezaproject/corteza-server/pkg/http"
 	"github.com/cortezaproject/corteza-server/pkg/logger"
 	"github.com/cortezaproject/corteza-server/pkg/mail"
 	"github.com/cortezaproject/corteza-server/pkg/monitor"
+	"github.com/cortezaproject/corteza-server/pkg/scheduler"
 	"github.com/cortezaproject/corteza-server/pkg/sentry"
 )
 
@@ -51,6 +53,8 @@ func (app *App) Setup(log *zap.Logger, opts *app.Options) (err error) {
 
 	monitor.Setup(app.log, opts.Monitor)
 
+	scheduler.Setup(log, eventbus.Default(), 0)
+
 	return
 }
 
@@ -75,6 +79,14 @@ func (app *App) Activate(ctx context.Context) (err error) {
 		return err
 	}
 
+	// Run scheduler
+	scheduler.Run(ctx)
+
+	// Load corredor scripts
+	if err = corredor.Service().Load(ctx); err != nil {
+		return
+	}
+
 	return nil
 }
 

pkg/corredor/deferred.go

@@ -1 +0,0 @@
-package corredor

pkg/scheduler/helpers.go

@@ -0,0 +1,60 @@
+package scheduler
+
+import (
+	"time"
+
+	"github.com/getsentry/sentry-go"
+	"github.com/gorhill/cronexpr"
+)
+
+// OnInterval parses all given strings as crontab expressions (ii) and returns true if any of them matches current time
+func OnInterval(ee ...string) bool {
+	var (
+		// This function will likely to be called exactly on a minute (00 sec) or a few milliseconds after
+		// Round it up to the smallest unit that cronexpr package supports
+		currTime = now().Truncate(time.Second)
+
+		// For cron expression reference we need to subtract 1ns
+		// this will cause Next() fn to include next nanosecond (if it matches)
+		cronRef = currTime.Add(-time.Nanosecond)
+	)
+
+	// At least one of the given expressions should match
+	for _, e := range ee {
+		exp, err := cronexpr.Parse(e)
+		if err != nil {
+			sentry.CaptureException(err)
+			return false
+		}
+
+		if currTime.Equal(exp.Next(cronRef)) {
+			return true
+		}
+	}
+
+	return false
+}
+
+// OnTimestamp parses all given strings as RFC3339 timestamps and returns true if any of them matches current time
+func OnTimestamp(tt ...string) bool {
+	var (
+		// This function will likely to be called exactly on a minute (00 sec) or a few milliseconds after
+		// Round it up to the smallest unit that cronexpr package supports
+		currTime = now().Truncate(time.Second)
+	)
+
+	for _, t := range tt {
+		ts, err := time.Parse(time.RFC3339, t)
+
+		if err != nil {
+			sentry.CaptureException(err)
+			return false
+		}
+
+		if currTime.Equal(ts.Round(time.Second)) {
+			return true
+		}
+	}
+
+	return false
+}

pkg/scheduler/service.go

@@ -0,0 +1,128 @@
+package scheduler
+
+import (
+	"context"
+	"time"
+
+	"go.uber.org/zap"
+
+	"github.com/cortezaproject/corteza-server/pkg/eventbus"
+	"github.com/cortezaproject/corteza-server/pkg/sentry"
+)
+
+type (
+	service struct {
+		log        *zap.Logger
+		events     []eventbus.Event
+		interval   time.Duration
+		dispatcher dispatcher
+
+		// Simple chan to control if service is running or not
+		active chan bool
+	}
+
+	dispatcher interface {
+		Dispatch(ctx context.Context, ev eventbus.Event)
+	}
+)
+
+const (
+	defaultInterval = time.Minute
+)
+
+var (
+	now = func() time.Time { return time.Now() }
+
+	gService *service
+)
+
+// Setup configures global scheduling service
+func Setup(log *zap.Logger, d dispatcher, interval time.Duration) {
+	if gService != nil {
+		// shut it down
+		gService.active <- false
+	}
+
+	gService = NewService(log, d, interval)
+}
+
+func OnTick(events ...eventbus.Event) {
+	gService.OnTick(events...)
+}
+
+func Run(ctx context.Context) {
+	gService.Run(ctx)
+}
+
+func NewService(log *zap.Logger, d dispatcher, interval time.Duration) *service {
+	// Fix interval to positive number
+	if interval == 0 {
+		interval = defaultInterval
+	}
+
+	var svc = &service{
+		log:        log.Named("scheduler"),
+		interval:   interval,
+		dispatcher: d,
+	}
+
+	return svc
+}
+
+// Register all events that should fire on tick (interval)
+func (svc *service) OnTick(events ...eventbus.Event) {
+	svc.events = append(svc.events, events...)
+}
+
+// Run starts event scheduler service
+func (svc service) Run(ctx context.Context) {
+	if svc.active != nil {
+		return
+	}
+
+	svc.active = make(chan bool, 1)
+	go func() {
+		defer sentry.Recover()
+
+		nextTick := now().Truncate(svc.interval).Add(svc.interval)
+
+		svc.log.Info(
+			"starting",
+			zap.Time("delay", nextTick),
+			zap.Duration("interval", svc.interval),
+		)
+
+		// Wait until start of the next interval
+		time.Sleep(nextTick.Sub(now()))
+		svc.log.Info("started")
+
+		// start with first interval
+		svc.dispatch(ctx)
+		ticker := time.NewTicker(svc.interval)
+		defer ticker.Stop()
+		defer svc.log.Info("stopped")
+
+		for {
+			select {
+			case <-svc.active:
+				svc.log.Info("unactivated")
+				return
+			case <-ctx.Done():
+				svc.log.Info("done")
+				return
+			case <-ticker.C:
+				svc.dispatch(ctx)
+			}
+		}
+
+	}()
+}
+
+func (svc service) dispatch(ctx context.Context) {
+	for _, ev := range svc.events {
+		go func(ev eventbus.Event) {
+			sentry.Recover()
+			svc.dispatcher.Dispatch(ctx, ev)
+		}(ev)
+	}
+}