read_replicate/common.sh
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env bash
# Shared helpers for Capgo read-replica maintenance.
# The current topology has one Google Cloud SQL subscriber from Supabase. Google
# handles replication from that instance to downstream regional replicas.
REPLICA_SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
REPLICA_ENV_FILE="${REPLICA_ENV_FILE:-${REPLICA_SCRIPT_DIR}/../internal/cloudflare/.env.prod}"
# Capgo production defaults for the single Google Cloud SQL eu-2 subscriber.
# DB URLs come from .env.prod; pub/sub/region are fixed so bun scripts need zero exports.
# Live subscription name is capgo_google_eu_2_sub (not capgo_google_eu_2).
# Slot name is discovered from the live subscription when present (fallback: <sub>_slot).
: "${READ_REPLICA_PUBLICATION_NAME:=capgo_google_eu_2_pub}"
: "${READ_REPLICA_SUBSCRIPTION_NAME:=capgo_google_eu_2_sub}"
: "${READ_REPLICA_REGION:=eu_2}"
REPLICA_TABLES=(
"orgs"
"stripe_info"
"org_users"
"apps"
"app_versions"
"channels"
"channel_devices"
"delta_files"
"delta_sessions"
"notifications"
"onboarding_demo_data"
"version_files"
)
REPLICA_PRIORITY_TABLES=(
"delta_sessions"
"orgs"
"stripe_info"
"org_users"
"apps"
"app_versions"
"channels"
"notifications"
"onboarding_demo_data"
)
REPLICA_DEFERRED_TABLES=(
"channel_devices"
"delta_files"
"version_files"
)
ensure_env_file() {
if [[ ! -f "$REPLICA_ENV_FILE" ]]; then
echo "Error: $REPLICA_ENV_FILE not found"
exit 1
fi
}
get_env_value() {
local key="$1"
local line
if [[ -n "${!key:-}" ]]; then
echo "${!key}"
return 0
fi
ensure_env_file
while IFS= read -r line; do
line="${line#"${line%%[![:space:]]*}"}"
[[ -z "$line" ]] && continue
[[ "${line:0:1}" == "#" ]] && continue
if [[ "$line" == "${key}="* ]]; then
echo "${line#*=}"
return 0
fi
done < "$REPLICA_ENV_FILE"
return 1
}
ensure_sslrootcert_system() {
local url="$1"
if [[ "$url" == *"sslmode=require"* ]]; then
printf "%s" "$url"
return 0
fi
if [[ "$url" == *"sslrootcert="* ]]; then
printf "%s" "$url"
return 0
fi
if [[ "$url" == *"?"* ]]; then
printf "%s" "${url}&sslrootcert=system"
else
printf "%s" "${url}?sslrootcert=system"
fi
}
ensure_connect_timeout() {
local url="$1"
local timeout="${2:-10}"
if [[ "$url" == *"connect_timeout="* ]]; then
printf "%s" "$url"
elif [[ "$url" == *"?"* ]]; then
printf "%s&connect_timeout=%s" "$url" "$timeout"
else
printf "%s?connect_timeout=%s" "$url" "$timeout"
fi
}
normalize_cloudsql_ssl() {
local url="$1"
if [[ "$url" == *"sslmode=no-verify"* ]]; then
echo "==> WARNING: PostgreSQL 17 rejects sslmode=no-verify." >&2
echo "==> Downgrading this connection to sslmode=require for this command." >&2
printf "%s" "${url/sslmode=no-verify/sslmode=require}"
return 0
fi
if [[ "$url" == *"sslmode=verify-full"* ]]; then
echo "==> WARNING: Google Cloud SQL URLs using IP hosts usually fail with sslmode=verify-full." >&2
echo "==> Downgrading this connection to sslmode=require for this command." >&2
printf "%s" "${url/sslmode=verify-full/sslmode=require}"
return 0
fi
printf "%s" "$url"
}
extract_host() {
local url="$1"
echo "$url" | sed -E 's|.*@([^/:?]+).*|\1|'
}
sanitize_identifier_part() {
local value="$1"
value="${value//-/_}"
value="${value//[^A-Za-z0-9_]/_}"
printf '%s' "$value" | tr '[:upper:]' '[:lower:]'
}
uri_decode() {
perl -e 'my $s = shift; $s =~ tr/+/ /; $s =~ s/%([0-9A-Fa-f]{2})/chr(hex($1))/eg; print $s;' "$1"
}
libpq_escape_value() {
local value="$1"
local escaped_quote
escaped_quote=$'\\\''
value="${value//\\/\\\\}"
value="${value//\'/$escaped_quote}"
printf "'%s'" "$value"
}
sql_literal_escape() {
local value="$1"
printf "%s" "${value//\'/''}"
}
validate_public_identifier() {
local value="$1"
local label="${2:-identifier}"
if [[ ! "$value" =~ ^[A-Za-z_][A-Za-z0-9_]*$ ]]; then
echo "Error: invalid ${label}: ${value}"
exit 1
fi
}
load_source_db_url() {
local db_url
if ! db_url="$(get_env_value "MAIN_SUPABASE_DB_URL")"; then
echo "Error: MAIN_SUPABASE_DB_URL not set in $REPLICA_ENV_FILE"
exit 1
fi
db_url="${db_url//ssl=false/sslmode=disable}"
db_url="$(normalize_cloudsql_ssl "$db_url")"
printf "%s" "$db_url"
}
build_source_connection_parts() {
local db_url="$1"
local source_password_encoded
SOURCE_USER=$(echo "$db_url" | sed -E 's|postgresql://([^:]+):.*|\1|')
source_password_encoded=$(echo "$db_url" | sed -E 's|postgresql://[^:]+:(.*)@[^@]+$|\1|')
SOURCE_PASSWORD="$(uri_decode "$source_password_encoded")"
_HOST_PORT_DB=$(echo "$db_url" | sed -E 's|.*@([^@]+)$|\1|')
SOURCE_HOST=$(echo "$_HOST_PORT_DB" | sed -E 's|([^:]+):.*|\1|')
SOURCE_PORT=$(echo "$_HOST_PORT_DB" | sed -E 's|[^:]+:([0-9]+)/.*|\1|')
SOURCE_DB=$(echo "$_HOST_PORT_DB" | sed -E 's|[^/]+/([^?]+).*|\1|')
if [[ "$SOURCE_USER" == postgres.* ]]; then
PROJECT_ID=$(echo "$SOURCE_USER" | sed -E 's|postgres\.(.+)|\1|')
SOURCE_HOST="db.${PROJECT_ID}.supabase.co"
SOURCE_PORT="5432"
SOURCE_USER="postgres"
echo "==> Converted Supabase pooler URL to direct connection: $SOURCE_HOST:$SOURCE_PORT"
elif [[ "$SOURCE_PORT" == "6543" ]]; then
SOURCE_PORT="5432"
echo "==> Changed Supabase source port from 6543 to 5432 for direct replication"
fi
SOURCE_SSLMODE='require'
# The Supabase postgres role has statement_timeout=2min. The subscription's
# initial table copy (COPY over this connection) can run longer, and a
# timeout restarts the copy from zero forever, so disable it here.
SOURCE_DB_URL="postgresql://${SOURCE_USER}:${source_password_encoded}@${SOURCE_HOST}:${SOURCE_PORT}/${SOURCE_DB}?sslmode=${SOURCE_SSLMODE}"
SOURCE_CONNECTION_STRING="host=$(libpq_escape_value "$SOURCE_HOST")
port=$(libpq_escape_value "$SOURCE_PORT")
dbname=$(libpq_escape_value "$SOURCE_DB")
user=$(libpq_escape_value "$SOURCE_USER")
password=$(libpq_escape_value "$SOURCE_PASSWORD")
sslmode=$(libpq_escape_value "$SOURCE_SSLMODE")
connect_timeout=10
keepalives=1
keepalives_idle=10
keepalives_interval=5
keepalives_count=3
options='-c statement_timeout=0'"
}
load_source() {
local db_url
db_url="$(load_source_db_url)"
build_source_connection_parts "$db_url"
}
load_replica_target() {
local forced_key="${READ_REPLICA_TARGET_ENV:-}"
local key
local value
local matches_file
local preferred_matches=()
local candidate_keys=(
READ_REPLICATE_GOOGLE_EU1
READ_REPLICA_DB_URL
GOOGLE_READ_REPLICA_DB_URL
GOOGLE_PRIMARY_REPLICA_DB_URL
GOOGLE_REPLICA_DB_URL
GOOGLE_DB_URL
GOOGLE_EU_2
GOOGLE_EU
GOOGLE_PRIMARY
GOOGLE_REPLICA
)
local count
ensure_env_file
if [[ -n "$forced_key" ]]; then
if ! value="$(get_env_value "$forced_key")"; then
echo "Error: READ_REPLICA_TARGET_ENV=$forced_key is not set in $REPLICA_ENV_FILE"
exit 1
fi
if [[ "$value" != postgresql://* ]]; then
echo "Error: $forced_key does not look like a PostgreSQL URL"
exit 1
fi
REPLICA_TARGET_ENV="$forced_key"
REPLICA_TARGET_DB_URL="$(ensure_connect_timeout "$(ensure_sslrootcert_system "$(normalize_cloudsql_ssl "$value")")" 10)"
REPLICA_TARGET_HOST="$(extract_host "$value")"
return 0
fi
for key in "${candidate_keys[@]}"; do
if value="$(get_env_value "$key")" && [[ "$value" == postgresql://* ]]; then
preferred_matches+=("$key")
fi
done
if [[ ${#preferred_matches[@]} -gt 1 ]]; then
echo "Error: multiple preferred Google read-replica database URLs found. Refusing to guess."
echo "Set READ_REPLICA_TARGET_ENV to one of:"
printf ' %s\n' "${preferred_matches[@]}"
exit 1
fi
if [[ ${#preferred_matches[@]} -eq 1 ]]; then
key="${preferred_matches[0]}"
value="$(get_env_value "$key")"
REPLICA_TARGET_ENV="$key"
REPLICA_TARGET_DB_URL="$(ensure_connect_timeout "$(ensure_sslrootcert_system "$(normalize_cloudsql_ssl "$value")")" 10)"
REPLICA_TARGET_HOST="$(extract_host "$value")"
return 0
fi
matches_file="$(mktemp)"
grep -E '^(READ_REPLICATE_[A-Z0-9_]*|GOOGLE_[A-Z0-9_]*|CLOUDSQL_[A-Z0-9_]*)=postgresql://' "$REPLICA_ENV_FILE" \
| cut -d'=' -f1 > "$matches_file" || true
count=$(wc -l < "$matches_file" | tr -d ' ')
if [[ "$count" == "1" ]]; then
key="$(cat "$matches_file")"
rm -f "$matches_file"
value="$(get_env_value "$key")"
REPLICA_TARGET_ENV="$key"
REPLICA_TARGET_DB_URL="$(ensure_connect_timeout "$(ensure_sslrootcert_system "$(normalize_cloudsql_ssl "$value")")" 10)"
REPLICA_TARGET_HOST="$(extract_host "$value")"
return 0
fi
if [[ "$count" -gt 1 ]]; then
echo "Error: multiple Google/Cloud SQL database URLs found. Refusing to guess."
echo "Set READ_REPLICA_TARGET_ENV to one of:"
sed 's/^/ /' "$matches_file"
rm -f "$matches_file"
exit 1
fi
rm -f "$matches_file"
echo "Error: no Google read-replica PostgreSQL URL found in $REPLICA_ENV_FILE"
echo "Set READ_REPLICA_DB_URL or READ_REPLICA_TARGET_ENV."
exit 1
}
replica_region_name() {
local host_part
if [[ -n "${READ_REPLICA_REGION:-}" ]]; then
sanitize_identifier_part "$READ_REPLICA_REGION"
return 0
fi
host_part="${REPLICA_TARGET_HOST%%.*}"
sanitize_identifier_part "$host_part"
}
# psql with statement/lock timeouts disabled (large COPY / INDEX on manifest).
psql_no_timeout() {
local url="$1"
shift
PGOPTIONS="${PGOPTIONS:+${PGOPTIONS} }-c statement_timeout=0 -c lock_timeout=0" \
psql-17 "$url" "$@"
}
# Terminate any walsender holding the slot, then drop it. Retries until inactive.
drop_source_slot_with_retry() {
local slot_name="${1:-${REPLICA_SLOT_NAME}}"
local source_url="${2:-${SOURCE_DB_URL}}"
local max_attempts="${3:-15}"
if [[ -z "$slot_name" || -z "$source_url" ]]; then
echo "Error: drop_source_slot_with_retry requires slot name and source URL"
exit 1
fi
echo "==> Dropping source slot ${slot_name} if present..."
psql-17 "$source_url" -v ON_ERROR_STOP=1 \
-v slot_name="$slot_name" \
-v max_attempts="$max_attempts" <<'SQL'
DO $$
DECLARE
target_slot text := :'slot_name';
max_attempts int := :'max_attempts'::int;
slot record;
attempt int := 0;
BEGIN
LOOP
attempt := attempt + 1;
SELECT slot_name, active, active_pid
INTO slot
FROM pg_replication_slots
WHERE slot_name = target_slot;
IF slot.slot_name IS NULL THEN
RAISE NOTICE 'No source slot named %', target_slot;
RETURN;
END IF;
IF NOT COALESCE(slot.active, false) THEN
PERFORM pg_drop_replication_slot(slot.slot_name);
RAISE NOTICE 'Dropped slot: %', slot.slot_name;
RETURN;
END IF;
RAISE NOTICE 'Slot % still active (pid %, attempt %/%)',
slot.slot_name, slot.active_pid, attempt, max_attempts;
IF slot.active_pid IS NOT NULL THEN
PERFORM pg_terminate_backend(slot.active_pid);
END IF;
IF attempt >= max_attempts THEN
RAISE EXCEPTION 'replication slot % still active for PID % after % attempts',
slot.slot_name, slot.active_pid, attempt;
END IF;
PERFORM pg_sleep(2);
END LOOP;
END
$$;
SQL
}
discover_publication_name() {
local existing
local count
if [[ -n "${READ_REPLICA_PUBLICATION_NAME:-}" ]]; then
printf "%s" "$READ_REPLICA_PUBLICATION_NAME"
return 0
fi
if [[ -n "${REPLICA_TARGET_DB_URL:-}" ]]; then
existing=$(psql-17 "$REPLICA_TARGET_DB_URL" -t -A -c "
SELECT DISTINCT unnest(subpublications)
FROM pg_subscription;
" 2>/dev/null || true)
count=$(printf '%s\n' "$existing" | sed '/^$/d' | wc -l | tr -d ' ')
if [[ "$count" == "1" ]]; then
printf "%s" "$existing"
return 0
fi
fi
existing=$(psql-17 "$SOURCE_DB_URL" -t -A -c "
SELECT pubname
FROM pg_publication
WHERE pubname IN ('capgo_google_eu_2_pub', 'capgo_google_replicate')
ORDER BY CASE pubname
WHEN 'capgo_google_eu_2_pub' THEN 0
ELSE 1
END
LIMIT 1;
" 2>/dev/null || true)
if [[ -n "$existing" ]]; then
printf "%s" "$existing"
else
printf "%s" "capgo_google_eu_2_pub"
fi
}
discover_subscription() {
local default_name="${1:-}"
local rows
local count
local subname
local slotname
local configured_name="${READ_REPLICA_SUBSCRIPTION_NAME:-}"
# If a configured name exists on the target, use it (and its live slot).
if [[ -n "$configured_name" ]]; then
slotname=$(psql-17 "$REPLICA_TARGET_DB_URL" -t -A -c "
SELECT COALESCE(subslotname, '')
FROM pg_subscription
WHERE subname = '${configured_name}'
LIMIT 1;
" 2>/dev/null || true)
if [[ -n "$slotname" || -n "$(psql-17 "$REPLICA_TARGET_DB_URL" -t -A -c "
SELECT 1 FROM pg_subscription WHERE subname = '${configured_name}' LIMIT 1;
" 2>/dev/null || true)" ]]; then
REPLICA_SUBSCRIPTION_NAME="$configured_name"
if [[ -n "$slotname" ]]; then
REPLICA_SLOT_NAME="$slotname"
else
REPLICA_SLOT_NAME="${READ_REPLICA_SLOT_NAME:-${configured_name}_slot}"
fi
return 0
fi
echo "==> Configured subscription ${configured_name} not found on target; discovering live subscription..."
fi
# Prefer a subscription that currently has an apply worker (non-null pid).
rows=$(psql-17 "$REPLICA_TARGET_DB_URL" -t -A -F '|' -c "
SELECT s.subname, COALESCE(s.subslotname, ''), CASE WHEN st.pid IS NULL THEN 0 ELSE 1 END AS has_worker
FROM pg_subscription s
LEFT JOIN pg_stat_subscription st ON st.subname = s.subname AND st.pid IS NOT NULL
ORDER BY has_worker DESC, s.subname;
" 2>/dev/null || true)
count=$(printf '%s\n' "$rows" | sed '/^$/d' | wc -l | tr -d ' ')
if [[ "$count" == "1" ]]; then
subname="${rows%%|*}"
slotname="${rows#*|}"
slotname="${slotname%%|*}"
REPLICA_SUBSCRIPTION_NAME="$subname"
if [[ -n "$slotname" ]]; then
REPLICA_SLOT_NAME="$slotname"
else
REPLICA_SLOT_NAME="${READ_REPLICA_SLOT_NAME:-${subname}_slot}"
fi
return 0
fi
if [[ "$count" -gt 1 ]]; then
# If exactly one has a worker, use that. Otherwise refuse to guess.
local worker_rows
worker_rows=$(printf '%s\n' "$rows" | awk -F'|' '$3 == 1 { print }')
local worker_count
worker_count=$(printf '%s\n' "$worker_rows" | sed '/^$/d' | wc -l | tr -d ' ')
if [[ "$worker_count" == "1" ]]; then
subname="${worker_rows%%|*}"
slotname="${worker_rows#*|}"
slotname="${slotname%%|*}"
echo "==> Multiple subscriptions found; using live worker subscription ${subname}"
REPLICA_SUBSCRIPTION_NAME="$subname"
if [[ -n "$slotname" ]]; then
REPLICA_SLOT_NAME="$slotname"
else
REPLICA_SLOT_NAME="${READ_REPLICA_SLOT_NAME:-${subname}_slot}"
fi
return 0
fi
echo "Error: multiple subscriptions found on $REPLICA_TARGET_ENV. Refusing to guess."
echo "$rows" | sed 's/^/ /'
echo "Set READ_REPLICA_SUBSCRIPTION_NAME and READ_REPLICA_SLOT_NAME."
exit 1
fi
REPLICA_SUBSCRIPTION_NAME="${configured_name:-${default_name:-capgo_google_subscription}}"
REPLICA_SLOT_NAME="${READ_REPLICA_SLOT_NAME:-${REPLICA_SUBSCRIPTION_NAME}_slot}"
}
print_target_summary() {
echo "==> Replica target env: ${REPLICA_TARGET_ENV}"
echo "==> Replica target host: ${REPLICA_TARGET_HOST}"
echo "==> Subscription: ${REPLICA_SUBSCRIPTION_NAME:-unknown}"
echo "==> Slot: ${REPLICA_SLOT_NAME:-unknown}"
}