diff --git a/corteza/corteza.go b/corteza/corteza.go index 0b984f08e..b3a294821 100644 --- a/corteza/corteza.go +++ b/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 } diff --git a/pkg/corredor/deferred.go b/pkg/corredor/deferred.go deleted file mode 100644 index 3f28d564e..000000000 --- a/pkg/corredor/deferred.go +++ /dev/null @@ -1 +0,0 @@ -package corredor diff --git a/pkg/scheduler/helpers.go b/pkg/scheduler/helpers.go new file mode 100644 index 000000000..d76da26c1 --- /dev/null +++ b/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 +} diff --git a/pkg/scheduler/service.go b/pkg/scheduler/service.go new file mode 100644 index 000000000..69f339aad --- /dev/null +++ b/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) + } +}