#!/usr/bin/env bash

set -Eeuo pipefail
export LANG=C
umask 077

EXPECTED_HOSTNAME="sqjcmqtt01"
EXPECTED_IPV4="10.2.203.108"
TARGET_PORT_RE=':(1883|4369|4370|5370|8083|8883|8084|18083)([[:space:]]|$)'
TEMP_ROOT=
EMQX_PID=
SUB_PID=
COOKIE=
EMQX_ENV=()

die() {
  printf 'ERROR: %s\n' "$*" >&2
  exit 1
}

cleanup() {
  set +e
  [[ -z "${SUB_PID:-}" ]] || kill "${SUB_PID}" >/dev/null 2>&1
  if ((${#EMQX_ENV[@]} > 0)); then
    runuser -u emqx -- env "${EMQX_ENV[@]}" emqx stop >/dev/null 2>&1
  fi
  if [[ -n "${EMQX_PID:-}" ]]; then
    kill "${EMQX_PID}" >/dev/null 2>&1
    for _ in {1..20}; do
      kill -0 "${EMQX_PID}" >/dev/null 2>&1 || break
      sleep 0.25
    done
    kill -KILL "${EMQX_PID}" >/dev/null 2>&1
    wait "${EMQX_PID}" >/dev/null 2>&1
  fi
  pkill -u emqx -x epmd >/dev/null 2>&1
  systemctl disable --now emqx.service mosquitto.service >/dev/null 2>&1
  [[ -z "${TEMP_ROOT:-}" ]] || rm -rf -- "${TEMP_ROOT}"
  COOKIE=
}

unit_is_quiet() {
  local unit=$1 enabled active
  enabled=$(systemctl is-enabled "${unit}" 2>/dev/null || true)
  active=$(systemctl is-active "${unit}" 2>/dev/null || true)
  case "${enabled}" in disabled|masked) ;; *) return 1 ;; esac
  case "${active}" in inactive|failed) ;; *) return 1 ;; esac
}

[[ "${EUID}" -eq 0 ]] || die "must run as root"
[[ "$(hostname -s)" == "${EXPECTED_HOSTNAME}" ]] || die "hostname mismatch"
ip -o -4 addr show scope global | awk '{print $4}' | cut -d/ -f1 |
  grep -Fxq "${EXPECTED_IPV4}" || die "expected IPv4 is not configured"

for command in emqx mosquitto_pub mosquitto_sub openssl runuser ss timeout; do
  command -v "${command}" >/dev/null || die "required command not found: ${command}"
done
getent passwd emqx >/dev/null || die "emqx account not found"
[[ "$(rpm -q --qf '%{NAME}-%{VERSION}-%{RELEASE}.%{ARCH}' emqx)" == \
  "emqx-5.8.9-1.el8.x86_64" ]] || die "EMQX NEVRA mismatch"
rpm -q mosquitto libatomic cjson libwebsockets >/dev/null ||
  die "an installed dependency is missing"
unit_is_quiet emqx.service || die "emqx.service is not disabled/inactive"
unit_is_quiet mosquitto.service || die "mosquitto.service is not disabled/inactive"
[[ -z "$(ss -lntupH | grep -E "${TARGET_PORT_RE}" || true)" ]] ||
  die "an EMQX target port is occupied before the test"
[[ -z "$(pgrep -u emqx -x epmd || true)" ]] ||
  die "an emqx-owned epmd process already exists"
[[ -z "$(pgrep -fa 'beam.smp|emqx-q081@127.0.0.1' || true)" ]] ||
  die "an EMQX test process already exists"
[[ -z "$(find /var/tmp -maxdepth 1 -type d -name 'emqx-q081-pubsub.*' -print -quit)" ]] ||
  die "a stale EMQX pubsub temporary root exists"

TEMP_ROOT=$(mktemp -d /var/tmp/emqx-q081-pubsub.XXXXXX)
install -d -m 0700 -o emqx -g emqx "${TEMP_ROOT}/data"
chown -R emqx:emqx "${TEMP_ROOT}"
COOKIE=$(openssl rand -hex 32)
TOPIC="q081/$(date +%s)/$(openssl rand -hex 6)"
PAYLOAD="Q081_$(openssl rand -hex 16)"

EMQX_ENV=(
  'EMQX_NODE__NAME=emqx-q081@127.0.0.1'
  'EMQX_CLUSTER__DISCOVERY_STRATEGY=manual'
  'EMQX_LISTENERS__TCP__DEFAULT__BIND="127.0.0.1:1883"'
  'EMQX_LISTENERS__SSL__DEFAULT__ENABLE=false'
  'EMQX_LISTENERS__WS__DEFAULT__ENABLE=false'
  'EMQX_LISTENERS__WSS__DEFAULT__ENABLE=false'
  'EMQX_DASHBOARD__LISTENERS__HTTP__ENABLE=false'
  'ERL_EPMD_ADDRESS=127.0.0.1'
  'ERL_AFLAGS=-kernel inet_dist_use_interface {127,0,0,1}'
  "EMQX_NODE__COOKIE=${COOKIE}"
  "EMQX_NODE__DATA_DIR=${TEMP_ROOT}/data"
)

trap cleanup EXIT
runuser -u emqx -- env "${EMQX_ENV[@]}" emqx check_config
printf 'TEMP_CHECK_CONFIG_RC=0\n'

runuser -u emqx -- env "${EMQX_ENV[@]}" emqx foreground \
  >"${TEMP_ROOT}/foreground.log" 2>&1 &
EMQX_PID=$!

for _ in {1..60}; do
  ss -lntH | grep -Eq '127\.0\.0\.1:1883([[:space:]]|$)' && break
  kill -0 "${EMQX_PID}" >/dev/null 2>&1 ||
    die "temporary EMQX exited before listening"
  sleep 0.5
done
ss -lntH | grep -Eq '127\.0\.0\.1:1883([[:space:]]|$)' ||
  die "temporary EMQX did not bind loopback 1883"

LISTENERS_DURING=$(ss -lntupH | grep -E "${TARGET_PORT_RE}" || true)
printf '=== LISTENERS DURING TEST ===\n%s\n' "${LISTENERS_DURING}"
if printf '%s\n' "${LISTENERS_DURING}" |
  grep -Ev '127\.0\.0\.1:(1883|4369|4370|5370)([[:space:]]|$)' |
  grep -q .; then
  die "a temporary EMQX listener is not bound to loopback"
fi

timeout 15 mosquitto_sub -h 127.0.0.1 -p 1883 \
  -t "${TOPIC}" -C 1 >"${TEMP_ROOT}/received.txt" &
SUB_PID=$!
sleep 1
mosquitto_pub -h 127.0.0.1 -p 1883 -t "${TOPIC}" -m "${PAYLOAD}"
wait "${SUB_PID}" || die "mosquitto_sub did not receive a message"
SUB_PID=

RECEIVED=$(cat "${TEMP_ROOT}/received.txt")
printf 'PUBSUB_RECEIVED=%s\n' "${RECEIVED}"
[[ "${RECEIVED}" == "${PAYLOAD}" ]] || die "pubsub payload mismatch"
printf 'PUBSUB_MATCH=PASS\n'
runuser -u emqx -- env "${EMQX_ENV[@]}" emqx ctl status

cleanup
trap - EXIT
sleep 1

printf '=== FINAL STATE ===\n'
[[ ! -e "${TEMP_ROOT}" ]] || die "temporary root remains"
printf 'TEMP_ROOT_ABSENT\n'
[[ -z "$(pgrep -fa 'beam.smp|emqx-q081@127.0.0.1' || true)" ]] ||
  die "temporary EMQX process remains"
printf 'TARGET_PROCESSES_ZERO\n'
unit_is_quiet emqx.service || die "emqx.service is not disabled/inactive after cleanup"
unit_is_quiet mosquitto.service || die "mosquitto.service is not disabled/inactive after cleanup"
printf 'TARGET_UNITS_DISABLED_INACTIVE\n'
[[ -z "$(ss -lntupH | grep -E "${TARGET_PORT_RE}" || true)" ]] ||
  die "an EMQX target listener remains after cleanup"
printf 'TARGET_LISTENERS_ZERO\n'
printf 'EMQX01_LOOPBACK_PUBSUB_PASS\n'