#!/usr/bin/env -S uv run -q
# /// script
# requires-python = ">=3.14"
# dependencies = [
#   "pyyaml>=6.0",
# ]
# ///
"""
Weekly automated test: dispatch deploy then destroy for each config in parallel.

Before dispatching the terraform destroy workflow, destroys all app-delivery
applications in the cluster using the destroy-region-application.yaml workflow
in tidbcloud/app-delivery (step 1: app-layer services, step 2: infra/platform
services).  Each step triggers the workflow, waits for it to complete, waits
for the auto-created PR to merge, then verifies deletion in ArgoCD by querying
ArgoCD Application CRDs directly on the GKE CI cluster via gcloud + kubectl
(no ArgoCD CLI or SSO required).

Reads test-configs.yaml, dispatches every deploy workflow concurrently, waits
for each to finish, then empties any Terraform-managed AWS S3 buckets before
triggering destroy for the successful ones and waits again. Exits 0 if all
pass, 1 if any fail.

Failures are intentionally left for the on-failure job in each workflow to pick
up — this script does not create issues or call the diagnose agent.

Region existence check (auto-resume):
    Before each deploy, the script checks whether the region's ArgoCD cluster
    secret ({env}-seed-{region-name}) already exists on the CI GKE cluster.
    If it does, the deploy is skipped and the run goes straight to destroy.
    This handles the case where a previous weekly test run left a partially-
    deployed region behind (e.g. the destroy step failed mid-way).
    If the existence check itself fails, the run is aborted immediately.

Active region recovery:
    Before a new region is deployed, its generated configuration is recorded at
    weekly-test/active/{env}/{base-name}.json in the Terraform state bucket.  A
    later run that finds this record skips deployment and retries destruction of
    the recorded region.  The record is deleted only after destruction succeeds.

Usage:
    uv run scripts/run-weekly-test.py --run-id "$GITHUB_RUN_ID" [path/to/test-configs.yaml]
    uv run scripts/run-weekly-test.py --run-id <original-run-id> --destroy-only [path/to/test-configs.yaml]
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import subprocess
import sys
import tempfile
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Protocol

import yaml

REPO = "tidbcloud/auto-deploy"
APP_DELIVERY_REPO = "tidbcloud/app-delivery"
DEFAULT_REF = "main"
STATE_BUCKET = "autonomous-deployment-terraform-state"
ACTIVE_REGION_PREFIX = "weekly-test/active"
AWS_ROLE_NAME = "auto-deploy-cli"
AWS_DESTROY_ROLE_NAME = "auto-destroy-cli"
AWS_ADMIN_POLICY_ARN = "arn:aws:iam::aws:policy/AdministratorAccess"
AWS_BACKEND_ACCOUNT = "182399704004"
AWS_WORKFLOWS = {
    "deploy-aws-byoc-region.yml",
    "deploy-aws-premium-region.yml",
}
BYOC_AWS_WORKFLOWS = {
    "deploy-aws-byoc-region.yml",
}
ALICLOUD_WORKFLOWS = {
    "deploy-alicloud-premium-region.yml",
}
WORKFLOWS_REQUIRING_CONFIRM = {
    "deploy-azure-byoc-region.yml",
}
WORKFLOW_STATE_KEY_PATTERNS = {
    "deploy-aws-byoc-region.yml": "aws/byoc/{env}-{region}.json",
    "deploy-aws-premium-region.yml": "aws/premium/{env}-{region}.json",
    "deploy-alicloud-premium-region.yml": "alicloud/premium/{env}-{region}.json",
}
WORKFLOW_REGION_NAME_INPUTS = {
    "deploy-aws-byoc-region.yml": "region-name",
    "deploy-aws-premium-region.yml": "region-name",
    "deploy-alicloud-premium-region.yml": "region-name",
    "deploy-azure-premium-region.yml": "logical-region-name",
    "deploy-azure-byoc-region.yml": "logical-region-name",
}
REGION_NAME_SUFFIX_LENGTH = 6
BASE36_ALPHABET = "0123456789abcdefghijklmnopqrstuvwxyz"
AWS_EKS_IAM_ROLE_NAME_PREFIX_MAX_LENGTH = 38

# GKE clusters hosting ArgoCD, keyed by environment.
# ArgoCD Application CRDs are read directly via kubectl — no ArgoCD CLI needed.
GKE_CLUSTERS_BY_ENV: dict[str, tuple[str, str]] = {
    "dev": ("dev-cicd-cluster-us-central1", "us-central1"),
    "staging": ("staging-cicd-cluster-us-central1", "us-central1"),
    "prod": ("cicd-cluster-us-central1", "us-central1"),
}
GCP_PROJECT = "pingcap-dbaas-build"


# ── Types ─────────────────────────────────────────────────────────────────────────


@dataclass
class Config:
    workflow: str
    env: str
    logical_region_name: str
    params: dict[str, str]
    generated_region_prefix: str | None = None
    # Optional: if set, app-delivery apps are destroyed before terraform destroy.
    # Fields: cloud (aws/azure/alicloud), author (default: tidbcloud-bot),
    #         timeout (per-step seconds, default: 1800)
    # The logical region and cluster names are derived from this config.
    app_destroy: dict | None = None

    @property
    def label(self) -> str:
        return self.logical_region_name

    @property
    def state_key(self) -> str | None:
        pattern = WORKFLOW_STATE_KEY_PATTERNS.get(self.workflow)
        if not pattern:
            return None
        return pattern.format(env=self.env, region=self.logical_region_name)


@dataclass
class StepResult:
    run_id: int | None = None
    conclusion: str = "unknown"
    url: str = ""


@dataclass
class ConfigResult:
    config: Config
    deploy: StepResult = field(default_factory=StepResult)
    destroy: StepResult = field(default_factory=StepResult)

    @property
    def passed(self) -> bool:
        return (
            self.deploy.conclusion in ("success", "skipped")
            and self.destroy.conclusion == "success"
        )


@dataclass
class ActiveRegionRecord:
    slot: str
    run_id: int
    config: Config

    def to_json(self) -> str:
        return json.dumps(
            {
                "version": 1,
                "slot": self.slot,
                "run_id": self.run_id,
                "config": {
                    "workflow": self.config.workflow,
                    "env": self.config.env,
                    "logical_region_name": self.config.logical_region_name,
                    "params": self.config.params,
                    "app_destroy": self.config.app_destroy,
                },
            },
            sort_keys=True,
        )

    @classmethod
    def from_json(cls, value: str) -> ActiveRegionRecord:
        data = json.loads(value)
        if data.get("version") != 1:
            raise ValueError("unsupported active region record version")

        config = data.get("config")
        if not isinstance(config, dict):
            raise ValueError("active region record has no config")
        params = config.get("params")
        app_destroy = config.get("app_destroy")
        if not isinstance(params, dict):
            raise ValueError("active region record params must be an object")
        if app_destroy is not None and not isinstance(app_destroy, dict):
            raise ValueError("active region record app_destroy must be an object")

        return cls(
            slot=str(data["slot"]),
            run_id=int(data["run_id"]),
            config=Config(
                workflow=str(config["workflow"]),
                env=str(config["env"]),
                logical_region_name=str(config["logical_region_name"]),
                params={str(key): str(item) for key, item in params.items()},
                app_destroy=app_destroy,
            ),
        )


class ActiveRegionStore(Protocol):
    def load(self, slot: str) -> ActiveRegionRecord | None:
        raise NotImplementedError

    def state_exists(self, config: Config) -> bool:
        raise NotImplementedError

    def create(self, slot: str, record: ActiveRegionRecord) -> None:
        raise NotImplementedError

    def delete(self, slot: str) -> None:
        raise NotImplementedError


class CommandRunner(Protocol):
    def __call__(
        self,
        command: list[str],
        *,
        check: bool = True,
        env: dict[str, str] | None = None,
    ) -> subprocess.CompletedProcess[str]:
        raise NotImplementedError


class ConfigExecutor(Protocol):
    def __call__(
        self,
        config: Config,
        *,
        skip_deploy: bool,
        ref: str,
    ) -> ConfigResult:
        raise NotImplementedError


def _base36(value: int) -> str:
    if value < 0:
        raise ValueError("base36 value must not be negative")
    if value == 0:
        return "0"

    encoded = ""
    while value:
        value, remainder = divmod(value, len(BASE36_ALPHABET))
        encoded = BASE36_ALPHABET[remainder] + encoded
    return encoded


def _region_name_suffix(run_id: int) -> str:
    if run_id <= 0:
        raise ValueError("run ID must be positive")

    digest = hashlib.sha256(str(run_id).encode("ascii")).digest()
    value = int.from_bytes(digest, byteorder="big") % (
        len(BASE36_ALPHABET) ** REGION_NAME_SUFFIX_LENGTH
    )
    return _base36(value).zfill(REGION_NAME_SUFFIX_LENGTH)


def _materialize_region_name(config: Config, run_id: int) -> None:
    name_input = WORKFLOW_REGION_NAME_INPUTS.get(config.workflow)
    if name_input is None:
        raise ValueError(f"No region name input mapping for {config.workflow}")

    region_prefix = config.generated_region_prefix or config.logical_region_name
    region_name = f"{region_prefix}-{_region_name_suffix(run_id)}"
    config.logical_region_name = region_name
    config.params = {**config.params, name_input: region_name}
    if config.app_destroy is not None:
        config.app_destroy = {
            **config.app_destroy,
            "region": region_name,
            "cluster": f"{config.env}-seed-{region_name}",
        }


def _validate_derived_region_names(config: Config) -> None:
    if config.workflow not in AWS_WORKFLOWS:
        return

    iam_role_name_prefix = f"{config.env}-seed-{config.logical_region_name}-cluster-"
    if len(iam_role_name_prefix) > AWS_EKS_IAM_ROLE_NAME_PREFIX_MAX_LENGTH:
        fixed_length = (
            len(config.env)
            + len("-seed-")
            + 1  # Separator between the configured prefix and generated suffix.
            + REGION_NAME_SUFFIX_LENGTH
            + len("-cluster-")
        )
        max_generated_prefix_length = (
            AWS_EKS_IAM_ROLE_NAME_PREFIX_MAX_LENGTH - fixed_length
        )
        raise ValueError(
            f"derived EKS IAM role name prefix {iam_role_name_prefix!r} is "
            f"{len(iam_role_name_prefix)} characters; limit is "
            f"{AWS_EKS_IAM_ROLE_NAME_PREFIX_MAX_LENGTH}; generated-region-prefix "
            f"must be at most {max_generated_prefix_length} characters for "
            f"environment {config.env!r}"
        )


def run_tracked_config(
    config: Config,
    *,
    run_id: int,
    store: ActiveRegionStore,
    execute: ConfigExecutor,
    ref: str = DEFAULT_REF,
    skip_deploy: bool = False,
) -> ConfigResult:
    slot = f"{config.env}/{config.logical_region_name}"
    active_region = store.load(slot)
    if active_region is not None:
        if (
            active_region.config.workflow != config.workflow
            or active_region.config.env != config.env
        ):
            raise ValueError(
                f"active region record for {slot} does not match weekly config"
            )
        if not store.state_exists(active_region.config):
            print(
                f"[{config.label}] active region {active_region.config.label} has no "
                "terraform state; removing the stale record",
                flush=True,
            )
            store.delete(slot)
            return ConfigResult(
                config=active_region.config,
                deploy=StepResult(conclusion="skipped"),
                destroy=StepResult(conclusion="success"),
            )
        print(
            f"[{config.label}] found active region {active_region.config.label} "
            f"from run {active_region.run_id}; running cleanup only",
            flush=True,
        )
        result = execute(active_region.config, skip_deploy=True, ref=ref)
        if result.passed:
            store.delete(slot)
        return result

    _materialize_region_name(config, run_id)
    _validate_derived_region_names(config)
    store.create(
        slot,
        ActiveRegionRecord(slot=slot, run_id=run_id, config=config),
    )
    print(
        f"[{config.label}] recorded active region for slot {slot}",
        flush=True,
    )
    result = execute(config, skip_deploy=skip_deploy, ref=ref)
    if result.passed:
        store.delete(slot)
    return result


# ── GitHub helpers ────────────────────────────────────────────────────────────────


def _dispatch(workflow_file: str, inputs: dict[str, str], destroy: bool, ref: str = DEFAULT_REF, env: str = "dev") -> int:
    """Trigger a workflow_dispatch event and return the new run ID."""
    full_inputs: dict[str, str] = {
        **inputs,
        "environment": env,
        "preview": "false",
        "destroy": "true" if destroy else "false",
    }
    if destroy and workflow_file in WORKFLOWS_REQUIRING_CONFIRM:
        # deploy-azure-byoc-region.yml requires a `confirm` input equal to the
        # logical region name when destroy=true.  AWS/other workflows do not
        # define this input and will reject it with HTTP 422.
        logical_region = (
            inputs.get("logical-region-name")
            or inputs.get("region-name")
            or ""
        )
        if logical_region:
            full_inputs["confirm"] = logical_region
    before = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

    gh_args = ["workflow", "run", workflow_file, "--repo", REPO, "--ref", ref]
    for key, value in full_inputs.items():
        gh_args.extend(["-f", f"{key}={value}"])

    result = subprocess.run(["gh", *gh_args], capture_output=True, text=True)
    if result.returncode != 0:
        raise RuntimeError(
            f"Failed to trigger {workflow_file}: {result.stderr.strip()}"
        )

    # Poll until GitHub registers the new run (up to ~15 seconds).
    for _ in range(5):
        time.sleep(3)
        out = subprocess.run(
            [
                "gh",
                "run",
                "list",
                "--workflow",
                workflow_file,
                "--repo",
                REPO,
                "--limit",
                "5",
                "--json",
                "databaseId,createdAt",
            ],
            capture_output=True,
            text=True,
            check=True,
        ).stdout
        runs = json.loads(out)
        recent = [r for r in runs if r.get("createdAt", "") >= before]
        if recent:
            return int(recent[0]["databaseId"])

    raise RuntimeError(
        f"Could not identify new run for {workflow_file} after triggering — "
        "no run with createdAt >= trigger time was found."
    )


def _wait_for_run(run_id: int, repo: str = REPO) -> str:
    """Block until the run finishes. Returns the conclusion string."""
    result = subprocess.run(
        ["gh", "run", "watch", str(run_id), "--repo", repo, "--exit-status"],
        capture_output=True,
        text=True,
    )
    if result.returncode == 0:
        return "success"
    # Fetch actual conclusion from the API.
    out = subprocess.run(
        [
            "gh",
            "run",
            "view",
            str(run_id),
            "--repo",
            repo,
            "--json",
            "conclusion",
        ],
        capture_output=True,
        text=True,
    ).stdout
    try:
        return json.loads(out).get("conclusion") or "failure"
    except Exception:
        return "failure"


def _run_url(run_id: int, repo: str = REPO) -> str:
    return f"https://github.com/{repo}/actions/runs/{run_id}"


def _run(
    command: list[str], *, check: bool = True, env: dict[str, str] | None = None
) -> subprocess.CompletedProcess[str]:
    result = subprocess.run(command, capture_output=True, text=True, env=env)
    if check and result.returncode != 0:
        raise RuntimeError(
            f"Command failed ({' '.join(command)}): "
            f"{result.stderr.strip() or result.stdout.strip()}"
        )
    return result


class S3ActiveRegionStore:
    def __init__(
        self,
        *,
        bucket: str,
        aws_env: dict[str, str],
        run_command: CommandRunner = _run,
    ) -> None:
        self.bucket = bucket
        self.aws_env = aws_env
        self.run_command = run_command

    @staticmethod
    def _key(slot: str) -> str:
        return f"{ACTIVE_REGION_PREFIX}/{slot}.json"

    def load(self, slot: str) -> ActiveRegionRecord | None:
        key = self._key(slot)
        result = self.run_command(
            [
                "aws",
                "s3",
                "cp",
                f"s3://{self.bucket}/{key}",
                "-",
                "--region",
                "us-west-2",
            ],
            check=False,
            env=self.aws_env,
        )
        if result.returncode == 0:
            record = ActiveRegionRecord.from_json(result.stdout)
            if record.slot != slot:
                raise ValueError(
                    f"active region record slot {record.slot!r} does not match {slot!r}"
                )
            return record
        if "404" in result.stderr or "NoSuchKey" in result.stderr:
            return None
        raise RuntimeError(
            f"Failed to read active region record {key}: {result.stderr.strip()}"
        )

    def state_exists(self, config: Config) -> bool:
        if config.state_key is None:
            return True

        result = self.run_command(
            [
                "aws",
                "s3api",
                "head-object",
                "--bucket",
                self.bucket,
                "--key",
                config.state_key,
                "--region",
                "us-west-2",
            ],
            check=False,
            env=self.aws_env,
        )
        if result.returncode == 0:
            return True
        if "404" in result.stderr or "Not Found" in result.stderr:
            return False
        raise RuntimeError(
            f"Failed to check terraform state {config.state_key}: "
            f"{result.stderr.strip()}"
        )

    def create(self, slot: str, record: ActiveRegionRecord) -> None:
        if record.slot != slot:
            raise ValueError(
                f"active region record slot {record.slot!r} does not match {slot!r}"
            )
        key = self._key(slot)
        fd, body_path = tempfile.mkstemp(prefix="weekly-test-active-region-")
        try:
            with os.fdopen(fd, "w") as fh:
                fh.write(record.to_json())
            self.run_command(
                [
                    "aws",
                    "s3api",
                    "put-object",
                    "--bucket",
                    self.bucket,
                    "--key",
                    key,
                    "--body",
                    body_path,
                    "--if-none-match",
                    "*",
                    "--region",
                    "us-west-2",
                ],
                env=self.aws_env,
            )
        finally:
            Path(body_path).unlink(missing_ok=True)

    def delete(self, slot: str) -> None:
        key = self._key(slot)
        self.run_command(
            [
                "aws",
                "s3api",
                "delete-object",
                "--bucket",
                self.bucket,
                "--key",
                key,
                "--region",
                "us-west-2",
            ],
            env=self.aws_env,
        )


def _aws_backend_credentials() -> str:
    token_file = os.environ.get("AWS_WEB_IDENTITY_TOKEN_FILE", "").strip()
    if not token_file:
        raise RuntimeError(
            "AWS_WEB_IDENTITY_TOKEN_FILE is not set; weekly test cannot track active regions"
        )

    lines = [
        "[default]",
        f"role_arn = arn:aws:iam::{AWS_BACKEND_ACCOUNT}:role/{AWS_ROLE_NAME}",
        f"web_identity_token_file = {token_file}",
        "",
    ]
    fd, credentials_path = tempfile.mkstemp(prefix="weekly-test-backend-credentials-")
    with os.fdopen(fd, "w") as fh:
        fh.write("\n".join(lines))
    return credentials_path


def _aws_credentials_from_config(config: Config) -> str | None:
    if config.workflow not in AWS_WORKFLOWS:
        return None

    controlplane_account = config.params.get("aws-account", "").strip()
    if not controlplane_account:
        return None

    token_file = os.environ.get("AWS_WEB_IDENTITY_TOKEN_FILE", "").strip()
    if not token_file:
        raise RuntimeError(
            "AWS_WEB_IDENTITY_TOKEN_FILE is not set; weekly test cannot prepare AWS profiles for S3 cleanup"
        )

    dataplane_accounts: list[str] = []
    for key in ("dataplane-account", "o11y-global-account"):
        value = config.params.get(key, "").strip()
        if value:
            dataplane_accounts.append(value)
    for value in config.params.get("dataplane-accounts", "").split(","):
        account = value.strip()
        if account:
            dataplane_accounts.append(account)

    accounts = [controlplane_account]
    for account in dataplane_accounts:
        if account not in accounts:
            accounts.append(account)

    lines = [
        "[default]",
        f"role_arn = arn:aws:iam::{AWS_BACKEND_ACCOUNT}:role/{AWS_ROLE_NAME}",
        f"web_identity_token_file = {token_file}",
        "",
        f"[{controlplane_account}]",
        f"role_arn = arn:aws:iam::{controlplane_account}:role/{AWS_ROLE_NAME}",
        f"web_identity_token_file = {token_file}",
        "",
    ]
    for account in accounts[1:]:
        lines.extend(
            [
                f"[{account}]",
                f"role_arn = arn:aws:iam::{account}:role/{AWS_ROLE_NAME}",
                f"source_profile = {controlplane_account}",
                "",
            ]
        )

    fd, credentials_path = tempfile.mkstemp(prefix="weekly-test-aws-credentials-")
    with os.fdopen(fd, "w") as fh:
        fh.write("\n".join(lines))
    return credentials_path


def _aws_env_with_credentials(credentials_path: str) -> dict[str, str]:
    env = os.environ.copy()
    env["AWS_SHARED_CREDENTIALS_FILE"] = credentials_path
    return env


def _manage_byoc_admin_policy(config: Config, attach: bool) -> None:
    """Attach or detach AdministratorAccess on the dataplane auto-deploy-cli role.

    Uses the auto-destroy-cli role on the dataplane account, which has
    permission to attach/detach policies on auto-deploy-cli.
    """
    if config.workflow not in BYOC_AWS_WORKFLOWS:
        return

    dataplane_account = config.params.get("dataplane-account", "").strip()
    if not dataplane_account:
        return

    token_file = os.environ.get("AWS_WEB_IDENTITY_TOKEN_FILE", "").strip()
    if not token_file:
        raise RuntimeError(
            "AWS_WEB_IDENTITY_TOKEN_FILE is not set; cannot manage BYOC admin policy"
        )

    controlplane_account = config.params.get("aws-account", "").strip()
    if not controlplane_account:
        raise RuntimeError("aws-account not set in config params")

    # Build a temporary credentials file that can assume auto-destroy-cli
    # on the dataplane account via the controlplane account.
    lines = [
        "[controlplane]",
        f"role_arn = arn:aws:iam::{controlplane_account}:role/{AWS_ROLE_NAME}",
        f"web_identity_token_file = {token_file}",
        "",
        "[destroy-cli]",
        f"role_arn = arn:aws:iam::{dataplane_account}:role/{AWS_DESTROY_ROLE_NAME}",
        "source_profile = controlplane",
        "",
    ]
    fd, credentials_path = tempfile.mkstemp(prefix="weekly-test-destroy-cli-")
    with os.fdopen(fd, "w") as fh:
        fh.write("\n".join(lines))

    try:
        aws_env = _aws_env_with_credentials(credentials_path)
        action = "attach" if attach else "detach"
        action_verb = "attaching" if attach else "detaching"
        cmd = [
            "aws", "iam",
            f"{action}-role-policy",
            "--role-name", AWS_ROLE_NAME,
            "--policy-arn", AWS_ADMIN_POLICY_ARN,
            "--profile", "destroy-cli",
        ]
        print(
            f"[{config.label}] {action_verb} AdministratorAccess on "
            f"{dataplane_account}/{AWS_ROLE_NAME} ...",
            flush=True,
        )
        result = _run(cmd, env=aws_env, check=False)
        if result.returncode != 0:
            # Detach may fail if policy is already detached (NoSuchEntity);
            # attach may fail if already attached.  Both are idempotent.
            stderr = result.stderr.strip()
            if "NoSuchEntity" in stderr:
                print(
                    f"[{config.label}] policy already {action}ed; continuing",
                    flush=True,
                )
            else:
                raise RuntimeError(
                    f"Failed to {action} admin policy: {stderr}"
                )
        else:
            print(
                f"[{config.label}] AdministratorAccess {action}ed successfully",
                flush=True,
            )
    finally:
        Path(credentials_path).unlink(missing_ok=True)


def _load_terraform_state(config: Config, aws_env: dict[str, str]) -> dict:
    if not config.state_key:
        raise RuntimeError(f"No terraform state key mapping defined for workflow {config.workflow}")
    result = _run(
        [
            "aws",
            "s3",
            "cp",
            f"s3://{STATE_BUCKET}/{config.state_key}",
            "-",
            "--region",
            "us-west-2",
        ],
        env=aws_env,
    )
    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError as exc:
        raise RuntimeError(f"Failed to parse terraform state {config.state_key}: {exc}") from exc


def _s3_buckets_from_state(state: dict) -> list[dict[str, str]]:
    buckets: list[dict[str, str]] = []

    for resource in state.get("resources", []):
        if resource.get("type") != "aws_s3_bucket":
            continue
        for instance in resource.get("instances", []):
            attributes = instance.get("attributes") or {}
            identity = instance.get("identity") or {}
            bucket = attributes.get("bucket") or identity.get("bucket")
            region = attributes.get("bucket_region") or attributes.get("region") or identity.get("region")
            account_id = identity.get("account_id")
            if bucket and region and account_id:
                buckets.append(
                    {
                        "bucket": str(bucket),
                        "region": str(region),
                        "account_id": str(account_id),
                    }
                )

    deduped: list[dict[str, str]] = []
    seen: set[tuple[str, str, str]] = set()
    for bucket in buckets:
        key = (bucket["bucket"], bucket["region"], bucket["account_id"])
        if key in seen:
            continue
        seen.add(key)
        deduped.append(bucket)
    return deduped


def _vpcs_from_state(state: dict) -> list[dict[str, str]]:
    vpcs: list[dict[str, str]] = []

    for resource in state.get("resources", []):
        if resource.get("type") != "aws_vpc":
            continue
        for instance in resource.get("instances", []):
            attributes = instance.get("attributes") or {}
            identity = instance.get("identity") or {}
            vpc_id = attributes.get("id") or identity.get("id")
            region = attributes.get("region") or identity.get("region")
            account_id = identity.get("account_id")
            if vpc_id and region and account_id:
                vpcs.append(
                    {
                        "vpc_id": str(vpc_id),
                        "region": str(region),
                        "account_id": str(account_id),
                    }
                )

    deduped: list[dict[str, str]] = []
    seen: set[tuple[str, str, str]] = set()
    for vpc in vpcs:
        key = (vpc["vpc_id"], vpc["region"], vpc["account_id"])
        if key in seen:
            continue
        seen.add(key)
        deduped.append(vpc)
    return deduped


def _delete_bucket_versions(bucket: str, region: str, profile: str, aws_env: dict[str, str]) -> int:
    removed = 0
    key_marker = ""
    version_marker = ""

    while True:
        cmd = [
            "aws",
            "s3api",
            "list-object-versions",
            "--bucket",
            bucket,
            "--profile",
            profile,
            "--region",
            region,
            "--output",
            "json",
        ]
        if key_marker:
            cmd.extend(["--key-marker", key_marker])
        if version_marker:
            cmd.extend(["--version-id-marker", version_marker])

        data = json.loads(_run(cmd, env=aws_env).stdout or "{}")
        objects = [
            {"Key": item["Key"], "VersionId": item["VersionId"]}
            for item in data.get("Versions", [])
        ]
        objects.extend(
            {"Key": item["Key"], "VersionId": item["VersionId"]}
            for item in data.get("DeleteMarkers", [])
        )

        if objects:
            _run(
                [
                    "aws",
                    "s3api",
                    "delete-objects",
                    "--bucket",
                    bucket,
                    "--profile",
                    profile,
                    "--region",
                    region,
                    "--delete",
                    json.dumps({"Objects": objects, "Quiet": True}),
                ],
                env=aws_env,
            )
            removed += len(objects)

        if not data.get("IsTruncated"):
            return removed
        key_marker = data.get("NextKeyMarker", "")
        version_marker = data.get("NextVersionIdMarker", "")


def _assert_bucket_empty(bucket: str, region: str, profile: str, aws_env: dict[str, str]) -> None:
    objects = json.loads(
        _run(
            [
                "aws",
                "s3api",
                "list-objects-v2",
                "--bucket",
                bucket,
                "--profile",
                profile,
                "--region",
                region,
                "--max-items",
                "1",
                "--output",
                "json",
            ],
            env=aws_env,
        ).stdout
        or "{}"
    )
    if objects.get("Contents"):
        raise RuntimeError(f"Bucket {bucket} still has live objects after cleanup")

    versions = json.loads(
        _run(
            [
                "aws",
                "s3api",
                "list-object-versions",
                "--bucket",
                bucket,
                "--profile",
                profile,
                "--region",
                region,
                "--max-items",
                "1",
                "--output",
                "json",
            ],
            env=aws_env,
        ).stdout
        or "{}"
    )
    if versions.get("Versions") or versions.get("DeleteMarkers"):
        raise RuntimeError(f"Bucket {bucket} is still not empty after cleanup")


def _delete_nlb_endpoint_service_dependencies(
    nlb_arn: str, region: str, profile: str, aws_env: dict[str, str]
) -> None:
    """
    Find any VPC Endpoint Service configurations backed by the given NLB, delete
    all VPC Endpoints connected to each service, then delete the service itself.
    This must be done before deleting the NLB or AWS will return ResourceInUse.
    """
    svc_data = json.loads(
        _run(
            [
                "aws",
                "ec2",
                "describe-vpc-endpoint-service-configurations",
                "--profile",
                profile,
                "--region",
                region,
                "--output",
                "json",
            ],
            env=aws_env,
        ).stdout
        or "{}"
    )

    services = [
        svc
        for svc in svc_data.get("ServiceConfigurations", [])
        if nlb_arn in (svc.get("NetworkLoadBalancerArns") or [])
    ]
    if not services:
        return

    for svc in services:
        svc_id = svc["ServiceId"]
        svc_name = svc.get("ServiceName", svc_id)
        print(f"found VPC Endpoint Service {svc_id} ({svc_name}) referencing NLB; cleaning up ...", flush=True)

        # Find and delete all VPC Endpoints connected to this service.
        ep_data = json.loads(
            _run(
                [
                    "aws",
                    "ec2",
                    "describe-vpc-endpoints",
                    "--profile",
                    profile,
                    "--region",
                    region,
                    "--filters",
                    f"Name=service-name,Values={svc_name}",
                    "Name=vpc-endpoint-state,Values=available,pending,pending-acceptance",
                    "--output",
                    "json",
                ],
                env=aws_env,
            ).stdout
            or "{}"
        )
        endpoint_ids = [ep["VpcEndpointId"] for ep in ep_data.get("VpcEndpoints", [])]
        if endpoint_ids:
            print(f"deleting {len(endpoint_ids)} VPC Endpoint(s): {endpoint_ids} ...", flush=True)
            _run(
                [
                    "aws",
                    "ec2",
                    "delete-vpc-endpoints",
                    "--profile",
                    profile,
                    "--region",
                    region,
                    "--vpc-endpoint-ids",
                    *endpoint_ids,
                    "--output",
                    "json",
                ],
                env=aws_env,
            )
            # Wait for all endpoints to reach a terminal state before deleting the service.
            # AWS removes the endpoint record entirely once deleted, so
            # InvalidVpcEndpointId.NotFound is treated as fully deleted.
            deadline = time.time() + 300
            while time.time() < deadline:
                time.sleep(10)
                result = _run(
                    [
                        "aws",
                        "ec2",
                        "describe-vpc-endpoints",
                        "--profile",
                        profile,
                        "--region",
                        region,
                        "--vpc-endpoint-ids",
                        *endpoint_ids,
                        "--query",
                        "VpcEndpoints[?State!='deleted'].[VpcEndpointId,State]",
                        "--output",
                        "json",
                    ],
                    env=aws_env,
                    check=False,
                )
                if result.returncode != 0:
                    if "InvalidVpcEndpointId.NotFound" in result.stderr:
                        break  # endpoint no longer exists — fully deleted
                    raise RuntimeError(
                        f"Command failed (aws ec2 describe-vpc-endpoints): {result.stderr.strip()}"
                    )
                poll = json.loads(result.stdout or "[]")
                if not poll:
                    break
                print(f"waiting for VPC Endpoints to be deleted: {poll} ...", flush=True)
            else:
                raise RuntimeError(
                    f"VPC Endpoints {endpoint_ids} for service {svc_id} did not reach deleted state within 300s"
                )

        print(f"deleting VPC Endpoint Service {svc_id} ...", flush=True)
        _run(
            [
                "aws",
                "ec2",
                "delete-vpc-endpoint-service-configurations",
                "--profile",
                profile,
                "--region",
                region,
                "--service-ids",
                svc_id,
                "--output",
                "json",
            ],
            env=aws_env,
        )


def _delete_vpc_load_balancers(vpc_id: str, region: str, profile: str, aws_env: dict[str, str]) -> None:
    data = json.loads(
        _run(
            [
                "aws",
                "elbv2",
                "describe-load-balancers",
                "--profile",
                profile,
                "--region",
                region,
                "--output",
                "json",
            ],
            env=aws_env,
        ).stdout
        or "{}"
    )

    load_balancers = [
        lb
        for lb in data.get("LoadBalancers", [])
        if lb.get("VpcId") == vpc_id
    ]
    if not load_balancers:
        return

    print(
        f"deleting {len(load_balancers)} load balancer(s) in {vpc_id} before terraform destroy",
        flush=True,
    )
    for load_balancer in load_balancers:
        arn = load_balancer["LoadBalancerArn"]
        name = load_balancer.get("LoadBalancerName", arn)
        _delete_nlb_endpoint_service_dependencies(arn, region, profile, aws_env)
        print(f"disabling deletion protection for {name} ...", flush=True)
        _run(
            [
                "aws",
                "elbv2",
                "modify-load-balancer-attributes",
                "--load-balancer-arn",
                arn,
                "--profile",
                profile,
                "--region",
                region,
                "--attributes",
                "Key=deletion_protection.enabled,Value=false",
            ],
            env=aws_env,
        )
        print(f"deleting load balancer {name} ...", flush=True)
        _run(
            [
                "aws",
                "elbv2",
                "delete-load-balancer",
                "--load-balancer-arn",
                arn,
                "--profile",
                profile,
                "--region",
                region,
            ],
            env=aws_env,
        )


def _empty_s3_bucket(bucket: str, region: str, profile: str, aws_env: dict[str, str]) -> None:
    print(
        f"emptying s3://{bucket} in {region} using profile {profile} ...",
        flush=True,
    )
    head = _run(
        ["aws", "s3api", "head-bucket", "--bucket", bucket, "--profile", profile, "--region", region],
        check=False,
        env=aws_env,
    )
    if head.returncode != 0:
        print(f"s3://{bucket} does not exist or is not accessible; skipping", flush=True)
        return
    _run(
        [
            "aws",
            "s3api",
            "put-bucket-policy",
            "--bucket",
            bucket,
            "--profile",
            profile,
            "--region",
            region,
            "--policy",
            json.dumps(
                {
                    "Version": "2012-10-17",
                    "Statement": [
                        {
                            "Sid": "DenyBucketWritesDuringWeeklyCleanup",
                            "Effect": "Deny",
                            "Principal": "*",
                            "Action": [
                                "s3:AbortMultipartUpload",
                                "s3:PutObject",
                                "s3:RestoreObject",
                            ],
                            "Resource": [
                                f"arn:aws:s3:::{bucket}",
                                f"arn:aws:s3:::{bucket}/*",
                            ],
                        }
                    ],
                }
            ),
        ],
        env=aws_env,
    )
    _run(
        [
            "aws",
            "s3",
            "rm",
            f"s3://{bucket}",
            "--recursive",
            "--profile",
            profile,
            "--region",
            region,
        ],
        env=aws_env,
    )
    removed_versions = _delete_bucket_versions(bucket, region, profile, aws_env)
    if removed_versions:
        print(
            f"removed {removed_versions} versioned object(s)/delete marker(s) from s3://{bucket}",
            flush=True,
        )
    _assert_bucket_empty(bucket, region, profile, aws_env)


def _cleanup_terraform_s3_buckets(config: Config, label: str) -> None:
    credentials_path = _aws_credentials_from_config(config)
    if not credentials_path:
        return

    try:
        aws_env = _aws_env_with_credentials(credentials_path)
        state = _load_terraform_state(config, aws_env)
        vpcs = _vpcs_from_state(state)
        for vpc in vpcs:
            _delete_vpc_load_balancers(
                vpc_id=vpc["vpc_id"],
                region=vpc["region"],
                profile=vpc["account_id"],
                aws_env=aws_env,
            )
        buckets = _s3_buckets_from_state(state)
        if not buckets:
            print(f"[{label}] no S3 buckets found in terraform state; skipping cleanup", flush=True)
            return
        print(f"[{label}] emptying {len(buckets)} S3 bucket(s) before terraform destroy", flush=True)
        for bucket in buckets:
            _empty_s3_bucket(
                bucket=bucket["bucket"],
                region=bucket["region"],
                profile=bucket["account_id"],
                aws_env=aws_env,
            )
    finally:
        Path(credentials_path).unlink(missing_ok=True)


# ── AliCloud OSS cleanup helpers ─────────────────────────────────────────────────


def _alicloud_credentials_from_config(config: Config) -> str | None:
    """Write an aliyun CLI config file to a temp path. Returns the path or None."""
    if config.workflow not in ALICLOUD_WORKFLOWS:
        return None

    controlplane_account = config.params.get("alicloud-account", "").strip()
    if not controlplane_account:
        return None

    token_file = os.environ.get("AWS_WEB_IDENTITY_TOKEN_FILE", "").strip()
    if not token_file:
        raise RuntimeError(
            "AWS_WEB_IDENTITY_TOKEN_FILE is not set; weekly test cannot prepare AliCloud profiles for OSS cleanup"
        )

    alicloud_region = config.params.get("alicloud-region", "ap-southeast-1").strip()

    dataplane_accounts: list[str] = []
    for value in config.params.get("dataplane-accounts", "").split(","):
        account = value.strip()
        if account:
            dataplane_accounts.append(account)

    profiles: list[dict] = [
        {
            "name": "oidc",
            "mode": "OIDC",
            "sts_region": "ap-southeast-1",
            "ram_role_arn": f"acs:ram::{controlplane_account}:role/auto-deploy-cli",
            "ram_session_name": f"{alicloud_region}-auto-deploy",
            "expired_seconds": 3600,
            "region_id": alicloud_region,
            "output_format": "json",
            "language": "en",
            "oidc_provider_arn": f"acs:ram::{controlplane_account}:oidc-provider/cicd-gcp-oidc-provider",
            "oidc_token_file": token_file,
        },
        {
            "name": controlplane_account,
            "mode": "ChainableRamRoleArn",
            "sts_region": "ap-southeast-1",
            "ram_role_arn": f"acs:ram::{controlplane_account}:role/auto-deploy-cli",
            "ram_session_name": f"{alicloud_region}-auto-deploy",
            "source_profile": "oidc",
            "expired_seconds": 3600,
            "region_id": alicloud_region,
            "output_format": "json",
            "language": "en",
        },
    ]
    for account in dataplane_accounts:
        profiles.append(
            {
                "name": account,
                "mode": "ChainableRamRoleArn",
                "sts_region": "ap-southeast-1",
                "ram_role_arn": f"acs:ram::{account}:role/auto-deploy-cli",
                "ram_session_name": f"{alicloud_region}-auto-deploy",
                "source_profile": controlplane_account,
                "expired_seconds": 3600,
                "region_id": alicloud_region,
                "output_format": "json",
                "language": "en",
            }
        )

    config_data = {"current": controlplane_account, "profiles": profiles}
    fd, config_path = tempfile.mkstemp(prefix="weekly-test-aliyun-config-", suffix=".json")
    with os.fdopen(fd, "w") as fh:
        json.dump(config_data, fh, indent=2)
    return config_path


def _oss_buckets_from_state(state: dict) -> list[dict[str, str]]:
    buckets: list[dict[str, str]] = []

    for resource in state.get("resources", []):
        if resource.get("type") != "alicloud_oss_bucket":
            continue
        for instance in resource.get("instances", []):
            attributes = instance.get("attributes") or {}
            identity = instance.get("identity") or {}
            bucket = attributes.get("bucket") or attributes.get("id")
            # location is "oss-<region>", e.g. "oss-ap-southeast-3"
            location = attributes.get("location", "")
            region = location.removeprefix("oss-") if location.startswith("oss-") else location
            account_id = identity.get("account_id") or identity.get("owner_id") or ""
            if bucket:
                buckets.append(
                    {
                        "bucket": str(bucket),
                        "region": region,
                        "account_id": account_id,
                    }
                )

    deduped: list[dict[str, str]] = []
    seen: set[str] = set()
    for b in buckets:
        if b["bucket"] not in seen:
            seen.add(b["bucket"])
            deduped.append(b)
    return deduped


def _empty_oss_bucket(bucket: str, region: str, profile: str, aliyun_config_path: str) -> None:
    print(
        f"emptying oss://{bucket} in {region} using profile {profile} ...",
        flush=True,
    )
    # Remove all current-version objects recursively.
    _run(
        [
            "aliyun", "--profile", profile, "--config-path", aliyun_config_path,
            "oss", "rm", f"oss://{bucket}/", "-r", "-f",
        ]
    )
    # Remove versioned objects and delete markers if versioning is enabled.
    result = _run(
        [
            "aliyun", "--profile", profile, "--config-path", aliyun_config_path,
            "oss", "ls", f"oss://{bucket}/", "--all-versions", "-r",
        ],
        check=False,
    )
    if result.returncode != 0:
        # --all-versions not supported or bucket already empty; treat as done.
        return
    versioned_objects = [
        line.split()[-1]
        for line in result.stdout.splitlines()
        if line.strip() and not line.strip().startswith("LastModifiedTime")
    ]
    if versioned_objects:
        print(
            f"removing {len(versioned_objects)} versioned object(s)/delete marker(s) from oss://{bucket}",
            flush=True,
        )
        _run(
            [
                "aliyun", "--profile", profile, "--config-path", aliyun_config_path,
                "oss", "rm", f"oss://{bucket}/", "-r", "-f", "--all-versions",
            ]
        )


def _cleanup_terraform_oss_buckets(config: Config, label: str) -> None:
    aliyun_config_path = _alicloud_credentials_from_config(config)
    if not aliyun_config_path:
        return

    # Terraform state is stored in S3; use a minimal AWS backend credentials file.
    token_file = os.environ.get("AWS_WEB_IDENTITY_TOKEN_FILE", "").strip()
    if not token_file:
        raise RuntimeError(
            "AWS_WEB_IDENTITY_TOKEN_FILE is not set; weekly test cannot load terraform state for OSS cleanup"
        )
    aws_lines = [
        "[default]",
        f"role_arn = arn:aws:iam::{AWS_BACKEND_ACCOUNT}:role/{AWS_ROLE_NAME}",
        f"web_identity_token_file = {token_file}",
        "",
    ]
    fd, aws_credentials_path = tempfile.mkstemp(prefix="weekly-test-aws-backend-credentials-")
    with os.fdopen(fd, "w") as fh:
        fh.write("\n".join(aws_lines))

    try:
        aws_env = _aws_env_with_credentials(aws_credentials_path)
        state = _load_terraform_state(config, aws_env)
        buckets = _oss_buckets_from_state(state)
        if not buckets:
            print(f"[{label}] no OSS buckets found in terraform state; skipping cleanup", flush=True)
            return
        print(f"[{label}] emptying {len(buckets)} OSS bucket(s) before terraform destroy", flush=True)
        controlplane_account = config.params.get("alicloud-account", "").strip()
        for bucket in buckets:
            profile = bucket["account_id"] or controlplane_account
            _empty_oss_bucket(
                bucket=bucket["bucket"],
                region=bucket["region"],
                profile=profile,
                aliyun_config_path=aliyun_config_path,
            )
    finally:
        Path(aliyun_config_path).unlink(missing_ok=True)
        Path(aws_credentials_path).unlink(missing_ok=True)


# ── App-delivery destroy helpers ──────────────────────────────────────────────────


def _fetch_destroy_script() -> str:
    """Fetch destory-region.sh from tidbcloud/app-delivery."""
    result = subprocess.run(
        [
            "gh",
            "api",
            f"repos/{APP_DELIVERY_REPO}/contents/.github/scripts/destory-region.sh",
            "--jq",
            ".content",
        ],
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        raise RuntimeError(
            f"Failed to fetch destory-region.sh: {result.stderr.strip()}"
        )
    # .content is base64-encoded; decode it
    import base64

    return base64.b64decode(result.stdout.strip()).decode()


def _parse_apps(
    script_content: str, cloud: str, env: str, region: str, cluster: str
) -> list[dict]:
    """Extract and evaluate the YAML apps list embedded in the bash heredoc."""
    env_path_prefix = "" if env == "prod" else f"{env}/"

    match = re.search(r"apps=\$\(cat <<EOF\n(.*?)\nEOF\n\)", script_content, re.DOTALL)
    if not match:
        raise RuntimeError("Could not find apps heredoc in destory-region.sh")

    yaml_str = match.group(1)
    replacements = {
        "${env_path_prefix}": env_path_prefix,
        "${logicalRegionName}": region,
        "${cloud}": cloud,
        "${env}": env,
        "${cluster}": cluster,
    }
    for var, val in replacements.items():
        yaml_str = yaml_str.replace(var, val)

    data = yaml.safe_load(yaml_str)
    return data["apps"]


# ── AWS cleanup helpers ───────────────────────────────────────────────────────────


def _filter_apps_by_step(apps: list[dict], cloud: str, step: int) -> list[dict]:
    """Return apps that match the given step and cloud."""
    result = []
    for app in apps:
        app_cloud = app.get("cloud")
        app_step = app.get("step")
        if app_cloud and app_cloud != cloud:
            continue
        if app_step is None or str(app_step) != str(step):
            continue
        result.append(app)
    return result


def _trigger_app_destroy_workflow(
    cloud: str, env: str, region: str, cluster: str, step: int, author: str
) -> int:
    """Trigger destroy-region-application.yaml and return the run ID directly."""
    result = subprocess.run(
        [
            "gh",
            "api",
            f"repos/{APP_DELIVERY_REPO}/actions/workflows/destroy-region-application.yaml/dispatches",
            "--method",
            "POST",
            "-H",
            "Accept: application/vnd.github+json",
            "-f",
            "ref=main",
            "-F",
            "return_run_details=true",
            "-f",
            f"inputs[cloud]={cloud}",
            "-f",
            f"inputs[env]={env}",
            "-f",
            f"inputs[logicalRegionName]={region}",
            "-f",
            f"inputs[cluster]={cluster}",
            "-f",
            f"inputs[step]={step}",
            "-f",
            f"inputs[author]={author}",
        ],
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        raise RuntimeError(
            f"Failed to trigger app destroy workflow (step {step}): {result.stderr.strip()}"
        )
    data = json.loads(result.stdout)
    return int(data["workflow_run_id"])


def _get_pr_from_run_logs(run_id: int) -> int | None:
    """Extract the PR number from the workflow run logs."""
    result = subprocess.run(
        ["gh", "run", "view", str(run_id), "--repo", APP_DELIVERY_REPO, "--log"],
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        return None
    for line in result.stdout.splitlines():
        m = re.search(r"Pull Request Number - (\d+)", line)
        if m:
            return int(m.group(1))
    return None


def _wait_for_pr_merged(pr_number: int, label: str, timeout: int = 1800) -> None:
    """Block until the PR is merged."""
    start = time.time()
    while time.time() - start < timeout:
        result = subprocess.run(
            [
                "gh",
                "pr",
                "view",
                str(pr_number),
                "--repo",
                APP_DELIVERY_REPO,
                "--json",
                "state,mergedAt",
            ],
            capture_output=True,
            text=True,
        )
        if result.returncode != 0:
            raise RuntimeError(
                f"[{label}] Failed to check PR #{pr_number}: {result.stderr.strip()}"
            )
        data = json.loads(result.stdout)
        state = data.get("state", "")
        if state == "MERGED" or data.get("mergedAt"):
            print(f"[{label}] PR #{pr_number} merged", flush=True)
            return
        if state == "CLOSED":
            raise RuntimeError(f"[{label}] PR #{pr_number} closed without merging")
        print(f"[{label}] PR #{pr_number} state={state}; waiting 30s...", flush=True)
        time.sleep(30)
    raise TimeoutError(f"[{label}] PR #{pr_number} not merged within {timeout}s")


def _gke_get_credentials(env: str) -> None:
    """
    Authenticate kubectl against the CI GKE cluster for the given environment.
    Raises RuntimeError if gcloud fails.
    """
    gke_cluster, gke_region = GKE_CLUSTERS_BY_ENV[env]
    result = subprocess.run(
        [
            "gcloud",
            "container",
            "clusters",
            "get-credentials",
            gke_cluster,
            "--region",
            gke_region,
            "--project",
            GCP_PROJECT,
        ],
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        raise RuntimeError(
            f"gcloud get-credentials failed for {gke_cluster}: {result.stderr.strip()}"
        )


def _region_exists(env: str, region_name: str) -> bool:
    """
    Return True if the region's ArgoCD cluster secret already exists on the CI
    GKE cluster, indicating a previous deploy left the region partially or fully
    deployed.  The secret name follows the convention used by check-region-conflict:
    {env}-seed-{region_name} in the argocd namespace.

    Raises RuntimeError if gcloud or kubectl cannot be reached.
    """
    _gke_get_credentials(env)
    secret_name = f"{env}-seed-{region_name}"
    result = subprocess.run(
        ["kubectl", "get", "secret", secret_name, "-n", "argocd"],
        capture_output=True,
        text=True,
    )
    if result.returncode == 0:
        return True
    if "NotFound" in result.stderr:
        return False
    raise RuntimeError(
        f"kubectl get secret {secret_name} failed: {result.stderr.strip()}"
    )


def _gke_list_argocd_apps(env: str, cluster: str) -> list[str]:
    """
    Return ArgoCD app names for the given destination cluster by querying
    Application CRDs directly on the GKE CI cluster via gcloud + kubectl.
    Raises RuntimeError if gcloud or kubectl fails.
    """
    _gke_get_credentials(env)

    result = subprocess.run(
        ["kubectl", "get", "applications.argoproj.io", "-n", "argocd", "-o", "json"],
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        raise RuntimeError(f"kubectl get applications failed: {result.stderr.strip()}")

    data = json.loads(result.stdout)
    return [
        item["metadata"]["name"]
        for item in data.get("items", [])
        if item.get("spec", {}).get("destination", {}).get("name") == cluster
    ]


def _wait_for_apps_deleted(
    env: str,
    cluster: str,
    step_apps: list[dict],
    label: str,
    timeout: int = 1800,
) -> None:
    """Poll ArgoCD (via GKE kubectl) until all step apps are gone from the cluster."""
    app_names = [a["name"] for a in step_apps]

    print(
        f"[{label}] waiting for {len(app_names)} app(s) to disappear from ArgoCD "
        f"on cluster {cluster} ...",
        flush=True,
    )
    start = time.time()
    while time.time() - start < timeout:
        apps = _gke_list_argocd_apps(env, cluster)
        still_present = [a for a in apps if any(name in a for name in app_names)]
        if not still_present:
            print(f"[{label}] all step apps removed from ArgoCD", flush=True)
            return
        print(
            f"[{label}] still present in ArgoCD: {still_present}; waiting 30s...",
            flush=True,
        )
        time.sleep(30)
    raise TimeoutError(
        f"[{label}] apps not deleted from ArgoCD within {timeout}s: {still_present}"
    )


def _destroy_region_apps(app_destroy: dict, env: str, label: str) -> None:
    """
    Destroy all app-delivery apps for the region in two ordered steps:
      step 1 — app-layer services
      step 2 — infra/platform services

    Each step: trigger destroy-region-application.yaml, wait for workflow +
    PR merge, then verify deletion in ArgoCD (or sleep if argocd unavailable).
    """
    cloud = app_destroy["cloud"]
    region = app_destroy["region"]
    cluster = app_destroy["cluster"]
    author = app_destroy.get("author", "tidbcloud-bot")
    timeout = int(app_destroy.get("timeout", 1800))

    print(f"[{label}] fetching app list from app-delivery ...", flush=True)
    script_content = _fetch_destroy_script()
    all_apps = _parse_apps(script_content, cloud, env, region, cluster)
    print(f"[{label}] parsed {len(all_apps)} apps from destory-region.sh", flush=True)

    for step in (1, 2):
        step_apps = _filter_apps_by_step(all_apps, cloud, step)
        if not step_apps:
            print(f"[{label}] step {step}: no apps — skipping", flush=True)
            continue

        names = [a["name"] for a in step_apps]
        print(
            f"[{label}] step {step}: destroying {len(step_apps)} app(s): {names}",
            flush=True,
        )

        run_id = _trigger_app_destroy_workflow(
            cloud, env, region, cluster, step, author
        )
        url = _run_url(run_id, APP_DELIVERY_REPO)
        print(f"[{label}] step {step}: workflow started — {url}", flush=True)

        conclusion = _wait_for_run(run_id, APP_DELIVERY_REPO)
        if conclusion != "success":
            raise RuntimeError(
                f"[{label}] step {step}: workflow {run_id} ended with {conclusion!r}"
            )
        print(f"[{label}] step {step}: workflow complete", flush=True)

        pr_number = _get_pr_from_run_logs(run_id)
        if pr_number:
            print(
                f"[{label}] step {step}: waiting for PR #{pr_number} to merge ...",
                flush=True,
            )
            _wait_for_pr_merged(pr_number, label=label, timeout=timeout)
        else:
            print(
                f"[{label}] step {step}: no PR found in logs (no changes?); skipping PR wait",
                flush=True,
            )

        _wait_for_apps_deleted(env, cluster, step_apps, label=label, timeout=timeout)
        print(f"[{label}] step {step}: complete", flush=True)

    print(f"[{label}] all app-delivery apps destroyed", flush=True)


# ── Per-config test loop ──────────────────────────────────────────────────────────


def run_one_config(config: Config, *, skip_deploy: bool = False, ref: str = DEFAULT_REF) -> ConfigResult:
    result = ConfigResult(config=config)

    # Auto-resume: if the region's ArgoCD cluster secret already exists, a
    # previous run left it deployed.  Skip deploy and go straight to destroy.
    if not skip_deploy:
        print(f"[{config.label}] checking whether region already exists ...", flush=True)
        try:
            if _region_exists(config.env, config.label):
                print(
                    f"[{config.label}] region already exists (ArgoCD secret found) "
                    f"— skipping deploy, proceeding directly to destroy",
                    flush=True,
                )
                skip_deploy = True
            else:
                print(f"[{config.label}] region does not exist — will deploy", flush=True)
        except Exception as exc:
            result.deploy.conclusion = "check_failed"
            print(f"[{config.label}] region-exists check failed: {exc}", flush=True)
            return result

    # For BYOC AWS: ensure auto-deploy-cli does NOT have admin policy during deploy,
    # so we test with the same permissions as production.
    if not skip_deploy:
        try:
            _manage_byoc_admin_policy(config, attach=False)
        except Exception as exc:
            result.deploy.conclusion = "policy_detach_failed"
            print(f"[{config.label}] pre-deploy admin policy detach failed: {exc}", flush=True)
            return result

    if skip_deploy:
        result.deploy.conclusion = "skipped"
    else:
        # Deploy.
        print(f"[{config.label}] dispatching deploy ...", flush=True)
        try:
            run_id = _dispatch(config.workflow, config.params, destroy=False, ref=ref, env=config.env)
        except RuntimeError as exc:
            result.deploy.conclusion = "dispatch_failed"
            print(f"[{config.label}] deploy dispatch error: {exc}", flush=True)
            return result

        result.deploy.run_id = run_id
        result.deploy.url = _run_url(run_id)
        print(f"[{config.label}] deploy started: {result.deploy.url}", flush=True)

        result.deploy.conclusion = _wait_for_run(run_id)
        print(
            f"[{config.label}] deploy {result.deploy.conclusion}: {result.deploy.url}",
            flush=True,
        )

        if result.deploy.conclusion != "success":
            # on-failure job in the workflow will handle diagnosis automatically.
            return result

    # For BYOC AWS: attach admin policy before destroy so auto-deploy-cli
    # has permission to delete dataplane resources.
    try:
        _manage_byoc_admin_policy(config, attach=True)
    except Exception as exc:
        result.destroy.conclusion = "policy_attach_failed"
        print(f"[{config.label}] pre-destroy admin policy attach failed: {exc}", flush=True)
        return result

    # Destroy app-delivery apps before terraform destroy.
    if config.app_destroy:
        try:
            _destroy_region_apps(config.app_destroy, env=config.env, label=config.label)
        except Exception as exc:
            result.destroy.conclusion = "app_destroy_failed"
            print(f"[{config.label}] app-delivery destroy failed: {exc}", flush=True)
            return result

    if config.workflow in AWS_WORKFLOWS:
        try:
            _cleanup_terraform_s3_buckets(config, label=config.label)
        except Exception as exc:
            result.destroy.conclusion = "bucket_cleanup_failed"
            print(f"[{config.label}] bucket cleanup failed: {exc}", flush=True)
            return result
    elif config.workflow in ALICLOUD_WORKFLOWS:
        try:
            _cleanup_terraform_oss_buckets(config, label=config.label)
        except Exception as exc:
            result.destroy.conclusion = "bucket_cleanup_failed"
            print(f"[{config.label}] OSS bucket cleanup failed: {exc}", flush=True)
            return result

    # Terraform destroy.
    print(f"[{config.label}] dispatching terraform destroy ...", flush=True)
    try:
        run_id = _dispatch(config.workflow, config.params, destroy=True, ref=ref, env=config.env)
    except RuntimeError as exc:
        result.destroy.conclusion = "dispatch_failed"
        print(f"[{config.label}] destroy dispatch error: {exc}", flush=True)
        return result

    result.destroy.run_id = run_id
    result.destroy.url = _run_url(run_id)
    print(f"[{config.label}] destroy started: {result.destroy.url}", flush=True)

    result.destroy.conclusion = _wait_for_run(run_id)
    print(
        f"[{config.label}] destroy {result.destroy.conclusion}: {result.destroy.url}",
        flush=True,
    )

    # For BYOC AWS: detach admin policy after destroy to leave the role clean.
    try:
        _manage_byoc_admin_policy(config, attach=False)
    except Exception as exc:
        print(f"[{config.label}] post-destroy admin policy detach failed (non-fatal): {exc}", flush=True)

    return result


# ── Entry point ───────────────────────────────────────────────────────────────────


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Weekly automated test: deploy then destroy each config in parallel."
    )
    parser.add_argument(
        "config_path",
        nargs="?",
        help="Path to test-configs.yaml (default: autopilot/test-configs.yaml)",
    )
    parser.add_argument(
        "--destroy-only",
        action="store_true",
        help="Skip deploy and go straight to destroy (for cleaning up a partially-deployed region).",
    )
    parser.add_argument(
        "--ref",
        default=DEFAULT_REF,
        help="Git ref (branch/tag/SHA) to dispatch workflows from (default: main).",
    )
    parser.add_argument(
        "--only",
        default=None,
        metavar="LABEL",
        help="Run only the config whose label matches this value (e.g. us-east-1-wkly).",
    )
    parser.add_argument(
        "--run-id",
        required=True,
        type=int,
        help="Parent weekly workflow GITHUB_RUN_ID used to generate unique logical region names.",
    )
    args = parser.parse_args()

    config_path = (
        Path(args.config_path)
        if args.config_path
        else Path(__file__).parent.parent / "test-configs.yaml"
    )
    if not config_path.exists():
        print(f"Config file not found: {config_path}", file=sys.stderr)
        sys.exit(1)

    raw = yaml.safe_load(config_path.read_text())
    configs = [
        Config(
            workflow=c["workflow"],
            env=c["env"],
            logical_region_name=c["logical-region-name"],
            params=c.get("params", {}),
            generated_region_prefix=c.get("generated-region-prefix"),
            app_destroy=c.get("app_destroy"),
        )
        for c in raw.get("configs", [])
    ]

    if not configs:
        print("No configs found.", file=sys.stderr)
        sys.exit(1)

    if args.only:
        configs = [c for c in configs if c.label == args.only]
        if not configs:
            print(f"No config found with label {args.only!r}.", file=sys.stderr)
            sys.exit(1)
        print(f"--only {args.only!r}: running 1 config.\n", flush=True)

    for config in configs:
        if config.workflow not in WORKFLOW_REGION_NAME_INPUTS:
            parser.error(f"No region name input mapping for {config.workflow}")

    if args.destroy_only:
        print(
            f"--destroy-only: skipping deploy, running destroy for {len(configs)} config(s) ...\n",
            flush=True,
        )
    else:
        print(f"Running {len(configs)} test config(s) in parallel ...\n", flush=True)

    backend_credentials_path = _aws_backend_credentials()
    try:
        store = S3ActiveRegionStore(
            bucket=STATE_BUCKET,
            aws_env=_aws_env_with_credentials(backend_credentials_path),
        )
        results: list[ConfigResult] = []
        with ThreadPoolExecutor(max_workers=len(configs)) as pool:
            futures = {
                pool.submit(
                    run_tracked_config,
                    config,
                    run_id=args.run_id,
                    store=store,
                    execute=run_one_config,
                    ref=args.ref,
                    skip_deploy=args.destroy_only,
                ): config
                for config in configs
            }
            for future in as_completed(futures):
                results.append(future.result())
    finally:
        Path(backend_credentials_path).unlink(missing_ok=True)

    # Print summary table.
    print("\n" + "=" * 70)
    print(f"{'CONFIG':<35} {'DEPLOY':<12} {'DESTROY':<12}")
    print("=" * 70)
    any_failed = False
    for r in sorted(results, key=lambda x: x.config.label):
        deploy_str = r.deploy.conclusion
        deploy_ran_ok = r.deploy.conclusion in ("success", "skipped")
        destroy_str = r.destroy.conclusion if deploy_ran_ok else "-"
        print(f"{r.config.label:<35} {deploy_str:<12} {destroy_str:<12}")
        if not r.passed:
            any_failed = True
    print("=" * 70)

    if any_failed:
        print("\nSome configs failed. Diagnosis issues will be created automatically.")
        sys.exit(1)
    else:
        print("\nAll configs passed.")


if __name__ == "__main__":
    main()