refactor: remove bluebird + native scheduler (#5702)

package.json

@@ -52,7 +52,6 @@
     "auto-load": "3.0.4",
     "aws-sdk": "2.1208.0",
     "bcryptjs-then": "1.0.1",
-    "bluebird": "3.7.2",
     "body-parser": "1.20.0",
     "chalk": "4.1.2",
     "cheerio": "1.0.0-rc.12",
@@ -81,7 +80,6 @@
     "filesize": "6.1.0",
     "fs-extra": "9.0.1",
     "getos": "3.2.1",
-    "graphile-worker": "0.13.0",
     "graphql": "16.3.0",
     "graphql-list-fields": "2.0.2",
     "graphql-rate-limit-directive": "2.0.2",

server/app/data.yml

@@ -78,25 +78,6 @@ defaults:
     search:
       maxHits: 100
     maintainerEmail: security@requarks.io
-jobs:
-  - task: background
-    identifier: purge-uploads
-    pattern: '*/15 * * * *'
-    payload:
-      name: purgeUploads
-      data: {}
-  # - task: simple
-  #   identifier: letest
-  #   pattern: '* * * * *'
-  #   payload:
-  #     name: bob
-  #     data: {}
-  # - task: simple
-  #   identifier: letest2
-  #   pattern: '* * * * *'
-  #   payload:
-  #     name: bob
-  #     data: {}
 groups:
   defaultPermissions:
     - 'read:pages'

server/core/auth.js

@@ -4,8 +4,9 @@ const _ = require('lodash')
 const jwt = require('jsonwebtoken')
 const ms = require('ms')
 const { DateTime } = require('luxon')
-const Promise = require('bluebird')
-const crypto = Promise.promisifyAll(require('crypto'))
+const util = require('node:util')
+const crypto = require('node:crypto')
+const randomBytes = util.promisify(crypto.randomBytes)
 const pem2jwk = require('pem-jwk').pem2jwk
 
 const securityHelper = require('../helpers/security')
@@ -366,7 +367,7 @@ module.exports = {
   async regenerateCertificates () {
     WIKI.logger.info('Regenerating certificates...')
 
-    _.set(WIKI.config, 'sessionSecret', (await crypto.randomBytesAsync(32)).toString('hex'))
+    _.set(WIKI.config, 'sessionSecret', (await randomBytes(32)).toString('hex'))
     const certs = crypto.generateKeyPairSync('rsa', {
       modulusLength: 2048,
       publicKeyEncoding: {

server/core/db.js

 const _ = require('lodash')
 const autoload = require('auto-load')
 const path = require('path')
-const Promise = require('bluebird')
 const Knex = require('knex')
 const fs = require('fs')
 const Objection = require('objection')
 
 const migrationSource = require('../db/migrator-source')
 const migrateFromLegacy = require('../db/legacy')
+const { setTimeout } = require('timers/promises')
 
 /**
  * ORM DB module
@@ -118,7 +118,7 @@ module.exports = {
               WIKI.logger.error(`Database Connection Error: ${err.message}`)
             }
             WIKI.logger.warn(`Will retry in 3 seconds... [Attempt ${++conAttempts} of 10]`)
-            await new Promise(resolve => setTimeout(resolve, 3000))
+            await setTimeout(3000)
             await initTasks.connect()
           } else {
             throw err

server/core/kernel.js

@@ -74,9 +74,9 @@ module.exports = {
     await WIKI.db.commentProviders.initProvider()
     await WIKI.db.sites.reloadCache()
     await WIKI.db.storage.initTargets()
-    await WIKI.scheduler.start()
 
     await WIKI.db.subscribeToNotifications()
+    await WIKI.scheduler.start()
   },
   /**
    * Graceful shutdown

server/core/logger.js

@@ -33,6 +33,9 @@ LEVELS.forEach(lvl => {
           message: msg
         })
       } else {
+        if (msg instanceof Error) {
+          msg = msg.stack
+        }
         formatted = chalk`${new Date().toISOString()} {dim [${WIKI.INSTANCE_ID}]} {${LEVELCOLORS[lvl]}.bold ${lvl}}: ${msg}`
       }
 

server/core/scheduler.js

 const { DynamicThreadPool } = require('poolifier')
-const { v4: uuid } = require('uuid')
 const os = require('node:os')
-const path = require('node:path')
+const { setTimeout } = require('node:timers/promises')
 
 module.exports = {
   pool: null,
-  runner: null,
   maxWorkers: 1,
+  activeWorkers: 0,
   async init () {
     this.maxWorkers = WIKI.config.scheduler.workers === 'auto' ? os.cpus().length : WIKI.config.scheduler.workers
     WIKI.logger.info(`Initializing Worker Pool (Limit: ${this.maxWorkers})...`)
@@ -19,46 +18,55 @@ module.exports = {
   },
   async start () {
     WIKI.logger.info('Starting Scheduler...')
-    // this.runner = await run({
-    //   pgPool: new Pool({
-    //     ...(typeof WIKI.db.config === 'string') ? {
-    //       connectionString: WIKI.db.config
-    //     } : WIKI.db.config,
-    //     max: this.maxWorkers + 2
-    //   }),
-    //   schema: WIKI.config.db.schemas.scheduler,
-    //   concurrency: this.maxWorkers,
-    //   noHandleSignals: true,
-    //   logger: new Logger(scope => {
-    //     return (level, message, meta) => {
-    //       const prefix = (scope?.workerId) ? `[${scope.workerId}] ` : ''
-    //       WIKI.logger[level](`${prefix}${message}`, meta)
-    //     }
-    //   }),
-    //   parsedCronItems: parseCronItems(WIKI.data.jobs),
-    //   taskList: {
-    //     simple: async (payload, helpers) => {
-    //       // TODO: Handle task
-    //     },
-    //     background: async (payload, helpers) => {
-    //       try {
-    //         await this.pool.execute({
-    //           id: helpers.job.id,
-    //           name: payload.name,
-    //           data: payload.data
+    WIKI.db.listener.addChannel('scheduler', payload => {
+      switch (payload.event) {
+        case 'newJob': {
+          if (this.activeWorkers < this.maxWorkers) {
+            this.activeWorkers++
+            this.processJob()
+          }
+          break
+        }
+      }
+    })
+    // await WIKI.db.knex('jobs').insert({
+    //   task: 'test',
+    //   payload: { foo: 'bar' }
     // })
-    //       } catch (err) {
-    //         helpers.logger.warn(`Failed job: ${err.message}`)
-    //         throw err
-    //       }
-    //     }
-    //   }
+    // WIKI.db.listener.publish('scheduler', {
+    //   source: WIKI.INSTANCE_ID,
+    //   event: 'newJob'
     // })
     WIKI.logger.info('Scheduler: [ STARTED ]')
   },
+  async processJob () {
+    try {
+      await WIKI.db.knex.transaction(async trx => {
+        const jobs = await trx('jobs')
+          .where('id', WIKI.db.knex.raw('(SELECT id FROM jobs ORDER BY id FOR UPDATE SKIP LOCKED LIMIT 1)'))
+          .returning('*')
+          .del()
+        if (jobs && jobs.length === 1) {
+          const job = jobs[0]
+          WIKI.logger.info(`Processing new job ${job.id}: ${job.task}...`)
+          if (job.useWorker) {
+            await this.pool.execute({
+              id: job.id,
+              name: job.task,
+              data: job.payload
+            })
+          } else {
+
+          }
+        }
+      })
+    } catch (err) {
+      WIKI.logger.warn(err)
+    }
+  },
   async stop () {
     WIKI.logger.info('Stopping Scheduler...')
-    // await this.runner.stop()
+    await this.pool.stop()
     WIKI.logger.info('Scheduler: [ STOPPED ]')
   }
 }

server/core/servers.js

@@ -2,7 +2,6 @@ const fs = require('fs-extra')
 const http = require('http')
 const https = require('https')
 const { ApolloServer } = require('apollo-server-express')
-const Promise = require('bluebird')
 const _ = require('lodash')
 const io = require('socket.io')
 const { ApolloServerPluginLandingPageGraphQLPlayground, ApolloServerPluginLandingPageProductionDefault } = require('apollo-server-core')
@@ -188,11 +187,11 @@ module.exports = {
   async stopServers () {
     this.closeConnections()
     if (this.http) {
-      await Promise.fromCallback(cb => { this.http.close(cb) })
+      await new Promise(resolve => this.http.close(resolve))
       this.http = null
     }
     if (this.https) {
-      await Promise.fromCallback(cb => { this.https.close(cb) })
+      await new Promise(resolve => this.https.close(resolve))
       this.https = null
     }
     this.graph = null
@@ -205,7 +204,7 @@ module.exports = {
     switch (srv) {
       case 'http':
         if (this.http) {
-          await Promise.fromCallback(cb => { this.http.close(cb) })
+          await new Promise(resolve => this.http.close(resolve))
           this.http = null
         }
         this.initHTTP()
@@ -213,7 +212,7 @@ module.exports = {
         break
       case 'https':
         if (this.https) {
-          await Promise.fromCallback(cb => { this.https.close(cb) })
+          await new Promise(resolve => this.https.close(resolve))
           this.https = null
         }
         this.initHTTPS()

server/db/migrations/3.0.0.js

@@ -145,7 +145,9 @@ exports.up = async knex => {
     .createTable('jobs', table => {
       table.uuid('id').notNullable().primary().defaultTo(knex.raw('gen_random_uuid()'))
       table.string('task').notNullable()
+      table.boolean('useWorker').notNullable().defaultTo(false)
       table.jsonb('payload')
+      table.timestamp('waitUntil')
       table.timestamp('createdAt').notNullable().defaultTo(knex.fn.now())
       table.timestamp('updatedAt').notNullable().defaultTo(knex.fn.now())
     })

server/graph/resolvers/system.js

 const _ = require('lodash')
-const Promise = require('bluebird')
-const getos = Promise.promisify(require('getos'))
-const os = require('os')
+const util = require('node:util')
+const getos = util.promisify(require('getos'))
+const os = require('node:os')
 const filesize = require('filesize')
 const path = require('path')
 const fs = require('fs-extra')
@@ -105,12 +105,12 @@ module.exports = {
     currentVersion () {
       return WIKI.version
     },
-    async dbVersion () {
-      return _.get(WIKI.models, 'knex.client.version', 'Unknown Version')
-    },
     dbHost () {
       return WIKI.config.db.host
     },
+    dbVersion () {
+      return _.get(WIKI.db, 'knex.client.version', 'Unknown Version')
+    },
     hostname () {
       return os.hostname()
     },

server/helpers/security.js

-const Promise = require('bluebird')
-const crypto = require('crypto')
+const util = require('node:util')
+const crypto = require('node:crypto')
+const randomBytes = util.promisify(crypto.randomBytes)
 const passportJWT = require('passport-jwt')
 
 module.exports = {
@@ -17,11 +18,7 @@ module.exports = {
    * @returns
    */
   async generateToken (length) {
-    return Promise.fromCallback(clb => {
-      crypto.randomBytes(length, clb)
-    }).then(buf => {
-      return buf.toString('hex')
-    })
+    return (await randomBytes(length)).toString('hex')
   },
 
   extractJWT: passportJWT.ExtractJwt.fromExtractors([

server/models/assets.js

-/* global WIKI */
-
 const Model = require('objection').Model
 const moment = require('moment')
 const path = require('path')
 const fs = require('fs-extra')
 const _ = require('lodash')
 const assetHelper = require('../helpers/asset')
-const Promise = require('bluebird')
 
 /**
  * Users model
@@ -199,9 +196,8 @@ module.exports = class Asset extends Model {
     } catch (err) {
       return false
     }
-    const sendFile = Promise.promisify(res.sendFile, {context: res})
     res.type(path.extname(assetPath))
-    await sendFile(cachePath, { dotfiles: 'deny' })
+    await new Promise(resolve => res.sendFile(cachePath, { dotfiles: 'deny' }, resolve))
     return true
   }
 

server/modules/storage/azure/storage.js

 const { BlobServiceClient, StorageSharedKeyCredential } = require('@azure/storage-blob')
 const stream = require('stream')
-const Promise = require('bluebird')
-const pipeline = Promise.promisify(stream.pipeline)
+const util = require('node:util')
+const pipeline = util.promisify(stream.pipeline)
 const pageHelper = require('../../../helpers/page.js')
 const _ = require('lodash')
 

server/modules/storage/disk/common.js

 const fs = require('fs-extra')
-const path = require('path')
-const stream = require('stream')
-const Promise = require('bluebird')
-const pipeline = Promise.promisify(stream.pipeline)
+const path = require('node:path')
+const stream = require('node:stream')
+const util = require('node:util')
+const pipeline = util.promisify(stream.pipeline)
 const klaw = require('klaw')
 const mime = require('mime-types').lookup
 const _ = require('lodash')

server/modules/storage/disk/storage.js

 const fs = require('fs-extra')
-const path = require('path')
+const path = require('node:path')
 const tar = require('tar-fs')
-const zlib = require('zlib')
-const stream = require('stream')
+const zlib = require('node:zlib')
+const stream = require('node:stream')
 const _ = require('lodash')
-const Promise = require('bluebird')
-const pipeline = Promise.promisify(stream.pipeline)
+const util = require('node:util')
+const pipeline = util.promisify(stream.pipeline)
 const moment = require('moment')
 
 const pageHelper = require('../../../helpers/page')

server/modules/storage/gcs/storage.js

 const { BlobServiceClient, StorageSharedKeyCredential } = require('@azure/storage-blob')
-const stream = require('stream')
-const Promise = require('bluebird')
-const pipeline = Promise.promisify(stream.pipeline)
+const stream = require('node:stream')
+const util = require('node:util')
+const pipeline = util.promisify(stream.pipeline)
 const pageHelper = require('../../../helpers/page.js')
 const _ = require('lodash')
 

server/modules/storage/git/storage.js

-const path = require('path')
+const path = require('node:path')
 const sgit = require('simple-git/promise')
 const fs = require('fs-extra')
 const _ = require('lodash')
-const stream = require('stream')
-const Promise = require('bluebird')
-const pipeline = Promise.promisify(stream.pipeline)
+const stream = require('node:stream')
+const util = require('node:util')
+const pipeline = util.promisify(stream.pipeline)
 const klaw = require('klaw')
-const os = require('os')
+const os = require('node:os')
 
 const pageHelper = require('../../../helpers/page')
 const assetHelper = require('../../../helpers/asset')

server/modules/storage/s3/storage.js

 const S3 = require('aws-sdk/clients/s3')
-const stream = require('stream')
-const Promise = require('bluebird')
-const pipeline = Promise.promisify(stream.pipeline)
+const stream = require('node:stream')
+const util = require('node:util')
+const pipeline = util.promisify(stream.pipeline)
 const _ = require('lodash')
 const pageHelper = require('../../../helpers/page.js')
 

server/modules/storage/sftp/storage.js

 const SSH2Promise = require('ssh2-promise')
 const _ = require('lodash')
-const path = require('path')
-const stream = require('stream')
-const Promise = require('bluebird')
-const pipeline = Promise.promisify(stream.pipeline)
+const path = require('node:path')
+const stream = require('node:stream')
+const util = require('node:util')
+const pipeline = util.promisify(stream.pipeline)
 const pageHelper = require('../../../helpers/page.js')
 
 const getFilePath = (page, pathKey) => {