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

package bootstrap

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync/atomic"
	"testing"

	"github.com/redis/go-redis/v9"

	"proxy-pool/internal/config"
	"proxy-pool/internal/platform/admission"
)

type recordingAdmissionScripter struct {
	redis.Scripter
	keys []string
	args []any
}

func (client *recordingAdmissionScripter) EvalSha(
	_ context.Context,
	_ string,
	keys []string,
	args ...any,
) *redis.Cmd {
	client.keys = append([]string(nil), keys...)
	client.args = append([]any(nil), args...)
	return redis.NewCmdResult("ok", nil)
}

func TestProductionInfrastructureRejectsInvalidStorageWithoutLeakingURLs(t *testing.T) {
	t.Parallel()
	postgresSecret := "postgres-secret"
	_, err := (&productionInfrastructure{}).Open(context.Background(), &config.Config{
		Admin:   config.Listener{Enabled: true},
		Storage: config.Storage{PostgresURL: "postgres://user:" + postgresSecret + "@%zz"},
	})
	if !errors.Is(err, ErrPostgresConfiguration) || strings.Contains(err.Error(), postgresSecret) {
		t.Fatalf("Open(invalid PostgreSQL) error = %v", err)
	}

	redisSecret := "redis-secret"
	_, err = (&productionInfrastructure{}).Open(context.Background(), &config.Config{
		Distribution: config.Distribution{Listener: config.Listener{Enabled: true}},
		Storage:      config.Storage{RedisURL: "redis://user:" + redisSecret + "@%zz"},
	})
	if !errors.Is(err, ErrRedisConfiguration) || strings.Contains(err.Error(), redisSecret) {
		t.Fatalf("Open(invalid Redis) error = %v", err)
	}
}

func TestNewDistributionAdmitterPassesConfiguredLimits(t *testing.T) {
	t.Parallel()
	client := &recordingAdmissionScripter{}
	limiter, err := newDistributionAdmitter(client, "controller-a", config.Limits{
		RequestsPerMinute:          321,
		RequestsPerMinutePerClient: 17,
	})
	if err != nil {
		t.Fatalf("newDistributionAdmitter(): %v", err)
	}
	if err := limiter.Admit(context.Background(), "client-a"); err != nil {
		t.Fatalf("Admit(): %v", err)
	}
	if len(client.keys) != 1 || len(client.args) != 4 {
		t.Fatalf("Redis admission call = keys:%v args:%v", client.keys, client.args)
	}
	if got := fmt.Sprintf("%s|%v|%v|%v", client.keys[0], client.args[0], client.args[1], client.args[2]); got != "pp:{admission}:controller-a:window|60000|321|17" {
		t.Fatalf("Redis admission inputs = %q", got)
	}
}

func TestNewDistributionAdmitterAllowsAllWhenQuotasAreDisabled(t *testing.T) {
	t.Parallel()
	limiter, err := newDistributionAdmitter(nil, "controller-a", config.Limits{})
	if err != nil {
		t.Fatalf("newDistributionAdmitter(): %v", err)
	}
	if _, ok := limiter.(admission.AllowAll); !ok {
		t.Fatalf("limiter type = %T, want admission.AllowAll", limiter)
	}
	if err := limiter.Admit(context.Background(), "client-a"); err != nil {
		t.Fatalf("AllowAll.Admit(): %v", err)
	}
}

func TestSelectMetricsReadinessPreservesDistributionWhenAdminStoreFails(t *testing.T) {
	t.Parallel()
	adminCalls := &atomic.Int64{}
	activityCalls := &atomic.Int64{}
	adminReady := readinessFunc(func(context.Context) error {
		adminCalls.Add(1)
		return ErrPostgresUnavailable
	})
	activityReady := readinessFunc(func(context.Context) error {
		activityCalls.Add(1)
		return nil
	})
	selected := selectMetricsReadiness(&config.Config{
		Admin:        config.Listener{Enabled: true},
		Distribution: config.Distribution{Listener: config.Listener{Enabled: true}},
	}, adminReady, activityReady)
	if err := selected.Ready(context.Background()); err != nil {
		t.Fatalf("Ready() error = %v", err)
	}
	if adminCalls.Load() != 0 || activityCalls.Load() != 1 {
		t.Fatalf("readiness calls = admin:%d activity:%d", adminCalls.Load(), activityCalls.Load())
	}
}

func TestSelectMetricsReadinessUsesAdminStoresWithoutDistribution(t *testing.T) {
	t.Parallel()
	wantErr := errors.New("admin unavailable")
	selected := selectMetricsReadiness(
		&config.Config{Admin: config.Listener{Enabled: true}},
		readinessFunc(func(context.Context) error { return wantErr }),
		readinessFunc(func(context.Context) error { return nil }),
	)
	if err := selected.Ready(context.Background()); !errors.Is(err, wantErr) {
		t.Fatalf("Ready() error = %v, want %v", err, wantErr)
	}
}

type readinessFunc func(context.Context) error

func (function readinessFunc) Ready(ctx context.Context) error { return function(ctx) }

func TestProductionRedisSizingUsesConfigurationBounds(t *testing.T) {
	t.Parallel()
	configuration := &config.Config{
		Distribution: config.Distribution{Extraction: config.Extraction{
			MaxCountPerRequest: 100, ReserveForGateway: 5_000,
		}},
		Upstreams: map[string]config.Upstream{
			"provider-a": {Enabled: true, Pool: config.Pool{MaxSize: 3_000}, Fetch: config.Fetch{MaxInFlight: 2}},
			"provider-b": {Enabled: true, Pool: config.Pool{MaxSize: 2_000}, Fetch: config.Fetch{MaxInFlight: 1}},
			"disabled":   {Pool: config.Pool{MaxSize: 50_000}, Fetch: config.Fetch{MaxInFlight: 3}},
		},
	}
	if got := credentialCapacity(configuration); got != 5_000 {
		t.Fatalf("credentialCapacity() = %d, want 5000", got)
	}
	if got := providerCredentialCapacity(configuration); got != 158_000 {
		t.Fatalf("providerCredentialCapacity() = %d, want 158000", got)
	}
	if got := candidateScan(configuration); got != 5_100 {
		t.Fatalf("candidateScan() = %d, want 5100", got)
	}
	configuration.Distribution.Extraction = config.Extraction{MaxCountPerRequest: 1}
	if got := candidateScan(configuration); got != redisMinimumScan {
		t.Fatalf("candidateScan(minimum) = %d, want %d", got, redisMinimumScan)
	}
}

func TestMaxInventoryScanSupportsPoolGrowthAfterReload(t *testing.T) {
	t.Parallel()
	configuration := &config.Config{Upstreams: map[string]config.Upstream{
		"provider-a": {Enabled: true, Pool: config.Pool{MaxSize: 100}},
	}}

	if got := maxInventoryScan(configuration); got != config.MaximumPoolSize {
		t.Fatalf("maxInventoryScan(initial small pool) = %d, want %d", got, config.MaximumPoolSize)
	}
}