api/deploy/sync-replica-data.sh

#!/usr/bin/env bash
# Compare + repair row data on logical-replication subscribers vs publisher.
#
# replicas:sync only ships DDL. If a subscriber missed initial copy (permissions,
# stalled subscription, import before replication caught up), rows stay missing
# until repaired. This script makes publisher the source of truth for core tables.
#
# Usage (on publisher, same deploy/replicas.env as sync-replica-schema.sh):
#   bash deploy/sync-replica-data.sh --check
#   bash deploy/sync-replica-data.sh --repair
#   bash deploy/sync-replica-data.sh --repair-if-needed
set -euo pipefail

API_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
CONFIG_FILE="${REPLICAS_ENV_FILE:-$API_DIR/deploy/replicas.env}"
MODE="${1:---repair-if-needed}"

# Tables that must match publisher for public API (order: parents before children).
SYNC_TABLES=(setting_templates tenants settings plugin_catalog_items plugin_catalog_prices plugin_purchase_requests)
APP_DB_USER="${REPLICA_APP_DB_USER:-montime}"

case "$MODE" in
  --check|--repair|--repair-if-needed) ;;
  *)
    echo "Usage: $0 [--check|--repair|--repair-if-needed]" >&2
    exit 1
    ;;
esac

command -v psql >/dev/null || { echo "psql not found" >&2; exit 1; }
command -v pg_dump >/dev/null || { echo "pg_dump not found" >&2; exit 1; }

if [[ -f "$CONFIG_FILE" ]]; then
  # shellcheck disable=SC1090
  set -a && source "$CONFIG_FILE" && set +a
fi

if [[ -z "${REPLICA_DSNS:-}" ]]; then
  echo "REPLICA_DSNS is empty. Create $CONFIG_FILE from deploy/replicas.env.example" >&2
  exit 1
fi

if [[ -z "${DATABASE_URL:-}" && -f "$API_DIR/.env" ]]; then
  # shellcheck disable=SC1091
  set -a && source "$API_DIR/.env" && set +a
fi

# Prisma DATABASE_URL often has @ in the password (09128081392@Montime).
# libpq URI parsing then treats host as "Montime@127.0.0.1". Convert to keyword form.
uri_to_conninfo() {
  local u="${1%%\?*}"
  if [[ "$u" != postgresql://* && "$u" != postgres://* ]]; then
    echo "$u"
    return 0
  fi
  local rest="${u#*://}"
  if [[ "$rest" =~ ^([^:]+):(.+)@([^:/]+):([0-9]+)/([^/?]+) ]]; then
    printf 'host=%s port=%s dbname=%s user=%s password=%s' \
      "${BASH_REMATCH[3]}" "${BASH_REMATCH[4]}" "${BASH_REMATCH[5]}" \
      "${BASH_REMATCH[1]}" "${BASH_REMATCH[2]}"
    return 0
  fi
  echo "$u"
}

resolve_publisher_dsn() {
  if [[ -n "${PUBLISHER_DSN:-}" ]]; then
    echo "$PUBLISHER_DSN"
    return 0
  fi
  if [[ -n "${PGHOST:-}" && -n "${PGUSER:-}" ]]; then
    echo "host=${PGHOST} port=${PGPORT:-5432} dbname=${PGDATABASE:-superadmin} user=${PGUSER} password=${PGPASSWORD:-}"
    return 0
  fi
  if [[ -n "${DATABASE_URL:-}" ]]; then
    echo "$DATABASE_URL"
    return 0
  fi
  echo "Set DATABASE_URL or PUBLISHER_DSN (publisher Postgres)." >&2
  return 1
}

PUBLISHER="$(resolve_publisher_dsn)" || exit 1
PUBLISHER_MODE="dsn"
PUB_PORT=""
PUB_DB="superadmin"

mapfile -t REPLICAS < <(
  printf '%s\n' "$REPLICA_DSNS" | tr ',' '\n' | sed 's/^[[:space:]]*//;s/[[:space:]]*$//' | grep -v '^$'
)

run_psql() {
  psql "$@" </dev/null
}

libpq_dsn() {
  uri_to_conninfo "$1"
}

label() {
  local dsn="$1"
  if [[ "$dsn" == *"://"* ]]; then
    sed -E 's#^[a-z]+://([^@]*@)?##; s#\?.*$##' <<< "$dsn"
    return
  fi
  local host port db
  host="$(sed -nE 's/.*host=([^ ]+).*/\1/p' <<< "$dsn")"
  port="$(sed -nE 's/.*port=([^ ]+).*/\1/p' <<< "$dsn")"
  db="$(sed -nE 's/.*dbname=([^ ]+).*/\1/p' <<< "$dsn")"
  echo "${host:-?}:${port:-5432}/${db:-?}"
}

as_postgres() {
  if command -v runuser >/dev/null; then
    runuser -u postgres -- "$@"
  else
    sudo -n -u postgres -- "$@"
  fi
}

publisher_port() {
  local src="${PUBLISHER:-}${DATABASE_URL:-}"
  if [[ "$src" =~ :([0-9]{4,5})/ ]]; then
    echo "${BASH_REMATCH[1]}"
    return
  fi
  if [[ "$src" =~ port=([0-9]+) ]]; then
    echo "${BASH_REMATCH[1]}"
    return
  fi
  echo "15432"
}

probe_dsn() {
  local dsn="$1"
  run_psql "$(libpq_dsn "$dsn")" -v ON_ERROR_STOP=1 -At -c "SELECT 1" >/dev/null
}

connect_publisher() {
  PUB_PORT="$(publisher_port)"
  echo "publisher probe port=$PUB_PORT dsn=$(label "$PUBLISHER")"

  if probe_dsn "$PUBLISHER"; then
    PUBLISHER_MODE="dsn"
    echo "publisher: TCP/DSN ok"
    return 0
  fi
  echo "publisher: TCP/DSN failed (cluster is often unix-socket only on this host)"

  local sock="host=/var/run/postgresql port=${PUB_PORT} dbname=${PUB_DB} user=postgres"
  if probe_dsn "$sock"; then
    PUBLISHER="$sock"
    PUBLISHER_MODE="socket"
    echo "publisher: unix socket as postgres ok"
    return 0
  fi

  if as_postgres psql -p "$PUB_PORT" -d "$PUB_DB" -v ON_ERROR_STOP=1 -At -c "SELECT 1" </dev/null >/dev/null; then
    PUBLISHER_MODE="sudo"
    echo "publisher: runuser/sudo -u postgres ok (port $PUB_PORT)"
    return 0
  fi

  echo "Cannot connect to publisher." >&2
  echo "Tried: $(label "$PUBLISHER")  then unix socket  then sudo -u postgres -p $PUB_PORT" >&2
  echo "Debug:" >&2
  run_psql "$(libpq_dsn "$PUBLISHER")" -c "SELECT 1" || true
  as_postgres psql -p "$PUB_PORT" -d "$PUB_DB" -c "SELECT 1" || true
  ss -lntp 2>/dev/null | grep -E "$PUB_PORT|postgres" || netstat -lntp 2>/dev/null | grep postgres || true
  exit 1
}

publisher_psql() {
  if [[ "$PUBLISHER_MODE" == "sudo" ]]; then
    as_postgres psql -p "$PUB_PORT" -d "$PUB_DB" "$@" </dev/null
  else
    run_psql "$(libpq_dsn "$PUBLISHER")" "$@"
  fi
}

publisher_dump() {
  if [[ "$PUBLISHER_MODE" == "sudo" ]]; then
    as_postgres pg_dump -p "$PUB_PORT" -d "$PUB_DB" "$@"
  else
    pg_dump "$(libpq_dsn "$PUBLISHER")" "$@"
  fi
}

connect_publisher

REPL_PUBLISHER_USER="${REPL_PUBLISHER_USER:-repl_user}"

grant_publisher_repl_user() {
  echo "publisher: GRANT SELECT on all tables to ${REPL_PUBLISHER_USER}"
  publisher_psql -v ON_ERROR_STOP=1 -q -c "
    GRANT USAGE ON SCHEMA public TO ${REPL_PUBLISHER_USER};
    GRANT SELECT ON ALL TABLES IN SCHEMA public TO ${REPL_PUBLISHER_USER};
    GRANT SELECT ON ALL SEQUENCES IN SCHEMA public TO ${REPL_PUBLISHER_USER};
    ALTER DEFAULT PRIVILEGES IN SCHEMA public
      GRANT SELECT ON TABLES TO ${REPL_PUBLISHER_USER};
  " || true
}

grant_publisher_repl_user

table_count() {
  local dsn="$1" table="$2"
  if [[ "$dsn" == "$PUBLISHER" ]]; then
    publisher_psql -v ON_ERROR_STOP=1 -At -c \
      "SELECT count(*)::bigint FROM public.\"$table\""
    return
  fi
  run_psql "$(libpq_dsn "$dsn")" -v ON_ERROR_STOP=1 -At -c \
    "SELECT count(*)::bigint FROM public.\"$table\""
}

subscription_names() {
  local dsn="$1"
  run_psql "$(libpq_dsn "$dsn")" -v ON_ERROR_STOP=1 -At -c \
    "SELECT subname FROM pg_subscription ORDER BY subname"
}

grant_app_user() {
  local dsn="$1"
  run_psql "$(libpq_dsn "$dsn")" -v ON_ERROR_STOP=1 -q -c "
    GRANT USAGE ON SCHEMA public TO ${APP_DB_USER};
    GRANT SELECT ON ALL TABLES IN SCHEMA public TO ${APP_DB_USER};
    GRANT SELECT ON ALL SEQUENCES IN SCHEMA public TO ${APP_DB_USER};
    ALTER DEFAULT PRIVILEGES IN SCHEMA public
      GRANT SELECT ON TABLES TO ${APP_DB_USER};
  " 2>/dev/null || true
}

grant_replica_writer() {
  local dsn="$1"
  local user=""
  if [[ "$dsn" == *"://"* ]]; then
    user="$(sed -E 's#^[a-z]+://([^:/]+).*#\1#' <<< "$dsn")"
  else
    user="$(sed -nE 's/.*user=([^ ]+).*/\1/p' <<< "$dsn")"
  fi
  [[ -n "$user" ]] || return 0
  echo "  grant write to $user"
  run_psql "$(libpq_dsn "$dsn")" -v ON_ERROR_STOP=1 -q -c "
    GRANT USAGE ON SCHEMA public TO ${user};
    GRANT SELECT, INSERT, UPDATE, DELETE, TRUNCATE ON ALL TABLES IN SCHEMA public TO ${user};
    GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO ${user};
  " || true
}

copy_table_from_publisher() {
  local replica_dsn="$1" table="$2"
  local dump
  dump="$(mktemp)"
  publisher_dump \
    -t "$table" \
    --data-only \
    --no-owner \
    --no-privileges \
    --column-inserts \
    --disable-triggers \
    >"$dump"
  local bytes
  bytes="$(wc -c <"$dump" | tr -d ' ')"
  echo "  dumped $table (${bytes} bytes)"
  if [[ "${bytes:-0}" -lt 20 ]]; then
    echo "  ! pg_dump produced empty output for $table" >&2
    rm -f "$dump"
    return 1
  fi
  run_psql "$(libpq_dsn "$replica_dsn")" -v ON_ERROR_STOP=1 -f "$dump"
  rm -f "$dump"
}

repair_replica() {
  local replica_dsn="$1"
  local lbl drifted=()
  local table pub replica

  lbl="$(label "$replica_dsn")"
  echo "== repair $lbl"

  mapfile -t subs < <(subscription_names "$replica_dsn")
  if ((${#subs[@]} == 0)); then
    echo "  ! no subscription on replica — skip data repair (create subscription first)" >&2
    return 1
  fi

  for table in "${SYNC_TABLES[@]}"; do
    pub="$(table_count "$PUBLISHER" "$table")"
    replica="$(table_count "$replica_dsn" "$table")"
    if [[ "$pub" != "$replica" ]]; then
      drifted+=("$table")
      echo "  drift $table publisher=$pub replica=$replica"
    fi
  done

  if ((${#drifted[@]} == 0)); then
    echo "  data already in sync"
    grant_app_user "$replica_dsn"
    return 0
  fi

  # TRUNCATE tenants CASCADE wipes settings — recopy the whole core set.
  drifted=("${SYNC_TABLES[@]}")

  for sub in "${subs[@]}"; do
    echo "  disabling subscription $sub"
    run_psql "$(libpq_dsn "$replica_dsn")" -v ON_ERROR_STOP=1 -q \
      -c "ALTER SUBSCRIPTION \"${sub}\" DISABLE"
  done

  grant_replica_writer "$replica_dsn"

  for table in "${drifted[@]}"; do
    echo "  truncating + copying $table"
    run_psql "$(libpq_dsn "$replica_dsn")" -v ON_ERROR_STOP=1 -q \
      -c "TRUNCATE TABLE public.\"$table\" CASCADE"
    copy_table_from_publisher "$replica_dsn" "$table"
    pub="$(table_count "$PUBLISHER" "$table")"
    replica="$(table_count "$replica_dsn" "$table")"
    if [[ "$pub" != "$replica" ]]; then
      echo "  ! repair failed for $table (publisher=$pub replica=$replica)" >&2
      for sub in "${subs[@]}"; do
        run_psql "$(libpq_dsn "$replica_dsn")" -v ON_ERROR_STOP=1 -q \
          -c "ALTER SUBSCRIPTION \"${sub}\" ENABLE" || true
      done
      return 1
    fi
    echo "  ok $table=$replica"
  done

  grant_app_user "$replica_dsn"

  for sub in "${subs[@]}"; do
    echo "  enabling subscription $sub"
    run_psql "$(libpq_dsn "$replica_dsn")" -v ON_ERROR_STOP=1 -q \
      -c "ALTER SUBSCRIPTION \"${sub}\" ENABLE"
    run_psql "$(libpq_dsn "$replica_dsn")" -v ON_ERROR_STOP=1 -q \
      -c "ALTER SUBSCRIPTION \"${sub}\" REFRESH PUBLICATION WITH (copy_data = false)" || true
  done

  echo "  repair done"
}

check_replicas() {
  local needs_repair=0
  local dsn lbl table pub replica

  for dsn in "${REPLICAS[@]}"; do
    lbl="$(label "$dsn")"
    echo "== $lbl"
    run_psql "$(libpq_dsn "$dsn")" -v ON_ERROR_STOP=1 -At -c \
      "SELECT subname || '  received=' || COALESCE(received_lsn::text, '-')
              || '  latest=' || COALESCE(latest_end_lsn::text, '-')
        FROM pg_stat_subscription" 2>/dev/null || echo "  (no pg_stat_subscription)"

    for table in "${SYNC_TABLES[@]}"; do
      pub="$(table_count "$PUBLISHER" "$table")"
      replica="$(table_count "$dsn" "$table")"
      if [[ "$pub" == "$replica" ]]; then
        echo "  $table OK ($pub)"
      else
        echo "  $table DRIFT publisher=$pub replica=$replica"
        needs_repair=1
      fi
    done
  done

  return "$needs_repair"
}

LOCK_FILE="${REPLICA_DATA_LOCK:-$API_DIR/deploy/.replica-data.lock}"
# --repair is operator-initiated; never block on flock (avoids /tmp permission fights with pm2 user).
if [[ "$MODE" == "--repair-if-needed" ]] && command -v flock >/dev/null; then
  mkdir -p "$(dirname "$LOCK_FILE")"
  touch "$LOCK_FILE" 2>/dev/null || true
  chmod 666 "$LOCK_FILE" 2>/dev/null || true
  if exec 9>>"$LOCK_FILE" 2>/dev/null; then
    if ! flock -n 9; then
      echo "replica data sync already running — skip"
      exit 0
    fi
  else
    echo "  ! lock unavailable ($LOCK_FILE) — continuing without flock" >&2
  fi
fi

if [[ "$MODE" == "--check" ]]; then
  check_replicas
  exit $?
fi

if [[ "$MODE" == "--repair-if-needed" ]]; then
  if check_replicas; then
    echo
    echo "All replicas in sync."
    exit 0
  fi
  echo
  echo "Drift detected — repairing..."
fi

failed=0
for dsn in "${REPLICAS[@]}"; do
  repair_replica "$dsn" || failed=1
done

echo
if ((failed == 0)); then
  echo "Replica data sync complete."
  exit 0
fi

echo "Some replicas failed data repair." >&2
exit 1