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

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.