<commit_before>const assert = require('assert');
const EventEmitter = require('events');
const Promise = require('bluebird');
const uuid = require('uuid');

const Worker = require('./worker');
const plans = require('./plans');
const Attorney = require('./attorney');

const expiredJobSuffix = plans.expiredJobSuffix;
const completedJobSuffix = plans.completedJobSuffix;
const failedJobSuffix = plans.failedJobSuffix;

const events = {
  job: 'job',
  failed: 'failed',
  error: 'error'
};

class Manager extends EventEmitter {
  constructor(db, config){
    super();

    this.config = config;
    this.db = db;

    this.events = events;
    this.subscriptions = {};

    this.nextJobCommand = plans.fetchNextJob(config.schema);
    this.insertJobCommand = plans.insertJob(config.schema);
    this.completeJobCommand = plans.completeJob(config.schema);
    this.completeJobsCommand = plans.completeJobs(config.schema);
    this.cancelJobCommand = plans.cancelJob(config.schema);
    this.cancelJobsCommand = plans.cancelJobs(config.schema);
    this.failJobCommand = plans.failJob(config.schema);
    this.failJobsCommand = plans.failJobs(config.schema);

    // exported api to index
    this.functions = [
      this.fetch,
      this.complete,
      this.cancel,
      this.fail,
      this.publish,
      this.subscribe,
      this.unsubscribe,
      this.onComplete,
      this.offComplete,
      this.onExpire,
      this.offExpire,
      this.onFail,
      this.offFail,
      this.fetchFailed,
      this.fetchExpired,
      this.fetchCompleted
    ];
  }

  stop() {
    Object.keys(this.subscriptions).forEach(name => this.unsubscribe(name));
    this.subscriptions = {};
    return Promise.resolve(true);
  }

  subscribe(name, ...args){
    return Attorney.checkSubscribeArgs(name, args)
      .then(({options, callback}) => this.watch(name, options, callback));
  }

  onExpire(name, ...args) {
    // unwrapping job in callback here because we love our customers
    return Attorney.checkSubscribeArgs(name, args)
      .then(({options, callback}) => this.watch(name + expiredJobSuffix, options, job => callback(job.data)));
  }

  onComplete(name, ...args) {
    return Attorney.checkSubscribeArgs(name, args)
      .then(({options, callback}) => this.watch(name + completedJobSuffix, options, callback));
  }

  onFail(name, ...args) {
    return Attorney.checkSubscribeArgs(name, args)
      .then(({options, callback}) => this.watch(name + failedJobSuffix, options, callback));
  }

  watch(name, options, callback){
    assert(!(name in this.subscriptions), 'this job has already been subscribed on this instance.');

    options.batchSize = options.batchSize || options.teamSize;

    if('newJobCheckInterval' in options || 'newJobCheckIntervalSeconds' in options)
      options = Attorney.applyNewJobCheckInterval(options);
    else
      options.newJobCheckInterval = this.config.newJobCheckInterval;

    let subscription = this.subscriptions[name] = {worker:null};

    let onError = error => this.emit(events.error, error);

    let complete = (job, error, response) => {
      if(!error)
        return this.complete(job.id, response);

      return this.fail(job.id)
        .then(() => this.emit(events.failed, {job, error}));
    };

    let respond = jobs => {
      if (!jobs) return;

      if (!Array.isArray(jobs))<commit_msg>pass the error supplied to the done callback on through the fail handler<commit_after>const assert = require('assert');
const EventEmitter = require('events');
const Promise = require('bluebird');
const uuid = require('uuid');

const Worker = require('./worker');
const plans = require('./plans');
const Attorney = require('./attorney');

const expiredJobSuffix = plans.expiredJobSuffix;
const completedJobSuffix = plans.completedJobSuffix;
const failedJobSuffix = plans.failedJobSuffix;

const events = {
  job: 'job',
  failed: 'failed',
  error: 'error'
};

class Manager extends EventEmitter {
  constructor(db, config){
    super();

    this.config = config;
    this.db = db;

    this.events = events;
    this.subscriptions = {};

    this.nextJobCommand = plans.fetchNextJob(config.schema);
    this.insertJobCommand = plans.insertJob(config.schema);
    this.completeJobCommand = plans.completeJob(config.schema);
    this.completeJobsCommand = plans.completeJobs(config.schema);
    this.cancelJobCommand = plans.cancelJob(config.schema);
    this.cancelJobsCommand = plans.cancelJobs(config.schema);
    this.failJobCommand = plans.failJob(config.schema);
    this.failJobsCommand = plans.failJobs(config.schema);

    // exported api to index
    this.functions = [
      this.fetch,
      this.complete,
      this.cancel,
      this.fail,
      this.publish,
      this.subscribe,
      this.unsubscribe,
      this.onComplete,
      this.offComplete,
      this.onExpire,
      this.offExpire,
      this.onFail,
      this.offFail,
      this.fetchFailed,
      this.fetchExpired,
      this.fetchCompleted
    ];
  }

  stop() {
    Object.keys(this.subscriptions).forEach(name => this.unsubscribe(name));
    this.subscriptions = {};
    return Promise.resolve(true);
  }

  subscribe(name, ...args){
    return Attorney.checkSubscribeArgs(name, args)
      .then(({options, callback}) => this.watch(name, options, callback));
  }

  onExpire(name, ...args) {
    // unwrapping job in callback here because we love our customers
    return Attorney.checkSubscribeArgs(name, args)
      .then(({options, callback}) => this.watch(name + expiredJobSuffix, options, job => callback(job.data)));
  }

  onComplete(name, ...args) {
    return Attorney.checkSubscribeArgs(name, args)
      .then(({options, callback}) => this.watch(name + completedJobSuffix, options, callback));
  }

  onFail(name, ...args) {
    return Attorney.checkSubscribeArgs(name, args)
      .then(({options, callback}) => this.watch(name + failedJobSuffix, options, callback));
  }

  watch(name, options, callback){
    assert(!(name in this.subscriptions), 'this job has already been subscribed on this instance.');

    options.batchSize = options.batchSize || options.teamSize;

    if('newJobCheckInterval' in options || 'newJobCheckIntervalSeconds' in options)
      options = Attorney.applyNewJobCheckInterval(options);
    else
      options.newJobCheckInterval = this.config.newJobCheckInterval;

    let subscription = this.subscriptions[name] = {worker:null};

    let onError = error => this.emit(events.error, error);

    let complete = (job, error, response) => {
      if(!error)
        return this.complete(job.id, response);

      return this.fail(job.id, error)
        .then(() => this.emit(events.failed, {job, error}));
    };

    let respond = jobs => {
      if (!jobs) return;

      if (!Array.isArray(jobs))
{'fraction_non_alphanumeric': 0.13332351208014143, 'fraction_numerical': 0.0, 'mean_word_length': 3.50796812749004, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1624208', 'n_tokens_mistral': 2025, 'n_tokens_neox': 1976, 'n_words': 432}