thamanyah/tests/events_test.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"
)

// 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
}