main.go

Rama main · réplica de lectura del repositorio original.

package main

import (
	"context"
	"errors"
	"fmt"
	"net"
	"net/http"
	"os"
	"os/signal"
	"path/filepath"
	"strconv"
	"sync"
	"syscall"
	"time"

	"github.com/n0rdy/forq/api"
	"github.com/n0rdy/forq/common"
	"github.com/n0rdy/forq/configs"
	"github.com/n0rdy/forq/db"
	"github.com/n0rdy/forq/jobs/cleanup"
	"github.com/n0rdy/forq/jobs/maintenance"
	metricsJobs "github.com/n0rdy/forq/jobs/metrics"
	"github.com/n0rdy/forq/metrics"
	"github.com/n0rdy/forq/services"
	"github.com/n0rdy/forq/ui"

	_ "github.com/golang-migrate/migrate/v4/database/sqlite"

	"github.com/golang-migrate/migrate/v4"
	"github.com/golang-migrate/migrate/v4/source/iofs"
	"github.com/rs/zerolog/log"
)

const (
	minAuthSecretLength = 32
)

func main() {
	env := getEnv()
	authSecret := getAuthSecret()
	metricsEnabled, metricsAuthSecret := getMetricsConfigs(authSecret)
	queueTtlHours, dlqTtlHours := getTtlConfigs()
	apiAddr, uiAddr := getServerAddrs()
	trustProxyHeaders := getTrustProxyHeaders()

	dbPath := getDbPath()
	log.Info().Msgf("using database file at: %s", dbPath)

	runMigrations(dbPath)

	appConfigs := configs.NewAppConfig(metricsEnabled, queueTtlHours, dlqTtlHours)

	repo, err := db.NewSQLiteRepo(dbPath, appConfigs)
	if err != nil {
		log.Fatal().Err(err).Msg("failed to create SQLite repository")
	}
	defer repo.Close()

	monitoringService := services.NewMonitoringService(repo)
	metricsService := metrics.NewMetricsService(metricsEnabled)
	queuesService := services.NewQueuesService(repo)
	messagesService := services.NewMessagesService(metricsService, repo, appConfigs)
	sessionsService := services.NewSessionsService()
	defer sessionsService.Close()
	// Separate throttle state per server: the API and UI are split into
	// distinct servers so exposing the API can't affect the UI. A shared
	// instance would re-couple them - behind a proxy every request collapses
	// to one client IP, so bad-key floods on the public API would surface as
	// lockouts on the UI login path (harmless today thanks to compare-first
	// auth, but coupling them invites a regression).
	apiThrottlingService := services.NewThrottlingService()
	defer apiThrottlingService.Close()
	uiThrottlingService := services.NewThrottlingService()
	defer uiThrottlingService.Close()

	expiredMessagesCleanupJob := cleanup.NewExpiredMessagesCleanupJob(metricsService, repo, appConfigs.JobsIntervals.ExpiredMessagesCleanupMs)
	defer expiredMessagesCleanupJob.Close()
	expiredDlqMessagesCleanupJob := cleanup.NewExpiredDlqMessagesCleanupJob(metricsService, repo, appConfigs.JobsIntervals.ExpiredDlqMessagesCleanupMs)
	defer expiredDlqMessagesCleanupJob.Close()
	failedMessagesCleanupJob := cleanup.NewFailedMessagesCleanupJob(metricsService, repo, appConfigs.JobsIntervals.FailedMessagesCleanupMs)
	defer failedMessagesCleanupJob.Close()
	failedDlqMessagesCleanupJob := cleanup.NewFailedDlqMessagesCleanupJob(metricsService, repo, appConfigs.JobsIntervals.FailedDqlMessagesCleanupMs)
	defer failedDlqMessagesCleanupJob.Close()
	staleMessagesCleanupJob := cleanup.NewStaleMessagesCleanupJob(metricsService, repo, appConfigs.JobsIntervals.StaleMessagesCleanupMs)
	defer staleMessagesCleanupJob.Close()
	dbOptimizationJob := maintenance.NewDbOptimizationJob(repo, appConfigs.JobsIntervals.DbOptimizationMs, appConfigs.JobsIntervals.DbOptimizationMaxDurationMs)
	defer dbOptimizationJob.Close()

	if metricsEnabled {
		queuesDepthMetricsJob := metricsJobs.NewQueuesDepthMetricsJob(metricsService, repo, appConfigs.JobsIntervals.QueuesDepthMetricsMs)
		defer queuesDepthMetricsJob.Close()
	}

	// shutdownCtx is cancelled on SIGINT/SIGTERM. It is also the BaseContext of
	// both servers, so request contexts (incl. in-flight long polls) are cancelled
	// immediately on shutdown instead of holding Shutdown() for up to 30s.
	shutdownCtx, stopSignals := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stopSignals()

	serverFailedCh := make(chan struct{})
	var serverFailedOnce sync.Once

	apiRouter := api.NewRouter(monitoringService, messagesService, apiThrottlingService, authSecret, metricsEnabled, metricsAuthSecret, env, trustProxyHeaders)

	var apiProtocols http.Protocols
	apiProtocols.SetUnencryptedHTTP2(true)
	apiProtocols.SetHTTP1(true)

	apiServer := &http.Server{
		Addr:              apiAddr,
		Handler:           http.TimeoutHandler(apiRouter.NewRouter(), appConfigs.ServerConfig.Timeouts.Handle, "timeout"),
		WriteTimeout:      appConfigs.ServerConfig.Timeouts.Write,
		ReadTimeout:       appConfigs.ServerConfig.Timeouts.Read,
		ReadHeaderTimeout: appConfigs.ServerConfig.Timeouts.ReadHeader,
		IdleTimeout:       appConfigs.ServerConfig.Timeouts.Idle,
		Protocols:         &apiProtocols,
		BaseContext:       func(net.Listener) context.Context { return shutdownCtx },
	}

	uiRouter := ui.NewRouter(messagesService, sessionsService, queuesService, uiThrottlingService, authSecret, env, trustProxyHeaders)

	var uiProtocols http.Protocols
	uiProtocols.SetUnencryptedHTTP2(true)
	uiProtocols.SetHTTP1(true)

	uiServer := &http.Server{
		Addr:              uiAddr,
		Handler:           http.TimeoutHandler(uiRouter.NewRouter(), appConfigs.ServerConfig.Timeouts.Handle, "timeout"),
		WriteTimeout:      appConfigs.ServerConfig.Timeouts.Write,
		ReadTimeout:       appConfigs.ServerConfig.Timeouts.Read,
		ReadHeaderTimeout: appConfigs.ServerConfig.Timeouts.ReadHeader,
		IdleTimeout:       appConfigs.ServerConfig.Timeouts.Idle,
		Protocols:         &uiProtocols,
		BaseContext:       func(net.Listener) context.Context { return shutdownCtx },
	}

	// Start API server
	go func() {
		log.Info().Msgf("Starting API server on %s", apiAddr)
		err := apiServer.ListenAndServe()
		if err != nil && !errors.Is(err, http.ErrServerClosed) {
			log.Warn().Err(err).Msg("API server failed")
			serverFailedOnce.Do(func() { close(serverFailedCh) })
		}
	}()

	// Start UI server
	go func() {
		log.Info().Msgf("Starting UI server on %s", uiAddr)
		err := uiServer.ListenAndServe()
		if err != nil && !errors.Is(err, http.ErrServerClosed) {
			log.Warn().Err(err).Msg("UI server failed")
			serverFailedOnce.Do(func() { close(serverFailedCh) })
		}
	}()

	// Block until a shutdown signal arrives or one of the servers dies.
	select {
	case <-shutdownCtx.Done():
		log.Info().Msg("shutdown signal received")
	case <-serverFailedCh:
		log.Warn().Msg("server failure, shutting down")
	}

	// cancel shutdownCtx on BOTH trigger paths: on the server-failure path it
	// is not cancelled yet, and in-flight request contexts (incl. long polls
	// on the surviving server) hang off it via BaseContext - without this they
	// would hold Shutdown until its deadline instead of returning immediately
	stopSignals()

	// In-flight requests see their contexts cancelled via BaseContext, so this
	// deadline only needs to cover response writing, not the 30s long poll.
	gracefulCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	if err := apiServer.Shutdown(gracefulCtx); err != nil {
		log.Warn().Err(err).Msg("graceful API server shutdown failed, closing forcefully")
		if err := apiServer.Close(); err != nil {
			log.Warn().Err(err).Msg("failed to close API server")
		}
	}
	if err := uiServer.Shutdown(gracefulCtx); err != nil {
		log.Warn().Err(err).Msg("graceful UI server shutdown failed, closing forcefully")
		if err := uiServer.Close(); err != nil {
			log.Warn().Err(err).Msg("failed to close UI server")
		}
	}

	log.Info().Msg("servers stopped, closing jobs and database")
	// jobs and repo are closed by the deferred Close() calls above (LIFO: jobs
	// first, repo last).
}

func getEnv() string {
	env := os.Getenv("FORQ_ENV")
	if env == "" {
		env = common.ProEnv // Default to production environment
	}

	if !common.SupportedEnvs[env] {
		log.Fatal().Msgf("unsupported environment: %s", env)
	}
	return env
}

func getAuthSecret() string {
	authSecret := os.Getenv("FORQ_AUTH_SECRET")
	if authSecret == "" {
		log.Fatal().Msg("auth secret is not provided: set FORQ_AUTH_SECRET environment variable")
	}
	if len(authSecret) < minAuthSecretLength {
		log.Fatal().Msgf("auth secret is too short: must be at least %d characters", minAuthSecretLength)
	}
	return authSecret
}

func getTrustProxyHeaders() bool {
	v := os.Getenv("FORQ_TRUST_PROXY_HEADERS")
	if v == "" {
		return false
	}
	trust, err := strconv.ParseBool(v)
	if err != nil {
		log.Fatal().Err(err).Msg("failed to parse FORQ_TRUST_PROXY_HEADERS env var")
	}
	if trust {
		log.Warn().Msg("FORQ_TRUST_PROXY_HEADERS=true: client IP will be read from X-Forwarded-For. " +
			"Forq MUST be behind a reverse proxy that strips or replaces incoming X-Forwarded-For from clients, " +
			"otherwise IPs are spoofable and throttling can be bypassed.")
	}
	return trust
}

func getMetricsConfigs(authSecret string) (bool, string) {
	metricsEnabledEnv := os.Getenv("FORQ_METRICS_ENABLED")
	if metricsEnabledEnv == "" {
		return false, "" // Metrics disabled by default
	}

	metricsEnabled, err := strconv.ParseBool(metricsEnabledEnv)
	if err != nil {
		log.Fatal().Err(err).Msg("failed to parse FORQ_METRICS_ENABLED env var")
	}

	if !metricsEnabled {
		return false, ""
	}

	metricsAuthSecret := os.Getenv("FORQ_METRICS_AUTH_SECRET")
	if metricsAuthSecret == "" {
		log.Fatal().Msg("FORQ_METRICS_AUTH_SECRET env var is required when metrics are enabled")
	}
	if len(metricsAuthSecret) < minAuthSecretLength {
		log.Fatal().Msgf("metrics auth secret is too short: must be at least %d characters", minAuthSecretLength)
	}
	// The metrics secret is a low-privilege scrape credential; keeping it
	// distinct from the full-access API token limits the blast radius of a
	// leaked Prometheus config.
	if metricsAuthSecret == authSecret {
		log.Fatal().Msg("FORQ_METRICS_AUTH_SECRET must differ from FORQ_AUTH_SECRET")
	}
	return true, metricsAuthSecret
}

func getTtlConfigs() (int, int) {
	queueTtlEnv := os.Getenv("FORQ_QUEUE_TTL_HOURS")
	dlqTtlEnv := os.Getenv("FORQ_DLQ_TTL_HOURS")

	// Default values
	queueTtlHours := 24   // 1 day
	dlqTtlHours := 7 * 24 // 7 days

	if queueTtlEnv != "" {
		parsed, err := strconv.Atoi(queueTtlEnv)
		if err != nil {
			log.Fatal().Err(err).Msg("failed to parse FORQ_QUEUE_TTL_HOURS env var")
		}
		if parsed < 1 {
			log.Fatal().Msg("FORQ_QUEUE_TTL_HOURS must be at least 1 hour")
		}
		queueTtlHours = parsed
	}

	if dlqTtlEnv != "" {
		parsed, err := strconv.Atoi(dlqTtlEnv)
		if err != nil {
			log.Fatal().Err(err).Msg("failed to parse FORQ_DLQ_TTL_HOURS env var")
		}
		if parsed < 1 {
			log.Fatal().Msg("FORQ_DLQ_TTL_HOURS must be at least 1 hour")
		}
		dlqTtlHours = parsed
	}

	return queueTtlHours, dlqTtlHours
}

func getServerAddrs() (string, string) {
	apiAddr := os.Getenv("FORQ_API_ADDR")
	if apiAddr == "" {
		apiAddr = "localhost:8080" // safe default localhost only
	}

	uiAddr := os.Getenv("FORQ_UI_ADDR")
	if uiAddr == "" {
		uiAddr = "localhost:8081" // safe default localhost only
	}

	return apiAddr, uiAddr
}

func getDbPath() string {
	dbPath := os.Getenv("FORQ_DB_PATH")
	if dbPath == "" {
		log.Fatal().Msg("database path is not provided: set FORQ_DB_PATH environment variable")
	}

	absPath, err := filepath.Abs(dbPath)
	if err != nil {
		log.Fatal().Err(err).Msgf("failed to resolve absolute path for FORQ_DB_PATH=%s", dbPath)
	}

	if err := os.MkdirAll(filepath.Dir(absPath), 0755); err != nil {
		log.Fatal().Err(err).Msgf("failed to create directory for database file at %s", absPath)
	}

	return absPath
}

func runMigrations(dbPath string) {
	// x-no-tx-wrap=true to disable transaction wrapping for PRAGMA statements, as otherwise it fails:
	// https://github.com/golang-migrate/migrate/issues/346
	dbURL := fmt.Sprintf("sqlite://file:%s?cache=shared&mode=rwc&x-no-tx-wrap=true", dbPath)

	source, err := iofs.New(db.MigrationsFS, "migrations")
	if err != nil {
		log.Fatal().Err(err).Msg("failed to create migrations source")
	}

	m, err := migrate.NewWithSourceInstance("iofs", source, dbURL)
	if err != nil {
		log.Fatal().Err(err).Msg("failed to create migration instance")
	}
	defer func() {
		srcErr, dbErr := m.Close()
		if srcErr != nil {
			log.Warn().Err(srcErr).Msg("failed to close migration source")
		}
		if dbErr != nil {
			log.Warn().Err(dbErr).Msg("failed to close migration database connection")
		}
	}()

	err = m.Up()
	if err != nil {
		if errors.Is(err, migrate.ErrNoChange) {
			log.Info().Msg("no migrations to run")
			return
		}
		log.Fatal().Err(err).Msg("failed to run migrations")
	} else {
		log.Info().Msg("migrations applied successfully")
	}
}