refactor: standardize ingest http handler [OM-87] (#3675)

Author: turip

app/common/openmeter_server.go

@@ -107,9 +107,9 @@ func ServerProvisionTopics(conf config.EventsConfiguration) []pkgkafka.TopicConf
 func NewIngestService(
 	collector ingest.Collector,
 	logger *slog.Logger,
-) (*ingest.Service, error) {
-	return &ingest.Service{
+) (ingest.Service, error) {
+	return ingest.NewService(ingest.Config{
 		Collector: collector,
 		Logger:    logger,
-	}, nil
+	})
 }

cmd/server/main.go

@@ -19,10 +19,8 @@ import (
 	"github.com/openmeterio/openmeter/app/common"
 	"github.com/openmeterio/openmeter/app/config"
 	"github.com/openmeterio/openmeter/openmeter/debug"
-	"github.com/openmeterio/openmeter/openmeter/ingest/ingestdriver"
 	"github.com/openmeterio/openmeter/openmeter/ingest/kafkaingest"
 	"github.com/openmeterio/openmeter/openmeter/namespace"
-	"github.com/openmeterio/openmeter/openmeter/namespace/namespacedriver"
 	"github.com/openmeterio/openmeter/openmeter/server"
 	"github.com/openmeterio/openmeter/openmeter/server/router"
 	"github.com/openmeterio/openmeter/pkg/errorsx"
@@ -112,14 +110,6 @@ func main() {
 		os.Exit(1)
 	}
 
-	// Initialize HTTP Ingest handler
-	ingestHandler := ingestdriver.NewIngestEventsHandler(
-		app.IngestService.IngestEvents,
-		namespacedriver.StaticNamespaceDecoder(app.NamespaceManager.GetDefaultNamespace()),
-		nil,
-		errorsx.NewSlogHandler(logger),
-	)
-
 	// Initialize debug connector
 	debugConnector := debug.NewDebugConnector(app.StreamingConnector)
 
@@ -165,7 +155,7 @@ func main() {
 			FeatureConnector:            app.EntitlementRegistry.Feature,
 			GrantConnector:              app.EntitlementRegistry.Grant,
 			GrantRepo:                   app.EntitlementRegistry.GrantRepo,
-			IngestHandler:               ingestHandler,
+			IngestService:               app.IngestService,
 			Logger:                      logger,
 			MeterManageService:          app.MeterManageService,
 			MeterEventService:           app.MeterEventService,

cmd/server/wire.go

@@ -56,7 +56,7 @@ type Application struct {
 	FeatureConnector                 feature.FeatureConnector
 	FeatureFlags                     ffx.Service
 	IngestCollector                  ingest.Collector
-	IngestService                    *ingest.Service
+	IngestService                    ingest.Service
 	KafkaProducer                    *kafka.Producer
 	KafkaMetrics                     *kafkametrics.Metrics
 	KafkaIngestNamespaceHandler      *kafkaingest.NamespaceHandler

cmd/server/wire_gen.go

@@ -0,0 +1,50 @@
+package httpdriver
+
+import (
+	"context"
+	"net/http"
+
+	"github.com/openmeterio/openmeter/pkg/framework/commonhttp"
+	"github.com/openmeterio/openmeter/pkg/framework/transport/httptransport/encoder"
+)
+
+type ErrorInvalidContentType struct {
+	ContentType string
+}
+
+func (e ErrorInvalidContentType) Error() string {
+	// return "invalid content type"
+
+	return "invalid content type: " + e.ContentType
+}
+
+func (e ErrorInvalidContentType) Message() string {
+	return "invalid content type: " + e.ContentType
+}
+
+func (e ErrorInvalidContentType) Details() map[string]any {
+	return map[string]any{
+		"contentType": e.ContentType,
+	}
+}
+
+type ErrorInvalidEvent struct {
+	Err error
+}
+
+func (e ErrorInvalidEvent) Error() string {
+	// return "invalid event"
+
+	return "invalid event: " + e.Err.Error()
+}
+
+func (e ErrorInvalidEvent) Message() string {
+	return "invalid event: " + e.Err.Error()
+}
+
+func errorEncoder() encoder.ErrorEncoder {
+	return func(ctx context.Context, err error, w http.ResponseWriter, r *http.Request) bool {
+		return commonhttp.HandleErrorIfTypeMatches[ErrorInvalidContentType](ctx, http.StatusBadRequest, err, w) ||
+			commonhttp.HandleErrorIfTypeMatches[ErrorInvalidEvent](ctx, http.StatusBadRequest, err, w)
+	}
+}

openmeter/ingest/httpdriver/errors.go

@@ -0,0 +1,47 @@
+package httpdriver
+
+import (
+	"context"
+	"errors"
+	"net/http"
+
+	"github.com/openmeterio/openmeter/openmeter/ingest"
+	"github.com/openmeterio/openmeter/openmeter/namespace/namespacedriver"
+	"github.com/openmeterio/openmeter/pkg/framework/commonhttp"
+	"github.com/openmeterio/openmeter/pkg/framework/transport/httptransport"
+)
+
+type Handler interface {
+	IngestHandler
+}
+
+type IngestHandler interface {
+	IngestEvents() IngestEventsHandler
+}
+
+type handler struct {
+	service          ingest.Service
+	namespaceDecoder namespacedriver.NamespaceDecoder
+	options          []httptransport.HandlerOption
+}
+
+func (h *handler) resolveNamespace(ctx context.Context) (string, error) {
+	ns, ok := h.namespaceDecoder.GetNamespace(ctx)
+	if !ok {
+		return "", commonhttp.NewHTTPError(http.StatusInternalServerError, errors.New("internal server error"))
+	}
+
+	return ns, nil
+}
+
+func New(
+	namespaceDecoder namespacedriver.NamespaceDecoder,
+	service ingest.Service,
+	options ...httptransport.HandlerOption,
+) Handler {
+	return &handler{
+		namespaceDecoder: namespaceDecoder,
+		service:          service,
+		options:          options,
+	}
+}

openmeter/ingest/httpdriver/handler.go

@@ -0,0 +1,109 @@
+package httpdriver
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http"
+
+	"github.com/cloudevents/sdk-go/v2/event"
+
+	"github.com/openmeterio/openmeter/api"
+	"github.com/openmeterio/openmeter/openmeter/ingest"
+	"github.com/openmeterio/openmeter/pkg/framework/commonhttp"
+	"github.com/openmeterio/openmeter/pkg/framework/transport/httptransport"
+)
+
+type (
+	IngestEventsRequest  = ingest.IngestEventsRequest
+	IngestEventsResponse = struct{}
+	IngestEventsHandler  httptransport.Handler[IngestEventsRequest, IngestEventsResponse]
+)
+
+func (h *handler) IngestEvents() IngestEventsHandler {
+	return httptransport.NewHandler(
+		func(ctx context.Context, r *http.Request) (IngestEventsRequest, error) {
+			var req ingest.IngestEventsRequest
+
+			namespace, err := h.resolveNamespace(ctx)
+			if err != nil {
+				return req, err
+			}
+
+			req.Namespace = namespace
+
+			contentType := r.Header.Get("Content-Type")
+
+			switch contentType {
+			case "application/json":
+				var apiRequest api.IngestEventsBody
+
+				err := json.NewDecoder(r.Body).Decode(&apiRequest)
+				if err != nil {
+					return req, ErrorInvalidEvent{
+						Err: err,
+					}
+				}
+
+				// Try to parse as a single event
+				e, err := apiRequest.AsEvent()
+				if err == nil {
+					req.Events = []event.Event{e}
+				} else {
+					// Try to parse as a batch of events
+					e, err := apiRequest.AsIngestEventsBody1()
+					if err == nil {
+						req.Events = e
+					}
+				}
+
+				// If we still don't have any events, return an error
+				if len(req.Events) == 0 {
+					return req, ErrorInvalidEvent{
+						Err: errors.New("no events found"),
+					}
+				}
+			case "application/cloudevents+json":
+				var apiRequest api.IngestEventsApplicationCloudeventsPlusJSONRequestBody
+
+				err := json.NewDecoder(r.Body).Decode(&apiRequest)
+				if err != nil {
+					return req, ErrorInvalidEvent{
+						Err: err,
+					}
+				}
+
+				req.Events = []event.Event{apiRequest}
+			case "application/cloudevents-batch+json":
+				var apiRequest api.IngestEventsApplicationCloudeventsBatchPlusJSONBody
+
+				err := json.NewDecoder(r.Body).Decode(&apiRequest)
+				if err != nil {
+					return req, ErrorInvalidEvent{
+						Err: err,
+					}
+				}
+
+				req.Events = apiRequest
+			default:
+				return req, ErrorInvalidContentType{ContentType: contentType}
+			}
+
+			return req, nil
+		},
+		func(ctx context.Context, params IngestEventsRequest) (IngestEventsResponse, error) {
+			_, err := h.service.IngestEvents(ctx, params)
+			if err != nil {
+				return IngestEventsResponse{}, err
+			}
+
+			return IngestEventsResponse{}, nil
+		},
+		commonhttp.EmptyResponseEncoder[IngestEventsResponse](http.StatusNoContent),
+		httptransport.AppendOptions(
+			h.options,
+			httptransport.WithOperationName("IngestEvents"),
+			httptransport.WithErrorEncoder(errorEncoder()),
+		)...,
+	)
+}

openmeter/ingest/httpdriver/ingest.go

@@ -1,4 +1,4 @@
-package ingestdriver_test
+package httpdriver_test
 
 import (
 	"bytes"
@@ -16,27 +16,25 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/openmeterio/openmeter/openmeter/ingest"
-	"github.com/openmeterio/openmeter/openmeter/ingest/ingestdriver"
+	"github.com/openmeterio/openmeter/openmeter/ingest/httpdriver"
 	"github.com/openmeterio/openmeter/openmeter/namespace/namespacedriver"
-	"github.com/openmeterio/openmeter/pkg/errorsx"
 )
 
 func TestIngestEvents(t *testing.T) {
 	collector := ingest.NewInMemoryCollector()
 
-	service := ingest.Service{
+	ingestSvc, err := ingest.NewService(ingest.Config{
 		Collector: collector,
 		Logger:    slog.Default(),
-	}
+	})
+	require.NoError(t, err)
 
-	handler := ingestdriver.NewIngestEventsHandler(
-		service.IngestEvents,
+	handler := httpdriver.New(
 		namespacedriver.StaticNamespaceDecoder("test"),
-		nil,
-		errorsx.NewNopHandler(),
+		ingestSvc,
 	)
 
-	server := httptest.NewServer(handler)
+	server := httptest.NewServer(handler.IngestEvents())
 	client := server.Client()
 
 	now := time.Date(2023, 0o6, 15, 14, 33, 0o0, 0o0, time.UTC)
@@ -49,7 +47,7 @@ func TestIngestEvents(t *testing.T) {
 
 	var buf bytes.Buffer
 
-	err := json.NewEncoder(&buf).Encode(ev)
+	err = json.NewEncoder(&buf).Encode(ev)
 	require.NoError(t, err)
 
 	resp, err := client.Post(server.URL, "application/cloudevents+json", &buf)
@@ -74,19 +72,18 @@ func TestIngestEvents(t *testing.T) {
 func TestIngestEvents_InvalidEvent(t *testing.T) {
 	collector := ingest.NewInMemoryCollector()
 
-	service := ingest.Service{
+	ingestSvc, err := ingest.NewService(ingest.Config{
 		Collector: collector,
 		Logger:    slog.Default(),
-	}
+	})
+	require.NoError(t, err)
 
-	handler := ingestdriver.NewIngestEventsHandler(
-		service.IngestEvents,
+	handler := httpdriver.New(
 		namespacedriver.StaticNamespaceDecoder("test"),
-		nil,
-		errorsx.NewNopHandler(),
+		ingestSvc,
 	)
 
-	server := httptest.NewServer(handler)
+	server := httptest.NewServer(handler.IngestEvents())
 	client := server.Client()
 
 	resp, err := client.Post(server.URL, "application/cloudevents+json", bytes.NewBuffer([]byte(`invalid`)))
@@ -100,19 +97,18 @@ func TestIngestEvents_InvalidEvent(t *testing.T) {
 func TestBatchHandler(t *testing.T) {
 	collector := ingest.NewInMemoryCollector()
 
-	service := ingest.Service{
+	ingestSvc, err := ingest.NewService(ingest.Config{
 		Collector: collector,
 		Logger:    slog.Default(),
-	}
+	})
+	require.NoError(t, err)
 
-	handler := ingestdriver.NewIngestEventsHandler(
-		service.IngestEvents,
+	handler := httpdriver.New(
 		namespacedriver.StaticNamespaceDecoder("test"),
-		nil,
-		errorsx.NewNopHandler(),
+		ingestSvc,
 	)
 
-	server := httptest.NewServer(handler)
+	server := httptest.NewServer(handler.IngestEvents())
 	client := server.Client()
 
 	now := time.Date(2023, 0o6, 15, 14, 33, 0o0, 0o0, time.UTC)
@@ -130,7 +126,7 @@ func TestBatchHandler(t *testing.T) {
 	}
 
 	var buf bytes.Buffer
-	err := json.NewEncoder(&buf).Encode(events)
+	err = json.NewEncoder(&buf).Encode(events)
 	require.NoError(t, err)
 
 	resp, err := client.Post(server.URL, "application/cloudevents-batch+json", &buf)