Process Upload VOD

.github/workflows/build.yaml

@@ -55,7 +55,7 @@ jobs:
       - name: Build
         run: |
           mkdir -p build/ releases/
-          make all GO_BUILD_DIR="build/"
+          make build GO_BUILD_DIR="build/"
 
       - name: Archive binaries for windows
         if: matrix.platform == 'windows'
@@ -126,7 +126,7 @@ jobs:
       - name: Build
         run: |
           mkdir -p build/ releases/
-          GOARCH="${{ matrix.arch }}" make -j4 all GO_BUILD_DIR="build/"
+          GOARCH="${{ matrix.arch }}" make -j4 build GO_BUILD_DIR="build/"
           cd build/
           for file in $(find . -type f -perm -a+x)
           do
@@ -167,4 +167,4 @@ jobs:
         uses: actions/upload-artifact@master
         with:
           name: release-artifacts
-          path: releases/
+          path: releases/

Makefile

@@ -3,8 +3,16 @@ GO_BUILD_DIR?=build/
 ldflags := -X 'github.com/livepeer/catalyst-api/config.Version=$(shell git rev-parse HEAD)'
 
 .PHONY: all
-all: build-server
+all: build fmt test
 
-.PHONY: build-server
-build-server:
+.PHONY: build
+build:
 	go build -ldflags="$(ldflags)" -o "$(GO_BUILD_DIR)catalyst-api" cmd/http-server/http-server.go
+
+.PHONY: fmt
+fmt:
+	go fmt ./...
+
+.PHONY: test
+test:
+	go test ./...

cmd/http-server/http-server.go

@@ -7,7 +7,7 @@ import (
 	"net/http"
 	"os"
 
-	log "github.com/go-kit/kit/log"
+	"github.com/go-kit/kit/log"
 
 	"github.com/julienschmidt/httprouter"
 	"github.com/livepeer/catalyst-api/config"
@@ -18,6 +18,7 @@ import (
 
 func main() {
 	port := flag.Int("port", 4949, "Port to listen on")
+	mistPort := flag.Int("mist-port", 4242, "Port to listen on")
 	mistJson := flag.Bool("j", false, "Print application info as JSON. Used by Mist to present flags in its UI.")
 	flag.Parse()
 
@@ -26,26 +27,35 @@ func main() {
 		return
 	}
 
-	listen := fmt.Sprintf("0.0.0.0:%d", *port)
-	router := StartCatalystAPIRouter()
+	mc := &handlers.MistClient{
+		ApiUrl:          fmt.Sprintf("http://localhost:%d/api2", *mistPort),
+		TriggerCallback: fmt.Sprintf("http://localhost:%d/api/mist/trigger", *port),
+	}
+
+	listen := fmt.Sprintf("localhost:%d", *port)
+	router := StartCatalystAPIRouter(mc)
 
 	stdlog.Println("Starting Catalyst API version", config.Version, "listening on", listen)
 	err := http.ListenAndServe(listen, router)
 	stdlog.Fatal(err)
 
 }
 
-func StartCatalystAPIRouter() *httprouter.Router {
+func StartCatalystAPIRouter(mc *handlers.MistClient) *httprouter.Router {
 	router := httprouter.New()
 
 	var logger log.Logger
 	logger = log.NewLogfmtLogger(log.NewSyncWriter(os.Stderr))
 	logger = log.With(logger, "ts", log.DefaultTimestampUTC)
 	withLogging := middleware.LogRequest(logger)
 
-	router.GET("/ok", withLogging(middleware.IsAuthorized(handlers.CatalystAPIHandlers.Ok())))
-	router.POST("/api/vod", withLogging(middleware.IsAuthorized(handlers.CatalystAPIHandlers.UploadVOD())))
-  router.POST("/api/mist/trigger", withLogging(handlers.MistCallbackHandlers.Trigger()))
+	sc := make(map[string]handlers.StreamInfo)
+	catalystApiHandlers := &handlers.CatalystAPIHandlersCollection{MistClient: mc, StreamCache: sc}
+	mistCallbackHandlers := &handlers.MistCallbackHandlersCollection{MistClient: mc, StreamCache: sc}
+
+	router.GET("/ok", withLogging(middleware.IsAuthorized(catalystApiHandlers.Ok())))
+	router.POST("/api/vod", withLogging(middleware.IsAuthorized(catalystApiHandlers.UploadVOD())))
+	router.POST("/api/mist/trigger", withLogging(mistCallbackHandlers.Trigger()))
 
 	return router
 }

cmd/http-server/http-server_test.go

@@ -8,7 +8,7 @@ import (
 
 func TestInitServer(t *testing.T) {
 	require := require.New(t)
-	router := StartCatalystAPIRouter()
+	router := StartCatalystAPIRouter(nil)
 
 	handle, _, _ := router.Lookup("GET", "/ok")
 	require.NotNil(handle)

handlers/client.go

@@ -11,9 +11,17 @@ import (
 	"sync"
 )
 
+type MistAPIClient interface {
+	AddStream(streamName, sourceUrl string) error
+	PushStart(streamName, targetURL string) error
+	DeleteStream(streamName string) error
+	AddTrigger(streamName, triggerName string) error
+	DeleteTrigger(streamName, triggerName string) error
+}
+
 type MistClient struct {
-	apiUrl          string
-	triggerCallback string
+	ApiUrl          string
+	TriggerCallback string
 	configMu        sync.Mutex
 }
 
@@ -47,7 +55,7 @@ func (mc *MistClient) AddTrigger(streamName, triggerName string) error {
 	if err != nil {
 		return err
 	}
-	c := commandAddTrigger(streamName, triggerName, mc.triggerCallback, triggers)
+	c := commandAddTrigger(streamName, triggerName, mc.TriggerCallback, triggers)
 	resp, err := mc.sendCommand(c)
 	return validateAddTrigger(streamName, triggerName, resp, err)
 }
@@ -97,7 +105,7 @@ func (mc *MistClient) sendCommand(command interface{}) (string, error) {
 		return "", err
 	}
 	payload := payloadFor(c)
-	resp, err := http.Post(mc.apiUrl, "application/x-www-form-urlencoded", bytes.NewBuffer([]byte(payload)))
+	resp, err := http.Post(mc.ApiUrl, "application/json", bytes.NewBuffer([]byte(payload)))
 	if err != nil {
 		return "", err
 	}
@@ -200,7 +208,7 @@ func commandUpdateTrigger(streamName, triggerName string, currentTriggers Trigge
 	triggersMap := currentTriggers
 
 	triggers := triggersMap[triggerName]
-	triggers = filterTriggersWithoutStream(triggers, streamName)
+	triggers = deleteAllTriggersFor(triggers, streamName)
 	if len(replaceTrigger.Streams) != 0 {
 		triggers = append(triggers, replaceTrigger)
 	}
@@ -209,7 +217,7 @@ func commandUpdateTrigger(streamName, triggerName string, currentTriggers Trigge
 	return MistConfig{Config{Triggers: triggersMap}}
 }
 
-func filterTriggersWithoutStream(triggers []ConfigTrigger, streamName string) []ConfigTrigger {
+func deleteAllTriggersFor(triggers []ConfigTrigger, streamName string) []ConfigTrigger {
 	var res []ConfigTrigger
 	for _, t := range triggers {
 		f := false

handlers/handlers.go

@@ -3,9 +3,9 @@ package handlers
 import (
 	"encoding/json"
 	"fmt"
+	"github.com/livepeer/catalyst-api/clients"
 	"io"
 	"io/ioutil"
-	"log"
 	"math/rand"
 	"mime"
 	"net/http"
@@ -17,25 +17,27 @@ import (
 	"github.com/xeipuuv/gojsonschema"
 )
 
-type CatalystAPIHandlersCollection struct{}
+type StreamInfo struct {
+	callbackUrl string
+}
 
-var CatalystAPIHandlers = CatalystAPIHandlersCollection{}
+type CatalystAPIHandlersCollection struct {
+	MistClient  MistAPIClient
+	StreamCache map[string]StreamInfo
+}
 
 func (d *CatalystAPIHandlersCollection) Ok() httprouter.Handle {
 	return func(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
 		io.WriteString(w, "OK")
 	}
 }
 
-var processUpload = processUploadVOD
-
 func (d *CatalystAPIHandlersCollection) 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": {
@@ -75,11 +77,15 @@ func (d *CatalystAPIHandlersCollection) UploadVOD() httprouter.Handle {
 	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"`
+			Outputs         struct {
+				SourceMp4          bool `json:"source_mp4"`
+				SourceSegments     bool `json:"source_segments"`
+				TranscodedSegments bool `json:"transcoded_segments"`
+			} `json:"outputs,omitempty"`
 		} `json:"output_locations,omitempty"`
 	}
 
@@ -103,10 +109,32 @@ func (d *CatalystAPIHandlersCollection) UploadVOD() httprouter.Handle {
 			return
 		}
 
-		if err := processUpload(uploadVODRequest.Url); err != nil {
+		// find source segment URL
+		var tURL string
+		for _, o := range uploadVODRequest.OutputLocations {
+			if o.Outputs.SourceSegments {
+				tURL = o.URL
+				break
+			}
+		}
+		if tURL == "" {
+			errors.WriteHTTPBadRequest(w, "Invalid request payload", fmt.Errorf("no source segment URL in request"))
+			return
+		}
+
+		streamName := randomStreamName("catalyst_vod_")
+		d.StreamCache[streamName] = StreamInfo{callbackUrl: uploadVODRequest.CallbackUrl}
+
+		// process the request
+		if err := d.processUploadVOD(streamName, uploadVODRequest.Url, tURL); err != nil {
 			errors.WriteHTTPInternalServerError(w, "Cannot process upload VOD request", err)
 		}
 
+		callbackClient := clients.NewCallbackClient()
+		if err := callbackClient.SendTranscodeStatus(uploadVODRequest.CallbackUrl, clients.TranscodeStatusPreparing, 0.0); err != nil {
+			errors.WriteHTTPInternalServerError(w, "Cannot send transcode status", err)
+		}
+
 		io.WriteString(w, fmt.Sprint(len(uploadVODRequest.OutputLocations)))
 	}
 }
@@ -129,41 +157,17 @@ func HasContentType(r *http.Request, mimetype string) bool {
 	return false
 }
 
-func processUploadVOD(url string) error {
-	// TODO: This function is only a scaffold for now
-
-	// TODO: Update hostnames and ports
-	mc := MistClient{apiUrl: "http://localhost:4242/api2", triggerCallback: "http://host.docker.internal:8080/api/mist/trigger"}
-
-	streamName := randomStreamName("catalyst_vod_")
-	if err := mc.AddStream(streamName, url); err != nil {
+func (d *CatalystAPIHandlersCollection) processUploadVOD(streamName, sourceURL, targetURL string) error {
+	if err := d.MistClient.AddStream(streamName, sourceURL); err != nil {
 		return err
 	}
-
-	// TODO: Move it to `Trigger()`
-	defer mc.DeleteStream(streamName)
-
-	if err := mc.AddTrigger(streamName, "PUSH_END"); err != nil {
+	if err := d.MistClient.AddTrigger(streamName, "PUSH_END"); err != nil {
 		return err
 	}
-	// TODO: Move it to `Trigger()`
-	defer mc.DeleteTrigger(streamName, "PUSH_END")
-
-	if err := mc.AddTrigger(streamName, "RECORDING_END"); err != nil {
+	if err := d.MistClient.PushStart(streamName, targetURL); err != nil {
 		return err
 	}
 
-	// TODO: Move it to `Trigger()`
-	defer mc.DeleteTrigger(streamName, "RECORDING_END")
-
-	// TODO: Change the output to the value from the request instead of the hardcoded "/media/recording/result.ts"
-	if err := mc.PushStart(streamName, "/media/recording/result.ts"); err != nil {
-		return err
-	}
-
-	// TODO: After moving cleanup to `Trigger()`, this is no longer needed
-	time.Sleep(10 * time.Second)
-
 	return nil
 }
 
@@ -179,21 +183,50 @@ func randomStreamName(prefix string) string {
 	return fmt.Sprintf("%s%s", prefix, string(res))
 }
 
-type MistCallbackHandlersCollection struct{}
-
-var MistCallbackHandlers = MistCallbackHandlersCollection{}
+type MistCallbackHandlersCollection struct {
+	MistClient  MistAPIClient
+	StreamCache map[string]StreamInfo
+}
 
 func (d *MistCallbackHandlersCollection) Trigger() httprouter.Handle {
 	return func(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
-		log.Println("Received Mist Trigger")
+		if t := req.Header.Get("X-Trigger"); t != "PUSH_END" {
+			errors.WriteHTTPBadRequest(w, "Unsupported X-Trigger", fmt.Errorf("unknown trigger '%s'", t))
+			return
+		}
 		payload, err := ioutil.ReadAll(req.Body)
 		if err != nil {
 			errors.WriteHTTPInternalServerError(w, "Cannot read payload", err)
 			return
 		}
+		lines := strings.Split(strings.TrimSuffix(string(payload), "\n"), "\n")
+		if len(lines) < 2 {
+			errors.WriteHTTPBadRequest(w, "Bad request payload", fmt.Errorf("unknown payload '%s'", string(payload)))
+			return
+		}
 
-		// TODO: Handle trigger results: 1) Check the trigger name, 2) Call callbackURL, 3) Perform stream cleanup
-		fmt.Println(string(payload))
-		io.WriteString(w, "OK")
+		// stream name is the second line in the Mist Trigger payload
+		s := lines[1]
+		// when uploading is done, remove trigger and stream from Mist
+		errT := d.MistClient.DeleteTrigger(s, "PUSH_END")
+		errS := d.MistClient.DeleteStream(s)
+		if errT != nil {
+			errors.WriteHTTPInternalServerError(w, fmt.Sprintf("Cannot remove PUSH_END trigger for stream '%s'", s), errT)
+			return
+		}
+		if errS != nil {
+			errors.WriteHTTPInternalServerError(w, fmt.Sprintf("Cannot remove stream '%s'", s), errS)
+			return
+		}
+
+		callbackClient := clients.NewCallbackClient()
+		if err := callbackClient.SendTranscodeStatus(d.StreamCache[s].callbackUrl, clients.TranscodeStatusTranscoding, 0.0); err != nil {
+			errors.WriteHTTPInternalServerError(w, "Cannot send transcode status", err)
+		}
+
+		delete(d.StreamCache, s)
+
+		// TODO: add timeout for the stream upload
+		// TODO: start transcoding
 	}
 }