Working sync engine!

engine.go

@@ -4,7 +4,9 @@ import (
 	"encoding/json"
 	"log/slog"
 	"net/http"
+	"sync"
 
+	"github.com/cespare/xxhash"
 	"github.com/wisplite/tether/reactivity"
 	"gorm.io/gorm"
 )
@@ -12,9 +14,11 @@ import (
 type Engine struct {
 	db           *gorm.DB
 	dbType       string // sqlite or postgres
-	mutations    map[string]func(ctx *MutationCtx) error
-	queries      map[string]func(ctx *QueryCtx) error
+	mutations    map[string]func(ctx *MutationCtx) interface{}
+	queries      map[string]func(ctx *QueryCtx) interface{}
 	dependencies map[string][]string
+	hashMu       sync.RWMutex
+	queryHashes  map[string]uint64
 	tracker      *reactivity.Tracker
 }
 
@@ -24,7 +28,7 @@ func NewEngine(db *gorm.DB, dbType string) *Engine {
 	if dbType != "sqlite" && dbType != "postgres" {
 		panic("Invalid database type")
 	}
-	e := &Engine{db: db, dbType: dbType, mutations: make(map[string]func(ctx *MutationCtx) error), queries: make(map[string]func(ctx *QueryCtx) error), dependencies: make(map[string][]string), tracker: tracker}
+	e := &Engine{db: db, dbType: dbType, mutations: make(map[string]func(ctx *MutationCtx) interface{}), queries: make(map[string]func(ctx *QueryCtx) interface{}), dependencies: make(map[string][]string), queryHashes: make(map[string]uint64), tracker: tracker}
 	db.Callback().Create().After("gorm:create").Register("tether:after_create", func(tx *gorm.DB) {
 		if dbType == "postgres" {
 			return
@@ -34,12 +38,12 @@ func NewEngine(db *gorm.DB, dbType string) *Engine {
 	return e
 }
 
-func (e *Engine) RegisterMutation(name string, mutation func(ctx *MutationCtx) error) {
+func (e *Engine) RegisterMutation(name string, mutation func(ctx *MutationCtx) interface{}) {
 	e.mutations[name] = mutation // stores the mutation in the list of valid mutations
 	slog.Debug("Registered mutation", "name", name)
 }
 
-func (e *Engine) RegisterQuery(name string, query func(ctx *QueryCtx) error, dependencies []string) {
+func (e *Engine) RegisterQuery(name string, query func(ctx *QueryCtx) interface{}, dependencies []string) {
 	e.queries[name] = query // stores the query in the list of valid queries
 	for _, dependency := range dependencies {
 		e.dependencies[dependency] = append(e.dependencies[dependency], name)
@@ -78,11 +82,16 @@ func (e *Engine) InvalidateTable(tableName string) {
 	for _, query := range dependentQueries {
 		slog.Debug("Invalidating query", "query", query)
 		subscriptions := e.tracker.GetQuerySubscriptions(query)
+		slog.Debug("Subscriptions", "subscriptions", subscriptions)
 		for _, subscription := range subscriptions {
 			slog.Debug("Invalidating subscription", "subscription", subscription["clientID"])
 			params := map[string]interface{}{}
-			json.Unmarshal([]byte(subscription["params"]), &params)
-			_, err := e.ExecuteQuery(query, params, subscription["clientID"])
+			err := json.Unmarshal([]byte(subscription["params"]), &params)
+			if err != nil {
+				slog.Error("Failed to unmarshal params", "error", err)
+				continue
+			}
+			_, err = e.ExecuteQuery(query, params, subscription["clientID"])
 			if err != nil {
 				slog.Error("Failed to execute query", "error", err)
 				continue
@@ -101,9 +110,31 @@ func (e *Engine) ExecuteQuery(query string, params map[string]interface{}, clien
 		4. Calculate hash for every query
 		5. Send the updated queries if hash changed
 	*/
+	paramsJSON, err := json.Marshal(params)
+	if err != nil {
+		return nil, err
+	}
+	cacheKey := query + "?" + string(paramsJSON)
+	e.hashMu.Lock()
+	lastHash := e.queryHashes[cacheKey]
+	e.hashMu.Unlock()
 	slog.Debug("Executing query", "query", query, "params", params)
-	e.queries[query](&QueryCtx{DB: e.db, AuthCtx: &AuthCtx{UserID: "", IsLoggedIn: true}, Params: params})
-	return nil, nil
+	result := e.queries[query](&QueryCtx{DB: e.db, AuthCtx: &AuthCtx{UserID: "", IsLoggedIn: true}, Params: params})
+	responseJSON, err := json.Marshal(map[string]interface{}{"type": "query", "location": query, "data": result})
+	if err != nil {
+		return nil, err
+	}
+	queryHash := xxhash.Sum64(responseJSON)
+	if lastHash == queryHash {
+		return result, nil
+	}
+
+	e.hashMu.Lock()
+	e.queryHashes[cacheKey] = queryHash
+	e.hashMu.Unlock()
+
+	e.tracker.SendMessage(clientID, responseJSON)
+	return result, nil
 }
 
 func (e *Engine) ExecuteMutation(mutation string, params map[string]interface{}, clientID string) (interface{}, error) {
@@ -114,8 +145,12 @@ func (e *Engine) ExecuteMutation(mutation string, params map[string]interface{},
 func (e *Engine) OnReceiveMessage(clientID string, msg map[string]interface{}) error {
 	slog.Debug("Received message", "from", clientID, "message", msg)
 	switch msg["type"] {
-	case "query":
-		e.tracker.SubscribeToQuery(clientID, msg["location"].(string), msg["params"].(map[string]string))
+	case "subscribe":
+		paramsJSON, err := json.Marshal(msg["params"])
+		if err != nil {
+			return err
+		}
+		e.tracker.SubscribeToQuery(clientID, msg["location"].(string), string(paramsJSON))
 	case "mutation":
 		e.ExecuteMutation(msg["location"].(string), msg["params"].(map[string]interface{}), clientID)
 	}