package aws

import (
	"context"
	"errors"
	"fmt"
	"maps"
	"math"
	"math/rand"
	"slices"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	awsmiddleware "github.com/aws/aws-sdk-go-v2/aws/middleware"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/credentials"
	"github.com/aws/aws-sdk-go-v2/service/ec2"
	"github.com/aws/smithy-go"
	smithymiddleware "github.com/aws/smithy-go/middleware"
	"golang.org/x/sync/errgroup"

	"security-center-backend/internal/modules/caasm/collector"
	model "security-center-backend/internal/modules/caasm/model"
)

type AWSCollector struct {
	cfg            aws.Config
	cloudAccountID uint
	region         string
	allowedRegions []string
	domainScope    model.CollectionScopeMode
	reporter       collector.EventReporter
	// install once to avoid duplicate middleware registrations
	apiMiddlewareInstalled bool
	apiCallStatsMu         sync.Mutex
	apiCallStats           map[string]*awsAPICallStats
	executionGateInstalled bool
	rateLimiterInstalled   bool

	throttleEnabled bool
	rateLimiter     *awsRateLimiter
	adaptive        *adaptiveThrottle
	retryConfig     throttleRetryConfig
	maxConcurrency  int
	serviceLimits   map[string]model.ThrottlingServiceSpec
	circuitBreaker  *model.CircuitBreakerConfig
	executionGate   func(context.Context) error
	resumeCursor    collector.ResumeCursor
}

type regionalServiceCollector struct {
	name   string
	budget time.Duration
	fn     func(context.Context, string, *collector.AssetResult) error
}

type throttleRetryConfig struct {
	maxAttempts int
	baseDelay   time.Duration
	maxDelay    time.Duration
}

type awsAPICallStats struct {
	service         string
	operation       string
	region          string
	count           int
	totalDurationMS int64
	minDurationMS   int64
	maxDurationMS   int64
}

type awsRateLimiter struct {
	mu               sync.Mutex
	buckets          map[string]*tokenBucket
	defaultCapacity  float64
	defaultRefillPer float64 // tokens per second
	serviceLimits    map[string]model.ThrottlingServiceSpec
}

type tokenBucket struct {
	tokens     float64
	capacity   float64
	refillRate float64 // tokens per second
	lastRefill time.Time
}

type adaptiveThrottle struct {
	mu               sync.Mutex
	counts           map[string]int
	windowStart      map[string]time.Time
	coolOffUntil     map[string]time.Time
	threshold        int
	window           time.Duration
	coolOff          time.Duration
	maxCoolOffSleep  time.Duration
	maxConcurrency   int
	minConcurrency   int
	lastDrop         map[string]time.Time
	breakerOpenUntil map[string]time.Time
	breakerThreshold int
	breakerCoolOff   time.Duration
}

// NewAWSCollector creates a new collector for a specific AWS account and region.
// It assumes credentials are provided via access key/secret or IAM role.
func NewAWSCollector(ctx context.Context, accessKey, secretKey, region string, cloudAccountID uint, allowedRegions []string) (*AWSCollector, error) {
	var cfg aws.Config
	var err error

	if accessKey != "" && secretKey != "" {
		cfg, err = config.LoadDefaultConfig(ctx,
			config.WithRegion(region),
			config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(accessKey, secretKey, "")),
		)
	} else {
		// Fallback to default chain (env vars, profile, IAM role)
		cfg, err = config.LoadDefaultConfig(ctx, config.WithRegion(region))
	}

	if err != nil {
		return nil, fmt.Errorf("unable to load SDK config: %v", err)
	}

	defaultMaxConcurrency := model.DefaultThrottleMaxConcurrency
	return &AWSCollector{
		cfg:             cfg,
		cloudAccountID:  cloudAccountID,
		region:          region,
		allowedRegions:  allowedRegions,
		domainScope:     model.CollectionScopeAll,
		throttleEnabled: true,
		rateLimiter:     newAWSRateLimiter(model.DefaultThrottleTokenCapacity, model.DefaultThrottleTokenRefillPerSec),
		adaptive:        newAdaptiveThrottleWithConcurrency(model.DefaultThrottleAdaptiveThreshold, time.Duration(model.DefaultThrottleAdaptiveWindowSec)*time.Second, time.Duration(model.DefaultThrottleCoolOffSec)*time.Second, defaultMaxConcurrency, model.DefaultThrottleMinConcurrency),
		maxConcurrency:  defaultMaxConcurrency,
		retryConfig: throttleRetryConfig{
			maxAttempts: 6,
			baseDelay:   500 * time.Millisecond,
			maxDelay:    20 * time.Second,
		},
		serviceLimits: maps.Clone(model.DefaultThrottleServiceLimits()),
	}, nil
}

// NewAWSCollectorWithConfig builds a collector from a preconfigured AWS config (e.g., assumed role).
func NewAWSCollectorWithConfig(cfg aws.Config, region string, cloudAccountID uint, allowedRegions []string) (*AWSCollector, error) {
	if region != "" {
		cfg.Region = region
	}
	defaultMaxConcurrency := model.DefaultThrottleMaxConcurrency
	return &AWSCollector{
		cfg:             cfg,
		cloudAccountID:  cloudAccountID,
		region:          cfg.Region,
		allowedRegions:  allowedRegions,
		domainScope:     model.CollectionScopeAll,
		throttleEnabled: true,
		rateLimiter:     newAWSRateLimiter(model.DefaultThrottleTokenCapacity, model.DefaultThrottleTokenRefillPerSec),
		adaptive:        newAdaptiveThrottleWithConcurrency(model.DefaultThrottleAdaptiveThreshold, time.Duration(model.DefaultThrottleAdaptiveWindowSec)*time.Second, time.Duration(model.DefaultThrottleCoolOffSec)*time.Second, defaultMaxConcurrency, model.DefaultThrottleMinConcurrency),
		maxConcurrency:  defaultMaxConcurrency,
		retryConfig: throttleRetryConfig{
			maxAttempts: 6,
			baseDelay:   500 * time.Millisecond,
			maxDelay:    20 * time.Second,
		},
		serviceLimits: maps.Clone(model.DefaultThrottleServiceLimits()),
	}, nil
}

func (c *AWSCollector) Provider() string {
	return "aws"
}

func (c *AWSCollector) SetEventReporter(reporter collector.EventReporter) {
	c.reporter = reporter
}

func (c *AWSCollector) SetExecutionGate(gate func(context.Context) error) {
	c.executionGate = gate
}

func (c *AWSCollector) SetResumeCursor(cursor *collector.ResumeCursor) {
	if cursor == nil {
		c.resumeCursor = collector.ResumeCursor{}
		return
	}
	c.resumeCursor = cloneResumeCursor(*cursor)
}

func (c *AWSCollector) SetDomainCollectionScope(scope model.CollectionScopeMode) {
	if scope == "" {
		c.domainScope = model.CollectionScopeAll
		return
	}
	c.domainScope = scope
}

func (c *AWSCollector) report(eventType, message string, metadata map[string]interface{}) {
	if c.reporter != nil {
		c.reporter.Report(eventType, message, metadata)
	}
}

func (c *AWSCollector) checkExecutionGate(ctx context.Context) error {
	if ctx != nil {
		if err := ctx.Err(); err != nil {
			return err
		}
	}
	if c.executionGate == nil {
		return nil
	}
	return c.executionGate(ctx)
}

// ensureAPIMiddleware installs a smithy middleware that records every AWS API call
// as a sync job event (success/failure + metadata) so /sync-jobs/:id/events shows
// detailed observability for the collector.
func (c *AWSCollector) ensureAPIMiddleware() {
	if c.apiMiddlewareInstalled {
		return
	}
	c.cfg.APIOptions = append(c.cfg.APIOptions, c.apiCallLoggerMiddleware)
	c.apiMiddlewareInstalled = true
}

func (c *AWSCollector) ensureExecutionGateMiddleware() {
	if c.executionGateInstalled {
		return
	}
	c.cfg.APIOptions = append(c.cfg.APIOptions, c.apiExecutionGateMiddleware)
	c.executionGateInstalled = true
}

func (c *AWSCollector) apiExecutionGateMiddleware(stack *smithymiddleware.Stack) error {
	return stack.Initialize.Add(smithymiddleware.InitializeMiddlewareFunc("CAASMExecutionGate", func(ctx context.Context, in smithymiddleware.InitializeInput, next smithymiddleware.InitializeHandler) (smithymiddleware.InitializeOutput, smithymiddleware.Metadata, error) {
		if err := c.checkExecutionGate(ctx); err != nil {
			return smithymiddleware.InitializeOutput{}, smithymiddleware.Metadata{}, err
		}
		return next.HandleInitialize(ctx, in)
	}), smithymiddleware.Before)
}

func (c *AWSCollector) apiCallLoggerMiddleware(stack *smithymiddleware.Stack) error {
	return stack.Finalize.Add(smithymiddleware.FinalizeMiddlewareFunc("CAASMApiCallLogger", func(ctx context.Context, in smithymiddleware.FinalizeInput, next smithymiddleware.FinalizeHandler) (smithymiddleware.FinalizeOutput, smithymiddleware.Metadata, error) {
		start := time.Now()
		opName := smithymiddleware.GetOperationName(ctx)
		serviceID := awsmiddleware.GetServiceID(ctx)
		throttleServiceKey := c.normalizeThrottleServiceKey(serviceID)
		region := awsmiddleware.GetRegion(ctx)

		out, meta, err := next.HandleFinalize(ctx, in)

		durationMS := time.Since(start).Milliseconds()
		if c.reporter != nil {
			metadata := map[string]interface{}{
				"service":              serviceID,
				"throttle_service_key": throttleServiceKey,
				"operation":            opName,
				"region":               region,
				"duration_ms":          durationMS,
			}
			if err != nil {
				metadata["error"] = err.Error()
				if errorCode := awsAPIErrorCode(err); errorCode != "" {
					metadata["aws_error_code"] = errorCode
				}
				eventType, message := classifyAWSAPIEvent(serviceID, throttleServiceKey, opName, err)
				c.reporter.Report(eventType, message, metadata)
			}
		}
		if err == nil {
			c.recordAPISuccess(serviceID, opName, region, durationMS)
		}

		return out, meta, err
	}), smithymiddleware.Before)
}

func (c *AWSCollector) recordAPISuccess(serviceID, operation, region string, durationMS int64) {
	serviceID = strings.TrimSpace(serviceID)
	operation = strings.TrimSpace(operation)
	region = strings.TrimSpace(region)
	if serviceID == "" && operation == "" && region == "" {
		return
	}
	if durationMS < 0 {
		durationMS = 0
	}

	key := fmt.Sprintf("%s|%s|%s", serviceID, operation, region)
	c.apiCallStatsMu.Lock()
	defer c.apiCallStatsMu.Unlock()

	if c.apiCallStats == nil {
		c.apiCallStats = make(map[string]*awsAPICallStats)
	}
	stats := c.apiCallStats[key]
	if stats == nil {
		stats = &awsAPICallStats{
			service:       serviceID,
			operation:     operation,
			region:        region,
			minDurationMS: durationMS,
		}
		c.apiCallStats[key] = stats
	}
	stats.count++
	stats.totalDurationMS += durationMS
	if durationMS < stats.minDurationMS {
		stats.minDurationMS = durationMS
	}
	if durationMS > stats.maxDurationMS {
		stats.maxDurationMS = durationMS
	}
}

func (c *AWSCollector) flushAPISuccessEvents() {
	c.apiCallStatsMu.Lock()
	stats := make([]awsAPICallStats, 0, len(c.apiCallStats))
	for _, item := range c.apiCallStats {
		if item != nil && item.count > 0 {
			stats = append(stats, *item)
		}
	}
	c.apiCallStats = nil
	c.apiCallStatsMu.Unlock()

	if c.reporter == nil || len(stats) == 0 {
		return
	}

	sort.Slice(stats, func(i, j int) bool {
		if stats[i].service != stats[j].service {
			return stats[i].service < stats[j].service
		}
		if stats[i].region != stats[j].region {
			return stats[i].region < stats[j].region
		}
		return stats[i].operation < stats[j].operation
	})

	for _, item := range stats {
		avgDurationMS := int64(0)
		if item.count > 0 {
			avgDurationMS = item.totalDurationMS / int64(item.count)
		}
		c.reporter.Report("info", "AWS API successes summarized", map[string]interface{}{
			"aggregation":       "aws_api_success",
			"service":           item.service,
			"operation":         item.operation,
			"region":            item.region,
			"success_count":     item.count,
			"total_duration_ms": item.totalDurationMS,
			"avg_duration_ms":   avgDurationMS,
			"min_duration_ms":   item.minDurationMS,
			"max_duration_ms":   item.maxDurationMS,
		})
	}
}

func awsAPIErrorCode(err error) string {
	var apiErr smithy.APIError
	if !errors.As(err, &apiErr) {
		return ""
	}
	return strings.TrimSpace(apiErr.ErrorCode())
}

func classifyAWSAPIEvent(serviceID, throttleServiceKey, operation string, err error) (string, string) {
	if err == nil {
		return "info", fmt.Sprintf("AWS %s.%s success", serviceID, operation)
	}
	if eventType, message, ok := classifyS3MissingConfigurationEvent(throttleServiceKey, operation, awsAPIErrorCode(err)); ok {
		return eventType, message
	}
	return "error", fmt.Sprintf("AWS %s.%s failed", serviceID, operation)
}

func classifyAWSHelperEvent(serviceKey, operation string, err error) (string, string, bool) {
	errorCode := awsAPIErrorCode(err)
	switch {
	case serviceKey == "route53" && operation == "ListTagsForResource" && strings.EqualFold(errorCode, "NoSuchHostedZone"):
		return "info", "Route53 hosted zone disappeared before tag retrieval", true
	case serviceKey == "cloudfront" && operation == "GetDistribution" && strings.EqualFold(errorCode, "NoSuchDistribution"):
		return "info", "CloudFront distribution disappeared before tag retrieval", true
	case serviceKey == "cloudfront" && operation == "ListTagsForResource" && strings.EqualFold(errorCode, "NoSuchDistribution"):
		return "info", "CloudFront distribution disappeared before tag retrieval", true
	case serviceKey == "elasticloadbalancing" && operation == "DescribeTags" && strings.EqualFold(errorCode, "LoadBalancerNotFound"):
		return "info", "ELB load balancer disappeared before tag retrieval", true
	case serviceKey == "rds" && operation == "ListTagsForResource" && strings.EqualFold(errorCode, "DBInstanceNotFound"):
		return "info", "RDS instance disappeared before tag retrieval", true
	default:
		return "", "", false
	}
}

func classifyS3MissingConfigurationEvent(serviceKey, operation, errorCode string) (string, string, bool) {
	if serviceKey != "s3" {
		return "", "", false
	}

	switch {
	case operation == "GetPublicAccessBlock" && strings.EqualFold(errorCode, "NoSuchPublicAccessBlockConfiguration"):
		return "info", "S3 bucket public access block not configured", true
	case operation == "GetBucketTagging" && strings.EqualFold(errorCode, "NoSuchTagSet"):
		return "info", "S3 bucket tagging not configured", true
	case operation == "GetBucketPolicyStatus" && (strings.EqualFold(errorCode, "NoSuchBucketPolicy") || strings.EqualFold(errorCode, "NoSuchBucketPolicyStatus")):
		return "info", "S3 bucket policy not configured", true
	case operation == "GetBucketWebsite" && strings.EqualFold(errorCode, "NoSuchWebsiteConfiguration"):
		return "info", "S3 bucket website hosting not configured", true
	default:
		return "", "", false
	}
}

func (c *AWSCollector) ensureRateLimitMiddleware() {
	if c.rateLimiterInstalled || !c.throttleEnabled {
		return
	}
	c.cfg.APIOptions = append(c.cfg.APIOptions, c.apiRateLimitMiddleware)
	c.rateLimiterInstalled = true
}

func (c *AWSCollector) apiRateLimitMiddleware(stack *smithymiddleware.Stack) error {
	// Token bucket gating in Initialize
	if err := stack.Initialize.Add(smithymiddleware.InitializeMiddlewareFunc("CAASMApiRateLimiter", func(ctx context.Context, in smithymiddleware.InitializeInput, next smithymiddleware.InitializeHandler) (smithymiddleware.InitializeOutput, smithymiddleware.Metadata, error) {
		if !c.throttleEnabled || c.rateLimiter == nil {
			return next.HandleInitialize(ctx, in)
		}
		serviceID := awsmiddleware.GetServiceID(ctx)
		region := awsmiddleware.GetRegion(ctx)
		key := c.rateLimiterKey(serviceID, region)

		if c.adaptive != nil {
			if delay := c.adaptive.coolOffDelay(key); delay > 0 {
				select {
				case <-time.After(delay):
				case <-ctx.Done():
					return smithymiddleware.InitializeOutput{}, smithymiddleware.Metadata{}, ctx.Err()
				}
			}
		}

		if err := c.rateLimiter.Wait(ctx, key, c.throttlingSpecForKey(key)); err != nil {
			return smithymiddleware.InitializeOutput{}, smithymiddleware.Metadata{}, err
		}
		return next.HandleInitialize(ctx, in)
	}), smithymiddleware.Before); err != nil {
		return err
	}

	// Throttling-aware retry in Finalize
	return stack.Finalize.Add(smithymiddleware.FinalizeMiddlewareFunc("CAASMThrottleRetry", func(ctx context.Context, in smithymiddleware.FinalizeInput, next smithymiddleware.FinalizeHandler) (smithymiddleware.FinalizeOutput, smithymiddleware.Metadata, error) {
		serviceID := awsmiddleware.GetServiceID(ctx)
		region := awsmiddleware.GetRegion(ctx)
		key := c.rateLimiterKey(serviceID, region)

		attempt := 0
		for {
			attempt++
			out, meta, err := next.HandleFinalize(ctx, in)
			if err == nil || !c.shouldRetry(err) || !c.throttleEnabled {
				return out, meta, err
			}

			if attempt >= c.retryConfig.maxAttempts {
				c.handleThrottleEvent(key, serviceID, region, attempt, 0, err, true)
				return out, meta, err
			}

			delay := c.computeBackoff(attempt)
			c.handleThrottleEvent(key, serviceID, region, attempt, delay, err, false)

			select {
			case <-time.After(delay):
				continue
			case <-ctx.Done():
				return out, meta, ctx.Err()
			}
		}
	}), smithymiddleware.Before)
}

func (c *AWSCollector) shouldRetry(err error) bool {
	var apiErr smithy.APIError
	if !errors.As(err, &apiErr) {
		return false
	}
	code := strings.ToLower(apiErr.ErrorCode())
	if code == "throttling" ||
		code == "throttlingexception" ||
		code == "requestlimitexceeded" ||
		code == "slowdown" ||
		code == "toomanyrequestsexception" {
		return true
	}
	if apiErr.ErrorFault() == smithy.FaultServer {
		return true
	}
	return false
}

func (c *AWSCollector) computeBackoff(attempt int) time.Duration {
	if attempt < 1 {
		attempt = 1
	}
	jitter := 0.7 + rand.Float64()*0.6 // 0.7–1.3x
	delay := float64(c.retryConfig.baseDelay) * math.Pow(2, float64(attempt-1)) * jitter
	if delay > float64(c.retryConfig.maxDelay) {
		delay = float64(c.retryConfig.maxDelay)
	}
	return time.Duration(delay)
}

func (c *AWSCollector) handleThrottleEvent(key, serviceID, region string, attempt int, delay time.Duration, err error, terminal bool) {
	circuitState := "closed"
	if c.adaptive != nil {
		circuitState = c.adaptive.record(key)
	}
	if c.reporter != nil {
		metadata := map[string]interface{}{
			"service":               serviceID,
			"region":                region,
			"attempt":               attempt,
			"slowdown_source":       "aws_throttle",
			"throttle_state":        "backing_off",
			"circuit_breaker_state": circuitState,
			"limiter_key":           key,
		}
		if delay > 0 {
			metadata["backoff_ms"] = delay.Milliseconds()
		}
		if spec := c.throttlingSpecForKey(key); spec.TokenCapacity > 0 || spec.TokenRefillPerSec > 0 || spec.MaxConcurrency > 0 || spec.MinConcurrency > 0 {
			metadata["service_limit_applied"] = true
		}
		if err != nil {
			metadata["error"] = err.Error()
		}
		if terminal {
			c.reporter.Report("error", "AWS throttling persisted; giving up", metadata)
		} else {
			c.reporter.Report("warning", "AWS throttled, backing off", metadata)
		}
	}
}

func (c *AWSCollector) rateLimiterKey(serviceID, region string) string {
	svc := c.normalizeThrottleServiceKey(serviceID)
	if region == "" {
		region = c.region
	}
	return fmt.Sprintf("aws:%s:%s", svc, region)
}

func (c *AWSCollector) normalizeThrottleServiceKey(serviceID string) string {
	svc := strings.ToLower(strings.TrimSpace(serviceID))
	svc = strings.ReplaceAll(svc, " ", "")
	switch svc {
	case "apigateway", "apigatewayv2":
		return "apigateway"
	case "elasticloadbalancing", "elasticloadbalancingv2":
		return "elasticloadbalancing"
	case "route53", "route53domains":
		return "route53"
	default:
		return svc
	}
}

// SetThrottleEnabled toggles throttling/rate-limit protections (per-account override).
func (c *AWSCollector) SetThrottleEnabled(enabled bool) {
	c.throttleEnabled = enabled
}

// ConfigureRateLimiter adjusts capacity/refill tokens per second.
func (c *AWSCollector) ConfigureRateLimiter(capacity int, refillPerSec float64) {
	if capacity <= 0 || refillPerSec <= 0 || c.rateLimiter == nil {
		return
	}
	c.rateLimiter.setDefaults(float64(capacity), refillPerSec)
}

func (c *AWSCollector) ConfigureServiceLimits(serviceLimits map[string]model.ThrottlingServiceSpec) {
	if serviceLimits == nil {
		c.serviceLimits = map[string]model.ThrottlingServiceSpec{}
		if c.rateLimiter != nil {
			c.rateLimiter.setServiceLimits(nil)
		}
		return
	}
	c.serviceLimits = maps.Clone(serviceLimits)
	if c.rateLimiter != nil {
		c.rateLimiter.setServiceLimits(c.serviceLimits)
	}
}

func (c *AWSCollector) ConfigureCircuitBreaker(cfg *model.CircuitBreakerConfig) {
	if cfg == nil {
		c.circuitBreaker = nil
		if c.adaptive != nil {
			c.adaptive.configureCircuitBreaker(0, 0)
		}
		return
	}
	copyCfg := *cfg
	c.circuitBreaker = &copyCfg
	if c.adaptive != nil {
		c.adaptive.configureCircuitBreaker(copyCfg.Threshold, time.Duration(copyCfg.CoolOffSec)*time.Second)
	}
}

// ConfigureAdaptive updates adaptive throttling thresholds and concurrency bounds.
func (c *AWSCollector) ConfigureAdaptive(threshold int, window, coolOff time.Duration, maxConc, minConc int) {
	if threshold <= 0 {
		threshold = 3
	}
	if window <= 0 {
		window = time.Minute
	}
	if coolOff <= 0 {
		coolOff = 30 * time.Second
	}
	if maxConc <= 0 {
		maxConc = 5
	}
	if minConc <= 0 || minConc > maxConc {
		minConc = 2
	}

	if c.adaptive == nil {
		c.adaptive = newAdaptiveThrottleWithConcurrency(threshold, window, coolOff, maxConc, minConc)
	} else {
		c.adaptive.configure(threshold, window, coolOff, maxConc, minConc)
	}
	c.maxConcurrency = maxConc
	if c.circuitBreaker != nil {
		c.adaptive.configureCircuitBreaker(c.circuitBreaker.Threshold, time.Duration(c.circuitBreaker.CoolOffSec)*time.Second)
	}
}

func (c *AWSCollector) reportThrottlingConfig() {
	metadata := map[string]interface{}{
		"enabled":              c.throttleEnabled,
		"token_capacity":       0,
		"token_refill_per_sec": 0.0,
		"adaptive_threshold":   0,
		"adaptive_window_sec":  0,
		"cool_off_sec":         0,
		"max_concurrency":      c.maxConcurrency,
		"min_concurrency":      0,
		"service_limit_keys":   []string{},
	}

	if c.rateLimiter != nil {
		metadata["token_capacity"] = int(c.rateLimiter.defaultCapacity)
		metadata["token_refill_per_sec"] = c.rateLimiter.defaultRefillPer
	}
	if c.adaptive != nil {
		metadata["adaptive_threshold"] = c.adaptive.threshold
		metadata["adaptive_window_sec"] = int(c.adaptive.window / time.Second)
		metadata["cool_off_sec"] = int(c.adaptive.coolOff / time.Second)
		metadata["min_concurrency"] = c.adaptive.minConcurrency
	}
	if c.circuitBreaker != nil {
		metadata["circuit_breaker_threshold"] = c.circuitBreaker.Threshold
		metadata["circuit_breaker_cool_off_sec"] = c.circuitBreaker.CoolOffSec
	}
	if len(c.serviceLimits) > 0 {
		keys := make([]string, 0, len(c.serviceLimits))
		for key := range c.serviceLimits {
			keys = append(keys, key)
		}
		slices.Sort(keys)
		metadata["service_limit_keys"] = keys
		metadata["service_limit_count"] = len(keys)
	}

	c.report("info", "AWS throttling config applied", metadata)
}

func newAWSRateLimiter(capacity int, refillPerSec float64) *awsRateLimiter {
	if capacity <= 0 {
		capacity = 20
	}
	if refillPerSec <= 0 {
		refillPerSec = 10
	}
	return &awsRateLimiter{
		buckets:          make(map[string]*tokenBucket),
		defaultCapacity:  float64(capacity),
		defaultRefillPer: refillPerSec,
		serviceLimits:    map[string]model.ThrottlingServiceSpec{},
	}
}

func (r *awsRateLimiter) setDefaults(capacity float64, refillPerSec float64) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if capacity > 0 {
		r.defaultCapacity = capacity
	}
	if refillPerSec > 0 {
		r.defaultRefillPer = refillPerSec
	}
}

func (r *awsRateLimiter) setServiceLimits(serviceLimits map[string]model.ThrottlingServiceSpec) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if serviceLimits == nil {
		r.serviceLimits = map[string]model.ThrottlingServiceSpec{}
		return
	}
	r.serviceLimits = maps.Clone(serviceLimits)
}

func (r *awsRateLimiter) Wait(ctx context.Context, key string, spec model.ThrottlingServiceSpec) error {
	r.mu.Lock()
	bucket, ok := r.buckets[key]
	if !ok {
		capacity := r.defaultCapacity
		refill := r.defaultRefillPer
		if spec.TokenCapacity > 0 {
			capacity = float64(spec.TokenCapacity)
		}
		if spec.TokenRefillPerSec > 0 {
			refill = spec.TokenRefillPerSec
		}
		bucket = &tokenBucket{
			tokens:     capacity,
			capacity:   capacity,
			refillRate: refill,
			lastRefill: time.Now(),
		}
		r.buckets[key] = bucket
	}

	for {
		now := time.Now()
		bucket.refill(now)
		if bucket.tokens >= 1 {
			bucket.tokens -= 1
			r.mu.Unlock()
			return nil
		}
		waitFor := bucket.timeUntilNext()
		if waitFor <= 0 {
			waitFor = 50 * time.Millisecond
		}
		r.mu.Unlock()

		select {
		case <-time.After(waitFor):
			r.mu.Lock()
			continue
		case <-ctx.Done():
			return ctx.Err()
		}
	}
}

func (b *tokenBucket) refill(now time.Time) {
	if b.refillRate <= 0 {
		return
	}
	elapsed := now.Sub(b.lastRefill).Seconds()
	if elapsed <= 0 {
		return
	}
	b.tokens += elapsed * b.refillRate
	if b.tokens > b.capacity {
		b.tokens = b.capacity
	}
	b.lastRefill = now
}

func (b *tokenBucket) timeUntilNext() time.Duration {
	if b.refillRate <= 0 {
		return time.Second
	}
	needed := 1 - b.tokens
	if needed <= 0 {
		return 0
	}
	seconds := needed / b.refillRate
	return time.Duration(seconds * float64(time.Second))
}

func newAdaptiveThrottle(threshold int, window, coolOff time.Duration) *adaptiveThrottle {
	return newAdaptiveThrottleWithConcurrency(threshold, window, coolOff, 5, 2)
}

func newAdaptiveThrottleWithConcurrency(threshold int, window, coolOff time.Duration, maxConc, minConc int) *adaptiveThrottle {
	if threshold <= 0 {
		threshold = 3
	}
	if window <= 0 {
		window = time.Minute
	}
	if coolOff <= 0 {
		coolOff = 30 * time.Second
	}
	if maxConc <= 0 {
		maxConc = 5
	}
	if minConc <= 0 || minConc > maxConc {
		minConc = 2
	}
	return &adaptiveThrottle{
		counts:           make(map[string]int),
		windowStart:      make(map[string]time.Time),
		coolOffUntil:     make(map[string]time.Time),
		threshold:        threshold,
		window:           window,
		coolOff:          coolOff,
		maxCoolOffSleep:  5 * time.Second,
		maxConcurrency:   maxConc,
		minConcurrency:   minConc,
		lastDrop:         make(map[string]time.Time),
		breakerOpenUntil: make(map[string]time.Time),
	}
}

func (a *adaptiveThrottle) record(key string) string {
	now := time.Now()
	a.mu.Lock()
	defer a.mu.Unlock()

	start := a.windowStart[key]
	if start.IsZero() || now.Sub(start) > a.window {
		a.windowStart[key] = now
		a.counts[key] = 0
	}
	a.counts[key]++
	state := "cooling_off"

	if a.counts[key] >= a.threshold {
		a.coolOffUntil[key] = now.Add(a.coolOff)
		a.lastDrop[key] = now
		a.counts[key] = 0
		a.windowStart[key] = now
	}
	if a.breakerThreshold > 0 && a.counts[key] >= a.breakerThreshold {
		a.breakerOpenUntil[key] = now.Add(a.breakerCoolOff)
		a.counts[key] = 0
		a.windowStart[key] = now
		state = "open"
	}
	return state
}

func (a *adaptiveThrottle) coolOffDelay(key string) time.Duration {
	a.mu.Lock()
	defer a.mu.Unlock()
	until, ok := a.coolOffUntil[key]
	now := time.Now()
	if breakerUntil, breakerOpen := a.breakerOpenUntil[key]; breakerOpen {
		if now.After(breakerUntil) {
			delete(a.breakerOpenUntil, key)
		} else {
			delay := time.Until(breakerUntil)
			if delay > a.maxCoolOffSleep {
				return a.maxCoolOffSleep
			}
			return delay
		}
	}
	if !ok {
		return 0
	}
	if now.After(until) {
		delete(a.coolOffUntil, key)
		return 0
	}
	delay := time.Until(until)
	if delay > a.maxCoolOffSleep {
		return a.maxCoolOffSleep
	}
	return delay
}

// currentConcurrency returns the allowed concurrency based on cool-off state.
func (a *adaptiveThrottle) currentConcurrency(key string) int {
	a.mu.Lock()
	defer a.mu.Unlock()
	if until, ok := a.breakerOpenUntil[key]; ok && time.Now().Before(until) {
		return 1
	}
	if until, ok := a.coolOffUntil[key]; ok && time.Now().Before(until) {
		if a.minConcurrency > 0 {
			return a.minConcurrency
		}
		return 1
	}
	if a.maxConcurrency > 0 {
		return a.maxConcurrency
	}
	return 5
}

func (a *adaptiveThrottle) configure(threshold int, window, coolOff time.Duration, maxConc, minConc int) {
	if threshold <= 0 {
		threshold = 3
	}
	if window <= 0 {
		window = time.Minute
	}
	if coolOff <= 0 {
		coolOff = 30 * time.Second
	}
	if maxConc <= 0 {
		maxConc = 5
	}
	if minConc <= 0 || minConc > maxConc {
		minConc = 2
	}
	a.mu.Lock()
	a.threshold = threshold
	a.window = window
	a.coolOff = coolOff
	a.maxConcurrency = maxConc
	a.minConcurrency = minConc
	a.mu.Unlock()
}

func (a *adaptiveThrottle) configureCircuitBreaker(threshold int, coolOff time.Duration) {
	a.mu.Lock()
	defer a.mu.Unlock()
	a.breakerThreshold = threshold
	a.breakerCoolOff = coolOff
}

// currentConcurrencyLimit computes the concurrency limit for a given throttling key.
func (c *AWSCollector) currentConcurrencyLimit(key string) int {
	if spec := c.throttlingSpecForKey(key); spec.MaxConcurrency > 0 {
		if c.adaptive != nil {
			if lim := c.adaptive.currentConcurrency(key); lim > 0 {
				if lim > spec.MaxConcurrency {
					return spec.MaxConcurrency
				}
				if spec.MinConcurrency > 0 && lim < spec.MinConcurrency {
					return spec.MinConcurrency
				}
				return lim
			}
		}
		return spec.MaxConcurrency
	}
	if c.adaptive != nil {
		if lim := c.adaptive.currentConcurrency(key); lim > 0 {
			return lim
		}
	}
	if c.maxConcurrency > 0 {
		return c.maxConcurrency
	}
	return 5
}

func (c *AWSCollector) throttlingSpecForKey(key string) model.ThrottlingServiceSpec {
	if len(c.serviceLimits) == 0 {
		return model.ThrottlingServiceSpec{}
	}
	parts := strings.Split(key, ":")
	if len(parts) < 2 {
		return model.ThrottlingServiceSpec{}
	}
	serviceKey := parts[1]
	spec, ok := c.serviceLimits[serviceKey]
	if !ok {
		return model.ThrottlingServiceSpec{}
	}
	return spec
}

type memoryChunkSink struct {
	mu     sync.Mutex
	result *collector.AssetResult
}

func newMemoryChunkSink() *memoryChunkSink {
	return &memoryChunkSink{result: &collector.AssetResult{}}
}

func (s *memoryChunkSink) PersistChunk(ctx context.Context, chunk *collector.AssetChunk) error {
	if s == nil || chunk == nil {
		return nil
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.result == nil {
		s.result = &collector.AssetResult{}
	}
	s.result.Merge(&chunk.Result)
	return nil
}

func (s *memoryChunkSink) Result() *collector.AssetResult {
	if s == nil {
		return &collector.AssetResult{}
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.result == nil {
		s.result = &collector.AssetResult{}
	}
	return s.result
}

func (c *AWSCollector) Collect(ctx context.Context) (*collector.AssetResult, error) {
	sink := newMemoryChunkSink()
	summary, err := c.CollectChunks(ctx, sink)
	if err != nil {
		return nil, err
	}
	result := sink.Result()
	if result == nil {
		result = &collector.AssetResult{}
	}
	if result.CollectedAt.IsZero() {
		result.CollectedAt = summary.CollectedAt
	}
	c.normalizeCollectedResult(result, result.CollectedAt)
	return result, nil
}

func (c *AWSCollector) CollectChunks(ctx context.Context, sink collector.ChunkSink) (*collector.CollectionSummary, error) {
	if sink == nil {
		return nil, errors.New("chunk sink is required")
	}
	c.ensureExecutionGateMiddleware()
	c.ensureAPIMiddleware()
	c.ensureRateLimitMiddleware()
	defer c.flushAPISuccessEvents()
	if err := c.checkExecutionGate(ctx); err != nil {
		return nil, err
	}

	collectedAt := time.Now()
	summary := &collector.CollectionSummary{
		CollectedAt: collectedAt,
	}
	cursor := cloneResumeCursor(c.resumeCursor)
	if cursor.WindowStartedAt == "" {
		cursor.WindowStartedAt = collectedAt.UTC().Format(time.RFC3339Nano)
	}
	sequence := 0
	var chunkMu sync.Mutex
	c.reportThrottlingConfig()
	emitChunk := func(scope, target string, checkpoint collector.ChunkCheckpoint, result *collector.AssetResult) error {
		chunkMu.Lock()
		defer chunkMu.Unlock()
		if result == nil {
			result = &collector.AssetResult{}
		}
		if result.CollectedAt.IsZero() {
			result.CollectedAt = collectedAt
		}
		c.normalizeCollectedResult(result, collectedAt)
		checkpoint.Eligible = checkpoint.Eligible && len(result.PartialFailures) == 0

		nextCursor := cloneResumeCursor(cursor)
		advanceResumeCursor(&nextCursor, checkpoint)
		meta := map[string]interface{}{
			"collector":      "aws",
			"chunk_boundary": scope,
			"chunk_target":   target,
			"sequence":       sequence + 1,
			"collected_at":   collectedAt.UTC().Format(time.RFC3339Nano),
			"resume_cursor":  nextCursor,
		}
		if checkpoint.GlobalService != "" {
			meta["global_service"] = checkpoint.GlobalService
		}
		if checkpoint.Region != "" {
			meta["region"] = checkpoint.Region
		}
		if checkpoint.Route53ZoneID != "" {
			meta["route53_zone_id"] = checkpoint.Route53ZoneID
		}
		if checkpoint.Route53Zone != "" {
			meta["route53_zone"] = checkpoint.Route53Zone
		}
		if scope == "global_service_registry" && target == "aws_public_exposure_coverage" {
			meta["coverage_registry"] = awsPublicExposureCoverageMetadata()
		}

		chunk := &collector.AssetChunk{
			Result:     *result,
			Scope:      scope,
			Target:     target,
			Sequence:   sequence + 1,
			Metadata:   meta,
			Checkpoint: checkpoint,
		}
		if err := sink.PersistChunk(ctx, chunk); err != nil {
			return err
		}
		sequence++
		cursor = nextCursor
		summary.ChunkCount++
		summary.PartialFailures = append(summary.PartialFailures, result.PartialFailures...)
		return nil
	}

	c.report("info", "Starting global services collection", nil)
	if !c.hasCompletedGlobal("public_exposure_registry") {
		if err := emitChunk("global_service_registry", "aws_public_exposure_coverage", collector.ChunkCheckpoint{
			GlobalService: "public_exposure_registry",
			Eligible:      true,
		}, &collector.AssetResult{CollectedAt: collectedAt}); err != nil {
			return summary, err
		}
	}
	if route53Summary, err := c.collectRoute53Chunks(ctx, sink); err != nil {
		if route53Summary != nil {
			summary.PartialFailures = append(summary.PartialFailures, route53Summary.PartialFailures...)
			summary.ChunkCount += route53Summary.ChunkCount
		}
		if shouldContinueAfterCollectionError(err) {
			summary.PartialFailures = append(summary.PartialFailures, collector.CollectionIssue{
				Scope:   "global_service",
				Target:  "route53",
				Message: err.Error(),
			})
			c.report("warning", "Route53 collection degraded to partial failure", map[string]interface{}{
				"global_service": "route53",
				"error":          err.Error(),
			})
		} else {
			return summary, err
		}
	} else if route53Summary != nil {
		summary.PartialFailures = append(summary.PartialFailures, route53Summary.PartialFailures...)
		summary.ChunkCount += route53Summary.ChunkCount
	}
	if err := c.checkExecutionGate(ctx); err != nil {
		return summary, err
	}
	if c.hasCompletedGlobal("cloudfront") {
		c.report("info", "Skipping CloudFront due to resume checkpoint", map[string]interface{}{"global_service": "cloudfront"})
	} else {
		if cloudFrontSummary, err := c.collectCloudFrontChunks(ctx, sink); err != nil {
			if cloudFrontSummary != nil {
				summary.PartialFailures = append(summary.PartialFailures, cloudFrontSummary.PartialFailures...)
				summary.ChunkCount += cloudFrontSummary.ChunkCount
			}
			if shouldContinueAfterCollectionError(err) {
				summary.PartialFailures = append(summary.PartialFailures, collector.CollectionIssue{
					Scope:   "global_service",
					Target:  "cloudfront",
					Message: err.Error(),
				})
				c.report("warning", "CloudFront collection degraded to partial failure", map[string]interface{}{
					"global_service": "cloudfront",
					"error":          err.Error(),
				})
			} else {
				return summary, err
			}
		} else if cloudFrontSummary != nil {
			summary.PartialFailures = append(summary.PartialFailures, cloudFrontSummary.PartialFailures...)
			summary.ChunkCount += cloudFrontSummary.ChunkCount
		}
	}
	if err := c.checkExecutionGate(ctx); err != nil {
		return summary, err
	}
	if c.hasCompletedGlobal("globalaccelerator") {
		c.report("info", "Skipping Global Accelerator due to resume checkpoint", map[string]interface{}{"global_service": "globalaccelerator"})
	} else {
		if globalAcceleratorSummary, err := c.collectGlobalAcceleratorChunks(ctx, sink); err != nil {
			if globalAcceleratorSummary != nil {
				summary.PartialFailures = append(summary.PartialFailures, globalAcceleratorSummary.PartialFailures...)
				summary.ChunkCount += globalAcceleratorSummary.ChunkCount
			}
			if shouldContinueAfterCollectionError(err) {
				summary.PartialFailures = append(summary.PartialFailures, collector.CollectionIssue{
					Scope:   "global_service",
					Target:  "globalaccelerator",
					Message: err.Error(),
				})
				c.report("warning", "Global Accelerator collection degraded to partial failure", map[string]interface{}{
					"global_service": "globalaccelerator",
					"error":          err.Error(),
				})
			} else {
				return summary, err
			}
		} else if globalAcceleratorSummary != nil {
			summary.PartialFailures = append(summary.PartialFailures, globalAcceleratorSummary.PartialFailures...)
			summary.ChunkCount += globalAcceleratorSummary.ChunkCount
		}
	}
	if err := c.checkExecutionGate(ctx); err != nil {
		return summary, err
	}
	if c.hasCompletedGlobal("s3") {
		c.report("info", "Skipping S3 due to resume checkpoint", map[string]interface{}{"global_service": "s3"})
	} else {
		if s3Summary, err := c.collectS3Chunks(ctx, sink); err != nil {
			if s3Summary != nil {
				summary.PartialFailures = append(summary.PartialFailures, s3Summary.PartialFailures...)
				summary.ChunkCount += s3Summary.ChunkCount
			}
			if shouldContinueAfterCollectionError(err) {
				summary.PartialFailures = append(summary.PartialFailures, collector.CollectionIssue{
					Scope:   "global_service",
					Target:  "s3",
					Message: err.Error(),
				})
				c.report("warning", "S3 collection degraded to partial failure", map[string]interface{}{
					"global_service": "s3",
					"error":          err.Error(),
				})
			} else {
				return summary, err
			}
		} else if s3Summary != nil {
			summary.PartialFailures = append(summary.PartialFailures, s3Summary.PartialFailures...)
			summary.ChunkCount += s3Summary.ChunkCount
		}
	}

	if err := c.checkExecutionGate(ctx); err != nil {
		return summary, err
	}
	regions, err := c.getEnabledRegions(ctx)
	if err != nil {
		if shouldContinueAfterCollectionError(err) {
			summary.PartialFailures = append(summary.PartialFailures, collector.CollectionIssue{
				Scope:   "global",
				Target:  "ec2:describe_regions",
				Message: fmt.Sprintf("failed to list regions: %v", err),
			})
			c.report("warning", "Failed to list enabled regions; continuing with global inventory only", map[string]interface{}{
				"service": "EC2",
				"error":   err.Error(),
			})
			regions = nil
		} else {
			return summary, fmt.Errorf("failed to list regions: %w", err)
		}
	}

	c.report("info", fmt.Sprintf("Collecting from regions: %v", regions), nil)
	limit := c.maxConcurrency
	if limit <= 0 {
		limit = 5
	}

	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(limit)

	for _, region := range regions {
		region := region
		if c.hasCompletedRegion(region) {
			c.report("info", "Skipping region due to resume checkpoint", map[string]interface{}{"region": region})
			continue
		}
		if err := c.checkExecutionGate(ctx); err != nil {
			return summary, err
		}
		if !hasTimeBudget(ctx, 2*time.Minute) {
			budgetErr := c.lowTimeBudgetError(ctx, region, "region", 2*time.Minute)
			summary.PartialFailures = append(summary.PartialFailures, collector.CollectionIssue{
				Scope:   "region",
				Target:  region,
				Message: budgetErr.Error(),
			})
			c.report("warning", "Stopping collection early due to low time budget", map[string]interface{}{
				"remaining_ms":    timeRemainingMillis(ctx),
				"slowdown_source": "configured_budget",
				"budget_state":    "low_time_budget",
				"region":          region,
			})
			break
		}
		g.Go(func() error {
			c.report("info", fmt.Sprintf("Starting collection for region %s", region), map[string]interface{}{"region": region})
			regionCompleted, err := c.collectRegionalServiceChunks(ctx, region, func(serviceName string, checkpoint collector.ChunkCheckpoint, result *collector.AssetResult) error {
				scope := "region_service"
				target := fmt.Sprintf("%s/%s", region, normalizeRegionalServiceTarget(serviceName))
				return emitChunk(scope, target, checkpoint, result)
			})
			if err != nil {
				if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
					return err
				}
				c.report("error", fmt.Sprintf("Error collecting from region %s: %v", region, err), map[string]interface{}{"region": region})
				return err
			}
			if !regionCompleted {
				c.report("warning", fmt.Sprintf("Region %s ended before completion; checkpoint not advanced", region), map[string]interface{}{
					"region":          region,
					"slowdown_source": "configured_budget",
					"budget_state":    "low_time_budget",
				})
			}
			c.report("info", fmt.Sprintf("Finished collection for region %s", region), map[string]interface{}{"region": region})
			return nil
		})
	}

	if err := g.Wait(); err != nil {
		return summary, err
	}

	return summary, nil
}

func shouldContinueAfterCollectionError(err error) bool {
	if err == nil {
		return false
	}
	return !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded)
}

func appendAWSPartialFailure(result *collector.AssetResult, scope, target, message string) {
	if result == nil || message == "" {
		return
	}
	result.PartialFailures = append(result.PartialFailures, collector.CollectionIssue{
		Scope:   scope,
		Target:  target,
		Message: message,
	})
}

func cloneResumeCursor(cursor collector.ResumeCursor) collector.ResumeCursor {
	zoneProgress := make(map[string]collector.Route53ZoneProgress, len(cursor.Route53ZoneProgress))
	for zoneID, progress := range cursor.Route53ZoneProgress {
		zoneProgress[zoneID] = progress
	}
	return collector.ResumeCursor{
		WindowStartedAt:       cursor.WindowStartedAt,
		CompletedGlobal:       append([]string(nil), cursor.CompletedGlobal...),
		CompletedRegions:      append([]string(nil), cursor.CompletedRegions...),
		Route53CompletedZones: append([]string(nil), cursor.Route53CompletedZones...),
		Route53ZoneProgress:   zoneProgress,
	}
}

func appendUniqueString(values []string, next string) []string {
	next = strings.TrimSpace(next)
	if next == "" || slices.Contains(values, next) {
		return values
	}
	return append(values, next)
}

func advanceResumeCursor(cursor *collector.ResumeCursor, checkpoint collector.ChunkCheckpoint) {
	if cursor == nil || !checkpoint.Eligible {
		return
	}
	if checkpoint.GlobalService != "" {
		cursor.CompletedGlobal = appendUniqueString(cursor.CompletedGlobal, checkpoint.GlobalService)
	}
	if checkpoint.Region != "" {
		cursor.CompletedRegions = appendUniqueString(cursor.CompletedRegions, checkpoint.Region)
	}
	if checkpoint.Route53ZoneID != "" {
		if cursor.Route53ZoneProgress == nil {
			cursor.Route53ZoneProgress = make(map[string]collector.Route53ZoneProgress)
		}
		if checkpoint.Route53ZoneComplete {
			cursor.Route53CompletedZones = appendUniqueString(cursor.Route53CompletedZones, checkpoint.Route53ZoneID)
			delete(cursor.Route53ZoneProgress, checkpoint.Route53ZoneID)
			return
		}
		if checkpoint.Route53NextRecordName != "" && checkpoint.Route53NextRecordType != "" {
			cursor.Route53ZoneProgress[checkpoint.Route53ZoneID] = collector.Route53ZoneProgress{
				ZoneID:               checkpoint.Route53ZoneID,
				ZoneName:             checkpoint.Route53Zone,
				NextRecordName:       checkpoint.Route53NextRecordName,
				NextRecordType:       checkpoint.Route53NextRecordType,
				NextRecordIdentifier: checkpoint.Route53NextRecordIdentifier,
			}
		}
	}
}

func (c *AWSCollector) hasCompletedGlobal(service string) bool {
	return slices.Contains(c.resumeCursor.CompletedGlobal, service)
}

func (c *AWSCollector) hasCompletedRegion(region string) bool {
	return slices.Contains(c.resumeCursor.CompletedRegions, region)
}

func (c *AWSCollector) hasCompletedRoute53Zone(zoneID string) bool {
	return slices.Contains(c.resumeCursor.Route53CompletedZones, zoneID)
}

func (c *AWSCollector) route53ZoneProgress(zoneID string) (collector.Route53ZoneProgress, bool) {
	if c.resumeCursor.Route53ZoneProgress == nil {
		return collector.Route53ZoneProgress{}, false
	}
	progress, ok := c.resumeCursor.Route53ZoneProgress[zoneID]
	return progress, ok
}

func (c *AWSCollector) normalizeCollectedResult(result *collector.AssetResult, collectedAt time.Time) {
	if result == nil {
		return
	}
	if result.CollectedAt.IsZero() {
		result.CollectedAt = collectedAt
	}
	for i := range result.DomainAssets {
		d := &result.DomainAssets[i]
		if d.DomainClass == "" {
			d.DomainClass = model.DomainClassUnknown
		}
		if d.ZoneType == "" {
			d.ZoneType = "public"
		}
		if d.Status == "" {
			d.Status = "active"
		}
		if d.LastSyncedAt == nil {
			ts := result.CollectedAt
			d.LastSyncedAt = &ts
		}
		if d.FirstSeenAt == nil {
			ts := result.CollectedAt
			d.FirstSeenAt = &ts
		}
		if d.LastSeenAt == nil {
			ts := result.CollectedAt
			d.LastSeenAt = &ts
		}
	}
}

func (c *AWSCollector) getEnabledRegions(ctx context.Context) ([]string, error) {
	// Use the default config to list regions
	svc := ec2.NewFromConfig(c.cfg)

	input := &ec2.DescribeRegionsInput{
		AllRegions: aws.Bool(false), // Only enabled regions
	}

	resp, err := svc.DescribeRegions(ctx, input)
	if err != nil {
		return nil, err
	}

	var regions []string
	for _, r := range resp.Regions {
		if r.RegionName != nil {
			regionName := *r.RegionName
			// If allowedRegions is set, filter
			if len(c.allowedRegions) > 0 {
				allowed := false
				for _, ar := range c.allowedRegions {
					if ar == regionName {
						allowed = true
						break
					}
				}
				if !allowed {
					continue
				}
			}
			regions = append(regions, regionName)
		}
	}

	return regions, nil
}

func (c *AWSCollector) regionalServicePlan() []regionalServiceCollector {
	return []regionalServiceCollector{
		{name: "EC2", budget: 90 * time.Second, fn: c.collectEC2},
		{name: "ELB", budget: 90 * time.Second, fn: c.collectELB},
		{name: "Classic ELB", budget: 45 * time.Second, fn: c.collectClassicELB},
		{name: "Lightsail", budget: 45 * time.Second, fn: c.collectLightsail},
		{name: "RDS", budget: 60 * time.Second, fn: c.collectRDS},
		{name: "VPC", budget: 60 * time.Second, fn: c.collectVPC},
		{name: "API Gateway", budget: 45 * time.Second, fn: c.collectAPIGateway},
		{name: "ECS", budget: 45 * time.Second, fn: c.collectECS},
		{name: "EKS", budget: 30 * time.Second, fn: c.collectEKS},
		{name: "ElastiCache", budget: 30 * time.Second, fn: c.collectElastiCache},
		{name: "Redshift", budget: 20 * time.Second, fn: c.collectRedshift},
		{name: "App Runner", budget: 30 * time.Second, fn: c.collectAppRunner},
		{name: "OpenSearch", budget: 30 * time.Second, fn: c.collectOpenSearch},
		{name: "Lambda Function URL", budget: 30 * time.Second, fn: c.collectLambdaFunctionURLs},
		{name: "Transfer Family", budget: 30 * time.Second, fn: c.collectTransferFamily},
		{name: "Elastic Beanstalk", budget: 30 * time.Second, fn: c.collectElasticBeanstalk},
		{name: "AppSync", budget: 30 * time.Second, fn: c.collectAppSync},
		{name: "MSK", budget: 30 * time.Second, fn: c.collectMSK},
		{name: "Cognito Hosted Domain", budget: 30 * time.Second, fn: c.collectCognitoHostedDomains},
	}
}

func normalizeRegionalServiceTarget(service string) string {
	svc := strings.ToLower(strings.TrimSpace(service))
	svc = strings.ReplaceAll(svc, " ", "_")
	svc = strings.ReplaceAll(svc, "-", "_")
	return svc
}

func (c *AWSCollector) collectRegionalServices(ctx context.Context, region string, result *collector.AssetResult) error {
	for _, spec := range c.regionalServicePlan() {
		serviceKey := normalizeRegionalServiceTarget(spec.name)
		if err := c.checkExecutionGate(ctx); err != nil {
			return err
		}
		if !hasTimeBudget(ctx, spec.budget) {
			budgetErr := c.lowTimeBudgetError(ctx, region, serviceKey, spec.budget)
			appendAWSPartialFailure(
				result,
				"service",
				fmt.Sprintf("%s:%s", region, serviceKey),
				fmt.Sprintf("failed to collect %s: %v", spec.name, budgetErr),
			)
			c.report("warning", "Stopping region early (time budget)", map[string]interface{}{
				"region":          region,
				"remaining_ms":    timeRemainingMillis(ctx),
				"slowdown_source": "configured_budget",
				"budget_state":    "low_time_budget",
				"service":         serviceKey,
			})
			return budgetErr
		}
		if spec.fn == nil {
			continue
		}
		if err := spec.fn(ctx, region, result); err != nil {
			return fmt.Errorf("failed to collect %s: %v", spec.name, err)
		}
	}
	return nil
}

func (c *AWSCollector) collectRegionalServiceChunks(
	ctx context.Context,
	region string,
	emit func(serviceName string, checkpoint collector.ChunkCheckpoint, result *collector.AssetResult) error,
) (bool, error) {
	return c.collectRegionalServiceChunksWithPlan(ctx, region, c.regionalServicePlan(), emit)
}

func (c *AWSCollector) collectRegionalServiceChunksWithPlan(
	ctx context.Context,
	region string,
	plan []regionalServiceCollector,
	emit func(serviceName string, checkpoint collector.ChunkCheckpoint, result *collector.AssetResult) error,
) (bool, error) {
	recordServiceIssue := func(service string, result *collector.AssetResult, err error) {
		if err == nil {
			return
		}
		message := fmt.Sprintf("failed to collect %s: %v", service, err)
		appendAWSPartialFailure(result, "service", fmt.Sprintf("%s:%s", region, normalizeRegionalServiceTarget(service)), message)
		c.report("warning", fmt.Sprintf("Partial failure collecting %s in %s", service, region), map[string]interface{}{
			"region":  region,
			"service": service,
			"error":   err.Error(),
		})
	}

	if len(plan) == 0 {
		return true, nil
	}

	completedAll := true
	regionEligible := true

	for idx, spec := range plan {
		if err := c.checkExecutionGate(ctx); err != nil {
			return false, err
		}
		if !hasTimeBudget(ctx, spec.budget) {
			serviceResult := &collector.AssetResult{}
			serviceKey := normalizeRegionalServiceTarget(spec.name)
			budgetErr := c.lowTimeBudgetError(ctx, region, serviceKey, spec.budget)
			recordServiceIssue(spec.name, serviceResult, budgetErr)
			c.report("warning", "Stopping region early (time budget)", map[string]interface{}{
				"region":          region,
				"remaining_ms":    timeRemainingMillis(ctx),
				"slowdown_source": "configured_budget",
				"budget_state":    "low_time_budget",
				"service":         serviceKey,
			})
			completedAll = false
			regionEligible = false
			if emit != nil {
				if err := emit(spec.name, collector.ChunkCheckpoint{Eligible: false}, serviceResult); err != nil {
					return false, err
				}
			}
			break
		}

		serviceResult := &collector.AssetResult{}
		if spec.fn != nil {
			if err := spec.fn(ctx, region, serviceResult); err != nil {
				if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
					return false, err
				}
				recordServiceIssue(spec.name, serviceResult, err)
			}
		}

		serviceEligible := len(serviceResult.PartialFailures) == 0
		regionEligible = regionEligible && serviceEligible
		checkpoint := collector.ChunkCheckpoint{Eligible: false}
		if idx == len(plan)-1 && completedAll {
			checkpoint.Region = region
			checkpoint.Eligible = regionEligible
		}
		if emit != nil {
			if err := emit(spec.name, checkpoint, serviceResult); err != nil {
				return false, err
			}
		}
	}

	return completedAll, nil
}

// Helper to create base asset
func (c *AWSCollector) newBaseAsset(nativeID, name, assetType, status, region string) model.Asset {
	cid := uint64(c.cloudAccountID)
	now := time.Now()
	return model.Asset{
		CloudAccountID: &cid,
		NativeID:       nativeID,
		Name:           name,
		AssetType:      model.AssetType(assetType),
		Status:         model.AssetStatus(status),
		Region:         region,
		Provider:       "aws",
		FirstSeenAt:    &now,
		LastSeenAt:     &now,
	}
}

// normalizeServiceName standardizes service identifiers for consistent UI mapping.
func (c *AWSCollector) normalizeServiceName(svc string) string {
	svc = strings.ToLower(strings.TrimSpace(svc))
	switch svc {
	case "s3", "s3_bucket", "bucket":
		return "S3"
	case "cloudfront", "cloudfront_distribution", "cdn":
		return "CloudFront"
	case "api_gateway", "apigw", "api_gateway_v2":
		return "API Gateway"
	case "elb", "alb", "nlb", "load_balancer":
		return "ELB"
	case "rds", "aurora", "database":
		return "RDS"
	case "redshift":
		return "Redshift"
	case "elasticache", "redis", "memcached":
		return "ElastiCache"
	case "eks", "eks_cluster":
		return "EKS"
	case "ecs", "ecs_cluster", "ecs_service", "ecs_task":
		return "ECS"
	case "vpc_endpoint", "vpce":
		return "VPC Endpoint"
	case "ec2", "ec2_instance":
		return "EC2"
	case "vpc":
		return "VPC"
	case "route53", "hosted_zone":
		return "Route53"
	case "dynamodb":
		return "DynamoDB"
	case "lambda":
		return "Lambda"
	case "iam":
		return "IAM"
	case "kms":
		return "KMS"
	case "sns":
		return "SNS"
	case "sqs":
		return "SQS"
	default:
		// Capitalize first letter as fallback
		if len(svc) > 0 {
			return strings.ToUpper(svc[:1]) + svc[1:]
		}
		return svc
	}
}

// setService assigns a normalized service label to the asset.
func (c *AWSCollector) setService(asset *model.Asset, svc string) {
	if asset == nil || svc == "" {
		return
	}
	asset.Service = c.normalizeServiceName(svc)
}

// extractEnvironment extracts environment from resource tags
func (c *AWSCollector) extractEnvironment(tags map[string]string) string {
	keys := []string{"Environment", "environment", "Env", "env", "Stage", "stage"}
	for _, k := range keys {
		if v, ok := tags[k]; ok {
			return c.normalizeEnvironment(v)
		}
	}
	return "" // Return empty string if no environment tag found
}

// normalizeEnvironment normalizes environment values to standard levels
func (c *AWSCollector) normalizeEnvironment(env string) string {
	env = strings.ToLower(strings.TrimSpace(env))

	// Exact matches first
	switch env {
	case "prod", "production":
		return "prod"
	case "staging", "stage", "stg":
		return "staging"
	case "dev", "development":
		return "dev"
	case "test", "testing", "qa":
		return "test"
	}

	// Substring matches for compound values
	if strings.Contains(env, "prod") {
		return "prod"
	}
	if strings.Contains(env, "stag") {
		return "staging"
	}
	if strings.Contains(env, "dev") {
		return "dev"
	}
	if strings.Contains(env, "test") || strings.Contains(env, "qa") {
		return "test"
	}

	// Return "other" for unrecognized environments
	return "other"
}

// hasTimeBudget checks if the context deadline has at least threshold remaining.
func hasTimeBudget(ctx context.Context, threshold time.Duration) bool {
	if ctx == nil {
		return true
	}
	deadline, ok := ctx.Deadline()
	if !ok {
		return true
	}
	return time.Until(deadline) > threshold
}

func timeRemainingMillis(ctx context.Context) int64 {
	if ctx == nil {
		return -1
	}
	if deadline, ok := ctx.Deadline(); ok {
		return time.Until(deadline).Milliseconds()
	}
	return -1
}

func (c *AWSCollector) lowTimeBudgetError(ctx context.Context, region, service string, threshold time.Duration) error {
	return fmt.Errorf(
		"incomplete collection: low time budget before %s in region %s; remaining_ms=%d threshold_ms=%d",
		service,
		region,
		timeRemainingMillis(ctx),
		threshold.Milliseconds(),
	)
}

// extractCriticality extracts criticality from resource tags
func (c *AWSCollector) extractCriticality(tags map[string]string) string {
	keys := []string{"Criticality", "criticality", "Priority", "priority", "Importance", "importance", "Tier", "tier"}
	for _, k := range keys {
		if v, ok := tags[k]; ok {
			return c.normalizeCriticality(v)
		}
	}
	return ""
}

// normalizeCriticality normalizes criticality values to standard levels
func (c *AWSCollector) normalizeCriticality(crit string) string {
	crit = strings.ToLower(strings.TrimSpace(crit))

	// Exact matches first
	switch crit {
	case "critical", "p0":
		return "critical"
	case "high", "p1":
		return "high"
	case "medium", "p2", "moderate":
		return "medium"
	case "low", "p3", "minimal":
		return "low"
	}

	// Substring matches for compound values (e.g., "critical-priority")
	if strings.Contains(crit, "critical") {
		return "critical"
	}
	if strings.Contains(crit, "high") {
		return "high"
	}
	if strings.Contains(crit, "medium") || strings.Contains(crit, "moderate") {
		return "medium"
	}
	if strings.Contains(crit, "low") || strings.Contains(crit, "minimal") {
		return "low"
	}

	// Return normalized value if no match
	return crit
}