Feat: AsyncLocalStorage for schema changing

bin/lamassu-migrate

@@ -21,12 +21,12 @@ const getMigrateFile = () => {
   })
 }
 
-db.none(createMigration)
-  .then(() => Promise.all([db.oneOrNone(select), getMigrateFile()]))
+db.$none(createMigration)
+  .then(() => Promise.all([db.$oneOrNone(select), getMigrateFile()]))
   .then(([qResult, migrateFile]) => {
     process.env.SKIP_SERVER_LOGS = !(qResult && qResult.data.migrations.find(({ title }) => title === '1572524820075-server-support-logs.js'))
     if (!qResult && migrateFile) {
-      return db.none('insert into migrations (id, data) values (1, $1)', [migrateFile])
+      return db.$none('insert into migrations (id, data) values (1, $1)', [migrateFile])
     }
   })
   .then(() => migrate.run())

lib/app.js

@@ -3,6 +3,7 @@ const http = require('http')
 const https = require('https')
 const argv = require('minimist')(process.argv.slice(2))
 
+const { asyncLocalStorage, defaultStore } = require('./async-storage')
 const routes = require('./routes')
 const logger = require('./logger')
 const poller = require('./poller')
@@ -19,26 +20,29 @@ const version = require('../package.json').version
 logger.info('Version: %s', version)
 
 function run () {
-  let count = 0
-  let handler
+  const store = defaultStore()
+  return asyncLocalStorage.run(store, () => {
+    let count = 0
+    let handler
 
-  const errorHandler = err => {
-    count += 1
-    logger.error(err)
-    logger.error('[%d] Retrying in 10s...', count)
-  }
+    const errorHandler = err => {
+      count += 1
+      logger.error(err)
+      logger.error('[%d] Retrying in 10s...', count)
+    }
 
-  const runner = () =>
-    settingsLoader.loadLatest()
-      .then(settings => {
-        clearInterval(handler)
-        return loadSanctions(settings)
-          .then(() => startServer(settings))
-      })
-      .catch(errorHandler)
+    const runner = () =>
+      settingsLoader.loadLatest()
+        .then(settings => {
+          clearInterval(handler)
+          return loadSanctions(settings)
+            .then(() => startServer(settings))
+        })
+        .catch(errorHandler)
 
-  handler = setInterval(runner, 10000)
-  return runner()
+    handler = setInterval(runner, 10000)
+    return runner()
+  })
 }
 
 function loadSanctions (settings) {

lib/async-storage.js

@@ -0,0 +1,12 @@
+const { AsyncLocalStorage } = require('async_hooks')
+const asyncLocalStorage = new AsyncLocalStorage()
+
+const defaultStore = (a = null) => {
+  const store = new Map()
+  store.set('schema', 'public')
+  store.set('defaultSchema', 'ERROR_SCHEMA')
+  if (a) store.set('a', 'a')
+  return store
+}
+
+module.exports = { asyncLocalStorage, defaultStore }

lib/db.js

@@ -1,9 +1,11 @@
 const Pgp = require('pg-promise')
 const uuid = require('uuid')
 const _ = require('lodash/fp')
+
 const psqlUrl = require('../lib/options').postgresql
 const logger = require('./logger')
 const eventBus = require('./event-bus')
+const { asyncLocalStorage, defaultStore } = require('./async-storage')
 
 const stripDefaultDbFuncs = dbCtx => {
   return {
@@ -87,9 +89,13 @@ eventBus.subscribe('log', args => {
 
   const sql = `insert into server_logs
   (id, device_id, message, log_level, meta) values ($1, $2, $3, $4, $5) returning *`
-
-  db.$one(sql, [uuid.v4(), '', msgToSave, level, meta])
-    .then(_.mapKeys(_.camelCase))
+  // need to set AsyncLocalStorage (ALS) for this function as well
+  // because this module is imported before ALS is set up on app.js
+  const store = defaultStore(true)
+  asyncLocalStorage.run(store, () => {
+    db.$one(sql, [uuid.v4(), '', msgToSave, level, meta])
+      .then(_.mapKeys(_.camelCase))
+  })
 })
 
 module.exports = db

lib/new-admin/admin-server.js

@@ -11,6 +11,7 @@ const cookieParser = require('cookie-parser')
 const { ApolloServer, AuthenticationError } = require('apollo-server-express')
 const _ = require('lodash/fp')
 
+const asyncLocalStorage = require('../async-storage')
 const options = require('../options')
 const users = require('../users')
 
@@ -96,12 +97,17 @@ const certOptions = {
 }
 
 function run () {
-  const serverPort = devMode ? 8070 : 443
+  const store = new Map()
+  asyncLocalStorage.run(store, () => {
+    store.set('schema', 'public')
+    store.set('defaultSchema', 'ERROR_SCHEMA')
+    const serverPort = devMode ? 8070 : 443
 
-  const serverLog = `lamassu-admin-server listening on port ${serverPort}`
+    const serverLog = `lamassu-admin-server listening on port ${serverPort}`
 
-  const webServer = https.createServer(certOptions, app)
-  webServer.listen(serverPort, () => console.log(serverLog))
+    const webServer = https.createServer(certOptions, app)
+    webServer.listen(serverPort, () => console.log(serverLog))
+  })
 }
 
 module.exports = { run }