initial proposal for pubsub/event message close #8

Author: dstpierre

cache/cache.go

@@ -42,6 +42,23 @@ func (c *Cache) Set(key string, value string) error {
 	return nil
 }
 
+func (c *Cache) GetTyped(key string, v interface{}) error {
+	s, err := c.Get(key)
+	if err != nil {
+		return err
+	}
+
+	return json.Unmarshal([]byte(s), v)
+}
+
+func (c *Cache) SetTyped(key string, v interface{}) error {
+	b, err := json.Marshal(v)
+	if err != nil {
+		return err
+	}
+	return c.Set(key, string(b))
+}
+
 func (c *Cache) Inc(key string, by int64) (int64, error) {
 	return c.Rdb.IncrBy(c.Ctx, key, by).Result()
 }
@@ -70,14 +87,16 @@ func (c *Cache) Subscribe(send chan internal.Command, token, channel string, clo
 				return
 			}
 
-			// for non DB events we change the type to MsgTypeChanOut
-			if !msg.IsDBEvent() {
+			// TODO: this will need more thinking
+			if msg.Type == internal.MsgTypeChanIn {
 				msg.Type = internal.MsgTypeChanOut
-			} else if c.HasPermission(token, channel, msg.Data) == false {
+			} else if msg.IsSystemEvent {
+
+			} else if msg.IsDBEvent() && c.HasPermission(token, channel, msg.Data) == false {
 				continue
 			}
 			send <- msg
-		case _ = <-close:
+		case <-close:
 			_ = pubsub.Close()
 			return
 		}
@@ -93,6 +112,22 @@ func (c *Cache) Publish(msg internal.Command) error {
 	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
 	defer cancel()
 
+	// Publish the event to system so server-side function can trigger
+	go func(sysmsg internal.Command) {
+		sysmsg.IsSystemEvent = true
+		b, err := json.Marshal(sysmsg)
+		if err != nil {
+			log.Println("error marshaling the system msg: ", err)
+			return
+		}
+
+		sysctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+		defer cancel()
+		if err := c.Rdb.Publish(sysctx, "sbsys", string(b)).Err(); err != nil {
+			log.Println("error publishing to system channel: ", err)
+		}
+	}(msg)
+
 	return c.Rdb.Publish(ctx, msg.Channel, string(b)).Err()
 }
 
@@ -129,8 +164,8 @@ func (c *Cache) PublishDocument(channel, typ string, v interface{}) {
 }
 
 func (c *Cache) HasPermission(token, repo, payload string) bool {
-	me, ok := internal.Tokens[token]
-	if !ok {
+	var me internal.Auth
+	if err := c.GetTyped(token, &me); err != nil {
 		return false
 	}
 

function/management.go

@@ -138,6 +138,32 @@ func List(db *mongo.Database) ([]ExecData, error) {
 	return results, nil
 }
 
+func ListByTrigger(db *mongo.Database, trigger string) ([]ExecData, error) {
+	opt := &options.FindOptions{}
+	opt.SetProjection(bson.M{"h": 0})
+
+	filter := bson.M{"tr": trigger}
+	ctx := context.Background()
+	cur, err := db.Collection("sb_functions").Find(ctx, filter, opt)
+	if err != nil {
+		return nil, err
+	}
+	defer cur.Close(ctx)
+
+	var results []ExecData
+	for cur.Next(ctx) {
+		var ed ExecData
+		err := cur.Decode(&ed)
+		if err != nil {
+			return nil, err
+		}
+
+		results = append(results, ed)
+	}
+
+	return results, nil
+}
+
 func Delete(db *mongo.Database, name string) error {
 	filter := bson.M{"name": name}
 

function/runtime.go

@@ -1,6 +1,7 @@
 package function
 
 import (
+	"encoding/json"
 	"fmt"
 	"staticbackend/db"
 	"staticbackend/internal"
@@ -11,10 +12,11 @@ import (
 )
 
 type ExecutionEnvironment struct {
-	Auth internal.Auth
-	DB   *mongo.Database
-	Base *db.Base
-	Data ExecData
+	Auth     internal.Auth
+	DB       *mongo.Database
+	Base     *db.Base
+	Volatile internal.PubSuber
+	Data     ExecData
 }
 
 type Result struct {
@@ -28,6 +30,7 @@ func (env *ExecutionEnvironment) Execute() error {
 
 	env.addHelpers(vm)
 	env.addDatabaseFunctions(vm)
+	env.addVolatileFunctions(vm)
 
 	result, err := vm.RunString(env.Data.Code)
 	if err != nil {
@@ -255,3 +258,37 @@ func (*ExecutionEnvironment) clean(doc map[string]interface{}) error {
 
 	return nil
 }
+
+func (env *ExecutionEnvironment) addVolatileFunctions(vm *goja.Runtime) {
+	vm.Set("send", func(call goja.FunctionCall) goja.Value {
+		if len(call.Arguments) != 3 {
+			return vm.ToValue(Result{Content: "argument missmatch: you need 3 arguments for send(type, data, channel)"})
+		}
+
+		var typ, channel string
+		if err := vm.ExportTo(call.Argument(0), &typ); err != nil {
+			return vm.ToValue(Result{Content: "the first argument should be a string"})
+		} else if err := vm.ExportTo(call.Argument(2), &channel); err != nil {
+			return vm.ToValue(Result{Content: "the third argument should be a string"})
+		}
+
+		b, err := json.Marshal(call.Argument(1).Export())
+		if err != nil {
+			return vm.ToValue(Result{Content: fmt.Sprintf("error converting your data: %v", err)})
+		}
+
+		msg := internal.Command{
+			SID:     env.Data.ID.Hex(),
+			Type:    typ,
+			Data:    string(b),
+			Channel: channel,
+			Token:   env.Auth.ReconstructToken(),
+		}
+
+		if err := env.Volatile.Publish(msg); err != nil {
+			return vm.ToValue(Result{Content: fmt.Sprintf("error publishing your message: %v", err)})
+		}
+
+		return vm.ToValue(Result{OK: true})
+	})
+}

function/subscriber.go

@@ -0,0 +1,89 @@
+package function
+
+import (
+	"fmt"
+	"log"
+	"staticbackend/internal"
+
+	"go.mongodb.org/mongo-driver/bson/primitive"
+)
+
+type Subscriber struct {
+	PubSub     internal.PubSuber
+	GetExecEnv func(token string) (ExecutionEnvironment, error)
+}
+
+// Start starts the system event subscription.
+// This channel is responsible of executing functions that match the
+// topic/trigger
+func (sub *Subscriber) Start() {
+	receiver := make(chan internal.Command)
+	close := make(chan bool)
+
+	go sub.PubSub.Subscribe(receiver, "", "sbsys", close)
+
+	for {
+		select {
+		case msg := <-receiver:
+			go sub.process(msg)
+		case _ = <-close:
+			log.Println("system event channel closed?!?")
+			break
+		}
+	}
+}
+
+func (sub *Subscriber) process(msg internal.Command) {
+	switch msg.Type {
+	case internal.MsgTypeChanOut,
+		internal.MsgTypeDBCreated,
+		internal.MsgTypeDBUpdated,
+		internal.MsgTypeDBDeleted:
+		sub.handleRealtimeEvents(msg)
+	}
+}
+
+func (sub *Subscriber) handleRealtimeEvents(msg internal.Command) {
+	exe, err := sub.GetExecEnv(msg.Token)
+	if err != nil {
+		log.Println("cannot retrieve base from token")
+		return
+	}
+
+	var ids []primitive.ObjectID
+
+	key := fmt.Sprintf("%s:%s", exe.DB.Name(), msg.Type)
+	if err := sub.PubSub.GetTyped(key, &ids); err != nil {
+		funcs, err := ListByTrigger(exe.DB, msg.Type)
+		if err != nil {
+			log.Println("error getting functions by trigger: ", err)
+			return
+		}
+
+		for _, fn := range funcs {
+			if err := sub.PubSub.SetTyped("fn_"+fn.ID.Hex(), fn); err != nil {
+				log.Println("error adding function  to cache: ", err)
+				return
+			}
+
+			ids = append(ids, fn.ID)
+		}
+
+		sub.PubSub.SetTyped(key, ids)
+	}
+
+	for _, id := range ids {
+		var fn ExecData
+		if err := sub.PubSub.GetTyped("fn_"+id.Hex(), &fn); err != nil {
+			log.Println("error getting function out of cache: ", err)
+			return
+		}
+
+		exe.Data = fn
+		go func(ex ExecutionEnvironment) {
+			if err := ex.Execute(); err != nil {
+				log.Printf(`executing "%s" function failed: %v"`, ex.Data.FunctionName, err)
+			}
+		}(exe)
+	}
+}

hub.go

@@ -106,8 +106,8 @@ func (h *Hub) getTargets(msg internal.Command) (sockets []*Socket, payload inter
 			return
 		}
 
-		_, ok := internal.Tokens[pl.Token]
-		if !ok {
+		var a internal.Auth
+		if err := volatile.GetTyped(pl.Token, &a); err != nil {
 			payload = internal.Command{Type: internal.MsgTypeError, Data: "invalid token"}
 		} else {
 			payload = internal.Command{Type: internal.MsgTypeToken, Data: pl.Token}

internal/account.go

@@ -2,6 +2,7 @@ package internal
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/gbrlsnchs/jwt/v3"
@@ -16,6 +17,11 @@ type Auth struct {
 	UserID    primitive.ObjectID
 	Email     string
 	Role      int
+	Token     string
+}
+
+func (auth Auth) ReconstructToken() string {
+	return fmt.Sprintf("%s|%s", auth.UserID.Hex(), auth.Token)
 }
 
 // JWTPayload contains the current user token

internal/data.go

@@ -16,9 +16,9 @@ type BaseConfig struct {
 }
 
 var (
-	Tokens     map[string]Auth       = make(map[string]Auth)
-	Bases      map[string]BaseConfig = make(map[string]BaseConfig)
-	HashSecret                       = jwt.NewHS256([]byte(os.Getenv("JWT_SECRET")))
+	//Tokens     map[string]Auth       = make(map[string]Auth)
+	//Bases      map[string]BaseConfig = make(map[string]BaseConfig)
+	HashSecret = jwt.NewHS256([]byte(os.Getenv("JWT_SECRET")))
 )
 
 const (
@@ -48,11 +48,12 @@ const (
 )
 
 type Command struct {
-	SID     string `json:"sid"`
-	Type    string `json:"type"`
-	Data    string `json:"data"`
-	Channel string `json:"channel"`
-	Token   string `json:"token"`
+	SID           string `json:"sid"`
+	Type          string `json:"type"`
+	Data          string `json:"data"`
+	Channel       string `json:"channel"`
+	Token         string `json:"token"`
+	IsSystemEvent bool   `json:"-"`
 }
 
 func (msg Command) IsDBEvent() bool {

internal/pubsuber.go

@@ -4,6 +4,8 @@ package internal
 type PubSuber interface {
 	Get(key string) (string, error)
 	Set(key string, value string) error
+	GetTyped(key string, v interface{}) error
+	SetTyped(key string, v interface{}) error
 	Inc(key string, by int64) (int64, error)
 	Dec(key string, by int64) (int64, error)
 	Subscribe(send chan Command, token, channel string, close chan bool)

membership.go

@@ -80,11 +80,16 @@ func login(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	internal.Tokens[token] = internal.Auth{
+	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
 	}
 
 	respond(w, http.StatusOK, string(jwtBytes))
@@ -195,11 +200,15 @@ func createUser(db *mongo.Database, accountID primitive.ObjectID, email, passwor
 		return nil, tok, err
 	}
 
-	internal.Tokens[token] = internal.Auth{
+	auth := internal.Auth{
 		AccountID: tok.AccountID,
 		UserID:    tok.ID,
 		Email:     tok.Email,
 		Role:      role,
+		Token:     tok.Token,
+	}
+	if err := volatile.SetTyped(token, auth); err != nil {
+		return nil, tok, err
 	}
 
 	return jwtBytes, tok, nil
@@ -445,11 +454,16 @@ func sudoGetTokenFromAccountID(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	internal.Tokens[token] = internal.Auth{
+	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
 	}
 
 	respond(w, http.StatusOK, string(jwtBytes))