Add notifier to DB library

lib/www/server/lib/db/connection.js

@@ -1,5 +1,5 @@
 const { Pool, Client, types } = require('pg');
-
+const createSubscriber = require('pg-listen');
 const cfg = require("../config");
 
 const pool = new Pool(cfg.db);
@@ -14,6 +14,12 @@ numericTypeOIDs.forEach(oid => {
 // 	types.setTypeParser(oid, function (v) { return JSON.parse(v); });
 // })
 
+function makeSubscriber (opts = cfg.db) {
+	const subscriber = createSubscriber(opts);
+	process.on("exit", () => subscriber.close());
+	return subscriber;
+}
+
 const transaction = {
 	async begin (client) {
 		return await client.query("BEGIN;");
@@ -72,6 +78,7 @@ async function fetchRow (cursor) {
 
 module.exports = {
 	pool,
+	makeSubscriber,
 	transaction,
 	setSurvey,
 	schema2pid,

lib/www/server/lib/db/notify.js

@@ -0,0 +1,36 @@
+const { makeSubscriber } = require('./connection');
+
+
+async function listen (addChannels, callback) {
+
+	const client = makeSubscriber();
+
+	client.events.on("error", (err) => {
+		console.error("Postgres LISTEN subscriber error", err);
+		setTimeout(() => client.connect(), 5000);
+	});
+
+	client.connect();
+
+	if (!Array.isArray(addChannels)) {
+		addChannels = [addChannels];
+	}
+
+	for (const channel of addChannels) {
+		await client.listenTo(channel);
+		client.notifications.on(channel, (payload) => {
+			const data = {
+				channel,
+				_received: new Date(),
+				payload
+			};
+			callback(data);
+		});
+	}
+
+	return client;
+}
+
+module.exports = {
+	listen
+};