/* ============================================================================
   MySQL Change Data Capture (CDC) User Setup Script
   ============================================================================

   Overview:
   ---------------------------------------------------------------------------
   This script provides a robust setup for Change Data Capture (CDC) in MySQL by:
   - Validating that a CDC user exists with proper authentication plugin.
   - Granting required privileges to the CDC user.
   - Enabling binary logging for non-cloud-managed instances.

   Key Features:
   ---------------------------------------------------------------------------
   1. Enables binary logging for non-RDS or unmanaged MySQL deployments.
   2. Validates that the CDC user exists with correct authentication plugin.
   3. Validates that specified schemas and tables exist.
   4. Grants essential privileges for CDC functionality:
      - REPLICATION SLAVE
      - REPLICATION CLIENT
      - SELECT on mysql.user
      - SELECT on specified tables or all tables in specified schemas

   Important Notice – MUST BE RUN AS ROOT/ADMIN:
   ---------------------------------------------------------------------------
   ⚠ This script is intended to be executed using the root user or an account with full administrative privileges (e.g., SUPER, ALL PRIVILEGES). ⚠

   While it may be technically possible to run parts of this script under a user with specific privileges (such as CREATE USER or GRANT OPTION), doing so is **not recommended**.
   Only an admin account ensures that all required operations—including global configurations, replication settings, and security requirements—are applied correctly and reliably.

   For consistent, secure, and error-free execution, it is strongly advised to run this script with a root or equivalent administrative user.

   Summary of the required privileges that an admin account has:
   ---------------------------------------------------------------------------
   The executing user must have the following privileges to ensure complete and error-free execution:
   - SUPER for configuring replication and global settings. (Needed only for on-prem MySQL, not for managed services)
   - GRANT OPTION for assigning privileges to the replication user.
   - EXECUTE, CREATE, ALTER, DROP ROUTINE for managing stored procedures and functions.
   - SELECT on mysql.user for validation.
   - SELECT on target databases for data access.
   - REPLICATION SLAVE and REPLICATION CLIENT for granting such permissions.
   - RELOAD for FLUSH PRIVILEGES.

   Preconditions:
   ---------------------------------------------------------------------------
   - Supported MySQL versions: 5.7.44 and >=8.0.
   - CDC user must already exist with appropriate authentication plugin.
   - Target schemas and tables must already exist.

   Usage Instructions:
   ---------------------------------------------------------------------------
   1. Create the CDC user with appropriate authentication plugin before running this script.
        - MySQL 8.x: Use 'caching_sha2_password'
        - MySQL 5.x: Use 'sha256_password'
   2. (Optional) Test with debug procedure first: CALL debug_setup_cdc_user(@username, @tablesList);
   3. Verify that the CDC user has been granted the correct privileges.
   5. Test the CDC connection using the configured user to confirm proper setup.

============================================================================ */

-- =====================================================================
-- STORED PROCEDURE DEFINITIONS
-- =====================================================================

-- ==========================
-- 1. User Setup Procedure
-- ==========================
-- Purpose: Validates CDC user exists with appropriate authentication plugin and grants privileges
-- Parameters:
--   p_username: CDC username (must already exist)
--   p_tablesList: Comma-separated list of tables (e.g., '`db1`.`table1`,`db2`.*,`db3`.`table2`')
--                 Format: `schema`.`table` for specific table, `schema`.* for all tables in schema
-- Error Handling: Validates MySQL version, checks user, schema and table existence, handles plugin mismatches

DROP PROCEDURE IF EXISTS lakeflow_setup_cdc_user;

DELIMITER $$

CREATE PROCEDURE lakeflow_setup_cdc_user(
    IN p_username VARCHAR(255),
    IN p_tablesList MEDIUMTEXT
)
BEGIN
    -- Variable declarations for user validation
    DECLARE v_plugin VARCHAR(64);            -- Current plugin for existing user
    DECLARE v_required_plugin VARCHAR(64);   -- Required plugin based on MySQL version
    DECLARE err_msg VARCHAR(500);            -- Dynamic error message construction

    -- Variables for table/schema parsing loop
    DECLARE v_pos INT DEFAULT 1;             -- Current position in tables list
    DECLARE v_table_count INT;               -- Total number of table entries to process
    DECLARE v_table_entry VARCHAR(256);      -- Current table entry being processed
    DECLARE v_schema_name VARCHAR(128);      -- Extracted schema name
    DECLARE v_table_name VARCHAR(128);       -- Extracted table name
    DECLARE v_is_wildcard BOOLEAN;           -- True if entry is schema.*
    DECLARE v_count INT;                     -- Counter for existence checks

    -- Variables for MySQL version detection
    DECLARE v_mysql_version VARCHAR(50);
    DECLARE v_clean_version VARCHAR(50);
    DECLARE v_mysql_major_version INT;

    -- Detect MySQL version
    SET v_mysql_version = VERSION();
    SET v_clean_version = SUBSTRING_INDEX(v_mysql_version, '-', 1);
    SET v_mysql_major_version = CAST(SUBSTRING_INDEX(v_clean_version, '.', 1) AS UNSIGNED);

    -- Set required authentication plugin based on MySQL version
    IF v_mysql_major_version >= 8 THEN
        SET v_required_plugin := 'caching_sha2_password';
    ELSEIF v_mysql_major_version = 5 THEN
        SET v_required_plugin := 'sha256_password';
ELSE
        -- Unsupported version - abort with detailed error message
        SET err_msg = CONCAT('Cannot create a user for CDC. Unsupported MySQL version: ', v_mysql_version, '. Only MySQL 5.x and >=8.0 are supported.');
        SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = err_msg;
END IF;

    -- =========================
    -- Step 1: Validate CDC user exists
    -- =========================

    -- Detect MySQL version to determine expected authentication plugin
    -- MySQL >=8 uses caching_sha2_password
    -- MySQL 5.7.44 uses sha256_password

    -- Check if user exists and retrieve authentication plugin
SELECT plugin INTO v_plugin
FROM mysql.user
WHERE user = p_username AND host = '%'
    LIMIT 1;

-- User validation logic
IF v_plugin IS NULL THEN
        -- User doesn't exist - throw error
        SET err_msg = CONCAT('!!! ERROR: User ''', p_username, '''@''%'' does NOT exist. Please create the user before running this script.');
        SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = err_msg;
ELSE
        -- User exists - validate plugin compatibility
        IF v_plugin != v_required_plugin THEN
            -- Plugin mismatch detected - this could cause authentication failures
            SET err_msg = CONCAT('User exists with wrong plugin type. Expected: ', v_required_plugin, ', Found: ', v_plugin);
            SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = err_msg;
END IF;
END IF;

    -- =========================
    -- Step 2: Validate tables and schemas
    -- =========================
    -- Verify all specified tables/schemas exist before granting privileges
    -- Format: `schema`.`table` or `schema`.*

    -- Count total number of table entries to process
    SET v_table_count = CASE
    WHEN p_tablesList IS NULL OR p_tablesList = 'NULL' OR p_tablesList = '' THEN 0
    ELSE (LENGTH(p_tablesList) - LENGTH(REPLACE(p_tablesList, ',', '')) + 1)
END;

    -- Loop through each table entry for validation
    SET v_pos = 1;
    WHILE v_pos <= v_table_count DO
        -- Extract current table entry (format: `schema`.`table` or `schema`.*)
        SET v_table_entry = TRIM(SUBSTRING_INDEX(SUBSTRING_INDEX(p_tablesList, ',', v_pos), ',', -1));

        -- Parse schema and table: use `.` between backticked parts as delimiter
        IF v_table_entry LIKE '%`.*' THEN
            -- Wildcard: `schema`.*
            SET v_schema_name = SUBSTRING_INDEX(v_table_entry, '`.', 1);
            SET v_is_wildcard = TRUE;
ELSE
            -- Specific table: `schema`.`table`
            SET v_schema_name = SUBSTRING_INDEX(v_table_entry, '`.`', 1);
            SET v_table_name = SUBSTRING_INDEX(v_table_entry, '`.`', -1);
            SET v_is_wildcard = FALSE;
END IF;

        -- Remove backticks from schema name
        SET v_schema_name = REPLACE(REPLACE(v_schema_name, '`', ''), '`', '');

        -- Remove backticks from table name (if not wildcard)
        IF NOT v_is_wildcard THEN
            SET v_table_name = REPLACE(REPLACE(v_table_name, '`', ''), '`', '');
END IF;

        -- Validate schema exists
SELECT COUNT(*) INTO v_count
FROM INFORMATION_SCHEMA.SCHEMATA
WHERE SCHEMA_NAME = v_schema_name;

IF v_count = 0 THEN
            SET err_msg = CONCAT('!!! ERROR: Catalog ''', v_schema_name, ''' does NOT exist. Aborting grants.');
            SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = err_msg;
END IF;

        -- Validate specific table exists (skip for wildcard entries)
        IF NOT v_is_wildcard THEN
SELECT COUNT(*) INTO v_count
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_SCHEMA = v_schema_name AND TABLE_NAME = v_table_name;

IF v_count = 0 THEN
                SET err_msg = CONCAT('!!! ERROR: Table ''', v_schema_name, '.', v_table_name, ''' does NOT exist. Aborting grants.');
                SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = err_msg;
END IF;
END IF;

        -- Move to next table entry
        SET v_pos = v_pos + 1;
END WHILE;

    -- =========================
    -- Step 3: Grant privileges
    -- =========================
    -- Grant minimal required privileges for CDC operations

    -- Global replication privileges required for CDC
    -- REPLICATION SLAVE: Required to read binary logs
    -- REPLICATION CLIENT: Required to execute SHOW MASTER STATUS and SHOW SLAVE STATUS
    SET @s := CONCAT(
        'GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO ''',
        p_username, '''@''%'';'
    );
PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt;

-- Grant SELECT on mysql.user table
-- Required to read authentication plugin information
-- Limited to plugin and user columns to minimize security exposure
SET @s := CONCAT(
        'GRANT SELECT (plugin,user) ON mysql.user TO ''',
        p_username, '''@''%'';'
    );
PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt;

-- Grant SELECT privileges on specified tables/schemas for CDC data reading
-- Loop through table entries again for privilege grants
SET v_pos = 1;
    WHILE v_pos <= v_table_count DO
        -- Extract current table entry
        SET v_table_entry = TRIM(SUBSTRING_INDEX(SUBSTRING_INDEX(p_tablesList, ',', v_pos), ',', -1));

        -- Grant SELECT on the table or schema
        -- The table entry is already in the correct format: `schema`.`table` or `schema`.*
        SET @s := CONCAT('GRANT SELECT ON ', v_table_entry, ' TO ''', p_username, '''@''%'';');
PREPARE stmt FROM @s;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;

-- Move to next table entry
SET v_pos = v_pos + 1;
END WHILE;

    -- Apply all privilege changes immediately
    FLUSH PRIVILEGES;
END $$
DELIMITER ;

-- ==========================
-- 2. CDC Setup Procedure
-- ==========================
-- Purpose: Configure MySQL binary logging for CDC operations
-- Handles both standalone and managed MySQL services
-- Sets appropriate binlog parameters based on MySQL version

DROP PROCEDURE IF EXISTS lakeflow_cdc_setup;

DELIMITER $$

CREATE PROCEDURE lakeflow_cdc_setup()
BEGIN
    -- Variable for environment detection
    DECLARE superuser_count INT DEFAULT 0;

    -- Variables for MySQL version detection
    DECLARE v_mysql_version VARCHAR(50);
    DECLARE v_clean_version VARCHAR(50);
    DECLARE v_mysql_major_version INT;

    -- Detect MySQL version
    SET v_mysql_version = VERSION();
    SET v_clean_version = SUBSTRING_INDEX(v_mysql_version, '-', 1);
    SET v_mysql_major_version = CAST(SUBSTRING_INDEX(v_clean_version, '.', 1) AS UNSIGNED);

    -- Detect managed MySQL services (RDS, Azure, GCP)
    -- rdsadmin: Amazon RDS
    -- azure_superuser: Azure Database for MySQL
    -- cloudsqlsuperuser: Google Cloud SQL
SELECT COUNT(*) INTO superuser_count
FROM mysql.user
WHERE user IN ('rdsadmin', 'azure_superuser', 'cloudsqlsuperuser');

-- Configuration logic based on service type
IF superuser_count = 0 THEN
        -- Standalone MySQL server - can configure directly

        -- MySQL >=8.0: Use SET PERSIST for permanent configuration
        -- These settings survive server restarts
        IF v_mysql_major_version >= 8 THEN
            -- Binary log format: ROW format captures all column changes
            SET @sql = 'SET PERSIST binlog_format = ''ROW''';
PREPARE stmt FROM @sql;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;

-- Binary log row image: FULL ensures complete before/after images
SET @sql = 'SET PERSIST binlog_row_image = ''FULL''';
PREPARE stmt FROM @sql;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;

SELECT 'Persistent configuration applied successfully.' as info;

-- MySQL 5.x: Use SET GLOBAL, settings lost after restart
-- Manual my.cnf configuration required for persistence
ELSEIF v_mysql_major_version = 5 THEN
            SET @sql = 'SET GLOBAL binlog_format = ''ROW''';
PREPARE stmt FROM @sql;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;

SET @sql = 'SET GLOBAL binlog_row_image = ''FULL''';
PREPARE stmt FROM @sql;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;

SELECT '
Non-persistent configuration applied for MySQL 5.x.

Note:
- These settings will be lost after a MySQL server restart.
- To make them permanent, update the /etc/my.cnf file as shown below or rerun this procedure after each restart.

Steps to persist the configuration:
1. Open the my.cnf file with root privileges:
   sudo vi /etc/my.cnf

2. Under the [mysqld] section, add or update the following lines:

   [mysqld]
   binlog_format=ROW
   binlog_row_image=FULL

3. Save the file and restart MySQL:
   sudo systemctl restart mysqld

This will ensure binary logging and replication settings are retained across restarts.
' AS info;

ELSE
            -- Unsupported MySQL version
            SIGNAL SQLSTATE '45000'
                SET MESSAGE_TEXT = 'Cannot enable binlogging. Unsupported MySQL version detected. Only MySQL 5.x and >=8 are supported.';
END IF;
ELSE
        -- Managed service detected - configuration must be done through service-specific tools
        SIGNAL SQLSTATE '45000'
            SET MESSAGE_TEXT = 'Cloud/managed service detected. Use parameter/server group tools for binlog setup. See documentation for details.';
END IF;
END $$

DELIMITER ;
-- =====================================================================
--  USAGE EXAMPLES
-- =====================================================================
SELECT '--- Example Usage Instructions ---' AS info
UNION ALL
SELECT 'Grant permissions to specific tables: CALL lakeflow_setup_cdc_user(''cdc_user'', ''`mydb1`.`mytable1`, `mydb2`.`mytable2`'')'
UNION ALL
SELECT 'Grant permissions to all tables in specific schemas: CALL lakeflow_setup_cdc_user(''cdc_user'', ''`mydb1`.*, `mydb2`.*'')'
UNION ALL
SELECT 'Grant only replication permissions: CALL lakeflow_setup_cdc_user(''cdc_user'', ''NULL'')'
UNION ALL
SELECT 'Grant permissions to specific table for a schema and all tables in another schemas: CALL lakeflow_setup_cdc_user(''cdc_user'', ''`mydb1`.*, `mydb2`.`table2`'')'
UNION ALL
SELECT 'Setup binary logging for CDC: CALL lakeflow_cdc_setup()';

