#!/bin/sh
set -eu

# see https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-migration-post-feature-upgrade

# VARS ----------------------------------------- #
ES_URL="${ES_URL:-http://localhost:9200}"
TERM_WIDTH=$(tput cols 2>/dev/null || echo 120)
DISK_EMPTY_SPACE_FACTOR=1.25
REINDEX_SLICES_NUM=5
REINDEX_SLICES_DYNAMIC=${REINDEX_SLICES_DYNAMIC:-1}
# Default batch size: smaller batches reduce memory and merge pressure
# Can be overridden by env var REINDEX_BATCH_SIZE
REINDEX_BATCH_SIZE=${REINDEX_BATCH_SIZE:-1000}
# Multiplier used in aggressive mode to compute slices
REINDEX_SLICES_MULTIPLIER=${REINDEX_SLICES_MULTIPLIER:-7}
# When set to 1 enable very aggressive optimizations (disable refresh, replicas,
# drop analyzers in templates, large batch size). Use with caution.
AGGRESSIVE_REINDEX=${AGGRESSIVE_REINDEX:-0}
# Requests per second for reindex. -1 = unlimited. Set to a number to throttle.
# Default to unlimited (-1). Set via env REINDEX_RPS to a numeric throttle if desired.
REINDEX_RPS=${REINDEX_RPS:--1}
# Optional global hard cap for batch size to avoid huge bulks (docs)
REINDEX_BATCH_SIZE_MAX=${REINDEX_BATCH_SIZE_MAX:-10000}
# If set to 1, the script will skip ingest pipelines during reindex (faster)
SKIP_PIPELINE=${SKIP_PIPELINE:-0}
SINGLE_NODE=${SINGLE_NODE:-}
# Accumulate perf mode payloads to ensure we can restore settings on exit
RESTORE_PAYLOADS=""
# Define migrations as colon-separated values: source:destination:pipeline:mapping_var
# The optional fourth field names an env var (containing JSON) to merge into
# the base mappings for that destination. Leave empty to use defaults.
migrates='status-*;status;default_pipeline;STATUS_MAPPINGS_JSON
lastlog-*;lastlog;lastlog;
sbc-rtp-stats-*;rtp;default_pipeline;RTP_MAPPINGS_JSON
alarms-*;alarms;default_pipeline;ALARMS_MAPPINGS_JSON
logstashh-*;calls;default_pipeline;CALLS_MAPPINGS_JSON
collectd-*;collectd;default_pipeline;COLLECTD_MAPPINGS_JSON'

# CLI argument parsing: allow overriding `ES_URL` from the command line.
print_usage() {
  cat <<EOF
Usage: $(basename "$0") [--es-url URL] [-u URL] [--help]

Options:
  --es-url URL    Elasticsearch URL (overrides environment ES_URL)
  -u URL          Short form for --es-url
  --reindex-batch-size N, --batch-size N, -b N
                  Number of docs per bulk request during reindex (overrides REINDEX_BATCH_SIZE)
  --reindex-slices-multiplier N
                  Multiplier for computing slices when dynamic slices enabled (overrides REINDEX_SLICES_MULTIPLIER)
  --reindex-slices-num N
                  Default fixed number of slices when dynamic slices disabled (overrides REINDEX_SLICES_NUM)
  --reindex-slices-dynamic N
                  Enable (1) or disable (0) dynamic slice calculation (overrides REINDEX_SLICES_DYNAMIC)
  --reindex-rps N  Requests per second throttle for reindex (-1 = unlimited) (overrides REINDEX_RPS)
  --reindex-batch-size-max N
                  Global hard cap for batch size (overrides REINDEX_BATCH_SIZE_MAX)
  --aggressive-reindex / --no-aggressive-reindex
                  Enable/disable aggressive reindex tuning (overrides AGGRESSIVE_REINDEX)
  --skip-pipeline / --no-skip-pipeline
                  Skip ingest pipelines during reindex (overrides SKIP_PIPELINE)
  --disk-empty-space-factor FLOAT
                  Disk empty space multiplier used for safety checks (overrides DISK_EMPTY_SPACE_FACTOR)
  --single-node / --no-single-node
                  Force single-node cluster behavior (overrides SINGLE_NODE detection)
  -h, --help      Show this help message

Environment variables:
  ES_URL                    Elasticsearch URL used by the script (default: http://localhost:9200)
  REINDEX_BATCH_SIZE
                            Number of docs per bulk request during reindex (default: 1000)
  REINDEX_RPS               Requests per second throttle for reindex (-1 = unlimited) (default: -1)
  REINDEX_SLICES_NUM
                            Default fixed number of slices when dynamic slices disabled (default: 5)
                            Must be a positive integer.
  REINDEX_SLICES_DYNAMIC
                            Enable (1) or disable (0) dynamic slice calculation (default: 1)
                            Must be 0 or 1.
  REINDEX_SLICES_MULTIPLIER
                            Multiplier for computing slices when dynamic slices enabled (default: 7)
                            Must be a positive integer.
  DISK_EMPTY_SPACE_FACTOR
                            Disk empty space multiplier used for safety checks (default: 1.25)
                            Must be a positive number.
EOF
}

while [ "$#" -gt 0 ]; do
  case "$1" in
    --es-url)
      ES_URL="$2"
      shift 2
      ;;
    --reindex-batch-size|--batch-size|-b)
      REINDEX_BATCH_SIZE="$2"
      shift 2
      ;;
    --reindex-batch-size=*)
      REINDEX_BATCH_SIZE="${1#*=}"
      shift
      ;;
    --reindex-slices-multiplier)
      REINDEX_SLICES_MULTIPLIER="$2"
      shift 2
      ;;
    --reindex-slices-multiplier=*)
      REINDEX_SLICES_MULTIPLIER="${1#*=}"
      shift
      ;;
    --reindex-slices-num)
      REINDEX_SLICES_NUM="$2"
      shift 2
      ;;
    --reindex-slices-num=*)
      REINDEX_SLICES_NUM="${1#*=}"
      shift
      ;;
    --reindex-slices-dynamic)
      REINDEX_SLICES_DYNAMIC="$2"
      shift 2
      ;;
    --reindex-slices-dynamic=*)
      REINDEX_SLICES_DYNAMIC="${1#*=}"
      shift
      ;;
    --reindex-rps)
      REINDEX_RPS="$2"
      shift 2
      ;;
    --reindex-rps=*)
      REINDEX_RPS="${1#*=}"
      shift
      ;;
    --reindex-batch-size-max)
      REINDEX_BATCH_SIZE_MAX="$2"
      shift 2
      ;;
    --reindex-batch-size-max=*)
      REINDEX_BATCH_SIZE_MAX="${1#*=}"
      shift
      ;;
    --aggressive-reindex)
      AGGRESSIVE_REINDEX=1
      shift
      ;;
    --aggressive-reindex=*)
      AGGRESSIVE_REINDEX="${1#*=}"
      shift
      ;;
    --no-aggressive-reindex)
      AGGRESSIVE_REINDEX=0
      shift
      ;;
    --skip-pipeline)
      SKIP_PIPELINE=1
      shift
      ;;
    --skip-pipeline=*)
      SKIP_PIPELINE="${1#*=}"
      shift
      ;;
    --no-skip-pipeline)
      SKIP_PIPELINE=0
      shift
      ;;
    --disk-empty-space-factor)
      DISK_EMPTY_SPACE_FACTOR="$2"
      shift 2
      ;;
    --disk-empty-space-factor=*)
      DISK_EMPTY_SPACE_FACTOR="${1#*=}"
      shift
      ;;
    --single-node)
      SINGLE_NODE=1
      shift
      ;;
    --no-single-node)
      SINGLE_NODE=0
      shift
      ;;
    --es-url=*)
      ES_URL="${1#*=}"
      shift
      ;;
    -u)
      ES_URL="$2"
      shift 2
      ;;
    -h|--help)
      print_usage
      exit 0
      ;;
    --)
      shift
      break
      ;;
    -* )
      # Unknown flag; show usage and exit with error
      echo "Unknown option: $1" >&2
      print_usage >&2
      exit 1
      ;;
    *)
      # positional args - break leaving them for later if needed
      break
      ;;
  esac
done

# STYLING -------------------------------------- #
RED="$(printf '\033[31m')"
CYAN="$(printf '\033[36m')"
GREEN="$(printf '\033[32m')"
YELLOW="$(printf '\033[33m')"
BLUE="$(printf '\033[34m')"
PURPLE="$(printf '\033[35m')"
WHITE="$(printf '\033[37m')"

REDB="$(printf '\033[41m')"

BOLD="$(printf '\033[1m')"
DIM="$(printf '\033[2m')"
RESET="$(printf '\033[0m')"
BOLD="$(printf '\033[1m')"

# --- LOGGING HELPERS ----------------------------------------------
# Write logs to stderr so stdout remains valid JSON for callers like `jq`.
log_info()    { printf "${BLUE}%s${RESET}\n"   "[INFO]    $*" >&2; }
log_success() { printf "${GREEN}%s${RESET}\n"  "[OK]      $*" >&2; }
log_warn()    { printf "${YELLOW}%s${RESET}\n" "[WARN]    $*" >&2; }
log_error()   { printf "${RED}%s${RESET}\n"    "[ERROR]   $*" >&2; }
log_curl()   { printf "${PURPLE}%s${RESET}\n"  "[CURL]    curl $*" >&2; }

# --- PROMPT HELPER ------------------------------------------------
prompt_checkpoint() {
  message="$1"
  hint="${2:-}"

  printf '%s\n' "${REDB:-}${WHITE:-}[CONFIRM] ${message}${RESET:-}"

  if [ -n "$hint" ]; then
    printf '%s\n' "${YELLOW:-}          Hint:"
    printf '%s\n' "${hint}${RESET:-}"
  fi

  printf '%s\n' "${DIM:-}          Press Enter to continue...${RESET:-}"
  read dummy
}

# --- PROGRESS HELPER ----------------------------------------------
progress() {
  current="$1"
  total="$2"
  percent=$((current * 100 / total))
  filled=$((percent / 2))  # bar width = 50 chars

  bar=""
  i=0
  while [ "$i" -lt "$filled" ]; do
    bar="${bar}="
    i=$((i + 1))
  done
  while [ "$i" -lt 50 ]; do
    bar="${bar} "
    i=$((i + 1))
  done

  # choose color for filled part
  if [ "$percent" -lt 50 ]; then
    color="$YELLOW"
  else
    color="$GREEN"
  fi

  printf "[${PURPLE}${current}/${total}${RESET}]     [%-50s] %3d%%\n" "${color}${bar}${RESET}" "$percent"
}

# -------------------------------------------------------
# Dynamic table printer
# Usage: print_table "Header1\tHeader2\tHeader3\tHeader4" "10:20:50:auto" <<EOF
# row1col1\trow1col2\trow1col3\trow1col4
# -------------------------------------------------------
print_table() {
  header=$1
  shift
  ratios=$1
  shift || true

  oldIFS=$IFS
  IFS=':'
  set -- $ratios
  IFS=$oldIFS
  cols_count=$#
  total_fixed=0
  auto_cols=0

  for r in "$@"; do
    case $r in
      auto) auto_cols=$((auto_cols + 1)) ;;
      *) total_fixed=$((total_fixed + r)) ;;
    esac
  done

  remain=$((TERM_WIDTH - total_fixed - (cols_count * 3) - 1))
  if [ "$remain" -lt 10 ]; then
    remain=10
  fi
  if [ "$auto_cols" -gt 0 ]; then
    auto_width=$((remain / auto_cols))
  else
    auto_width=0
  fi

  widths=""
  for r in "$@"; do
    case $r in
      auto) widths="$widths $auto_width" ;;
      *) widths="$widths $r" ;;
    esac
  done

  awk -F'\t' \
    -v W="$widths" \
    -v N="$cols_count" \
    -v HEADER="$header" \
    -v BOLD="$BOLD" \
    -v BLUE="$BLUE" \
    -v RESET="$RESET" '
      BEGIN {
        split(W, w, " ")
        split(HEADER, h, "\t")
      }

      function print_border() {
        border = "+"
        for (i = 1; i <= N; i++) {
          width = (w[i] > 0 ? w[i] : 15)
          for (j = 0; j < width + 2; j++) border = border "-"
          border = border "+"
        }
        print border
      }

      BEGIN {
        print_border()
        printf "|"
        for (i = 1; i <= length(h) && i <= N; i++) {
          width = (w[i] > 0 ? w[i] : 15)
          text = h[i]
          if (length(text) > width) text = substr(text, 1, width - 4) "..."
          # printf " %-*s |", width, text
          printf " %s%-*s%s |", BOLD BLUE, width, text, RESET
        }
        printf "\n"
        print_border()
      }

      {
        printf "|"
        for (i = 1; i <= NF && i <= N; i++) {
          width = (w[i] > 0 ? w[i] : 15)
          text = $i
          if (length(text) > width) text = substr(text, 1, width - 4) "..."
          printf " %-*s |", width, text
        }
        printf "\n"
      }

      END {
        print_border()
        printf "\n"
      }'
}

check_for_required_cli_tools() {
  MISSING=0
  local REQUIRED_TOOLS="awk curl jq"

  for tool in $REQUIRED_TOOLS; do
    if ! command -v "$tool" >/dev/null 2>&1; then
      log_error "Error: $tool is not installed or not in PATH."
      MISSING=1
    fi
  done

  if [ "$MISSING" -eq 1 ]; then
    log_info "Please install the missing tool(s) and re-run this script."
    log_info "For Debian/Ubuntu:"
    log_info "  sudo apt install $REQUIRED_TOOLS"
    exit 1
  fi

  log_info "All dependencies are installed."
}

wait_for_active_shards() {
  log_info "Wait for all shards to be active"

  while :; do
      status=$(call_es GET "_cluster/health?level=shards")

      # Extract overall cluster status plus unassigned + initializing shards
      cluster_status=$(echo "$status" | jq -r '.status // "red"')
      unassigned=$(echo "$status" | jq -r '.unassigned_shards')
      initializing=$(echo "$status" | jq -r '.initializing_shards')

      log_info "Cluster status: $cluster_status; Initializing: $initializing, Unassigned: $unassigned"

      # Accept both green and yellow as healthy states (yellow may occur when
      # replicas are unassigned, which is fine during single-node operation or
      # temporary cluster transitions). Continue waiting only while status is red.
      if [ "$cluster_status" = "green" ] || [ "$cluster_status" = "yellow" ]; then
        break
      fi

      sleep 2
  done

  log_success "Cluster health is ${cluster_status} ✅"
}

get_largest_index_size() {
  index_patterns=""

  log_info "Getting largest index size" >&2

  while IFS=";" read -r source destination script mapping_var || [ -n "$source" ]; do
    if [ -z "$index_patterns" ]; then
      index_patterns="$source"
    else
      index_patterns="$index_patterns,$source"
    fi
  done <<EOF
$migrates
EOF

  local resp=$(call_es GET "_cat/indices/${source}?format=json&bytes=b&h=index,store.size&s=store.size:desc&index=${index_patterns}")

  index=$(echo "$resp" | jq -r '.[0].index')
  size=$(echo "$resp" | jq -r '.[0]["store.size"] // 0')

  echo "${index} ${size}"
}

get_available_disk_space() {
  df -B1 / | awk 'NR==2 {print $4}'
}

# Convert bytes to a human-readable string with unit and full-bytes in brackets
human_bytes() {
  bytes="$1"
  if [ -z "${bytes}" ]; then
    printf "0B [0 bytes]"
    return
  fi

  # Use awk to convert to human-readable using 1024 base
  awk -v b="$bytes" 'BEGIN{
    split("B KB MB GB TB PB EB", u, " ")
    val = b + 0
    i = 1
    while (val >= 1024 && i < 7) { val = val / 1024; i++ }
    # choose unit without a space for bytes
    unit = u[i]
    if (unit == "B") {
      printf "%d%s [%d bytes]", val, unit, b
    } else {
      printf "%.2f%s [%d bytes]", val, unit, b
    }
  }'
}

# Format seconds into H:M:S
format_seconds() {
  local secs=$1
  printf '%d:%02d:%02d' $((secs/3600)) $(((secs%3600)/60)) $((secs%60))
}

validate_required_disk_space() {
  result=$(get_largest_index_size)
  largest_index=$(echo "$result" | awk '{print $1}')
  largest_size=$(echo "$result" | awk '{print $2}')
  safe_size=$(awk -v s="$largest_size" -v f="$DISK_EMPTY_SPACE_FACTOR" 'BEGIN {printf "%d", s*f+0.999}')
  total_available=$(get_available_disk_space)

  human_largest=$(human_bytes "$largest_size")
  human_safe=$(human_bytes "$safe_size")
  human_total=$(human_bytes "$total_available")

  log_info "Largest index: $largest_index, ${YELLOW}${human_largest}${RESET}"
  log_info "Safe size with factor $DISK_EMPTY_SPACE_FACTOR: ${YELLOW}${human_safe}${RESET}"
  log_info "Total available disk space: ${YELLOW}${human_total}${RESET}"


  if [ "$total_available" -lt "$safe_size" ]; then
    log_error "Error: Not enough disk space to start reindex migration process"
    prompt_checkpoint "Insufficient disk space detected. Please free up enough space before continuing the migration process."
    validate_required_disk_space
  fi

  log_success "Disk space check passed. Ready to start migration."
}

check_reindex_free_disk_safety() {
  local size=$1
  local safe_size=$(awk -v s="$size" -v f="$DISK_EMPTY_SPACE_FACTOR" 'BEGIN {printf "%d", s*f+0.999}')
  local available_space=$(get_available_disk_space)

  human_req=$(human_bytes "$size")
  human_safe=$(human_bytes "$safe_size")
  human_available=$(human_bytes "$available_space")

  log_info "Required space for reindex ${YELLOW}${human_req}${RESET}"
  log_info "Safe size with factor $DISK_EMPTY_SPACE_FACTOR: ${YELLOW}${human_safe}${RESET}"
  log_info "Total available across cluster ${YELLOW}${human_available}${RESET}"

  if [ "$available_space" -lt "$safe_size" ]; then
    log_error "Error: Not enough disk space to start reindex migration process"
    log_error "Please free up enough space and run the script again."
    exit 1
  fi
}

start_the_migration_feature() {
  resp=$(call_es POST "_migration/system_features")

  if echo "$resp" | jq -e '.accepted == true or .state != null' >/dev/null 2>&1; then
    log_success "Started the migration feature successfully."
  else
    log_warn "Failed to start the migration feature"
    log_warn "$(echo "$resp" | jq -c '.')"
  fi
}

es_version() {
  echo $(call_es GET "") | jq -r '.version.number'
}

call_es() {
  local METHOD="$1"
  local ENDPOINT="$2"
  local DATA="${3:-}"
  local RESP=""

  log_curl "--silent --show-error -X $METHOD '$ES_URL/$ENDPOINT'${DATA:+ -H 'Content-Type: application/json' -d '$DATA'}" >&2

  if [ -n "$DATA" ]; then
    RESP=$(curl --silent --show-error -X "$METHOD" \
      "$ES_URL/$ENDPOINT" \
      -H "Content-Type: application/json" \
      -d "$DATA" 2>&1) || true
  else
    RESP=$(curl --silent --show-error -X "$METHOD" "$ES_URL/$ENDPOINT" 2>&1) || true
  fi

  if [ -z "$RESP" ] || ! echo "$RESP" | jq empty >/dev/null 2>&1 || \
    echo "$RESP" | jq -e 'has("error")' >/dev/null 2>&1; then
      log_error "$(printf 'Elasticsearch request failed or invalid response:\n%s' "$RESP" | sed '1!s/^/          /')" >&2
      prompt_checkpoint "Please ensure Elasticsearch is running and press Enter to retry..." >&2
      RESP=$(call_es "$METHOD" "$ENDPOINT" "$DATA")
  fi

  printf '%s' "$RESP"
}

# Check whether an index exists without treating 404 as an error (returns 0 if exists, 1 otherwise)
index_exists() {
  local idx="$1"
  # Use curl to check HTTP status code to avoid call_es error handling for 404
  http_code=$(curl --silent --show-error -o /dev/null -w '%{http_code}' "$ES_URL/_cat/indices?format=json&index=$idx" 2>/dev/null || echo "000")
  if [ "$http_code" = "200" ]; then
    # additionally ensure the body isn't empty/null; fetch body quietly
    body=$(curl --silent --show-error "$ES_URL/_cat/indices?format=json&index=$idx" 2>/dev/null || echo "")
    # if body is an array with items, consider exists; otherwise treat as non-existent
    if echo "$body" | jq -e 'if type=="array" then length>0 else false end' >/dev/null 2>&1; then
      return 0
    fi
  fi
  return 1
}

# Check whether a data stream exists without treating 404 as an error (returns 0 if exists, 1 otherwise)
data_stream_exists() {
  local ds="$1"
  http_code=$(curl --silent --show-error -o /dev/null -w '%{http_code}' "$ES_URL/_data_stream/$ds" 2>/dev/null || echo "000")
  if [ "$http_code" = "200" ]; then
    body=$(curl --silent --show-error "$ES_URL/_data_stream/$ds" 2>/dev/null || echo "")
    if echo "$body" | jq -e 'has("data_streams") and (.data_streams | length > 0)' >/dev/null 2>&1; then
      return 0
    fi
  fi
  return 1
}

# Common mappings JSON used in multiple template creations
# Uses jq to build a canonical mappings object so we can reuse it across
# several places where templates are created for data streams.
# full definition is defined here: https://gitlab.frafos.net/mirror/moki/-/blob/ae87caddd1d410e8a4691036b4475a35f56d65ca/packages/server/src/services/elasticsearchService.ts#L306
MAPPINGS_JSON=$(jq -n '
{
  "dynamic_templates": [
    {
      "attrs_template": {
        "path_match": "attrs.vars.*",
        "mapping": {
          "type": "{dynamic_type}",
          "index": true
        }
      }
    },
    { "strings_template": { "match_mapping_type": "string", "mapping": { "type": "keyword", "ignore_above": 256, "doc_values": true } } }
  ],
  "properties": {
    "@timestamp": { "type": "date" },
    "attrs": {
      "properties": {
        "from": { "type": "keyword", "ignore_above": 256, "doc_values": true }
      }
    }
  }
}')

CALLS_MAPPINGS_JSON=$(jq -n '
{
  "dynamic_templates": [
    {
      "attrs_template": {
        "path_match": "attrs.vars.*",
        "mapping": {
          "type": "{dynamic_type}",
          "index": true
        }
      }
    },
    { "strings_template": { "match_mapping_type": "string", "mapping": { "type": "keyword", "ignore_above": 256, "doc_values": true } } }
  ],
  "properties": {
            URIs: {
              index: false,
              type: "keyword",
            },
            termination: {
              type: "keyword",
            },
            "tls-cn": {
              type: "keyword",
            },
            headers: {
              properties: {
                http_accept: {
                  index: false,
                  type: "keyword",
                },
                content_type: {
                  index: false,
                  type: "keyword",
                },
                request_path: {
                  index: false,
                  type: "keyword",
                },
                http_version: {
                  index: false,
                  type: "keyword",
                },
                request_method: {
                  index: false,
                  type: "keyword",
                },
                http_host: {
                  index: false,
                  type: "keyword",
                },
                request_uri: {
                  index: false,
                  type: "keyword",
                },
                content_length: {
                  index: false,
                  type: "float",
                },
                http_user_agent: {
                  index: false,
                  type: "keyword",
                },
              },
            },
            countIP: {
              index: false,
              type: "float",
            },
            geoip: {
              dynamic: false,
              type: "object",
              properties: {
                ip: {
                  type: "ip",
                },
                latitude: {
                  type: "float",
                },
                country_name: {
                  type: "keyword",
                },
                city_name: {
                  type: "keyword",
                },
                country_code2: {
                  type: "keyword",
                },
                location: {
                  type: "geo_point",
                },
                location_all: {
                  type: "geo_point",
                },
                longitude: {
                  type: "float",
                },
              },
            },
            exceeded: {
              type: "keyword",
            },
            StatesCA: {
              type: "float",
            },
            success: {
              type: "float",
              index: false,
            },
            failure: {
              type: "float",
              index: false,
            },
            SumFailureSuccess: {
              type: "float",
              index: false,
            },
            CallAttempts: {
              type: "float",
              index: false,
            },
            CallEnd: {
              type: "float",
              index: false,
            },
            AnsweredCalls: {
              type: "float",
              index: false,
            },
            countCall: {
              index: false,
              type: "float",
            },
            countCalls: {
              index: false,
              type: "text",
            },
            type: {
              type: "keyword",
            },
            countURI: {
              index: false,
              type: "float",
            },
            countURIs: {
              index: false,
              type: "text",
            },
            attrs: {
              properties: {
                reg_ts: {
                  index: false,
                  type: "date",
                },
                "rtp-MOScqex-avg-b": {
                  index: false,
                  type: "float",
                },
                ua_expire: {
                  index: false,
                  type: "date",
                },
                "rtp-MOScqex-avg-a": {
                  index: false,
                  type: "float",
                },
                dest_ca_name: {
                  type: "keyword",
                },
                tags: {
                  type: "keyword",
                },
                value: {
                  type: "long",
                },
                "rtp-stats-a": {
                  enabled: false,
                },
                durationGroup: {
                  type: "keyword",
                },
                calls: {
                  type: "float",
                },
                call_to: {
                  type: "float",
                },
                call_from: {
                  type: "float",
                },
                bits_from: {
                  type: "float",
                },
                bits_to: {
                  type: "float",
                },
                type: {
                  type: "keyword",
                },
                conf_id: {
                  type: "keyword",
                },
                count: {
                  type: "float",
                },
                "rtp-stats-b": {
                  enabled: false,
                },
                src_ca_name: {
                  copy_to: "attrs.all_copy",

                  type: "keyword",
                },
                src_ca_id: {
                  copy_to: "attrs.all_copy",

                  type: "keyword",
                },
                "src-port": {
                  type: "long",
                },
                "rtp-lossavg": {
                  type: "float",
                },
                source: {
                  copy_to: "attrs.all_copy",

                  type: "keyword",
                },
                "to-ua": {
                  type: "keyword",
                },
                "from-domain": {
                  type: "keyword",
                },
                originator: {
                  type: "keyword",
                },
                reg_expire: {
                  index: false,
                  type: "date",
                },
                "rtp-lossavg-a": {
                  index: false,
                  type: "float",
                },
                "r-uri": {
                  type: "text",
                  copy_to: "attrs.all_copy",
                  search_analyzer: "whitespace",
                  fields: {
                    keyword: {
                      type: "keyword",
                    },
                  },
                },
                "rtp-lossavg-b": {
                  index: false,
                  type: "float",
                },
                "rtp-direction": {
                  type: "keyword",
                },
                duration: {
                  type: "float",
                },
                hostname: {
                  type: "keyword",
                },
                sourceSubnets: {
                  type: "keyword",
                },
                xcallid: {
                  norms: false,

                  type: "keyword",
                },
                "call-id": {
                  norms: false,

                  type: "keyword",
                },
                from: {
                  type: "text",
                  copy_to: "attrs.all_copy",
                  search_analyzer: "whitespace",
                  fields: {
                    keyword: {
                      type: "keyword",
                    },
                  },
                },
                receiver_ip: {
                  type: "ip",
                },
                dest_dns_only: {
                  type: "boolean",
                },
                "rtp-MOScqex-min": {
                  type: "float",
                },
                "rtp-MOScqex-avg": {
                  type: "float",
                },
                "sip-code": {
                  type: "keyword",
                },
                "from-ua": {
                  norms: false,

                  type: "keyword",
                },
                "x-org-connid": {
                  norms: false,

                  type: "keyword",
                },
                "source-alias": {
                  norms: false,

                  type: "keyword",
                },
                method: {
                  type: "keyword",
                },
                dest_state: {
                  type: "keyword",
                },
                transport: {
                  type: "keyword",
                },
                dst_cc: {
                  type: "keyword",
                },
                "sub-type": {
                  type: "keyword",
                },
                dst_ca_name: {
                  copy_to: "attrs.all_copy",

                  type: "keyword",
                },
                dst_ca_id: {
                  copy_to: "attrs.all_copy",

                  type: "keyword",
                },
                dest_ca_uuid: {
                  type: "keyword",
                },
                ca_uuid: {
                  type: "keyword",
                },
                all_copy: {
                  type: "text",

                  search_analyzer: "whitespace",
                  analyzer: "analyzerWhiteSpaceOnly",
                },
                sbc: {
                  type: "keyword",
                },
                "r-uri-shorted": {
                  type: "keyword",
                },
                src_rlm_name: {
                  type: "keyword",
                },
                dst_rlm_name: {
                  type: "keyword",
                },
                to: {
                  type: "text",
                  copy_to: "attrs.all_copy",
                  search_analyzer: "whitespace",
                  fields: {
                    keyword: {
                      type: "keyword",
                    },
                  },
                },
                "rtp-lossmax": {
                  type: "float",
                },
                "rtp-lossmax-a": {
                  index: false,
                  type: "float",
                },
                "rtp-MOScqex-min-a": {
                  index: false,
                  type: "float",
                },
                "rtp-lossmax-b": {
                  index: false,
                  type: "float",
                },
                "rtp-MOScqex-min-b": {
                  index: false,
                  type: "float",
                },
                start_tm: {
                  index: false,
                  type: "text",
                },
                connect_tm: {
                  index: false,
                  type: "text",
                },
                end_tm: {
                  index: false,
                  type: "text",
                },
                version: {
                  type: "text",
                  fields: {
                    keyword: {
                      type: "keyword",
                      ignore_above: 256,
                    },
                  },
                },
              },
            },
            "@timestamp": {
              type: "date",
            },
            countReg: {
              index: false,
              type: "float",
            },
            countRegs: {
              index: false,
              type: "text",
            },
            durationSum: {
              index: false,
              type: "float",
            },
            id: {
              type: "keyword",
            },
            events: {
              properties: {
                rate_1m: {
                  index: false,
                  type: "float",
                },
                rate_15m: {
                  index: false,
                  type: "float",
                },
                count: {
                  index: false,
                  type: "integer",
                },
                rate_5m: {
                  index: false,
                  type: "float",
                },
              },
            },
            ts: {
              format: "epoch_second",
              type: "date",
            },
          }
}')

ALARMS_MAPPINGS_JSON=$(jq -n '
{
  "dynamic_templates": [
    {
      "attrs_template": {
        "path_match": "attrs.vars.*",
        "mapping": {
          "type": "{dynamic_type}",
          "index": true
        }
      }
    },
    { "strings_template": { "match_mapping_type": "string", "mapping": { "type": "keyword", "ignore_above": 256, "doc_values": true } } }
  ],
  "properties": {
            "@timestamp": {
              type: "date",
            },
            type: {
              type: "keyword",
            },
            severity: {
              type: "keyword",
            },
            exceeded: {
              type: "keyword",
            },
            "exceeded-by": {
              type: "keyword",
            },
          }
}')

COLLECTD_MAPPINGS_JSON=$(jq -n '
{
  "dynamic_templates": [
    {
      "attrs_template": {
        "path_match": "attrs.vars.*",
        "mapping": {
          "type": "{dynamic_type}",
          "index": true
        }
      }
    },
    { "strings_template": { "match_mapping_type": "string", "mapping": { "type": "keyword", "ignore_above": 256, "doc_values": true } } }
  ],
  "properties": {
            "@timestamp": {
              type: "date",
            },
            values: {
              type: "keyword",
            },
            plugin_instance: {
              type: "keyword",
            },
            plugin: {
              type: "keyword",
            },
            host: {
              type: "keyword",
            },
            collectd_type: {
              type: "keyword",
            },
            type: {
              type: "keyword",
            },
            countReg: {
              index: false,
              type: "float",
            },
            countRegs: {
              index: false,
              type: "keyword",
            },
            countCall: {
              index: false,
              type: "float",
            },
            countCalls: {
              index: false,
              type: "keyword",
            },
            tags: {
              type: "keyword",
            },
            type_instance: {
              type: "keyword",
            },
            dstypes: {
              index: false,
              type: "keyword",
            },
            dsnames: {
              index: false,
              type: "keyword",
            },
            interval: {
              type: "float",
            },
            value: {
              type: "float",
            },
            longterm: {
              type: "float",
            },
            midterm: {
              type: "float",
            },
            shortterm: {
              type: "float",
            },
            bitsfrom: {
              type: "long",
            },
            rx: {
              type: "float",
            },
            tx: {
              type: "float",
            },
            time: {
              index: false,
              type: "keyword",
            },
            attrs: {
              properties: {
                hostname: {
                  type: "keyword",
                },
                sbc: {
                  type: "keyword",
                },
                tags: {
                  type: "keyword",
                },
                type: {
                  type: "keyword",
                },
                calls: {
                  type: "float",
                },
                callstarts: {
                  type: "float",
                },
                bits: {
                  type: "float",
                },
                regs: {
                  type: "float",
                },
                uas_trans: {
                  type: "float",
                },
                uac_trans: {
                  type: "float",
                },
                callstartto: {
                  type: "float",
                },
                callstartfrom: {
                  type: "float",
                },
                callto: {
                  type: "float",
                },
                callsto: {
                  type: "float",
                },
                callfrom: {
                  type: "float",
                },
                callsfrom: {
                  type: "float",
                },
                bitsto: {
                  type: "float",
                },
                bitsfrom: {
                  type: "float",
                },
                realm: {
                  type: "keyword",
                },
                src_ca_id: {
                  type: "keyword",
                },
                dst_ca_id: {
                  type: "keyword",
                },
              },
            },
          }
}')


RTP_MAPPINGS_JSON=$(jq -n '
{
  "dynamic_templates": [
    {
      "attrs_template": {
        "path_match": "attrs.vars.*",
        "mapping": {
          "type": "{dynamic_type}",
          "index": true
        }
      }
    },
    { "strings_template": { "match_mapping_type": "string", "mapping": { "type": "keyword", "ignore_above": 256, "doc_values": true } } }
  ],
  "properties": {
            "@timestamp": {
              type: "date",
            },
            id: {
              type: "keyword",
            },
            ts: {
              format: "epoch_second",
              type: "date",
            },
            type: {
              type: "keyword",
            },
            attrs: {
              properties: {
                hostname: {
                  type: "keyword",
                },
                release: {
                  type: "keyword",
                },
                "rtp-remote-ca": {
                  type: "keyword",
                },
                "rtp-remote-realm": {
                  type: "keyword",
                },
                "rtp-remote-ip": {
                  type: "ip",
                },
                "rtp-remote-port": {
                  type: "integer",
                },
                "rtp-recv-packets": {
                  type: "unsigned_long",
                  index: false,
                },
                "rtp-recv-bytes": {
                  type: "unsigned_long",
                  index: false,
                },
                "rtp-recv-gaps": {
                  type: "unsigned_long",
                  index: false,
                },
                "rtp-recv-lost-percentage": {
                  type: "float",
                  index: false,
                },
                "rtp-recv-jitter": {
                  type: "float",
                  index: false,
                },
                "rtp-send-packets": {
                  type: "unsigned_long",
                  index: false,
                },
                "rtp-send-bytes": {
                  type: "unsigned_long",
                  index: false,
                },
                sbc: {
                  type: "keyword",
                },
                type: {
                  type: "keyword",
                },
              },
            },
          }
}')

STATUS_MAPPINGS_JSON=$(jq -n '
{
  "dynamic_templates": [
    {
      "attrs_template": {
        "path_match": "attrs.vars.*",
        "mapping": {
          "type": "{dynamic_type}",
          "index": true
        }
      }
    },
    { "strings_template": { "match_mapping_type": "string", "mapping": { "type": "keyword", "ignore_above": 256, "doc_values": true } } }
  ],
  "properties": {
            "@timestamp": {
              type: "date",
              index: true,
            },
            host: {
              type: "keyword",
              index: false,
            },
            metric: {
              type: "keyword",
              index: true,
            },
            counter: {
              type: "float",
              index: false,
            },
            gauge: {
              type: "float",
              index: false,
            },
            aggregated_histogram: {
              properties: {
                count: { type: "long" },
                sum: { type: "double" },
                buckets: {
                  type: "nested",
                  properties: {
                    count: { type: "long" },
                    upper_limit: { type: "keyword" },
                  },
                },
              },
            },
            "@version": {
              type: "float",
              index: false,
            },
            tags: {
              properties: {
                type: {
                  type: "keyword",
                  index: false,
                },
                component_kind: {
                  type: "keyword",
                  index: false,
                },
                component_type: {
                  type: "keyword",
                  index: false,
                },
                component_id: {
                  type: "keyword",
                  index: true,
                },
                endpoint: {
                  type: "keyword",
                  index: false,
                },
                status: {
                  type: "keyword",
                  index: false,
                },
              },
            },
          }
}')

get_index_size() {
  local idx="$1"
  call_es GET "${idx}/_stats/store?human=true" | jq -r '._all.total.store.size'
}

fetch_cluster_deprecations() {
  log_info "Fetching deprecations from Elasticsearch..."
  DEPRECATIONS=$(call_es GET "_migration/deprecations")

  if [ -z "$DEPRECATIONS" ]; then
    log_error "No response from Elasticsearch. Is it running at $ES_URL?"
    prompt_checkpoint "Please confirm and hit enter"
  fi

  log_info "Deprecations fetched successfully."

  CLUSTER_DEPRECATIONS=$(echo "$DEPRECATIONS" | jq -c '
    [
      to_entries[] as $section |
      if ($section.value | type) == "object" then
        ($section.value | to_entries[] |
          . as $entry |
          ($entry.value[]? | . + { section: $section.key, name: $entry.key }))
      elif ($section.value | type) == "array" then
        ($section.value[]? | . + { section: $section.key })
      else empty end
    ]
  ')

  if [ "$(echo "$CLUSTER_DEPRECATIONS" | jq -r 'if type=="array" then length else 0 end')" -eq 0 ]; then
    log_success "No cluster deprecations found. Skipping..."
    return 0
  fi

  CLUSTER_DEPRECATIONS_LEVEL_COUNTS=$(echo "$CLUSTER_DEPRECATIONS" | jq -r '
    group_by(.level) | map({(.[0].level): length}) | add
  ')

  log_warn "Node settings deprecations by level:"
  {
    echo "$CLUSTER_DEPRECATIONS_LEVEL_COUNTS" | jq -r 'to_entries[] | [.key, .value] | @tsv'
  } | print_table "Level\tCount" "15:auto"

  log_warn "Node settings deprecation details:"
  {
    echo "$CLUSTER_DEPRECATIONS" | jq -r 'to_entries[] | [(.key + 1 | tostring), .value.level, .value.section, .value.name, .value.message, .value.url] | @tsv'
  } | print_table "#\tLevel\tSection\tName\tMessage\tDetails" "3:9:15:40:60:auto"
}

fix_deprecated_slow_log_setting() {
  local SEARCH_PATTERNS="Setting \\[index\\.(indexing|search)\\.slowlog\\.level\\] is deprecated"

  if echo "$CLUSTER_DEPRECATIONS" | jq -r '.[].message?' | grep -Eq "$SEARCH_PATTERNS"; then
    log_warn "Detected deprecations for slowlog level settings (index.indexing.slowlog.level or index.search.slowlog.level)."
  else
    log_success "No slowlog level deprecations found."
    return 0
  fi

  log_info "Fixing this deprecated settings"
  local RESP=$(call_es PUT "_settings" "{\"settings\":{\"index.indexing.slowlog.level\": null,\"index.search.slowlog.level\": null}}")

  if echo "$RESP" | jq -e '.acknowledged == true' >/dev/null; then
    log_success "Index setting applied successfully."
  else
    log_error "Failed to apply setting setting."
  fi

  RESP=$(call_es GET "_settings")

  if echo "$RESP" | jq -e 'to_entries | any(.value.settings.index.indexing.slowlog.level != null or .value.settings.index.search.slowlog.level != null)' >/dev/null; then
    log_error "Some indices still have slowlog level settings defined."
    prompt_checkpoint "Click confirm to retry applying the settings."
    # recursion call
    fetch_cluster_deprecations
    fix_deprecated_slow_log_setting
  else
    log_success "Verification: Slowlog settings successfully removed from all indices."
  fi

}

fix_disk_water_mark_deprecation() {
  local SEARCH_TEXT="Disk watermarks do not treat single-node clusters differently"

  if ! echo "$CLUSTER_DEPRECATIONS" | jq -r '.[].message?' | grep -qF "$SEARCH_TEXT"; then
    log_success "No disk watermark deprecation."
    return 0
  fi

  log_warn "Deprecation found: Disk watermark single-node behavior change."
  log_warn "The setting 'cluster.routing.allocation.disk.watermark.enable_for_single_data_node' cannot be updated dynamically through the Cluster Settings API."
  log_warn "It must be manually configured in the elasticsearch.yml file and requires a node restart to take effect."

  prompt_checkpoint "Please acknowledge that you understand this limitation and press Enter to continue..."
}

es_8_upgrade_checkpoint() {
  ES_VERSION=$(es_version)

  if ! echo "$ES_VERSION" | grep -Eq '^8(\.|$)'; then
    log_error "Not on Elasticsearch 8.x — current version is $ES_VERSION"
    log_warn "Please upgrade Elasticsearch to 8.x before running this migration."
    log_warn "This script expects an Elasticsearch 8.x cluster reachable at ${ES_URL}."
    log_warn "If you manage Elasticsearch with Docker, a container or orchestration step is required outside this script."
    prompt_checkpoint "Make sure you've upgraded to Elasticsearch 8.x and press Enter to retry..."
    es_8_upgrade_checkpoint
  fi
}

default_pipeline() {
  local pipeline_name="default_pipeline"

  log_info "Creating ingest pipeline ${pipeline_name} for logstashh reindex to calls datastream"

  local resp=$(call_es PUT "_ingest/pipeline/$pipeline_name" @- <<'EOF'
{
  "description": "Clean and normalize documents with native processors for logstashh reindex to calls datastream",
  "processors": [
    {
      "set": {
        "if": "ctx.host instanceof Map",
        "field": "host",
        "value": "{{host.name}}"
      }
    },
    {
      "set": {
        "if": "ctx['@timestamp'] == null",
        "field": "@timestamp",
        "value": "{{_ingest.timestamp}}"
      }
    },
    {
      "script": {
        "if": "ctx.month instanceof String",
        "lang": "painless",
        "source": "
          Map m = ['january':1,'february':2,'march':3,'april':4,'may':5,'june':6,'july':7,'august':8,'september':9,'october':10,'november':11,'december':12];
          String s = ctx.month.toLowerCase();
          if (m.containsKey(s)) {
            ctx.month = m[s];
          } else {
            ctx.remove('month');
          }
        "
      }
    },
    {
      "json": {
        "field": "attrs.rtp-stats-a",
        "target_field": "attrs.rtp-stats-a",
        "if": "ctx.attrs instanceof Map && ctx.attrs.containsKey('rtp-stats-a')",
        "ignore_failure": true
      }
    },
    {
      "json": {
        "field": "attrs.rtp-stats-b",
        "target_field": "attrs.rtp-stats-b",
        "if": "ctx.attrs instanceof Map && ctx.attrs.containsKey('rtp-stats-b')",
        "ignore_failure": true
      }
    }
  ]
}
EOF
  )

  if echo "$resp" | jq -e '.acknowledged == true' >/dev/null; then
    log_success "Pipeline ${pipeline_name} created successfully"
  else
    log_error "Failed to create pipeline ${pipeline_name}"
    log_error "$resp"
    exit 1
  fi
}

delete_last_created_backing_index() {
  local destination=$1

  last_index=$(call_es GET "_data_stream/${destination}" | jq -r '.data_streams[0].indices[-1].index_name')
  log_info "Deleting last datastream index ${YELLOW}${last_index}${RESET} for ${GREEN}${destination}${RESET} before exit"
  resp=$(call_es POST "${destination}/_rollover")
  resp=$(call_es DELETE $last_index)

  if echo "$resp" | jq -e '.acknowledged == true' >/dev/null; then
    log_success "datastream index ${YELLOW}${last_index}${RESET} was deleted successfully"
  else
    log_warn "Failed to delete ${YELLOW}${last_index}${RESET}"
    log_error $resp
  fi
}

calc_progress_bar() {
  local task=$1
  id=$(echo "$task" | jq -r '.id // .task.id // ""')
  # support description in either the object or nested under .task
  description=$(echo "$task" | jq -r '.description // .task.description // ""' 2>/dev/null || echo "")
  # collapse whitespace and truncate for display
  description=$(printf '%s' "$description" | tr '\n' ' ' | sed -E 's/[[:space:]]+/ /g' | awk '{ if (length($0) > 60) {print substr($0,1,57) "..."} else print $0 }')
  run_ns=$(echo "$task" | jq -r '.running_time_in_nanos // 0')
  created=$(echo "$task" | jq -r '.status.created // 0')
  total=$(echo "$task" | jq -r '.status.total // 1')
  percent_num=$(awk -v t=$total "BEGIN { if(t>0) print ($created / t) * 100; else print 0 }")
  percent=$(awk "BEGIN { printf \"%.2f\", $percent_num }")
  elapsed_sec=$(awk "BEGIN { print $run_ns / 1000000000 }")
  speed=$(awk -v e=$elapsed_sec "BEGIN { if(e>0) printf \"%.2f\", $created / e; else printf \"0\" }")
  eta_sec=$(awk -v p=$percent_num "BEGIN { if(p > 0) printf \"%.0f\", ($elapsed_sec * (100 - p)) / p; else printf \"--\" }")
  filled=$(awk "BEGIN { printf \"%d\", $percent_num / 2 }")
  empty=$((50 - filled))
  progress_bar=$(printf "%s%s" "$(awk -v n="$filled" 'BEGIN{while(n--) printf "#"}')" "$(awk -v n="$empty" 'BEGIN{while(n--) printf "-"}')")

  printf "${CYAN}[PRGS]    %s: %s | ${RESET}[%s] ${YELLOW}%5.1f%%${RESET} | ${GREEN}%s${RESET} docs | ${PURPLE}%s${RESET} docs/s | ETA: ${BLUE}%ss${RESET}\n" \
    "$id" "$description" "$progress_bar" "$percent" "$created" "$speed" "$eta_sec"
}

# Temporarily set index settings to favor fast indexing (disable refresh, set replicas to 0)
# Returns: "<backing_index>|<orig_refresh>|<orig_replicas>" or empty string if none
set_index_perf_mode() {
  local ds="$1"
  last_index=$(call_es GET "_data_stream/$ds" | jq -r '.data_streams[0].indices[-1].index_name // empty') || true
  if [ -z "$last_index" ]; then
    echo ""
    return 0
  fi

  # Capture original settings
  orig_refresh=$(call_es GET "$last_index/_settings" | jq -r '.[].settings.index.refresh_interval // "1s"') || orig_refresh="1s"
  orig_replicas=$(call_es GET "$last_index/_settings" | jq -r '.[].settings.index.number_of_replicas // "1"') || orig_replicas="1"

  # log_info "Temporarily setting refresh_interval=-1 and number_of_replicas=0 on ${YELLOW}${last_index}${RESET}"
  call_es PUT "$last_index/_settings" '{"index":{"refresh_interval":"-1","number_of_replicas":"0"}}' >/dev/null || true

  payload="$last_index|$orig_refresh|$orig_replicas"
  # Store payload in global list so the trap can restore if script is interrupted
  if [ -n "$payload" ]; then
    if [ -z "$RESTORE_PAYLOADS" ]; then
      RESTORE_PAYLOADS="$payload"
    else
      RESTORE_PAYLOADS="${RESTORE_PAYLOADS}
$payload"
    fi
  fi

  echo "$payload"
}

# Ensure we restore any index settings if the script exits unexpectedly
on_exit() {
  if [ -n "$RESTORE_PAYLOADS" ]; then
    log_warn "Script exiting — attempting to restore index settings for modified indices..."
    echo "$RESTORE_PAYLOADS" | while IFS= read -r pl; do
      [ -n "$pl" ] || continue
      restore_index_settings "$pl" || true
    done
  fi
}
trap 'on_exit' EXIT INT TERM

# Restore index settings after reindex. Expects argument in the format returned by set_index_perf_mode
restore_index_settings() {
  local payload="$1"
  [ -z "$payload" ] && return 0
  last_index=$(echo "$payload" | awk -F'|' '{print $1}')
  orig_refresh=$(echo "$payload" | awk -F'|' '{print $2}')
  orig_replicas=$(echo "$payload" | awk -F'|' '{print $3}')

  log_info "Restoring settings on ${YELLOW}${last_index}${RESET} -> refresh_interval=${orig_refresh}, number_of_replicas=${orig_replicas}"
  call_es PUT "$last_index/_settings" "{\"index\":{\"refresh_interval\":\"$orig_refresh\",\"number_of_replicas\":\"$orig_replicas\"}}" >/dev/null || true
}

poll_task_progress() {
  parent_task_id=$1

  log_info "Reindex started with parent task ID: $parent_task_id"

  resp=$(call_es GET "_tasks/$parent_task_id")
  log_info "Waiting $(echo "$resp" | jq -r '.task.description') to complete..."

  start_ms=$(echo "$resp" | jq -r '.task.start_time_in_millis // now * 1000')
  start_time=$(date -d @"$(expr "$start_ms" / 1000)" "+%H:%M:%S")
  log_info "Started at: $start_time"

  previous_lines=0

  while :; do
    resp=$(curl --silent --show-error -X GET "$ES_URL/_tasks/$parent_task_id")
    completed=$(echo $resp | jq -r '.completed')

    [ "$completed" = true ] && break

    status=$(curl --silent --show-error -X GET "$ES_URL/_tasks?parent_task_id=$parent_task_id&detailed=true&group_by=parents")
    num_slices=$(echo "$status" | jq '.tasks | length')

    if [ "$num_slices" -le 0 ]; then
      sleep 1
      log_info "Waiting task to start"
      continue
    fi

    # Build all progress lines first (no printing inside)
    progress_lines=$(
      echo "$status" | jq -r '.tasks | to_entries[] | .key' | while IFS= read -r task_id; do
        task=$(echo "$status" | jq -r --arg id "$task_id" '.tasks[$id]')
        calc_progress_bar "$task"
      done
    )

    overall=$(calc_progress_bar "$(echo $resp | jq -r '.task')")
    progress_lines=$(printf "%s\n%s" "$progress_lines" "$overall")

    # Move cursor up to the start of the previous block and clear each line
    # This ensures we don't leave stale lines when the new progress block
    # contains fewer lines than the previous one.
    if [ "$previous_lines" -gt 0 ]; then
      tput cuu $previous_lines
      i=0
      while [ $i -lt "$previous_lines" ]; do
        tput el      # erase to end of line
        tput cud1    # move cursor down 1 line
        i=$((i+1))
      done
      # Move back to top of the cleared area so new output prints there
      tput cuu $previous_lines
    fi

    # Print current progress lines
    printf "%s\n" "$progress_lines"
    previous_lines=$(echo "$progress_lines" | wc -l)

    # Poll less frequently to reduce load on the cluster
    sleep 2
  done

  end_time=$(date +%s)
  duration=$((end_time - (start_ms / 1000)))
  echo ""
  log_success "Ended at: $(date '+%H:%M:%S'), Duration: ${duration}s"
}

reindex_old_indices() {
  log_info "The reindex operation transforms an index into a new, compatible one. It will copy all of the existing documents into a new index and remove the old one. Depending on size and resources, reindexing may take extended time and your data will be in a read-only state until the job has completed."
  prompt_checkpoint "During reindex, script will try to fix broken docs, otherwise they will be dropped"
  ALL_INDEX_STATS=""

  while IFS=";" read -r source destination pipeline mapping_var; do
    # Print effective tuning configuration for the run
    log_info "Reindex tuning: AGGRESSIVE_REINDEX=${AGGRESSIVE_REINDEX:-0}, REINDEX_RPS=${REINDEX_RPS:-}, SKIP_PIPELINE=${SKIP_PIPELINE:-0}, REINDEX_SLICES_NUM=${REINDEX_SLICES_NUM:-}, REINDEX_BATCH_SIZE=${REINDEX_BATCH_SIZE:-}, REINDEX_BATCH_SIZE_MAX=${REINDEX_BATCH_SIZE_MAX:-}"

    # GET INDICES BY PREFIX ------------------------ #
    indices=$(call_es GET "_cat/indices?format=json&index=${source}&s=index&h=index,store.size&bytes=b")
    indices_count=$(echo $indices | jq 'length')

    if [ $indices_count -lt 1 ]; then
      log_warn "No matching indices found for pattern ${source}. Skipping...";
      continue;
    fi

    if ! [ "$destination" = "status" -o "$destination" = "lastlog" ]; then
      if ! call_es GET "_data_stream" | jq -e --arg ds "$destination" '[.data_streams[]?.name | select(. == $ds)] | length > 0' > /dev/null; then
        log_info "Copy template and settings from $(echo $indices | jq -r .[0].index) to $destination"
        local template_name="${destination}-template"

        log_info "Creating index template '$template_name' for data stream '$destination' ..."
        # Prepare mappings for template. If a mapping_var is provided as the
        # fourth field in the `migrates` entry, attempt to read that variable
        # (it should contain a JSON object) and merge it into the base
        # $MAPPINGS_JSON. Otherwise use the default mappings.
        if [ -n "${mapping_var:-}" ]; then
          # Retrieve the named variable's content (may contain newlines).
          # Use eval to expand the variable by name in a POSIX-compatible way.
          eval "extra_json=\${$mapping_var:-}" || extra_json='{}'
          if ! echo "$extra_json" | jq empty >/dev/null 2>&1; then
            log_warn "Mapping variable ${mapping_var} is invalid JSON; falling back to empty object"
            extra_json='{}'
          fi
          MAPPINGS_FOR_TEMPLATE=$(jq -n --argjson base "$MAPPINGS_JSON" --argjson extra "$extra_json" '
            $base
            | .dynamic_templates = ((($base.dynamic_templates // []) + ($extra.dynamic_templates // [])))
            | .properties = ($base.properties // {}) + ($extra.properties // {})
          ')
        else
          MAPPINGS_FOR_TEMPLATE="$MAPPINGS_JSON"
        fi
        # If running a single-node cluster we should set replicas to 0 in the template
        if [ "${SINGLE_NODE:-0}" -eq 1 ]; then
          if [ "${AGGRESSIVE_REINDEX:-0}" -eq 1 ]; then
            # Aggressive: drop analyzers (use keyword analyzer) in template to speed indexing
            local body=$(jq -nc \
              --arg pattern "${destination}*" \
              --argjson mappings "$MAPPINGS_FOR_TEMPLATE" \
              '{
                "index_patterns": [$pattern],
                "data_stream": {},
                "template": {
                  "settings": { "index": { "number_of_replicas": "0" } , "analysis": { "analyzer": { "default": { "type": "keyword" } } } },
                  "mappings": $mappings
                }
              }')
          else
            local body=$(jq -nc \
              --arg pattern "${destination}*" \
              --argjson mappings "$MAPPINGS_FOR_TEMPLATE" \
              '{
                "index_patterns": [$pattern],
                "data_stream": {},
                "template": {
                  "settings": { "index": { "number_of_replicas": "0" } },
                  "mappings": $mappings
                }
              }')
          fi
        else
          if [ "${AGGRESSIVE_REINDEX:-0}" -eq 1 ]; then
            local body=$(jq -nc \
              --arg pattern "${destination}*" \
              --argjson mappings "$MAPPINGS_FOR_TEMPLATE" \
              '{
                "index_patterns": [$pattern],
                "data_stream": {},
                "template": {
                  "settings": { "index": { "number_of_replicas": "0" }, "analysis": { "analyzer": { "default": { "type": "keyword" } } } },
                  "mappings": $mappings
                }
              }')
          else
            local body=$(jq -nc \
              --arg pattern "${destination}*" \
              --argjson mappings "$MAPPINGS_FOR_TEMPLATE" \
              '{
                "index_patterns": [$pattern],
                "data_stream": {},
                "template": {
                  "settings": { "index": { "number_of_replicas": "0" } },
                  "mappings": $mappings
                }
              }')
          fi
        fi
        # If a mapping_var was specified for this migrates entry, ensure the
        # analyzer used by common mappings exists in the template settings
        # (adds analyzerWhiteSpaceOnly). This mirrors the `calls` behavior but
        # is driven by the fourth field of the migrates entry.
        if [ -n "${mapping_var:-}" ]; then
          analysis_json='{"analyzer":{"analyzerWhiteSpaceOnly":{"type":"custom","tokenizer":"whitespace"}}}'
          body=$(printf '%s' "$body" | jq --argjson analysis "$analysis_json" '.template.settings.index.analysis = ((.template.settings.index.analysis // {}) + $analysis)')
        fi

        resp=$(call_es PUT "_index_template/${template_name}" "$body")
        if echo "$resp" | jq -e '.acknowledged == true' >/dev/null; then
          log_success "'${template_name}' was created successfully"
        else
          log_error "Failed to create template '${template_name}'"
          log_error $resp
          exit 1
        fi

        log_info "Create new data stream $destination"
        local resp=$(call_es PUT "_data_stream/$destination")

        if echo "$resp" | jq -e '.acknowledged == true' >/dev/null; then
          log_success "data stream: ${YELLOW}'${destination}'${RESET} was created successfully"
        else
          log_error "Failed to create data stream '${destination}'"
          log_error $resp
          exit 1
        fi
      fi

      if [ -n "$pipeline" ] && type "$pipeline" >/dev/null 2>&1; then
        # Create the ingest pipeline. Call the function dynamically
          "$pipeline"
      fi
    fi

    for index in $(echo $indices | jq -c '.[]'); do
      local index_name=$(echo $index | jq -r '.index')

      if [ $destination = "status" -o $destination = "lastlog" ]; then
        log_info "Deleting ${index_name} index. Old status data is not needed anymore."
        local resp=$(call_es DELETE $index_name)

        if echo "$resp" | jq -e '.acknowledged == true' >/dev/null; then
          log_success "Index ${YELLOW}${index_name}${RESET} was deleted successfully"
        else
          log_warn "Failed to delete index ${YELLOW}${index_name}${RESET}"
          log_error $resp
        fi

        continue;
      fi

      local version=$(call_es GET $index_name | jq -r '.[].settings.index.version.created // "unknown"')
      log_info "Reindexing $index_name. Current index version: $version"

      if [ "$version" = "unknown" ]; then
        log_warn "Skipping ${index}: could not detect version."
        continue
      fi

      if [ "$version" -ge 8000000 ]; then
        log_warn "${index} is already v${version}, skipping."
        continue
      fi

      check_reindex_free_disk_safety $(echo $index | jq -r '.["store.size"]')
      log_info "Reindexing source: $index_name -> dest: ${destination}"

      docs_before=$(call_es GET "$index_name/_count" "" | jq -r '.count')
      size_before=$(get_index_size "$index_name")

      # Temporarily set index to high-throughput mode (disable refresh, replicas=0)
      perf_payload=$(set_index_perf_mode "$destination") || perf_payload=""

      # Ensure pipeline arg and batch_size have sane defaults so we always send a request body
      if [ "${SKIP_PIPELINE:-0}" -eq 1 ]; then
        PIPELINE_ARG=""
      else
        PIPELINE_ARG="$pipeline"
      fi

      # Ensure REINDEX_BATCH_SIZE is numeric; fallback to default
      if ! echo "${REINDEX_BATCH_SIZE}" | grep -Eq '^[0-9]+'; then
        REINDEX_BATCH_SIZE=10000
      fi

      batch_size=${REINDEX_BATCH_SIZE}

      # Apply global hard cap for batch size if configured
      if ! echo "${REINDEX_BATCH_SIZE_MAX}" | grep -Eq '^[0-9]+'; then
        REINDEX_BATCH_SIZE_MAX=100000
      fi
      if [ "$batch_size" -gt "$REINDEX_BATCH_SIZE_MAX" ]; then
        log_warn "Capping batch_size ${batch_size} to REINDEX_BATCH_SIZE_MAX=${REINDEX_BATCH_SIZE_MAX}"
        batch_size=$REINDEX_BATCH_SIZE_MAX
      fi

      # --- Dynamic slices calculation per-index ---
      # Compute a sensible slices count based on index primary shards and multiplier.
      # Falls back to configured REINDEX_SLICES_NUM when REINDEX_SLICES_DYNAMIC is 0.
      if [ "${REINDEX_SLICES_DYNAMIC:-1}" -eq 1 ]; then
        shards=$(call_es GET "$index_name/_settings" | jq -r '.[].settings.index.number_of_shards // 1' 2>/dev/null || echo 1)
        # ensure numeric
        if ! echo "$shards" | grep -Eq '^[0-9]+'; then shards=1; fi
        slices_calc=$((shards * REINDEX_SLICES_MULTIPLIER))
        if [ "$slices_calc" -lt 1 ]; then slices_calc=1; fi
        REINDEX_SLICES_USED=$slices_calc
      else
        REINDEX_SLICES_USED=$REINDEX_SLICES_NUM
      fi

      # Compute per-slice scroll size (size param). Keep per-slice bulks reasonable
      # so high slice counts don't produce huge concurrent bulk sizes.
      per_slice_size=$batch_size
      if [ "$REINDEX_SLICES_USED" -gt 1 ]; then
        # integer division, ensure at least 100
        per_slice_size=$((batch_size / REINDEX_SLICES_USED))
        if [ "$per_slice_size" -lt 100 ]; then per_slice_size=100; fi
      fi

      log_info "Using slices=${REINDEX_SLICES_USED}, per_slice_size=${per_slice_size} (global batch_size=${batch_size})"

      resp=$(call_es POST "_reindex?slices=${REINDEX_SLICES_USED}&wait_for_completion=false&requests_per_second=${REINDEX_RPS}" "$(jq -n \
        --arg src "$index_name" \
        --arg dest "$destination" \
        --arg pipeline "$PIPELINE_ARG" \
        --argjson batch_size ${per_slice_size} \
        '{
          source: { index: $src, size: $batch_size },
          dest: (
            {
              index: $dest,
              op_type: "create"
            } + if $pipeline != "" then { pipeline: $pipeline } else {} end
          ),
          conflicts: "proceed"
        }')"
      )

      if ! echo "$resp" | jq -e '.task' >/dev/null; then
        log_error "Failed start reindexing for ${index_name}"
        log_error "$resp"
        delete_last_created_backing_index $destination
        exit 1
      fi

      local task_id=$(echo "$resp" | jq -r '.task')

      poll_task_progress $task_id

      # Restore index settings to their original values (refresh interval & replicas)
      if [ -n "${perf_payload}" ]; then
        restore_index_settings "${perf_payload}" || true
      fi

      local reps=$(call_es GET "_tasks/$task_id")
      failures_count=$(echo "$resp" | jq '.response.failures | length')
      failures=$(echo "$resp" | jq -r --arg red "$RED" --arg reset "$RESET" '
          .response.failures[]? |
          [
            ((if .cause.caused_by.type then .cause.caused_by.type else .cause.type end)),
            (.cause.reason + (if .cause.caused_by.reason then " -> " + .cause.caused_by.reason else "" end))
          ] |
          @tsv |
          "\($red)\(.)\($reset)"
      ')

      if [ "$failures_count" -gt 0 ]; then
          log_error "Reindex task $task_id has $failures_count errors. Exiting."
          {
            echo "$failures"
          } | print_table "Type\tError" "34:auto"
          delete_last_created_backing_index $destination
          exit 1
      fi

      # Ensure the newly created backing index uses 0 replicas (user requested)
      last_index=$(call_es GET "_data_stream/${destination}" | jq -r '.data_streams[0].indices[-1].index_name')
      if [ -n "$last_index" ]; then
        log_info "Setting number_of_replicas=0 on ${last_index}"
        call_es PUT "${last_index}/_settings" '{"index":{"number_of_replicas":"0"}}' >/dev/null || true
      fi

      log_success "Successfully reindexed ${index_name} → ${destination}"
      docs_after=$(call_es GET "${last_index}/_count" "" | jq -r '.count')
      size_after=$(get_index_size "$destination")

      INDEX_STATS="${index_name}\t${destination}\t${docs_before}\t${docs_after}\t$((docs_before - docs_after))\t${size_before}\t${size_after}"

      log_info "Deleting old index ${index_name}"
      local resp=$(call_es DELETE $index_name | jq .)

      if echo "$resp" | jq -e '.acknowledged == true' >/dev/null; then
        log_success "'${index_name}' was deleted successfully"
      else
        log_warn "Failed to delete '${index_name}'"
      fi

      log_info "Creating new backing index for upcoming events"
      resp=$(call_es POST "${destination}/_rollover")

      {
        echo $INDEX_STATS
      } | print_table "Old Index\tNew Index\tDocs Before\tDocs After\tDropped Docs\tSize Before\tSize After" "50:50:15:15:15:15:auto"

      ALL_INDEX_STATS="${ALL_INDEX_STATS}\n${INDEX_STATS}"
    done

    log_info "Found ${indices_count} indices to migrate"
  done <<EOF
$migrates
EOF

  if [ -n "$(printf "%s" "$ALL_INDEX_STATS" | tr -d '[:space:]')" ]; then
    log_success "Reindexing and cleanup complete."
    {
      echo $ALL_INDEX_STATS
    } | print_table "Old Index\tNew Index\tDocs Before\tDocs After\tDropped Docs\tSize Before\tSize After" "50:50:15:15:15:15:auto"
  fi
}

reindex_legacy_backing_indices() {
  log_info "Reindex all legacy backing indices for a data stream. This operation occurs in a persistent task. The persistent task ID is returned immediately and the reindexing work is completed in that task."
  local resp=$(call_es GET "_migration/deprecations" | jq -r '.data_streams | keys[]')

  for ds in $resp; do
    log_info "Reindexing legacy backing indices for data stream: ${YELLOW}$ds${RESET}"

    log_info "Calling _migration/reindex can take some time to respond,...it is not stuck"
    local resp=$(call_es POST "_migration/reindex" \
      "{\"source\":{\"index\":\"${ds}\"},\"mode\":\"upgrade\"}"
    )

    if echo "$resp" | jq -e '.acknowledged == true' >/dev/null 2>&1; then
      log_success "Successfully acknowledged reindex for $ds"
    else
      log_error "Reindex for ${YELLOW}$ds${RESET} not acknowledged or failed"
      log_error "$resp"
      exit 1
    fi
  done
}

reindex_legacy_indexes(){
  # Migrate the rest of the indices with legacy settings
  local index_settings=$(call_es GET "_migration/deprecations" | jq -r '.index_settings | keys[]')
  for index in $index_settings; do
    log_info "Reindexing legacy index settings for index: ${YELLOW}$index${RESET}"

    # Try calling the migration API for this index. If it fails because the
    # target is not a data stream, fall back to plain index reindexing.
    payload=$(jq -n --arg ind "$index" '{source: { index: $ind }, mode: "upgrade"}')
    log_info "Attempting _migration/reindex for ${index}"
    resp=$(curl --silent --show-error -X POST "$ES_URL/_migration/reindex" -H "Content-Type: application/json" -d "$payload" 2>&1 || true)

    if echo "$resp" | jq -e 'has("error")' >/dev/null 2>&1; then
      err_type=$(echo "$resp" | jq -r '.error.type // ""')
      err_reason=$(echo "$resp" | jq -r '.error.reason // ""')
      # If migration API indicates this is not a data stream, fall back
      if [ "$err_type" = "resource_not_found_exception" ] && echo "$err_reason" | grep -q "Data stream named"; then
        log_info "${index} is not a data stream; falling back to plain index reindex"
        reindex_unmanaged_indices "$index"
        STEP=$((STEP + 1))
        progress "$STEP" "$TOTAL_STEPS"
      else
        log_error "_migration/reindex failed for ${index}:"
        log_error "$resp"
        exit 1
      fi
    else
      if echo "$resp" | jq -e '.acknowledged == true' >/dev/null 2>&1; then
        log_success "Successfully acknowledged reindex for $index"
      else
        log_error "Reindex for ${YELLOW}$index${RESET} not acknowledged or failed"
        log_error "$resp"
        exit 1
      fi
    fi
  done
}

reindex_unmanaged_indices() {
  # If an index name is passed as the first arg, reindex only that index.
  local target_index="$1"
  if [ -n "$target_index" ]; then
    log_info "Reindexing single unmanaged index: ${target_index} (no pipeline will be used)"
    # fetch index info for the single target
    all_indices_json=$(call_es GET "_cat/indices?format=json&index=${target_index}&h=index,store.size&bytes=b")
  else
    log_info "Reindexing indices not covered by migrates (no pipeline will be used)"
    # Get all non-system indices with their sizes
    all_indices_json=$(call_es GET "_cat/indices?format=json&h=index,store.size&bytes=b")
  fi

  for row in $(echo "$all_indices_json" | jq -c '.[]'); do
    index_name=$(echo "$row" | jq -r '.index')
    index_size=$(echo "$row" | jq -r '."store.size"')

    # Skip hidden/system indices
    case "$index_name" in
      \.*) log_info "Skipping system/hidden index ${index_name}"; continue ;;
    esac

    # Check if index matches any 'source' pattern from $migrates
    is_managed=0
    while IFS=";" read -r src dst pl || [ -n "$src" ]; do
      case "$index_name" in
        $src) is_managed=1; break ;;
      esac
    done <<EOF
$migrates
EOF

    if [ "$is_managed" -eq 1 ]; then
      # If the caller requested a specific target index, do not skip even if it's in migrates
      if [ -n "$target_index" ] && [ "$target_index" = "$index_name" ]; then
        log_warn "Index ${index_name} is managed by migrates but was requested explicitly — proceeding"
      else
        log_info "Index ${index_name} is managed by migrates, skipping here"
        continue
      fi
    fi

    # Detect index created version
    version=$(call_es GET "$index_name" | jq -r '.[].settings.index.version.created // "unknown"') || version="unknown"
    if [ "$version" = "unknown" ]; then
      log_warn "Skipping ${index_name}: could not detect version"
      continue
    fi
    # TODO: the version check should be dynamic based on ES version
    if [ "$version" -ge 8000000 ]; then
      log_warn "${index_name} is already v${version}, skipping."
      continue
    fi

    # Capture original replicas (to apply later) and ensure enough disk for reindex
    orig_replicas=$(call_es GET "$index_name/_settings" | jq -r '.[].settings.index.number_of_replicas // "1"' 2>/dev/null || echo "1")
    # normalize numeric (some templates use strings)
    if ! echo "$orig_replicas" | grep -Eq '^[0-9]+'; then orig_replicas=1; fi

    check_reindex_free_disk_safety "$index_size"

    # Choose a destination index name different from the source to avoid invalid reindex target
    dest_index="${index_name}-reindexed"
    if [ "$dest_index" = "$index_name" ]; then
      dest_index="${index_name}-reindexed-$(date +%s)"
    fi

    # If the destination exists, pick a unique name (few attempts)
    attempt=0
    while index_exists "$dest_index"; do
      attempt=$((attempt + 1))
      dest_index="${index_name}-reindexed-$(date +%s)-${attempt}"
      if [ $attempt -gt 5 ]; then
        log_error "Unable to find unique destination name for ${index_name}, skipping this index"
        continue 2
      fi
      sleep 1
    done

    log_info "Reindexing unmanaged index ${index_name} -> ${dest_index} (no pipeline)"

    # Create the destination index explicitly and set replicas=0 and refresh_interval=-1 for fast indexing
    create_body=$(jq -n --argjson r 0 '{settings: {index: {number_of_replicas: $r, refresh_interval: "-1"}}}')
    log_info "Creating destination index ${dest_index} with replicas=0 and refresh_interval=-1"
    call_es PUT "$dest_index" "$create_body" >/dev/null || true

    # Compute slices similar to managed reindex logic
    if [ "${REINDEX_SLICES_DYNAMIC:-1}" -eq 1 ]; then
      shards=$(call_es GET "$index_name/_settings" | jq -r '.[].settings.index.number_of_shards // 1' 2>/dev/null || echo 1)
      if ! echo "$shards" | grep -Eq '^[0-9]+'; then shards=1; fi
      slices_calc=$((shards * REINDEX_SLICES_MULTIPLIER))
      if [ "$slices_calc" -lt 1 ]; then slices_calc=1; fi
      REINDEX_SLICES_USED=$slices_calc
    else
      REINDEX_SLICES_USED=$REINDEX_SLICES_NUM
    fi

    per_slice_size=${REINDEX_BATCH_SIZE}
    if [ "$REINDEX_SLICES_USED" -gt 1 ]; then
      per_slice_size=$((REINDEX_BATCH_SIZE / REINDEX_SLICES_USED))
      if [ "$per_slice_size" -lt 100 ]; then per_slice_size=100; fi
    fi

    log_info "Using slices=${REINDEX_SLICES_USED}, per_slice_size=${per_slice_size} (global batch_size=${REINDEX_BATCH_SIZE})"

    resp=$(call_es POST "_reindex?slices=${REINDEX_SLICES_USED}&wait_for_completion=false&requests_per_second=${REINDEX_RPS}" "$(jq -n \
      --arg src "$index_name" \
      --arg dest "$dest_index" \
      --argjson batch_size ${per_slice_size} \
      '{ source: { index: $src, size: $batch_size }, dest: { index: $dest, op_type: "create" }, conflicts: "proceed" }')")

    if ! echo "$resp" | jq -e '.task' >/dev/null; then
      log_error "Failed to start reindexing for ${index_name}"
      log_error "$resp"
      # Attempt to cleanup any partially created destination
      call_es DELETE "$dest_index" >/dev/null 2>&1 || true
      continue
    fi

    task_id=$(echo "$resp" | jq -r '.task')
    poll_task_progress "$task_id"

    # After successful reindex, try to delete old index and add alias from dest to original name
    log_info "Deleting old index ${index_name}"
    del_resp=$(call_es DELETE "$index_name" | jq .)
    if echo "$del_resp" | jq -e '.acknowledged == true' >/dev/null; then
      log_success "Deleted old index ${index_name}"

      # Wait (with timeout) until the index disappears from _cat/indices to reduce a race window
      wait_secs=0
      while index_exists "$index_name"; do
        sleep 1
        wait_secs=$((wait_secs + 1))
        if [ $wait_secs -gt 10 ]; then
          log_warn "Index ${index_name} still appears in cluster after ${wait_secs}s; proceeding to create alias anyway"
          break
        fi
      done

      # Create alias from new index to original name so callers still use same index name
      alias_payload=$(jq -n --arg dest "$dest_index" --arg alias "$index_name" '{ actions: [ { add: { index: $dest, alias: $alias } } ] }')
      call_es POST "_aliases" "$alias_payload" >/dev/null || true
    else
      log_warn "Failed to delete old index ${index_name}, leaving reindexed index ${dest_index} in place"
    fi

    # Ensure the newly created index uses the same number_of_replicas as the original index (as in set_index_perf_mode)
    if [ -n "$orig_replicas" ]; then
      log_info "Setting number_of_replicas=${orig_replicas} on ${dest_index}"
      call_es PUT "${dest_index}/_settings" "$(jq -n --arg r "$orig_replicas" '{index: { number_of_replicas: ($r | tonumber) } }')" >/dev/null || true
    fi

    log_success "Reindexed unmanaged index ${index_name} -> ${dest_index}"
  done
}

es_9_upgrade_checkpoint() {
  log_info "Checking current ES version..."

  ES_VERSION=$(es_version)
  log_info "Current ES version $ES_VERSION"

  if ! echo "$ES_VERSION" | grep -Eq '^9(\.|$)'; then
    log_error "Not on Elasticsearch 9.x — current version is $ES_VERSION"
    prompt_checkpoint "Make sure you've upgraded to Elasticsearch 9.x"
    es_9_upgrade_checkpoint
  fi

  log_success "Elasticsearch v9 is ready"
}

wait_for_all_shards_recovery() {
  log_info "Waiting for all shards of all indices (including data streams) to finish recovery..."

  while :; do
    recovery_json=$(call_es GET "_cat/recovery?format=json")

    total_shards=$(echo "$recovery_json" | jq '. | length')
    done_shards=0
    recovering_shards=0
    total_percent=0

    idx=0
    while [ "$idx" -lt "$total_shards" ]; do
      stage=$(echo "$recovery_json" | jq -r ".[$idx].stage")
      percent=$(echo "$recovery_json" | jq -r ".[$idx].percent")

      # Handle null or empty percent
      if [ "$percent" = "null" ] || [ -z "$percent" ]; then
        percent=0
      else
        # Remove decimal part
        percent=$(echo "$percent" | cut -d'.' -f1)
      fi

      total_percent=$((total_percent + percent))

      if [ "$stage" = "done" ]; then
        done_shards=$((done_shards + 1))
      else
        recovering_shards=$((recovering_shards + 1))
      fi

      idx=$((idx + 1))
    done

    avg_percent=0
    if [ "$total_shards" -gt 0 ]; then
      avg_percent=$((total_percent / total_shards))
    fi

    echo "Recovery status: $done_shards / $total_shards shards done, $recovering_shards recovering, avg progress: $avg_percent%"

    if [ "$recovering_shards" -eq 0 ]; then
      log_success "All shards of all indices are recovered!"
      break
    fi

    sleep 5
  done
}

# ---------------------------------------------- #
# RUN THE SCRIPT                                 #
# ---------------------------------------------- #
main() {
  clear
  printf "${BOLD}ES upgrade script${RESET}\n"
  printf -- "----------------------------------------\n\n"

  TOTAL_STEPS=11
  STEP=0

  progress "$STEP" "$TOTAL_STEPS"

  check_for_required_cli_tools
  STEP=$((STEP + 1))
  progress "$STEP" "$TOTAL_STEPS"

  wait_for_active_shards
  validate_required_disk_space
  STEP=$((STEP + 1))
  progress "$STEP" "$TOTAL_STEPS"

  ES_VERSION=$(es_version)
  log_info "Current ES version $ES_VERSION"

  log_info "Detecting cluster node count..." # Detect cluster node count early and set SINGLE_NODE flag for later logic
  NODE_COUNT=$(call_es GET "_cat/nodes?format=json" | jq 'length' 2>/dev/null || echo 1)
  if [ "$NODE_COUNT" -eq 1 ]; then
    log_info "Single-node cluster detected"
    SINGLE_NODE=1
  else
    log_info "Multi-node cluster detected ($NODE_COUNT nodes)"
    SINGLE_NODE=0
  fi

  if [ "${AGGRESSIVE_REINDEX:-0}" -eq 1 ]; then
    log_warn "AGGRESSIVE_REINDEX=1 -> enabling maximum-throughput settings"
    log_warn " - refresh_interval will be set to -1 during reindex"
    log_warn " - number_of_replicas will be set to 0 during reindex"
    log_warn " - batch size: ${REINDEX_BATCH_SIZE} docs, slices=${REINDEX_SLICES_NUM}, requests_per_second=-1"
    log_warn "For best results use SSD/NVMe storage and ensure sufficient CPU and IO capacity"
  fi

  # ES 7 CHECKS ---------------------------------- #
  if echo "$ES_VERSION" | grep -Eq '^7(\.|$)'; then
    start_the_migration_feature
    STEP=$((STEP + 1))
    progress "$STEP" "$TOTAL_STEPS"

    fetch_cluster_deprecations

    fix_deprecated_slow_log_setting
    STEP=$((STEP + 1))
    progress "$STEP" "$TOTAL_STEPS"

    fix_disk_water_mark_deprecation
    STEP=$((STEP + 1))
    progress "$STEP" "$TOTAL_STEPS"

    log_success "Elasticsearch 7 deprecation migration completed successfully."
  else
    log_info "Skip Elasticsearch 7 checks"
    STEP=$((STEP + 3))
    progress "$STEP" "$TOTAL_STEPS"
  fi

  # Record ES8 migration start if we need to run ES8 steps
  es8_start=0
  if ! echo "$ES_VERSION" | grep -Eq '^9(\.|$)'; then
    es8_start=$(date +%s)
  fi

  log_info "Checking current ES version..."
  ES_VERSION=$(es_version)
  log_info "Current ES version $ES_VERSION"

  # ES 8 CHECKS ---------------------------------- #
  if ! echo "$ES_VERSION" | grep -Eq '^9(\.|$)'; then
    es_8_upgrade_checkpoint

    wait_for_active_shards
    log_success "Elasticsearch v8 is ready"
    STEP=$((STEP + 1))
    progress "$STEP" "$TOTAL_STEPS"

    start_the_migration_feature
    STEP=$((STEP + 1))
    progress "$STEP" "$TOTAL_STEPS"

    fetch_cluster_deprecations

    reindex_legacy_backing_indices
    STEP=$((STEP + 1))
    progress "$STEP" "$TOTAL_STEPS"

    reindex_old_indices
    STEP=$((STEP + 1))
    progress "$STEP" "$TOTAL_STEPS"
    reindex_legacy_indexes

    wait_for_all_shards_recovery
    es8_end=$(date +%s)
    es8_elapsed=$((es8_end - es8_start))
    log_success "ES 7->8 migration elapsed: $(format_seconds $es8_elapsed) (${es8_elapsed}s)"
  else
    log_info "Skip Elasticsearch 8 checks"
    STEP=$((STEP + 4))
    progress "$STEP" "$TOTAL_STEPS"
  fi

  # ES9 upgrade steps
  es9_start=$(date +%s)
  es_9_upgrade_checkpoint
  STEP=$((STEP + 1))
  progress "$STEP" "$TOTAL_STEPS"

  fetch_cluster_deprecations

  reindex_legacy_backing_indices
  STEP=$((STEP + 1))
  progress "$STEP" "$TOTAL_STEPS"
  reindex_legacy_indexes
  wait_for_all_shards_recovery

  es9_end=$(date +%s)
  es9_elapsed=$((es9_end - es9_start))
  log_success "ES 8->9 migration elapsed: $(format_seconds $es9_elapsed) (${es9_elapsed}s)"

  log_success "All tasks completed successfully."
}

{
  main
} 2>&1 | tee runtime.log