Simple kue cleanup script

gistfile1.js

  // this is a simple log action
  function QueueActionLog(message) {
    this.message = message || 'QueueActionLog :: got a action for job id(%s)';

    this.apply = function(job) {
      console.log(util.format(this.message, job.id));
      return true;
    };
  }

  // remove item action
  function QueueActionRemove(age) {
    this.age = age;

    this.apply = function(job) {
      job.remove();
      return true;
    };
  }

  // filter by age
  function QueueFilterAge(age) {
    this.now = new Date();
    this.age = age;

    this.test = function(job) {
      var created = new Date(parseInt(job.created_at));
      var age = parseInt(this.now - created);

      return age > this.age;

    };
  }

  // the queue iterator
  var queueIterator = function(ids, queueFilterChain, queueActionChain) {
    _.each(ids, function(id, index) {

      // get the kue job
      kue.Job.get(id, function(err, job) {
        var filterIterator = function(filter) { return filter.test(job) };
        var actionIterator = function(filter) { return filter.apply(job) };

        // apply filter chain
        if(_.every(queueFilterChain, filterIterator)) {

          // apply action chain
          _.every(queueActionChain, actionIterator);
        }
      });
    });
  };

  setInterval(function() {
    var ki = new kue;
    
    ki.failed(function(err, ids) {
      queueIterator(
        ids,
        [new QueueFilterAge(jobs.CLEANUP_MAX_FAILED_TIME)],
        [new QueueActionLog('Going to remove job id(%s) for being failed to long'),
          new QueueActionRemove()]
      );
    });

    ki.active(function(err, ids) {
      queueIterator(
        ids,
        [new QueueFilterAge(jobs.CLEANUP_MAX_ACTIVE_TIME)],
        [new QueueActionLog('Going to remove job id(%s) for being active to long'),
          new QueueActionRemove()]
      );
    });

    ki.complete(function(err, ids) {
      queueIterator(
        ids,
        [new QueueFilterAge(jobs.CLEANUP_MAX_COMPLETE_TIME)],
        [new QueueActionLog('Going to remove job id(%s) for being complete to long'),
          new QueueActionRemove()]
      );
    });
  }, jobs.CLEANUP_INTERVAL);