//nolint:gocritic // Runtime rendering deliberately receives immutable activation contracts by value.
package byoc

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/netip"
	"path/filepath"
	"strings"

	"gopkg.in/yaml.v3"

	"github.com/tidbcloud/controlplane9/internal/proof"
	"github.com/tidbcloud/controlplane9/internal/sourcebacked"
)

const (
	byocRuntimeNamespace           = "cp-byoc"
	byocDataplaneMetadataNamespace = "byoc-dataplane-metadata"
	byocTiCDCMetadataNamespace     = "byoc-ticdc-service-meta"
	byocRuntimeImagePullSecret     = "cp9-staging-ecr-pull"
	byocRuntimeImageSourceNS       = "cp9-premium"
	byocRegionalAPIRouteName       = "byoc-regional-api"
	byocRegionalAPIPathPrefix      = "/cp9/byoc/api/premium"
	byocRegionalProxyRouteName     = "byoc-regional-dataplane-proxy"
	byocRegionalProxyPathPrefix    = "/cp9/byoc/dataplane/proxy"
	byocRegionalAPIServiceAddress  = "http://shared-regional-kong-ingress-controller-kong-proxy.shared-regional-kong-system.svc.cluster.local/cp9/byoc"
	premiumDataplaneNamespace      = "cp9-premium"
	premiumDataplaneControllerName = "dataplane-controller"
)

var byocRuntimeNamespaces = []string{
	byocRuntimeNamespace,
	byocDataplaneMetadataNamespace,
	byocTiCDCMetadataNamespace,
}

var byocRuntimeDeployments = []string{
	"byoc-premium-metadb",
	"byoc-config-center-regional",
	"byoc-dataplane-controller",
	"byoc-dataplane-server",
	"byoc-cluster-ng-server",
	"byoc-cluster-ng-executor",
	"byoc-ticdc-service-controller",
	"byoc-ticdc-service-server",
}

type regionalRuntimeNetwork struct {
	VPCID                   string
	SubnetID                string
	EndpointSecurityGroupID string
	PublicAccessCIDR        string
}

type runtimeDeploymentMetadata struct {
	Name       string `json:"name"`
	Generation int64  `json:"generation"`
}

type runtimeDeploymentSpec struct {
	Replicas int32 `json:"replicas"`
}

type runtimeDeploymentStatus struct {
	ObservedGeneration int64 `json:"observedGeneration"`
	AvailableReplicas  int32 `json:"availableReplicas"`
}

type runtimeDeployment struct {
	Metadata runtimeDeploymentMetadata `json:"metadata"`
	Spec     runtimeDeploymentSpec     `json:"spec"`
	Status   runtimeDeploymentStatus   `json:"status"`
}

type runtimeDeploymentsResponse struct {
	Items []runtimeDeployment `json:"items"`
}

type runtimeImagePullSecret struct {
	Data map[string]string `json:"data"`
	Type string            `json:"type"`
}

func (d *AWSDriver) applyRegionalRuntime(ctx context.Context, input Input, journal *Journal) (*StageResult, error) {
	network, err := runtimeNetworkFromJournal(input, journal)
	if err != nil {
		return nil, err
	}
	overlay := filepath.Join(input.RepoRoot, "personas", "app-delivery", "envs", "byoc", "runtime")
	kustomized, err := d.kubernetesRunner.Run(ctx, Command{
		Name: "kubectl",
		Args: kubeArgs(input, "kustomize", overlay, "--load-restrictor", "LoadRestrictionsNone"),
	})
	if err != nil {
		return nil, fmt.Errorf("render isolated BYOC regional runtime: %w", err)
	}
	manifest, digest, err := renderRegionalRuntime(input, network, kustomized)
	if err != nil {
		return nil, err
	}
	if err := d.copyRegionalRuntimeImagePullSecret(ctx, input); err != nil {
		return nil, err
	}
	if err := d.quiescePremiumDataplaneController(ctx, input); err != nil {
		return nil, err
	}
	applyOutput, err := d.kubernetesRunner.Run(ctx, Command{
		Name:  "kubectl",
		Args:  kubeArgs(input, "apply", "--server-side", "--force-conflicts", "-f", "-"),
		Stdin: manifest,
	})
	if err != nil {
		if detail := strings.TrimSpace(commandFailureText(applyOutput, err)); detail != "" {
			return nil, fmt.Errorf("apply isolated BYOC regional runtime: %s: %w", detail, err)
		}
		return nil, fmt.Errorf("apply isolated BYOC regional runtime: %w", err)
	}
	for _, deployment := range byocRuntimeDeployments {
		if _, err := d.kubernetesRunner.Run(ctx, Command{
			Name: "kubectl",
			Args: kubeArgs(
				input,
				"-n",
				byocRuntimeNamespace,
				"rollout",
				"status",
				"deployment/"+deployment,
				"--timeout=20m",
			),
		}); err != nil {
			return nil, fmt.Errorf("wait for BYOC regional deployment %s: %w", deployment, err)
		}
	}
	return &StageResult{Artifacts: map[string]string{
		"namespace":                     byocRuntimeNamespace,
		"namespaces":                    strings.Join(byocRuntimeNamespaces, ","),
		"manifest_digest":               digest,
		"deployments":                   strings.Join(byocRuntimeDeployments, ","),
		"control_plane_provider_config": "byoc-control-plane",
		"customer_provider_config":      input.CustomerAccountID,
		"control_plane_vpc_id":          network.VPCID,
		"endpoint_security_group_id":    network.EndpointSecurityGroupID,
		"premium_dataplane_controller":  "quiesced",
	}}, nil
}

func (d *AWSDriver) revalidateRegionalRuntime(ctx context.Context, input Input, record *StageRecord) (*StageResult, error) {
	if record == nil || record.Artifacts["namespace"] != byocRuntimeNamespace ||
		record.Artifacts["namespaces"] != strings.Join(byocRuntimeNamespaces, ",") || record.Artifacts["manifest_digest"] == "" ||
		record.Artifacts["premium_dataplane_controller"] != "quiesced" {
		return nil, ErrNeedsReconcile
	}
	overlay := filepath.Join(input.RepoRoot, "personas", "app-delivery", "envs", "byoc", "runtime")
	kustomized, err := d.kubernetesRunner.Run(ctx, Command{
		Name: "kubectl",
		Args: kubeArgs(input, "kustomize", overlay, "--load-restrictor", "LoadRestrictionsNone"),
	})
	if err != nil {
		return nil, fmt.Errorf("render isolated BYOC regional runtime for revalidation: %w", err)
	}
	network := regionalRuntimeNetwork{
		VPCID:                   record.Artifacts["control_plane_vpc_id"],
		SubnetID:                input.RemoteSubnetID,
		EndpointSecurityGroupID: record.Artifacts["endpoint_security_group_id"],
	}
	publicAccessCIDR, err := remotePublicAccessCIDR(input.RemotePublicIP)
	if err != nil {
		return nil, err
	}
	network.PublicAccessCIDR = publicAccessCIDR
	if network.VPCID == "" || network.EndpointSecurityGroupID == "" {
		return nil, ErrNeedsReconcile
	}
	manifest, digest, err := renderRegionalRuntime(input, network, kustomized)
	if err != nil {
		return nil, err
	}
	if digest != record.Artifacts["manifest_digest"] {
		return nil, ErrNeedsReconcile
	}
	if !d.regionalRuntimeImagePullSecretCurrent(ctx, input) {
		return nil, ErrNeedsReconcile
	}
	if !d.premiumDataplaneControllerQuiesced(ctx, input) {
		return nil, ErrNeedsReconcile
	}
	if output, err := d.kubernetesRunner.Run(ctx, Command{
		Name: "kubectl", Args: kubeArgs(input, "diff", "-f", "-"), Stdin: manifest,
	}); err != nil {
		if strings.Contains(err.Error(), "exit status 1") {
			return nil, ErrNeedsReconcile
		}
		return nil, fmt.Errorf("observe isolated BYOC regional runtime diff: %w (%s)", err, strings.TrimSpace(string(output)))
	}
	args := []string{"-n", byocRuntimeNamespace, "get", "deployments"}
	args = append(args, byocRuntimeDeployments...)
	args = append(args, "-o", "json")
	output, err := d.kubernetesRunner.Run(ctx, Command{Name: "kubectl", Args: kubeArgs(input, args...)})
	if err != nil {
		return nil, ErrNeedsReconcile
	}
	var response runtimeDeploymentsResponse
	if err := json.Unmarshal(output, &response); err != nil {
		return nil, fmt.Errorf("parse BYOC regional deployment status: %w", err)
	}
	if len(response.Items) != len(byocRuntimeDeployments) {
		return nil, ErrNeedsReconcile
	}
	for _, item := range response.Items {
		if item.Status.ObservedGeneration < item.Metadata.Generation || item.Status.AvailableReplicas < item.Spec.Replicas {
			return nil, ErrNeedsReconcile
		}
	}
	return &StageResult{Artifacts: cloneArtifacts(record.Artifacts)}, nil
}

func (d *AWSDriver) quiescePremiumDataplaneController(ctx context.Context, input Input) error {
	// cluster-ng hard-codes child regional resources with mixed service-type
	// labels. Both dataplane controllers watch cluster-wide, so the Premium
	// controller must be quiesced while this exclusive BYOC runtime owns them.
	output, err := d.kubernetesRunner.Run(ctx, Command{
		Name: "kubectl",
		Args: kubeArgs(
			input,
			"-n", premiumDataplaneNamespace,
			"scale", "deployment/"+premiumDataplaneControllerName,
			"--replicas=0",
		),
	})
	if err != nil {
		if detail := strings.TrimSpace(commandFailureText(output, err)); detail != "" {
			return fmt.Errorf("quiesce Premium dataplane controller for exclusive BYOC ownership: %s: %w", detail, err)
		}
		return fmt.Errorf("quiesce Premium dataplane controller for exclusive BYOC ownership: %w", err)
	}
	return nil
}

func (d *AWSDriver) premiumDataplaneControllerQuiesced(ctx context.Context, input Input) bool {
	output, err := d.kubernetesRunner.Run(ctx, Command{
		Name: "kubectl",
		Args: kubeArgs(
			input,
			"-n", premiumDataplaneNamespace,
			"get", "deployment/"+premiumDataplaneControllerName,
			"-o", "json",
		),
	})
	if err != nil {
		return false
	}
	var deployment runtimeDeployment
	return json.Unmarshal(output, &deployment) == nil && deployment.Spec.Replicas == 0
}

func renderRegionalRuntime(input Input, network regionalRuntimeNetwork, kustomized []byte) ([]byte, string, error) {
	documents, err := decodeYAMLDocuments(kustomized)
	if err != nil {
		return nil, "", fmt.Errorf("parse BYOC regional runtime: %w", err)
	}
	controlPlaneImages, err := regionalRuntimeControlPlaneImages(input)
	if err != nil {
		return nil, "", err
	}
	targetRegistry := fmt.Sprintf("%s.dkr.ecr.%s.amazonaws.com", input.CustomerAccountID, input.ImageSyncRegion)
	for _, document := range documents {
		kind := stringMapValue(document, "kind")
		metadata := ensureNestedMap(document, "metadata")
		labels := ensureNestedMap(metadata, "labels")
		labels["app.kubernetes.io/managed-by"] = "cp9-byoc"
		labels["cp9.tidbcloud.com/owner"] = boundedKubernetesLabelValue(input.Owner)
		labels["cp9.tidbcloud.com/session"] = boundedKubernetesLabelValue(input.Session)
		annotations := ensureNestedMap(metadata, "annotations")
		annotations["cp9.tidbcloud.com/owner"] = input.Owner
		annotations["cp9.tidbcloud.com/session"] = input.Session
		name := stringMapValue(metadata, "name")
		switch kind {
		case "ConfigMap":
			data := nestedMap(document, "data")
			switch name {
			case "byoc-regional-server-root-config":
				if err := patchDataplaneRootConfig(data, input, network); err != nil {
					return nil, "", err
				}
			case "byoc-account-list":
				data["account-list.yaml"] = fmt.Sprintf(
					"accounts:\n  - accountID: %q\n    roleName: %s\n    externalID: %q\n",
					input.CustomerAccountID,
					customerDataplaneManagerRoleName,
					"",
				)
			case "byoc-cluster-ng-server-root-config", "byoc-cluster-ng-executor-root-config":
				if err := patchClusterNGConfig(data, name, input, targetRegistry); err != nil {
					return nil, "", err
				}
			case "byoc-config-center-config":
				patchConfigCenterConfig(data, input)
			case "byoc-ticdc-service-config":
				if err := patchTiCDCConfig(data, input, targetRegistry); err != nil {
					return nil, "", err
				}
			default:
				if strings.HasPrefix(name, "byoc-regional-server-") && strings.HasSuffix(name, "-template-config") {
					for key, value := range data {
						data[key] = strings.ReplaceAll(fmt.Sprint(value), defaultCoreImageRegistry, targetRegistry)
					}
				}
			}
		case "Deployment":
			template := ensureNestedMap(ensureNestedMap(document, "spec"), "template")
			templateLabels := ensureNestedMap(ensureNestedMap(template, "metadata"), "labels")
			templateLabels["cp9.tidbcloud.com/persona"] = "byoc"
			if err := patchRegionalRuntimeResourceRequests(document); err != nil {
				return nil, "", fmt.Errorf("patch %s runtime resource requests: %w", name, err)
			}
			if image, ok := controlPlaneImages[name]; ok {
				if err := patchRegionalRuntimeImage(document, image); err != nil {
					return nil, "", fmt.Errorf("patch %s control-plane image: %w", name, err)
				}
			}
			isClusterNG := name == "byoc-cluster-ng-server" || name == "byoc-cluster-ng-executor"
			if isClusterNG ||
				name == "byoc-dataplane-controller" || name == "byoc-dataplane-server" ||
				name == "byoc-ticdc-service-controller" || name == "byoc-ticdc-service-server" {
				if err := patchAWSRoleDeployment(document, input, !isClusterNG); err != nil {
					return nil, "", err
				}
			}
		case "RoleBinding", "ClusterRoleBinding":
			if name == "byoc-leader-election-rolebinding" || name == "byoc-dedicated-dataplane-manager-rolebinding" {
				patchServiceAccountRoleBinding(document, "byoc-dataplane-manager")
			}
			if name == "byoc-ticdc-service-rolebinding" {
				patchServiceAccountRoleBinding(document, "byoc-ticdc-service")
			}
		}
	}
	documents = append(regionalRuntimeControlDocuments(input), documents...)
	manifest, err := encodeYAMLDocuments(documents)
	if err != nil {
		return nil, "", fmt.Errorf("encode BYOC regional runtime: %w", err)
	}
	digest := sha256.Sum256(manifest)
	return manifest, "sha256:" + hex.EncodeToString(digest[:]), nil
}

func regionalRuntimeControlPlaneImages(input Input) (map[string]string, error) {
	registry, err := sourcebacked.DefaultRegistry()
	if err != nil {
		return nil, err
	}
	var selectedImages map[string]string
	if strings.TrimSpace(input.StateDir) != "" {
		summaryPath := proof.ImageSourceSummaryPath(input.StateDir, nil)
		summary, err := proof.LoadImageSourceSummary(summaryPath)
		if err != nil {
			return nil, fmt.Errorf("load BYOC image source summary: %w", err)
		}
		selectedImages = summary.Images
	}
	services := map[string]string{
		"byoc-config-center-regional":   "config-center-regional",
		"byoc-dataplane-controller":     "dataplane-manager",
		"byoc-dataplane-server":         "dataplane-manager",
		"byoc-cluster-ng-server":        "cluster-ng-server",
		"byoc-cluster-ng-executor":      "cluster-ng-executor",
		"byoc-ticdc-service-controller": "ticdc-service",
		"byoc-ticdc-service-server":     "ticdc-service",
	}
	images := make(map[string]string, len(services))
	for deployment, service := range services {
		contract, ok := registry.Contract(service)
		if !ok || strings.TrimSpace(contract.StagingImage) == "" {
			return nil, fmt.Errorf("source-backed service %s has no staging image", service)
		}
		image := strings.TrimSpace(contract.StagingImage)
		if selectedImages != nil {
			image = strings.TrimSpace(selectedImages[service])
			if image == "" {
				return nil, fmt.Errorf("BYOC image source summary has no selected image for %s", service)
			}
		}
		images[deployment] = image
	}
	return images, nil
}

func patchRegionalRuntimeImage(document map[string]any, image string) error {
	podSpec := ensureNestedMap(ensureNestedMap(ensureNestedMap(document, "spec"), "template"), "spec")
	containers, ok := podSpec["containers"].([]any)
	if !ok || len(containers) == 0 {
		return errors.New("deployment has no containers")
	}
	container, ok := containers[0].(map[string]any)
	if !ok {
		return errors.New("deployment container is invalid")
	}
	container["image"] = image
	if strings.HasPrefix(strings.TrimSpace(image), "cp9.local/") {
		container["imagePullPolicy"] = "IfNotPresent"
		return nil
	}
	var imagePullSecrets []any
	if value, exists := podSpec["imagePullSecrets"]; exists {
		var ok bool
		imagePullSecrets, ok = value.([]any)
		if !ok {
			return errors.New("deployment image pull secrets are invalid")
		}
	}
	podSpec["imagePullSecrets"] = upsertNamedObject(
		imagePullSecrets,
		map[string]any{"name": byocRuntimeImagePullSecret},
	)
	return nil
}

func patchRegionalRuntimeResourceRequests(document map[string]any) error {
	podSpec := ensureNestedMap(ensureNestedMap(ensureNestedMap(document, "spec"), "template"), "spec")
	containers, ok := podSpec["containers"].([]any)
	if !ok || len(containers) == 0 {
		return errors.New("deployment has no containers")
	}
	for _, value := range containers {
		container, ok := value.(map[string]any)
		if !ok {
			return errors.New("deployment container is invalid")
		}
		resources, ok := container["resources"].(map[string]any)
		if !ok {
			continue
		}
		requests, ok := resources["requests"].(map[string]any)
		if !ok {
			continue
		}
		if _, exists := requests["cpu"]; exists {
			requests["cpu"] = "25m"
		}
	}
	return nil
}

func (d *AWSDriver) copyRegionalRuntimeImagePullSecret(ctx context.Context, input Input) error {
	needsSecret, err := regionalRuntimeNeedsImagePullSecret(input)
	if err != nil {
		return err
	}
	if !needsSecret {
		return nil
	}
	namespace, err := encodeYAMLDocuments(regionalRuntimeControlDocuments(input)[:1])
	if err != nil {
		return fmt.Errorf("encode BYOC runtime namespace: %w", err)
	}
	if _, err := d.kubernetesRunner.Run(ctx, Command{
		Name: "kubectl", Args: kubeArgs(input, "apply", "-f", "-"), Stdin: namespace,
	}); err != nil {
		return fmt.Errorf("apply BYOC runtime namespace: %w", err)
	}
	secret, err := d.readRegionalRuntimeImagePullSecret(ctx, input, byocRuntimeImageSourceNS)
	if err != nil {
		return fmt.Errorf("read control-plane staging image pull secret: %w", err)
	}
	document := map[string]any{
		"apiVersion": "v1",
		"kind":       "Secret",
		"metadata": map[string]any{
			"name":      byocRuntimeImagePullSecret,
			"namespace": byocRuntimeNamespace,
		},
		"data": secret.Data,
		"type": secret.Type,
	}
	manifest, err := encodeYAMLDocuments([]map[string]any{document})
	if err != nil {
		return fmt.Errorf("encode BYOC runtime image pull secret: %w", err)
	}
	if _, err := d.kubernetesRunner.Run(ctx, Command{
		Name: "kubectl", Args: kubeArgs(input, "apply", "-f", "-"), Stdin: manifest,
	}); err != nil {
		return fmt.Errorf("apply BYOC runtime image pull secret: %w", err)
	}
	return nil
}

func (d *AWSDriver) regionalRuntimeImagePullSecretCurrent(ctx context.Context, input Input) bool {
	needsSecret, err := regionalRuntimeNeedsImagePullSecret(input)
	if err != nil {
		return false
	}
	if !needsSecret {
		return true
	}
	source, err := d.readRegionalRuntimeImagePullSecret(ctx, input, byocRuntimeImageSourceNS)
	if err != nil {
		return false
	}
	target, err := d.readRegionalRuntimeImagePullSecret(ctx, input, byocRuntimeNamespace)
	if err != nil {
		return false
	}
	sourceData, _ := json.Marshal(source)
	targetData, _ := json.Marshal(target)
	return bytes.Equal(sourceData, targetData)
}

func regionalRuntimeNeedsImagePullSecret(input Input) (bool, error) {
	images, err := regionalRuntimeControlPlaneImages(input)
	if err != nil {
		return false, err
	}
	for _, image := range images {
		if !strings.HasPrefix(strings.TrimSpace(image), "cp9.local/") {
			return true, nil
		}
	}
	return false, nil
}

func (d *AWSDriver) readRegionalRuntimeImagePullSecret(
	ctx context.Context,
	input Input,
	namespace string,
) (*runtimeImagePullSecret, error) {
	output, err := d.kubernetesRunner.Run(ctx, Command{
		Name: "kubectl",
		Args: kubeArgs(
			input,
			"-n",
			namespace,
			"get",
			"secret",
			byocRuntimeImagePullSecret,
			"-o",
			"json",
		),
	})
	if err != nil {
		return nil, err
	}
	var secret runtimeImagePullSecret
	if err := json.Unmarshal(output, &secret); err != nil {
		return nil, err
	}
	if len(secret.Data) == 0 || secret.Type == "" {
		return nil, errors.New("staging image pull secret is empty")
	}
	return &secret, nil
}

func patchDataplaneRootConfig(data map[string]any, input Input, network regionalRuntimeNetwork) error {
	root, err := parseEmbeddedYAML(data, "root-config.yaml")
	if err != nil {
		return fmt.Errorf("patch BYOC dataplane root config: %w", err)
	}
	resourceLabels := ensureNestedMap(root, "resourceLabels")
	predicate := ensureNestedMap(root, "predicate")
	matchLabels := ensureNestedMap(predicate, "matchLabels")
	root["namespace"] = byocDataplaneMetadataNamespace
	root["accountListPath"] = ""
	matchLabels["tags.tidbcloud.com/tenant"] = input.TenantID
	delete(matchLabels, "tags.tidbcloud.com/servicetype")
	matchExpressions := []any{}
	if value, exists := predicate["matchExpressions"]; exists {
		var ok bool
		matchExpressions, ok = value.([]any)
		if !ok {
			return errors.New("patch BYOC dataplane root config: predicate.matchExpressions is invalid")
		}
	}
	predicate["matchExpressions"] = removeMatchExpression(
		matchExpressions,
		"tags.tidbcloud.com/servicetype",
	)
	resourceLabels["tags.tidbcloud.com/tenant"] = input.TenantID
	delete(resourceLabels, "tags.tidbcloud.com/shared-pool")
	resourceLabels["tags.tidbcloud.com/usedby"] = "cp9-byoc"
	additionalTags := ensureNestedMap(root, "additionalCloudTags")
	additionalTags["ManagedBy"] = "PingCAP"
	additionalTags["cp9-managed"] = "true"
	additionalTags["cp9-persona"] = "byoc"
	additionalTags["cp9-owner"] = input.Owner
	additionalTags["cp9-session"] = input.Session
	additionalTags["cp9-region"] = input.Region
	additionalTags["cp9-account"] = input.CustomerAccountID
	additionalTags["tags.tidbcloud.com/tenant"] = input.TenantID
	delete(additionalTags, "tags.tidbcloud.com/shared-pool")
	dns := ensureNestedMap(ensureNestedMap(root, "dnsProviderConfig"), "awsRoute53")
	dns["accountID"] = input.CustomerAccountID
	dns["roleArn"] = customerDataplaneManagerRoleARN(input.CustomerAccountID)
	dns["zoneId"] = input.TiDBHostedZoneID
	dns["domain"] = input.TiDBHostedZoneName
	region := ensureNestedMap(root, "regionConfig")
	region["logicalRegionName"] = byocLogicalRegionName(input)
	region["name"] = input.Region
	zones := make([]any, 0, len(input.AvailabilityZones))
	zoneNames := make(map[string]struct{}, len(input.AvailabilityZones))
	for _, zone := range input.AvailabilityZones {
		zones = append(zones, zone.Name)
		zoneNames[zone.Name] = struct{}{}
	}
	region["zones"] = zones
	var nodeZones []any
	if value, exists := region["nodeZones"]; exists {
		var ok bool
		nodeZones, ok = value.([]any)
		if !ok {
			return errors.New("patch BYOC dataplane root config: regionConfig.nodeZones is invalid")
		}
	}
	filteredNodeZones := make([]any, 0, len(nodeZones))
	for _, value := range nodeZones {
		name := fmt.Sprint(value)
		if _, ok := zoneNames[name]; ok {
			filteredNodeZones = append(filteredNodeZones, name)
		}
	}
	if len(filteredNodeZones) > 0 && len(filteredNodeZones) == len(zones) {
		filteredNodeZones = filteredNodeZones[:len(filteredNodeZones)-1]
	}
	region["nodeZones"] = filteredNodeZones
	root["controlPlaneCrossplaneProviderConfig"] = "byoc-control-plane"
	root["dataPlaneRoleName"] = customerDataplaneManagerRoleName
	appBox := ensureNestedMap(root, "appBoxConfig")
	appBox["enableK8sPublicAccess"] = true
	appBox["apiserverPublicAccessCidrs"] = []any{network.PublicAccessCIDR}
	appBox["authorizedSubjects"] = []any{byocControlPlaneRoleARN(input)}
	adminNodeRole := ensureNestedMap(appBox, "adminNodeRoleConfig")
	adminNodeRole["roleName"] = "tidbcloud-eks-node-role"
	pca := ensureNestedMap(appBox, "pcaConfig")
	pca["caIdentifier"] = input.PCAARN
	pca["region"] = input.Region
	byocConfig := ensureNestedMap(root, "byoc")
	privateLink := ensureNestedMap(byocConfig, "privateLink")
	privateLink["controlPlaneVPCID"] = network.VPCID
	privateLink["controlPlaneSubnetIDs"] = []any{network.SubnetID}
	privateLink["controlPlaneSecurityGroupIDs"] = []any{network.EndpointSecurityGroupID}
	byocConfig["nodeRoleName"] = "tidbcloud-eks-node-role"
	return writeEmbeddedYAML(data, "root-config.yaml", root)
}

func patchClusterNGConfig(data map[string]any, name string, input Input, targetRegistry string) error {
	root, err := parseEmbeddedYAML(data, "config.yaml")
	if err != nil {
		return fmt.Errorf("patch %s config: %w", name, err)
	}
	server := ensureNestedMap(root, "server")
	server["region"] = input.Region
	image := ensureNestedMap(server, "image")
	image["registry"] = targetRegistry
	image["repoPrefix"] = "tidbcloud"
	biz := ensureNestedMap(server, "biz")
	biz["defaultDataplaneAccount"] = input.CustomerAccountID
	biz["o11yBizType"] = "byoc"
	biz["disableO11y"] = true
	biz["enablePca"] = true
	biz["kongDataplanePlugins"] = "auth-server,audit-kong"
	controlPlaneMetering := ensureNestedMap(server, "controlPlaneMetering")
	controlPlaneMetering["plan"] = "byoc"
	mysql := ensureNestedMap(server, "mysql")
	mysql["host"] = "byoc-premium-metadb.cp-byoc.svc.cluster.local"
	server["dataPlaneManagerAddr"] = "byoc-dataplane-manager.cp-byoc:10000"
	server["configCenterAddr"] = "byoc-config-center-regional.cp-byoc.svc.cluster.local:8080"
	if name == "byoc-cluster-ng-server-root-config" {
		server["executorGrpcAddr"] = "byoc-cluster-ng-executor:10002"
	}
	return writeEmbeddedYAML(data, "config.yaml", root)
}

func patchConfigCenterConfig(data map[string]any, input Input) {
	config := strings.ReplaceAll(
		fmt.Sprint(data["config.toml"]),
		"premium-metadb.cp9-system.svc.cluster.local",
		"byoc-premium-metadb.cp-byoc.svc.cluster.local",
	)
	if !strings.Contains(config, "\n[regional]\n") && !strings.HasPrefix(config, "[regional]\n") {
		config = strings.TrimRight(config, "\n") + fmt.Sprintf(
			"\n\n[regional]\nlogical_region = %q\n",
			byocLogicalRegionName(input),
		)
	} else {
		config = replaceTOMLString(config, "logical_region", byocLogicalRegionName(input))
	}
	data["config.toml"] = config
}

func replaceTOMLString(config, key, value string) string {
	lines := strings.Split(config, "\n")
	prefix := key + " ="
	for index, line := range lines {
		if strings.HasPrefix(strings.TrimSpace(line), prefix) {
			lines[index] = fmt.Sprintf("%s = %q", key, value)
		}
	}
	return strings.Join(lines, "\n")
}

func patchTiCDCConfig(data map[string]any, input Input, targetRegistry string) error {
	root, err := parseEmbeddedYAML(data, "config.yaml")
	if err != nil {
		return fmt.Errorf("patch BYOC TiCDC service config: %w", err)
	}
	root["region"] = input.Region
	root["namespace"] = byocTiCDCMetadataNamespace
	root["clusterNSPrefix"] = "cp9-byoc"
	root["ticdcBaseImage"] = replaceImageRegistry(fmt.Sprint(root["ticdcBaseImage"]), targetRegistry)
	root["ticdcPreStopImage"] = replaceImageRegistry(fmt.Sprint(root["ticdcPreStopImage"]), targetRegistry)
	root["clusterServiceBaseURL"] = "http://byoc-cluster-ng-server:80"
	root["dataPlaneManagerServiceBaseURL"] = "http://byoc-dataplane-manager.cp-byoc:80"
	return writeEmbeddedYAML(data, "config.yaml", root)
}

func replaceImageRegistry(image, targetRegistry string) string {
	parts := strings.SplitN(strings.TrimSpace(image), "/", 2)
	if len(parts) != 2 {
		return image
	}
	return targetRegistry + "/" + parts[1]
}

func patchAWSRoleDeployment(document map[string]any, input Input, isolateAuthenticator bool) error {
	spec := ensureNestedMap(ensureNestedMap(ensureNestedMap(document, "spec"), "template"), "spec")
	containers, ok := spec["containers"].([]any)
	if !ok || len(containers) == 0 {
		return errors.New("BYOC dataplane deployment has no containers")
	}
	container, ok := containers[0].(map[string]any)
	if !ok {
		return errors.New("BYOC dataplane deployment container is invalid")
	}
	var env []any
	if value, exists := container["env"]; exists {
		var ok bool
		env, ok = value.([]any)
		if !ok {
			return errors.New("BYOC dataplane deployment environment is invalid")
		}
	}
	env = upsertNamedObject(env, map[string]any{"name": "AWS_CONFIG_FILE", "value": "/etc/cp9-byoc-aws/config"})
	env = upsertNamedObject(env, map[string]any{"name": "AWS_SDK_LOAD_CONFIG", "value": "1"})
	env = upsertNamedObject(env, map[string]any{"name": "AWS_REGION", "value": input.Region})
	env = upsertNamedObject(env, map[string]any{"name": "AWS_DEFAULT_REGION", "value": input.Region})
	env = upsertNamedObject(env, map[string]any{"name": "AWS_ROLE_ARN", "value": byocControlPlaneRoleARN(input)})
	if isolateAuthenticator {
		env = upsertNamedObject(env, map[string]any{
			"name":  "PATH",
			"value": "/opt/cp9-byoc/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
		})
	}
	container["env"] = env
	var mounts []any
	if value, exists := container["volumeMounts"]; exists {
		var ok bool
		mounts, ok = value.([]any)
		if !ok {
			return errors.New("BYOC dataplane deployment volume mounts are invalid")
		}
	}
	mounts = upsertNamedObject(mounts, map[string]any{
		"name":      "byoc-aws-runtime-config",
		"mountPath": "/etc/cp9-byoc-aws/config",
		"subPath":   "config",
		"readOnly":  true,
	})
	if isolateAuthenticator {
		mounts = upsertNamedObject(mounts, map[string]any{
			"name":      "byoc-aws-authenticator-wrapper",
			"mountPath": "/opt/cp9-byoc/bin",
			"readOnly":  true,
		})
	}
	container["volumeMounts"] = mounts
	var volumes []any
	if value, exists := spec["volumes"]; exists {
		var ok bool
		volumes, ok = value.([]any)
		if !ok {
			return errors.New("BYOC dataplane deployment volumes are invalid")
		}
	}
	volumes = upsertNamedObject(volumes, map[string]any{
		"name": "byoc-aws-runtime-config",
		"configMap": map[string]any{
			"name": "byoc-aws-runtime-config",
		},
	})
	if isolateAuthenticator {
		volumes = upsertNamedObject(volumes, map[string]any{
			"name": "byoc-aws-authenticator-wrapper",
			"configMap": map[string]any{
				"name":        "byoc-aws-authenticator-wrapper",
				"defaultMode": 365,
			},
		})
	}
	spec["volumes"] = volumes
	return nil
}

func patchServiceAccountRoleBinding(document map[string]any, serviceAccount string) {
	subjects, ok := document["subjects"].([]any)
	if !ok {
		return
	}
	for _, value := range subjects {
		subject, ok := value.(map[string]any)
		if !ok || stringMapValue(subject, "kind") != "ServiceAccount" {
			continue
		}
		subject["name"] = serviceAccount
		subject["namespace"] = byocRuntimeNamespace
	}
}

func regionalRuntimeControlDocuments(input Input) []map[string]any {
	labels := map[string]any{
		"app.kubernetes.io/managed-by": "cp9-byoc",
		"cp9.tidbcloud.com/owner":      boundedKubernetesLabelValue(input.Owner),
		"cp9.tidbcloud.com/session":    boundedKubernetesLabelValue(input.Session),
	}
	annotations := map[string]any{
		"cp9.tidbcloud.com/owner":   input.Owner,
		"cp9.tidbcloud.com/session": input.Session,
	}
	cpRole := byocControlPlaneRoleARN(input)
	customerRole := customerDataplaneManagerRoleARN(input.CustomerAccountID)
	credentials := map[string]any{"source": "PodIdentity"}
	providerConfig := func(name string, roles ...string) map[string]any {
		chain := make([]any, 0, len(roles))
		for _, role := range roles {
			chain = append(chain, map[string]any{"roleARN": role})
		}
		return map[string]any{
			"apiVersion": "aws.upbound.io/v1beta1",
			"kind":       "ProviderConfig",
			"metadata":   map[string]any{"name": name, "labels": labels, "annotations": annotations},
			"spec": map[string]any{
				"assumeRoleChain": chain,
				"credentials":     credentials,
			},
		}
	}
	awsConfig := fmt.Sprintf(
		"[default]\nrole_arn=%s\ncredential_source=Ec2InstanceMetadata\nregion=%s\nrole_session_name=cp9-byoc-dataplane-manager\n",
		cpRole,
		input.Region,
	)
	return []map[string]any{
		{
			"apiVersion": "v1",
			"kind":       "Namespace",
			"metadata":   map[string]any{"name": byocRuntimeNamespace, "labels": labels, "annotations": annotations},
		},
		{
			"apiVersion": "v1",
			"kind":       "Namespace",
			"metadata":   map[string]any{"name": byocDataplaneMetadataNamespace, "labels": labels, "annotations": annotations},
		},
		{
			"apiVersion": "v1",
			"kind":       "Namespace",
			"metadata":   map[string]any{"name": byocTiCDCMetadataNamespace, "labels": labels, "annotations": annotations},
		},
		{
			"apiVersion": "v1",
			"kind":       "ConfigMap",
			"metadata": map[string]any{
				"name":        "byoc-aws-runtime-config",
				"namespace":   byocRuntimeNamespace,
				"labels":      labels,
				"annotations": annotations,
			},
			"data": map[string]any{"config": awsConfig},
		},
		{
			"apiVersion": "v1",
			"kind":       "ConfigMap",
			"metadata": map[string]any{
				"name":        "byoc-aws-authenticator-wrapper",
				"namespace":   byocRuntimeNamespace,
				"labels":      labels,
				"annotations": annotations,
			},
			"data": map[string]any{
				"aws-iam-authenticator": "#!/bin/sh\n# Keep the service SDK role config out of the authenticator's explicit assume-role hop.\nunset AWS_CONFIG_FILE AWS_PROFILE AWS_DEFAULT_PROFILE AWS_SDK_LOAD_CONFIG AWS_ROLE_ARN\nexec /usr/local/bin/aws-iam-authenticator \"$@\"\n",
			},
		},
		{
			"apiVersion": "gateway.networking.k8s.io/v1",
			"kind":       "HTTPRoute",
			"metadata": map[string]any{
				"name":      byocRegionalAPIRouteName,
				"namespace": byocRuntimeNamespace,
				"labels":    labels,
				"annotations": map[string]any{
					"cp9.tidbcloud.com/owner":   input.Owner,
					"cp9.tidbcloud.com/session": input.Session,
					"konghq.com/strip-path":     "true",
				},
			},
			"spec": map[string]any{
				"parentRefs": []any{map[string]any{
					"name": "shared-regional-kong", "namespace": "shared-regional-kong-system",
				}},
				"rules": []any{map[string]any{
					"matches": []any{map[string]any{
						"path": map[string]any{"type": "PathPrefix", "value": byocRegionalAPIPathPrefix},
					}},
					"backendRefs": []any{map[string]any{
						"name": "byoc-cluster-ng-server", "port": 80,
					}},
				}},
			},
		},
		{
			"apiVersion": "gateway.networking.k8s.io/v1",
			"kind":       "HTTPRoute",
			"metadata": map[string]any{
				"name":      byocRegionalProxyRouteName,
				"namespace": byocRuntimeNamespace,
				"labels":    labels,
				"annotations": map[string]any{
					"cp9.tidbcloud.com/owner":   input.Owner,
					"cp9.tidbcloud.com/session": input.Session,
					"konghq.com/strip-path":     "true",
				},
			},
			"spec": map[string]any{
				"parentRefs": []any{map[string]any{
					"name": "shared-regional-kong", "namespace": "shared-regional-kong-system",
				}},
				"rules": []any{map[string]any{
					"matches": []any{map[string]any{
						"path": map[string]any{"type": "PathPrefix", "value": byocRegionalProxyPathPrefix},
					}},
					"backendRefs": []any{map[string]any{
						"name": "byoc-cluster-ng-server-proxy", "port": 80,
					}},
				}},
			},
		},
		providerConfig("byoc-control-plane", cpRole),
		providerConfig(input.CustomerAccountID, cpRole, customerRole),
		providerConfig("dns-"+input.CustomerAccountID, cpRole, customerRole),
	}
}

func decodeYAMLDocuments(data []byte) ([]map[string]any, error) {
	decoder := yaml.NewDecoder(bytes.NewReader(data))
	documents := make([]map[string]any, 0)
	for {
		var document map[string]any
		err := decoder.Decode(&document)
		if errors.Is(err, io.EOF) {
			break
		}
		if err != nil {
			return nil, err
		}
		if len(document) > 0 {
			documents = append(documents, document)
		}
	}
	return documents, nil
}

func encodeYAMLDocuments(documents []map[string]any) ([]byte, error) {
	var output bytes.Buffer
	encoder := yaml.NewEncoder(&output)
	encoder.SetIndent(2)
	for _, document := range documents {
		if err := encoder.Encode(document); err != nil {
			return nil, err
		}
	}
	if err := encoder.Close(); err != nil {
		return nil, err
	}
	return output.Bytes(), nil
}

func parseEmbeddedYAML(data map[string]any, key string) (map[string]any, error) {
	value, ok := data[key]
	if !ok {
		return nil, fmt.Errorf("ConfigMap data key %s is missing", key)
	}
	var root map[string]any
	if err := yaml.Unmarshal(fmt.Append(nil, value), &root); err != nil {
		return nil, err
	}
	return root, nil
}

func writeEmbeddedYAML(data map[string]any, key string, root map[string]any) error {
	value, err := yaml.Marshal(root)
	if err != nil {
		return err
	}
	data[key] = string(value)
	return nil
}

func nestedMap(value map[string]any, key string) map[string]any {
	if value == nil {
		return nil
	}
	result, ok := value[key].(map[string]any)
	if !ok {
		return nil
	}
	return result
}

func ensureNestedMap(value map[string]any, key string) map[string]any {
	if result, ok := value[key].(map[string]any); ok {
		return result
	}
	result := make(map[string]any)
	value[key] = result
	return result
}

func stringMapValue(value map[string]any, key string) string {
	if value == nil {
		return ""
	}
	return strings.TrimSpace(fmt.Sprint(value[key]))
}

func upsertNamedObject(values []any, replacement map[string]any) []any {
	name := stringMapValue(replacement, "name")
	for index, value := range values {
		object, ok := value.(map[string]any)
		if ok && stringMapValue(object, "name") == name {
			values[index] = replacement
			return values
		}
	}
	return append(values, replacement)
}

func removeMatchExpression(values []any, key string) []any {
	filtered := make([]any, 0, len(values))
	for _, value := range values {
		object, ok := value.(map[string]any)
		if !ok || stringMapValue(object, "key") != key {
			filtered = append(filtered, value)
		}
	}
	return filtered
}

func kubeArgs(input Input, args ...string) []string {
	base := []string{"--kubeconfig", input.KubeconfigPath, "--context", input.KubeContext}
	return append(base, args...)
}

func runtimeNetworkFromJournal(input Input, journal *Journal) (regionalRuntimeNetwork, error) {
	if journal == nil {
		return regionalRuntimeNetwork{}, errors.New("BYOC activation journal is required for regional runtime")
	}
	record := journal.Stages[StageNetworkAssets]
	if record == nil || record.State != StageDone {
		return regionalRuntimeNetwork{}, errors.New("BYOC network assets must be ready before regional runtime")
	}
	network := regionalRuntimeNetwork{
		VPCID:                   record.Artifacts["vpc_id"],
		SubnetID:                input.RemoteSubnetID,
		EndpointSecurityGroupID: record.Artifacts["security_group_id"],
	}
	publicAccessCIDR, err := remotePublicAccessCIDR(input.RemotePublicIP)
	if err != nil {
		return regionalRuntimeNetwork{}, err
	}
	network.PublicAccessCIDR = publicAccessCIDR
	if network.VPCID == "" || network.SubnetID == "" || network.EndpointSecurityGroupID == "" {
		return regionalRuntimeNetwork{}, errors.New("BYOC network stage artifacts are incomplete")
	}
	return network, nil
}

func remotePublicAccessCIDR(value string) (string, error) {
	address, err := netip.ParseAddr(strings.TrimSpace(value))
	if err != nil || !address.Is4() {
		return "", fmt.Errorf("BYOC remote host public IP %q is not a valid IPv4 address", value)
	}
	return address.String() + "/32", nil
}