Typed request body (#168)

We should have a type request body. 

This PR parses the body to specific type. 

Passing to handlers (ingest,query runner) methods will added in the next
PRs.

quesma/quesma/matchers.go

@@ -79,7 +79,14 @@ func matchedAgainstPattern(configuration config.QuesmaConfiguration) mux.Request
 func matchAgainstKibanaAlerts() mux.RequestMatcher {
 	return mux.RequestMatcherFunc(func(req *mux.Request) bool {
 
-		if req.JSON == nil {
+		var query mux.JSON
+
+		switch req.ParsedBody.(type) {
+
+		case mux.JSON:
+			query = req.ParsedBody.(mux.JSON)
+
+		default:
 			return true
 		}
 
@@ -120,7 +127,7 @@ func matchAgainstKibanaAlerts() mux.RequestMatcher {
 			return false
 		}
 
-		q := req.JSON["query"].(map[string]interface{})
+		q := query["query"].(map[string]interface{})
 
 		return !findKibanaAlertField(q)
 	})

quesma/quesma/matchers_test.go

@@ -1,7 +1,7 @@
 package quesma
 
 import (
-	"encoding/json"
+	"context"
 	"github.com/stretchr/testify/assert"
 	"mitmproxy/quesma/quesma/mux"
 	"testing"
@@ -161,16 +161,9 @@ func TestMatchAgainstKibanaAlerts(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(tt *testing.T) {
 
-			var parsed mux.ParsedJSON
-			if test.body != "" {
+			req := &mux.Request{Body: test.body}
 
-				err := json.Unmarshal([]byte(test.body), &parsed)
-				if err != nil {
-					tt.Fatal(err)
-				}
-			}
-
-			req := &mux.Request{Body: test.body, JSON: parsed}
+			mux.ParseRequestBody(context.TODO(), req)
 
 			actual := matchAgainstKibanaAlerts().Matches(req)
 			assert.Equal(t, test.expected, actual)

quesma/quesma/mux/body.go

@@ -0,0 +1,67 @@
+package mux
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+)
+
+type JSON map[string]interface{}
+type NDJSON []JSON
+type Unknown []error
+
+type RequestBody interface {
+	isParsedRequestBody() // this is a marker method
+}
+
+func (j JSON) isParsedRequestBody()    {}
+func (n NDJSON) isParsedRequestBody()  {}
+func (e Unknown) isParsedRequestBody() {}
+
+func ParseRequestBody(ctx context.Context, req *Request) {
+
+	var errors []error
+
+	// maybe it's a JSON
+
+	if len(req.Body) > 1 && req.Body[0] == '{' {
+		parsedBody := make(JSON)
+		if err := json.Unmarshal([]byte(req.Body), &parsedBody); err != nil {
+			errors = append(errors, fmt.Errorf("error while parsing JSON %s", err))
+		} else {
+			req.ParsedBody = JSON(parsedBody)
+			return
+		}
+	}
+
+	// maybe it's a NDJSON
+
+	if len(req.Body) > 1 && req.Body[0] == '{' {
+
+		var ndjson NDJSON
+
+		var err error
+		for _, line := range strings.Split(req.Body, "\n") {
+
+			parsedLine := make(JSON)
+
+			err = json.Unmarshal([]byte(line), &parsedLine)
+			if err != nil {
+				errors = append(errors, fmt.Errorf("error while parsing NDJSON %s", err))
+				break
+			}
+
+			ndjson = append(ndjson, parsedLine)
+		}
+		if err == nil {
+			req.ParsedBody = ndjson
+			return
+		}
+	}
+
+	// if nothing else, it's unknown
+
+	req.ParsedBody = Unknown(errors)
+
+}

quesma/quesma/mux/mux.go

@@ -24,21 +24,19 @@ type (
 		Meta       map[string]string
 		StatusCode int
 	}
-	ParsedJSON map[string]interface{}
-	Request    struct {
+
+	Request struct {
 		Method string
 		Path   string
 		Params map[string]string
 
 		Headers     http.Header
 		QueryParams url.Values
 
-		// body can be:
-
-		Body   string
-		JSON   ParsedJSON
-		NDJSON []ParsedJSON
+		Body       string
+		ParsedBody RequestBody
 	}
+
 	Handler func(ctx context.Context, req *Request) (*Result, error)
 
 	RequestMatcher interface {

quesma/quesma/quesma.go

@@ -3,7 +3,6 @@ package quesma
 import (
 	"bytes"
 	"context"
-	"encoding/json"
 	"fmt"
 	"io"
 	"mitmproxy/quesma/clickhouse"
@@ -146,24 +145,7 @@ func (r *router) reroute(ctx context.Context, w http.ResponseWriter, req *http.R
 		Body:        string(reqBody),
 	}
 
-	// TODO omit _bulk requests from parsing JSON
-	if !strings.Contains(req.URL.Path, "_bulk") {
-		// try to parse the body as JSON
-		// we should rely here on the content type header
-		// or not?
-
-		if len(quesmaRequest.Body) > 1 && quesmaRequest.Body[0] == '{' {
-			parsedBody := make(map[string]interface{})
-			if err := json.Unmarshal(reqBody, &parsedBody); err != nil {
-				logger.ErrorWithCtx(ctx).Msgf("Error parsing request body as a JSON: %v", err)
-			} else {
-				quesmaRequest.JSON = parsedBody
-			}
-		}
-	}
-
-	// TODO parse other types of content types
-	// It will be implemented after the next SLAM session
+	mux.ParseRequestBody(ctx, quesmaRequest)
 
 	handler, found := router.Matches(quesmaRequest)