proxy-pool/internal/controller/bootstrap/infrastructure.go

package bootstrap

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"strings"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/redis/go-redis/v9"

	"proxy-pool/internal/adapters/postgresadmin"
	"proxy-pool/internal/adapters/redisactivity"
	"proxy-pool/internal/adapters/redisprovider"
	"proxy-pool/internal/config"
	controllerProvider "proxy-pool/internal/controller/provider"
	"proxy-pool/internal/platform/credentials"
	platformMetrics "proxy-pool/internal/platform/metrics"
)

const (
	redisNamespace        = "controller"
	redisOperationTTL     = 30 * time.Second
	redisMinimumScan      = 4_096
	redisMaximumScan      = config.MaximumPoolSize
	redisCleanupLimit     = 1_024
	providerLeaseTTL      = 15 * time.Second
	providerRenewEvery    = 3 * time.Second
	providerRetryInterval = 100 * time.Millisecond
	providerPermitGrace   = 5 * time.Second
)

var (
	ErrPostgresConfiguration = errors.New("invalid PostgreSQL configuration")
	ErrPostgresUnavailable   = errors.New("PostgreSQL unavailable")
	ErrRedisConfiguration    = errors.New("invalid Redis configuration")
	ErrRedisUnavailable      = errors.New("Redis unavailable")
)

type productionInfrastructure struct {
	holderID  string
	namespace string
}

func (infrastructure *productionInfrastructure) Open(
	ctx context.Context,
	configuration *config.Config,
) (_ ports, resultErr error) {
	if ctx == nil || configuration == nil {
		return ports{}, ErrInvalidOptions
	}
	namespace, err := resolveRedisNamespace(infrastructure.namespace)
	if err != nil {
		return ports{}, err
	}
	var postgresPool *pgxpool.Pool
	var redisClient *redis.Client
	closeResources := func() error {
		var closeErr error
		if redisClient != nil {
			closeErr = redisClient.Close()
		}
		if postgresPool != nil {
			postgresPool.Close()
		}
		return closeErr
	}
	defer func() {
		if resultErr != nil {
			_ = closeResources()
		}
	}()

	opened := ports{close: closeResources}
	if configuration.Admin.Enabled {
		if strings.TrimSpace(configuration.Storage.PostgresURL) == "" {
			return ports{}, ErrPostgresConfiguration
		}
		poolConfig, err := pgxpool.ParseConfig(configuration.Storage.PostgresURL)
		if err != nil {
			return ports{}, ErrPostgresConfiguration
		}
		poolConfig.ConnConfig.RuntimeParams["application_name"] = "proxy-controller"
		postgresPool, err = pgxpool.NewWithConfig(ctx, poolConfig)
		if err != nil {
			return ports{}, ErrPostgresUnavailable
		}
		if err = postgresPool.Ping(ctx); err != nil {
			return ports{}, contextOr(ctx, ErrPostgresUnavailable)
		}
		if err = postgresadmin.ApplyMigrations(ctx, postgresPool); err != nil {
			return ports{}, err
		}
		opened.state, err = postgresadmin.New(postgresPool)
		if err != nil {
			return ports{}, err
		}
	}

	providersEnabled := hasEnabledUpstream(configuration)
	if configuration.Distribution.Enabled || configuration.Admin.Enabled || providersEnabled {
		if strings.TrimSpace(configuration.Storage.RedisURL) == "" {
			return ports{}, ErrRedisConfiguration
		}
		redisOptions, err := redis.ParseURL(configuration.Storage.RedisURL)
		if err != nil {
			return ports{}, ErrRedisConfiguration
		}
		redisClient = redis.NewClient(redisOptions)
		if err = redisClient.Ping(ctx).Err(); err != nil {
			return ports{}, contextOr(ctx, ErrRedisUnavailable)
		}
		credentialStore, err := credentials.NewMemoryStore(providerCredentialCapacity(configuration))
		if err != nil {
			return ports{}, err
		}
		adapter, err := redisactivity.New(redisClient, redisactivity.Options{
			Namespace:          namespace,
			Credentials:        credentialStore,
			OperationTTL:       redisOperationTTL,
			MaxCandidateScan:   candidateScan(configuration),
			MaxRuntimeCounters: credentialCapacity(configuration),
			MaxInventoryScan:   maxInventoryScan(configuration),
			CleanupLimit:       redisCleanupLimit,
		})
		if err != nil {
			return ports{}, err
		}
		opened.activity = adapter
		opened.readiness = redisReadiness{client: redisClient}
		opened.credentials = credentialStore
		if providersEnabled {
			stats, statsErr := controllerProvider.NewStatsRecorder(config.MaximumUpstreams)
			if statsErr != nil {
				return ports{}, statsErr
			}
			opened.providerResults = stats
			holderID, holderErr := resolveHolderID(infrastructure.holderID)
			if holderErr != nil {
				return ports{}, holderErr
			}
			opened.coordinator, err = redisprovider.New(redisClient, redisprovider.Options{
				Namespace: namespace, HolderID: holderID,
				LeaseTTL: providerLeaseTTL, RenewEvery: providerRenewEvery,
				RetryInterval: providerRetryInterval, PermitGrace: providerPermitGrace,
			})
			if err != nil {
				return ports{}, err
			}
		}
	}
	if configuration.Metrics.Enabled {
		opened.metricsReadiness = selectMetricsReadiness(
			configuration,
			storeReadiness{postgres: postgresPool, redis: redisClient},
			redisReadiness{client: redisClient},
		)
	}
	return opened, nil
}

func resolveRedisNamespace(configured string) (string, error) {
	if strings.TrimSpace(configured) != configured {
		return "", ErrInvalidOptions
	}
	if configured == "" {
		return redisNamespace, nil
	}
	return configured, nil
}

func selectMetricsReadiness(
	configuration *config.Config,
	admin, activity platformMetrics.ReadinessChecker,
) platformMetrics.ReadinessChecker {
	if configuration.Distribution.Enabled || hasEnabledUpstream(configuration) {
		return activity
	}
	if configuration.Admin.Enabled {
		return admin
	}
	return alwaysReady{}
}

type alwaysReady struct{}

func (alwaysReady) Ready(ctx context.Context) error {
	if ctx == nil {
		return ErrInvalidOptions
	}
	return ctx.Err()
}

type storeReadiness struct {
	postgres *pgxpool.Pool
	redis    *redis.Client
}

func (readiness storeReadiness) Ready(ctx context.Context) error {
	if ctx == nil {
		return ErrInvalidOptions
	}
	if readiness.postgres != nil {
		if err := readiness.postgres.Ping(ctx); err != nil {
			return contextOr(ctx, ErrPostgresUnavailable)
		}
	}
	if readiness.redis != nil {
		if err := readiness.redis.Ping(ctx).Err(); err != nil {
			return contextOr(ctx, ErrRedisUnavailable)
		}
	}
	return nil
}

type redisReadiness struct {
	client *redis.Client
}

func (readiness redisReadiness) Ready(ctx context.Context) error {
	if ctx == nil || readiness.client == nil {
		return ErrRedisUnavailable
	}
	if err := readiness.client.Ping(ctx).Err(); err != nil {
		return contextOr(ctx, ErrRedisUnavailable)
	}
	return nil
}

func credentialCapacity(configuration *config.Config) int {
	capacity := 0
	maximum := int(^uint(0) >> 1)
	for _, upstream := range configuration.Upstreams {
		if !upstream.Enabled || upstream.Pool.MaxSize <= 0 {
			continue
		}
		if capacity > maximum-upstream.Pool.MaxSize {
			return maximum
		}
		capacity += upstream.Pool.MaxSize
	}
	if capacity == 0 {
		return 1
	}
	return capacity
}

func providerCredentialCapacity(configuration *config.Config) int {
	capacity := 0
	maximum := int(^uint(0) >> 1)
	for _, upstream := range configuration.Upstreams {
		if upstream.Pool.MaxSize <= 0 {
			continue
		}
		maxInFlight := upstream.Fetch.MaxInFlight
		if maxInFlight <= 0 {
			maxInFlight = 1
		}
		if upstream.Pool.MaxSize > maximum/maxInFlight {
			return maximum
		}
		leases := upstream.Pool.MaxSize * maxInFlight
		if capacity > maximum-leases {
			return maximum
		}
		capacity += leases
	}
	if capacity == 0 {
		return 1
	}
	return capacity
}

func maxInventoryScan(_ *config.Config) int {
	return config.MaximumPoolSize
}

func hasEnabledUpstream(configuration *config.Config) bool {
	if configuration == nil {
		return false
	}
	for _, upstream := range configuration.Upstreams {
		if upstream.Enabled {
			return true
		}
	}
	return false
}

func resolveHolderID(configured string) (string, error) {
	if strings.TrimSpace(configured) != configured {
		return "", ErrInvalidOptions
	}
	if configured != "" {
		return configured, nil
	}
	var entropy [16]byte
	if _, err := rand.Read(entropy[:]); err != nil {
		return "", errors.Join(ErrStartup, err)
	}
	return "controller-" + hex.EncodeToString(entropy[:]), nil
}

func candidateScan(configuration *config.Config) int {
	configured := configuration.Distribution.Extraction
	if configured.MaxCountPerRequest > int(^uint(0)>>1)-configured.ReserveForGateway {
		return int(^uint(0) >> 1)
	}
	value := configured.MaxCountPerRequest + configured.ReserveForGateway
	if value < redisMinimumScan {
		return redisMinimumScan
	}
	return value
}

func contextOr(ctx context.Context, fallback error) error {
	if ctx != nil && ctx.Err() != nil {
		return ctx.Err()
	}
	return fallback
}