Postgres Imports - WAL Streaming

# Enable logical replication
wal_level = logical

# Set maximum replication slots
max_replication_slots = 10

# Set maximum WAL senders
max_wal_senders = 10

# Enable logical replication workers
max_logical_replication_workers = 10

# Allow replication connections
host replication replication_user source_ip/32 md5

# On systemd systems
sudo systemctl restart postgresql

# On older systems
sudo service postgresql restart

-- Create replication user
CREATE USER replication_user WITH REPLICATION LOGIN;

-- Grant necessary permissions
GRANT CONNECT ON DATABASE your_database TO replication_user;
GRANT USAGE ON SCHEMA public TO replication_user;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO replication_user;

-- Grant permissions for future tables
ALTER DEFAULT PRIVILEGES IN SCHEMA public
    GRANT SELECT ON TABLES TO replication_user;

-- Create publication for all tables
CREATE PUBLICATION planetscale_migration FOR ALL TABLES;

-- Or create publication for specific tables
CREATE PUBLICATION planetscale_migration FOR TABLE
    table1, table2, table3;

-- Verify publication
SELECT * FROM pg_publication;

# Export schema from source
pg_dump -h source-host \
        -p source-port \
        -U source-username \
        -d source-database \
        --schema-only \
        --no-owner \
        --no-privileges \
        -f schema.sql

# Import schema to PlanetScale
psql -h planetscale-host \
     -p planetscale-port \
     -U planetscale-username \
     -d planetscale-database \
     -f schema.sql

-- Run on source database in a dedicated session
-- This creates the slot and returns a snapshot name
SELECT slot_name, snapshot_name, confirmed_flush_lsn
FROM pg_create_logical_replication_slot('planetscale_slot', 'pgoutput');

# Export data using the consistent snapshot from Step 6
pg_dump -h source-host \
        -p source-port \
        -U source-username \
        -d source-database \
        --data-only \
        --no-owner \
        --no-privileges \
        --snapshot=<snapshot_name_from_step_6> \
        -f data.sql

# Import data to PlanetScale
psql -h planetscale-host \
     -p planetscale-port \
     -U planetscale-username \
     -d planetscale-database \
     -f data.sql

-- Create subscription referencing the pre-created slot
CREATE SUBSCRIPTION planetscale_subscription
    CONNECTION 'host=source-host port=source-port
                dbname=source-database user=replication_user
                password=replication-password'
    PUBLICATION planetscale_migration
    WITH (
        copy_data = false,
        create_slot = false,
        slot_name = 'planetscale_slot'
    );

-- Verify subscription is active
SELECT subname, subenabled, subslotname FROM pg_subscription;

-- Confirm replication is streaming
SELECT subname, received_lsn, latest_end_lsn, latest_end_time
FROM pg_stat_subscription;

-- On source database
SELECT * FROM pg_replication_slots;

-- On PlanetScale database
SELECT
    subname,
    received_lsn,
    latest_end_lsn,
    latest_end_time
FROM pg_stat_subscription;

-- Check for subscription errors
SELECT * FROM pg_stat_subscription
WHERE last_msg_failure_time IS NOT NULL;

-- Compare row counts between source and target
-- Run on both databases
SELECT
    schemaname,
    tablename,
    n_tup_ins as estimated_rows
FROM pg_stat_user_tables
ORDER BY schemaname, tablename;

SELECT schemaname, sequencename, last_value + increment_by
  AS next_value
  FROM pg_sequences;

 schemaname |   sequencename   | next_value
------------+------------------+------------
 public     | users_id_seq     |        105
 public     | posts_id_seq     |       1417
 public     | followers_id_seq |       3014

ERROR:  duplicate key value violates unique constraint "XXXX"
DETAIL:  Key (id)=(ZZZZ) already exists.

SELECT 'SELECT setval(''' || schemaname || '.' || sequencename || ''', '
       || (last_value + 10000) || ');' AS query
FROM pg_sequences;

                      query
--------------------------------------------------
 SELECT setval('public.users_id_seq', 10104);
 SELECT setval('public.posts_id_seq', 11416);
 SELECT setval('public.followers_id_seq', 13013);

-- Check that latest data is present
SELECT count(*), max(updated_at) FROM your_main_table;

DROP SUBSCRIPTION planetscale_subscription;

DROP PUBLICATION planetscale_migration;

-- Check existing slots
SELECT * FROM pg_replication_slots;

-- Drop unused slots
SELECT pg_drop_replication_slot('slot_name');

-- Check subscription worker status
SELECT * FROM pg_stat_subscription;

-- Restart subscription if needed
ALTER SUBSCRIPTION planetscale_subscription DISABLE;
ALTER SUBSCRIPTION planetscale_subscription ENABLE;

DROP SUBSCRIPTION planetscale_subscription;

DO $$
DECLARE r RECORD;
BEGIN
  FOR r IN
    SELECT roname FROM pg_replication_origin
    WHERE roname NOT IN (
      SELECT 'pg_' || oid::text FROM pg_subscription
    )
  LOOP
    PERFORM pg_replication_origin_drop(r.roname);
  END LOOP;
END;
$$;