metadata-agregator/cmd/server/main.go

package main

import (
	"context"
	"flag"
	"fmt"
	"net"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/logging"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/rs/zerolog/log"
	"google.golang.org/grpc"
	"google.golang.org/grpc/health"
	healthpb "google.golang.org/grpc/health/grpc_health_v1"
	"google.golang.org/grpc/reflection"

	"github.com/metadata-agregator/internal/config"
	applog "github.com/metadata-agregator/internal/logging"
	"github.com/metadata-agregator/internal/metrics"
	"github.com/metadata-agregator/internal/provider/musicbrainz"
	"github.com/metadata-agregator/internal/repository/postgres"
	"github.com/metadata-agregator/internal/server"
	"github.com/metadata-agregator/internal/service"
	metadatav1 "github.com/metadata-agregator/pkg/gen/metadata/v1"
)

func main() {
	configPath := flag.String("config", "", "path to config file")
	flag.Parse()

	cfg, err := config.Load(*configPath)
	if err != nil {
		fmt.Fprintf(os.Stderr, "failed to load config: %v\n", err)
		os.Exit(1)
	}

	applog.Init(cfg.Logging)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	services, cleanup := buildServices(ctx, cfg)
	defer cleanup()

	srvMetrics := metrics.NewServerMetrics()

	addr := fmt.Sprintf(":%d", cfg.Server.Port)

	lis, err := net.Listen("tcp", addr)
	if err != nil {
		log.Fatal().Err(err).Str("addr", addr).Msg("failed to listen")
	}

	grpcServer := grpc.NewServer(
		grpc.ChainUnaryInterceptor(
			applog.RequestIDInterceptor(),
			applog.PeerInfoInterceptor(),
			srvMetrics.UnaryServerInterceptor(),
			logging.UnaryServerInterceptor(applog.InterceptorLogger(), applog.LoggingOpts()...),
		),
		grpc.ChainStreamInterceptor(
			applog.StreamRequestIDInterceptor(),
			srvMetrics.StreamServerInterceptor(),
			logging.StreamServerInterceptor(applog.InterceptorLogger(), applog.LoggingOpts()...),
		),
	)
	metadatav1.RegisterMetadataServiceServer(grpcServer, server.NewMetadataServer(services))
	healthpb.RegisterHealthServer(grpcServer, health.NewServer())
	reflection.Register(grpcServer)

	srvMetrics.InitializeMetrics(grpcServer)

	if cfg.Metrics.Enabled {
		go srvMetrics.StartHTTPServer(ctx, cfg.Metrics.Port)
	}

	go gracefulShutdown(grpcServer, cancel)

	log.Info().Str("addr", addr).Msg("gRPC server listening")
	if err := grpcServer.Serve(lis); err != nil {
		log.Fatal().Err(err).Msg("failed to serve")
	}
}

func buildServices(ctx context.Context, cfg *config.Config) (map[metadatav1.Provider]*service.MetadataService, func()) {
	mb := musicbrainz.New()
	services := make(map[metadatav1.Provider]*service.MetadataService)

	dbURL := cfg.Database.DSN()
	if dbURL == "" {
		dbURL = os.Getenv("DATABASE_URL")
	}

	if dbURL == "" {
		log.Warn().Msg("no database configured, running in provider-only mode")
		services[metadatav1.Provider_PROVIDER_MUSICBRAINZ] = service.NewMetadataService(
			&noopArtistRepo{},
			&noopAlbumRepo{},
			&noopTrackRepo{},
			mb,
		)
		return services, func() {}
	}

	pool, err := connectDB(ctx, dbURL)
	if err != nil {
		log.Fatal().Err(err).Msg("database connection failed")
	}

	artistRepo := postgres.NewArtistRepository(pool)
	albumRepo := postgres.NewAlbumRepository(pool)
	trackRepo := postgres.NewTrackRepository(pool)

	services[metadatav1.Provider_PROVIDER_MUSICBRAINZ] = service.NewMetadataService(
		artistRepo,
		albumRepo,
		trackRepo,
		mb,
	)

	log.Info().Msg("database connected, caching enabled")
	return services, func() { pool.Close() }
}

func connectDB(ctx context.Context, dbURL string) (*pgxpool.Pool, error) {
	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()

	pgxCfg, err := pgxpool.ParseConfig(dbURL)
	if err != nil {
		return nil, err
	}

	pgxCfg.MaxConns = 10
	pgxCfg.MinConns = 2

	log.Debug().Msg("connecting to database")

	pool, err := pgxpool.NewWithConfig(ctx, pgxCfg)
	if err != nil {
		return nil, err
	}

	if err := pool.Ping(ctx); err != nil {
		pool.Close()
		return nil, err
	}

	return pool, nil
}

func gracefulShutdown(srv *grpc.Server, cancel context.CancelFunc) {
	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
	sig := <-sigCh
	log.Info().Str("signal", sig.String()).Msg("shutting down")
	cancel()
	srv.GracefulStop()
}