#!/bin/bash
# Usage: repro.sh BINDIR LABEL SECONDS [CLIENTS]
# Primary + one streaming standby over unix sockets, the standby named in
# synchronous_standby_names, pgbench one-row inserts at SYNC_COMMIT (default remote_write).
set -euo pipefail
here=$(cd "$(dirname "$0")" && pwd)
bindir=$1
label=$2
secs=$3
clients=${4:-${CLIENTS:-1}}
base=/tmp/walsnd-repro/$label
sock=$base/sock
pport=55432
sport=55433
export PATH=$bindir:$PATH

if [ -d "$base" ]; then
  pg_ctl -D "$base/standby" stop -m immediate >/dev/null 2>&1 || true
  pg_ctl -D "$base/primary" stop -m immediate >/dev/null 2>&1 || true
  rm -rf "$base"
fi
mkdir -p "$sock" "$base/out"

initdb -D "$base/primary" -A trust -U postgres >/dev/null
cat >> "$base/primary/postgresql.conf" <<CONF
listen_addresses = ''
unix_socket_directories = '$sock'
port = $pport
wal_level = replica
max_wal_senders = 4
synchronous_standby_names = 's1'
synchronous_commit = ${SYNC_COMMIT:-remote_write}
${EXTRA_PRIMARY_CONF:-}
shared_buffers = 256MB
logging_collector = off
CONF
pg_ctl -D "$base/primary" -l "$base/primary.log" -w start >/dev/null

pg_basebackup -h "$sock" -p $pport -U postgres -D "$base/standby" -X stream >/dev/null
touch "$base/standby/standby.signal"
cat >> "$base/standby/postgresql.conf" <<CONF
port = $sport
primary_conninfo = 'host=$sock port=$pport user=postgres application_name=s1'
hot_standby = on
${EXTRA_STANDBY_CONF:-}
CONF
pg_ctl -D "$base/standby" -l "$base/standby.log" -w start >/dev/null

psql_p() { psql -h "$sock" -p $pport -U postgres -X -q -A -t "$@"; }

for _ in $(seq 1 50); do
  st=$(psql_p -c "select sync_state from pg_stat_replication where application_name = 's1'")
  [ "$st" = "sync" ] && break
  sleep 0.2
done
echo "standby sync_state=$st"
psql_p -c "select version()"
psql_p -c "create table t (id bigserial primary key, v int)"
echo 'insert into t (v) values (1);' > "$base/insert.sql"
if [ -n "${THINK_MS:-}" ]; then echo "\\sleep ${THINK_MS} ms" >> "$base/insert.sql"; fi

cat > "$base/sample.sql" <<'SQL'
select to_char(clock_timestamp(), 'HH24:MI:SS.MS') as at,
       pg_wal_lsn_diff(pg_current_wal_flush_lsn(), r.sent_lsn) as unsent,
       pg_wal_lsn_diff(r.sent_lsn, r.write_lsn) as unwritten,
       pg_wal_lsn_diff(r.write_lsn, r.flush_lsn) as unflushed,
       coalesce(a.wait_event, 'cpu') as walsender_wait,
       (select count(*) from pg_stat_activity where wait_event = 'SyncRep') as syncrep,
       (select coalesce(string_agg(coalesce(wait_event_type || ':' || wait_event, state), ','), '-') from pg_stat_activity where application_name = 'pgbench') as client
from pg_stat_replication r join pg_stat_activity a using (pid)
\watch 0.05
SQL
cat > "$base/sample_standby.sql" <<'SQL'
select to_char(clock_timestamp(), 'HH24:MI:SS.MS') as at,
       written_lsn, flushed_lsn,
       (select coalesce(wait_event, 'cpu') from pg_stat_activity where backend_type = 'walreceiver') as walreceiver_wait,
       (select coalesce(wait_event, 'cpu') from pg_stat_activity where backend_type = 'startup') as startup_wait
from pg_stat_wal_receiver
\watch 0.05
SQL
psql_p -F ' ' -f "$base/sample.sql" > "$base/samples.txt" 2>&1 &
psql -h "$sock" -p $sport -U postgres -X -q -A -t -F ' ' -f "$base/sample_standby.sql" > "$base/samples_standby.txt" 2>&1 &

cd "$base/out"
pgbench -h "$sock" -p $pport -U postgres -n -M prepared -c "$clients" -j "$clients" -T "$secs" \
  -f "$base/insert.sql" --log --log-prefix=tx postgres > "$base/pgbench.txt" 2>&1
grep -E 'number of transactions actually processed|latency average|tps' "$base/pgbench.txt" || cat "$base/pgbench.txt"

cat "$base"/out/tx.* | awk -v label="$label" '
  { n++; us=$3; if (us > max) max = us; if (us > 100000) { s++; stalls = stalls " " int(us/1000) "ms" } if (us > 10000) t++ }
  END { printf "%s: commits=%d over10ms=%d over100ms=%d max=%.1fms\n", label, n, t, s, max/1000; if (s) print "stalls:" stalls }'

python3 "$here/analyse.py" "$base" || true
pg_ctl -D "$base/standby" stop -m fast >/dev/null
pg_ctl -D "$base/primary" stop -m fast >/dev/null