#!/bin/bash

# ==============================================================================
# Script Name: obs-cloud-cli.sh
# Description: Public Cloud OTel exporter management in Cloudera Manager
# Dependencies: curl, jq
# ==============================================================================

DEFAULT_CM_BASE_URL="https://localhost:7183"
DEFAULT_API_VERSION="v31"
DEFAULT_SCOPE="cluster"
DEFAULT_SCOPES="cluster host mgmt"

CM_API_STALENESS="v41"
CM_POLL_INTERVAL=5
CM_POLL_TIMEOUT=600

PARAM_EXPORTERS="otelcol_exporters"
PARAM_RECEIVERS="otelcol_receivers"
PARAM_SERVICE="otelcol_service"
PARAM_PROCESSORS="otelcol_processors"
PARAM_EXTENSIONS="otelcol_extensions"
PARAM_SHOULD_COLLECT="otelcol_should_collect"
PARAM_RTM_LOGS_SERVICE="otelcol_rtm_logs_service"

# Signals a command can act on. Default (no --signal) acts on both.
DEFAULT_SIGNALS="metrics logs"

SERVICEMONITOR_ROLE_TYPE="SERVICEMONITOR"
SERVICEMONITOR_RCG="MGMT-SERVICEMONITOR-BASE"

ALLHOSTS_TARGET="allHosts"

DEBUG=false

log() { [[ "$DEBUG" == true ]] && echo "$*"; }

_pager() {
  if command -v less >/dev/null 2>&1; then
    less -R
  elif command -v more >/dev/null 2>&1; then
    more
  else
    cat
  fi
}

usage() {
  local cmd
  cmd=$(basename "$0")
  cat <<EOF | _pager

${cmd}()                                                         ${cmd}()

NAME
       ${cmd} - Manage Public Cloud OTel exporter configuration in CM.

DESCRIPTION
       A command-line tool to manage OpenTelemetry collector exporter and service
       pipeline configuration for Public Cloud clusters in Cloudera Manager.

       Scope model:
         - cluster: roleConfigGroups under a given cluster/service
         - mgmt:    Service Monitor role config group under /cm/service
         - host:    host-level collector config on /cm/allHosts/config

       In cluster scope, omit --role to fan out to all roleConfigGroups, and
       omit --service to fan out to every OTel-supported service in the cluster.
       If --cluster is omitted and the deployment has exactly one cluster, it is
       auto-detected. In mgmt scope only the Service Monitor group is targeted by
       default (use --role to target a specific mgmt group). In host scope the
       single /cm/allHosts/config object is targeted (no --role/--cluster/--service);
       host-level changes take effect without a restart/reload. Discovery via
       list-roles still lists all groups.

       --scope accepts multiple scopes in one command: a comma list
       (--scope cluster,host) or 'all' (cluster+mgmt+host); each scope is applied
       with its own rules. When --scope is omitted, commands apply to BOTH host
       and cluster (default: $DEFAULT_SCOPES).

SYNOPSIS
       ${cmd} [global options] <command> [command options]

GLOBAL OPTIONS
       --url <value>
          Cloudera Manager base URL.
          Default: $DEFAULT_CM_BASE_URL

       --api-version <vNN>
          CM API version used for read/write config endpoints.
          Default: $DEFAULT_API_VERSION

       --scope cluster|mgmt|host|<list>|all
          Configuration scope(s). Single value, comma list (cluster,host), or
          'all' (cluster+mgmt+host).
          Default when omitted: $DEFAULT_SCOPES

       --user <value>
          Workload username for CM API authentication. Required.

       --pass <value>
          Workload password for CM API authentication. Required.

       --cluster <value>
          Cluster name for --scope cluster commands. If omitted and the
          deployment has exactly one cluster, it is auto-detected; with multiple
          clusters it is required. Ignored for --scope mgmt/host.

       --service <value>
          Service name (for example hbase). In --scope cluster, omit it to fan
          out the command to every OTel-supported service in the cluster;
          provide it to target a single service. Ignored for --scope mgmt/host.

       --role <value>
          Single roleConfigGroup to target. In cluster scope, if omitted the
          command fans out to all OTel-supported roleConfigGroups in the
          service (unsupported groups such as Gateway are skipped). Passing an
          unsupported group explicitly returns an error. In mgmt scope the
          Service Monitor group is targeted by default.

       --exporters-key <value>
       --receivers-key <value>
       --service-key <value>
          Override CM property keys.
          Defaults: $PARAM_EXPORTERS, $PARAM_RECEIVERS, $PARAM_SERVICE

       --servicemonitor-role-type <value>
          Role type used to locate the Service Monitor group in mgmt scope.
          Default: $SERVICEMONITOR_ROLE_TYPE

       --servicemonitor-rcg <value>
          Fallback Service Monitor role config group name in mgmt scope.
          Default: $SERVICEMONITOR_RCG

       --signal metrics|logs|both
          Which telemetry signal(s) a command acts on. Definitions live at host in
          the shared otelcol_* params; the signal selects the pipeline section —
          metrics -> otelcol_service, logs -> otelcol_rtm_logs_service.
          Default: both

       --refresh
          After the command's business logic completes, detect stale CM config
          and apply it once: stale cluster services get a no-downtime Deploy
          Client Configs + Refresh; the Management Service, if stale, is
          restarted. Anything FRESH is skipped. Each command is polled to
          completion. Requires full CM administrator rights (see refresh-config).
          Default: false (config is written but not applied).

       --debug
          Enable debug output.

       --help | -h
          Show this help.

COMMANDS
       list-clusters
       list-services
       list-roles
       get-host-configs
       get-metric-configs
       set-should-collect
       list-exporters
       add-exporter
       update-exporter
       remove-exporter
       list-extensions
       add-extension
       update-extension
       remove-extension
       list-processors
       add-processor
       update-processor
       remove-processor
       refresh-config

EXAMPLES
       ${cmd} list-clusters --user workload_user --pass secret

       ${cmd} list-services --cluster Cluster\ 1 --user workload_user --pass secret

       ${cmd} list-roles --cluster Cluster\ 1 --service hbase --user workload_user --pass secret

       ${cmd} get-metric-configs --scope mgmt --user workload_user --pass secret

       # Host-level collector config (/cm/allHosts/config); no restart needed.
       ${cmd} get-metric-configs --scope host --user workload_user --pass secret
       ${cmd} update-exporter --scope host --file ./exporter-update.yaml \\
              --user workload_user --pass secret

       # Fan out to every OTel-supported service in the cluster (omit --service).
       ${cmd} add-exporter --scope cluster --cluster Cluster\ 1 \\
              --file ./exporter-add.yaml --user workload_user --pass secret

       ${cmd} add-exporter --scope cluster --cluster Cluster\ 1 --service hbase \\
              --file ./exporter-add.yaml --user workload_user --pass secret

       ${cmd} add-exporter --scope mgmt --role MGMT-SERVICEMONITOR-BASE \\
              --file ./exporter-add.yaml --user workload_user --pass secret

       ${cmd} update-exporter --scope cluster --cluster Cluster\ 1 --service hbase \\
              --file ./exporter-update.yaml --user workload_user --pass secret

       ${cmd} remove-exporter --scope mgmt --exporter otlphttp/partner \\
              --user workload_user --pass secret

       ${cmd} add-extension --scope cluster --cluster Cluster\ 1 --service hbase \\
              --file ./extension-add.yaml --user workload_user --pass secret

       ${cmd} add-processor --scope cluster --cluster Cluster\ 1 --service hbase \\
              --file ./processor-add.yaml --exporter otlphttp/partner \\
              --user workload_user --pass secret

       ${cmd} update-processor --scope cluster --cluster Cluster\ 1 --service hdfs \\
              --file ./processor-update.yaml --user workload_user --pass secret

       # Apply pending stale config on demand (needs full CM admin rights).
       # Pass --cluster on Public Cloud where the account can't list clusters.
       ${cmd} refresh-config --cluster manvi-pulse-dh --user admin --pass secret

EOF
  exit 0
}

usage_add_exporter() {
  local cmd
  cmd=$(basename "$0")
  cat <<EOF | _pager
NAME
       add-exporter - Add exporter block(s) and pipeline(s).

SYNOPSIS
       ${cmd} add-exporter [scope options] --file <yaml>

INPUT MODES
       --file <yaml>
          Generic exporter YAML. Supports either:
            1) Root exporter blocks (preferred), for example:
               otlphttp/partner:
                 endpoint: "https://otel.partner.example/v1/metrics"
                 headers:
                   api-key: "replace-me"

            2) Full exporters section:
               exporters:
                 otlphttp/partner:
                   endpoint: "https://otel.partner.example/v1/metrics"

NOTES
       - The exporter DEFINITION is always written once to host (otelcol_exporters);
         the single per-host collector shares it across metrics and logs pipelines.
         Any auth.authenticator it references must exist in host otelcol_extensions.
       - Pipelines are then wired per --signal, referencing the host-defined exporter:
           metrics -> otelcol_service. cluster/mgmt: metrics/<exp>-\$ROLE_NAME
             (inherits metrics/\$ROLE_NAME; the \$ROLE_NAME macro keeps it role-unique).
             host: metrics/<exp>-self-metrics and metrics/<exp>-hostmetrics.
           logs -> otelcol_rtm_logs_service on RCGs that already run a logs pipeline:
             logs/<exp>-\$ROLE_NAME inherits the existing logs receivers/processors.
       - Target RCGs are those that already run the base pipeline for the signal
         (metrics/\$ROLE_NAME or logs/\$ROLE_NAME) — no role-type allowlist, so newly
         OTel-enabled groups are picked up automatically.
EOF
  exit 0
}

usage_update_exporter() {
  local cmd
  cmd=$(basename "$0")
  cat <<EOF | _pager
NAME
       update-exporter - Replace existing exporter block(s).

SYNOPSIS
       ${cmd} update-exporter [scope options] --file <yaml>

NOTES
       - Updates exporter YAML blocks only.
       - Service pipelines are left unchanged.
       - All named exporters in input must already exist in each target roleConfigGroup.
EOF
  exit 0
}

usage_remove_exporter() {
  local cmd
  cmd=$(basename "$0")
  cat <<EOF | _pager
NAME
       remove-exporter - Remove exporter block(s) and default pipeline(s).

SYNOPSIS
       ${cmd} remove-exporter [scope options] --exporter <name> [--exporter <name> ...]

NOTES
       - Removes named exporter block from exporters section.
       - Removes default derived pipeline for that exporter from service section.
       - When --role is omitted, command fans out to all roleConfigGroups in scope.
EOF
  exit 0
}

usage_add_extension() {
  local cmd
  cmd=$(basename "$0")
  cat <<EOF | _pager
NAME
       add-extension - Add authenticator/extension block(s).

SYNOPSIS
       ${cmd} add-extension [scope options] --file <yaml>

INPUT MODES
       --file <yaml>
          Generic extensions YAML. Root blocks or a full extensions: section:
            oauth2client/partner:
              client_id: "..."
              client_secret: "..."
              token_url: "https://idp.example/oauth2/token"

NOTES
       - Blocks are appended to each target roleConfigGroup's otelcol_extensions.
       - Section-only: service.extensions is NOT modified (CM enables via defaults).
       - Reference the extension from an exporter's auth.authenticator.
       - When --role is omitted, command fans out to all roleConfigGroups in scope.
EOF
  exit 0
}

usage_update_extension() {
  local cmd
  cmd=$(basename "$0")
  cat <<EOF | _pager
NAME
       update-extension - Replace existing extension block(s).

SYNOPSIS
       ${cmd} update-extension [scope options] --file <yaml>

NOTES
       - Replaces the named extension block(s) in otelcol_extensions.
       - All named extensions in input must already exist in each target RCG.
EOF
  exit 0
}

usage_remove_extension() {
  local cmd
  cmd=$(basename "$0")
  cat <<EOF | _pager
NAME
       remove-extension - Remove extension block(s).

SYNOPSIS
       ${cmd} remove-extension [scope options] --extension <name> [--extension <name> ...]

NOTES
       - Removes named block(s) from otelcol_extensions (section-only).
EOF
  exit 0
}

usage_add_processor() {
  local cmd
  cmd=$(basename "$0")
  cat <<EOF | _pager
NAME
       add-processor - Add processor block(s), optionally link to a pipeline.

SYNOPSIS
       ${cmd} add-processor [scope options] --file <yaml> [--exporter <name> ...]
       ${cmd} add-processor [scope options] --processor <name> --exporter <name> [...]

MODES
       --file <yaml>
          Define processor block(s) in otelcol_processors (no pipeline change).
       --file <yaml> --exporter <name>
          Define AND link each processor into that exporter's pipeline processors: [ ].
       --processor <name> --exporter <name>
          Link an already-defined processor into that exporter's pipeline.

NOTES
       - Linking co-writes otelcol_service. Pipeline must exist (run add-exporter first).
       - When --role is omitted, command fans out to all roleConfigGroups in scope.
EOF
  exit 0
}

usage_update_processor() {
  local cmd
  cmd=$(basename "$0")
  cat <<EOF | _pager
NAME
       update-processor - Replace existing processor block(s).

SYNOPSIS
       ${cmd} update-processor [scope options] --file <yaml>

NOTES
       - Replaces the named processor block(s) in otelcol_processors.
       - Pipelines are left unchanged. Common path for editing filter/\$ROLE_NAME.
EOF
  exit 0
}

usage_remove_processor() {
  local cmd
  cmd=$(basename "$0")
  cat <<EOF | _pager
NAME
       remove-processor - Remove or unlink processor(s).

SYNOPSIS
       ${cmd} remove-processor [scope options] --processor <name> [--processor <name> ...]
       ${cmd} remove-processor [scope options] --processor <name> --exporter <name> [...]

MODES
       --processor <name>
          Delete the definition from otelcol_processors and unlink it from all
          pipelines in the target roleConfigGroup(s).
       --processor <name> --exporter <name>
          Unlink from that exporter's pipeline only (keep the definition).
EOF
  exit 0
}

usage_set_should_collect() {
  local cmd
  cmd=$(basename "$0")
  cat <<EOF | _pager
NAME
       set-should-collect - Enable or disable OTel metric collection for role(s).

SYNOPSIS
       ${cmd} set-should-collect [scope options] --enable
       ${cmd} set-should-collect [scope options] --disable

DESCRIPTION
       Sets the '$PARAM_SHOULD_COLLECT' flag on the target roleConfigGroup(s).
       This flag controls whether Cloudera Manager generates OTel collector
       config for the role. If it is false, exporter/service config written to
       the role is ignored at runtime (no snippet is produced).

OPTIONS
       --enable    Set $PARAM_SHOULD_COLLECT=true.
       --disable   Set $PARAM_SHOULD_COLLECT=false.

NOTES
       - In mgmt scope only the Service Monitor group is targeted (default).
       - On public cloud the Service Monitor default is false, so enable it
         for mgmt metric export to take effect.
       - Combine with --refresh to apply immediately (this flag is not
         refreshable, so it triggers a restart of the affected role).
EOF
  exit 0
}

usage_refresh_config() {
  local cmd
  cmd=$(basename "$0")
  cat <<EOF | _pager
NAME
       refresh-config - Detect stale Cloudera Manager config and apply it.

SYNOPSIS
       ${cmd} refresh-config [--cluster <name>] [--url ...] --user <admin> --pass <secret>

DESCRIPTION
       Reads configuration staleness from CM and applies whatever is pending:

         • Stale cluster services -> a no-downtime "Deploy Client Configuration
           and Refresh" (clusters/{cluster}/commands/deployClientConfigsAndRefresh).
         • The Management Service, if stale -> a restart
           (cm/service/commands/restart).
         • Anything already FRESH is skipped.

       Each submitted command is polled to completion. Same apply step performed
       by --refresh on other commands; run it on demand when you did not pass
       --refresh, or when config went stale by other means. Writes no OTel config.

NOTES
       - --cluster is optional. Given, it checks/applies that cluster directly;
         omitted, it enumerates all clusters CM lists. On Public Cloud the account
         often cannot list clusters, so pass --cluster <name> there. The
         Management Service is always checked either way.
       - Requires full Cloudera Manager administrator rights. A workload_user can
         read/write config but cannot deploy/refresh services or restart mgmt.
EOF
  exit 0
}

show_command_help() {
  case "$1" in
    add-exporter) usage_add_exporter ;;
    update-exporter) usage_update_exporter ;;
    remove-exporter) usage_remove_exporter ;;
    add-extension) usage_add_extension ;;
    update-extension) usage_update_extension ;;
    remove-extension) usage_remove_extension ;;
    add-processor) usage_add_processor ;;
    update-processor) usage_update_processor ;;
    remove-processor) usage_remove_processor ;;
    set-should-collect) usage_set_should_collect ;;
    refresh-config) usage_refresh_config ;;
    *) usage ;;
  esac
}

require_auth() {
  if [[ -z "$CM_USER" || -z "$CM_PASS" ]]; then
    echo "❌ Error: --user and --pass are required."
    exit 1
  fi
}

urlencode() {
  jq -rn --arg v "$1" '$v|@uri'
}

normalize_base_url() {
  CM_BASE_URL="${CM_BASE_URL%/}"
  if [[ "$CM_BASE_URL" =~ ^(https?://.+)/api/(v[0-9]+)$ ]]; then
    local parsed_base="${BASH_REMATCH[1]}"
    local parsed_ver="${BASH_REMATCH[2]}"
    CM_BASE_URL="$parsed_base"
    if [[ "$API_VERSION_SET" == false ]]; then
      API_VERSION="$parsed_ver"
    fi
  fi
}

api_get() {
  local path="$1"
  local out
  out=$(curl -k -s -u "$CM_USER:$CM_PASS" "${CM_BASE_URL}/api/${API_VERSION}${path}")
  if [[ -z "$out" ]]; then
    echo "❌ Error: Empty response for GET ${path}"
    exit 1
  fi
  echo "$out"
}

# POST a JSON body to a versioned API path. Prints the response body on success
# and exits non-zero on HTTP failure.
api_post() {
  local path="$1"
  local body="$2"
  local tmp_file http_code
  tmp_file=$(mktemp)
  log "➡️  POST ${path}"
  log "$body"
  http_code=$(curl -k -s -o "$tmp_file" -w "%{http_code}" -u "$CM_USER:$CM_PASS" \
    -X POST "${CM_BASE_URL}/api/${API_VERSION}${path}" \
    -H "Content-Type: application/json" \
    -d "$body")
  if [[ "$http_code" -ge 200 && "$http_code" -lt 300 ]]; then
    cat "$tmp_file"
    rm -f "$tmp_file"
    return 0
  fi
  echo "❌ POST ${path} failed (HTTP $http_code)."
  cat "$tmp_file"
  rm -f "$tmp_file"
  return 1
}

send_batch_payload() {
  local payload="$1"
  local tmp_file
  tmp_file=$(mktemp)
  local http_code
  log "➡️  Sending /api/v15/batch payload:"
  log "$payload"
  http_code=$(curl -k -s -o "$tmp_file" -w "%{http_code}" -u "$CM_USER:$CM_PASS" \
    -X POST "${CM_BASE_URL}/api/v15/batch" \
    -H "Content-Type: application/json" \
    -d "$payload")
  if [[ "$http_code" -ge 200 && "$http_code" -lt 300 ]]; then
    log "✅ Batch request succeeded (HTTP $http_code)."
  else
    echo "❌ Batch request failed (HTTP $http_code)."
    cat "$tmp_file"
    rm -f "$tmp_file"
    exit 1
  fi
  rm -f "$tmp_file"
}

get_config_value_from_response() {
  local response="$1"
  local param_name="$2"
  echo "$response" | jq -r --arg T "$param_name" \
    '.items[] | select(.name == $T) | if .value != null then .value else .default end' 2>/dev/null
}

get_child_indent() {
  local config="$1"
  local wrapper_key="$2"
  echo "$config" | awk -v key="$wrapper_key" '
    $0 ~ "^[[:space:]]*" key "[[:space:]]*$" { found=1; next }
    found && /^[[:space:]]+[^[:space:]]/ {
      spaces = 0
      line = $0
      while (substr(line, spaces+1, 1) == " ") spaces++
      # Skip comment lines when measuring child indentation
      if (substr(line, spaces+1, 1) == "#") next
      print spaces
      exit
    }
  '
}

parse_exporter_names() {
  local config="$1"
  local root_indent="$2"
  while IFS= read -r line; do
    [[ -z "${line// }" ]] && continue
    # Ignore comment lines (first non-space char is '#')
    local stripped
    stripped=$(echo "$line" | sed 's/^[[:space:]]*//')
    [[ "${stripped:0:1}" == "#" ]] && continue
    local line_indent
    line_indent=$(echo "$line" | sed 's/^\( *\).*/\1/' | wc -c)
    line_indent=$((line_indent - 1))
    if [[ "$line_indent" -eq "$root_indent" ]] && echo "$line" | grep -qE '^[[:space:]]*[^[:space:]]+:[[:space:]]*$'; then
      local name
      name=$(echo "$line" | sed 's/:[[:space:]]*$//' | sed 's/^[[:space:]]*//')
      [[ -n "$name" ]] && echo "$name"
    fi
  done <<< "$config"
}

parse_root_names() {
  local block="$1"
  parse_exporter_names "$block" 0
}

parse_section_child_names() {
  local section="$1"
  local wrapper="$2"
  local child_indent
  child_indent=$(get_child_indent "$section" "$wrapper")
  if [[ -z "$child_indent" ]]; then
    return 0
  fi
  parse_exporter_names "$section" "$child_indent"
}

indent_block() {
  local spaces="$1"
  local block="$2"
  local pad=""
  local i
  for ((i=0; i<spaces; i++)); do
    pad="${pad} "
  done
  while IFS= read -r line; do
    if [[ -z "$line" ]]; then
      echo ""
    else
      echo "${pad}${line}"
    fi
  done <<< "$block"
}

append_blocks_under_wrapper() {
  local section="$1"
  local wrapper="$2"
  local block="$3"

  local out="$section"
  if [[ -z "$out" || "$out" == "null" ]]; then
    out="${wrapper}"
  fi
  if ! echo "$out" | grep -qE "^[[:space:]]*${wrapper}[[:space:]]*$"; then
    out="${wrapper}
${out}"
  fi

  local child_indent
  child_indent=$(get_child_indent "$out" "$wrapper")
  if [[ -z "$child_indent" ]]; then
    child_indent=2
  fi
  local indented
  indented=$(indent_block "$child_indent" "$block")
  printf '%s\n%s\n' "$out" "$indented"
}

remove_named_block() {
  local config="$1"
  local target_name="$2"
  local root_indent="$3"

  echo "$config" | awk \
    -v target="$target_name" \
    -v rindent="$root_indent" \
    'BEGIN { skip=0 }
     {
       indent = 0
       line = $0
       while (substr(line, indent+1, 1) == " ") indent++
       # A key at root indent toggles skip; comment lines never toggle it
       if (indent == rindent && $0 !~ /^[[:space:]]*$/ && substr(line, indent+1, 1) != "#") {
         name = $0
         gsub(/^[[:space:]]+/, "", name)
         gsub(/:[[:space:]]*$/, "", name)
         skip = (name == target) ? 1 : 0
       }
       if (!skip) print $0
     }'
}

remove_named_block_from_section() {
  local section="$1"
  local wrapper="$2"
  local name="$3"
  local child_indent
  child_indent=$(get_child_indent "$section" "$wrapper")
  if [[ -z "$child_indent" ]]; then
    echo "$section"
    return
  fi
  remove_named_block "$section" "$name" "$child_indent"
}

pipeline_exists() {
  local service_val="$1"
  local pipeline_name="$2"
  echo "$service_val" | grep -qF "${pipeline_name}:"
}

ensure_service_structure() {
  local service_val="$1"
  local out="$service_val"
  if [[ -z "$out" || "$out" == "null" ]]; then
    out="service:
  pipelines:"
    echo "$out"
    return
  fi
  if ! echo "$out" | grep -qE '^[[:space:]]*service:[[:space:]]*$'; then
    out="service:
${out}"
  fi
  if ! echo "$out" | grep -qE '^[[:space:]]*pipelines:[[:space:]]*$'; then
    out="${out}
  pipelines:"
  fi
  echo "$out"
}

# Extract a list-valued field (receivers|processors|exporters) from inside a named
# pipeline block, normalized to inline form "[a, b]". Handles BOTH inline lists
# ("receivers: [x, y]") and block-style YAML sequences ("receivers:" then "- x").
# Returns empty if the field is absent. Example: [prometheus/$ROLE_NAME].
extract_pipeline_list_field() {
  local service_val="$1"
  local pipeline_name="$2"
  local field="$3"
  echo "$service_val" | awk \
    -v pipeline="${pipeline_name}:" \
    -v field="${field}:" '
    function printlist(){ out="["; for(i=0;i<n;i++){out=out arr[i]; if(i<n-1) out=out ", "}; out=out "]"; print out }
    BEGIN { in_pipe=0; in_field=0; n=0 }
    {
      if (in_pipe==0) { if (index($0, pipeline)) in_pipe=1; next }
      line=$0; s=0; while (substr(line,s+1,1)==" ") s++
      trimmed=substr(line,s+1)
      if (in_field==1) {
        if (trimmed ~ /^-[[:space:]]/ || trimmed ~ /^-[[:space:]]*$/) {
          item=trimmed; sub(/^-[[:space:]]*/,"",item)
          gsub(/^[\047"]+/,"",item); gsub(/[\047"]+$/,"",item)
          if (item != "") arr[n++]=item
          next
        } else { printlist(); in_field=0; exit }
      }
      if (index(trimmed, field) == 1) {
        if (match($0, /\[[^]]*\]/)) { print substr($0, RSTART, RLENGTH); exit }
        in_field=1; n=0; next
      }
      if (trimmed ~ "^[a-zA-Z]+/[^:]*:[[:space:]]*$") exit
    }
    END { if (in_field==1) printlist() }'
}
# Default source pipeline to inherit receivers/processors from, per scope. Single
# quotes keep the literal $ROLE_NAME CM macro intact (not a shell variable here).
default_source_pipeline() {
  if [[ "$SCOPE" == "host" ]]; then echo 'metrics/hostmetrics'; else echo 'metrics/$ROLE_NAME'; fi
}

# Receivers used only when the source pipeline is absent (nothing to inherit).
default_fallback_receivers() {
  if [[ "$SCOPE" == "host" ]]; then echo '[hostmetrics]'; else echo '[prometheus/$ROLE_NAME]'; fi
}

append_pipeline_block() {
  local service_val="$1"
  local pipeline_name="$2"
  local exporter_name="$3"
  local source_pipeline="${4:-}"
  [[ -z "$source_pipeline" ]] && source_pipeline=$(default_source_pipeline)

  local out
  out=$(ensure_service_structure "$service_val")
  if pipeline_exists "$out" "$pipeline_name"; then
    echo "$out"
    return
  fi

  local pipeline_indent
  pipeline_indent=$(get_child_indent "$out" "pipelines:")
  if [[ -z "$pipeline_indent" ]]; then
    pipeline_indent=4
  fi
  local child_indent=$((pipeline_indent + 2))

  local pad_pipeline=""
  local pad_child=""
  local i
  for ((i=0; i<pipeline_indent; i++)); do pad_pipeline="${pad_pipeline} "; done
  for ((i=0; i<child_indent; i++)); do pad_child="${pad_child} "; done

  # Inherit receivers/processors from the source pipeline when present; otherwise
  # fall back to a scope-appropriate default receiver. Both the source name and the
  # fallback preserve the literal $ROLE_NAME macro for cluster/mgmt scope.
  local inherited_receivers inherited_processors
  inherited_receivers=$(extract_pipeline_list_field "$out" "$source_pipeline" 'receivers')
  inherited_processors=$(extract_pipeline_list_field "$out" "$source_pipeline" 'processors')
  [[ -z "$inherited_receivers" ]] && inherited_receivers=$(default_fallback_receivers)

  if [[ -n "$inherited_processors" ]]; then
    printf '%s\n%s%s:\n%sreceivers: %s\n%sprocessors: %s\n%sexporters: [%s]\n' \
      "$out" \
      "$pad_pipeline" "$pipeline_name" \
      "$pad_child" "$inherited_receivers" \
      "$pad_child" "$inherited_processors" \
      "$pad_child" "$exporter_name"
  else
    printf '%s\n%s%s:\n%sreceivers: %s\n%sexporters: [%s]\n' \
      "$out" \
      "$pad_pipeline" "$pipeline_name" \
      "$pad_child" "$inherited_receivers" \
      "$pad_child" "$exporter_name"
  fi
}

remove_pipeline_block() {
  local service_val="$1"
  local pipeline_name="$2"
  local pipeline_indent
  pipeline_indent=$(get_child_indent "$service_val" "pipelines:")
  if [[ -z "$pipeline_indent" ]]; then
    echo "$service_val"
    return
  fi

  echo "$service_val" | awk \
    -v target="${pipeline_name}:" \
    -v pindent="$pipeline_indent" \
    'BEGIN { skip=0 }
     {
       line = $0
       spaces = 0
       while (substr(line, spaces+1, 1) == " ") spaces++
       trimmed = substr(line, spaces+1)

       if (skip) {
         if (spaces == pindent && trimmed != "" && trimmed !~ /^#/) {
           skip = 0
         } else {
           next
         }
       }

       if (spaces == pindent && trimmed == target) {
         skip = 1
         next
       }
       print
     }'
}

# Slugify a component name into a pipeline-safe token (drops the '/' etc.).
sanitize_name() {
  local name="$1" safe
  safe=$(echo "$name" | sed 's/[^[:alnum:]_-]/-/g; s/--*/-/g; s/^-//; s/-$//')
  [[ -z "$safe" ]] && safe="custom-exporter"
  echo "$safe"
}

default_pipeline_for_exporter() {
  echo "metrics/$(sanitize_name "$1")"
}

# Host source pipelines whose receivers/processors a newly-added exporter mirrors.
# Each becomes its own dedicated pipeline for the exporter so both self-metrics and
# host metrics are forwarded with the correct per-source processors.
host_source_pipelines() {
  echo 'metrics/self-metrics'
  echo 'metrics/hostmetrics'
}

exporter_pipelines_for() {
  local exporter="$1" signal="${2:-metrics}" safe
  safe=$(sanitize_name "$exporter")
  if [[ "$signal" == "logs" ]]; then
    printf 'logs/%s-$ROLE_NAME\t%s\n' "$safe" 'logs/$ROLE_NAME'
  elif [[ "$SCOPE" == "host" ]]; then
    local src suffix
    while IFS= read -r src; do
      [[ -z "$src" ]] && continue
      suffix="${src#metrics/}"
      printf 'metrics/%s-%s\t%s\n' "$safe" "$suffix" "$src"
    done < <(host_source_pipelines)
  else
    printf 'metrics/%s-$ROLE_NAME\t%s\n' "$safe" 'metrics/$ROLE_NAME'
  fi
}

read_input_file_block() {
  # Errors go to stderr: this runs inside $(...), so stdout is captured by the caller
  # and an `exit 1` here only leaves the subshell — callers must check the status
  # (|| exit 1). Printing to stdout would swallow the message into the captured value.
  if [[ -z "$CONFIG_FILE" ]]; then
    echo "❌ Error: --file is required." >&2
    exit 1
  fi
  if [[ ! -f "$CONFIG_FILE" ]]; then
    echo "❌ Error: File '$CONFIG_FILE' not found." >&2
    exit 1
  fi
  if grep -q $'^\t' "$CONFIG_FILE"; then
    echo "❌ Error: Tabs found in '$CONFIG_FILE'. Use spaces only." >&2
    exit 1
  fi
  local content
  content=$(cat "$CONFIG_FILE")
  if [[ -z "$content" ]]; then
    echo "❌ Error: '$CONFIG_FILE' is empty." >&2
    exit 1
  fi
  echo "$content"
}

# Read --file and return the named child blocks for the given wrapper.
# Accepts either root blocks (no wrapper line) or a full "<wrapper>:" section.
resolve_input_block() {
  local wrapper="$1"
  if [[ -z "$CONFIG_FILE" ]]; then
    echo "❌ Error: --file is required." >&2
    exit 1
  fi

  local content
  content=$(read_input_file_block) || exit 1

  if echo "$content" | grep -qE "^[[:space:]]*${wrapper}[[:space:]]*$"; then
    local child_indent
    child_indent=$(get_child_indent "$content" "$wrapper")
    if [[ -z "$child_indent" ]]; then
      echo "❌ Error: '${wrapper}' section found but no child blocks were detected." >&2
      exit 1
    fi

    local extracted
    extracted=$(echo "$content" | awk -v wrapper="$wrapper" -v cindent="$child_indent" '
      BEGIN { in_wrap=0 }
      {
        line=$0
        spaces=0
        while (substr(line, spaces+1, 1) == " ") spaces++
        trimmed=substr(line, spaces+1)
        if (!in_wrap && trimmed == wrapper) { in_wrap=1; next }
        if (in_wrap) {
          if (trimmed != "" && substr(line, spaces+1, 1) != "#" && spaces < cindent) exit
          if (trimmed == "") { print ""; next }
          if (spaces >= cindent) print substr(line, cindent+1)
        }
      }')
    if [[ -z "${extracted// }" ]]; then
      echo "❌ Error: Could not extract blocks under '${wrapper}'." >&2
      exit 1
    fi
    echo "$extracted"
    return
  fi

  echo "$content"
}

resolve_input_exporter_block() {
  resolve_input_block "exporters:"
}

# ---- Pipeline field helpers (inline [ ... ] lists), ported from obs-cli.sh ----

# Add a value to a bracketed list field inside a named pipeline block.
# Inserts the field before exporters: if it does not exist yet.
add_to_pipeline_field() {
  local service_val="$1"
  local pipeline_name="$2"
  local field="$3"
  local name="$4"
  local safe_name
  safe_name=$(printf '%s' "$name" | sed 's/[&\]/\\&/g')
  echo "$service_val" | awk \
    -v pipeline="${pipeline_name}:" \
    -v field="$field:" \
    -v name="$safe_name" \
    'BEGIN { in_pipe=0; found_field=0 }
    {
      if (index($0, pipeline) && in_pipe==0) { in_pipe=1 }
      if (in_pipe) {
        trimmed = $0
        gsub(/^[[:space:]]+/, "", trimmed)
        if (index(trimmed, field) == 1) {
          if ($0 ~ /\[\]/) {
            sub(/\[\]/, "[" name "]")
          } else {
            sub(/\]$/, ", " name "]")
          }
          found_field=1
          in_pipe=0
        }
        if (!found_field && index(trimmed, "exporters:") == 1) {
          indent_str = $0
          sub(/[^ ].*/, "", indent_str)
          print indent_str field " [" name "]"
          in_pipe=0
        }
      }
      print
    }'
}

# Remove a value from a bracketed list field inside a named pipeline block.
remove_from_pipeline_field() {
  local service_val="$1"
  local pipeline_name="$2"
  local field="$3"
  local name="$4"
  local escaped
  # Escape ERE metacharacters so the name matches literally in the awk gsub regexes
  # below. Do NOT escape '/': it is not an ERE metacharacter, and '\/' makes awk warn
  # "escape sequence \/ treated as plain /" (surfaces for names like attributes/partner).
  escaped=$(printf '%s\n' "$name" | sed 's/[[\.*^$()+?{|]/\\&/g')
  echo "$service_val" | awk \
    -v pipeline="${pipeline_name}:" \
    -v field="$field:" \
    -v esc="$escaped" \
    'BEGIN { in_pipe=0 }
    {
      if (index($0, pipeline) && in_pipe==0) { in_pipe=1 }
      if (in_pipe) {
        trimmed = $0
        gsub(/^[[:space:]]+/, "", trimmed)
        if (index(trimmed, field) == 1) {
          gsub(", " esc, "")
          gsub(esc ", ", "")
          gsub("\\[" esc "\\]", "[]")
          in_pipe=0
        }
      }
      print
    }'
}

require_cluster_service_for_scope() {
  if [[ "$SCOPE" == "cluster" ]]; then
    # --cluster is always required in cluster scope. --service may be omitted to
    # fan out across all OTel-supported services (handled by run_scoped_handler,
    # which sets SERVICE before invoking the handler), so only require it here
    # once a specific service is in play.
    if [[ -z "$CLUSTER" ]]; then
      echo "❌ Error: --cluster is required for --scope cluster."
      exit 1
    fi
    if [[ -z "$SERVICE" ]]; then
      echo "❌ Error: --service is required for --scope cluster (or omit it at dispatch to fan out to all services)."
      exit 1
    fi
  fi
}

list_scope_rcgs_response() {
  if [[ "$SCOPE" == "mgmt" ]]; then
    api_get "/cm/service/roleConfigGroups"
  elif [[ "$SCOPE" == "host" ]]; then
    # No roleConfigGroups in host scope; expose the allHosts config object so
    # callers that only read this (e.g. handle_list_roles) degrade gracefully.
    api_get "/cm/allHosts/config"
  else
    api_get "/clusters/${CLUSTER}/services/${SERVICE}/roleConfigGroups"
  fi
}

# Prints the target role config group name(s) to stdout, one per line.
# All diagnostics go to stderr, and hard errors exit non-zero, so callers must
# use: rcgs=$(list_target_rcgs) || exit 1
list_target_rcgs() {
  # Host scope has no roleConfigGroups: the config lives on the single
  # /cm/allHosts/config object. Emit one sentinel token so the per-RCG loops in
  # the handlers run exactly once against that object.
  if [[ "$SCOPE" == "host" ]]; then
    if [[ -n "$ROLE" ]]; then
      echo "❌ Error: --role is not applicable in host scope." >&2
      exit 1
    fi
    echo "$ALLHOSTS_TARGET"
    return 0
  fi

  local resp
  resp=$(list_scope_rcgs_response)

  # Guard against unexpected/error responses (e.g. wrong cluster/service name
  # or auth failure) where .items is missing/null.
  if ! echo "$resp" | jq -e 'has("items") and (.items != null)' >/dev/null 2>&1; then
    echo "❌ Error: Unexpected response listing role config groups for scope '$SCOPE'." >&2
    echo "   Verify --cluster/--service names (see 'list-services') and credentials." >&2
    log "Response: $resp"
    exit 1
  fi

  if [[ -n "$ROLE" ]]; then
    local found
    found=$(echo "$resp" | jq -r --arg role "$ROLE" '(.items // [])[] | select(.name == $role) | .name')
    if [[ -z "$found" ]]; then
      echo "❌ Error: roleConfigGroup '$ROLE' not found for selected scope." >&2
      exit 1
    fi
    echo "$ROLE"
  elif [[ "$SCOPE" == "mgmt" ]]; then
    # In mgmt scope, target ONLY the Service Monitor role config group by
    # default (not every mgmt role). Match by role type to be robust to
    # naming, falling back to the well-known base group name.
    local smon
    smon=$(echo "$resp" | jq -r --arg t "$SERVICEMONITOR_ROLE_TYPE" \
      '(.items // [])[] | select(.roleType == $t) | .name')
    if [[ -z "$smon" ]]; then
      smon=$(echo "$resp" | jq -r --arg n "$SERVICEMONITOR_RCG" \
        '(.items // [])[] | select(.name == $n) | .name')
    fi
    if [[ -z "$smon" ]]; then
      echo "❌ Error: Service Monitor role config group (roleType '$SERVICEMONITOR_ROLE_TYPE' / name '$SERVICEMONITOR_RCG') not found in mgmt scope." >&2
      exit 1
    fi
    echo "$smon"
  else
    # cluster fan-out: return ALL role config groups in the service. Whether a
    # group actually participates is decided by the consuming loop via base-pipeline
    # presence (e.g. metrics/$ROLE_NAME in otelcol_service), not a role-type
    # allowlist — so newly OTel-enabled groups are picked up automatically.
    echo "$resp" | jq -r '(.items // [])[].name'
  fi
}

service_has_base_pipeline() {
  local service_val="$1" signal="${2:-metrics}"
  if [[ "$signal" == "logs" ]]; then
    pipeline_exists "$service_val" 'logs/$ROLE_NAME'
  else
    pipeline_exists "$service_val" 'metrics/$ROLE_NAME'
  fi
}

list_supported_services() {
  local svc_resp
  svc_resp=$(api_get "/clusters/${CLUSTER}/services")
  if ! echo "$svc_resp" | jq -e 'has("items") and (.items != null)' >/dev/null 2>&1; then
    echo "❌ Error: Unexpected response listing services for cluster '${CLUSTER}'." >&2
    echo "   Verify --cluster (see 'list-clusters') and credentials." >&2
    log "Response: $svc_resp"
    exit 1
  fi
  echo "$svc_resp" | jq -r '(.items // [])[].name'
}

rcg_config_path() {
  local rcg="$1"
  if [[ "$SCOPE" == "mgmt" ]]; then
    echo "/cm/service/roleConfigGroups/${rcg}/config"
  elif [[ "$SCOPE" == "host" ]]; then
    echo "/cm/allHosts/config"
  else
    echo "/clusters/${CLUSTER}/services/${SERVICE}/roleConfigGroups/${rcg}/config"
  fi
}

resolve_cluster_or_detect() {
  [[ -n "$CLUSTER" ]] && return 0
  local resp names count
  resp=$(api_get "/clusters")
  if ! echo "$resp" | jq -e 'has("items") and (.items != null)' >/dev/null 2>&1; then
    echo "❌ Error: could not list clusters to auto-detect --cluster." >&2
    exit 1
  fi
  names=$(echo "$resp" | jq -r '(.items // [])[].name')
  count=$(printf '%s\n' "$names" | grep -c .)
  if [[ "$count" -eq 1 ]]; then
    CLUSTER="$names"
    log "ℹ️  Auto-detected cluster '$CLUSTER'."
  elif [[ "$count" -eq 0 ]]; then
    echo "❌ Error: no clusters found; cannot auto-detect --cluster." >&2
    exit 1
  else
    echo "❌ Error: multiple clusters found; pass --cluster. Options:" >&2
    printf '%s\n' "$names" | sed 's/^/     - /' >&2
    exit 1
  fi
}

run_scoped_handler() {
  local handler="$1"
  if [[ "$SCOPE" == "cluster" ]]; then
    resolve_cluster_or_detect
    if [[ -z "$SERVICE" ]]; then
      local services
      services=$(list_supported_services) || exit 1
      if [[ -z "${services// }" ]]; then
        echo "❌ Error: No services found in cluster '$CLUSTER'."
        exit 1
      fi
      local svc
      while IFS= read -r svc; do
        [[ -z "$svc" ]] && continue
        SERVICE="$svc"
        # During wiring the fan-out banner is noise (most services don't
        # participate); the affected-targets summary covers what changed.
        # QUIET_FANOUT suppresses it for read-only cross-scope scans (e.g. the
        # list-processors linkage index) that fan out purely to gather data.
        if [[ "${WIRE_ACTIVE:-0}" != 1 && "${QUIET_FANOUT:-0}" != 1 ]]; then
          echo ""
          echo "########## Service: $svc ##########"
        fi
        "$handler"
      done <<< "$services"
      SERVICE=""
    else
      "$handler"
    fi
  else
    "$handler"
  fi
}

# Run a scoped handler once per scope in SCOPES (multi-scope support). Each scope
# runs with its own rules via run_scoped_handler (cluster service fan-out, etc.).
run_multiscope_handler() {
  local handler="$1"
  local sc multi=false
  [[ ${#SCOPES[@]} -gt 1 ]] && multi=true
  for sc in "${SCOPES[@]}"; do
    SCOPE="$sc"
    if [[ "$multi" == true ]]; then
      echo ""
      echo "========== SCOPE: $sc =========="
    fi
    if [[ "$sc" == "mgmt" || "$sc" == "host" ]]; then
      if [[ -n "$CLUSTER" || -n "$SERVICE" ]]; then
        log "ℹ️  --cluster/--service ignored for --scope $sc."
      fi
    fi
    run_scoped_handler "$handler"
  done
}

# Build the ordered, de-duplicated list of scopes to act on (SCOPES). --scope
# accepts a comma-separated list, the keyword 'all' (cluster+mgmt+host), or a
# single value. When --scope is omitted, default to host + cluster.
build_scopes() {
  local raw token seen
  if [[ "$SCOPE_SET" != true ]]; then
    raw="$DEFAULT_SCOPES"
  elif [[ "$SCOPE" == "all" ]]; then
    raw="cluster mgmt host"
  else
    raw="${SCOPE//,/ }"
  fi
  SCOPES=()
  seen=" "   # space-delimited set, set -u safe (avoids expanding an empty array)
  for token in $raw; do
    case "$token" in
      cluster|mgmt|host)
        if [[ "$seen" != *" $token "* ]]; then
          SCOPES+=("$token")
          seen="${seen}${token} "
        fi
        ;;
      *)
        echo "❌ Error: invalid --scope '$token'. Use cluster, mgmt, host, a comma list (cluster,host), or all."
        exit 1
        ;;
    esac
  done
  if [[ ${#SCOPES[@]} -eq 0 ]]; then
    echo "❌ Error: no valid scope resolved from --scope '$SCOPE'."
    exit 1
  fi
  # Keep SCOPE pointing at a valid single scope for help / single-scope uses.
  SCOPE="${SCOPES[0]}"
}

# Build the ordered, de-duplicated list of signals (SIGNALS). --signal accepts
# metrics|logs|both (or a comma list). Default (omitted) acts on both.
build_signals() {
  local raw token seen
  if [[ "$SIGNAL_SET" != true ]]; then
    raw="$DEFAULT_SIGNALS"
  elif [[ "$SIGNAL" == "both" ]]; then
    raw="metrics logs"
  else
    raw="${SIGNAL//,/ }"
  fi
  SIGNALS=()
  seen=" "
  for token in $raw; do
    case "$token" in
      metrics|logs)
        if [[ "$seen" != *" $token "* ]]; then
          SIGNALS+=("$token")
          seen="${seen}${token} "
        fi
        ;;
      *)
        echo "❌ Error: invalid --signal '$token'. Use metrics, logs, or both."
        exit 1
        ;;
    esac
  done
  if [[ ${#SIGNALS[@]} -eq 0 ]]; then
    echo "❌ Error: no valid signal resolved from --signal '$SIGNAL'."
    exit 1
  fi
}

# Given a signal, echo the service-section param key that holds its pipelines.
signal_service_key() {
  if [[ "$1" == "logs" ]]; then echo "$PARAM_RTM_LOGS_SERVICE"; else echo "$PARAM_SERVICE"; fi
}

# Given a signal, echo the pipeline-name prefix used for its pipelines.
signal_pipeline_prefix() {
  if [[ "$1" == "logs" ]]; then echo "logs"; else echo "metrics"; fi
}

# Prints post-write guidance only. The actual apply is done once at the tail via
# apply_stale_config (staleness-gated deploy+refresh / mgmt restart). Here we just
# note the host no-op, or (without --refresh) remind the user to re-run with it.
apply_config_if_requested() {
  local rcgs="$1"

  # Host-level config (/cm/allHosts/config) takes effect without any role
  # refresh/restart, so there is nothing to apply here.
  if [[ "$SCOPE" == "host" ]]; then
    if [[ "$REFRESH" == true && "${SUPPRESS_APPLY_NOTE:-0}" != 1 ]]; then
      echo "ℹ️  Host-level config applies without a restart/reload; --refresh is a no-op for --scope host."
    fi
    return 0
  fi

  if [[ "$REFRESH" != true ]]; then
    # During a wiring run the reminder is printed once by print_affected_summary.
    [[ "${SUPPRESS_APPLY_NOTE:-0}" != 1 ]] && echo "ℹ️  Config written but not yet applied. Re-run with --refresh to apply it to the affected role(s) so the change takes effect."
    return 0
  fi

  # --refresh was passed: the single staleness-gated apply runs once at the tail
  # (apply_stale_config wait), so nothing to do inline here.
  return 0
}

fetch_rcg_config() {
  local rcg="$1"
  local path
  path=$(rcg_config_path "$rcg")
  api_get "${path}?view=full"
}

is_yaml_block_param() {
  case "$1" in
    "$PARAM_EXPORTERS"|"$PARAM_RECEIVERS"|"$PARAM_SERVICE"|"$PARAM_PROCESSORS"|"$PARAM_EXTENSIONS"|"$PARAM_RTM_LOGS_SERVICE")
      return 0 ;;
    *)
      return 1 ;;
  esac
}

build_batch_put_item() {
  local config_path="$1"
  local message_encoded="$2"
  local exporters_val="$3"
  local service_val="$4"

  # Both values here are YAML block params; restore a single trailing newline.
  if [[ -n "$exporters_val" && "$exporters_val" != *$'\n' ]]; then
    exporters_val="${exporters_val}"$'\n'
  fi
  if [[ -n "$service_val" && "$service_val" != *$'\n' ]]; then
    service_val="${service_val}"$'\n'
  fi

  local body
  if [[ -n "$service_val" ]]; then
    body=$(jq -n \
      --arg ekey "$PARAM_EXPORTERS" --arg eval "$exporters_val" \
      --arg skey "$PARAM_SERVICE" --arg sval "$service_val" \
      '{items:[{name:$ekey,value:$eval},{name:$skey,value:$sval}]}')
  else
    body=$(jq -n \
      --arg ekey "$PARAM_EXPORTERS" --arg eval "$exporters_val" \
      '{items:[{name:$ekey,value:$eval}]}')
  fi

  jq -n \
    --arg url "/api/${API_VERSION}${config_path}?message=${message_encoded}" \
    --argjson body "$body" \
    '{method:"PUT",url:$url,body:$body,contentType:"application/json"}'
}

# Generic PUT item builder. Writes one key/value, or two if name2 is provided.
#   args: config_path message_encoded name1 value1 [name2 value2]
build_put_item() {
  local config_path="$1"
  local message_encoded="$2"
  local name1="$3"
  local value1="$4"
  local name2="${5:-}"
  local value2="${6:-}"

  # Restore a single trailing newline on YAML block params (see
  # is_yaml_block_param); leave scalar params such as should-collect untouched.
  if is_yaml_block_param "$name1" && [[ -n "$value1" && "$value1" != *$'\n' ]]; then
    value1="${value1}"$'\n'
  fi
  if [[ -n "$name2" ]] && is_yaml_block_param "$name2" && [[ -n "$value2" && "$value2" != *$'\n' ]]; then
    value2="${value2}"$'\n'
  fi

  local body
  if [[ -n "$name2" ]]; then
    body=$(jq -n \
      --arg n1 "$name1" --arg v1 "$value1" \
      --arg n2 "$name2" --arg v2 "$value2" \
      '{items:[{name:$n1,value:$v1},{name:$n2,value:$v2}]}')
  else
    body=$(jq -n \
      --arg n1 "$name1" --arg v1 "$value1" \
      '{items:[{name:$n1,value:$v1}]}')
  fi

  jq -n \
    --arg url "/api/${API_VERSION}${config_path}?message=${message_encoded}" \
    --argjson body "$body" \
    '{method:"PUT",url:$url,body:$body,contentType:"application/json"}'
}

send_batch_items() {
  local items_json_lines="$1"
  local payload
  payload=$(echo "$items_json_lines" | jq -s '{items:.}')
  send_batch_payload "$payload"
}

handle_list_clusters() {
  [[ "$SHOW_HELP" == true ]] && usage
  local resp
  resp=$(api_get "/clusters")
  echo ""
  echo "Clusters:"
  echo "$resp" | jq -r '.items[] | "  - \(.name) (\(.fullVersion // "unknown"))"'
  echo ""
}

handle_list_services() {
  [[ "$SHOW_HELP" == true ]] && usage
  if [[ -z "$CLUSTER" ]]; then
    echo "❌ Error: --cluster is required for list-services."
    exit 1
  fi
  local resp
  resp=$(api_get "/clusters/${CLUSTER}/services")
  echo ""
  printf "%-35s %-25s %s\n" "SERVICE NAME" "TYPE" "DISPLAY NAME"
  printf "%-35s %-25s %s\n" "------------" "----" "------------"
  echo "$resp" | jq -r '.items[] | [.name, .type, (.displayName // "")] | @tsv' \
    | while IFS=$'\t' read -r n t d; do
        printf "%-35s %-25s %s\n" "$n" "$t" "$d"
      done
  echo ""
}

handle_list_roles() {
  [[ "$SHOW_HELP" == true ]] && usage
  require_cluster_service_for_scope
  local resp
  resp=$(list_scope_rcgs_response)
  echo ""
  printf "%-45s %-30s %s\n" "ROLE CONFIG GROUP" "ROLE TYPE" "BASE"
  printf "%-45s %-30s %s\n" "-----------------" "---------" "----"
  echo "$resp" | jq -r '.items[] | [.name, (.roleType // ""), (.base // false)] | @tsv' \
    | while IFS=$'\t' read -r n t b; do
        printf "%-45s %-30s %s\n" "$n" "$t" "$b"
      done
  echo ""
}

handle_get_host_configs() {
  [[ "$SHOW_HELP" == true ]] && usage
  local resp
  resp=$(api_get "/cm/allHosts/config?view=full")
  echo ""
  echo "=== /cm/allHosts/config otelcol_* ==="
  echo "$resp" | jq -r '.items[] | select(.name|startswith("otelcol_")) | "[\(.name)]\n\(.value // .default)\n"'
  echo ""
}

handle_get_metric_configs() {
  [[ "$SHOW_HELP" == true ]] && usage
  require_cluster_service_for_scope
  local rcgs
  rcgs=$(list_target_rcgs) || exit 1
  if [[ -z "$rcgs" ]]; then
    echo "❌ Error: No roleConfigGroups found."
    exit 1
  fi
  while IFS= read -r rcg; do
    [[ -z "$rcg" ]] && continue
    local resp exp recv proc ext svc
    resp=$(fetch_rcg_config "$rcg")
    exp=$(get_config_value_from_response "$resp" "$PARAM_EXPORTERS")
    recv=$(get_config_value_from_response "$resp" "$PARAM_RECEIVERS")
    proc=$(get_config_value_from_response "$resp" "$PARAM_PROCESSORS")
    ext=$(get_config_value_from_response "$resp" "$PARAM_EXTENSIONS")
    svc=$(get_config_value_from_response "$resp" "$PARAM_SERVICE")
    echo ""
    echo "=== RCG: $rcg ==="
    echo ""
    echo "--- $PARAM_EXPORTERS ---"
    echo "${exp:-<not configured>}"
    echo ""
    echo "--- $PARAM_RECEIVERS ---"
    echo "${recv:-<not configured>}"
    echo ""
    echo "--- $PARAM_PROCESSORS ---"
    echo "${proc:-<not configured>}"
    echo ""
    echo "--- $PARAM_EXTENSIONS ---"
    echo "${ext:-<not configured>}"
    echo ""
    echo "--- $PARAM_SERVICE ---"
    echo "${svc:-<not configured>}"
    echo ""
  done <<< "$rcgs"
}

handle_set_should_collect() {
  [[ "$SHOW_HELP" == true ]] && usage_set_should_collect
  require_cluster_service_for_scope

  if [[ -z "$COLLECT_VALUE" ]]; then
    echo "❌ Error: provide --enable or --disable."
    exit 1
  fi

  local rcgs
  rcgs=$(list_target_rcgs) || exit 1
  if [[ -z "$rcgs" ]]; then
    echo "❌ Error: No roleConfigGroups found."
    exit 1
  fi

  local msg_encoded
  msg_encoded=$(urlencode "Modified OpenTelemetry Collector metric collection flag")
  local items=""
  while IFS= read -r rcg; do
    [[ -z "$rcg" ]] && continue
    local cfg_path item
    cfg_path=$(rcg_config_path "$rcg")
    item=$(build_put_item "$cfg_path" "$msg_encoded" "$PARAM_SHOULD_COLLECT" "$COLLECT_VALUE")
    items="${items}${item}"$'\n'
  done <<< "$rcgs"

  send_batch_items "$items"
  if [[ "$COLLECT_VALUE" == "true" ]]; then
    echo "✅ Enabled $PARAM_SHOULD_COLLECT in scope '$SCOPE'."
  else
    echo "✅ Disabled $PARAM_SHOULD_COLLECT in scope '$SCOPE'."
  fi
  apply_config_if_requested "$rcgs"
}

classify_exporter_type() {
  local name="$1"
  if [[ "$name" == */* ]]; then
    echo "${name%%/*}"
  else
    echo "generic"
  fi
}

extract_authenticator_refs() {
  local block="$1"
  printf '%s\n' "$block" \
    | sed -n 's/^[[:space:]]*authenticator:[[:space:]]*//p' \
    | sed 's/[[:space:]]*$//; s/^"//; s/"$//; s/^'\''//; s/'\''$//' \
    | grep -v '^$' \
    | sort -u
}

validate_exporter_extension_refs() {
  local exporters_val="$1" extensions_val="$2" label="${3:-}"
  local ext_names refs ref missing=0
  ext_names=$(parse_section_child_names "$extensions_val" "extensions:")
  refs=$(extract_authenticator_refs "$exporters_val")
  while IFS= read -r ref; do
    [[ -z "$ref" ]] && continue
    if ! printf '%s\n' "$ext_names" | grep -qxF "$ref"; then
      echo "❌ Error: exporter references authenticator '$ref' not defined in extensions${label:+ ($label)}."
      echo "   Define it first with add-extension so the collector config stays valid."
      missing=1
    fi
  done <<< "$refs"
  [[ "$missing" -eq 0 ]]
}

note_host_only_definition_listing() {
  local noun="$1"
  if [[ "${SCOPE_SET:-false}" == true || -n "${CLUSTER:-}" || -n "${SERVICE:-}" || -n "${ROLE:-}" ]]; then
    echo "ℹ️  $noun are defined at host (allHosts); --scope/--cluster/--service/--role do not apply to this listing."
  fi
}

handle_list_exporters() {
  [[ "$SHOW_HELP" == true ]] && usage
  note_host_only_definition_listing "Exporters"
  local exp names
  exp=$(host_config_value "$PARAM_EXPORTERS")
  names=$(parse_section_child_names "$exp" "exporters:")

  echo ""
  printf "%-45s %-40s %s\n" "ROLE CONFIG GROUP" "EXPORTER" "TYPE"
  printf "%-45s %-40s %s\n" "-----------------" "--------" "----"
  if [[ -z "$names" ]]; then
    printf "%-45s %-40s %s\n" "$ALLHOSTS_TARGET" "<none>" "-"
  else
    while IFS= read -r n; do
      [[ -z "$n" ]] && continue
      printf "%-45s %-40s %s\n" "$ALLHOSTS_TARGET" "$n" "$(classify_exporter_type "$n")"
    done <<< "$names"
  fi
  echo ""
}

# Verify that newly added exporters are present in each target RCG after write.
# Returns 0 when all exporters are present; non-zero otherwise.
verify_added_exporters() {
  local rcgs="$1"
  local expected_names="$2"
  local has_missing=0

  echo ""
  echo "Post-write verification (otelcol_exporters):"
  printf "%-45s %-35s %s\n" "ROLE CONFIG GROUP" "EXPORTER" "STATUS"
  printf "%-45s %-35s %s\n" "-----------------" "--------" "------"

  while IFS= read -r rcg; do
    [[ -z "$rcg" ]] && continue
    while IFS= read -r name; do
      [[ -z "$name" ]] && continue

      local found=0
      local attempt
      for attempt in 1 2 3; do
        local resp exp curr_names
        resp=$(fetch_rcg_config "$rcg")
        exp=$(get_config_value_from_response "$resp" "$PARAM_EXPORTERS")
        curr_names=$(parse_section_child_names "$exp" "exporters:")
        if echo "$curr_names" | grep -qxF "$name"; then
          found=1
          break
        fi
        sleep 1
      done

      if [[ "$found" -eq 1 ]]; then
        printf "%-45s %-35s %s\n" "$rcg" "$name" "PRESENT"
      else
        printf "%-45s %-35s %s\n" "$rcg" "$name" "MISSING"
        has_missing=1
      fi
    done <<< "$expected_names"
  done <<< "$rcgs"

  echo ""
  if [[ "$has_missing" -eq 1 ]]; then
    echo "❌ Verification failed: one or more exporters are missing after write."
    echo "   The batch request may have partially failed per item."
    return 1
  fi

  echo "✅ Verification passed: all added exporters are present."
  return 0
}

host_config_value() {
  local param="$1" resp save="$SCOPE"
  SCOPE="host"; resp=$(fetch_rcg_config "$ALLHOSTS_TARGET"); SCOPE="$save"
  get_config_value_from_response "$resp" "$param"
}

# PUT a single host param value to /cm/allHosts/config.
put_host_param() {
  local param="$1" value="$2" msg="$3" save="$SCOPE" cfg item
  SCOPE="host"; cfg=$(rcg_config_path "$ALLHOSTS_TARGET"); SCOPE="$save"
  item=$(build_put_item "$cfg" "$msg" "$param" "$value")
  send_batch_items "${item}"$'\n'
}

ensure_exporter_defined_host() {
  local input_block="$1" input_names="$2"
  local curr_exp curr_ext curr_names n
  curr_exp=$(host_config_value "$PARAM_EXPORTERS")
  curr_ext=$(host_config_value "$PARAM_EXTENSIONS")
  curr_names=$(parse_section_child_names "$curr_exp" "exporters:")
  while IFS= read -r n; do
    [[ -z "$n" ]] && continue
    if echo "$curr_names" | grep -qxF "$n"; then
      echo "❌ Error: Exporter '$n' already defined at host. Use update-exporter to modify it, or remove-exporter first."
      exit 1
    fi
  done <<< "$input_names"
  validate_exporter_extension_refs "$input_block" "$curr_ext" "host" || exit 1
  local updated
  updated=$(append_blocks_under_wrapper "$curr_exp" "exporters:" "$input_block")
  put_host_param "$PARAM_EXPORTERS" "$updated" "$(urlencode "Modified OpenTelemetry Collector Exporters Section")"
  echo "✅ Defined exporter(s) at host: $(echo "$input_names" | tr '\n' ' ')"
}

# Remove exporter definition(s) from host otelcol_exporters.
remove_exporter_defined_host() {
  local names="$1" curr updated n changed=0
  HOST_DEF_REMOVED=0
  curr=$(host_config_value "$PARAM_EXPORTERS")
  updated="$curr"
  while IFS= read -r n; do
    [[ -z "$n" ]] && continue
    if parse_section_child_names "$updated" "exporters:" | grep -qxF "$n"; then
      updated=$(remove_named_block_from_section "$updated" "exporters:" "$n")
      changed=1
    fi
  done <<< "$names"
  if [[ "$changed" -eq 1 ]]; then
    put_host_param "$PARAM_EXPORTERS" "$updated" "$(urlencode "Modified OpenTelemetry Collector Exporters Section")"
    HOST_DEF_REMOVED=1
    echo "✅ Removed exporter definition(s) from host."
  fi
}

# Globals consumed by _wire_pipelines_current (invoked via run_scoped_handler).
WIRE_NAMES=""
WIRE_SIGNAL="metrics"
WIRE_MODE="add"    # add | remove

WIRE_ACTIVE=0        # 1 while wiring: suppress run_scoped_handler service banners
SUPPRESS_APPLY_NOTE=0 # 1: apply_config_if_requested stays quiet (summary prints once)
AFFECTED=()          # human labels of targets actually written this command

# Record an affected target (called only for RCGs that were actually written).
record_affected() {
  local rcg="$1" signal="$2" label
  if [[ "$SCOPE" == "host" ]]; then label="host: allHosts [$signal]"
  elif [[ "$SCOPE" == "mgmt" ]]; then label="mgmt: $rcg [$signal]"
  else label="cluster: $SERVICE / $rcg [$signal]"; fi
  AFFECTED+=("$label")
}

# Print the consolidated summary of affected targets + a single apply reminder.
print_affected_summary() {
  echo ""
  if [[ ${#AFFECTED[@]} -eq 0 ]]; then
    echo "ℹ️  No matching targets were affected (no participating role config groups)."
    return 0
  fi
  echo "Affected targets:"
  printf '   - %s\n' "${AFFECTED[@]}"
  if [[ "$REFRESH" != true ]]; then
    echo "ℹ️  Config written but not yet applied. Re-run with --refresh to apply it to the affected role(s)."
  fi
}

_wire_pipelines_current() {
  local rcgs; rcgs=$(list_target_rcgs) || exit 1
  [[ -z "$rcgs" ]] && return 0
  local svc_key msg items="" rcg
  svc_key=$(signal_service_key "$WIRE_SIGNAL")
  msg=$(urlencode "Modified OpenTelemetry Collector Service Pipelines")
  while IFS= read -r rcg; do
    [[ -z "$rcg" ]] && continue
    local resp curr_svc updated_svc
    resp=$(fetch_rcg_config "$rcg")
    curr_svc=$(get_config_value_from_response "$resp" "$svc_key")
    if [[ "$SCOPE" != "host" ]] && ! service_has_base_pipeline "$curr_svc" "$WIRE_SIGNAL"; then
      continue
    fi
    updated_svc=$(ensure_service_structure "$curr_svc")
    local base_svc="$updated_svc"
    local n
    while IFS= read -r n; do
      [[ -z "$n" ]] && continue
      local pl_line pname src
      while IFS= read -r pl_line; do
        [[ -z "$pl_line" ]] && continue
        pname="${pl_line%%$'\t'*}"
        src="${pl_line#*$'\t'}"
        if [[ "$WIRE_MODE" == "remove" ]]; then
          updated_svc=$(remove_pipeline_block "$updated_svc" "$pname")
        else
          # host metrics: only mirror sources that actually exist; cluster/logs use
          # the CM template source (metrics/$ROLE_NAME or logs/$ROLE_NAME).
          if [[ "$SCOPE" == "host" && "$WIRE_SIGNAL" == "metrics" ]] && ! pipeline_exists "$updated_svc" "$src"; then
            continue
          fi
          updated_svc=$(append_pipeline_block "$updated_svc" "$pname" "$n" "$src")
        fi
      done < <(exporter_pipelines_for "$n" "$WIRE_SIGNAL")
    done <<< "$WIRE_NAMES"
    local cfg item svc_changed=0
    [[ "$updated_svc" != "$base_svc" ]] && svc_changed=1
    cfg=$(rcg_config_path "$rcg")
    local sc_item_written=0
    if [[ "$WIRE_MODE" == "add" && "$SCOPE" == "mgmt" ]]; then
      local cur_sc
      cur_sc=$(get_config_value_from_response "$resp" "$PARAM_SHOULD_COLLECT")
      if [[ "$cur_sc" != "true" ]]; then
        echo "ℹ️  Enabling $PARAM_SHOULD_COLLECT on '$rcg' (was '${cur_sc:-unset}')."
        item=$(build_put_item "$cfg" "$msg" "$svc_key" "$updated_svc" "$PARAM_SHOULD_COLLECT" "true")
        sc_item_written=1
      fi
    fi
    # Skip RCGs where nothing actually changed — a redundant PUT would falsely mark the
    # service stale (add: idempotent re-add; remove: pipeline wasn't there).
    if [[ "$svc_changed" -eq 0 && "$sc_item_written" -eq 0 ]]; then
      continue
    fi
    [[ "$sc_item_written" -eq 0 ]] && item=$(build_put_item "$cfg" "$msg" "$svc_key" "$updated_svc")
    items="${items}${item}"$'\n'
    record_affected "$rcg" "$WIRE_SIGNAL"
  done <<< "$rcgs"
  [[ -z "${items//[$'\n ']}" ]] && return 0
  send_batch_items "$items"
  apply_config_if_requested "$rcgs"
}

# Wire (add|remove) metrics pipelines across each requested --scope.
wire_metrics_pipelines() {
  local names="$1" mode="$2" sc
  WIRE_NAMES="$names"; WIRE_SIGNAL="metrics"; WIRE_MODE="$mode"
  for sc in "${SCOPES[@]}"; do
    SCOPE="$sc"
    run_scoped_handler _wire_pipelines_current
  done
}

# Wire (add|remove) logs pipelines on logs-supporting cluster RCGs (scope-independent;
# a "logs-supporting" RCG is one whose otelcol_rtm_logs_service already has a pipeline).
wire_logs_pipelines() {
  local names="$1" mode="$2"
  WIRE_NAMES="$names"; WIRE_SIGNAL="logs"; WIRE_MODE="$mode"
  SCOPE="cluster"; SERVICE=""
  run_scoped_handler _wire_pipelines_current
}

handle_add_exporter() {
  [[ "$SHOW_HELP" == true ]] && usage_add_exporter

  local input_block input_names
  input_block=$(resolve_input_exporter_block) || exit 1
  input_names=$(parse_root_names "$input_block")
  if [[ -z "$input_names" ]]; then
    echo "❌ Error: Could not parse exporter names from input."
    exit 1
  fi

  # 1) Definition always at host (shared across signals / collectors).
  ensure_exporter_defined_host "$input_block" "$input_names"

  # 2) Pipeline wiring per requested signal (quiet fan-out + affected summary).
  WIRE_ACTIVE=1; SUPPRESS_APPLY_NOTE=1; AFFECTED=()
  local sig
  for sig in "${SIGNALS[@]}"; do
    if [[ "$sig" == "metrics" ]]; then
      wire_metrics_pipelines "$input_names" add
    else
      wire_logs_pipelines "$input_names" add
    fi
  done
  WIRE_ACTIVE=0
  print_affected_summary
  echo ""
  echo "✅ add-exporter complete (signals: ${SIGNALS[*]})."
}

handle_update_exporter() {
  [[ "$SHOW_HELP" == true ]] && usage_update_exporter

  local input_block input_names
  input_block=$(resolve_input_exporter_block) || exit 1
  input_names=$(parse_root_names "$input_block")
  if [[ -z "$input_names" ]]; then
    echo "❌ Error: Could not parse exporter names from input."
    exit 1
  fi

  # Definitions live at host; update there. Pipelines are unchanged.
  local curr_exp curr_ext curr_names n updated_exp
  curr_exp=$(host_config_value "$PARAM_EXPORTERS")
  curr_ext=$(host_config_value "$PARAM_EXTENSIONS")
  curr_names=$(parse_section_child_names "$curr_exp" "exporters:")
  while IFS= read -r n; do
    [[ -z "$n" ]] && continue
    if ! echo "$curr_names" | grep -qxF "$n"; then
      echo "❌ Error: Exporter '$n' is not defined at host (define it with add-exporter first)."
      exit 1
    fi
  done <<< "$input_names"
  validate_exporter_extension_refs "$input_block" "$curr_ext" "host" || exit 1

  updated_exp="$curr_exp"
  while IFS= read -r n; do
    [[ -z "$n" ]] && continue
    updated_exp=$(remove_named_block_from_section "$updated_exp" "exporters:" "$n")
  done <<< "$input_names"
  updated_exp=$(append_blocks_under_wrapper "$updated_exp" "exporters:" "$input_block")
  put_host_param "$PARAM_EXPORTERS" "$updated_exp" "$(urlencode "Modified OpenTelemetry Collector Exporters Section")"
  echo "✅ Updated exporter definition(s) at host."
}

handle_remove_exporter() {
  [[ "$SHOW_HELP" == true ]] && usage_remove_exporter
  if [[ ${#EXPORTERS[@]} -eq 0 ]]; then
    echo "❌ Error: At least one --exporter is required."
    exit 1
  fi
  local names; names=$(printf '%s\n' "${EXPORTERS[@]}")
  HOST_DEF_REMOVED=0

  # 1) Unwire pipelines per requested signal (quiet fan-out + affected summary).
  WIRE_ACTIVE=1; SUPPRESS_APPLY_NOTE=1; AFFECTED=()
  local sig metrics_done=false
  for sig in "${SIGNALS[@]}"; do
    if [[ "$sig" == "metrics" ]]; then
      wire_metrics_pipelines "$names" remove
      metrics_done=true
    else
      wire_logs_pipelines "$names" remove
    fi
  done
  WIRE_ACTIVE=0

  # 2) Remove the host definition only when the metrics signal is in play (host
  #    holds the shared definition). --signal logs leaves the definition intact.
  if [[ "$metrics_done" == true ]]; then
    remove_exporter_defined_host "$names"
  fi

  # Nothing matched: not wired into any target pipeline and (for metrics) not defined
  # at host. Report and exit without a spurious refresh apply (no config was changed).
  if [[ ${#AFFECTED[@]} -eq 0 && "$HOST_DEF_REMOVED" != 1 ]]; then
    echo "❌ Error: exporter(s) [$(echo "$names" | tr '\n' ' ')] not found for signal(s): ${SIGNALS[*]} — not wired into any pipeline, nor defined at host. Nothing to remove."
    exit 1
  fi

  print_affected_summary
  echo ""
  echo "✅ remove-exporter complete (signals: ${SIGNALS[*]})."
}

handle_list_extensions() {
  [[ "$SHOW_HELP" == true ]] && usage
  note_host_only_definition_listing "Extensions"
  local ext names
  ext=$(host_config_value "$PARAM_EXTENSIONS")
  names=$(parse_section_child_names "$ext" "extensions:")

  echo ""
  printf "%-45s %-40s %s\n" "ROLE CONFIG GROUP" "EXTENSION" "TYPE"
  printf "%-45s %-40s %s\n" "-----------------" "---------" "----"
  if [[ -z "$names" ]]; then
    printf "%-45s %-40s %s\n" "$ALLHOSTS_TARGET" "<none>" "-"
  else
    while IFS= read -r n; do
      [[ -z "$n" ]] && continue
      printf "%-45s %-40s %s\n" "$ALLHOSTS_TARGET" "$n" "$(classify_exporter_type "$n")"
    done <<< "$names"
  fi
  echo ""
}

# Extensions are shared definitions: always defined at host (/cm/allHosts/config),
# where the single per-host collector can reference them from any pipeline. --scope
# does not apply.
handle_add_extension() {
  [[ "$SHOW_HELP" == true ]] && usage_add_extension

  local input_block input_names
  input_block=$(resolve_input_block "extensions:") || exit 1
  input_names=$(parse_root_names "$input_block")
  if [[ -z "$input_names" ]]; then
    echo "❌ Error: Could not parse extension names from input."
    exit 1
  fi

  local curr_ext curr_names n updated_ext
  curr_ext=$(host_config_value "$PARAM_EXTENSIONS")
  curr_names=$(parse_section_child_names "$curr_ext" "extensions:")
  while IFS= read -r n; do
    [[ -z "$n" ]] && continue
    if echo "$curr_names" | grep -qxF "$n"; then
      echo "❌ Error: Extension '$n' already defined at host."
      exit 1
    fi
  done <<< "$input_names"

  updated_ext=$(append_blocks_under_wrapper "$curr_ext" "extensions:" "$input_block")
  put_host_param "$PARAM_EXTENSIONS" "$updated_ext" "$(urlencode "Modified OpenTelemetry Collector Extensions Section")"
  echo "✅ Defined extension(s) at host: $(echo "$input_names" | tr '\n' ' ')"
}

handle_update_extension() {
  [[ "$SHOW_HELP" == true ]] && usage_update_extension

  local input_block input_names
  input_block=$(resolve_input_block "extensions:") || exit 1
  input_names=$(parse_root_names "$input_block")
  if [[ -z "$input_names" ]]; then
    echo "❌ Error: Could not parse extension names from input."
    exit 1
  fi

  local curr_ext curr_names n updated_ext
  curr_ext=$(host_config_value "$PARAM_EXTENSIONS")
  curr_names=$(parse_section_child_names "$curr_ext" "extensions:")
  while IFS= read -r n; do
    [[ -z "$n" ]] && continue
    if ! echo "$curr_names" | grep -qxF "$n"; then
      echo "❌ Error: Extension '$n' is not defined at host (add it first)."
      exit 1
    fi
  done <<< "$input_names"

  updated_ext="$curr_ext"
  while IFS= read -r n; do
    [[ -z "$n" ]] && continue
    updated_ext=$(remove_named_block_from_section "$updated_ext" "extensions:" "$n")
  done <<< "$input_names"
  updated_ext=$(append_blocks_under_wrapper "$updated_ext" "extensions:" "$input_block")
  put_host_param "$PARAM_EXTENSIONS" "$updated_ext" "$(urlencode "Modified OpenTelemetry Collector Extensions Section")"
  echo "✅ Updated extension(s) at host."
}

handle_remove_extension() {
  [[ "$SHOW_HELP" == true ]] && usage_remove_extension
  if [[ ${#EXTENSIONS[@]} -eq 0 ]]; then
    echo "❌ Error: At least one --extension is required."
    exit 1
  fi

  local curr_ext curr_names curr_exp in_use_refs updated_ext ex
  curr_ext=$(host_config_value "$PARAM_EXTENSIONS")
  curr_names=$(parse_section_child_names "$curr_ext" "extensions:")
  # Authenticators referenced by host exporters; removing one still in use breaks
  # the collector.
  curr_exp=$(host_config_value "$PARAM_EXPORTERS")
  in_use_refs=$(extract_authenticator_refs "$curr_exp")

  updated_ext="$curr_ext"
  for ex in "${EXTENSIONS[@]}"; do
    if ! echo "$curr_names" | grep -qxF "$ex"; then
      echo "❌ Error: Extension '$ex' not defined at host."
      exit 1
    fi
    if printf '%s\n' "$in_use_refs" | grep -qxF "$ex"; then
      echo "❌ Error: Extension '$ex' is still referenced as an authenticator by a host exporter."
      echo "   Update or remove that exporter first so the collector config stays valid."
      exit 1
    fi
    updated_ext=$(remove_named_block_from_section "$updated_ext" "extensions:" "$ex")
  done
  put_host_param "$PARAM_EXTENSIONS" "$updated_ext" "$(urlencode "Modified OpenTelemetry Collector Extensions Section")"
  echo "✅ Removed extension(s) from host."
}

list_pipeline_names() {
  local service_val="$1"
  echo "$service_val" | awk '
    /^[[:space:]]+[^[:space:]].*:[[:space:]]*$/ {
      line=$0; s=0; while (substr(line,s+1,1)==" ") s++
      if (substr(line,s+1,1)=="#") next
      name=substr(line,s+1); sub(/:[[:space:]]*$/,"",name)
      if (name ~ /^(metrics|logs|traces)/) print name
    }'
}

# Exact processor names inside a pipeline's `processors:` list, one per line.
# Uses extract_pipeline_list_field (returns "[a, b]") and splits it, so matching is
# by whole name — never a substring (attributes/partner must not match
# attributes/partner-2).
pipeline_processor_names() {
  local svc="$1" pl="$2" list
  list=$(extract_pipeline_list_field "$svc" "$pl" "processors")
  list="${list#[}"; list="${list%]}"
  printf '%s' "$list" | tr ',' '\n' | sed 's/^[[:space:]]*//; s/[[:space:]]*$//' | grep -v '^$'
}

PROC_LINK_PAIRS=""   # newline "proc<TAB>pipeline" rows across all scanned scopes
PROC_LINK_BUILT=0

# For the current $SCOPE (+ $SERVICE in cluster fan-out), append every
# proc<TAB>pipeline pair found in this scope's RCG service sections (both signals).
_index_proc_links_current() {
  local rcgs; rcgs=$(list_target_rcgs) || exit 1
  [[ -z "$rcgs" ]] && return 0
  local rcg
  while IFS= read -r rcg; do
    [[ -z "$rcg" ]] && continue
    local resp svc_m svc_l svc pl p
    resp=$(fetch_rcg_config "$rcg")
    svc_m=$(get_config_value_from_response "$resp" "$PARAM_SERVICE")
    svc_l=$(get_config_value_from_response "$resp" "$PARAM_RTM_LOGS_SERVICE")
    svc="$svc_m
$svc_l"
    while IFS= read -r pl; do
      [[ -z "$pl" ]] && continue
      while IFS= read -r p; do
        [[ -z "$p" ]] && continue
        PROC_LINK_PAIRS="${PROC_LINK_PAIRS}${p}"$'\t'"${pl}"$'\n'
      done <<< "$(pipeline_processor_names "$svc" "$pl")"
    done <<< "$(list_pipeline_names "$svc")"
  done <<< "$rcgs"
}

# Build PROC_LINK_PAIRS once, fanning out over every requested scope (banners
# suppressed via QUIET_FANOUT). Restores SCOPE/SERVICE afterwards.
build_proc_link_index() {
  [[ "$PROC_LINK_BUILT" == 1 ]] && return 0
  local save_scope="$SCOPE" save_service="$SERVICE" sc
  QUIET_FANOUT=1
  for sc in "${SCOPES[@]}"; do
    SCOPE="$sc"; SERVICE=""
    run_scoped_handler _index_proc_links_current
  done
  QUIET_FANOUT=0
  SCOPE="$save_scope"; SERVICE="$save_service"
  PROC_LINK_BUILT=1
}

# Distinct pipelines a processor is linked into (across all scanned scopes), joined
# with ", ". Empty if none.
proc_linked_pipelines() {
  local name="$1"
  printf '%s' "$PROC_LINK_PAIRS" | awk -F'\t' -v n="$name" '$1==n{print $2}' \
    | awk '!seen[$0]++' | paste -sd, - | sed 's/,/, /g'
}

handle_list_processors() {
  [[ "$SHOW_HELP" == true ]] && usage
  build_proc_link_index
  local proc names
  proc=$(host_config_value "$PARAM_PROCESSORS")
  names=$(parse_section_child_names "$proc" "processors:")

  echo ""
  printf "%-40s %-35s %s\n" "ROLE CONFIG GROUP" "PROCESSOR" "LINKED PIPELINES"
  printf "%-40s %-35s %s\n" "-----------------" "---------" "----------------"
  if [[ -z "$names" ]]; then
    printf "%-40s %-35s %s\n" "$ALLHOSTS_TARGET" "<none>" "-"
    echo ""
    return 0
  fi
  while IFS= read -r n; do
    [[ -z "$n" ]] && continue
    local linked
    linked=$(proc_linked_pipelines "$n")
    printf "%-40s %-35s %s\n" "$ALLHOSTS_TARGET" "$n" "${linked:-(not linked)}"
  done <<< "$names"
  echo ""
}

# ---- processor definition (host) + signal-aware pipeline linking -----------

# Ensure processor definition(s) exist at host otelcol_processors (shared by the
# per-host collector across signals). Definitions are host-level and signal/scope-
# independent, so a name already defined at host is a hard error (mirrors add-extension).
ensure_processor_defined_host() {
  local input_block="$1" input_names="$2"
  local curr curr_names n updated
  curr=$(host_config_value "$PARAM_PROCESSORS")
  curr_names=$(parse_section_child_names "$curr" "processors:")
  while IFS= read -r n; do
    [[ -z "$n" ]] && continue
    if echo "$curr_names" | grep -qxF "$n"; then
      echo "❌ Error: Processor '$n' already defined at host. Use update-processor to modify it, or remove-processor first."
      exit 1
    fi
  done <<< "$input_names"
  updated=$(append_blocks_under_wrapper "$curr" "processors:" "$input_block")
  put_host_param "$PARAM_PROCESSORS" "$updated" "$(urlencode "Modified OpenTelemetry Collector Processors Section")"
  echo "✅ Defined processor(s) at host: $(echo "$input_names" | tr '\n' ' ')"
}

# Return 0 if every name (newline-separated) is defined at host otelcol_processors.
processor_defined_at_host() {
  local names="$1" curr_names n
  curr_names=$(parse_section_child_names "$(host_config_value "$PARAM_PROCESSORS")" "processors:")
  while IFS= read -r n; do
    [[ -z "$n" ]] && continue
    echo "$curr_names" | grep -qxF "$n" || return 1
  done <<< "$names"
  return 0
}

# Globals consumed by _link_processors_current (invoked via run_scoped_handler).
PLINK_EXPORTERS=""   # newline-separated exporter names whose pipelines to touch
PLINK_PROCS=""       # newline-separated processor names to link/unlink
PLINK_SIGNAL="metrics"
PLINK_MODE="add"     # add | remove
PLINK_MATCHED=" "    # space-delimited set of exporters that matched >=1 pipeline (add mode)

# For the current $SCOPE + $SERVICE and PLINK_SIGNAL, add/remove PLINK_PROCS in the
# processors list of each PLINK_EXPORTERS pipeline. Writes only the service section.
_link_processors_current() {
  local rcgs; rcgs=$(list_target_rcgs) || exit 1
  [[ -z "$rcgs" ]] && return 0
  local svc_key msg items="" rcg
  svc_key=$(signal_service_key "$PLINK_SIGNAL")
  msg=$(urlencode "Modified OpenTelemetry Collector Service Pipelines")
  while IFS= read -r rcg; do
    [[ -z "$rcg" ]] && continue
    local resp curr_svc updated_svc
    resp=$(fetch_rcg_config "$rcg")
    curr_svc=$(get_config_value_from_response "$resp" "$svc_key")
    if [[ "$SCOPE" != "host" ]] && ! service_has_base_pipeline "$curr_svc" "$PLINK_SIGNAL"; then
      continue
    fi
    updated_svc=$(ensure_service_structure "$curr_svc")
    local base_svc="$updated_svc"
    local ex pl_line pl found_pl p
    for ex in $PLINK_EXPORTERS; do
      found_pl=0
      while IFS= read -r pl_line; do
        [[ -z "$pl_line" ]] && continue
        pl="${pl_line%%$'\t'*}"
        pipeline_exists "$updated_svc" "$pl" || continue
        found_pl=1
        while IFS= read -r p; do
          [[ -z "$p" ]] && continue
          if [[ "$PLINK_MODE" == "remove" ]]; then
            updated_svc=$(remove_from_pipeline_field "$updated_svc" "$pl" "processors" "$p")
          else
            updated_svc=$(add_to_pipeline_field "$updated_svc" "$pl" "processors" "$p")
          fi
        done <<< "$PLINK_PROCS"
      done < <(exporter_pipelines_for "$ex" "$PLINK_SIGNAL")
      # An exporter can legitimately exist in only one signal (e.g. a logs-only
      # exporter). Don't error on per-RCG/per-signal absence — just record which
      # exporters matched a pipeline somewhere; link_processors_signals errors after
      # the fact (add mode) for any exporter that matched no pipeline at all.
      [[ "$found_pl" -eq 1 && "$PLINK_MATCHED" != *" $ex "* ]] && PLINK_MATCHED="${PLINK_MATCHED}${ex} "
    done
    # Only write + record RCGs whose service section actually changed. Unlinking a
    # processor that isn't in the pipeline (or a pass where nothing matched) leaves the
    # section untouched — no redundant PUT, no false "affected", no false staleness.
    [[ "$updated_svc" == "$base_svc" ]] && continue
    local cfg item
    cfg=$(rcg_config_path "$rcg")
    item=$(build_put_item "$cfg" "$msg" "$svc_key" "$updated_svc")
    items="${items}${item}"$'\n'
    record_affected "$rcg" "$PLINK_SIGNAL"
  done <<< "$rcgs"
  [[ -z "${items//[$'\n ']}" ]] && return 0
  send_batch_items "$items"
  apply_config_if_requested "$rcgs"
}

link_processors_signals() {
  local mode="$1" sig sc
  PLINK_MODE="$mode"
  PLINK_MATCHED=" "
  for sig in "${SIGNALS[@]}"; do
    PLINK_SIGNAL="$sig"
    if [[ "$sig" == "metrics" ]]; then
      for sc in "${SCOPES[@]}"; do
        SCOPE="$sc"
        run_scoped_handler _link_processors_current
      done
    else
      SCOPE="cluster"; SERVICE=""
      run_scoped_handler _link_processors_current
    fi
  done
  if [[ "$mode" == "add" ]]; then
    local ex unmatched=()
    for ex in $PLINK_EXPORTERS; do
      [[ "$PLINK_MATCHED" == *" $ex "* ]] || unmatched+=("$ex")
    done
    if [[ ${#unmatched[@]} -gt 0 ]]; then
      echo "❌ Error: No pipeline found for exporter(s) in signal(s) [${SIGNALS[*]}] — run add-exporter first:"
      for ex in "${unmatched[@]}"; do echo "   - $ex"; done
      exit 1
    fi
  fi
}

_scrub_processors_current() {
  local rcgs; rcgs=$(list_target_rcgs) || exit 1
  [[ -z "$rcgs" ]] && return 0
  local svc_key msg items="" rcg
  svc_key=$(signal_service_key "$PLINK_SIGNAL")
  msg=$(urlencode "Modified OpenTelemetry Collector Service Pipelines")
  while IFS= read -r rcg; do
    [[ -z "$rcg" ]] && continue
    local resp curr_svc updated_svc base_svc pl p
    resp=$(fetch_rcg_config "$rcg")
    curr_svc=$(get_config_value_from_response "$resp" "$svc_key")
    [[ -z "$curr_svc" || "$curr_svc" == "null" ]] && continue
    updated_svc=$(ensure_service_structure "$curr_svc")
    base_svc="$updated_svc"
    while IFS= read -r pl; do
      [[ -z "$pl" ]] && continue
      while IFS= read -r p; do
        [[ -z "$p" ]] && continue
        updated_svc=$(remove_from_pipeline_field "$updated_svc" "$pl" "processors" "$p")
      done <<< "$PLINK_PROCS"
    done <<< "$(list_pipeline_names "$updated_svc")"
    [[ "$updated_svc" == "$base_svc" ]] && continue
    local cfg item
    cfg=$(rcg_config_path "$rcg")
    item=$(build_put_item "$cfg" "$msg" "$svc_key" "$updated_svc")
    items="${items}${item}"$'\n'
    record_affected "$rcg" "$PLINK_SIGNAL"
  done <<< "$rcgs"
  [[ -z "${items//[$'\n ']}" ]] && return 0
  send_batch_items "$items"
  apply_config_if_requested "$rcgs"
}

scrub_processors_signals() {
  local sig sc
  for sig in "${SIGNALS[@]}"; do
    PLINK_SIGNAL="$sig"
    if [[ "$sig" == "metrics" ]]; then
      for sc in "${SCOPES[@]}"; do
        SCOPE="$sc"
        run_scoped_handler _scrub_processors_current
      done
    else
      SCOPE="cluster"; SERVICE=""
      run_scoped_handler _scrub_processors_current
    fi
  done
}

handle_add_processor() {
  [[ "$SHOW_HELP" == true ]] && usage_add_processor

  if [[ -z "$CONFIG_FILE" && ${#PROCESSORS[@]} -eq 0 ]]; then
    echo "❌ Error: Provide --file to define processor(s), or --processor <name> --exporter <name> to link an existing one."
    exit 1
  fi
  if [[ -n "$CONFIG_FILE" && ${#PROCESSORS[@]} -gt 0 ]]; then
    echo "❌ Error: --file and --processor cannot be used together."
    exit 1
  fi
  if [[ ${#PROCESSORS[@]} -gt 0 && ${#EXPORTERS[@]} -eq 0 ]]; then
    echo "❌ Error: --exporter is required when linking with --processor."
    exit 1
  fi

  if [[ ${#EXPORTERS[@]} -gt 0 ]]; then
    local _ex _exp_defined
    _exp_defined=$(parse_section_child_names "$(host_config_value "$PARAM_EXPORTERS")" "exporters:")
    for _ex in "${EXPORTERS[@]}"; do
      if ! echo "$_exp_defined" | grep -qxF "$_ex"; then
        echo "❌ Error: Exporter '$_ex' is not defined at host. Run add-exporter first."
        exit 1
      fi
    done
  fi

  local link_names=""
  if [[ -n "$CONFIG_FILE" ]]; then
    local input_block input_names
    input_block=$(resolve_input_block "processors:") || exit 1
    input_names=$(parse_root_names "$input_block")
    if [[ -z "$input_names" ]]; then
      echo "❌ Error: Could not parse processor names from input."
      exit 1
    fi
    # Definition always at host.
    ensure_processor_defined_host "$input_block" "$input_names"
    link_names="$input_names"
  else
    link_names=$(printf '%s\n' "${PROCESSORS[@]}")
    if ! processor_defined_at_host "$link_names"; then
      echo "❌ Error: processor(s) not defined at host. Define them first with: add-processor --file <yaml>."
      exit 1
    fi
  fi

  # Optionally link into the exporter's pipeline(s) per signal/scope.
  if [[ ${#EXPORTERS[@]} -gt 0 ]]; then
    PLINK_EXPORTERS=$(printf '%s\n' "${EXPORTERS[@]}")
    PLINK_PROCS="$link_names"
    WIRE_ACTIVE=1; SUPPRESS_APPLY_NOTE=1; AFFECTED=()
    link_processors_signals add
    WIRE_ACTIVE=0
    print_affected_summary
  fi
  echo ""
  echo "✅ add-processor complete."
}

handle_update_processor() {
  [[ "$SHOW_HELP" == true ]] && usage_update_processor

  local input_block input_names
  input_block=$(resolve_input_block "processors:") || exit 1
  input_names=$(parse_root_names "$input_block")
  if [[ -z "$input_names" ]]; then
    echo "❌ Error: Could not parse processor names from input."
    exit 1
  fi

  local curr curr_names n updated
  curr=$(host_config_value "$PARAM_PROCESSORS")
  curr_names=$(parse_section_child_names "$curr" "processors:")
  while IFS= read -r n; do
    [[ -z "$n" ]] && continue
    if ! echo "$curr_names" | grep -qxF "$n"; then
      echo "❌ Error: Processor '$n' is not defined at host (add it first)."
      exit 1
    fi
  done <<< "$input_names"

  updated="$curr"
  while IFS= read -r n; do
    [[ -z "$n" ]] && continue
    updated=$(remove_named_block_from_section "$updated" "processors:" "$n")
  done <<< "$input_names"
  updated=$(append_blocks_under_wrapper "$updated" "processors:" "$input_block")
  put_host_param "$PARAM_PROCESSORS" "$updated" "$(urlencode "Modified OpenTelemetry Collector Processors Section")"
  echo "✅ Updated processor(s) at host."
}

handle_remove_processor() {
  [[ "$SHOW_HELP" == true ]] && usage_remove_processor
  if [[ ${#PROCESSORS[@]} -eq 0 ]]; then
    echo "❌ Error: At least one --processor is required."
    exit 1
  fi
  local names; names=$(printf '%s\n' "${PROCESSORS[@]}")

  if [[ ${#EXPORTERS[@]} -gt 0 ]]; then
    # Unlink from the given exporter pipeline(s) per signal/scope; keep definition.
    PLINK_EXPORTERS=$(printf '%s\n' "${EXPORTERS[@]}")
    PLINK_PROCS="$names"
    WIRE_ACTIVE=1; SUPPRESS_APPLY_NOTE=1; AFFECTED=()
    link_processors_signals remove
    WIRE_ACTIVE=0
    if [[ ${#AFFECTED[@]} -eq 0 ]]; then
      echo "ℹ️  Nothing to unlink: processor(s) [$(echo "$names" | tr '\n' ' ')] not linked to the given exporter pipeline(s) for signal(s) [${SIGNALS[*]}]. Nothing changed."
      exit 0
    fi
    print_affected_summary
    echo ""
    echo "✅ Unlinked processor(s) from exporter pipeline(s)."
  else
    local curr n
    curr=$(host_config_value "$PARAM_PROCESSORS")
    while IFS= read -r n; do
      [[ -z "$n" ]] && continue
      if ! parse_section_child_names "$curr" "processors:" | grep -qxF "$n"; then
        echo "❌ Error: Processor '$n' not defined at host."
        echo "   Defined at host: $(parse_section_child_names "$curr" "processors:" | tr '\n' ' ')"
        exit 1
      fi
    done <<< "$names"

    PLINK_PROCS="$names"
    WIRE_ACTIVE=1; SUPPRESS_APPLY_NOTE=1; AFFECTED=()
    scrub_processors_signals
    WIRE_ACTIVE=0

    local updated
    updated="$curr"
    while IFS= read -r n; do
      [[ -z "$n" ]] && continue
      updated=$(remove_named_block_from_section "$updated" "processors:" "$n")
    done <<< "$names"
    put_host_param "$PARAM_PROCESSORS" "$updated" "$(urlencode "Modified OpenTelemetry Collector Processors Section")"

    print_affected_summary
    echo ""
    echo "✅ Removed processor definition(s) from host and unlinked from all pipelines."
  fi
}

cm_v41_get() {
  [[ "${DEBUG:-false}" == true ]] && echo "   [debug] GET /api/$CM_API_STALENESS$1" >&2
  curl -k -s -u "$CM_USER:$CM_PASS" "$CM_BASE_URL/api/$CM_API_STALENESS$1"
}

cm_get() {
  local ver="${API_VERSION:-$DEFAULT_API_VERSION}"
  [[ "${DEBUG:-false}" == true ]] && echo "   [debug] GET /api/$ver$1" >&2
  curl -k -s -u "$CM_USER:$CM_PASS" "$CM_BASE_URL/api/$ver$1"
}

cm_v41_post() {
  local path="$1" body="${2:-}" tmp_file http_code
  tmp_file=$(mktemp)
  if [[ -n "$body" ]]; then
    http_code=$(curl -k -s -o "$tmp_file" -w "%{http_code}" -u "$CM_USER:$CM_PASS" -X POST \
      -H "Content-Type: application/json" -d "$body" \
      "$CM_BASE_URL/api/$CM_API_STALENESS$path")
  else
    http_code=$(curl -k -s -o "$tmp_file" -w "%{http_code}" -u "$CM_USER:$CM_PASS" -X POST \
      "$CM_BASE_URL/api/$CM_API_STALENESS$path")
  fi
  if [[ "$http_code" == "401" || "$http_code" == "403" ]]; then
    rm -f "$tmp_file"
    printf '{"message":"Permission denied (HTTP %s) — this command requires full Cloudera Manager administrator rights."}' "$http_code"
    return 0
  fi
  cat "$tmp_file"
  rm -f "$tmp_file"
}

cm_wait_command() {
  local id="$1" waited=0 resp active success msg
  [[ -z "$id" || "$id" == "null" ]] && return 0
  while :; do
    resp=$(cm_v41_get "/commands/$id")
    active=$(echo "$resp" | jq -r '.active // false' 2>/dev/null)
    [[ "$active" != "true" ]] && break
    if [[ "$waited" -ge "$CM_POLL_TIMEOUT" ]]; then
      echo "   ⏳ Command $id still running after ${CM_POLL_TIMEOUT}s; not waiting further."
      return 0
    fi
    sleep "$CM_POLL_INTERVAL"; waited=$((waited + CM_POLL_INTERVAL))
  done
  success=$(echo "$resp" | jq -r '.success // "unknown"' 2>/dev/null)
  msg=$(echo "$resp" | jq -r '.resultMessage // ""' 2>/dev/null)
  if [[ "$success" == "true" ]]; then
    echo "   ✅ Command $id succeeded${msg:+ — $msg}"
  else
    echo "   ❌ Command $id did not succeed (success=$success)${msg:+ — $msg}"
  fi
}

_target_clusters() {
  if [[ -n "${CLUSTER:-}" ]]; then
    printf '%s\n' "$CLUSTER"
  else
    cm_get "/clusters" | jq -r '.items[]?.name' 2>/dev/null
  fi
}

_scan_stale() {
  local target="$1" cl enc svc_json stale any=1
  SCAN_CLUSTERS=""
  SCAN_MGMT=$(cm_v41_get "/cm/service?view=full" | jq -r '.configStalenessStatus // "FRESH"' 2>/dev/null)
  [[ "${DEBUG:-false}" == true ]] && echo "   [debug] mgmt configStalenessStatus=${SCAN_MGMT:-<empty>}" >&2
  [[ -n "$SCAN_MGMT" && "$SCAN_MGMT" != "FRESH" ]] && any=0
  while IFS= read -r cl; do
    [[ -z "$cl" ]] && continue
    enc=$(urlencode "$cl")
    svc_json=$(cm_v41_get "/clusters/$enc/services?view=full")
    stale=$(echo "$svc_json" | jq -r '[.items[]? | select((.configStalenessStatus // "FRESH") != "FRESH") | .name] | join(",")' 2>/dev/null)
    [[ "${DEBUG:-false}" == true ]] && echo "   [debug] cluster '$cl': $(echo -n "$svc_json" | wc -c | tr -d ' ') bytes, stale=[${stale}]" >&2
    if [[ -n "$stale" ]]; then
      SCAN_CLUSTERS+="${cl}"$'\t'"${stale}"$'\n'
      any=0
    fi
  done <<< "$target"
  return $any
}

apply_stale_config() {
  local wait_compute="${1:-}"
  require_auth
  echo ""
  echo "🔎 refresh: checking Cloudera Manager for stale configuration..."

  # Enumerate target clusters ONCE (reused for the hint and every scan below).
  local clusters
  clusters=$(_target_clusters)
  [[ "${DEBUG:-false}" == true ]] && echo "   [debug] clusters=[$(echo "$clusters" | tr '\n' ' ' | sed 's/ *$//')]" >&2
  # On Public Cloud the account often cannot list clusters (v41 /clusters is empty);
  # point the user at --cluster so the cluster path isn't silently skipped. Mgmt is
  # still scanned regardless.
  if [[ -z "${clusters//[$'\n' ]}" && -z "${CLUSTER:-}" ]]; then
    echo "   ℹ️  CM returned no clusters for this account. Pass --cluster <name> to check/apply a specific cluster (the Management Service is still checked)."
  fi

  # Scan once; in wait mode re-scan until staleness appears (CM computes it async
  # after a write). Each poll is one pass — not three.
  local waited=0
  while ! _scan_stale "$clusters"; do
    if [[ "$wait_compute" != "wait" ]]; then
      echo "   ℹ️  Nothing stale — nothing to do."
      return 0
    fi
    if [[ "$waited" -ge 60 ]]; then
      echo "   ℹ️  Nothing stale — configuration already applied (nothing to do)."
      return 0
    fi
    sleep 5; waited=$((waited + 5))
  done

  # --- Cluster apply: POST deployClientConfigsAndRefresh (from scanned data) ---
  local line cl enc stale_svcs resp id
  while IFS= read -r line; do
    [[ -z "$line" ]] && continue
    cl="${line%%$'\t'*}"; stale_svcs="${line#*$'\t'}"
    enc=$(urlencode "$cl")
    echo ""
    echo "Apply plan (cluster '$cl'): DEPLOY CLIENT CONFIGS + REFRESH"
    printf '   • %s\n' ${stale_svcs//,/ }
    resp=$(cm_v41_post "/clusters/$enc/commands/deployClientConfigsAndRefresh")
    id=$(echo "$resp" | jq -r '.id // empty' 2>/dev/null)
    if [[ -z "$id" ]]; then
      echo "   ❌ deployClientConfigsAndRefresh failed: $(echo "$resp" | jq -r '.message // .' 2>/dev/null)"
    else
      echo "   🔄 Submitted deployClientConfigsAndRefresh (command $id)..."
      cm_wait_command "$id"
    fi
  done <<< "$SCAN_CLUSTERS"

  # --- Mgmt apply: restart if the scan found it stale (mgmt has no refresh) ---
  if [[ -n "$SCAN_MGMT" && "$SCAN_MGMT" != "FRESH" ]]; then
    echo ""
    echo "Apply plan (Management Service): RESTART (mgmt is $SCAN_MGMT)"
    resp=$(cm_v41_post "/cm/service/commands/restart")
    id=$(echo "$resp" | jq -r '.id // empty' 2>/dev/null)
    if [[ -z "$id" ]]; then
      echo "   ❌ mgmt restart failed: $(echo "$resp" | jq -r '.message // .' 2>/dev/null)"
    else
      echo "   ♻️  Restarting Management Service (command $id)..."
      cm_wait_command "$id"
    fi
  else
    echo "   ℹ️  Management Service is FRESH — no mgmt action needed."
  fi

  echo ""
  echo "✅ refresh: apply step complete."
}

handle_refresh_config() {
  [[ "$SHOW_HELP" == true ]] && usage_refresh_config
  apply_stale_config
}

# When sourced as a library (e.g. by tests), stop here: all functions are
# defined above, but no argument parsing or command dispatch should run.
if [[ -n "${OBS_CLOUD_CLI_LIBRARY_MODE:-}" ]]; then
  return 0 2>/dev/null || exit 0
fi

if [[ $# -eq 0 ]]; then
  usage
fi

COMMAND=""
CM_BASE_URL="$DEFAULT_CM_BASE_URL"
CM_USER=""
CM_PASS=""
API_VERSION="$DEFAULT_API_VERSION"
API_VERSION_SET=false
SCOPE="$DEFAULT_SCOPE"
SCOPE_SET=false
SCOPES=()
SIGNAL=""
SIGNAL_SET=false
SIGNALS=()
CLUSTER=""
SERVICE=""
ROLE=""
CONFIG_FILE=""
SHOW_HELP=false
REFRESH=false
COLLECT_VALUE=""
EXPORTERS=()
EXTENSIONS=()
PROCESSORS=()

while [[ $# -gt 0 ]]; do
  case "$1" in
    --url) CM_BASE_URL="$2"; shift 2 ;;
    --user) CM_USER="$2"; shift 2 ;;
    --pass) CM_PASS="$2"; shift 2 ;;
    --api-version) API_VERSION="$2"; API_VERSION_SET=true; shift 2 ;;
    --scope) SCOPE="$2"; SCOPE_SET=true; shift 2 ;;
    --signal) SIGNAL="$2"; SIGNAL_SET=true; shift 2 ;;
    --cluster) CLUSTER="$2"; shift 2 ;;
    --service) SERVICE="$2"; shift 2 ;;
    --role) ROLE="$2"; shift 2 ;;
    --file) CONFIG_FILE="$2"; shift 2 ;;
    --exporter) EXPORTERS+=("$2"); shift 2 ;;
    --extension) EXTENSIONS+=("$2"); shift 2 ;;
    --processor) PROCESSORS+=("$2"); shift 2 ;;
    --exporters-key) PARAM_EXPORTERS="$2"; shift 2 ;;
    --receivers-key) PARAM_RECEIVERS="$2"; shift 2 ;;
    --service-key) PARAM_SERVICE="$2"; shift 2 ;;
    --processors-key) PARAM_PROCESSORS="$2"; shift 2 ;;
    --extensions-key) PARAM_EXTENSIONS="$2"; shift 2 ;;
    --servicemonitor-role-type) SERVICEMONITOR_ROLE_TYPE="$2"; shift 2 ;;
    --servicemonitor-rcg) SERVICEMONITOR_RCG="$2"; shift 2 ;;
    --refresh) REFRESH=true; shift ;;
    --enable) COLLECT_VALUE="true"; shift ;;
    --disable) COLLECT_VALUE="false"; shift ;;
    --debug) DEBUG=true; shift ;;
    --help|-h) SHOW_HELP=true; shift ;;
    -*)
      echo "❌ Unknown option: '$1'"
      if [[ -n "$COMMAND" ]]; then
        show_command_help "$COMMAND"
      else
        usage
      fi
      ;;
    *)
      if [[ -z "$COMMAND" ]]; then
        COMMAND="$1"
        shift
      else
        echo "❌ Unexpected argument: '$1'"
        show_command_help "$COMMAND"
      fi
      ;;
  esac
done

if [[ -z "$COMMAND" ]]; then
  usage
fi

build_scopes
build_signals

if [[ "$SHOW_HELP" == true ]]; then
  if [[ -n "$COMMAND" ]]; then
    show_command_help "$COMMAND"
  else
    usage
  fi
fi

normalize_base_url
require_auth

# ==============================================================================
# EXECUTE COMMAND
# ==============================================================================

case "$COMMAND" in
  list-clusters)
    handle_list_clusters
    ;;
  list-services)
    handle_list_services
    ;;
  list-roles)
    run_multiscope_handler handle_list_roles
    ;;
  get-host-configs)
    handle_get_host_configs
    ;;
  get-metric-configs)
    run_multiscope_handler handle_get_metric_configs
    ;;
  set-should-collect)
    run_multiscope_handler handle_set_should_collect
    ;;
  list-exporters)
    handle_list_exporters
    ;;
  add-exporter)
    handle_add_exporter
    ;;
  update-exporter)
    handle_update_exporter
    ;;
  remove-exporter)
    handle_remove_exporter
    ;;
  list-extensions)
    handle_list_extensions
    ;;
  add-extension)
    handle_add_extension
    ;;
  update-extension)
    handle_update_extension
    ;;
  remove-extension)
    handle_remove_extension
    ;;
  list-processors)
    handle_list_processors
    ;;
  add-processor)
    handle_add_processor
    ;;
  update-processor)
    handle_update_processor
    ;;
  remove-processor)
    handle_remove_processor
    ;;
  refresh-config)
    handle_refresh_config
    ;;
  *)
    echo "❌ Unknown command: '$COMMAND'"
    usage
    ;;
esac

# --refresh: after a mutating command wrote config, apply any resulting stale
# config once. Read-only (list-*/get-*) and refresh-config (applies via its own
# handler) are intentionally excluded.
if [[ "$REFRESH" == true ]]; then
  case "$COMMAND" in
    set-should-collect|add-exporter|update-exporter|remove-exporter|\
    add-extension|update-extension|remove-extension|\
    add-processor|update-processor|remove-processor)
      apply_stale_config wait
      ;;
  esac
fi