add support for postgresql as an alternative to mysql #25

config/test.json

@@ -4,14 +4,29 @@
   },
   "mysql": {
     "host": "localhost",
+    "dialect": "mysql",
     "user": "jambones_test",
     "database": "jambones_test",
     "password": "jambones_test"
   },
+  "postgres": {
+    "host": "localhost",
+    "dialect": "postgres",
+    "user": "postgres",
+    "database": "jambones_test",
+    "password": "jambones_test"
+  },
   "write-mysql": {
     "host": "localhost",
     "user": "jambones_write_test",
     "database": "jambones_test",
     "password": "jambones_test"
+  },
+  "write-postgres": {
+    "host": "localhost",
+    "dialect": "postgres",
+    "user": "postgres",
+    "database": "jambones_test",
+    "password": "jambones_test"
   }
 }

index.js

@@ -1,56 +1,97 @@
-const mysql = require('mysql2');
+const { Sequelize } = require('sequelize');
+const fs = require('fs');
 const cacheActivator = require('./lib/cache-activator');
 
 const pingDb = async(pool) => {
-  return new Promise((resolve, reject) => {
-    pool.getConnection((err, conn) => {
-      if (err) return reject(err);
-      conn.ping((err) => {
-        pool.releaseConnection(conn);
-        if (err) return reject(err);
-        resolve();
-      });
-    });
-  });
+  await pool.authenticate();
 };
 
-module.exports = function(mysqlConfig, logger, writeMysqlConfig = null)  {
-  const pool = mysql.createPool(mysqlConfig);
+module.exports = function(mysqlConfig, logger)  {
   let writePool = null;
-  let writeConfiguration = writeMysqlConfig;
+  const { host, user, port, password, database, connectionLimit, dialect } = mysqlConfig;
+
+  const config = {
+    host,
+    dialect: process.env.JAMBONES_DB_DIALECT || dialect || 'mysql', /* one of 'mysql' | 'postgres' | 'mariadb'  */
+    port,
+    pool: {
+      max: connectionLimit || 10,
+      min: 0,
+      acquire: 30000,
+      idle: 10000
+    }
+  };
+
+  // allow including database ssl certificate
+  if (process.env.JAMBONES_DB_CA_CERT && fs.fileExistsSync(process.env.JAMBONES_DB_CA_CERT)) {
+    config.ssl = {
+      ca : fs.readFileSync(process.env.JAMBONES_DB_CA_CERT)
+    };
+  }
+
+  // create new Sequelize connection
+  const pool = new Sequelize(database, user, password, config);
+
+  pool.promise = () => pool;
+  pool.execute = () => pool.query;
+
   if (process.env.JAMBONES_MYSQL_WRITE_HOST &&
     process.env.JAMBONES_MYSQL_WRITE_USER &&
     process.env.JAMBONES_MYSQL_WRITE_PASSWORD &&
     process.env.JAMBONES_MYSQL_WRITE_DATABASE) {
-    writeConfiguration = {
+    const user = process.env.JAMBONES_MYSQL_WRITE_USER;
+    const password = process.env.JAMBONES_MYSQL_WRITE_PASSWORD;
+    const database = process.env.JAMBONES_MYSQL_WRITE_DATABASE;
+    const writeConfiguration = {
       host: process.env.JAMBONES_MYSQL_WRITE_HOST,
+      dialect: 'mysql',
       port: process.env.JAMBONES_MYSQL_WRITE_PORT || 3306,
-      user: process.env.JAMBONES_MYSQL_WRITE_USER,
-      password: process.env.JAMBONES_MYSQL_WRITE_PASSWORD,
-      database: process.env.JAMBONES_MYSQL_WRITE_DATABASE,
-      connectionLimit: process.env.JAMBONES_MYSQL_WRITE_CONNECTION_LIMIT || 10
+      pool: {
+        max: process.env.JAMBONES_MYSQL_WRITE_CONNECTION_LIMIT || 10,
+        min: 0,
+        acquire: 30000,
+        idle: 10000
+      }
     };
+    // create new Sequelize connection for write pool
+    writePool = new Sequelize(database, user, password, writeConfiguration);
+    // test connection to database write pool
+    writePool.authenticate().catch((err) => { throw err; });
   }
-  if (writeMysqlConfig) {
-    writePool = mysql.createPool(writeConfiguration);
-    writePool.getConnection((err, conn) => {
-      if (err) throw err;
-      conn.ping((err) => {
-        if (err) return logger.error(err, `Error pinging mysql at ${JSON.stringify(writeConfiguration)}`);
-      });
-    });
+
+  if (process.env.JAMBONES_POSTGRES_WRITE_HOST &&
+    process.env.JAMBONES_POSTGRES_WRITE_USER &&
+    process.env.JAMBONES_POSTGRES_WRITE_PASSWORD &&
+    process.env.JAMBONES_POSTGRES_WRITE_DATABASE) {
+    const user = process.env.JAMBONES_POSTGRES_WRITE_USER;
+    const password = process.env.JAMBONES_POSTGRES_WRITE_PASSWORD;
+    const database = process.env.JAMBONES_POSTGRES_WRITE_DATABASE;
+    const writeConfiguration = {
+      host: process.env.JAMBONES_POSTGRES_WRITE_HOST,
+      dialect: 'postgres',
+      port: process.env.JAMBONES_POSTGRES_WRITE_PORT || 3306,
+      pool: {
+        max: process.env.JAMBONES_POSTGRES_WRITE_CONNECTION_LIMIT || 10,
+        min: 0,
+        acquire: 30000,
+        idle: 10000
+      }
+    };
+    // create new Sequelize connection for write pool
+    writePool = new Sequelize(database, user, password, writeConfiguration);
+    // test connection to database write pool
+    writePool.authenticate().catch((err) => { throw err; });
   }
+
+
   // Cache activation for read only.
-  if (process.env.JAMBONES_MYSQL_REFRESH_TTL)
+  if (process.env.JAMBONES_DB_REFRESH_TTL)
     cacheActivator.activate(pool);
 
   logger = logger || {info: () => {}, error: () => {}, debug: () => {}};
-  pool.getConnection((err, conn) => {
-    if (err) throw err;
-    conn.ping((err) => {
-      if (err) return logger.error(err, `Error pinging mysql at ${JSON.stringify(mysqlConfig)}`);
-    });
-  });
+
+  // test connection to the database
+  pool.authenticate().catch((err) => { throw err; });
 
   return {
     pool,

lib/add-sbc-address.js

@@ -1,6 +1,6 @@
 const debug = require('debug')('jambonz:db-helpers');
 const { v4: uuid } = require('uuid');
-
+const { QueryTypes } = require('sequelize');
 
 const sql = `INSERT into sbc_addresses 
 (sbc_address_sid, ipv4, port, tls_port, wss_port, last_updated) 
@@ -16,16 +16,26 @@ wss_port = ?, last_updated = NOW() WHERE sbc_address_sid = ?`;
  */
 async function addSbcAddress(pool, logger, ipv4, port = 5060, tls_port = null, wss_port = null) {
   try {
-    const pp = pool.promise();
     debug(`select with ${ipv4}`);
-    const [r] = await pp.execute('SELECT * FROM sbc_addresses where ipv4 = ?', [ipv4]);
+    const r = await pool.query('SELECT * FROM sbc_addresses where ipv4 = ?', {
+      raw: true,
+      replacements: [ipv4],
+      type: QueryTypes.SELECT
+    });
+
     debug(`results from searching for sbc address ${ipv4}: ${JSON.stringify(r)}`);
     if (r.length > 0) {
-      const [r2] = await pp.execute(updateSql, [port, tls_port, wss_port, r[0].sbc_address_sid]);
+      const r2 = await pool.query(updateSql, {
+        replacements:  [port, tls_port, wss_port, r[0].sbc_address_sid],
+        type: QueryTypes.UPDATE
+      });
       debug(`results from Updating sbc address ${ipv4} port: ${port} 
       tls_port: ${tls_port} wss_port:${wss_port} : ${JSON.stringify(r2)}`);
     } else {
-      const [r2] = await pp.execute(sql, [uuid(), ipv4, port, tls_port, wss_port]);
+      const r2 = await pool.query(sql, {
+        replacements:  [uuid(), ipv4, port, tls_port, wss_port],
+        type: QueryTypes.INSERT
+      });
       debug(`results from inserting sbc address ${ipv4} port: ${port} 
       tls_port: ${tls_port} wss_port:${wss_port} : ${JSON.stringify(r2)}`);
     }

lib/add-smpp-address.js

@@ -1,4 +1,5 @@
 const debug = require('debug')('jambonz:db-helpers');
+const { QueryTypes } = require('sequelize');
 const { v4: uuid } = require('uuid');
 
 
@@ -13,12 +14,21 @@ values (?, ?)`;
  */
 async function addSmppAddress(pool, logger, ipv4) {
   try {
-    const pp = pool.promise();
     debug(`select with ${ipv4}`);
-    const [r] = await pp.execute('SELECT * FROM smpp_addresses where ipv4 = ?', [ipv4]);
+    const r = await pool.query('SELECT * FROM smpp_addresses where ipv4 = ?', {
+      raw: true,
+      replacements: [ipv4],
+      type: QueryTypes.SELECT
+    });
     debug(`results from searching for smpp address ${ipv4}: ${JSON.stringify(r)}`);
+
     if (r.length > 0) return;
-    const [r2] = await pp.execute(sql, [uuid(), ipv4]);
+
+    const r2 = await pool.query(sql, {
+      replacements: [uuid(), ipv4],
+      type: QueryTypes.INSERT
+    });
+
     debug(`results from inserting smpp address ${ipv4}: ${JSON.stringify(r2)}`);
   } catch (err) {
     debug(err);

lib/clean-sbc-addresses.js

@@ -1,6 +1,6 @@
 const debug = require('debug')('jambonz:db-helpers');
-
-
+const { QueryTypes } = require('sequelize');
+const dialect = process.env.JAMBONES_DB_DIALECT  || 'mysql';
 /**
  * Clean the sbc address after long time no update
  * @param {*} pool
@@ -9,11 +9,17 @@ const debug = require('debug')('jambonz:db-helpers');
  */
 async function cleanSbcAddresses(pool, logger) {
   try {
-    const pp = pool.promise();
-    const sql = `DELETE FROM sbc_addresses WHERE last_updated IS NULL OR
-      last_updated < DATE_SUB(NOW(), INTERVAL 
-      '${process.env.DEAD_SBC_IN_SECOND || 3600}' SECOND)`;
-    const [r] = await pp.execute(sql);
+    let sql;
+    if (dialect == 'mysql') {
+      sql = `DELETE FROM sbc_addresses WHERE last_updated IS NULL OR
+        last_updated < DATE_SUB(NOW(), INTERVAL  '${process.env.DEAD_SBC_IN_SECOND || 3600}' SECOND)`;
+    } else {
+      sql = `DELETE FROM sbc_addresses WHERE last_updated IS NULL OR 
+     last_updated < NOW() - INTERVAL '${process.env.DEAD_SBC_IN_SECOND || 3600} second'`;
+    }
+
+
+    const r = await pool.query(sql, { type: QueryTypes.DELETE });
     debug(`results from cleaning for sbc address ${JSON.stringify(r)}`);
   } catch (err) {
     debug(err);

lib/lookup-account-by-phone-number.js

@@ -1,4 +1,5 @@
 const debug = require('debug')('jambonz:db-helpers');
+const { QueryTypes } = require('sequelize');
 
 const sql =
 `SELECT *
@@ -12,13 +13,18 @@ WHERE acc.account_sid = (SELECT account_sid from phone_numbers WHERE number = ?)
  * @param {*} logger
  */
 async function lookupAccountByPhoneNumber(pool, logger, phoneNumber) {
-  const pp = pool.promise();
-  const [r] = await pp.execute({sql, nestTables: true}, [phoneNumber]);
+  const r = await pool.query(sql, {
+    raw: true,
+    nest: false,
+    replacements: [phoneNumber],
+    type: QueryTypes.SELECT
+  });
   debug(`results: ${JSON.stringify(r)}`);
+
   if (r.length > 0) {
-    const obj = r[0].acc;
-    Object.assign(obj, {registration_hook: r[0].rh});
-    if (!obj.registration_hook.url) delete obj.registration_hook;
+    const obj = r[0]; // r[0].acc;
+    //Object.assign(obj, {registration_hook: r[0].rh});
+    //if (!obj.registration_hook.url) delete obj.registration_hook;
     logger.debug(`retrieved account: ${JSON.stringify(obj)}`);
     return obj;
   }

lib/lookup-account-by-sid.js

@@ -1,4 +1,5 @@
 const debug = require('debug')('jambonz:db-helpers');
+const { QueryTypes } = require('sequelize');
 
 const sql =
 `SELECT *
@@ -12,13 +13,18 @@ WHERE acc.account_sid = ?`;
  * @param {*} logger
  */
 async function lookupAccountBySid(pool, logger, account_sid) {
-  const pp = pool.promise();
-  const [r] = await pp.execute({sql, nestTables: true}, [account_sid]);
+  const r = await pool.query(sql, {
+    raw: true,
+    nest: true,
+    replacements: [account_sid],
+    type: QueryTypes.SELECT
+  });
   debug(`results: ${JSON.stringify(r)}`);
   if (r.length > 0) {
-    const obj = r[0].acc;
-    Object.assign(obj, {registration_hook: r[0].rh});
-    if (!obj.registration_hook.url) delete obj.registration_hook;
+    const obj = r[0];
+    //const obj = r[0].acc;
+    //Object.assign(obj, {registration_hook: r[0].rh});
+    //if (!obj.registration_hook.url) delete obj.registration_hook;
     logger.debug(`retrieved account: ${JSON.stringify(obj)}`);
     return obj;
   }

lib/lookup-account-by-sip-realm.js

@@ -1,4 +1,5 @@
 const debug = require('debug')('jambonz:db-helpers');
+const { QueryTypes } = require('sequelize');
 
 const sql =
 `SELECT *
@@ -12,13 +13,17 @@ WHERE acc.sip_realm = ?`;
  * @param {*} logger
  */
 async function lookupAccountBySipRealm(pool, logger, realm) {
-  const pp = pool.promise();
-  const [r] = await pp.execute({sql, nestTables: true}, [realm]);
+  const r = await pool.query(sql, {
+    raw: true,
+    nest: true,
+    replacements: [realm],
+    type: QueryTypes.SELECT
+  });
   debug(`results: ${JSON.stringify(r)}`);
   if (r.length > 0) {
-    const obj = r[0].acc;
-    Object.assign(obj, {registration_hook: r[0].rh});
-    if (!obj.registration_hook.url) delete obj.registration_hook;
+    const obj = r[0]; //r[0].acc;
+    //Object.assign(obj, {registration_hook: r[0].rh});
+    //if (!obj.registration_hook.url) delete obj.registration_hook;
     logger.debug(`retrieved account: ${JSON.stringify(obj)}`);
     return obj;
   }
@@ -31,12 +36,18 @@ async function lookupAccountBySipRealm(pool, logger, realm) {
   if (arr.length > 2) {
     arr.shift();
     const superDomain = arr.join('.');
-    const [r] =  await pp.execute({sql, nestTables: true}, [superDomain]);
+    const r = await pool.query(sql, {
+      raw: true,
+      nest: true,
+      replacements: [superDomain],
+      type: QueryTypes.SELECT
+    });
     debug(`results: ${JSON.stringify(r)}`);
     if (r.length === 1) {
-      const obj = r[0].acc;
-      Object.assign(obj, {registration_hook: r[0].rh});
-      if (!obj.registration_hook.url) delete obj.registration_hook;
+      const obj = r[0]; //r[0].acc;
+      debug(`obj results: ${JSON.stringify(obj)}`);
+      //Object.assign(obj, {registration_hook: r[0].rh});
+      //if (!obj.registration_hook.url) delete obj.registration_hook;
       logger.debug(`retrieved account: ${JSON.stringify(obj)}`);
       return obj;
     }

lib/lookup-account-capacities-by-sid.js

@@ -1,4 +1,5 @@
 const debug = require('debug')('jambonz:db-helpers');
+const { QueryTypes } = require('sequelize');
 
 const sql =
 `select prod.category, ap.quantity
@@ -9,9 +10,12 @@ const sql =
  and ap.product_sid = prod.product_sid`;
 
 async function lookupAccountSettingsBySid(pool, logger, account_sid) {
-  const pp = pool.promise();
-  const [r] = await pp.query(sql, [account_sid]);
-  debug(`results: ${JSON.stringify(r)}`);
+  const r = await pool.query(sql, {
+    raw: true,
+    replacements: [account_sid],
+    type: QueryTypes.SELECT
+  });
+  debug(`results: ${JSON.stringify(r[0])}`);
   return r;
 }