diff --git a/.env.example b/.env.example index 497921d..5199f75 100644 --- a/.env.example +++ b/.env.example @@ -22,6 +22,7 @@ MQTT_RETAINED="__MQTT_RETAINED__" PROTOCOLS_CHIRPSTACK_ENABLED="__PROTOCOLS_CHIRPSTACK_ENABLED__" PROTOCOLS_TTN_ENABLED="__PROTOCOLS_TTN_ENABLED__" PROTOCOLS_HELIUM_ENABLED="__PROTOCOLS_HELIUM_ENABLED__" +PROTOCOLS_API_ENABLED="__PROTOCOLS_API_ENABLED__" # SMS Protocol Configuration PROTOCOLS_SMS_ENABLED="__PROTOCOLS_SMS_ENABLED__" @@ -38,4 +39,4 @@ PROTOCOLS_WEBSOCKET_PORT="__PROTOCOLS_WEBSOCKET_PORT__" # MQTT Protocol Configuration PROTOCOLS_MQTT_PROTOCOL_ENABLED="__PROTOCOLS_MQTT_PROTOCOL_ENABLED__" -PROTOCOLS_MQTT_PROTOCOL_PORT="__PROTOCOLS_MQTT_PROTOCOL_PORT__" \ No newline at end of file +PROTOCOLS_MQTT_PROTOCOL_PORT="__PROTOCOLS_MQTT_PROTOCOL_PORT__" diff --git a/cmd/mpa/cmd/serve.go b/cmd/mpa/cmd/serve.go index 7b2b3ba..eee7324 100644 --- a/cmd/mpa/cmd/serve.go +++ b/cmd/mpa/cmd/serve.go @@ -30,6 +30,7 @@ import ( "github.com/Space-DF/mpa-service/internal/mqtt" "github.com/Space-DF/mpa-service/internal/protocols/handlers" "github.com/Space-DF/mpa-service/internal/protocols/lorawan" + nonlorawan "github.com/Space-DF/mpa-service/internal/protocols/non-lorawan" mqttprotocol "github.com/Space-DF/mpa-service/internal/protocols/transport/mqtt" "github.com/Space-DF/mpa-service/internal/protocols/transport/socketio" "github.com/Space-DF/mpa-service/internal/protocols/transport/websocket" @@ -157,6 +158,14 @@ func runServe(cmd *cobra.Command, args []string) { transportCount++ } + // 4. API HTTP Transport for non-LoRaWAN devices + if cfg.Protocols.API.Enabled { + apiHandler := nonlorawan.NewHandler(deviceService, logger) + handlerManager.Register(apiHandler) + logger.Infof("Registered API transport handler at path: %s", apiHandler.Path()) + transportCount++ + } + // 2. WebSocket Transport if cfg.Protocols.WebSocket.Enabled { wsHandler := websocket.NewHandler(deviceService, websocket.Config{ @@ -243,12 +252,12 @@ func runServe(cmd *cobra.Command, args []string) { e.Use(otelecho.Middleware("mpa-service")) // Security and operational middleware - e.Use(middleware.Logger()) + e.Use(middleware.RequestLogger()) e.Use(middleware.Recover()) e.Use(middleware.CORS()) // Request timeout middleware (prevent slow loris attacks) - e.Use(middleware.TimeoutWithConfig(middleware.TimeoutConfig{ + e.Use(middleware.ContextTimeoutWithConfig(middleware.ContextTimeoutConfig{ Timeout: 30 * time.Second, })) diff --git a/configs/config.yaml b/configs/config.yaml index df3059b..5e9aedd 100644 --- a/configs/config.yaml +++ b/configs/config.yaml @@ -49,6 +49,9 @@ protocols: helium: enabled: false + api: + enabled: true + # OpenTelemetry Configuration opentelemetry: endpoint: "signoz-otel-collector:4317" diff --git a/go.mod b/go.mod index 2046b9c..b064eae 100644 --- a/go.mod +++ b/go.mod @@ -7,7 +7,7 @@ require ( github.com/googollee/go-socket.io v1.7.0 github.com/gorilla/websocket v1.5.3 github.com/joho/godotenv v1.5.1 - github.com/labstack/echo/v4 v4.13.4 + github.com/labstack/echo/v4 v4.15.3 github.com/spf13/cobra v1.9.1 github.com/spf13/viper v1.20.1 go.opentelemetry.io/contrib/instrumentation/github.com/labstack/echo/otelecho v0.63.0 @@ -33,9 +33,9 @@ require ( github.com/google/uuid v1.6.0 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.3 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect - github.com/labstack/gommon v0.4.2 // indirect + github.com/labstack/gommon v0.5.0 // indirect github.com/mattn/go-colorable v0.1.14 // indirect - github.com/mattn/go-isatty v0.0.20 // indirect + github.com/mattn/go-isatty v0.0.22 // indirect github.com/pelletier/go-toml/v2 v2.2.3 // indirect github.com/sagikazarmark/locafero v0.7.0 // indirect github.com/sourcegraph/conc v0.3.0 // indirect @@ -56,7 +56,7 @@ require ( golang.org/x/sync v0.22.0 // indirect golang.org/x/sys v0.47.0 // indirect golang.org/x/text v0.41.0 // indirect - golang.org/x/time v0.12.0 // indirect + golang.org/x/time v0.15.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect google.golang.org/grpc v1.83.2 // indirect diff --git a/go.sum b/go.sum index ff5b5da..788df9b 100644 --- a/go.sum +++ b/go.sum @@ -48,14 +48,14 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= -github.com/labstack/echo/v4 v4.13.4 h1:oTZZW+T3s9gAu5L8vmzihV7/lkXGZuITzTQkTEhcXEA= -github.com/labstack/echo/v4 v4.13.4/go.mod h1:g63b33BZ5vZzcIUF8AtRH40DrTlXnx4UMC8rBdndmjQ= -github.com/labstack/gommon v0.4.2 h1:F8qTUNXgG1+6WQmqoUWnz8WiEU60mXVVw0P4ht1WRA0= -github.com/labstack/gommon v0.4.2/go.mod h1:QlUFxVM+SNXhDL/Z7YhocGIBYOiwB0mXm1+1bAPHPyU= +github.com/labstack/echo/v4 v4.15.3 h1:lIdG4kK5RdMyhwCwSc4AmSQsLBb3AVwok6S8PX/9kwQ= +github.com/labstack/echo/v4 v4.15.3/go.mod h1:Xzp1Ns1RA2c9fY7nSgUJkpkUZGNbEIVHZbtbOMPktBI= +github.com/labstack/gommon v0.5.0 h1:6VSQ2NOzsnEJ5W6+84E0RbcaDDmgB6NIAzWCczTEe6c= +github.com/labstack/gommon v0.5.0/go.mod h1:Rzlg7HHy1maLfzBYGg9NZcVuz1sA68HHhLjhcEllYE0= github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE= github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8= -github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= -github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/mattn/go-isatty v0.0.22 h1:j8l17JJ9i6VGPUFUYoTUKPSgKe/83EYU2zBC7YNKMw4= +github.com/mattn/go-isatty v0.0.22/go.mod h1:ZXfXG4SQHsB/w3ZeOYbR0PrPwLy+n6xiMrJlRFqopa4= github.com/pelletier/go-toml/v2 v2.2.3 h1:YmeHyLY8mFWbdkNWwpr+qIL2bEqT0o95WSdkNHvL12M= github.com/pelletier/go-toml/v2 v2.2.3/go.mod h1:MfCQTFTvCcUyyvvwm1+G6H/jORL20Xlb6rzQu9GuUkc= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= @@ -130,13 +130,12 @@ golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= -golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= -golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE= -golang.org/x/time v0.12.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa h1:Kjn0N0tCrDgiAFW+lGO4JZ3ck44CehvJQMAwj9QF0G8= diff --git a/internal/config/config.go b/internal/config/config.go index 35a295a..b507120 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -32,6 +32,7 @@ type ProtocolsConfig struct { ChirpStack ChirpStackConfig `mapstructure:"chirpstack"` TTN TTNConfig `mapstructure:"ttn"` Helium HeliumConfig `mapstructure:"helium"` + API APIConfig `mapstructure:"api"` } // Protocol-specific configurations @@ -67,6 +68,10 @@ type HeliumConfig struct { Enabled bool `mapstructure:"enabled" env:"PROTOCOLS_HELIUM_ENABLED"` } +type APIConfig struct { + Enabled bool `mapstructure:"enabled" env:"PROTOCOLS_API_ENABLED"` +} + // LoRaWANConfig defines configuration for LoRaWAN providers type LoRaWANConfig struct { Providers map[string]LoRaWANProviderConfig `mapstructure:"providers"` @@ -153,6 +158,7 @@ func setDefaults(vp *viper.Viper) { vp.SetDefault("protocols.chirpstack.enabled", false) vp.SetDefault("protocols.ttn.enabled", false) vp.SetDefault("protocols.helium.enabled", false) + vp.SetDefault("protocols.api.enabled", true) // OpenTelemetry defaults vp.SetDefault("opentelemetry.enabled", false) diff --git a/internal/protocols/non-lorawan/handler.go b/internal/protocols/non-lorawan/handler.go new file mode 100644 index 0000000..fc7a2be --- /dev/null +++ b/internal/protocols/non-lorawan/handler.go @@ -0,0 +1,160 @@ +package nonlorawan + +import ( + "encoding/base64" + "encoding/json" + "io" + "net/http" + "regexp" + "strings" + "time" + + "github.com/Space-DF/mpa-service/internal/logger" + "github.com/Space-DF/mpa-service/internal/protocols/handlers" + "github.com/Space-DF/mpa-service/internal/services" + "github.com/labstack/echo/v4" +) + +// Handler implements the HTTP API ingress for non-LoRaWAN devices. +type Handler struct { + deviceService *services.DeviceService + logger *logger.Logger +} + +const maxRequestSize = 1024 * 1024 + +var tenantIDPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{1,64}$`) + +type RequestPayload struct { + SerialNumber string `json:"serialNumber"` + Payload string `json:"payload"` + ReceivedAt string `json:"received_at,omitempty"` +} + +func NewHandler(deviceService *services.DeviceService, logger *logger.Logger) handlers.ProtocolHandler { + return &Handler{ + deviceService: deviceService, + logger: logger, + } +} + +func (h Handler) Name() string { + return "api" +} + +func (h Handler) Path() string { + return "/api/http" +} + +func (h Handler) Method() string { + return "POST" +} + +func (h *Handler) Handle(c echo.Context) error { + startTime := time.Now() + + contentType := c.Request().Header.Get("Content-Type") + if contentType != "" && !strings.HasPrefix(contentType, "application/json") { + h.logger.Warnf("api: invalid content type: %s", contentType) + return echo.NewHTTPError(http.StatusBadRequest, "Content-Type must be application/json") + } + + c.Request().Body = http.MaxBytesReader(c.Response().Writer, c.Request().Body, maxRequestSize) + body, err := io.ReadAll(c.Request().Body) + if err != nil { + h.logger.Errorf("api: error reading request body: %v", err) + if strings.Contains(err.Error(), "http: request body too large") { + return echo.NewHTTPError(http.StatusRequestEntityTooLarge, "Request body too large") + } + return echo.NewHTTPError(http.StatusBadRequest, "Failed to read request body") + } + + var requestPayload RequestPayload + if err := json.Unmarshal(body, &requestPayload); err != nil { + h.logger.Warnf("api: invalid JSON payload: %v", err) + return echo.NewHTTPError(http.StatusBadRequest, "Invalid JSON payload") + } + + serialNumber := strings.TrimSpace(requestPayload.SerialNumber) + if serialNumber == "" { + return echo.NewHTTPError(http.StatusBadRequest, "serialNumber is required") + } + + encodedPayload := strings.TrimSpace(requestPayload.Payload) + if encodedPayload == "" { + return echo.NewHTTPError(http.StatusBadRequest, "payload is required") + } + if _, err := base64.StdEncoding.DecodeString(encodedPayload); err != nil { + return echo.NewHTTPError(http.StatusBadRequest, "payload must be a valid base64 string") + } + + tenantID := extractTenantID(c.Request().Host) + if tenantID == "" { + h.logger.Errorf("api: invalid organization slug") + return echo.NewHTTPError(http.StatusBadRequest, "Invalid subdomain: must include organization slug (e.g., {org-slug}.localhost)") + } + + receivedAt := strings.TrimSpace(requestPayload.ReceivedAt) + if receivedAt == "" { + receivedAt = startTime.UTC().Format(time.RFC3339) + } + + transportMetadata := map[string]interface{}{ + "transport": "http", + "api_source": "api", + "serial_number": serialNumber, + "tenant_id": tenantID, + "received_at": receivedAt, + } + + if err := h.deviceService.ProcessHTTPMessage(c.Request(), body, transportMetadata); err != nil { + h.logger.Errorf("api: error processing message: %v", err) + return echo.NewHTTPError(http.StatusInternalServerError, "Failed to process message") + } + + return c.JSON(http.StatusOK, map[string]interface{}{ + "status": "success", + "source": "api", + "serialNumber": serialNumber, + "message": "Message processed successfully", + "processing_time_ms": time.Since(startTime).Milliseconds(), + }) +} + +func (h *Handler) HealthCheck(c echo.Context) error { + healthStatus := h.deviceService.GetHealthStatus() + + if !healthStatus["mqtt_connected"].(bool) { + return echo.NewHTTPError(http.StatusServiceUnavailable, "MQTT client not connected") + } + + return c.JSON(http.StatusOK, map[string]interface{}{ + "source": "api", + "status": "healthy", + "endpoint": h.Path(), + "mqtt": "connected", + "timestamp": time.Now().UTC().Format(time.RFC3339), + }) +} + +func extractTenantID(host string) string { + host = strings.TrimSpace(host) + if host == "" { + return "" + } + + if colonIndex := strings.Index(host, ":"); colonIndex != -1 { + host = host[:colonIndex] + } + + parts := strings.Split(host, ".") + if len(parts) < 2 { + return "" + } + + tenantID := strings.TrimSpace(parts[0]) + if tenantIDPattern.MatchString(tenantID) { + return tenantID + } + return "" +} diff --git a/internal/services/device_service.go b/internal/services/device_service.go index 9e17769..d3ded21 100644 --- a/internal/services/device_service.go +++ b/internal/services/device_service.go @@ -87,7 +87,9 @@ func (ds *DeviceService) forwardToMQTT(payload []byte, request *http.Request, me if mqttMessage.Metadata == nil { mqttMessage.Metadata = make(map[string]interface{}) } - mqttMessage.Metadata["request_path"] = request.URL.Path + if request.URL != nil { + mqttMessage.Metadata["request_path"] = request.URL.Path + } mqttMessage.Metadata["request_method"] = request.Method if userAgent := request.Header.Get("User-Agent"); userAgent != "" { mqttMessage.Metadata["user_agent"] = userAgent