#!/usr/bin/env bash
# Verifies every factual claim in:
#   https://www.heatware.net/postgresql/active-active-replication-guide/
# Builds TWO throwaway PostgreSQL clusters, sets up genuine bidirectional logical
# replication between them, demonstrates both convergence and permanent divergence,
# prints PASS/FAIL per claim, and removes both clusters on exit.
#
# Usage:  ./verify-active-active-replication-guide.sh
# Requires: PostgreSQL 16+ binaries on PATH (initdb, pg_ctl, psql).
#           origin=none needs PostgreSQL 16 or later.

set -uo pipefail

PASS=0; FAIL=0
ok()  { printf 'PASS  %s\n' "$1"; PASS=$((PASS+1)); }
bad() { printf 'FAIL  %s\n' "$1"; FAIL=$((FAIL+1)); }
check(){ if [ "$2" = "$3" ]; then ok "$1 (got: $3)"; else bad "$1 (expected: $2, got: $3)"; fi; }

command -v initdb >/dev/null 2>&1 || { echo "initdb not on PATH"; exit 1; }

DIR_A="$(mktemp -d "${TMPDIR:-/tmp}/pgaa.XXXXXX")"
DIR_B="$(mktemp -d "${TMPDIR:-/tmp}/pgab.XXXXXX")"
PORT_A=""; PORT_B=""
cleanup() {
  [ -n "$PORT_A" ] && pg_ctl -D "$DIR_A" -m immediate stop >/dev/null 2>&1
  [ -n "$PORT_B" ] && pg_ctl -D "$DIR_B" -m immediate stop >/dev/null 2>&1
  rm -rf "$DIR_A" "$DIR_B"
}
trap cleanup EXIT

pick() { for p in $(seq "$1" "$(( $1 + 60 ))"); do
           if ! (exec 3<>/dev/tcp/127.0.0.1/$p) 2>/dev/null; then echo "$p"; return; fi
         done; }
PORT_A="$(pick 56110)"; PORT_B="$(pick 56210)"
[ -n "$PORT_A" ] && [ -n "$PORT_B" ] || { echo "no free ports"; exit 1; }

OPTS="-c listen_addresses=127.0.0.1 -c unix_socket_directories='' -c wal_level=logical -c track_commit_timestamp=on"
for pair in "$DIR_A:$PORT_A" "$DIR_B:$PORT_B"; do
  d="${pair%%:*}"; p="${pair##*:}"
  initdb -D "$d" -U postgres --no-sync >/dev/null 2>&1 || { echo "initdb failed"; exit 1; }
  pg_ctl -D "$d" -o "-p $p $OPTS" -l "$d/server.log" start >/dev/null 2>&1
done
sleep 3

A(){ psql -h 127.0.0.1 -p "$PORT_A" -U postgres -X -Atq -c "$1" 2>&1; }
B(){ psql -h 127.0.0.1 -p "$PORT_B" -U postgres -X -Atq -c "$1" 2>&1; }

echo "== environment =="
A "select version()"
echo

echo "== Claim 1: BDR is not part of stock PostgreSQL =="
check "no 'bdr' schema exists"          "0" "$(A "select count(*) from pg_namespace where nspname='bdr'")"
check "no bdr.* functions exist"        "0" "$(A "select count(*) from pg_proc p join pg_namespace n on n.oid=p.pronamespace where n.nspname='bdr'")"
check "'bdr' is not an available extension" "0" "$(A "select count(*) from pg_available_extensions where name='bdr'")"
for fn in bdr_group_create bdr_group_join bdr_part_by_node_names; do
  check "no function named $fn anywhere" "0" "$(A "select count(*) from pg_proc where proname='$fn'")"
done

echo
echo "== Claim 2: track_commit_timestamp is a server setting, not a BDR function argument =="
check "track_commit_timestamp is a GUC in pg_settings" "1" "$(A "select count(*) from pg_settings where name='track_commit_timestamp'")"
check "track_commit_timestamp is enabled here"         "on" "$(A "show track_commit_timestamp")"

echo
echo "== Claim 3: origin=none is accepted, and builds bidirectional replication =="
A "create table t(id int primary key, note text)" >/dev/null
B "create table t(id int primary key, note text)" >/dev/null
A "create publication p for table t" >/dev/null
B "create publication p for table t" >/dev/null
R1=$(A "create subscription s_from_b connection 'host=127.0.0.1 port=$PORT_B user=postgres dbname=postgres' publication p with (origin=none, copy_data=false)")
R2=$(B "create subscription s_from_a connection 'host=127.0.0.1 port=$PORT_A user=postgres dbname=postgres' publication p with (origin=none, copy_data=false)")
case "$R1$R2" in
  *ERROR*) bad "CREATE SUBSCRIPTION ... origin=none failed: $R1 $R2";;
  *)       ok  "CREATE SUBSCRIPTION ... (origin=none) accepted on both nodes";;
esac
check "subscription on A records origin=none" "none" "$(A "select suborigin from pg_subscription where subname='s_from_b'")"
check "subscription on B records origin=none" "none" "$(B "select suborigin from pg_subscription where subname='s_from_a'")"
sleep 4

echo
echo "== Claim 4: writes on both nodes converge, with no infinite echo loop =="
A "insert into t values (1,'written on A')" >/dev/null
B "insert into t values (2,'written on B')" >/dev/null
CONV="f"
for _ in $(seq 1 30); do
  sleep 1
  [ "$(A "select count(*) from t")" = "2" ] && [ "$(B "select count(*) from t")" = "2" ] && { CONV="t"; break; }
done
check "both nodes converged on 2 rows" "t" "$CONV"
check "node A has the row written on B" "1" "$(A "select count(*) from t where note='written on B'")"
check "node B has the row written on A" "1" "$(B "select count(*) from t where note='written on A'")"
sleep 5
check "row count on A is stable, not growing (no echo loop)" "2" "$(A "select count(*) from t")"
check "row count on B is stable, not growing (no echo loop)" "2" "$(B "select count(*) from t")"

echo
echo "== Claim 5: a same-key write on both nodes diverges permanently and stalls replication =="
A "alter subscription s_from_b disable" >/dev/null
B "alter subscription s_from_a disable" >/dev/null
sleep 2
A "insert into t values (99,'A version')" >/dev/null
B "insert into t values (99,'B version')" >/dev/null
A "alter subscription s_from_b enable" >/dev/null
B "alter subscription s_from_a enable" >/dev/null

ERRS="f"
for _ in $(seq 1 40); do
  sleep 1
  EA=$(A "select coalesce(sum(apply_error_count),0) from pg_stat_subscription_stats")
  EB=$(B "select coalesce(sum(apply_error_count),0) from pg_stat_subscription_stats")
  [ "${EA:-0}" -gt 0 ] && [ "${EB:-0}" -gt 0 ] && { ERRS="t"; break; }
done
check "both nodes report apply errors" "t" "$ERRS"
check "node A kept its own version of key 99" "A version" "$(A "select note from t where id=99")"
check "node B kept its own version of key 99" "B version" "$(B "select note from t where id=99")"
ok "the two nodes now permanently disagree about row 99 (no automatic resolution)"

if grep -qi "duplicate key value violates unique constraint" "$DIR_A/server.log"; then
  ok "server log on A shows the unique-constraint conflict"
else
  bad "expected a duplicate-key error in A's server log"
fi

echo
echo "-- a clean write made while the conflict is unresolved --"
A "insert into t values (500,'post-conflict write on A')" >/dev/null
sleep 12
check "the clean row did NOT reach B: replication is stalled" "0" "$(B "select count(*) from t where id=500")"
EA2=$(A "select coalesce(sum(apply_error_count),0) from pg_stat_subscription_stats")
check "apply_error_count is still climbing (retry loop)" "t" "$(A "select ${EA2:-0} > 0")"

echo
echo "== Claim 6: per-conflict statistics are a PostgreSQL 18 feature, absent here =="
NCONF=$(A "select count(*) from information_schema.columns where table_name='pg_stat_subscription_stats' and column_name like 'confl\_%'")
VER=$(A "show server_version_num")
if [ "${VER:-0}" -lt 180000 ]; then
  check "no confl_* columns on this pre-18 server" "0" "$NCONF"
else
  check "confl_* columns present on PostgreSQL 18+" "t" "$(A "select ${NCONF:-0} > 0")"
fi

echo
echo "-------------------------------------------"
printf 'PASSED: %d   FAILED: %d\n' "$PASS" "$FAIL"
[ "$FAIL" -eq 0 ] || exit 1