-- Verify Configuration
SELECT name, setting FROM pg_settings
WHERE name IN ('wal_level', 'rds.logical_replication', 'max_replication_slots');

-- SHOW wal_level;
-- SHOW max_replication_slots;
-- SELECT * FROM pg_replication_slots;

-- Create Dedicated CDC User
CREATE USER cdc_user WITH REPLICATION PASSWORD 'secure_password';

-- Grant necessary permissions
GRANT USAGE ON SCHEMA public TO cdc_user;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO cdc_user;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO cdc_user;

-- For RDS, grant additional role
GRANT rds_replication TO cdc_user;

-- Set Up Tables
CREATE TABLE student (id SERIAL PRIMARY KEY, name VARCHAR(100));

-- Configure Tables for CDC to identify row where Primary Key is not available in table, use only when:
-- Your table has no primary key, or
-- You want Debezium to capture before-image data of UPDATEs (i.e., old values), or
-- You want to support UPDATEs/DELETEs on columns not part of a key and still emit useful Kafka records
ALTER TABLE public.student REPLICA IDENTITY FULL;

-- Create replication slot (KafkaConnector can create it automatically)
-- SELECT pg_create_logical_replication_slot('debezium_slot', 'pgoutput');

-- Create publication for specific table (KafkaConnector can create it automatically)
-- CREATE PUBLICATION debezium_publication FOR TABLE public.student;

-- Test CDC Functionality
-- Insert sample data
INSERT INTO student (name) VALUES ('John Doe'), ('Jane Smith');
INSERT INTO student (name) VALUES ('New Student');
UPDATE student SET name = 'Updated Name' WHERE id = 1;
DELETE FROM student WHERE id = 2;

-- Insert many random data into student table if don't want to add one by one
INSERT INTO student (name)
SELECT SUBSTR('ABCDEFGHIJKLMNOPQRSTUVWXYZ', ((random() * (26 - 20 + 1) + 1)::integer), 20)
FROM generate_series(1, 10);

SELECT * FROM student;

-- Monitor and Troubleshoot
-- Check Replication Slots
SELECT slot_name, plugin, slot_type, database, active
FROM pg_replication_slots;

-- Verify replication slot
SELECT * FROM pg_replication_slots;

-- Check Publications
SELECT * FROM pg_publication;
SELECT oid,pubname FROM pg_publication;
SELECT * FROM pg_publication_tables;
SELECT oid,pubname FROM pg_publication;
SELECT * FROM pg_publication_tables WHERE pubname = 'debezium_publication';

-- Cleanup Commands (if needed)
-- If you want to delete a replication slot
SELECT pg_drop_replication_slot('debezium_slot');
-- To delete a publication
DROP PUBLICATION IF EXISTS debezium_publication;