ansible-worker/files/rabbit-client.sh-beforedelayed

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

# --- Config (override via env) ---
RMQ_USER="${RMQ_USER:-admin}"
RMQ_PASS="${RMQ_PASS:-change_me}"
RMQ_HOST="${RMQ_HOST:-localhost}"
RMQ_PORT="${RMQ_PORT:-15672}"
VHOST="${VHOST:-app}"

RMQ_HOST="10.210.12.2"

# Mode A: read from a queue (default)
QUEUE="${QUEUE:-queue_deviceconfig}"


# Mode B: bind temp queue to an exchange + routing key and consume from it
EXCHANGE="${EXCHANGE:-}"          # empty => Mode B OFF
ROUTING_KEY="${ROUTING_KEY:-}"    # irrelevant when EXCHANGE is empty
# Polling interval when no messages
SLEEP_SECS="${SLEEP_SECS:-1}"

# Paths
APP_ROOT="/opt/containers/ansible-worker/app"
NBPLAY="${APP_ROOT}/bin/nbplay"

# --- Helpers ---
api() {
  local method="$1"; shift
  local path="$1"; shift
  local data="${1:-}"
  if [[ -n "$data" ]]; then
    curl -sS -u "$RMQ_USER:$RMQ_PASS" -H "content-type:application/json" -X "$method" "http://$RMQ_HOST:$RMQ_PORT$path" -d "$data"
  else
    curl -sS -u "$RMQ_USER:$RMQ_PASS" -H "content-type:application/json" -X "$method" "http://$RMQ_HOST:$RMQ_PORT$path"
  fi
}

has_jq() { command -v jq >/dev/null 2>&1; }

urlenc() { printf '%s' "$1" | sed -e 's, ,%20,g' -e 's,/,%2F,g'; }

log() { printf '[consumer] %s\n' "$*"; }
warn() { printf '[consumer][WARN] %s\n' "$*" >&2; }
err() { printf '[consumer][ERROR] %s\n' "$*" >&2; }

# Safely append string options into an array using eval (so quotes are honored).
append_opts() {
  local opts_str="$1"
  # shellcheck disable=SC2206
  local -n _arr_ref=$2
  if [[ -n "$opts_str" ]]; then
    # We build an eval that appends parsed words to the array.
    eval '_arr_ref+=('"$opts_str"')'
  fi
}

# Given a JSON object payload, extract fields and dispatch nbplay
dispatch_task() {
  local json="$1"
  local device task playbook task_options
  device="$(jq -er '.inscope_device // empty' <<<"$json")" || device=""
  task="$(jq -er '.task_name // empty' <<<"$json")" || task=""
  task_options="$(jq -r '.task_options // empty' <<<"$json")" || task_options=""

  if [[ -z "$device" || -z "$task" ]]; then
    warn "payload missing required keys (inscope_device/task_name). Skipping."
    return 0
  fi

  # --- Special case: task "update-rebootin_N" (N=0..99)
  # If matched, run playbook "update-rebootin.yml" and append "-erebootin=N" at the end of nbplay args.
  local base_task="$task"
  local -a extra_nbplay_opts=()
  local n=""
  if [[ "$task" =~ ^update-rebootin_([0-9]{1,2})$ ]]; then
    n="${BASH_REMATCH[1]}"
    base_task="update-rebootin"
    extra_nbplay_opts+=("-erebootin=${n}")
  fi

  playbook="${APP_ROOT}/${base_task}.yml"
  if [[ ! -f "$playbook" ]]; then
    warn "playbook not found: $playbook (device='$device', task='$task'). Skipping."
    return 0
  fi

  log "→ Running $(basename "$playbook") on '${device}' with options: ${task_options:-<none>}"

  # Build command array and append options (respect quoted args)
  local -a cmd=( "$NBPLAY" "$playbook" "$device" )
  append_opts "$task_options" cmd

  # Append any special-case options at the very end (e.g., -erebootin=N)
  if ((${#extra_nbplay_opts[@]})); then
    cmd+=("${extra_nbplay_opts[@]}")
  fi

  if ! "${cmd[@]}"; then
    err "playbook failed (task='${task}', device='${device}')"
    # keep consuming; message already acked by HTTP get endpoint
  fi
}

cleanup_queue=""
cleanup() {
  if [[ -n "$cleanup_queue" ]]; then
    log "Cleaning up temp queue: $cleanup_queue"
    api DELETE "/api/queues/$(urlenc "$VHOST")/$cleanup_queue" >/dev/null || true
  fi
}
trap cleanup EXIT

# --- Setup: decide queue source ---
if [[ -n "$EXCHANGE" ]]; then
  cleanup_queue="tmp.$(hostname -s).$$.$(date +%s)"
  log "Declaring temp queue: $cleanup_queue (exclusive, auto-delete)"
  api PUT "/api/queues/$(urlenc "$VHOST")/$cleanup_queue" '{
    "auto_delete": true, "durable": false, "arguments": {}, "exclusive": true
  }' >/dev/null

  log "Binding temp queue to exchange '$EXCHANGE' with routing key '$ROUTING_KEY'"
  api POST "/api/bindings/$(urlenc "$VHOST")/e/$EXCHANGE/q/$cleanup_queue" "{
    \"routing_key\": \"$ROUTING_KEY\", \"arguments\": {}
  }" >/dev/null

  QUEUE="$cleanup_queue"
  log "Consuming from bound temp queue: $QUEUE"
else
  log "Consuming directly from queue: $QUEUE (vhost: $VHOST)"
fi

log "Press Ctrl+C to stop."

# --- Consume loop ---
while :; do
  RESP="$(api POST "/api/queues/$(urlenc "$VHOST")/$QUEUE/get" '{
    "count": 1, "ackmode": "ack_requeue_false", "encoding": "auto", "truncate": 1000000
  }')"

  # Empty array => no messages
  if [[ "$RESP" == "[]" || -z "$RESP" ]]; then
    sleep "$SLEEP_SECS"
    continue
  fi

  # Process one message (array of length 1)
  if has_jq; then
    echo "$RESP" | jq -c '.[]' | while read -r item; do
      payload=$(printf '%s' "$item" | jq -r '.payload')

      # Decode if payload is a quoted JSON string
      if jq -e . >/dev/null 2>&1 <<<"$payload"; then
        ptype="$(printf '%s' "$payload" | jq -r 'type')"
        if [[ "$ptype" == "string" ]]; then
          decoded="$(printf '%s' "$payload" | jq -r .)"
          if jq -e . >/dev/null 2>&1 <<<"$decoded"; then
            dispatch_task "$decoded"
          else
            warn "payload is a string but not JSON after unquote; skipping."
          fi
        else
          dispatch_task "$payload"
        fi
      else
        warn "payload is not JSON; skipping."
      fi
    done
  else
    # Fallback: cannot parse JSON without jq
    warn "jq not found; cannot parse JSON payloads. Exiting."
    exit 3
  fi
done