Define new API endpoint + test

cmd/http-server/http-server.go

@@ -55,6 +55,7 @@ func StartCatalystAPIRouter(mc *handlers.MistClient) *httprouter.Router {
 
 	router.GET("/ok", withLogging(middleware.IsAuthorized(catalystApiHandlers.Ok())))
 	router.POST("/api/vod", withLogging(middleware.IsAuthorized(catalystApiHandlers.UploadVOD())))
+	router.POST("/api/transcode/file", withLogging(middleware.IsAuthorized(catalystApiHandlers.TranscodeSegment())))
 	router.POST("/api/mist/trigger", withLogging(mistCallbackHandlers.Trigger()))
 
 	return router

errors/errors.go

@@ -4,6 +4,9 @@ import (
 	"encoding/json"
 	"log"
 	"net/http"
+	"strings"
+
+	"github.com/xeipuuv/gojsonschema"
 )
 
 type apiError struct {
@@ -37,3 +40,15 @@ func WriteHTTPUnsupportedMediaType(w http.ResponseWriter, msg string, err error)
 func WriteHTTPInternalServerError(w http.ResponseWriter, msg string, err error) apiError {
 	return writeHttpError(w, msg, http.StatusInternalServerError, err)
 }
+
+func WriteHTTPBadBodySchema(where string, w http.ResponseWriter, errors []gojsonschema.ResultError) apiError {
+	sb := strings.Builder{}
+	sb.WriteString("Body validation error in ")
+	sb.WriteString(where)
+	sb.WriteString(" ")
+	for i := 0; i < len(errors); i++ {
+		sb.WriteString(errors[i].String())
+		sb.WriteString(" ")
+	}
+	return writeHttpError(w, sb.String(), http.StatusBadRequest, nil)
+}

handlers/handlers.go

@@ -3,7 +3,6 @@ package handlers
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/livepeer/catalyst-api/clients"
 	"io"
 	"io/ioutil"
 	"math/rand"
@@ -12,6 +11,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/livepeer/catalyst-api/clients"
+
 	"github.com/julienschmidt/httprouter"
 	"github.com/livepeer/catalyst-api/errors"
 	"github.com/xeipuuv/gojsonschema"
@@ -32,62 +33,39 @@ func (d *CatalystAPIHandlersCollection) Ok() httprouter.Handle {
 	}
 }
 
-func (d *CatalystAPIHandlersCollection) UploadVOD() httprouter.Handle {
-	schemaLoader := gojsonschema.NewStringLoader(`{
-		"type": "object",
-		"properties": {
-			"url": { "type": "string", "format": "uri" },
-		  	"callback_url": { "type": "string", "format": "uri" },
-		  	"output_locations": {
-				"type": "array",
-				"items": {
-					"oneOf": [
-						{
-							"type": "object",
-			  				"properties": {
-								"type": { "type": "string", "const": "object_store" },
-								"url": { "type": "string", "format": "uri" }
-				  			},
-							"required": [ "type", "url" ],
-							"additional_properties": false
-						},
-						{
-							"type": "object",
-			  				"properties": {
-								"type": { "type": "string", "const": "pinata" },
-								"pinata_access_key": { "type": "string", "minLength": 1 }
-				  			},
-							"required": [ "type", "pinata_access_key" ],
-							"additional_properties": false
-						}
-					]
-				},
-				"minItems": 1
-		  	}
-		},
-		"required": [ "url", "callback_url", "output_locations" ],
-		"additional_properties": false
-	}`)
-
-	schema, err := gojsonschema.NewSchema(schemaLoader)
-	if err != nil {
-		panic(err)
-	}
+func (d *CatalystAPIHandlersCollection) TranscodeSegment() httprouter.Handle {
+	schema := inputSchemasCompiled["TranscodeSegment"]
+	return func(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
+		var transcodeRequest TranscodeSegmentRequest
+		payload, err := ioutil.ReadAll(req.Body)
+		if err != nil {
+			errors.WriteHTTPInternalServerError(w, "Cannot read body", err)
+			return
+		}
+		result, err := schema.Validate(gojsonschema.NewBytesLoader(payload))
+		if err != nil {
+			errors.WriteHTTPInternalServerError(w, "body schema validation failed", err)
+			return
+		}
+		if !result.Valid() {
+			errors.WriteHTTPBadBodySchema("TranscodeSegment", w, result.Errors())
+			return
+		}
+		if err := json.Unmarshal(payload, &transcodeRequest); err != nil {
+			errors.WriteHTTPBadRequest(w, "Invalid request payload", err)
+			return
+		}
 
-	type UploadVODRequest struct {
-		Url             string `json:"url"`
-		CallbackUrl     string `json:"callback_url"`
-		OutputLocations []struct {
-			Type            string `json:"type"`
-			URL             string `json:"url"`
-			PinataAccessKey string `json:"pinata_access_key"`
-			Outputs         struct {
-				SourceMp4          bool `json:"source_mp4"`
-				SourceSegments     bool `json:"source_segments"`
-				TranscodedSegments bool `json:"transcoded_segments"`
-			} `json:"outputs,omitempty"`
-		} `json:"output_locations,omitempty"`
+		callbackClient := clients.NewCallbackClient()
+		if err := callbackClient.SendTranscodeStatusError(transcodeRequest.CallbackUrl, "NYI - not yet implemented"); err != nil {
+			errors.WriteHTTPInternalServerError(w, "error send transcode error", err)
+		}
+		io.WriteString(w, "OK") // TODO later
 	}
+}
+
+func (d *CatalystAPIHandlersCollection) UploadVOD() httprouter.Handle {
+	schema := inputSchemasCompiled["UploadVOD"]
 
 	return func(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
 		var uploadVODRequest UploadVODRequest

handlers/handlers_test.go

@@ -2,12 +2,17 @@ package handlers
 
 import (
 	"bytes"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/julienschmidt/httprouter"
+	"github.com/livepeer/catalyst-api/clients"
 	"github.com/stretchr/testify/require"
 )
 
@@ -24,6 +29,157 @@ func TestOKHandler(t *testing.T) {
 	require.Equal(rr.Body.String(), "OK")
 }
 
+func TestSegmentCallback(t *testing.T) {
+	var jsonData = `{
+		"source_location": "http://localhost/input",
+		"callback_url": "CALLBACK_URL",
+		"manifestID": "somestream",
+		"profiles": [
+			{
+				"name": "720p",
+				"width": 1280,
+				"height": 720,
+				"bitrate": 700000,
+				"fps": 30
+			}, {
+				"name": "360p",
+				"width": 640,
+				"height": 360,
+				"bitrate": 200000,
+				"fps": 30
+			}
+		],
+		"verificationFreq": 1
+	}`
+
+	callbacks := make(chan string, 10)
+	callbackServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		payload, err := ioutil.ReadAll(r.Body)
+		if err != nil {
+			fmt.Printf("WebhookReceiver error reading req body\n")
+			w.WriteHeader(451)
+			return
+		}
+		w.WriteHeader(200)
+		callbacks <- string(payload)
+	}))
+	defer callbackServer.Close()
+	jsonData = strings.ReplaceAll(jsonData, "CALLBACK_URL", callbackServer.URL)
+
+	catalystApiHandlers := CatalystAPIHandlersCollection{MistClient: StubMistClient{}}
+
+	router := httprouter.New()
+
+	req, _ := http.NewRequest("POST", "/api/transcode/file", bytes.NewBuffer([]byte(jsonData)))
+	req.Header.Set("Content-Type", "application/json")
+
+	rr := httptest.NewRecorder()
+	router.POST("/api/transcode/file", catalystApiHandlers.TranscodeSegment())
+	router.ServeHTTP(rr, req)
+
+	require.Equal(t, 200, rr.Result().StatusCode)
+	require.Equal(t, "OK", rr.Body.String())
+
+	// Wait for callback
+	select {
+	case data := <-callbacks:
+		message := &clients.TranscodeStatusMessage{}
+		err := json.Unmarshal([]byte(data), message)
+		require.NoErrorf(t, err, "json unmarshal failed, src=%s", data)
+		require.Equal(t, "error", message.Status)
+		require.Equal(t, "NYI - not yet implemented", message.Error)
+	case <-time.After(300 * time.Millisecond):
+		require.FailNow(t, "Callback not fired by handler")
+	}
+}
+
+func TestSegmentBodyFormat(t *testing.T) {
+	require := require.New(t)
+
+	badRequests := [][]byte{
+		// missing source_location
+		[]byte(`{
+			"callback_url": "http://localhost:8080/callback",
+			"manifestID": "somestream",
+			"profiles": [{"name": "t","width": 1280,"height": 720,"bitrate": 70000,"fps": 30}],
+			"verificationFreq": 1
+		}`),
+		// missing callback_url
+		[]byte(`{
+			"source_location": "http://localhost/input",
+			"manifestID": "somestream",
+			"profiles": [{"name": "t","width": 1280,"height": 720,"bitrate": 70000,"fps": 30}],
+			"verificationFreq": 1
+		}`),
+		// missing manifestID
+		[]byte(`{
+			"source_location": "http://localhost/input",
+			"callback_url": "http://localhost:8080/callback",
+			"profiles": [{"name": "t","width": 1280,"height": 720,"bitrate": 70000,"fps": 30}],
+			"verificationFreq": 1
+		}`),
+		// missing profiles
+		[]byte(`{
+			"source_location": "http://localhost/input",
+			"callback_url": "http://localhost:8080/callback",
+			"manifestID": "somestream",
+			"verificationFreq": 1
+		}`),
+		// missing name
+		[]byte(`{
+			"source_location": "http://localhost/input",
+			"callback_url": "http://localhost:8080/callback",
+			"manifestID": "somestream",
+			"profiles": [{"width": 1280,"height": 720,"bitrate": 70000,"fps": 30}],
+			"verificationFreq": 1
+		}`),
+		// missing width
+		[]byte(`{
+			"source_location": "http://localhost/input",
+			"callback_url": "http://localhost:8080/callback",
+			"manifestID": "somestream",
+			"profiles": [{"name": "t","height": 720,"bitrate": 70000,"fps": 30}],
+			"verificationFreq": 1
+		}`),
+		// missing height
+		[]byte(`{
+			"source_location": "http://localhost/input",
+			"callback_url": "http://localhost:8080/callback",
+			"manifestID": "somestream",
+			"profiles": [{"name": "t","width": 1280,"bitrate": 70000,"fps": 30}],
+			"verificationFreq": 1
+		}`),
+		// missing bitrate
+		[]byte(`{
+			"source_location": "http://localhost/input",
+			"callback_url": "http://localhost:8080/callback",
+			"manifestID": "somestream",
+			"profiles": [{"name": "t","width": 1280,"height": 720,"fps": 30}],
+			"verificationFreq": 1
+		}`),
+		// missing verificationFreq
+		[]byte(`{
+			"source_location": "http://localhost/input",
+			"callback_url": "http://localhost:8080/callback",
+			"manifestID": "somestream",
+			"profiles": [{"name": "t","width": 1280,"height": 720,"bitrate": 70000,"fps": 30}]
+		}`),
+	}
+
+	catalystApiHandlers := CatalystAPIHandlersCollection{MistClient: StubMistClient{}}
+	router := httprouter.New()
+
+	router.POST("/api/transcode/file", catalystApiHandlers.TranscodeSegment())
+	for _, payload := range badRequests {
+		req, _ := http.NewRequest("POST", "/api/transcode/file", bytes.NewBuffer(payload))
+		req.Header.Set("Content-Type", "application/json")
+
+		rr := httptest.NewRecorder()
+		router.ServeHTTP(rr, req)
+		require.Equal(400, rr.Result().StatusCode, string(payload))
+	}
+}
+
 func TestSuccessfulVODUploadHandler(t *testing.T) {
 	require := require.New(t)
 

handlers/json_schema.go

@@ -0,0 +1,24 @@
+package handlers
+
+import "github.com/xeipuuv/gojsonschema"
+
+var inputSchemas map[string]string = map[string]string{
+	"TranscodeSegment": TranscodeSegmentRequestSchemaDefinition,
+	"UploadVOD":        UploadVODRequestSchemaDefinition,
+}
+
+func compileJsonSchemas() map[string]*gojsonschema.Schema {
+	compiled := make(map[string]*gojsonschema.Schema, 0)
+	for name, text := range inputSchemas {
+		schema, err := gojsonschema.NewSchema(gojsonschema.NewStringLoader(text))
+		if err != nil {
+			// rase panic on program start
+			panic(err) // fix schema text
+		}
+		compiled[name] = schema
+	}
+	return compiled
+}
+
+// Run compile step on program start:
+var inputSchemasCompiled map[string]*gojsonschema.Schema = compileJsonSchemas()

handlers/segmenting.go

@@ -0,0 +1,52 @@
+package handlers
+
+type UploadVODRequest struct {
+	Url             string `json:"url"`
+	CallbackUrl     string `json:"callback_url"`
+	OutputLocations []struct {
+		Type            string `json:"type"`
+		URL             string `json:"url"`
+		PinataAccessKey string `json:"pinata_access_key"`
+		Outputs         struct {
+			SourceMp4          bool `json:"source_mp4"`
+			SourceSegments     bool `json:"source_segments"`
+			TranscodedSegments bool `json:"transcoded_segments"`
+		} `json:"outputs,omitempty"`
+	} `json:"output_locations,omitempty"`
+}
+
+var UploadVODRequestSchemaDefinition string = `{
+	"type": "object",
+	"properties": {
+		"url": { "type": "string", "format": "uri" },
+			"callback_url": { "type": "string", "format": "uri" },
+			"output_locations": {
+			"type": "array",
+			"items": {
+				"oneOf": [
+					{
+						"type": "object",
+							"properties": {
+							"type": { "type": "string", "const": "object_store" },
+							"url": { "type": "string", "format": "uri" }
+							},
+						"required": [ "type", "url" ],
+						"additional_properties": false
+					},
+					{
+						"type": "object",
+							"properties": {
+							"type": { "type": "string", "const": "pinata" },
+							"pinata_access_key": { "type": "string", "minLength": 1 }
+							},
+						"required": [ "type", "pinata_access_key" ],
+						"additional_properties": false
+					}
+				]
+			},
+			"minItems": 1
+			}
+	},
+	"required": [ "url", "callback_url", "output_locations" ],
+	"additional_properties": false
+}`