#!/usr/bin/env bash

set -euo pipefail

readonly NAMESPACE="tidb-serverless"
readonly SOURCE_STS="tikv-api"
readonly SOURCE_CONTAINER="tikv-api"
readonly SERVICE_ACCOUNT="backup-restore"
readonly DATA_MOUNT_PATH="/data"
readonly CLUSTER_ID_MOUNT_PATH="/var/run/packed-backup-selection/cluster-ids"
readonly MAX_CLUSTER_ID_FILE_BYTES=$((900 * 1024))
readonly DEFAULT_TTL_SECONDS_AFTER_FINISHED=$((30 * 24 * 60 * 60))

usage() {
  cat <<'EOF'
Usage:
  generate-pack-backup-offline.sh [OPTIONS] -- PACK-BACKUP-ARGS...

Options:
  --name=NAME                    Job group name (defaults to pack-backup-offline-<time>)
  --workflow-name=NAME           Stable workflow RUN_ID label
  --shards=N                     Generate N zero-based shard Jobs
  --image=IMAGE                  Container image (defaults to tikv-api's image)
  --node-group=GROUP             Node group label and toleration (default: general)
  --cpu.request=QUANTITY         CPU request
  --cpu.limit=QUANTITY           CPU limit (defaults to the CPU request)
  --mem.request=QUANTITY         Memory request
  --mem.limit=QUANTITY           Memory limit (defaults to the memory request)
  --ebs.request=QUANTITY         Use a temporary EBS volume of this size for /data
  --ebs.storage-class=NAME       EBS StorageClass (defaults to tikv-api's data class)
  --cluster-id-in-file=PATH      Embed and mount cluster IDs from a local file
  --cluster-id-config-map=NAME   Mount cluster IDs from an existing ConfigMap
  --cluster-id-dfs-object-key=KEY
                                 Read cluster IDs directly from DFS_S3_BUCKET
  --output=FORMAT                Output yaml or deterministic json (default: yaml)
  -h, --help                     Show this help

The generated Job YAML is written to stdout and is never applied to the cluster.
The script injects /cse-ctl pack-backup, --offline, and --data-dir /data.
Shard Jobs share the label app.kubernetes.io/instance=<job-group-name>.

Example:
  generate-pack-backup-offline.sh \
    --shards=4 --cpu.request=500m --mem.request=2Gi --ebs.request=100Gi -- \
    --backup-name 20260717/000044.meta \
    --keyspace-name 3Ccz3CgjfFciuUg
EOF
}

die() {
  echo "error: $*" >&2
  exit 1
}

take_value() {
  local option="$1"
  local value="${2-}"
  [[ -n "$value" ]] || die "$option requires a value"
  printf '%s' "$value"
}

job_name=""
workflow_name=""
shards=""
shards_set=false
image_override=""
node_group="general"
cpu_request=""
cpu_limit=""
mem_request=""
mem_limit=""
ebs_request=""
ebs_storage_class=""
cluster_id_in_file=""
cluster_id_config_map=""
cluster_id_dfs_object_key=""
output_format="yaml"
cse_args=()

seen_generator_options="|"
for argument in "$@"; do
  [[ "$argument" != -- ]] || break
  option="${argument%%=*}"
  case "$option" in
    --name|--workflow-name|--shards|--image|--node-group|--cpu.request|--cpu.limit|--ebs.request|--ebs.storage-class|--cluster-id-in-file|--cluster-id-config-map|--cluster-id-dfs-object-key|--output)
      canonical_option="$option"
      ;;
    --mem.request|--memory.request)
      canonical_option="--memory.request"
      ;;
    --mem.limit|--memory.limit)
      canonical_option="--memory.limit"
      ;;
    --ebs)
      canonical_option="--ebs.request"
      ;;
    *)
      continue
      ;;
  esac
  if [[ "$seen_generator_options" == *"|${canonical_option}|"* ]]; then
    die "generator option '$canonical_option' must not be repeated"
  fi
  seen_generator_options+="${canonical_option}|"
done

while (($# > 0)); do
  case "$1" in
    --)
      shift
      cse_args=("$@")
      break
      ;;
    --name=*)
      job_name="$(take_value --name "${1#*=}")"
      ;;
    --name)
      shift
      job_name="$(take_value --name "${1-}")"
      ;;
    --workflow-name=*)
      workflow_name="$(take_value --workflow-name "${1#*=}")"
      ;;
    --workflow-name)
      shift
      workflow_name="$(take_value --workflow-name "${1-}")"
      ;;
    --shards=*)
      shards="$(take_value --shards "${1#*=}")"
      shards_set=true
      ;;
    --shards)
      shift
      shards="$(take_value --shards "${1-}")"
      shards_set=true
      ;;
    --image=*)
      image_override="$(take_value --image "${1#*=}")"
      ;;
    --image)
      shift
      image_override="$(take_value --image "${1-}")"
      ;;
    --node-group=*)
      node_group="$(take_value --node-group "${1#*=}")"
      ;;
    --node-group)
      shift
      node_group="$(take_value --node-group "${1-}")"
      ;;
    --cpu.request=*)
      cpu_request="$(take_value --cpu.request "${1#*=}")"
      ;;
    --cpu.request)
      shift
      cpu_request="$(take_value --cpu.request "${1-}")"
      ;;
    --cpu.limit=*)
      cpu_limit="$(take_value --cpu.limit "${1#*=}")"
      ;;
    --cpu.limit)
      shift
      cpu_limit="$(take_value --cpu.limit "${1-}")"
      ;;
    --mem.request=*|--memory.request=*)
      mem_request="$(take_value "${1%%=*}" "${1#*=}")"
      ;;
    --mem.request|--memory.request)
      option="$1"
      shift
      mem_request="$(take_value "$option" "${1-}")"
      ;;
    --mem.limit=*|--memory.limit=*)
      mem_limit="$(take_value "${1%%=*}" "${1#*=}")"
      ;;
    --mem.limit|--memory.limit)
      option="$1"
      shift
      mem_limit="$(take_value "$option" "${1-}")"
      ;;
    --ebs.request=*|--ebs=*)
      ebs_request="$(take_value "${1%%=*}" "${1#*=}")"
      ;;
    --ebs.request|--ebs)
      option="$1"
      shift
      ebs_request="$(take_value "$option" "${1-}")"
      ;;
    --ebs.storage-class=*)
      ebs_storage_class="$(take_value --ebs.storage-class "${1#*=}")"
      ;;
    --ebs.storage-class)
      shift
      ebs_storage_class="$(take_value --ebs.storage-class "${1-}")"
      ;;
    --cluster-id-in-file=*)
      cluster_id_in_file="$(take_value --cluster-id-in-file "${1#*=}")"
      ;;
    --cluster-id-in-file)
      shift
      cluster_id_in_file="$(take_value --cluster-id-in-file "${1-}")"
      ;;
    --cluster-id-config-map=*)
      cluster_id_config_map="$(take_value --cluster-id-config-map "${1#*=}")"
      ;;
    --cluster-id-config-map)
      shift
      cluster_id_config_map="$(take_value --cluster-id-config-map "${1-}")"
      ;;
    --cluster-id-dfs-object-key=*)
      cluster_id_dfs_object_key="$(take_value --cluster-id-dfs-object-key "${1#*=}")"
      ;;
    --cluster-id-dfs-object-key)
      shift
      cluster_id_dfs_object_key="$(take_value --cluster-id-dfs-object-key "${1-}")"
      ;;
    --output=*)
      output_format="$(take_value --output "${1#*=}")"
      ;;
    --output)
      shift
      output_format="$(take_value --output "${1-}")"
      ;;
    -h|--help)
      usage
      exit 0
      ;;
    *)
      die "unknown generator option '$1' (cse-ctl arguments must follow --)"
      ;;
  esac
  shift
done

for value_name in job_name workflow_name shards image_override node_group cpu_request cpu_limit mem_request mem_limit ebs_request ebs_storage_class cluster_id_in_file cluster_id_config_map cluster_id_dfs_object_key output_format; do
  value="${!value_name}"
  [[ "$value" != "" ]] || continue
  [[ "$value" != -* ]] || die "${value_name//_/.} must not start with '-'"
done

[[ ${#cse_args[@]} -gt 0 ]] || die "missing pack-backup arguments after --"
[[ "$output_format" == yaml || "$output_format" == json ]] \
  || die "--output must be yaml or json"

if [[ "$shards_set" == true ]]; then
  [[ "$shards" =~ ^[0-9]+$ ]] || die "--shards must be a positive integer"
  shard_total=$((10#$shards))
  ((shard_total > 0)) || die "--shards must be a positive integer"
  ((shard_total <= 4294967295)) || die "--shards exceeds cse-ctl's u32 range"
else
  shard_total=1
fi

case "${cse_args[0]}" in
  cse-ctl|/cse-ctl|pack-backup)
    die "do not pass cse-ctl or pack-backup; the script injects them"
    ;;
esac

if [[ -n "$cpu_limit" && -z "$cpu_request" ]]; then
  die "--cpu.limit requires --cpu.request"
fi
if [[ -n "$mem_limit" && -z "$mem_request" ]]; then
  die "--mem.limit requires --mem.request"
fi
if [[ -n "$ebs_storage_class" && -z "$ebs_request" ]]; then
  die "--ebs.storage-class requires --ebs.request"
fi
cluster_id_input_option_count=0
[[ -z "$cluster_id_in_file" ]] || cluster_id_input_option_count=$((cluster_id_input_option_count + 1))
[[ -z "$cluster_id_config_map" ]] || cluster_id_input_option_count=$((cluster_id_input_option_count + 1))
[[ -z "$cluster_id_dfs_object_key" ]] || cluster_id_input_option_count=$((cluster_id_input_option_count + 1))
if ((cluster_id_input_option_count > 1)); then
  die "cluster ID input options are mutually exclusive"
fi

has_offline=false
backup_name=""
backup_name_count=0
for ((index = 0; index < ${#cse_args[@]}; index++)); do
  arg="${cse_args[index]}"
  case "$arg" in
    --backup-name)
      ((index + 1 < ${#cse_args[@]})) || die "--backup-name requires a value"
      value="${cse_args[index + 1]}"
      [[ -n "$value" && "$value" != -* ]] || die "--backup-name requires a value"
      backup_name="$value"
      backup_name_count=$((backup_name_count + 1))
      index=$((index + 1))
      ;;
    --backup-name=*)
      value="${arg#*=}"
      [[ -n "$value" ]] || die "--backup-name requires a value"
      backup_name="$value"
      backup_name_count=$((backup_name_count + 1))
      ;;
    --use-id|--use-id=*)
      die "do not pass --use-id; the script uses each Job name as its pack ID"
      ;;
    --data-dir|--data-dir=*)
      die "do not pass --data-dir; the script injects --data-dir $DATA_MOUNT_PATH"
      ;;
    --keyspace-select.shard-number|--keyspace-select.shard-number=*|--keyspace-select.shard-total|--keyspace-select.shard-total=*)
      die "do not pass keyspace shard flags; use --shards"
      ;;
    --keyspace-select.cluster-id-in|--keyspace-select.cluster-id-in=*)
      die "do not pass --keyspace-select.cluster-id-in directly; use a cluster ID input option"
      ;;
    --offline)
      has_offline=true
      ;;
  esac
done
((backup_name_count == 1)) || die "pack-backup arguments must contain exactly one --backup-name"
if [[ "$has_offline" == false ]]; then
  cse_args+=(--offline)
fi

if [[ -z "$job_name" ]]; then
  job_name="pack-backup-offline-$(date +%Y%m%d-%H%M%S)"
fi

if ((${#job_name} > 63)) || [[ ! "$job_name" =~ ^[a-z0-9]([-a-z0-9]*[a-z0-9])?$ ]]; then
  die "invalid Kubernetes Job group name '$job_name'"
fi
if [[ -n "$workflow_name" ]] \
  && { ((${#workflow_name} > 63)) \
    || [[ ! "$workflow_name" =~ ^[a-z0-9]([-a-z0-9]*[a-z0-9])?$ ]]; }; then
  die "invalid workflow name '$workflow_name'"
fi
if ((${#node_group} > 63)) \
  || [[ ! "$node_group" =~ ^[A-Za-z0-9]([-A-Za-z0-9_.]*[A-Za-z0-9])?$ ]]; then
  die "invalid Kubernetes node-group label value '$node_group'"
fi

if [[ "$shards_set" == true ]]; then
  longest_shard_name="${job_name}-shard-$((shard_total - 1))-of-${shard_total}"
  ((${#longest_shard_name} <= 63)) || die "sharded Job name '$longest_shard_name' exceeds 63 characters"
fi

has_cluster_id_in=false
mount_cluster_id_in=false
embed_cluster_id_config_map=false
cluster_id_input=""
if [[ -n "$cluster_id_in_file" ]]; then
  [[ -f "$cluster_id_in_file" ]] || die "--cluster-id-in-file is not a regular file"
  [[ -r "$cluster_id_in_file" ]] || die "--cluster-id-in-file is not readable"
  cluster_id_file_bytes="$(wc -c <"$cluster_id_in_file")"
  ((cluster_id_file_bytes > 0)) || die "--cluster-id-in-file is empty"
  ((cluster_id_file_bytes <= MAX_CLUSTER_ID_FILE_BYTES)) \
    || die "--cluster-id-in-file exceeds ${MAX_CLUSTER_ID_FILE_BYTES} bytes"
  cluster_id_config_map="${job_name}-cluster-ids"
  ((${#cluster_id_config_map} <= 63)) \
    || die "cluster ID ConfigMap name '$cluster_id_config_map' exceeds 63 characters"
  has_cluster_id_in=true
  mount_cluster_id_in=true
  cluster_id_input="$CLUSTER_ID_MOUNT_PATH"
  embed_cluster_id_config_map=true
elif [[ -n "$cluster_id_config_map" ]]; then
  if ((${#cluster_id_config_map} > 63)) \
    || [[ ! "$cluster_id_config_map" =~ ^[a-z0-9]([-a-z0-9]*[a-z0-9])?$ ]]; then
    die "invalid cluster ID ConfigMap name '$cluster_id_config_map'"
  fi
  has_cluster_id_in=true
  mount_cluster_id_in=true
  cluster_id_input="$CLUSTER_ID_MOUNT_PATH"
elif [[ -n "$cluster_id_dfs_object_key" ]]; then
  has_cluster_id_in=true
  cluster_id_input="dfs:${cluster_id_dfs_object_key}"
fi

command -v kubectl >/dev/null || die "kubectl is required"
command -v jq >/dev/null || die "jq is required"

sts_json="$(kubectl --request-timeout=30s -n "$NAMESPACE" \
  get statefulset "$SOURCE_STS" -o json)"

container_json="$(jq -ce --arg name "$SOURCE_CONTAINER" '
  (.spec.template.spec.containers // [] | map(select(.name == $name)) | first)
    // error("container " + $name + " not found in source StatefulSet")
' <<<"$sts_json")"

if [[ -n "$image_override" ]]; then
  image="$image_override"
  image_pull_policy="Always"
else
  image="$(jq -er '.image // error("source container has no image")' <<<"$container_json")"
  image_pull_policy="$(jq -r '.imagePullPolicy // "IfNotPresent"' <<<"$container_json")"
fi

env_json="$(jq -ce '
  [
    "DFS_PREFIX",
    "DFS_S3_BUCKET",
    "DFS_S3_REGION",
    "CSE_MASTER_KEY_ID",
    "CSE_MASTER_KEY_CIPHER_TEXT",
    "CSE_MASTER_KEY_VENDOR"
  ] as $required
  | [
      $required[] as $name
      | ((.env // []) | map(select(.name == $name))) as $matches
      | if ($matches | length) == 1 then
          $matches[0]
        else
          error("expected exactly one " + $name + " entry in the source container")
        end
    ]
' <<<"$container_json")"

dfs_s3_bucket="$(jq -er '
  map(select(.name == "DFS_S3_BUCKET"))[0]
  | if (.value | type) == "string" and (.value | length) > 0 then .value
    else error("DFS_S3_BUCKET must have a non-empty literal value") end
' <<<"$env_json")"
dfs_prefix="$(jq -er '
  map(select(.name == "DFS_PREFIX"))[0]
  | if (.value | type) == "string" then .value
    else error("DFS_PREFIX must have a literal value") end
' <<<"$env_json")"

if [[ "$dfs_s3_bucket" == */* || "$dfs_s3_bucket" == *"?"* \
  || "$dfs_s3_bucket" == *"#"* || "$dfs_s3_bucket" =~ [[:space:]] ]]; then
  die "invalid literal DFS_S3_BUCKET value"
fi
while [[ "$dfs_prefix" == /* ]]; do dfs_prefix="${dfs_prefix#/}"; done
while [[ "$dfs_prefix" == */ ]]; do dfs_prefix="${dfs_prefix%/}"; done
if [[ "$dfs_prefix" == *//* || "$dfs_prefix" == *"?"* \
  || "$dfs_prefix" == *"#"* || "$dfs_prefix" =~ [[:space:]] ]]; then
  die "invalid literal DFS_PREFIX value"
fi
[[ "$backup_name" != /* && "$backup_name" != */ && "$backup_name" != *//* ]] || die "invalid --backup-name for pack prefix"

pack_prefix_root="s3://${dfs_s3_bucket}/"
if [[ -n "$dfs_prefix" ]]; then
  pack_prefix_root+="${dfs_prefix}/"
fi
pack_prefix_root+="packed_backup/native/${backup_name}/"

resources_json="$(jq -cn \
  --arg cpu_request "$cpu_request" \
  --arg cpu_limit "$cpu_limit" \
  --arg mem_request "$mem_request" \
  --arg mem_limit "$mem_limit" '
  {}
  | if $cpu_request != "" then
      .requests.cpu = $cpu_request
      | .limits.cpu = (if $cpu_limit != "" then $cpu_limit else $cpu_request end)
    else . end
  | if $mem_request != "" then
      .requests.memory = $mem_request
      | .limits.memory = (if $mem_limit != "" then $mem_limit else $mem_request end)
    else . end
')"

if [[ -n "$ebs_request" ]]; then
  if [[ -z "$ebs_storage_class" ]]; then
    ebs_storage_class="$(jq -r '
      [.spec.volumeClaimTemplates[]?.spec.storageClassName | select(. != null and . != "")]
      | first // empty
    ' <<<"$sts_json")"
  fi
  [[ -n "$ebs_storage_class" ]] || die "source StatefulSet has no EBS StorageClass; use --ebs.storage-class"

  volume_json="$(jq -cn \
    --arg size "$ebs_request" \
    --arg storage_class "$ebs_storage_class" '
    {
      name: "data",
      ephemeral: {
        volumeClaimTemplate: {
          spec: {
            accessModes: ["ReadWriteOnce"],
            storageClassName: $storage_class,
            volumeMode: "Filesystem",
            resources: {requests: {storage: $size}}
          }
        }
      }
    }
  ')"
else
  volume_json='{"name":"data","emptyDir":{}}'
fi

command_base_json="$(jq -cn --args '$ARGS.positional' -- \
  /cse-ctl pack-backup "${cse_args[@]}")"

cluster_id_jq_args=(
  --arg cluster_id_config_map "$cluster_id_config_map"
  --arg cluster_id_input "$cluster_id_input"
  --arg cluster_ids ""
  --argjson has_cluster_id_in "$has_cluster_id_in"
  --argjson mount_cluster_id_in "$mount_cluster_id_in"
  --argjson embed_cluster_id_config_map "$embed_cluster_id_config_map"
)
if [[ "$embed_cluster_id_config_map" == true ]]; then
  cluster_id_jq_args=(
    --arg cluster_id_config_map "$cluster_id_config_map"
    --arg cluster_id_input "$cluster_id_input"
    --rawfile cluster_ids "$cluster_id_in_file"
    --argjson has_cluster_id_in true
    --argjson mount_cluster_id_in true
    --argjson embed_cluster_id_config_map true
  )
fi

output_command=(kubectl create --dry-run=client --validate=false -f - -o yaml)
if [[ "$output_format" == json ]]; then
  output_command=(cat)
fi

# The client dry-run is only used for the optional JSON-to-YAML conversion.
jq -cn \
  --arg name "$job_name" \
  --arg workflow_name "$workflow_name" \
  --arg namespace "$NAMESPACE" \
  --arg image "$image" \
  --arg image_pull_policy "$image_pull_policy" \
  --arg service_account "$SERVICE_ACCOUNT" \
  --arg node_group "$node_group" \
  --argjson ttl_seconds_after_finished "$DEFAULT_TTL_SECONDS_AFTER_FINISHED" \
  --arg data_mount_path "$DATA_MOUNT_PATH" \
  --arg pack_prefix_root "$pack_prefix_root" \
  --argjson command_base "$command_base_json" \
  --argjson env "$env_json" \
  --argjson resources "$resources_json" \
  --argjson volume "$volume_json" \
  --argjson shard_total "$shard_total" \
  --argjson sharding_enabled "$shards_set" \
  "${cluster_id_jq_args[@]}" '
  def workflow_labels:
    if $workflow_name == "" then {}
    else {"cse.tidbcloud.com/workflow-name": $workflow_name}
    end;
  def stage_labels($job_name):
    workflow_labels + {
      "cse.tidbcloud.com/stage": "pack",
      "cse.tidbcloud.com/job-name": $job_name
    };
  {
    apiVersion: "v1",
    kind: "List",
    items: (
      (if $embed_cluster_id_config_map then [{
        apiVersion: "v1",
        kind: "ConfigMap",
        metadata: {
          name: $cluster_id_config_map,
          namespace: $namespace,
          labels: (
            {
              "app.kubernetes.io/name": "cse-pack-backup-offline",
              "app.kubernetes.io/instance": $name,
              "cse.tidbcloud.com/resource-role": "cluster-selection"
            }
            + workflow_labels
          )
        },
        data: {
          "cluster-ids": $cluster_ids
        }
      }] else [] end)
      + [
        range(0; $shard_total) as $shard
        | ($shard | tostring) as $shard_number
        | ($shard_total | tostring) as $shard_total_text
        | (if $sharding_enabled then
             $name + "-shard-" + $shard_number + "-of-" + $shard_total_text
           else
             $name
           end) as $job_name
        | {
          apiVersion: "batch/v1",
          kind: "Job",
          metadata: {
            name: $job_name,
            namespace: $namespace,
            labels: (
              {
                "app.kubernetes.io/name": "cse-pack-backup-offline",
                "app.kubernetes.io/instance": $name,
                "cse.tidbcloud.com/shard-number": $shard_number,
                "cse.tidbcloud.com/shard-total": $shard_total_text
              }
              + stage_labels($job_name)
            ),
            annotations: {
              "cse.tidbcloud.com/pack-id": $job_name,
              "cse.tidbcloud.com/pack-backup-prefix": ($pack_prefix_root + $job_name + "/")
            }
          },
          spec: {
            backoffLimit: 0,
            ttlSecondsAfterFinished: $ttl_seconds_after_finished,
            template: {
              metadata: {
                labels: (
                  {
                    "app.kubernetes.io/name": "cse-pack-backup-offline",
                    "app.kubernetes.io/instance": $name,
                    "cse.tidbcloud.com/shard-number": $shard_number,
                    "cse.tidbcloud.com/shard-total": $shard_total_text
                  }
                  + stage_labels($job_name)
                )
              },
              spec: {
                restartPolicy: "Never",
                serviceAccountName: $service_account,
                nodeSelector: {
                  "kubernetes.io/arch": "arm64",
                  "serverless.tidbcloud.com/node": $node_group
                },
                tolerations: [{
                  key: "serverless.tidbcloud.com/node",
                  operator: "Equal",
                  value: $node_group,
                  effect: "NoSchedule"
                }],
                containers: [(
                  {
                    name: "cse-pack-backup-offline",
                    image: $image,
                    imagePullPolicy: $image_pull_policy,
                    command: (
                      [
                        "/bin/bash", "-c",
                        "\"$@\" 2>&1 | grep -Ev \"$0\"\nexit \"${PIPESTATUS[0]}\"",
                        "[0-9A-F]{16}"
                      ]
                      + $command_base
                      + (if $has_cluster_id_in then [
                          "--keyspace-select.cluster-id-in",
                          $cluster_id_input
                        ] else [] end)
                      + (if $sharding_enabled then [
                          "--keyspace-select.shard-number", $shard_number,
                          "--keyspace-select.shard-total", $shard_total_text
                        ] else [] end)
                      + [
                          "--use-id", $job_name,
                          "--data-dir", $data_mount_path
                        ]
                    ),
                    env: $env,
                    volumeMounts: (
                      [{name: "data", mountPath: $data_mount_path}]
                      + (if $mount_cluster_id_in then [{
                          name: "cluster-id-in",
                          mountPath: $cluster_id_input,
                          subPath: "cluster-ids",
                          readOnly: true
                        }] else [] end)
                    )
                  }
                  + if ($resources | length) > 0 then {resources: $resources} else {} end
                )],
                volumes: (
                  [(
                    $volume
                    | if has("ephemeral") then
                        .ephemeral.volumeClaimTemplate.metadata.labels =
                          stage_labels($job_name)
                      else .
                      end
                  )]
                  + (if $mount_cluster_id_in then [{
                      name: "cluster-id-in",
                      configMap: {
                        name: $cluster_id_config_map,
                        defaultMode: 292
                      }
                    }] else [] end)
                )
              }
            }
          }
        }
      ]
    )
  }
' | "${output_command[@]}"