<commit_before>'use strict';

const Firebase = require('firebase');
const lodashDebounce = require('lodash.debounce');
const once = require('lodash.once');
const asyncLib = require('async');

const noop = () => undefined;

const verifier = require('./verifier');
const events = require('events');

const DEFAULT_MAX_WORKER = 2;

// TODO: should be saved in FB db and shared between verifiers.
const DEFAULT_PRESENCE_DELAY = 30000;
const DEFAULT_TASK_TIMEOUT = 6000;

module.exports = class Queue extends events.EventEmitter {

  constructor(firebaseClient, dockerClient, options) {
    super();

    options = options || {};

    this.ref = firebaseClient;
    this.dockerClient = dockerClient;
    this.logger = options.logger || console;

    this.imageTag = options.imageTag;
    this.opts = {
      presenceDelay: options.presenceDelay || DEFAULT_PRESENCE_DELAY,
      taskTimeout: options.taskTimeout || DEFAULT_TASK_TIMEOUT,
      maxWorker: options.maxWorker || DEFAULT_MAX_WORKER
    };

    if (this.opts.maxWorker < 1) {
      this.opts.maxWorker = 1;
    }

    this.queueName = this.ref.key();
    this.singpathRef = this.ref.root().child('singpath');
    this.tasksRef = this.ref.child('tasks');
    this.workersRef = this.ref.child('workers');

    this.taskQueue = asyncLib.queue((task, callback) => {
      this.runTask(task).then(
        result => callback(undefined, result)
      ).catch(
        err => callback(err)
      );
    }, this.opts.maxWorker);

    this.authData = undefined;
    this.ref.onAuth(authData => {
      this.authData = authData;

      if (authData && authData.uid) {
        this.emit('loggedIn', authData);
      } else {
        this.emit('loggedOut', authData);
      }
    });
  }

  isLoggedIn() {
    return this.authData && this.authData.uid;
  }

  isWorker() {
    return (
      this.isLoggedIn() &&
      this.authData.auth &&
      this.authData.auth.isWorker &&
      this.authData.auth.queue === this.queueName
    );
  }

  /**
   * Authenticate the firebase client with the custom token.
   *
   * Note only one firebase client per process. You cannot use two firebase
   * client with different token.
   *
   * @param  {string}  token
   * @return {Promise}
   */
  auth(token) {
    return new Promise((resolve, reject) => {
      this.ref.authWithCustomToken(token, (err, authData) => {
        if (err) {
          reject(err);
        } else {
          this.authData = authData;
          resolve(authData);
        }
      });
    });
  }

  /**
   * Push a new task to the firebase queue.
   *
   * To use for developpment.
   *
   * @param  {object}  payload Verification request
   *                           (should include a language, a solution and a test)
   * @return {Promise}         Promise resolving to the new task reference.
   */
  pushToQueue(payload) {
    if (!this.isLoggedIn()) {
      return Promise.reject(new Error('No user logged in.'));
    }

    return promisedPush(this.tasksRef, {
      started: false,
      completed: false,
      consumed: false,
      owner: this.authData.uid,
      payload: payload,
      createdAt: Firebase.ServerValue.TIMESTAMP
    });
  }

  /**
   * Register a worker for a queue and start a timer to update the worker
   * presence.
   *
   * @return {Promise} Resolve when the worker is registered, to a function to
   *                   deregister it.
   */
  registerWorker(opts) {
    opts = opts || {};

    if (!this.isWorker()) {
      return Promise.reject(new Error('The user is not logged in as a worker for this queue'));
    }

    const timer = Object.assign({
      start: setInterval,
      cancel: clearInterval
    }, opts.timer || {});

    return promisedSet(this.workersRef.child(this.authData.uid), {
      startedAt: Firebase.ServerValue.TIMESTAMP,
      presence: Firebase.ServerValue.TIMESTAMP
    }).then(ref => {
      let intervalId, stopTimer;

      this.logger.info('Worker registered.');

      intervalId = timer.start(() => {
        this.updatePresence().catch(stopTimer);
      }, this.opts.presenceDelay);

      stopTimer = () => {
        if (intervalId !== undefined) {
          timer.cancel(intervalId);
          intervalId = undefined;
          this.logger.debug('Stopping updating presence.');
        }
      };

      return () => {
        stopTimer();

        if (!this.isWorker()) {
          return Promise.reject(new Error('The user is not logged in as a worker for this queue'));
        }

        return promisedSet(ref, null).then(() => this.logger.info('Worker removed.'));
      };
    });
  }

  /**
   * Update the worker presence.
   *
   * @return {Promise} Resolve when the presence is updated
   */
  updatePresence() {
    if (!this.isWorker()) {
      return Promise.reject(new Error('The user is not logged in as a worker for this queue'));
    }

    return promisedSet(
      this.workersRef.child(this.authData.uid).child('presence'),
      Firebase.ServerValue.TIMESTAMP
    ).then(
      () => this.logger.debug('Worker presence updated')
    ).catch(err => {
      this.logger.error('Failed to update worker presence: %s', err.toString());
      return Promise.reject(err);
    });
  }

  /**
   * Start watching the task queue and workers' presence.
   *
   * Any opened task or opened task added or updated later, should be sheduled
   * to be verified.
   *
   * Any worker craching should be removed from the workers list.
   *
   * Returns a promise resolving when the worker is registered and the watch
   * starts. It will resolve to a fn that will stop the watch when called.
   *
   * Note that you do need not it to call it the auth token expire.
   *
   * It will reject if the worker couldn't register itself.
   *
   * To deal with Auth token expiring, you should listen for "watchStopped"
   * event and then restart watching with a new token.
   *
   * @return {Promise}
   */
  watch() {
    return this.registerWorker().then(deregister => {
      let cancel = noop;

      const failureHandler = once(err => {
        this.logger.error('Watch on new task failed unexpectively: %s', err.toString());
        cancel(err);
      });

      const stopWorkerWatch = this.monitorWorkers(failureHandler);
      const stopPendingTaskWatch = this.monitorPendingTask(failureHandler);

      cancel = (err) => {
        this.emit('watchStopped', err);
        this.logger.info('Watch on new task stopped.');

        return Promise.all([
          deregister, stopWorkerWatch, stopPendingTaskWatch, () => this.reset()
        ].map(fn => {
          try {
            return fn();
          } catch (e) {
            return Promise.reject(e);
          }
        }));
      };

      this.emit('watchStarted', cancel);
      this.logger.info('Starting watching for new task.');
      return cancel;
    });
  }

  /**
   * Monitor task queue to shedule any (re)opened task.
   *
   * Any tasks pending in the queue, tasks added and task unclaimed later.
   *
   * @param  {Function} failHandler
   * @return {Function}
   */
  monitorPendingTask(failHandler) {
    const query = this.tasksRef.orderByChild('started').equalTo(false);
    const eventTypes = ['child_added', 'child_changed'];
    const handler = snapshot => this.sheduleTask(snapshot.key(), snapshot.val());

    eventTypes.forEach(type => query.on(type, handler, failHandler));

    return () => {
      return eventTypes.map(type => {
        try {
          return query.off(type, handler);
        } catch (e) {
          this.logger.error(e.toString());
        }
      });
    };
  }

  /**
   * Schedule run of a new task.
   *
   * The task will be run immedialy or enqueue if if there are to many concurent
   * task running.
   *
   * @param  {string} key  Task id
   * @param  {Object} data Task body
   * @return {Promise}
   */
  sheduleTask(key, data) {
    if (!this.isWorker()) {
      return Promise.reject(new Error('The user is not logged in as a worker for this queue'));
    }

    const language = data && data.payload && data.payload.language;
    if (!verifier.support(language)) {
      this.logger.info('Task ("%s") language ("%s") is not supported', key, language);
      return Promise.reject(new Error('Unsupported language'));
    }

    const lastTry = data && data.tries && data.tries[this.authData.uid];
    if (lastTry) {
      this.logger.info('Already failed to run task. Skipping it (%s).', key);
      return Promise.reject(new Error('Already failed with that worker'));
    }

    this.logger.info('Task ("%s") run scheduled', key);
    this.logger.debug('Task ("%s") run scheduled with "%j"', key, data);

    return new Promise((resolve, reject) => {
      this.taskQueue.push({key, data}, (err, result) => {
        if (err) {
          return reject(err);
        }

        resolve(result);
      });
    });
  }

  /**
   * Process a task.
   *
   * @param  {Object} task Task key and body.
   * @return {Promise}     Resolve when the queue is empty.
   */
  runTask(task) {
    if (
      !task ||
      !task.key ||
      !task.data ||
      !task.data.payload
    ) {
      return Promise.reject(new Error('Unvalid task'));
    }

    const skip = {};

    return this.claimTask(task).catch(
      () => Promise.reject(skip)
    ).then(
      () => verifier.verify(this.dockerClient, task.data.payload, {
        logger: this.logger,
        imageTag: this.imageTag
      })
    ).then(results => {
      this.logger.info('Task ("%s") run.', task.key);
      this.logger.debug('Task ("%s") run: "%j".', task.key, results);

      return this.saveTaskResults(task, results);
    }).catch(err => {
      if (err === skip) {
        return;
      }

      this.logger.info('Task ("%s") failed running: %s.\n%s', task.key, err.toString(), err.stack);
      return this.removeTaskClaim(task).then(() => Promise.reject(err));
    }).catch(err => {
      this.logger.error(err);
      return Promise.reject(err);
    });
  }

  /**
   * Claim a task.
   *
   * Resolve when the task is claimed.
   *
   * @param  {Object} task Task key and body.
   * @return {Promise}     Resolve when the
   */
  claimTask(task) {
    if (!this.isWorker()) {
      return Promise.reject(new Error('The user is not logged in as a worker for this queue'));
    }

    return promisedUpdate(this.tasksRef.child(task.key), {
      worker: this.authData.uid,
      started: true,
      startedAt: Firebase.ServerValue.TIMESTAMP
    }).then(
      () => this.logger.info('Task ("%s") claimed.', task.key)
    ).catch(err => {
      this.logger.debug('Failed to claim task ("%s"): %s', task.key, err.toString());
      return Promise.reject(err);
    });
  }

  /**
   * Remove claim on a task
   *
   * @param  {Object} task Task key and body
   * @return {Promise}     Resolve when claim is removed.
   */<commit_msg>Fix saving worker try.<commit_after>'use strict';

const Firebase = require('firebase');
const lodashDebounce = require('lodash.debounce');
const once = require('lodash.once');
const asyncLib = require('async');

const noop = () => undefined;

const verifier = require('./verifier');
const events = require('events');

const DEFAULT_MAX_WORKER = 2;

// TODO: should be saved in FB db and shared between verifiers.
const DEFAULT_PRESENCE_DELAY = 30000;
const DEFAULT_TASK_TIMEOUT = 6000;

module.exports = class Queue extends events.EventEmitter {

  constructor(firebaseClient, dockerClient, options) {
    super();

    options = options || {};

    this.ref = firebaseClient;
    this.dockerClient = dockerClient;
    this.logger = options.logger || console;

    this.imageTag = options.imageTag;
    this.opts = {
      presenceDelay: options.presenceDelay || DEFAULT_PRESENCE_DELAY,
      taskTimeout: options.taskTimeout || DEFAULT_TASK_TIMEOUT,
      maxWorker: options.maxWorker || DEFAULT_MAX_WORKER
    };

    if (this.opts.maxWorker < 1) {
      this.opts.maxWorker = 1;
    }

    this.queueName = this.ref.key();
    this.singpathRef = this.ref.root().child('singpath');
    this.tasksRef = this.ref.child('tasks');
    this.workersRef = this.ref.child('workers');

    this.taskQueue = asyncLib.queue((task, callback) => {
      this.runTask(task).then(
        result => callback(undefined, result)
      ).catch(
        err => callback(err)
      );
    }, this.opts.maxWorker);

    this.authData = undefined;
    this.ref.onAuth(authData => {
      this.authData = authData;

      if (authData && authData.uid) {
        this.emit('loggedIn', authData);
      } else {
        this.emit('loggedOut', authData);
      }
    });
  }

  isLoggedIn() {
    return this.authData && this.authData.uid;
  }

  isWorker() {
    return (
      this.isLoggedIn() &&
      this.authData.auth &&
      this.authData.auth.isWorker &&
      this.authData.auth.queue === this.queueName
    );
  }

  /**
   * Authenticate the firebase client with the custom token.
   *
   * Note only one firebase client per process. You cannot use two firebase
   * client with different token.
   *
   * @param  {string}  token
   * @return {Promise}
   */
  auth(token) {
    return new Promise((resolve, reject) => {
      this.ref.authWithCustomToken(token, (err, authData) => {
        if (err) {
          reject(err);
        } else {
          this.authData = authData;
          resolve(authData);
        }
      });
    });
  }

  /**
   * Push a new task to the firebase queue.
   *
   * To use for developpment.
   *
   * @param  {object}  payload Verification request
   *                           (should include a language, a solution and a test)
   * @return {Promise}         Promise resolving to the new task reference.
   */
  pushToQueue(payload) {
    if (!this.isLoggedIn()) {
      return Promise.reject(new Error('No user logged in.'));
    }

    return promisedPush(this.tasksRef, {
      started: false,
      completed: false,
      consumed: false,
      owner: this.authData.uid,
      payload: payload,
      createdAt: Firebase.ServerValue.TIMESTAMP
    });
  }

  /**
   * Register a worker for a queue and start a timer to update the worker
   * presence.
   *
   * @return {Promise} Resolve when the worker is registered, to a function to
   *                   deregister it.
   */
  registerWorker(opts) {
    opts = opts || {};

    if (!this.isWorker()) {
      return Promise.reject(new Error('The user is not logged in as a worker for this queue'));
    }

    const timer = Object.assign({
      start: setInterval,
      cancel: clearInterval
    }, opts.timer || {});

    return promisedSet(this.workersRef.child(this.authData.uid), {
      startedAt: Firebase.ServerValue.TIMESTAMP,
      presence: Firebase.ServerValue.TIMESTAMP
    }).then(ref => {
      let intervalId, stopTimer;

      this.logger.info('Worker registered.');

      intervalId = timer.start(() => {
        this.updatePresence().catch(stopTimer);
      }, this.opts.presenceDelay);

      stopTimer = () => {
        if (intervalId !== undefined) {
          timer.cancel(intervalId);
          intervalId = undefined;
          this.logger.debug('Stopping updating presence.');
        }
      };

      return () => {
        stopTimer();

        if (!this.isWorker()) {
          return Promise.reject(new Error('The user is not logged in as a worker for this queue'));
        }

        return promisedSet(ref, null).then(() => this.logger.info('Worker removed.'));
      };
    });
  }

  /**
   * Update the worker presence.
   *
   * @return {Promise} Resolve when the presence is updated
   */
  updatePresence() {
    if (!this.isWorker()) {
      return Promise.reject(new Error('The user is not logged in as a worker for this queue'));
    }

    return promisedSet(
      this.workersRef.child(this.authData.uid).child('presence'),
      Firebase.ServerValue.TIMESTAMP
    ).then(
      () => this.logger.debug('Worker presence updated')
    ).catch(err => {
      this.logger.error('Failed to update worker presence: %s', err.toString());
      return Promise.reject(err);
    });
  }

  /**
   * Start watching the task queue and workers' presence.
   *
   * Any opened task or opened task added or updated later, should be sheduled
   * to be verified.
   *
   * Any worker craching should be removed from the workers list.
   *
   * Returns a promise resolving when the worker is registered and the watch
   * starts. It will resolve to a fn that will stop the watch when called.
   *
   * Note that you do need not it to call it the auth token expire.
   *
   * It will reject if the worker couldn't register itself.
   *
   * To deal with Auth token expiring, you should listen for "watchStopped"
   * event and then restart watching with a new token.
   *
   * @return {Promise}
   */
  watch() {
    return this.registerWorker().then(deregister => {
      let cancel = noop;

      const failureHandler = once(err => {
        this.logger.error('Watch on new task failed unexpectively: %s', err.toString());
        cancel(err);
      });

      const stopWorkerWatch = this.monitorWorkers(failureHandler);
      const stopPendingTaskWatch = this.monitorPendingTask(failureHandler);

      cancel = (err) => {
        this.emit('watchStopped', err);
        this.logger.info('Watch on new task stopped.');

        return Promise.all([
          deregister, stopWorkerWatch, stopPendingTaskWatch, () => this.reset()
        ].map(fn => {
          try {
            return fn();
          } catch (e) {
            return Promise.reject(e);
          }
        }));
      };

      this.emit('watchStarted', cancel);
      this.logger.info('Starting watching for new task.');
      return cancel;
    });
  }

  /**
   * Monitor task queue to shedule any (re)opened task.
   *
   * Any tasks pending in the queue, tasks added and task unclaimed later.
   *
   * @param  {Function} failHandler
   * @return {Function}
   */
  monitorPendingTask(failHandler) {
    const query = this.tasksRef.orderByChild('started').equalTo(false);
    const eventTypes = ['child_added', 'child_changed'];
    const handler = snapshot => this.sheduleTask(snapshot.key(), snapshot.val());

    eventTypes.forEach(type => query.on(type, handler, failHandler));

    return () => {
      return eventTypes.map(type => {
        try {
          return query.off(type, handler);
        } catch (e) {
          this.logger.error(e.toString());
        }
      });
    };
  }

  /**
   * Schedule run of a new task.
   *
   * The task will be run immedialy or enqueue if if there are to many concurent
   * task running.
   *
   * @param  {string} key  Task id
   * @param  {Object} data Task body
   * @return {Promise}
   */
  sheduleTask(key, data) {
    if (!this.isWorker()) {
      return Promise.reject(new Error('The user is not logged in as a worker for this queue'));
    }

    const language = data && data.payload && data.payload.language;
    if (!verifier.support(language)) {
      this.logger.info('Task ("%s") language ("%s") is not supported', key, language);
      return Promise.reject(new Error('Unsupported language'));
    }

    const lastTry = data && data.tries && data.tries[this.authData.uid];
    if (lastTry) {
      this.logger.info('Already failed to run task. Skipping it (%s).', key);
      return Promise.reject(new Error('Already failed with that worker'));
    }

    this.logger.info('Task ("%s") run scheduled', key);
    this.logger.debug('Task ("%s") run scheduled with "%j"', key, data);

    return new Promise((resolve, reject) => {
      this.taskQueue.push({key, data}, (err, result) => {
        if (err) {
          return reject(err);
        }

        resolve(result);
      });
    });
  }

  /**
   * Process a task.
   *
   * @param  {Object} task Task key and body.
   * @return {Promise}     Resolve when the queue is empty.
   */
  runTask(task) {
    if (
      !task ||
      !task.key ||
      !task.data ||
      !task.data.payload
    ) {
      return Promise.reject(new Error('Unvalid task'));
    }

    const skip = {};

    return this.claimTask(task).catch(
      () => Promise.reject(skip)
    ).then(
      () => verifier.verify(this.dockerClient, task.data.payload, {
        logger: this.logger,
        imageTag: this.imageTag
      })
    ).then(results => {
      this.logger.info('Task ("%s") run.', task.key);
      this.logger.debug('Task ("%s") run: "%j".', task.key, results);

      return this.saveTaskResults(task, results);
    }).catch(err => {
      if (err === skip) {
        return;
      }

      this.logger.info('Task ("%s") failed running: %s.\n%s', task.key, err.toString(), err.stack);
      return this.removeTaskClaim(task).then(() => Promise.reject(err));
    }).catch(err => {
      this.logger.error(err);
      return Promise.reject(err);
    });
  }

  /**
   * Claim a task.
   *
   * Resolve when the task is claimed.
   *
   * @param  {Object} task Task key and body.
   * @return {Promise}     Resolve when the
   */
  claimTask(task) {
    if (!this.isWorker()) {
      return Promise.reject(new Error('The user is not logged in as a worker for this queue'));
    }

    return promisedUpdate(this.tasksRef.child(task.key), {
      worker: this.authData.uid,
      started: true,
      startedAt: Firebase.ServerValue.TIMESTAMP
    }).then(() => {
      () => this.logger.info('Task ("%s") claimed.', task.key);

      if (task.data) {
        task.data.worker = this.authData.uid;
      }
    }).catch(err => {
      this.logger.debug('Failed to claim task ("%s"): %s', task.key, err.toString());
      return Promise.reject(err);
    });
  }

  /**
   * Remove claim on a task
   *
   * @param  {Object} task Task key and body
   * @return {Promise}     Resolve when claim is removed.
   */
{'fraction_non_alphanumeric': 0.13669429097605892, 'fraction_numerical': 0.0011049723756906078, 'mean_word_length': 2.442313787638669, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 70, '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': '5951979', 'n_tokens_mistral': 6590, 'n_tokens_neox': 6454, 'n_words': 1739}