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

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

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

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

# Mode B: bind temp queue to an exchange + routing key and consume from it
EXCHANGE="${EXCHANGE:-}"        # e.g. "amq.topic" or "my-exchange". Leave empty to skip binding mode.
ROUTING_KEY="${ROUTING_KEY:-#}" # pattern for topic/direct, default is catch-all "#"

# Polling interval when no messages
SLEEP_SECS="${SLEEP_SECS:-1}"

# Pretty-print with jq if available
USE_JQ="${USE_JQ:-auto}"        # auto|yes|no
USE_COLOR="${USE_COLOR:-yes}"   # yes|no -> 'yes' forces color with jq -C

# 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; }

pretty() {
  if [[ "$USE_JQ" == "yes" ]] || { [[ "$USE_JQ" == "auto" ]] && has_jq; }; then
    if [[ "$USE_COLOR" == "yes" ]]; then jq -C .; else jq .; fi
  else
    cat
  fi
}

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; }

# Map task_name -> absolute playbook path
map_playbook() {
  case "$1" in
    uptime)      printf '%s/uptime.yml' "${APP_ROOT}" ;;
    reset_aths)  printf '%s/resetradios.yml' "${APP_ROOT}" ;;
    *)           return 1 ;;
  esac
}

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

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

  if ! playbook="$(map_playbook "$task")"; then
    warn "unknown task_name='$task' for device='$device' (no-op)."
    return 0
  fi

  # Optional delay: "none" -> no delay; integer -> sleep seconds
  if [[ -n "$delay" && "$delay" != "none" ]]; then
    if [[ "$delay" =~ ^[0-9]+$ ]]; then
      log "delaying ${delay}s before running '${task}' on '${device}'"
      sleep "$delay"
    else
      warn "task_delay value '$delay' not numeric/'none' (ignoring)."
    fi
  fi

  log "→ Running playbook $(basename "$playbook") on host '${device}'"
  if ! "${NBPLAY}" "$playbook" "$device"; then
    err "playbook failed (task='${task}', device='${device}')"
    # do not exit; keep consuming
  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-by-one (the API returns an array)
  if has_jq && { [[ "$USE_JQ" == "yes" ]] || [[ "$USE_JQ" == "auto" ]]; }; then
    echo "$RESP" | jq -c '.[]' | while read -r item; do
      payload=$(printf '%s' "$item" | jq -r '.payload')
      rk=$(printf '%s' "$item" | jq -r '.routing_key')
      ex=$(printf '%s' "$item" | jq -r '.exchange')

      echo "-----"
      echo "exchange: ${ex:-\"\"}"
      echo "routing_key: $rk"
      echo "payload:"

      # Decode + pretty print + DISPATCH
      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
            printf '%s' "$decoded" | pretty
            dispatch_task "$decoded"
          else
            printf '%s\n' "$decoded"
            # not JSON -> no dispatch
          fi
        else
          printf '%s' "$payload" | pretty
          dispatch_task "$payload"
        fi
      else
        printf '%s\n' "$payload"
        # not JSON -> no dispatch
      fi
    done
  else
    # Minimal fallback: print raw response only
    echo "$RESP"
  fi
done