Job queue handler abstraction

JobHandler.js

const Bull = use("Rocketseat/Bull");
const moment = require("moment");
const Redis = use("Redis");

/**
 * Prepare queue configuration. Options provided overrides
 * the field value of the default configuration.
 *
 * @param {object} [options]
 */
const prepareConfig = (...options) => {
  const defaultConfig = { attempts: 3 };
  return Object.assign({}, defaultConfig, options);
};

/**
 * Dispatch a job to the default queue provider.
 *
 * @param {object} job
 * @param {String} job.key
 * @param {object} job.options
 * @param {Object} data
 */
const dispatch = (job, data, options = {}) =>
  Bull.add(job.key, data, prepareConfig(job.options, options));

/**
 * Schedule a job to the time specified.
 *
 * @param {object} job
 * @param {String} job.key
 * @param {object} job.options
 * @param {Object} data
 * @param {Date} datetime
 */
const schedule = (job, data, datetime, options = {}) =>
  Bull.schedule(job.key, data, datetime, prepareConfig(job.options, options));

/**
 * Schedule a job but throttle it's execution for the specified
 * delay (in milliseconds).
 *
 * @param {String} id
 * @param {object} job
 * @param {String} job.key
 * @param {Number} delay
 * @param {object} job.options
 * @param {Object} data
 */
const scheduleButthrottleFor = async (id, job, data, delay = 360000, options = {}) => {
  const key = `${job.key}-${id}-${delay}`;
  const exists = await Redis.get(key);

  if (!exists) {
    await Redis.set(key, id, "EX", delay);
    const scheduleAt = moment().add(delay, "millisecond").format("YYYY-MM-DD HH:mm:ss");

    Bull.schedule(
      job.key,
      data,
      scheduleAt,
      prepareConfig(job.options, options)
    );
  }
};

module.exports = { dispatch, schedule, scheduleButthrottleFor };