FahdShalhoub
260 lines
9.3 KiB
Go
260 lines
9.3 KiB
Go
package tests
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"path"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/aws/aws-sdk-go-v2/aws"
|
|
awsconfig "github.com/aws/aws-sdk-go-v2/config"
|
|
"github.com/aws/aws-sdk-go-v2/credentials"
|
|
"github.com/aws/aws-sdk-go-v2/service/sns"
|
|
"github.com/aws/aws-sdk-go-v2/service/sqs"
|
|
)
|
|
|
|
// These scenarios stand in for EventBridge, not for a client: they publish the
|
|
// same "MediaConvert Job State Change" event AWS would put on the topic when a
|
|
// job changes state. That is the seam cms owns — the topic, the queue, the
|
|
// consumer and the record it updates. The EventBridge rule that feeds the
|
|
// topic in AWS is Pulumi configuration, and is not exercised here.
|
|
//
|
|
// This is the only place the suite reaches for the AWS SDK. Everything a real
|
|
// client does still goes over plain HTTP.
|
|
|
|
const (
|
|
defaultEndpointURL = "http://localhost.localstack.cloud:4566"
|
|
defaultTopicARN = "arn:aws:sns:us-east-1:000000000000:mediaconvert-job-events"
|
|
defaultRegion = "us-east-1"
|
|
defaultOutputBucket = "encoded-bucket"
|
|
defaultPlaybackBaseURL = "http://localhost.localstack.cloud:4566/encoded-bucket"
|
|
// The suite's own queue on the catalogue topic cms announces ready videos
|
|
// to. The suite reads a queue rather than the topic because that is what a
|
|
// subscriber sees — SNS has nothing to poll.
|
|
//
|
|
// Deliberately not discovery's queue: a consumer destroys what it reads, so
|
|
// sharing one would have the suite and the running service race for every
|
|
// announcement and each see about half. Pinned by the localstack branch in
|
|
// infrastructure/main.go; LocalStack always uses account 000000000000.
|
|
defaultCatalogueQueueURL = "http://localhost.localstack.cloud:4566/000000000000/tests-catalogue-events"
|
|
)
|
|
|
|
// playbackBaseURL is the public host the encoded output is served from — the
|
|
// CloudFront distribution in AWS, LocalStack's own S3 endpoint here. It has to
|
|
// match what cms is configured with, since the whole point of the rewrite is
|
|
// that the two agree.
|
|
func playbackBaseURL() string {
|
|
return strings.TrimRight(envOr("PLAYBACK_BASE_URL", defaultPlaybackBaseURL), "/")
|
|
}
|
|
|
|
// playlistPathFor is where the HLS output group writes the master playlist for
|
|
// a given source key: one folder per video, named after the source object.
|
|
// QueueEncodingJob derives the job's destination the same way.
|
|
func playlistPathFor(storageKey string) string {
|
|
base := strings.TrimSuffix(storageKey, path.Ext(storageKey))
|
|
return fmt.Sprintf("s3://%s/%s/index.m3u8", envOr("MEDIACONVERT_OUTPUT_BUCKET", defaultOutputBucket), base)
|
|
}
|
|
|
|
// playbackURLFor is what cms should end up storing for that same key.
|
|
func playbackURLFor(storageKey string) string {
|
|
base := strings.TrimSuffix(storageKey, path.Ext(storageKey))
|
|
return fmt.Sprintf("%s/%s/index.m3u8", playbackBaseURL(), base)
|
|
}
|
|
|
|
func envOr(name, fallback string) string {
|
|
if v := strings.TrimSpace(os.Getenv(name)); v != "" {
|
|
return v
|
|
}
|
|
return fallback
|
|
}
|
|
|
|
// jobStateChange is the EventBridge event MediaConvert emits. Only the members
|
|
// cms reads are modelled; AWS sends a good deal more.
|
|
type jobStateChange struct {
|
|
Version string `json:"version"`
|
|
ID string `json:"id"`
|
|
DetailType string `json:"detail-type"`
|
|
Source string `json:"source"`
|
|
Account string `json:"account"`
|
|
Time time.Time `json:"time"`
|
|
Region string `json:"region"`
|
|
Resources []string `json:"resources"`
|
|
Detail jobStateDetail `json:"detail"`
|
|
}
|
|
|
|
type jobStateDetail struct {
|
|
Timestamp int64 `json:"timestamp"`
|
|
AccountID string `json:"accountId"`
|
|
Queue string `json:"queue"`
|
|
JobID string `json:"jobId"`
|
|
Status string `json:"status"`
|
|
OutputGroupDetails []outputGroup `json:"outputGroupDetails,omitempty"`
|
|
}
|
|
|
|
// outputGroup is one output group's result. An HLS group reports the manifests
|
|
// it wrote under playlistFilePaths; the segments themselves are not listed.
|
|
type outputGroup struct {
|
|
Type string `json:"type"`
|
|
PlaylistFilePaths []string `json:"playlistFilePaths,omitempty"`
|
|
}
|
|
|
|
// eventPublisher puts job state changes on the topic cms's queue subscribes to.
|
|
type eventPublisher struct {
|
|
sns *sns.Client
|
|
topicARN string
|
|
}
|
|
|
|
// awsConfig is the SDK configuration both ends of the event seam use: the
|
|
// publisher that stands in for EventBridge, and the reader that stands in for
|
|
// whoever consumes the catalogue queue.
|
|
func awsConfig(ctx context.Context) (aws.Config, error) {
|
|
return awsconfig.LoadDefaultConfig(ctx,
|
|
awsconfig.WithRegion(envOr("AWS_REGION", defaultRegion)),
|
|
// LocalStack accepts any credentials; these keep the SDK from hunting
|
|
// for a profile that a developer's machine may not have.
|
|
awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(
|
|
envOr("AWS_ACCESS_KEY_ID", "test"), envOr("AWS_SECRET_ACCESS_KEY", "test"), "")),
|
|
)
|
|
}
|
|
|
|
func newEventPublisher(ctx context.Context) (*eventPublisher, error) {
|
|
cfg, err := awsConfig(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
endpoint := envOr("AWS_ENDPOINT_URL", defaultEndpointURL)
|
|
|
|
return &eventPublisher{
|
|
sns: sns.NewFromConfig(cfg, func(o *sns.Options) { o.BaseEndpoint = aws.String(endpoint) }),
|
|
topicARN: envOr("MEDIACONVERT_EVENTS_TOPIC_ARN", defaultTopicARN),
|
|
}, nil
|
|
}
|
|
|
|
// publishJobState sends the event MediaConvert would send. playlistPath is the
|
|
// master playlist an HLS output group reports on COMPLETE; pass "" to send a
|
|
// COMPLETE that names no playlist, which is what a job with no HLS group does.
|
|
func (p *eventPublisher) publishJobState(ctx context.Context, jobID, state, playlistPath string) error {
|
|
event := jobStateChange{
|
|
Version: "0",
|
|
ID: fmt.Sprintf("test-%d", time.Now().UnixNano()),
|
|
DetailType: "MediaConvert Job State Change",
|
|
Source: "aws.mediaconvert",
|
|
Account: "000000000000",
|
|
Time: time.Now().UTC(),
|
|
Region: envOr("AWS_REGION", defaultRegion),
|
|
Resources: []string{fmt.Sprintf("arn:aws:mediaconvert:%s:000000000000:jobs/%s", envOr("AWS_REGION", defaultRegion), jobID)},
|
|
Detail: jobStateDetail{
|
|
Timestamp: time.Now().UnixMilli(),
|
|
AccountID: "000000000000",
|
|
JobID: jobID,
|
|
Status: state,
|
|
},
|
|
}
|
|
|
|
if playlistPath != "" {
|
|
event.Detail.OutputGroupDetails = []outputGroup{{
|
|
Type: "HLS_GROUP",
|
|
PlaylistFilePaths: []string{playlistPath},
|
|
}}
|
|
}
|
|
|
|
body, err := json.Marshal(event)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
_, err = p.sns.Publish(ctx, &sns.PublishInput{
|
|
TopicArn: aws.String(p.topicARN),
|
|
Message: aws.String(string(body)),
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("publishing to %s: %w", p.topicARN, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// --- the catalogue queue ----------------------------------------------------
|
|
|
|
// catalogueAnnouncement is what cms puts on the catalogue queue when a video
|
|
// becomes ready. This is the published contract, so the suite spells it out
|
|
// rather than importing it: a change here is a change other services see.
|
|
type catalogueAnnouncement struct {
|
|
VideoID string `json:"videoId"`
|
|
Title string `json:"title"`
|
|
PlaybackURL string `json:"playbackUrl"`
|
|
Categories []string `json:"categories"`
|
|
}
|
|
|
|
// catalogueReader stands in for the read side, draining its own queue on the
|
|
// catalogue topic. It consumes what it reads, exactly as a real subscriber
|
|
// would. The subscription delivers raw, so a message body is the announcement
|
|
// itself with no SNS envelope to unwrap.
|
|
type catalogueReader struct {
|
|
sqs *sqs.Client
|
|
queueURL string
|
|
}
|
|
|
|
func newCatalogueReader(ctx context.Context) (*catalogueReader, error) {
|
|
cfg, err := awsConfig(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
endpoint := envOr("AWS_ENDPOINT_URL", defaultEndpointURL)
|
|
|
|
return &catalogueReader{
|
|
sqs: sqs.NewFromConfig(cfg, func(o *sqs.Options) { o.BaseEndpoint = aws.String(endpoint) }),
|
|
queueURL: envOr("CATALOGUE_QUEUE_URL", defaultCatalogueQueueURL),
|
|
}, nil
|
|
}
|
|
|
|
// await waits for the announcement naming videoID and returns it. Messages for
|
|
// other videos are consumed and discarded on the way: scenarios run one at a
|
|
// time, so anything else on the queue is a leftover, and leaving it would only
|
|
// make the next scenario wade through it. It returns nil if nothing names that
|
|
// video before the deadline.
|
|
func (r *catalogueReader) await(ctx context.Context, videoID string, within time.Duration) (*catalogueAnnouncement, error) {
|
|
deadline := time.Now().Add(within)
|
|
|
|
for time.Now().Before(deadline) {
|
|
output, err := r.sqs.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
|
|
QueueUrl: aws.String(r.queueURL),
|
|
MaxNumberOfMessages: 10,
|
|
WaitTimeSeconds: 2,
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("receiving from %s: %w", r.queueURL, err)
|
|
}
|
|
|
|
var found *catalogueAnnouncement
|
|
for _, message := range output.Messages {
|
|
if message.Body == nil || message.ReceiptHandle == nil {
|
|
continue
|
|
}
|
|
|
|
var announcement catalogueAnnouncement
|
|
if err := json.Unmarshal([]byte(*message.Body), &announcement); err == nil &&
|
|
announcement.VideoID == videoID {
|
|
found = &announcement
|
|
}
|
|
|
|
if _, err := r.sqs.DeleteMessage(ctx, &sqs.DeleteMessageInput{
|
|
QueueUrl: aws.String(r.queueURL),
|
|
ReceiptHandle: message.ReceiptHandle,
|
|
}); err != nil {
|
|
return nil, fmt.Errorf("acknowledging a message on %s: %w", r.queueURL, err)
|
|
}
|
|
}
|
|
|
|
if found != nil {
|
|
return found, nil
|
|
}
|
|
}
|
|
|
|
return nil, nil
|
|
}
|