#!/bin/sh
set -eu

usage() {
  echo "usage: $0 <kustomize-path> <namespace-or-> <kubeconfig> <context> <old=new>..." >&2
  exit 2
}

[ "$#" -ge 4 ] || usage

kustomize_path=$1
namespace=$2
kubeconfig=$3
context=$4
shift 4

if [ "${CP9_AWS_RUNTIME_IDENTITY:-}" = "ec2-instance-profile" ]; then
  unset AWS_ACCESS_KEY_ID AWS_SECRET_ACCESS_KEY AWS_SESSION_TOKEN AWS_SECURITY_TOKEN
  unset AWS_PROFILE AWS_DEFAULT_PROFILE AWS_SHARED_CREDENTIALS_FILE AWS_CONFIG_FILE
  unset AWS_EC2_METADATA_DISABLED AWS_WEB_IDENTITY_TOKEN_FILE AWS_ROLE_ARN
fi

tmp=$(mktemp)
config_phase=$(mktemp)
workload_phase=$(mktemp)
serverless_auth0_secret=
serverless_auth0_meta=
dedicated_proxy_secret=
dedicated_proxy_meta=
cleanup() {
  rm -f "$tmp" "$config_phase" "$workload_phase"
  [ -z "$serverless_auth0_secret" ] || rm -f "$serverless_auth0_secret"
  [ -z "$serverless_auth0_meta" ] || rm -f "$serverless_auth0_meta"
  [ -z "$dedicated_proxy_secret" ] || rm -f "$dedicated_proxy_secret"
  [ -z "$dedicated_proxy_meta" ] || rm -f "$dedicated_proxy_meta"
}
trap cleanup EXIT
print_applied_yaml=${CP9_PRINT_APPLIED_YAML:-false}
if [ "$print_applied_yaml" = "true" ]; then
  exec 3>&1
  exec 1>&2
fi

verbose_external() {
  [ "${CP9_VERBOSE:-0}" != "1" ] || printf 'cp9 verbose: external %s\n' "$1" >&2
}

# Fetches a Secrets Manager SecretString into $2, using the {profile,region,secret_id}
# triple that a session-context meta block wrote as JSON to $1. $3 is a label used
# only in the redacted verbose line. Shared by the Starter serverless and Dedicated
# proxy Auth0 paths, whose meta blocks emit exactly these three keys.
fetch_session_secret_string() {
  fss_values=$(mktemp)
  python3 - "$1" >"$fss_values" <<'PY_FETCH_SESSION_SECRET'
import json
import sys

meta = json.load(open(sys.argv[1]))
print(meta["profile"])
print(meta["region"])
print(meta["secret_id"])
PY_FETCH_SESSION_SECRET
  {
    IFS= read -r fss_profile
    IFS= read -r fss_region
    IFS= read -r fss_secret_id
  } <"$fss_values"
  rm -f "$fss_values"
  verbose_external "aws secretsmanager get-secret-value [$3 secret-id redacted]"
  if [ "${CP9_AWS_RUNTIME_IDENTITY:-}" = "ec2-instance-profile" ]; then
    aws secretsmanager get-secret-value \
      --region "$fss_region" \
      --secret-id "$fss_secret_id" \
      --query SecretString \
      --output text >"$2"
  else
    aws --profile "$fss_profile" secretsmanager get-secret-value \
      --region "$fss_region" \
      --secret-id "$fss_secret_id" \
      --query SecretString \
      --output text >"$2"
  fi
}

strip_remote_human_aws_credentials() {
  if [ "${CP9_AWS_RUNTIME_IDENTITY:-}" = "ec2-instance-profile" ]; then
    python3 "$(dirname "$0")/strip-human-aws-credentials.py" "$tmp"
  fi
}

disable_dedicated_provider_conversion_webhooks() {
  case "$kustomize_path" in
    */envs/dev-full/dedicated-ticdc-api-crds|envs/dev-full/dedicated-ticdc-api-crds) ;;
    *) return 0 ;;
  esac
  python3 - "$tmp" <<'PY'
import re
import sys

path = sys.argv[1]
documents = []
for document in re.split(r"(?m)^---\s*$", open(path).read()):
    lines = document.strip().splitlines()
    if not lines:
        continue
    if "kind: CustomResourceDefinition" in lines:
        try:
            start = lines.index("  conversion:")
            end = next(i for i in range(start + 1, len(lines)) if lines[i].startswith("  group:"))
        except ValueError:
            pass
        else:
            lines[start:end] = ["  conversion:", "    strategy: None"]
        try:
            versions = lines.index("  versions:")
        except ValueError:
            pass
        else:
            starts = [i for i in range(versions + 1, len(lines)) if lines[i].startswith("  - ")]
            for start, end in zip(starts, starts[1:] + [len(lines)]):
                if "    storage: false" not in lines[start:end]:
                    continue
                for index in range(start, end):
                    if lines[index] == "    served: true":
                        lines[index] = "    served: false"
    documents.append("\n".join(lines))
open(path, "w").write("\n---\n".join(documents) + "\n")
PY
}

patch_image_pull_secret_before_apply() {
  image_pull_secret=${CP9_IMAGE_PULL_SECRET:-}
  if [ -z "$image_pull_secret" ] || [ "$namespace" = "-" ] || [ -z "$namespace" ]; then
    return 0
  fi

  workload_patch=$(
    python3 - "$image_pull_secret" <<'PY'
import json
import sys

print(json.dumps({
    "spec": {
        "template": {
            "spec": {
                "imagePullSecrets": [{"name": sys.argv[1]}],
            },
        },
    },
}))
PY
  )
  python3 "$(dirname "$0")/patch-local-manifests.py" "$tmp" \
    --kind Deployment --kind StatefulSet --kind DaemonSet \
    --patch "$workload_patch"

  cronjob_patch=$(
    python3 - "$image_pull_secret" <<'PY'
import json
import sys

print(json.dumps({
    "spec": {
        "jobTemplate": {
            "spec": {
                "template": {
                    "spec": {
                        "imagePullSecrets": [{"name": sys.argv[1]}],
                    },
                },
            },
        },
    },
}))
PY
  )
  python3 "$(dirname "$0")/patch-local-manifests.py" "$tmp" \
    --kind CronJob \
    --patch "$cronjob_patch"
}

patch_workload_annotation() {
  manifest=$1
  annotation_key=$2
  stamp=$3
  if [ -z "$stamp" ] || [ ! -s "$manifest" ]; then
    return 0
  fi
  workload_patch=$(
    python3 - "$annotation_key" "$stamp" <<'PY'
import json
import sys

print(json.dumps({
    "spec": {
        "template": {
            "metadata": {
                "annotations": {sys.argv[1]: sys.argv[2]},
            },
        },
    },
}))
PY
  )
  python3 "$(dirname "$0")/patch-local-manifests.py" "$manifest" \
    --kind Deployment --kind StatefulSet --kind DaemonSet \
    --patch "$workload_patch"

  cronjob_patch=$(
    python3 - "$annotation_key" "$stamp" <<'PY'
import json
import sys

print(json.dumps({
    "spec": {
        "jobTemplate": {
            "spec": {
                "template": {
                    "metadata": {
                        "annotations": {sys.argv[1]: sys.argv[2]},
                    },
                },
            },
        },
    },
}))
PY
  )
  python3 "$(dirname "$0")/patch-local-manifests.py" "$manifest" \
    --kind CronJob \
    --patch "$cronjob_patch"
}

patch_image_rollout_stamp() {
  patch_workload_annotation \
    "$tmp" \
    "cp9.tidbcloud.com/source-image-stamp" \
    "${CP9_IMAGE_ROLLOUT_STAMP:-}"
}

should_stage_workloads() {
  case "$kustomize_path" in
    */envs/dev-full/regional-server|envs/dev-full/regional-server|\
    */envs/dev-full/dedicated-cluster-controller|envs/dev-full/dedicated-cluster-controller|\
    */envs/dev-full/dedicated-cluster-control-server|envs/dev-full/dedicated-cluster-control-server|\
    */envs/dev-full/dedicated-global-service|envs/dev-full/dedicated-global-service|\
    */envs/dev-full/serverless-service|envs/dev-full/serverless-service|\
    */envs/dev-full/serverless-global-service|envs/dev-full/serverless-global-service|\
    */envs/dev-full/account-svc-config|envs/dev-full/account-svc-config)
      return 0
      ;;
    *)
      return 1
      ;;
  esac
}

split_manifest_phases() {
  python3 - "$tmp" "$config_phase" "$workload_phase" <<'PY'
import re
import sys
from pathlib import Path

source_path, config_path, workload_path = map(Path, sys.argv[1:4])
workload_kinds = {"CronJob", "DaemonSet", "Deployment", "Job", "StatefulSet"}
config_documents = []
workload_documents = []

for document in re.split(r"(?m)^---\s*$", source_path.read_text()):
    document = document.strip()
    if not document:
        continue
    match = re.search(r"(?m)^kind:\s*(\S+)\s*$", document)
    if match and match.group(1) in workload_kinds:
        workload_documents.append(document)
    else:
        config_documents.append(document)

def write(path, documents):
    path.write_text(("\n---\n".join(documents) + "\n") if documents else "")

write(config_path, config_documents)
write(workload_path, workload_documents)
PY
}

patch_runtime_config_rollout_stamp() {
  if [ ! -s "$workload_phase" ] || [ "$namespace" = "-" ] || [ -z "$namespace" ]; then
    return 0
  fi
  KUBECONFIG="$kubeconfig" python3 - \
    "$workload_phase" "$context" "$namespace" "$(dirname "$0")/patch-local-manifests.py" <<'PY'
import hashlib
import json
import os
import subprocess
import sys

manifest, context, namespace, patcher = sys.argv[1:5]
workload_kinds = {"CronJob", "DaemonSet", "Deployment", "StatefulSet"}
pod_spec_paths = {
    "CronJob": ("spec", "jobTemplate", "spec", "template", "spec"),
    "DaemonSet": ("spec", "template", "spec"),
    "Deployment": ("spec", "template", "spec"),
    "StatefulSet": ("spec", "template", "spec"),
}


def decode_json_stream(text):
    decoder = json.JSONDecoder()
    offset = 0
    while offset < len(text):
        while offset < len(text) and text[offset].isspace():
            offset += 1
        if offset == len(text):
            return
        value, offset = decoder.raw_decode(text, offset)
        yield value


def nested(value, path):
    for key in path:
        value = (value or {}).get(key)
    return value or {}


def add_reference(references, kind, value, *name_keys):
    if not isinstance(value, dict):
        return
    name = next(
        (str(value.get(key) or "").strip() for key in name_keys if value.get(key)),
        "",
    )
    if not name:
        return
    key = (kind, name)
    references[key] = references.get(key, False) or not bool(value.get("optional", False))


def workload_references(workload):
    pod_spec = nested(workload, pod_spec_paths[workload["kind"]])
    references = {}
    reference_fields = {
        "configMap": ("ConfigMap", ("name",)),
        "configMapKeyRef": ("ConfigMap", ("name",)),
        "configMapRef": ("ConfigMap", ("name",)),
        "nodePublishSecretRef": ("Secret", ("name",)),
        "secret": ("Secret", ("secretName", "name")),
        "secretKeyRef": ("Secret", ("name",)),
        "secretRef": ("Secret", ("name",)),
    }

    def scan(value):
        if isinstance(value, list):
            for child in value:
                scan(child)
            return
        if not isinstance(value, dict):
            return
        for field, child in value.items():
            if field in reference_fields:
                kind, name_keys = reference_fields[field]
                add_reference(references, kind, child, *name_keys)
            elif field == "imagePullSecrets":
                for source in child or []:
                    name = str((source or {}).get("name") or "").strip()
                    if name:
                        references[("Secret", name)] = True
            scan(child)

    scan(pod_spec)
    return references


rendered = subprocess.run(
    [
        "kubectl",
        "patch",
        "--local",
        "-f",
        manifest,
        "--type=merge",
        "--patch={}",
        "-o",
        "json",
    ],
    check=True,
    env={**os.environ, "KUBECONFIG": os.devnull},
    stdout=subprocess.PIPE,
    text=True,
).stdout
workloads = [
    item
    for item in decode_json_stream(rendered)
    if item.get("kind") in workload_kinds
]
references_by_workload = [workload_references(workload) for workload in workloads]
all_references = {}
for references in references_by_workload:
    for key, required in references.items():
        all_references[key] = all_references.get(key, False) or required

live = {}
if all_references:
    resources = [f"{kind.lower()}/{name}" for kind, name in sorted(all_references)]
    result = subprocess.run(
        [
            "kubectl",
            "--context",
            context,
            "--namespace",
            namespace,
            "get",
            *resources,
            "--ignore-not-found",
            "-o",
            "json",
        ],
        check=True,
        stdout=subprocess.PIPE,
        text=True,
    )
    if result.stdout.strip():
        response = json.loads(result.stdout)
        items = (response.get("items") or []) if response.get("kind") == "List" else [response]
        live = {
            (item.get("kind"), str((item.get("metadata") or {}).get("name") or "")): item
            for item in items
        }

missing = [
    f"{kind}/{name}"
    for (kind, name), required in all_references.items()
    if required and (kind, name) not in live
]
if missing:
    raise SystemExit("required workload configuration not found: " + ", ".join(sorted(missing)))

for workload, references in zip(workloads, references_by_workload):
    items = []
    for kind, name in sorted(references):
        item = live.get((kind, name))
        if item is None:
            items.append({"kind": kind, "name": name, "missing": True})
            continue
        metadata = item.get("metadata") or {}
        items.append({
            "kind": item.get("kind"),
            "name": metadata.get("name"),
            "type": item.get("type"),
            "data": item.get("data") or {},
            "binaryData": item.get("binaryData") or {},
        })
    payload = json.dumps(items, sort_keys=True, separators=(",", ":")).encode()
    stamp = hashlib.sha256(payload).hexdigest()
    template = {"metadata": {"annotations": {
        "cp9.tidbcloud.com/runtime-config-sha256": stamp,
    }}}
    if workload["kind"] == "CronJob":
        patch = {"spec": {"jobTemplate": {"spec": {"template": template}}}}
    else:
        patch = {"spec": {"template": template}}
    subprocess.run(
        [
            sys.executable,
            patcher,
            manifest,
            "--kind",
            workload["kind"],
            "--name",
            workload["metadata"]["name"],
            "--patch",
            json.dumps(patch),
            "--require-match",
        ],
        check=True,
    )
PY
}

print_live_yaml() {
  if [ "$print_applied_yaml" != "true" ]; then
    return 0
  fi
  if [ "$namespace" = "-" ] || [ -z "$namespace" ]; then
    KUBECONFIG="$kubeconfig" kubectl --context "$context" get -f "$tmp" -o yaml >&3
    return 0
  fi
  KUBECONFIG="$kubeconfig" kubectl --context "$context" --namespace "$namespace" get -f "$tmp" -o yaml >&3
}

patch_session_aws_credentials_before_apply() {
  session_context=${CP9_SESSION_CONTEXT:-}
  case "$kustomize_path" in
    */envs/dev-full/regional-server|envs/dev-full/regional-server)
      deployment_name=regional-server
      container_name=regional-server
      ;;
    */envs/dev-full/dedicated-cluster-controller|envs/dev-full/dedicated-cluster-controller)
      deployment_name=cluster-controller
      container_name=controller
      ;;
    */envs/dev-full/dedicated-o11y-agent-control-server|envs/dev-full/dedicated-o11y-agent-control-server)
      deployment_name=o11y-agent-control-server
      container_name=agent-control-server
      ;;
    *) return 0 ;;
  esac
  if [ -z "$session_context" ] || [ "$namespace" = "-" ] || [ -z "$namespace" ]; then
    return 0
  fi
  if [ "${CP9_AWS_RUNTIME_IDENTITY:-}" = "ec2-instance-profile" ]; then
    return 0
  fi

  deployment_patch=$(
    python3 - "$container_name" "$session_context" <<'PY'
import json
import sys

container_name, ctx_path = sys.argv[1:3]
ctx = json.load(open(ctx_path))
region = str(((ctx.get("target") or {}).get("region") or (ctx.get("host") or {}).get("region") or "")).strip()
if not region:
    raise SystemExit("session AWS credentials patch requires target.region")

patch = {
    "spec": {
        "template": {
            "spec": {
                "containers": [
                    {
                        "name": container_name,
                        "env": [
                            {"name": "AWS_SHARED_CREDENTIALS_FILE", "value": "/var/run/cp9/aws/creds"},
                            {"name": "AWS_REGION", "value": region},
                            {"name": "AWS_DEFAULT_REGION", "value": region},
                        ],
                        "volumeMounts": [
                            {
                                "name": "aws-creds",
                                "mountPath": "/var/run/cp9/aws",
                                "readOnly": True,
                            },
                        ],
                    }
                ],
                "volumes": [
                    {
                        "name": "aws-creds",
                        "secret": {"secretName": "aws-creds"},
                    },
                ],
            }
        }
    }
}
print(json.dumps(patch))
PY
  )
  python3 "$(dirname "$0")/patch-local-manifests.py" "$tmp" \
    --kind Deployment \
    --name "$deployment_name" \
    --patch "$deployment_patch" \
    --require-match
}

default_missing_namespaces() {
  if [ "$namespace" = "-" ] || [ -z "$namespace" ]; then
    return 0
  fi

  python3 - "$tmp" "$namespace" <<'PY'
import pathlib
import re
import sys

path = pathlib.Path(sys.argv[1])
namespace = sys.argv[2]
text = path.read_text()

cluster_scoped_kinds = {
    "ClusterRole",
    "ClusterRoleBinding",
    "CustomResourceDefinition",
    "KongClusterPlugin",
    "MutatingWebhookConfiguration",
    "Namespace",
    "PersistentVolume",
    "Provider",
    "ProviderConfig",
    "StorageClass",
    "ValidatingWebhookConfiguration",
}


def kind_of(document: str) -> str:
    match = re.search(r"(?m)^kind: ([^\n]+)$", document)
    return match.group(1).strip() if match else ""


def metadata_bounds(lines):
    start = None
    for index, line in enumerate(lines):
        if line == "metadata:":
            start = index
            break
    if start is None:
        return None
    end = len(lines)
    for index in range(start + 1, len(lines)):
        if lines[index] and not lines[index].startswith((" ", "-")):
            end = index
            break
    return start, end


def has_metadata_namespace(lines: list[str], start: int, end: int) -> bool:
    return any(line.startswith("  namespace:") for line in lines[start + 1 : end])


rendered = []
for document in re.split(r"\n---\n", text):
    kind = kind_of(document)
    if not kind or kind in cluster_scoped_kinds:
        rendered.append(document)
        continue

    lines = document.splitlines()
    bounds = metadata_bounds(lines)
    if bounds is None:
        rendered.append(document)
        continue
    start, end = bounds
    if has_metadata_namespace(lines, start, end):
        rendered.append(document)
        continue

    lines.insert(start + 1, f"  namespace: {namespace}")
    rendered.append("\n".join(lines))

path.write_text("\n---\n".join(rendered))
PY
}

patch_regional_server_root_config() {
  session_context=${CP9_SESSION_CONTEXT:-}
  case "$kustomize_path" in
    */envs/dev-full/regional-server|envs/dev-full/regional-server) ;;
    *) return 0 ;;
  esac
  if [ -z "$session_context" ] || [ "$namespace" = "-" ] || [ -z "$namespace" ]; then
    return 0
  fi

  root_config=$(mktemp)
  root_patch=$(mktemp)

  KUBECONFIG="$kubeconfig" kubectl --context "$context" --namespace "$namespace" \
    get configmap regional-server-root-config -o json >"$root_config"

  python3 - "$root_config" "$root_patch" "$session_context" <<'PY'
import json
import os
import re
import sys

root_config_path, root_patch_path, ctx_path = sys.argv[1:4]
configmap = json.load(open(root_config_path))
ctx = json.load(open(ctx_path))
root = ((configmap.get("data") or {}).get("root-config.yaml") or "")
if not root:
    raise SystemExit("regional-server-root-config missing data.root-config.yaml")


def replace_line(text, pattern, replacement, desc):
    updated, count = re.subn(pattern, replacement, text, count=1, flags=re.MULTILINE)
    if count != 1:
        raise SystemExit(f"unable to patch regional-server root config {desc}")
    return updated


def replace_sequence(text, parent_indent, key, item_indent, value, desc):
    pattern = re.compile(
        rf"(?m)^{re.escape(parent_indent)}{re.escape(key)}:\n"
        rf"(?P<body>(?:{re.escape(item_indent)}- .*\n)+)"
    )
    replacement = f"{parent_indent}{key}:\n{item_indent}- {json.dumps(value)}\n"
    updated, count = pattern.subn(replacement, text, count=1)
    if count != 1:
        raise SystemExit(f"unable to patch regional-server root config {desc}")
    return updated


def set_mapping_values(text, key, values, desc):
    lines = text.splitlines()
    try:
        start = next(index for index, line in enumerate(lines) if line == f"{key}:")
    except StopIteration as exc:
        raise SystemExit(f"unable to patch regional-server root config {desc}") from exc
    end = len(lines)
    for index in range(start + 1, len(lines)):
        if lines[index] and not lines[index].startswith(" "):
            end = index
            break
    block = lines[start + 1 : end]
    for map_key, map_value in values.items():
        rendered = f"  {map_key}: {json.dumps(str(map_value))}"
        replaced = False
        for offset, line in enumerate(block):
            if line.startswith(f"  {map_key}:"):
                block[offset] = rendered
                replaced = True
                break
        if not replaced:
            block.append(rendered)
    return "\n".join(lines[: start + 1] + block + lines[end:]) + "\n"


def aws_tag_label_value(value):
    value = str(value or "").strip()
    if not value:
        return "cp9"
    value = re.sub(r"[^A-Za-z0-9-]+", "-", value).strip("-")
    if not value:
        return "cp9"
    if len(value) > 35:
        value = value[:35].strip("-")
        if not value:
            return "cp9"
    return value


session_name = str(ctx.get("session_name") or (ctx.get("host") or {}).get("cluster_name") or "").strip()
if not session_name:
    raise SystemExit("CP9_SESSION_CONTEXT.session_name is required for regional-server root config")
role_arn = str((((ctx.get("terraform") or {}).get("iam") or {}).get("role_arn")) or "").strip()
match = re.match(r"^arn:aws:iam::(?P<account>\d{12}):role/(?P<role>[^/]+)$", role_arn)
if not match:
    raise SystemExit("CP9_SESSION_CONTEXT terraform.iam.role_arn must be an IAM role ARN")
account_id = match.group("account")
role_name = match.group("role")
host = ctx.get("host") or {}
remote = host.get("remote") or {}
remote_region = remote.get("region") if isinstance(remote, dict) else remote if isinstance(remote, str) else ""
region = str(
    host.get("region")
    or remote_region
    or ((ctx.get("target") or {}).get("region"))
    or ""
).strip()
if not region:
    raise SystemExit("regional-server root config requires host.region or target.region")
component_pca_arn = str((((ctx.get("terraform") or {}).get("session_product_pca") or {}).get("arn")) or "").strip()
if not component_pca_arn:
    raise SystemExit("CP9_SESSION_CONTEXT terraform.session_product_pca.arn is required for regional-server root config")
role_segment = []
last_hyphen = False
for ch in session_name:
    if ch.isalnum():
        role_segment.append(ch)
        last_hyphen = False
    elif not last_hyphen:
        role_segment.append("-")
        last_hyphen = True
role_segment = "".join(role_segment).strip("-")
eks_service_role_name = (role_segment + "-eks-service").strip("-")[:64].strip("-")
if not eks_service_role_name:
    raise SystemExit("unable to derive regional-server EKS service role name")

public_access_cidr = "0.0.0.0/0"
session_tags = ctx.get("session_tags") or {}
ttl_expires_at = str(ctx.get("ttl_expires_at") or "").strip()

root = replace_line(root, r"^env:\s*.*$", f"env: {json.dumps(session_name)}", "env")
root = replace_line(
    root,
    r"^clusterControllerAssumeRoleName:\s*.*$",
    f"clusterControllerAssumeRoleName: {json.dumps(role_name)}",
    "clusterControllerAssumeRoleName",
)
root = replace_line(
    root,
    r"^readonlyClusterControllerAssumeRoleName:\s*.*$",
    f"readonlyClusterControllerAssumeRoleName: {json.dumps(role_name)}",
    "readonlyClusterControllerAssumeRoleName",
)
root = replace_line(
    root,
    r"^dpControllerAssumeRoleName:\s*.*$",
    f"dpControllerAssumeRoleName: {json.dumps(role_name)}",
    "dpControllerAssumeRoleName",
)
root = replace_line(
    root,
    r"^ticdcControllerAssumeRoleName:\s*.*$",
    f"ticdcControllerAssumeRoleName: {json.dumps(role_name)}",
    "ticdcControllerAssumeRoleName",
)
root = replace_line(root, r"^  serviceRoleName:\s*.*$", f"  serviceRoleName: {json.dumps(eks_service_role_name)}", "appBoxConfig.serviceRoleName")
root = replace_line(root, r"^  enableK8sPublicAccess:\s*.*$", "  enableK8sPublicAccess: true", "appBoxConfig.enableK8sPublicAccess")
root = replace_sequence(root, "  ", "apiserverPublicAccessCidrs", "    ", public_access_cidr, "appBoxConfig.apiserverPublicAccessCidrs")
root = replace_sequence(root, "  ", "authorizedSubjects", "    ", role_arn, "appBoxConfig.authorizedSubjects")
root = replace_line(
    root,
    r"^componentPCAARN:\s*.*$",
    f"componentPCAARN: {component_pca_arn}",
    "componentPCAARN",
)

# Dedicated SQL DNS is independent of the optional Kong HTTPS exposure.
public_dns = (ctx.get("terraform") or {}).get("dedicated_public_dns") or {}
public_zone_id = str(public_dns.get("zone_id") or "").strip()
public_zone_base = str(public_dns.get("domain") or "").strip().rstrip(".")
if str((ctx.get("host") or {}).get("mode") or "") == "remote-ec2-k3d" and not (public_zone_id and public_zone_base):
    raise SystemExit("terraform.dedicated_public_dns is required for remote Dedicated SQL endpoints")
if public_zone_id and public_zone_base:
    root = replace_line(
        root,
        r"^publicHostedZoneDomain:\s*.*$",
        f"publicHostedZoneDomain: {json.dumps(public_zone_base)}",
        "publicHostedZoneDomain",
    )
    root = replace_line(
        root,
        r"^publicHostedZoneID:\s*.*$",
        f"publicHostedZoneID: {json.dumps(public_zone_id)}",
        "publicHostedZoneID",
    )
root = set_mapping_values(
    root,
    "resourceLabels",
    {
        "tags.tidbcloud.com/cp9-managed": "true",
        "tags.tidbcloud.com/cp9-owner": session_tags.get("cp9-owner", ""),
        "tags.tidbcloud.com/cp9-session": session_tags.get("cp9-session", ""),
        "tags.tidbcloud.com/cp9-ttl": aws_tag_label_value(ttl_expires_at),
    },
    "resourceLabels",
)

json.dump({"data": {"root-config.yaml": root}}, open(root_patch_path, "w"))
PY

  KUBECONFIG="$kubeconfig" kubectl --context "$context" --namespace "$namespace" \
    patch configmap regional-server-root-config --type=merge -p "$(cat "$root_patch")"
  rm -f "$root_config" "$root_patch"
}

patch_dedicated_cluster_controller_runtime() {
  session_context=${CP9_SESSION_CONTEXT:-}
  case "$kustomize_path" in
    */envs/dev-full/dedicated-cluster-controller|envs/dev-full/dedicated-cluster-controller) ;;
    *) return 0 ;;
  esac
  if [ -z "$session_context" ] || [ "$namespace" = "-" ] || [ -z "$namespace" ]; then
    return 0
  fi

  if ! KUBECONFIG="$kubeconfig" kubectl --context "$context" --namespace "$namespace" \
    get configmap cluster-controller-root-config >/dev/null 2>&1; then
    return 0
  fi
  root_config=$(mktemp)
  root_patch=$(mktemp)

  KUBECONFIG="$kubeconfig" kubectl --context "$context" --namespace "$namespace" \
    get configmap cluster-controller-root-config -o json >"$root_config"

  python3 - "$root_config" "$root_patch" "$session_context" <<'PY'
import json
import os
import re
import sys

root_config_path, patch_path, ctx_path = sys.argv[1:4]
configmap = json.load(open(root_config_path))
ctx = json.load(open(ctx_path))
root = ((configmap.get("data") or {}).get("root-config.yaml") or "")
if not root:
    raise SystemExit("cluster-controller-root-config missing data.root-config.yaml")

role_arn = str((((ctx.get("terraform") or {}).get("iam") or {}).get("role_arn")) or "").strip()
match = re.match(r"^arn:aws:iam::(?P<account>\d{12}):role/(?P<role>[^/]+)$", role_arn)
if not match:
    raise SystemExit("CP9_SESSION_CONTEXT terraform.iam.role_arn must be an IAM role ARN")
account_id = match.group("account")
role_name = match.group("role")
host = ctx.get("host") or {}
remote = host.get("remote") or {}
remote_region = remote.get("region") if isinstance(remote, dict) else remote if isinstance(remote, str) else ""
region = str(
    host.get("region")
    or remote_region
    or ((ctx.get("target") or {}).get("region"))
    or ""
).strip()
if not region:
    raise SystemExit("cluster-controller root config requires host.region or target.region")
component_pca_arn = str((((ctx.get("terraform") or {}).get("session_product_pca") or {}).get("arn")) or "").strip()
if not component_pca_arn:
    raise SystemExit("CP9_SESSION_CONTEXT terraform.session_product_pca.arn is required for cluster-controller root config")
session_name = str(ctx.get("session_name") or (ctx.get("host") or {}).get("cluster_name") or "").strip()
if not session_name:
    raise SystemExit("CP9_SESSION_CONTEXT.session_name is required for cluster-controller root config")

role_segment = []
last_hyphen = False
for ch in session_name:
    if ch.isalnum():
        role_segment.append(ch)
        last_hyphen = False
    elif not last_hyphen:
        role_segment.append("-")
        last_hyphen = True
role_segment = "".join(role_segment).strip("-")
eks_service_role_name = (role_segment + "-eks-service").strip("-")[:64].strip("-")
if not eks_service_role_name:
    raise SystemExit("unable to derive cluster-controller EKS service role name")


def replace_line(text, pattern, replacement, desc):
    updated, count = re.subn(pattern, replacement, text, count=1, flags=re.MULTILINE)
    if count != 1:
        raise SystemExit(f"unable to patch cluster-controller root config {desc}")
    return updated


root = replace_line(
    root,
    r"^  serviceRoleARN:\s*.*$",
    f"  serviceRoleARN: arn:aws:iam::{account_id}:role/{eks_service_role_name}",
    "fullManage.serviceRoleARN",
)
root = replace_line(
    root,
    r"^  serviceRoleName:\s*.*$",
    f"  serviceRoleName: {json.dumps(eks_service_role_name)}",
    "fullManage.serviceRoleName",
)
root = replace_line(
    root,
    r"^  assumeRoleName:\s*.*$",
    f"  assumeRoleName: {json.dumps(role_name)}",
    "fullManage.assumeRoleName",
)
root = replace_line(
    root,
    r"^  readonlyAssumeRoleName:\s*.*$",
    f"  readonlyAssumeRoleName: {json.dumps(role_name)}",
    "fullManage.readonlyAssumeRoleName",
)

# Dedicated SQL DNS is independent of the optional Kong HTTPS exposure.
public_dns = (ctx.get("terraform") or {}).get("dedicated_public_dns") or {}
public_zone_id = str(public_dns.get("zone_id") or "").strip()
public_zone_base = str(public_dns.get("domain") or "").strip().rstrip(".")
if str((ctx.get("host") or {}).get("mode") or "") == "remote-ec2-k3d" and not (public_zone_id and public_zone_base):
    raise SystemExit("terraform.dedicated_public_dns is required for remote Dedicated SQL endpoints")
if public_zone_id and public_zone_base:
    root = replace_line(
        root,
        r"^  publicHostedZoneDomain:\s*.*$",
        f"  publicHostedZoneDomain: {json.dumps(public_zone_base)}",
        "fullManage.publicHostedZoneDomain",
    )
    root = replace_line(
        root,
        r"^  publicHostedZoneID:\s*.*$",
        f"  publicHostedZoneID: {json.dumps(public_zone_id)}",
        "fullManage.publicHostedZoneID",
    )
if re.search(r"(?m)^  awsClusterServerRoleARN:", root):
    root = replace_line(
        root,
        r"^  awsClusterServerRoleARN:\s*.*$",
        f"  awsClusterServerRoleARN: {role_arn}",
        "fullManage.awsClusterServerRoleARN",
    )
else:
    root = replace_line(
        root,
        r"^  assumeRoleName:\s*(.*)$",
        rf"  assumeRoleName: \1\n  awsClusterServerRoleARN: {role_arn}",
        "fullManage.awsClusterServerRoleARN",
    )
root = replace_line(
    root,
    r"^  componentPCAARN:\s*.*$",
    f"  componentPCAARN: {component_pca_arn}",
    "fullManage.componentPCAARN",
)

json.dump({"data": {"root-config.yaml": root}}, open(patch_path, "w"))
PY

  KUBECONFIG="$kubeconfig" kubectl --context "$context" --namespace "$namespace" \
    patch configmap cluster-controller-root-config --type=merge -p "$(cat "$root_patch")"
  rm -f "$root_config" "$root_patch"
}

patch_starter_local_global_runtime_config() {
  session_context=${CP9_SESSION_CONTEXT:-}
  patch_serverless_auth0=false
  case "$kustomize_path" in
    */envs/dev-full/serverless-service|envs/dev-full/serverless-service)
      configmaps="serverless-service serverless-service-brworker"
      patch_serverless_auth0=true
      ;;
    */envs/dev-full/serverless-global-service|envs/dev-full/serverless-global-service)
      configmaps="serverless-global-service-config serverless-global-service"
      ;;
    *) return 0 ;;
  esac
  if [ -z "$session_context" ] || [ "$namespace" = "-" ] || [ -z "$namespace" ]; then
    return 0
  fi

  if [ "$patch_serverless_auth0" = "true" ]; then
    serverless_auth0_secret=$(mktemp)
    serverless_auth0_meta=$(mktemp)
    python3 - "$session_context" "$serverless_auth0_meta" <<'PY_AUTH0_META'
import json
import os
import pathlib
import sys

ctx_path, meta_path = sys.argv[1:3]
ctx = json.load(open(ctx_path))
outputs_path = str(((ctx.get("terraform") or {}).get("outputs_path")) or "").strip()
if not outputs_path or not pathlib.Path(outputs_path).is_file():
    raise SystemExit("Starter serverless Auth0 config requires terraform.outputs_path")
outputs = json.load(open(outputs_path))
runtime_patch_inputs = ((outputs.get("runtime_patch_inputs") or {}).get("value") or {})
aws = runtime_patch_inputs.get("aws") or {}
account_secret = ((outputs.get("account_config_secret") or {}).get("value") or {})
remote = ((ctx.get("host") or {}).get("remote") or {})
profile = str(remote.get("profile") or aws.get("profile") or "").strip()
region = str(((ctx.get("target") or {}).get("region")) or aws.get("region") or "").strip()
if not region:
    raise SystemExit("Starter serverless Auth0 config requires target.region")
if os.environ.get("CP9_AWS_RUNTIME_IDENTITY") != "ec2-instance-profile" and not profile:
    raise SystemExit("Starter serverless Auth0 config requires host.remote.profile or runtime_patch_inputs.aws.profile")
secret_id = str(account_secret.get("name") or "").strip()
if not secret_id:
    raise SystemExit("Starter serverless Auth0 config requires terraform output account_config_secret.name")
json.dump({"profile": profile, "region": region, "secret_id": secret_id}, open(meta_path, "w"))
PY_AUTH0_META
    fetch_session_secret_string "$serverless_auth0_meta" "$serverless_auth0_secret" 'Starter serverless Auth0'
    [ -s "$serverless_auth0_secret" ] || {
      echo "Starter serverless Auth0 secret is empty" >&2
      exit 1
    }
  fi

  patched=false
  for configmap in $configmaps; do
    config_json=$(mktemp)
    config_patch=$(mktemp)
    if ! KUBECONFIG="$kubeconfig" kubectl --context "$context" --namespace "$namespace" \
      get configmap "$configmap" -o json >"$config_json" 2>/dev/null; then
      rm -f "$config_json" "$config_patch"
      continue
    fi

    python3 - "$config_json" "$config_patch" "$session_context" "$configmap" "$serverless_auth0_secret" <<'PY'
import json
import pathlib
import re
import sys

config_path, patch_path, ctx_path, configmap, auth0_secret_path = sys.argv[1:6]
configmap_json = json.load(open(config_path))
ctx = json.load(open(ctx_path))
config = ((configmap_json.get("data") or {}).get("config.yaml") or "")
if not config:
    raise SystemExit(f"{configmap} missing data.config.yaml")

component_pca_arn = str((((ctx.get("terraform") or {}).get("session_product_pca") or {}).get("arn")) or "").strip()
if not re.match(r"^arn:aws:acm-pca:[^:]+:\d{12}:certificate-authority/.+", component_pca_arn):
    raise SystemExit("CP9_SESSION_CONTEXT terraform.session_product_pca.arn must be an ACM PCA ARN")
authority_id = component_pca_arn.rsplit("/", 1)[-1]
if authority_id == "cp9-local" or authority_id.startswith("cp9-"):
    raise SystemExit("CP9_SESSION_CONTEXT terraform.session_product_pca.arn must not be a cp9-local placeholder")
runtime_kms_arn = str((((ctx.get("terraform") or {}).get("session_cmk") or {}).get("arn")) or "").strip()
if not re.match(r"^arn:aws(?:-[a-z0-9-]+)?:kms:[^:]+:\d{12}:key/.+", runtime_kms_arn):
    raise SystemExit(
        "CP9_SESSION_CONTEXT terraform.session_cmk.arn "
        "must be a KMS key ARN"
    )

outputs_path = str(((ctx.get("terraform") or {}).get("outputs_path")) or "").strip()
outputs = {}
if outputs_path:
    try:
        outputs = json.load(open(outputs_path))
    except FileNotFoundError:
        outputs = {}
remote_eks = ((outputs.get("remote_eks") or {}).get("value") or {})
remote_cluster = str(remote_eks.get("cluster_name") or "").strip()
session_name = str(ctx.get("session_name") or "").strip()
if not session_name:
    raise SystemExit("CP9_SESSION_CONTEXT session_name is required")
if not remote_cluster:
    remote_cluster = f"{session_name}-starter-regional"

cluster_proxy_url = "dbaas-proxy.tidb-admin.svc/proxy/namespace/tidb-serverless/cluster/serverless-cluster"

config, count = re.subn(
    r"(?m)^(?P<indent>\s*)pcaArn:\s*.*$",
    lambda match: f"{match.group('indent')}pcaArn: {component_pca_arn}",
    config,
    count=1,
)
if count != 1:
    raise SystemExit(f"{configmap} config missing pcaArn")
config, count = re.subn(
    r"(?m)^(?P<indent>\s*)cmkArn:\s*.*$",
    lambda match: f"{match.group('indent')}cmkArn: {runtime_kms_arn}",
    config,
    count=1,
)
if count != 1:
    raise SystemExit(f"{configmap} config missing cmkArn")

requires_local_k8s_config = configmap in {
    "serverless-service",
    "serverless-service-brworker",
}

if requires_local_k8s_config:
    config, count = re.subn(
        r"(?m)^(?P<indent>\s*)stage:\s*.*$",
        lambda match: (
            f"{match.group('indent')}stage: staging\n"
            f"{match.group('indent')}stashingAreaBucketPrefix: {session_name}\n"
            f"{match.group('indent')}dmStorageBucketPrefix: {session_name}"
        ),
        config,
        count=1,
    )
    if count != 1:
        raise SystemExit(f"{configmap} config missing server.stage")

if requires_local_k8s_config or "kubernetes:" in config:
    config, count = re.subn(
        r"(?m)^(?P<indent>\s*(?:-\s*)?)kubernetes:\s*.*$",
        lambda match: f"{match.group('indent')}kubernetes: {remote_cluster}",
        config,
        count=1,
    )
    if count != 1:
        raise SystemExit(f"{configmap} config missing or invalid k8sConfig kubernetes")

if requires_local_k8s_config or "clusterProxyUrl:" in config:
    config, count = re.subn(
        r"(?m)^(?P<indent>\s*)clusterProxyUrl:\s*.*$",
        lambda match: f"{match.group('indent')}clusterProxyUrl: {cluster_proxy_url}",
        config,
        count=1,
    )
    if count != 1:
        raise SystemExit(f"{configmap} config missing or invalid k8sConfig clusterProxyUrl")

if requires_local_k8s_config:
    regional_url_path = pathlib.Path(ctx_path).parent.parent / "reports" / "evidence" / "starter-serverless-regional-service-url.txt"
    if regional_url_path.is_file():
        regional_url = regional_url_path.read_text().strip()
        if not re.match(r"^https?://[A-Za-z0-9.-]+(?::[0-9]+)?$", regional_url):
            raise SystemExit("Starter regional service URL is invalid")
        config, count = re.subn(
            r"(?m)^(?P<indent>\s*)k8sIngressUrl:\s*.*$",
            lambda match: f"{match.group('indent')}k8sIngressUrl: {regional_url}",
            config,
            count=1,
        )
        if count != 1:
            raise SystemExit(f"{configmap} config missing k8sIngressUrl")

    raw_secret = open(auth0_secret_path).read().strip()
    body_match = re.search(r"(?m)^auth0:\s*\n((?:[ \t]+.*(?:\n|$))*)", raw_secret)
    if not body_match:
        raise SystemExit("Starter serverless Auth0 secret missing top-level auth0 block")
    body = body_match.group(1)
    auth0 = {}
    for key in ("auth0Secret", "proxyAuth0ClientSecret"):
        match = re.search(rf"(?m)^\s*{re.escape(key)}:\s*(.*?)\s*$", body)
        if match:
            value = match.group(1).strip()
            if (value.startswith('"') and value.endswith('"')) or (value.startswith("'") and value.endswith("'")):
                value = value[1:-1]
            auth0[key] = value
    for key in ("auth0Secret", "proxyAuth0ClientSecret"):
        value = auth0.get(key, "").strip()
        if not value or value == "cp9-local":
            raise SystemExit(f"Starter serverless Auth0 secret missing auth0.{key}")
        config, count = re.subn(
            rf"(?m)^(?P<indent>\s*){re.escape(key)}:\s*.*$",
            lambda match, secret=value, field=key: f"{match.group('indent')}{field}: {json.dumps(secret)}",
            config,
            count=1,
        )
        if count != 1:
            raise SystemExit(f"{configmap} config missing auth0.{key}")

if configmap == "serverless-service-brworker":
    proxy_host_path = pathlib.Path(ctx_path).parent.parent / "reports" / "evidence" / "starter-dbaas-proxy-host.txt"
    if not proxy_host_path.is_file() or not proxy_host_path.read_text().strip():
        config, count = re.subn(
            r"(?m)^(?P<indent>[ \t]*)k8sConfig:\s*\n(?:(?P=indent)[ \t]+.*(?:\n|$))+",
            lambda match: f"{match.group('indent')}k8sConfig: []\n",
            config,
            count=1,
        )
        if count != 1:
            raise SystemExit(f"{configmap} config missing k8sConfig")

json.dump({"data": {"config.yaml": config}}, open(patch_path, "w"))
PY

    KUBECONFIG="$kubeconfig" kubectl --context "$context" --namespace "$namespace" \
      patch configmap "$configmap" --type=merge -p "$(cat "$config_patch")"
    patched=true
    rm -f "$config_json" "$config_patch"
  done

  if [ "$patch_serverless_auth0" = "true" ]; then
    rm -f "$serverless_auth0_secret" "$serverless_auth0_meta"
    serverless_auth0_secret=
    serverless_auth0_meta=
  fi

  if [ "$patched" != "true" ]; then
    echo "expected Starter local/global ConfigMap for $kustomize_path" >&2
    exit 1
  fi

}

patch_dedicated_proxy_auth_secret() {
  case "$kustomize_path" in
    */envs/dev-full/dedicated-cluster-control-server|envs/dev-full/dedicated-cluster-control-server) ;;
    *) return 0 ;;
  esac

  session_context=${CP9_SESSION_CONTEXT:-}
  if [ -z "$session_context" ]; then
    echo "CP9_SESSION_CONTEXT is required for Dedicated proxy Auth0" >&2
    exit 1
  fi

  dedicated_proxy_secret=$(mktemp)
  dedicated_proxy_meta=$(mktemp)
  python3 - "$session_context" "$dedicated_proxy_meta" <<'PY_DEDICATED_PROXY_META'
import json
import os
import pathlib
import sys

ctx_path, meta_path = sys.argv[1:3]
ctx = json.load(open(ctx_path))
outputs_path = str(((ctx.get("terraform") or {}).get("outputs_path")) or "").strip()
if not outputs_path or not pathlib.Path(outputs_path).is_file():
    raise SystemExit("Dedicated proxy Auth0 requires terraform.outputs_path")
outputs = json.load(open(outputs_path))
account_secret = ((outputs.get("account_config_secret") or {}).get("value") or {})
secret_id = str(account_secret.get("name") or "").strip()
if not secret_id:
    raise SystemExit("Dedicated proxy Auth0 requires terraform output account_config_secret.name")

runtime_patch_inputs = ((outputs.get("runtime_patch_inputs") or {}).get("value") or {})
aws = runtime_patch_inputs.get("aws") or {}
remote = ((ctx.get("host") or {}).get("remote") or {})
profile = str(remote.get("profile") or aws.get("profile") or "").strip()
region = str(((ctx.get("target") or {}).get("region")) or aws.get("region") or "").strip()
if not region:
    raise SystemExit("Dedicated proxy Auth0 requires target.region")
if os.environ.get("CP9_AWS_RUNTIME_IDENTITY") != "ec2-instance-profile" and not profile:
    raise SystemExit("Dedicated proxy Auth0 requires host.remote.profile")

json.dump({"profile": profile, "region": region, "secret_id": secret_id}, open(meta_path, "w"))
PY_DEDICATED_PROXY_META

  fetch_session_secret_string "$dedicated_proxy_meta" "$dedicated_proxy_secret" 'Dedicated proxy Auth0'

  python3 - "$dedicated_proxy_secret" <<'PY_DEDICATED_PROXY_SECRET' |
import base64
import json
import re
import sys

text = open(sys.argv[1]).read()

def block(name):
    match = re.search(rf"(?m)^{re.escape(name)}:\n((?:  .*(?:\n|$))*)", text)
    return match.group(1) if match else ""

def value(body, key):
    match = re.search(rf"(?m)^  {re.escape(key)}:\s*(.*?)\s*$", body)
    if not match:
        return ""
    result = match.group(1).strip()
    if (result.startswith('"') and result.endswith('"')) or (result.startswith("'") and result.endswith("'")):
        result = result[1:-1]
    return result

proxy = block("proxy")
auth0 = block("auth0")
client_id = value(proxy, "auth0ClientID") or value(auth0, "proxyAuth0ClientID")
client_secret = value(proxy, "auth0ClientSecret") or value(auth0, "proxyAuth0ClientSecret")
if not client_id or not client_secret or client_secret == "cp9-local":
    raise SystemExit("Dedicated proxy Auth0 session secret is incomplete")

def encoded(raw):
    return base64.b64encode(raw.encode()).decode()

print(json.dumps({
    "apiVersion": "v1",
    "kind": "Secret",
    "metadata": {"name": "dedicated-proxy-auth0-client"},
    "type": "Opaque",
    "data": {
        "Auth0ClientID": encoded(client_id),
        "Auth0ClientSecret": encoded(client_secret),
    },
}))
PY_DEDICATED_PROXY_SECRET
    KUBECONFIG="$kubeconfig" kubectl --context "$context" --namespace "$namespace" apply -f -

  rm -f "$dedicated_proxy_secret" "$dedicated_proxy_meta"
  dedicated_proxy_secret=
  dedicated_proxy_meta=
}

patch_account_secret_store_config() {
  case "$kustomize_path" in
    */envs/dev-full/account-svc-config|envs/dev-full/account-svc-config) ;;
    *) return 0 ;;
  esac
  session_context=${CP9_SESSION_CONTEXT:-}
  if [ -z "$session_context" ] || [ "$namespace" = "-" ] || [ -z "$namespace" ]; then
    return 0
  fi
  if ! command -v aws >/dev/null 2>&1; then
    echo "aws CLI is required to materialize the account-central session secret" >&2
    exit 1
  fi

  config_json=$(mktemp)
  config_patch=$(mktemp)
  account_meta=$(mktemp)
  aws_creds_json=$(mktemp)
  aws_env=$(mktemp)
  account_m2m_secret_json=$(mktemp)
  shared_bundle_secret_json=$(mktemp)
  session_secret_payload=$(mktemp)
  if ! KUBECONFIG="$kubeconfig" kubectl --context "$context" --namespace "$namespace" \
    get configmap account-central-conf -o json >"$config_json" 2>/dev/null; then
    rm -f "$config_json" "$config_patch" "$account_meta" "$aws_creds_json" "$aws_env" "$account_m2m_secret_json" "$shared_bundle_secret_json" "$session_secret_payload"
    echo "expected account-central-conf ConfigMap for $kustomize_path" >&2
    exit 1
  fi
  if [ "${CP9_AWS_RUNTIME_IDENTITY:-}" != "ec2-instance-profile" ]; then
    if ! KUBECONFIG="$kubeconfig" kubectl --context "$context" --namespace "$namespace" \
      get secret cp9-aws-credentials -o json >"$aws_creds_json" 2>/dev/null; then
      rm -f "$config_json" "$config_patch" "$account_meta" "$aws_creds_json" "$aws_env" "$account_m2m_secret_json" "$shared_bundle_secret_json" "$session_secret_payload"
      echo "expected cp9-aws-credentials Secret before account-svc; check premium-source-aws-credentials dependency" >&2
      exit 1
    fi
  fi

  python3 - "$config_json" "$account_meta" "$session_context" <<'PY_ACCOUNT_META'
import json
import os
import pathlib
import sys

config_path, meta_path, ctx_path = sys.argv[1:4]
configmap_json = json.load(open(config_path))
ctx = json.load(open(ctx_path))
config = ((configmap_json.get("data") or {}).get("config.yaml") or "")
if not config:
    raise SystemExit("account-central-conf missing data.config.yaml")

region = str(((ctx.get("target") or {}).get("region")) or "").strip()
if not region:
    raise SystemExit("account-central config requires target.region")
outputs_path = str(((ctx.get("terraform") or {}).get("outputs_path")) or "").strip()
if not outputs_path or not pathlib.Path(outputs_path).is_file():
    raise SystemExit("account-central config requires terraform.outputs_path")
outputs = json.loads(pathlib.Path(outputs_path).read_text())

def output_value(name):
    raw = outputs.get(name) or {}
    if not isinstance(raw, dict):
        return None
    return raw.get("value")

account_secret = output_value("account_config_secret") or {}
session_secret_name = str((account_secret or {}).get("name") or "").strip()
if not session_secret_name:
    raise SystemExit("terraform output account_config_secret.name is required for account-central")

host = ctx.get("host") or {}
remote = host.get("remote") or {}
bootstrap_profile = str(remote.get("profile") or "").strip()

json.dump({
    "bootstrap_profile": bootstrap_profile,
    "region": region,
    "session_secret_name": session_secret_name,
}, open(meta_path, "w"))
PY_ACCOUNT_META

  region=$(python3 - "$account_meta" <<'PY'
import json
import sys
print(json.load(open(sys.argv[1]))["region"])
PY
)
  bootstrap_profile=$(python3 - "$account_meta" <<'PY'
import json
import sys
print(json.load(open(sys.argv[1]))["bootstrap_profile"])
PY
)
  account_m2m_secret_key=cp9/dev/auth0/account-central-management-m2m
  shared_bundle_secret_key=cp9/dev/auth0/shared-bundle
  session_secret_name=$(python3 - "$account_meta" <<'PY'
import json
import sys
print(json.load(open(sys.argv[1]))["session_secret_name"])
PY
)

  if [ "${CP9_AWS_RUNTIME_IDENTITY:-}" != "ec2-instance-profile" ]; then
    python3 - "$aws_creds_json" "$aws_env" "$region" <<'PY_ACCOUNT_AWS'
import base64
import json
import shlex
import sys

secret_path, env_path, region = sys.argv[1:4]
secret = json.load(open(secret_path))
data = secret.get("data") or {}

def decode(name):
    value = data.get(name) or ""
    if not value:
        return ""
    return base64.b64decode(value).decode().strip()

values = {
    "AWS_ACCESS_KEY_ID": decode("AWS_ACCESS_KEY_ID"),
    "AWS_SECRET_ACCESS_KEY": decode("AWS_SECRET_ACCESS_KEY"),
    "AWS_SESSION_TOKEN": decode("AWS_SESSION_TOKEN"),
    "AWS_REGION": region,
    "AWS_DEFAULT_REGION": region,
}

if not values["AWS_ACCESS_KEY_ID"] or not values["AWS_SECRET_ACCESS_KEY"]:
    raise SystemExit("cp9-aws-credentials is missing AWS access key material")

with open(env_path, "w") as out:
    for key, value in values.items():
        if value:
            out.write(f"export {key}={shlex.quote(value)}\n")
PY_ACCOUNT_AWS
    . "$aws_env"
  fi

  verbose_external 'aws secretsmanager get-secret-value [account management M2M secret-id redacted]'
  verbose_external 'aws secretsmanager get-secret-value [shared Auth0 bundle secret-id redacted]'
  AWS_REGION="$region" AWS_DEFAULT_REGION="$region" \
    aws secretsmanager get-secret-value --secret-id "$account_m2m_secret_key" --output json >"$account_m2m_secret_json"
  AWS_REGION="$region" AWS_DEFAULT_REGION="$region" \
    aws secretsmanager get-secret-value --secret-id "$shared_bundle_secret_key" --output json >"$shared_bundle_secret_json"

  python3 - "$config_json" "$config_patch" "$account_m2m_secret_json" "$shared_bundle_secret_json" "$session_secret_payload" "$account_meta" <<'PY_ACCOUNT'
import json
import re
import sys

config_path, patch_path, shared_secret_path, serverless_secret_path, payload_path, meta_path = sys.argv[1:7]
configmap_json = json.load(open(config_path))
shared_secret = json.load(open(shared_secret_path)).get("SecretString") or ""
serverless_secret = json.load(open(serverless_secret_path)).get("SecretString") or ""
meta = json.load(open(meta_path))
config = ((configmap_json.get("data") or {}).get("config.yaml") or "")
if not config:
    raise SystemExit("account-central-conf missing data.config.yaml")

region = meta["region"]
session_secret_name = meta["session_secret_name"]

def block(text, name):
    match = re.search(rf"(?m)^{re.escape(name)}:\n((?:  .*(?:\n|$))*)", text)
    return match.group(1) if match else ""

def scalar_from_block(body, key, default=""):
    match = re.search(rf"(?m)^  {re.escape(key)}:\s*(.*?)\s*$", body)
    if not match:
        return default
    value = match.group(1).strip()
    if "#" in value:
        value = value.split("#", 1)[0].strip()
    if (value.startswith('"') and value.endswith('"')) or (value.startswith("'") and value.endswith("'")):
        value = value[1:-1]
    return value

def scalar(text, section, key, default=""):
    return scalar_from_block(block(text, section), key, default)

server_block = block(config, "server")
if not server_block:
    raise SystemExit("account config.yaml missing top-level server block")

auth0_block = block(config, "auth0")
if not auth0_block:
    raise SystemExit("account config.yaml missing top-level auth0 block")

def secret_auth0_values(text):
    values = {}
    try:
        obj = json.loads(text)
        auth0 = obj.get("auth0", obj) if isinstance(obj, dict) else {}
        if isinstance(auth0, dict):
            for key, value in auth0.items():
                values[str(key)] = str(value)
    except Exception:
        body = block(text, "auth0")
        for key in [
            "auth0ManagementClientID",
            "auth0ManagementClientSecret",
            "auth0DatabaseConnectionName",
            "auth0Domain",
            "auth0Audience",
            "auth0Secret",
            "managementPortalAuth0Domain",
            "managementPortalAuth0Audience",
            "managementPortalAuth0Secret",
            "managementPortalAuth0ClientAudience",
            "proxyAuth0Domain",
            "proxyAuth0Audience",
            "proxyAuth0ClientID",
            "proxyAuth0ClientSecret",
        ]:
            value = scalar_from_block(body, key, "")
            if not value:
                # Accept flat one-level YAML as a defensive fallback.
                match = re.search(rf"(?m)^{re.escape(key)}:\s*(.*?)\s*$", text)
                if match:
                    value = match.group(1).strip()
            if value:
                values[key] = value
    return values

shared_auth0 = secret_auth0_values(shared_secret)
for required in ["auth0ManagementClientID", "auth0ManagementClientSecret"]:
    if not shared_auth0.get(required, "").strip():
        raise SystemExit(f"shared account-central M2M secret missing auth0.{required}")
serverless_auth0 = secret_auth0_values(serverless_secret)
for required in ["auth0Secret", "proxyAuth0ClientSecret"]:
    if not serverless_auth0.get(required, "").strip():
        raise SystemExit(f"shared serverless Auth0 secret missing auth0.{required}")
shared_auth0.update(serverless_auth0)

def yaml_string(value):
    return json.dumps(str(value))

def auth0_value(key, default=""):
    return shared_auth0.get(key) or scalar_from_block(auth0_block, key, default)

poc_days = scalar_from_block(server_block, "pocManualBackupReservedDays", "7")
if not re.fullmatch(r"[0-9]+", poc_days):
    poc_days = "7"

session_secret = "\n".join([
    "server:",
    f"  frontendDNS: {yaml_string(scalar_from_block(server_block, 'frontendDNS', 'http://127.0.0.1'))}",
    f"  backendDNS: {yaml_string(scalar_from_block(server_block, 'backendDNS', 'http://127.0.0.1'))}",
    f"  stage: {yaml_string(scalar_from_block(server_block, 'stage', 'local'))}",
    f"  pocManualBackupReservedDays: {poc_days}",
    "  secret:",
    f"    secretKey: {yaml_string(session_secret_name)}",
    f"    region: {yaml_string(region)}",
    "auth0:",
    f"  auth0ManagementClientID: {yaml_string(auth0_value('auth0ManagementClientID'))}",
    f"  auth0ManagementClientSecret: {yaml_string(auth0_value('auth0ManagementClientSecret'))}",
    f"  auth0DatabaseConnectionName: {yaml_string(auth0_value('auth0DatabaseConnectionName', 'custom-email'))}",
    f"  auth0Domain: {yaml_string(auth0_value('auth0Domain', 'tidb-dev.us.auth0.com'))}",
    f"  auth0Audience: {yaml_string(auth0_value('auth0Audience'))}",
    f"  auth0Secret: {yaml_string(auth0_value('auth0Secret'))}",
    f"  managementPortalAuth0Domain: {yaml_string(auth0_value('managementPortalAuth0Domain', 'tidb-dev.us.auth0.com'))}",
    f"  managementPortalAuth0Audience: {yaml_string(auth0_value('managementPortalAuth0Audience'))}",
    f"  managementPortalAuth0Secret: {yaml_string(auth0_value('managementPortalAuth0Secret'))}",
    f"  managementPortalAuth0ClientAudience: {yaml_string(auth0_value('managementPortalAuth0ClientAudience'))}",
    f"  proxyAuth0Domain: {yaml_string(auth0_value('proxyAuth0Domain', 'tidb-dev.us.auth0.com'))}",
    f"  proxyAuth0Audience: {yaml_string(auth0_value('proxyAuth0Audience', 'dev-proxy-service'))}",
    f"  proxyAuth0ClientID: {yaml_string(auth0_value('proxyAuth0ClientID', 'qetSCop5DfAOsK2S7WKTCre9iBvSdNmo'))}",
    f"  proxyAuth0ClientSecret: {yaml_string(auth0_value('proxyAuth0ClientSecret'))}",
    "proxy:",
    f"  auth0Domain: {yaml_string(auth0_value('proxyAuth0Domain', 'tidb-dev.us.auth0.com'))}",
    f"  auth0Audience: {yaml_string(auth0_value('proxyAuth0Audience', 'dev-proxy-service'))}",
    f"  auth0ClientID: {yaml_string(auth0_value('proxyAuth0ClientID', 'qetSCop5DfAOsK2S7WKTCre9iBvSdNmo'))}",
    f"  auth0ClientSecret: {yaml_string(auth0_value('proxyAuth0ClientSecret'))}",
    "",
])
open(payload_path, "w").write(session_secret)

# account-central reads server.secret from AWS Secrets Manager at startup and
# merges that YAML into its config. Point the runtime at the session-scoped
# secret we materialize above, not the partial shared Auth0 secret. The shared
# secret remains the source of the M2M credential only; the session secret keeps
# enough server/auth0 fields to survive account's current partial-merge behavior.
secret_block = f"  secret:\n    secretKey: {session_secret_name}\n    region: {region}\n"
if re.search(r"(?m)^  secret:\n", config):
    config = re.sub(r"(?m)^  secret:\n(?:    .*\n)+", secret_block, config, count=1)
else:
    config, count = re.subn(r"(?m)^server:\n", "server:\n" + secret_block, config, count=1)
    if count != 1:
        raise SystemExit("account config.yaml missing top-level server block")

json.dump({"data": {"config.yaml": config}}, open(patch_path, "w"))
PY_ACCOUNT

  AWS_REGION="$region" AWS_DEFAULT_REGION="$region" \
    aws secretsmanager put-secret-value --secret-id "$session_secret_name" --secret-string "file://$session_secret_payload" >/dev/null

  KUBECONFIG="$kubeconfig" kubectl --context "$context" --namespace "$namespace" \
    patch configmap account-central-conf --type=merge -p "$(cat "$config_patch")"
  rm -f "$config_json" "$config_patch" "$account_meta" "$aws_creds_json" "$aws_env" "$account_m2m_secret_json" "$shared_bundle_secret_json" "$session_secret_payload"

}

patch_dedicated_session_contract_before_apply() {
  session_context=${CP9_SESSION_CONTEXT:-}
  case "$kustomize_path" in
    */envs/dev-full/dedicated-cluster-control-server|envs/dev-full/dedicated-cluster-control-server|\
    */envs/dev-full/dedicated-cluster-controller|envs/dev-full/dedicated-cluster-controller|\
    */envs/dev-full/dedicated-ticdc-service|envs/dev-full/dedicated-ticdc-service|\
    */envs/dev-full/dedicated-global-service|envs/dev-full/dedicated-global-service|\
    */envs/dev-full/regional-server|envs/dev-full/regional-server) ;;
    *) return 0 ;;
  esac
  if [ -z "$session_context" ]; then
    echo "CP9_SESSION_CONTEXT is required before applying Dedicated cluster controllers" >&2
    exit 1
  fi

  python3 - "$tmp" "$session_context" "$kustomize_path" <<'PY'
import ipaddress
import json
import os
import re
import sys

manifest_path, ctx_path, kustomize_path = sys.argv[1:4]
ctx = json.load(open(ctx_path))
session = str((ctx.get("session_tags") or {}).get("cp9-session") or ctx.get("session_name") or "").strip()
owner = str((ctx.get("session_tags") or {}).get("cp9-owner") or "").strip()
ttl = str(ctx.get("ttl_expires_at") or "").strip()
if not session or not owner or not ttl:
    raise SystemExit("cp9-owner, cp9-session, and ttl_expires_at are required before applying Dedicated controllers")
text = open(manifest_path).read()
path = kustomize_path.rstrip("/")

if path.endswith("dedicated-cluster-control-server"):
    text, count = re.subn(r"(?m)^(\s*- --ns-prefix=).*$", rf"\g<1>{session}-", text)
    if count != 1:
        raise SystemExit(f"expected one cluster-control-server --ns-prefix argument, found {count}")

if path.endswith(("dedicated-cluster-controller", "dedicated-ticdc-service", "dedicated-global-service", "regional-server")):
    target_account = str((ctx.get("target") or {}).get("account") or "").strip()
    role_arn = str(((ctx.get("terraform") or {}).get("iam") or {}).get("role_arn") or "").strip()
    role_match = re.fullmatch(r"arn:aws:iam::(\d{12}):role/[^/]+", role_arn)
    if not re.fullmatch(r"\d{12}", target_account):
        raise SystemExit("CP9_SESSION_CONTEXT.target.account must be a 12-digit AWS account id")
    if not role_match or role_match.group(1) != target_account:
        raise SystemExit(
            "CP9_SESSION_CONTEXT terraform.iam.role_arn "
            "must be an IAM role ARN in target.account"
        )
    target_region = str(((ctx.get("target") or {}).get("region")) or "").strip()
    if not target_region:
        raise SystemExit("CP9_SESSION_CONTEXT.target.region is required for Dedicated runtime")
    terraform = ctx.get("terraform") or {}
    pca = terraform.get("session_product_pca") or {}
    pca_arn = str(pca.get("arn") or "").strip()
    arn_partition = r"aws(?:-[a-z0-9-]+)?"
    if not re.fullmatch(
        rf"arn:{arn_partition}:acm-pca:{re.escape(target_region)}:{target_account}:"
        r"certificate-authority/[0-9a-fA-F-]+",
        pca_arn,
    ):
        raise SystemExit(
            "CP9_SESSION_CONTEXT.terraform.session_product_pca.arn "
            "must be a target-account ACM PCA ARN"
        )

if path.endswith("dedicated-global-service"):
    kms = ((ctx.get("terraform") or {}).get("session_cmk") or {})
    kms_arn = str(kms.get("arn") or "").strip()
    if not re.fullmatch(
        rf"arn:{arn_partition}:kms:{re.escape(target_region)}:{target_account}:"
        r"key/[0-9a-fA-F-]+",
        kms_arn,
    ):
        raise SystemExit(
            "CP9_SESSION_CONTEXT.terraform.session_cmk.arn "
            "must be a target-account KMS key ARN"
        )
    outputs_path = str((ctx.get("terraform") or {}).get("outputs_path") or "").strip()
    outputs = json.load(open(outputs_path)) if outputs_path else {}
    cluster_certificates_bucket = str(
        (
            (outputs.get("cluster_certificates_storage") or {}).get("value")
            or {}
        ).get("bucket")
        or ""
    ).strip()
    if not cluster_certificates_bucket:
        raise SystemExit(
            "Dedicated global-service requires terraform output "
            "cluster_certificates_storage.bucket"
        )

if path.endswith("dedicated-cluster-controller"):
    role_segment = re.sub(r"[^A-Za-z0-9]+", "-", session).strip("-")
    eks_service_role_name = (role_segment + "-eks-service").strip("-")[:64].strip("-")
    if not eks_service_role_name:
        raise SystemExit("unable to derive Dedicated EKS service role name")

    text, service_role_count = re.subn(
        r"(?m)^(\s*serviceRoleARN:\s*).*$",
        rf"\g<1>arn:aws:iam::{target_account}:role/{eks_service_role_name}",
        text,
    )
    text, cluster_server_role_count = re.subn(
        r"(?m)^(\s*awsClusterServerRoleARN:\s*).*$",
        lambda match: match.group(1) + role_arn,
        text,
    )
    if service_role_count != 1 or cluster_server_role_count != 1:
        raise SystemExit(
            "Dedicated cluster-controller manifest must contain exactly one "
            "serviceRoleARN and one awsClusterServerRoleARN"
        )
    text, pca_count = re.subn(
        r"(?m)^(\s*componentPCAARN:\s*).*$",
        lambda match: match.group(1) + pca_arn,
        text,
    )
    if pca_count != 1:
        raise SystemExit(
            "Dedicated cluster-controller manifest must contain exactly one componentPCAARN"
        )

    text, count = re.subn(
        r"(?m)^(\s*)tagSpecification:\s*Vendor=PingCAP\s*$",
        lambda match: (
            f"{match.group(1)}tagSpecification: Vendor=PingCAP\n"
            f"{match.group(1)}tagSpecification_cp9_managed: cp9-managed=true\n"
            f"{match.group(1)}tagSpecification_cp9_owner: cp9-owner={owner}\n"
            f"{match.group(1)}tagSpecification_cp9_session: cp9-session={session}\n"
            f"{match.group(1)}tagSpecification_cp9_ttl: cp9-ttl={ttl}"
        ),
        text,
    )
    if count != 2:
        raise SystemExit(f"expected two Dedicated StorageClass tagSpecification entries, found {count}")

if path.endswith("dedicated-ticdc-service"):
    role_name = role_arn.rsplit("/", 1)[-1]
    replacements = {
        "clusterNSPrefix": session,
        "assumeRoleName": role_name,
        "externalID": session,
        "CP9_RUNTIME_PROVIDER_SOURCE": (
            "PodIdentity"
            if os.environ.get("CP9_AWS_RUNTIME_IDENTITY") == "ec2-instance-profile"
            else "Secret"
        ),
    }
    for key, value in replacements.items():
        if key.startswith("CP9_RUNTIME_"):
            count = text.count(key)
            text = text.replace(key, value)
        else:
            text, count = re.subn(
                rf"(?m)^(\s*{key}:\s*).*$",
                lambda match, value=value: match.group(1) + json.dumps(value),
                text,
            )
        if count != 1:
            raise SystemExit(f"Dedicated TiCDC manifest must contain exactly one {key}")

if path.endswith("regional-server"):

    remote_public_ip = str((((ctx.get("host") or {}).get("remote") or {}).get("public_ip")) or "").strip()
    try:
        parsed_public_ip = ipaddress.ip_address(remote_public_ip)
    except ValueError as exc:
        raise SystemExit(
            "CP9_SESSION_CONTEXT host.remote.public_ip must be a valid public IPv4 address"
        ) from exc
    if parsed_public_ip.version != 4 or not parsed_public_ip.is_global:
        raise SystemExit(
            "CP9_SESSION_CONTEXT host.remote.public_ip must be a valid public IPv4 address"
        )
    public_cidr = f"{parsed_public_ip}/32"
    allowlist_count = text.count("CP9_REMOTE_PUBLIC_CIDR")
    if allowlist_count != 2:
        raise SystemExit(
            "regional-server Dedicated template must contain exactly two CP9_REMOTE_PUBLIC_CIDR entries"
        )
    text = text.replace("CP9_REMOTE_PUBLIC_CIDR", public_cidr)

    text, subject_count = re.subn(
        r"(?m)^(\s*-\s*)arn:aws:iam::\d{12}:role/[^\s]+$",
        lambda match: match.group(1) + role_arn,
        text,
        count=1,
    )
    text, pca_count = re.subn(
        r"(?m)^(\s*componentPCAARN:\s*).*$",
        lambda match: match.group(1) + pca_arn,
        text,
    )
    if subject_count != 1 or pca_count != 1:
        raise SystemExit(
            "regional-server manifest must contain exactly one authorized subject "
            "and one componentPCAARN"
        )

if path.endswith("dedicated-global-service"):
    text, pca_count = re.subn(
        r"(?m)^(\s*pcaArn:\s*).*$",
        lambda match: match.group(1) + pca_arn,
        text,
    )
    text, kms_count = re.subn(
        r"(?m)^(\s*cmkArn:\s*).*$",
        lambda match: match.group(1) + kms_arn,
        text,
    )
    text, cluster_certificates_count = re.subn(
        r"(?m)^(\s*clusterCertsConfig:\n\s*bucket:\s*).*$",
        lambda match: match.group(1) + cluster_certificates_bucket,
        text,
        count=1,
    )
    if pca_count != 1 or kms_count != 1 or cluster_certificates_count != 1:
        raise SystemExit(
            "Dedicated global-service manifest must contain exactly one pcaArn, "
            "one cmkArn, and one cluster certificates bucket"
        )
    # The session CMK is regional, so the service must call KMS in the key's region.
    text = text.replace("CP9_RUNTIME_REGION", target_region)

open(manifest_path, "w").write(text)
PY

}

patch_starter_session_contract_before_apply() {
  session_context=${CP9_SESSION_CONTEXT:-}
  case "$kustomize_path" in
    */envs/dev-full/serverless-service|envs/dev-full/serverless-service|\
    */envs/dev-full/serverless-global-service|envs/dev-full/serverless-global-service) ;;
    *) return 0 ;;
  esac
  if [ -z "$session_context" ]; then
    echo "CP9_SESSION_CONTEXT is required for Starter runtime config" >&2
    exit 1
  fi

  python3 - "$tmp" "$session_context" <<'PY'
import json
import re
import sys

manifest_path, ctx_path = sys.argv[1:3]
ctx = json.load(open(ctx_path))
target = ctx.get("target") or {}
account_id = str(target.get("account") or "").strip()
region = str(target.get("region") or "").strip()
terraform = ctx.get("terraform") or {}
pca_arn = str(((terraform.get("session_product_pca") or {}).get("arn")) or "").strip()
kms_arn = str(((terraform.get("session_cmk") or {}).get("arn")) or "").strip()
arn_partition = r"aws(?:-[a-z0-9-]+)?"
if not re.fullmatch(r"\d{12}", account_id) or not region:
    raise SystemExit("Starter runtime requires target.account and target.region")
if not re.fullmatch(
    rf"arn:{arn_partition}:acm-pca:{re.escape(region)}:{account_id}:"
    r"certificate-authority/[0-9a-fA-F-]+",
    pca_arn,
):
    raise SystemExit("Starter runtime requires a target-account session_product_pca ARN")
if not re.fullmatch(
    rf"arn:{arn_partition}:kms:{re.escape(region)}:{account_id}:key/[0-9a-fA-F-]+",
    kms_arn,
):
    raise SystemExit("Starter runtime requires a target-account session_cmk ARN")

text = open(manifest_path).read()
text, pca_count = re.subn(
    r"(?m)^(\s*pcaArn:\s*).*$",
    lambda match: match.group(1) + pca_arn,
    text,
)
text, kms_count = re.subn(
    r"(?m)^(\s*cmkArn:\s*).*$",
    lambda match: match.group(1) + kms_arn,
    text,
)
if pca_count < 1 or kms_count < 1:
    raise SystemExit(
        "Starter runtime manifest must contain at least one pcaArn and one cmkArn"
    )
# The session CMK is regional, so the service must call KMS in the key's region.
text = text.replace("CP9_RUNTIME_REGION", region)
open(manifest_path, "w").write(text)
PY
}

patch_account_contract_before_apply() {
  session_context=${CP9_SESSION_CONTEXT:-}
  case "$kustomize_path" in
    */envs/dev-full/premium-metadb|envs/dev-full/premium-metadb|\
    */envs/dev-full/dataflow|envs/dev-full/dataflow|\
    */envs/dev-full/dedicated-dataflow-svc|envs/dev-full/dedicated-dataflow-svc|\
    */envs/dev-full/account-provider|envs/dev-full/account-provider|\
    */envs/dev-full/cluster-ng-server|envs/dev-full/cluster-ng-server|\
    */envs/dev-full/cluster-ng-executor|envs/dev-full/cluster-ng-executor|\
    */envs/dev-full/nextgen-global-service|envs/dev-full/nextgen-global-service) ;;
    *) return 0 ;;
  esac
  if [ -z "$session_context" ]; then
    echo "CP9_SESSION_CONTEXT is required for account-scoped resources" >&2
    exit 1
  fi

  python3 - "$tmp" "$session_context" "$kustomize_path" <<'PY'
import json
import re
import sys

manifest_path, ctx_path, kustomize_path = sys.argv[1:4]
ctx = json.load(open(ctx_path))
account_id = str(((ctx.get("target") or {}).get("account") or "")).strip()
if not re.fullmatch(r"\d{12}", account_id):
    raise SystemExit("CP9_SESSION_CONTEXT.target.account must be a 12-digit AWS account id")

text = open(manifest_path).read()
account_placeholder = "000000000000"
count = text.count(account_placeholder)
if count == 0:
    raise SystemExit("Premium account-scoped manifest is missing the account placeholder")
text = text.replace(account_placeholder, account_id)
path = kustomize_path.rstrip("/")
if "__CP9_SESSION_NAME__" in text:
    session_name = str(ctx.get("session_name") or "").strip()
    if not session_name:
        raise SystemExit("CP9_SESSION_CONTEXT.session_name is required for account-scoped O11Y")
    text = text.replace("__CP9_SESSION_NAME__", session_name)
pca_placeholder = "CP9_SESSION_PCA_ARN"
pca_arn = ""
if pca_placeholder in text:
    target_region = str(((ctx.get("target") or {}).get("region") or "")).strip()
    pca_arn = str(
        (((ctx.get("terraform") or {}).get("session_product_pca") or {}).get("arn") or "")
    ).strip()
    if not re.fullmatch(
        rf"arn:aws(?:-[a-z0-9-]+)?:acm-pca:{re.escape(target_region)}:{account_id}:"
        r"certificate-authority/[0-9a-fA-F-]+",
        pca_arn,
    ):
        raise SystemExit("Premium account-scoped manifest requires a target-account session PCA ARN")
    text = text.replace(pca_placeholder, pca_arn)

if path.endswith("/dataflow") or path.endswith("/dedicated-dataflow-svc"):
    target = ctx.get("target") or {}
    region = str(target.get("region") or "").strip()
    session_name = str(ctx.get("session_name") or "").strip()
    terraform = ctx.get("terraform") or {}
    pca = terraform.get("session_product_pca") or {}
    kms = terraform.get("session_cmk") or {}
    pca_arn = str(pca.get("arn") or "").strip()
    pca_region = str(pca.get("region") or "").strip()
    kms_arn = str(kms.get("arn") or "").strip()
    kms_region = str(kms.get("region") or "").strip()

    if not region:
        raise SystemExit("CP9_SESSION_CONTEXT.target.region is required for Dataflow")
    if not session_name:
        raise SystemExit("CP9_SESSION_CONTEXT.session_name is required for Dataflow")
    if pca_region != region:
        raise SystemExit(
            "CP9_SESSION_CONTEXT.terraform.session_product_pca.region "
            f"must equal target.region {region!r}"
        )
    if kms_region != region:
        raise SystemExit(
            "CP9_SESSION_CONTEXT.terraform.session_cmk.region "
            f"must equal target.region {region!r}"
        )

    arn_partition = r"aws(?:-[a-z0-9-]+)?"
    if not re.fullmatch(
        rf"arn:{arn_partition}:acm-pca:{re.escape(region)}:{account_id}:"
        r"certificate-authority/[0-9a-fA-F-]+",
        pca_arn,
    ):
        raise SystemExit(
            "CP9_SESSION_CONTEXT.terraform.session_product_pca.arn "
            "must be a target-account ACM PCA ARN"
        )
    text, pca_count = re.subn(
        r"(?m)^(\s*pcaArn:\s*).+$",
        lambda match: match.group(1) + pca_arn,
        text,
    )
    if pca_count != 1:
        raise SystemExit(
            f"Dataflow manifest must contain exactly one pcaArn; found {pca_count}"
        )

    if path.endswith("/dedicated-dataflow-svc"):
        outputs_path = str(terraform.get("outputs_path") or "").strip()
        if not outputs_path:
            raise SystemExit("CP9_SESSION_CONTEXT.terraform.outputs_path is required for Dedicated Dataflow")
        outputs = json.load(open(outputs_path))
        account_secret = ((outputs.get("account_config_secret") or {}).get("value") or {})
        account_secret_name = str(account_secret.get("name") or "").strip()
        if not account_secret_name:
            raise SystemExit("terraform output account_config_secret.name is required for Dedicated Dataflow")

        replacements = {
            "CP9_SESSION_NAME": session_name,
            "CP9_RUNTIME_REGION": region,
            "CP9_ACCOUNT_CONFIG_SECRET": account_secret_name,
        }
        for old, new in replacements.items():
            count = text.count(old)
            if count == 0:
                raise SystemExit(f"Dedicated Dataflow manifest missing {old}")
            text = text.replace(old, new)

if path.endswith("/nextgen-global-service"):
    outputs_path = str(((ctx.get("terraform") or {}).get("outputs_path") or "")).strip()
    outputs = json.load(open(outputs_path)) if outputs_path else {}
    bucket = str(
        (((outputs.get("cluster_certificates_storage") or {}).get("value") or {}).get("bucket") or "")
    ).strip()
    if not pca_arn:
        raise SystemExit("nextgen global-service requires session_product_pca.arn")
    if not bucket:
        raise SystemExit("nextgen global-service requires cluster_certificates_storage.bucket")
    replacements = [
        (r"(?m)^(\s*pcaArn:\s*).*$", pca_arn),
        (r"(?m)^(\s*clusterCertsBucket:\s*).*$", bucket),
        (r"(?m)^(\s*clusterCertsConfig:\n\s*bucket:\s*).*$", bucket),
        (r"(?m)^(\s*defaultKey:\s*).*$", "ca.pem"),
    ]
    for pattern, value in replacements:
        text, count = re.subn(
            pattern,
            lambda match, value=value: match.group(1) + json.dumps(value),
            text,
            count=1,
        )
        if count != 1:
            raise SystemExit(f"nextgen global-service manifest missing {pattern}")
# Every account-scoped service carrying a cmkArn talks to real AWS KMS, so the session
# CMK has to reach all of them, not just Dataflow. The account substitution above only
# fixes the account segment, which would leave a service pointing at a key id that does
# not exist; real KMS answers that with AccessDenied, reading like a permissions bug.
# The key is regional, so any service holding its ARN must also target that region.
target_region = str((ctx.get("target") or {}).get("region") or "").strip()
session_cmk_placeholder = (
    f"arn:aws:kms:CP9_RUNTIME_REGION:{account_id}:key/cp9-session-placeholder"
)
if session_cmk_placeholder in text or re.search(r"(?m)^\s*cmkArn:\s*", text):
    session_cmk_arn = str(
        ((ctx.get("terraform") or {}).get("session_cmk") or {}).get("arn") or ""
    ).strip()
    if not re.fullmatch(
        rf"arn:aws(?:-[a-z0-9-]+)?:kms:{re.escape(target_region)}:{account_id}:key/[0-9a-fA-F-]+",
        session_cmk_arn,
    ):
        raise SystemExit(
            "CP9_SESSION_CONTEXT.terraform.session_cmk.arn must be a target-account "
            "KMS key ARN for every account-scoped service that uses a cmkArn"
        )
    text = text.replace(session_cmk_placeholder, session_cmk_arn)
    if re.search(r"(?m)^\s*cmkArn:\s*", text):
        text, cmk_count = re.subn(
            r"(?m)^(\s*cmkArn:\s*).*$",
            lambda match: match.group(1) + session_cmk_arn,
            text,
        )
        if cmk_count != 1:
            raise SystemExit(f"manifest must contain exactly one cmkArn; found {cmk_count}")

text = text.replace("CP9_RUNTIME_REGION", target_region)

open(manifest_path, "w").write(text)
PY
}

patch_metering_session_contract_before_apply() {
  case "$kustomize_path" in
    */envs/dev-full/cluster-ng-server|envs/dev-full/cluster-ng-server|\
    */envs/dev-full/cluster-ng-executor|envs/dev-full/cluster-ng-executor|\
    */envs/dev-full/nextgen-metering-server|envs/dev-full/nextgen-metering-server) ;;
    *) return 0 ;;
  esac

  session_context=${CP9_SESSION_CONTEXT:-}
  if [ -z "$session_context" ] || [ ! -f "$session_context" ]; then
    echo "CP9_SESSION_CONTEXT is required for Premium metering" >&2
    return 1
  fi

  python3 - "$tmp" "$session_context" <<'PY'
from datetime import datetime
import json
import pathlib
import re
import sys

manifest_path = pathlib.Path(sys.argv[1])
ctx = json.loads(pathlib.Path(sys.argv[2]).read_text())
target = ctx.get("target") or {}
terraform = ctx.get("terraform") or {}
metering = terraform.get("metering") or {}

session_name = str(ctx.get("session_name") or "").strip()
account = str(target.get("account") or "").strip()
region = str(target.get("region") or "").strip()
bucket = str(metering.get("bucket") or "").strip()
read_write_role = str(metering.get("read_write_role_arn") or "").strip()
writer_role = str(metering.get("writer_role_arn") or "").strip()

if not session_name:
    raise SystemExit("CP9_SESSION_CONTEXT.session_name is required for Premium metering")
if not re.fullmatch(r"[0-9]{12}", account):
    raise SystemExit("CP9_SESSION_CONTEXT.target.account must be a 12-digit AWS account id")
if not re.fullmatch(r"[a-z]{2}(?:-gov)?-[a-z]+-[0-9]", region):
    raise SystemExit("CP9_SESSION_CONTEXT.target.region must be an AWS region")
if not re.fullmatch(r"[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]", bucket):
    raise SystemExit("CP9_SESSION_CONTEXT.terraform.metering.bucket must be an S3 bucket name")
if bucket != f"{session_name}-metering-{account}":
    raise SystemExit("Premium metering bucket must be scoped to the Session and target account")

role_prefix = rf"arn:aws(?:-[a-z0-9-]+)?:iam::{account}:role/cp9-{re.escape(session_name)}-metering-"
for name, value, suffix in (
    ("read_write_role_arn", read_write_role, "read-write"),
    ("writer_role_arn", writer_role, "writer"),
):
    if not re.fullmatch(role_prefix + suffix, value):
        raise SystemExit(
            f"CP9_SESSION_CONTEXT.terraform.metering.{name} must be a Session-scoped target-account IAM role ARN"
        )

created_at = str(ctx.get("created_at") or "").strip()
start_time = ""
if created_at:
    try:
        parsed = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
    except ValueError as exc:
        raise SystemExit(f"CP9_SESSION_CONTEXT.created_at must be RFC3339: {exc}") from exc
    if parsed.tzinfo is None:
        raise SystemExit("CP9_SESSION_CONTEXT.created_at must include timezone")
    start_time = str(int(parsed.timestamp()) // 60 * 60)

values = {
    "__CP9_TARGET_REGION__": region,
    "__CP9_METERING_BUCKET__": bucket,
    "__CP9_METERING_READ_WRITE_ROLE_ARN__": read_write_role,
    "__CP9_METERING_WRITER_ROLE_ARN__": writer_role,
    "__CP9_METERING_START_TIME__": start_time,
}

text = manifest_path.read_text()
required = sorted(
    set(re.findall(r"__(?:CP9_TARGET_REGION|CP9_METERING_[A-Z0-9_]+)__", text))
)
unknown = sorted(set(required) - set(values))
if unknown:
    raise SystemExit("unknown Premium metering placeholders: " + ", ".join(unknown))
missing = [placeholder for placeholder in required if not values[placeholder]]
if missing:
    raise SystemExit("missing Premium metering session values: " + ", ".join(missing))
for placeholder in required:
    text = text.replace(placeholder, values[placeholder])
unresolved = sorted(set(re.findall(r"__CP9_METERING_[A-Z0-9_]+__", text)))
if unresolved:
    raise SystemExit("unresolved Premium metering placeholders: " + ", ".join(unresolved))
manifest_path.write_text(text)
PY
}

patch_o11y_session_contract_before_apply() {
  case "$kustomize_path" in
    */envs/dev-full/o11y-platform|envs/dev-full/o11y-platform|\
    */envs/dev-full/o11y-platform/gateway|envs/dev-full/o11y-platform/gateway|\
    */envs/dev-full/premium-o11y-agent-control-server|envs/dev-full/premium-o11y-agent-control-server|\
    */envs/dev-full/dedicated-o11y-agent-control-server|envs/dev-full/dedicated-o11y-agent-control-server|\
    */envs/dev-full/regional-server|envs/dev-full/regional-server) ;;
    *) return 0 ;;
  esac

  session_context=${CP9_SESSION_CONTEXT:-}
  if [ -z "$session_context" ] || [ ! -f "$session_context" ]; then
    echo "CP9_SESSION_CONTEXT is required for the O11Y platform" >&2
    return 1
  fi

  python3 - "$tmp" "$session_context" <<'PY'
import json
import pathlib
import re
import sys
import urllib.parse

manifest_path = pathlib.Path(sys.argv[1])
ctx = json.loads(pathlib.Path(sys.argv[2]).read_text())
terraform = ctx.get("terraform") or {}
outputs_path = pathlib.Path(str(terraform.get("outputs_path") or ""))
outputs = json.loads(outputs_path.read_text()) if outputs_path.is_file() else {}


def output(name):
    raw = outputs.get(name) or {}
    return raw.get("value") if isinstance(raw, dict) else None


target = ctx.get("target") or {}
remote = ((ctx.get("host") or {}).get("remote") or {})
gcp = (((ctx.get("host") or {}).get("regional") or {}).get("gcp") or {})
remote_eks = output("remote_eks") or {}
storage = output("o11y_platform_storage") or {}
public_host = urllib.parse.urlparse(str(remote.get("public_https_url") or "")).hostname or ""
runtime_role = str(
    (remote_eks.get("runtime_role_arn") if isinstance(remote_eks, dict) else "")
    or output("iam_role_arn")
    or ""
).strip()

values = {
    "__CP9_SESSION_NAME__": str(ctx.get("session_name") or "").strip(),
    "__CP9_TARGET_ACCOUNT_ID__": str(target.get("account") or "").strip(),
    "__CP9_TARGET_REGION__": str(target.get("region") or "").strip(),
    "__CP9_O11Y_ENDPOINT_SERVICE_ID__": str(remote.get("o11y_endpoint_service_id") or "").strip(),
    "__CP9_PUBLIC_HOST__": public_host,
    "__CP9_RUNTIME_ROLE_ARN__": runtime_role,
    "__CP9_REMOTE_RUNTIME_ROLE_ARN__": str(remote.get("runtime_role_arn") or "").strip(),
    "__CP9_O11Y_BUCKET_NAME__": str(storage.get("bucket") if isinstance(storage, dict) else "").strip(),
    "__CP9_GCP_REGIONAL_ENDPOINT_URL__": str(
        gcp.get("regional_endpoint_url") or "https://disabled.invalid"
    ).strip(),
}

text = manifest_path.read_text()
required = sorted(set(re.findall(r"__CP9_[A-Z0-9_]+__", text)))
unknown = sorted(set(required) - set(values))
if unknown:
    raise SystemExit("unknown O11Y session placeholders: " + ", ".join(unknown))
missing = [placeholder for placeholder in required if not values[placeholder]]
if missing:
    raise SystemExit("missing O11Y session values: " + ", ".join(missing))

account = values["__CP9_TARGET_ACCOUNT_ID__"]
region = values["__CP9_TARGET_REGION__"]
if "__CP9_TARGET_ACCOUNT_ID__" in required and not re.fullmatch(r"[0-9]{12}", account):
    raise SystemExit("CP9_SESSION_CONTEXT.target.account must be a 12-digit AWS account id")
if "__CP9_TARGET_REGION__" in required and not re.fullmatch(r"[a-z]{2}(?:-gov)?-[a-z]+-[0-9]", region):
    raise SystemExit("CP9_SESSION_CONTEXT.target.region must be an AWS region")
if "__CP9_O11Y_ENDPOINT_SERVICE_ID__" in required and not re.fullmatch(
    r"vpce-svc-[0-9a-f]+", values["__CP9_O11Y_ENDPOINT_SERVICE_ID__"]
):
    raise SystemExit("O11Y endpoint service id must be an AWS vpce-svc id")
if "__CP9_GCP_REGIONAL_ENDPOINT_URL__" in required:
    parsed_gcp_endpoint = urllib.parse.urlparse(values["__CP9_GCP_REGIONAL_ENDPOINT_URL__"])
    if parsed_gcp_endpoint.scheme != "https" or not parsed_gcp_endpoint.hostname or parsed_gcp_endpoint.path:
        raise SystemExit("GCP Regional endpoint must be one HTTPS origin")
for placeholder in ("__CP9_RUNTIME_ROLE_ARN__", "__CP9_REMOTE_RUNTIME_ROLE_ARN__"):
    if placeholder in required and not re.fullmatch(
        r"arn:aws(?:-[a-z0-9-]+)?:iam::[0-9]{12}:role/.+", values[placeholder]
    ):
        raise SystemExit(f"{placeholder} must be an IAM role ARN")

for placeholder in required:
    text = text.replace(placeholder, values[placeholder])
unresolved = sorted(set(re.findall(r"__CP9_[A-Z0-9_]+__", text)))
if unresolved:
    raise SystemExit("unresolved O11Y session placeholders: " + ", ".join(unresolved))
manifest_path.write_text(text)
PY

}

kubectl kustomize "$kustomize_path" >"$tmp"

for replacement in "$@"; do
  old=${replacement%%=*}
  new=${replacement#*=}
  python3 - "$tmp" "$old" "$new" <<'PY'
import pathlib
import sys

path = pathlib.Path(sys.argv[1])
old = sys.argv[2]
new = sys.argv[3]
text = path.read_text()
count = text.count(old)
if count == 0:
    print(f"expected image replacement not found: {old}", file=sys.stderr)
    sys.exit(1)
text = text.replace(old, new)
path.write_text(text)
PY
done

python3 - "$tmp" <<'PY'
import pathlib
import sys

path = pathlib.Path(sys.argv[1])
text = path.read_text().replace("imagePullPolicy: Always", "imagePullPolicy: IfNotPresent")
path.write_text(text)
PY

patch_image_rollout_stamp
default_missing_namespaces
strip_remote_human_aws_credentials
disable_dedicated_provider_conversion_webhooks
patch_dedicated_session_contract_before_apply
patch_starter_session_contract_before_apply
patch_account_contract_before_apply
patch_metering_session_contract_before_apply
patch_o11y_session_contract_before_apply
patch_image_pull_secret_before_apply
patch_session_aws_credentials_before_apply

if [ "${CP9_RENDER_ONLY:-false}" = "true" ]; then
  cat "$tmp"
  exit 0
fi

if should_stage_workloads; then
  split_manifest_phases
  if [ -s "$config_phase" ]; then
    KUBECONFIG="$kubeconfig" kubectl --context "$context" \
      apply --server-side --force-conflicts -f "$config_phase"
  fi

  # Each patch_* below self-gates on an exact "$kustomize_path" match and returns
  # 0 silently otherwise, so it only fires for the one overlay it owns:
  #   patch_dedicated_proxy_auth_secret       <- envs/dev-full/dedicated-cluster-control-server
  #   patch_starter_local_global_runtime_config <- envs/dev-full/serverless-service
  #   patch_account_secret_store_config       <- envs/dev-full/account-svc-config
  # Triggering a near-neighbour overlay (dedicated-cluster-controller,
  # dedicated-cluster-control-server-support, serverless-global-service) is not an
  # error and produces no output -- it just does nothing. Check this list before
  # concluding a patch "did not work".
  patch_dedicated_proxy_auth_secret
  patch_regional_server_root_config
  patch_dedicated_cluster_controller_runtime
  patch_starter_local_global_runtime_config
  patch_account_secret_store_config

  if [ -s "$workload_phase" ]; then
    patch_runtime_config_rollout_stamp
    KUBECONFIG="$kubeconfig" kubectl --context "$context" \
      apply --server-side --force-conflicts -f "$workload_phase"
  fi
else
  KUBECONFIG="$kubeconfig" kubectl --context "$context" \
    apply --server-side --force-conflicts -f "$tmp"
fi

print_live_yaml