thamanyah/cms/main.go

package main

import (
	"context"
	"fmt"
	"log"
	"net/http"
	"os"
	"thamanyah/cms/v2/internal/consumers"
	"thamanyah/cms/v2/internal/db"
	"thamanyah/cms/v2/internal/db/repositories"
	"thamanyah/cms/v2/internal/handlers"
	"thamanyah/cms/v2/internal/services"

	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/mediaconvert"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/aws-sdk-go-v2/service/sns"
	"github.com/aws/aws-sdk-go-v2/service/sqs"
	_ "github.com/lib/pq"
	httpSwagger "github.com/swaggo/http-swagger/v2"

	_ "thamanyah/cms/v2/docs"
)

//	@title			Thamanyah CMS API
//	@version		1.0
//	@description	JSON API for ingesting videos into the Thamanyah catalogue. Uploads are two-step: presign, PUT the file straight to S3, then register the video to queue transcoding.
//	@BasePath		/

func main() {
	if len(os.Args) > 1 && os.Args[1] == "migrate" {
		runMigrate()
		return
	}
	runServer()
}

func runMigrate() {
	concreteDBClient := db.CreateDBConnection(requireDBConnectionString())
	defer db.CloseConnection(concreteDBClient)
	db.Migrate(concreteDBClient)
	log.Println("migrations applied successfully")
}

func runServer() {
	awsConfig, err := config.LoadDefaultConfig(context.Background())
	if err != nil {
		panic(fmt.Errorf("failed To Load S3 Config: %s", err))
	}

	bucketName := os.Getenv("S3_BUCKET")
	if bucketName == "" {
		panic(fmt.Errorf("missing required env var: S3_BUCKET"))
	}

	mediaConvertRole := os.Getenv("MEDIACONVERT_ROLE_ARN")
	if mediaConvertRole == "" {
		panic(fmt.Errorf("missing required env var: MEDIACONVERT_ROLE_ARN"))
	}

	mediaConvertInputBucket := os.Getenv("MEDIACONVERT_INPUT_BUCKET")
	if mediaConvertInputBucket == "" {
		panic(fmt.Errorf("missing required env var: MEDIACONVERT_INPUT_BUCKET"))
	}

	mediaConvertOutputBucket := os.Getenv("MEDIACONVERT_OUTPUT_BUCKET")
	if mediaConvertOutputBucket == "" {
		panic(fmt.Errorf("missing required env var: MEDIACONVERT_OUTPUT_BUCKET"))
	}

	mediaConvertEventsQueueURL := os.Getenv("MEDIACONVERT_EVENTS_QUEUE_URL")
	if mediaConvertEventsQueueURL == "" {
		panic(fmt.Errorf("missing required env var: MEDIACONVERT_EVENTS_QUEUE_URL"))
	}

	// The topic ready videos are announced on, which fans out to whatever has
	// subscribed — the read side's queue, today.
	catalogueEventsTopicARN := os.Getenv("CATALOGUE_EVENTS_TOPIC_ARN")
	if catalogueEventsTopicARN == "" {
		panic(fmt.Errorf("missing required env var: CATALOGUE_EVENTS_TOPIC_ARN"))
	}

	// 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 <bucket>.<endpoint host> there, which neither Docker's DNS nor
	// LocalStack's own wildcard domain serves from inside a container, so those
	// deployments need path-style URLs. Left off against real S3, which has
	// been steering away from path-style for new buckets.
	s3Options := func(o *s3.Options) {
		if os.Getenv("AWS_ENDPOINT_URL") != "" || os.Getenv("AWS_ENDPOINT_URL_S3") != "" {
			o.UsePathStyle = true
		}
	}

	concreteS3Client := &services.S3Concrete{
		S3Client: s3.NewFromConfig(awsConfig, s3Options),
		Bucket:   bucketName,
	}
	concreteS3Client.AssertSuccessfulConnection(context.Background())

	services.S3Client = concreteS3Client

	services.MediaConvertClient = &services.MediaConvertConcrete{
		MediaConvertClient: mediaconvert.NewFromConfig(awsConfig),
		Role:               mediaConvertRole,
		InputBucket:        mediaConvertInputBucket,
		OutputBucket:       mediaConvertOutputBucket,
	}

	concreteSQSClient := &services.SQSConcrete{
		SQSClient: sqs.NewFromConfig(awsConfig),
		QueueURL:  mediaConvertEventsQueueURL,
	}
	concreteSQSClient.AssertSuccessfulConnection(context.Background())

	services.SQSClient = concreteSQSClient

	concreteCatalogueClient := &services.CatalogueConcrete{
		SNSClient: sns.NewFromConfig(awsConfig),
		TopicARN:  catalogueEventsTopicARN,
	}
	concreteCatalogueClient.AssertSuccessfulConnection(context.Background())

	services.CatalogueClient = concreteCatalogueClient

	concreteDBClient := db.CreateDBConnection(requireDBConnectionString())
	defer db.CloseConnection(concreteDBClient)
	db.AssertSuccessfulConnection(context.Background(), concreteDBClient)

	repositories.VideoRepo = repositories.VideoRepository{
		SQLDB: concreteDBClient,
	}

	repositories.CatagoriesRepo = repositories.CatagoriesRepository{
		SQLDB: concreteDBClient,
	}

	// MediaConvert reports job state changes to a topic that fans out to this
	// service's queue; the consumer runs alongside the HTTP server for the
	// lifetime of the process, so a video's status catches up with its
	// transcoding job without anything calling back into this service.
	consumerCtx, stopConsumer := context.WithCancel(context.Background())
	defer stopConsumer()
	go consumers.MediaConvertEvents{PlaybackBaseURL: playbackBaseURL}.Run(consumerCtx)

	mux := http.NewServeMux()

	mux.HandleFunc("GET /health", handlers.Health)
	mux.HandleFunc("GET /api/categories", handlers.ListCategories)
	mux.HandleFunc("POST /api/videos/presign", handlers.PresignVideoUpload)
	mux.HandleFunc("POST /api/videos", handlers.CompleteVideoUpload)
	mux.HandleFunc("GET /api/videos/{id}", handlers.GetVideo)

	// Swagger UI and the generated spec. The UI assets are embedded in the
	// binary by swaggo/files, so this needs no static directory on disk.
	mux.Handle("GET /swagger/", httpSwagger.WrapHandler)

	addr := ":8081"
	log.Printf("listening on %s", addr)
	if err := http.ListenAndServe(addr, mux); err != nil {
		log.Fatal(err)
	}
}

func requireDBConnectionString() string {
	dbHost := os.Getenv("DB_HOST")
	if dbHost == "" {
		panic(fmt.Errorf("missing required env var: DB_HOST"))
	}

	dbPort := os.Getenv("DB_PORT")
	if dbPort == "" {
		panic(fmt.Errorf("missing required env var: DB_PORT"))
	}

	dbUser := os.Getenv("DB_USER")
	if dbUser == "" {
		panic(fmt.Errorf("missing required env var: DB_USER"))
	}

	dbPassword := os.Getenv("DB_PASSWORD")
	if dbPassword == "" {
		panic(fmt.Errorf("missing required env var: DB_PASSWORD"))
	}

	dbName := os.Getenv("DB_NAME")
	if dbName == "" {
		panic(fmt.Errorf("missing required env var: DB_NAME"))
	}

	// Defaults to require: RDS terminates TLS, and only a local Postgres that
	// serves none (LocalStack's RDS emulation) has any business overriding it.
	dbSSLMode := os.Getenv("DB_SSLMODE")
	if dbSSLMode == "" {
		dbSSLMode = "require"
	}

	return fmt.Sprintf(
		"host=%s port=%s user=%s password=%s dbname=%s sslmode=%s",
		dbHost, dbPort, dbUser, dbPassword, dbName, dbSSLMode,
	)
}