thamanyah/discovery/main.go

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