#!/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'
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 true; do
  CURL_STATUS=$(curl -s -o /dev/null -w %{http_code} http://localhost:${CONNECT_REST_PORT}/connectors)
  printf '%s\n' "$(date): Kafka Connect listener HTTP state: ${CURL_STATUS}. Waiting for 200..."
  if [ "${CURL_STATUS}" -eq 200 ]; then
    break
  fi
  sleep 5
done

cat << EOF > /tmp/config.json
{
  "connector.class": "org.apache.camel.kafkaconnector.salesforce.CamelSalesforceSourceConnector",
  "topics": "${KAFKA_TOPICS}",
  "camel.sink.endpoint.rawPayload": "true",
  "camel.sink.endpoint.sObjectName": "${SF_SOBJECT_NAME}",
  "camel.sink.endpoint.sObjectIdName": "${SF_SOBJECT_ID_NAME}",
  "camel.sink.path.operationName": "${SF_OPERATION_NAME}"
  "camel.component.salesforce.login-config.clientId": "${SF_CONSUMER_KEY}",
  "camel.component.salesforce.login-config.clientSecret": "${SF_CONSUMER_SECRET}",
  "camel.component.salesforce.login-config.loginUrl": "${SF_INSTANCE}",
  "camel.component.salesforce.login-config.refreshToken": "${SF_REFRESH_TOKEN}",
  "camel.component.salesforce.login-config.type":"${SF_AUTH_TYPE}",
  "camel.component.salesforce.format": "JSON",
  "camel.sink.endpoint.contentType": "JSON",
  "key.converter": "${CONNECT_KEY_CONVERTER}",
  "value.converter": "${CONNECT_VALUE_CONVERTER}"
}
EOF

printf '%s\n' "Creating Kafka Connect ${CONNECTOR_NAME} connector"
if [ -n "${DEBUG_MODE}" ]; then
  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
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

sleep infinity