//nolint:gocritic // Destroy lifecycle boundaries deliberately receive immutable activation contracts by value.
package byoc

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

	awssdk "github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/cloudformation"
	"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/iam"
	"github.com/aws/aws-sdk-go-v2/service/resourcegroupstaggingapi"
	taggingtypes "github.com/aws/aws-sdk-go-v2/service/resourcegroupstaggingapi/types"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/aws-sdk-go-v2/service/ssm"
	"github.com/aws/smithy-go"

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

type DestroyResult struct {
	DeletedDataplanes                  []string `json:"deleted_dataplanes,omitempty"`
	DeletedStacks                      []string `json:"deleted_stacks"`
	LeaseReleased                      bool     `json:"lease_released"`
	SelectorReleased                   bool     `json:"selector_released"`
	AbsenceProved                      bool     `json:"absence_proved"`
	RegionalRuntimeHostGone            bool     `json:"regional_runtime_host_gone,omitempty"`
	RetainedAccountSetupStacks         []string `json:"retained_account_setup_stacks,omitempty"`
	RetainedImageCache                 string   `json:"retained_image_cache"`
	RetainedDNSPCA                     string   `json:"retained_dns_pca"`
	RetainedControlPlaneSetupPrincipal string   `json:"retained_control_plane_setup_principal"`
}

type destroyCloseoutProof struct {
	AccountLease            bool
	SelectorSlot            bool
	AllowForeign            bool
	RegionalRuntimeHostGone bool
}

// DestroySessionRequest defers all nonterminal input and driver construction
// until after the lifecycle lock and authoritative journal load. Build must be
// read-only; the returned driver performs mutations only inside the same lock.
type DestroySessionRequest struct {
	StateDir                         string
	Session                          string
	ControlPlaneAccountID            string
	CustomerAccountID                string
	ImageSyncRegion                  string
	Build                            func(*Journal) (Input, *AWSDriver, error)
	ObserveTerminalRemoteRuntimeHost func(context.Context, *Journal) (bool, error)
}

var dataplaneIDPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:-]{0,190}$`)
var s3BucketNamePattern = regexp.MustCompile(`^[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]$`)

// DestroySession owns the complete closeout serialization boundary: the lock
// is acquired before journal absence or state is observed and remains held
// through the final terminal checkpoint. A terminal retry never calls Build;
// callers may opt into one exact host observation to migrate or promote the
// terminal proof used by the enclosing destroy lifecycle.
func DestroySession(ctx context.Context, request DestroySessionRequest) (*DestroyResult, error) {
	if strings.TrimSpace(request.StateDir) == "" || strings.TrimSpace(request.Session) == "" {
		return nil, errors.New("BYOC destroy requires state directory and session identity")
	}
	release, err := acquireLifecycleLock(request.StateDir)
	if err != nil {
		return nil, err
	}
	defer release()
	journal, err := LoadJournal(request.StateDir)
	if errors.Is(err, os.ErrNotExist) {
		now := time.Now().UTC()
		journal = &Journal{
			APIVersion: JournalAPIVersion,
			Kind:       JournalKind,
			Session:    request.Session,
			State:      StateDestroyed,
			Stages:     make(map[Stage]*StageRecord),
			CreatedAt:  now,
			UpdatedAt:  now,
		}
		if err := saveJournal(request.StateDir, journal); err != nil {
			return nil, fmt.Errorf("persist empty BYOC lifecycle destroy fence: %w", err)
		}
		return terminalDestroyResult(request, false), nil
	}
	if err != nil {
		return nil, err
	}
	if journal.Session != request.Session {
		return nil, fmt.Errorf(
			"BYOC activation journal belongs to session %q, not %q",
			journal.Session,
			request.Session,
		)
	}
	if journal.State == StateDestroyed {
		if !journalRemoteRuntimeHostGone(journal) && stageAttempted(journal, StageRegionalRuntime) &&
			request.ObserveTerminalRemoteRuntimeHost != nil {
			gone, observeErr := request.ObserveTerminalRemoteRuntimeHost(ctx, journal)
			if observeErr != nil {
				return nil, fmt.Errorf("observe terminal BYOC remote host: %w", observeErr)
			}
			if journal.RemoteRuntimeHostGone == nil || gone {
				journal.RemoteRuntimeHostGone = &gone
				if err := saveJournal(request.StateDir, journal); err != nil {
					return nil, fmt.Errorf("persist terminal BYOC remote-host observation: %w", err)
				}
			}
		}
		return terminalDestroyResult(request, journalRemoteRuntimeHostGone(journal)), nil
	}
	if request.Build == nil {
		return nil, errors.New("BYOC destroy requires a nonterminal input builder")
	}
	input, driver, err := request.Build(journal)
	if err != nil {
		return nil, err
	}
	if driver == nil {
		return nil, errors.New("BYOC destroy input builder returned no driver")
	}
	if input.StateDir != request.StateDir || input.Session != request.Session {
		return nil, errors.New("BYOC destroy input builder changed the locked session identity")
	}
	return driver.destroyLocked(ctx, input, journal)
}

func terminalDestroyResult(request DestroySessionRequest, remoteRuntimeHostGone bool) *DestroyResult {
	result := newDestroyResult(Input{
		ControlPlaneAccountID: request.ControlPlaneAccountID,
		CustomerAccountID:     request.CustomerAccountID,
		ImageSyncRegion:       request.ImageSyncRegion,
	})
	result.LeaseReleased = true
	result.SelectorReleased = true
	result.AbsenceProved = true
	result.RegionalRuntimeHostGone = remoteRuntimeHostGone
	return result
}

// destroyLocked removes the activation closure in reverse ownership order. Its
// caller must hold the session lifecycle lock across the authoritative journal
// read. It never deletes customer-owned DNS/PCA, the account-level ECR cache,
// or the shared inert control-plane setup principal; the cross-account selector
// slot is released last.
func (d *AWSDriver) destroyLocked(ctx context.Context, input Input, journal *Journal) (*DestroyResult, error) {
	if d == nil || journal == nil {
		return nil, errors.New("BYOC destroy requires the activation driver and journal")
	}
	fingerprint, err := Fingerprint(input)
	if err != nil {
		return nil, err
	}
	if journal.Session != input.Session || journal.Fingerprint != fingerprint {
		return nil, errors.New("BYOC destroy input does not match the bound activation journal")
	}
	result := newDestroyResult(input)
	if journal.State == StateDestroyed {
		// The checkpoint is published only after the live two-account closure and
		// both reservation releases were proved. Later enclosing-destroy phases
		// may already have removed the Kubernetes and AWS observation substrate,
		// so a retry must not depend on those external readers still existing.
		if err := saveJournal(input.StateDir, journal); err != nil {
			return result, fmt.Errorf("persist terminal BYOC destroy checkpoint: %w", err)
		}
		result.LeaseReleased = true
		result.SelectorReleased = true
		result.AbsenceProved = true
		result.RegionalRuntimeHostGone = journalRemoteRuntimeHostGone(journal)
		return result, nil
	}
	protectedMutationAttempted := destroyHasReservationProtectedMutation(journal)
	leaseAttempted := stageAttempted(journal, StageLease)
	held := destroyCloseoutProof{AllowForeign: true}
	if protectedMutationAttempted || leaseAttempted {
		selectorSlot, selectorErr := d.readSelectorSlot(ctx, input)
		if selectorErr != nil {
			return nil, fmt.Errorf("validate live BYOC selector-slot reservation before destroy: %w", selectorErr)
		}
		if selectorSlotBoundToInput(selectorSlot, input, fingerprint) {
			held.SelectorSlot = true
		}
		lease, leaseErr := d.readLease(ctx, input)
		if leaseErr != nil {
			return nil, fmt.Errorf("validate live bound lease before BYOC destroy: %w", leaseErr)
		}
		if leaseBoundToInput(lease, input, fingerprint) {
			held.AccountLease = true
		}
	}
	if stageAttempted(journal, StageRegionalRuntime) {
		hostGone, observeErr := d.remoteRuntimeHostGoneForDestroy(ctx, input)
		if observeErr != nil {
			return result, observeErr
		}
		held.RegionalRuntimeHostGone = hostGone
	}
	if protectedMutationAttempted && (!held.SelectorSlot || !held.AccountLease) {
		if err := d.proveDestroyAbsence(ctx, input, journal, held); err != nil {
			required := make([]string, 0, 2)
			if !held.SelectorSlot {
				required = append(required, "live bound selector-slot reservation")
			}
			if !held.AccountLease {
				required = append(required, "live bound lease")
			}
			return result, fmt.Errorf(
				"BYOC destroy requires the %s before protected mutation; exact zero-state retry proof failed: %w",
				strings.Join(required, " and "),
				err,
			)
		}
		return d.finishDestroyCloseout(ctx, input, journal, result, held)
	}
	if stageAttempted(journal, StageCatalog) {
		bindCatalogArtifactsForDestroy(input, journal)
		if err := d.requireNoActiveBYOCClusters(ctx, input, journal); err != nil {
			return result, err
		}
		dataplanes, err := d.catalogDataplaneIDs(ctx, input, journal)
		if err != nil {
			return result, err
		}
		for _, id := range dataplanes {
			if err := d.deleteDataplaneThroughDPM(ctx, input, id); err != nil {
				return result, err
			}
			result.DeletedDataplanes = append(result.DeletedDataplanes, id)
		}
		if err := d.deleteCatalog(ctx, input, journal); err != nil {
			return result, err
		}
	}
	if stageAttempted(journal, StageRegionalRuntime) {
		if !held.RegionalRuntimeHostGone {
			if err := d.deleteRegionalRuntime(ctx, input); err != nil {
				return result, err
			}
		}
	}
	if stageAttempted(journal, StageNetworkAssets) {
		if err := d.bindNetworkAssetsForDestroy(ctx, input, journal); err != nil {
			return result, err
		}
		if err := d.deleteNetworkAssets(ctx, input, journal); err != nil {
			return result, err
		}
	}
	if stageAttempted(journal, StageControlPlaneRole) {
		if err := d.bindControlPlaneRoleForDestroy(ctx, input, journal); err != nil {
			return result, err
		}
		if err := d.deleteControlPlaneRole(ctx, input, journal); err != nil {
			return result, err
		}
	}
	if stageAttempted(journal, StageAccountSetup) {
		if accountSetupIsRetained(journal.Stages[StageAccountSetup]) {
			result.RetainedAccountSetupStacks = append([]string(nil), accountSetupStackNames...)
		} else {
			if err := d.bindAccountSetupForDestroy(ctx, input, journal); err != nil {
				return result, err
			}
			stacks, err := d.deleteAccountSetup(ctx, input, journal)
			if err != nil {
				return result, err
			}
			result.DeletedStacks = stacks
		}
	}
	if err := d.deleteCustomerResidualVolumes(ctx, input); err != nil {
		return result, err
	}
	return d.finishDestroyCloseout(ctx, input, journal, result, held)
}

//nolint:gocritic // Destroy result construction copies the immutable activation contract by value.
func newDestroyResult(input Input) *DestroyResult {
	retainedPrincipal := controlPlaneAccountSetupPrincipalName + " (shared inert account prerequisite)"
	if strings.TrimSpace(input.ControlPlaneAccountID) != "" {
		retainedPrincipal = controlPlaneAccountSetupPrincipalARN(input.ControlPlaneAccountID) +
			" (shared inert account prerequisite)"
	}
	return &DestroyResult{
		RetainedImageCache: fmt.Sprintf(
			"%s.dkr.ecr.%s.amazonaws.com (ManagedBy=PingCAP account cache)",
			input.CustomerAccountID,
			input.ImageSyncRegion,
		),
		RetainedDNSPCA:                     "customer-owned TiDB/O11Y hosted zones, delegation, and PCA",
		RetainedControlPlaneSetupPrincipal: retainedPrincipal,
	}
}

// finishDestroyCloseout proves the session-owned closure absent while holding
// any exact reservations, releases only those exact reservations, proves that
// this activation no longer owns either slot, and durably checkpoints terminal
// cleanup. Foreign reservations may appear at either global coordination key;
// they are never deleted and do not make this already-clean session non-empty.
func (d *AWSDriver) finishDestroyCloseout(
	ctx context.Context,
	input Input,
	journal *Journal,
	result *DestroyResult,
	held destroyCloseoutProof,
) (*DestroyResult, error) {
	fingerprint, err := Fingerprint(input)
	if err != nil {
		return result, err
	}
	if err := d.proveDestroyAbsence(ctx, input, journal, held); err != nil {
		return result, err
	}
	if held.AccountLease {
		if err := d.releaseLease(ctx, input, fingerprint); err != nil {
			return result, err
		}
	}
	if held.SelectorSlot {
		if err := d.releaseSelectorSlot(ctx, input, fingerprint); err != nil {
			return result, err
		}
	}
	result.LeaseReleased = true
	result.SelectorReleased = true
	if err := d.proveDestroyAbsence(ctx, input, journal, destroyCloseoutProof{
		AllowForeign:            true,
		RegionalRuntimeHostGone: held.RegionalRuntimeHostGone,
	}); err != nil {
		return result, err
	}
	journal.State = StateDestroyed
	journal.RemoteRuntimeHostGone = new(held.RegionalRuntimeHostGone)
	if err := saveJournal(input.StateDir, journal); err != nil {
		return result, fmt.Errorf("persist terminal BYOC destroy checkpoint: %w", err)
	}
	result.AbsenceProved = true
	return result, nil
}

func journalRemoteRuntimeHostGone(journal *Journal) bool {
	return journal != nil && journal.RemoteRuntimeHostGone != nil && *journal.RemoteRuntimeHostGone
}

func destroyHasReservationProtectedMutation(journal *Journal) bool {
	if journal == nil {
		return false
	}
	for _, stage := range []Stage{
		StageAccountSetup,
		StageControlPlaneRole,
		StageNetworkAssets,
		StageRegionalRuntime,
		StageCatalog,
	} {
		if stageAttempted(journal, stage) {
			return true
		}
	}
	return false
}

func stageAttempted(journal *Journal, stage Stage) bool {
	if journal == nil {
		return false
	}
	record := journal.Stages[stage]
	return record != nil && (record.State != StagePending || len(record.Artifacts) > 0)
}

func bindCatalogArtifactsForDestroy(input Input, journal *Journal) {
	record := journal.Stages[StageCatalog]
	if record == nil {
		record = &StageRecord{State: StageFailed}
		journal.Stages[StageCatalog] = record
	}
	if record.Artifacts == nil {
		record.Artifacts = make(map[string]string)
	}
	for key, value := range catalogArtifacts(input) {
		if record.Artifacts[key] == "" {
			record.Artifacts[key] = value
		}
	}
}

func (d *AWSDriver) bindAccountSetupForDestroy(ctx context.Context, input Input, journal *Journal) error {
	record := journal.Stages[StageAccountSetup]
	if record == nil {
		return errors.New("BYOC account-setup stage is missing from the activation journal")
	}
	if accountSetupIsRetained(record) {
		return nil
	}
	if record.Artifacts == nil {
		record.Artifacts = make(map[string]string)
	}
	for _, name := range accountSetupStackNames {
		key := "stack." + name + ".id"
		if record.Artifacts[key] != "" {
			continue
		}
		if record.State == StageDone {
			return fmt.Errorf("completed BYOC setup stack %s has no bound stack ID", name)
		}
		stack, err := d.describeStack(ctx, input, name)
		if err != nil {
			if isAWSNotFound(err) {
				continue
			}
			return fmt.Errorf("observe crash-window BYOC setup stack %s for destroy: %w", name, err)
		}
		if err := d.validateCrashWindowStackForDestroy(ctx, input, record, name, stack); err != nil {
			return fmt.Errorf("bind crash-window BYOC setup stack %s for destroy: %w", name, err)
		}
		record.Artifacts[key] = stack.StackID
	}
	return nil
}

func (d *AWSDriver) validateCrashWindowStackForDestroy(
	ctx context.Context,
	input Input,
	record *StageRecord,
	stackName string,
	stack *cloudFormationStack,
) error {
	intent, err := journaledStackCreateIntent(input, record, stackName)
	if err != nil {
		return err
	}
	createdAt, err := time.Parse(time.RFC3339Nano, stack.CreationTime)
	if err != nil || createdAt.Before(intent.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 := validateStackOwnership(input, stackName, stack); err != nil {
		return err
	}
	if stackParametersDigest(stack.Parameters) != intent.ParametersDigest {
		return fmt.Errorf("unbound BYOC setup stack %s parameters do not match its journaled create intent", stackName)
	}
	templateDigest, err := d.stackTemplateDigest(ctx, input, stackName, stack.StackID)
	if err != nil {
		return err
	}
	if templateDigest != intent.TemplateDigest {
		return fmt.Errorf("unbound BYOC setup stack %s template does not match its journaled create intent", stackName)
	}
	return nil
}

func (d *AWSDriver) bindControlPlaneRoleForDestroy(ctx context.Context, input Input, journal *Journal) error {
	record := journal.Stages[StageControlPlaneRole]
	if record == nil {
		return errors.New("BYOC role stage is missing from the activation journal")
	}
	roleName := byocControlPlaneRoleName(input.Session)
	role, err := d.readIAMRole(ctx, input, roleName)
	if err != nil && !isAWSNotFound(err) {
		return fmt.Errorf("observe control-plane BYOC role for destroy: %w", err)
	}
	if role != nil {
		if err := validateOwnedIAMRole(role, roleName, byocControlPlaneRoleARN(input), input); err != nil {
			return err
		}
	}
	if record.Artifacts == nil {
		record.Artifacts = make(map[string]string)
	}
	want := map[string]string{
		"role_name":                   roleName,
		"remote_runtime_role_name":    roleNameFromARN(input.RemoteRuntimeRoleARN),
		"remote_inline_policy_name":   remoteBYOCAssumePolicyName(input.Session),
		"customer_inline_policy_name": "cp9-byoc-assume-customer",
		"network_inline_policy_name":  "cp9-byoc-control-plane-network",
	}
	for key, value := range want {
		if existing := record.Artifacts[key]; existing != "" && existing != value {
			return fmt.Errorf("BYOC role artifact %s is not bound to this activation", key)
		}
		record.Artifacts[key] = value
	}
	return nil
}

func (d *AWSDriver) bindNetworkAssetsForDestroy(ctx context.Context, input Input, journal *Journal) error {
	record := journal.Stages[StageNetworkAssets]
	if record == nil {
		return errors.New("BYOC network stage is missing from the activation journal")
	}
	if record.Artifacts["security_group_id"] != "" {
		return nil
	}
	vpcID, err := d.controlPlaneVPCID(ctx, input)
	if err != nil {
		return err
	}
	group, err := d.findEndpointSecurityGroup(ctx, input, vpcID, endpointSecurityGroupName(input.Session))
	if err != nil {
		return err
	}
	if group == nil {
		return nil
	}
	if !group.Owned {
		return fmt.Errorf("existing BYOC endpoint security group %s is not owned by this session", group.GroupID)
	}
	if record.Artifacts == nil {
		record.Artifacts = make(map[string]string)
	}
	record.Artifacts["security_group_id"] = group.GroupID
	record.Artifacts["security_group_name"] = group.GroupName
	record.Artifacts["vpc_id"] = group.VPCID
	return nil
}

func (d *AWSDriver) requireNoActiveBYOCClusters(ctx context.Context, input Input, journal *Journal) error {
	a, err := boundCatalogArtifacts(input, journal)
	if err != nil {
		return err
	}
	query := fmt.Sprintf("SELECT COUNT(*) FROM clusters WHERE region_id=%s AND deleted_at IS NULL;", a["region_id"])
	output, err := d.runGlobalCatalogQuery(ctx, input, query)
	if err != nil {
		return fmt.Errorf("check BYOC clusters before destroy: %w", err)
	}
	count, err := strconv.Atoi(strings.TrimSpace(string(output)))
	if err != nil || count != 0 {
		return fmt.Errorf(
			"BYOC Region still has %q active product cluster(s); delete them through BYOC-A before cp9 destroy",
			strings.TrimSpace(string(output)),
		)
	}
	return nil
}

func (d *AWSDriver) catalogDataplaneIDs(ctx context.Context, input Input, journal *Journal) ([]string, error) {
	a, err := boundCatalogArtifacts(input, journal)
	if err != nil {
		return nil, err
	}
	query := fmt.Sprintf(
		"SELECT DISTINCT dataplane_id FROM resource_pools WHERE region_id=%s AND deleted_at IS NULL AND dataplane_id IS NOT NULL AND dataplane_id<>'' ORDER BY dataplane_id;",
		a["region_id"],
	)
	output, err := d.runGlobalCatalogQuery(ctx, input, query)
	if err != nil {
		return nil, fmt.Errorf("discover Empty BYOC dataplane: %w", err)
	}
	var result []string
	for field := range strings.FieldsSeq(string(output)) {
		if !dataplaneIDPattern.MatchString(field) {
			return nil, fmt.Errorf("global catalog returned unsafe BYOC dataplane id %q", field)
		}
		result = append(result, field)
	}
	return result, nil
}

func (d *AWSDriver) deleteDataplaneThroughDPM(ctx context.Context, input Input, dataplaneID string) error {
	name := boundedKubernetesName("cp9-byoc-delete-dataplane-" + dataplaneID)
	clusterNames, err := d.dataplaneClusterNames(ctx, input, dataplaneID)
	if err != nil {
		return err
	}
	if _, err := d.kubernetesRunner.Run(
		ctx,
		Command{Name: "kubectl", Args: kubeArgs(input, "-n", byocRuntimeNamespace, "delete", "job", name, "--ignore-not-found=true", "--wait=true")},
	); err != nil {
		return fmt.Errorf("reset BYOC dataplane delete job %s: %w", dataplaneID, err)
	}
	script := fmt.Sprintf(`set -eu
url="%s"
code="$(curl -sS -o /tmp/delete.out -w '%%{http_code}' -X DELETE "$url")"
case "$code" in 200|202|204|404) ;; *) cat /tmp/delete.out >&2; exit 51;; esac
for i in $(seq 1 270); do
  code="$(curl -sS -o /tmp/get.out -w '%%{http_code}' "$url")"
  case "$code" in 404) exit 0;; 200) sleep 10;; *) cat /tmp/get.out >&2; exit 52;; esac
done
echo "dataplane deletion did not converge" >&2
exit 53
`, byocDataplaneManagerURL(dataplaneID))
	document := map[string]any{
		"apiVersion": "batch/v1", "kind": "Job",
		"metadata": map[string]any{"name": name, "namespace": byocRuntimeNamespace, "labels": roleProofLabels(input)},
		"spec": map[string]any{"backoffLimit": 0, "activeDeadlineSeconds": 3000, "template": map[string]any{
			"metadata": map[string]any{"labels": roleProofLabels(input)},
			"spec": map[string]any{"restartPolicy": "Never", "containers": []any{map[string]any{
				"name": "delete", "image": "curlimages/curl:8.16.0", "command": []any{"/bin/sh", "-ceu"}, "args": []any{script},
			}}},
		}},
	}
	manifest, err := encodeYAMLDocuments([]map[string]any{document})
	if err != nil {
		return err
	}
	if _, err := d.kubernetesRunner.Run(ctx, Command{Name: "kubectl", Args: kubeArgs(input, "apply", "-f", "-"), Stdin: manifest}); err != nil {
		return fmt.Errorf("apply BYOC dataplane delete job %s: %w", dataplaneID, err)
	}
	var lastWaitErr error
	for range 20 {
		_, waitErr := d.kubernetesRunner.Run(
			ctx,
			Command{
				Name: "kubectl",
				Args: kubeArgs(input, "-n", byocRuntimeNamespace, "wait", "--for=condition=complete", "job/"+name, "--timeout=2m"),
			},
		)
		if waitErr == nil {
			return nil
		}
		lastWaitErr = waitErr
		state, stateErr := d.dataplaneDeleteJobState(ctx, input, name)
		if stateErr != nil {
			return fmt.Errorf("observe Empty BYOC dataplane %s delete job: %w", dataplaneID, stateErr)
		}
		if state.Status.Succeeded > 0 {
			return nil
		}
		if state.Status.Failed > 0 || state.Status.Active == 0 {
			return fmt.Errorf("delete Empty BYOC dataplane %s through DPM: %w", dataplaneID, lastWaitErr)
		}
		if err := d.deleteDetachedDataplaneENIs(ctx, input, clusterNames); err != nil {
			return fmt.Errorf("delete exact detached BYOC dataplane ENIs for %s: %w", dataplaneID, err)
		}
	}
	return fmt.Errorf("delete Empty BYOC dataplane %s through DPM: %w", dataplaneID, lastWaitErr)
}

type dataplaneDeleteJobState struct {
	Active    int `json:"active"`
	Succeeded int `json:"succeeded"`
	Failed    int `json:"failed"`
}

type dataplaneDeleteJobStatus struct {
	Status dataplaneDeleteJobState `json:"status"`
}

func (d *AWSDriver) dataplaneDeleteJobState(ctx context.Context, input Input, name string) (dataplaneDeleteJobStatus, error) {
	var state dataplaneDeleteJobStatus
	output, err := d.kubernetesRunner.Run(ctx, Command{Name: "kubectl", Args: kubeArgs(
		input, "-n", byocRuntimeNamespace, "get", "job", name, "-o", "json",
	)})
	if err != nil {
		return state, err
	}
	if err := json.Unmarshal(output, &state); err != nil {
		return state, fmt.Errorf("decode dataplane delete job %s: %w", name, err)
	}
	return state, nil
}

func (d *AWSDriver) dataplaneClusterNames(ctx context.Context, input Input, dataplaneID string) ([]string, error) {
	selector := strings.Join([]string{
		"cloud.pingcap.com/dataplaneId=" + dataplaneID,
		"cp9-session=" + input.Session,
		"cp9-account=" + input.CustomerAccountID,
	}, ",")
	output, err := d.kubernetesRunner.Run(ctx, Command{Name: "kubectl", Args: kubeArgs(
		input, "get", "clusters.eks.aws.upbound.io", "-l", selector, "-o", "json",
	)})
	if err != nil {
		return nil, fmt.Errorf("discover exact BYOC EKS cluster for dataplane %s: %w", dataplaneID, err)
	}
	var list runtimeObjectList
	if err := json.Unmarshal(output, &list); err != nil {
		return nil, fmt.Errorf("decode exact BYOC EKS clusters for dataplane %s: %w", dataplaneID, err)
	}
	names := make([]string, 0, len(list.Items))
	for _, item := range list.Items {
		labels := item.Metadata.Labels
		name := strings.TrimSpace(item.Metadata.Name)
		if item.Kind != "Cluster" || name == "" || labels["cloud.pingcap.com/dataplaneId"] != dataplaneID ||
			labels["cp9-session"] != input.Session || labels["cp9-account"] != input.CustomerAccountID ||
			item.Metadata.Annotations["crossplane.io/external-name"] != name {
			return nil, fmt.Errorf("BYOC EKS cluster discovery returned an object outside dataplane %s ownership", dataplaneID)
		}
		names = append(names, name)
	}
	slices.Sort(names)
	return names, nil
}

func (d *AWSDriver) deleteDetachedDataplaneENIs(ctx context.Context, input Input, clusterNames []string) error {
	if len(clusterNames) == 0 {
		return nil
	}
	clients, err := d.clients(ctx, input.CustomerProfile, input.Region)
	if err != nil {
		return err
	}
	for _, clusterName := range clusterNames {
		var nextToken *string
		for {
			response, describeErr := clients.ec2.DescribeNetworkInterfaces(ctx, &ec2.DescribeNetworkInterfacesInput{
				Filters: []ec2types.Filter{
					{Name: awssdk.String("tag:cluster.k8s.amazonaws.com/name"), Values: []string{clusterName}},
					{Name: awssdk.String("status"), Values: []string{"available"}},
				},
				NextToken: nextToken,
			})
			if describeErr != nil {
				return fmt.Errorf("describe detached ENIs for exact cluster %s: %w", clusterName, describeErr)
			}
			for _, interfaceID := range detachedDataplaneENIIDs(response.NetworkInterfaces, clusterName) {
				if _, deleteErr := clients.ec2.DeleteNetworkInterface(ctx, &ec2.DeleteNetworkInterfaceInput{
					NetworkInterfaceId: awssdk.String(interfaceID),
				}); deleteErr != nil && !networkInterfaceNotFoundForDestroy(deleteErr) {
					return fmt.Errorf("delete exact detached ENI %s for cluster %s: %w", interfaceID, clusterName, deleteErr)
				}
			}
			nextToken = response.NextToken
			if strings.TrimSpace(awssdk.ToString(nextToken)) == "" {
				break
			}
		}
	}
	return nil
}

func detachedDataplaneENIIDs(interfaces []ec2types.NetworkInterface, clusterName string) []string {
	ids := make([]string, 0, len(interfaces))
	for _, networkInterface := range interfaces {
		tags := make(map[string]string, len(networkInterface.TagSet))
		for _, tag := range networkInterface.TagSet {
			tags[awssdk.ToString(tag.Key)] = awssdk.ToString(tag.Value)
		}
		if strings.TrimSpace(awssdk.ToString(networkInterface.NetworkInterfaceId)) == "" ||
			string(networkInterface.Status) != "available" || networkInterface.Attachment != nil ||
			awssdk.ToBool(networkInterface.RequesterManaged) || string(networkInterface.InterfaceType) != "interface" ||
			tags["cluster.k8s.amazonaws.com/name"] != clusterName ||
			tags["eks:eni:owner"] != "amazon-vpc-cni" ||
			strings.TrimSpace(tags["node.k8s.amazonaws.com/instance_id"]) == "" {
			continue
		}
		ids = append(ids, awssdk.ToString(networkInterface.NetworkInterfaceId))
	}
	slices.Sort(ids)
	return ids
}

func networkInterfaceNotFoundForDestroy(err error) bool {
	if err == nil {
		return false
	}
	var apiError smithy.APIError
	return errors.As(err, &apiError) && apiError.ErrorCode() == "InvalidNetworkInterfaceID.NotFound"
}

func byocDataplaneManagerURL(dataplaneID string) string {
	return "http://byoc-dataplane-manager.cp-byoc.svc.cluster.local:80/api/v1/dataplanes/" + dataplaneID
}

func (d *AWSDriver) deleteCatalog(ctx context.Context, input Input, journal *Journal) error {
	a, err := boundCatalogArtifacts(input, journal)
	if err != nil {
		return err
	}
	sql := fmt.Sprintf(`START TRANSACTION;
DELETE FROM resource_pool_strategies WHERE region_id=%s;
DELETE FROM resource_pools WHERE region_id=%s;
DELETE FROM region_ops_audit_log_settings WHERE region_id=%s;
DELETE FROM default_provider_identities WHERE region_id=%s AND service_plan='BYOC';
DELETE FROM provider_identities WHERE region_id=%s;
DELETE FROM region_tenants WHERE region_id=%s;
DELETE FROM region_service_plans WHERE region_id=%s AND service_plan='BYOC';
DELETE FROM rcu_quota WHERE id=%s AND tenant_id=%s AND service_plan='BYOC'
AND NOT EXISTS (
  SELECT 1 FROM region_tenants rt
  JOIN region_service_plans rsp ON rsp.id=rt.region_service_plan_id
  WHERE rt.tenant_id=%s AND rt.deleted_at IS NULL
    AND rsp.service_plan='BYOC' AND rsp.deleted_at IS NULL
);
DELETE o FROM account.feature_flag_orgs o
JOIN account.feature_flags f ON f.id=o.flag_id
WHERE f.flag_name=%s AND o.org_id=%s
AND NOT EXISTS (
  SELECT 1 FROM region_tenants rt
  JOIN region_service_plans rsp ON rsp.id=rt.region_service_plan_id
  WHERE rt.tenant_id=%s AND rt.deleted_at IS NULL
    AND rsp.service_plan='BYOC' AND rsp.deleted_at IS NULL
);
UPDATE regions SET updated_at=NOW(3), deleted_at=COALESCE(deleted_at,NOW(3)), status='inactive', is_public=0
WHERE id=%s AND name=%s AND service_plan='BYOC';
COMMIT;
`, a["region_id"], a["region_id"], a["region_id"], a["region_id"], a["region_id"], a["region_id"], a["region_id"], a["rcu_quota_id"], input.TenantID, input.TenantID,
		sqlQuote(a[byocUIFeatureFlagArtifact]), input.TenantID, input.TenantID,
		a["region_id"], sqlQuote(a["logical_region_name"]))
	if _, err := d.kubernetesRunner.Run(ctx, Command{
		Name: "kubectl",
		Args: kubeArgs(
			input,
			"-n",
			"cp9-system",
			"exec",
			"-i",
			"deployment/premium-metadb",
			"--",
			"env",
			"MYSQL_PWD=cp9-local",
			"mysql",
			"-uroot",
			byocGlobalDatabase,
		),
		Stdin: []byte(sql),
	}); err != nil {
		return fmt.Errorf("delete BYOC global catalog: %w", err)
	}
	return nil
}

func (d *AWSDriver) deleteRegionalRuntime(ctx context.Context, input Input) error {
	if err := d.validateRegionalRuntimeOwnershipForDestroy(ctx, input); err != nil {
		return err
	}
	if _, err := d.kubernetesRunner.Run(ctx, Command{
		Name: "kubectl",
		Args: kubeArgs(
			input,
			"-n",
			"kong-system",
			"delete",
			"httproute",
			byocRegionalAPIRouteName,
			"--ignore-not-found=true",
			"--wait=true",
		),
	}); err != nil {
		return fmt.Errorf("delete BYOC regional API route: %w", err)
	}
	if err := d.deleteRegionalRuntimeBuckets(ctx, input); err != nil {
		return err
	}
	for _, namespaces := range regionalRuntimeNamespaceDeletionBatches() {
		namespaceArgs := []string{"delete", "namespace"}
		namespaceArgs = append(namespaceArgs, namespaces...)
		namespaceArgs = append(namespaceArgs, "--ignore-not-found=true", "--wait=true", "--timeout=20m")
		if _, err := d.kubernetesRunner.Run(ctx, Command{Name: "kubectl", Args: kubeArgs(input, namespaceArgs...)}); err != nil {
			return fmt.Errorf("delete isolated BYOC regional runtime namespaces %s: %w", strings.Join(namespaces, ","), err)
		}
	}
	selector := "cp9.tidbcloud.com/session=" + boundedKubernetesLabelValue(input.Session)
	if _, err := d.kubernetesRunner.Run(
		ctx,
		Command{
			Name: "kubectl",
			Args: kubeArgs(
				input,
				"delete",
				"clusterrole,clusterrolebinding",
				"-l",
				selector,
				"--ignore-not-found=true",
				"--wait=true",
				"--timeout=10m",
			),
		},
	); err != nil {
		return fmt.Errorf("delete BYOC cluster-scoped runtime RBAC: %w", err)
	}
	names := []string{"byoc-control-plane", input.CustomerAccountID, "dns-" + input.CustomerAccountID}
	args := []string{"delete", "providerconfig.aws.upbound.io"}
	args = append(args, names...)
	args = append(args, "--ignore-not-found=true", "--wait=true", "--timeout=10m")
	if _, err := d.kubernetesRunner.Run(ctx, Command{Name: "kubectl", Args: kubeArgs(input, args...)}); err != nil {
		return fmt.Errorf("delete BYOC ProviderConfigs: %w", err)
	}
	return nil
}

type regionalRuntimeBucketReference struct {
	Name string `json:"name"`
}

type regionalRuntimeBucketStatus struct {
	Bucket regionalRuntimeBucketReference `json:"bucket"`
}

type regionalRuntimeBucket struct {
	Metadata runtimeObjectMetadata       `json:"metadata"`
	Status   regionalRuntimeBucketStatus `json:"status"`
}

type regionalRuntimeBucketList struct {
	Items []regionalRuntimeBucket `json:"items"`
}

func (d *AWSDriver) deleteRegionalRuntimeBuckets(ctx context.Context, input Input) error {
	output, err := d.kubernetesRunner.Run(ctx, Command{Name: "kubectl", Args: kubeArgs(
		input,
		"-n", byocDataplaneMetadataNamespace,
		"get", "buckets.dedicated.regionalserver.tidbcloud.com",
		"-o", "json",
	)})
	if err != nil {
		if isCommandNotFound(output, err) {
			return nil
		}
		return fmt.Errorf("list BYOC regional runtime buckets for destroy: %w", err)
	}
	var list regionalRuntimeBucketList
	if err := json.Unmarshal(output, &list); err != nil {
		return fmt.Errorf("decode BYOC regional runtime buckets for destroy: %w", err)
	}
	if len(list.Items) == 0 {
		return nil
	}
	clients, err := d.clients(ctx, input.CustomerProfile, input.Region)
	if err != nil {
		return err
	}
	cleaner := &cpaws.ResourceCleaner{S3: clients.s3}
	names := make([]string, 0, len(list.Items))
	for _, bucket := range list.Items {
		bucketName := strings.TrimSpace(bucket.Status.Bucket.Name)
		tags := make(map[string]string)
		tagOutput, tagErr := clients.s3.GetBucketTagging(ctx, &s3.GetBucketTaggingInput{Bucket: awssdk.String(bucketName)})
		if tagErr != nil {
			if !isAWSNotFound(tagErr) {
				return fmt.Errorf("read exact BYOC bucket %s tags before destroy: %w", bucketName, tagErr)
			}
		} else {
			for _, tag := range tagOutput.TagSet {
				tags[awssdk.ToString(tag.Key)] = awssdk.ToString(tag.Value)
			}
		}
		if err := validateRegionalRuntimeBucketOwnership(bucket, tags, input, tagErr != nil); err != nil {
			return err
		}
		if tagErr == nil {
			if _, err := cleaner.Delete(ctx, &cpaws.Resource{
				ARN: "arn:aws:s3:::" + bucketName, Type: cpaws.ResourceS3, Region: input.Region,
				SessionID: input.Session, Owner: input.Owner,
			}); err != nil {
				return fmt.Errorf("delete exact BYOC regional bucket %s: %w", bucketName, err)
			}
		}
		names = append(names, bucket.Metadata.Name)
	}
	slices.Sort(names)
	for _, name := range names {
		if _, err := d.kubernetesRunner.Run(ctx, Command{Name: "kubectl", Args: kubeArgs(
			input,
			"-n", byocDataplaneMetadataNamespace,
			"annotate", "bucket.dedicated.regionalserver.tidbcloud.com", name,
			"tidbcloud.com/confirm-deletion=true", "--overwrite",
		)}); err != nil {
			return fmt.Errorf("confirm exact BYOC regional bucket %s deletion: %w", name, err)
		}
	}
	deleteArgs := []string{"-n", byocDataplaneMetadataNamespace, "delete", "bucket.dedicated.regionalserver.tidbcloud.com"}
	deleteArgs = append(deleteArgs, names...)
	deleteArgs = append(deleteArgs, "--ignore-not-found=true", "--wait=true", "--timeout=20m")
	if _, err := d.kubernetesRunner.Run(ctx, Command{Name: "kubectl", Args: kubeArgs(input, deleteArgs...)}); err != nil {
		return fmt.Errorf("delete exact BYOC regional bucket resources: %w", err)
	}
	return nil
}

func validateRegionalRuntimeBucketOwnership(
	bucket regionalRuntimeBucket,
	tags map[string]string,
	input Input,
	externalGone bool,
) error {
	name := strings.TrimSpace(bucket.Metadata.Name)
	bucketName := strings.TrimSpace(bucket.Status.Bucket.Name)
	labels := bucket.Metadata.Labels
	dataplaneID := labels["tags.tidbcloud.com/dataplane-id"]
	if name == "" || !s3BucketNamePattern.MatchString(bucketName) || !dataplaneIDPattern.MatchString(dataplaneID) ||
		labels["tags.tidbcloud.com/environment"] != "cp9" ||
		labels["tags.tidbcloud.com/servicetype"] != "byoc" ||
		labels["tags.tidbcloud.com/usedby"] != "cp9-byoc" ||
		labels["tags.tidbcloud.com/tenant"] != input.TenantID {
		return fmt.Errorf("BYOC regional bucket %s is outside session %s ownership", name, input.Session)
	}
	if externalGone {
		return nil
	}
	if tags["cp9-session"] != input.Session || tags["cp9-account"] != input.CustomerAccountID ||
		tags["cp9-owner"] != boundedKubernetesLabelValue(input.Owner) || tags["cp9-managed"] != "true" ||
		tags["cp9-persona"] != "byoc" || tags["dataplane-id"] != dataplaneID ||
		tags["crossplane-name"] != bucketName {
		return fmt.Errorf("BYOC regional bucket %s AWS tags are outside session %s ownership", bucketName, input.Session)
	}
	return nil
}

func regionalRuntimeNamespaceDeletionBatches() [][]string {
	// The controllers in cp-byoc own finalizers in both metadata namespaces.
	// Keep them alive until their dependents have reached terminal deletion.
	return [][]string{
		{byocDataplaneMetadataNamespace, byocTiCDCMetadataNamespace},
		{byocRuntimeNamespace},
	}
}

// ObserveRemoteRuntimeHostGoneForDestroy proves whether the journal-bound EC2
// host can no longer contain the isolated k3d runtime without constructing the
// rest of the activation driver.
func ObserveRemoteRuntimeHostGoneForDestroy(ctx context.Context, input Input) (bool, error) {
	return remoteRuntimeHostGoneForDestroy(ctx, ExecRunner{}, input)
}

// remoteRuntimeHostGoneForDestroy proves whether the journal-bound EC2 host
// can no longer contain the isolated k3d runtime. Only exact not-found,
// explicit empty, shutting-down, or terminated observations count as gone.
func (d *AWSDriver) remoteRuntimeHostGoneForDestroy(ctx context.Context, input Input) (bool, error) {
	return remoteRuntimeHostGoneForDestroy(ctx, d.runner, input)
}

func remoteRuntimeHostGoneForDestroy(ctx context.Context, runner Runner, input Input) (bool, error) {
	driver := &AWSDriver{runner: runner}
	identity, err := driver.getCallerIdentity(ctx, input.ControlPlaneProfile, input.Region)
	if err != nil {
		return false, fmt.Errorf("validate control-plane account before remote instance destroy observation: %w", err)
	}
	if identity.Account != strings.TrimSpace(input.ControlPlaneAccountID) {
		return false, fmt.Errorf(
			"control-plane profile resolved to account %s, want journal-bound account %s",
			identity.Account,
			strings.TrimSpace(input.ControlPlaneAccountID),
		)
	}
	clients, err := driver.clients(ctx, input.ControlPlaneProfile, input.Region)
	if err != nil {
		return false, fmt.Errorf("load control-plane AWS clients before BYOC runtime destroy: %w", err)
	}
	response, err := clients.ec2.DescribeInstances(ctx, &ec2.DescribeInstancesInput{InstanceIds: []string{input.RemoteInstanceID}})
	if err != nil {
		if remoteInstanceNotFoundForDestroy(err) {
			return true, nil
		}
		return false, fmt.Errorf("observe control-plane remote instance before BYOC runtime destroy: %w", err)
	}
	if response.Reservations == nil {
		return false, errors.New("control-plane remote instance destroy observation omitted Reservations")
	}
	if len(response.Reservations) == 0 {
		return true, nil
	}
	instances := make([]ec2types.Instance, 0, 1)
	for index, reservation := range response.Reservations {
		if len(reservation.Instances) == 0 {
			return false, fmt.Errorf(
				"control-plane remote instance destroy observation reservation %d omitted Instances",
				index,
			)
		}
		instances = append(instances, reservation.Instances...)
	}
	if len(instances) != 1 || awssdk.ToString(instances[0].InstanceId) != input.RemoteInstanceID {
		return false, fmt.Errorf(
			"control-plane remote instance destroy observation returned %d instance(s), want exactly %s",
			len(instances),
			input.RemoteInstanceID,
		)
	}
	if instances[0].State == nil {
		return false, errors.New("control-plane remote instance destroy observation omitted state")
	}
	switch instances[0].State.Name {
	case ec2types.InstanceStateNameShuttingDown, ec2types.InstanceStateNameTerminated:
		return true, nil
	default:
		return false, nil
	}
}

func remoteInstanceNotFoundForDestroy(err error) bool {
	if err == nil {
		return false
	}
	var apiError smithy.APIError
	return errors.As(err, &apiError) && apiError.ErrorCode() == "InvalidInstanceID.NotFound"
}

type runtimeObjectMetadata struct {
	Name        string            `json:"name"`
	Labels      map[string]string `json:"labels"`
	Annotations map[string]string `json:"annotations"`
}

type runtimeObject struct {
	Kind     string                `json:"kind"`
	Metadata runtimeObjectMetadata `json:"metadata"`
}

type runtimeObjectList struct {
	Items []runtimeObject `json:"items"`
}

func (d *AWSDriver) validateRegionalRuntimeOwnershipForDestroy(ctx context.Context, input Input) error {
	output, err := d.kubernetesRunner.Run(ctx, Command{Name: "kubectl", Args: kubeArgs(
		input, "-n", "kong-system", "get", "httproute", byocRegionalAPIRouteName, "-o", "json",
	)})
	if err != nil {
		if !isCommandNotFound(output, err) {
			return fmt.Errorf("observe BYOC regional API route for destroy: %w", err)
		}
	} else {
		var object struct {
			Metadata runtimeObjectMetadata `json:"metadata"`
		}
		if err := json.Unmarshal(output, &object); err != nil {
			return fmt.Errorf("parse BYOC regional API route for destroy: %w", err)
		}
		if err := validateRuntimeObjectOwnership("HTTPRoute", byocRegionalAPIRouteName, object.Metadata, input); err != nil {
			return err
		}
	}

	for _, namespace := range byocRuntimeNamespaces {
		output, err := d.kubernetesRunner.Run(ctx, Command{Name: "kubectl", Args: kubeArgs(input, "get", "namespace", namespace, "-o", "json")})
		if err != nil {
			if isCommandNotFound(output, err) {
				continue
			}
			return fmt.Errorf("observe BYOC runtime namespace %s for destroy: %w", namespace, err)
		}
		var object struct {
			Metadata runtimeObjectMetadata `json:"metadata"`
		}
		if err := json.Unmarshal(output, &object); err != nil {
			return fmt.Errorf("parse BYOC runtime namespace %s for destroy: %w", namespace, err)
		}
		if namespace == byocRuntimeNamespace && isBYOCBaselineNamespace(object.Metadata) {
			continue
		}
		if err := validateRuntimeObjectOwnership("namespace", namespace, object.Metadata, input); err != nil {
			return err
		}
	}

	selector := "cp9.tidbcloud.com/session=" + boundedKubernetesLabelValue(input.Session)
	output, err = d.kubernetesRunner.Run(
		ctx,
		Command{Name: "kubectl", Args: kubeArgs(input, "get", "clusterrole,clusterrolebinding", "-l", selector, "-o", "json")},
	)
	if err != nil {
		if !isCommandNotFound(output, err) {
			return fmt.Errorf("observe BYOC cluster-scoped runtime RBAC for destroy: %w", err)
		}
	} else {
		var list runtimeObjectList
		if err := json.Unmarshal(output, &list); err != nil {
			return fmt.Errorf("parse BYOC cluster-scoped runtime RBAC for destroy: %w", err)
		}
		for _, object := range list.Items {
			if err := validateRuntimeObjectOwnership(object.Kind, object.Metadata.Name, object.Metadata, input); err != nil {
				return err
			}
		}
	}

	for _, name := range []string{"byoc-control-plane", input.CustomerAccountID, "dns-" + input.CustomerAccountID} {
		output, err := d.kubernetesRunner.Run(
			ctx,
			Command{Name: "kubectl", Args: kubeArgs(input, "get", "providerconfig.aws.upbound.io", name, "-o", "json")},
		)
		if err != nil {
			if isCommandNotFound(output, err) {
				continue
			}
			return fmt.Errorf("observe BYOC ProviderConfig %s for destroy: %w", name, err)
		}
		var object struct {
			Metadata runtimeObjectMetadata `json:"metadata"`
		}
		if err := json.Unmarshal(output, &object); err != nil {
			return fmt.Errorf("parse BYOC ProviderConfig %s for destroy: %w", name, err)
		}
		if err := validateRuntimeObjectOwnership("ProviderConfig", name, object.Metadata, input); err != nil {
			return err
		}
	}
	return nil
}

func validateRuntimeObjectOwnership(kind, expectedName string, metadata runtimeObjectMetadata, input Input) error {
	if metadata.Name != expectedName ||
		metadata.Labels["app.kubernetes.io/managed-by"] != "cp9-byoc" ||
		metadata.Labels["cp9.tidbcloud.com/owner"] != boundedKubernetesLabelValue(input.Owner) ||
		metadata.Labels["cp9.tidbcloud.com/session"] != boundedKubernetesLabelValue(input.Session) ||
		metadata.Annotations["cp9.tidbcloud.com/owner"] != input.Owner ||
		metadata.Annotations["cp9.tidbcloud.com/session"] != input.Session {
		return fmt.Errorf("BYOC %s %s is not owned by session %s", kind, expectedName, input.Session)
	}
	return nil
}

func (d *AWSDriver) deleteNetworkAssets(ctx context.Context, input Input, journal *Journal) error {
	record := journal.Stages[StageNetworkAssets]
	if record == nil || record.Artifacts["security_group_id"] == "" {
		return nil
	}
	clients, err := d.clients(ctx, input.ControlPlaneProfile, input.Region)
	if err != nil {
		return err
	}
	_, err = clients.ec2.DeleteSecurityGroup(ctx, &ec2.DeleteSecurityGroupInput{GroupId: awssdk.String(record.Artifacts["security_group_id"])})
	if err != nil && !isAWSNotFound(err) {
		return fmt.Errorf("delete BYOC endpoint security group: %w", err)
	}
	return nil
}

func (d *AWSDriver) deleteControlPlaneRole(ctx context.Context, input Input, journal *Journal) error {
	record := journal.Stages[StageControlPlaneRole]
	if record == nil {
		return errors.New("BYOC role stage is missing from the activation journal")
	}
	remoteRole := record.Artifacts["remote_runtime_role_name"]
	remotePolicy := record.Artifacts["remote_inline_policy_name"]
	roleName := record.Artifacts["role_name"]
	if remoteRole == "" || remotePolicy == "" || roleName != byocControlPlaneRoleName(input.Session) {
		return errors.New("BYOC role stage has incomplete bound identities")
	}
	clients, err := d.clients(ctx, input.ControlPlaneProfile, input.Region)
	if err != nil {
		return err
	}
	for _, policy := range []struct{ role, name string }{
		{remoteRole, remotePolicy},
		{roleName, record.Artifacts["network_inline_policy_name"]},
		{roleName, record.Artifacts["customer_inline_policy_name"]},
	} {
		if policy.name == "" {
			return errors.New("BYOC role stage has an incomplete inline policy identity")
		}
		_, err := clients.iam.DeleteRolePolicy(ctx, &iam.DeleteRolePolicyInput{
			RoleName: awssdk.String(policy.role), PolicyName: awssdk.String(policy.name),
		})
		if err != nil && !isAWSNotFound(err) {
			return fmt.Errorf("delete BYOC inline policy %s/%s: %w", policy.role, policy.name, err)
		}
	}
	if _, err := clients.iam.DeleteRole(ctx, &iam.DeleteRoleInput{RoleName: awssdk.String(roleName)}); err != nil &&
		!isAWSNotFound(err) {
		return fmt.Errorf("delete control-plane BYOC role: %w", err)
	}
	return nil
}

func (d *AWSDriver) deleteAccountSetup(ctx context.Context, input Input, journal *Journal) ([]string, error) {
	record := journal.Stages[StageAccountSetup]
	if record == nil {
		return nil, errors.New("BYOC account-setup stage is missing from the activation journal")
	}
	if accountSetupIsRetained(record) {
		return nil, nil
	}
	var deleted []string
	clients, err := d.clients(ctx, input.CustomerProfile, input.Region)
	if err != nil {
		return nil, err
	}
	for _, name := range slices.Backward(accountSetupStackNames) {
		id := record.Artifacts["stack."+name+".id"]
		if id == "" {
			continue
		}
		if _, err := clients.cfn.DeleteStack(ctx, &cloudformation.DeleteStackInput{StackName: awssdk.String(id)}); err != nil &&
			!isAWSNotFound(err) {
			return deleted, fmt.Errorf("delete bound BYOC setup stack %s: %w", name, err)
		}
		if err := d.waitForAccountSetupStackDeletion(ctx, input, name, id); err != nil {
			return deleted, fmt.Errorf("wait for bound BYOC setup stack %s deletion: %w", name, err)
		}
		deleted = append(deleted, id)
	}
	return deleted, nil
}

func (d *AWSDriver) waitForAccountSetupStackDeletion(
	ctx context.Context,
	input Input,
	stackName string,
	stackID string,
) error {
	clients, err := d.clients(ctx, input.CustomerProfile, input.Region)
	if err != nil {
		return err
	}
	waitErr := cloudformation.NewStackDeleteCompleteWaiter(clients.cfn).Wait(
		ctx, &cloudformation.DescribeStacksInput{StackName: awssdk.String(stackID)}, 10*time.Minute,
	)
	if waitErr == nil || isAWSNotFound(waitErr) {
		return nil
	}
	if observeErr := d.confirmAccountSetupStackDeleted(ctx, input, stackName, stackID); observeErr != nil {
		return errors.Join(waitErr, observeErr)
	}
	return nil
}

func (d *AWSDriver) confirmAccountSetupStackDeleted(
	ctx context.Context,
	input Input,
	stackName string,
	stackID string,
) error {
	stack, observeErr := d.describeStack(ctx, input, stackID)
	if observeErr != nil {
		if isAWSNotFound(observeErr) {
			return nil
		}
		return fmt.Errorf("observe bound stack deletion postcondition: %w", observeErr)
	}
	if stack.StackID != stackID || stack.StackName != stackName {
		return fmt.Errorf(
			"bound stack deletion postcondition returned identity %s/%s, want %s/%s",
			stack.StackName,
			stack.StackID,
			stackName,
			stackID,
		)
	}
	if stack.StackStatus != "DELETE_COMPLETE" {
		return fmt.Errorf("bound stack %s deletion status is %s, want DELETE_COMPLETE", stackName, stack.StackStatus)
	}
	return nil
}

func (d *AWSDriver) releaseSelectorSlot(ctx context.Context, input Input, fingerprint string) error {
	slot, err := d.readSelectorSlot(ctx, input)
	if err != nil {
		return err
	}
	if slot == nil {
		return errors.New("refusing to release a missing BYOC selector-slot reservation")
	}
	if !selectorSlotBoundToInput(slot, input, fingerprint) {
		return errors.New("refusing to release a BYOC selector-slot reservation not bound to this activation")
	}
	clients, err := d.clients(ctx, input.ControlPlaneProfile, accountLeaseRegion(input))
	if err != nil {
		return err
	}
	if _, err := clients.ssm.DeleteParameter(ctx, &ssm.DeleteParameterInput{Name: awssdk.String(selectorSlotParameterName(input))}); err != nil {
		return fmt.Errorf("release BYOC selector-slot reservation: %w", err)
	}
	return nil
}

func (d *AWSDriver) releaseLease(ctx context.Context, input Input, fingerprint string) error {
	lease, err := d.readLease(ctx, input)
	if err != nil {
		return err
	}
	if lease == nil {
		return errors.New("refusing to release a missing BYOC account lease")
	}
	if !leaseBoundToInput(lease, input, fingerprint) {
		return errors.New("refusing to release a BYOC account lease not bound to this activation")
	}
	clients, err := d.clients(ctx, input.CustomerProfile, accountLeaseRegion(input))
	if err != nil {
		return err
	}
	if _, err := clients.ssm.DeleteParameter(ctx, &ssm.DeleteParameterInput{Name: awssdk.String(accountLeaseParameterName)}); err != nil {
		return fmt.Errorf("release BYOC customer account lease: %w", err)
	}
	return nil
}

func (d *AWSDriver) proveDestroyAbsence(
	ctx context.Context,
	input Input,
	journal *Journal,
	held destroyCloseoutProof,
) error {
	fingerprint, fingerprintErr := Fingerprint(input)
	if fingerprintErr != nil {
		return fmt.Errorf("fingerprint BYOC reservations during absence proof: %w", fingerprintErr)
	}
	selectorSlot, err := d.readSelectorSlot(ctx, input)
	if err != nil {
		return fmt.Errorf("observe BYOC selector-slot reservation during absence proof: %w", err)
	}
	if held.SelectorSlot {
		if !selectorSlotBoundToInput(selectorSlot, input, fingerprint) {
			return errors.New("BYOC selector-slot absence proof failed: exact bound reservation is not held")
		}
	} else if selectorSlot != nil && (!held.AllowForeign || selectorSlotBoundToInput(selectorSlot, input, fingerprint)) {
		return errors.New("BYOC selector-slot absence proof failed: reservation remains")
	}
	lease, err := d.readLease(ctx, input)
	if err != nil {
		return fmt.Errorf("observe BYOC account lease during absence proof: %w", err)
	}
	if held.AccountLease {
		if !leaseBoundToInput(lease, input, fingerprint) {
			return fmt.Errorf("BYOC account lease absence proof failed: exact bound lease is not held")
		}
	} else if lease != nil && (!held.AllowForeign || leaseBoundToInput(lease, input, fingerprint)) {
		return fmt.Errorf("BYOC account lease absence proof failed: lease remains")
	}
	if stageAttempted(journal, StageAccountSetup) && !accountSetupIsRetained(journal.Stages[StageAccountSetup]) {
		accountRecord := journal.Stages[StageAccountSetup]
		for _, name := range accountSetupStackNames {
			id := accountRecord.Artifacts["stack."+name+".id"]
			if id == "" {
				continue
			}
			if err := d.confirmAccountSetupStackDeleted(ctx, input, name, id); err != nil {
				return fmt.Errorf("prove BYOC setup stack %s absence: %w", name, err)
			}
		}
	}
	if stageAttempted(journal, StageControlPlaneRole) {
		_, err := d.readIAMRole(ctx, input, byocControlPlaneRoleName(input.Session))
		if err == nil {
			return errors.New("control-plane BYOC role remains after destroy")
		}
		if !isAWSNotFound(err) {
			return fmt.Errorf("observe control-plane BYOC role absence: %w", err)
		}
	}
	if stageAttempted(journal, StageNetworkAssets) {
		networkRecord := journal.Stages[StageNetworkAssets]
		groupID := networkRecord.Artifacts["security_group_id"]
		if groupID != "" {
			clients, clientErr := d.clients(ctx, input.ControlPlaneProfile, input.Region)
			if clientErr != nil {
				return clientErr
			}
			_, err := clients.ec2.DescribeSecurityGroups(ctx, &ec2.DescribeSecurityGroupsInput{GroupIds: []string{groupID}})
			if err == nil {
				return errors.New("BYOC endpoint security group remains after destroy")
			}
			if !isAWSNotFound(err) {
				return fmt.Errorf("observe BYOC endpoint security group absence: %w", err)
			}
		}
	}
	if stageAttempted(journal, StageRegionalRuntime) && !held.RegionalRuntimeHostGone {
		for _, namespace := range byocRuntimeNamespaces {
			if _, err := d.kubernetesRunner.Run(
				ctx,
				Command{Name: "kubectl", Args: kubeArgs(input, "get", "namespace", namespace, "-o", "name")},
			); err == nil ||
				!isCommandNotFound(nil, err) {
				return fmt.Errorf("isolated BYOC runtime namespace %s remains after destroy", namespace)
			}
		}
		selector := "cp9.tidbcloud.com/session=" + boundedKubernetesLabelValue(input.Session)
		output, err := d.kubernetesRunner.Run(
			ctx,
			Command{Name: "kubectl", Args: kubeArgs(input, "get", "clusterrole,clusterrolebinding", "-l", selector, "-o", "name")},
		)
		if err != nil && !isCommandNotFound(output, err) {
			return fmt.Errorf("prove BYOC cluster-scoped runtime RBAC absence: %w", err)
		}
		if strings.TrimSpace(string(output)) != "" {
			return errors.New("BYOC cluster-scoped runtime RBAC remains after destroy")
		}
		for _, name := range []string{"byoc-control-plane", input.CustomerAccountID, "dns-" + input.CustomerAccountID} {
			output, err := d.kubernetesRunner.Run(
				ctx,
				Command{Name: "kubectl", Args: kubeArgs(input, "get", "providerconfig.aws.upbound.io", name, "-o", "name")},
			)
			if err == nil {
				return fmt.Errorf("BYOC ProviderConfig %s remains after destroy", name)
			}
			if !isCommandNotFound(output, err) {
				return fmt.Errorf("prove BYOC ProviderConfig %s absence: %w", name, err)
			}
		}
	}
	if stageAttempted(journal, StageCatalog) {
		if err := d.proveCatalogAbsent(ctx, input, journal); err != nil {
			return err
		}
	}
	// The control-plane account also owns the same-session Dev full baseline.
	// Its BYOC closure is proved above through journal-bound roles, networking,
	// Kubernetes objects, catalog rows, and the selector reservation; a broad
	// cp9-session scan would incorrectly make unrelated baseline resources gate
	// BYOC cleanup. The distinct customer account has no such shared baseline,
	// so its broad session-tag scan remains the unknown-resource backstop.
	clients, err := d.clients(ctx, input.CustomerProfile, input.Region)
	if err != nil {
		return fmt.Errorf("scan tagged BYOC survivors with customer profile %s: %w", input.CustomerProfile, err)
	}
	paginator := resourcegroupstaggingapi.NewGetResourcesPaginator(clients.tagging, &resourcegroupstaggingapi.GetResourcesInput{
		TagFilters: []taggingtypes.TagFilter{{Key: awssdk.String("cp9-session"), Values: []string{input.Session}}},
	})
	for paginator.HasMorePages() {
		page, pageErr := paginator.NextPage(ctx)
		if pageErr != nil {
			return fmt.Errorf("scan tagged BYOC survivors with customer profile %s: %w", input.CustomerProfile, pageErr)
		}
		for _, resource := range page.ResourceTagMappingList {
			resourceARN := awssdk.ToString(resource.ResourceARN)
			if held.AccountLease && isBYOCLeaseParameterARN(resourceARN, input) {
				continue
			}
			if retainedAccountSetupStackARN(journal, resourceARN) {
				continue
			}
			live, liveErr := d.customerTaggedResourceLive(ctx, input, resourceARN)
			if liveErr != nil {
				return fmt.Errorf("directly verify tagged BYOC resource absence: %w", liveErr)
			}
			if !live {
				continue
			}
			return fmt.Errorf(
				"customer profile %s still observes a non-reservation cp9-session tagged BYOC resource",
				input.CustomerProfile,
			)
		}
	}
	return nil
}

func accountSetupIsRetained(record *StageRecord) bool {
	return record != nil && record.Artifacts[accountSetupRetainedKey] == "true"
}

func retainedAccountSetupStackARN(journal *Journal, arn string) bool {
	if journal == nil || !accountSetupIsRetained(journal.Stages[StageAccountSetup]) {
		return false
	}
	for _, name := range accountSetupStackNames {
		if journal.Stages[StageAccountSetup].Artifacts["stack."+name+".id"] == arn {
			return true
		}
	}
	return false
}

func (d *AWSDriver) deleteCustomerResidualVolumes(ctx context.Context, input Input) error {
	clients, err := d.clients(ctx, input.CustomerProfile, input.Region)
	if err != nil {
		return err
	}
	paginator := resourcegroupstaggingapi.NewGetResourcesPaginator(clients.tagging, &resourcegroupstaggingapi.GetResourcesInput{
		ResourceTypeFilters: []string{"ec2:volume"},
		TagFilters:          []taggingtypes.TagFilter{{Key: awssdk.String("cp9-session"), Values: []string{input.Session}}},
	})
	cleaner := &cpaws.ResourceCleaner{EC2: clients.ec2}
	for paginator.HasMorePages() {
		page, pageErr := paginator.NextPage(ctx)
		if pageErr != nil {
			return fmt.Errorf("scan exact BYOC residual volumes: %w", pageErr)
		}
		for _, mapping := range page.ResourceTagMappingList {
			arn := awssdk.ToString(mapping.ResourceARN)
			if err := validateCustomerResidualTags(mapping.Tags, input); err != nil {
				return fmt.Errorf("refuse residual volume %s cleanup: %w", arn, err)
			}
			if cpaws.ParseResourceType(arn) != cpaws.ResourceEBSVolume {
				return fmt.Errorf("customer residual volume scan returned unsupported resource %s", arn)
			}
			if _, err := cleaner.Delete(ctx, &cpaws.Resource{
				ARN: arn, Type: cpaws.ResourceEBSVolume, Region: input.Region,
				SessionID: input.Session, Owner: input.Owner,
			}); err != nil {
				return fmt.Errorf("delete exact BYOC residual volume %s: %w", arn, err)
			}
		}
	}
	return nil
}

func validateCustomerResidualTags(tags []taggingtypes.Tag, input Input) error {
	values := make(map[string]string, len(tags))
	for _, tag := range tags {
		values[awssdk.ToString(tag.Key)] = awssdk.ToString(tag.Value)
	}
	if values["cp9-session"] != input.Session || values["cp9-account"] != input.CustomerAccountID ||
		values["cp9-owner"] != boundedKubernetesLabelValue(input.Owner) || values["cp9-managed"] != "true" ||
		values["cp9-persona"] != "byoc" || values["environment"] != "cp9" || values["usedby"] != "cp9-byoc" {
		return errors.New("AWS tags are outside the exact BYOC session ownership boundary")
	}
	return nil
}

func (d *AWSDriver) customerTaggedResourceLive(ctx context.Context, input Input, arn string) (bool, error) {
	clients, err := d.clients(ctx, input.CustomerProfile, input.Region)
	if err != nil {
		return false, err
	}
	switch cpaws.ParseResourceType(arn) {
	case cpaws.ResourceEC2Instance:
		id := customerTaggedResourceID(arn, "instance/")
		response, err := clients.ec2.DescribeInstances(ctx, &ec2.DescribeInstancesInput{InstanceIds: []string{id}})
		if err != nil {
			if awsErrorCodeMatches(err, "InvalidInstanceID.NotFound") {
				return false, nil
			}
			return false, err
		}
		for _, reservation := range response.Reservations {
			for _, instance := range reservation.Instances {
				if awssdk.ToString(instance.InstanceId) == id && instance.State != nil &&
					instance.State.Name != ec2types.InstanceStateNameTerminated &&
					instance.State.Name != ec2types.InstanceStateNameShuttingDown {
					return true, nil
				}
			}
		}
		return false, nil
	case cpaws.ResourceEBSVolume:
		id := customerTaggedResourceID(arn, "volume/")
		response, err := clients.ec2.DescribeVolumes(ctx, &ec2.DescribeVolumesInput{VolumeIds: []string{id}})
		if err != nil {
			if awsErrorCodeMatches(err, "InvalidVolume.NotFound") {
				return false, nil
			}
			return false, err
		}
		return len(response.Volumes) > 0, nil
	case cpaws.ResourceENI:
		id := customerTaggedResourceID(arn, "network-interface/")
		response, err := clients.ec2.DescribeNetworkInterfaces(ctx, &ec2.DescribeNetworkInterfacesInput{NetworkInterfaceIds: []string{id}})
		if err != nil {
			if networkInterfaceNotFoundForDestroy(err) {
				return false, nil
			}
			return false, err
		}
		return len(response.NetworkInterfaces) > 0, nil
	case cpaws.ResourceNATGateway:
		id := customerTaggedResourceID(arn, "natgateway/")
		response, err := clients.ec2.DescribeNatGateways(ctx, &ec2.DescribeNatGatewaysInput{NatGatewayIds: []string{id}})
		if err != nil {
			if awsErrorCodeMatches(err, "NatGatewayNotFound") {
				return false, nil
			}
			return false, err
		}
		for _, gateway := range response.NatGateways {
			if gateway.State != ec2types.NatGatewayStateDeleted && gateway.State != ec2types.NatGatewayStateFailed {
				return true, nil
			}
		}
		return false, nil
	case cpaws.ResourceVPCEndpoint:
		id := customerTaggedResourceID(arn, "vpc-endpoint/")
		response, err := clients.ec2.DescribeVpcEndpoints(ctx, &ec2.DescribeVpcEndpointsInput{VpcEndpointIds: []string{id}})
		if err != nil {
			if awsErrorCodeMatches(err, "InvalidVpcEndpointId.NotFound") {
				return false, nil
			}
			return false, err
		}
		for _, endpoint := range response.VpcEndpoints {
			if endpoint.State != ec2types.StateDeleted && endpoint.State != ec2types.StateFailed {
				return true, nil
			}
		}
		return false, nil
	case cpaws.ResourceVPCEndpointService:
		id := customerTaggedResourceID(arn, "vpc-endpoint-service/")
		response, err := clients.ec2.DescribeVpcEndpointServiceConfigurations(
			ctx, &ec2.DescribeVpcEndpointServiceConfigurationsInput{ServiceIds: []string{id}},
		)
		if err != nil {
			if awsErrorCodeMatches(err, "InvalidVpcEndpointServiceId.NotFound") {
				return false, nil
			}
			return false, err
		}
		return len(response.ServiceConfigurations) > 0, nil
	case cpaws.ResourceS3:
		bucket := customerTaggedResourceID(arn, "")
		_, err := clients.s3.GetBucketTagging(ctx, &s3.GetBucketTaggingInput{Bucket: awssdk.String(bucket)})
		if err != nil {
			if isAWSNotFound(err) {
				return false, nil
			}
			return false, err
		}
		return true, nil
	default:
		return true, nil
	}
}

func customerTaggedResourceID(arn, prefix string) string {
	parts := strings.SplitN(strings.TrimSpace(arn), ":", 6)
	if len(parts) != 6 {
		return ""
	}
	return strings.TrimPrefix(parts[5], prefix)
}

func awsErrorCodeMatches(err error, codes ...string) bool {
	if err == nil {
		return false
	}
	var apiError smithy.APIError
	if !errors.As(err, &apiError) {
		return false
	}
	return slices.Contains(codes, apiError.ErrorCode())
}

func isBYOCLeaseParameterARN(value string, input Input) bool {
	parts := strings.SplitN(strings.TrimSpace(value), ":", 6)
	return len(parts) == 6 && parts[2] == "ssm" && parts[3] == accountLeaseRegion(input) &&
		parts[4] == input.CustomerAccountID &&
		parts[5] == "parameter"+accountLeaseParameterName
}

func (d *AWSDriver) proveCatalogAbsent(ctx context.Context, input Input, journal *Journal) error {
	a, err := boundCatalogArtifacts(input, journal)
	if err != nil {
		return err
	}
	query := fmt.Sprintf(`SELECT
(SELECT COUNT(*) FROM regions WHERE id=%s AND deleted_at IS NULL),
(SELECT COUNT(*) FROM region_service_plans WHERE region_id=%s AND service_plan='BYOC' AND deleted_at IS NULL),
(SELECT COUNT(*) FROM region_tenants WHERE region_id=%s AND deleted_at IS NULL),
((SELECT COUNT(*) FROM provider_identities WHERE region_id=%s AND deleted_at IS NULL)+(SELECT COUNT(*) FROM default_provider_identities WHERE region_id=%s AND service_plan='BYOC' AND deleted_at IS NULL)),
((SELECT COUNT(*) FROM resource_pools WHERE region_id=%s AND deleted_at IS NULL)+(SELECT COUNT(*) FROM resource_pool_strategies WHERE region_id=%s AND deleted_at IS NULL)),
(SELECT COUNT(*) FROM clusters WHERE region_id=%s AND deleted_at IS NULL),
(SELECT COUNT(*) FROM rcu_quota q
 WHERE q.id=%s AND q.tenant_id=%s AND q.service_plan='BYOC' AND q.deleted_at IS NULL
   AND NOT EXISTS (
     SELECT 1 FROM region_tenants rt
     JOIN region_service_plans rsp ON rsp.id=rt.region_service_plan_id
     WHERE rt.tenant_id=%s AND rt.deleted_at IS NULL
       AND rsp.service_plan='BYOC' AND rsp.deleted_at IS NULL
   )),
(SELECT COUNT(*) FROM account.feature_flags f
 JOIN account.feature_flag_orgs o ON o.flag_id=f.id
 WHERE f.flag_name=%s AND o.org_id=%s
   AND NOT EXISTS (
     SELECT 1 FROM region_tenants rt
     JOIN region_service_plans rsp ON rsp.id=rt.region_service_plan_id
     WHERE rt.tenant_id=%s AND rt.deleted_at IS NULL
       AND rsp.service_plan='BYOC' AND rsp.deleted_at IS NULL
   ));`, a["region_id"], a["region_id"], a["region_id"], a["region_id"], a["region_id"], a["region_id"], a["region_id"], a["region_id"], a["rcu_quota_id"], input.TenantID, input.TenantID,
		sqlQuote(a[byocUIFeatureFlagArtifact]), input.TenantID, input.TenantID)
	output, err := d.runGlobalCatalogQuery(ctx, input, query)
	if err != nil {
		return fmt.Errorf("prove BYOC global catalog absence: %w", err)
	}
	if strings.Join(strings.Fields(string(output)), "\t") != "0\t0\t0\t0\t0\t0\t0\t0" {
		return fmt.Errorf("BYOC global catalog absence proof returned %q", strings.TrimSpace(string(output)))
	}
	return nil
}

func (d *AWSDriver) runGlobalCatalogQuery(ctx context.Context, input Input, query string) ([]byte, error) {
	return d.kubernetesRunner.Run(ctx, Command{
		Name: "kubectl",
		Args: kubeArgs(
			input,
			"-n",
			"cp9-system",
			"exec",
			"deployment/premium-metadb",
			"--",
			"env",
			"MYSQL_PWD=cp9-local",
			"mysql",
			"-uroot",
			"-Nse",
			query,
			byocGlobalDatabase,
		),
	})
}

func boundCatalogArtifacts(input Input, journal *Journal) (map[string]string, error) {
	if journal == nil || journal.Stages[StageCatalog] == nil {
		return nil, errors.New("BYOC catalog stage is missing from activation journal")
	}
	artifacts := journal.Stages[StageCatalog].Artifacts
	want := catalogArtifacts(input)
	for key, value := range want {
		if artifacts[key] != value {
			return nil, fmt.Errorf("BYOC catalog artifact %s is not bound to this activation", key)
		}
	}
	return artifacts, nil
}