#!/usr/bin/env bash
# surinch.com · 盈尺：MySQL 复制延迟演示 v1.0 —— 一个大事务 vs 分批小事务，对比 Seconds_Behind_Source 与心跳延迟
# 仅限测试实例！会在源库建库 repl_lab、写 100 万行自造数据并整表 UPDATE。不要对生产或生产从库执行。
# 测试环境：MySQL 8.0.46 源库(3306) + 从库(3307)，异步复制，replica_parallel_workers=4。
# 用法：
#   SRC="mysql --defaults-file=/path/source.cnf" REP="mysql --defaults-file=/path/replica.cnf" ./mysql-replication-lag-demo-v1.0.sh init
#   ... ./mysql-replication-lag-demo-v1.0.sh big   out-big.txt     # 一条 UPDATE 改 100 万行
#   ... ./mysql-replication-lag-demo-v1.0.sh chunk out-chunk.txt   # 按主键每 1 万行一批，共 100 批
# 输出：<out>（源库耗时、追平时间、两端 SUM 校验）和 <out>.poll（从库每 0.5 秒一行：心跳延迟、正在应用落后秒数、忙碌 worker、SBS）
set -euo pipefail
SRC=${SRC:?请设置 SRC=源库 mysql 客户端命令}; REP=${REP:?请设置 REP=从库 mysql 客户端命令}
MODE=${1:?init|big|chunk}; OUT=${2:-out-$MODE.txt}
S="$SRC -N"; Q="$REP -N"; Q2="$REP"
if [ "$MODE" = init ]; then
  $SRC <<'SQL'
CREATE DATABASE IF NOT EXISTS repl_lab;
USE repl_lab;
DROP TABLE IF EXISTS orders; DROP TABLE IF EXISTS heartbeat;
CREATE TABLE orders(id INT PRIMARY KEY, user_id INT NOT NULL, status VARCHAR(16) NOT NULL,
  amount DECIMAL(10,2) NOT NULL, note VARCHAR(64) NOT NULL, updated_at DATETIME(6) NOT NULL);
CREATE TABLE heartbeat(id TINYINT PRIMARY KEY, ts DATETIME(6) NOT NULL);
INSERT INTO heartbeat VALUES (1, NOW(6));
SET SESSION cte_max_recursion_depth=1000000;
INSERT INTO orders
WITH RECURSIVE seq(n) AS (SELECT 1 UNION ALL SELECT n+1 FROM seq WHERE n < 1000000)
SELECT n, n % 50000, ELT(1 + n % 4,'paid','shipped','refunded','created'), ROUND((n % 997) * 1.37, 2),
       CONCAT('synthetic-order-', LPAD(n, 12, '0')), '2026-10-01 00:00:00' FROM seq;
SQL
  echo "init done; 等从库追平后再跑 big / chunk"; exit 0
fi
( while true; do $S -e "UPDATE repl_lab.heartbeat SET ts=NOW(6) WHERE id=1" 2>/dev/null; sleep 0.5; done ) & HB=$!
( while true; do
  $Q -e "SELECT DATE_FORMAT(NOW(6),'%H:%i:%s.%f'),
    (SELECT ROUND(TIMESTAMPDIFF(MICROSECOND, ts, NOW(6))/1e6,2) FROM repl_lab.heartbeat WHERE id=1) AS hb_lag_s,
    (SELECT MAX(ROUND(TIMESTAMPDIFF(MICROSECOND, APPLYING_TRANSACTION_ORIGINAL_COMMIT_TIMESTAMP, NOW(6))/1e6,2)) FROM performance_schema.replication_applier_status_by_worker WHERE APPLYING_TRANSACTION<>'') AS applying_age_s,
    (SELECT COUNT(*) FROM performance_schema.replication_applier_status_by_worker WHERE APPLYING_TRANSACTION<>'') AS busy_workers" 2>/dev/null | tr '\n' ' '
  $Q2 -e "SHOW REPLICA STATUS\G" 2>/dev/null | awk -F': ' '/Seconds_Behind_Source/{printf "SBS=%s ",$2} /Replica_SQL_Running_State/{printf "| %s",$2}'
  echo; sleep 0.5; done ) > "$OUT.poll" & PL=$!
trap 'kill $HB $PL 2>/dev/null || true' EXIT
sleep 3
echo "start $(date +%T.%N)" > "$OUT"
t0=$(date +%s%N)
if [ "$MODE" = big ]; then
  $S -e "UPDATE repl_lab.orders SET amount=amount+1, updated_at=NOW(6)"
else
  for i in $(seq 0 99); do lo=$((i*10000+1)); hi=$(((i+1)*10000))
    $S -e "UPDATE repl_lab.orders SET amount=amount+1, updated_at=NOW(6) WHERE id BETWEEN $lo AND $hi"; done
fi
echo "source_elapsed_ms=$(( ($(date +%s%N) - t0)/1000000 ))" >> "$OUT"
echo "source_done $(date +%T.%N)" >> "$OUT"
ok=0; for k in $(seq 1 240); do
  l=$($Q -e "SELECT TIMESTAMPDIFF(MICROSECOND, ts, NOW(6))/1e6 FROM repl_lab.heartbeat WHERE id=1")
  if awk "BEGIN{exit !($l<1)}"; then ok=$((ok+1)); else ok=0; fi
  [ $ok -ge 4 ] && break; sleep 0.5; done
echo "replica_caught_up $(date +%T.%N)" >> "$OUT"
sleep 1
$Q -e "SELECT SUM(amount) FROM repl_lab.orders" | sed 's/^/replica_sum_amount=/' >> "$OUT"
$S -e "SELECT SUM(amount) FROM repl_lab.orders" | sed 's/^/source_sum_amount=/' >> "$OUT"
cat "$OUT"