package cmd

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"regexp"
	"strings"
	"time"

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

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

const (
	purgeDefaultRegion = "us-west-2"
)

// purgeSessionPattern matches a DNS-compatible cp9 session name (3-32 chars).
var purgeSessionPattern = regexp.MustCompile(`^[a-z][a-z0-9-]{1,30}[a-z0-9]$`)
var purgeAccountIDPattern = regexp.MustCompile(`^\d{12}$`)

// purgeOperatorConfig holds the operator/environment configuration for cp9
// purge. Per the accepted ADRs, the account id, cleanup role ARN, and AWS
// profile are operator/environment configuration, not ordinary public args, so
// that ordinary callers cannot accidentally change the cleanup account.
type purgeOperatorConfig struct {
	AccountID      string
	RoleARN        string
	Profile        string
	Region         string
	KnownResources *purgeOperatorKnownResources
}

type purgeOperatorKnownResources struct {
	Resources     []cpaws.Resource
	PrimaryRegion string
}

func (c purgeOperatorConfig) knownResources() []cpaws.Resource {
	if c.KnownResources == nil {
		return nil
	}
	return c.KnownResources.Resources
}

func (c purgeOperatorConfig) knownResourcesPrimaryRegion() string {
	if c.KnownResources == nil {
		return ""
	}
	return c.KnownResources.PrimaryRegion
}

func loadPurgeOperatorConfig(region string) purgeOperatorConfig {
	return purgeOperatorConfig{
		AccountID: strings.TrimSpace(os.Getenv("CP9_CLEANUP_ACCOUNT_ID")),
		RoleARN:   strings.TrimSpace(os.Getenv("CP9_CLEANUP_ROLE_ARN")),
		Profile:   strings.TrimSpace(os.Getenv("CP9_CLEANUP_AWS_PROFILE")),
		Region:    firstNonEmptyString(strings.TrimSpace(region), purgeDefaultRegion),
	}
}

func validatePurgeOperatorConfig(opCfg purgeOperatorConfig) error {
	if !purgeAccountIDPattern.MatchString(opCfg.AccountID) {
		return cp9errors.New(
			cp9errors.CategoryConfig,
			"cp9 cleanup: CP9_CLEANUP_ACCOUNT_ID must be an explicit 12-digit AWS account id",
		)
	}
	rolePrefix := fmt.Sprintf("arn:aws:iam::%s:role/", opCfg.AccountID)
	if !strings.HasPrefix(opCfg.RoleARN, rolePrefix) || strings.TrimSpace(strings.TrimPrefix(opCfg.RoleARN, rolePrefix)) == "" {
		return cp9errors.Newf(
			cp9errors.CategoryConfig,
			"cp9 cleanup: CP9_CLEANUP_ROLE_ARN must be an IAM role ARN in account %s",
			opCfg.AccountID,
		)
	}
	return nil
}

// purgeSurvivor identifies one resource still present after cleanup, so the
// report can name what blocked a scope instead of only a count.
type purgeSurvivor struct {
	Type string `json:"type"`
	ARN  string `json:"arn"`
}

type purgeTrustedParentClaim struct {
	Type   string `json:"type"`
	ID     string `json:"id"`
	Parent string `json:"parent"`
}

type purgeScopePlan struct {
	SurvivorResources   []cpaws.Resource
	TrustedParentClaims []purgeTrustedParentClaim
	NukeResourceTypes   []string
}

type purgeCandidate struct {
	Owner               string                    `json:"owner"`
	Session             string                    `json:"session"`
	TTL                 string                    `json:"ttl,omitempty"`
	NukeConfigOK        bool                      `json:"nuke_config_ok"`
	NukeConfigError     string                    `json:"nuke_config_error,omitempty"`
	NukeResourceTypes   []string                  `json:"nuke_resource_types,omitempty"`
	TrustedParentClaims []purgeTrustedParentClaim `json:"trusted_parent_claims,omitempty"`
	// Destructive-run outcome (populated only when dry-run=false).
	Deleted        int             `json:"deleted,omitempty"`
	Survivors      int             `json:"survivors,omitempty"`
	SurvivorDetail []purgeSurvivor `json:"survivor_detail,omitempty"`
	Blocked        bool            `json:"blocked,omitempty"`
	BlockReason    string          `json:"block_reason,omitempty"`
	StaleNoop      bool            `json:"stale_noop,omitempty"`
	Deferred       bool            `json:"deferred_live_dependency,omitempty"`
	DeferredReason string          `json:"deferred_reason,omitempty"`

	retryExpiredDependency bool
}

// survivorDetails projects survivor resources into the report shape.
func survivorDetails(rs []cpaws.Resource) []purgeSurvivor {
	out := make([]purgeSurvivor, 0, len(rs))
	for _, r := range rs {
		out = append(out, purgeSurvivor{Type: string(r.Type), ARN: r.ARN})
	}
	return out
}

// formatSurvivors renders survivors as "type arn; type arn" for a one-line reason.
func formatSurvivors(rs []cpaws.Resource) string {
	if len(rs) == 0 {
		return "(details unavailable)"
	}
	parts := make([]string, 0, len(rs))
	for _, r := range rs {
		id := strings.TrimSpace(r.ARN)
		if id == "" {
			id = "(no arn)"
		}
		parts = append(parts, fmt.Sprintf("%s %s", r.Type, id))
	}
	return strings.Join(parts, "; ")
}

type purgeResult struct {
	Mode               string                 `json:"mode"`
	DryRun             bool                   `json:"dry_run"`
	Region             string                 `json:"region"`
	AccountID          string                 `json:"account_id"`
	RoleARN            string                 `json:"role_arn"`
	Candidates         []purgeCandidate       `json:"candidates"`
	OrphanEKSProviders *orphanOIDCSweepResult `json:"orphan_eks_oidc_providers,omitempty"`
	OrphanOIDCError    string                 `json:"orphan_eks_oidc_error,omitempty"`
	Note               string                 `json:"note,omitempty"`
}

// FormatText implements model.TextFormatter for human-readable output.
func (r *purgeResult) FormatText() string {
	var b strings.Builder
	fmt.Fprintf(&b, "cp9 purge (%s mode, dry-run=%t)\n", r.Mode, r.DryRun)
	fmt.Fprintf(&b, "  account: %s\n", r.AccountID)
	fmt.Fprintf(&b, "  region:  %s\n", r.Region)
	fmt.Fprintf(&b, "  candidates: %d\n", len(r.Candidates))
	for i := range r.Candidates {
		c := &r.Candidates[i]
		status := "boundary-ok"
		switch {
		case c.Blocked:
			status = "blocked: " + c.BlockReason
		case c.Deferred:
			status = "deferred_live_dependency: " + c.DeferredReason
		case c.StaleNoop:
			status = "stale/no-op"
		case !c.NukeConfigOK:
			status = "blocked: " + c.NukeConfigError
		case !r.DryRun:
			status = fmt.Sprintf("deleted=%d survivors=%d", c.Deleted, c.Survivors)
		}
		ttl := c.TTL
		if ttl == "" {
			ttl = "-"
		}
		fmt.Fprintf(&b, "    - owner=%s session=%s ttl=%s [%s]\n", c.Owner, c.Session, ttl, status)
		if len(c.NukeResourceTypes) > 0 {
			fmt.Fprintf(&b, "        nuke_resource_types: %s\n", strings.Join(c.NukeResourceTypes, ", "))
		}
		for _, claim := range c.TrustedParentClaims {
			fmt.Fprintf(&b, "        trusted-parent: %s %s via %s\n", claim.Type, claim.ID, claim.Parent)
		}
		for _, s := range c.SurvivorDetail {
			fmt.Fprintf(&b, "        survivor: %s %s\n", s.Type, s.ARN)
		}
	}
	if r.OrphanEKSProviders != nil {
		fmt.Fprintf(&b, "  orphan EKS OIDC: scanned=%d eligible=%d deleted=%d\n",
			r.OrphanEKSProviders.Scanned,
			r.OrphanEKSProviders.Eligible,
			r.OrphanEKSProviders.Deleted,
		)
	}
	if r.OrphanOIDCError != "" {
		fmt.Fprintf(&b, "  orphan EKS OIDC blocked: %s\n", r.OrphanOIDCError)
	}
	if r.Note != "" {
		fmt.Fprintf(&b, "  note: %s\n", r.Note)
	}
	return b.String()
}

// purgeCmd implements the top-level `cp9 purge` cloud-cleanup command.
//
// purge is the DESTRUCTIVE, act-on-provable-scope half of cleanup: it cleans
// resources tagged with cp9-owner+cp9-session, plus untagged resources it can
// attach to a trusted parent (ADR 0001), and refuses anything it cannot prove.
// Its complement is `cp9 inventory` (read-only attribution of standalone
// orphaned resources purge will not touch). See docs/adr/0007 for the boundary.
//
// Mode is inferred from inputs: no owner/session => broad expired mode; paired
// owner+session => exact mode (TTL ignored); exactly one is an error. dry-run
// (the default) is fully read-only: it reports candidates and the cleanup
// boundary without mutating tags or deleting resources. Account id, cleanup role
// ARN, and AWS profile are operator/environment configuration.
func purgeCmd() model.NormalCmd {
	return func(args model.ArgVals, cc *model.Cli, env *model.Env, _ []model.ParsedCmd) (runErr error) {
		syncVerbose(env)

		dryRun := !strings.EqualFold(strings.TrimSpace(args.GetRaw("dry-run")), "false")
		owner := strings.TrimSpace(args.GetRaw("owner"))
		session := strings.TrimSpace(args.GetRaw("session"))
		opCfg := loadPurgeOperatorConfig(args.GetRaw("region"))

		mode, err := cpaws.ResolvePurgeMode(owner, session)
		if err != nil {
			return cp9errors.Newf(cp9errors.CategoryConfig, "cp9 purge: %v", err)
		}
		if mode == cpaws.PurgeModeExact && !purgeSessionPattern.MatchString(session) {
			return cp9errors.Newf(cp9errors.CategoryConfig,
				"cp9 purge: session %q is not a valid cp9 session name", session)
		}
		if path := strings.TrimSpace(args.GetRaw("known-resources")); path != "" {
			if mode != cpaws.PurgeModeExact {
				return cp9errors.New(cp9errors.CategoryConfig, "cp9 purge: known-resources requires exact owner/session mode")
			}
			known, loadErr := loadPurgeKnownResources(path, owner, session, opCfg.AccountID)
			if loadErr != nil {
				return cp9errors.Newf(cp9errors.CategoryConfig, "cp9 purge: %v", loadErr)
			}
			opCfg.KnownResources = known
		}

		ctx := context.Background()
		if !dryRun {
			var cancel context.CancelFunc
			ctx, cancel = context.WithTimeout(ctx, purgeMaxSeconds())
			defer cancel()
		}

		// Fail-closed safety guard: confirm we are running as the expected
		// cleanup role in the expected account before discovering anything.
		// A mismatch is a systemic error even in dry-run mode.
		if err := verifyPurgeIdentity(ctx, opCfg); err != nil {
			return err
		}
		if !dryRun {
			lease, deferred, err := acquirePurgeAccountLease(ctx, mode, opCfg)
			if err != nil {
				return cp9errors.Newf(cp9errors.CategoryResource, "cp9 purge: %v", err)
			}
			if deferred {
				return model.Output(cc, env, &purgeResult{
					Mode:      string(mode),
					DryRun:    false,
					Region:    opCfg.Region,
					AccountID: opCfg.AccountID,
					RoleARN:   opCfg.RoleARN,
					Note:      "deferred_account_lock: another destructive cleanup owns this account; scheduled cleanup will retry",
				})
			}
			ctx = lease.ctx
			defer func() {
				if err := lease.close(); err != nil {
					runErr = errors.Join(runErr, cp9errors.Newf(cp9errors.CategoryResource,
						"cp9 purge: cleanup account lease: %v", err))
				}
			}()
		}

		scopes, iamInventory, err := resolvePurgeScopes(ctx, mode, owner, session, opCfg)
		if err != nil {
			return err
		}
		candidates := make([]purgeCandidate, 0, len(scopes))
		for i := range scopes {
			candidates = append(candidates, buildPurgeCandidate(&scopes[i], opCfg))
		}

		result := &purgeResult{
			Mode:       string(mode),
			DryRun:     dryRun,
			Region:     opCfg.Region,
			AccountID:  opCfg.AccountID,
			RoleARN:    opCfg.RoleARN,
			Candidates: candidates,
		}

		if dryRun {
			result.Note = "dry-run: read-only; no tags mutated and no resources deleted"
			anyBlocked := false
			if mode == cpaws.PurgeModeBroad {
				oidc, oidcErr := purgeOrphanOIDCSweep(ctx, opCfg, false)
				result.OrphanEKSProviders = &oidc
				if oidcErr != nil {
					result.OrphanOIDCError = oidcErr.Error()
					anyBlocked = true
				}
			}
			if mode == cpaws.PurgeModeExact {
				anyBlocked = populatePurgeDryRunPlans(ctx, result, scopes, opCfg)
			}
			if err := model.Output(cc, env, result); err != nil {
				return err
			}
			if anyBlocked {
				return cp9errors.New(cp9errors.CategoryResource,
					"cp9 purge: one or more dry-run candidates were blocked")
			}
			return nil
		}
		return runPurgeDestructive(ctx, cc, env, result, scopes, opCfg, iamInventory)
	}
}

func loadPurgeKnownResources(path, owner, session, accountID string) (*purgeOperatorKnownResources, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return nil, fmt.Errorf("read known-resources handoff %s: %w", path, err)
	}
	var receipt destroyCloudCleanupHandoffReceipt
	if err := json.Unmarshal(data, &receipt); err != nil {
		return nil, fmt.Errorf("parse known-resources handoff %s: %w", path, err)
	}
	if receipt.Version != destroyCleanupHandoffReceiptVersion || receipt.Owner != owner ||
		receipt.SessionID != session || receipt.AccountID != accountID {
		return nil, fmt.Errorf("known-resources handoff scope does not match exact purge owner/session/account")
	}
	regions := normalizedDestroyHandoffValues(receipt.Regions)
	if len(regions) == 0 {
		return nil, fmt.Errorf("known-resources handoff has no cleanup region")
	}
	result := make([]cpaws.Resource, 0, len(receipt.KnownResources))
	for _, resource := range receipt.KnownResources {
		if strings.TrimSpace(resource.ARN) == "" || resource.Type == "" || resource.Type == cpaws.ResourceUnknown {
			return nil, fmt.Errorf("known-resources handoff contains an unsupported resource identity")
		}
		if resource.Owner != owner || resource.SessionID != session {
			return nil, fmt.Errorf("known resource %q has conflicting owner/session", resource.ARN)
		}
		parts := strings.SplitN(resource.ARN, ":", 6)
		if len(parts) >= 5 && parts[0] == "arn" && parts[4] != "" && parts[4] != accountID {
			return nil, fmt.Errorf("known resource %q is outside cleanup account %q", resource.ARN, accountID)
		}
		result = append(result, resource)
	}
	return &purgeOperatorKnownResources{Resources: result, PrimaryRegion: regions[0]}, nil
}

func populatePurgeDryRunPlans(
	ctx context.Context,
	result *purgeResult,
	scopes []cpaws.ExpiredManagedScope,
	opCfg purgeOperatorConfig,
) bool {
	if result == nil {
		return false
	}
	anyBlocked := false
	for i := range result.Candidates {
		if i >= len(scopes) {
			break
		}
		candidate := &result.Candidates[i]
		plan, err := purgePlanScope(ctx, purgeScopeFor(&scopes[i], opCfg), opCfg)
		if err != nil {
			candidate.Blocked = true
			candidate.BlockReason = err.Error()
			anyBlocked = true
			continue
		}
		candidate.SurvivorDetail = survivorDetails(plan.SurvivorResources)
		candidate.TrustedParentClaims = append([]purgeTrustedParentClaim(nil), plan.TrustedParentClaims...)
		candidate.NukeResourceTypes = append([]string(nil), plan.NukeResourceTypes...)
	}
	return anyBlocked
}

func verifyPurgeIdentity(ctx context.Context, opCfg purgeOperatorConfig) error {
	if err := validatePurgeOperatorConfig(opCfg); err != nil {
		return err
	}
	awsCfg, err := cmdLoadAWSConfig(ctx, opCfg.Profile, opCfg.Region)
	if err != nil {
		return cp9errors.Newf(cp9errors.CategoryResource, "cp9 purge: load aws config: %v", err)
	}
	out, err := cmdNewSTSClient(&awsCfg).GetCallerIdentity(ctx, &sts.GetCallerIdentityInput{})
	if err != nil {
		return cp9errors.Newf(cp9errors.CategoryResource, "cp9 purge: sts GetCallerIdentity: %v", err)
	}
	account := awssdk.ToString(out.Account)
	if account != opCfg.AccountID {
		return cp9errors.Newf(cp9errors.CategoryPermanent,
			"cp9 purge: caller account %q does not match expected cleanup account %q", account, opCfg.AccountID)
	}
	if prefix := assumedRolePrefix(opCfg); prefix != "" && !strings.HasPrefix(awssdk.ToString(out.Arn), prefix) {
		return cp9errors.Newf(cp9errors.CategoryPermanent,
			"cp9 purge: caller %q is not the expected cleanup role %q", awssdk.ToString(out.Arn), opCfg.RoleARN)
	}
	return nil
}

// assumedRolePrefix derives the STS assumed-role ARN prefix from the cleanup
// role ARN, e.g. arn:aws:iam::ACC:role/cp9-developer =>
// arn:aws:sts::ACC:assumed-role/cp9-developer/.
func assumedRolePrefix(opCfg purgeOperatorConfig) string {
	idx := strings.LastIndex(opCfg.RoleARN, ":role/")
	if idx < 0 {
		return ""
	}
	roleName := opCfg.RoleARN[idx+len(":role/"):]
	if roleName == "" {
		return ""
	}
	return fmt.Sprintf("arn:aws:sts::%s:assumed-role/%s/", opCfg.AccountID, roleName)
}

// resolvePurgeScopes returns the owner/session candidates for the run: a single
// explicit pair in exact mode, or the discovered expired cp9-managed scopes in
// broad mode. Discovery is read-only.
func resolvePurgeScopes(
	ctx context.Context,
	mode cpaws.PurgeMode,
	owner, session string,
	opCfg purgeOperatorConfig,
) ([]cpaws.ExpiredManagedScope, *cpaws.IAMInventory, error) {
	switch mode {
	case cpaws.PurgeModeExact:
		return []cpaws.ExpiredManagedScope{{Owner: owner, Session: session}}, nil, nil
	case cpaws.PurgeModeBroad:
		awsCfg, err := cmdLoadAWSConfig(ctx, opCfg.Profile, opCfg.Region)
		if err != nil {
			return nil, nil, cp9errors.Newf(cp9errors.CategoryResource, "cp9 purge: load aws config: %v", err)
		}
		now := time.Now()
		regional, err := cmdFindManagedScopes(ctx, cmdNewTaggingClient(&awsCfg))
		if err != nil {
			return nil, nil, cp9errors.Newf(cp9errors.CategoryResource, "cp9 purge: discover expired sessions: %v", err)
		}
		asgScopes, err := cmdFindManagedASGScopes(ctx, cmdNewAutoScalingClient(&awsCfg))
		if err != nil {
			return nil, nil, cp9errors.Newf(cp9errors.CategoryResource, "cp9 purge: discover Auto Scaling managed scopes: %v", err)
		}
		merged := cpaws.MergeManagedScopes(regional, asgScopes)
		// Resource Groups Tagging API is regional and never returns IAM, so a
		// session whose only survivors are IAM entities is invisible above. Read the
		// account's IAM once: it both completes discovery and becomes the snapshot
		// every candidate reuses instead of re-scanning IAM per session.
		iamClient := cpaws.NewIAMInventoryClient(&awsCfg)
		inventory, invErr := cpaws.ScanIAMInventory(ctx, iamClient, iamClient)
		if invErr != nil {
			// Degrade rather than block: the tag-index candidates above are still
			// valid work, and IAM-only sessions were already invisible before this.
			verbose.Logf("cp9 purge: iam inventory scan degraded: %v", invErr)
		}
		if inventory != nil {
			merged = cpaws.MergeManagedScopes(merged, inventory.ManagedScopes())
		}
		expired := cpaws.FilterExpiredManagedScopes(merged, now)
		req := &cpaws.NukeRunRequest{
			Profile:   opCfg.Profile,
			Region:    opCfg.Region,
			AccountID: opCfg.AccountID,
		}
		if err := purgeMaterializeNukeCredentials(ctx, req); err != nil {
			return nil, nil, cp9errors.Newf(cp9errors.CategoryResource,
				"cp9 purge: materialize broad stale verifier credentials: %v", err)
		}
		iamComplete := invErr == nil && inventory != nil && inventory.Complete()
		for i := range expired {
			if len(expired[i].Resources) == 0 {
				continue
			}
			live, stale, verifyErr := filterPostNukeStaleSurvivorsWithWarnings(
				ctx,
				expired[i].Resources,
				req,
				"purge-broad-discovery",
			)
			if verifyErr != nil {
				verbose.Logf("cp9 purge: broad scope owner=%q session=%q cannot prove stale: %v",
					expired[i].Owner, expired[i].Session, verifyErr)
				continue
			}
			expired[i].Resources = live
			expired[i].VerifiedStale = len(live) == 0 && iamComplete
			if expired[i].VerifiedStale {
				verbose.Logf("cp9 purge: broad scope owner=%q session=%q is stale/no-op resources=%d",
					expired[i].Owner, expired[i].Session, len(stale))
			}
		}
		return expired, inventory, nil
	default:
		return nil, nil, cp9errors.Newf(cp9errors.CategoryConfig, "cp9 purge: unknown mode %q", mode)
	}
}

// buildPurgeCandidate builds the report row for a scope, proving the safe
// owner/session aws-nuke boundary is expressible (read-only) for it.
func buildPurgeCandidate(scope *cpaws.ExpiredManagedScope, opCfg purgeOperatorConfig) purgeCandidate {
	candidate := purgeCandidate{Owner: scope.Owner, Session: scope.Session, StaleNoop: scope.VerifiedStale}
	if !scope.TTL.IsZero() {
		candidate.TTL = scope.TTL.UTC().Format(time.RFC3339)
	}
	_, err := cpaws.GenerateNukeConfig(&cpaws.NukeConfigRequest{
		AccountID: opCfg.AccountID,
		Scope:     purgeScopeFor(scope, opCfg),
	})
	if err != nil {
		candidate.NukeConfigError = err.Error()
	} else {
		candidate.NukeConfigOK = true
	}
	return candidate
}

// purgeScopeFor converts a discovered/explicit candidate into a cleanup Scope.
func purgeScopeFor(scope *cpaws.ExpiredManagedScope, opCfg purgeOperatorConfig) cpaws.Scope {
	return cpaws.Scope{
		Owner:     scope.Owner,
		SessionID: scope.Session,
		Regions:   []string{opCfg.Region},
	}
}