#!/usr/bin/env bash

printf '%s\n' "Setting programmatically derived runtime variables"
if [ "${Environment}" != 'local' ]; 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}"
export CONNECT_PRODUCER_SSL_TRUSTSTORE_LOCATION="${CACERT_FILE}"

# Connector topic config
export  DATABASE_HISTORY_KAFKA_TOPIC="_${DATABASE_SERVER_NAME}_history"


# TODO: convert secrets to JSON objects and use secrets provider instead
if [[ -z "${DB_PASSWORD}" ]]; then
  SECRET_NAME="${Environment}/${SERVICE_NAME}/${SECRETS_MANAGER_DB_KEY}"
  DB_PASSWORD=$(aws secretsmanager get-secret-value --secret-id ${SECRET_NAME} --region ${AWS_REGION} --query 'SecretString' --output text)
fi
sed -i "s/<REPLACE_DB_HOST>/${DB_HOST}/g" /etc/debezium-mysql/mysql.properties
sed -i "s/<REPLACE_DB_PORT>/${DB_PORT}/g" /etc/debezium-mysql/mysql.properties
sed -i "s/<REPLACE_DB_USER>/${DB_USER}/g" /etc/debezium-mysql/mysql.properties
sed -i "s/<REPLACE_DB_PASSWORD>/${DB_PASSWORD}/g" /etc/debezium-mysql/mysql.properties

export KAFKA_JMX_OPTS="${KAFKA_JMX_OPTS}"
printf '%s %s \n' 'KAFKA_JMX_OPTS set to ' "${KAFKA_JMX_OPTS}"

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 -s -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})" == "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 '\n%s\n' "Curl HTTP status code is ${CURL_STATUS}"
  if [ "${CURL_STATUS}" -ne 200 ] && [ "${CURL_STATUS}" -ne 201 ]; then
    printf '\n%s\n' "$(date): Updating ${CONNECTOR_NAME} did not return 200 or 201 (returned ${CURL_STATUS}): Exiting..."
    exit 1
  else
    printf '\n%s\n' "$(date): ${CONNECTOR_NAME} connector successfully updated"
  fi
fi

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