Add upload VOD endpoint (#4)

* Add upload VOD endpoint

* Use errors module

* Use internal/external structs instead of direct unmarshalling

* Address review comments

* Get rid of go-livepeer dependency
* Use httprouter instead of http
* Fix typos
* Use jsonschema

* Remove useless schema directive

cmd/http-server/http-server.go

@@ -4,26 +4,25 @@ import (
 	"log"
 	"net/http"
 
+	"github.com/julienschmidt/httprouter"
 	"github.com/livepeer/dms-api/handlers"
 	"github.com/livepeer/dms-api/middleware"
 )
 
 func main() {
-	server := StartDMSAPIServer("localhost:8080")
-	err := server.ListenAndServe()
+	listen := "localhost:8080"
+	router := StartDMSAPIRouter()
+	log.Println("DMS API server listening on", listen)
+
+	err := http.ListenAndServe(listen, router)
 	log.Fatal(err)
 }
 
-func StartDMSAPIServer(addr string) http.Server {
-	server := http.Server{Addr: addr}
-
-	mux := http.DefaultServeMux
-	http.DefaultServeMux = http.NewServeMux()
-
-	mux.Handle("/ok", middleware.IsAuthorized(handlers.DMSAPIHandlers.Ok()))
+func StartDMSAPIRouter() *httprouter.Router {
+	router := httprouter.New()
 
-	log.Println("DMS API server listening on", addr)
-	server.Handler = mux
+	router.GET("/ok", middleware.IsAuthorized(handlers.DMSAPIHandlers.Ok()))
+	router.POST("/api/vod", middleware.IsAuthorized(handlers.DMSAPIHandlers.UploadVOD()))
 
-	return server
+	return router
 }

cmd/http-server/http-server_test.go

@@ -10,9 +10,8 @@ func TestInitServer(t *testing.T) {
 	require := require.New(t)
 
 	const listen = "localhost:8081"
-	server := StartDMSAPIServer(listen)
+	router := StartDMSAPIRouter()
 
-	require.Equal(server.Addr, listen)
-
-	server.Close()
+	handle, _, _ := router.Lookup("GET", "/ok")
+	require.NotNil(handle)
 }

errors/errors.go

@@ -25,3 +25,15 @@ func writeHttpError(w http.ResponseWriter, msg string, status int, err error) ap
 func WriteHTTPUnauthorized(w http.ResponseWriter, msg string, err error) apiError {
 	return writeHttpError(w, msg, http.StatusUnauthorized, err)
 }
+
+func WriteHTTPBadRequest(w http.ResponseWriter, msg string, err error) apiError {
+	return writeHttpError(w, msg, http.StatusBadRequest, err)
+}
+
+func WriteHTTPUnsupportedMediaType(w http.ResponseWriter, msg string, err error) apiError {
+	return writeHttpError(w, msg, http.StatusUnsupportedMediaType, err)
+}
+
+func WriteHTTPInternalServerError(w http.ResponseWriter, msg string, err error) apiError {
+	return writeHttpError(w, msg, http.StatusInternalServerError, err)
+}

go.mod

@@ -2,9 +2,18 @@ module github.com/livepeer/dms-api
 
 go 1.18
 
+require (
+	github.com/julienschmidt/httprouter v1.3.0
+	github.com/stretchr/testify v1.8.0
+	github.com/xeipuuv/gojsonschema v1.2.0
+)
+
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/kr/pretty v0.3.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/stretchr/testify v1.8.0 // indirect
+	github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f // indirect
+	github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect
+	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -1,14 +1,38 @@
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/julienschmidt/httprouter v1.3.0 h1:U0609e9tgbseu3rBINet9P48AI/D3oJs4dN7jwJOQ1U=
+github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
+github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0=
+github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/rogpeppe/go-internal v1.6.1 h1:/FiVV8dS/e+YqF2JvO3yXRFbBLTIuSDkuC7aBOAvL+k=
+github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f h1:J9EGpcZtP0E/raorCMxlFGSTBrsSlaDGf3jU/qvAE2c=
+github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
+github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 h1:EzJWgHovont7NscjpAxXsDA8S8BMYve8Y5+7cuRE7R0=
+github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415/go.mod h1:GwrjFmJcFw6At/Gs6z4yjiIwzuJ1/+UwLxMQDVQXShQ=
+github.com/xeipuuv/gojsonschema v1.2.0 h1:LhYJRs+L4fBtjZUfuSZIKGeVu0QRy8e5Xi7D17UxZ74=
+github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

handlers/handlers.go

@@ -1,16 +1,122 @@
 package handlers
 
 import (
+	"encoding/json"
+	"fmt"
 	"io"
+	"io/ioutil"
+	"mime"
 	"net/http"
+	"strings"
+
+	"github.com/julienschmidt/httprouter"
+	"github.com/livepeer/dms-api/errors"
+	"github.com/xeipuuv/gojsonschema"
 )
 
 type DMSAPIHandlersCollection struct{}
 
 var DMSAPIHandlers = DMSAPIHandlersCollection{}
 
-func (d *DMSAPIHandlersCollection) Ok() http.HandlerFunc {
-	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+func (d *DMSAPIHandlersCollection) Ok() httprouter.Handle {
+	return func(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
 		io.WriteString(w, "OK")
-	})
+	}
+}
+
+func (d *DMSAPIHandlersCollection) UploadVOD() httprouter.Handle {
+	schemaLoader := gojsonschema.NewStringLoader(`{
+		"type": "object",
+		"properties": {
+			"url": { "type": "string", "format": "uri" },
+		  	"callback_url": { "type": "string", "format": "uri" },
+		  	"mp4_output": { "type": "boolean" },
+		  	"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)
+	}
+
+	type UploadVODRequest struct {
+		Url             string `json:"url"`
+		CallbackUrl     string `json:"callback_url"`
+		Mp4Output       bool   `json:"mp4_output"`
+		OutputLocations []struct {
+			Type            string `json:"type"`
+			URL             string `json:"url"`
+			PinataAccessKey string `json:"pinata_access_key"`
+		} `json:"output_locations,omitempty"`
+	}
+
+	return func(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
+		var uploadVODRequest UploadVODRequest
+
+		if !HasContentType(req, "application/json") {
+			errors.WriteHTTPUnsupportedMediaType(w, "Requires application/json content type", nil)
+			return
+		} else if payload, err := ioutil.ReadAll(req.Body); err != nil {
+			errors.WriteHTTPInternalServerError(w, "Cannot read payload", err)
+			return
+		} else if result, err := schema.Validate(gojsonschema.NewBytesLoader(payload)); err != nil {
+			errors.WriteHTTPInternalServerError(w, "Cannot validate payload", err)
+			return
+		} else if !result.Valid() {
+			errors.WriteHTTPBadRequest(w, "Invalid request payload", nil)
+			return
+		} else if err := json.Unmarshal(payload, &uploadVODRequest); err != nil {
+			errors.WriteHTTPBadRequest(w, "Invalid request payload", err)
+			return
+		}
+
+		// Do something with uploadVODRequest
+		io.WriteString(w, fmt.Sprint(len(uploadVODRequest.OutputLocations)))
+	}
+}
+
+func HasContentType(r *http.Request, mimetype string) bool {
+	contentType := r.Header.Get("Content-Type")
+	if contentType == "" {
+		return mimetype == "application/octet-stream"
+	}
+
+	for _, v := range strings.Split(contentType, ",") {
+		t, _, err := mime.ParseMediaType(v)
+		if err != nil {
+			break
+		}
+		if t == mimetype {
+			return true
+		}
+	}
+	return false
 }

handlers/handlers_test.go

@@ -1,20 +1,144 @@
 package handlers
 
 import (
+	"bytes"
 	"net/http"
 	"net/http/httptest"
 	"testing"
 
+	"github.com/julienschmidt/httprouter"
 	"github.com/stretchr/testify/require"
 )
 
 func TestOKHandler(t *testing.T) {
 	require := require.New(t)
 
+	router := httprouter.New()
 	req, _ := http.NewRequest("GET", "/ok", nil)
 	rr := httptest.NewRecorder()
-	h := DMSAPIHandlers.Ok()
-	h.ServeHTTP(rr, req)
+	router.GET("/ok", DMSAPIHandlers.Ok())
+	router.ServeHTTP(rr, req)
 
 	require.Equal(rr.Body.String(), "OK")
 }
+
+func TestSuccessfulVODUploadHandler(t *testing.T) {
+	require := require.New(t)
+
+	var jsonData = []byte(`{
+		"url": "http://localhost/input",
+		"callback_url": "http://localhost/callback",
+		"output_locations": [
+			{
+				"type": "object_store",
+				"url": "memory://localhost/output"
+			},
+			{
+				"type": "pinata",
+				"pinata_access_key": "abc"
+			}
+		]
+	}`)
+
+	router := httprouter.New()
+
+	req, _ := http.NewRequest("POST", "/api/vod", bytes.NewBuffer(jsonData))
+	req.Header.Set("Content-Type", "application/json")
+
+	rr := httptest.NewRecorder()
+	router.POST("/api/vod", DMSAPIHandlers.UploadVOD())
+	router.ServeHTTP(rr, req)
+
+	require.Equal(rr.Result().StatusCode, 200)
+	require.Equal(rr.Body.String(), "2")
+}
+
+func TestInvalidPayloadVODUploadHandler(t *testing.T) {
+	require := require.New(t)
+
+	badRequests := [][]byte{
+		// missing url
+		[]byte(`{
+			"callback_url": "http://localhost/callback",
+			"output_locations": [ { "type": "object_store", "url": "memory://localhost/output" } ]
+		}`),
+		// missing callback_url
+		[]byte(`{
+			"url": "http://localhost/input",
+			"output_locations": [ { "type": "object_store", "url": "memory://localhost/output" } ]
+		}`),
+		// missing output_locatoins
+		[]byte(`{
+			"url": "http://localhost/input",
+			"callback_url": "http://localhost/callback"
+		}`),
+		// invalid url
+		[]byte(`{
+			"url": "x://}]:&7@localhost/",
+			"callback_url": "http://localhost/callback",
+			"output_locations": [ { "type": "object_store", "url": "memory://localhost/output" } ]
+		}`),
+		// invalid callback_url
+		[]byte(`{
+			"url": "http://localhost/input",
+			"callback_url": "x://}]:&7@localhost/",
+			"output_locations": [ { "type": "object_store", "url": "memory://localhost/output" } ]
+		}`),
+		// invalid output_location's object_store url
+		[]byte(`{
+			"url": "http://localhost/input",
+			"callback_url": "http://localhost/callback",
+			"output_locations": [ { "type": "object_store", "url": "x://}]:&7@localhost/" } ]
+		}`),
+		// invalid output_location type
+		[]byte(`{
+			"url": "http://localhost/input",
+			"callback_url": "http://localhost/callback",
+			"output_locations": [ { "type": "foo", "url": "http://localhost/" } ]
+		}`),
+		// invalid output_location's pinata params
+		[]byte(`{
+			"url": "http://localhost/input",
+			"callback_url": "http://localhost/callback",
+			"output_locations": [ { "type": "pinata", "pinata_access_key": "" } ]
+		}`),
+	}
+
+	router := httprouter.New()
+
+	router.POST("/api/vod", DMSAPIHandlers.UploadVOD())
+	for _, payload := range badRequests {
+		req, _ := http.NewRequest("POST", "/api/vod", bytes.NewBuffer(payload))
+		req.Header.Set("Content-Type", "application/json")
+
+		rr := httptest.NewRecorder()
+		router.ServeHTTP(rr, req)
+		require.Equal(rr.Result().StatusCode, 400)
+	}
+}
+
+func TestWrongContentTypeVODUploadHandler(t *testing.T) {
+	require := require.New(t)
+
+	var jsonData = []byte(`{
+		"url": "http://localhost/input",
+		"callback_url": "http://localhost/callback",
+		"output_locations": [
+			{
+				"type": "object_store",
+				"url": "http://localhost/"
+			}
+		]
+	}`)
+
+	router := httprouter.New()
+	req, _ := http.NewRequest("POST", "/api/vod", bytes.NewBuffer(jsonData))
+	req.Header.Set("Content-Type", "json")
+
+	rr := httptest.NewRecorder()
+	router.POST("/api/vod", DMSAPIHandlers.UploadVOD())
+	router.ServeHTTP(rr, req)
+
+	require.Equal(rr.Result().StatusCode, 415)
+	require.JSONEq(rr.Body.String(), `{"error": "Requires application/json content type"}`)
+}