proxy-pool/internal/adapters/redisactivity/extract_integration_test.go

//go:build integration

package redisactivity

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/url"
	"sync"
	"testing"
	"time"

	"proxy-pool/internal/domain/activitypool"
	extractionDomain "proxy-pool/internal/domain/extraction"
	proxyDomain "proxy-pool/internal/domain/proxy"
	"proxy-pool/internal/platform/credentials"
)

func TestRedisExtractAppliesEveryFilterAndBuildsCredentialURL(t *testing.T) {
	fixture := newRedisTestFixture(t)
	now := redisTestNow()
	reference, err := fixture.Credentials.Put(context.Background(), "target", credentials.Value{
		Username: "user@tenant", Password: "p:/@ss",
	})
	if err != nil {
		t.Fatalf("Credentials.Put(): %v", err)
	}
	target := testProxy("target", "192.0.2.10")
	target.Username = "user@tenant"
	target.SecretRef = reference.SecretRef
	target.CredentialVersion = reference.CredentialVersion
	target.Tags = map[string]string{"region": "cn", "carrier": "ct"}
	seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(2*time.Second), time.Minute, target)

	other := testProxy("other", "192.0.2.11")
	other.Scheme = proxyDomain.SchemeSOCKS5
	other.Tags = map[string]string{"region": "us", "carrier": "cu"}
	seedRedisAvailable(t, fixture.Adapter, "provider-b", now, now.Add(2*time.Second), time.Minute, other)

	result, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{
		RequestID: "req-filter", ClientID: "client-a", Requested: 2,
		Fulfillment: extractionDomain.Partial, Now: now.Add(3 * time.Second),
		Protocols: []string{"http"}, Regions: []string{"cn"}, Carriers: []string{"ct"},
		Upstreams: []string{"provider-a"},
	})
	if err != nil || result.Returned != 1 || result.Items[0].ID != "target" {
		t.Fatalf("Extract(filtered) = %+v, %v", result, err)
	}
	parsed, err := url.Parse(result.Items[0].URL)
	if err != nil {
		t.Fatalf("parse extracted URL: %v", err)
	}
	password, hasPassword := parsed.User.Password()
	if parsed.User.Username() != "user@tenant" || !hasPassword || password != "p:/@ss" {
		t.Fatalf("URL credentials = (%q, %q, %t)", parsed.User.Username(), password, hasPassword)
	}
}

func TestRedisExtractSupportsPartialAndAllOrNothing(t *testing.T) {
	fixture := newRedisTestFixture(t)
	now := redisTestNow()
	candidate := testProxy("proxy-a", "192.0.2.10")
	seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(2*time.Second), time.Minute, candidate)

	_, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{
		RequestID: "req-all", ClientID: "client-a", Requested: 2,
		Fulfillment: extractionDomain.AllOrNothing, Now: now.Add(3 * time.Second),
	})
	if !errors.Is(err, extractionDomain.ErrInsufficientProxies) {
		t.Fatalf("Extract(allOrNothing) error = %v", err)
	}
	result, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{
		RequestID: "req-partial", ClientID: "client-a", Requested: 2,
		Fulfillment: extractionDomain.Partial, Now: now.Add(3 * time.Second),
	})
	if err != nil || result.Returned != 1 || result.Items[0].ID != "proxy-a" {
		t.Fatalf("Extract(partial) = %+v, %v", result, err)
	}
}

func TestRedisExtractReturnsEmptyItemsForValidZeroResults(t *testing.T) {
	t.Run("gateway reserve", func(t *testing.T) {
		fixture := newRedisTestFixture(t)
		now := redisTestNow()
		seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(time.Second), time.Minute,
			testProxy("proxy-a", "192.0.2.10"))

		result, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{
			RequestID: "req-empty-reserve", ClientID: "client-a", Requested: 1,
			Fulfillment: extractionDomain.Partial, Now: now.Add(2 * time.Second), ReserveForGateway: 1,
		})
		if err != nil || result.Returned != 0 || len(result.Items) != 0 {
			t.Fatalf("Extract(reserved empty) = %+v, %v", result, err)
		}
	})

	t.Run("zero requested with idempotency", func(t *testing.T) {
		fixture := newRedisTestFixture(t)
		now := redisTestNow()
		command := extractionDomain.Command{
			RequestID: "req-empty-zero", ClientID: "client-a", IdempotencyKey: "idem-empty-zero",
			Requested: 0, Fulfillment: extractionDomain.Partial, Now: now,
		}
		result, err := fixture.Adapter.Extract(context.Background(), command)
		if err != nil || result.Requested != 0 || result.Returned != 0 || len(result.Items) != 0 {
			t.Fatalf("Extract(zero requested) = %+v, %v", result, err)
		}
		command.RequestID = "req-empty-zero-replay"
		command.Now = now.Add(time.Second)
		replayed, err := fixture.Adapter.Extract(context.Background(), command)
		if err != nil || replayed.Requested != 0 || replayed.Returned != 0 || len(replayed.Items) != 0 {
			t.Fatalf("Extract(zero requested replay) = %+v, %v", replayed, err)
		}
		command.RequestID = "req-empty-zero-conflict"
		command.Requested = 1
		if _, err := fixture.Adapter.Extract(context.Background(), command); !errors.Is(err, extractionDomain.ErrIdempotencyConflict) {
			t.Fatalf("Extract(zero requested conflict) error = %v", err)
		}
	})
}

func TestRedisExtractAppliesTTLHealthAgeAndGatewayReserve(t *testing.T) {
	fixture := newRedisTestFixture(t)
	now := redisTestNow()
	observedAt := now.Add(-30 * time.Second)

	short := testProxy("short", "192.0.2.10")
	seedRedisAvailable(t, fixture.Adapter, "provider-a", observedAt, now.Add(-time.Second), 35*time.Second, short)
	stale := testProxy("stale", "192.0.2.11")
	seedRedisAvailable(t, fixture.Adapter, "provider-a", observedAt, now.Add(-10*time.Second), 2*time.Minute, stale)
	fresh := testProxy("fresh", "192.0.2.12")
	seedRedisAvailable(t, fixture.Adapter, "provider-a", observedAt, now.Add(-time.Second), 2*time.Minute, fresh)

	result, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{
		RequestID: "req-policy", ClientID: "client-a", Requested: 2,
		Fulfillment: extractionDomain.Partial, Now: now,
		MinRemainingTTL: 30 * time.Second, MaxHealthCheckAge: 5 * time.Second,
	})
	if err != nil || result.Returned != 1 || result.Items[0].ID != "fresh" {
		t.Fatalf("Extract(policy) = %+v, %v", result, err)
	}

	reserveFixture := newRedisTestFixture(t)
	for index := range 3 {
		candidate := testProxy(fmt.Sprintf("reserve-%d", index), fmt.Sprintf("192.0.2.%d", index+20))
		seedRedisAvailable(t, reserveFixture.Adapter, "provider-a", now, now.Add(time.Second), time.Minute, candidate)
	}
	reserved, err := reserveFixture.Adapter.Extract(context.Background(), extractionDomain.Command{
		RequestID: "req-reserve", ClientID: "client-a", Requested: 2,
		Fulfillment: extractionDomain.Partial, Now: now.Add(2 * time.Second), ReserveForGateway: 2,
	})
	if err != nil || reserved.Returned != 1 {
		t.Fatalf("Extract(reserve) = %+v, %v", reserved, err)
	}
}

func TestRedisExtractIdempotencyReplayConflictAndExpiryBound(t *testing.T) {
	fixture := newRedisTestFixture(t)
	now := redisTestNow()
	seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(time.Second), time.Minute,
		testProxy("proxy-a", "192.0.2.10"))
	command := extractionDomain.Command{
		RequestID: "req-first", ClientID: "client-a", IdempotencyKey: "idem-12345678",
		IdempotencyTTL: 5 * time.Minute, Requested: 1,
		Fulfillment: extractionDomain.Partial, Now: now.Add(2 * time.Second),
	}
	first, err := fixture.Adapter.Extract(context.Background(), command)
	if err != nil || first.Returned != 1 {
		t.Fatalf("first Extract() = %+v, %v", first, err)
	}
	command.RequestID = "req-replay"
	command.Now = now.Add(10 * time.Second)
	replayed, err := fixture.Adapter.Extract(context.Background(), command)
	if err != nil || replayed.Returned != 1 || !replayed.ExtractedAt.Equal(first.ExtractedAt) {
		t.Fatalf("replayed Extract() = %+v, %v", replayed, err)
	}
	command.RequestID = "req-conflict"
	command.Requested = 2
	if _, err := fixture.Adapter.Extract(context.Background(), command); !errors.Is(err, extractionDomain.ErrIdempotencyConflict) {
		t.Fatalf("conflicting Extract() error = %v", err)
	}

	expiryFixture := newRedisTestFixture(t)
	realNow := time.Now().UTC().Truncate(time.Millisecond)
	shortLived := testProxy("short-lived", "192.0.2.30")
	shortLived.State = proxyDomain.StateAvailable
	if _, err := expiryFixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{
		ObservedAt: realNow, ConfiguredTTL: 2 * time.Second, MaxSize: 10,
		Proxies: []proxyDomain.Proxy{shortLived},
	}); err != nil {
		t.Fatalf("UpsertFetched(short-lived): %v", err)
	}
	expiryCommand := extractionDomain.Command{
		RequestID: "req-expiry-first", ClientID: "client-a", IdempotencyKey: "idem-expiry",
		IdempotencyTTL: time.Minute, Requested: 1,
		Fulfillment: extractionDomain.Partial, Now: realNow.Add(time.Millisecond),
	}
	original, err := expiryFixture.Adapter.Extract(context.Background(), expiryCommand)
	if err != nil || original.Returned != 1 {
		t.Fatalf("Extract(short-lived) = %+v, %v", original, err)
	}
	time.Sleep(2200 * time.Millisecond)
	replacementAt := time.Now().UTC().Truncate(time.Millisecond)
	if _, err := expiryFixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{
		ObservedAt: replacementAt, ConfiguredTTL: time.Minute, MaxSize: 10,
		Proxies: []proxyDomain.Proxy{shortLived},
	}); err != nil {
		t.Fatalf("UpsertFetched(replacement): %v", err)
	}
	expiryCommand.RequestID = "req-expiry-second"
	expiryCommand.Now = replacementAt.Add(time.Millisecond)
	again, err := expiryFixture.Adapter.Extract(context.Background(), expiryCommand)
	if err != nil || again.Returned != 1 || again.ExtractedAt.Equal(original.ExtractedAt) {
		t.Fatalf("Extract(after idempotency expiry) = %+v, %v", again, err)
	}
}

func TestRedisExtractIdempotencyDigestCanonicalizesBusinessFilters(t *testing.T) {
	fixture := newRedisTestFixture(t)
	now := redisTestNow()
	candidate := testProxy("proxy-a", "192.0.2.10")
	candidate.Tags = map[string]string{"region": "cn", "carrier": "ct"}
	seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(time.Second), time.Minute, candidate)

	command := extractionDomain.Command{
		RequestID: "req-canonical-first", ClientID: "client-a", SourceIP: "192.0.2.100",
		IdempotencyKey: "idem-canonical", IdempotencyTTL: time.Minute,
		Requested: 1, Fulfillment: extractionDomain.Partial, Now: now.Add(2 * time.Second),
		MinRemainingTTL: time.Second, MaxHealthCheckAge: time.Minute,
		Protocols: []string{"http", "http"}, Regions: []string{"cn", "cn"},
		Carriers: []string{"ct", "ct"}, Upstreams: []string{"provider-a", "provider-a"},
	}
	first, err := fixture.Adapter.Extract(context.Background(), command)
	if err != nil || first.Returned != 1 {
		t.Fatalf("first Extract() = %+v, %v", first, err)
	}

	command.RequestID = "req-canonical-replay"
	command.SourceIP = "198.51.100.200"
	command.Now = now.Add(10 * time.Second)
	command.IdempotencyTTL = 2 * time.Minute
	command.MinRemainingTTL = 2 * time.Second
	command.MaxHealthCheckAge = 2 * time.Minute
	command.Protocols = []string{"http"}
	command.Regions = []string{"cn"}
	command.Carriers = []string{"ct"}
	command.Upstreams = []string{"provider-a"}
	replayed, err := fixture.Adapter.Extract(context.Background(), command)
	if err != nil || replayed.Returned != 1 || !replayed.ExtractedAt.Equal(first.ExtractedAt) ||
		replayed.Items[0].ID != first.Items[0].ID {
		t.Fatalf("replayed Extract() = %+v, %v", replayed, err)
	}
}

func TestRedisExtractReplaysTheCommittedRequestOperation(t *testing.T) {
	fixture := newRedisTestFixture(t)
	now := redisTestNow()
	seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(time.Second), time.Minute,
		testProxy("proxy-a", "192.0.2.10"))
	command := extractionDomain.Command{
		RequestID: "req-operation-replay", ClientID: "client-a", Requested: 1,
		Fulfillment: extractionDomain.Partial, Now: now.Add(2 * time.Second),
	}

	first, err := fixture.Adapter.Extract(context.Background(), command)
	if err != nil || first.Returned != 1 {
		t.Fatalf("first Extract() = %+v, %v", first, err)
	}
	command.Now = now.Add(3 * time.Second)
	replayed, err := fixture.Adapter.Extract(context.Background(), command)
	if err != nil || replayed.Returned != 1 || !replayed.ExtractedAt.Equal(first.ExtractedAt) ||
		replayed.Items[0].ID != first.Items[0].ID {
		t.Fatalf("replayed Extract() = %+v, %v", replayed, err)
	}
}

func TestRedisExtractIsExclusiveUnderConcurrentRaces(t *testing.T) {
	fixture := newRedisTestFixture(t)
	now := redisTestNow()
	for iteration := range 100 {
		candidate := testProxy(fmt.Sprintf("race-%d", iteration), fmt.Sprintf("198.51.100.%d", iteration+1))
		candidate.State = proxyDomain.StateAvailable
		if _, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{
			ObservedAt: now, ConfiguredTTL: 10 * time.Minute, MaxSize: 10,
			Proxies: []proxyDomain.Proxy{candidate},
		}); err != nil {
			t.Fatalf("iteration %d UpsertFetched(): %v", iteration, err)
		}

		results := make(chan extractionDomain.Result, 2)
		errorsCh := make(chan error, 2)
		var workers sync.WaitGroup
		for worker := range 2 {
			workers.Add(1)
			go func(worker int) {
				defer workers.Done()
				result, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{
					RequestID: fmt.Sprintf("req-race-%d-%d", iteration, worker),
					ClientID:  fmt.Sprintf("client-%d", worker), Requested: 1,
					Fulfillment: extractionDomain.Partial, Now: now.Add(time.Second),
				})
				results <- result
				errorsCh <- err
			}(worker)
		}
		workers.Wait()
		close(results)
		close(errorsCh)
		for err := range errorsCh {
			if err != nil {
				t.Fatalf("iteration %d Extract(): %v", iteration, err)
			}
		}
		returned := 0
		for result := range results {
			returned += result.Returned
		}
		if returned != 1 {
			t.Fatalf("iteration %d total returned = %d, want 1", iteration, returned)
		}
	}
}

func TestRedisExtractScanBudgetExhaustionDoesNotMutateCandidates(t *testing.T) {
	fixture := newRedisTestFixture(t)
	now := redisTestNow()
	for index, ttl := range []time.Duration{3 * time.Minute, 2 * time.Minute, time.Minute} {
		candidate := testProxy(fmt.Sprintf("scan-%d", index), fmt.Sprintf("203.0.113.%d", index+1))
		candidate.State = proxyDomain.StateAvailable
		candidate.Tags["region"] = "none"
		if index == 2 {
			candidate.Tags["region"] = "target"
		}
		if _, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{
			ObservedAt: now, ConfiguredTTL: ttl, MaxSize: 10, Proxies: []proxyDomain.Proxy{candidate},
		}); err != nil {
			t.Fatalf("UpsertFetched(scan-%d): %v", index, err)
		}
	}
	bounded, err := New(fixture.Client, Options{
		Namespace: fixture.Namespace, Credentials: fixture.Credentials,
		OperationTTL: time.Minute, MaxCandidateScan: 2, CleanupLimit: 16,
	})
	if err != nil {
		t.Fatalf("New(bounded): %v", err)
	}
	_, err = bounded.Extract(context.Background(), extractionDomain.Command{
		RequestID: "req-scan", ClientID: "client-a", Requested: 1,
		Fulfillment: extractionDomain.AllOrNothing, Now: now.Add(time.Second),
		Regions: []string{"target", "other"},
	})
	if !errors.Is(err, extractionDomain.ErrStoreUnavailable) {
		t.Fatalf("Extract(scan exhausted) error = %v", err)
	}

	remaining, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{
		RequestID: "req-after-scan", ClientID: "client-a", Requested: 3,
		Fulfillment: extractionDomain.AllOrNothing, Now: now.Add(time.Second),
	})
	if err != nil || remaining.Returned != 3 {
		t.Fatalf("Extract(after scan exhaustion) = %+v, %v", remaining, err)
	}
}

func TestRedisExtractDoesNotConsumeIncompleteRecords(t *testing.T) {
	tests := []struct {
		name   string
		mutate func(map[string]any)
	}{
		{name: "empty host", mutate: func(record map[string]any) { record["host"] = "" }},
		{name: "missing creation time", mutate: func(record map[string]any) { record["createdAtMs"] = float64(0) }},
		{name: "negative latency", mutate: func(record map[string]any) { record["latencyNs"] = float64(-1) }},
		{name: "negative concurrency", mutate: func(record map[string]any) { record["maxConcurrency"] = float64(-1) }},
		{name: "invalid state", mutate: func(record map[string]any) { record["state"] = "BROKEN" }},
		{name: "invalid credential version", mutate: func(record map[string]any) { record["credentialVersion"] = float64(1) }},
		{name: "invalid last success", mutate: func(record map[string]any) { record["lastSuccessAtMs"] = float64(-1) }},
		{name: "foreign index key", mutate: func(record map[string]any) { record["indexKeys"] = []any{"pp:{other}:available"} }},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			fixture := newRedisTestFixture(t)
			now := redisTestNow()
			seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(time.Second), time.Minute,
				testProxy("corrupt", "192.0.2.10"))

			raw, err := fixture.Client.HGet(context.Background(), fixture.Adapter.keys.records, "corrupt").Result()
			if err != nil {
				t.Fatalf("HGet(corrupt): %v", err)
			}
			var stored map[string]any
			if err := json.Unmarshal([]byte(raw), &stored); err != nil {
				t.Fatalf("decode stored record: %v", err)
			}
			test.mutate(stored)
			expectedState := stored["state"]
			corrupted, err := json.Marshal(stored)
			if err != nil {
				t.Fatalf("encode corrupt record: %v", err)
			}
			if err := fixture.Client.HSet(context.Background(), fixture.Adapter.keys.records, "corrupt", corrupted).Err(); err != nil {
				t.Fatalf("HSet(corrupt): %v", err)
			}

			result, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{
				RequestID: "req-corrupt", ClientID: "client-a", Requested: 1,
				Fulfillment: extractionDomain.Partial, Now: now.Add(2 * time.Second),
			})
			if err != nil || result.Returned != 0 || len(result.Items) != 0 {
				t.Fatalf("Extract(corrupt) = %+v, %v", result, err)
			}

			after, err := fixture.Client.HGet(context.Background(), fixture.Adapter.keys.records, "corrupt").Result()
			if err != nil {
				t.Fatalf("HGet(corrupt after extraction): %v", err)
			}
			if err := json.Unmarshal([]byte(after), &stored); err != nil {
				t.Fatalf("decode record after extraction: %v", err)
			}
			if stored["state"] != expectedState {
				t.Fatalf("corrupt record state = %q, want unchanged %q", stored["state"], expectedState)
			}
		})
	}
}

func seedRedisAvailable(
	t *testing.T,
	adapter *Adapter,
	upstreamID string,
	observedAt time.Time,
	availableAt time.Time,
	ttl time.Duration,
	candidate proxyDomain.Proxy,
) {
	t.Helper()
	candidate.State = proxyDomain.StateFetched
	if _, err := adapter.UpsertFetched(context.Background(), upstreamID, activitypool.FetchedBatch{
		ObservedAt: observedAt, ConfiguredTTL: ttl, MaxSize: 100,
		Proxies: []proxyDomain.Proxy{candidate},
	}); err != nil {
		t.Fatalf("UpsertFetched(%s): %v", candidate.ID, err)
	}
	if _, err := adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{
		ProxyID: candidate.ID, CheckedAt: availableAt.Add(-time.Millisecond), NextState: proxyDomain.StateChecking,
	}); err != nil {
		t.Fatalf("ApplyHealth(%s, checking): %v", candidate.ID, err)
	}
	if _, err := adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{
		ProxyID: candidate.ID, CheckedAt: availableAt, NextState: proxyDomain.StateAvailable,
	}); err != nil {
		t.Fatalf("ApplyHealth(%s, available): %v", candidate.ID, err)
	}
}