gis Database

Parameters

Name	Type	Mode
p_ids	ARRAY	IN
p_min_event_id	bigint	IN
p_max_event_id	bigint	IN
p_max_action_tstamp_tx	timestamp with time zone	IN
replay_count	integer	OUT

Definition

DECLARE
    sqltemplate text;
    p_central_id text;
    p_clone_id text;
    p_sync_id uuid;
    rec record;
    p_counter integer;
    dblink_connection_name text;
    dblink_msg text;
BEGIN
    -- Do the replay ONLY if there are ids to replay
    -- We do NOT want to insert a new lizsync.history item if p_ids IS NULL
    -- This means there were no changes in the central since last sync
    -- the next sync will still use the last central->clone sync history item data
    RAISE NOTICE 'Check if there are some central logs since last sync: %', p_ids;
    IF p_ids IS NOT NULL THEN

-- Get central server id
        SELECT server_id::text INTO p_central_id
        FROM central_lizsync.server_metadata
        LIMIT 1;
        -- RAISE NOTICE 'Central server id = %', p_central_id;

-- Get clone server id
        SELECT server_id::text INTO p_clone_id
        FROM lizsync.server_metadata
        LIMIT 1;
        -- RAISE NOTICE 'Clone server id = %', p_clone_id;

-- Create dblink connection
        dblink_connection_name = (md5(((random())::text || (clock_timestamp())::text)))::text;
        -- RAISE NOTICE 'dblink_connection_name %', dblink_connection_name;
        SELECT dblink_connect(
            dblink_connection_name,
            'central_server'
        )
        INTO dblink_msg;

-- Add item in CENTRAL history table
        sqltemplate = format(
            'SELECT lizsync.insert_history_item(
                ''%1$s''::text, ''%2$s''::text,
                %3$s, %4$s, Cast(''%5$s'' AS TIMESTAMP WITH TIME ZONE),
                ''partial'', ''pending''
            ) AS sync_id
            ',
            p_central_id, p_clone_id,
            p_min_event_id, p_max_event_id, p_max_action_tstamp_tx
        );
        SELECT sync_id FROM dblink(
            dblink_connection_name,
            sqltemplate
        ) AS foo(sync_id uuid)
        INTO p_sync_id;
        -- RAISE NOTICE 'SYNC ID = %', p_sync_id;

-- Replay SQL queries in clone db
        -- We disable triggers to avoid adding more rows to the local audit logged_actions table
        sqltemplate = '';
        p_counter = 0;
        sqltemplate = concat(sqltemplate, ' SET session_replication_role = replica;');
        FOR rec IN
            SELECT event_id, action
            FROM temp_central_audit
            ORDER BY tid
        LOOP
            p_counter = p_counter + 1;
            -- RAISE NOTICE '%', rec.event_id;
            sqltemplate = concat(sqltemplate, rec.action || ';');
        END LOOP;
        sqltemplate = concat(sqltemplate, ' SET session_replication_role = DEFAULT;');
        -- RAISE NOTICE 'SQL TEMPLATE %', sqltemplate;
        -- RAISE NOTICE 'p_counter %', p_counter;

--RAISE NOTICE '%', sqltemplate;
        IF p_counter > 0 THEN
            EXECUTE sqltemplate
            ;
        END IF;

-- Update central audit logged actions
        -- To tell these actions have been replayed by this clone
        -- run this function with dblink as it must be done by the central server with security definer
        -- it allows the clone to write data back to the protected table logged_actions
        sqltemplate = concat(
            'SELECT lizsync.update_central_logs_add_clone_id(',
            quote_literal(p_clone_id),
            ', ',
            quote_literal(p_sync_id), '::uuid',
            ', ',
            '(', quote_literal(p_ids::text), ')::integer[]',
            ')'
        );
        RAISE NOTICE 'UPDATE CLONE ID = %', sqltemplate;

-- Update central lizsync.logged_actions
        -- we need to use dblink and not dblink_exec
        -- else error "statement returning results not allowed"
        PERFORM * FROM dblink(
            dblink_connection_name,
            sqltemplate
        ) AS foo(update_status boolean)
        ;

-- Modify central server synchronisation item central->clone
        -- to mark it as 'done'
        sqltemplate = format(
            'SELECT lizsync.update_history_item(''%1$s''::uuid, ''done'', NULL)',
            p_sync_id
        )
        ;

PERFORM * FROM dblink(
            dblink_connection_name,
            sqltemplate
        ) AS foo(update_status boolean)
        ;

-- Disconnect dblink
        SELECT dblink_disconnect(dblink_connection_name)
        INTO dblink_msg;

ELSE
        p_counter = 0;
        RAISE NOTICE 'No central logs since last sync';
    END IF;

-- Sync done !
    RETURN QUERY
    SELECT p_counter;
END;

replay_central_logs_to_clone

Description

Parameters

Definition