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

NAMESPACE="kafka"
KAFKA_NAME="kafka"
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"

echo "==> Checking prerequisites"
command -v kubectl >/dev/null || { echo "kubectl not found"; exit 1; }
command -v helm    >/dev/null || { echo "helm not found";    exit 1; }

echo "==> Verifying cluster reachable"
kubectl cluster-info --request-timeout=5s >/dev/null

echo "==> Step 1/5 : Create namespace"
kubectl create namespace "${NAMESPACE}" --dry-run=client -o yaml | kubectl apply -f -

echo "==> Step 2/5 : Install Strimzi operator"
helm repo add strimzi https://strimzi.io/charts/ 2>/dev/null || true
helm repo update strimzi

helm upgrade --install strimzi-kafka-operator strimzi/strimzi-kafka-operator \
  --namespace "${NAMESPACE}" \
  --set watchAnyNamespace=false \
  --wait --timeout 5m

echo "==> Waiting for Strimzi Cluster Operator to be ready"
kubectl rollout status deployment/strimzi-cluster-operator -n "${NAMESPACE}" --timeout=120s

echo "==> Waiting for Strimzi CRDs to be fully established"
for crd in kafkas.kafka.strimzi.io kafkanodepools.kafka.strimzi.io kafkausers.kafka.strimzi.io; do
  until kubectl get crd "${crd}" -o jsonpath='{.status.conditions[?(@.type=="Established")].status}' 2>/dev/null | grep -q "True"; do
    sleep 2
  done
  echo "    - ${crd} established"
done

echo "==> Step 3/5 : Apply Kafka cluster (KRaft, TLS, SCRAM-SHA-512)"
kubectl apply -f "${SCRIPT_DIR}/kafka.yaml"

echo "==> Waiting for Kafka cluster to be Ready (3-5 min)"
kubectl wait kafka/"${KAFKA_NAME}" \
  --for=condition=Ready \
  --timeout=10m \
  -n "${NAMESPACE}"

echo "==> Step 4/5 : Apply KafkaUsers"
kubectl apply -f "${SCRIPT_DIR}/kafka-users.yaml"

echo "==> Waiting for KafkaUsers to be Ready"
for user in kafka-admin kafka-client; do
  echo "    - waiting for ${user}"
  kubectl wait kafkauser/"${user}" \
    --for=condition=Ready \
    --timeout=120s \
    -n "${NAMESPACE}"
done

echo ""
echo "==> Step 5/5 : Deployment complete"
echo ""
kubectl get pods -n "${NAMESPACE}"
echo ""
echo "Bootstrap (TLS + SCRAM-SHA-512, cluster-internal):"
echo "  kafka-kafka-bootstrap.${NAMESPACE}.svc.cluster.local:9093"
echo ""
echo "Get CA cert (import on client side):"
echo "  kubectl -n ${NAMESPACE} get secret kafka-cluster-ca-cert \\"
echo "    -o jsonpath='{.data.ca\\.crt}' | base64 -d > kafka-ca.crt"
echo ""
echo "Get SCRAM credentials:"
echo "  # Admin"
echo "  kubectl -n ${NAMESPACE} get secret kafka-admin -o jsonpath='{.data.password}' | base64 -d"
echo "  # Client"
echo "  kubectl -n ${NAMESPACE} get secret kafka-client -o jsonpath='{.data.password}' | base64 -d"
echo ""
echo "Sample client config (properties):"
echo "  bootstrap.servers=kafka-kafka-bootstrap.${NAMESPACE}.svc.cluster.local:9093"
echo "  security.protocol=SASL_SSL"
echo "  ssl.truststore.type=PEM"
echo "  ssl.truststore.certificates=<contents of kafka-ca.crt>"
echo "  sasl.mechanism=SCRAM-SHA-512"
echo "  sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \\"
echo "    username=\"kafka-client\" password=\"<password>\";"
echo ""