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
}