feat: add redis activity upsert and health

internal/adapters/redisactivity/codec.go

@@ -25,6 +25,7 @@ type proxyRecord struct {
 	Port              int64             `json:"port"`
 	Username          string            `json:"username,omitempty"`
 	Password          string            `json:"password,omitempty"`
+	CredentialVersion string            `json:"credentialVersion,omitempty"`
 	SourceUpstream    string            `json:"sourceUpstream"`
 	CreatedAtMS       int64             `json:"createdAtMs"`
 	ExpiresAtMS       int64             `json:"expiresAtMs"`
@@ -36,6 +37,7 @@ type proxyRecord struct {
 	State             string            `json:"state"`
 	Tags              map[string]string `json:"tags,omitempty"`
 	OwnerWorkerID     string            `json:"ownerWorkerId,omitempty"`
+	IndexKeys         []string          `json:"indexKeys,omitempty"`
 }
 
 type ownershipRecord struct {
@@ -204,6 +206,11 @@ func validateProxyRecord(record proxyRecord) error {
 		!validScheme(record.Scheme) || !validProxyState(record.State) {
 		return ErrInvalidRecord
 	}
+	for _, key := range record.IndexKeys {
+		if key == "" || !strings.Contains(key, "{activity}") {
+			return ErrInvalidRecord
+		}
+	}
 	return nil
 }
 

internal/adapters/redisactivity/health.go

@@ -0,0 +1,57 @@
+package redisactivity
+
+import (
+	"context"
+
+	"proxy-pool/internal/domain/activitypool"
+)
+
+var _ activitypool.HealthStore = (*Adapter)(nil)
+
+func (a *Adapter) ApplyHealth(ctx context.Context, update activitypool.HealthUpdate) (activitypool.Entry, error) {
+	if ctx == nil {
+		return activitypool.Entry{}, activitypool.ErrInvalidHealthUpdate
+	}
+	if err := ctx.Err(); err != nil {
+		return activitypool.Entry{}, err
+	}
+	if a == nil || update.ProxyID == "" || update.CheckedAt.IsZero() || update.Latency < 0 ||
+		!validProxyState(string(update.NextState)) {
+		return activitypool.Entry{}, activitypool.ErrInvalidHealthUpdate
+	}
+	operationID, err := newOperationID()
+	if err != nil {
+		return activitypool.Entry{}, err
+	}
+	result, err := runScript(ctx, a.client, healthScript, []string{
+		a.keys.records, a.keys.unique, a.keys.idkeys, a.keys.expiry, a.keys.available,
+		a.keys.inventory, a.keys.owners, a.keys.ownerExpiry, a.keys.operation(operationID),
+	}, update.CheckedAt.UnixMilli(), string(update.NextState), int64(update.Latency),
+		a.options.CleanupLimit, operationTTLMillis(a.options.OperationTTL), update.ProxyID)
+	if err != nil {
+		return activitypool.Entry{}, err
+	}
+	var reply healthScriptReply
+	if err := decodeScriptResult(result, &reply); err != nil {
+		return activitypool.Entry{}, err
+	}
+	switch reply.Status {
+	case scriptNotFound:
+		return activitypool.Entry{}, activitypool.ErrActivityNotFound
+	case scriptStale:
+		return activitypool.Entry{}, activitypool.ErrStaleHealthUpdate
+	case scriptInvalid:
+		return activitypool.Entry{}, activitypool.ErrInvalidHealthUpdate
+	case scriptOK:
+		if reply.Record == "" {
+			return activitypool.Entry{}, invalidScriptReply("health reply omitted record")
+		}
+		record, err := decodeProxyRecord(reply.Record)
+		if err != nil {
+			return activitypool.Entry{}, invalidScriptReply("health reply contained an invalid record")
+		}
+		return proxyRecordEntry(record), nil
+	default:
+		return activitypool.Entry{}, invalidScriptReply("unexpected health status")
+	}
+}

internal/adapters/redisactivity/scripts.go

@@ -2,8 +2,12 @@ package redisactivity
 
 import (
 	"context"
+	"crypto/rand"
+	_ "embed"
+	"encoding/hex"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/redis/go-redis/v9"
 
@@ -17,6 +21,7 @@ const (
 	scriptInvalid      scriptStatus = "invalid"
 	scriptNotFound     scriptStatus = "not_found"
 	scriptConflict     scriptStatus = "conflict"
+	scriptStale        scriptStatus = "stale"
 	scriptUnavailable  scriptStatus = "unavailable"
 	scriptInsufficient scriptStatus = "insufficient"
 )
@@ -49,6 +54,17 @@ type maintenanceScriptReply struct {
 	Count  int          `json:"count"`
 }
 
+//go:embed scripts/upsert.lua
+var upsertSource string
+
+//go:embed scripts/health.lua
+var healthSource string
+
+var (
+	upsertScript = redis.NewScript(upsertSource)
+	healthScript = redis.NewScript(healthSource)
+)
+
 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 {
@@ -62,3 +78,35 @@ func runScript(ctx context.Context, client redis.Scripter, script *redis.Script,
 	}
 	return result, nil
 }
+
+func newOperationID() (string, error) {
+	var value [16]byte
+	if _, err := rand.Read(value[:]); err != nil {
+		return "", fmt.Errorf("create redis activity operation ID: %w", err)
+	}
+	return hex.EncodeToString(value[:]), nil
+}
+
+func operationTTLMillis(ttl time.Duration) int64 {
+	milliseconds := ttl / time.Millisecond
+	if ttl%time.Millisecond != 0 {
+		milliseconds++
+	}
+	return int64(milliseconds)
+}
+
+func decodeScriptResult(result any, destination any) error {
+	var payload string
+	switch value := result.(type) {
+	case string:
+		payload = value
+	case []byte:
+		payload = string(value)
+	default:
+		return errors.Join(extractionDomain.ErrStoreUnavailable, errors.New("invalid Redis script reply type"))
+	}
+	if err := decodeJSON(payload, destination); err != nil {
+		return errors.Join(extractionDomain.ErrStoreUnavailable, fmt.Errorf("decode Redis script reply: %w", err))
+	}
+	return nil
+}

internal/adapters/redisactivity/scripts/health.lua

@@ -0,0 +1,169 @@
+local records_key = KEYS[1]
+local unique_key = KEYS[2]
+local idkeys_key = KEYS[3]
+local expiry_key = KEYS[4]
+local available_key = KEYS[5]
+local inventory_key = KEYS[6]
+local owners_key = KEYS[7]
+local owner_expiry_key = KEYS[8]
+local operation_key = KEYS[9]
+
+local checked_at_ms = tonumber(ARGV[1])
+local next_state = ARGV[2]
+local latency_ns = tonumber(ARGV[3])
+local cleanup_limit = tonumber(ARGV[4])
+local operation_ttl_ms = tonumber(ARGV[5])
+local proxy_id = ARGV[6]
+
+local committed = redis.call('GET', operation_key)
+if committed then
+    return committed
+end
+
+local function is_managed(state)
+    return state == 'FETCHED' or state == 'CHECKING' or state == 'AVAILABLE' or
+        state == 'SUSPECT' or state == 'DRAINING'
+end
+
+local function decrement_inventory(upstream)
+    if not upstream or upstream == '' then
+        return
+    end
+    local value = redis.call('HINCRBY', inventory_key, upstream, -1)
+    if value < 0 then
+        redis.call('HSET', inventory_key, upstream, 0)
+    end
+end
+
+local function remove_available(id, record)
+    redis.call('ZREM', available_key, id)
+    for _, index_key in ipairs(record and record.indexKeys or {}) do
+        redis.call('ZREM', index_key, id)
+    end
+end
+
+local function remove_proxy(id)
+    local raw = redis.call('HGET', records_key, id)
+    local record = nil
+    if raw then
+        record = cjson.decode(raw)
+        remove_available(id, record)
+        if is_managed(record.state) then
+            decrement_inventory(record.sourceUpstream)
+        end
+    else
+        redis.call('ZREM', available_key, id)
+    end
+    local digest = redis.call('HGET', idkeys_key, id)
+    if digest and redis.call('HGET', unique_key, digest) == id then
+        redis.call('HDEL', unique_key, digest)
+    end
+    redis.call('HDEL', idkeys_key, id)
+    redis.call('HDEL', records_key, id)
+    redis.call('ZREM', expiry_key, id)
+    redis.call('HDEL', owners_key, id)
+    redis.call('ZREM', owner_expiry_key, id)
+end
+
+local function cleanup_expired()
+    local expired = redis.call('ZRANGEBYSCORE', expiry_key, '-inf', checked_at_ms, 'LIMIT', 0, cleanup_limit)
+    for _, id in ipairs(expired) do
+        remove_proxy(id)
+    end
+end
+
+local function touch(key, expires_at_ms)
+    if redis.call('EXISTS', key) == 0 then
+        return
+    end
+    local current = redis.call('PEXPIRETIME', key)
+    if current < expires_at_ms then
+        redis.call('PEXPIREAT', key, expires_at_ms)
+    end
+end
+
+local function finish(reply, hard_expiry_ms)
+    local encoded = cjson.encode(reply)
+    redis.call('SET', operation_key, encoded, 'PX', operation_ttl_ms)
+    if hard_expiry_ms then
+        local operation_expiry_ms = redis.call('PEXPIRETIME', operation_key)
+        if operation_expiry_ms > hard_expiry_ms then
+            redis.call('PEXPIREAT', operation_key, hard_expiry_ms)
+        end
+    end
+    return encoded
+end
+
+local transitions = {
+    FETCHED = {CHECKING = true, EXPIRED = true, REMOVED = true},
+    CHECKING = {AVAILABLE = true, UNHEALTHY = true, EXPIRED = true, REMOVED = true},
+    AVAILABLE = {SUSPECT = true, DRAINING = true, EXTRACTED = true, EXPIRED = true},
+    SUSPECT = {AVAILABLE = true, UNHEALTHY = true, DRAINING = true, EXPIRED = true},
+    DRAINING = {EXPIRED = true, UNHEALTHY = true, REMOVED = true},
+    UNHEALTHY = {CHECKING = true, REMOVED = true, EXPIRED = true},
+    EXTRACTED = {EXPIRED = true, REMOVED = true},
+    EXPIRED = {REMOVED = true},
+    REMOVED = {},
+}
+
+cleanup_expired()
+
+local raw = redis.call('HGET', records_key, proxy_id)
+if not raw then
+    return finish({status = 'not_found'})
+end
+local record = cjson.decode(raw)
+if tonumber(record.expiresAtMs) <= checked_at_ms then
+    remove_proxy(proxy_id)
+    return finish({status = 'not_found'})
+end
+
+local last_checked_at_ms = tonumber(record.lastCheckedAtMs or '0')
+if checked_at_ms < last_checked_at_ms then
+    return finish({status = 'stale'})
+end
+if checked_at_ms == last_checked_at_ms then
+    if record.state ~= next_state then
+        return finish({status = 'stale'})
+    end
+    return finish({status = 'ok', record = raw}, tonumber(record.expiresAtMs))
+end
+if record.state ~= next_state and not (transitions[record.state] and transitions[record.state][next_state]) then
+    return finish({status = 'invalid'})
+end
+
+local was_managed = is_managed(record.state)
+local will_be_managed = is_managed(next_state)
+remove_available(proxy_id, record)
+record.state = next_state
+record.lastCheckedAtMs = checked_at_ms
+record.latencyNs = latency_ns
+if next_state == 'AVAILABLE' then
+    record.lastSuccessAtMs = checked_at_ms
+end
+if was_managed and not will_be_managed then
+    decrement_inventory(record.sourceUpstream)
+elseif not was_managed and will_be_managed then
+    redis.call('HINCRBY', inventory_key, record.sourceUpstream, 1)
+end
+
+local encoded = cjson.encode(record)
+redis.call('HSET', records_key, proxy_id, encoded)
+local owned = (record.ownerWorkerId and record.ownerWorkerId ~= '') or redis.call('HEXISTS', owners_key, proxy_id) == 1
+if next_state == 'AVAILABLE' and not owned and tonumber(record.usableUntilMs) > checked_at_ms then
+    redis.call('ZADD', available_key, record.usableUntilMs, proxy_id)
+    for _, index_key in ipairs(record.indexKeys or {}) do
+        redis.call('ZADD', index_key, record.usableUntilMs, proxy_id)
+        touch(index_key, tonumber(record.expiresAtMs))
+    end
+end
+touch(records_key, tonumber(record.expiresAtMs))
+touch(unique_key, tonumber(record.expiresAtMs))
+touch(idkeys_key, tonumber(record.expiresAtMs))
+touch(expiry_key, tonumber(record.expiresAtMs))
+touch(available_key, tonumber(record.expiresAtMs))
+touch(inventory_key, tonumber(record.expiresAtMs))
+touch(owners_key, tonumber(record.expiresAtMs))
+touch(owner_expiry_key, tonumber(record.expiresAtMs))
+
+return finish({status = 'ok', record = encoded}, tonumber(record.expiresAtMs))

internal/adapters/redisactivity/scripts/upsert.lua

@@ -0,0 +1,192 @@
+local records_key = KEYS[1]
+local unique_key = KEYS[2]
+local idkeys_key = KEYS[3]
+local expiry_key = KEYS[4]
+local available_key = KEYS[5]
+local inventory_key = KEYS[6]
+local owners_key = KEYS[7]
+local owner_expiry_key = KEYS[8]
+local operation_key = KEYS[9]
+
+local now_ms = tonumber(ARGV[1])
+local cleanup_limit = tonumber(ARGV[2])
+local max_size = tonumber(ARGV[3])
+local operation_ttl_ms = tonumber(ARGV[4])
+local candidates = cjson.decode(ARGV[5])
+
+local committed = redis.call('GET', operation_key)
+if committed then
+    return committed
+end
+
+local function is_managed(state)
+    return state == 'FETCHED' or state == 'CHECKING' or state == 'AVAILABLE' or
+        state == 'SUSPECT' or state == 'DRAINING'
+end
+
+local function decrement_inventory(upstream)
+    if not upstream or upstream == '' then
+        return
+    end
+    local value = redis.call('HINCRBY', inventory_key, upstream, -1)
+    if value < 0 then
+        redis.call('HSET', inventory_key, upstream, 0)
+    end
+end
+
+local function remove_available(proxy_id, record)
+    redis.call('ZREM', available_key, proxy_id)
+    local indexes = record and record.indexKeys or {}
+    for _, index_key in ipairs(indexes) do
+        redis.call('ZREM', index_key, proxy_id)
+    end
+end
+
+local function remove_proxy(proxy_id)
+    local raw = redis.call('HGET', records_key, proxy_id)
+    local record = nil
+    if raw then
+        record = cjson.decode(raw)
+        remove_available(proxy_id, record)
+        if is_managed(record.state) then
+            decrement_inventory(record.sourceUpstream)
+        end
+    else
+        redis.call('ZREM', available_key, proxy_id)
+    end
+    local digest = redis.call('HGET', idkeys_key, proxy_id)
+    if digest and redis.call('HGET', unique_key, digest) == proxy_id then
+        redis.call('HDEL', unique_key, digest)
+    end
+    redis.call('HDEL', idkeys_key, proxy_id)
+    redis.call('HDEL', records_key, proxy_id)
+    redis.call('ZREM', expiry_key, proxy_id)
+    redis.call('HDEL', owners_key, proxy_id)
+    redis.call('ZREM', owner_expiry_key, proxy_id)
+end
+
+local function cleanup_expired()
+    local expired = redis.call('ZRANGEBYSCORE', expiry_key, '-inf', now_ms, 'LIMIT', 0, cleanup_limit)
+    for _, proxy_id in ipairs(expired) do
+        remove_proxy(proxy_id)
+    end
+end
+
+local function touch(key, expires_at_ms)
+    if redis.call('EXISTS', key) == 0 then
+        return
+    end
+    local current = redis.call('PEXPIRETIME', key)
+    if current < expires_at_ms then
+        redis.call('PEXPIREAT', key, expires_at_ms)
+    end
+end
+
+local function add_available(proxy_id, record)
+    if record.state ~= 'AVAILABLE' or (record.ownerWorkerId and record.ownerWorkerId ~= '') or
+        tonumber(record.usableUntilMs) <= now_ms then
+        return
+    end
+    redis.call('ZADD', available_key, record.usableUntilMs, proxy_id)
+    for _, index_key in ipairs(record.indexKeys or {}) do
+        redis.call('ZADD', index_key, record.usableUntilMs, proxy_id)
+        touch(index_key, tonumber(record.expiresAtMs))
+    end
+end
+
+local function finish(reply)
+    local encoded = cjson.encode(reply)
+    redis.call('SET', operation_key, encoded, 'PX', operation_ttl_ms)
+    return encoded
+end
+
+cleanup_expired()
+
+for _, candidate in ipairs(candidates) do
+    local mapped = redis.call('HGET', idkeys_key, candidate.proxyId)
+    if mapped and mapped ~= candidate.uniqueDigest then
+        return finish({status = 'invalid', accepted = 0, inserted = 0, refreshed = 0, dropped = 0})
+    end
+end
+
+local accepted = #candidates
+local inserted = 0
+local refreshed = 0
+local dropped = 0
+local max_expiry_ms = 0
+
+for _, candidate in ipairs(candidates) do
+    local incumbent_id = redis.call('HGET', unique_key, candidate.uniqueDigest)
+    local current_raw = incumbent_id and redis.call('HGET', records_key, incumbent_id) or nil
+    if incumbent_id and not current_raw then
+        remove_proxy(incumbent_id)
+        redis.call('HDEL', unique_key, candidate.uniqueDigest)
+        incumbent_id = nil
+    end
+
+    if current_raw then
+        local current = cjson.decode(current_raw)
+        if tonumber(current.expiresAtMs) <= now_ms then
+            remove_proxy(incumbent_id)
+            incumbent_id = nil
+            current_raw = nil
+        elseif current.state == 'EXTRACTED' or current.sourceUpstream ~= candidate.upstream then
+            refreshed = refreshed + 1
+        else
+            local incoming = cjson.decode(candidate.record)
+            remove_available(incumbent_id, current)
+            incoming.id = current.id
+            incoming.createdAtMs = current.createdAtMs
+            incoming.state = current.state
+            incoming.lastCheckedAtMs = current.lastCheckedAtMs
+            incoming.lastSuccessAtMs = current.lastSuccessAtMs
+            incoming.latencyNs = current.latencyNs
+            incoming.ownerWorkerId = current.ownerWorkerId
+            local encoded = cjson.encode(incoming)
+            redis.call('HSET', records_key, incumbent_id, encoded)
+            redis.call('ZADD', expiry_key, incoming.expiresAtMs, incumbent_id)
+            add_available(incumbent_id, incoming)
+            if tonumber(incoming.expiresAtMs) > max_expiry_ms then
+                max_expiry_ms = tonumber(incoming.expiresAtMs)
+            end
+            refreshed = refreshed + 1
+        end
+    end
+
+    if not incumbent_id then
+        local current_size = tonumber(redis.call('HGET', inventory_key, candidate.upstream) or '0')
+        if current_size >= max_size then
+            dropped = dropped + 1
+        else
+            local incoming = cjson.decode(candidate.record)
+            redis.call('HSET', records_key, candidate.proxyId, candidate.record)
+            redis.call('HSET', unique_key, candidate.uniqueDigest, candidate.proxyId)
+            redis.call('HSET', idkeys_key, candidate.proxyId, candidate.uniqueDigest)
+            redis.call('ZADD', expiry_key, incoming.expiresAtMs, candidate.proxyId)
+            if is_managed(incoming.state) then
+                redis.call('HINCRBY', inventory_key, candidate.upstream, 1)
+            end
+            add_available(candidate.proxyId, incoming)
+            if tonumber(incoming.expiresAtMs) > max_expiry_ms then
+                max_expiry_ms = tonumber(incoming.expiresAtMs)
+            end
+            inserted = inserted + 1
+        end
+    end
+end
+
+if max_expiry_ms > 0 then
+    touch(records_key, max_expiry_ms)
+    touch(unique_key, max_expiry_ms)
+    touch(idkeys_key, max_expiry_ms)
+    touch(expiry_key, max_expiry_ms)
+    touch(available_key, max_expiry_ms)
+    touch(inventory_key, max_expiry_ms)
+    touch(owners_key, max_expiry_ms)
+    touch(owner_expiry_key, max_expiry_ms)
+end
+
+return finish({
+    status = 'ok', accepted = accepted, inserted = inserted,
+    refreshed = refreshed, dropped = dropped,
+})

internal/adapters/redisactivity/upsert.go

@@ -0,0 +1,258 @@
+package redisactivity
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"proxy-pool/internal/domain/activitypool"
+	extractionDomain "proxy-pool/internal/domain/extraction"
+	proxyDomain "proxy-pool/internal/domain/proxy"
+	"proxy-pool/internal/platform/credentials"
+)
+
+const maxUpsertScriptBatch = 256
+
+type upsertCandidate struct {
+	ProxyID      string `json:"proxyId"`
+	UniqueDigest string `json:"uniqueDigest"`
+	Upstream     string `json:"upstream"`
+	Record       string `json:"record"`
+}
+
+var _ activitypool.Upserter = (*Adapter)(nil)
+
+func (a *Adapter) UpsertFetched(ctx context.Context, upstreamID string, batch activitypool.FetchedBatch) (activitypool.UpsertResult, error) {
+	var result activitypool.UpsertResult
+	if ctx == nil {
+		return result, activitypool.ErrInvalidBatch
+	}
+	if err := ctx.Err(); err != nil {
+		return result, err
+	}
+	if a == nil || upstreamID == "" || batch.ObservedAt.IsZero() || batch.ConfiguredTTL < 0 ||
+		batch.AllocationSafetyMargin < 0 || batch.MaxSize <= 0 ||
+		(batch.ConfiguredTTL > 0 && batch.AllocationSafetyMargin >= batch.ConfiguredTTL) {
+		return result, activitypool.ErrInvalidBatch
+	}
+	for _, candidate := range batch.Proxies {
+		if candidate.SourceUpstream != "" && candidate.SourceUpstream != upstreamID {
+			return result, activitypool.ErrInvalidBatch
+		}
+	}
+
+	prepared := make([]upsertCandidate, 0, len(batch.Proxies))
+	seenIDs := make(map[string]string, len(batch.Proxies))
+	for _, candidate := range batch.Proxies {
+		if err := ctx.Err(); err != nil {
+			return activitypool.UpsertResult{}, err
+		}
+		item, accepted, err := a.prepareUpsertCandidate(ctx, upstreamID, batch, candidate)
+		if err != nil {
+			return activitypool.UpsertResult{}, err
+		}
+		if !accepted {
+			result.Dropped++
+			continue
+		}
+		if digest, exists := seenIDs[item.ProxyID]; exists && digest != item.UniqueDigest {
+			return activitypool.UpsertResult{}, activitypool.ErrInvalidBatch
+		}
+		seenIDs[item.ProxyID] = item.UniqueDigest
+		prepared = append(prepared, item)
+	}
+	result.Accepted = len(prepared)
+
+	for start := 0; start < len(prepared); start += maxUpsertScriptBatch {
+		end := min(start+maxUpsertScriptBatch, len(prepared))
+		reply, err := a.upsertChunk(ctx, batch.ObservedAt, batch.MaxSize, prepared[start:end])
+		if err != nil {
+			return activitypool.UpsertResult{}, err
+		}
+		if reply.Accepted != end-start || reply.Inserted < 0 || reply.Refreshed < 0 || reply.Dropped < 0 ||
+			reply.Inserted+reply.Refreshed+reply.Dropped != reply.Accepted {
+			return activitypool.UpsertResult{}, invalidScriptReply("invalid upsert counters")
+		}
+		result.Inserted += reply.Inserted
+		result.Refreshed += reply.Refreshed
+		result.Dropped += reply.Dropped
+	}
+	return result, nil
+}
+
+func (a *Adapter) prepareUpsertCandidate(
+	ctx context.Context,
+	upstreamID string,
+	batch activitypool.FetchedBatch,
+	candidate proxyDomain.Proxy,
+) (upsertCandidate, bool, error) {
+	if !validCandidateIdentity(candidate) {
+		return upsertCandidate{}, false, nil
+	}
+	expiresAt := proxyDomain.EffectiveExpiry(batch.ObservedAt, candidate.ExpiresAt, 0, batch.ConfiguredTTL)
+	if expiresAt == nil {
+		return upsertCandidate{}, false, nil
+	}
+	usableUntil := expiresAt.Add(-batch.AllocationSafetyMargin)
+	if !usableUntil.After(batch.ObservedAt) || usableUntil.UnixMilli() <= batch.ObservedAt.UnixMilli() {
+		return upsertCandidate{}, false, nil
+	}
+	if candidate.MaxConcurrency < 0 || (candidate.State != "" && !validProxyState(string(candidate.State))) {
+		return upsertCandidate{}, false, activitypool.ErrInvalidBatch
+	}
+	if (candidate.SecretRef == "") != (candidate.CredentialVersion == "") {
+		return upsertCandidate{}, false, activitypool.ErrInvalidBatch
+	}
+
+	password := ""
+	if candidate.SecretRef != "" {
+		value, err := a.credentials.Resolve(ctx, credentials.Reference{
+			SecretRef: candidate.SecretRef, CredentialVersion: candidate.CredentialVersion,
+		})
+		if err != nil {
+			return upsertCandidate{}, false, fmt.Errorf("resolve proxy credential: %w", err)
+		}
+		candidate.Username = value.Username
+		password = value.Password
+	}
+	candidate.SourceUpstream = upstreamID
+	candidate.ExpiresAt = expiresAt
+	candidate.UsableUntil = &usableUntil
+	if candidate.CreatedAt.IsZero() {
+		candidate.CreatedAt = batch.ObservedAt.UTC()
+	}
+	if candidate.State == "" {
+		candidate.State = proxyDomain.StateFetched
+	}
+	uniqueKey := candidate.UniqueKey()
+	if candidate.ID == "" {
+		candidate.ID = stableAdapterProxyID(uniqueKey)
+	}
+	record := proxyRecord{
+		Version: recordVersion, ID: candidate.ID, Scheme: string(candidate.Scheme), Host: candidate.Host,
+		Port: int64(candidate.Port), Username: candidate.Username, Password: password,
+		CredentialVersion: candidate.CredentialVersion, SourceUpstream: upstreamID,
+		CreatedAtMS: candidate.CreatedAt.UnixMilli(), ExpiresAtMS: expiresAt.UnixMilli(),
+		UsableUntilMS: usableUntil.UnixMilli(), LatencyNS: int64(candidate.Latency),
+		MaxConcurrency: candidate.MaxConcurrency, State: string(candidate.State),
+		Tags: cloneTags(candidate.Tags), IndexKeys: a.availableIndexKeys(candidate),
+	}
+	if candidate.LastCheckedAt != nil {
+		record.LastCheckedAtMS = candidate.LastCheckedAt.UnixMilli()
+	}
+	if candidate.LastSuccessAt != nil {
+		record.LastSuccessAtMS = candidate.LastSuccessAt.UnixMilli()
+	}
+	encoded, err := encodeProxyRecord(record)
+	if err != nil {
+		if errors.Is(err, ErrInvalidRecord) {
+			return upsertCandidate{}, false, activitypool.ErrInvalidBatch
+		}
+		return upsertCandidate{}, false, err
+	}
+	return upsertCandidate{
+		ProxyID: candidate.ID, UniqueDigest: digestToken(uniqueKey), Upstream: upstreamID, Record: encoded,
+	}, true, nil
+}
+
+func (a *Adapter) upsertChunk(
+	ctx context.Context,
+	observedAt time.Time,
+	maxSize int,
+	candidates []upsertCandidate,
+) (upsertScriptReply, error) {
+	operationID, err := newOperationID()
+	if err != nil {
+		return upsertScriptReply{}, err
+	}
+	payload, err := json.Marshal(candidates)
+	if err != nil {
+		return upsertScriptReply{}, fmt.Errorf("encode Redis upsert candidates: %w", err)
+	}
+	result, err := runScript(ctx, a.client, upsertScript, []string{
+		a.keys.records, a.keys.unique, a.keys.idkeys, a.keys.expiry, a.keys.available,
+		a.keys.inventory, a.keys.owners, a.keys.ownerExpiry, a.keys.operation(operationID),
+	}, observedAt.UnixMilli(), a.options.CleanupLimit, maxSize, operationTTLMillis(a.options.OperationTTL), string(payload))
+	if err != nil {
+		return upsertScriptReply{}, err
+	}
+	var reply upsertScriptReply
+	if err := decodeScriptResult(result, &reply); err != nil {
+		return upsertScriptReply{}, err
+	}
+	switch reply.Status {
+	case scriptOK:
+		return reply, nil
+	case scriptInvalid:
+		return upsertScriptReply{}, activitypool.ErrInvalidBatch
+	default:
+		return upsertScriptReply{}, invalidScriptReply("unexpected upsert status")
+	}
+}
+
+func (a *Adapter) availableIndexKeys(candidate proxyDomain.Proxy) []string {
+	keys := []string{a.keys.protocol(string(candidate.Scheme))}
+	if region := candidate.Tags["region"]; region != "" {
+		keys = append(keys, a.keys.region(region))
+	}
+	if carrier := candidate.Tags["carrier"]; carrier != "" {
+		keys = append(keys, a.keys.carrier(carrier))
+	}
+	return append(keys, a.keys.upstream(candidate.SourceUpstream))
+}
+
+func proxyRecordEntry(record proxyRecord) activitypool.Entry {
+	createdAt := time.UnixMilli(record.CreatedAtMS).UTC()
+	expiresAt := time.UnixMilli(record.ExpiresAtMS).UTC()
+	usableUntil := time.UnixMilli(record.UsableUntilMS).UTC()
+	proxy := proxyDomain.Proxy{
+		ID: record.ID, Scheme: proxyDomain.Scheme(record.Scheme), Host: record.Host, Port: uint16(record.Port),
+		Username: record.Username, CredentialVersion: record.CredentialVersion,
+		SourceUpstream: record.SourceUpstream, CreatedAt: createdAt, ExpiresAt: &expiresAt,
+		UsableUntil: &usableUntil, Latency: time.Duration(record.LatencyNS),
+		MaxConcurrency: record.MaxConcurrency, State: proxyDomain.State(record.State), Tags: cloneTags(record.Tags),
+	}
+	if record.LastCheckedAtMS > 0 {
+		value := time.UnixMilli(record.LastCheckedAtMS).UTC()
+		proxy.LastCheckedAt = &value
+	}
+	if record.LastSuccessAtMS > 0 {
+		value := time.UnixMilli(record.LastSuccessAtMS).UTC()
+		proxy.LastSuccessAt = &value
+	}
+	return activitypool.Entry{
+		Proxy: proxy, UsableUntil: usableUntil,
+		OwnerWorkerID: record.OwnerWorkerID, State: proxyDomain.State(record.State),
+	}
+}
+
+func validCandidateIdentity(candidate proxyDomain.Proxy) bool {
+	if candidate.Host == "" || candidate.Port == 0 {
+		return false
+	}
+	return validScheme(string(candidate.Scheme))
+}
+
+func stableAdapterProxyID(uniqueKey string) string {
+	digest := sha256.Sum256([]byte(uniqueKey))
+	return "px_" + hex.EncodeToString(digest[:12])
+}
+
+func cloneTags(tags map[string]string) map[string]string {
+	if tags == nil {
+		return nil
+	}
+	cloned := make(map[string]string, len(tags))
+	for key, value := range tags {
+		cloned[key] = value
+	}
+	return cloned
+}
+
+func invalidScriptReply(reason string) error {
+	return errors.Join(extractionDomain.ErrStoreUnavailable, errors.New(reason))
+}

internal/adapters/redisactivity/upsert_integration_test.go

@@ -0,0 +1,230 @@
+//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"},
+	}
+}