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" ) // 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" ) // 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 } func newEventPublisher(ctx context.Context) (*eventPublisher, error) { endpoint := envOr("AWS_ENDPOINT_URL", defaultEndpointURL) region := envOr("AWS_REGION", defaultRegion) cfg, err := awsconfig.LoadDefaultConfig(ctx, awsconfig.WithRegion(region), // 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"), "")), ) if err != nil { return nil, err } 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 }