apiVersion: batch/v1
kind: CronJob
metadata:
  name: rabbitmq-apps-bootstrap
  namespace: rabbitmq
spec:
  schedule: "*/10 * * * *"
  successfulJobsHistoryLimit: 1
  failedJobsHistoryLimit: 2
  concurrencyPolicy: Forbid
  jobTemplate:
    spec:
      backoffLimit: 0
      activeDeadlineSeconds: 600
      template:
        metadata:
          annotations:
            sidecar.istio.io/inject: "false"
        spec:
          restartPolicy: OnFailure
          serviceAccountName: rabbitmq
          automountServiceAccountToken: false
          volumes:
            - name: sa-token
              projected:
                sources:
                  - serviceAccountToken:
                      path: token
                      expirationSeconds: 3600
          containers:
            - name: bootstrap
              image: alpine:3.20
              volumeMounts:
                - name: sa-token
                  mountPath: /var/run/secrets/tokens
                  readOnly: true
              command: ["/bin/sh", "-ec"]
              args:
                - |
                  apk add --no-cache curl jq >/dev/null
                  VAULT_ADDR="http://vault-vault-contour.vault.svc:8200"
                  JWT="$(cat /var/run/secrets/tokens/token)"
                  VAULT_TOKEN="$(curl -sS --request POST \
                    --data "{\"role\":\"rabbitmq\",\"jwt\":\"${JWT}\"}" \
                    "${VAULT_ADDR}/v1/auth/kubernetes/login" | jq -r '.auth.client_token')"
                  [ -n "${VAULT_TOKEN}" ] && [ "${VAULT_TOKEN}" != "null" ]

                  admin_json="$(curl -sS -H "X-Vault-Token: ${VAULT_TOKEN}" "${VAULT_ADDR}/v1/secrets/data/rabbitmq/auth")"
                  admin_user="$(echo "${admin_json}" | jq -r '.data.data.username')"
                  admin_pass="$(echo "${admin_json}" | jq -r '.data.data.password')"
                  list_json="$(curl -sS -H "X-Vault-Token: ${VAULT_TOKEN}" "${VAULT_ADDR}/v1/secrets/metadata/rabbitmq/apps?list=true")"
                  for app in $(echo "${list_json}" | jq -r '.data.keys[]?' | sed 's#/$##'); do
                    app_json="$(curl -sS -H "X-Vault-Token: ${VAULT_TOKEN}" "${VAULT_ADDR}/v1/secrets/data/rabbitmq/apps/${app}")"
                    username="$(echo "${app_json}" | jq -r '.data.data.username')"
                    password="$(echo "${app_json}" | jq -r '.data.data.password')"
                    [ -z "${username}" ] && username="${app}"
                    [ -z "${password}" ] && continue

                    curl -sS -u "${admin_user}:${admin_pass}" -H "content-type:application/json" \
                      -X PUT "http://rabbitmq.rabbitmq.svc.cluster.local:15672/api/users/${username}" \
                      -d "{\"password\":\"${password}\",\"tags\":\"\"}" >/dev/null

                    vhosts_count="$(echo "${app_json}" | jq -r '(.data.data.vhosts // []) | length')"
                    if [ "${vhosts_count}" -gt 0 ]; then
                      echo "${app_json}" | jq -c '.data.data.vhosts[]' | while read -r vhost_item; do
                        vhost="$(echo "${vhost_item}" | jq -r '.name // "/"')"
                        configure="$(echo "${vhost_item}" | jq -r '.permissions.configure // ".*"')"
                        write="$(echo "${vhost_item}" | jq -r '.permissions.write // ".*"')"
                        read="$(echo "${vhost_item}" | jq -r '.permissions.read // ".*"')"
                        vhost_uri="$(jq -rn --arg v "${vhost}" '$v|@uri')"

                        curl -sS -u "${admin_user}:${admin_pass}" -H "content-type:application/json" \
                          -X PUT "http://rabbitmq.rabbitmq.svc.cluster.local:15672/api/vhosts/${vhost_uri}" \
                          -d '{}' >/dev/null
                        curl -sS -u "${admin_user}:${admin_pass}" -H "content-type:application/json" \
                          -X PUT "http://rabbitmq.rabbitmq.svc.cluster.local:15672/api/permissions/${vhost_uri}/${username}" \
                          -d "{\"configure\":\"${configure}\",\"write\":\"${write}\",\"read\":\"${read}\"}" >/dev/null
                      done
                    else
                      vhost="$(echo "${app_json}" | jq -r '.data.data.vhost // "/"')"
                      configure="$(echo "${app_json}" | jq -r '.data.data.permissions.configure // ".*"')"
                      write="$(echo "${app_json}" | jq -r '.data.data.permissions.write // ".*"')"
                      read="$(echo "${app_json}" | jq -r '.data.data.permissions.read // ".*"')"
                      vhost_uri="$(jq -rn --arg v "${vhost}" '$v|@uri')"

                      curl -sS -u "${admin_user}:${admin_pass}" -H "content-type:application/json" \
                        -X PUT "http://rabbitmq.rabbitmq.svc.cluster.local:15672/api/vhosts/${vhost_uri}" \
                        -d '{}' >/dev/null
                      curl -sS -u "${admin_user}:${admin_pass}" -H "content-type:application/json" \
                        -X PUT "http://rabbitmq.rabbitmq.svc.cluster.local:15672/api/permissions/${vhost_uri}/${username}" \
                        -d "{\"configure\":\"${configure}\",\"write\":\"${write}\",\"read\":\"${read}\"}" >/dev/null
                    fi
                  done
---
apiVersion: batch/v1
kind: CronJob
metadata:
  name: minio-apps-bootstrap
  namespace: minio
spec:
  schedule: "*/10 * * * *"
  successfulJobsHistoryLimit: 1
  failedJobsHistoryLimit: 2
  concurrencyPolicy: Forbid
  jobTemplate:
    spec:
      template:
        metadata:
          annotations:
            sidecar.istio.io/inject: "false"
        spec:
          restartPolicy: OnFailure
          serviceAccountName: minio-sa
          automountServiceAccountToken: false
          volumes:
            - name: sa-token
              projected:
                sources:
                  - serviceAccountToken:
                      path: token
                      expirationSeconds: 3600
          containers:
            - name: bootstrap
              image: alpine:3.20
              volumeMounts:
                - name: sa-token
                  mountPath: /var/run/secrets/tokens
                  readOnly: true
              command: ["/bin/sh", "-ec"]
              args:
                - |
                  apk add --no-cache curl jq >/dev/null
                  curl -fsSL -o /usr/local/bin/mc https://dl.min.io/client/mc/release/linux-amd64/mc
                  chmod +x /usr/local/bin/mc

                  VAULT_ADDR="http://vault-vault-contour.vault.svc:8200"
                  JWT="$(cat /var/run/secrets/tokens/token)"
                  VAULT_TOKEN="$(curl -sS --request POST \
                    --data "{\"role\":\"minio\",\"jwt\":\"${JWT}\"}" \
                    "${VAULT_ADDR}/v1/auth/kubernetes/login" | jq -r '.auth.client_token')"
                  [ -n "${VAULT_TOKEN}" ] && [ "${VAULT_TOKEN}" != "null" ]

                  admin_json="$(curl -sS -H "X-Vault-Token: ${VAULT_TOKEN}" "${VAULT_ADDR}/v1/secrets/data/minio/admin")"
                  root_user="$(echo "${admin_json}" | jq -r '.data.data.rootUser')"
                  root_pass="$(echo "${admin_json}" | jq -r '.data.data.rootPassword')"
                  /usr/local/bin/mc alias set local http://minio.minio.svc.cluster.local:9000 "${root_user}" "${root_pass}"

                  list_json="$(curl -sS -H "X-Vault-Token: ${VAULT_TOKEN}" "${VAULT_ADDR}/v1/secrets/metadata/minio/apps?list=true")"
                  for app in $(echo "${list_json}" | jq -r '.data.keys[]?' | sed 's#/$##'); do
                    app_json="$(curl -sS -H "X-Vault-Token: ${VAULT_TOKEN}" "${VAULT_ADDR}/v1/secrets/data/minio/apps/${app}")"
                    username="$(echo "${app_json}" | jq -r '.data.data.username')"
                    password="$(echo "${app_json}" | jq -r '.data.data.password')"
                    access_key="$(echo "${app_json}" | jq -r '.data.data.access_key // .data.data.username // empty')"
                    secret_key="$(echo "${app_json}" | jq -r '.data.data.secret_key // .data.data.password // empty')"
                    policy="$(echo "${app_json}" | jq -r '.data.data.policy // empty')"
                    [ -z "${username}" ] && username="${app}"
                    [ -z "${access_key}" ] && access_key="${username}"
                    [ -z "${secret_key}" ] && secret_key="${password}"
                    [ -z "${secret_key}" ] && continue

                    /usr/local/bin/mc admin user add local "${access_key}" "${secret_key}" >/dev/null 2>&1 || true
                    /usr/local/bin/mc admin user enable local "${access_key}" >/dev/null 2>&1 || true
                    if [ -n "${policy}" ]; then
                      /usr/local/bin/mc admin policy attach local "${policy}" --user "${access_key}" >/dev/null 2>&1 || true
                    fi

                    echo "${app_json}" | jq -r '.data.data.buckets[]? | if type=="string" then . else .name // empty end' | while read -r bucket; do
                      [ -z "${bucket}" ] && continue
                      /usr/local/bin/mc mb --ignore-existing "local/${bucket}" >/dev/null 2>&1 || true
                    done
                  done
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: kafka-bootstrap-exec
  namespace: kafka
rules:
  - apiGroups: [""]
    resources: ["pods"]
    verbs: ["get", "list"]
  - apiGroups: [""]
    resources: ["pods/exec"]
    verbs: ["create", "get"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: kafka-bootstrap-exec
  namespace: kafka
subjects:
  - kind: ServiceAccount
    name: kafka-kafka-contour
    namespace: kafka
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: Role
  name: kafka-bootstrap-exec
---
apiVersion: batch/v1
kind: CronJob
metadata:
  name: kafka-apps-bootstrap
  namespace: kafka
spec:
  schedule: "*/10 * * * *"
  successfulJobsHistoryLimit: 1
  failedJobsHistoryLimit: 2
  concurrencyPolicy: Forbid
  jobTemplate:
    spec:
      template:
        metadata:
          annotations:
            sidecar.istio.io/inject: "false"
        spec:
          restartPolicy: OnFailure
          serviceAccountName: kafka-kafka-contour
          automountServiceAccountToken: true
          volumes:
            - name: sa-token
              projected:
                sources:
                  - serviceAccountToken:
                      path: token
                      expirationSeconds: 3600
          containers:
            - name: bootstrap
              image: alpine:3.20
              volumeMounts:
                - name: sa-token
                  mountPath: /var/run/secrets/tokens
                  readOnly: true
              command: ["/bin/sh", "-ec"]
              args:
                - |
                  set -e
                  apk add --no-cache bash curl jq kubectl >/dev/null

                  VAULT_ADDR="http://vault-vault-contour.vault.svc:8200"
                  JWT="$(cat /var/run/secrets/tokens/token)"
                  VAULT_TOKEN="$(curl -sS --request POST \
                    --data "{\"role\":\"kafka\",\"jwt\":\"${JWT}\"}" \
                    "${VAULT_ADDR}/v1/auth/kubernetes/login" | jq -r '.auth.client_token')"
                  [ -n "${VAULT_TOKEN}" ] && [ "${VAULT_TOKEN}" != "null" ]

                  bootstrap_json="$(curl -sS -H "X-Vault-Token: ${VAULT_TOKEN}" "${VAULT_ADDR}/v1/secrets/data/kafka/bootstrap")"
                  inter_broker_password="$(echo "${bootstrap_json}" | jq -r '.data.data.interBrokerPassword')"
                  [ -n "${inter_broker_password}" ] && [ "${inter_broker_password}" != "null" ]
                  list_json="$(curl -sS -H "X-Vault-Token: ${VAULT_TOKEN}" "${VAULT_ADDR}/v1/secrets/metadata/kafka/apps?list=true")"
                  target_pod="kafka-kafka-contour-controller-0"
                  if ! kubectl -n kafka get pod "${target_pod}" >/dev/null 2>&1; then
                    echo "Kafka controller pod not found"
                    exit 1
                  fi
                  target_bootstrap="${target_pod}.kafka-kafka-contour-controller-headless.kafka.svc.cluster.local:9094"

                  admin_props="$(mktemp)"
                  printf "%s\n" \
                    "security.protocol=SASL_PLAINTEXT" \
                    "sasl.mechanism=PLAIN" \
                    "sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='inter_broker_user' password='${inter_broker_password}';" \
                    "default.api.timeout.ms=60000" \
                    "request.timeout.ms=60000" \
                    > "${admin_props}"
                  kubectl -n kafka exec -i "${target_pod}" -c kafka -- /bin/bash -lc 'cat > /tmp/admin.properties' < "${admin_props}"
                  rm -f "${admin_props}"
                  kubectl -n kafka exec "${target_pod}" -c kafka -- /bin/bash -lc 'test -s /tmp/admin.properties'

                  for app in $(echo "${list_json}" | jq -r '.data.keys[]?' | sed 's#/$##'); do
                    app_json="$(curl -sS -H "X-Vault-Token: ${VAULT_TOKEN}" "${VAULT_ADDR}/v1/secrets/data/kafka/apps/${app}")"
                    username="$(echo "${app_json}" | jq -r '.data.data.username')"
                    password="$(echo "${app_json}" | jq -r '.data.data.password')"
                    [ -z "${username}" ] && username="${app}"
                    [ -z "${password}" ] && continue

                    echo "Reconciling Kafka user ${username}"
                    user_reconciled=false
                    attempt=1
                    while [ "${attempt}" -le 3 ]; do
                      if kubectl -n kafka exec "${target_pod}" -c kafka -- /bin/bash -lc "\
                        timeout 60 /opt/bitnami/kafka/bin/kafka-configs.sh --bootstrap-server ${target_bootstrap} --command-config /tmp/admin.properties \
                          --alter --add-config 'SCRAM-SHA-512=[password=${password}]' \
                          --entity-type users --entity-name '${username}'
                      " >/dev/null; then
                        user_reconciled=true
                        break
                      fi
                      echo "Kafka user ${username} reconcile attempt ${attempt}/3 failed"
                      attempt=$((attempt + 1))
                      sleep 5
                    done
                    if [ "${user_reconciled}" != "true" ]; then
                      echo "Kafka user ${username} reconcile failed, continue"
                      continue
                    fi

                    echo "${app_json}" | jq -c '.data.data.topics[]?' | while read -r topic_item; do
                      topic_name="$(echo "${topic_item}" | jq -r '.name // empty')"
                      partitions="$(echo "${topic_item}" | jq -r '.partitions // 3')"
                      replication_factor="$(echo "${topic_item}" | jq -r '.replication_factor // 1')"
                      topic_configs="$(echo "${topic_item}" | jq -r '(.configs // {}) | to_entries | map("\(.key)=\(.value|tostring)") | join(",")')"
                      [ -z "${topic_name}" ] && continue

                      echo "Reconciling Kafka topic ${topic_name}"
                      if ! kubectl -n kafka exec "${target_pod}" -c kafka -- /bin/bash -lc "\
                        timeout 40 /opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server ${target_bootstrap} --command-config /tmp/admin.properties \
                          --create --if-not-exists --topic '${topic_name}' --partitions '${partitions}' --replication-factor '${replication_factor}'
                      " >/dev/null; then
                        echo "Kafka topic ${topic_name} create/reconcile failed, continue"
                        continue
                      fi

                      if [ -n "${topic_configs}" ]; then
                        if ! kubectl -n kafka exec "${target_pod}" -c kafka -- /bin/bash -lc "\
                          timeout 40 /opt/bitnami/kafka/bin/kafka-configs.sh --bootstrap-server ${target_bootstrap} --command-config /tmp/admin.properties \
                            --alter --entity-type topics --entity-name '${topic_name}' --add-config '${topic_configs}'
                        " >/dev/null; then
                          echo "Kafka topic ${topic_name} config reconcile failed, continue"
                        fi
                      fi
                    done
                  done
