Migração PostgreSQL com Zero Downtime

-- ORIGEM: versão, tamanho e bancos
SELECT version();
SELECT datname, pg_size_pretty(pg_database_size(datname)) AS tamanho
FROM pg_database WHERE datallowconn ORDER BY pg_database_size(datname) DESC;

-- 20 maiores tabelas (dimensiona carga inicial)
SELECT relid::regclass AS tabela, pg_size_pretty(pg_total_relation_size(relid)) AS total, n_live_tup
FROM pg_stat_user_tables ORDER BY pg_total_relation_size(relid) DESC LIMIT 20;

-- Extensões (todas precisam existir no destino, na mesma versão ou superior)
SELECT extname, extversion FROM pg_extension ORDER BY 1;

-- Tabelas SEM chave primária (bloqueiam UPDATE/DELETE na replicação lógica)
SELECT c.oid::regclass AS tabela, c.relreplident
FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE c.relkind IN ('r','p') AND n.nspname NOT IN ('pg_catalog','information_schema')
  AND NOT EXISTS (SELECT 1 FROM pg_constraint k WHERE k.conrelid = c.oid AND k.contype = 'p');

-- Large objects (não replicam logicamente)
SELECT count(*) AS large_objects FROM pg_largeobject_metadata;

-- Unlogged (u), matviews (m) e foreign tables (f) — não replicam logicamente
SELECT c.relkind, c.relpersistence, count(*)
FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE n.nspname NOT IN ('pg_catalog','information_schema') AND c.relkind IN ('r','p','m','f')
GROUP BY 1,2;

-- Collation (PG15+: compara versão registrada com a da biblioteca atual)
SELECT datname, datcollate, datctype, datcollversion,
       pg_database_collation_actual_version(oid) AS versao_atual
FROM pg_database WHERE datallowconn;

-- Transações longas (atrasam criação de slots e snapshots)
SELECT pid, usename, now() - xact_start AS duracao, state, left(query, 80)
FROM pg_stat_activity WHERE xact_start < now() - interval '5 minutes' ORDER BY xact_start;

# Taxa de geração de WAL (amostra de 5 min extrapolada para 1 h)
L1=$(psql -h $OLD -d $DB -Atc "SELECT pg_current_wal_lsn()")
sleep 300
psql -h $OLD -d $DB -Atc "SELECT pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), '$L1') * 12) AS wal_por_hora"

# Versão da glibc (mudança de glibc 2.28+ altera ordenação de texto)
ldd --version | head -1

-- ORIGEM
CREATE ROLE replicator WITH LOGIN REPLICATION PASSWORD 'troque-esta-senha';
-- Necessário para a cópia inicial da replicação lógica
GRANT USAGE ON SCHEMA public TO replicator;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO replicator;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO replicator;

# pg_hba.conf da ORIGEM — física usa a pseudo-base "replication"; lógica conecta no banco
hostssl  replication  replicator  10.0.2.0/24  scram-sha-256
hostssl  app          replicator  10.0.2.0/24  scram-sha-256

psql -h $OLD -U postgres -c "SELECT pg_reload_conf();"

# Teste de conexão física e lógica a partir do DESTINO
psql "host=$OLD user=$REPL replication=true sslmode=require" -c "IDENTIFY_SYSTEM;"
psql "host=$OLD user=$REPL dbname=$DB replication=database sslmode=require" -c "IDENTIFY_SYSTEM;"

-- ORIGEM
ALTER SYSTEM SET wal_level = 'logical';            -- restart
ALTER SYSTEM SET max_replication_slots = 20;       -- restart
ALTER SYSTEM SET max_wal_senders = 20;             -- restart
ALTER SYSTEM SET wal_log_hints = on;               -- restart; necessário para pg_rewind
ALTER SYSTEM SET max_slot_wal_keep_size = '200GB'; -- PG13+; protege o disco
ALTER SYSTEM SET wal_keep_size = '4GB';            -- PG13+
ALTER SYSTEM SET idle_in_transaction_session_timeout = '10min';

-- DESTINO (também precisa de wal_level=logical para a replicação reversa)
ALTER SYSTEM SET wal_level = 'logical';
ALTER SYSTEM SET max_replication_slots = 20;
ALTER SYSTEM SET max_wal_senders = 20;
ALTER SYSTEM SET max_worker_processes = 32;
ALTER SYSTEM SET max_logical_replication_workers = 16;
ALTER SYSTEM SET max_sync_workers_per_subscription = 6;         -- tabelas copiadas em paralelo
ALTER SYSTEM SET max_parallel_apply_workers_per_subscription = 4; -- PG16+

-- Parâmetros pendentes de restart
SELECT name, setting, pending_restart FROM pg_settings WHERE pending_restart;

sudo systemctl restart postgresql@14-main   # origem, na janela combinada
psql -h $OLD -Atc "SHOW wal_level;"          # esperado: logical

; /etc/pgbouncer/pgbouncer.ini
[databases]
app = host=old-primary.internal port=5432 dbname=app

[pgbouncer]
listen_addr = 0.0.0.0
listen_port = 6432
pool_mode = transaction        ; PAUSE em modo session espera clientes desconectarem
admin_users = pgbouncer_admin
query_wait_timeout = 120       ; tempo máximo que um cliente espera na fila (cobre o cutover)
server_reset_query = DISCARD ALL
server_lifetime = 3600

# Ensaio (homologação): medir quanto tempo o PAUSE leva
time psql -h $PGB -p 6432 -U pgbouncer_admin pgbouncer -c "PAUSE app;"
psql -h $PGB -p 6432 -U pgbouncer_admin pgbouncer -c "SHOW POOLS;"
psql -h $PGB -p 6432 -U pgbouncer_admin pgbouncer -c "RESUME app;"

-- ORIGEM: top 50 queries por tempo total (requer pg_stat_statements)
\copy (SELECT queryid, calls, round(total_exec_time) AS total_ms, round(mean_exec_time::numeric,2) AS media_ms, rows, left(query,200) AS query FROM pg_stat_statements ORDER BY total_exec_time DESC LIMIT 50) TO 'baseline_origem.csv' CSV HEADER

-- Throughput de referência
SELECT datname, xact_commit, xact_rollback, tup_inserted, tup_updated, tup_deleted, now()
FROM pg_stat_database WHERE datname = 'app';

# Roles/globais (senhas não são exportáveis em RDS: use --no-role-passwords e redefina)
pg_dumpall -h $OLD -U postgres --roles-only --no-role-passwords -f roles.sql
psql -h $NEW -U postgres -v ON_ERROR_STOP=1 -f roles.sql

# Schema dividido por seção
pg_dump -h $OLD -U postgres -d $DB --schema-only --section=pre-data  -f pre.sql
pg_dump -h $OLD -U postgres -d $DB --schema-only --section=post-data -f post.sql

createdb -h $NEW -U postgres -O app_owner -T template0 -E UTF8 $DB
psql -h $NEW -U postgres -d $DB -v ON_ERROR_STOP=1 -f pre.sql

# split_post.py — separa PK/UNIQUE (antes da carga) do restante (depois da carga)
import re
sql = open('post.sql').read()
stmts = [s.strip() + ';' for s in sql.split(';\n') if s.strip() and not s.strip().startswith('--')]
keys = [s for s in stmts if re.search(r'ADD CONSTRAINT \S+ (PRIMARY KEY|UNIQUE) ', s)]
rest = [s for s in stmts if s not in keys]
open('post_keys.sql', 'w').write('\n\n'.join(keys) + '\n')
open('post_rest.sql', 'w').write('\n\n'.join(rest) + '\n')
print(len(keys), 'chaves |', len(rest), 'demais objetos')

python3 split_post.py
psql -h $NEW -U postgres -d $DB -v ON_ERROR_STOP=1 -f post_keys.sql
# post_rest.sql (índices, FKs, triggers, policies) será aplicado na etapa B4

-- ORIGEM: tabelas sem PK
ALTER TABLE public.log_eventos ADD PRIMARY KEY (id);                 -- ideal (bloqueia a tabela durante a validação)
ALTER TABLE public.auditoria   REPLICA IDENTITY USING INDEX auditoria_uk; -- índice UNIQUE sobre colunas NOT NULL
ALTER TABLE public.staging_raw REPLICA IDENTITY FULL;               -- último recurso: UPDATE/DELETE fazem seq scan no destino

-- Publicação (requer superuser para FOR ALL TABLES)
CREATE PUBLICATION mig_pub FOR ALL TABLES;
-- PG15+: por schema
-- CREATE PUBLICATION mig_pub FOR TABLES IN SCHEMA public, vendas;
-- Tabelas particionadas com estrutura diferente no destino:
-- CREATE PUBLICATION mig_pub FOR ALL TABLES WITH (publish_via_partition_root = true);

SELECT schemaname, tablename FROM pg_publication_tables WHERE pubname = 'mig_pub' ORDER BY 1,2;

-- DESTINO
CREATE SUBSCRIPTION mig_sub
  CONNECTION 'host=old-primary.internal port=5432 dbname=app user=replicator sslmode=require'
  PUBLICATION mig_pub
  WITH (copy_data = true,
        slot_name = 'mig_sub',
        binary = true,            -- PG14+; cópia inicial binária a partir do PG16
        streaming = parallel,     -- PG16+ (em PG14/15 use 'on')
        disable_on_error = true); -- PG15+: para em erro em vez de entrar em loop

-- Progresso por tabela: i=init, d=copiando, f=copiado, s=sincronizando, r=pronto
SELECT srsubstate, count(*) FROM pg_subscription_rel GROUP BY 1;
SELECT srrelid::regclass, srsubstate FROM pg_subscription_rel WHERE srsubstate <> 'r';

-- COPY em andamento (PG14+)
SELECT relid::regclass, bytes_processed, bytes_total, tuples_processed FROM pg_stat_progress_copy;

-- Erros acumulados (PG15+; PG18 inclui contadores de conflito)
SELECT * FROM pg_stat_subscription_stats;

-- ORIGEM: publicação inicial parcial
CREATE PUBLICATION mig_pub FOR TABLE public.clientes, public.produtos;
-- ... após o lote ficar 'r' no destino:
ALTER PUBLICATION mig_pub ADD TABLE public.pedidos, public.itens_pedido;

-- DESTINO: sincroniza apenas as tabelas novas
ALTER SUBSCRIPTION mig_sub REFRESH PUBLICATION WITH (copy_data = true);

# Somente após todas as tabelas em 'r'
psql -h $NEW -d $DB -Atc "SELECT count(*) FROM pg_subscription_rel WHERE srsubstate <> 'r';"   # esperado: 0

# Índices e constraints restantes (em paralelo por arquivo, se desejar)
psql -h $NEW -U postgres -d $DB -v ON_ERROR_STOP=1 -f post_rest.sql

# Estatísticas
vacuumdb -h $NEW -U postgres -d $DB --analyze-only --jobs=8

-- DESTINO: índices inválidos (CONCURRENTLY interrompido)
SELECT indexrelid::regclass FROM pg_index WHERE NOT indisvalid;
-- FKs criadas como NOT VALID podem ser validadas sem lock exclusivo
ALTER TABLE public.itens_pedido VALIDATE CONSTRAINT itens_pedido_pedido_id_fkey;

-- ORIGEM: lag por slot lógico
SELECT s.slot_name, s.active, s.wal_status,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), s.confirmed_flush_lsn)) AS lag,
       r.write_lag, r.flush_lag, r.replay_lag
FROM pg_replication_slots s
LEFT JOIN pg_stat_replication r ON r.pid = s.active_pid
WHERE s.slot_type = 'logical';

-- DESTINO
SELECT subname, pid, received_lsn, latest_end_lsn, latest_end_time FROM pg_stat_subscription;

-- 1. DESTINO primeiro
ALTER TABLE public.clientes ADD COLUMN segmento text;
-- 2. ORIGEM depois
ALTER TABLE public.clientes ADD COLUMN segmento text;

-- Tabela nova: criar nos dois lados e então, no DESTINO:
ALTER SUBSCRIPTION mig_sub REFRESH PUBLICATION;

set -euo pipefail
PGBA="psql -h $PGB -p 6432 -U pgbouncer_admin pgbouncer"

# (1) Congelar escrita: pooler + fencing da role da aplicação na origem
$PGBA -c "PAUSE app;"
psql -h $OLD -U postgres -d $DB -c "ALTER ROLE $APP_ROLE SET default_transaction_read_only = on;"
psql -h $OLD -U postgres -d $DB -c "SELECT pg_terminate_backend(pid) FROM pg_stat_activity
  WHERE datname = '$DB' AND usename = '$APP_ROLE';"

# (2) Esperar o subscriber alcançar o LSN atual da origem
LSN=$(psql -h $OLD -U postgres -d $DB -Atc "SELECT pg_current_wal_lsn()")
until [ "$(psql -h $OLD -U postgres -d $DB -Atc \
  "SELECT confirmed_flush_lsn >= '$LSN' FROM pg_replication_slots WHERE slot_name = 'mig_sub'")" = "t" ]; do
  sleep 1
done
echo "Destino alcançou $LSN"

# (3) Sincronizar sequências (script da etapa 09 · Sequências)
psql -h $OLD -U postgres -d $DB -Atq -f gera_setval.sql > seq_sync.sql
psql -h $NEW -U postgres -d $DB -v ON_ERROR_STOP=1 -f seq_sync.sql

# (4) Encerrar o fluxo origem -> destino (mantém o slot para auditoria)
psql -h $NEW -U postgres -d $DB -c "ALTER SUBSCRIPTION mig_sub DISABLE;"

# (5) Replicação reversa ANTES de liberar escrita (caminho de rollback)
psql -h $NEW -U postgres -d $DB -c "CREATE PUBLICATION rev_pub FOR ALL TABLES;"
psql -h $OLD -U postgres -d $DB -c "CREATE SUBSCRIPTION rev_sub
  CONNECTION 'host=$NEW port=5432 dbname=$DB user=$REPL sslmode=require'
  PUBLICATION rev_pub WITH (copy_data = false, origin = none);"   # origin: PG16+

# (6) Repontar e liberar
sudo sed -i "s/host=$OLD/host=$NEW/" /etc/pgbouncer/pgbouncer.ini
$PGBA -c "RELOAD;" -c "RECONNECT app;" -c "RESUME app;"

-- DESTINO (após o período de rollback, ex.: 7 dias)
DROP SUBSCRIPTION mig_sub;          -- também remove o slot remoto na origem
DROP PUBLICATION rev_pub;

-- ORIGEM antiga
DROP SUBSCRIPTION rev_sub;
DROP PUBLICATION mig_pub;
SELECT slot_name FROM pg_replication_slots;   -- esperado: vazio

# Debian/Ubuntu (repositório PGDG) — instale também as extensões usadas na origem
sudo apt install -y postgresql-18 postgresql-18-postgis-3 postgresql-18-pg-stat-kcache

# Conferir configurações da origem
sudo -u postgres $BIN_OLD/pg_controldata $PGDATA_OLD | grep -i checksum   # 0 = desabilitado
psql -U postgres -Atc "SELECT datname, pg_encoding_to_char(encoding), datcollate, datctype FROM pg_database;"

# Criar o cluster novo espelhando a origem
sudo -u postgres $BIN_NEW/initdb -D $PGDATA_NEW \
  --encoding=UTF8 --locale=pt_BR.UTF-8 \
  --no-data-checksums      # PG18 habilita checksums por padrão; espelhe a origem!

sudo -u postgres bash -c "cd /tmp && $BIN_NEW/pg_upgrade \
  -b $BIN_OLD -B $BIN_NEW -d $PGDATA_OLD -D $PGDATA_NEW \
  -o '-c config_file=/etc/postgresql/14/main/postgresql.conf' \
  -O '-c config_file=/etc/postgresql/18/main/postgresql.conf' \
  --link --jobs=$(nproc) --check"
# Saída esperada: *Clusters are compatible*
# Em caso de falha, os detalhes ficam em $PGDATA_NEW/pg_upgrade_output.d/

# (0) Snapshot/backup: com --link, a origem NÃO pode mais ser iniciada depois que o novo cluster subir
# (1) Parar a aplicação (PgBouncer PAUSE) e o cluster antigo
psql -h $PGB -p 6432 -U pgbouncer_admin pgbouncer -c "PAUSE app;"
sudo systemctl stop postgresql@14-main

# (2) Upgrade
sudo -u postgres bash -c "cd /tmp && $BIN_NEW/pg_upgrade \
  -b $BIN_OLD -B $BIN_NEW -d $PGDATA_OLD -D $PGDATA_NEW \
  -o '-c config_file=/etc/postgresql/14/main/postgresql.conf' \
  -O '-c config_file=/etc/postgresql/18/main/postgresql.conf' \
  --link --jobs=$(nproc)"
#   --clone : XFS/Btrfs com reflink — rápido e a origem continua íntegra
#   --swap  : PG18 — move diretórios; mais rápido que --link em muitos arquivos

# (3) Subir o novo (mesma porta da antiga) e liberar
sudo systemctl start postgresql@18-main
psql -h $PGB -p 6432 -U pgbouncer_admin pgbouncer -c "RESUME app;"

# PG18 preserva as estatísticas do otimizador; só completa as ausentes (ex.: estatísticas estendidas)
sudo -u postgres $BIN_NEW/vacuumdb --all --analyze-in-stages --missing-stats-only --jobs=8
# Versões anteriores ao 18: ANALYZE completo em estágios
# sudo -u postgres $BIN_NEW/vacuumdb --all --analyze-in-stages --jobs=8

# Scripts gerados pelo pg_upgrade (no diretório onde foi executado)
sudo -u postgres psql -f /tmp/update_extensions.sql
# Após o período de rollback:
sudo -u postgres /tmp/delete_old_cluster.sh

# 1. Conferir que o standby aplicou o checkpoint final da origem (valores idênticos)
sudo -u postgres $BIN_OLD/pg_controldata $PGDATA_OLD | grep "Latest checkpoint location"
ssh standby1 "sudo -u postgres $BIN_OLD/pg_controldata $PGDATA_OLD | grep 'Latest checkpoint location'"

# 2. No PRIMÁRIO, a partir do diretório pai dos clusters
cd /var/lib/postgresql
rsync --archive --delete --hard-links --size-only --no-inc-recursive \
  14 18 standby1:/var/lib/postgresql

# 3. No standby: standby.signal + primary_conninfo no cluster 18 e iniciar
ssh standby1 "sudo -u postgres touch /var/lib/postgresql/18/main/standby.signal"

-- ORIGEM
CREATE ROLE debezium WITH LOGIN REPLICATION PASSWORD 'troque-esta-senha';
GRANT USAGE ON SCHEMA public TO debezium;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;
CREATE PUBLICATION dbz_pub FOR ALL TABLES;

-- Heartbeat: faz o slot avançar mesmo quando as tabelas capturadas não recebem escrita
CREATE TABLE public.dbz_heartbeat (id int PRIMARY KEY, ts timestamptz);
INSERT INTO public.dbz_heartbeat VALUES (1, now());
GRANT UPDATE ON public.dbz_heartbeat TO debezium;

-- Sinais: snapshots incrementais sob demanda
CREATE TABLE public.dbz_signal (id varchar(42) PRIMARY KEY, type varchar(32) NOT NULL, data varchar(2048));
GRANT INSERT, SELECT ON public.dbz_signal TO debezium;

{
  "name": "pg-source-app",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "old-primary.internal",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "${file:/opt/secrets/pg.properties:password}",
    "database.dbname": "app",
    "topic.prefix": "app",
    "plugin.name": "pgoutput",
    "slot.name": "dbz_app",
    "publication.name": "dbz_pub",
    "publication.autocreate.mode": "disabled",
    "snapshot.mode": "initial",
    "table.include.list": "public.*",
    "heartbeat.interval.ms": "10000",
    "heartbeat.action.query": "UPDATE public.dbz_heartbeat SET ts = now() WHERE id = 1",
    "signal.data.collection": "public.dbz_signal",
    "decimal.handling.mode": "precise",
    "tombstones.on.delete": "true"
  }
}

curl -sS -X POST -H "Content-Type: application/json" --data @pg-source.json http://connect:8083/connectors
curl -sS http://connect:8083/connectors/pg-source-app/status | jq '.connector.state, .tasks[].state'
# Snapshot incremental de uma tabela (sem bloquear a origem)
psql -h $OLD -d $DB -c "INSERT INTO public.dbz_signal VALUES ('snap-pedidos-1','execute-snapshot',
  '{\"data-collections\": [\"public.pedidos\"], \"type\": \"incremental\"}');"

{
  "name": "pg-sink-app",
  "config": {
    "connector.class": "io.debezium.connector.jdbc.JdbcSinkConnector",
    "topics.regex": "app\\.public\\.(?!dbz_).*",
    "connection.url": "jdbc:postgresql://new-primary.internal:5432/app",
    "connection.username": "app_writer",
    "connection.password": "${file:/opt/secrets/pg.properties:target_password}",
    "insert.mode": "upsert",
    "primary.key.mode": "record_key",
    "delete.enabled": "true",
    "schema.evolution": "none",
    "tasks.max": "4",
    "transforms": "route",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "app\\.public\\.(.*)",
    "transforms.route.replacement": "$1"
  }
}

curl -sS -X POST -H "Content-Type: application/json" --data @pg-sink.json http://connect:8083/connectors
# Lag do consumidor (sink) — deve tender a 0
kafka-consumer-groups.sh --bootstrap-server kafka:9092 --describe --group connect-pg-sink-app