work towards decoupling api from router

internal/api/api.go

@@ -0,0 +1,274 @@
+package api
+
+import (
+	"context"
+	_ "embed"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/jwetzell/showbridge-go/internal/common"
+	"github.com/jwetzell/showbridge-go/internal/config"
+	"github.com/jwetzell/showbridge-go/internal/schema"
+)
+
+type ApiServer struct {
+	config             config.ApiConfig
+	serverMu           sync.Mutex
+	server             *http.Server
+	shutdown           context.CancelFunc
+	logger             *slog.Logger
+	configurableRouter config.Configurable
+	eventRouter        common.EventRouter
+}
+
+func NewApiServer(configurableRouter config.Configurable, eventRouter common.EventRouter) *ApiServer {
+	return &ApiServer{
+		configurableRouter: configurableRouter,
+		eventRouter:        eventRouter,
+		logger:             slog.Default().With("component", "api"),
+	}
+}
+
+func (as *ApiServer) Start(config config.ApiConfig) {
+	as.config = config
+	if !as.config.Enabled {
+		as.logger.Warn("not enabled")
+		return
+	}
+	as.logger.Debug("starting", "port", as.config.Port)
+	mux := http.NewServeMux()
+	mux.HandleFunc("/ws", as.handleWebsocket)
+	mux.HandleFunc("/health", as.handleHealthHTTP)
+	mux.HandleFunc("/api/v1/config", as.handleConfigHTTP)
+	mux.HandleFunc("/schema/config.schema.json", handleConfigSchema)
+	mux.HandleFunc("/schema/routes.schema.json", handleRoutesSchema)
+	mux.HandleFunc("/schema/modules.schema.json", handleModulesSchema)
+	mux.HandleFunc("/schema/processors.schema.json", handleProcessorsSchema)
+
+	as.serverMu.Lock()
+	defer as.serverMu.Unlock()
+	as.server = &http.Server{
+		Addr:    fmt.Sprintf(":%d", as.config.Port),
+		Handler: mux,
+	}
+
+	go func() {
+		as.server.ListenAndServe()
+		as.shutdown()
+	}()
+}
+
+func (as *ApiServer) Stop() {
+	if as.server == nil {
+		return
+	}
+	as.logger.Debug("stopping")
+	as.serverMu.Lock()
+	defer as.serverMu.Unlock()
+	if as.server != nil {
+		apiShutdownCtx, apiShutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
+		as.shutdown = apiShutdownCancel
+		as.server.Shutdown(apiShutdownCtx)
+		<-apiShutdownCtx.Done()
+		as.server = nil
+	}
+}
+
+func (as *ApiServer) handleHealthHTTP(w http.ResponseWriter, req *http.Request) {
+	switch req.Method {
+	case http.MethodGet:
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.WriteHeader(http.StatusOK)
+	case http.MethodOptions:
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.Header().Set("Access-Control-Allow-Methods", "GET, OPTIONS")
+		w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
+		w.WriteHeader(http.StatusOK)
+	default:
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.WriteHeader(http.StatusMethodNotAllowed)
+	}
+}
+
+func (as *ApiServer) handleConfigHTTP(w http.ResponseWriter, req *http.Request) {
+
+	switch req.Method {
+	case http.MethodGet:
+		configJSON, err := json.Marshal(as.configurableRouter.GetRunningConfig())
+		if err != nil {
+			http.Error(w, "Internal server error", http.StatusInternalServerError)
+			return
+		}
+
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.Header().Set("Content-Type", "application/json")
+		w.Write(configJSON)
+	case http.MethodPut:
+		//TODO(jwetzell): again way too much marshaling
+		cfgBytes, err := io.ReadAll(req.Body)
+		if err != nil {
+			http.Error(w, "Bad request", http.StatusBadRequest)
+			return
+		}
+		cfgMap := make(map[string]any)
+		err = json.Unmarshal(cfgBytes, &cfgMap)
+		if err != nil {
+			http.Error(w, "Bad request", http.StatusBadRequest)
+			return
+		}
+
+		err = schema.ApplyDefaults(&cfgMap)
+
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+
+		err = schema.ValidateConfig(cfgMap)
+
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+
+		validCfgBytes, err := json.Marshal(cfgMap)
+		if err != nil {
+			http.Error(w, "Internal server error", http.StatusInternalServerError)
+			return
+		}
+
+		var newConfig config.Config
+		err = json.Unmarshal(validCfgBytes, &newConfig)
+		if err != nil {
+			http.Error(w, "Bad request", http.StatusBadRequest)
+			return
+		}
+		err, moduleErrors, routeErrors := as.configurableRouter.UpdateConfig(newConfig, true)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusConflict)
+			return
+		}
+		if len(moduleErrors) > 0 || len(routeErrors) > 0 {
+			errorResponse := struct {
+				ModuleErrors []config.ModuleError `json:"moduleErrors,omitempty"`
+				RouteErrors  []config.RouteError  `json:"routeErrors,omitempty"`
+			}{
+				ModuleErrors: moduleErrors,
+				RouteErrors:  routeErrors,
+			}
+			errorResponseJSON, err := json.Marshal(errorResponse)
+			if err != nil {
+				http.Error(w, "Internal server error", http.StatusInternalServerError)
+				return
+			}
+			w.Header().Set("Access-Control-Allow-Origin", "*")
+			w.Header().Set("Content-Type", "application/json")
+			w.WriteHeader(http.StatusBadRequest)
+			w.Write(errorResponseJSON)
+			return
+		}
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.WriteHeader(http.StatusOK)
+	case http.MethodOptions:
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.Header().Set("Access-Control-Allow-Methods", "GET, PUT, OPTIONS")
+		w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
+		w.WriteHeader(http.StatusOK)
+	default:
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.WriteHeader(http.StatusMethodNotAllowed)
+	}
+}
+
+func handleConfigSchema(w http.ResponseWriter, req *http.Request) {
+	switch req.Method {
+	case http.MethodGet:
+		schemaJSON, err := json.Marshal(schema.ConfigSchema)
+		if err != nil {
+			http.Error(w, "Internal server error", http.StatusInternalServerError)
+			return
+		}
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.Header().Set("Content-Type", "application/json")
+		w.Write(schemaJSON)
+	case http.MethodOptions:
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.Header().Set("Access-Control-Allow-Methods", "GET, OPTIONS")
+		w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
+		w.WriteHeader(http.StatusOK)
+	default:
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.WriteHeader(http.StatusMethodNotAllowed)
+	}
+}
+
+func handleRoutesSchema(w http.ResponseWriter, req *http.Request) {
+	switch req.Method {
+	case http.MethodGet:
+		schemaJSON, err := json.Marshal(schema.RoutesConfigSchema)
+		if err != nil {
+			http.Error(w, "Internal server error", http.StatusInternalServerError)
+			return
+		}
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.Header().Set("Content-Type", "application/json")
+		w.Write(schemaJSON)
+	case http.MethodOptions:
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.Header().Set("Access-Control-Allow-Methods", "GET, OPTIONS")
+		w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
+		w.WriteHeader(http.StatusOK)
+	default:
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.WriteHeader(http.StatusMethodNotAllowed)
+	}
+}
+
+func handleModulesSchema(w http.ResponseWriter, req *http.Request) {
+	switch req.Method {
+	case http.MethodGet:
+		schemaJSON, err := json.Marshal(schema.GetModulesSchema())
+		if err != nil {
+			http.Error(w, "Internal server error", http.StatusInternalServerError)
+			return
+		}
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.Header().Set("Content-Type", "application/json")
+		w.Write(schemaJSON)
+	case http.MethodOptions:
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.Header().Set("Access-Control-Allow-Methods", "GET, OPTIONS")
+		w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
+		w.WriteHeader(http.StatusOK)
+	default:
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.WriteHeader(http.StatusMethodNotAllowed)
+	}
+}
+
+func handleProcessorsSchema(w http.ResponseWriter, req *http.Request) {
+	switch req.Method {
+	case http.MethodGet:
+		schemaJSON, err := json.Marshal(schema.GetProcessorsSchema())
+		if err != nil {
+			http.Error(w, "Internal server error", http.StatusInternalServerError)
+			return
+		}
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.Header().Set("Content-Type", "application/json")
+		w.Write(schemaJSON)
+	case http.MethodOptions:
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.Header().Set("Access-Control-Allow-Methods", "GET, OPTIONS")
+		w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
+		w.WriteHeader(http.StatusOK)
+	default:
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.WriteHeader(http.StatusMethodNotAllowed)
+	}
+}

internal/api/websocket.go

@@ -0,0 +1,82 @@
+package api
+
+import (
+	"encoding/json"
+	"net/http"
+
+	"github.com/gorilla/websocket"
+	"github.com/jwetzell/showbridge-go/internal/common"
+)
+
+var upgrader = websocket.Upgrader{
+	CheckOrigin: func(r *http.Request) bool {
+		return true
+	},
+}
+
+type WebsocketEventDestination struct {
+	conn *websocket.Conn
+}
+
+func (d WebsocketEventDestination) Send(event common.Event) error {
+	eventJSON, err := event.ToJSON()
+	if err != nil {
+		return err
+	}
+	return d.conn.WriteMessage(websocket.TextMessage, eventJSON)
+}
+
+func (d WebsocketEventDestination) Is(dest common.EventDestination) bool {
+	other, ok := dest.(WebsocketEventDestination)
+	if !ok {
+		return false
+	}
+	return d.conn == other.conn
+}
+
+func (as *ApiServer) handleWebsocket(w http.ResponseWriter, req *http.Request) {
+	conn, err := upgrader.Upgrade(w, req, nil)
+	if err != nil {
+		as.logger.Error("websocket upgrade error", "error", err)
+		return
+	}
+	defer conn.Close()
+
+	eventDestination := WebsocketEventDestination{conn: conn}
+
+	as.eventRouter.AddEventDestination(eventDestination)
+READ_LOOP:
+	for {
+		messageType, message, err := conn.ReadMessage()
+		if err != nil {
+			_, ok := err.(*websocket.CloseError)
+			if ok {
+				break READ_LOOP
+			}
+		}
+
+		switch messageType {
+		case websocket.TextMessage, websocket.BinaryMessage:
+			event := common.Event{}
+			err = json.Unmarshal(message, &event)
+			if err != nil {
+				as.logger.Error("websocket message unmarshal error", "error", err)
+				continue
+			}
+			as.eventRouter.HandleEvent(event, WebsocketEventDestination{conn: conn})
+		case websocket.CloseMessage:
+			break READ_LOOP
+		case websocket.PingMessage:
+			err = conn.WriteMessage(websocket.PongMessage, nil)
+			if err != nil {
+				as.logger.Error("websocket pong error", "error", err)
+			}
+		default:
+			as.logger.Warn("unsupported websocket message type", "type", messageType)
+			continue
+		}
+
+	}
+	//NOTE(jwetzell): remove ws connection
+	as.eventRouter.RemoveEventDestination(eventDestination)
+}

internal/common/events.go

@@ -0,0 +1,26 @@
+package common
+
+import (
+	"encoding/json"
+)
+
+type Event struct {
+	Type  string `json:"type"`
+	Data  any    `json:"data,omitempty"`
+	Error string `json:"error,omitempty"`
+}
+
+func (e Event) ToJSON() ([]byte, error) {
+	return json.Marshal(e)
+}
+
+type EventDestination interface {
+	Send(event Event) error
+	Is(dest EventDestination) bool
+}
+
+type EventRouter interface {
+	HandleEvent(event Event, source EventDestination)
+	AddEventDestination(dest EventDestination)
+	RemoveEventDestination(dest EventDestination)
+}

internal/common/routing.go

@@ -1,6 +1,8 @@
 package common
 
-import "context"
+import (
+	"context"
+)
 
 type RouteIO interface {
 	HandleInput(ctx context.Context, sourceId string, payload any) (bool, []RouteIOError)

internal/config/config.go

@@ -5,3 +5,8 @@ type Config struct {
 	Modules []ModuleConfig `json:"modules"`
 	Routes  []RouteConfig  `json:"routes"`
 }
+
+type Configurable interface {
+	UpdateConfig(newConfig Config, triggerChangeChannel bool) (error, []ModuleError, []RouteError)
+	GetRunningConfig() Config
+}

internal/config/module.go

@@ -5,3 +5,9 @@ type ModuleConfig struct {
 	Type   string `json:"type"`
 	Params Params `json:"params,omitempty"`
 }
+
+type ModuleError struct {
+	Index  int          `json:"index"`
+	Config ModuleConfig `json:"config"`
+	Error  string       `json:"error"`
+}

internal/config/route.go

@@ -4,3 +4,9 @@ type RouteConfig struct {
 	Input      string            `json:"input"`
 	Processors []ProcessorConfig `json:"processors"`
 }
+
+type RouteError struct {
+	Index  int         `json:"index"`
+	Config RouteConfig `json:"config"`
+	Error  string      `json:"error"`
+}

internal/module/module.go

@@ -10,12 +10,6 @@ import (
 	"github.com/jwetzell/showbridge-go/internal/config"
 )
 
-type ModuleError struct {
-	Index  int                 `json:"index"`
-	Config config.ModuleConfig `json:"config"`
-	Error  string              `json:"error"`
-}
-
 type ModuleRegistration struct {
 	Type         string             `json:"type"`
 	Title        string             `json:"title,omitempty"`

internal/route/route.go

@@ -13,11 +13,6 @@ import (
 	"go.opentelemetry.io/otel/trace"
 )
 
-type RouteError struct {
-	Index  int                `json:"index"`
-	Config config.RouteConfig `json:"config"`
-	Error  string             `json:"error"`
-}
 type Route struct {
 	input      string
 	processors []processor.Processor