#!/bin/bash

#
set -u
PGBIN="${PGBIN:-$(pg_config --bindir 2>/dev/null)}"
if [ -z "$PGBIN" ] || [ ! -x "$PGBIN/initdb" ]; then
	echo "set PGBIN to a PostgreSQL bin directory (no pg_config in PATH)" >&2
	exit 1
fi
PUB=/tmp/dl_pub; SUB=/tmp/dl_sub; PP=${PP:-55971}; SP=${SP:-55972}
ROWS=${ROWS:-3000000}
SPLIT=$((ROWS/2))

echo "### binaries: $PGBIN"
"$PGBIN/postgres" --version

for d in $PUB $SUB; do "$PGBIN/pg_ctl" -D $d stop -m immediate >/dev/null 2>&1; done
# pg_ctl -l appends, so a previous run would otherwise leave stale evidence.
rm -rf $PUB $SUB /tmp/dlp.log /tmp/dls.log
"$PGBIN/initdb" -D $PUB --no-sync -A trust >/dev/null 2>&1
"$PGBIN/initdb" -D $SUB --no-sync -A trust >/dev/null 2>&1
"$PGBIN/pg_ctl" -D $PUB -o "-p $PP -k /tmp -c wal_level=logical" -l /tmp/dlp.log start >/dev/null
"$PGBIN/pg_ctl" -D $SUB -o "-p $SP -k /tmp -c wal_level=logical -c deadlock_timeout=1s -c log_lock_waits=on" -l /tmp/dls.log start >/dev/null
sleep 2
P(){ "$PGBIN/psql" -p $PP -h /tmp -Xqt -d postgres -c "$1" 2>&1|sed 's/^ *//;/^$/d'; }
S(){ "$PGBIN/psql" -p $SP -h /tmp -Xqt -d postgres -c "$1" 2>&1|sed 's/^ *//;/^$/d'; }


echo "building a $ROWS row table..."
P "CREATE TABLE t (a int primary key, pad text);
   INSERT INTO t SELECT g, repeat('x',200) FROM generate_series(1,$ROWS) g;
   CREATE PUBLICATION p_hi FOR TABLE t WHERE (a >= $SPLIT);
   CREATE PUBLICATION p_lo FOR TABLE t WHERE (a < $SPLIT);" >/dev/null
S "CREATE TABLE t (a int primary key, pad text);" >/dev/null

echo
echo "### sub2 syncs first, then is disabled (REFRESH TABLE requires that) ###"
S "CREATE SUBSCRIPTION sub2 CONNECTION 'host=/tmp port=$PP dbname=postgres' PUBLICATION p_hi" >/dev/null 2>&1
for i in $(seq 1 120); do
	[ "$(S "SELECT count(*) FROM pg_subscription_rel WHERE srsubstate <> 'r'"|tr -d ' ')" = "0" ] && break; sleep 1
done
echo "  sub2 synced: $(S 'SELECT count(*) FROM t') rows"
S "ALTER SUBSCRIPTION sub2 DISABLE" >/dev/null
for i in $(seq 1 30); do
	[ "$(S "SELECT count(*) FROM pg_stat_subscription WHERE subname='sub2' AND pid IS NOT NULL"|tr -d ' ')" = "0" ] && break; sleep 1
done
echo "  sub2 disabled, workers gone"

echo
echo "### sub1 starts copying the same table ###"
S "CREATE SUBSCRIPTION sub1 CONNECTION 'host=/tmp port=$PP dbname=postgres' PUBLICATION p_lo" >/dev/null 2>&1
for i in $(seq 1 60); do
	n=$(S "SELECT count(*) FROM pg_stat_activity WHERE backend_type='logical replication tablesync worker'"|tr -d ' ')
	[ "$n" != "0" ] && break
	sleep 0.3
done
echo "  tablesync workers running: $(S "SELECT count(*) FROM pg_stat_activity WHERE backend_type='logical replication tablesync worker'")"
echo "  lock it holds on t: $(S "SELECT coalesce(string_agg(DISTINCT l.mode,','),'none') FROM pg_locks l JOIN pg_stat_activity a ON a.pid=l.pid WHERE l.relation='t'::regclass AND a.backend_type LIKE 'logical replication%'")"

echo
echo "### now: ALTER SUBSCRIPTION sub2 REFRESH TABLE t ###"
out=$(S "ALTER SUBSCRIPTION sub2 REFRESH TABLE t")
echo "  ${out:-<accepted>}"

echo
echo "### server log ###"
grep -iE "deadlock|still waiting|process .* acquired" /tmp/dls.log | tail -8 | sed 's/^/  /'

echo
if grep -q "deadlock detected" /tmp/dls.log; then
	victim=$(grep -oE "process [0-9]+ detected deadlock" /tmp/dls.log | head -1 | grep -oE "[0-9]+")
	echo "### RESULT: DEADLOCK REPRODUCED (victim pid $victim) ###"
else
	echo "### RESULT: no deadlock; the command was resolved on its own merits ###"
fi

for d in $PUB $SUB; do "$PGBIN/pg_ctl" -D $d stop -m immediate >/dev/null 2>&1; done
rm -rf $PUB $SUB