get_central_audit_logs

DECLARE

    p_clone_id text;

    p_excluded_columns_text text;

    sqltemplate text;

    sqltext text;

    dblink_connection_name text;

    dblink_msg text;

BEGIN

​

    IF p_excluded_columns IS NULL THEN

        p_excluded_columns_text = '';

    ELSE

        p_excluded_columns_text = array_to_string(p_excluded_columns, '@');

    END IF;

​

    -- Get clone server id

    SELECT server_id::text INTO p_clone_id

    FROM lizsync.server_metadata

    LIMIT 1;

​

    IF p_clone_id IS NULL THEN

        RETURN QUERY

        SELECT

            NULL, NULL, NULL, NULL,

            NULL, NULL, NULL, NULL, NULL, NULL

        LIMIT 0;

    END IF;

​

    -- Create dblink connection

    dblink_connection_name = (md5(((random())::text || (clock_timestamp())::text)))::text;

    SELECT dblink_connect(

        dblink_connection_name,

        'central_server'

    )

    INTO dblink_msg;

​

    sqltemplate = '

​

        WITH

        cid AS (

            SELECT server_id::text AS p_central_id

            FROM lizsync.server_metadata

            LIMIT 1

        ),

        last_sync AS (

            SELECT

                max_action_tstamp_tx,

                max_event_id

            FROM lizsync.history

            WHERE True

            AND server_from = (SELECT p_central_id FROM cid)

            AND ''%1$s'' = ANY (server_to)

            AND sync_status = ''done''

            ORDER BY sync_time DESC

            LIMIT 1

        ),

        tables AS (

            SELECT sync_tables

            FROM lizsync.synchronized_tables

            WHERE server_id = ''%1$s''::uuid

            LIMIT 1

        )

        SELECT

            a.event_id,

            a.action_tstamp_tx AS action_tstamp_tx,

            extract(epoch from a.action_tstamp_tx)::numeric AS action_tstamp_epoch,

            concat(a.schema_name, ''.'', a.table_name) AS ident,

            a.action AS action_type,

            CASE

                WHEN a.sync_data->>''origin'' IS NULL THEN ''central''

                ELSE ''clone''

            END AS origine,

            Coalesce(

                lizsync.get_event_sql(

                    a.event_id,

                    ''%2$s''::text,

                    array_cat(

                        string_to_array(''%3$s'',''@''),

                        array_remove(akeys(a.changed_fields), s)

                    )

                ),

                ''''

            ) AS action,

            s AS updated_field,

            (a.row_data->''%2$s'')::uuid AS uid,

            CASE

                WHEN a.sync_data->>''action_tstamp_tx'' IS NOT NULL

                AND a.sync_data->>''origin'' IS NOT NULL

                    THEN extract(epoch from Cast(a.sync_data->>''action_tstamp_tx'' AS TIMESTAMP WITH TIME ZONE))::numeric

                ELSE extract(epoch from a.action_tstamp_tx)::numeric

            END AS original_action_tstamp_tx

        FROM lizsync.logged_actions AS a

        -- Create as many lines as there are changed fields in UPDATE

        LEFT JOIN skeys(a.changed_fields) AS s ON TRUE,

        last_sync, tables

​

        WHERE True

​

        -- modifications do not come from clone database

        AND (a.sync_data->>''origin'' != ''%1$s'' OR a.sync_data->>''origin'' IS NULL)

​

        -- modifications have not yet been replayed in the clone database

        AND (NOT (a.sync_data->''replayed_by'' ? ''%1$s'') OR a.sync_data->''replayed_by'' = jsonb_build_object() )

​

        -- modifications after the last synchronisation

        -- MAX_ACTION_TSTAMP_TX Par ex: 2019-04-20 12:00:00+02

        AND a.action_tstamp_tx > last_sync.max_action_tstamp_tx

​

        -- Event ID is bigger than last sync event id

        -- MAX_EVENT_ID

        AND a.event_id > last_sync.max_event_id

​

        -- only for tables synchronised by the clone server ID

        AND sync_tables ? concat(''"'', a.schema_name, ''"."'', a.table_name, ''"'')

​

        ORDER BY a.event_id

        ;

    ';

​

    sqltext = format(sqltemplate,

        p_clone_id,

        p_uid_field,

        p_excluded_columns_text

    );

    --RAISE NOTICE '%', sqltext;

​

    RETURN QUERY

    SELECT *

    FROM dblink(

        dblink_connection_name,

        sqltext

    ) AS t(

        event_id bigint, action_tstamp_tx timestamp with time zone, action_tstamp_epoch numeric,

        ident text, action_type text, origine text, action text, updated_field text,

        uid uuid, original_action_tstamp_tx numeric

    )

    ;

​

END;

​