#!/bin/bash
#
# Eon GCP Source/Restore Onboarding Script
# ----------------------------------------
# Creates or updates a "source" or "restore" environment in GCP:
#   - One "source" service account, custom roles
#   - Impersonation for scanning SA (in scanning project)
#   - Impersonation for an AWS Control Plane GCP SA (if you have set up a GCP SA to represent your AWS account)
#
# Usage (Source Example - Project):
#   ./gcp-eon-setup.sh \
#       --account-type Source \
#       --project-id my-gcp-project \
#       --eon-account-id 6b2e5764-a189-43ae-a173-8ad44f6bf160 \
#       --scanning-project-id rachel-scanning-project \
#       --control-plane-service-account <SVC_ACCOUNT_EMAIL> \
#       --verbose
#
# Usage (Source Example - Organization):
#   ./gcp-eon-setup.sh \
#       --account-type Source \
#       --organization-id 123456789123 \
#       --project-id management-project-within-org \
#       --eon-account-id 6b2e5764-a189-43ae-a173-8ad44f6bf160 \
#       --scanning-project-id rachel-scanning-project \
#       --control-plane-service-account <SVC_ACCOUNT_EMAIL> \
#       --verbose
#
# Usage (Source Example - Folder):
#   ./gcp-eon-setup.sh \
#       --account-type Source \
#       --organization-id 123456789123 \
#       --folder-id 987654321987 \
#       --project-id management-project-within-folder \
#       --eon-account-id 6b2e5764-a189-43ae-a173-8ad44f6bf160 \
#       --scanning-project-id rachel-scanning-project \
#       --control-plane-service-account <SVC_ACCOUNT_EMAIL> \
#       --verbose
#
# Usage (Restore Example):
#   ./gcp-eon-setup.sh \
#       --account-type Restore \
#       --project-id my-gcp-project \
#       --eon-account-id 6b2e5764-a189-43ae-a173-8ad44f6bf160 \
#       --control-plane-service-account <SVC_ACCOUNT_EMAIL> \
#       --host-project-id shared-vpc-host-project \
#       --verbose
#
# -E (errtrace) makes the ERR trap fire inside functions too — without it the
# trap would only see the main() call site, not the real failing command.
set -eE


ACCOUNT_TYPE=""
PROJECT_ID=""
SCANNING_PROJECT_ID=""
SA_NAME=""
SCANNING_SA_NAME=""
CONTROL_PLANE_SA=""    # GCP SA that your AWS control plane uses (via WIF or otherwise)
EON_ACCOUNT_ID=""
ORGANIZATION_ID=""     # GCP organization ID for organization-level deployments
FOLDER_ID=""           # GCP folder ID for folder-level deployments
ENABLE_GCS_NOTIFICATION_MANAGEMENT=false  # Enable GCS bucket notification management for CDC
ENABLE_GCS_INVENTORY_MANAGEMENT=true      # Enable GCS inventory management for inventory backup
ENABLE_BIGQUERY_AUTO_CHANGE_HISTORY=false  # Enable automatic BigQuery change history enablement
RESTRICT_TARGET_RESTORE_GCS_BY_TAG="${RESTRICT_TARGET_RESTORE_GCS_BY_TAG:-false}"  # Require eon-allow-restore:true tag on target GCS restore buckets
EXCLUDE_PROJECT_PATTERNS=""  # Comma-separated glob patterns for project IDs to exclude
NODE_SA_GROUP_EMAIL=""       # Google Group email for node SAs (org/folder restore)
USER_PROVIDED_NODE_SA_GROUP_EMAIL=false
INCLUDE_LABEL_FILTERS=""  # Comma-separated label filters to include (e.g., "env=prod,team=data,temp")
EXCLUDE_LABEL_FILTERS=""  # Comma-separated label filters to exclude (e.g., "env=staging,sandbox")
INCLUDE_TAG_FILTERS=""    # Comma-separated RM tag filters to include (e.g., "env/production,team/data")
EXCLUDE_TAG_FILTERS=""    # Comma-separated RM tag filters to exclude (e.g., "env/staging,sandbox/true")
VERBOSE=false
DESTROY=false           # --destroy: tear down resources instead of creating them
DRY_RUN=false           # --dry-run: with --destroy, print actions without executing
DESTROY_REMOVED=0       # count of resources actually deleted (for the summary)
DESTROY_FAILURES=0      # count of genuine deletion failures (not "already gone")
EON_GCP_SOURCE_VERSION="1.5.15"
EON_GCP_RESTORE_VERSION="1.7.4"
FIRESTORE_LOCATION="nam5"
# Resource type toggles - set via environment variables. On a source account ENABLE_<TYPE>_BACKUP sets a type's backup
# and ENABLE_<TYPE> is its deprecated alias (resolve_source_backup_toggles); restore accounts read ENABLE_<TYPE>. Whether
# an alias was set is captured before its default fills it in, so a conflict with the new name can be told apart.
BACKUP_ALIASES_SET=" ${ENABLE_GCS:+ENABLE_GCS} ${ENABLE_GCE:+ENABLE_GCE} ${ENABLE_CLOUDSQL:+ENABLE_CLOUDSQL} ${ENABLE_BIGQUERY:+ENABLE_BIGQUERY} ${ENABLE_SAP_HANA:+ENABLE_SAP_HANA} "
ENABLE_GCS="${ENABLE_GCS:-true}"
ENABLE_GCE="${ENABLE_GCE:-true}"
ENABLE_GCE_NATIVE_BACKUP="${ENABLE_GCE_NATIVE_BACKUP:-true}"
ENABLE_CLOUDSQL="${ENABLE_CLOUDSQL:-true}"
ENABLE_BIGQUERY="${ENABLE_BIGQUERY:-true}"
ENABLE_SAP_HANA="${ENABLE_SAP_HANA:-false}"
# Per-type discovery toggles (source only). A type's backup toggle above gates its backup roles, which need its
# discovery. SAP HANA has no discovery of its own: its VMs are discovered through Compute Engine. Which ones the caller
# set, even to an empty value, is captured before the defaults fill them in, so a restore account refuses any it was
# given.
DISCOVERY_TOGGLES_SET=" ${ENABLE_GCE_DISCOVERY+ENABLE_GCE_DISCOVERY} ${ENABLE_GCS_DISCOVERY+ENABLE_GCS_DISCOVERY} ${ENABLE_CLOUDSQL_DISCOVERY+ENABLE_CLOUDSQL_DISCOVERY} ${ENABLE_BIGQUERY_DISCOVERY+ENABLE_BIGQUERY_DISCOVERY} "
ENABLE_GCE_DISCOVERY="${ENABLE_GCE_DISCOVERY:-true}"
ENABLE_GCS_DISCOVERY="${ENABLE_GCS_DISCOVERY:-true}"
ENABLE_CLOUDSQL_DISCOVERY="${ENABLE_CLOUDSQL_DISCOVERY:-true}"
ENABLE_BIGQUERY_DISCOVERY="${ENABLE_BIGQUERY_DISCOVERY:-true}"
ENABLE_GWS="${ENABLE_GWS:-false}"
# Set true for the first (primary) Eon restore account in this project; it
# provisions the resources shared by all Eon restore accounts in the project.
# Set false for any additional restore account in a project that already has one,
# so it reuses the existing shared resources.
IS_PRIMARY_RESTORE_ACCOUNT="${IS_PRIMARY_RESTORE_ACCOUNT:-true}"

####
# Colors
end="\033[0m"
green="\033[0;32m"
blue="\033[0;34m"
red="\033[0;31m"

##############################################
# 2) Logging + Helper Functions
##############################################
log_info()    { echo -e "\033[1;34m[INFO] $1\033[0m"; }
log_success() { echo -e "\033[1;32m[SUCCESS] $1\033[0m"; }
log_error()   { echo -e "\033[1;31m[ERROR] $1\033[0m"; }
log_warn()    { echo -e "\033[1;33m[WARN] $1\033[0m"; }
log_verbose() {
    if [ "$VERBOSE" = true ]; then
        echo -e "\033[1;36m[VERBOSE] $1\033[0m"
    fi
}

# Make partial failures unmistakable. The script runs many steps under `set -e`
# (and _gcloud calls `exit` on gcloud errors); without this an error scrolls past
# in long output and the operator assumes onboarding completed when it didn't
# (e.g. the Firestore DB was never created). The ERR trap records where it broke;
# the EXIT trap prints a loud FAILED banner on any non-zero exit.
EON_FAILED_LINE=0
EON_FAILED_CMD=""
on_err() {
    EON_FAILED_LINE="$1"
    EON_FAILED_CMD="$2"
}
on_exit() {
    local rc=$?
    if [ "$rc" -ne 0 ]; then
        echo
        echo -e "\033[1;31m================================================================================\033[0m"
        # The banner has to match the operation: in --destroy mode a non-zero exit
        # means teardown was incomplete, not that anything was created.
        if [ "$DESTROY" = true ]; then
            echo -e "\033[1;31m❌ EON GCP DESTROY FAILED (exit $rc)\033[0m"
            if [ "$EON_FAILED_LINE" -ne 0 ]; then
                echo -e "\033[1;31m   Failed at line $EON_FAILED_LINE: $EON_FAILED_CMD\033[0m"
            fi
            echo -e "\033[1;31m   Destroy did NOT complete — some resources may not have been removed.\033[0m"
            echo -e "\033[1;31m   Fix the error above and re-run; the script is idempotent and will skip\033[0m"
            echo -e "\033[1;31m   resources that are already gone.\033[0m"
        else
            echo -e "\033[1;31m❌ EON GCP ONBOARDING FAILED (exit $rc)\033[0m"
            if [ "$EON_FAILED_LINE" -ne 0 ]; then
                echo -e "\033[1;31m   Failed at line $EON_FAILED_LINE: $EON_FAILED_CMD\033[0m"
            fi
            echo -e "\033[1;31m   Onboarding did NOT complete — earlier resources may be partially created.\033[0m"
            echo -e "\033[1;31m   Fix the error above and re-run; the script is idempotent and will skip\033[0m"
            echo -e "\033[1;31m   resources that already exist.\033[0m"
        fi
        echo -e "\033[1;31m================================================================================\033[0m"
    fi
}
trap 'on_err "$LINENO" "$BASH_COMMAND"' ERR
trap on_exit EXIT

# Resolves each source backup toggle into ENABLE_<TYPE>, which the rest of the script reads: ENABLE_<TYPE>_BACKUP wins,
# then the deprecated ENABLE_<TYPE>, then the default already in ENABLE_<TYPE>.
resolve_source_backup_toggles() {
  local legacy_var backup_var
  for legacy_var in ENABLE_GCE ENABLE_GCS ENABLE_CLOUDSQL ENABLE_BIGQUERY ENABLE_SAP_HANA; do
    backup_var="${legacy_var}_BACKUP"
    [[ -z "${!backup_var:-}" ]] && continue
    if [[ "$ACCOUNT_TYPE" != "source" ]]; then
      log_error "Error: $backup_var applies to Source accounts only; Restore accounts read $legacy_var."
      exit 1
    fi
    # Backup roles are created only for exactly "true", so any other value would silently leave the type unprotected.
    if [[ "${!backup_var}" != "true" && "${!backup_var}" != "false" ]]; then
      log_error "Error: $backup_var must be 'true' or 'false' (got '${!backup_var}')."
      exit 1
    fi
    if [[ "$BACKUP_ALIASES_SET" == *" $legacy_var "* && "${!legacy_var}" != "${!backup_var}" ]]; then
      log_error "Error: $backup_var=${!backup_var} conflicts with its deprecated alias $legacy_var=${!legacy_var}. Set only $backup_var."
      exit 1
    fi
    printf -v "$legacy_var" '%s' "${!backup_var}"
  done
}

# Prints the name the user set a source type's backup through, so an error names the variable they can see in their
# command: the deprecated alias when only it was set, otherwise ENABLE_<TYPE>_BACKUP.
source_backup_var_name() {
  local legacy_var="$1"
  local backup_var="${legacy_var}_BACKUP"
  if [[ -z "${!backup_var:-}" && "$BACKUP_ALIASES_SET" == *" $legacy_var "* ]]; then
    echo "$legacy_var"
  else
    echo "$backup_var"
  fi
}

# Build GCP service account description as JSON.
# Args: version method [var_name var_value ...]
# The description is capped at 256 characters, so only values that differ from the default are written: a *_disc var
# (discovery, default on) only when "false", every other var only when "true".
build_gcp_description_json() {
  local version="$1"
  local method="$2"
  shift 2
  
  # Build vars object from remaining key-value pairs
  local vars_json="{}"
  if [ $# -gt 0 ]; then
    # Build vars JSON object using jq with dynamic key-value pairs
    while [ $# -gt 0 ]; do
      local var_name="$1"
      local var_value="$2"
      shift 2
      if [[ "$var_name" == *_disc ]]; then
        [[ "$var_value" != "false" ]] && continue
      else
        [[ "$var_value" == "false" || -z "$var_value" ]] && continue
      fi
      vars_json=$(echo "$vars_json" | jq --arg key "$var_name" --arg val "$var_value" '. + {($key): $val}')
    done
  fi

  # Build final JSON structure
  jq -c -n \
    --arg version "$version" \
    --arg method "$method" \
    --argjson vars "$vars_json" \
    '{version: $version, method: $method, vars: $vars}'
}

show_help() {
  log_info "Usage: $0 [options]"
  echo ""
  echo "Options:"
  echo "  --account-type <Source|Restore>               (Required) Which account type logic to run."
  echo "  --project-id <PROJECT_ID>                     (Required) GCP project to hold roles/SA."
  echo "  --eon-account-id <STRING>                     (Required) Registered account ID to prevent confused deputy."
  echo "  --control-plane-service-account <EMAIL>       (Required) The GCP Workload‑Identity SA email that will impersonate the Eon SA."
  echo "  --scanning-project-id <SCAN_PROJ>             (Required for Source) GCP project that contains the eon‑scan‑* service account."
  echo "  --organization-id <ORG_ID>                    (Optional) GCP organization ID for organization-level deployments."
  echo "  --folder-id <FOLDER_ID>                       (Optional) GCP folder ID for folder-level deployments. Requires organization_id to be set."
  echo "  --exclude-project-patterns <PATTERNS>         (Optional) Comma-separated glob patterns for project IDs to exclude (e.g., internal-*,test-*)."
  echo "  --node-sa-group-email <EMAIL>                (Optional for Restore) Google Group email for node SAs. Auto-created from org domain if not provided for org/folder scope."
  echo "  --include-label-filters <FILTERS>             (Optional) Comma-separated label filters to include projects (e.g., env=prod,team=data,temp)."
  echo "                                                  Format: key=value or key (key-only matches any value). Projects must match at least one filter."
  echo "  --exclude-label-filters <FILTERS>             (Optional) Comma-separated label filters to exclude projects (e.g., env=staging,sandbox)."
  echo "                                                  Format: key=value or key (key-only matches any value). Projects matching any filter are excluded."
  echo "  --include-tag-filters <FILTERS>               (Optional) Comma-separated Resource Manager tag filters to include projects (e.g., env/production,team/data)."
  echo "                                                  Format: tagKey/tagValue. Projects must match at least one filter."
  echo "  --exclude-tag-filters <FILTERS>               (Optional) Comma-separated Resource Manager tag filters to exclude projects (e.g., env/staging)."
  echo "                                                  Format: tagKey/tagValue. Projects matching any filter are excluded."
  echo "  --enable-gcs-notification-management          (Optional for Source) Allow Eon to manage GCS bucket notifications for CDC."
  echo "  --no-gcs-inventory-management                 (Optional for Source) Disable Eon from managing GCS bucket inventory reports for inventory backup (enabled by default)."
  echo "  --enable-bigquery-auto-change-history          (Optional for Source) Allow Eon to automatically enable BigQuery change history."
  echo "  --enable-gws                                  (Optional for Restore) Enable Google Workspace restore permissions."
  echo "  --host-project-id <HOST_PROJECT>              (Optional for Restore) Host project ID for Shared VPC."
  echo "  --firestore-location <LOCATION>               (Optional for Restore) Firestore location for Eon command results."
  echo "  --restrict-target-restore-gcs-by-tag          (Optional for Restore) Allow GCS restore writes only to buckets tagged eon-allow-restore:true."
  echo "  --destroy                                     Delete the Eon resources this script created for the account (reverse of onboarding)."
  echo "  --dry-run                                     With --destroy: print exactly what would be deleted, deleting nothing."
  echo "  --verbose                                     Print verbose logs."
  echo "  --help                                        Show help."
  echo ""
  echo "Environment variables (true|false):"
  echo "  Source backup:    ENABLE_GCE_BACKUP, ENABLE_GCS_BACKUP, ENABLE_CLOUDSQL_BACKUP, ENABLE_BIGQUERY_BACKUP (default true),"
  echo "                    ENABLE_SAP_HANA_BACKUP (default false), ENABLE_GCE_NATIVE_BACKUP (default true)."
  echo "                    ENABLE_GCE, ENABLE_GCS, ENABLE_CLOUDSQL, ENABLE_BIGQUERY and ENABLE_SAP_HANA are deprecated aliases."
  echo "  Source discovery: ENABLE_GCE_DISCOVERY, ENABLE_GCS_DISCOVERY, ENABLE_CLOUDSQL_DISCOVERY, ENABLE_BIGQUERY_DISCOVERY"
  echo "                    (default true). A type's backup requires its discovery; SAP HANA backup requires ENABLE_GCE_DISCOVERY."
  echo "  Restore:          ENABLE_GCE, ENABLE_GCS, ENABLE_CLOUDSQL, ENABLE_BIGQUERY (default true), ENABLE_SAP_HANA (default false);"
  echo "                    the *_BACKUP and *_DISCOVERY toggles are refused."
  echo ""
  log_info "Example (Source - Project):"
  echo "  $0 --account-type Source --project-id my-gcp-project \\"
  echo "     --scanning-project-id rachel-scanning-project --control-plane-service-account my-aws-controlplane@my-gcp-project.iam.gserviceaccount.com \\"
  echo "     --eon-account-id 6b2e5764-a189-43ae-a173-8ad44f6bf160 --verbose"
  echo ""
  log_info "Example (Source - Organization):"
  echo "  $0 --account-type Source --organization-id my-gcp-org --project-id management-project-within-org \\"
  echo "     --scanning-project-id rachel-scanning-project --control-plane-service-account my-aws-controlplane@my-gcp-project.iam.gserviceaccount.com \\"
  echo "     --eon-account-id 6b2e5764-a189-43ae-a173-8ad44f6bf160 --verbose"
  echo ""
  log_info "Example (Source - Folder):"
  echo "  $0 --account-type Source --organization-id my-gcp-org --folder-id my-folder --project-id management-project-within-folder \\"
  echo "     --scanning-project-id rachel-scanning-project --control-plane-service-account my-aws-controlplane@my-gcp-project.iam.gserviceaccount.com \\"
  echo "     --eon-account-id 6b2e5764-a189-43ae-a173-8ad44f6bf160 --verbose"
  echo ""
  log_info "Example (Restore - Project):"
  echo "  $0 --account-type Restore --project-id my-gcp-project \\"
  echo "     --control-plane-service-account my-aws-controlplane@my-gcp-project.iam.gserviceaccount.com \\"
  echo "     --eon-account-id 6b2e5764-a189-43ae-a173-8ad44f6bf160 \\"
  echo "     --host-project-id shared-vpc-host-project --verbose"
  echo ""
  log_info "Example (Restore - Organization):"
  echo "  $0 --account-type Restore --organization-id 123456789123 --project-id management-project-within-org \\"
  echo "     --control-plane-service-account my-aws-controlplane@my-gcp-project.iam.gserviceaccount.com \\"
  echo "     --eon-account-id 6b2e5764-a189-43ae-a173-8ad44f6bf160 --verbose"
  echo ""
  log_info "Example (Restore - Folder):"
  echo "  $0 --account-type Restore --organization-id 123456789123 --folder-id 987654321987 --project-id management-project-within-folder \\"
  echo "     --control-plane-service-account my-aws-controlplane@my-gcp-project.iam.gserviceaccount.com \\"
  echo "     --eon-account-id 6b2e5764-a189-43ae-a173-8ad44f6bf160 --verbose"
  echo ""
  log_info "Example (Destroy - Source, dry run first):"
  echo "  $0 --account-type Source --destroy --dry-run --project-id my-gcp-project \\"
  echo "     --scanning-project-id rachel-scanning-project --control-plane-service-account my-aws-controlplane@my-gcp-project.iam.gserviceaccount.com \\"
  echo "     --eon-account-id 6b2e5764-a189-43ae-a173-8ad44f6bf160"
  echo ""
  log_info "Example (Destroy - Restore, organization):"
  echo "  $0 --account-type Restore --destroy --organization-id 123456789123 --project-id management-project-within-org \\"
  echo "     --control-plane-service-account my-aws-controlplane@my-gcp-project.iam.gserviceaccount.com \\"
  echo "     --eon-account-id 6b2e5764-a189-43ae-a173-8ad44f6bf160"
}

parse_arguments() {
  while [[ "$#" -gt 0 ]]; do
    case "$1" in
      --account-type)
        if [[ -n "$2" && ! "$2" =~ ^-- ]]; then
          ACCOUNT_TYPE=$(echo "$2" | tr '[:upper:]' '[:lower:]')
          shift 2
        else
          log_error "Error: --account-type requires a value (Source|Restore)."
          show_help
          exit 1
        fi
        ;;
      --project-id)
        if [[ -n "$2" && ! "$2" =~ ^-- ]]; then
          PROJECT_ID="$2"
          shift 2
        else
          log_error "Error: --project-id requires a value."
          show_help
          exit 1
        fi
        ;;
      --organization-id)
        if [[ -n "$2" && ! "$2" =~ ^-- ]]; then
          ORGANIZATION_ID="$2"
          shift 2
        else
          log_error "Error: --organization-id requires a value."
          show_help
          exit 1
        fi
        ;;
      --folder-id)
        if [[ -n "$2" && ! "$2" =~ ^-- ]]; then
          FOLDER_ID="$2"
          shift 2
        else
          log_error "Error: --folder-id requires a value."
          show_help
          exit 1
        fi
        ;;
      --scanning-project-id)
        if [[ -n "$2" && ! "$2" =~ ^-- ]]; then
          SCANNING_PROJECT_ID="$2"
          shift 2
        else
          log_error "Error: --scanning-project-id requires a value."
          show_help
          exit 1
        fi
        ;;
      --control-plane-service-account)
        if [[ -n "$2" && ! "$2" =~ ^-- ]]; then
          CONTROL_PLANE_SA="$2"
          shift 2
        else
          log_error "Error: --control-plane-service-account requires a value."
          show_help
          exit 1
        fi
        ;;
      --eon-account-id)
        if [[ -n "$2" && ! "$2" =~ ^-- ]]; then
          EON_ACCOUNT_ID="$2"
          shift 2
        else
          log_error "Error: --eon-account-id requires a value."
          show_help
          exit 1
        fi
        ;;
      --enable-gcs-notification-management)
        ENABLE_GCS_NOTIFICATION_MANAGEMENT=true
        shift
        ;;
      --no-gcs-inventory-management)
        ENABLE_GCS_INVENTORY_MANAGEMENT=false
        shift
        ;;
      --enable-bigquery-auto-change-history)
        ENABLE_BIGQUERY_AUTO_CHANGE_HISTORY=true
        shift
        ;;
      --enable-gws)
        ENABLE_GWS=true
        shift
        ;;
      --restrict-target-restore-gcs-by-tag)
        RESTRICT_TARGET_RESTORE_GCS_BY_TAG=true
        shift
        ;;
      --exclude-project-patterns)
        if [[ -n "$2" && ! "$2" =~ ^-- ]]; then
          EXCLUDE_PROJECT_PATTERNS="$2"
          shift 2
        else
          log_error "Error: --exclude-project-patterns requires a value."
          show_help
          exit 1
        fi
        ;;
      --node-sa-group-email)
        if [[ -n "$2" && ! "$2" =~ ^-- ]]; then
          NODE_SA_GROUP_EMAIL="$2"
          USER_PROVIDED_NODE_SA_GROUP_EMAIL=true
          shift 2
        else
          log_error "Error: --node-sa-group-email requires a value."
          show_help
          exit 1
        fi
        ;;
      --include-label-filters)
        if [[ -n "$2" && ! "$2" =~ ^-- ]]; then
          INCLUDE_LABEL_FILTERS="$2"
          shift 2
        else
          log_error "Error: --include-label-filters requires a value."
          show_help
          exit 1
        fi
        ;;
      --exclude-label-filters)
        if [[ -n "$2" && ! "$2" =~ ^-- ]]; then
          EXCLUDE_LABEL_FILTERS="$2"
          shift 2
        else
          log_error "Error: --exclude-label-filters requires a value."
          show_help
          exit 1
        fi
        ;;
      --include-tag-filters)
        if [[ -n "$2" && ! "$2" =~ ^-- ]]; then
          INCLUDE_TAG_FILTERS="$2"
          shift 2
        else
          log_error "Error: --include-tag-filters requires a value."
          show_help
          exit 1
        fi
        ;;
      --exclude-tag-filters)
        if [[ -n "$2" && ! "$2" =~ ^-- ]]; then
          EXCLUDE_TAG_FILTERS="$2"
          shift 2
        else
          log_error "Error: --exclude-tag-filters requires a value."
          show_help
          exit 1
        fi
        ;;
      --host-project-id)
        if [[ -n "$2" && ! "$2" =~ ^-- ]]; then
          HOST_PROJECT_ID="$2"
          shift 2
        else
          log_error "Error: --host-project-id requires a value."
          show_help
          exit 1
        fi
        ;;
      --firestore-location)
        if [[ -n "$2" && ! "$2" =~ ^-- ]]; then
          FIRESTORE_LOCATION="$2"
          shift 2
        else
          log_error "Error: --firestore-location requires a value."
          show_help
          exit 1
        fi
        ;;
      --destroy)
        DESTROY=true
        shift
        ;;
      --dry-run)
        DRY_RUN=true
        shift
        ;;
      --verbose)
        VERBOSE=true
        shift
        ;;
      --help)
        show_help
        exit 0
        ;;
      *)
        log_error "Unknown parameter: $1"
        show_help
        exit 1
        ;;
    esac
  done
}

function blue {
    echo -e "${blue}${*}${end}"
}

function green {
    echo -e "${green}${*}${end}"
}

function red {
    echo -e "${red}${*}${end}"
}


validate_input() {
  # Must have an account type
  if [[ "$ACCOUNT_TYPE" != "source" && "$ACCOUNT_TYPE" != "restore" ]]; then
    log_error "Error: --account-type must be 'Source' or 'Restore'."
    show_help
    exit 1
  fi

  # --dry-run only previews a teardown; without --destroy it would silently fall
  # through to a live create run, which is never what the user intended.
  if [[ "$DRY_RUN" == "true" && "$DESTROY" != "true" ]]; then
    log_error "Error: --dry-run can only be used together with --destroy."
    show_help
    exit 1
  fi

  if [[ -z "$EON_ACCOUNT_ID" ]]; then
    log_error "Error: --eon-account-id is required."
    exit 1
  fi

  if [[ -z "$PROJECT_ID" ]]; then
    log_error "Error: --project-id is required."
    show_help
    exit 1
  fi

  if [[ "$DESTROY" != "true" && -z "$CONTROL_PLANE_SA" ]]; then
    log_error "Error: --control-plane-service-account is required."
    show_help
    exit 1
  fi

  # For source account type, scanning-project-id is required
  if [[ "$ACCOUNT_TYPE" == "source" && -z "$SCANNING_PROJECT_ID" ]]; then
    log_error "Error: --scanning-project-id is required for Source account type."
    show_help
    exit 1
  fi

  # For source account type, host-project-id should not be provided
  if [[ "$ACCOUNT_TYPE" == "source" && -n "$HOST_PROJECT_ID" ]]; then
    log_error "Error: --host-project-id can only be used with Restore account type."
    show_help
    exit 1
  fi

  # The GWS restore SA only makes sense for a Restore account.
  if [[ "$ACCOUNT_TYPE" == "source" && "$ENABLE_GWS" == "true" ]]; then
    log_error "Error: --enable-gws can only be used with Restore account type."
    show_help
    exit 1
  fi

  if [[ "$ACCOUNT_TYPE" == "source" && "$RESTRICT_TARGET_RESTORE_GCS_BY_TAG" == "true" ]]; then
    log_error "Error: --restrict-target-restore-gcs-by-tag can only be used with Restore account type."
    show_help
    exit 1
  fi

  # A Google Workspace restore writes only into Google Workspace, so Eon does not
  # request storage, compute, or database restore permissions you would not use.
  if [[ "$ENABLE_GWS" == "true" ]]; then
    ENABLE_GCS=false
    ENABLE_GCE=false
    ENABLE_GCE_NATIVE_BACKUP=false
    ENABLE_CLOUDSQL=false
    ENABLE_BIGQUERY=false
    RESTRICT_TARGET_RESTORE_GCS_BY_TAG=false
  fi

  # When folder ID is provided, organization ID is required
  if [[ -n "$FOLDER_ID" && -z "$ORGANIZATION_ID" ]]; then
    log_error "Error: --organization-id is required when --folder-id is provided."
    show_help
    exit 1
  fi

  # Any value other than "true" or "false" is treated as false, which would skip
  # creating the shared resources. Require an explicit value to avoid silently
  # misconfiguring the account.
  if [[ "$IS_PRIMARY_RESTORE_ACCOUNT" != "true" && "$IS_PRIMARY_RESTORE_ACCOUNT" != "false" ]]; then
    log_error "Error: IS_PRIMARY_RESTORE_ACCOUNT must be 'true' or 'false' (got '$IS_PRIMARY_RESTORE_ACCOUNT')."
    exit 1
  fi

  if [[ "$RESTRICT_TARGET_RESTORE_GCS_BY_TAG" != "true" && "$RESTRICT_TARGET_RESTORE_GCS_BY_TAG" != "false" ]]; then
    log_error "Error: RESTRICT_TARGET_RESTORE_GCS_BY_TAG must be 'true' or 'false' (got '$RESTRICT_TARGET_RESTORE_GCS_BY_TAG')."
    exit 1
  fi

  # The description records a discovery toggle only when it is exactly "false", so any other value would be set up as
  # off but recorded as on.
  local discovery_var
  for discovery_var in ENABLE_GCE_DISCOVERY ENABLE_GCS_DISCOVERY ENABLE_CLOUDSQL_DISCOVERY ENABLE_BIGQUERY_DISCOVERY; do
    if [[ "$ACCOUNT_TYPE" != "source" && "$DISCOVERY_TOGGLES_SET" == *" $discovery_var "* ]]; then
      log_error "Error: $discovery_var applies to Source accounts only; Restore accounts have no discovery toggles."
      exit 1
    fi
    if [[ "${!discovery_var}" != "true" && "${!discovery_var}" != "false" ]]; then
      log_error "Error: $discovery_var must be 'true' or 'false' (got '${!discovery_var}')."
      exit 1
    fi
  done

  resolve_source_backup_toggles

  # Eon backs up only resources it has discovered, so a type's backup roles are useless without its discovery roles.
  # Teardown removes every role whatever the toggles say, so it skips the check.
  if [[ "$ACCOUNT_TYPE" == "source" && "$DESTROY" != "true" ]]; then
    local backup_var
    for backup_var in ENABLE_GCE ENABLE_GCS ENABLE_CLOUDSQL ENABLE_BIGQUERY; do
      discovery_var="${backup_var}_DISCOVERY"
      if [[ "${!backup_var}" == "true" && "${!discovery_var}" != "true" ]]; then
        log_error "Error: $(source_backup_var_name "$backup_var")=true requires $discovery_var=true: Eon backs up only resources it discovers."
        exit 1
      fi
    done
    if [[ "$ENABLE_SAP_HANA" == "true" && "$ENABLE_GCE_DISCOVERY" != "true" ]]; then
      log_error "Error: $(source_backup_var_name ENABLE_SAP_HANA)=true requires ENABLE_GCE_DISCOVERY=true: SAP HANA VMs are discovered through Compute Engine."
      exit 1
    fi
  fi
}

check_gcloud_installed() {
  if command -v gcloud &> /dev/null; then
      log_success "✔ gcloud CLI is already installed."
  else
      log_error "⚠ gcloud CLI is not installed. Please install manually and re-run the script"
      exit 1
  fi
}

login_to_gcp() {
  set +e
  local activeAcct
  activeAcct=$(gcloud auth list --filter=status:ACTIVE --format="value(account)" 2>/dev/null)
  set -e

  if [[ -n "$activeAcct" ]]; then
    log_info "Already logged into GCP as: $activeAcct. Skipping 'gcloud auth login'."
    return
  fi

  log_info "No active GCP account found. Logging into GCP..."
  set +e
  gcloud auth login --quiet
  local retVal=$?
  set -e

  if [ $retVal -ne 0 ]; then
    log_error "❌ GCP login failed. Please check your credentials."
    exit 1
  fi
  log_success "✔ Successfully logged into GCP."
}

refresh_gcp_auth() {
  log_info "Refreshing GCP authentication..."
  set +e
  gcloud auth print-access-token --quiet >/dev/null 2>&1
  local retVal=$?
  set -e

  if [ $retVal -ne 0 ]; then
    log_warn "Token refresh failed, attempting re-authentication..."
    set +e
    gcloud auth login --quiet
    retVal=$?
    set -e
    if [ $retVal -ne 0 ]; then
      log_error "❌ Re-authentication failed. Please run 'gcloud auth login' manually."
      exit 1
    fi
  fi
  log_success "✔ GCP authentication refreshed."
}

is_token_expired_error() {
  local error_output="$1"
  if echo "$error_output" | grep -qiE "token.*expired|token.*invalid|credentials.*expired|unauthorized|unauthenticated|UNAUTHENTICATED|invalid_grant"; then
    return 0
  fi
  return 1
}

is_billing_error() {
  local error_output="$1"
  if echo "$error_output" | grep -qiE "billing.*not.*enabled|billing.*disabled|billing.*account.*not|enable.*billing|BILLING_DISABLED|billingNotEnabled|budget.*limit|requires.*billing"; then
    return 0
  fi
  return 1
}

matches_any_glob_pattern() {
  local project_id="$1"
  local patterns="$2"
  
  if [[ -z "$patterns" ]]; then
    return 1
  fi
  
  IFS=',' read -ra pattern_array <<< "$patterns"
  for pattern in "${pattern_array[@]}"; do
    pattern=$(echo "$pattern" | xargs)
    if [[ -z "$pattern" ]]; then
      continue
    fi
    if [[ "$project_id" == $pattern ]]; then
      return 0
    fi
  done
  return 1
}

filter_excluded_projects() {
  local projects="$1"
  local exclude_patterns="$2"
  
  if [[ -z "$exclude_patterns" ]]; then
    echo "$projects"
    return
  fi
  
  local filtered_projects=""
  while IFS= read -r project; do
    [[ -z "$project" ]] && continue
    if matches_any_glob_pattern "$project" "$exclude_patterns"; then
      log_verbose "Excluding project: $project (matches exclusion pattern)" >&2
    else
      if [[ -n "$filtered_projects" ]]; then
        filtered_projects="${filtered_projects}"$'\n'"${project}"
      else
        filtered_projects="$project"
      fi
    fi
  done <<< "$projects"
  
  echo "$filtered_projects"
}

project_matches_label_filter() {
  local labels_json="$1"
  local filter_key="$2"
  local filter_value="$3"

  local key_exists
  key_exists=$(echo "$labels_json" | jq --arg k "$filter_key" 'if .labels then .labels | has($k) else false end' 2>/dev/null)

  if [[ "$key_exists" != "true" ]]; then
    return 1
  fi

  if [[ -z "$filter_value" ]]; then
    return 0
  fi

  local label_val
  label_val=$(echo "$labels_json" | jq -r --arg k "$filter_key" '.labels[$k]' 2>/dev/null)

  if [[ "$label_val" == "$filter_value" ]]; then
    return 0
  fi

  return 1
}

project_matches_any_label_filter() {
  local labels_json="$1"
  local filters="$2"

  IFS=',' read -ra filter_array <<< "$filters"
  for filter in "${filter_array[@]}"; do
    filter=$(echo "$filter" | xargs)
    [[ -z "$filter" ]] && continue

    local key value
    if [[ "$filter" == *"="* ]]; then
      key="${filter%%=*}"
      value="${filter#*=}"
    else
      key="$filter"
      value=""
    fi

    if project_matches_label_filter "$labels_json" "$key" "$value"; then
      return 0
    fi
  done
  return 1
}

# Filter projects by labels and Resource Manager tags with OR semantics.
# - Within each group (labels or tags), filters are OR-combined.
# - Across the two groups, includes are OR-combined (a project survives if it
#   matches at least one include label filter OR at least one include tag filter).
# - Excludes are OR-combined across both groups (any match removes the project).
filter_projects_by_labels_and_tags() {
  local projects="$1"
  local include_label_filters="$2"
  local exclude_label_filters="$3"
  local include_tag_filters="$4"
  local exclude_tag_filters="$5"

  if [[ -z "$include_label_filters" && -z "$exclude_label_filters" && -z "$include_tag_filters" && -z "$exclude_tag_filters" ]]; then
    echo "$projects"
    return
  fi

  local has_includes=""
  if [[ -n "$include_label_filters" || -n "$include_tag_filters" ]]; then
    has_includes="true"
  fi

  local filtered_projects=""
  while IFS= read -r project; do
    [[ -z "$project" ]] && continue

    # One describe call fetches both labels and projectNumber. Failures
    # (deleted project, IAM gap, transient API error) are logged with the
    # project ID and gcloud's error text instead of being silently
    # swallowed — otherwise include-filtered runs would drop projects
    # without any user-visible signal.
    local labels_json project_number bindings_json
    local describe_out describe_rc=0
    describe_out=$(gcloud projects describe "$project" --format="json(labels,projectNumber)" 2>&1) || describe_rc=$?
    if [[ $describe_rc -ne 0 ]]; then
      log_warn "Failed to describe project '$project'; treating as having no labels/tags. gcloud said: ${describe_out}" >&2
      labels_json='{"labels":{}}'
      project_number=""
    else
      labels_json=$(echo "$describe_out" | jq -c '{labels: (.labels // {})}')
      project_number=$(echo "$describe_out" | jq -r '.projectNumber // ""')
    fi
    bindings_json='[]'
    if [[ -n "$project_number" ]]; then
      local bindings_out bindings_rc=0
      bindings_out=$(gcloud resource-manager tags bindings list \
        --parent="//cloudresourcemanager.googleapis.com/projects/${project_number}" \
        --location=global --format=json 2>&1) || bindings_rc=$?
      if [[ $bindings_rc -ne 0 ]]; then
        log_warn "Failed to list tag bindings for project '$project'; treating as having no tags. gcloud said: ${bindings_out}" >&2
      else
        bindings_json="$bindings_out"
      fi
    fi

    if [[ -n "$has_includes" ]]; then
      local include_pass=""
      if [[ -n "$include_label_filters" ]] && project_matches_any_label_filter "$labels_json" "$include_label_filters"; then
        include_pass="true"
      fi
      if [[ -z "$include_pass" && -n "$include_tag_filters" ]] && project_matches_any_tag_filter "$bindings_json" "$include_tag_filters"; then
        include_pass="true"
      fi
      if [[ -z "$include_pass" ]]; then
        log_verbose "Excluding project: $project (does not match any include label or tag filter)" >&2
        continue
      fi
    fi

    if [[ -n "$exclude_label_filters" ]] && project_matches_any_label_filter "$labels_json" "$exclude_label_filters"; then
      log_verbose "Excluding project: $project (matches exclude label filter)" >&2
      continue
    fi
    if [[ -n "$exclude_tag_filters" ]] && project_matches_any_tag_filter "$bindings_json" "$exclude_tag_filters"; then
      log_verbose "Excluding project: $project (matches exclude tag filter)" >&2
      continue
    fi

    if [[ -n "$filtered_projects" ]]; then
      filtered_projects="${filtered_projects}"$'\n'"${project}"
    else
      filtered_projects="$project"
    fi
  done <<< "$projects"

  echo "$filtered_projects"
}

project_matches_tag_filter() {
  local bindings_json="$1"
  local filter_key="$2"
  local filter_value="$3"

  local match
  match=$(echo "$bindings_json" | jq -r --arg k "$filter_key" --arg v "$filter_value" '
    [.[] | .tagValueNamespacedName // "" | split("/") | if length >= 3 then {key: .[1], value: .[2]} else empty end]
    | any(.key == $k and .value == $v)
  ' 2>/dev/null)

  if [[ "$match" == "true" ]]; then
    return 0
  fi
  return 1
}

project_matches_any_tag_filter() {
  local bindings_json="$1"
  local filters="$2"

  IFS=',' read -ra filter_array <<< "$filters"
  for filter in "${filter_array[@]}"; do
    filter=$(echo "$filter" | xargs)
    [[ -z "$filter" ]] && continue

    local key value
    key="${filter%%/*}"
    value="${filter#*/}"

    if project_matches_tag_filter "$bindings_json" "$key" "$value"; then
      return 0
    fi
  done
  return 1
}

# Track projects with billing issues for summary at end
PROJECTS_WITH_BILLING_ISSUES=""


add_billing_issue_project() {
  local project="$1"
  if [[ -z "$PROJECTS_WITH_BILLING_ISSUES" ]]; then
    PROJECTS_WITH_BILLING_ISSUES="$project"
  elif ! echo "$PROJECTS_WITH_BILLING_ISSUES" | grep -q "$project"; then
    PROJECTS_WITH_BILLING_ISSUES="${PROJECTS_WITH_BILLING_ISSUES}, $project"
  fi
}


print_billing_issues_summary() {
  if [[ -n "$PROJECTS_WITH_BILLING_ISSUES" ]]; then
    echo ""
    log_warn "════════════════════════════════════════════════════════════════════════"
    log_warn "BILLING NOTICE: Some projects have billing disabled"
    log_warn "════════════════════════════════════════════════════════════════════════"
    log_warn "The following projects require billing to be enabled for full API access:"
    log_warn "  $PROJECTS_WITH_BILLING_ISSUES"
    log_warn ""
    log_warn "APIs like Compute Engine require an active billing account."
    log_warn "To enable billing, visit: https://console.cloud.google.com/billing"
    log_warn "Or re-run this script after billing is restored."
    log_warn "════════════════════════════════════════════════════════════════════════"
    echo ""
  fi
}


select_gcp_project() {
  log_info "Selecting GCP project: $PROJECT_ID"
  set +e
  output=$(_gcloud config set project "$PROJECT_ID" --quiet 2>&1)
  retVal=$?
  set -e
  if [ $retVal -ne 0 ]; then
    echo "$output"
    log_error "❌ Failed to select project $PROJECT_ID."
    exit 1
  fi
  log_success "✔ Selected GCP project: $PROJECT_ID."
}

_gcloud() {
  log_verbose "\$ gcloud $*"
  # Keep gcloud stdout ($output) and stderr ($output_err) separate: gcloud can emit
  # non-fatal warnings (e.g. its Regional Access Boundary probe) to stderr while still
  # exiting 0, and merging the streams corrupts downstream JSON parsing of $output.
  local err_file
  err_file=$(mktemp)
  set +e
  output=$(gcloud "$@" --quiet --verbosity=error 2>"$err_file")
  local retVal=$?
  set -e
  output_err=$(<"$err_file")
  rm -f "$err_file"

  if [ $retVal -ne 0 ]; then
    # Check if error is due to token expiration
    if is_token_expired_error "$output_err"; then
      refresh_gcp_auth

      # Retry the command after refreshing
      log_verbose "\$ gcloud $* (retry after auth refresh)"
      err_file=$(mktemp)
      set +e
      output=$(gcloud "$@" --quiet --verbosity=error 2>"$err_file")
      retVal=$?
      set -e
      output_err=$(<"$err_file")
      rm -f "$err_file"

      if [ $retVal -ne 0 ]; then
        [ -n "$output_err" ] && echo "$output_err"
        [ -n "$output" ] && echo "$output"
        log_error "Error, exiting."
        exit $retVal
      fi
    else
      [ -n "$output_err" ] && echo "$output_err"
      [ -n "$output" ] && echo "$output"
      log_error "Error, exiting."
      exit $retVal
    fi
  fi

  if [ "$VERBOSE" = true ]; then
    [ -n "$output_err" ] && echo "$output_err"
    echo "$output"
  fi
}

# _gcloud_try is the non-fatal sibling of _gcloud, used only by teardown.
# Unlike _gcloud it NEVER exits the script: it returns gcloud's exit code and
# leaves stdout in $output and stderr in $output_err. It preserves the
# token-expiry refresh + retry behavior so long teardowns survive credential expiry.
_gcloud_try() {
  log_verbose "\$ gcloud $*"
  local err_file
  err_file=$(mktemp)
  set +e
  output=$(gcloud "$@" --quiet --verbosity=error 2>"$err_file")
  local retVal=$?
  set -e
  output_err=$(<"$err_file")
  rm -f "$err_file"

  if [ $retVal -ne 0 ] && is_token_expired_error "$output_err"; then
    refresh_gcp_auth
    log_verbose "\$ gcloud $* (retry after auth refresh)"
    err_file=$(mktemp)
    set +e
    output=$(gcloud "$@" --quiet --verbosity=error 2>"$err_file")
    retVal=$?
    set -e
    output_err=$(<"$err_file")
    rm -f "$err_file"
  fi

  if [ "$VERBOSE" = true ]; then
    [ -n "$output_err" ] && echo "$output_err"
    [ -n "$output" ] && echo "$output"
  fi
  return $retVal
}

# is_not_found_error returns 0 when gcloud output indicates the target is
# already absent, so teardown can treat it as a benign skip rather than a failure.
is_not_found_error() {
  # Match resource-absent messages only. The bare phrase "could not find" is
  # intentionally excluded because gcloud also uses it for credential/project
  # config errors, which must NOT be treated as benign skips. "not found!" and
  # "does not contain the binding" cover remove-iam-policy-binding when the
  # binding is already gone.
  echo "$1" | grep -qiE 'NOT_FOUND|does not exist|was not found|not found!|does not contain the binding|Resource .* not found'
}

# log_warn_captured_output warns both captured gcloud streams (stderr then stdout,
# each only when present) so a best-effort failure never hides detail gcloud wrote
# to either stream.
log_warn_captured_output() {
  [ -n "$output_err" ] && log_warn "  ${output_err}"
  [ -n "$output" ] && log_warn "  ${output}"
}

# destroy_exec is the single chokepoint for every destructive gcloud call.
# Usage: destroy_exec "<human description>" -- <gcloud args...>
#   --dry-run     : prints "[DRY-RUN] would delete: <desc>" and returns 0.
#   not-found     : prints a skip line and returns 0 (idempotent).
#   real failure  : logs WARN, increments DESTROY_FAILURES, returns 0 (best-effort).
#   success       : increments DESTROY_REMOVED.
# Always returns 0 so a single failure never aborts the run under `set -e`.
destroy_exec() {
  local description="$1"
  shift
  if [[ "${1:-}" == "--" ]]; then
    shift
  fi

  if [[ "$DRY_RUN" == "true" ]]; then
    log_info "[DRY-RUN] would delete: $description"
    return 0
  fi

  if _gcloud_try "$@"; then
    DESTROY_REMOVED=$((DESTROY_REMOVED + 1))
    log_success "Deleted: $description"
    return 0
  fi

  if is_not_found_error "$output_err"; then
    log_info "Already absent (skip): $description"
    return 0
  fi

  DESTROY_FAILURES=$((DESTROY_FAILURES + 1))
  log_warn "Failed to delete: $description"
  log_warn_captured_output
  return 0
}

# delete_role deletes a custom role at project or organization scope (best-effort).
delete_role() {
  # Custom roles exist only at project or organization scope (never folder).
  local role_name="$1"
  local scope_type="$2"
  local scope_id="$3"
  if [[ "$scope_type" == "project" ]]; then
    destroy_exec "custom role $role_name (project $scope_id)" -- iam roles delete "$role_name" --project="$scope_id"
  elif [[ "$scope_type" == "organization" ]]; then
    destroy_exec "custom role $role_name (org $scope_id)" -- iam roles delete "$role_name" --organization="$scope_id"
  else
    log_warn "delete_role: invalid scope '$scope_type' for role '$role_name'"
  fi
}

# delete_sa deletes a service account (best-effort). project must be the SA's home project.
delete_sa() {
  local sa_email="$1"
  local project="$2"
  local existing_sa

  if _gcloud_try iam service-accounts list --project="$project" --filter="email:$sa_email" --format="value(email)"; then
    existing_sa="$(echo "$output" | grep -Fx "$sa_email" || true)"
    if [[ -z "$existing_sa" ]]; then
      log_info "Already absent (skip): service account $sa_email"
      return 0
    fi
  fi

  destroy_exec "service account $sa_email" -- iam service-accounts delete "$sa_email" --project="$project"
}

# remove_binding removes an IAM policy binding of <role_name> to <member> at the
# given scope. --all removes the binding regardless of any IAM condition the
# create path attached (e.g. RestrictEonSnapshots). Best-effort.
remove_binding() {
  local member="$1"
  local role_name="$2"
  local scope_type="${3:-project}"
  local scope_id="${4:-$PROJECT_ID}"
  local full_role
  case "$scope_type" in
    project)
      full_role="projects/$scope_id/roles/$role_name"
      destroy_exec "binding $full_role -> $member (project $scope_id)" -- \
        projects remove-iam-policy-binding "$scope_id" --member="$member" --role="$full_role" --all
      ;;
    organization)
      full_role="organizations/$scope_id/roles/$role_name"
      destroy_exec "binding $full_role -> $member (org $scope_id)" -- \
        organizations remove-iam-policy-binding "$scope_id" --member="$member" --role="$full_role" --all
      ;;
    folder)
      full_role="organizations/$ORGANIZATION_ID/roles/$role_name"
      destroy_exec "binding $full_role -> $member (folder $scope_id)" -- \
        resource-manager folders remove-iam-policy-binding "$scope_id" --member="$member" --role="$full_role" --all
      ;;
    *)
      log_warn "remove_binding: invalid scope '$scope_type' for role '$role_name'"
      ;;
  esac
}

remove_project_binding_with_full_role() {
  local member="$1"
  local full_role="$2"
  local project="${3:-$PROJECT_ID}"
  destroy_exec "binding $full_role -> $member (project $project)" -- \
    projects remove-iam-policy-binding "$project" --member="$member" --role="$full_role" --all
}

line_in_list() {
  local needle="$1"
  local haystack="$2"
  local line
  while IFS= read -r line; do
    if [[ "$line" == "$needle" ]]; then
      return 0
    fi
  done <<< "$haystack"
  return 1
}

get_policy_members_for_role() {
  local scope_type="$1"
  local scope_id="$2"
  local full_role="$3"

  case "$scope_type" in
    project)
      _gcloud_try projects get-iam-policy "$scope_id" \
        --flatten="bindings[].members" \
        --filter="bindings.role:${full_role}" \
        --format="value(bindings.members)"
      ;;
    organization)
      _gcloud_try organizations get-iam-policy "$scope_id" \
        --flatten="bindings[].members" \
        --filter="bindings.role:${full_role}" \
        --format="value(bindings.members)"
      ;;
    folder)
      _gcloud_try resource-manager folders get-iam-policy "$scope_id" \
        --flatten="bindings[].members" \
        --filter="bindings.role:${full_role}" \
        --format="value(bindings.members)"
      ;;
    *)
      output=""
      log_warn "get_policy_members_for_role: invalid scope '$scope_type'"
      return 1
      ;;
  esac
}

remove_deleted_sa_bindings_for_full_role() {
  local sa_email="$1"
  local full_role="$2"
  local scope_type="$3"
  local scope_id="$4"

  if ! get_policy_members_for_role "$scope_type" "$scope_id" "$full_role"; then
    log_warn "Could not inspect $scope_type $scope_id for deleted service account bindings on $full_role."
    return
  fi

  local member
  while IFS= read -r member; do
    [[ -z "$member" ]] && continue
    [[ "$member" == deleted:serviceAccount:${sa_email}\?uid=* ]] || continue
    case "$scope_type" in
      project)
        destroy_exec "binding $full_role -> $member (project $scope_id)" -- \
          projects remove-iam-policy-binding "$scope_id" --member="$member" --role="$full_role" --all
        ;;
      organization)
        destroy_exec "binding $full_role -> $member (org $scope_id)" -- \
          organizations remove-iam-policy-binding "$scope_id" --member="$member" --role="$full_role" --all
        ;;
      folder)
        destroy_exec "binding $full_role -> $member (folder $scope_id)" -- \
          resource-manager folders remove-iam-policy-binding "$scope_id" --member="$member" --role="$full_role" --all
        ;;
    esac
  done <<< "$output"
}

remove_deleted_sa_bindings() {
  local sa_email="$1"
  local role_name="$2"
  local scope_type="${3:-project}"
  local scope_id="${4:-$PROJECT_ID}"
  local full_role

  case "$scope_type" in
    project)
      full_role="projects/$scope_id/roles/$role_name"
      ;;
    organization)
      full_role="organizations/$scope_id/roles/$role_name"
      ;;
    folder)
      full_role="organizations/$ORGANIZATION_ID/roles/$role_name"
      ;;
    *)
      log_warn "remove_deleted_sa_bindings: invalid scope '$scope_type' for role '$role_name'"
      return
      ;;
  esac

  remove_deleted_sa_bindings_for_full_role "$sa_email" "$full_role" "$scope_type" "$scope_id"
}

role_has_non_target_bindings() {
  local full_role="$1"
  local target_members="$2"
  shift 2

  local scope_type scope_id member members
  while [[ "$#" -gt 0 ]]; do
    scope_type="$1"
    scope_id="$2"
    shift 2

    if ! get_policy_members_for_role "$scope_type" "$scope_id" "$full_role"; then
      log_warn "Could not inspect $scope_type $scope_id for $full_role; skipping role deletion."
      return 0
    fi
    members="$output"

    while IFS= read -r member; do
      [[ -z "$member" ]] && continue
      if ! line_in_list "$member" "$target_members"; then
        log_info "Custom role $full_role is still bound to $member at $scope_type $scope_id; skipping role deletion."
        return 0
      fi
    done <<< "$members"
  done

  return 1
}

members_have_non_target_binding() {
  local full_role="$1"
  local target_members="$2"
  local members="$3"
  local context="$4"
  local member

  while IFS= read -r member; do
    [[ -z "$member" ]] && continue
    if ! line_in_list "$member" "$target_members"; then
      log_info "Custom role $full_role is still bound to $member at $context; skipping role deletion."
      return 0
    fi
  done <<< "$members"

  return 1
}

org_has_non_target_role_bindings() {
  local org_id="$1"
  local full_role="$2"
  local target_members="$3"
  local asset_output members

  if ! _gcloud_try asset search-all-iam-policies \
    --scope="organizations/$org_id" \
    --query="policy:$full_role" \
    --format=json; then
    log_warn "Could not inspect organization $org_id for $full_role via Cloud Asset; skipping role deletion."
    return 0
  fi

  asset_output="$output"
  if ! members=$(echo "$asset_output" | jq -r --arg role "$full_role" '.[]?.policy.bindings[]? | select(.role == $role) | .members[]?' 2>/dev/null); then
    log_warn "Could not parse Cloud Asset IAM policy search for $full_role; skipping role deletion."
    return 0
  fi

  members_have_non_target_binding "$full_role" "$target_members" "$members" "organization $org_id"
}

delete_role_if_no_other_bindings() {
  local role_name="$1"
  local scope_type="$2"
  local scope_id="$3"
  local target_members="$4"
  shift 4

  local full_role
  case "$scope_type" in
    project)
      full_role="projects/$scope_id/roles/$role_name"
      ;;
    organization)
      full_role="organizations/$scope_id/roles/$role_name"
      if org_has_non_target_role_bindings "$scope_id" "$full_role" "$target_members"; then
        return
      fi
      ;;
    *)
      delete_role "$role_name" "$scope_type" "$scope_id"
      return
      ;;
  esac

  if role_has_non_target_bindings "$full_role" "$target_members" "$@"; then
    return
  fi

  delete_role "$role_name" "$scope_type" "$scope_id"
}

list_project_service_account_emails() {
  local project="$1"
  if ! _gcloud_try iam service-accounts list --project="$project" --format="value(email)"; then
    log_warn "Could not list service accounts in project $project; keeping shared resources."
    return 1
  fi
}

project_has_other_source_accounts() {
  local project="$1"
  local current_emails="$2"
  local email

  if ! list_project_service_account_emails "$project"; then
    return 0
  fi

  while IFS= read -r email; do
    [[ -z "$email" ]] && continue
    case "$email" in
      eon-restore-*|eon-scan-*|eon-vault-*|eon-gws-*)
        continue
        ;;
      eon-*@${project}.iam.gserviceaccount.com)
        if ! line_in_list "$email" "$current_emails"; then
          log_info "Found another Eon Source service account in project $project: $email"
          return 0
        fi
        ;;
    esac
  done <<< "$output"

  return 1
}

project_has_other_restore_accounts() {
  local project="$1"
  local current_emails="$2"
  local email

  if ! list_project_service_account_emails "$project"; then
    return 0
  fi

  while IFS= read -r email; do
    [[ -z "$email" ]] && continue
    case "$email" in
      eon-restore-*@${project}.iam.gserviceaccount.com)
        if ! line_in_list "$email" "$current_emails"; then
          log_info "Found another Eon Restore service account in project $project: $email"
          return 0
        fi
        ;;
    esac
  done <<< "$output"

  return 1
}

restore_description_var() {
  local restore_sa="$1"
  local project="$2"
  local var_key="$3"
  local description var_value

  if ! _gcloud_try iam service-accounts describe "$restore_sa" --project="$project" --format="value(description)"; then
    return 1
  fi

  description="$output"
  if [[ -z "$description" ]]; then
    return 1
  fi

  if ! var_value=$(echo "$description" | jq -er --arg key "$var_key" '.vars[$key] // empty' 2>/dev/null); then
    return 2
  fi

  if [[ -z "$var_value" ]]; then
    return 2
  fi

  output="$var_value"
}

restore_node_group_email_from_description() {
  restore_description_var "$1" "$2" "node_sa_group"
}

restore_host_project_id_from_description() {
  restore_description_var "$1" "$2" "host_project_id"
}

restore_sa_description_readable() {
  local restore_sa="$1"
  local project="$2"

  _gcloud_try iam service-accounts describe "$restore_sa" --project="$project" --format="value(description)"
}

restore_gws_enabled_from_description() {
  local restore_sa="$1"
  local project="$2"
  local description gws_enabled

  if ! restore_sa_description_readable "$restore_sa" "$project"; then
    return 1
  fi

  description="$output"
  if [[ -z "$description" ]]; then
    return 2
  fi

  if ! gws_enabled=$(echo "$description" | jq -r '.vars.gws // empty' 2>/dev/null); then
    return 2
  fi

  if [[ "$gws_enabled" != "true" && "$gws_enabled" != "false" ]]; then
    return 2
  fi

  output="$gws_enabled"
}

detect_restore_gws_for_destroy() {
  if [[ "$ENABLE_GWS" == "true" ]]; then
    return 0
  fi

  local standard_restore_sa="eon-restore-${EON_ACCOUNT_ID:0:16}@${PROJECT_ID}.iam.gserviceaccount.com"
  local lookup_status=0

  set +e
  restore_gws_enabled_from_description "$standard_restore_sa" "$PROJECT_ID" || lookup_status=$?
  set -e
  if [[ "$lookup_status" -eq 0 ]]; then
    if [[ "$output" == "true" ]]; then
      ENABLE_GWS=true
      log_info "Detected Google Workspace restore mode from restore SA description: $standard_restore_sa"
    fi
    return 0
  fi

  if [[ "$lookup_status" -ne 1 ]]; then
    return 0
  fi

  if restore_sa_description_readable "$standard_restore_sa" "$PROJECT_ID"; then
    return 0
  fi

  local gws_restore_sa="eon-restore-${EON_ACCOUNT_ID:0:12}-gws@${PROJECT_ID}.iam.gserviceaccount.com"
  if restore_gws_enabled_from_description "$gws_restore_sa" "$PROJECT_ID" && [[ "$output" == "true" ]]; then
    ENABLE_GWS=true
    log_info "Detected Google Workspace restore mode from restore SA description: $gws_restore_sa"
  fi
  return 0
}

# destroy_summary prints the run summary and, for a real (non-dry-run) destroy,
# exits non-zero if any genuine failures occurred so automation can detect it.
destroy_summary() {
  local account_type="$1"
  echo "--------------------------------------------------------------------------------"
  if [[ "$DRY_RUN" == "true" ]]; then
    echo "[${account_type} DESTROY DRY-RUN COMPLETE] No resources were modified."
  else
    echo "[${account_type} DESTROY COMPLETE] removed=${DESTROY_REMOVED} failed=${DESTROY_FAILURES}"
  fi
  echo "--------------------------------------------------------------------------------"
  if [[ "$DRY_RUN" != "true" && "$DESTROY_FAILURES" -gt 0 ]]; then
    log_warn "${DESTROY_FAILURES} resource(s) could not be deleted. Review the warnings above and re-run."
    exit 1
  fi
}

# Canonical Eon custom-role catalogs. These MUST mirror the roles created in
# handle_source_account / handle_restore_account. destroy_*_account deletes the
# full catalog (capability-flag-agnostic); roles never created are benign skips.
SOURCE_ROLE_CATALOG=(
  EonDiscoveryRole
  EonOrgDiscoveryRole
  EonGceCaptureRole
  EonGceMachineImageRole
  EonGceDiscoveryRole
  EonGceBackupRole
  EonGceSnapshotManagerRole
  EonGceConsistencyGroupRole
  EonGceInstantSnapshotGroupRole
  EonGceVaultCaptureRole
  EonGcsDiscoveryRole
  EonGcsBackupRole
  EonGcsInventoryUserRole
  EonGcsNotificationManagerRole
  EonGcsInventoryBucketCreatorRole
  EonGcsInventoryManagerRole
  EonGcsInventoryIamManagerRole
  EonCloudSqlDiscoveryRole
  EonCloudSqlBackupRole
  EonCloudSqlSnapshotConsumerRole
  EonBigQueryDiscoveryRole
  EonBigQueryBackupRole
  EonSapHanaRole
)

RESTORE_ROLE_CATALOG=(
  EonOrgDiscoveryRole
  EonRestoreRole
  EonRestoreFirestoreRole
  EonRestoreVmServiceAccountManagerRole
  EonBigQueryExportWritesRole
  EonExportWritesRole
  EonRestoreNodeRole
  EonRestoreNodeFirestoreRole
  EonBigQueryRestoreNodeRole
  EonGcsRestoreNodeRole
  EonGceRestoreRole
  EonGceRestoreNodeRole
  EonSecretManagerRole
  EonSecretsManagerRole
  EonComputeManagerRole
  EonBatchManagerRole
  EonBigQueryRestoreRole
  EonCloudSqlRestoreRole
)

DISCOVERY_ROLE_PERMISSIONS="\
iam.serviceAccounts.get,\
resourcemanager.projects.get\
"

EON_ORG_DISCOVERY_ROLE_PERMISSIONS="\
resourcemanager.projects.list,\
resourcemanager.projects.get,\
resourcemanager.folders.list,\
resourcemanager.folders.get\
"

# Read-only asset enumeration for GCE
EON_GCE_DISCOVERY_ROLE_PERMISSIONS="\
compute.instances.list,\
compute.instances.get,\
compute.disks.get,\
compute.disks.list,\
compute.snapshots.get,\
compute.snapshots.list\
"

# Snapshot and machine-image creation permissions for GCE
# Also includes permissions for crash-consistent multi-disk snapshots:
#   compute.resourcePolicies.get           - read CG membership of disks
#   compute.disks.addResourcePolicies      - add disks to consistency groups
#   compute.disks.removeResourcePolicies   - remove disks from CGs during cleanup (no IAM condition available for this permission)
#   compute.regionOperations.get           - poll regional async operations (ISG create/delete)
#   compute.instantSnapshots.list          - list instant snapshots in a group
#   compute.instantSnapshots.useReadOnly   - create standard snapshots from instant snapshots
#   compute.resourcePolicies.useReadOnly   - attach CG policy when creating ISG
#   compute.disks.useReadOnly              - read disk data for snapshot creation
EON_GCE_CAPTURE_ROLE_PERMISSIONS="\
compute.instances.useReadOnly,\
compute.disks.createSnapshot,\
compute.globalOperations.get,\
cloudkms.cryptoKeyVersions.useToEncrypt,\
cloudkms.cryptoKeyVersions.useToDecrypt,\
compute.resourcePolicies.get,\
compute.disks.addResourcePolicies,\
compute.disks.removeResourcePolicies,\
compute.regionOperations.get,\
compute.instantSnapshots.list,\
compute.instantSnapshots.useReadOnly,\
compute.resourcePolicies.useReadOnly,\
compute.disks.useReadOnly,\
iam.serviceAccounts.actAs\
"
EON_GCE_MACHINE_IMAGE_ROLE_PERMISSIONS="compute.machineImages.create,compute.machineImages.delete,compute.machineImages.get,compute.machineImages.setIamPolicy"

EON_GCE_CONSISTENCY_GROUP_ROLE_PERMISSIONS="\
compute.resourcePolicies.create,\
compute.resourcePolicies.delete,\
compute.resourcePolicies.use\
"

EON_GCE_INSTANT_SNAPSHOT_GROUP_ROLE_PERMISSIONS="\
compute.instantSnapshotGroups.create,\
compute.instantSnapshotGroups.delete,\
compute.instantSnapshotGroups.get\
"

EON_GCE_VAULT_CAPTURE_ROLE_PERMISSIONS="\
compute.disks.createSnapshot,\
compute.disks.useReadOnly,\
compute.disks.get,\
compute.instantSnapshots.useReadOnly,\
cloudkms.cryptoKeyVersions.useToEncrypt,\
cloudkms.cryptoKeyVersions.useToDecrypt\
"

EON_GCE_SNAPSHOT_MANAGER_ROLE_PERMISSIONS="\
compute.snapshots.create,\
compute.snapshots.get,\
compute.snapshots.list,\
compute.snapshots.delete,\
compute.snapshots.setLabels,\
compute.snapshots.setIamPolicy\
"

EON_CLOUDSQL_DISCOVERY_ROLE_PERMISSIONS="\
cloudsql.instances.list,\
cloudsql.instances.get,\
cloudsql.backupRuns.get,\
cloudsql.backupRuns.list,\
cloudsql.users.list\
"

EON_CLOUDSQL_BACKUP_ROLE_PERMISSIONS="\
cloudsql.backupRuns.create,\
cloudsql.backupRuns.delete\
"

EON_CLOUDSQL_SNAPSHOT_CONSUMER_ROLE_PERMISSIONS="\
cloudsql.instances.restoreBackup,\
cloudsql.backupRuns.get\
"

EON_GCE_BACKUP_ROLE_PERMISSIONS="\
compute.disks.useReadOnly,\
cloudkms.cryptoKeyVersions.useToDecrypt,\
cloudkms.cryptoKeyVersions.useToEncrypt\
"

RESTORE_ROLE_PERMISSIONS="\
iam.serviceAccounts.get,\
storage.buckets.list,\
storage.buckets.get,\
compute.regions.list,\
compute.subnetworks.list,\
compute.machineTypes.list,\
compute.diskTypes.list,\
cloudkms.keyRings.list,\
cloudkms.cryptoKeys.list,\
pubsub.topics.create,\
pubsub.topics.get,\
pubsub.topics.delete,\
pubsub.topics.publish,\
pubsub.topics.attachSubscription,\
pubsub.subscriptions.create,\
pubsub.subscriptions.get,\
pubsub.subscriptions.delete,\
compute.instances.get,\
compute.instances.list,\
compute.disks.use,\
compute.disks.create,\
compute.disks.list,\
compute.disks.get,\
compute.disks.setLabels,\
compute.disks.delete,\
compute.subnetworks.use,\
compute.networks.get,\
compute.networks.list,\
compute.subnetworks.useExternalIp,\
compute.subnetworks.get,\
compute.addresses.list,\
compute.addresses.get,\
compute.regions.get,\
compute.zoneOperations.get,\
resourcemanager.projects.get,\
compute.projects.get,\
compute.machineTypes.get,\
logging.logEntries.list\
"

EON_RESTORE_FIRESTORE_ROLE_PERMISSIONS="\
datastore.entities.get,\
datastore.entities.list,\
datastore.entities.delete\
"

EON_RESTORE_VM_SERVICE_ACCOUNT_MANAGER_ROLE_PERMISSIONS="compute.instances.setServiceAccount"

RESTORE_STORAGE_WRITE_ACCESS_ROLE="\
storage.objects.create,\
storage.objects.delete\
"

EON_SECRETS_MANAGER_ROLE_PERMISSIONS="\
secretmanager.secrets.get,\
secretmanager.versions.access,\
secretmanager.secrets.create,\
secretmanager.versions.add\
"

EON_COMPUTE_MANAGER_ROLE_PERMISSIONS="\
compute.instances.create,\
compute.instances.setMetadata,\
compute.instances.setLabels,\
compute.instances.setMachineType,\
compute.instances.deleteAccessConfig,\
compute.instances.stop,\
compute.instances.delete,\
compute.instances.start,\
compute.instances.getSerialPortOutput,\
compute.instances.update,\
compute.instances.deleteTagBinding,\
compute.instances.listTagBindings,\
compute.addresses.create,\
compute.addresses.delete,\
compute.addresses.use,\
compute.addresses.createInternal,\
compute.addresses.deleteInternal,\
compute.addresses.useInternal,\
compute.addresses.setLabels\
"

EON_BATCH_MANAGER_ROLE_PERMISSIONS="\
batch.jobs.create,\
batch.jobs.get,\
batch.jobs.list,\
batch.jobs.delete,\
batch.operations.get,\
batch.operations.list,\
batch.states.report,\
batch.tasks.get\
"

EON_GCS_NOTIFICATION_MANAGER_ROLE_PERMISSIONS="\
storage.buckets.update\
"

EON_GCS_INVENTORY_USER_ROLE_PERMISSIONS="\
storageinsights.reportConfigs.get,\
storageinsights.reportConfigs.list\
"

EON_GCS_INVENTORY_MANAGER_ROLE_PERMISSIONS="\
storage.buckets.get,\
storage.buckets.getIamPolicy,\
storage.buckets.getObjectInsights,\
storage.buckets.list,\
storage.objects.create,\
storage.objects.list,\
storageinsights.reportConfigs.create,\
storageinsights.reportConfigs.delete,\
resourcemanager.projects.get\
"

EON_GCS_INVENTORY_BUCKET_CREATOR_ROLE_PERMISSIONS="\
storage.buckets.create\
"

EON_GCS_INVENTORY_IAM_MANAGER_ROLE_PERMISSIONS="\
storage.buckets.setIamPolicy\
"

EON_GCS_DISCOVERY_ROLE_PERMISSIONS="\
storage.buckets.list,\
storage.buckets.get,\
monitoring.timeSeries.list\
"

EON_GCS_BACKUP_ROLE_PERMISSIONS="\
storage.objects.list,\
storage.objects.get\
"

EON_BIGQUERY_DISCOVERY_ROLE_PERMISSIONS="\
bigquery.datasets.get,\
bigquery.tables.list,\
bigquery.tables.get,\
bigquery.routines.list,\
bigquery.routines.get,\
bigquery.jobs.create,\
bigquery.jobs.get\
"

EON_BIGQUERY_BACKUP_ROLE_PERMISSIONS="\
bigquery.datasets.get,\
bigquery.tables.list,\
bigquery.tables.get,\
bigquery.routines.list,\
bigquery.routines.get,\
bigquery.jobs.create,\
bigquery.tables.getData,\
bigquery.tables.export,\
bigquery.jobs.get\
"

EON_BIGQUERY_RESTORE_ROLE_PERMISSIONS="\
bigquery.datasets.create,\
bigquery.datasets.get,\
bigquery.tables.create,\
bigquery.tables.get,\
bigquery.tables.list,\
bigquery.tables.update,\
bigquery.tables.updateData,\
bigquery.tables.getData,\
bigquery.tables.export,\
bigquery.routines.create,\
bigquery.routines.list,\
bigquery.routines.update,\
bigquery.jobs.create,\
bigquery.jobs.get\
"

EON_SAP_HANA_ROLE_PERMISSIONS="\
storage.buckets.get,\
storage.objects.get,\
storage.objects.list,\
storage.objects.create,\
storage.objects.delete,\
compute.instances.get,\
compute.instances.list,\
pubsub.subscriptions.consume,\
datastore.databases.get,\
datastore.entities.get,\
datastore.entities.list,\
datastore.entities.create,\
datastore.entities.update,\
datastore.indexes.get,\
datastore.indexes.list\
"

RESTORE_NODE_ROLE_PERMISSIONS="\
pubsub.subscriptions.consume,\
batch.states.report,\
artifactregistry.repositories.downloadArtifacts,\
logging.logEntries.create,\
logging.logEntries.route\
"

EON_RESTORE_NODE_FIRESTORE_ROLE_PERMISSIONS="\
datastore.entities.create,\
datastore.entities.update\
"

EON_CLOUDSQL_RESTORE_ROLE_PERMISSIONS="\
cloudsql.instances.create,\
cloudsql.instances.get,\
cloudsql.instances.update,\
cloudsql.users.update,\
cloudsql.users.list,\
cloudsql.users.create\
"

EON_GCS_RESTORE_NODE_ROLE_PERMISSIONS="\
storage.objects.create,\
storage.objects.delete,\
storage.multipartUploads.create,\
storage.multipartUploads.abort\
"

EON_GCE_RESTORE_NODE_ROLE_PERMISSIONS="\
cloudkms.cryptoKeyVersions.useToEncrypt,\
cloudkms.cryptoKeyVersions.useToDecrypt\
"

EON_GCE_RESTORE_ROLE_PERMISSIONS="\
compute.instances.create,\
compute.instances.attachDisk,\
compute.instances.detachDisk,\
compute.instances.setLabels,\
compute.instances.stop,\
compute.instances.createTagBinding\
"

EON_BIGQUERY_RESTORE_NODE_ROLE_PERMISSIONS="\
bigquery.datasets.create,\
bigquery.datasets.get,\
bigquery.jobs.create,\
bigquery.tables.create,\
bigquery.tables.get,\
bigquery.tables.getData,\
bigquery.tables.list,\
bigquery.tables.update,\
bigquery.tables.updateData,\
bigquery.routines.create,\
bigquery.routines.list,\
bigquery.routines.update\
"

EON_RESTORE_TO_SHARED_SUBNETWORK_PERMISSIONS="\
compute.subnetworks.use,\
compute.subnetworks.list,\
compute.networks.get,\
compute.networks.list,\
compute.subnetworks.useExternalIp,\
compute.subnetworks.get,\
compute.addresses.list\
"

SOURCE_SA_EMAIL=""

REQUIRED_APIS=(
  iam.googleapis.com
  serviceusage.googleapis.com
  cloudresourcemanager.googleapis.com
  secretmanager.googleapis.com
  compute.googleapis.com
  cloudkms.googleapis.com
  logging.googleapis.com
  monitoring.googleapis.com
  storage.googleapis.com
)

RESTORE_REQUIRED_APIS=(
  pubsub.googleapis.com
  firestore.googleapis.com
  batch.googleapis.com
)

# Google Workspace APIs (Gmail, Drive, People, Admin SDK, Groups Settings, Groups Migration) Eon uses to
# restore your data back into Google Workspace. Enabled only when Google Workspace restore is selected.
GWS_RESTORE_REQUIRED_APIS=(
  gmail.googleapis.com
  drive.googleapis.com
  people.googleapis.com
  admin.googleapis.com
  groupssettings.googleapis.com
  groupsmigration.googleapis.com
)

role_exists() {
  local role_name="$1"
  local scope_type="$2"
  local scope_id="$3"

  if [[ "$scope_type" == "project" ]]; then
    _gcloud iam roles list --project="$scope_id" \
      --filter="name=projects/$scope_id/roles/$role_name" \
      --show-deleted --format=json
  elif [[ "$scope_type" == "organization" ]]; then
    _gcloud iam roles list --organization="$scope_id" \
      --filter="name=organizations/$scope_id/roles/$role_name" \
      --show-deleted --format=json
  else
    log_error "Invalid scope type: $scope_type"
    return 1
  fi

  local c
  c=$(echo "$output" | jq length)
  [ "$c" -eq 1 ]
}

role_exists_in_project() {
  local role_name="$1"
  local project="$2"
  role_exists "$role_name" "project" "$project"
  return $?
}

role_exists_in_organization() {
  local role_name="$1"
  local org="$2"
  role_exists "$role_name" "organization" "$org"
  return $?
}

role_deleted() {
  local role_name="$1"
  local scope_type="$2"
  local scope_id="$3"

  if [[ "$scope_type" == "project" ]]; then
    _gcloud iam roles list --project="$scope_id" \
      --filter="name=projects/$scope_id/roles/$role_name" \
      --format=json
  elif [[ "$scope_type" == "organization" ]]; then
    _gcloud iam roles list --organization="$scope_id" \
      --filter="name=organizations/$scope_id/roles/$role_name" \
      --format=json
  else
    log_error "Invalid scope_id type: $scope_type"
    return 1
  fi

  local c
  c=$(echo "$output" | jq length)
  [ "$c" -eq 0 ]
}

role_deleted_in_project() {
  local role_name="$1"
  local project="$2"
  role_deleted "$role_name" "project" "$project"
  return $?
}

role_deleted_in_organization() {
  local role_name="$1"
  local org="$2"
  role_deleted "$role_name" "organization" "$org"
  return $?
}

create_or_update_role() {
  local role_name="$1"
  local perms="$2"
  local scope_type="$3"
  local scope_id="$4"

  if [[ "$scope_type" == "project" ]]; then
    if role_exists_in_project "$role_name" "$scope_id"; then
      if role_deleted_in_project "$role_name" "$scope_id"; then
        log_info "Undeleting custom role: $role_name in project $scope_id"
        _gcloud iam roles undelete "$role_name" --project="$scope_id"
      fi
      log_info "Updating custom role: $role_name in project $scope_id"
      _gcloud iam roles update "$role_name" \
        --project="$scope_id" \
        --permissions="$perms" \
        --stage="GA"
    else
      log_info "Creating custom role: $role_name in project $scope_id"
      _gcloud iam roles create "$role_name" \
        --project="$scope_id" \
        --permissions="$perms" \
        --stage="GA"
  fi
  elif [[ "$scope_type" == "organization" ]]; then
    if role_exists_in_organization "$role_name" "$scope_id"; then
      if role_deleted_in_organization "$role_name" "$scope_id"; then
        log_info "Undeleting custom role: $role_name in organization $scope_id"
        _gcloud iam roles undelete "$role_name" --organization="$scope_id"
      fi
      log_info "Updating custom role: $role_name in organization $scope_id"
      _gcloud iam roles update "$role_name" \
        --organization="$scope_id" \
        --permissions="$perms" \
        --stage="GA"
    else
      log_info "Creating custom role: $role_name in organization $scope_id"
      _gcloud iam roles create "$role_name" \
        --organization="$scope_id" \
        --permissions="$perms" \
        --stage="GA"
    fi
  else
    log_error "Invalid scope_id type: $scope_type"
    return 1
  fi
}

create_or_update_role_in_project() {
  local role_name="$1"
  local perms="$2"
  local project="$3"
  create_or_update_role "$role_name" "$perms" "project" "$project"
}

create_or_update_role_in_organization() {
  local role_name="$1"
  local perms="$2"
  local org="$3"
  create_or_update_role "$role_name" "$perms" "organization" "$org"
}

build_inventory_insights_collector_iam_condition() {
  printf '^:^title=RestrictEonGcsInventoryInsightsCollectorIam:description=OnlyStorageInsightsCollectorGrant:expression=api.getAttribute("iam.googleapis.com/modifiedGrantsByRole", []).hasOnly(["roles/storage.insightsCollectorService"])'
}

build_inventory_object_creator_iam_condition() {
  printf '^:^title=RestrictEonGcsInventoryObjectCreatorIam:description=OnlyInventoryDestinationObjectCreatorGrant:expression=api.getAttribute("iam.googleapis.com/modifiedGrantsByRole", []).hasOnly(["roles/storage.objectCreator"]) && resource.name.startsWith("projects/_/buckets/eon-inventory-destination-")'
}

build_inventory_bucket_creator_iam_condition() {
  printf 'title=RestrictEonGcsInventoryBucketCreator,description=OnlyInventoryDestinationBuckets,expression=resource.name.startsWith("projects/_/buckets/eon-inventory-destination-")'
}

build_restore_firestore_iam_condition() {
  printf 'title=RestrictEonRestoreFirestore,description=OnlyEonRestoreCommandFirestore,expression=resource.type == "firestore.googleapis.com/Database" && (resource.name.endsWith("/databases/eon-cmd-results") || resource.name.endsWith("/databases/eon-sap-hana-cmds-%s"))' "$PROJECT_ID"
}

restore_gcs_target_tag_key() {
  if [[ -n "$ORGANIZATION_ID" ]]; then
    printf '%s/eon-allow-restore' "$ORGANIZATION_ID"
  else
    printf '%s/eon-allow-restore' "$PROJECT_ID"
  fi
}

build_restore_gcs_target_tag_iam_condition() {
  local tag_key
  tag_key="$(restore_gcs_target_tag_key)"
  printf '^:^title=RestrictEonRestoreGcsTargetTag:description=OnlyBucketsTaggedEonAllowRestore:expression=resource.matchTag("%s", "true")' "$tag_key"
}

# The account-id suffix keeps tag keys unique across Eon accounts sharing a project/organization.
gce_managed_tag_namespaced_key() {
  if [[ -n "$ORGANIZATION_ID" ]]; then
    printf '%s/eon-managed-%s' "$ORGANIZATION_ID" "${EON_ACCOUNT_ID:0:8}"
  else
    printf '%s/eon-managed-%s' "$PROJECT_ID" "${EON_ACCOUNT_ID:0:8}"
  fi
}

build_eon_managed_instances_iam_condition() {
  local title="$1"
  local description="$2"
  printf '^:^title=%s:description=%s:expression=resource.name.extract("/instances/{instance}").startsWith("eon-") || resource.name.extract("/addresses/{address}").startsWith("eon-") || resource.matchTag("%s", "true")' \
    "$title" "$description" "$(gce_managed_tag_namespaced_key)"
}

bind_iam_member() {
  local member="$1"
  local role_name="$2"
  local cond_expr="${3:-None}"   # default = None
  local scope_type="${4:-project}"
  local scope_id="${5:-$PROJECT_ID}"

  local full_role
  if [[ "$scope_type" == "project" ]]; then
    full_role="projects/$scope_id/roles/$role_name"
    log_info "Binding role '$full_role' to $member"

    if [[ "$cond_expr" == "None" ]]; then
      _gcloud projects add-iam-policy-binding "$scope_id" \
          --member="$member" \
          --role="$full_role" \
          --condition=None
    else
      _gcloud projects add-iam-policy-binding "$scope_id" \
          --member="$member" \
          --role="$full_role" \
          --condition="$cond_expr"
    fi
  elif [[ "$scope_type" == "organization" ]]; then
    full_role="organizations/$scope_id/roles/$role_name"
    log_info "Binding role '$full_role' to $member"

    if [[ "$cond_expr" == "None" ]]; then
      _gcloud organizations add-iam-policy-binding "$scope_id" \
          --member="$member" \
          --role="$full_role" \
          --condition=None
    else
      _gcloud organizations add-iam-policy-binding "$scope_id" \
          --member="$member" \
          --role="$full_role" \
          --condition="$cond_expr"
    fi
  elif [[ "$scope_type" == "folder" ]]; then
    full_role="organizations/$ORGANIZATION_ID/roles/$role_name"  # Folder roles are defined at org level
    log_info "Binding role '$full_role' to $member at folder level"

    if [[ "$cond_expr" == "None" ]]; then
      _gcloud resource-manager folders add-iam-policy-binding "$scope_id" \
          --member="$member" \
          --role="$full_role" \
          --condition=None
    else
      _gcloud resource-manager folders add-iam-policy-binding "$scope_id" \
          --member="$member" \
          --role="$full_role" \
          --condition="$cond_expr"
    fi
  else
    log_error "Invalid scope_id type: $scope_type"
    return 1
  fi
}

bind_role_to_sa() {
  local sa_email="$1"
  shift
  bind_iam_member "serviceAccount:$sa_email" "$@"
}

remove_iam_member_bindings() {
  local member="$1"
  local role_name="$2"
  local scope_type="${3:-project}"
  local scope_id="${4:-$PROJECT_ID}"

  local full_role
  case "$scope_type" in
    project)
      full_role="projects/$scope_id/roles/$role_name"
      log_info "Removing existing bindings for role '$full_role' from $member"
      if _gcloud_try projects remove-iam-policy-binding "$scope_id" --member="$member" --role="$full_role" --all; then
        log_success "Removed existing bindings for role '$full_role' from $member"
      elif is_not_found_error "$output_err"; then
        log_info "No existing bindings for role '$full_role' on $member"
      else
        log_warn "Failed to remove existing bindings for role '$full_role' from $member"
        log_warn_captured_output
      fi
      ;;
    organization)
      full_role="organizations/$scope_id/roles/$role_name"
      log_info "Removing existing bindings for role '$full_role' from $member"
      if _gcloud_try organizations remove-iam-policy-binding "$scope_id" --member="$member" --role="$full_role" --all; then
        log_success "Removed existing bindings for role '$full_role' from $member"
      elif is_not_found_error "$output_err"; then
        log_info "No existing bindings for role '$full_role' on $member"
      else
        log_warn "Failed to remove existing bindings for role '$full_role' from $member"
        log_warn_captured_output
      fi
      ;;
    folder)
      full_role="organizations/$ORGANIZATION_ID/roles/$role_name"
      log_info "Removing existing bindings for role '$full_role' from $member at folder level"
      if _gcloud_try resource-manager folders remove-iam-policy-binding "$scope_id" --member="$member" --role="$full_role" --all; then
        log_success "Removed existing bindings for role '$full_role' from $member at folder level"
      elif is_not_found_error "$output_err"; then
        log_info "No existing bindings for role '$full_role' on $member at folder level"
      else
        log_warn "Failed to remove existing bindings for role '$full_role' from $member at folder level"
        log_warn_captured_output
      fi
      ;;
    *)
      log_warn "remove_iam_member_bindings: invalid scope '$scope_type' for role '$role_name'"
      ;;
  esac
}

bind_role_to_group() {
  local group_email="$1"
  shift
  bind_iam_member "group:$group_email" "$@"
}

get_org_domain() {
  local org_id="$1"
  set +e
  local domain
  domain=$(gcloud organizations describe "$org_id" --format="value(displayName)" --quiet 2>/dev/null)
  set -e
  echo "$domain"
}

default_restore_node_group_email() {
  local org_id="$1"
  local org_domain
  org_domain=$(get_org_domain "$org_id")
  if [[ -z "$org_domain" ]]; then
    return 1
  fi
  echo "eon-restore-nodes-${EON_ACCOUNT_ID:0:8}$(restore_group_email_suffix)@${org_domain}"
}

# group_exists checks whether a Cloud Identity group exists.
# Returns 0 if it exists, non-zero otherwise. No logging beyond verbose;
# no errors printed — the caller decides what to do with a "not found".
group_exists() {
  local group_email="$1"
  log_verbose "Checking if group '$group_email' exists..."
  set +e
  gcloud identity groups describe "$group_email" --quiet --verbosity=error >/dev/null 2>&1
  local retVal=$?
  set -e
  return $retVal
}

# create_group runs `gcloud identity groups create` and returns gcloud's exit
# code. Unlike _gcloud, this does NOT exit on failure — the caller handles
# the error so it can present a contextual message.
create_group() {
  local group_email="$1"
  local org_id="$2"
  log_info "Creating group '$group_email'..."
  set +e
  gcloud identity groups create "$group_email" \
    --organization="$org_id" \
    --display-name="Eon Restore Node Service Accounts" \
    --labels="cloudidentity.googleapis.com/groups.discussion_forum" \
    --quiet --verbosity=error
  local retVal=$?
  set -e
  if [ $retVal -eq 0 ]; then
    log_success "Created group: $group_email"
  fi
  return $retVal
}

# ensure_group_exists is the idempotent composition: describe → create if missing.
# Kept so existing call sites continue to work after the preflight has already
# created the group (the describe path hits and this becomes a no-op).
ensure_group_exists() {
  local group_email="$1"
  local org_id="$2"

  if group_exists "$group_email"; then
    log_info "Group '$group_email' already exists."
    return 0
  fi

  if ! create_group "$group_email" "$org_id"; then
    log_error "Failed to create group '$group_email'."
    exit 1
  fi
}

add_sa_to_group() {
  local sa_email="$1"
  local group_email="$2"

  set +e
  local existing
  existing=$(gcloud identity groups memberships list \
    --group-email="$group_email" \
    --filter="preferredMemberKey.id=$sa_email" \
    --format="value(name)" --quiet --verbosity=error 2>/dev/null)
  set -e

  if [ -n "$existing" ]; then
    log_verbose "SA '$sa_email' is already a member of group '$group_email'"
    return 0
  fi

  log_info "Adding $sa_email to group $group_email..."
  _gcloud identity groups memberships add \
    --group-email="$group_email" \
    --member-email="$sa_email" \
    --roles=MEMBER
  log_success "Added $sa_email to group $group_email"
}

remove_sa_from_group() {
  local sa_email="$1"
  local group_email="$2"

  if [[ "$DRY_RUN" == "true" ]]; then
    log_info "[DRY-RUN] would delete: group membership $sa_email in $group_email"
    return 0
  fi

  if ! _gcloud_try identity groups memberships list \
    --group-email="$group_email" \
    --filter="preferredMemberKey.id=$sa_email" \
    --format="value(name)"; then
    if is_not_found_error "$output_err"; then
      log_info "Already absent (skip): group membership $sa_email in $group_email"
      return 0
    fi

    DESTROY_FAILURES=$((DESTROY_FAILURES + 1))
    log_warn "Failed to find group membership $sa_email in $group_email"
    log_warn_captured_output
    return 0
  fi

  local membership_name
  membership_name="${output%%$'\n'*}"
  if [[ -z "$membership_name" ]]; then
    log_info "Already absent (skip): group membership $sa_email in $group_email"
    return 0
  fi

  destroy_exec "group membership $sa_email in $group_email" -- \
    identity groups memberships delete --group-email="$group_email" --member-email="$sa_email"
}

# preflight_group_create_permission verifies upfront that the operator can
# create (or has) the Google Group that the org/folder restore flow needs to
# collapse per-project IAM bindings. Runs only for restore + org/folder.
#
# Failure mode 1 (no --node-sa-group-email passed): we auto-generate the
# group email and attempt to create it. If create fails on permissions, we
# print a contextual error explaining how to grant the perm or how to
# pre-create the group.
#
# Failure mode 2 (--node-sa-group-email passed): the operator owns the
# group. We only verify it is reachable. If describe fails, we print an
# error explaining how to create it or how to drop the flag.
preflight_group_create_permission() {
  if [[ "$ACCOUNT_TYPE" != "restore" ]]; then
    return 0
  fi
  if [[ -z "$ORGANIZATION_ID" && -z "$FOLDER_ID" ]]; then
    # project-scope restore: no Google Group is used
    return 0
  fi

  log_info "Preflight: verifying Cloud Identity group permissions..."

  # Enable Cloud Identity in the management project only. The rest of the
  # APIs are enabled later by ensure_required_apis. This call may itself
  # fail (e.g. operator lacks serviceusage.services.enable) — that is the
  # same error path the original flow would have hit later, so we let
  # enable_api's existing behavior handle it.
  enable_api "cloudidentity.googleapis.com" "$PROJECT_ID"

  # Resolve / compute the group email if the operator did not provide one.
  if [[ "$USER_PROVIDED_NODE_SA_GROUP_EMAIL" == "false" ]]; then
    if ! NODE_SA_GROUP_EMAIL=$(default_restore_node_group_email "$ORGANIZATION_ID"); then
      log_error "Failed to determine organization domain. Please provide --node-sa-group-email."
      exit 1
    fi
    log_info "Auto-generated group email: $NODE_SA_GROUP_EMAIL"
  fi

  # Fast path: group already exists. No create needed → no create-permission
  # needed. We are done.
  if group_exists "$NODE_SA_GROUP_EMAIL"; then
    log_success "✓ Group '$NODE_SA_GROUP_EMAIL' already exists; no creation needed."
    return 0
  fi

  # Group does not exist (or we cannot see it).
  if [[ "$USER_PROVIDED_NODE_SA_GROUP_EMAIL" == "true" ]]; then
    # Operator passed a group email but it cannot be described.
    log_error "Google Group '$NODE_SA_GROUP_EMAIL' was not found in your organization,"
    log_error "or your gcloud user does not have permission to view it."
    log_error ""
    log_error "How to fix this — pick one:"
    log_error ""
    log_error "  1) Create the group in the Google Workspace Admin Console:"
    log_error "       a. Create a group with the email '$NODE_SA_GROUP_EMAIL'."
    log_error "       b. Add your gcloud user as a group OWNER."
    log_error "       c. Re-run this script unchanged."
    log_error ""
    log_error "  2) Let Eon auto-create the group: re-run this script without the"
    log_error "     '--node-sa-group-email' flag. Your gcloud user will need the"
    log_error "     'cloudidentity.groups.create' permission ('Groups Admin' role in"
    log_error "     Google Workspace, or 'roles/cloudidentity.groupsEditor' IAM role)."
    exit 1
  fi

  # Operator did not pass a group email — auto-create path. Attempt the
  # create; if it fails, the most likely cause is missing permission.
  if ! create_group "$NODE_SA_GROUP_EMAIL" "$ORGANIZATION_ID"; then
    log_error "Cannot create Google Group '$NODE_SA_GROUP_EMAIL' in organization '$ORGANIZATION_ID'."
    log_error ""
    log_error "What went wrong:"
    log_error "  Eon needs to create a Google Group to bind IAM roles to restore-node service"
    log_error "  accounts across all projects in your org/folder. Creating a Cloud Identity"
    log_error "  group requires the 'cloudidentity.groups.create' permission, which your"
    log_error "  current gcloud user does not have."
    log_error ""
    log_error "How to fix this — pick one:"
    log_error ""
    log_error "  1) Grant your user the permission and re-run this script:"
    log_error "       Easiest: assign the 'Groups Admin' role in the Google Workspace Admin"
    log_error "       Console (admin.google.com -> Account -> Admin roles)."
    log_error "       IAM alternative: grant 'roles/cloudidentity.groupsEditor' at the"
    log_error "       organization level."
    log_error ""
    log_error "  2) Create the group yourself and pass it to the script:"
    log_error "       a. In the Google Workspace Admin Console, create a group with email"
    log_error "          '$NODE_SA_GROUP_EMAIL' (or any email under your org's domain)."
    log_error "       b. Add your gcloud user as a group OWNER (not just member) — this"
    log_error "          script will add service accounts to the group, which requires"
    log_error "          owner/manager rights on the group."
    log_error "       c. Re-run this script with:"
    log_error "            --node-sa-group-email <the-email-you-created>"
    exit 1
  fi

  log_success "✓ Group '$NODE_SA_GROUP_EMAIL' created. Continuing onboarding."
}

enable_api() {
  local api="$1"
  local project="${2:-$PROJECT_ID}"
  local max_retries=5
  local retry_count=0
  local enabled_services
  local enable_output
  local retVal

  # Check if API is already enabled
  set +e
  enabled_services=$(gcloud services list --enabled --project="$project" --format="value(config.name)" --quiet 2>/dev/null)
  set -e

  if echo "$enabled_services" | grep -q "^${api}$"; then
    log_verbose "API ${api} already enabled in ${project}"
    return 0
  fi

  log_info "Enabling API ${api} in project ${project}..."

  while [ $retry_count -lt $max_retries ]; do
    # Attempt to enable the API
    set +e
    enable_output=$(gcloud services enable "${api}" --project="${project}" --quiet 2>&1)
    retVal=$?
    set -e

    if [ $retVal -eq 0 ]; then
      sleep 5
      log_success "API ${api} enabled in ${project}"
      return 0
    fi

    # Handle token expiration - refresh and retry
    if is_token_expired_error "$enable_output"; then
      refresh_gcp_auth
      retry_count=$((retry_count + 1))
      continue
    fi

    # Handle billing disabled - continue with warning (common in non-prod with budget limits)
    if is_billing_error "$enable_output"; then
      log_warn "Billing is disabled for project ${project} - skipping API ${api}"
      log_warn "Some APIs require billing. Enable billing when ready and re-run the script."
      add_billing_issue_project "$project"
      return 0  # Return success to continue with other operations
    fi

    # Handle gcloud response parsing errors by using async mode
    if echo "$enable_output" | grep -q "string indices must be integers"; then
      log_info "Enabling API ${api} in async mode..."

      set +e
      gcloud services enable "${api}" --project="${project}" --async --quiet 2>/dev/null
      set -e

      log_info "Waiting for API ${api} activation to complete..."
      sleep 15

      # Verify the API was enabled
      set +e
      enabled_services=$(gcloud services list --enabled --project="$project" --format="value(config.name)" --quiet 2>/dev/null)
      set -e

      if echo "$enabled_services" | grep -q "^${api}$"; then
        log_success "API ${api} enabled in ${project}"
        return 0
      fi

      retry_count=$((retry_count + 1))
      if [ $retry_count -lt $max_retries ]; then
        log_info "API activation in progress, retrying verification (${retry_count}/${max_retries})..."
        sleep 5
      fi
      continue
    fi

    # Other errors - log and fail
    echo "$enable_output"
    log_error "Failed to enable API ${api} in ${project}"
    return 1
  done

  # Final verification after retries
  set +e
  enabled_services=$(gcloud services list --enabled --project="$project" --format="value(config.name)" --quiet 2>/dev/null)
  set -e

  if echo "$enabled_services" | grep -q "^${api}$"; then
    log_success "API ${api} enabled in ${project}"
    return 0
  fi

  log_error "Failed to enable API ${api} in ${project} after ${max_retries} attempts. Please enable it manually via the GCP Console."
  return 1
}

# ensure_cloud_asset_api enables the Cloud Asset API on the management project
# before org-scoped role cleanup runs. The destroy path's "is this shared org
# role still bound to another Eon account?" check (org_has_non_target_role_bindings)
# relies on `gcloud asset search-all-iam-policies`, which fails with
# SERVICE_DISABLED when the API is off — causing every org-level role deletion to
# be skipped (fail-safe). The API and the search call are free of charge; only
# org-scoped destroys reach the Cloud Asset code path.
ensure_cloud_asset_api() {
  local api="cloudasset.googleapis.com"

  if [[ "$DRY_RUN" == "true" ]]; then
    local enabled_services
    set +e
    enabled_services=$(gcloud services list --enabled --project="$PROJECT_ID" --format="value(config.name)" --quiet 2>/dev/null)
    set -e
    if echo "$enabled_services" | grep -q "^${api}$"; then
      log_info "[DRY-RUN] ${api} already enabled in ${PROJECT_ID}; org-level role-usage checks will run."
    else
      log_info "[DRY-RUN] would enable ${api} in ${PROJECT_ID}. Until it is enabled, the org-level role-usage checks below are skipped (fail-safe); the real run enables it first."
    fi
    return 0
  fi

  if ! enable_api "$api" "$PROJECT_ID"; then
    log_warn "Could not enable ${api} in ${PROJECT_ID}; org-level custom roles may be skipped during cleanup."
    return 0
  fi

  # API activation can lag a few minutes before search-all-iam-policies stops
  # returning SERVICE_DISABLED. Poll a cheap probe so the first destroy run
  # doesn't blanket-skip role deletion; a timeout is tolerated since the
  # cleanup's own fail-safe still protects against deleting a shared role.
  local attempt
  for attempt in $(seq 1 18); do
    if _gcloud_try asset search-all-iam-policies \
      --scope="organizations/$ORGANIZATION_ID" \
      --query="policy:roles/owner" \
      --limit=1 \
      --format=json; then
      log_verbose "Cloud Asset API ready after ${attempt} probe(s)."
      return 0
    fi
    if ! echo "$output_err" | grep -qiE "SERVICE_DISABLED|has not been used in project|is disabled"; then
      # Not a propagation error (e.g. missing org-level permission). Stop polling
      # and let the cleanup's own check surface it.
      log_verbose "Cloud Asset probe returned a non-propagation error; continuing."
      return 0
    fi
    log_verbose "Waiting for ${api} activation to propagate (${attempt}/18)..."
    sleep 5
  done

  log_warn "${api} still not ready after enabling; org-level role cleanup may be skipped this run. Re-running destroy shortly will complete it."
  return 0
}

get_projects_under_organization() {
  local org_id="$1"
  log_verbose "Getting all projects under organization ${org_id}..." >&2
  set +e
  local direct_projects
  direct_projects=$(gcloud projects list --filter="parent.id=${org_id} AND parent.type=organization" --format="value(projectId)" --quiet --verbosity=error 2>&1)
  local retVal=$?
  set -e

  if [ $retVal -ne 0 ]; then
    log_error "Failed to list direct projects under organization ${org_id}: ${direct_projects}" >&2
    return 1
  fi

  # Count direct projects
  local direct_count=0
  if [ -n "$direct_projects" ]; then
    direct_count=$(echo "$direct_projects" | grep -c '^' 2>/dev/null || echo "0")
  fi
  log_verbose "Found ${direct_count} projects directly under organization ${org_id}" >&2

  # Get all folders under the organization
  set +e
  local folders_output
  folders_output=$(gcloud resource-manager folders list --organization="${org_id}" --format="value(name)" --quiet --verbosity=error 2>&1)
  local folders_retVal=$?
  set -e

  local folder_projects=""
  local folder_count=0
  local total_folder_projects=0

  if [ $folders_retVal -eq 0 ] && [ -n "$folders_output" ]; then
    folder_count=$(echo "$folders_output" | grep -c '^' 2>/dev/null || echo "0")
    log_verbose "Found ${folder_count} folders under organization, discovering projects in each folder..." >&2

    # Process each folder and get projects within it
    while IFS= read -r folder_path; do
      [[ -z "$folder_path" ]] && continue

      # Extract folder ID from the full path (folders/123456789)
      local folder_id
      folder_id=$(echo "$folder_path" | sed 's|folders/||g')

      if [[ -n "$folder_id" ]]; then
        log_verbose "Getting projects under folder ${folder_id}..." >&2
        set +e
        local fp
        fp=$(get_projects_under_folder "$folder_id")
        local fp_retVal=$?
        set -e

        if [ $fp_retVal -eq 0 ] && [ -n "$fp" ]; then
          local fp_count
          fp_count=$(echo "$fp" | grep -c '^' 2>/dev/null || echo "0")
          log_verbose "Found ${fp_count} projects under folder ${folder_id}" >&2
          total_folder_projects=$((total_folder_projects + fp_count))

          if [ -n "$folder_projects" ]; then
            folder_projects="${folder_projects}"$'\n'"${fp}"
          else
            folder_projects="${fp}"
          fi
        fi
      fi
    done <<< "$folders_output"
  else
    log_verbose "No folders found under organization ${org_id} or failed to list folders" >&2
  fi

  # Combine direct projects and folder projects
  local all_projects=""
  if [ -n "$direct_projects" ]; then
    all_projects="$direct_projects"
  fi

  if [ -n "$folder_projects" ]; then
    if [ -n "$all_projects" ]; then
      all_projects="${all_projects}"$'\n'"${folder_projects}"
    else
      all_projects="$folder_projects"
    fi
  fi

  # Remove duplicates and empty lines, then output
  local final_projects=""
  if [ -n "$all_projects" ]; then
    final_projects=$(echo "$all_projects" | sort -u | grep -v '^$')
    local final_count
    final_count=$(echo "$final_projects" | grep -c '^' 2>/dev/null || echo "0")
    log_info "Recursive discovery completed: ${final_count} total projects (${direct_count} direct + ${total_folder_projects} in folders)" >&2
    
    if [[ -n "$EXCLUDE_PROJECT_PATTERNS" ]]; then
      final_projects=$(filter_excluded_projects "$final_projects" "$EXCLUDE_PROJECT_PATTERNS")
      local filtered_count
      filtered_count=$(echo "$final_projects" | grep -c '^' 2>/dev/null || echo "0")
      log_info "After exclusion filters: ${filtered_count} projects remaining" >&2
    fi

    if [[ -n "$INCLUDE_LABEL_FILTERS" || -n "$EXCLUDE_LABEL_FILTERS" || -n "$INCLUDE_TAG_FILTERS" || -n "$EXCLUDE_TAG_FILTERS" ]]; then
      final_projects=$(filter_projects_by_labels_and_tags "$final_projects" "$INCLUDE_LABEL_FILTERS" "$EXCLUDE_LABEL_FILTERS" "$INCLUDE_TAG_FILTERS" "$EXCLUDE_TAG_FILTERS")
      local label_tag_filtered_count
      label_tag_filtered_count=$(echo "$final_projects" | grep -c '^' 2>/dev/null || echo "0")
      log_info "After label/tag filters: ${label_tag_filtered_count} projects remaining" >&2
    fi
    
    echo "$final_projects"
  else
    log_info "Recursive discovery completed: 0 projects found" >&2
  fi
}

get_projects_under_folder() {
  local folder_id="$1"
  log_verbose "Getting all projects under folder ${folder_id}..." >&2
  set +e
  local direct_projects
  direct_projects=$(gcloud projects list --filter="parent.id=${folder_id} AND parent.type=folder" --format="value(projectId)" --quiet --verbosity=error 2>&1)
  local retVal=$?
  set -e
  
  if [ $retVal -ne 0 ]; then
    log_error "Failed to list direct projects under folder ${folder_id}: ${direct_projects}" >&2
    return 1
  fi
  
  # Count direct projects
  local direct_count=0
  if [ -n "$direct_projects" ]; then
    direct_count=$(echo "$direct_projects" | grep -c '^' 2>/dev/null || echo "0")
  fi
  log_verbose "Found ${direct_count} projects directly under folder ${folder_id}" >&2
  
  # Get all sub-folders under this folder
  set +e
  local subfolders_output
  subfolders_output=$(gcloud resource-manager folders list --folder="${folder_id}" --format="value(name)" --quiet --verbosity=error 2>&1)
  local subfolders_retVal=$?
  set -e
  
  local subfolder_projects=""
  local subfolder_count=0
  local total_subfolder_projects=0
  
  if [ $subfolders_retVal -eq 0 ] && [ -n "$subfolders_output" ]; then
    subfolder_count=$(echo "$subfolders_output" | grep -c '^' 2>/dev/null || echo "0")
    log_verbose "Found ${subfolder_count} sub-folders under folder ${folder_id}, discovering projects in each..." >&2
    
    # Process each sub-folder and get projects within it recursively
    while IFS= read -r subfolder_path; do
      [[ -z "$subfolder_path" ]] && continue
      
      # Extract folder ID from the full path (folders/123456789)
      local subfolder_id
      subfolder_id=$(echo "$subfolder_path" | sed 's|folders/||g')
      
      if [[ -n "$subfolder_id" ]]; then
        log_verbose "Recursively getting projects under sub-folder ${subfolder_id}..." >&2
        set +e
        local sfp
        sfp=$(get_projects_under_folder "$subfolder_id")
        local sfp_retVal=$?
        set -e
        
        if [ $sfp_retVal -eq 0 ] && [ -n "$sfp" ]; then
          local sfp_count
          sfp_count=$(echo "$sfp" | grep -c '^' 2>/dev/null || echo "0")
          log_verbose "Found ${sfp_count} projects under sub-folder ${subfolder_id} (recursive)" >&2
          total_subfolder_projects=$((total_subfolder_projects + sfp_count))
          
          if [ -n "$subfolder_projects" ]; then
            subfolder_projects="${subfolder_projects}"$'\n'"${sfp}"
          else
            subfolder_projects="${sfp}"
          fi
        fi
      fi
    done <<< "$subfolders_output"
  else
    log_verbose "No sub-folders found under folder ${folder_id} or failed to list sub-folders" >&2
  fi
  
  # Combine direct projects and sub-folder projects
  local all_projects=""
  if [ -n "$direct_projects" ]; then
    all_projects="$direct_projects"
  fi
  
  if [ -n "$subfolder_projects" ]; then
    if [ -n "$all_projects" ]; then
      all_projects="${all_projects}"$'\n'"${subfolder_projects}"
    else
      all_projects="$subfolder_projects"
    fi
  fi
  
  # Remove duplicates and empty lines, then output
  local final_projects=""
  if [ -n "$all_projects" ]; then
    final_projects=$(echo "$all_projects" | sort -u | grep -v '^$')
    local final_count
    final_count=$(echo "$final_projects" | grep -c '^' 2>/dev/null || echo "0")
    log_verbose "Recursive discovery in folder ${folder_id} completed: ${final_count} total projects (${direct_count} direct + ${total_subfolder_projects} in sub-folders)" >&2
    
    if [[ -n "$EXCLUDE_PROJECT_PATTERNS" ]]; then
      final_projects=$(filter_excluded_projects "$final_projects" "$EXCLUDE_PROJECT_PATTERNS")
      local filtered_count
      filtered_count=$(echo "$final_projects" | grep -c '^' 2>/dev/null || echo "0")
      log_verbose "After exclusion filters in folder ${folder_id}: ${filtered_count} projects remaining" >&2
    fi

    echo "$final_projects"
  else
    log_verbose "Recursive discovery in folder ${folder_id} completed: 0 projects found" >&2
  fi
}

get_projects_under_organization_for_destroy() {
  local org_id="$1"
  local saved_exclude_project_patterns="$EXCLUDE_PROJECT_PATTERNS"
  local saved_include_label_filters="$INCLUDE_LABEL_FILTERS"
  local saved_exclude_label_filters="$EXCLUDE_LABEL_FILTERS"
  local saved_include_tag_filters="$INCLUDE_TAG_FILTERS"
  local saved_exclude_tag_filters="$EXCLUDE_TAG_FILTERS"
  local projects retVal

  EXCLUDE_PROJECT_PATTERNS=""
  INCLUDE_LABEL_FILTERS=""
  EXCLUDE_LABEL_FILTERS=""
  INCLUDE_TAG_FILTERS=""
  EXCLUDE_TAG_FILTERS=""

  set +e
  projects=$(get_projects_under_organization "$org_id")
  retVal=$?
  set -e

  EXCLUDE_PROJECT_PATTERNS="$saved_exclude_project_patterns"
  INCLUDE_LABEL_FILTERS="$saved_include_label_filters"
  EXCLUDE_LABEL_FILTERS="$saved_exclude_label_filters"
  INCLUDE_TAG_FILTERS="$saved_include_tag_filters"
  EXCLUDE_TAG_FILTERS="$saved_exclude_tag_filters"

  if [ $retVal -ne 0 ]; then
    return $retVal
  fi
  echo "$projects"
}

get_projects_under_folder_for_destroy() {
  local folder_id="$1"
  local saved_exclude_project_patterns="$EXCLUDE_PROJECT_PATTERNS"
  local projects retVal

  EXCLUDE_PROJECT_PATTERNS=""

  set +e
  projects=$(get_projects_under_folder "$folder_id")
  retVal=$?
  set -e

  EXCLUDE_PROJECT_PATTERNS="$saved_exclude_project_patterns"

  if [ $retVal -ne 0 ]; then
    return $retVal
  fi
  echo "$projects"
}

enable_apis_in_all_projects() {
  local projects="$1"
  shift
  local apis=("$@")
  
  if [[ -z "$projects" ]]; then
    log_warn "No projects found to enable APIs in"
    return
  fi
  
  # Count non-empty projects
  local project_count=0
  while IFS= read -r project; do
    [[ -z "$project" ]] && continue
    ((project_count+=1))
  done <<< "$projects"
  
  if [[ $project_count -eq 0 ]]; then
    log_info "No projects found to enable APIs in"
    return
  fi
  
  log_info "Enabling required APIs in ${project_count} projects..."
  
  while IFS= read -r project; do
    [[ -z "$project" ]] && continue
    log_info "Processing project: ${project}"
    
    # Check if project is accessible and active
    set +e
    gcloud projects describe "${project}" --format="value(lifecycleState)" --verbosity=error >/dev/null 2>&1
    local retVal=$?
    set -e
    
    if [ $retVal -ne 0 ]; then
      log_warn "Skipping project ${project} - cannot access or project may not exist"
      continue
    fi
    
    for api in "${apis[@]}"; do
      # Add error handling for individual API enabling
      set +e
      enable_api "${api}" "${project}"
      local api_retVal=$?
      set -e
      
      if [ $api_retVal -ne 0 ]; then
        log_warn "Failed to enable API ${api} in project ${project} - continuing with next API"
      fi
    done
  done <<< "$projects"
}

ensure_required_apis() {
  log_info "Checking required APIs …"
  
  # For source account type with organization or folder, enable APIs in all projects
  if [[ "$ACCOUNT_TYPE" == "source" ]]; then
    local all_apis=("${REQUIRED_APIS[@]}" "sqladmin.googleapis.com")
    if [[ "$ENABLE_GCS_INVENTORY_MANAGEMENT" == "true" ]]; then
      all_apis+=("storageinsights.googleapis.com")
    fi
    if [[ "$ENABLE_BIGQUERY_DISCOVERY" == "true" || "$ENABLE_BIGQUERY" == "true" ]]; then
      all_apis+=("bigquery.googleapis.com")
    fi
    local target_projects=""
    
    if [[ -n "$ORGANIZATION_ID" && -n "$FOLDER_ID" ]]; then
      log_info "Organization and folder specified - enabling APIs in all projects under folder ${FOLDER_ID}..."
      target_projects=$(get_projects_under_folder "$FOLDER_ID")
      if [[ -n "$INCLUDE_LABEL_FILTERS" || -n "$EXCLUDE_LABEL_FILTERS" || -n "$INCLUDE_TAG_FILTERS" || -n "$EXCLUDE_TAG_FILTERS" ]]; then
        target_projects=$(filter_projects_by_labels_and_tags "$target_projects" "$INCLUDE_LABEL_FILTERS" "$EXCLUDE_LABEL_FILTERS" "$INCLUDE_TAG_FILTERS" "$EXCLUDE_TAG_FILTERS")
      fi
      if [[ -n "$target_projects" ]]; then
        log_verbose "Found projects: $(echo "$target_projects" | tr '\n' ' ')"
        enable_apis_in_all_projects "$target_projects" "${all_apis[@]}"
      fi
      log_info "Enabling APIs in management project ${PROJECT_ID}..."
      for api in "${all_apis[@]}"; do
        enable_api "${api}" "${PROJECT_ID}"
      done
    elif [[ -n "$ORGANIZATION_ID" ]]; then
      log_info "Organization specified - enabling APIs in all projects under organization ${ORGANIZATION_ID}..."
      target_projects=$(get_projects_under_organization "$ORGANIZATION_ID")
      if [[ -n "$target_projects" ]]; then
        log_verbose "Found projects: $(echo "$target_projects" | tr '\n' ' ')"
        enable_apis_in_all_projects "$target_projects" "${all_apis[@]}"
      fi
      log_info "Enabling APIs in management project ${PROJECT_ID}..."
      for api in "${all_apis[@]}"; do
        enable_api "${api}" "${PROJECT_ID}"
      done
    else
      # Single project mode - enable APIs in management project only
      for api in "${REQUIRED_APIS[@]}"; do
        enable_api "${api}"
      done
      enable_api "sqladmin.googleapis.com"
      if [[ "$ENABLE_GCS_INVENTORY_MANAGEMENT" == "true" ]]; then
        enable_api "storageinsights.googleapis.com"
      fi
      if [[ "$ENABLE_BIGQUERY_DISCOVERY" == "true" || "$ENABLE_BIGQUERY" == "true" ]]; then
        enable_api "bigquery.googleapis.com"
      fi
    fi
  elif [[ "$ACCOUNT_TYPE" == "restore" ]]; then
    local all_apis=("${REQUIRED_APIS[@]}" "${RESTORE_REQUIRED_APIS[@]}")
    if [[ "$ENABLE_GWS" == "true" ]]; then
      all_apis+=("${GWS_RESTORE_REQUIRED_APIS[@]}")
    else
      all_apis+=("sqladmin.googleapis.com" "bigquery.googleapis.com")
    fi
    local target_projects=""

    if [[ -n "$ORGANIZATION_ID" && -n "$FOLDER_ID" ]]; then
      log_info "Organization and folder specified - enabling APIs in all projects under folder ${FOLDER_ID}..."
      target_projects=$(get_projects_under_folder "$FOLDER_ID")
      if [[ -n "$INCLUDE_LABEL_FILTERS" || -n "$EXCLUDE_LABEL_FILTERS" || -n "$INCLUDE_TAG_FILTERS" || -n "$EXCLUDE_TAG_FILTERS" ]]; then
        target_projects=$(filter_projects_by_labels_and_tags "$target_projects" "$INCLUDE_LABEL_FILTERS" "$EXCLUDE_LABEL_FILTERS" "$INCLUDE_TAG_FILTERS" "$EXCLUDE_TAG_FILTERS")
      fi
      if [[ -n "$target_projects" ]]; then
        log_verbose "Found projects: $(echo "$target_projects" | tr '\n' ' ')"
        enable_apis_in_all_projects "$target_projects" "${all_apis[@]}"
      fi
      log_info "Enabling APIs in management project ${PROJECT_ID}..."
      for api in "${all_apis[@]}"; do
        enable_api "${api}" "${PROJECT_ID}"
      done
    elif [[ -n "$ORGANIZATION_ID" ]]; then
      log_info "Organization specified - enabling APIs in all projects under organization ${ORGANIZATION_ID}..."
      target_projects=$(get_projects_under_organization "$ORGANIZATION_ID")
      if [[ -n "$target_projects" ]]; then
        log_verbose "Found projects: $(echo "$target_projects" | tr '\n' ' ')"
        enable_apis_in_all_projects "$target_projects" "${all_apis[@]}"
      fi
      log_info "Enabling APIs in management project ${PROJECT_ID}..."
      for api in "${all_apis[@]}"; do
        enable_api "${api}" "${PROJECT_ID}"
      done
    else
      # Single project mode - enable APIs in management project only
      for api in "${all_apis[@]}"; do
        enable_api "${api}"
      done
    fi
  fi
  
  log_success "✓ All required APIs are enabled"
}

handle_source_account() {
  local roles_scope
  local roles_scope_id
  local binding_roles_scope
  local binding_roles_scope_id
  if [[ -n "$ORGANIZATION_ID" ]]; then
    roles_scope="organization"
    roles_scope_id="$ORGANIZATION_ID"
    if [[ -n "$FOLDER_ID" ]]; then
      binding_roles_scope="folder"
      binding_roles_scope_id="$FOLDER_ID"
    else
      binding_roles_scope="organization"
      binding_roles_scope_id="$ORGANIZATION_ID"
    fi
  else
    roles_scope="project"
    roles_scope_id="$PROJECT_ID"
    binding_roles_scope="project"
    binding_roles_scope_id="$PROJECT_ID"
  fi

  local eon_account_id="$EON_ACCOUNT_ID"
  SA_NAME="eon-${eon_account_id:0:16}"
  SCANNING_SA_NAME="eon-scan-${eon_account_id:0:16}"

  SOURCE_SA_EMAIL="${SA_NAME}@${PROJECT_ID}.iam.gserviceaccount.com"
  log_info "Checking/creating Source SA: $SOURCE_SA_EMAIL"

  _gcloud iam service-accounts list --project="$PROJECT_ID" --filter="email:$SOURCE_SA_EMAIL" --format=json
  local count
  count=$(echo "$output" | jq length)
  if [ "$count" -eq 1 ]; then
    log_info "SA '$SOURCE_SA_EMAIL' already exists; ensuring version in description."
  else
    log_info "Creating SA '$SOURCE_SA_EMAIL'..."
    _gcloud iam service-accounts create "$SA_NAME" \
      --project="$PROJECT_ID" \
      --display-name "Eon Discovery [Full EON ID: $EON_ACCOUNT_ID]"
    log_success "Created SA: $SOURCE_SA_EMAIL"
  fi
  BACKUP_SA_NAME="${SA_NAME}-bk"
  BACKUP_SA_EMAIL="${BACKUP_SA_NAME}@${PROJECT_ID}.iam.gserviceaccount.com"

  _gcloud iam service-accounts list --project="$PROJECT_ID" \
          --filter="email:$BACKUP_SA_EMAIL" --format=json
  if [ "$(echo "$output" | jq length)" -eq 1 ]; then
      log_info "Backup SA '$BACKUP_SA_EMAIL' already exists; ensuring version in description."
  else
      log_info "Creating Backup SA '$BACKUP_SA_EMAIL'..."
      _gcloud iam service-accounts create "$BACKUP_SA_NAME" \
          --project="$PROJECT_ID" \
          --display-name "Eon Backup [Full EON ID: $EON_ACCOUNT_ID]"
      log_success "Created SA: $BACKUP_SA_EMAIL"
  fi

  local scanning_sa="${SCANNING_SA_NAME}@${SCANNING_PROJECT_ID}.iam.gserviceaccount.com"

  local discovery_role="EonDiscoveryRole"
  create_or_update_role "$discovery_role" "$DISCOVERY_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
  bind_role_to_sa "$SOURCE_SA_EMAIL" "$discovery_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"

  # Folder scope: the management project (where the SA lives) may sit outside
  # the folder being onboarded, so the folder-level binding above doesn't
  # reach it. Without a binding on the management project itself, Eon's
  # CheckCloudAccountPermissions cannot resolve the SA after impersonation
  # (iam.serviceAccounts.get on the SA, resourcemanager.projects.get on the
  # project). Bind the discovery role on the management project to cover both.
  if [[ -n "$FOLDER_ID" ]]; then
    log_info "Binding 'organizations/$ORGANIZATION_ID/roles/$discovery_role' to serviceAccount:$SOURCE_SA_EMAIL on management project '$PROJECT_ID'"
    _gcloud projects add-iam-policy-binding "$PROJECT_ID" \
      --member="serviceAccount:$SOURCE_SA_EMAIL" \
      --role="organizations/$ORGANIZATION_ID/roles/$discovery_role" \
      --condition=None
  fi

  if [[ -n "$ORGANIZATION_ID" ]]; then
    local org_discovery_role="EonOrgDiscoveryRole"
    create_or_update_role "$org_discovery_role" "$EON_ORG_DISCOVERY_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    bind_role_to_sa "$SOURCE_SA_EMAIL" "$org_discovery_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"
  fi

  if [[ "$ENABLE_GCE" == "true" ]]; then
    log_info "Setting up GCE roles..."
    # Machine-image perms in their own role so a bare-name condition can fence
    # them (the dedicated-role fix for EON-12807). Created + bound before the
    # capture role is updated below, so the perms are never dropped mid-upgrade.
    local gce_machine_image_role="EonGceMachineImageRole"
    create_or_update_role "$gce_machine_image_role" "$EON_GCE_MACHINE_IMAGE_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    bind_role_to_sa "$SOURCE_SA_EMAIL" "$gce_machine_image_role" \
      'title=RestrictEonMachineImages,expression=resource.name.extract("machineImages/{name}").startsWith("eon-"),description=OnlyEonMachineImages' \
      "$binding_roles_scope" "$binding_roles_scope_id"

    # Create capture role before slimming discovery to avoid a permission gap
    local gce_capture_role="EonGceCaptureRole"
    create_or_update_role "$gce_capture_role" "$EON_GCE_CAPTURE_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    bind_role_to_sa "$SOURCE_SA_EMAIL" "$gce_capture_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"

    local gce_backup_role="EonGceBackupRole"
    create_or_update_role "$gce_backup_role" "$EON_GCE_BACKUP_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    bind_role_to_sa "$BACKUP_SA_EMAIL" "$gce_backup_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"

    local gce_snapshots_role="EonGceSnapshotManagerRole"
    create_or_update_role "$gce_snapshots_role" "$EON_GCE_SNAPSHOT_MANAGER_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    bind_role_to_sa "$SOURCE_SA_EMAIL"  "$gce_snapshots_role" \
      'title=RestrictEonSnapshots,expression=resource.type == "compute.googleapis.com/Snapshot" && resource.name.extract("snapshots/{name}").startsWith("eon-"),description=OnlyEonServiceSnapshots' \
      "$binding_roles_scope" "$binding_roles_scope_id"

    local gce_consistency_group_role="EonGceConsistencyGroupRole"
    create_or_update_role "$gce_consistency_group_role" "$EON_GCE_CONSISTENCY_GROUP_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    bind_role_to_sa "$SOURCE_SA_EMAIL" "$gce_consistency_group_role" \
      'title=RestrictEonConsistencyGroups,expression=resource.name.extract("resourcePolicies/{name}").startsWith("eon-cg-"),description=OnlyEonConsistencyGroups' \
      "$binding_roles_scope" "$binding_roles_scope_id"

    local gce_instant_snapshot_group_role="EonGceInstantSnapshotGroupRole"
    create_or_update_role "$gce_instant_snapshot_group_role" "$EON_GCE_INSTANT_SNAPSHOT_GROUP_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    bind_role_to_sa "$SOURCE_SA_EMAIL" "$gce_instant_snapshot_group_role" \
      'title=RestrictEonInstantSnapshotGroups,expression=resource.name.extract("instantSnapshotGroups/{name}").startsWith("eon-isg-"),description=OnlyEonInstantSnapshotGroups' \
      "$binding_roles_scope" "$binding_roles_scope_id"

    if [[ "$ENABLE_GCE_NATIVE_BACKUP" == "true" ]]; then
      local vault_sa="eon-vault-${EON_ACCOUNT_ID:0:16}@${SCANNING_PROJECT_ID}.iam.gserviceaccount.com"
      local gce_vault_capture_role="EonGceVaultCaptureRole"
      create_or_update_role "$gce_vault_capture_role" "$EON_GCE_VAULT_CAPTURE_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
      bind_role_to_sa "$vault_sa" "$gce_vault_capture_role" \
        'title=RestrictEonInstantSnapshots,expression=resource.type != "compute.googleapis.com/InstantSnapshot" || resource.name.extract("instantSnapshots/{name}").startsWith("eon-"),description=RestrictInstantSnapshotAccessToEonResources' \
        "$binding_roles_scope" "$binding_roles_scope_id"
    fi

    log_success "✓ GCE roles configured"
  else
    log_info "Skipping GCE roles (ENABLE_GCE_BACKUP=$ENABLE_GCE)"
  fi

  if [[ "$ENABLE_GCS" == "true" ]]; then
    log_info "Setting up GCS roles..."
    local gcs_backup_role="EonGcsBackupRole"
    create_or_update_role "$gcs_backup_role" "$EON_GCS_BACKUP_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    bind_role_to_sa "$BACKUP_SA_EMAIL" "$gcs_backup_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"

    local gcs_inventory_user_role="EonGcsInventoryUserRole"
    create_or_update_role "$gcs_inventory_user_role" "$EON_GCS_INVENTORY_USER_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    bind_role_to_sa "$SOURCE_SA_EMAIL" "$gcs_inventory_user_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"

    if [[ "$ENABLE_GCS_NOTIFICATION_MANAGEMENT" == "true" ]]; then
      local gcs_notification_role="EonGcsNotificationManagerRole"
      create_or_update_role "$gcs_notification_role" "$EON_GCS_NOTIFICATION_MANAGER_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
      bind_role_to_sa "$SOURCE_SA_EMAIL" "$gcs_notification_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"
      log_success "✓ GCS bucket notification management permissions granted to $SOURCE_SA_EMAIL"
    fi

    if [[ "$ENABLE_GCS_INVENTORY_MANAGEMENT" == "true" ]]; then
      local gcs_inventory_bucket_creator_role="EonGcsInventoryBucketCreatorRole"
      create_or_update_role "$gcs_inventory_bucket_creator_role" "$EON_GCS_INVENTORY_BUCKET_CREATOR_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
      bind_role_to_sa "$SOURCE_SA_EMAIL" "$gcs_inventory_bucket_creator_role" \
        "$(build_inventory_bucket_creator_iam_condition)" \
        "$binding_roles_scope" "$binding_roles_scope_id"
      local gcs_inventory_manager_role="EonGcsInventoryManagerRole"
      create_or_update_role "$gcs_inventory_manager_role" "$EON_GCS_INVENTORY_MANAGER_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
      bind_role_to_sa "$SOURCE_SA_EMAIL" "$gcs_inventory_manager_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"
      local gcs_inventory_iam_manager_role="EonGcsInventoryIamManagerRole"
      create_or_update_role "$gcs_inventory_iam_manager_role" "$EON_GCS_INVENTORY_IAM_MANAGER_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
      remove_iam_member_bindings "serviceAccount:$SOURCE_SA_EMAIL" "$gcs_inventory_iam_manager_role" "$binding_roles_scope" "$binding_roles_scope_id"
      bind_role_to_sa "$SOURCE_SA_EMAIL" "$gcs_inventory_iam_manager_role" \
        "$(build_inventory_insights_collector_iam_condition)" \
        "$binding_roles_scope" "$binding_roles_scope_id"
      bind_role_to_sa "$SOURCE_SA_EMAIL" "$gcs_inventory_iam_manager_role" \
        "$(build_inventory_object_creator_iam_condition)" \
        "$binding_roles_scope" "$binding_roles_scope_id"
      log_success "✓ GCS inventory management permissions granted to $SOURCE_SA_EMAIL"
    fi
    log_success "✓ GCS roles configured"
  else
    log_info "Skipping GCS roles (ENABLE_GCS_BACKUP=$ENABLE_GCS)"
  fi

  if [[ "$ENABLE_CLOUDSQL" == "true" ]]; then
    log_info "Setting up CloudSQL roles..."
    # Bound before the discovery roles are updated below, so re-running over an older setup never leaves the SA without
    # the backup-run permissions the Cloud SQL discovery role used to hold.
    local cloudsql_backup_role="EonCloudSqlBackupRole"
    create_or_update_role "$cloudsql_backup_role" "$EON_CLOUDSQL_BACKUP_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    bind_role_to_sa "$SOURCE_SA_EMAIL" "$cloudsql_backup_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"


    local cloudsql_consumer_role="EonCloudSqlSnapshotConsumerRole"
    create_or_update_role "$cloudsql_consumer_role" "$EON_CLOUDSQL_SNAPSHOT_CONSUMER_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    bind_role_to_sa "$scanning_sa" "$cloudsql_consumer_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"

    log_success "✓ CloudSQL roles configured"
  else
    log_info "Skipping CloudSQL roles (ENABLE_CLOUDSQL_BACKUP=$ENABLE_CLOUDSQL)"
  fi

  if [[ "$ENABLE_BIGQUERY" == "true" ]]; then
    log_info "Setting up BigQuery roles..."
    # Add bigquery.tables.update only when auto change history is enabled.
    # Must be done here (after argument parsing), not at top-level variable definition.
    if [ "$ENABLE_BIGQUERY_AUTO_CHANGE_HISTORY" = "true" ]; then
      EON_BIGQUERY_BACKUP_ROLE_PERMISSIONS="${EON_BIGQUERY_BACKUP_ROLE_PERMISSIONS},bigquery.tables.update"
    fi

    local bigquery_backup_role="EonBigQueryBackupRole"
    create_or_update_role "$bigquery_backup_role" "$EON_BIGQUERY_BACKUP_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    bind_role_to_sa "$BACKUP_SA_EMAIL" "$bigquery_backup_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"
    log_success "✓ BigQuery roles configured"
  else
    log_info "Skipping BigQuery roles (ENABLE_BIGQUERY_BACKUP=$ENABLE_BIGQUERY)"
  fi

  # The discovery roles come after the backup blocks, so a backup role that took over permissions an older discovery
  # role held (GCE capture, Cloud SQL backup) is bound before the discovery role drops them.
  local discovery_toggle discovery_role discovery_permissions
  for discovery_toggle in ENABLE_GCE_DISCOVERY ENABLE_GCS_DISCOVERY ENABLE_CLOUDSQL_DISCOVERY ENABLE_BIGQUERY_DISCOVERY; do
    case "$discovery_toggle" in
      ENABLE_GCE_DISCOVERY) discovery_role="EonGceDiscoveryRole"; discovery_permissions="$EON_GCE_DISCOVERY_ROLE_PERMISSIONS" ;;
      ENABLE_GCS_DISCOVERY) discovery_role="EonGcsDiscoveryRole"; discovery_permissions="$EON_GCS_DISCOVERY_ROLE_PERMISSIONS" ;;
      ENABLE_CLOUDSQL_DISCOVERY) discovery_role="EonCloudSqlDiscoveryRole"; discovery_permissions="$EON_CLOUDSQL_DISCOVERY_ROLE_PERMISSIONS" ;;
      ENABLE_BIGQUERY_DISCOVERY) discovery_role="EonBigQueryDiscoveryRole"; discovery_permissions="$EON_BIGQUERY_DISCOVERY_ROLE_PERMISSIONS" ;;
    esac
    if [[ "${!discovery_toggle}" == "true" ]]; then
      log_info "Setting up $discovery_role..."
      create_or_update_role "$discovery_role" "$discovery_permissions" "$roles_scope" "$roles_scope_id"
      bind_role_to_sa "$SOURCE_SA_EMAIL" "$discovery_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"
    else
      log_info "Skipping $discovery_role ($discovery_toggle=${!discovery_toggle})"
    fi
  done
  log_success "✓ Discovery roles configured"

  if [[ "$ENABLE_SAP_HANA" == "true" ]]; then
    log_info "Setting up SAP HANA roles..."
    local sap_hana_role="EonSapHanaRole"
    create_or_update_role "$sap_hana_role" "$EON_SAP_HANA_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    bind_role_to_sa "$SOURCE_SA_EMAIL" "$sap_hana_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"
    log_success "✓ SAP HANA roles configured"

    log_info "Setting up SAP HANA Pub/Sub and Firestore infrastructure..."
    local sap_topic="eon-sap-hana-cmds-${PROJECT_ID}"
    local sap_sub="${sap_topic}-sub"
    local sap_db="${sap_topic}"

    _gcloud pubsub topics list --project="$PROJECT_ID" \
      --filter="name=projects/${PROJECT_ID}/topics/${sap_topic}" --format="value(name)"
    if [ -n "$output" ]; then
      log_info "Pub/Sub topic '$sap_topic' already exists."
    else
      _gcloud pubsub topics create "$sap_topic" --project="$PROJECT_ID"
      log_success "Created Pub/Sub topic: $sap_topic"
    fi

    _gcloud pubsub subscriptions list --project="$PROJECT_ID" \
      --filter="name=projects/${PROJECT_ID}/subscriptions/${sap_sub}" --format="value(name)"
    if [ -n "$output" ]; then
      log_info "Pub/Sub subscription '$sap_sub' already exists."
    else
      _gcloud pubsub subscriptions create "$sap_sub" \
        --project="$PROJECT_ID" \
        --topic="$sap_topic" \
        --ack-deadline=30 \
        --message-retention-duration=7d
      log_success "Created Pub/Sub subscription: $sap_sub"
    fi

    _gcloud firestore databases list --project="$PROJECT_ID" \
      --filter="name=projects/${PROJECT_ID}/databases/${sap_db}" --format="value(name)"
    if [ -n "$output" ]; then
      log_info "Firestore database '$sap_db' already exists."
    else
      _gcloud firestore databases create \
        --project="$PROJECT_ID" \
        --location="${FIRESTORE_LOCATION:-nam5}" \
        --database="$sap_db" \
        --type=firestore-native
      log_success "Created Firestore database: $sap_db"
    fi
    log_success "✓ SAP HANA infrastructure configured"
  else
    log_info "Skipping SAP HANA roles (ENABLE_SAP_HANA_BACKUP=$ENABLE_SAP_HANA)"
  fi

  # Update SA description with version and capabilities AFTER all roles are configured,
  # so that capabilities and IAM permissions stay in sync on partial failure.
  local gcs_notification_value="false"
  if [ "$ENABLE_GCS_NOTIFICATION_MANAGEMENT" = "true" ]; then
    gcs_notification_value="true"
  fi

  local gcs_inventory_value="false"
  if [ "$ENABLE_GCS" = "true" ] && [ "$ENABLE_GCS_INVENTORY_MANAGEMENT" = "true" ]; then
    gcs_inventory_value="true"
  fi

  local bigquery_auto_change_history_value="false"
  if [ "$ENABLE_BIGQUERY_AUTO_CHANGE_HISTORY" = "true" ]; then
    bigquery_auto_change_history_value="true"
  fi

  local description_json
  description_json=$(build_gcp_description_json \
    "$EON_GCP_SOURCE_VERSION" \
    "bash" \
    "gcs_notif" "$gcs_notification_value" \
    "gcs_inv" "$gcs_inventory_value" \
    "bq_change_history" "$bigquery_auto_change_history_value" \
    "gcs" "$ENABLE_GCS" \
    "gce" "$ENABLE_GCE" \
    "gce_native_backup" "$ENABLE_GCE_NATIVE_BACKUP" \
    "cloudsql" "$ENABLE_CLOUDSQL" \
    "bigquery" "$ENABLE_BIGQUERY" \
    "sap_hana" "$ENABLE_SAP_HANA" \
    "gce_disc" "$ENABLE_GCE_DISCOVERY" \
    "gcs_disc" "$ENABLE_GCS_DISCOVERY" \
    "cloudsql_disc" "$ENABLE_CLOUDSQL_DISCOVERY" \
    "bigquery_disc" "$ENABLE_BIGQUERY_DISCOVERY")
  _gcloud iam service-accounts update "$SOURCE_SA_EMAIL" --project="$PROJECT_ID" --description="$description_json"
  log_info "Set/Updated description for $SOURCE_SA_EMAIL to JSON format"

  log_info "Granting scanning SA '$scanning_sa' ability to impersonate $BACKUP_SA_EMAIL"
  _gcloud iam service-accounts add-iam-policy-binding "$BACKUP_SA_EMAIL" \
      --member="serviceAccount:$scanning_sa" \
      --role="roles/iam.serviceAccountTokenCreator"
  log_success "Scanning SA can now impersonate Eon Backup SA."

  log_info "Granting AWS control-plane GCP SA '$CONTROL_PLANE_SA' ability to impersonate $SOURCE_SA_EMAIL"
  _gcloud iam service-accounts add-iam-policy-binding "$SOURCE_SA_EMAIL" \
    --member="serviceAccount:$CONTROL_PLANE_SA" \
    --role="roles/iam.serviceAccountTokenCreator"
  log_success "AWS control plane GCP SA can now impersonate Eon Discovery SA."

  echo "--------------------------------------------------------------------------------"
  echo "[SOURCE ACCOUNT SETUP COMPLETE]"
  if [[ -n "$ORGANIZATION_ID" ]]; then
    echo "Organization ID:     $ORGANIZATION_ID"
  fi
  if [[ -n "$FOLDER_ID" ]]; then
    echo "Folder ID:           $FOLDER_ID"
  fi
  echo "Project:             $PROJECT_ID"
  echo "Eon Discovery SA:    $SOURCE_SA_EMAIL"
  echo "Eon Backup SA:       $BACKUP_SA_EMAIL"
  echo "Scanning Project ID: $SCANNING_PROJECT_ID"
  echo "Control Plane SA:    $CONTROL_PLANE_SA"
  echo "GCS CDC Enabled:     $ENABLE_GCS_NOTIFICATION_MANAGEMENT"
  echo "GCS Inventory Mgmt:  $ENABLE_GCS_INVENTORY_MANAGEMENT"
  if [[ -n "$EXCLUDE_PROJECT_PATTERNS" ]]; then
    echo "Exclude Patterns:    $EXCLUDE_PROJECT_PATTERNS"
  fi
  if [[ -n "$INCLUDE_LABEL_FILTERS" ]]; then
    echo "Include Labels:      $INCLUDE_LABEL_FILTERS"
  fi
  if [[ -n "$EXCLUDE_LABEL_FILTERS" ]]; then
    echo "Exclude Labels:      $EXCLUDE_LABEL_FILTERS"
  fi
  if [[ -n "$INCLUDE_TAG_FILTERS" ]]; then
    echo "Include Tags:        $INCLUDE_TAG_FILTERS"
  fi
  if [[ -n "$EXCLUDE_TAG_FILTERS" ]]; then
    echo "Exclude Tags:        $EXCLUDE_TAG_FILTERS"
  fi
  echo "--------------------------------------------------------------------------------"

  print_billing_issues_summary
  log_success "✔ Source account onboarding finished."
}

# ------------------ Restore account naming helpers -------------------------
# These helpers name the restore account's service accounts, node SA group, and custom
# roles. For a Google Workspace restore they return distinct names so they don't conflict
# with a separate standard GCP restore set up for the same Eon account in this project.

# restore_sa_id returns the restore SA id, using a distinct name for a Google Workspace restore.
restore_sa_id() {
  local eon_account_id="$1"
  if [[ "$ENABLE_GWS" == "true" ]]; then
    printf '%s-gws' "${eon_account_id:0:12}"
  else
    printf '%s' "${eon_account_id:0:16}"
  fi
}

# restore_role_name returns a restore custom-role id, using a distinct name for a Google Workspace restore.
restore_role_name() {
  if [[ "$ENABLE_GWS" == "true" ]]; then
    printf '%sGws' "$1"
  else
    printf '%s' "$1"
  fi
}

restore_catalog_role_name() {
  case "$1" in
    EonRestoreRole|EonRestoreFirestoreRole|EonRestoreVmServiceAccountManagerRole|EonBigQueryExportWritesRole|EonExportWritesRole|EonRestoreNodeRole|EonRestoreNodeFirestoreRole|EonSecretManagerRole|EonSecretsManagerRole|EonComputeManagerRole|EonBatchManagerRole|EonRestoreToSharedSubnetworkRole)
      restore_role_name "$1"
      ;;
    *)
      printf '%s' "$1"
      ;;
  esac
}

# restore_group_email_suffix returns the suffix inserted before "@" in the node SA group email (distinct for a Google Workspace restore, empty otherwise).
restore_group_email_suffix() {
  if [[ "$ENABLE_GWS" == "true" ]]; then
    printf -- '-gws'
  fi
}

# destroy_source_account reverses handle_source_account: remove IAM bindings,
# delete custom roles, delete service accounts, then delete SAP HANA data
# resources. Scope detection mirrors handle_source_account exactly.
destroy_source_account() {
  local roles_scope roles_scope_id binding_roles_scope binding_roles_scope_id
  if [[ -n "$ORGANIZATION_ID" ]]; then
    roles_scope="organization"
    roles_scope_id="$ORGANIZATION_ID"
    if [[ -n "$FOLDER_ID" ]]; then
      binding_roles_scope="folder"
      binding_roles_scope_id="$FOLDER_ID"
    else
      binding_roles_scope="organization"
      binding_roles_scope_id="$ORGANIZATION_ID"
    fi
  else
    roles_scope="project"
    roles_scope_id="$PROJECT_ID"
    binding_roles_scope="project"
    binding_roles_scope_id="$PROJECT_ID"
  fi

  local id16="${EON_ACCOUNT_ID:0:16}"
  local source_sa="eon-${id16}@${PROJECT_ID}.iam.gserviceaccount.com"
  local backup_sa="eon-${id16}-bk@${PROJECT_ID}.iam.gserviceaccount.com"
  local scanning_sa="eon-scan-${id16}@${SCANNING_PROJECT_ID}.iam.gserviceaccount.com"
  local vault_sa="eon-vault-${id16}@${SCANNING_PROJECT_ID}.iam.gserviceaccount.com"
  local source_target_members="serviceAccount:$source_sa"
  source_target_members+=$'\n'"serviceAccount:$backup_sa"
  source_target_members+=$'\n'"serviceAccount:$scanning_sa"
  source_target_members+=$'\n'"serviceAccount:$vault_sa"
  local source_project_emails="$source_sa"
  source_project_emails+=$'\n'"$backup_sa"
  local has_other_source_accounts=false
  if project_has_other_source_accounts "$PROJECT_ID" "$source_project_emails"; then
    has_other_source_accounts=true
  fi

  log_info "Removing Source IAM bindings (scope: $binding_roles_scope)..."
  # These lists map each role to the member SA it was bound to in
  # handle_source_account; SOURCE_ROLE_CATALOG is the authoritative flat list
  # used below for role deletion. Keep both in sync with the create handler.
  # Roles bound to the discovery (source) SA in handle_source_account.
  local source_sa_roles=(
    EonDiscoveryRole EonOrgDiscoveryRole
    EonGceCaptureRole EonGceMachineImageRole EonGceDiscoveryRole EonGceSnapshotManagerRole
    EonGceConsistencyGroupRole EonGceInstantSnapshotGroupRole
    EonGcsDiscoveryRole EonGcsInventoryUserRole EonGcsNotificationManagerRole EonGcsInventoryBucketCreatorRole EonGcsInventoryManagerRole EonGcsInventoryIamManagerRole
    EonCloudSqlDiscoveryRole EonCloudSqlBackupRole EonBigQueryDiscoveryRole EonSapHanaRole
  )
  # Roles bound to the backup SA.
  local backup_sa_roles=(EonGceBackupRole EonGcsBackupRole EonBigQueryBackupRole)

  local r
  for r in "${source_sa_roles[@]}"; do
    remove_binding "serviceAccount:$source_sa" "$r" "$binding_roles_scope" "$binding_roles_scope_id"
    remove_deleted_sa_bindings "$source_sa" "$r" "$binding_roles_scope" "$binding_roles_scope_id"
  done
  for r in "${backup_sa_roles[@]}"; do
    remove_binding "serviceAccount:$backup_sa" "$r" "$binding_roles_scope" "$binding_roles_scope_id"
    remove_deleted_sa_bindings "$backup_sa" "$r" "$binding_roles_scope" "$binding_roles_scope_id"
  done
  remove_binding "serviceAccount:$scanning_sa" "EonCloudSqlSnapshotConsumerRole" "$binding_roles_scope" "$binding_roles_scope_id"
  remove_deleted_sa_bindings "$scanning_sa" "EonCloudSqlSnapshotConsumerRole" "$binding_roles_scope" "$binding_roles_scope_id"
  remove_binding "serviceAccount:$vault_sa" "EonGceVaultCaptureRole" "$binding_roles_scope" "$binding_roles_scope_id"
  remove_deleted_sa_bindings "$vault_sa" "EonGceVaultCaptureRole" "$binding_roles_scope" "$binding_roles_scope_id"

  # Folder scope also bound EonDiscoveryRole on the management project itself.
  if [[ "$binding_roles_scope" == "folder" ]]; then
    remove_project_binding_with_full_role \
      "serviceAccount:$source_sa" \
      "organizations/$ORGANIZATION_ID/roles/EonDiscoveryRole" \
      "$PROJECT_ID"
    remove_deleted_sa_bindings_for_full_role \
      "$source_sa" \
      "organizations/$ORGANIZATION_ID/roles/EonDiscoveryRole" \
      "project" \
      "$PROJECT_ID"
  fi

  log_info "Deleting Source custom roles (scope: $roles_scope)..."
  for r in "${SOURCE_ROLE_CATALOG[@]}"; do
    if [[ "$binding_roles_scope" == "folder" ]]; then
      delete_role_if_no_other_bindings "$r" "$roles_scope" "$roles_scope_id" "$source_target_members" \
        "$binding_roles_scope" "$binding_roles_scope_id" \
        "project" "$PROJECT_ID"
    else
      delete_role_if_no_other_bindings "$r" "$roles_scope" "$roles_scope_id" "$source_target_members" \
        "$binding_roles_scope" "$binding_roles_scope_id"
    fi
  done

  # Deleting the SAs also removes their SA-level policy bindings (the
  # serviceAccountTokenCreator impersonation grants for scanning/control-plane
  # SAs), so those need no separate cleanup.
  log_info "Deleting Source service accounts..."
  local failures_before_source_sa_delete=$DESTROY_FAILURES
  delete_sa "$source_sa" "$PROJECT_ID"
  delete_sa "$backup_sa" "$PROJECT_ID"

  # SAP HANA data resources (no-op when SAP HANA was never enabled).
  if [[ "$has_other_source_accounts" == "true" ]]; then
    log_info "Skipping shared SAP HANA Pub/Sub/Firestore resources because other Eon Source accounts remain in project $PROJECT_ID."
  elif [[ "$DESTROY_FAILURES" -gt "$failures_before_source_sa_delete" ]]; then
    log_warn "Skipping shared SAP HANA Pub/Sub/Firestore resources because Source service account deletion failed."
  else
    local sap_topic="eon-sap-hana-cmds-${PROJECT_ID}"
    destroy_exec "Pub/Sub subscription ${sap_topic}-sub" -- pubsub subscriptions delete "${sap_topic}-sub" --project="$PROJECT_ID"
    destroy_exec "Pub/Sub topic ${sap_topic}" -- pubsub topics delete "$sap_topic" --project="$PROJECT_ID"
    destroy_exec "Firestore database ${sap_topic}" -- firestore databases delete --database="$sap_topic" --project="$PROJECT_ID"
  fi

  destroy_summary "SOURCE"
}

handle_restore_account() {
  # Determine scope for role creation and binding (same pattern as handle_source_account)
  local roles_scope
  local roles_scope_id
  local binding_roles_scope
  local binding_roles_scope_id
  if [[ -n "$ORGANIZATION_ID" ]]; then
    roles_scope="organization"
    roles_scope_id="$ORGANIZATION_ID"
    if [[ -n "$FOLDER_ID" ]]; then
      binding_roles_scope="folder"
      binding_roles_scope_id="$FOLDER_ID"
    else
      binding_roles_scope="organization"
      binding_roles_scope_id="$ORGANIZATION_ID"
    fi
  else
    roles_scope="project"
    roles_scope_id="$PROJECT_ID"
    binding_roles_scope="project"
    binding_roles_scope_id="$PROJECT_ID"
  fi

  local eon_account_id="$EON_ACCOUNT_ID"

  SA_NAME="eon-restore-$(restore_sa_id "$eon_account_id")"
  NODE_SA_NAME="${SA_NAME}nd"
  SCANNING_SA_NAME="eon-scan-${eon_account_id:0:16}"
  GWS_RESTORE_SA_NAME="eon-gws-${eon_account_id:0:16}-rs"

  RESTORE_SA_EMAIL="${SA_NAME}@${PROJECT_ID}.iam.gserviceaccount.com"
  log_info "Checking/creating Restore SA: $RESTORE_SA_EMAIL"

  _gcloud iam service-accounts list --project="$PROJECT_ID" --filter="email:$RESTORE_SA_EMAIL" --format=json
  local count
  count=$(echo "$output" | jq length)
  if [ "$count" -eq 1 ]; then
    log_info "SA '$RESTORE_SA_EMAIL' already exists; ensuring version in description."
  else
    log_info "Creating SA '$RESTORE_SA_EMAIL'..."
    _gcloud iam service-accounts create "$SA_NAME" \
      --project="$PROJECT_ID" \
      --display-name "Eon Restore [Full EON ID: $EON_ACCOUNT_ID]"
    log_success "Created SA: $RESTORE_SA_EMAIL"
  fi

  RESTORE_NODE_SA_EMAIL="${NODE_SA_NAME}@${PROJECT_ID}.iam.gserviceaccount.com"
  log_info "Checking/creating Restore Node SA: $RESTORE_NODE_SA_EMAIL"

  local restore_node_display="Eon Restore Node [Full EON ID: $EON_ACCOUNT_ID]"
  [[ "$ENABLE_GWS" == "true" ]] && restore_node_display="Eon GWS Restore Node [Full EON ID: $EON_ACCOUNT_ID]"

  _gcloud iam service-accounts list --project="$PROJECT_ID" --filter="email:$RESTORE_NODE_SA_EMAIL" --format=json
  local count
  count=$(echo "$output" | jq length)
  if [ "$count" -eq 1 ]; then
    log_info "SA '$RESTORE_NODE_SA_EMAIL' already exists; ensuring version in description."
  else
    log_info "Creating SA '$RESTORE_NODE_SA_EMAIL'..."
    _gcloud iam service-accounts create "$NODE_SA_NAME" \
      --project="$PROJECT_ID" \
      --display-name "$restore_node_display"
    log_success "Created SA: $RESTORE_NODE_SA_EMAIL"
  fi

  # Build list of all node SA emails. For org/folder scope, GCP Compute Engine can't
  # attach a SA from project A to a VM in project B, so we create a node SA in every
  # discovered project. For project scope this array contains only the management SA.
  ALL_NODE_SA_EMAILS=("$RESTORE_NODE_SA_EMAIL")

  if [[ "$roles_scope" == "organization" ]]; then
    local discovered_projects=""
    if [[ -n "$FOLDER_ID" ]]; then
      discovered_projects=$(get_projects_under_folder "$FOLDER_ID")
      if [[ -n "$INCLUDE_LABEL_FILTERS" || -n "$EXCLUDE_LABEL_FILTERS" || -n "$INCLUDE_TAG_FILTERS" || -n "$EXCLUDE_TAG_FILTERS" ]]; then
        discovered_projects=$(filter_projects_by_labels_and_tags "$discovered_projects" "$INCLUDE_LABEL_FILTERS" "$EXCLUDE_LABEL_FILTERS" "$INCLUDE_TAG_FILTERS" "$EXCLUDE_TAG_FILTERS")
      fi
    else
      discovered_projects=$(get_projects_under_organization "$ORGANIZATION_ID")
    fi
    if [[ -n "$discovered_projects" ]]; then
      while IFS= read -r project; do
        [[ -z "$project" ]] && continue
        [[ "$project" == "$PROJECT_ID" ]] && continue  # skip management project (already created above)
        local node_sa_email="${NODE_SA_NAME}@${project}.iam.gserviceaccount.com"
        log_info "Creating restore node SA in project: $project"
        _gcloud iam service-accounts list --project="$project" --filter="email:$node_sa_email" --format=json
        local sa_count
        sa_count=$(echo "$output" | jq length)
        if [ "$sa_count" -eq 0 ]; then
          _gcloud iam service-accounts create "$NODE_SA_NAME" \
            --project="$project" \
            --display-name "$restore_node_display"
          log_success "Created node SA: $node_sa_email"
        else
          log_info "Node SA '$node_sa_email' already exists"
        fi
        ALL_NODE_SA_EMAILS+=("$node_sa_email")
      done <<< "$discovered_projects"
    fi
  fi

  # For org/folder scope, add all node SAs to the Google Group that the
  # preflight (preflight_group_create_permission) already created/verified.
  # Grouping all node SAs into one Cloud Identity group collapses N IAM
  # bindings per role into 1, avoiding the IAM policy size limit.
  if [[ "$roles_scope" == "organization" ]]; then
    for node_sa in "${ALL_NODE_SA_EMAILS[@]}"; do
      add_sa_to_group "$node_sa" "$NODE_SA_GROUP_EMAIL"
    done
    log_success "✓ All node SAs added to group $NODE_SA_GROUP_EMAIL"
  fi

  # Create org discovery role if org-scoped
  if [[ "$roles_scope" == "organization" ]]; then
    log_info "Creating organization-level discovery role..."
    # Role that lets Eon list the projects and folders in your organization. It is
    # shared with Eon's other onboarding flows, so it keeps a single common name.
    create_or_update_role "EonOrgDiscoveryRole" "$EON_ORG_DISCOVERY_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    bind_role_to_sa "$RESTORE_SA_EMAIL" "EonOrgDiscoveryRole" "None" "$binding_roles_scope" "$binding_roles_scope_id"
  fi

  local restore_role; restore_role="$(restore_role_name "EonRestoreRole")"
  create_or_update_role "$restore_role" "$RESTORE_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
  bind_role_to_sa "$RESTORE_SA_EMAIL" "$restore_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"

  local restore_firestore_role; restore_firestore_role="$(restore_role_name "EonRestoreFirestoreRole")"
  create_or_update_role "$restore_firestore_role" "$EON_RESTORE_FIRESTORE_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
  bind_role_to_sa "$RESTORE_SA_EMAIL" "$restore_firestore_role" \
    "$(build_restore_firestore_iam_condition)" \
    "$binding_roles_scope" "$binding_roles_scope_id"

  local restore_vm_sa_manager_role; restore_vm_sa_manager_role="$(restore_role_name "EonRestoreVmServiceAccountManagerRole")"
  create_or_update_role "$restore_vm_sa_manager_role" "$EON_RESTORE_VM_SERVICE_ACCOUNT_MANAGER_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
  remove_iam_member_bindings "serviceAccount:$RESTORE_SA_EMAIL" "$restore_vm_sa_manager_role" "$binding_roles_scope" "$binding_roles_scope_id"
  bind_role_to_sa "$RESTORE_SA_EMAIL" "$restore_vm_sa_manager_role" \
    "$(build_eon_managed_instances_iam_condition "RestrictEonRestoreVmServiceAccount" "OnlyEonManagedInstances")" \
    "$binding_roles_scope" "$binding_roles_scope_id"

  if [[ "$ENABLE_BIGQUERY" == "true" ]]; then
    local restore_storage_write_access_role; restore_storage_write_access_role="$(restore_role_name "EonBigQueryExportWritesRole")"
    create_or_update_role "$restore_storage_write_access_role" "$RESTORE_STORAGE_WRITE_ACCESS_ROLE" "$roles_scope" "$roles_scope_id"
    local export_writes_expr="resource.service == \"storage.googleapis.com\""
    if [[ "$binding_roles_scope" != "project" ]]; then
      export_writes_expr="resource.type == \"storage.googleapis.com/Object\" && resource.name.startsWith(\"projects/_/buckets/eon-\")"
    fi
    local export_writes_condition="title=EonBigQueryExportWrites,description=WriteResultsFromBigQuery,expression=$export_writes_expr"
    local legacy_export_writes_role; legacy_export_writes_role="$(restore_role_name "EonExportWritesRole")"
    remove_iam_member_bindings "serviceAccount:$RESTORE_SA_EMAIL" "$restore_storage_write_access_role" "$binding_roles_scope" "$binding_roles_scope_id"
    remove_iam_member_bindings "serviceAccount:$RESTORE_SA_EMAIL" "$legacy_export_writes_role" "$binding_roles_scope" "$binding_roles_scope_id"
    bind_role_to_sa "$RESTORE_SA_EMAIL" "$restore_storage_write_access_role" \
      "$export_writes_condition" \
      "$binding_roles_scope" "$binding_roles_scope_id"
  fi

  local restore_node_role; restore_node_role="$(restore_role_name "EonRestoreNodeRole")"
  create_or_update_role "$restore_node_role" "$RESTORE_NODE_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
  local restore_node_firestore_role; restore_node_firestore_role="$(restore_role_name "EonRestoreNodeFirestoreRole")"
  create_or_update_role "$restore_node_firestore_role" "$EON_RESTORE_NODE_FIRESTORE_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"

  if [[ "$ENABLE_BIGQUERY" == "true" ]]; then
    log_info "Setting up BigQuery restore node role..."
    local bigquery_restore_node_role; bigquery_restore_node_role="EonBigQueryRestoreNodeRole"
    create_or_update_role "$bigquery_restore_node_role" "$EON_BIGQUERY_RESTORE_NODE_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    log_success "BigQuery restore node role configured"
  fi

  if [[ "$ENABLE_GCS" == "true" ]]; then
    log_info "Setting up GCS restore node role..."
    local gcs_restore_node_role; gcs_restore_node_role="EonGcsRestoreNodeRole"
    create_or_update_role "$gcs_restore_node_role" "$EON_GCS_RESTORE_NODE_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    log_success "GCS restore node role configured"
  fi

  if [[ "$ENABLE_GCE" == "true" ]]; then
    log_info "Setting up GCE restore role..."
    local gce_restore_role; gce_restore_role="EonGceRestoreRole"
    create_or_update_role "$gce_restore_role" "$EON_GCE_RESTORE_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    bind_role_to_sa "$RESTORE_SA_EMAIL" "$gce_restore_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"
    log_success "GCE restore role configured"

    log_info "Setting up GCE restore node role..."
    local gce_restore_node_role; gce_restore_node_role="EonGceRestoreNodeRole"
    create_or_update_role "$gce_restore_node_role" "$EON_GCE_RESTORE_NODE_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    log_success "GCE restore node role configured"

    log_info "Setting up Eon-managed instances tag..."
    local managed_tag_namespaced managed_tag_parent managed_tag_key_name managed_tag_value_name
    managed_tag_namespaced="$(gce_managed_tag_namespaced_key)"
    if [[ -n "$ORGANIZATION_ID" ]]; then
      managed_tag_parent="organizations/$ORGANIZATION_ID"
    else
      managed_tag_parent="projects/$PROJECT_ID"
    fi
    if _gcloud_try resource-manager tags keys describe "$managed_tag_namespaced" --format="value(name)"; then
      managed_tag_key_name="$output"
      log_info "Tag key '$managed_tag_namespaced' already exists."
    else
      _gcloud resource-manager tags keys create "eon-managed-${EON_ACCOUNT_ID:0:8}" \
        --parent="$managed_tag_parent" \
        --description="Marks GCE instances managed by Eon restore jobs ($EON_ACCOUNT_ID)" \
        --format="value(name)"
      managed_tag_key_name="$output"
    fi
    if _gcloud_try resource-manager tags values describe "$managed_tag_namespaced/true" --format="value(name)"; then
      managed_tag_value_name="$output"
      log_info "Tag value '$managed_tag_namespaced/true' already exists."
    else
      _gcloud resource-manager tags values create "true" \
        --parent="$managed_tag_key_name" \
        --description="Instance is managed by an Eon restore job" \
        --format="value(name)"
      managed_tag_value_name="$output"
    fi
    _gcloud resource-manager tags values add-iam-policy-binding "$managed_tag_value_name" \
      --member="serviceAccount:$RESTORE_SA_EMAIL" \
      --role="roles/resourcemanager.tagUser"
    log_success "Eon-managed instances tag configured ($managed_tag_value_name)"
  fi

  local secrets_role; secrets_role="$(restore_role_name "EonSecretsManagerRole")"
  create_or_update_role "$secrets_role" "$EON_SECRETS_MANAGER_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
  bind_role_to_sa "$RESTORE_SA_EMAIL"  "$secrets_role" \
    "title=RestrictEonSecrets,description=OnlyEonServiceSecrets,expression=resource.name.extract(\"secrets/{secret}\").startsWith(\"eon-\")" \
    "$binding_roles_scope" "$binding_roles_scope_id"

  # Bind node SA roles — use Google Group for org/folder scope, individual SAs for project scope
  if [[ "$roles_scope" == "organization" ]]; then
    log_info "Binding node SA roles to group $NODE_SA_GROUP_EMAIL..."
    bind_role_to_group "$NODE_SA_GROUP_EMAIL" "$restore_node_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"
    bind_role_to_group "$NODE_SA_GROUP_EMAIL" "$restore_node_firestore_role" \
      "$(build_restore_firestore_iam_condition)" \
      "$binding_roles_scope" "$binding_roles_scope_id"
    bind_role_to_group "$NODE_SA_GROUP_EMAIL" "$secrets_role" \
      "title=RestrictEonSecretsNode,description=OnlyEonServiceSecretsNode,expression=resource.name.extract(\"secrets/{secret}\").startsWith(\"eon-\")" \
      "$binding_roles_scope" "$binding_roles_scope_id"
    if [[ "$ENABLE_BIGQUERY" == "true" ]]; then
      bind_role_to_group "$NODE_SA_GROUP_EMAIL" "$bigquery_restore_node_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"
    fi
    if [[ "$ENABLE_GCS" == "true" ]]; then
      local gcs_restore_node_condition="None"
      if [[ "$RESTRICT_TARGET_RESTORE_GCS_BY_TAG" == "true" ]]; then
        gcs_restore_node_condition="$(build_restore_gcs_target_tag_iam_condition)"
        remove_iam_member_bindings "group:$NODE_SA_GROUP_EMAIL" "$gcs_restore_node_role" "$binding_roles_scope" "$binding_roles_scope_id"
      fi
      bind_role_to_group "$NODE_SA_GROUP_EMAIL" "$gcs_restore_node_role" "$gcs_restore_node_condition" "$binding_roles_scope" "$binding_roles_scope_id"
    fi
    if [[ "$ENABLE_GCE" == "true" ]]; then
      bind_role_to_group "$NODE_SA_GROUP_EMAIL" "$gce_restore_node_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"
    fi
    log_success "✓ Node SA roles bound to group"
  else
    for node_sa in "${ALL_NODE_SA_EMAILS[@]}"; do
      bind_role_to_sa "$node_sa" "$restore_node_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"
      bind_role_to_sa "$node_sa" "$restore_node_firestore_role" \
        "$(build_restore_firestore_iam_condition)" \
        "$binding_roles_scope" "$binding_roles_scope_id"
      bind_role_to_sa "$node_sa" "$secrets_role" \
        "title=RestrictEonSecretsNode,description=OnlyEonServiceSecretsNode,expression=resource.name.extract(\"secrets/{secret}\").startsWith(\"eon-\")" \
        "$binding_roles_scope" "$binding_roles_scope_id"
      if [[ "$ENABLE_BIGQUERY" == "true" ]]; then
        bind_role_to_sa "$node_sa" "$bigquery_restore_node_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"
      fi
      if [[ "$ENABLE_GCS" == "true" ]]; then
        local gcs_restore_node_condition="None"
        if [[ "$RESTRICT_TARGET_RESTORE_GCS_BY_TAG" == "true" ]]; then
          gcs_restore_node_condition="$(build_restore_gcs_target_tag_iam_condition)"
          remove_iam_member_bindings "serviceAccount:$node_sa" "$gcs_restore_node_role" "$binding_roles_scope" "$binding_roles_scope_id"
        fi
        bind_role_to_sa "$node_sa" "$gcs_restore_node_role" "$gcs_restore_node_condition" "$binding_roles_scope" "$binding_roles_scope_id"
      fi
      if [[ "$ENABLE_GCE" == "true" ]]; then
        bind_role_to_sa "$node_sa" "$gce_restore_node_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"
      fi
    done
  fi

  # When the management project is outside the folder/org, node SAs and the restore SA
  # need project-level EonRestoreNodeRole on the management project for Firestore access.
  if [[ "$roles_scope" == "organization" ]]; then
    local org_restore_firestore_role="organizations/$roles_scope_id/roles/$restore_firestore_role"
    local org_node_firestore_role="organizations/$roles_scope_id/roles/$restore_node_firestore_role"
    log_info "Granting Firestore access on management project $PROJECT_ID..."
    _gcloud projects add-iam-policy-binding "$PROJECT_ID" \
      --member="serviceAccount:$RESTORE_SA_EMAIL" \
      --role="$org_restore_firestore_role" \
      --condition="$(build_restore_firestore_iam_condition)"
    _gcloud projects add-iam-policy-binding "$PROJECT_ID" \
      --member="group:$NODE_SA_GROUP_EMAIL" \
      --role="$org_node_firestore_role" \
      --condition="$(build_restore_firestore_iam_condition)"
  fi

  local compute_role; compute_role="$(restore_role_name "EonComputeManagerRole")"
  create_or_update_role "$compute_role" "$EON_COMPUTE_MANAGER_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
  remove_iam_member_bindings "serviceAccount:$RESTORE_SA_EMAIL" "$compute_role" "$binding_roles_scope" "$binding_roles_scope_id"
  bind_role_to_sa "$RESTORE_SA_EMAIL"  "$compute_role" \
    "$(build_eon_managed_instances_iam_condition "RestrictEonComputeInstances" "OnlyEonServiceResources")" \
    "$binding_roles_scope" "$binding_roles_scope_id"

  local batch_role; batch_role="$(restore_role_name "EonBatchManagerRole")"
  create_or_update_role "$batch_role" "$EON_BATCH_MANAGER_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
  bind_role_to_sa "$RESTORE_SA_EMAIL" "$batch_role" \
    "title=RestrictEonBatchJobs,description=OnlyEonBatchJobs,expression=resource.name.extract(\"/jobs/{job}\") == \"\" || resource.name.extract(\"/jobs/{job}\").startsWith(\"eon-\")" \
    "$binding_roles_scope" "$binding_roles_scope_id"

  if [[ "$ENABLE_BIGQUERY" == "true" ]]; then
    log_info "Setting up BigQuery restore role..."
    local bigquery_restore_role; bigquery_restore_role="EonBigQueryRestoreRole"
    create_or_update_role "$bigquery_restore_role" "$EON_BIGQUERY_RESTORE_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    bind_role_to_sa "$RESTORE_SA_EMAIL" "$bigquery_restore_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"
    log_success "BigQuery restore role configured"
  fi

  if [[ "$ENABLE_CLOUDSQL" == "true" ]]; then
    log_info "Setting up CloudSQL restore role..."
    local cloudsql_restore_role; cloudsql_restore_role="EonCloudSqlRestoreRole"
    create_or_update_role "$cloudsql_restore_role" "$EON_CLOUDSQL_RESTORE_ROLE_PERMISSIONS" "$roles_scope" "$roles_scope_id"
    bind_role_to_sa "$RESTORE_SA_EMAIL" "$cloudsql_restore_role" "None" "$binding_roles_scope" "$binding_roles_scope_id"
    log_success "CloudSQL restore role configured"
  fi

  GWS_RESTORE_SA_EMAIL=""
  if [[ "$ENABLE_GWS" == "true" ]]; then
    # Service account Eon uses to restore your data back into Google Workspace (Gmail and Drive).
    GWS_RESTORE_SA_EMAIL="${GWS_RESTORE_SA_NAME}@${PROJECT_ID}.iam.gserviceaccount.com"
    log_info "Checking/creating Google Workspace Restore SA: $GWS_RESTORE_SA_EMAIL"
    _gcloud iam service-accounts list --project="$PROJECT_ID" --filter="email:$GWS_RESTORE_SA_EMAIL" --format=json
    local gws_count
    gws_count=$(echo "$output" | jq length)
    if [ "$gws_count" -eq 1 ]; then
      log_info "SA '$GWS_RESTORE_SA_EMAIL' already exists."
    else
      log_info "Creating SA '$GWS_RESTORE_SA_EMAIL'..."
      _gcloud iam service-accounts create "$GWS_RESTORE_SA_NAME" \
        --project="$PROJECT_ID" \
        --display-name "Eon GWS Restore [Full EON ID: $EON_ACCOUNT_ID]" \
        --description "Used by Eon to restore data back into your Google Workspace."
      log_success "Created SA: $GWS_RESTORE_SA_EMAIL"
    fi
  fi

  # Build and set JSON description for restore SA (after all roles are created/bound,
  # so the description never advertises capabilities that aren't actually granted)
  local restore_description_json node_sa_group_description_value=""
  if [[ "$USER_PROVIDED_NODE_SA_GROUP_EMAIL" == "true" ]]; then
    node_sa_group_description_value="${NODE_SA_GROUP_EMAIL:-}"
  fi
  restore_description_json=$(build_gcp_description_json \
    "$EON_GCP_RESTORE_VERSION" \
    "bash" \
    "gcs" "$ENABLE_GCS" \
    "gce" "$ENABLE_GCE" \
    "gce_native_backup" "$ENABLE_GCE_NATIVE_BACKUP" \
    "cloudsql" "$ENABLE_CLOUDSQL" \
    "bigquery" "$ENABLE_BIGQUERY" \
    "sap_hana" "$ENABLE_SAP_HANA" \
    "gws" "$ENABLE_GWS" \
    "gcs_restore_tag" "$RESTRICT_TARGET_RESTORE_GCS_BY_TAG" \
    "host_project_id" "${HOST_PROJECT_ID:-}" \
    "node_sa_group" "$node_sa_group_description_value")
  _gcloud iam service-accounts update "$RESTORE_SA_EMAIL" --project="$PROJECT_ID" --description="$restore_description_json"
  log_info "Set/Updated description for $RESTORE_SA_EMAIL to JSON format"

  # SA impersonation bindings (always SA-level, scope-independent)
  log_info "Granting restore SA '$RESTORE_SA_EMAIL' the ability to impersonate itself"
  _gcloud iam service-accounts add-iam-policy-binding "$RESTORE_SA_EMAIL" \
    --member="serviceAccount:$RESTORE_SA_EMAIL" \
    --role="roles/iam.serviceAccountTokenCreator"
  log_success "Restore SA can now impersonate itself."

  # Grant serviceAccountUser and self-impersonation for ALL node SAs.
  # SA-level bindings require --project matching the SA's home project.
  for node_sa in "${ALL_NODE_SA_EMAILS[@]}"; do
    local sa_project="${node_sa#*@}"
    sa_project="${sa_project%.iam.gserviceaccount.com}"

    log_info "Granting restore SA '$RESTORE_SA_EMAIL' the ability to assign $node_sa to resources"
    _gcloud iam service-accounts add-iam-policy-binding "$node_sa" \
      --project="$sa_project" \
      --member="serviceAccount:$RESTORE_SA_EMAIL" \
      --role="roles/iam.serviceAccountUser"
    log_success "Restore SA can now assign $node_sa to resources."

    log_info "Granting restore node SA '$node_sa' the ability to impersonate itself"
    _gcloud iam service-accounts add-iam-policy-binding "$node_sa" \
      --project="$sa_project" \
      --member="serviceAccount:$node_sa" \
      --role="roles/iam.serviceAccountTokenCreator"
    log_success "Restore Node SA $node_sa can now impersonate itself."

    if [[ "$ENABLE_GWS" == "true" ]]; then
      log_info "Granting restore node SA '$node_sa' the ability to impersonate $GWS_RESTORE_SA_EMAIL"
      _gcloud iam service-accounts add-iam-policy-binding "$GWS_RESTORE_SA_EMAIL" \
        --project="$PROJECT_ID" \
        --member="serviceAccount:$node_sa" \
        --role="roles/iam.serviceAccountTokenCreator"
      log_success "Restore Node SA $node_sa can now impersonate $GWS_RESTORE_SA_EMAIL."
    fi
  done

  log_info "Granting AWS control-plane GCP SA '$CONTROL_PLANE_SA' ability to impersonate $RESTORE_SA_EMAIL"
  _gcloud iam service-accounts add-iam-policy-binding "$RESTORE_SA_EMAIL" \
    --member="serviceAccount:$CONTROL_PLANE_SA" \
    --role="roles/iam.serviceAccountTokenCreator"
  log_success "AWS control plane GCP SA can now impersonate Restore SA (via WIF)."

  if [[ "$ENABLE_GWS" == "true" ]]; then
    # Lets Eon confirm it can access your Google Workspace during onboarding and
    # restore data back into it.
    log_info "Granting AWS control-plane GCP SA '$CONTROL_PLANE_SA' ability to impersonate $GWS_RESTORE_SA_EMAIL"
    _gcloud iam service-accounts add-iam-policy-binding "$GWS_RESTORE_SA_EMAIL" \
      --project="$PROJECT_ID" \
      --member="serviceAccount:$CONTROL_PLANE_SA" \
      --role="roles/iam.serviceAccountTokenCreator"
    log_success "AWS control plane GCP SA can now impersonate GWS Restore SA (via WIF)."
  fi

  if [[ -n "$HOST_PROJECT_ID" ]]; then
    log_info "Setting up Shared VPC support with host project: $HOST_PROJECT_ID"

    local shared_subnet_role; shared_subnet_role="$(restore_role_name "EonRestoreToSharedSubnetworkRole")"
    create_or_update_role_in_project "$shared_subnet_role" "$EON_RESTORE_TO_SHARED_SUBNETWORK_PERMISSIONS" "$HOST_PROJECT_ID"

    log_info "Binding shared subnet role to restore SA in host project"
    _gcloud projects add-iam-policy-binding "$HOST_PROJECT_ID" \
      --member="serviceAccount:$RESTORE_SA_EMAIL" \
      --role="projects/$HOST_PROJECT_ID/roles/$shared_subnet_role" \
      --condition=None
  fi

  # This database is shared across restore accounts in this project; only the
  # primary account (is_primary_restore_account) creates it, the rest reuse it.
  DB_ID="eon-cmd-results"
  if [[ "$IS_PRIMARY_RESTORE_ACCOUNT" == "true" ]]; then
    # Check if the Firestore database already exists
    if gcloud firestore databases list \
         --project="$PROJECT_ID" \
         --filter="name=projects/${PROJECT_ID}/databases/${DB_ID}" \
         --format="value(name)" | grep -q "^projects/${PROJECT_ID}/databases/${DB_ID}$"; then
      log_info "Firestore database '$DB_ID' already exists in project '$PROJECT_ID'. Skipping creation."
    else
      log_info "Creating Firestore database '$DB_ID' in region '$FIRESTORE_LOCATION'…"
      _gcloud firestore databases create \
        --project="$PROJECT_ID" \
        --location="$FIRESTORE_LOCATION" \
        --database="$DB_ID" \
        --type=firestore-native
      log_success "✅ Database '$DB_ID' created."
    fi
  else
    # Reusing the shared resources created by another restore account in this
    # project. Verify they exist before continuing, so this account does not end
    # up referencing a Firestore database that was never created.
    if gcloud firestore databases list \
         --project="$PROJECT_ID" \
         --filter="name=projects/${PROJECT_ID}/databases/${DB_ID}" \
         --format="value(name)" | grep -q "^projects/${PROJECT_ID}/databases/${DB_ID}$"; then
      log_info "is_primary_restore_account is false; reusing existing Firestore database '$DB_ID' in project '$PROJECT_ID'."
    else
      log_error "Error: IS_PRIMARY_RESTORE_ACCOUNT is false but shared Firestore database '$DB_ID' does not exist in project '$PROJECT_ID'. Run the primary restore account with IS_PRIMARY_RESTORE_ACCOUNT=true."
      exit 1
    fi
  fi

  echo "--------------------------------------------------------------------------------"
  echo "[RESTORE ACCOUNT SETUP COMPLETE]"
  echo "Project:              $PROJECT_ID"
  if [[ -n "$ORGANIZATION_ID" ]]; then
    echo "Organization ID:      $ORGANIZATION_ID"
  fi
  if [[ -n "$FOLDER_ID" ]]; then
    echo "Folder ID:            $FOLDER_ID"
  fi
  echo "Roles Scope:          $roles_scope"
  echo "Binding Scope:        $binding_roles_scope"
  echo "Restore SA:           $RESTORE_SA_EMAIL"
  echo "Restore Node SA:      $RESTORE_NODE_SA_EMAIL"
  if [[ -n "$GWS_RESTORE_SA_EMAIL" ]]; then
    echo "GWS Restore SA:       $GWS_RESTORE_SA_EMAIL"
  fi
  if [[ -n "$NODE_SA_GROUP_EMAIL" ]]; then
    echo "Node SA Group:        $NODE_SA_GROUP_EMAIL"
  fi
  echo "Control Plane SA:     $CONTROL_PLANE_SA"
  if [[ -n "$HOST_PROJECT_ID" ]]; then
    echo "Host Project ID:      $HOST_PROJECT_ID (Shared VPC)"
  fi
  if [[ -n "$INCLUDE_LABEL_FILTERS" ]]; then
    echo "Include Labels:       $INCLUDE_LABEL_FILTERS"
  fi
  if [[ -n "$EXCLUDE_LABEL_FILTERS" ]]; then
    echo "Exclude Labels:       $EXCLUDE_LABEL_FILTERS"
  fi
  if [[ -n "$INCLUDE_TAG_FILTERS" ]]; then
    echo "Include Tags:         $INCLUDE_TAG_FILTERS"
  fi
  if [[ -n "$EXCLUDE_TAG_FILTERS" ]]; then
    echo "Exclude Tags:         $EXCLUDE_TAG_FILTERS"
  fi
  echo "--------------------------------------------------------------------------------"

  print_billing_issues_summary
  log_success "✔ Restore account onboarding complete."
}

# destroy_restore_account reverses handle_restore_account: remove IAM bindings
# (restore SA + node SAs/group + host project), delete custom roles, delete all
# service accounts (incl. per-project node SAs swept across org/folder), then
# delete the Firestore command-results database. Scope detection and project
# enumeration mirror handle_restore_account exactly.
destroy_restore_account() {
  local roles_scope roles_scope_id binding_roles_scope binding_roles_scope_id
  if [[ -n "$ORGANIZATION_ID" ]]; then
    roles_scope="organization"
    roles_scope_id="$ORGANIZATION_ID"
    if [[ -n "$FOLDER_ID" ]]; then
      binding_roles_scope="folder"
      binding_roles_scope_id="$FOLDER_ID"
    else
      binding_roles_scope="organization"
      binding_roles_scope_id="$ORGANIZATION_ID"
    fi
  else
    roles_scope="project"
    roles_scope_id="$PROJECT_ID"
    binding_roles_scope="project"
    binding_roles_scope_id="$PROJECT_ID"
  fi

  detect_restore_gws_for_destroy

  local restore_sa_id_value
  restore_sa_id_value="$(restore_sa_id "$EON_ACCOUNT_ID")"
  local restore_sa="eon-restore-${restore_sa_id_value}@${PROJECT_ID}.iam.gserviceaccount.com"
  local node_sa_name="eon-restore-${restore_sa_id_value}nd"
  local node_sa="${node_sa_name}@${PROJECT_ID}.iam.gserviceaccount.com"
  local gws_restore_sa=""
  if [[ "$ENABLE_GWS" == "true" ]]; then
    gws_restore_sa="eon-gws-${EON_ACCOUNT_ID:0:16}-rs@${PROJECT_ID}.iam.gserviceaccount.com"
  fi
  local restore_target_members="serviceAccount:$restore_sa"
  restore_target_members+=$'\n'"serviceAccount:$node_sa"
  local restore_project_emails="$restore_sa"
  restore_project_emails+=$'\n'"$node_sa"
  local restore_sa_emails=("$restore_sa")
  local has_other_restore_accounts=false
  if project_has_other_restore_accounts "$PROJECT_ID" "$restore_project_emails"; then
    has_other_restore_accounts=true
  fi

  # Build the per-project SA email lists across scope. Destroy sweeps all
  # accessible projects under the requested org/folder for both the current
  # node-SA name and the legacy regular restore SA name, independent of current
  # include/exclude filters that may have changed since onboarding.
  local node_sa_emails=("$node_sa")
  if [[ "$roles_scope" == "organization" ]]; then
    local discovered_projects=""
    if [[ -n "$FOLDER_ID" ]]; then
      discovered_projects=$(get_projects_under_folder_for_destroy "$FOLDER_ID")
    else
      discovered_projects=$(get_projects_under_organization_for_destroy "$ORGANIZATION_ID")
    fi
    if [[ -n "$discovered_projects" ]]; then
      while IFS= read -r project; do
        [[ -z "$project" ]] && continue
        [[ "$project" == "$PROJECT_ID" ]] && continue
        local discovered_restore_sa="eon-restore-${restore_sa_id_value}@${project}.iam.gserviceaccount.com"
        local discovered_node_sa="${node_sa_name}@${project}.iam.gserviceaccount.com"
        restore_sa_emails+=("$discovered_restore_sa")
        node_sa_emails+=("$discovered_node_sa")
        restore_target_members+=$'\n'"serviceAccount:$discovered_restore_sa"
        restore_target_members+=$'\n'"serviceAccount:$discovered_node_sa"
      done <<< "$discovered_projects"
    fi
  fi

  local restore_role restore_firestore_role restore_vm_sa_manager_role export_writes_role legacy_export_writes_role restore_node_role restore_node_firestore_role secrets_role compute_role batch_role shared_subnet_role
  restore_role="$(restore_catalog_role_name "EonRestoreRole")"
  restore_firestore_role="$(restore_catalog_role_name "EonRestoreFirestoreRole")"
  restore_vm_sa_manager_role="$(restore_catalog_role_name "EonRestoreVmServiceAccountManagerRole")"
  export_writes_role="$(restore_catalog_role_name "EonBigQueryExportWritesRole")"
  legacy_export_writes_role="$(restore_catalog_role_name "EonExportWritesRole")"
  restore_node_role="$(restore_catalog_role_name "EonRestoreNodeRole")"
  restore_node_firestore_role="$(restore_catalog_role_name "EonRestoreNodeFirestoreRole")"
  secrets_role="$(restore_catalog_role_name "EonSecretsManagerRole")"
  compute_role="$(restore_catalog_role_name "EonComputeManagerRole")"
  batch_role="$(restore_catalog_role_name "EonBatchManagerRole")"
  shared_subnet_role="$(restore_catalog_role_name "EonRestoreToSharedSubnetworkRole")"

  # These lists map each role to the member it was bound to in
  # handle_restore_account; RESTORE_ROLE_CATALOG is the authoritative base-role
  # list used below for role deletion. Keep both in sync with the create handler.
  # Removal is intentionally unconditional (capability- and scope-agnostic):
  # roles never bound for this scope/config — e.g. EonOrgDiscoveryRole at project
  # scope — are benign "already absent" skips, just like the catalog deletion.
  # Roles bound to the restore SA in handle_restore_account.
  local restore_sa_roles=(
    EonOrgDiscoveryRole "$restore_role" "$restore_firestore_role" "$restore_vm_sa_manager_role" "$export_writes_role" "$legacy_export_writes_role" "$secrets_role"
    EonSecretManagerRole "$compute_role" "$batch_role" EonGceRestoreRole EonBigQueryRestoreRole EonCloudSqlRestoreRole
  )
  # Roles bound to node SAs (project scope) or the node-SA group (org/folder scope).
  local node_roles=("$restore_node_role" "$restore_node_firestore_role" "$secrets_role" EonSecretManagerRole EonBigQueryRestoreNodeRole EonGcsRestoreNodeRole EonGceRestoreNodeRole)

  if [[ "$roles_scope" == "organization" && -z "$NODE_SA_GROUP_EMAIL" ]]; then
    if restore_node_group_email_from_description "$restore_sa" "$PROJECT_ID"; then
      NODE_SA_GROUP_EMAIL="$output"
      log_info "Using node SA group email from restore SA description: $NODE_SA_GROUP_EMAIL"
    elif NODE_SA_GROUP_EMAIL=$(default_restore_node_group_email "$ORGANIZATION_ID"); then
      log_info "Using auto-derived default node SA group email: $NODE_SA_GROUP_EMAIL"
    fi
  fi

  if [[ -z "$HOST_PROJECT_ID" ]]; then
    local host_project_lookup_status=0
    if restore_host_project_id_from_description "$restore_sa" "$PROJECT_ID"; then
      HOST_PROJECT_ID="$output"
      log_info "Using host project ID from restore SA description: $HOST_PROJECT_ID"
    else
      host_project_lookup_status=$?
      if [[ "$host_project_lookup_status" -eq 1 ]]; then
        log_warn "No --host-project-id provided and restore SA description could not be read. If Restore onboarding used Shared VPC, re-run destroy with --host-project-id to remove host-project resources."
      fi
    fi
  fi

  log_info "Removing Restore IAM bindings (scope: $binding_roles_scope)..."
  local r node restore_email restore_project
  for restore_email in "${restore_sa_emails[@]}"; do
    for r in "${restore_sa_roles[@]}"; do
      remove_binding "serviceAccount:$restore_email" "$r" "$binding_roles_scope" "$binding_roles_scope_id"
      remove_deleted_sa_bindings "$restore_email" "$r" "$binding_roles_scope" "$binding_roles_scope_id"
    done
  done

  if [[ "$roles_scope" == "organization" ]]; then
    local org_restore_firestore_role="organizations/$roles_scope_id/roles/$restore_firestore_role"
    local org_node_firestore_role="organizations/$roles_scope_id/roles/$restore_node_firestore_role"
    if [[ -n "$NODE_SA_GROUP_EMAIL" ]]; then
      restore_target_members+=$'\n'"group:$NODE_SA_GROUP_EMAIL"
      for r in "${node_roles[@]}"; do
        remove_binding "group:$NODE_SA_GROUP_EMAIL" "$r" "$binding_roles_scope" "$binding_roles_scope_id"
      done
      # Management-project Firestore bindings created for org/folder scope.
      remove_project_binding_with_full_role "group:$NODE_SA_GROUP_EMAIL" "$org_node_firestore_role" "$PROJECT_ID"
    else
      log_warn "No --node-sa-group-email provided; skipping group binding removal. Pass it to clean up org/folder node-group bindings."
    fi
    for node in "${node_sa_emails[@]}"; do
      remove_project_binding_with_full_role "serviceAccount:$node" "$org_node_firestore_role" "$PROJECT_ID"
      remove_deleted_sa_bindings_for_full_role "$node" "$org_node_firestore_role" "project" "$PROJECT_ID"
    done
    for restore_email in "${restore_sa_emails[@]}"; do
      restore_project="${restore_email#*@}"
      restore_project="${restore_project%.iam.gserviceaccount.com}"
      remove_project_binding_with_full_role "serviceAccount:$restore_email" "$org_restore_firestore_role" "$restore_project"
    done
  fi

  if [[ "$roles_scope" == "organization" ]]; then
    # Older org/folder onboarding bound node SAs directly before the group-based
    # model existed. Keep sweeping those direct bindings so upgrades destroy
    # cleanly after the node SAs have been deleted.
    for node in "${node_sa_emails[@]}"; do
      for r in "${node_roles[@]}"; do
        remove_binding "serviceAccount:$node" "$r" "$binding_roles_scope" "$binding_roles_scope_id"
        remove_deleted_sa_bindings "$node" "$r" "$binding_roles_scope" "$binding_roles_scope_id"
      done
    done
  else
    for node in "${node_sa_emails[@]}"; do
      for r in "${node_roles[@]}"; do
        remove_binding "serviceAccount:$node" "$r" "$binding_roles_scope" "$binding_roles_scope_id"
        remove_deleted_sa_bindings "$node" "$r" "$binding_roles_scope" "$binding_roles_scope_id"
      done
    done
  fi

  # Shared VPC host-project role + binding (only when --host-project-id was used).
  if [[ -n "$HOST_PROJECT_ID" ]]; then
    remove_binding "serviceAccount:$restore_sa" "$shared_subnet_role" "project" "$HOST_PROJECT_ID"
  fi

  log_info "Deleting Restore custom roles (scope: $roles_scope)..."
  for r in "${RESTORE_ROLE_CATALOG[@]}"; do
    local role_to_delete
    role_to_delete="$(restore_catalog_role_name "$r")"
    if [[ "$binding_roles_scope" == "folder" ]]; then
      delete_role_if_no_other_bindings "$role_to_delete" "$roles_scope" "$roles_scope_id" "$restore_target_members" \
        "$binding_roles_scope" "$binding_roles_scope_id" \
        "project" "$PROJECT_ID"
    elif [[ "$binding_roles_scope" == "organization" ]]; then
      delete_role_if_no_other_bindings "$role_to_delete" "$roles_scope" "$roles_scope_id" "$restore_target_members" \
        "$binding_roles_scope" "$binding_roles_scope_id" \
        "project" "$PROJECT_ID"
    else
      delete_role_if_no_other_bindings "$role_to_delete" "$roles_scope" "$roles_scope_id" "$restore_target_members" \
        "$binding_roles_scope" "$binding_roles_scope_id"
    fi
  done
  if [[ -n "$HOST_PROJECT_ID" ]]; then
    delete_role_if_no_other_bindings "$shared_subnet_role" "project" "$HOST_PROJECT_ID" \
      "serviceAccount:$restore_sa" \
      "project" "$HOST_PROJECT_ID"
  fi

  # Tag value before key (a key with values cannot be deleted); deletion fails while
  # instances still carry the tag - re-run destroy once they are gone.
  local managed_tag_namespaced managed_tag_resource
  managed_tag_namespaced="$(gce_managed_tag_namespaced_key)"
  if _gcloud_try resource-manager tags values describe "$managed_tag_namespaced/true" --format="value(name)"; then
    managed_tag_resource="$output"
    destroy_exec "Eon-managed instances tag value $managed_tag_namespaced/true" -- resource-manager tags values delete "$managed_tag_resource"
  else
    log_info "Already absent (skip): Eon-managed instances tag value $managed_tag_namespaced/true"
  fi
  if _gcloud_try resource-manager tags keys describe "$managed_tag_namespaced" --format="value(name)"; then
    managed_tag_resource="$output"
    destroy_exec "Eon-managed instances tag key $managed_tag_namespaced" -- resource-manager tags keys delete "$managed_tag_resource"
  else
    log_info "Already absent (skip): Eon-managed instances tag key $managed_tag_namespaced"
  fi

  if [[ "$roles_scope" == "organization" && -n "$NODE_SA_GROUP_EMAIL" ]]; then
    log_info "Removing Restore node service accounts from group $NODE_SA_GROUP_EMAIL..."
    for node in "${node_sa_emails[@]}"; do
      remove_sa_from_group "$node" "$NODE_SA_GROUP_EMAIL"
    done
  fi

  log_info "Deleting Restore service accounts..."
  local failures_before_restore_sa_delete=$DESTROY_FAILURES
  for restore_email in "${restore_sa_emails[@]}"; do
    restore_project="${restore_email#*@}"
    restore_project="${restore_project%.iam.gserviceaccount.com}"
    delete_sa "$restore_email" "$restore_project"
  done
  if [[ -n "$gws_restore_sa" ]]; then
    delete_sa "$gws_restore_sa" "$PROJECT_ID"
  fi
  for node in "${node_sa_emails[@]}"; do
    local sa_project="${node#*@}"
    sa_project="${sa_project%.iam.gserviceaccount.com}"
    delete_sa "$node" "$sa_project"
  done

  # Firestore command-results database.
  if [[ "$has_other_restore_accounts" == "true" ]]; then
    log_info "Skipping shared Firestore database eon-cmd-results because other Eon Restore accounts remain in project $PROJECT_ID."
  elif [[ "$DESTROY_FAILURES" -gt "$failures_before_restore_sa_delete" ]]; then
    log_warn "Skipping shared Firestore database eon-cmd-results because Restore service account deletion failed."
  else
    destroy_exec "Firestore database eon-cmd-results" -- firestore databases delete --database="eon-cmd-results" --project="$PROJECT_ID"
  fi

  destroy_summary "RESTORE"
}

main() {
  parse_arguments "$@"
  validate_input
  check_gcloud_installed
  login_to_gcp
  select_gcp_project

  if [[ "$DESTROY" == "true" ]]; then
    # Org-scoped teardown deletes shared org-level custom roles only after
    # confirming no other Eon account still uses them, via Cloud Asset. Ensure
    # the (free) API is enabled first, otherwise every org role is skipped.
    if [[ -n "$ORGANIZATION_ID" ]]; then
      ensure_cloud_asset_api
    fi
    case "$ACCOUNT_TYPE" in
      "source")
        destroy_source_account
        ;;
      "restore")
        destroy_restore_account
        ;;
      *)
        log_error "Invalid account type: $ACCOUNT_TYPE. Must be 'Source' or 'Restore'."
        exit 1
        ;;
    esac
    return
  fi

  preflight_group_create_permission
  ensure_required_apis

  case "$ACCOUNT_TYPE" in
    "source")
      handle_source_account
      ;;
    "restore")
      handle_restore_account
      ;;
    *)
      log_error "Invalid account type: $ACCOUNT_TYPE. Must be 'Source' or 'Restore'."
      exit 1
      ;;
  esac
}

if [[ "${BASH_SOURCE[0]}" == "${0}" ]]; then
  main "$@"
fi