//go:build integration package redisactivity import ( "context" "errors" "fmt" "testing" "time" "proxy-pool/internal/domain/activitypool" proxyDomain "proxy-pool/internal/domain/proxy" "proxy-pool/internal/platform/credentials" ) func TestRedisUpsertEnforcesMaxSizeWithoutLeavingRejectedMappings(t *testing.T) { fixture := newRedisTestFixture(t) now := redisTestNow() first := testProxy("proxy-a", "192.0.2.10") second := testProxy("proxy-b", "192.0.2.11") result, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ ObservedAt: now, ConfiguredTTL: 30 * time.Second, MaxSize: 1, Proxies: []proxyDomain.Proxy{first, second}, }) if err != nil || result.Accepted != 2 || result.Inserted != 1 || result.Dropped != 1 { t.Fatalf("UpsertFetched() = %+v, %v", result, err) } retry, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ ObservedAt: now.Add(time.Second), ConfiguredTTL: 30 * time.Second, MaxSize: 2, Proxies: []proxyDomain.Proxy{second}, }) if err != nil || retry.Inserted != 1 || retry.Refreshed != 0 { t.Fatalf("UpsertFetched(capacity retry) = %+v, %v", retry, err) } } func TestRedisUpsertChunksLargeProviderResponsesWithGlobalCapacity(t *testing.T) { fixture := newRedisTestFixture(t) now := redisTestNow() proxies := make([]proxyDomain.Proxy, 300) for index := range proxies { proxies[index] = testProxy("proxy-"+fmt.Sprint(index), fmt.Sprintf("192.0.2.%d", index%250+1)) proxies[index].Port = uint16(10_000 + index) } result, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 275, Proxies: proxies, }) if err != nil || result.Accepted != 300 || result.Inserted != 275 || result.Dropped != 25 { t.Fatalf("UpsertFetched(large batch) = %+v, %v", result, err) } } func TestRedisUpsertPreservesIncumbentLifecycleAndRuntimeHealth(t *testing.T) { fixture := newRedisTestFixture(t) now := redisTestNow() candidate := testProxy("proxy-a", "192.0.2.10") first, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ ObservedAt: now, ConfiguredTTL: 30 * time.Second, AllocationSafetyMargin: 3 * time.Second, MaxSize: 10, Proxies: []proxyDomain.Proxy{candidate}, }) if err != nil || first.Inserted != 1 { t.Fatalf("first UpsertFetched() = %+v, %v", first, err) } if _, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ ProxyID: "proxy-a", CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateChecking, }); err != nil { t.Fatalf("ApplyHealth(checking): %v", err) } healthyAt := now.Add(2 * time.Second) if _, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ ProxyID: "proxy-a", CheckedAt: healthyAt, NextState: proxyDomain.StateAvailable, Latency: 25 * time.Millisecond, }); err != nil { t.Fatalf("ApplyHealth(available): %v", err) } refreshed, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ ObservedAt: now.Add(3 * time.Second), ConfiguredTTL: time.Minute, AllocationSafetyMargin: 5 * time.Second, MaxSize: 10, Proxies: []proxyDomain.Proxy{candidate}, }) if err != nil || refreshed.Refreshed != 1 || refreshed.Inserted != 0 { t.Fatalf("same-provider refresh = %+v, %v", refreshed, err) } entry, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ ProxyID: "proxy-a", CheckedAt: healthyAt, NextState: proxyDomain.StateAvailable, }) if err != nil || entry.State != proxyDomain.StateAvailable || entry.Proxy.Latency != 25*time.Millisecond || entry.Proxy.SourceUpstream != "provider-a" || entry.Proxy.ExpiresAt == nil || !entry.Proxy.ExpiresAt.Equal(now.Add(63*time.Second)) { t.Fatalf("refreshed entry = %+v, %v", entry, err) } duplicate, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-b", activitypool.FetchedBatch{ ObservedAt: now.Add(4 * time.Second), ConfiguredTTL: 5 * time.Minute, MaxSize: 10, Proxies: []proxyDomain.Proxy{candidate}, }) if err != nil || duplicate.Refreshed != 1 || duplicate.Inserted != 0 { t.Fatalf("cross-provider duplicate = %+v, %v", duplicate, err) } entry, err = fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ ProxyID: "proxy-a", CheckedAt: healthyAt, NextState: proxyDomain.StateAvailable, }) if err != nil || entry.Proxy.SourceUpstream != "provider-a" || entry.Proxy.ExpiresAt == nil || !entry.Proxy.ExpiresAt.Equal(now.Add(63*time.Second)) { t.Fatalf("incumbent entry = %+v, %v", entry, err) } replacementAt := now.Add(64 * time.Second) replaced, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-b", activitypool.FetchedBatch{ ObservedAt: replacementAt, ConfiguredTTL: time.Minute, MaxSize: 10, Proxies: []proxyDomain.Proxy{candidate}, }) if err != nil || replaced.Inserted != 1 || replaced.Refreshed != 0 { t.Fatalf("expired incumbent replacement = %+v, %v", replaced, err) } entry, err = fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ ProxyID: "proxy-a", CheckedAt: replacementAt.Add(time.Second), NextState: proxyDomain.StateChecking, }) if err != nil || entry.Proxy.SourceUpstream != "provider-b" { t.Fatalf("replacement entry = %+v, %v", entry, err) } } func TestRedisUpsertResolvesCredentialsBeforeCommit(t *testing.T) { fixture := newRedisTestFixture(t) now := redisTestNow() candidate := testProxy("proxy-a", "192.0.2.10") candidate.Username = "user" candidate.SecretRef = "cred_missing" candidate.CredentialVersion = "v1" _, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 10, Proxies: []proxyDomain.Proxy{candidate}, }) if !errors.Is(err, credentials.ErrCredentialMissing) { t.Fatalf("UpsertFetched(missing credential) error = %v", err) } candidate.SecretRef = "" candidate.CredentialVersion = "" retry, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 10, Proxies: []proxyDomain.Proxy{candidate}, }) if err != nil || retry.Inserted != 1 || retry.Refreshed != 0 { t.Fatalf("UpsertFetched(after resolution failure) = %+v, %v", retry, err) } reference, err := fixture.Credentials.Put(context.Background(), "proxy-b", credentials.Value{ Username: "user", Password: "password", }) if err != nil { t.Fatalf("Credentials.Put(): %v", err) } withCredential := testProxy("proxy-b", "192.0.2.11") withCredential.Username = "user" withCredential.SecretRef = reference.SecretRef withCredential.CredentialVersion = reference.CredentialVersion stored, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 10, Proxies: []proxyDomain.Proxy{withCredential}, }) if err != nil || stored.Inserted != 1 { t.Fatalf("UpsertFetched(resolved credential) = %+v, %v", stored, err) } } func TestRedisHealthTransitionsAreMonotonicAndIdempotent(t *testing.T) { fixture := newRedisTestFixture(t) now := redisTestNow() if _, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 10, Proxies: []proxyDomain.Proxy{testProxy("proxy-a", "192.0.2.10")}, }); err != nil { t.Fatalf("UpsertFetched(): %v", err) } if _, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ ProxyID: "proxy-a", CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateChecking, }); err != nil { t.Fatalf("ApplyHealth(checking): %v", err) } checkedAt := now.Add(2 * time.Second) available, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ ProxyID: "proxy-a", CheckedAt: checkedAt, NextState: proxyDomain.StateAvailable, Latency: 30 * time.Millisecond, }) if err != nil || available.State != proxyDomain.StateAvailable || available.Proxy.LastSuccessAt == nil || !available.Proxy.LastSuccessAt.Equal(checkedAt) { t.Fatalf("ApplyHealth(available) = %+v, %v", available, err) } replayed, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ ProxyID: "proxy-a", CheckedAt: checkedAt, NextState: proxyDomain.StateAvailable, Latency: time.Second, }) if err != nil || replayed.Proxy.Latency != 30*time.Millisecond { t.Fatalf("ApplyHealth(replay) = %+v, %v", replayed, err) } _, err = fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ ProxyID: "proxy-a", CheckedAt: checkedAt, NextState: proxyDomain.StateSuspect, }) if !errors.Is(err, activitypool.ErrStaleHealthUpdate) { t.Fatalf("ApplyHealth(conflicting replay) error = %v", err) } _, err = fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ ProxyID: "proxy-a", CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateSuspect, }) if !errors.Is(err, activitypool.ErrStaleHealthUpdate) { t.Fatalf("ApplyHealth(stale) error = %v", err) } } func redisTestNow() time.Time { return time.Now().UTC().Add(time.Hour).Truncate(time.Millisecond) } func testProxy(id, host string) proxyDomain.Proxy { return proxyDomain.Proxy{ ID: id, Scheme: proxyDomain.SchemeHTTP, Host: host, Port: 8080, State: proxyDomain.StateFetched, Tags: map[string]string{"region": "cn", "carrier": "ct"}, } }