-- ============================================================================
-- ZODAN (Zero Downtime Add Node) - Spock Extension
-- Version: 1.0.0
-- Required Spock Version: 5.0.9 or later
-- Mixed versions: the new node must run the same or a newer major.minor than
--                 every existing node (for example a 6.0.x node joining a
--                 cluster of 5.0.x nodes, or of 5.0.x and 6.0.x nodes).
--                 Existing nodes may be mixed; a cluster becomes mixed the
--                 moment a newer node joins it.
-- ============================================================================
-- Adds a new node to the cluster of Spock.
-- NOTE: Run the add_node procedure on the new node you are adding, not on an existing cluster node.

-- Usage:
-- CALL add_node(
--   'source_node_name',
--   'source_node_dsn',
--   'new_node_name',
--   'new_node_dsn',
--   'new_node_location', -- optional
--   'new_node_country',  -- optional
--   new_node_info        -- optional (jsonb)
-- );

-- Example:
-- CALL add_node(
--   'n1',
--   'host=127.0.0.1 dbname=pgedge port=5431 user=pgedge password=pgedge',
--   'n4',
--   'host=127.0.0.1 dbname=pgedge port=5434 user=pgedge password=pgedge'
-- );


-- ============================================================================

-- ============================================================================
-- Function: sync_timeout
-- Purpose : Resolve the budget, in seconds, for a single synchronisation wait.
--           Every wait in this script that bounds a whole synchronisation step
--           reads its limit from here, so an operator can raise them all at
--           once with "SET spock.sync_timeout = '2h'" instead of editing this
--           file.
-- Returns  : integer, seconds. Always positive, so callers need no special case
--           for an unbounded wait.
-- ============================================================================
CREATE OR REPLACE FUNCTION spock.sync_timeout()
RETURNS integer LANGUAGE sql STABLE AS $$
    SELECT COALESCE(
               (SELECT NULLIF(setting::integer, 0) FROM pg_settings
                 WHERE name = 'spock.sync_timeout'),
               180);
$$;

-- ============================================================================
-- Function: gen_sub_name
-- Purpose : Generate a subscription name following the sub_{provider}_{subscriber}
--           convention (i.e. sub_{remote}_{local}).
-- Arguments:
--   provider_node   - The node supplying data (remote/provider).
--   subscriber_node - The node receiving data (local/subscriber).
-- Returns  : text, e.g. 'sub_n1_n2' where n1 is the provider and n2 is the subscriber.
-- ============================================================================
CREATE OR REPLACE FUNCTION spock.gen_sub_name(
    provider_node text,
    subscriber_node text
) RETURNS text LANGUAGE sql IMMUTABLE AS $$
    SELECT 'sub_' || provider_node || '_' || subscriber_node;
$$;

-- ============================================================================
-- Function: version_to_array
-- Purpose : Convert a Spock version string (e.g. '5.0.10', '5.0.4-devel') into
--           an integer array (e.g. {5,0,10}) so versions can be compared
--           numerically. A plain text comparison is wrong because it orders
--           versions lexically (e.g. '5.0.10' < '5.0.4').
-- Arguments:
--   v - The version string. Any non-numeric suffix (such as '-devel') is ignored.
-- Returns  : int[], the dotted numeric components in order.
-- ============================================================================
CREATE OR REPLACE FUNCTION spock.version_to_array(v text)
RETURNS int[] LANGUAGE sql IMMUTABLE AS $$
    SELECT string_to_array((regexp_match(v, '[0-9]+(?:\.[0-9]+)*'))[1], '.')::int[];
$$;

-- ============================================================================
-- Function: version_major_minor
-- Purpose : Extract the {major, minor} components of a Spock version string as
--           an int[] (e.g. '5.0.10' and '5.0.4' both yield {5,0}). Used to
--           allow rolling upgrades: nodes may differ in patch level as long as
--           their major.minor matches.
-- Arguments:
--   v - The version string. Any non-numeric suffix (such as '-devel') is ignored.
-- Returns  : int[] of length up to 2, the major and minor components.
-- ============================================================================
CREATE OR REPLACE FUNCTION spock.version_major_minor(v text)
RETURNS int[] LANGUAGE sql IMMUTABLE AS $$
    SELECT (spock.version_to_array(v))[1:2];
$$;

-- ============================================================================
-- Function: progress_lsn_column
-- Purpose : Name of the spock.progress column that holds the LSN of the last
--           commit applied from a peer, for a node running the given Spock
--           version.  Spock 6.0.0 renamed it from remote_lsn to
--           remote_commit_lsn, so a query sent to a remote node over dblink
--           has to use the name that node understands.
-- Arguments:
--   v - The remote node's Spock version string.
-- Returns  : 'remote_lsn' for versions before 6.0.0, else 'remote_commit_lsn'.
-- ============================================================================
CREATE OR REPLACE FUNCTION spock.progress_lsn_column(v text)
RETURNS text LANGUAGE sql IMMUTABLE AS $$
    SELECT CASE WHEN spock.version_to_array(v) < ARRAY[6,0,0]
                THEN 'remote_lsn' ELSE 'remote_commit_lsn' END;
$$;

-- ============================================================================
-- Procedure: check_spock_version_compatibility
-- Purpose: Verify all nodes have the same Spock version before adding a node
-- ============================================================================
CREATE OR REPLACE PROCEDURE spock.check_spock_version_compatibility(
    src_dsn text,
    new_node_dsn text,
    verb boolean DEFAULT false
) LANGUAGE plpgsql AS $$
DECLARE
    min_required_version text := '5.0.9';
    src_version text;
    new_version text;
    node_rec RECORD;
    remotesql text;
    node_version text;
    version_mismatch boolean := false;   -- any existing node differs in major.minor
    existing_versions text := NULL;
BEGIN
    -- Get source node Spock version
    remotesql := 'SELECT extversion FROM pg_extension WHERE extname = ''spock''';
    IF verb THEN
        RAISE NOTICE 'Checking Spock version on source node';
    END IF;
    SELECT * FROM dblink(src_dsn, remotesql) AS t(version text) INTO src_version;

    IF src_version IS NULL THEN
        RAISE EXCEPTION 'Spock extension not found on source node';
    END IF;

    -- Check source node has required version (compare numerically, not lexically)
    IF spock.version_to_array(src_version) < spock.version_to_array(min_required_version) THEN
        RAISE EXCEPTION 'Spock version mismatch: source node has version %, but minimum required version is %. Please upgrade all nodes to at least %.',
            src_version, min_required_version, min_required_version;
    END IF;

    -- Get new node Spock version
    IF verb THEN
        RAISE NOTICE 'Checking Spock version on new node';
    END IF;
    SELECT * FROM dblink(new_node_dsn, remotesql) AS t(version text) INTO new_version;

    IF new_version IS NULL THEN
        RAISE EXCEPTION 'Spock extension not found on new node';
    END IF;

    -- Check new node has required version (compare numerically, not lexically)
    IF spock.version_to_array(new_version) < spock.version_to_array(min_required_version) THEN
        RAISE EXCEPTION 'Spock version mismatch: new node has version %, but minimum required version is %. Please upgrade all nodes to at least %.',
            new_version, min_required_version, min_required_version;
    END IF;

    -- Patch differences are fine.  The new node may run a newer major.minor
    -- than the existing nodes (e.g. a 6.0.x node joining 5.0.x nodes), but
    -- never an older one: the new node's sync worker knows how to talk to
    -- older providers, not the other way around.  The source is checked
    -- here, every other existing node in the loop below.
    IF spock.version_major_minor(new_version) < spock.version_major_minor(src_version) THEN
        RAISE EXCEPTION 'Spock version mismatch: new node has version %, but source node has version %. The new node must run the same or a newer major.minor version than every existing node.',
            new_version, src_version;
    END IF;

    -- Check all existing nodes in cluster
    FOR node_rec IN
        SELECT node_name, if_dsn
        FROM dblink(src_dsn,
            'SELECT n.node_name, i.if_dsn FROM spock.node n JOIN spock.node_interface i ON n.node_id = i.if_nodeid'
        ) AS t(node_name text, if_dsn text)
    LOOP
        SELECT * FROM dblink(node_rec.if_dsn, remotesql) AS t(version text) INTO node_version;

        IF node_version IS NULL THEN
            RAISE EXCEPTION 'Spock extension not found on node %', node_rec.node_name;
        END IF;

        IF spock.version_to_array(node_version) < spock.version_to_array(min_required_version) THEN
            RAISE EXCEPTION 'Spock version mismatch: node % has version %, but required version is at least %. All nodes must have version % or later.',
                node_rec.node_name, node_version, min_required_version, min_required_version;
        END IF;

        -- Existing nodes may differ from each other (a cluster is mixed from
        -- the first newer node on), but none may be newer than the new node.
        IF spock.version_major_minor(new_version) < spock.version_major_minor(node_version) THEN
            RAISE EXCEPTION 'Spock version mismatch: new node has version %, but node % has version %. The new node must run the same or a newer major.minor version than every existing node.',
                new_version, node_rec.node_name, node_version;
        END IF;

        IF spock.version_major_minor(node_version) IS DISTINCT FROM spock.version_major_minor(new_version) THEN
            version_mismatch := true;
        END IF;
        existing_versions := concat_ws(', ', existing_versions, node_rec.node_name || ' ' || node_version);
    END LOOP;

    IF version_mismatch THEN
        RAISE NOTICE 'Mixed-version add: new node runs Spock %, existing nodes run: %', new_version, existing_versions;
    END IF;

    IF verb THEN
        RAISE NOTICE 'Version check passed: All nodes running Spock version % or later. Source version is %, new node version is %', min_required_version, src_version, new_version;
    END IF;
END;
$$;

-- ============================================================================
-- Procedure: get_spock_nodes
-- Purpose : Retrieves all Spock nodes and their DSNs from a remote cluster.
-- Arguments:
--   remote_dsn - DSN string to connect to the remote cluster.
--   verb - Verbose output flag
-- Usage    : CALL get_spock_nodes('host=... dbname=... user=... password=...', true);
-- ============================================================================

DROP PROCEDURE IF EXISTS spock.get_spock_nodes(text, boolean);
CREATE OR REPLACE PROCEDURE spock.get_spock_nodes(src_dsn text, verb boolean)
LANGUAGE plpgsql
AS
$$
BEGIN
    -- Build and execute remote SQL to fetch node details and DSNs
    -- Note: Procedures cannot return tables directly, so we'll use RAISE NOTICE
    -- to display the results or store them in a temporary table

    IF verb THEN
        RAISE NOTICE E'[STEP] get_spock_nodes: Retrieved nodes from remote DSN: %', src_dsn;
    END IF;

    -- Create a temporary table to store results
    CREATE TEMP TABLE IF NOT EXISTS temp_spock_nodes (
        node_id    integer,
        node_name  text,
        location   text,
        country    text,
        info       text,
        dsn        text
    );

    -- Clear previous results
    DELETE FROM temp_spock_nodes;

    -- Insert results into temp table
    IF verb THEN
        RAISE NOTICE '[QUERY] SELECT n.node_id, n.node_name, n.location, n.country, n.info, i.if_dsn FROM spock.node n JOIN spock.node_interface i ON n.node_id = i.if_nodeid';
    END IF;
    INSERT INTO temp_spock_nodes
    SELECT *
    FROM dblink(
        src_dsn,
        'SELECT n.node_id, n.node_name, n.location, n.country, n.info, i.if_dsn
         FROM spock.node n
         JOIN spock.node_interface i ON n.node_id = i.if_nodeid'
    ) AS t(
        node_id integer,
        node_name text,
        location text,
        country text,
        info text,
        dsn text
    );

    IF verb THEN
        RAISE NOTICE 'Retrieved % nodes from remote cluster', (SELECT count(*) FROM temp_spock_nodes);
    END IF;
END;
$$;

-- ============================================================================
-- Temporary table to store spock nodes information
-- ============================================================================
CREATE TEMP TABLE IF NOT EXISTS temp_spock_nodes (
    node_id integer,
    node_name text,
    location text,
    country text,
    info text,
    dsn text
);

-- ============================================================================
-- Procedure: create_sub
-- Purpose : Creates a Spock subscription on a remote node via dblink.
-- Arguments:
--   node_dsn              - DSN string to connect to the remote node.
--   subscription_name     - Name of the subscription to create.
--   provider_dsn          - DSN string of the provider node.
--   replication_sets      - Text array of replication sets.
--   synchronize_structure - Whether to synchronize structure (boolean).
--   synchronize_data      - Whether to synchronize data (boolean).
--   forward_origins       - Text array of origins to forward.
--   apply_delay           - Interval for apply delay.
--   enabled               - Whether to enable the subscription (boolean).
--   force_text_transfer   - Whether to force text transfer (boolean).
--   verb                  - Verbose output flag
-- Usage    : CALL create_sub(...);
-- ============================================================================

CREATE OR REPLACE PROCEDURE spock.create_sub(
    node_dsn text,
    subscription_name text,
    provider_dsn text,
    replication_sets text,
    synchronize_structure boolean,
    synchronize_data boolean,
    forward_origins text,
    apply_delay interval,
    enabled boolean,
    force_text_transfer boolean,
    verb boolean
)
LANGUAGE plpgsql
AS
$$
DECLARE
    sid oid;
    remotesql text;
    exists_count int;
    skip_schema_list text;
    existing_schemas text;
    remotesql_schema text;
BEGIN
    -- Auto-detect existing schemas on new node when synchronize_structure is true
    skip_schema_list := 'ARRAY[]::text[]';

    IF synchronize_structure THEN
        BEGIN
            -- Query existing schemas on the new node (excluding system schemas).
            -- lolor is excluded here: Spock already keeps it out of the
            -- structure dump globally, and putting it into skip_schema would
            -- also exclude the lolor tables from the initial data sync.
            remotesql_schema := 'SELECT string_agg(schema_name, '','') as schemas
                                FROM information_schema.schemata
                                WHERE schema_name NOT IN (''information_schema'', ''pg_catalog'', ''pg_toast'', ''spock'', ''public'', ''lolor'')
                                AND schema_name NOT LIKE ''pg_temp_%''
                                AND schema_name NOT LIKE ''pg_toast_temp_%''';

            IF verb THEN
                RAISE NOTICE '[QUERY] Detecting existing schemas on new node: %', remotesql_schema;
            END IF;

            SELECT schemas INTO existing_schemas FROM dblink(node_dsn, remotesql_schema) AS t(schemas text);

            IF existing_schemas IS NOT NULL AND existing_schemas != '' THEN
                skip_schema_list := 'ARRAY[''' || replace(existing_schemas, ',', ''',''') || ''']::text[]';
                IF verb THEN
                    RAISE NOTICE '[INFO] Found existing schemas to skip: %', existing_schemas;
                END IF;
            ELSE
                IF verb THEN
                    RAISE NOTICE '[INFO] No existing user schemas found on new node';
                END IF;
            END IF;

        EXCEPTION
            WHEN OTHERS THEN
                IF verb THEN
                    RAISE NOTICE '[WARNING] Failed to detect existing schemas: %', SQLERRM;
                END IF;
                skip_schema_list := 'ARRAY[]::text[]';
        END;
    END IF;

    remotesql := format(
        'SELECT spock.sub_create(
            subscription_name := %L,
            provider_dsn := %L,
            replication_sets := %s,
            synchronize_structure := %L,
            synchronize_data := %L,
            forward_origins := %s,
            apply_delay := %L,
            enabled := %L,
            force_text_transfer := %L,
            skip_schema := %s
        )',
        subscription_name,
        provider_dsn,
        replication_sets,
        synchronize_structure::text,
        synchronize_data::text,
        forward_origins,
        apply_delay::text,
        enabled::text,
        force_text_transfer::text,
        skip_schema_list
    );

    IF verb THEN
        RAISE NOTICE '[QUERY] SQL: %', remotesql;
    END IF;

    BEGIN
        SELECT * FROM dblink(node_dsn, remotesql) AS t(sid oid) INTO sid;

    EXCEPTION
        WHEN OTHERS THEN
            IF verb THEN
                RAISE EXCEPTION E'
            [STEP 3] Subscription "%" creation failed on remote node! Error: %
            ', subscription_name, SQLERRM;
            END IF;
    END;
END;
$$;


-- ============================================================================


-- ============================================================================
-- Procedure: create_replication_slot
-- Purpose : Creates a logical replication slot on a remote node via dblink.
-- Arguments:
--   node_dsn   - DSN string to connect to the remote node.
--   slot_name  - Name of the replication slot to create.
--   verb       - Verbose output flag
--   plugin     - Logical decoding plugin (default: 'spock_output').
-- Usage    : CALL create_replication_slot(...);
-- ============================================================================

CREATE OR REPLACE PROCEDURE spock.create_replication_slot(
    node_dsn text,
    slot_name text,
    verb boolean,
    plugin text DEFAULT 'spock_output'
)
LANGUAGE plpgsql
AS
$$
DECLARE
    remotesql      text;
    result         RECORD;
    exists_count   int;
    remote_version int;
BEGIN
    -- ============================================================================
    -- Step 1: Check if replication slot already exists on remote node
    -- ============================================================================
    remotesql := format(
        'SELECT count(*) FROM pg_replication_slots WHERE slot_name = %L',
        slot_name
    );

    IF verb THEN
        RAISE NOTICE '[QUERY] %', remotesql;
    END IF;

    SELECT * FROM dblink(node_dsn, remotesql) AS t(count int) INTO exists_count;

    IF exists_count > 0 THEN
        IF verb THEN
            RAISE NOTICE E'
        [STEP 1] Replication slot "%" already exists on remote node. Skipping creation.',
        slot_name;
        END IF;
        RETURN;
    END IF;


    -- ============================================================================
    -- Step 2: Build remote SQL for replication slot creation
    --         On PG17+ pass failover := true so the slot is picked up by the
    --         native slotsync worker (sync_replication_slots = on) and is
    --         synchronized to physical standbys, matching what Spock does
    --         on its own CREATE_REPLICATION_SLOT path (see spock_sync.c).
    -- ============================================================================
    SELECT v INTO remote_version
      FROM dblink(node_dsn, 'SHOW server_version_num') AS t(v int);

    IF remote_version >= 170000 THEN
        remotesql := format(
            'SELECT slot_name, lsn '
            'FROM pg_create_logical_replication_slot(%L, %L, false, false, true)',
            slot_name, plugin
        );
    ELSE
        remotesql := format(
            'SELECT slot_name, lsn '
            'FROM pg_create_logical_replication_slot(%L, %L)',
            slot_name, plugin
        );
    END IF;

    IF verb THEN
        RAISE NOTICE '[QUERY] %', remotesql;
    END IF;


    -- ============================================================================
    -- Step 3: Execute replication slot creation on remote node using dblink
    -- ============================================================================
    BEGIN
        SELECT * FROM dblink(node_dsn, remotesql) AS t(slot_name text, lsn pg_lsn) INTO result;
        IF verb THEN
            RAISE NOTICE E'
        [STEP 3] Created replication slot "%" with plugin "%" on remote node.',
        slot_name, plugin;
        END IF;
    EXCEPTION
        WHEN OTHERS THEN
            IF verb THEN
                RAISE NOTICE E'
            [STEP 3] Replication slot "%" may already exist or creation failed. Error: %
            ', slot_name, SQLERRM;
            END IF;
    END;
END;
$$;


-- ============================================================================


-- ============================================================================
-- Function: sync_event
-- Purpose : Triggers a sync event on a remote node and returns the resulting LSN.
-- Arguments:
--   node_dsn - DSN string to connect to the remote node.
-- Returns  : pg_lsn (the LSN of the sync event)
-- Usage    : SELECT sync_event('host=... dbname=... user=... password=...');
-- ============================================================================

-- ============================================================================
-- Procedure: sync_event
-- Purpose : Triggers a sync event on a remote node and returns the resulting LSN.
-- Arguments:
--   node_dsn - DSN string to connect to the remote node.
--   verb - Verbose output flag
--   sync_lsn - OUT parameter to receive the LSN
-- Usage    : CALL sync_event('host=... dbname=... user=... password=...', true, NULL);
-- ============================================================================

CREATE OR REPLACE PROCEDURE spock.sync_event(
    node_dsn text,
    verb boolean,
    INOUT sync_lsn pg_lsn DEFAULT NULL
)
LANGUAGE plpgsql
AS
$$
DECLARE
    sync_rec RECORD;
    remotesql text;
BEGIN
    -- Build remote SQL to trigger sync event
    remotesql := 'SELECT spock.sync_event();';

    IF verb THEN
        RAISE NOTICE '[QUERY] %', remotesql;
    END IF;

    -- Execute remote SQL and capture the returned LSN
    SELECT * FROM dblink(node_dsn, remotesql) AS t(lsn pg_lsn) INTO sync_rec;

    IF verb THEN
        RAISE NOTICE E'[STEP] Sync event triggered on remote node: % with LSN %', node_dsn, sync_rec.lsn;
    END IF;

    sync_lsn := sync_rec.lsn;
END;
$$;


-- ============================================================================






-- ============================================================================


-- ============================================================================
-- Procedure: create_node
-- Purpose  : Creates a Spock node on a remote cluster via dblink.
-- Arguments:
--   node_name - Name of the node to create.
--   dsn       - DSN string to connect to the remote cluster.
--   location  - Location of the node (default: 'NY').
--   country   - Country of the node (default: 'USA').
--   info      - Additional node info (default: '{}'::jsonb).
-- Returns   : void
-- Usage     : CALL create_node(...);
-- ============================================================================

CREATE OR REPLACE PROCEDURE spock.create_node(
    node_name text,
    dsn text,
    verb boolean,
    location text DEFAULT 'NY',
    country text DEFAULT 'USA',
    info jsonb DEFAULT '{}'::jsonb
)
LANGUAGE plpgsql
AS
$$
DECLARE
    joinid oid;
    remotesql text;
    exists_count int;
BEGIN
    IF verb THEN
        RAISE NOTICE 'Checking if node % exists on remote cluster', node_name;
    END IF;
    -- ============================================================================
    -- Step 1: Check if node already exists on remote cluster
    -- ============================================================================
    remotesql := format(
        'SELECT count(*) FROM spock.node WHERE node_name = %L',
        node_name
    );

    IF verb THEN
        RAISE NOTICE '[QUERY] %', remotesql;
    END IF;

    SELECT * FROM dblink(dsn, remotesql) AS t(count int) INTO exists_count;

    IF verb THEN
        RAISE NOTICE E'
    [STEP 1] Remote node existence check for node "%": found % record(s).
    ', node_name, exists_count;
    END IF;

    IF exists_count > 0 THEN
        IF verb THEN
            RAISE NOTICE 'Node % already exists remotely. Skipping creation.', node_name;
        END IF;
        RETURN;
    END IF;


    -- ============================================================================
    -- Step 2: Build remote SQL for node creation
    -- ============================================================================
    remotesql := format(
        'SELECT spock.node_create(
            node_name := %L,
            dsn := %L,
            location := %L,
            country := %L,
            info := %L::jsonb
        )',
        node_name, dsn, location, country, info::text
    );

    IF verb THEN
        RAISE NOTICE '[QUERY] %', remotesql;
    END IF;


    -- ============================================================================
    -- Step 3: Execute node creation on remote cluster using dblink
    -- ============================================================================
    BEGIN
        SELECT * FROM dblink(dsn, remotesql) AS t(joinid oid) INTO joinid;

        IF joinid IS NOT NULL THEN
            IF verb THEN
                RAISE NOTICE 'Node % created remotely with id % and DSN: %', node_name, joinid, dsn;
            END IF;
        ELSE
            IF verb THEN
                RAISE EXCEPTION E'
            [STEP 3] Node "%" creation failed remotely!
            ', node_name;
            END IF;
        END IF;
    EXCEPTION
        WHEN OTHERS THEN
            IF verb THEN
                RAISE EXCEPTION E'
            [STEP 3] Node "%" creation failed remotely! Error: %
            ', node_name, SQLERRM;
            END IF;
    END;

END;
$$;

-- ============================================================================


-- ============================================================================
-- Procedure: enable_sub
-- Purpose : Enables a Spock subscription on a remote node via dblink.
-- Arguments:
--   node_dsn   - DSN string to connect to the remote node.
--   sub_name   - Name of the subscription to enable.
--   verb       - Verbose output flag
--   immediate  - Whether to enable immediately (default: true).
-- Usage    : CALL enable_sub(node_dsn, sub_name, true, true);
-- ============================================================================

CREATE OR REPLACE PROCEDURE spock.enable_sub(
    node_dsn text,
    sub_name text,
    verb boolean,
    immediate boolean DEFAULT true
)
LANGUAGE plpgsql
AS
$$
DECLARE
    remotesql text;
BEGIN
    -- ============================================================================
    -- Step 1: Build remote SQL for enabling subscription
    -- ============================================================================
    remotesql := format(
        'SELECT spock.sub_enable(subscription_name := %L, immediate := %L);',
        sub_name, immediate::text
    );

    IF verb THEN
        RAISE NOTICE '[QUERY] %', remotesql;
    END IF;

    -- ============================================================================
    -- Step 2: Execute enabling subscription on remote node using dblink
    -- ============================================================================
    PERFORM * FROM dblink(node_dsn, remotesql) AS t(result text);

    IF verb THEN
        RAISE NOTICE E'
    [STEP 2] Enabled subscription "%" on remote node: %
    ', sub_name, node_dsn;
    END IF;
END;
$$;

-- ============================================================================
-- Procedure: verify_subscription_replicating
-- Purpose : Verifies that a subscription is actively replicating after being enabled
-- Arguments:
--   node_dsn           - DSN of the node where subscription exists
--   subscription_name  - Name of the subscription to verify
--   verb              - Verbose output flag
--   max_attempts      - Maximum verification attempts in seconds (default: 120 = 2 minutes)
-- Usage    : CALL spock.verify_subscription_replicating(node_dsn, 'sub_name', true);
-- Notes   : Raises exception if subscription fails to reach 'replicating' status within timeout
-- ============================================================================
CREATE OR REPLACE PROCEDURE spock.verify_subscription_replicating(
    node_dsn text,
    subscription_name text,
    verb boolean DEFAULT true,
    max_attempts integer DEFAULT 120  -- 2 minutes (120 seconds)
) LANGUAGE plpgsql AS $$
DECLARE
    sub_status text;
    verify_count integer := 0;
BEGIN
    LOOP
        verify_count := verify_count + 1;

        -- Check subscription status on the target node
        SELECT status INTO sub_status
        FROM dblink(node_dsn,
            format('SELECT status FROM spock.sub_show_status() WHERE subscription_name = %L',
                   subscription_name)) AS t(status text);

        IF sub_status = 'replicating' THEN
            IF verb THEN
                RAISE NOTICE '    SUCCESS: %', rpad('Verified subscription ' || subscription_name || ' is replicating', 120, ' ');
            END IF;
            EXIT;
        ELSIF verify_count >= max_attempts THEN
            RAISE EXCEPTION 'Subscription % verification timeout after % seconds (final status: %)',
                          subscription_name, max_attempts, COALESCE(sub_status, 'unknown');
        ELSE
            IF verb THEN
                RAISE NOTICE '    ⏳ %', rpad('Waiting for subscription ' || subscription_name || ' to start replicating (status: ' || COALESCE(sub_status, 'unknown') || ', attempt ' || verify_count || '/' || max_attempts || ')', 120, ' ');
            END IF;
            PERFORM pg_sleep(1);
        END IF;
    END LOOP;
END;
$$;

-- ============================================================================
-- Procedure: show_all_nodes
-- Purpose : Shows comprehensive node status across all nodes in cluster
-- Arguments:
--   cluster_dsn - DSN of any node in the cluster to query cluster state
--   verb        - Verbose output flag
-- Usage    : CALL spock.show_all_nodes('host=127.0.0.1 dbname=pgedge port=5431 user=pgedge password=pgedge', true);
-- ============================================================================
CREATE OR REPLACE PROCEDURE spock.show_all_nodes(
    cluster_dsn text,
    verb boolean DEFAULT true
) LANGUAGE plpgsql AS $$
DECLARE
    node_rec RECORD;
    total_nodes integer := 0;
BEGIN
    IF verb THEN
        RAISE NOTICE '';
        RAISE NOTICE 'COMPREHENSIVE NODE STATUS REPORT';
        RAISE NOTICE '====================================';
    END IF;

    -- Get all nodes from the cluster using the provided DSN
    FOR node_rec IN
        SELECT node_id, node_name, location, country, info, dsn
        FROM dblink(cluster_dsn, 'SELECT n.node_id, n.node_name, n.location, n.country, n.info, i.if_dsn FROM spock.node n JOIN spock.node_interface i ON n.node_id = i.if_nodeid ORDER BY n.node_name')
        AS t(node_id integer, node_name text, location text, country text, info text, dsn text)
    LOOP
        total_nodes := total_nodes + 1;

        IF verb THEN
            RAISE NOTICE '    [NODE] %: % (%s, %s) - %s', node_rec.node_name, node_rec.node_id, node_rec.location, node_rec.country, regexp_replace(node_rec.dsn, ' password=pgedge', ' password=***');
        END IF;
    END LOOP;

    -- Summary
    IF verb THEN
        RAISE NOTICE '';
        RAISE NOTICE 'NODE STATUS SUMMARY';
        RAISE NOTICE '==================';
        RAISE NOTICE 'Total nodes: %', total_nodes;
    END IF;

    -- Raise exception if no nodes are found
    IF total_nodes = 0 THEN
        RAISE EXCEPTION 'No nodes found in the cluster';
    END IF;
END;
$$;

-- ============================================================================
-- Procedure: show_all_subscription_status
-- Purpose : Shows comprehensive subscription status across all nodes in the cluster
-- Arguments:
--   cluster_dsn - DSN of any node in the cluster to query cluster state
--   verb        - Verbose output flag
-- Usage    : CALL spock.show_all_subscription_status('host=127.0.0.1 dbname=pgedge port=5431 user=pgedge password=pgedge', true);
-- ============================================================================
CREATE OR REPLACE PROCEDURE spock.show_all_subscription_status(
    cluster_dsn text,
    verb boolean DEFAULT true
) LANGUAGE plpgsql AS $$
DECLARE
    node_rec RECORD;
    sub_rec RECORD;
    total_subscriptions integer := 0;
    replicating_count integer := 0;
    error_count integer := 0;
BEGIN
    IF verb THEN
        RAISE NOTICE '';
        RAISE NOTICE 'COMPREHENSIVE SUBSCRIPTION STATUS REPORT';
        RAISE NOTICE '==========================================';
    END IF;

    -- Get all nodes from the cluster using the provided DSN
    FOR node_rec IN
        SELECT node_id, node_name, location, country, info, dsn
        FROM dblink(cluster_dsn, 'SELECT n.node_id, n.node_name, n.location, n.country, n.info, i.if_dsn FROM spock.node n JOIN spock.node_interface i ON n.node_id = i.if_nodeid ORDER BY n.node_name')
        AS t(node_id integer, node_name text, location text, country text, info text, dsn text)
    LOOP
        IF verb THEN
            RAISE NOTICE '';
            RAISE NOTICE 'Node: % (DSN: %)', node_rec.node_name, node_rec.dsn;
            RAISE NOTICE 'Subscriptions:';
        END IF;

        -- Get all subscriptions from this node
        FOR sub_rec IN
            SELECT subscription_name, status, provider_node, provider_dsn, slot_name, replication_sets, forward_origins
            FROM dblink(node_rec.dsn, 'SELECT * FROM spock.sub_show_status() ORDER BY subscription_name')
            AS t(subscription_name text, status text, provider_node text, provider_dsn text,
                 slot_name text, replication_sets text[], forward_origins text)
        LOOP
            total_subscriptions := total_subscriptions + 1;

            CASE sub_rec.status
                WHEN 'replicating' THEN
                    replicating_count := replicating_count + 1;
                    IF verb THEN
                        RAISE NOTICE '  [OK] %: % (provider: %, sets: %)',
                            sub_rec.subscription_name, sub_rec.status,
                            sub_rec.provider_node,
                            array_to_string(sub_rec.replication_sets, ',');
                    END IF;
                WHEN 'disabled' THEN
                    IF verb THEN
                        RAISE NOTICE '  [DISABLED] %: % (provider: %, sets: %)',
                            sub_rec.subscription_name, sub_rec.status,
                            sub_rec.provider_node,
                            array_to_string(sub_rec.replication_sets, ',');
                    END IF;
                WHEN 'initializing' THEN
                    IF verb THEN
                        RAISE NOTICE '  [INITIALIZING] %: % (provider: %, sets: %)',
                            sub_rec.subscription_name, sub_rec.status,
                            sub_rec.provider_node,
                            array_to_string(sub_rec.replication_sets, ',');
                    END IF;
                WHEN 'syncing' THEN
                    IF verb THEN
                        RAISE NOTICE '  [SYNCING] %: % (provider: %, sets: %)',
                            sub_rec.subscription_name, sub_rec.status,
                            sub_rec.provider_node,
                            array_to_string(sub_rec.replication_sets, ',');
                    END IF;
                WHEN 'down' THEN
                    error_count := error_count + 1;
                    IF verb THEN
                        RAISE NOTICE '  [ERROR] %: % (provider: %, sets: %)',
                            sub_rec.subscription_name, sub_rec.status,
                            sub_rec.provider_node,
                            array_to_string(sub_rec.replication_sets, ',');
                    END IF;
                ELSE
                    IF verb THEN
                        RAISE NOTICE '  [UNKNOWN] %: % (provider: %, sets: %)',
                            sub_rec.subscription_name, sub_rec.status,
                            sub_rec.provider_node,
                            array_to_string(sub_rec.replication_sets, ',');
                    END IF;
            END CASE;
        END LOOP;

        -- Check if node has no subscriptions
        IF NOT FOUND THEN
            IF verb THEN
                RAISE NOTICE '  (no subscriptions found)';
            END IF;
        END IF;
    END LOOP;

    -- Summary
    IF verb THEN
        RAISE NOTICE '';
        RAISE NOTICE 'SUBSCRIPTION STATUS SUMMARY';
        RAISE NOTICE '==========================';
        RAISE NOTICE 'Total subscriptions: %', total_subscriptions;
        RAISE NOTICE 'Replicating: %', replicating_count;
        RAISE NOTICE 'With errors/issues: %', error_count;

        IF total_subscriptions > 0 THEN
            RAISE NOTICE 'Success rate: %%%',
                round((replicating_count::numeric / total_subscriptions::numeric) * 100, 1);
        END IF;

        IF replicating_count = total_subscriptions AND total_subscriptions > 0 THEN
            RAISE NOTICE 'SUCCESS: All subscriptions are replicating successfully!';
        ELSIF error_count > 0 THEN
            RAISE NOTICE 'WARNING: Some subscriptions have issues - check details above';
        ELSE
            RAISE NOTICE 'INFO: Subscriptions are in various states - check details above';
        END IF;
    END IF;

    -- Raise exception if no subscriptions are replicating
    IF total_subscriptions > 0 AND replicating_count = 0 THEN
        RAISE EXCEPTION 'No subscriptions are in replicating state after node addition';
    END IF;
END;
$$;

-- ============================================================================
-- Procedure to monitor lag using dblink
-- ============================================================================

CREATE OR REPLACE PROCEDURE spock.monitor_lag_with_dblink(
    src_node_name text,
    new_node_name text,
    new_node_dsn text,
    verb boolean DEFAULT true
)
LANGUAGE plpgsql
AS $$
DECLARE
    lag_interval interval;
    lag_bytes bigint;
    max_wait_seconds integer := 60;
    start_time timestamp := clock_timestamp();
    elapsed_interval interval;
    loop_count integer := 0;
    lag_sql text;
    lag_result record;
BEGIN
    RAISE NOTICE '    Monitoring lag from % to % using dblink...', src_node_name, new_node_name;
    LOOP
        loop_count := loop_count + 1;
        lag_sql := format(
            'SELECT now() - commit_timestamp AS lag_interval, replication_lag_bytes AS lag_bytes FROM spock.lag_tracker WHERE origin_name = %L AND receiver_name = %L',
            src_node_name, new_node_name
        );
        -- Use dblink to get lag from remote node
        EXECUTE format(
            'SELECT * FROM dblink(%L, %L) AS t(lag_interval interval, lag_bytes bigint)',
            new_node_dsn, lag_sql
        ) INTO lag_result;

        lag_interval := lag_result.lag_interval;
        lag_bytes := lag_result.lag_bytes;
        elapsed_interval := clock_timestamp() - start_time;

        RAISE NOTICE '% → % lag: % (bytes: %, elapsed: %, loop: %)',
            src_node_name, new_node_name,
            COALESCE(lag_interval::text, 'NULL'),
            COALESCE(lag_bytes::text, 'NULL'),
            elapsed_interval::text, loop_count;

        EXIT WHEN lag_interval IS NOT NULL AND (extract(epoch FROM lag_interval) < 59 OR lag_bytes = 0);
        IF extract(epoch FROM elapsed_interval) > max_wait_seconds THEN
            RAISE NOTICE 'Timeout reached (% seconds) - exiting lag monitoring', max_wait_seconds;
            EXIT;
        END IF;
        PERFORM pg_sleep(1);
    END LOOP;

    IF lag_interval IS NOT NULL AND (extract(epoch FROM lag_interval) < 59 OR lag_bytes = 0) THEN
        IF lag_bytes = 0 THEN
            RAISE NOTICE '    OK: Replication lag monitoring completed (lag_bytes = 0)';
        ELSE
            RAISE NOTICE '    OK: Replication lag monitoring completed';
        END IF;
    ELSE
        RAISE NOTICE '    - Replication lag monitoring timed out';
    END IF;
EXCEPTION
    WHEN OTHERS THEN
        RAISE EXCEPTION '%', '    ✗ Replication lag monitoring failed' || ' (error: ' || SQLERRM || ')';
END;
$$;

CREATE OR REPLACE PROCEDURE spock.wait_for_replication_catchup_with_dblink(
    src_node_name text,
    new_node_name text,
    new_node_dsn text,
    verb boolean DEFAULT true,
    max_wait_seconds integer DEFAULT 180
)
LANGUAGE plpgsql
AS $$
DECLARE
    lag_bytes bigint;
    received_lsn pg_lsn;
    remote_write_lsn pg_lsn;
    lag_sql text;
    wait_started timestamptz := clock_timestamp();
BEGIN
    IF verb THEN
        RAISE NOTICE '    Waiting for replication catchup from % to %...', src_node_name, new_node_name;
    END IF;

    LOOP
        lag_sql := format(
            'SELECT '
            'MAX(remote_insert_lsn) AS remote_write_lsn, '
            'MAX(received_lsn) AS received_lsn, '
            'COALESCE(MAX(remote_insert_lsn - received_lsn), 0) AS lag_bytes '
            'FROM spock.lag_tracker '
            'WHERE origin_name = %L AND receiver_name = %L',
            src_node_name, new_node_name
        );

        EXECUTE format(
            'SELECT * FROM dblink(%L, %L) AS t(remote_write_lsn pg_lsn, received_lsn pg_lsn, lag_bytes bigint)',
            new_node_dsn, lag_sql
        ) INTO remote_write_lsn, received_lsn, lag_bytes;

        IF verb THEN
            RAISE NOTICE '    Catchup % -> %: remote_write_lsn=%, received_lsn=%, lag_bytes=%, elapsed=%',
                src_node_name,
                new_node_name,
                COALESCE(remote_write_lsn::text, '<null>'),
                COALESCE(received_lsn::text, '<null>'),
                COALESCE(lag_bytes::text, '<null>'),
                clock_timestamp() - wait_started;
        END IF;

        EXIT WHEN remote_write_lsn IS NOT NULL
              AND received_lsn IS NOT NULL
              AND lag_bytes <= 0;

        IF EXTRACT(EPOCH FROM (clock_timestamp() - wait_started)) >= max_wait_seconds THEN
            RAISE EXCEPTION 'Replication catchup timeout for % -> % after % seconds (remote_write_lsn=%, received_lsn=%, lag_bytes=%)',
                src_node_name,
                new_node_name,
                max_wait_seconds,
                COALESCE(remote_write_lsn::text, '<null>'),
                COALESCE(received_lsn::text, '<null>'),
                COALESCE(lag_bytes::text, '<null>');
        END IF;

        PERFORM pg_sleep(1);
    END LOOP;

    IF verb THEN
        RAISE NOTICE '    OK: Replication catchup complete for % -> %', src_node_name, new_node_name;
    END IF;
END;
$$;

-- ============================================================================

--
-- Utility routine to correctly extract database name from the DSN string.
--
-- The purpose here is to centralise this specific logic: people may complain
-- about more flexibility in writing the DSN: using upper-case letters in
-- keywords, as an example.
--
CREATE OR REPLACE FUNCTION spock.extract_dbname_from_dsn(dsn text)
RETURNS text AS $$
DECLARE
	dbname text;
BEGIN
	dbname := substring(dsn from 'dbname=([^\s]+)');
    IF dbname IS NOT NULL THEN
        dbname := TRIM(BOTH '''' FROM dbname);
    END IF;
    IF dbname IS NULL THEN
		-- We can't rely on the PGDATABASE environment variable here.
		-- Also, it seems unreliable to guess or use a default name.
		-- So, complain.
        RAISE EXCEPTION 'Exiting add_node: Database name must be explicitly included into the DSN string %', dsn;
    END IF;

	RETURN dbname;
END;
$$ LANGUAGE plpgsql IMMUTABLE;

-- ============================================================================
-- Function: node_repset_list
-- Purpose : Return a node's replication sets as an SQL array literal suitable
--           for the replication_sets argument of spock.create_sub().
--
--           Subscriptions used to be pinned to the three built-in sets
--           (default, default_insert_only, ddl_sql), which silently dropped
--           every table that only lives in a user-created set.
--
--           This must always be called with the DSN of the subscription's
--           PROVIDER. A subscription's replication set names are resolved on
--           the provider, by get_replication_sets() in spock_repset.c, and a
--           name the provider does not have is an ERROR there ("replication
--           set %s not found"), not something it ignores. The walsender then
--           refuses START_REPLICATION and the apply worker retries forever.
--           Replication sets are per-node local state, so two existing nodes
--           can legitimately have different ones.
-- Arguments:
--   node_dsn - DSN of the provider node.
-- Returns  : text, e.g. ARRAY['default', 'default_insert_only', 'ddl_sql']
-- ============================================================================
CREATE OR REPLACE FUNCTION spock.node_repset_list(node_dsn text)
RETURNS text LANGUAGE plpgsql AS $$
DECLARE
    sets text[];
BEGIN
    SELECT array_agg(set_name ORDER BY set_name)
      INTO sets
      FROM dblink(node_dsn,
               'SELECT rs.set_name
                  FROM spock.replication_set rs
                  JOIN spock.local_node ln ON rs.set_nodeid = ln.node_id')
        AS t(set_name text);

    -- A Spock node always has the built-in sets, so an empty result means the
    -- DSN does not point at a node. Fall back rather than build an empty array.
    IF sets IS NULL OR array_length(sets, 1) IS NULL THEN
        sets := ARRAY['default', 'default_insert_only', 'ddl_sql'];
    END IF;

    RETURN 'ARRAY[' ||
           (SELECT string_agg(quote_literal(s), ', ' ORDER BY i)
              FROM unnest(sets) WITH ORDINALITY AS u(s, i)) ||
           ']';
END;
$$;

-- ============================================================================
-- Function: repset_diff
-- Purpose : List the replication set entries that differ between two nodes.
--           An entry is a set definition, a table membership (with its column
--           list and row filter) or a sequence membership, rendered as text so
--           the two sides can be compared directly.
--
--           Used after the mirroring to report drift: the source is a moving
--           target, and a join can run for hours, so somebody may well add a
--           table to a replication set there while it is in progress. That is
--           reported rather than treated as a failure, because the new node is
--           otherwise complete and the remedy is a single repset_add_table on
--           it. Naming the exact entries is what makes that practical.
-- Arguments:
--   left_dsn  - DSN of the first node (the source, in add_node's use).
--   right_dsn - DSN of the second node (the new node).
-- Returns  : rows of (side, entry), where side is 'source only' or 'new only'.
-- ============================================================================
CREATE OR REPLACE FUNCTION spock.repset_diff(left_dsn text, right_dsn text)
RETURNS TABLE (side text, entry text) LANGUAGE plpgsql AS $$
DECLARE
    entries_sql CONSTANT text :=
        'SELECT rs.set_name || '' ['' || concat_ws('','',
                    CASE WHEN rs.replicate_insert   THEN ''insert'' END,
                    CASE WHEN rs.replicate_update   THEN ''update'' END,
                    CASE WHEN rs.replicate_delete   THEN ''delete'' END,
                    CASE WHEN rs.replicate_truncate THEN ''truncate'' END) || '']''
           FROM spock.replication_set rs
           JOIN spock.local_node ln ON rs.set_nodeid = ln.node_id
         UNION ALL
         SELECT rs.set_name || '' '' ||
                quote_ident(n.nspname) || ''.'' || quote_ident(c.relname) ||
                coalesce('' cols='' || rst.set_att_list::text, '''') ||
                coalesce('' filter='' || pg_get_expr(rst.set_row_filter, rst.set_reloid), '''')
           FROM spock.replication_set_table rst
           JOIN spock.replication_set rs ON rs.set_id = rst.set_id
           JOIN spock.local_node ln ON rs.set_nodeid = ln.node_id
           JOIN pg_catalog.pg_class c ON c.oid = rst.set_reloid
           JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
         UNION ALL
         SELECT rs.set_name || '' sequence '' ||
                quote_ident(n.nspname) || ''.'' || quote_ident(c.relname)
           FROM spock.replication_set_seq rss
           JOIN spock.replication_set rs ON rs.set_id = rss.set_id
           JOIN spock.local_node ln ON rs.set_nodeid = ln.node_id
           JOIN pg_catalog.pg_class c ON c.oid = rss.set_seqoid
           JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace';
    left_entries  text[];
    right_entries text[];
BEGIN
    SELECT coalesce(array_agg(e), ARRAY[]::text[]) INTO left_entries
      FROM dblink(left_dsn, entries_sql) AS t(e text);
    SELECT coalesce(array_agg(e), ARRAY[]::text[]) INTO right_entries
      FROM dblink(right_dsn, entries_sql) AS t(e text);

    RETURN QUERY
        SELECT 'source only', e
          FROM unnest(left_entries) AS e
         WHERE e <> ALL (right_entries)
         UNION ALL
        SELECT 'new only', e
          FROM unnest(right_entries) AS e
         WHERE e <> ALL (left_entries)
         ORDER BY 1, 2;
END;
$$;

-- ============================================================================
-- Function: sync_repsets_from_source
-- Purpose : Make the new node's replication sets an exact copy of the source
--           node's: the set definitions, the table memberships (including
--           column lists and row filters) and the sequence memberships.
--
--           This has to run after the structure and data sync, because the
--           tables must exist locally before they can be added to a set.
--
--           Nothing in Spock copies replication set membership. What actually
--           populates the sets on a freshly joined node is AutoDDL firing
--           while pg_restore replays the structure dump, which routes each
--           table by policy (primary key or replica identity to 'default',
--           otherwise 'default_insert_only') rather than by what the source
--           says. That derived membership is wiped here and replaced by the
--           source's, so a table the user deliberately removed from a set on
--           the source does not reappear on the new node, and extension-owned
--           tables such as lolor's (which AutoDDL skips entirely, because it
--           ignores anything created by CREATE EXTENSION) no longer have to be
--           added by hand.
--           Everything here runs on the new node, but it must NOT be called
--           directly from add_node: add_node holds one transaction open across
--           all of its phases, so the AccessShareLock these queries take on
--           spock.local_node would be held for the rest of the run. The later
--           phases reach the new node over dblink, in a second session, and
--           spock.sub_enable() there wants ShareUpdateExclusiveLock on the same
--           catalog. The two sessions would wait on each other forever, with
--           the deadlock detector unable to see it because add_node is blocked
--           on a network read. add_node therefore invokes this over dblink too,
--           so every statement commits and releases its locks immediately.
-- Arguments:
--   src_dsn       - DSN of the source node.
--   new_node_name - Name of the node being added (for messages only).
--   verb          - Verbose output flag.
-- ============================================================================
CREATE OR REPLACE FUNCTION spock.sync_repsets_from_source(
    src_dsn       text,
    new_node_name text,
    verb          boolean
) RETURNS text LANGUAGE plpgsql AS $$
DECLARE
    rec           RECORD;
    created       integer := 0;
    altered       integer := 0;
    dropped       integer := 0;
    cleared       integer := 0;
    added         integer := 0;
    added_seqs    integer := 0;
    missing       text[]  := ARRAY[]::text[];
BEGIN

    -- ------------------------------------------------------------------
    -- 1. Drop the membership AutoDDL derived during the structure restore.
    --    The source's memberships are replayed below; keeping both would
    --    leave two sources of truth. include_partitions is false throughout
    --    so that the source's rows are mirrored one for one: if the source
    --    has a partition in a set, it has its own row there.
    -- ------------------------------------------------------------------
    FOR rec IN
        SELECT rs.set_name, rst.set_reloid
          FROM spock.replication_set_table rst
          JOIN spock.replication_set rs ON rs.set_id = rst.set_id
          JOIN spock.local_node ln ON rs.set_nodeid = ln.node_id
    LOOP
        PERFORM spock.repset_remove_table(rec.set_name::name, rec.set_reloid, false);
        cleared := cleared + 1;
    END LOOP;

    FOR rec IN
        SELECT rs.set_name, rss.set_seqoid
          FROM spock.replication_set_seq rss
          JOIN spock.replication_set rs ON rs.set_id = rss.set_id
          JOIN spock.local_node ln ON rs.set_nodeid = ln.node_id
    LOOP
        PERFORM spock.repset_remove_seq(rec.set_name::name, rec.set_seqoid);
        cleared := cleared + 1;
    END LOOP;

    IF cleared > 0 THEN
        RAISE NOTICE '    OK: %', rpad('Cleared ' || cleared || ' auto-derived replication set entries', 120, ' ');
    END IF;

    -- ------------------------------------------------------------------
    -- 2. Replication set definitions. The three built-in sets already exist
    --    (spock.node_create() creates them), so those are altered into shape
    --    and the rest are created.
    --
    --    The source is read once, into a temp table, and both the create/alter
    --    pass and the drop pass below work from that snapshot. Reading it twice
    --    would let a set created on the source in between be created here and
    --    then immediately dropped again.
    -- ------------------------------------------------------------------
    CREATE TEMP TABLE IF NOT EXISTS temp_src_repsets (
        set_name text PRIMARY KEY,
        ins boolean,
        upd boolean,
        del boolean,
        trunc boolean
    );
    TRUNCATE temp_src_repsets;

    INSERT INTO temp_src_repsets
        SELECT * FROM dblink(src_dsn,
                   'SELECT rs.set_name, rs.replicate_insert, rs.replicate_update,
                           rs.replicate_delete, rs.replicate_truncate
                      FROM spock.replication_set rs
                      JOIN spock.local_node ln ON rs.set_nodeid = ln.node_id')
            AS t(set_name text, ins boolean, upd boolean, del boolean, trunc boolean);

    FOR rec IN SELECT * FROM temp_src_repsets
    LOOP
        IF EXISTS (SELECT 1
                     FROM spock.replication_set rs
                     JOIN spock.local_node ln ON rs.set_nodeid = ln.node_id
                    WHERE rs.set_name = rec.set_name) THEN
            PERFORM spock.repset_alter(rec.set_name::name, rec.ins, rec.upd,
                                       rec.del, rec.trunc);
            altered := altered + 1;
        ELSE
            PERFORM spock.repset_create(rec.set_name::name, rec.ins, rec.upd,
                                        rec.del, rec.trunc);
            created := created + 1;
            IF verb THEN
                RAISE NOTICE '[INFO] Created replication set % on %', rec.set_name, new_node_name;
            END IF;
        END IF;
    END LOOP;

    -- Any set the source does not have should not exist here either.
    FOR rec IN
        SELECT rs.set_name
          FROM spock.replication_set rs
          JOIN spock.local_node ln ON rs.set_nodeid = ln.node_id
         WHERE rs.set_name NOT IN (SELECT set_name FROM temp_src_repsets)
    LOOP
        PERFORM spock.repset_drop(rec.set_name::name, true);
        dropped := dropped + 1;
        IF verb THEN
            RAISE NOTICE '[INFO] Dropped replication set % (absent on source)', rec.set_name;
        END IF;
    END LOOP;

    RAISE NOTICE '    OK: %', rpad('Replication sets: ' || created || ' created, ' ||
                                   altered || ' updated, ' || dropped || ' dropped', 120, ' ');

    -- ------------------------------------------------------------------
    -- 3. Table membership, with column lists and row filters.
    -- ------------------------------------------------------------------
    FOR rec IN
        SELECT * FROM dblink(src_dsn,
                   'SELECT rs.set_name,
                           quote_ident(n.nspname) || ''.'' || quote_ident(c.relname),
                           rst.set_att_list,
                           pg_get_expr(rst.set_row_filter, rst.set_reloid)
                      FROM spock.replication_set_table rst
                      JOIN spock.replication_set rs ON rs.set_id = rst.set_id
                      JOIN spock.local_node ln ON rs.set_nodeid = ln.node_id
                      JOIN pg_catalog.pg_class c ON c.oid = rst.set_reloid
                      JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace')
            AS t(set_name text, relname text, att_list text[], row_filter text)
    LOOP
        IF to_regclass(rec.relname) IS NULL THEN
            missing := missing || rec.relname;
            CONTINUE;
        END IF;

        PERFORM spock.repset_add_table(rec.set_name::name,
                                       rec.relname::regclass,
                                       false,          -- synchronize_data
                                       rec.att_list,
                                       rec.row_filter,
                                       false);         -- include_partitions
        added := added + 1;
    END LOOP;

    IF array_length(missing, 1) IS NOT NULL THEN
        RAISE NOTICE '    [FAILED] %', rpad('Tables replicated by the source are missing on ' || new_node_name, 60, ' ');
        RAISE EXCEPTION 'Exiting add_node: the source node replicates tables that do not exist on %: %. The new node cannot join until they are present.',
            new_node_name, array_to_string(missing, ', ');
    END IF;

    -- ------------------------------------------------------------------
    -- 4. Sequence membership.
    -- ------------------------------------------------------------------
    FOR rec IN
        SELECT * FROM dblink(src_dsn,
                   'SELECT rs.set_name,
                           quote_ident(n.nspname) || ''.'' || quote_ident(c.relname)
                      FROM spock.replication_set_seq rss
                      JOIN spock.replication_set rs ON rs.set_id = rss.set_id
                      JOIN spock.local_node ln ON rs.set_nodeid = ln.node_id
                      JOIN pg_catalog.pg_class c ON c.oid = rss.set_seqoid
                      JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace')
            AS t(set_name text, relname text)
    LOOP
        IF to_regclass(rec.relname) IS NULL THEN
            missing := missing || rec.relname;
            CONTINUE;
        END IF;

        PERFORM spock.repset_add_seq(rec.set_name::name, rec.relname::regclass, false);
        added_seqs := added_seqs + 1;
    END LOOP;

    IF array_length(missing, 1) IS NOT NULL THEN
        RAISE NOTICE '    [FAILED] %', rpad('Sequences replicated by the source are missing on ' || new_node_name, 60, ' ');
        RAISE EXCEPTION 'Exiting add_node: the source node replicates sequences that do not exist on %: %.',
            new_node_name, array_to_string(missing, ', ');
    END IF;

    RAISE NOTICE '    OK: %', rpad('Replication set membership: ' || added || ' tables, ' ||
                                   added_seqs || ' sequences copied from source', 120, ' ');

    -- Drift against the source is reported by the caller, not here: this runs
    -- in a dblink session, and dblink discards remote NOTICE and WARNING, so a
    -- complaint raised at this point would never reach the operator.
    RETURN 'replication sets: ' || created || ' created, ' || altered ||
           ' updated, ' || dropped || ' dropped; membership: ' || added ||
           ' tables, ' || added_seqs || ' sequences copied from source';
END;
$$;

-- ============================================================================
-- Procedure to verify prerequisites for adding a new node
-- (Combines Phase 1: Validating source node prerequisites and Phase 2: Validating new node prerequisites)
-- ============================================================================
CREATE OR REPLACE PROCEDURE spock.verify_node_prerequisites(
    src_node_name text,
    src_dsn text,
    new_node_name text,
    new_node_dsn text,
    verb boolean
) LANGUAGE plpgsql AS $$
DECLARE
    src_exists integer;
    src_sub_exists integer;
    src_repset_exists integer;
    new_exists integer;
    new_sub_exists integer;
    new_repset_exists integer;
    new_db_name text;
    new_db_exists boolean;
BEGIN
    RAISE NOTICE 'Phase 1: Validating source and new node prerequisites';

    -- Verify that add_node is being called on the new node, not on an existing cluster node
    DECLARE
        local_sysid bigint;
        remote_sysid bigint;
        local_dbname text;
        remote_dbname text;
    BEGIN
        SELECT system_identifier INTO local_sysid FROM pg_control_system();
        local_dbname := current_database();

        SELECT * FROM dblink(new_node_dsn,
            'SELECT system_identifier FROM pg_control_system()')
            AS t(sysid bigint) INTO remote_sysid;
        SELECT * FROM dblink(new_node_dsn,
            'SELECT current_database()')
            AS t(dbname text) INTO remote_dbname;

        IF local_sysid != remote_sysid OR local_dbname != remote_dbname THEN
            RAISE NOTICE '    [FAILED] %', rpad('add_node must be run on the new node', 60, ' ');
            RAISE EXCEPTION 'Exiting add_node: This procedure must be run on the new node being added. The new_node_dsn (%) does not match the current database connection. Please connect to the new node and re-run.', new_node_dsn;
        ELSE
            RAISE NOTICE '    OK: %', rpad('Verifying add_node is running on the new node', 120, ' ');
        END IF;
    END;

    -- Check if database specified in new_node_dsn exists on new node

    SELECT spock.extract_dbname_from_dsn(new_node_dsn) INTO new_db_name;
    BEGIN
        SELECT EXISTS(SELECT 1 FROM dblink(new_node_dsn, 'SELECT 1') AS t(dummy int)) INTO new_db_exists;
        RAISE NOTICE '    OK: %', rpad('Checking database ' || new_db_name || ' exists on new node', 120, ' ');
    EXCEPTION
        WHEN OTHERS THEN
            RAISE NOTICE '    [FAILED] %', rpad('Database ' || new_db_name || ' does not exist on new node', 60, ' ');
            RAISE EXCEPTION 'Exiting add_node: Database % does not exist on new node. Please create it first.', new_db_name;
    END;

    -- Check lolor state on the destination. Having the lolor extension
    -- installed is fine (and required when the source replicates lolor
    -- tables), but pre-existing large object data would conflict with the
    -- data synchronized from the source.
    DECLARE
        new_lolor_installed boolean;
        lolor_row_count bigint;
        src_lolor_repset_tables integer;
        remotesql text;
    BEGIN
        remotesql := 'SELECT EXISTS(SELECT 1 FROM pg_catalog.pg_extension WHERE extname = ''lolor'')';
        SELECT * FROM dblink(new_node_dsn, remotesql) AS t(installed boolean) INTO new_lolor_installed;

        IF new_lolor_installed THEN
            remotesql := 'SELECT (SELECT count(*) FROM lolor.pg_largeobject) + (SELECT count(*) FROM lolor.pg_largeobject_metadata)';
            SELECT * FROM dblink(new_node_dsn, remotesql) AS t(row_count bigint) INTO lolor_row_count;

            IF lolor_row_count > 0 THEN
                RAISE NOTICE '    [FAILED] %', rpad('Database ' || new_db_name || ' has pre-existing large object data in the lolor tables', 120, ' ');
                RAISE EXCEPTION 'Exiting add_node: Database % on new node has pre-existing large object data in the lolor tables. The lolor tables must be empty so the large object data from the source node can be synchronized.', new_db_name;
            ELSE
                RAISE NOTICE '    OK: %', rpad('Checking database ' || new_db_name || ' lolor tables are empty', 120, ' ');
            END IF;
        END IF;

        -- If the source node replicates lolor tables, the new node must have
        -- lolor installed or the initial data synchronization will fail.
        remotesql := 'SELECT count(*) FROM spock.tables WHERE nspname = ''lolor'' AND set_name IS NOT NULL';
        SELECT * FROM dblink(src_dsn, remotesql) AS t(count integer) INTO src_lolor_repset_tables;

        IF src_lolor_repset_tables > 0 AND NOT new_lolor_installed THEN
            RAISE NOTICE '    [FAILED] %', rpad('Source node replicates lolor tables but database ' || new_db_name || ' does not have lolor installed', 120, ' ');
            RAISE EXCEPTION 'Exiting add_node: Source node replicates lolor tables but database % on new node does not have the lolor extension installed. Please run CREATE EXTENSION lolor on the new node first.', new_db_name;
        ELSE
            RAISE NOTICE '    OK: %', rpad('Checking lolor requirements for database ' || new_db_name, 120, ' ');
        END IF;
    END;

    -- Check if database has user-created tables in user-created schemas
    DECLARE
        user_table_count integer;
        remotesql text;
    BEGIN
        -- lolor tables are excluded: they belong to the lolor extension and
        -- their emptiness is enforced by the lolor check above.
        remotesql := 'SELECT count(*) FROM pg_tables WHERE schemaname NOT IN (''information_schema'', ''pg_catalog'', ''pg_toast'', ''spock'', ''lolor'') AND schemaname NOT LIKE ''pg_temp_%'' AND schemaname NOT LIKE ''pg_toast_temp_%''';
        SELECT * FROM dblink(new_node_dsn, remotesql) AS t(count integer) INTO user_table_count;

        IF user_table_count > 0 THEN
            RAISE NOTICE '    [FAILED] %', rpad('Database ' || new_db_name || ' has ' || user_table_count || ' user-created tables', 60, ' ');
            RAISE EXCEPTION 'Exiting add_node: Database % on new node has user-created tables. It must be a freshly created database with no user tables (only system and extension tables allowed).', new_db_name;
        ELSE
            RAISE NOTICE '    OK: %', rpad('Checking database ' || new_db_name || ' has no user-created tables', 120, ' ');
        END IF;
    END;

    -- Check that new node has all users that source node has
    DECLARE
        missing_users text := '';
        user_rec RECORD;
        user_exists boolean;
        src_users_sql text;
        check_user_sql text;
    BEGIN
        src_users_sql := 'SELECT rolname FROM pg_roles WHERE rolcanlogin = true AND rolname NOT IN (''postgres'', ''rdsadmin'', ''rdsrepladmin'', ''rds_superuser'') ORDER BY rolname';

        FOR user_rec IN
            SELECT * FROM dblink(src_dsn, src_users_sql) AS t(rolname text)
        LOOP
            -- Build the SQL with the rolname embedded (properly escaped)
            check_user_sql := format('SELECT EXISTS(SELECT 1 FROM pg_roles WHERE rolname = %L AND rolcanlogin = true)', user_rec.rolname);
            SELECT * FROM dblink(new_node_dsn, check_user_sql) AS t(exists boolean) INTO user_exists;

            IF NOT user_exists THEN
                IF missing_users = '' THEN
                    missing_users := user_rec.rolname;
                ELSE
                    missing_users := missing_users || ', ' || user_rec.rolname;
                END IF;
            END IF;
        END LOOP;

        IF missing_users != '' THEN
            RAISE NOTICE '    [FAILED] %', rpad('New node missing users: ' || missing_users, 60, ' ');
            RAISE EXCEPTION 'Exiting add_node: New node is missing the following users that exist on source node: %. Please create these users on the new node before adding it to the cluster.', missing_users;
        ELSE
            RAISE NOTICE '    OK: %', rpad('Checking new node has all source node users', 120, ' ');
        END IF;
    END;

    -- Check: all nodes, included in the cluster, have only enabled subscriptions.
	--
	-- Connect to each node in the cluster and pass through the spock.subscription
	-- table to check subscriptions statuses. Using it we try to avoid cases
	-- when somewhere in the middle a crash or disconnection happens that may
	-- be aggravated by add_node.
    DECLARE
		status_rec     record;
		dsn_rec        record;
		dsns_sql       text;
		sub_status_sql text;
    BEGIN
        dsns_sql := 'SELECT if_dsn,node_name
					 FROM spock.node JOIN spock.node_interface
					 ON (if_nodeid = node_id)
					 WHERE node_id NOT IN (SELECT node_id FROM spock.local_node)';
		sub_status_sql := 'SELECT sub_name, sub_enabled FROM spock.subscription';

        FOR dsn_rec IN SELECT * FROM dblink(src_dsn, dsns_sql)
													AS t(dsn text, node name)
		LOOP
			FOR status_rec IN SELECT * FROM dblink(dsn_rec.dsn, sub_status_sql)
													AS t(name text, status text)
			LOOP
			    IF status_rec.status != 't' THEN
                    RAISE EXCEPTION '    [FAILED] %', rpad('Node ' || dsn_rec.node || ' has disabled subscription ' || status_rec.name, 60, ' ');
                ELSIF verb THEN
                    RAISE NOTICE '    OK: %', rpad('Node with DSN ' || dsn_rec.dsn || ' has enabled subscription ' || status_rec.name, 120, ' ');
                END IF;
			END LOOP;
        END LOOP;
		RAISE NOTICE '    OK: %', rpad('Checking each Spock node has only active subscriptions', 120, ' ');
    END;

    -- Validating new node prerequisites
    SELECT count(*) INTO new_exists FROM spock.node WHERE node_name = new_node_name;
    IF new_exists > 0 THEN
        RAISE NOTICE '    %[FAILED]', rpad('Checking new node "' || new_node_name || '" already exists', 60, ' ');
        RAISE EXCEPTION 'Exiting add_node: New node % already exists', new_node_name;
    ELSE
        RAISE NOTICE '    OK: %', rpad('Checking new node ' || new_node_name || ' does not exist', 120, ' ');
    END IF;
    SELECT count(*) INTO new_sub_exists FROM spock.subscription s JOIN spock.node n ON s.sub_origin = n.node_id WHERE n.node_name = new_node_name;
    IF new_sub_exists > 0 THEN
        RAISE NOTICE '    %[FAILED]', rpad('Checking new node "' || new_node_name || '" has subscriptions', 60, ' ');
        RAISE EXCEPTION 'Exiting add_node: New node % has subscriptions', new_node_name;
    ELSE
        RAISE NOTICE '    OK: %', rpad('Checking new node ' || new_node_name || ' has no subscriptions', 120, ' ');
    END IF;
    SELECT count(*) INTO new_repset_exists FROM spock.replication_set rs JOIN spock.node n ON rs.set_nodeid = n.node_id WHERE n.node_name = new_node_name;
    IF new_repset_exists > 0 THEN
        RAISE NOTICE '    %[FAILED]', rpad('Checking new node "' || new_node_name || '" has replication sets', 60, ' ');
        RAISE EXCEPTION 'Exiting add_node: New node % has replication sets', new_node_name;
    ELSE
        RAISE NOTICE '    OK: %', rpad('Checking new node ' || new_node_name || ' has no replication sets', 120, ' ');
    END IF;
END;
$$;

-- ============================================================================
-- Procedure to create nodes only
-- ============================================================================
CREATE OR REPLACE PROCEDURE spock.create_nodes_only(
    src_node_name text,
    src_dsn text,
    new_node_name text,
    new_node_dsn text,
    verb boolean,
    new_node_location text,
    new_node_country text,
    new_node_info jsonb,
    INOUT initial_node_count integer
) LANGUAGE plpgsql AS $$
BEGIN
    -- Phase 2: Creating nodes
    RAISE NOTICE 'Phase 2: Creating nodes';
    BEGIN
        CALL spock.create_node(src_node_name, src_dsn, verb);
        RAISE NOTICE '    OK: %', rpad('Creating source node ' || src_node_name || '...', 120, ' ');
    EXCEPTION
        WHEN OTHERS THEN
            RAISE NOTICE '    ✗ %', rpad('Creating source node ' || src_node_name  || ' (error: ' || SQLERRM || ')', 120, ' ');
            RAISE;
    END;

    BEGIN
        CALL spock.create_node(new_node_name, new_node_dsn, verb, new_node_location, new_node_country, new_node_info);
        RAISE NOTICE '    OK: %', rpad('Creating new node ' || new_node_name || '...', 120, ' ');
    EXCEPTION
        WHEN OTHERS THEN
            RAISE EXCEPTION '    ✗ %', rpad('Creating new node ' || new_node_name || ' (error: ' || SQLERRM || ')', 120, ' ');
    END;

    -- Get initial node count from source node using inline dblink
    SELECT count INTO initial_node_count
    FROM dblink(src_dsn, 'SELECT count(*) FROM spock.node')
        AS t(count integer);
END;
$$;

-- ============================================================================
-- Procedure to configure cross-node replication
--
-- Not called by add_node, which creates its slots inline in Phase 3, so its
-- notice carries no phase number.
-- ============================================================================
CREATE OR REPLACE PROCEDURE spock.create_replication_slots(
    src_node_name text,
    src_dsn text,
    new_node_name text,
    new_node_dsn text,
    verb boolean
) LANGUAGE plpgsql AS $$
DECLARE
    rec RECORD;
    dbname text;
    slot_name text;
BEGIN
    RAISE NOTICE 'Configuring cross-node replication';
    BEGIN
        FOR rec IN SELECT * FROM temp_spock_nodes LOOP
            IF rec.node_name = src_node_name THEN
                CONTINUE;
            END IF;

		    -- Extract dbname and handle both quoted and unquoted values
			SELECT spock.extract_dbname_from_dsn(rec.dsn) INTO dbname;

		    -- Remove single quotes if present
            IF dbname IS NOT NULL THEN
                dbname := TRIM(BOTH '''' FROM dbname);
            END IF;

		    -- Remove single quotes if present
            IF dbname IS NOT NULL THEN
                dbname := TRIM(BOTH '''' FROM dbname);
            END IF;
            IF dbname IS NULL THEN
                dbname := 'pgedge';
            END IF;

            slot_name := spock.spock_gen_slot_name(
							dbname, rec.node_name,
							spock.gen_sub_name(rec.node_name, new_node_name));

            CALL spock.create_replication_slot(
                rec.dsn,
                slot_name,
                verb,
                'spock_output'
            );
            RAISE NOTICE '    OK: %', rpad('Creating replication slot ' || slot_name || '...', 120, ' ');
        END LOOP;
    EXCEPTION
        WHEN OTHERS THEN
            RAISE NOTICE '%', '    ✗ Creating replication slots...' || ' (error: ' || SQLERRM || ')';
            RAISE;
    END;
END;
$$;


-- ============================================================================
-- Procedure to create disabled subscriptions and slots
-- ============================================================================
CREATE OR REPLACE PROCEDURE spock.create_disable_subscriptions_and_slots(
    src_node_name text,  -- Source node name
    src_dsn text,        -- Source node DSN
    new_node_name text,  -- New node name
    new_node_dsn text,   -- New node DSN
    verb boolean         -- Verbose flag
) LANGUAGE plpgsql AS $$
DECLARE
    rec                RECORD;
    subscription_count integer := 0;
    remotesql          text;
    dbname             text;
    slot_name          text;
	sub_name           text;
    _slot_lsn          pg_lsn;
    _catchup_lsn       pg_lsn;
    repsets            text;
BEGIN
    RAISE NOTICE 'Phase 3: Creating disabled subscriptions and slots';

    -- Get all existing nodes (excluding source and new)
    CALL spock.get_spock_nodes(src_dsn, verb);

    -- Create temporary table to store sync LSNs
    CREATE TEMP TABLE IF NOT EXISTS temp_sync_lsns (
        origin_node text PRIMARY KEY,
        sync_lsn text NOT NULL,
        slot_lsn pg_lsn
    );

    -- Check if there are any "other" nodes (not source, not new)
    IF (SELECT count(*) FROM temp_spock_nodes WHERE node_name != src_node_name AND node_name != new_node_name) = 0 THEN
        -- 2-node scenario: trigger sync event on source node and store it
        BEGIN
            RAISE NOTICE '    - 2-node scenario';
            SELECT * INTO remotesql
            FROM dblink(src_dsn, 'SELECT spock.sync_event()') AS t(sync_lsn text);

            -- Store the sync LSN for later use when enabling subscriptions
            INSERT INTO temp_sync_lsns (origin_node, sync_lsn)
            VALUES (src_node_name, remotesql)
            ON CONFLICT (origin_node) DO UPDATE SET sync_lsn = EXCLUDED.sync_lsn;

            RAISE NOTICE '    OK: %', rpad('Triggering sync event on node ' || src_node_name || ' (LSN: ' || remotesql || ')', 120, ' ');
        EXCEPTION
            WHEN OTHERS THEN
                RAISE EXCEPTION '    ✗ %', rpad('Triggering sync event on node ' || src_node_name || ' (error: ' || SQLERRM || ')', 120, ' ');
        END;

        RAISE NOTICE '    - 2-node scenario: sync event stored, skipping disabled subscriptions';
        RETURN;
    END IF;

    -- For each "other" node (not source, not new), create disabled subscription and slot
    FOR rec IN SELECT * FROM temp_spock_nodes
	           WHERE node_name != src_node_name AND node_name != new_node_name
	LOOP
        -- Create replication slot on the "other" node
        BEGIN
            -- Extract dbname and handle both quoted and unquoted values
            SELECT spock.extract_dbname_from_dsn(rec.dsn) INTO dbname;

            -- Remove single quotes if present
            IF dbname IS NOT NULL THEN
                dbname := TRIM(BOTH '''' FROM dbname);
            END IF;

            -- Remove single quotes if present
            IF dbname IS NOT NULL THEN
                dbname := TRIM(BOTH '''' FROM dbname);
            END IF;
            IF dbname IS NULL THEN dbname := 'pgedge'; END IF;

			slot_name := spock.spock_gen_slot_name(
							dbname, rec.node_name,
							spock.gen_sub_name(rec.node_name, new_node_name));

            remotesql := format('SELECT slot_name, lsn FROM pg_create_logical_replication_slot(%L, ''spock_output'');', slot_name);
            IF verb THEN
                RAISE NOTICE '    Remote SQL for slot creation: %', remotesql;
            END IF;

            SELECT lsn INTO _slot_lsn
                FROM dblink(rec.dsn, remotesql) AS t(slot_name text, lsn pg_lsn);
            RAISE NOTICE '    OK: %', rpad('Creating replication slot ' || slot_name || ' (LSN: ' || _slot_lsn || ')' || ' on node ' || rec.node_name, 120, ' ');
        EXCEPTION
            WHEN OTHERS THEN
                RAISE EXCEPTION '    ✗ %', rpad('Creating replication slot ' || slot_name || ' on node ' || rec.node_name || ' (error: ' || SQLERRM || ')', 120, ' ');
        END;

        -- Trigger a sync event on rec *after* slot creation. This anchors a
        -- real commit at LSN > _slot_lsn, which gives the catchup wait below
        -- a reachable target even when rec is otherwise idle (the slot's
        -- consistent_point typically lands on a RUNNING_XACTS record, not on
        -- a commit, so remote_commit_lsn alone may never reach _slot_lsn).
        -- The same LSN is reused by Phase 8 wait_for_sync_event on new.
        BEGIN
            SELECT t.sync_lsn::pg_lsn INTO _catchup_lsn
            FROM dblink(rec.dsn, 'SELECT spock.sync_event(true)') AS t(sync_lsn text);

            INSERT INTO temp_sync_lsns (origin_node, sync_lsn, slot_lsn)
            VALUES (rec.node_name, _catchup_lsn::text, _slot_lsn)
            ON CONFLICT (origin_node) DO UPDATE
                SET sync_lsn = EXCLUDED.sync_lsn,
                    slot_lsn = EXCLUDED.slot_lsn;

            RAISE NOTICE '    OK: %', rpad('Triggering sync event on node ' || rec.node_name || ' (LSN: ' || _catchup_lsn || ')', 120, ' ');
        EXCEPTION
            WHEN OTHERS THEN
                RAISE EXCEPTION '    ✗ %', rpad('Triggering sync event on node ' || rec.node_name || ' (error: ' || SQLERRM || ')', 120, ' ');
        END;

        -- Wait until src has applied rec changes through _catchup_lsn. Since
        -- _catchup_lsn > _slot_lsn and apply is in commit order, reaching it
        -- proves every rec commit with commit_lsn <= _slot_lsn is committed
        -- on src and visible to the Phase 5 src->new COPY snapshot.
        --
        -- We must use remote_commit_lsn (apply progress at last applied
        -- commit). received_lsn is unsafe here: it advances on keepalive
        -- messages reporting the remote walsender position even when no rec
        -- changes have been applied on src yet.
        DECLARE
            src_progress_lsn         pg_lsn;
            wait_started             timestamptz := clock_timestamp();
            wait_budget              integer  := spock.sync_timeout();
            progress_sql             text;
            src_lsn_column           text;
            v_prev_statement_timeout text;
        BEGIN
            -- The column name depends on the source node's Spock version
            -- (remote_lsn before 6.0.0, remote_commit_lsn from 6.0.0 on).
            SELECT spock.progress_lsn_column(t.v) INTO src_lsn_column
              FROM dblink(src_dsn, 'SELECT extversion FROM pg_extension WHERE extname = ''spock''') AS t(v text);

            progress_sql := format(
                'SELECT p.%I '
                'FROM spock.progress p '
                'JOIN spock.node n ON n.node_id = p.remote_node_id '
                'WHERE p.node_id = (SELECT node_id FROM spock.node_info()) '
                '  AND n.node_name = %L',
                src_lsn_column, rec.node_name);

            RAISE NOTICE '    - Waiting for source node % to apply % changes up to sync LSN %...',
                         src_node_name, rec.node_name, _catchup_lsn;

            LOOP
                BEGIN
                    -- Bound each remote probe so dblink calls cannot hang forever.
                    v_prev_statement_timeout := current_setting('statement_timeout', true);
                    PERFORM set_config('statement_timeout', '5s', true);

                    SELECT * FROM dblink(src_dsn, progress_sql)
                        AS t(lsn pg_lsn) INTO src_progress_lsn;

                    PERFORM set_config('statement_timeout', coalesce(v_prev_statement_timeout, '0'), true);
                EXCEPTION
                    WHEN OTHERS THEN
					    -- Let user know if something wrong happens
						IF verb THEN
					      RAISE NOTICE 'An error happened: %', SQLERRM;
						END IF;
                        -- Transient probe failure; restore timeout and retry.
                        PERFORM set_config('statement_timeout', coalesce(v_prev_statement_timeout, '0'), true);
                        src_progress_lsn := NULL;
                END;

                EXIT WHEN src_progress_lsn IS NOT NULL
                          AND src_progress_lsn >= _catchup_lsn;

                IF clock_timestamp() - wait_started
                       > make_interval(secs => wait_budget) THEN
                    -- Hard failure: a post-slot sync_event commit was emitted
                    -- on rec, so this LSN must reach src unless replication
                    -- is broken. Proceeding would risk losing rec rows in
                    -- (src_progress_lsn, _slot_lsn] from the src->new COPY.
                    RAISE EXCEPTION 'Timed out waiting for source node % to apply % changes through sync LSN % (last seen: %)',
                                    src_node_name, rec.node_name, _catchup_lsn,
                                    COALESCE(src_progress_lsn::text, 'none');
                END IF;

                PERFORM pg_sleep(0.5);
            END LOOP;

            RAISE NOTICE '    OK: %', rpad(
                'Source node ' || src_node_name || ' applied ' || rec.node_name
                || ' changes up to ' || COALESCE(src_progress_lsn::text, 'unknown')
                || ' (needed >= ' || _catchup_lsn || ')', 120, ' ');
        END;

        -- Create disabled subscription on new node from "other" node
        BEGIN
			sub_name := spock.gen_sub_name(rec.node_name, new_node_name);
            -- Drop stale replication origin from a previous add_node cycle
            -- so create_sub starts fresh at 0/0 (avoids stale-LSN data loss).
            BEGIN
                PERFORM dblink_exec(
                    new_node_dsn,
                    format($dsql$
                        DO $x$
                        BEGIN
                            IF EXISTS (SELECT 1 FROM pg_replication_origin
                                       WHERE roname = %L) THEN
                                PERFORM pg_replication_origin_drop(%L);
                            END IF;
                        END $x$
                        $dsql$,
                        slot_name, slot_name)
                );
                RAISE NOTICE '    OK: Dropped stale origin % on new node (if existed)',
                    slot_name;
            EXCEPTION
                WHEN OTHERS THEN
                    RAISE WARNING '    Could not drop stale origin % on new node: %',
                        slot_name, SQLERRM;
            END;
            -- Every replication set this provider has, not just the three
            -- built-ins: a table that only lives in a user-created set would
            -- otherwise never reach the new node, with no error reported. The
            -- list has to come from this provider rather than from the source,
            -- because the provider is where the names are resolved and it
            -- errors out on one it does not have.
            repsets := spock.node_repset_list(rec.dsn);

            CALL spock.create_sub(
                new_node_dsn,                                 -- Create on new node
                sub_name, 									  -- sub_<new_node>_<other_node>
                rec.dsn,                                      -- Provider is other node
                repsets,                                      -- Replication sets (this provider's)
                false,                                        -- synchronize_structure
                false,                                        -- synchronize_data
                'ARRAY[]::text[]',                            -- forward_origins
                '0'::interval,                                -- apply_delay
                false,                                        -- enabled (disabled)
                false,                                        -- force_text_transfer
                verb                                          -- verbose
            );
            RAISE NOTICE '    ✓ %', rpad('Creating initial subscription ' || sub_name || ' on new node ' || new_node_name || ' (provider: ' || rec.node_name || ')', 120, ' ');
            PERFORM pg_sleep(5);
            subscription_count := subscription_count + 1;
        EXCEPTION
            WHEN OTHERS THEN
                RAISE NOTICE '    ✗ %', rpad('Creating initial subscription ' || sub_name || ' on new node ' || new_node_name || ' (provider: ' || rec.node_name || ') (error: ' || SQLERRM || ')', 120, ' ');
        END;
    END LOOP;

    IF subscription_count = 0 THEN
        RAISE NOTICE '    - No disabled subscriptions created';
    END IF;
END;
$$;

-- ============================================================================
-- Procedure to enable disabled subscriptions
-- ============================================================================
CREATE OR REPLACE PROCEDURE spock.enable_disabled_subscriptions(
    src_node_name text,
    src_dsn text,
    new_node_name text,
    new_node_dsn text,
    verb boolean
) LANGUAGE plpgsql AS $$
DECLARE
    rec      RECORD;
	sub_name text;
BEGIN
    RAISE NOTICE 'Phase 9: Enabling disabled subscriptions';

    -- Check if this is a 2-node scenario (only source and new node)
    IF (SELECT count(*) FROM temp_spock_nodes WHERE node_name != src_node_name AND node_name != new_node_name) = 0 THEN
        -- 2-node scenario: enable the disabled subscription from source to new node

		sub_name := spock.gen_sub_name(src_node_name, new_node_name);

        BEGIN
            CALL spock.enable_sub( new_node_dsn, sub_name, verb, true);

            -- Wait for the sync event that was captured when subscription was created
            -- This ensures the subscription starts replicating from the correct sync point
            DECLARE
                sync_lsn text;
                timeout_s integer := spock.sync_timeout();
                temp_table_exists boolean;
            BEGIN
                -- Check if temp_sync_lsns table exists
                SELECT EXISTS (
                    SELECT 1 FROM pg_tables
                    WHERE tablename = 'temp_sync_lsns'
                    AND schemaname = 'pg_temp'
                ) INTO temp_table_exists;

                IF temp_table_exists THEN
                    -- Get the stored sync LSN from when subscription was created
                    SELECT tsl.sync_lsn INTO sync_lsn
                    FROM temp_sync_lsns tsl
                    WHERE tsl.origin_node = src_node_name;

                    IF sync_lsn IS NOT NULL THEN
                        IF verb THEN
                            RAISE NOTICE '    OK: %', rpad('Using stored sync event from origin node ' || src_node_name || ' (LSN: ' || sync_lsn || ')...', 120, ' ');
                        END IF;

                        -- Wait for this sync event on the new node where the subscription exists
                        DECLARE
                            sync_ok text;
                        BEGIN
                            SELECT * INTO sync_ok FROM dblink(new_node_dsn,
                                format('CALL spock.wait_for_sync_event(true, %L, %L::pg_lsn, %s, true)',
                                       src_node_name, sync_lsn, timeout_s)) AS t(result text);

                            IF sync_ok IS NULL OR sync_ok::boolean IS NOT TRUE THEN
                                RAISE EXCEPTION 'wait_for_sync_event timed out for % on new node %', src_node_name, new_node_name;
                            END IF;

                            IF verb THEN
                                RAISE NOTICE '    OK: %', rpad('Sync event from ' || src_node_name || ' confirmed on new node ' || new_node_name, 120, ' ');
                            END IF;
                        END;
                    ELSE
                        RAISE NOTICE '    WARNING: %', rpad('No stored sync LSN found for ' || src_node_name || ', skipping sync wait', 120, ' ');
                    END IF;
                ELSE
                    -- For 2-node scenario where temp_sync_lsns doesn't exist, skip sync wait
                    IF verb THEN
                        RAISE NOTICE '    INFO: %', rpad('2-node scenario detected, no sync LSN table available, skipping sync wait', 120, ' ');
                    END IF;
                END IF;
            END;

            -- Verify subscription is replicating after enabling (2-node scenario)
            CALL spock.verify_subscription_replicating(
                new_node_dsn,
                spock.gen_sub_name(src_node_name, new_node_name),
                verb,
				180
            );

            RAISE NOTICE '    ✓ %', rpad('Enabling subscription ' || sub_name || '...', 120, ' ');
        EXCEPTION
            WHEN OTHERS THEN
                RAISE EXCEPTION '    ✗ %', rpad('Enabling subscription ' || sub_name || ' (error: ' || SQLERRM || ')', 120, ' ');
        END;
        RETURN;
    END IF;

    -- Multi-node scenario: original logic for other nodes
    -- Enable the initially disabled subscriptions
    BEGIN
        DECLARE
            subscription_count integer := 0;
        BEGIN
            FOR rec IN SELECT * FROM temp_spock_nodes LOOP
                IF rec.node_name = src_node_name THEN
                    CONTINUE;  -- Skip source node as it's handled separately
                END IF;
                IF rec.node_name = new_node_name THEN
                    CONTINUE;  -- Skip new node to avoid self-subscription
                END IF;

				sub_name := spock.gen_sub_name(rec.node_name, new_node_name);

                CALL spock.enable_sub(new_node_dsn, sub_name, verb, true);

                -- Wait for the sync event that was captured when subscription was created
                -- This ensures the subscription starts replicating from the correct sync point
                DECLARE
                    sync_lsn text;
                    timeout_s integer := spock.sync_timeout();
                BEGIN
                    -- Get the stored sync LSN from when subscription was created
                    SELECT tsl.sync_lsn INTO sync_lsn
                    FROM temp_sync_lsns tsl
                    WHERE tsl.origin_node = rec.node_name;

                    IF sync_lsn IS NOT NULL THEN
                        IF verb THEN
                            RAISE NOTICE '    OK: %', rpad('Using stored sync event from origin node ' || rec.node_name || ' (LSN: ' || sync_lsn || ')...', 120, ' ');
                        END IF;

                        -- Wait for this sync event on the new node where the subscription exists
                        DECLARE
                            sync_ok text;
                        BEGIN
                            SELECT * INTO sync_ok FROM dblink(new_node_dsn,
                                format('CALL spock.wait_for_sync_event(true, %L, %L::pg_lsn, %s, true)',
                                       rec.node_name, sync_lsn, timeout_s)) AS t(result text);

                            IF sync_ok IS NULL OR sync_ok::boolean IS NOT TRUE THEN
                                RAISE EXCEPTION 'wait_for_sync_event timed out for % on new node %', rec.node_name, new_node_name;
                            END IF;

                            IF verb THEN
                                RAISE NOTICE '    OK: %', rpad('Sync event from ' || rec.node_name || ' confirmed on new node ' || new_node_name, 120, ' ');
                            END IF;
                        END;
                    ELSE
                        RAISE NOTICE '    WARNING: %', rpad('No stored sync LSN found for ' || rec.node_name || ', skipping sync wait', 120, ' ');
                    END IF;
                END;

                -- Verify subscription is replicating after enabling
                CALL spock.verify_subscription_replicating(
                    new_node_dsn,
                    spock.gen_sub_name(rec.node_name, new_node_name),
                    verb,
					180
                );

                RAISE NOTICE '    ✓ %', rpad('Enabling subscription ' || sub_name || '...', 120, ' ');
                subscription_count := subscription_count + 1;
            END LOOP;

            IF subscription_count = 0 THEN
                RAISE NOTICE '    - %', rpad('No subscriptions to enable', 120, ' ');
            END IF;
        END;
    EXCEPTION
        WHEN OTHERS THEN
            RAISE EXCEPTION '%', '    ✗ Enabling disabled subscriptions...' || ' (error: ' || SQLERRM || ')';
    END;
END;
$$;

-- ============================================================================
-- Procedure to create a subscription from new node to source node only
-- ============================================================================
CREATE OR REPLACE PROCEDURE spock.create_sub_on_new_node_to_src_node(
    src_node_name   text,  -- Source node name
    src_dsn         text,  -- Source node DSN
    new_node_name   text,  -- New node name
    new_node_dsn    text,  -- New node DSN
    verb            boolean  -- Verbose flag
) LANGUAGE plpgsql AS $$
DECLARE
    rec                RECORD;
    subscription_count integer := 0;
	sub_name           text;
    repsets            text;
BEGIN
    RAISE NOTICE 'Phase 10: Creating subscriptions from all other nodes to new node';

    -- The new node is the provider for every subscription created here, and by
    -- this point its replication sets have been mirrored from the source.
    repsets := spock.node_repset_list(new_node_dsn);

    -- Get all existing nodes (excluding new node)
    CALL spock.get_spock_nodes(src_dsn, verb);

    -- For each existing node (excluding new node), create subscription TO the new node
    FOR rec IN SELECT * FROM temp_spock_nodes WHERE node_name != new_node_name LOOP
	    sub_name := spock.gen_sub_name(new_node_name, rec.node_name);
        BEGIN
            CALL spock.create_sub(
                rec.dsn,                                      -- Create on existing node
                sub_name, 									  -- sub_<new_node>_<existing_node>
                new_node_dsn,                                 -- Provider is new node
                repsets,                                      -- Replication sets (mirrored from source)
                false,                                        -- synchronize_structure
                false,                                        -- synchronize_data
                'ARRAY[]::text[]',                            -- forward_origins
                '0'::interval,                                -- apply_delay
                true,                                         -- enabled
                false,                                        -- force_text_transfer
                verb                                          -- verbose
            );
            RAISE NOTICE '    ✓ %', rpad('Creating subscription ' || sub_name || ' on node ' || rec.node_name || '...', 120, ' ');
            -- Allow the apply worker on rec.node time to come up and
            -- create its slot on new_node before the next iteration
            -- (and before subsequent zodan phases poke this state).
            PERFORM pg_sleep(5);
            subscription_count := subscription_count + 1;
        EXCEPTION
            WHEN OTHERS THEN
                RAISE EXCEPTION '    ✗ %', rpad('Creating subscription ' || sub_name || ' on node ' || rec.node_name || ' (error: ' || SQLERRM || ')', 120, ' ');
        END;
    END LOOP;

    IF subscription_count = 0 THEN
        RAISE NOTICE '    - No subscriptions created (no other nodes found)';
    ELSE
        RAISE NOTICE '    OK: Created % subscriptions from other nodes to new node', subscription_count;
    END IF;
END;
$$;

-- ============================================================================
-- Procedure to create new to source node subscription
--
-- Not part of the add_node sequence and not called by it: add_node's Phase 10
-- loops over every node except the new one, which covers the source as well.
-- Its notice therefore carries no phase number.
-- ============================================================================
CREATE OR REPLACE PROCEDURE spock.create_new_to_source_subscription(
    src_node_name text,
    src_dsn text,
    new_node_name text,
    new_node_dsn text,
    verb boolean
) LANGUAGE plpgsql AS $$
DECLARE
    sub_name text := spock.gen_sub_name(new_node_name, src_node_name);
    repsets  text;
BEGIN
    RAISE NOTICE 'Creating new to source node subscription';

    -- The new node is the provider, so the list comes from it.
    repsets := spock.node_repset_list(new_node_dsn);

    -- Create subscription from new node to source node (enabled with sync)
    CALL spock.create_sub(
        src_dsn,
        sub_name,
        new_node_dsn,
        repsets,
        false,   -- synchronize_structure
        false,   -- synchronize_data
        'ARRAY[]::text[]',
        '0'::interval,
        true,   -- enabled
        false,   -- force_text_transfer
        verb
    );
    RAISE NOTICE '    ✓ %', rpad('Creating subscription ' || sub_name || ' on node ' || new_node_name || '...', 120, ' ');
    PERFORM pg_sleep(5);
END;
$$;

-- ============================================================================
-- Procedure to create source to new node subscription
-- ============================================================================
CREATE OR REPLACE PROCEDURE spock.create_source_to_new_subscription(
    src_node_name   text,  -- Source node name
    src_dsn         text,  -- Source node DSN
    new_node_name   text,  -- New node name
    new_node_dsn    text,  -- New node DSN
    verb            boolean  -- Verbose flag
) LANGUAGE plpgsql AS $$
DECLARE
    sub_name text := spock.gen_sub_name(src_node_name, new_node_name);
    repsets  text;
BEGIN
    RAISE NOTICE 'Phase 5: Creating source to new node subscription';

    -- The source is the provider for this subscription.
    repsets := spock.node_repset_list(src_dsn);

    -- Create subscription from source to new node (enabled with sync)
    CALL spock.create_sub(
        new_node_dsn,                                 -- Create on new node
        sub_name, 									  -- sub_<new_node>_<src_node>
        src_dsn,                                      -- Provider is source node
        repsets,                                      -- Replication sets (all of the source's)
        true,                                         -- synchronize_structure
        true,                                         -- synchronize_data
        'ARRAY[]::text[]',                            -- forward_origins
        '0'::interval,                                -- apply_delay
        true,                                         -- enabled
        false,                                        -- force_text_transfer
        verb                                          -- verbose
    );
    RAISE NOTICE '    ✓ %', rpad('Creating subscription ' || sub_name || ' on node ' || new_node_name || '...', 120, ' ');
    PERFORM pg_sleep(5);
END;
$$;

-- ============================================================================
-- Procedure to trigger sync events on other nodes and wait on source
-- ============================================================================
CREATE OR REPLACE PROCEDURE spock.trigger_sync_on_other_nodes_and_wait_on_source(
    src_node_name text,  -- Source node name
    src_dsn text,        -- Source node DSN
    new_node_name text,  -- New node name
    new_node_dsn text,   -- New node DSN
    verb boolean         -- Verbose flag
) LANGUAGE plpgsql AS $$
DECLARE
    rec RECORD;
    sync_lsn pg_lsn;
    timeout_s integer := spock.sync_timeout();
    remotesql text;
BEGIN
    RAISE NOTICE 'Phase 4: Triggering sync events on other nodes and waiting on source';

    -- Check if this is a 2-node scenario (only source and new node)
    IF (SELECT count(*) FROM temp_spock_nodes WHERE node_name != src_node_name AND node_name != new_node_name) = 0 THEN
        RAISE NOTICE '    - No other nodes exist, skipping sync events';
        RETURN;
    END IF;

    -- Multi-node scenario: trigger sync on "other" nodes and wait on source
    FOR rec IN SELECT * FROM temp_spock_nodes WHERE node_name != src_node_name AND node_name != new_node_name LOOP
        -- Trigger sync event on "other" node
        BEGIN
            remotesql := 'SELECT spock.sync_event();';
            IF verb THEN
                RAISE NOTICE '    Remote SQL for sync event on %: %', rec.node_name, remotesql;
            END IF;

            SELECT * FROM dblink(rec.dsn, remotesql) AS t(lsn pg_lsn) INTO sync_lsn;
            RAISE NOTICE '    OK: %', rpad('Triggering sync event on node ' || rec.node_name || ' (LSN: ' || sync_lsn || ')...', 120, ' ');
        EXCEPTION
            WHEN OTHERS THEN
                RAISE EXCEPTION '    ✗ %', rpad('Triggering sync event on node ' || rec.node_name || ' (error: ' || SQLERRM || ')', 120, ' ');
        END;

        -- Wait for sync event on source node
        BEGIN
            remotesql := format('CALL spock.wait_for_sync_event(true, %L, %L::pg_lsn, %s, true);',
                               rec.node_name, sync_lsn, timeout_s);
            IF verb THEN
                RAISE NOTICE '    Remote SQL for waiting sync event: %', remotesql;
            END IF;

            DECLARE
                sync_ok text;
            BEGIN
                SELECT * INTO sync_ok FROM dblink(src_dsn, remotesql) AS t(result text);
                IF sync_ok IS NULL OR sync_ok::boolean IS NOT TRUE THEN
                    RAISE EXCEPTION 'wait_for_sync_event timed out for % on source node %', rec.node_name, src_node_name;
                END IF;
                RAISE NOTICE '    OK: %', rpad('Sync event from ' || rec.node_name || ' confirmed on source node ' || src_node_name, 120, ' ');
            END;
        EXCEPTION
            WHEN OTHERS THEN
                RAISE EXCEPTION '    ✗ %', rpad('Waiting for sync event from ' || rec.node_name || ' on source node ' || src_node_name || ' (error: ' || SQLERRM || ')', 120, ' ');
        END;
    END LOOP;
END;
$$;

-- ============================================================================
-- Procedure to check commit timestamp and advance replication slot
-- ============================================================================
CREATE OR REPLACE PROCEDURE spock.check_commit_timestamp_and_advance_slot(
    src_node_name text,  -- Source node name
    src_dsn text,        -- Source node DSN
    new_node_name text,  -- New node name
    new_node_dsn text,   -- New node DSN
    verb boolean         -- Verbose flag
) LANGUAGE plpgsql AS $$
DECLARE
    rec RECORD;
    slot_lsn pg_lsn;
    slot_name text;
    dbname text;
    remotesql text;
    src_rec RECORD;
    src_commit_lsn pg_lsn;
    src_slot_name text;
    src_dbname text;
    current_lsn pg_lsn;
    target_lsn pg_lsn;
    v_sub_name name;
    v_subid oid;
    v_pending_sync integer;
    v_wait_started timestamptz;
    v_sub_status text;
BEGIN
    RAISE NOTICE 'Phase 7: Checking commit timestamp and advancing replication slot';

    -- Wait for src->new COPY to complete so resume_lsn is written to spock.progress.
    v_sub_name := spock.gen_sub_name(src_node_name, new_node_name)::name;
    RAISE NOTICE '    - Waiting for subscription % to reach READY...', v_sub_name;
    BEGIN
        -- Avoid rare hangs in C-level sub_wait_for_sync by using a bounded SQL loop.
        SELECT sub_id INTO v_subid
        FROM spock.subscription
        WHERE sub_name = v_sub_name;

        IF v_subid IS NULL THEN
            RAISE WARNING '    - Subscription % not found on new node; continuing', v_sub_name;
        ELSE
            v_wait_started := clock_timestamp();
            LOOP
                SELECT count(*) INTO v_pending_sync
                FROM spock.local_sync_status
                WHERE sync_subid = v_subid
                  AND sync_status NOT IN ('y', 'r');

                SELECT status INTO v_sub_status
                FROM spock.sub_show_status()
                WHERE subscription_name = v_sub_name;

                IF v_pending_sync = 0 THEN
                    RAISE NOTICE '    - Subscription % is READY', v_sub_name;
                    EXIT;
                END IF;

                -- Advisory wait: report and carry on rather than failing the
                -- whole operation. Only the budget comes from the GUC.
                IF clock_timestamp() - v_wait_started
                       > make_interval(secs => spock.sync_timeout()) THEN
                    RAISE WARNING '    - Timed out waiting for % to become READY (pending rows: %, status: %); continuing',
                        v_sub_name, v_pending_sync, coalesce(v_sub_status, '<unknown>');
                    EXIT;
                END IF;

                PERFORM pg_sleep(1);
            END LOOP;
        END IF;
    EXCEPTION
        WHEN OTHERS THEN
            RAISE WARNING '    - READY wait for subscription % failed: %; proceeding anyway', v_sub_name, SQLERRM;
            RAISE WARNING '    - Current subscription status snapshot: %',
                coalesce((SELECT string_agg(subscription_name || ':' || status, ', ')
                          FROM spock.sub_show_status()), '<none>');
    END;

    -- Check src->new slot; only advance if it is NOT active (defensive).
    BEGIN
        RAISE NOTICE '    - Checking source-to-new subscription slot...';

        -- Get source node info and extract dbname
        FOR src_rec IN SELECT * FROM temp_spock_nodes WHERE node_name = src_node_name LOOP
            SELECT spock.extract_dbname_from_dsn(src_rec.dsn) INTO src_dbname;
            IF src_dbname IS NOT NULL THEN
                src_dbname := TRIM(BOTH '''' FROM src_dbname);
            END IF;
            IF src_dbname IS NULL THEN
                src_dbname := 'pgedge';
            END IF;

            -- Generate slot name: spk_<dbname>_<src>_sub_<src>_<new>
            src_slot_name := spock.spock_gen_slot_name(
                src_dbname, src_node_name,
                spock.gen_sub_name(src_node_name, new_node_name));

            RAISE NOTICE '    Looking for slot % on source node', src_slot_name;

            -- Check if slot exists on source node and whether it is active
            DECLARE
                v_slot_active boolean;
            BEGIN
                remotesql := format(
                    'SELECT restart_lsn, active FROM pg_replication_slots WHERE slot_name = %L',
                    src_slot_name);
                SELECT * FROM dblink(src_dsn, remotesql)
                    AS t(lsn pg_lsn, active boolean)
                    INTO current_lsn, v_slot_active;

                IF current_lsn IS NULL THEN
                    RAISE NOTICE '    Slot % not found on source node', src_slot_name;
                ELSIF v_slot_active THEN
                    -- Subscription is running; slot and origin managed by the apply worker
                    RAISE NOTICE '    Slot % is active (subscription running) — no advance needed', src_slot_name;
                ELSE
                    -- Slot exists but is not active (unusual).  Advance defensively.
                    RAISE NOTICE '    Slot % found at LSN % (inactive)', src_slot_name, current_lsn;

                    SELECT p.remote_commit_lsn INTO target_lsn
                    FROM spock.progress p
                    JOIN spock.node n ON n.node_id = p.remote_node_id
                    WHERE n.node_name = src_node_name;

                    IF target_lsn IS NOT NULL AND target_lsn > current_lsn THEN
                        RAISE NOTICE '    Snapshot LSN for %: %', src_node_name, target_lsn;
                        remotesql := format('SELECT pg_replication_slot_advance(%L, %L::pg_lsn)', src_slot_name, target_lsn);
                        PERFORM * FROM dblink(src_dsn, remotesql) AS t(result text);
                        RAISE NOTICE '    OK: Advanced slot % on source node from % to %', src_slot_name, current_lsn, target_lsn;

                        IF NOT EXISTS (
                            SELECT 1 FROM pg_replication_origin WHERE roname = src_slot_name
                        ) THEN
                            RAISE WARNING '    Origin % not found on new node; creating it now', src_slot_name;
                            PERFORM pg_replication_origin_create(src_slot_name);
                        END IF;
                        PERFORM pg_replication_origin_advance(src_slot_name, target_lsn);
                        RAISE NOTICE '    OK: Advanced replication origin % on new node to %', src_slot_name, target_lsn;
                    ELSE
                        RAISE NOTICE '    Slot % already at or beyond snapshot LSN', src_slot_name;
                    END IF;
                END IF;
            END;
        END LOOP;
    EXCEPTION
        WHEN OTHERS THEN
            RAISE WARNING 'Could not check source-to-new slot: %', SQLERRM;
    END;

    -- Check if this is a 2-node scenario (only source and new node)
    IF (SELECT count(*) FROM temp_spock_nodes WHERE node_name != src_node_name AND node_name != new_node_name) = 0 THEN
        RAISE NOTICE '    - No other nodes exist, skipping additional commitment checks';
        RETURN;
    END IF;

    -- Multi-node scenario: check commit timestamp for "other" nodes to new node
    FOR rec IN SELECT * FROM temp_spock_nodes WHERE node_name != src_node_name AND node_name != new_node_name LOOP
        BEGIN
            IF EXISTS (SELECT 1 FROM pg_class WHERE relname = 'temp_sync_lsns' AND relpersistence = 't') THEN
                -- Get the stored slot LSN from when the slot was created
                SELECT tsl.slot_lsn INTO slot_lsn
                FROM temp_sync_lsns tsl
                WHERE tsl.origin_node = rec.node_name;

                IF slot_lsn IS NOT NULL THEN
                    RAISE NOTICE '    OK: %', rpad('Found slot LSN for ' || rec.node_name || ' (LSN: ' || slot_lsn || ')...', 120, ' ');
                ELSE
                    RAISE NOTICE '    - %', rpad('No slot LSN found for ' || rec.node_name || '->' || new_node_name, 120, ' ');
                    CONTINUE;
                END IF;
            END IF;
        EXCEPTION
            WHEN OTHERS THEN
                -- Can't continue the process because slot_lsn is needed
                -- to advance LR slot properly.
                RAISE EXCEPTION '    ✗ %', rpad('Checking slot LSN for ' || rec.node_name || '->' || new_node_name || ' (error: ' || SQLERRM || ')', 120, ' ');
        END;

        -- Advance replication slot based on commit timestamp
        BEGIN
            -- Extract dbname and handle both quoted and unquoted values
            SELECT spock.extract_dbname_from_dsn(rec.dsn) INTO dbname;

            -- Remove single quotes if present
            IF dbname IS NOT NULL THEN
                dbname := TRIM(BOTH '''' FROM dbname);
            END IF;

            -- Remove single quotes if present
            IF dbname IS NOT NULL THEN
                dbname := TRIM(BOTH '''' FROM dbname);
            END IF;
            IF dbname IS NULL THEN dbname := 'pgedge'; END IF;

			slot_name := spock.spock_gen_slot_name(dbname, rec.node_name, spock.gen_sub_name(rec.node_name, new_node_name));

            -- First check if slot exists and get current LSN
            remotesql := format('SELECT restart_lsn FROM pg_replication_slots WHERE slot_name = %L', slot_name);
            IF verb THEN
                RAISE NOTICE '    Remote SQL for slot check: %', remotesql;
            END IF;

            DECLARE
                current_lsn pg_lsn;
                target_lsn pg_lsn;
            BEGIN
                SELECT * FROM dblink(rec.dsn, remotesql) AS t(lsn pg_lsn) INTO current_lsn;

                IF current_lsn IS NULL THEN
                    RAISE NOTICE '    - Slot % does not exist, skipping advancement', slot_name;
                    CONTINUE;
                END IF;

                -- Advance the slot to resume_lsn: the last commit from this node
                -- that N1 had applied at snapshot time (stored in N3's spock.progress).
                SELECT p.remote_commit_lsn INTO target_lsn
                FROM spock.progress p
                JOIN spock.node n ON n.node_id = p.remote_node_id
                WHERE n.node_name = rec.node_name;

                IF target_lsn IS NULL THEN
                    RAISE NOTICE '    WARNING: No spock.progress entry for %, falling back to pg_current_wal_lsn()', rec.node_name;
                    remotesql := 'SELECT pg_current_wal_lsn()';
                    SELECT * FROM dblink(rec.dsn, remotesql) AS t(lsn pg_lsn) INTO target_lsn;
                END IF;

                IF target_lsn IS NULL OR target_lsn <= current_lsn THEN
                    RAISE NOTICE '    - Slot % already at or beyond resume_lsn LSN (current: %, target: %)', slot_name, current_lsn, target_lsn;
                    CONTINUE;
                END IF;

                -- Advance the slot
                remotesql := format('SELECT pg_replication_slot_advance(%L, %L::pg_lsn)', slot_name, target_lsn);
                IF verb THEN
                    RAISE NOTICE '    Remote SQL for slot advancement: %', remotesql;
                END IF;

                PERFORM * FROM dblink(rec.dsn, remotesql) AS t(result text);
                RAISE NOTICE '    OK: %', rpad('Advanced slot ' || slot_name || ' from ' || current_lsn || ' to ' || target_lsn, 120, ' ');

                -- Advance the replication origin on the new node (subscriber side)
                -- directly, not via dblink; the origin is local to the new node.
                IF NOT EXISTS (
                    SELECT 1 FROM pg_replication_origin WHERE roname = slot_name
                ) THEN
                    RAISE WARNING '    Origin % not found on new node; creating it now (was it created in Phase 3?)', slot_name;
                    PERFORM pg_replication_origin_create(slot_name);
                END IF;
                PERFORM pg_replication_origin_advance(slot_name, target_lsn);
                RAISE NOTICE '    OK: %', rpad('Advanced replication origin ' || slot_name || ' on new node to ' || target_lsn, 120, ' ');
            END;
        EXCEPTION
            WHEN OTHERS THEN
                -- Can't continue the process because of an error during the
                -- slot advancement operation
                RAISE EXCEPTION '    ✗ %', rpad('Advancing slot ' || slot_name || ' to LSN ' || slot_lsn || ' (error: ' || SQLERRM || ')', 120, ' ');
        END;
    END LOOP;
END;
$$;

-- ============================================================================
-- Procedure to trigger sync on source node and wait for it on new node using sync_event and wait_for_sync_event
-- ============================================================================
CREATE OR REPLACE PROCEDURE spock.trigger_source_sync_and_wait_on_new_node(
    src_node_name text,
    src_dsn text,
    new_node_name text,
    new_node_dsn text,
    verb boolean,
    sync_check_on_new_node boolean DEFAULT false
) LANGUAGE plpgsql AS $$
DECLARE
    remotesql text;
    sync_lsn pg_lsn;
    timeout_s integer := spock.sync_timeout();
BEGIN
    RAISE NOTICE 'Phase 6: Triggering sync on source node and waiting on new node';

    -- Trigger sync event on source node and wait for it on new node
    BEGIN
        remotesql := 'SELECT spock.sync_event();';
        IF verb THEN
            RAISE NOTICE '    Remote SQL for sync_event on source node %: %', src_node_name, remotesql;
        END IF;
        SELECT * FROM dblink(src_dsn, remotesql) AS t(lsn pg_lsn) INTO sync_lsn;
        RAISE NOTICE '    OK: %', rpad('Triggered sync_event on source node ' || src_node_name || ' (LSN: ' || sync_lsn || ')...', 120, ' ');
    EXCEPTION
        WHEN OTHERS THEN
            RAISE EXCEPTION '    ✗ %', rpad('Triggering sync_event on source node ' || src_node_name || ' (error: ' || SQLERRM || ')', 120, ' ');
    END;

    -- Wait for sync event on new node
    BEGIN
        remotesql := format('CALL spock.wait_for_sync_event(true, %L, %L::pg_lsn, %s, true);', src_node_name, sync_lsn, timeout_s);
        IF verb THEN
            RAISE NOTICE '    Remote SQL for wait_for_sync_event on new node %: %', new_node_name, remotesql;
        END IF;
        DECLARE
            sync_ok text;
        BEGIN
            SELECT * INTO sync_ok FROM dblink(new_node_dsn, remotesql) AS t(result text);
            IF sync_ok IS NULL OR sync_ok::boolean IS NOT TRUE THEN
                RAISE EXCEPTION 'wait_for_sync_event timed out for % on new node %', src_node_name, new_node_name;
            END IF;
            RAISE NOTICE '    OK: %', rpad('Sync event from ' || src_node_name || ' confirmed on new node ' || new_node_name, 120, ' ');
        END;
    EXCEPTION
        WHEN OTHERS THEN
            RAISE EXCEPTION '    ✗ %', rpad('Unable to wait for sync event from ' || src_node_name || ' on new node ' || new_node_name || ' (error: ' || SQLERRM || ')', 120, ' ');
    END;
END;
$$;

-- ============================================================================
-- Procedure to present final cluster state
-- ============================================================================
CREATE OR REPLACE PROCEDURE spock.present_final_cluster_state(
    src_dsn            text,
    initial_node_count integer,
    verb boolean DEFAULT false
) LANGUAGE plpgsql AS $$
DECLARE
    node_rec RECORD;
    sub_rec  RECORD;
    rec RECORD;
    wait_count integer := 0;
    wait_started timestamptz := clock_timestamp();
    -- A budget in seconds, not a number of attempts: each iteration costs a
    -- dblink round trip per node on top of the sleep, so on a loaded cluster an
    -- attempt count is not a time bound at all.
    wait_budget integer := spock.sync_timeout();
BEGIN
    -- Let remote subscriptions update their subscription's state.
    COMMIT;

    -- Phase 10: Presenting final cluster state
    RAISE NOTICE 'Phase 11: Presenting final cluster state';

    -- Wait for replication to be active
    RAISE NOTICE '    Waiting for replication to be active...';
    LOOP
        wait_count := wait_count + 1;
        sub_rec := NULL;

        -- Pass through all subscriptions and detect if some of them is not
		-- replicating.
        FOR node_rec IN SELECT dsn
            FROM dblink(src_dsn, '
                        SELECT i.if_dsn
                        FROM spock.node n JOIN spock.node_interface i
                        ON n.node_id = i.if_nodeid ORDER BY n.node_name'
            ) AS t(dsn text)
        LOOP
            SELECT sub_name, status INTO sub_rec
                FROM dblink(node_rec.dsn, '
                            SELECT subscription_name, status
                            FROM spock.sub_show_status()
                            WHERE status <> ''replicating''
                            ORDER BY subscription_name LIMIT 1'
                ) AS t(sub_name text, status text);
            EXIT WHEN sub_rec IS NOT NULL;
        END LOOP;

        IF sub_rec IS NULL THEN
            RAISE NOTICE '    OK: Replication is active';
            EXIT;
        ELSIF sub_rec.status IN ('disabled', 'down') THEN
            RAISE EXCEPTION 'Subscription % entered terminal state % while waiting for replication to become active',
                sub_rec.sub_name, sub_rec.status;
        ELSIF clock_timestamp() - wait_started
                  > make_interval(secs => wait_budget) THEN
            RAISE EXCEPTION 'Timeout waiting for subscription % to become active after % seconds (current status: %)',
                sub_rec.sub_name, wait_budget, sub_rec.status;
        ELSE
            RAISE NOTICE '    Waiting for replication... (subscription: %, status: %, attempt %, elapsed %)',
                sub_rec.sub_name, sub_rec.status, wait_count,
                clock_timestamp() - wait_started;
            PERFORM pg_sleep(1);
        END IF;
    END LOOP;

    -- Print nodes in psql-style table format
    RAISE NOTICE '';
    RAISE NOTICE 'Current Spock Nodes:';
    RAISE NOTICE ' node_id | node_name | location | country | info';
    RAISE NOTICE '---------+-----------+----------+---------+------';

    IF verb THEN
        RAISE NOTICE '[QUERY] SELECT node_id, node_name, location, country, info FROM spock.node ORDER BY node_id';
    END IF;
    FOR rec IN SELECT node_id, node_name, location, country, info FROM spock.node ORDER BY node_id LOOP
        RAISE NOTICE ' % | % | % | % | %',
            rpad(rec.node_id::text, 7, ' '),
            rpad(rec.node_name, 9, ' '),
            rpad(COALESCE(rec.location, ''), 9, ' '),
            rpad(COALESCE(rec.country, ''), 7, ' '),
            COALESCE(rec.info::text, '');
    END LOOP;
    RAISE NOTICE '';

    -- Show subscription status
    RAISE NOTICE 'Subscription Status:';
    IF verb THEN
        RAISE NOTICE '[QUERY] SELECT * FROM spock.sub_show_status()';
    END IF;
    FOR rec IN SELECT * FROM spock.sub_show_status() LOOP
        RAISE NOTICE '  %: % (provider: %)', rec.subscription_name, rec.status, rec.provider_node;
    END LOOP;
    RAISE NOTICE '';
END;
$$;

-- ============================================================================
-- Procedure to monitor replication lag
-- ============================================================================
CREATE OR REPLACE PROCEDURE spock.monitor_replication_lag(
    src_node_name text,
    new_node_name text,
    new_node_dsn text,
    verb boolean
) LANGUAGE plpgsql AS $$
BEGIN
    -- Phase 11: Monitor replication lag
    RAISE NOTICE 'Phase 12: Monitoring replication lag';
    CALL spock.monitor_lag_with_dblink(src_node_name, new_node_name, new_node_dsn, verb);
END;
$$;

/* =============================================================================
 * Procedure: add_node
 *
 * Description:
 *   Adds a new node to an existing Spock-based logical replication cluster.
 *   The procedure:
 *     - Creates the new node
 *     - Sets up replication subscriptions between the new node and existing nodes
 *     - Creates necessary replication slots
 *     - Triggers synchronization events
 *     - Advances replication slots to ensure consistent replication
 *     - Enables all subscriptions for bidirectional replication
 *
 * Parameters:
 *   src_node_name     - Name of the source node initiating the add operation
 *   src_dsn           - DSN of the source node
 *   new_node_name     - Name of the node to be added
 *   new_node_dsn      - DSN of the new node
 *   verb			   - verbose, show more information during the process (default: false)
 *   new_node_location - Location (default: 'NY')
 *   new_node_country  - Country (default: 'USA')
 *   new_node_info     - JSONB object with additional metadata (default: '{}')
 *
 * Notes:
 *   - Assumes `create_node`, `create_sub`, `enable_sub`, `create_replication_slot`,
 *     `sync_event`, `wait_for_sync_event`, `get_commit_timestamp`,
 *     and `advance_replication_slot` are already defined and available.
 *   - Ensures minimal interruption and consistency using sync + slot advance.
 * =============================================================================
 */
CREATE OR REPLACE PROCEDURE spock.add_node(
    src_node_name text,
    src_dsn text,
    new_node_name text,
    new_node_dsn text,
    verb boolean DEFAULT false,
    new_node_location text DEFAULT 'NY',
    new_node_country text DEFAULT 'USA',
    new_node_info jsonb DEFAULT '{}'::jsonb
)
LANGUAGE plpgsql
AS
$$
DECLARE
    initial_node_count integer;
    repset_summary     text;
BEGIN
    -- Phase 0: Check Spock version compatibility across all nodes
    -- Example: Ensure all nodes are running the same Spock version before proceeding
    CALL spock.check_spock_version_compatibility(src_dsn, new_node_dsn, verb);

    -- Phase 1: Verify prerequisites for source and new node.
    -- Example: Ensure n1 (source) and n4 (new) are ready before adding n4 to cluster n1,n2,n3.
    CALL spock.verify_node_prerequisites(src_node_name, src_dsn, new_node_name, new_node_dsn, verb);

    -- Phase 2: Create node objects in the cluster.
    -- Example: Register n4 as a new node alongside n1, n2, n3.
    CALL spock.create_nodes_only(src_node_name, src_dsn, new_node_name, new_node_dsn, verb, new_node_location, new_node_country, new_node_info, initial_node_count);

    -- Phase 3: Create disabled subscriptions and replication slots.
    -- Example: Prepare n4 for replication but keep subscriptions disabled initially.
    CALL spock.create_disable_subscriptions_and_slots(src_node_name, src_dsn, new_node_name, new_node_dsn, verb);

    -- Phase 4: Trigger sync events on other nodes and wait on source.
    -- Example: Sync n2 and n3, then wait for n1 to acknowledge before proceeding with n4.
    CALL spock.trigger_sync_on_other_nodes_and_wait_on_source(src_node_name, src_dsn, new_node_name, new_node_dsn, verb);

    -- Phase 5: Create subscription from source to new node.
    -- Example: Set up n1 to replicate to n4.
    CALL spock.create_source_to_new_subscription(src_node_name, src_dsn, new_node_name, new_node_dsn, verb);

    -- Phase 6: Trigger sync on source node and wait on new node.
    -- Example: Ensure n1 and n4 are fully synchronized before continuing.
    CALL spock.trigger_source_sync_and_wait_on_new_node(src_node_name, src_dsn, new_node_name, new_node_dsn, verb, true);

    -- Phase 7: Check commit timestamp and advance replication slot.
    -- Example: Confirm n4 is caught up to n1's latest changes.
    CALL spock.check_commit_timestamp_and_advance_slot(src_node_name, src_dsn, new_node_name, new_node_dsn, verb);

    -- Phase 8: Mirror the source node's replication sets onto the new node.
    -- Runs after the structure and data sync (the tables have to exist) and
    -- before the new node starts acting as a provider in Phase 10, otherwise it
    -- would send only whatever AutoDDL happened to route into the built-in
    -- sets during the restore.
    --
    -- Invoked over dblink rather than called directly even though add_node is
    -- already running on the new node: see sync_repsets_from_source() for why
    -- holding its catalog locks for the rest of this transaction deadlocks
    -- against the dblink session that Phase 9 enables subscriptions from.
    RAISE NOTICE 'Phase 8: Mirroring replication sets from source node';
    SELECT summary INTO repset_summary
      FROM dblink(new_node_dsn,
               format('SELECT spock.sync_repsets_from_source(%L, %L, %L::boolean)',
                      src_dsn, new_node_name, verb)) AS t(summary text);
    RAISE NOTICE '    OK: %', rpad(repset_summary, 120, ' ');

    -- Report, but do not fail on, replication sets that moved on the source
    -- while this join was running. A join can take hours; refusing to finish
    -- because somebody added a table in the meantime would throw away all of
    -- that work, when the new node is otherwise complete and the fix is one
    -- repset_add_table away. Name the exact entries so it can be applied.
    DECLARE
        drift_rec  RECORD;
        drift_list text := '';
    BEGIN
        FOR drift_rec IN
            SELECT side, entry FROM spock.repset_diff(src_dsn, new_node_dsn)
        LOOP
            drift_list := drift_list || E'\n      ' || drift_rec.side || ': ' || drift_rec.entry;
        END LOOP;

        IF drift_list <> '' THEN
            RAISE WARNING 'Replication sets on % no longer match %: the source changed while the node was joining.%',
                new_node_name, src_node_name, drift_list;
            RAISE WARNING 'Reconcile with spock.repset_add_table()/spock.repset_remove_table() on %; add_node has otherwise completed.',
                new_node_name;
        ELSE
            RAISE NOTICE '    OK: %', rpad('Replication sets on ' || new_node_name ||
                                           ' match ' || src_node_name, 120, ' ');
        END IF;
    END;

    -- Phase 9: Enable previously disabled subscriptions.
    -- Example: Activate replication paths for n4.
    CALL spock.enable_disabled_subscriptions(src_node_name, src_dsn, new_node_name, new_node_dsn, verb);

    -- Phase 10: Create subscription from new node to source node.
    -- Example: Set up n4 to replicate back to n1 for bidirectional sync.
    CALL spock.create_sub_on_new_node_to_src_node(src_node_name, src_dsn, new_node_name, new_node_dsn, verb);

    -- Phase 11: Present final cluster state.
    -- Example: Show n1, n2, n3, n4 as fully connected and synchronized.
    CALL spock.present_final_cluster_state(src_dsn, initial_node_count, verb);

    -- Phase 12: Monitor replication lag.
    -- Example: Check that n4 is keeping up with n1, n2, n3 after joining.
    CALL spock.monitor_replication_lag(src_node_name, new_node_name, new_node_dsn, verb);

    -- Phase 13: Show comprehensive node status across all nodes.
    -- Example: Display all nodes in n1, n2, n3, n4, n5 cluster.
    CALL spock.show_all_nodes(src_dsn, verb);

    -- Phase 14: Show comprehensive subscription status across all nodes.
    -- Example: Display status of all subscriptions in n1, n2, n3, n4, n5 cluster.
    CALL spock.show_all_subscription_status(src_dsn, verb);
END;
$$;

-- ============================================================================
-- Procedure: health_check
-- Purpose : Validate cluster health before or after ZODAN node addition
--           Similar to pg_upgrade -c (check) option
-- Arguments:
--   src_node_name  - Source node name
--   src_dsn        - Source node DSN
--   new_node_name  - New node name (optional for cluster-wide check)
--   new_node_dsn   - New node DSN (optional for cluster-wide check)
--   check_type     - Check type: 'pre' (before add_node) or 'post' (after add_node)
--   verb           - Verbose output flag
-- Usage    :
--   -- Pre-check before adding a node
--   CALL spock.health_check('n1', 'host=localhost dbname=pgedge port=5431 user=pgedge password=pgedge',
--                           'n3', 'host=localhost dbname=pgedge port=5433 user=pgedge password=pgedge',
--                           'pre', true);
--
--   -- Post-check after adding a node
--   CALL spock.health_check('n1', 'host=localhost dbname=pgedge port=5431 user=pgedge password=pgedge',
--                           'n3', 'host=localhost dbname=pgedge port=5433 user=pgedge password=pgedge',
--                           'post', true);
--
--   -- Cluster-wide check (no new node)
--   CALL spock.health_check('n1', 'host=localhost dbname=pgedge port=5431 user=pgedge password=pgedge',
--                           NULL, NULL, 'pre', true);
-- ============================================================================
CREATE OR REPLACE PROCEDURE spock.health_check(
    src_node_name text,
    src_dsn text,
    new_node_name text DEFAULT NULL,
    new_node_dsn text DEFAULT NULL,
    check_type text DEFAULT 'pre',
    verb boolean DEFAULT false
) LANGUAGE plpgsql AS $$
DECLARE
    checks_passed integer := 0;
    checks_failed integer := 0;
    check_result text;
    node_rec RECORD;
    remotesql text;
    result_value text;
    result_count integer;
    src_version text;
    new_version text;
    sub_count text;
    user_table_count text;
    new_lolor_installed boolean;
    lolor_row_count bigint;
    src_lolor_repset_tables integer;
BEGIN
    RAISE NOTICE '';
    RAISE NOTICE '================================================================================';
    RAISE NOTICE 'ZODAN CLUSTER HEALTH CHECK (%-CHECK)', upper(check_type);
    RAISE NOTICE '================================================================================';

    -- ========================================================================
    -- Check 1: Spock version compatibility (if new node provided)
    -- ========================================================================
    IF new_node_dsn IS NOT NULL THEN
        BEGIN
            -- Call existing check_spock_version_compatibility procedure
            CALL spock.check_spock_version_compatibility(src_dsn, new_node_dsn, false);
            RAISE NOTICE 'PASS: Spock version compatibility check';
            checks_passed := checks_passed + 1;
        EXCEPTION WHEN OTHERS THEN
            RAISE NOTICE 'FAIL: Spock version compatibility - %', SQLERRM;
            checks_failed := checks_failed + 1;
        END;
    END IF;

    -- ========================================================================
    -- Check 2: Node connectivity - Source node
    -- ========================================================================
    BEGIN
        remotesql := 'SELECT 1';
        SELECT * FROM dblink(src_dsn, remotesql) AS t(result integer) INTO result_value;
        RAISE NOTICE 'PASS: Source node % connectivity', src_node_name;
        checks_passed := checks_passed + 1;
    EXCEPTION WHEN OTHERS THEN
        RAISE NOTICE 'FAIL: Source node % connectivity - %', src_node_name, SQLERRM;
        checks_failed := checks_failed + 1;
    END;

    -- ========================================================================
    -- Check 3: Node connectivity - New node (if provided)
    -- ========================================================================
    IF new_node_dsn IS NOT NULL THEN
        BEGIN
            remotesql := 'SELECT 1';
            SELECT * FROM dblink(new_node_dsn, remotesql) AS t(result integer) INTO result_value;
            RAISE NOTICE 'PASS: New node % connectivity', new_node_name;
            checks_passed := checks_passed + 1;
        EXCEPTION WHEN OTHERS THEN
            RAISE NOTICE 'FAIL: New node % connectivity - %', new_node_name, SQLERRM;
            checks_failed := checks_failed + 1;
        END;
    END IF;

    -- ========================================================================
    -- Check 4: Spock extension installation - Source node
    -- ========================================================================
    BEGIN
        remotesql := 'SELECT extversion FROM pg_extension WHERE extname = ''spock''';
        SELECT * FROM dblink(src_dsn, remotesql) AS t(version text) INTO src_version;
        IF src_version IS NOT NULL THEN
            RAISE NOTICE 'PASS: Spock extension on source node (version %)', src_version;
            checks_passed := checks_passed + 1;
        ELSE
            RAISE NOTICE 'FAIL: Spock extension not installed on source node';
            checks_failed := checks_failed + 1;
        END IF;
    EXCEPTION WHEN OTHERS THEN
        RAISE NOTICE 'FAIL: Spock extension check on source - %', SQLERRM;
        checks_failed := checks_failed + 1;
    END;

    -- ========================================================================
    -- Check 5: Spock extension installation - New node (if provided)
    -- ========================================================================
    IF new_node_dsn IS NOT NULL THEN
        BEGIN
            remotesql := 'SELECT extversion FROM pg_extension WHERE extname = ''spock''';
            SELECT * FROM dblink(new_node_dsn, remotesql) AS t(version text) INTO new_version;
            IF new_version IS NOT NULL THEN
                RAISE NOTICE 'PASS: Spock extension on new node (version %)', new_version;
                checks_passed := checks_passed + 1;
            ELSE
                RAISE NOTICE 'FAIL: Spock extension not installed on new node';
                checks_failed := checks_failed + 1;
            END IF;
        EXCEPTION WHEN OTHERS THEN
            RAISE NOTICE 'FAIL: Spock extension check on new node - %', SQLERRM;
            checks_failed := checks_failed + 1;
        END;
    END IF;

    -- ========================================================================
    -- Check 6: Cluster node enumeration
    -- ========================================================================
    BEGIN
        remotesql := 'SELECT count(*) FROM spock.node';
        SELECT * FROM dblink(src_dsn, remotesql) AS t(node_count integer) INTO result_count;
        RAISE NOTICE 'PASS: Cluster has % nodes', result_count;
        checks_passed := checks_passed + 1;
    EXCEPTION WHEN OTHERS THEN
        RAISE NOTICE 'FAIL: Cluster node enumeration - %', SQLERRM;
        checks_failed := checks_failed + 1;
    END;

    -- ========================================================================
    -- Check 7: Active subscriptions on each node
    -- ========================================================================
    BEGIN
        FOR node_rec IN
            SELECT node_name, if_dsn
            FROM dblink(src_dsn,
                'SELECT n.node_name, i.if_dsn FROM spock.node n JOIN spock.node_interface i ON n.node_id = i.if_nodeid'
            ) AS t(node_name text, if_dsn text)
        LOOP
            BEGIN
                remotesql := 'SELECT count(*) FROM spock.subscription WHERE sub_enabled = true';
                SELECT * FROM dblink(node_rec.if_dsn, remotesql) AS t(sub_count text) INTO sub_count;
                RAISE NOTICE 'PASS: Node % has % active subscriptions', node_rec.node_name, COALESCE(sub_count, '0');
                checks_passed := checks_passed + 1;
            EXCEPTION WHEN OTHERS THEN
                RAISE NOTICE 'FAIL: Node % subscription check - %', node_rec.node_name, SQLERRM;
                checks_failed := checks_failed + 1;
            END;
        END LOOP;
    EXCEPTION WHEN OTHERS THEN
        RAISE NOTICE 'FAIL: Cluster subscription check - %', SQLERRM;
        checks_failed := checks_failed + 1;
    END;

    -- ========================================================================
    -- Check 8: Database prerequisites (pre-check only)
    -- ========================================================================
    IF check_type = 'pre' AND new_node_dsn IS NOT NULL THEN
        -- Check 8a: lolor state. Having lolor installed is OK, but the lolor
        -- tables must be empty, and if the source replicates lolor tables the
        -- new node must have lolor installed.
        BEGIN
            remotesql := 'SELECT EXISTS(SELECT 1 FROM pg_catalog.pg_extension WHERE extname = ''lolor'')';
            SELECT * FROM dblink(new_node_dsn, remotesql) AS t(installed boolean) INTO new_lolor_installed;

            IF new_lolor_installed THEN
                remotesql := 'SELECT (SELECT count(*) FROM lolor.pg_largeobject) + (SELECT count(*) FROM lolor.pg_largeobject_metadata)';
                SELECT * FROM dblink(new_node_dsn, remotesql) AS t(row_count bigint) INTO lolor_row_count;
            ELSE
                lolor_row_count := 0;
            END IF;

            remotesql := 'SELECT count(*) FROM spock.tables WHERE nspname = ''lolor'' AND set_name IS NOT NULL';
            SELECT * FROM dblink(src_dsn, remotesql) AS t(count integer) INTO src_lolor_repset_tables;

            IF lolor_row_count > 0 THEN
                RAISE NOTICE 'FAIL: Destination database has pre-existing large object data in the lolor tables';
                checks_failed := checks_failed + 1;
            ELSIF src_lolor_repset_tables > 0 AND NOT new_lolor_installed THEN
                RAISE NOTICE 'FAIL: Source node replicates lolor tables but the destination database does not have the lolor extension installed';
                checks_failed := checks_failed + 1;
            ELSE
                RAISE NOTICE 'PASS: Destination database lolor state is compatible with add_node';
                checks_passed := checks_passed + 1;
            END IF;
        EXCEPTION WHEN OTHERS THEN
            RAISE NOTICE 'FAIL: lolor check - %', SQLERRM;
            checks_failed := checks_failed + 1;
        END;

        -- Check 8b: Verify database is empty (no user tables)
        BEGIN
            remotesql := $pg_tables$
                SELECT count(*) FROM pg_tables
                WHERE schemaname NOT IN ('information_schema', 'pg_catalog', 'pg_toast', 'spock', 'lolor')
                AND schemaname NOT LIKE 'pg_temp_%'
                AND schemaname NOT LIKE 'pg_toast_temp_%'
            $pg_tables$;
            SELECT * FROM dblink(new_node_dsn, remotesql) AS t(table_count text) INTO user_table_count;

            IF user_table_count IS NOT NULL AND user_table_count::integer = 0 THEN
                RAISE NOTICE 'PASS: New node database is empty (fresh database)';
                checks_passed := checks_passed + 1;
            ELSE
                RAISE NOTICE 'FAIL: New node database has % user-created tables', user_table_count;
                checks_failed := checks_failed + 1;
            END IF;
        EXCEPTION WHEN OTHERS THEN
            RAISE NOTICE 'FAIL: Database emptiness check - %', SQLERRM;
            checks_failed := checks_failed + 1;
        END;
    END IF;

    -- ========================================================================
    -- Check 9: Replication status (post-check only)
    -- ========================================================================
    IF check_type = 'post' AND new_node_name IS NOT NULL AND new_node_dsn IS NOT NULL THEN
        BEGIN
            remotesql := 'SELECT count(*) FROM spock.subscription WHERE sub_enabled = true';
            SELECT * FROM dblink(new_node_dsn, remotesql) AS t(sub_count text) INTO sub_count;

            IF sub_count IS NOT NULL AND sub_count::integer > 0 THEN
                RAISE NOTICE 'PASS: New node % has active subscriptions', new_node_name;
                checks_passed := checks_passed + 1;
            ELSE
                RAISE NOTICE 'FAIL: New node % has no active subscriptions', new_node_name;
                checks_failed := checks_failed + 1;
            END IF;
        EXCEPTION WHEN OTHERS THEN
            RAISE NOTICE 'FAIL: Post-addition replication check - %', SQLERRM;
            checks_failed := checks_failed + 1;
        END;
    END IF;

    -- ========================================================================
    -- Summary
    -- ========================================================================
    RAISE NOTICE '';
    RAISE NOTICE '================================================================================';
    RAISE NOTICE 'HEALTH CHECK SUMMARY';
    RAISE NOTICE '================================================================================';
    RAISE NOTICE 'Checks Passed: %', checks_passed;
    RAISE NOTICE 'Checks Failed: %', checks_failed;
    RAISE NOTICE 'Total Checks:  %', checks_passed + checks_failed;
    RAISE NOTICE '';

    IF checks_failed > 0 THEN
        RAISE NOTICE 'RESULT: FAILED - Please resolve issues before proceeding';
        RAISE EXCEPTION 'Health check failed with % failed checks', checks_failed;
    ELSE
        RAISE NOTICE 'RESULT: PASSED - Cluster is ready for ZODAN node addition';
    END IF;
END;
$$;