First implementation of React UI

.gitignore

@@ -16,4 +16,21 @@
 
 
 # IDE configuration
-.idea
+.idea
+
+# misc
+.DS_Store
+.vscode
+
+*.log
+
+_bin/
+
+go.work*
+
+# temp testing
+hive/  
+
+*__debug_bin*
+yarn.lock
+node_modules

api/bridge_handler.go

@@ -1,18 +1,19 @@
 package api
 
 import (
-	"bytes"
 	"context"
-	"encoding/binary"
+	"encoding/json"
 	"fmt"
+	"log"
+	"net/http"
+	"sync"
+
 	"github.com/go-chi/chi/v5"
 	"github.com/ledgerwatch/diagnostics"
 	"github.com/ledgerwatch/diagnostics/api/internal"
+	"github.com/ledgerwatch/diagnostics/internal/erigon_node"
 	"github.com/ledgerwatch/diagnostics/internal/sessions"
 	"github.com/pkg/errors"
-	"io"
-	"log"
-	"net/http"
 )
 
 var _ http.Handler = &UIHandler{}
@@ -23,112 +24,137 @@ type BridgeHandler struct {
 }
 
 func (h BridgeHandler) Bridge(w http.ResponseWriter, r *http.Request) {
-	pin, err := retrievePinFromURL(r)
-	if err != nil {
-		http.Error(w, "Error parsing session PIN", http.StatusBadRequest)
-	}
-
-	nodeSession, ok := h.cache.FindNodeSession(pin)
-	if !ok {
-		log.Printf("Session with specified PIN %d not found\n", pin)
-		internal.EncodeError(w, r, diagnostics.AsBadRequestErr(errors.Errorf("Session with specified PIN %d not found", pin)))
-		return
-	}
 
 	//Sends a success Message to the Node client, to receive more information
 	flusher, _ := w.(http.Flusher)
-	fmt.Fprintf(w, "SUCCESS\n")
 	ctx, cancel := context.WithCancel(r.Context())
 	defer cancel()
 	defer r.Body.Close()
 
-	nodeSession.Connect(r.RemoteAddr)
-	defer nodeSession.Disconnect()
-
 	// Update the request context with the connection context.
 	// If the connection is closed by the server, it will also notify everything that waits on the request context.
 	*r = *r.WithContext(ctx)
 
 	w.WriteHeader(http.StatusOK)
 	flusher.Flush()
 
-	var versionBytes [8]byte
-	if _, err := io.ReadFull(r.Body, versionBytes[:]); err != nil {
-		log.Printf("Error reading version bytes: %v\n", err)
-		internal.EncodeError(w, r, diagnostics.AsBadRequestErr(errors.Errorf("Error reading version bytes: %v", err)))
+	connectionInfo := struct {
+		Version  uint64               `json:"version"`
+		Sessions []string             `json:"sessions"`
+		Nodes    []*sessions.NodeInfo `json:"nodes"`
+	}{}
+
+	err := json.NewDecoder(r.Body).Decode(&connectionInfo)
+
+	if err != nil {
+		log.Printf("Error reading connection info: %v\n", err)
+		internal.EncodeError(w, r, diagnostics.AsBadRequestErr(errors.Errorf("Error reading connection info: %v", err)))
 		return
 	}
 
-	nodeSession.SupportVersion = binary.BigEndian.Uint64(versionBytes[:])
-
-	// Read data from the request body
-	var writeBuf bytes.Buffer
-	for request := range nodeSession.RequestCh {
-		request.Lock.Lock()
-		url := request.Url
-		request.Lock.Unlock()
-		log.Printf("Sending request %s\n", url)
-		writeBuf.Reset()
-		fmt.Fprint(&writeBuf, url)
-		if _, err := w.Write(writeBuf.Bytes()); err != nil {
-			log.Printf("Writing metrics request: %v\n", err)
-			request.Lock.Lock()
-			request.Served = true
-			request.Response = nil
-			request.Err = fmt.Sprintf("writing metrics request: %v", err)
-			request.Retries++
-			if request.Retries < 15 {
-				select {
-				case nodeSession.RequestCh <- request:
-				default:
-				}
+	requestMap := map[string]*erigon_node.NodeRequest{}
+	requestMutex := sync.Mutex{}
+
+	for _, node := range connectionInfo.Nodes {
+		nodeSession, ok := h.cache.FindNodeSession(node.Id)
+
+		if !ok {
+			nodeSession, err = h.cache.CreateNodeSession(node)
+
+			if err != nil {
+				log.Printf("Error creating node session: %v\n", err)
+				internal.EncodeError(w, r, diagnostics.AsBadRequestErr(fmt.Errorf("error creating node session: %w", err)))
+				return
+
 			}
-			request.Lock.Unlock()
-			return
 		}
-		flusher.Flush()
-		// Read the response
-		var sizeBuf [4]byte
-		if _, err := io.ReadFull(r.Body, sizeBuf[:]); err != nil {
-			log.Printf("Reading size of metrics response: %v\n", err)
-			request.Lock.Lock()
-			request.Served = true
-			request.Response = nil
-			request.Err = fmt.Sprintf("reading size of metrics response: %v", err)
-			request.Retries++
-			if request.Retries < 15 {
-				select {
-				case nodeSession.RequestCh <- request:
-				default:
+
+		nodeSession.AttachSessions(connectionInfo.Sessions)
+
+		nodeSession.Connect(r.RemoteAddr)
+
+		go func() {
+			defer nodeSession.Disconnect()
+
+			for request := range nodeSession.RequestCh {
+				rpcRequest := request.Request
+
+				bytes, err := json.Marshal(rpcRequest)
+
+				if err != nil {
+					request.Responses <- &erigon_node.Response{
+						Last: true,
+						Error: &erigon_node.Error{
+							Message: fmt.Errorf("failed to marshal request: %w", err).Error(),
+						},
+					}
+					continue
 				}
-			}
-			request.Lock.Unlock()
-			return
-		}
-		metricsBuf := make([]byte, binary.BigEndian.Uint32(sizeBuf[:]))
-		if _, err := io.ReadFull(r.Body, metricsBuf); err != nil {
-			log.Printf("Reading metrics response: %v\n", err)
-			request.Lock.Lock()
-			request.Served = true
-			request.Response = nil
-			request.Err = fmt.Sprintf("reading metrics response: %v", err)
-			request.Retries++
-			if request.Retries < 15 {
-				select {
-				case nodeSession.RequestCh <- request:
-				default:
+
+				//fmt.Printf("Sending request %s\n", string(bytes))
+
+				requestMutex.Lock()
+				requestMap[rpcRequest.Id] = request
+				requestMutex.Unlock()
+
+				if _, err := w.Write(bytes); err != nil {
+					requestMutex.Lock()
+					delete(requestMap, rpcRequest.Id)
+					requestMutex.Unlock()
+
+					fmt.Println(request.Retries, err)
+					request.Retries++
+					if request.Retries < 15 {
+						select {
+						case nodeSession.RequestCh <- request:
+						default:
+						}
+					} else {
+						request.Responses <- &erigon_node.Response{
+							Last: true,
+							Error: &erigon_node.Error{
+								Message: fmt.Errorf("failed to write metrics request: %w", err).Error(),
+							},
+						}
+					}
+					continue
 				}
+
+				flusher.Flush()
 			}
-			request.Lock.Unlock()
-			return
-		}
-		request.Lock.Lock()
-		request.Served = true
-		request.Response = metricsBuf
-		request.Err = ""
-		request.Lock.Unlock()
+		}()
 	}
 
+	decoder := json.NewDecoder(r.Body)
+
+	for {
+		var response erigon_node.Response
+
+		if err = decoder.Decode(&response); err != nil {
+			fmt.Printf("can't read response: %v\n", err)
+			continue
+		}
+
+		requestMutex.Lock()
+		request, ok := requestMap[response.Id]
+		requestMutex.Unlock()
+
+		if !ok {
+			continue
+		}
+
+		if response.Error != nil {
+			response.Last = true
+		}
+
+		request.Responses <- &response
+
+		if response.Last {
+			requestMutex.Lock()
+			delete(requestMap, response.Id)
+			requestMutex.Unlock()
+		}
+	}
 }
 
 func NewBridgeHandler(cacheSvc sessions.CacheService) BridgeHandler {
@@ -137,7 +163,7 @@ func NewBridgeHandler(cacheSvc sessions.CacheService) BridgeHandler {
 		cache:  cacheSvc,
 	}
 
-	r.Post("/{pin}", r.Bridge)
+	r.Post("/", r.Bridge)
 
 	return *r
 }

api/helpers.go

@@ -1,36 +1,30 @@
 package api
 
-import (
-	"fmt"
-	"log"
-	"net/http"
-	"net/url"
-	"path"
-	"strconv"
-)
-
-func retrievePinFromURL(r *http.Request) (uint64, error) {
-	parsedURL, err := url.Parse(r.URL.Path)
-	if err != nil {
-		log.Println("Error parsing URL:", err)
-		return 0, fmt.Errorf("Error parsing URL: %w", err)
+/*func retrievePinFromURL(r *http.Request) (pins []uint64, err error) {
+
+	for _, session := range strings.Split(r.URL.Query().Get("sessions"), ",") {
+		pin, err := strconv.ParseUint(session, 10, 64)
+
+		if err != nil {
+			log.Printf("Error parsing session pin %s: %v\n", session, err)
+			return pins, fmt.Errorf("error parsing session pin %s: %w", session, err)
+		}
+
+		pins = append(pins, pin)
 	}
 
-	lastPathItem := path.Base(parsedURL.Path)
-	pin, err := strconv.ParseUint(lastPathItem, 10, 64)
-	if err != nil {
-		log.Printf("Error parsing session pin %s: %v\n", lastPathItem, err)
-		return 0, fmt.Errorf("Error parsing session pin %s: %w", lastPathItem, err)
+	if len(pins) == 0 {
+		err = fmt.Errorf("no sessions")
 	}
 
-	return pin, nil
+	return pins, err
 }
 
 func retrieveSizeStrFrom(r *http.Request) (uint64, error) {
 	sizeStr := r.Form.Get("size")
 	size, err := strconv.ParseUint(sizeStr, 10, 64)
 	if err != nil {
-		return 0, fmt.Errorf("Parsing size %s: %w", sizeStr, err)
+		return 0, fmt.Errorf("parsing size %s: %w", sizeStr, err)
 	}
 
 	var offset uint64
@@ -39,4 +33,4 @@ func retrieveSizeStrFrom(r *http.Request) (uint64, error) {
 	}
 
 	return offset, nil
-}
+}*/

api/internal/end_points.go

@@ -2,7 +2,5 @@ package internal
 
 const (
 	HealthCheckEndPoint = "/healthcheck"
-	UIEndPoint          = "/ui"
-	ScriptEndPoint      = "/script"
-	SupportEndPoint     = "/support"
+	BridgeEndPoint      = "/bridge"
 )

api/internal/error.go

@@ -2,8 +2,9 @@ package internal
 
 import (
 	"encoding/json"
-	"github.com/ledgerwatch/diagnostics"
 	"net/http"
+
+	"github.com/ledgerwatch/diagnostics"
 )
 
 type Error struct {
@@ -24,7 +25,7 @@ func marshalError(r *http.Request, err error) Error {
 
 	} else {
 		code = http.StatusInternalServerError
-		message = "internal server error"
+		message = err.Error()
 	}
 
 	return Error{