//nolint:gocritic // The AWS lifecycle deliberately passes immutable activation contracts by value across every boundary.
package byoc

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"maps"
	"os"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"sync"
	"time"

	awssdk "github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/acmpca"
	acmpcatypes "github.com/aws/aws-sdk-go-v2/service/acmpca/types"
	"github.com/aws/aws-sdk-go-v2/service/cloudformation"
	cloudformationtypes "github.com/aws/aws-sdk-go-v2/service/cloudformation/types"
	"github.com/aws/aws-sdk-go-v2/service/ec2"
	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
	"github.com/aws/aws-sdk-go-v2/service/ecr"
	ecrtypes "github.com/aws/aws-sdk-go-v2/service/ecr/types"
	"github.com/aws/aws-sdk-go-v2/service/route53"
	"github.com/aws/aws-sdk-go-v2/service/ssm"
	ssmtypes "github.com/aws/aws-sdk-go-v2/service/ssm/types"
	"github.com/aws/aws-sdk-go-v2/service/sts"
	"github.com/aws/smithy-go"
	"gopkg.in/yaml.v3"

	"github.com/tidbcloud/controlplane9/internal/byoc/accountsetup"
)

const (
	accountLeaseParameterName   = "/tidbcloud/cp9/byoc-account-lease"
	accountLeaseAPIVersion      = "cp9.byoc.lease/v1alpha1"
	selectorSlotParameterPrefix = "/tidbcloud/cp9/byoc-selector-slots/sha256/"
	selectorSlotAPIVersion      = "cp9.byoc.selector-slot/v1alpha1"
	accountSetupStackDeploy     = "tidbcloud-byoc-setup-deploy"
	accountSetupStackDataPlane  = "tidbcloud-byoc-setup-dataplane"
	accountSetupStackO11Y       = "tidbcloud-byoc-setup-o11y"
	accountSetupRetainedKey     = "retained"
	globalBaselineNamespace     = "cp9-global"
	githubRunnerGoogleAccountID = "114667344163696279999"
	o11yGlobalRoleARNs          = "arn:aws:iam::557537366020:role/globalserver-role-780c8f0,arn:aws:iam::380838443567:role/tidbcloud-global-apigw"
	PinnedCraneVersion          = "v0.20.6"
	PinnedCraneModule           = "github.com/google/go-containerregistry/cmd/crane@" + PinnedCraneVersion
)

// accountSetupCommandWrapper preserves the pinned official script byte for
// byte while adding the repository-mandated cp9 ownership tags to each of its
// three CloudFormation deploys. All other AWS CLI calls pass through exactly.
const accountSetupCommandWrapper = `set -Eeuo pipefail
aws() {
  if [[ "${1-}" == "cloudformation" && "${2-}" == "deploy" ]]; then
    local arg previous="" stack_name="" expected="" observed="" error_text="" error_file=""
    local describe_status=0
    for arg in "$@"; do
      if [[ "$previous" == "--stack-name" ]]; then
        stack_name="$arg"
      fi
      case "$arg" in
        --stack-name=*) stack_name="${arg#--stack-name=}" ;;
      esac
      previous="$arg"
    done
    case "$stack_name" in
      tidbcloud-byoc-setup-deploy) expected="${CP9_SETUP_EXPECTED_DEPLOY_STACK_ID}" ;;
      tidbcloud-byoc-setup-dataplane) expected="${CP9_SETUP_EXPECTED_DATAPLANE_STACK_ID}" ;;
      tidbcloud-byoc-setup-o11y) expected="${CP9_SETUP_EXPECTED_O11Y_STACK_ID}" ;;
      *) printf 'refusing unrecognized BYOC setup stack %s\n' "$stack_name" >&2; return 65 ;;
    esac
    error_file="$(mktemp)"
    if observed="$(command aws cloudformation describe-stacks --stack-name "$stack_name" --query 'Stacks[0].StackId' --output text 2>"$error_file")"; then
      rm -f "$error_file"
      if [[ "$expected" == "ABSENT" ]]; then
        printf 'refusing to adopt stack %s created after the leased preflight\n' "$stack_name" >&2
        return 66
      fi
      if [[ "$observed" != "$expected" ]]; then
        printf 'refusing changed stack identity for %s\n' "$stack_name" >&2
        return 67
      fi
    else
      describe_status=$?
      error_text="$(cat "$error_file")"
      rm -f "$error_file"
      if [[ "$expected" != "ABSENT" || "$error_text" != *"does not exist"* ]]; then
        printf '%s\n' "$error_text" >&2
        return "$describe_status"
      fi
    fi
    local actual_account=""
    actual_account="$(command aws sts get-caller-identity --query Account --output text)"
    if [[ -z "${CP9_SETUP_EXPECTED_CUSTOMER_ACCOUNT_ID-}" || "$actual_account" != "${CP9_SETUP_EXPECTED_CUSTOMER_ACCOUNT_ID-}" ]]; then
      printf 'refusing BYOC setup deploy for customer account %s while authenticated as %s\n' \
        "${CP9_SETUP_EXPECTED_CUSTOMER_ACCOUNT_ID-<unset>}" "${actual_account:-<unset>}" >&2
      return 68
    fi
    command aws "$@" --tags "cp9-owner=${CP9_SETUP_OWNER}" "cp9-session=${CP9_SETUP_SESSION}"
  else
    command aws "$@"
  fi
}
export -f aws
exec bash "$@"
`

var accountSetupStackNames = []string{
	accountSetupStackDeploy,
	accountSetupStackDataPlane,
	accountSetupStackO11Y,
}

type AWSDriverOptions struct {
	Runner           Runner
	KubernetesRunner Runner
	ImageRunner      Runner
	Bundle           *accountsetup.Bundle
	ImageClosure     *ImageClosure
}

type AWSDriver struct {
	runner                 Runner
	kubernetesRunner       Runner
	imageRunner            Runner
	bundle                 *accountsetup.Bundle
	imageClosure           *ImageClosure
	clientsMu              sync.Mutex
	clientsByProfileRegion map[string]*byocAWSClients
}

func NewAWSDriver(options AWSDriverOptions) (*AWSDriver, error) {
	runner := options.Runner
	if runner == nil {
		runner = ExecRunner{}
	}
	kubernetesRunner := options.KubernetesRunner
	if kubernetesRunner == nil {
		kubernetesRunner = runner
	}
	imageRunner := options.ImageRunner
	if imageRunner == nil {
		imageRunner = runner
	}
	if options.Bundle == nil {
		return nil, errors.New("pinned BYOC account-setup bundle is required")
	}
	if options.ImageClosure != nil && options.ImageClosure.Digest != closureDigest(options.ImageClosure.Images) {
		return nil, errors.New("BYOC image closure digest does not match its image list")
	}
	return &AWSDriver{
		runner:                 runner,
		kubernetesRunner:       kubernetesRunner,
		imageRunner:            imageRunner,
		bundle:                 options.Bundle,
		imageClosure:           options.ImageClosure,
		clientsByProfileRegion: make(map[string]*byocAWSClients),
	}, nil
}

//nolint:gocritic // The immutable lifecycle contract is deliberately copied into the read-only boundary.
func (d *AWSDriver) Preflight(ctx context.Context, input Input, journal *Journal) (*PreflightResult, error) {
	if input.BundleCommit != d.bundle.UpstreamCommit || input.BundleDigest != d.bundle.Digest {
		return nil, fmt.Errorf(
			"BYOC account-setup bundle pin mismatch: input %s/%s, embedded %s/%s",
			input.BundleCommit,
			input.BundleDigest,
			d.bundle.UpstreamCommit,
			d.bundle.Digest,
		)
	}
	if d.imageClosure != nil && input.ImageClosureDigest != d.imageClosure.Digest {
		return nil, fmt.Errorf("BYOC image closure pin mismatch: input %s, discovered %s", input.ImageClosureDigest, d.imageClosure.Digest)
	}
	if err := d.validatePinnedCrane(ctx); err != nil {
		return nil, err
	}
	controlIdentity, err := d.getCallerIdentity(ctx, input.ControlPlaneProfile, input.Region)
	if err != nil {
		return nil, fmt.Errorf("resolve control-plane AWS identity: %w", err)
	}
	if controlIdentity.Account != input.ControlPlaneAccountID {
		return nil, fmt.Errorf("control-plane profile resolved to account %s, want %s", controlIdentity.Account, input.ControlPlaneAccountID)
	}
	customerIdentity, err := d.getCallerIdentity(ctx, input.CustomerProfile, input.Region)
	if err != nil {
		return nil, fmt.Errorf("resolve customer AWS identity: %w", err)
	}
	if customerIdentity.Account != input.CustomerAccountID {
		return nil, fmt.Errorf("customer account-setup profile resolved to account %s, want %s", customerIdentity.Account, input.CustomerAccountID)
	}
	if customerIdentity.Account == controlIdentity.Account {
		return nil, errors.New("BYOC control-plane and customer profiles resolved to the same AWS account")
	}
	accountSetupPrincipalARN, err := d.validateControlPlaneAccountSetupPrincipal(ctx, input)
	if err != nil {
		return nil, err
	}

	for _, zone := range []struct {
		name string
		id   string
	}{
		{name: input.TiDBHostedZoneName, id: input.TiDBHostedZoneID},
		{name: input.O11YHostedZoneName, id: input.O11YHostedZoneID},
	} {
		if err := d.validateHostedZone(ctx, input, zone.id, zone.name); err != nil {
			return nil, err
		}
	}
	if err := d.validatePCA(ctx, input); err != nil {
		return nil, err
	}
	if err := d.validateAvailabilityZones(ctx, input); err != nil {
		return nil, err
	}
	vpcID, subnetAZID, err := d.validateControlPlaneNetwork(ctx, input)
	if err != nil {
		return nil, err
	}
	if err := d.validateBaselineKubernetes(ctx, input); err != nil {
		return nil, err
	}
	if err := d.validateCatalogSelectionBoundary(ctx, input, journal); err != nil {
		return nil, err
	}
	if err := d.validateCatalogIdentityBoundary(ctx, input); err != nil {
		return nil, err
	}
	if err := d.validateBYOCRuntimeNamespace(ctx, input, journal); err != nil {
		return nil, err
	}

	fingerprint, err := Fingerprint(input)
	if err != nil {
		return nil, err
	}
	selectorSlot, err := d.readSelectorSlot(ctx, input)
	if err != nil {
		return nil, fmt.Errorf("observe BYOC selector-slot reservation: %w", err)
	}
	if selectorSlot != nil && !selectorSlotBoundToInput(selectorSlot, input, fingerprint) {
		return nil, fmt.Errorf(
			"BYOC selector for tenant %s, aws/%s, and service plan BYOC is reserved by session %q on %q",
			input.TenantID,
			input.Region,
			selectorSlot.Session,
			selectorSlot.Machine,
		)
	}
	lease, err := d.readLease(ctx, input)
	if err != nil {
		return nil, fmt.Errorf("observe customer account lease: %w", err)
	}
	if lease != nil && !leaseBoundToInput(lease, input, fingerprint) {
		return nil, fmt.Errorf("customer account %s is leased by session %q on %q", input.CustomerAccountID, lease.Session, lease.Machine)
	}
	for _, stackName := range accountSetupStackNames {
		if err := d.validatePreflightStack(ctx, input, journal, stackName); err != nil {
			return nil, err
		}
	}
	if input.ReuseAccountSetup {
		if err := d.validateRetainedAccountSetupSemantics(ctx, input); err != nil {
			return nil, err
		}
	}
	imageSourceEvidence, err := d.preflightImageSources(ctx, input, journal)
	if err != nil {
		return nil, err
	}

	observed := map[string]string{
		"control_plane_vpc_id":                  vpcID,
		"control_plane_subnet_az_id":            subnetAZID,
		"control_plane_account_setup_principal": accountSetupPrincipalARN,
		"catalog_selector_slot":                 observedPresence(selectorSlot != nil),
		"catalog_selector_slot_name":            selectorSlotParameterName(input),
		"account_lease":                         observedPresence(lease != nil),
		"account_lease_region":                  accountLeaseRegion(input),
		"account_setup_bundle":                  d.bundle.Digest,
		"image_tool":                            "crane@" + PinnedCraneVersion,
		"catalog_selector":                      "unambiguous",
	}
	maps.Copy(observed, imageSourceEvidence)

	return &PreflightResult{
		ControlPlaneCallerARN: controlIdentity.ARN,
		CustomerCallerARN:     customerIdentity.ARN,
		Observed:              observed,
	}, nil
}

func (d *AWSDriver) validatePinnedCrane(ctx context.Context) error {
	output, err := d.imageRunner.Run(ctx, Command{Name: "crane", Args: []string{"version"}})
	install := "go install " + PinnedCraneModule
	if err != nil {
		return fmt.Errorf(
			"required BYOC image tool crane %s is unavailable; run `%s` on the configured execution plane: %w",
			PinnedCraneVersion,
			install,
			err,
		)
	}
	version := strings.TrimSpace(string(output))
	if version != PinnedCraneVersion {
		return fmt.Errorf(
			"BYOC image tool crane version %s, want %s; run `%s` on the configured execution plane",
			version,
			PinnedCraneVersion,
			install,
		)
	}
	return nil
}

//nolint:gocritic // The immutable lifecycle contract is deliberately copied into the observation boundary.
func (d *AWSDriver) Revalidate(ctx context.Context, input Input, stage Stage, record *StageRecord) (*StageResult, error) {
	switch stage {
	case StageLease:
		fingerprint, err := Fingerprint(input)
		if err != nil {
			return nil, err
		}
		selectorSlot, err := d.readSelectorSlot(ctx, input)
		if err != nil {
			return nil, err
		}
		if !selectorSlotBoundToInput(selectorSlot, input, fingerprint) {
			return nil, ErrNeedsReconcile
		}
		lease, err := d.readLease(ctx, input)
		if err != nil {
			return nil, err
		}
		if !leaseBoundToInput(lease, input, fingerprint) {
			return nil, ErrNeedsReconcile
		}
		return &StageResult{Artifacts: map[string]string{
			"parameter_name":          accountLeaseParameterName,
			"region":                  accountLeaseRegion(input),
			"version":                 fmt.Sprint(lease.Version),
			"selector_parameter_name": selectorSlotParameterName(input),
			"selector_region":         accountLeaseRegion(input),
			"selector_version":        fmt.Sprint(selectorSlot.Version),
		}}, nil
	case StageAccountSetup:
		if record == nil || record.Artifacts["bundle_commit"] != d.bundle.UpstreamCommit ||
			record.Artifacts["bundle_digest"] != d.bundle.Digest {
			return nil, ErrNeedsReconcile
		}
		if (record.Artifacts[accountSetupRetainedKey] == "true") != input.ReuseAccountSetup {
			return nil, ErrNeedsReconcile
		}
		artifacts := cloneArtifacts(record.Artifacts)
		for _, stackName := range accountSetupStackNames {
			boundID := record.Artifacts["stack."+stackName+".id"]
			creationTime := record.Artifacts["stack."+stackName+".creation_time"]
			if boundID == "" || creationTime == "" {
				return nil, ErrNeedsReconcile
			}
			stack, err := d.describeStack(ctx, input, boundID)
			if err != nil {
				if isAWSNotFound(err) {
					return nil, ErrNeedsReconcile
				}
				return nil, err
			}
			if err := validateStackIdentity(input, stackName, stack); err != nil {
				return nil, err
			}
			if stack.StackID != boundID || !sameStackCreationTime(stack.CreationTime, creationTime) {
				return nil, fmt.Errorf("bound BYOC setup stack %s identity or creation time changed", stackName)
			}
			var validationErr error
			if input.ReuseAccountSetup {
				validationErr = validateRetainedStackParameters(input, stackName, stack)
			} else {
				validationErr = validateStackParameters(input, stackName, stack)
			}
			if validationErr != nil {
				return nil, validationErr
			}
			if !input.ReuseAccountSetup {
				if err := d.validateStackTemplate(ctx, input, stackName, stack.StackID); err != nil {
					return nil, err
				}
			}
			artifacts["stack."+stackName+".id"] = stack.StackID
			artifacts["stack."+stackName+".creation_time"] = stack.CreationTime
		}
		if input.ReuseAccountSetup {
			if err := d.validateRetainedAccountSetupSemantics(ctx, input); err != nil {
				return nil, err
			}
		}
		return &StageResult{Artifacts: artifacts}, nil
	case StageImageMirror:
		return d.revalidateImageMirror(ctx, input, record)
	case StageControlPlaneRole:
		return d.revalidateControlPlaneRole(ctx, input, record)
	case StageNetworkAssets:
		return d.revalidateNetworkAssets(ctx, input, record)
	case StageRegionalRuntime:
		return d.revalidateRegionalRuntime(ctx, input, record)
	case StageCatalog:
		return d.revalidateCatalog(ctx, input, record)
	case StageRoleChainProof:
		return d.revalidateRoleChainProof(ctx, input, record)
	default:
		return nil, ErrNeedsReconcile
	}
}

//nolint:gocritic // The immutable lifecycle contract is deliberately copied into the mutation boundary.
func (d *AWSDriver) Apply(ctx context.Context, input Input, stage Stage, journal *Journal) (*StageResult, error) {
	switch stage {
	case StageLease:
		return d.acquireLease(ctx, input, journal)
	case StageAccountSetup:
		return d.applyAccountSetup(ctx, input, journal)
	case StageImageMirror:
		return d.applyImageMirror(ctx, input, journal)
	case StageControlPlaneRole:
		return d.applyControlPlaneRole(ctx, input)
	case StageNetworkAssets:
		return d.applyNetworkAssets(ctx, input)
	case StageRegionalRuntime:
		return d.applyRegionalRuntime(ctx, input, journal)
	case StageCatalog:
		return d.applyCatalog(ctx, input, journal)
	case StageRoleChainProof:
		return d.applyRoleChainProof(ctx, input)
	default:
		return nil, fmt.Errorf("BYOC stage %s is not implemented by the AWS driver", stage)
	}
}

type callerIdentity struct {
	Account string
	ARN     string
	UserID  string
}

func (d *AWSDriver) getCallerIdentity(ctx context.Context, profile, region string) (*callerIdentity, error) {
	clients, err := d.clients(ctx, profile, region)
	if err != nil {
		return nil, err
	}
	output, err := clients.sts.GetCallerIdentity(ctx, &sts.GetCallerIdentityInput{})
	if err != nil {
		return nil, err
	}
	identity := callerIdentity{
		Account: awssdk.ToString(output.Account),
		ARN:     awssdk.ToString(output.Arn),
		UserID:  awssdk.ToString(output.UserId),
	}
	if identity.Account == "" || identity.ARN == "" {
		return nil, errors.New("sts get-caller-identity returned an incomplete identity")
	}
	return &identity, nil
}

//nolint:gocritic // Hosted-zone validation receives the immutable activation contract by value.
func (d *AWSDriver) validateHostedZone(ctx context.Context, input Input, zoneID, zoneName string) error {
	clients, err := d.clients(ctx, input.CustomerProfile, input.Region)
	if err != nil {
		return fmt.Errorf("read customer hosted zone %s: %w", zoneID, err)
	}
	response, err := clients.route53.GetHostedZone(ctx, &route53.GetHostedZoneInput{Id: awssdk.String(zoneID)})
	if err != nil {
		return fmt.Errorf("read customer hosted zone %s: %w", zoneID, err)
	}
	if response.HostedZone == nil ||
		normalizeHostedZoneID(awssdk.ToString(response.HostedZone.Id)) != normalizeHostedZoneID(zoneID) ||
		normalizeDNSName(awssdk.ToString(response.HostedZone.Name)) != normalizeDNSName(zoneName) {
		return fmt.Errorf("customer hosted zone %s does not match configured name %q", zoneID, zoneName)
	}
	if response.HostedZone.Config != nil && response.HostedZone.Config.PrivateZone {
		return fmt.Errorf("customer hosted zone %s must be public", zoneID)
	}
	dnsOutput, err := d.runner.Run(ctx, Command{Name: "dig", Args: []string{"+short", "NS", normalizeDNSName(zoneName)}})
	if err != nil {
		return fmt.Errorf("resolve public delegation for %s: %w", zoneName, err)
	}
	configured := []string(nil)
	if response.DelegationSet != nil {
		configured = normalizedDNSSet(response.DelegationSet.NameServers)
	}
	observed := normalizedDNSSet(strings.Fields(string(dnsOutput)))
	if len(configured) == 0 || !slices.Equal(configured, observed) {
		return fmt.Errorf("public delegation for %s does not match hosted zone %s", zoneName, zoneID)
	}
	return nil
}

//nolint:gocritic // PCA validation receives the immutable activation contract by value.
func (d *AWSDriver) validatePCA(ctx context.Context, input Input) error {
	clients, err := d.clients(ctx, input.CustomerProfile, input.Region)
	if err != nil {
		return fmt.Errorf("read customer PCA: %w", err)
	}
	response, err := clients.acmpca.DescribeCertificateAuthority(ctx, &acmpca.DescribeCertificateAuthorityInput{
		CertificateAuthorityArn: awssdk.String(input.PCAARN),
	})
	if err != nil {
		return fmt.Errorf("read customer PCA: %w", err)
	}
	authority := response.CertificateAuthority
	if authority == nil || awssdk.ToString(authority.Arn) != input.PCAARN || authority.Status != acmpcatypes.CertificateAuthorityStatusActive {
		return fmt.Errorf("customer PCA must be the configured ACTIVE certificate authority")
	}
	return nil
}

//nolint:gocritic // Zone validation receives the immutable activation contract by value.
func (d *AWSDriver) validateAvailabilityZones(ctx context.Context, input Input) error {
	if len(input.AvailabilityZones) < 2 {
		return errors.New("BYOC customer handoff requires at least two availability zones so node zones can be a strict subset")
	}
	names := make([]string, 0, len(input.AvailabilityZones))
	for _, zone := range input.AvailabilityZones {
		names = append(names, zone.Name)
	}
	clients, err := d.clients(ctx, input.CustomerProfile, input.Region)
	if err != nil {
		return fmt.Errorf("read customer availability zones: %w", err)
	}
	response, err := clients.ec2.DescribeAvailabilityZones(ctx, &ec2.DescribeAvailabilityZonesInput{ZoneNames: names})
	if err != nil {
		return fmt.Errorf("read customer availability zones: %w", err)
	}
	observed := make(map[string]string, len(response.AvailabilityZones))
	for _, zone := range response.AvailabilityZones {
		if zone.State == ec2types.AvailabilityZoneStateAvailable {
			observed[awssdk.ToString(zone.ZoneName)] = awssdk.ToString(zone.ZoneId)
		}
	}
	for _, expected := range input.AvailabilityZones {
		if observed[expected.Name] != expected.ID {
			return fmt.Errorf(
				"customer availability zone %s resolves to %q, want physical zone id %q",
				expected.Name,
				observed[expected.Name],
				expected.ID,
			)
		}
	}
	return nil
}

type ecrRepositoryDescription struct {
	Name string
	ARN  string
}

//nolint:gocritic // Network validation receives the immutable activation contract by value.
func (d *AWSDriver) validateControlPlaneNetwork(ctx context.Context, input Input) (controlPlaneVPCID, subnetAZID string, err error) {
	clients, err := d.clients(ctx, input.ControlPlaneProfile, input.Region)
	if err != nil {
		return "", "", fmt.Errorf("read control-plane remote instance: %w", err)
	}
	instances, err := clients.ec2.DescribeInstances(ctx, &ec2.DescribeInstancesInput{InstanceIds: []string{input.RemoteInstanceID}})
	if err != nil {
		return "", "", fmt.Errorf("read control-plane remote instance: %w", err)
	}
	var remoteInstances []ec2types.Instance
	for _, reservation := range instances.Reservations {
		remoteInstances = append(remoteInstances, reservation.Instances...)
	}
	if len(remoteInstances) != 1 || awssdk.ToString(remoteInstances[0].InstanceId) != input.RemoteInstanceID {
		return "", "", fmt.Errorf("control-plane remote instance %s did not resolve to exactly one instance", input.RemoteInstanceID)
	}
	remoteInstance := remoteInstances[0]
	remoteState := ec2types.InstanceStateName("")
	if remoteInstance.State != nil {
		remoteState = remoteInstance.State.Name
	}
	if remoteState != ec2types.InstanceStateNameRunning {
		return "", "", fmt.Errorf("control-plane remote instance %s is %q, want running", input.RemoteInstanceID, remoteState)
	}
	if awssdk.ToString(remoteInstance.SubnetId) != input.RemoteSubnetID {
		return "", "", fmt.Errorf(
			"control-plane remote instance %s is in subnet %q, want configured remote subnet %q",
			input.RemoteInstanceID,
			awssdk.ToString(remoteInstance.SubnetId),
			input.RemoteSubnetID,
		)
	}
	if awssdk.ToString(remoteInstance.VpcId) == "" {
		return "", "", fmt.Errorf("control-plane remote instance %s has no VPC", input.RemoteInstanceID)
	}
	if !slices.ContainsFunc(remoteInstance.SecurityGroups, func(group ec2types.GroupIdentifier) bool {
		return awssdk.ToString(group.GroupId) == input.RemoteSecurityGroupID
	}) {
		return "", "", fmt.Errorf(
			"control-plane remote instance %s is not attached to security group %s",
			input.RemoteInstanceID,
			input.RemoteSecurityGroupID,
		)
	}

	subnets, err := clients.ec2.DescribeSubnets(ctx, &ec2.DescribeSubnetsInput{SubnetIds: []string{input.RemoteSubnetID}})
	if err != nil {
		return "", "", fmt.Errorf("read control-plane remote subnet: %w", err)
	}
	if len(subnets.Subnets) != 1 || awssdk.ToString(subnets.Subnets[0].SubnetId) != input.RemoteSubnetID ||
		subnets.Subnets[0].State != ec2types.SubnetStateAvailable {
		return "", "", fmt.Errorf("control-plane remote subnet %s is not available", input.RemoteSubnetID)
	}
	subnet := subnets.Subnets[0]
	if awssdk.ToString(subnet.VpcId) != awssdk.ToString(remoteInstance.VpcId) {
		return "", "", fmt.Errorf(
			"control-plane remote instance %s reports VPC %s but subnet %s belongs to VPC %s",
			input.RemoteInstanceID,
			awssdk.ToString(remoteInstance.VpcId),
			awssdk.ToString(subnet.SubnetId),
			awssdk.ToString(subnet.VpcId),
		)
	}
	remoteAvailabilityZone := ""
	if remoteInstance.Placement != nil {
		remoteAvailabilityZone = awssdk.ToString(remoteInstance.Placement.AvailabilityZone)
	}
	if awssdk.ToString(subnet.AvailabilityZone) != remoteAvailabilityZone {
		return "", "", fmt.Errorf(
			"control-plane remote instance %s reports availability zone %s but subnet %s belongs to %s",
			input.RemoteInstanceID,
			remoteAvailabilityZone,
			awssdk.ToString(subnet.SubnetId),
			awssdk.ToString(subnet.AvailabilityZone),
		)
	}
	zoneAllowed := false
	for _, zone := range input.AvailabilityZones {
		if zone.ID == awssdk.ToString(subnet.AvailabilityZoneId) {
			zoneAllowed = true
			break
		}
	}
	if !zoneAllowed {
		return "", "", fmt.Errorf(
			"control-plane subnet physical zone %s is absent from the BYOC customer handoff",
			awssdk.ToString(subnet.AvailabilityZoneId),
		)
	}
	groups, err := clients.ec2.DescribeSecurityGroups(ctx, &ec2.DescribeSecurityGroupsInput{
		GroupIds: []string{input.RemoteSecurityGroupID},
	})
	if err != nil {
		return "", "", fmt.Errorf("read control-plane remote security group: %w", err)
	}
	if len(groups.SecurityGroups) != 1 || awssdk.ToString(groups.SecurityGroups[0].GroupId) != input.RemoteSecurityGroupID ||
		awssdk.ToString(groups.SecurityGroups[0].VpcId) != awssdk.ToString(subnet.VpcId) {
		return "", "", fmt.Errorf(
			"control-plane remote security group %s is not in VPC %s",
			input.RemoteSecurityGroupID,
			awssdk.ToString(subnet.VpcId),
		)
	}
	return awssdk.ToString(subnet.VpcId), awssdk.ToString(subnet.AvailabilityZoneId), nil
}

//nolint:gocritic // Baseline validation receives the immutable activation contract by value.
func (d *AWSDriver) validateBaselineKubernetes(ctx context.Context, input Input) error {
	// Reported per namespace rather than as one combined get: both are Dev Full
	// baseline resources, and a session created before cp-byoc became a baseline
	// will be missing only that one. Naming it is the difference between an
	// actionable error and a kubectl failure the operator has to re-run by hand.
	for _, namespace := range []string{globalBaselineNamespace, byocRuntimeNamespace} {
		_, err := d.kubernetesRunner.Run(ctx, Command{
			Name: "kubectl",
			Args: kubeArgs(input, "get", "namespace", namespace, "-o", "name"),
		})
		if err != nil {
			return fmt.Errorf(
				"observe Dev Full baseline namespace %s (applied by the session Tiltfile): %w",
				namespace, err,
			)
		}
	}
	return nil
}

//nolint:gocritic // Runtime validation receives the immutable activation contract by value.
func (d *AWSDriver) validateBYOCRuntimeNamespace(ctx context.Context, input Input, journal *Journal) error {
	for _, expectedName := range byocRuntimeNamespaces {
		output, err := d.kubernetesRunner.Run(ctx, Command{
			Name: "kubectl",
			Args: kubeArgs(input, "get", "namespace", expectedName, "-o", "json"),
		})
		if err != nil {
			if isCommandNotFound(output, err) {
				continue
			}
			return fmt.Errorf("observe BYOC runtime namespace %s: %w", expectedName, err)
		}
		var namespace runtimeObject
		if err := json.Unmarshal(output, &namespace); err != nil {
			return fmt.Errorf("parse BYOC runtime namespace %s: %w", expectedName, err)
		}
		record := (*StageRecord)(nil)
		if journal != nil {
			record = journal.Stages[StageRegionalRuntime]
		}
		ownedIntent := record != nil && (record.State == StageRunning || record.State == StageFailed || record.State == StageDone)
		ownedNamespace := namespace.Metadata.Name == expectedName &&
			namespace.Metadata.Labels["app.kubernetes.io/managed-by"] == "cp9-byoc" &&
			namespace.Metadata.Annotations["cp9.tidbcloud.com/owner"] == input.Owner &&
			namespace.Metadata.Annotations["cp9.tidbcloud.com/session"] == input.Session
		if expectedName == byocRuntimeNamespace && isBYOCBaselineNamespace(namespace.Metadata) {
			continue
		}
		if !ownedIntent || !ownedNamespace {
			return fmt.Errorf("foreign BYOC runtime namespace %s already exists; CP9 will not adopt or update it", expectedName)
		}
	}
	return nil
}

func isBYOCBaselineNamespace(metadata runtimeObjectMetadata) bool {
	return metadata.Name == byocRuntimeNamespace &&
		metadata.Labels["app.kubernetes.io/managed-by"] == "cp9" &&
		metadata.Labels["cp9.tidbcloud.com/baseline"] == "dev-full"
}

type selectorSlotReservation struct {
	APIVersion            string `json:"apiVersion"`
	Provider              string `json:"provider"`
	Region                string `json:"region"`
	ServicePlan           string `json:"service_plan"`
	TenantID              string `json:"tenant_id"`
	ControlPlaneAccountID string `json:"control_plane_account_id"`
	CustomerAccountID     string `json:"customer_account_id"`
	Session               string `json:"session"`
	Owner                 string `json:"owner"`
	Machine               string `json:"machine"`
	Fingerprint           string `json:"fingerprint"`
	ExpiresAt             string `json:"expires_at,omitempty"`
	Version               int64  `json:"-"`
}

//nolint:gocritic // Selector-slot naming copies the immutable activation contract before hashing it.
func selectorSlotParameterName(input Input) string {
	tenantID := strings.TrimSpace(input.TenantID)
	if parsed, err := strconv.ParseUint(tenantID, 10, 64); err == nil && parsed > 0 {
		tenantID = strconv.FormatUint(parsed, 10)
	}
	canonical := strings.Join([]string{"aws", strings.TrimSpace(input.Region), "BYOC", tenantID}, "\x00")
	digest := sha256.Sum256([]byte(canonical))
	return selectorSlotParameterPrefix + hex.EncodeToString(digest[:])
}

//nolint:gocritic // Selector-slot serialization copies the immutable activation contract before persistence.
func selectorSlotValueJSON(input Input, fingerprint string) string {
	data, _ := json.Marshal(selectorSlotReservation{
		APIVersion: selectorSlotAPIVersion, Provider: "aws", Region: input.Region, ServicePlan: "BYOC",
		TenantID: input.TenantID, ControlPlaneAccountID: input.ControlPlaneAccountID,
		CustomerAccountID: input.CustomerAccountID, Session: input.Session, Owner: input.Owner,
		Machine: input.Machine, Fingerprint: fingerprint, ExpiresAt: input.LeaseExpiresAt,
	})
	return string(data)
}

//nolint:gocritic // Selector-slot comparison receives the immutable activation contract by value.
func selectorSlotBoundToInput(slot *selectorSlotReservation, input Input, fingerprint string) bool {
	return slot != nil &&
		slot.APIVersion == selectorSlotAPIVersion &&
		slot.Provider == "aws" &&
		slot.Region == input.Region &&
		slot.ServicePlan == "BYOC" &&
		slot.TenantID == input.TenantID &&
		slot.ControlPlaneAccountID == input.ControlPlaneAccountID &&
		slot.CustomerAccountID == input.CustomerAccountID &&
		slot.Session == input.Session &&
		slot.Owner == input.Owner &&
		slot.Machine == input.Machine &&
		slot.Fingerprint == fingerprint
}

type accountLease struct {
	APIVersion  string `json:"apiVersion"`
	Session     string `json:"session"`
	Owner       string `json:"owner"`
	Machine     string `json:"machine"`
	Fingerprint string `json:"fingerprint"`
	ExpiresAt   string `json:"expires_at,omitempty"`
	Version     int64  `json:"-"`
}

//nolint:gocritic // Lease serialization copies the immutable activation contract before persistence.
func leaseValueJSON(input Input, fingerprint string) string {
	data, _ := json.Marshal(accountLease{
		APIVersion:  accountLeaseAPIVersion,
		Session:     input.Session,
		Owner:       input.Owner,
		Machine:     input.Machine,
		Fingerprint: fingerprint,
		ExpiresAt:   input.LeaseExpiresAt,
	})
	return string(data)
}

//nolint:gocritic // Lease comparison receives the immutable activation contract by value.
func leaseBoundToInput(lease *accountLease, input Input, fingerprint string) bool {
	return lease != nil &&
		lease.APIVersion == accountLeaseAPIVersion &&
		lease.Session == input.Session &&
		lease.Owner == input.Owner &&
		lease.Machine == input.Machine &&
		lease.Fingerprint == fingerprint
}

//nolint:gocritic,nilnil // Input is immutable; nil is the read-only representation of an unreserved selector slot.
func (d *AWSDriver) readSelectorSlot(ctx context.Context, input Input) (*selectorSlotReservation, error) {
	name := selectorSlotParameterName(input)
	clients, err := d.clients(ctx, input.ControlPlaneProfile, accountLeaseRegion(input))
	if err != nil {
		return nil, err
	}
	response, err := clients.ssm.GetParameter(ctx, &ssm.GetParameterInput{Name: awssdk.String(name)})
	if err != nil {
		if isAWSNotFound(err) {
			return nil, nil
		}
		return nil, err
	}
	if response.Parameter == nil {
		return nil, errors.New("BYOC selector-slot response omitted parameter")
	}
	if awssdk.ToString(response.Parameter.Name) != name {
		return nil, fmt.Errorf("unexpected BYOC selector-slot parameter %q", awssdk.ToString(response.Parameter.Name))
	}
	var slot selectorSlotReservation
	if err := json.Unmarshal([]byte(awssdk.ToString(response.Parameter.Value)), &slot); err != nil {
		return nil, fmt.Errorf("parse control-plane BYOC selector-slot value: %w", err)
	}
	slot.Version = response.Parameter.Version
	return &slot, nil
}

//nolint:gocritic,nilnil // Input is immutable; a nil lease is the read-only representation of an unleased account.
func (d *AWSDriver) readLease(ctx context.Context, input Input) (*accountLease, error) {
	clients, err := d.clients(ctx, input.CustomerProfile, accountLeaseRegion(input))
	if err != nil {
		return nil, err
	}
	response, err := clients.ssm.GetParameter(ctx, &ssm.GetParameterInput{Name: awssdk.String(accountLeaseParameterName)})
	if err != nil {
		if isAWSNotFound(err) {
			return nil, nil
		}
		return nil, err
	}
	if response.Parameter == nil {
		return nil, errors.New("customer account lease response omitted parameter")
	}
	if awssdk.ToString(response.Parameter.Name) != accountLeaseParameterName {
		return nil, fmt.Errorf("unexpected customer account lease parameter %q", awssdk.ToString(response.Parameter.Name))
	}
	var lease accountLease
	if err := json.Unmarshal([]byte(awssdk.ToString(response.Parameter.Value)), &lease); err != nil {
		return nil, fmt.Errorf("parse customer account lease value: %w", err)
	}
	lease.Version = response.Parameter.Version
	return &lease, nil
}

//nolint:gocritic // Lease acquisition receives the immutable activation contract by value.
func (d *AWSDriver) acquireLease(ctx context.Context, input Input, journal *Journal) (*StageResult, error) {
	fingerprint := ""
	if journal != nil {
		fingerprint = journal.Fingerprint
	}
	if fingerprint == "" {
		var err error
		fingerprint, err = Fingerprint(input)
		if err != nil {
			return nil, err
		}
	}
	selectorSlot, err := d.acquireSelectorSlot(ctx, input, fingerprint)
	if err != nil {
		return nil, err
	}
	value := leaseValueJSON(input, fingerprint)
	clients, err := d.clients(ctx, input.CustomerProfile, accountLeaseRegion(input))
	if err != nil {
		return nil, err
	}
	if _, err := clients.ssm.PutParameter(ctx, &ssm.PutParameterInput{
		Name:      awssdk.String(accountLeaseParameterName),
		Type:      ssmtypes.ParameterTypeString,
		Value:     awssdk.String(value),
		Tier:      ssmtypes.ParameterTierStandard,
		Overwrite: awssdk.Bool(false),
		Tags: []ssmtypes.Tag{
			{Key: awssdk.String("cp9-owner"), Value: awssdk.String(input.Owner)},
			{Key: awssdk.String("cp9-session"), Value: awssdk.String(input.Session)},
		},
	}); err != nil {
		lease, readErr := d.readLease(ctx, input)
		if readErr != nil || !leaseBoundToInput(lease, input, fingerprint) {
			return nil, fmt.Errorf("atomically acquire customer account lease: %w", err)
		}
	}
	lease, err := d.readLease(ctx, input)
	if err != nil {
		return nil, err
	}
	if !leaseBoundToInput(lease, input, fingerprint) {
		return nil, errors.New("customer account lease was not bound to this session")
	}
	return &StageResult{Artifacts: map[string]string{
		"parameter_name":          accountLeaseParameterName,
		"region":                  accountLeaseRegion(input),
		"version":                 fmt.Sprint(lease.Version),
		"selector_parameter_name": selectorSlotParameterName(input),
		"selector_region":         accountLeaseRegion(input),
		"selector_version":        fmt.Sprint(selectorSlot.Version),
	}}, nil
}

//nolint:gocritic // Selector-slot acquisition receives the immutable activation contract by value.
func (d *AWSDriver) acquireSelectorSlot(
	ctx context.Context,
	input Input,
	fingerprint string,
) (*selectorSlotReservation, error) {
	name := selectorSlotParameterName(input)
	clients, err := d.clients(ctx, input.ControlPlaneProfile, accountLeaseRegion(input))
	if err != nil {
		return nil, err
	}
	if _, err := clients.ssm.PutParameter(ctx, &ssm.PutParameterInput{
		Name:      awssdk.String(name),
		Type:      ssmtypes.ParameterTypeString,
		Value:     awssdk.String(selectorSlotValueJSON(input, fingerprint)),
		Tier:      ssmtypes.ParameterTierStandard,
		Overwrite: awssdk.Bool(false),
		Tags: []ssmtypes.Tag{
			{Key: awssdk.String("cp9-owner"), Value: awssdk.String(input.Owner)},
			{Key: awssdk.String("cp9-session"), Value: awssdk.String(input.Session)},
		},
	}); err != nil {
		slot, readErr := d.readSelectorSlot(ctx, input)
		if readErr != nil || !selectorSlotBoundToInput(slot, input, fingerprint) {
			return nil, fmt.Errorf(
				"atomically reserve BYOC selector slot for tenant %s, aws/%s, and service plan BYOC: %w",
				input.TenantID,
				input.Region,
				err,
			)
		}
	}
	slot, err := d.readSelectorSlot(ctx, input)
	if err != nil {
		return nil, err
	}
	if !selectorSlotBoundToInput(slot, input, fingerprint) {
		return nil, errors.New("BYOC selector slot was not bound to this session")
	}
	return slot, nil
}

// accountLeaseRegion anchors both SSM coordination records in one fixed region
// per AWS partition. Every target region in the same customer account therefore
// contends on one account-global lease, while selector peers contend across
// customer accounts.
//
//nolint:gocritic // Partition selection receives the immutable activation contract by value.
func accountLeaseRegion(input Input) string {
	switch {
	case strings.HasPrefix(input.PCAARN, "arn:aws-cn:"):
		return "cn-north-1"
	case strings.HasPrefix(input.PCAARN, "arn:aws-us-gov:"):
		return "us-gov-west-1"
	case strings.HasPrefix(input.PCAARN, "arn:aws-iso:"):
		return "us-iso-east-1"
	case strings.HasPrefix(input.PCAARN, "arn:aws-iso-b:"):
		return "us-isob-east-1"
	default:
		return "us-east-1"
	}
}

type cloudFormationParameter struct {
	ParameterKey   string
	ParameterValue string
}

type cloudFormationTag struct {
	Key   string
	Value string
}

type cloudFormationStack struct {
	StackID      string
	StackName    string
	CreationTime string
	StackStatus  string
	Parameters   []cloudFormationParameter
	Tags         []cloudFormationTag
}

//nolint:gocritic // Stack observation receives the immutable activation contract by value.
func (d *AWSDriver) describeStack(ctx context.Context, input Input, nameOrID string) (*cloudFormationStack, error) {
	clients, err := d.clients(ctx, input.CustomerProfile, input.Region)
	if err != nil {
		return nil, err
	}
	response, err := clients.cfn.DescribeStacks(ctx, &cloudformation.DescribeStacksInput{StackName: awssdk.String(nameOrID)})
	if err != nil {
		return nil, err
	}
	if len(response.Stacks) != 1 {
		return nil, fmt.Errorf("CloudFormation describe-stacks returned %d stacks for %s", len(response.Stacks), nameOrID)
	}
	observed := response.Stacks[0]
	stack := &cloudFormationStack{
		StackID:     awssdk.ToString(observed.StackId),
		StackName:   awssdk.ToString(observed.StackName),
		StackStatus: string(observed.StackStatus),
		Parameters:  make([]cloudFormationParameter, 0, len(observed.Parameters)),
		Tags:        make([]cloudFormationTag, 0, len(observed.Tags)),
	}
	if observed.CreationTime != nil {
		stack.CreationTime = observed.CreationTime.UTC().Format(time.RFC3339Nano)
	}
	for _, parameter := range observed.Parameters {
		stack.Parameters = append(stack.Parameters, cloudFormationParameter{
			ParameterKey: awssdk.ToString(parameter.ParameterKey), ParameterValue: awssdk.ToString(parameter.ParameterValue),
		})
	}
	for _, tag := range observed.Tags {
		stack.Tags = append(stack.Tags, cloudFormationTag{Key: awssdk.ToString(tag.Key), Value: awssdk.ToString(tag.Value)})
	}
	return stack, nil
}

//nolint:gocritic // Stack validation receives the immutable activation contract by value.
func (d *AWSDriver) validatePreflightStack(ctx context.Context, input Input, journal *Journal, stackName string) error {
	stack, err := d.describeStack(ctx, input, stackName)
	if err != nil {
		if isAWSNotFound(err) {
			if input.ReuseAccountSetup {
				return fmt.Errorf("customer-owned BYOC setup stack %s is required when reuse_existing_stacks is true", stackName)
			}
			return nil
		}
		return fmt.Errorf("observe BYOC setup stack %s: %w", stackName, err)
	}
	return d.validateObservedPreflightStack(ctx, input, journal, stackName, stack)
}

func (d *AWSDriver) validateObservedPreflightStack(
	ctx context.Context,
	input Input,
	journal *Journal,
	stackName string,
	stack *cloudFormationStack,
) error {
	var record *StageRecord
	if journal != nil {
		record = journal.Stages[StageAccountSetup]
	}
	if input.ReuseAccountSetup {
		if err := validateStackIdentity(input, stackName, stack); err != nil {
			return err
		}
		if record != nil {
			boundID := record.Artifacts["stack."+stackName+".id"]
			creationTime := record.Artifacts["stack."+stackName+".creation_time"]
			if boundID != "" && (creationTime == "" || stack.StackID != boundID || !sameStackCreationTime(stack.CreationTime, creationTime)) {
				return fmt.Errorf("retained BYOC setup stack %s changed identity or creation time", stackName)
			}
			if retained := record.Artifacts[accountSetupRetainedKey]; retained != "" && retained != "true" {
				return fmt.Errorf("BYOC setup stack %s journal ownership mode changed", stackName)
			}
		}
		if err := validateRetainedStackParameters(input, stackName, stack); err != nil {
			return err
		}
		return nil
	}
	if record == nil {
		return fmt.Errorf("foreign BYOC setup stack %s already exists; CP9 will not adopt or update it", stackName)
	}
	boundID := record.Artifacts["stack."+stackName+".id"]
	if boundID != "" {
		creationTime := record.Artifacts["stack."+stackName+".creation_time"]
		if creationTime == "" || stack.StackID != boundID || !sameStackCreationTime(stack.CreationTime, creationTime) {
			return fmt.Errorf("bound BYOC setup stack %s changed identity or creation time", stackName)
		}
		if err := validateStackIdentity(input, stackName, stack); err != nil {
			return err
		}
		if err := validateStackParameters(input, stackName, stack); err != nil {
			return err
		}
		return d.validateStackTemplate(ctx, input, stackName, stack.StackID)
	}
	if record.State != StageRunning && record.State != StageFailed {
		return fmt.Errorf("foreign BYOC setup stack %s already exists; CP9 will not adopt or update it", stackName)
	}
	observedAt, err := d.validateStackCreateIntent(input, record, stackName)
	if err != nil {
		return err
	}
	createdAt, err := time.Parse(time.RFC3339Nano, stack.CreationTime)
	if err != nil || createdAt.Before(observedAt.Add(-time.Minute)) {
		return fmt.Errorf("unbound BYOC setup stack %s predates this activation create intent", stackName)
	}
	if err := validateStackIdentity(input, stackName, stack); err != nil {
		return err
	}
	if err := validateStackParameters(input, stackName, stack); err != nil {
		return err
	}
	if err := d.validateStackTemplate(ctx, input, stackName, stack.StackID); err != nil {
		return err
	}
	return nil
}

func (d *AWSDriver) validateStackTemplate(ctx context.Context, input Input, stackName, stackID string) error {
	remoteDigest, err := d.stackTemplateDigest(ctx, input, stackName, stackID)
	if err != nil {
		return err
	}
	localName := templateFileForStack(stackName)
	localBody, err := d.bundle.Read(localName)
	if err != nil {
		return err
	}
	localDigest, err := canonicalTemplateDigest(localBody)
	if err != nil {
		return fmt.Errorf("canonicalize pinned template %s: %w", localName, err)
	}
	if localDigest != remoteDigest {
		return fmt.Errorf("BYOC setup stack %s template does not match pinned bundle", stackName)
	}
	return nil
}

func (d *AWSDriver) stackTemplateDigest(ctx context.Context, input Input, stackName, stackID string) (string, error) {
	clients, err := d.clients(ctx, input.CustomerProfile, input.Region)
	if err != nil {
		return "", fmt.Errorf("read template for BYOC setup stack %s: %w", stackName, err)
	}
	response, err := clients.cfn.GetTemplate(ctx, &cloudformation.GetTemplateInput{
		StackName: awssdk.String(stackID), TemplateStage: cloudformationtypes.TemplateStageOriginal,
	})
	if err != nil {
		return "", fmt.Errorf("read template for BYOC setup stack %s: %w", stackName, err)
	}
	remoteBody := []byte(awssdk.ToString(response.TemplateBody))
	remoteDigest, err := canonicalTemplateDigest(remoteBody)
	if err != nil {
		return "", fmt.Errorf("canonicalize live template for stack %s: %w", stackName, err)
	}
	return remoteDigest, nil
}

func validateStackIdentity(input Input, stackName string, stack *cloudFormationStack) error {
	if stack == nil || stack.StackName != stackName {
		return fmt.Errorf("BYOC setup stack %s has an unexpected name", stackName)
	}
	parts := strings.SplitN(stack.StackID, ":", 6)
	if len(parts) != 6 || parts[0] != "arn" || parts[2] != "cloudformation" ||
		parts[3] != input.Region || parts[4] != input.CustomerAccountID ||
		!strings.HasPrefix(parts[5], "stack/"+stackName+"/") {
		return fmt.Errorf("BYOC setup stack %s has an unexpected account, region, or ARN", stackName)
	}
	return nil
}

func sameStackCreationTime(actual, expected string) bool {
	actualTime, actualErr := time.Parse(time.RFC3339Nano, actual)
	expectedTime, expectedErr := time.Parse(time.RFC3339Nano, expected)
	return actualErr == nil && expectedErr == nil && actualTime.Equal(expectedTime)
}

func canonicalTemplateDigest(data []byte) (string, error) {
	var document yaml.Node
	if err := yaml.Unmarshal(data, &document); err != nil {
		return "", err
	}
	value, err := canonicalTemplateNode(&document)
	if err != nil {
		return "", err
	}
	canonical, err := json.Marshal(value)
	if err != nil {
		return "", err
	}
	digest := sha256.Sum256(canonical)
	return hex.EncodeToString(digest[:]), nil
}

type canonicalTemplateEntry struct {
	Key   any `json:"key"`
	Value any `json:"value"`
}

// canonicalTemplateNode preserves YAML tag identity because CloudFormation's
// short-form intrinsic functions (!Sub, !Ref, !GetAtt, and peers) are semantic,
// not presentation. Mapping order and comments remain intentionally ignored.
func canonicalTemplateNode(node *yaml.Node) (any, error) {
	if node == nil {
		return nil, errors.New("template contains a nil YAML node")
	}
	switch node.Kind {
	case yaml.DocumentNode:
		if len(node.Content) != 1 {
			return nil, errors.New("template must contain exactly one YAML document")
		}
		return canonicalTemplateNode(node.Content[0])
	case yaml.MappingNode:
		if len(node.Content)%2 != 0 {
			return nil, errors.New("template contains an invalid YAML mapping")
		}
		entries := make([]canonicalTemplateEntry, 0, len(node.Content)/2)
		for index := 0; index < len(node.Content); index += 2 {
			key, err := canonicalTemplateNode(node.Content[index])
			if err != nil {
				return nil, err
			}
			value, err := canonicalTemplateNode(node.Content[index+1])
			if err != nil {
				return nil, err
			}
			entries = append(entries, canonicalTemplateEntry{Key: key, Value: value})
		}
		slices.SortFunc(entries, func(left, right canonicalTemplateEntry) int {
			leftKey, _ := json.Marshal(left.Key)
			rightKey, _ := json.Marshal(right.Key)
			return bytes.Compare(leftKey, rightKey)
		})
		return map[string]any{"kind": "mapping", "tag": node.Tag, "entries": entries}, nil
	case yaml.SequenceNode:
		items := make([]any, 0, len(node.Content))
		for _, child := range node.Content {
			item, err := canonicalTemplateNode(child)
			if err != nil {
				return nil, err
			}
			items = append(items, item)
		}
		return map[string]any{"kind": "sequence", "tag": node.Tag, "items": items}, nil
	case yaml.ScalarNode:
		return map[string]any{"kind": "scalar", "tag": node.Tag, "value": node.Value}, nil
	case yaml.AliasNode:
		return nil, errors.New("template YAML aliases are not supported by exact canonicalization")
	default:
		return nil, fmt.Errorf("template contains unsupported YAML node kind %d", node.Kind)
	}
}

func validateStackParameters(input Input, stackName string, stack *cloudFormationStack) error {
	if err := validateStackOwnership(input, stackName, stack); err != nil {
		return err
	}
	return validateAccountSetupStackParameters(input, stackName, stack)
}

// validateRetainedStackParameters checks a customer-owned stack CP9 did not
// create. It deliberately asserts less than validateStackParameters: ownership
// tags are not required, the template body is not compared, and only the
// parameters in requiredRetainedStackParameters are pinned. Reuse therefore
// trades template equivalence for customer ownership, which is why it is opt-in.
func validateRetainedStackParameters(input Input, stackName string, stack *cloudFormationStack) error {
	if stack == nil {
		return errors.New("CloudFormation stack is required")
	}
	if !retainedStackStatusIsReusable(stack.StackStatus) {
		return fmt.Errorf("retained BYOC setup stack %s is not in a reusable stable state: %s", stackName, stack.StackStatus)
	}
	actual := make(map[string]string, len(stack.Parameters))
	for _, parameter := range stack.Parameters {
		actual[parameter.ParameterKey] = parameter.ParameterValue
	}
	for key, expected := range requiredRetainedStackParameters(input, stackName) {
		value, ok := actual[key]
		if !ok {
			return fmt.Errorf("retained BYOC setup stack %s is missing required parameter %s", stackName, key)
		}
		if value != expected {
			return fmt.Errorf(
				"retained BYOC setup stack %s parameter %s=%q, but CP9 requires %q for this activation",
				stackName, key, value, expected,
			)
		}
	}
	return nil
}

func retainedStackStatusIsReusable(status string) bool {
	switch status {
	case "CREATE_COMPLETE", "UPDATE_COMPLETE", "UPDATE_ROLLBACK_COMPLETE", "IMPORT_COMPLETE":
		return true
	default:
		return false
	}
}

func requiredRetainedStackParameters(input Input, stackName string) map[string]string {
	switch stackName {
	case accountSetupStackDeploy:
		return map[string]string{
			"GithubRunnerGoogleAccountId": githubRunnerGoogleAccountID,
			"O11yHostedZoneId":            input.O11YHostedZoneID,
		}
	case accountSetupStackDataPlane:
		return map[string]string{
			"HostedZoneId":    input.TiDBHostedZoneID,
			"PCAArn":          input.PCAARN,
			"ClinicAccountId": input.ClinicAccountID,
			// CP9 grants external-DNS through the dataplane manager role instead of
			// the node role, so a retained stack that enables the node-role policy is
			// not compatible with this activation rather than merely different.
			"EnableExternalDNSNodeRolePolicy": "false",
		}
	case accountSetupStackO11Y:
		return map[string]string{"O11yHostedZoneId": input.O11YHostedZoneID}
	default:
		return nil
	}
}

// validateRetainedAccountSetupSemantics is the compensating control for the
// checks validateRetainedStackParameters skips. It proves one thing: the
// dataplane manager role exists under its expected identity and safely trusts
// the control-plane account. It proves nothing about the deploy or o11y stack
// contents, so widening reuse to depend on more of those stacks needs a
// matching proof added here.
func (d *AWSDriver) validateRetainedAccountSetupSemantics(ctx context.Context, input Input) error {
	role, err := d.readIAMRoleWithProfile(ctx, input, input.CustomerProfile, customerDataplaneManagerRoleName)
	if err != nil {
		return fmt.Errorf("read retained customer BYOC DPM role: %w", err)
	}
	if role.RoleName != customerDataplaneManagerRoleName ||
		role.ARN != customerDataplaneManagerRoleARN(input.CustomerAccountID) {
		return fmt.Errorf("retained customer BYOC DPM role has an unexpected identity")
	}
	if !customerDPMTrustAllowsControlPlane(role.AssumeRolePolicyDocument, input.ControlPlaneAccountID) {
		return fmt.Errorf(
			"retained customer BYOC DPM role does not safely allow control-plane account %s",
			input.ControlPlaneAccountID,
		)
	}
	return nil
}

func validateAccountSetupStackParameters(input Input, stackName string, stack *cloudFormationStack) error {
	actual := make(map[string]string, len(stack.Parameters))
	for _, parameter := range stack.Parameters {
		actual[parameter.ParameterKey] = parameter.ParameterValue
	}
	expectedParameters := expectedStackParameters(input, stackName)
	if len(stack.Parameters) != len(expectedParameters) || len(actual) != len(expectedParameters) {
		return fmt.Errorf("BYOC setup stack %s parameter set does not match the pinned official script", stackName)
	}
	for key, expected := range expectedParameters {
		if actual[key] != expected {
			return fmt.Errorf("BYOC setup stack %s parameter %s=%q, want %q", stackName, key, actual[key], expected)
		}
	}
	return nil
}

func validateStackOwnership(input Input, stackName string, stack *cloudFormationStack) error {
	if stack == nil {
		return errors.New("CloudFormation stack is required")
	}
	if !strings.HasSuffix(stack.StackStatus, "_COMPLETE") {
		return fmt.Errorf("BYOC setup stack %s is not complete: %s", stackName, stack.StackStatus)
	}
	tags := make(map[string]string, len(stack.Tags))
	for _, tag := range stack.Tags {
		tags[tag.Key] = tag.Value
	}
	if tags["cp9-owner"] != input.Owner || tags["cp9-session"] != input.Session {
		return fmt.Errorf("BYOC setup stack %s ownership tags do not match this activation", stackName)
	}
	return nil
}

func expectedStackParameters(input Input, stackName string) map[string]string {
	switch stackName {
	case accountSetupStackDeploy:
		return map[string]string{
			"ControlPlaneAccountId":       input.ControlPlaneAccountID,
			"GithubRunnerGoogleAccountId": githubRunnerGoogleAccountID,
			"O11yHostedZoneId":            input.O11YHostedZoneID,
			"AdditionalO11yHostedZoneIds": "",
		}
	case accountSetupStackDataPlane:
		return map[string]string{
			"ControlPlaneAccountId":           input.ControlPlaneAccountID,
			"HostedZoneId":                    input.TiDBHostedZoneID,
			"PCAArn":                          input.PCAARN,
			"ClinicAccountId":                 input.ClinicAccountID,
			"EnableExternalDNSNodeRolePolicy": "false",
			"AdditionalPCAArns":               "",
			"AdditionalHostedZoneIds":         "",
		}
	case accountSetupStackO11Y:
		return map[string]string{
			"O11yHostedZoneId":            input.O11YHostedZoneID,
			"O11yGlobalRoleArns":          o11yGlobalRoleARNs,
			"AdditionalO11yHostedZoneIds": "",
		}
	default:
		return nil
	}
}

func stackIntentPrefix(stackName string) string {
	return "intent." + stackName + "."
}

func expectedStackParametersDigest(input Input, stackName string) string {
	parameters := expectedStackParameters(input, stackName)
	ordered := make([]cloudFormationParameter, 0, len(parameters))
	for key, value := range parameters {
		ordered = append(ordered, cloudFormationParameter{ParameterKey: key, ParameterValue: value})
	}
	return stackParametersDigest(ordered)
}

func stackParametersDigest(parameters []cloudFormationParameter) string {
	ordered := append([]cloudFormationParameter(nil), parameters...)
	slices.SortFunc(ordered, func(left, right cloudFormationParameter) int {
		if left.ParameterKey == right.ParameterKey {
			return strings.Compare(left.ParameterValue, right.ParameterValue)
		}
		return strings.Compare(left.ParameterKey, right.ParameterKey)
	})
	data, _ := json.Marshal(ordered)
	digest := sha256.Sum256(data)
	return hex.EncodeToString(digest[:])
}

func (d *AWSDriver) pinnedStackTemplateDigest(stackName string) (string, error) {
	body, err := d.bundle.Read(templateFileForStack(stackName))
	if err != nil {
		return "", err
	}
	digest, err := canonicalTemplateDigest(body)
	if err != nil {
		return "", fmt.Errorf("canonicalize pinned template for stack %s: %w", stackName, err)
	}
	return digest, nil
}

func (d *AWSDriver) validateStackCreateIntent(
	input Input,
	record *StageRecord,
	stackName string,
) (time.Time, error) {
	intent, err := journaledStackCreateIntent(input, record, stackName)
	if err != nil {
		return time.Time{}, err
	}
	templateDigest, err := d.pinnedStackTemplateDigest(stackName)
	if err != nil {
		return time.Time{}, err
	}
	if intent.TemplateDigest != templateDigest ||
		intent.ParametersDigest != expectedStackParametersDigest(input, stackName) {
		return time.Time{}, fmt.Errorf("unbound BYOC setup stack %s does not match its durable create intent", stackName)
	}
	return intent.ObservedAt, nil
}

type stackCreateIntent struct {
	ObservedAt       time.Time
	TemplateDigest   string
	ParametersDigest string
}

func journaledStackCreateIntent(input Input, record *StageRecord, stackName string) (stackCreateIntent, error) {
	if record == nil || record.Artifacts == nil {
		return stackCreateIntent{}, fmt.Errorf("unbound BYOC setup stack %s has no durable create intent", stackName)
	}
	prefix := stackIntentPrefix(stackName)
	intent := stackCreateIntent{
		TemplateDigest:   record.Artifacts[prefix+"template_digest"],
		ParametersDigest: record.Artifacts[prefix+"parameters_digest"],
	}
	if record.Artifacts[prefix+"expected_id"] != "ABSENT" ||
		record.Artifacts[prefix+"account_id"] != input.CustomerAccountID ||
		record.Artifacts[prefix+"region"] != input.Region ||
		!validSHA256Hex(intent.TemplateDigest) ||
		!validSHA256Hex(intent.ParametersDigest) {
		return stackCreateIntent{}, fmt.Errorf("unbound BYOC setup stack %s does not match its durable create intent", stackName)
	}
	observedAt, err := time.Parse(time.RFC3339Nano, record.Artifacts[prefix+"observed_at"])
	if err != nil {
		return stackCreateIntent{}, fmt.Errorf("unbound BYOC setup stack %s has an invalid create intent timestamp", stackName)
	}
	intent.ObservedAt = observedAt
	return intent, nil
}

func validSHA256Hex(value string) bool {
	decoded, err := hex.DecodeString(value)
	return err == nil && len(decoded) == sha256.Size
}

func (d *AWSDriver) applyAccountSetup(ctx context.Context, input Input, journal *Journal) (*StageResult, error) {
	if journal == nil || journal.Stages == nil {
		return nil, errors.New("running BYOC activation journal is required for account setup")
	}
	record := journal.Stages[StageAccountSetup]
	if record == nil || record.State != StageRunning || record.StartedAt == nil {
		return nil, errors.New("running BYOC account-setup stage journal is required")
	}
	if record.Artifacts == nil {
		record.Artifacts = make(map[string]string)
	}
	if input.ReuseAccountSetup {
		return d.bindRetainedAccountSetup(ctx, input, record)
	}
	// Repeat the fixed-name reads after both selector and account reservations are held. The global
	// read-only preflight catches ordinary conflicts. The wrapper repeats each
	// observed boundary immediately before its corresponding official deploy.
	boundaryEnv := make([]string, 0, len(accountSetupStackNames))
	for _, stackName := range accountSetupStackNames {
		expectedID := "ABSENT"
		stack, err := d.describeStack(ctx, input, stackName)
		if err != nil {
			if !isAWSNotFound(err) {
				return nil, fmt.Errorf("observe BYOC setup stack %s: %w", stackName, err)
			}
		} else {
			if err := d.validateObservedPreflightStack(ctx, input, journal, stackName, stack); err != nil {
				return nil, err
			}
			if stack.StackStatus == "ROLLBACK_COMPLETE" {
				if record.Artifacts["stack."+stackName+".id"] != "" {
					return nil, fmt.Errorf("bound BYOC setup stack %s is ROLLBACK_COMPLETE and cannot be recreated", stackName)
				}
				if err := d.deleteRollbackCompleteAccountSetupStack(ctx, input, stackName, stack); err != nil {
					return nil, err
				}
			} else {
				expectedID = stack.StackID
			}
		}
		prefix := stackIntentPrefix(stackName)
		if record.Artifacts[prefix+"observed_at"] == "" {
			if expectedID != "ABSENT" {
				return nil, fmt.Errorf("unbound BYOC setup stack %s has no prior durable create intent", stackName)
			}
			templateDigest, digestErr := d.pinnedStackTemplateDigest(stackName)
			if digestErr != nil {
				return nil, digestErr
			}
			record.Artifacts[prefix+"expected_id"] = "ABSENT"
			record.Artifacts[prefix+"observed_at"] = time.Now().UTC().Format(time.RFC3339Nano)
			record.Artifacts[prefix+"account_id"] = input.CustomerAccountID
			record.Artifacts[prefix+"region"] = input.Region
			record.Artifacts[prefix+"template_digest"] = templateDigest
			record.Artifacts[prefix+"parameters_digest"] = expectedStackParametersDigest(input, stackName)
		}
		if _, intentErr := d.validateStackCreateIntent(input, record, stackName); intentErr != nil {
			return nil, intentErr
		}
		boundaryEnv = append(boundaryEnv, expectedStackBoundaryEnv(stackName)+"="+expectedID)
	}
	if err := saveJournal(input.StateDir, journal); err != nil {
		return nil, fmt.Errorf("persist per-stack BYOC account-setup create intents: %w", err)
	}
	digestSegment := strings.TrimPrefix(d.bundle.Digest, "sha256:")
	dir, err := filepath.Abs(filepath.Join(StateDir(input.StateDir), "account-setup", digestSegment))
	if err != nil {
		return nil, fmt.Errorf("resolve pinned BYOC account-setup directory: %w", err)
	}
	if err := d.bundle.Materialize(dir); err != nil {
		return nil, err
	}
	command := Command{
		Name: "bash",
		Args: []string{
			"-c",
			accountSetupCommandWrapper,
			"cp9-byoc-account-setup",
			filepath.Join(dir, accountsetup.ScriptName),
			"--control-plane-id", input.ControlPlaneAccountID,
			"--clinic-id", input.ClinicAccountID,
			"--tidb-hz-id", input.TiDBHostedZoneID,
			"--o11y-hz-id", input.O11YHostedZoneID,
			"--pca-arn", input.PCAARN,
		},
		Dir: dir,
		Env: append([]string{
			"AWS_PROFILE=" + input.CustomerProfile,
			"AWS_REGION=" + input.Region,
			"AWS_DEFAULT_REGION=" + input.Region,
			"AWS_PAGER=",
			"CP9_SETUP_OWNER=" + input.Owner,
			"CP9_SETUP_SESSION=" + input.Session,
			"CP9_SETUP_EXPECTED_CUSTOMER_ACCOUNT_ID=" + input.CustomerAccountID,
		}, boundaryEnv...),
	}
	if _, err := d.runner.Run(ctx, command); err != nil {
		return nil, fmt.Errorf("run pinned official BYOC account setup: %w", err)
	}
	artifacts := cloneArtifacts(record.Artifacts)
	artifacts["bundle_commit"] = d.bundle.UpstreamCommit
	artifacts["bundle_digest"] = d.bundle.Digest
	for _, stackName := range accountSetupStackNames {
		stack, err := d.describeStack(ctx, input, stackName)
		if err != nil {
			return nil, fmt.Errorf("bind BYOC setup stack %s: %w", stackName, err)
		}
		if err := validateStackIdentity(input, stackName, stack); err != nil {
			return nil, err
		}
		if err := validateStackParameters(input, stackName, stack); err != nil {
			return nil, err
		}
		if err := d.validateStackTemplate(ctx, input, stackName, stack.StackID); err != nil {
			return nil, err
		}
		artifacts["stack."+stackName+".id"] = stack.StackID
		artifacts["stack."+stackName+".creation_time"] = stack.CreationTime
	}
	return &StageResult{Artifacts: artifacts}, nil
}

func (d *AWSDriver) bindRetainedAccountSetup(
	ctx context.Context,
	input Input,
	record *StageRecord,
) (*StageResult, error) {
	artifacts := cloneArtifacts(record.Artifacts)
	if artifacts == nil {
		artifacts = make(map[string]string)
	}
	artifacts[accountSetupRetainedKey] = "true"
	artifacts["bundle_commit"] = d.bundle.UpstreamCommit
	artifacts["bundle_digest"] = d.bundle.Digest
	for _, stackName := range accountSetupStackNames {
		stack, err := d.describeStack(ctx, input, stackName)
		if err != nil {
			return nil, fmt.Errorf("bind retained BYOC setup stack %s: %w", stackName, err)
		}
		if err := validateStackIdentity(input, stackName, stack); err != nil {
			return nil, err
		}
		if err := validateRetainedStackParameters(input, stackName, stack); err != nil {
			return nil, err
		}
		artifacts["stack."+stackName+".id"] = stack.StackID
		artifacts["stack."+stackName+".creation_time"] = stack.CreationTime
	}
	if err := d.validateRetainedAccountSetupSemantics(ctx, input); err != nil {
		return nil, err
	}
	return &StageResult{Artifacts: artifacts}, nil
}

// deleteRollbackCompleteAccountSetupStack removes only an unbound stack that
// the durable create intent and live validation have already proven belongs to
// this activation. CloudFormation cannot update a create-time
// ROLLBACK_COMPLETE stack, so an identical official deploy must recreate it.
//
//nolint:gocritic // Input is the immutable activation contract at the mutation boundary.
func (d *AWSDriver) deleteRollbackCompleteAccountSetupStack(
	ctx context.Context,
	input Input,
	stackName string,
	stack *cloudFormationStack,
) error {
	if stack == nil || stack.StackStatus != "ROLLBACK_COMPLETE" || strings.TrimSpace(stack.StackID) == "" {
		return fmt.Errorf("BYOC setup stack %s is not an exact ROLLBACK_COMPLETE recovery candidate", stackName)
	}
	clients, err := d.clients(ctx, input.CustomerProfile, input.Region)
	if err != nil {
		return err
	}
	if _, err := clients.cfn.DeleteStack(
		ctx,
		&cloudformation.DeleteStackInput{StackName: awssdk.String(stack.StackID)},
	); err != nil &&
		!isAWSNotFound(err) {
		return fmt.Errorf("delete rollback-complete BYOC setup stack %s: %w", stackName, err)
	}
	if err := d.waitForAccountSetupStackDeletion(ctx, input, stackName, stack.StackID); err != nil {
		return fmt.Errorf("wait for rollback-complete BYOC setup stack %s deletion: %w", stackName, err)
	}
	return nil
}

func expectedStackBoundaryEnv(stackName string) string {
	switch stackName {
	case accountSetupStackDeploy:
		return "CP9_SETUP_EXPECTED_DEPLOY_STACK_ID"
	case accountSetupStackDataPlane:
		return "CP9_SETUP_EXPECTED_DATAPLANE_STACK_ID"
	case accountSetupStackO11Y:
		return "CP9_SETUP_EXPECTED_O11Y_STACK_ID"
	default:
		return "CP9_SETUP_EXPECTED_UNKNOWN_STACK_ID"
	}
}

type sourceImageManifest struct {
	Source string `json:"source"`
	Digest string `json:"digest"`
}

func (d *AWSDriver) preflightImageSources(
	ctx context.Context,
	input Input,
	journal *Journal,
) (evidence map[string]string, resultErr error) {
	if d.imageClosure == nil {
		return map[string]string{}, nil
	}
	if len(d.imageClosure.Images) == 0 {
		return nil, errors.New("BYOC image closure is empty")
	}
	if input.ImageClosureDigest != d.imageClosure.Digest {
		return nil, fmt.Errorf("BYOC image closure pin mismatch: input %s, discovered %s", input.ImageClosureDigest, d.imageClosure.Digest)
	}
	if journal != nil && journal.Preflight != nil && stageAttempted(journal, StageLease) {
		digests, err := sourceImageDigestsFromPreflight(d.imageClosure, journal)
		if err != nil {
			return nil, fmt.Errorf("existing BYOC activation lacks valid pinned source image manifests: %w", err)
		}
		return sourceImageEvidence(d.imageClosure, digests)
	}
	authDir, err := resetRegistryAuthDirectory(input.StateDir)
	if err != nil {
		return nil, err
	}
	defer func() {
		if err := os.RemoveAll(authDir); err != nil {
			evidence = nil
			resultErr = errors.Join(resultErr, fmt.Errorf("remove temporary BYOC registry credentials: %w", err))
		}
		if err := d.cleanupImageRunner(ctx); err != nil {
			evidence = nil
			resultErr = errors.Join(resultErr, fmt.Errorf("remove remote temporary BYOC registry credentials: %w", err))
		}
	}()
	if err := d.loginSourceECRRegistries(ctx, input, authDir); err != nil {
		return nil, err
	}

	digests := make(map[string]string, len(d.imageClosure.Images))
	for _, image := range d.imageClosure.Images {
		digest, err := d.craneDigest(ctx, image.Source, authDir)
		if err != nil {
			return nil, fmt.Errorf("resolve exact source manifest for %s: %w", image.Source, err)
		}
		if !validSHA256Digest(digest) {
			return nil, fmt.Errorf("source registry returned an invalid digest for %s", image.Source)
		}
		digests[image.Source] = digest
	}
	return sourceImageEvidence(d.imageClosure, digests)
}

func sourceImageEvidence(closure *ImageClosure, digests map[string]string) (map[string]string, error) {
	manifestSetDigest, err := sourceImageManifestSetDigest(closure, digests)
	if err != nil {
		return nil, err
	}
	evidence := map[string]string{
		"image_source_count":   fmt.Sprint(len(closure.Images)),
		"image_sources_digest": manifestSetDigest,
	}
	for _, image := range closure.Images {
		evidence["image."+imageArtifactKey(image.Source)+".source_digest"] = digests[image.Source]
	}
	return evidence, nil
}

func sourceImageDigestsFromPreflight(closure *ImageClosure, journal *Journal) (map[string]string, error) {
	if closure == nil || journal == nil || journal.Preflight == nil {
		return nil, errors.New("BYOC image mirror requires a completed source-manifest preflight")
	}
	if journal.Preflight.Observed["image_source_count"] != fmt.Sprint(len(closure.Images)) {
		return nil, errors.New("BYOC source-manifest preflight image count does not match the pinned closure")
	}
	digests := make(map[string]string, len(closure.Images))
	for _, image := range closure.Images {
		digest := journal.Preflight.Observed["image."+imageArtifactKey(image.Source)+".source_digest"]
		if !validSHA256Digest(digest) {
			return nil, fmt.Errorf("BYOC source-manifest preflight is missing an exact digest for %s", image.Source)
		}
		digests[image.Source] = digest
	}
	manifestSetDigest, err := sourceImageManifestSetDigest(closure, digests)
	if err != nil {
		return nil, err
	}
	if journal.Preflight.Observed["image_sources_digest"] != manifestSetDigest {
		return nil, errors.New("BYOC source-manifest preflight digest does not match its image evidence")
	}
	return digests, nil
}

func sourceImageManifestSetDigest(closure *ImageClosure, digests map[string]string) (string, error) {
	manifests := make([]sourceImageManifest, 0, len(closure.Images))
	for _, image := range closure.Images {
		digest := digests[image.Source]
		if !validSHA256Digest(digest) {
			return "", fmt.Errorf("BYOC source manifest digest is missing or invalid for %s", image.Source)
		}
		manifests = append(manifests, sourceImageManifest{Source: image.Source, Digest: digest})
	}
	slices.SortFunc(manifests, func(a, b sourceImageManifest) int {
		return strings.Compare(a.Source, b.Source)
	})
	data, err := json.Marshal(manifests)
	if err != nil {
		return "", fmt.Errorf("encode BYOC source image manifests: %w", err)
	}
	digest := sha256.Sum256(data)
	return "sha256:" + hex.EncodeToString(digest[:]), nil
}

func resetRegistryAuthDirectory(stateDir string) (string, error) {
	authDir := filepath.Join(StateDir(stateDir), "registry-auth")
	if err := os.RemoveAll(authDir); err != nil {
		return "", fmt.Errorf("remove stale BYOC registry credentials: %w", err)
	}
	if err := os.MkdirAll(authDir, 0o700); err != nil {
		return "", fmt.Errorf("create BYOC registry auth directory: %w", err)
	}
	if err := os.Chmod(authDir, 0o700); err != nil {
		return "", fmt.Errorf("secure BYOC registry auth directory: %w", err)
	}
	return authDir, nil
}

func (d *AWSDriver) loginSourceECRRegistries(ctx context.Context, input Input, authDir string) error {
	loggedIn := make(map[string]struct{})
	for _, image := range d.imageClosure.Images {
		sourceRegistry := imageRegistry(image.Source)
		if sourceRegion := ecrRegistryRegion(sourceRegistry); sourceRegion != "" {
			if _, ok := loggedIn[sourceRegistry]; ok {
				continue
			}
			if err := d.loginECR(ctx, input.CustomerProfile, sourceRegion, sourceRegistry, authDir); err != nil {
				return fmt.Errorf("authenticate source ECR registry %s: %w", sourceRegistry, err)
			}
			loggedIn[sourceRegistry] = struct{}{}
		}
	}
	return nil
}

func (d *AWSDriver) applyImageMirror(
	ctx context.Context,
	input Input,
	journal *Journal,
) (result *StageResult, resultErr error) {
	if d.imageClosure == nil || len(d.imageClosure.Images) == 0 {
		return nil, errors.New("BYOC image closure is required")
	}
	if input.ImageClosureDigest != d.imageClosure.Digest {
		return nil, fmt.Errorf("BYOC image closure pin mismatch: input %s, discovered %s", input.ImageClosureDigest, d.imageClosure.Digest)
	}
	sourceDigests, err := sourceImageDigestsFromPreflight(d.imageClosure, journal)
	if err != nil {
		return nil, err
	}
	authDir, err := resetRegistryAuthDirectory(input.StateDir)
	if err != nil {
		return nil, err
	}
	defer func() {
		if err := os.RemoveAll(authDir); err != nil {
			result = nil
			resultErr = errors.Join(resultErr, fmt.Errorf("remove temporary BYOC registry credentials: %w", err))
		}
		if err := d.cleanupImageRunner(ctx); err != nil {
			result = nil
			resultErr = errors.Join(resultErr, fmt.Errorf("remove remote temporary BYOC registry credentials: %w", err))
		}
	}()

	targetRegistry := fmt.Sprintf("%s.dkr.ecr.%s.amazonaws.com", input.CustomerAccountID, input.ImageSyncRegion)
	if err := d.loginSourceECRRegistries(ctx, input, authDir); err != nil {
		return nil, err
	}
	if err := d.loginECR(ctx, input.CustomerProfile, input.ImageSyncRegion, targetRegistry, authDir); err != nil {
		return nil, fmt.Errorf("authenticate customer ECR registry %s: %w", targetRegistry, err)
	}

	artifacts := map[string]string{
		"closure_digest": d.imageClosure.Digest,
		"image_count":    fmt.Sprint(len(d.imageClosure.Images)),
		"registry":       targetRegistry,
	}
	for _, image := range d.imageClosure.Images {
		sourceDigest := sourceDigests[image.Source]
		if err := d.ensureCustomerECRRepository(ctx, input, image.Repository); err != nil {
			return nil, err
		}
		target := targetRegistry + "/" + image.Repository + ":" + image.Tag
		immutableSource := imageSourceWithoutTag(image.Source) + "@" + sourceDigest
		targetDigest, err := d.craneDigest(ctx, target, authDir)
		if err != nil || targetDigest != sourceDigest {
			if _, err := d.imageRunner.Run(ctx, Command{
				Name: "crane",
				Args: []string{"copy", immutableSource, target},
				Env:  []string{"DOCKER_CONFIG=" + authDir},
			}); err != nil {
				return nil, fmt.Errorf("copy BYOC image %s to customer ECR: %w", image.Source, err)
			}
			targetDigest, err = d.craneDigest(ctx, target, authDir)
			if err != nil {
				return nil, fmt.Errorf("verify customer ECR manifest for %s: %w", image.Source, err)
			}
		}
		if targetDigest != sourceDigest {
			return nil, fmt.Errorf("customer ECR digest mismatch for %s: copied manifest is not the exact source manifest", image.Source)
		}
		key := imageArtifactKey(image.Source)
		artifacts["image."+key+".repository"] = image.Repository
		artifacts["image."+key+".tag"] = image.Tag
		artifacts["image."+key+".digest"] = sourceDigest
	}
	return &StageResult{Artifacts: artifacts}, nil
}

func (d *AWSDriver) revalidateImageMirror(ctx context.Context, input Input, record *StageRecord) (*StageResult, error) {
	if d.imageClosure == nil || record == nil || record.Artifacts["closure_digest"] != d.imageClosure.Digest ||
		input.ImageClosureDigest != d.imageClosure.Digest {
		return nil, ErrNeedsReconcile
	}
	artifacts := map[string]string{
		"closure_digest": d.imageClosure.Digest,
		"image_count":    fmt.Sprint(len(d.imageClosure.Images)),
		"registry":       fmt.Sprintf("%s.dkr.ecr.%s.amazonaws.com", input.CustomerAccountID, input.ImageSyncRegion),
	}
	validatedRepositories := make(map[string]struct{})
	for _, image := range d.imageClosure.Images {
		if _, ok := validatedRepositories[image.Repository]; !ok {
			if err := d.validateCustomerECRRepository(ctx, input, image.Repository); err != nil {
				if errors.Is(err, errCustomerECRRepositoryDrift) || isAWSNotFound(err) {
					return nil, ErrNeedsReconcile
				}
				return nil, err
			}
			validatedRepositories[image.Repository] = struct{}{}
		}
		key := imageArtifactKey(image.Source)
		expectedDigest := record.Artifacts["image."+key+".digest"]
		if !validSHA256Digest(expectedDigest) {
			return nil, ErrNeedsReconcile
		}
		clients, err := d.clients(ctx, input.CustomerProfile, input.ImageSyncRegion)
		if err != nil {
			return nil, err
		}
		response, err := clients.ecr.DescribeImages(ctx, &ecr.DescribeImagesInput{
			RepositoryName: awssdk.String(image.Repository),
			ImageIds:       []ecrtypes.ImageIdentifier{{ImageTag: awssdk.String(image.Tag)}},
		})
		if err != nil {
			if isAWSNotFound(err) {
				return nil, ErrNeedsReconcile
			}
			return nil, fmt.Errorf("observe customer ECR image %s:%s: %w", image.Repository, image.Tag, err)
		}
		if len(response.ImageDetails) != 1 || awssdk.ToString(response.ImageDetails[0].ImageDigest) != expectedDigest {
			return nil, ErrNeedsReconcile
		}
		artifacts["image."+key+".repository"] = image.Repository
		artifacts["image."+key+".tag"] = image.Tag
		artifacts["image."+key+".digest"] = expectedDigest
	}
	return &StageResult{Artifacts: artifacts}, nil
}

func (d *AWSDriver) loginECR(ctx context.Context, profile, region, registry, authDir string) error {
	clients, err := d.clients(ctx, profile, region)
	if err != nil {
		return err
	}
	response, err := clients.ecr.GetAuthorizationToken(ctx, &ecr.GetAuthorizationTokenInput{})
	if err != nil {
		return err
	}
	if len(response.AuthorizationData) != 1 || response.AuthorizationData[0].AuthorizationToken == nil {
		return errors.New("ECR returned an empty temporary password")
	}
	decoded, err := base64.StdEncoding.DecodeString(awssdk.ToString(response.AuthorizationData[0].AuthorizationToken))
	if err != nil {
		return fmt.Errorf("decode ECR authorization token: %w", err)
	}
	_, password, ok := strings.Cut(string(decoded), ":")
	if !ok || password == "" {
		return errors.New("ECR returned an invalid temporary password")
	}
	_, err = d.imageRunner.Run(ctx, Command{
		Name:  "crane",
		Args:  []string{"auth", "login", registry, "--username", "AWS", "--password-stdin"},
		Env:   []string{"DOCKER_CONFIG=" + authDir},
		Stdin: []byte(password),
	})
	return err
}

func (d *AWSDriver) craneDigest(ctx context.Context, reference, authDir string) (string, error) {
	output, err := d.imageRunner.Run(ctx, Command{
		Name: "crane",
		Args: []string{"digest", reference},
		Env:  []string{"DOCKER_CONFIG=" + authDir},
	})
	if err != nil {
		return "", err
	}
	return strings.TrimSpace(string(output)), nil
}

func (d *AWSDriver) cleanupImageRunner(ctx context.Context) error {
	cleaner, ok := d.imageRunner.(interface {
		Cleanup(ctx context.Context) error
	})
	if !ok {
		return nil
	}
	cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
	defer cancel()
	return cleaner.Cleanup(cleanupCtx)
}

func (d *AWSDriver) ensureCustomerECRRepository(ctx context.Context, input Input, repository string) error {
	observed, err := d.readCustomerECRRepository(ctx, input, repository)
	if err != nil && !isAWSNotFound(err) {
		return err
	}
	if observed == nil {
		clients, clientErr := d.clients(ctx, input.CustomerProfile, input.ImageSyncRegion)
		if clientErr != nil {
			return clientErr
		}
		if _, err := clients.ecr.CreateRepository(ctx, &ecr.CreateRepositoryInput{
			RepositoryName: awssdk.String(repository),
			Tags:           []ecrtypes.Tag{{Key: awssdk.String("ManagedBy"), Value: awssdk.String("PingCAP")}},
		}); err != nil {
			return fmt.Errorf("create customer ECR repository %s: %w", repository, err)
		}
		observed, err = d.readCustomerECRRepository(ctx, input, repository)
		if err != nil {
			return fmt.Errorf("read back created customer ECR repository %s: %w", repository, err)
		}
	}
	if err := validateCustomerECRRepositoryIdentity(input, repository, observed); err != nil {
		return err
	}
	managed, err := d.readCustomerECRManagedTag(ctx, input, observed.ARN)
	if err != nil {
		return err
	}
	if managed {
		return nil
	}
	clients, err := d.clients(ctx, input.CustomerProfile, input.ImageSyncRegion)
	if err != nil {
		return err
	}
	if _, err := clients.ecr.TagResource(ctx, &ecr.TagResourceInput{
		ResourceArn: awssdk.String(observed.ARN),
		Tags:        []ecrtypes.Tag{{Key: awssdk.String("ManagedBy"), Value: awssdk.String("PingCAP")}},
	}); err != nil {
		return fmt.Errorf("tag existing customer ECR repository %s: %w", repository, err)
	}
	managed, err = d.readCustomerECRManagedTag(ctx, input, observed.ARN)
	if err != nil {
		return err
	}
	if !managed {
		return fmt.Errorf("customer ECR repository %s did not retain ManagedBy=PingCAP", repository)
	}
	return nil
}

var errCustomerECRRepositoryDrift = errors.New("customer ECR repository contract drift")

func (d *AWSDriver) readCustomerECRRepository(
	ctx context.Context,
	input Input,
	repository string,
) (*ecrRepositoryDescription, error) {
	clients, err := d.clients(ctx, input.CustomerProfile, input.ImageSyncRegion)
	if err != nil {
		return nil, fmt.Errorf("observe customer ECR repository %s: %w", repository, err)
	}
	response, err := clients.ecr.DescribeRepositories(ctx, &ecr.DescribeRepositoriesInput{RepositoryNames: []string{repository}})
	if err != nil {
		return nil, fmt.Errorf("observe customer ECR repository %s: %w", repository, err)
	}
	if len(response.Repositories) != 1 {
		return nil, fmt.Errorf("%w: customer ECR repository %s did not resolve exactly once", errCustomerECRRepositoryDrift, repository)
	}
	return &ecrRepositoryDescription{
		Name: awssdk.ToString(response.Repositories[0].RepositoryName),
		ARN:  awssdk.ToString(response.Repositories[0].RepositoryArn),
	}, nil
}

func validateCustomerECRRepositoryIdentity(input Input, repository string, observed *ecrRepositoryDescription) error {
	parts := strings.SplitN(input.PCAARN, ":", 3)
	if len(parts) != 3 || parts[0] != "arn" || strings.TrimSpace(parts[1]) == "" {
		return fmt.Errorf("derive customer ECR partition from PCA ARN %q", input.PCAARN)
	}
	expectedARN := fmt.Sprintf(
		"arn:%s:ecr:%s:%s:repository/%s",
		parts[1],
		input.ImageSyncRegion,
		input.CustomerAccountID,
		repository,
	)
	if observed == nil || observed.Name != repository || observed.ARN != expectedARN {
		return fmt.Errorf(
			"%w: customer ECR repository %s identity is not %s",
			errCustomerECRRepositoryDrift,
			repository,
			expectedARN,
		)
	}
	return nil
}

func (d *AWSDriver) readCustomerECRManagedTag(
	ctx context.Context,
	input Input,
	repositoryARN string,
) (bool, error) {
	clients, err := d.clients(ctx, input.CustomerProfile, input.ImageSyncRegion)
	if err != nil {
		return false, fmt.Errorf("read customer ECR repository tags for %s: %w", repositoryARN, err)
	}
	response, err := clients.ecr.ListTagsForResource(ctx, &ecr.ListTagsForResourceInput{ResourceArn: awssdk.String(repositoryARN)})
	if err != nil {
		return false, fmt.Errorf("read customer ECR repository tags for %s: %w", repositoryARN, err)
	}
	if len(response.Tags) == 0 {
		return false, nil
	}
	if len(response.Tags) != 1 || awssdk.ToString(response.Tags[0].Key) != "ManagedBy" || awssdk.ToString(response.Tags[0].Value) != "PingCAP" {
		return false, fmt.Errorf(
			"%w: customer ECR repository %s tags are not exactly ManagedBy=PingCAP",
			errCustomerECRRepositoryDrift,
			repositoryARN,
		)
	}
	return true, nil
}

func (d *AWSDriver) validateCustomerECRRepository(ctx context.Context, input Input, repository string) error {
	observed, err := d.readCustomerECRRepository(ctx, input, repository)
	if err != nil {
		return err
	}
	if err := validateCustomerECRRepositoryIdentity(input, repository, observed); err != nil {
		return err
	}
	managed, err := d.readCustomerECRManagedTag(ctx, input, observed.ARN)
	if err != nil {
		return err
	}
	if !managed {
		return fmt.Errorf("%w: customer ECR repository %s is missing ManagedBy=PingCAP", errCustomerECRRepositoryDrift, repository)
	}
	return nil
}

func imageRegistry(source string) string {
	if index := strings.Index(source, "/"); index > 0 {
		return source[:index]
	}
	return ""
}

func ecrRegistryRegion(registry string) string {
	const marker = ".dkr.ecr."
	_, after, ok := strings.Cut(registry, marker)
	if !ok {
		return ""
	}
	remainder := after
	if suffix := strings.Index(remainder, ".amazonaws.com"); suffix > 0 {
		return remainder[:suffix]
	}
	return ""
}

func imageSourceWithoutTag(source string) string {
	lastSlash := strings.LastIndex(source, "/")
	if digestIndex := strings.LastIndex(source, "@sha256:"); digestIndex > lastSlash {
		return source[:digestIndex]
	}
	lastColon := strings.LastIndex(source, ":")
	if lastColon > lastSlash {
		return source[:lastColon]
	}
	return source
}

func validSHA256Digest(value string) bool {
	if !strings.HasPrefix(value, "sha256:") || len(value) != len("sha256:")+64 {
		return false
	}
	_, err := hex.DecodeString(strings.TrimPrefix(value, "sha256:"))
	return err == nil
}

func imageArtifactKey(source string) string {
	digest := sha256.Sum256([]byte(source))
	return hex.EncodeToString(digest[:8])
}

func isAWSNotFound(err error) bool {
	if err == nil {
		return false
	}
	var apiError smithy.APIError
	if errors.As(err, &apiError) {
		switch apiError.ErrorCode() {
		case "ParameterNotFound", "RepositoryNotFoundException", "ImageNotFoundException", "NoSuchEntity",
			"InvalidGroup.NotFound", "ResourceNotFoundException", "NoSuchBucket":
			return true
		case "ValidationError":
			return strings.Contains(strings.ToLower(apiError.ErrorMessage()), "does not exist")
		}
	}
	return false
}

func isCommandNotFound(output []byte, err error) bool {
	if err == nil {
		return false
	}
	text := strings.ToLower(commandFailureText(output, err))
	return strings.Contains(text, "not found") || strings.Contains(text, "notfound")
}

func normalizeHostedZoneID(value string) string {
	return strings.TrimPrefix(strings.TrimSpace(value), "/hostedzone/")
}

func normalizeDNSName(value string) string {
	return strings.ToLower(strings.TrimSuffix(strings.TrimSpace(value), "."))
}

func normalizedDNSSet(values []string) []string {
	normalized := make([]string, 0, len(values))
	for _, value := range values {
		value = normalizeDNSName(value)
		if value != "" {
			normalized = append(normalized, value)
		}
	}
	slices.Sort(normalized)
	return slices.Compact(normalized)
}

func observedPresence(present bool) string {
	if present {
		return "present"
	}
	return "absent"
}

func templateFileForStack(stackName string) string {
	switch stackName {
	case accountSetupStackDeploy:
		return "tidbcloud-byoc-setup-deploy.yaml"
	case accountSetupStackDataPlane:
		return "tidbcloud-byoc-setup-dataplane.yaml"
	case accountSetupStackO11Y:
		return "tidbcloud-byoc-setup-o11y.yaml"
	default:
		return ""
	}
}