package cmd

import (
	"context"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/aws/aws-sdk-go-v2/service/ec2"
	"github.com/innerr/ticat/pkg/core/model"

	cpaws "github.com/tidbcloud/controlplane9/internal/aws"
	cp9errors "github.com/tidbcloud/controlplane9/internal/errors"
)

// purgeNukeDefaultWaitRetries mirrors the destroy-path default for aws-nuke waits.
const purgeNukeDefaultWaitRetries = 10

// ponytail: two workers keep broad sweeps under the schedule interval without
// overwhelming the account-global IAM API; raise only with live broad proof.
const purgeMaxConcurrentScopes = 2

// purgeNukeWaitRetries resolves the aws-nuke wait-retry count from operator env
// (CP9_CLEANUP_AWS_NUKE_MAX_WAIT_RETRIES), defaulting to the destroy-path value.
func purgeNukeWaitRetries() int {
	if raw := strings.TrimSpace(os.Getenv("CP9_CLEANUP_AWS_NUKE_MAX_WAIT_RETRIES")); raw != "" {
		if n, err := strconv.Atoi(raw); err == nil && n > 0 {
			return n
		}
	}
	return purgeNukeDefaultWaitRetries
}

// purgeScopeOutcome is the result of destructively cleaning one owner/session scope.
type purgeScopeOutcome struct {
	Deleted   int
	Survivors int
	StaleNoop bool
	// SurvivorResources are the resources still present after cleanup converged,
	// carried up so the operator-facing report can name what blocked the scope
	// instead of only reporting a count.
	SurvivorResources []cpaws.Resource
}

// purgeExecuteScope is the per-scope destructive executor seam. Tests override it
// to exercise the orchestration without AWS; the default implementation composes
// the proven internal/aws + destroy-path cleanup primitives.
var purgeExecuteScope = realPurgeExecuteScope

// purgePlanScope is the per-scope read-only planner seam for exact dry-runs.
// Tests override it so command tests do not need live AWS.
var purgePlanScope = realPurgePlanScope

var (
	purgePlanBuildReaper = purgeScopedReaper
	purgePlanEKSClaims   = planPurgeTrustedParentEKSClaims
)

// runPurgeDestructive runs destructive cleanup for each candidate scope. It is
// fail-closed per candidate: a blocked or failed candidate is recorded and the
// run continues; if any candidate was blocked or left survivors, the command
// exits non-zero at the end (ADR 0002).
func runPurgeDestructive(
	ctx context.Context,
	cc *model.Cli,
	env *model.Env,
	result *purgeResult,
	scopes []cpaws.ExpiredManagedScope,
	opCfg purgeOperatorConfig,
	iamInventory *cpaws.IAMInventory,
) error {
	runCtx, cancel := context.WithTimeout(ctx, purgeMaxSeconds())
	defer cancel()

	workers := make(chan struct{}, purgeMaxConcurrentScopes)
	execute := func(i int, finalAttempt bool) {
		candidate := &result.Candidates[i]
		candidate.Blocked = false
		candidate.BlockReason = ""
		candidate.Deferred = false
		candidate.DeferredReason = ""
		outcome, err := purgeExecuteScope(runCtx, purgeScopeFor(&scopes[i], opCfg), opCfg, iamInventory)
		if err != nil {
			deferred, retry := purgeDependencyDisposition(err, time.Now().UTC())
			switch {
			case deferred:
				candidate.Deferred = true
				candidate.DeferredReason = err.Error()
			case retry && !finalAttempt:
				candidate.retryExpiredDependency = true
				candidate.BlockReason = err.Error()
			default:
				candidate.Blocked = true
				candidate.BlockReason = err.Error()
			}
			return
		}
		candidate.Deleted = outcome.Deleted
		candidate.Survivors = outcome.Survivors
		candidate.StaleNoop = outcome.StaleNoop
		if outcome.Survivors > 0 {
			candidate.Blocked = true
			candidate.SurvivorDetail = survivorDetails(outcome.SurvivorResources)
			candidate.BlockReason = fmt.Sprintf("%d survivor(s) remain after cleanup: %s",
				outcome.Survivors, formatSurvivors(outcome.SurvivorResources))
		}
	}
	var wg sync.WaitGroup
	for i := range scopes {
		candidate := &result.Candidates[i]
		if !candidate.NukeConfigOK {
			candidate.StaleNoop = false
			candidate.Blocked = true
			candidate.BlockReason = "cleanup boundary not expressible: " + candidate.NukeConfigError
			continue
		}
		if scopes[i].VerifiedStale {
			candidate.StaleNoop = true
			continue
		}
		select {
		case workers <- struct{}{}:
		case <-runCtx.Done():
			candidate.Blocked = true
			candidate.BlockReason = "cleanup run deadline exhausted before candidate: " + runCtx.Err().Error()
			continue
		}
		wg.Go(func() {
			defer func() { <-workers }()
			execute(i, false)
		})
	}
	wg.Wait()
	for i := range result.Candidates {
		if !result.Candidates[i].retryExpiredDependency {
			continue
		}
		result.Candidates[i].retryExpiredDependency = false
		execute(i, true)
	}

	var orphanOIDCErr error
	if result.Mode == string(cpaws.PurgeModeBroad) {
		oidc, err := purgeOrphanOIDCSweep(runCtx, opCfg, true)
		result.OrphanEKSProviders = &oidc
		if err != nil {
			orphanOIDCErr = err
			result.OrphanOIDCError = err.Error()
		}
	}

	anyBlocked := orphanOIDCErr != nil
	anyDeferred := false
	for i := range result.Candidates {
		anyBlocked = anyBlocked || result.Candidates[i].Blocked
		anyDeferred = anyDeferred || result.Candidates[i].Deferred
	}
	switch {
	case anyBlocked:
		result.Note = "destructive run completed with blocked candidates (see block_reason)"
	case anyDeferred:
		result.Note = "destructive run completed; live dependencies were deferred for the next scheduled pass"
	default:
		result.Note = "destructive run completed; all candidates reached zero survivors"
	}
	if err := model.Output(cc, env, result); err != nil {
		return err
	}
	if anyBlocked {
		return cp9errors.New(cp9errors.CategoryResource,
			"cp9 purge: one or more candidates or the orphan EKS OIDC sweep were blocked")
	}
	return nil
}

func purgeDependencyDisposition(err error, now time.Time) (deferred, retry bool) {
	var conflict *nukeScopeConflictError
	if !errors.As(err, &conflict) || conflict.session == "" || conflict.session == conflict.wantSession {
		return false, false
	}
	ttl, parseErr := cpaws.ParseCP9TTL(conflict.ttl)
	if parseErr != nil {
		return false, false
	}
	if ttl.After(now) {
		return true, false
	}
	return false, true
}

// realPurgeExecuteScope performs plan -> tag-assurance -> aws-nuke delete ->
// zero-survivor verify for one scope, reusing the proven destroy-path cleanup
// primitives. It is intentionally not exercised by unit tests (it requires live
// AWS and the aws-nuke binary); the orchestration around it is what is tested.
func realPurgeExecuteScope(
	ctx context.Context,
	scope cpaws.Scope,
	opCfg purgeOperatorConfig,
	iamInventory *cpaws.IAMInventory,
) (outcome purgeScopeOutcome, runErr error) {
	scope = scope.Normalize()
	if err := scope.Validate(); err != nil {
		return purgeScopeOutcome{}, err
	}

	req := &cpaws.NukeRunRequest{
		Profile:        opCfg.Profile,
		Region:         opCfg.Region,
		AccountID:      opCfg.AccountID,
		MaxWaitRetries: purgeNukeWaitRetries(),
	}
	if err := purgeMaterializeNukeCredentials(ctx, req); err != nil {
		return purgeScopeOutcome{}, err
	}

	// The command already verified we run as the cleanup role, so the reaper and
	// aws-nuke use the ambient (now frozen) credentials directly.
	//nolint:contextcheck // the scoped reaper manages its own per-region context, matching newScopedReaper
	reaper, err := purgeScopedReaper(opCfg, iamInventory)
	if err != nil {
		return purgeScopeOutcome{}, fmt.Errorf("build scoped reaper: %w", err)
	}

	initial, err := reaper.Reap(ctx, scope, true, nil)
	if err != nil {
		return purgeScopeOutcome{}, fmt.Errorf("initial scan: %w", err)
	}
	var knownVerifier cpaws.KnownResourceVerifier
	if verifier, ok := reaper.Scanner.(cpaws.KnownResourceVerifier); ok {
		knownVerifier = verifier
	}
	validatedKnown, err := validatePurgeKnownResources(
		ctx,
		knownVerifier,
		scope,
		initial.Survivors,
		opCfg.knownResources(),
		opCfg.Region,
		opCfg.knownResourcesPrimaryRegion(),
	)
	if err != nil {
		return purgeScopeOutcome{}, err
	}
	defer func() {
		if runErr != nil || outcome.Survivors != 0 || len(validatedKnown) == 0 {
			return
		}
		deleted, err := ensurePurgeKnownResourcesGone(
			ctx,
			knownVerifier,
			scope,
			reaper.Cleaner,
			validatedKnown,
			purgeMaxPasses(),
			func() { time.Sleep(purgeRetrySleep()) },
		)
		if err != nil {
			outcome = purgeScopeOutcome{}
			runErr = fmt.Errorf("known-resource direct zero verification: %w", err)
			return
		}
		outcome.Deleted += deleted
	}()
	survivors, _, err := filterPostNukeStaleSurvivorsWithWarnings(
		ctx,
		initial.Survivors,
		req,
		"purge-initial-scan",
	)
	if err != nil {
		return purgeScopeOutcome{}, fmt.Errorf("verify initial scan survivors: %w", err)
	}
	survivors, err = mergePurgeKnownResourceSurvivors(ctx, knownVerifier, scope, survivors, validatedKnown)
	if err != nil {
		return purgeScopeOutcome{}, fmt.Errorf("merge known-resource survivors: %w", err)
	}
	if len(survivors) == 0 {
		return purgeScopeOutcome{Deleted: 0, Survivors: 0, StaleNoop: true}, nil
	}

	// Terminate the session's control plane before anything is deleted, including
	// the EKS teardown below: crossplane on that host reconciles the dataplane
	// resources this is about to remove, and EKS deletion is the longest window it
	// has to recreate them. Never fatal -- see purgeTerminateSessionHost.
	purgeTerminateSessionHost(ctx, scope, req)

	// Read and validate the complete EKS role graph before any of this
	// candidate's new trusted-parent tags are applied. Tagging is deferred until
	// the existing VPC/EKS/LB assurance steps also pass, so a graph conflict
	// cannot block the candidate after EKS has already been mutated.
	eksRoleChildren, err := planNukeEKSRoleChildrenForExecution(ctx, scope, req)
	if err != nil {
		return purgeScopeOutcome{}, fmt.Errorf("eks trusted-parent assurance: %w", err)
	}
	survivors = mergePurgeResources(survivors, eksRoleChildren)
	initialCount := len(survivors)

	// Trusted-parent tag assurance before aws-nuke (ADR 0001).
	vpcChildren, err := prepareNukeVPCChildren(ctx, survivors, scope, req)
	if err != nil {
		return purgeScopeOutcome{}, fmt.Errorf("vpc tag assurance: %w", err)
	}
	if _, err := prepareNukeEKSResources(ctx, survivors, scope, req); err != nil {
		return purgeScopeOutcome{}, fmt.Errorf("eks tag assurance: %w", err)
	}
	if _, err := prepareNukeLoadBalancers(ctx, survivors, scope, req); err != nil {
		return purgeScopeOutcome{}, fmt.Errorf("load balancer tag assurance: %w", err)
	}
	if _, err := prepareNukeOrphanEBSVolumes(ctx, survivors, scope, req); err != nil {
		return purgeScopeOutcome{}, fmt.Errorf("ebs volume tag assurance: %w", err)
	}
	if err := tagPlannedNukeEKSRoleChildren(ctx, eksRoleChildren, scope, req); err != nil {
		return purgeScopeOutcome{}, fmt.Errorf("eks trusted-parent tagging: %w", err)
	}
	if _, survivors, err = destroyCleanupEKSResources(ctx, scope, survivors, req); err != nil {
		return purgeScopeOutcome{}, fmt.Errorf("eks direct cleanup: %w", err)
	}

	// aws-nuke does not reliably delete ACTIVE root ACM-PCA certificate
	// authorities, so purge deletes them directly (disable then delete). This
	// moves them to DELETED, which the post-nuke stale check recognizes as gone.
	handledPCA, err := purgeDeleteACMPCACertAuthorities(ctx, survivors, req)
	if err != nil {
		return purgeScopeOutcome{}, fmt.Errorf("acm-pca cleanup: %w", err)
	}
	survivors = withoutHandledPurgeSurvivors(survivors, handledPCA)

	// A CloudFormation stack is only ever deleted here: the reaper's verify pass
	// runs dry, and aws-nuke did not remove session-owned stacks.
	goneStacks, err := destroyCleanupCloudFormationStacks(ctx, reaper, scope, survivors)
	if err != nil {
		return purgeScopeOutcome{}, fmt.Errorf("cloudformation cleanup: %w", err)
	}
	survivors = withoutHandledPurgeSurvivors(survivors, goneStacks)

	// S3 and Secrets Manager are safe to direct-clean up front. Launch templates
	// and IAM must wait until EKS drains: nodegroups can still use launch
	// templates, and cluster service roles remain in use while clusters are
	// ACTIVE/DELETING. Treat both as a direct-cleaned tail so a late launch
	// template survivor cannot pin the aws-nuke convergence loop.
	if _, err := destroyCleanupS3Buckets(ctx, reaper, scope, survivors); err != nil {
		return purgeScopeOutcome{}, fmt.Errorf("s3 direct cleanup: %w", err)
	}
	if _, survivors, err = destroyCleanupSecretsManagerResources(ctx, reaper, scope, survivors); err != nil {
		return purgeScopeOutcome{}, fmt.Errorf("secretsmanager direct cleanup: %w", err)
	}
	if len(survivors) == 0 {
		return purgeScopeOutcome{Deleted: initialCount, Survivors: 0}, nil
	}
	if len(nonDirectTailCleanupResources(survivors)) == 0 {
		if survivors, err = purgeCleanupDirectTail(ctx, reaper, scope, survivors, req); err != nil {
			return purgeScopeOutcome{}, err
		}
		if survivors, err = purgeVerifyFinalSurvivors(ctx, reaper, scope, survivors, req); err != nil {
			return purgeScopeOutcome{}, err
		}
		return purgeScopeOutcome{
			Deleted:           initialCount - len(survivors),
			Survivors:         len(survivors),
			SurvivorResources: survivors,
		}, nil
	}

	// Delete serverless resources aws-nuke intentionally skips in this shared
	// account (DynamoDB / Lambda / SQS / ECR). A real destroy relies on the
	// in-cluster operator to remove them; the purge has no cluster, so they survive
	// every run — and Lambda-in-VPC ENIs also block the VPC teardown.
	if _, err := purgeDeleteServerlessResources(ctx, survivors, req); err != nil {
		return purgeScopeOutcome{}, fmt.Errorf("serverless cleanup: %w", err)
	}

	configPath, err := purgeWriteNukeConfig(scope, opCfg, survivors, vpcChildren)
	if err != nil {
		return purgeScopeOutcome{}, err
	}
	defer func() { _ = os.Remove(configPath) }()
	req.ConfigPath = configPath

	// One aws-nuke delete pass rarely removes everything: dependency-ordered
	// resources (e.g. a VPC behind subnets/ENIs/NAT gateways) need several
	// delete -> verify passes to converge. Loop until zero survivors, the
	// max-seconds deadline, or the pass cap (mirrors the legacy
	// cleanup_until_stable retry loop).
	deadline := time.Now().Add(purgeMaxSeconds())
	var lastRemaining []cpaws.Resource
	deleteVerify := func(pass int) (int, error) {
		// Re-scan every pass: endpoint-service and load-balancer deletion are
		// eventually consistent, and their requester-managed ENIs pin the VPC.
		if _, err := purgeDeleteVPCLoadBalancers(ctx, survivors, req); err != nil {
			fmt.Fprintf(os.Stderr, "purge: vpc load balancer cleanup (pass %d) errored, verifying and retrying: %v\n", pass, err)
		}
		req.Delete = true
		if out, err := destroyNukeRunner.Run(ctx, req); err != nil {
			// aws-nuke exits fatal on a single DependencyViolation (e.g. an ENI
			// still releasing after a load balancer delete). Do not abort the
			// scope — fall through to finalize + verify; a later pass retries once
			// the dependency clears. The final survivor count is the verdict.
			//
			// A listing failure is different in kind: aws-nuke could not enumerate a
			// type, which on its own says nothing about the deletes. Report the types
			// and the exit reason in one line instead of echoing the whole run output
			// on every pass, and leave the verdict where it already lives -- the
			// survivor count. Do not claim the deletes succeeded: this output proves
			// only that every error line was a listing failure.
			if types, listingOnly := cpaws.NukeListingDegradedTypes(out); listingOnly {
				fmt.Fprintf(os.Stderr,
					"purge: aws-nuke delete (pass %d) could not list %s (%v); survivor count decides\n",
					pass, strings.Join(types, ", "), err)
			} else {
				fmt.Fprintf(os.Stderr, "purge: aws-nuke delete (pass %d) errored, verifying and retrying: %v\n", pass, err)
			}
		}
		if _, err := finalizeNukeVPCRootsAfterDelete(ctx, survivors, scope, req); err != nil {
			// A direct DeleteVpc can DependencyViolation while a load balancer's ENIs
			// or security group are still releasing after the LB delete. Don't abort
			// the scope — verify and retry on the next pass once the dependency clears.
			fmt.Fprintf(os.Stderr, "purge: finalize vpc roots (pass %d) errored, verifying and retrying: %v\n", pass, err)
		}
		verify, err := reaper.Reap(ctx, scope, true, nil)
		if err != nil {
			return 0, fmt.Errorf("verify scan (pass %d): %w", pass, err)
		}
		// ACM-PCA deletion can lag behind the initial disable/delete request.
		// Retry the direct cleanup against the latest verified survivors before
		// deciding this pass is still blocked.
		handledPCA, err := purgeDeleteACMPCACertAuthorities(ctx, verify.Survivors, req)
		if err != nil {
			return 0, fmt.Errorf("acm-pca cleanup (pass %d): %w", pass, err)
		}
		verify.Survivors = withoutHandledPurgeSurvivors(verify.Survivors, handledPCA)
		goneStacks, err := destroyCleanupCloudFormationStacks(ctx, reaper, scope, verify.Survivors)
		if err != nil {
			return 0, fmt.Errorf("cloudformation cleanup (pass %d): %w", pass, err)
		}
		verify.Survivors = withoutHandledPurgeSurvivors(verify.Survivors, goneStacks)
		remaining, _, err := filterPostNukeStaleSurvivorsWithWarnings(
			ctx, verify.Survivors, req, fmt.Sprintf("purge-verify-pass-%d", pass))
		if err != nil {
			return 0, fmt.Errorf("verify survivors (pass %d): %w", pass, err)
		}
		lastRemaining = remaining
		return len(nonDirectTailCleanupResources(remaining)), nil
	}
	withinDeadline := func() bool { return ctx.Err() == nil && time.Now().Before(deadline) }
	sleeper := func() { time.Sleep(purgeRetrySleep()) }

	if _, _, err := purgeRetryUntilStable(deleteVerify, purgeMaxPasses(), withinDeadline, sleeper); err != nil {
		return purgeScopeOutcome{}, err
	}
	// The direct tail waits for EKS specifically, not for an empty survivor set.
	// Nodegroups can still use a launch template and cluster service roles stay in
	// use while a cluster is ACTIVE/DELETING, so deleting those early wedges
	// nodegroup and ENI teardown. Waiting on *every* other survivor was too strong:
	// one resource the reaper cannot delete then suppressed the cleanup of every
	// launch template and IAM resource behind it.
	if !purgeEKSTeardownPending(lastRemaining) {
		if lastRemaining, err = purgeCleanupDirectTail(ctx, reaper, scope, lastRemaining, req); err != nil {
			return purgeScopeOutcome{}, err
		}
		if lastRemaining, err = purgeVerifyFinalSurvivors(ctx, reaper, scope, lastRemaining, req); err != nil {
			return purgeScopeOutcome{}, err
		}
	}
	return purgeScopeOutcome{
		Deleted:           initialCount - len(lastRemaining),
		Survivors:         len(lastRemaining),
		SurvivorResources: lastRemaining,
	}, nil
}

func purgeVerifyFinalSurvivors(
	ctx context.Context,
	reaper *cpaws.Reaper,
	scope cpaws.Scope,
	survivors []cpaws.Resource,
	req *cpaws.NukeRunRequest,
) ([]cpaws.Resource, error) {
	// Direct tail cleanup performs fresh Resource Groups Tagging API scans. Those
	// scans can reintroduce stale rows already removed during the delete/verify
	// loop, so apply the same direct-service verification to their result before
	// deciding the final outcome.
	handledPCA, err := purgeDeleteACMPCACertAuthorities(ctx, survivors, req)
	if err != nil {
		return nil, fmt.Errorf("final acm-pca cleanup: %w", err)
	}
	survivors = withoutHandledPurgeSurvivors(survivors, handledPCA)
	goneStacks, err := destroyCleanupCloudFormationStacks(ctx, reaper, scope, survivors)
	if err != nil {
		return nil, fmt.Errorf("final cloudformation cleanup: %w", err)
	}
	survivors = withoutHandledPurgeSurvivors(survivors, goneStacks)
	filtered, _, err := filterPostNukeStaleSurvivorsWithWarnings(
		ctx,
		survivors,
		req,
		"purge-final-iam-verify",
	)
	if err != nil {
		return nil, fmt.Errorf("final survivor verification: %w", err)
	}
	return filtered, nil
}

// purgeEKSTeardownPending reports whether an EKS cluster or nodegroup is still
// present. Those are the resources that genuinely pin the direct tail: a
// nodegroup keeps using its launch template, and a cluster keeps using its
// service role, until both are gone.
func purgeEKSTeardownPending(resources []cpaws.Resource) bool {
	return slices.ContainsFunc(resources, func(resource cpaws.Resource) bool {
		return resource.Type == cpaws.ResourceEKSCluster || resource.Type == cpaws.ResourceEKSNodegroup
	})
}

func nonDirectTailCleanupResources(resources []cpaws.Resource) []cpaws.Resource {
	filtered := make([]cpaws.Resource, 0, len(resources))
	for _, resource := range resources {
		if resource.Type == cpaws.ResourceEC2PlacementGroup ||
			resource.Type == cpaws.ResourceLaunchTemplate ||
			destroyDirectIAMCleanupType(resource.Type) {
			continue
		}
		filtered = append(filtered, resource)
	}
	return filtered
}

func purgeCleanupDirectTail(
	ctx context.Context,
	reaper *cpaws.Reaper,
	scope cpaws.Scope,
	resources []cpaws.Resource,
	req *cpaws.NukeRunRequest,
) ([]cpaws.Resource, error) {
	_, survivors, err := destroyCleanupEC2TailResources(
		ctx, reaper, scope, resources, cpaws.ResourceEC2PlacementGroup)
	if err != nil {
		return nil, fmt.Errorf("placement group direct cleanup: %w", err)
	}
	_, survivors, err = destroyCleanupEC2TailResources(
		ctx, reaper, scope, survivors, cpaws.ResourceLaunchTemplate)
	if err != nil {
		return nil, fmt.Errorf("launch template direct cleanup: %w", err)
	}
	_, survivors, err = destroyCleanupIAMResources(ctx, reaper, scope, survivors, req)
	if err != nil {
		return nil, fmt.Errorf("iam direct cleanup: %w", err)
	}
	return survivors, nil
}

// validatePurgeKnownResources turns the handoff list into a zero-verification
// set only after either the ordinary owner/session scanner or an exact,
// read-only owning-service lookup independently proves every applicable
// resource. The handoff is evidence of what must reach zero; it is never, by
// itself, authority to adopt or delete an ARN.
func validatePurgeKnownResources(
	ctx context.Context,
	verifier cpaws.KnownResourceVerifier,
	scope cpaws.Scope,
	discovered []cpaws.Resource,
	known []cpaws.Resource,
	region string,
	primaryRegion string,
) ([]cpaws.Resource, error) {
	if len(known) == 0 {
		return nil, nil
	}
	discoveredByARN := make(map[string]cpaws.Resource, len(discovered))
	for _, resource := range discovered {
		discoveredByARN[resource.ARN] = resource
	}
	seen := make(map[string]struct{}, len(known))
	validated := make([]cpaws.Resource, 0, len(known))
	for _, expected := range known {
		if expected.Region != "" && expected.Region != region {
			continue
		}
		if expected.Region == "" && region != primaryRegion {
			continue
		}
		if expected.Region == "" {
			expected.Region = primaryRegion
		}
		if _, ok := seen[expected.ARN]; ok {
			continue
		}
		proven, ok := discoveredByARN[expected.ARN]
		if !ok {
			if verifier == nil {
				return nil, fmt.Errorf(
					"known handoff resource %q is not proven by scoped discovery and no direct verifier is configured",
					expected.ARN,
				)
			}
			exists, err := verifier.VerifyKnownResource(ctx, scope, &expected)
			if err != nil {
				return nil, fmt.Errorf("direct ownership proof for known handoff resource %q: %w", expected.ARN, err)
			}
			if !exists {
				// Direct service absence is terminal proof. Preserve the ID so the
				// idempotent direct-zero pass independently observes it as absent.
				seen[expected.ARN] = struct{}{}
				validated = append(validated, expected)
				continue
			}
			seen[expected.ARN] = struct{}{}
			validated = append(validated, expected)
			continue
		}
		if proven.Type != expected.Type {
			return nil, fmt.Errorf(
				"known handoff resource %q type %q conflicts with scoped discovery type %q",
				expected.ARN,
				expected.Type,
				proven.Type,
			)
		}
		seen[expected.ARN] = struct{}{}
		validated = append(validated, proven)
	}
	return validated, nil
}

// ensurePurgeKnownResourcesGone uses direct, idempotent service deletion as
// the terminal existence verifier for IDs whose ownership was already proven
// by validatePurgeKnownResources. A green result therefore covers both the
// final tag scan and every persisted handoff ID.
func ensurePurgeKnownResourcesGone(
	ctx context.Context,
	verifier cpaws.KnownResourceVerifier,
	scope cpaws.Scope,
	cleaner cpaws.Cleaner,
	resources []cpaws.Resource,
	maxAttempts int,
	wait func(),
) (int, error) {
	if len(resources) == 0 {
		return 0, nil
	}
	if verifier == nil {
		return 0, fmt.Errorf("known-resource ownership verifier is not configured")
	}
	if cleaner == nil {
		return 0, fmt.Errorf("known-resource direct verifier is not configured")
	}
	if maxAttempts < 1 {
		maxAttempts = 1
	}
	deletedResources := 0
	for _, resource := range resources {
		deletedOnce := false
		for attempt := 1; attempt <= maxAttempts; attempt++ {
			exists, err := verifier.VerifyKnownResource(ctx, scope, &resource)
			if err != nil {
				return deletedResources, fmt.Errorf("revalidate known resource %s ownership: %w", resource.ARN, err)
			}
			if !exists {
				if deletedOnce {
					deletedResources++
				}
				break
			}
			deleted, err := cleaner.Delete(ctx, &resource)
			if err != nil {
				return deletedResources, fmt.Errorf("verify known resource %s: %w", resource.ARN, err)
			}
			if !deleted {
				return deletedResources, fmt.Errorf(
					"known resource %s was proven present but direct deletion reported it absent",
					resource.ARN,
				)
			}
			deletedOnce = true
			if attempt == maxAttempts {
				return deletedResources, fmt.Errorf(
					"known resource %s still existed after %d direct verification attempts",
					resource.ARN,
					maxAttempts,
				)
			}
			if wait != nil {
				wait()
			}
		}
	}
	return deletedResources, nil
}

func withoutHandledPurgeSurvivors(survivors []cpaws.Resource, handled map[string]struct{}) []cpaws.Resource {
	if len(survivors) == 0 || len(handled) == 0 {
		return survivors
	}
	filtered := make([]cpaws.Resource, 0, len(survivors))
	for _, survivor := range survivors {
		if _, ok := handled[survivor.ARN]; ok {
			continue
		}
		filtered = append(filtered, survivor)
	}
	return filtered
}

// purgeRetryUntilStable runs delete -> verify passes until verify reports zero
// survivors, the pass cap is hit, or the deadline passes. It returns the final
// survivor count and the number of passes run. deleteVerify performs one pass
// and returns the post-pass survivor count; withinDeadline and sleeper are
// injected so the loop is unit-testable without real time.
func purgeRetryUntilStable(
	deleteVerify func(pass int) (int, error),
	maxPasses int,
	withinDeadline func() bool,
	sleeper func(),
) (survivors, passes int, err error) {
	if maxPasses < 1 {
		maxPasses = 1
	}
	last := 0
	for pass := 1; pass <= maxPasses; pass++ {
		if pass > 1 && !withinDeadline() {
			return last, pass - 1, nil
		}
		count, runErr := deleteVerify(pass)
		if runErr != nil {
			return count, pass, runErr
		}
		last = count
		if count == 0 {
			return 0, pass, nil
		}
		if pass < maxPasses {
			sleeper()
		}
	}
	return last, maxPasses, nil
}

func purgeMaxPasses() int {
	if raw := strings.TrimSpace(os.Getenv("CP9_CLEANUP_MAX_PASSES")); raw != "" {
		if n, err := strconv.Atoi(raw); err == nil && n > 0 {
			return n
		}
	}
	return 8
}

func purgeMaxSeconds() time.Duration {
	if raw := strings.TrimSpace(os.Getenv("CP9_CLEANUP_MAX_SECONDS")); raw != "" {
		if n, err := strconv.Atoi(raw); err == nil && n > 0 {
			return time.Duration(n) * time.Second
		}
	}
	return 21600 * time.Second
}

func purgeRetrySleep() time.Duration {
	if raw := strings.TrimSpace(os.Getenv("CP9_CLEANUP_RETRY_SLEEP_SECONDS")); raw != "" {
		if n, err := strconv.Atoi(raw); err == nil && n > 0 {
			return time.Duration(n) * time.Second
		}
	}
	return 20 * time.Second
}

// purgeScopedReaper builds a region-scoped reaper using the ambient cleanup-role
// credentials. It is context-free like newScopedReaper because the reaper manages
// its own per-region context internally.
func purgeScopedReaper(opCfg purgeOperatorConfig, iamInventory *cpaws.IAMInventory) (*cpaws.Reaper, error) {
	return newScopedAssumeRoleReaperWithIAMInventory(opCfg.Profile, "", false, iamInventory)
}

// purgeMaterializeNukeCredentials freezes the cleanup-role credentials into the
// run request so aws-nuke and the tag-assurance helpers use static credentials.
func purgeMaterializeNukeCredentials(ctx context.Context, req *cpaws.NukeRunRequest) error {
	awsCfg, err := cmdLoadAWSConfig(ctx, req.Profile, req.Region)
	if err != nil {
		return fmt.Errorf("load aws-nuke credentials: %w", err)
	}
	if err := cpaws.FreezeConfigCredentials(ctx, &awsCfg); err != nil {
		return fmt.Errorf("freeze aws-nuke credentials: %w", err)
	}
	creds, err := awsCfg.Credentials.Retrieve(ctx)
	if err != nil {
		return fmt.Errorf("retrieve frozen aws-nuke credentials: %w", err)
	}
	if strings.TrimSpace(creds.AccessKeyID) == "" || strings.TrimSpace(creds.SecretAccessKey) == "" {
		return fmt.Errorf("aws-nuke credentials are incomplete")
	}
	req.AccessKeyID = creds.AccessKeyID
	req.SecretAccessKey = creds.SecretAccessKey
	req.SessionToken = creds.SessionToken
	req.Profile = ""
	return nil
}

// filterPurgeDryRunStaleSurvivors applies the same direct-service, read-only
// absence proof used after destructive cleanup. Resource Groups Tagging API can
// retain a DELETED ACM-PCA row for its restoration window and EKS pod identity
// association rows after their owning cluster is gone; neither is a live
// survivor and neither should make an independent exact dry-run report a false
// leak. Live resources remain in the plan.
func filterPurgeDryRunStaleSurvivors(
	ctx context.Context,
	survivors []cpaws.Resource,
	opCfg purgeOperatorConfig,
) ([]cpaws.Resource, error) {
	if len(survivors) == 0 {
		return survivors, nil
	}
	req := &cpaws.NukeRunRequest{
		Profile:   opCfg.Profile,
		Region:    opCfg.Region,
		AccountID: opCfg.AccountID,
	}
	if err := purgeMaterializeNukeCredentials(ctx, req); err != nil {
		return nil, fmt.Errorf("materialize exact dry-run verifier credentials: %w", err)
	}
	filtered, _, err := filterPostNukeStaleSurvivorsWithWarnings(
		ctx,
		survivors,
		req,
		"purge-dry-run-plan",
	)
	if err != nil {
		return nil, fmt.Errorf("verify exact dry-run survivors: %w", err)
	}
	return filtered, nil
}

// realPurgePlanScope performs the exact-mode dry-run cleanup plan without
// mutating tags or deleting resources. It mirrors the first read-only scan and
// trusted-parent EC2 discovery used by the destructive purge path so operators
// can prove coverage before running dry-run=false.
func realPurgePlanScope(
	ctx context.Context,
	scope cpaws.Scope,
	opCfg purgeOperatorConfig,
) (purgeScopePlan, error) {
	scope = scope.Normalize()
	if err := scope.Validate(); err != nil {
		return purgeScopePlan{}, err
	}

	reaper, err := purgePlanBuildReaper(opCfg, nil)
	if err != nil {
		return purgeScopePlan{}, fmt.Errorf("build scoped reaper: %w", err)
	}
	initial, err := reaper.Reap(ctx, scope, true, nil)
	if err != nil {
		return purgeScopePlan{}, fmt.Errorf("initial scan: %w", err)
	}
	var knownVerifier cpaws.KnownResourceVerifier
	if verifier, ok := reaper.Scanner.(cpaws.KnownResourceVerifier); ok {
		knownVerifier = verifier
	}
	validatedKnown, err := validatePurgeKnownResources(
		ctx,
		knownVerifier,
		scope,
		initial.Survivors,
		opCfg.knownResources(),
		opCfg.Region,
		opCfg.knownResourcesPrimaryRegion(),
	)
	if err != nil {
		return purgeScopePlan{}, err
	}
	survivors, err := filterPurgeDryRunStaleSurvivors(ctx, initial.Survivors, opCfg)
	if err != nil {
		return purgeScopePlan{}, err
	}
	survivors, err = mergePurgeKnownResourceSurvivors(ctx, knownVerifier, scope, survivors, validatedKnown)
	if err != nil {
		return purgeScopePlan{}, err
	}
	claims, err := planPurgeTrustedParentEC2Claims(ctx, survivors, scope, opCfg)
	if err != nil {
		return purgeScopePlan{}, err
	}
	asgClaims, err := planPurgeTrustedParentASGClaims(ctx, survivors, scope, opCfg)
	if err != nil {
		return purgeScopePlan{}, err
	}
	claims = append(claims, asgClaims...)
	eksResources, eksClaims, err := purgePlanEKSClaims(ctx, scope, opCfg)
	if err != nil {
		return purgeScopePlan{}, err
	}
	claims = append(claims, eksClaims...)
	survivors = mergePurgeResources(survivors, eksResources)
	preparedIDs := make([]string, 0, len(claims))
	for _, claim := range claims {
		preparedIDs = append(preparedIDs, preparedNukeIDForTrustedParentClaim(claim))
	}
	return purgeScopePlan{
		SurvivorResources:   survivors,
		TrustedParentClaims: claims,
		NukeResourceTypes:   destroyNukeResourceTypes(survivors, preparedIDs),
	}, nil
}

func mergePurgeKnownResourceSurvivors(
	ctx context.Context,
	verifier cpaws.KnownResourceVerifier,
	scope cpaws.Scope,
	survivors []cpaws.Resource,
	validatedKnown []cpaws.Resource,
) ([]cpaws.Resource, error) {
	if len(validatedKnown) == 0 {
		return survivors, nil
	}
	byARN := make(map[string]struct{}, len(survivors))
	for _, survivor := range survivors {
		byARN[survivor.ARN] = struct{}{}
	}
	for _, resource := range validatedKnown {
		if _, ok := byARN[resource.ARN]; ok {
			continue
		}
		if verifier == nil {
			// A resource which came from the tag scan but was removed by the
			// direct stale-resource filter already has terminal absence proof.
			continue
		}
		exists, err := verifier.VerifyKnownResource(ctx, scope, &resource)
		if err != nil {
			return nil, fmt.Errorf("final direct proof for known handoff resource %q: %w", resource.ARN, err)
		}
		if !exists {
			continue
		}
		byARN[resource.ARN] = struct{}{}
		survivors = append(survivors, resource)
	}
	return survivors, nil
}

func preparedNukeIDForTrustedParentClaim(claim purgeTrustedParentClaim) string {
	switch claim.Type {
	case "AutoScalingGroup":
		return nukePreparedAutoScalingGroupID(claim.ID)
	default:
		return claim.ID
	}
}

func planPurgeTrustedParentEC2Claims(
	ctx context.Context,
	survivors []cpaws.Resource,
	scope cpaws.Scope,
	opCfg purgeOperatorConfig,
) ([]purgeTrustedParentClaim, error) {
	clients := map[string]destroyEC2VPCChildAPI{}
	clientFor := func(region string) (destroyEC2VPCChildAPI, error) {
		region = firstNonEmptyString(strings.TrimSpace(region), strings.TrimSpace(opCfg.Region), firstNonEmptyString(scope.Regions...))
		if region == "" {
			return nil, fmt.Errorf("trusted-parent EC2 dry-run planning requires region")
		}
		if client := clients[region]; client != nil {
			return client, nil
		}
		cfg, err := cmdLoadAWSConfig(ctx, opCfg.Profile, region)
		if err != nil {
			return nil, fmt.Errorf("load EC2 client for trusted-parent dry-run planning in %s: %w", region, err)
		}
		cfg.Region = region
		client := destroyNewEC2Client(cfg)
		clients[region] = client
		return client, nil
	}

	var claims []purgeTrustedParentClaim
	for _, resource := range survivors {
		if resource.Type != cpaws.ResourceVPC {
			continue
		}
		if strings.TrimSpace(resource.Owner) != "" && !cpaws.OwnerTagMatches(scope.Owner, resource.Owner) {
			return nil, fmt.Errorf(
				"refusing trusted-parent EC2 dry-run planning for %s with owner=%q, want %q",
				resource.ARN,
				resource.Owner,
				scope.Owner,
			)
		}
		if strings.TrimSpace(resource.SessionID) != "" && resource.SessionID != scope.SessionID {
			return nil, fmt.Errorf(
				"refusing trusted-parent EC2 dry-run planning for %s with session=%q, want %q",
				resource.ARN,
				resource.SessionID,
				scope.SessionID,
			)
		}
		vpcID := ec2ResourceIDFromARN(resource.ARN, "vpc/")
		if vpcID == "" {
			return nil, fmt.Errorf("cannot parse VPC ID from %s for trusted-parent EC2 dry-run planning", resource.ARN)
		}
		client, err := clientFor(resource.Region)
		if err != nil {
			return nil, err
		}
		if err := scopedVPCTagsForDryRun(ctx, client, vpcID, scope); err != nil {
			return nil, err
		}
		vpcClaims, err := planPurgeVPCChildEC2Claims(ctx, client, vpcID, scope)
		if err != nil {
			return nil, err
		}
		claims = append(claims, vpcClaims...)
	}
	return claims, nil
}

func planPurgeTrustedParentASGClaims(
	ctx context.Context,
	survivors []cpaws.Resource,
	scope cpaws.Scope,
	opCfg purgeOperatorConfig,
) ([]purgeTrustedParentClaim, error) {
	ec2Clients := map[string]destroyEC2VPCChildAPI{}
	asgClients := map[string]destroyAutoScalingVPCChildAPI{}
	regionFor := func(region string) (string, error) {
		region = firstNonEmptyString(strings.TrimSpace(region), strings.TrimSpace(opCfg.Region), firstNonEmptyString(scope.Regions...))
		if region == "" {
			return "", fmt.Errorf("trusted-parent ASG dry-run planning requires region")
		}
		return region, nil
	}
	ec2ClientFor := func(region string) (destroyEC2VPCChildAPI, error) {
		region, err := regionFor(region)
		if err != nil {
			return nil, err
		}
		if client := ec2Clients[region]; client != nil {
			return client, nil
		}
		cfg, err := cmdLoadAWSConfig(ctx, opCfg.Profile, region)
		if err != nil {
			return nil, fmt.Errorf("load EC2 client for trusted-parent ASG dry-run planning in %s: %w", region, err)
		}
		cfg.Region = region
		client := destroyNewEC2Client(cfg)
		ec2Clients[region] = client
		return client, nil
	}
	asgClientFor := func(region string) (destroyAutoScalingVPCChildAPI, error) {
		region, err := regionFor(region)
		if err != nil {
			return nil, err
		}
		if client := asgClients[region]; client != nil {
			return client, nil
		}
		cfg, err := cmdLoadAWSConfig(ctx, opCfg.Profile, region)
		if err != nil {
			return nil, fmt.Errorf("load Auto Scaling client for trusted-parent ASG dry-run planning in %s: %w", region, err)
		}
		cfg.Region = region
		client := destroyNewAutoScalingClient(cfg)
		asgClients[region] = client
		return client, nil
	}

	var claims []purgeTrustedParentClaim
	for _, resource := range survivors {
		if resource.Type != cpaws.ResourceVPC {
			continue
		}
		if strings.TrimSpace(resource.Owner) != "" && !cpaws.OwnerTagMatches(scope.Owner, resource.Owner) {
			return nil, fmt.Errorf(
				"refusing trusted-parent ASG dry-run planning for %s with owner=%q, want %q",
				resource.ARN,
				resource.Owner,
				scope.Owner,
			)
		}
		if strings.TrimSpace(resource.SessionID) != "" && resource.SessionID != scope.SessionID {
			return nil, fmt.Errorf(
				"refusing trusted-parent ASG dry-run planning for %s with session=%q, want %q",
				resource.ARN,
				resource.SessionID,
				scope.SessionID,
			)
		}
		vpcID := ec2ResourceIDFromARN(resource.ARN, "vpc/")
		if vpcID == "" {
			return nil, fmt.Errorf("cannot parse VPC ID from %s for trusted-parent ASG dry-run planning", resource.ARN)
		}
		ec2Client, err := ec2ClientFor(resource.Region)
		if err != nil {
			return nil, err
		}
		asgClient, err := asgClientFor(resource.Region)
		if err != nil {
			return nil, err
		}
		if err := scopedVPCTagsForDryRun(ctx, ec2Client, vpcID, scope); err != nil {
			return nil, err
		}
		vpcClaims, err := planPurgeVPCChildASGClaims(ctx, ec2Client, asgClient, vpcID, scope)
		if err != nil {
			return nil, err
		}
		claims = append(claims, vpcClaims...)
	}
	return claims, nil
}

func scopedVPCTagsForDryRun(
	ctx context.Context,
	client destroyEC2VPCChildAPI,
	vpcID string,
	scope cpaws.Scope,
) error {
	out, err := client.DescribeVpcs(ctx, &ec2.DescribeVpcsInput{
		VpcIds: []string{vpcID},
	})
	if err != nil {
		return fmt.Errorf("describe scoped VPC %s before dry-run child planning: %w", vpcID, err)
	}
	if len(out.Vpcs) != 1 {
		return fmt.Errorf("describe scoped VPC %s returned %d VPCs", vpcID, len(out.Vpcs))
	}
	tags := ec2TagMap(out.Vpcs[0].Tags)
	if tags[cpaws.TagSession] != scope.SessionID ||
		strings.TrimSpace(tags[cpaws.TagOwner]) == "" ||
		!cpaws.OwnerTagMatches(scope.Owner, tags[cpaws.TagOwner]) {
		return fmt.Errorf(
			"refusing to plan children for VPC %s with tags owner=%q session=%q, want owner=%q session=%q",
			vpcID,
			tags[cpaws.TagOwner],
			tags[cpaws.TagSession],
			scope.Owner,
			scope.SessionID,
		)
	}
	return nil
}

func planPurgeVPCChildEC2Claims(
	ctx context.Context,
	client destroyEC2VPCChildAPI,
	vpcID string,
	scope cpaws.Scope,
) ([]purgeTrustedParentClaim, error) {
	ids, err := nukeVPCChildInstanceIDs(ctx, client, vpcID, scope, "dry-run child planning")
	if err != nil {
		return nil, err
	}
	claims := make([]purgeTrustedParentClaim, 0, len(ids))
	for _, id := range ids {
		claims = append(claims, purgeTrustedParentClaim{
			Type:   "EC2Instance",
			ID:     id,
			Parent: vpcID,
		})
	}
	return claims, nil
}

func planPurgeVPCChildASGClaims(
	ctx context.Context,
	ec2Client destroyEC2VPCChildAPI,
	asgClient destroyAutoScalingVPCChildAPI,
	vpcID string,
	scope cpaws.Scope,
) ([]purgeTrustedParentClaim, error) {
	names, err := nukeVPCChildAutoScalingGroupNames(ctx, ec2Client, asgClient, vpcID, scope, "dry-run child planning")
	if err != nil {
		return nil, err
	}
	claims := make([]purgeTrustedParentClaim, 0, len(names))
	for _, name := range names {
		claims = append(claims, purgeTrustedParentClaim{
			Type:   "AutoScalingGroup",
			ID:     name,
			Parent: vpcID,
		})
	}
	return claims, nil
}

func purgeWriteNukeConfig(
	scope cpaws.Scope,
	opCfg purgeOperatorConfig,
	survivors []cpaws.Resource,
	vpcChildren []string,
) (string, error) {
	types := destroyNukeResourceTypes(survivors, vpcChildren)
	name := fmt.Sprintf("cp9-purge-%s-%s.yaml", sanitizePurgePathPart(scope.Owner), sanitizePurgePathPart(scope.SessionID))
	configPath := filepath.Join(os.TempDir(), name)
	if err := cpaws.WriteNukeConfig(configPath, &cpaws.NukeConfigRequest{
		AccountID:     opCfg.AccountID,
		Scope:         scope,
		ResourceTypes: types,
	}); err != nil {
		return "", fmt.Errorf("write aws-nuke config: %w", err)
	}
	return configPath, nil
}

func sanitizePurgePathPart(s string) string {
	return strings.Map(func(r rune) rune {
		switch {
		case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-':
			return r
		default:
			return '_'
		}
	}, s)
}