dougal-software/lib/www/server/lib/db/connection.js

const { Pool, Client, types } = require('pg');
const createSubscriber = require('pg-listen');
const cfg = require("../config");

const pool = new Pool(cfg.db);

const numericTypeOIDs = [ 20, 21, 23, 700, 701, 1700 ];
numericTypeOIDs.forEach(oid => {
	types.setTypeParser(oid, function (v) { return Number(v); });
})

// const jsonTypeOIDs = [ 114, 3802 ];
// jsonTypeOIDs.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;");
	},

	async commit (client) {
		return await client.query("COMMIT;");
	},

	async rollback (client) {
		return await client.query("ROLLBACK;");
	}
};

async function setSurvey (projectId, client) {
	if (!client) {
		client = await pool.connect();
	}
	if (projectId) {
		await client.query("CALL set_survey($1);", [projectId]);
	} else {
		await client.query("SET search_path TO public;");
	}
	return client;
}

async function schema2pid (schema, client) {
	if (!client) {
		client = await pool.connect();
	}
	const res = await client.query("SELECT pid FROM projects WHERE schema = $1", [schema]);
	client.release();
	return res.rows[0] && res.rows[0].pid;;
}

/** Fetch one row from a database cursor.
 *
 * @a cursor A query cursor
 *
 * @returns A row object if there are any rows left to consume.
 * @returns Undefined when all the rows (if any) have been consumed.
 */
async function fetchRow (cursor) {
	return new Promise((resolve, reject) => {
		cursor.read(1, (err, rows) => {
			if (err) {
				reject(err);
			} else if (rows.length) {
				resolve(rows[0]);
			} else {
				resolve(); // undefined
			}
		});
	});
}

module.exports = {
	pool,
	makeSubscriber,
	transaction,
	setSurvey,
	schema2pid,
	fetchRow
};