feat: add mcpulse command

Author: sirrobot01

cmd/mcpulse/main.go

@@ -0,0 +1,222 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"flag"
+	"fmt"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"syscall"
+	"time"
+
+	"github.com/golang-migrate/migrate/v4"
+	githubPgx "github.com/golang-migrate/migrate/v4/database/pgx/v5"
+	_ "github.com/golang-migrate/migrate/v4/source/file"
+	_ "github.com/jackc/pgx/v5/stdlib"
+	"github.com/sirrobot01/mcpulse/internal/auth"
+	"github.com/sirrobot01/mcpulse/internal/config"
+	"github.com/sirrobot01/mcpulse/internal/database"
+	"github.com/sirrobot01/mcpulse/internal/ingestion"
+	"github.com/sirrobot01/mcpulse/internal/metrics"
+	"github.com/sirrobot01/mcpulse/internal/server"
+	grpcserver "github.com/sirrobot01/mcpulse/internal/server/grpc"
+	"go.uber.org/zap"
+	"google.golang.org/grpc"
+)
+
+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)
+	}
+
+	if err := cfg.Validate(); err != nil {
+		_, _ = fmt.Fprintf(os.Stderr, "invalid config: %v\n", err)
+		os.Exit(1)
+	}
+
+	logger, err := initLogger(cfg.Logging)
+	if err != nil {
+		_, _ = fmt.Fprintf(os.Stderr, "failed to initialize logger: %v\n", err)
+		os.Exit(1)
+	}
+	defer func(logger *zap.Logger) {
+		_ = logger.Sync()
+	}(logger)
+
+	logger.Info("starting MCPulse server",
+		zap.String("http_addr", fmt.Sprintf(":%d", cfg.Server.Port)),
+		zap.String("grpc_addr", fmt.Sprintf(":%d", cfg.Server.GRPCPort)),
+	)
+
+	promMetrics := metrics.NewMetrics()
+
+	ctx := context.Background()
+	dbConfig := database.Config{
+		Host:               cfg.Database.Host,
+		Port:               cfg.Database.Port,
+		Database:           cfg.Database.Database,
+		Username:           cfg.Database.Username,
+		Password:           cfg.Database.Password,
+		MaxConnections:     cfg.Database.MaxConnections,
+		MaxIdleConnections: cfg.Database.MaxIdleConnections,
+		MaxLifetime:        cfg.Database.MaxLifetime,
+	}
+
+	if err := runMigrations(dbConfig, cfg.Database.MigrationsPath, logger); err != nil {
+		logger.Fatal("failed to run database migrations", zap.Error(err))
+	}
+
+	db, err := database.New(ctx, dbConfig)
+	if err != nil {
+		logger.Fatal("failed to connect to database", zap.Error(err))
+	}
+	defer db.Close()
+
+	logger.Info("connected to database")
+
+	repo := database.NewRepository(db)
+
+	authRepo := auth.NewRepository(db)
+	jwtService := auth.NewJWTService(
+		cfg.Auth.JWTSecret,
+		cfg.Auth.AccessTokenTTL,
+		cfg.Auth.RefreshTokenTTL,
+	)
+	authService := auth.NewService(authRepo, jwtService)
+	authMiddleware := auth.NewMiddleware(jwtService, authRepo)
+
+	logger.Info("authentication enabled",
+		zap.Duration("access_token_ttl", cfg.Auth.AccessTokenTTL),
+		zap.Duration("refresh_token_ttl", cfg.Auth.RefreshTokenTTL),
+	)
+
+	ingestService := ingestion.NewService(cfg.Ingestion, cfg.Privacy, repo, promMetrics, logger)
+
+	srv := server.NewServer(cfg, repo, ingestService, promMetrics, logger, authService, authMiddleware)
+	gSrv := grpcserver.New(cfg, repo, ingestService, promMetrics, logger, authService)
+
+	serverErrors := make(chan error, 1)
+	go func() {
+		serverErrors <- srv.Start()
+	}()
+
+	grpcErrors := make(chan error, 1)
+	go func() {
+		if err := gSrv.Start(); err != nil && !errors.Is(err, grpc.ErrServerStopped) {
+			grpcErrors <- err
+		}
+		close(grpcErrors)
+	}()
+
+	shutdown := make(chan os.Signal, 1)
+	signal.Notify(shutdown, os.Interrupt, syscall.SIGTERM)
+
+	select {
+	case err := <-serverErrors:
+		logger.Fatal("server error", zap.Error(err))
+	case err := <-grpcErrors:
+		if err != nil {
+			logger.Fatal("gRPC server error", zap.Error(err))
+		}
+	case sig := <-shutdown:
+		logger.Info("received shutdown signal", zap.String("signal", sig.String()))
+
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		gSrv.Stop()
+
+		if err := srv.Stop(ctx); err != nil {
+			logger.Error("failed to gracefully shutdown server", zap.Error(err))
+		}
+		logger.Info("shutdown complete")
+	}
+}
+
+func runMigrations(cfg database.Config, migrationsPath string, logger *zap.Logger) error {
+	if migrationsPath == "" {
+		migrationsPath = "migrations"
+	}
+
+	absPath, err := filepath.Abs(migrationsPath)
+	if err != nil {
+		return fmt.Errorf("resolve migrations path: %w", err)
+	}
+
+	if info, statErr := os.Stat(absPath); statErr != nil {
+		if os.IsNotExist(statErr) {
+			return fmt.Errorf("migrations path does not exist: %s", absPath)
+		}
+		return fmt.Errorf("stat migrations path: %w", statErr)
+	} else if !info.IsDir() {
+		return fmt.Errorf("migrations path is not a directory: %s", absPath)
+	}
+
+	sourceURL := fmt.Sprintf("file://%s", absPath)
+
+	db, err := sql.Open("pgx/v5", database.ConnectionString(cfg))
+	if err != nil {
+		return fmt.Errorf("open database for migrations: %w", err)
+	}
+	defer func(db *sql.DB) {
+		err := db.Close()
+		if err != nil {
+			logger.Warn("failed to close database after migrations", zap.Error(err))
+		}
+	}(db)
+
+	driver, err := githubPgx.WithInstance(db, &githubPgx.Config{})
+	if err != nil {
+		return fmt.Errorf("create migration driver: %w", err)
+	}
+
+	m, err := migrate.NewWithDatabaseInstance(sourceURL, cfg.Database, driver)
+	if err != nil {
+		return fmt.Errorf("create migrator: %w", err)
+	}
+	defer func() {
+		if sourceErr, dbErr := m.Close(); sourceErr != nil || dbErr != nil {
+			logger.Warn("migration close error", zap.Error(errors.Join(sourceErr, dbErr)))
+		}
+	}()
+
+	if err := m.Up(); err != nil && !errors.Is(err, migrate.ErrNoChange) {
+		return fmt.Errorf("run migrations: %w", err)
+	}
+
+	logger.Info("database migrations applied", zap.String("path", absPath))
+	return nil
+}
+
+func initLogger(cfg config.LoggingConfig) (*zap.Logger, error) {
+	var zapConfig zap.Config
+
+	if cfg.Format == "json" {
+		zapConfig = zap.NewProductionConfig()
+	} else {
+		zapConfig = zap.NewDevelopmentConfig()
+	}
+
+	switch cfg.Level {
+	case "debug":
+		zapConfig.Level = zap.NewAtomicLevelAt(zap.DebugLevel)
+	case "info":
+		zapConfig.Level = zap.NewAtomicLevelAt(zap.InfoLevel)
+	case "warn":
+		zapConfig.Level = zap.NewAtomicLevelAt(zap.WarnLevel)
+	case "error":
+		zapConfig.Level = zap.NewAtomicLevelAt(zap.ErrorLevel)
+	default:
+		zapConfig.Level = zap.NewAtomicLevelAt(zap.InfoLevel)
+	}
+
+	return zapConfig.Build()
+}