gulp.task('webpack', function() {
  const { result$, next, end } = parallelRunner(
    __filename,
    function(worker, configName) {
      return Promise.resolve(
        worker.compile(
          null,
          [configName]
        )
      )
    },
    isProduction ? PRODUCTION_PARALLEL_NUMBER : true
  )
  Object.keys(configNameToFileMap).forEach((name) => next(name))
  end()
  return result$
})

return new Promise((resolve, reject) => {
  webpack(configs, function(error, stats) {
    let errors
    if (stats) {
      log(stats.toString(Object.assign({
        // https://webpack.js.org/configuration/stats/
        chunks: false,
        colors: colors.supportsColor,
      }, configs[0].stats)))
    }

    if (error || stats.hasErrors()) {
      // errors should be serializable since it will go through processes
      errors = error ? error.toString() : stats.toJson().errors
    }

    if (callback) {
      callback(errors)
    }
    if (errors) {
      reject(errors)
    } else {
      resolve()
    }
  })
})

// Inspired by terser-webpack-plugin.

const os = require('os')
const rx = require('rxjs')
const rxOperators = require('rxjs/operators')
const JestWorker = require('jest-worker').default
const through2 = require('through2')
const log = require('fancy-log')

const { Subject, from } = rx
const { mergeMap, share } = rxOperators

function getAvailableNumberOfCores(parallel) {
  const cpus = os.cpus() || { length: 1 }

  return parallel === true
    ? cpus.length - 1
    : Math.min(Number(parallel) || 0, cpus.length - 1)
}

function parallelRunner(
  module,
  taskCallback,
  parallel = true
) {
  const availableNumberOfCores = getAvailableNumberOfCores(parallel)
  let concurrency = 1 // which means mergeMap will behave as concatMap
  let worker
  let total = 0
  let completed = 0
  let allScheduled = false

  log('Parallel Running: ', module)
  log('Available Number of Cores: ', availableNumberOfCores)

  // setup worker
  if (availableNumberOfCores > 0) {
    const numWorkers = availableNumberOfCores
    concurrency = numWorkers
    log('Number of Workers: ', numWorkers)
    worker = new JestWorker(module, { numWorkers, enableWorkerThreads: true })

    const workerStdout = worker.getStdout()
    if (workerStdout) {
      workerStdout.on('data', (chunk) => {
        return process.stdout.write(chunk)
      })
    }

    const workerStderr = worker.getStderr()

    if (workerStderr) {
      workerStderr.on('data', (chunk) => {
        return process.stderr.write(chunk)
      })
    }
  }

  // handle concurrency with rxjs
  const scheduled = new Subject()
  const consumed = new Subject()

  const result$ = scheduled.pipe(
    mergeMap((data) => {
      // data is actually consumed here
      consumed.next(null)
      // worker[methodName] can only be invoked with serializable data
      // and returned value could be just plain RESULT or Promise<RESULT>
      return from(taskCallback(worker || require(module), data))
    }, concurrency),
    share()
  )
  result$.subscribe({
    complete: function() {
      if (worker) {
        worker.end()
      }
    },
    next: function() {
      completed += 1
      if (allScheduled && completed === total) {
        scheduled.complete()
      }
    },
    error: function(err) {
      throw err
    }
  })

  return {
    result$,
    consumed$: consumed.asObservable(),
    next: (data) => {
      scheduled.next(data)
      total += 1
    },
    complete: () => { allScheduled = true }
  }
}

function gulpParallelRunner(module, taskCallback, parallel) {
  const {
    result$,
    consumed$,
    next,
    complete
  } = parallelRunner(module, taskCallback, parallel)
  let afterComplete, stream, afterConsume

  consumed$.subscribe(() => {
    // `afterComplete was defined` means there is no more data
    if (!afterComplete && afterConsume) {
      afterConsume()
    }
  })

  result$.subscribe({
    complete: () => {
      if (afterComplete) {
        afterComplete()
      }
    },
    next: (data) => {
      stream.push(data)
      // if returned value is false means stream ends or meets highWaterMark
      // but we don't care since we use rxjs to control concurrency
    }
  })

  const flush = function(cb) {
    afterComplete = cb
    complete()
  }
  const transform = function(file, enc, afterTransform) {
    if (!stream) {
      stream = this
    }
    if (!afterConsume) {
      afterConsume = afterTransform
    }
    next(file)
  }
  return through2.obj(transform, flush)
}

// Staticng has CPU limit of 5 on k8s, so we can't use os.cpus().length which
// reports the number of online CPUs, but running with 4 threads is fast enough.
// https://github.com/nodejs/node/issues/28762#issuecomment-513730856
const PRODUCTION_PARALLEL_NUMBER = 4

module.exports = {
  parallelRunner,
  gulpParallelRunner,
  PRODUCTION_PARALLEL_NUMBER,
}

function parallelTerser(needMangle) {
  const options = generateTerserOptions(needMangle)
  return gulpParallelRunner(
    require.resolve('terser'),
    function(worker, file) {
      return Promise.resolve(worker.minify({
        [file.path]: file.contents.toString('utf-8')
      }, options)).then((result) => {
        if ('error' in result) {
          throw new Error(result.error.message)
        }
        file.contents = 'from' in Buffer ? Buffer.from(result.code) : new Buffer(result.code)
        return file
      })
    },
    isProduction ? PRODUCTION_PARALLEL_NUMBER : true
  )
}