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 . 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, ) }