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

script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
module_dir="${TF_MODULE_DIR:-$(cd "$script_dir/.." && pwd)}"

endpoints="$(terraform -chdir="$module_dir" output -json public_endpoints)"
credentials="$(terraform -chdir="$module_dir" output -json service_credentials)"

public_ip="$(jq -r '.public_ip' <<<"$endpoints")"
mysql_port="$(jq -r '.mysql.port' <<<"$endpoints")"
kafka_public_port="$(jq -r '.kafka_anonymous.port' <<<"$endpoints")"
kafka_sasl_port="$(jq -r '.kafka_sasl_plain.port' <<<"$endpoints")"

mysql_user="$(jq -r '.mysql.user' <<<"$credentials")"
mysql_password="$(jq -r '.mysql.password' <<<"$credentials")"
kafka_user="$(jq -r '.kafka.username' <<<"$credentials")"
kafka_password="$(jq -r '.kafka.password' <<<"$credentials")"

mysql_image="mysql:8.4.11@sha256:b3b90af2a6552ae30c266fdb7d5dd55f3afb72404bb78d37fe8a23eb857fd3fb"
kafka_image="apache/kafka:3.9.2@sha256:05b4616e0702ef2729327705d54ad6b50ea70b271c4b730fabd2320789fb7b02"
pulsar_image="apachepulsar/pulsar:3.3.9@sha256:edb08c608371bc69a92a0edda255e869cef404577b9ca995cbe98e0c0de10e0e"

tmp_dir="$(mktemp -d)"
cleanup() {
  find "$tmp_dir" -mindepth 1 -delete
  rmdir "$tmp_dir"
}
trap cleanup EXIT

echo "Checking MySQL public endpoint..."
mysql_settings="$(docker run --rm \
  -e MYSQL_PWD="$mysql_password" \
  "$mysql_image" \
  mysql --connect-timeout=10 -h "$public_ip" -P "$mysql_port" -u "$mysql_user" \
  -Nse "SELECT @@gtid_mode, @@binlog_format, @@binlog_row_image, COUNT(*) FROM smoke.healthcheck")"
read -r gtid_mode binlog_format row_image smoke_rows <<<"$mysql_settings"
if [[ "$gtid_mode" != "ON" || "$binlog_format" != "ROW" || "$row_image" != "FULL" || "$smoke_rows" -lt 1 ]]; then
  echo "Unexpected MySQL replication settings or missing Smoke rows." >&2
  exit 1
fi
if docker run --rm -e MYSQL_PWD=deliberately-wrong "$mysql_image" \
  mysql --connect-timeout=10 -h "$public_ip" -P "$mysql_port" -u "$mysql_user" \
  -Nse "SELECT 1" >/dev/null 2>&1; then
  echo "MySQL accepted an incorrect password." >&2
  exit 1
fi

message="anonymous-$(date +%s)"
echo "Checking anonymous Kafka Proxy endpoint..."
printf '%s\n' "$message" | docker run --rm -i "$kafka_image" \
  /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server "$public_ip:$kafka_public_port" \
  --topic date-integration-smoke >/dev/null
docker run --rm "$kafka_image" \
  timeout 30 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server "$public_ip:$kafka_public_port" \
  --topic date-integration-smoke \
  --from-beginning --max-messages 1 >/dev/null

cat >"$tmp_dir/kafka-sasl.properties" <<EOF
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$kafka_user" password="$kafka_password";
EOF
chmod 0600 "$tmp_dir/kafka-sasl.properties"

echo "Checking SASL/PLAIN Kafka Proxy endpoint..."
printf 'sasl-%s\n' "$(date +%s)" | docker run --rm -i \
  -v "$tmp_dir/kafka-sasl.properties:/tmp/client.properties:ro" \
  "$kafka_image" \
  /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server "$public_ip:$kafka_sasl_port" \
  --producer.config /tmp/client.properties \
  --topic date-integration-smoke >/dev/null
docker run --rm -v "$tmp_dir/kafka-sasl.properties:/tmp/client.properties:ro" \
  "$kafka_image" timeout 30 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server "$public_ip:$kafka_sasl_port" \
  --consumer.config /tmp/client.properties \
  --topic date-integration-smoke --from-beginning --max-messages 1 >/dev/null

sed "s/password=\"$kafka_password\"/password=\"deliberately-wrong\"/" \
  "$tmp_dir/kafka-sasl.properties" >"$tmp_dir/kafka-sasl-wrong.properties"
chmod 0600 "$tmp_dir/kafka-sasl-wrong.properties"
if timeout 30 docker run --rm \
  -v "$tmp_dir/kafka-sasl-wrong.properties:/tmp/client.properties:ro" \
  "$kafka_image" /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server "$public_ip:$kafka_sasl_port" \
  --command-config /tmp/client.properties --list >/dev/null 2>&1; then
  echo "Kafka Proxy accepted incorrect SASL credentials." >&2
  exit 1
fi

if jq -e '.pulsar_binary != null' <<<"$endpoints" >/dev/null; then
  pulsar_binary_port="$(jq -r '.pulsar_binary.port' <<<"$endpoints")"
  pulsar_admin_port="$(jq -r '.pulsar_admin_http.port' <<<"$endpoints")"
  pulsar_admin_token="$(jq -r '.pulsar.admin_token' <<<"$credentials")"
  pulsar_client_token="$(jq -r '.pulsar.client_token' <<<"$credentials")"

  echo "Checking Pulsar Proxy binary endpoint..."
timeout 90 docker run --rm "$pulsar_image" \
  bin/pulsar-client \
  --url "pulsar://$public_ip:$pulsar_binary_port" \
  --auth-plugin org.apache.pulsar.client.impl.auth.AuthenticationToken \
  --auth-params "token:$pulsar_client_token" \
  produce persistent://public/default/date-integration-smoke \
  --messages "smoke-$(date +%s)" >/dev/null
timeout 90 docker run --rm "$pulsar_image" \
  bin/pulsar-client \
  --url "pulsar://$public_ip:$pulsar_binary_port" \
  --auth-plugin org.apache.pulsar.client.impl.auth.AuthenticationToken \
  --auth-params "token:$pulsar_client_token" \
  consume persistent://public/default/date-integration-smoke \
  --subscription-name "smoke-$(date +%s)" \
  --subscription-position Earliest \
  --num-messages 1 >/dev/null

if timeout 45 docker run --rm "$pulsar_image" \
  bin/pulsar-client \
  --url "pulsar://$public_ip:$pulsar_binary_port" \
  --auth-plugin org.apache.pulsar.client.impl.auth.AuthenticationToken \
  --auth-params token:deliberately-wrong \
  produce persistent://public/default/date-integration-smoke \
  --messages should-fail >/dev/null 2>&1; then
  echo "Pulsar Proxy accepted an incorrect client token." >&2
  exit 1
fi

echo "Checking Pulsar Proxy Admin HTTP endpoint..."
curl -fsS --max-time 15 \
  -H "Authorization: Bearer $pulsar_admin_token" \
  "http://$public_ip:$pulsar_admin_port/admin/v2/persistent/public/default/date-integration-smoke/partitions" >/dev/null

unauthenticated_status="$(curl -sS -o /dev/null -w '%{http_code}' --max-time 15 \
  "http://$public_ip:$pulsar_admin_port/admin/v2/persistent/public/default/date-integration-smoke/partitions")"
if [[ "$unauthenticated_status" != "401" && "$unauthenticated_status" != "403" ]]; then
  echo "Expected unauthenticated Pulsar request to fail, got HTTP $unauthenticated_status." >&2
  exit 1
fi

wrong_token_status="$(curl -sS -o /dev/null -w '%{http_code}' --max-time 15 \
  -H 'Authorization: Bearer deliberately-wrong' \
  "http://$public_ip:$pulsar_admin_port/admin/v2/persistent/public/default/date-integration-smoke/partitions")"
  if [[ "$wrong_token_status" != "401" && "$wrong_token_status" != "403" ]]; then
  echo "Expected invalid Pulsar token to fail, got HTTP $wrong_token_status." >&2
  exit 1
  fi
fi

if jq -e '.schema_registry != null' <<<"$endpoints" >/dev/null; then
  schema_endpoint="$(jq -r '.schema_registry.endpoint' <<<"$endpoints")"
  schema_user="$(jq -r '.schema_registry.username' <<<"$credentials")"
  schema_password="$(jq -r '.schema_registry.password' <<<"$credentials")"
  echo "Checking authenticated Schema Registry compatibility endpoint..."
  curl -fsS --max-time 15 -u "$schema_user:$schema_password" "$schema_endpoint/subjects" >/dev/null

  unauthenticated_status="$(curl -sS -o /dev/null -w '%{http_code}' --max-time 15 \
    "$schema_endpoint/subjects")"
  if [[ "$unauthenticated_status" != "401" ]]; then
    echo "Expected unauthenticated Schema Registry request to return HTTP 401, got $unauthenticated_status." >&2
    exit 1
  fi

  wrong_credentials_status="$(curl -sS -o /dev/null -w '%{http_code}' --max-time 15 \
    -u "$schema_user:deliberately-wrong" "$schema_endpoint/subjects")"
  if [[ "$wrong_credentials_status" != "401" ]]; then
    echo "Expected invalid Schema Registry credentials to return HTTP 401, got $wrong_credentials_status." >&2
    exit 1
  fi
fi

echo "All public service smoke tests passed."