package main import ( "context" "fmt" "log" "net/http" "os" "thamanyah/discovery/internal/consumers" "thamanyah/discovery/internal/db" "thamanyah/discovery/internal/db/repositories" "thamanyah/discovery/internal/handlers" "thamanyah/discovery/internal/services" "time" awsconfig "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/service/sqs" _ "github.com/lib/pq" "github.com/redis/go-redis/v9" httpSwagger "github.com/swaggo/http-swagger/v2" _ "thamanyah/discovery/docs" ) // @title Thamanyah Discovery API // @version 1.0 // @description JSON API for browsing the Thamanyah catalogue. Read-side counterpart to the CMS, which is what ingests videos. // @description // @description The catalogue search is `GET /api/videos`, listed below: title is matched lexically with the last word as a prefix, categories narrow to videos filed under any one of them, limit defaults to 20 and is capped at 100, and cursor is the opaque `nextCursor` of a previous search. // @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) if err := db.Migrate(concreteDBClient); err != nil { panic(err) } log.Println("migrations applied successfully") } func runServer() { // The queue subscribed to cms's catalogue topic. This is the only AWS // service discovery talks to. catalogueQueueURL := os.Getenv("CATALOGUE_EVENTS_QUEUE_URL") if catalogueQueueURL == "" { panic(fmt.Errorf("missing required env var: CATALOGUE_EVENTS_QUEUE_URL")) } awsConfig, err := awsconfig.LoadDefaultConfig(context.Background()) if err != nil { panic(err) } concreteSQSClient := &services.SQSConcrete{ SQSClient: sqs.NewFromConfig(awsConfig), QueueURL: catalogueQueueURL, } concreteSQSClient.AssertSuccessfulConnection(context.Background()) services.SQSClient = concreteSQSClient // The ElastiCache Redis node the catalogue search is answered from. Not an // AWS API call — the address resolves inside the VPC and the security // group is what grants access — so it needs nothing from the task role and // nothing from awsConfig. // // The address is required because a missing one is a deployment mistake, // not a choice; whether the node actually answers is a different question, // and one the service deliberately survives getting "no" to. redisAddress := os.Getenv("REDIS_ADDR") if redisAddress == "" { panic(fmt.Errorf("missing required env var: REDIS_ADDR")) } concreteCacheClient := &services.RedisConcrete{ Redis: redis.NewClient(&redis.Options{ Addr: redisAddress, // Per-call deadlines on top of the budget RedisConcrete already // applies, so a connection that hangs rather than refusing cannot // tie up a goroutine past the request that opened it. Retries are // off for the same reason the budget is small: a second attempt at // a cache costs more than the query it is saving. DialTimeout: 100 * time.Millisecond, ReadTimeout: 100 * time.Millisecond, WriteTimeout: 100 * time.Millisecond, MaxRetries: -1, }), } defer func() { _ = concreteCacheClient.Close() }() concreteCacheClient.AssertSuccessfulConnection(context.Background()) services.CacheClient = concreteCacheClient concreteDBClient := db.CreateDBConnection(requireDBConnectionString()) defer db.CloseConnection(concreteDBClient) db.AssertSuccessfulConnection(context.Background(), concreteDBClient) repositories.VideoRepo = repositories.VideoRepository{ SQLDB: concreteDBClient, } // Consumes announcements for the lifetime of the process. Cancelled when // runServer returns, which today only happens if the server itself fails. consumerCtx, stopConsumer := context.WithCancel(context.Background()) defer stopConsumer() go consumers.CatalogueEvents{}.Run(consumerCtx) mux := http.NewServeMux() mux.HandleFunc("GET /health", handlers.Health) mux.HandleFunc("GET /api/videos/{id}", handlers.GetVideo) // The catalogue search. GET rather than POST because a search is safe and // idempotent, and its parameters — title, categories, limit, cursor — fit // a query string. ServeMux keeps this apart from the {id} pattern above: // the more specific one wins. mux.HandleFunc("GET /api/videos", handlers.SearchVideos) // 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 := ":8080" 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, ) }