#!/usr/bin/env bash

printf '%s\n' "Setting programmatically derived runtime variables"
if [ "${Environment}" = 'qa' ] || [ "${Environment}" = 'prod' ]; then
  # Assume this is a Fargate task. Find the private IP address without installing additional packages.
  DNS_NAME=$(curl -s ${ECS_CONTAINER_METADATA_URI_V4}/task | jq -r '.Containers[0].Networks[0].PrivateDNSName')
  printf '%s\n' "Task DNS name is ${DNS_NAME}"
  export CONNECT_REST_ADVERTISED_HOST_NAME="${DNS_NAME}"
else
  # Provide dev and local settings
  export CONNECT_REST_ADVERTISED_HOST_NAME='localhost'
  # Source the dotenv file volume in the container. Alter if you change the path where .env is mounted.
  . /etc/kafka-connect/connector_config.properties
fi

# This must be unique per worker cluster
export CONNECT_GROUP_ID="kafka_connect_${CONNECTOR_NAME}_${CONNECT_GROUP_UNIQUE_IDENTIFIER}"
export CONNECT_CONFIG_STORAGE_TOPIC="_${CONNECT_GROUP_ID}-configs"
export CONNECT_OFFSET_STORAGE_TOPIC="_${CONNECT_GROUP_ID}-offsets"
export CONNECT_STATUS_STORAGE_TOPIC="_${CONNECT_GROUP_ID}-status"

# Find Java and cacerts
JAVA_HOME=$(java -XshowSettings:properties -version 2>&1 > /dev/null | grep -i java.home | awk '{print $3}')
CACERT_FILE=$(find "${JAVA_HOME}" -type f -name "cacerts" | awk 'NR==1{print $1}')
export CONNECT_SSL_TRUSTSTORE_LOCATION="${CACERT_FILE}"
export CONNECT_CONSUMER_SSL_TRUSTSTORE_LOCATION="${CACERT_FILE}"

printf '%s\n' 'Launching Kafka Connect worker'
/etc/confluent/docker/run &

printf '%s\n' "Waiting for Kafka Connect to start listening on localhost:${CONNECT_REST_PORT}"
while ! /etc/confluent/docker/healthcheck.sh; do sleep 5; done

printf '%s\n' 'Generate the connector config...'
# Populate the template
dub template /tmp/config.template.json /tmp/config.json


printf '%s\n' "Creating Kafka Connect ${CONNECTOR_NAME} connector"
if [[ "${DEBUG_MODE}" == "true" ]]; then
  printf '%s\n' "Connector config:"
  cat /tmp/config.json
  printf '%s\n' "Printing raw curl output"
  curl -i -X PUT -H "Content-Type:application/json" \
    http://localhost:${CONNECT_REST_PORT}/connectors/${CONNECTOR_NAME}/config \
    -d @/tmp/config.json
  # When debugging force exit if connector creation failed
  # or potentially failed kconnect clusters might be deployed
  # if DEBUG_MODE is set to true
  [[ $(curl -s -o /dev/null -w %{http_code} http://localhost:${CONNECT_REST_PORT}/connectors/${CONNECTOR_NAME}) -eq 200 ]] || exit 1
else
  CURL_STATUS=$(curl -s -o /dev/null -i -X PUT -H "Content-Type:application/json" -w "%{http_code}" \
    http://localhost:${CONNECT_REST_PORT}/connectors/${CONNECTOR_NAME}/config \
    -d @/tmp/config.json)
  printf '%s\n' "Curl HTTP status code is ${CURL_STATUS}"
  if [ "${CURL_STATUS}" -ne 200 ] && [ "${CURL_STATUS}" -ne 201 ]; then
    printf '%s\n' "$(date): Updating ${CONNECTOR_NAME} did not return 200 or 201 (returned ${CURL_STATUS}): Exiting..."
    exit 1
  else
    printf '%s\n' "$(date): ${CONNECTOR_NAME} connector successfully updated"
  fi
fi

printf '%s\n' "$(date): ${CONNECTOR_NAME} connector started"
sleep infinity