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])

Parameter	Description
db	PostgreSQL db to use
channel	Channel for LISTEN/NOTIFY on the PostgreSQL database, cannot be used by more than one application on the same database.
options	Optional object containing options. See below.

Option	Description
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)

Parameter	Description
query	SELECT query to run and observe. May contain placeholders following pg-promise
params	The 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.
triggers	function(change). The trigger function, see below.
callback	Callback 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:

Field	Description
table	String, name of the table that changed. This will always be in lowercase.
insert	For INSERT, true
delete	For DELETE, true
update	For 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.
row	The row values, for UPDATE, the NEW row values
old	For 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:

Field	Description
added	array of rows that are in new_rows, but not in old_rows
changed	array of rows from new_rows existed in old_rows but have changed
deleted	array 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.

Field	Description
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.