send incoming websocket data to engine

engine.go

@@ -36,5 +36,10 @@ func (e *Engine) CreateTable(name string, schema interface{}) {
 }
 
 func (e *Engine) Handle(w http.ResponseWriter, r *http.Request) {
-	reactivity.Handle(w, r)
+	reactivity.Handle(w, r, e)
+}
+
+func (e *Engine) OnReceiveMessage(msg map[string]interface{}) error {
+	slog.Debug("Received message", "message", msg)
+	return nil
 }

reactivity/websocket.go

@@ -1,18 +1,24 @@
 package reactivity
 
 import (
+	"encoding/json"
 	"log/slog"
 	"net/http"
 
 	"github.com/gorilla/websocket"
 )
 
+// MessageReceiver receives decoded WebSocket payloads. Implemented by tether.Engine.
+type MessageReceiver interface {
+	OnReceiveMessage(msg map[string]interface{}) error
+}
+
 var upgrader = websocket.Upgrader{}
 
-func Handle(w http.ResponseWriter, r *http.Request) {
+func Handle(w http.ResponseWriter, r *http.Request, e MessageReceiver) {
 	ws, err := upgrader.Upgrade(w, r, nil)
 	if err != nil {
-		slog.Error("Failed to upgrade to websocket", "error", err)
+		slog.Error("WS: Failed to upgrade to websocket", "error", err)
 		return
 	}
 	defer ws.Close()
@@ -20,9 +26,22 @@ func Handle(w http.ResponseWriter, r *http.Request) {
 	for {
 		_, message, err := ws.ReadMessage()
 		if err != nil {
-			slog.Error("Failed to read message", "error", err)
+			slog.Error("WS: Failed to read message", "error", err)
+			return
+		}
+		slog.Debug("WS: Received message", "message", string(message))
+
+		var msg map[string]interface{}
+		err = json.Unmarshal(message, &msg)
+		if err != nil {
+			slog.Error("WS: Failed to unmarshal message", "error", err)
+			return
+		}
+		slog.Debug("WS: Unmarshalled message", "message", msg)
+		err = e.OnReceiveMessage(msg)
+		if err != nil {
+			slog.Error("WS: Failed to on receive message", "error", err)
 			return
 		}
-		slog.Debug("Received message", "message", string(message))
 	}
 }