proxy-pool/internal/controller/health/scheduler.go

package health

import (
	"crypto/sha256"
	"encoding/binary"
	"errors"
	"sort"
	"strings"
	"time"

	healthDomain "proxy-pool/internal/domain/health"
	proxyDomain "proxy-pool/internal/domain/proxy"
)

var (
	ErrInvalidSchedulePolicy  = errors.New("invalid health schedule policy")
	ErrInvalidScheduleRequest = errors.New("invalid health schedule request")
)

// SchedulePolicy is scoped to one effective upstream Check configuration.
// MaxInFlight is enforced by the eventual shared task store; Planner uses the
// observed count only to ensure one planning pass never exceeds that ceiling.
type SchedulePolicy struct {
	Interval    time.Duration
	Jitter      int
	MaxInFlight int
	Timeout     time.Duration
	MaxAttempts int
}

// Task contracts live in the health domain so every shared store can use the
// same boundary without importing Controller orchestration code.
type Candidate = healthDomain.Candidate
type Priority = healthDomain.Priority
type PlannedTask = healthDomain.PlannedTask

const (
	PriorityFetched   = healthDomain.PriorityFetched
	PrioritySuspect   = healthDomain.PrioritySuspect
	PriorityUnhealthy = healthDomain.PriorityUnhealthy
	PriorityAvailable = healthDomain.PriorityAvailable
)

// Planner is stateless and safe for concurrent callers. Its lack of internal
// queues makes maxInFlight and batch bounds explicit at the storage boundary.
type Planner struct {
	policy SchedulePolicy
}

func NewPlanner(policy SchedulePolicy) (*Planner, error) {
	if policy.Interval <= 0 || policy.Jitter < 0 || policy.Jitter > 100 || policy.MaxInFlight <= 0 ||
		policy.Timeout <= 0 || policy.MaxAttempts <= 0 {
		return nil, ErrInvalidSchedulePolicy
	}
	return &Planner{policy: policy}, nil
}

// NextDue derives a stable, symmetric jitter from the check identity. A
// process restart therefore does not synchronize all proxy checks into one
// burst, while the same proxy/profile remains predictably distributed.
func (planner *Planner) NextDue(checkedAt time.Time, identity string) (time.Time, error) {
	if planner == nil || checkedAt.IsZero() || strings.TrimSpace(identity) != identity || identity == "" {
		return time.Time{}, ErrInvalidScheduleRequest
	}
	base := checkedAt.UTC().Add(planner.policy.Interval)
	if planner.policy.Jitter == 0 {
		return base, nil
	}
	interval := int64(planner.policy.Interval)
	span := (interval/100)*int64(planner.policy.Jitter) + (interval%100)*int64(planner.policy.Jitter)/100
	if span <= 0 {
		return base, nil
	}
	digest := sha256.Sum256([]byte(identity))
	value := binary.BigEndian.Uint64(digest[:8])
	offset := time.Duration(value % uint64(span+1))
	if value&1 == 0 {
		return base.Add(offset), nil
	}
	return base.Add(-offset), nil
}

// Plan returns at most min(maxTasks, MaxInFlight-inFlight) overdue tasks.
// Fetched inventory is prioritized over suspect inventory, followed by normal
// available inventory, so recovery work does not starve first-use validation.
func (planner *Planner) Plan(now time.Time, inFlight, maxTasks int, candidates []Candidate) ([]PlannedTask, error) {
	if planner == nil || now.IsZero() || inFlight < 0 || maxTasks <= 0 {
		return nil, ErrInvalidScheduleRequest
	}
	if inFlight >= planner.policy.MaxInFlight || len(candidates) == 0 {
		return []PlannedTask{}, nil
	}
	limit := planner.policy.MaxInFlight - inFlight
	if maxTasks < limit {
		limit = maxTasks
	}
	eligible := make([]plannedCandidate, 0, min(limit, len(candidates)))
	for _, candidate := range candidates {
		priority, include, err := planner.validateCandidate(candidate)
		if err != nil {
			return nil, err
		}
		if !include || candidate.DueAt.After(now) {
			continue
		}
		eligible = append(eligible, plannedCandidate{candidate: candidate, priority: priority})
	}
	sort.Slice(eligible, func(left, right int) bool {
		if eligible[left].priority != eligible[right].priority {
			return eligible[left].priority < eligible[right].priority
		}
		if !eligible[left].candidate.DueAt.Equal(eligible[right].candidate.DueAt) {
			return eligible[left].candidate.DueAt.Before(eligible[right].candidate.DueAt)
		}
		return healthDomain.CandidateIdentity(eligible[left].candidate) < healthDomain.CandidateIdentity(eligible[right].candidate)
	})
	if len(eligible) > limit {
		eligible = eligible[:limit]
	}
	deadline := now.UTC().Add(planner.policy.Timeout)
	result := make([]PlannedTask, len(eligible))
	for index, item := range eligible {
		result[index] = PlannedTask{
			Candidate: item.candidate, Priority: item.priority, Deadline: deadline, Attempts: planner.policy.MaxAttempts,
		}
	}
	return result, nil
}

type plannedCandidate struct {
	candidate Candidate
	priority  Priority
}

func (planner *Planner) validateCandidate(candidate Candidate) (Priority, bool, error) {
	if strings.TrimSpace(candidate.ProxyID) != candidate.ProxyID || candidate.ProxyID == "" || candidate.DueAt.IsZero() {
		return 0, false, ErrInvalidScheduleRequest
	}
	switch candidate.Level {
	case healthDomain.LevelBasic, healthDomain.LevelEgress:
		if candidate.RoutingName != "" || candidate.TargetURL != "" {
			return 0, false, ErrInvalidScheduleRequest
		}
	case healthDomain.LevelTarget:
		if _, err := healthDomain.NormalizeTargetProfile(healthDomain.TargetProfile{
			RoutingName: candidate.RoutingName, TargetURL: candidate.TargetURL,
		}); err != nil {
			return 0, false, ErrInvalidScheduleRequest
		}
	default:
		return 0, false, ErrInvalidScheduleRequest
	}
	switch candidate.State {
	case proxyDomain.StateFetched:
		return PriorityFetched, true, nil
	case proxyDomain.StateSuspect:
		return PrioritySuspect, true, nil
	case proxyDomain.StateUnhealthy:
		return PriorityUnhealthy, true, nil
	case proxyDomain.StateAvailable:
		return PriorityAvailable, true, nil
	default:
		return 0, false, nil
	}
}