FEAT: Return Playback URL in Get Videos

cms/internal/consumers/mediaconvert.go

@@ -10,6 +10,7 @@ import (
 	"encoding/json"
 	"errors"
 	"log"
+	"strings"
 	"thamanyah/cms/v2/internal/db/repositories"
 	"thamanyah/cms/v2/internal/models"
 	"thamanyah/cms/v2/internal/services"
@@ -30,9 +31,61 @@ type jobStateChange struct {
 	Detail     struct {
 		JobID  string `json:"jobId"`
 		Status string `json:"status"`
+		// Present on COMPLETE only, and only for output groups that produced
+		// something. It is how the service learns where the output landed
+		// rather than having to guess the path back from the job settings.
+		OutputGroupDetails []outputGroup `json:"outputGroupDetails"`
 	} `json:"detail"`
 }
 
+// outputGroup is one output group's result. An HLS group lists the manifests
+// it wrote under playlistFilePaths, master first; the segments are not listed.
+type outputGroup struct {
+	Type              string   `json:"type"`
+	PlaylistFilePaths []string `json:"playlistFilePaths"`
+}
+
+// MediaConvertEvents consumes MediaConvert job state changes.
+type MediaConvertEvents struct {
+	// PlaybackBaseURL is the public host the encoded output is served from —
+	// the CloudFront distribution in front of the output bucket. MediaConvert
+	// reports s3:// paths into a bucket that blocks all public access, so a
+	// path is only usable to a player once it has been rewritten onto this.
+	PlaybackBaseURL string
+}
+
+// hlsPlaylist picks the master playlist out of a finished job's output groups.
+// It returns "" when the job produced no HLS group, which is not an error: a
+// job configured with only a file group legitimately has no playlist.
+func hlsPlaylist(groups []outputGroup) string {
+	for _, group := range groups {
+		for _, playlist := range group.PlaylistFilePaths {
+			if strings.HasSuffix(playlist, ".m3u8") {
+				return playlist
+			}
+		}
+	}
+	return ""
+}
+
+// playbackURL rewrites an s3://bucket/key path onto the public delivery host.
+// The bucket is dropped rather than checked: the job writes to exactly one
+// output bucket, and that bucket is what PlaybackBaseURL fronts.
+func (c MediaConvertEvents) playbackURL(s3URI string) (string, bool) {
+	const scheme = "s3://"
+
+	if !strings.HasPrefix(s3URI, scheme) {
+		return "", false
+	}
+
+	_, key, found := strings.Cut(strings.TrimPrefix(s3URI, scheme), "/")
+	if !found || key == "" {
+		return "", false
+	}
+
+	return strings.TrimRight(c.PlaybackBaseURL, "/") + "/" + key, true
+}
+
 // statusForJobState maps MediaConvert's job states onto the catalogue's. The
 // second return is false for states that are not an outcome — the job is still
 // running, and the record should stay where it is rather than being rewritten
@@ -48,9 +101,9 @@ func statusForJobState(state string) (models.VideoStatus, bool) {
 	}
 }
 
-// RunMediaConvertEvents consumes job state changes until ctx is cancelled. It
-// is meant to be run in its own goroutine for the lifetime of the process.
-func RunMediaConvertEvents(ctx context.Context) {
+// Run consumes job state changes until ctx is cancelled. It is meant to be run
+// in its own goroutine for the lifetime of the process.
+func (c MediaConvertEvents) Run(ctx context.Context) {
 	log.Println("mediaconvert job events consumer started")
 
 	for {
@@ -71,7 +124,7 @@ func RunMediaConvertEvents(ctx context.Context) {
 		}
 
 		for _, message := range messages {
-			if !handleJobEvent(ctx, message.Body) {
+			if !c.handleJobEvent(ctx, message.Body) {
 				// Left on the queue on purpose: it becomes visible again when
 				// the visibility timeout expires, and reaches the dead-letter
 				// queue if it keeps failing.
@@ -91,7 +144,7 @@ func RunMediaConvertEvents(ctx context.Context) {
 // (a body that will never parse, an event for a job nobody has) is finished
 // with, however little it accomplished: leaving it on the queue would only
 // stall the events behind it.
-func handleJobEvent(ctx context.Context, body string) bool {
+func (c MediaConvertEvents) handleJobEvent(ctx context.Context, body string) bool {
 	var event jobStateChange
 	if err := json.Unmarshal([]byte(body), &event); err != nil {
 		log.Printf("Discarding A Job Event That Is Not JSON: %s", err)
@@ -116,7 +169,25 @@ func handleJobEvent(ctx context.Context, body string) bool {
 		return true
 	}
 
-	err := repositories.VideoRepo.UpdateVideoStatusByJobID(ctx, event.Detail.JobID, status)
+	// Only a finished job has output. A job that failed is recorded with no
+	// playback URL, which also clears one from an earlier attempt.
+	playbackURL := ""
+	if status == models.VideoStatusReady {
+		if playlist := hlsPlaylist(event.Detail.OutputGroupDetails); playlist != "" {
+			rewritten, ok := c.playbackURL(playlist)
+			if !ok {
+				log.Printf("Could Not Turn A Playlist Path Into A Playback URL: job=%q path=%q",
+					event.Detail.JobID, playlist)
+			}
+			playbackURL = rewritten
+		} else {
+			// Worth knowing about: the video is playable in principle but the
+			// catalogue has nothing to point a player at.
+			log.Printf("A Finished Job Reported No HLS Playlist: job=%q", event.Detail.JobID)
+		}
+	}
+
+	err := repositories.VideoRepo.UpdateVideoOutcomeByJobID(ctx, event.Detail.JobID, status, playbackURL)
 	if errors.Is(err, repositories.ErrVideoNotFound) {
 		// Not a fault: the topic carries every job in the account, including
 		// ones this service never submitted.
@@ -130,6 +201,6 @@ func handleJobEvent(ctx context.Context, body string) bool {
 		return false
 	}
 
-	log.Printf("Recorded A Job Outcome: job=%q status=%q", event.Detail.JobID, status)
+	log.Printf("Recorded A Job Outcome: job=%q status=%q playback=%q", event.Detail.JobID, status, playbackURL)
 	return true
 }