From 9f3a51a7c0846e7a3a50fc9eddb584b781cc1edc Mon Sep 17 00:00:00 2001 From: youfak Date: Wed, 29 Jul 2026 15:25:18 +0800 Subject: [PATCH] feat: add redis activity adapter foundation --- deploy/docker-compose.test.yml | 14 + go.mod | 10 +- go.sum | 22 ++ internal/adapters/redisactivity/adapter.go | 70 ++++ .../adapters/redisactivity/adapter_test.go | 232 ++++++++++++ internal/adapters/redisactivity/codec.go | 350 ++++++++++++++++++ internal/adapters/redisactivity/keys.go | 80 ++++ internal/adapters/redisactivity/scripts.go | 64 ++++ .../adapters/redisactivity/testredis_test.go | 99 +++++ scripts/test-redis.ps1 | 33 ++ 10 files changed, 973 insertions(+), 1 deletion(-) create mode 100644 deploy/docker-compose.test.yml create mode 100644 internal/adapters/redisactivity/adapter.go create mode 100644 internal/adapters/redisactivity/adapter_test.go create mode 100644 internal/adapters/redisactivity/codec.go create mode 100644 internal/adapters/redisactivity/keys.go create mode 100644 internal/adapters/redisactivity/scripts.go create mode 100644 internal/adapters/redisactivity/testredis_test.go create mode 100644 scripts/test-redis.ps1 diff --git a/deploy/docker-compose.test.yml b/deploy/docker-compose.test.yml new file mode 100644 index 0000000..838943b --- /dev/null +++ b/deploy/docker-compose.test.yml @@ -0,0 +1,14 @@ +name: proxy-pool-test + +services: + redis: + image: redis:8.2-alpine + command: ["redis-server", "--appendonly", "no", "--save", ""] + ports: + - "127.0.0.1:16379:6379" + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 1s + timeout: 1s + retries: 30 + start_period: 1s diff --git a/go.mod b/go.mod index 766da3e..64e67ae 100644 --- a/go.mod +++ b/go.mod @@ -2,4 +2,12 @@ module proxy-pool go 1.26.0 -require go.yaml.in/yaml/v4 v4.0.0-rc.3 +require ( + github.com/redis/go-redis/v9 v9.19.0 + go.yaml.in/yaml/v4 v4.0.0-rc.3 +) + +require ( + github.com/cespare/xxhash/v2 v2.3.0 // indirect + go.uber.org/atomic v1.11.0 // indirect +) diff --git a/go.sum b/go.sum index 1b5b7f5..9b3c79c 100644 --- a/go.sum +++ b/go.sum @@ -1,2 +1,24 @@ +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= +github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/redis/go-redis/v9 v9.19.0 h1:XPVaaPSnG6RhYf7p+rmSa9zZfeVAnWsH5h3lxthOm/k= +github.com/redis/go-redis/v9 v9.19.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= +github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.yaml.in/yaml/v4 v4.0.0-rc.3 h1:3h1fjsh1CTAPjW7q/EMe+C8shx5d8ctzZTrLcs/j8Go= go.yaml.in/yaml/v4 v4.0.0-rc.3/go.mod h1:aZqd9kCMsGL7AuUv/m/PvWLdg5sjJsZ4oHDEnfPPfY0= +golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc= +golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= diff --git a/internal/adapters/redisactivity/adapter.go b/internal/adapters/redisactivity/adapter.go new file mode 100644 index 0000000..48339ce --- /dev/null +++ b/internal/adapters/redisactivity/adapter.go @@ -0,0 +1,70 @@ +package redisactivity + +import ( + "errors" + "fmt" + "reflect" + "regexp" + "strings" + "time" + + "github.com/redis/go-redis/v9" + + "proxy-pool/internal/platform/credentials" +) + +var ( + ErrInvalidOptions = errors.New("invalid redis activity adapter options") + namespacePattern = regexp.MustCompile(`^[A-Za-z0-9._-]+$`) +) + +type Options struct { + Namespace string + Credentials credentials.Store + OperationTTL time.Duration + MaxCandidateScan int + CleanupLimit int +} + +type Adapter struct { + client redis.Scripter + credentials credentials.Store + keys keyspace + options Options +} + +func New(client redis.Scripter, options Options) (*Adapter, error) { + options.Namespace = strings.TrimSpace(options.Namespace) + if nilInterface(client) || nilInterface(options.Credentials) || + !namespacePattern.MatchString(options.Namespace) || options.OperationTTL <= 0 || + options.MaxCandidateScan <= 0 || options.CleanupLimit <= 0 { + return nil, ErrInvalidOptions + } + return &Adapter{ + client: client, + credentials: options.Credentials, + keys: newKeyspace(options.Namespace), + options: options, + }, nil +} + +func (a *Adapter) Format(state fmt.State, _ rune) { + if a == nil { + _, _ = state.Write([]byte("redisactivity.Adapter")) + return + } + _, _ = fmt.Fprintf(state, "redisactivity.Adapter{Namespace:%q}", a.options.Namespace) +} + +func nilInterface(value any) bool { + if value == nil { + return true + } + reflected := reflect.ValueOf(value) + switch reflected.Kind() { + case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice: + return reflected.IsNil() + default: + return false + } +} diff --git a/internal/adapters/redisactivity/adapter_test.go b/internal/adapters/redisactivity/adapter_test.go new file mode 100644 index 0000000..64f78f3 --- /dev/null +++ b/internal/adapters/redisactivity/adapter_test.go @@ -0,0 +1,232 @@ +package redisactivity + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + "time" + + "github.com/redis/go-redis/v9" + + extractionDomain "proxy-pool/internal/domain/extraction" + proxyDomain "proxy-pool/internal/domain/proxy" + "proxy-pool/internal/platform/credentials" +) + +func TestRunScriptPreservesContextCancellation(t *testing.T) { + t.Parallel() + client := redis.NewClient(&redis.Options{Addr: "127.0.0.1:0"}) + t.Cleanup(func() { _ = client.Close() }) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err := runScript(ctx, client, redis.NewScript("return 1"), nil) + if !errors.Is(err, context.Canceled) || errors.Is(err, extractionDomain.ErrStoreUnavailable) { + t.Fatalf("runScript() error = %v, want only context cancellation", err) + } +} + +func TestNewRejectsInvalidDependenciesAndOptions(t *testing.T) { + t.Parallel() + client := redis.NewClient(&redis.Options{Addr: "127.0.0.1:0"}) + t.Cleanup(func() { _ = client.Close() }) + credentialStore, err := credentials.NewMemoryStore(10) + if err != nil { + t.Fatalf("NewMemoryStore(): %v", err) + } + valid := Options{ + Namespace: "test-a", Credentials: credentialStore, + OperationTTL: time.Minute, MaxCandidateScan: 2048, CleanupLimit: 128, + } + var typedNilClient *redis.Client + var typedNilCredentials *credentials.MemoryStore + + tests := []struct { + name string + client redis.Scripter + options Options + }{ + {name: "nil client", options: valid}, + {name: "typed nil client", client: typedNilClient, options: valid}, + {name: "nil credentials", client: client, options: withCredentials(valid, nil)}, + {name: "typed nil credentials", client: client, options: withCredentials(valid, typedNilCredentials)}, + {name: "empty namespace", client: client, options: withNamespace(valid, "")}, + {name: "braces in namespace", client: client, options: withNamespace(valid, "tenant{other}")}, + {name: "colon in namespace", client: client, options: withNamespace(valid, "tenant:other")}, + {name: "zero operation ttl", client: client, options: withOperationTTL(valid, 0)}, + {name: "zero candidate scan", client: client, options: withMaxCandidateScan(valid, 0)}, + {name: "negative cleanup limit", client: client, options: withCleanupLimit(valid, -1)}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + if adapter, err := New(tt.client, tt.options); err == nil || adapter != nil { + t.Fatalf("New() = (%v, %v), want nil adapter and error", adapter, err) + } + }) + } +} + +func TestNewBuildsClusterSafeKeyspaceAndHashesDynamicTokens(t *testing.T) { + t.Parallel() + client := redis.NewClient(&redis.Options{Addr: "127.0.0.1:0"}) + t.Cleanup(func() { _ = client.Close() }) + credentialStore, err := credentials.NewMemoryStore(10) + if err != nil { + t.Fatalf("NewMemoryStore(): %v", err) + } + adapter, err := New(client, Options{ + Namespace: " test-a ", Credentials: credentialStore, + OperationTTL: time.Minute, MaxCandidateScan: 2048, CleanupLimit: 128, + }) + if err != nil { + t.Fatalf("New(): %v", err) + } + + if got := adapter.keys.records; got != "pp:{activity}:test-a:records" { + t.Fatalf("records key = %q", got) + } + staticKeys := []string{ + adapter.keys.records, adapter.keys.unique, adapter.keys.idkeys, + adapter.keys.expiry, adapter.keys.available, adapter.keys.owners, + adapter.keys.ownerExpiry, adapter.keys.epoch, adapter.keys.inventory, + } + for _, key := range staticKeys { + if strings.Count(key, "{activity}") != 1 || strings.Count(key, "{") != 1 || strings.Count(key, "}") != 1 { + t.Fatalf("key %q does not contain exactly one fixed hash tag", key) + } + } + + raw := "tenant:{unsafe}:TOKEN" + dynamicKeys := []string{ + adapter.keys.idempotency(raw, raw), + adapter.keys.operation(raw), + adapter.keys.protocol(raw), + adapter.keys.region(raw), + adapter.keys.carrier(raw), + adapter.keys.upstream(raw), + } + for _, key := range dynamicKeys { + if strings.Contains(key, raw) || strings.Count(key, "{activity}") != 1 || strings.Count(key, "{") != 1 { + t.Fatalf("dynamic key is unsafe: %q", key) + } + } + if got := digestToken(raw); len(got) != 64 || got != digestToken(raw) || strings.Contains(got, raw) { + t.Fatalf("digestToken() = %q", got) + } +} + +func TestProxyRecordCodecIsDeterministicStrictAndRedacted(t *testing.T) { + t.Parallel() + record := proxyRecord{ + Version: 1, ID: "proxy-a", Scheme: string(proxyDomain.SchemeHTTP), + Host: "192.0.2.10", Port: 8080, Username: "user", Password: "top-secret", + SourceUpstream: "provider-a", CreatedAtMS: 1_000, ExpiresAtMS: 61_000, + UsableUntilMS: 58_000, LastCheckedAtMS: 2_000, LastSuccessAtMS: 2_000, + LatencyNS: int64(25 * time.Millisecond), MaxConcurrency: 8, + State: string(proxyDomain.StateAvailable), Tags: map[string]string{"region": "cn", "carrier": "ct"}, + } + first, err := encodeProxyRecord(record) + if err != nil { + t.Fatalf("encodeProxyRecord(): %v", err) + } + second, err := encodeProxyRecord(record) + if err != nil || first != second { + t.Fatalf("deterministic encode = (%q, %q, %v)", first, second, err) + } + decoded, err := decodeProxyRecord(first) + if err != nil || decoded.Password != "top-secret" || decoded.State != string(proxyDomain.StateAvailable) { + t.Fatalf("decodeProxyRecord() = (%+v, %v)", decoded, err) + } + if formatted := fmt.Sprintf("%+v", record); strings.Contains(formatted, "top-secret") || !strings.Contains(formatted, "") { + t.Fatalf("proxyRecord formatting leaked password: %s", formatted) + } + + invalid := []string{ + strings.Replace(first, `"state":"AVAILABLE"`, `"state":"UNKNOWN"`, 1), + strings.Replace(first, `"latencyNs":25000000`, `"latencyNs":-1`, 1), + strings.TrimSuffix(first, "}") + `,"unexpected":true}`, + } + for _, payload := range invalid { + if _, err := decodeProxyRecord(payload); err == nil { + t.Fatalf("decodeProxyRecord(%s) error = nil", payload) + } + } +} + +func TestOwnershipAndIdempotencyCodecsValidateAndRedact(t *testing.T) { + t.Parallel() + assignment := ownershipRecord{ + Version: 1, ProxyID: "proxy-a", WorkerID: "worker-a", + Epoch: 2, AssignmentVersion: 3, ExpiresAtMS: 5_000, Draining: true, + } + payload, err := encodeOwnershipRecord(assignment) + if err != nil { + t.Fatalf("encodeOwnershipRecord(): %v", err) + } + if decoded, err := decodeOwnershipRecord(payload); err != nil || decoded != assignment { + t.Fatalf("decodeOwnershipRecord() = (%+v, %v)", decoded, err) + } + if _, err := decodeOwnershipRecord(strings.Replace(payload, `"epoch":2`, `"epoch":0`, 1)); err == nil { + t.Fatal("decodeOwnershipRecord() accepted zero epoch") + } + + idempotency := idempotencyRecord{ + Version: 1, RequestDigest: digestToken("request"), ExpiresAtMS: 10_000, + Result: extractionDomain.Result{ + Requested: 1, Returned: 1, ExtractedAt: time.UnixMilli(1_000).UTC(), + Items: []extractionDomain.Candidate{{ + ID: "proxy-a", Protocol: "http", Host: "192.0.2.10", Port: 8080, + Username: "user", Password: "top-secret", Upstream: "provider-a", State: extractionDomain.Extracted, + ExpiresAt: time.UnixMilli(10_000).UTC(), + }}, + }, + } + payload, err = encodeIdempotencyRecord(idempotency) + if err != nil { + t.Fatalf("encodeIdempotencyRecord(): %v", err) + } + decodedID, err := decodeIdempotencyRecord(payload) + if err != nil || decodedID.Result.Items[0].Password != "top-secret" { + t.Fatalf("decodeIdempotencyRecord() = (%+v, %v)", decodedID, err) + } + invalidPayloads := []string{ + strings.Replace(payload, `"port":8080`, `"port":65537`, 1), + strings.Replace(payload, `"state":"EXTRACTED","expiresAtMs":10000`, `"state":"EXTRACTED","expiresAtMs":10000,"lastCheckedAtMs":-1`, 1), + } + for _, invalidPayload := range invalidPayloads { + if _, err := decodeIdempotencyRecord(invalidPayload); err == nil { + t.Fatalf("decodeIdempotencyRecord(%s) error = nil", invalidPayload) + } + } + if formatted := fmt.Sprintf("%+v", idempotency); strings.Contains(formatted, "top-secret") || !strings.Contains(formatted, "") { + t.Fatalf("idempotencyRecord formatting leaked password: %s", formatted) + } +} + +func withCredentials(options Options, store credentials.Store) Options { + options.Credentials = store + return options +} + +func withNamespace(options Options, namespace string) Options { + options.Namespace = namespace + return options +} + +func withOperationTTL(options Options, ttl time.Duration) Options { + options.OperationTTL = ttl + return options +} + +func withMaxCandidateScan(options Options, limit int) Options { + options.MaxCandidateScan = limit + return options +} + +func withCleanupLimit(options Options, limit int) Options { + options.CleanupLimit = limit + return options +} diff --git a/internal/adapters/redisactivity/codec.go b/internal/adapters/redisactivity/codec.go new file mode 100644 index 0000000..2c199d7 --- /dev/null +++ b/internal/adapters/redisactivity/codec.go @@ -0,0 +1,350 @@ +package redisactivity + +import ( + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "strings" + "time" + + extractionDomain "proxy-pool/internal/domain/extraction" + proxyDomain "proxy-pool/internal/domain/proxy" +) + +var ErrInvalidRecord = errors.New("invalid redis activity record") + +const recordVersion = 1 + +type proxyRecord struct { + Version int `json:"version"` + ID string `json:"id"` + Scheme string `json:"scheme"` + Host string `json:"host"` + Port int64 `json:"port"` + Username string `json:"username,omitempty"` + Password string `json:"password,omitempty"` + SourceUpstream string `json:"sourceUpstream"` + CreatedAtMS int64 `json:"createdAtMs"` + ExpiresAtMS int64 `json:"expiresAtMs"` + UsableUntilMS int64 `json:"usableUntilMs"` + LastCheckedAtMS int64 `json:"lastCheckedAtMs,omitempty"` + LastSuccessAtMS int64 `json:"lastSuccessAtMs,omitempty"` + LatencyNS int64 `json:"latencyNs"` + MaxConcurrency int64 `json:"maxConcurrency"` + State string `json:"state"` + Tags map[string]string `json:"tags,omitempty"` + OwnerWorkerID string `json:"ownerWorkerId,omitempty"` +} + +type ownershipRecord struct { + Version int `json:"version"` + ProxyID string `json:"proxyId"` + WorkerID string `json:"workerId"` + Epoch uint64 `json:"epoch"` + AssignmentVersion uint64 `json:"assignmentVersion"` + ExpiresAtMS int64 `json:"expiresAtMs"` + Draining bool `json:"draining"` +} + +type idempotencyRecord struct { + Version int + RequestDigest string + ExpiresAtMS int64 + Result extractionDomain.Result +} + +type idempotencyWire struct { + Version int `json:"version"` + RequestDigest string `json:"requestDigest"` + ExpiresAtMS int64 `json:"expiresAtMs"` + Result extractionWire `json:"result"` +} + +type extractionWire struct { + Requested int `json:"requested"` + Returned int `json:"returned"` + ExtractedAtMS int64 `json:"extractedAtMs,omitempty"` + Items []candidateWire `json:"items"` +} + +type candidateWire struct { + ID string `json:"id"` + Protocol string `json:"protocol"` + Host string `json:"host"` + Port int64 `json:"port"` + Username string `json:"username,omitempty"` + Password string `json:"password,omitempty"` + Region string `json:"region,omitempty"` + Carrier string `json:"carrier,omitempty"` + Upstream string `json:"upstream"` + OwnerWorkerID string `json:"ownerWorkerId,omitempty"` + URL string `json:"url,omitempty"` + State string `json:"state"` + ExpiresAtMS int64 `json:"expiresAtMs"` + CheckedAtMS int64 `json:"lastCheckedAtMs,omitempty"` +} + +func (record proxyRecord) Format(state fmt.State, _ rune) { + _, _ = fmt.Fprintf(state, + "redisactivity.proxyRecord{ID:%q, Scheme:%q, Host:%q, Port:%d, Username:%q, Password:, State:%q}", + record.ID, record.Scheme, record.Host, record.Port, record.Username, record.State, + ) +} + +func (record ownershipRecord) Format(state fmt.State, _ rune) { + _, _ = fmt.Fprintf(state, + "redisactivity.ownershipRecord{ProxyID:%q, WorkerID:%q, Epoch:%d, Version:%d, Draining:%t}", + record.ProxyID, record.WorkerID, record.Epoch, record.AssignmentVersion, record.Draining, + ) +} + +func (record idempotencyRecord) Format(state fmt.State, _ rune) { + _, _ = fmt.Fprintf(state, + "redisactivity.idempotencyRecord{RequestDigest:%q, Returned:%d, Credentials:}", + record.RequestDigest, record.Result.Returned, + ) +} + +func encodeProxyRecord(record proxyRecord) (string, error) { + if err := validateProxyRecord(record); err != nil { + return "", err + } + return encodeJSON(record) +} + +func decodeProxyRecord(payload string) (proxyRecord, error) { + var record proxyRecord + if err := decodeJSON(payload, &record); err != nil { + return proxyRecord{}, err + } + if err := validateProxyRecord(record); err != nil { + return proxyRecord{}, err + } + return record, nil +} + +func encodeOwnershipRecord(record ownershipRecord) (string, error) { + if err := validateOwnershipRecord(record); err != nil { + return "", err + } + return encodeJSON(record) +} + +func decodeOwnershipRecord(payload string) (ownershipRecord, error) { + var record ownershipRecord + if err := decodeJSON(payload, &record); err != nil { + return ownershipRecord{}, err + } + if err := validateOwnershipRecord(record); err != nil { + return ownershipRecord{}, err + } + return record, nil +} + +func encodeIdempotencyRecord(record idempotencyRecord) (string, error) { + if err := validateIdempotencyRecord(record); err != nil { + return "", err + } + wire := idempotencyWire{ + Version: record.Version, RequestDigest: record.RequestDigest, ExpiresAtMS: record.ExpiresAtMS, + Result: extractionToWire(record.Result), + } + return encodeJSON(wire) +} + +func decodeIdempotencyRecord(payload string) (idempotencyRecord, error) { + var wire idempotencyWire + if err := decodeJSON(payload, &wire); err != nil { + return idempotencyRecord{}, err + } + if err := validateIdempotencyWire(wire); err != nil { + return idempotencyRecord{}, err + } + record := idempotencyRecord{ + Version: wire.Version, RequestDigest: wire.RequestDigest, ExpiresAtMS: wire.ExpiresAtMS, + Result: extractionFromWire(wire.Result), + } + if err := validateIdempotencyRecord(record); err != nil { + return idempotencyRecord{}, err + } + return record, nil +} + +func encodeJSON(value any) (string, error) { + payload, err := json.Marshal(value) + if err != nil { + return "", errors.Join(ErrInvalidRecord, err) + } + return string(payload), nil +} + +func decodeJSON(payload string, destination any) error { + decoder := json.NewDecoder(strings.NewReader(payload)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(destination); err != nil { + return errors.Join(ErrInvalidRecord, err) + } + if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) { + if err == nil { + err = errors.New("multiple JSON values") + } + return errors.Join(ErrInvalidRecord, err) + } + return nil +} + +func validateProxyRecord(record proxyRecord) error { + if record.Version != recordVersion || record.ID == "" || record.Host == "" || + record.Port <= 0 || record.Port > 65_535 || record.SourceUpstream == "" || + record.CreatedAtMS <= 0 || record.ExpiresAtMS <= 0 || record.UsableUntilMS <= 0 || + record.UsableUntilMS > record.ExpiresAtMS || record.LastCheckedAtMS < 0 || + record.LastSuccessAtMS < 0 || record.LatencyNS < 0 || record.MaxConcurrency < 0 || + !validScheme(record.Scheme) || !validProxyState(record.State) { + return ErrInvalidRecord + } + return nil +} + +func validateOwnershipRecord(record ownershipRecord) error { + if record.Version != recordVersion || record.ProxyID == "" || record.WorkerID == "" || + record.Epoch == 0 || record.AssignmentVersion == 0 || record.ExpiresAtMS <= 0 { + return ErrInvalidRecord + } + return nil +} + +func validateIdempotencyRecord(record idempotencyRecord) error { + if record.Version != recordVersion { + return invalidRecord("unsupported idempotency version") + } + if !validDigest(record.RequestDigest) { + return invalidRecord("invalid request digest") + } + if record.ExpiresAtMS <= 0 { + return invalidRecord("invalid idempotency expiry") + } + if record.Result.Requested < 0 || record.Result.Returned < 0 || + record.Result.Returned != len(record.Result.Items) || record.Result.Returned > record.Result.Requested { + return invalidRecord("invalid extraction counters") + } + if record.Result.Returned > 0 && record.Result.ExtractedAt.IsZero() { + return invalidRecord("missing extraction time") + } + for _, candidate := range record.Result.Items { + if candidate.ID == "" || candidate.Host == "" || candidate.Port == 0 || candidate.Upstream == "" { + return invalidRecord("incomplete extraction candidate identity") + } + if !validScheme(candidate.Protocol) || candidate.State != extractionDomain.Extracted { + return invalidRecord("invalid extraction candidate state") + } + if candidate.ExpiresAt.IsZero() || candidate.ExpiresAt.UnixMilli() <= 0 || + (!candidate.LastCheckedAt.IsZero() && candidate.LastCheckedAt.UnixMilli() <= 0) { + return invalidRecord("invalid extraction candidate time") + } + } + return nil +} + +func validateIdempotencyWire(wire idempotencyWire) error { + if wire.Version != recordVersion || !validDigest(wire.RequestDigest) || wire.ExpiresAtMS <= 0 { + return invalidRecord("invalid idempotency wire header") + } + if wire.Result.Requested < 0 || wire.Result.Returned < 0 || + wire.Result.Returned != len(wire.Result.Items) || wire.Result.Returned > wire.Result.Requested { + return invalidRecord("invalid extraction wire counters") + } + if wire.Result.Returned > 0 && wire.Result.ExtractedAtMS <= 0 { + return invalidRecord("invalid extraction wire time") + } + for _, candidate := range wire.Result.Items { + if candidate.ID == "" || candidate.Host == "" || candidate.Port <= 0 || candidate.Port > 65_535 || + candidate.Upstream == "" || !validScheme(candidate.Protocol) || + extractionDomain.State(candidate.State) != extractionDomain.Extracted || + candidate.ExpiresAtMS <= 0 || candidate.CheckedAtMS < 0 { + return invalidRecord("invalid extraction wire candidate") + } + } + return nil +} + +func invalidRecord(reason string) error { + return fmt.Errorf("%w: %s", ErrInvalidRecord, reason) +} + +func extractionToWire(result extractionDomain.Result) extractionWire { + wire := extractionWire{ + Requested: result.Requested, Returned: result.Returned, + Items: make([]candidateWire, 0, len(result.Items)), + } + if !result.ExtractedAt.IsZero() { + wire.ExtractedAtMS = result.ExtractedAt.UnixMilli() + } + for _, candidate := range result.Items { + item := candidateWire{ + ID: candidate.ID, Protocol: candidate.Protocol, Host: candidate.Host, Port: int64(candidate.Port), + Username: candidate.Username, Password: candidate.Password, Region: candidate.Region, + Carrier: candidate.Carrier, Upstream: candidate.Upstream, OwnerWorkerID: candidate.OwnerWorkerID, + URL: candidate.URL, State: string(candidate.State), ExpiresAtMS: candidate.ExpiresAt.UnixMilli(), + } + if !candidate.LastCheckedAt.IsZero() { + item.CheckedAtMS = candidate.LastCheckedAt.UnixMilli() + } + wire.Items = append(wire.Items, item) + } + return wire +} + +func extractionFromWire(wire extractionWire) extractionDomain.Result { + result := extractionDomain.Result{ + Requested: wire.Requested, Returned: wire.Returned, + Items: make([]extractionDomain.Candidate, 0, len(wire.Items)), + } + if wire.ExtractedAtMS > 0 { + result.ExtractedAt = time.UnixMilli(wire.ExtractedAtMS).UTC() + } + for _, item := range wire.Items { + candidate := extractionDomain.Candidate{ + ID: item.ID, Protocol: item.Protocol, Host: item.Host, Port: uint16(item.Port), + Username: item.Username, Password: item.Password, Region: item.Region, + Carrier: item.Carrier, Upstream: item.Upstream, OwnerWorkerID: item.OwnerWorkerID, + URL: item.URL, State: extractionDomain.State(item.State), ExpiresAt: time.UnixMilli(item.ExpiresAtMS).UTC(), + } + if item.CheckedAtMS > 0 { + candidate.LastCheckedAt = time.UnixMilli(item.CheckedAtMS).UTC() + } + result.Items = append(result.Items, candidate) + } + return result +} + +func validDigest(value string) bool { + if len(value) != sha256HexSize { + return false + } + _, err := hex.DecodeString(value) + return err == nil +} + +func validScheme(value string) bool { + switch proxyDomain.Scheme(value) { + case proxyDomain.SchemeHTTP, proxyDomain.SchemeHTTPS, proxyDomain.SchemeSOCKS5: + return true + default: + return false + } +} + +func validProxyState(value string) bool { + switch proxyDomain.State(value) { + case proxyDomain.StateFetched, proxyDomain.StateChecking, proxyDomain.StateAvailable, + proxyDomain.StateSuspect, proxyDomain.StateDraining, proxyDomain.StateUnhealthy, + proxyDomain.StateExtracted, proxyDomain.StateExpired, proxyDomain.StateRemoved: + return true + default: + return false + } +} + +const sha256HexSize = 64 diff --git a/internal/adapters/redisactivity/keys.go b/internal/adapters/redisactivity/keys.go new file mode 100644 index 0000000..1a19ce0 --- /dev/null +++ b/internal/adapters/redisactivity/keys.go @@ -0,0 +1,80 @@ +package redisactivity + +import ( + "crypto/sha256" + "encoding/hex" + "strconv" +) + +const redisKeyPrefix = "pp:{activity}:" + +type keyspace struct { + prefix string + records string + unique string + idkeys string + expiry string + available string + owners string + ownerExpiry string + epoch string + inventory string +} + +func newKeyspace(namespace string) keyspace { + prefix := redisKeyPrefix + namespace + return keyspace{ + prefix: prefix, + records: prefix + ":records", + unique: prefix + ":unique", + idkeys: prefix + ":idkeys", + expiry: prefix + ":expiry", + available: prefix + ":available", + owners: prefix + ":owners", + ownerExpiry: prefix + ":owner-expiry", + epoch: prefix + ":epoch", + inventory: prefix + ":inventory", + } +} + +func (keys keyspace) idempotency(clientID, idempotencyKey string) string { + return keys.prefix + ":idem:" + digestParts(clientID, idempotencyKey) +} + +func (keys keyspace) operation(operationID string) string { + return keys.prefix + ":op:" + digestToken(operationID) +} + +func (keys keyspace) protocol(value string) string { + return keys.facet("protocol", value) +} + +func (keys keyspace) region(value string) string { + return keys.facet("region", value) +} + +func (keys keyspace) carrier(value string) string { + return keys.facet("carrier", value) +} + +func (keys keyspace) upstream(value string) string { + return keys.facet("upstream", value) +} + +func (keys keyspace) facet(name, value string) string { + return keys.prefix + ":" + name + ":" + digestToken(value) +} + +func digestToken(value string) string { + return digestParts(value) +} + +func digestParts(values ...string) string { + digest := sha256.New() + for _, value := range values { + _, _ = digest.Write([]byte(strconv.Itoa(len(value)))) + _, _ = digest.Write([]byte{':'}) + _, _ = digest.Write([]byte(value)) + } + return hex.EncodeToString(digest.Sum(nil)) +} diff --git a/internal/adapters/redisactivity/scripts.go b/internal/adapters/redisactivity/scripts.go new file mode 100644 index 0000000..d3c9d2c --- /dev/null +++ b/internal/adapters/redisactivity/scripts.go @@ -0,0 +1,64 @@ +package redisactivity + +import ( + "context" + "errors" + "fmt" + + "github.com/redis/go-redis/v9" + + extractionDomain "proxy-pool/internal/domain/extraction" +) + +type scriptStatus string + +const ( + scriptOK scriptStatus = "ok" + scriptInvalid scriptStatus = "invalid" + scriptNotFound scriptStatus = "not_found" + scriptConflict scriptStatus = "conflict" + scriptUnavailable scriptStatus = "unavailable" + scriptInsufficient scriptStatus = "insufficient" +) + +type upsertScriptReply struct { + Status scriptStatus `json:"status"` + Accepted int `json:"accepted"` + Inserted int `json:"inserted"` + Refreshed int `json:"refreshed"` + Dropped int `json:"dropped"` +} + +type healthScriptReply struct { + Status scriptStatus `json:"status"` + Record string `json:"record,omitempty"` +} + +type extractScriptReply struct { + Status scriptStatus `json:"status"` + Record string `json:"record,omitempty"` +} + +type ownershipScriptReply struct { + Status scriptStatus `json:"status"` + Record string `json:"record,omitempty"` +} + +type maintenanceScriptReply struct { + Status scriptStatus `json:"status"` + Count int `json:"count"` +} + +func runScript(ctx context.Context, client redis.Scripter, script *redis.Script, keys []string, args ...any) (any, error) { + result, err := script.Run(ctx, client, keys, args...).Result() + if err != nil { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return nil, err + } + return nil, errors.Join( + extractionDomain.ErrStoreUnavailable, + fmt.Errorf("run redis activity script: %w", err), + ) + } + return result, nil +} diff --git a/internal/adapters/redisactivity/testredis_test.go b/internal/adapters/redisactivity/testredis_test.go new file mode 100644 index 0000000..1e8b7de --- /dev/null +++ b/internal/adapters/redisactivity/testredis_test.go @@ -0,0 +1,99 @@ +//go:build integration + +package redisactivity + +import ( + "context" + "fmt" + "os" + "sync/atomic" + "testing" + "time" + + "github.com/redis/go-redis/v9" + + "proxy-pool/internal/platform/credentials" +) + +var testNamespaceSequence atomic.Uint64 + +type redisTestFixture struct { + Adapter *Adapter + Client *redis.Client + Credentials *credentials.MemoryStore + Namespace string +} + +func TestRedisFixtureUsesIsolatedNamespace(t *testing.T) { + fixture := newRedisTestFixture(t) + key := fixture.Adapter.keys.operation("fixture-probe") + if err := fixture.Client.Set(t.Context(), key, "ok", time.Minute).Err(); err != nil { + t.Fatalf("write isolated fixture key: %v", err) + } + if value, err := fixture.Client.Get(t.Context(), key).Result(); err != nil || value != "ok" { + t.Fatalf("read isolated fixture key = (%q, %v)", value, err) + } +} + +func newRedisTestFixture(t *testing.T) redisTestFixture { + t.Helper() + redisURL := os.Getenv("PROXY_POOL_TEST_REDIS_URL") + if redisURL == "" { + t.Skip("PROXY_POOL_TEST_REDIS_URL is not set") + } + redisOptions, err := redis.ParseURL(redisURL) + if err != nil { + t.Fatalf("parse PROXY_POOL_TEST_REDIS_URL: %v", err) + } + client := redis.NewClient(redisOptions) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := client.Ping(ctx).Err(); err != nil { + _ = client.Close() + t.Fatalf("ping test Redis: %v", err) + } + credentialStore, err := credentials.NewMemoryStore(10_000) + if err != nil { + _ = client.Close() + t.Fatalf("NewMemoryStore(): %v", err) + } + namespace := fmt.Sprintf("it-%d-%d-%d", os.Getpid(), time.Now().UnixNano(), testNamespaceSequence.Add(1)) + adapter, err := New(client, Options{ + Namespace: namespace, Credentials: credentialStore, + OperationTTL: time.Minute, MaxCandidateScan: 2_048, CleanupLimit: 128, + }) + if err != nil { + _ = client.Close() + t.Fatalf("New(): %v", err) + } + t.Cleanup(func() { + cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cleanupCancel() + if err := deleteRedisNamespace(cleanupCtx, client, redisKeyPrefix+namespace+":*"); err != nil { + t.Errorf("clean Redis test namespace: %v", err) + } + _ = client.Close() + }) + return redisTestFixture{ + Adapter: adapter, Client: client, Credentials: credentialStore, Namespace: namespace, + } +} + +func deleteRedisNamespace(ctx context.Context, client *redis.Client, pattern string) error { + var cursor uint64 + for { + keys, next, err := client.Scan(ctx, cursor, pattern, 128).Result() + if err != nil { + return err + } + if len(keys) > 0 { + if err := client.Unlink(ctx, keys...).Err(); err != nil { + return err + } + } + cursor = next + if cursor == 0 { + return nil + } + } +} diff --git a/scripts/test-redis.ps1 b/scripts/test-redis.ps1 new file mode 100644 index 0000000..bc8d60f --- /dev/null +++ b/scripts/test-redis.ps1 @@ -0,0 +1,33 @@ +$ErrorActionPreference = "Stop" + +$repositoryRoot = Split-Path -Parent $PSScriptRoot +$composeFile = Join-Path $repositoryRoot "deploy/docker-compose.test.yml" +$previousRedisURL = [Environment]::GetEnvironmentVariable("PROXY_POOL_TEST_REDIS_URL", "Process") + +try { + docker compose -f $composeFile up -d --wait + if ($LASTEXITCODE -ne 0) { + throw "starting Redis test fixture failed with exit code $LASTEXITCODE" + } + + $env:PROXY_POOL_TEST_REDIS_URL = "redis://127.0.0.1:16379/15" + Push-Location $repositoryRoot + try { + go test -count=1 -tags=integration -timeout 60s ./internal/adapters/redisactivity + if ($LASTEXITCODE -ne 0) { + throw "Redis integration tests failed with exit code $LASTEXITCODE" + } + } + finally { + Pop-Location + } +} +finally { + if ($null -eq $previousRedisURL) { + Remove-Item Env:PROXY_POOL_TEST_REDIS_URL -ErrorAction SilentlyContinue + } + else { + $env:PROXY_POOL_TEST_REDIS_URL = $previousRedisURL + } + docker compose -f $composeFile down --remove-orphans +}