Feat: abstract extended queries; strip out default funcs from db obj

lib/db.js

@@ -29,6 +29,9 @@ const pgp = Pgp({
     obj.$one = (query, variables) => extendedQueries.one(obj, query, variables)
     obj.$none = (query, variables) => extendedQueries.none(obj, query, variables)
     obj.$any = (query, variables) => extendedQueries.any(obj, query, variables)
+    // when opts is not defined "cb" occupies the "opts" spot of the arguments
+    obj.$tx = (opts, cb) => typeof opts === 'function' ? extendedQueries.tx(obj, {}, opts) : extendedQueries.tx(obj, opts, cb)
+    obj.$task = (opts, cb) => typeof opts === 'function' ? extendedQueries.task(obj, {}, opts) : extendedQueries.task(obj, opts, cb)
   },
   error: (err, e) => {
     if (e.cn) logger.error('Database not reachable.')
@@ -40,13 +43,13 @@ const pgp = Pgp({
   }
 })
 
-const db = pgp(psqlUrl)
+const db = extendedQueries.stripDefaultDbFuncs(pgp(psqlUrl))
 
 eventBus.subscribe('log', args => {
   if (process.env.SKIP_SERVER_LOGS) return
 
   const { level, message, meta } = args
-  const msgToSave = message ? message : _.get('message', meta)
+  const msgToSave = message || _.get('message', meta)
 
   const sql = `insert into server_logs
   (id, device_id, message, log_level, meta) values ($1, $2, $3, $4, $5) returning *`

lib/extendedQueries.js

@@ -1,9 +1,36 @@
-const getSchema = () => 'public'
+const logger = require('./logger')
 
-const getDefaultSchema = () => 'public'
+// TODO use asynclocalstorage
+const getSchema = () => {
+  return 'public'
+}
+
+const getDefaultSchema = () => {
+  return 'ERROR_SCHEMA'
+}
+
+const schemaNotFound = query => {
+  logger.error(`Schema for query '${query}' has not been found and the query has returned null`)
+}
+
+const stripDefaultDbFuncs = dbCtx => {
+  return {
+    ctx: dbCtx.ctx,
+    query: dbCtx.$query,
+    result: dbCtx.$result,
+    many: dbCtx.$many,
+    oneOrNone: dbCtx.$oneOrNone,
+    one: dbCtx.$one,
+    none: dbCtx.$none,
+    any: dbCtx.$any,
+    tx: dbCtx.$tx,
+    task: dbCtx.$task
+  }
+}
 
 const any = (obj, query, variables) => {
   const schema = getSchema()
+  if (!schema) throw new Error(schemaNotFound(query))
   return obj.taskEx({ schema }, t => {
     return t.any(query, variables).then(res => {
       return t.none('set search_path to $1~', [getDefaultSchema()]).then(() => {
@@ -15,6 +42,7 @@ const any = (obj, query, variables) => {
 
 const none = (obj, query, variables) => {
   const schema = getSchema()
+  if (!schema) throw new Error(schemaNotFound(query))
   return obj.taskEx({ schema }, t => {
     return t.none(query, variables).then(res => {
       return t.none('set search_path to $1~', [getDefaultSchema()]).then(() => {
@@ -26,6 +54,7 @@ const none = (obj, query, variables) => {
 
 const one = (obj, query, variables) => {
   const schema = getSchema()
+  if (!schema) throw new Error(schemaNotFound(query))
   return obj.taskEx({ schema }, t => {
     return t.one(query, variables).then(res => {
       return t.none('set search_path to $1~', [getDefaultSchema()]).then(() => {
@@ -37,6 +66,7 @@ const one = (obj, query, variables) => {
 
 const oneOrNone = (obj, query, variables) => {
   const schema = getSchema()
+  if (!schema) throw new Error(schemaNotFound(query))
   return obj.taskEx({ schema }, t => {
     return t.oneOrNone(query, variables).then(res => {
       return t.none('set search_path to $1~', [getDefaultSchema()]).then(() => {
@@ -48,6 +78,7 @@ const oneOrNone = (obj, query, variables) => {
 
 const manyOrNone = (obj, query, variables) => {
   const schema = getSchema()
+  if (!schema) throw new Error(schemaNotFound(query))
   return obj.taskEx({ schema }, t => {
     return t.manyOrNone(query, variables).then(res => {
       return t.none('set search_path to $1~', [getDefaultSchema()]).then(() => {
@@ -59,6 +90,7 @@ const manyOrNone = (obj, query, variables) => {
 
 const many = (obj, query, variables) => {
   const schema = getSchema()
+  if (!schema) throw new Error(schemaNotFound(query))
   return obj.taskEx({ schema }, t => {
     return t.many(query, variables).then(res => {
       return t.none('set search_path to $1~', [getDefaultSchema()]).then(() => {
@@ -70,6 +102,7 @@ const many = (obj, query, variables) => {
 
 const result = (obj, query, variables, cb, thisArg) => {
   const schema = getSchema()
+  if (!schema) throw new Error(schemaNotFound(query))
   return obj.taskEx({ schema }, t => {
     return t.result(query, variables, cb, thisArg).then(res => {
       return t.none('set search_path to $1~', [getDefaultSchema()]).then(() => {
@@ -79,10 +112,14 @@ const result = (obj, query, variables, cb, thisArg) => {
   })
 }
 
-const query = (obj, query, values, qrm) => {
+const query = (obj, query, variables, qrm, throwOnError) => {
   const schema = getSchema()
+  if (!schema) {
+    if (throwOnError) throw new Error(schemaNotFound(query))
+    return Promise.resolve(schemaNotFound(query))
+  }
   return obj.taskEx({ schema }, t => {
-    return t.query(query, values, qrm).then(res => {
+    return t.query(query, variables, qrm).then(res => {
       return t.none('set search_path to $1~', [getDefaultSchema()]).then(() => {
         return res
       })
@@ -90,6 +127,18 @@ const query = (obj, query, values, qrm) => {
   })
 }
 
+const tx = (obj, opts, cb) => {
+  return obj.tx(opts, t => {
+    return cb(stripDefaultDbFuncs(t))
+  })
+}
+
+const task = (obj, opts, cb) => {
+  return obj.task(opts, t => {
+    return cb(stripDefaultDbFuncs(t))
+  })
+}
+
 module.exports = {
   any,
   manyOrNone,
@@ -98,5 +147,8 @@ module.exports = {
   oneOrNone,
   one,
   result,
-  query
+  query,
+  tx,
+  task,
+  stripDefaultDbFuncs
 }

package.json

@@ -60,7 +60,7 @@
     "p-each-series": "^1.0.0",
     "p-retry": "^4.4.0",
     "pg-native": "^3.0.0",
-    "pg-promise": "^7.4.1",
+    "pg-promise": "^10.10.2",
     "pify": "^3.0.0",
     "pretty-ms": "^2.1.0",
     "promise-sequential": "^1.1.1",