Notify studio when recording completes

handlers/misttriggers/push_end.go

@@ -2,8 +2,11 @@ package misttriggers
 
 import (
 	"fmt"
+	"log"
 	"net/http"
+	"net/url"
 	"strings"
+	"time"
 
 	"github.com/livepeer/catalyst-api/cache"
 	"github.com/livepeer/catalyst-api/clients"
@@ -40,6 +43,8 @@ func (d *MistCallbackHandlersCollection) TriggerPushEnd(w http.ResponseWriter, r
 		d.TranscodingPushEnd(w, req, streamName, destination, actualDestination, pushStatus)
 	case Segmenting:
 		d.SegmentingPushEnd(w, req, streamName)
+	case Recording:
+		d.RecordingPushEnd(w, req, streamName, actualDestination, pushStatus)
 	default:
 		// Not related to API logic
 	}
@@ -82,6 +87,23 @@ func (d *MistCallbackHandlersCollection) TranscodingPushEnd(w http.ResponseWrite
 	}
 }
 
+func (d *MistCallbackHandlersCollection) RecordingPushEnd(w http.ResponseWriter, req *http.Request, streamName, actualDestination, pushStatus string) {
+	var err error
+	pushSuccess := pushStatus == "null"
+	event := &clients.RecordingEvent{
+		Event:      "end",
+		Timestamp:  time.Now().UnixMilli(),
+		StreamName: streamName,
+		Hostname:   req.Host,
+		Success:    &pushSuccess,
+	}
+	if event.RecordingId, err = uuidFromPushUrl(actualDestination); err != nil {
+		log.Printf("RecordingPushEnd extract uuid failed %v", err)
+		return
+	}
+	go clients.DefaultCallbackClient.SendRecordingEvent(event)
+}
+
 func (d *MistCallbackHandlersCollection) SegmentingPushEnd(w http.ResponseWriter, req *http.Request, streamName string) {
 	// when uploading is done, remove trigger and stream from Mist
 	defer cache.DefaultStreamCache.Segmenting.Remove(streamName)
@@ -109,3 +131,15 @@ func (d *MistCallbackHandlersCollection) SegmentingPushEnd(w http.ResponseWriter
 	// TODO: Start Transcoding (stubbed for now with below method)
 	stubTranscodingCallbacksForStudio(callbackUrl)
 }
+
+func uuidFromPushUrl(uri string) (string, error) {
+	pushUrl, err := url.Parse(uri)
+	if err != nil {
+		return "", err
+	}
+	path := strings.Split(pushUrl.EscapedPath(), "/")
+	if len(path) < 4 {
+		return "", fmt.Errorf("push url path malformed: element count %d %s", len(path), pushUrl.EscapedPath())
+	}
+	return path[len(path)-2], nil
+}

handlers/misttriggers/triggers_test.go

@@ -68,6 +68,45 @@ func TestRecordingStart(t *testing.T) {
 	}
 }
 
+func TestRecordingCompleted(t *testing.T) {
+	testStartTime := time.Now().UnixMilli()
+	mistCallbackHandlers := &MistCallbackHandlersCollection{MistClient: clients.StubMistClient{}}
+	callbackHappened := make(chan bool, 10)
+	callbackServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		payload, err := io.ReadAll(r.Body)
+		require.NoError(t, err)
+		w.WriteHeader(200)
+		message := clients.RecordingEvent{}
+		err = json.Unmarshal(payload, &message)
+		require.NoError(t, err)
+		require.Equal(t, "videoSomeStreamName", message.StreamName)
+		require.Equal(t, "0b152108-0bee-4333-8cb7-e859b800c57f", message.RecordingId)
+		require.Equal(t, "end", message.Event)
+		require.NotNil(t, message.Success)
+		require.True(t, *message.Success)
+		require.GreaterOrEqual(t, message.Timestamp, testStartTime)
+		require.Less(t, message.Timestamp, testStartTime+2)
+		callbackHappened <- true
+	}))
+	defer callbackServer.Close()
+	config.RecordingCallback = callbackServer.URL
+
+	router := httprouter.New()
+	router.POST("/api/mist/trigger", mistCallbackHandlers.Trigger())
+	pushOutTriggerPayload := "123\nvideoSomeStreamName\ns3+https://creds:passwd@s3.storage.com/region/livepeer-recordings-bucket/$stream/0b152108-0bee-4333-8cb7-e859b800c57f/index.m3u8\ns3+https://creds:passwd@s3.storage.com/region/livepeer-recordings-bucket/videoSomeStreamName/0b152108-0bee-4333-8cb7-e859b800c57f/index.m3u8\n[]\nnull"
+	req, _ := http.NewRequest("POST", "/api/mist/trigger", bytes.NewBuffer([]byte(pushOutTriggerPayload)))
+	req.Header.Set("X-Trigger", "PUSH_END")
+	req.Header.Set("Host", "test.livepeer.monster")
+	rr := httptest.NewRecorder()
+	router.ServeHTTP(rr, req)
+	require.Equal(t, 200, rr.Result().StatusCode)
+	select {
+	case <-callbackHappened:
+	case <-time.After(1 * time.Second):
+		require.FailNow(t, "no callback happened")
+	}
+}
+
 type StreamSample struct {
 	streamName string
 	expected   PipelineId