#!/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

if [[ "${SNOWFLAKE_CONFIGURATION}" == "true" ]]; then
  # Use BouncyCastle for SnowflakeJDBC to decrypt password-encrypted private keys
  export KAFKA_OPTS="-Dnet.snowflake.jdbc.enableBouncyCastle=TRUE"
  if [ "${Environment}" == 'local' ]; then
    # Configure Snowflake connection from local environment including private key
    export DATABASE_CONNECTION_URL="jdbc:snowflake://${SNOWFLAKE_SUBDOMAIN}.snowflakecomputing.com/?warehouse=${SNOWFLAKE_WAREHOUSE}&db=${SNOWFLAKE_DATABASE}&role=${SNOWFLAKE_ROLE}&schema=${SNOWFLAKE_SCHEMA}&user=${SNOWFLAKE_USER}&private_key_file=/home/appuser/snowflake_private_key/${SNOWFLAKE_PRIVATE_KEY_FILENAME}&private_key_pwd=${SNOWFLAKE_PRIVATE_KEY_PASSPHRASE}"
  else
    # Configure Snowflake connection from AWS environment including private key
    export DATABASE_CONNECTION_URL="jdbc:snowflake://${SNOWFLAKE_SUBDOMAIN}.snowflakecomputing.com/?warehouse=${SNOWFLAKE_WAREHOUSE}&db=${SNOWFLAKE_DATABASE}&role=${SNOWFLAKE_ROLE}&schema=${SNOWFLAKE_SCHEMA}&user=${SNOWFLAKE_USER}&private_key_base64=\${aws:${Environment}/${SERVICE_NAME}/SNOWFLAKE_PRIVATE_KEY:private_key_b64}&private_key_pwd=\${aws:${Environment}/${SERVICE_NAME}/SNOWFLAKE_PRIVATE_KEY:private_key_passphrase}"
  fi
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}"

# Configure transform list when set
FIELD_NAME_FOR_KEY=${FIELD_NAME_FOR_KEY:-}
FIELD_NAME_FOR_VALUE=${FIELD_NAME_FOR_VALUE:-}
TRANSFORMS=""
if [ -n "$FIELD_NAME_FOR_KEY" ]; then
  TRANSFORMS='createKey,extractInt'
fi

if [ -n "$FIELD_NAME_FOR_VALUE" ]; then
  TRANSFORMS="$TRANSFORMS,extractField"
fi
printf '%s\n' "Configured SMTs as: ${TRANSFORMS}"
export TRANSFORMS

if [[ "${SKIP_CONNECTOR_CONFIGURATION}" == "true" ]]; then
  printf '%s\n' 'Launching Kafka Connect worker. Connector configuration is not required.'
  /etc/confluent/docker/run
else
  printf '%s\n' 'Launching Kafka Connect worker'
  /etc/confluent/docker/run &
fi

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