--
-- Copyright [2024 - 2025] Confluent Inc.
--

-- Confluent Oracle XStream CDC connector readiness script (version 1.0)
--
-- This script validates that all database pre-requisites for the Oracle XStream CDC connector as documented in the 'Oracle Database Prerequisites' section of the documentation are met.
--
-- This script does not make any modifications to the database.
--
-- Parameters
-- 1 - Capture database user (default: C##CFLTADMIN)
-- 2 - Connect database user (default: C##CFLTUSER)
-- 3 - Outbound server name (default: XOUT)
-- 4 - Pluggable database (PDB) name in case of multi-tenant database (default: NULL). Ignored when role = DOWNSTREAM.
-- 5 - Role being validated (default: LOCAL): LOCAL | SOURCE | DOWNSTREAM
--     LOCAL:  capture and outbound run on the source database (today's setup).
--     SOURCE:     source database in a downstream-capture topology. Outbound and capture process checks are skipped.
--     DOWNSTREAM: downstream database that hosts capture and outbound in a downstream-capture topology.
--                 PDB and snapshot-privilege checks are skipped (snapshot privs are validated on the source).

SET SERVEROUTPUT ON;
SET VERIFY OFF;
SET LINESIZE 300;

DECLARE
    -- Types
    TYPE varchar2_tt IS TABLE OF VARCHAR2(128);

    -- Constants
    k_db_version NUMBER := DBMS_DB_VERSION.VERSION;
    k_new_line VARCHAR2(10) := CHR(13) || CHR(10);
    k_docs_base_url VARCHAR2(1000) := 'https://docs.confluent.io/cloud/current/connectors/cc-oracle-xstream-cdc-source';
    k_role_local CONSTANT VARCHAR2(30) := 'LOCAL';
    k_role_source CONSTANT VARCHAR2(30) := 'SOURCE';
    k_role_downstream CONSTANT VARCHAR2(30) := 'DOWNSTREAM';

    -- Global variables
    g_capture_user DBA_USERS.USERNAME%TYPE := UPPER(NVL('&1', 'C##CFLTADMIN'));
    g_connect_user DBA_USERS.USERNAME%TYPE := UPPER(NVL('&2', 'C##CFLTUSER'));
    g_outbound_server DBA_XSTREAM_OUTBOUND.SERVER_NAME%TYPE := UPPER(NVL('&3', 'XOUT'));
    g_capture_process DBA_CAPTURE.CAPTURE_NAME%TYPE;
    g_pdb_name V$PDBS.NAME%TYPE := UPPER(NVL('&4', NULL));
    g_role VARCHAR2(30) := UPPER(NVL('&5', k_role_local));
    g_is_multitenant BOOLEAN;
    g_is_rac BOOLEAN;
    g_is_rds BOOLEAN;
    g_required_system_privs varchar2_tt := varchar2_tt('CREATE SESSION');
    g_capture_type VARCHAR2(16);
    g_use_db_link DBA_CAPTURE.USE_DATABASE_LINK%TYPE;
    g_source_database DBA_CAPTURE.SOURCE_DATABASE%TYPE;

-- ### Define all functions and procedures ###

-- Quotes a string
FUNCTION quote(p_input IN VARCHAR2) RETURN VARCHAR2
IS
BEGIN
    RETURN '''' || p_input || '''';
END;

-- Constructs container prefix for messages
FUNCTION msg_container_prefix(p_container IN VARCHAR2 DEFAULT NULL) RETURN VARCHAR2
IS
BEGIN
    IF NOT g_is_multitenant THEN
        RETURN '';
    ELSIF p_container IS NULL THEN
        RETURN '(Container: CDB$ROOT) ';
    ELSE
        RETURN '(Container: ' || p_container || ') ';
    END IF;
END;

-- Logs exception
PROCEDURE log_exception(p_msg IN VARCHAR2, p_errcode IN NUMBER, p_errmsg IN VARCHAR2)
IS
BEGIN
    dbms_output.put_line('ERROR: ' || p_msg);
    dbms_output.put_line('Error code: ' || p_errcode || ', Message: ' || p_errmsg);
END;

-- Checks if current user has enough privilege to run the script
PROCEDURE check_dba_role
IS
    l_result NUMBER;
BEGIN
    SELECT 1
    INTO l_result
    FROM dual
    WHERE USER IN
    (
        SELECT GRANTEE
        FROM DBA_ROLE_PRIVS
        START WITH GRANTED_ROLE = 'DBA'
        CONNECT BY PRIOR GRANTEE = GRANTED_ROLE
    );
EXCEPTION
    WHEN no_data_found THEN
        RAISE_APPLICATION_ERROR(-20001, 'ERROR: Current user does not have DBA role. Please execute the script using a user having DBA role.');
    WHEN others THEN
        log_exception('Failed to check if current user has DBA role.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
        raise;
END;

-- Checks if DB architecture is Multitenant
FUNCTION is_db_multitenant RETURN BOOLEAN
IS
    l_is_cdb V$DATABASE.CDB%TYPE;
BEGIN
    SELECT CDB
    INTO l_is_cdb
    FROM V$DATABASE;

    IF l_is_cdb = 'YES' THEN
        dbms_output.put_line(k_new_line || 'Detected multitenant database architecture.');
        RETURN TRUE;
    ELSE
        dbms_output.put_line(k_new_line || 'Detected non-multitenant database architecture.');
        RETURN FALSE;
    END IF;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to fetch database architecture type.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
        raise;
END;

-- Checks if the provided PDB exists
FUNCTION pdb_exists(p_pdb_name IN VARCHAR2) RETURN BOOLEAN
IS
    l_count NUMBER;
BEGIN
    SELECT COUNT(*)
    INTO l_count
    FROM DBA_PDBS
    WHERE PDB_NAME = p_pdb_name;

    RETURN l_count > 0;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to check if PDB ' || quote(p_pdb_name) || ' exists.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
        raise;
END;

-- Checks if database is RAC (Real Application Clusters) or single instance
FUNCTION is_db_rac RETURN BOOLEAN
IS
    l_is_rac BOOLEAN;
BEGIN
    l_is_rac := DBMS_UTILITY.IS_CLUSTER_DATABASE;
    IF l_is_rac THEN
        dbms_output.put_line(k_new_line || 'Detected RAC (Real Application Clusters) database.');
    ELSE
        dbms_output.put_line(k_new_line || 'Detected single instance database.');
    END IF;

    RETURN l_is_rac;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to check if database is RAC (Real Application Clusters).', SQLCODE, SUBSTR(SQLERRM, 1, 128));
        RETURN FALSE;
END;

-- Checks if database is Amazon RDS
FUNCTION is_db_rds RETURN BOOLEAN
IS
    l_is_rdsadmin_package       NUMBER;
    l_is_rdsadmin_schema        NUMBER;
    l_has_rds_internal_objects  NUMBER;
    l_is_rds                    BOOLEAN := FALSE;
BEGIN
  -- Check if RDSADMIN schema exists
  SELECT COUNT(*) INTO l_is_rdsadmin_schema
  FROM DBA_USERS
  WHERE USERNAME = 'RDSADMIN';

  -- Check if RDSADMIN package exists
  SELECT COUNT(*)
  INTO l_is_rdsadmin_package
  FROM DBA_PROCEDURES
  WHERE OBJECT_TYPE = 'PACKAGE' AND OBJECT_NAME = 'RDSADMIN';

  l_is_rds := l_is_rdsadmin_schema > 0
               AND l_is_rdsadmin_package > 0;

  IF l_is_rds THEN
    DBMS_OUTPUT.PUT_LINE('Detected Amazon RDS for Oracle instance.');
  END IF;

  RETURN l_is_rds;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to check if the database instance is running on Amazon RDS.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
        RETURN FALSE;
END;

-- Checks if XStream is enabled
PROCEDURE validate_xstream
IS
    l_db_xstream V$PARAMETER.VALUE%TYPE;
    l_xstream_docs_url VARCHAR2(1000) := k_docs_base_url || '/prereqs-validation.html#connect-oracle-xstream-cdc-source-prereqs-enable-xstream';
BEGIN
    dbms_output.put_line(k_new_line || 'Validating if XStream is enabled.');
    SELECT MIN(value)
    INTO l_db_xstream
    FROM GV$PARAMETER
    WHERE name = 'enable_goldengate_replication';

    IF l_db_xstream = 'TRUE' THEN
        dbms_output.put_line('SUCCESS: XStream is enabled.');
    ELSE
        dbms_output.put_line('FAILED: XStream is not enabled.');
        dbms_output.put_line('Please refer to the documentation for the procedure to enable XStream: ' || l_xstream_docs_url);
    END IF;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to validate if XStream is enabled.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Resolves the capture process and source database from the outbound server.
-- Silent on miss: validate_outbound_server reports the missing outbound case later.
PROCEDURE resolve_capture_process
IS
BEGIN
    SELECT CAPTURE_NAME, SOURCE_DATABASE
    INTO g_capture_process, g_source_database
    FROM DBA_XSTREAM_OUTBOUND
    WHERE SERVER_NAME = g_outbound_server;
EXCEPTION
    WHEN NO_DATA_FOUND THEN
        NULL;
    WHEN others THEN
        log_exception('Failed to resolve capture process for outbound server ' || quote(g_outbound_server) || '.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Resolves 'DOWNSTREAM_REAL_TIME_MINE' on the capture process and stores the mining mode
PROCEDURE resolve_downstream_real_time_mine
IS
    l_value DBA_CAPTURE_PARAMETERS.VALUE%TYPE;
    l_set_by_user DBA_CAPTURE_PARAMETERS.SET_BY_USER%TYPE;
BEGIN
    IF g_capture_process IS NULL THEN
        dbms_output.put_line(k_new_line || 'INFO: Cannot determine ''DOWNSTREAM_REAL_TIME_MINE''; capture process could not be resolved from the outbound server.');
        RETURN;
    END IF;

    dbms_output.put_line(k_new_line || 'Validating ''DOWNSTREAM_REAL_TIME_MINE'' for capture process ' || quote(g_capture_process) || '.');

    SELECT VALUE, SET_BY_USER
    INTO l_value, l_set_by_user
    FROM DBA_CAPTURE_PARAMETERS
    WHERE CAPTURE_NAME = g_capture_process
    AND PARAMETER = 'DOWNSTREAM_REAL_TIME_MINE';

    IF UPPER(l_value) = 'Y' THEN
        g_capture_type := 'REAL_TIME';
        dbms_output.put_line('INFO: ''DOWNSTREAM_REAL_TIME_MINE'' is set to ''Y'' (set_by_user=' || quote(l_set_by_user) || '). Treating the capture as real-time downstream.');
    ELSE
        g_capture_type := 'ARCHIVED';
        dbms_output.put_line('INFO: ''DOWNSTREAM_REAL_TIME_MINE'' is set to ''N'' (set_by_user=' || quote(l_set_by_user) || '). Treating the capture as archived-log downstream.');
    END IF;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to validate ''DOWNSTREAM_REAL_TIME_MINE'' for capture process ' || quote(g_capture_process) || '.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Checks if log mode is set to 'ARCHIVELOG'
PROCEDURE validate_log_mode
IS
    l_db_log_mode V$DATABASE.LOG_MODE%TYPE;
    l_log_mode_docs_url VARCHAR2(1000) := k_docs_base_url || '/prereqs-validation.html#configure-archivelog-mode';
BEGIN
    dbms_output.put_line(k_new_line || 'Validating database log mode.');
    SELECT LOG_MODE
    INTO l_db_log_mode
    FROM V$DATABASE;

    IF l_db_log_mode = 'ARCHIVELOG' THEN
        dbms_output.put_line('SUCCESS: Database is set to ''ARCHIVELOG'' mode.');
    ELSIF g_role = k_role_downstream AND g_capture_type = 'ARCHIVED' THEN
        dbms_output.put_line('INFO: Database is not set to ''ARCHIVELOG'' mode (current: ' || quote(l_db_log_mode) || '). Acceptable because the capture is archived-log downstream.');
    ELSE
        dbms_output.put_line('FAILED: Database is not set to ''ARCHIVELOG'' mode. Current mode: ' || quote(l_db_log_mode) || '.');
        dbms_output.put_line('Please refer to the documentation for the procedure to set the database to ARCHIVELOG mode: ' || l_log_mode_docs_url);
    END IF;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to validate database log mode.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Checks if supplemental logs is enabled or not
PROCEDURE validate_supplemental_logging
IS
    l_supp_log_min V$DATABASE.SUPPLEMENTAL_LOG_DATA_MIN%TYPE;
    l_supp_log_all V$DATABASE.SUPPLEMENTAL_LOG_DATA_ALL%TYPE;
    l_supp_docs_url VARCHAR2(1000) := k_docs_base_url || '/prereqs-validation.html#configure-supplemental-logging';
BEGIN
    dbms_output.put_line(k_new_line || 'Validating supplemental logging.');
    SELECT SUPPLEMENTAL_LOG_DATA_MIN, SUPPLEMENTAL_LOG_DATA_ALL
    INTO l_supp_log_min, l_supp_log_all
    FROM V$DATABASE;

    IF l_supp_log_all = 'YES' THEN
        dbms_output.put_line('WARN: ''ALL COLUMN'' supplemental logging is enabled at the database level. Confluent recommends enabling minimal supplemental logging at the database level and ''ALL COLUMN'' supplemental logging for the specific tables that require change data capture.');
        dbms_output.put_line('Please refer to the documentation for the procedure to enable supplemental logging: ' || l_supp_docs_url);
    ELSIF l_supp_log_min = 'NO' THEN
        dbms_output.put_line('FAILED: Minimal supplemental logging is not enabled and is required for the connector to function. Confluent recommends enabling minimal supplemental logging at the database level and ''ALL COLUMN'' supplemental logging for the specific tables that require change data capture.');
        dbms_output.put_line('Please refer to the documentation for the procedure to enable supplemental logging: ' || l_supp_docs_url);
    ELSE
        dbms_output.put_line('INFO: Please ensure that ''ALL COLUMN'' supplemental logging is enabled for the specific tables that require change data capture.');
        dbms_output.put_line('Please refer to the documentation for the procedure to enable supplemental logging: ' || l_supp_docs_url);
    END IF;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to validate supplemental logging.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Checks the archive log retention. Can only get it for Managed RDS at the moment.
PROCEDURE validate_archive_log_retention
IS
    l_retention_hours NUMBER;
    l_rds_docs_url varchar2(1000) := k_docs_base_url || '/prereqs-validation.html#configure-archivelog-mode-rds';
BEGIN
    dbms_output.put_line(k_new_line || 'Validating archive log retention period.');

    IF g_is_rds THEN
        -- Using execute immediate otherwise it will fail to compile for non RDS DBs
        EXECUTE IMMEDIATE 'SELECT value FROM rdsadmin.rds_configuration WHERE name =''archivelog retention hours''' INTO l_retention_hours;

        IF l_retention_hours < 24 THEN
            dbms_output.put_line('WARN: Archive log retention is set to ' || quote(l_retention_hours) || ' hours. Confluent recommends setting the archive log retention to at least 24 hours.');
            dbms_output.put_line('Please refer to the documentation for steps to configure archive log retention: ' || l_rds_docs_url);
        ELSE
            dbms_output.put_line('SUCCESS: ARCHIVELOG mode is enabled, and archive log retention is set to ' || quote(l_retention_hours) || ' hours.');
        END IF;
    ELSE
        dbms_output.put_line('INFO: Confluent recommends that archive logs be retained for at least 24 hours.');
    END IF;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to fetch archive log retention.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;


-- Checks if 'GLOBAL_NAMES' is set to TRUE on every instance and reports the database GLOBAL_NAME
PROCEDURE validate_global_names
IS
    l_global_names V$PARAMETER.VALUE%TYPE;
    l_global_name GLOBAL_NAME.GLOBAL_NAME%TYPE;
    l_docs_url VARCHAR2(1000) := k_docs_base_url || '/prereqs-validation.html#downstream-capture-database-prerequisites';
BEGIN
    dbms_output.put_line(k_new_line || 'Validating ''GLOBAL_NAMES'' parameter and database GLOBAL_NAME.');
    SELECT MIN(value)
    INTO l_global_names
    FROM GV$PARAMETER
    WHERE name = 'global_names';

    SELECT global_name
    INTO l_global_name
    FROM GLOBAL_NAME;

    IF UPPER(l_global_names) = 'TRUE' THEN
        dbms_output.put_line('SUCCESS: ''GLOBAL_NAMES'' is set to ''TRUE''. Database GLOBAL_NAME is ' || quote(l_global_name) || '.');
    ELSE
        dbms_output.put_line('WARN: ''GLOBAL_NAMES'' is set to ' || quote(l_global_names) || '. Confluent recommends setting it to ''TRUE'' so the database GLOBAL_NAME identifies the source consistently across instances. Database GLOBAL_NAME is ' || quote(l_global_name) || '.');
        dbms_output.put_line('Please refer to the documentation for the steps to set this parameter: ' || l_docs_url);
    END IF;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to validate ''GLOBAL_NAMES''.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Checks if 'REMOTE_LOGIN_PASSWORDFILE' is set to EXCLUSIVE or SHARED
PROCEDURE validate_remote_login_passwordfile
IS
    l_bad_count NUMBER;
    l_docs_url VARCHAR2(1000) := k_docs_base_url || '/prereqs-validation.html#step-1-configure-database-prerequisites-on-both-databases';
BEGIN
    dbms_output.put_line(k_new_line || 'Validating ''REMOTE_LOGIN_PASSWORDFILE'' parameter.');

    SELECT COUNT(*)
    INTO l_bad_count
    FROM GV$PARAMETER
    WHERE name = 'remote_login_passwordfile'
    AND UPPER(value) NOT IN ('EXCLUSIVE', 'SHARED');

    IF l_bad_count = 0 THEN
        dbms_output.put_line('SUCCESS: ''REMOTE_LOGIN_PASSWORDFILE'' is set to ''EXCLUSIVE'' or ''SHARED''.');
    ELSE
        dbms_output.put_line('WARN: ''REMOTE_LOGIN_PASSWORDFILE'' is not set to ''EXCLUSIVE'' or ''SHARED''. Redo transport authenticates via the password file; set it to ''EXCLUSIVE'' or ''SHARED'' so redo can ship between the source and the downstream capture database. On a RAC database, ensure every instance has this parameter set to ''EXCLUSIVE'' or ''SHARED''.');
        dbms_output.put_line('Please refer to the documentation for the steps to set this parameter: ' || l_docs_url);
    END IF;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to validate ''REMOTE_LOGIN_PASSWORDFILE''.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Checks if standby redo log groups are present (required for real-time downstream capture)
PROCEDURE validate_standby_logfile
IS
    l_count NUMBER;
    l_min_bytes NUMBER;
    l_max_bytes NUMBER;
    l_docs_url VARCHAR2(1000) := k_docs_base_url || '/prereqs-validation.html#step-2-set-up-the-downstream-database';
BEGIN
    dbms_output.put_line(k_new_line || 'Validating standby redo logs.');

    SELECT COUNT(*), MIN(BYTES), MAX(BYTES)
    INTO l_count, l_min_bytes, l_max_bytes
    FROM V$STANDBY_LOG;

    IF l_count = 0 THEN
        dbms_output.put_line('FAILED: No standby redo log groups are configured. Real-time downstream capture requires standby redo logs.');
        dbms_output.put_line('Please refer to the documentation for the steps to add standby redo log groups: ' || l_docs_url);
    ELSE
        dbms_output.put_line('SUCCESS: Found ' || l_count || ' standby redo log group(s). Sizes range from ' || ROUND(l_min_bytes / 1024 / 1024) || ' MB to ' || ROUND(l_max_bytes / 1024 / 1024) || ' MB.');
        dbms_output.put_line('INFO: Each standby redo log file must be at least as large as the largest online redo log file on the source database, and the downstream must have at least (groups_per_thread + 1) standby redo log groups per source thread. Verify both manually against the source.');
    END IF;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to validate standby redo logs.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Populates required system privileges based on architecture
PROCEDURE init_required_sys_privs
IS
BEGIN
    IF g_is_multitenant AND g_pdb_name IS NOT NULL THEN
        g_required_system_privs := g_required_system_privs MULTISET UNION varchar2_tt('SET CONTAINER');
    END IF;
END;

-- Checks if the provided database user exists
FUNCTION user_exists(p_user IN VARCHAR2) RETURN BOOLEAN
IS
    l_count NUMBER;
    l_docs_url VARCHAR2(1000) := k_docs_base_url || '/prereqs-validation.html#configure-database-users';
BEGIN
    dbms_output.put_line(k_new_line || 'Validating if the database user ' || quote(p_user) || ' exists.');

    IF g_is_multitenant THEN
        SELECT COUNT(*)
        INTO l_count
        FROM DBA_USERS
        WHERE USERNAME = p_user
        AND COMMON = 'YES';

        IF l_count > 0 THEN
            dbms_output.put_line('SUCCESS: The database user ' || quote(p_user) || ' exists and is a common user.');
            RETURN TRUE;
        ELSE
            dbms_output.put_line('FAILED: The database user ' || quote(p_user) || ' does not exist or is not a common user.');
            dbms_output.put_line('Please refer to the documentation for the steps to create and configure the database user: ' || l_docs_url);
            RETURN FALSE;
        END IF;
    ELSE
        SELECT COUNT(*)
        INTO l_count
        FROM DBA_USERS
        WHERE USERNAME = p_user;

        IF l_count > 0 THEN
            dbms_output.put_line('SUCCESS: The database user ' || quote(p_user) || ' exists.');
            RETURN TRUE;
        ELSE
            dbms_output.put_line('FAILED: The database user ' || quote(p_user) || ' does not exist.');
            dbms_output.put_line('Please refer to the documentation for the steps to create and configure the database user: ' || l_docs_url);
            RETURN FALSE;
        END IF;
    END IF;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to validate if the database user ' || quote(p_user) || ' exists.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
        raise;
END;

-- Checks if a user has a specific system privilege in the given container
FUNCTION has_system_privilege(p_privilege IN VARCHAR2, p_user IN VARCHAR2, p_container IN VARCHAR2 DEFAULT NULL) RETURN BOOLEAN
IS
    l_count NUMBER;
BEGIN
    IF p_container IS NULL THEN
        SELECT COUNT(*)
        INTO l_count
        FROM DBA_SYS_PRIVS dsp
        WHERE dsp.PRIVILEGE = p_privilege
        AND (
            GRANTEE = p_user OR GRANTEE IN (
                SELECT GRANTED_ROLE
                FROM DBA_ROLE_PRIVS
                CONNECT BY PRIOR GRANTED_ROLE = GRANTEE
                START WITH GRANTEE = p_user
            )
        );
    ELSE
        SELECT COUNT(*)
        INTO l_count
        FROM CDB_SYS_PRIVS
        WHERE PRIVILEGE = p_privilege
        AND CON_ID = CON_NAME_TO_ID(p_container)
        AND (
            GRANTEE = p_user OR GRANTEE IN (
                SELECT GRANTED_ROLE
                FROM CDB_ROLE_PRIVS
                WHERE CON_ID = CON_NAME_TO_ID(p_container)
                CONNECT BY PRIOR GRANTED_ROLE = GRANTEE AND PRIOR CON_ID = CON_ID
                START WITH GRANTEE = p_user
            )
        );
    END IF;

    RETURN l_count > 0;
END;

-- Checks if user has all the required system privileges in place
PROCEDURE validate_system_privileges(p_user IN VARCHAR2)
IS
    l_docs_url VARCHAR2(1000) := k_docs_base_url || '/prereqs-validation.html#configure-database-users';
    l_containers varchar2_tt := varchar2_tt(NULL);
BEGIN
    l_containers(1) := NULL;
    IF g_is_multitenant AND g_pdb_name IS NOT NULL THEN
        l_containers.EXTEND;
        l_containers(2) := g_pdb_name;
    END IF;

    FOR c_idx IN 1..l_containers.COUNT LOOP
        FOR p_idx IN 1..g_required_system_privs.COUNT LOOP
            dbms_output.put_line(k_new_line || msg_container_prefix(l_containers(c_idx)) || 'Validating ' || quote(g_required_system_privs(p_idx)) || ' privilege for database user ' || quote(p_user) || '.');
            IF has_system_privilege(g_required_system_privs(p_idx), p_user, l_containers(c_idx)) THEN
                dbms_output.put_line('SUCCESS: ' || msg_container_prefix(l_containers(c_idx)) || 'Database user ' || quote(p_user) || ' has the required system privilege ' || quote(g_required_system_privs(p_idx)) || '.');
            ELSE
                dbms_output.put_line('FAILED: ' || msg_container_prefix(l_containers(c_idx)) || 'Database user ' || quote(p_user) || ' is missing the required system privilege ' || quote(g_required_system_privs(p_idx)) || '.');
                dbms_output.put_line('Please refer to the documentation for steps to grant this access: ' || l_docs_url);
            END IF;
        END LOOP;
    END LOOP;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to validate system privileges for database user ' || quote(p_user) || '.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Checks if capture user has been granted privileges to be an XStream administrator
PROCEDURE validate_xstream_admin
IS
    l_count NUMBER;
    l_xstream_admin_docs_url VARCHAR2(1000) := k_docs_base_url || '/prereqs-validation.html#configure-database-users';
BEGIN
    dbms_output.put_line(k_new_line || 'Validating XStream administrator privileges for database user ' || quote(g_capture_user) || '.');
    SELECT COUNT(*)
    INTO l_count
    FROM DBA_XSTREAM_ADMINISTRATOR
    WHERE USERNAME = g_capture_user
    AND PRIVILEGE_TYPE = 'CAPTURE'
    AND GRANT_SELECT_PRIVILEGES = 'YES';

    IF l_count > 0 THEN
        dbms_output.put_line('SUCCESS: Database user ' || quote(g_capture_user) || ' has the required XStream administrator privileges.');
    ELSE
        dbms_output.put_line('FAILED: Database user ' || quote(g_capture_user) || ' does not have the required XStream administrator privileges.');
        dbms_output.put_line('Please refer to the documentation for steps to grant XStream administrator privileges: ' || l_xstream_admin_docs_url);
    END IF;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to validate XStream administrator privileges for database user ' || quote(g_capture_user) || '.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Reads 'USE_DATABASE_LINK' on the capture process and stores it in g_use_db_link
PROCEDURE resolve_use_database_link
IS
    l_value DBA_CAPTURE.USE_DATABASE_LINK%TYPE;
BEGIN
    IF g_capture_process IS NULL THEN
        dbms_output.put_line(k_new_line || 'INFO: Cannot determine ''USE_DATABASE_LINK''; capture process could not be resolved from the outbound server.');
        RETURN;
    END IF;

    dbms_output.put_line(k_new_line || 'Validating ''USE_DATABASE_LINK'' for capture process ' || quote(g_capture_process) || '.');

    SELECT USE_DATABASE_LINK
    INTO l_value
    FROM DBA_CAPTURE
    WHERE CAPTURE_NAME = g_capture_process;

    g_use_db_link := UPPER(l_value);
    IF g_use_db_link = 'YES' THEN
        dbms_output.put_line('INFO: ''USE_DATABASE_LINK'' is ''YES''. Database link grant and link presence checks will run.');
    ELSE
        dbms_output.put_line('INFO: ''USE_DATABASE_LINK'' is ' || quote(l_value) || '. Database link grant and link presence checks will be skipped.');
    END IF;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to read ''USE_DATABASE_LINK'' for capture process ' || quote(g_capture_process) || '.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Checks if 'DBMS_XSTREAM_AUTH.GRANT_REMOTE_ADMIN_ACCESS' has been granted to the capture user
PROCEDURE validate_access_from_remote
IS
    l_access_from_remote DBA_STREAMS_ADMINISTRATOR.ACCESS_FROM_REMOTE%TYPE;
    l_docs_url VARCHAR2(1000) := k_docs_base_url || '/prereqs-validation.html#step-3-set-up-the-source-database';
BEGIN
    dbms_output.put_line(k_new_line || 'Validating ''DBMS_XSTREAM_AUTH.GRANT_REMOTE_ADMIN_ACCESS'' for database user ' || quote(g_capture_user) || '.');

    SELECT ACCESS_FROM_REMOTE
    INTO l_access_from_remote
    FROM DBA_STREAMS_ADMINISTRATOR
    WHERE USERNAME = g_capture_user;

    IF l_access_from_remote = 'YES' THEN
        dbms_output.put_line('SUCCESS: Database user ' || quote(g_capture_user) || ' has been granted ''DBMS_XSTREAM_AUTH.GRANT_REMOTE_ADMIN_ACCESS''.');
    ELSE
        dbms_output.put_line('WARN: Database user ' || quote(g_capture_user) || ' has not been granted ''DBMS_XSTREAM_AUTH.GRANT_REMOTE_ADMIN_ACCESS''. Link-based downstream capture setup will not work until this grant is in place.');
        dbms_output.put_line('Please refer to the documentation for the steps to grant remote admin access: ' || l_docs_url);
    END IF;
EXCEPTION
    WHEN NO_DATA_FOUND THEN
        dbms_output.put_line('WARN: Database user ' || quote(g_capture_user) || ' has not been granted ''DBMS_XSTREAM_AUTH.GRANT_REMOTE_ADMIN_ACCESS''. Link-based downstream capture setup will not work until this grant is in place.');
        dbms_output.put_line('Please refer to the documentation for the steps to grant remote admin access: ' || l_docs_url);
    WHEN others THEN
        log_exception('Failed to validate ''DBMS_XSTREAM_AUTH.GRANT_REMOTE_ADMIN_ACCESS'' for database user ' || quote(g_capture_user) || '.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Checks if capture user has 'CREATE DATABASE LINK' privilege
PROCEDURE validate_create_database_link_grant
IS
    l_container VARCHAR2(128) := CASE WHEN g_pdb_name IS NOT NULL THEN g_pdb_name ELSE NULL END;
    l_docs_url VARCHAR2(1000) := k_docs_base_url || '/prereqs-validation.html#step-4-create-xstream-out-on-the-downstream';
BEGIN
    IF g_use_db_link IS NULL THEN
        dbms_output.put_line(k_new_line || 'INFO: Skipping ''CREATE DATABASE LINK'' grant check; ''USE_DATABASE_LINK'' could not be determined.');
        RETURN;
    ELSIF g_use_db_link = 'NO' THEN
        dbms_output.put_line(k_new_line || 'Skipping ''CREATE DATABASE LINK'' grant check; capture process has ''USE_DATABASE_LINK'' = ''NO''.');
        RETURN;
    END IF;

    dbms_output.put_line(k_new_line || 'Validating ''CREATE DATABASE LINK'' privilege for database user ' || quote(g_capture_user) || '.');

    IF has_system_privilege('CREATE DATABASE LINK', g_capture_user, l_container) THEN
        dbms_output.put_line('SUCCESS: ' || msg_container_prefix(l_container) || 'Database user ' || quote(g_capture_user) || ' has the ''CREATE DATABASE LINK'' privilege.');
    ELSE
        dbms_output.put_line('FAILED: ' || msg_container_prefix(l_container) || 'Database user ' || quote(g_capture_user) || ' is missing the ''CREATE DATABASE LINK'' privilege required to create the link the capture process uses.');
        dbms_output.put_line('Please refer to the documentation for the steps to grant this access: ' || l_docs_url);
    END IF;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to validate ''CREATE DATABASE LINK'' privilege for database user ' || quote(g_capture_user) || '.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Probes DBA_DB_LINKS for public and private database links matching the source
PROCEDURE validate_source_db_links
IS
    l_match_pattern DBA_DB_LINKS.DB_LINK%TYPE;
    l_public_count NUMBER := 0;
    l_private_count NUMBER := 0;
    l_docs_url VARCHAR2(1000) := k_docs_base_url || '/prereqs-validation.html#step-4-create-xstream-out-on-the-downstream';
BEGIN
    IF g_use_db_link IS NULL THEN
        dbms_output.put_line(k_new_line || 'INFO: Skipping source database link presence check; ''USE_DATABASE_LINK'' could not be determined.');
        RETURN;
    ELSIF g_use_db_link = 'NO' THEN
        dbms_output.put_line(k_new_line || 'Skipping source database link presence check; capture process has ''USE_DATABASE_LINK'' = ''NO''.');
        RETURN;
    END IF;

    IF g_source_database IS NULL THEN
        dbms_output.put_line(k_new_line || 'INFO: Skipping source database link presence check; ''DBA_CAPTURE.SOURCE_DATABASE'' is empty.');
        RETURN;
    END IF;

    dbms_output.put_line(k_new_line || 'Validating source database links matching ' || quote(g_source_database) || '.');

    l_match_pattern := UPPER(REGEXP_SUBSTR(g_source_database, '^[^.]+')) || '%';

    FOR r IN (
        SELECT OWNER, DB_LINK, HOST
        FROM DBA_DB_LINKS
        WHERE (OWNER = 'PUBLIC' OR OWNER = g_capture_user)
        AND UPPER(DB_LINK) LIKE l_match_pattern
        ORDER BY OWNER, DB_LINK
    ) LOOP
        dbms_output.put_line('INFO: Found database link ' || quote(r.DB_LINK) || ' (owner=' || quote(r.OWNER) || ', host=' || quote(r.HOST) || ').');
        IF r.OWNER = 'PUBLIC' THEN
            l_public_count := l_public_count + 1;
        ELSE
            l_private_count := l_private_count + 1;
        END IF;
    END LOOP;

    IF l_public_count = 0 THEN
        dbms_output.put_line('WARN: No PUBLIC database link matching ' || quote(l_match_pattern) || ' found.');
    END IF;
    IF l_private_count = 0 THEN
        dbms_output.put_line('WARN: No private database link owned by ' || quote(g_capture_user) || ' matching ' || quote(l_match_pattern) || ' found.');
    END IF;
    IF l_public_count = 0 OR l_private_count = 0 THEN
        dbms_output.put_line('Capture-through-link requires both a public and a private database link to the source. Please refer to the documentation for the steps to set up the database links: ' || l_docs_url);
    END IF;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to probe source database links.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Checks if a user has the specified role in the given container
FUNCTION has_role(p_role IN VARCHAR2, p_user IN VARCHAR2, p_container IN VARCHAR2 DEFAULT NULL) RETURN BOOLEAN
IS
    l_count NUMBER;
BEGIN
    IF p_container IS NULL THEN
        SELECT COUNT(*)
        INTO l_count
        FROM DBA_ROLE_PRIVS
        WHERE GRANTED_ROLE = p_role
        CONNECT BY PRIOR GRANTED_ROLE = GRANTEE
        START WITH GRANTEE = p_user;
    ELSE
        SELECT COUNT(*)
        INTO l_count
        FROM CDB_ROLE_PRIVS
        WHERE GRANTED_ROLE = p_role AND CON_ID = CON_NAME_TO_ID(p_container)
        CONNECT BY PRIOR GRANTED_ROLE = GRANTEE AND PRIOR CON_ID = CON_ID
        START WITH GRANTEE = p_user AND CON_ID = CON_NAME_TO_ID(p_container);
    END IF;

    RETURN l_count > 0;
END;

-- Checks if connect user has 'SELECT_CATALOG_ROLE'
PROCEDURE validate_select_catalog_role
IS
    l_privs_docs_url VARCHAR2(1000) := k_docs_base_url || '/prereqs-validation.html#configure-database-users';
    l_container VARCHAR2(128) := CASE WHEN g_pdb_name IS NOT NULL THEN g_pdb_name ELSE NULL END;
BEGIN
    dbms_output.put_line(k_new_line || 'Validating ''SELECT_CATALOG_ROLE'' role for database user ' || quote(g_connect_user) || '.');

    IF has_role('SELECT_CATALOG_ROLE', g_connect_user, l_container) THEN
        dbms_output.put_line('SUCCESS: ' || msg_container_prefix(l_container) || 'Database user ' || quote(g_connect_user) || ' has ''SELECT_CATALOG_ROLE'' role.');
    ELSE
        dbms_output.put_line('FAILED: ' || msg_container_prefix(l_container) || 'Database user ' || quote(g_connect_user) || ' is missing ''SELECT_CATALOG_ROLE'' role.');
        dbms_output.put_line('Please refer to the documentation for steps to grant this role: ' || l_privs_docs_url);
    END IF;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to validate ''SELECT_CATALOG_ROLE'' role for database user ' || quote(g_connect_user) || '.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Checks the user privileges for snapshot
PROCEDURE validate_snapshot_privileges
IS
    l_docs_url VARCHAR2(1000) := k_docs_base_url || '/prereqs-validation.html#configure-database-users';
    l_container VARCHAR2(128) := CASE WHEN g_pdb_name IS NOT NULL THEN g_pdb_name ELSE NULL END;
BEGIN
    dbms_output.put_line(k_new_line || 'Validating snapshot privileges for database user ' || quote(g_connect_user) || '.');

    IF has_system_privilege('SELECT ANY TABLE', g_connect_user, l_container) AND has_system_privilege('LOCK ANY TABLE', g_connect_user, l_container) THEN
        dbms_output.put_line('SUCCESS: ' || msg_container_prefix(l_container) || 'Database user ' || quote(g_connect_user) || ' has the required system privileges ''SELECT ANY TABLE'' and ''LOCK ANY TABLE''.');
    ELSE
        dbms_output.put_line('INFO: ' || msg_container_prefix(l_container) || 'Database user ' || quote(g_connect_user) || ' does not have the ''SELECT ANY TABLE'' and ''LOCK ANY TABLE'' system privileges. Either grant these system privileges or grant the ''SELECT'' object privilege on the tables included in the capture set.');
        dbms_output.put_line('Please refer to the documentation for the steps to grant this access: ' || l_docs_url);
    END IF;

    IF has_system_privilege('FLASHBACK ANY TABLE', g_connect_user, l_container) THEN
        dbms_output.put_line('SUCCESS: ' || msg_container_prefix(l_container) || 'Database user ' || quote(g_connect_user) || ' has the system privilege ''FLASHBACK ANY TABLE'' required to snapshot table data.');
    ELSE
        dbms_output.put_line('INFO: ' || msg_container_prefix(l_container) || 'Database user ' || quote(g_connect_user) || ' does not have the ''FLASHBACK ANY TABLE'' system privilege. Either grant this system privilege or grant the ''FLASHBACK'' object privilege on the tables included in the capture set if the connector is configured to snapshot table data.');
        dbms_output.put_line('Please refer to the documentation for the steps to grant this access: ' || l_docs_url);
    END IF;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to validate snapshot privileges.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Checks the outbound server details
PROCEDURE validate_outbound_server
IS
    l_connect_user DBA_XSTREAM_OUTBOUND.CONNECT_USER%TYPE;
    l_status DBA_XSTREAM_OUTBOUND.STATUS%TYPE;
    l_docs_url VARCHAR2(1000) := k_docs_base_url || '/prereqs-validation.html#create-xstream-out';
BEGIN
    dbms_output.put_line(k_new_line || 'Validating outbound server: ' || quote(g_outbound_server));
    SELECT CONNECT_USER, STATUS
    INTO l_connect_user, l_status
    FROM DBA_XSTREAM_OUTBOUND
    WHERE SERVER_NAME = g_outbound_server;

    IF l_connect_user != g_connect_user THEN
        dbms_output.put_line('FAILED: The connect user ' || quote(l_connect_user) || ' for outbound server ' || quote(g_outbound_server) || ' does not match the connect user ' || quote(g_connect_user) || ' provided in the input.');
        dbms_output.put_line('Please refer to the documentation for the steps to change the connect user for the outbound server: ' || l_docs_url);
    END IF;

    IF g_capture_process IS NULL THEN
        dbms_output.put_line('WARN: The outbound server ' || quote(g_outbound_server) || ' does not have an associated capture process.');
    END IF;

    IF l_status = 'ABORTED' THEN
        dbms_output.put_line('FAILED: The outbound server is in ''ABORTED'' status due to an error.');
        dbms_output.put_line('To view details, run the following query on the ''DBA_APPLY'' view: ''SELECT STATUS, ERROR_NUMBER, ERROR_MESSAGE FROM DBA_APPLY WHERE PURPOSE = ''XStream Out'' AND APPLY_NAME = ' || quote(g_outbound_server) || '''');
    ELSIF l_status = 'DISABLED' THEN
        dbms_output.put_line('INFO: The outbound server is currently in ''DISABLED'' status and not running.');
        dbms_output.put_line('You can start the outbound server manually using the ''DBMS_XSTREAM_ADM.START_OUTBOUND'' procedure, or it will start automatically when the connector attaches to it during startup.');
    ELSE
        dbms_output.put_line('SUCCESS: The outbound server has been successfully validated.');
    END IF;
EXCEPTION
    WHEN NO_DATA_FOUND THEN
        dbms_output.put_line('FAILED: The outbound server ' || quote(g_outbound_server) || ' does not exist.');
        dbms_output.put_line('Please refer to the documentation for the steps to set up the outbound server: ' || l_docs_url);
    WHEN others THEN
        log_exception('Failed to validate outbound server ' || quote(g_outbound_server) || '.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Checks the capture process details
PROCEDURE validate_capture_process
IS
    l_capture_user DBA_CAPTURE.CAPTURE_USER%TYPE;
    l_status DBA_CAPTURE.STATUS%TYPE;
    l_docs_url VARCHAR2(1000) := k_docs_base_url || '/prereqs-validation.html#create-xstream-out';
BEGIN
    dbms_output.put_line(k_new_line || 'Validating capture process: ' || quote(g_capture_process));
    SELECT CAPTURE_USER, STATUS
    INTO l_capture_user, l_status
    FROM DBA_CAPTURE
    WHERE CAPTURE_NAME = g_capture_process;

    IF l_capture_user != g_capture_user THEN
        dbms_output.put_line('FAILED: The capture user ' || quote(l_capture_user) || ' for capture process ' || quote(g_capture_process) || ' does not match the capture user ' || quote(g_capture_user) || ' provided in the input.');
        dbms_output.put_line('Please refer to the documentation for the steps to change the capture user for the capture process: ' || l_docs_url);
    END IF;

    IF l_status = 'ABORTED' THEN
        dbms_output.put_line('FAILED: The capture process is in ''ABORTED'' status due to an error.');
        dbms_output.put_line('To view details, run the following query on the ''DBA_CAPTURE'' view: ''SELECT STATUS, ERROR_NUMBER, ERROR_MESSAGE FROM DBA_CAPTURE WHERE PURPOSE = ''XStream Out'' AND CAPTURE_NAME = ' || quote(g_capture_process) || '''');
    ELSIF l_status = 'DISABLED' THEN
        dbms_output.put_line('INFO: The capture process is currently in ''DISABLED'' status and not running.');
        dbms_output.put_line('You can start the capture process manually using the ''DBMS_CAPTURE_ADM.START_CAPTURE'' procedure, or it may start automatically when the connector attaches to it during startup.');
    ELSE
        dbms_output.put_line('SUCCESS: The capture process has been successfully validated.');
    END IF;
EXCEPTION
    WHEN NO_DATA_FOUND THEN
        dbms_output.put_line('FAILED: The capture process ' || quote(g_capture_process) || ' does not exist.');
        dbms_output.put_line('Please refer to the documentation for the steps to set up the capture process: ' || l_docs_url);
    WHEN others THEN
        log_exception('Failed to validate capture process ' || quote(g_capture_process) || '.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Checks if the 'USE_RAC_SERVICE' parameter is enabled for the capture process (only on RAC databases)
PROCEDURE validate_rac_enabled
IS
    l_rac_service_enabled DBA_CAPTURE_PARAMETERS.VALUE%TYPE;
    l_docs_url VARCHAR2(1000) := k_docs_base_url || '/prereqs-validation.html#capture-changes-from-oracle-rac';
BEGIN
    dbms_output.put_line(k_new_line || 'Validating if the ''USE_RAC_SERVICE'' parameter is enabled for the capture process ' || quote(g_capture_process) || '.');
    SELECT VALUE
    INTO l_rac_service_enabled
    FROM DBA_CAPTURE_PARAMETERS
    WHERE CAPTURE_NAME = g_capture_process
    AND PARAMETER = 'USE_RAC_SERVICE';

    IF l_rac_service_enabled = 'N' THEN
        dbms_output.put_line('FAILED: The capture parameter ''USE_RAC_SERVICE'' is not set to ''Y'' for the capture process ' || quote(g_capture_process) || '.');
        dbms_output.put_line('Please refer to the documentation for the steps to capture changes from an Oracle RAC database: ' || l_docs_url);
    ELSE
        dbms_output.put_line('SUCCESS: The capture parameter ''USE_RAC_SERVICE'' is set to ''Y'' for the capture process ' || quote(g_capture_process) || '.');
    END IF;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to validate if the ''USE_RAC_SERVICE'' parameter is enabled for the capture process ' || quote(g_capture_process) || '.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Checks if the outbound server's queue exists (catches partial-setup states like manual queue drops)
PROCEDURE validate_queue
IS
    l_queue_owner DBA_XSTREAM_OUTBOUND.QUEUE_OWNER%TYPE;
    l_queue_name  DBA_XSTREAM_OUTBOUND.QUEUE_NAME%TYPE;
    l_count       NUMBER;
    l_docs_url VARCHAR2(1000) := k_docs_base_url || '/prereqs-validation.html#create-xstream-out';
BEGIN
    dbms_output.put_line(k_new_line || 'Validating queue for outbound server ' || quote(g_outbound_server) || '.');

    SELECT QUEUE_OWNER, QUEUE_NAME
    INTO l_queue_owner, l_queue_name
    FROM DBA_XSTREAM_OUTBOUND
    WHERE SERVER_NAME = g_outbound_server;

    SELECT COUNT(*)
    INTO l_count
    FROM DBA_QUEUES
    WHERE OWNER = l_queue_owner
    AND NAME = l_queue_name;

    IF l_count > 0 THEN
        dbms_output.put_line('SUCCESS: Queue ' || quote(l_queue_owner || '.' || l_queue_name) || ' for outbound server ' || quote(g_outbound_server) || ' exists.');
    ELSE
        dbms_output.put_line('FAILED: Queue ' || quote(l_queue_owner || '.' || l_queue_name) || ' for outbound server ' || quote(g_outbound_server) || ' does not exist.');
        dbms_output.put_line('Please refer to the documentation for the steps to set up the outbound server: ' || l_docs_url);
    END IF;
EXCEPTION
    WHEN NO_DATA_FOUND THEN
        -- Outbound row missing: validate_outbound_server already reported it.
        dbms_output.put_line('INFO: Could not check the queue because the outbound server ' || quote(g_outbound_server) || ' does not exist.');
    WHEN others THEN
        log_exception('Failed to validate queue for outbound server ' || quote(g_outbound_server) || '.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- Logs 'LOG_ARCHIVE_CONFIG' and every enabled 'LOG_ARCHIVE_DEST_n' as INFO
PROCEDURE log_redo_transport_info
IS
BEGIN
    dbms_output.put_line(k_new_line || 'Logging redo transport configuration.');

    FOR r IN (
        SELECT d.INST_ID AS inst_id,
               d.NAME AS name,
               d.VALUE AS value
        FROM GV$PARAMETER d
        LEFT JOIN GV$PARAMETER s
            ON s.INST_ID = d.INST_ID
           AND s.NAME = REPLACE(d.NAME, 'log_archive_dest_', 'log_archive_dest_state_')
           AND d.NAME LIKE 'log_archive_dest_%'
           AND d.NAME NOT LIKE 'log_archive_dest_state_%'
        WHERE (d.NAME = 'log_archive_config' AND d.VALUE IS NOT NULL)
           OR (d.NAME LIKE 'log_archive_dest_%'
               AND d.NAME NOT LIKE 'log_archive_dest_state_%'
               AND d.VALUE IS NOT NULL
               AND UPPER(NVL(s.VALUE, 'ENABLE')) = 'ENABLE')
        ORDER BY d.INST_ID, d.NAME
    ) LOOP
        dbms_output.put_line('INFO: ' || CASE WHEN g_is_rac THEN '(Instance ' || r.inst_id || ') ' END || r.name || ' = ' || quote(r.value));
    END LOOP;
EXCEPTION
    WHEN others THEN
        log_exception('Failed to log redo transport configuration.', SQLCODE, SUBSTR(SQLERRM, 1, 128));
END;

-- MAIN SCRIPT
BEGIN
    dbms_output.put_line('Running prerequisite checks for the connector on Oracle Database ' || quote(k_db_version) || ' for role ' || quote(g_role) || ', XStream capture user ' || quote(g_capture_user) || ' and connect user ' || quote(g_connect_user) || '.');

    IF g_role NOT IN (k_role_local, k_role_source, k_role_downstream) THEN
        dbms_output.put_line('FAILED: Invalid role: ' || quote(g_role) || '.');
        dbms_output.put_line('Expected one of: ''LOCAL'', ''SOURCE'', ''DOWNSTREAM''.');
        RETURN;
    END IF;

    IF k_db_version < 19 OR k_db_version >= 23 THEN
        dbms_output.put_line('FAILED: The connector does not support Oracle Database version: ' || quote(k_db_version) || '.');
        dbms_output.put_line('Please refer to the documentation for supported database versions: ' || k_docs_base_url || '/cc-oracle-xstream-cdc-source.html#supported-versions');
        RETURN;
    END IF;

    check_dba_role();

    g_is_rds := is_db_rds();

    g_is_multitenant := is_db_multitenant();

    IF g_is_rds AND g_is_multitenant THEN
      dbms_output.put_line('FAILED: The connector does not support container databases (CDB) on Amazon RDS for Oracle.');
      RETURN;
    END IF;

    IF g_is_rds AND g_role != k_role_local THEN
      dbms_output.put_line('FAILED: The connector does not support Amazon RDS for Oracle in a downstream-capture topology (role ' || quote(g_role) || ').');
      RETURN;
    END IF;

    IF g_role = k_role_source THEN
        IF g_is_multitenant THEN
            dbms_output.put_line('INFO: Source database is multitenant. The downstream database must also be multitenant.');
        ELSE
            dbms_output.put_line('INFO: Source database is non-CDB. The downstream database must also be non-CDB.');
        END IF;
    ELSIF g_role = k_role_downstream THEN
        IF g_is_multitenant THEN
            dbms_output.put_line('INFO: Downstream database is multitenant. The source database must also be multitenant.');
        ELSE
            dbms_output.put_line('INFO: Downstream database is non-CDB. The source database must also be non-CDB.');
        END IF;
    END IF;

    IF g_role = k_role_downstream THEN
        g_pdb_name := NULL;
    ELSIF g_is_multitenant AND g_pdb_name IS NOT NULL THEN
        IF pdb_exists(g_pdb_name) THEN
            dbms_output.put_line('Using Pluggable Database (PDB) name: ' || quote(g_pdb_name) || '.');
        ELSE
            dbms_output.put_line('FAILED: The Pluggable Database (PDB) name ' || quote(g_pdb_name) || ' does not exist in the database.');
            dbms_output.put_line('Please verify that the specified PDB name is correct and present in the database.');
            RETURN;
        END IF;
    ELSIF g_pdb_name IS NOT NULL THEN
        g_pdb_name := NULL;
    END IF;

    g_is_rac := is_db_rac();

    validate_xstream();

    IF g_role != k_role_source THEN
        resolve_capture_process();
    END IF;

    IF g_role = k_role_downstream THEN
        resolve_downstream_real_time_mine();
    END IF;

    validate_log_mode();

    IF g_role != k_role_downstream THEN
        validate_supplemental_logging();
        validate_archive_log_retention();
    ELSE
        dbms_output.put_line(k_new_line || 'INFO: Skipping supplemental logging and archive log retention checks for role ' || quote(g_role) || ' (validated on the source database).');
    END IF;

    IF g_role != k_role_local THEN
        validate_global_names();
        validate_remote_login_passwordfile();
    END IF;

    IF g_role = k_role_downstream AND g_capture_type = 'REAL_TIME' THEN
        validate_standby_logfile();
    ELSIF g_role = k_role_downstream AND g_capture_type = 'ARCHIVED' THEN
        dbms_output.put_line(k_new_line || 'INFO: Skipping standby redo log check; capture is archived-log downstream and does not require standby redo logs.');
    END IF;

    init_required_sys_privs();

    IF g_role = k_role_downstream THEN
        resolve_use_database_link();
    END IF;

    IF user_exists(g_capture_user) THEN
        validate_system_privileges(g_capture_user);
        validate_xstream_admin();
        IF g_role = k_role_source THEN
            validate_access_from_remote();
        ELSIF g_role = k_role_downstream THEN
            validate_create_database_link_grant();
            validate_source_db_links();
        END IF;
    END IF;

    IF user_exists(g_connect_user) THEN
        validate_system_privileges(g_connect_user);
        validate_select_catalog_role();
        IF g_role != k_role_downstream THEN
            validate_snapshot_privileges();
        ELSE
            dbms_output.put_line(k_new_line || 'INFO: Skipping snapshot privileges check for role ' || quote(g_role) || ' (validated on the source database).');
        END IF;
    END IF;

    IF g_role != k_role_source THEN
        validate_outbound_server();

        IF g_capture_process IS NOT NULL THEN
            validate_capture_process();

            IF g_is_rac THEN
                validate_rac_enabled();
            END IF;
        END IF;

        validate_queue();
    ELSE
        dbms_output.put_line(k_new_line || 'INFO: Skipping outbound server, capture process and queue checks for role ' || quote(g_role) || ' (validated on the downstream database).');
    END IF;

    IF g_role != k_role_local THEN
        log_redo_transport_info();
    END IF;

    dbms_output.put_line(k_new_line || 'Finished script execution.');
END;
/