added function execution history and fixed bugs

Author: dstpierre

function/management.go

@@ -38,6 +38,7 @@ func Add(db *mongo.Database, data ExecData) (string, error) {
 	data.ID = primitive.NewObjectID()
 	data.Version = 1
 	data.LastUpdated = time.Now()
+	data.History = make([]ExecHistory, 0)
 
 	ctx := context.Background()
 	res, err := db.Collection("sb_functions").InsertOne(ctx, data)
@@ -190,3 +191,17 @@ func Delete(db *mongo.Database, name string) error {
 
 	return nil
 }
+
+func Ran(db *mongo.Database, id primitive.ObjectID, rh ExecHistory) error {
+	filter := bson.M{internal.FieldID: id}
+	update := bson.M{
+		"$set":  bson.M{"lr": time.Now()},
+		"$push": bson.M{"h": rh},
+	}
+	ctx := context.Background()
+	res := db.Collection("sb_functions").FindOneAndUpdate(ctx, filter, update)
+	if err := res.Err(); err != nil {
+		return err
+	}
+	return nil
+}

function/runtime.go

@@ -2,9 +2,13 @@ package function
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
+	"log"
+	"net/http"
 	"staticbackend/db"
 	"staticbackend/internal"
+	"time"
 
 	"github.com/dop251/goja"
 	"go.mongodb.org/mongo-driver/bson/primitive"
@@ -17,41 +21,89 @@ type ExecutionEnvironment struct {
 	Base     *db.Base
 	Volatile internal.PubSuber
 	Data     ExecData
+
+	CurrentRun ExecHistory
 }
 
 type Result struct {
 	OK      bool        `json:"ok"`
 	Content interface{} `json:"content"`
 }
 
-func (env *ExecutionEnvironment) Execute() error {
+func (env *ExecutionEnvironment) Execute(data interface{}) error {
 	vm := goja.New()
 	vm.SetFieldNameMapper(goja.TagFieldNameMapper("json", true))
 
 	env.addHelpers(vm)
 	env.addDatabaseFunctions(vm)
 	env.addVolatileFunctions(vm)
 
-	result, err := vm.RunString(env.Data.Code)
-	if err != nil {
+	if _, err := vm.RunString(env.Data.Code); err != nil {
 		return err
 	}
 
 	handler, ok := goja.AssertFunction(vm.Get("handle"))
 	if !ok {
-		return fmt.Errorf(`unable to find function "handle": %v`, err)
+		return errors.New(`unable to find function "handle"`)
+	}
+
+	args, err := env.prepareArguments(vm, data)
+	if err != nil {
+		return fmt.Errorf("error preparing argument: %v", err)
+	}
+
+	env.CurrentRun = ExecHistory{
+		ID:      primitive.NewObjectID().Hex(),
+		Version: env.Data.Version,
+		Started: time.Now(),
+		Output:  make([]string, 0),
 	}
 
-	resp, err := handler(goja.Undefined())
+	env.CurrentRun.Output = append(env.CurrentRun.Output, "Function started")
+
+	_, err = handler(goja.Undefined(), args...)
+	go env.complete(err)
 	if err != nil {
 		return fmt.Errorf("error executing your function: %v", err)
 	}
 
-	fmt.Println("resp", resp)
-	fmt.Println("result", result)
 	return nil
 }
 
+func (env *ExecutionEnvironment) prepareArguments(vm *goja.Runtime, data interface{}) ([]goja.Value, error) {
+	var args []goja.Value
+
+	// for "web" trigger we prepare the body, query string and headers
+	r, ok := data.(*http.Request)
+	if ok {
+		defer r.Body.Close()
+
+		// let's ready the HTTP body
+		if r.Header.Get("Content-Type") == "application/json" {
+			var v interface{}
+			if err := json.NewDecoder(r.Body).Decode(&v); err != nil {
+				return nil, err
+			}
+
+			args = append(args, vm.ToValue(v))
+		} else if r.Header.Get("Content-Type") == "application/x-www-form-urlencoded" {
+			if err := r.ParseForm(); err != nil {
+				return nil, err
+			}
+			args = append(args, vm.ToValue(r.Form))
+		}
+
+		args = append(args, vm.ToValue(r.URL.Query()))
+		args = append(args, vm.ToValue(r.Header))
+
+		return args, nil
+	}
+
+	// system or custom event/topic, we send only the 1st argument (body)
+	args = append(args, vm.ToValue(data))
+	return args, nil
+}
+
 func (env *ExecutionEnvironment) addHelpers(vm *goja.Runtime) {
 	vm.Set("log", func(call goja.FunctionCall) goja.Value {
 		if len(call.Arguments) == 0 {
@@ -62,7 +114,7 @@ func (env *ExecutionEnvironment) addHelpers(vm *goja.Runtime) {
 		for _, v := range call.Arguments {
 			params = append(params, v.Export())
 		}
-		fmt.Println(params...)
+		env.CurrentRun.Output = append(env.CurrentRun.Output, fmt.Sprint(params...))
 		return goja.Undefined()
 	})
 }
@@ -292,3 +344,21 @@ func (env *ExecutionEnvironment) addVolatileFunctions(vm *goja.Runtime) {
 		return vm.ToValue(Result{OK: true})
 	})
 }
+
+func (env *ExecutionEnvironment) complete(err error) {
+	env.CurrentRun.Completed = time.Now()
+	env.CurrentRun.Success = err == mongo.ErrNilCursor
+
+	env.CurrentRun.Output = append(env.CurrentRun.Output, "Function completed")
+
+	// add the error in the last output entry
+	if err != nil {
+		env.CurrentRun.Output = append(env.CurrentRun.Output, err.Error())
+	}
+
+	//TODO: this needs to be regrouped and ran un batch
+	if err := Ran(env.DB, env.Data.ID, env.CurrentRun); err != nil {
+		//TODO: do something with those error
+		log.Println("error logging function complete: ", err)
+	}
+}

function/scheduler.go

@@ -152,7 +152,7 @@ func (ts *TaskScheduler) execFunction(curDB *mongo.Database, auth internal.Auth,
 		Data:     fn,
 	}
 
-	if err := exe.Execute(); err != nil {
+	if err := exe.Execute(task.Name); err != nil {
 		log.Printf("error executing function %s: %v", task.Value, err)
 	}
 }

function/subscriber.go

@@ -46,7 +46,8 @@ func (sub *Subscriber) process(msg internal.Command) {
 func (sub *Subscriber) handleRealtimeEvents(msg internal.Command) {
 	exe, err := sub.GetExecEnv(msg.Token)
 	if err != nil {
-		log.Println("cannot retrieve base from token")
+		log.Println("cannot retrieve base from token", msg.Token)
+		log.Println(err)
 		return
 	}
 
@@ -81,7 +82,7 @@ func (sub *Subscriber) handleRealtimeEvents(msg internal.Command) {
 
 		exe.Data = fn
 		go func(ex ExecutionEnvironment) {
-			if err := ex.Execute(); err != nil {
+			if err := ex.Execute(msg); err != nil {
 				log.Printf(`executing "%s" function failed: %v"`, ex.Data.FunctionName, err)
 			}
 		}(exe)

functions.go

@@ -111,7 +111,7 @@ func (f *functions) exec(w http.ResponseWriter, r *http.Request) {
 		Base: f.base,
 		Data: fn,
 	}
-	if err := env.Execute(); err != nil {
+	if err := env.Execute(r); err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}

membership.go

@@ -87,10 +87,17 @@ func login(w http.ResponseWriter, r *http.Request) {
 		Role:      tok.Role,
 		Token:     tok.Token,
 	}
+
+	//TODO: find a good way to find all occurences of those two
+	// and make them easily callable via a shared function
 	if err := volatile.SetTyped(token, auth); err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
+	if err := volatile.SetTyped("base:"+token, conf); err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
 
 	respond(w, http.StatusOK, string(jwtBytes))
 }
@@ -141,12 +148,32 @@ func register(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	jwtBytes, _, err := createAccountAndUser(db, l.Email, l.Password, 0)
+	jwtBytes, tok, err := createAccountAndUser(db, l.Email, l.Password, 0)
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
-	respond(w, http.StatusOK, string(jwtBytes))
+
+	token := string(jwtBytes)
+
+	auth := internal.Auth{
+		AccountID: tok.AccountID,
+		UserID:    tok.ID,
+		Email:     tok.Email,
+		Role:      tok.Role,
+		Token:     tok.Token,
+	}
+
+	if err := volatile.SetTyped(token, auth); err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	if err := volatile.SetTyped("base:"+token, conf); err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	respond(w, http.StatusOK, token)
 }
 
 func createAccountAndUser(db *mongo.Database, email, password string, role int) ([]byte, internal.Token, error) {

realtime/broker.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"staticbackend/internal"
 	"strings"
+	"time"
 
 	"github.com/google/uuid"
 )
@@ -192,7 +193,18 @@ func (b *Broker) getTargets(msg internal.Command) (sockets []chan internal.Comma
 
 		go b.pubsub.Subscribe(sender, msg.Token, msg.Data, closesub)
 
-		payload = internal.Command{Type: internal.MsgTypeJoined, Data: msg.Data}
+		joinedMsg := internal.Command{
+			Type:    internal.MsgTypeJoined,
+			Data:    msg.SID,
+			Channel: msg.Data,
+		}
+		// make sure the subscription had time to kick-off
+		go func(m internal.Command) {
+			time.Sleep(250 * time.Millisecond)
+			b.pubsub.Publish(joinedMsg)
+		}(joinedMsg)
+
+		payload = internal.Command{Type: internal.MsgTypeOk, Data: msg.Data}
 	case internal.MsgTypePresence:
 		v, err := b.pubsub.Get(msg.Data)
 		if err != nil {

server.go

@@ -3,6 +3,7 @@ package staticbackend
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"log"
 	"net/http"
@@ -72,9 +73,25 @@ func Start(dbHost, port string) {
 			return key, nil
 		}
 
-		if _, err := middleware.ValidateAuthKey(client, volatile, ctx, key); err != nil {
+		auth, err := middleware.ValidateAuthKey(client, volatile, ctx, key)
+		if err != nil {
 			return "", err
 		}
+
+		// set base:token useful when executing pubsub event message / function
+		conf, ok := ctx.Value(middleware.ContextBase).(internal.BaseConfig)
+		if !ok {
+			return "", errors.New("could not find base config")
+		}
+
+		//TODO: Lots of repetition of this, needs to be refactor
+		if err := volatile.SetTyped(key, auth); err != nil {
+			return "", err
+		}
+		if err := volatile.SetTyped("base:"+key, conf); err != nil {
+			return "", err
+		}
+
 		return key, nil
 	}, volatile)
 
@@ -162,6 +179,7 @@ func Start(dbHost, port string) {
 	http.Handle("/fn/add", middleware.Chain(http.HandlerFunc(f.add), stdRoot...))
 	http.Handle("/fn/update", middleware.Chain(http.HandlerFunc(f.update), stdRoot...))
 	http.Handle("/fn/delete/", middleware.Chain(http.HandlerFunc(f.del), stdRoot...))
+	http.Handle("/fn/del/", middleware.Chain(http.HandlerFunc(f.del), stdRoot...))
 	http.Handle("/fn/info/", middleware.Chain(http.HandlerFunc(f.info), stdRoot...))
 	http.Handle("/fn/exec", middleware.Chain(http.HandlerFunc(f.exec), stdAuth...))
 	http.Handle("/fn", middleware.Chain(http.HandlerFunc(f.list), stdRoot...))
@@ -213,11 +231,13 @@ func initServices(dbHost string) {
 
 		var conf internal.BaseConfig
 		if err := volatile.GetTyped("base:"+token, &conf); err != nil {
+			log.Println("cannot find base")
 			return exe, err
 		}
 
 		var auth internal.Auth
 		if err := volatile.GetTyped(token, &auth); err != nil {
+			log.Println("cannot find auth")
 			return exe, err
 		}