feat: scheduler + worker pool

config.sample.yml

@@ -122,3 +122,11 @@ dataPath: ./data
 # file uploads.
 
 bodyParserLimit: 5mb
+
+# ---------------------------------------------------------------------
+# Workers Limit
+# ---------------------------------------------------------------------
+# Maximum number of workers that can run background cpu-intensive jobs.
+# Leave to 'auto' to use CPU cores count as maximum.
+
+workers: auto

package.json

@@ -88,7 +88,6 @@
     "he": "1.2.0",
     "highlight.js": "10.3.1",
     "i18next": "19.8.3",
-    "i18next-express-middleware": "2.0.0",
     "i18next-node-fs-backend": "2.1.3",
     "image-size": "0.9.2",
     "js-base64": "3.7.2",
@@ -153,6 +152,7 @@
     "pg-pubsub": "0.8.0",
     "pg-query-stream": "4.2.4",
     "pg-tsquery": "8.4.0",
+    "poolifier": "2.2.0",
     "pug": "3.0.2",
     "punycode": "2.1.1",
     "puppeteer-core": "17.1.3",

server/app/data.yml

@@ -27,34 +27,16 @@ defaults:
     logLevel: info
     logFormat: default
     offline: false
+    dataPath: ./data
     bodyParserLimit: 5mb
+    workers: auto
     # DB defaults
     api:
       isEnabled: false
-    graphEndpoint: 'https://graph.requarks.io'
-    lang:
-      code: en
-      autoUpdate: true
-      namespaces: []
-      namespacing: false
-      rtl: false
-    telemetry:
-      clientId: ''
-      isEnabled: false
-    title: Wiki.js
-    company: ''
-    contentLicense: ''
-    logoUrl: https://static.requarks.io/logo/wikijs-butterfly.svg
     mail:
       host: ''
       secure: true
       verifySSL: true
-    nav:
-      mode: 'MIXED'
-    theming:
-      theme: 'default'
-      iconset: 'md'
-      darkMode: false
     auth:
       autoLogin: false
       enforce2FA: false
@@ -63,34 +45,30 @@ defaults:
       audience: 'urn:wiki.js'
       tokenExpiration: '30m'
       tokenRenewal: '14d'
-    features:
-      featurePageRatings: true
-      featurePageComments: true
-      featurePersonalWikis: true
     security:
-      securityOpenRedirect: true
-      securityIframe: true
-      securityReferrerPolicy: true
-      securityTrustProxy: true
-      securitySRI: true
-      securityHSTS: false
-      securityHSTSDuration: 300
-      securityCSP: false
-      securityCSPDirectives: ''
-    server:
-      sslRedir: false
-    uploads:
-      maxFileSize: 5242880
-      maxFiles: 10
-      scanSVG: true
-      forceDownload: true
+      corsMode: 'OFF'
+      corsConfig: ''
+      enforceCsp: false
+      trustProxy: false
+      enforceHsts: false
+      disallowFloc: true
+      hstsDuration: 0
+      cspDirectives: ''
+      uploadScanSVG: true
+      disallowIframe: true
+      uploadMaxFiles: 20
+      authJwtAudience: 'urn:wiki.js'
+      authJwtExpiration: '30m'
+      uploadMaxFileSize: 10485760
+      forceAssetDownload: true
+      disallowOpenRedirect: true
+      authJwtRenewablePeriod: '14d'
+      enforceSameOriginReferrerPolicy: true
     flags:
       ldapdebug: false
       sqllog: false
     # System defaults
     channel: NEXT
-    setup: false
-    dataPath: ./data
     cors:
       credentials: true
       maxAge: 600
@@ -99,10 +77,6 @@ defaults:
     search:
       maxHits: 100
     maintainerEmail: security@requarks.io
-localeNamespaces:
-  - admin
-  - auth
-  - common
 jobs:
   purgeUploads:
     onInit: true

server/core/kernel.js

 const _ = require('lodash')
 const EventEmitter = require('eventemitter2').EventEmitter2
 
+let isShuttingDown = false
+
 /* global WIKI */
 
 module.exports = {
@@ -31,9 +33,8 @@ module.exports = {
    */
   async preBootWeb() {
     try {
-      WIKI.sideloader = await require('./sideloader').init()
       WIKI.cache = require('./cache').init()
-      WIKI.scheduler = require('./scheduler').init()
+      WIKI.scheduler = await require('./scheduler').init()
       WIKI.servers = require('./servers')
       WIKI.events = {
         inbound: new EventEmitter(),
@@ -83,6 +84,8 @@ module.exports = {
    * Graceful shutdown
    */
   async shutdown (devMode = false) {
+    if (isShuttingDown) { return }
+    isShuttingDown = true
     if (WIKI.servers) {
       await WIKI.servers.stopServers()
     }
@@ -99,6 +102,7 @@ module.exports = {
       await WIKI.asar.unload()
     }
     if (!devMode) {
+      WIKI.logger.info('Terminating process...')
       process.exit(0)
     }
   }

server/core/localization.js

-const _ = require('lodash')
-const dotize = require('dotize')
-const i18nMW = require('i18next-express-middleware')
-const i18next = require('i18next')
-const Promise = require('bluebird')
-const fs = require('fs-extra')
-const path = require('path')
-const yaml = require('js-yaml')
-
-/* global WIKI */
-
-module.exports = {
-  engine: null,
-  namespaces: [],
-  init() {
-    this.namespaces = WIKI.data.localeNamespaces
-    this.engine = i18next
-    this.engine.init({
-      load: 'languageOnly',
-      ns: this.namespaces,
-      defaultNS: 'common',
-      saveMissing: false,
-      lng: WIKI.config.lang.code,
-      fallbackLng: 'en'
-    })
-
-    // Load current language + namespaces
-    this.refreshNamespaces(true)
-
-    return this
-  },
-  /**
-   * Attach i18n middleware for Express
-   *
-   * @param {Object} app Express Instance
-   */
-  attachMiddleware (app) {
-    app.use(i18nMW.handle(this.engine))
-  },
-  /**
-   * Get all entries for a specific locale and namespace
-   *
-   * @param {String} locale Locale code
-   * @param {String} namespace Namespace
-   */
-  async getByNamespace(locale, namespace) {
-    if (this.engine.hasResourceBundle(locale, namespace)) {
-      let data = this.engine.getResourceBundle(locale, namespace)
-      return _.map(dotize.convert(data), (value, key) => {
-        return {
-          key,
-          value
-        }
-      })
-    } else {
-      throw new Error('Invalid locale or namespace')
-    }
-  },
-  /**
-   * Load entries from the DB for a single locale
-   *
-   * @param {String} locale Locale code
-   * @param {*} opts Additional options
-   */
-  async loadLocale(locale, opts = { silent: false }) {
-    const res = await WIKI.models.locales.query().findOne('code', locale)
-    if (res) {
-      if (_.isPlainObject(res.strings)) {
-        _.forOwn(res.strings, (data, ns) => {
-          this.namespaces.push(ns)
-          this.engine.addResourceBundle(locale, ns, data, true, true)
-        })
-      }
-    } else if (!opts.silent) {
-      throw new Error('No such locale in local store.')
-    }
-
-    // -> Load dev locale files if present
-    if (WIKI.IS_DEBUG) {
-      try {
-        const devEntriesRaw = await fs.readFile(path.join(WIKI.SERVERPATH, `locales/${locale}.yml`), 'utf8')
-        if (devEntriesRaw) {
-          const devEntries = yaml.safeLoad(devEntriesRaw)
-          _.forOwn(devEntries, (data, ns) => {
-            this.namespaces.push(ns)
-            this.engine.addResourceBundle(locale, ns, data, true, true)
-          })
-          WIKI.logger.info(`Loaded dev locales from ${locale}.yml`)
-        }
-      } catch (err) {
-        // ignore
-      }
-    }
-  },
-  /**
-   * Reload all namespaces for all active locales from the DB
-   *
-   * @param {Boolean} silent No error on fail
-   */
-  async refreshNamespaces (silent = false) {
-    await this.loadLocale(WIKI.config.lang.code, { silent })
-    if (WIKI.config.lang.namespacing) {
-      for (let ns of WIKI.config.lang.namespaces) {
-        await this.loadLocale(ns, { silent })
-      }
-    }
-  },
-  /**
-   * Set the active locale
-   *
-   * @param {String} locale Locale code
-   */
-  async setCurrentLocale(locale) {
-    await Promise.fromCallback(cb => {
-      return this.engine.changeLanguage(locale, cb)
-    })
-  }
-}

server/core/scheduler.js

 const PgBoss = require('pg-boss')
+const { DynamicThreadPool } = require('poolifier')
+const os = require('node:os')
 
 /* global WIKI */
 
 module.exports = {
+  pool: null,
   scheduler: null,
-  jobs: [],
-  init () {
+  async init () {
     WIKI.logger.info('Initializing Scheduler...')
     this.scheduler = new PgBoss({
-      ...WIKI.models.knex.client.connectionSettings,
+      db: {
+        close: () => Promise.resolve('ok'),
+        executeSql: async (text, values) => {
+          try {
+            const resource = await WIKI.models.knex.client.pool.acquire().promise
+            const res = await resource.query(text, values)
+            WIKI.models.knex.client.pool.release(resource)
+            return res
+          } catch (err) {
+            WIKI.logger.error('Failed to acquire DB connection during scheduler query execution.')
+            WIKI.logger.error(err)
+          }
+        }
+      },
+      // ...WIKI.models.knex.client.connectionSettings,
       application_name: 'Wiki.js Scheduler',
       schema: WIKI.config.db.schemas.scheduler,
       uuid: 'v4'
     })
+
+    const maxWorkers = WIKI.config.workers === 'auto' ? os.cpus().length : WIKI.config.workers
+    WIKI.logger.info(`Initializing Worker Pool (Max ${maxWorkers})...`)
+    this.pool = new DynamicThreadPool(1, maxWorkers, './server/worker.js', {
+      errorHandler: (err) => WIKI.logger.warn(err),
+      exitHandler: () => WIKI.logger.debug('A worker has gone offline.'),
+      onlineHandler: () => WIKI.logger.debug('New worker is online.')
+    })
     return this
   },
   async start () {
     WIKI.logger.info('Starting Scheduler...')
     await this.scheduler.start()
+    this.scheduler.work('*', async job => {
+      return this.pool.execute({
+        id: job.id,
+        name: job.name,
+        data: job.data
+      })
+    })
     WIKI.logger.info('Scheduler: [ STARTED ]')
   },
   async stop () {
     WIKI.logger.info('Stopping Scheduler...')
     await this.scheduler.stop()
+    await this.pool.destroy()
     WIKI.logger.info('Scheduler: [ STOPPED ]')
   }
 }

server/web.js

@@ -18,7 +18,6 @@ module.exports = async () => {
   // ----------------------------------------
 
   WIKI.auth = require('./core/auth').init()
-  WIKI.lang = require('./core/localization').init()
   WIKI.mail = require('./core/mail').init()
   WIKI.system = require('./core/system').init()
 
@@ -134,12 +133,6 @@ module.exports = async () => {
   app.use(bodyParser.urlencoded({ extended: false, limit: '1mb' }))
 
   // ----------------------------------------
-  // Localization
-  // ----------------------------------------
-
-  WIKI.lang.attachMiddleware(app)
-
-  // ----------------------------------------
   // View accessible data
   // ----------------------------------------
 

server/worker.js

+const { ThreadWorker } = require('poolifier')
+
+module.exports = new ThreadWorker(async (job) => {
+  return { ok: true }
+}, { async: true })