Merge branch 'next' into user-status-refactor

bin/cli-jobs

@@ -12,7 +12,7 @@ const mongoose = require('../services/mongoose');
 const kue = require('../services/kue');
 
 util.onshutdown([
-  () => mongoose.disconnect()
+  () => mongoose.disconnect(),
 ]);
 
 /**
@@ -20,17 +20,17 @@ util.onshutdown([
  */
 function processJobs() {
 
-  // Start the scraper processor.
-  scraper.process();
-
-  // Start the mail processor.
-  mailer.process();
-
   // The scraper only needs to shutdown when the scraper has actually been
   // started.
   util.onshutdown([
     () => kue.Task.shutdown()
   ]);
+
+  // Start the scraper processor.
+  scraper.process();
+
+  // Start the mail processor.
+  mailer.process();
 }
 
 /**
@@ -48,22 +48,13 @@ function removeJob(job) {
   }));
 }
 
-/**
- * Removes the jobs passed in and returns a promise.
- * @param  {Array} jobs array of jobs
- * @return {Promise}
- */
-function removeJobs(jobs) {
-  return Promise.all(jobs.map(removeJob));
-}
-
 /**
  * Get the top n jobs with a specific state.
  * @param  {String} [state='complete'] state to list jobs by
  * @param  {Number} limit              limit of jobs to load
  * @return {Promise}
  */
-function rangeJobsByState(state = 'complete', limit) {
+function rangeJobsByState(state, limit) {
   return new Promise((resolve, reject) => {
     kue.Job.rangeByState(state, 0, limit, 'asc', (err, jobs) => {
       if (err) {
@@ -75,22 +66,52 @@ function rangeJobsByState(state = 'complete', limit) {
   });
 }
 
+async function getJobBatch(n, includeStuck) {
+  let jobs = [];
+
+  jobs = await rangeJobsByState('complete', n);
+
+  if (includeStuck) {
+    jobs = jobs.concat(await rangeJobsByState('failed', n));
+  }
+
+  return jobs;
+}
+
 /**
  * Cleans up the jobs that are in the queue.
  */
 async function cleanupJobs(options) {
+
+  // The scraper only needs to shutdown when the scraper has actually been
+  // started.
+  util.onshutdown([
+    () => kue.Task.shutdown()
+  ]);
+
   const n = 100;
 
   try {
-    const joblists = await Promise.all([
-      rangeJobsByState('complete', n),
-      options.stuck ? rangeJobsByState('failed', n) : false
-    ]);
 
-    await joblists.filter((jobs) => jobs).map(removeJobs);
+    // Connect to redis by establishing a queue.
+    kue.Task.connect();
+
+    let jobCount = 0;
+    let jobs = await getJobBatch(n, options.stuck);
+
+    while (jobs.length > 0) {
+
+      // Remove all the jobs.
+      await Promise.all(jobs.map((job) => removeJob(job)));
+
+      jobCount += jobs.length;
+
+      // Get the next batch of jobs.
+      jobs = await getJobBatch(n, options.stuck);
+    }
 
     util.shutdown();
-    console.log('Removed old jobs');
+    console.log(`Removed ${jobCount} jobs`);
   } catch (err) {
     console.error(err);
     util.shutdown(1);