#!/usr/bin/env bash
compose_file="schema-test-docker-compose.yml"
config_file="schema-test-source-config.json"
connect_log="connect.log"
records_file="records.json"

function compose(){
    sudo docker-compose -f ${compose_file} $*
}

function clean(){
    echo "Shutting down services"
    compose down -v
    echo "Deleting files"
    rm ${compose_file} ${config_file} ${connect_log} ${records_file}
}

if [[ ${1} = 'clean' ]]; then
    clean
    exit 0
fi

if [[ -z "$(which jq)" ]]; then
    echo "You'll need jq installed, see https://stedolan.github.io/jq/download/"
    exit 1
fi

if [[ -z "$(which docker-compose)" ]]; then
    echo "You'll need docker-compose (and docker) installed, see https://docs.docker.com/compose/install/"
    exit 1
fi

echo "creating ${config_file}"
tee ./${config_file} <<EOF
{
  "name": "source-test",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "ddl.parser.mode": "antlr",
    "database.history.store.only.monitored.tables.ddl": "true",
    "database.history.skip.unparseable.ddl": "true",
    "tasks.max": "1",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.user": "******",
    "database.password": "*******",
    "database.server.id": "1337",
    "database.serverTimezone": "Europe/Berlin",
    "table.whitelist": "test.schema_test",
    "database.history.kafka.bootstrap.servers": "broker:9092",
    "database.history.kafka.topic": "cdc-history-test",
    "include.schema.changes": "true",
    "min.row.count.to.stream.results": "0",
    "database.history.kafka.recovery.poll.interval.ms": "1000",
    "database.history.kafka.recovery.attempts": "12000",
    "max.batch.size": "16384",
    "max.queue.size": "65536",
    "decimal.handling.mode": "precise",
    "snapshot.mode": "when_needed",
    "connect.timeout.ms": "120000",
    "database.server.name": "test-db",
    "transforms": "route",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "(.*)\\\\.(.*)\\\\.(.*)",
    "transforms.route.replacement": "cdc-data-\$2-\$3"
  }
}
EOF

echo "creating ${compose_file}"
tee ./${compose_file} <<EOF
# Based on https://github.com/confluentinc/cp-docker-images/blob/master/examples/cp-all-in-one/docker-compose.yml
version: '2'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:5.0.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  broker:
    image: confluentinc/cp-kafka:5.0.0
    hostname: broker
    container_name: broker
    depends_on:
      - zookeeper
    ports:
      - 9092:9092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_GROUP_MIN_SESSION_TIMEOUT_MS: 1000

  schema-registry:
    image: confluentinc/cp-schema-registry:5.0.0-beta30
    hostname: schema-registry
    container_name: schema-registry
    depends_on:
      - zookeeper
      - broker
    ports:
      - 8081:8081
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181'

  rest-proxy:
    image: confluentinc/cp-kafka-rest:5.0.0
    depends_on:
      - zookeeper
      - broker
      - schema-registry
    ports:
      - 8082:8082
    hostname: rest-proxy
    container_name: rest-proxy
    environment:
      KAFKA_REST_HOST_NAME: rest-proxy
      KAFKA_REST_BOOTSTRAP_SERVERS: 'broker:9092'
      KAFKA_REST_LISTENERS: "http://0.0.0.0:8082"
      KAFKA_REST_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'

  connect:
    image: debezium/connect:0.8.3.Final
    depends_on:
      - mysql
      - broker
    ports:
    - 9595:9595/tcp
    - 9200:9200/tcp
    - 8083:8083/tcp
    environment:
      BOOTSTRAP_SERVERS: "broker:9092"
      HOST_NAME: "connect"
      ADVERTISED_HOST_NAME: "connect"
      ADVERTISED_PORT: "8083"
      GROUP_ID: "connect-worker-1"
      OFFSET_FLUSH_INTERVAL_MS: "30000"
      OFFSET_FLUSH_TIMEOUT_MS: "25000"
      CONFIG_STORAGE_TOPIC: "kafka-connect-configs-worker-1"
      OFFSET_STORAGE_TOPIC: "kafka-connect-offsets-worker-1"
      CONNECT_STATUS_STORAGE_TOPIC: "kafka-connect-status-worker-1"
      KEY_CONVERTER: "io.confluent.connect.avro.AvroConverter"
      VALUE_CONVERTER: "io.confluent.connect.avro.AvroConverter"
      INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE: "true"
      CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE: "true"
      CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'
      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: "1"
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: "1"
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: "1"
      CONNECT_INTERNAL_KEY_CONVERTER_SCHEMAS_ENABLE: "false"
      CONNECT_INTERNAL_VALUE_CONVERTER_SCHEMAS_ENABLE: "false"
      CONNECT_ACCESS_CONTROL_ALLOW_ORIGIN: "*"
      CONNECT_ACCESS_CONTROL_ALLOW_METHODS: "GET,POST,PUT,DELETE,HEAD"
      KAFKA_HEAP_OPTS: "-Xms2G -Xmx20G -Xloggc:/var/log/jvm-logs/verbose-gc.log -verbose:gc -XX:+PrintGCDateStamps -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/jvm-logs"
      KAFKA_JMX_PORT: 9200
      KAFKA_JMX_OPTS: '-Dcom.sun.management.jmxremote=true -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false'
      KAFKA_JMX_HOSTNAME: "localhost"

  mysql:
    image: mysql:5.6
    ports:
      - 3306:3306
    environment:
      MYSQL_ROOT_PASSWORD: root
      TZ: Europe/Berlin
    command: --log-bin=mysqld-bin --binlog-format=row --server-id=1
EOF

echo "docker-compose up"
compose up >/dev/null 2>/dev/null &

echo "Wait for kafka to be up (mysql will be long ready by then)"

wait_attempts=300
until (curl -s "http://localhost:8082/topics" >/dev/null); do
    sleep 0.1s
    wait_attempts="$((wait_attempts - 1))"
    if [[ "${wait_attempts}" = "0" ]]; then
        echo "timeout during waiting for kafka!"
        exit 1
    fi
done

echo "Create table in database"
(
mysql -u root -proot -h 127.0.0.1 -P 3306 <<EOF
    create schema if not exists test;

    use test;

    create table schema_test (
        id int primary key,
        record_type enum('internal', 'external') not null default 'internal',
        comment varchar(64)
    );
EOF
) || exit

echo "Start connector"
curl -s -X POST -H "Content-Type: application/json" --data "@${config_file}" "localhost:8083/connectors" || exit
echo ""

echo "Insert data into table"
(
mysql -u root -proot -h 127.0.0.1 -P 3306 test <<EOF
    insert into schema_test values
        (1, 'internal', 'I\'m not null'),
        (2, 'external', 'I\'m not null')
    ;
    insert into schema_test (id, comment) values (3, 'I use the default');

EOF
) || exit

echo "Wait until schema subject published"
wait_attempts=60
while
    error_code="$(curl -s localhost:8081/subjects/cdc-data-test-schema_test-value/versions/ | jq '.error_code?')"
    [[ "${error_code}" = "40401" ]]  # this error code means schema not found
do
    sleep 1s
    wait_attempts="$((wait_attempts - 1))"
    if [[ "${wait_attempts}" = "0" ]]; then
        echo "timeout during waiting for kafka!"
        exit 1
    fi
done

echo "Wait until version published"
wait_attempts=60
while
    versions="$(curl -s localhost:8081/subjects/cdc-data-test-schema_test-value/versions/)"
    [[ "${versions}" = "[]" ]]
do
    sleep 1s
    wait_attempts="$((wait_attempts - 1))"
    if [[ "${wait_attempts}" = "0" ]]; then
        echo "timeout during waiting for schema!"
        exit 1
    fi
done

echo "Register consumer"
curl -s -X POST -H "Content-Type: application/vnd.kafka.v2+json" --data '{"name": "my_consumer_instance", "format": "avro", "auto.offset.reset": "earliest"}' http://localhost:8082/consumers/my_avro_consumer || exit

echo "Subscribe to topic"
curl -s -X POST -H "Content-Type: application/vnd.kafka.v2+json" --data '{"topics":["cdc-data-test-schema_test"]}' http://localhost:8082/consumers/my_avro_consumer/instances/my_consumer_instance/subscription || exit

echo "Get messages"
curl -s -X GET -H "Accept: application/vnd.kafka.avro.v2+json" http://localhost:8082/consumers/my_avro_consumer/instances/my_consumer_instance/records | jq '.' > ${records_file} || exit
cat ${records_file} | jq -c '.[].value.after["test_db.test.schema_test.Value"]'

echo "Creating additional table"
(
mysql -u root -proot -h 127.0.0.1 -P 3306 <<EOF
    use test;

    create table migration_test (
        id varchar(20) null,
        mgb_no varchar(20) null
    );
EOF
) || exit

echo "Adding records to new table"
(
mysql -u root -proot -h 127.0.0.1 -P 3306 test <<EOF
    insert into migration_test values
            ('1', '42'),
            ('2', '23')
        ;
EOF
) || exit

echo "Creating unique index"
(
mysql -u root -proot -h 127.0.0.1 -P 3306 <<EOF
    use test;
    create unique index migration_test_mgb_no_uindex
    on migration_test (mgb_no);
;
EOF
) || exit

echo "Alter table and insert some rows"
(
mysql -u ***** -p***** -h 127.0.0.1 -P 3306 <<EOF
    use test;
    alter table schema_test add column one_more varchar(100) after comment;

    insert into schema_test values
        (23, 'internal', 'I\'m not null', 'blubb')
    ;
;
EOF
)

echo "Wait until second schema version published"
wait_attempts=60
while
    new_versions="$(curl -s localhost:8081/subjects/cdc-data-test-schema_test-value/versions/)"
    [[ "${versions}" = "${new_versions}" ]]
do
    sleep 1s
    wait_attempts="$((wait_attempts - 1))"
    if [[ "${wait_attempts}" = "0" ]]; then
        echo "timeout during waiting for schema!"
        exit 1
    fi
done

echo "Get messages"
curl -s -X GET -H "Accept: application/vnd.kafka.avro.v2+json" http://localhost:8082/consumers/my_avro_consumer/instances/my_consumer_instance/records | jq '.'  >> ${records_file} || exit
cat ${records_file} | jq -c '.[].value.after["test_db.test.schema_test.Value"]'

echo "Show schema"
curl -s localhost:8081/subjects/cdc-data-test-schema_test-value/versions/latest | jq -r '.schema | fromjson | .fields[] | select( .name == "before") | .type[1].fields' || exit

echo "select data from mysql"
mysql -u root -proot -h 127.0.0.1 -P 3306 test -e "select * from schema_test"

echo "dumping connect log at ${connect_log}, records are at ${records_file}"
compose logs connect >> ${connect_log}

printf "remove files and shut down? (yN): "
read ans
if [[ "$ans" = "y" || "$ans" = "Y" ]]; then
    clean
    exit 0
fi

echo "You can run test_add_columns_after.sh clean to clean up afterwards"