Home

Awesome

pg-query-observer

Observe PostgreSQL query for changes.

Requires PostgresSQL version 9.3 or above.

Usage

var pgp = require('pg-promise')();

import PgQueryObserver from 'pg-query-observer';


const connection = 'postgres://localhost/db';

async function start() {
  try {
    let db = await pgp(connection);

    let query_observer = new PgQueryObserver(db, 'myapp');

    async function cleanupAndExit() {
      await query_observer.cleanup();
      pgp.end();
      process.exit();
    }

    process.on('SIGTERM', cleanupAndExit);
    process.on('SIGINT', cleanupAndExit);

    // Show notifications

    let query = 'SELECT id AS _id, * FROM test';
    let params = [];

    function triggers(change) {
      console.log('triggers', change);
      return true;
    }

    let handle = await query_observer.notify(query, params, triggers, diff => {
      console.log(diff);
    });

    console.log('initial rows', handle.getRows());

    // ... when finished observing the query

    // await handle.stop();

    // ... when finished observing altogether

    // await query_observer.cleanup();
    // await pgp.end();
  }
  catch(err) {
    console.error(err);
  }
}

process.on('unhandledRejection', (err, p) => console.log(err.stack));

start();

constructor(db, channel, [options])

ParameterDescription
dbPostgreSQL db to use
channelChannel for LISTEN/NOTIFY on the PostgreSQL database, cannot be used by more than one application on the same database.
optionsOptional object containing options. See below.
OptionDescription
trigger_delay(default 200ms): passed through to PgTableObserver.
keyfield(default _id): field to use as a unique keyfield to determine the differences.
initial_cached(default true): If a query is already being observed with the same query/params combination, if will use the cached rows as the initial rows (e.g. with handle.getRows()). If your triggers are correctly defined this should be fine. Turn this option off to load fresh rows from the database for the initial dataset to be sure they are up to date.

let handle = async notify(query, params, triggers, callback)

ParameterDescription
querySELECT query to run and observe. May contain placeholders following pg-promise
paramsThe parameters to the query, following pg-promise. Single values will be $1. Array elements will be $1..$n. Object properties will be $*property* where ** is one of (), [], {} or //. See pg-promise for details.
triggersfunction(change). The trigger function, see below.
callbackCallback function, see below.

triggers function

This function will be called whenever there is a change to one of the underlying tables of the query. You should determine if this change requires a rerun of the query. If so, you should return true.

One parameter is passed, change. It contains the following fields:

FieldDescription
tableString, name of the table that changed. This will always be in lowercase.
insertFor INSERT, true
deleteFor DELETE, true
updateFor UPDATE, an object that contains the old and new values of each changed column. If a column score changed from 10 to 20, change.update.score.from would be 10 and change.update.score.to would be 20.
rowThe row values, for UPDATE, the NEW row values
oldFor UPDATE, the OLD row values

callback function

Whenever observer is triggered, the query will be reran. The callback will be called with one parameter, diff. It will contain only the difference between the last time the query was run.

Diff will contain thee fields:

FieldDescription
addedarray of rows that are in new_rows, but not in old_rows
changedarray of rows from new_rows existed in old_rows but have changed
deletedarray if the key's from the rows from old_rows that don't exist in new_rows

return value

On success, returns an object with the following fields.

FieldDescription
async stop()async function(). Call to stop the observer.
async refresh()async function(). Call to refresh the query.
getRows()Get current full set of rows.

async cleanup()

Stop observing and cleanup triggers from the database.