`change_logs.sql` - automatically track changes to Postgres tables.

change_logs.sql

/*
change_logs: use triggers to automatically track changes to certain columns in
database tables.

To enable change tracking for a table:

    select change_logs_track_table(
        -- table name
        'table_to_track',
        -- primary key column
        'id',
        -- a list of columns to track
        array['foo', 'bar']
    );

This will do two things:

1. Setup a trigger after inserts, updates, and deletes on 'table_to_track'
   which calls 'change_logs_create(...)' each time there is a change to the
   table.

2. Write a row to 'change_logs_track_table' with the table name, primary key
   column, and columns to log.

The 'change_logs_track_table(...)' is idempotent and can safely be called
multiple times with different columns to track.

Considerations:

* The "age" calculation can be expensive for columns updated infrequently on a
  table which is updated frequently (because, currently, there is no index on
  the column names in each change).

* Only necessary columns should be tracked, as it will get expensive to log all
  changes to all columns.
*/

create table if not exists change_logs (
    id bigserial not null,
    tbl varchar(64) not null,
    pk varchar(64) not null,
    ts timestamp with time zone not null,
    user_id int,
    old jsonb null,
    new jsonb null,
    age jsonb null
);

create index if not exists change_logs_tbl_pk_ts on change_logs (tbl, pk, ts);

create table if not exists change_logs_tracked_tables (
    table_name varchar(64) primary key,
    pk_column varchar(64) not null,
    logged_columns text[] not null
);

/*
(internal) selects `_cols` from `_obj`, returning a new object containing only
the columns in `_cols`.

`_cols` can contain a wildcard `*` to select all columns, and a column name
can be prefixed with '-' to exclude it from the result.
> _change_logs_join_columns(array['*', '-foo'], '{"foo": "bar", "baz": 42}')
{'baz': 42}
*/
create or replace function _change_logs_jsonb_filter_object(_cols text[], _obj jsonb)
returns jsonb as $pgsql$
declare
    res jsonb := jsonb_build_object();
    col text;
    cols_to_log text[];
begin
    if _obj is null then
        return null;
    end if;

    cols_to_log := case
        when '*' = any(_cols) then array(select jsonb_object_keys(_obj)) || _cols
        else _cols
    end;

    foreach col in array cols_to_log loop
        if col = '*' then
            continue;
        end if;
        res := case
            when col like '-%' then res - substring(col, 2)
            else res || jsonb_build_object(col, _obj->col)
        end;
    end loop;
    return res;
end;
$pgsql$ language plpgsql returns null on null input;

/*
(internal) change_logs_create: creates a record in the `change_logs` table.

Called by `change_logs_insert_update_delete_trigger` each time a row is inserted,
updated, or deleted in a table which is tracked by `change_logs_track_table`.
*/
create or replace function change_logs_create(_table text, _old jsonb, _new jsonb)
returns void as $pgsql$
declare
    def change_logs_tracked_tables;
    _pk text;
    col text;
    old_to_add jsonb;
    new_to_add jsonb;
    ages jsonb;
    did_change boolean;
begin
    def := (
        select row(t.*)
        from change_logs_tracked_tables as t
        where table_name = _table
    );
    if def is null then
        return;
    end if;

    _pk := COALESCE(_new->>(def.pk_column), _old->>(def.pk_column));
    if _pk is null THEN
        raise exception 'Primary key % for table % not found in old=% or new=%', def.pk_column, _table, _old, _new;
    end if;

    _old := _change_logs_jsonb_filter_object(def.logged_columns, _old);
    _new := _change_logs_jsonb_filter_object(def.logged_columns, _new);

    -- Handle new rows
    if _old is null then
        insert into change_logs (
            tbl, pk,
            ts,
            user_id,
            old, new,
            age
        ) values (
            _table, _pk,
            now(),
            nullif(current_setting('change_logs.current_user_id', true), '')::integer,
            null, _new,
            null
        );
        return;
    end if;

    -- Handle changes to existing rows
    if _new is not null then
        old_to_add := jsonb_build_object();
        new_to_add := jsonb_build_object();
        ages := jsonb_build_object();

        did_change := false;
        foreach col in array array(select jsonb_object_keys(_old)) loop
            if (_old->col) is distinct from (_new->col) then
                did_change := true;
                old_to_add := old_to_add || jsonb_build_object(col, _old->col);
                new_to_add := new_to_add || jsonb_build_object(col, _new->col);
                ages := ages || jsonb_build_object(col, (
                    select now() - ts
                    from change_logs
                    where
                        tbl = _table and
                        pk = _pk and
                        new ? col
                    order by ts desc
                    limit 1
                ));
            end if;
        end loop;

        if not did_change then
            return;
        end if;

        insert into change_logs (
            tbl, pk,
            ts,
            user_id,
            old, new,
            age
        ) values (
            _table, _pk,
            now(),
            nullif(current_setting('change_logs.current_user_id', true), '')::integer,
            old_to_add, new_to_add,
            ages
        );
        return;
    end if;

    -- Handle deleted rows
    if _new is null then
        ages := jsonb_build_object();

        foreach col in array array(select jsonb_object_keys(_old)) loop
            if (_old->col) is distinct from (_new->col) then
                ages := ages || jsonb_build_object(col, (
                    select now() - ts
                    from change_logs
                    where
                        tbl = _table and
                        pk = _pk and
                        new ? col
                    order by ts desc
                    limit 1
                ));
            end if;
        end loop;

        insert into change_logs (
            tbl, pk,
            ts,
            user_id,
            old, new,
            age
        ) values (
            _table, _pk,
            now(),
            nullif(current_setting('change_logs.current_user_id', true), '')::integer,
            _old, null,
            null
        );
        return;
    end if;
end;
$pgsql$ language plpgsql;

/*
(internal) change_logs_insert_update_delete_trigger: trigger added to all
tables tracked with `change_logs_track_table`.
*/
create or replace function change_logs_insert_update_delete_trigger()
returns trigger as $pgsql$
begin
    perform change_logs_create(
        TG_TABLE_NAME::regclass::text,
        case TG_OP when 'INSERT' then NULL else row_to_json(OLD)::jsonb end,
        case TG_OP when 'DELETE' then NULL else row_to_json(NEW)::jsonb end
    );
    return new;
end;
$pgsql$ language plpgsql;

/*
(internal) change_logs_assert_column: asserts that `_col` exists on `_table`.

When `_col='*'`, this function asserts only that `_table` exists.

When `_col like '-%'`, this function will strip the '-' before asserting that
`_col` exists on `_table`.
*/
create or replace function change_logs_assert_column(_table text, _col text)
returns void as $pgsql$
declare
    cols text[];
begin
    if _col = '*' then
        return;
    end if;

    if _col like '-%' then
        _col := substring(_col, 2);
    end if;

    cols := (
      SELECT array_agg(column_name)
      FROM information_schema.columns
      WHERE table_name=_table
    );

    if cols is null then
        raise exception 'Table not found: "%"', _table;
    end if;

    if not (_col = any(cols)) then
        raise exception 'Column "%" not found on table "%"', _col, _table;
    end if;
end;
$pgsql$ language plpgsql;

/*
change_logs_track_table: track changes to `_cols` on `_table`, with rows identified by
primary key column `_pk_col`.

Note: if the table has previously been tracked with `change_logs_track_tables`, the
old `_pk_col` will be overwritten, and `_cols` will be added to the list of tracked columns.

For example::

    > select * from change_logs_track_table('users', 'id', array['username', 'email']);
    table_name | pk_column | logged_columns
    -----------+-----------+----------------
    users      | id        | {username, email}

    > select * from change_logs_track_table('users', 'unknown_column', array[...]);
    ERROR: Column "unknown_column" not found on table "users"

    > select * from change_logs_track_table('unknown_table', 'id', array[...]);
    ERROR: Table not found: "unknown_table"

The `_cols` array can contain the special value `*`, which will track all columns.

If any cols in `_cols` are prefixed with a '-', they will be removed from the
list of tracked columns. These columns should appear after the '*'.

For example::

    > select * from change_logs_track_table('users', 'id', array['*', '-password']);
    table_name | pk_column | logged_columns
    -----------+-----------+----------------
    users      | id        | {'*', '-password'}
*/
create or replace function change_logs_track_table(_table text, _pk_col text, _cols text[])
returns change_logs_tracked_tables as $pgsql$
begin
    perform change_logs_assert_column(_table, _pk_col);

    execute 'drop trigger if exists ' || quote_ident(_table || '_change_logs_tracker') || ' ' ||
        'on ' || quote_ident(_table);
    execute 'create trigger ' || quote_ident(_table || '_change_logs_tracker') || ' ' ||
        'after insert or update or delete ' ||
        'on ' || quote_ident(_table) || ' ' ||
        'for each row execute procedure change_logs_insert_update_delete_trigger()';
    insert into change_logs_tracked_tables values (_table, _pk_col, array[]::text[])
    on conflict (table_name) do update
    set
        pk_column = _pk_col;

    return change_logs_track_columns(_table, _cols);
end;
$pgsql$ language plpgsql;

/*
change_logs_track_columns: adds `_cols` to the list of columns tracked on `_table`.

Returns an error if `_table` is not tracked.

For example::

    > select * from change_logs_track_columns('users', array['full_name']);
    table_name | pk_column | logged_columns
    -----------+-----------+----------------
    users      | id        | {username, email, full_name}

    > select * from change_logs_track_columns('users', array['unknown_column']);
    ERROR: Column "unknown_column" not found on table "users"

    > select * from change_logs_track_columns('unknown_table', array['email']);
    ERROR:  Table not found: "unknown_table"

The `_cols` array can contain the special value `*`, which will track all columns.

If any cols in `_cols` are prefixed with a '-', they will be removed from the
list of tracked columns. These columns should appear after the '*'.

For example::

    > select * from change_logs_track_columns('users', array['*', '-password']);
    table_name | pk_column | logged_columns
    -----------+-----------+----------------
    users      | id        | {'*', '-password'}
*/
create or replace function change_logs_track_columns(_table text, _cols text[])
returns change_logs_tracked_tables as $pgsql$
declare
  res record;
  _col text;
begin
    foreach _col in array _cols loop
       perform change_logs_assert_column(_table, _col);
    end loop;

    update change_logs_tracked_tables
    set logged_columns = logged_columns || (select array(
        select col
        from unnest(_cols) as x(col)
        where not (col = any(logged_columns))
    ))
    where table_name = _table
    returning * into res;

    if res is null then
        raise exception 'Table "%" not logged (hint: use `change_logs_track_table("%", ''%''::text[])`', _table, _table, _cols;
    end if;

    return res;
end;
$pgsql$ language plpgsql;

/*
change_logs_set_user_id: sets the current user id to `_id`, which will be
included in any change logs incurred by the current transaction.

Example:

    > select change_logs_set_user_id(1);
*/
create or replace function change_logs_set_user_id(_id int)
returns void as $pgsql$
begin
    perform set_config('change_logs.current_user_id', _id::text, false);
end;
$pgsql$ language plpgsql;

select change_logs_track_table('change_logs_tracked_tables', 'table_name', array['table_name', 'pk_column', 'logged_columns']);