feat: handle event propagation via DB (HA)

config.sample.yml

@@ -126,6 +126,15 @@ uploads:
 offline: false
 
 # ---------------------------------------------------------------------
+# High-Availability
+# ---------------------------------------------------------------------
+# Set to true if you have multiple concurrent instances running off the
+# same DB (e.g. Kubernetes pods / load balanced instances). Leave false
+# otherwise.
+
+ha: false
+
+# ---------------------------------------------------------------------
 # Data Path
 # ---------------------------------------------------------------------
 # Writeable data path used for cache and temporary user uploads.

dev/build/config.yml

@@ -16,3 +16,4 @@ ssl:
   domain: $(LETSENCRYPT_DOMAIN)
   subscriberEmail: $(LETSENCRYPT_EMAIL)
 logLevel: info
+ha: $(HA_ACTIVE)

dev/index.js

@@ -5,7 +5,6 @@
 // Licensed under AGPLv3
 // ===========================================
 
-const Promise = require('bluebird')
 const _ = require('lodash')
 const chalk = require('chalk')
 
@@ -60,16 +59,9 @@ const init = {
     })
   },
   async reload() {
-    console.warn(chalk.yellow('--- Stopping scheduled jobs...'))
-    if (global.WIKI.scheduler) {
-      global.WIKI.scheduler.stop()
-    }
-    console.warn(chalk.yellow('--- Closing DB connections...'))
-    await global.WIKI.models.knex.destroy()
-    console.warn(chalk.yellow('--- Closing Server connections...'))
-    if (global.WIKI.servers) {
-      await global.WIKI.servers.stopServers()
-    }
+    console.warn(chalk.yellow('--- Gracefully stopping server...'))
+    await global.WIKI.kernel.shutdown()
+
     console.warn(chalk.yellow('--- Purging node modules cache...'))
 
     global.WIKI = {}

package.json

@@ -69,6 +69,7 @@
     "dotize": "0.3.0",
     "elasticsearch6": "npm:@elastic/elasticsearch@6",
     "elasticsearch7": "npm:@elastic/elasticsearch@7",
+    "eventemitter2": "6.0.0",
     "emoji-regex": "9.0.0",
     "express": "4.17.1",
     "express-brute": "1.0.1",
@@ -146,6 +147,7 @@
     "pg": "8.0.2",
     "pg-hstore": "2.3.3",
     "pg-query-stream": "3.0.6",
+    "pg-pubsub": "0.5.0",
     "pg-tsquery": "8.1.0",
     "pug": "2.0.4",
     "punycode": "2.1.1",

server/app/data.yml

@@ -28,6 +28,7 @@ defaults:
       maxFileSize: 5242880
       maxFiles: 10
     offline: false
+    ha: false
     # DB defaults
     api:
       isEnabled: false

server/core/db.js

@@ -17,6 +17,7 @@ const migrateFromBeta = require('../db/beta')
 module.exports = {
   Objection,
   knex: null,
+  listener: null,
   /**
    * Initialize DB
    *
@@ -182,5 +183,55 @@ module.exports = {
       ...this,
       ...models
     }
+  },
+  /**
+   * Subscribe to database LISTEN / NOTIFY for multi-instances events
+   */
+  async subscribeToNotifications () {
+    const useHA = (WIKI.config.ha === true || WIKI.config.ha === 'true' || WIKI.config.ha === 1 || WIKI.config.ha === '1')
+    if (!useHA) {
+      return
+    } else if (WIKI.config.db.type !== 'postgres') {
+      WIKI.logger.warn(`Database engine doesn't support pub/sub. Will not handle concurrent instances: [ DISABLED ]`)
+      return
+    }
+
+    const PGPubSub = require('pg-pubsub')
+
+    this.listener = new PGPubSub(this.knex.client.connectionSettings, {
+      log (ev) {
+        WIKI.logger.debug(ev)
+      }
+    })
+    this.listener.addChannel('wiki', payload => {
+      if (_.has(payload.event) && payload.source !== WIKI.INSTANCE_ID) {
+        WIKI.events.emit(payload.event, payload.value)
+      }
+    })
+    WIKI.events.onAny(this.notifyViaDB)
+
+    WIKI.logger.info(`High-Availability Listener initialized successfully: [ OK ]`)
+  },
+  /**
+   * Unsubscribe from database LISTEN / NOTIFY
+   */
+  async unsubscribeToNotifications () {
+    if (this.listener) {
+      WIKI.events.offAny(this.notifyViaDB)
+      this.listener.close()
+    }
+  },
+  /**
+   * Publish event via database NOTIFY
+   *
+   * @param {string} event Event fired
+   * @param {object} value Payload of the event
+   */
+  notifyViaDB (event, value) {
+    this.listener.publish('wiki', {
+      source: WIKI.INSTANCE_ID,
+      event,
+      value
+    })
   }
 }

server/core/kernel.js

 const _ = require('lodash')
-const EventEmitter = require('events')
+const EventEmitter = require('eventemitter2').EventEmitter2
 
 /* global WIKI */
 
@@ -37,6 +37,7 @@ module.exports = {
       WIKI.servers = require('./servers')
       WIKI.sideloader = require('./sideloader').init()
       WIKI.events = new EventEmitter()
+      await WIKI.models.subscribeToNotifications()
     } catch (err) {
       WIKI.logger.error(err)
       process.exit(1)
@@ -91,5 +92,21 @@ module.exports = {
       WIKI.logger.warn(err)
       WIKI.telemetry.sendError(err)
     })
+  },
+  /**
+   * Graceful shutdown
+   */
+  async shutdown () {
+    if (WIKI.models) {
+      await WIKI.models.unsubscribeToNotifications()
+      await WIKI.models.knex.client.pool.destroy()
+      await WIKI.models.knex.destroy()
+    }
+    if (WIKI.scheduler) {
+      WIKI.scheduler.stop()
+    }
+    if (WIKI.servers) {
+      await WIKI.servers.stopServers()
+    }
   }
 }

server/index.js

@@ -4,11 +4,13 @@
 // ===========================================
 
 const path = require('path')
+const nanoid = require('nanoid')
 
 let WIKI = {
   IS_DEBUG: process.env.NODE_ENV === 'development',
   IS_MASTER: true,
   ROOTPATH: process.cwd(),
+  INSTANCE_ID: nanoid(10),
   SERVERPATH: path.join(process.cwd(), 'server'),
   Error: require('./helpers/error'),
   configSvc: require('./core/config'),