diff --git a/cms/docs/docs.go b/cms/docs/docs.go index c57b810..e882f0d 100644 --- a/cms/docs/docs.go +++ b/cms/docs/docs.go @@ -433,6 +433,11 @@ const docTemplate = `{ "type": "string", "example": "1755300000000-abcdef" }, + "playbackUrl": { + "description": "PlaybackURL is the HLS master playlist, ready for an m3u8-capable\nplayer. Empty until the transcoding job reports success, and for a job\nthat failed.", + "type": "string", + "example": "https://d111111abcdef8.cloudfront.net/videos/a1b2c3d4e5f6/index.m3u8" + }, "sizeBytes": { "type": "integer", "example": 60 diff --git a/cms/docs/swagger.json b/cms/docs/swagger.json index a8d7b73..f405bfd 100644 --- a/cms/docs/swagger.json +++ b/cms/docs/swagger.json @@ -426,6 +426,11 @@ "type": "string", "example": "1755300000000-abcdef" }, + "playbackUrl": { + "description": "PlaybackURL is the HLS master playlist, ready for an m3u8-capable\nplayer. Empty until the transcoding job reports success, and for a job\nthat failed.", + "type": "string", + "example": "https://d111111abcdef8.cloudfront.net/videos/a1b2c3d4e5f6/index.m3u8" + }, "sizeBytes": { "type": "integer", "example": 60 diff --git a/cms/docs/swagger.yaml b/cms/docs/swagger.yaml index c8dc9e0..f283ce4 100644 --- a/cms/docs/swagger.yaml +++ b/cms/docs/swagger.yaml @@ -152,6 +152,13 @@ definitions: mediaConvertJobId: example: 1755300000000-abcdef type: string + playbackUrl: + description: |- + PlaybackURL is the HLS master playlist, ready for an m3u8-capable + player. Empty until the transcoding job reports success, and for a job + that failed. + example: https://d111111abcdef8.cloudfront.net/videos/a1b2c3d4e5f6/index.m3u8 + type: string sizeBytes: example: 60 type: integer diff --git a/cms/internal/api/videos.go b/cms/internal/api/videos.go index 4b26fc1..bd12e3b 100644 --- a/cms/internal/api/videos.go +++ b/cms/internal/api/videos.go @@ -61,16 +61,20 @@ type CompleteResponse struct { // deliberately a separate type, so the shape POST publishes and the shape GET // publishes can diverge without one silently dragging the other with it. type VideoResponse struct { - ID string `json:"id" example:"0199f3a1-7c2e-7b21-9f0d-1a2b3c4d5e6f"` - Title string `json:"title" example:"Inside the Newsroom"` - Description string `json:"description" example:"A behind-the-scenes look at the evening bulletin."` - CategoryIDs []int16 `json:"categoryIds" example:"1,2"` - Tags string `json:"tags" example:"media, press, riyadh"` - FileName string `json:"fileName" example:"interview-cut.mov"` - StorageKey string `json:"storageKey" example:"videos/a1b2c3d4e5f6.mov"` - MediaConvertJobID string `json:"mediaConvertJobId" example:"1755300000000-abcdef"` - Status string `json:"status" enums:"processing,ready,failed" example:"processing"` - SizeBytes int64 `json:"sizeBytes" example:"60"` - CreatedAt time.Time `json:"createdAt"` - UpdatedAt time.Time `json:"updatedAt"` + ID string `json:"id" example:"0199f3a1-7c2e-7b21-9f0d-1a2b3c4d5e6f"` + Title string `json:"title" example:"Inside the Newsroom"` + Description string `json:"description" example:"A behind-the-scenes look at the evening bulletin."` + CategoryIDs []int16 `json:"categoryIds" example:"1,2"` + Tags string `json:"tags" example:"media, press, riyadh"` + FileName string `json:"fileName" example:"interview-cut.mov"` + StorageKey string `json:"storageKey" example:"videos/a1b2c3d4e5f6.mov"` + MediaConvertJobID string `json:"mediaConvertJobId" example:"1755300000000-abcdef"` + Status string `json:"status" enums:"processing,ready,failed" example:"processing"` + // PlaybackURL is the HLS master playlist, ready for an m3u8-capable + // player. Empty until the transcoding job reports success, and for a job + // that failed. + PlaybackURL string `json:"playbackUrl" example:"https://d111111abcdef8.cloudfront.net/videos/a1b2c3d4e5f6/index.m3u8"` + SizeBytes int64 `json:"sizeBytes" example:"60"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` } diff --git a/cms/internal/consumers/mediaconvert.go b/cms/internal/consumers/mediaconvert.go index 4ce4f7b..3e8d769 100644 --- a/cms/internal/consumers/mediaconvert.go +++ b/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 } diff --git a/cms/internal/db/migrations/0004_add_videos_playback_url.down.sql b/cms/internal/db/migrations/0004_add_videos_playback_url.down.sql new file mode 100644 index 0000000..03a1da2 --- /dev/null +++ b/cms/internal/db/migrations/0004_add_videos_playback_url.down.sql @@ -0,0 +1 @@ +ALTER TABLE videos DROP COLUMN playback_url; diff --git a/cms/internal/db/migrations/0004_add_videos_playback_url.up.sql b/cms/internal/db/migrations/0004_add_videos_playback_url.up.sql new file mode 100644 index 0000000..11d4167 --- /dev/null +++ b/cms/internal/db/migrations/0004_add_videos_playback_url.up.sql @@ -0,0 +1,5 @@ +-- Where the finished HLS playlist is served from, written by the job events +-- consumer when MediaConvert reports the job complete. Empty until then, and +-- for videos whose job failed — so it is DEFAULT '' rather than nullable: the +-- absence of a playback URL is "not ready yet", not "unknown". +ALTER TABLE videos ADD COLUMN playback_url TEXT NOT NULL DEFAULT ''; diff --git a/cms/internal/db/repositories/videos.go b/cms/internal/db/repositories/videos.go index 111f921..b347d04 100644 --- a/cms/internal/db/repositories/videos.go +++ b/cms/internal/db/repositories/videos.go @@ -15,15 +15,18 @@ type VideoRepository struct { var VideoRepo VideoRepository -// UpdateVideoStatusByJobID moves the video a transcoding job belongs to into a -// new status. It returns ErrVideoNotFound when no video carries that job id, -// which the consumer treats as a message to drop rather than a fault. -func (svc VideoRepository) UpdateVideoStatusByJobID(ctx context.Context, jobID string, status models.VideoStatus) error { +// UpdateVideoOutcomeByJobID records what became of a transcoding job: the +// status it ended in, and the playback URL its output is served from. A job +// that failed has no output, so playbackURL is empty for those and the column +// is cleared along with the status. It returns ErrVideoNotFound when no video +// carries that job id, which the consumer treats as a message to drop rather +// than a fault. +func (svc VideoRepository) UpdateVideoOutcomeByJobID(ctx context.Context, jobID string, status models.VideoStatus, playbackURL string) error { result, err := svc.SQLDB.ExecContext(ctx, ` UPDATE videos - SET status = $1, updated_at = now() - WHERE mediaconvert_job_id = $2 - `, string(status), jobID) + SET status = $1, playback_url = $2, updated_at = now() + WHERE mediaconvert_job_id = $3 + `, string(status), playbackURL, jobID) if err != nil { return err } @@ -45,11 +48,11 @@ func (svc VideoRepository) GetVideoByID(ctx context.Context, id string) (models. var v models.Video err := svc.SQLDB.QueryRowContext(ctx, ` - SELECT id, title, description, tags, file_name, storage_key, mediaconvert_job_id, status, size_bytes, created_at, updated_at + SELECT id, title, description, tags, file_name, storage_key, mediaconvert_job_id, status, playback_url, size_bytes, created_at, updated_at FROM videos WHERE id = $1 `, id).Scan(&v.ID, &v.Title, &v.Description, &v.Tags, &v.FileName, &v.StorageKey, - &v.MediaConvertJobID, &v.Status, &v.SizeBytes, &v.CreatedAt, &v.UpdatedAt) + &v.MediaConvertJobID, &v.Status, &v.PlaybackURL, &v.SizeBytes, &v.CreatedAt, &v.UpdatedAt) if errors.Is(err, sql.ErrNoRows) { return models.Video{}, ErrVideoNotFound } diff --git a/cms/internal/handlers/videos.go b/cms/internal/handlers/videos.go index 7649c5f..492b9bf 100644 --- a/cms/internal/handlers/videos.go +++ b/cms/internal/handlers/videos.go @@ -309,6 +309,7 @@ func GetVideo(w http.ResponseWriter, r *http.Request) { StorageKey: video.StorageKey, MediaConvertJobID: video.MediaConvertJobID, Status: video.Status.String(), + PlaybackURL: video.PlaybackURL, SizeBytes: video.SizeBytes, CreatedAt: video.CreatedAt, UpdatedAt: video.UpdatedAt, diff --git a/cms/internal/models/video.go b/cms/internal/models/video.go index 128753e..7fcd11c 100644 --- a/cms/internal/models/video.go +++ b/cms/internal/models/video.go @@ -12,6 +12,7 @@ type Video struct { StorageKey string MediaConvertJobID string Status VideoStatus + PlaybackURL string SizeBytes int64 CreatedAt time.Time UpdatedAt time.Time diff --git a/cms/internal/services/mediaconvert.go b/cms/internal/services/mediaconvert.go index fb3be6c..f5e827b 100644 --- a/cms/internal/services/mediaconvert.go +++ b/cms/internal/services/mediaconvert.go @@ -3,6 +3,8 @@ package services import ( "context" "fmt" + "path" + "strings" "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/service/mediaconvert" @@ -15,6 +17,11 @@ type MediaConvert interface { QueueEncodingJob(ctx context.Context, key string) (string, error) } +// segmentLengthSeconds is how long each HLS segment runs. Six seconds is the +// common default: long enough to keep the playlist and the request count down, +// short enough that a player can switch rendition without a long stall. +const segmentLengthSeconds = 6 + type MediaConvertConcrete struct { MediaConvertClient *mediaconvert.Client Role string @@ -24,7 +31,14 @@ type MediaConvertConcrete struct { func (svc MediaConvertConcrete) QueueEncodingJob(ctx context.Context, key string) (string, error) { input := fmt.Sprintf("s3://%s/%s", svc.InputBucket, key) - destination := fmt.Sprintf("s3://%s/%s", svc.OutputBucket, key) + + // HlsGroupSettings.Destination is a *filename base*, not a directory and + // not a finished name: MediaConvert appends each output's NameModifier and + // the extensions it needs. Giving each video its own folder keeps the + // master playlist, the variant playlists and the segments together, and + // keeps two videos from interleaving their segments in one prefix. + base := strings.TrimSuffix(key, path.Ext(key)) + destination := fmt.Sprintf("s3://%s/%s/index", svc.OutputBucket, base) job, err := svc.MediaConvertClient.CreateJob(ctx, &mediaconvert.CreateJobInput{ Role: &svc.Role, @@ -42,24 +56,42 @@ func (svc MediaConvertConcrete) QueueEncodingJob(ctx context.Context, key string }, OutputGroups: []types.OutputGroup{ { - Name: aws.String("File Group"), + Name: aws.String("HLS Group"), OutputGroupSettings: &types.OutputGroupSettings{ - Type: types.OutputGroupTypeFileGroupSettings, - FileGroupSettings: &types.FileGroupSettings{ - Destination: &destination, + Type: types.OutputGroupTypeHlsGroupSettings, + HlsGroupSettings: &types.HlsGroupSettings{ + Destination: &destination, + SegmentLength: aws.Int32(segmentLengthSeconds), + // 0 lets the last segment be as short as it needs to + // be rather than padding the ones before it. + MinSegmentLength: aws.Int32(0), }, }, Outputs: []types.Output{ { + // One rendition. A ladder is another entry here per + // rendition, each with its own NameModifier and + // height — the master playlist lists whatever is + // present, so nothing downstream changes. + NameModifier: aws.String("_720p"), ContainerSettings: &types.ContainerSettings{ - Container: types.ContainerTypeMp4, + Container: types.ContainerTypeM3u8, + M3u8Settings: &types.M3u8Settings{}, }, VideoDescription: &types.VideoDescription{ + Height: aws.Int32(720), CodecSettings: &types.VideoCodecSettings{ Codec: types.VideoCodecH264, H264Settings: &types.H264Settings{ RateControlMode: types.H264RateControlModeQvbr, - MaxBitrate: aws.Int32(5000000), + MaxBitrate: aws.Int32(3000000), + // A segment has to start on a keyframe, + // so the GOP is pinned to a divisor of + // the segment length. Left to follow the + // source, segments come out uneven and + // players stall at the joins. + GopSizeUnits: types.H264GopSizeUnitsSeconds, + GopSize: aws.Float64(2), }, }, }, diff --git a/cms/main.go b/cms/main.go index f64f675..74a7cca 100644 --- a/cms/main.go +++ b/cms/main.go @@ -73,6 +73,15 @@ func runServer() { panic(fmt.Errorf("missing required env var: MEDIACONVERT_EVENTS_QUEUE_URL")) } + // The public host the encoded output is served from — the CloudFront + // distribution in front of the output bucket. MediaConvert reports where + // it wrote as an s3:// path into a bucket that blocks public access, so + // this is what makes a finished video reachable by a player. + playbackBaseURL := os.Getenv("PLAYBACK_BASE_URL") + if playbackBaseURL == "" { + panic(fmt.Errorf("missing required env var: PLAYBACK_BASE_URL")) + } + // AWS_ENDPOINT_URL is only ever set when pointing at something other than // real S3 — LocalStack, in docker-compose. Virtual-host addressing would // resolve . there, which neither Docker's DNS nor @@ -126,7 +135,7 @@ func runServer() { // transcoding job without anything calling back into this service. consumerCtx, stopConsumer := context.WithCancel(context.Background()) defer stopConsumer() - go consumers.RunMediaConvertEvents(consumerCtx) + go consumers.MediaConvertEvents{PlaybackBaseURL: playbackBaseURL}.Run(consumerCtx) mux := http.NewServeMux() diff --git a/docker-compose.yml b/docker-compose.yml index 64045b0..c46a813 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -29,6 +29,13 @@ services: # infrastructure/Pulumi.local.yaml works from the host; the alias # makes the same name resolve here from inside the network. - localhost.localstack.cloud + # S3 Control prefixes its endpoint host with the caller's account id + # — the SDK does this even for a custom endpoint — so the bucket-tag + # read goes to .localhost.localstack.cloud. The public + # wildcard points that at 127.0.0.1, which inside the network is the + # calling container, not this one. LocalStack's account is always + # 000000000000, so one more alias covers it. + - 000000000000.localhost.localstack.cloud # Provisions the buckets, the Postgres instance and the IAM roles inside # LocalStack, by running infrastructure/ — the same program that deploys @@ -93,6 +100,10 @@ services: # of the process. Pinned by the localstack branch in infrastructure/main.go; # LocalStack always uses account 000000000000. - MEDIACONVERT_EVENTS_QUEUE_URL=${MEDIACONVERT_EVENTS_QUEUE_URL:-http://localhost.localstack.cloud:4566/000000000000/cms-mediaconvert-events} + # Where finished HLS output is served from. In AWS this is the CloudFront + # distribution; there is none under LocalStack, so the encoded bucket is + # addressed directly — path-style, for the same reason S3 is elsewhere. + - PLAYBACK_BASE_URL=${PLAYBACK_BASE_URL:-http://localhost.localstack.cloud:4566/encoded-bucket} # Postgres: the RDS instance LocalStack provisions, which runs inside the # localstack container and speaks plain TCP — hence sslmode=disable. - DB_HOST=${DB_HOST:-localhost.localstack.cloud}