feat: queue handling

config.sample.yml

@@ -42,6 +42,15 @@ redis:
 # ---------------------------------------------------------------------
 # Background Workers
 # ---------------------------------------------------------------------
-
 # Leave 0 for auto based on CPU cores
+
 workers: 0
+
+# ---------------------------------------------------------------------
+# High Availability
+# ---------------------------------------------------------------------
+# Read the docs BEFORE changing these settings!
+
+ha:
+  nodeuid: primary
+  readonly: false

server/app/data.yml

@@ -21,6 +21,12 @@ defaults:
       db: 0
       password: null
     workers: 0
+    ha:
+      nodeuid: primary
+      readonly: false
+queues:
+  - gitSync
+  - uplClearTemp
 authProviders:
   - local
   - microsoft

server/index.js

@@ -48,18 +48,27 @@ if (numWorkers > numCPUs) {
 }
 
 if (cluster.isMaster) {
+  wiki.logger.info('--------------------------')
   wiki.logger.info('Wiki.js is initializing...')
+  wiki.logger.info('--------------------------')
 
-  require('./master')
+  require('./master').then(() => {
+    // -> Create background workers
+    for (let i = 0; i < numWorkers; i++) {
+      cluster.fork()
+    }
 
-  for (let i = 0; i < numWorkers; i++) {
+    // -> Queue post-init tasks
-    cluster.fork()
+
-  }
+    wiki.queue.uplClearTemp.add({}, {
+      repeat: { cron: '*/15 * * * *' }
+    })
+  })
 
   cluster.on('exit', (worker, code, signal) => {
-    wiki.logger.info(`Worker #${worker.id} died.`)
+    wiki.logger.info(`Background Worker #${worker.id} was terminated.`)
   })
 } else {
-  wiki.logger.info(`Background Worker #${cluster.worker.id} is starting...`)
+  wiki.logger.info(`Background Worker #${cluster.worker.id} is initializing...`)
   require('./worker')
 }

server/master.js

@@ -4,20 +4,23 @@
 
 const Promise = require('bluebird')
 
+wiki.redis = require('./modules/redis').init()
+wiki.queue = require('./modules/queue').init()
+
 module.exports = Promise.join(
   wiki.db.onReady,
-  wiki.configSvc.loadFromDb()
+  wiki.configSvc.loadFromDb(),
+  wiki.queue.clean()
 ).then(() => {
   // ----------------------------------------
   // Load global modules
   // ----------------------------------------
 
   wiki.disk = require('./modules/disk').init()
-  wiki.entries = require('./modules/entries').init()
+  wiki.docs = require('./modules/documents').init()
   wiki.git = require('./modules/git').init(false)
   wiki.lang = require('i18next')
   wiki.mark = require('./modules/markdown')
-  wiki.redis = require('./modules/redis').init()
   wiki.search = require('./modules/search').init()
   wiki.upl = require('./modules/uploads').init()
 
@@ -75,7 +78,7 @@ module.exports = Promise.join(
   // Passport Authentication
   // ----------------------------------------
 
-  require('./modules/auth')(passport)
+  require('./modules/auth').init(passport)
   wiki.rights = require('./modules/rights')
   // wiki.rights.init()
 

server/modules/auth.js

@@ -4,87 +4,89 @@
 
 const _ = require('lodash')
 
-module.exports = (passport) => {
+module.exports = {
+  init(passport) {
   // Serialization user methods
 
-  passport.serializeUser(function (user, done) {
+    passport.serializeUser(function (user, done) {
-    done(null, user._id)
+      done(null, user._id)
-  })
-
-  passport.deserializeUser(function (id, done) {
-    wiki.db.User.findById(id).then((user) => {
-      if (user) {
-        done(null, user)
-      } else {
-        done(new Error(wiki.lang.t('auth:errors:usernotfound')), null)
-      }
-      return true
-    }).catch((err) => {
-      done(err, null)
     })
-  })
 
-  // Load authentication strategies
+    passport.deserializeUser(function (id, done) {
-
+      wiki.db.User.findById(id).then((user) => {
-  wiki.config.authStrategies = {
+        if (user) {
-    list: _.pickBy(wiki.config.auth, strategy => strategy.enabled),
+          done(null, user)
-    socialEnabled: (_.chain(wiki.config.auth).omit('local').filter(['enabled', true]).value().length > 0)
+        } else {
-  }
+          done(new Error(wiki.lang.t('auth:errors:usernotfound')), null)
-
+        }
-  _.forOwn(wiki.config.authStrategies.list, (strategyConfig, strategyName) => {
-    strategyConfig.callbackURL = `${wiki.config.site.host}/login/${strategyName}/callback`
-    require(`../authentication/${strategyName}`)(passport, strategyConfig)
-    wiki.logger.info(`Authentication Provider ${_.upperFirst(strategyName)}: OK`)
-  })
-
-  // Create Guest account for first-time
-
-  return wiki.db.User.findOne({
-    where: {
-      provider: 'local',
-      email: 'guest@example.com'
-    }
-  }).then((c) => {
-    if (c < 1) {
-      return wiki.db.User.create({
-        provider: 'local',
-        email: 'guest@example.com',
-        name: 'Guest',
-        password: '',
-        role: 'guest'
-      }).then(() => {
-        wiki.logger.info('[AUTH] Guest account created successfully!')
         return true
       }).catch((err) => {
-        wiki.logger.error('[AUTH] An error occured while creating guest account:')
+        done(err, null)
-        wiki.logger.error(err)
-        return err
       })
+    })
+
+    // Load authentication strategies
+
+    wiki.config.authStrategies = {
+      list: _.pickBy(wiki.config.auth, strategy => strategy.enabled),
+      socialEnabled: (_.chain(wiki.config.auth).omit('local').filter(['enabled', true]).value().length > 0)
     }
-  })
 
-  // .then(() => {
+    _.forOwn(wiki.config.authStrategies.list, (strategyConfig, strategyName) => {
-  //   if (process.env.WIKI_JS_HEROKU) {
+      strategyConfig.callbackURL = `${wiki.config.site.host}/login/${strategyName}/callback`
-  //     return wiki.db.User.findOne({ provider: 'local', email: process.env.WIKI_ADMIN_EMAIL }).then((c) => {
+      require(`../authentication/${strategyName}`)(passport, strategyConfig)
-  //       if (c < 1) {
+      wiki.logger.info(`Authentication Provider ${_.upperFirst(strategyName)}: OK`)
-  //         // Create root admin account (HEROKU ONLY)
+    })
 
-  //         return wiki.db.User.create({
+    // Create Guest account for first-time
-  //           provider: 'local',
+
-  //           email: process.env.WIKI_ADMIN_EMAIL,
+    return wiki.db.User.findOne({
-  //           name: 'Administrator',
+      where: {
-  //           password: '$2a$04$MAHRw785Xe/Jd5kcKzr3D.VRZDeomFZu2lius4gGpZZ9cJw7B7Mna', // admin123 (default)
+        provider: 'local',
-  //           role: 'admin'
+        email: 'guest@example.com'
-  //         }).then(() => {
+      }
-  //           wiki.logger.info('[AUTH] Root admin account created successfully!')
+    }).then((c) => {
-  //           return true
+      if (c < 1) {
-  //         }).catch((err) => {
+        return wiki.db.User.create({
-  //           wiki.logger.error('[AUTH] An error occured while creating root admin account:')
+          provider: 'local',
-  //           wiki.logger.error(err)
+          email: 'guest@example.com',
-  //           return err
+          name: 'Guest',
-  //         })
+          password: '',
-  //       } else { return true }
+          role: 'guest'
-  //     })
+        }).then(() => {
-  //   } else { return true }
+          wiki.logger.info('[AUTH] Guest account created successfully!')
-  // })
+          return true
+        }).catch((err) => {
+          wiki.logger.error('[AUTH] An error occured while creating guest account:')
+          wiki.logger.error(err)
+          return err
+        })
+      }
+    })
+
+    // .then(() => {
+    //   if (process.env.WIKI_JS_HEROKU) {
+    //     return wiki.db.User.findOne({ provider: 'local', email: process.env.WIKI_ADMIN_EMAIL }).then((c) => {
+    //       if (c < 1) {
+    //         // Create root admin account (HEROKU ONLY)
+
+    //         return wiki.db.User.create({
+    //           provider: 'local',
+    //           email: process.env.WIKI_ADMIN_EMAIL,
+    //           name: 'Administrator',
+    //           password: '$2a$04$MAHRw785Xe/Jd5kcKzr3D.VRZDeomFZu2lius4gGpZZ9cJw7B7Mna', // admin123 (default)
+    //           role: 'admin'
+    //         }).then(() => {
+    //           wiki.logger.info('[AUTH] Root admin account created successfully!')
+    //           return true
+    //         }).catch((err) => {
+    //           wiki.logger.error('[AUTH] An error occured while creating root admin account:')
+    //           wiki.logger.error(err)
+    //           return err
+    //         })
+    //       } else { return true }
+    //     })
+    //   } else { return true }
+    // })
+  }
 }

server/modules/config.js

@@ -13,9 +13,6 @@ module.exports = {
 
   /**
    * Load root config from disk
-   *
-   * @param {any} confPaths
-   * @returns
    */
   init() {
     let confPaths = {

server/modules/db.js

@@ -63,12 +63,32 @@ module.exports = {
 
     require(path.join(dbModelsPath, '_relations.js'))(self)
 
-    // Sync DB
+    // Set init tasks
 
-    self.onReady = (wiki.IS_MASTER) ? self.inst.sync({
+    let initTasks = {
-      force: false,
+      // -> Sync DB Schemas
-      logging: false
+      syncSchemas() {
-    }) : Promise.resolve()
+        return self.inst.sync({
+          force: false,
+          logging: false
+        })
+      },
+      // -> Set Connection App Name
+      setAppName() {
+        return self.inst.query(`set application_name = 'Wiki.js'`, { raw: true })
+      }
+    }
+
+    let initTasksQueue = (wiki.IS_MASTER) ? [
+      initTasks.syncSchemas,
+      initTasks.setAppName
+    ] : [
+      initTasks.setAppName
+    ]
+
+    // Perform init tasks
+
+    self.onReady = Promise.each(initTasksQueue, t => t())
 
     return self
   }

server/modules/documents.js

@@ -10,7 +10,7 @@ const _ = require('lodash')
 const entryHelper = require('../helpers/entry')
 
 /**
- * Entries Model
+ * Documents Model
  */
 module.exports = {
 

server/modules/queue.js

@@ -0,0 +1,37 @@
+'use strict'
+
+/* global wiki */
+
+const Bull = require('bull')
+const Promise = require('bluebird')
+
+module.exports = {
+  init() {
+    wiki.data.queues.forEach(queueName => {
+      this[queueName] = new Bull(queueName, {
+        prefix: `q-${wiki.config.ha.nodeuid}`,
+        redis: wiki.config.redis
+      })
+    })
+    return this
+  },
+  clean() {
+    return Promise.each(wiki.data.queues, queueName => {
+      return new Promise((resolve, reject) => {
+        let keyStream = wiki.redis.scanStream({
+          match: `q-${wiki.config.ha.nodeuid}:${queueName}:*`
+        })
+        keyStream.on('data', resultKeys => {
+          if (resultKeys.length > 0) {
+            wiki.redis.del(resultKeys)
+          }
+        })
+        keyStream.on('end', resolve)
+      })
+    }).then(() => {
+      wiki.logger.info('Purging old queue jobs: OK')
+    }).catch(err => {
+      wiki.logger.error(err)
+    })
+  }
+}

server/modules/redis.js

@@ -19,7 +19,11 @@ module.exports = {
    */
   init() {
     if (isPlainObject(wiki.config.redis)) {
-      return new Redis(wiki.config.redis)
+      let red = new Redis(wiki.config.redis)
+      red.on('ready', () => {
+        wiki.logger.info('Redis connection: OK')
+      })
+      return red
     } else {
       wiki.logger.error('Invalid Redis configuration!')
       process.exit(1)

server/queues/git-sync.js

@@ -61,5 +61,8 @@ module.exports = (job, done) => {
     })
 
     return jobCbStreamDocs
+  }).then(() => {
+    wiki.logger.info('Git remote repository sync: DONE')
+    return true
   })
 }

server/queues/upl-clear-temp.js

@@ -22,5 +22,8 @@ module.exports = (job, done) => {
         }
       })
     })
+  }).then(() => {
+    wiki.logger.info('Purging temporary upload files: DONE')
+    return true
   })
 }

server/worker.js

@@ -2,69 +2,68 @@
 
 /* global wiki */
 
-const path = require('path')
-
-wiki.REPOPATH = path.resolve(wiki.ROOTPATH, wiki.config.paths.repo)
-wiki.DATAPATH = path.resolve(wiki.ROOTPATH, wiki.config.paths.data)
-wiki.UPLTEMPPATH = path.join(wiki.DATAPATH, 'temp-upload')
-
-// ----------------------------------------
-// Load global modules
-// ----------------------------------------
-
-// wiki.upl = require('./modules/uploads-agent').init()
-// wiki.git = require('./modules/git').init()
-// wiki.entries = require('./modules/entries').init()
-wiki.lang = require('i18next')
-wiki.mark = require('./modules/markdown')
-
-// ----------------------------------------
-// Load local modules
-// ----------------------------------------
-
 const Promise = require('bluebird')
-const i18nBackend = require('i18next-node-fs-backend')
 
-// ----------------------------------------
+module.exports = Promise.join(
-// Localization Engine
+  wiki.db.onReady,
-// ----------------------------------------
+  wiki.configSvc.loadFromDb(['features', 'git', 'logging', 'site', 'uploads'])
-
-wiki.lang.use(i18nBackend).init({
-  load: 'languageOnly',
-  ns: ['common', 'admin', 'auth', 'errors', 'git'],
-  defaultNS: 'common',
-  saveMissing: false,
-  preload: [wiki.config.lang],
-  lng: wiki.config.lang,
-  fallbackLng: 'en',
-  backend: {
-    loadPath: path.join(wiki.SERVERPATH, 'locales/{{lng}}/{{ns}}.json')
-  }
-})
-
-// ----------------------------------------
-// Start Queues
-// ----------------------------------------
-
-const Bull = require('bull')
-const autoload = require('auto-load')
-
-let queues = autoload(path.join(wiki.SERVERPATH, 'queues'))
-
-Promise.join(
-  wiki.db.onReady
-  // wiki.upl.initialScan()
 ).then(() => {
+  const path = require('path')
+
+  wiki.REPOPATH = path.resolve(wiki.ROOTPATH, wiki.config.paths.repo)
+  wiki.DATAPATH = path.resolve(wiki.ROOTPATH, wiki.config.paths.data)
+  wiki.UPLTEMPPATH = path.join(wiki.DATAPATH, 'temp-upload')
+
+  // ----------------------------------------
+  // Load global modules
+  // ----------------------------------------
+
+  // wiki.upl = require('./modules/uploads-agent').init()
+  // wiki.git = require('./modules/git').init()
+  // wiki.entries = require('./modules/entries').init()
+  wiki.lang = require('i18next')
+  wiki.mark = require('./modules/markdown')
+
+  // ----------------------------------------
+  // Localization Engine
+  // ----------------------------------------
+
+  const i18nBackend = require('i18next-node-fs-backend')
+  wiki.lang.use(i18nBackend).init({
+    load: 'languageOnly',
+    ns: ['common', 'admin', 'auth', 'errors', 'git'],
+    defaultNS: 'common',
+    saveMissing: false,
+    preload: [wiki.config.lang],
+    lng: wiki.config.lang,
+    fallbackLng: 'en',
+    backend: {
+      loadPath: path.join(wiki.SERVERPATH, 'locales/{{lng}}/{{ns}}.json')
+    }
+  })
+
+  // ----------------------------------------
+  // Start Queues
+  // ----------------------------------------
+
+  const Bull = require('bull')
+  const autoload = require('auto-load')
+
+  let queues = autoload(path.join(wiki.SERVERPATH, 'queues'))
+
   for (let queueName in queues) {
-    new Bull(queueName, { redis: wiki.config.redis }).process(queues[queueName])
+    new Bull(queueName, {
+      prefix: `q-${wiki.config.ha.nodeuid}`,
+      redis: wiki.config.redis
+    }).process(queues[queueName])
   }
-})
 
-// ----------------------------------------
+  // ----------------------------------------
-// Shutdown gracefully
+  // Shutdown gracefully
-// ----------------------------------------
+  // ----------------------------------------
 
-process.on('disconnect', () => {
+  process.on('disconnect', () => {
-  wiki.logger.warn('Lost connection to Master. Exiting...')
+    wiki.logger.warn('Lost connection to Master. Exiting...')
-  process.exit()
+    process.exit()
+  })
 })