#!/usr/bin/env bash
set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
DEFAULT_PROJECT_DIR="$(cd "${SCRIPT_DIR}/../../.." && pwd)"

ACTION=""
TARGET_HOST=""
PROJECT_DIR="${DEFAULT_PROJECT_DIR}"
SSH_USER="root"
SSH_PORT="22"
SSH_KEY="${HOME}/.ssh/id_rsa"

CLUSTER_NAME="next-gen-cluster"
TIUP_VERSION="v8.5.1"
KEYSPACE_COUNT="2"
TIDB_BIN=""
PD_ENDPOINT=""
CDC_SERVER=""
KEYSPACE_TIDB_BASE_PORT=""
KEYSPACE_STATUS_BASE_PORT=""
PATCH_TIKV=""
PATCH_TIDB=""
PATCH_PD=""
PATCH_CDC=""
SKIP_MINIO="0"

CREATE_CHANGEFEED="0"
SOURCE_KEYSPACE="1"
CHANGEFEED_ID="test-sink"
SINK_URI=""
SINK_DEPLOYED_KEYSPACE="0"
SINK_HOST=""
SINK_USER="root"
SINK_PASSWORD=""

SSH_KEY_SET="0"
TIDB_BIN_SET="0"
PD_ENDPOINT_SET="0"
CDC_SERVER_SET="0"
KEYSPACE_TIDB_BASE_PORT_SET="0"
KEYSPACE_STATUS_BASE_PORT_SET="0"
PATCH_TIKV_SET="0"
PATCH_TIDB_SET="0"
PATCH_PD_SET="0"
PATCH_CDC_SET="0"
REMOTE_EXEC="0"

GENERATED_TOPO=""
CDC_CLI_BIN=""
TIDB_PATCH_BIN=""

RUNTIME_ROOT="/tmp/next-gen-with-cdc"
GENERATED_ROOT="${RUNTIME_ROOT}/skill-generated"
MINIO_RUNTIME_ROOT="/tmp/minio/next-gen-cdc"

log() {
    printf '[ticdc-next-gen] %s\n' "$*"
}

die() {
    printf '[ticdc-next-gen] ERROR: %s\n' "$*" >&2
    exit 1
}

usage() {
    cat <<'USAGE'
Usage:
  ticdc_next_gen.sh deploy [options]
  ticdc_next_gen.sh destroy [options]

Common options:
  --host <ip-or-hostname>              Deployment host (required)
  --project-dir <path>                 Directory containing topo.yaml/start-minio.sh (default: repo root)
  --ssh-user <user>                    SSH user for target host (default: root)
  --ssh-port <port>                    SSH port (default: 22)
  --ssh-key <path>                     SSH private key path (default: $HOME/.ssh/id_rsa)

Deploy options:
  --cluster-name <name>                TiUP cluster name (default: next-gen-cluster)
  --tiup-version <version>             TiDB version passed to tiup deploy (default: v8.5.1)
  --keyspace-count <n>                 Number of keyspace TiDB servers to launch (default: 2)
  --tidb-bin <path>                    tidb-server binary path (default: extracted from --patch-tidb)
  --pd-endpoint <host:port>            PD endpoint for keyspace TiDB; default from generated topo
  --cdc-server <host:port>             CDC API server; default from generated topo
  --keyspace-tidb-base-port <port>     Base port for keyspace TiDB; default from topo tidb_servers.port
  --keyspace-status-base-port <port>   Base status port for keyspace TiDB; default from topo tidb_servers.status_port
  --patch-tikv <path>                  tikv tar.gz package path
  --patch-tidb <path>                  tidb tar.gz package path
  --patch-pd <path>                    pd tar.gz package path
  --patch-cdc <path>                   cdc tar.gz package path
  --skip-minio                         Skip start-minio.sh

Changefeed options (deploy only):
  --create-changefeed                  Create changefeed after deployment
  --source-keyspace <n>                Source keyspace index (default: 1 => keyspace1)
  --changefeed-id <id>                 Changefeed id (default: test-sink)
  --sink-uri <uri>                     Explicit sink URI (e.g. mysql://root@127.0.0.1:14002)
  --sink-deployed-keyspace <n>         Auto resolve sink-uri to mysql://<user>@<host>:<port>
  --sink-host <ip-or-hostname>         Host used for auto sink-uri (default: 127.0.0.1)
  --sink-user <user>                   User used for auto sink-uri (default: root)
  --sink-password <password>           Password used for auto sink-uri

Examples:
  # Deploy on target host via SSH
  bash skills/ticdc-next-gen-deploy/scripts/ticdc_next_gen.sh deploy \
    --host 10.2.7.6 --project-dir /root/project --keyspace-count 2

  # Deploy + create changefeed to keyspace2
  bash skills/ticdc-next-gen-deploy/scripts/ticdc_next_gen.sh deploy \
    --host 10.2.7.6 --project-dir /root/project \
    --keyspace-count 2 --create-changefeed --source-keyspace 1 \
    --sink-deployed-keyspace 2 --changefeed-id test-sink

  # Destroy deployment on target host
  bash skills/ticdc-next-gen-deploy/scripts/ticdc_next_gen.sh destroy \
    --host 10.2.7.6 --project-dir /root/project

Notes:
  All actions run on the target host via SSH.
  Missing tiup/minio/mc will be auto-installed from official sources on the execution host.
  Auto-install requires Linux + curl or wget + outbound internet access.
  Patch package paths can be passed via --patch-*; if omitted, script prompts for each path and verifies file existence.
  Changefeed CLI uses cdc binary extracted from --patch-cdc tar.gz.
USAGE
}

need_arg() {
    local option="$1"
    local value="${2-}"
    if [[ -z "${value}" ]]; then
        die "${option} requires a value"
    fi
}

is_positive_int() {
    [[ "$1" =~ ^[1-9][0-9]*$ ]]
}

is_non_negative_int() {
    [[ "$1" =~ ^[0-9]+$ ]]
}

ensure_command() {
    command -v "$1" >/dev/null 2>&1 || die "Missing command: $1"
}

ensure_file() {
    [[ -f "$1" ]] || die "File not found: $1"
}

has_command() {
    command -v "$1" >/dev/null 2>&1
}

prepend_path_once() {
    local dir="$1"
    case ":$PATH:" in
        *":${dir}:"*) ;;
        *) export PATH="${dir}:$PATH" ;;
    esac
}

detect_linux_arch() {
    local os
    local arch
    os="$(uname -s)"
    arch="$(uname -m)"

    [[ "$os" == "Linux" ]] || die "Automatic dependency installation only supports Linux hosts."

    case "$arch" in
        x86_64 | amd64) printf 'amd64\n' ;;
        aarch64 | arm64) printf 'arm64\n' ;;
        armv7l) printf 'arm\n' ;;
        ppc64le) printf 'ppc64le\n' ;;
        s390x) printf 's390x\n' ;;
        *) die "Unsupported Linux architecture for MinIO download: ${arch}" ;;
    esac
}

download_to_file() {
    local url="$1"
    local output="$2"

    if has_command curl; then
        curl -fsSL "$url" -o "$output" >/dev/null 2>&1
        return $?
    fi
    if has_command wget; then
        wget -qO "$output" "$url" >/dev/null 2>&1
        return $?
    fi

    die "Neither curl nor wget is available. Install one of them before running this script."
}

select_install_bin_dir() {
    if [[ -d "/usr/local/bin" && -w "/usr/local/bin" ]]; then
        printf '/usr/local/bin\n'
        return
    fi

    local user_bin="${HOME}/.local/bin"
    mkdir -p "$user_bin"
    prepend_path_once "$user_bin"
    printf '%s\n' "$user_bin"
}

install_binary_from_urls() {
    local cmd_name="$1"
    shift

    local bin_dir
    local target
    local tmp
    local url
    local downloaded="0"

    bin_dir="$(select_install_bin_dir)"
    target="${bin_dir}/${cmd_name}"
    tmp="$(mktemp)"

    for url in "$@"; do
        if download_to_file "$url" "$tmp"; then
            downloaded="1"
            break
        fi
    done

    if [[ "$downloaded" != "1" ]]; then
        rm -f "$tmp"
        die "Failed to download ${cmd_name} binary from official sources."
    fi

    cp "$tmp" "$target"
    chmod +x "$target"
    rm -f "$tmp"

    prepend_path_once "$bin_dir"
    has_command "$cmd_name" || die "${cmd_name} installation failed: command not found after install."
}

install_tiup_if_missing() {
    if has_command tiup; then
        return
    fi

    log "tiup is missing, installing from official installer"

    local installer
    installer="$(mktemp)"
    download_to_file "https://tiup-mirrors.pingcap.com/install.sh" "$installer" || {
        rm -f "$installer"
        die "Failed to download TiUP installer."
    }

    sh "$installer"
    rm -f "$installer"

    prepend_path_once "${HOME}/.tiup/bin"
    has_command tiup || die "TiUP installation completed but tiup command is still unavailable."
}

install_minio_if_missing() {
    if has_command minio; then
        return
    fi

    local arch
    arch="$(detect_linux_arch)"
    log "minio is missing, installing binary for linux-${arch}"
    install_binary_from_urls "minio" \
        "https://dl.min.io/server/minio/release/linux-${arch}/minio" \
        "https://dl.minio.io/server/minio/release/linux-${arch}/minio"
}

install_mc_if_missing() {
    if has_command mc; then
        return
    fi

    local arch
    arch="$(detect_linux_arch)"
    log "mc is missing, installing binary for linux-${arch}"
    install_binary_from_urls "mc" \
        "https://dl.min.io/client/mc/release/linux-${arch}/mc" \
        "https://dl.minio.io/client/mc/release/linux-${arch}/mc"
}

ensure_runtime_dependencies() {
    ensure_command sed
    ensure_command awk
    ensure_command tar
    ensure_command find

    install_tiup_if_missing

    if [[ "$SKIP_MINIO" == "0" ]]; then
        install_minio_if_missing
        install_mc_if_missing
    fi
}

build_keyspace_prealloc_list() {
    local values=()
    local i
    for ((i = 1; i <= KEYSPACE_COUNT; i++)); do
        values+=("\"keyspace${i}\"")
    done
    local IFS=", "
    printf '%s' "${values[*]}"
}

ssh_target() {
    printf '%s@%s' "$SSH_USER" "$TARGET_HOST"
}

print_ssh_copy_id_hint() {
    local pub_key="${SSH_KEY}.pub"
    if [[ -f "$pub_key" ]]; then
        printf '[ticdc-next-gen] Hint: run ssh-copy-id first:\n' >&2
        printf '  ssh-copy-id -i %q -p %q %q\n' "$pub_key" "$SSH_PORT" "$(ssh_target)" >&2
    else
        printf '[ticdc-next-gen] Hint: generate %s and run ssh-copy-id before deploy.\n' "$pub_key" >&2
    fi
}

ensure_ssh_auth() {
    ensure_command ssh
    if ! ssh -o BatchMode=yes -o ConnectTimeout=8 -p "$SSH_PORT" -i "$SSH_KEY" "$(ssh_target)" "true" >/dev/null 2>&1; then
        print_ssh_copy_id_hint
        die "SSH key authentication failed for $(ssh_target)"
    fi
}

yaml_first_value_in_section() {
    local file="$1"
    local section="$2"
    local key="$3"

    awk -v section="$section" -v key="$key" '
        $1 == section ":" {
            in_section = 1
            next
        }
        in_section && $1 == "-" && $2 == key ":" {
            print $3
            exit
        }
        in_section && $1 == key ":" {
            print $2
            exit
        }
        in_section && $0 ~ /^[A-Za-z_][A-Za-z0-9_]*:/ {
            in_section = 0
        }
    ' "$file"
}

yaml_section_exists() {
    local file="$1"
    local section="$2"

    awk -v section="$section" '
        $1 == section ":" {
            found = 1
            exit
        }
        END {
            if (found) {
                exit 0
            }
            exit 1
        }
    ' "$file"
}

ensure_observability_sections() {
    local topo_file="$1"
    local has_monitoring="0"
    local has_grafana="0"

    if yaml_section_exists "$topo_file" "monitoring_servers"; then
        has_monitoring="1"
    fi
    if yaml_section_exists "$topo_file" "grafana_servers"; then
        has_grafana="1"
    fi

    if [[ "$has_monitoring" == "1" ]]; then
        local monitoring_host
        local monitoring_port
        monitoring_host="$(yaml_first_value_in_section "$topo_file" "monitoring_servers" "host")"
        monitoring_port="$(yaml_first_value_in_section "$topo_file" "monitoring_servers" "port")"
        [[ -n "$monitoring_host" && -n "$monitoring_port" ]] || die "monitoring_servers exists but no valid host/port entry was found."
    else
        cat >>"$topo_file" <<EOF

monitoring_servers:
  - host: ${TARGET_HOST}
    port: 15000
EOF
        log "monitoring_servers missing in topo, appended default ${TARGET_HOST}:15000"
    fi

    if [[ "$has_grafana" == "1" ]]; then
        local grafana_host
        local grafana_port
        grafana_host="$(yaml_first_value_in_section "$topo_file" "grafana_servers" "host")"
        grafana_port="$(yaml_first_value_in_section "$topo_file" "grafana_servers" "port")"
        [[ -n "$grafana_host" && -n "$grafana_port" ]] || die "grafana_servers exists but no valid host/port entry was found."
    else
        cat >>"$topo_file" <<EOF

grafana_servers:
  - host: ${TARGET_HOST}
    port: 33000
EOF
        log "grafana_servers missing in topo, appended default ${TARGET_HOST}:33000"
    fi
}

tidb_patch_extract_dir() {
    printf '%s/tidb-bin/%s\n' "$GENERATED_ROOT" "$CLUSTER_NAME"
}

cdc_patch_extract_dir() {
    printf '%s/cdc-bin/%s\n' "$GENERATED_ROOT" "$CLUSTER_NAME"
}

default_generated_topo_path() {
    printf '%s/topo-%s.yaml\n' "$GENERATED_ROOT" "$CLUSTER_NAME"
}

print_troubleshooting_paths() {
    local topo_path="$GENERATED_TOPO"
    local keyspace_config_dir="${GENERATED_ROOT}/keyspace-config"
    local tidb_extract_dir
    local cdc_extract_dir

    if [[ -z "$topo_path" ]]; then
        topo_path="$(default_generated_topo_path)"
    fi

    tidb_extract_dir="$(tidb_patch_extract_dir)"
    cdc_extract_dir="$(cdc_patch_extract_dir)"

    log "Troubleshooting paths:"
    log "  runtime_root: ${RUNTIME_ROOT}"
    log "  generated_root: ${GENERATED_ROOT}"
    log "  generated_topo: ${topo_path}"
    log "  keyspace_config_dir: ${keyspace_config_dir}"
    log "  keyspace_log_files: ${RUNTIME_ROOT}/tidb-ks*.log"
    log "  keyspace_stdout_logs: ${RUNTIME_ROOT}/tidb-ks*.stdout.log"
    log "  tidb_patch_extract_dir: ${tidb_extract_dir}"
    log "  cdc_patch_extract_dir: ${cdc_extract_dir}"
    if [[ "$SKIP_MINIO" == "0" ]]; then
        log "  minio_runtime_dir: ${MINIO_RUNTIME_ROOT}"
    else
        log "  minio_runtime_dir: ${MINIO_RUNTIME_ROOT} (minio skipped in this run)"
    fi
}

prepare_generated_topo() {
    local runtime_dir="$GENERATED_ROOT"
    mkdir -p "$runtime_dir"

    local source_topo="${PROJECT_DIR}/topo.yaml"
    local generated_topo="${runtime_dir}/topo-${CLUSTER_NAME}.yaml"
    local deploy_dir="/DATA/disk/tidb/deploy"
    local data_dir="/DATA/disk/tidb/data"
    ensure_file "$source_topo"
    cp "$source_topo" "$generated_topo"

    awk -v host="$TARGET_HOST" '
        {
            if ($1 == "host:") {
                indent = substr($0, 1, index($0, "host:") - 1)
                print indent "host: " host
                next
            }
            print
        }
    ' "$generated_topo" >"${generated_topo}.tmp"
    mv "${generated_topo}.tmp" "$generated_topo"

    awk -v deploy_dir="$deploy_dir" -v data_dir="$data_dir" '
        {
            if ($1 == "deploy_dir:") {
                indent = substr($0, 1, index($0, "deploy_dir:") - 1)
                print indent "deploy_dir: \"" deploy_dir "\""
                next
            }
            if ($1 == "data_dir:") {
                indent = substr($0, 1, index($0, "data_dir:") - 1)
                print indent "data_dir: \"" data_dir "\""
                next
            }
            print
        }
    ' "$generated_topo" >"${generated_topo}.tmp"
    mv "${generated_topo}.tmp" "$generated_topo"

    ensure_observability_sections "$generated_topo"

    local prealloc_line
    prealloc_line="    keyspace.pre-alloc: [$(build_keyspace_prealloc_list)]"
    awk -v replacement="$prealloc_line" '
        /^[[:space:]]*keyspace\.pre-alloc:/ {
            print replacement
            next
        }
        { print }
    ' "$generated_topo" >"${generated_topo}.tmp"
    mv "${generated_topo}.tmp" "$generated_topo"

    GENERATED_TOPO="$generated_topo"
}

resolve_patch_path() {
    local component="$1"
    local current_path="$2"
    local option_name="$3"
    local value="$current_path"

    while true; do
        if [[ -n "$value" ]]; then
            if [[ -f "$value" ]]; then
                printf '%s\n' "$value"
                return
            fi
            log "File not found for ${component} patch: ${value}"
        fi

        if [[ ! -t 0 ]]; then
            die "Missing ${component} patch package path. Please pass ${option_name} <path> and ensure the file exists."
        fi

        printf '[ticdc-next-gen] Enter %s patch package path (%s): ' "$component" "$option_name" >&2
        if ! IFS= read -r value; then
            die "Failed to read input for ${component} patch package path."
        fi
    done
}

ensure_packages() {
    PATCH_TIKV="$(resolve_patch_path "tikv" "$PATCH_TIKV" "--patch-tikv")"
    PATCH_TIDB="$(resolve_patch_path "tidb" "$PATCH_TIDB" "--patch-tidb")"
    PATCH_PD="$(resolve_patch_path "pd" "$PATCH_PD" "--patch-pd")"
    PATCH_CDC="$(resolve_patch_path "cdc" "$PATCH_CDC" "--patch-cdc")"
}

derive_runtime_settings_from_topo() {
    local topo_file="$1"

    if [[ "$PD_ENDPOINT_SET" == "0" ]]; then
        local pd_host
        local pd_port
        pd_host="$(yaml_first_value_in_section "$topo_file" "pd_servers" "host")"
        pd_port="$(yaml_first_value_in_section "$topo_file" "pd_servers" "client_port")"
        [[ -n "$pd_host" && -n "$pd_port" ]] || die "Unable to derive PD endpoint from ${topo_file}; use --pd-endpoint"
        PD_ENDPOINT="${pd_host}:${pd_port}"
    fi

    if [[ "$KEYSPACE_TIDB_BASE_PORT_SET" == "0" ]]; then
        KEYSPACE_TIDB_BASE_PORT="$(yaml_first_value_in_section "$topo_file" "tidb_servers" "port")"
        [[ -n "$KEYSPACE_TIDB_BASE_PORT" ]] || die "Unable to derive tidb base port from ${topo_file}; use --keyspace-tidb-base-port"
    fi

    if [[ "$KEYSPACE_STATUS_BASE_PORT_SET" == "0" ]]; then
        KEYSPACE_STATUS_BASE_PORT="$(yaml_first_value_in_section "$topo_file" "tidb_servers" "status_port")"
        [[ -n "$KEYSPACE_STATUS_BASE_PORT" ]] || die "Unable to derive tidb status base port from ${topo_file}; use --keyspace-status-base-port"
    fi

    if [[ "$CREATE_CHANGEFEED" == "1" && "$CDC_SERVER_SET" == "0" ]]; then
        local cdc_host
        local cdc_port
        cdc_host="$(yaml_first_value_in_section "$topo_file" "cdc_servers" "host")"
        cdc_port="$(yaml_first_value_in_section "$topo_file" "cdc_servers" "port")"
        [[ -n "$cdc_host" && -n "$cdc_port" ]] || die "Unable to derive CDC server from ${topo_file}; use --cdc-server"
        CDC_SERVER="${cdc_host}:${cdc_port}"
    fi
}

extract_tidb_binary_from_patch() {
    if [[ -n "$TIDB_PATCH_BIN" && -x "$TIDB_PATCH_BIN" ]]; then
        printf '%s\n' "$TIDB_PATCH_BIN"
        return 0
    fi

    [[ -n "$PATCH_TIDB" ]] || return 1
    [[ -f "$PATCH_TIDB" ]] || return 1

    local runtime_dir
    runtime_dir="$(tidb_patch_extract_dir)"
    rm -rf "$runtime_dir"
    mkdir -p "$runtime_dir"

    tar -xzf "$PATCH_TIDB" -C "$runtime_dir"

    local candidate=""
    candidate="$(find "$runtime_dir" -type f -name tidb-server -perm -u+x 2>/dev/null | head -n1 || true)"
    if [[ -z "$candidate" ]]; then
        candidate="$(find "$runtime_dir" -type f -name tidb-server 2>/dev/null | head -n1 || true)"
    fi
    [[ -n "$candidate" ]] || return 1

    chmod +x "$candidate" >/dev/null 2>&1 || true
    [[ -x "$candidate" ]] || return 1

    TIDB_PATCH_BIN="$candidate"
    printf '%s\n' "$TIDB_PATCH_BIN"
}

resolve_tidb_bin() {
    if [[ "$TIDB_BIN_SET" == "1" ]]; then
        ensure_file "$TIDB_BIN"
        [[ -x "$TIDB_BIN" ]] || die "--tidb-bin is not executable: ${TIDB_BIN}"
        log "Using user-provided tidb binary: ${TIDB_BIN}"
        return
    fi

    local candidate=""
    candidate="$(extract_tidb_binary_from_patch || true)"
    if [[ -n "$candidate" ]]; then
        TIDB_BIN="$candidate"
        log "Using tidb binary extracted from --patch-tidb: ${TIDB_BIN}"
        return
    fi
    die "Cannot extract tidb-server from --patch-tidb (${PATCH_TIDB}). Please provide a valid patch package or pass --tidb-bin explicitly."
}

wait_for_port() {
    local host="$1"
    local port="$2"
    local timeout_seconds="$3"
    local i

    for ((i = 1; i <= timeout_seconds; i++)); do
        if bash -c ":</dev/tcp/${host}/${port}" >/dev/null 2>&1; then
            return 0
        fi
        sleep 1
    done
    return 1
}

extract_cdc_binary_from_patch() {
    if [[ -n "$CDC_CLI_BIN" && -x "$CDC_CLI_BIN" ]]; then
        return
    fi

    local runtime_dir
    runtime_dir="$(cdc_patch_extract_dir)"
    rm -rf "$runtime_dir"
    mkdir -p "$runtime_dir"

    tar -xzf "$PATCH_CDC" -C "$runtime_dir"

    local candidate=""
    candidate="$(find "$runtime_dir" -type f -name cdc -perm -u+x 2>/dev/null | head -n1 || true)"
    if [[ -z "$candidate" ]]; then
        candidate="$(find "$runtime_dir" -type f -name cdc 2>/dev/null | head -n1 || true)"
    fi
    if [[ -z "$candidate" ]]; then
        candidate="$(find "$runtime_dir" -type f -name '*cdc*' -perm -u+x 2>/dev/null | head -n1 || true)"
    fi

    [[ -n "$candidate" ]] || die "Cannot find cdc binary after extracting ${PATCH_CDC}."
    chmod +x "$candidate" >/dev/null 2>&1 || true
    [[ -x "$candidate" ]] || die "Extracted cdc binary is not executable: ${candidate}"

    CDC_CLI_BIN="$candidate"
    log "Using CDC CLI binary from patch: ${CDC_CLI_BIN}"
}

run_cdc_cli() {
    extract_cdc_binary_from_patch
    "$CDC_CLI_BIN" cli "$@"
}

deploy_cluster() {
    local topo_file="$1"

    log "Deploying cluster ${CLUSTER_NAME} with ${topo_file}"
    tiup cluster deploy "$CLUSTER_NAME" "$TIUP_VERSION" "$topo_file" \
        --user "$SSH_USER" \
        -i "$SSH_KEY" \
        --ignore-config-check \
        --yes

    log "Patching cluster binaries"
    tiup cluster patch "$CLUSTER_NAME" "$PATCH_TIKV" -R tikv --offline --yes
    tiup cluster patch "$CLUSTER_NAME" "$PATCH_TIDB" -R tidb --offline --yes
    tiup cluster patch "$CLUSTER_NAME" "$PATCH_PD" -R pd --offline --yes
    tiup cluster patch "$CLUSTER_NAME" "$PATCH_CDC" -R cdc --offline --yes

    log "Starting cluster ${CLUSTER_NAME}"
    tiup cluster start "$CLUSTER_NAME"
    tiup cluster display "$CLUSTER_NAME"
}

ensure_keyspace_config() {
    local index="$1"
    local runtime_dir="${GENERATED_ROOT}/keyspace-config"
    mkdir -p "$runtime_dir"

    local template="${PROJECT_DIR}/tidb-keyspace${index}.toml"
    if [[ ! -f "$template" ]]; then
        template="${PROJECT_DIR}/tidb-keyspace1.toml"
    fi
    ensure_file "$template"

    local output="${runtime_dir}/tidb-keyspace${index}.toml"
    sed -E "s/^keyspace-name = \".*\"/keyspace-name = \"keyspace${index}\"/" "$template" >"$output"
    printf '%s\n' "$output"
}

start_keyspace_tidbs() {
    resolve_tidb_bin

    local runtime_dir="$RUNTIME_ROOT"
    mkdir -p "$runtime_dir"

    local i
    for ((i = 1; i <= KEYSPACE_COUNT; i++)); do
        local config_file
        config_file="$(ensure_keyspace_config "$i")"

        local port=$((KEYSPACE_TIDB_BASE_PORT + i))
        local status_port=$((KEYSPACE_STATUS_BASE_PORT + i))
        local stdout_file="${runtime_dir}/tidb-ks${i}.stdout.log"
        local log_file="${runtime_dir}/tidb-ks${i}.log"

        log "Starting keyspace${i} TiDB at ${TARGET_HOST}:${port}"
        if has_command setsid; then
            setsid "$TIDB_BIN" \
                --config "$config_file" \
                -P "$port" \
                --store=tikv \
                --path="$PD_ENDPOINT" \
                -status "$status_port" \
                -log-file "$log_file" >"$stdout_file" 2>&1 </dev/null &
        else
            nohup "$TIDB_BIN" \
                --config "$config_file" \
                -P "$port" \
                --store=tikv \
                --path="$PD_ENDPOINT" \
                -status "$status_port" \
                -log-file "$log_file" >"$stdout_file" 2>&1 </dev/null &
        fi

        local pid=$!
        disown "$pid" 2>/dev/null || true
        sleep 1
        if ! kill -0 "$pid" >/dev/null 2>&1; then
            die "keyspace${i} TiDB failed to start. Check ${stdout_file}"
        fi

        if ! wait_for_port "127.0.0.1" "$port" 180; then
            die "keyspace${i} TiDB did not listen on 127.0.0.1:${port} in time. Check ${log_file} and ${stdout_file}"
        fi
    done
}

resolve_sink_uri() {
    if [[ -n "$SINK_URI" ]]; then
        printf '%s\n' "$SINK_URI"
        return
    fi

    if [[ "$SINK_DEPLOYED_KEYSPACE" == "0" ]]; then
        die "Provide --sink-uri or --sink-deployed-keyspace when --create-changefeed is set"
    fi

    if ((SINK_DEPLOYED_KEYSPACE > KEYSPACE_COUNT)); then
        die "--sink-deployed-keyspace (${SINK_DEPLOYED_KEYSPACE}) is greater than --keyspace-count (${KEYSPACE_COUNT})"
    fi

    local sink_host="$SINK_HOST"
    if [[ -z "$sink_host" ]]; then
        sink_host="127.0.0.1"
    fi

    local sink_port=$((KEYSPACE_TIDB_BASE_PORT + SINK_DEPLOYED_KEYSPACE))
    local auth
    if [[ -n "$SINK_PASSWORD" ]]; then
        auth="${SINK_USER}:${SINK_PASSWORD}@"
    else
        auth="${SINK_USER}@"
    fi

    printf 'mysql://%s%s:%s\n' "$auth" "$sink_host" "$sink_port"
}

create_changefeed() {
    if ((SOURCE_KEYSPACE > KEYSPACE_COUNT)); then
        die "--source-keyspace (${SOURCE_KEYSPACE}) is greater than --keyspace-count (${KEYSPACE_COUNT})"
    fi

    [[ -n "$CDC_SERVER" ]] || die "CDC server is empty; use --cdc-server"

    local sink_uri
    sink_uri="$(resolve_sink_uri)"

    log "Creating changefeed ${CHANGEFEED_ID}: keyspace${SOURCE_KEYSPACE} -> ${sink_uri}"
    run_cdc_cli changefeed --server="$CDC_SERVER" \
        -k "keyspace${SOURCE_KEYSPACE}" \
        create \
        --sink-uri="$sink_uri" \
        -c "$CHANGEFEED_ID"
}

remove_path() {
    local path="$1"
    if [[ $(id -u) -eq 0 ]]; then
        rm -rf "$path" || true
    elif has_command sudo; then
        sudo rm -rf "$path" || true
    else
        log "Skipping cleanup for ${path} (need root/sudo)"
    fi
}

destroy_local() {
    install_tiup_if_missing

    log "Destroying cluster ${CLUSTER_NAME}"
    tiup cluster destroy "$CLUSTER_NAME" --yes || true

    pkill -9 -x minio >/dev/null 2>&1 || true
    pkill -9 -x tidb-server >/dev/null 2>&1 || true
    pkill -9 -x local-kms >/dev/null 2>&1 || true

    remove_path "$RUNTIME_ROOT"
    remove_path "$MINIO_RUNTIME_ROOT"

    log "Destroy finished"
}

deploy_local() {
    ensure_runtime_dependencies

    [[ -d "$PROJECT_DIR" ]] || die "Directory not found: ${PROJECT_DIR}"
    ensure_file "${PROJECT_DIR}/topo.yaml"
    ensure_file "${PROJECT_DIR}/start-minio.sh"

    ensure_packages

    if [[ "$SKIP_MINIO" == "0" ]]; then
        log "Starting minio"
        (cd "$PROJECT_DIR" && bash ./start-minio.sh)
    fi

    prepare_generated_topo
    derive_runtime_settings_from_topo "$GENERATED_TOPO"

    deploy_cluster "$GENERATED_TOPO"
    start_keyspace_tidbs

    if [[ "$CREATE_CHANGEFEED" == "1" ]]; then
        create_changefeed
    fi

    log "Deployment finished"
    print_troubleshooting_paths
}

validate_inputs() {
    [[ "$ACTION" == "deploy" || "$ACTION" == "destroy" ]] || die "Action must be deploy or destroy"
    [[ -n "$TARGET_HOST" ]] || die "--host is required"
    is_positive_int "$SSH_PORT" || die "--ssh-port must be a positive integer"
    [[ "$REMOTE_EXEC" == "0" || "$REMOTE_EXEC" == "1" ]] || die "internal flag --_remote-exec is invalid"

    if [[ "$ACTION" == "deploy" ]]; then
        is_positive_int "$KEYSPACE_COUNT" || die "--keyspace-count must be a positive integer"

        if [[ "$PD_ENDPOINT_SET" == "1" ]]; then
            [[ "$PD_ENDPOINT" == *:* ]] || die "--pd-endpoint must be host:port"
        fi

        if [[ "$CDC_SERVER_SET" == "1" ]]; then
            [[ "$CDC_SERVER" == *:* ]] || die "--cdc-server must be host:port"
        fi

        if [[ "$KEYSPACE_TIDB_BASE_PORT_SET" == "1" ]]; then
            is_positive_int "$KEYSPACE_TIDB_BASE_PORT" || die "--keyspace-tidb-base-port must be a positive integer"
        fi

        if [[ "$KEYSPACE_STATUS_BASE_PORT_SET" == "1" ]]; then
            is_positive_int "$KEYSPACE_STATUS_BASE_PORT" || die "--keyspace-status-base-port must be a positive integer"
        fi

        if [[ "$CREATE_CHANGEFEED" == "1" ]]; then
            is_positive_int "$SOURCE_KEYSPACE" || die "--source-keyspace must be a positive integer"
        fi

        is_non_negative_int "$SINK_DEPLOYED_KEYSPACE" || die "--sink-deployed-keyspace must be >= 0"
    fi
}

build_remote_args() {
    local args=("$ACTION" "--_remote-exec")

    args+=(--host "$TARGET_HOST")
    args+=(--project-dir "$PROJECT_DIR")
    args+=(--ssh-user "$SSH_USER")
    args+=(--ssh-port "$SSH_PORT")
    if [[ "$SSH_KEY_SET" == "1" ]]; then
        args+=(--ssh-key "$SSH_KEY")
    fi

    if [[ "$ACTION" == "deploy" ]]; then
        args+=(--cluster-name "$CLUSTER_NAME")
        args+=(--tiup-version "$TIUP_VERSION")
        args+=(--keyspace-count "$KEYSPACE_COUNT")

        if [[ "$TIDB_BIN_SET" == "1" ]]; then
            args+=(--tidb-bin "$TIDB_BIN")
        fi
        if [[ "$PD_ENDPOINT_SET" == "1" ]]; then
            args+=(--pd-endpoint "$PD_ENDPOINT")
        fi
        if [[ "$CDC_SERVER_SET" == "1" ]]; then
            args+=(--cdc-server "$CDC_SERVER")
        fi
        if [[ "$KEYSPACE_TIDB_BASE_PORT_SET" == "1" ]]; then
            args+=(--keyspace-tidb-base-port "$KEYSPACE_TIDB_BASE_PORT")
        fi
        if [[ "$KEYSPACE_STATUS_BASE_PORT_SET" == "1" ]]; then
            args+=(--keyspace-status-base-port "$KEYSPACE_STATUS_BASE_PORT")
        fi
        if [[ "$PATCH_TIKV_SET" == "1" ]]; then
            args+=(--patch-tikv "$PATCH_TIKV")
        fi
        if [[ "$PATCH_TIDB_SET" == "1" ]]; then
            args+=(--patch-tidb "$PATCH_TIDB")
        fi
        if [[ "$PATCH_PD_SET" == "1" ]]; then
            args+=(--patch-pd "$PATCH_PD")
        fi
        if [[ "$PATCH_CDC_SET" == "1" ]]; then
            args+=(--patch-cdc "$PATCH_CDC")
        fi
        if [[ "$SKIP_MINIO" == "1" ]]; then
            args+=(--skip-minio)
        fi

        if [[ "$CREATE_CHANGEFEED" == "1" ]]; then
            args+=(--create-changefeed)
            args+=(--source-keyspace "$SOURCE_KEYSPACE")
            args+=(--changefeed-id "$CHANGEFEED_ID")

            if [[ -n "$SINK_URI" ]]; then
                args+=(--sink-uri "$SINK_URI")
            fi
            if [[ "$SINK_DEPLOYED_KEYSPACE" != "0" ]]; then
                args+=(--sink-deployed-keyspace "$SINK_DEPLOYED_KEYSPACE")
            fi
            if [[ -n "$SINK_HOST" ]]; then
                args+=(--sink-host "$SINK_HOST")
            fi
            args+=(--sink-user "$SINK_USER")
            if [[ -n "$SINK_PASSWORD" ]]; then
                args+=(--sink-password "$SINK_PASSWORD")
            fi
        fi
    fi

    if [[ "$ACTION" == "destroy" ]]; then
        args+=(--cluster-name "$CLUSTER_NAME")
    fi

    printf '%q ' "${args[@]}"
}

run_remote_action() {
    ensure_ssh_auth

    local ssh_opts=(-o BatchMode=yes -o ConnectTimeout=8 -p "$SSH_PORT" -i "$SSH_KEY")
    local target
    target="$(ssh_target)"

    local remote_script="/tmp/ticdc-next-gen-$$-$RANDOM.sh"
    log "Copying helper script to ${target}:${remote_script}"
    # shellcheck disable=SC2029
    ssh "${ssh_opts[@]}" "$target" "cat > '$remote_script' && chmod +x '$remote_script'" <"$0"

    local remote_cmd
    remote_cmd="${remote_script} $(build_remote_args)"

    log "Executing ${ACTION} on ${target}"
    # shellcheck disable=SC2029
    if ! ssh -tt "${ssh_opts[@]}" "$target" "$remote_cmd"; then
        # shellcheck disable=SC2029
        ssh "${ssh_opts[@]}" "$target" "rm -f '$remote_script'" >/dev/null 2>&1 || true
        die "Remote ${ACTION} failed"
    fi

    # shellcheck disable=SC2029
    ssh "${ssh_opts[@]}" "$target" "rm -f '$remote_script'" >/dev/null 2>&1 || true
}

parse_args() {
    if [[ $# -lt 1 ]]; then
        usage
        exit 1
    fi

    ACTION="$1"
    shift

    case "$ACTION" in
        deploy | destroy)
            ;;
        -h | --help | help)
            usage
            exit 0
            ;;
        *)
            die "Unknown action: ${ACTION}"
            ;;
    esac

    while [[ $# -gt 0 ]]; do
        case "$1" in
            --host)
                need_arg "$1" "${2-}"
                TARGET_HOST="$2"
                shift 2
                ;;
            --project-dir)
                need_arg "$1" "${2-}"
                PROJECT_DIR="$2"
                shift 2
                ;;
            --ssh-user)
                need_arg "$1" "${2-}"
                SSH_USER="$2"
                shift 2
                ;;
            --ssh-port)
                need_arg "$1" "${2-}"
                SSH_PORT="$2"
                shift 2
                ;;
            --ssh-key)
                need_arg "$1" "${2-}"
                SSH_KEY="$2"
                SSH_KEY_SET="1"
                shift 2
                ;;
            --cluster-name)
                need_arg "$1" "${2-}"
                CLUSTER_NAME="$2"
                shift 2
                ;;
            --tiup-version)
                need_arg "$1" "${2-}"
                TIUP_VERSION="$2"
                shift 2
                ;;
            --keyspace-count)
                need_arg "$1" "${2-}"
                KEYSPACE_COUNT="$2"
                shift 2
                ;;
            --tidb-bin)
                need_arg "$1" "${2-}"
                TIDB_BIN="$2"
                TIDB_BIN_SET="1"
                shift 2
                ;;
            --pd-endpoint)
                need_arg "$1" "${2-}"
                PD_ENDPOINT="$2"
                PD_ENDPOINT_SET="1"
                shift 2
                ;;
            --cdc-server)
                need_arg "$1" "${2-}"
                CDC_SERVER="$2"
                CDC_SERVER_SET="1"
                shift 2
                ;;
            --keyspace-tidb-base-port)
                need_arg "$1" "${2-}"
                KEYSPACE_TIDB_BASE_PORT="$2"
                KEYSPACE_TIDB_BASE_PORT_SET="1"
                shift 2
                ;;
            --keyspace-status-base-port)
                need_arg "$1" "${2-}"
                KEYSPACE_STATUS_BASE_PORT="$2"
                KEYSPACE_STATUS_BASE_PORT_SET="1"
                shift 2
                ;;
            --patch-tikv)
                need_arg "$1" "${2-}"
                PATCH_TIKV="$2"
                PATCH_TIKV_SET="1"
                shift 2
                ;;
            --patch-tidb)
                need_arg "$1" "${2-}"
                PATCH_TIDB="$2"
                PATCH_TIDB_SET="1"
                shift 2
                ;;
            --patch-pd)
                need_arg "$1" "${2-}"
                PATCH_PD="$2"
                PATCH_PD_SET="1"
                shift 2
                ;;
            --patch-cdc)
                need_arg "$1" "${2-}"
                PATCH_CDC="$2"
                PATCH_CDC_SET="1"
                shift 2
                ;;
            --skip-minio)
                SKIP_MINIO="1"
                shift 1
                ;;
            --create-changefeed)
                CREATE_CHANGEFEED="1"
                shift 1
                ;;
            --source-keyspace)
                need_arg "$1" "${2-}"
                SOURCE_KEYSPACE="$2"
                shift 2
                ;;
            --changefeed-id)
                need_arg "$1" "${2-}"
                CHANGEFEED_ID="$2"
                shift 2
                ;;
            --sink-uri)
                need_arg "$1" "${2-}"
                SINK_URI="$2"
                shift 2
                ;;
            --sink-deployed-keyspace)
                need_arg "$1" "${2-}"
                SINK_DEPLOYED_KEYSPACE="$2"
                shift 2
                ;;
            --sink-host)
                need_arg "$1" "${2-}"
                SINK_HOST="$2"
                shift 2
                ;;
            --sink-user)
                need_arg "$1" "${2-}"
                SINK_USER="$2"
                shift 2
                ;;
            --sink-password)
                need_arg "$1" "${2-}"
                SINK_PASSWORD="$2"
                shift 2
                ;;
            --_remote-exec)
                REMOTE_EXEC="1"
                shift 1
                ;;
            -h | --help)
                usage
                exit 0
                ;;
            *)
                die "Unknown option: $1"
                ;;
        esac
    done
}

main() {
    parse_args "$@"
    validate_inputs

    if [[ "$REMOTE_EXEC" == "1" ]]; then
        case "$ACTION" in
            deploy)
                deploy_local
                ;;
            destroy)
                destroy_local
                ;;
            *)
                die "Unsupported action: ${ACTION}"
                ;;
        esac
        return
    fi

    run_remote_action
}

main "$@"