first version of a task scheduler close #9

Author: dstpierre

function/scheduler.go

@@ -0,0 +1,180 @@
+package function
+
+import (
+	"context"
+	"log"
+	"staticbackend/db"
+	"staticbackend/internal"
+	"time"
+
+	"github.com/go-co-op/gocron"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+	"go.mongodb.org/mongo-driver/mongo"
+)
+
+type TaskScheduler struct {
+	Client    *mongo.Client
+	Volatile  internal.PubSuber
+	Scheduler *gocron.Scheduler
+}
+
+const (
+	TaskTypeFunction = "function"
+	TaskTypeMessage  = "message"
+)
+
+type Task struct {
+	ID       primitive.ObjectID `bson:"_id" json:"id"`
+	Name     string             `bson:"name" json:"name"`
+	Type     string             `bson:"type" json:"type"`
+	Value    string             `bson:"value" json:"value"`
+	Meta     interface{}        `bson:"meta" json:"meta"`
+	Interval string             `bson:"invertal" json:"interval"`
+	LastRun  time.Time          `bson:"last" json:"last"`
+
+	BaseName string `bson:"-" json:"base"`
+}
+
+type MetaMessage struct {
+	Data    string `bson:"data" json:"data"`
+	Channel string `bson:"channel" json:"channel"`
+}
+
+func (ts *TaskScheduler) Start() {
+	tasks, err := ts.listTasks()
+	if err != nil {
+		log.Println("error loading tasks: %v", err)
+		return
+	}
+
+	ts.Scheduler = gocron.NewScheduler(time.UTC)
+	ts.Scheduler.TagsUnique()
+
+	for _, task := range tasks {
+		_, err := ts.Scheduler.Cron(task.Interval).Tag(task.ID.Hex()).Do(ts.run, task)
+		if err != nil {
+			log.Println("error scheduling this task: %s -> %v", task.ID.Hex(), err)
+		}
+	}
+}
+
+func (ts *TaskScheduler) listTasks() ([]Task, error) {
+	bases, err := internal.ListDatabases(ts.Client.Database("sbsys"))
+	if err != nil {
+		return nil, err
+	}
+
+	filter := bson.M{}
+
+	//TODO: Might be worth doing this concurrently
+	var results []Task
+
+	ctx := context.Background()
+
+	for _, base := range bases {
+		db := ts.Client.Database(base.Name)
+		cur, err := db.Collection("sb_tasks").Find(ctx, filter)
+		if err != nil {
+			return nil, err
+		}
+		defer cur.Close(ctx)
+
+		var tasks []Task
+
+		for cur.Next(ctx) {
+			var t Task
+			if err := cur.Decode(&t); err != nil {
+				return nil, err
+			}
+
+			t.BaseName = base.Name
+
+			tasks = append(tasks, t)
+		}
+		if err := cur.Err(); err != nil {
+			return nil, err
+		}
+
+		results = append(results, tasks...)
+	}
+
+	return results, nil
+}
+
+func (ts *TaskScheduler) run(task Task) {
+	curDB := ts.Client.Database(task.BaseName)
+
+	// the task must run as the root base user
+	var auth internal.Auth
+	if err := ts.Volatile.GetTyped("root:"+task.BaseName, &auth); err != nil {
+		tok, err := internal.GetRootForBase(curDB)
+		if err != nil {
+			log.Printf("error finding root token for base %s: %v\n", task.BaseName, err)
+			return
+		}
+
+		auth = internal.Auth{
+			AccountID: tok.AccountID,
+			UserID:    tok.ID,
+			Email:     tok.Email,
+			Role:      tok.Role,
+			Token:     tok.Token,
+		}
+
+		if err := ts.Volatile.SetTyped("root:"+task.BaseName, auth); err != nil {
+			log.Printf("error setting auth inside TaskScheduler.run: ", err)
+			return
+		}
+	}
+
+	switch task.Type {
+	case TaskTypeFunction:
+		ts.execFunction(curDB, auth, task)
+	case TaskTypeMessage:
+		ts.sendMessage(curDB, auth, task)
+	}
+}
+
+func (ts *TaskScheduler) execFunction(curDB *mongo.Database, auth internal.Auth, task Task) {
+
+	fn, err := GetForExecution(curDB, task.Value)
+	if err != nil {
+		log.Printf("cannot find function %s on task %s", task.Value, task.ID.Hex())
+		return
+	}
+
+	exe := &ExecutionEnvironment{
+		Auth:     auth,
+		DB:       curDB,
+		Base:     &db.Base{PublishDocument: ts.Volatile.PublishDocument},
+		Volatile: ts.Volatile,
+		Data:     fn,
+	}
+
+	if err := exe.Execute(); err != nil {
+		log.Printf("error executing function %s: %v", task.Value, err)
+	}
+}
+
+func (ts *TaskScheduler) sendMessage(curDB *mongo.Database, auth internal.Auth, task Task) {
+	token := auth.ReconstructToken()
+
+	meta, ok := task.Meta.(MetaMessage)
+	if !ok {
+		log.Println("unable to get meta data for type MetaMessage for task: ", task.ID.Hex())
+		return
+	}
+
+	msg := internal.Command{
+		SID:     task.ID.Hex(),
+		Type:    task.Value,
+		Data:    meta.Data,
+		Channel: meta.Channel,
+		Token:   token,
+	}
+
+	if err := ts.Volatile.Publish(msg); err != nil {
+		log.Println("error publishing message from task", task.ID.Hex(), err)
+	}
+}

go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/aws/aws-sdk-go v1.27.2
 	github.com/dop251/goja v0.0.0-20210804101310-32956a348b49
 	github.com/gbrlsnchs/jwt/v3 v3.0.0-rc.1
+	github.com/go-co-op/gocron v1.6.2 // indirect
 	github.com/go-redis/redis/v8 v8.4.4
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/uuid v1.1.4

go.sum

@@ -19,6 +19,8 @@ github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWo
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
 github.com/gbrlsnchs/jwt/v3 v3.0.0-rc.1 h1:/opyYiz6HZoBVAU8ypemFOTtzuKFE9kiKstP6RYE1Z4=
 github.com/gbrlsnchs/jwt/v3 v3.0.0-rc.1/go.mod h1:JEL7eYb4ETfz9AYni+/4BV09MrMgGwju0G/k4XF8QMg=
+github.com/go-co-op/gocron v1.6.2 h1:x5g1tWnWcXIZesdosJJcbziRi4XG6tKB92yKLUpoBkU=
+github.com/go-co-op/gocron v1.6.2/go.mod h1:DbJm9kdgr1sEvWpHCA7dFFs/PGHPMil9/97EXCRPr4k=
 github.com/go-redis/redis/v8 v8.4.4 h1:fGqgxCTR1sydaKI00oQf3OmkU/DIe/I/fYXvGklCIuc=
 github.com/go-redis/redis/v8 v8.4.4/go.mod h1:nA0bQuF0i5JFx4Ta9RZxGKXFrQ8cRWntra97f0196iY=
 github.com/go-sourcemap/sourcemap v2.1.3+incompatible h1:W1iEw64niKVGogNgBN3ePyLFfuisuzeidWPMPWmECqU=
@@ -112,6 +114,8 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/go-internal v1.1.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.2.2/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
@@ -127,6 +131,7 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stripe/stripe-go/v71 v71.44.0 h1:kACtWvhEOQ0THj5okxVqEN+YxcUzRK3ls5vuuMxX4xA=
 github.com/stripe/stripe-go/v71 v71.44.0/go.mod h1:BXYwMQe+xjYomcy5/qaTGyoyVMTP3wDCHa7DVFvg8+Y=
 github.com/tidwall/pretty v1.0.0 h1:HsD+QiTn7sK6flMKIvNmpqz1qrpP3Ps6jOKIKMooyg4=
@@ -228,3 +233,4 @@ gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

internal/account.go

@@ -88,6 +88,15 @@ func FindRootToken(db *mongo.Database, id, accountID primitive.ObjectID, token s
 	return
 }
 
+func GetRootForBase(db *mongo.Database) (tok Token, err error) {
+	filter := bson.M{
+		FieldRole: 100,
+	}
+	sr := db.Collection("sb_tokens").FindOne(ctx, filter)
+	err = sr.Decode(&tok)
+	return
+}
+
 func FindTokenByEmail(db *mongo.Database, email string) (tok Token, err error) {
 	sr := db.Collection("sb_tokens").FindOne(ctx, bson.M{"email": email})
 	err = sr.Decode(&tok)
@@ -128,3 +137,31 @@ func DatabaseExists(db *mongo.Database, name string) (bool, error) {
 	}
 	return count > 0, nil
 }
+
+func ListDatabases(db *mongo.Database) ([]BaseConfig, error) {
+	filter := bson.M{FieldIsActive: true}
+
+	ctx := context.Background()
+
+	cur, err := db.Collection("bases").Find(ctx, filter)
+	if err != nil {
+		return nil, err
+	}
+	defer cur.Close(ctx)
+
+	var results []BaseConfig
+
+	for cur.Next(ctx) {
+		var bc BaseConfig
+		if err := cur.Decode(&bc); err != nil {
+			return nil, err
+		}
+
+		results = append(results, bc)
+	}
+	if err := cur.Err(); err != nil {
+		return nil, err
+	}
+
+	return results, nil
+}

internal/data.go

@@ -26,6 +26,8 @@ const (
 	FieldAccountID = "accountId"
 	FieldOwnerID   = "sb_owner"
 	FieldToken     = "token"
+	FieldIsActive  = "active"
+	FieldRole      = "role"
 )
 
 const (