#!/usr/bin/env bash
# Restore a PostgreSQL dump into a throwaway, network-isolated container and check it.
# Usage: restore-test.sh <dump-file> [postgres-major-version] [freshness-check ...]
#   freshness-check: table.column, e.g. public.orders.created_at (newest value is reported)
# Exit code: 0 = pass, 1 = restore or integrity failure.
set -euo pipefail

DUMP=$(realpath "$1"); PGVER=${2:-16}; shift $(( $# >= 2 ? 2 : 1 ))
NAME="restore-test-$$"
PSQL=(docker exec -i "$NAME" psql -U postgres -d restored -v ON_ERROR_STOP=1 -At)

cleanup() { docker rm -f "$NAME" >/dev/null 2>&1 || true; }
trap cleanup EXIT

echo "== file"
echo "size:  $(stat -c %s "$DUMP") bytes"
echo "age:   $(( ( $(date +%s) - $(stat -c %Y "$DUMP") ) / 3600 )) hours"

# --network none: the restored data can't talk to anything. tmpfs: nothing touches disk.
docker run -d --name "$NAME" --network none --tmpfs /var/lib/postgresql/data \
  -e POSTGRES_PASSWORD=unused -v "$DUMP":/backup.dump:ro "postgres:$PGVER" >/dev/null
# The image runs a socket-only temporary server during init; TCP readiness means the real one is up.
until docker exec "$NAME" pg_isready -h 127.0.0.1 -U postgres -q 2>/dev/null; do sleep 1; done
docker exec "$NAME" createdb -U postgres restored

echo "== restore"
start=$(date +%s)
if head -c 5 "$DUMP" | grep -q PGDMP; then
  # Custom-format dump: pg_restore reports every error; --exit-on-error makes any of them fatal.
  docker exec "$NAME" pg_restore -U postgres -d restored --no-owner --exit-on-error /backup.dump
else
  docker exec "$NAME" psql -U postgres -d restored -v ON_ERROR_STOP=1 -q -f /backup.dump >/dev/null
fi
echo "ok in $(( $(date +%s) - start ))s"

echo "== integrity"
# Reading every row forces Postgres to decode every page and TOAST value; corruption errors out here.
"${PSQL[@]}" <<'SQL'
CREATE EXTENSION IF NOT EXISTS amcheck;
DO $$
DECLARE r record; n bigint;
BEGIN
  FOR r IN SELECT c.oid::regclass AS t FROM pg_class c JOIN pg_namespace s ON s.oid = c.relnamespace
           WHERE c.relkind = 'r' AND s.nspname NOT IN ('pg_catalog', 'information_schema') LOOP
    EXECUTE format('SELECT count(*) FROM (SELECT * FROM %s) x', r.t) INTO n;
  END LOOP;
  -- B-tree index structure, and that every heap tuple is present in its index.
  FOR r IN SELECT c.oid::regclass AS i FROM pg_index x JOIN pg_class c ON c.oid = x.indexrelid
           JOIN pg_am a ON a.oid = c.relam JOIN pg_namespace s ON s.oid = c.relnamespace
           WHERE a.amname = 'btree' AND s.nspname NOT IN ('pg_catalog', 'information_schema') LOOP
    PERFORM bt_index_check(r.i, true);
  END LOOP;
END $$;
SQL
echo "all tables read, btree indexes ok"

echo "== row counts"
"${PSQL[@]}" <<'SQL'
SELECT format('%s.%s %s', schemaname, relname,
  (xpath('/row/n/text()', query_to_xml(format('SELECT count(*) AS n FROM %I.%I', schemaname, relname), false, true, '')))[1]::text)
FROM pg_stat_user_tables ORDER BY 1;
SQL

if (( $# )); then
  echo "== freshness"
  for spec in "$@"; do
    col=${spec##*.}; tbl=${spec%.*}
    newest=$("${PSQL[@]}" -c "SELECT max($col) FROM $tbl")
    echo "$tbl.$col newest: ${newest:-<empty>}"
  done
fi
echo "== PASS"