#!/usr/bin/env bash
# Keep logical-replication subscribers in schema sync with the publisher.
#
# Logical replication ships rows, not DDL. When a migration adds a column on the
# publisher, every subscriber keeps crashing with
#   "logical replication target relation ... is missing replicated columns"
# and the whole subscription stalls until the column exists there too.
#
# _prisma_migrations is itself replicated, so `prisma migrate deploy` on a
# subscriber reports "No pending migrations" even when its tables lag behind.
# Applied migrations are therefore tracked per subscriber in _replica_ddl_applied.
#
# Usage:
#   bash deploy/sync-replica-schema.sh --init      # baseline: mark current migrations as applied
#   bash deploy/sync-replica-schema.sh --dry-run   # list what would run
#   bash deploy/sync-replica-schema.sh             # apply to replicas, then migrate publisher
#   bash deploy/sync-replica-schema.sh --check     # replication + schema drift report
set -euo pipefail

API_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
MIGRATIONS_DIR="$API_DIR/prisma/migrations"
CONFIG_FILE="${REPLICAS_ENV_FILE:-$API_DIR/deploy/replicas.env}"
MODE="${1:-apply}"

case "$MODE" in
  apply|--init|--dry-run|--check) ;;
  *)
    echo "Usage: $0 [--init|--dry-run|--check]" >&2
    exit 1
    ;;
esac

command -v psql >/dev/null || { echo "psql 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

# One DSN per line or comma (NOT spaces — keyword conninfo has spaces).
mapfile -t REPLICAS < <(
  printf '%s\n' "$REPLICA_DSNS" | tr ',' '\n' | sed 's/^[[:space:]]*//;s/[[:space:]]*$//' | grep -v '^$'
)

# psql would otherwise swallow the loop input it is called from
run_psql() {
  psql "$@" </dev/null
}

# host:port/db only — DSNs carry passwords
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:-?}"
}

# every list feeding comm is sorted with the same collation
migration_names() {
  find "$MIGRATIONS_DIR" -mindepth 1 -maxdepth 1 -type d -printf '%f\n' | LC_ALL=C sort
}

ensure_marker_table() {
  run_psql "$1" -v ON_ERROR_STOP=1 -q -c "
    CREATE TABLE IF NOT EXISTS _replica_ddl_applied (
      migration_name text PRIMARY KEY,
      applied_at timestamptz NOT NULL DEFAULT now()
    );"
}

applied_names() {
  run_psql "$1" -v ON_ERROR_STOP=1 -At -c \
    "SELECT migration_name FROM _replica_ddl_applied" | LC_ALL=C sort
}

pending_names() {
  comm -23 <(migration_names) <(applied_names "$1")
}

apply_migration() {
  local dsn="$1" name="$2"
  local sql_file="$MIGRATIONS_DIR/$name/migration.sql"
  [[ -f "$sql_file" ]] || { echo "  ! missing $sql_file" >&2; return 1; }

  local bundle
  bundle="$(mktemp)"
  cat "$sql_file" > "$bundle"
  printf "\nINSERT INTO _replica_ddl_applied (migration_name) VALUES ('%s') ON CONFLICT DO NOTHING;\n" \
    "$name" >> "$bundle"

  run_psql "$dsn" -v ON_ERROR_STOP=1 -q --single-transaction -f "$bundle"
  rm -f "$bundle"
}

column_fingerprint() {
  run_psql "$(libpq_dsn "$1")" -v ON_ERROR_STOP=1 -At -c "
    SELECT table_name || '.' || column_name
      FROM information_schema.columns
     WHERE table_schema = 'public'
       AND table_name NOT IN ('_replica_ddl_applied')" | LC_ALL=C sort
}

# Prisma DATABASE_URL uses ?schema=public; libpq rejects that query param.
libpq_dsn() {
  local u="$1"
  echo "${u%%\?*}"
}

if [[ "$MODE" == "--check" ]]; then
  for dsn in "${REPLICAS[@]}"; do
    echo "== $(label "$dsn")"
    run_psql "$dsn" -v ON_ERROR_STOP=1 -At -c "
      SELECT subname || '  received=' || COALESCE(received_lsn::text, '-')
             || '  latest=' || COALESCE(latest_end_lsn::text, '-')
             || '  last_msg=' || COALESCE(last_msg_receipt_time::text, '-')
        FROM pg_stat_subscription" || true

    if [[ -n "${DATABASE_URL:-}" ]]; then
      drift="$(comm -23 <(column_fingerprint "$DATABASE_URL") <(column_fingerprint "$dsn") || true)"
      if [[ -n "$drift" ]]; then
        echo "  ! columns present on publisher but missing here:"
        sed 's/^/    /' <<< "$drift"
      else
        echo "  columns match publisher"
      fi
    fi
  done
  exit 0
fi

for dsn in "${REPLICAS[@]}"; do
  echo "== $(label "$dsn")"
  ensure_marker_table "$dsn"

  if [[ "$MODE" == "--init" ]]; then
    while read -r name; do
      run_psql "$dsn" -v ON_ERROR_STOP=1 -q -c \
        "INSERT INTO _replica_ddl_applied (migration_name) VALUES ('$name') ON CONFLICT DO NOTHING"
    done < <(migration_names)
    echo "  baseline recorded (no SQL executed)"
    continue
  fi

  pending="$(pending_names "$dsn")"
  if [[ -z "$pending" ]]; then
    echo "  up to date"
    continue
  fi

  while read -r name; do
    if [[ "$MODE" == "--dry-run" ]]; then
      echo "  would apply $name"
    else
      echo "  applying $name"
      apply_migration "$dsn" "$name"
    fi
  done <<< "$pending"
done

if [[ "$MODE" == "--dry-run" || "$MODE" == "--init" ]]; then
  exit 0
fi

echo "== publisher"
cd "$API_DIR"
npx prisma migrate deploy

echo
echo "Replicas apply DDL first, so the publisher can stream the new columns right away."
echo "Verify with: bash deploy/sync-replica-schema.sh --check"