#!/bin/sh
# Rehearses native MySQL GTID replication and a documented manual promotion.
# This is an isolated topology check; it does not provide automatic failover.
set -eu

TASK_DOCKER=${DOCKER:-docker}
TASK_MYSQL_IMAGE=${ORBIT_TEST_MYSQL_IMAGE:-mysql:8.4}
TASK_ORBIT=${ORBIT_BIN:-bin/orbit}
TASK_MIGRATE=${ORBIT_MIGRATE_BIN:-bin/orbit-migrate}
TASK_ADMIN=${ORBIT_ADMIN_BIN:-bin/orbit-admin}
TASK_SOURCE_NAME="orbit-ha-source-$$"
TASK_REPLICA_NAME="orbit-ha-replica-$$"
TASK_DIR=$(mktemp -d "${TMPDIR:-/tmp}/orbit-ha.XXXXXX")
TASK_SERVER_PID=
TASK_SOURCE_PORT=${ORBIT_TEST_SOURCE_PORT:-}
TASK_REPLICA_PORT=${ORBIT_TEST_REPLICA_PORT:-}
TASK_LAST_HOST_PORT=
TASK_SOURCE_IP=

cleanup() {
	if [ -n "$TASK_SERVER_PID" ]; then
		kill "$TASK_SERVER_PID" >/dev/null 2>&1 || true
		wait "$TASK_SERVER_PID" 2>/dev/null || true
	fi
	"$TASK_DOCKER" rm -f "$TASK_SOURCE_NAME" "$TASK_REPLICA_NAME" >/dev/null 2>&1 || true
	for task_artifact in "$TASK_DIR"/*; do
		if [ -f "$task_artifact" ]; then unlink "$task_artifact"; fi
	done
	rmdir "$TASK_DIR" 2>/dev/null || true
}
trap cleanup EXIT HUP INT TERM

umask 077
if [ ! -x "$TASK_ORBIT" ] || [ ! -x "$TASK_MIGRATE" ] || [ ! -x "$TASK_ADMIN" ]; then
	printf '%s\n' 'orbit, orbit-migrate and orbit-admin binaries are required; build them before this rehearsal.' >&2
	exit 2
fi

TASK_SOURCE_PASSWORD=$(openssl rand -hex 24)
TASK_REPLICA_PASSWORD=$(openssl rand -hex 24)
TASK_REPLICA_USER_PASSWORD=$(openssl rand -hex 16)
TASK_DATABASE=orbit_test

start_mysql() {
	container=$1
	password=$2
	host_port=$3
	server_id=$4
	port_mapping="127.0.0.1::3306"
	if [ -n "$host_port" ]; then port_mapping="127.0.0.1:$host_port:3306"; fi
	"$TASK_DOCKER" run -d --name "$container" --network bridge \
		-e MYSQL_ROOT_PASSWORD="$password" -e MYSQL_DATABASE="$TASK_DATABASE" \
		-p "$port_mapping" "$TASK_MYSQL_IMAGE" --mysqlx=OFF --server-id="$server_id" --log-bin=mysql-bin \
		--relay-log=relay-bin --binlog-format=ROW --gtid-mode=ON \
		--enforce-gtid-consistency=ON --log-replica-updates=ON >/dev/null
	ready=0
	for task_attempt in $(seq 1 90); do
		if "$TASK_DOCKER" exec "$container" sh -c \
			'MYSQL_PWD="$MYSQL_ROOT_PASSWORD" mysql -uroot --protocol=tcp -h127.0.0.1 -P3306 -e "SELECT 1" >/dev/null 2>&1'; then
			ready=1
			break
		fi
		if ! "$TASK_DOCKER" inspect -f '{{.State.Running}}' "$container" 2>/dev/null | grep -qx true; then
			break
		fi
		sleep 1
	done
	if [ "$ready" -ne 1 ]; then
		printf 'MySQL container did not become ready: %s\n' "$container" >&2
		"$TASK_DOCKER" logs --tail 40 "$container" >&2 || true
		exit 1
	fi
	TASK_LAST_HOST_PORT=$("$TASK_DOCKER" port "$container" 3306/tcp | sed 's/.*://')
}

source_mysql() {
	"$TASK_DOCKER" exec -e MYSQL_PWD="$TASK_SOURCE_PASSWORD" "$TASK_SOURCE_NAME" \
		mysql -uroot --protocol=tcp -h127.0.0.1 -P3306 "$@"
}

replica_mysql() {
	"$TASK_DOCKER" exec -e MYSQL_PWD="$TASK_REPLICA_PASSWORD" "$TASK_REPLICA_NAME" \
		mysql -uroot --protocol=tcp -h127.0.0.1 -P3306 "$@"
}

start_mysql "$TASK_SOURCE_NAME" "$TASK_SOURCE_PASSWORD" "$TASK_SOURCE_PORT" 1
TASK_SOURCE_PORT=$TASK_LAST_HOST_PORT
TASK_SOURCE_IP=$("$TASK_DOCKER" inspect -f '{{range.NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "$TASK_SOURCE_NAME")
start_mysql "$TASK_REPLICA_NAME" "$TASK_REPLICA_PASSWORD" "$TASK_REPLICA_PORT" 2
TASK_REPLICA_PORT=$TASK_LAST_HOST_PORT
if [ "$TASK_SOURCE_PORT" = "$TASK_REPLICA_PORT" ]; then
	printf '%s\n' 'source and replica host ports collided; set ORBIT_TEST_SOURCE_PORT and ORBIT_TEST_REPLICA_PORT explicitly' >&2
	exit 1
fi
TASK_SOURCE_DSN="root:$TASK_SOURCE_PASSWORD@tcp(127.0.0.1:$TASK_SOURCE_PORT)/$TASK_DATABASE"
TASK_REPLICA_DSN="root:$TASK_REPLICA_PASSWORD@tcp(127.0.0.1:$TASK_REPLICA_PORT)/$TASK_DATABASE"

ORBIT_MYSQL_DSN="$TASK_SOURCE_DSN" "$TASK_MIGRATE" up >/dev/null
ORBIT_MYSQL_DSN="$TASK_SOURCE_DSN" "$TASK_ADMIN" bootstrap \
	--application ha-rehearsal --namespace default --environment test --queue failover \
	--output "$TASK_DIR/credentials.json" >/dev/null

read TASK_TOKEN TASK_QUEUE <<EOF
$(python3 - "$TASK_DIR/credentials.json" <<'PY'
import json
import sys

with open(sys.argv[1], encoding="utf-8") as handle:
    value = json.load(handle)
print(value["token"], value["queue_id"])
PY
)
EOF

source_mysql -e "CREATE USER IF NOT EXISTS 'orbit_replica'@'%' IDENTIFIED BY '$TASK_REPLICA_USER_PASSWORD'; GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'orbit_replica'@'%'; FLUSH PRIVILEGES"
if ! "$TASK_DOCKER" exec -e MYSQL_PWD="$TASK_SOURCE_PASSWORD" "$TASK_SOURCE_NAME" \
	mysqldump -uroot --protocol=tcp -h127.0.0.1 -P3306 \
	--single-transaction --set-gtid-purged=ON --routines --triggers --events --hex-blob \
	--no-tablespaces "$TASK_DATABASE" > "$TASK_DIR/orbit.sql" 2>"$TASK_DIR/mysqldump.log"; then
	cat "$TASK_DIR/mysqldump.log" >&2
	exit 1
fi
"$TASK_DOCKER" exec -i -e MYSQL_PWD="$TASK_REPLICA_PASSWORD" "$TASK_REPLICA_NAME" \
	mysql -uroot --protocol=tcp -h127.0.0.1 -P3306 "$TASK_DATABASE" < "$TASK_DIR/orbit.sql"

replica_mysql -e "CHANGE REPLICATION SOURCE TO SOURCE_HOST='$TASK_SOURCE_IP', SOURCE_PORT=3306, SOURCE_USER='orbit_replica', SOURCE_PASSWORD='$TASK_REPLICA_USER_PASSWORD', SOURCE_AUTO_POSITION=1, GET_SOURCE_PUBLIC_KEY=1; START REPLICA"
replica_mysql -e "SET GLOBAL super_read_only=ON; SET GLOBAL read_only=ON"

ORBIT_MYSQL_DSN="$TASK_SOURCE_DSN" ORBIT_HTTP_ADDR=127.0.0.1:0 "$TASK_ORBIT" >"$TASK_DIR/source-server.log" 2>&1 &
TASK_SERVER_PID=$!
TASK_HTTP_PORT=
for task_attempt in $(seq 1 30); do
	TASK_HTTP_PORT=$(sed -n 's/.*"address":"127.0.0.1:\([0-9]*\)".*/\1/p' "$TASK_DIR/source-server.log" | head -n 1)
	if [ -n "$TASK_HTTP_PORT" ]; then break; fi
	if ! kill -0 "$TASK_SERVER_PID" 2>/dev/null; then cat "$TASK_DIR/source-server.log" >&2; exit 1; fi
	sleep 1
done
if [ -z "$TASK_HTTP_PORT" ]; then cat "$TASK_DIR/source-server.log" >&2; exit 1; fi

TASK_OPERATION_ID=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
cat >"$TASK_DIR/task.json" <<EOF
{"queue_id":"$TASK_QUEUE","handler":"ha.rehearsal","handler_version":"v1","payload_version":"v1","payload":{"phase":"source"},"recovery_safety":"RECONCILE_ONLY","retry":{"max_attempts":1}}
EOF
TASK_SOURCE_RESPONSE=$(curl -fsS --max-time 5 -X POST "http://127.0.0.1:$TASK_HTTP_PORT/v1/tasks" \
	-H "Authorization: Bearer $TASK_TOKEN" -H 'Content-Type: application/json' \
	-H "Idempotency-Key: $TASK_OPERATION_ID" --data-binary @"$TASK_DIR/task.json")
TASK_SOURCE_ID=$(python3 - "$TASK_SOURCE_RESPONSE" <<'PY'
import json
import sys

value = json.loads(sys.argv[1])
task = value.get("data", {}).get("task", {})
if not task.get("id"):
    raise SystemExit("source did not accept task")
print(task["id"])
PY
)
kill "$TASK_SERVER_PID" >/dev/null 2>&1 || true
wait "$TASK_SERVER_PID" 2>/dev/null || true
TASK_SERVER_PID=

TASK_REPLICA_ROWS=0
for task_attempt in $(seq 1 60); do
	TASK_REPLICA_ROWS=$(replica_mysql --batch --skip-column-names "$TASK_DATABASE" -e \
		"SELECT COUNT(*) FROM pulse_tasks" 2>/dev/null || true)
	if [ "$TASK_REPLICA_ROWS" = 1 ]; then break; fi
	sleep 1
done
if [ "$TASK_REPLICA_ROWS" != 1 ]; then
	replica_mysql -e 'SHOW REPLICA STATUS\G' >&2 || true
	exit 1
fi

TASK_REPLICATION_STATUS=$(replica_mysql -e 'SHOW REPLICA STATUS\G' 2>/dev/null | \
	awk -F': ' '/Replica_IO_Running:/{io=$2} /Replica_SQL_Running:/{sql=$2} END{print io ":" sql}')
case "$TASK_REPLICATION_STATUS" in
	Yes:Yes) ;;
	*) printf 'replication was not healthy: %s\n' "$TASK_REPLICATION_STATUS" >&2; exit 1 ;;
esac

"$TASK_DOCKER" stop "$TASK_SOURCE_NAME" >/dev/null
TASK_SOURCE_STOPPED=true
replica_mysql -e 'STOP REPLICA; RESET REPLICA ALL; SET GLOBAL super_read_only=OFF; SET GLOBAL read_only=OFF'
ORBIT_MYSQL_DSN="$TASK_REPLICA_DSN" "$TASK_ADMIN" verify >/dev/null

ORBIT_MYSQL_DSN="$TASK_REPLICA_DSN" ORBIT_HTTP_ADDR=127.0.0.1:0 "$TASK_ORBIT" >"$TASK_DIR/replica-server.log" 2>&1 &
TASK_SERVER_PID=$!
TASK_HTTP_PORT=
for task_attempt in $(seq 1 30); do
	TASK_HTTP_PORT=$(sed -n 's/.*"address":"127.0.0.1:\([0-9]*\)".*/\1/p' "$TASK_DIR/replica-server.log" | head -n 1)
	if [ -n "$TASK_HTTP_PORT" ]; then break; fi
	if ! kill -0 "$TASK_SERVER_PID" 2>/dev/null; then cat "$TASK_DIR/replica-server.log" >&2; exit 1; fi
	sleep 1
done
if [ -z "$TASK_HTTP_PORT" ]; then cat "$TASK_DIR/replica-server.log" >&2; exit 1; fi

TASK_REPLAY_RESPONSE=$(curl -fsS --max-time 5 -X POST "http://127.0.0.1:$TASK_HTTP_PORT/v1/tasks" \
	-H "Authorization: Bearer $TASK_TOKEN" -H 'Content-Type: application/json' \
	-H "Idempotency-Key: $TASK_OPERATION_ID" --data-binary @"$TASK_DIR/task.json")
TASK_REPLAY_ID=$(python3 - "$TASK_REPLAY_RESPONSE" <<'PY'
import json
import sys

value = json.loads(sys.argv[1])
task = value.get("data", {}).get("task", {})
if not task.get("id"):
    raise SystemExit("promoted replica did not accept idempotent replay")
print(task["id"])
PY
)
if [ "$TASK_REPLAY_ID" != "$TASK_SOURCE_ID" ]; then
	printf 'idempotent replay changed task: source=%s promoted=%s\n' "$TASK_SOURCE_ID" "$TASK_REPLAY_ID" >&2
	exit 1
fi

printf '{"replication_io_sql":"%s","replicated_task_rows":%s,"source_stopped":true,"manual_promotion_verified":true,"idempotent_replay_same_task":true,"automatic_failover":false}\n' \
	"$TASK_REPLICATION_STATUS" "$TASK_REPLICA_ROWS"